From 7fabc237159c77d89b8acb46dcd7f51bbb6085c2 Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Mon, 15 Jun 2026 11:39:31 +0800 Subject: [PATCH] fix(t12): close payment delivery and governance remediation --- .github/workflows/ci.yml | 15 +- admin/app.py | 9 + admin/config.py | 137 +++++- admin/routes/web_public.py | 42 +- admin/tests/test_order_deletion.py | 29 +- admin/tests/test_order_info_form.py | 6 +- admin/tests/test_web_public.py | 7 + data/channel_sync/dao_extension.py | 240 +--------- data/channel_sync/webhook_server.py | 48 +- data/notifications/dispatcher.py | 98 +++- data/notifications/email_service.py | 51 +- data/notifications/smtp_sender.py | 64 +++ data/orders/dao.py | 40 +- data/orders/deletion_service.py | 18 + data/orders/models.py | 1 + data/orders/public_flow.py | 2 + data/orders/schema.py | 6 + data/orders/tests/test_models.py | 3 + data/payments/providers/alipay.py | 4 + data/payments/tests/test_provider_alipay.py | 2 + data/payments/tests/test_webhook.py | 87 +++- data/share/permission.py | 53 ++- data/share/tests/test_permission.py | 157 +++++- docs/ACTIVE_REMEDIATION_2026-06-13.md | 36 +- docs/CURRENT_STATE.md | 12 +- docs/DELIVERY_SERVICE_DESIGN.md | 8 +- docs/PAYMENT_PROVIDER_ONBOARDING.md | 2 +- reports/PROJECT_SYSTEM_REVIEW_2026-06-14.md | 499 ++++++++++++++++++++ scripts/backup_restore_smoke.py | 2 +- scripts/check_coverage_gate.py | 4 +- scripts/dev-verify.sh | 3 + scripts/gaokao-delivery-dispatch.py | 19 +- tests/test_delivery_dispatcher.py | 75 ++- tests/test_delivery_notification.py | 25 +- tests/test_smtp_sender.py | 63 +++ 35 files changed, 1529 insertions(+), 338 deletions(-) create mode 100644 data/notifications/smtp_sender.py create mode 100644 reports/PROJECT_SYSTEM_REVIEW_2026-06-14.md create mode 100644 tests/test_smtp_sender.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index db04e8d..581c84f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -61,19 +61,10 @@ jobs: python --version pytest --version - - name: Run tests with coverage - # T5.5: 基线门禁 = 整体覆盖率 >=60%，核心覆盖率 >=80%（由 check_coverage_gate.py 校验） + - name: Run unified quality gate + # P1-7: CI 与本地统一走 scripts/dev-verify.sh，避免 coverage / lint / mypy 口径漂移 run: | - pytest admin/tests tests data \ - --cov=admin \ - --cov=data \ - --cov=skills \ - --cov=scripts \ - --cov-report=term-missing \ - --cov-report=xml \ - --cov-fail-under=60 \ - -v - python scripts/check_coverage_gate.py coverage.xml + PYTHON_BIN=python bash scripts/dev-verify.sh - name: Upload coverage artifact if: always() && matrix.python-version == '3.11' && hashFiles('coverage.xml') != '' diff --git a/admin/app.py b/admin/app.py index eaebb6c..8bbfbc9 100644 --- a/admin/app.py +++ b/admin/app.py @@ -27,6 +27,7 @@ from admin.config import ( Settings, is_default_admin_password_secure, is_jwt_secret_secure, + is_portal_token_secret_secure, load_settings, ) from admin.db import bootstrap_admin, ensure_schema @@ -70,6 +71,14 @@ def _validate_and_log_settings(settings: Settings) -> None: f"default admin password insecure in prod: {admin_reason}" ) logger.warning("默认管理员密码提示: %s", admin_reason) + portal_secure, portal_reason = is_portal_token_secret_secure(settings) + if not portal_secure: + if settings.env == "prod": + logger.error( + "Portal token secret 不安全: %s — 拒绝启动生产环境!", portal_reason + ) + raise RuntimeError(f"portal token secret insecure in prod: {portal_reason}") + logger.warning("Portal token secret 提示: %s", portal_reason) logger.info( "Admin API 启动: env=%s db=%s jwt_exp_min=%d", settings.env, diff --git a/admin/config.py b/admin/config.py index 3dac4a6..83d97b1 100644 --- a/admin/config.py +++ b/admin/config.py @@ -16,8 +16,17 @@ from fastapi import Request # 安全占位密钥（仅 dev 环境）。生产必须显式设置 GAOKAO_JWT_SECRET。 _DEV_JWT_SECRET = "dev-only-do-not-use-in-prod-please-override-via-env" +# Portal token 独立 dev 占位密钥。**与后台 JWT 密钥不同**,P2-4 验收要求。 +# 生产环境必须显式设置 GAOKAO_PORTAL_TOKEN_SECRET。 +_DEV_PORTAL_TOKEN_SECRET = "dev-only-portal-token-secret-do-not-use-in-prod" +# Payment webhook dev 默认密钥。生产环境必须显式设置,否则 fail-closed (P2-5)。 +_DEV_PAYMENT_WEBHOOK_SECRET = "dev-mock-payment-secret" _DEFAULT_ADMIN_PASSWORD = "admin123" _MIN_ADMIN_PASSWORD_LENGTH = 10 +# Webhook secret 最低长度门槛 (生产环境) +_MIN_PROD_WEBHOOK_SECRET_LEN = 16 +# Portal token secret 最低长度门槛 (生产环境) +_MIN_PORTAL_TOKEN_SECRET_LEN = 32 @dataclass(frozen=True) @@ -37,13 +46,102 @@ class Settings: payment_app_id: str payment_private_key_path: str payment_alipay_public_key_path: str + smtp_host: str + smtp_port: int + smtp_sender: str + smtp_username: str + smtp_password: str + smtp_use_tls: bool + smtp_use_ssl: bool jwt_secret: str + portal_token_secret: str # P2-4: 与后台 jwt_secret 分离 jwt_algorithm: str jwt_expire_minutes: int default_admin_username: str default_admin_password: str +def _resolve_payment_webhook_secret(env: str) -> str: + """解析支付 webhook secret,根据环境返回实际值。 + + dev/test 保留 dev 默认值 ``_DEV_PAYMENT_WEBHOOK_SECRET``,便于本地开发/测试; + 生产环境不读取默认值,强制要求显式设置环境变量。 + """ + raw = os.getenv("GAOKAO_PAYMENT_WEBHOOK_SECRET") + if raw is None or raw == "": + if env == "prod": + # 生产环境未设置环境变量 → 抛错,防止 Settings 实例化成功后再次绕过校验 + raise RuntimeError( + "GAOKAO_PAYMENT_WEBHOOK_SECRET 必须在生产环境显式设置 " + "(当前缺失或为空),已拒绝启动 (P2-5 fail-closed)" + ) + return _DEV_PAYMENT_WEBHOOK_SECRET + return raw + + +def _enforce_payment_webhook_secret_policy(settings: Settings) -> None: + """生产环境 webhook secret fail-closed 校验 (P2-5)。 + + 不接受: + - 缺失/空字符串 (由 ``_resolve_payment_webhook_secret`` 处理,但为防御性双重检查) + - dev 占位 secret (``_DEV_PAYMENT_WEBHOOK_SECRET``) + - 长度 < ``_MIN_PROD_WEBHOOK_SECRET_LEN`` 字符 + """ + if settings.env != "prod": + return + secret = settings.payment_webhook_secret + if not secret: + raise RuntimeError( + "GAOKAO_PAYMENT_WEBHOOK_SECRET 缺失,生产环境必须显式设置 (P2-5 fail-closed)" + ) + if secret == _DEV_PAYMENT_WEBHOOK_SECRET: + raise RuntimeError( + "GAOKAO_PAYMENT_WEBHOOK_SECRET 仍使用 dev 默认值,生产环境禁止 " + "(P2-5 fail-closed)" + ) + if len(secret) < _MIN_PROD_WEBHOOK_SECRET_LEN: + raise RuntimeError( + "GAOKAO_PAYMENT_WEBHOOK_SECRET 长度 {} 小于生产环境最低要求 {} " + "(P2-5 fail-closed)".format(len(secret), _MIN_PROD_WEBHOOK_SECRET_LEN) + ) + + +def _resolve_portal_token_secret(env: str) -> str: + raw = os.getenv("GAOKAO_PORTAL_TOKEN_SECRET") + if raw is None or raw == "": + if env == "prod": + raise RuntimeError( + "GAOKAO_PORTAL_TOKEN_SECRET 必须在生产环境显式设置 " + "(当前缺失或为空),已拒绝启动 (P2-4 fail-closed)" + ) + return _DEV_PORTAL_TOKEN_SECRET + return raw + + +def is_portal_token_secret_secure(settings: Settings) -> tuple[bool, str]: + secret = settings.portal_token_secret + if settings.env == "prod": + if secret == _DEV_PORTAL_TOKEN_SECRET: + return False, "生产环境禁止使用默认 portal token secret" + if len(secret) < _MIN_PORTAL_TOKEN_SECRET_LEN: + return False, "portal token secret 长度必须 >= 32" + if secret == settings.jwt_secret: + return False, "portal token secret 必须与 JWT secret 分离" + if settings.env == "dev" and secret == _DEV_PORTAL_TOKEN_SECRET: + return False, "dev 环境使用占位 portal token secret（仅本地开发可接受）" + if len(secret) < _MIN_PORTAL_TOKEN_SECRET_LEN: + return False, "portal token secret 长度必须 >= 32" + if secret == settings.jwt_secret: + return False, "portal token secret 必须与 JWT secret 分离" + return True, "ok" + + +def _enforce_portal_token_secret_policy(settings: Settings) -> None: + secure, reason = is_portal_token_secret_secure(settings) + if not secure and settings.env == "prod": + raise RuntimeError(f"GAOKAO_PORTAL_TOKEN_SECRET invalid in prod: {reason}") + + def load_settings() -> Settings: """从环境变量加载配置。 @@ -54,21 +152,34 @@ def load_settings() -> Settings: - GAOKAO_SHARE_REPORT_DIR : 分享报告 JSON 目录，默认 data/share/reports - GAOKAO_PAYMENT_PROVIDER : 支付 provider，默认 mock - GAOKAO_PAYMENT_BASE_URL : 支付相关回跳基础 URL，默认 http://testserver - - GAOKAO_PAYMENT_WEBHOOK_SECRET : 支付 webhook 独立签名密钥，默认 dev mock secret + - GAOKAO_PAYMENT_WEBHOOK_SECRET : 支付 webhook 独立签名密钥。生产环境 (GAOKAO_ENV=prod) + 必须显式设置强密钥,否则启动 fail-closed (P2-5)。 + dev 保留 ``dev-mock-payment-secret`` 默认值。 - GAOKAO_PAYMENT_NOTIFY_URL : 真实 provider 异步通知地址，默认空 - GAOKAO_PAYMENT_RETURN_URL : 真实 provider 浏览器返回地址，默认空 - GAOKAO_PAYMENT_APP_ID : 真实 provider 应用 ID，默认空 - GAOKAO_PAYMENT_PRIVATE_KEY_PATH : 真实 provider 私钥路径，默认空 - GAOKAO_PAYMENT_ALIPAY_PUBLIC_KEY_PATH : 支付宝公钥路径，默认空 + - GAOKAO_SMTP_HOST : SMTP 主机，默认空 + - GAOKAO_SMTP_PORT : SMTP 端口，默认 25 + - GAOKAO_SMTP_SENDER : 发件邮箱，默认空 + - GAOKAO_SMTP_USER : SMTP 用户名，默认空 + - GAOKAO_SMTP_PASS : SMTP 密码，默认空 + - GAOKAO_SMTP_USE_TLS : 是否启用 STARTTLS，默认 false + - GAOKAO_SMTP_USE_SSL : 是否启用 SMTPS，默认 false - GAOKAO_JWT_SECRET : HS256 密钥，默认 dev 占位（启动日志 WARN） + - GAOKAO_PORTAL_TOKEN_SECRET : Portal token 独立签名密钥 (P2-4)。 + 与 ``GAOKAO_JWT_SECRET`` **分离**:默认 dev 占位, + 生产环境必须显式设置且与 JWT secret 不同。 - GAOKAO_JWT_EXP_MIN : JWT 过期时间（分钟），默认 60 - GAOKAO_ADMIN_USER : 默认管理员用户名，默认 admin - GAOKAO_ADMIN_PASS : 默认管理员密码，默认 admin123（仅本地开发占位） - Returns: - Settings: 不可变配置实例 + Raises: + RuntimeError: 生产环境下 GAOKAO_PAYMENT_WEBHOOK_SECRET 缺失/默认值/长度不足 + 时 fail-closed。 """ - return Settings( + settings = Settings( env=os.getenv("GAOKAO_ENV", "dev"), db_path=os.getenv("GAOKAO_DB_PATH", "data/orders/admin.db"), orders_db_path=os.getenv("GAOKAO_ORDERS_DB_PATH", "data/orders.db"), @@ -76,8 +187,8 @@ def load_settings() -> Settings: share_report_dir=os.getenv("GAOKAO_SHARE_REPORT_DIR", "data/share/reports"), payment_provider=os.getenv("GAOKAO_PAYMENT_PROVIDER", "mock"), payment_base_url=os.getenv("GAOKAO_PAYMENT_BASE_URL", "http://testserver"), - payment_webhook_secret=os.getenv( - "GAOKAO_PAYMENT_WEBHOOK_SECRET", "dev-mock-payment-secret" + payment_webhook_secret=_resolve_payment_webhook_secret( + os.getenv("GAOKAO_ENV", "dev") ), payment_notify_url=os.getenv("GAOKAO_PAYMENT_NOTIFY_URL", ""), payment_return_url=os.getenv("GAOKAO_PAYMENT_RETURN_URL", ""), @@ -86,12 +197,26 @@ def load_settings() -> Settings: payment_alipay_public_key_path=os.getenv( "GAOKAO_PAYMENT_ALIPAY_PUBLIC_KEY_PATH", "" ), + smtp_host=os.getenv("GAOKAO_SMTP_HOST", ""), + smtp_port=int(os.getenv("GAOKAO_SMTP_PORT", "25")), + smtp_sender=os.getenv("GAOKAO_SMTP_SENDER", ""), + smtp_username=os.getenv("GAOKAO_SMTP_USER", ""), + smtp_password=os.getenv("GAOKAO_SMTP_PASS", ""), + smtp_use_tls=os.getenv("GAOKAO_SMTP_USE_TLS", "false").lower() == "true", + smtp_use_ssl=os.getenv("GAOKAO_SMTP_USE_SSL", "false").lower() == "true", jwt_secret=os.getenv("GAOKAO_JWT_SECRET", _DEV_JWT_SECRET), + portal_token_secret=_resolve_portal_token_secret( + os.getenv("GAOKAO_ENV", "dev") + ), jwt_algorithm=os.getenv("GAOKAO_JWT_ALGORITHM", "HS256"), jwt_expire_minutes=int(os.getenv("GAOKAO_JWT_EXP_MIN", "60")), default_admin_username=os.getenv("GAOKAO_ADMIN_USER", "admin"), default_admin_password=os.getenv("GAOKAO_ADMIN_PASS", _DEFAULT_ADMIN_PASSWORD), ) + # 生产环境 post-load 校验:webhook / portal token secret 必须满足强度门槛。 + _enforce_payment_webhook_secret_policy(settings) + _enforce_portal_token_secret_policy(settings) + return settings def is_jwt_secret_secure(settings: Settings) -> tuple: diff --git a/admin/routes/web_public.py b/admin/routes/web_public.py index 7a43634..ba12a58 100644 --- a/admin/routes/web_public.py +++ b/admin/routes/web_public.py @@ -24,6 +24,7 @@ from data.customer_portal.token import ( issue_portal_token, verify_portal_token, ) +from data.notifications.email_service import DeliveryNotificationService from data.orders.dao import OrderNotFound, OrdersDAO from data.orders.intake_schema import IntakePayload from data.orders.intake_store import IntakeStore @@ -114,7 +115,7 @@ def create_public_order_endpoint( with OrdersDAO.connect(settings.orders_db_path) as dao: order = create_public_order(dao, payload) - portal_token = issue_portal_token(order.id, settings.jwt_secret) + portal_token = issue_portal_token(order.id, settings.portal_token_secret) try: checkout = payment_service.create_checkout(order.id, portal_token=portal_token) except PaymentError as exc: @@ -357,7 +358,7 @@ def _payment_service(settings: Settings) -> PaymentService: def _resolve_order_from_token(token: str, settings: Settings) -> Order: try: - payload = verify_portal_token(token, settings.jwt_secret) + payload = verify_portal_token(token, settings.portal_token_secret) except PortalTokenError as exc: raise HTTPException(status_code=401, detail=str(exc)) from exc with OrdersDAO.connect(settings.orders_db_path) as dao: @@ -375,6 +376,15 @@ def _build_portal_context(order: Order, settings: Settings) -> dict[str, Any]: intake = intake_store.get(order.id) finally: intake_store.close() + notification_service = DeliveryNotificationService.for_db(settings.orders_db_path) + try: + sent_station_events = notification_service.list_events( + order.id, + status="sent", + channel="station", + ) + finally: + notification_service.close() stage = "pending_payment" report_html_ready = bool(order.audit_report and Path(order.audit_report).is_file()) @@ -402,6 +412,15 @@ def _build_portal_context(order: Order, settings: Settings) -> dict[str, Any]: stage = "info_submitted" title, subtitle = _STAGE_META[stage] + latest_station_notice: dict[str, Any] | None = None + if sent_station_events: + try: + latest_payload = json.loads(sent_station_events[-1].payload_json) + except json.JSONDecodeError: + latest_payload = {} + station_notice = latest_payload.get("station_notice") + if isinstance(station_notice, dict): + latest_station_notice = station_notice return { "stage": stage, "stage_title": title, @@ -411,6 +430,7 @@ def _build_portal_context(order: Order, settings: Settings) -> dict[str, Any]: "report_html_ready": report_html_ready, "report_pdf_ready": report_pdf_ready, "order": order, + "station_notice": latest_station_notice, } @@ -540,6 +560,7 @@ def _render_checkout_page(service_version: str) -> str:
+ @@ -557,6 +578,7 @@ def _render_checkout_page(service_version: str) -> str: amount_cents: {amount}, customer_name: form.get('customer_name'), customer_phone: form.get('customer_phone'), + customer_email: form.get('customer_email') || null, candidate_name: form.get('candidate_name'), candidate_province: form.get('candidate_province'), notes: form.get('notes'), @@ -673,7 +695,7 @@ def _complete_simulated_payment( if payment is None: raise HTTPException(status_code=404, detail="payment not found") try: - portal_payload = verify_portal_token(token, settings.jwt_secret) + portal_payload = verify_portal_token(token, settings.portal_token_secret) except PortalTokenError as exc: raise HTTPException(status_code=401, detail=str(exc)) from exc if ( @@ -777,6 +799,19 @@ def _render_status_page(token: str, context: dict[str, Any]) -> str: ) if not report_links: report_links = "
  • 报告生成中，交付物就绪后这里会显示查看/下载入口。
    • " + station_notice_html = "" + station_notice = context.get("station_notice") + if isinstance(station_notice, dict): + title = escape(str(station_notice.get("title") or "站内通知")) + body = escape(str(station_notice.get("body") or "")) + sent_at = escape(str(station_notice.get("sent_at") or "")) + station_notice_html = f""" +
    +

    通知已发送

    +

    {title}

    +

    {body}

    +

    发送时间：{sent_at}

    +
    """ return f""" 订单状态 @@ -788,6 +823,7 @@ def _render_status_page(token: str, context: dict[str, Any]) -> str:

    服务版本：{escape(order.service_version)}

    当前订单状态：{escape(order.status)}

    + {station_notice_html}

    下一步操作

      diff --git a/admin/tests/test_order_deletion.py b/admin/tests/test_order_deletion.py index fe44b5c..728f662 100644 --- a/admin/tests/test_order_deletion.py +++ b/admin/tests/test_order_deletion.py @@ -20,6 +20,7 @@ def _seed_order(db_path: str, order_id: str = "GKO-20260614-DELETE") -> Order: customer_name="张家长", customer_phone="13800138000", customer_wechat="wx-parent-01", + customer_email="parent@example.com", candidate_name="张三", candidate_province="湖南", notes="需要删除的测试订单", @@ -78,7 +79,8 @@ def test_admin_delete_order_removes_artifacts_and_related_records( notification_service = DeliveryNotificationService.for_db(settings.orders_db_path) try: - assert len(notification_service.list_events(order.id)) == 1 + events = notification_service.list_events(order.id) + assert {event.channel for event in events} == {"station", "email"} finally: notification_service.close() @@ -120,6 +122,12 @@ def test_admin_anonymize_order_masks_pii_but_keeps_order( client, auth_headers, settings ): order = _seed_order(settings.orders_db_path, order_id="GKO-20260614-ANON") + _mark_paid(settings, order) + IntakeStore.for_db(settings.orders_db_path).save( + order_id=order.id, + payload={"candidate_score": 578, "guardian_notes": "contains pii"}, + submit=True, + ) resp = client.delete( f"/api/orders/{order.id}?mode=anonymize&reason=retention_expired", @@ -135,9 +143,28 @@ def test_admin_anonymize_order_masks_pii_but_keeps_order( assert anonymized.customer_name == "已匿名化" assert anonymized.customer_phone is None assert anonymized.customer_wechat is None + assert anonymized.customer_email is None assert anonymized.candidate_name == "匿名考生" assert anonymized.notes == "[ANONYMIZED] retention_expired" + intake_store = IntakeStore.for_db(settings.orders_db_path) + try: + intake = intake_store.get(order.id) + finally: + intake_store.close() + assert intake is not None + assert intake.payload == {} + + payment_service = PaymentService.for_db( + settings.orders_db_path, + base_url=settings.payment_base_url, + webhook_secret=settings.payment_webhook_secret, + ) + payment = payment_service.get_payment_by_order(order.id) + assert payment is not None + assert payment.callback_payload is None + assert payment.checkout_token is None + deletion_service = OrderDeletionService.for_db(settings.orders_db_path) try: assert deletion_service.audit_count(order.id) == 1 diff --git a/admin/tests/test_order_info_form.py b/admin/tests/test_order_info_form.py index 778738e..1db2ee4 100644 --- a/admin/tests/test_order_info_form.py +++ b/admin/tests/test_order_info_form.py @@ -41,7 +41,7 @@ def _mark_paid(settings, order: Order) -> None: def test_order_info_form_accepts_draft_and_submit(client, settings): order = _seed_order(settings.orders_db_path) _mark_paid(settings, order) - token = issue_portal_token(order.id, settings.jwt_secret) + token = issue_portal_token(order.id, settings.portal_token_secret) page = client.get(f"/portal/{token}/info") assert page.status_code == 200, page.text @@ -98,7 +98,7 @@ def test_order_info_form_becomes_read_only_after_report_ready( ): order = _seed_order(settings.orders_db_path, order_id="GKO-20260614-INFO-LOCK") _mark_paid(settings, order) - token = issue_portal_token(order.id, settings.jwt_secret) + token = issue_portal_token(order.id, settings.portal_token_secret) report_path = tmp_path / "locked-report.html" pdf_path = tmp_path / "locked-report.pdf" @@ -134,7 +134,7 @@ def test_order_info_form_becomes_read_only_after_report_ready( def test_submit_requires_consent_fields(client, settings): order = _seed_order(settings.orders_db_path, order_id="GKO-20260614-INFO-CONSENT") _mark_paid(settings, order) - token = issue_portal_token(order.id, settings.jwt_secret) + token = issue_portal_token(order.id, settings.portal_token_secret) resp = client.post( f"/portal/{token}/info", diff --git a/admin/tests/test_web_public.py b/admin/tests/test_web_public.py index 9b20c86..ea16e81 100644 --- a/admin/tests/test_web_public.py +++ b/admin/tests/test_web_public.py @@ -41,6 +41,7 @@ def test_public_create_order_endpoint(client): "amount_cents": 9900, "customer_name": "张家长", "customer_phone": "13800138000", + "customer_email": "parent@example.com", "candidate_name": "张三", "candidate_province": "湖南", }, @@ -55,6 +56,9 @@ def test_public_create_order_endpoint(client): assert body["next_step"] == "payment" assert body["checkout_url"].startswith("/pay/mock/") assert "/portal/" in body["portal_status_url"] + with OrdersDAO.connect(client.app.state.settings.orders_db_path) as dao: + created = dao.get(body["order_id"]) + assert created.customer_email == "parent@example.com" def test_public_create_order_rejects_missing_contact(client): @@ -84,6 +88,7 @@ def test_public_create_order_rejects_price_tampering(client): "amount_cents": 1, "customer_name": "张家长", "customer_phone": "13800138000", + "customer_email": "parent@example.com", "candidate_province": "湖南", }, ) @@ -104,6 +109,7 @@ def test_mock_payment_complete_rejects_wrong_portal_token(client): "amount_cents": 9900, "customer_name": "张家长", "customer_phone": "13800138000", + "customer_email": "parent@example.com", "candidate_province": "湖南", }, ) @@ -123,6 +129,7 @@ def test_payment_return_redirects_to_portal_status(client): "amount_cents": 9900, "customer_name": "张家长", "customer_phone": "13800138000", + "customer_email": "parent@example.com", "candidate_province": "湖南", }, ) diff --git a/data/channel_sync/dao_extension.py b/data/channel_sync/dao_extension.py index 82278de..9e1f7ff 100644 --- a/data/channel_sync/dao_extension.py +++ b/data/channel_sync/dao_extension.py @@ -1,97 +1,17 @@ -"""订单 DAO 扩展 (T8.1 配套) +"""Channel Sync 与 OrdersDAO 兼容层。 -T4.2 主 DAO 仍为 ``todo``；为不阻塞 T8.1，本模块在 -``data/orders/schema.py`` 既有 ``orders`` / ``order_status_history`` 表 -之上提供 Channel Sync 直接需要的两个能力: - -- :func:`upsert_by_external_id` — 按 ``(source, external_id)`` 唯一索引幂等写入 -- :func:`insert_status_history` — 状态机转换历史 - -依赖: - -- :class:`data.orders.models.Order` -- :class:`data.orders.state_machine.is_valid_transition` / :class:`InvalidStateTransition` - -T4.2 落地后，本模块可删除并由 T4.2 接管。 +历史上 T8.1 为避免阻塞，先在本模块内实现了独立的 upsert/status_history 逻辑。 +T4.2 完成后，订单真相已经迁移到 ``data.orders.dao.OrdersDAO``，本文件只保留 +向后兼容入口，避免 channel_sync / 旧测试继续绕过主 DAO。 """ from __future__ import annotations -import json import sqlite3 -from dataclasses import dataclass -from typing import Any, Optional +from typing import Optional -from data.orders.models import Order -from data.orders.state_machine import ( - InvalidStateTransition, - assert_valid_transition, -) - - -@dataclass -class UpsertResult: - """upsert 的返回结构。""" - - order_id: str - action: str # 'inserted' | 'updated' | 'unchanged' | 'illegal_transition' - old_status: Optional[str] = None - new_status: Optional[str] = None - error: Optional[str] = None - - -def _row_to_dict(row) -> dict: - """把 fetchone() 拿到的 tuple 转成 dict，用 cursor.description 取列名。""" - if row is None: - return {} - if hasattr(row, "keys"): - return dict(row) - # 普通 tuple：依赖调用方已设 row_factory 或我们用 description 拼 - cur = getattr(_row_to_dict, "_last_cur", None) - if cur is None or cur.description is None: - raise RuntimeError( - "row is a plain tuple but no cursor.description is available; " - "set conn.row_factory = sqlite3.Row before querying" - ) - return dict(zip([d[0] for d in cur.description], row)) - - -def _row_to_order(row) -> Order: - """把数据库行转换为 Order（借助 from_db_row 解析加密字段）。""" - if row is None: - raise ValueError("cannot convert None row to Order") - if hasattr(row, "keys"): - return Order.from_db_row(dict(row)) - # tuple: 需要列名 - cur = getattr(_row_to_order, "_last_cur", None) - if cur is None or cur.description is None: - raise RuntimeError( - "row is a plain tuple but no cursor.description available; " - "set conn.row_factory = sqlite3.Row before querying" - ) - return Order.from_db_row(dict(zip([d[0] for d in cur.description], row))) - - -def _coerce_value(key: str, value: Any) -> Any: - """保证 tags / candidate_subjects 落盘为 JSON 字符串。""" - if key in ("tags", "candidate_subjects") and isinstance(value, (list, tuple)): - return json.dumps(list(value), ensure_ascii=False) - return value - - -def _column_list() -> str: - """``orders`` 表的可写列清单（与 schema.py 对齐）。""" - return ( - "id, source, external_id, service_version, amount_cents, status, " - "status_updated_at, customer_name, customer_phone_enc, " - "customer_phone_hash, customer_wechat, candidate_name, " - "candidate_id_card_enc, candidate_province, candidate_score, " - "candidate_rank, candidate_subjects, candidate_interests, " - "candidate_strong_subjects, candidate_weak_subjects, " - "candidate_family, assigned_consultant, plan_file, audit_report, " - "pdf_path, created_at, paid_at, started_at, delivered_at, " - "completed_at, notes, tags, upgrade_from" - ) +from data.orders.dao import OrdersDAO, UpsertResult +from data.orders.models import Order, utc_now_iso def upsert_by_external_id( @@ -101,125 +21,8 @@ def upsert_by_external_id( actor: str = "xianyu_webhook", reason: Optional[str] = None, ) -> UpsertResult: - """按 (source, external_id) 唯一索引写入或更新订单。 - - 行为: - - - **不存在** → 插入新行 + 写 status_history(from=None → status) - - **已存在且状态可推进** → 更新 status / status_updated_at + 写 status_history - - **已存在且状态非法转换** → 不写库，返回 ``action='illegal_transition'`` - - **已存在且状态不变** → ``action='unchanged'``，不写 status_history - - 入参: - conn: 启用了 ``PRAGMA foreign_keys = ON`` 的 sqlite3 连接 - order: 待写入的 Order；id 与外部订单号会用于查重 - actor: 操作者标签（写入 status_history） - reason: 可选原因描述 - """ - if not order.external_id: - return UpsertResult( - order_id=order.id, - action="illegal_transition", - error="external_id 缺失，无法做幂等 upsert", - ) - - # 1) 查询是否已存在（统一用 sqlite3.Row 工厂） - prior_factory = conn.row_factory - conn.row_factory = sqlite3.Row - try: - row = conn.execute( - "SELECT * FROM orders WHERE source=? AND external_id=? LIMIT 1", - (order.source, order.external_id), - ).fetchone() - finally: - conn.row_factory = prior_factory - - if row is None: - # INSERT - db_row = order.to_db_row() - # 过滤掉 schema 中不存在的列（防御性） - valid_cols = set(_column_list().split(", ")) - db_row = {k: _coerce_value(k, v) for k, v in db_row.items() if k in valid_cols} - cols = list(db_row.keys()) - placeholders = ",".join("?" for _ in cols) - values = [db_row[c] for c in cols] - conn.execute( - f"INSERT INTO orders ({','.join(cols)}) VALUES ({placeholders})", - values, - ) - insert_status_history( - conn, - order_id=order.id, - from_status=None, - to_status=order.status, - actor=actor, - reason=reason or "channel_sync_upsert_insert", - ) - conn.commit() - return UpsertResult( - order_id=order.id, - action="inserted", - old_status=None, - new_status=order.status, - ) - - # 2) 已存在：判断状态转换 - existing = _row_to_order(row) - old_status = existing.status - if old_status == order.status: - # 状态未变：可选择更新业务字段（amount / 客户信息） - # 但 Channel Sync 场景下避免覆盖人工录入的字段，只更新可空字段 - return UpsertResult( - order_id=existing.id, - action="unchanged", - old_status=old_status, - new_status=order.status, - ) - try: - assert_valid_transition(old_status, order.status) - except InvalidStateTransition as e: - conn.commit() - return UpsertResult( - order_id=existing.id, - action="illegal_transition", - old_status=old_status, - new_status=order.status, - error=str(e), - ) - - # 3) 合法的状态推进 - conn.execute( - """ - UPDATE orders SET - status=?, - status_updated_at=?, - paid_at = COALESCE(paid_at, ?), - completed_at = COALESCE(completed_at, ?) - WHERE id=? - """, - ( - order.status, - order.status_updated_at or order.created_at, - order.paid_at, - order.completed_at, - existing.id, - ), - ) - insert_status_history( - conn, - order_id=existing.id, - from_status=old_status, - to_status=order.status, - actor=actor, - reason=reason or f"channel_sync_update_{order.source}", - ) - conn.commit() - return UpsertResult( - order_id=existing.id, - action="updated", - old_status=old_status, - new_status=order.status, - ) + """兼容旧调用，统一委托给 OrdersDAO.upsert_by_external_id。""" + return OrdersDAO(conn).upsert_by_external_id(order, actor=actor, reason=reason) def insert_status_history( @@ -232,18 +35,15 @@ def insert_status_history( reason: Optional[str] = None, changed_at: Optional[str] = None, ) -> int: - """插入一条 order_status_history 记录，返回 rowid。""" - from data.orders.models import utc_now_iso # local import 避免循环 - - if changed_at is None: - changed_at = utc_now_iso() - cur = conn.execute( - """ - INSERT INTO order_status_history( - order_id, from_status, to_status, actor, reason, changed_at - ) VALUES (?, ?, ?, ?, ?, ?) - """, - (order_id, from_status, to_status, actor, reason, changed_at), + """兼容旧调用，统一走 OrdersDAO 的状态历史写入口。""" + return OrdersDAO(conn)._insert_status_history( + order_id=order_id, + from_status=from_status, + to_status=to_status, + actor=actor, + reason=reason, + changed_at=changed_at or utc_now_iso(), ) - # 不在此处 commit，由调用方统一事务 - return int(cur.lastrowid or 0) + + +__all__ = ["UpsertResult", "upsert_by_external_id", "insert_status_history"] diff --git a/data/channel_sync/webhook_server.py b/data/channel_sync/webhook_server.py index 03c99a4..3b0edcf 100644 --- a/data/channel_sync/webhook_server.py +++ b/data/channel_sync/webhook_server.py @@ -119,26 +119,52 @@ def _open_db(db_path: str) -> sqlite3.Connection: return conn -# 用工厂函数为每个 server 实例创建一个 DB 连接（简单做法，单进程） -_DB_CONN: Optional[sqlite3.Connection] = None +# 按 db_path 隔离的连接缓存（修复 P1-4） +# +# 旧实现是模块级单例 ``_DB_CONN``,不区分 db_path: +# 1) 先 make_server(A.db) 会缓存连接到 A.db +# 2) 之后 make_server(B.db) 仍然返回 A.db 的连接 +# → 第二个 server 的所有写入都进了 A.db,跨库污染 +# +# 修复后: 连接按 db_path 索引,每个路径独立持有一个连接; +# 单进程下 webhook server 持有的连接数 = 实际 db_path 数量。 +# 该结构仍然保留 ``close_db_for_tests`` 旧 API(关闭全部),新增 +# ``close_db_for_path(path)`` 便于按需 teardown。 +_DB_CONNS: dict[str, sqlite3.Connection] = {} _DB_CONN_LOCK = threading.Lock() def _get_db(db_path: str) -> sqlite3.Connection: - global _DB_CONN + """按 ``db_path`` 取连接;该路径未缓存则新建并应用 schema。""" with _DB_CONN_LOCK: - if _DB_CONN is None: - _DB_CONN = _open_db(db_path) - return _DB_CONN + existing = _DB_CONNS.get(db_path) + if existing is not None: + return existing + conn = _open_db(db_path) + _DB_CONNS[db_path] = conn + return conn def close_db_for_tests() -> None: - """单测 teardown 关闭全局连接。""" - global _DB_CONN + """单测 teardown:关闭所有缓存连接(覆盖旧 API,关闭全部)。""" with _DB_CONN_LOCK: - if _DB_CONN is not None: - _DB_CONN.close() - _DB_CONN = None + for conn in _DB_CONNS.values(): + try: + conn.close() + except Exception: + pass + _DB_CONNS.clear() + + +def close_db_for_path(db_path: str) -> None: + """单测按路径 teardown;未注册路径为 no-op。""" + with _DB_CONN_LOCK: + conn = _DB_CONNS.pop(db_path, None) + if conn is not None: + try: + conn.close() + except Exception: + pass def _client_ip(headers, client_address=None) -> str: diff --git a/data/notifications/dispatcher.py b/data/notifications/dispatcher.py index e74b360..1a14a4e 100644 --- a/data/notifications/dispatcher.py +++ b/data/notifications/dispatcher.py @@ -8,6 +8,8 @@ from data.notifications.email_service import ( DeliveryNotificationEvent, DeliveryNotificationService, ) +from data.notifications.smtp_sender import SMTPDeliverySender +from data.orders.models import utc_now_iso @dataclass @@ -18,12 +20,26 @@ class DispatchResult: class DeliveryDispatcher: - def __init__(self, service: DeliveryNotificationService) -> None: + def __init__( + self, + service: DeliveryNotificationService, + *, + email_sender: SMTPDeliverySender | None = None, + ) -> None: self._service = service + self._email_sender = email_sender @classmethod - def for_db(cls, db_path: str) -> "DeliveryDispatcher": - return cls(DeliveryNotificationService.for_db(db_path)) + def for_db( + cls, + db_path: str, + *, + email_sender: SMTPDeliverySender | None = None, + ) -> "DeliveryDispatcher": + return cls( + DeliveryNotificationService.for_db(db_path), + email_sender=email_sender, + ) def close(self) -> None: self._service.close() @@ -43,27 +59,89 @@ class DeliveryDispatcher: ) for event in events: result.processed += 1 - failure_reason = self._validate_event(event) + payload, failure_reason = self._validate_event(event) if failure_reason is not None: self._service.mark_failed( event.order_id, failure_reason, event_type=event.event_type ) result.failed += 1 continue - self._service.mark_sent(event.order_id, event_type=event.event_type) + assert payload is not None + sent_at = utc_now_iso() + try: + rendered_payload = self._deliver_event( + event, + payload=payload, + sent_at=sent_at, + ) + except Exception as exc: + self._service.mark_failed( + event.order_id, + str(exc), + event_type=event.event_type, + ) + result.failed += 1 + continue + self._service.mark_sent( + event.order_id, + event_type=event.event_type, + payload_json=json.dumps(rendered_payload, ensure_ascii=False), + sent_at=sent_at, + ) result.sent += 1 return result @staticmethod - def _validate_event(event: DeliveryNotificationEvent) -> str | None: + def _validate_event( + event: DeliveryNotificationEvent, + ) -> tuple[dict[str, object] | None, str | None]: try: payload = json.loads(event.payload_json) except json.JSONDecodeError: - return "delivery payload invalid" + return None, "delivery payload invalid" report_path = payload.get("audit_report") or payload.get("plan_file") pdf_path = payload.get("pdf_path") if not report_path or not Path(str(report_path)).is_file(): - return "delivery artifact missing" + return None, "delivery artifact missing" if not pdf_path or not Path(str(pdf_path)).is_file(): - return "delivery artifact missing" - return None + return None, "delivery artifact missing" + if event.channel == "email" and not payload.get("customer_email"): + return None, "delivery recipient missing" + return payload, None + + def _deliver_event( + self, + event: DeliveryNotificationEvent, + *, + payload: dict[str, object], + sent_at: str, + ) -> dict[str, object]: + rendered = dict(payload) + if event.channel == "station": + rendered["station_notice"] = { + "title": "报告已就绪", + "body": f"订单 {event.order_id} 的志愿报告已就绪，可在当前状态页查看在线报告并下载 PDF。", + "sent_at": sent_at, + } + return rendered + if event.channel == "email": + if self._email_sender is None: + raise ValueError("email sender not configured") + recipient = str(payload["customer_email"]) + subject = f"高考志愿报告已就绪 - {event.order_id}" + body = ( + f"您好，订单 {event.order_id} 的志愿报告已就绪。" + "请登录当前 portal 状态页查看在线报告并下载 PDF。" + ) + send_result = self._email_sender.send_report_ready( + recipient=recipient, + order_id=event.order_id, + subject=subject, + body=body, + ) + rendered["email_notice"] = { + **send_result, + "sent_at": sent_at, + } + return rendered + return rendered diff --git a/data/notifications/email_service.py b/data/notifications/email_service.py index e53e500..d11f9f6 100644 --- a/data/notifications/email_service.py +++ b/data/notifications/email_service.py @@ -93,21 +93,58 @@ class DeliveryNotificationService: def notify_report_ready( self, order_id: str, payload_json: str, channel: str = "station" + ) -> None: + self.notify_event( + order_id, + event_type="report_ready", + channel=channel, + payload_json=payload_json, + ) + + def notify_event( + self, + order_id: str, + *, + event_type: str, + channel: str, + payload_json: str, ) -> None: try: self._conn.execute( - "INSERT INTO delivery_notifications(order_id, event_type, channel, payload_json, status, attempt_count, last_attempt_at, failure_reason, created_at) VALUES (?, 'report_ready', ?, ?, 'ready', 1, ?, NULL, ?)", - (order_id, channel, payload_json, utc_now_iso(), utc_now_iso()), + "INSERT INTO delivery_notifications(order_id, event_type, channel, payload_json, status, attempt_count, last_attempt_at, failure_reason, created_at) VALUES (?, ?, ?, ?, 'ready', 1, ?, NULL, ?)", + ( + order_id, + event_type, + channel, + payload_json, + utc_now_iso(), + utc_now_iso(), + ), ) self._conn.commit() except sqlite3.IntegrityError: self._conn.rollback() - def mark_sent(self, order_id: str, event_type: str = "report_ready") -> None: - self._conn.execute( - "UPDATE delivery_notifications SET status='sent', last_attempt_at=?, failure_reason=NULL WHERE order_id=? AND event_type=?", - (utc_now_iso(), order_id, event_type), - ) + def mark_sent( + self, + order_id: str, + event_type: str = "report_ready", + *, + payload_json: str | None = None, + sent_at: str | None = None, + ) -> None: + if sent_at is None: + sent_at = utc_now_iso() + if payload_json is None: + self._conn.execute( + "UPDATE delivery_notifications SET status='sent', last_attempt_at=?, failure_reason=NULL WHERE order_id=? AND event_type=?", + (sent_at, order_id, event_type), + ) + else: + self._conn.execute( + "UPDATE delivery_notifications SET status='sent', payload_json=?, last_attempt_at=?, failure_reason=NULL WHERE order_id=? AND event_type=?", + (payload_json, sent_at, order_id, event_type), + ) self._conn.commit() def mark_failed( diff --git a/data/notifications/smtp_sender.py b/data/notifications/smtp_sender.py new file mode 100644 index 0000000..5b3893f --- /dev/null +++ b/data/notifications/smtp_sender.py @@ -0,0 +1,64 @@ +from __future__ import annotations + +import smtplib +from dataclasses import dataclass +from email.message import EmailMessage + + +@dataclass(frozen=True) +class SMTPSettings: + host: str + port: int + sender: str + username: str = "" + password: str = "" + use_tls: bool = False + use_ssl: bool = False + + +class SMTPDeliverySender: + def __init__(self, settings: SMTPSettings) -> None: + self._settings = settings + + def send_report_ready( + self, + *, + recipient: str, + order_id: str, + subject: str, + body: str, + ) -> dict[str, str]: + if not self._settings.host: + raise ValueError("smtp host not configured") + if not self._settings.sender: + raise ValueError("smtp sender not configured") + message = EmailMessage() + message["From"] = self._settings.sender + message["To"] = recipient + message["Subject"] = subject + message.set_content(body) + + if self._settings.use_ssl: + client: smtplib.SMTP = smtplib.SMTP_SSL( + self._settings.host, + self._settings.port, + timeout=10, + ) + else: + client = smtplib.SMTP( + self._settings.host, + self._settings.port, + timeout=10, + ) + with client: + if self._settings.use_tls: + client.starttls() + if self._settings.username: + client.login(self._settings.username, self._settings.password) + client.send_message(message) + return { + "recipient": recipient, + "order_id": order_id, + "subject": subject, + "body": body, + } diff --git a/data/orders/dao.py b/data/orders/dao.py index 83dce27..8a616d1 100644 --- a/data/orders/dao.py +++ b/data/orders/dao.py @@ -118,6 +118,7 @@ _WRITABLE_COLUMNS: tuple[str, ...] = ( "customer_phone_enc", "customer_phone_hash", "customer_wechat", + "customer_email", "candidate_name", "candidate_id_card_enc", "candidate_province", @@ -352,7 +353,7 @@ class OrdersDAO: ) -> Order: """按主键更新订单业务字段（非 status 字段）。 - 适用字段：customer_name / customer_wechat / candidate_name / + 适用字段：customer_name / customer_wechat / customer_email / candidate_name / candidate_province / candidate_score / candidate_rank / candidate_subjects / candidate_interests / candidate_strong_subjects / candidate_weak_subjects / candidate_family / assigned_consultant / @@ -456,6 +457,7 @@ class OrdersDAO: customer_name=original.customer_name, customer_phone=original.customer_phone, customer_wechat=original.customer_wechat, + customer_email=original.customer_email, candidate_name=original.candidate_name, candidate_id_card=original.candidate_id_card, candidate_province=original.candidate_province, @@ -581,18 +583,32 @@ class OrdersDAO: and order.pdf_path ): notifier = DeliveryNotificationService.from_connection(self._conn) - notifier.notify_report_ready( - order_id, - json.dumps( - { - "order_id": order_id, - "audit_report": order.audit_report, - "plan_file": order.plan_file, - "pdf_path": order.pdf_path, - }, - ensure_ascii=False, - ), + payload = json.dumps( + { + "order_id": order_id, + "audit_report": order.audit_report, + "plan_file": order.plan_file, + "pdf_path": order.pdf_path, + }, + ensure_ascii=False, ) + notifier.notify_report_ready(order_id, payload) + if order.customer_email: + notifier.notify_event( + order_id, + event_type="report_ready_email", + channel="email", + payload_json=json.dumps( + { + "order_id": order_id, + "audit_report": order.audit_report, + "plan_file": order.plan_file, + "pdf_path": order.pdf_path, + "customer_email": order.customer_email, + }, + ensure_ascii=False, + ), + ) return order def _insert_status_history( diff --git a/data/orders/deletion_service.py b/data/orders/deletion_service.py index 042f4b0..1d6a4a8 100644 --- a/data/orders/deletion_service.py +++ b/data/orders/deletion_service.py @@ -84,6 +84,7 @@ class OrderDeletionService: customer_phone_enc=NULL, customer_phone_hash=NULL, customer_wechat=NULL, + customer_email=NULL, candidate_name=?, candidate_id_card_enc=NULL, candidate_interests=NULL, @@ -102,6 +103,16 @@ class OrderDeletionService: order_id, ), ) + if self._table_exists("payments"): + self._conn.execute( + "UPDATE payments SET checkout_token=NULL, callback_payload=NULL WHERE order_id=?", + (order_id,), + ) + if self._table_exists("order_intakes"): + self._conn.execute( + "UPDATE order_intakes SET payload_json='{}', updated_at=? WHERE order_id=?", + (now, order_id), + ) self._insert_audit( order_id=order_id, action="anonymize", @@ -135,6 +146,13 @@ class OrderDeletionService: (order_id, action, actor, reason, files_deleted, utc_now_iso()), ) + def _table_exists(self, table_name: str) -> bool: + row = self._conn.execute( + "SELECT 1 FROM sqlite_master WHERE type='table' AND name=? LIMIT 1", + (table_name,), + ).fetchone() + return row is not None + @staticmethod def _delete_artifacts(*paths: str | None) -> int: deleted = 0 diff --git a/data/orders/models.py b/data/orders/models.py index 7ca8d81..1c3f7d2 100644 --- a/data/orders/models.py +++ b/data/orders/models.py @@ -59,6 +59,7 @@ class Order: customer_phone: Optional[str] = None # 明文（API 入口接收） customer_phone_hash: Optional[str] = None # 自动派生 customer_wechat: Optional[str] = None + customer_email: Optional[str] = None # 考生 candidate_name: Optional[str] = None diff --git a/data/orders/public_flow.py b/data/orders/public_flow.py index 40a8c7f..70ebd09 100644 --- a/data/orders/public_flow.py +++ b/data/orders/public_flow.py @@ -29,6 +29,7 @@ class PublicOrderCreate(BaseModel): customer_name: str = Field(min_length=1) customer_phone: Optional[str] = None customer_wechat: Optional[str] = None + customer_email: Optional[str] = None candidate_name: Optional[str] = None candidate_province: str = Field(min_length=1) notes: Optional[str] = None @@ -62,6 +63,7 @@ def create_public_order(dao: OrdersDAO, request: PublicOrderCreate) -> Order: customer_name=request.customer_name, customer_phone=request.customer_phone, customer_wechat=request.customer_wechat, + customer_email=request.customer_email, candidate_name=request.candidate_name, candidate_province=request.candidate_province, notes=request.notes, diff --git a/data/orders/schema.py b/data/orders/schema.py index 169fe75..9d04435 100644 --- a/data/orders/schema.py +++ b/data/orders/schema.py @@ -28,6 +28,7 @@ CREATE TABLE IF NOT EXISTS orders ( customer_phone_enc TEXT, customer_phone_hash TEXT, customer_wechat TEXT, + customer_email TEXT, -- 考生信息 candidate_name TEXT, @@ -100,6 +101,11 @@ def apply_schema(db_path: str | Path) -> sqlite3.Connection: try: conn.execute("PRAGMA foreign_keys = ON") conn.executescript(SCHEMA_SQL) + columns = { + row[1] for row in conn.execute("PRAGMA table_info(orders)").fetchall() + } + if "customer_email" not in columns: + conn.execute("ALTER TABLE orders ADD COLUMN customer_email TEXT") conn.commit() except Exception: conn.close() diff --git a/data/orders/tests/test_models.py b/data/orders/tests/test_models.py index 4328eab..4991b5f 100644 --- a/data/orders/tests/test_models.py +++ b/data/orders/tests/test_models.py @@ -32,6 +32,7 @@ def test_order_auto_derives_phone_hash(): service_version="basic", amount_cents=1000, customer_phone="13800001234", + customer_email="parent@example.com", ) assert order.customer_phone_hash == hash_for_index("13800001234") @@ -43,6 +44,7 @@ def test_order_to_db_row_encrypts_phone(): service_version="basic", amount_cents=1000, customer_phone="13800001234", + customer_email="parent@example.com", ) row = order.to_db_row() # 明文不出现在 DB 行 @@ -52,6 +54,7 @@ def test_order_to_db_row_encrypts_phone(): assert row["customer_phone_enc"] != "13800001234" # hash 字段保留 assert row["customer_phone_hash"] == hash_for_index("13800001234") + assert row["customer_email"] == "parent@example.com" # 密文可解 assert decrypt(row["customer_phone_enc"]) == "13800001234" diff --git a/data/payments/providers/alipay.py b/data/payments/providers/alipay.py index 341be24..8e40ddc 100644 --- a/data/payments/providers/alipay.py +++ b/data/payments/providers/alipay.py @@ -87,6 +87,8 @@ class AlipayProvider: provider_trade_no: str, ) -> tuple[dict[str, Any], str]: payload = { + "app_id": self.app_id, + "notify_id": f"notify_{payment_id}", "out_trade_no": payment_id, "trade_no": provider_trade_no, "total_amount": self._format_amount(amount_cents), @@ -107,6 +109,8 @@ class AlipayProvider: "amount_cents": amount_cents, "provider_trade_no": str(payload.get("trade_no") or ""), "status": str(payload.get("trade_status") or "TRADE_SUCCESS"), + "app_id": str(payload.get("app_id") or ""), + "notify_id": str(payload.get("notify_id") or ""), } def sign_payload(self, payload: dict[str, Any]) -> str: diff --git a/data/payments/tests/test_provider_alipay.py b/data/payments/tests/test_provider_alipay.py index 2c3b936..3288c96 100644 --- a/data/payments/tests/test_provider_alipay.py +++ b/data/payments/tests/test_provider_alipay.py @@ -90,4 +90,6 @@ def test_alipay_provider_verifies_signed_webhook_payload(tmp_path): "amount_cents": 9900, "provider_trade_no": "ALI-TRADE-001", "status": "TRADE_SUCCESS", + "app_id": "20260001", + "notify_id": "notify_pay_123", } diff --git a/data/payments/tests/test_webhook.py b/data/payments/tests/test_webhook.py index fd44a43..b43e426 100644 --- a/data/payments/tests/test_webhook.py +++ b/data/payments/tests/test_webhook.py @@ -1,8 +1,12 @@ from __future__ import annotations +from typing import Any, cast + +import pytest + from data.orders.dao import OrdersDAO from data.orders.models import Order -from data.payments.service import PaymentService +from data.payments.service import PaymentError, PaymentService def _seed_order(db_path: str, order_id: str = "GKO-20260614-WEBHOOK") -> Order: @@ -78,3 +82,84 @@ def test_mock_payment_webhook_rejects_amount_mismatch(client, settings): with OrdersDAO.connect(settings.orders_db_path) as dao: unchanged = dao.get(order.id) assert unchanged.status == "pending" + + +def test_handle_webhook_rejects_non_success_status(settings): + order = _seed_order(settings.orders_db_path, order_id="GKO-20260614-WEBHOOK-STATUS") + service = PaymentService.for_db( + settings.orders_db_path, + base_url=settings.payment_base_url, + webhook_secret=settings.payment_webhook_secret, + ) + checkout = service.create_checkout(order.id, portal_token="portal-token") + payload, headers = service.provider.build_webhook_request( + payment_id=checkout.payment_id, + amount_cents=order.amount_cents, + provider_trade_no="MOCK-WAITING-001", + ) + payload = cast(dict[str, Any], payload) + headers = cast(dict[str, str], headers) + payload["status"] = "failed" + headers["X-Mock-Signature"] = service.provider.sign_payload(payload) + + with pytest.raises(PaymentError, match="payment status not successful"): + service.handle_webhook(payload, headers["X-Mock-Signature"]) + + with OrdersDAO.connect(settings.orders_db_path) as dao: + unchanged = dao.get(order.id) + assert unchanged.status == "pending" + + +def test_handle_webhook_rejects_app_id_mismatch(settings): + order = _seed_order(settings.orders_db_path, order_id="GKO-20260614-WEBHOOK-APPID") + service = PaymentService.for_db( + settings.orders_db_path, + base_url=settings.payment_base_url, + webhook_secret=settings.payment_webhook_secret, + ) + setattr(service.provider, "app_id", "expected-app-id") + checkout = service.create_checkout(order.id, portal_token="portal-token") + payload, headers = service.provider.build_webhook_request( + payment_id=checkout.payment_id, + amount_cents=order.amount_cents, + provider_trade_no="MOCK-APPID-001", + ) + payload = cast(dict[str, Any], payload) + headers = cast(dict[str, str], headers) + payload["app_id"] = "wrong-app-id" + payload["notify_id"] = "notify-001" + headers["X-Mock-Signature"] = service.provider.sign_payload(payload) + + with pytest.raises(PaymentError, match="payment app_id mismatch"): + service.handle_webhook(payload, headers["X-Mock-Signature"]) + + with OrdersDAO.connect(settings.orders_db_path) as dao: + unchanged = dao.get(order.id) + assert unchanged.status == "pending" + + +def test_handle_webhook_rejects_missing_notify_id_for_bound_provider(settings): + order = _seed_order(settings.orders_db_path, order_id="GKO-20260614-WEBHOOK-NOTIFY") + service = PaymentService.for_db( + settings.orders_db_path, + base_url=settings.payment_base_url, + webhook_secret=settings.payment_webhook_secret, + ) + setattr(service.provider, "app_id", "expected-app-id") + checkout = service.create_checkout(order.id, portal_token="portal-token") + payload, headers = service.provider.build_webhook_request( + payment_id=checkout.payment_id, + amount_cents=order.amount_cents, + provider_trade_no="MOCK-NOTIFY-001", + ) + payload = cast(dict[str, Any], payload) + headers = cast(dict[str, str], headers) + payload["app_id"] = "expected-app-id" + headers["X-Mock-Signature"] = service.provider.sign_payload(payload) + + with pytest.raises(PaymentError, match="payment notify_id missing"): + service.handle_webhook(payload, headers["X-Mock-Signature"]) + + with OrdersDAO.connect(settings.orders_db_path) as dao: + unchanged = dao.get(order.id) + assert unchanged.status == "pending" diff --git a/data/share/permission.py b/data/share/permission.py index f7eb513..f6cb270 100644 --- a/data/share/permission.py +++ b/data/share/permission.py @@ -41,10 +41,46 @@ POLICY_ADMIN_ALIAS = "edit" # key: permission 等级 # value: dict(可见字段集合, 允许能力集合, 是否脱敏姓名) # -# 字段集合 None = "全部可见 (除内部字段外默认通过)"; -# iterable = "只允许 iterable 列出的字段被前端渲染"。 -# 这样未来加新 PII (电话 / 身份证) 时, 只要在 read/comment 行内把它们 -# 排除即可, 不必改 ShareLinkService。 +# 字段集合 iterable = "只允许 iterable 列出的字段被前端渲染"。 +# P1-6 起不再允许 edit/admin 走 None 透传,避免未来新增敏感字段自动外泄。 + +_EDIT_VISIBLE_FIELDS = { + "report_id", + "permission", + "owner_id", + "share_url", + "created_at_iso", + "expires_at_iso", + "candidate_name", + "customer_name", + "student_name", + "name", + "title", + "summary", + "recommendations", + "volunteers", + "score", + "rank", + "year", + "province", + "candidate_phone", + "customer_phone", + "customer_email", + "customer_wechat", + "candidate_id_card", + "candidate_subjects", + "candidate_interests", + "candidate_strong_subjects", + "candidate_weak_subjects", + "candidate_family", + "assigned_consultant", + "service_version", + "plan_file", + "audit_report", + "pdf_path", + "notes", + "tags", +} _POLICY_TABLE: dict[str, dict[str, Any]] = { "read": { @@ -99,8 +135,8 @@ _POLICY_TABLE: dict[str, dict[str, Any]] = { "can_comment": True, "can_edit": True, "mask_name": False, - # edit 允许业务字段透传，但内部敏感字段仍由 _ALWAYS_HIDDEN_FIELDS 拦截。 - "visible_fields": None, + # P1-6: edit 必须显式 allowlist，不能再走 None 透传。 + "visible_fields": _EDIT_VISIBLE_FIELDS, }, "admin": { # alias -> edit @@ -108,7 +144,7 @@ _POLICY_TABLE: dict[str, dict[str, Any]] = { "can_comment": True, "can_edit": True, "mask_name": False, - "visible_fields": None, + "visible_fields": _EDIT_VISIBLE_FIELDS, }, } @@ -126,6 +162,9 @@ _PII_NAME_FIELDS = ("candidate_name", "customer_name", "student_name", "name") _ALWAYS_HIDDEN_FIELDS = frozenset( { "password_hash", + "api_token", + "access_token", + "refresh_token", "internal_note", "note", "debug_info", diff --git a/data/share/tests/test_permission.py b/data/share/tests/test_permission.py index 1194e19..21c3d67 100644 --- a/data/share/tests/test_permission.py +++ b/data/share/tests/test_permission.py @@ -135,7 +135,7 @@ def test_policy_edit_caps(): _eq(p.can_comment, True, "edit can comment") _eq(p.can_edit, True, "edit can edit") _eq(p.mask_name, False, "edit does NOT mask name") - _eq(p.visible_fields, None, "edit visible_fields=None (default pass)") + _truthy(p.visible_fields is not None, "edit uses explicit visible_fields allowlist") def test_policy_admin_alias_to_edit(): @@ -189,8 +189,8 @@ def test_allows_field_for_edit(): _eq(p.allows_field("candidate_phone"), True, "edit allows phone") _eq( p.allows_field("password_hash"), - True, - "edit: visible_fields=None -> pass (上层负责拦截 hash)", + False, + "edit: password_hash 不在 allowlist 且仍受内部字段规则保护", ) @@ -316,7 +316,7 @@ def test_render_edit_name_unmasked_in_payload(): def test_render_admin_aliases_to_edit(): - """admin 视为 edit, 姓名不脱敏, 字段全通过。""" + """admin 视为 edit, 姓名不脱敏, 但字段仍受显式 allowlist 约束。""" out = render_report_payload("admin", _SAMPLE_REPORT) _eq(out["permission"], "edit", "admin perm normalized to edit in payload") _eq(out["payload"]["candidate_name"], "李明", "admin: name intact") @@ -390,8 +390,155 @@ def test_render_visible_fields_echo(): def test_render_edit_visible_fields_none(): + """P1-6 修复后: edit 必须使用显式 allowlist, 不再返回 None pass-through. + + 旧契约 (visible_fields=None) 会让未来新增的敏感字段自动外泄, + 这里改为 RED 断言 visible_fields 为非 None 且为显式集合. + """ out = render_report_payload("edit", _SAMPLE_REPORT) - _eq(out["visible_fields"], None, "edit: visible_fields=None means pass-through") + _truthy( + out["visible_fields"] is not None, + "edit: visible_fields must be explicit allowlist, not None", + ) + + +def test_render_admin_visible_fields_none(): + """P1-6: admin (alias->edit) 也必须使用显式 allowlist.""" + out = render_report_payload("admin", _SAMPLE_REPORT) + _truthy( + out["visible_fields"] is not None, + "admin: visible_fields must be explicit allowlist, not None", + ) + + +# --------------------------------------------------------------------------- +# P1-6 回归测试: 新增敏感字段不应自动外泄 +# --------------------------------------------------------------------------- + + +def test_policy_edit_uses_explicit_allowlist(): + """P1-6: edit.permission 策略必须返回非 None 的显式 allowlist.""" + p = PermissionPolicy.for_permission("edit") + _truthy( + p.visible_fields is not None, + "edit: visible_fields must be explicit frozenset, not None pass-through", + ) + + +def test_policy_admin_uses_explicit_allowlist(): + """P1-6: admin (alias->edit) 同样必须返回非 None allowlist.""" + p = PermissionPolicy.for_permission("admin") + _truthy( + p.visible_fields is not None, + "admin: visible_fields must be explicit frozenset, not None pass-through", + ) + + +def test_edit_does_not_leak_new_sensitive_field(): + """P1-6 核心回归: edit 模式下, payload 里塞入一个全新的敏感字段 (不在任何 + allowlist 中) 不应自动出现在对外 payload 中. + + 这条用例模拟"未来某个 commit 在 report 里新增了 candidate_bank_card / + api_token / home_address 等敏感字段, edit/admin 是否会无声地把它外暴". + """ + report = { + "report_id": "R-NEW", + "candidate_name": "李明", + "candidate_bank_card": "6222020200001234567", # 新增敏感字段 + "candidate_home_address": "长沙市岳麓区某街道 88 号", # 新增敏感字段 + "api_token": "sk_live_secret_abcdef", # 新增敏感字段 + } + for perm in ("edit", "admin"): + out = render_report_payload(perm, report) + payload = out["payload"] + _eq( + "candidate_bank_card" in payload, + False, + f"{perm}: 新增敏感字段 candidate_bank_card 不应自动外泄", + ) + _eq( + "candidate_home_address" in payload, + False, + f"{perm}: 新增敏感字段 candidate_home_address 不应自动外泄", + ) + _eq( + "api_token" in payload, + False, + f"{perm}: 新增敏感字段 api_token 不应自动外泄", + ) + + +def test_edit_allows_field_rejects_unknown_field(): + """P1-6: edit 的 allows_field 必须对未知字段返回 False (allowlist 语义).""" + p = PermissionPolicy.for_permission("edit") + # 已知业务字段仍然允许 + _eq(p.allows_field("report_id"), True, "edit: report_id allowed") + _eq(p.allows_field("recommendations"), True, "edit: recommendations allowed") + # 未来新增的、显式 allowlist 中不存在的字段必须被拒绝 + _eq( + p.allows_field("candidate_bank_card"), + False, + "edit: 新增敏感字段 candidate_bank_card 被显式 allowlist 拒绝", + ) + _eq( + p.allows_field("api_token"), + False, + "edit: 新增敏感字段 api_token 被显式 allowlist 拒绝", + ) + + +def test_admin_allows_field_rejects_unknown_field(): + """P1-6: admin (alias->edit) 的 allows_field 同样拒绝未知字段.""" + p = PermissionPolicy.for_permission("admin") + _eq( + p.allows_field("candidate_bank_card"), + False, + "admin: 新增敏感字段 candidate_bank_card 被显式 allowlist 拒绝", + ) + + +def test_edit_still_hides_internal_fields(): + """P1-6 修复后, edit/admin 的隐式内部敏感字段 (password_hash 等) 仍必须被拦截. + + 切换为 allowlist 后, 这条由 _ALWAYS_HIDDEN_FIELDS + 显式 allowlist 共同保证: + 即便未来有人在 allowlist 里不小心漏掉 password_hash, 也仍被黑名单兜底. + """ + report = { + "report_id": "R-INT", + "candidate_name": "李明", + "password_hash": "hashed-secret", + "internal_note": "ops only", + "raw_payload": {"deep": "data"}, + } + for perm in ("edit", "admin"): + out = render_report_payload(perm, report) + payload = out["payload"] + _eq( + "password_hash" in payload, + False, + f"{perm}: password_hash 仍必须被拦截", + ) + _eq( + "internal_note" in payload, + False, + f"{perm}: internal_note 仍必须被拦截", + ) + _eq( + "raw_payload" in payload, + False, + f"{perm}: raw_payload 仍必须被拦截", + ) + + +def test_edit_allowlist_superset_of_comment(): + """P1-6: edit 的 allowlist 必须是 comment allowlist 的超集, 保证业务字段透传.""" + edit_p = PermissionPolicy.for_permission("edit") + comment_p = PermissionPolicy.for_permission("comment") + _eq( + comment_p.visible_fields.issubset(edit_p.visible_fields), + True, + "edit allowlist 必须是 comment allowlist 的超集", + ) # --------------------------------------------------------------------------- diff --git a/docs/ACTIVE_REMEDIATION_2026-06-13.md b/docs/ACTIVE_REMEDIATION_2026-06-13.md index 90c1944..df505c0 100644 --- a/docs/ACTIVE_REMEDIATION_2026-06-13.md +++ b/docs/ACTIVE_REMEDIATION_2026-06-13.md @@ -1,6 +1,6 @@ # ACTIVE_REMEDIATION -最后更新: 2026-06-13 +最后更新: 2026-06-15 状态词: 当前仍然有效的问题清单（仅保留未解决项） 真相源: `docs/CURRENT_STATE.md` 来源基线: @@ -108,12 +108,13 @@ - 站内查看 + PDF 下载最小交付闭环 - `DeliveryDispatcher` + `scripts/gaokao-delivery-dispatch.py` - `ready -> sent` / 缺文件 `-> failed` 的最小执行链 +- 站内通知发送器：dispatch 时真实生成 `station_notice` 内容，并在 portal 状态页向用户展示“通知已发送” +- 真实邮件通知发送器：`SMTPDeliverySender` + `customer_email` 采集链 + `email` 通道 dispatch（本地 stub SMTP 已验证） - dispatcher/watchdog runbook + systemd/cron 样例（`docs/DELIVERY_RETENTION_OPS_RUNBOOK.md` / `deploy/systemd` / `deploy/cron`） 仍缺： -- 真实邮件/站内通知发送执行器 -- 自动重试调度 / 告警链（watchdog/systemd/cron 样例已补，生产安装未完成） +- 自动重试调度 / 生产告警链（watchdog/systemd/cron 样例已补，生产安装未完成） - 面向用户的独立通知审计页 - 对账/退款与交付失败补偿联动 @@ -206,25 +207,25 @@ - `data.crowd_db.quality_summary` 已补 province-level summary，后续可接产品文案/页面 - 继续提升非湖南省份高置信数据密度 -#### B-2 本地验证环境仍未完全固化为单命令体验 +#### B-2 本地验证环境已基本统一，仍待生产 runner 长期观察 严重度: P1 -当前状态: 部分解决，仍待完善 +当前状态: 已解决（本地/CI 统一脚本门禁已落地） 已解决： - requirements-admin / requirements-dev 已修正 -- CI 覆盖率门禁已对齐真实阈值 +- `scripts/dev-verify.sh` 已成为本地统一入口 +- `.github/workflows/ci.yml` 已统一走 `scripts/dev-verify.sh` +- `scripts/check_coverage_gate.py` 已纳入 dev-verify，core coverage gate 与本地/CI 对齐 仍缺： -- `make verify` / `scripts/dev-verify.sh` 统一入口 -- 本地一键创建 venv + 安装依赖 + 运行门禁 +- 生产 runner / Python 3.12 长期观察（当前已通过 skip 兼容 SMTP stdlib 移除） 建议动作： -- 新建 `scripts/dev-verify.sh` -- README 增加标准开发环境初始化命令 +- 后续若迁移 Python 3.12+，将 SMTP stub 从 stdlib `smtpd/asyncore` 迁移到 `aiosmtpd` #### B-3 历史评审文档仍可能被误读为当前阻塞 @@ -236,16 +237,15 @@ - 对齐评审报告已加历史快照说明 - `CURRENT_STATE.md` 已建立 +已解决： + +- `docs/AUDIT_REPORT_2026-06-11.md` 已标注历史快照 +- `docs/REMEDIATION_TASK_BOARD_2026-06-11.md` 已标注历史快照 +- `reports/PROJECT_SYSTEM_REVIEW_2026-06-13.md` 已标注历史快照与真相源跳转 + 仍缺： -- 其他历史文档（尤其旧 audit / remediation / review 报告）未全部统一加“历史快照”页眉 - -建议动作： - -- 对 `docs/AUDIT_REPORT_2026-06-11.md` -- 对 `docs/REMEDIATION_TASK_BOARD_2026-06-11.md` -- 对 `reports/PROJECT_SYSTEM_REVIEW_2026-06-13.md` - 补统一快照声明与真相源跳转 +- 无 --- diff --git a/docs/CURRENT_STATE.md b/docs/CURRENT_STATE.md index a18bab2..d4b1c4c 100644 --- a/docs/CURRENT_STATE.md +++ b/docs/CURRENT_STATE.md @@ -1,7 +1,7 @@ # CURRENT_STATE -最后更新: 2026-06-13 -状态词: 完成（v2.1 当前实现闭环） +最后更新: 2026-06-15 +状态词: 本地验证完成（v2.1 已交付增强 + T12 本地修复闭环已收口，待 commit/push） 真相源优先级: 1. 本文件 @@ -19,7 +19,7 @@ - 已完成: 人工服务运营增强系统（闲鱼 / 微信 / 学校渠道） - 已完成: 管理后台、订单、分享、渠道同步、AI 审核、CI/CD、性能与安全加固 - 未完成: 用户端 Web 自助支付 / 资料填写 / 站内交付的完整商业闭环（T12 进行中） -- 已完成但仅限本地验证: `mock` / `alipay_sim` / `alipay` 三层支付代码链与 notify/return 路由 +- 已完成且已本地验证: `mock` / `alipay_sim` / `alipay` 三层支付代码链与 notify/return 路由；退款状态闭环、portal token secret 分离、payment webhook fail-closed、删除/匿名化扩围、分享 allowlist、channel_sync DB 隔离与 DAO 真相收敛 - 未完成且仍阻塞线上验收: 真实支付宝商户凭据、公网 notify_url、备案域名、真实支付 acceptance 一句话： @@ -70,7 +70,7 @@ - 真实支付接入与回调验签的线上 acceptance - `info_submitted -> serving` 自动处理主链 +本地自动桥接 `info_submitted -> serving` 已落地；仍缺真实 worker/生成服务/线上调度收口 -- 完整交付通知/发送器与调度 +- 站内通知发送器、真实邮件通知发送器与调度已落地；仍缺生产告警链与独立通知审计页 - 前台删除工单流程与更完整的产品化交互 因此： @@ -105,10 +105,10 @@ ### P0 / P1 真实缺口 - 用户端 Web 自助支付闭环缺失 -- 邮件 / 站内自动交付尚未形成当前主系统标准能力 +- 生产告警链与独立通知审计页仍未形成当前主系统标准能力（站内通知 + 邮件通知发送器已落地） - 上传入口仍以后台 / CLI / 内部入口为主，用户前台入口不足 - 业务数据备份 / 恢复 / 密钥托管已有本地基线与验证，但异机备份和生产接入仍不足 -- 隐私政策 / 服务协议 / 监护人同意 / 数据保留与删除流程未形成正式产品闭环 +- 隐私政策 / 服务协议 / 监护人同意 / 数据保留与删除流程仍缺前台/客服自助工单与正式法务版本 ### P2 工程改进 diff --git a/docs/DELIVERY_SERVICE_DESIGN.md b/docs/DELIVERY_SERVICE_DESIGN.md index ab133b3..46706f7 100644 --- a/docs/DELIVERY_SERVICE_DESIGN.md +++ b/docs/DELIVERY_SERVICE_DESIGN.md @@ -56,6 +56,8 @@ - `scripts/gaokao-delivery-dispatch.py` - `scripts/gaokao-delivery-watchdog.py`（失败返回 exit code `2`） - `ready -> sent` / `缺文件 -> failed` / `attempt_count` 递增 +- `station` 通道 dispatch 时会真实生成 `station_notice`，portal 状态页展示“通知已发送”消息卡片 +- `email` 通道使用 `SMTPDeliverySender` 发送真实邮件通知，支持 `GAOKAO_SMTP_*` 配置，并已通过本地 stub SMTP 验证 - `docs/DELIVERY_RETENTION_OPS_RUNBOOK.md` - `deploy/systemd/gaokao-delivery-{dispatch,watchdog}.{service,timer}` - `deploy/cron/gaokao-jobs.crontab` @@ -80,10 +82,10 @@ ## 7. 下一步实施建议 -1. 增加邮件或站内通知真实发送器（二选一先闭环） -2. 增加失败重试阈值与告警推送 +1. 增加失败重试阈值与告警推送 +2. 增加面向用户的独立通知审计页 3. 把 `sent` 语义拆成“可投递校验通过”与“真实渠道已发送” -4. 再决定是否扩到多通道 +4. 再决定是否扩到更多通道 ## 8. 生产化接入口径 diff --git a/docs/PAYMENT_PROVIDER_ONBOARDING.md b/docs/PAYMENT_PROVIDER_ONBOARDING.md index b043419..180c69b 100644 --- a/docs/PAYMENT_PROVIDER_ONBOARDING.md +++ b/docs/PAYMENT_PROVIDER_ONBOARDING.md @@ -19,7 +19,7 @@ - 配置键： - `GAOKAO_PAYMENT_PROVIDER` - `GAOKAO_PAYMENT_BASE_URL` - - `GAOKAO_PAYMENT_WEBHOOK_SECRET` + - `GAOKAO_PAYMENT_WEBHOOK_SECRET` (生产环境必须显式设置强密钥,启动 fail-closed;P2-5) - `GAOKAO_PAYMENT_NOTIFY_URL` - `GAOKAO_PAYMENT_RETURN_URL` - `GAOKAO_PAYMENT_APP_ID` diff --git a/reports/PROJECT_SYSTEM_REVIEW_2026-06-14.md b/reports/PROJECT_SYSTEM_REVIEW_2026-06-14.md new file mode 100644 index 0000000..1d365fb --- /dev/null +++ b/reports/PROJECT_SYSTEM_REVIEW_2026-06-14.md @@ -0,0 +1,499 @@ +# gaokao-volunteer-system 系统性 Review 报告（2026-06-14） + +**评审对象**: `/home/long/project/gaokao-volunteer-system` +**评审方式**: 当前文档真相核对 + 关键源码结构审查 + 安全/测试/运维专项审查 + 本地门禁复核 +**评审标准**: 严格标准；以“真实业务闭环、数据安全、验证可信度、可运维性”为主轴 +**当前真相源**: `docs/CURRENT_STATE.md` + +--- + +## 1. 结论 + +**结论**：项目已经具备较完整的“运营后台 + 订单 + 支付抽象 + portal + 分享 + 渠道同步 + 交付事件”骨架，核心订单状态机与 DAO 边界也明显优于很多同规模项目；但按严格标准看，**系统仍未达到“关键链路一致性可靠、支付/删除/交付语义闭环、质量门禁可信、灾备与部署可验证”的稳态**。 + +**总体评级**：**有明显工程基础，但不建议把当前版本描述为“已完成的、可放心放量的商业闭环系统”**。 + +**最重要判断**： + +1. **订单/支付/退款/交付**仍有多处语义裂缝，主系统真相源不统一。 +2. **隐私删除/匿名化**与文档承诺不一致，存在误报“已匿名化/已删除”的合规风险。 +3. **CI / 本地 / coverage gate / Docker / 备份恢复**没有形成一致、可复现、可证明的验证链。 +4. **支付回调与分享公开面**仍有显著安全收敛空间。 + +本次未发现必须立即停机级别的 **P0** 远程利用证据，但存在多项 **P1**，足以阻止“系统已成熟闭环”的结论。 + +--- + +## 2. 评审范围 + +### 2.1 文档与约束 + +- `README.md` +- `docs/CURRENT_STATE.md` +- `docs/LEGAL_PRIVACY_BASELINE.md` +- `docs/DATA_RETENTION_AND_DELETION.md` +- `docs/BACKUP_AND_RECOVERY_PLAN.md` +- `docs/DELIVERY_SERVICE_DESIGN.md` +- `docker-compose.yml` +- `Dockerfile` +- `.github/workflows/ci.yml` +- `codecov.yml` +- `pytest.ini` +- `scripts/dev-verify.sh` +- `scripts/backup_verify.sh` +- `scripts/check_coverage_gate.py` + +### 2.2 核心代码链路 + +- `admin/app.py` +- `admin/auth.py` +- `admin/config.py` +- `admin/password.py` +- `admin/db.py` +- `admin/routes/auth.py` +- `admin/routes/orders.py` +- `admin/routes/web_public.py` +- `data/orders/dao.py` +- `data/orders/state_machine.py` +- `data/orders/public_flow.py` +- `data/orders/deletion_service.py` +- `data/orders/intake_store.py` +- `data/orders/crypto.py` +- `data/orders/masking.py` +- `data/payments/service.py` +- `data/payments/dao.py` +- `data/payments/providers/alipay.py` +- `data/payments/providers/alipay_sim.py` +- `data/channel_sync/webhook_server.py` +- `data/share/permission.py` +- `data/customer_portal/token.py` +- `data/notifications/dispatcher.py` + +### 2.3 测试与验证样本 + +- `admin/tests/test_web_public_alipay_sim_e2e.py` +- `admin/tests/test_order_status_page.py` +- `admin/tests/test_routes_orders.py` +- `data/payments/tests/test_webhook.py` +- `tests/test_delivery_dispatcher.py` +- `tests/test_retention_cleanup.py` +- `tests/test_t5_performance.py` + +--- + +## 3. 严重级别定义 + +- **P0**: 已存在可直接导致系统失控/严重数据泄漏/资金错误且缺乏现实缓解 +- **P1**: 高优先级问题；会破坏主链路一致性、支付/隐私/恢复可信度，阻止成熟上线结论 +- **P2**: 中优先级问题；不会立即击穿主链，但会持续积累运维、测试或安全债务 +- **P3**: 正向亮点或低优先改进项 + +--- + +## 4. 关键问题清单 + +## 4.1 P1 问题 + +### P1-1 支付回调存在双写裂缝：payment 已记 paid，但 order 可能未推进 + +**证据** + +- `data/payments/service.py::PaymentService.handle_webhook()` 先更新 `payments.status=paid` +- `data/payments/dao.py::PaymentDAO.update_status()` 内部立即 `commit` +- 随后才在新的 `OrdersDAO` 上下文里把订单从 `pending` 推到 `paid` + +**风险** + +这是典型跨表双写一致性问题。任一时刻如果支付表提交成功、订单推进失败（订单异常、锁冲突、状态先被别处推进、进程中断），系统会留下： + +- `payments.status = paid` +- `orders.status != paid` + +之后 portal、退款、统计、人工判断都会基于不同真相源工作。 + +**结论** + +这是当前最重要的业务一致性问题之一。 + +--- + +### P1-2 删除/匿名化与文档承诺不一致，存在“看起来已删，实际上没删”的合规风险 + +**证据** + +- `docs/DATA_RETENTION_AND_DELETION.md` 要求订单、资料提交、报告文件、通知/支付审计都应清理或匿名化 +- `data/orders/deletion_service.py::OrderDeletionService.anonymize_order()` 仅更新 `orders` 主表部分字段 +- 未覆盖： + - `data/orders/intake_store.py` 中 `order_intakes.payload_json` + - `data/payments/dao.py` 中 `payments.callback_payload` + - `audit_report` / `pdf_path` 对应文件 + +**风险** + +系统当前容易把“订单主表去标识化”误报成“订单数据已匿名化完成”。对于高考场景中的未成年人/监护人资料，这是明显不够的。 + +**结论** + +当前删除/匿名化能力不足以支撑强隐私合规表述。 + +--- + +### P1-3 真实支付宝回调校验过弱，只验签 + 金额，不校验应用/商户/状态边界 + +**证据** + +- `admin/routes/web_public.py` 的 `/api/public/payments/alipay/notify` 直接把表单解析结果交给 `PaymentService.handle_webhook()` +- `data/payments/service.py::handle_webhook()` 只校验：签名、`payment_id`、金额一致 +- `data/payments/providers/alipay.py` 仅标准化 `out_trade_no/trade_no/total_amount/trade_status` 并做 RSA 验签 +- 未见对 `app_id`、商户标识、允许状态白名单、`notify_id`、`gmt_payment` 等做二次校验 + +**风险** + +系统会把“签名有效且金额匹配”的回调近似当成“业务上确认为成功支付”。这对真实支付系统不够。 + +**结论** + +真实支付回调仍未达到严格上线标准。 + +--- + +### P1-4 Webhook server 的数据库连接缓存按模块全局单例保存，且不区分 `db_path` + +**证据** + +- `data/channel_sync/webhook_server.py` 中 `_DB_CONN` 为模块级单例 +- `_get_db(db_path)` 只在 `_DB_CONN is None` 时首次打开连接；后续不同 `db_path` 直接复用旧连接 +- `make_server(..., db_path=...)` 虽允许传不同库路径，但连接缓存不会切换 + +**风险** + +同一进程先后启动多个 webhook server / 测试实例 / 不同 DB 场景时，后者可能静默写入前一个库。问题隐蔽，影响审计、幂等和订单入库正确性。 + +**结论** + +这是高风险的状态污染问题。 + +--- + +### P1-5 退款域模型未闭环：状态机有 `refunded`，支付服务只把 payment 记到 `refund_pending` + +**证据** + +- `data/orders/state_machine.py` 支持订单进入 `refunded` 终态 +- `data/payments/service.py::request_refund()` 仅把 payment 更新到 `refund_pending` +- 未见生产调用路径把订单主状态同步推进到 `refunded` +- `admin/routes/web_public.py` portal 阶段判断优先看 payment 的退款状态，而不是订单主状态 + +**风险** + +订单域与支付域的退款真相源分裂。之后交付限制、统计、保留期清理、人工判断都可能不一致。 + +**结论** + +退款闭环目前不完整。 + +--- + +### P1-6 公开分享 `edit/admin` 权限是 allow-by-default，未来新增字段极易外泄 + +**证据** + +- `data/share/permission.py` 中 `edit/admin` 的 `visible_fields=None` +- `render_report_payload()` 在该模式下除 denylist 外默认全部透传 +- `_ALWAYS_HIDDEN_FIELDS` 仅覆盖少数内部字段，不覆盖未来可能新增的手机号、身份证、顾问备注等 + +**风险** + +一旦报告 payload 新增敏感字段，公开分享链接可能自动带出这些信息，而不需要任何显式审批。 + +**结论** + +分享策略对未来演进不够安全。 + +--- + +### P1-7 覆盖率门禁口径互相冲突，验证链本身不可信 + +**证据** + +- `.github/workflows/ci.yml`：`--cov-fail-under=60` +- `scripts/dev-verify.sh`：`--cov-fail-under=80` +- `scripts/check_coverage_gate.py`：`overall=80%`、`core=100%` +- `codecov.yml`：也声明整体 80%、核心 100% + +**风险** + +同一改动可能出现： + +- CI 通过 +- 本地失败 +- codecov 标准不同 +- gate 脚本又给出另一套结论 + +这种门禁不稳定，会持续削弱团队对测试结论的信任。 + +**结论** + +当前质量门禁存在制度级不一致。 + +--- + +### P1-8 备份恢复只做到“复制文件 + 打印表名”，还不是恢复演练 + +**证据** + +- `docs/BACKUP_AND_RECOVERY_PLAN.md` 自认仅是“最小恢复基线”，并明确尚无自动化恢复验证 +- `scripts/backup_verify.sh` 只复制 DB / 文件，然后验证 SQLite 文件可打开、文件可枚举 +- 未见 CI 或自动化测试真正执行“恢复后启动应用并验证 portal / 订单 / 报告” + +**风险** + +当前只能证明“文件能拷贝”，不能证明“系统能恢复并可用”。这对任何需要对外交付或数据保留承诺的系统都不够。 + +**结论** + +灾备能力目前停留在文档与文件级验证，不是服务级验证。 + +--- + +## 4.2 P2 问题 + +### P2-1 公共下单先写订单、后建支付，失败时会遗留孤儿 `pending` 订单 + +**证据** + +- `admin/routes/web_public.py::create_public_order_endpoint()` 先 `create_public_order()`，再 `PaymentService.create_checkout()` +- `data/orders/public_flow.py` 直接先把订单写入库 +- `data/payments/service.py::create_checkout()` 再独立创建 payment + +**风险** + +支付初始化失败时，用户请求失败，但库里已经有订单，且没有补偿动作。 + +--- + +### P2-2 渠道同步仍在走已声明被替代的 `dao_extension`，存在双实现漂移风险 + +**证据** + +- `data/orders/dao.py` 已提供 `OrdersDAO.upsert_by_external_id()` +- `data/channel_sync/webhook_server.py` 与 `data/channel_sync/poller.py` 仍直接使用 `data.channel_sync.dao_extension.upsert_by_external_id` + +**风险** + +同一业务规则同时存在两套实现，后续修改很容易只修一套。 + +--- + +### P2-3 交付执行器把“文件存在”直接标记为 `sent`，语义过度乐观 + +**证据** + +- `data/notifications/dispatcher.py::dispatch_ready_events()` 校验文件存在后直接 `mark_sent` +- `docs/DELIVERY_SERVICE_DESIGN.md` 又明确说明“邮件/渠道真实发送执行器”尚未完成 + +**风险** + +`sent` 更像“已投递成功”，但当前实际含义只是“本地文件通过检查”。这会误导运营状态与告警语义。 + +--- + +### P2-4 Portal token 与后台 JWT 共用同一根 secret + +**证据** + +- `admin/routes/web_public.py` 创建和校验 portal token 时使用 `settings.jwt_secret` +- `data/customer_portal/token.py` 是独立 HMAC token 格式 +- `admin/auth.py` 后台 JWT 也使用同一 secret + +**风险** + +轮换、隔离、泄露影响面都被耦合，不符合边界最小化原则。 + +--- + +### P2-5 payment webhook secret 仍有固定开发默认值，启动阶段也没有 prod fail-closed 校验 + +**证据** + +- `admin/config.py` 默认 `GAOKAO_PAYMENT_WEBHOOK_SECRET=dev-mock-payment-secret` +- `data/payments/service.py::for_db()` 也带同样默认值 +- `admin/app.py` 启动只校验 JWT secret 与默认管理员密码，没有校验 payment webhook secret + +**风险** + +在 mock / alipay_sim 场景中，如果配置漂移到公网环境，伪造回调成本会明显降低。 + +--- + +### P2-6 pytest 把 `data/` 直接纳入测试发现根，测试边界偏混杂 + +**证据** + +- `pytest.ini`: `testpaths = admin/tests tests data` +- CI 也直接跑 `pytest admin/tests tests data` + +**风险** + +业务代码目录与测试入口绑定过紧，不利于区分 unit / integration / e2e / script verification 的语义。 + +--- + +### P2-7 部署链只有单容器存活探测，没有业务冒烟、恢复验证、监控告警闭环 + +**证据** + +- `docker-compose.yml` 仅定义一个服务、一个数据卷、一个 `/health` healthcheck +- `Dockerfile` 仅安装运行依赖并直接启动 `admin.app` +- 文档里有监控设计，但未见对应自动化验证或可运行集成 + +**风险** + +当前最多只能说明“进程活着”，不能说明“关键业务链在部署后可用”。 + +--- + +## 4.3 P3 亮点 + +### P3-1 OrdersDAO 把状态机、时间戳、历史记录、交付触发收敛到一条主写路径 + +**证据** + +- `data/orders/dao.py::transition_status()` 单事务处理：状态校验、状态更新、history、时间戳 +- `delivered` 且交付物齐备时会顺带落 `delivery_notifications` + +**评价** + +这是当前仓库最稳的边界之一。它明显优于把状态历史散落在各个 route/service 中维护。 + +--- + +### P3-2 FastAPI app 装配较整洁，配置校验与 schema 初始化集中在启动期 + +**证据** + +- `admin/app.py::create_app()` 结构清晰 +- `_validate_and_log_settings()` 对 JWT / 默认管理员密码做环境分级处理 +- `_setup_database()` 统一做 schema/bootstrap + +**评价** + +应用装配层没有明显失控，是后续继续治理的好基础。 + +--- + +### P3-3 管理员密码与分享密码使用 PBKDF2 + 随机盐 + 恒时比较 + +**证据** + +- `admin/password.py` +- `data/share/short_link.py` + +**评价** + +密码存储方案比很多轻量项目更稳，属于正向设计。 + +--- + +### P3-4 登录失败统一口径，且已有基础限流 + +**证据** + +- `admin/routes/auth.py` 以 `username@client_ip` 做失败计数与 `retry_after` +- 普通错误统一返回 `AUTH_INVALID_CREDENTIALS` + +**评价** + +已具备最低限度的暴力破解缓解与账户枚举抑制。 + +--- + +## 5. 验证与复核结果 + +### 5.1 命令复核 + +在仓库根执行了以下检查： + +```bash +python3 -m pytest admin/tests tests data -q +python3 -m ruff check . --exclude .venv,.worktrees +python3 -m mypy . +python3 scripts/check_coverage_gate.py coverage.xml + +./.venv/bin/python -m pytest admin/tests tests data -q +./.venv/bin/python -m ruff check . --exclude .venv,.worktrees +./.venv/bin/python -m mypy . +./.venv/bin/python scripts/check_coverage_gate.py coverage.xml +``` + +### 5.2 观察到的结果 + +1. **系统 Python 环境**缺少 `pytest / ruff / mypy`，全局命令不可直接复现。 +2. **`.venv` 环境下**： + - `pytest`: **639 passed, 5 failed** + - 失败主要来自： + - `tests/test_delivery_dispatcher.py` + - `tests/test_retention_cleanup.py` + - `tests/test_t5_performance.py` +3. 失败原因并非随机： + - 若测试通过 `subprocess.run(["python3", ...])` 启动脚本，脚本会落到系统 Python，随后因 `admin.__init__ -> admin.app -> import uvicorn` 报 `ModuleNotFoundError: uvicorn` + - `locust` 可执行文件缺失，导致性能测试无法启动 +4. `ruff check`：**通过** +5. `mypy`：未见硬错误输出，但存在 **8 条 `annotation-unchecked` 提示**，集中在测试文件 +6. `scripts/check_coverage_gate.py coverage.xml`：**失败**，报 `missing core coverage entries`，说明当前 coverage artifact 与 gate 脚本的核心文件映射未对齐。 + +### 5.3 这组结果说明什么 + +- 项目不是“完全不可测”，因为大部分测试可在 `.venv` 中通过。 +- 但它也**不是稳定可复现的验证闭环**： + - 脚本子进程环境与测试 runner 环境脱节 + - 性能验证依赖工具未稳定声明/注入 + - coverage gate 与覆盖率产物不一致 + +这与前述 P1/P2 结论一致：**系统已有工程基础，但门禁可信度仍不够强。** + +--- + +## 6. 优先级建议 + +## 第一优先级（立即处理） + +1. **修复支付 webhook 的双写一致性问题** +2. **补齐删除/匿名化的真实覆盖范围，统一 delete vs anonymize 语义** +3. **补强真实支付宝回调业务校验** +4. **修复 webhook server 的全局 DB 连接污染问题** +5. **统一 coverage / CI / 本地 gate 口径** + +## 第二优先级（短期完成） + +6. **完成退款主状态闭环** +7. **把分享权限改为 allowlist，而不是 edit/admin 全量透传** +8. **拆分 portal secret 与 admin JWT secret** +9. **为 payment webhook secret 增加生产环境启动门禁** +10. **把渠道同步彻底 cutover 到 `OrdersDAO.upsert_by_external_id()`** + +## 第三优先级（中期治理） + +11. **把交付状态拆成“已验证可投递”与“真实已发送”** +12. **补真实 restore drill / Docker 业务冒烟 / 监控告警最小闭环** +13. **收紧 pytest 入口，明确 unit/integration/e2e 语义边界** + +--- + +## 7. 最终判断 + +如果问题是： + +> 这个仓库有没有认真做工程？ + +答案是：**有，而且部分核心边界（尤其 orders DAO / 状态机）做得不错。** + +如果问题是： + +> 这个系统是否已经达到“关键闭环可信、合规表述稳妥、验证链稳定、运维可托底”的严格标准？ + +答案是：**还没有。** + +更准确的表述应当是： + +> 这是一个已经具备明显产品化骨架、但仍处于“关键一致性 / 支付安全 / 删除合规 / 验证闭环 / 灾备落地”强化期的系统。当前适合继续收敛与修复，不适合对外宣称为完全成熟闭环。 \ No newline at end of file diff --git a/scripts/backup_restore_smoke.py b/scripts/backup_restore_smoke.py index 8247d2b..c98ad00 100644 --- a/scripts/backup_restore_smoke.py +++ b/scripts/backup_restore_smoke.py @@ -167,7 +167,7 @@ def run_restore_smoke(backup_dir: str | Path) -> dict[str, object]: reason="report_ready", ) - token = issue_portal_token(order.id, settings.jwt_secret) + token = issue_portal_token(order.id, settings.portal_token_secret) status_page = client.get(f"/portal/{token}/status") if status_page.status_code != 200 or "报告已就绪" not in status_page.text: raise RuntimeError( diff --git a/scripts/check_coverage_gate.py b/scripts/check_coverage_gate.py index 6d6eebe..0001283 100644 --- a/scripts/check_coverage_gate.py +++ b/scripts/check_coverage_gate.py @@ -10,8 +10,8 @@ from pathlib import Path OVERALL_MIN = 0.80 CORE_MIN = 1.00 CORE_FILES = ( - "gaokao-spec-checker/scripts/spec_checker_v2.py", - "gaokao-college-advisor/scripts/gaokao_visual_report.py", + "skills/gaokao-spec-checker/scripts/spec_checker_v2.py", + "skills/gaokao-college-advisor/scripts/gaokao_visual_report.py", ) diff --git a/scripts/dev-verify.sh b/scripts/dev-verify.sh index 7a87298..5a24c22 100644 --- a/scripts/dev-verify.sh +++ b/scripts/dev-verify.sh @@ -41,6 +41,9 @@ run_checks() { --cov-fail-under=80 \ -q + log "running core coverage verifier" + python scripts/check_coverage_gate.py coverage.xml + log "running ruff" python -m ruff check . --exclude .venv,.worktrees diff --git a/scripts/gaokao-delivery-dispatch.py b/scripts/gaokao-delivery-dispatch.py index 31e99d4..3fd155f 100644 --- a/scripts/gaokao-delivery-dispatch.py +++ b/scripts/gaokao-delivery-dispatch.py @@ -13,6 +13,7 @@ if str(ROOT) not in sys.path: def main() -> int: from admin.config import load_settings from data.notifications.dispatcher import DeliveryDispatcher + from data.notifications.smtp_sender import SMTPDeliverySender, SMTPSettings parser = argparse.ArgumentParser( description="Dispatch delivery notification events" @@ -22,7 +23,23 @@ def main() -> int: args = parser.parse_args() settings = load_settings() - dispatcher = DeliveryDispatcher.for_db(settings.orders_db_path) + email_sender = None + if args.channel == "email" and settings.smtp_host: + email_sender = SMTPDeliverySender( + SMTPSettings( + host=settings.smtp_host, + port=settings.smtp_port, + sender=settings.smtp_sender, + username=settings.smtp_username, + password=settings.smtp_password, + use_tls=settings.smtp_use_tls, + use_ssl=settings.smtp_use_ssl, + ) + ) + dispatcher = DeliveryDispatcher.for_db( + settings.orders_db_path, + email_sender=email_sender, + ) try: result = dispatcher.dispatch_ready_events( channel=args.channel, limit=args.limit diff --git a/tests/test_delivery_dispatcher.py b/tests/test_delivery_dispatcher.py index 31f714e..e580aa5 100644 --- a/tests/test_delivery_dispatcher.py +++ b/tests/test_delivery_dispatcher.py @@ -1,5 +1,6 @@ from __future__ import annotations +import json import subprocess from pathlib import Path @@ -13,7 +14,12 @@ from data.payments.service import PaymentService PROJECT_ROOT = Path(__file__).resolve().parents[1] -def _seed_order(db_path: str, order_id: str = "GKO-20260614-DISPATCH") -> Order: +def _seed_order( + db_path: str, + order_id: str = "GKO-20260614-DISPATCH", + *, + customer_email: str | None = None, +) -> Order: order = Order( id=order_id, source="web", @@ -22,6 +28,7 @@ def _seed_order(db_path: str, order_id: str = "GKO-20260614-DISPATCH") -> Order: status="pending", customer_name="张家长", customer_phone="13800138000", + customer_email=customer_email, candidate_name="张三", candidate_province="湖南", ) @@ -89,6 +96,72 @@ def test_dispatch_ready_station_event_marks_sent(settings, tmp_path): notification_service.close() assert event.status == "sent" assert event.attempt_count == 1 + payload = json.loads(event.payload_json) + station_notice = payload.get("station_notice") + assert station_notice is not None + assert station_notice["title"] == "报告已就绪" + assert order.id in station_notice["body"] + assert station_notice["sent_at"] == event.last_attempt_at + + +class _FakeEmailSender: + def __init__(self) -> None: + self.sent: list[dict[str, str]] = [] + + def send_report_ready( + self, *, recipient: str, order_id: str, subject: str, body: str + ) -> dict[str, str]: + payload = { + "recipient": recipient, + "order_id": order_id, + "subject": subject, + "body": body, + } + self.sent.append(payload) + return payload + + +def test_dispatch_ready_email_event_marks_sent(settings, tmp_path): + order = _seed_order( + settings.orders_db_path, + order_id="GKO-20260614-DISPATCH-EMAIL", + customer_email="parent@example.com", + ) + _mark_paid(settings, order) + IntakeStore.for_db(settings.orders_db_path).save( + order_id=order.id, payload={"candidate_score": 578}, submit=True + ) + _attach_ready_delivery(settings, tmp_path, order.id) + + from data.notifications.dispatcher import DeliveryDispatcher + + email_sender = _FakeEmailSender() + dispatcher = DeliveryDispatcher.for_db( + settings.orders_db_path, + email_sender=email_sender, + ) + try: + result = dispatcher.dispatch_ready_events(channel="email") + finally: + dispatcher.close() + + assert result.processed == 1 + assert result.sent == 1 + assert result.failed == 0 + assert len(email_sender.sent) == 1 + assert email_sender.sent[0]["recipient"] == "parent@example.com" + + notification_service = DeliveryNotificationService.for_db(settings.orders_db_path) + try: + event = notification_service.list_events(order.id, channel="email")[0] + finally: + notification_service.close() + assert event.status == "sent" + payload = json.loads(event.payload_json) + email_notice = payload.get("email_notice") + assert email_notice is not None + assert email_notice["recipient"] == "parent@example.com" + assert email_notice["subject"] def test_dispatch_ready_station_event_marks_failed_when_pdf_missing(settings, tmp_path): diff --git a/tests/test_delivery_notification.py b/tests/test_delivery_notification.py index dda9ffa..a522bab 100644 --- a/tests/test_delivery_notification.py +++ b/tests/test_delivery_notification.py @@ -7,7 +7,12 @@ from data.orders.models import Order from data.payments.service import PaymentService -def _seed_order(db_path: str, order_id: str = "GKO-20260614-NOTIFY") -> Order: +def _seed_order( + db_path: str, + order_id: str = "GKO-20260614-NOTIFY", + *, + customer_email: str | None = None, +) -> Order: order = Order( id=order_id, source="web", @@ -16,6 +21,7 @@ def _seed_order(db_path: str, order_id: str = "GKO-20260614-NOTIFY") -> Order: status="pending", customer_name="张家长", customer_phone="13800138000", + customer_email=customer_email, candidate_name="张三", candidate_province="湖南", ) @@ -83,7 +89,11 @@ def test_report_ready_transition_creates_notification_event( def test_dao_delivered_transition_also_creates_notification_event(settings, tmp_path): - order = _seed_order(settings.orders_db_path, order_id="GKO-20260614-NOTIFY-DAO") + order = _seed_order( + settings.orders_db_path, + order_id="GKO-20260614-NOTIFY-DAO", + customer_email="parent@example.com", + ) _mark_paid(settings, order) IntakeStore.for_db(settings.orders_db_path).save( order_id=order.id, payload={"candidate_score": 578}, submit=True @@ -111,11 +121,12 @@ def test_dao_delivered_transition_also_creates_notification_event(settings, tmp_ events = notification_service.list_events(order.id) finally: notification_service.close() - assert len(events) == 1 - assert events[0].event_type == "report_ready" - assert events[0].status == "ready" - assert events[0].attempt_count == 1 - assert events[0].failure_reason is None + assert len(events) == 2 + event_types = {(event.channel, event.event_type) for event in events} + assert ("station", "report_ready") in event_types + assert ("email", "report_ready_email") in event_types + email_event = next(event for event in events if event.channel == "email") + assert "parent@example.com" in email_event.payload_json def test_delivery_notification_tracks_failure_and_sent_status(settings, tmp_path): diff --git a/tests/test_smtp_sender.py b/tests/test_smtp_sender.py new file mode 100644 index 0000000..cad4a59 --- /dev/null +++ b/tests/test_smtp_sender.py @@ -0,0 +1,63 @@ +from __future__ import annotations + +import sys +import threading +import time +from typing import cast + +import pytest + +from data.notifications.smtp_sender import SMTPDeliverySender, SMTPSettings + + +@pytest.mark.skipif( + sys.version_info >= (3, 12), reason="stdlib smtpd/asyncore removed in Python 3.12+" +) +def test_smtp_delivery_sender_sends_message_over_local_stub(): + import asyncore + import smtpd + + class _CapturingSMTPServer(smtpd.SMTPServer): + def __init__(self, localaddr, remoteaddr): + super().__init__(localaddr, remoteaddr) + self.messages: list[dict[str, object]] = [] + + def process_message(self, peer, mailfrom, rcpttos, data, **kwargs): + self.messages.append({ + "peer": peer, + "mailfrom": mailfrom, + "rcpttos": rcpttos, + "data": data, + }) + return None + + server = _CapturingSMTPServer(("127.0.0.1", 0), None) + port = server.socket.getsockname()[1] + thread = threading.Thread( + target=asyncore.loop, kwargs={"timeout": 0.1}, daemon=True + ) + thread.start() + try: + sender = SMTPDeliverySender( + SMTPSettings( + host="127.0.0.1", + port=port, + sender="noreply@example.com", + ) + ) + result = sender.send_report_ready( + recipient="parent@example.com", + order_id="GKO-EMAIL-STUB", + subject="报告已就绪", + body="请查看 portal。", + ) + deadline = time.time() + 2 + while time.time() < deadline and not server.messages: + time.sleep(0.05) + assert server.messages, "stub smtp server did not receive message" + assert result["recipient"] == "parent@example.com" + raw = cast(bytes, server.messages[0]["data"]) + assert b"Subject: =?utf-8?" in raw or b"Subject: " in raw + assert b"parent@example.com" in raw + finally: + server.close()