限制扫码登录场景值长度

.env.example

@@ -15,3 +15,6 @@ WECHAT_APP_SECRET=
 PUBLIC_BASE_URL=http://127.0.0.1:8000
 UPLOAD_DIR=uploads
 
+ERP_API_BASE_URL=http://127.0.0.1:8000
+ERP_QR_LOGIN_SHARED_SECRET=replace-with-same-secret-on-erp-backend
+ERP_LOGIN_QRCODE_DIR=erp-login-qrcodes

app/config.py

@@ -27,6 +27,9 @@ class Settings(BaseSettings):
     wechat_qr_env_version: str = "release"
     public_base_url: str = "http://127.0.0.1:8000"
     upload_dir: str = "uploads"
+    erp_api_base_url: str = "http://127.0.0.1:8000"
+    erp_qr_login_shared_secret: str = ""
+    erp_login_qrcode_dir: str = "erp-login-qrcodes"
 
     @property
     def database_url(self) -> str:

app/main.py

@@ -16,6 +16,7 @@ from app.routers import (
     dashboard,
     devices,
     equipment,
+    erp_login,
     monitor,
     notices,
     people,
@@ -92,6 +93,7 @@ def create_app() -> FastAPI:
     app.include_router(dashboard.router)
     app.include_router(devices.router)
     app.include_router(equipment.router)
+    app.include_router(erp_login.router)
     app.include_router(monitor.router)
     app.include_router(notices.router)
     app.include_router(reconciliation.router)

app/routers/erp_login.py

@@ -0,0 +1,199 @@
+import json
+from typing import Any
+from urllib.parse import urlsplit
+from uuid import uuid4
+
+import httpx
+from fastapi import APIRouter, Header, HTTPException, Query, Request
+
+from app.config import settings
+from app.schemas import (
+    ErpLoginActionResponse,
+    ErpLoginConfirmRequest,
+    ErpLoginPreviewResponse,
+    ErpLoginQrcodeRequest,
+    ErpLoginQrcodeResponse,
+)
+from app.services.erp_login_security import (
+    SERVICE_NONCE_HEADER,
+    SERVICE_SIGNATURE_HEADER,
+    SERVICE_TIMESTAMP_HEADER,
+    signed_headers,
+    verify_service_signature,
+)
+from app.services.wechat import WechatConfigError, create_miniapp_qrcode, get_phone_number
+from app.timezone import now
+
+
+router = APIRouter(prefix="/api/erp-login", tags=["erp-login"])
+
+ERP_LOGIN_PAGE = "pages/erpLoginConfirm/erpLoginConfirm"
+WECHAT_SCENE_MAX_LENGTH = 32
+
+
+def _json_body(payload: dict[str, Any]) -> bytes:
+    return json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
+
+
+def _erp_url(path: str) -> str:
+    return f"{settings.erp_api_base_url.rstrip('/')}{path}"
+
+
+def _require_session_id(session_id: int | None) -> int:
+    if session_id is None:
+        raise HTTPException(status_code=400, detail="缺少ERP扫码登录会话ID")
+    return session_id
+
+
+def _scene(ticket: str, session_id: int) -> str:
+    scene = f"t={ticket}&s={session_id}"
+    if len(scene) > WECHAT_SCENE_MAX_LENGTH:
+        raise HTTPException(status_code=422, detail="ERP扫码登录二维码参数过长")
+    return scene
+
+
+def _action_response(data: dict[str, Any]) -> ErpLoginActionResponse:
+    return ErpLoginActionResponse(
+        status=str(data.get("status") or ""),
+        failure_reason=data.get("failure_reason"),
+    )
+
+
+async def _read_erp_json(response: httpx.Response) -> dict[str, Any]:
+    try:
+        response.raise_for_status()
+    except httpx.HTTPStatusError as exc:
+        raise HTTPException(status_code=502, detail="ERP扫码登录服务暂时不可用") from exc
+    try:
+        data = response.json()
+    except ValueError as exc:
+        raise HTTPException(status_code=502, detail="ERP扫码登录响应无效") from exc
+    if not isinstance(data, dict):
+        raise HTTPException(status_code=502, detail="ERP扫码登录响应无效")
+    return data
+
+
+async def _get_erp_json(path: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
+    url = _erp_url(path)
+    body = b""
+    headers = signed_headers(
+        "GET",
+        urlsplit(url).path,
+        body,
+        settings.erp_qr_login_shared_secret,
+    )
+    try:
+        async with httpx.AsyncClient(timeout=15) as client:
+            response = await client.get(url, params=params, headers=headers)
+    except httpx.RequestError as exc:
+        raise HTTPException(status_code=502, detail="ERP扫码登录服务暂时不可用") from exc
+    return await _read_erp_json(response)
+
+
+async def _post_erp_json(path: str, payload: dict[str, Any]) -> dict[str, Any]:
+    url = _erp_url(path)
+    body = _json_body(payload)
+    headers = signed_headers(
+        "POST",
+        urlsplit(url).path,
+        body,
+        settings.erp_qr_login_shared_secret,
+    )
+    headers["content-type"] = "application/json"
+    try:
+        async with httpx.AsyncClient(timeout=15) as client:
+            response = await client.post(url, content=body, headers=headers)
+    except httpx.RequestError as exc:
+        raise HTTPException(status_code=502, detail="ERP扫码登录服务暂时不可用") from exc
+    return await _read_erp_json(response)
+
+
+@router.post("/qrcode", response_model=ErpLoginQrcodeResponse)
+async def create_erp_login_qrcode(
+    payload: ErpLoginQrcodeRequest,
+    request: Request,
+    timestamp: str | None = Header(default=None, alias=SERVICE_TIMESTAMP_HEADER),
+    nonce: str | None = Header(default=None, alias=SERVICE_NONCE_HEADER),
+    signature: str | None = Header(default=None, alias=SERVICE_SIGNATURE_HEADER),
+) -> ErpLoginQrcodeResponse:
+    body = await request.body()
+    verify_service_signature(
+        "POST",
+        request.url.path,
+        timestamp,
+        nonce,
+        body,
+        signature,
+        settings.erp_qr_login_shared_secret,
+    )
+    scene = _scene(payload.ticket, payload.session_id)
+    try:
+        qr_url = await create_miniapp_qrcode(
+            key=f"erp-login-{payload.session_id}",
+            page=ERP_LOGIN_PAGE,
+            scene=scene,
+            public_base_url=settings.public_base_url,
+            directory=settings.erp_login_qrcode_dir,
+            label="ERP扫码登录",
+        )
+    except WechatConfigError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+    except RuntimeError as exc:
+        raise HTTPException(status_code=502, detail=str(exc)) from exc
+    return ErpLoginQrcodeResponse(qr_url=qr_url, page=ERP_LOGIN_PAGE, scene=scene)
+
+
+@router.get("/sessions/{ticket}", response_model=ErpLoginPreviewResponse)
+async def preview_erp_login_session(
+    ticket: str,
+    session_id: int | None = Query(default=None),
+) -> ErpLoginPreviewResponse:
+    session_id = _require_session_id(session_id)
+    data = await _get_erp_json(
+        f"/api/auth/qr-login/sessions/{session_id}/preview",
+        params={"ticket": ticket},
+    )
+    await _post_erp_json(
+        f"/api/auth/qr-login/sessions/{session_id}/scanned",
+        {"ticket": ticket},
+    )
+    return ErpLoginPreviewResponse(**data)
+
+
+@router.post("/sessions/{ticket}/confirm", response_model=ErpLoginActionResponse)
+async def confirm_erp_login_session(
+    ticket: str,
+    payload: ErpLoginConfirmRequest,
+    session_id: int | None = Query(default=None),
+) -> ErpLoginActionResponse:
+    session_id = _require_session_id(session_id)
+    try:
+        phone = await get_phone_number(payload.phone_code)
+    except WechatConfigError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+    except RuntimeError as exc:
+        raise HTTPException(status_code=502, detail=str(exc)) from exc
+
+    data = await _post_erp_json(
+        f"/api/auth/qr-login/sessions/{session_id}/confirm",
+        {
+            "ticket": ticket,
+            "phone": phone,
+            "confirmed_at": now().isoformat(),
+            "nonce": uuid4().hex,
+        },
+    )
+    return _action_response(data)
+
+
+@router.post("/sessions/{ticket}/cancel", response_model=ErpLoginActionResponse)
+async def cancel_erp_login_session(
+    ticket: str,
+    session_id: int | None = Query(default=None),
+) -> ErpLoginActionResponse:
+    session_id = _require_session_id(session_id)
+    data = await _post_erp_json(
+        f"/api/auth/qr-login/sessions/{session_id}/cancel",
+        {"ticket": ticket},
+    )
+    return _action_response(data)

app/schemas.py

@@ -778,6 +778,35 @@ class ReconciliationEntryUpdate(BaseModel):
     return_qty: float | None = Field(default=None, ge=0)
 
 
+class ErpLoginQrcodeRequest(BaseModel):
+    ticket: str = Field(min_length=1, max_length=16)
+    session_id: int = Field(ge=1)
+
+
+class ErpLoginQrcodeResponse(BaseModel):
+    qr_url: str | None = None
+    page: str
+    scene: str
+
+
+class ErpLoginPreviewResponse(BaseModel):
+    session_id: int
+    system_name: str
+    started_at: datetime
+    expires_at: datetime
+    request_ip_hint: str | None = None
+    user_agent_hint: str | None = None
+
+
+class ErpLoginConfirmRequest(BaseModel):
+    phone_code: str = Field(min_length=1, max_length=256)
+
+
+class ErpLoginActionResponse(BaseModel):
+    status: str
+    failure_reason: str | None = None
+
+
 PersonnelOut.model_rebuild()
 TokenResponse.model_rebuild()
 LoginSelectionResponse.model_rebuild()

app/services/erp_login_security.py

@@ -0,0 +1,99 @@
+import hashlib
+import hmac
+import time
+from uuid import uuid4
+
+from fastapi import HTTPException
+
+
+SERVICE_TIMESTAMP_HEADER = "X-ERP-QR-Timestamp"
+SERVICE_NONCE_HEADER = "X-ERP-QR-Nonce"
+SERVICE_SIGNATURE_HEADER = "X-ERP-QR-Signature"
+
+_seen_nonces: dict[str, int] = {}
+
+
+def build_service_signature(
+    method: str,
+    path: str,
+    timestamp: str,
+    nonce: str,
+    body: bytes,
+    secret: str,
+) -> str:
+    message = "\n".join([
+        method.upper(),
+        path,
+        timestamp,
+        nonce,
+        body.decode("utf-8"),
+    ])
+    return hmac.new(secret.encode("utf-8"), message.encode("utf-8"), hashlib.sha256).hexdigest()
+
+
+def signed_headers(method: str, path: str, body: bytes, secret: str) -> dict[str, str]:
+    timestamp = str(int(time.time()))
+    nonce = uuid4().hex
+    signature = build_service_signature(method, path, timestamp, nonce, body, secret)
+    return {
+        SERVICE_TIMESTAMP_HEADER: timestamp,
+        SERVICE_NONCE_HEADER: nonce,
+        SERVICE_SIGNATURE_HEADER: signature,
+    }
+
+
+def _cleanup_seen_nonces(now_ts: int, ttl_seconds: int) -> None:
+    expired = [
+        nonce
+        for nonce, seen_at in _seen_nonces.items()
+        if now_ts - seen_at > ttl_seconds
+    ]
+    for nonce in expired:
+        _seen_nonces.pop(nonce, None)
+
+
+def verify_service_signature(
+    method: str,
+    path: str,
+    timestamp: str | None,
+    nonce: str | None,
+    body: bytes,
+    signature: str | None,
+    secret: str,
+    ttl_seconds: int = 300,
+) -> None:
+    if not secret:
+        raise HTTPException(status_code=500, detail="ERP扫码登录服务密钥未配置")
+
+    try:
+        timestamp_value = int(str(timestamp))
+    except (TypeError, ValueError) as exc:
+        raise HTTPException(status_code=401, detail="ERP服务签名无效") from exc
+
+    now_ts = int(time.time())
+    if abs(now_ts - timestamp_value) > ttl_seconds:
+        raise HTTPException(status_code=401, detail="ERP服务签名已过期")
+
+    _cleanup_seen_nonces(now_ts, ttl_seconds)
+    nonce_value = str(nonce or "")
+    if not nonce_value or nonce_value in _seen_nonces:
+        raise HTTPException(status_code=401, detail="ERP服务签名无效")
+
+    try:
+        expected = build_service_signature(
+            method,
+            path,
+            str(timestamp),
+            nonce_value,
+            body,
+            secret,
+        )
+    except UnicodeDecodeError as exc:
+        raise HTTPException(status_code=401, detail="ERP服务签名无效") from exc
+
+    if not hmac.compare_digest(expected, str(signature or "")):
+        raise HTTPException(status_code=401, detail="ERP服务签名无效")
+
+    # This in-process nonce store only protects one running process. Production
+    # multi-worker deployments should move replay tracking to shared storage.
+    _seen_nonces[nonce_value] = now_ts

app/services/wechat.py

@@ -190,6 +190,24 @@ async def create_device_qrcode(
     scene: str,
     public_base_url: str,
     label: str | None = None,
+) -> str | None:
+    return await create_miniapp_qrcode(
+        key=device_no,
+        page=page,
+        scene=scene,
+        public_base_url=public_base_url,
+        directory="qrcodes",
+        label=label,
+    )
+
+
+async def create_miniapp_qrcode(
+    key: str,
+    page: str,
+    scene: str,
+    public_base_url: str,
+    directory: str,
+    label: str | None = None,
 ) -> str | None:
     token = await get_access_token()
     url = f"https://api.weixin.qq.com/wxa/getwxacodeunlimit?access_token={token}"
@@ -205,12 +223,15 @@ async def create_device_qrcode(
     content_type = response.headers.get("content-type", "")
     if "application/json" in content_type:
         data = response.json()
-        raise RuntimeError(data.get("errmsg") or "生成设备小程序码失败")
+        raise RuntimeError(data.get("errmsg") or "生成小程序码失败")
 
-    qr_dir = settings.upload_path / "qrcodes"
+    qr_dir = settings.upload_path / directory
     qr_dir.mkdir(parents=True, exist_ok=True)
     suffix = _qrcode_file_suffix(content_type)
-    file_name = _qrcode_file_name(device_no, suffix)
+    file_name = _qrcode_file_name(key, suffix)
     file_path = Path(qr_dir) / file_name
-    file_path.write_bytes(_draw_qrcode_label(response.content, suffix, label or device_no))
-    return f"{public_base_url.rstrip('/')}/uploads/qrcodes/{quote(file_name, safe='')}"
+    label_text = label if label is not None else key
+    file_path.write_bytes(_draw_qrcode_label(response.content, suffix, label_text))
+    encoded_directory = quote(directory.strip("/"), safe="")
+    encoded_file_name = quote(file_name, safe="")
+    return f"{public_base_url.rstrip('/')}/uploads/{encoded_directory}/{encoded_file_name}"

tests/test_erp_login.py

@@ -0,0 +1,392 @@
+import asyncio
+import hmac
+import json
+import time
+from hashlib import sha256
+from uuid import uuid4
+
+import httpx
+from fastapi.testclient import TestClient
+
+from app.config import settings
+from app.main import create_app
+from app.services import wechat
+
+
+SERVICE_SECRET = "test-erp-secret"
+TIMESTAMP_HEADER = "X-ERP-QR-Timestamp"
+NONCE_HEADER = "X-ERP-QR-Nonce"
+SIGNATURE_HEADER = "X-ERP-QR-Signature"
+
+
+def _json_bytes(payload: dict) -> bytes:
+    return json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
+
+
+def _test_signature(
+    method: str,
+    path: str,
+    timestamp: str,
+    nonce: str,
+    body: bytes,
+    secret: str = SERVICE_SECRET,
+) -> str:
+    message = "\n".join([
+        method.upper(),
+        path,
+        timestamp,
+        nonce,
+        body.decode("utf-8"),
+    ])
+    return hmac.new(secret.encode("utf-8"), message.encode("utf-8"), sha256).hexdigest()
+
+
+def _signed_headers(
+    method: str,
+    path: str,
+    body: bytes,
+    secret: str = SERVICE_SECRET,
+) -> dict[str, str]:
+    timestamp = str(int(time.time()))
+    nonce = uuid4().hex
+    return {
+        TIMESTAMP_HEADER: timestamp,
+        NONCE_HEADER: nonce,
+        SIGNATURE_HEADER: _test_signature(method, path, timestamp, nonce, body, secret),
+        "content-type": "application/json",
+    }
+
+
+def _client(monkeypatch, erp_base_url: str = "http://erp.local") -> TestClient:
+    monkeypatch.setattr(settings, "erp_qr_login_shared_secret", SERVICE_SECRET)
+    monkeypatch.setattr(settings, "erp_api_base_url", erp_base_url)
+    monkeypatch.setattr(settings, "erp_login_qrcode_dir", "erp-login-qrcodes")
+    monkeypatch.setattr(settings, "public_base_url", "https://miniapp.example.com")
+    return TestClient(create_app())
+
+
+def test_erp_login_router_registered(monkeypatch):
+    client = _client(monkeypatch)
+
+    response = client.get("/api/erp-login/sessions/ticket-abc")
+
+    assert response.status_code == 400
+    assert response.json()["detail"] == "缺少ERP扫码登录会话ID"
+
+
+def test_qrcode_rejects_missing_or_bad_service_signature(monkeypatch):
+    client = _client(monkeypatch)
+    body = _json_bytes({"ticket": "ticket-abc", "session_id": 123})
+
+    missing = client.post(
+        "/api/erp-login/qrcode",
+        content=body,
+        headers={"content-type": "application/json"},
+    )
+    bad_headers = _signed_headers("POST", "/api/erp-login/qrcode", body)
+    bad_headers[SIGNATURE_HEADER] = "bad-signature"
+    bad = client.post("/api/erp-login/qrcode", content=body, headers=bad_headers)
+
+    assert missing.status_code == 401
+    assert bad.status_code == 401
+
+
+def test_qrcode_signed_request_uses_expected_miniapp_qrcode_contract(monkeypatch):
+    from app.routers import erp_login
+
+    calls = []
+
+    async def fake_create_miniapp_qrcode(**kwargs):
+        calls.append(kwargs)
+        return "https://miniapp.example.com/uploads/erp-login-qrcodes/erp-login-123.png"
+
+    monkeypatch.setattr(erp_login, "create_miniapp_qrcode", fake_create_miniapp_qrcode)
+    client = _client(monkeypatch)
+    body = _json_bytes({"ticket": "ticket-abc", "session_id": 123})
+
+    response = client.post(
+        "/api/erp-login/qrcode",
+        content=body,
+        headers=_signed_headers("POST", "/api/erp-login/qrcode", body),
+    )
+
+    assert response.status_code == 200
+    assert response.json() == {
+        "qr_url": "https://miniapp.example.com/uploads/erp-login-qrcodes/erp-login-123.png",
+        "page": "pages/erpLoginConfirm/erpLoginConfirm",
+        "scene": "t=ticket-abc&s=123",
+    }
+    assert "ticket-abc" not in calls[0]["key"]
+    assert calls == [{
+        "key": "erp-login-123",
+        "page": "pages/erpLoginConfirm/erpLoginConfirm",
+        "scene": "t=ticket-abc&s=123",
+        "public_base_url": "https://miniapp.example.com",
+        "directory": "erp-login-qrcodes",
+        "label": "ERP扫码登录",
+    }]
+    assert len(response.json()["scene"]) <= 32
+
+
+def test_qrcode_requires_session_id(monkeypatch):
+    from app.routers import erp_login
+
+    calls = []
+
+    async def fake_create_miniapp_qrcode(**kwargs):
+        calls.append(kwargs)
+        return "https://miniapp.example.com/uploads/erp-login-qrcodes/erp-login-ticket-abc.png"
+
+    monkeypatch.setattr(erp_login, "create_miniapp_qrcode", fake_create_miniapp_qrcode)
+    client = _client(monkeypatch)
+    body = _json_bytes({"ticket": "ticket-abc"})
+
+    response = client.post(
+        "/api/erp-login/qrcode",
+        content=body,
+        headers=_signed_headers("POST", "/api/erp-login/qrcode", body),
+    )
+
+    assert response.status_code == 422
+    assert calls == []
+
+
+def test_qrcode_rejects_scene_values_over_wechat_limit(monkeypatch):
+    from app.routers import erp_login
+
+    calls = []
+
+    async def fake_create_miniapp_qrcode(**kwargs):
+        calls.append(kwargs)
+        return "https://miniapp.example.com/uploads/erp-login-qrcodes/erp-login-long.png"
+
+    monkeypatch.setattr(erp_login, "create_miniapp_qrcode", fake_create_miniapp_qrcode)
+    client = _client(monkeypatch)
+    body = _json_bytes({"ticket": "x" * 17, "session_id": 123456789})
+
+    response = client.post(
+        "/api/erp-login/qrcode",
+        content=body,
+        headers=_signed_headers("POST", "/api/erp-login/qrcode", body),
+    )
+
+    assert response.status_code == 422
+    assert calls == []
+
+
+def test_preview_requires_session_id(monkeypatch):
+    client = _client(monkeypatch)
+
+    response = client.get("/api/erp-login/sessions/ticket-abc")
+
+    assert response.status_code == 400
+    assert response.json()["detail"] == "缺少ERP扫码登录会话ID"
+
+
+def test_preview_marks_erp_session_scanned(monkeypatch):
+    from app.routers import erp_login
+
+    original_async_client = httpx.AsyncClient
+    calls = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        calls.append((request.method, request.url.path, request.content))
+        if request.method == "GET":
+            assert request.url.path == "/api/auth/qr-login/sessions/123/preview"
+            assert request.url.params["ticket"] == "ticket-abc"
+            return httpx.Response(
+                200,
+                json={
+                    "session_id": 123,
+                    "system_name": "嘉恒智能五金 ERP",
+                    "started_at": "2026-07-05T08:00:00",
+                    "expires_at": "2026-07-05T08:02:00",
+                    "request_ip_hint": "127.0.0.1",
+                    "user_agent_hint": "Chrome",
+                },
+            )
+        if request.method == "POST":
+            assert request.url.path == "/api/auth/qr-login/sessions/123/scanned"
+            assert json.loads(request.content.decode("utf-8")) == {"ticket": "ticket-abc"}
+            return httpx.Response(200, json={"status": "SCANNED"})
+        raise AssertionError(f"Unexpected request: {request.method} {request.url.path}")
+
+    transport = httpx.MockTransport(handler)
+
+    def fake_async_client(**kwargs):
+        _ = kwargs
+        return original_async_client(transport=transport)
+
+    monkeypatch.setattr(erp_login.httpx, "AsyncClient", fake_async_client)
+    client = _client(monkeypatch)
+
+    response = client.get("/api/erp-login/sessions/ticket-abc?session_id=123")
+
+    assert response.status_code == 200
+    assert [call[0] for call in calls] == ["GET", "POST"]
+
+
+def test_confirm_gets_phone_signs_erp_request_with_base_path_and_hides_auth_session(monkeypatch):
+    from app.routers import erp_login
+
+    async def fake_get_phone_number(phone_code: str) -> str:
+        assert phone_code == "wx-phone-code"
+        return "13800000000"
+
+    captured: dict[str, object] = {}
+    original_async_client = httpx.AsyncClient
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        captured["url"] = str(request.url)
+        captured["body"] = request.content
+        captured["headers"] = request.headers
+        assert request.url.path == "/erp-prefix/api/auth/qr-login/sessions/123/confirm"
+        raw_header_names = {name.decode("ascii") for name, _ in request.headers.raw}
+        assert TIMESTAMP_HEADER in raw_header_names
+        assert NONCE_HEADER in raw_header_names
+        assert SIGNATURE_HEADER in raw_header_names
+        assert "x-service-timestamp" not in request.headers
+        timestamp = request.headers[TIMESTAMP_HEADER]
+        nonce = request.headers[NONCE_HEADER]
+        expected = _test_signature(
+            "POST",
+            "/erp-prefix/api/auth/qr-login/sessions/123/confirm",
+            timestamp,
+            nonce,
+            request.content,
+        )
+        assert hmac.compare_digest(request.headers[SIGNATURE_HEADER], expected)
+        return httpx.Response(
+            200,
+            json={
+                "status": "confirmed",
+                "failure_reason": None,
+                "access_token": "must-not-leak",
+                "user": {"phone": "13800000000"},
+            },
+        )
+
+    transport = httpx.MockTransport(handler)
+
+    def fake_async_client(**kwargs):
+        _ = kwargs
+        return original_async_client(transport=transport)
+
+    monkeypatch.setattr(erp_login, "get_phone_number", fake_get_phone_number)
+    monkeypatch.setattr(erp_login.httpx, "AsyncClient", fake_async_client)
+    client = _client(monkeypatch, erp_base_url="http://erp.local/erp-prefix")
+
+    response = client.post(
+        "/api/erp-login/sessions/ticket-abc/confirm?session_id=123",
+        json={"phone_code": "wx-phone-code"},
+    )
+
+    assert response.status_code == 200
+    assert response.json() == {"status": "confirmed", "failure_reason": None}
+    assert "access_token" not in response.json()
+    erp_body = json.loads(captured["body"].decode("utf-8"))
+    assert erp_body["ticket"] == "ticket-abc"
+    assert erp_body["phone"] == "13800000000"
+    assert erp_body["confirmed_at"]
+    assert erp_body["nonce"]
+
+
+def test_erp_http_status_error_uses_generic_gateway_error(monkeypatch):
+    from app.routers import erp_login
+
+    original_async_client = httpx.AsyncClient
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        assert request.url.path == "/api/auth/qr-login/sessions/123/preview"
+        return httpx.Response(403, json={"detail": "do-not-leak"})
+
+    transport = httpx.MockTransport(handler)
+
+    def fake_async_client(**kwargs):
+        _ = kwargs
+        return original_async_client(transport=transport)
+
+    monkeypatch.setattr(erp_login.httpx, "AsyncClient", fake_async_client)
+    client = _client(monkeypatch)
+
+    response = client.get("/api/erp-login/sessions/ticket-abc?session_id=123")
+
+    assert response.status_code == 502
+    assert response.json()["detail"] == "ERP扫码登录服务暂时不可用"
+
+
+def test_erp_request_error_uses_generic_gateway_error(monkeypatch):
+    from app.routers import erp_login
+
+    original_async_client = httpx.AsyncClient
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        raise httpx.ConnectError("boom", request=request)
+
+    transport = httpx.MockTransport(handler)
+
+    def fake_async_client(**kwargs):
+        _ = kwargs
+        return original_async_client(transport=transport)
+
+    monkeypatch.setattr(erp_login.httpx, "AsyncClient", fake_async_client)
+    client = _client(monkeypatch)
+
+    response = client.get("/api/erp-login/sessions/ticket-abc?session_id=123")
+
+    assert response.status_code == 502
+    assert response.json()["detail"] == "ERP扫码登录服务暂时不可用"
+
+
+def test_erp_invalid_json_uses_invalid_response_error(monkeypatch):
+    from app.routers import erp_login
+
+    original_async_client = httpx.AsyncClient
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        assert request.url.path == "/api/auth/qr-login/sessions/123/preview"
+        return httpx.Response(200, content=b"not-json")
+
+    transport = httpx.MockTransport(handler)
+
+    def fake_async_client(**kwargs):
+        _ = kwargs
+        return original_async_client(transport=transport)
+
+    monkeypatch.setattr(erp_login.httpx, "AsyncClient", fake_async_client)
+    client = _client(monkeypatch)
+
+    response = client.get("/api/erp-login/sessions/ticket-abc?session_id=123")
+
+    assert response.status_code == 502
+    assert response.json()["detail"] == "ERP扫码登录响应无效"
+
+
+def test_create_device_qrcode_delegates_to_miniapp_qrcode_directory(monkeypatch):
+    calls = []
+
+    async def fake_create_miniapp_qrcode(**kwargs):
+        calls.append(kwargs)
+        return "https://miniapp.example.com/uploads/qrcodes/device.png"
+
+    monkeypatch.setattr(wechat, "create_miniapp_qrcode", fake_create_miniapp_qrcode, raising=False)
+
+    result = asyncio.run(
+        wechat.create_device_qrcode(
+            "28#",
+            page="pages/clock/clock",
+            scene="mold=abc",
+            public_base_url="https://miniapp.example.com",
+            label="模具 28#",
+        )
+    )
+
+    assert result == "https://miniapp.example.com/uploads/qrcodes/device.png"
+    assert calls == [{
+        "key": "28#",
+        "page": "pages/clock/clock",
+        "scene": "mold=abc",
+        "public_base_url": "https://miniapp.example.com",
+        "directory": "qrcodes",
+        "label": "模具 28#",
+    }]