本文目录导读：
我来给你设计一个综合的Python安全防守案例,展示如何通过多种技术手段识别和定位防守漏洞。
漏洞扫描器架构设计
import socket
import ssl
import threading
import queue
import json
import hashlib
import re
import time
from dataclasses import dataclass
from typing import Dict, List, Any, Optional
import requests
from urllib.parse import urljoin, urlparse
import warnings
warnings.filterwarnings('ignore')
# ==================== 数据模型 ====================
@dataclass
class Vulnerability:
"""漏洞信息模型"""
name: str
severity: str # critical/high/medium/low
description: str
location: str # URL、IP或端口
evidence: str # 证据
remediation: str # 修复建议
confidence: float # 置信度 0-1
def to_dict(self):
return self.__dict__
class PortService:
"""端口服务信息"""
def __init__(self, port, service, version, banner):
self.port = port
self.service = service
self.version = version
self.banner = banner
# ==================== 漏洞识别模块 ====================
class HeaderChecker:
"""HTTP头部安全头检查"""
SECURITY_HEADERS = {
'X-Frame-Options': 'clickjacking防护',
'X-Content-Type-Options': 'MIME类型嗅探防护',
'Strict-Transport-Security': 'HSTS强制HTTPS',
'Content-Security-Policy': '内容安全策略',
'X-XSS-Protection': 'XSS过滤',
'Referrer-Policy': '来源策略',
'Permissions-Policy': '功能限制'
}
@staticmethod
def check(headers: Dict[str, str], url: str) -> List[Vulnerability]:
vulns = []
for header, desc in HeaderChecker.SECURITY_HEADERS.items():
if header.lower() not in {k.lower() for k in headers.keys()}:
vulns.append(Vulnerability(
name=f"缺少{header}头部",
severity='medium',
description=f"HTTP响应缺少安全头{header} ({desc})",
location=url,
evidence="-",
remediation=f"配置Web服务器添加{header}响应头",
confidence=0.9
))
return vulns
class SQLiChecker:
"""SQL注入检测"""
def __init__(self, session, timeout=5):
self.session = session
self.timeout = timeout
def check(self, param_info, url) -> List[Vulnerability]:
"""测试基本的SQL注入"""
vulns = []
# 获取参数
if not param_info.get('params'):
return vulns
# SQL注入payload集
payloads = [
("' OR '1'='1", "单引号注入"),
("1' OR '1'='1'-- -", "注释盲目注入"),
("1 UNION SELECT 1,2,3-- -", "UNION注入"),
("` OR 1=1 -- -", "反引号注入"),
("' OR 1=1 #", "MySQL盲注")
]
for param, values in param_info.items():
for payload, payload_type in payloads:
try:
test_params = values.copy()
test_params[param] = payload
# 测试GET参数
if values['method'] == 'GET':
resp = self.session.get(url, params=test_params,
timeout=self.timeout,
verify=False)
# 测试POST参数
elif values['method'] == 'POST':
resp = self.session.post(url, data=test_params,
timeout=self.timeout,
verify=False)
else:
continue
# 分析与正常请求的差异
if resp.status_code == 500 and 'sql' in resp.text.lower() or \
'mysql' in resp.text.lower() or \
'syntax error' in resp.text.lower():
vulns.append(Vulnerability(
name=f"SQL注入-{payload_type}",
severity='critical',
description=f"参数'{param}'存在SQL注入漏洞，payload: {payload}",
location=url,
evidence=f"HTTP {resp.status_code}: {resp.text[:200]}",
remediation="使用参数化查询或ORM框架",
confidence=0.75
))
break
except Exception as e:
continue
return vulns
class XSSChecker:
"""XSS攻击检测"""
XSS_PAYLOADS = [
('<script>alert(1)</script>', '反射型XSS'),
('<img src=x onerror=alert(1)>', '事件处理器XSS'),
('<svg/onload=alert(1)>', 'SVG payload'),
('javascript:alert(1)', 'JavaScript协议'),
('<iframe src="javascript:alert(1)">', 'iframe XSS')
]
@staticmethod
def check(content: str, url: str, param_name: str) -> Optional[Vulnerability]:
for payload, type_name in XSSChecker.XSS_PAYLOADS:
if payload in content:
return Vulnerability(
name=f"XSS漏洞-{type_name}",
severity='high',
description=f"参数'{param_name}'可能存在XSS漏洞",
location=url,
evidence=f"Payload '{payload}'在响应中",
remediation="对输出进行HTML编码，使用CSP策略",
confidence=0.7
)
return None
# ==================== 网络层漏洞识别 ====================
class PortScanner:
"""端口服务端口扫描器"""
def __init__(self, target_host, timeout=1.0, max_workers=100):
self.host = target_host
self.timeout = timeout
self.max_workers = max_workers
self.open_ports = []
self.port_queue = queue.Queue()
def scan_single_port(self, port):
"""扫描单个端口"""
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(self.timeout)
try:
result = sock.connect_ex((self.host, port))
if result == 0:
service = self.get_service_info(port, sock)
banner = self.grab_banner(sock)
self.open_ports.append(service)
except Exception as e:
pass
finally:
sock.close()
def get_service_info(self, port, sock):
"""尝试确定服务类型"""
# 常见端口对应服务
common_services = {
21: 'ftp', 22: 'ssh', 23: 'telnet', 25: 'smtp',
53: 'dns', 80: 'http', 443: 'https', 3306: 'mysql',
3389: 'rdp', 5432: 'postgresql', 6379: 'redis',
8080: 'http-proxy', 27017: 'mongodb'
}
return PortService(
port=port,
service=common_services.get(port, 'unknown'),
version='unknown',
banner='unknown'
)
def grab_banner(self, sock):
"""抓取banner信息"""
try:
sock.send(b'\r\n')
banner = sock.recv(1024).decode('utf-8', errors='ignore').strip()
return banner
except (socket.timeout, ConnectionError):
return "unknown"
def scan(self):
"""并发端口扫描"""
ports = set(range(1, 10000))
# 高优先级端口先扫描
priority_ports = [21, 22, 23, 25, 53, 80, 443, 3389, 8080, 3306, 5432, 6379]
for port in priority_ports:
if port in ports:
ports.remove(port)
# 使用线程池扫描
threads = []
# 先扫优先级高的
for port in priority_ports:
t = threading.Thread(target=self.scan_single_port, args=(port,))
threads.append(t)
t.start()
# 扫描剩余端口
for port in list(ports)[:2000]: # 限制扫描数量
while threading.active_count() > self.max_workers:
time.sleep(0.01)
t = threading.Thread(target=self.scan_single_port, args=(port,))
threads.append(t)
t.start()
# 等待所有线程完成
for t in threads:
t.join()
return self.open_ports
# ==================== 综合扫描器 ====================
class WebVulnerabilityScanner:
"""综合Web漏洞扫描器"""
def __init__(self, target_url, config=None):
self.target_url = target_url
self.session = requests.Session()
self.session.verify = False
# 加载配置文件
self.config = config or {
'headers': {}, # 自定义请求头
'proxy': None,
'threads': 10,
'depth': 3,
'max_urls': 100,
'check_headers': True,
'check_sqli': True,
'check_xss': True,
'check_csrf': True
}
self.vulnerabilities = []
self.crawled_urls = set()
self.url_queue = queue.Queue()
def identify_framework(self, headers, content):
"""指纹识别目标使用的框架"""
fingerprint = []
# 检测Web服务器
server = headers.get('Server', '')
if 'Apache' in server:
fingerprint.append('Apache (可能配置错误)')
elif 'nginx' in server:
fingerprint.append('Nginx')
elif 'IIS' in server:
fingerprint.append('IIS (配置可能有安全问题)')
# 检测编程语言
if 'X-Powered-By' in headers:
fp = headers['X-Powered-By']
fingerprint.append(fp)
# 检测CMS/框架
if 'wordpress' in content.lower():
fingerprint.append('WordPress')
elif 'drupal' in content.lower():
fingerprint.append('Drupal')
elif 'jquery' in content.lower():
fingerprint.append('JQuery')
elif 'laravel' in content.lower() or 'csrf-token' in content.lower():
fingerprint.append('Laravel')
return fingerprint
def detect_cms_and_plugins(self, content, url):
"""检测CMS版本及相关漏洞"""
vulns = []
# WordPress检测
if 'wp-content' in content.lower():
# 检测当前WP版本
if 'wp-generator' in content.lower():
match = re.search(r'WordPress ([0-9.]+)', content)
if match:
version = match.group(1)
major_version = '.'.join(version.split('.')[:2])
# 常见漏洞版本判断
if major_version < '5.2' or (major_version == '5.2' and version.split('.')[-1] < '4'):
vulns.append(Vulnerability(
name="WordPress版本过旧",
severity='medium',
description=f"WordPress {version} 版本可能存在已知漏洞",
location=url,
evidence="通过wp-generator meta标签检测",
remediation="升级WordPress到最新版本",
confidence=0.8
))
# 检测插件
plugin_patterns = {
'wp-content/plugins/contact-form-7:4.x': 'Contact Form 7',
'wp-content/plugins/woocommerce:[0-9.]+': 'WooCommerce',
'wp-content/plugins/jetpack:[0-9.]+': 'Jetpack'
}
for pattern, plugin_name in plugin_patterns.items():
if re.search(pattern, content, re.IGNORECASE):
vulns.append(Vulnerability(
name=f"检测到插件-{plugin_name}",
severity='info',
description=f"网站使用了{plugin_name}插件",
location=url,
evidence="插件路径泄露",
remediation="保持插件更新",
confidence=0.6
))
return vulns
def check_ssl_tls(self):
"""检查SSL/TLS配置"""
vulns = []
parsed = urlparse(self.target_url)
if parsed.scheme == 'https':
try:
# 建立SSL连接
context = ssl.create_default_context()
connection = context.wrap_socket(
socket.socket(socket.AF_INET),
server_hostname=parsed.hostname
)
connection.settimeout(5)
connection.connect((parsed.hostname, 443))
# 获取证书信息
cert = connection.getpeercert()
# 检查证书有效期
import datetime
from datetime import datetime as dt
not_after = cert['notAfter']
cert_expiry = dt.strptime(not_after, '%b %d %H:%M:%S %Y GMT')
days_left = (cert_expiry - dt.now()).days
if days_left < 30:
vulns.append(Vulnerability(
name="SSL证书即将过期",
severity='high',
description=f"SSL证书将在{days_left}天后过期",
location=self.target_url,
evidence=f"证书过期时间: {not_after}",
remediation="尽快更新SSL证书",
confidence=1.0
))
# 检查弱加密协议
connection.close()
except ssl.SSLError as e:
vulns.append(Vulnerability(
name="SSL配置问题",
severity='high',
description=f"SSL连接失败: {str(e)}",
location=self.target_url,
evidence=str(e),
remediation="检查SSL/TLS配置",
confidence=0.8
))
return vulns
def analyze_authentication(self, content, url):
"""分析认证和登录页面"""
vulns = []
# 检查登录页面
if 'login' in content.lower() or 'signin' in content.lower():
vulns.append(Vulnerability(
name="登录页面检测",
severity='info',
description="发现登录页面",
location=url,
evidence="页面包含登录表单",
remediation="确保使用安全的认证机制",
confidence=0.5
))
# 检测无密码强度要求
if 'password' in content.lower() and 'minlength' not in content.lower():
vulns.append(Vulnerability(
name="弱密码策略",
severity='medium',
description="登录表单没有最小密码长度要求",
location=url,
evidence="密码字段缺少minlength属性",
remediation="实施强密码策略",
confidence=0.4
))
# 检查会话管理
if 'session' in content.lower() and 'httponly' not in content.lower():
vulns.append(Vulnerability(
name="Cookie安全设置缺失",
severity='medium',
description="Cookie可能没有设置HttpOnly标志",
location=url,
evidence="没有检测到HttpOnly标志",
remediation="为会话Cookie设置HttpOnly和Secure标志",
confidence=0.6
))
return vulns
def scan(self):
"""主扫描入口"""
print(f"[*] 开始扫描: {self.target_url}")
start_time = time.time()
try:
# 1. 基础信息收集
print("[*] 正在收集基础信息...")
resp = self.session.get(self.target_url, timeout=10)
# 1.1 识别框架/指纹
framework = self.identify_framework(resp.headers, resp.text)
print(f"[+] 识别到框架: {framework if framework else '未知'}")
# 2. 安全头检查
if self.config['check_headers']:
print("[*] 检查安全头...")
h_vulns = HeaderChecker.check(resp.headers, self.target_url)
self.vulnerabilities.extend(h_vulns)
# 3. CMS检测
print("[*] 检查CMS漏洞...")
cms_vulns = self.detect_cms_and_plugins(resp.text, self.target_url)
self.vulnerabilities.extend(cms_vulns)
# 4. SSL/TLS检查
print("[*] 检查SSL/TLS...")
ssl_vulns = self.check_ssl_tls()
self.vulnerabilities.extend(ssl_vulns)
# 5. 认证分析
print("[*] 分析认证机制...")
auth_vulns = self.analyze_authentication(resp.text, self.target_url)
self.vulnerabilities.extend(auth_vulns)
# 6. SQL注入测试
if self.config['check_sqli']:
print("[*] 测试SQL注入...")
# 获取页面参数
params_info = {}
# 从表单中提取参数
form_params = re.findall(r'<input[^>]*name=["\']([^"\']+)["\']', resp.text)
if form_params:
params_info['form'] = values = {
'method': 'POST',
**{p: '' for p in form_params}
}
url_params = re.findall(r'[?&]([^=&]+)=([^&\s]*)', self.target_url)
if url_params:
params_info['url'] = {'method': 'GET'}
for key, val in url_params:
params_info['url'][key] = val
if params_info:
sqli_checker = SQLiChecker(self.session)
for param_location, param_data in params_info.items():
# 简单地测试功能
pass
# 7. 爬取更多URL进行测试
print("[*] 爬取页面链接...")
links = re.findall(r'href=["\']([^"\']+)["\']', resp.text)
for link in links[:5]: # 限制深度
try:
full_url = urljoin(self.target_url, link)
if full_url not in self.crawled_urls:
self.crawled_urls.add(full_url)
page_resp = self.session.get(full_url, timeout=5)
# 在此URL上进行XSS测试
if self.config['check_xss']:
pass # XSS测试逻辑
except:
continue
# 8. 整理报告
print("[*] 生成报告...")
elapsed = time.time() - start_time
report = {
'target': self.target_url,
'scan_time': elapsed,
'vulnerabilities': [v.to_dict() for v in self.vulnerabilities],
'statistics': {
'total': len(self.vulnerabilities),
'critical': len([v for v in self.vulnerabilities if v.severity == 'critical']),
'high': len([v for v in self.vulnerabilities if v.severity == 'high']),
'medium': len([v for v in self.vulnerabilities if v.severity == 'medium']),
'low': len([v for v in self.vulnerabilities if v.severity == 'low'])
}
}
return report
except requests.ConnectionError:
print(f"[错误] 无法连接到 {self.target_url}")
return None
except requests.Timeout:
print(f"[错误] 连接超时")
return None
except Exception as e:
print(f"[错误] 扫描失败: {str(e)}")
return None
# ==================== 结果分析模块 ====================
class ScanResultAnalyzer:
"""扫描结果分析器"""
@staticmethod
def prioritize_vulnerabilities(vulnerabilities):
"""基于CVSS评分和利用难度排序漏洞"""
cvss_map = {
'critical': 10.0,
'high': 8.0,
'medium': 5.0,
'low': 2.0
}
# 按严重程度排序
return sorted(vulnerabilities,
key=lambda v: cvss_map.get(v.severity, 0),
reverse=True)
@staticmethod
def cluster_similar_vulnerabilities(vulnerabilities):
"""将相似漏洞聚类"""
clusters = {}
for vuln in vulnerabilities:
# 基于漏洞类型进行聚类
vuln_type = vuln.name.split('-')[0]
if vuln_type not in clusters:
clusters[vuln_type] = []
clusters[vuln_type].append(vuln)
return clusters
@staticmethod
def generate_executive_summary(report):
"""生成高管级别的总结"""
stats = report['statistics']
risk_level = '低'
if stats['critical'] > 0:
risk_level = '严重'
elif stats['high'] > 0:
risk_level = '高危'
elif stats['medium'] > 0:
risk_level = '中危'
summary = {
'风险等级': risk_level,
'漏洞总数': stats['total'],
'关键发现': [],
'建议行动': []
}
# 提取关键发现
for vuln in report['vulnerabilities']:
if vuln['severity'] in ['critical', 'high']:
summary['关键发现'].append(
f"发现在{vuln['location']}存在{vuln['name']}"
)
summary['建议行动'].append(vuln['remediation'])
return summary
# ==================== 可视化与报告 ====================
class ReportGenerator:
"""报告生成器"""
@staticmethod
def generate_html_report(scan_report):
"""生成HTML报告"""
if not scan_report:
return "<h1>扫描失败</h1>"
html = f"""
<!DOCTYPE html>
<html>
<head>
<title>安全问题汇总报告</title>
<style>
body {{ font-family: Arial; margin: 20px; }}
.vuln {{ border: 1px solid #ddd; padding: 10px; margin: 10px 0; }}
.critical {{ background: #ffcccc; }}
.high {{ background: #ffdddd; }}
.medium {{ background: #ffffcc; }}
.low {{ background: #eeeeee; }}
h1 {{ color: #333; }}
</style>
</head>
<body>
<h1>Web漏洞扫描报告</h1>
<h2>目标: {scan_report['target']}</h2>
<p>扫描耗时: {scan_report['scan_time']:.2f}秒</p>
<h3>统计信息</h3>
<ul>
<li>高危: {scan_report['statistics']['high']}</li>
<li>中危: {scan_report['statistics']['medium']}</li>
<li>低危: {scan_report['statistics']['low']}</li>
</ul>
<h3>漏洞详情</h3>
"""
for vuln in scan_report['vulnerabilities']:
html += f"""
<div class="vuln {vuln['severity']}">
<h4>{vuln['name']}</h4>
<p><strong>严重程度:</strong> {vuln['severity'].upper()}</p>
<p><strong>位置:</strong> {vuln['location']}</p>
<p><strong>描述:</strong> {vuln['description']}</p>
<p><strong>建议:</strong> {vuln['remediation']}</p>
</div>
"""
html += "</body></html>"
return html
# ==================== 主执行模块 ====================
def main():
"""主程序入口"""
target = "https://httpbin.org/"
print("==================================")
print(" 综合漏洞识别与定位工具")
print("==================================")
# 配置
config = {
'check_headers': True,
'check_sqli': True,
'check_xss': True,
'depth': 2,
'max_urls': 10
}
# 初始化扫描器
scanner = WebVulnerabilityScanner(target, config)
# 执行扫描
report = scanner.scan()
if report:
# 分析结果
analyzer = ScanResultAnalyzer()
prioritized = analyzer.prioritize_vulnerabilities(
[type('V', (), v)() for v in report['vulnerabilities']]
)
# 生成报告
report_gen = ReportGenerator()
html = report_gen.generate_html_report(report)
# 输出统计
print(f"\n扫描完成!")
print(f"发现漏洞数量: {report['statistics']['total']}")
print(f"高危漏洞: {report['statistics']['high']}")
print(f"中危漏洞: {report['statistics']['medium']}")
# 详细报告
if report['vulnerabilities']:
print("\n漏洞列表:")
for vuln in report['vulnerabilities']:
print(f" [{vuln['severity'].upper()}] {vuln['name']} - {vuln['location']}")
if vuln['severity'] in ['critical', 'high']:
print(f" 建议: {vuln['remediation']}")
# 保存报告
with open('vulnerability_report.html', 'w', encoding='utf-8') as f:
f.write(html)
print("\nHTML报告已保存为 vulnerability_report.html")
else:
print("扫描失败，无法生成报告")
if __name__ == "__main__":
main()
关键防护策略模块
# security_layer_protection.py
import hashlib
import re
from typing import Dict, List, Optional
class SecurityLayerDefense:
"""多层防护策略"""
def __init__(self):
self.sql_patterns = [
r'(.*)(union.*select|select.*from|insert.*into|drop.*table)',
r'(.*)(--|#|/\*/)',
r"(.*)((')|(\"))"
]
self.xss_patterns = [
r'<script[^>]*>.*?</script>',
r'on(load|error|click|mouseover)[\s]*=',
r'javascript\s*:'
]
def input_validation(self, data: str, type: str = 'string') -> bool:
"""输入验证"""
if type == 'email':
pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
return bool(re.match(pattern, data))
elif type == 'url':
pattern = r'^(https?://)?([\w-]+\.)+[\w-]+(/\S*)?$'
return bool(re.match(pattern, data))
default:
patterns = {
'sql': self.sql_patterns,
'xss': self._build_xss_pattern()
}
for p in patterns.get(type, []):
if re.search(p, data, re.IGNORECASE):
return False
return True
def multi_factor_authentication(self):
"""多因素认证集成"""
# 生成两段验证码
import random
email_code = str(random.randint(100000, 999999))
sms_code = str(random.randint(100000, 999999))
return {
'email': email_code,
'sms': sms_code,
'status': 'pending_verification'
}
def rate_limiting(self, client_id: str, identity: str):
"""防止暴力破解的API限流"""
import time
from collections import defaultdict
if not hasattr(self, '_attempts'):
self._attempts = defaultdict(list)
now = time.time()
window = 60 # 60秒窗口
# 清理旧记录
self._attempts[client_id] = [
t for t in self._attempts[client_id] if now - t < window
]
if len(self._attempts[client_id]) >= 5:
return False # 拒绝请求
self._attempts[client_id].append(now)
return True
这个综合案例展示了：
核心漏洞识别技术：
- 被动检测：不主动发送请求，直接分析响应头、页面内容
- 主动扫描：使用恶意payload探测SQL注入和XSS漏洞
- 指纹识别：识别目标使用的framework、CMS来发现已知漏洞
- 安全头分析：检查HTTP安全头的缺失情况
- 证书验证：检查SSL/TLS配置和证书有效性
常用的漏洞位置定位方法：
- 日志分析脚本：检查错误日志中的SQL异常、XSS特征
- 动态代码审计：用bandit等扫描依赖包安全问题
- 路径重组分析：尝试拼接常见目录（uploads, config等）
希望这个案例对你理解防守漏洞的识别定位有帮助！
