RBAC、Scopes 与 API Key

第 16 章 · RBAC、OAuth2 Scopes 与 API Key

本章目标：在 svc-demo 毕业项目基础上，实现 RBAC（基于角色的访问控制）；使用 FastAPI OAuth2 Scopes 与 SecurityScopes 做细粒度授权；支持 API Key Header（X-API-Key）供机器对机器调用；落地对象级权限（仅可编辑自己创建的资源）；编写 require_permission 依赖与 can_edit_item 服务函数；种子 admin / editor / viewer 三角色；与 Django ch16 / Flask ch16 对照。

学时建议：5～6 小时（含 2 小时权限联调）

前置：本模块 ch07 JWT 与 OAuth2 Password Flow、ch12 svc-demo MVP；ch14 OpenAPI 定制可选（Scopes 会出现在 /docs）。

16.1 为什么需要 RBAC

ch12 毕业项目仅区分「已登录 / 未登录」。团队扩大后需细分能力：

┌─────────┐  M:N  ┌─────────┐  M:N  ┌────────────┐
│  User   │◄────►│  Role   │◄────►│ Permission │
└────┬────┘      └─────────┘      └────────────┘
     │ 1:N
     ▼
┌─────────┐  对象级：created_by_id == user.id
│ Product │
└─────────┘

svc-demo 为教学项目；权限码采用 资源:动作（如 product:write），不映射任何真实业务系统。域名统一使用 api.example.com、registry.example.com。

16.2 数据模型：Role 与 Permission

svc_demo/models/rbac.py（SQLAlchemy 2.0 异步）：

from sqlalchemy import Column, ForeignKey, Integer, String, Table
from sqlalchemy.orm import Mapped, mapped_column, relationship
from svc_demo.db.base import Base

user_roles = Table(
    "user_roles", Base.metadata,
    Column("user_id", ForeignKey("users.id"), primary_key=True),
    Column("role_id", ForeignKey("roles.id"), primary_key=True),
)
role_permissions = Table(
    "role_permissions", Base.metadata,
    Column("role_id", ForeignKey("roles.id"), primary_key=True),
    Column("permission_id", ForeignKey("permissions.id"), primary_key=True),
)

class Permission(Base):
    __tablename__ = "permissions"
    id: Mapped[int] = mapped_column(primary_key=True)
    code: Mapped[str] = mapped_column(String(64), unique=True)
    description: Mapped[str] = mapped_column(String(128), default="")

class Role(Base):
    __tablename__ = "roles"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(32), unique=True)
    label: Mapped[str] = mapped_column(String(64))
    permissions: Mapped[list["Permission"]] = relationship(
        secondary=role_permissions, lazy="selectin"
    )

扩展 User（svc_demo/models/user.py）：

class User(Base):
    roles: Mapped[list["Role"]] = relationship(
        secondary=user_roles, lazy="selectin"
    )

    def has_permission(self, code: str) -> bool:
        return any(p.code == code for r in self.roles for p in r.permissions)

    def has_role(self, name: str) -> bool:
        return any(r.name == name for r in self.roles)

Alembic 迁移后执行 python -m svc_demo.scripts.seed_rbac 写入三角色与权限码。

16.3 OAuth2 Scopes 与 SecurityScopes

FastAPI 原生支持 OAuth2 Scopes：在 OpenAPI 中声明，JWT scope 字段或自定义 Claims 映射到权限码。

svc_demo/core/security.py：

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

oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="/api/v1/auth/token",
    scopes={
        "product:read": "读取商品",
        "product:write": "创建/编辑商品",
        "user:admin": "用户管理",
    },
)

async def get_current_user(
    security_scopes: SecurityScopes,
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db),
) -> User:
    payload = decode_access_token(token)
    user = await user_repo.get_by_id(db, payload["sub"])
    if user is None:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail="无效令牌")

    # JWT scope 为空时回退到 RBAC 权限表
    token_scopes = set(payload.get("scope", "").split())
    for scope in security_scopes.scopes:
        if scope in token_scopes:
            continue
        if not user.has_permission(scope):
            raise HTTPException(
                status.HTTP_403_FORBIDDEN,
                detail=f"缺少权限: {scope}",
                headers={"WWW-Authenticate": security_scopes.scope_str},
            )
    return user

路由声明 Scopes：

@router.get(
    "/products",
    dependencies=[Security(get_current_user, scopes=["product:read"])],
)
async def list_products(...):
    ...

@router.post(
    "/products",
    dependencies=[Security(get_current_user, scopes=["product:write"])],
)
async def create_product(...):
    ...

访问 https://api.example.com/docs 可看到各端点所需 Scopes，便于前端与第三方对接。

16.4 API Key Header：机器对机器

B2B 集成、定时任务、内部微服务常使用 API Key，不走用户密码流。

svc_demo/models/api_key.py：

class ApiKey(Base):
    __tablename__ = "api_keys"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(64))
    key_hash: Mapped[str] = mapped_column(String(128), unique=True)  # 存 SHA-256
    role_id: Mapped[int] = mapped_column(ForeignKey("roles.id"))
    is_active: Mapped[bool] = mapped_column(default=True)
    expires_at: Mapped[datetime | None]

svc_demo/core/api_key.py：

from fastapi import Header, HTTPException, status
from fastapi.security import APIKeyHeader

api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)

async def get_user_or_api_key(
    x_api_key: str | None = Depends(api_key_header),
    token: str | None = Depends(oauth2_scheme_optional),
    db: AsyncSession = Depends(get_db),
) -> User | ApiKeyPrincipal:
    if x_api_key:
        principal = await verify_api_key(db, x_api_key)
        if principal is None:
            raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail="无效 API Key")
        return principal
    if token:
        return await get_current_user_from_token(db, token)
    raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail="需要认证")

Webhook 回调示例：

curl -H "X-API-Key: sk-demo-xxxx" https://api.example.com/api/v1/products