当前位置：首页 > news >正文

【安全】API安全最佳实践：从认证到防护的完整指南

news 2026/5/28 5:35:46

一、API安全概述

1.1 API安全的重要性

API是现代应用的核心接口，保护API安全至关重要：

数据保护：防止敏感数据泄露
身份认证：确保只有授权用户访问
防止攻击：抵御各种安全威胁
合规要求：满足行业安全标准

1.2 常见API安全威胁

威胁类型	描述	风险等级
SQL注入	通过输入注入恶意SQL	高
XSS攻击	跨站脚本攻击	高
CSRF攻击	跨站请求伪造	中
身份伪造	伪造身份进行请求	高
拒绝服务	耗尽服务器资源	高
数据泄露	敏感信息泄露	高

二、认证与授权

2.1 JWT认证

# JWT认证实现 import jwt from datetime import datetime, timedelta class JWTAuthenticator: def __init__(self, secret_key): self.secret_key = secret_key def generate_token(self, user_id, expires_hours=24): payload = { "user_id": user_id, "exp": datetime.utcnow() + timedelta(hours=expires_hours), "iat": datetime.utcnow() } return jwt.encode(payload, self.secret_key, algorithm="HS256") def verify_token(self, token): try: payload = jwt.decode(token, self.secret_key, algorithms=["HS256"]) return payload except jwt.ExpiredSignatureError: raise ValueError("Token已过期") except jwt.InvalidTokenError: raise ValueError("无效的Token")

2.2 OAuth2授权

# OAuth2客户端实现 from oauthlib.oauth2 import WebApplicationClient class OAuth2Client: def __init__(self, client_id, client_secret, authorization_url, token_url): self.client = WebApplicationClient(client_id) self.client_secret = client_secret self.authorization_url = authorization_url self.token_url = token_url def get_authorization_url(self, redirect_uri, scope): return self.client.prepare_request_uri( self.authorization_url, redirect_uri=redirect_uri, scope=scope ) def fetch_token(self, redirect_uri, code): return self.client.fetch_token( self.token_url, client_secret=self.client_secret, code=code, redirect_uri=redirect_uri )

2.3 API密钥管理

# API密钥管理 import uuid from datetime import datetime class APIKeyManager: def __init__(self): self.keys = {} def generate_key(self, user_id): api_key = str(uuid.uuid4()) self.keys[api_key] = { "user_id": user_id, "created_at": datetime.now(), "last_used": None, "is_active": True } return api_key def validate_key(self, api_key): if api_key not in self.keys: return False if not self.keys[api_key]["is_active"]: return False self.keys[api_key]["last_used"] = datetime.now() return True def revoke_key(self, api_key): if api_key in self.keys: self.keys[api_key]["is_active"] = False

三、输入验证与过滤

3.1 请求参数验证

# 请求参数验证 from pydantic import BaseModel, ValidationError class UserCreateRequest(BaseModel): username: str email: str password: str age: int = None class Config: schema_extra = { "example": { "username": "john_doe", "email": "john@example.com", "password": "secure_password", "age": 25 } } def validate_request(data): try: return UserCreateRequest(**data) except ValidationError as e: raise ValueError(f"请求参数验证失败: {e}")

3.2 SQL注入防护

# SQL注入防护 import psycopg2 class SafeDatabase: def __init__(self, connection_string): self.connection_string = connection_string def query_user(self, user_id): conn = psycopg2.connect(self.connection_string) cursor = conn.cursor() # 使用参数化查询 query = "SELECT * FROM users WHERE id = %s" cursor.execute(query, (user_id,)) result = cursor.fetchone() conn.close() return result

3.3 XSS防护

# XSS防护 from html import escape class XSSProtector: def sanitize_input(self, input_string): return escape(input_string) def sanitize_html(self, html_content): # 使用更强大的HTML清理库 import bleach return bleach.clean(html_content)

四、安全头与HTTPS

4.1 安全头配置

# 安全头配置 from fastapi import FastAPI, Response app = FastAPI() @app.middleware("http") async def security_headers(request, call_next): response = await call_next(request) response.headers["X-Content-Type-Options"] = "nosniff" response.headers["X-Frame-Options"] = "DENY" response.headers["X-XSS-Protection"] = "1; mode=block" response.headers["Content-Security-Policy"] = "default-src 'self'" response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains" return response

4.2 HTTPS配置

# Nginx HTTPS配置 server { listen 443 ssl; server_name api.example.com; ssl_certificate /path/to/certificate.crt; ssl_certificate_key /path/to/private.key; ssl_protocols TLSv1.2 TLSv1.3; ssl_ciphers HIGH:!aNULL:!MD5; location / { proxy_pass http://localhost:8000; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; } }

五、速率限制与防护

5.1 速率限制实现

# 速率限制 from fastapi import FastAPI, Request, HTTPException from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from datetime import datetime from collections import defaultdict app = FastAPI() security = HTTPBearer() # 存储请求计数 request_counts = defaultdict(lambda: {"count": 0, "last_reset": datetime.now()}) def rate_limit(request: Request, limit=100, window_seconds=3600): client_ip = request.client.host now = datetime.now() if (now - request_counts[client_ip]["last_reset"]).seconds > window_seconds: request_counts[client_ip] = {"count": 1, "last_reset": now} else: request_counts[client_ip]["count"] += 1 if request_counts[client_ip]["count"] > limit: raise HTTPException(status_code=429, detail="请求过于频繁") @app.get("/api/data") async def get_data(request: Request): rate_limit(request) return {"data": "some data"}

5.2 WAF配置

# WAF规则配置 class WAF: def __init__(self): self.rules = [ {"pattern": r"SELECT.*FROM.*WHERE", "action": "block"}, {"pattern": r"UNION.*SELECT", "action": "block"}, {"pattern": r"<script.*>", "action": "sanitize"}, {"pattern": r"../", "action": "block"} ] def inspect_request(self, request): for rule in self.rules: if re.search(rule["pattern"], request, re.IGNORECASE): if rule["action"] == "block": raise ValueError("请求被WAF拦截") elif rule["action"] == "sanitize": request = re.sub(rule["pattern"], "", request) return request

六、日志与监控

6.1 安全日志记录

# 安全日志记录 import logging from pythonjsonlogger import jsonlogger class SecurityLogger: def __init__(self): self.logger = logging.getLogger("security") self.logger.setLevel(logging.INFO) handler = logging.FileHandler("security.log") formatter = jsonlogger.JsonFormatter( "%(asctime)s %(levelname)s %(event_type)s %(message)s" ) handler.setFormatter(formatter) self.logger.addHandler(handler) def log_login_attempt(self, user_id, success, ip_address): self.logger.info( f"Login attempt for user {user_id}", extra={ "event_type": "login_attempt", "user_id": user_id, "success": success, "ip_address": ip_address } ) def log_security_event(self, event_type, details): self.logger.info( f"Security event: {event_type}", extra={ "event_type": event_type, **details } )

6.2 异常检测

# 异常检测 import numpy as np class AnomalyDetector: def __init__(self): self.request_patterns = {} def record_request(self, client_ip, request_type, timestamp): if client_ip not in self.request_patterns: self.request_patterns[client_ip] = [] self.request_patterns[client_ip].append({ "type": request_type, "timestamp": timestamp }) def detect_anomaly(self, client_ip): patterns = self.request_patterns.get(client_ip, []) if len(patterns) < 10: return False timestamps = [p["timestamp"] for p in patterns[-10:]] intervals = np.diff(timestamps) # 检测请求频率异常 if np.mean(intervals) < 0.1: # 平均间隔小于100ms return True return False

七、实战案例：安全API设计

7.1 安全API实现

class SecureAPI: def __init__(self): self.jwt_authenticator = JWTAuthenticator("super_secret_key") self.security_logger = SecurityLogger() self.rate_limiter = RateLimiter() def authenticate(self, token): try: payload = self.jwt_authenticator.verify_token(token) return payload["user_id"] except Exception as e: self.security_logger.log_security_event("auth_failure", {"error": str(e)}) raise def protected_endpoint(self, request): # 1. 速率限制 self.rate_limiter.check(request.client.host) # 2. 身份认证 token = request.headers.get("Authorization", "").replace("Bearer ", "") user_id = self.authenticate(token) # 3. 参数验证 data = validate_request(request.json()) # 4. 业务逻辑 result = self.process_request(data) # 5. 日志记录 self.security_logger.log_security_event( "api_call", {"user_id": user_id, "endpoint": request.path} ) return result

7.2 安全审计

# 安全审计 class SecurityAuditor: def __init__(self): pass def audit_logs(self, start_time, end_time): # 分析安全日志 logs = self._load_logs(start_time, end_time) suspicious_activities = [] for log in logs: if log["event_type"] == "auth_failure": suspicious_activities.append(log) return { "total_logs": len(logs), "suspicious_count": len(suspicious_activities), "suspicious_details": suspicious_activities[:10] }