1
你 → [前台: 有身份证?] → [闸机: 能去5楼?] → 进办公室
2
      ↓ 没有               ↓ 不能
3
    401 Unauthorized     403 Forbidden

1
import bcrypt
2

3
# ========== 注册时 ==========
4
hashed = bcrypt.hashpw(
5
    b"myPassword123",
6
    bcrypt.gensalt(rounds=10)  # rounds=10 故意变慢 ~100ms，增加暴力破解成本
7
)
8
# 存进数据库的：b'$2b$10$N9qo8uLOickgx2ZMRZoMyeIjZIMs3xXcJ3...'
9

10
# ========== 登录时 ==========
11
is_valid = bcrypt.checkpw(b"myPassword123", hashed)
12
# True → 密码正确，False → 密码错误

1
# 注册
2
hashed = bcrypt.hashpw(password.encode(), bcrypt.gensalt(rounds=10))
3

4
# 登录
5
is_ok = bcrypt.checkpw(password.encode(), stored_hash.encode())

1
$2b$10$N9qo8uLOickgx2ZMRZoMyeIjZIMs3xXcJ3xMBXmq...
2
 ─┬─ ─┬─ ───────────┬────────────── ──────────┬──────────
3
  │   │       盐 (22 字符)              哈希值 (31 字符)
4
  │   └─ 成本因子 rounds=10 → 2^10 = 1024 次迭代
5
  └───── 版本号 $2b$

1
import jwt
2
from datetime import datetime, timedelta, timezone
3

4
JWT_SECRET = os.getenv("JWT_SECRET")   # ← 生产环境必须走环境变量！
5
JWT_ALGORITHM = "HS256"
6
JWT_EXPIRES_HOURS = 2
7

8
def create_token(user_id: int, username: str, role: str) -> str:
9
    """签发 JWT 门票"""
10
    now = datetime.now(timezone.utc)
11
    payload = {
12
        "sub": str(user_id),       # subject = 谁的门票
13
        "username": username,      # 购票人
14
        "role": role,              # VIP区还是普通区
15
        "iat": now,                # issued at = 签发时间
16
        "exp": now + timedelta(hours=JWT_EXPIRES_HOURS),  # 2小时后过期
17
    }
18
    return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)
19

20
# 结果长这样：
21
# eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxIiwidXNlcm5hbWUiOiJqb2huIn0.xxxxxx
22
#  ↑ Header              ↑ Payload                    ↑ Signature

1
# 签发
2
payload = {
3
    "sub": str(user_id), "role": role,
4
    "iat": datetime.now(timezone.utc),
5
    "exp": datetime.now(timezone.utc) + timedelta(hours=2)
6
}
7
token = jwt.encode(payload, SECRET, algorithm="HS256")
8

9
# 验证
10
try:
11
    payload = jwt.decode(token, SECRET, algorithms=["HS256"])
12
except jwt.ExpiredSignatureError:
13
    # 门票过期了
14
except jwt.InvalidTokenError:
15
    # 假票！

1
import os
2

3
JWT_SECRET = os.getenv("JWT_SECRET")
4
JWT_ALGORITHM = "HS256"
5
JWT_EXPIRES_HOURS = 2
6

7
# 🔥 fail-fast：没密钥直接崩溃，绝不带病运行
8
if not JWT_SECRET:
9
    raise ValueError("❌ JWT_SECRET 环境变量未设置，服务器拒绝启动！")

1
import os
2
JWT_SECRET = os.getenv("JWT_SECRET")
3
if not JWT_SECRET:
4
    raise ValueError("JWT_SECRET 未设置")

1
from fastapi import Depends, HTTPException
2
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
3
import jwt, hashlib
4

5
security = HTTPBearer()   # 自动从请求头提取 Bearer Token
6

7
def get_current_user(
8
    credentials: HTTPAuthorizationCredentials = Depends(security),
9
    db: Session = Depends(get_db),
10
) -> dict:
11
    token = credentials.credentials
12

13
    # ① 查黑名单（不是先验签，先看有没有挂失）
14
    token_hash = hashlib.sha256(token.encode()).hexdigest()[:64]
15
    if db.query(RevokedToken).filter(RevokedToken.token_hash == token_hash).first():
16
        # HTTPException 是 FastAPI 的异常类——主动抛出一个受控的 HTTP 错误（状态码 + detail）
17
        raise HTTPException(status_code=401, detail="Token 已被撤销，请重新登录")
18

19
    # ② 验签
20
    try:
21
        payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM])
22
    except jwt.ExpiredSignatureError:
23
        raise HTTPException(status_code=401, detail="门票已过期，请重新登录")
24
    except jwt.InvalidTokenError:
25
        raise HTTPException(status_code=401, detail="无效的门票")
26

27
    # ③ 验过了，把身份信息贴在身上
28
    return {
29
        "id": int(payload["sub"]),
30
        "username": payload["username"],
31
        "role": payload["role"],
32
    }
33

34

35
# ========== 使用：一行保护任意路由 ==========
36
@app.get("/vip-room")
37
def enter_vip(user: dict = Depends(get_current_user)):
38
    return {"message": f"欢迎 VIP {user['username']}"}

1
security = HTTPBearer()
2

3
def get_current_user(creds = Depends(security), db = Depends(get_db)):
4
    # ① 查黑名单
5
    token_hash = hashlib.sha256(creds.credentials.encode()).hexdigest()[:64]
6
    if db.query(RevokedToken).filter(RevokedToken.token_hash == token_hash).first():
7
        raise HTTPException(401, "Token 已撤销")
8
    # ② 验签
9
    payload = jwt.decode(creds.credentials, SECRET, algorithms=["HS256"])
10
    return {"id": int(payload["sub"]), "username": payload["username"], "role": payload["role"]}
11

12
# 保护路由
13
@app.get("/xxx")
14
def xxx(user = Depends(get_current_user)):
15
    ...

1
def require_role(*allowed_roles: str):
2
    """
3
    角色守卫工厂 —— 用法:
4
        @app.get("/admin")
5
        def admin(user: dict = require_role("ADMIN")):
6
            ...
7
    """
8
    def checker(user: dict = Depends(get_current_user)):
9
        if user["role"] not in allowed_roles:
10
            raise HTTPException(status_code=403, detail=f"权限不足，需要角色: {allowed_roles}")
11
        return user
12
    return Depends(checker)
13

14

15
# ========== 使用 ==========
16
@app.delete("/articles/{id}")
17
def delete_article(user: dict = require_role("ADMIN", "EDITOR")):
18
    return {"message": f"{user['username']} 删除了文章"}
19

20
@app.get("/admin")
21
def admin_only(user: dict = require_role("ADMIN")):
22
    return {"message": f"欢迎管理员 {user['username']}", "stats": "机密数据"}

1
def require_role(*roles):
2
    def checker(user = Depends(get_current_user)):
3
        if user["role"] not in roles:
4
            raise HTTPException(403, "权限不足")
5
        return user
6
    return Depends(checker)
7

8
# 仅管理员
9
Depends(require_role("ADMIN"))
10
# 多角色
11
Depends(require_role("ADMIN", "EDITOR"))

1
from sqlalchemy.orm import Session
2
from fastapi import APIRouter, Depends, HTTPException
3
from models import User, RevokedToken
4

5
router = APIRouter(prefix="/auth", tags=["认证"])
6

7
# ========== Pydantic 请求体 ==========
8
class RegisterDto(BaseModel):
9
    username: str
10
    password: str
11

12
class LoginDto(BaseModel):
13
    username: str
14
    password: str
15

16
class TokenResponse(BaseModel):
17
    access_token: str
18
    token_type: str = "bearer"
19
    username: str
20
    role: str
21

22
# ========== 注册 ==========
23
@router.post("/register", response_model=TokenResponse)
24
def register(dto: RegisterDto, db: Session = Depends(get_db)):
25
    # ① 查重
26
    if db.query(User).filter(User.username == dto.username).first():
27
        raise HTTPException(status_code=400, detail="用户名已被注册")
28
    # ② 密码绞肉
29
    hashed = bcrypt.hashpw(dto.password.encode(), bcrypt.gensalt(rounds=10))
30
    # ③ 入库
31
    user = User(username=dto.username, password=hashed.decode(), role="USER")
32
    db.add(user); db.commit(); db.refresh(user)
33
    # ④ 发门票
34
    token = create_token(user.id, user.username, user.role)
35
    return TokenResponse(access_token=token, username=user.username, role=user.role)
36

37
# ========== 登录 ==========
38
@router.post("/login", response_model=TokenResponse)
39
def login(dto: LoginDto, db: Session = Depends(get_db)):
40
    # ① 查档案
41
    user = db.query(User).filter(User.username == dto.username).first()
42
    if not user:
43
        raise HTTPException(status_code=401, detail="账号或密码错误")
44
    # ② 对暗号
45
    if not bcrypt.checkpw(dto.password.encode(), user.password.encode()):
46
        raise HTTPException(status_code=401, detail="账号或密码错误")
47
    # ③ 发门票
48
    token = create_token(user.id, user.username, user.role)
49
    return TokenResponse(access_token=token, username=user.username, role=user.role)
50

51
# ========== 登出 ==========
52
@router.post("/logout")
53
def logout(
54
    credentials: HTTPAuthorizationCredentials = Depends(security),
55
    db: Session = Depends(get_db),
56
):
57
    token = credentials.credentials
58
    token_hash = _hash_token(token)   # SHA256前64位，不存完整Token
59

60
    # 解析过期时间放进黑名单，方便定时清理
61
    try:
62
        payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM],
63
                             options={"verify_exp": False})
64
        expires_at = datetime.fromtimestamp(payload["exp"], tz=timezone.utc)
65
    except jwt.InvalidTokenError:
66
        expires_at = datetime.now(timezone.utc) + timedelta(hours=JWT_EXPIRES_HOURS)
67

68
    # 避免重复挂失
69
    if not db.query(RevokedToken).filter(RevokedToken.token_hash == token_hash).first():
70
        revoked = RevokedToken(token_hash=token_hash, expires_at=expires_at)
71
        db.add(revoked); db.commit()
72

73
    return {"message": "已退出登录，Token 已失效"}

1
class User(Base):
2
    """👤 用户（认证系统）"""
3
    __tablename__ = "users"
4
    id = Column(Integer, primary_key=True)
5
    username = Column(String(50), unique=True, nullable=False, index=True)
6
    password = Column(String(200), nullable=False)  # bcrypt 哈希值
7
    role = Column(String(20), default="USER")
8
    created_at = Column(DateTime, default=datetime.now)
9

10

11
class RevokedToken(Base):
12
    """🚫 Token 黑名单（挂失的 Token）"""
13
    __tablename__ = "revoked_tokens"
14
    id = Column(Integer, primary_key=True)
15
    token_hash = Column(String(64), unique=True, index=True)  # SHA256前64位，不存完整Token
16
    revoked_at = Column(DateTime, server_default=func.now())
17
    expires_at = Column(DateTime)  # 自然过期时间，方便定时清理

1
# RegisterDto / LoginDto / TokenResponse
2
class RegisterDto(BaseModel):
3
    username: str
4
    password: str
5

6
class TokenResponse(BaseModel):
7
    access_token: str
8
    token_type: str = "bearer"
9
    username: str
10
    role: str
11

12
# 注册四步
13
user = User(username=dto.username, password=bcrypt.hashpw(dto.password.encode(), bcrypt.gensalt(rounds=10)).decode(), role="USER")
14
db.add(user); db.commit(); db.refresh(user)
15
token = create_token(user.id, user.username, user.role)
16

17
# 登录三步
18
user = db.query(User).filter(User.username == dto.username).first()
19
if not user or not bcrypt.checkpw(dto.password.encode(), user.password.encode()):
20
    raise HTTPException(401, "账号或密码错误")
21
token = create_token(user.id, user.username, user.role)
22

23
# 登出两步
24
token_hash = _hash_token(credentials.credentials)
25
db.add(RevokedToken(token_hash=token_hash, expires_at=...))
26
db.commit()

1
token_hash = _hash_token(credentials.credentials)
2
if db.query(RevokedToken).filter(RevokedToken.token_hash == token_hash).first():
3
    raise HTTPException(status_code=401, detail="Token 已被撤销，请重新登录")

1
前端 ──Access Token──→ API（闸机秒过，无状态）
2
   ←── 401 过期 ──
3
前端 ──Refresh Token──→ /refresh（查档案）→ 新 Access Token

1
# 黑名单验证（嵌入 get_current_user）
2
token_hash = hashlib.sha256(token.encode()).hexdigest()[:64]
3
if db.query(RevokedToken).filter(RevokedToken.token_hash == token_hash).first():
4
    raise HTTPException(401, "Token 已撤销")
5

6
# 登出时加入黑名单
7
token_hash = _hash_token(token)
8
payload = jwt.decode(token, SECRET, algorithms=["HS256"], options={"verify_exp": False})
9
expires_at = datetime.fromtimestamp(payload["exp"], tz=timezone.utc)
10
db.add(RevokedToken(token_hash=token_hash, expires_at=expires_at))

1
app = FastAPI(
2
    title="Study Python API",
3
    description=DESCRIPTION,        # 显示在文档顶部的 Markdown
4
    version="0.1.0",
5
    openapi_tags=TAGS_METADATA,     # 按标签分组路由
6
    docs_url="/docs",               # Swagger UI
7
    redoc_url="/redoc",             # ReDoc 备选
8
)

1
from fastapi.middleware.cors import CORSMiddleware
2

3
app.add_middleware(
4
    CORSMiddleware,
5
    allow_origins=["*"],         # 学习阶段允许所有来源，生产要限制
6
    allow_credentials=True,
7
    allow_methods=["*"],
8
    allow_headers=["*"],
9
)

1
from fastapi import Request
2
from fastapi.responses import JSONResponse
3

4
@app.exception_handler(Exception)
5
async def global_exception_handler(request: Request, exc: Exception):
6
    """兜底异常处理：防止 500 裸奔返回给前端"""
7
    print(f"[异常] {request.method} {request.url.path} → {type(exc).__name__}: {exc}")
8
    return JSONResponse(
9
        status_code=500,
10
        content={"detail": f"服务器内部错误: {type(exc).__name__}"},
11
    )

1
from fastapi.testclient import TestClient
2
from main import app
3

4
client = TestClient(app)
5

6
# 1. 注册 → 200
7
r = client.post("/auth/register", json={"username": "testuser", "password": "test123"})
8
assert r.status_code == 200
9
assert "access_token" in r.json()
10

11
# 2. 登录 → 200
12
r = client.post("/auth/login", json={"username": "testuser", "password": "test123"})
13
token = r.json()["access_token"]
14
assert r.status_code == 200
15

16
# 3. /me → 200
17
r = client.get("/auth/me", headers={"Authorization": f"Bearer {token}"})
18
assert r.status_code == 200
19
assert r.json()["username"] == "testuser"
20

21
# 4. /admin（普通用户）→ 403
22
r = client.get("/auth/admin", headers={"Authorization": f"Bearer {token}"})
23
assert r.status_code == 403
24

25
# 5. 登出 → 200
26
r = client.post("/auth/logout", headers={"Authorization": f"Bearer {token}"})
27
assert r.status_code == 200
28

29
# 6. 登出后 /me → 401
30
r = client.get("/auth/me", headers={"Authorization": f"Bearer {token}"})
31
assert r.status_code == 401
32

33
# 7. 错误密码 → 401
34
r = client.post("/auth/login", json={"username": "testuser", "password": "wrong"})
35
assert r.status_code == 401
36

37
print("✅ 全部测试通过！")

1
# 注册
2
hashed = bcrypt.hashpw(password.encode(), bcrypt.gensalt(rounds=10))
3
# 登录
4
bcrypt.checkpw(password.encode(), stored_hash.encode())  # → True/False

1
payload = {
2
    "sub": str(user_id), "role": role,
3
    "iat": datetime.now(timezone.utc),
4
    "exp": datetime.now(timezone.utc) + timedelta(hours=2)
5
}
6
token = jwt.encode(payload, SECRET, algorithm="HS256")

1
security = HTTPBearer()
2

3
def get_current_user(creds = Depends(security), db = Depends(get_db)):
4
    # ① 黑名单
5
    token_hash = hashlib.sha256(creds.credentials.encode()).hexdigest()[:64]
6
    if db.query(RevokedToken).filter(RevokedToken.token_hash == token_hash).first():
7
        raise HTTPException(401, "Token 已撤销")
8
    # ② 验签
9
    payload = jwt.decode(creds.credentials, SECRET, algorithms=["HS256"])
10
    return {"id": int(payload["sub"]), "username": payload["username"], "role": payload["role"]}
11

12
@app.get("/xxx")
13
def xxx(user = Depends(get_current_user)):
14
    ...

1
def require_role(*roles):
2
    def checker(user = Depends(get_current_user)):
3
        if user["role"] not in roles:
4
            raise HTTPException(403, "权限不足")
5
        return user
6
    return Depends(checker)
7

8
@app.delete("/xxx")
9
def xxx(user = require_role("ADMIN")):
10
    ...

1
Authorization: Bearer <your-jwt>

1
import os
2
JWT_SECRET = os.getenv("JWT_SECRET")
3
if not JWT_SECRET:
4
    raise ValueError("JWT_SECRET 未设置")