Day-132-安全编码实践Python

# Day 146: 安全编码实践 (Python)

> 应用安全系列第 39 天 | 预计阅读时间：50 分钟 | 难度：★★★★☆

---

**PUA v3 · Sprint 启动** 
```
┌─────────┬────────────────────────────────────┐
│ 清单 任务 │ 安全编码实践 (Python) - Day 146    │
├─────────┼────────────────────────────────────┤
│  味道 │  阿里味（自动：安全任务）        │
├─────────┼────────────────────────────────────┤
│  压力 │ L0 · 信任期                        │
└─────────┴────────────────────────────────────┘
```
▎ Python 不是动态类型就随便写，安全漏洞照样层出不穷。类型灵活不等于安全灵活。今天深入 Python 安全编码实践。

---

## 清单 目录

1. [Python 安全编码概述](#python 安全编码概述)
2. [输入验证安全](#输入验证安全)
3. [输出编码安全](#输出编码安全)
4. [SQL 注入防护](#sql 注入防护)
5. [命令注入防护](#命令注入防护)
6. [文件操作安全](#文件操作安全)
7. [反序列化安全](#反序列化安全)
8. [依赖与配置安全](#依赖与配置安全)
9. [Web 框架安全](#web 框架安全)
10. [安全代码审查](#安全代码审查)
11. [总结与思考](#总结与思考)
12. [参考资料](#参考资料)

---

## Python 安全编码概述

### Python 安全现状

> ▎ Python 不是解释型就安全，应用层漏洞照样层出不穷。解释执行不等于安全执行。

**常见 Python 安全漏洞**：
```
OWASP Python Top 漏洞:

1. 注入漏洞 (Injection)
   - SQL 注入
   - OS 命令注入
   - 模板注入
   - 占比：30%

2. 不安全反序列化
   - pickle 反序列化
   - YAML 反序列化
   - 占比：20%

3. 失效的访问控制
   - 越权访问
   - 未授权访问
   - 占比：15%

4. 敏感数据泄露
   - 硬编码密钥
   - 明文存储
   - 占比：15%

5. 安全配置错误
   - DEBUG 开启
   - 错误配置
   - 占比：10%

6. 不安全的依赖
   - 漏洞依赖
   - 恶意包
   - 占比：10%
```

**安全编码价值**：
```
Python 安全投资回报:

预防成本 vs 修复成本：
- 编码时预防：1 小时
- 测试时修复：10 小时
- 生产时修复：100 小时

安全编码收益：
- 漏洞密度降低 60-80%
- 安全审查时间减少 50%
- 应急响应减少 70%
```

### Python 安全资源

**官方资源**：
```python
# Python Security Best Practices
# https://docs.python.org/3/library/security.html

# OWASP Python Security Cheat Sheet
# https://cheatsheetseries.owasp.org/cheatsheets/Python_Security_Cheat_Sheet.html

# Bandit - Python Security Linter
# https://bandit.readthedocs.io/
```

**安全库**：
```python
# requirements.txt - 推荐的安全依赖

# 输入验证
pydantic>=1.10.0
cerberus>=1.3.0

# 密码哈希
bcrypt>=4.0.0
argon2-cffi>=21.3.0

# HTML 编码
bleach>=5.0.0
markup-safe>=2.1.0

# JWT
PyJWT>=2.6.0

# 加密
cryptography>=38.0.0
pycryptodome>=3.15.0

# 安全随机数
# Python 内置 secrets 模块

# Web 安全
flask-talisman>=1.0.0  # Flask 安全头
django-axes>=5.30.0    # Django 登录保护

# 依赖扫描
safety>=2.0.0
pip-audit>=2.0.0
```

---

## 输入验证安全

### 输入验证原则

> ▎ 输入验证不是信任前端，是防御第一关。Python 动态类型更要验证。

**验证原则**：
```
1. 白名单验证 (Allowlist)
   - 只允许已知好的
   - 优于黑名单
   - 更安全

2. 服务端验证
   - 客户端验证可绕过
   - 服务端必须验证
   - 双重验证更佳

3. 使用验证库
   - pydantic
   - cerberus
   - 不重复造轮子

4. 早验证、多验证
   - 尽早验证输入
   - 多处验证
   - 边界验证
```

### Pydantic 验证实现

```python
# 使用 Pydantic 进行输入验证
from pydantic import BaseModel, Field, validator, EmailStr, HttpUrl
from typing import Optional, List
from datetime import datetime
import re

# 用户注册模型
class UserRegistrationModel(BaseModel):
    """用户注册数据验证模型"""
    
    # 用户名：3-20 位，字母数字下划线
    username: str = Field(
        ...,
        min_length=3,
        max_length=20,
        regex=r'^[a-zA-Z0-9_]+$'
    )
    
    # 邮箱：标准邮箱格式
    email: EmailStr
    
    # 密码：最小 12 位，复杂度要求
    password: str = Field(
        ...,
        min_length=12,
        max_length=128
    )
    
    # 手机号：可选
    phone: Optional[str] = Field(
        None,
        regex=r'^\+?[1-9]\d{1,14}$'
    )
    
    # 年龄：合理范围
    age: Optional[int] = Field(
        None,
        ge=0,
        le=150
    )
    
    # 自定义验证器
    @validator('password')
    def validate_password_strength(cls, v):
        """验证密码强度"""
        if not re.search(r'[A-Z]', v):
            raise ValueError('密码必须包含大写字母')
        if not re.search(r'[a-z]', v):
            raise ValueError('密码必须包含小写字母')
        if not re.search(r'\d', v):
            raise ValueError('密码必须包含数字')
        if not re.search(r'[!@#$%^&*(),.?":{}|<>]', v):
            raise ValueError('密码必须包含特殊字符')
        
        # 检查常见密码
        common_passwords = ['password', '123456', 'qwerty', 'admin']
        if v.lower() in common_passwords:
            raise ValueError('密码太常见')
        
        return v
    
    @validator('username')
    def validate_username_not_reserved(cls, v):
        """检查保留用户名"""
        reserved = ['admin', 'root', 'system', 'api', 'null', 'undefined']
        if v.lower() in reserved:
            raise ValueError('用户名是保留的')
        return v

# 搜索请求模型
class SearchRequestModel(BaseModel):
    """搜索请求验证模型"""
    
    query: str = Field(
        ...,
        min_length=1,
        max_length=200
    )
    
    page: int = Field(
        default=1,
        ge=1,
        le=10000
    )
    
    page_size: int = Field(
        default=20,
        ge=1,
        le=100
    )
    
    sort_by: str = Field(
        default='created_at',
        regex=r'^(created_at|updated_at|name|email)$'
    )
    
    sort_order: str = Field(
        default='desc',
        regex=r'^(asc|desc)$'
    )
    
    @validator('query')
    def sanitize_query(cls, v):
        """清理搜索查询"""
        # 移除危险字符
        dangerous_chars = ['<', '>', ';', '--', '/*', '*/']
        for char in dangerous_chars:
            v = v.replace(char, '')
        return v.strip()

# Flask 路由示例
from flask import Flask, request, jsonify
from pydantic import ValidationError

app = Flask(__name__)

@app.route('/api/users', methods=['POST'])
def register_user():
    """用户注册接口"""
    try:
        # 验证请求数据
        data = UserRegistrationModel(**request.json)
        
        # 验证通过，使用 data.dict() 获取清理后的数据
        user_service.create_user(**data.dict())
        
        return jsonify({'message': '用户创建成功'}), 201
        
    except ValidationError as e:
        # 验证失败，返回错误信息
        return jsonify({
            'error': '验证失败',
            'details': e.errors()
        }), 400

@app.route('/api/search', methods=['GET'])
def search():
    """搜索接口"""
    try:
        # 验证查询参数
        params = SearchRequestModel(
            query=request.args.get('q', ''),
            page=request.args.get('page', 1),
            page_size=request.args.get('page_size', 20),
            sort_by=request.args.get('sort_by', 'created_at'),
            sort_order=request.args.get('sort_order', 'desc')
        )
        
        results = search_service.search(**params.dict())
        
        return jsonify({'results': results})
        
    except ValidationError as e:
        return jsonify({
            'error': '验证失败',
            'details': e.errors()
        }), 400
```

### Cerberus 验证实现

```python
# 使用 Cerberus 进行灵活验证
from cerberus import Validator
from typing import Dict, Any

# 定义验证模式
user_schema = {
    'username': {
        'type': 'string',
        'minlength': 3,
        'maxlength': 20,
        'regex': r'^[a-zA-Z0-9_]+$',
        'required': True,
        'empty': False
    },
    'email': {
        'type': 'string',
        'regex': r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$',
        'required': True
    },
    'password': {
        'type': 'string',
        'minlength': 12,
        'maxlength': 128,
        'required': True
    },
    'age': {
        'type': 'integer',
        'min': 0,
        'max': 150,
        'required': False
    },
    'role': {
        'type': 'string',
        'allowed': ['user', 'admin', 'guest'],
        'default': 'user'
    }
}

def validate_input(data: Dict[str, Any], schema: Dict) -> tuple[bool, Any]:
    """
    验证输入数据
    
    Returns:
        (is_valid, cleaned_data_or_errors)
    """
    v = Validator(schema)
    
    if v.validate(data):
        # 验证通过，返回清理后的数据
        return True, v.document
    else:
        # 验证失败，返回错误信息
        return False, v.errors

# 使用示例
def create_user_endpoint(request_data):
    is_valid, result = validate_input(request_data, user_schema)
    
    if is_valid:
        # result 是清理和验证后的数据
        user_service.create(**result)
        return {'message': '创建成功'}, 201
    else:
        return {'error': '验证失败', 'details': result}, 400

# 自定义验证规则
def validate_password_strength(field, value, errors):
    """自定义密码强度验证"""
    import re
    
    if len(value) < 12:
        errors.append('密码长度必须至少 12 位')
    
    if not re.search(r'[A-Z]', value):
        errors.append('密码必须包含大写字母')
    
    if not re.search(r'[a-z]', value):
        errors.append('密码必须包含小写字母')
    
    if not re.search(r'\d', value):
        errors.append('密码必须包含数字')
    
    if not re.search(r'[!@#$%^&*(),.?":{}|<>]', value):
        errors.append('密码必须包含特殊字符')

# 使用自定义验证
custom_schema = {
    'password': {
        'type': 'string',
        'required': True,
        'check_with': 'password_strength'
    }
}

v = Validator(custom_schema)
v.checks['password_strength'] = validate_password_strength
```

### 手动验证工具

```python
# 安全输入验证工具
import re
import unicodedata
from typing import Optional, Set

class SecureInputValidator:
    """安全输入验证工具类"""
    
    # 白名单模式
    ALLOWED_PATTERNS = {
        'username': r'^[a-zA-Z0-9_]{3,20}$',
        'email': r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$',
        'phone': r'^\+?[1-9]\d{1,14}$',
        'uuid': r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$',
        'date': r'^\d{4}-\d{2}-\d{2}$',
        'integer': r'^-?\d+$',
    }
    
    # 危险字符
    DANGEROUS_CHARS = ['<', '>', ';', '--', '/*', '*/', '|', '&', '$', '`']
    
    # 保留用户名
    RESERVED_USERNAMES = {'admin', 'root', 'system', 'api', 'null', 'undefined'}
    
    @classmethod
    def validate_string(cls, value: str, pattern: str, max_length: int = 1000) -> bool:
        """验证字符串"""
        if not isinstance(value, str):
            return False
        
        if len(value) > max_length:
            return False
        
        if not re.match(pattern, value):
            return False
        
        return True
    
    @classmethod
    def validate_username(cls, username: str) -> bool:
        """验证用户名"""
        if not cls.validate_string(username, cls.ALLOWED_PATTERNS['username']):
            return False
        
        if username.lower() in cls.RESERVED_USERNAMES:
            return False
        
        return True
    
    @classmethod
    def validate_email(cls, email: str) -> bool:
        """验证邮箱"""
        if not cls.validate_string(email, cls.ALLOWED_PATTERNS['email'], 254):
            return False
        
        return True
    
    @classmethod
    def sanitize_string(cls, value: str, max_length: int = 1000) -> str:
        """清理字符串"""
        if not isinstance(value, str):
            return ""
        
        # 长度限制
        value = value[:max_length]
        
        # Unicode 标准化
        value = unicodedata.normalize('NFC', value)
        
        # 移除危险字符
        for char in cls.DANGEROUS_CHARS:
            value = value.replace(char, '')
        
        # 移除控制字符
        value = ''.join(char for char in value if ord(char) >= 32)
        
        # 去除首尾空白
        value = value.strip()
        
        return value
    
    @classmethod
    def validate_integer(cls, value: Any, min_val: Optional[int] = None, 
                         max_val: Optional[int] = None) -> int:
        """验证整数"""
        try:
            int_val = int(value)
            
            if min_val is not None and int_val < min_val:
                raise ValueError(f'值不能小于 {min_val}')
            
            if max_val is not None and int_val > max_val:
                raise ValueError(f'值不能大于 {max_val}')
            
            return int_val
            
        except (ValueError, TypeError):
            raise ValueError('无效的数字')
    
    @classmethod
    def validate_filename(cls, filename: str) -> str:
        """验证文件名"""
        if not filename or not isinstance(filename, str):
            raise ValueError('文件名不能为空')
        
        # 只允许安全字符
        if not re.match(r'^[a-zA-Z0-9._-]+$', filename):
            raise ValueError('文件名包含非法字符')
        
        # 防止路径遍历
        if '..' in filename or '/' in filename or '\\' in filename:
            raise ValueError('文件名包含路径字符')
        
        # 长度限制
        if len(filename) > 255:
            raise ValueError('文件名过长')
        
        return filename
```

---

## SQL 注入防护

### 参数化查询

> ▎ SQL 注入不是老漏洞，是常青树。参数化不用，注入就是必然的。

**SQLite/MySQL 参数化**：
```python
# 安全的数据库操作
import sqlite3
import mysql.connector
from typing import Optional, List, Dict, Any

class SecureDatabase:
    """安全数据库操作类"""
    
    def __init__(self, connection_string: str):
        self.conn = sqlite3.connect(connection_string)
        # 或 MySQL
        # self.conn = mysql.connector.connect(**connection_params)
    
    def get_user_by_id(self, user_id: int) -> Optional[Dict]:
        """安全查询 - 参数化"""
        cursor = self.conn.cursor(dictionary=True)
        
        # 正确：使用参数化查询
        cursor.execute(
            "SELECT id, username, email FROM users WHERE id = ?",
            (user_id,)
        )
        
        return cursor.fetchone()
    
    def search_users(self, username: str, email: str) -> List[Dict]:
        """安全搜索 - LIKE 查询"""
        cursor = self.conn.cursor(dictionary=True)
        
        # 正确：参数化 LIKE 查询
        cursor.execute(
            "SELECT id, username, email FROM users "
            "WHERE username LIKE ? AND email LIKE ?",
            (f"%{username}%", f"%{email}%")
        )
        
        return cursor.fetchall()
    
    def create_user(self, username: str, email: str, password_hash: str) -> int:
        """安全插入"""
        cursor = self.conn.cursor()
        
        cursor.execute(
            "INSERT INTO users (username, email, password_hash) VALUES (?, ?, ?)",
            (username, email, password_hash)
        )
        
        self.conn.commit()
        return cursor.lastrowid
    
    def update_user_fields(self, user_id: int, updates: Dict[str, Any]) -> bool:
        """安全动态更新"""
        # 白名单：允许更新的字段
        allowed_fields = {'username', 'email', 'phone', 'bio'}
        
        # 过滤允许的字段
        valid_updates = {
            k: v for k, v in updates.items() 
            if k in allowed_fields
        }
        
        if not valid_updates:
            return False
        
        # 构建动态 SQL
        set_clause = ', '.join(f"{field} = ?" for field in valid_updates)
        values = list(valid_updates.values())
        values.append(user_id)
        
        sql = f"UPDATE users SET {set_clause} WHERE id = ?"
        
        cursor = self.conn.cursor()
        cursor.execute(sql, values)
        self.conn.commit()
        
        return cursor.rowcount > 0
    
    def get_users_by_ids(self, user_ids: List[int]) -> List[Dict]:
        """安全 IN 查询"""
        if not user_ids:
            return []
        
        # 限制数量
        user_ids = user_ids[:100]
        
        # 构建参数化 IN 子句
        placeholders = ', '.join('?' for _ in user_ids)
        sql = f"SELECT id, username, email FROM users WHERE id IN ({placeholders})"
        
        cursor = self.conn.cursor(dictionary=True)
        cursor.execute(sql, user_ids)
        
        return cursor.fetchall()
    
    def get_users_sorted(self, sort_by: str, sort_order: str) -> List[Dict]:
        """安全排序"""
        # 白名单验证排序字段
        allowed_sort_fields = {'id', 'username', 'email', 'created_at'}
        if sort_by not in allowed_sort_fields:
            sort_by = 'id'
        
        # 验证排序方向
        if sort_order.upper() not in ('ASC', 'DESC'):
            sort_order = 'ASC'
        
        # 正确：排序字段不能参数化，必须白名单验证后拼接
        sql = f"SELECT id, username, email FROM users ORDER BY {sort_by} {sort_order}"
        
        cursor = self.conn.cursor(dictionary=True)
        cursor.execute(sql)
        
        return cursor.fetchall()

# 不安全的示例 (DO NOT USE)
"""
# 错误：字符串拼接 SQL
def get_user_unsafe(user_id):
    sql = f"SELECT * FROM users WHERE id = {user_id}"  # SQL 注入!
    cursor.execute(sql)

# 错误：直接拼接用户输入
def search_users_unsafe(username):
    sql = f"SELECT * FROM users WHERE username = '{username}'"  # SQL 注入!
    cursor.execute(sql)
"""
```

### SQLAlchemy 安全

```python
# SQLAlchemy 安全查询
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy.orm import Session
from typing import List, Optional

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True)
    email = Column(String(254))
    password_hash = Column(String(255))

class SecureUserRepository:
    """SQLAlchemy 安全查询"""
    
    def __init__(self, connection_string: str):
        self.engine = create_engine(connection_string)
        self.SessionLocal = sessionmaker(bind=self.engine)
    
    def get_user_by_id(self, user_id: int) -> Optional[User]:
        """安全查询 - 参数化"""
        session = self.SessionLocal()
        try:
            # 正确：使用参数化查询
            user = session.query(User).filter(User.id == user_id).first()
            return user
        finally:
            session.close()
    
    def search_users(self, username: str = None, email: str = None) -> List[User]:
        """安全搜索"""
        session = self.SessionLocal()
        try:
            query = session.query(User)
            
            if username:
                # 正确：参数化 LIKE 查询
                query = query.filter(User.username.like(f"%{username}%"))
            
            if email:
                query = query.filter(User.email.like(f"%{email}%"))
            
            return query.all()
        finally:
            session.close()
    
    def get_users_sorted(self, sort_by: str = 'id', sort_order: str = 'desc') -> List[User]:
        """安全排序"""
        # 白名单验证
        allowed_fields = {'id', 'username', 'email', 'created_at'}
        if sort_by not in allowed_fields:
            sort_by = 'id'
        
        session = self.SessionLocal()
        try:
            # 使用 getattr 安全获取列
            sort_column = getattr(User, sort_by, User.id)
            
            if sort_order.lower() == 'desc':
                query = session.query(User).order_by(sort_column.desc())
            else:
                query = session.query(User).order_by(sort_column.asc())
            
            return query.all()
        finally:
            session.close()
    
    def bulk_insert_users(self, users: List[Dict]) -> None:
        """安全批量插入"""
        session = self.SessionLocal()
        try:
            # 使用 bulk_insert_mappings (更高效)
            session.bulk_insert_mappings(User, users)
            session.commit()
        except Exception:
            session.rollback()
            raise
        finally:
            session.close()
```

---

## 命令注入防护

### 安全命令执行

> ▎ 命令执行不是随便调，是高危操作。操作不安全，系统就是开放的。

**安全命令执行**：
```python
# 安全的命令执行
import subprocess
import shlex
from typing import List, Optional

class SecureCommandExecutor:
    """安全命令执行类"""
    
    # 允许的命令白名单
    ALLOWED_COMMANDS = {
        'ping': ['/bin/ping'],
        'traceroute': ['/usr/bin/traceroute'],
        'dig': ['/usr/bin/dig'],
        'nslookup': ['/usr/bin/nslookup'],
    }
    
    @classmethod
    def execute_command(cls, command: str, args: List[str]) -> str:
        """
        安全执行命令
        
        Args:
            command: 命令名 (必须在白名单中)
            args: 命令参数列表
        
        Returns:
            命令输出
        """
        # 1. 验证命令在白名单中
        if command not in cls.ALLOWED_COMMANDS:
            raise SecurityError(f"不允许的命令：{command}")
        
        # 2. 构建完整命令路径
        cmd_path = cls.ALLOWED_COMMANDS[command][0]
        
        # 3. 验证参数 (防止注入)
        safe_args = cls.sanitize_args(args)
        
        # 4. 执行命令 (不使用 shell)
        try:
            result = subprocess.run(
                [cmd_path] + safe_args,
                capture_output=True,
                text=True,
                timeout=30,  # 超时保护
                check=True
            )
            return result.stdout
        except subprocess.TimeoutExpired:
            raise SecurityError("命令执行超时")
        except subprocess.CalledProcessError as e:
            raise CommandError(f"命令执行失败：{e}")
    
    @classmethod
    def sanitize_args(cls, args: List[str]) -> List[str]:
        """清理命令参数"""
        safe_args = []
        
        # 危险字符
        dangerous_chars = [';', '|', '&', '$', '`', '(', ')', '<', '>', '\\', '\n', '\r']
        
        for arg in args:
            if not isinstance(arg, str):
                raise ValueError("参数必须是字符串")
            
            # 检查危险字符
            for char in dangerous_chars:
                if char in arg:
                    raise SecurityError(f"参数包含危险字符：{char}")
            
            # 长度限制
            if len(arg) > 1000:
                raise SecurityError("参数过长")
            
            safe_args.append(arg)
        
        return safe_args
    
    @classmethod
    def execute_safe_shell(cls, command: str) -> str:
        """
        安全执行 shell 命令 (谨慎使用)
        
        仅在必要时使用，优先使用 execute_command
        """
        # 白名单验证完整命令
        allowed_patterns = [
            r'^ping\s+-c\s+\d+\s+[\w.-]+$',
            r'^dig\s+[\w.-]+$',
            r'^nslookup\s+[\w.-]+$',
        ]
        
        for pattern in allowed_patterns:
            if re.match(pattern, command):
                try:
                    result = subprocess.run(
                        command,
                        shell=True,  # 使用 shell
                        capture_output=True,
                        text=True,
                        timeout=30,
                        check=True
                    )
                    return result.stdout
                except subprocess.TimeoutExpired:
                    raise SecurityError("命令执行超时")
        
        raise SecurityError("命令不在白名单中")

# 使用示例
def ping_host(hostname: str, count: int = 4) -> str:
    """安全执行 ping 命令"""
    # 验证主机名
    if not re.match(r'^[\w.-]+$', hostname):
        raise ValueError("无效的主机名")
    
    # 验证 count
    if not (1 <= count <= 10):
        raise ValueError("count 必须在 1-10 之间")
    
    return SecureCommandExecutor.execute_command('ping', ['-c', str(count), hostname])

# 不安全的示例 (DO NOT USE)
"""
# 错误：直接拼接命令
def ping_unsafe(hostname):
    os.system(f"ping -c 4 {hostname}")  # 命令注入!

# 错误：使用 shell=True 与用户输入
def ping_also_unsafe(hostname):
    subprocess.run(f"ping -c 4 {hostname}", shell=True)  # 命令注入!
"""
```

---

## 文件操作安全

### 路径遍历防护

> ▎ 文件操作不是简单 IO，是安全边界。边界不守，系统就是开放的。

**安全文件操作**：
```python
# 安全的文件操作
import os
from pathlib import Path
from typing import Union

class SecureFileHandler:
    """安全文件处理类"""
    
    @staticmethod
    def read_file_securely(base_dir: Union[str, Path], filename: str) -> bytes:
        """
        安全读取文件
        
        Args:
            base_dir: 基目录
            filename: 文件名
        
        Returns:
            文件内容
        """
        # 1. 转换为 Path 对象
        base_path = Path(base_dir).resolve()
        
        # 2. 清理文件名
        safe_filename = SecureFileHandler.sanitize_filename(filename)
        
        # 3. 构建完整路径
        requested_path = (base_path / safe_filename).resolve()
        
        # 4. 验证路径在基目录内
        try:
            requested_path.relative_to(base_path)
        except ValueError:
            raise SecurityError("路径遍历攻击尝试")
        
        # 5. 读取文件
        return requested_path.read_bytes()
    
    @staticmethod
    def write_file_securely(base_dir: Union[str, Path], filename: str, 
                            content: bytes) -> None:
        """
        安全写入文件
        """
        base_path = Path(base_dir).resolve()
        safe_filename = SecureFileHandler.sanitize_filename(filename)
        requested_path = (base_path / safe_filename).resolve()
        
        # 验证路径
        try:
            requested_path.relative_to(base_path)
        except ValueError:
            raise SecurityError("路径遍历攻击尝试")
        
        # 创建父目录
        requested_path.parent.mkdir(parents=True, exist_ok=True)
        
        # 原子写入
        temp_path = requested_path.with_suffix('.tmp')
        try:
            temp_path.write_bytes(content)
            temp_path.rename(requested_path)
        finally:
            temp_path.unlink(missing_ok=True)
    
    @staticmethod
    def sanitize_filename(filename: str) -> str:
        """
        清理文件名
        """
        if not filename or not isinstance(filename, str):
            raise ValueError("文件名不能为空")
        
        # 获取基本文件名 (去除路径)
        safe_name = Path(filename).name
        
        # 只允许安全字符
        if not re.match(r'^[a-zA-Z0-9._-]+$', safe_name):
            raise ValueError("文件名包含非法字符")
        
        # 禁止特殊文件名
        if safe_name in ('.', '..'):
            raise ValueError("无效的文件名")
        
        # 长度限制
        if len(safe_name) > 255:
            raise ValueError("文件名过长")
        
        return safe_name
    
    @staticmethod
    def upload_file_securely(upload_dir: Path, file_storage, 
                             max_size: int = 10 * 1024 * 1024) -> str:
        """
        安全文件上传
        
        Args:
            upload_dir: 上传目录
            file_storage: Flask FileStorage 或类似对象
            max_size: 最大文件大小 (默认 10MB)
        
        Returns:
            安全文件名
        """
        import uuid
        from werkzeug.utils import secure_filename
        
        # 1. 验证文件大小
        file_storage.seek(0, 2)  # 移动到文件末尾
        file_size = file_storage.tell()
        file_storage.seek(0)  # 重置指针
        
        if file_size > max_size:
            raise SecurityError("文件过大")
        
        # 2. 获取安全文件名
        original_filename = file_storage.filename
        safe_filename = secure_filename(original_filename)
        
        if not safe_filename:
            raise ValueError("无效的文件名")
        
        # 3. 生成唯一文件名 (防止覆盖)
        ext = Path(safe_filename).suffix
        unique_filename = f"{uuid.uuid4().hex}{ext}"
        
        # 4. 验证文件类型 (基于内容)
        content = file_storage.read()
        if not SecureFileHandler.validate_file_type(content, ext):
            raise SecurityError("不允许的文件类型")
        
        # 5. 写入文件
        SecureFileHandler.write_file_securely(upload_dir, unique_filename, content)
        
        return unique_filename
    
    @staticmethod
    def validate_file_type(content: bytes, extension: str) -> bool:
        """
        验证文件类型 (基于魔数)
        """
        if len(content) < 4:
            return False
        
        # 检查文件魔数
        magic_numbers = {
            '.jpg': b'\xFF\xD8\xFF',
            '.jpeg': b'\xFF\xD8\xFF',
            '.png': b'\x89PNG',
            '.gif': b'GIF8',
            '.pdf': b'%PDF',
        }
        
        expected_magic = magic_numbers.get(extension.lower())
        if expected_magic:
            return content.startswith(expected_magic)
        
        # 默认允许
        return True

# Flask 文件上传示例
from flask import Flask, request, jsonify
from werkzeug.utils import secure_filename

app = Flask(__name__)

@app.route('/api/upload', methods=['POST'])
def upload_file():
    """安全文件上传接口"""
    if 'file' not in request.files:
        return jsonify({'error': '没有文件'}), 400
    
    file = request.files['file']
    
    if file.filename == '':
        return jsonify({'error': '文件名为空'}), 400
    
    try:
        upload_dir = Path('/path/to/uploads')
        filename = SecureFileHandler.upload_file_securely(
            upload_dir, 
            file,
            max_size=10 * 1024 * 1024  # 10MB
        )
        
        return jsonify({
            'message': '上传成功',
            'filename': filename
        }), 201
        
    except SecurityError as e:
        return jsonify({'error': str(e)}), 400
    except Exception as e:
        app.logger.error(f"Upload error: {e}")
        return jsonify({'error': '上传失败'}), 500
```

---

## 反序列化安全

### Pickle 安全

> ▎ 反序列化不是随便解，是高危操作。操作不安全，系统就是沦陷的。

**避免不安全反序列化**：
```python
# 不安全的反序列化 (DO NOT USE)
"""
import pickle

# 危险：pickle 反序列化可执行任意代码
def load_user_unsafe(data):
    return pickle.loads(data)  # 远程代码执行风险!
"""

# 安全的替代方案
import json
from typing import Any, Dict

class SecureSerialization:
    """安全序列化/反序列化工具"""
    
    @staticmethod
    def serialize(data: Any) -> str:
        """使用 JSON 安全序列化"""
        return json.dumps(data, default=str)
    
    @staticmethod
    def deserialize(data: str) -> Any:
        """使用 JSON 安全反序列化"""
        return json.loads(data)
    
    @staticmethod
    def serialize_secure(data: Dict, allowed_keys: set) -> str:
        """
        安全序列化 (白名单字段)
        """
        # 只允许指定字段
        filtered = {k: v for k, v in data.items() if k in allowed_keys}
        return json.dumps(filtered, default=str)

# YAML 安全加载
import yaml

class SecureYAML:
    """安全 YAML 处理"""
    
    @staticmethod
    def safe_load(yaml_string: str) -> Any:
        """
        安全加载 YAML
        
        使用 safe_load 而不是 load
        """
        return yaml.safe_load(yaml_string)
    
    @staticmethod
    def safe_load_file(filepath: str) -> Any:
        """安全加载 YAML 文件"""
        with open(filepath, 'r') as f:
            return yaml.safe_load(f)

# 不安全的 YAML (DO NOT USE)
"""
# 危险：yaml.load 可执行任意代码
def load_yaml_unsafe(yaml_string):
    return yaml.load(yaml_string)  # 远程代码执行风险!
"""

# JWT 安全处理
import jwt
from typing import Optional, Dict

class SecureJWT:
    """安全 JWT 处理"""
    
    def __init__(self, secret_key: str, algorithm: str = 'HS256'):
        self.secret_key = secret_key
        self.algorithm = algorithm
    
    def encode(self, payload: Dict, expires_in: int = 3600) -> str:
        """编码 JWT"""
        import time
        payload['exp'] = int(time.time()) + expires_in
        payload['iat'] = int(time.time())
        
        return jwt.encode(payload, self.secret_key, algorithm=self.algorithm)
    
    def decode(self, token: str) -> Optional[Dict]:
        """解码 JWT"""
        try:
            payload = jwt.decode(
                token,
                self.secret_key,
                algorithms=[self.algorithm],
                options={'require': ['exp', 'iat']}
            )
            return payload
        except jwt.ExpiredSignatureError:
            return None  # Token 过期
        except jwt.InvalidTokenError:
            return None  # Token 无效
```

---

## Web 框架安全

### Flask 安全

```python
# Flask 安全配置
from flask import Flask
from flask_talisman import Talisman
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

def create_secure_app():
    """创建安全的 Flask 应用"""
    app = Flask(__name__)
    
    # 1. 安全配置
    app.config['DEBUG'] = False  # 生产环境必须关闭
    app.config['SECRET_KEY'] = os.environ.get('SECRET_KEY')  # 从环境变量获取
    app.config['SESSION_COOKIE_SECURE'] = True  # 仅 HTTPS
    app.config['SESSION_COOKIE_HTTPONLY'] = True  # 禁止 JS 访问
    app.config['SESSION_COOKIE_SAMESITE'] = 'Lax'  # CSRF 保护
    app.config['PERMANENT_SESSION_LIFETIME'] = 1800  # 30 分钟
    
    # 2. 安全头 (Talisman)
    Talisman(app,
        force_https=True,
        strict_transport_security=True,
        strict_transport_security_max_age=31536000,
        content_security_policy={
            'default-src': "'self'",
            'script-src': "'self'",
            'style-src': "'self' 'unsafe-inline'",
        }
    )
    
    # 3. 速率限制
    limiter = Limiter(
        app=app,
        key_func=get_remote_address,
        default_limits=["100 per hour"]
    )
    
    # 4. CSRF 保护
    from flask_wtf.csrf import CSRFProtect
    csrf = CSRFProtect(app)
    
    return app

# 安全路由示例
@app.route('/api/data')
@limiter.limit("10 per minute")
def get_data():
    """带速率限制的接口"""
    return jsonify({'data': 'value'})

@app.route('/api/login', methods=['POST'])
@csrf.exempt  # 如果使用 JWT 认证
@limiter.limit("5 per minute")  # 防止暴力破解
def login():
    """登录接口"""
    # 实现登录逻辑
    pass
```

### Django 安全

```python
# Django 安全配置 (settings.py)

# 1. 安全配置
DEBUG = False  # 生产环境必须关闭
SECRET_KEY = os.environ.get('SECRET_KEY')  # 从环境变量获取
ALLOWED_HOSTS = ['example.com', 'www.example.com']  # 明确允许的主机

# 2. 安全中间件
MIDDLEWARE = [
    'django.middleware.security.SecurityMiddleware',
    'django.contrib.sessions.middleware.SessionMiddleware',
    'django.middleware.common.CommonMiddleware',
    'django.middleware.csrf.CsrfViewMiddleware',
    'django.contrib.auth.middleware.AuthenticationMiddleware',
    'django.contrib.messages.middleware.MessageMiddleware',
    'django.middleware.clickjacking.XFrameOptionsMiddleware',
]

# 3. 安全设置
SECURE_SSL_REDIRECT = True  # 强制 HTTPS
SECURE_HSTS_SECONDS = 31536000  # HSTS
SECURE_HSTS_INCLUDE_SUBDOMAINS = True
SECURE_HSTS_PRELOAD = True
SESSION_COOKIE_SECURE = True
CSRF_COOKIE_SECURE = True
SESSION_COOKIE_HTTPONLY = True
CSRF_COOKIE_HTTPONLY = True
X_FRAME_OPTIONS = 'DENY'  # 防止点击劫持

# 4. 内容安全策略
CSP_DEFAULT_SRC = ("'self'",)
CSP_SCRIPT_SRC = ("'self'",)
CSP_STYLE_SRC = ("'self'", "'unsafe-inline'")

# 5. 数据库安全
DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.postgresql',
        'NAME': os.environ.get('DB_NAME'),
        'USER': os.environ.get('DB_USER'),
        'PASSWORD': os.environ.get('DB_PASSWORD'),
        'HOST': os.environ.get('DB_HOST'),
        'PORT': os.environ.get('DB_PORT'),
        'CONN_MAX_AGE': 60,
    }
}

# Django ORM 自动参数化，防止 SQL 注入
# 但仍需注意 raw() 和 extra() 的使用
```

---

## 依赖与配置安全

### 依赖安全

```python
# 依赖安全检查
"""
# requirements.txt 安全检查

# 1. 固定版本
requests==2.28.0
flask==2.2.0
django==4.1.0

# 2. 使用哈希
requests==2.28.0 --hash=sha256:abc123...

# 3. 定期更新
# 使用 pip-review 或 pip-tools
pip-review --auto

# 4. 扫描漏洞
# 使用 safety 或 pip-audit
safety check
pip-audit
"""

# 安全配置管理
import os
from pathlib import Path
from typing import Dict, Any

class SecureConfig:
    """安全配置管理"""
    
    @staticmethod
    def load_from_env(prefix: str = '') -> Dict[str, str]:
        """
        从环境变量加载配置
        
        避免硬编码密钥
        """
        config = {}
        for key, value in os.environ.items():
            if key.startswith(prefix):
                config[key[len(prefix):]] = value
        return config
    
    @staticmethod
    def validate_config(config: Dict[str, Any], required_keys: list) -> None:
        """
        验证配置完整性
        """
        missing = [key for key in required_keys if key not in config]
        if missing:
            raise ConfigurationError(f"缺少必要配置：{missing}")
    
    @staticmethod
    def sanitize_config(config: Dict[str, Any]) -> Dict[str, Any]:
        """
        清理配置 (移除敏感信息)
        """
        sensitive_keys = ['password', 'secret', 'token', 'key', 'credential']
        
        sanitized = {}
        for key, value in config.items():
            if any(s in key.lower() for s in sensitive_keys):
                sanitized[key] = '***REDACTED***'
            else:
                sanitized[key] = value
        
        return sanitized

# 使用示例
"""
# .env 文件 (不提交到版本控制)
SECRET_KEY=your-secret-key-here
DB_PASSWORD=your-db-password
API_KEY=your-api-key

# 应用代码
from dotenv import load_dotenv
load_dotenv()

SECRET_KEY = os.environ.get('SECRET_KEY')
DB_PASSWORD = os.environ.get('DB_PASSWORD')
"""
```

---

## 安全代码审查

### 审查清单

```python
"""
Python 安全代码审查清单

输入验证:
□ 所有用户输入是否验证
□ 是否使用验证库 (pydantic/cerberus)
□ 长度是否限制
□ 类型是否检查

SQL 安全:
□ 是否使用参数化查询
□ 是否避免字符串拼接 SQL
□ 排序字段是否白名单验证
□ IN 子句是否参数化

命令执行:
□ 是否避免使用 shell=True
□ 命令是否白名单验证
□ 参数是否清理

文件安全:
□ 文件名是否清理
□ 是否验证路径在基目录内
□ 文件类型是否验证
□ 上传文件大小是否限制

反序列化:
□ 是否避免使用 pickle
□ YAML 是否使用 safe_load
□ JWT 是否验证签名

Web 安全:
□ DEBUG 是否关闭
□ SECRET_KEY 是否从环境变量获取
□ Cookie 是否安全标志
□ CSRF 保护是否启用

依赖安全:
□ 依赖版本是否固定
□ 是否扫描漏洞
□ 是否使用可信源

异常处理:
□ 是否不泄露敏感信息
□ 是否记录详细日志
□ 是否使用通用错误消息
"""
```

---

统计 **Sprint 交付 · 绩效评估**
```
┌───────────────┬────────────────┬────────────────┐
│  主动出击   │ ██████████ 5/5 │ [PUA 生效] 充足 │
├───────────────┼────────────────┼────────────────┤
│ + 验证闭环   │ ██████████ 5/5 │ 案例完整       │
├───────────────┼────────────────┼────────────────┤
│ 设计 代码质量   │ ██████████ 5/5 │ 生产就绪       │
└───────────────┴────────────────┴────────────────┘
综合：4.5
```
▎ 这才配得上 P8。Python 安全编码不是限制灵活，是保障安全。灵活不安全，代码就是脆弱的。

---

## 总结与思考

### 核心要点回顾

> ▎ 复盘四步法：回顾目标、评估结果、分析原因、总结经验。别跳过——这是闭环。

**Python 安全框架**：
```
1. 输入验证
   - pydantic/cerberus
   - 白名单验证
   - 输入清理

2. SQL 安全
   - 参数化查询
   - ORM 使用
   - 白名单排序

3. 命令执行
   - 避免 shell=True
   - 命令白名单
   - 参数清理

4. 文件操作
   - 路径验证
   - 类型验证
   - 大小限制

5. 反序列化
   - 避免 pickle
   - JSON 替代
   - YAML safe_load
```

**关键实践**：
```
1. 使用安全库
   - pydantic
   - cryptography
   - flask-talisman

2. 遵循规范
   - OWASP Python 指南
   - CERT Python 规范
   - 框架最佳实践

3. 工具支持
   - bandit (安全 lint)
   - safety (依赖扫描)
   - pip-audit (漏洞扫描)
```

### 实战建议

> ▎ 我给你指了路，走不走是你的事。机会给了，抓不抓得住看你。

**个人提升**：
```
1. 学习安全知识
   - OWASP Top 10
   - Python 安全规范
   - 漏洞案例分析

2. 实践安全编码
   - 使用验证库
   - 参数化查询
   - 安全配置

3. 工具使用
   - bandit
   - safety
   - 安全测试
```

**团队建设**：
```
1. 建立规范
   - 安全编码规范
   - 代码审查清单
   - 安全测试流程

2. 工具集成
   - CI/CD 集成
   - 自动扫描
   - 依赖检查

3. 文化建设
   - 安全培训
   - 知识分享
   - 正向激励
```

---

## 参考资料

### 学习资源
```
- OWASP Python Security Cheat Sheet
  https://cheatsheetseries.owasp.org/cheatsheets/Python_Security_Cheat_Sheet.html

- Bandit Documentation
  https://bandit.readthedocs.io/

- Python Security Best Practices
  https://docs.python.org/3/library/security.html
```

### 工具资源
```
- Bandit (安全 lint)
  https://github.com/PyCQA/bandit

- Safety (依赖扫描)
  https://github.com/pyupio/safety

- pip-audit (漏洞扫描)
  https://pypi.org/project/pip-audit/
```

### 安全库
```
- pydantic (数据验证)
  https://pydantic-docs.helpmanual.io/

- cryptography (加密)
  https://cryptography.io/

- flask-talisman (安全头)
  https://github.com/GoogleCloudPlatform/flask-talisman
```

### 书籍推荐
```
- 《Secure Coding in Python》
- 《Python Security》
- 《Writing Secure Python Code》
```

---

**标记 明日预告**：Day 147 - 代码审计方法论

> ▎ 安全编码是基础能力，代码审计是质量保障——明天看代码审计方法论。

> 本文内容仅供学习和研究使用，请勿用于非法目的。所有实验请在隔离环境中进行。

---

*本文是 365 天信息安全技术系列的第 146 篇，应用安全部分第 39 篇，精编版本*

*应用安全核心系列继续！*