MoviePilot/app/startup/composition/security.py

"""
认证、用户与 PassKey 服务的启动组合。
"""

from dataclasses import dataclass
from functools import partial
from typing import cast

from app.adapters.web.security.access import (
    reset_superuser_token_payload_provider,
    reset_token_identity_validator,
    set_superuser_token_payload_provider,
    set_token_identity_validator,
)
from app.application.configuration import get_configured_system_config
from app.application.security.auth import (
    AuthService,
    AuthUserRepository,
    build_superuser_token_payload,
    configure_auth_service,
    reset_auth_service,
    validate_token_identity,
)
from app.application.security.passkey import (
    PASSKEY_CHALLENGE_TTL_SECONDS,
    PasskeyService,
    configure_passkey_challenge_cache,
    configure_passkey_service,
    reset_passkey_challenge_cache,
    reset_passkey_service,
)
from app.application.security.user import configure_user_lookups, reset_user_lookups
from app.application.security.widget import WidgetScopeFactory
from app.db.adapters.user import SqlAlchemyUserRepository
from app.db.adapters.widget import widget_scope
from app.db.oper.passkey import PassKeyOper
from app.db.oper.systemconfig import SystemConfigOper
from app.db.session import SessionFactory
from app.runtime.cache import TTLCache
from app.runtime.log import logger
from app.runtime.settings import get_runtime_setting, update_runtime_setting
from app.startup.composition.context import (
    RepositoryFactory,
    StandaloneRepositoryFactory,
)
from app.startup.composition.database import build_transactional_user_repository


@dataclass(frozen=True, slots=True)
class SecurityComposition:
    """
    保存认证领域运行时投影所需的持久化工厂。

    :param user_repository: 按当前请求会话构造用户仓储的工厂
    :param passkey_repository: 按当前请求会话构造通行密钥仓储的工厂
    :param standalone_user: 为请求外独立操作构造用户仓储的工厂
    :param system_config: 构造独立配置存储访问端口的工厂
    :param passkey: 构造独立通行密钥存储访问端口的工厂
    :param widget_scope: 创建小组件专用授权的独占短事务；为空时不开放
    """

    user_repository: RepositoryFactory
    passkey_repository: RepositoryFactory
    standalone_user: StandaloneRepositoryFactory
    system_config: StandaloneRepositoryFactory
    passkey: StandaloneRepositoryFactory
    widget_scope: WidgetScopeFactory | None = None


def _backfill_superuser_setting(users: AuthUserRepository) -> None:
    """
    用现有数据库管理员补全 V2 升级后缺失的 SUPERUSER。
    """
    if str(get_runtime_setting("SUPERUSER") or "").strip():
        return
    user = users.get_active_superuser()
    if user is None:
        return
    success, message = update_runtime_setting("SUPERUSER", user.name)
    if success is False:
        logger.warning(
            f"检测到数据库超级管理员 {user.name}，但自动补全 SUPERUSER 失败："
            f"{message or '未知错误'}"
        )
        return
    logger.info(f"已根据数据库超级管理员自动补全 SUPERUSER：{user.name}")


def configure_security_services() -> SecurityComposition:
    """
    构造并登记认证、用户查询和 PassKey 服务。
    """
    users = build_transactional_user_repository()
    _backfill_superuser_setting(users)
    configure_user_lookups(
        by_id=lambda user_id: build_transactional_user_repository().get_by_id(user_id),
        by_name=lambda username: build_transactional_user_repository().get_by_name(username),
        by_channel=lambda **bindings: build_transactional_user_repository().find_name_by_bindings(bindings),
    )
    configure_auth_service(
        AuthService(
            users=users,
            config=get_configured_system_config(),
            passkeys=PassKeyOper(),
        )
    )
    configure_passkey_challenge_cache(
        TTLCache(
            region="passkey_challenge",
            maxsize=4096,
            ttl=PASSKEY_CHALLENGE_TTL_SECONDS,
        )
    )
    configure_passkey_service(PasskeyService(repository=PassKeyOper()))
    return SecurityComposition(
        user_repository=SqlAlchemyUserRepository,
        passkey_repository=cast(RepositoryFactory, PassKeyOper),
        standalone_user=build_transactional_user_repository,
        system_config=SystemConfigOper,
        passkey=PassKeyOper,
        widget_scope=partial(widget_scope, SessionFactory),
    )


def configure_security_access() -> None:
    """
    登记 Web 认证边界所需的令牌载荷和当前身份校验端口。
    """
    set_superuser_token_payload_provider(build_superuser_token_payload)
    set_token_identity_validator(validate_token_identity)


def reset_security_services() -> None:
    """
    按发布逆序撤销认证、用户查询和 PassKey 服务。
    """
    reset_passkey_service()
    reset_passkey_challenge_cache()
    reset_auth_service()
    reset_user_lookups()


def reset_security_access() -> None:
    """
    撤销 Web 认证边界的令牌载荷和当前身份校验端口。
    """
    reset_superuser_token_payload_provider()
    reset_token_identity_validator()