pigway-frame-flow/pigway_frame_flow/extensions/webhooks.py

"""Signed, bounded, durable event delivery; independent of frame writes."""

import hashlib
import hmac
import json
import secrets
import time
from datetime import UTC, datetime, timedelta
from uuid import uuid4

from fastapi import Depends
from pydantic import Field

from pigway_frame_flow.core.auth import require
from pigway_frame_flow.errors import ErrorCode as E
from pigway_frame_flow.errors import FlowError
from pigway_frame_flow.models import StrictModel
from pigway_frame_flow.storage.database import encode, now_iso


class WebhookInput(StrictModel):
    callback_url: str = Field(max_length=2048)
    event_types: list[str] = Field(min_length=1, max_length=20)
    enabled: bool = True


class Webhooks:
    def __init__(self, service):
        self.service = service
        self.last = 0.0
        service.db.conn.executescript("""CREATE TABLE IF NOT EXISTS webhook_deliveries (
        subscription_id TEXT,event_id TEXT,status TEXT,attempts INTEGER DEFAULT 0,next_at TEXT,error_code TEXT,
        PRIMARY KEY(subscription_id,event_id));""")

    def tick(self):
        if time.monotonic() - self.last < 60:
            return
        self.last = time.monotonic()
        db = self.service.db
        for sub in db.all("SELECT * FROM webhook_subscriptions WHERE enabled=1"):
            types = json.loads(sub["event_types"])
            for event in db.all(
                "SELECT * FROM events e WHERE created_at>=? AND NOT EXISTS (SELECT 1 FROM webhook_deliveries d WHERE d.event_id=e.id AND d.subscription_id=?) ORDER BY created_at LIMIT 500",
                (sub["created_at"], sub["id"]),
            ):
                if event["event_type"] not in types:
                    continue
                db.execute(
                    "INSERT OR IGNORE INTO webhook_deliveries VALUES(?,?,'QUEUED',0,?,NULL)",
                    (sub["id"], event["id"], now_iso()),
                )
        for row in db.all(
            "SELECT d.*,s.callback_url FROM webhook_deliveries d JOIN webhook_subscriptions s ON s.id=d.subscription_id WHERE s.enabled=1 AND d.status IN ('QUEUED','RETRY') AND d.next_at<=? ORDER BY d.next_at LIMIT 2",
            (now_iso(),),
        ):
            event = db.one("SELECT * FROM events WHERE id=?", (row["event_id"],))
            if not event:
                continue
            keyfile = self.service.settings.config_dir / (
                "webhook-" + row["subscription_id"] + ".secret"
            )
            try:
                key = keyfile.read_text().strip()
                payload = encode(
                    {
                        "id": event["id"],
                        "type": event["event_type"],
                        "timestamp": event["created_at"],
                        "data": json.loads(event["payload"]),
                    }
                ).encode()
                signature = hmac.new(key.encode(), payload, hashlib.sha256).hexdigest()
                self.service.assets.fetch_url(
                    row["callback_url"],
                    method="POST",
                    body=payload,
                    headers={
                        "Content-Type": "application/json",
                        "X-Frame-Flow-Event-ID": event["id"],
                        "X-Frame-Flow-Signature": "sha256=" + signature,
                    },
                    limit=64 * 1024,
                    accepted_status=(200, 201, 202, 204),
                )
                status, error = "DELIVERED", None
            except Exception:
                status, error = (
                    ("FAILED" if row["attempts"] >= 2 else "RETRY"),
                    "DELIVERY_FAILED",
                )
            db.execute(
                "UPDATE webhook_deliveries SET status=?,attempts=attempts+1,next_at=?,error_code=? WHERE subscription_id=? AND event_id=?",
                (
                    status,
                    (
                        datetime.now(UTC) + timedelta(seconds=60 * 2 ** row["attempts"])
                    ).isoformat(),
                    error,
                    row["subscription_id"],
                    row["event_id"],
                ),
            )


def install_webhooks(app, service):
    worker = Webhooks(service)
    service.webhooks = worker
    db = service.db

    @app.get(
        "/api/v1/webhooks", tags=["Webhooks"], dependencies=[Depends(require("admin"))]
    )
    def list_hooks():
        return [dict(row) for row in db.all("SELECT * FROM webhook_subscriptions")]

    @app.post(
        "/api/v1/webhooks",
        status_code=201,
        tags=["Webhooks"],
        dependencies=[Depends(require("admin"))],
    )
    def add_hook(payload: WebhookInput):
        from urllib.parse import urlsplit

        parsed = urlsplit(payload.callback_url)
        if (
            parsed.scheme != "https"
            or not parsed.hostname
            or parsed.username
            or parsed.password
            or parsed.port not in (None, 443)
            or parsed.hostname not in service.settings.image_url_hosts
        ):
            raise FlowError(
                E.INVALID_REQUEST, "Webhook requires an explicitly allowed HTTPS host"
            )
        allowed = {
            "operation.completed",
            "operation.failed",
            "display.changed",
            "device.online",
            "device.offline",
            "device.low_battery",
        }
        if set(payload.event_types) - allowed:
            raise FlowError(E.INVALID_REQUEST, "Unsupported event type")
        id = uuid4().hex
        key = secrets.token_urlsafe(32)
        import os

        path = service.settings.config_dir / ("webhook-" + id + ".secret")
        with os.fdopen(
            os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), "w"
        ) as f:
            f.write(key)
        db.execute(
            "INSERT INTO webhook_subscriptions VALUES(?,?,?,?,?,?)",
            (
                id,
                "admin",
                payload.callback_url,
                encode(payload.event_types),
                int(payload.enabled),
                now_iso(),
            ),
        )
        service.settings.additional_secrets.append(key)
        return {"id": id, "signing_secret": key}

    @app.delete(
        "/api/v1/webhooks/{id}",
        status_code=204,
        tags=["Webhooks"],
        dependencies=[Depends(require("admin"))],
    )
    def delete_hook(id: str):
        if not db.one("SELECT id FROM webhook_subscriptions WHERE id=?", (id,)):
            raise FlowError(E.NOT_FOUND, "Webhook not found", 404)
        db.execute("DELETE FROM webhook_subscriptions WHERE id=?", (id,))
        (service.settings.config_dir / ("webhook-" + id + ".secret")).unlink(
            missing_ok=True
        )

    @app.get(
        "/api/v1/webhooks/{id}/deliveries",
        tags=["Webhooks"],
        dependencies=[Depends(require("admin"))],
    )
    def deliveries(id: str):
        return [
            dict(row)
            for row in db.all(
                "SELECT * FROM webhook_deliveries WHERE subscription_id=? ORDER BY next_at DESC LIMIT 100",
                (id,),
            )
        ]