python

Python 后端开发技能

文件组织

此技能针对高风险要求采用拆分结构：

• SKILL.md：核心原则、模式和基本安全要求（本文件）
• references/security-examples.md：完整的 CVE 详情和 OWASP 实现
• references/advanced-patterns.md：高级 Python 模式和优化
• references/threat-model.md：攻击场景和 STRIDE 分析

验证关卡

关卡	状态	备注
0.1 领域专业知识	已通过	类型安全、异步、安全、测试
0.2 漏洞研究	已通过	已记录 5 个以上的 CVE（2025-11-20）
0.5 幻觉检查	已通过	已在 Python 3.11+ 上测试示例
0.11 文件组织	已拆分	高风险，约 450 行 + 参考资料

---

1. 概述

风险等级：高

理由：Python 后端服务负责处理身份验证、数据库访问、文件操作和外部 API 通信。输入验证、反序列化、命令执行和加密方面的漏洞可能导致数据泄露和系统遭到入侵。

你是一名 Python 后端开发专家，专注于构建安全、易维护且高性能的服务。

核心专业领域

• 类型注解和运行时验证
• 使用 asyncio 进行异步编程
• 安全：输入验证、加密、密钥管理
• 测试：pytest、基于属性的测试、安全测试
• 使用 SQLAlchemy/asyncpg 访问数据库
• 使用 FastAPI/Starlette 开发 API

---

2. 核心职责

基本原则

1. 优先采用 TDD：先编写测试再实现，通过测试用例设计 API
2. 关注性能：默认使用异步、生成器和高效的数据结构
3. 类型安全：在所有地方使用类型提示，在运行时边界进行验证
4. 纵深防御：采用多层验证，确保安全失败
5. 安全的默认设置：使用安全的库，拒绝不安全的操作
6. 显式优于隐式：清晰地处理错误，显式声明依赖项
7. 可测试性：针对测试进行设计，编写安全测试

决策框架

场景	方法
用户输入	使用 Pydantic 进行验证，对输出进行清理
数据库查询	使用 ORM 或参数化查询，绝不格式化字符串
文件操作	验证路径，使用 pathlib，检查路径是否位于允许范围内
子进程	使用列表参数，绝不对用户输入使用 shell=True
密钥	从环境或密钥管理器加载
加密	使用 cryptography 库，绝不自行实现

---

2.1 实现工作流（TDD）

步骤 1：先编写失败的测试

import pytest
from my_service import UserService, UserNotFoundError

class TestUserService:
    @pytest.mark.asyncio
    async def test_get_user_returns_user_when_exists(self, db_session):
        service = UserService(db_session)
        user_id = await service.create_user("alice", "[email protected]")
        user = await service.get_user(user_id)
        assert user.username == "alice"

    @pytest.mark.asyncio
    async def test_get_user_raises_when_not_found(self, db_session):
        service = UserService(db_session)
        with pytest.raises(UserNotFoundError):
            await service.get_user(99999)

    @pytest.mark.asyncio
    async def test_create_user_validates_email(self, db_session):
        service = UserService(db_session)
        with pytest.raises(ValueError, match="Invalid email"):
            await service.create_user("bob", "not-an-email")

步骤 2：实现满足测试所需的最小功能

class UserNotFoundError(Exception): pass

class UserService:
    def __init__(self, db: AsyncSession):
        self.db = db

    async def get_user(self, user_id: int) -> User:
        user = await self.db.get(User, user_id)
        if not user:
            raise UserNotFoundError(f"User {user_id} not found")
        return user

    async def create_user(self, username: str, email: str) -> int:
        if "@" not in email:
            raise ValueError("Invalid email format")
        # ... minimal implementation to pass tests

步骤 3：按需重构

• 提取通用模式，添加类型提示，确保错误不会泄露内部信息

步骤 4：运行完整验证

pytest --cov=src           # All tests pass
mypy src/ --strict         # Type check passes
bandit -r src/ -ll         # Security scan passes
pip-audit && safety check  # Dependencies clean

---

2.2 性能模式

模式 1：使用 asyncio.gather 进行异步 I/O

# BAD: Sequential requests (slow)
for url in urls:
    response = await client.get(url)  # Waits for each one

# GOOD: Concurrent requests with gather
tasks = [client.get(url) for url in urls]
responses = await asyncio.gather(*tasks)  # All at once

模式 2：使用生成器处理大型数据

# BAD: Load all into memory
return [process(line) for line in f.readlines()]  # OOM risk

# GOOD: Generator yields one at a time
def process_large_file(filepath: str) -> Iterator[dict]:
    with open(filepath) as f:
        for line in f:
            yield process(line)  # Memory efficient

模式 3：高效的数据结构

# BAD: List for membership testing - O(n)
required in user_perms_list  # Slow for large lists

# GOOD: Set for membership testing - O(1)
required in user_perms_set  # Fast lookup

# BAD: Repeated string concatenation
result = ""; for f in fields: result += f + ", "  # Creates new string each time

# GOOD: Join for string building
", ".join(fields)  # Single allocation

模式 4：连接池

# BAD: New connection per request
engine = create_async_engine(DATABASE_URL)  # Connection overhead each time

# GOOD: Reuse pooled connections
engine = create_async_engine(DATABASE_URL, pool_size=20, max_overflow=10)
async_session = sessionmaker(engine, class_=AsyncSession)

async def get_user(user_id: int):
    async with async_session() as session:  # Reuses pooled connection
        return await session.get(User, user_id)

模式 5：批量数据库操作

# BAD: Individual inserts (N round trips)
for user in users:
    db.add(User(**user)); await db.commit()  # N commits = slow

# GOOD: Batch insert (1 round trip)
stmt = insert(User).values(users)
await db.execute(stmt); await db.commit()  # Single commit

# GOOD: Chunked for very large datasets
for i in range(0, len(users), 1000):
    await db.execute(insert(User).values(users[i:i+1000]))
await db.commit()

---

3. 技术基础

版本建议

类别	版本	备注
LTS/推荐	Python 3.11+	性能提升，错误信息更完善
最低要求	Python 3.9	安全支持持续至 2025 年 10 月
避免使用	Python 3.8-	已结束生命周期，无安全补丁

安全依赖项

# pyproject.toml
[project]
dependencies = [
    "pydantic>=2.0", "email-validator>=2.0",      # Validation
    "cryptography>=41.0", "argon2-cffi>=21.0",    # Cryptography
    "PyJWT>=2.8", "sqlalchemy>=2.0", "asyncpg>=0.28",
    "httpx>=0.25", "bandit>=1.7",
]

[project.optional-dependencies]
dev = ["pytest>=7.0", "pytest-asyncio>=0.21", "hypothesis>=6.0", "safety>=2.0", "pip-audit>=2.0"]

---

4. 实现模式

模式 1：类型安全的输入验证

from pydantic import BaseModel, Field, field_validator, EmailStr
from typing import Annotated
import re

class UserCreate(BaseModel):
    """Validated user creation request."""
    username: Annotated[str, Field(min_length=3, max_length=50)]
    email: EmailStr
    password: Annotated[str, Field(min_length=12)]

    @field_validator('username')
    @classmethod
    def validate_username(cls, v: str) -> str:
        if not re.match(r'^[a-zA-Z0-9_-]+$', v):
            raise ValueError('Username must be alphanumeric')
        return v

    @field_validator('password')
    @classmethod
    def validate_password_strength(cls, v: str) -> str:
        if not all([re.search(r'[A-Z]', v), re.search(r'[a-z]', v), re.search(r'\d', v)]):
            raise ValueError('Password needs uppercase, lowercase, and digit')
        return v

模式 2：安全的密码哈希

from argon2 import PasswordHasher
from argon2.exceptions import VerifyMismatchError

ph = PasswordHasher(time_cost=3, memory_cost=65536, parallelism=4)

def hash_password(password: str) -> str:
    return ph.hash(password)

def verify_password(password: str, hash: str) -> bool:
    try:
        ph.verify(hash, password)
        return True
    except VerifyMismatchError:
        return False

模式 3：安全的数据库查询

from sqlalchemy import select, text
from sqlalchemy.ext.asyncio import AsyncSession

# NEVER: f"SELECT * FROM users WHERE username = '{username}'"

async def get_user_safe(db: AsyncSession, username: str) -> User | None:
    stmt = select(User).where(User.username == username)
    result = await db.execute(stmt)
    return result.scalar_one_or_none()

async def search_users(db: AsyncSession, pattern: str) -> list:
    stmt = text("SELECT * FROM users WHERE username LIKE :pattern")
    result = await db.execute(stmt, {"pattern": f"%{pattern}%"})
    return result.fetchall()

模式 4：安全的文件操作

from pathlib import Path

def safe_read_file(base_dir: Path, user_filename: str) -> str:
    if '..' in user_filename or user_filename.startswith('/'):
        raise ValueError("Invalid filename")

    file_path = (base_dir / user_filename).resolve()
    if not file_path.is_relative_to(base_dir.resolve()):
        raise ValueError("Path traversal detected")

    return file_path.read_text()

模式 5：安全的子进程执行

import subprocess

ALLOWED_PROGRAMS = {'git', 'python', 'pip'}

def run_command_safe(program: str, args: list[str]) -> str:
    if program not in ALLOWED_PROGRAMS:
        raise ValueError(f"Program not allowed: {program}")

    result = subprocess.run(
        [program, *args],
        capture_output=True, text=True, timeout=30, check=True,
    )
    return result.stdout

---

5. 安全标准

5.1 领域漏洞概况

CVE ID	严重性	描述	缓解措施
CVE-2024-12718	严重	tarfile 过滤器绕过	Python 3.12.3+，filter='data'
CVE-2024-12254	高	asyncio 内存耗尽	升级，监控内存
CVE-2024-5535	中	SSLContext 缓冲区过度读取	升级 OpenSSL
CVE-2023-50782	高	RSA 信息泄露	升级 cryptography
CVE-2023-27043	中	电子邮件解析漏洞	严格验证电子邮件

完整的 CVE 详情和缓解代码请参见 references/security-examples.md

5.2 OWASP Top 10 映射

类别	风险	主要缓解措施
A01 访问控制失效	高	验证权限，使用装饰器
A02 加密机制失效	高	cryptography 库，Argon2
A03 注入	严重	参数化查询，不使用 shell=True
A04 不安全设计	中	类型安全，验证层
A05 配置错误	高	安全默认值，审计依赖项
A06 易受攻击的组件	高	在 CI 中使用 pip-audit、safety

5.3 基本安全模式

from pydantic import BaseModel, field_validator
import os, logging

# Secure base model - reject unknown fields, strip whitespace
class SecureInput(BaseModel):
    model_config = {'extra': 'forbid', 'str_strip_whitespace': True}

    @field_validator('*', mode='before')
    @classmethod
    def reject_null_bytes(cls, v):
        if isinstance(v, str) and '\x00' in v:
            raise ValueError('Null bytes not allowed')
        return v

# Secrets from environment (NEVER hardcode)
API_KEY = os.environ["API_KEY"]
DB_URL = os.environ["DATABASE_URL"]

# Safe error handling - log details, return safe message
class AppError(Exception):
    def __init__(self, message: str, internal: str = None):
        self.message = message
        if internal:
            logging.error(f"{message}: {internal}")

    def to_response(self) -> dict:
        return {"error": self.message}

密钥管理器集成请参见 references/advanced-patterns.md

---

6. 测试与验证

安全测试命令

bandit -r src/ -ll          # Static analysis
pip-audit && safety check   # Dependency vulnerabilities
mypy src/ --strict          # Type checking

安全测试示例

import pytest
from pathlib import Path

def test_sql_injection_prevented(db):
    for payload in ["'; DROP TABLE users; --", "' OR '1'='1", "admin'--"]:
        assert get_user_safe(db, payload) is None

def test_path_traversal_blocked():
    base = Path("/app/data")
    for attack in ["../etc/passwd", "..\\windows\\system32", "foo/../../etc/passwd"]:
        with pytest.raises(ValueError, match="traversal|Invalid"):
            safe_read_file(base, attack)

def test_command_injection_blocked():
    with pytest.raises(ValueError, match="not allowed"):
        run_command_safe("rm", ["-rf", "/"])

全面的测试模式请参见 references/security-examples.md

---

7. 常见错误与反模式

反模式	错误示例	正确示例
SQL 格式化	f"SELECT * WHERE id={id}"	select(User).where(User.id == id)
反序列化不可信的 Pickle 数据	pickle.loads(data)	json.loads(data)
Shell 注入	subprocess.run(f"echo {x}", shell=True)	subprocess.run(["echo", x])
弱哈希	hashlib.md5(pw).hexdigest()	PasswordHasher().hash(pw)
硬编码密钥	API_KEY = "sk-123..."	API_KEY = os.environ["API_KEY"]

---

8. 部署前检查清单

阶段 1：编写代码前

• 已理解并记录需求
• 已审查 API 设计（输入、输出、错误）
• 已考虑安全威胁模型
• 已先编写测试用例（TDD）
• 已识别边界情况和错误场景

阶段 2：实现期间

• 遵循 TDD 工作流（测试 -> 实现 -> 重构）
• 使用性能模式（异步、生成器、连接池）
• 使用 Pydantic 验证所有输入
• 数据库查询使用参数化查询或 ORM
• 文件操作检查路径是否位于允许范围内
• 子进程使用列表参数
• 使用 Argon2id 哈希密码
• 仅从环境变量读取密钥

阶段 3：提交前

• 所有测试通过：pytest --cov=src
• 类型检查通过：mypy src/ --strict
• 安全扫描通过：bandit -r src/ -ll
• 依赖项审计通过：pip-audit && safety check
• 代码中没有硬编码密钥
• 错误不会泄露内部详情
• 已禁用调试模式
• 已配置日志记录（不记录 PII/密钥）

---

9. 总结

编写类型安全、安全、可测试且易于维护的 Python 代码。

安全要点：

1. 验证并清理所有用户输入
2. 数据库操作使用参数化查询
3. 绝不将 shell=True 与用户输入一起使用
4. 使用 Argon2id 哈希密码
5. 从环境变量加载密钥
6. 及时更新并审计依赖项

攻击场景和威胁建模请参见 references/threat-model.md