From 5a1c582fe389f61ef5af1c6536caf598cc8ce155 Mon Sep 17 00:00:00 2001 From: proscar87 Date: Tue, 4 Aug 2026 21:06:05 -0600 Subject: [PATCH 1/3] feat(server): add outbound webhook dispatcher for note sync events Adds a WebhookService that sends HMAC-SHA256 signed HTTP POST notifications to configured endpoints, enabling integrations with Home Assistant, n8n, Zapier, etc. Dispatch is fire-and-forget: it subscribes on the existing LocalEventBus and never blocks or fails the request that triggered the event. - New `webhooks` config section (WebhookConfig/WebhookEndpointConfig) with per-endpoint url/secret/events, matching the config.yaml shape from #167. - Fires `note.sync_completed` off the existing NoteUpdatedEvent (the point where a synced .note file is durably written), signed via `X-Supernote-Signature: sha256=` using hmac.compare_digest-safe verification. - Each endpoint POST has an explicit 10s timeout and up to 3 attempts with linear backoff for network errors/5xx; 4xx responses are not retried. All failures are logged (never the secret or signature) and swallowed. Fixes #167 Co-Authored-By: Claude Fable 5 --- config-example.yaml | 14 ++ supernote/server/app.py | 5 + supernote/server/config.py | 49 +++++ supernote/server/services/webhook.py | 145 +++++++++++++ tests/server/services/test_webhook.py | 293 ++++++++++++++++++++++++++ tests/server/test_config.py | 48 +++++ 6 files changed, 554 insertions(+) create mode 100644 supernote/server/services/webhook.py create mode 100644 tests/server/services/test_webhook.py diff --git a/config-example.yaml b/config-example.yaml index e5230e48..9d21419a 100644 --- a/config-example.yaml +++ b/config-example.yaml @@ -46,3 +46,17 @@ auth: # Whether to allow remote password reset via email. # When disabled, the public password reset endpoint returns 403. enable_remote_password_reset: false + +# Outbound webhook dispatch (optional). +# Sends a signed HTTP POST to each endpoint when a subscribed event occurs, +# e.g. to trigger a Home Assistant, n8n, or Zapier automation on note sync. +# webhooks: +# enabled: true +# endpoints: +# - url: "https://n8n.homelab.local/webhook/supernote" +# # Shared secret used to HMAC-SHA256 sign each payload. Sent to the +# # endpoint in the `X-Supernote-Signature: sha256=` header so it +# # can verify the request came from this server and was not altered. +# secret: "CHANGE_ME_TO_A_SECURE_RANDOM_STRING" +# events: +# - "note.sync_completed" diff --git a/supernote/server/app.py b/supernote/server/app.py index 6c014fed..84394659 100644 --- a/supernote/server/app.py +++ b/supernote/server/app.py @@ -53,6 +53,7 @@ from .services.search import SearchService from .services.summary import SummaryService from .services.user import UserService +from .services.webhook import WebhookService from .socket import setup_socketio from .utils.hashing import get_md5_hash from .utils.prompt_loader import PROMPT_LOADER @@ -365,6 +366,9 @@ def create_app(config: ServerConfig) -> web.Application: ) app["processor_service"] = processor_service + webhook_service = WebhookService(config.webhooks, event_bus) + app["webhook_service"] = webhook_service + # Register modules processor_service.register_modules( hashing=PageHashingModule(file_service=file_service), @@ -467,6 +471,7 @@ async def on_startup_handler(app: web.Application) -> None: logger.info("Starting background services...") await processor_service.start() + webhook_service.start() logger.info("Startup sequence complete.") app["mcp_task"] = mcp_task diff --git a/supernote/server/config.py b/supernote/server/config.py index b36e0b5d..ee4f257b 100644 --- a/supernote/server/config.py +++ b/supernote/server/config.py @@ -52,6 +52,46 @@ class Config(BaseConfig): code_generation_options = [TO_DICT_ADD_OMIT_NONE_FLAG] # type: ignore[list-item] +@dataclass +class WebhookEndpointConfig(DataClassYAMLMixin): + """A single outbound webhook subscriber.""" + + url: str = "" + """The HTTP(S) endpoint that receives the webhook POST request.""" + + secret: str = "" + """Shared secret used to HMAC-SHA256 sign the payload. + + Never logged. Sent to receivers via the `X-Supernote-Signature` header so + they can verify the payload was not tampered with in transit. + """ + + events: list[str] = field(default_factory=list) + """Event names this endpoint wants to receive, e.g. `note.sync_completed`.""" + + class Config(BaseConfig): + omit_none = True + code_generation_options = [TO_DICT_ADD_OMIT_NONE_FLAG] # type: ignore[list-item] + + +@dataclass +class WebhookConfig(DataClassYAMLMixin): + """Outbound webhook dispatch configuration.""" + + enabled: bool = False + """Whether outbound webhook dispatch is enabled. + + Env Var: `SUPERNOTE_WEBHOOKS_ENABLED` + """ + + endpoints: list[WebhookEndpointConfig] = field(default_factory=list) + """Configured webhook subscribers.""" + + class Config(BaseConfig): + omit_none = True + code_generation_options = [TO_DICT_ADD_OMIT_NONE_FLAG] # type: ignore[list-item] + + @dataclass class ServerConfig(DataClassYAMLMixin): host: str = "0.0.0.0" @@ -155,6 +195,9 @@ class ServerConfig(DataClassYAMLMixin): Env Var: `SUPERNOTE_METRICS_PATH` """ + webhooks: WebhookConfig = field(default_factory=WebhookConfig) + """Outbound webhook dispatch configuration.""" + @property def configured_base_url(self) -> str | None: """Get the explicitly configured base URL, or None if unset. @@ -344,6 +387,12 @@ def load( config.metrics_path = metrics_path logger.info(f"Using SUPERNOTE_METRICS_PATH: {config.metrics_path}") + if os.getenv("SUPERNOTE_WEBHOOKS_ENABLED"): + config.webhooks.enabled = _get_bool_env( + "SUPERNOTE_WEBHOOKS_ENABLED", config.webhooks.enabled + ) + logger.info(f"Webhooks Enabled: {config.webhooks.enabled}") + if config.trace_log_file is None: config.trace_log_file = str( Path(config.storage_dir) / "system" / "trace.log" diff --git a/supernote/server/services/webhook.py b/supernote/server/services/webhook.py new file mode 100644 index 00000000..616b9390 --- /dev/null +++ b/supernote/server/services/webhook.py @@ -0,0 +1,145 @@ +"""Outbound webhook dispatch for server events. + +Listens on the internal :class:`~supernote.server.events.LocalEventBus` and, +for every configured endpoint subscribed to a given event name, sends a +signed HTTP POST notification. Delivery is best-effort: a slow or failing +endpoint is logged and skipped, it never raises back into the caller and +never blocks or delays the operation that triggered the event (e.g. a +device sync request). +""" + +import asyncio +import hashlib +import hmac +import json +import logging +import time +from typing import Any + +import aiohttp + +from ..config import WebhookConfig, WebhookEndpointConfig +from ..events import Event, LocalEventBus, NoteUpdatedEvent + +logger = logging.getLogger(__name__) + +# Name of the event dispatched when a .note file finishes syncing to the +# server. Fired from the same point the processing pipeline is enqueued +# (see NoteUpdatedEvent), i.e. as soon as the device's upload completes. +EVENT_NOTE_SYNC_COMPLETED = "note.sync_completed" + +SIGNATURE_HEADER = "X-Supernote-Signature" +EVENT_HEADER = "X-Supernote-Event" + +REQUEST_TIMEOUT_SECONDS = 10.0 +MAX_ATTEMPTS = 3 +RETRY_BACKOFF_SECONDS = 1.0 + + +def sign_payload(secret: str, body: bytes) -> str: + """Compute the `sha256=` HMAC signature for a webhook payload body.""" + digest = hmac.new(secret.encode("utf-8"), body, hashlib.sha256).hexdigest() + return f"sha256={digest}" + + +def verify_signature(secret: str, body: bytes, signature: str) -> bool: + """Verify a `sha256=` signature against `body` using constant-time comparison.""" + expected = sign_payload(secret, body) + return hmac.compare_digest(expected, signature) + + +class WebhookService: + """Dispatches signed outbound webhook notifications for subscribed events.""" + + def __init__(self, config: WebhookConfig, event_bus: LocalEventBus) -> None: + self.config = config + self.event_bus = event_bus + + def start(self) -> None: + """Subscribe to the events that can trigger webhook dispatch.""" + if not self.config.enabled or not self.config.endpoints: + logger.debug("Webhook dispatch disabled or no endpoints configured.") + return + self.event_bus.subscribe(NoteUpdatedEvent, self._handle_note_updated) + logger.info( + f"WebhookService started with {len(self.config.endpoints)} endpoint(s)." + ) + + async def _handle_note_updated(self, event: Event) -> None: + """Translate a NoteUpdatedEvent into a note.sync_completed webhook.""" + if not isinstance(event, NoteUpdatedEvent): + return + payload = { + "event": EVENT_NOTE_SYNC_COMPLETED, + "timestamp": int(time.time()), + "user_id": event.user_id, + "file": { + "id": event.file_id, + "path": event.file_path, + }, + } + await self.dispatch(EVENT_NOTE_SYNC_COMPLETED, payload) + + async def dispatch(self, event_name: str, payload: dict[str, Any]) -> None: + """Send `payload` to every configured endpoint subscribed to `event_name`. + + Never raises. Endpoints are notified concurrently and independently; + one endpoint failing has no effect on delivery to the others. + """ + endpoints = [ep for ep in self.config.endpoints if event_name in ep.events] + if not endpoints: + return + await asyncio.gather( + *(self._send(endpoint, event_name, payload) for endpoint in endpoints) + ) + + async def _send( + self, + endpoint: WebhookEndpointConfig, + event_name: str, + payload: dict[str, Any], + ) -> None: + """POST `payload` to a single endpoint, retrying transient failures. + + Client errors (4xx) are not retried. Network errors, timeouts, and + server errors (5xx) are retried up to MAX_ATTEMPTS with a linear + backoff. All failures are logged (without the secret or signature) + and swallowed. + """ + body = json.dumps(payload).encode("utf-8") + headers = {"Content-Type": "application/json", EVENT_HEADER: event_name} + if endpoint.secret: + headers[SIGNATURE_HEADER] = sign_payload(endpoint.secret, body) + + timeout = aiohttp.ClientTimeout(total=REQUEST_TIMEOUT_SECONDS) + for attempt in range(1, MAX_ATTEMPTS + 1): + try: + async with aiohttp.ClientSession(timeout=timeout) as session: + async with session.post( + endpoint.url, data=body, headers=headers + ) as response: + if response.status < 400: + return + if response.status < 500: + logger.warning( + f"Webhook {event_name} to {endpoint.url} rejected " + f"with status {response.status}; not retrying." + ) + return + logger.warning( + f"Webhook {event_name} to {endpoint.url} failed with " + f"status {response.status} (attempt {attempt}/{MAX_ATTEMPTS})." + ) + except (aiohttp.ClientError, TimeoutError) as e: + logger.warning( + f"Webhook {event_name} to {endpoint.url} failed: " + f"{type(e).__name__} (attempt {attempt}/{MAX_ATTEMPTS})." + ) + + if attempt < MAX_ATTEMPTS: + await asyncio.sleep(RETRY_BACKOFF_SECONDS * attempt) + + logger.error( + f"Webhook {event_name} to {endpoint.url} failed after " + f"{MAX_ATTEMPTS} attempts; giving up." + ) diff --git a/tests/server/services/test_webhook.py b/tests/server/services/test_webhook.py new file mode 100644 index 00000000..60ed416a --- /dev/null +++ b/tests/server/services/test_webhook.py @@ -0,0 +1,293 @@ +"""Tests for the outbound webhook dispatcher service.""" + +import asyncio +import hashlib +import hmac +import json +from typing import Any + +import pytest + +from supernote.server.config import WebhookConfig, WebhookEndpointConfig +from supernote.server.events import LocalEventBus, NoteUpdatedEvent +from supernote.server.services import webhook as webhook_module +from supernote.server.services.webhook import ( + EVENT_HEADER, + EVENT_NOTE_SYNC_COMPLETED, + MAX_ATTEMPTS, + SIGNATURE_HEADER, + WebhookService, + sign_payload, + verify_signature, +) + + +class _FakeResponse: + def __init__(self, status: int) -> None: + self.status = status + + async def __aenter__(self) -> "_FakeResponse": + return self + + async def __aexit__(self, *exc: object) -> bool: + return False + + +class _FakeResponseCtx: + """Async context manager returned by `session.post(...)`.""" + + def __init__(self, result: int | BaseException) -> None: + self._result = result + + async def __aenter__(self) -> _FakeResponse: + if isinstance(self._result, BaseException): + raise self._result + return _FakeResponse(self._result) + + async def __aexit__(self, *exc: object) -> bool: + return False + + +class _FakeTransport: + """Records every POST made across (possibly many) fake sessions. + + `results` is consumed in order, one entry per POST call. If it runs out, + subsequent calls default to a 200 response. + """ + + def __init__(self) -> None: + self._results: list[int | BaseException] = [] + self.calls: list[dict[str, Any]] = [] + + def queue_results(self, *results: int | BaseException) -> None: + """Set the canned per-POST results, consumed in order.""" + self._results = list(results) + + def new_session(self, **_: Any) -> "_FakeClientSession": + return _FakeClientSession(self) + + def next_result(self) -> int | BaseException: + if self._results: + return self._results.pop(0) + return 200 + + +class _FakeClientSession: + def __init__(self, transport: _FakeTransport) -> None: + self._transport = transport + + async def __aenter__(self) -> "_FakeClientSession": + return self + + async def __aexit__(self, *exc: object) -> bool: + return False + + def post(self, url: str, data: bytes, headers: dict[str, str]) -> _FakeResponseCtx: + self._transport.calls.append({"url": url, "data": data, "headers": headers}) + return _FakeResponseCtx(self._transport.next_result()) + + +@pytest.fixture +def fake_transport(monkeypatch: pytest.MonkeyPatch) -> _FakeTransport: + """Patch aiohttp.ClientSession used by the webhook module with a fake.""" + transport = _FakeTransport() + monkeypatch.setattr( + webhook_module.aiohttp, "ClientSession", lambda **kw: transport.new_session() + ) + # Retries use a real (but zeroed-out) asyncio.sleep so the suite stays + # fast without mocking asyncio.sleep globally. + monkeypatch.setattr(webhook_module, "RETRY_BACKOFF_SECONDS", 0.0) + return transport + + +async def _drain_event_loop(iterations: int = 20) -> None: + """Let scheduled asyncio tasks (e.g. LocalEventBus.publish's fire-and-forget + handler task, and the child tasks asyncio.gather creates for each webhook + endpoint) actually run to completion before asserting on their effects. + """ + for _ in range(iterations): + await asyncio.sleep(0) + + +def _endpoint(**overrides: Any) -> WebhookEndpointConfig: + defaults: dict[str, Any] = { + "url": "https://example.com/webhook", + "secret": "top-secret", + "events": [EVENT_NOTE_SYNC_COMPLETED], + } + defaults.update(overrides) + return WebhookEndpointConfig(**defaults) + + +def test_sign_payload_is_deterministic_hmac_sha256() -> None: + body = b'{"event": "note.sync_completed"}' + signature = sign_payload("my-secret", body) + + expected = hmac.new(b"my-secret", body, hashlib.sha256).hexdigest() + assert signature == f"sha256={expected}" + + +def test_verify_signature_accepts_matching_signature() -> None: + body = b"payload-bytes" + signature = sign_payload("shared-secret", body) + assert verify_signature("shared-secret", body, signature) is True + + +def test_verify_signature_rejects_tampered_body() -> None: + body = b"payload-bytes" + signature = sign_payload("shared-secret", body) + assert verify_signature("shared-secret", b"tampered-bytes", signature) is False + + +def test_verify_signature_rejects_wrong_secret() -> None: + body = b"payload-bytes" + signature = sign_payload("shared-secret", body) + assert verify_signature("wrong-secret", body, signature) is False + + +async def test_dispatch_sends_signed_post_to_subscribed_endpoint( + fake_transport: _FakeTransport, +) -> None: + endpoint = _endpoint() + config = WebhookConfig(enabled=True, endpoints=[endpoint]) + service = WebhookService(config, LocalEventBus()) + + payload = {"event": EVENT_NOTE_SYNC_COMPLETED, "user_id": 1} + await service.dispatch(EVENT_NOTE_SYNC_COMPLETED, payload) + + assert len(fake_transport.calls) == 1 + call = fake_transport.calls[0] + assert call["url"] == endpoint.url + assert json.loads(call["data"]) == payload + assert call["headers"][EVENT_HEADER] == EVENT_NOTE_SYNC_COMPLETED + assert call["headers"][SIGNATURE_HEADER] == sign_payload( + endpoint.secret, call["data"] + ) + + +async def test_dispatch_skips_endpoints_not_subscribed_to_event( + fake_transport: _FakeTransport, +) -> None: + endpoint = _endpoint(events=["task.created"]) + config = WebhookConfig(enabled=True, endpoints=[endpoint]) + service = WebhookService(config, LocalEventBus()) + + await service.dispatch(EVENT_NOTE_SYNC_COMPLETED, {"event": "note.sync_completed"}) + + assert fake_transport.calls == [] + + +async def test_dispatch_omits_signature_header_when_no_secret( + fake_transport: _FakeTransport, +) -> None: + endpoint = _endpoint(secret="") + config = WebhookConfig(enabled=True, endpoints=[endpoint]) + service = WebhookService(config, LocalEventBus()) + + await service.dispatch(EVENT_NOTE_SYNC_COMPLETED, {"event": "note.sync_completed"}) + + assert SIGNATURE_HEADER not in fake_transport.calls[0]["headers"] + + +async def test_note_updated_event_triggers_note_sync_completed_webhook( + fake_transport: _FakeTransport, +) -> None: + endpoint = _endpoint() + config = WebhookConfig(enabled=True, endpoints=[endpoint]) + event_bus = LocalEventBus() + service = WebhookService(config, event_bus) + service.start() + + await event_bus.publish( + NoteUpdatedEvent(file_id=42, user_id=7, file_path="Note/Daily/2026-08-04.note") + ) + # LocalEventBus dispatches via asyncio.create_task; let it run. + await _drain_event_loop() + + assert len(fake_transport.calls) == 1 + body = json.loads(fake_transport.calls[0]["data"]) + assert body["event"] == EVENT_NOTE_SYNC_COMPLETED + assert body["user_id"] == 7 + assert body["file"] == {"id": 42, "path": "Note/Daily/2026-08-04.note"} + assert "timestamp" in body + + +async def test_start_does_not_subscribe_when_disabled( + fake_transport: _FakeTransport, +) -> None: + config = WebhookConfig(enabled=False, endpoints=[_endpoint()]) + event_bus = LocalEventBus() + service = WebhookService(config, event_bus) + service.start() + + await event_bus.publish( + NoteUpdatedEvent(file_id=1, user_id=1, file_path="Note/a.note") + ) + await _drain_event_loop() + + assert fake_transport.calls == [] + + +async def test_send_does_not_raise_on_repeated_server_errors( + fake_transport: _FakeTransport, +) -> None: + fake_transport.queue_results(500, 500, 500) + endpoint = _endpoint() + config = WebhookConfig(enabled=True, endpoints=[endpoint]) + service = WebhookService(config, LocalEventBus()) + + # Must not raise even though every attempt fails. + await service.dispatch(EVENT_NOTE_SYNC_COMPLETED, {"event": "x"}) + + assert len(fake_transport.calls) == MAX_ATTEMPTS + + +async def test_send_does_not_raise_on_timeout(fake_transport: _FakeTransport) -> None: + fake_transport.queue_results(TimeoutError(), TimeoutError(), TimeoutError()) + endpoint = _endpoint() + config = WebhookConfig(enabled=True, endpoints=[endpoint]) + service = WebhookService(config, LocalEventBus()) + + await service.dispatch(EVENT_NOTE_SYNC_COMPLETED, {"event": "x"}) + + assert len(fake_transport.calls) == MAX_ATTEMPTS + + +async def test_send_does_not_retry_client_errors( + fake_transport: _FakeTransport, +) -> None: + fake_transport.queue_results(404, 200, 200) + endpoint = _endpoint() + config = WebhookConfig(enabled=True, endpoints=[endpoint]) + service = WebhookService(config, LocalEventBus()) + + await service.dispatch(EVENT_NOTE_SYNC_COMPLETED, {"event": "x"}) + + assert len(fake_transport.calls) == 1 + + +async def test_send_recovers_after_transient_failure( + fake_transport: _FakeTransport, +) -> None: + fake_transport.queue_results(500, 200) + endpoint = _endpoint() + config = WebhookConfig(enabled=True, endpoints=[endpoint]) + service = WebhookService(config, LocalEventBus()) + + await service.dispatch(EVENT_NOTE_SYNC_COMPLETED, {"event": "x"}) + + assert len(fake_transport.calls) == 2 + + +async def test_dispatch_does_not_leak_secret_or_full_signature_in_logs( + fake_transport: _FakeTransport, caplog: pytest.LogCaptureFixture +) -> None: + fake_transport.queue_results(500, 500, 500) + endpoint = _endpoint(secret="super-secret-value") + config = WebhookConfig(enabled=True, endpoints=[endpoint]) + service = WebhookService(config, LocalEventBus()) + + with caplog.at_level("WARNING"): + await service.dispatch(EVENT_NOTE_SYNC_COMPLETED, {"event": "x"}) + + assert "super-secret-value" not in caplog.text diff --git a/tests/server/test_config.py b/tests/server/test_config.py index 9f97b8a5..902f58d4 100644 --- a/tests/server/test_config.py +++ b/tests/server/test_config.py @@ -125,6 +125,54 @@ def test_configured_base_url_from_env(tmp_path: Path) -> None: assert config.base_url == "https://env.example.com" +def test_server_config_webhooks_default_disabled(tmp_path: Path) -> None: + """Webhooks are disabled with no endpoints by default.""" + config_dir = tmp_path / "config" + config = ServerConfig.load(config_dir) + + assert config.webhooks.enabled is False + assert config.webhooks.endpoints == [] + + +def test_server_config_load_webhooks_from_file(tmp_path: Path) -> None: + """Webhook endpoints are parsed from the config file.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + config_file = config_dir / "config.yaml" + + data = { + "webhooks": { + "enabled": True, + "endpoints": [ + { + "url": "https://n8n.homelab.local/webhook/supernote", + "secret": "hmac-secret-key", + "events": ["note.sync_completed"], + } + ], + } + } + with open(config_file, "w") as f: + yaml.safe_dump(data, f) + + config = ServerConfig.load(config_dir) + + assert config.webhooks.enabled is True + assert len(config.webhooks.endpoints) == 1 + endpoint = config.webhooks.endpoints[0] + assert endpoint.url == "https://n8n.homelab.local/webhook/supernote" + assert endpoint.secret == "hmac-secret-key" + assert endpoint.events == ["note.sync_completed"] + + +def test_server_config_webhooks_enabled_env_var_override(tmp_path: Path) -> None: + """SUPERNOTE_WEBHOOKS_ENABLED overrides the config file value.""" + config_dir = tmp_path / "config" + with patch.dict(os.environ, {"SUPERNOTE_WEBHOOKS_ENABLED": "true"}): + config = ServerConfig.load(config_dir) + assert config.webhooks.enabled is True + + def test_server_config_proxy_env_vars(tmp_path: Path) -> None: """Test that proxy configuration can be set via environment variables.""" config_dir = tmp_path / "config" From 6d45bdd70945ee0eeb991f1d541ceaa228ba2bea Mon Sep 17 00:00:00 2001 From: proscar87 Date: Tue, 4 Aug 2026 21:11:37 -0600 Subject: [PATCH 2/3] fix(server): isolate unexpected webhook errors from the caller asyncio.gather without return_exceptions propagates the first exception, so an unexpected error while dispatching to one endpoint would surface in the sync flow that triggered it, contradicting the documented never-raises contract. Collect the results instead and log anything unexpected. Co-Authored-By: Claude Fable 5 --- supernote/server/services/webhook.py | 11 +++++++++-- tests/server/services/test_webhook.py | 22 ++++++++++++++++++++++ 2 files changed, 31 insertions(+), 2 deletions(-) diff --git a/supernote/server/services/webhook.py b/supernote/server/services/webhook.py index 616b9390..b0e511f5 100644 --- a/supernote/server/services/webhook.py +++ b/supernote/server/services/webhook.py @@ -89,9 +89,16 @@ async def dispatch(self, event_name: str, payload: dict[str, Any]) -> None: endpoints = [ep for ep in self.config.endpoints if event_name in ep.events] if not endpoints: return - await asyncio.gather( - *(self._send(endpoint, event_name, payload) for endpoint in endpoints) + results = await asyncio.gather( + *(self._send(endpoint, event_name, payload) for endpoint in endpoints), + return_exceptions=True, ) + for endpoint, result in zip(endpoints, results, strict=True): + if isinstance(result, BaseException): + logger.error( + f"Webhook {event_name} to {endpoint.url} raised " + f"{type(result).__name__}: {result}" + ) async def _send( self, diff --git a/tests/server/services/test_webhook.py b/tests/server/services/test_webhook.py index 60ed416a..fe659672 100644 --- a/tests/server/services/test_webhook.py +++ b/tests/server/services/test_webhook.py @@ -291,3 +291,25 @@ async def test_dispatch_does_not_leak_secret_or_full_signature_in_logs( await service.dispatch(EVENT_NOTE_SYNC_COMPLETED, {"event": "x"}) assert "super-secret-value" not in caplog.text + + +async def test_dispatch_never_raises_on_unexpected_error( + caplog: pytest.LogCaptureFixture, +) -> None: + """An unexpected error while sending must not reach the caller.""" + config = WebhookConfig( + enabled=True, + endpoints=[ + WebhookEndpointConfig( + url="https://example.com/hook", + events=[EVENT_NOTE_SYNC_COMPLETED], + ), + ], + ) + service = WebhookService(config, LocalEventBus()) + + # A payload that cannot be serialized fails outside the retried network + # errors, so it exercises the unexpected-error path. + await service.dispatch(EVENT_NOTE_SYNC_COMPLETED, {"bad": object()}) + + assert "TypeError" in caplog.text From 820c3abe3f9c6df6660009fe7f6aa19a23928393 Mon Sep 17 00:00:00 2001 From: proscar87 Date: Wed, 5 Aug 2026 13:01:24 -0600 Subject: [PATCH 3/3] refactor(server): address webhook dispatcher review feedback Addresses allenporter's review comments on PR #217: - Extract the note.sync_completed webhook body into a documented NoteSyncCompletedPayload/WebhookFileVO dataclass pair in supernote/models/webhook.py, instead of an inline dict. - Drop the internal, per-user user_id from the outbound payload; it's a private DB identifier that shouldn't leave the server. - Remove verify_signature(), which was dead code (this feature only implements the sending side; nothing in-repo calls it). - An endpoint with no `events` configured now defaults to receiving every event, instead of receiving nothing. - Rewrite the test suite to POST against a real aiohttp.web.Application test receiver (matching the fake-server pattern already used elsewhere in the suite, e.g. tests/server/conftest.py) instead of mocking aiohttp.ClientSession, and switch monkeypatch to unittest.mock.patch throughout. Co-Authored-By: Claude Fable 5 --- config-example.yaml | 2 + supernote/models/__init__.py | 1 + supernote/models/webhook.py | 41 +++++ supernote/server/config.py | 5 +- supernote/server/services/webhook.py | 40 +++-- tests/models/test_webhook.py | 29 ++++ tests/server/services/test_webhook.py | 240 ++++++++++++-------------- 7 files changed, 213 insertions(+), 145 deletions(-) create mode 100644 supernote/models/webhook.py create mode 100644 tests/models/test_webhook.py diff --git a/config-example.yaml b/config-example.yaml index 9d21419a..a8fc9ecf 100644 --- a/config-example.yaml +++ b/config-example.yaml @@ -58,5 +58,7 @@ auth: # # endpoint in the `X-Supernote-Signature: sha256=` header so it # # can verify the request came from this server and was not altered. # secret: "CHANGE_ME_TO_A_SECURE_RANDOM_STRING" +# # Event names this endpoint wants to receive. Omit (or leave empty) +# # to receive every event. # events: # - "note.sync_completed" diff --git a/supernote/models/__init__.py b/supernote/models/__init__.py index 541fdd65..bfaa0d78 100644 --- a/supernote/models/__init__.py +++ b/supernote/models/__init__.py @@ -12,4 +12,5 @@ "summary", "system", "user", + "webhook", ] diff --git a/supernote/models/webhook.py b/supernote/models/webhook.py new file mode 100644 index 00000000..ccb58ad9 --- /dev/null +++ b/supernote/models/webhook.py @@ -0,0 +1,41 @@ +"""Data models for outbound webhook payloads. + +See :mod:`supernote.server.services.webhook` for the dispatcher that builds +and signs these payloads before sending them to configured endpoints. +""" + +from dataclasses import dataclass + +from mashumaro.mixins.json import DataClassJSONMixin + + +@dataclass +class WebhookFileVO(DataClassJSONMixin): + """File metadata included in a `note.sync_completed` webhook payload.""" + + id: int + """The internal ID of the synced file.""" + + path: str + """The full path of the file, relative to the user's root directory.""" + + +@dataclass +class NoteSyncCompletedPayload(DataClassJSONMixin): + """Payload sent for the `note.sync_completed` outbound webhook event. + + Corresponds to `EVENT_NOTE_SYNC_COMPLETED` in + `supernote.server.services.webhook`. + + Intentionally excludes the internal `user_id` (a private, per-user + database identifier) -- see that module for discussion. + """ + + event: str + """The webhook event name. Always `note.sync_completed` today.""" + + timestamp: int + """Unix timestamp (seconds) of when the event was dispatched.""" + + file: WebhookFileVO + """Metadata about the file that finished syncing.""" diff --git a/supernote/server/config.py b/supernote/server/config.py index ee4f257b..47092020 100644 --- a/supernote/server/config.py +++ b/supernote/server/config.py @@ -67,7 +67,10 @@ class WebhookEndpointConfig(DataClassYAMLMixin): """ events: list[str] = field(default_factory=list) - """Event names this endpoint wants to receive, e.g. `note.sync_completed`.""" + """Event names this endpoint wants to receive, e.g. `note.sync_completed`. + + If empty (the default), the endpoint receives every event. + """ class Config(BaseConfig): omit_none = True diff --git a/supernote/server/services/webhook.py b/supernote/server/services/webhook.py index b0e511f5..9b704f23 100644 --- a/supernote/server/services/webhook.py +++ b/supernote/server/services/webhook.py @@ -18,6 +18,8 @@ import aiohttp +from supernote.models.webhook import NoteSyncCompletedPayload, WebhookFileVO + from ..config import WebhookConfig, WebhookEndpointConfig from ..events import Event, LocalEventBus, NoteUpdatedEvent @@ -42,12 +44,6 @@ def sign_payload(secret: str, body: bytes) -> str: return f"sha256={digest}" -def verify_signature(secret: str, body: bytes, signature: str) -> bool: - """Verify a `sha256=` signature against `body` using constant-time comparison.""" - expected = sign_payload(secret, body) - return hmac.compare_digest(expected, signature) - - class WebhookService: """Dispatches signed outbound webhook notifications for subscribed events.""" @@ -69,24 +65,30 @@ async def _handle_note_updated(self, event: Event) -> None: """Translate a NoteUpdatedEvent into a note.sync_completed webhook.""" if not isinstance(event, NoteUpdatedEvent): return - payload = { - "event": EVENT_NOTE_SYNC_COMPLETED, - "timestamp": int(time.time()), - "user_id": event.user_id, - "file": { - "id": event.file_id, - "path": event.file_path, - }, - } - await self.dispatch(EVENT_NOTE_SYNC_COMPLETED, payload) + # Note: intentionally does not include event.user_id. It's an + # internal database identifier, not something we want to hand to + # third-party endpoints. See PR #217 for discussion on how + # per-user webhook delivery might work in a future iteration. + payload = NoteSyncCompletedPayload( + event=EVENT_NOTE_SYNC_COMPLETED, + timestamp=int(time.time()), + file=WebhookFileVO(id=event.file_id, path=event.file_path), + ) + await self.dispatch(EVENT_NOTE_SYNC_COMPLETED, payload.to_dict()) async def dispatch(self, event_name: str, payload: dict[str, Any]) -> None: """Send `payload` to every configured endpoint subscribed to `event_name`. - Never raises. Endpoints are notified concurrently and independently; - one endpoint failing has no effect on delivery to the others. + An endpoint with no `events` configured is subscribed to everything + (see `WebhookEndpointConfig.events`). Never raises: endpoints are + notified concurrently and independently, and one endpoint failing has + no effect on delivery to the others. """ - endpoints = [ep for ep in self.config.endpoints if event_name in ep.events] + endpoints = [ + ep + for ep in self.config.endpoints + if not ep.events or event_name in ep.events + ] if not endpoints: return results = await asyncio.gather( diff --git a/tests/models/test_webhook.py b/tests/models/test_webhook.py new file mode 100644 index 00000000..ea93b38a --- /dev/null +++ b/tests/models/test_webhook.py @@ -0,0 +1,29 @@ +"""Tests for outbound webhook payload models.""" + +from supernote.models.webhook import NoteSyncCompletedPayload, WebhookFileVO + + +def test_note_sync_completed_payload_serializes_to_dict() -> None: + payload = NoteSyncCompletedPayload( + event="note.sync_completed", + timestamp=1730000000, + file=WebhookFileVO(id=42, path="Note/Daily/2026-08-04.note"), + ) + + assert payload.to_dict() == { + "event": "note.sync_completed", + "timestamp": 1730000000, + "file": {"id": 42, "path": "Note/Daily/2026-08-04.note"}, + } + + +def test_note_sync_completed_payload_excludes_user_id() -> None: + """The payload has no field for the internal, per-user database ID.""" + payload = NoteSyncCompletedPayload( + event="note.sync_completed", + timestamp=1730000000, + file=WebhookFileVO(id=1, path="a.note"), + ) + + assert "user_id" not in payload.to_dict() + assert not hasattr(payload, "user_id") diff --git a/tests/server/services/test_webhook.py b/tests/server/services/test_webhook.py index fe659672..4a5cc3b3 100644 --- a/tests/server/services/test_webhook.py +++ b/tests/server/services/test_webhook.py @@ -4,13 +4,16 @@ import hashlib import hmac import json +from collections.abc import Generator from typing import Any +from unittest.mock import patch import pytest +from aiohttp import web +from pytest_aiohttp import AiohttpClient from supernote.server.config import WebhookConfig, WebhookEndpointConfig from supernote.server.events import LocalEventBus, NoteUpdatedEvent -from supernote.server.services import webhook as webhook_module from supernote.server.services.webhook import ( EVENT_HEADER, EVENT_NOTE_SYNC_COMPLETED, @@ -18,86 +21,72 @@ SIGNATURE_HEADER, WebhookService, sign_payload, - verify_signature, ) -class _FakeResponse: - def __init__(self, status: int) -> None: - self.status = status +class _StallResponse: + """Sentinel response: the receiver never replies to the request. - async def __aenter__(self) -> "_FakeResponse": - return self - - async def __aexit__(self, *exc: object) -> bool: - return False - - -class _FakeResponseCtx: - """Async context manager returned by `session.post(...)`.""" - - def __init__(self, result: int | BaseException) -> None: - self._result = result + Used to exercise the dispatcher's own client-side timeout, rather than + a status code. + """ - async def __aenter__(self) -> _FakeResponse: - if isinstance(self._result, BaseException): - raise self._result - return _FakeResponse(self._result) - async def __aexit__(self, *exc: object) -> bool: - return False +STALL = _StallResponse() -class _FakeTransport: - """Records every POST made across (possibly many) fake sessions. +class WebhookReceiver: + """A real aiohttp server standing in for a third-party webhook receiver. - `results` is consumed in order, one entry per POST call. If it runs out, - subsequent calls default to a 200 response. + Records every POST it gets and lets tests script the response to each + call in turn (an HTTP status, or STALL to force a client timeout). + Using a real server -- rather than mocking aiohttp.ClientSession -- + exercises the dispatcher's actual request construction, connection + handling, and timeout/retry behavior end-to-end. """ def __init__(self) -> None: - self._results: list[int | BaseException] = [] + self.url = "" self.calls: list[dict[str, Any]] = [] - - def queue_results(self, *results: int | BaseException) -> None: - """Set the canned per-POST results, consumed in order.""" - self._results = list(results) - - def new_session(self, **_: Any) -> "_FakeClientSession": - return _FakeClientSession(self) - - def next_result(self) -> int | BaseException: - if self._results: - return self._results.pop(0) - return 200 - - -class _FakeClientSession: - def __init__(self, transport: _FakeTransport) -> None: - self._transport = transport - - async def __aenter__(self) -> "_FakeClientSession": - return self - - async def __aexit__(self, *exc: object) -> bool: - return False - - def post(self, url: str, data: bytes, headers: dict[str, str]) -> _FakeResponseCtx: - self._transport.calls.append({"url": url, "data": data, "headers": headers}) - return _FakeResponseCtx(self._transport.next_result()) + self._responses: list[int | _StallResponse] = [] + + def queue_responses(self, *responses: int | _StallResponse) -> None: + """Set the canned per-POST responses, consumed in order (default: 200).""" + self._responses = list(responses) + + async def handle(self, request: web.Request) -> web.Response: + body = await request.read() + self.calls.append( + {"url": str(request.url), "data": body, "headers": dict(request.headers)} + ) + response = self._responses.pop(0) if self._responses else 200 + if isinstance(response, _StallResponse): + # Outlives the (patched, short) client request timeout used by + # every test that queues this, so the client sees a real + # asyncio.TimeoutError instead of a canned status. + await asyncio.sleep(2) + return web.Response(status=200) + return web.Response(status=response) @pytest.fixture -def fake_transport(monkeypatch: pytest.MonkeyPatch) -> _FakeTransport: - """Patch aiohttp.ClientSession used by the webhook module with a fake.""" - transport = _FakeTransport() - monkeypatch.setattr( - webhook_module.aiohttp, "ClientSession", lambda **kw: transport.new_session() - ) - # Retries use a real (but zeroed-out) asyncio.sleep so the suite stays - # fast without mocking asyncio.sleep globally. - monkeypatch.setattr(webhook_module, "RETRY_BACKOFF_SECONDS", 0.0) - return transport +async def webhook_receiver(aiohttp_client: AiohttpClient) -> WebhookReceiver: + """Spin up a real aiohttp server to receive webhook POSTs in tests.""" + receiver = WebhookReceiver() + app = web.Application() + app.router.add_post("/webhook", receiver.handle) + client = await aiohttp_client(app) + receiver.url = str(client.make_url("/webhook")) + return receiver + + +@pytest.fixture(autouse=True) +def no_retry_backoff() -> Generator[None]: + """Use a real (but zeroed-out) asyncio.sleep for retry backoff so the + suite stays fast without mocking asyncio.sleep globally. + """ + with patch("supernote.server.services.webhook.RETRY_BACKOFF_SECONDS", 0.0): + yield async def _drain_event_loop(iterations: int = 20) -> None: @@ -109,9 +98,9 @@ async def _drain_event_loop(iterations: int = 20) -> None: await asyncio.sleep(0) -def _endpoint(**overrides: Any) -> WebhookEndpointConfig: +def _endpoint(url: str, **overrides: Any) -> WebhookEndpointConfig: defaults: dict[str, Any] = { - "url": "https://example.com/webhook", + "url": url, "secret": "top-secret", "events": [EVENT_NOTE_SYNC_COMPLETED], } @@ -127,36 +116,18 @@ def test_sign_payload_is_deterministic_hmac_sha256() -> None: assert signature == f"sha256={expected}" -def test_verify_signature_accepts_matching_signature() -> None: - body = b"payload-bytes" - signature = sign_payload("shared-secret", body) - assert verify_signature("shared-secret", body, signature) is True - - -def test_verify_signature_rejects_tampered_body() -> None: - body = b"payload-bytes" - signature = sign_payload("shared-secret", body) - assert verify_signature("shared-secret", b"tampered-bytes", signature) is False - - -def test_verify_signature_rejects_wrong_secret() -> None: - body = b"payload-bytes" - signature = sign_payload("shared-secret", body) - assert verify_signature("wrong-secret", body, signature) is False - - async def test_dispatch_sends_signed_post_to_subscribed_endpoint( - fake_transport: _FakeTransport, + webhook_receiver: WebhookReceiver, ) -> None: - endpoint = _endpoint() + endpoint = _endpoint(webhook_receiver.url) config = WebhookConfig(enabled=True, endpoints=[endpoint]) service = WebhookService(config, LocalEventBus()) payload = {"event": EVENT_NOTE_SYNC_COMPLETED, "user_id": 1} await service.dispatch(EVENT_NOTE_SYNC_COMPLETED, payload) - assert len(fake_transport.calls) == 1 - call = fake_transport.calls[0] + assert len(webhook_receiver.calls) == 1 + call = webhook_receiver.calls[0] assert call["url"] == endpoint.url assert json.loads(call["data"]) == payload assert call["headers"][EVENT_HEADER] == EVENT_NOTE_SYNC_COMPLETED @@ -166,33 +137,47 @@ async def test_dispatch_sends_signed_post_to_subscribed_endpoint( async def test_dispatch_skips_endpoints_not_subscribed_to_event( - fake_transport: _FakeTransport, + webhook_receiver: WebhookReceiver, +) -> None: + endpoint = _endpoint(webhook_receiver.url, events=["task.created"]) + config = WebhookConfig(enabled=True, endpoints=[endpoint]) + service = WebhookService(config, LocalEventBus()) + + await service.dispatch(EVENT_NOTE_SYNC_COMPLETED, {"event": "note.sync_completed"}) + + assert webhook_receiver.calls == [] + + +async def test_dispatch_endpoint_with_no_events_receives_everything( + webhook_receiver: WebhookReceiver, ) -> None: - endpoint = _endpoint(events=["task.created"]) + """An endpoint with no `events` configured defaults to receiving all events.""" + endpoint = _endpoint(webhook_receiver.url, events=[]) config = WebhookConfig(enabled=True, endpoints=[endpoint]) service = WebhookService(config, LocalEventBus()) await service.dispatch(EVENT_NOTE_SYNC_COMPLETED, {"event": "note.sync_completed"}) + await service.dispatch("task.created", {"event": "task.created"}) - assert fake_transport.calls == [] + assert len(webhook_receiver.calls) == 2 async def test_dispatch_omits_signature_header_when_no_secret( - fake_transport: _FakeTransport, + webhook_receiver: WebhookReceiver, ) -> None: - endpoint = _endpoint(secret="") + endpoint = _endpoint(webhook_receiver.url, secret="") config = WebhookConfig(enabled=True, endpoints=[endpoint]) service = WebhookService(config, LocalEventBus()) await service.dispatch(EVENT_NOTE_SYNC_COMPLETED, {"event": "note.sync_completed"}) - assert SIGNATURE_HEADER not in fake_transport.calls[0]["headers"] + assert SIGNATURE_HEADER not in webhook_receiver.calls[0]["headers"] async def test_note_updated_event_triggers_note_sync_completed_webhook( - fake_transport: _FakeTransport, + webhook_receiver: WebhookReceiver, ) -> None: - endpoint = _endpoint() + endpoint = _endpoint(webhook_receiver.url) config = WebhookConfig(enabled=True, endpoints=[endpoint]) event_bus = LocalEventBus() service = WebhookService(config, event_bus) @@ -204,18 +189,19 @@ async def test_note_updated_event_triggers_note_sync_completed_webhook( # LocalEventBus dispatches via asyncio.create_task; let it run. await _drain_event_loop() - assert len(fake_transport.calls) == 1 - body = json.loads(fake_transport.calls[0]["data"]) + assert len(webhook_receiver.calls) == 1 + body = json.loads(webhook_receiver.calls[0]["data"]) assert body["event"] == EVENT_NOTE_SYNC_COMPLETED - assert body["user_id"] == 7 assert body["file"] == {"id": 42, "path": "Note/Daily/2026-08-04.note"} assert "timestamp" in body + # The internal, per-user database ID must never leave the server. + assert "user_id" not in body async def test_start_does_not_subscribe_when_disabled( - fake_transport: _FakeTransport, + webhook_receiver: WebhookReceiver, ) -> None: - config = WebhookConfig(enabled=False, endpoints=[_endpoint()]) + config = WebhookConfig(enabled=False, endpoints=[_endpoint(webhook_receiver.url)]) event_bus = LocalEventBus() service = WebhookService(config, event_bus) service.start() @@ -225,65 +211,68 @@ async def test_start_does_not_subscribe_when_disabled( ) await _drain_event_loop() - assert fake_transport.calls == [] + assert webhook_receiver.calls == [] async def test_send_does_not_raise_on_repeated_server_errors( - fake_transport: _FakeTransport, + webhook_receiver: WebhookReceiver, ) -> None: - fake_transport.queue_results(500, 500, 500) - endpoint = _endpoint() + webhook_receiver.queue_responses(500, 500, 500) + endpoint = _endpoint(webhook_receiver.url) config = WebhookConfig(enabled=True, endpoints=[endpoint]) service = WebhookService(config, LocalEventBus()) # Must not raise even though every attempt fails. await service.dispatch(EVENT_NOTE_SYNC_COMPLETED, {"event": "x"}) - assert len(fake_transport.calls) == MAX_ATTEMPTS + assert len(webhook_receiver.calls) == MAX_ATTEMPTS -async def test_send_does_not_raise_on_timeout(fake_transport: _FakeTransport) -> None: - fake_transport.queue_results(TimeoutError(), TimeoutError(), TimeoutError()) - endpoint = _endpoint() +async def test_send_does_not_raise_on_timeout( + webhook_receiver: WebhookReceiver, +) -> None: + webhook_receiver.queue_responses(STALL, STALL, STALL) + endpoint = _endpoint(webhook_receiver.url) config = WebhookConfig(enabled=True, endpoints=[endpoint]) service = WebhookService(config, LocalEventBus()) - await service.dispatch(EVENT_NOTE_SYNC_COMPLETED, {"event": "x"}) + with patch("supernote.server.services.webhook.REQUEST_TIMEOUT_SECONDS", 0.05): + await service.dispatch(EVENT_NOTE_SYNC_COMPLETED, {"event": "x"}) - assert len(fake_transport.calls) == MAX_ATTEMPTS + assert len(webhook_receiver.calls) == MAX_ATTEMPTS async def test_send_does_not_retry_client_errors( - fake_transport: _FakeTransport, + webhook_receiver: WebhookReceiver, ) -> None: - fake_transport.queue_results(404, 200, 200) - endpoint = _endpoint() + webhook_receiver.queue_responses(404, 200, 200) + endpoint = _endpoint(webhook_receiver.url) config = WebhookConfig(enabled=True, endpoints=[endpoint]) service = WebhookService(config, LocalEventBus()) await service.dispatch(EVENT_NOTE_SYNC_COMPLETED, {"event": "x"}) - assert len(fake_transport.calls) == 1 + assert len(webhook_receiver.calls) == 1 async def test_send_recovers_after_transient_failure( - fake_transport: _FakeTransport, + webhook_receiver: WebhookReceiver, ) -> None: - fake_transport.queue_results(500, 200) - endpoint = _endpoint() + webhook_receiver.queue_responses(500, 200) + endpoint = _endpoint(webhook_receiver.url) config = WebhookConfig(enabled=True, endpoints=[endpoint]) service = WebhookService(config, LocalEventBus()) await service.dispatch(EVENT_NOTE_SYNC_COMPLETED, {"event": "x"}) - assert len(fake_transport.calls) == 2 + assert len(webhook_receiver.calls) == 2 async def test_dispatch_does_not_leak_secret_or_full_signature_in_logs( - fake_transport: _FakeTransport, caplog: pytest.LogCaptureFixture + webhook_receiver: WebhookReceiver, caplog: pytest.LogCaptureFixture ) -> None: - fake_transport.queue_results(500, 500, 500) - endpoint = _endpoint(secret="super-secret-value") + webhook_receiver.queue_responses(500, 500, 500) + endpoint = _endpoint(webhook_receiver.url, secret="super-secret-value") config = WebhookConfig(enabled=True, endpoints=[endpoint]) service = WebhookService(config, LocalEventBus()) @@ -301,7 +290,7 @@ async def test_dispatch_never_raises_on_unexpected_error( enabled=True, endpoints=[ WebhookEndpointConfig( - url="https://example.com/hook", + url="https://example.invalid/hook", events=[EVENT_NOTE_SYNC_COMPLETED], ), ], @@ -309,7 +298,8 @@ async def test_dispatch_never_raises_on_unexpected_error( service = WebhookService(config, LocalEventBus()) # A payload that cannot be serialized fails outside the retried network - # errors, so it exercises the unexpected-error path. + # errors, so it exercises the unexpected-error path. No real request is + # ever attempted, so this doesn't need the fake webhook receiver. await service.dispatch(EVENT_NOTE_SYNC_COMPLETED, {"bad": object()}) assert "TypeError" in caplog.text