将项目文件整理到 refer 目录

2026-07-03 22:26:00 +08:00
parent 49880e785f
commit 9f57af4d37
1223 changed files with 98 additions and 0 deletions
@@ -0,0 +1,189 @@
+"""访问门控 —— 支持管理员令牌 + 动态 UUID 两种凭证。
+
+部署方式(优先级从高到低):
+1. ADMIN_TOKEN: 管理员初始令牌(如 admin7226132)。验证通过后进入管理页,
+   可创建普通 UUID 供合伙人使用,管理员本身也可访问全部功能。
+2. 动态 UUID: 管理员通过 /admin/uuids 创建的访问 UUID,持久化在
+   data/user_data/access_uuids.json。
+3. ACCESS_UUID: 遗留单共享 UUID,保持向后兼容。
+
+以上全部留空则门控不启用。
+"""
+from __future__ import annotations
+
+import hmac
+import logging
+import secrets
+import time
+from enum import Enum
+from typing import Any
+
+from fastapi import HTTPException, Request
+from pydantic import BaseModel
+
+from app.config import settings
+from app import uuid_store
+
+logger = logging.getLogger(__name__)
+
+
+class AuthRole(str, Enum):
+    ADMIN = "admin"
+    USER = "user"
+
+
+# 内存中的令牌缓存:{token: {"role": role, "expires_at": float}}
+_token_cache: dict[str, dict[str, Any]] = {}
+
+# 令牌默认有效期:7 天
+TOKEN_TTL_SECONDS = 7 * 24 * 60 * 60
+
+
+class VerifyIn(BaseModel):
+    credential: str
+
+
+class VerifyOut(BaseModel):
+    valid: bool
+    role: str | None = None
+    token: str | None = None
+
+
+class AuthStatusOut(BaseModel):
+    enabled: bool
+    verified: bool
+    role: str | None = None
+
+
+class UuidRecordOut(BaseModel):
+    uuid: str
+    label: str
+    enabled: bool
+    created_at: int
+
+
+class UuidCreateIn(BaseModel):
+    label: str = ""
+
+
+def access_control_enabled() -> bool:
+    """是否启用了任何访问门控。"""
+    return bool(settings.admin_token) or bool(settings.access_uuid)
+
+
+def admin_mode_enabled() -> bool:
+    """是否启用了管理员令牌模式。"""
+    return bool(settings.admin_token)
+
+
+def _constant_time_compare(a: str, b: str) -> bool:
+    """常量时间字符串比较,降低时序攻击风险。"""
+    return hmac.compare_digest(a.encode(), b.encode())
+
+
+def verify_admin_token(credential: str | None) -> bool:
+    """校验是否为管理员令牌。"""
+    if not credential or not settings.admin_token:
+        return False
+    return _constant_time_compare(credential.strip(), settings.admin_token.strip())
+
+
+def verify_uuid(credential: str | None) -> bool:
+    """校验输入是否为有效访问 UUID(遗留单 UUID 或动态 UUID)。"""
+    if not credential:
+        return False
+    stripped = credential.strip()
+    # 动态 UUID
+    if uuid_store.exists(stripped):
+        return True
+    # 遗留单共享 UUID
+    if settings.access_uuid and _constant_time_compare(stripped, settings.access_uuid.strip()):
+        return True
+    return False
+
+
+def verify_credential(credential: str | None) -> AuthRole | None:
+    """校验任意凭证,返回对应角色;无效返回 None。"""
+    if not credential:
+        return None
+    if verify_admin_token(credential):
+        return AuthRole.ADMIN
+    if verify_uuid(credential):
+        return AuthRole.USER
+    return None
+
+
+def create_access_token(role: AuthRole) -> str:
+    """生成一个随机访问令牌并缓存。"""
+    token = secrets.token_urlsafe(32)
+    _token_cache[token] = {"role": role, "expires_at": time.time() + TOKEN_TTL_SECONDS}
+    return token
+
+
+def validate_access_token(token: str | None) -> AuthRole | None:
+    """校验令牌是否有效,返回角色。"""
+    if not token:
+        return None
+    cached = _token_cache.get(token)
+    if cached is None:
+        return None
+    expires_at = cached.get("expires_at", 0)
+    if expires_at > 0 and time.time() > expires_at:
+        _token_cache.pop(token, None)
+        return None
+    return cached.get("role")
+
+
+def revoke_access_token(token: str | None) -> None:
+    """使指定令牌失效。"""
+    if token:
+        _token_cache.pop(token, None)
+
+
+def get_access_token_from_request(request: Request) -> str | None:
+    """从请求头或 query 参数中提取访问令牌。"""
+    header = request.headers.get("X-Access-Token")
+    if header:
+        return header.strip()
+    return request.query_params.get("access_token")
+
+
+def require_access(request: Request, allowed_roles: set[AuthRole] | None = None) -> AuthRole:
+    """FastAPI 依赖:未通过校验时抛出 401。
+
+    allowed_roles: 仅允许指定角色访问;None 表示 admin/user 均可。
+    """
+    if not access_control_enabled():
+        return AuthRole.ADMIN  # 未启用门控时视为最高权限
+    token = get_access_token_from_request(request)
+    role = validate_access_token(token)
+    if role is None:
+        raise HTTPException(status_code=401, detail="访问令牌无效或已过期")
+    if allowed_roles is not None and role not in allowed_roles:
+        raise HTTPException(status_code=403, detail="权限不足")
+    return role
+
+
+def require_admin(request: Request) -> AuthRole:
+    """FastAPI 依赖:仅管理员可访问。"""
+    return require_access(request, allowed_roles={AuthRole.ADMIN})
+
+
+def is_public_path(path: str) -> bool:
+    """判断请求路径是否属于白名单(无需校验)。"""
+    public_prefixes = (
+        "/health",
+        "/api/auth/",
+        "/assets/",
+        "/index.html",
+        "/favicon.ico",
+        "/robots.txt",
+        "/manifest.json",
+    )
+    lowered = path.lower()
+    return lowered.startswith(public_prefixes) or lowered == "/"
+
+
+def is_admin_path(path: str) -> bool:
+    """判断是否为管理员接口路径。"""
+    return path.lower().startswith("/api/admin/")