shaguabijia-app-server/app/core/jiguang.py

"""极光一键登录服务端核验 + RSA 解密手机号。

链路:
  Android SDK loginAuth → loginToken
  → 本服务 POST 极光 /v1/web/loginTokenVerify (Basic Auth = AppKey:MasterSecret)
  → 极光返回 RSA 公钥加密的手机号 (base64)
  → 用配套 RSA 私钥解密
  → 返回明文手机号

参考实现:试验后台 shaguabijia-server/app/api/auth.py。
"""
from __future__ import annotations

import base64
import logging
from pathlib import Path

import httpx
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.hazmat.primitives.asymmetric.rsa import RSAPrivateKey

from app.core.config import settings

logger = logging.getLogger("shagua.jg")


class JiguangError(Exception):
    """极光接口 / 解密失败的统一异常,api 层 catch 后翻成 4xx/5xx。"""


_private_key_cache: RSAPrivateKey | None = None


def _load_private_key() -> RSAPrivateKey:
    """懒加载 RSA 私钥并缓存。首次调用时如果文件不存在抛 JiguangError,
    让单个请求失败 + 日志,而不是 import 时炸进程。"""
    global _private_key_cache
    if _private_key_cache is not None:
        return _private_key_cache

    path = Path(settings.JG_PRIVATE_KEY_PATH)
    if not path.is_file():
        raise JiguangError(
            f"RSA private key not found at {path}. "
            f"Set JG_PRIVATE_KEY_PATH or place the .pem file there."
        )
    key = serialization.load_pem_private_key(path.read_bytes(), password=None)
    if not isinstance(key, RSAPrivateKey):
        raise JiguangError("loaded key is not an RSA private key")
    _private_key_cache = key
    return key


def _verify_login_token(login_token: str) -> str:
    """调极光 REST,返回 RSA 加密后的 base64 手机号字符串。"""
    if not settings.JG_APP_KEY or not settings.JG_MASTER_SECRET:
        raise JiguangError("JG_APP_KEY / JG_MASTER_SECRET not configured")

    auth_b64 = base64.b64encode(
        f"{settings.JG_APP_KEY}:{settings.JG_MASTER_SECRET}".encode()
    ).decode()

    try:
        resp = httpx.post(
            settings.JG_VERIFY_ENDPOINT,
            json={"loginToken": login_token, "exID": ""},
            headers={
                "Content-Type": "application/json",
                "Authorization": f"Basic {auth_b64}",
            },
            timeout=settings.JG_REQUEST_TIMEOUT_SEC,
        )
    except httpx.HTTPError as e:
        logger.exception("[JG] http error calling %s", settings.JG_VERIFY_ENDPOINT)
        raise JiguangError(f"jg http error: {e}") from e

    if resp.status_code != 200:
        body = resp.text[:500]
        logger.error("[JG] HTTP %s body=%s", resp.status_code, body)
        raise JiguangError(f"jg http {resp.status_code}")

    data = resp.json()
    code = data.get("code")
    if code != 8000:
        logger.error("[JG] verify failed code=%s content=%s", code, data.get("content"))
        raise JiguangError(f"jg verify failed code={code}")

    encrypted = data.get("phone") or ""
    if not encrypted:
        logger.error("[JG] response missing phone field: %s", data)
        raise JiguangError("jg response missing phone")
    return encrypted


def _decrypt_phone(encrypted_b64: str) -> str:
    """RSA 解密 base64 密文。

    极光文档没明说 padding。实测 PKCS1v15 在 OAEP 密文上会"假成功"返回垃圾
    (UnicodeDecodeError 0x8e),因此按 OAEP-SHA1 → OAEP-SHA256 → PKCS1v15 顺序
    尝试,谁能解出 11 位纯数字手机号就用谁。

    base64 padding 兼容:JG 各运营商可能返回不带 `=` 的 base64,先按 4 字节对齐补齐。
    """
    private_key = _load_private_key()
    padded_b64 = encrypted_b64 + "=" * (-len(encrypted_b64) % 4)
    ciphertext = base64.b64decode(padded_b64)

    paddings = [
        ("OAEP-SHA1", padding.OAEP(
            mgf=padding.MGF1(algorithm=hashes.SHA1()),
            algorithm=hashes.SHA1(),
            label=None,
        )),
        ("OAEP-SHA256", padding.OAEP(
            mgf=padding.MGF1(algorithm=hashes.SHA256()),
            algorithm=hashes.SHA256(),
            label=None,
        )),
        ("PKCS1v15", padding.PKCS1v15()),
    ]

    last_err: Exception | None = None
    for name, pad in paddings:
        try:
            pt = private_key.decrypt(ciphertext, pad)
        except Exception as e:
            last_err = e
            logger.debug("[JG] decrypt padding=%s failed: %s", name, e)
            continue
        try:
            phone = pt.decode("utf-8")
        except UnicodeDecodeError:
            # PKCS1v15 在错误密文上会"假成功"返回垃圾;跳到下一种 padding
            logger.debug(
                "[JG] decrypt padding=%s yielded non-UTF8 bytes head=%s",
                name, pt[:4].hex(),
            )
            continue
        if phone.isdigit() and len(phone) == 11:
            logger.info("[JG] decrypted with padding=%s", name)
            return phone
        if phone.isascii() and phone.isprintable():
            logger.warning(
                "[JG] decrypted with padding=%s, non-standard format: %r", name, phone,
            )
            return phone
        logger.debug("[JG] decrypt padding=%s yielded non-phone string: %r", name, phone[:20])

    raise JiguangError(f"all RSA paddings failed; last error: {last_err}")


def verify_and_get_phone(login_token: str) -> str:
    """对外暴露的唯一函数:loginToken → 明文手机号。失败抛 JiguangError。"""
    encrypted = _verify_login_token(login_token)
    return _decrypt_phone(encrypted)


def mask_phone(phone: str) -> str:
    """日志用,不打明文。"""
    if len(phone) < 5:
        return "***"
    return phone[:3] + "****" + phone[-2:]