Archived
snowy-yang
6c3181341b
迁移路由（/api/v1/*，与 Flask 同端点逐字段对齐）： - account/vip：admin/user 双路径 + 5 分钟缓存（降级结果 30s）， VIP 探测单飞锁与 Flask 同源跨栈共享；未登录可访问（同 Flask） - global/playlists、global/playlist：管理员闸（require_admin，403 对齐） - web/deps.py 新增 require_admin；_resolve_user_cookies 支持可选用户（匿名审计保底） 测试：46/46（新增 10 项鉴权依赖：有效/管理员/封禁/禁用/顶号/伪造 Cookie、 401/403 闸）；ruff All checks passed；pyright 0 errors。 契约奇偶校验：16 项全 OK（新增 account:vip、global:playlists/playlist、 非管理员 403）；uvicorn 实跑双栈 account/vip 均 200、未授权歌单 403。
91 lines
2.8 KiB
Python
91 lines
2.8 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""FastAPI 依赖：会话解析、当前用户、登录校验。
|
|
|
|
与 Flask 侧 auth_system 的语义对齐（迁移期两栈共用同一 Cookie、同一站点库）：
|
|
- 会话 Cookie 经 core.session 解码（Flask 兼容签名），密钥同源
|
|
- 封禁账号（status 0/2）：按未登录处理
|
|
- 顶号校验：会话内的 `site_uid_seq` 必须与库中 `login_seq` 一致，否则视为失效
|
|
- 剩余次数/账本预扣属付费流程，待付费路由迁移时再补（此处先覆盖免费端点）
|
|
"""
|
|
from dataclasses import dataclass
|
|
from typing import Annotated, Any
|
|
|
|
from fastapi import Cookie, Depends
|
|
|
|
from core import repo
|
|
from core import session as sess
|
|
from web.errors import AppError
|
|
|
|
_SESSION_COOKIE = 'session'
|
|
|
|
|
|
def secret_key() -> str:
|
|
"""会话签名密钥（与 Flask 同源：data/secret.key）。"""
|
|
import auth_system
|
|
return auth_system.get_secret_key()
|
|
|
|
|
|
@dataclass
|
|
class SiteUser:
|
|
id: int
|
|
username: str
|
|
role: str
|
|
credits: int
|
|
|
|
@property
|
|
def is_admin(self) -> bool:
|
|
return self.role == 'admin'
|
|
|
|
@property
|
|
def unlimited(self) -> bool:
|
|
return self.is_admin or self.credits == -1
|
|
|
|
|
|
def _parse_cookie(session_cookie: str | None) -> dict[str, Any] | None:
|
|
return sess.loads(secret_key(), session_cookie or '')
|
|
|
|
|
|
async def current_user(
|
|
session_cookie: Annotated[str | None, Cookie(alias=_SESSION_COOKIE)] = None,
|
|
) -> SiteUser | None:
|
|
"""当前登录用户（未登录/失效返回 None）。"""
|
|
data = _parse_cookie(session_cookie)
|
|
uid = sess.get_uid(data)
|
|
if not uid:
|
|
return None
|
|
row = await repo.get_user_by_id(uid)
|
|
if row is None:
|
|
return None
|
|
status = row.get('status', 1)
|
|
if status in (0, 2): # 禁用/封禁：按未登录处理
|
|
return None
|
|
# 顶号校验：库有 login_seq 时会话版本必须严格一致
|
|
db_seq = int(row.get('login_seq') or 0)
|
|
sess_seq = int((data or {}).get(sess.SESSION_KEY + '_seq') or 0)
|
|
if db_seq and sess_seq != db_seq:
|
|
return None
|
|
return SiteUser(
|
|
id=int(row['id']),
|
|
username=row['username'],
|
|
role=row['role'],
|
|
credits=int(row.get('duration_seconds') or 0),
|
|
)
|
|
|
|
|
|
async def require_login(
|
|
user: Annotated[SiteUser | None, Depends(current_user)],
|
|
) -> SiteUser:
|
|
"""登录闸：未登录抛 401（与 Flask login_required 同形包体），已登录返回用户。"""
|
|
if user is None:
|
|
raise AppError('请先登录后再使用解析功能', 401, error_code='unauthorized')
|
|
return user
|
|
|
|
|
|
async def require_admin(
|
|
user: Annotated[SiteUser | None, Depends(current_user)],
|
|
) -> SiteUser:
|
|
"""管理员闸：非管理员抛 403（与 Flask admin_required 同形包体）。"""
|
|
if user is None or not user.is_admin:
|
|
raise AppError('需要管理员权限', 403, error_code='forbidden')
|
|
return user
|