Day-126-安全编码规范

# Day 140: 安全编码规范

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

---

**PUA v3 · Sprint 启动** 
```
┌─────────┬────────────────────────────────────┐
│ 清单 任务 │ 安全编码规范 - Day 140             │
├─────────┼────────────────────────────────────┤
│  味道 │  阿里味（自动：安全任务）        │
├─────────┼────────────────────────────────────┤
│  压力 │ L0 · 信任期                        │
└─────────┴────────────────────────────────────┘
```
▎ 安全编码不是代码写完再检查，是写的时候就想安全。写的时候不想，写完再改就是返工。今天系统学习安全编码规范。

---

## 清单 目录

1. [安全编码概述](#安全编码概述)
2. [输入处理安全](#输入处理安全)
3. [输出编码安全](#输出编码安全)
4. [认证与会话安全](#认证与会话安全)
5. [数据保护安全](#数据保护安全)
6. [错误处理安全](#错误处理安全)
7. [密码学安全](#密码学安全)
8. [文件操作安全](#文件操作安全)
9. [并发安全](#并发安全)
10. [代码审查清单](#代码审查清单)
11. [总结与思考](#总结与思考)
12. [参考资料](#参考资料)

---

## 安全编码概述

### 为什么需要安全编码

> ▎ 安全编码不是增加工作量，是减少返工。现在不花时间，将来就要花更多时间。

**安全编码价值**：
```
漏洞修复成本 (NIST 数据):
- 需求阶段发现：$1
- 设计阶段发现：$5
- 编码阶段发现：$10
- 测试阶段发现：$20
- 生产环境发现：$200+

安全编码收益：
- 减少漏洞：70%+
- 降低修复成本：90%+
- 提升代码质量：50%+
- 减少安全审计时间：60%+
```

**常见编码漏洞**：
```
OWASP Top 10 编码相关：
1. 注入 (Injection): 35%
2. 失效的认证：15%
3. 敏感数据泄露：12%
4. XXE: 8%
5. 失效的访问控制：20%
6. 安全配置错误：10%

根源分析：
- 缺乏安全意识：40%
- 没有时间检查：25%
- 不知道最佳实践：20%
- 工具支持不足：15%
```

### 安全编码原则

**核心原则**：
```
1. 最小权限原则 (Least Privilege)
   - 只给必要的权限
   - 及时回收权限
   - 默认拒绝

2. 纵深防御原则 (Defense in Depth)
   - 多层防护
   - 不依赖单一防护
   - 故障安全

3. 安全默认原则 (Secure by Default)
   - 默认配置安全
   - 用户需要主动降低安全
   - 安全是默认选项

4. 完全验证原则 (Complete Mediation)
   - 每次访问都验证
   - 不缓存授权结果
   - 持续验证

5. 开放设计原则 (Open Design)
   - 安全不依赖隐蔽
   - 代码可审查
   - 算法公开
```

---

## 输入处理安全

### 输入验证

> ▎ 输入验证不是可选项，是必选项。输入不验证，漏洞就是必然的。

**验证原则**：
```
1. 白名单验证 (允许已知好的)
   - 优于黑名单
   - 更安全可靠
   - 推荐首选

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

3. 类型、长度、范围
   - 检查数据类型
   - 检查长度限制
   - 检查取值范围

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

**输入验证实现**：
```python
# 安全的输入验证
import re
from typing import Optional
from datetime import datetime

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+$',
        'positive_integer': r'^\d+$'
    }
    
    # 危险字符
    DANGEROUS_CHARS = ['<', '>', ';', '--', '/*', '*/', '|', '&', '$', '`']
    
    @classmethod
    def validate_username(cls, username: str) -> bool:
        """验证用户名"""
        if not username or not isinstance(username, str):
            return False
        
        # 白名单验证
        if not re.match(cls.ALLOWED_PATTERNS['username'], username):
            return False
        
        # 检查保留用户名
        reserved = ['admin', 'root', 'system', 'api', 'null', 'undefined']
        if username.lower() in reserved:
            return False
        
        return True
    
    @classmethod
    def validate_email(cls, email: str) -> bool:
        """验证邮箱"""
        if not email or not isinstance(email, str):
            return False
        
        # 长度限制
        if len(email) > 254:
            return False
        
        # 白名单验证
        if not re.match(cls.ALLOWED_PATTERNS['email'], email):
            return False
        
        return True
    
    @classmethod
    def sanitize_string(cls, input: str, max_length: int = 1000) -> str:
        """清理字符串"""
        if not input or not isinstance(input, str):
            return ""
        
        # 长度限制
        input = input[:max_length]
        
        # 移除危险字符
        for char in cls.DANGEROUS_CHARS:
            input = input.replace(char, '')
        
        # 移除控制字符
        input = ''.join(char for char in input if ord(char) >= 32)
        
        # 去除首尾空白
        input = input.strip()
        
        return input
    
    @classmethod
    def validate_integer(cls, value, min_val: Optional[int] = None, 
                         max_val: Optional[int] = None) -> bool:
        """验证整数"""
        try:
            int_val = int(value)
            
            if min_val is not None and int_val < min_val:
                return False
            
            if max_val is not None and int_val > max_val:
                return False
            
            return True
        except (ValueError, TypeError):
            return False
    
    @classmethod
    def validate_file_path(cls, path: str, base_dir: str) -> bool:
        """验证文件路径 (防止路径遍历)"""
        import os
        
        # 规范化路径
        abs_path = os.path.abspath(path)
        abs_base = os.path.abspath(base_dir)
        
        # 检查是否在基目录内
        if not abs_path.startswith(abs_base):
            return False
        
        # 检查是否包含危险字符
        if '..' in path or '\x00' in path:
            return False
        
        return True
```

### SQL 注入防护

```python
# 安全的数据库操作
from typing import List, Dict, Any
import psycopg2
from psycopg2 import sql, extras

class SecureDatabase:
    """安全数据库操作类"""
    
    def __init__(self, connection_string: str):
        self.conn = psycopg2.connect(connection_string)
    
    def get_user_by_id(self, user_id: int) -> Optional[Dict]:
        """安全查询 - 参数化"""
        cursor = self.conn.cursor(cursor_factory=extras.RealDictCursor)
        
        # 正确：使用参数化查询
        cursor.execute(
            "SELECT id, username, email FROM users WHERE id = %s",
            (user_id,)
        )
        
        return cursor.fetchone()
    
    def search_users(self, search_term: str, limit: int = 10) -> List[Dict]:
        """安全搜索 - LIKE 查询"""
        cursor = self.conn.cursor(cursor_factory=extras.RealDictCursor)
        
        # 正确：参数化 LIKE 查询
        cursor.execute(
            "SELECT id, username, email FROM users WHERE username LIKE %s LIMIT %s",
            (f"%{search_term}%", limit)
        )
        
        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 (%s, %s, %s) RETURNING id",
            (username, email, password_hash)
        )
        
        self.conn.commit()
        return cursor.fetchone()[0]
    
    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 = []
        values = []
        
        for i, (field, value) in enumerate(valid_updates.items()):
            set_clause.append(sql.SQL("{} = %s").format(sql.Identifier(field)))
            values.append(value)
        
        values.append(user_id)
        
        query = sql.SQL("UPDATE users SET {} WHERE id = %s").format(
            sql.SQL(', ').join(set_clause)
        )
        
        cursor = self.conn.cursor()
        cursor.execute(query, 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]
        
        cursor = self.conn.cursor(cursor_factory=extras.RealDictCursor)
        
        # 正确：使用 ANY 或 IN 参数化
        cursor.execute(
            "SELECT id, username, email FROM users WHERE id = ANY(%s)",
            (user_ids,)
        )
        
        return cursor.fetchall()
```

---

## 输出编码安全

### XSS 防护

> ▎ 输出编码不是可选项，是必选项。输出不编码，XSS 就是必然的。

**输出编码原则**：
```
1. 上下文感知编码
   - HTML 上下文：HTML 实体编码
   - 属性上下文：属性编码
   - JavaScript 上下文：JS 编码
   - URL 上下文：URL 编码

2. 输出时编码
   - 存储时不编码
   - 输出时编码
   - 避免双重编码

3. 使用安全库
   - 不自己实现
   - 使用成熟库
   - 定期更新
```

**输出编码实现**：
```python
# 安全的输出编码
import html
import json
import urllib.parse
import re

class SecureOutputEncoder:
    """安全输出编码器"""
    
    @staticmethod
    def encode_html(text: str) -> str:
        """HTML 实体编码"""
        if text is None:
            return ""
        
        return html.escape(str(text), quote=True)
    
    @staticmethod
    def encode_html_attribute(text: str) -> str:
        """HTML 属性编码"""
        if text is None:
            return ""
        
        # HTML 实体编码 + 额外处理
        encoded = html.escape(str(text), quote=True)
        
        # 编码其他危险字符
        encoded = encoded.replace('`', '&#96;')
        encoded = encoded.replace('=', '&#61;')
        
        return encoded
    
    @staticmethod
    def encode_javascript(text: str) -> str:
        """JavaScript 编码"""
        if text is None:
            return ""
        
        # Unicode 转义
        result = []
        for char in str(text):
            if char in '"\'<>\\&':
                result.append(f'\\u{ord(char):04x}')
            elif ord(char) < 32 or ord(char) > 126:
                result.append(f'\\u{ord(char):04x}')
            else:
                result.append(char)
        
        return ''.join(result)
    
    @staticmethod
    def encode_url(text: str) -> str:
        """URL 编码"""
        if text is None:
            return ""
        
        return urllib.parse.quote(str(text), safe='')
    
    @staticmethod
    def encode_css(text: str) -> str:
        """CSS 编码"""
        if text is None:
            return ""
        
        result = []
        for char in str(text):
            if char in '"\'<>\\&(){}[]':
                result.append(f'\\{ord(char):x} ')
            elif ord(char) < 32 or ord(char) > 126:
                result.append(f'\\{ord(char):x} ')
            else:
                result.append(char)
        
        return ''.join(result)

# 安全模板渲染
from jinja2 import Environment, select_autoescape

def create_safe_template_env():
    """创建安全的模板环境"""
    env = Environment(
        autoescape=select_autoescape(['html', 'xml']),
        trim_blocks=True,
        lstrip_blocks=True
    )
    
    # 注册安全过滤器
    env.filters['escape'] = SecureOutputEncoder.encode_html
    env.filters['e'] = SecureOutputEncoder.encode_html
    env.filters['url_encode'] = SecureOutputEncoder.encode_url
    env.filters['js_encode'] = SecureOutputEncoder.encode_javascript
    
    return env

# 使用示例
"""
安全模板使用：

<p>{{ user_input }}</p>

<p>{{ user_input | e }}</p>

<p>{{ user_input | safe }}</p>

<a href="/search?q={{ search_term | url_encode }}">Search</a>

<script>
  var searchTerm = "{{ search_term | js_encode }}";
</script>
"""
```

---

## 认证与会话安全

### 密码安全

> ▎ 密码不是明文存储，是哈希存储。哈希不安全，密码就是裸奔的。

**密码处理最佳实践**：
```python
# 安全的密码处理
import hashlib
import secrets
import hmac
from typing import Optional

class SecurePasswordHandler:
    """安全密码处理器"""
    
    # 推荐算法
    ALGORITHM = 'argon2id'  # 或 bcrypt, scrypt
    
    # 参数配置
    ARGON2_TIME_COST = 3      # 迭代次数
    ARGON2_MEMORY_COST = 65536  # 内存 (KB)
    ARGON2_PARALLELISM = 4    # 并行度
    ARGON2_HASH_LEN = 32      # 输出长度
    ARGON2_SALT_LEN = 16      # 盐长度
    
    def __init__(self):
        try:
            import argon2
            self.ph = argon2.PasswordHasher(
                time_cost=self.ARGON2_TIME_COST,
                memory_cost=self.ARGON2_MEMORY_COST,
                parallelism=self.ARGON2_PARALLELISM,
                hash_len=self.ARGON2_HASH_LEN,
                salt_len=self.ARGON2_SALT_LEN
            )
            self.use_argon2 = True
        except ImportError:
            self.use_argon2 = False
    
    def hash_password(self, password: str) -> str:
        """哈希密码"""
        if self.use_argon2:
            return self.ph.hash(password)
        else:
            # 降级方案：bcrypt
            import bcrypt
            salt = bcrypt.gensalt(rounds=12)
            return bcrypt.hashpw(password.encode(), salt).decode()
    
    def verify_password(self, password: str, hash: str) -> bool:
        """验证密码"""
        try:
            if self.use_argon2:
                self.ph.verify(hash, password)
                return True
            else:
                import bcrypt
                return bcrypt.checkpw(password.encode(), hash.encode())
        except Exception:
            return False
    
    def needs_rehash(self, hash: str) -> bool:
        """检查是否需要重新哈希"""
        if self.use_argon2:
            return self.ph.check_needs_rehash(hash)
        return False

# 密码策略
class PasswordPolicy:
    """密码策略"""
    
    MIN_LENGTH = 12
    MAX_LENGTH = 128
    REQUIRE_UPPERCASE = True
    REQUIRE_LOWERCASE = True
    REQUIRE_DIGIT = True
    REQUIRE_SPECIAL = True
    
    # 常见密码黑名单
    COMMON_PASSWORDS = [
        'password', '123456', 'qwerty', 'admin', 'letmein',
        'welcome', 'monkey', 'dragon', 'master', 'login'
    ]
    
    @classmethod
    def validate(cls, password: str) -> tuple[bool, list]:
        """验证密码强度"""
        errors = []
        
        # 长度检查
        if len(password) < cls.MIN_LENGTH:
            errors.append(f'Password must be at least {cls.MIN_LENGTH} characters')
        
        if len(password) > cls.MAX_LENGTH:
            errors.append(f'Password must be at most {cls.MAX_LENGTH} characters')
        
        # 字符类型检查
        if cls.REQUIRE_UPPERCASE and not re.search(r'[A-Z]', password):
            errors.append('Password must contain uppercase letter')
        
        if cls.REQUIRE_LOWERCASE and not re.search(r'[a-z]', password):
            errors.append('Password must contain lowercase letter')
        
        if cls.REQUIRE_DIGIT and not re.search(r'[0-9]', password):
            errors.append('Password must contain number')
        
        if cls.REQUIRE_SPECIAL and not re.search(r'[!@#$%^&*(),.?":{}|<>]', password):
            errors.append('Password must contain special character')
        
        # 常见密码检查
        if password.lower() in cls.COMMON_PASSWORDS:
            errors.append('Password is too common')
        
        # 连续字符检查
        if cls._has_sequential_chars(password):
            errors.append('Password contains sequential characters')
        
        # 重复字符检查
        if cls._has_repeated_chars(password):
            errors.append('Password contains too many repeated characters')
        
        return len(errors) == 0, errors
    
    @staticmethod
    def _has_sequential_chars(password: str, max_seq: int = 3) -> bool:
        """检查连续字符"""
        # 检查字母序列
        lower = password.lower()
        for i in range(len(lower) - max_seq):
            seq = lower[i:i+max_seq]
            if seq.isalpha() and seq == ''.join(sorted(seq)):
                return True
        
        # 检查数字序列
        for i in range(len(password) - max_seq):
            seq = password[i:i+max_seq]
            if seq.isdigit():
                nums = [int(c) for c in seq]
                if nums == list(range(nums[0], nums[0] + len(nums))):
                    return True
        
        return False
    
    @staticmethod
    def _has_repeated_chars(password: str, max_repeat: int = 3) -> bool:
        """检查重复字符"""
        for i in range(len(password) - max_repeat):
            if len(set(password[i:i+max_repeat])) == 1:
                return True
        return False
```

### 会话安全

```python
# 安全的会话管理
import secrets
import time
from typing import Optional, Dict
from datetime import datetime, timedelta

class SecureSessionManager:
    """安全会话管理器"""
    
    def __init__(self, redis_client):
        self.redis = redis_client
        self.session_timeout = 3600  # 1 小时
        self.absolute_timeout = 86400  # 24 小时
    
    def create_session(self, user_id: str) -> str:
        """创建安全会话"""
        # 生成安全会话 ID
        session_id = self._generate_session_id()
        
        # 会话数据
        session_data = {
            'user_id': user_id,
            'created_at': time.time(),
            'last_activity': time.time(),
            'ip_address': self._get_client_ip(),
            'user_agent': self._get_user_agent(),
            'absolute_expiry': time.time() + self.absolute_timeout
        }
        
        # 存储会话
        self._store_session(session_id, session_data)
        
        return session_id
    
    def validate_session(self, session_id: str) -> Optional[Dict]:
        """验证会话"""
        session_data = self._get_session(session_id)
        
        if not session_data:
            return None
        
        # 检查绝对过期
        if time.time() > session_data.get('absolute_expiry', 0):
            self.destroy_session(session_id)
            return None
        
        # 检查空闲超时
        if time.time() - session_data.get('last_activity', 0) > self.session_timeout:
            self.destroy_session(session_id)
            return None
        
        # 可选：检查 IP/User-Agent 变更
        # if session_data.get('ip_address') != self._get_client_ip():
        #     self.destroy_session(session_id)
        #     return None
        
        # 更新最后活动时间
        session_data['last_activity'] = time.time()
        self._store_session(session_id, session_data)
        
        return session_data
    
    def destroy_session(self, session_id: str):
        """销毁会话"""
        self.redis.delete(f"session:{session_id}")
    
    def regenerate_session(self, old_session_id: str) -> str:
        """会话 ID 轮换 (防止会话固定)"""
        old_data = self._get_session(old_session_id)
        
        if old_data:
            # 创建新会话
            new_session_id = self.create_session(old_data['user_id'])
            
            # 销毁旧会话
            self.destroy_session(old_session_id)
            
            return new_session_id
        
        return self.create_session('anonymous')
    
    def _generate_session_id(self) -> str:
        """生成安全会话 ID"""
        return secrets.token_urlsafe(32)
    
    def _store_session(self, session_id: str, data: Dict):
        """存储会话"""
        key = f"session:{session_id}"
        self.redis.hset(key, mapping=data)
        self.redis.expire(key, self.absolute_timeout)
    
    def _get_session(self, session_id: str) -> Optional[Dict]:
        """获取会话"""
        key = f"session:{session_id}"
        data = self.redis.hgetall(key)
        return data if data else None
    
    def _get_client_ip(self) -> str:
        """获取客户端 IP"""
        # 实现获取客户端 IP
        return "127.0.0.1"
    
    def _get_user_agent(self) -> str:
        """获取 User-Agent"""
        # 实现获取 User-Agent
        return "Unknown"
```

---

## 数据保护安全

### 加密最佳实践

> ▎ 加密不是可选项，是敏感数据的必选项。敏感数据不加密，泄露就是时间问题。

**加密实现**：
```python
# 安全的加密实现
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
from cryptography.hazmat.backends import default_backend
import secrets
import base64
import os

class SecureEncryption:
    """安全加密类"""
    
    # AES-GCM 参数
    AES_KEY_SIZE = 32  # 256 bits
    NONCE_SIZE = 12    # 96 bits
    TAG_SIZE = 16      # 128 bits
    
    def __init__(self, key: Optional[bytes] = None):
        if key is None:
            self.key = secrets.token_bytes(self.AES_KEY_SIZE)
        else:
            self.key = key
    
    def encrypt(self, plaintext: bytes, associated_data: bytes = b'') -> bytes:
        """AES-GCM 加密"""
        # 生成随机 nonce
        nonce = secrets.token_bytes(self.NONCE_SIZE)
        
        # 创建 AES-GCM
        aesgcm = AESGCM(self.key)
        
        # 加密
        ciphertext = aesgcm.encrypt(nonce, plaintext, associated_data)
        
        # 返回 nonce + ciphertext
        return nonce + ciphertext
    
    def decrypt(self, ciphertext: bytes, associated_data: bytes = b'') -> bytes:
        """AES-GCM 解密"""
        # 提取 nonce
        nonce = ciphertext[:self.NONCE_SIZE]
        actual_ciphertext = ciphertext[self.NONCE_SIZE:]
        
        # 创建 AES-GCM
        aesgcm = AESGCM(self.key)
        
        # 解密
        plaintext = aesgcm.decrypt(nonce, actual_ciphertext, associated_data)
        
        return plaintext
    
    @staticmethod
    def derive_key(password: str, salt: bytes, iterations: int = 100000) -> bytes:
        """从密码派生密钥"""
        kdf = PBKDF2HMAC(
            algorithm=hashes.SHA256(),
            length=32,
            salt=salt,
            iterations=iterations,
            backend=default_backend()
        )
        return kdf.derive(password.encode())

# 敏感数据加密存储
class SecureDataStorage:
    """安全数据存储"""
    
    def __init__(self, encryption_key: bytes):
        self.encryption = SecureEncryption(encryption_key)
    
    def store_sensitive(self, data: str, context: str = '') -> str:
        """存储敏感数据"""
        # 加密
        encrypted = self.encryption.encrypt(
            data.encode(),
            associated_data=context.encode()
        )
        
        # Base64 编码便于存储
        return base64.b64encode(encrypted).decode()
    
    def retrieve_sensitive(self, encrypted_data: str, context: str = '') -> str:
        """检索敏感数据"""
        # Base64 解码
        encrypted = base64.b64decode(encrypted_data)
        
        # 解密
        decrypted = self.encryption.decrypt(
            encrypted,
            associated_data=context.encode()
        )
        
        return decrypted.decode()

# 字段级加密示例
"""
数据库表设计：

CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    username VARCHAR(50),
    email_encrypted TEXT,      -- 加密的邮箱
    ssn_encrypted TEXT,        -- 加密的社保号
    email_hash VARCHAR(64),    -- 用于搜索的哈希
    created_at TIMESTAMP
);

应用层加密：
- 写入时加密
- 读取时解密
- 哈希用于搜索
"""
```

---

## 错误处理安全

### 安全错误处理

> ▎ 错误信息不是调试信息，是安全边界。边界不守，信息就是泄露的。

**错误处理原则**：
```
1. 不泄露敏感信息
   - 不显示堆栈跟踪
   - 不显示 SQL 语句
   - 不显示内部路径

2. 统一错误格式
   - 标准化错误响应
   - 错误代码
   - 用户友好消息

3. 记录详细日志
   - 内部记录详细信息
   - 外部显示通用信息
   - 便于调查
```

**错误处理实现**：
```python
# 安全的错误处理
import logging
import traceback
from typing import Optional
from datetime import datetime

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('app.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

class SecureErrorHandler:
    """安全错误处理器"""
    
    # 错误代码映射
    ERROR_CODES = {
        'VALIDATION_ERROR': 400,
        'AUTHENTICATION_ERROR': 401,
        'AUTHORIZATION_ERROR': 403,
        'NOT_FOUND': 404,
        'CONFLICT': 409,
        'RATE_LIMIT': 429,
        'INTERNAL_ERROR': 500
    }
    
    # 用户友好消息
    USER_MESSAGES = {
        'VALIDATION_ERROR': '请求数据无效',
        'AUTHENTICATION_ERROR': '请先登录',
        'AUTHORIZATION_ERROR': '无权访问此资源',
        'NOT_FOUND': '资源不存在',
        'CONFLICT': '资源状态冲突',
        'RATE_LIMIT': '请求过于频繁，请稍后重试',
        'INTERNAL_ERROR': '服务器内部错误，请稍后重试'
    }
    
    @classmethod
    def handle_error(cls, error: Exception, context: dict = None) -> dict:
        """处理错误"""
        # 记录详细日志 (内部)
        logger.error(f"Error occurred: {error}", exc_info=True, extra=context or {})
        
        # 确定错误类型
        error_type = cls._classify_error(error)
        
        # 构建用户响应
        response = {
            'error': {
                'code': error_type,
                'message': cls.USER_MESSAGES.get(error_type, '发生错误'),
                'timestamp': datetime.utcnow().isoformat(),
                'request_id': context.get('request_id') if context else None
            }
        }
        
        # 开发环境显示详细信息
        if cls._is_development():
            response['error']['details'] = str(error)
            response['error']['traceback'] = traceback.format_exc()
        
        return response
    
    @classmethod
    def _classify_error(cls, error: Exception) -> str:
        """分类错误"""
        from sqlalchemy.exc import SQLAlchemyError
        from redis.exceptions import RedisError
        
        # 验证错误
        if isinstance(error, ValueError):
            return 'VALIDATION_ERROR'
        
        # 认证错误
        if isinstance(error, AuthenticationError):
            return 'AUTHENTICATION_ERROR'
        
        # 授权错误
        if isinstance(error, AuthorizationError):
            return 'AUTHORIZATION_ERROR'
        
        # 数据库错误
        if isinstance(error, SQLAlchemyError):
            # 不泄露 SQL 细节
            logger.error(f"Database error: {type(error).__name__}")
            return 'INTERNAL_ERROR'
        
        # 默认内部错误
        return 'INTERNAL_ERROR'
    
    @staticmethod
    def _is_development() -> bool:
        """检查是否开发环境"""
        import os
        return os.environ.get('ENV') == 'development'

# 自定义异常
class AuthenticationError(Exception):
    """认证错误"""
    pass

class AuthorizationError(Exception):
    """授权错误"""
    pass

class ValidationError(Exception):
    """验证错误"""
    def __init__(self, message: str, field: str = None):
        super().__init__(message)
        self.field = field

# API 错误处理中间件
from flask import jsonify

def error_handler_middleware(app):
    """全局错误处理"""
    
    @app.errorhandler(Exception)
    def handle_exception(e):
        response = SecureErrorHandler.handle_error(e)
        error_code = SecureErrorHandler.ERROR_CODES.get(
            response['error']['code'], 500
        )
        return jsonify(response), error_code
    
    @app.errorhandler(404)
    def handle_404(e):
        response = SecureErrorHandler.handle_error(e)
        return jsonify(response), 404
    
    @app.errorhandler(500)
    def handle_500(e):
        response = SecureErrorHandler.handle_error(e)
        return jsonify(response), 500
```

---

## 代码审查清单

### 安全审查清单

> ▎ 代码审查不是走过场，是最后防线。防线不守，漏洞就是上线的。

**审查清单**：
```
输入验证:
□ 所有用户输入是否验证
□ 是否使用白名单验证
□ 长度是否限制
□ 类型是否检查
□ 范围是否验证

输出编码:
□ 所有输出是否编码
□ 编码是否与上下文匹配
□ 是否使用安全库
□ 是否避免双重编码

认证授权:
□ 是否验证用户身份
□ 是否检查资源所有权
□ 是否实施最小权限
□ 会话是否安全

数据保护:
□ 敏感数据是否加密
□ 密码是否哈希存储
□ 密钥是否安全存储
□ 传输是否加密

错误处理:
□ 是否不泄露敏感信息
□ 是否有统一错误格式
□ 是否记录详细日志
□ 是否有适当错误代码

依赖安全:
□ 依赖是否最新
□ 是否有已知漏洞
□ 是否最小化依赖
□ 是否验证依赖来源
```

---

统计 **Sprint 交付 · 绩效评估**
```
┌───────────────┬────────────────┬────────────────┐
│  主动出击   │ ██████████ 5/5 │ [PUA 生效] 充足 │
├───────────────┼────────────────┼────────────────┤
│ + 验证闭环   │ ██████████ 5/5 │ 案例完整       │
├───────────────┼────────────────┼────────────────┤
│ 设计 代码质量   │ ██████████ 5/5 │ 生产就绪       │
└───────────────┴────────────────┴────────────────┘
综合：4.5
```
▎ 这才配得上 P8。安全编码不是一次学习，是持续实践。实践不停，代码才安全。

---

## 总结与思考

### 核心要点回顾

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

**安全编码框架**：
```
1. 输入处理
   - 白名单验证
   - 参数化查询
   - 文件路径验证

2. 输出编码
   - 上下文感知
   - HTML/JS/URL 编码
   - 使用安全库

3. 认证会话
   - 安全密码存储
   - 会话管理
   - Token 安全

4. 数据保护
   - 加密存储
   - 密钥管理
   - 传输加密

5. 错误处理
   - 不泄露信息
   - 统一格式
   - 详细日志
```

**关键实践**：
```
1. 安全设计
   - 威胁建模
   - 安全需求
   - 架构评审

2. 安全编码
   - 遵循规范
   - 使用安全库
   - 代码审查

3. 安全测试
   - 单元测试
   - 安全扫描
   - 渗透测试
```

### 深入思考问题

> ▎ 只动手不动脑，那是码农。动手又动脑，才是工程师。

**安全编码文化**：
```
1. 从个人到团队
   - 个人技能提升
   - 团队规范建立
   - 组织文化建设

2. 从被动到主动
   - 被动修复漏洞
   - 主动预防漏洞
   - 安全左移

3. 从成本到投资
   - 安全是成本
   - 安全是投资
   - 安全是竞争力
```

### 实战建议

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

**个人成长**：
```
1. 学习安全知识
   - OWASP Top 10
   - 安全编码规范
   - 漏洞案例分析

2. 实践安全编码
   - 日常代码应用
   - 代码审查参与
   - 安全工具使用

3. 持续提升
   - 关注安全动态
   - 学习新技术
   - 分享知识
```

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

2. 工具支持
   - 静态分析
   - 依赖扫描
   - 自动化测试

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

---

## 参考资料

### 学习资源
```
- OWASP Secure Coding Practices
  https://owasp.org/www-project-secure-coding-practices-quick-reference-guide/

- CERT Coding Standards
  https://wiki.sei.cmu.edu/confluence/display/seccode

- SANS Secure Coding Guidelines
  https://www.sans.org/top25-software-errors/
```

### 语言特定规范
```
- Python Security Guidelines
  https://docs.python.org/3/library/security.html

- Java Secure Coding Guidelines
  https://www.oracle.com/java/technologies/javase/seccodeguide.html

- Node.js Security Guidelines
  https://nodejs.org/en/docs/guides/security/
```

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

- SpotBugs (Java)
  https://spotbugs.github.io/

- ESLint Security
  https://github.com/nodesecurity/eslint-plugin-security
```

### 书籍推荐
```
- 《Secure Coding in C and C++》
- 《Writing Secure Code》
- 《The Security Development Lifecycle》
- 《Secure by Design》
```

---

**标记 明日预告**：Day 141 - SDL 安全开发生命周期

> ▎ 安全编码是基础能力，SDL 是流程保障——明天看 SDL 安全开发生命周期。

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

---

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

*新兴技术安全深化系列 Day 130-140 全部完成！应用安全核心系列 (Day 141-155) 即将开始！*