fix(rewrite,wire): decrypt reads whose result set spans a join

docs/runbook.md

@@ -107,6 +107,28 @@ kubectl delete ns $NS
 # apply your proxy Deployment/Service/Secret manifests here
 ```
 
+### Recovering from a read-path ciphertext leak
+
+If a read-path gap let raw `$KKP$` envelopes reach Keycloak (symptom:
+`kkp_ciphertext_passthrough_total` climbing, by `reason` — `empty-plan`,
+`no-plan`, `no-portal`, or `ambiguous-column`), Keycloak's user cache stores the
+ciphertext and serves it back on later reads, and any row Keycloak then writes
+back carries a leaked ciphertext as its value (a double-encrypted row).
+
+After rolling out the proxy version that closes the gap:
+
+1. Flush Keycloak's user cache so it re-reads through the fixed proxy — the
+   `clear-user-cache` step above (section 4, step 1). Until this runs, cached
+   ciphertext keeps being served.
+2. Confirm the incident is actually closed before standing down: watch
+   `kkp_ciphertext_passthrough_total` stay flat, and check
+   `kkp_double_encrypted_total` — any non-zero value inventories rows that were
+   written back while the leak was open and need cleanup (delete/re-save via the
+   admin REST API, as in section 4, step 2). The encrypt path now refuses to
+   re-encrypt a value that already parses as an envelope, so no *new*
+   double-encrypted rows are produced once the fix is live; the counter only
+   reflects the backlog.
+
 ## 5. Recovery: proxy down
 
 The Service IP routes to the running replicas via the regular Service

internal/rewrite/analyze.go

@@ -53,9 +53,23 @@ type Analysis struct {
 	// simple-protocol query string. Set by AnalyzeAll only; empty for
 	// Analyze, whose callers already hold the (single) statement text.
 	SQL string
-	// Table is the target table (INSERT/UPDATE/DELETE) or the first FROM table
-	// (SELECT). Empty if no single table applies.
+	// Table is the target table (INSERT/UPDATE/DELETE) or, for a SELECT, the
+	// first FROM item when that item is a plain range variable. It is empty
+	// when the first FROM item is an explicit JOIN expression, whose first
+	// child is the join rather than a relation. A comma join (FROM a, b) keeps
+	// Table non-empty, but it names only the first of several relations —
+	// prefer FromTables, which lists them all. Table is retained for the write
+	// path and the conformance corpus.
 	Table string
+	// FromTables lists every base relation the statement reads, with join
+	// trees, sub-selects and set-operation arms walked and CTE references
+	// resolved to their underlying relations; for INSERT/UPDATE/DELETE it is
+	// the target relation. The read planner resolves each result column against
+	// this set, so a result set assembled by a join still decrypts: Keycloak's
+	// group-members listing (USER_GROUP_MEMBERSHIP join USER_ENTITY) selects
+	// every USER_ENTITY column and used to hand Keycloak raw envelopes because
+	// Table is empty for a join.
+	FromTables []string
 	// WriteColumns are the columns whose values are written: INSERT column list
 	// and UPDATE SET targets, each with the parameter position of its value.
 	WriteColumns []ColumnParam
@@ -135,6 +149,7 @@ func analyzeNode(node *pg.Node) *Analysis {
 
 func analyzeInsert(ins *pg.InsertStmt) *Analysis {
 	a := &Analysis{Kind: KindInsert, Table: ins.GetRelation().GetRelname()}
+	a.FromTables = targetTables(a.Table)
 
 	var params []int
 	if sel := ins.GetSelectStmt().GetSelectStmt(); sel != nil && len(sel.GetValuesLists()) > 0 {
@@ -155,6 +170,7 @@ func analyzeInsert(ins *pg.InsertStmt) *Analysis {
 
 func analyzeUpdate(upd *pg.UpdateStmt) *Analysis {
 	a := &Analysis{Kind: KindUpdate, Table: upd.GetRelation().GetRelname()}
+	a.FromTables = targetTables(a.Table)
 	for _, t := range upd.GetTargetList() {
 		rt := t.GetResTarget()
 		a.WriteColumns = append(a.WriteColumns, ColumnParam{Column: rt.GetName(), Param: paramNumber(rt.GetVal())})
@@ -170,12 +186,125 @@ func analyzeSelect(sel *pg.SelectStmt) *Analysis {
 			a.Table = rv.GetRelname()
 		}
 	}
+	a.FromTables = selectTables(sel)
 	a.FilterColumns, a.LikeFilterColumns = whereColumns(sel.GetWhereClause())
 	return a
 }
 
+// targetTables returns a write statement's target relation as its lone read
+// source, so a plain RETURNING result set decrypts like any other read. It
+// does not walk an UPDATE ... FROM / DELETE ... USING auxiliary relation, so a
+// RETURNING column drawn from one of those is not resolved — Keycloak issues no
+// such statement against a PII table, and the read-path leak detector would
+// flag any envelope that slipped through.
+func targetTables(table string) []string {
+	if table == "" {
+		return nil
+	}
+	return []string{table}
+}
+
+// selectTables collects every base relation a SELECT reads from, in the order
+// they appear, without duplicates. CTE references are resolved to the relations
+// their query reads rather than reported as base tables.
+func selectTables(sel *pg.SelectStmt) []string {
+	c := &tableCollector{seen: make(map[string]bool), cteActive: make(map[string]bool)}
+	c.walkSelect(sel, nil)
+	return c.tables
+}
+
+// tableCollector accumulates the base relations of a SELECT. seen deduplicates
+// the result; cteActive guards against a WITH RECURSIVE cycle while a CTE
+// reference is being resolved.
+type tableCollector struct {
+	tables    []string
+	seen      map[string]bool
+	cteActive map[string]bool
+}
+
+// walkSelect descends a SELECT, its set-operation arms (UNION and friends) and
+// its FROM items. ctes is the CTE scope in effect, extended by this select's
+// own WITH clause; a nested WITH shadows outer names.
+func (c *tableCollector) walkSelect(sel *pg.SelectStmt, ctes map[string]*pg.SelectStmt) {
+	if sel == nil {
+		return
+	}
+	ctes = extendCTEScope(ctes, sel.GetWithClause())
+	c.walkSelect(sel.GetLarg(), ctes)
+	c.walkSelect(sel.GetRarg(), ctes)
+	for _, item := range sel.GetFromClause() {
+		c.walkFromItem(item, ctes)
+	}
+}
+
+// walkFromItem walks one FROM item: a range variable is a base relation (or a
+// CTE reference to resolve), a join expression has two sides, and a sub-select
+// carries its own FROM clause.
+func (c *tableCollector) walkFromItem(n *pg.Node, ctes map[string]*pg.SelectStmt) {
+	if n == nil {
+		return
+	}
+	switch {
+	case n.GetRangeVar() != nil:
+		c.addRelation(n.GetRangeVar().GetRelname(), ctes)
+	case n.GetJoinExpr() != nil:
+		je := n.GetJoinExpr()
+		c.walkFromItem(je.GetLarg(), ctes)
+		c.walkFromItem(je.GetRarg(), ctes)
+	case n.GetRangeSubselect() != nil:
+		c.walkSelect(n.GetRangeSubselect().GetSubquery().GetSelectStmt(), ctes)
+	}
+}
+
+// addRelation records a base relation, or resolves a CTE reference to the
+// relations its query reads — a CTE alias is never reported as a base table.
+func (c *tableCollector) addRelation(name string, ctes map[string]*pg.SelectStmt) {
+	if name == "" {
+		return
+	}
+	key := strings.ToUpper(name)
+	if q, ok := ctes[key]; ok {
+		if c.cteActive[key] {
+			return // WITH RECURSIVE self-reference: already resolving it.
+		}
+		c.cteActive[key] = true
+		c.walkSelect(q, ctes)
+		delete(c.cteActive, key)
+		return
+	}
+	if c.seen[key] {
+		return
+	}
+	c.seen[key] = true
+	c.tables = append(c.tables, name)
+}
+
+// extendCTEScope returns the CTE scope augmented with a WITH clause's CTEs,
+// mapping each upper-cased CTE name to its query. The parent scope is left
+// untouched so sibling selects do not see this level's names.
+func extendCTEScope(parent map[string]*pg.SelectStmt, wc *pg.WithClause) map[string]*pg.SelectStmt {
+	if wc == nil || len(wc.GetCtes()) == 0 {
+		return parent
+	}
+	scope := make(map[string]*pg.SelectStmt, len(parent)+len(wc.GetCtes()))
+	for k, v := range parent {
+		scope[k] = v
+	}
+	for _, node := range wc.GetCtes() {
+		cte := node.GetCommonTableExpr()
+		if cte == nil {
+			continue
+		}
+		if q := cte.GetCtequery().GetSelectStmt(); q != nil {
+			scope[strings.ToUpper(cte.GetCtename())] = q
+		}
+	}
+	return scope
+}
+
 func analyzeDelete(del *pg.DeleteStmt) *Analysis {
 	a := &Analysis{Kind: KindDelete, Table: del.GetRelation().GetRelname()}
+	a.FromTables = targetTables(a.Table)
 	a.FilterColumns, a.LikeFilterColumns = whereColumns(del.GetWhereClause())
 	return a
 }

internal/rewrite/fromtables_test.go

@@ -0,0 +1,162 @@
+package rewrite
+
+import "testing"
+
+// FromTables is what lets the read planner decrypt a result set the backend
+// assembled from more than one relation. A join's first FROM item is the join
+// expression, not a range variable, so `Table` is empty for it — the
+// group-members listing leaked every USER_ENTITY column on stage for exactly
+// that reason.
+
+func eqTables(got, want []string) bool {
+	if len(got) != len(want) {
+		return false
+	}
+	for i := range got {
+		if got[i] != want[i] {
+			return false
+		}
+	}
+	return true
+}
+
+func TestAnalyzeFromTablesJoin(t *testing.T) {
+	t.Parallel()
+
+	a, err := Analyze(`select u.EMAIL from USER_GROUP_MEMBERSHIP m join USER_ENTITY u on u.ID=m.USER_ID where m.GROUP_ID=$1`)
+	if err != nil {
+		t.Fatalf("Analyze: %v", err)
+	}
+	if a.Table != "" {
+		t.Errorf("table: got %q, want empty for a join", a.Table)
+	}
+	if want := []string{"user_group_membership", "user_entity"}; !eqTables(a.FromTables, want) {
+		t.Errorf("from tables: got %v, want %v", a.FromTables, want)
+	}
+}
+
+func TestAnalyzeFromTablesNestedJoin(t *testing.T) {
+	t.Parallel()
+
+	a, err := Analyze(`select u.EMAIL from FEDERATED_IDENTITY f join USER_ENTITY u on u.ID=f.USER_ID ` +
+		`left join USER_ATTRIBUTE at on at.USER_ID=u.ID where f.FEDERATED_USER_ID=$1`)
+	if err != nil {
+		t.Fatalf("Analyze: %v", err)
+	}
+	want := []string{"federated_identity", "user_entity", "user_attribute"}
+	if !eqTables(a.FromTables, want) {
+		t.Errorf("from tables: got %v, want %v", a.FromTables, want)
+	}
+}
+
+func TestAnalyzeFromTablesCommaJoin(t *testing.T) {
+	t.Parallel()
+
+	a, err := Analyze(`select u.EMAIL from USER_ENTITY u, USER_ATTRIBUTE at where at.USER_ID=u.ID`)
+	if err != nil {
+		t.Fatalf("Analyze: %v", err)
+	}
+	if want := []string{"user_entity", "user_attribute"}; !eqTables(a.FromTables, want) {
+		t.Errorf("from tables: got %v, want %v", a.FromTables, want)
+	}
+}
+
+func TestAnalyzeFromTablesSubselect(t *testing.T) {
+	t.Parallel()
+
+	a, err := Analyze(`select x.EMAIL from (select ue.EMAIL from USER_ENTITY ue) x`)
+	if err != nil {
+		t.Fatalf("Analyze: %v", err)
+	}
+	if want := []string{"user_entity"}; !eqTables(a.FromTables, want) {
+		t.Errorf("from tables: got %v, want %v", a.FromTables, want)
+	}
+}
+
+func TestAnalyzeFromTablesSetOperation(t *testing.T) {
+	t.Parallel()
+
+	a, err := Analyze(`select EMAIL from USER_ENTITY union all select FEDERATED_USERNAME from FEDERATED_IDENTITY`)
+	if err != nil {
+		t.Fatalf("Analyze: %v", err)
+	}
+	if want := []string{"user_entity", "federated_identity"}; !eqTables(a.FromTables, want) {
+		t.Errorf("from tables: got %v, want %v", a.FromTables, want)
+	}
+}
+
+func TestAnalyzeFromTablesDeduplicatesSelfJoin(t *testing.T) {
+	t.Parallel()
+
+	a, err := Analyze(`select a.EMAIL from USER_ENTITY a join USER_ENTITY b on b.ID=a.FEDERATION_LINK`)
+	if err != nil {
+		t.Fatalf("Analyze: %v", err)
+	}
+	if want := []string{"user_entity"}; !eqTables(a.FromTables, want) {
+		t.Errorf("from tables: got %v, want %v", a.FromTables, want)
+	}
+}
+
+// TestAnalyzeFromTablesCTE: a CTE alias is not a base relation. `FROM m` must
+// resolve to the relations the CTE's query reads (USER_ENTITY), never report
+// the alias `m` — which would leave the read planner unable to decrypt and
+// (worse) misrepresent what the statement actually touches.
+func TestAnalyzeFromTablesCTE(t *testing.T) {
+	t.Parallel()
+
+	a, err := Analyze(`with m as (select ue.EMAIL from USER_ENTITY ue) select x.EMAIL from m x`)
+	if err != nil {
+		t.Fatalf("Analyze: %v", err)
+	}
+	if want := []string{"user_entity"}; !eqTables(a.FromTables, want) {
+		t.Errorf("from tables: got %v, want %v (CTE alias must resolve to its base relation)", a.FromTables, want)
+	}
+}
+
+// TestAnalyzeFromTablesCTEJoinedWithBase: a CTE reference joined with a real
+// relation resolves the CTE to its underlying table and keeps the base table,
+// in order, without the alias.
+func TestAnalyzeFromTablesCTEJoinedWithBase(t *testing.T) {
+	t.Parallel()
+
+	a, err := Analyze(`with m as (select fue.USER_ID from FED_USER_ENTITY fue) ` +
+		`select u.EMAIL from USER_ENTITY u join m on m.USER_ID=u.ID`)
+	if err != nil {
+		t.Fatalf("Analyze: %v", err)
+	}
+	if want := []string{"user_entity", "fed_user_entity"}; !eqTables(a.FromTables, want) {
+		t.Errorf("from tables: got %v, want %v", a.FromTables, want)
+	}
+}
+
+// TestAnalyzeFromTablesRecursiveCTETerminates: a WITH RECURSIVE self-reference
+// must not send the walker into an infinite loop.
+func TestAnalyzeFromTablesRecursiveCTETerminates(t *testing.T) {
+	t.Parallel()
+
+	a, err := Analyze(`with recursive m as (select ue.ID from USER_ENTITY ue union all select ue2.ID from USER_ENTITY ue2 join m on m.ID=ue2.ID) select ID from m`)
+	if err != nil {
+		t.Fatalf("Analyze: %v", err)
+	}
+	if want := []string{"user_entity"}; !eqTables(a.FromTables, want) {
+		t.Errorf("from tables: got %v, want %v", a.FromTables, want)
+	}
+}
+
+func TestAnalyzeFromTablesOnWrites(t *testing.T) {
+	t.Parallel()
+
+	for _, sql := range []string{
+		`UPDATE user_entity SET email = $1 WHERE id = $2`,
+		`INSERT INTO user_entity (id, email) VALUES ($1, $2)`,
+		`DELETE FROM user_entity WHERE id = $1`,
+	} {
+		a, err := Analyze(sql)
+		if err != nil {
+			t.Fatalf("Analyze(%s): %v", sql, err)
+		}
+		if want := []string{"user_entity"}; !eqTables(a.FromTables, want) {
+			t.Errorf("%s: from tables got %v, want %v", sql, a.FromTables, want)
+		}
+	}
+}