webhook: per-topic filter on the bridge + pilotctl --topics

teovl · teovl · commit f7e83ffa828c · 2026-05-12T18:18:17.000-07:00
The webhook plugin subscribed to "*" and forwarded every bus event,
including high-volume internal topics (tunnel.relay_activated,
conn.established, security.*, pubsub.*). Operators who only wanted
the human-meaningful events (message.received, file.received,
handshake.received, trust.changed) had no way to filter at the source.

This change:

- Service gains a topics map&lt;string, struct{}&gt; read by the bridge
  goroutine before each Emit. Nil = forward all = pre-filter default
  (backwards-compatible — existing webhook configs unchanged).
- Persisted to ~/.pilot/webhook_topics (one topic per line, # comments
  and blank lines ignored). Survives daemon restart.
- WebhookManager interface extended: SetTopics / Topics. All four
  existing test stubs updated.
- New IPC opcodes CmdSetWebhookTopics / CmdSetWebhookTopicsOK
  (0x1D / 0x1E). Driver gets a SetWebhookTopics method.
- pilotctl set-webhook accepts --topics t1,t2,... and --clear-topics.
  The URL and topic filter are independent IPCs so neither flag
  mutates the other.

Tests cover topicSet normalization, persist round-trip, comment
handling, and the SetTopics/Topics operator surface. Existing webhook
tests still pass.
diff --git a/cmd/daemon/main.go b/cmd/daemon/main.go
@@ -230,7 +230,9 @@ func main() {
 // than in the plugin to keep the plugin free of pkg/daemon imports.
 type webhookManagerAdapter struct{ svc *webhook.Service }
 
-func (a webhookManagerAdapter) SetURL(url string) { a.svc.SetURL(url) }
+func (a webhookManagerAdapter) SetURL(url string)            { a.svc.SetURL(url) }
+func (a webhookManagerAdapter) SetTopics(topics []string)    { a.svc.SetTopics(topics) }
+func (a webhookManagerAdapter) Topics() []string             { return a.svc.Topics() }
 func (a webhookManagerAdapter) Stats() daemon.WebhookStats {
 	s := a.svc.Stats()
 	return daemon.WebhookStats{Dropped: s.Dropped, CircuitSkips: s.CircuitSkips}
diff --git a/cmd/pilotctl/main.go b/cmd/pilotctl/main.go
@@ -2896,17 +2896,48 @@ func cmdClearHostname() {
 }
 
 func cmdSetWebhook(args []string) {
-	if len(args) < 1 {
-		fatalCode("invalid_argument", "usage: pilotctl set-webhook <url>")
+	// Flags accepted after the positional <url>:
+	//   --topics <a,b,c>   only forward events with these topics (empty = forward all)
+	//   --clear-topics     clear any previously-set topic filter (= forward all)
+	var url string
+	var topics []string
+	var topicsSet, clearTopics bool
+	for i := 0; i < len(args); i++ {
+		switch args[i] {
+		case "--topics":
+			if i+1 >= len(args) {
+				fatalCode("invalid_argument", "--topics requires a comma-separated list")
+			}
+			topicsSet = true
+			for _, t := range strings.Split(args[i+1], ",") {
+				if tt := strings.TrimSpace(t); tt != "" {
+					topics = append(topics, tt)
+				}
+			}
+			i++
+		case "--clear-topics":
+			clearTopics = true
+		default:
+			if url == "" && !strings.HasPrefix(args[i], "--") {
+				url = args[i]
+			}
+		}
+	}
+	if url == "" {
+		fatalCode("invalid_argument", "usage: pilotctl set-webhook <url> [--topics t1,t2,...] [--clear-topics]")
 	}
-	url := args[0]
 	if !strings.HasPrefix(url, "http://") && !strings.HasPrefix(url, "https://") {
 		fatalCode("invalid_argument", "webhook URL must start with http:// or https://")
 	}
 
 	// Persist to config so it survives daemon restart
 	cfg := loadConfig()
 	cfg["webhook"] = url
+	if topicsSet {
+		cfg["webhook_topics"] = topics
+	} else if clearTopics {
+		delete(cfg, "webhook_topics")
+	}
 	if err := saveConfig(cfg); err != nil {
 		fatalCode("internal", "save config: %v", err)
 	}
@@ -2916,19 +2947,38 @@ func cmdSetWebhook(args []string) {
 	d, err := driver.Connect(getSocket())
 	if err == nil {
 		_, err = d.SetWebhook(url)
-		d.Close()
 		if err == nil {
 			applied = true
+			// Topic filter is a separate IPC call; apply when --topics
+			// or --clear-topics was given. Don't touch the daemon's
+			// current topic filter if neither flag was set.
+			if topicsSet {
+				_, _ = d.SetWebhookTopics(topics)
+			} else if clearTopics {
+				_, _ = d.SetWebhookTopics(nil)
+			}
 		}
+		d.Close()
 	}
 
 	if jsonOutput {
-		outputOK(map[string]interface{}{
+		out := map[string]interface{}{
 			"webhook": url,
 			"applied": applied,
-		})
+		}
+		if topicsSet {
+			out["topics"] = topics
+		} else if clearTopics {
+			out["topics_cleared"] = true
+		}
+		outputOK(out)
 	} else {
 		fmt.Printf("webhook set: %s\n", url)
+		if topicsSet {
+			fmt.Printf("topic filter: %s\n", strings.Join(topics, ","))
+		} else if clearTopics {
+			fmt.Printf("topic filter cleared (forwarding all events)\n")
+		}
 		if applied {
 			fmt.Printf("applied to running daemon\n")
 		} else {
diff --git a/pkg/daemon/contract.go b/pkg/daemon/contract.go
@@ -145,6 +145,15 @@ type WebhookManager interface {
 	// delivery (no-op until set again).
 	SetURL(url string)
 
+	// SetTopics hot-swaps the event-topic allow-list. Empty/nil =
+	// forward every event (legacy pre-filter behavior). A non-empty
+	// list filters at the bridge before each Emit.
+	SetTopics(topics []string)
+
+	// Topics returns a snapshot of the current allow-list for
+	// introspection. Nil/empty = "forward all."
+	Topics() []string
+
 	// Stats returns dispatcher counters for daemon Info. All-zero
 	// when no client is configured (nil-safe at the plugin level).
 	Stats() WebhookStats
diff --git a/pkg/daemon/daemon.go b/pkg/daemon/daemon.go
@@ -1772,6 +1772,24 @@ func (d *Daemon) SetWebhookURL(url string) {
 	d.webhookManager.SetURL(url)
 }
 
+// SetWebhookTopics hot-swaps the event-topic allow-list. Empty/nil =
+// forward all events (legacy). No-op if no manager is registered.
+func (d *Daemon) SetWebhookTopics(topics []string) {
+	if d.webhookManager == nil {
+		return
+	}
+	d.webhookManager.SetTopics(topics)
+}
+
+// WebhookTopics returns the currently-configured topic allow-list (or
+// nil for "forward all"). Used by daemon Info / pilotctl introspection.
+func (d *Daemon) WebhookTopics() []string {
+	if d.webhookManager == nil {
+		return nil
+	}
+	return d.webhookManager.Topics()
+}
+
 // RegisterWebhookManager wires the L11 webhook plugin into the daemon
 // (T4.1). cmd/daemon (composition root) constructs the plugin and
 // calls this with an adapter satisfying the daemon-local interface.
diff --git a/pkg/daemon/ipc.go b/pkg/daemon/ipc.go
@@ -47,9 +47,11 @@ const (
 	CmdDeregisterOK      byte = 0x18
 	CmdSetTags           byte = 0x19
 	CmdSetTagsOK         byte = 0x1A
-	CmdSetWebhook        byte = 0x1B
-	CmdSetWebhookOK      byte = 0x1C
-	CmdNetwork           byte = 0x1F
+	CmdSetWebhook         byte = 0x1B
+	CmdSetWebhookOK       byte = 0x1C
+	CmdSetWebhookTopics   byte = 0x1D
+	CmdSetWebhookTopicsOK byte = 0x1E
+	CmdNetwork            byte = 0x1F
 	CmdNetworkOK         byte = 0x20
 	CmdHealth            byte = 0x21
 	CmdHealthOK          byte = 0x22
@@ -630,6 +632,8 @@ func (s *IPCServer) dispatch(conn *ipcConn, cmd byte, reqID uint64, payload []by
 		s.handleSetTags(conn, reqID, payload)
 	case CmdSetWebhook:
 		s.handleSetWebhook(conn, reqID, payload)
+	case CmdSetWebhookTopics:
+		s.handleSetWebhookTopics(conn, reqID, payload)
 	case CmdNetwork:
 		s.handleNetwork(conn, reqID, payload)
 	case CmdHealth:
@@ -1179,6 +1183,25 @@ func (s *IPCServer) handleSetWebhook(conn *ipcConn, reqID uint64, payload []byte
 	}
 }
 
+// handleSetWebhookTopics swaps the event-topic allow-list. Payload is
+// JSON: either an array of strings (the topic names) or empty/null
+// (clears the filter — forwards every event, legacy default).
+func (s *IPCServer) handleSetWebhookTopics(conn *ipcConn, reqID uint64, payload []byte) {
+	var topics []string
+	if len(payload) > 0 {
+		if err := json.Unmarshal(payload, &topics); err != nil {
+			s.sendError(conn, reqID, fmt.Sprintf("invalid topics payload: %v", err))
+			return
+		}
+	}
+	s.daemon.SetWebhookTopics(topics)
+	result := map[string]interface{}{"topics": topics}
+	data, _ := json.Marshal(result)
+	if err := conn.writeReply(CmdSetWebhookTopicsOK, reqID, data); err != nil {
+		slog.Debug("IPC set_webhook_topics reply failed", "err", err)
+	}
+}
+
 // Handshake IPC sub-commands
 const (
 	SubHandshakeSend    byte = 0x01
diff --git a/pkg/daemon/zz_info_health_metrics_bug_test.go b/pkg/daemon/zz_info_health_metrics_bug_test.go
@@ -22,6 +22,8 @@ import (
 type infoFakeWebhookManager struct{ stats WebhookStats }
 
 func (f *infoFakeWebhookManager) SetURL(string)       {}
+func (f *infoFakeWebhookManager) SetTopics([]string)  {}
+func (f *infoFakeWebhookManager) Topics() []string    { return nil }
 func (f *infoFakeWebhookManager) Stats() WebhookStats { return f.stats }
 
 func TestInfoZerosWebhookCountersWhenNoManagerRegistered(t *testing.T) {
diff --git a/pkg/daemon/zz_ipc_test.go b/pkg/daemon/zz_ipc_test.go
@@ -357,6 +357,8 @@ type fakeWebhookManager struct {
 }
 
 func (f *fakeWebhookManager) SetURL(url string)   { f.urls = append(f.urls, url) }
+func (f *fakeWebhookManager) SetTopics([]string)  {}
+func (f *fakeWebhookManager) Topics() []string    { return nil }
 func (f *fakeWebhookManager) Stats() WebhookStats { return WebhookStats{} }
 
 func TestHandleSetWebhookEmptyPayloadClearsWebhookAndRepliesOK(t *testing.T) {
diff --git a/pkg/daemon/zz_retx_test.go b/pkg/daemon/zz_retx_test.go
@@ -21,6 +21,8 @@ import (
 type retxFakeWebhookManager struct{ urls []string }
 
 func (f *retxFakeWebhookManager) SetURL(url string)   { f.urls = append(f.urls, url) }
+func (f *retxFakeWebhookManager) SetTopics([]string)  {}
+func (f *retxFakeWebhookManager) Topics() []string    { return nil }
 func (f *retxFakeWebhookManager) Stats() WebhookStats { return WebhookStats{} }
 
 func TestSetWebhookURLDelegatesToRegisteredManager(t *testing.T) {
diff --git a/pkg/driver/driver.go b/pkg/driver/driver.go
@@ -317,6 +317,20 @@ func (d *Driver) SetWebhook(url string) (map[string]interface{}, error) {
 	return d.jsonRPC(msg, cmdSetWebhookOK, "set_webhook")
 }
 
+// SetWebhookTopics sets or clears the daemon's webhook event-topic
+// allow-list. Empty / nil slice clears the filter (forward every
+// event, legacy default). Non-empty slice filters at the bridge.
+func (d *Driver) SetWebhookTopics(topics []string) (map[string]interface{}, error) {
+	body, err := json.Marshal(topics)
+	if err != nil {
+		return nil, err
+	}
+	msg := make([]byte, 1+len(body))
+	msg[0] = cmdSetWebhookTopics
+	copy(msg[1:], body)
+	return d.jsonRPC(msg, cmdSetWebhookTopicsOK, "set_webhook_topics")
+}
+
 // RotateKey asks the daemon to rotate its Ed25519 identity at the registry.
 // The daemon generates a new keypair, signs proof of the current key, calls
 // registry.RotateKey, then atomically swaps and persists the new identity.
diff --git a/pkg/driver/ipc.go b/pkg/driver/ipc.go
@@ -41,9 +41,11 @@ const (
 	cmdDeregisterOK      byte = 0x18
 	cmdSetTags           byte = 0x19
 	cmdSetTagsOK         byte = 0x1A
-	cmdSetWebhook        byte = 0x1B
-	cmdSetWebhookOK      byte = 0x1C
-	cmdNetwork           byte = 0x1F
+	cmdSetWebhook         byte = 0x1B
+	cmdSetWebhookOK       byte = 0x1C
+	cmdSetWebhookTopics   byte = 0x1D
+	cmdSetWebhookTopicsOK byte = 0x1E
+	cmdNetwork            byte = 0x1F
 	cmdNetworkOK         byte = 0x20
 	cmdHealth            byte = 0x21
 	cmdHealthOK          byte = 0x22
@@ -183,7 +185,7 @@ func (c *ipcClient) readLoop() {
 			c.dispatchPush(cmd, payload)
 		case cmdBindOK, cmdDialOK, cmdError, cmdInfoOK, cmdHandshakeOK,
 			cmdResolveHostnameOK, cmdSetHostnameOK, cmdSetVisibilityOK,
-			cmdDeregisterOK, cmdSetTagsOK, cmdSetWebhookOK, cmdNetworkOK,
+			cmdDeregisterOK, cmdSetTagsOK, cmdSetWebhookOK, cmdSetWebhookTopicsOK, cmdNetworkOK,
 			cmdHealthOK, cmdManagedOK, cmdRotateKeyOK, cmdBroadcastOK:
 			// Known response cmds: route to pending for the in-flight sendAndWait.
 			select {
diff --git a/plugins/webhook/service.go b/plugins/webhook/service.go
@@ -20,6 +20,10 @@ import (
 //
 // The URL can be hot-swapped at runtime via SetURL — that path is
 // invoked from cmd/daemon when IPC's set-webhook handler fires.
+//
+// As of the topic-filter change, the bridge optionally consults a
+// per-Service allow-list before forwarding; empty allow-list = legacy
+// "forward every event" behavior. See SetTopics for the operator path.
 type Service struct {
 	mu sync.Mutex
 
@@ -30,6 +34,11 @@ type Service struct {
 	client *Client
 	cancel func()
 	done   chan struct{}
+
+	// topics is the optional event-topic allow-list. nil/empty = forward
+	// all events (legacy behavior). When set, the bridge loop matches
+	// ev.Topic against this set and drops misses. Read/written under mu.
+	topics map[string]struct{}
 }
 
 // Stats is the snapshot of per-Client counters needed by daemon's
@@ -69,10 +78,31 @@ func (s *Service) Start(_ context.Context, deps coreapi.Deps) error {
 			url = persisted
 		}
 	}
+	// Load the persisted topic allow-list, if any. Missing file = nil
+	// map = forward-all (pre-filter behavior).
+	if topics, err := LoadPersistedTopics(); err == nil && len(topics) > 0 {
+		s.topics = topicSet(topics)
+	}
 	s.startClientLocked(url)
 	return nil
 }
 
+// topicSet converts a list into the lookup set used on the hot path.
+// Caller owns the input slice; the returned map is a fresh allocation.
+func topicSet(topics []string) map[string]struct{} {
+	m := make(map[string]struct{}, len(topics))
+	for _, t := range topics {
+		if t == "" {
+			continue
+		}
+		m[t] = struct{}{}
+	}
+	if len(m) == 0 {
+		return nil
+	}
+	return m
+}
+
 func (s *Service) Stop(_ context.Context) error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
@@ -98,6 +128,41 @@ func (s *Service) SetURL(url string) {
 	}
 }
 
+// SetTopics swaps the event-topic allow-list. An empty / nil slice
+// disables filtering (forward every event — pre-filter default).
+// Persists to ~/.pilot/webhook_topics so it survives daemon restart.
+// The bridge goroutine doesn't have to be restarted — it reads the
+// filter from the Service on every event under s.mu.
+func (s *Service) SetTopics(topics []string) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.topics = topicSet(topics)
+	if err := SavePersistedTopics(topics); err != nil {
+		slog.Warn("failed to persist webhook topics", "err", err)
+	}
+	if len(s.topics) == 0 {
+		slog.Info("webhook topic filter cleared (forwarding all events)")
+	} else {
+		slog.Info("webhook topic filter updated", "count", len(s.topics))
+	}
+}
+
+// Topics returns the current allow-list as a sorted slice for
+// inspection by daemon Info() / pilotctl introspection. Nil/empty
+// return means "forward all."
+func (s *Service) Topics() []string {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if len(s.topics) == 0 {
+		return nil
+	}
+	out := make([]string, 0, len(s.topics))
+	for t := range s.topics {
+		out = append(out, t)
+	}
+	return out
+}
+
 // Stats returns dispatcher counters for the daemon's Info() response.
 // Nil-safe across "no client configured" and "service stopped."
 func (s *Service) Stats() Stats {
@@ -138,6 +203,7 @@ func (s *Service) startClientLocked(url string) {
 	s.done = done
 	wc := s.client
 	events := s.deps.Events
+	svc := s // captured for the filter probe; the map is read under svc.mu
 	go func() {
 		defer close(done)
 		// L11 panic boundary: a panic in Emit (or in the channel
@@ -146,6 +212,17 @@ func (s *Service) startClientLocked(url string) {
 		// restart the bridge.
 		defer coreapi.RecoverPlugin("webhook", "bridgeLoop", events, nil)
 		for ev := range ch {
+			// Topic filter, if configured. Holding mu briefly is cheap
+			// here — the map lookup is O(1) and contention is bounded
+			// by the rare SetTopics call. nil map = forward everything.
+			svc.mu.Lock()
+			topics := svc.topics
+			svc.mu.Unlock()
+			if topics != nil {
+				if _, ok := topics[ev.Topic]; !ok {
+					continue
+				}
+			}
 			wc.Emit(ev.Topic, ev.Payload)
 		}
 	}()
diff --git a/plugins/webhook/webhook.go b/plugins/webhook/webhook.go
diff --git a/plugins/webhook/zz_topic_filter_test.go b/plugins/webhook/zz_topic_filter_test.go
diff --git a/tests/testenv.go b/tests/testenv.go

Original file line number	Diff line number	Diff line change
`@@ -357,6 +357,8 @@ type fakeWebhookManager struct {`
`357`	`357`	`}`
`358`	`358`
`359`	`359`	`func (f *fakeWebhookManager) SetURL(url string) { f.urls = append(f.urls, url) }`
	`360`	`+func (f *fakeWebhookManager) SetTopics([]string) {}`
	`361`	`+func (f *fakeWebhookManager) Topics() []string { return nil }`
`360`	`362`	`func (f *fakeWebhookManager) Stats() WebhookStats { return WebhookStats{} }`
`361`	`363`
`362`	`364`	`func TestHandleSetWebhookEmptyPayloadClearsWebhookAndRepliesOK(t *testing.T) {`