这几年接手的Python后端项目,十有八九逃不开认证、权限、限流这三件套。你随便起一个新接口,自己心里就得过三关:你是谁,你能碰什么,你一分钟能用几次。这三件事不做好,后面全是补窟窿——被爬虫刷爆短信接口、离职员工还拿老token调生产接口、一个用户一秒请求200次把数据库拖垮,我都见过。
其实这些概念离我们一点都不远。连一次公司WiFi,先弹出来一个认证页面;想删除一个被占用的文件,Windows会提示你需要管理员权限;开放平台开放一个API,文档里必然有一行“每分钟最多调用60次”。它们对应的就是认证、权限和限流。
这篇文章我准备用Python把这条链路完整串一遍,从最基础的概念区分开始,到选型,再到基于FastAPI的落地实现,最后把限流算法和分布式改造讲透。适合正在写Web API、想做小程序后端,或者准备面试梳理这一类问题的朋友。
1. 先把概念说清楚:认证、权限、限流各管哪一段
1.1 认证:先证明“你是你”
认证英文叫Authentication,核心问题就是“你是谁”。登录是认证,录入指纹是认证,扫码也是认证。在Web系统里,HTTP请求是无状态的,第一次登录成功之后,后续每个请求服务端都得知道这次请求是谁发来的,这就是认证要解决的。
Python里的认证实现方式,说白了就是生成凭证和校验凭证。凭证可以是服务端存在内存里的Session ID,也可以是无状态的JWT(JSON Web Token),还可以是给机器用的API Key。凭证本身没有业务意义,它的意义是“经过服务器签名或存储、可以证明身份的东西”。
一个经常被问的易混点是Authentication和Authorization的区别:认证管“你是谁”,授权管“你能干什么”。别把这两个词混在一起说,面试时很容易减分。
企业内部还有网络层面的认证,比如RADIUS协议配合802.1X做准入控制,Python里也能通过pyrad这类库对接,但Web应用层面最常用的还是上面这三种。下文的代码我统一用JWT示范,因为前后端分离项目里它最好落地。
1.2 权限:证明完了还要看“你能碰什么”
认证通过后,下一个问题是授权。英文叫Authorization,但中文社区经常把“权限”和“授权”混着用。权限控制的本质是一张映射表:什么样的用户,能对什么样的资源,执行什么样的操作。
最常见的模型是RBAC(Role-Based Access Control,基于角色的访问控制)。RBAC把权限挂到角色上,角色再挂到用户上。为什么要多一层角色?因为如果权限直接挂到人身上,一个1000人的公司,权限调一次要改1000条记录;有了角色,只需要改角色和用户的关联关系。这个思想跟操作系统的用户组、文件权限组本质上是一样的,只是Web系统里我们把“文件权限”换成了“接口权限”和“数据权限”。
权限的控制粒度至少要分清三层:接口级权限(能不能调用某个URL)、模块级权限(能不能看到某个菜单)、行级权限(能不能看到某一条订单数据)。很多系统只做了前两层,行级没做,结果普通用户调接口把别人订单也捞出来了。这种问题比接口权限缺失更隐蔽,也更危险。
1.3 限流:管住频率,别让系统被冲垮
限流(Rate Limiting)跟前两者维度不同,它不关心你是谁,关心的是你用多快。就像高速收费站,车能过,但每个ETC通道一分钟过多少辆是有上限的,超出就得排队。
为什么要限流?最典型的是登录接口。如果不限流,攻击者可以对用户名密码做一个字典爆破,一秒打几千次;短信验证码接口不限流,能被脚本刷到供应商欠费;开放API不限流,某个大客户写了个死循环,能把整个服务拖垮。限流是保护系统可用性的一道闸门,也是控制成本的一种手段。
在具体实现上,限流会有固定窗口、滑动窗口、令牌桶、漏桶等算法,后面第4章我会详细对比。先记住限流有两种最常见的维度:按IP限流和按用户限流,生产环境通常两者结合用。
需要模型API调用? 免费领10W Token,多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. 技术选型:Python生态里怎么选组件
2.1 认证方案选型:Session、JWT还是OAuth2
先说结论:自建用户体系、前后端分离的API项目,用JWT最稳妥;传统服务端渲染的老项目,用Session继承成本最低;将来要做第三方开放平台,直接上OAuth2,别自己发明协议。
| 方案 | 状态存储 | 适合场景 | 撤销难易 | 跨域/移动端适配 |
|---|---|---|---|---|
| Session-Cookie | 服务端 | 传统Web应用 | 容易 | 一般 |
| JWT | 无状态 | 前后端分离API、移动端 | 较难 | 好 |
| OAuth2 | 授权服务器统一管理 | 第三方授权、开放平台 | 由授权服务器控制 | 好 |
| API Key | 服务端记录 | M2M内部服务 | 容易 | 好 |
表格里“撤销难易”对JWT要特别解释一下。JWT无状态,签发之后在过期前服务器没法主动让token失效。解决办法是引入token版本号,或者维护一个黑名单,但那等于又引入了状态。所以很多团队“看起来用了JWT,实际还是把session换了个地方存”,这也是我建议你上线之前想清楚的一点:token能不能撤销,比选什么格式更重要。
Python这边,PyJWT是纯JWT编解码,python-jose支持的算法更多,Authlib适合做OAuth2客户端/服务端。密码哈希我用passlib配合bcrypt,后面代码里会用到。
2.2 权限模型选型:RBAC先跑通,再考虑ABAC
权限模型别一上来就上重武器。我的建议:90%的后台管理系统,RBAC就够用。权限表、角色表、用户角色关联表,三张表跑通,再配合FastAPI的一个依赖注入函数,接口权限就能控制住了。
ABAC(基于属性的访问控制)适合什么场景?风控、多租户、复杂规则过滤。比如“当用户所在区域为华东,且订单金额大于5000时,允许审批”,这种条件组合用RBAC写起来很痛苦。Python里有开源引擎Casbin的Python版,规则写在model.conf里,支持RBAC/ABAC/ACL,功能很强,但学习成本也不低,小项目不建议为了炫技引入。
如果你看到项目里权限直接写在每个接口的一堆if里,那就说明权限模型还没建立起来。先统一成角色,再谈扩展,这个顺序不要反过来。
2.3 限流组件选型:轮子还是自己造
Python限流生态没有Java那边那么统一。Java有Sentinel、Guava RateLimiter,Python这边常见路线:
limits库:算法库,支持固定窗口、滑动窗口、令牌桶等,不绑定框架。slowapi:FastAPI的限流插件,底层就是limits,装饰器写法很舒服。Flask-Limiter:Flask场景的经典选择。- 自研Redis版:用INCR+EXPIRE,或者Lua脚本,需要控制原子性。
我的建议是先了解算法,再决定用哪个库。因为限流的坑多数不在“调用库”阶段,而在“算法选错导致误杀”或者“多实例下计数不准”这些看不见的地方。
3. 落地实现:用FastAPI逐步搭一套认证+权限+限流
3.1 先搭骨架:依赖安装与目录结构
我选FastAPI做演示,理由很直接:类型提示好、依赖注入方便、自带OpenAPI文档,能把认证、权限、限流三件事用最少的代码串起来。你先建一个虚拟环境,再装这些依赖:
bash复制pip install fastapi uvicorn pyjwt passlib[bcrypt] redis limits
目录结构我习惯按功能拆,而不是按文件类型拆:
code复制app/
main.py # FastAPI入口
auth.py # 认证:密码哈希、JWT签发与校验
rbac.py # 权限:角色依赖
rate_limit.py # 限流:内存版与Redis版
这样每个文件只干一件事,后面加功能不打架。
3.2 认证链路:登录后签发JWT,请求头里校验JWT
先写一个简单的用户存储,实际项目替换成数据库就行:
python复制# auth.py
from datetime import datetime, timedelta, timezone
from typing import Optional
import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from passlib.context import CryptContext
SECRET_KEY = "please-change-me"
ALGORITHM = "HS256"
TOKEN_EXPIRE_MINUTES = 30
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
bearer_scheme = HTTPBearer()
# 演示用内存用户表,生产环境换成数据库查询
USERS = {
"admin": {
"username": "admin",
"password_hash": pwd_context.hash("admin123"),
"role": "admin",
},
"alice": {
"username": "alice",
"password_hash": pwd_context.hash("alice123"),
"role": "user",
},
}
def hash_password(password: str) -> str:
return pwd_context.hash(password)
def verify_password(plain_password: str, password_hash: str) -> bool:
return pwd_context.verify(plain_password, password_hash)
def create_token(username: str, role: str) -> str:
payload = {
"sub": username,
"role": role,
"exp": datetime.now(timezone.utc) + timedelta(minutes=TOKEN_EXPIRE_MINUTES),
}
return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
def get_current_user(
credentials: HTTPAuthorizationCredentials = Depends(bearer_scheme),
) -> dict:
token = credentials.credentials
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username = payload.get("sub")
user = USERS.get(username)
if user is None:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="用户不存在")
return user
except jwt.ExpiredSignatureError:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Token已过期")
except jwt.InvalidTokenError:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Token无效")
密码为什么要用bcrypt而不是直接MD5?因为用户数据库一旦泄露,MD5可以用彩虹表批量逆出明文。bcrypt是慢哈希,每次计算故意设计得慢,还带随机盐,相同密码也会得到不同哈希值,攻击成本高得多。这个细节面试经常问,工程里更不能省。
注意SECRET_KEY别写死在代码里,我从配置中心或环境变量读取。密钥一旦泄露,任何人都能伪造token,这是认证体系最核心的资产。
然后在main.py里写登录接口:
python复制# main.py
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from auth import USERS, create_token, verify_password, get_current_user
app = FastAPI()
@app.post("/login")
def login(form_data: OAuth2PasswordRequestForm = Depends()):
user = USERS.get(form_data.username)
if not user or not verify_password(form_data.password, user["password_hash"]):
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="用户名或密码错误")
token = create_token(user["username"], user["role"])
return {"access_token": token, "token_type": "bearer"}
@app.get("/me")
def me(current_user: dict = Depends(get_current_user)):
return {"username": current_user["username"], "role": current_user["role"]}
跑一下uvicorn main:app --reload,先调/login拿token,再带上Authorization: Bearer <token>调/me,就能看到当前用户信息了。这一步跑通,认证闭环就完成了。
3.3 权限链路:用依赖注入实现角色控制
权限校验必须放在认证之后。FastAPI里最简单的方式是定义一个依赖工厂:
python复制# rbac.py
from fastapi import Depends, HTTPException, status
from auth import get_current_user
def require_roles(*roles: str):
def checker(user: dict = Depends(get_current_user)):
if user["role"] not in roles:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="权限不足")
return user
return checker
使用的时候:
python复制@app.get("/admin/users")
def list_users(user: dict = Depends(require_roles("admin"))):
return {"user": user["username"], "message": "管理员可见"}
@app.delete("/orders/{order_id}")
def delete_order(order_id: int, user: dict = Depends(require_roles("admin"))):
# 演示代码,实际请删除对应记录
return {"deleted": order_id, "operator": user["username"]}
这里有个很容易搞混的点:认证失败返回401,权限不足返回403。401的意思是“你没证明你是谁”,403的意思是“我知道你是谁,但你不够格”。如果你看到接口对未登录和登录了但没权限的用户都返回403,多半是权限校验没有建立在认证之上。
行级权限怎么处理?关键要拿到“当前用户”和“资源属主”两个维度。比如用户只能查自己的订单:
python复制@app.get("/orders/{order_id}")
def get_order(order_id: int, user: dict = Depends(get_current_user)):
order = fake_order_db.get(order_id)
if order is None:
raise HTTPException(status_code=404, detail="订单不存在")
if order["owner"] != user["username"] and user["role"] != "admin":
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="无权查看该订单")
return order
这条规则的意思是:普通用户只能碰自己的数据,管理员可以跨用户。很多权限漏洞就出在这一层被省略,前端只隐藏了入口,后端接口却照常返回全部数据。
3.4 限流链路:先上内存版,再切Redis版
内存版最简单,我先写一个基于字典的固定窗口计数器,适合单机开发环境:
python复制# rate_limit.py
import threading
import time
from fastapi import HTTPException, Request
class FixedWindowLimiter:
def __init__(self, max_requests: int, window_seconds: int):
self.max_requests = max_requests
self.window_seconds = window_seconds
self._buckets: dict[str, list] = {}
self._lock = threading.Lock()
def check(self, key: str) -> bool:
now = time.time()
with self._lock:
window_start = self._buckets.get(key, [0, 0])
if now - window_start[0] > self.window_seconds:
self._buckets[key] = [now, 1]
return True
if window_start[1] >= self.max_requests:
return False
window_start[1] += 1
return True
rate_limiter = FixedWindowLimiter(max_requests=5, window_seconds=60)
然后在main.py里引入并使用:
python复制from rate_limit import rate_limiter
@app.post("/login")
def login(request: Request, form_data: OAuth2PasswordRequestForm = Depends()):
client_ip = request.client.host
if not rate_limiter.check(client_ip):
raise HTTPException(status_code=429, detail="请求过于频繁,请稍后再试")
# ...后续登录逻辑
登录接口按IP限流是最基础的动作,不然密码爆破脚本能把你的登录接口打穿。但只按IP也有问题,公司里几十个人共享一个出口IP,误伤率很高。更好的做法是IP+用户名双维度限流,比如同一IP一分钟不能超过100次,同一用户名一分钟只允许尝试5次。这样既防爆破,又不容易把同一栋楼的人全部误杀。
生产环境用内存版不行:FastAPI起两个worker,每个worker的内存是独立的,等于限流额度翻倍。所以多实例场景下必须用Redis。Redis版的固定窗口两条命令就能实现:
python复制import redis
r = redis.Redis(host="localhost", port=6379, decode_responses=True)
def redis_fixed_window(key: str, max_requests: int, window_seconds: int) -> bool:
current = r.incr(key)
if current == 1:
r.expire(key, window_seconds)
return current <= max_requests
两条命令要保证原子性,理想情况应该放进Lua脚本里执行,这样并发请求不会出现“同时incr”导致的误判。Redis官方也推荐这种做法,第4章我会给完整脚本。
4. 进阶细节:限流算法原理与分布式改造
4.1 四种限流算法怎么选:固定窗口、滑动窗口、令牌桶、漏桶
先看一张对比表,这张表是我面试别人时最常用的考点:
| 算法 | 实现难度 | 是否允许突发 | 输出是否平滑 | 典型场景 |
|---|---|---|---|---|
| 固定窗口 | 低 | 窗口边界可双倍突发 | 不平滑 | 简单计数、快速上线 |
| 滑动窗口 | 中 | 基本削平突发 | 较平滑 | 常规API限流 |
| 令牌桶 | 中 | 允许有限突发 | 平滑 | 一般网关、开放API |
| 漏桶 | 中 | 不允许突发 | 严格匀速 | 保护脆弱下游 |
固定窗口的问题最典型:假设限制每分钟100次,第59秒来了100次,第61秒又来了100次,中间只隔着2秒,却通过了200次请求。这就是窗口边界突发。滑动窗口就是为了解决这个问题,它在Redis里常用ZSET记录每次请求的时间戳,取窗口内的数量做判断:
python复制import time
def sliding_window(user_key: str, max_requests: int, window_seconds: int) -> bool:
key = f"sliding:{user_key}"
now = time.time()
pipeline = r.pipeline()
pipeline.zremrangebyscore(key, 0, now - window_seconds)
pipeline.zadd(key, {str(now): now})
pipeline.expire(key, window_seconds)
pipeline.zcard(key)
results = pipeline.execute()
count = results[-1]
return count <= max_requests
这段代码把过期时间窗口之外的记录先清理掉,再插入当前请求,然后统计窗口内的记录数。实际项目建议用随机串做ZSET成员,避免同一毫秒内多个请求覆盖同一个member。
令牌桶的思路是:用一个桶装令牌,以固定速率往桶里放令牌,请求来了必须拿走一个令牌才能通过。桶有容量上限,所以允许短时间突发,但桶空了就得等。它比滑动窗口更接近大多数API网关的默认行为。如果你用limits库,limits.strategies里有现成的令牌桶实现,不必自己造。
漏桶大家可能用得少,它的核心是请求先进队列,按固定速率往外发放。下游系统能力固定时用漏桶最保险,比如外呼平台每秒最多能处理10通电话,你用漏桶就能把上游的突发流量强制舒缓成匀速。
4.2 网关层限流和应用层限流的分工
很多同学问:Nginx都能限流,为什么还要在Python里做?答案是维度不同。
Nginx的limit_req是按IP或Nginx变量限流的,它工作在HTTP入口,拦截最暴力最底层的流量很划算,几行配置就能挡住大流量攻击。但它不知道业务用户是谁,无法按用户、按接口维度限流。
应用层限流则可以拿到完整上下文:当前登录用户、用户等级、API类型。比如普通用户每分钟100次,付费用户每分钟1000次,这种策略只能放在应用层。所以成熟系统的常态是两层配合:Nginx先挡住IP洪峰,应用层按用户维度做精细控制。遇到超大流量时,甚至可以再往前的CDN层面动手,但那属于运维架构的范畴了。
Python社区经常有人问“有没有Sentinel这样的东西”。Java里Sentinel是限流、熔断、降级一体的框架,Python这边没有完全对等的计划,但拆开来看都有替代:限流用limits+Redis,熔断可以自己写一个计数器或状态机,降级可以配合信号量限制并发。很多小型团队把“限流”和“熔断”混为一谈,其实限流是限制进入量,熔断是发现下游故障时快速失败,保护的不是自己而是下游。别在用词上把这两个概念搞混,架构评审时比较容易出问题。
4.3 分布式限流的原子性与一致性:Redis Lua脚本
多实例部署后,进程内限流肯定不准,Redis限流成为标配。但INCR和EXPIRE分开执行,会在极端并发下有窗口漏洞——两个请求同时在INCR之前读到旧值,可能把计数覆盖掉。正确做法是把判断、计数、过期合成一个Lua脚本,利用Redis单线程执行脚本的原子性:
lua复制local key = KEYS[1]
local limit = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local current = redis.call('INCR', key)
if current == 1 then
redis.call('EXPIRE', key, window)
end
if current > limit then
return 0
end
return 1
在Python里调用:
python复制import redis
r = redis.Redis(host="localhost", port=6379, decode_responses=True)
rate_limit_script = """
local key = KEYS[1]
local limit = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local current = redis.call('INCR', key)
if current == 1 then
redis.call('EXPIRE', key, window)
end
if current > limit then
return 0
end
return 1
"""
def check_rate_limit(user_key: str, limit: int, window: int) -> bool:
result = r.eval(rate_limit_script, 1, f"rate:{user_key}", limit, window)
return result == 1
这段Lua最终实现的是固定窗口限流,胜在原子且只用了两次Redis操作。如果要精确削峰,再把第4.1节的滑动窗口逻辑搬进Lua。这里有个取舍:Redis限流通常能扛住绝大多数业务,但单个Redis实例本身也可能成为瓶颈,极大规模场景会退化为本地限流加Redis兜底的混合模式。那种方案注意点很多,我说一句实在话:绝大多数团队还没到那一步,先把Redis版做好,性能问题等压测数据说话。
5. 常见问题与排查经验
5.1 认证环节容易踩的坑
我把实际项目里见过最多的认证问题整理成一个速查表:
| 现象 | 可能原因 | 处理建议 |
|---|---|---|
| 前端带token但接口一直401 | Header少了Bearer 前缀 |
检查Authorization格式,标准是Bearer <token> |
| token过期后前端仍显示登录 | 前端没有拦截401并跳登录页 | 在请求库统一处理401响应 |
| 日志里能看到用户密码 | 框架默认记录请求体 | 关闭access log的请求体或做脱敏 |
| 同一用户旧token一直有效 | JWT无状态地天然存在 | 引入token版本号或黑名单 |
| 修改密码后老token仍能用 | 同上 | 用户表存token_version,签发时写入payload |
这里特别提醒一个容易被忽略的点:JWT的payload只是Base64编码,不是加密。任何拿到token的人都可能解码查看内容,所以不要在payload里放手机号、邮箱、身份证这类敏感字段。签名只是防篡改,不是防窥视。
5.2 权限控制的边界问题
权限控制最大的坑不在技术,而在“你以为控制住了”。最常见的情况是前端把按钮隐藏了,后端接口却没有校验。前端隐藏只是体验优化,后端校验才是安全边界,这个顺序不要颠倒。
第二个高频问题是在路由里绑定了接口权限,但数据权限没做。比如“只有管理员能查看用户列表”做到了,但普通用户传入一个订单ID,照样能查别人订单。解决办法是每个查询都带上owner = current_user.username条件,管理员再放开过滤条件。
第三是权限编码没有统一。有人把角色判断写死成if user.username == "admin",一旦以后有多个管理员,这套代码就得全改。角色和用户必须分开,权限模型的核心是“以角色为中间层”,不是“以用户名为中间层”。
5.3 限流失效与误杀:排查思路与经验
排查限流问题,我一般先问三个问题:按什么维度限的?阈值是多少?响应是什么?然后分情况看:
| 现象 | 排查思路 |
|---|---|
| 限流完全没生效 | 看限流逻辑是否真的在业务代码之前执行,多worker部署是否用了内存限流 |
| 正常用户被误杀 | 检查是否只按IP限流,出口IP是否被多人共享 |
| 被限流后仍大量请求进入 | 检查限流是否发生在业务代码之后,正确顺序是中间件/依赖靠前 |
| Redis key不断膨胀 | 确认是否忘记设置过期时间,以及key命名是否包含用户ID |
| 压测时误差很大 | 检查Redis网络耗时,考虑用pipeline或Lua合并操作 |
还有个实操经验:限流响应头最好带上剩余配额,比如X-RateLimit-Remaining。这样前端能提前给用户提示,而不是等429来了再处理;后端排查时看这个头也能判断是不是限流逻辑本身出了问题。登录接口的限流阈值不要拍脑袋,可以先从压测数据反推:正常情况下你希望同一用户名一分钟允许多少次密码尝试,低于这个值就会被误杀。
6. 写在最后的几点实战心得
我做了这些年后端,一个很深的感受是:认证、权限、限流这三件事,难点从来不在“会不会写代码”,而在于设计时有没有把他们当成一个整体来考虑。认证是所有请求的入口,权限是业务数据的边界,限流是系统最后一道防线。只要你在项目第一天就把current_user依赖留好、把角色模型定下来、给登录接口加上限流,后面就不会出现“三天两头补安全窟窿”的状态。
再分享一个小技巧:上线前可以用压测工具模拟一下限流阈值,观察返回429的时间点和系统CPU、内存的表现,能帮你找到很多开发环境发现不了的配置问题。如果你正在做一个新项目,我建议把这三个功能做成公共依赖,而不是散落在各个业务接口里,长期维护下来会省非常多事。
