feat(webhooks): managed webhook subscriptions (phase 2) (#258)

Phase 2 of #256: DB-backed, operator-managed webhook subscriptions delivered
through the same bus as the phase-1 config webhook. Multiple endpoints, runtime
management via REST (no restart), per-endpoint event filter, signing secret, and
delivery-status observability.

- store: webhook_subscriptions table (migration 021) + CRUD + RecordDelivery.
  The HMAC secret is envelope-encrypted under the master key (per-row DEK wrapped
  under the master, secret sealed under the DEK — the cas table's scheme), so it
  never leaves the server in cleartext and is never returned by the API.
- webhook.Dispatcher refactored from a single static target to fan-out: it
  delivers each event to the optional static config target plus all matching
  managed subscriptions from a SubscriptionSource, with per-target retry and
  status recording. StoreSource resolves active subscriptions and decrypts their
  secrets at delivery time. Phase-1 config behavior is preserved.
- api: owner-scoped REST CRUD at /api/v1/webhook-subscriptions (admin sees all,
  operators see their own). The secret is write-only (omit to keep, "" to clear,
  value to set); responses expose only has_secret. config.ValidateWebhookURL is
  exported and reused for URL/SSRF validation.
- serve.go: the dispatcher now always runs (managed subs are created at runtime)
  with a store-backed source; cert.expiring still folds into the bus.
- api/openapi.yaml: the CRUD endpoints + schemas; contract tests validate the
  responses. docs/webhooks.md documents managed subscriptions.

The route-scoping registry and the cross-operator leak test are extended to
cover the new collection and single-resource routes.

Web UI for subscriptions remains a follow-up (still tracked in #256).

Refs #256

Author: juev

api/openapi.yaml

@@ -198,6 +198,86 @@ paths:
                 items:
                   $ref: "#/components/schemas/AuditEntry"
 
+  /api/v1/webhook-subscriptions:
+    get:
+      operationId: listWebhookSubscriptions
+      summary: List webhook subscriptions visible to the operator.
+      responses:
+        "200":
+          description: Subscriptions.
+          content:
+            application/json:
+              schema:
+                type: array
+                items:
+                  $ref: "#/components/schemas/WebhookSubscription"
+    post:
+      operationId: createWebhookSubscription
+      summary: Create a webhook subscription.
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: "#/components/schemas/WebhookSubscriptionRequest"
+      responses:
+        "201":
+          description: Created subscription.
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/WebhookSubscription"
+        "400":
+          $ref: "#/components/responses/Error"
+
+  /api/v1/webhook-subscriptions/{id}:
+    get:
+      operationId: getWebhookSubscription
+      summary: Get a webhook subscription by id.
+      parameters:
+        - { in: path, name: id, required: true, schema: { type: string } }
+      responses:
+        "200":
+          description: Subscription.
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/WebhookSubscription"
+        "404":
+          $ref: "#/components/responses/Error"
+    patch:
+      operationId: updateWebhookSubscription
+      summary: Update a webhook subscription.
+      parameters:
+        - { in: path, name: id, required: true, schema: { type: string } }
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: "#/components/schemas/WebhookSubscriptionRequest"
+      responses:
+        "200":
+          description: Updated subscription.
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/WebhookSubscription"
+        "400":
+          $ref: "#/components/responses/Error"
+        "404":
+          $ref: "#/components/responses/Error"
+    delete:
+      operationId: deleteWebhookSubscription
+      summary: Delete a webhook subscription.
+      parameters:
+        - { in: path, name: id, required: true, schema: { type: string } }
+      responses:
+        "204":
+          description: Deleted.
+        "404":
+          $ref: "#/components/responses/Error"
+
 # Outbound webhooks the server POSTs to an operator-configured endpoint (#256).
 # Each delivery carries X-Nebula-Event (type), X-Nebula-Delivery (id), and
 # X-Nebula-Signature (sha256=<hmac> over the raw body). The handler/scanner emit
@@ -604,3 +684,59 @@ components:
           format: date-time
         data:
           $ref: "#/components/schemas/CertExpiringData"
+
+    # --- Managed webhook subscriptions (#256 phase 2) ---
+
+    WebhookSubscriptionRequest:
+      type: object
+      required: [url]
+      properties:
+        url:
+          type: string
+        events:
+          type: ["array", "null"]
+          items:
+            type: string
+        active:
+          type: boolean
+        allow_private:
+          type: boolean
+        secret:
+          type: string
+          description: Write-only HMAC secret. Omit to keep, "" to clear, value to set.
+
+    WebhookSubscription:
+      type: object
+      required: [id, owner_operator_id, url, active, allow_private, has_secret, consecutive_failures, created_at, updated_at]
+      properties:
+        id:
+          type: string
+        owner_operator_id:
+          type: string
+        url:
+          type: string
+        events:
+          type: ["array", "null"]
+          items:
+            type: string
+        active:
+          type: boolean
+        allow_private:
+          type: boolean
+        has_secret:
+          type: boolean
+        last_delivery_at:
+          type: string
+          format: date-time
+        last_status:
+          type: string
+        last_error:
+          type: string
+        consecutive_failures:
+          type: integer
+        created_at:
+          type: string
+          format: date-time
+        updated_at:
+          type: string
+          format: date-time

docs/webhooks.md

@@ -24,7 +24,52 @@ webhooks:
 `url` is validated at startup (must be http/https; loopback/private/link-local
 targets are rejected unless `allow_private: true`).
 
-## Events (phase 1)
+The config webhook is a single static target. For multiple endpoints, runtime
+management, and per-endpoint delivery status, use **managed subscriptions**
+below — both are delivered through the same bus.
+
+## Managed subscriptions
+
+Subscriptions are operator-owned rows managed through the REST API; they need no
+config and take effect immediately (no restart). Each is one delivery target
+with its own URL, event filter, signing secret, and delivery status.
+
+```
+GET    /api/v1/webhook-subscriptions          # list (admin: all; operator: own)
+POST   /api/v1/webhook-subscriptions          # create
+GET    /api/v1/webhook-subscriptions/{id}     # get
+PATCH  /api/v1/webhook-subscriptions/{id}     # update
+DELETE /api/v1/webhook-subscriptions/{id}     # delete
+```
+
+Create body (all but `url` optional):
+
+```json
+{
+  "url": "https://hooks.example.com/team-a",
+  "events": ["host.enrolled", "cert.expiring"],
+  "active": true,
+  "allow_private": false,
+  "secret": "<hmac secret>"
+}
+```
+
+- `events` empty/omitted means all events.
+- `secret` is **write-only**: it is stored envelope-encrypted under
+  `NEBULA_MGMT_MASTER_KEY` (the same scheme as CA keys) and never returned.
+  Responses expose only `has_secret`. On update, omit `secret` to keep it, send
+  `""` to clear it, or a new value to replace it. A non-empty secret requires
+  the master key to be configured.
+- `url` is SSRF-validated like the config webhook; `allow_private: true` opts a
+  private/loopback target in.
+
+Each subscription tracks `last_delivery_at`, `last_status` (`ok`/`failed`),
+`last_error`, and `consecutive_failures` for observability.
+
+## Events
+
+| Event | Fires when | `data` fields |
+|---|---|---|
 
 | Event | Fires when | `data` fields |
 |---|---|---|
@@ -76,9 +121,8 @@ Reject deliveries that do not match.
   as a notification stream, not a system of record — the audit log remains the
   authoritative history.
 
-## Not in phase 1
+## Not yet
 
-- Managed subscriptions (multiple endpoints, CRUD, per-subscription status) —
-  phase 1 is a single config-driven subscription.
+- A Web UI for managing subscriptions (the REST API exists today).
 - `ca.expiring` and other CA-lifecycle events.
 - A persistent dead-letter queue and delivery dashboards.

internal/api/metrics.go

@@ -72,6 +72,9 @@ const (
 	auditOperatorAPIKeyCreate      = "operator.api_key.create"
 	auditOperatorAPIKeyRevoke      = "operator.api_key.revoke"
 	auditSettingsEnforce2FA        = "settings.enforce_2fa"
+	auditWebhookSubCreate          = "webhook_subscription.create"
+	auditWebhookSubUpdate          = "webhook_subscription.update"
+	auditWebhookSubDelete          = "webhook_subscription.delete"
 )
 
 // Audit reason codes used in the `details` field of host.auth.failed audit
@@ -197,6 +200,7 @@ func newMetrics(s store.Store) *metrics {
 		auditOperatorCreate, auditOperatorDisable, auditOperatorEnable,
 		auditOperatorAPIKeyCreate, auditOperatorAPIKeyRevoke,
 		auditSettingsEnforce2FA,
+		auditWebhookSubCreate, auditWebhookSubUpdate, auditWebhookSubDelete,
 	} {
 		m.auditEntries.WithLabelValues(action).Add(0)
 	}

internal/api/scoping_property_test.go

@@ -63,9 +63,10 @@ var protectedGETScoping = map[string]listScoping{
 	"/api/v1/agent/updates": publicRoute, // agent poll, signed-poll auth — not operator bearer
 
 	// Collection reads — must scope non-admins to owned CAs.
-	"/api/v1/networks": scopedToOwner,
-	"/api/v1/hosts":    scopedToOwner,
-	"/api/v1/cas":      scopedToOwner,
+	"/api/v1/networks":              scopedToOwner,
+	"/api/v1/hosts":                 scopedToOwner,
+	"/api/v1/cas":                   scopedToOwner,
+	"/api/v1/webhook-subscriptions": scopedToOwner,
 
 	// Admin-only reads — non-admin gets 403, no tenant data in the body.
 	"/api/v1/operators":               adminOnly,
@@ -75,10 +76,11 @@ var protectedGETScoping = map[string]listScoping{
 	"/api/v1/settings":                adminOnly,
 
 	// Single-resource reads — guarded per-row by canAccess*.
-	"/api/v1/networks/{id}":          singleResource,
-	"/api/v1/networks/{id}/firewall": singleResource,
-	"/api/v1/hosts/{id}":             singleResource,
-	"/api/v1/cas/{id}":               singleResource,
+	"/api/v1/networks/{id}":              singleResource,
+	"/api/v1/networks/{id}/firewall":     singleResource,
+	"/api/v1/hosts/{id}":                 singleResource,
+	"/api/v1/cas/{id}":                   singleResource,
+	"/api/v1/webhook-subscriptions/{id}": singleResource,
 }
 
 // TestProtectedGETRoutesAreClassified fails the moment setupRoutes registers
@@ -116,12 +118,16 @@ func TestProtectedGETRoutesAreClassified(t *testing.T) {
 func TestListEndpointsScopeToOwner(t *testing.T) {
 	srv, st := newTestServer(t)
 
-	keyA, _, caA := createOperatorWithCA(t, srv)
-	keyB, _, caB := createOperatorWithCA(t, srv)
+	keyA, opA, caA := createOperatorWithCA(t, srv)
+	keyB, opB, caB := createOperatorWithCA(t, srv)
 	// Distinct CIDRs/IPs so the seed is robust to any nebula_ip uniqueness
 	// scheme; ownership is what the test cares about, not addressing.
 	seedNetworkAndHost(t, st, caA.ID, "a", "10.10.0.0/24", "10.10.0.10")
 	seedNetworkAndHost(t, st, caB.ID, "b", "10.20.0.0/24", "10.20.0.10")
+	// Webhook subscriptions carry no ca_id, so embed the owner's CA id in the
+	// URL to satisfy the generic owner-leak assertions below.
+	seedWebhookSub(t, st, opA.ID, "https://example.com/"+caA.ID)
+	seedWebhookSub(t, st, opB.ID, "https://example.com/"+caB.ID)
 
 	for route, scoping := range protectedGETScoping {
 		switch scoping {
@@ -178,6 +184,20 @@ func concretePath(route string) string {
 	return strings.ReplaceAll(route, "{id}", "test-admin")
 }
 
+// seedWebhookSub inserts a webhook subscription owned by ownerID directly via
+// the store, bypassing the create handler's owner derivation so cross-operator
+// fixtures can be set up.
+func seedWebhookSub(t *testing.T, st *store.SQLiteStore, ownerID, url string) {
+	t.Helper()
+	require.NoError(t, st.CreateWebhookSubscription(context.Background(), &models.WebhookSubscription{
+		ID:              uuid.New().String(),
+		OwnerOperatorID: ownerID,
+		URL:             url,
+		Active:          true,
+		CreatedAt:       time.Now(),
+	}))
+}
+
 // seedNetworkAndHost inserts a network and a host bound to caID directly via
 // the store, bypassing the create handlers (which would re-derive ownership
 // from the caller). This lets the scoping tests set up cross-operator

internal/api/server.go

@@ -239,6 +239,12 @@ func (s *Server) setupRoutes() {
 		r.Post("/api/v1/cas/{id}/rotate", s.handleRotateCA)
 		r.Get("/api/v1/settings", s.handleGetSettings)
 		r.Patch("/api/v1/settings", s.handlePatchSettings)
+
+		r.Get("/api/v1/webhook-subscriptions", s.handleListWebhookSubscriptions)
+		r.Post("/api/v1/webhook-subscriptions", s.handleCreateWebhookSubscription)
+		r.Get("/api/v1/webhook-subscriptions/{id}", s.handleGetWebhookSubscription)
+		r.Patch("/api/v1/webhook-subscriptions/{id}", s.handleUpdateWebhookSubscription)
+		r.Delete("/api/v1/webhook-subscriptions/{id}", s.handleDeleteWebhookSubscription)
 	})
 
 	s.router = r

internal/api/webhook_subs_contract_test.go

@@ -0,0 +1,50 @@
+package api
+
+import (
+	"bytes"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+)
+
+// TestContract_WebhookSubscriptions validates the CRUD responses against the
+// OpenAPI spec, so the managed-subscription API cannot drift from its contract.
+func TestContract_WebhookSubscriptions(t *testing.T) {
+	v := loadContract(t)
+	srv, _ := newTestServer(t)
+
+	// Create.
+	body := `{"url":"https://hooks.example.com/c","events":["host.enrolled","cert.rotated"],"secret":"s3cret"}`
+	req := httptest.NewRequest(http.MethodPost, "/api/v1/webhook-subscriptions", bytes.NewBufferString(body))
+	authRequest(req)
+	rec := serve(srv, req)
+	if rec.Code != http.StatusCreated {
+		t.Fatalf("create: %d / %s", rec.Code, rec.Body.String())
+	}
+	assertContract(t, v, req, rec)
+	var sub struct {
+		ID string `json:"id"`
+	}
+	_ = json.Unmarshal(rec.Body.Bytes(), &sub)
+
+	// List.
+	lreq := httptest.NewRequest(http.MethodGet, "/api/v1/webhook-subscriptions", nil)
+	authRequest(lreq)
+	assertContract(t, v, lreq, serve(srv, lreq))
+
+	// Get.
+	greq := httptest.NewRequest(http.MethodGet, "/api/v1/webhook-subscriptions/"+sub.ID, nil)
+	authRequest(greq)
+	assertContract(t, v, greq, serve(srv, greq))
+
+	// Update.
+	ureq := httptest.NewRequest(http.MethodPatch, "/api/v1/webhook-subscriptions/"+sub.ID,
+		bytes.NewBufferString(`{"url":"https://hooks.example.com/d","active":false}`))
+	authRequest(ureq)
+	urec := serve(srv, ureq)
+	if urec.Code != http.StatusOK {
+		t.Fatalf("update: %d / %s", urec.Code, urec.Body.String())
+	}
+	assertContract(t, v, ureq, urec)
+}