190 lines
5.3 KiB
Python
190 lines
5.3 KiB
Python
"""访问门控 —— 支持管理员令牌 + 动态 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/")
|