duobibi-app-mock/duobibi(mock)backend/duobibi-server/app/api/auth.py

"""极光一键登录验证 endpoint。

链路:
  Android 客户端 SDK loginAuth → 拿到 loginToken
  → POST 本 endpoint
  → 调极光 REST /v1/web/loginTokenVerify(Basic Auth = AppKey:MasterSecret)
  → 极光返回 RSA 公钥加密的手机号(base64)
  → 用部署在服务器上的对应 RSA 私钥解密
  → 返回明文手机号给客户端

验通路阶段只返回手机号,不入库、不签 session、不做 CPS 归因。
跑通后再补持久化和绑定逻辑。

RSA 私钥位置:`$JG_PRIVATE_KEY_PATH` 环境变量,默认 /opt/duobibi-server/secrets/jverify_rsa_private.pem。
私钥不入 git,部署时单独 scp 到服务器。
"""
from __future__ import annotations

import base64
import json
import logging
import os
import time
import urllib.error
import urllib.request

from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel, Field

from app import db

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

router = APIRouter(prefix="/api/v1/auth")

# 占坑期硬编码,跟 llm_client.py 的 key 风格一致(后续统一挪到 env)
JG_APP_KEY = "966b451a8d9cfe12d173ea9d"
JG_MASTER_SECRET = "dae6d856c7556ac1b8f2deda"
JG_VERIFY_ENDPOINT = "https://api.verification.jpush.cn/v1/web/loginTokenVerify"
JG_REQUEST_TIMEOUT_SEC = 15

JG_PRIVATE_KEY_PATH = os.environ.get(
    "JG_PRIVATE_KEY_PATH",
    "/opt/duobibi-server/secrets/jverify_rsa_private.pem",
)

_private_key_cache = None


def _get_private_key():
    """懒加载 RSA 私钥并缓存。首次调用时如果文件不存在会抛 FileNotFoundError,
    让请求 500 + 日志,而不是 import 时炸进程。"""
    global _private_key_cache
    if _private_key_cache is None:
        with open(JG_PRIVATE_KEY_PATH, "rb") as f:
            _private_key_cache = serialization.load_pem_private_key(f.read(), password=None)
    return _private_key_cache


class JverifyLoginRequest(BaseModel):
    login_token: str = Field(..., description="客户端 loginAuth 拿到的 loginToken")
    operator: str = Field(..., description="CM/CU/CT,客户端透传过来用于日志")


class JverifyLoginResponse(BaseModel):
    phone: str


def _call_jg_login_token_verify(login_token: str) -> str:
    """调极光 REST,返回 RSA 加密后的 base64 手机号字符串。"""
    body = json.dumps({"loginToken": login_token, "exID": ""}).encode("utf-8")
    auth = base64.b64encode(f"{JG_APP_KEY}:{JG_MASTER_SECRET}".encode()).decode()
    req = urllib.request.Request(
        JG_VERIFY_ENDPOINT,
        data=body,
        method="POST",
        headers={
            "Content-Type": "application/json",
            "Authorization": f"Basic {auth}",
        },
    )
    try:
        with urllib.request.urlopen(req, timeout=JG_REQUEST_TIMEOUT_SEC) as resp:
            raw = resp.read()
    except urllib.error.HTTPError as e:
        body_text = ""
        try:
            body_text = e.read().decode("utf-8", errors="replace")
        except Exception:
            pass
        logger.error("[JG] HTTP %s %s body=%s", e.code, e.reason, body_text[:500])
        raise HTTPException(status_code=502, detail=f"jg http error: {e.code}")

    data = json.loads(raw)
    code = data.get("code")
    if code != 8000:
        logger.error("[JG] verify failed code=%s content=%s", code, data.get("content"))
        raise HTTPException(status_code=400, detail=f"jg verify failed code={code}")

    # JG 返回里加密手机号在 "phone" 字段(已验证)
    encrypted = data.get("phone") or ""
    if not encrypted:
        logger.error("[JG] response missing phone field: %s", data)
        raise HTTPException(status_code=500, detail="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 = _get_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:
            # decrypt 在 PKCS1v15 上对错误密文会"假成功"返回垃圾,跳到下一种 padding
            logger.debug("[JG] decrypt padding=%s yielded non-UTF8 bytes head=%s", name, pt[:4].hex())
            continue
        # 验证是不是合法手机号(中国大陆 11 位纯数字 / 或带 + 前缀)
        if phone.isdigit() and len(phone) == 11:
            logger.info("[JG] decrypted with padding=%s", name)
            return phone
        # 不是 11 位纯数字但是 ASCII 可打印,也接受(留个口子,后续看实际格式)
        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 RuntimeError(f"all RSA paddings failed; last error: {last_err}")


@router.post("/jverify-login", response_model=JverifyLoginResponse)
def jverify_login(req: JverifyLoginRequest) -> JverifyLoginResponse:
    logger.info(
        "jverify_login operator=%s token_len=%d",
        req.operator,
        len(req.login_token),
    )
    encrypted = _call_jg_login_token_verify(req.login_token)
    try:
        phone = _decrypt_phone(encrypted)
    except Exception as e:
        logger.error("[JG] decrypt failed: %s", e, exc_info=True)
        raise HTTPException(status_code=500, detail="phone decrypt failed")

    # upsert user(占坑期账号体系:phone 当主键)。失败不阻塞登录响应,只 log
    try:
        db.upsert_user(phone, int(time.time() * 1000))
    except Exception as e:
        logger.error("[JG] upsert_user failed (login still succeeds): %s", e)

    # 日志里只打掩码,明文只返回客户端
    masked = (phone[:3] + "****" + phone[-2:]) if len(phone) >= 5 else "***"
    logger.info("jverify_login ok operator=%s phone=%s", req.operator, masked)
    return JverifyLoginResponse(phone=phone)