跳到主要内容

安全认证

前言​

安全认证分两个概念：

  • 认证（Authentication）：验证"你是谁"（用户名密码、Token 等）。
  • 授权（Authorization）：认证通过后，判断"你能访问什么"。
注意

OAuth 2 是授权协议而非认证协议；常见的"登录"流程是 OAuth 2 + JWT 的组合。

本章整理 FastAPI 中常用的认证鉴权方案：HTTPBasic、APIKey、OAuth 2（客户端模式 + 密码模式）+ JWT。

HTTPBasic​

HTTP Basic 是最基础的 HTTP 认证方式：客户端在 Authorization 头中携带 base64 编码的 用户名:密码。使用简单，但因为 base64 可逆，明文传输易被破解，生产环境一般配合 HTTPS 使用。

import secrets

from fastapi import APIRouter, Depends
from fastapi.responses import PlainTextResponse
from fastapi.security import HTTPBasic, HTTPBasicCredentials

router = APIRouter(prefix="/chapter9", tags=["安全认证"])

security = HTTPBasic()


@router.get("/login1")
async def login(credentials: HTTPBasicCredentials = Depends(security)):
    # 用 secrets.compare_digest 做恒定时间比较，避免时序攻击
    username_ok = secrets.compare_digest(credentials.username, "zhangsan")
    password_ok = secrets.compare_digest(credentials.password, "123456")
    if username_ok and password_ok:
        return PlainTextResponse(status_code=200, content="login success")
    return PlainTextResponse(status_code=401, content="login failed")

APIKey​

APIKey 鉴权是基于固定 key 值比对的方式。FastAPI 提供三种来源：APIKeyHeader、APIKeyQuery、APIKeyCookie。

from fastapi import APIRouter, Depends, Request, Security
from fastapi.exceptions import HTTPException
from fastapi.security import APIKeyHeader, APIKeyQuery

router = APIRouter(prefix="/chapter9", tags=["安全认证"])


class APIKey:
    API_KEY_HEADER = "XTOKEN"
    API_KEY_HEADER_NAME = "X-TOKEN"
    api_key_header_token = APIKeyHeader(
        name=API_KEY_HEADER_NAME, scheme_name="API Key Header", auto_error=True
    )

    API_KEY_QUERY = "XQUERY"
    API_KEY_QUERY_NAME = "X-QUERY"
    api_key_query_token = APIKeyQuery(
        name=API_KEY_QUERY_NAME, scheme_name="API Key Query", auto_error=True
    )

    async def __call__(
        self,
        request: Request,
        api_key_header: str = Security(api_key_header_token),
        api_key_query: str = Security(api_key_query_token),
    ):
        if api_key_header != self.API_KEY_HEADER:
            raise HTTPException(status_code=401, detail="API Key Header Error")
        if api_key_query != self.API_KEY_QUERY:
            raise HTTPException(status_code=401, detail="API Key Query Error")
        return True


apikeyauth = APIKey()


@router.get("/login2")
async def login2(request: Request, auth: bool = Depends(apikeyauth)):
    if auth:
        return PlainTextResponse(status_code=200, content="login success")
    return PlainTextResponse(status_code=401, content="login failed")

简单动态 API Key​

某些 API 需要鉴权但又不想引入数据库、OAuth2 等复杂机制时，可以用 API Key；担心泄露，可以按算法动态生成 key。这里做个 demo：按"年月日时 + 固定字符串"生成 key，一个 key 过整点就失效。

import hashlib
from datetime import datetime, timedelta, timezone

import uvicorn
from fastapi import Depends, FastAPI, Request, Security
from fastapi.exceptions import HTTPException
from fastapi.responses import PlainTextResponse
from fastapi.security import APIKeyHeader

app = FastAPI()


def calculate_hash(src: str, alg: str = "md5"):
    src_enc = src.encode("utf-8")
    if alg == "sha256":
        return hashlib.sha256(src_enc).hexdigest()
    return hashlib.md5(src_enc).hexdigest()


class APIKey:
    API_KEY_HEADER_NAME = "X-TOKEN"
    api_key_header_token = APIKeyHeader(
        name=API_KEY_HEADER_NAME, scheme_name="API Key Header", auto_error=True
    )

    async def __call__(
        self,
        request: Request,
        api_key_header: str = Security(api_key_header_token),
    ):
        # 东八区当前小时，整点后 key 自动失效
        dt_ymdh = datetime.now(tz=timezone(timedelta(hours=8))).strftime("%Y%m%d%H")
        # 先按 md5 算法，再按 sha256 算法
        secret = calculate_hash(calculate_hash(f"{dt_ymdh}_1234qwerASDF"), alg="sha256")
        if api_key_header != secret:
            raise HTTPException(status_code=401, detail="API Key Header Error")
        return True


apikeyauth = APIKey()


@app.get("/login")
async def login(request: Request, auth: bool = Depends(apikeyauth)):
    if auth:
        return PlainTextResponse(status_code=200, content="login success")
    return PlainTextResponse(status_code=401, content="login failed")

测试：

# 1. 用命令行生成 key
echo -en "$(date +%Y%m%d%H)_1234qwerASDF" | md5sum | awk '{printf $1}' | sha256sum | awk '{print $1}'

# 2. 请求
curl 'http://127.0.0.1:8000/login' -H 'accept: application/json' -H 'X-TOKEN: f569b0c11f9ab85946a674eddbb4a111aaa52aab81d1815d26292612314256e7'

OAuth 2 与 JWT​

OAuth 是一种开放协议，允许用户让第三方应用以安全且标准的方式获取其在某网站/应用上存储的受保护资源。OAuth 2 与 OAuth 1 不兼容。

注意：OAuth 2 是授权协议，不是认证协议。认证验证身份；授权在认证后判断可访问的资源范围。常见方案是 OAuth 2 + JWT：服务端先签发 token，客户端携带 token 访问受保护资源。

OAuth 2 优点：

  • 避免在授权给第三方应用时泄露用户信息。
  • 结合 token 机制，可设置访问范围和有效期。

JWT​

JWT 是一个字符串，由 Header、Payload、Signature 三部分组成，每部分经过 base64url 编码，签名保证内容未被篡改。

FastAPI 官方推荐使用 JOSE 规范的 python-jose 库生成 JWT：

python -m pip install python-jose
from datetime import datetime, timedelta, timezone

from jose import jwt

SECRET_KEY = "qwerasdf"
ALGORITHM = "HS256"


class TokenUtils:
    @staticmethod
    def token_encode(data: dict):
        return jwt.encode(data, SECRET_KEY, algorithm=ALGORITHM)

    @staticmethod
    def token_decode(token: str):
        return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])


if __name__ == "__main__":
    data = {
        "iss": "zhangsan",  # issuer, token 颁发者
        "sub": "1",         # subject, token 的主题（用户标识）
        "aud": "lisi",      # audience, token 的接收者
        "iat": datetime.now(timezone.utc),  # issued at, 创建时间
        "nbf": datetime.now(timezone.utc),  # not before, 生效时间
        "exp": datetime.now(timezone.utc) + timedelta(minutes=15),  # 过期时间
        "jti": "1234567890",  # JWT ID
    }
    token = TokenUtils.token_encode(data)
    print(token)
    print(TokenUtils.token_decode(token))
备注

datetime.utcnow() 已废弃，统一使用 datetime.now(timezone.utc)。

OAuth 2 参数说明​

授权过程中 OAuth 2 规范需要携带指定配置参数：

请求参数​

参数名描述
client_id第三方应用的 ID，通常在开放平台申请后分配，URL 中必填
client_secret第三方应用的密钥，申请授权时必填
username参与授权的用户主体用户名
password参与授权的用户主体密码
response_type授权类型，授权码模式中一般固定为 token/code
grant_type授权方式，用于授权码模式和密码模式
redirect_uri授权完成后重定向的客户端 URL
scope申请的权限范围
state客户端状态，授权服务器原样返回，用于辨识请求来源

响应参数​

参数名描述
code授权服务器产生的临时随机码，一般只能用一次，用于换取 access_token
access_token客户端访问资源服务器的凭据
refresh_tokenaccess_token 过期后用于刷新续期，一般在授权码模式中返回
token_typeaccess_token 的类型（如 bearer）
expires_inaccess_token 的过期时间（秒）
scope权限范围
state与请求中的 state 对应，原样返回

OAuth 2 主体角色​

  • 资源所有者（Resource Owner）：又称用户。
  • 用户代理（User Agent）：通常是浏览器、APP。
  • 客户端（Client）：申请资源时使用的应用程序。
  • 授权服务器（Authorization Server）：发放 access_token 的服务端。
  • 资源服务器（Resource Server）：托管受保护资源的服务器。

中小型应用中授权服务器和资源服务器通常在同一服务内；一个授权服务器可以颁发多个资源服务器可接受的令牌。

根据参与角色不同，授权模式主要分为：客户端模式（Client Credentials）、密码模式（Password）、授权码模式（Authorization Code）、简化模式（Implicit）。

OAuth 2 客户端模式​

客户端模式适用于服务间调用（没有用户参与）：客户端拿自己的 client_id + client_secret 换 token。

备注

OAuth2 的 token 端点是表单提交（application/x-www-form-urlencoded），需要先安装 python-multipart。

from datetime import datetime, timedelta, timezone
from typing import Optional

from fastapi import APIRouter, Depends, Form, Request, status
from fastapi.exceptions import HTTPException
from fastapi.openapi.models import OAuthFlows as OAuthFlowsModel
from fastapi.security import OAuth2
from fastapi.security.utils import get_authorization_scheme_param
from jose import JWTError, jwt
from pydantic import ValidationError

router = APIRouter(prefix="/chapter9", tags=["安全认证"])

fake_client_db = {
    "zhangsan": {
        "client_id": "zhangsan",
        "client_secret": "123456",
    }
}

SECRET_KEY = "yJk48ijYXHVjSvgGosj6"
ALGORITHM = "HS256"


class TokenUtils:
    @staticmethod
    def token_encode(data: dict):
        return jwt.encode(data, SECRET_KEY, algorithm=ALGORITHM)

    @staticmethod
    def token_decode(token: str):
        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])
        except (JWTError, ValidationError):
            raise credentials_exception
        return payload


class OAuth2ClientCredentialsBearer(OAuth2):
    def __init__(
        self,
        tokenUrl: str,
        scheme_name: Optional[str] = None,
        scopes: Optional[dict[str, str]] = None,
        description: Optional[str] = None,
        auto_error: bool = True,
    ):
        if not scopes:
            scopes = {}
        flows = OAuthFlowsModel(
            clientCredentials={
                "tokenUrl": tokenUrl,
                "scopes": scopes,
            }
        )
        super().__init__(
            flows=flows,
            scheme_name=scheme_name,
            description=description,
            auto_error=auto_error,
        )

    async def __call__(self, request: Request) -> Optional[str]:
        authorization: str = request.headers.get("Authorization")
        scheme, param = get_authorization_scheme_param(authorization)
        if not authorization or scheme.lower() != "bearer":
            if self.auto_error:
                raise HTTPException(
                    status_code=status.HTTP_401_UNAUTHORIZED,
                    detail="Not authenticated",
                    headers={"WWW-Authenticate": "Bearer"},
                )
            return None
        return param


oauth2_scheme = OAuth2ClientCredentialsBearer(tokenUrl="/chapter9/oauth2/authorize")


# OAuth2 的 token 端点接收的是表单数据（application/x-www-form-urlencoded）
class OAuth2ClientCredentialsRequestForm:
    def __init__(
        self,
        grant_type: str = Form(..., pattern="client_credentials"),
        scope: str = Form(""),
        client_id: str = Form(...),
        client_secret: str = Form(...),
    ):
        self.grant_type = grant_type
        self.scopes = scope.split()
        self.client_id = client_id
        self.client_secret = client_secret


@router.post("/oauth2/authorize", summary="请求授权URL地址")
async def authorize(client_data: OAuth2ClientCredentialsRequestForm = Depends()):
    if client_data.client_id not in fake_client_db:
        raise HTTPException(status_code=400, detail="客户端ID不存在")
    clientinfo = fake_client_db.get(client_data.client_id)
    if client_data.client_secret != clientinfo["client_secret"]:
        raise HTTPException(status_code=400, detail="客户端密钥错误")

    data = {
        "iss": "client_id",
        "sub": "dilibili",
        "client_id": client_data.client_id,
        "exp": datetime.now(timezone.utc) + timedelta(minutes=10),
    }
    token = TokenUtils.token_encode(data)
    return {
        "access_token": token,
        "token_type": "bearer",
        "expires_in": 600,
        "scope": "all",
    }


@router.get("/get/clientinfo", summary="获取客户端信息(受保护资源)")
async def get_clientinfo(token: str = Depends(oauth2_scheme)):
    payload = TokenUtils.token_decode(token)
    client_id = payload.get("client_id")
    if client_id not in fake_client_db:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Could not validate credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
    clientinfo = fake_client_db.get(client_id)
    return {
        "client_id": clientinfo["client_id"],
        "client_secret": clientinfo["client_secret"],
    }

测试：

# 1. 获取 token（注意这里是表单提交）
curl -s -X POST 'http://127.0.0.1:8000/chapter9/oauth2/authorize' \
  -d 'grant_type=client_credentials&client_id=zhangsan&client_secret=123456&scope=all' | python -m json.tool

# 2. 使用 token 访问受保护资源
curl -X GET 'http://127.0.0.1:8000/chapter9/get/clientinfo' \
  -H 'Authorization: Bearer <access_token>'

OAuth 2 密码模式 + JWT（标准登录流程）​

密码模式是最常见的"用户名密码登录"流程：用户提交用户名密码换取 token，之后携带 token 访问受保护接口。FastAPI 内置了 OAuth2PasswordBearer 和 OAuth2PasswordRequestForm 简化实现。

from datetime import datetime, timedelta, timezone
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

fake_users_db = {
    "zhangsan": {
        "username": "zhangsan",
        # 实际项目中密码必须哈希存储，这里用 passlib 生成：
        # pwd_context.hash("123456")
        "hashed_password": "$2b$12$...",
        "disabled": False,
    }
}

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/login")


class Token(BaseModel):
    access_token: str
    token_type: str


class User(BaseModel):
    username: str
    email: str | None = None
    disabled: bool | None = None


class UserInDB(User):
    hashed_password: str


def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)


def get_user(db, username: str) -> UserInDB | None:
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)
    return None


def authenticate_user(db, username: str, password: str) -> UserInDB | None:
    user = get_user(db, username)
    if not user or not verify_password(password, user.hashed_password):
        return None
    return user


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


async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]) -> User:
    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 | None = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    user = get_user(fake_users_db, username)
    if user is None:
        raise credentials_exception
    return user


app = FastAPI()


@app.post("/login")
async def login(form_data: Annotated[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="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token = create_access_token(data={"sub": user.username})
    return Token(access_token=access_token, token_type="bearer")


@app.get("/users/me")
async def read_users_me(current_user: Annotated[User, Depends(get_current_user)]):
    return current_user

测试：

# 1. 登录获取 token
curl -X POST 'http://127.0.0.1:8000/login' \
  -d 'username=zhangsan&password=123456'

# 2. 携带 token 访问受保护接口
curl -X GET 'http://127.0.0.1:8000/users/me' \
  -H 'Authorization: Bearer <access_token>'
注意

依赖安装：pip install python-jose passlib[bcrypt]。实际项目请使用数据库保存用户和哈希密码，并妥善保管 SECRET_KEY（放到环境变量/配置中心，见 读取应用配置）。

参考​