引言 网络安全是数字化时代的核心挑战。随着攻击手段的不断演进，了解常见漏洞、攻击方法和防御策略变得至关重要。本文将深入分析网络安全领域的核心主题，帮助读者构建更安全的系统。
一、OWASP Top 10深度解析 1.1 注入漏洞 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 # ========== SQL注入 ========== # 不安全的代码示例 def get_user_by_id(user_id): """危险：直接拼接SQL""" query = f"SELECT * FROM users WHERE id = {user_id}" return db.execute(query) # 攻击示例 # user_id = "1 OR 1=1" # 实际执行的SQL: SELECT * FROM users WHERE id = 1 OR 1=1 # 结果：返回所有用户数据 # 安全的代码示例 import sqlite3 from typing import Optional class UserRepository: """安全的用户仓库""" def __init__(self, db_path: str): self.conn = sqlite3.connect(db_path) def get_user_by_id(self, user_id: int) -> Optional[dict]: """使用参数化查询""" cursor = self.conn.cursor() # 使用?占位符，数据库驱动会正确转义参数 query = "SELECT * FROM users WHERE id = ?" cursor.execute(query, (user_id,)) row = cursor.fetchone() if row: return { 'id': row[0], 'username': row[1], 'email': row[2] } return None def search_users(self, keyword: str) -> list: """安全的搜索功能""" cursor = self.conn.cursor() query = """ SELECT id, username, email FROM users WHERE username LIKE ? OR email LIKE ? """ # 使用通配符进行模糊搜索 pattern = f"%{keyword}%" cursor.execute(query, (pattern, pattern)) return cursor.fetchall() # ========== SQL注入检测与防御 ========== class SQLInjectionDetector: """SQL注入检测器""" # 常见SQL注入特征 INJECTION_PATTERNS = [ r"(\%27)|(\')", # 单引号 r"(\-\-)|(#)", # 注释符 r"(\bor\b|\band\b).*?=.*?", # 逻辑运算符 r"(\bunion\b.*\bselect\b)", # UNION查询 r"(\bselect\b.*\bfrom\b)", # SELECT查询 r"(\bexec\b|\bexecute\b)", # 执行命令 r"(;|\bxp_cmdshell\b)", # 命令分隔符和存储过程 r"(\bdrop\b|\bdelete\b|\btruncate\b)", # 危险操作 ] @classmethod def detect_injection(cls, input_string: str) -> bool: """检测SQL注入""" import re for pattern in cls.INJECTION_PATTERNS: if re.search(pattern, input_string, re.IGNORECASE): return True return False @classmethod def sanitize_input(cls, input_string: str) -> str: """清理输入""" import re # 移除危险字符 sanitized = re.sub(r"[\'\"\;\-\-]", "", input_string) # 限制长度 max_length = 100 sanitized = sanitized[:max_length] return sanitized # ========== ORM安全使用 ========== from sqlalchemy import create_engine, Column, Integer, String from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker Base = declarative_base() class User(Base): """用户模型""" __tablename__ = 'users' id = Column(Integer, primary_key=True) username = Column(String(50), nullable=False) email = Column(String(100), nullable=False) class SecureUserRepository: """使用ORM的安全仓库""" def __init__(self, database_url: str): self.engine = create_engine(database_url) Session = sessionmaker(bind=self.engine) self.session = Session() def get_user_by_id(self, user_id: int) -> Optional[User]: """ORM自动处理参数化""" return self.session.query(User).filter( User.id == user_id ).first() def search_users(self, keyword: str) -> list: """安全的搜索""" return self.session.query(User).filter( (User.username.like(f"%{keyword}%")) | (User.email.like(f"%{keyword}%")) ).all() 1.2 跨站脚本攻击(XSS) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 # ========== XSS攻击类型 ========== """ 1. 反射型XSS (Reflected XSS) - 恶意代码通过URL参数反射 - 例如: http://example.com/search?q=<script>alert('XSS')</script> 2. 存储型XSS (Stored XSS) - 恶意代码存储在服务器，每次访问时执行 - 例如: 评论、用户资料等可存储的内容 3. DOM型XSS (DOM-based XSS) - 恶意代码通过DOM操作执行 - 例如: #<img src=x onerror=alert('XSS')> """ # ========== XSS防御 ========== import html import re from typing import Dict, Any class XSSProtection: """XSS防护""" @staticmethod def escape_html(text: str) -> str: """HTML转义""" return html.escape(text, quote=True) @staticmethod def escape_js(text: str) -> str: """JavaScript字符串转义""" escape_map = { '\\': '\\\\', '"': '\\"', "'": "\\'", '\n': '\\n', '\r': '\\r', '\t': '\\t', '\b': '\\b', '\f': '\\f', } for char, escaped in escape_map.items(): text = text.replace(char, escaped) return text @staticmethod def sanitize_html(content: str, allowed_tags: list = None) -> str: """HTML净化，只允许安全的标签""" if allowed_tags is None: allowed_tags = ['p', 'br', 'strong', 'em', 'u', 'a'] import bleach return bleach.clean( content, tags=allowed_tags, attributes={ 'a': ['href', 'title'], '*': ['class'] }, strip=True ) @staticmethod def validate_url(url: str) -> bool: """验证URL安全性""" import urllib.parse try: parsed = urllib.parse.urlparse(url) # 检查协议 if parsed.scheme not in ['http', 'https']: return False # 检查是否为JavaScript伪协议 if parsed.scheme.lower() == 'javascript': return False return True except Exception: return False # Web框架集成示例 from flask import Flask, request, render_template_string app = Flask(__name__) # 不安全的示例 @app.route('/unsafe') def unsafe_search(): query = request.args.get('q', '') # 危险：直接渲染用户输入 return render_template_string(f'<p>搜索结果: {query}</p>') # 安全的示例 @app.route('/safe') def safe_search(): query = request.args.get('q', '') # 转义用户输入 safe_query = XSSProtection.escape_html(query) # 或使用模板引擎的自动转义 return render_template_string( '<p>搜索结果: {{ query }}</p>', query=safe_query ) # Content Security Policy (CSP) @app.after_request def add_security_headers(response): """添加安全头""" csp = ( "default-src 'self'; " "script-src 'self' 'unsafe-inline' https://cdn.example.com; " "style-src 'self' 'unsafe-inline'; " "img-src 'self' data: https:; " "font-src 'self' data:; " "connect-src 'self'; " "frame-ancestors 'none';" ) response.headers['Content-Security-Policy'] = csp response.headers['X-Content-Type-Options'] = 'nosniff' response.headers['X-Frame-Options'] = 'DENY' response.headers['X-XSS-Protection'] = '1; mode=block' return response # ========== 输入验证 ========== class InputValidator: """输入验证器""" @staticmethod def validate_username(username: str) -> bool: """验证用户名""" # 长度检查 if len(username) < 3 or len(username) > 20: return False # 字符检查：只允许字母、数字、下划线 import re pattern = r'^[a-zA-Z0-9_]+$' return bool(re.match(pattern, username)) @staticmethod def validate_email(email: str) -> bool: """验证邮箱""" import re pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$' return bool(re.match(pattern, email)) @staticmethod def validate_phone(phone: str) -> bool: """验证手机号""" import re pattern = r'^1[3-9]\d{9}$' return bool(re.match(pattern, phone)) 1.3 CSRF防护 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 # ========== CSRF攻击原理 ========== """ CSRF (Cross-Site Request Forgery) 跨站请求伪造 攻击场景： 1. 用户登录了银行网站 bank.com 2. 用户访问了恶意网站 evil.com 3. evil.com 包含指向 bank.com 的请求 <img src="http://bank.com/transfer?to=attacker&amount=10000"> 4. 浏览器自动携带 bank.com 的 cookie 5. 银行网站处理转账请求 防御措施： 1. CSRF Token 2. SameSite Cookie属性 3. 验证Referer/Origin头 4. 双重提交Cookie """ # ========== CSRF防护实现 ========== import secrets from flask import Flask, request, session, render_template app = Flask(__name__) app.secret_key = secrets.token_hex(32) class CSRFProtection: """CSRF防护""" @staticmethod def generate_token() -> str: """生成CSRF Token""" return secrets.token_hex(32) @staticmethod def validate_token(token: str, stored_token: str) -> bool: """验证Token""" return secrets.compare_digest(token, stored_token) # 在Flask中使用CSRF保护 @app.before_request def csrf_protect(): """CSRF保护中间件""" # 豁免安全的方法 if request.method in ['GET', 'HEAD', 'OPTIONS', 'TRACE']: return # 检查CSRF Token token = request.headers.get('X-CSRF-Token') or request.form.get('csrf_token') if not token or token != session.get('csrf_token'): return 'Invalid CSRF token', 403 @app.route('/form') def show_form(): """显示表单""" # 生成并存储CSRF Token csrf_token = CSRFProtection.generate_token() session['csrf_token'] = csrf_token return f''' <form method="POST" action="/submit"> <input type="hidden" name="csrf_token" value="{csrf_token}"> <input type="text" name="username"> <button type="submit">提交</button> </form> ''' @app.route('/submit', methods=['POST']) def handle_submit(): """处理表单提交""" # Token已在中间件验证 username = request.form.get('username') # 业务逻辑 return f'Hello, {username}!' # ========== SameSite Cookie ========== @app.route('/login', methods=['POST']) def login(): """登录""" # 设置SameSite属性 response = app.make_response('Login successful') # Strict: 严格模式，只在同站点请求中发送 response.set_cookie( 'session_id', value='session_data', httponly=True, secure=True, samesite='Strict' ) return response # ========== 双重Cookie验证 ========== class DoubleSubmitCookie: """双重Cookie CSRF防护""" @staticmethod def generate_token(): """生成Token并设置到Cookie""" token = secrets.token_hex(32) @app.after_this_request def add_cookie(response): response.set_cookie( 'csrf_token', token, httponly=True, secure=True, samesite='Strict' ) return response return token @staticmethod def validate(request): """验证Token""" cookie_token = request.cookies.get('csrf_token') request_token = request.form.get('csrf_token') or \ request.headers.get('X-CSRF-Token') return secrets.compare_digest(cookie_token, request_token) 二、其他常见Web漏洞 2.1 文件上传漏洞 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 # ========== 文件上传安全 ========== import os import magic from werkzeug.utils import secure_filename from typing import Tuple class SecureFileUpload: """安全的文件上传""" # 允许的文件类型 ALLOWED_EXTENSIONS = { 'image': {'jpg', 'jpeg', 'png', 'gif', 'webp'}, 'document': {'pdf', 'doc', 'docx', 'txt'}, 'media': {'mp4', 'mp3', 'wav'} } # 允许的MIME类型 ALLOWED_MIME_TYPES = { 'image/jpeg', 'image/png', 'image/gif', 'image/webp', 'application/pdf', 'video/mp4', 'audio/mpeg' } # 最大文件大小 (10MB) MAX_FILE_SIZE = 10 * 1024 * 1024 def __init__(self, upload_dir: str): self.upload_dir = upload_dir os.makedirs(upload_dir, exist_ok=True) def validate_file(self, file) -> Tuple[bool, str]: """验证文件""" # 检查文件名 filename = secure_filename(file.filename) if not filename: return False, "Invalid filename" # 检查文件扩展名 ext = filename.rsplit('.', 1)[1].lower() if '.' in filename else '' if ext not in set().union(*self.ALLOWED_EXTENSIONS.values()): return False, f"File extension '{ext}' not allowed" # 检查文件大小 file.seek(0, os.SEEK_END) file_size = file.tell() file.seek(0) if file_size > self.MAX_FILE_SIZE: return False, f"File size exceeds {self.MAX_FILE_SIZE} bytes" # 检查MIME类型 file_content = file.read() mime_type = magic.from_buffer(file_content, mime=True) if mime_type not in self.ALLOWED_MIME_TYPES: return False, f"MIME type '{mime_type}' not allowed" # 验证扩展名和MIME类型是否匹配 ext_mime_map = { 'jpg': 'image/jpeg', 'jpeg': 'image/jpeg', 'png': 'image/png', 'gif': 'image/gif', 'pdf': 'application/pdf' } if ext in ext_mime_map and ext_mime_map[ext] != mime_type: return False, "File extension does not match content" return True, "Valid file" def save_file(self, file) -> Tuple[bool, str]: """保存文件""" # 验证文件 is_valid, message = self.validate_file(file) if not is_valid: return False, message # 生成安全的文件名 import uuid original_filename = secure_filename(file.filename) ext = original_filename.rsplit('.', 1)[1].lower() safe_filename = f"{uuid.uuid4().hex}.{ext}" # 保存文件 filepath = os.path.join(self.upload_dir, safe_filename) file.seek(0) file.save(filepath) # 设置文件权限 os.chmod(filepath, 0o644) return True, safe_filename # Flask路由示例 from flask import Flask, request, jsonify app = Flask(__name__) uploader = SecureFileUpload('/var/uploads') @app.route('/upload', methods=['POST']) def upload_file(): """文件上传接口""" if 'file' not in request.files: return jsonify({'error': 'No file provided'}), 400 file = request.files['file'] if file.filename == '': return jsonify({'error': 'Empty filename'}), 400 # 保存文件 success, message = uploader.save_file(file) if success: return jsonify({ 'success': True, 'filename': message }) else: return jsonify({ 'success': False, 'error': message }), 400 2.2 权限绕过 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 # ========== 权限控制 ========== from functools import wraps from flask import session, request, jsonify class PermissionChecker: """权限检查器""" # 权限定义 PERMISSIONS = { 'user:read': '查看用户信息', 'user:write': '修改用户信息', 'user:delete': '删除用户', 'admin:panel': '访问管理面板', 'system:config': '修改系统配置' } # 角色权限映射 ROLE_PERMISSIONS = { 'guest': ['user:read'], 'user': ['user:read', 'user:write'], 'moderator': ['user:read', 'user:write', 'user:delete'], 'admin': ['admin:panel', 'system:config'], 'superadmin': ['*'] # 所有权限 } @classmethod def has_permission(cls, role: str, required_permission: str) -> bool: """检查角色是否有权限""" if role not in cls.ROLE_PERMISSIONS: return False permissions = cls.ROLE_PERMISSIONS[role] # 超级管理员拥有所有权限 if '*' in permissions: return True return required_permission in permissions # 装饰器实现权限控制 def require_permission(permission: str): """权限检查装饰器""" def decorator(f): @wraps(f) def decorated_function(*args, **kwargs): # 从session获取用户角色 role = session.get('role', 'guest') # 检查权限 if not PermissionChecker.has_permission(role, permission): return jsonify({ 'error': 'Permission denied', 'required': permission }), 403 return f(*args, **kwargs) return decorated_function return decorator # 资源所有者检查 def require_owner_or_admin(model_name: str, id_param: str = 'id'): """资源所有者或管理员权限检查""" def decorator(f): @wraps(f) def decorated_function(*args, **kwargs): user_id = session.get('user_id') role = session.get('role', 'guest') # 管理员跳过检查 if role == 'superadmin': return f(*args, **kwargs) # 获取资源ID resource_id = kwargs.get(id_param) or request.view_args.get(id_param) # 检查资源所有者 from database import get_db db = get_db() resource = db.query(model_name).get(resource_id) if resource and resource.owner_id == user_id: return f(*args, **kwargs) return jsonify({'error': 'Resource access denied'}), 403 return decorated_function return decorator # 使用示例 @app.route('/users/<int:id>', methods=['GET']) @require_permission('user:read') def get_user(id): """获取用户信息""" return jsonify({'user': f'User {id}'}) @app.route('/users/<int:id>', methods=['PUT']) @require_permission('user:write') @require_owner_or_admin('User', 'id') def update_user(id): """更新用户信息""" return jsonify({'success': True}) @app.route('/users/<int:id>', methods=['DELETE']) @require_permission('user:delete') def delete_user(id): """删除用户""" return jsonify({'success': True}) # ========== 水平越权防护 ========== class AccessControl: """访问控制""" @staticmethod def check_resource_access(user_id: int, resource_id: int, resource_type: str) -> bool: """检查资源访问权限""" from database import get_db db = get_db() # 检查资源是否存在 resource = db.query(resource_type).get(resource_id) if not resource: return False # 检查资源所有者 if resource.owner_id == user_id: return True # 检查共享权限 if hasattr(resource, 'shared_with'): if user_id in resource.shared_with: return True return False @staticmethod def check_id_or_id_list(user_input, max_length: int = 1000) -> bool: """检查ID参数，防止ID遍历攻击""" # 单个ID if isinstance(user_input, int): return 0 < user_input < max_length # ID列表 if isinstance(user_input, list): if len(user_input) > 100: # 限制列表长度 return False return all( isinstance(id, int) and 0 < id < max_length for id in user_input ) return False 三、渗透测试与漏洞挖掘 3.1 信息收集 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 # ========== 信息收集工具 ========== import requests import dns.resolver import socket from typing import List, Dict, Set from urllib.parse import urlparse import re class InformationGathering: """信息收集""" def __init__(self, target: str): self.target = target self.results = { 'subdomains': set(), 'open_ports': [], 'technologies': [], 'sensitive_files': [], 'emails': set(), 'social_accounts': set() } def enumerate_subdomains(self) -> Set[str]: """子域名枚举""" wordlist = [ 'www', 'mail', 'ftp', 'admin', 'blog', 'api', 'dev', 'staging', 'test', 'app', 'mobile' ] for subdomain in wordlist: domain = f"{subdomain}.{self.target}" try: # DNS查询 dns.resolver.resolve(domain, 'A') self.results['subdomains'].add(domain) except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer): continue return self.results['subdomains'] def scan_ports(self, ports: List[int] = None) -> List[int]: """端口扫描""" if ports is None: ports = [21, 22, 23, 25, 53, 80, 110, 443, 445, 3306, 3389, 8080] open_ports = [] for port in ports: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(1) try: sock.connect((self.target, port)) open_ports.append(port) except socket.error: pass finally: sock.close() self.results['open_ports'] = open_ports return open_ports def identify_technologies(self) -> List[str]: """技术栈识别""" url = f"http://{self.target}" try: response = requests.get(url, timeout=10) # 检查响应头 headers = response.headers technologies = [] # Server头 if 'Server' in headers: technologies.append(f"Server: {headers['Server']}") # X-Powered-By头 if 'X-Powered-By' in headers: technologies.append(f"PoweredBy: {headers['X-Powered-By']}") # 检查HTML中的技术特征 content = response.text # WordPress if 'wp-content' in content: technologies.append("WordPress") # jQuery if 'jquery' in content.lower(): technologies.append("jQuery") # React if 'react' in content.lower(): technologies.append("React") # Vue if 'vue' in content.lower(): technologies.append("Vue") self.results['technologies'] = technologies except requests.RequestException: pass return self.results['technologies'] def find_sensitive_files(self) -> List[str]: """敏感文件扫描""" sensitive_paths = [ '/robots.txt', '/.git/config', '/.env', '/wp-config.php', '/web.config', '/.htaccess', '/admin', '/phpmyadmin', '/backup.zip', '/database.sql' ] found_files = [] for path in sensitive_paths: url = f"http://{self.target}{path}" try: response = requests.get(url, timeout=5) if response.status_code in [200, 301, 302]: found_files.append(path) except requests.RequestException: continue self.results['sensitive_files'] = found_files return found_files def extract_emails(self) -> Set[str]: """提取邮箱地址""" url = f"http://{self.target}" try: response = requests.get(url, timeout=10) content = response.text # 邮箱正则 email_pattern = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b' emails = set(re.findall(email_pattern, content, re.IGNORECASE)) self.results['emails'] = emails except requests.RequestException: pass return self.results['emails'] def generate_report(self) -> Dict: """生成报告""" return { 'target': self.target, 'subdomains': list(self.results['subdomains']), 'open_ports': self.results['open_ports'], 'technologies': self.results['technologies'], 'sensitive_files': self.results['sensitive_files'], 'emails': list(self.results['emails']), 'social_accounts': list(self.results['social_accounts']) } 3.2 漏洞扫描 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 # ========== 漏洞扫描器 ========== import requests from typing import Dict, List import re class VulnerabilityScanner: """漏洞扫描器""" def __init__(self, target: str): self.target = target self.vulnerabilities = [] def scan_sql_injection(self, urls: List[str]) -> List[Dict]: """SQL注入扫描""" payloads = [ "'", "' OR '1'='1", "' OR '1'='1'--", "admin'--", "1' ORDER BY 1--", "1' UNION SELECT NULL--" ] found = [] for url in urls: for payload in payloads: try: # 测试GET参数 if '?' in url: test_url = f"{url}{payload}" response = requests.get(test_url, timeout=5) # 检查SQL错误特征 sql_errors = [ "You have an error in your SQL syntax", "Warning: mysql_fetch_array()", "ORA-01756: quoted string not properly terminated", "Unclosed quotation mark after the character string" ] if any(error in response.text for error in sql_errors): found.append({ 'type': 'SQL Injection', 'url': url, 'payload': payload, 'severity': 'High' }) except requests.RequestException: continue self.vulnerabilities.extend(found) return found def scan_xss(self, urls: List[str]) -> List[Dict]: """XSS扫描""" payloads = [ "<script>alert('XSS')</script>", "<img src=x onerror=alert('XSS')>", "<svg onload=alert('XSS')>", "javascript:alert('XSS')", "'><script>alert('XSS')</script>" ] found = [] for url in urls: for payload in payloads: try: # URL编码payload import urllib.parse encoded_payload = urllib.parse.quote(payload) test_url = f"{url}{encoded_payload}" response = requests.get(test_url, timeout=5) # 检查payload是否被反射 if payload in response.text or encoded_payload in response.text: found.append({ 'type': 'XSS', 'url': url, 'payload': payload, 'severity': 'High' }) except requests.RequestException: continue self.vulnerabilities.extend(found) return found def check_security_headers(self, url: str) -> Dict: """安全头检查""" try: response = requests.get(url, timeout=10) headers = response.headers missing_headers = [] # 检查必要的安全头 security_headers = { 'X-Frame-Options': '防止点击劫持', 'X-Content-Type-Options': '防止MIME类型嗅探', 'X-XSS-Protection': 'XSS过滤', 'Content-Security-Policy': '内容安全策略', 'Strict-Transport-Security': '强制HTTPS' } for header, description in security_headers.items(): if header not in headers: missing_headers.append({ 'header': header, 'description': description }) return { 'url': url, 'missing_headers': missing_headers, 'severity': 'Medium' if missing_headers else 'Low' } except requests.RequestException: return {'error': 'Failed to check headers'} def scan_directory_traversal(self, base_url: str) -> List[Dict]: """目录遍历扫描""" payloads = [ '../../../etc/passwd', '..\\..\\..\\windows\\win.ini', '....//....//....//etc/passwd', '%2e%2e%2f%2e%2e%2f%2e%2e%2fetc%2fpasswd' ] found = [] # 常见的可测试端点 test_paths = [ '/download', '/file', '/image', '/document' ] for path in test_paths: for payload in payloads: try: test_url = f"{base_url}{path}?file={payload}" response = requests.get(test_url, timeout=5) # 检查响应内容 if 'root:' in response.text or '[extensions]' in response.text: found.append({ 'type': 'Directory Traversal', 'url': test_url, 'payload': payload, 'severity': 'High' }) except requests.RequestException: continue self.vulnerabilities.extend(found) return found def generate_report(self) -> Dict: """生成扫描报告""" # 统计漏洞 vuln_count = { 'Critical': 0, 'High': 0, 'Medium': 0, 'Low': 0 } for vuln in self.vulnerabilities: severity = vuln.get('severity', 'Low') vuln_count[severity] = vuln_count.get(severity, 0) + 1 return { 'target': self.target, 'total_vulnerabilities': len(self.vulnerabilities), 'by_severity': vuln_count, 'vulnerabilities': self.vulnerabilities } 四、安全编码最佳实践 4.1 密码安全 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 # ========== 密码哈希与验证 ========== import bcrypt import secrets import hashlib from typing import Tuple class PasswordSecurity: """密码安全""" @staticmethod def hash_password(password: str) -> str: """哈希密码""" # 生成盐值 salt = bcrypt.gensalt(rounds=12) # 哈希密码 hashed = bcrypt.hashpw(password.encode('utf-8'), salt) return hashed.decode('utf-8') @staticmethod def verify_password(password: str, hashed: str) -> bool: """验证密码""" try: return bcrypt.checkpw( password.encode('utf-8'), hashed.encode('utf-8') ) except Exception: return False @staticmethod def validate_password_strength(password: str) -> Tuple[bool, list]: """验证密码强度""" errors = [] # 长度检查 if len(password) < 8: errors.append("密码长度至少8位") if len(password) > 128: errors.append("密码长度不超过128位") # 复杂度检查 has_upper = any(c.isupper() for c in password) has_lower = any(c.islower() for c in password) has_digit = any(c.isdigit() for c in password) has_special = any(c in "!@#$%^&*()_+-=[]{}|;:,.<>?" for c in password) if not (has_upper and has_lower): errors.append("密码必须包含大小写字母") if not has_digit: errors.append("密码必须包含数字") if not has_special: errors.append("密码必须包含特殊字符") # 常见弱密码检查 common_passwords = [ 'password', '12345678', 'qwerty', 'abc123', 'admin', 'welcome', 'password123' ] if password.lower() in common_passwords: errors.append("密码过于常见") return len(errors) == 0, errors @staticmethod def generate_reset_token() -> str: """生成密码重置令牌""" return secrets.token_urlsafe(32) @staticmethod def hash_reset_token(token: str) -> str: """哈希重置令牌用于存储""" return hashlib.sha256(token.encode()).hexdigest() # ========== 密码策略 ========== class PasswordPolicy: """密码策略""" def __init__( self, min_length: int = 8, max_length: int = 128, require_uppercase: bool = True, require_lowercase: bool = True, require_digit: bool = True, require_special: bool = True, expire_days: int = 90, prevent_reuse: int = 5 ): self.min_length = min_length self.max_length = max_length self.require_uppercase = require_uppercase self.require_lowercase = require_lowercase self.require_digit = require_digit self.require_special = require_special self.expire_days = expire_days self.prevent_reuse = prevent_reuse def validate(self, password: str, user_history: list = None) -> Tuple[bool, list]: """验证密码""" errors = [] # 长度 if len(password) < self.min_length: errors.append(f"密码长度至少{self.min_length}位") if len(password) > self.max_length: errors.append(f"密码长度不超过{self.max_length}位") # 复杂度 if self.require_uppercase and not any(c.isupper() for c in password): errors.append("密码必须包含大写字母") if self.require_lowercase and not any(c.islower() for c in password): errors.append("密码必须包含小写字母") if self.require_digit and not any(c.isdigit() for c in password): errors.append("密码必须包含数字") if self.require_special and not any( c in "!@#$%^&*()_+-=[]{}|;:,.<>?" for c in password ): errors.append("密码必须包含特殊字符") # 历史密码检查 if user_history and self.prevent_reuse: recent_passwords = user_history[:self.prevent_reuse] for old_hashed in recent_passwords: if PasswordSecurity.verify_password(password, old_hashed): errors.append(f"不能使用最近{self.prevent_reuse}次使用过的密码") break return len(errors) == 0, errors 4.2 敏感数据保护 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 # ========== 敏感数据加密 ========== from cryptography.fernet import Fernet from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2 import os import base64 from typing import Any import json class DataEncryption: """数据加密""" @staticmethod def generate_key() -> bytes: """生成加密密钥""" return Fernet.generate_key() @staticmethod def encrypt_data(data: str, key: bytes) -> str: """加密数据""" f = Fernet(key) encrypted = f.encrypt(data.encode()) return base64.b64encode(encrypted).decode() @staticmethod def decrypt_data(encrypted_data: str, key: bytes) -> str: """解密数据""" f = Fernet(key) encrypted = base64.b64decode(encrypted_data.encode()) decrypted = f.decrypt(encrypted) return decrypted.decode() @staticmethod def encrypt_dict(data: dict, key: bytes) -> str: """加密字典数据""" json_str = json.dumps(data) return DataEncryption.encrypt_data(json_str, key) @staticmethod def decrypt_dict(encrypted_data: str, key: bytes) -> dict: """解密字典数据""" json_str = DataEncryption.decrypt_data(encrypted_data, key) return json.loads(json_str) class SecureStorage: """安全存储""" def __init__(self, encryption_key: bytes): self.encryption_key = encryption_key def store_sensitive_data(self, key: str, data: Any): """存储敏感数据""" # 序列化数据 if isinstance(data, dict): encrypted = DataEncryption.encrypt_dict(data, self.encryption_key) else: encrypted = DataEncryption.encrypt_data(str(data), self.encryption_key) # 存储到数据库或文件 # 这里简化为内存存储 self.storage[key] = encrypted def retrieve_sensitive_data(self, key: str) -> Any: """检索敏感数据""" encrypted = self.storage.get(key) if not encrypted: return None # 尝试解密为字典 try: return DataEncryption.decrypt_dict(encrypted, self.encryption_key) except: return DataEncryption.decrypt_data(encrypted, self.encryption_key) # ========== 日志脱敏 ========== import re class LogSanitizer: """日志脱敏""" # 敏感信息正则模式 PATTERNS = { 'email': r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', 'phone': r'\b1[3-9]\d{9}\b', 'id_card': r'\b\d{17}[\dXx]\b', 'credit_card': r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b', 'password': r'(password|pwd|passwd)\s*[=:]\s*\S+', 'token': r'(token|api_key|secret)\s*[=:]\s*\S+', 'ip': r'\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b' } @classmethod def sanitize(cls, log_message: str) -> str: """脱敏日志""" sanitized = log_message for pattern_type, pattern in cls.PATTERNS.items(): if pattern_type == 'email': sanitized = re.sub( pattern, lambda m: f"{m.group(0)[0]}***@{m.group(0).split('@')[1]}", sanitized ) elif pattern_type == 'phone': sanitized = re.sub( pattern, lambda m: f"{m.group(0)[:3]}****{m.group(0)[7:]}", sanitized ) elif pattern_type == 'id_card': sanitized = re.sub( pattern, lambda m: f"{m.group(0)[:6]}********{m.group(0)[14:]}", sanitized ) elif pattern_type in ['password', 'token', 'credit_card']: sanitized = re.sub( pattern, lambda m: f"{m.group(0).split('=')[0].split(':')[0]}=***", sanitized ) elif pattern_type == 'ip': sanitized = re.sub( pattern, lambda m: f"{m.group(0).rsplit('.', 1)[0]}.*", sanitized ) return sanitized @classmethod def sanitize_dict(cls, data: dict) -> dict: """脱敏字典数据""" sanitized = {} for key, value in data.items(): if isinstance(value, str): sanitized[key] = cls.sanitize(value) elif isinstance(value, dict): sanitized[key] = cls.sanitize_dict(value) else: sanitized[key] = value return sanitized 总结 网络安全是一个持续的过程，需要：
...