From 1cff7e06aac1e7f01b63682aecb57f611732f23c 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 01:28:12 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9EERP=E6=89=AB=E7=A0=81?= =?UTF-8?q?=E7=99=BB=E5=BD=95=E5=B0=8F=E7=A8=8B=E5=BA=8F=E5=90=8E=E7=AB=AF?= =?UTF-8?q?=E4=BB=A3=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/config.py | 3 + app/main.py | 2 + app/routers/erp_login.py | 205 ++++++++++++++++++++++++++ app/schemas.py | 29 ++++ app/services/erp_login_security.py | 99 +++++++++++++ app/services/wechat.py | 31 +++- tests/test_erp_login.py | 222 +++++++++++++++++++++++++++++ 7 files changed, 586 insertions(+), 5 deletions(-) create mode 100644 app/routers/erp_login.py create mode 100644 app/services/erp_login_security.py create mode 100644 tests/test_erp_login.py diff --git a/app/config.py b/app/config.py index 10baaa0..58c3c4a 100644 --- a/app/config.py +++ b/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: diff --git a/app/main.py b/app/main.py index 2d672bd..a190eaa 100644 --- a/app/main.py +++ b/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) diff --git a/app/routers/erp_login.py b/app/routers/erp_login.py new file mode 100644 index 0000000..390d526 --- /dev/null +++ b/app/routers/erp_login.py @@ -0,0 +1,205 @@ +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" + + +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 | None) -> str: + scene = f"ticket={ticket}" + if session_id is not None: + scene = f"{scene}&session_id={session_id}" + 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 ""), + 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: + detail = _erp_status_error_detail(exc.response) or "ERP扫码登录服务暂时不可用" + raise HTTPException(status_code=502, detail=detail) 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.ticket}", + 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}, + ) + 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) diff --git a/app/schemas.py b/app/schemas.py index 5ae2205..b6f8664 100644 --- a/app/schemas.py +++ b/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=256) + session_id: int | None = None + + +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() diff --git a/app/services/erp_login_security.py b/app/services/erp_login_security.py new file mode 100644 index 0000000..707ade5 --- /dev/null +++ b/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-service-timestamp" +SERVICE_NONCE_HEADER = "x-service-nonce" +SERVICE_SIGNATURE_HEADER = "x-service-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 diff --git a/app/services/wechat.py b/app/services/wechat.py index de28b2a..96fbc7e 100644 --- a/app/services/wechat.py +++ b/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}" diff --git a/tests/test_erp_login.py b/tests/test_erp_login.py new file mode 100644 index 0000000..23531ce --- /dev/null +++ b/tests/test_erp_login.py @@ -0,0 +1,222 @@ +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" + + +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 { + "x-service-timestamp": timestamp, + "x-service-nonce": nonce, + "x-service-signature": _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["x-service-signature"] = "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-ticket-abc.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-ticket-abc.png", + "page": "pages/erpLoginConfirm/erpLoginConfirm", + "scene": "ticket=ticket-abc&session_id=123", + } + assert calls == [{ + "key": "erp-login-ticket-abc", + "page": "pages/erpLoginConfirm/erpLoginConfirm", + "scene": "ticket=ticket-abc&session_id=123", + "public_base_url": "https://miniapp.example.com", + "directory": "erp-login-qrcodes", + "label": "ERP扫码登录", + }] + + +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_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" + timestamp = request.headers["x-service-timestamp"] + nonce = request.headers["x-service-nonce"] + expected = _test_signature( + "POST", + "/erp-prefix/api/auth/qr-login/sessions/123/confirm", + timestamp, + nonce, + request.content, + ) + assert hmac.compare_digest(request.headers["x-service-signature"], 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_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#", + }]