SwaggerHound/swagger-hound.py

# -*- coding: utf-8 -*-
# https://github.com/evilc0deooo/SwaggerHound

import json
import sys
import csv
import argparse
import requests
import re
import random
import urllib3
from datetime import datetime
from urllib.parse import urlparse
from loguru import logger

# 禁用安全请求警告
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

logger.remove()
handler_id = logger.add(sys.stderr, level='DEBUG')  # 设置输出级别

now_time = datetime.now().strftime("%Y%m%d_%H%M%S")

proxies = {
    'https': 'http://127.0.0.1:7890',
    'http': 'http://127.0.0.1:7890'
}

# 开启代理
SET_PROXY = False

header_agents = [
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_3) AppleWebKit/537.75.14 (KHTML, like Gecko) Version/7.0.3 Safari/7046A194A',
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Code/1.96.2 Chrome/128.0.6613.186 Electron/32.2.6 Safari/537.36'
]


def http_req(url, method='get', **kwargs):
    kwargs.setdefault('verify', False)
    kwargs.setdefault('timeout', (10.1, 30.1))
    kwargs.setdefault('allow_redirects', False)

    headers = kwargs.get('headers', {})
    headers.setdefault('User-Agent', random.choice(header_agents))
    # 不允许缓存，每次请求都获取服务器上最新的资源
    headers.setdefault('Cache-Control', 'max-age=0')
    kwargs['headers'] = headers
    if SET_PROXY:
        kwargs['proxies'] = proxies

    conn = getattr(requests, method)(url, **kwargs)
    return conn


def check_page(url):
    """
    检查当前页面
    """
    res = http_req(url, method='get')
    if '<html' in res.text:
        logger.debug('[+] 输入为 swagger 首页，开始解析 api 文档地址')
        return 3  # swagger-html
    elif '"parameters"' in res.text:
        logger.debug('[+] 输入为 api 文档地址，开始构造请求发包')
        return 2  # api_docs
    elif '"location"' in res.text:
        logger.debug('[+] 输入为 resource 地址，开始解析 api 文档地址')
        return 1  # resource


def fill_parameters(parameters, url):
    """
    填充测试数据并替换 URL 中的占位符
    """
    filled_params = {}
    path_params = {}
    for param in parameters:
        param_name = param['name']
        param_in = param['in']
        param_type = param['type']

        # 根据类型填充默认值
        if param_type == 'string':
            value = 'a'
        elif param_type == 'integer':
            value = 1
        elif param_type == 'boolean':
            value = True
        else:
            value = ''

        if param_in == 'query':
            filled_params[param_name] = value
        elif param_in == 'path':
            path_params[param_name] = value
            filled_params[param_name] = value
        elif param_in == 'body':
            if 'body' not in filled_params:
                filled_params['body'] = {}
            filled_params['body'][param_name] = value

    # 替换 URL 中的占位符
    for key, value in path_params.items():
        url = url.replace(f'{{{key}}}', str(value))

    return filled_params, url


def get_api_docs_path(resource_url):
    """
    输入 resource 解析 api 文档 url
    """
    domain = urlparse(resource_url)
    domain = domain.scheme + '://' + domain.netloc
    try:
        res = http_req(resource_url, method='get')
        resources = json.loads(res.text)
    except Exception as e:
        logger.error(f'[-] {resource_url} error info {e}')
        return []

    paths = []
    if isinstance(resources, tuple):
        if 'apis' in resources.keys():  # 版本不同, 格式不一样
            for api_docs in resources.get('apis', {}):
                paths.append(domain + api_docs['path'])
            return paths
    else:
        for i in resources:
            paths.append(domain + i['location'])
        return paths


def output_to_csv(data):
    _f = open(f'{now_time}.csv', 'a', newline='', encoding='utf-8')  # 使用追加模式
    writer = csv.writer(_f)
    writer.writerow(data)


def go_resources(url):
    """
    解析 swagger-resources 获取 api-docs
    """
    try:
        _domain = urlparse(url)
        domain = _domain.scheme + '://' + _domain.netloc
        domain_path = _domain.path
        stripped_path = domain_path.strip('/')
        res = http_req(url)
        data = json.loads(res.text)
        for _i in data:
            location = _i.get('location')
            # 判断如果不存在路径则直接进行拼接
            target = domain + location
            if len(stripped_path) > 0:
                # 如果存在路径则在原路径上继续拼接
                target = url.rsplit('/', 1)[0] + location
            go_api_docs(target)  # 调用 api_docs 扫描全部接口

    except Exception as e:
        logger.error(f'[-] {url} error info {e}')


def go_swagger_html(url):
    """
    解析 swagger-ui.html 获取 api 接口路径
    """
    response = http_req(url)
    response.raise_for_status()
    html_content = response.text
    # 在 swagger-initializer.js 中获取 swagger.json 接口
    initializer_pattern = r'<script\s+src=["\']([^"\']*swagger-initializer\.js[^"\']*)["\']'
    initializer_match = re.search(initializer_pattern, html_content)
    if initializer_match:
        js_file_path = initializer_match.group(1)
        if js_file_path.startswith('http'):
            js_file_url = js_file_path
        else:
            _domain = urlparse(url)
            domain = _domain.scheme + '://' + _domain.netloc
            domain_path = _domain.path
            stripped_path = domain_path.strip('/')
            if len(stripped_path) > 0:
                base_url = url.rsplit('/', 1)[0]
                js_file_url = f'{base_url}/{js_file_path.lstrip("/")}'
            else:
                js_file_url = f'{domain}/{js_file_path.lstrip("/")}'

        js_response = http_req(js_file_url)
        js_response.raise_for_status()
        js_content = js_response.text

        # 正则获取 defaultDefinitionUrl 的值 swagger.json 接口路径
        js_pattern = r'const\s+defaultDefinitionUrl\s*=\s*["\']([^"\']+)["\'];'
        js_match = re.search(js_pattern, js_content)
        if js_match:
            api_docs_path = js_match.group(1)
            go_api_docs(api_docs_path)
            return

    # 未找到 swagger-initializer.js 文件或 defaultDefinitionUrl 定义, 则尝试查找 springfox.js 文件
    springfox_pattern = r'<script\s+src=["\']([^"\']*springfox\.js[^"\']*)["\']'
    springfox_match = re.search(springfox_pattern, html_content)
    if not springfox_match:
        logger.debug('[-] 未找到 swagger-initializer.js 和 springfox.js 文件路径')
        return

    # 获取 springfox.js 文件的相对或绝对路径
    springfox_file_path = springfox_match.group(1)
    if springfox_file_path.startswith('http'):
        springfox_file_url = springfox_file_path
    else:
        base_url = url.rsplit('/', 1)[0]
        springfox_file_url = f'{base_url}/{springfox_file_path.lstrip("/")}'

    # 发送请求获取 springfox.js 文件内容
    springfox_response = http_req(springfox_file_url)
    springfox_response.raise_for_status()
    springfox_content = springfox_response.text
    if "/swagger-resources" in springfox_content:
        base_url = url.rsplit('/', 1)[0]
        resource_url = f"{base_url}/swagger-resources"
        go_resources(resource_url)
        return


def go_api_docs(url):
    """
    开始 api-docs 解析并扫描
    """
    try:
        domain = urlparse(url)
        domain = domain.scheme + '://' + domain.netloc

        res = http_req(url)
        if res.status_code != 200:
            logger.error(f'[-] {url} req status is {res.status_code}')
            return
        try:
            data = json.loads(res.text)
        except json.JSONDecodeError:
            # 遇到 html 标签内存在双引号 json.loads 无法格式化, 需要特殊处理
            data = res.text.replace("'", '"')
            result = re.sub(r'<[^>]*>', lambda match: match.group(0).replace('"', "'"), data)
            data = json.loads(result, strict=False)

        if 'basePath' in data.keys():
            base_path = data['basePath']
        elif 'servers' in data.keys():
            base_path = data['servers']['url']
        else:
            base_path = ''

        paths = data.get('paths', {})
        definitions = data.get('definitions', {})
        swagger_result = []
        for path, methods in paths.items():
            for method, details in methods.items():  # get / post / put / update / delete / head...
                if method.upper() not in ['GET', 'POST']:  # http 请求方式白名单
                    continue
                req_path = domain + base_path + path
                summary = details.get('summary', path)  # 概要信息
                consumes = details.get('consumes', [])  # 数据请求类型 application/json
                params = details.get('parameters', [])
                logger.debug(f'test on {summary} => {method} => {req_path}')
                param_info = []
                for param in params:
                    param_name = param.get('name')
                    param_in = param.get('in')
                    schema = param.get('schema')
                    # 判断是否存在自定义的模型或对象
                    if schema and '$ref' in schema:
                        ref = schema['$ref'].split('/')[-1]
                        if ref in definitions:  # 如果在 definitions 中声明了参数属性，则去 definitions 定义中获取参数及属性信息
                            # 递归处理定义中的属性
                            for prop_name, prop_details in definitions[ref].get('properties', {}).items():
                                param_info.append({
                                    'name': prop_name,
                                    'in': param_in,
                                    'type': prop_details.get('type')
                                })
                    else:
                        param_type = param.get('type')
                        param_info.append({
                            'name': param_name,
                            'in': param_in,
                            'type': param_type
                        })

                # 解析 swagger 获取到所有需要的数据
                swagger_result.append({
                    'summary': summary,
                    'req_path': req_path,
                    'method': method,
                    'consumes': consumes,
                    'parameters': param_info
                })

        black_list_status = [401, 404, 502, 503]  # 状态码黑名单
        for item in swagger_result:
            summary = item['summary']
            req_path = item['req_path']
            method = item['method']
            consumes = item['consumes']
            parameters = item['parameters']
            # 生成发送的 Body 数据
            filled_params, new_url = fill_parameters(parameters, req_path)
            headers = {}

            if 'application/json' in consumes:
                headers = {'Content-Type': 'application/json'}
            if method.lower() == 'get':
                response = http_req(new_url, method='get', params=filled_params)
                if response.status_code in black_list_status:
                    logger.debug(f'[-] {method} {new_url} req status is {response.status_code}')
                    continue
                if response.status_code == 200:
                    logger.debug(f'[+] {method} {new_url} req status is {response.status_code}')
                    write_result = [url, new_url, summary, method, consumes, filled_params, response.status_code, response.text]
                    output_to_csv(write_result)

            elif method.lower() == 'post':
                if 'body' in filled_params:
                    response = http_req(new_url, method='post', json=filled_params['body'], headers=headers)
                    if response.status_code in black_list_status:
                        logger.debug(f'[-] {method} {new_url} req status is {response.status_code}')
                        continue
                    if response.status_code == 200:
                        logger.debug(f'[+] {method} {new_url} req status is {response.status_code}')
                        write_result = [url, new_url, summary, method, consumes, filled_params, response.status_code, response.text]
                        output_to_csv(write_result)

                else:
                    response = http_req(new_url, method='post', params=filled_params, headers=headers)
                    if response.status_code in black_list_status:
                        logger.debug(f'[-] {method} {new_url} req status is {response.status_code}')
                        continue
                    if response.status_code == 200:
                        logger.debug(f'[+] {method} {new_url} req status is {response.status_code}')
                        write_result = [url, new_url, summary, method, consumes, filled_params, response.status_code, response.text]
                        output_to_csv(write_result)


    except Exception as e:
        logger.error(f'[-] {url} error info {e}')


def run(target):
    """
    执行程序
    """
    url_type = check_page(target)
    if url_type == 1:
        logger.success('working on {}'.format(target), 'type: source')
        go_resources(target)
    elif url_type == 2:
        logger.success('working on {}'.format(target), 'type: api-docs')
        go_api_docs(target)
    else:
        logger.success('working on {}'.format(target), 'type: html')
        go_swagger_html(target)


if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('-u', '--url', dest='target_url', help='resource 地址 or api 文档地址 or swagger 首页地址')
    parser.add_argument('-f', '--file', dest='url_file', help='批量测试')
    args = parser.parse_args()

    logger.add('debug.log', format='{time:YYYY-MM-DD at HH:mm:ss} | {level} | {message}')
    if args.target_url:
        run(args.target_url)
    elif args.url_file:
        with open(args.url_file, 'r') as f:
            urls = [line.strip() for line in f.readlines()]
        for target_url in urls:
            print(target_url)
            run(target_url)