Python 安全编程：输入验证、加密与常见漏洞防护

from typing import Optional
import re

class InputValidator:
    """输入验证工具类"""
    
    @staticmethod
    def validate_username(username: str) -> bool:
        """用户名：字母数字下划线，3-20位"""
        if not username:
            return False
        return bool(re.match(r'^[a-zA-Z0-9_]{3,20}$', username))
    
    @staticmethod
    def validate_email(email: str) -> bool:
        """邮箱格式验证"""
        pattern = r'^[\w\.-]+@[\w\.-]+\.\w+$'
        return bool(re.match(pattern, email))
    
    @staticmethod
    def sanitize_filename(filename: str) -> Optional[str]:
        """清理文件名，防止路径遍历"""
        filename = filename.strip().replace('..', '')
        # 移除路径分隔符
        filename = filename.replace('/', '').replace('\\', '')
        # 只允许字母数字和常见扩展名
        if re.match(r'^[\w\-\.]+\.[a-zA-Z0-9]{1,10}$', filename):
            return filename
        return None

# Pydantic 模型验证（推荐用于 Web API）
from pydantic import BaseModel, EmailStr, Field, validator

class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=20, regex=r'^[\w]+$')
    email: EmailStr
    age: int = Field(..., ge=0, le=150)
    
    @validator('username')
    def username_not_admin(cls, v):
        if v.lower() in ('admin', 'root', 'superuser'):
            raise ValueError('保留用户名')
        return v

import sqlite3

# ❌ 极其危险！字符串拼接 SQL
def get_user_unsafe(username):
    conn = sqlite3.connect('app.db')
    cursor = conn.cursor()
    # 攻击者输入：' OR '1'='1
    cursor.execute(f"SELECT * FROM users WHERE name = '{username}'")
    return cursor.fetchall()

# ✅ 参数化查询（始终使用）
def get_user_safe(username):
    conn = sqlite3.connect('app.db')
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users WHERE name = ?", (username,))
    return cursor.fetchall()

# SQLAlchemy ORM 自动参数化
from sqlalchemy import select
from sqlalchemy.orm import Session

def get_user_orm(session: Session, username: str):
    stmt = select(User).where(User.name == username)
    return session.execute(stmt).scalar_one_or_none()

# 使用 bcrypt（推荐）
# pip install bcrypt
import bcrypt

def hash_password(password: str) -> str:
    """生成密码哈希"""
    salt = bcrypt.gensalt(rounds=12)
    hashed = bcrypt.hashpw(password.encode(), salt)
    return hashed.decode()

def verify_password(password: str, hashed: str) -> bool:
    """验证密码"""
    return bcrypt.checkpw(password.encode(), hashed.encode())

# 使用 Python 3.11+ 的 hashlib.scrypt（内建）
import hashlib
import secrets

def hash_password_scrypt(password: str) -> str:
    salt = secrets.token_hex(16)
    hashed = hashlib.scrypt(
        password.encode(),
        salt=salt.encode(),
        n=16384, r=8, p=1, dklen=64
    ).hex()
    return f"{salt}${hashed}"

# pip install pyjwt
import jwt
import datetime
from typing import Optional

SECRET_KEY = secrets.token_urlsafe(32)   # 生产环境应存环境变量
ALGORITHM = "HS256"

def create_access_token(data: dict, expires_delta: Optional[datetime.timedelta] = None):
    to_encode = data.copy()
    expire = datetime.datetime.utcnow() + (expires_delta or datetime.timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def verify_token(token: str) -> Optional[dict]:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except jwt.ExpiredSignatureError:
        return None
    except jwt.InvalidTokenError:
        return None

import urllib.request
import ssl

# ✅ 验证服务器证书（默认行为，不要关闭）
resp = urllib.request.urlopen("https://example.com")

# ❌ 永远不要这样做！
ctx = ssl.create_default_context()
ctx.check_hostname = False    # 危险！
ctx.verify_mode = ssl.CERT_NONE   # 极其危险！

# 自签名证书
ctx = ssl.create_default_context(cafile="/path/to/ca.crt")
resp = urllib.request.urlopen("https://self-signed.example.com", context=ctx)

# requests 库同样
def safe_request():
    import requests
    # verify=True 是默认值，保持即可
    resp = requests.get("https://api.example.com", verify=True)
    return resp.json()

from pathlib import Path

# ❌ 危险：用户可能传入 ../../../etc/passwd
user_file = request.args.get('file')
content = open(f"/uploads/{user_file}").read()

# ✅ 安全：限制在指定目录内
def safe_read_file(base_dir: Path, user_path: str) -> str:
    target = (base_dir / user_path).resolve()
    # 确保最终在指定目录内
    if not str(target).startswith(str(base_dir.resolve())):
        raise ValueError("非法路径")
    return target.read_text()

# 使用 werkzeug.utils.secure_filename（Flask）
from werkzeug.utils import secure_filename

filename = secure_filename(user_input)   # 清理危险字符

import pickle
import json

# ❌ pickle 不安全！不要反序列化不受信任的数据
# attacker_payload = pickle.loads(untrusted_data)   # 可能执行任意代码

# ✅ 使用 JSON（安全但功能有限）
data = json.loads(untrusted_json)

# ✅ 使用 MessagePack（二进制，相对安全）
# msgpack.unpackb(data, strict_map_key=True)

# 如果必须用 pickle，签名验证
import hmac

def safe_pickle_loads(data: bytes, secret: bytes):
    if len(data) < 32:
        raise ValueError("数据太短")
    
    signature, payload = data[:32], data[32:]
    expected = hmac.new(secret, payload, 'sha256').digest()
    
    if not hmac.compare_digest(signature, expected):
        raise ValueError("签名无效")
    
    return pickle.loads(payload)

import subprocess

# ❌ 极其危险！shell=True + 用户输入
user_input = request.form.get('filename')
subprocess.run(f"rm -rf {user_input}", shell=True)
# 用户输入：'; rm -rf /; echo '

# ✅ 安全：参数列表，禁用 shell
subprocess.run(["rm", "-rf", user_input])   # 即使注入也只是一个文件名

# ✅ 更安全：校验参数
ALLOWED_FILES = {"log1.txt", "log2.txt"}
if user_input not in ALLOWED_FILES:
    raise ValueError("不允许的文件")

# ❌ eval/exec 极其危险
eval(user_input)   # 可以执行任何 Python 代码！

# ✅ 用 ast.literal_eval 解析字面量
import ast
result = ast.literal_eval("[1, 2, 3]")   # 安全：只允许字面量

import secrets
import string

# 生成安全令牌
token = secrets.token_urlsafe(32)   # 43字符的 URL 安全字符串
hex_token = secrets.token_hex(32)   # 64字符的十六进制

# 生成随机密码
alphabet = string.ascii_letters + string.digits + "!@#$%^&*"
password = ''.join(secrets.choice(alphabet) for _ in range(20))

# 生成随机选择（如抽奖）
participants = ['Alice', 'Bob', 'Charlie']
winner = secrets.choice(participants)

# 比较令牌（防时序攻击）
import hmac
if hmac.compare_digest(user_token, expected_token):
    print("验证通过")