From 0aab4fde1bc1d953a6a0148e6d745597ef9e9ec0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B1=A4=E5=AD=A6=E4=BC=9A?= Date: Mon, 6 Jul 2026 10:42:35 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E6=AD=A3ERP=E6=89=AB=E7=A0=81?= =?UTF-8?q?=E7=99=BB=E5=BD=95=E6=9C=8D=E5=8A=A1=E7=AD=BE=E5=90=8D=E5=A5=91?= =?UTF-8?q?=E7=BA=A6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/routers/erp_login.py | 13 +-------- app/services/erp_login_security.py | 6 ++-- tests/test_erp_login.py | 46 +++++++++++++++++++++++++----- 3 files changed, 43 insertions(+), 22 deletions(-) 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 = []