diff --git a/app/routers/erp_login.py b/app/routers/erp_login.py index 390d526..f7537c7 100644 --- a/app/routers/erp_login.py +++ b/app/routers/erp_login.py @@ -51,16 +51,6 @@ def _scene(ticket: str, session_id: int | None) -> str: return scene -def _erp_status_error_detail(response: httpx.Response) -> Any: - try: - data = response.json() - except ValueError: - return None - if isinstance(data, dict): - return data.get("detail") - return None - - def _action_response(data: dict[str, Any]) -> ErpLoginActionResponse: return ErpLoginActionResponse( status=str(data.get("status") or ""), @@ -72,8 +62,7 @@ async def _read_erp_json(response: httpx.Response) -> dict[str, Any]: try: response.raise_for_status() except httpx.HTTPStatusError as exc: - detail = _erp_status_error_detail(exc.response) or "ERP扫码登录服务暂时不可用" - raise HTTPException(status_code=502, detail=detail) from exc + raise HTTPException(status_code=502, detail="ERP扫码登录服务暂时不可用") from exc try: data = response.json() except ValueError as exc: diff --git a/app/services/erp_login_security.py b/app/services/erp_login_security.py index 707ade5..d4f320b 100644 --- a/app/services/erp_login_security.py +++ b/app/services/erp_login_security.py @@ -6,9 +6,9 @@ from uuid import uuid4 from fastapi import HTTPException -SERVICE_TIMESTAMP_HEADER = "x-service-timestamp" -SERVICE_NONCE_HEADER = "x-service-nonce" -SERVICE_SIGNATURE_HEADER = "x-service-signature" +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] = {} diff --git a/tests/test_erp_login.py b/tests/test_erp_login.py index 23531ce..5e0e753 100644 --- a/tests/test_erp_login.py +++ b/tests/test_erp_login.py @@ -14,6 +14,9 @@ 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: @@ -47,9 +50,9 @@ def _signed_headers( timestamp = str(int(time.time())) nonce = uuid4().hex return { - "x-service-timestamp": timestamp, - "x-service-nonce": nonce, - "x-service-signature": _test_signature(method, path, timestamp, nonce, body, secret), + TIMESTAMP_HEADER: timestamp, + NONCE_HEADER: nonce, + SIGNATURE_HEADER: _test_signature(method, path, timestamp, nonce, body, secret), "content-type": "application/json", } @@ -81,7 +84,7 @@ def test_qrcode_rejects_missing_or_bad_service_signature(monkeypatch): headers={"content-type": "application/json"}, ) bad_headers = _signed_headers("POST", "/api/erp-login/qrcode", body) - bad_headers["x-service-signature"] = "bad-signature" + bad_headers[SIGNATURE_HEADER] = "bad-signature" bad = client.post("/api/erp-login/qrcode", content=body, headers=bad_headers) assert missing.status_code == 401 @@ -147,8 +150,13 @@ def test_confirm_gets_phone_signs_erp_request_with_base_path_and_hides_auth_sess captured["body"] = request.content captured["headers"] = request.headers assert request.url.path == "/erp-prefix/api/auth/qr-login/sessions/123/confirm" - timestamp = request.headers["x-service-timestamp"] - nonce = request.headers["x-service-nonce"] + 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", @@ -156,7 +164,7 @@ def test_confirm_gets_phone_signs_erp_request_with_base_path_and_hides_auth_sess nonce, request.content, ) - assert hmac.compare_digest(request.headers["x-service-signature"], expected) + assert hmac.compare_digest(request.headers[SIGNATURE_HEADER], expected) return httpx.Response( 200, json={ @@ -192,6 +200,30 @@ def test_confirm_gets_phone_signs_erp_request_with_base_path_and_hides_auth_sess 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_create_device_qrcode_delegates_to_miniapp_qrcode_directory(monkeypatch): calls = []