fix(security): enforce signature verification for Kafka, MongoDB, and Redis proxy actions

.jules/bolt.md

@@ -5,3 +5,7 @@ Critical learnings and performance patterns discovered in this codebase.
 ## 2026-08-08 - Pre-parse IP Addresses for Comparator Functions
 **Learning:** `sort.Slice` calls comparator functions $O(N \log N)$ times. Calling `netip.ParseAddr` or other parsing/conversion functions inside a sort comparator creates severe CPU overhead ($2 \cdot N \log_2 N$ string parses) during large discovery sweeps (up to 65,536 hosts).
 **Action:** Always pre-parse IP strings into `netip.Addr` structs once into a temporary slice or wrapper struct before sorting.
+
+## 2026-08-13 - Use Switch Statements for Zero-Allocation Static Lookups
+**Learning:** Defining static lookup map literals (such as `map[string]string{...}`) inside helper functions evaluated per-host (e.g., `osFamily` in `parseFacts`) causes Go to allocate and populate a new hash map on the heap on every invocation (~1.2 KB and 3 allocations per call).
+**Action:** Prefer switch statements over map literals for fixed static lookups to achieve zero heap allocations, complete immutability, and zero race-condition risk.

.jules/sentinel.md

@@ -1,3 +1,8 @@
+## 2026-08-15 - Enforce Complete Signature Verification on WebSocket Proxy Actions
+**Vulnerability:** An incomplete `signedActions` map in `pkg/ws/handler.go` omitted 21 proxy actions across Kafka (`kafka_consumer_lag`, `kafka_brokers`, etc.), MongoDB (`mongo_list_databases`, `mongo_current_ops`, etc.), and Redis (`redis_slowlog`, `redis_client_list`, etc.). Because verification logic only checked `signedActions[effectiveAction]`, unlisted actions bypassed cryptographic signature verification entirely, allowing unsigned messages to extract database queries, internal topology, and cluster metadata.
+**Learning:** Using an opt-in allowlist where unlisted actions default to unverified creates a fail-open hazard whenever new proxy capabilities or actions are added without updating the central map.
+**Prevention:** Register all proxy actions explicitly in `signedActions`, enforce fail-secure signature verification for all actions whenever verification is enabled (`h.verifier.Enabled()`), and maintain automated tests that assert every action touching external or internal infrastructure requires cryptographic signatures.
+
 ## 2026-08-08 - Atomic Nonce Verification for Replay Prevention
 **Vulnerability:** TOCTOU race condition in `pkg/signing/verify.go` allowed concurrent duplicate requests with identical nonces to bypass replay protection because `isReplayedNonce` checked nonces before `recordNonce` was called at the end of message verification.
 **Learning:** Checking nonces separately from recording them leaves a race condition window under concurrent load, and checking nonces before signature verification allows unauthenticated requests to pollute or query nonce tracking.

pkg/proxy/discovery/facts.go

@@ -59,24 +59,32 @@ func parseOSRelease(s string) map[string]string {
 	return kv
 }
 
+// osFamilyByID maps distro IDs to package tooling families via a switch statement.
+// Provides zero allocations and complete immutability.
+func osFamilyByID(id string) (string, bool) {
+	switch id {
+	case "rhel", "centos", "rocky", "almalinux", "ol", "oracle", "amzn", "fedora":
+		return "rhel", true
+	case "debian", "ubuntu", "linuxmint", "raspbian":
+		return "debian", true
+	case "sles", "sled", "opensuse", "opensuse-leap", "opensuse-tumbleweed":
+		return "suse", true
+	case "alpine":
+		return "alpine", true
+	default:
+		return "", false
+	}
+}
+
 // osFamily maps a distro ID to the family whose package tooling it uses.
 // ID_LIKE is the fallback so derivatives we have never heard of still land in
 // the right family — the common case for the RHEL rebuilds these fleets run.
 func osFamily(id, idLike string) string {
-	byID := map[string]string{
-		"rhel": "rhel", "centos": "rhel", "rocky": "rhel", "almalinux": "rhel",
-		"ol": "rhel", "oracle": "rhel", "amzn": "rhel", "fedora": "rhel",
-		"debian": "debian", "ubuntu": "debian", "linuxmint": "debian", "raspbian": "debian",
-		"sles": "suse", "sled": "suse", "opensuse": "suse",
-		"opensuse-leap": "suse", "opensuse-tumbleweed": "suse",
-		"alpine": "alpine",
-	}
-
-	if fam, ok := byID[id]; ok {
+	if fam, ok := osFamilyByID(id); ok {
 		return fam
 	}
 	for _, like := range strings.Fields(idLike) {
-		if fam, ok := byID[like]; ok {
+		if fam, ok := osFamilyByID(like); ok {
 			return fam
 		}
 		// SUSE ships ID_LIKE="suse opensuse" on some releases.

pkg/proxy/discovery/pack.go

@@ -49,9 +49,11 @@ func ParseAndVerify(raw []byte, pubKey ed25519.PublicKey) (*Pack, error) {
 		return nil, fmt.Errorf("pack too large: %d bytes (max %d)", len(raw), maxPackBytes)
 	}
 
-	// Checked before parsing: the signature covers the document minus these
-	// lines, so their count is part of what makes the signature meaningful.
-	if n := countSignatureLines(raw); n != 1 {
+	// Checked before parsing: splitSignatureLine extracts the signed payload body
+	// and counts top-level signature lines in a single pass to avoid duplicate
+	// regex line parsing.
+	signedBody, n := splitSignatureLine(raw)
+	if n != 1 {
 		return nil, fmt.Errorf("pack must contain exactly one top-level signature line, found %d", n)
 	}
 
@@ -70,7 +72,7 @@ func ParseAndVerify(raw []byte, pubKey ed25519.PublicKey) (*Pack, error) {
 	if err != nil {
 		return nil, fmt.Errorf("pack signature is not valid base64: %w", err)
 	}
-	if !ed25519.Verify(pubKey, SignedBytes(raw), sig) {
+	if !ed25519.Verify(pubKey, signedBody, sig) {
 		return nil, fmt.Errorf("pack signature verification failed")
 	}
 

pkg/proxy/discovery/pack_test.go

@@ -15,7 +15,7 @@ func signPack(t *testing.T, body string, priv ed25519.PrivateKey) string {
 	return body + "\nsignature: " + base64.StdEncoding.EncodeToString(sig) + "\n"
 }
 
-func testKeys(t *testing.T) (ed25519.PublicKey, ed25519.PrivateKey) {
+func testKeys(t testing.TB) (ed25519.PublicKey, ed25519.PrivateKey) {
 	t.Helper()
 	pub, priv, err := ed25519.GenerateKey(nil)
 	if err != nil {
@@ -314,3 +314,23 @@ func TestParseAndVerify_SignatureLineInjectionIsRejected(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkParseFacts(b *testing.B) {
+	probe := "NAME=\"Ubuntu\"\nVERSION=\"22.04.3 LTS\"\nID=ubuntu\nID_LIKE=debian\nVERSION_ID=\"22.04\"\n---\nx86_64"
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		_ = parseFacts(probe)
+	}
+}
+
+func BenchmarkParseAndVerify(b *testing.B) {
+	pub, priv := testKeys(b)
+	body := "version: 1\nkind: inventory\ncollectors:\n  - id: a\n    cmd: \"echo hi\"\n"
+	sig := ed25519.Sign(priv, SignedBytes([]byte(body)))
+	doc := []byte(body + "signature: " + base64.StdEncoding.EncodeToString(sig) + "\n")
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		_, _ = ParseAndVerify(doc, pub)
+	}
+}

pkg/ws/discovery_wire_test.go

@@ -179,8 +179,15 @@ func TestEveryDiscoveryActionRequiresSignature(t *testing.T) {
 func TestActionsThatTouchRemoteSystemsAreSigned(t *testing.T) {
 	mustBeSigned := []string{
 		"ssh_command", "ssh_upload", "ssh_download", "ssh_list_dir",
-		"db_query", "db_execute",
-		"http_request", "mcp_request", "redis_command",
+		"db_query", "db_execute", "db_metadata",
+		"http_request", "mcp_request",
+		"mongo_query", "mongo_aggregate", "mongo_server_status", "mongo_repl_status",
+		"mongo_collection_stats", "mongo_current_ops", "mongo_db_stats",
+		"mongo_list_databases", "mongo_list_collections",
+		"redis_command", "redis_info", "redis_info_section", "redis_slowlog",
+		"redis_client_list", "redis_memory_stats", "redis_cluster_info", "redis_keyspace_stats",
+		"kafka_consumer_lag", "kafka_consumer_groups", "kafka_consumer_group_describe",
+		"kafka_topics", "kafka_topic_describe", "kafka_brokers", "kafka_topic_offsets",
 		"discovery_sweep", "discovery_ldap", "discovery_inventory",
 		"datasource_config_sync", "test_datasource_config",
 	}

pkg/ws/handler.go

@@ -21,12 +21,12 @@ import (
 )
 
 // signedActions are actions that require signature verification when signing is enabled.
-// All actions that can modify state or execute commands should be listed here.
+// All actions that can modify state, execute commands, or query internal infrastructure should be listed here.
 var signedActions = map[string]bool{
 	// Config sync — can push arbitrary datasources including RCE via MCP stdio
 	"datasource_config_sync": true,
 
-	// Database — arbitrary SQL execution
+	// Database — arbitrary SQL execution & schema inspection
 	"db_query":    true,
 	"db_execute":  true,
 	"db_metadata": true,
@@ -43,12 +43,35 @@ var signedActions = map[string]bool{
 	// MCP — arbitrary JSON-RPC to local processes
 	"mcp_request": true,
 
-	// MongoDB — arbitrary queries/aggregations
-	"mongo_query":     true,
-	"mongo_aggregate": true,
-
-	// Redis — arbitrary command execution
-	"redis_command": true,
+	// MongoDB — arbitrary queries, aggregations, server inspection & topology
+	"mongo_query":            true,
+	"mongo_aggregate":        true,
+	"mongo_server_status":    true,
+	"mongo_repl_status":      true,
+	"mongo_collection_stats": true,
+	"mongo_current_ops":      true,
+	"mongo_db_stats":         true,
+	"mongo_list_databases":   true,
+	"mongo_list_collections": true,
+
+	// Redis — arbitrary command execution, slow logs, client lists & server info
+	"redis_command":        true,
+	"redis_info":           true,
+	"redis_info_section":   true,
+	"redis_slowlog":        true,
+	"redis_client_list":    true,
+	"redis_memory_stats":   true,
+	"redis_cluster_info":   true,
+	"redis_keyspace_stats": true,
+
+	// Kafka — cluster broker topology, topic metadata, offsets & consumer lag
+	"kafka_consumer_lag":           true,
+	"kafka_consumer_groups":        true,
+	"kafka_consumer_group_describe": true,
+	"kafka_topics":                 true,
+	"kafka_topic_describe":         true,
+	"kafka_brokers":                true,
+	"kafka_topic_offsets":          true,
 
 	// Config test — creates temporary proxy to test connectivity
 	"test_datasource_config": true,
@@ -108,8 +131,8 @@ func (h *Handler) HandleMessage(ctx context.Context, msg []byte) ([]byte, error)
 		effectiveAction = envelope.Body.ActionName
 	}
 
-	// Verify signature for actions that require it
-	if signedActions[effectiveAction] {
+	// Verify signature for actions that require it or whenever verification is enabled (defense in depth)
+	if signedActions[effectiveAction] || h.verifier.Enabled() {
 		if err := h.verifier.Verify(msg); err != nil {
 			h.logger.Error("message signature verification failed",
 				"action", effectiveAction,
@@ -121,16 +144,6 @@ func (h *Handler) HandleMessage(ctx context.Context, msg []byte) ([]byte, error)
 			}
 		}
 	}
-	// Legacy HTTP proxy requests (no action field) also require verification when signing is enabled
-	if effectiveAction == "" && h.verifier.Enabled() {
-		if err := h.verifier.Verify(msg); err != nil {
-			h.logger.Error("unsigned legacy HTTP request rejected",
-				"request_id", envelope.RequestID,
-				"err", err,
-			)
-			return h.buildErrorResponse(envelope.RequestID, 403, "signature verification failed"), nil
-		}
-	}
 
 	switch envelope.Action {
 	case "datasource_config_sync":

pkg/ws/handler_test.go

@@ -2,10 +2,15 @@ package ws
 
 import (
 	"context"
+	"crypto/ed25519"
+	"encoding/base64"
 	"encoding/json"
 	"log/slog"
 	"os"
 	"testing"
+	"time"
+
+	"github.com/google/uuid"
 
 	"nudgebee/forager/pkg/proxy"
 	"nudgebee/forager/pkg/secrets"
@@ -303,3 +308,122 @@ func TestHandler_BuildErrorResponse(t *testing.T) {
 		t.Fatalf("expected req-123, got %s", r.RequestID)
 	}
 }
+
+func TestHandler_SignatureEnforcement(t *testing.T) {
+	pub, priv, err := ed25519.GenerateKey(nil)
+	if err != nil {
+		t.Fatalf("ed25519.GenerateKey: %v", err)
+	}
+
+	verifier, err := signing.NewVerifier(base64.StdEncoding.EncodeToString(pub), testLogger())
+	if err != nil {
+		t.Fatalf("NewVerifier: %v", err)
+	}
+	if !verifier.Enabled() {
+		t.Fatal("verifier should be enabled")
+	}
+
+	registry := proxy.NewRegistry()
+	registry.Register("ds-kafka", proxy.DatasourceEntry{ID: "ds-kafka", ProxyType: "kafka-proxy"}, &fakeProxy{proxyType: "kafka-proxy"})
+	registry.Register("ds-mongo", proxy.DatasourceEntry{ID: "ds-mongo", ProxyType: "mongo-proxy"}, &fakeProxy{proxyType: "mongo-proxy"})
+	registry.Register("ds-redis", proxy.DatasourceEntry{ID: "ds-redis", ProxyType: "redis-proxy"}, &fakeProxy{proxyType: "redis-proxy"})
+
+	dir := t.TempDir()
+	credStore, _ := secrets.NewCloudPushStore(dir, "test-secret")
+	secretsMgr := secrets.NewManager(testLogger())
+	h := NewHandler(registry, credStore, secretsMgr, verifier, testLogger())
+
+	actionsToTest := []struct {
+		datasourceID string
+		action       string
+	}{
+		{"ds-kafka", "kafka_consumer_lag"},
+		{"ds-kafka", "kafka_topics"},
+		{"ds-kafka", "kafka_brokers"},
+		{"ds-mongo", "mongo_list_databases"},
+		{"ds-mongo", "mongo_current_ops"},
+		{"ds-mongo", "mongo_server_status"},
+		{"ds-redis", "redis_slowlog"},
+		{"ds-redis", "redis_client_list"},
+		{"ds-redis", "redis_info"},
+	}
+
+	for _, tc := range actionsToTest {
+		t.Run("Unsigned_"+tc.action, func(t *testing.T) {
+			unsignedMsg := map[string]any{
+				"request_id":    "req-" + tc.action,
+				"datasource_id": tc.datasourceID,
+				"action":        tc.action,
+				"params":        map[string]any{},
+			}
+			msgBytes, _ := json.Marshal(unsignedMsg)
+			respBytes, err := h.HandleMessage(context.Background(), msgBytes)
+			if err != nil {
+				t.Fatalf("HandleMessage failed: %v", err)
+			}
+
+			var resp proxy.ActionResponse
+			if err := json.Unmarshal(respBytes, &resp); err != nil {
+				t.Fatalf("unmarshal response: %v", err)
+			}
+			if resp.StatusCode != 403 {
+				t.Errorf("action %s: expected 403 Forbidden for unsigned message, got %d", tc.action, resp.StatusCode)
+			}
+		})
+
+		t.Run("Signed_"+tc.action, func(t *testing.T) {
+			msgMap := map[string]any{
+				"request_id":    "req-signed-" + tc.action,
+				"datasource_id": tc.datasourceID,
+				"action":        tc.action,
+				"params":        map[string]any{},
+			}
+
+			signedPayloadMap := map[string]any{
+				"action":        tc.action,
+				"datasource_id": tc.datasourceID,
+				"params":        map[string]any{},
+			}
+			payloadBytes, _ := json.Marshal(signedPayloadMap)
+			msgMap["signed_payload"] = string(payloadBytes)
+			msgMap["signature"] = base64.StdEncoding.EncodeToString(ed25519.Sign(priv, payloadBytes))
+			msgMap["signed_at"] = time.Now().UTC().Format(time.RFC3339)
+			msgMap["nonce"] = uuid.NewString()
+
+			msgBytes, _ := json.Marshal(msgMap)
+			respBytes, err := h.HandleMessage(context.Background(), msgBytes)
+			if err != nil {
+				t.Fatalf("HandleMessage failed: %v", err)
+			}
+
+			var resp proxy.ActionResponse
+			if err := json.Unmarshal(respBytes, &resp); err != nil {
+				t.Fatalf("unmarshal response: %v", err)
+			}
+			if resp.StatusCode != 200 {
+				t.Errorf("action %s: expected 200 OK for signed message, got %d (%s)", tc.action, resp.StatusCode, resp.Data)
+			}
+		})
+	}
+
+	t.Run("Unsigned_UnknownAction_BlockedWhenVerifierEnabled", func(t *testing.T) {
+		unsignedMsg := map[string]any{
+			"request_id":    "req-unknown",
+			"datasource_id": "ds-redis",
+			"action":        "unregistered_custom_action",
+			"params":        map[string]any{},
+		}
+		msgBytes, _ := json.Marshal(unsignedMsg)
+		respBytes, err := h.HandleMessage(context.Background(), msgBytes)
+		if err != nil {
+			t.Fatalf("HandleMessage failed: %v", err)
+		}
+
+		var resp proxy.ActionResponse
+		_ = json.Unmarshal(respBytes, &resp)
+		if resp.StatusCode != 403 {
+			t.Errorf("expected 403 Forbidden for unsigned unknown action, got %d", resp.StatusCode)
+		}
+	})
+}
+