MoviePilot/app/application/security/auth.py

"""
认证用户查询、管理员令牌构造与插件一次性登录票据。
"""

import copy
import secrets
import threading
import time
from collections.abc import Mapping
from datetime import timedelta
from typing import Any, Optional, Protocol, cast

from app.application.configuration import get_api_runtime_config_snapshot, get_chain_runtime_config_snapshot
from app.application.security.token import create_access_token
from app.application.security.user import FrozenJson

# 站点目录由资源扩展提供，静态环境可能没有生成模块，保留现有导入检查豁免。
from app.application.site.sites import SitesHelper  # pylint: disable=import-error,no-name-in-module
from app.foundation.singleton import Singleton
from app.schemas.token import Token as _SchemaToken
from app.schemas.token import TokenPayload as _SchemaTokenPayload
from app.schemas.user import UserPermissions


class AuthTicketStore(metaclass=Singleton):
    """
    插件认证一次性票据存储。
    """

    _ttl_seconds = 120
    _max_items = 1024

    def __init__(self) -> None:
        """
        初始化内存票据缓存。
        """
        self._tickets: dict[str, dict[str, Any]] = {}
        self._lock = threading.RLock()

    def create(self, user_id: int, provider_id: str, metadata: Optional[dict[str, Any]] = None) -> str:
        """
        创建短时一次性登录票据。

        :param user_id: 已通过插件认证的本地用户 ID
        :param provider_id: 认证提供方 ID
        :param metadata: 插件侧附加信息
        :return: 一次性票据字符串
        """
        ticket = secrets.token_urlsafe(32)
        now = time.time()
        with self._lock:
            self._tickets[ticket] = {
                "user_id": int(user_id),
                "provider_id": provider_id,
                "metadata": copy.deepcopy(metadata) if metadata is not None else {},
                "created_at": now,
            }
            self._cleanup(now)
        return ticket

    def consume(self, ticket: str) -> Optional[dict[str, Any]]:
        """
        消费并删除一次性登录票据。

        :param ticket: 登录票据
        :return: 票据数据，票据不存在或过期时返回 None
        """
        if not ticket:
            return None
        now = time.time()
        with self._lock:
            data = self._tickets.pop(ticket, None)
            self._cleanup(now)
        if not data:
            return None
        if now - float(data.get("created_at") or 0) > self._ttl_seconds:
            return None
        return copy.deepcopy(data)

    def _cleanup(self, now: Optional[float] = None) -> None:
        """
        清理过期或过量的票据缓存。

        :param now: 当前时间戳，未传入时自动读取
        """
        current = time.time() if now is None else now
        expired = [
            key
            for key, value in self._tickets.items()
            if current - float(value.get("created_at") or 0) > self._ttl_seconds
        ]
        for key in expired:
            self._tickets.pop(key, None)
        if len(self._tickets) <= self._max_items:
            return
        ordered = sorted(
            self._tickets.items(),
            key=lambda item: float(item[1].get("created_at") or 0),
        )
        for key, _ in ordered[: len(self._tickets) - self._max_items]:
            self._tickets.pop(key, None)


def create_plugin_auth_ticket(user_id: int, provider_id: str, metadata: Optional[dict[str, Any]] = None) -> str:
    """
    为插件认证成功的用户创建一次性登录票据。

    :param user_id: 本地用户 ID
    :param provider_id: 认证提供方 ID
    :param metadata: 插件侧附加信息
    :return: 一次性票据字符串
    """
    return AuthTicketStore().create(user_id=user_id, provider_id=provider_id, metadata=metadata)


def consume_plugin_auth_ticket(ticket: str) -> Optional[dict[str, Any]]:
    """
    消费插件认证登录票据。

    :param ticket: 登录票据
    :return: 票据数据，票据不存在或过期时返回 None
    """
    return AuthTicketStore().consume(ticket)


class AuthUser(Protocol):
    """
    认证服务需要的最小用户投影。
    """

    @property
    def id(self) -> int:
        """
        返回用户 ID。

        :return: 用户的数据库主键
        """

    @property
    def name(self) -> str:
        """
        返回用户名。

        :return: 本地用户名
        """

    @property
    def is_active(self) -> bool:
        """
        返回账号启用状态。

        :return: 账号启用时为 True
        """

    @property
    def is_superuser(self) -> bool:
        """
        返回超级用户状态。

        :return: 具备超级管理员身份时为 True
        """

    @property
    def avatar(self) -> Optional[str]:
        """
        返回用户头像。

        :return: 用户头像；未设置时为 None
        """

    @property
    def permissions(self) -> Mapping[str, FrozenJson]:
        """
        返回只读权限快照。

        :return: 可供鉴权读取的权限快照
        """


class AuthUserRepository(Protocol):
    """
    认证服务的用户数据端口。
    """

    def get_by_name(self, name: str) -> Optional[AuthUser]:
        """
        按用户名查询用户。

        :param name: 待查询或更新的本地用户名
        :return: 匹配的用户投影；不存在时返回 None
        """

    def get_by_id(self, user_id: int) -> Optional[AuthUser]:
        """
        按 ID 查询用户。

        :param user_id: 目标用户的数据库主键
        :return: 匹配的用户投影；不存在时返回 None
        """

    def get_active_superuser(self) -> Optional[AuthUser]:
        """
        返回按稳定顺序选出的启用超级管理员。

        :return: 稳定顺序中的首个启用超级管理员；不存在时返回 None
        """


class AuthPasskeyRepository(Protocol):
    """
    认证提供方查询端口。
    """

    def list(self) -> list[Any]:
        """
        返回已启用的 PassKey。

        :return: 已启用的凭证列表；无凭证时为空列表
        """


class AuthConfigRepository(Protocol):
    """
    认证配置读取端口。
    """

    def get(self, key: Any) -> Any:
        """
        读取配置值。

        :param key: 认证服务需要读取的系统配置键
        :return: 配置值；缺失时沿用配置仓库的空值约定
        """


class AuthService:
    """
    认证应用服务，编排用户、配置和 PassKey 端口。
    """

    def __init__(
        self,
        users: AuthUserRepository,
        config: AuthConfigRepository,
        passkeys: AuthPasskeyRepository,
    ) -> None:
        """
        注入认证所需的数据端口。
        """
        self._users = users
        self._config = config
        self._passkeys = passkeys
        self._superuser_binding_name: str | None = None
        self._superuser_binding_id: int | None = None

    def get_user_by_id(self, user_id: int) -> Optional[AuthUser]:
        """
        按 ID 查询本地用户。
        """
        return self._users.get_by_id(user_id)

    def has_passkey(self) -> bool:
        """
        判断系统是否已有 PassKey。
        """
        return bool(self._passkeys.list())

    def build_superuser_token_payload(self) -> _SchemaTokenPayload:
        """
        从持久化用户和站点认证状态构造超级用户令牌载荷。
        """
        configured_name = str(
            get_chain_runtime_config_snapshot().superuser or ""
        ).strip()
        if (
            self._superuser_binding_id is not None
            and configured_name == self._superuser_binding_name
        ):
            # 配置保存用户名；持久化 ID 保证管理员改名不会让管理员级集成失效。
            user = self._users.get_by_id(self._superuser_binding_id)
        else:
            user = (
                self._users.get_by_name(configured_name)
                if configured_name
                else self._users.get_active_superuser()
            )
            if user:
                self._superuser_binding_name = configured_name
                self._superuser_binding_id = user.id
        if not user or not user.is_active or not user.is_superuser:
            if not configured_name:
                raise PermissionError(
                    "未配置 SUPERUSER，且数据库中没有可用超级管理员"
                )
            raise PermissionError(
                "SUPERUSER 对应用户不存在、未启用或非超级管理员"
            )
        return _SchemaTokenPayload(
            sub=user.id,
            username=user.name,
            super_user=user.is_superuser,
            level=SitesHelper().auth_level,
            purpose="authentication",
        )

    def validate_token_identity(self, payload: _SchemaTokenPayload) -> None:
        """
        按当前持久化用户状态校验令牌身份与权限声明。
        """
        if payload.sub is None:
            raise PermissionError("用户不存在或已禁用")
        user = self._users.get_by_id(payload.sub)
        if not user or not user.is_active:
            raise PermissionError("用户不存在或已禁用")
        if payload.username != user.name or payload.super_user != user.is_superuser:
            raise PermissionError("令牌身份或权限上下文不匹配")

    def build_token_response(self, user: AuthUser) -> _SchemaToken:
        """
        使用统一逻辑构造登录 Token 响应。
        """
        level = SitesHelper().auth_level
        config = get_api_runtime_config_snapshot()
        return _SchemaToken(
            access_token=create_access_token(
                userid=user.id,
                username=user.name,
                super_user=user.is_superuser,
                expires_delta=timedelta(minutes=config.access_token_expire_minutes),
                level=level,
            ),
            token_type="bearer",
            super_user=user.is_superuser,
            user_id=user.id,
            user_name=user.name,
            avatar=user.avatar,
            level=level,
            permissions=cast(UserPermissions, dict(user.permissions)),
        )


_configured_auth_service: AuthService | None = None


def configure_auth_service(service: AuthService) -> None:
    """
    由启动组合根登记认证应用服务。
    """
    global _configured_auth_service
    _configured_auth_service = service


def reset_auth_service() -> None:
    """
    清除当前应用生命周期 的认证应用服务。
    """
    global _configured_auth_service
    _configured_auth_service = None


def _get_auth_service() -> AuthService:
    """
    返回启动阶段登记的认证应用服务。
    """
    if _configured_auth_service is None:
        raise RuntimeError("认证服务尚未配置")
    return _configured_auth_service


def get_configured_auth_service() -> AuthService:
    """
    返回启动阶段登记的认证服务。
    """
    return _get_auth_service()


def build_superuser_token_payload() -> _SchemaTokenPayload:
    """
    使用启动组合根注入的认证服务构造超级用户令牌载荷。
    """
    return _get_auth_service().build_superuser_token_payload()


def validate_token_identity(payload: _SchemaTokenPayload) -> None:
    """
    使用启动组合根注入的认证服务校验当前令牌身份。
    """
    _get_auth_service().validate_token_identity(payload)


def build_token_response(user: AuthUser) -> _SchemaToken:
    """
    使用启动组合根注入的认证服务构造登录 Token 响应。
    """
    return _get_auth_service().build_token_response(user)