FastAPI实战：构建认证、权限与限流三位一体的安全API服务

做后端这些年，我见过太多业务代码写得飞起，最后却在认证、权限、限流这三件事上翻车的项目。早年我自己做过一个内部管理后台，接口全部裸奔，只要绕过前端按钮就能直接调接口改数据，被同事调侃“这不是后台，是后门”。后来补课才明白，一套服务的健壮程度，往往不取决于业务逻辑多花哨，而取决于这三个功能做得有多扎实。今天就用 Python + FastAPI 的实际代码，把身份认证、接口权限、流量限流从零到一拆开来讲，理清设计思路，也把我踩过的坑一并交代清楚。适合正在写接口、准备上线的Python后端开发，也适合想从零搭一套安全可用的API服务的同学。

1. 认证是第一关：确认“你是谁”

1.1 认证和授权别混为一谈

很多人在设计接口时把认证和授权混在一起说，其实这是两件完全不同的事。认证（Authentication）解决的是“你是谁”的问题，你拿用户名密码过来，我给你验明正身；授权（Authorization）解决的是“你能干什么”的问题，你身份确认后，我决定让不让你执行某个操作。先有认证，后有授权，顺序不能乱。

举个直观的例子：公司门禁刷卡，刷完卡门开了，这是认证；进了门之后，普通员工只能进办公区，财务室、机房需要更高权限才能进，这是授权。在系统里也一样，登录接口负责认证，业务接口前面的权限校验负责授权。

常见的认证方案有 Session-Cookie、JWT、OAuth2 第三方登录等。如果做前后端分离的 API 服务，JWT（JSON Web Token）基本是首选，因为它无状态、跨语言、扩展方便。用 Session 的话，服务端要维护会话状态，分布式部署时还得引入 Redis 存 session，麻烦不少。JWT 把用户信息签进 Token 里，服务端只需要验签，不需要存状态，天然适合多实例部署。

下表对比了几种方案的特点：

1.2 用户密码的保存：第一道安全底线

认证绕不开账号体系，账号体系最脆弱的地方就是密码存储。很多新手教程里直接拿 MD5 存密码，这在大厂安全评审里是会被直接打回的。MD5 属于快速哈希，计算速度太快，黑客拿到数据库后可以用彩虹表和暴力破解在很短时间内还原大量弱口令。正确的做法是使用专门的密码哈希算法，比如 bcrypt、argon2，它们的特点是“慢”，一次哈希计算要几十到几百毫秒，攻击者想暴力破解的成本会大幅上升。

Python 生态里最省事的组合是 passlib + bcrypt。安装依赖：

bash复制pip install fastapi uvicorn pyjwt passlib[bcrypt] python-multipart redis

密码哈希和校验的代码很简单：

python复制from passlib.context import CryptContext

pwd_ctx = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hash_password(password: str) -> str:
    return pwd_ctx.hash(password)

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

这里有个容易踩坑的细节：bcrypt 哈希结果是固定 60 个字符，但数据库中存放密文字段的长度建议直接给到 100 或 255，别刚好吃一个 60，避免将来换算法或加前缀时字段不够用。另外，bcrypt 会自动生成盐并存储在哈希串中，所以你不需要自己维护盐值，这也是它比“加盐MD5”省心得多的地方。

注意：给用户建表时，密码字段我习惯命名为 password_hash，不叫 password，时刻提醒自己和别人：这里存的是哈希值，不是明文。

1.3 签发与校验 Token：完整 JWT 流程

JWT 的结构分为三部分：Header（头部）、Payload（载荷）、Signature（签名），三部分用点号连接。Header 里声明算法，Payload 里放用户标识和过期时间，Signature 用服务端密钥对前两部分签名。客户端拿到 Token 后，每次请求在 Authorization 头里带上，服务端验签通过就认可这个身份。

签发 Token 的代码：

python复制import jwt
from datetime import datetime, timedelta, timezone

SECRET_KEY = "请改成环境变量里的随机密钥"
ALGORITHM = "HS256"

def create_access_token(user_id: str, roles: list, expires_minutes: int = 30) -> str:
    payload = {
        "sub": user_id,
        "roles": roles,
        "iat": datetime.now(timezone.utc),
        "exp": datetime.now(timezone.utc) + timedelta(minutes=expires_minutes),
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

sub 是 JWT 标准字段，代表主题，通常存放用户唯一标识。roles 是我额外放的，用于后续权限判断。iat 表示签发时间，exp 表示过期时间，这里必须用 UTC 时间，否则多台服务器时区不一致会导致 Token 提前过期或延迟过期，排查起来非常头疼。

校验 Token 的依赖函数是认证模块的核心。我使用 FastAPI 的 HTTPBearer 来解析 Authorization 头：

python复制from typing import Optional
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt

security = HTTPBearer(auto_error=False)

async def get_current_user(credentials: Optional[HTTPAuthorizationCredentials] = Depends(security)):
    if credentials is None:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="缺少登录凭证")
    token = credentials.credentials
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="登录已过期")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="无效的登录凭证")
    return {
        "user_id": payload.get("sub"),
        "roles": payload.get("roles", []),
    }

这里有个细节我特别想说：HTTPBearer 默认 auto_error=True，如果请求没带 Authorization 头，FastAPI 会直接返回 403 而非 401。从语义上说，缺少凭证应该返回 401 Unauthorized，403 表示你身份已确认但没权限，两者不能混。所以我把 auto_error 设为 False，自己判空并手动抛 401，这样对外表现才规范。

重要：JWT 的 Payload 只是 Base64 编码，不是加密。任何拿到 Token 的人都能解码看到里面的内容。所以不要在 Payload 里放手机号、身份证号这类敏感信息，只放用户ID和角色这类非敏感标识。

登录接口就是把身份校验和 Token 签发串起来：

python复制from fastapi import HTTPException, status

@app.post("/auth/login")
def login(username: str, password: str):
    user = get_user_by_username(username)  # 伪代码，实际查库
    if not user or not verify_password(password, user["password_hash"]):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="用户名或密码错误")
    token = create_access_token(user_id=user["id"], roles=user["roles"])
    return {"access_token": token, "token_type": "bearer"}

我习惯在密码不对时统一返回“用户名或密码错误”，不告诉调用者到底是用户名不存在还是密码错了，避免被用来探测已注册账号。

2. 权限控制：知道你是谁之后，决定你能做什么

2.1 从操作系统权限到 RBAC 模型

权限控制这个概念不只 Web 开发里有，操作系统里更能见其原理。你在 Windows 里删除一个文件，提示“你需要来自 Administrators 的权限才能对此文件夹进行更改”，本质就是当前用户这个主体，对文件这个客体执行删除操作时，系统检查了访问控制列表，发现你不在允许列表里，于是拒绝执行。Web 系统的权限设计，底层逻辑完全一样。

最经典的权限模型是 RBAC（Role-Based Access Control，基于角色的访问控制）。它引入“角色”这个中间层，把用户和权限解耦。用户关联角色，角色关联权限，而不是直接让用户绑定权限。为什么要多这一层？因为实际业务中权限点数量多且会变动，如果用户直接绑权限，每新增一个权限就要给几百个用户改数据，管理成本极高。有了角色之后，比如“运营”角色统一配置权限，批量调整即可，非常灵活。

RBAC 的核心三要素是：用户、角色、权限。一次权限校验的问题可以拆成三步：这个用户有哪些角色？这个角色有哪些权限？当前操作需要哪个权限？全部对上了才放行。

更细一点的还有 ABAC（基于属性的访问控制），根据用户属性、资源属性、环境条件动态决策，适合复杂场景，但实现成本高。大多数业务系统先把 RBAC 做扎实就够了，不要一上来就堆复杂度。

2.2 用依赖注入实现接口级权限校验

FastAPI 的依赖注入系统非常适合做权限校验。上一节写的 get_current_user 是一个依赖，权限校验可以再包一层依赖，传入要求的角色，然后复用 get_current_user 的结果。

python复制from typing import Callable

def require_roles(*required_roles: str) -> Callable:
    async def dependency(user: dict = Depends(get_current_user)):
        user_roles = set(user.get("roles", []))
        if not user_roles.intersection(required_roles):
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="当前账号没有执行该操作的权限"
            )
        return user
    return dependency

用起来有两种姿势。第一种是把依赖挂在路由参数上：

python复制@app.get("/admin/users")
async def admin_list_users(user: dict = Depends(require_roles("admin"))):
    return {"message": "只有管理员能看到"}

第二种是放在路由装饰器的 dependencies 参数里：

python复制@app.delete("/admin/users/{user_id}", dependencies=[Depends(require_roles("admin"))])
async def delete_user(user_id: str):
    return {"message": "已删除"}

两者区别在于：第一种会在路径函数里拿到 user，适合权限校验后还要用登录信息的场景；第二种拿不到 user，适合只做校验、不关心是谁在操作的路由。我常用第一种，因为写日志或审计时要记录操作人。

我遇到过一个真实翻车案例：有个同事用装饰器实现权限校验，把函数包了一层，结果 FastAPI 拿到的是被装饰后的函数，丢失了原函数的参数签名，接口文档乱了，部分参数校验也失效。后来改成依赖注入，问题全解决。这不是说装饰器不能用，而是依赖注入是 FastAPI 原生支持的机制，类型提示、文档生成、依赖复用都比装饰器干净得多。

提示：权限不足时一定要抛 403，而不是 401。401 表示没认证或凭证无效，客户端会重新登录；403 表示已认证但没权限，客户端应该提示“无权操作”，而不是重新登录。

2.3 行级权限与数据范围控制

接口级权限只能控制“这个接口谁能调”，但很多场景下还要控制“这行数据谁能看”。经典例子：用户只能查看自己的订单，管理员可以查看全部订单。如果只做接口级校验，用户传一个别人的订单号就能看到别人数据，这叫水平越权，也称 IDOR（Insecure Direct Object Reference），安全测试里最常抓的就是这种漏洞。

行级权限的实现思路很简单：查询后比对资源属主，不匹配就拒绝。

python复制@app.get("/orders/{order_id}")
def get_order(order_id: str, user: dict = Depends(require_roles("user", "admin"))):
    order = get_order_from_db(order_id)  # 伪代码，实际查库
    if order is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="订单不存在")
    if order.owner_id != user["user_id"] and "admin" not in user["roles"]:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="无权查看该订单")
    return order

这里有个细节：先查订单再校验属主，如果订单不存在，返回 404。如果校验属主不通过，返回 403。不能把“订单不存在”和“无权访问”统一返回 404，否则会暴露接口探测逻辑；但反过来，如果业务上不想让用户确认某个订单是否存在，可以统一返回 404。具体业务具体定，但要注意一致性。

行级权限还有一种批量场景，比如列表接口，普通用户只能查到自己名下的数据。这时候不要在代码里逐条过滤，最好在 SQL 层直接加条件：

python复制orders = db.query(Order)
if "admin" not in user["roles"]:
    orders = orders.filter(Order.owner_id == user["user_id"])

如果在查询后再逐条过滤，数据量大时会有严重的性能问题，而且容易漏过滤。条件下沉到 SQL 层，既高效又不容易犯错。列级权限也是类似思路，对敏感字段（手机号、身份证）进行脱敏或限制返回，实现时可以通过序列化器动态选择返回字段，或者用两个查询模型做区分。

3. 限流：从源头保护后端资源

3.1 为什么要限流，以及常见算法区别

限流这件事，做的好不好，直接决定服务能不能扛过突发的流量高峰。攻击者可以用脚本以每秒几百次的频率调用你的登录接口，暴力尝试密码；某个活动上线后，用户集中点击，瞬间涌进大量请求；内部定时任务凌晨并发跑数，把数据库打满。这些场景下，业务代码再优化也顶不住无限流量，必须在入口处控制速率。

常见限流算法有四种：固定窗口、滑动窗口、令牌桶、漏桶。

固定窗口的问题我举个具体例子：限制每分钟 100 次。某个请求在 10:00:59 来临，这是第一分钟的最后一秒；紧接着 10:01:00 又来了一个请求，这已经是第二分钟了，计数器归零重新计数。结果就是在 10:00:59 到 10:01:00 这一秒内，可能放行了 200 个请求，这就是“临界突刺”，会绕过限流。

滑动窗口通过更细粒度的时间片来避免临界问题，但存储成本高了一些。令牌桶则允许一定的突发流量，同时保证长时间平均速率可控，所以实际项目里应用最普遍。我的经验是：对外接口用令牌桶，内部调用用固定窗口也够了，看对平滑度的要求。

3.2 单机限流：内存令牌桶的实现

先写一个单机版令牌桶。核心逻辑：桶里最多放 capacity 个令牌，每秒补充 refill_per_second 个令牌，请求来了消费一个令牌，桶空就拒绝。

python复制import time
import threading
from collections import defaultdict

class MemoryTokenBucket:
    def __init__(self, capacity: int, refill_per_second: float):
        self.capacity = capacity
        self.refill_per_second = refill_per_second
        self.buckets = defaultdict(self._create_bucket)
        self.lock = threading.Lock()

    def _create_bucket(self):
        return {"tokens": self.capacity, "last_refill": time.monotonic()}

    def allow(self, key: str) -> bool:
        with self.lock:
            bucket = self.buckets[key]
            now = time.monotonic()
            elapsed = now - bucket["last_refill"]
            bucket["tokens"] = min(self.capacity, bucket["tokens"] + elapsed * self.refill_per_second)
            bucket["last_refill"] = now
            if bucket["tokens"] >= 1:
                bucket["tokens"] -= 1
                return True
            return False

time.monotonic() 是我特别强调的，它不受系统时间调整影响，专门用于计算时间间隔，比 time.time() 更可靠。线程锁保证并发环境下计数器安全。

在 FastAPI 里接入很简单，定义一个全局令牌桶对象，然后在接口里检查：

python复制bucket = MemoryTokenBucket(capacity=20, refill_per_second=5)

@app.get("/api/important")
def important_api(user: dict = Depends(get_current_user)):
    if not bucket.allow(user["user_id"]):
        raise HTTPException(status_code=status.HTTP_429_TOO_MANY_REQUESTS, detail="请求过于频繁")
    return {"message": "ok"}

单机版最大的坑在于内存无限增长。每个新用户都会在 self.buckets 里留下一个条目，如果不做清理，长时间运行后内存会被占满。简单方案是给每个 key 加最后访问时间，定期把长期不活跃的 key 删除，或者用 LRU 缓存结构来替代默认字典。我踩过一次，一个服务跑了两个月，内存从 200MB 涨到 2GB，清理逻辑加上后才恢复正常。

注意：单机限流只对单实例有效。如果服务水平扩展成多个实例，每个实例维护各自的计数器，总限流阈值会被放大 N 倍，这时候必须用分布式限流。

3.3 分布式限流：Redis 固定窗口和 Lua 脚本

多实例部署时，限流计数需要统一存储，Redis 是最常用的选择。固定窗口 + Redis 实现非常简洁：以“用户ID + 当前时间窗口”作为 Key，用 INCR 自增计数，第一次自增时设置过期时间。

直接用 Python 实现会有并发问题：INCR 之后判断是否第一次，然后 EXPIRE，两步之间如果实例挂掉，Key 可能变成永不过期。更稳妥的做法是把判断和过期设置写进 Lua 脚本，利用 Redis 单线程执行脚本的原子性保证不出岔子。

python复制import redis
import time

r = redis.Redis(host="localhost", port=6379, decode_responses=True)

RATE_LIMIT_SCRIPT = """
local current = redis.call('incr', KEYS[1])
if current == 1 then
    redis.call('expire', KEYS[1], ARGV[1])
end
return current
"""

rate_limit_sha = r.script_load(RATE_LIMIT_SCRIPT)

def redis_fixed_window(key: str, limit: int, window_seconds: int) -> bool:
    window = int(time.time()) // window_seconds
    redis_key = f"rate_limit:{key}:{window}"
    current = r.evalsha(rate_limit_sha, 1, redis_key, window_seconds)
    return int(current) <= limit

窗口的计算通过 int(time.time()) // window_seconds 实现，每过 window_seconds 秒，窗口编号加一，自然切换。这个方案的好处是代码极短、性能好、易于理解。

但我还是要提醒：固定窗口有临界突刺问题，Redis 版也一样。如果业务对平滑度要求高，可以改成 Redis 的 Sorted Set（ZSET）实现滑动窗口：Score 为请求时间戳，每次请求前删除窗口外的旧记录，再统计窗口内数量。代价是每个 Key 会存大量时间戳，内存开销更大。大部分场景固定窗口够用，滑动窗口是升级选项，不必一上来就上。

限流的 Key 设计也要花心思。用用户 ID，可以区分不同用户；用 IP，可以防单个 IP 狂刷，但 NAT 环境下很多用户共享一个出口 IP，容易误伤。我通常的组合策略是：登录用户按用户 ID 限流，匿名接口按 IP 限流，两套规则独立配置。

4. 实战中容易踩的坑与排查实录

4.1 认证模块的坑

令牌过期时间设置。JWT 一旦签发，在过期前无法主动让它失效。如果用户改了密码，老 Token 依然有效，这是很危险的事。我常用的缓解方案是在用户表加一个 password_changed_at 字段，Token 里带上该字段的值，校验时对比一下，密码修改后这个时间戳变化，旧 Token 自然失效。另一种方案是维护 Token 黑名单，但这就引入了状态存储，等于部分放弃了 JWT 的优势，权衡之后再定。

decode 异常只捕获了一部分。写校验代码时，我只写了 ExpiredSignatureError 和 InvalidTokenError，但实际 pyjwt 抛的异常类型很多，比如 DecodeError、MissingRequiredClaimError。稳妥做法是捕获 jose 的基础异常类，比如 jwt.PyJWTError，统一返回 401。我之前捕获漏了一种异常类型，线上日志里经常出现 500，排查半天才发现是某个字段缺失导致的。

时区不一致导致 Token 签发出错。有一次我用了本地时间而不是 UTC，测试环境一切正常，上线后和部署在其他时区的服务联调，发现 Token 时而有效时而无效，最后发现 iat 和 exp 都用的本地时间，导致签名校验时对不上。统一用 datetime.now(timezone.utc) 后问题消失。

4.2 权限模块的坑

角色变了 Token 不更新。JWT 里存了 roles，角色信息存的是签发时的快照。如果用户在 Token 有效期内被降级或删号，他手里的旧 Token 照样能访问高权限接口，直到过期。解决思路：敏感接口从数据库实时查一次角色，而不是完全信任 Token 里的快照。比如新增一个依赖函数，根据 user_id 重新查库，再检查权限。

权限点命名混乱。项目初期权限字符串随手写，前端传 updateUser，后端校验 user_update，两边对不上，调试起来非常痛苦。权限点建议统一格式：模块:操作，比如 user:create、order:delete，用冒号分隔，集中维护在常量类里，前后端共享这一套命名。

依赖注入和装饰器的混用问题。我在 2.2 里提过，权限校验用依赖注入比装饰器稳。快速排查方法：接口文档里如果某个接口的参数列表异常或者 401/403 行为不对，优先检查是不是被装饰器包裹后破坏了函数签名。

4.3 限流模块的坑

阈值设置不合理误伤用户。限流阈值不是拍脑袋定的，要根据业务压测数据来。比如登录接口正常用户一分钟最多点三次登录按钮，阈值可以定 10 次/分钟；上传接口高峰期可能很频繁，要预留更多空间。阈值设太低，用户正常操作都被 429，体验极差；设太高，限流形同虚设。我用一个简单的办法：上线前先压测，记录 P99 的请求速率，阈值为这个值的 3 到 5 倍。

限流 Key 粒度太粗。用 IP 限流时，如果某个公司出口 IP 下有三五百人，一人请求一次就把 IP 的限额用光了，整个公司都进不来。排查时先看日志里被拒请求的分布，如果集中在同一个 IP 段，说明 Key 粒度有问题。策略是拆成“IP + 用户ID”组合 Key，或者对已登录用户优先按用户 ID 限流。

忘掉清理导致的内存泄漏。单机内存限流必须有过期清理逻辑，哪怕只是最简单的定时把超过 30 分钟没活动的 key 清掉。数据库里如果开了慢日志，经常能看到限流相关查询，也可能是 Key 设计不合理，比如把时间戳精确到纳秒导致每个请求都创建新 Key。

5. 上线前我每次都会检查的几件事

做过的项目多了，慢慢养成了一个习惯：每次发布前按清单自查一遍，能省去很多线上事故。这几项是我一定会过一遍的。

第一，数据库用户表密码字段长度是否足够。现在多项目从 SQLite 迁到 MySQL，就出现过字段长度太短存不下 bcrypt 哈希的情况。建表时直接给 255，一劳永逸。

第二，SECRET_KEY 是不是从环境变量导入。写死在代码里的密钥，一旦代码库泄露，所有 Token 都可以被伪造。生产环境务必从配置中心或环境变量读取，并定期轮换。

第三，所有需要登录的接口是否都加上了认证依赖。我常用一个笨办法：启动服务后，不带 Token 把所有接口请求一遍，返回 401 就通过，返回 200 或 500 就有遗漏。写个脚本自动跑，每次上线前执行一次。

第四，限流的阈值和 Key 设计是否合理。确认登录接口、下单接口、短信发送接口这些高风险接口都接入了限流，并且 Key 能区分用户维度，不会误伤正常用户。

第五，用 curl 实测三个状态码：未登录返回 401，登录但无权限返回 403，超过限流阈值返回 429。这三个状态码是安全体系对外暴露的语言，必须准确无误。

检查完这些，我才敢点下发布按钮。这套体系从简单到复杂都可以用，小项目用内存限流和 JWT，大项目升级 Redis 和 RBAC，技术选型有梯度，落地就有底气。