""" 与传输框架无关的令牌、密码和对称加密能力。 """ import base64 import datetime import hashlib import hmac import importlib import json import os import traceback from datetime import timedelta from typing import Any, Optional, Union import bcrypt import jwt from Crypto.Cipher import AES from Crypto.Util.Padding import pad from cryptography.fernet import Fernet from app.application.configuration import TokenRuntimeConfig, get_token_runtime_config from app.runtime.log import logger from app.schemas.token import TokenPayload BCRYPT_PASSWORD_MAX_BYTES = 72 BCRYPT_ROUNDS = 12 ALGORITHM = "HS256" def _token_config() -> TokenRuntimeConfig: """ 读取启动快照；未装配时保留旧插件的独立调用兼容。 """ try: return get_token_runtime_config() except RuntimeError: legacy_settings = importlib.import_module("app.runtime.config").settings return TokenRuntimeConfig( secret_key=legacy_settings.SECRET_KEY, resource_secret_key=legacy_settings.RESOURCE_SECRET_KEY, access_token_expire_minutes=legacy_settings.ACCESS_TOKEN_EXPIRE_MINUTES, resource_access_token_expire_seconds=( legacy_settings.RESOURCE_ACCESS_TOKEN_EXPIRE_SECONDS ), ) class PasswordTooLongError(ValueError): """ 密码的 UTF-8 字节长度超过 bcrypt 可安全处理的上限。 """ class TokenValidationError(ValueError): """ 令牌缺失、签名无效或用途不符合调用方要求。 """ def _encode_bcrypt_password( password: str, *, allow_legacy_truncation: bool = False, ) -> bytes: """ 编码 bcrypt 密码；仅验证既有哈希时允许按历史语义截断。 """ password_bytes = password.encode("utf-8") if len(password_bytes) > BCRYPT_PASSWORD_MAX_BYTES: if allow_legacy_truncation: return password_bytes[:BCRYPT_PASSWORD_MAX_BYTES] raise PasswordTooLongError( f"密码 UTF-8 编码后不能超过 {BCRYPT_PASSWORD_MAX_BYTES} 字节" ) return password_bytes def create_access_token( userid: Union[str, Any], username: str, super_user: Optional[bool] = False, expires_delta: Optional[timedelta] = None, level: Optional[int] = 1, purpose: Optional[str] = "authentication", agent_channel: Optional[str] = None, agent_channel_user: Optional[str] = None, ) -> str: """ 创建带身份、权限等级和用途声明的 JWT 访问令牌。 宿主 Agent 代渠道用户调用内部 API 时，可把发起会话的渠道及渠道用户标识写入签名声明； 两项需同时提供且令牌用途为认证时才会写入，资源令牌不携带这组声明。 :param userid: 令牌主体的 MoviePilot 用户 ID :param username: 令牌主体的用户名 :param super_user: 是否为超级用户 :param expires_delta: 自定义有效期；None 时按用途使用默认有效期 :param level: 权限级别 :param purpose: 令牌用途，authentication 或 resource :param agent_channel: 宿主 Agent 会话所属的消息渠道 :param agent_channel_user: 发起该会话的用户在渠道内的原始标识 :return: 已签名的 JWT 字符串 :raises ValueError: 自定义有效期不是正数 """ config = _token_config() if purpose == "resource": default_expire = timedelta( seconds=config.resource_access_token_expire_seconds ) secret_key = config.resource_secret_key else: default_expire = timedelta(minutes=config.access_token_expire_minutes) secret_key = config.secret_key if expires_delta is not None: if expires_delta.total_seconds() <= 0: raise ValueError("过期时间必须为正数") expire = datetime.datetime.now(datetime.UTC) + expires_delta else: expire = datetime.datetime.now(datetime.UTC) + default_expire now = datetime.datetime.now(datetime.UTC) payload = { "exp": expire, "iat": now, "sub": str(userid), "username": username, "super_user": super_user, "level": level, "purpose": purpose, } if purpose == "authentication" and agent_channel and agent_channel_user: payload["agent_channel"] = agent_channel payload["agent_channel_user"] = agent_channel_user encoded: str = jwt.encode(payload, secret_key, algorithm=ALGORITHM) return encoded def decode_access_token( token: str | None, purpose: str = "authentication", ) -> TokenPayload: """ 校验 JWT 签名和用途并返回框架无关的令牌载荷。 """ config = _token_config() if not token: raise TokenValidationError(f"{purpose} token not found") secret_key = ( config.resource_secret_key if purpose == "resource" else config.secret_key ) try: payload = jwt.decode(token, secret_key, algorithms=[ALGORITHM]) token_payload = TokenPayload(**payload) if token_payload.purpose != purpose: raise jwt.InvalidTokenError("令牌用途不匹配") return token_payload except ( jwt.DecodeError, jwt.InvalidTokenError, jwt.ImmatureSignatureError, ) as error: raise TokenValidationError("token校验不通过") from error def verify_password(plain_password: str, hashed_password: str) -> bool: """ 验证既有 bcrypt 哈希，并保留超长历史密码的截断语义。 """ try: verified: bool = bcrypt.checkpw( _encode_bcrypt_password( plain_password, allow_legacy_truncation=True, ), hashed_password.encode("ascii"), ) return verified except (UnicodeEncodeError, ValueError): return False def get_password_hash(password: str) -> str: """ 使用 ``$2b$`` 前缀和 cost 12 生成可持久化的 bcrypt 哈希。 """ hashed: bytes = bcrypt.hashpw( _encode_bcrypt_password(password), bcrypt.gensalt(rounds=BCRYPT_ROUNDS, prefix=b"2b"), ) return hashed.decode("ascii") def decrypt(data: bytes, key: bytes) -> Optional[bytes]: """ 使用 Fernet 解密二进制数据，失败时记录诊断并返回空值。 """ try: decrypted: bytes = Fernet(key).decrypt(data) return decrypted except Exception as error: logger.error(f"解密失败：{str(error)} - {traceback.format_exc()}") return None def encrypt_message(message: str, key: bytes) -> str: """ 使用 Fernet 加密文本并返回可传输字符串。 """ encrypted: str = Fernet(key).encrypt(message.encode()).decode() return encrypted def hash_sha256(message: str) -> str: """ 返回文本的 SHA-256 十六进制摘要。 """ return hashlib.sha256(message.encode()).hexdigest() def aes_decrypt(data: str, key: str) -> str: """ 按历史 AES-256-CBC 合同解密 Base64 文本。 """ if not data: return "" raw_data = base64.b64decode(data) iv = raw_data[:16] encrypted = raw_data[16:] cipher = AES.new(key.encode("utf-8"), AES.MODE_CBC, iv) result = cipher.decrypt(encrypted) padding = result[-1] if padding < 1 or padding > AES.block_size: return "" return result[:-padding].decode("utf-8") def aes_encrypt(data: str, key: str) -> str: """ 按历史 AES-256-CBC 合同加密文本并返回 Base64 字符串。 """ if not data: return "" cipher = AES.new(key.encode("utf-8"), AES.MODE_CBC) # CBC 按字节分块；先编码再填充，保留 ASCII 密文并支持多字节文本。 encoded = data.encode("utf-8") result = cipher.encrypt(pad(encoded, AES.block_size)) return base64.b64encode(bytes(cipher.iv) + result).decode("utf-8") def nexusphp_encrypt(data_str: str, key: bytes) -> str: """ 生成 NexusPHP 兼容的 AES-CBC 加密载荷。 """ iv = os.urandom(16) iv_base64 = base64.b64encode(iv) cipher = AES.new(key, AES.MODE_CBC, iv) ciphertext = cipher.encrypt(pad(data_str.encode(), AES.block_size)) ciphertext_base64 = base64.b64encode(ciphertext) mac = hmac.new( key, msg=iv_base64 + ciphertext_base64, digestmod=hashlib.sha256, ).hexdigest() payload = json.dumps({ "iv": iv_base64.decode(), "value": ciphertext_base64.decode(), "mac": mac, "tag": "", }) return base64.b64encode(payload.encode()).decode()