04. Web 服务开发:认证、权限、流式、限流

读完这篇你能:

给 API 加 JWT 认证,用户登录拿 token,后续请求带 token

写 RBAC 权限系统(普通用户/管理员能干不同的事)

实现 ChatGPT 式流式输出(SSE),token 边生成边返回

用 slowapi 做限流,防止单用户打垮系统

对应架构层:应用层(进阶)

前置知识:01. Web 基础、02. 数据库、03. 缓存与存储

预计用时:4 小时(含动手)

0. 为什么算法工程师要学这个

前三篇你写了:

01:能跑的 API
02:数据能存 Postgres
03:加了 Redis 缓存、S3 文件、pgvector

但这个 API 有四个致命问题:

谁都能调——没有登录,任何人都能调你的 LLM 接口,烧你的钱
所有人权限一样——没法区分免费用户、付费用户、管理员
体验差——LLM 生成要 10 秒,用户盯着 loading 看 10 秒,中间没有任何反馈
会被打垮——一个脚本一秒发 1000 个请求,你的 LLM 额度瞬间耗尽

这一篇就是解决这四件事:

问题	技术	行业叫法
谁都能调	JWT token	认证(Authentication)
权限一样	RBAC	授权(Authorization)
体验差	SSE 流式	Streaming Response
会被打垮	slowapi	Rate Limiting

学完这一篇,你的代码就具备了生产级 LLM 后端服务的骨架。

1. 核心概念(最少必要理论)

1.1 认证 vs 授权(必背)

概念	英文	回答的问题	例子
认证	Authentication / AuthN	你是谁?	登录(账号密码)
授权	Authorization / AuthZ	你能干什么?	管理员能删用户,普通用户不能

记忆法:AuthN 里的 N = "Who"(身份),AuthZ 里的 Z = "Zero or Not"(能不能)。

1.2 认证方式对比

方式	怎么工作	优缺点	什么时候用
Session + Cookie	服务端存 session,客户端存 cookie id	简单,但服务端有状态,扩展难	传统 Web 应用
JWT	客户端存 token,服务端无状态	无状态、跨域、跨服务	前后端分离、API、移动端
OAuth 2.0	用第三方登录(Google、GitHub)	不用自己管账号	SSO、第三方登录
API Key	长期有效的密钥	简单,但泄露风险高	服务间调用、开发期

本篇讲 JWT——LLM 应用最常用。

1.3 JWT 是什么

JWT(JSON Web Token) 是一种紧凑、自包含的 token 格式:

eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4iLCJyb2xlIjoiYWRtaW4ifQ.SflKxwRJSMeKKF2QT4fwpMeJf36POk6yJV_adQssw5c

三部分(. 分隔):

Header:{"alg": "HS256", "typ": "JWT"} —— 算法声明
Payload:{"sub": "123", "name": "John", "role": "admin", "exp": 1717200000} —— 真正的数据
Signature:用密钥(只有服务端知道)对前两部分签名

工作流程:

用户 POST /login {email, password}
服务端验证密码 → 生成 JWT(签名)→ 返回
客户端存 token(localStorage / cookie)
后续请求:Authorization: Bearer <token>
服务端验证签名 + 检查过期 → 解析出 user_id

JWT 的核心特性:

无状态:服务端不存 session,任何服务实例都能验证
不可篡改:改 payload 会导致签名不匹配
可读:payload 是 Base64,任何人能解码看到内容(所以不要放密码)
有过期时间:exp 字段

类比:JWT 像机场的"登机牌"——上面有你的名字、航班、座位,机场每个工作人员扫一下都能验证真伪,不用打电话给售票处确认。

1.4 RBAC(Role-Based Access Control)

RBAC = 基于角色的权限控制。

不直接给用户分配权限,而是:

用户(User)──分配──▶ 角色(Role)──拥有──▶ 权限(Permission)

典型角色:

viewer(只读)
user(读写自己的)
editor(读写所有人的)
admin(全部权限 + 管理用户)

FastAPI 里的实现:

from fastapi import Depends

def require_role(*roles):
    def checker(user: User = Depends(get_current_user)):
        if user.role not in roles:
            raise HTTPException(403, "Permission denied")
        return user
    return checker

@app.get("/api/users")
def list_users(_: User = Depends(require_role("admin"))):
    ...

1.5 流式响应(SSE)

问题:LLM 生成 1000 token 要 10 秒,用户盯着 loading 看 10 秒——体验灾难。

解决:Server-Sent Events(SSE),让 token 边生成边推送。

原理:

普通 HTTP:
   客户端请求 → 服务端等 10 秒生成完 → 一次性返回 1000 token

SSE 流式:
   客户端请求 → 服务端开一个长连接 → 每生成一个 token 就推一段
   客户端在 1 秒后就开始看到字蹦出来

SSE 数据格式:

data: {"token":"你"}

data: {"token":"好"}

data: {"token":","}

data: [DONE]

每行 data: 开头,空行分隔。

vs WebSocket:

维度	SSE	WebSocket
方向	服务端 → 客户端(单向)	双向
协议	HTTP	单独协议(握手后升级)
复杂度	低	高
用途	LLM 流式输出、推送	聊天、协同编辑

LLM 应用默认用 SSE,因为生成是单向的(服务端 → 客户端)。双向交互(多轮聊天实时同步)才用 WebSocket。

1.6 限流(Rate Limiting)

为什么:一个恶意用户一秒发 1000 个请求,LLM 额度瞬间烧光,正常用户全被拖累。

限流策略:

算法	怎么工作	优缺点
固定窗口	每分钟最多 N 次	简单,但窗口边界有突刺(59 秒发 N + 1 秒发 N = 2N)
滑动窗口	任意时刻往前看 1 分钟	平滑,但实现复杂
令牌桶(Token Bucket)	桶里放令牌,每秒加 N 个,每个请求拿 1 个	突发友好,行业主流
漏桶(Leaky Bucket)	请求像水滴,以固定速率漏出	严格平滑

限流维度:

全局限流(整个 API)
用户级限流(free: 10/分钟,paid: 100/分钟)
IP 级限流(防爬虫)

Python 里用 slowapi:

from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)

@app.post("/api/summarize")
@limiter.limit("10/minute")
async def summarize(req, request: Request):
    ...

2. 实战:把前 3 篇整合成完整服务

2.1 装包

pip install "passlib[bcrypt]" python-jose[cryptography] slowapi

2.2 密码 Hash 工具

绝对不能明文存密码。新建 security.py:

from passlib.context import CryptContext
from jose import jwt, JWTError
from datetime import datetime, timedelta, timezone

SECRET_KEY = "CHANGE_ME_TO_A_LONG_RANDOM_STRING"  # 生产用环境变量
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24  # 24 小时

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

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

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_ctx.verify(plain, hashed)

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

def decode_token(token: str) -> dict | None:
    try:
        return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    except JWTError:
        return None

生成 SECRET_KEY:

python -c "import secrets; print(secrets.token_urlsafe(32))"

2.3 User 模型加 role 字段

改 models.py:

from enum import Enum

class UserRole(str, Enum):
    viewer = "viewer"
    user = "user"
    admin = "admin"

class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    email: str = Field(unique=True, index=True)
    name: str
    hashed_password: str
    role: UserRole = UserRole.user  # 默认普通用户
    created_at: datetime = Field(default_factory=datetime.utcnow)

2.4 登录接口

from fastapi.security import OAuth2PasswordBearer
from fastapi import Depends, HTTPException, status

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/login")

@app.post("/api/auth/register")
def register(email: str, password: str, name: str, session: Session = Depends(get_session)):
    if session.exec(select(User).where(User.email == email)).first():
        raise HTTPException(400, "Email already registered")
    user = User(email=email, name=name, hashed_password=hash_password(password))
    session.add(user)
    session.commit()
    session.refresh(user)
    return {"id": user.id, "email": user.email}

@app.post("/api/auth/login")
def login(email: str, password: str, session: Session = Depends(get_session)):
    user = session.exec(select(User).where(User.email == email)).first()
    if not user or not verify_password(password, user.hashed_password):
        raise HTTPException(401, "Invalid credentials")
    token = create_access_token({"sub": str(user.id), "role": user.role.value})
    return {"access_token": token, "token_type": "bearer"}

2.5 认证依赖

def get_current_user(
    token: str = Depends(oauth2_scheme),
    session: Session = Depends(get_session),
) -> User:
    payload = decode_token(token)
    if not payload:
        raise HTTPException(401, "Invalid token")
    user_id = int(payload["sub"])
    user = session.get(User, user_id)
    if not user:
        raise HTTPException(401, "User not found")
    return user

def require_role(*roles: UserRole):
    def checker(user: User = Depends(get_current_user)) -> User:
        if user.role not in roles:
            raise HTTPException(403, f"Requires role: {roles}")
        return user
    return checker

使用:

# 任何登录用户都能调
@app.get("/api/me")
def me(user: User = Depends(get_current_user)):
    return user

# 只有 admin 能调
@app.delete("/api/users/{user_id}")
def delete_user(user_id: int, _: User = Depends(require_role(UserRole.admin)),
                session: Session = Depends(get_session)):
    user = session.get(User, user_id)
    if user:
        session.delete(user)
        session.commit()
    return {"deleted": user_id}

测试:

# 登录拿 token
TOKEN=$(curl -s -X POST "http://localhost:8000/api/auth/[email protected]&password=xxx" | jq -r .access_token)

# 带 token 调接口
curl http://localhost:8000/api/me -H "Authorization: Bearer $TOKEN"

2.6 改 summarize 用认证用户

@app.post("/api/summarize")
async def summarize(
    req: SummarizeRequest,
    user: User = Depends(get_current_user),  # ✅ 必须登录
    session: Session = Depends(get_session),
):
    # 现在 user.id 是从 token 解出来的,不是 body 里传的
    cache_key = make_cache_key("summarize", user.id, req.text, req.max_length)
    cached = cache_get(cache_key)
    if cached:
        return cached

    resp = await client.chat.completions.create(...)
    # ... 写数据库记录、写缓存

2.7 流式输出(SSE)

核心:用 StreamingResponse 把生成器包装起来。

from fastapi.responses import StreamingResponse
import json

@app.post("/api/summarize/stream")
async def summarize_stream(
    req: SummarizeRequest,
    user: User = Depends(get_current_user),
):
    async def event_stream():
        full_text = ""
        try:
            stream = await client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[
                    {"role": "system", "content": "总结以下文本"},
                    {"role": "user", "content": req.text},
                ],
                stream=True,  # ✅ 关键:开启流式
            )
            async for chunk in stream:
                delta = chunk.choices[0].delta.content
                if delta:
                    full_text += delta
                    yield f"data: {json.dumps({'token': delta})}\n\n"

            yield f"data: {json.dumps({'done': True, 'full_text': full_text})}\n\n"
        except Exception as e:
            yield f"data: {json.dumps({'error': str(e)})}\n\n"

    return StreamingResponse(event_stream(), media_type="text/event-stream")

前端消费(JavaScript):

const response = await fetch('/api/summarize/stream', {
  method: 'POST',
  headers: {
    'Content-Type': 'application/json',
    'Authorization': `Bearer ${token}`,
  },
  body: JSON.stringify({ text: '...' }),
})

const reader = response.body.getReader()
const decoder = new TextDecoder()

while (true) {
  const { done, value } = await reader.read()
  if (done) break
  const text = decoder.decode(value)
  // 解析 "data: {...}\n\n" 格式
  text.split('\n\n').forEach(chunk => {
    if (chunk.startsWith('data: ')) {
      const data = JSON.parse(chunk.slice(6))
      if (data.token) {
        document.getElementById('output').textContent += data.token
      }
    }
  })
}

2.8 限流

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.post("/api/summarize")
@limiter.limit("10/minute")  # 每个 IP 每分钟 10 次
async def summarize(
    request: Request,  # ✅ 限流必须注入 request
    req: SummarizeRequest,
    user: User = Depends(get_current_user),
):
    ...

按用户限流(更精细):

def user_or_ip_key(request: Request) -> str:
    token = request.headers.get("Authorization", "").replace("Bearer ", "")
    payload = decode_token(token)
    if payload:
        return f"user:{payload['sub']}"
    return f"ip:{get_remote_address(request)}"

limiter = Limiter(key_func=user_or_ip_key)

3. 进阶:生产级改造

3.1 Refresh Token

Access token 短期(1 小时),Refresh token 长期(7 天):

Access token 泄露:1 小时后失效
Refresh token 泄露:可以一直刷 access token

REFRESH_TOKEN_EXPIRE_DAYS = 7

def create_refresh_token(data: dict) -> str:
    expire = datetime.now(timezone.utc) + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
    return jwt.encode({**data, "exp": expire, "type": "refresh"}, ...)

@app.post("/api/auth/refresh")
def refresh(refresh_token: str, session = Depends(get_session)):
    payload = decode_token(refresh_token)
    if not payload or payload.get("type") != "refresh":
        raise HTTPException(401, "Invalid refresh token")
    user = session.get(User, int(payload["sub"]))
    new_access = create_access_token({"sub": str(user.id), "role": user.role.value})
    return {"access_token": new_access}

3.2 OAuth 2.0(第三方登录)

用 Google 登录示例:

pip install authlib

from authlib.integrations.starlette_client import OAuth

oauth = OAuth()
oauth.register(
    name="google",
    client_id="xxx.apps.googleusercontent.com",
    client_secret="GOCSPX-xxx",
    server_metadata_url="https://accounts.google.com/.well-known/openid-configuration",
    client_kwargs={"scope": "openid email profile"},
)

@app.get("/api/auth/google/login")
async def google_login(request: Request):
    redirect_uri = "http://localhost:8000/api/auth/google/callback"
    return await oauth.google.authorize_redirect(request, redirect_uri)

@app.get("/api/auth/google/callback")
async def google_callback(request: Request, session: Session = Depends(get_session)):
    token = await oauth.google.authorize_access_token(request)
    user_info = token["userinfo"]
    # 根据 email 找/建用户,签发自己的 JWT
    ...

3.3 中间件(Middleware)

全局中间件,比如给每个请求加 X-Request-ID、记录响应时间:

import uuid

@app.middleware("http")
async def request_id_middleware(request: Request, call_next):
    request_id = request.headers.get("X-Request-ID") or str(uuid.uuid4())
    request.state.request_id = request_id

    start = time.time()
    response = await call_next(request)
    duration_ms = int((time.time() - start) * 1000)

    response.headers["X-Request-ID"] = request_id
    response.headers["X-Response-Time-ms"] = str(duration_ms)

    logger.info("request_completed",
                request_id=request_id,
                path=request.url.path,
                status=response.status_code,
                duration_ms=duration_ms)
    return response

3.4 全局异常处理

from fastapi.exceptions import RequestValidationError

@app.exception_handler(RequestValidationError)
async def validation_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(422, {"detail": exc.errors(), "body": exc.body})

@app.exception_handler(HTTPException)
async def http_handler(request: Request, exc: HTTPException):
    return JSONResponse(exc.status_code, {"detail": exc.detail})

@app.exception_handler(Exception)
async def fallback_handler(request: Request, exc: Exception):
    logger.exception("unhandled_exception", request_id=getattr(request.state, 'request_id', None))
    return JSONResponse(500, {"detail": "Internal Server Error"})

3.5 健康检查接口(给 K8s / LB 用)

@app.get("/health")
def health(session: Session = Depends(get_session)):
    # 实际去 ping 一下 DB / Redis
    try:
        session.exec(select(1))
        redis_client.ping()
        return {"status": "ok"}
    except Exception:
        raise HTTPException(503, "Dependencies unavailable")

K8s 健康探针、LB 后端健康检查都会调这个接口。

3.6 异步任务(Celery)

LLM 长任务(如批量摘要 100 篇文档)不能让 HTTP 请求等。用 Celery 异步:

pip install celery[redis]

from celery import Celery

celery_app = Celery("tasks", broker="redis://localhost:6379/0")

@celery_app.task
def long_running_task(doc_id: str):
    # 批量处理文档...
    return {"status": "done", "doc_id": doc_id}

@app.post("/api/docs/process")
async def process_docs(req):
    task = long_running_task.delay(req.doc_id)
    return {"task_id": task.id}

任务编排、状态查询、重试——见 07. 规模化架构。

4. 常见踩坑

坑 1:JWT 存 localStorage 被 XSS 偷

攻击者注入恶意 JS → 读 localStorage 拿到 token → 转发到自己的服务器

缓解:

严格的 CSP(Content Security Policy)
用 HttpOnly Cookie 存 token(JS 读不到)

但 HttpOnly Cookie 又有 CSRF 问题,要配合 SameSite 属性。没完美方案,看威胁模型。

坑 2:role 写在 token 里,改了不生效

用户是 user,你升级他成 admin——但他手里的 token 里还写着 role: user,一直要等 token 过期。

解决:

Access token TTL 设短(15 分钟)
或者把 role 不放 token 里,每次查 DB(慢但一致)
或者用 token 版本号:用户 role 改了,版本号 +1,旧 token 失效

坑 3:流式响应被代理截断

Nginx、Cloud Run 默认会缓冲响应。需要在响应头加:

return StreamingResponse(
    event_stream(),
    media_type="text/event-stream",
    headers={
        "Cache-Control": "no-cache",
        "X-Accel-Buffering": "no",  # Nginx: 关闭缓冲
    },
)

Cloud Run 也要开 --no-traffic-on-first-byte 之类的配置(具体见平台文档)。

坑 4:限流 key 设计不当

公司内网所有人都共享一个出口 IP → 全公司一起被限流

解决:优先按 user_id 限,IP 作为兜底。

坑 5:密码强度不校验

用户用 "123456" 作密码,bcrypt 也认真 hash 了,但泄露库一秒钟撞出来。

from passlib import pwd

def validate_password(password: str):
    if len(password) < 8:
        raise ValueError("Password too short")
    if not any(c.isupper() for c in password) or not any(c.islower() for c in password):
        raise ValueError("Mix upper and lower case")
    # 可以接 haveibeenpwned API 检查泄露

5. 检查清单

6. 下一步

应用层到这就完整了——你的 LLM 后端已经有认证、权限、缓存、流式、限流,这是 80% 的小型 SaaS 后端该有的东西。

下一站去部署层:

05. 容器化与部署:Docker 打包、Cloud Run 部署、CI/CD 自动上线

让全世界都能通过公网访问你的应用。

进阶学习资源

FastAPI 安全教程:https://fastapi.tiangolo.com/tutorial/security/
JWT.io(在线解码 JWT):https://jwt.io/
OWASP API Security Top 10:https://owasp.org/API-Security/
slowapi 文档:https://slowapi.readthedocs.io/
《OAuth 2 in Action》:Justin Richer
SSE 规范:https://html.spec.whatwg.org/multipage/server-sent-events.html

0. 为什么算法工程师要学这个​

1. 核心概念(最少必要理论)​

1.1 认证 vs 授权(必背)​

1.2 认证方式对比​

1.3 JWT 是什么​

1.4 RBAC(Role-Based Access Control)​

1.5 流式响应(SSE)​

1.6 限流(Rate Limiting)​

2. 实战:把前 3 篇整合成完整服务​

2.1 装包​

2.2 密码 Hash 工具​

2.3 User 模型加 role 字段​

2.4 登录接口​

2.5 认证依赖​

2.6 改 summarize 用认证用户​

2.7 流式输出(SSE)​

2.8 限流​

3. 进阶:生产级改造​

3.1 Refresh Token​

3.2 OAuth 2.0(第三方登录)​

3.3 中间件(Middleware)​

3.4 全局异常处理​

3.5 健康检查接口(给 K8s / LB 用)​

3.6 异步任务(Celery)​

4. 常见踩坑​

坑 1:JWT 存 localStorage 被 XSS 偷​

坑 2:role 写在 token 里,改了不生效​

坑 3:流式响应被代理截断​

坑 4:限流 key 设计不当​

坑 5:密码强度不校验​

5. 检查清单​

6. 下一步​

进阶学习资源​