Python + FastAPI 实现 API Key 鉴权：生成、哈希存储、撤销与限流

Kevin_need_java

277人浏览 · 2026-08-17 11:34:23

Kevin_need_java · 2026-08-17 11:34:23 发布

开发一个对外 API 时，我们通常需要回答几个问题：

谁可以调用接口？
如何识别调用方？
某个密钥泄露后如何单独撤销？
如何限制请求频率？
数据库泄露后，如何避免密钥被直接使用？

API Key 是一种常见的调用凭证，但仅仅判断请求中“有没有 Key”并不够。一套相对完整的实现还需要考虑生成、存储、权限、撤销、限流和日志脱敏。

OWASP 建议受保护接口要求客户端携带 API Key，请求过快时返回 429 Too Many Requests，并支持撤销违规或泄露的密钥。API Key 也不应该出现在 URL 中，因为 URL 可能进入服务器日志、浏览器历史和监控系统。OWASP REST Security Cheat Sheet

一、安装依赖

本文使用 Python 3.10 及以上版本。

pip install fastapi uvicorn

创建文件：

api-key-demo/
└── main.py

二、API Key 应该如何生成？

不要使用用户名、时间戳或自增 ID 充当密钥：

# 不安全示例
api_key = f"key_{user_id}_{int(time.time())}"

这类 Key 容易预测。

Python 标准库提供了适合生成安全随机令牌的 secrets 模块：

import secrets


def generate_api_key() -> str:
    random_part = secrets.token_urlsafe(32)
    return f"sk_demo_{random_part}"

前缀 sk_demo_ 不负责提供安全性，它的作用是：

帮助用户识别凭证类型；
方便密钥扫描工具发现误提交；
减少把其他 Token 当作 API Key 的情况。

真正的随机性来自 secrets.token_urlsafe(32)。

三、为什么不应该保存明文密钥？

如果数据库直接保存：

sk_demo_xxxxxxxxx

一旦数据库被读取，攻击者可以立即调用接口。

更合理的方案是：

创建密钥时只展示一次完整值；
数据库只保存不可逆摘要；
用户再次提交密钥时计算摘要；
使用摘要查找对应记录。

本文使用 HMAC-SHA256，并加入只保存在服务端环境变量中的 Pepper：

import hashlib
import hmac
import os


KEY_PEPPER = os.getenv(
    "API_KEY_PEPPER",
    "development-only-change-me"
)


def digest_api_key(api_key: str) -> str:
    return hmac.new(
        KEY_PEPPER.encode("utf-8"),
        api_key.encode("utf-8"),
        hashlib.sha256
    ).hexdigest()

生产环境必须设置一个足够随机的 API_KEY_PEPPER：

export API_KEY_PEPPER="替换为高强度随机值"

不要将真实 Pepper 提交到代码仓库。

四、定义密钥数据结构

为了方便演示，本文使用内存字典代替数据库：

from dataclasses import dataclass, field
from datetime import datetime, timezone


@dataclass
class APIKeyRecord:
    key_id: str
    key_digest: str
    key_prefix: str
    owner_id: str
    scopes: set[str] = field(default_factory=set)
    enabled: bool = True
    created_at: datetime = field(
        default_factory=lambda: datetime.now(timezone.utc)
    )
    last_used_at: datetime | None = None

创建存储容器：

api_keys: dict[str, APIKeyRecord] = {}

生产环境应换成 PostgreSQL、MySQL 等持久化数据库。

五、创建密钥

完整密钥只在创建时返回：

import secrets


def create_api_key(
    owner_id: str,
    scopes: set[str]
) -> tuple[str, APIKeyRecord]:
    plain_key = generate_api_key()
    key_digest = digest_api_key(plain_key)

    record = APIKeyRecord(
        key_id=f"key_{secrets.token_hex(8)}",
        key_digest=key_digest,
        key_prefix=plain_key[:16],
        owner_id=owner_id,
        scopes=scopes
    )

    api_keys[key_digest] = record

    return plain_key, record

key_prefix 只保存前面一小段，用于后台展示：

sk_demo_abcd1234...

它不能用于鉴权。

六、从 Authorization Header 读取密钥

推荐的请求形式：

Authorization: Bearer sk_demo_xxxxxxxxx

不要使用 URL 参数：

https://example.com/api?api_key=sk_demo_xxx

因为 URL 更容易进入：

访问日志；
浏览器历史；
代理服务器日志；
分析与监控平台；
Referer 信息。

使用 FastAPI 的 HTTPBearer：

from fastapi import Depends, HTTPException, status
from fastapi.security import (
    HTTPAuthorizationCredentials,
    HTTPBearer
)


bearer_scheme = HTTPBearer(auto_error=False)

实现鉴权依赖：

def authenticate_api_key(
    credentials: HTTPAuthorizationCredentials | None = Depends(
        bearer_scheme
    )
) -> APIKeyRecord:
    if credentials is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Missing API key"
        )

    plain_key = credentials.credentials
    key_digest = digest_api_key(plain_key)
    record = api_keys.get(key_digest)

    if record is None or not record.enabled:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid API key"
        )

    record.last_used_at = datetime.now(timezone.utc)

    return record

无论密钥不存在还是已停用，都返回相同错误，避免向调用者暴露过多内部状态。

七、添加权限范围

只有通过鉴权，并不意味着密钥可以调用所有接口。

例如可以定义：

models:read
chat:write
images:write
admin:write

创建权限检查函数：

def require_scope(required_scope: str):
    def checker(
        record: APIKeyRecord = Depends(authenticate_api_key)
    ) -> APIKeyRecord:
        if required_scope not in record.scopes:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Insufficient scope"
            )

        return record

    return checker

图片接口要求 images:write：

@app.post("/v1/images/generations")
def generate_image(
    record: APIKeyRecord = Depends(
        require_scope("images:write")
    )
):
    return {
        "message": "Task accepted",
        "owner_id": record.owner_id
    }

认证失败应该返回 401，已经认证但权限不足则返回 403。

八、实现简单限流

API Key 泄露后，攻击者可能高频调用接口。只有鉴权，没有限流，仍然可能产生大量资源消耗。

OWASP API Security Top 10 将不受限制的资源消耗列为重要风险，并建议限制每个客户端的调用频率、请求大小和执行时间。OWASP API Security Top 10

下面使用滑动窗口实现演示版限流：

import time
from collections import defaultdict, deque


RATE_LIMIT = 10
RATE_WINDOW_SECONDS = 60

request_times: dict[str, deque[float]] = defaultdict(deque)


def check_rate_limit(record: APIKeyRecord) -> None:
    now = time.monotonic()
    cutoff = now - RATE_WINDOW_SECONDS
    timestamps = request_times[record.key_id]

    while timestamps and timestamps[0] <= cutoff:
        timestamps.popleft()

    if len(timestamps) >= RATE_LIMIT:
        retry_after = max(
            1,
            int(RATE_WINDOW_SECONDS - (now - timestamps[0]))
        )

        raise HTTPException(
            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
            detail="Rate limit exceeded",
            headers={
                "Retry-After": str(retry_after)
            }
        )

    timestamps.append(now)

将鉴权和限流组合起来：

def authenticated_with_limit(
    record: APIKeyRecord = Depends(authenticate_api_key)
) -> APIKeyRecord:
    check_rate_limit(record)
    return record

需要注意：内存限流只适用于单进程演示。

如果部署多个实例，每个实例都有独立计数，应该使用 Redis 或 API Gateway 进行集中限流。

九、撤销密钥

密钥泄露时不能要求停掉整个账号，应该允许只撤销单个 Key：

def revoke_api_key(key_id: str) -> bool:
    for record in api_keys.values():
        if record.key_id == key_id:
            record.enabled = False
            return True

    return False

生产环境还可以记录：

revoked_at
revoked_by
revoke_reason

撤销后，下一次请求立即返回 401。

十、完整可运行代码

import hashlib
import hmac
import os
import secrets
import time
from collections import defaultdict, deque
from dataclasses import dataclass, field
from datetime import datetime, timezone

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import (
    HTTPAuthorizationCredentials,
    HTTPBearer
)


app = FastAPI(title="API Key Demo")

KEY_PEPPER = os.getenv(
    "API_KEY_PEPPER",
    "development-only-change-me"
)

RATE_LIMIT = 10
RATE_WINDOW_SECONDS = 60


@dataclass
class APIKeyRecord:
    key_id: str
    key_digest: str
    key_prefix: str
    owner_id: str
    scopes: set[str] = field(default_factory=set)
    enabled: bool = True
    created_at: datetime = field(
        default_factory=lambda: datetime.now(timezone.utc)
    )
    last_used_at: datetime | None = None


api_keys: dict[str, APIKeyRecord] = {}
request_times: dict[str, deque[float]] = defaultdict(deque)

bearer_scheme = HTTPBearer(auto_error=False)


def generate_api_key() -> str:
    return f"sk_demo_{secrets.token_urlsafe(32)}"


def digest_api_key(api_key: str) -> str:
    return hmac.new(
        KEY_PEPPER.encode("utf-8"),
        api_key.encode("utf-8"),
        hashlib.sha256
    ).hexdigest()


def create_api_key(
    owner_id: str,
    scopes: set[str]
) -> tuple[str, APIKeyRecord]:
    plain_key = generate_api_key()
    key_digest = digest_api_key(plain_key)

    record = APIKeyRecord(
        key_id=f"key_{secrets.token_hex(8)}",
        key_digest=key_digest,
        key_prefix=plain_key[:16],
        owner_id=owner_id,
        scopes=scopes
    )

    api_keys[key_digest] = record

    return plain_key, record


def authenticate_api_key(
    credentials: HTTPAuthorizationCredentials | None = Depends(
        bearer_scheme
    )
) -> APIKeyRecord:
    if credentials is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Missing API key"
        )

    key_digest = digest_api_key(credentials.credentials)
    record = api_keys.get(key_digest)

    if record is None or not record.enabled:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid API key"
        )

    record.last_used_at = datetime.now(timezone.utc)

    return record


def check_rate_limit(record: APIKeyRecord) -> None:
    now = time.monotonic()
    cutoff = now - RATE_WINDOW_SECONDS
    timestamps = request_times[record.key_id]

    while timestamps and timestamps[0] <= cutoff:
        timestamps.popleft()

    if len(timestamps) >= RATE_LIMIT:
        retry_after = max(
            1,
            int(RATE_WINDOW_SECONDS - (now - timestamps[0]))
        )

        raise HTTPException(
            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
            detail="Rate limit exceeded",
            headers={
                "Retry-After": str(retry_after)
            }
        )

    timestamps.append(now)


def require_scope(required_scope: str):
    def checker(
        record: APIKeyRecord = Depends(authenticate_api_key)
    ) -> APIKeyRecord:
        check_rate_limit(record)

        if required_scope not in record.scopes:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Insufficient scope"
            )

        return record

    return checker


@app.post("/demo/keys")
def issue_demo_key():
    plain_key, record = create_api_key(
        owner_id="user_123",
        scopes={
            "models:read",
            "images:write"
        }
    )

    return {
        "api_key": plain_key,
        "key_id": record.key_id,
        "warning": "完整密钥只显示一次，请安全保存"
    }


@app.get("/v1/models")
def list_models(
    record: APIKeyRecord = Depends(
        require_scope("models:read")
    )
):
    return {
        "data": [
            {"id": "example-model"}
        ],
        "key_id": record.key_id
    }


@app.post("/v1/images/generations")
def generate_image(
    record: APIKeyRecord = Depends(
        require_scope("images:write")
    )
):
    return {
        "message": "Task accepted",
        "owner_id": record.owner_id
    }


@app.delete("/demo/keys/{key_id}")
def revoke_key(key_id: str):
    for record in api_keys.values():
        if record.key_id == key_id:
            record.enabled = False
            return {"revoked": True}

    raise HTTPException(
        status_code=404,
        detail="Key not found"
    )

启动：

export API_KEY_PEPPER="请替换为随机值"
uvicorn main:app --reload

打开接口文档：

http://127.0.0.1:8000/docs

先调用 /demo/keys 创建密钥，再点击 Swagger 页面右上角的 Authorize，填入密钥进行测试。

/demo/keys 在本文中只是为了方便演示。生产环境的密钥创建接口必须由登录用户或管理员权限保护。

十一、生产环境注意事项

不要在日志中记录完整 Authorization Header。
不要把 Key 放在前端网页、移动应用安装包或公开仓库中。
为每个使用场景创建独立 Key，不要多人共享一个 Key。
支持撤销、轮换、到期时间和权限范围。
对请求频率、并发量和消费金额分别设置限制。
配置异常消费告警。
数据库只保存摘要和可展示前缀。
限流应使用 Redis 或网关等共享存储。
高价值操作不能只依赖 API Key，还需要更严格的授权校验。
第三方接口调用应设置连接超时和总超时。

亚马逊云科技技术品牌专区

更多推荐

第5篇_Python文件操作与异常处理：程序与外界交互的桥梁

读取一个不存在的文件 → Python 抛出把字符串当数字做除法 →TypeError整数除以零 →列表只有 3 个元素却要访问第 10 个 →IndexError用户输入了 “abc” 而程序期待一个数字 →ValueError这些错误，在 Python 中被称为异常（Exception）。异常不是 “程序坏了”，而是 “程序遇到了预期之外的情况”。默认行为：如果不对异常做任何处理，程序会打印错

亚马逊云科技技术品牌专区

第6篇_Python高级语法与标准库精髓：从会用到达精通

本篇是，我们将深入 Python 真正精华的部分 —— 那些让代码变得更简洁、更高效、更强大的语法和工具。学完本篇，你将从 “会写 Python” 迈向 “精通 Python”。准备好了吗？让我们开始！🚀。

亚马逊云科技技术品牌专区

一次请求获取 10 只跨市场股票 K 线：QuantDash 批量接口完全指南

摘要 QuantDash Python SDK 的 klines.batch() 方法支持单次 API 调用获取多只跨市场股票（A股/美股/港股）的 K 线数据，简化批量数据获取流程。传统方案存在爬虫限流、多市场代码不统一、复权计算复杂等问题，而 QuantDash 提供稳定数据源、统一代码格式、服务端自动复权（支持前复权等四种方式），并原生返回 Pandas DataFrame。5 行代码即可实