网络安全与漏洞深度分析：从攻击到防御的完整指南

  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
  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
  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)

  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

  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

  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'])
        }

  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
        }

  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

  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