From d61a05bad0bee98abec84ef05ff05da57270e1db Mon Sep 17 00:00:00 2001 From: Christian Krakau-Louis Date: Sat, 23 May 2026 18:06:29 +0200 Subject: [PATCH] feat: add outbound webhook event framework --- backend/alembic/env.py | 1 + ...b2c3d4e5f6a_add_webhook_event_framework.py | 137 ++++++ backend/app/api/api_v1/api.py | 2 + backend/app/api/api_v1/endpoints/settings.py | 7 +- backend/app/api/api_v1/endpoints/webhook.py | 17 + backend/app/api/api_v1/endpoints/webhooks.py | 228 +++++++++ backend/app/main.py | 17 + backend/app/models/webhook.py | 66 +++ backend/app/services/alert_rules.py | 34 ++ backend/app/services/webhook_events.py | 465 ++++++++++++++++++ backend/app/templates/settings.html | 299 +++++++++++ backend/app/tests/conftest.py | 1 + backend/app/tests/test_webhooks.py | 308 ++++++++++++ docs/milestones.md | 2 +- docs/reference/api.md | 46 +- docs/reference/database.md | 43 ++ docs/user_guide/settings.md | 15 + 17 files changed, 1674 insertions(+), 14 deletions(-) create mode 100644 backend/alembic/versions/1b2c3d4e5f6a_add_webhook_event_framework.py create mode 100644 backend/app/api/api_v1/endpoints/webhooks.py create mode 100644 backend/app/models/webhook.py create mode 100644 backend/app/services/webhook_events.py create mode 100644 backend/app/tests/test_webhooks.py diff --git a/backend/alembic/env.py b/backend/alembic/env.py index d030b14..1aa3456 100644 --- a/backend/alembic/env.py +++ b/backend/alembic/env.py @@ -29,6 +29,7 @@ import app.models.mail_source_import # noqa: E402, F401 import app.models.report # noqa: E402, F401 import app.models.setting # noqa: E402, F401 import app.models.user # noqa: E402, F401 +import app.models.webhook # noqa: E402, F401 # Import all models so that autogenerate can detect them from app.core.database import Base # noqa: E402 diff --git a/backend/alembic/versions/1b2c3d4e5f6a_add_webhook_event_framework.py b/backend/alembic/versions/1b2c3d4e5f6a_add_webhook_event_framework.py new file mode 100644 index 0000000..f1ee110 --- /dev/null +++ b/backend/alembic/versions/1b2c3d4e5f6a_add_webhook_event_framework.py @@ -0,0 +1,137 @@ +"""add webhook event framework + +Revision ID: 1b2c3d4e5f6a +Revises: 0a1b2c3d4e5f +Create Date: 2026-05-23 00:00:00.000000 + +""" + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = "1b2c3d4e5f6a" +down_revision: Union[str, Sequence[str], None] = "0a1b2c3d4e5f" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """Create outbound webhook endpoint and delivery tables.""" + op.create_table( + "webhook_endpoints", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("name", sa.String(length=120), nullable=False), + sa.Column("url", sa.Text(), nullable=False), + sa.Column("secret", sa.Text(), nullable=False), + sa.Column("event_types", sa.Text(), nullable=False), + sa.Column("enabled", sa.Boolean(), nullable=False), + sa.Column("max_attempts", sa.Integer(), nullable=False), + sa.Column("timeout_seconds", sa.Integer(), nullable=False), + sa.Column("created_at", sa.DateTime(), nullable=False), + sa.Column("updated_at", sa.DateTime(), nullable=True), + sa.Column("last_success_at", sa.DateTime(), nullable=True), + sa.Column("last_failure_at", sa.DateTime(), nullable=True), + sa.Column("failure_count", sa.Integer(), nullable=False), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index(op.f("ix_webhook_endpoints_id"), "webhook_endpoints", ["id"]) + op.create_index(op.f("ix_webhook_endpoints_enabled"), "webhook_endpoints", ["enabled"]) + op.create_index(op.f("ix_webhook_endpoints_created_at"), "webhook_endpoints", ["created_at"]) + op.create_index( + op.f("ix_webhook_endpoints_last_success_at"), + "webhook_endpoints", + ["last_success_at"], + ) + op.create_index( + op.f("ix_webhook_endpoints_last_failure_at"), + "webhook_endpoints", + ["last_failure_at"], + ) + op.create_index( + "ix_webhook_endpoints_enabled_events", + "webhook_endpoints", + ["enabled", "event_types"], + ) + + op.create_table( + "webhook_deliveries", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("endpoint_id", sa.Integer(), nullable=False), + sa.Column("event_type", sa.String(length=80), nullable=False), + sa.Column("payload", sa.Text(), nullable=False), + sa.Column("idempotency_key", sa.String(length=160), nullable=False), + sa.Column("status", sa.String(length=24), nullable=False), + sa.Column("attempt_count", sa.Integer(), nullable=False), + sa.Column("max_attempts", sa.Integer(), nullable=False), + sa.Column("next_attempt_at", sa.DateTime(), nullable=False), + sa.Column("last_attempt_at", sa.DateTime(), nullable=True), + sa.Column("delivered_at", sa.DateTime(), nullable=True), + sa.Column("last_status_code", sa.Integer(), nullable=True), + sa.Column("last_error", sa.Text(), nullable=True), + sa.Column("response_excerpt", sa.Text(), nullable=True), + sa.Column("created_at", sa.DateTime(), nullable=False), + sa.Column("updated_at", sa.DateTime(), nullable=True), + sa.ForeignKeyConstraint(["endpoint_id"], ["webhook_endpoints.id"]), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index(op.f("ix_webhook_deliveries_id"), "webhook_deliveries", ["id"]) + op.create_index( + op.f("ix_webhook_deliveries_endpoint_id"), "webhook_deliveries", ["endpoint_id"] + ) + op.create_index(op.f("ix_webhook_deliveries_event_type"), "webhook_deliveries", ["event_type"]) + op.create_index( + op.f("ix_webhook_deliveries_idempotency_key"), "webhook_deliveries", ["idempotency_key"] + ) + op.create_index(op.f("ix_webhook_deliveries_status"), "webhook_deliveries", ["status"]) + op.create_index( + op.f("ix_webhook_deliveries_next_attempt_at"), + "webhook_deliveries", + ["next_attempt_at"], + ) + op.create_index( + op.f("ix_webhook_deliveries_last_attempt_at"), + "webhook_deliveries", + ["last_attempt_at"], + ) + op.create_index( + op.f("ix_webhook_deliveries_delivered_at"), "webhook_deliveries", ["delivered_at"] + ) + op.create_index(op.f("ix_webhook_deliveries_created_at"), "webhook_deliveries", ["created_at"]) + op.create_index( + "ix_webhook_delivery_endpoint_idempotency", + "webhook_deliveries", + ["endpoint_id", "idempotency_key"], + unique=True, + ) + op.create_index( + "ix_webhook_delivery_due", + "webhook_deliveries", + ["status", "next_attempt_at"], + ) + + +def downgrade() -> None: + """Drop outbound webhook endpoint and delivery tables.""" + op.drop_index("ix_webhook_delivery_due", table_name="webhook_deliveries") + op.drop_index("ix_webhook_delivery_endpoint_idempotency", table_name="webhook_deliveries") + op.drop_index(op.f("ix_webhook_deliveries_created_at"), table_name="webhook_deliveries") + op.drop_index(op.f("ix_webhook_deliveries_delivered_at"), table_name="webhook_deliveries") + op.drop_index(op.f("ix_webhook_deliveries_last_attempt_at"), table_name="webhook_deliveries") + op.drop_index(op.f("ix_webhook_deliveries_next_attempt_at"), table_name="webhook_deliveries") + op.drop_index(op.f("ix_webhook_deliveries_status"), table_name="webhook_deliveries") + op.drop_index(op.f("ix_webhook_deliveries_idempotency_key"), table_name="webhook_deliveries") + op.drop_index(op.f("ix_webhook_deliveries_event_type"), table_name="webhook_deliveries") + op.drop_index(op.f("ix_webhook_deliveries_endpoint_id"), table_name="webhook_deliveries") + op.drop_index(op.f("ix_webhook_deliveries_id"), table_name="webhook_deliveries") + op.drop_table("webhook_deliveries") + + op.drop_index("ix_webhook_endpoints_enabled_events", table_name="webhook_endpoints") + op.drop_index(op.f("ix_webhook_endpoints_last_failure_at"), table_name="webhook_endpoints") + op.drop_index(op.f("ix_webhook_endpoints_last_success_at"), table_name="webhook_endpoints") + op.drop_index(op.f("ix_webhook_endpoints_created_at"), table_name="webhook_endpoints") + op.drop_index(op.f("ix_webhook_endpoints_enabled"), table_name="webhook_endpoints") + op.drop_index(op.f("ix_webhook_endpoints_id"), table_name="webhook_endpoints") + op.drop_table("webhook_endpoints") diff --git a/backend/app/api/api_v1/api.py b/backend/app/api/api_v1/api.py index f49a9a2..dd41591 100644 --- a/backend/app/api/api_v1/api.py +++ b/backend/app/api/api_v1/api.py @@ -15,6 +15,7 @@ from app.api.api_v1.endpoints import ( stats, tls_reports, webhook, + webhooks, ) api_router = APIRouter() @@ -34,3 +35,4 @@ api_router.include_router(mail_sources.router, prefix="/mail-sources", tags=["ma api_router.include_router(settings.router, prefix="/settings", tags=["settings"]) api_router.include_router(tls_reports.router, prefix="/tls-reports", tags=["tls-reports"]) api_router.include_router(webhook.router, prefix="/webhook", tags=["webhook"]) +api_router.include_router(webhooks.router, prefix="/webhooks", tags=["webhooks"]) diff --git a/backend/app/api/api_v1/endpoints/settings.py b/backend/app/api/api_v1/endpoints/settings.py index b3d7180..ffc5d61 100644 --- a/backend/app/api/api_v1/endpoints/settings.py +++ b/backend/app/api/api_v1/endpoints/settings.py @@ -29,7 +29,11 @@ from app.services.alert_history import ( record_alert_config_change, record_alert_evaluation, ) -from app.services.alert_rules import evaluate_alert_rules, send_current_alerts +from app.services.alert_rules import ( + enqueue_alert_webhook_events, + evaluate_alert_rules, + send_current_alerts, +) from app.services.notifications import send_notification from app.services.summary_notifications import build_summary, send_summary_notification @@ -496,6 +500,7 @@ async def evaluate_notification_alerts( _seed_defaults(db) alerts = evaluate_alert_rules(db) record_alert_evaluation(db, alerts) + enqueue_alert_webhook_events(db, alerts) return {"alerts": alerts} diff --git a/backend/app/api/api_v1/endpoints/webhook.py b/backend/app/api/api_v1/endpoints/webhook.py index ae45451..31b2cfe 100644 --- a/backend/app/api/api_v1/endpoints/webhook.py +++ b/backend/app/api/api_v1/endpoints/webhook.py @@ -17,6 +17,7 @@ from app.core.redaction import sanitize_for_log from app.services.dmarc_parser import DMARCParser from app.services.report_persistence import report_exists, save_parsed_report from app.services.report_store import ReportStore +from app.services.webhook_events import EVENT_REPORT_IMPORTED, enqueue_webhook_event logger = logging.getLogger(__name__) router = APIRouter() @@ -69,6 +70,22 @@ def _store_report(db: Session, store: ReportStore, report: Dict[str, Any]) -> st if report_id and (store.has_report(domain, report_id) or report_exists(db, domain, report_id)): return "duplicate" save_parsed_report(db, report) + try: + enqueue_webhook_event( + db, + event_type=EVENT_REPORT_IMPORTED, + payload={ + "domain": domain, + "report_id": report_id, + "org_name": report.get("org_name"), + "begin_date": report.get("begin_date"), + "end_date": report.get("end_date"), + "records": len(report.get("records") or []), + }, + idempotency_key=f"{EVENT_REPORT_IMPORTED}:{domain}:{report_id or 'unknown'}", + ) + except Exception as exc: # pylint: disable=broad-exception-caught + logger.warning("Failed to queue report-import webhook event: %s", sanitize_for_log(exc)) store.add_report(report) return "imported" diff --git a/backend/app/api/api_v1/endpoints/webhooks.py b/backend/app/api/api_v1/endpoints/webhooks.py new file mode 100644 index 0000000..aac61ae --- /dev/null +++ b/backend/app/api/api_v1/endpoints/webhooks.py @@ -0,0 +1,228 @@ +"""Admin endpoints for outbound webhook event delivery.""" + +from typing import Any, Dict, List, Optional + +from fastapi import APIRouter, Depends, HTTPException, Query, status +from pydantic import BaseModel +from sqlalchemy.orm import Session + +from app.core.database import get_db +from app.core.security import require_admin_auth +from app.models.webhook import WebhookDelivery, WebhookEndpoint +from app.services.webhook_events import ( + SUPPORTED_EVENT_TYPES, + create_webhook_endpoint, + deliver_due_webhooks, + delivery_to_dict, + endpoint_to_dict, + queue_test_webhook, + update_webhook_endpoint, +) + +router = APIRouter() + + +class WebhookEndpointCreate(BaseModel): + """Create payload for outbound webhook endpoints.""" + + name: str + url: str + secret: Optional[str] = None + event_types: List[str] = ["*"] + enabled: bool = True + max_attempts: int = 5 + timeout_seconds: int = 10 + + +class WebhookEndpointUpdate(BaseModel): + """Update payload for outbound webhook endpoints.""" + + name: Optional[str] = None + url: Optional[str] = None + secret: Optional[str] = None + event_types: Optional[List[str]] = None + enabled: Optional[bool] = None + max_attempts: Optional[int] = None + timeout_seconds: Optional[int] = None + + +class WebhookEndpointResponse(BaseModel): + """API-safe webhook endpoint metadata.""" + + id: int + name: str + url: str + event_types: List[str] + enabled: bool + max_attempts: int + timeout_seconds: int + created_at: Optional[str] + updated_at: Optional[str] + last_success_at: Optional[str] + last_failure_at: Optional[str] + failure_count: int + secret_configured: bool + url_encrypted: bool + secret: Optional[str] = None + + +class WebhookEndpointListResponse(BaseModel): + """List response for outbound webhook endpoints.""" + + endpoints: List[WebhookEndpointResponse] + supported_event_types: List[str] + + +class WebhookDeliveryResponse(BaseModel): + """API-safe webhook delivery metadata.""" + + id: int + endpoint_id: int + event_type: str + idempotency_key: str + status: str + attempt_count: int + max_attempts: int + next_attempt_at: Optional[str] + last_attempt_at: Optional[str] + delivered_at: Optional[str] + last_status_code: Optional[int] + last_error: Optional[str] + response_excerpt: Optional[str] + created_at: Optional[str] + updated_at: Optional[str] + + +class WebhookDeliveryListResponse(BaseModel): + """List response for outbound webhook deliveries.""" + + deliveries: List[WebhookDeliveryResponse] + + +class WebhookTestResponse(BaseModel): + """Response for a webhook test delivery.""" + + delivery: WebhookDeliveryResponse + + +@router.get("", response_model=WebhookEndpointListResponse) +async def list_webhook_endpoints( + db: Session = Depends(get_db), + _auth: dict = Depends(require_admin_auth), +) -> Dict[str, Any]: + """Return configured outbound webhook endpoints.""" + endpoints = db.query(WebhookEndpoint).order_by(WebhookEndpoint.created_at.desc()).all() + return { + "endpoints": [endpoint_to_dict(endpoint) for endpoint in endpoints], + "supported_event_types": SUPPORTED_EVENT_TYPES, + } + + +@router.post("", response_model=WebhookEndpointResponse) +async def create_webhook( + payload: WebhookEndpointCreate, + db: Session = Depends(get_db), + _auth: dict = Depends(require_admin_auth), +) -> Dict[str, Any]: + """Create an outbound webhook endpoint.""" + try: + endpoint, raw_secret = create_webhook_endpoint(db, **payload.model_dump()) + except ValueError as exc: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc + body = endpoint_to_dict(endpoint) + body["secret"] = raw_secret + return body + + +@router.put("/{endpoint_id}", response_model=WebhookEndpointResponse) +async def update_webhook( + endpoint_id: int, + payload: WebhookEndpointUpdate, + db: Session = Depends(get_db), + _auth: dict = Depends(require_admin_auth), +) -> Dict[str, Any]: + """Update an outbound webhook endpoint.""" + endpoint = db.query(WebhookEndpoint).filter(WebhookEndpoint.id == endpoint_id).first() + if endpoint is None: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, detail="Webhook endpoint not found" + ) + try: + endpoint, raw_secret = update_webhook_endpoint( + db, + endpoint, + **payload.model_dump(exclude_unset=True), + ) + except ValueError as exc: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc + body = endpoint_to_dict(endpoint) + body["secret"] = raw_secret + return body + + +@router.delete("/{endpoint_id}", response_model=WebhookEndpointResponse) +async def disable_webhook( + endpoint_id: int, + db: Session = Depends(get_db), + _auth: dict = Depends(require_admin_auth), +) -> Dict[str, Any]: + """Disable a webhook endpoint without deleting delivery history.""" + endpoint = db.query(WebhookEndpoint).filter(WebhookEndpoint.id == endpoint_id).first() + if endpoint is None: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, detail="Webhook endpoint not found" + ) + endpoint.enabled = False + db.commit() + db.refresh(endpoint) + body = endpoint_to_dict(endpoint) + body["secret"] = None + return body + + +@router.get("/deliveries", response_model=WebhookDeliveryListResponse) +async def list_webhook_deliveries( + endpoint_id: Optional[int] = None, + delivery_status: Optional[str] = Query(None, alias="status"), + limit: int = 50, + db: Session = Depends(get_db), + _auth: dict = Depends(require_admin_auth), +) -> Dict[str, Any]: + """Return recent outbound webhook deliveries.""" + query = db.query(WebhookDelivery) + if endpoint_id is not None: + query = query.filter(WebhookDelivery.endpoint_id == endpoint_id) + if delivery_status: + query = query.filter(WebhookDelivery.status == delivery_status) + deliveries = ( + query.order_by(WebhookDelivery.created_at.desc(), WebhookDelivery.id.desc()) + .limit(max(1, min(limit, 200))) + .all() + ) + return {"deliveries": [delivery_to_dict(delivery) for delivery in deliveries]} + + +@router.post("/{endpoint_id}/test", response_model=WebhookTestResponse) +async def test_webhook( + endpoint_id: int, + db: Session = Depends(get_db), + _auth: dict = Depends(require_admin_auth), +) -> Dict[str, Any]: + """Queue and immediately attempt a test delivery for an endpoint.""" + try: + delivery = queue_test_webhook(db, endpoint_id) + except ValueError as exc: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc + delivered = deliver_due_webhooks(db, endpoint_id=endpoint_id, limit=1) + return {"delivery": delivery_to_dict(delivered[0] if delivered else delivery)} + + +@router.post("/deliveries/process", response_model=WebhookDeliveryListResponse) +async def process_due_webhooks( + limit: int = 25, + db: Session = Depends(get_db), + _auth: dict = Depends(require_admin_auth), +) -> Dict[str, Any]: + """Attempt due pending webhook deliveries.""" + deliveries = deliver_due_webhooks(db, limit=max(1, min(limit, 100))) + return {"deliveries": [delivery_to_dict(delivery) for delivery in deliveries]} diff --git a/backend/app/main.py b/backend/app/main.py index 0e581cc..f92f727 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -19,6 +19,7 @@ import app.models.mail_source_import # noqa: F401 – ensure import history tab import app.models.report # noqa: F401 – ensure DMARCReport/ReportRecord tables are registered import app.models.setting # noqa: F401 – ensure Setting table is registered import app.models.user # noqa: F401 – ensure User table is registered +import app.models.webhook # noqa: F401 – ensure webhook tables are registered from app.api.api_v1.api import api_router from app.core.config import get_settings from app.core.database import Base, SessionLocal, engine @@ -42,6 +43,7 @@ from app.services.runtime_status import ( mark_scheduler_success, ) from app.services.summary_notifications import send_due_scheduled_summaries +from app.services.webhook_events import deliver_due_webhooks # Set up logging logger = logging.getLogger(__name__) @@ -290,6 +292,20 @@ def _send_due_summary_notifications() -> None: db.close() +def _deliver_due_webhook_events() -> None: + """Attempt due outbound webhook deliveries.""" + db = SessionLocal() + try: + deliveries = deliver_due_webhooks(db) + if deliveries: + delivered = sum(1 for item in deliveries if item.status == "delivered") + logger.info( + "Processed %d webhook deliveries (%d delivered)", len(deliveries), delivered + ) + finally: + db.close() + + def _next_sleep_seconds( min_sleep: int = 60, enabled_sources: Optional[List[MailSource]] = None ) -> int: @@ -318,6 +334,7 @@ async def scheduled_imap_polling(): try: enabled_sources = _poll_all_enabled_sources() _send_due_summary_notifications() + _deliver_due_webhook_events() mark_scheduler_success() except Exception as e: # pylint: disable=broad-exception-caught logger.error("Error in IMAP polling task: %s", str(e)) diff --git a/backend/app/models/webhook.py b/backend/app/models/webhook.py new file mode 100644 index 0000000..71a852b --- /dev/null +++ b/backend/app/models/webhook.py @@ -0,0 +1,66 @@ +from datetime import datetime + +from sqlalchemy import Boolean, Column, DateTime, ForeignKey, Index, Integer, String, Text + +from app.core.database import Base + + +class WebhookEndpoint(Base): + """Outbound webhook endpoint configured by an operator.""" + + __tablename__ = "webhook_endpoints" + + id = Column(Integer, primary_key=True, index=True) + name = Column(String(120), nullable=False) + url = Column(Text, nullable=False) + secret = Column(Text, nullable=False) + event_types = Column(Text, nullable=False, default="*") + enabled = Column(Boolean, default=True, nullable=False, index=True) + max_attempts = Column(Integer, default=5, nullable=False) + timeout_seconds = Column(Integer, default=10, nullable=False) + created_at = Column(DateTime, default=datetime.utcnow, nullable=False, index=True) + updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) + last_success_at = Column(DateTime, nullable=True, index=True) + last_failure_at = Column(DateTime, nullable=True, index=True) + failure_count = Column(Integer, default=0, nullable=False) + + __table_args__ = (Index("ix_webhook_endpoints_enabled_events", "enabled", "event_types"),) + + def __repr__(self): + return f"" + + +class WebhookDelivery(Base): + """Single outbound webhook delivery attempt state.""" + + __tablename__ = "webhook_deliveries" + + id = Column(Integer, primary_key=True, index=True) + endpoint_id = Column(Integer, ForeignKey("webhook_endpoints.id"), nullable=False, index=True) + event_type = Column(String(80), nullable=False, index=True) + payload = Column(Text, nullable=False) + idempotency_key = Column(String(160), nullable=False, index=True) + status = Column(String(24), nullable=False, default="pending", index=True) + attempt_count = Column(Integer, default=0, nullable=False) + max_attempts = Column(Integer, default=5, nullable=False) + next_attempt_at = Column(DateTime, default=datetime.utcnow, nullable=False, index=True) + last_attempt_at = Column(DateTime, nullable=True, index=True) + delivered_at = Column(DateTime, nullable=True, index=True) + last_status_code = Column(Integer, nullable=True) + last_error = Column(Text, nullable=True) + response_excerpt = Column(Text, nullable=True) + created_at = Column(DateTime, default=datetime.utcnow, nullable=False, index=True) + updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) + + __table_args__ = ( + Index( + "ix_webhook_delivery_endpoint_idempotency", + "endpoint_id", + "idempotency_key", + unique=True, + ), + Index("ix_webhook_delivery_due", "status", "next_attempt_at"), + ) + + def __repr__(self): + return f"" diff --git a/backend/app/services/alert_rules.py b/backend/app/services/alert_rules.py index ba838ee..29bf70b 100644 --- a/backend/app/services/alert_rules.py +++ b/backend/app/services/alert_rules.py @@ -2,6 +2,7 @@ from __future__ import annotations +import logging from datetime import datetime, timedelta, timezone from typing import Any, Dict, List, Optional @@ -13,6 +14,15 @@ from app.models.report import DMARCReport, ReportRecord from app.models.setting import Setting from app.services.alert_history import record_alert_evaluation from app.services.notifications import NotificationResult, send_notification +from app.services.webhook_events import ( + EVENT_ALERT_CREATED, + EVENT_COMPLIANCE_DROP, + EVENT_REPORTS_MISSING, + EVENT_SENDER_NEW, + enqueue_webhook_event, +) + +logger = logging.getLogger(__name__) def _truthy(value: Optional[str], default: bool = True) -> bool: @@ -235,10 +245,34 @@ def evaluate_alert_rules(db: Session) -> List[Dict[str, Any]]: return alerts +def enqueue_alert_webhook_events(db: Session, alerts: List[Dict[str, Any]]) -> None: + """Queue webhook events for alert-rule results without failing alert evaluation.""" + event_by_rule = { + "new_sender_source": EVENT_SENDER_NEW, + "missing_reports": EVENT_REPORTS_MISSING, + "compliance_drop": EVENT_COMPLIANCE_DROP, + } + for alert in alerts: + rule = alert.get("rule", "alert") + event_type = event_by_rule.get(rule, EVENT_ALERT_CREATED) + domain = alert.get("domain", "global") + idempotency_key = f"{event_type}:{domain}:{rule}:{alert.get('detail', '')}" + try: + enqueue_webhook_event( + db, + event_type=event_type, + payload=alert, + idempotency_key=idempotency_key, + ) + except Exception as exc: # pylint: disable=broad-exception-caught + logger.warning("Failed to queue alert webhook event: %s", exc) + + def send_current_alerts(db: Session) -> Dict[str, Any]: """Evaluate current alert rules and send one summary notification when needed.""" alerts = evaluate_alert_rules(db) record_alert_evaluation(db, alerts) + enqueue_alert_webhook_events(db, alerts) if not alerts: return { "alerts": [], diff --git a/backend/app/services/webhook_events.py b/backend/app/services/webhook_events.py new file mode 100644 index 0000000..97316b8 --- /dev/null +++ b/backend/app/services/webhook_events.py @@ -0,0 +1,465 @@ +"""Outbound webhook event creation, signing, and delivery.""" + +from __future__ import annotations + +import hashlib +import hmac +import json +import secrets +import urllib.error +import urllib.request +from dataclasses import dataclass +from datetime import datetime, timedelta +from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple +from urllib.parse import urlparse + +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session + +from app.core.credential_encryption import decrypt_secret, encrypt_secret, is_encrypted_secret +from app.models.webhook import WebhookDelivery, WebhookEndpoint + +EVENT_REPORT_IMPORTED = "dmarq.report.imported" +EVENT_SENDER_NEW = "dmarq.sender.new" +EVENT_COMPLIANCE_DROP = "dmarq.compliance.drop" +EVENT_REPORTS_MISSING = "dmarq.reports.missing" +EVENT_ALERT_CREATED = "dmarq.alert.created" +EVENT_ALERT_RESOLVED = "dmarq.alert.resolved" +EVENT_WEBHOOK_TEST = "dmarq.webhook.test" + +SUPPORTED_EVENT_TYPES = [ + EVENT_REPORT_IMPORTED, + EVENT_SENDER_NEW, + EVENT_COMPLIANCE_DROP, + EVENT_REPORTS_MISSING, + EVENT_ALERT_CREATED, + EVENT_ALERT_RESOLVED, + EVENT_WEBHOOK_TEST, +] + +DELIVERY_PENDING = "pending" +DELIVERY_DELIVERED = "delivered" +DELIVERY_FAILED = "failed" +DELIVERY_ABANDONED = "abandoned" + + +@dataclass +class DeliveryAttemptResult: + """Result returned by a webhook HTTP sender.""" + + status_code: int + body: str = "" + + +WebhookSender = Callable[[str, bytes, Dict[str, str], int], DeliveryAttemptResult] + + +def normalize_event_types(event_types: Iterable[str]) -> List[str]: + """Return validated event types for endpoint storage.""" + cleaned = sorted({item.strip() for item in event_types if item and item.strip()}) + if not cleaned: + return ["*"] + if "*" in cleaned: + return ["*"] + invalid = [item for item in cleaned if item not in SUPPORTED_EVENT_TYPES] + if invalid: + raise ValueError(f"Unsupported webhook event type: {', '.join(invalid)}") + return cleaned + + +def event_types_to_string(event_types: Iterable[str]) -> str: + """Serialize event types for storage.""" + return ",".join(normalize_event_types(event_types)) + + +def parse_event_types(value: str) -> List[str]: + """Parse stored event types.""" + return normalize_event_types((value or "*").split(",")) + + +def endpoint_matches_event(endpoint: WebhookEndpoint, event_type: str) -> bool: + """Return True when an endpoint should receive an event.""" + event_types = parse_event_types(endpoint.event_types) + return "*" in event_types or event_type in event_types + + +def generate_webhook_secret() -> str: + """Generate a signing secret for outbound webhook deliveries.""" + return secrets.token_urlsafe(32) + + +def _encrypt(value: str) -> str: + return encrypt_secret(value) or "" + + +def _decrypt(value: str) -> str: + return decrypt_secret(value) or "" + + +def _redact_url(url: str) -> str: + parsed = urlparse(url) + if not parsed.scheme or not parsed.netloc: + return "[invalid url]" + host = parsed.hostname or parsed.netloc + port = f":{parsed.port}" if parsed.port else "" + path = parsed.path or "/" + return f"{parsed.scheme}://{host}{port}{path}" + + +def validate_webhook_url(url: str) -> str: + """Validate and normalize an outbound webhook URL.""" + clean_url = url.strip() + parsed = urlparse(clean_url) + if parsed.scheme not in {"http", "https"} or not parsed.netloc: + raise ValueError("Webhook URL must be an absolute http or https URL") + return clean_url + + +def create_webhook_endpoint( + db: Session, + *, + name: str, + url: str, + secret: Optional[str] = None, + event_types: Iterable[str] = ("*",), + enabled: bool = True, + max_attempts: int = 5, + timeout_seconds: int = 10, +) -> Tuple[WebhookEndpoint, str]: + """Create a webhook endpoint and return the endpoint plus raw signing secret.""" + clean_name = name.strip() + if not clean_name: + raise ValueError("Webhook name is required") + clean_url = validate_webhook_url(url) + raw_secret = secret.strip() if secret else generate_webhook_secret() + if len(raw_secret) < 16: + raise ValueError("Webhook signing secret must be at least 16 characters") + + endpoint = WebhookEndpoint( + name=clean_name, + url=_encrypt(clean_url), + secret=_encrypt(raw_secret), + event_types=event_types_to_string(event_types), + enabled=enabled, + max_attempts=max(1, min(int(max_attempts or 5), 10)), + timeout_seconds=max(1, min(int(timeout_seconds or 10), 30)), + ) + db.add(endpoint) + db.commit() + db.refresh(endpoint) + return endpoint, raw_secret + + +def update_webhook_endpoint( + db: Session, + endpoint: WebhookEndpoint, + *, + name: Optional[str] = None, + url: Optional[str] = None, + secret: Optional[str] = None, + event_types: Optional[Iterable[str]] = None, + enabled: Optional[bool] = None, + max_attempts: Optional[int] = None, + timeout_seconds: Optional[int] = None, +) -> Tuple[WebhookEndpoint, Optional[str]]: + """Update a webhook endpoint. Return the endpoint and newly supplied/generated secret.""" + returned_secret = None + if name is not None: + clean_name = name.strip() + if not clean_name: + raise ValueError("Webhook name is required") + endpoint.name = clean_name + if url is not None and url.strip() and url != "**redacted**": + endpoint.url = _encrypt(validate_webhook_url(url)) + if secret is not None and secret.strip() and secret != "**redacted**": + returned_secret = secret.strip() + if len(returned_secret) < 16: + raise ValueError("Webhook signing secret must be at least 16 characters") + endpoint.secret = _encrypt(returned_secret) + if event_types is not None: + endpoint.event_types = event_types_to_string(event_types) + if enabled is not None: + endpoint.enabled = enabled + if max_attempts is not None: + endpoint.max_attempts = max(1, min(int(max_attempts), 10)) + if timeout_seconds is not None: + endpoint.timeout_seconds = max(1, min(int(timeout_seconds), 30)) + db.commit() + db.refresh(endpoint) + return endpoint, returned_secret + + +def _stable_json(value: Dict[str, Any]) -> str: + return json.dumps(value, sort_keys=True, separators=(",", ":"), default=str) + + +def build_event_payload(event_type: str, payload: Dict[str, Any]) -> Dict[str, Any]: + """Wrap event data in a stable, documented envelope.""" + now = datetime.utcnow().replace(microsecond=0).isoformat() + "Z" + return { + "event_type": event_type, + "created_at": now, + "data": payload, + } + + +def default_idempotency_key(event_type: str, payload: Dict[str, Any]) -> str: + """Build a deterministic idempotency key for an event payload.""" + digest = hashlib.sha256(_stable_json(payload).encode("utf-8")).hexdigest()[:32] + return f"{event_type}:{digest}" + + +def enqueue_webhook_event( + db: Session, + *, + event_type: str, + payload: Dict[str, Any], + idempotency_key: Optional[str] = None, +) -> List[WebhookDelivery]: + """Create pending deliveries for all enabled endpoints matching an event.""" + if event_type not in SUPPORTED_EVENT_TYPES: + raise ValueError(f"Unsupported webhook event type: {event_type}") + endpoints = db.query(WebhookEndpoint).filter(WebhookEndpoint.enabled.is_(True)).all() + event_payload = build_event_payload(event_type, payload) + key = idempotency_key or default_idempotency_key(event_type, event_payload) + deliveries: List[WebhookDelivery] = [] + + for endpoint in endpoints: + if not endpoint_matches_event(endpoint, event_type): + continue + delivery = WebhookDelivery( + endpoint_id=endpoint.id, + event_type=event_type, + payload=_stable_json(event_payload), + idempotency_key=key, + status=DELIVERY_PENDING, + max_attempts=endpoint.max_attempts, + ) + db.add(delivery) + try: + db.commit() + except IntegrityError: + db.rollback() + existing = ( + db.query(WebhookDelivery) + .filter( + WebhookDelivery.endpoint_id == endpoint.id, + WebhookDelivery.idempotency_key == key, + ) + .first() + ) + if existing: + deliveries.append(existing) + continue + db.refresh(delivery) + deliveries.append(delivery) + return deliveries + + +def _delivery_body(delivery: WebhookDelivery) -> bytes: + return delivery.payload.encode("utf-8") + + +def sign_delivery(secret: str, delivery: WebhookDelivery, timestamp: int, body: bytes) -> str: + """Return the v1 HMAC signature for a delivery.""" + signed = f"{timestamp}.{delivery.id}.".encode("utf-8") + body + digest = hmac.new(secret.encode("utf-8"), signed, hashlib.sha256).hexdigest() + return f"v1={digest}" + + +def build_delivery_headers(endpoint: WebhookEndpoint, delivery: WebhookDelivery) -> Dict[str, str]: + """Build outbound webhook headers with event metadata and HMAC signature.""" + body = _delivery_body(delivery) + timestamp = int(datetime.utcnow().timestamp()) + secret = _decrypt(endpoint.secret) + return { + "Content-Type": "application/json", + "User-Agent": "DMARQ-Webhooks/1.0", + "X-DMARQ-Event": delivery.event_type, + "X-DMARQ-Delivery": str(delivery.id), + "X-DMARQ-Idempotency-Key": delivery.idempotency_key, + "X-DMARQ-Timestamp": str(timestamp), + "X-DMARQ-Signature": sign_delivery(secret, delivery, timestamp, body), + } + + +def default_webhook_sender( + url: str, body: bytes, headers: Dict[str, str], timeout_seconds: int +) -> DeliveryAttemptResult: + """Send a webhook delivery using the Python standard library.""" + request = urllib.request.Request(url=url, data=body, headers=headers, method="POST") + try: + with urllib.request.urlopen(request, timeout=timeout_seconds) as response: # nosec B310 + response_body = response.read(4096).decode("utf-8", errors="replace") + return DeliveryAttemptResult(status_code=response.status, body=response_body) + except urllib.error.HTTPError as exc: + response_body = exc.read(4096).decode("utf-8", errors="replace") + return DeliveryAttemptResult(status_code=exc.code, body=response_body) + except urllib.error.URLError as exc: + raise ConnectionError(str(exc.reason)) from exc + + +def _backoff_for_attempt(attempt_count: int) -> timedelta: + seconds = min(3600, 60 * (2 ** max(0, attempt_count - 1))) + return timedelta(seconds=seconds) + + +def _response_excerpt(value: str) -> str: + return (value or "")[:500] + + +def deliver_webhook_delivery( + db: Session, + delivery: WebhookDelivery, + *, + sender: WebhookSender = default_webhook_sender, +) -> WebhookDelivery: + """Attempt one webhook delivery and persist retry state.""" + endpoint = db.query(WebhookEndpoint).filter(WebhookEndpoint.id == delivery.endpoint_id).first() + now = datetime.utcnow() + delivery.attempt_count = int(delivery.attempt_count or 0) + 1 + delivery.last_attempt_at = now + + if endpoint is None or not endpoint.enabled: + delivery.status = DELIVERY_ABANDONED + delivery.last_error = "Webhook endpoint is disabled or missing." + delivery.next_attempt_at = now + db.commit() + db.refresh(delivery) + return delivery + + try: + body = _delivery_body(delivery) + result = sender( + _decrypt(endpoint.url), + body, + build_delivery_headers(endpoint, delivery), + endpoint.timeout_seconds, + ) + delivery.last_status_code = result.status_code + delivery.response_excerpt = _response_excerpt(result.body) + if 200 <= result.status_code < 300: + delivery.status = DELIVERY_DELIVERED + delivery.delivered_at = now + delivery.last_error = None + delivery.next_attempt_at = now + endpoint.last_success_at = now + endpoint.failure_count = 0 + else: + delivery.last_error = f"HTTP {result.status_code}" + _mark_delivery_failure(delivery, endpoint, now) + except Exception as exc: # pylint: disable=broad-exception-caught + delivery.last_error = str(exc)[:500] + _mark_delivery_failure(delivery, endpoint, now) + + db.commit() + db.refresh(delivery) + return delivery + + +def _mark_delivery_failure( + delivery: WebhookDelivery, endpoint: WebhookEndpoint, now: datetime +) -> None: + endpoint.last_failure_at = now + endpoint.failure_count = int(endpoint.failure_count or 0) + 1 + if delivery.attempt_count >= delivery.max_attempts: + delivery.status = DELIVERY_FAILED + delivery.next_attempt_at = now + return + delivery.status = DELIVERY_PENDING + delivery.next_attempt_at = now + _backoff_for_attempt(delivery.attempt_count) + + +def deliver_due_webhooks( + db: Session, + *, + limit: int = 25, + endpoint_id: Optional[int] = None, + sender: WebhookSender = default_webhook_sender, +) -> List[WebhookDelivery]: + """Deliver pending webhook deliveries whose retry time has arrived.""" + now = datetime.utcnow() + query = db.query(WebhookDelivery).filter( + WebhookDelivery.status == DELIVERY_PENDING, + WebhookDelivery.next_attempt_at <= now, + ) + if endpoint_id is not None: + query = query.filter(WebhookDelivery.endpoint_id == endpoint_id) + deliveries = ( + query.order_by(WebhookDelivery.next_attempt_at, WebhookDelivery.id).limit(limit).all() + ) + return [deliver_webhook_delivery(db, delivery, sender=sender) for delivery in deliveries] + + +def queue_test_webhook(db: Session, endpoint_id: int) -> WebhookDelivery: + """Queue a one-off test delivery for a specific webhook endpoint.""" + endpoint = db.query(WebhookEndpoint).filter(WebhookEndpoint.id == endpoint_id).first() + if endpoint is None: + raise ValueError("Webhook endpoint not found") + payload = { + "endpoint_id": endpoint.id, + "endpoint_name": endpoint.name, + "message": "DMARQ webhook test delivery", + } + delivery = WebhookDelivery( + endpoint_id=endpoint.id, + event_type=EVENT_WEBHOOK_TEST, + payload=_stable_json(build_event_payload(EVENT_WEBHOOK_TEST, payload)), + idempotency_key=f"{EVENT_WEBHOOK_TEST}:{secrets.token_hex(16)}", + status=DELIVERY_PENDING, + max_attempts=endpoint.max_attempts, + ) + db.add(delivery) + db.commit() + db.refresh(delivery) + return delivery + + +def endpoint_to_dict(endpoint: WebhookEndpoint) -> Dict[str, Any]: + """Return an API-safe webhook endpoint representation.""" + raw_url = _decrypt(endpoint.url) + return { + "id": endpoint.id, + "name": endpoint.name, + "url": _redact_url(raw_url), + "event_types": parse_event_types(endpoint.event_types), + "enabled": endpoint.enabled, + "max_attempts": endpoint.max_attempts, + "timeout_seconds": endpoint.timeout_seconds, + "created_at": endpoint.created_at.isoformat() if endpoint.created_at else None, + "updated_at": endpoint.updated_at.isoformat() if endpoint.updated_at else None, + "last_success_at": ( + endpoint.last_success_at.isoformat() if endpoint.last_success_at else None + ), + "last_failure_at": ( + endpoint.last_failure_at.isoformat() if endpoint.last_failure_at else None + ), + "failure_count": endpoint.failure_count, + "secret_configured": bool(endpoint.secret), + "url_encrypted": is_encrypted_secret(endpoint.url), + } + + +def delivery_to_dict(delivery: WebhookDelivery) -> Dict[str, Any]: + """Return an API-safe delivery representation.""" + return { + "id": delivery.id, + "endpoint_id": delivery.endpoint_id, + "event_type": delivery.event_type, + "idempotency_key": delivery.idempotency_key, + "status": delivery.status, + "attempt_count": delivery.attempt_count, + "max_attempts": delivery.max_attempts, + "next_attempt_at": ( + delivery.next_attempt_at.isoformat() if delivery.next_attempt_at else None + ), + "last_attempt_at": ( + delivery.last_attempt_at.isoformat() if delivery.last_attempt_at else None + ), + "delivered_at": delivery.delivered_at.isoformat() if delivery.delivered_at else None, + "last_status_code": delivery.last_status_code, + "last_error": delivery.last_error, + "response_excerpt": delivery.response_excerpt, + "created_at": delivery.created_at.isoformat() if delivery.created_at else None, + "updated_at": delivery.updated_at.isoformat() if delivery.updated_at else None, + } diff --git a/backend/app/templates/settings.html b/backend/app/templates/settings.html index d0ffb9e..908b7a8 100644 --- a/backend/app/templates/settings.html +++ b/backend/app/templates/settings.html @@ -633,6 +633,151 @@ {% endcall %} {% endcall %} + + {% call card() %} + {% call card_header() %} + {% call card_title() %}Webhooks{% endcall %} + {% call card_description() %}Send signed operational events to downstream systems{% endcall %} + {% endcall %} + {% call card_content() %} +
+
+
+
+ + +
+
+ + +
+
+ +
+
+ + +
+
+ + +
+
+ + +
+
+ +
+ + +
+
+ + + +
+ + + + + + + + + + + + + + +
NameURLEventsStatusLast Result
+
+ +
+
+

Recent Deliveries

+
+ + +
+
+ +
+ +
+
+
+ {% endcall %} + {% endcall %} + {% call card() %} {% call card_header() %} @@ -681,6 +826,22 @@ function settingsApp() { importingCfZones: false, cfZones: [], showCfToken: false, + loadingWebhooks: false, + savingWebhook: false, + testingWebhookId: null, + disablingWebhookId: null, + processingWebhooks: false, + loadingWebhookDeliveries: false, + webhooks: [], + webhookDeliveries: [], + webhookEventTypes: [], + newWebhook: { + name: '', + url: '', + eventType: '*', + max_attempts: 5, + timeout_seconds: 10, + }, // Session cookie is sent automatically by the browser (httpOnly, same-origin). // No manual auth header needed for API calls from the UI. @@ -705,6 +866,8 @@ function settingsApp() { this.s = map; await this.loadAlertHistory(false); await this.loadConfigAudit(false); + await this.loadWebhooks(false); + await this.loadWebhookDeliveries(false); } catch (err) { this.showFlash('Error loading settings: ' + err.message, false); } @@ -951,6 +1114,142 @@ function settingsApp() { } }, + async loadWebhooks(showMessage = true) { + this.loadingWebhooks = true; + try { + const res = await fetch('/api/v1/webhooks', { headers: this.apiHeaders() }); + const data = await res.json().catch(() => ({})); + if (!res.ok) { + if (showMessage) this.showFlash('Webhook load failed: ' + (data.detail || res.statusText), false); + } else { + this.webhooks = data.endpoints || []; + this.webhookEventTypes = data.supported_event_types || []; + if (showMessage) this.showFlash('Webhooks refreshed.', true); + } + } catch (err) { + if (showMessage) this.showFlash('Error loading webhooks: ' + err.message, false); + } finally { + this.loadingWebhooks = false; + } + }, + + async createWebhook() { + this.savingWebhook = true; + try { + const payload = { + name: this.newWebhook.name, + url: this.newWebhook.url, + event_types: [this.newWebhook.eventType || '*'], + max_attempts: Number(this.newWebhook.max_attempts || 5), + timeout_seconds: Number(this.newWebhook.timeout_seconds || 10), + enabled: true, + }; + const res = await fetch('/api/v1/webhooks', { + method: 'POST', + headers: this.apiHeaders(), + body: JSON.stringify(payload), + }); + const data = await res.json().catch(() => ({})); + if (!res.ok) { + this.showFlash('Webhook create failed: ' + (data.detail || res.statusText), false); + } else { + this.newWebhook = { name: '', url: '', eventType: '*', max_attempts: 5, timeout_seconds: 10 }; + await this.loadWebhooks(false); + this.showFlash('Webhook created. Signing secret was generated and stored securely.', true); + } + } catch (err) { + this.showFlash('Error creating webhook: ' + err.message, false); + } finally { + this.savingWebhook = false; + } + }, + + async testWebhook(endpointId) { + this.testingWebhookId = endpointId; + try { + const res = await fetch(`/api/v1/webhooks/${endpointId}/test`, { + method: 'POST', + headers: this.apiHeaders(), + }); + const data = await res.json().catch(() => ({})); + if (!res.ok) { + this.showFlash('Webhook test failed: ' + (data.detail || res.statusText), false); + } else { + await this.loadWebhooks(false); + await this.loadWebhookDeliveries(false); + const status = data.delivery ? data.delivery.status : 'queued'; + this.showFlash(`Webhook test ${status}.`, status === 'delivered'); + } + } catch (err) { + this.showFlash('Error testing webhook: ' + err.message, false); + } finally { + this.testingWebhookId = null; + } + }, + + async disableWebhook(endpointId) { + this.disablingWebhookId = endpointId; + try { + const res = await fetch(`/api/v1/webhooks/${endpointId}`, { + method: 'DELETE', + headers: this.apiHeaders(), + }); + const data = await res.json().catch(() => ({})); + if (!res.ok) { + this.showFlash('Webhook disable failed: ' + (data.detail || res.statusText), false); + } else { + await this.loadWebhooks(false); + this.showFlash('Webhook disabled.', true); + } + } catch (err) { + this.showFlash('Error disabling webhook: ' + err.message, false); + } finally { + this.disablingWebhookId = null; + } + }, + + async loadWebhookDeliveries(showMessage = true) { + this.loadingWebhookDeliveries = true; + try { + const res = await fetch('/api/v1/webhooks/deliveries?limit=10', { + headers: this.apiHeaders(), + }); + const data = await res.json().catch(() => ({})); + if (!res.ok) { + if (showMessage) this.showFlash('Delivery history failed: ' + (data.detail || res.statusText), false); + } else { + this.webhookDeliveries = data.deliveries || []; + if (showMessage) this.showFlash('Webhook deliveries refreshed.', true); + } + } catch (err) { + if (showMessage) this.showFlash('Error loading webhook deliveries: ' + err.message, false); + } finally { + this.loadingWebhookDeliveries = false; + } + }, + + async processWebhooks() { + this.processingWebhooks = true; + try { + const res = await fetch('/api/v1/webhooks/deliveries/process', { + method: 'POST', + headers: this.apiHeaders(), + }); + const data = await res.json().catch(() => ({})); + if (!res.ok) { + this.showFlash('Webhook processing failed: ' + (data.detail || res.statusText), false); + } else { + await this.loadWebhooks(false); + await this.loadWebhookDeliveries(false); + this.showFlash(`${(data.deliveries || []).length} due webhook deliver${(data.deliveries || []).length === 1 ? 'y' : 'ies'} processed.`, true); + } + } catch (err) { + this.showFlash('Error processing webhooks: ' + err.message, false); + } finally { + this.processingWebhooks = false; + } + }, + showFlash(msg, ok) { this.flashMsg = msg; this.flashOk = ok; diff --git a/backend/app/tests/conftest.py b/backend/app/tests/conftest.py index 1976b63..ed0c15d 100644 --- a/backend/app/tests/conftest.py +++ b/backend/app/tests/conftest.py @@ -15,6 +15,7 @@ import app.models.mail_source_import # noqa: F401 # pylint: disable=unused-imp import app.models.report # noqa: F401 # pylint: disable=unused-import import app.models.setting # noqa: F401 # pylint: disable=unused-import import app.models.user # noqa: F401 # pylint: disable=unused-import +import app.models.webhook # noqa: F401 # pylint: disable=unused-import from app.core.database import Base, get_db from app.core.security import require_admin_auth from app.main import create_app diff --git a/backend/app/tests/test_webhooks.py b/backend/app/tests/test_webhooks.py new file mode 100644 index 0000000..a0cbe8e --- /dev/null +++ b/backend/app/tests/test_webhooks.py @@ -0,0 +1,308 @@ +import hmac +import json + +from fastapi.testclient import TestClient + +from app.models.webhook import WebhookDelivery +from app.services.webhook_events import ( + DELIVERY_ABANDONED, + DELIVERY_DELIVERED, + DELIVERY_FAILED, + DELIVERY_PENDING, + EVENT_ALERT_CREATED, + EVENT_REPORTS_MISSING, + EVENT_WEBHOOK_TEST, + DeliveryAttemptResult, + create_webhook_endpoint, + deliver_due_webhooks, + deliver_webhook_delivery, + endpoint_to_dict, + enqueue_webhook_event, + normalize_event_types, + queue_test_webhook, + sign_delivery, + update_webhook_endpoint, + validate_webhook_url, +) + + +def test_webhook_event_delivery_signs_and_records_success(db_session): + """Webhook deliveries include replay-resistant metadata and mark success.""" + endpoint, secret = create_webhook_endpoint( + db_session, + name="receiver", + url="https://receiver.example/webhook?token=hidden", + event_types=[EVENT_ALERT_CREATED], + ) + deliveries = enqueue_webhook_event( + db_session, + event_type=EVENT_ALERT_CREATED, + payload={"domain": "example.com", "detail": "alert"}, + idempotency_key="alert-example", + ) + assert len(deliveries) == 1 + + seen = {} + + def sender(url, body, headers, timeout_seconds): + seen["url"] = url + seen["body"] = body + seen["headers"] = headers + seen["timeout_seconds"] = timeout_seconds + expected = sign_delivery( + secret, + deliveries[0], + int(headers["X-DMARQ-Timestamp"]), + body, + ) + assert hmac.compare_digest(headers["X-DMARQ-Signature"], expected) + assert headers["X-DMARQ-Event"] == EVENT_ALERT_CREATED + assert headers["X-DMARQ-Idempotency-Key"] == "alert-example" + return DeliveryAttemptResult(status_code=204, body="") + + delivered = deliver_due_webhooks(db_session, sender=sender) + + assert delivered[0].status == DELIVERY_DELIVERED + assert json.loads(seen["body"])["data"]["domain"] == "example.com" + assert seen["url"] == "https://receiver.example/webhook?token=hidden" + db_session.refresh(endpoint) + assert endpoint.last_success_at is not None + assert endpoint.failure_count == 0 + + +def test_webhook_delivery_retries_then_fails(db_session): + """Transient failures remain pending with backoff, then fail at max attempts.""" + create_webhook_endpoint( + db_session, + name="receiver", + url="https://receiver.example/webhook", + event_types=[EVENT_ALERT_CREATED], + max_attempts=2, + ) + enqueue_webhook_event( + db_session, + event_type=EVENT_ALERT_CREATED, + payload={"domain": "example.com"}, + idempotency_key="retry-example", + )[0] + + def failing_sender(url, body, headers, timeout_seconds): + return DeliveryAttemptResult(status_code=503, body="try later") + + first = deliver_due_webhooks(db_session, sender=failing_sender)[0] + assert first.status == DELIVERY_PENDING + assert first.attempt_count == 1 + assert first.next_attempt_at > first.last_attempt_at + + first.next_attempt_at = first.last_attempt_at + db_session.commit() + second = deliver_due_webhooks(db_session, sender=failing_sender)[0] + assert second.status == DELIVERY_FAILED + assert second.attempt_count == 2 + assert second.last_status_code == 503 + + +def test_webhook_idempotency_skips_duplicate_deliveries(db_session): + """Repeated events with the same idempotency key reuse the existing delivery.""" + create_webhook_endpoint( + db_session, + name="receiver", + url="https://receiver.example/webhook", + event_types=[EVENT_ALERT_CREATED], + ) + first = enqueue_webhook_event( + db_session, + event_type=EVENT_ALERT_CREATED, + payload={"domain": "example.com"}, + idempotency_key="same-key", + ) + second = enqueue_webhook_event( + db_session, + event_type=EVENT_ALERT_CREATED, + payload={"domain": "example.com"}, + idempotency_key="same-key", + ) + + assert first[0].id == second[0].id + assert db_session.query(WebhookDelivery).count() == 1 + + +def test_webhook_validation_update_and_abandoned_delivery(db_session): + """Endpoint helpers validate input, update secrets, and abandon disabled endpoints.""" + assert normalize_event_types([]) == ["*"] + assert normalize_event_types(["*", EVENT_ALERT_CREATED]) == ["*"] + assert ( + validate_webhook_url(" https://receiver.example/hook ") == "https://receiver.example/hook" + ) + + for bad_events in [["bad.event"]]: + try: + normalize_event_types(bad_events) + except ValueError as exc: + assert "Unsupported webhook event type" in str(exc) + else: # pragma: no cover - defensive assertion shape + raise AssertionError("invalid event type was accepted") + + for bad_url in ["ftp://receiver.example/hook", "not-a-url"]: + try: + validate_webhook_url(bad_url) + except ValueError as exc: + assert "absolute http or https URL" in str(exc) + else: # pragma: no cover - defensive assertion shape + raise AssertionError("invalid webhook URL was accepted") + + endpoint, _secret = create_webhook_endpoint( + db_session, + name="receiver", + url="https://receiver.example/webhook", + event_types=[EVENT_ALERT_CREATED], + max_attempts=50, + timeout_seconds=50, + ) + updated, returned_secret = update_webhook_endpoint( + db_session, + endpoint, + name="receiver two", + url="https://receiver.example/updated?secret=hidden", + secret="a-new-secret-value", + event_types=[EVENT_REPORTS_MISSING], + enabled=False, + max_attempts=20, + timeout_seconds=40, + ) + + assert updated.name == "receiver two" + assert returned_secret == "a-new-secret-value" + assert updated.event_types == EVENT_REPORTS_MISSING + assert updated.max_attempts == 10 + assert updated.timeout_seconds == 30 + assert endpoint_to_dict(updated)["url"] == "https://receiver.example/updated" + + delivery = queue_test_webhook(db_session, updated.id) + result = deliver_webhook_delivery(db_session, delivery) + assert result.status == DELIVERY_ABANDONED + assert "disabled or missing" in result.last_error + + +def test_admin_webhook_endpoints_hide_secrets_and_show_delivery_status( + authed_client: TestClient, + monkeypatch, +): + """Operators can create, inspect, and test webhooks without secret leakage.""" + created = authed_client.post( + "/api/v1/webhooks", + json={ + "name": "ops", + "url": "https://ops.example/hooks/dmarq?secret=hidden", + "event_types": [EVENT_WEBHOOK_TEST], + }, + ) + assert created.status_code == 200 + body = created.json() + assert body["secret"] + assert "hidden" not in body["url"] + + listed = authed_client.get("/api/v1/webhooks") + assert listed.status_code == 200 + assert listed.json()["endpoints"][0]["url"] == "https://ops.example/hooks/dmarq" + assert body["secret"] not in listed.text + + def fake_deliver_due_webhooks(db, endpoint_id=None, limit=25): + delivery = ( + db.query(WebhookDelivery) + .filter(WebhookDelivery.endpoint_id == endpoint_id) + .order_by(WebhookDelivery.id.desc()) + .first() + ) + delivery.status = DELIVERY_DELIVERED + delivery.attempt_count = 1 + db.commit() + db.refresh(delivery) + return [delivery] + + monkeypatch.setattr( + "app.api.api_v1.endpoints.webhooks.deliver_due_webhooks", + fake_deliver_due_webhooks, + ) + + tested = authed_client.post(f"/api/v1/webhooks/{body['id']}/test") + assert tested.status_code == 200 + delivery = tested.json()["delivery"] + assert delivery["event_type"] == EVENT_WEBHOOK_TEST + assert delivery["status"] == DELIVERY_DELIVERED + + history = authed_client.get("/api/v1/webhooks/deliveries") + assert history.status_code == 200 + assert history.json()["deliveries"][0]["event_type"] == EVENT_WEBHOOK_TEST + + +def test_admin_webhook_update_disable_filter_and_process( + authed_client: TestClient, + monkeypatch, +): + """Admin API covers update, filtering, due processing, and not-found paths.""" + created = authed_client.post( + "/api/v1/webhooks", + json={"name": "ops", "url": "https://ops.example/hooks/dmarq"}, + ).json() + + bad_create = authed_client.post( + "/api/v1/webhooks", + json={"name": "bad", "url": "ftp://ops.example/hooks/dmarq"}, + ) + assert bad_create.status_code == 400 + + missing_update = authed_client.put("/api/v1/webhooks/9999", json={"name": "missing"}) + assert missing_update.status_code == 404 + + bad_update = authed_client.put( + f"/api/v1/webhooks/{created['id']}", + json={"event_types": ["not.supported"]}, + ) + assert bad_update.status_code == 400 + + updated = authed_client.put( + f"/api/v1/webhooks/{created['id']}", + json={ + "name": "ops updated", + "event_types": [EVENT_REPORTS_MISSING], + "max_attempts": 2, + "timeout_seconds": 3, + }, + ) + assert updated.status_code == 200 + assert updated.json()["name"] == "ops updated" + assert updated.json()["event_types"] == [EVENT_REPORTS_MISSING] + + missing_test = authed_client.post("/api/v1/webhooks/9999/test") + assert missing_test.status_code == 404 + + def fake_process(db, endpoint_id=None, limit=25): + delivery = queue_test_webhook(db, created["id"]) + delivery.status = DELIVERY_DELIVERED + delivery.attempt_count = 1 + db.commit() + db.refresh(delivery) + return [delivery] + + monkeypatch.setattr( + "app.api.api_v1.endpoints.webhooks.deliver_due_webhooks", + fake_process, + ) + + processed = authed_client.post("/api/v1/webhooks/deliveries/process?limit=500") + assert processed.status_code == 200 + assert processed.json()["deliveries"][0]["status"] == DELIVERY_DELIVERED + + endpoint_filtered = authed_client.get( + f"/api/v1/webhooks/deliveries?endpoint_id={created['id']}&status={DELIVERY_DELIVERED}" + ) + assert endpoint_filtered.status_code == 200 + assert len(endpoint_filtered.json()["deliveries"]) == 1 + + disabled = authed_client.delete(f"/api/v1/webhooks/{created['id']}") + assert disabled.status_code == 200 + assert disabled.json()["enabled"] is False + + missing_delete = authed_client.delete("/api/v1/webhooks/9999") + assert missing_delete.status_code == 404 diff --git a/docs/milestones.md b/docs/milestones.md index cf1dc99..b56c5ee 100644 --- a/docs/milestones.md +++ b/docs/milestones.md @@ -236,7 +236,7 @@ Goal: let DMARQ integrate cleanly into existing security and operations workflow Planned: - A stable, documented read-only API surface for posture and reporting queries. Delivered with scoped `reports:read`, `posture:read`, and `tls-reports:read` API tokens, public read-only endpoints, and per-token usage audit fields. -- Webhook event delivery for key events (new sender source, compliance drop, missing reports, alert lifecycle). +- Webhook event delivery for key events (new sender source, compliance drop, missing reports, alert lifecycle). Delivered with encrypted webhook endpoints, signed delivery headers, idempotency keys, retry/backoff state, test sends, and delivery inspection. - Integration templates for SIEM and ticketing workflows (export formats, payload schemas, examples). - Token/scoping model for API access that matches governance needs (service accounts, least privilege). diff --git a/docs/reference/api.md b/docs/reference/api.md index f514e65..81aeccc 100644 --- a/docs/reference/api.md +++ b/docs/reference/api.md @@ -510,17 +510,39 @@ The API uses versioning in the URL path (/api/v1/) to ensure backward compatibil ## Webhooks -DMARQ can notify your systems about events via webhooks: +DMARQ can notify downstream systems about operational events from +**Settings > Webhooks** or the admin API. -1. Navigate to **Settings** > **API Access** > **Webhooks** -2. Click **Add Webhook** -3. Configure: - - Destination URL - - Secret token (for verification) - - Events to subscribe to +| Endpoint | Purpose | +| --- | --- | +| `GET /api/v1/webhooks` | List endpoints and supported event types | +| `POST /api/v1/webhooks` | Create an endpoint | +| `PUT /api/v1/webhooks/{id}` | Update an endpoint | +| `DELETE /api/v1/webhooks/{id}` | Disable an endpoint while keeping delivery history | +| `POST /api/v1/webhooks/{id}/test` | Queue and attempt a test delivery | +| `GET /api/v1/webhooks/deliveries` | Inspect recent delivery attempts | +| `POST /api/v1/webhooks/deliveries/process` | Attempt due retries | -Supported events: -- `report.processed` - When a new report is processed -- `compliance.threshold` - When compliance falls below threshold -- `domain.added` - When a domain is added -- `domain.removed` - When a domain is removed +Supported event types: +- `dmarq.report.imported` +- `dmarq.sender.new` +- `dmarq.compliance.drop` +- `dmarq.reports.missing` +- `dmarq.alert.created` +- `dmarq.alert.resolved` +- `dmarq.webhook.test` + +Deliveries are signed with HMAC-SHA256 using the endpoint signing secret. +Receivers should verify these headers: + +| Header | Description | +| --- | --- | +| `X-DMARQ-Event` | Event type | +| `X-DMARQ-Delivery` | Delivery id | +| `X-DMARQ-Idempotency-Key` | Stable deduplication key | +| `X-DMARQ-Timestamp` | Unix timestamp used in the signature | +| `X-DMARQ-Signature` | `v1=` over `timestamp.delivery_id.body` | + +Non-2xx responses are retried with exponential backoff until the endpoint's +maximum attempt count is reached. Operators can inspect the delivery status, +last response code, error text, and response excerpt without reading logs. diff --git a/docs/reference/database.md b/docs/reference/database.md index 5afe60c..c11fa0c 100644 --- a/docs/reference/database.md +++ b/docs/reference/database.md @@ -135,6 +135,46 @@ and are never stored. | last_used_ip | VARCHAR(64) | Source IP from the last successful API use | | usage_count | INTEGER | Successful API use count | +### Webhook_Endpoints + +The `webhook_endpoints` table stores outbound webhook destinations. Target +URLs and signing secrets are encrypted at rest. + +| Column | Type | Description | +|--------|------|-------------| +| id | INTEGER | Primary key | +| name | VARCHAR(120) | Operator-facing endpoint name | +| url | TEXT | Encrypted destination URL | +| secret | TEXT | Encrypted signing secret | +| event_types | TEXT | Comma-separated event subscriptions, or `*` | +| enabled | BOOLEAN | Whether deliveries can be sent | +| max_attempts | INTEGER | Maximum attempts before a delivery fails | +| timeout_seconds | INTEGER | Per-request timeout | +| last_success_at | TIMESTAMP | Last successful delivery | +| last_failure_at | TIMESTAMP | Last failed delivery attempt | +| failure_count | INTEGER | Consecutive endpoint-level failures | + +### Webhook_Deliveries + +The `webhook_deliveries` table records delivery attempts and retry state. + +| Column | Type | Description | +|--------|------|-------------| +| id | INTEGER | Primary key | +| endpoint_id | INTEGER | Foreign key to webhook_endpoints.id | +| event_type | VARCHAR(80) | Delivered event type | +| payload | TEXT | Event envelope JSON | +| idempotency_key | VARCHAR(160) | Stable deduplication key per endpoint | +| status | VARCHAR(24) | pending, delivered, failed, or abandoned | +| attempt_count | INTEGER | Attempts already made | +| max_attempts | INTEGER | Maximum attempts for this delivery | +| next_attempt_at | TIMESTAMP | Next retry time | +| last_attempt_at | TIMESTAMP | Last attempt time | +| delivered_at | TIMESTAMP | Successful delivery time | +| last_status_code | INTEGER | Last HTTP status code | +| last_error | TEXT | Last sanitized error | +| response_excerpt | TEXT | Truncated downstream response | + ## DNS and Configuration Tables ### DNS_Records @@ -243,6 +283,9 @@ The schema includes several indexes to optimize query performance: - `ix_api_tokens_key_hash`: On api_tokens.key_hash - `ix_api_tokens_key_prefix`: On api_tokens.key_prefix - `ix_api_tokens_active_scope`: On api_tokens.active and api_tokens.scopes +- `ix_webhook_endpoints_enabled_events`: On webhook_endpoints.enabled and webhook_endpoints.event_types +- `ix_webhook_delivery_endpoint_idempotency`: Unique on webhook_deliveries.endpoint_id and idempotency_key +- `ix_webhook_delivery_due`: On webhook_deliveries.status and webhook_deliveries.next_attempt_at - `idx_activity_logs_timestamp`: On activity_logs.timestamp - `idx_activity_logs_user_id`: On activity_logs.user_id - `idx_system_logs_timestamp`: On system_logs.timestamp diff --git a/docs/user_guide/settings.md b/docs/user_guide/settings.md index a5ff247..519d725 100644 --- a/docs/user_guide/settings.md +++ b/docs/user_guide/settings.md @@ -86,6 +86,21 @@ Apprise supports email, Slack, Teams, Discord, generic webhooks, and many other targets through the same notification field. Add each destination on a separate line. +## Webhooks + +Use **Settings** > **Webhooks** when another system needs structured DMARQ +events instead of human-readable notifications. + +1. Add a name and HTTPS endpoint URL. +2. Choose all events or one event type. +3. Save the endpoint. +4. Use **Test** to send a signed test event. +5. Inspect **Recent Deliveries** to see status, attempts, response codes, and errors. + +DMARQ signs each delivery with `X-DMARQ-Signature` and includes +`X-DMARQ-Idempotency-Key` so receivers can reject replays and deduplicate +retries. Endpoint URLs and signing secrets are encrypted at rest. + ## API Access DMARQ provides an API for integration with other systems: