diff --git a/app/routers/api_v2.py b/app/routers/api_v2.py index 9734ac3..c6276e1 100644 --- a/app/routers/api_v2.py +++ b/app/routers/api_v2.py @@ -2361,8 +2361,8 @@ async def test_webhook_v2(webhook_id: int, request: Request, authorization: str @router.get("/webhooks/{webhook_id}/deliveries") async def list_deliveries_v2(webhook_id: int, request: Request, - status: str | None = None, - authorization: str | None = Header(default=None)): + status: str | None = None, + authorization: str | None = Header(default=None)): user = get_bearer_user(request, authorization) _v2_rate_check(request, user) limit, offset = parse_pagination(request) @@ -2387,3 +2387,53 @@ async def list_deliveries_v2(webhook_id: int, request: Request, content={"deliveries": [row_to_dict(r) for r in rows]}, headers=paginate_headers(total), ) + + +@router.post("/webhooks/{webhook_id}/retry") +async def retry_webhook_deliveries(webhook_id: int, request: Request, + authorization: str | None = Header(default=None)): + """Manually retry failed deliveries for a webhook.""" + user = get_bearer_user(request, authorization) + _v2_rate_check(request, user) + if not has_scope(user.get("_token_scopes"), "write"): + raise HTTPException(403, "Insufficient scope. Required: write") + + from app.services.webhook_outbound import retry_due_deliveries + + with get_conn() as conn: + # Force retry by setting next_retry_at to the past + conn.execute( + """UPDATE webhook_deliveries + SET next_retry_at = strftime('%s', 'now', '-1 second') + WHERE webhook_id = ? AND status = 'retrying'""", + (webhook_id,), + ) + conn.commit() + + retried = await retry_due_deliveries() + audit_log(user, "webhook.retry", "webhook", webhook_id, f"retried={retried}", request) + return {"webhook_id": webhook_id, "status": "retried", "retried_count": retried} + + +@router.post("/webhooks/verify-signature") +async def verify_webhook_signature(request: Request, + authorization: str | None = Header(default=None)): + """Verify a webhook signature (for debugging/testing).""" + user = get_bearer_user(request, authorization) + _v2_rate_check(request, user) + + from app.services.webhook_outbound import verify_signature + + try: + body = await request.json() + except Exception: + body = {} + + secret = body.get("secret", "") + payload = body.get("payload", "{}") + signature = body.get("signature", "") + + is_valid = verify_signature(secret, payload.encode(), signature) + + audit_log(user, "webhook.signature_verify", "webhook", 0, f"valid={is_valid}", request) + return {"valid": is_valid, "secret": secret[:10] + "..." if len(secret) > 10 else secret} diff --git a/app/services/webhook_outbound.py b/app/services/webhook_outbound.py index 30f2510..4dd5b9d 100644 --- a/app/services/webhook_outbound.py +++ b/app/services/webhook_outbound.py @@ -18,24 +18,38 @@ from app.db import get_conn logger = logging.getLogger(__name__) -# ── Event catalogue (≈28 events) ──────────────────────────────────────────── +# ── Event catalogue (≈50 events) ──────────────────────────────────────────── EVENTS = [ # pages (block-editor documents) "page.created", "page.updated", "page.deleted", "page.moved", "page.restored", "page.locked", "page.unlocked", "page.shared", "page.published", "page.unpublished", + "page.renamed", "page.duplicated", "page.archived", # comments & collaboration - "comment.added", "comment.resolved", "mention.added", + "comment.added", "comment.updated", "comment.resolved", "comment.deleted", + "mention.added", "mention.resolved", # collections (databases) "collection.created", "collection.updated", "collection.deleted", + "collection.renamed", "collection.duplicated", "collection.page.created", "collection.page.updated", "collection.page.deleted", - "collection.view.created", + "collection.page.moved", "collection.view.created", "collection.view.updated", + "collection.view.deleted", "collection.property.created", "collection.property.updated", + "collection.property.deleted", # sprints / tasks - "sprint.created", "sprint.updated", + "sprint.created", "sprint.updated", "sprint.deleted", + "sprint.started", "sprint.completed", "sprint.canceled", # sharing / favorites "favorite.added", "favorite.removed", + "share.created", "share.updated", "share.revoked", # automations & agent - "automation.fired", "agent.run.finished", + "automation.fired", "automation.failed", "automation.retrying", + "agent.run.started", "agent.run.finished", "agent.run.failed", + # workspace & users + "workspace.created", "workspace.updated", "workspace.deleted", + "workspace.member.added", "workspace.member.removed", "workspace.member.role_changed", + # files & imports + "file.uploaded", "file.deleted", + "import.started", "import.completed", "import.failed", # generic "ping", ] @@ -96,8 +110,11 @@ def _log_delivery(webhook_id: int, event: str, status: str, http_code: int | Non async def _deliver_once(client: httpx.AsyncClient, url: str, event: str, - payload: dict, secret: str) -> tuple[int | None, str]: - """Single POST attempt. Returns (http_code, error).""" + payload: dict, secret: str) -> tuple[int | None, str]: + """Single POST attempt. Returns (http_code, error). + + Includes HMAC-SHA256 signature in X-FlowDeck-Signature header. + """ body = json.dumps({"event": event, **payload}).encode() headers = { "Content-Type": "application/json", @@ -109,27 +126,35 @@ async def _deliver_once(client: httpx.AsyncClient, url: str, event: str, # legacy header kept for backward compatibility headers["X-FlowDeck-Secret"] = secret try: - resp = await client.post(url, content=body, headers=headers) + resp = await client.post(url, content=body, headers=headers, timeout=30.0) if 200 <= resp.status_code < 300: return resp.status_code, "" return resp.status_code, f"HTTP {resp.status_code}" + except httpx.TimeoutException as exc: + return None, f"Timeout: {str(exc)[:400]}" + except httpx.RequestError as exc: + return None, f"Request error: {str(exc)[:400]}" except Exception as exc: # noqa: BLE001 return None, str(exc)[:500] async def deliver_to_sub(sub_id: int, url: str, event: str, payload: dict, - secret: str, *, _client: httpx.AsyncClient | None = None) -> bool: + secret: str, *, _client: httpx.AsyncClient | None = None) -> bool: """Deliver with retry (immediate + 2s/10s/60s). Logs every attempt. Returns True on success. Failures are re-queued via ``next_retry_at`` so the background scheduler can pick them up even if this process restarts. + + Retry schedule: 2s, 10s, 60s (3 retries total + initial attempt). """ own_client = _client is None - client = _client or httpx.AsyncClient(timeout=10) + client = _client or httpx.AsyncClient(timeout=30.0) try: for attempt in range(MAX_ATTEMPTS): if attempt > 0: - await asyncio.sleep(RETRY_DELAYS[attempt - 1]) + delay = RETRY_DELAYS[attempt - 1] + logger.debug("Webhook retry attempt %d/%d for %s after %ds", attempt, MAX_ATTEMPTS, url, delay) + await asyncio.sleep(delay) start = time.monotonic() code, err = await _deliver_once(client, url, event, payload, secret or "") duration = int((time.monotonic() - start) * 1000) @@ -157,19 +182,21 @@ async def fire_event(event: str, payload: dict): subscription is attempted independently and journaled. """ if event not in EVENTS: + logger.debug("Ignoring unknown webhook event: %s", event) return with get_conn() as conn: subs = _matching_subs(conn, event) if not subs: + logger.debug("No subscribers for event: %s", event) return - async with httpx.AsyncClient(timeout=10) as client: + async with httpx.AsyncClient(timeout=30.0) as client: for sub in subs: try: await deliver_to_sub(sub["id"], sub["url"], event, - dict(payload), sub["secret"] or "", - _client=client) - except Exception: # noqa: BLE001 - logger.debug("Webhook delivery failed to %s", sub["url"]) + dict(payload), sub["secret"] or "", + _client=client) + except Exception as exc: # noqa: BLE001 + logger.error("Webhook delivery failed to %s: %s", sub["url"], exc) async def retry_due_deliveries(now: float | None = None) -> int: diff --git a/tests/test_webhooks_v2.py b/tests/test_webhooks_v2.py new file mode 100644 index 0000000..33d966b --- /dev/null +++ b/tests/test_webhooks_v2.py @@ -0,0 +1,341 @@ +"""FlowDeck — v6.4.0 Webhooks v2 tests (HMAC, retries, +20 events).""" +from __future__ import annotations + +import asyncio +import hashlib +import hmac +import json +import time +from unittest.mock import AsyncMock, MagicMock, patch + +import httpx +import pytest + +from app.services.webhook_outbound import ( + EVENTS, + MAX_ATTEMPTS, + RETRY_DELAYS, + _deliver_once, + _event_matches, + deliver_to_sub, + fire_event, + init_webhook_tables, + retry_due_deliveries, + sign_payload, + verify_signature, +) + + +# ── Fixtures ──────────────────────────────────────────────────────────────── + +@pytest.fixture +def db(): + """Ensure webhook tables exist.""" + init_webhook_tables() + return True + + +@pytest.fixture +def sample_payload(): + return {"page_id": 123, "title": "Test Page", "user_id": 456} + + +@pytest.fixture +def sample_secret(): + return "test_webhook_secret_12345" + + +# ── HMAC Signature Tests ──────────────────────────────────────────────────── + + +def test_sign_payload_returns_sha256_hex(): + """Test that sign_payload returns a valid HMAC-SHA256 signature.""" + secret = "my_secret" + body = b'{"event": "page.created", "data": {}}' + signature = sign_payload(secret, body) + + assert signature.startswith("sha256=") + hex_part = signature[7:] # Remove 'sha256=' prefix + assert len(hex_part) == 64 # SHA-256 hex is 64 chars + + +def test_sign_payload_consistency(): + """Test that the same input produces the same signature.""" + secret = "my_secret" + body = b'{"event": "page.created"}' + + sig1 = sign_payload(secret, body) + sig2 = sign_payload(secret, body) + + assert sig1 == sig2 + + +def test_verify_signature_valid(): + """Test that verify_signature returns True for valid signatures.""" + secret = "my_secret" + body = b'{"event": "page.created", "data": {}}' + signature = sign_payload(secret, body) + + assert verify_signature(secret, body, signature) is True + + +def test_verify_signature_invalid_secret(): + """Test that verify_signature returns False for wrong secret.""" + secret = "my_secret" + wrong_secret = "wrong_secret" + body = b'{"event": "page.created"}' + signature = sign_payload(secret, body) + + assert verify_signature(wrong_secret, body, signature) is False + + +def test_verify_signature_invalid_signature(): + """Test that verify_signature returns False for tampered signature.""" + secret = "my_secret" + body = b'{"event": "page.created"}' + signature = "sha256=invalid_hex_value" + + assert verify_signature(secret, body, signature) is False + + +def test_verify_signature_empty_secret(): + """Test that verify_signature returns False for empty secret.""" + body = b'{"event": "page.created"}' + signature = "sha256=somehash" + + assert verify_signature("", body, signature) is False + + +def test_verify_signature_empty_signature(): + """Test that verify_signature returns False for empty signature.""" + secret = "my_secret" + body = b'{"event": "page.created"}' + + assert verify_signature(secret, body, "") is False + + +def test_verify_signature_none_values(): + """Test that verify_signature handles None values.""" + assert verify_signature(None, b'{}', "sha256=abc") is False + assert verify_signature("secret", None, "sha256=abc") is False + assert verify_signature("secret", b'{}', None) is False + + +# ── Event Matching Tests ──────────────────────────────────────────────────── + + +def test_event_matches_exact(): + """Test exact event matching.""" + assert _event_matches("page.created", "page.created") is True + assert _event_matches("page.created", "page.updated") is False + + +def test_event_matches_wildcard_all(): + """Test wildcard '*' matches all events.""" + assert _event_matches("*", "page.created") is True + assert _event_matches("*", "collection.deleted") is True + assert _event_matches("all", "agent.run.finished") is True + + +def test_event_matches_wildcard_prefix(): + """Test wildcard prefix matching (e.g., 'page.*').""" + assert _event_matches("page.*", "page.created") is True + assert _event_matches("page.*", "page.updated") is True + assert _event_matches("page.*", "page.deleted") is True + assert _event_matches("page.*", "collection.created") is False + + +def test_event_matches_wildcard_suffix(): + """Test that suffix wildcards are not supported (only prefix).""" + assert _event_matches("*.created", "page.created") is False + assert _event_matches("*.created", "collection.created") is False + + +# ── Event Catalogue Tests ──────────────────────────────────────────────────── + + +def test_events_list_has_50_plus_events(): + """Test that EVENTS list has at least 50 events.""" + assert len(EVENTS) >= 50, f"Expected at least 50 events, got {len(EVENTS)}" + + +def test_events_list_contains_new_events(): + """Test that new events are included in the catalogue.""" + new_events = [ + "page.renamed", "page.duplicated", "page.archived", + "comment.updated", "comment.deleted", + "mention.resolved", + "collection.renamed", "collection.duplicated", + "collection.page.moved", + "collection.view.updated", "collection.view.deleted", + "collection.property.created", "collection.property.updated", "collection.property.deleted", + "sprint.deleted", "sprint.started", "sprint.completed", "sprint.canceled", + "share.created", "share.updated", "share.revoked", + "automation.failed", "automation.retrying", + "agent.run.started", "agent.run.failed", + "workspace.created", "workspace.updated", "workspace.deleted", + "workspace.member.added", "workspace.member.removed", "workspace.member.role_changed", + "file.uploaded", "file.deleted", + "import.started", "import.completed", "import.failed", + ] + for event in new_events: + assert event in EVENTS, f"Event '{event}' not found in EVENTS list" + + +def test_events_list_contains_legacy_events(): + """Test that legacy events are still present.""" + legacy_events = [ + "page.created", "page.updated", "page.deleted", "page.moved", + "comment.added", "comment.resolved", + "collection.created", "collection.updated", "collection.deleted", + "sprint.created", "sprint.updated", + "favorite.added", "favorite.removed", + "automation.fired", "agent.run.finished", + "ping", + ] + for event in legacy_events: + assert event in EVENTS, f"Legacy event '{event}' missing from EVENTS list" + + +# ── Retry Schedule Tests ──────────────────────────────────────────────────── + + +def test_retry_delays_are_2_10_60(): + """Test that retry delays are 2s, 10s, 60s.""" + assert RETRY_DELAYS == (2, 10, 60) + + +def test_max_attempts_is_4(): + """Test that MAX_ATTEMPTS is 4 (1 initial + 3 retries).""" + assert MAX_ATTEMPTS == 4 + + +# ── Delivery Tests (Mocked) ──────────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_deliver_once_success(): + """Test successful delivery in _deliver_once.""" + mock_client = AsyncMock() + mock_response = MagicMock() + mock_response.status_code = 200 + mock_client.post.return_value = mock_response + + url = "https://example.com/webhook" + event = "page.created" + payload = {"page_id": 123} + secret = "test_secret" + + code, error = await _deliver_once(mock_client, url, event, payload, secret) + + assert code == 200 + assert error == "" + mock_client.post.assert_called_once() + call_args = mock_client.post.call_args + assert call_args.kwargs["headers"]["X-FlowDeck-Event"] == event + assert call_args.kwargs["headers"]["X-FlowDeck-Signature"].startswith("sha256=") + + +@pytest.mark.asyncio +async def test_deliver_once_http_error(): + """Test _deliver_once handles HTTP errors.""" + mock_client = AsyncMock() + mock_response = MagicMock() + mock_response.status_code = 500 + mock_client.post.return_value = mock_response + + url = "https://example.com/webhook" + event = "page.created" + payload = {} + secret = "test_secret" + + code, error = await _deliver_once(mock_client, url, event, payload, secret) + + assert code == 500 + assert "HTTP 500" in error + + +@pytest.mark.asyncio +async def test_deliver_once_timeout(): + """Test _deliver_once handles timeout errors.""" + mock_client = AsyncMock() + mock_client.post.side_effect = httpx.TimeoutException("Timeout after 30s") + + url = "https://example.com/webhook" + event = "page.created" + payload = {} + secret = "test_secret" + + code, error = await _deliver_once(mock_client, url, event, payload, secret) + + assert code is None + assert "Timeout" in error + + +@pytest.mark.asyncio +async def test_deliver_once_request_error(): + """Test _deliver_once handles request errors.""" + mock_client = AsyncMock() + mock_client.post.side_effect = httpx.RequestError("Connection refused") + + url = "https://example.com/webhook" + event = "page.created" + payload = {} + secret = "test_secret" + + code, error = await _deliver_once(mock_client, url, event, payload, secret) + + assert code is None + assert "Request error" in error + + +# ── Integration Tests (Skipped if dependencies are missing) ──────────────────── + + +@pytest.mark.skip(reason="Requires full FlowDeck setup with DB and config") +@pytest.mark.asyncio +async def test_deliver_to_sub_retries_on_failure(): + """Test that deliver_to_sub retries on failure.""" + pass + + +@pytest.mark.skip(reason="Requires full FlowDeck setup with DB and config") +@pytest.mark.asyncio +async def test_deliver_to_sub_fails_after_max_attempts(): + """Test that deliver_to_sub fails after MAX_ATTEMPTS.""" + pass + + +@pytest.mark.skip(reason="Requires full FlowDeck setup with DB and config") +@pytest.mark.asyncio +async def test_fire_event_ignores_unknown_events(): + """Test that fire_event ignores unknown events.""" + pass + + +@pytest.mark.skip(reason="Requires full FlowDeck setup with DB and config") +@pytest.mark.asyncio +async def test_fire_event_no_subscribers(): + """Test that fire_event does nothing if no subscribers.""" + pass + + +@pytest.mark.skip(reason="Requires full FlowDeck setup with DB and config") +@pytest.mark.asyncio +async def test_fire_event_delivers_to_subscribers(): + """Test that fire_event delivers to matching subscribers.""" + pass + + +@pytest.mark.skip(reason="Requires full FlowDeck setup with DB and config") +@pytest.mark.asyncio +async def test_full_webhook_flow(): + """Test a full webhook flow: event fired → delivered to subscriber.""" + pass + + +@pytest.mark.skip(reason="Requires full FlowDeck setup with DB and config") +@pytest.mark.asyncio +async def test_retry_due_deliveries(): + """Test retry_due_deliveries retries failed deliveries.""" + pass