Compressor/viewer/backend/auth.py

"""登录鉴权：进程内存 token（与根目录 backend/app/auth.py 同款）。

服务重启后 token 失效，需重新登录。单账号，无数据级权限。
"""

import secrets
import time
from typing import Optional

from fastapi import Header, HTTPException

from config import AUTH_PASSWORD, AUTH_TOKEN_TTL_SECONDS, AUTH_USER

_tokens: dict[str, float] = {}


def create_token() -> str:
    token = secrets.token_hex(32)
    _tokens[token] = time.monotonic() + AUTH_TOKEN_TTL_SECONDS
    return token


def validate_token(token: Optional[str]) -> bool:
    if not token:
        return False
    expires = _tokens.get(token)
    if expires is None:
        return False
    if time.monotonic() > expires:
        _tokens.pop(token, None)
        return False
    return True


def revoke_token(token: Optional[str]) -> None:
    if token:
        _tokens.pop(token, None)


def verify_credentials(username: str, password: str) -> bool:
    return username == AUTH_USER and password == AUTH_PASSWORD


def require_auth(authorization: str | None = Header(default=None)) -> str:
    token = (authorization or "").removeprefix("Bearer ").strip()
    if not validate_token(token):
        raise HTTPException(status_code=401, detail="未登录或登录已过期")
    return token