mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-30 20:05:42 +08:00
6d4ce83837
严格检查全部宿主源码，清零类型错误预算并移除按模块类型豁免。 按原因、所属方法、错误码及语法指纹锁定必要兼容例外和站点类型桩， CI 阻断新增忽略、检查范围退化、本地桩替代真实接口及新的类型错误。 修复空值、异步和供应方类型边界，保持插件与 API 兼容合同， 并补充存储、订阅、通知、Agent 和工作流边界的回归覆盖。
118 lines
3.5 KiB
Python
118 lines
3.5 KiB
Python
"""
|
|
提供插件认证方式发现、外部认证跳转和一次性票据兑换接口。
|
|
"""
|
|
|
|
from typing import Any, cast
|
|
|
|
from fastapi import Depends, HTTPException, Request, Response
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app.adapters.web.security.access import set_or_refresh_resource_token_cookie
|
|
from app.api.dependencies.auth import get_auth_service
|
|
from app.api.response import (
|
|
RAW_RESPONSE_OPENAPI_KEY,
|
|
CompatibleCountParam,
|
|
CompatiblePageParam,
|
|
ResponseAPIRouter,
|
|
)
|
|
from app.application.plugin.runtime import get_plugin_manager
|
|
from app.application.security.auth import AuthService, consume_plugin_auth_ticket
|
|
from app.schemas.token import Token as _SchemaToken
|
|
from app.schemas.token import TokenPayload as _SchemaTokenPayload
|
|
from app.schemas.user import AuthProviderInfo as _SchemaAuthProviderInfo
|
|
|
|
router = ResponseAPIRouter()
|
|
|
|
|
|
# 沿用当前 Pydantic 动态基类的类型检查兼容处理。
|
|
class AuthExchangeRequest(BaseModel):
|
|
"""
|
|
插件认证票据兑换请求。
|
|
"""
|
|
|
|
ticket: str = Field(description="插件认证成功后签发的一次性票据，用于兑换宿主登录令牌")
|
|
|
|
|
|
def _system_auth_providers(service: AuthService) -> list[dict[str, Any]]:
|
|
"""
|
|
获取系统内建的匿名登录方式摘要。
|
|
|
|
:return: 系统认证提供方列表
|
|
"""
|
|
has_passkey = service.has_passkey()
|
|
return [
|
|
{
|
|
"id": "system:passkey",
|
|
"type": "system",
|
|
"method": "passkey",
|
|
"name": "通行密钥",
|
|
"icon": "material-symbols:passkey",
|
|
"enabled": has_passkey,
|
|
}
|
|
]
|
|
|
|
|
|
@router.get(
|
|
"/providers",
|
|
summary="查询登录认证提供方",
|
|
response_model=list[_SchemaAuthProviderInfo],
|
|
)
|
|
def auth_providers(service: AuthService = Depends(get_auth_service), page: CompatiblePageParam = None, count: CompatibleCountParam = None) -> list[dict[str, Any]]:
|
|
"""
|
|
查询系统和插件提供的登录认证入口。
|
|
|
|
:return: 认证提供方摘要列表
|
|
"""
|
|
providers = _system_auth_providers(service)
|
|
providers.extend(get_plugin_manager().get_plugin_auth_providers())
|
|
return [provider for provider in providers if provider.get("enabled", True)]
|
|
|
|
|
|
@router.post(
|
|
"/exchange",
|
|
summary="兑换插件认证登录票据",
|
|
response_model=_SchemaToken,
|
|
openapi_extra={RAW_RESPONSE_OPENAPI_KEY: True},
|
|
)
|
|
def auth_exchange(
|
|
request: Request,
|
|
response: Response,
|
|
body: AuthExchangeRequest,
|
|
service: AuthService = Depends(get_auth_service),
|
|
) -> _SchemaToken:
|
|
"""
|
|
将插件认证成功后生成的一次性票据兑换为系统 Token。
|
|
|
|
:param body: 票据兑换请求
|
|
:return: 标准登录 Token
|
|
"""
|
|
ticket_data = consume_plugin_auth_ticket(body.ticket)
|
|
if not ticket_data:
|
|
raise HTTPException(
|
|
status_code=401,
|
|
detail="认证票据无效或已过期",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
user = service.get_user_by_id(cast(int, ticket_data.get("user_id")))
|
|
if not user or not user.is_active:
|
|
raise HTTPException(
|
|
status_code=401,
|
|
detail="用户不存在或已禁用",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
token = service.build_token_response(user)
|
|
set_or_refresh_resource_token_cookie(
|
|
request,
|
|
response,
|
|
_SchemaTokenPayload(
|
|
sub=user.id,
|
|
username=user.name,
|
|
super_user=user.is_superuser,
|
|
level=token.level,
|
|
purpose="authentication",
|
|
),
|
|
)
|
|
return token
|