python-security

Python 安全 - 快速参考

不要使用此技能的情况

• 通用 OWASP 概念 - 使用 owasp 或 owasp-top-10 技能
• Node.js/TypeScript 安全 - 使用基础安全技能
• Java 安全 - 使用 java-security 技能
• 密钥管理 - 使用 secrets-management 技能

深入了解：使用 mcp__documentation__fetch_docs，并将技术指定为 fastapi 或 django，以获取框架专属的安全文档。

依赖项审计

# pip-audit - Official Python audit tool
pip-audit

# Safety - Check for known vulnerabilities
safety check

# pip-audit with requirements file
pip-audit -r requirements.txt

# Snyk for Python
snyk test

# Check outdated packages
pip list --outdated

CI/CD 集成

# GitHub Actions
- name: Security audit
  run: |
    pip install pip-audit safety
    pip-audit
    safety check

Bandit - 静态分析

# Run Bandit on project
bandit -r src/

# Generate JSON report
bandit -r src/ -f json -o bandit-report.json

# Skip specific tests
bandit -r src/ --skip B101,B601

# High severity only
bandit -r src/ -ll

Bandit 配置（.bandit）

# .bandit
skips: ['B101']  # Skip assert warnings in tests
exclude_dirs: ['tests', 'venv']

常见 Bandit 警告

代码	问题	修复方法
B101	使用了 assert	在生产环境中使用适当的验证
B105	硬编码密码	使用环境变量
B301	使用了 Pickle	使用 JSON 或更安全的序列化方式
B601	Shell 注入	使用带列表参数的 subprocess
B608	SQL 注入	使用参数化查询

SQL 注入防护

SQLAlchemy - 安全

# SAFE - Parameterized query
from sqlalchemy import text

result = db.execute(
    text("SELECT * FROM users WHERE email = :email"),
    {"email": email}
)

# SAFE - ORM query
user = db.query(User).filter(User.email == email).first()

# SAFE - FastAPI with SQLAlchemy
@app.get("/users/{user_id}")
async def get_user(user_id: int, db: Session = Depends(get_db)):
    return db.query(User).filter(User.id == user_id).first()

SQLAlchemy - 不安全

# UNSAFE - String formatting
query = f"SELECT * FROM users WHERE email = '{email}'"  # NEVER!
db.execute(query)

# UNSAFE - % formatting
query = "SELECT * FROM users WHERE email = '%s'" % email  # NEVER!

Django ORM - 安全

# SAFE - ORM querysets
User.objects.filter(email=email)
User.objects.get(pk=user_id)

# SAFE - Raw query with params
User.objects.raw("SELECT * FROM users WHERE email = %s", [email])

# SAFE - Extra with params
User.objects.extra(where=["email = %s"], params=[email])

XSS 防护

Django 模板（自动转义）

{{ user_input }}


{{ user_input|safe }}  

FastAPI/Jinja2

# Configure auto-escaping
from jinja2 import Environment, select_autoescape

env = Environment(
    autoescape=select_autoescape(['html', 'xml'])
)

手动清理

# Use bleach for HTML sanitization
import bleach

clean_html = bleach.clean(
    user_input,
    tags=['p', 'b', 'i', 'a'],
    attributes={'a': ['href']},
    strip=True
)

身份验证 - FastAPI

使用 python-jose 的 JWT

from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext

SECRET_KEY = os.environ["JWT_SECRET"]
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def create_access_token(data: dict) -> str:
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

def get_password_hash(password: str) -> str:
    return pwd_context.hash(password)

FastAPI OAuth2 依赖项

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception

    user = get_user(username)
    if user is None:
        raise credentials_exception
    return user

身份验证 - Django

Django 设置

# settings.py
PASSWORD_HASHERS = [
    'django.contrib.auth.hashers.Argon2PasswordHasher',
    'django.contrib.auth.hashers.PBKDF2PasswordHasher',
]

AUTH_PASSWORD_VALIDATORS = [
    {'NAME': 'django.contrib.auth.password_validation.MinimumLengthValidator',
     'OPTIONS': {'min_length': 12}},
    {'NAME': 'django.contrib.auth.password_validation.CommonPasswordValidator'},
    {'NAME': 'django.contrib.auth.password_validation.NumericPasswordValidator'},
]

SESSION_COOKIE_SECURE = True
SESSION_COOKIE_HTTPONLY = True
CSRF_COOKIE_SECURE = True

Django 速率限制

# Using django-ratelimit
from django_ratelimit.decorators import ratelimit

@ratelimit(key='ip', rate='5/m', method='POST', block=True)
def login_view(request):
    # login logic
    pass

输入验证

使用 Pydantic 的 FastAPI

from pydantic import BaseModel, EmailStr, Field, validator
import re

class CreateUserRequest(BaseModel):
    email: EmailStr
    password: str = Field(..., min_length=12, max_length=128)
    name: str = Field(..., min_length=2, max_length=100)

    @validator('password')
    def password_strength(cls, v):
        if not re.search(r'[A-Z]', v):
            raise ValueError('Password must contain uppercase')
        if not re.search(r'[a-z]', v):
            raise ValueError('Password must contain lowercase')
        if not re.search(r'\d', v):
            raise ValueError('Password must contain digit')
        if not re.search(r'[@$!%*?&]', v):
            raise ValueError('Password must contain special character')
        return v

    @validator('name')
    def name_alphanumeric(cls, v):
        if not re.match(r"^[a-zA-Z\s\-']+$", v):
            raise ValueError('Name contains invalid characters')
        return v

Django 表单

from django import forms
from django.core.validators import RegexValidator

class UserRegistrationForm(forms.Form):
    email = forms.EmailField(max_length=255)
    password = forms.CharField(
        min_length=12,
        max_length=128,
        widget=forms.PasswordInput
    )
    name = forms.CharField(
        min_length=2,
        max_length=100,
        validators=[RegexValidator(r"^[a-zA-Z\s\-']+$")]
    )

命令注入防护

import subprocess
import shlex

# SAFE - Use list arguments
subprocess.run(["ls", "-la", directory], check=True)

# SAFE - Use shlex.split for shell-like parsing
subprocess.run(shlex.split(f"ls -la {shlex.quote(directory)}"), check=True)

# UNSAFE - shell=True with user input
subprocess.run(f"ls -la {directory}", shell=True)  # NEVER!

# UNSAFE - os.system
os.system(f"ls -la {directory}")  # NEVER!

安全文件上传

FastAPI

from fastapi import UploadFile, HTTPException
import uuid

ALLOWED_TYPES = {"image/jpeg", "image/png", "application/pdf"}
MAX_SIZE = 10 * 1024 * 1024  # 10MB

@app.post("/upload")
async def upload_file(file: UploadFile):
    # Validate content type
    if file.content_type not in ALLOWED_TYPES:
        raise HTTPException(400, "File type not allowed")

    # Read and check size
    contents = await file.read()
    if len(contents) > MAX_SIZE:
        raise HTTPException(400, "File too large")

    # Generate safe filename
    ext = Path(file.filename).suffix if file.filename else ""
    safe_name = f"{uuid.uuid4()}{ext}"

    # Store outside web root
    file_path = UPLOAD_DIR / safe_name
    async with aiofiles.open(file_path, 'wb') as f:
        await f.write(contents)

    return {"filename": safe_name}

密钥管理

import os
from functools import lru_cache
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str
    jwt_secret: str
    api_key: str

    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"

@lru_cache
def get_settings():
    return Settings()

# Usage
settings = get_settings()
db_url = settings.database_url

# NEVER hardcode secrets
# JWT_SECRET = "hardcoded-secret"  # NEVER!

安全事件日志记录

import logging
from datetime import datetime

logger = logging.getLogger("security")

def log_login_attempt(username: str, success: bool, ip: str):
    logger.info(
        "Login attempt",
        extra={
            "user": username,
            "success": success,
            "ip": ip,
            "timestamp": datetime.utcnow().isoformat()
        }
    )

def log_access_denied(username: str, resource: str, ip: str):
    logger.warning(
        "Access denied",
        extra={
            "user": username,
            "resource": resource,
            "ip": ip
        }
    )

# NEVER log sensitive data
# logger.info(f"Password: {password}")  # NEVER!
# logger.info(f"Token: {token}")        # NEVER!

CORS 配置

FastAPI

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://myapp.com"],  # Not ["*"] in production
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
)

Django

# settings.py with django-cors-headers
CORS_ALLOWED_ORIGINS = [
    "https://myapp.com",
]
CORS_ALLOW_CREDENTIALS = True

反模式

反模式	有害原因	正确做法
shell=True in subprocess	命令注入	使用列表参数
f-string in SQL query	SQL 注入	使用参数化查询
pickle.loads(user_data)	任意代码执行	使用 JSON 或验证来源
eval(user_input)	代码注入	绝不要对用户输入使用 eval
硬编码密钥	密钥泄露	使用环境变量
生产环境中的 DEBUG=True	信息泄露	设置 DEBUG=False
{{user_input|safe}}	XSS 漏洞	除非已清理，否则避免使用

快速故障排查

问题	可能原因	解决方案
Bandit B105 警告	硬编码的密码字符串	移至环境变量
pip-audit 发现 CVE	易受攻击的软件包	更新到已修复的版本
浏览器中出现 CORS 错误	不允许该来源	将来源添加到允许列表
JWT 解码失败	密钥错误或已过期	检查密钥和过期时间
密码哈希不匹配	哈希算法不同	使用相同的上下文进行哈希/验证
速率限制无效	中间件顺序错误	在路由之前添加速率限制器

安全扫描命令

# Static analysis
bandit -r src/

# Dependency audit
pip-audit
safety check

# All-in-one with Snyk
snyk test

# Check for secrets
gitleaks detect
trufflehog git file://.

# Django security check
python manage.py check --deploy

相关技能

• OWASP Top 10:2025
• OWASP 通用指南
• 密钥管理
• 供应链安全