cli: add homelab edges — who-talks-to-whom investigation helper (v0.9.0)

Makes the goldmane_edges east-west trail (ADR-0014) reachable during incident
investigations without remembering the DB/creds/SQL. New top-level verb:

  homelab edges --ns <ns>         edges touching <ns> (either direction)
  homelab edges --src/--dst <ns>  directional egress / ingress peers
  homelab edges --peers-of <ns>   distinct peer namespaces of <ns>
  homelab edges --new-since 24h   first seen since a duration or date (YYYY-MM-DD)
  homelab edges --denied          only action='deny' (blocked / lateral movement)
  homelab edges --json --limit N  machine-readable / row cap (default 200)

Filters render to a single read-only SELECT against the `edge` table, run via
the dbaas CNPG primary pod (same exec path as `k8s db`). Namespace values are
validated to the k8s name charset (injection guard) before they reach SQL.

TDD: edges_test.go covers flag parsing, query building (each filter, AND
combination, peers-of shape, JSON wrapper), the new-since duration/date parser,
and namespace-validation / injection rejection. Smoke-tested live: --peers-of,
--new-since 24h, --denied, and --json all return correct rows.

Docs: runbook query section now leads with the CLI; cli/README gains a v0.9
section. VERSION v0.8.2 -> v0.9.0.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

cli/README.md

@@ -202,6 +202,21 @@ runs on the devvm, `setInputFiles` streams local files to the remote browser ove
 CDP — no `chmod`/staging-dir workaround. See `docs/architecture/chrome-service.md`
 and `docs/adr/0013`.
 
+### v0.9 verbs — edges (east-west "who-talks-to-whom" trail)
+
+Read-only investigation helper over the `goldmane_edges` CNPG trail (ADR-0014):
+filters render to a single safe `SELECT` (namespace values validated to the k8s
+name charset) run via the dbaas primary pod — the same exec path as `k8s db`.
+
+| Command | Tier | What it does |
+| --- | --- | --- |
+| `edges --ns <ns>` | read | edges touching `<ns>` (either direction) |
+| `edges --src <ns>` / `--dst <ns>` | read | directional: `<ns>`'s egress / ingress peers |
+| `edges --peers-of <ns>` | read | distinct peer namespaces of `<ns>` (both directions) |
+| `edges --new-since <24h\|7d\|YYYY-MM-DD>` | read | edges first seen since a duration or date |
+| `edges --denied` | read | only `action='deny'` edges (blocked / lateral-movement) |
+| `edges --json` / `--limit N` | read | JSON array output / row cap (default 200) |
+
 ## Build / install
 
 Built from source to `/usr/local/bin/homelab` during devvm provisioning

cli/VERSION

@@ -1 +1 @@
-v0.8.2
+v0.9.0

cli/cmd_edges.go

@@ -0,0 +1,69 @@
+package main
+
+import "fmt"
+
+func edgesCommands() []Command {
+	return []Command{
+		{Path: []string{"edges"}, Tier: TierRead,
+			Summary: "who-talks-to-whom trail: edges [--ns|--src|--dst|--peers-of N] [--new-since 24h] [--denied] [--json] [--limit N]",
+			Run:     edgesRun},
+	}
+}
+
+// edgesRun renders the filter flags to SQL and runs it read-only against the
+// goldmane_edges CNPG DB via the dbaas primary pod (same exec path as `k8s db`).
+func edgesRun(args []string) error {
+	for _, a := range args {
+		if a == "-h" || a == "--help" {
+			fmt.Print(edgesUsage())
+			return nil
+		}
+	}
+	o, err := parseEdgesArgs(args)
+	if err != nil {
+		return fmt.Errorf("%w\n\n%s", err, edgesUsage())
+	}
+	sql, err := buildEdgesQuery(o)
+	if err != nil {
+		return err
+	}
+	// pg-cluster-rw is a Service (not exec-able); resolve the primary POD.
+	pod, err := kubectlCapture("dbaas", "get", "pod", "-l", "cnpg.io/instanceRole=primary",
+		"-o", "jsonpath={.items[0].metadata.name}")
+	if err != nil || pod == "" {
+		return fmt.Errorf("could not resolve CNPG primary pod in dbaas: %v", err)
+	}
+	exec := []string{"exec", pod, "-c", "postgres", "--", "psql", "-U", "postgres", "-d", "goldmane_edges"}
+	if o.asJSON {
+		exec = append(exec, "-tAc", sql) // raw tuple → the JSON array
+	} else {
+		exec = append(exec, "-P", "pager=off", "-c", sql) // aligned table for humans
+	}
+	return kubectlStream("dbaas", exec...)
+}
+
+func edgesUsage() string {
+	return `homelab edges — query the who-talks-to-whom trail (goldmane_edges, ADR-0014)
+
+Usage: homelab edges [filters]
+
+Filters (AND-combined; namespace values are validated to the k8s name charset):
+  --ns NAME         edges touching NAME (either direction)
+  --src NAME        edges where source namespace = NAME
+  --dst NAME        edges where destination namespace = NAME
+  --peers-of NAME   distinct peer namespaces of NAME (both directions)
+  --new-since SPEC  first seen since SPEC: a duration (24h, 7d, 30m, 90s) or a date (YYYY-MM-DD)
+  --denied          only denied (action='deny') edges — blocked / lateral-movement attempts
+  --json            output a JSON array (for agents/pipelines)
+  --limit N         cap rows (default 200)
+
+Examples:
+  homelab edges --ns immich                # everything immich talks to / is talked to by
+  homelab edges --peers-of authentik       # authentik's peer namespaces
+  homelab edges --src recruiter-responder  # that namespace's egress peers
+  homelab edges --new-since 24h            # edges first seen in the last day
+  homelab edges --denied --json            # blocked flows, machine-readable
+
+Read-only SELECT against CNPG DB goldmane_edges via the dbaas primary pod.
+`
+}

cli/edges.go

@@ -0,0 +1,164 @@
+package main
+
+import (
+	"fmt"
+	"regexp"
+	"strconv"
+	"strings"
+)
+
+// edgesOpts is the parsed filter set for `homelab edges` (the who-talks-to-whom
+// investigation helper over the goldmane_edges trail; see ADR-0014).
+type edgesOpts struct {
+	ns       string // edges touching this namespace (either direction)
+	src      string // edges where src_ns = this
+	dst      string // edges where dst_ns = this
+	peersOf  string // distinct peers of this namespace (both directions)
+	newSince string // first_seen >= duration (24h/7d/30m) or date (YYYY-MM-DD)
+	denied   bool   // action = 'deny' only
+	asJSON   bool   // wrap result as a JSON array
+	limit    int    // row cap (default 200)
+}
+
+// parseEdgesArgs parses the edges flag surface. Unknown flags error out so a
+// typo surfaces instead of silently dumping the whole table.
+func parseEdgesArgs(args []string) (edgesOpts, error) {
+	o := edgesOpts{limit: 200}
+	i := 0
+	for i < len(args) {
+		a := args[i]
+		key, inline, hasInline := a, "", false
+		if eq := strings.IndexByte(a, '='); eq >= 0 {
+			key, inline, hasInline = a[:eq], a[eq+1:], true
+		}
+		needVal := func() (string, error) {
+			if hasInline {
+				return inline, nil
+			}
+			if i+1 < len(args) {
+				i++
+				return args[i], nil
+			}
+			return "", fmt.Errorf("flag %s needs a value", key)
+		}
+		var err error
+		switch key {
+		case "--ns":
+			o.ns, err = needVal()
+		case "--src":
+			o.src, err = needVal()
+		case "--dst":
+			o.dst, err = needVal()
+		case "--peers-of":
+			o.peersOf, err = needVal()
+		case "--new-since":
+			o.newSince, err = needVal()
+		case "--denied":
+			o.denied = true
+		case "--json":
+			o.asJSON = true
+		case "--limit":
+			var v string
+			if v, err = needVal(); err == nil {
+				if o.limit, err = strconv.Atoi(v); err != nil {
+					err = fmt.Errorf("--limit must be an integer: %q", v)
+				}
+			}
+		default:
+			return o, fmt.Errorf("unknown flag: %s", a)
+		}
+		if err != nil {
+			return o, err
+		}
+		i++
+	}
+	return o, nil
+}
+
+// nsRE is the safe namespace-token charset (k8s names + "Global"). Used as the
+// injection guard — anything else is rejected rather than quoted-and-hoped.
+var nsRE = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9_.-]*$`)
+
+func validateNS(s string) error {
+	if s == "" || len(s) > 63 || !nsRE.MatchString(s) {
+		return fmt.Errorf("invalid namespace name: %q", s)
+	}
+	return nil
+}
+
+// sqlStr renders a SQL string literal (belt-and-suspenders on top of validateNS).
+func sqlStr(s string) string { return "'" + strings.ReplaceAll(s, "'", "''") + "'" }
+
+var (
+	durRE  = regexp.MustCompile(`^(\d+)([smhd])$`)
+	dateRE = regexp.MustCompile(`^\d{4}-\d{2}-\d{2}([ T]\d{2}:\d{2}(:\d{2})?)?$`)
+)
+
+// newSinceCond turns a duration (24h/7d/30m/90s) or a date (YYYY-MM-DD[ HH:MM])
+// into a first_seen predicate.
+func newSinceCond(v string) (string, error) {
+	if m := durRE.FindStringSubmatch(v); m != nil {
+		unit := map[string]string{"s": "seconds", "m": "minutes", "h": "hours", "d": "days"}[m[2]]
+		return fmt.Sprintf("first_seen >= now() - interval '%s %s'", m[1], unit), nil
+	}
+	if dateRE.MatchString(v) {
+		return "first_seen >= " + sqlStr(v), nil
+	}
+	return "", fmt.Errorf("--new-since must be a duration (e.g. 24h, 7d, 30m) or a date (YYYY-MM-DD): %q", v)
+}
+
+// buildEdgesQuery renders the SQL for the given filters against the `edge` table.
+func buildEdgesQuery(o edgesOpts) (string, error) {
+	limit := o.limit
+	if limit <= 0 {
+		limit = 200
+	}
+
+	// peers-of is a distinct-peer summary, a different shape from the row list.
+	if o.peersOf != "" {
+		if err := validateNS(o.peersOf); err != nil {
+			return "", err
+		}
+		p := sqlStr(o.peersOf)
+		return fmt.Sprintf("SELECT DISTINCT peer, action FROM ("+
+			"SELECT dst_ns AS peer, action FROM edge WHERE src_ns = %s "+
+			"UNION SELECT src_ns AS peer, action FROM edge WHERE dst_ns = %s"+
+			") t ORDER BY peer LIMIT %d", p, p, limit), nil
+	}
+
+	var conds []string
+	for _, f := range []struct{ val, tmpl string }{
+		{o.ns, "(src_ns = %[1]s OR dst_ns = %[1]s)"},
+		{o.src, "src_ns = %s"},
+		{o.dst, "dst_ns = %s"},
+	} {
+		if f.val == "" {
+			continue
+		}
+		if err := validateNS(f.val); err != nil {
+			return "", err
+		}
+		conds = append(conds, fmt.Sprintf(f.tmpl, sqlStr(f.val)))
+	}
+	if o.denied {
+		conds = append(conds, "action = 'deny'")
+	}
+	if o.newSince != "" {
+		c, err := newSinceCond(o.newSince)
+		if err != nil {
+			return "", err
+		}
+		conds = append(conds, c)
+	}
+
+	q := "SELECT src_ns, dst_ns, action, flow_count, first_seen, last_seen FROM edge"
+	if len(conds) > 0 {
+		q += " WHERE " + strings.Join(conds, " AND ")
+	}
+	q += fmt.Sprintf(" ORDER BY first_seen DESC LIMIT %d", limit)
+
+	if o.asJSON {
+		q = "SELECT coalesce(json_agg(row_to_json(t)), '[]') FROM (" + q + ") t"
+	}
+	return q, nil
+}

cli/edges_test.go

@@ -0,0 +1,163 @@
+package main
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestParseEdgesArgs(t *testing.T) {
+	cases := []struct {
+		name string
+		args []string
+		want edgesOpts
+	}{
+		{"defaults", nil, edgesOpts{limit: 200}},
+		{"ns", []string{"--ns", "immich"}, edgesOpts{ns: "immich", limit: 200}},
+		{"ns equals", []string{"--ns=immich"}, edgesOpts{ns: "immich", limit: 200}},
+		{"src dst", []string{"--src", "a", "--dst", "b"}, edgesOpts{src: "a", dst: "b", limit: 200}},
+		{"peers-of", []string{"--peers-of", "authentik"}, edgesOpts{peersOf: "authentik", limit: 200}},
+		{"denied json", []string{"--denied", "--json"}, edgesOpts{denied: true, asJSON: true, limit: 200}},
+		{"new-since", []string{"--new-since", "24h"}, edgesOpts{newSince: "24h", limit: 200}},
+		{"limit", []string{"--limit", "50"}, edgesOpts{limit: 50}},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			got, err := parseEdgesArgs(c.args)
+			if err != nil {
+				t.Fatalf("parseEdgesArgs(%v) error: %v", c.args, err)
+			}
+			if got != c.want {
+				t.Fatalf("parseEdgesArgs(%v) = %+v, want %+v", c.args, got, c.want)
+			}
+		})
+	}
+}
+
+func TestParseEdgesArgsErrors(t *testing.T) {
+	for _, args := range [][]string{
+		{"--limit", "abc"},
+		{"--bogus"},
+	} {
+		if _, err := parseEdgesArgs(args); err == nil {
+			t.Errorf("parseEdgesArgs(%v) expected error, got nil", args)
+		}
+	}
+}
+
+func TestBuildEdgesQueryDefaults(t *testing.T) {
+	q, err := buildEdgesQuery(edgesOpts{limit: 200})
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, want := range []string{"FROM edge", "ORDER BY first_seen DESC", "LIMIT 200"} {
+		if !strings.Contains(q, want) {
+			t.Errorf("query %q missing %q", q, want)
+		}
+	}
+	if strings.Contains(q, "WHERE") {
+		t.Errorf("no-filter query should have no WHERE: %q", q)
+	}
+}
+
+func TestBuildEdgesQueryFilters(t *testing.T) {
+	cases := []struct {
+		name string
+		o    edgesOpts
+		want string
+	}{
+		{"ns both directions", edgesOpts{ns: "immich", limit: 10}, "(src_ns = 'immich' OR dst_ns = 'immich')"},
+		{"src only", edgesOpts{src: "authentik", limit: 10}, "src_ns = 'authentik'"},
+		{"dst only", edgesOpts{dst: "dbaas", limit: 10}, "dst_ns = 'dbaas'"},
+		{"denied", edgesOpts{denied: true, limit: 10}, "action = 'deny'"},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			q, err := buildEdgesQuery(c.o)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if !strings.Contains(q, "WHERE") || !strings.Contains(q, c.want) {
+				t.Errorf("query %q missing WHERE/%q", q, c.want)
+			}
+		})
+	}
+}
+
+func TestBuildEdgesQueryCombinedFiltersAnded(t *testing.T) {
+	q, err := buildEdgesQuery(edgesOpts{src: "a", denied: true, limit: 5})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !strings.Contains(q, "src_ns = 'a' AND action = 'deny'") {
+		t.Errorf("combined filters not AND'd: %q", q)
+	}
+}
+
+func TestBuildEdgesQueryPeersOf(t *testing.T) {
+	q, err := buildEdgesQuery(edgesOpts{peersOf: "authentik", limit: 100})
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, want := range []string{"DISTINCT", "src_ns = 'authentik'", "dst_ns = 'authentik'", "UNION"} {
+		if !strings.Contains(q, want) {
+			t.Errorf("peers-of query %q missing %q", q, want)
+		}
+	}
+}
+
+func TestBuildEdgesQueryJSON(t *testing.T) {
+	q, err := buildEdgesQuery(edgesOpts{asJSON: true, limit: 200})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !strings.Contains(q, "json_agg") || !strings.Contains(q, "row_to_json") {
+		t.Errorf("json query missing json_agg wrapper: %q", q)
+	}
+}
+
+func TestBuildEdgesQueryRejectsInjection(t *testing.T) {
+	for _, bad := range []string{"a'; DROP TABLE edge;--", "a b", "a;b", "a\"b"} {
+		if _, err := buildEdgesQuery(edgesOpts{ns: bad, limit: 10}); err == nil {
+			t.Errorf("buildEdgesQuery(ns=%q) expected validation error, got nil", bad)
+		}
+	}
+}
+
+func TestNewSinceCond(t *testing.T) {
+	cases := []struct {
+		in   string
+		want string
+	}{
+		{"24h", "first_seen >= now() - interval '24 hours'"},
+		{"7d", "first_seen >= now() - interval '7 days'"},
+		{"30m", "first_seen >= now() - interval '30 minutes'"},
+		{"2026-06-28", "first_seen >= '2026-06-28'"},
+	}
+	for _, c := range cases {
+		got, err := newSinceCond(c.in)
+		if err != nil {
+			t.Fatalf("newSinceCond(%q) error: %v", c.in, err)
+		}
+		if got != c.want {
+			t.Errorf("newSinceCond(%q) = %q, want %q", c.in, got, c.want)
+		}
+	}
+	for _, bad := range []string{"yesterday", "1y", "'; DROP", ""} {
+		if _, err := newSinceCond(bad); err == nil {
+			t.Errorf("newSinceCond(%q) expected error, got nil", bad)
+		}
+	}
+}
+
+func TestValidateNS(t *testing.T) {
+	for _, ok := range []string{"immich", "calico-system", "kube-system", "Global", "pg-cluster-rw"} {
+		if err := validateNS(ok); err != nil {
+			t.Errorf("validateNS(%q) unexpected error: %v", ok, err)
+		}
+	}
+	for _, bad := range []string{"", "a b", "a'b", "a;b", "../x", "a$b"} {
+		if err := validateNS(bad); err == nil {
+			t.Errorf("validateNS(%q) expected error, got nil", bad)
+		}
+	}
+}

cli/homelab.go

@@ -20,6 +20,7 @@ func buildRegistry() []Command {
 	reg = append(reg, deployCommands()...)
 	reg = append(reg, netCommands()...)
 	reg = append(reg, obsCommands()...)
+	reg = append(reg, edgesCommands()...)
 	reg = append(reg, usageCommands()...)
 	reg = append(reg, haCommands()...)
 	reg = append(reg, browserCommands()...)