diff --git a/backend/app.py b/backend/app.py index d040ca2..c0dc2ac 100644 --- a/backend/app.py +++ b/backend/app.py @@ -26,6 +26,7 @@ import uvicorn from db.session import sync_default_tools, sync_interface_definitions from services.knowledge import recover_interrupted_documents from services.post_call.worker import recover_interrupted_analyses, run_analysis_worker +from services.webhooks.worker import recover_interrupted_webhooks, run_webhook_worker from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware @@ -39,6 +40,7 @@ from routes import ( mcp_servers, model_registry, node_types, + site_settings, tools, voice_webrtc, voice_ws, @@ -51,14 +53,21 @@ async def lifespan(_app: FastAPI): await sync_default_tools() await recover_interrupted_documents() await recover_interrupted_analyses() + await recover_interrupted_webhooks() analysis_worker = asyncio.create_task( run_analysis_worker(), name="post-call-analysis-worker" ) + webhook_worker = asyncio.create_task( + run_webhook_worker(), name="analysis-webhook-worker" + ) try: yield finally: analysis_worker.cancel() - await asyncio.gather(analysis_worker, return_exceptions=True) + webhook_worker.cancel() + await asyncio.gather( + analysis_worker, webhook_worker, return_exceptions=True + ) await voice_webrtc.shutdown_active_sessions() @@ -81,6 +90,7 @@ app.include_router(knowledge_bases.router) app.include_router(mcp_servers.router) app.include_router(model_registry.router) app.include_router(node_types.router) +app.include_router(site_settings.router) app.include_router(tools.router) app.include_router(voice_webrtc.router) app.include_router(voice_ws.router) diff --git a/backend/db/models.py b/backend/db/models.py index 6933fe2..0361785 100644 --- a/backend/db/models.py +++ b/backend/db/models.py @@ -374,3 +374,51 @@ class ConversationArtifact(Base): DateTime(timezone=True), server_default=func.now() ) extra: Mapped[dict] = mapped_column(JSONB, default=dict) + + +class SiteSetting(Base): + """Deployment-wide settings, split into public values and write-only secrets.""" + + __tablename__ = "site_settings" + + key: Mapped[str] = mapped_column(String(64), primary_key=True) + value: Mapped[dict] = mapped_column(JSONB, default=dict) + secrets: Mapped[dict] = mapped_column(JSONB, default=dict) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), onupdate=func.now() + ) + + +class WebhookDelivery(Base): + """Immutable event snapshot plus its asynchronous delivery state.""" + + __tablename__ = "webhook_deliveries" + + id: Mapped[str] = mapped_column(String(40), primary_key=True) + event_id: Mapped[str] = mapped_column(String(40), unique=True) + event_type: Mapped[str] = mapped_column(String(128), index=True) + conversation_id: Mapped[str | None] = mapped_column( + String(40), + ForeignKey("conversation_sessions.id", ondelete="SET NULL"), + nullable=True, + index=True, + ) + payload: Mapped[dict] = mapped_column(JSONB) + url: Mapped[str] = mapped_column(String(2048)) + secrets: Mapped[dict] = mapped_column(JSONB, default=dict) + status: Mapped[str] = mapped_column(String(16), index=True, default="pending") + attempt_count: Mapped[int] = mapped_column(Integer, default=0) + next_attempt_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True, index=True + ) + last_status_code: Mapped[int | None] = mapped_column(Integer, nullable=True) + last_error: Mapped[str] = mapped_column(String(2048), default="") + delivered_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now() + ) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), onupdate=func.now() + ) diff --git a/backend/migrations/versions/20260807_0015_add_analysis_webhook.py b/backend/migrations/versions/20260807_0015_add_analysis_webhook.py new file mode 100644 index 0000000..be38e09 --- /dev/null +++ b/backend/migrations/versions/20260807_0015_add_analysis_webhook.py @@ -0,0 +1,86 @@ +"""add global analysis webhook settings and delivery queue + +Revision ID: 20260807_0015 +Revises: 20260807_0014 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + + +revision: str = "20260807_0015" +down_revision: str | Sequence[str] | None = "20260807_0014" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.create_table( + "site_settings", + sa.Column("key", sa.String(length=64), nullable=False), + sa.Column( + "value", + postgresql.JSONB(astext_type=sa.Text()), + server_default=sa.text("'{}'::jsonb"), + nullable=False, + ), + sa.Column( + "secrets", + postgresql.JSONB(astext_type=sa.Text()), + server_default=sa.text("'{}'::jsonb"), + nullable=False, + ), + sa.Column( + "updated_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.PrimaryKeyConstraint("key"), + ) + op.create_table( + "webhook_deliveries", + sa.Column("id", sa.String(length=40), nullable=False), + sa.Column("event_id", sa.String(length=40), nullable=False), + sa.Column("event_type", sa.String(length=128), nullable=False), + sa.Column("conversation_id", sa.String(length=40), nullable=True), + sa.Column("payload", postgresql.JSONB(astext_type=sa.Text()), nullable=False), + sa.Column("url", sa.String(length=2048), nullable=False), + sa.Column( + "secrets", + postgresql.JSONB(astext_type=sa.Text()), + server_default=sa.text("'{}'::jsonb"), + nullable=False, + ), + sa.Column("status", sa.String(length=16), server_default="pending", nullable=False), + sa.Column("attempt_count", sa.Integer(), server_default="0", nullable=False), + sa.Column("next_attempt_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("last_status_code", sa.Integer(), nullable=True), + sa.Column("last_error", sa.String(length=2048), server_default="", nullable=False), + sa.Column("delivered_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), + sa.ForeignKeyConstraint( + ["conversation_id"], ["conversation_sessions.id"], ondelete="SET NULL" + ), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("event_id"), + ) + op.create_index("ix_webhook_deliveries_event_type", "webhook_deliveries", ["event_type"]) + op.create_index("ix_webhook_deliveries_conversation_id", "webhook_deliveries", ["conversation_id"]) + op.create_index("ix_webhook_deliveries_status", "webhook_deliveries", ["status"]) + op.create_index("ix_webhook_deliveries_next_attempt_at", "webhook_deliveries", ["next_attempt_at"]) + + +def downgrade() -> None: + op.drop_index("ix_webhook_deliveries_next_attempt_at", table_name="webhook_deliveries") + op.drop_index("ix_webhook_deliveries_status", table_name="webhook_deliveries") + op.drop_index("ix_webhook_deliveries_conversation_id", table_name="webhook_deliveries") + op.drop_index("ix_webhook_deliveries_event_type", table_name="webhook_deliveries") + op.drop_table("webhook_deliveries") + op.drop_table("site_settings") diff --git a/backend/routes/site_settings.py b/backend/routes/site_settings.py new file mode 100644 index 0000000..5dba109 --- /dev/null +++ b/backend/routes/site_settings.py @@ -0,0 +1,126 @@ +"""Administration API for deployment-wide settings.""" + +from __future__ import annotations + +from datetime import UTC, datetime +import uuid + +from db.models import SiteSetting +from db.session import get_session +from fastapi import APIRouter, Depends, HTTPException +from schemas import ( + AnalysisWebhookSettingOut, + AnalysisWebhookSettingUpdate, + AnalysisWebhookTestIn, + AnalysisWebhookTestOut, +) +from services.auth import require_admin +from services.webhooks.delivery import deliver_webhook +from services.webhooks.events import ANALYSIS_WEBHOOK_KEY +from services.webhooks.security import UnsafeWebhookUrl, validate_webhook_url +from sqlalchemy.ext.asyncio import AsyncSession + + +router = APIRouter( + prefix="/api/settings", + tags=["site-settings"], + dependencies=[Depends(require_admin)], +) + + +def _setting_out(row: SiteSetting | None) -> AnalysisWebhookSettingOut: + values = dict(row.value or {}) if row else {} + secrets = dict(row.secrets or {}) if row else {} + return AnalysisWebhookSettingOut( + enabled=bool(values.get("enabled")), + url=str(values.get("url") or ""), + secret_configured=bool(secrets.get("secret")), + updated_at=row.updated_at if row else None, + ) + + +async def _safe_url_or_422(url: str) -> str: + try: + return await validate_webhook_url(url) + except UnsafeWebhookUrl as exc: + raise HTTPException(422, str(exc)) from exc + + +@router.get("/analysis-webhook", response_model=AnalysisWebhookSettingOut) +async def get_analysis_webhook(session: AsyncSession = Depends(get_session)): + return _setting_out(await session.get(SiteSetting, ANALYSIS_WEBHOOK_KEY)) + + +@router.put("/analysis-webhook", response_model=AnalysisWebhookSettingOut) +async def update_analysis_webhook( + body: AnalysisWebhookSettingUpdate, + session: AsyncSession = Depends(get_session), +): + row = await session.get(SiteSetting, ANALYSIS_WEBHOOK_KEY) + stored_secrets = dict(row.secrets or {}) if row else {} + incoming_secret = (body.secret or "").strip() + if body.clear_secret: + stored_secrets.pop("secret", None) + if incoming_secret: + stored_secrets["secret"] = incoming_secret + + if body.url: + await _safe_url_or_422(body.url) + if body.enabled and not stored_secrets.get("secret"): + raise HTTPException(422, "启用 Webhook 时必须配置签名密钥") + + if row is None: + row = SiteSetting(key=ANALYSIS_WEBHOOK_KEY) + session.add(row) + row.value = {"enabled": body.enabled, "url": body.url} + row.secrets = stored_secrets + await session.commit() + await session.refresh(row) + return _setting_out(row) + + +@router.post("/analysis-webhook/test", response_model=AnalysisWebhookTestOut) +async def test_analysis_webhook( + body: AnalysisWebhookTestIn, + session: AsyncSession = Depends(get_session), +): + url = await _safe_url_or_422(body.url) + row = await session.get(SiteSetting, ANALYSIS_WEBHOOK_KEY) + stored_secret = str((row.secrets or {}).get("secret") or "") if row else "" + secret = (body.secret or "").strip() or stored_secret + if not secret: + raise HTTPException(422, "请先填写或保存签名密钥") + + event_id = f"evt_{uuid.uuid4().hex}" + payload = { + "id": event_id, + "event": "call.analysis.completed", + "timestamp": datetime.now(UTC).isoformat(), + "data": { + "conversationId": "conv_test_001", + "assistantId": "asst_test_001", + "assistantName": "Webhook 测试助手", + "analysis": { + "customer_intent": "high", + "need_follow_up": True, + }, + }, + } + result = await deliver_webhook( + url=url, + secret=secret, + event_id=event_id, + payload=payload, + ) + message = ( + f"测试事件已送达，目标返回 HTTP {result.status_code}。" + if result.ok + else result.error or "测试事件发送失败。" + ) + return AnalysisWebhookTestOut( + ok=result.ok, + status_code=result.status_code, + latency_ms=result.latency_ms, + message=message, + payload=payload, + ) diff --git a/backend/schemas.py b/backend/schemas.py index c82e2f2..96e43ee 100644 --- a/backend/schemas.py +++ b/backend/schemas.py @@ -566,3 +566,38 @@ class ConversationListOut(CamelModel): total: int page: int page_size: int + + +# ---------- 站点级 Webhook ---------- +class AnalysisWebhookSettingOut(CamelModel): + enabled: bool = False + url: str = "" + secret_configured: bool = False + updated_at: datetime | None = None + + +class AnalysisWebhookSettingUpdate(CamelModel): + enabled: bool = False + url: str = Field(default="", max_length=2048) + secret: str | None = Field(default=None, max_length=512) + clear_secret: bool = False + + @model_validator(mode="after") + def validate_enabled_url(self): + self.url = self.url.strip() + if self.enabled and not self.url: + raise ValueError("启用 Webhook 时必须填写 URL") + return self + + +class AnalysisWebhookTestIn(CamelModel): + url: str = Field(min_length=1, max_length=2048) + secret: str | None = Field(default=None, max_length=512) + + +class AnalysisWebhookTestOut(CamelModel): + ok: bool + status_code: int | None = None + latency_ms: int | None = None + message: str + payload: dict[str, Any] diff --git a/backend/services/post_call/analyzer.py b/backend/services/post_call/analyzer.py index ba11215..64dbf04 100644 --- a/backend/services/post_call/analyzer.py +++ b/backend/services/post_call/analyzer.py @@ -10,6 +10,7 @@ import httpx from db.models import ConversationMessage, ConversationSession, ModelResource from db.session import SessionLocal from sqlalchemy import select +from services.webhooks.events import enqueue_analysis_completed ANALYSIS_TIMEOUT_SECONDS = 60.0 @@ -184,6 +185,7 @@ async def analyze_conversation(conversation_id: str) -> None: conversation.analysis_data = data conversation.analysis_status = "completed" conversation.analysis_error = "" + await enqueue_analysis_completed(session, conversation, result) await session.commit() except Exception as exc: async with SessionLocal() as session: @@ -192,4 +194,3 @@ async def analyze_conversation(conversation_id: str) -> None: conversation.analysis_status = "failed" conversation.analysis_error = str(exc)[:2048] await session.commit() - diff --git a/backend/services/webhooks/__init__.py b/backend/services/webhooks/__init__.py new file mode 100644 index 0000000..379127e --- /dev/null +++ b/backend/services/webhooks/__init__.py @@ -0,0 +1 @@ +"""Global webhook delivery helpers.""" diff --git a/backend/services/webhooks/delivery.py b/backend/services/webhooks/delivery.py new file mode 100644 index 0000000..4ff3178 --- /dev/null +++ b/backend/services/webhooks/delivery.py @@ -0,0 +1,86 @@ +"""Serialize, sign and send one webhook event.""" + +from __future__ import annotations + +from dataclasses import dataclass +import hashlib +import hmac +import json +import time +from typing import Any + +import httpx + +from services.webhooks.security import validate_webhook_url + + +WEBHOOK_TIMEOUT_SECONDS = 10.0 + + +@dataclass(frozen=True) +class DeliveryResult: + ok: bool + status_code: int | None + latency_ms: int + error: str = "" + + +def canonical_json(payload: dict[str, Any]) -> bytes: + return json.dumps( + payload, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ).encode("utf-8") + + +def webhook_signature(secret: str, timestamp: str, body: bytes) -> str: + signed = timestamp.encode("ascii") + b"." + body + digest = hmac.new(secret.encode("utf-8"), signed, hashlib.sha256).hexdigest() + return f"v1={digest}" + + +def should_retry(status_code: int | None) -> bool: + return status_code is None or status_code in {408, 429} or status_code >= 500 + + +async def deliver_webhook( + *, + url: str, + secret: str, + event_id: str, + payload: dict[str, Any], +) -> DeliveryResult: + """Deliver without following redirects; callers decide whether to retry.""" + safe_url = await validate_webhook_url(url) + body = canonical_json(payload) + timestamp = str(int(time.time())) + headers = { + "Content-Type": "application/json", + "X-AIVideo-Event-Id": event_id, + "X-AIVideo-Timestamp": timestamp, + "X-AIVideo-Signature": webhook_signature(secret, timestamp, body), + } + started = time.monotonic() + try: + async with httpx.AsyncClient( + timeout=WEBHOOK_TIMEOUT_SECONDS, + follow_redirects=False, + ) as client: + response = await client.post(safe_url, content=body, headers=headers) + latency_ms = round((time.monotonic() - started) * 1000) + if 200 <= response.status_code < 300: + return DeliveryResult(True, response.status_code, latency_ms) + return DeliveryResult( + False, + response.status_code, + latency_ms, + f"目标服务返回 HTTP {response.status_code}", + ) + except httpx.HTTPError as exc: + return DeliveryResult( + False, + None, + round((time.monotonic() - started) * 1000), + str(exc)[:2048], + ) diff --git a/backend/services/webhooks/events.py b/backend/services/webhooks/events.py new file mode 100644 index 0000000..734803c --- /dev/null +++ b/backend/services/webhooks/events.py @@ -0,0 +1,69 @@ +"""Build durable webhook events in the same transaction as domain updates.""" + +from __future__ import annotations + +from datetime import UTC, datetime +import uuid +from typing import Any + +from db.models import ConversationSession, SiteSetting, WebhookDelivery +from sqlalchemy.ext.asyncio import AsyncSession + + +ANALYSIS_WEBHOOK_KEY = "analysis_webhook" +ANALYSIS_COMPLETED_EVENT = "call.analysis.completed" + + +def analysis_completed_payload( + conversation: ConversationSession, + result: dict[str, Any], + *, + event_id: str, + timestamp: datetime, +) -> dict[str, Any]: + return { + "id": event_id, + "event": ANALYSIS_COMPLETED_EVENT, + "timestamp": timestamp.isoformat(), + "data": { + "conversationId": conversation.id, + "assistantId": conversation.assistant_id, + "assistantName": conversation.assistant_name, + "analysis": result, + }, + } + + +async def enqueue_analysis_completed( + session: AsyncSession, + conversation: ConversationSession, + result: dict[str, Any], +) -> WebhookDelivery | None: + setting = await session.get(SiteSetting, ANALYSIS_WEBHOOK_KEY) + values = dict(setting.value or {}) if setting else {} + secrets = dict(setting.secrets or {}) if setting else {} + url = str(values.get("url") or "") + secret = str(secrets.get("secret") or "") + if not values.get("enabled") or not url or not secret: + return None + + event_id = f"evt_{uuid.uuid4().hex}" + now = datetime.now(UTC) + delivery = WebhookDelivery( + id=f"dlv_{uuid.uuid4().hex}", + event_id=event_id, + event_type=ANALYSIS_COMPLETED_EVENT, + conversation_id=conversation.id, + payload=analysis_completed_payload( + conversation, + result, + event_id=event_id, + timestamp=now, + ), + url=url, + secrets={"secret": secret}, + status="pending", + next_attempt_at=now, + ) + session.add(delivery) + return delivery diff --git a/backend/services/webhooks/security.py b/backend/services/webhooks/security.py new file mode 100644 index 0000000..5a5dc6f --- /dev/null +++ b/backend/services/webhooks/security.py @@ -0,0 +1,61 @@ +"""Validate webhook destinations before every outbound request.""" + +from __future__ import annotations + +import asyncio +import ipaddress +import socket +from urllib.parse import urlsplit + + +class UnsafeWebhookUrl(ValueError): + """Raised when a URL could reach local or otherwise non-public infrastructure.""" + + def __init__(self, message: str, *, retryable: bool = False): + super().__init__(message) + self.retryable = retryable + + +def _public_ip(address: str) -> bool: + try: + return ipaddress.ip_address(address).is_global + except ValueError: + return False + + +async def validate_webhook_url(url: str) -> str: + """Require HTTPS and ensure every currently resolved address is public.""" + value = url.strip() + parsed = urlsplit(value) + if parsed.scheme != "https" or not parsed.hostname: + raise UnsafeWebhookUrl("Webhook URL 必须是完整的 HTTPS 地址") + if parsed.username or parsed.password: + raise UnsafeWebhookUrl("Webhook URL 不能包含用户名或密码") + + try: + port = parsed.port or 443 + except ValueError as exc: + raise UnsafeWebhookUrl("Webhook URL 端口无效") from exc + + try: + direct_ip = ipaddress.ip_address(parsed.hostname) + except ValueError: + direct_ip = None + if direct_ip is not None: + if not direct_ip.is_global: + raise UnsafeWebhookUrl("Webhook URL 不能指向内网或本机地址") + return value + + try: + records = await asyncio.to_thread( + socket.getaddrinfo, + parsed.hostname, + port, + type=socket.SOCK_STREAM, + ) + except OSError as exc: + raise UnsafeWebhookUrl("Webhook 域名暂时无法解析", retryable=True) from exc + addresses = {str(record[4][0]) for record in records} + if not addresses or any(not _public_ip(address) for address in addresses): + raise UnsafeWebhookUrl("Webhook 域名解析到了内网或非公网地址") + return value diff --git a/backend/services/webhooks/worker.py b/backend/services/webhooks/worker.py new file mode 100644 index 0000000..98dfa96 --- /dev/null +++ b/backend/services/webhooks/worker.py @@ -0,0 +1,120 @@ +"""PostgreSQL-backed webhook delivery worker with bounded retries.""" + +from __future__ import annotations + +import asyncio +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta + +from db.models import WebhookDelivery +from db.session import SessionLocal +from loguru import logger +from services.webhooks.delivery import DeliveryResult, deliver_webhook, should_retry +from services.webhooks.security import UnsafeWebhookUrl +from sqlalchemy import or_, select, update + + +POLL_INTERVAL_SECONDS = 2.0 +MAX_ATTEMPTS = 5 +RETRY_DELAYS_SECONDS = (30, 120, 600, 3600) + + +@dataclass(frozen=True) +class ClaimedDelivery: + id: str + event_id: str + url: str + secret: str + payload: dict + + +async def recover_interrupted_webhooks() -> None: + async with SessionLocal() as session: + await session.execute( + update(WebhookDelivery) + .where(WebhookDelivery.status == "processing") + .values(status="pending", next_attempt_at=datetime.now(UTC)) + ) + await session.commit() + + +async def claim_pending_webhook() -> ClaimedDelivery | None: + now = datetime.now(UTC) + async with SessionLocal() as session: + row = ( + await session.execute( + select(WebhookDelivery) + .where( + WebhookDelivery.status == "pending", + or_( + WebhookDelivery.next_attempt_at.is_(None), + WebhookDelivery.next_attempt_at <= now, + ), + ) + .order_by(WebhookDelivery.next_attempt_at, WebhookDelivery.created_at) + .with_for_update(skip_locked=True) + .limit(1) + ) + ).scalar_one_or_none() + if not row: + return None + row.status = "processing" + await session.commit() + return ClaimedDelivery( + id=row.id, + event_id=row.event_id, + url=row.url, + secret=str((row.secrets or {}).get("secret") or ""), + payload=dict(row.payload or {}), + ) + + +async def finish_delivery(delivery_id: str, result: DeliveryResult) -> None: + async with SessionLocal() as session: + row = await session.get(WebhookDelivery, delivery_id) + if not row or row.status != "processing": + return + row.attempt_count += 1 + row.last_status_code = result.status_code + row.last_error = result.error[:2048] + if result.ok: + row.status = "delivered" + row.delivered_at = datetime.now(UTC) + row.next_attempt_at = None + elif should_retry(result.status_code) and row.attempt_count < MAX_ATTEMPTS: + delay_index = min(row.attempt_count - 1, len(RETRY_DELAYS_SECONDS) - 1) + row.status = "pending" + row.next_attempt_at = datetime.now(UTC) + timedelta( + seconds=RETRY_DELAYS_SECONDS[delay_index] + ) + else: + row.status = "failed" + row.next_attempt_at = None + await session.commit() + + +async def run_webhook_worker() -> None: + logger.info("通话分析 Webhook worker 已启动") + while True: + try: + delivery = await claim_pending_webhook() + if delivery: + try: + result = await deliver_webhook( + url=delivery.url, + secret=delivery.secret, + event_id=delivery.event_id, + payload=delivery.payload, + ) + except UnsafeWebhookUrl as exc: + # DNS outages are transient; policy violations are permanent. + result = DeliveryResult( + False, None if exc.retryable else 400, 0, str(exc) + ) + await finish_delivery(delivery.id, result) + continue + except asyncio.CancelledError: + raise + except Exception as exc: + logger.error(f"Webhook worker 暂时不可用:{exc}") + await asyncio.sleep(POLL_INTERVAL_SECONDS) diff --git a/backend/tests/test_webhooks.py b/backend/tests/test_webhooks.py new file mode 100644 index 0000000..6c7b317 --- /dev/null +++ b/backend/tests/test_webhooks.py @@ -0,0 +1,82 @@ +from __future__ import annotations + +import hashlib +import hmac +from types import SimpleNamespace +import unittest +from unittest.mock import patch + +from services.webhooks.delivery import ( + canonical_json, + should_retry, + webhook_signature, +) +from services.webhooks.events import analysis_completed_payload +from services.webhooks.security import UnsafeWebhookUrl, validate_webhook_url + + +class WebhookSecurityTest(unittest.IsolatedAsyncioTestCase): + async def test_requires_https(self): + with self.assertRaises(UnsafeWebhookUrl): + await validate_webhook_url("http://hooks.example.com/events") + + async def test_rejects_loopback_ip(self): + with self.assertRaises(UnsafeWebhookUrl): + await validate_webhook_url("https://127.0.0.1/events") + + async def test_rejects_hostname_resolving_to_private_ip(self): + records = [(2, 1, 6, "", ("10.0.0.8", 443))] + with patch("services.webhooks.security.socket.getaddrinfo", return_value=records): + with self.assertRaises(UnsafeWebhookUrl): + await validate_webhook_url("https://hooks.example.com/events") + + async def test_accepts_hostname_resolving_to_public_ip(self): + records = [(2, 1, 6, "", ("8.8.8.8", 443))] + with patch("services.webhooks.security.socket.getaddrinfo", return_value=records): + result = await validate_webhook_url("https://hooks.example.com/events") + self.assertEqual(result, "https://hooks.example.com/events") + + +class WebhookDeliveryTest(unittest.TestCase): + def test_signature_uses_timestamp_dot_raw_body(self): + body = canonical_json({"b": 2, "a": "中文"}) + expected = hmac.new( + b"secret", + b"1700000000." + body, + hashlib.sha256, + ).hexdigest() + self.assertEqual( + webhook_signature("secret", "1700000000", body), + f"v1={expected}", + ) + + def test_only_transient_responses_are_retried(self): + for status in (None, 408, 429, 500, 503): + self.assertTrue(should_retry(status)) + for status in (200, 301, 400, 401, 404): + self.assertFalse(should_retry(status)) + + def test_analysis_event_has_stable_envelope(self): + conversation = SimpleNamespace( + id="conv_001", + assistant_id="asst_001", + assistant_name="销售助手", + ) + from datetime import UTC, datetime + + payload = analysis_completed_payload( + conversation, # type: ignore[arg-type] + {"customer_intent": "high"}, + event_id="evt_001", + timestamp=datetime(2026, 8, 7, tzinfo=UTC), + ) + self.assertEqual(payload["id"], "evt_001") + self.assertEqual(payload["event"], "call.analysis.completed") + self.assertEqual(payload["data"]["conversationId"], "conv_001") + self.assertEqual( + payload["data"]["analysis"], {"customer_intent": "high"} + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/frontend/src/components/pages/ProfilePage.tsx b/frontend/src/components/pages/ProfilePage.tsx index 43a13d7..06fab2f 100644 --- a/frontend/src/components/pages/ProfilePage.tsx +++ b/frontend/src/components/pages/ProfilePage.tsx @@ -17,6 +17,7 @@ import { SectionCard } from "@/components/editor/section-card"; import { ListPageLayout } from "@/components/layout/list-page-layout"; import { Button } from "@/components/ui/button"; import { Input } from "@/components/ui/input"; +import { siteSettingsApi } from "@/lib/api"; import { Select, SelectContent, @@ -26,6 +27,7 @@ import { } from "@/components/ui/select"; type TestStatus = "idle" | "loading" | "success" | "error"; +type SaveStatus = "loading" | "idle" | "saving" | "success" | "error"; type ProfileSectionId = "webhook" | "language"; const profileSections = [ @@ -35,22 +37,6 @@ const profileSections = [ const LANGUAGE_OPTIONS = [{ value: "zh-CN", label: "中文" }] as const; -function mockWebhookPayload() { - return { - id: "evt_mock_001", - event: "call.analysis.completed", - timestamp: new Date().toISOString(), - data: { - conversationId: "conv_mock_001", - assistantId: "asst_mock_001", - analysis: { - customer_intent: "high", - need_follow_up: true, - }, - }, - }; -} - function isHttpsUrl(value: string): boolean { try { return new URL(value).protocol === "https:"; @@ -66,10 +52,13 @@ function getAppScrollContainer(): HTMLElement | null { export function ProfilePage() { const [url, setUrl] = useState(""); const [secret, setSecret] = useState(""); + const [secretConfigured, setSecretConfigured] = useState(false); const [secretVisible, setSecretVisible] = useState(false); const [testStatus, setTestStatus] = useState("idle"); const [testMessage, setTestMessage] = useState(""); const [payload, setPayload] = useState | null>(null); + const [saveStatus, setSaveStatus] = useState("loading"); + const [saveMessage, setSaveMessage] = useState(""); const [language, setLanguage] = useState("zh-CN"); const [activeSection, setActiveSection] = useState("webhook"); @@ -80,6 +69,28 @@ export function ProfilePage() { }); const selectedAnchorRef = useRef(null); + useEffect(() => { + let active = true; + siteSettingsApi + .getAnalysisWebhook() + .then((setting) => { + if (!active) return; + setUrl(setting.url); + setSecretConfigured(setting.secretConfigured); + setSaveStatus("idle"); + }) + .catch((error: unknown) => { + if (!active) return; + setSaveStatus("error"); + setSaveMessage( + error instanceof Error ? error.message : "站点配置加载失败。", + ); + }); + return () => { + active = false; + }; + }, []); + useEffect(() => { const scrollContainer = getAppScrollContainer(); if (!scrollContainer) return; @@ -168,7 +179,7 @@ export function ProfilePage() { async function sendTestEvent() { if (!isHttpsUrl(url.trim())) { setTestStatus("error"); - setTestMessage("请填写完整的 HTTPS Webhook URL。测试不会发起网络请求。"); + setTestMessage("请填写完整的 HTTPS Webhook URL。"); setPayload(null); return; } @@ -176,10 +187,51 @@ export function ProfilePage() { setTestStatus("loading"); setTestMessage(""); setPayload(null); - await new Promise((resolve) => window.setTimeout(resolve, 700)); - setPayload(mockWebhookPayload()); - setTestStatus("success"); - setTestMessage("已在浏览器中生成测试事件，未向目标地址发送请求。"); + try { + const result = await siteSettingsApi.testAnalysisWebhook({ + url: url.trim(), + ...(secret.trim() ? { secret: secret.trim() } : {}), + }); + setPayload(result.payload); + setTestStatus(result.ok ? "success" : "error"); + setTestMessage( + result.latencyMs === null + ? result.message + : `${result.message} 耗时 ${result.latencyMs} ms。`, + ); + } catch (error) { + setTestStatus("error"); + setTestMessage(error instanceof Error ? error.message : "测试事件发送失败。"); + } + } + + async function saveWebhookSetting() { + const normalizedUrl = url.trim(); + if (normalizedUrl && !isHttpsUrl(normalizedUrl)) { + setSaveStatus("error"); + setSaveMessage("请填写完整的 HTTPS Webhook URL，或清空 URL 以停用。"); + return; + } + setSaveStatus("saving"); + setSaveMessage(""); + try { + const setting = await siteSettingsApi.updateAnalysisWebhook({ + enabled: Boolean(normalizedUrl), + url: normalizedUrl, + ...(secret.trim() ? { secret: secret.trim() } : {}), + }); + setSecret(""); + setSecretConfigured(setting.secretConfigured); + setSaveStatus("success"); + setSaveMessage( + setting.enabled + ? "Webhook 已启用，后续通话分析完成后会自动投递。" + : "Webhook 已停用。已保存的签名密钥不会被清除。", + ); + } catch (error) { + setSaveStatus("error"); + setSaveMessage(error instanceof Error ? error.message : "站点配置保存失败。"); + } } return ( @@ -207,7 +259,7 @@ export function ProfilePage() { } title="通话分析 Webhook" - description="站点级事件出口；后端接入后将在分析完成时执行 HTTP POST" + description="站点级事件出口；配置 URL 后将在每次通话分析完成时执行 HTTP POST" >
@@ -238,8 +292,16 @@ export function ProfilePage() { setSecret(event.target.value)} - placeholder="用于生成 HMAC-SHA256 签名" + onChange={(event) => { + setSecret(event.target.value); + setSaveStatus("idle"); + setSaveMessage(""); + }} + placeholder={ + secretConfigured + ? "已配置；留空表示不修改" + : "用于生成 HMAC-SHA256 签名" + } className="border-hairline-strong bg-background pr-10" />

- 计划使用 timestamp + raw body 计算签名；密钥不会出现在事件 - payload 中。 + 使用 timestamp + raw body 计算签名；保存后密钥不会返回浏览器。

@@ -263,7 +324,9 @@ export function ProfilePage() { type="button" variant="outline" className="gap-2 border-hairline-strong" - disabled={testStatus === "loading"} + disabled={ + testStatus === "loading" || saveStatus === "loading" + } onClick={() => void sendTestEvent()} > {testStatus === "loading" ? ( @@ -271,14 +334,39 @@ export function ProfilePage() { ) : ( )} - 生成测试事件 + 发送测试事件 - - 等待后端设置 API + {saveStatus === "loading" && ( + + + 正在加载配置 + + )} + {saveMessage && ( +
+ {saveMessage} +
+ )} + {testMessage && (
- Mock payload + 已发送的事件 payload 
diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts
index 6b79567..2eee52c 100644
--- a/frontend/src/lib/api.ts
+++ b/frontend/src/lib/api.ts
@@ -114,6 +114,45 @@ export const authApi = {
   me: () => request("/api/auth/me"),
 };
 
+// ---------- 站点级设置 ----------
+export type AnalysisWebhookSetting = {
+  enabled: boolean;
+  url: string;
+  secretConfigured: boolean;
+  updatedAt: string | null;
+};
+
+export type AnalysisWebhookTestResult = {
+  ok: boolean;
+  statusCode: number | null;
+  latencyMs: number | null;
+  message: string;
+  payload: Record;
+};
+
+export const siteSettingsApi = {
+  getAnalysisWebhook: () =>
+    request("/api/settings/analysis-webhook"),
+  updateAnalysisWebhook: (body: {
+    enabled: boolean;
+    url: string;
+    secret?: string;
+    clearSecret?: boolean;
+  }) =>
+    request("/api/settings/analysis-webhook", {
+      method: "PUT",
+      body: JSON.stringify(body),
+    }),
+  testAnalysisWebhook: (body: { url: string; secret?: string }) =>
+    request(
+      "/api/settings/analysis-webhook/test",
+      {
+        method: "POST",
+        body: JSON.stringify(body),
+      },
+    ),
+};
+
 // ---------- 接口定义驱动的模型注册表 ----------
 export type InterfaceField = {
   key: string;