FastAPI 是一个现代的、快速的基于 Python 3.7+ 构建 API 的 web 框架,它基于标准的 Python 类型提示。虽然它提供了许多强大的功能,但保护你的 API 端点对于保护你的应用程序和数据至关重要。在这篇文章中,我们将探讨各种技术来保护你的 FastAPI 应用程序。
项目结构让我们从定义我们安全API应用的项目结构开始:
fastapi_security/
├── app/
│ ├── __init__.py
│ ├── main.py
│ ├── auth.py
│ ├── models.py
│ └── config.py
├── requirements.txt
└── README.md
1. 设置您的环境
首先,创建一个虚拟环境并安装 FastAPI 和 Uvicorn:
python -m venv venv
source venv/bin/activate # 在 Windows 系统中使用 `venv\Scripts\activate`
pip install fastapi uvicorn python-jose passlib bcrypt
2. JWT 认证
JSON Web Tokens (JWT) 是一种紧凑且 URL 安全的表示声明的格式,用于在两个实体之间传输。这里是如何在 FastAPI 中实现 JWT 认证。
示例:**config.py**
SECRET_KEY = "你的密钥"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
示例: **models.py**
从 pydantic 导入 BaseModel
从 typing 导入 Optional
class Token(BaseModel):
access_token: str
token_type: str
class TokenData(BaseModel):
username: Optional[str] = None
class User(BaseModel):
username: str
email: str
full_name: Optional[str] = None
disabled: Optional[bool] = None
class UserInDB(User):
hashed_password: str
示例: **auth.py**
from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from fastapi import Depends, HTTPException, status
from app.models import TokenData, UserInDB, User
from app.config import SECRET_KEY, ALGORITHM, ACCESS_TOKEN_EXPIRE_MINUTES
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
# 模拟数据库
fake_users_db = {
"johndoe": {
"username": "johndoe",
"full_name": "John Doe",
"email": "johndoe@example.com",
"hashed_password": pwd_context.hash("secret"),
"disabled": False,
}
}
def verify_password(plain_password, hashed_password):
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password):
return pwd_context.hash(password)
def get_user(db, username: str):
if username in db:
user_dict = db[username]
return UserInDB(**user_dict)
def authenticate_user(fake_db, username: str, password: str):
user = get_user(fake_db, username)
if not user:
return False
if not verify_password(password, user.hashed_password):
return False
return user
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=15)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
async def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="无法验证凭证",
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
token_data = TokenData(username=username)
except JWTError:
raise credentials_exception
user = get_user(fake_users_db, username=token_data.username)
if user is None:
raise credentials_exception
return user
async def get_current_active_user(current_user: User = Depends(get_current_user)):
if current_user.disabled:
raise HTTPException(status_code=400, detail="无效用户")
return current_user
3. 创建安全端点
示例:**main.py**
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from datetime import timedelta
from app.auth import authenticate_user, create_access_token, get_current_active_user
from app.models import Token, User
from app.config import ACCESS_TOKEN_EXPIRE_MINUTES
app = FastAPI()
@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
user = authenticate_user(fake_users_db, form_data.username, form_data.password)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="用户名或密码错误",
headers={"WWW-Authenticate": "Bearer"},
)
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
data={"sub": user.username}, expires_delta=access_token_expires
)
return {"access_token": access_token, "token_type": "bearer"}
@app.get("/users/me", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_active_user)):
return current_user
4. HTTPS 和 CORS 配置
为了进一步保护你的API,使用HTTPS进行所有通信并正确配置CORS(跨源资源共享)。
启用 HTTPS为了启用 HTTPS,你需要一个证书。你可以生成一个自签名证书用于测试,或者使用 Let’s Encrypt 等服务来获取一个免费且受信任的证书。
使用 Uvicorn 和 SSL 的示例
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000 --ssl-keyfile=path/to/key.pem --ssl-certfile=path/to/cert.pem
配置CORS
CORS 是必要的,如果你的 API 将会从运行在不同域名上的 web 应用程序中访问。FastAPI 使得设置 CORS 变得非常简单。
示例:**main.py**
(带有CORS)
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
origins = [
"http://localhost",
"http://localhost:8080",
"https://yourdomain.com",
]
app.add_middleware(
CORSMiddleware,
allow_origins=origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
结论
保护您的 FastAPI 应用程序涉及多层防护,从实现 JWT 认证到配置 HTTPS 和 CORS。通过采用这些技术,您可以显著增强 API 端点的安全性,确保您的应用程序及其数据得到充分保护。
如果你觉得这篇文章对你有帮助,不妨关注我,获取更多关于保护你的 FastAPI 应用程序和在 API 开发中实施最佳实践的见解。
放心地为您的API提供安全保障!
Stackademic 🎓感谢您读到最后。在离开之前:
- 请考虑鼓掌和关注作者! 👏
- 关注我们 X | LinkedIn | YouTube | Discord
- 访问我们的其他平台: In Plain English | CoFeed | Differ
- 更多内容请访问 Stackademic.com
共同學(xué)習(xí),寫下你的評論
評論加載中...
作者其他優(yōu)質(zhì)文章