import argparse
import aiohttp
import asyncio
import time
from urllib.parse import urlparse

async def check_methods(session, url, semaphore, output_file):
    async with semaphore:
        try:
            # 自动补全协议头
            parsed = urlparse(url)
            if not parsed.scheme:
                url = 'http://' + url
                parsed = urlparse(url)
            
            # 如果仍然没有协议头，则使用https
            if not parsed.scheme:
                url = 'https://' + url
            
            async with session.options(url) as response:
                # 检查Allow或Access-Control-Allow-Methods头部
                allowed_methods = response.headers.get('Allow', '') or response.headers.get('Access-Control-Allow-Methods', '')
                if not allowed_methods:
                    return
                
                # 解析允许的方法并检查PUT/DELETE/MOVE
                allowed_methods = [method.strip().upper() for method in allowed_methods.split(',')]
                dangerous_methods = []
                for method in ['PUT', 'DELETE', 'MOVE']:
                    if method in allowed_methods:
                        dangerous_methods.append(method)
                
                if dangerous_methods:
                    print(f"\033[31m[+] 漏洞 {url} 允许危险方法: {', '.join(dangerous_methods)}\033[0m")
                    # 实时写入文件
                    with open(output_file, 'a', encoding='utf-8') as f:
                        f.write(f"{url} - 允许方法: {', '.join(dangerous_methods)}\n")
                else:
                    print(f"[+] 安全 {url}: {allowed_methods}")
        except aiohttp.ClientError:
            # 如果是http失败，尝试https
            if not url.startswith('https://'):
                try:
                    https_url = url.replace('http://', 'https://', 1)
                    async with session.options(https_url) as response:
                        allowed_methods = response.headers.get('Allow', '') or response.headers.get('Access-Control-Allow-Methods', '')
                        if not allowed_methods:
                            return
                        
                        allowed_methods = [method.strip().upper() for method in allowed_methods.split(',')]
                        dangerous_methods = []
                        for method in ['PUT', 'DELETE', 'MOVE']:
                            if method in allowed_methods:
                                dangerous_methods.append(method)
                        
                        if dangerous_methods:
                            print(f"\033[31m[+] 漏洞 {https_url} 允许危险方法: {', '.join(dangerous_methods)}\033[0m")
                            # 实时写入文件
                            with open(output_file, 'a', encoding='utf-8') as f:
                                f.write(f"{https_url} - 允许方法: {', '.join(dangerous_methods)}\n")
                        else:
                            print(f"[+] 安全 {https_url}: {allowed_methods}")
                except Exception:
                    pass
        except Exception:
            pass

async def process_urls(url_list, output_file, concurrency=50):
    # 清空或创建输出文件
    with open(output_file, 'w', encoding='utf-8') as f:
        pass
    
    # 配置TCP连接器和超时设置
    connector = aiohttp.TCPConnector(force_close=True)
    timeout = aiohttp.ClientTimeout(total=10)
    semaphore = asyncio.Semaphore(concurrency)
    
    async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
        tasks = [check_methods(session, url, semaphore, output_file) for url in url_list]
        await asyncio.gather(*tasks)

def read_urls_from_file(file_path):
    with open(file_path, 'r', encoding='utf-8') as f:
        return [line.strip() for line in f if line.strip()]

def main():
    parser = argparse.ArgumentParser(description='检测目标是否允许PUT/DELETE/MOVE等危险方法')
    parser.add_argument('-u', '--url', help='要检测的单个URL地址')
    parser.add_argument('-f', '--file', help='包含多个URL的文件路径')
    parser.add_argument('-o', '--output', default='2.txt', help='存在漏洞的URL输出文件(默认:2.txt)')
    parser.add_argument('-c', '--concurrency', type=int, default=50, help='并发连接数(默认:50)')
    
    args = parser.parse_args()
    
    if not args.url and not args.file:
        parser.print_help()
        return
    
    urls = []
    if args.url:
        urls.append(args.url)
    if args.file:
        urls.extend(read_urls_from_file(args.file))
    
    print(f"正在检测 {len(urls)} 个URL...\n")
    start_time = time.time()
    
    asyncio.run(process_urls(urls, args.output, args.concurrency))
    
    elapsed = time.time() - start_time
    print(f"\n检测完成! 总耗时: {elapsed:.2f}秒")
    print(f"存在漏洞的URL已实时保存到 {args.output}")

if __name__ == '__main__':
    main()

{/tabs-pane}
{tabs-pane label="如果支持PUT方法，批量写入"}

import argparse
import aiohttp
import asyncio
from urllib.parse import urlparse
import os

async def test_put_upload(session, url, semaphore, output_file):
    async with semaphore:
        try:
            # 自动补全协议头并规范化URL
            parsed = urlparse(url)
            if not parsed.scheme:
                url = f"http://{url}"
                parsed = urlparse(url)
            
            # 确保路径以/结尾（避免覆盖目录）
            path = parsed.path if parsed.path.endswith('/') else f"{parsed.path}/"
            target_url = f"{parsed.scheme}://{parsed.netloc}{path}1.txt"
            
            # 尝试PUT上传
            put_data = "123456"
            async with session.put(
                target_url,
                data=put_data,
                headers={'Content-Type': 'text/plain'},
                timeout=aiohttp.ClientTimeout(total=10)
            ) as put_response:
                # 检查PUT是否成功 (2xx状态码)
                if put_response.status not in range(200, 300):
                    print(f"[-] 上传失败 {target_url} (HTTP {put_response.status})")  # 白色
                    return
                
                # 验证文件是否可访问
                async with session.get(target_url, timeout=aiohttp.ClientTimeout(total=10)) as get_response:
                    if get_response.status == 200 and (await get_response.text()).strip() == put_data:
                        print(f"\033[31m[+] 成功 {target_url}\033[0m")  # 红色
                        with open(output_file, 'a') as f:
                            f.write(f"{target_url}\n")
                    else:
                        print(f"[-] 验证失败 {target_url} (HTTP {get_response.status})")  # 白色
        except Exception as e:
            print(f"[-] 错误 {url}: {str(e)}")  # 白色

async def process_urls(urls, output_file, concurrency=20):
    # 清空或创建输出文件
    with open(output_file, 'w') as f:
        pass
    
    # 配置HTTP客户端
    connector = aiohttp.TCPConnector(force_close=True)
    semaphore = asyncio.Semaphore(concurrency)
    
    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = [test_put_upload(session, url, semaphore, output_file) for url in urls]
        await asyncio.gather(*tasks)

def main():
    parser = argparse.ArgumentParser(description='测试URL是否允许PUT上传', formatter_class=argparse.RawTextHelpFormatter)
    parser.add_argument('-u', '--url', help='检测单个URL')
    parser.add_argument('-f', '--file', help='批量检测URL文件路径')
    parser.add_argument('-o', '--output', default='2.txt', help='输出文件路径（默认：2.txt）')
    parser.add_argument('-c', '--concurrency', type=int, default=20, 
                       help='并发数（仅在-f模式下有效，默认：20）')
    
    args = parser.parse_args()
    
    # 参数校验
    if not args.url and not args.file:
        parser.print_help()
        print("\n错误：必须指定-u或-f参数")
        return
    
    if args.url and args.file:
        print("错误：-u和-f参数不能同时使用")
        return
    
    # 准备URL列表
    urls = []
    if args.url:
        urls.append(args.url)
        args.concurrency = 1  # 单URL模式强制并发数为1
    elif args.file:
        if not os.path.exists(args.file):
            print(f"错误：输入文件 {args.file} 不存在")
            return
        with open(args.file, 'r') as f:
            urls = [line.strip() for line in f if line.strip()]
    
    print(f"正在测试 {len(urls)} 个URL...")
    asyncio.run(process_urls(urls, args.output, args.concurrency))
    print(f"\n测试完成！成功URL已保存到 {args.output}")

if __name__ == '__main__':
    main()

{/tabs-pane}