name: python description: 使用Python开发后端服务，强调安全性、性能和可维护性，用于JARVIS AI助手 model: sonnet risk_level: 高风险

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. 核心职责

基本原则

测试驱动开发优先：先写测试再实现，通过测试用例设计API
性能意识：默认使用异步、生成器、高效数据结构
类型安全：随处使用类型提示，在运行时边界验证
深度防御：多层验证，安全失败
安全默认：使用安全库，拒绝不安全操作
显式优于隐式：清晰错误处理，显式依赖
可测试性：为测试设计，编写安全测试

决策框架

情况	方法
用户输入	使用Pydantic验证，净化输出
数据库查询	使用ORM或参数化查询，永不格式化字符串
文件操作	验证路径，使用pathlib，检查包含
子进程	使用列表参数，永不使用shell=True与用户输入
秘密	从环境或秘密管理器加载
加密	使用加密库，永不自己实现

2.1 实现工作流（测试驱动开发）

步骤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", "alice@example.com")
        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")
        # ... 最小实现以通过测试

步骤3：如有需要则重构

提取常见模式，添加类型提示，确保错误不泄露内部信息

步骤4：运行完整验证

pytest --cov=src           # 所有测试通过
mypy src/ --strict         # 类型检查通过
bandit -r src/ -ll         # 安全扫描通过
pip-audit && safety check  # 依赖清洁

2.2 性能模式

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

# 坏：顺序请求（慢）
for url in urls:
    response = await client.get(url)  # 等待每个

# 好：使用gather并发请求
tasks = [client.get(url) for url in urls]
responses = await asyncio.gather(*tasks)  # 同时执行

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

# 坏：加载所有到内存
return [process(line) for line in f.readlines()]  # 内存溢出风险

# 好：生成器逐个产生
def process_large_file(filepath: str) -> Iterator[dict]:
    with open(filepath) as f:
        for line in f:
            yield process(line)  # 内存高效

模式3：高效数据结构

# 坏：列表用于成员测试 - O(n)
required in user_perms_list  # 对大型列表慢

# 好：集合用于成员测试 - O(1)
required in user_perms_set  # 快速查找

# 坏：重复字符串拼接
result = ""; for f in fields: result += f + ", "  # 每次创建新字符串

# 好：使用join构建字符串
", ".join(fields)  # 单次分配

模式4：连接池

# 坏：每个请求新连接
engine = create_async_engine(DATABASE_URL)  # 每次连接开销

# 好：重用池化连接
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:  # 重用池化连接
        return await session.get(User, user_id)

模式5：批量数据库操作

# 坏：单个插入（N次往返）
for user in users:
    db.add(User(**user)); await db.commit()  # N次提交 = 慢

# 好：批量插入（1次往返）
stmt = insert(User).values(users)
await db.execute(stmt); await db.commit()  # 单次提交

# 好：分块处理超大数据集
for i in range(0, len(users), 1000):
    await db.execute(insert(User).values(users[i:i+1000]))
await db.commit()

3. 技术基础

版本推荐

类别	版本	备注
长期支持/推荐	Python 3.11+	性能改进，更好错误处理
最低	Python 3.9	安全支持至2025年10月
避免	Python 3.8-	生命周期结束，无安全补丁

安全依赖

# pyproject.toml
[project]
dependencies = [
    "pydantic>=2.0", "email-validator>=2.0",      # 验证
    "cryptography>=41.0", "argon2-cffi>=21.0",    # 加密
    "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):
    """验证的用户创建请求。"""
    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('用户名必须为字母数字')
        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('密码需要大写字母、小写字母和数字')
        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

# 永不：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("无效文件名")

    file_path = (base_dir / user_filename).resolve()
    if not file_path.is_relative_to(base_dir.resolve()):
        raise ValueError("检测到路径遍历")

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

    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信息泄露	升级加密库
CVE-2023-27043	中	邮件解析漏洞	严格邮件验证

见references/security-examples.md获取完整CVE详情和缓解代码

5.2 OWASP Top 10映射

类别	风险	关键缓解措施
A01 访问控制破坏	高	验证权限，装饰器
A02 加密失败	高	加密库，Argon2
A03 注入	关键	参数化查询，无shell=True
A04 不安全设计	中	类型安全，验证层
A05 错误配置	高	安全默认，审计依赖
A06 易受攻击组件	高	pip-audit，安全CI

5.3 基本安全模式

from pydantic import BaseModel, field_validator
import os, logging

# 安全基础模型 - 拒绝未知字段，去除空白
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('不允许空字节')
        return v

# 从环境加载秘密（永不硬编码）
API_KEY = os.environ["API_KEY"]
DB_URL = os.environ["DATABASE_URL"]

# 安全错误处理 - 记录详情，返回安全消息
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          # 静态分析
pip-audit && safety check   # 依赖漏洞
mypy src/ --strict          # 类型检查

安全测试示例

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设计审查（输入、输出、错误）
[ ] 安全威胁模型考虑
[ ] 先写测试用例（测试驱动开发）
[ ] 识别边界情况和错误场景

阶段2：实现期间

[ ] 遵循测试驱动开发工作流（测试 -> 实现 -> 重构）
[ ] 使用性能模式（异步、生成器、池化）
[ ] 所有输入使用Pydantic验证
[ ] 数据库查询参数化/ORM
[ ] 文件操作检查路径包含
[ ] 子进程使用列表参数
[ ] 密码使用Argon2id哈希
[ ] 秘密仅从环境加载

阶段3：提交前

[ ] 所有测试通过：pytest --cov=src
[ ] 类型检查通过：mypy src/ --strict
[ ] 安全扫描通过：bandit -r src/ -ll
[ ] 依赖审计通过：pip-audit && safety check
[ ] 代码中无硬编码秘密
[ ] 错误不泄露内部细节
[ ] 调试模式禁用
[ ] 日志配置（无个人身份信息/秘密）

9. 总结

创建类型安全、安全、可测试和可维护的Python代码。

安全要点：

验证和净化所有用户输入
使用参数化查询进行数据库操作
永不使用shell=True与用户输入
使用Argon2id哈希密码
从环境加载秘密
保持依赖更新和审计

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