☰
FastAPI实战:认证、权限与限流的三层防线
2026/10/16 12:36:15 网站建设 项目流程

搞Web后端的人,迟早都会撞上这三件事:接口要不要登录才能调?登录之后谁能调?同一个IP一分钟能调多少次?这三件事就是标题里说的认证、权限和限流。我最近在一个内部管理系统上把这套东西完整落地了一遍,用的是FastAPI,但里面的思路放到Flask、Django上一样能平移。

先说结论:认证解决“你是谁”,权限解决“你能干什么”,限流解决“你一次能干多少”。很多新手容易把认证和权限混着写,结果就是接口里到处塞if判断,改一个角色权限要翻半天代码。这篇文章我把当时的完整方案拆开讲,从表结构设计到依赖注入,再到Redis限流,每一步都给出可复现的代码和踩坑记录,给正在写Python接口的开发者一份可以直接抄的作业。

1. 先把三件事拆开想:认证、权限、限流到底各管什么

很多项目之所以在后期变得难维护,根源在于一开始就没把这三件事的边界划清楚。我自己早期也犯过这个错:在业务函数里写死用户角色判断,API路由里塞IP频率统计,结果每个接口都像一个大杂烩。所以先花点时间把概念理清楚,后面写代码会顺很多。

1.1 三者的边界在哪里

认证(Authentication)的核心是“验证身份”。用户拿用户名密码来换一个凭证,比如Session ID或者JWT,之后每次请求带上这个凭证,服务端能确认“哦,这是张三”。认证只做身份确认,不回答任何“能不能”的问题。权限(Authorization)的核心是“资源授权”。它发生在认证之后,确认了“你是张三”,然后查“张三能不能删除这条订单”。权限判断的输入是用户身份和目标操作,输出是允许或拒绝。限流(Rate Limiting)跟前两者不一样,它跟“身份”没有必然关系,它管的是“请求频率”。哪怕你是一个合法的管理员,短时间打几百次请求也一样会被限制,目的就是保护服务端资源不被耗光。

一个很常见的误区是:限流属于安全模块,应该跟着认证走。实际上限流往往在认证之前就要生效,因为很多恶意请求根本不会走到登录那一步,直接就在网关层被拦掉了。我建议把限流看成是独立的横切面,跟认证权限解耦。

1.2 为什么项目里总把它们放一起实现

因为它们的另一个共同点是:都需要获取到请求的上下文信息。认证需要从Header里拿Token,权限需要知道当前用户和操作对象,限流需要知道IP或者用户ID。这些信息在每个请求进来之后都要被解析一遍,放一起做能用最小成本完成三次“请求前置处理”。

我用FastAPI做实践,是因为它对依赖注入的支持很舒服。依赖注入这个词听起来玄乎,其实你可以理解成“每个接口在正式执行之前可以声明好一些前置条件,框架会自动帮你执行”。比如一个删除接口声明了“当前用户必须是管理员”,FastAPI就会先进依赖,解析Token、查角色、比对权限,全部通过才进业务函数。这套机制天然适合把认证、权限、限流做成可复用的函数,而不是在每个路由里重复贴代码。

技术选型上,认证方案我最终选了JWT而不是Session。原因是项目后面要拆小程序端和Web端,JWT无状态、跨端方便,服务端不用维护Session存储。权限模型选了RBAC,角色-权限表结构清晰,新员工入职只要挂角色就行,不用一个一个接口去配。限流方案的选型纠结了一下,后面单独展开。

2. 认证:把“你是谁”这个事做扎实

认证部分我拆成三步走:密码校验、Token签发、Token解析。密码校验负责登录时确认身份,Token签发负责给认证通过的用户发一个“通行证”,Token解析负责在后面所有接口里读取这个通行证并确定用户。

用的核心库是python-jose来做JWT,passlib来做密码哈希。如果你用的是Flask,可以直接换Flask-JWT-Extended;用Django的话djangorestframework-simplejwt是同类方案,思路大同小异。

2.1 登录接口:密码校验 + 接头发放

先定义用户模型。我这里的项目用的SQLAlchemy,用户表字段简化成下面这样:

class User(Base): __tablename__ = "users" id = Column(Integer, primary_key=True, index=True) username = Column(String(50), unique=True, index=True) password_hash = Column(String(128)) role = Column(String(20), default="user") # admin / user / auditor is_active = Column(Boolean, default=True)

密码哈希用passlib的bcrypt方案,注册的时候这样存:

from passlib.context import CryptContext pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") def hash_password(plain_password: str) -> str: return pwd_context.hash(plain_password)

注意这个pwd_context建议单独放在一个模块里,因为登录校验、注册、改密码都要用到它,别在每个文件里重复创建,不然bcrypt的加载开销会浪费在每次请求上。

登录接口用OAuth2的标准表单方式接收用户名密码。FastAPI里可以直接用OAuth2PasswordRequestForm,好处是Swagger文档会自动渲染出一个“Authorize”按钮,调试接口很舒服。

@router.post("/login") def login(form_data: OAuth2PasswordRequestForm = Depends()): user = db.query(User).filter(User.username == form_data.username).first() if not user or not pwd_context.verify(form_data.password, user.password_hash): raise HTTPException(status_code=400, detail="用户名或密码错误") if not user.is_active: raise HTTPException(status_code=403, detail="账号已被禁用") access_token = create_access_token(subject=str(user.id), role=user.role) return {"access_token": access_token, "token_type": "bearer"}

这里有个细节:登录失败的提示信息我故意没有区分“用户不存在”和“密码错误”,统一返回“用户名或密码错误”。这是防枚举的标准做法,否则攻击者可以通过提示差异来判断一个用户名是否存在。

2.2 获取当前用户:Token解析依赖

Token签发之后,每个受保护接口都要能从Token里找回当前用户。这个动作我封装成一个FastAPI依赖:

from fastapi.security import OAuth2PasswordBearer from jose import JWTError, jwt oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login") def create_access_token(subject: str, role: str) -> str: expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) to_encode = {"sub": subject, "role": role, "exp": expire} return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) def get_current_user(token: str = Depends(oauth2_scheme)) -> User: credentials_exception = HTTPException( status_code=401, detail="凭证无效或已过期", headers={"WWW-Authenticate": "Bearer"}, ) try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) user_id = payload.get("sub") if user_id is None: raise credentials_exception user = db.query(User).filter(User.id == int(user_id)).first() if user is None: raise credentials_exception return user except JWTError: raise credentials_exception

注意exp过期时间字段我用的是datetime.utcnow()。JWT规范里exp要求是UTC时间,如果混用了本地时间,跨时区部署会出现令牌凭空多活或少活几小时的问题。这个细节坑过我好几次,建议直接写成常量配置。

2.3 密码哈希与密钥管理的几个坑

passlib的bcrypt是这个方案里最容易出问题的地方。我用的是passlib 1.7.4配合bcrypt 4.x,经常在创建CryptContext时报AttributeError: module 'bcrypt' has no attribute '__about__'。这是版本兼容性的坑,不是代码写错。解决办法有两个:要么把bcrypt锁在3.2.2版本,要么干脆换argon2。我在新项目里已经全面切到argon2了,纯Python环境兼容性更好,passlib的argon2方案没见过这种报错。

SECRET_KEY绝不能写死在代码里,更不能提交到Git仓库。我见过有人把JWT密钥直接放在config.py里然后推到公开仓库,之后所有token都能被解密伪造,等于认证系统整个裸奔。正确做法是用环境变量注入或者密钥管理服务,本地开发可以用secrets.token_urlsafe(32)生成一个临时密钥,线上环境务必用独立的密钥管理。

认证做好之后再去做别的接口就舒坦了,每个接口只要声明user: User = Depends(get_current_user),框架就会自动完成身份解析。

3. 权限:从“能登录”到“能干什么”

认证只告诉你“用户是谁”,权限管理要回答“这个用户能不能干这件事”。这一层我用RBAC模型(基于角色的访问控制)来实现。

RBAC的核心是给用户分配角色,角色绑定权限,而不是直接给用户绑权限。这样好处很明显:如果公司来了一批实习生,你要批量把某些接口的访问权收回,直接调角色就可以了,不用一个个用户去改权限字段。

3.1 用RBAC模型控制角色

我的项目里有三种角色:admin(管理员)、user(普通用户)、auditor(审计员)。权限矩阵大概是:

角色查看数据创建数据删除数据导出报表审计日志
admin允许允许允许允许允许
user允许允许不允许允许不允许
auditor允许不允许不允许不允许允许

数据库里我用了两张表来维护角色和权限:一张角色表roles,一张角色权限关联表role_permissions。用户表只存角色ID,不直接存权限列表。实际上如果你的角色数量很少(像我这个项目只有三种),完全不需要引入完整的RBAC五张表,直接在User表加一个role字段就够了。

3.2 依赖注入里做权限校验的完整写法

FastAPI里做权限控制最漂亮的一点就是依赖注入。我写了一个require_role函数来生成权限校验依赖:

def require_role(*allowed_roles): def role_checker(user: User = Depends(get_current_user)): if user.role not in allowed_roles: raise HTTPException( status_code=403, detail="权限不足,需要角色:" + ",".join(allowed_roles) ) return user return role_checker

然后路由里这样用:

@router.delete("/users/{user_id}") def delete_user( user_id: int, user: User = Depends(require_role("admin")), ): # 只有admin角色能走到这里 db.query(User).filter(User.id == user_id).delete() db.commit() return {"message": "删除成功"}

这里有个很关键的体验点:权限校验失败时返回403,和认证失败返回401要严格区分。401是“你没有登录或者凭证无效”,403是“你登录了但没资格”。前端拿到401会跳转登录页,拿到403会提示无权限,混用会导致整个前端的错误处理逻辑混乱。

更细粒度的权限可以做成权限点字符串,比如"user:delete"、"order:export",在require_permission里检查当前用户的角色对应的权限点集合:

ROLE_PERMISSIONS = { "admin": {"user:delete", "order:export", "audit:view"}, "user": {"order:create", "order:view"}, "auditor": {"audit:view", "report:view"}, } def require_permission(permission: str): def permission_checker(user: User = Depends(get_current_user)): user_perms = ROLE_PERMISSIONS.get(user.role, set()) if permission not in user_perms: raise HTTPException(status_code=403, detail=f"缺少权限:{permission}") return user return permission_checker

这样权限模型就从“角色”精确到了“操作点”,新接口上线时只要在权限映射表里加一行,不用改动业务代码。

不过要提醒一点:依赖注入可以做接口层面的权限控制,但数据级的权限它管不了。比如用户只能查看自己创建的订单,就不能只靠依赖注入了,必须在业务查询时加上order.creator_id == user.id这样的过滤条件。我管这叫“数据级权限”,RBAC管的是“功能权限”,两者要区分开。

3.3 扩展:数据级权限和权限缓存

数据级权限的实现方式,核心思路是拿到当前用户后去拼接查询条件。比如订单查询接口:

@router.get("/orders") def list_orders(user: User = Depends(get_current_user)): query = db.query(Order) if user.role != "admin": query = query.filter(Order.creator_id == user.id) return query.all()

这种“行级过滤”在ORM里写起来不复杂,但要注意别用错运算符。我见过有人用==过滤时因为类型不匹配(int和str)导致查不到数据,排查了半天才发现是前端的 user_id 传到后端变成了字符串,SQLAlchemy会自动转类型,但从API接收的参数类型要自己在Pydantic模型里声明好。

权限如果每个请求都去查一次数据库,在高并发下会放大数据库压力。我后来给角色的权限点加了一个缓存层,登录时把权限点集合写进Redis,TTL设为15分钟。缓存的代价是权限变更后需要等缓存过期才能生效,权衡之后我选择接受,因为我的权限变化频率很低,15分钟的延迟完全无感。如果你对即时性要求高,可以在修改角色权限时主动删掉相关用户的缓存key。

4. 限流:拦住高频请求的小事不小

限流这个模块,一开始我低估了它。第一版我只写了个简单的请求计数器,上线第二天就在高流量接口上出了乱子,然后才开始正经优化。这一节从简单方案讲到分布式方案。

4.1 最简单的窗口计数方案

如果你只用单机部署,Python的内存计数器就能实现基础限流。核心思路是:以IP+接口路径为key,记录当前时间窗口内已经发生的请求次数,超过阈值直接返回429。

import time from collections import defaultdict rate_store = defaultdict(list) def rate_limit_by_ip(ip: str, limit: int, window: int): now = time.time() req_times = [t for t in rate_store[ip] if now - t < window] req_times.append(now) rate_store[ip] = req_times return len(req_times) <= limit

这个方法有个明显缺陷:每个请求进来都要清一遍过期记录,如果窗口很大、请求量很高,rate_store会膨胀得极快,而且单机内存存不下全量请求记录。还有一个更隐蔽的问题:这个算法在窗口边界处有两个突刺,比如窗口是60秒,第一个59秒期间打了100个请求,然后第60秒初第61秒初又打了100个请求,按窗口算法这算合法,但真实场景已经冲击到服务了。

更稳的单机方案是令牌桶。我用的是slowapi这个库,它是limits库的FastAPI封装,用起来很干净:

from slowapi import Limiter from slowapi.util import get_remote_address limiter = Limiter(key_func=get_remote_address) @limiter.limit("5/minute") @router.get("/sensitive-data") def sensitive_data(): return {"data": "secret"}

这个装饰器的语法是“次数/时间单位”,5/minute就是每分钟5次。它默认用IP做key,也支持自定义key函数,比如按用户ID限流。

4.2 换用Redis做分布式限流

等你的服务从单机变成多副本部署之后,内存方案的局限就出来了:每个副本各自计数,两个副本加起来可以跑出双倍流量。我后来把计数服务器换成了Redis,做滑动窗口限流。

滑动窗口用Redis的ZSet实现非常合适。思路是这样:给每个限流key建一个ZSet,成员是请求时间戳,分数也是时间戳;每次请求来就把当前时间戳塞进去,再把窗口之前的旧记录清掉,然后统计集合长度。下面是关键代码:

import time import redis r = redis.Redis(host="redis_host", port=6379, decode_responses=True) def is_allowed(key: str, limit: int, window: int) -> bool: now = int(time.time() * 1000) # 毫秒时间戳 pipe = r.pipeline() pipe.zadd(key, {now: now}) pipe.zremrangebyscore(key, 0, now - window * 1000) pipe.zcard(key) pipe.expire(key, window) results = pipe.execute() return results[-2] <= limit

这里有几个细节值得说下。第一,用管道执行ZAdd、ZRemRangeByScore、ZCard、Expire四条命令,是为了让它们作为一个原子操作集执行,减少网络往返。第二,ZSet的成员用时间戳本身,虽然同一个毫秒内可能出现相同score,但因为成员值是唯一的,ZSet不会去重。第三,Expire一定要加,否则每个key都会永远留在Redis里,内存迟早爆掉。

毫秒时间戳在这里很重要。如果只用秒级时间戳,同一个秒内有多个请求时,ZAdd的member可能重复导致覆盖,限流就会漏计数。

4.3 限流放中间件还是业务层

限流实现的位置很有讲究。我倾向于把限流放在路由层或者说依赖层,不要混进业务函数里。FastAPI里我用的方案是把它做成一个依赖:

def rate_limit_dependency(limit: int, window: int): def rate_limiter(request: Request): client_ip = request.client.host key = f"rate:{client_ip}:{request.url.path}" if not is_allowed(key, limit, window): raise HTTPException(status_code=429, detail="请求过于频繁,请稍后再试") return rate_limiter

然后接口声明Depends(rate_limit_dependency(limit=10, window=60))。这样限流逻辑和业务逻辑完全隔离,测试时也能单独对限流器写单测。

还有一个容易被忽视的点:限流返回的HTTP状态码要统一用429。很多框架默认会返回403或400,但RFC 6585里已经明确了429代表“太多请求”。前端和网关如果已经约定好了429的语义,你随便换成403会造成前端误判成权限问题。同样的,响应体里最好带上Retry-After头,告诉客户端多久之后再重试,一些优秀的客户端库会自动遵守这个头做退避请求。

从真实用户的角度看,429错误意味着用户被“晾”在等待页面上了,所以限流参数不能拍脑袋定。我做过的内部系统,普通查询接口限流是每分钟120次:一个真人用户盯着一张报表看,操作频率不会超过每分钟20次,120次已经留出很大余量;而导出报表这种重接口我压到每分钟10次,因为导出一次就要跑大量数据库查询,不限制会拖垮整个库。

5. 三个功能串起来之后的常见问题记录

把认证、权限、限流都做好之后,真正的挑战在于它们之间的执行顺序和互相影响。这一节是我实际运行过程中踩过的坑,挑了几个有代表性的记录在这里。

5.1 常见坑速查表

现象根因解决办法
限流不生效,请求量还是很大限流放在了反向代理之前,代理的缓存响应没走后端把限流放在应用层,或者在网关层加同样的规则
Token过期后接口报500JWT的exp字段用了本地时间而非UTC统一用datetime.utcnow()或datetime.now(timezone.utc)
权限改了角色还不生效角色权限缓存未失效修改权限时主动删缓存key
多副本下限流被穿透内存限流没有共享状态换Redis集中计数
bcrypt报AttributeErrorpasslib和bcrypt版本不兼容锁bcrypt版本或换argon2
密码登录永远报“用户名或密码错误”密码哈希用的salt不同,存储时被截断确认password_hash字段长度别设成60,bcrypt哈希长度是60,有时候写60会截断

5.2 依赖顺序和执行顺序

FastAPI的依赖执行顺序和声明的顺序有关。我的接口一般这样组织:

@router.get("/orders/{order_id}") def get_order( order_id: int, user: User = Depends(require_role("admin", "user")), _=Depends(rate_limit_dependency(limit=30, window=60)), ): ...

我习惯把权限放前面、限流放后面。为什么刻意这样排:权限校验要查数据库(取用户),开销更大;限流只是Redis操作,更快。如果请求已经超频、本身就该被拒掉,那先走限流就可以省掉一次数据库查询。反过来如果你把权限放前面,超频请求来了还会白白打一次数据库。这不是绝对最优,但配合Redis做缓存时,请求能先快速被限流器拦截,压力会小很多。

不过要特别提醒:依赖注入的执行顺序不等于FastAPI整个请求生命周期的顺序。FastAPI先执行get_current_user,再执行后面的依赖,短路的机制是:如果前面抛异常,后面的依赖不会被执行。这是好事,意味着如果Token无效,权限校验直接抛401,限流根本不会执行。

5.3 性能与安全的平衡

这三件事全加齐后,最直观的感受就是每个请求的耗时变长了。认证需要解析JWT,权限需要查角色缓存,限流需要写Redis,这些全部加在一起,原本一个5ms的查询接口变成了20ms起步。好在这个开销是值得的。为了压缩这部分损耗,我做了两个优化:

一是把get_current_user里的用户查询加了一层缓存。JWT解析本身很快,瓶颈在“每次请求都从数据库查用户表”。我用Redis缓存了用户的基本信息和角色,TTL设置5分钟。带来的风险就是用户角色被变更后最多延迟5分钟生效,这个项目里是可接受的。

二是对令牌做过期时间的容忍度做了调优。JWT是解耦了服务端状态的,用户改密码之后旧的Token依然有效直到过期。我在改密码的接口里加了一个“令牌版本号”字段,存在JWT里,改密码时把版本号加一。这样老Token的版本号对不上,就会被get_current_user直接拒绝,不需要额外查库。

三是从网络层到应用层做好层级限流。我做过的最优配置是:网关层按IP限流(拦截扫描和刷量),应用层按用户+接口限流(拦截单用户高频点击),两层互相兜底。网关用反向代理的模块配置,应用层用Redis限流器,两边的阈值不要一样,网关层要宽松一些,否则用户根本没有机会走到应用层拿到友好的429提示,看到的是一长串网关错误页。

6. 写在最后的实操体会

这套系统从写第一版到现在稳定跑了大半年,我个人的体会是:认证、权限、限流这三个东西,真正难的不是某个单独功能的实现,而是把它们放进一个清晰的框架里,让它们互相配合又不互相干扰。

如果你现在是从零开始接一个新项目,我的建议是按“先认证、再限流、最后权限”的顺序来搭建:认证是一切的基础,限流可以比较早地保护你的开发环境不被扫爆,权限放到业务接口多起来之后再做也不迟。但如果你接手的是一个已经跑起来的旧项目,那就反过来,先把权限的硬编码判断收拢,再逐步补限流,最后再统一认证方案。

后端的信任链条是从“这个人是谁”开始的,认证、权限、限流这三环扣得越紧,系统就越扛得住流量和攻击。希望我这篇踩坑总结能帮你少走几段弯路。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询