探索 FastAPI 网络安全性关键策略保护你的应用免受常见网络攻击提升系统防御能力确保数据安全传输

威震华夏关云长

引言

FastAPI 是一个现代、高性能的 Web 框架，用于构建基于 Python 3.6+ 的 API，它具有出色的性能、易于使用和快速开发的特点。然而，随着网络攻击手段的不断演变，确保 FastAPI 应用的安全性变得至关重要。本文将深入探讨 FastAPI 中的网络安全关键策略，帮助你保护应用免受常见网络攻击，提升系统防御能力，并确保数据安全传输。

FastAPI 中的常见安全威胁

在深入安全策略之前，我们需要了解 FastAPI 应用可能面临的一些常见安全威胁：

1. 身份验证问题：弱密码、不安全的令牌管理和身份验证绕过
2. 授权漏洞：访问控制不当，允许用户访问未授权的资源
3. 数据泄露：敏感数据未经加密传输或存储
4. 注入攻击：SQL 注入、NoSQL 注入、命令注入等
5. 跨站脚本攻击 (XSS)：恶意脚本注入到网页中
6. 跨站请求伪造 (CSRF)：欺骗用户在已认证的网站上执行非预期操作
7. 跨域资源共享 (CORS) 配置错误：不当的 CORS 策略可能导致安全风险
8. 依赖项漏洞：使用存在已知漏洞的第三方库

FastAPI 安全基础设置

FastAPI 提供了一些基础的安全功能，可以通过简单的配置来增强应用的安全性。首先，让我们看一个基本的 FastAPI 应用，并逐步添加安全功能。

2. from fastapi import FastAPI
3. from fastapi.middleware.cors import CORSMiddleware
4. from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware
5. from fastapi.middleware.trustedhost import TrustedHostMiddleware
7. app = FastAPI()
9. # 强制使用 HTTPS
10. app.add_middleware(HTTPSRedirectMiddleware)
12. # 限制可访问的主机
13. app.add_middleware(
14.     TrustedHostMiddleware,
15.     allowed_hosts=["example.com", "*.example.com"]
16. )
18. # 配置 CORS
19. app.add_middleware(
20.     CORSMiddleware,
21.     allow_origins=["https://example.com"],
22.     allow_credentials=True,
23.     allow_methods=["*"],
24.     allow_headers=["*"],
25. )
27. @app.get("/")
28. def read_root():
29.     return {"message": "Hello World"}

复制代码

在上面的例子中，我们添加了三个中间件来增强安全性：

1. HTTPSRedirectMiddleware：强制所有请求使用 HTTPS
2. TrustedHostMiddleware：限制可访问的主机，防止 DNS 重新绑定攻击
3. CORSMiddleware：配置跨域资源共享策略

认证与授权策略

认证是验证用户身份的过程，而授权是确定已认证用户是否有权访问特定资源的过程。FastAPI 提供了灵活的安全工具来处理认证和授权。

OAuth2 密码流

OAuth2 是一个广泛使用的授权框架。以下是如何在 FastAPI 中实现 OAuth2 密码流的示例：

2. from fastapi import FastAPI, Depends, HTTPException, status
3. from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
4. from pydantic import BaseModel
5. from passlib.context import CryptContext
6. from datetime import datetime, timedelta
7. from typing import Optional
8. import jwt
10. app = FastAPI()
12. # 密码加密上下文
13. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
15. # OAuth2 密码流
16. oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
18. # 密钥和算法
19. SECRET_KEY = "your-secret-key"
20. ALGORITHM = "HS256"
21. ACCESS_TOKEN_EXPIRE_MINUTES = 30
23. # 用户模型
24. class User(BaseModel):
25.     username: str
26.     email: Optional[str] = None
27.     full_name: Optional[str] = None
28.     disabled: Optional[bool] = None
30. class UserInDB(User):
31.     hashed_password: str
33. # 模拟数据库
34. fake_users_db = {
35.     "johndoe": {
36.         "username": "johndoe",
37.         "full_name": "John Doe",
38.         "email": "johndoe@example.com",
39.         "hashed_password": pwd_context.hash("secret"),
40.         "disabled": False,
41.     }
42. }
44. # 验证密码
45. def verify_password(plain_password, hashed_password):
46.     return pwd_context.verify(plain_password, hashed_password)
48. # 获取用户
49. def get_user(db, username: str):
50.     if username in db:
51.         user_dict = db[username]
52.         return UserInDB(**user_dict)
54. # 认证用户
55. def authenticate_user(fake_db, username: str, password: str):
56.     user = get_user(fake_db, username)
57.     if not user:
58.         return False
59.     if not verify_password(password, user.hashed_password):
60.         return False
61.     return user
63. # 创建访问令牌
64. def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
65.     to_encode = data.copy()
66.     if expires_delta:
67.         expire = datetime.utcnow() + expires_delta
68.     else:
69.         expire = datetime.utcnow() + timedelta(minutes=15)
70.     to_encode.update({"exp": expire})
71.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
72.     return encoded_jwt
74. # 获取当前用户
75. async def get_current_user(token: str = Depends(oauth2_scheme)):
76.     credentials_exception = HTTPException(
77.         status_code=status.HTTP_401_UNAUTHORIZED,
78.         detail="Could not validate credentials",
79.         headers={"WWW-Authenticate": "Bearer"},
80.     )
81.     try:
82.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
83.         username: str = payload.get("sub")
84.         if username is None:
85.             raise credentials_exception
86.     except jwt.PyJWTError:
87.         raise credentials_exception
88.     user = get_user(fake_users_db, username=username)
89.     if user is None:
90.         raise credentials_exception
91.     return user
93. # 获取当前活跃用户
94. async def get_current_active_user(current_user: UserInDB = Depends(get_current_user)):
95.     if current_user.disabled:
96.         raise HTTPException(status_code=400, detail="Inactive user")
97.     return current_user
99. # 令牌端点
100. @app.post("/token")
101. async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
102.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
103.     if not user:
104.         raise HTTPException(
105.             status_code=status.HTTP_401_UNAUTHORIZED,
106.             detail="Incorrect username or password",
107.             headers={"WWW-Authenticate": "Bearer"},
108.         )
109.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
110.     access_token = create_access_token(
111.         data={"sub": user.username}, expires_delta=access_token_expires
112.     )
113.     return {"access_token": access_token, "token_type": "bearer"}
115. # 受保护的路由
116. @app.get("/users/me")
117. async def read_users_me(current_user: User = Depends(get_current_active_user)):
118.     return current_user

复制代码

这个示例实现了完整的 OAuth2 密码流，包括：

1. 密码哈希和验证
2. JWT 令牌的创建和验证
3. 用户认证和授权
4. 受保护的路由

API 密钥认证

对于简单的 API 认证，可以使用 API 密钥：

2. from fastapi import FastAPI, Depends, HTTPException, status
3. from fastapi.security import APIKeyHeader
5. app = FastAPI()
7. api_key_header = APIKeyHeader(name="X-API-KEY")
9. async def get_api_key(api_key: str = Depends(api_key_header)):
10.     if api_key != "your-secret-api-key":
11.         raise HTTPException(
12.             status_code=status.HTTP_401_UNAUTHORIZED,
13.             detail="Invalid API Key",
14.         )
15.     return api_key
17. @app.get("/protected-route")
18. async def protected_route(api_key: str = Depends(get_api_key)):
19.     return {"message": "This is a protected route"}

复制代码

数据加密与安全传输

HTTPS 配置

确保数据在传输过程中加密是至关重要的。在生产环境中，应始终使用 HTTPS。以下是如何在 FastAPI 中配置 HTTPS：

2. import uvicorn
3. from fastapi import FastAPI
5. app = FastAPI()
7. @app.get("/")
8. def read_root():
9.     return {"message": "Hello World"}
11. if __name__ == "__main__":
12.     uvicorn.run(
13.         "main:app",
14.         host="0.0.0.0",
15.         port=8000,
16.         ssl_keyfile="path/to/key.pem",
17.         ssl_certfile="path/to/cert.pem",
18.         reload=True
19.     )

复制代码

敏感数据加密

对于敏感数据，应该在存储前进行加密：

2. from cryptography.fernet import Fernet
4. # 生成密钥
5. key = Fernet.generate_key()
6. cipher_suite = Fernet(key)
8. # 加密数据
9. def encrypt_data(data: str) -> bytes:
10.     return cipher_suite.encrypt(data.encode())
12. # 解密数据
13. def decrypt_data(encrypted_data: bytes) -> str:
14.     return cipher_suite.decrypt(encrypted_data).decode()
16. # 使用示例
17. sensitive_data = "This is a secret message"
18. encrypted_data = encrypt_data(sensitive_data)
19. decrypted_data = decrypt_data(encrypted_data)
21. print(f"Original: {sensitive_data}")
22. print(f"Encrypted: {encrypted_data}")
23. print(f"Decrypted: {decrypted_data}")

复制代码

输入验证与 SQL 注入防护

FastAPI 使用 Pydantic 模型进行数据验证，这有助于防止许多类型的注入攻击。以下是如何安全地处理用户输入：

2. from fastapi import FastAPI, HTTPException, Depends
3. from pydantic import BaseModel, constr
4. from typing import Optional
5. import databases
6. import sqlalchemy
7. from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String
9. # 数据库配置
10. DATABASE_URL = "sqlite:///./test.db"
11. database = databases.Database(DATABASE_URL)
12. metadata = MetaData()
14. users = Table(
15.     "users",
16.     metadata,
17.     Column("id", Integer, primary_key=True),
18.     Column("name", String(50)),
19.     Column("email", String(50)),
20. )
22. engine = create_engine(DATABASE_URL)
23. metadata.create_all(engine)
25. app = FastAPI()
27. # 用户模型
28. class UserBase(BaseModel):
29.     name: constr(min_length=1, max_length=50)
30.     email: constr(min_length=1, max_length=50)
32. class UserCreate(UserBase):
33.     pass
35. class User(UserBase):
36.     id: int
37.    
38.     class Config:
39.         orm_mode = True
41. # 创建用户
42. @app.post("/users/", response_model=User)
43. async def create_user(user: UserCreate):
44.     query = users.insert().values(name=user.name, email=user.email)
45.     last_record_id = await database.execute(query)
46.     return {**user.dict(), "id": last_record_id}
48. # 获取用户
49. @app.get("/users/{user_id}", response_model=User)
50. async def read_user(user_id: int):
51.     query = users.select().where(users.c.id == user_id)
52.     user = await database.fetch_one(query)
53.     if user is None:
54.         raise HTTPException(status_code=404, detail="User not found")
55.     return user
57. # 安全搜索用户
58. @app.get("/users/", response_model=list[User])
59. async def search_users(name: Optional[str] = None, email: Optional[str] = None):
60.     query = users.select()
61.     if name:
62.         query = query.where(users.c.name == name)
63.     if email:
64.         query = query.where(users.c.email == email)
65.     results = await database.fetch_all(query)
66.     return results

复制代码

在这个例子中，我们：

1. 使用 Pydantic 模型验证输入数据
2. 使用参数化查询防止 SQL 注入
3. 对输入数据进行长度限制

CORS 与 CSRF 防护

CORS 配置

跨域资源共享 (CORS) 是一种机制，允许 Web 应用服务器进行跨域访问控制。以下是如何在 FastAPI 中安全地配置 CORS：

2. from fastapi import FastAPI
3. from fastapi.middleware.cors import CORSMiddleware
5. app = FastAPI()
7. # 安全的 CORS 配置
8. app.add_middleware(
9.     CORSMiddleware,
10.     allow_origins=["https://example.com", "https://www.example.com"],  # 只允许特定域名
11.     allow_credentials=True,
12.     allow_methods=["GET", "POST", "PUT", "DELETE"],  # 只允许特定方法
13.     allow_headers=["Authorization", "Content-Type"],  # 只允许特定头部
14.     max_age=600,  # 预检请求结果的缓存时间
15. )

复制代码

CSRF 防护

跨站请求伪造 (CSRF) 是一种攻击，迫使最终用户在当前已认证的 Web 应用程序上执行非预期的操作。以下是如何在 FastAPI 中实现 CSRF 防护：

2. from fastapi import FastAPI, Depends, HTTPException, Request, Response
3. from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
4. from secrets import token_hex
5. import uuid
7. app = FastAPI()
9. security = HTTPBearer()
11. # 存储 CSRF 令牌的简单字典
12. csrf_tokens = {}
14. # 生成 CSRF 令牌
15. def generate_csrf_token():
16.     return token_hex(32)
18. # 验证 CSRF 令牌
19. def verify_csrf_token(request: Request, csrf_token: str):
20.     # 从请求头获取 CSRF 令牌
21.     header_token = request.headers.get("X-CSRF-Token")
22.    
23.     # 比较令牌
24.     if not header_token or header_token != csrf_token:
25.         raise HTTPException(status_code=403, detail="CSRF token validation failed")
27. # 获取 CSRF 令牌
28. @app.get("/csrf-token")
29. async def get_csrf_token(request: Request):
30.     csrf_token = generate_csrf_token()
31.     session_id = str(uuid.uuid4())
32.     csrf_tokens[session_id] = csrf_token
33.    
34.     # 设置包含 CSRF 令牌的 cookie
35.     response = Response(content={"csrf_token": csrf_token})
36.     response.set_cookie(
37.         key="csrf_token",
38.         value=csrf_token,
39.         httponly=False,  # 允许 JavaScript 访问
40.         secure=True,     # 只通过 HTTPS 发送
41.         samesite="strict"  # 严格的同源策略
42.     )
43.     return response
45. # 受保护的路由
46. @app.post("/protected-route")
47. async def protected_route(
48.     request: Request,
49.     credentials: HTTPAuthorizationCredentials = Depends(security)
50. ):
51.     # 从 cookie 获取 CSRF 令牌
52.     csrf_token = request.cookies.get("csrf_token")
53.     if not csrf_token:
54.         raise HTTPException(status_code=403, detail="CSRF token not found")
55.    
56.     # 验证 CSRF 令牌
57.     verify_csrf_token(request, csrf_token)
58.    
59.     return {"message": "CSRF validation passed"}

复制代码

安全 Headers 设置

安全 HTTP 头可以帮助防止某些类型的攻击，如点击劫持、MIME 类型混淆等。以下是如何在 FastAPI 中设置安全头：

2. from fastapi import FastAPI, Response
3. from fastapi.middleware.gzip import GZipMiddleware
5. app = FastAPI()
7. # 添加 GZip 压缩中间件
8. app.add_middleware(GZipMiddleware, minimum_size=1000)
10. @app.middleware("http")
11. async def add_security_headers(request, call_next):
12.     response = await call_next(request)
13.    
14.     # 添加安全头
15.     response.headers["Strict-Transport-Security"] = "max-age=63072000; includeSubDomains; preload"
16.     response.headers["X-Content-Type-Options"] = "nosniff"
17.     response.headers["X-Frame-Options"] = "DENY"
18.     response.headers["X-XSS-Protection"] = "1; mode=block"
19.     response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
20.     response.headers["Content-Security-Policy"] = "default-src 'self'; script-src 'self' 'unsafe-inline' 'unsafe-eval'; style-src 'self' 'unsafe-inline'; img-src 'self' data:; font-src 'self'; connect-src 'self'; frame-ancestors 'none';"
21.     response.headers["Permissions-Policy"] = "camera=(), microphone=(), geolocation=(), payment=()"
22.    
23.     return response
25. @app.get("/")
26. def read_root():
27.     return {"message": "Hello World"}

复制代码

这些安全头的含义：

1. Strict-Transport-Security：强制浏览器使用 HTTPS
2. X-Content-Type-Options：防止 MIME 类型混淆攻击
3. X-Frame-Options：防止点击劫持攻击
4. X-XSS-Protection：启用浏览器内置的 XSS 过滤器
5. Referrer-Policy：控制 referrer 信息
6. Content-Security-Policy：防止代码注入攻击
7. Permissions-Policy：控制浏览器功能和 API 的访问权限

依赖注入与中间件的安全应用

FastAPI 的依赖注入系统可以用于实现安全功能，如认证、授权和输入验证。以下是一些示例：

安全依赖项

2. from fastapi import FastAPI, Depends, HTTPException, status
3. from fastapi.security import OAuth2PasswordBearer
4. from typing import List
6. app = FastAPI()
8. oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
10. # 模拟用户数据库
11. fake_users_db = {
12.     "alice": {"username": "alice", "role": "admin"},
13.     "bob": {"username": "bob", "role": "user"},
14. }
16. # 获取当前用户
17. async def get_current_user(token: str = Depends(oauth2_scheme)):
18.     # 这里应该验证 token 并获取用户信息
19.     # 简化示例，直接使用 token 作为用户名
20.     if token not in fake_users_db:
21.         raise HTTPException(
22.             status_code=status.HTTP_401_UNAUTHORIZED,
23.             detail="Invalid authentication credentials",
24.             headers={"WWW-Authenticate": "Bearer"},
25.         )
26.     return fake_users_db[token]
28. # 检查用户是否为管理员
29. async def get_current_admin(current_user: dict = Depends(get_current_user)):
30.     if current_user["role"] != "admin":
31.         raise HTTPException(
32.             status_code=status.HTTP_403_FORBIDDEN,
33.             detail="Not enough permissions"
34.         )
35.     return current_user
37. # 普通用户路由
38. @app.get("/user-profile")
39. async def user_profile(current_user: dict = Depends(get_current_user)):
40.     return {"username": current_user["username"], "role": current_user["role"]}
42. # 管理员路由
43. @app.get("/admin-panel")
44. async def admin_panel(current_admin: dict = Depends(get_current_admin)):
45.     return {"message": f"Welcome, {current_admin['username']}! You have admin access."}

复制代码

安全中间件

2. from fastapi import FastAPI, Request, HTTPException
3. from fastapi.middleware import Middleware
4. from fastapi.middleware.base import BaseHTTPMiddleware
5. import time
7. app = FastAPI()
9. # 速率限制中间件
10. class RateLimitMiddleware(BaseHTTPMiddleware):
11.     def __init__(self, app, rate_limit: int = 100, time_window: int = 60):
12.         super().__init__(app)
13.         self.rate_limit = rate_limit
14.         self.time_window = time_window
15.         self.requests = {}
17.     async def dispatch(self, request: Request, call_next):
18.         client_ip = request.client.host
19.         current_time = time.time()
20.         
21.         # 清理过期的请求记录
22.         if client_ip in self.requests:
23.             self.requests[client_ip] = [
24.                 req_time for req_time in self.requests[client_ip]
25.                 if current_time - req_time < self.time_window
26.             ]
27.         else:
28.             self.requests[client_ip] = []
29.         
30.         # 检查是否超过速率限制
31.         if len(self.requests[client_ip]) >= self.rate_limit:
32.             raise HTTPException(status_code=429, detail="Too many requests")
33.         
34.         # 记录当前请求
35.         self.requests[client_ip].append(current_time)
36.         
37.         response = await call_next(request)
38.         return response
40. # 添加速率限制中间件
41. app.add_middleware(RateLimitMiddleware, rate_limit=10, time_window=60)
43. @app.get("/")
44. async def read_root():
45.     return {"message": "Hello World"}

复制代码

日志记录与监控

安全日志记录和监控对于检测和响应安全事件至关重要。以下是如何在 FastAPI 中实现安全日志记录：

2. from fastapi import FastAPI, Request, HTTPException
3. from fastapi.responses import JSONResponse
4. import logging
5. from logging.handlers import RotatingFileHandler
6. import time
7. import traceback
9. app = FastAPI()
11. # 配置日志
12. logging.basicConfig(
13.     level=logging.INFO,
14.     format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
15.     handlers=[
16.         RotatingFileHandler('app.log', maxBytes=1024*1024, backupCount=5),
17.         logging.StreamHandler()
18.     ]
19. )
20. logger = logging.getLogger(__name__)
22. # 请求日志中间件
23. @app.middleware("http")
24. async def log_requests(request: Request, call_next):
25.     start_time = time.time()
26.    
27.     # 记录请求信息
28.     logger.info(f"Request: {request.method} {request.url}")
29.    
30.     try:
31.         response = await call_next(request)
32.         
33.         # 记录响应信息
34.         process_time = time.time() - start_time
35.         logger.info(f"Response: {response.status_code} - Processed in {process_time:.4f}s")
36.         
37.         return response
38.     except Exception as e:
39.         # 记录异常信息
40.         logger.error(f"Exception: {str(e)}")
41.         logger.error(traceback.format_exc())
42.         
43.         # 返回错误响应
44.         return JSONResponse(
45.             status_code=500,
46.             content={"message": "Internal Server Error"}
47.         )
49. # 安全事件日志
50. @app.exception_handler(HTTPException)
51. async def http_exception_handler(request: Request, exc: HTTPException):
52.     # 记录安全相关事件
53.     if exc.status_code in [401, 403, 404]:
54.         logger.warning(f"Security event: {exc.status_code} - {exc.detail} - IP: {request.client.host}")
55.    
56.     return JSONResponse(
57.         status_code=exc.status_code,
58.         content={"message": exc.detail}
59.     )
61. @app.get("/")
62. async def read_root():
63.     logger.info("Root endpoint accessed")
64.     return {"message": "Hello World"}
66. @app.get("/secure")
67. async def secure_endpoint():
68.     logger.info("Secure endpoint accessed")
69.     return {"message": "This is a secure endpoint"}

复制代码

安全测试方法

安全测试是确保 FastAPI 应用安全的重要环节。以下是一些常见的安全测试方法和工具：

单元测试

2. import pytest
3. from fastapi.testclient import TestClient
4. from main import app
6. client = TestClient(app)
8. def test_root_endpoint():
9.     response = client.get("/")
10.     assert response.status_code == 200
11.     assert response.json() == {"message": "Hello World"}
13. def test_secure_endpoint_without_auth():
14.     response = client.get("/secure")
15.     assert response.status_code == 401
17. def test_secure_endpoint_with_auth():
18.     # 获取令牌
19.     response = client.post(
20.         "/token",
21.         data={"username": "johndoe", "password": "secret"}
22.     )
23.     assert response.status_code == 200
24.     token = response.json()["access_token"]
25.    
26.     # 使用令牌访问安全端点
27.     response = client.get(
28.         "/secure",
29.         headers={"Authorization": f"Bearer {token}"}
30.     )
31.     assert response.status_code == 200
32.     assert response.json() == {"message": "This is a secure endpoint"}

复制代码

集成测试

2. import pytest
3. from fastapi.testclient import TestClient
4. from main import app
6. client = TestClient(app)
8. def test_sql_injection_protection():
9.     # 尝试 SQL 注入攻击
10.     malicious_input = "'; DROP TABLE users; --"
11.     response = client.post(
12.         "/users/",
13.         json={"name": malicious_input, "email": "test@example.com"}
14.     )
15.     # 应用应该正确处理输入，而不是执行 SQL 注入
16.     assert response.status_code == 422  # 验证错误
18. def test_xss_protection():
19.     # 尝试 XSS 攻击
20.     malicious_input = "<script>alert('XSS')</script>"
21.     response = client.post(
22.         "/users/",
23.         json={"name": malicious_input, "email": "test@example.com"}
24.     )
25.     # 应用应该正确处理输入，而不是执行脚本
26.     assert response.status_code == 422  # 验证错误

复制代码

使用安全测试工具

1. OWASP ZAP：开源的 Web 应用安全扫描器
2. Burp Suite：用于测试 Web 应用安全的集成平台
3. SQLMap：自动化的 SQL 注入检测工具
4. Bandit：Python 代码安全漏洞扫描器

以下是使用 Bandit 扫描 FastAPI 应用的示例：

2. # 安装 Bandit
3. pip install bandit
5. # 扫描 FastAPI 应用
6. bandit -r ./

复制代码

最佳实践与安全清单

以下是一些 FastAPI 安全最佳实践和安全清单：

安全清单

1. 认证与授权[ ] 使用强身份验证机制（如 OAuth2）[ ] 实现适当的访问控制[ ] 定期轮换密钥和令牌[ ] 使用安全的密码存储（如 bcrypt）
2. [ ] 使用强身份验证机制（如 OAuth2）
3. [ ] 实现适当的访问控制
4. [ ] 定期轮换密钥和令牌
5. [ ] 使用安全的密码存储（如 bcrypt）
6. 数据保护[ ] 使用 HTTPS 加密所有通信[ ] 加密敏感数据存储[ ] 实现适当的日志记录和监控[ ] 定期备份数据
7. [ ] 使用 HTTPS 加密所有通信
8. [ ] 加密敏感数据存储
9. [ ] 实现适当的日志记录和监控
10. [ ] 定期备份数据
11. 输入验证[ ] 验证所有用户输入[ ] 使用参数化查询防止 SQL 注入[ ] 限制文件上传类型和大小[ ] 实现适当的错误处理
12. [ ] 验证所有用户输入
13. [ ] 使用参数化查询防止 SQL 注入
14. [ ] 限制文件上传类型和大小
15. [ ] 实现适当的错误处理
16. 安全配置[ ] 配置适当的安全头[ ] 实现速率限制[ ] 禁用不必要的服务和功能[ ] 保持依赖项更新
17. [ ] 配置适当的安全头
18. [ ] 实现速率限制
19. [ ] 禁用不必要的服务和功能
20. [ ] 保持依赖项更新
21. 测试与监控[ ] 定期进行安全测试[ ] 实施安全监控和警报[ ] 定期审查日志[ ] 制定安全事件响应计划
22. [ ] 定期进行安全测试
23. [ ] 实施安全监控和警报
24. [ ] 定期审查日志
25. [ ] 制定安全事件响应计划

认证与授权

• [ ] 使用强身份验证机制（如 OAuth2）
• [ ] 实现适当的访问控制
• [ ] 定期轮换密钥和令牌
• [ ] 使用安全的密码存储（如 bcrypt）

数据保护

• [ ] 使用 HTTPS 加密所有通信
• [ ] 加密敏感数据存储
• [ ] 实现适当的日志记录和监控
• [ ] 定期备份数据

输入验证

• [ ] 验证所有用户输入
• [ ] 使用参数化查询防止 SQL 注入
• [ ] 限制文件上传类型和大小
• [ ] 实现适当的错误处理

安全配置

• [ ] 配置适当的安全头
• [ ] 实现速率限制
• [ ] 禁用不必要的服务和功能
• [ ] 保持依赖项更新

测试与监控

• [ ] 定期进行安全测试
• [ ] 实施安全监控和警报
• [ ] 定期审查日志
• [ ] 制定安全事件响应计划

安全最佳实践示例

2. from fastapi import FastAPI, Depends, HTTPException, status
3. from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
4. from fastapi.middleware.cors import CORSMiddleware
5. from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware
6. from fastapi.middleware.trustedhost import TrustedHostMiddleware
7. from fastapi.middleware.gzip import GZipMiddleware
8. from pydantic import BaseModel, constr, EmailStr
9. from passlib.context import CryptContext
10. from datetime import datetime, timedelta
11. from typing import Optional
12. import jwt
13. import secrets
14. import logging
15. from logging.handlers import RotatingFileHandler
17. app = FastAPI()
19. # 配置日志
20. logging.basicConfig(
21.     level=logging.INFO,
22.     format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
23.     handlers=[
24.         RotatingFileHandler('app.log', maxBytes=1024*1024, backupCount=5),
25.         logging.StreamHandler()
26.     ]
27. )
28. logger = logging.getLogger(__name__)
30. # 添加安全中间件
31. app.add_middleware(HTTPSRedirectMiddleware)
32. app.add_middleware(TrustedHostMiddleware, allowed_hosts=["example.com"])
33. app.add_middleware(GZipMiddleware, minimum_size=1000)
34. app.add_middleware(
35.     CORSMiddleware,
36.     allow_origins=["https://example.com"],
37.     allow_credentials=True,
38.     allow_methods=["GET", "POST", "PUT", "DELETE"],
39.     allow_headers=["Authorization", "Content-Type"],
40.     max_age=600,
41. )
43. # 安全配置
44. SECRET_KEY = secrets.token_urlsafe(32)
45. ALGORITHM = "HS256"
46. ACCESS_TOKEN_EXPIRE_MINUTES = 30
48. # 密码加密上下文
49. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
51. # OAuth2 密码流
52. oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
54. # 用户模型
55. class UserBase(BaseModel):
56.     username: constr(min_length=3, max_length=20)
57.     email: EmailStr
59. class UserCreate(UserBase):
60.     password: constr(min_length=8)
62. class User(UserBase):
63.     id: int
64.     is_active: bool = True
65.    
66.     class Config:
67.         orm_mode = True
69. # 模拟数据库
70. fake_users_db = {}
72. # 工具函数
73. def verify_password(plain_password, hashed_password):
74.     return pwd_context.verify(plain_password, hashed_password)
76. def get_password_hash(password):
77.     return pwd_context.hash(password)
79. def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
80.     to_encode = data.copy()
81.     if expires_delta:
82.         expire = datetime.utcnow() + expires_delta
83.     else:
84.         expire = datetime.utcnow() + timedelta(minutes=15)
85.     to_encode.update({"exp": expire})
86.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
87.     return encoded_jwt
89. # 认证和授权
90. async def get_current_user(token: str = Depends(oauth2_scheme)):
91.     credentials_exception = HTTPException(
92.         status_code=status.HTTP_401_UNAUTHORIZED,
93.         detail="Could not validate credentials",
94.         headers={"WWW-Authenticate": "Bearer"},
95.     )
96.     try:
97.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
98.         username: str = payload.get("sub")
99.         if username is None:
100.             raise credentials_exception
101.     except jwt.PyJWTError:
102.         raise credentials_exception
103.    
104.     user = fake_users_db.get(username)
105.     if user is None:
106.         raise credentials_exception
107.    
108.     return user
110. async def get_current_active_user(current_user: User = Depends(get_current_user)):
111.     if not current_user.get("is_active", True):
112.         raise HTTPException(status_code=400, detail="Inactive user")
113.     return current_user
115. # 端点
116. @app.post("/token")
117. async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
118.     user = fake_users_db.get(form_data.username)
119.     if not user or not verify_password(form_data.password, user["hashed_password"]):
120.         logger.warning(f"Failed login attempt for username: {form_data.username}")
121.         raise HTTPException(
122.             status_code=status.HTTP_401_UNAUTHORIZED,
123.             detail="Incorrect username or password",
124.             headers={"WWW-Authenticate": "Bearer"},
125.         )
126.    
127.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
128.     access_token = create_access_token(
129.         data={"sub": user["username"]}, expires_delta=access_token_expires
130.     )
131.    
132.     logger.info(f"User {form_data.username} logged in successfully")
133.     return {"access_token": access_token, "token_type": "bearer"}
135. @app.post("/users/", response_model=User)
136. async def create_user(user: UserCreate):
137.     if user.username in fake_users_db:
138.         raise HTTPException(
139.             status_code=400,
140.             detail="Username already registered"
141.         )
142.    
143.     hashed_password = get_password_hash(user.password)
144.     user_id = len(fake_users_db) + 1
145.    
146.     fake_users_db[user.username] = {
147.         "id": user_id,
148.         "username": user.username,
149.         "email": user.email,
150.         "hashed_password": hashed_password,
151.         "is_active": True,
152.     }
153.    
154.     logger.info(f"New user created: {user.username}")
155.     return fake_users_db[user.username]
157. @app.get("/users/me", response_model=User)
158. async def read_users_me(current_user: User = Depends(get_current_active_user)):
159.     return current_user
161. # 安全中间件
162. @app.middleware("http")
163. async def add_security_headers(request, call_next):
164.     response = await call_next(request)
165.    
166.     # 添加安全头
167.     response.headers["Strict-Transport-Security"] = "max-age=63072000; includeSubDomains; preload"
168.     response.headers["X-Content-Type-Options"] = "nosniff"
169.     response.headers["X-Frame-Options"] = "DENY"
170.     response.headers["X-XSS-Protection"] = "1; mode=block"
171.     response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
172.     response.headers["Content-Security-Policy"] = "default-src 'self'; script-src 'self' 'unsafe-inline' 'unsafe-eval'; style-src 'self' 'unsafe-inline'; img-src 'self' data:; font-src 'self'; connect-src 'self'; frame-ancestors 'none';"
173.     response.headers["Permissions-Policy"] = "camera=(), microphone=(), geolocation=(), payment=()"
174.    
175.     return response

复制代码

结论

FastAPI 提供了丰富的工具和功能来构建安全的 Web 应用程序。通过实施适当的安全策略，如强身份验证和授权机制、数据加密、输入验证、安全配置和持续监控，你可以显著提高 FastAPI 应用的安全性，保护其免受常见网络攻击，并确保数据安全传输。

记住，安全是一个持续的过程，而不是一次性的任务。定期审查和更新你的安全策略，保持对最新安全威胁的了解，并采用最佳实践，是确保你的 FastAPI 应用长期安全的关键。

通过本文介绍的技术和策略，你可以构建一个既强大又安全的 FastAPI 应用，为用户提供安全可靠的服务。