shanxia2.0/backend/app/core/login_crypto.py

"""登录凭据 RSA 解密（与前端 JSEncrypt PKCS#1 v1.5 兼容）。"""
from __future__ import annotations

import base64
import json
import logging
import time
from typing import Any, Dict, Optional, Tuple

from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import padding, rsa

from app.core.config import settings

logger = logging.getLogger(__name__)

_private_key = None  # type: ignore
_public_pem: Optional[str] = None


def ensure_initialized() -> None:
    """加载或生成 RSA 密钥对。"""
    global _private_key, _public_pem
    if _private_key is not None:
        return

    pem = (settings.LOGIN_RSA_PRIVATE_KEY_PEM or "").strip()
    # 兼容 .env 中单行保存 PEM（以 "\n" 形式转义换行）的场景
    if "\\n" in pem and "-----BEGIN" in pem:
        pem = pem.replace("\\n", "\n")
    if pem:
        _private_key = serialization.load_pem_private_key(
            pem.encode("utf-8"), password=None, backend=default_backend()
        )
        logger.info("登录 RSA：已自配置加载私钥")
    else:
        _private_key = rsa.generate_private_key(
            public_exponent=65537, key_size=2048, backend=default_backend()
        )
        logger.warning(
            "登录 RSA：未配置 LOGIN_RSA_PRIVATE_KEY_PEM，已生成进程内临时密钥；"
            "多进程/负载均衡下请配置固定 PEM。"
        )

    _public_pem = (
        _private_key.public_key()
        .public_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PublicFormat.SubjectPublicKeyInfo,
        )
        .decode("utf-8")
    )


def get_public_pem() -> str:
    ensure_initialized()
    assert _public_pem is not None
    return _public_pem


def decrypt_login_payload(b64_cipher: str) -> Dict[str, Any]:
    """Base64 密文 -> JSON 对象（须含 username、password、ts 毫秒时间戳）。"""
    ensure_initialized()
    raw = base64.b64decode(b64_cipher)
    decrypted = _private_key.decrypt(raw, padding.PKCS1v15())
    text = decrypted.decode("utf-8")
    return json.loads(text)


def _validate_ts(ts_val: Any) -> bool:
    if ts_val is None:
        return False
    try:
        client_ms = int(ts_val)
    except (TypeError, ValueError):
        return False
    now_ms = int(time.time() * 1000)
    skew_ms = max(30, int(settings.LOGIN_ENCRYPT_TS_SKEW_SEC)) * 1000
    return abs(now_ms - client_ms) <= skew_ms


def extract_login_credentials(body: dict) -> Tuple[Optional[str], Optional[str], Optional[str]]:
    """
    从请求体解析用户名、密码。
    返回 (username, password, error_message)；error_message 非空时表示失败。
    """
    enc = body.get("encryptedLogin")
    if isinstance(enc, str) and enc.strip():
        try:
            plain = decrypt_login_payload(enc.strip())
        except Exception:
            return None, None, "登录凭证无效或已过期"

        if not _validate_ts(plain.get("ts")):
            return None, None, "登录凭证已过期，请重新登录"

        username = (plain.get("username") or "").strip()
        password = plain.get("password")
        if password is None:
            password = ""
        elif not isinstance(password, str):
            password = str(password)

        if not username or not password:
            return None, None, "用户名和密码不能为空"
        return username, password, None

    if settings.LOGIN_ALLOW_PLAINTEXT_LOGIN:
        username = (body.get("username") or "").strip()
        password = body.get("password") or ""
        if not username or not password:
            return None, None, "用户名和密码不能为空"
        return username, password, None

    return None, None, "请使用加密方式登录"