From 6c91bdc93524f6940fca8e841d8aa4c565e4e45f Mon Sep 17 00:00:00 2001 From: Andrei Kvapil Date: Wed, 26 Aug 2026 15:11:21 +0200 Subject: [PATCH 1/4] fix(rewrite,wire): decrypt reads whose result set spans a join MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The read planner keyed the decrypt plan on `Analysis.Table`, the single relation behind a SELECT. A join has none: the first FROM item is the join expression, not a range variable, so `Table` stayed empty, `PlanRead` returned an empty plan and every PII column in the result reached Keycloak as a raw $KKP$ envelope. Keycloak takes that path for getGroupMembersStream — the admin REST group-members listing — which selects the full USER_ENTITY row through `USER_GROUP_MEMBERSHIP join USER_ENTITY`. On stage the proxy was flagging it itself: kkp_ciphertext_passthrough_total{reason="empty-plan"} and kkp_unrecognized_pii_sql_total climbing together, one "ciphertext passed through undecrypted" warning per member row. Ciphertext that reaches Keycloak this way outlives the request — it is what the user cache stores and later reads serve. Collect every base relation a statement reads into `FromTables`, walking join trees, sub-selects and set-operation arms, and resolve each result column against that set. A column two joined relations both configure is left alone: the associated data binds a ciphertext to one table, so guessing fails the authentication tag and errors the query — the passthrough detector flags those instead. `Table` keeps its meaning, so the write path is untouched: parameter encryption and the fail-loud rules still refuse to act on a statement whose target relation is ambiguous. Assisted-By: Claude Signed-off-by: Andrei Kvapil --- internal/rewrite/analyze.go | 71 ++++++++++- internal/rewrite/fromtables_test.go | 116 ++++++++++++++++++ internal/rewrite/read.go | 38 ++++-- internal/rewrite/read_test.go | 62 +++++++++- internal/wire/joined_read_test.go | 181 ++++++++++++++++++++++++++++ internal/wire/session.go | 8 +- 6 files changed, 457 insertions(+), 19 deletions(-) create mode 100644 internal/rewrite/fromtables_test.go create mode 100644 internal/wire/joined_read_test.go diff --git a/internal/rewrite/analyze.go b/internal/rewrite/analyze.go index 21e2be9..4bb54c8 100644 --- a/internal/rewrite/analyze.go +++ b/internal/rewrite/analyze.go @@ -54,8 +54,19 @@ type Analysis struct { // 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. + // (SELECT). Empty if no single table applies — notably when the FROM + // clause is a join, whose first item is the join expression rather than a + // range variable. Table string + // FromTables lists every base relation the statement reads, with join + // trees, sub-selects and set-operation arms walked; 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 +146,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 +167,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 +183,68 @@ func analyzeSelect(sel *pg.SelectStmt) *Analysis { a.Table = rv.GetRelname() } } + a.FromTables = selectTables(sel) a.FilterColumns, a.LikeFilterColumns = whereColumns(sel.GetWhereClause()) return a } +// targetTables wraps a write statement's target relation as its (single) read +// source, so a RETURNING result set decrypts like any other read. +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. +func selectTables(sel *pg.SelectStmt) []string { + var tables []string + seen := make(map[string]bool) + collectSelectTables(sel, &tables, seen) + return tables +} + +// collectSelectTables walks a SELECT and its set-operation arms (UNION and +// friends), descending into every FROM item. +func collectSelectTables(sel *pg.SelectStmt, tables *[]string, seen map[string]bool) { + if sel == nil { + return + } + collectSelectTables(sel.GetLarg(), tables, seen) + collectSelectTables(sel.GetRarg(), tables, seen) + for _, item := range sel.GetFromClause() { + collectFromItem(item, tables, seen) + } +} + +// collectFromItem walks one FROM item: a range variable is a base relation, a +// join expression has two sides, and a sub-select carries its own FROM clause. +func collectFromItem(n *pg.Node, tables *[]string, seen map[string]bool) { + if n == nil { + return + } + switch { + case n.GetRangeVar() != nil: + name := n.GetRangeVar().GetRelname() + if name == "" || seen[strings.ToUpper(name)] { + return + } + seen[strings.ToUpper(name)] = true + *tables = append(*tables, name) + case n.GetJoinExpr() != nil: + je := n.GetJoinExpr() + collectFromItem(je.GetLarg(), tables, seen) + collectFromItem(je.GetRarg(), tables, seen) + case n.GetRangeSubselect() != nil: + collectSelectTables(n.GetRangeSubselect().GetSubquery().GetSelectStmt(), tables, seen) + } +} + 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 } diff --git a/internal/rewrite/fromtables_test.go b/internal/rewrite/fromtables_test.go new file mode 100644 index 0000000..aeff0e5 --- /dev/null +++ b/internal/rewrite/fromtables_test.go @@ -0,0 +1,116 @@ +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) + } +} + +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) + } + } +} diff --git a/internal/rewrite/read.go b/internal/rewrite/read.go index e809dba..886000f 100644 --- a/internal/rewrite/read.go +++ b/internal/rewrite/read.go @@ -22,25 +22,45 @@ type ReadPlan struct { func (rp *ReadPlan) IsEmpty() bool { return len(rp.Fields) == 0 } // PlanRead maps a query's result columns (learned from RowDescription, in -// order) to the fields that may need decryption, given the queried table. A -// field is included when it is a configured PII column or an attribute value -// column; non-encrypted rows pass through harmlessly because decryption is -// marker-driven. An unknown (empty) table yields an empty plan. -func (p *Planner) PlanRead(table string, columns []string) *ReadPlan { - table = strings.ToUpper(table) +// order) to the fields that may need decryption, given the relations the +// statement reads from. A field is included when exactly one of those +// relations configures it as a PII column or an attribute value column; +// non-encrypted rows pass through harmlessly because decryption is +// marker-driven. A statement with no known relation yields an empty plan. +func (p *Planner) PlanRead(tables []string, columns []string) *ReadPlan { plan := &ReadPlan{} - if table == "" { + if len(tables) == 0 { return plan } for i, col := range columns { c := strings.ToUpper(col) - if p.isReadablePIIColumn(table, c) { - plan.Fields = append(plan.Fields, ReadField{Index: i, Table: table, Column: c}) + if owner, ok := p.columnOwner(tables, c); ok { + plan.Fields = append(plan.Fields, ReadField{Index: i, Table: owner, Column: c}) } } return plan } +// columnOwner reports which of the statement's relations configures the result +// column as PII. A column claimed by two joined relations is ambiguous: the +// associated data binds a ciphertext to one specific table, so decrypting under +// the wrong one fails the authentication tag and errors the query. Leave those +// alone — the read-path leak detector flags them instead. +func (p *Planner) columnOwner(tables []string, column string) (string, bool) { + owner := "" + for _, t := range tables { + t = strings.ToUpper(t) + if t == "" || !p.isReadablePIIColumn(t, column) { + continue + } + if owner != "" && owner != t { + return "", false + } + owner = t + } + return owner, owner != "" +} + func (p *Planner) isReadablePIIColumn(table, column string) bool { if _, ok := p.fields.ColumnRule(table, column); ok { return true diff --git a/internal/rewrite/read_test.go b/internal/rewrite/read_test.go index 8ea059f..b851488 100644 --- a/internal/rewrite/read_test.go +++ b/internal/rewrite/read_test.go @@ -4,6 +4,7 @@ import ( "testing" "github.com/cozystack/keycloak-kms-proxy/internal/config" + "github.com/cozystack/keycloak-kms-proxy/internal/crypto" ) func readFieldIndices(plan *ReadPlan) map[string]int { @@ -18,7 +19,7 @@ func TestPlanReadUserEntity(t *testing.T) { t.Parallel() p := NewPlanner(config.Default()) - plan := p.PlanRead("user_entity", []string{"id", "username", "email", "first_name"}) + plan := p.PlanRead([]string{"user_entity"}, []string{"id", "username", "email", "first_name"}) got := readFieldIndices(plan) want := map[string]int{"USERNAME": 1, "EMAIL": 2, "FIRST_NAME": 3} @@ -39,7 +40,7 @@ func TestPlanReadUppercaseColumns(t *testing.T) { t.Parallel() p := NewPlanner(config.Default()) - plan := p.PlanRead("USER_ENTITY", []string{"EMAIL"}) + plan := p.PlanRead([]string{"USER_ENTITY"}, []string{"EMAIL"}) if len(plan.Fields) != 1 || plan.Fields[0].Column != "EMAIL" || plan.Fields[0].Index != 0 { t.Fatalf("uppercase columns not matched: %+v", plan.Fields) } @@ -49,7 +50,7 @@ func TestPlanReadAttributeValues(t *testing.T) { t.Parallel() p := NewPlanner(config.Default()) - plan := p.PlanRead("user_attribute", []string{"name", "value", "long_value", "long_value_hash"}) + plan := p.PlanRead([]string{"user_attribute"}, []string{"name", "value", "long_value", "long_value_hash"}) got := readFieldIndices(plan) // value and long_value may be encrypted; name and the hash never are. @@ -65,7 +66,7 @@ func TestPlanReadNonPIITableEmpty(t *testing.T) { t.Parallel() p := NewPlanner(config.Default()) - if plan := p.PlanRead("realm", []string{"id", "name"}); !plan.IsEmpty() { + if plan := p.PlanRead([]string{"realm"}, []string{"id", "name"}); !plan.IsEmpty() { t.Fatalf("non-PII table produced read fields: %+v", plan.Fields) } } @@ -74,7 +75,58 @@ func TestPlanReadUnknownTableEmpty(t *testing.T) { t.Parallel() p := NewPlanner(config.Default()) - if plan := p.PlanRead("", []string{"email"}); !plan.IsEmpty() { + if plan := p.PlanRead(nil, []string{"email"}); !plan.IsEmpty() { t.Fatalf("unknown table produced read fields: %+v", plan.Fields) } } + +// TestPlanReadJoinedTables is the group-members read: the join brings in a +// relation with no PII of its own, and every PII column must still resolve to +// USER_ENTITY so the AAD matches what the write path used. +func TestPlanReadJoinedTables(t *testing.T) { + t.Parallel() + + p := NewPlanner(config.Default()) + plan := p.PlanRead( + []string{"user_group_membership", "user_entity"}, + []string{"id", "email", "first_name", "last_name", "username"}, + ) + + got := readFieldIndices(plan) + want := map[string]int{"EMAIL": 1, "FIRST_NAME": 2, "LAST_NAME": 3, "USERNAME": 4} + if len(got) != len(want) { + t.Fatalf("fields: got %+v, want %v", plan.Fields, want) + } + for col, idx := range want { + if got[col] != idx { + t.Errorf("%s index: got %d, want %d", col, got[col], idx) + } + } + for _, f := range plan.Fields { + if f.Table != "USER_ENTITY" { + t.Errorf("%s resolved to table %q, want USER_ENTITY", f.Column, f.Table) + } + } +} + +// TestPlanReadAmbiguousColumnSkipped: when two joined relations both configure +// the same column, the proxy cannot tell which AAD the value was sealed with. +// Guessing would fail the authentication tag and error the whole query, so the +// field is left for the passthrough detector to flag. +func TestPlanReadAmbiguousColumnSkipped(t *testing.T) { + t.Parallel() + + fs := config.New() + rule := config.Rule{Scheme: crypto.SchemeNonDeterministic} + fs.SetColumn("USER_ENTITY", "EMAIL", rule) + fs.SetColumn("FED_USER_ENTITY", "EMAIL", rule) + + p := NewPlanner(fs) + if plan := p.PlanRead([]string{"user_entity", "fed_user_entity"}, []string{"email"}); !plan.IsEmpty() { + t.Fatalf("ambiguous column planned anyway: %+v", plan.Fields) + } + // Unambiguous on its own. + if plan := p.PlanRead([]string{"user_entity"}, []string{"email"}); len(plan.Fields) != 1 { + t.Fatalf("single-table read lost its field: %+v", plan.Fields) + } +} diff --git a/internal/wire/joined_read_test.go b/internal/wire/joined_read_test.go new file mode 100644 index 0000000..646b279 --- /dev/null +++ b/internal/wire/joined_read_test.go @@ -0,0 +1,181 @@ +package wire + +import ( + "testing" + + "github.com/jackc/pgproto3/v2" + "github.com/prometheus/client_golang/prometheus/testutil" + + "github.com/cozystack/keycloak-kms-proxy/internal/crypto" + "github.com/cozystack/keycloak-kms-proxy/internal/observe" + "github.com/cozystack/keycloak-kms-proxy/internal/rewrite" +) + +// Regression tests for the joined read path. A SELECT whose FROM clause is a +// join has no single range variable, so the analyser left `Table` empty, the +// read planner returned an empty plan, and every USER_ENTITY column in the +// result — email, names, username — reached Keycloak as a raw $KKP$ envelope. +// Observed live on stage: kkp_ciphertext_passthrough_total{reason="empty-plan"} +// climbing in lockstep with the "unrecognised PII-touching SELECT" warning for +// the group-members listing below. + +// groupMembersSelect is the Hibernate SQL Keycloak issues for +// getGroupMembersStream (admin REST GET /groups/{id}/members), captured +// verbatim from the stage proxy's warning log. +const groupMembersSelect = "select u1_0.ID,u1_0.CREATED_TIMESTAMP,u1_0.EMAIL,u1_0.EMAIL_CONSTRAINT," + + "u1_0.EMAIL_VERIFIED,u1_0.ENABLED,u1_0.FEDERATION_LINK,u1_0.FIRST_NAME,u1_0.LAST_MODIFIED_TIMESTAMP," + + "u1_0.LAST_NAME,u1_0.NOT_BEFORE,u1_0.REALM_ID,u1_0.SERVICE_ACCOUNT_CLIENT_LINK,u1_0.USERNAME " + + "from USER_GROUP_MEMBERSHIP ugme1_0 join USER_ENTITY u1_0 on u1_0.ID=ugme1_0.USER_ID " + + "where ugme1_0.GROUP_ID=$1 order by u1_0.USERNAME offset $2 rows fetch first $3 rows only" + +// groupMembersColumns is the RowDescription the backend answers with, in the +// select list's order. +var groupMembersColumns = []string{ + "id", "created_timestamp", "email", "email_constraint", "email_verified", "enabled", + "federation_link", "first_name", "last_modified_timestamp", "last_name", "not_before", + "realm_id", "service_account_client_link", "username", +} + +func rowDescriptionFor(columns []string) *pgproto3.RowDescription { + fields := make([]pgproto3.FieldDescription, len(columns)) + for i, c := range columns { + fields[i] = pgproto3.FieldDescription{Name: []byte(c)} + } + return &pgproto3.RowDescription{Fields: fields} +} + +// storedValue produces the backend-stored ciphertext for a column value. +func storedValue(t *testing.T, s *Session, scheme crypto.Scheme, table, column, plaintext string) []byte { + t.Helper() + stored, err := s.cipher.Encrypt(scheme, []byte(plaintext), rewrite.AAD(table, column)) + if err != nil { + t.Fatalf("Encrypt(%s.%s): %v", table, column, err) + } + return []byte(stored) +} + +// groupMembersRow builds one DataRow of the group-members result with the +// PII columns stored as the backend holds them. +func groupMembersRow(t *testing.T, s *Session) *pgproto3.DataRow { + t.Helper() + const ( + deterministic = crypto.SchemeDeterministic + nonDeterministic = crypto.SchemeNonDeterministic + ) + return &pgproto3.DataRow{Values: [][]byte{ + []byte("00000000-0000-0000-0000-000000000001"), // id + []byte("1700000000000"), // created_timestamp + storedValue(t, s, deterministic, "USER_ENTITY", "EMAIL", "ivy@example.test"), // email + []byte("ivy@example.test"), // email_constraint + []byte("t"), // email_verified + []byte("t"), // enabled + nil, // federation_link + storedValue(t, s, nonDeterministic, "USER_ENTITY", "FIRST_NAME", "Ivy"), // first_name + []byte("1700000000000"), // last_modified_timestamp + storedValue(t, s, nonDeterministic, "USER_ENTITY", "LAST_NAME", "Nolan"), // last_name + []byte("0"), // not_before + []byte("00000000-0000-0000-0000-0000000000aa"), // realm_id + nil, // service_account_client_link + storedValue(t, s, deterministic, "USER_ENTITY", "USERNAME", "ivy"), // username + }} +} + +func assertGroupMemberDecrypted(t *testing.T, dr *pgproto3.DataRow) { + t.Helper() + for _, want := range []struct { + index int + name string + value string + }{ + {2, "email", "ivy@example.test"}, + {7, "first_name", "Ivy"}, + {9, "last_name", "Nolan"}, + {13, "username", "ivy"}, + } { + if got := string(dr.Values[want.index]); got != want.value { + t.Errorf("%s not decrypted on joined read: got %q, want %q", want.name, got, want.value) + } + } +} + +// TestJoinedSelectDecrypts is the group-members regression: the result set is +// assembled by a join, but every column in it belongs to USER_ENTITY and must +// decrypt. +func TestJoinedSelectDecrypts(t *testing.T) { + t.Parallel() + + s := newEncryptingSession(t) + if err := s.OnParse(&pgproto3.Parse{Name: "S_223", Query: groupMembersSelect}); err != nil { + t.Fatalf("OnParse: %v", err) + } + s.OnDescribe(&pgproto3.Describe{ObjectType: 'S', Name: "S_223"}) + s.OnBind(&pgproto3.Bind{DestinationPortal: "C_1", PreparedStatement: "S_223"}) + s.OnExecute(&pgproto3.Execute{Portal: "C_1"}) + s.OnSync() + + s.OnRowDescription(rowDescriptionFor(groupMembersColumns)) + dr := groupMembersRow(t, s) + if err := s.DecryptDataRow(dr); err != nil { + t.Fatalf("DecryptDataRow: %v", err) + } + assertGroupMemberDecrypted(t, dr) + s.OnCommandComplete() + s.OnReadyForQuery() +} + +// TestJoinedSelectReuseDecrypts covers the warm pgjdbc path for the same +// statement: once described, later executions carry no Describe at all and +// the plan must still be derivable from the cached columns. +func TestJoinedSelectReuseDecrypts(t *testing.T) { + t.Parallel() + + s := newEncryptingSession(t) + if err := s.OnParse(&pgproto3.Parse{Name: "S_223", Query: groupMembersSelect}); err != nil { + t.Fatalf("OnParse: %v", err) + } + s.OnDescribe(&pgproto3.Describe{ObjectType: 'S', Name: "S_223"}) + s.OnBind(&pgproto3.Bind{DestinationPortal: "C_1", PreparedStatement: "S_223"}) + s.OnExecute(&pgproto3.Execute{Portal: "C_1"}) + s.OnSync() + s.OnRowDescription(rowDescriptionFor(groupMembersColumns)) + s.OnCommandComplete() + s.OnReadyForQuery() + + s.OnBind(&pgproto3.Bind{DestinationPortal: "C_2", PreparedStatement: "S_223"}) + s.OnExecute(&pgproto3.Execute{Portal: "C_2"}) + s.OnSync() + dr := groupMembersRow(t, s) + if err := s.DecryptDataRow(dr); err != nil { + t.Fatalf("DecryptDataRow: %v", err) + } + assertGroupMemberDecrypted(t, dr) +} + +// TestJoinedSelectLeavesNoCiphertextPassthrough asserts the incident counter +// stays flat: it is the signal production alerts on, and it was the only +// evidence the group-members read was leaking. +func TestJoinedSelectLeavesNoCiphertextPassthrough(t *testing.T) { + observe.CiphertextPassthrough.Reset() + observe.UnrecognizedPIISQL.Add(0) + before := testutil.ToFloat64(observe.UnrecognizedPIISQL) + + s := newEncryptingSession(t) + if err := s.OnParse(&pgproto3.Parse{Name: "S_223", Query: groupMembersSelect}); err != nil { + t.Fatalf("OnParse: %v", err) + } + s.OnDescribe(&pgproto3.Describe{ObjectType: 'S', Name: "S_223"}) + s.OnBind(&pgproto3.Bind{DestinationPortal: "C_1", PreparedStatement: "S_223"}) + s.OnExecute(&pgproto3.Execute{Portal: "C_1"}) + s.OnSync() + s.OnRowDescription(rowDescriptionFor(groupMembersColumns)) + if err := s.DecryptDataRow(groupMembersRow(t, s)); err != nil { + t.Fatalf("DecryptDataRow: %v", err) + } + + if leaked := testutil.ToFloat64(observe.CiphertextPassthrough.WithLabelValues("empty-plan")); leaked != 0 { + t.Errorf("ciphertext passthrough counted %v times on the joined read", leaked) + } + if got := testutil.ToFloat64(observe.UnrecognizedPIISQL) - before; got != 0 { + t.Errorf("joined PII SELECT still counted as unrecognised %v times", got) + } +} diff --git a/internal/wire/session.go b/internal/wire/session.go index 95df02f..317ed6d 100644 --- a/internal/wire/session.go +++ b/internal/wire/session.go @@ -299,15 +299,15 @@ func (s *Session) OnRowDescription(rd *pgproto3.RowDescription) { func (s *Session) learnColumns(ps *PreparedStatement, columns []string) { ps.Columns = columns plan := s.planRead(ps, columns) - table := "" + var tables []string if ps.Analysis != nil { - table = ps.Analysis.Table + tables = ps.Analysis.FromTables } fields := 0 if plan != nil { fields = len(plan.Fields) } - debugf("kkp: rowdesc stmt=%q table=%q cols=%v readPlanFields=%d", ps.Name, table, columns, fields) + debugf("kkp: rowdesc stmt=%q tables=%v cols=%v readPlanFields=%d", ps.Name, tables, columns, fields) if (plan == nil || plan.IsEmpty()) && piiTouchedButNotPlanned(ps.SQL, columns) { observe.UnrecognizedPIISQL.Inc() @@ -322,7 +322,7 @@ func (s *Session) planRead(ps *PreparedStatement, columns []string) *rewrite.Rea if ps.Analysis == nil { return nil } - return s.planner.PlanRead(ps.Analysis.Table, columns) + return s.planner.PlanRead(ps.Analysis.FromTables, columns) } // ensureReadPlan resolves a portal's decrypt plan, deriving it from the From 711e1f9729be20dae6d01d95eeb9af6a2927f439 Mon Sep 17 00:00:00 2001 From: Andrei Kvapil Date: Wed, 26 Aug 2026 15:11:33 +0200 Subject: [PATCH 2/4] test(rewrite): gate the runtime corpus on the relations a SELECT reads The PII conformance gate keyed off the table the capture recorded, so a statement whose table the analyser had already failed to extract was skipped by the very test meant to catch that failure. The group-members join went in with an empty table column, was never checked, and leaked in production. Assert instead that every PII relation named after a FROM or JOIN keyword shows up in the statement's `FromTables`, and capture the two reads the stage proxy warned about: the group-members join and the FEDERATED_IDENTITY lookup. Assisted-By: Claude Signed-off-by: Andrei Kvapil --- internal/rewrite/runtime_corpus_test.go | 62 ++++++++++++++++++++++++ testdata/keycloak/26.0.0/runtime-sql.txt | 2 + 2 files changed, 64 insertions(+) diff --git a/internal/rewrite/runtime_corpus_test.go b/internal/rewrite/runtime_corpus_test.go index 83b2510..0a62e71 100644 --- a/internal/rewrite/runtime_corpus_test.go +++ b/internal/rewrite/runtime_corpus_test.go @@ -158,3 +158,65 @@ func TestRuntimeCorpusPIICoverage(t *testing.T) { } t.Logf("runtime corpus covers %d PII-touching statements", pii) } + +// TestRuntimeCorpusFromTableCoverage — every PII relation a SELECT reads from +// must show up in `FromTables`, whatever shape the FROM clause takes. +// +// TestRuntimeCorpusPIICoverage above keys off the *captured* table column, so a +// statement whose table the proxy already failed to extract is skipped by the +// very gate meant to catch it. That blind spot is how the group-members read +// (USER_GROUP_MEMBERSHIP join USER_ENTITY) shipped: recorded with an empty +// table, skipped by the conformance test, and leaking every USER_ENTITY column +// as a raw envelope in production. This test reads the SQL instead. +func TestRuntimeCorpusFromTableCoverage(t *testing.T) { + t.Parallel() + cases := loadRuntimeCorpus(t) + joined := 0 + for _, c := range cases { + a, err := Analyze(c.SQL) + if err != nil || a.Kind != KindSelect { + continue // parse failures are reported by their own test. + } + if len(a.FromTables) > 1 { + joined++ + } + for tbl := range piiTables { + if !readsFrom(c.SQL, tbl) || containsFold(a.FromTables, tbl) { + continue + } + t.Errorf("SELECT reads %s but it is missing from FromTables=%v (silent-passthrough hazard)\n sql=%s", + tbl, a.FromTables, truncateSQL(c.SQL, 200)) + } + } + if joined == 0 { + t.Fatal("no multi-relation SELECT in the corpus — the joined read path is uncovered") + } + t.Logf("runtime corpus covers %d multi-relation SELECTs", joined) +} + +// readsFrom reports whether the SQL names the relation right after a FROM or +// JOIN keyword — the positions that put it in the result set's scope. A +// mention inside a WHERE sub-query does not count: those rows never reach the +// client, so they need no decrypt plan. +func readsFrom(sql, table string) bool { + fields := strings.Fields(strings.ToLower(sql)) + for i := 1; i < len(fields); i++ { + prev := fields[i-1] + if prev != "from" && prev != "join" { + continue + } + if strings.TrimLeft(fields[i], "(") == strings.ToLower(table) { + return true + } + } + return false +} + +func containsFold(list []string, want string) bool { + for _, s := range list { + if strings.EqualFold(s, want) { + return true + } + } + return false +} diff --git a/testdata/keycloak/26.0.0/runtime-sql.txt b/testdata/keycloak/26.0.0/runtime-sql.txt index 5ffa072..74a7249 100644 --- a/testdata/keycloak/26.0.0/runtime-sql.txt +++ b/testdata/keycloak/26.0.0/runtime-sql.txt @@ -5,6 +5,7 @@ OTHER COMMIT SELECT select cr1_0.COMPOSITE,cr1_1.ID,cr1_1.CLIENT,cr1_1.CLIENT_REALM_CONSTRAINT,cr1_1.CLIENT_ROLE,cr1_1.DESCRIPTION,cr1_1.NAME,cr1_1.REALM_ID from COMPOSITE_ROLE cr1_0 join KEYCLOAK_ROLE cr1_1 on cr1_1.ID=cr1_0.CHILD_ROLE where cr1_0.COMPOSITE = any ($1) SELECT select cr1_0.COMPOSITE,cr1_1.ID,cr1_1.CLIENT,cr1_1.CLIENT_REALM_CONSTRAINT,cr1_1.CLIENT_ROLE,cr1_1.DESCRIPTION,cr1_1.NAME,cr1_1.REALM_ID from COMPOSITE_ROLE cr1_0 join KEYCLOAK_ROLE cr1_1 on cr1_1.ID=cr1_0.CHILD_ROLE where cr1_0.COMPOSITE=$1 SELECT select re1_0.ID,re1_0.ACCESS_CODE_LIFESPAN,re1_0.LOGIN_LIFESPAN,re1_0.USER_ACTION_LIFESPAN,re1_0.ACCESS_TOKEN_LIFESPAN,re1_0.ACCESS_TOKEN_LIFE_IMPLICIT,re1_0.ACCOUNT_THEME,re1_0.ADMIN_EVENTS_DETAILS_ENABLED,re1_0.ADMIN_EVENTS_ENABLED,re1_0.ADMIN_THEME,re1_0.ALLOW_USER_MANAGED_ACCESS,re1_0.BROWSER_FLOW,re1_0.CLIENT_AUTH_FLOW,re1_0.DEFAULT_LOCALE,re1_0.DEFAULT_ROLE,re1_0.DIRECT_GRANT_FLOW,re1_0.DOCKER_AUTH_FLOW,re1_0.DUPLICATE_EMAILS_ALLOWED,re1_0.EDIT_USERNAME_ALLOWED,re1_0.EMAIL_THEME,re1_0.ENABLED,re1_0.EVENTS_ENABLED,re1_0.EVENTS_EXPIRATION,re1_0.INTERNATIONALIZATION_ENABLED,re1_0.LOGIN_THEME,re1_0.LOGIN_WITH_EMAIL_ALLOWED,re1_0.MASTER_ADMIN_CLIENT,re1_0.NAME,re1_0.NOT_BEFORE,re1_0.OFFLINE_SESSION_IDLE_TIMEOUT,re1_0.OTP_POLICY_ALG,re1_0.OTP_POLICY_DIGITS,re1_0.OTP_POLICY_COUNTER,re1_0.OTP_POLICY_WINDOW,re1_0.OTP_POLICY_PERIOD,re1_0.OTP_POLICY_TYPE,re1_0.PASSWORD_POLICY,re1_0.REFRESH_TOKEN_MAX_REUSE,re1_0.REGISTRATION_ALLOWED,re1_0.REG_EMAIL_AS_USERNAME,re1_0.REGISTRATION_FLOW,re1_0.REMEMBER_ME,re1_0.RESET_CREDENTIALS_FLOW,re1_0.RESET_PASSWORD_ALLOWED,re1_0.REVOKE_REFRESH_TOKEN,re1_0.SSL_REQUIRED,re1_0.SSO_IDLE_TIMEOUT,re1_0.SSO_IDLE_TIMEOUT_REMEMBER_ME,re1_0.SSO_MAX_LIFESPAN,re1_0.SSO_MAX_LIFESPAN_REMEMBER_ME,re1_0.VERIFY_EMAIL,a1_0.REALM_ID,a1_0.NAME,a1_0.VALUE from REALM re1_0 left join REALM_ATTRIBUTE a1_0 on re1_0.ID=a1_0.REALM_ID where re1_0.ID=$1 +SELECT select u1_0.ID,u1_0.CREATED_TIMESTAMP,u1_0.EMAIL,u1_0.EMAIL_CONSTRAINT,u1_0.EMAIL_VERIFIED,u1_0.ENABLED,u1_0.FEDERATION_LINK,u1_0.FIRST_NAME,u1_0.LAST_MODIFIED_TIMESTAMP,u1_0.LAST_NAME,u1_0.NOT_BEFORE,u1_0.REALM_ID,u1_0.SERVICE_ACCOUNT_CLIENT_LINK,u1_0.USERNAME from USER_GROUP_MEMBERSHIP ugme1_0 join USER_ENTITY u1_0 on u1_0.ID=ugme1_0.USER_ID where ugme1_0.GROUP_ID=$1 order by u1_0.USERNAME offset $2 rows fetch first $3 rows only SELECT authentication_execution select e1_0.FLOW_ID,e1_0.ID,e1_0.AUTHENTICATOR,e1_0.AUTH_CONFIG,e1_0.AUTHENTICATOR_FLOW,e1_0.AUTH_FLOW_ID,e1_0.PRIORITY,e1_0.REALM_ID,e1_0.REQUIREMENT from AUTHENTICATION_EXECUTION e1_0 where e1_0.FLOW_ID = any ($1) SELECT authentication_flow select af1_0.REALM_ID,af1_0.ID,af1_0.ALIAS,af1_0.BUILT_IN,af1_0.DESCRIPTION,af1_0.PROVIDER_ID,af1_0.TOP_LEVEL from AUTHENTICATION_FLOW af1_0 where af1_0.REALM_ID=$1 SELECT authenticator_config select a1_0.REALM_ID,a1_0.ID,a1_0.ALIAS from AUTHENTICATOR_CONFIG a1_0 where a1_0.REALM_ID=$1 @@ -26,6 +27,7 @@ SELECT credential select c1_0.USER_ID,c1_0.ID,c1_0.CREATED_DATE,c1_0.CREDENTIAL_ SELECT credential select ce1_0.ID,ce1_0.CREATED_DATE,ce1_0.CREDENTIAL_DATA,ce1_0.PRIORITY,ce1_0.SALT,ce1_0.SECRET_DATA,ce1_0.TYPE,ce1_0.USER_ID,ce1_0.USER_LABEL from CREDENTIAL ce1_0 where ce1_0.USER_ID=$1 order by ce1_0.PRIORITY SELECT default_client_scope select dcsrme1_0.SCOPE_ID from DEFAULT_CLIENT_SCOPE dcsrme1_0 where dcsrme1_0.REALM_ID=$1 and dcsrme1_0.DEFAULT_SCOPE=$2 SELECT federated_identity select fie1_0.IDENTITY_PROVIDER,fie1_0.USER_ID,fie1_0.REALM_ID,fie1_0.TOKEN,fie1_0.FEDERATED_USER_ID,fie1_0.FEDERATED_USERNAME from FEDERATED_IDENTITY fie1_0 where fie1_0.USER_ID=$1 +SELECT federated_identity select fie1_0.IDENTITY_PROVIDER,fie1_0.USER_ID,fie1_0.REALM_ID,fie1_0.TOKEN,fie1_0.FEDERATED_USER_ID,fie1_0.FEDERATED_USERNAME from FEDERATED_IDENTITY fie1_0 where fie1_0.USER_ID=$1 SELECT identity_provider select count(ipe2_0.INTERNAL_ID) from IDENTITY_PROVIDER ipe1_0,IDENTITY_PROVIDER ipe2_0 where ipe1_0.REALM_ID=$1 SELECT keycloak_role select re1_0.ID,re1_0.CLIENT,re1_0.CLIENT_REALM_CONSTRAINT,re1_0.CLIENT_ROLE,re1_0.DESCRIPTION,re1_0.NAME,re1_0.REALM_ID from KEYCLOAK_ROLE re1_0 where re1_0.ID=$1 SELECT offline_client_session select pcse1_0.CLIENT_ID,pcse1_0.CLIENT_STORAGE_PROVIDER,pcse1_0.EXTERNAL_CLIENT_ID,pcse1_0.OFFLINE_FLAG,pcse1_0.USER_SESSION_ID,pcse1_0.DATA,pcse1_0.TIMESTAMP,pcse1_0.VERSION from OFFLINE_CLIENT_SESSION pcse1_0 where (pcse1_0.CLIENT_ID,pcse1_0.CLIENT_STORAGE_PROVIDER,pcse1_0.EXTERNAL_CLIENT_ID,pcse1_0.OFFLINE_FLAG,pcse1_0.USER_SESSION_ID) in (($1,$2,$3,$4,$5)) From 382855de72f9394684ea3e1f155a6579ae794618 Mon Sep 17 00:00:00 2001 From: Andrei Kvapil Date: Fri, 28 Aug 2026 12:23:50 +0200 Subject: [PATCH 3/4] fix(wire): flag ambiguous-column ciphertext passthrough and refuse double-encryption MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit On the joined read path a PII column claimed by two relations was skipped by the planner and left as a raw $KKP$ envelope. Both leak detectors gate on the plan being empty, but when another column in the join resolves the plan is non-empty, so the skipped column reached Keycloak with nothing counted or logged — a silent ciphertext leak. Carry ambiguous fields through the read plan (ReadPlan.Ambiguous) so DecryptDataRow can account for them: increment kkp_ciphertext_passthrough_total{reason="ambiguous-column"} and emit a WARN when such a field still holds an envelope. A wire-level regression asserts the counter increments on a one-ambiguous-one-unambiguous join. Also guard the write path: encryptParam now fails loud on a value that already parses as a ciphertext envelope, instead of producing a double-encrypted row when a leaked ciphertext is written back. The runbook gains a recovery note tying the read-leak fix to clear-user-cache and the kkp_double_encrypted_total backlog check. Signed-off-by: Andrei Kvapil --- docs/runbook.md | 22 +++++++++++ internal/rewrite/read.go | 43 ++++++++++++++------ internal/rewrite/read_test.go | 18 ++++++--- internal/wire/joined_read_test.go | 66 ++++++++++++++++++++++++++++++- internal/wire/transform.go | 44 +++++++++++++++++++++ internal/wire/transform_test.go | 22 +++++++++++ 6 files changed, 196 insertions(+), 19 deletions(-) diff --git a/docs/runbook.md b/docs/runbook.md index 94949c1..eeb1e06 100644 --- a/docs/runbook.md +++ b/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 diff --git a/internal/rewrite/read.go b/internal/rewrite/read.go index 886000f..b458c6c 100644 --- a/internal/rewrite/read.go +++ b/internal/rewrite/read.go @@ -15,18 +15,31 @@ type ReadField struct { // ReadPlan is the decrypt-on-DataRow plan for a result set. type ReadPlan struct { + // Fields are the result fields to decrypt, each resolved to exactly one + // owning relation. Fields []ReadField + // Ambiguous lists result fields whose column is configured as PII by more + // than one of the statement's relations. The proxy cannot tell which + // relation's associated data sealed the value, so it will not decrypt them + // (guessing wrong fails the authentication tag and errors the whole query). + // They are carried here rather than dropped so DecryptDataRow can count and + // warn on any raw envelope they still carry: in a join where another column + // resolves, the plan is non-empty and the whole-row empty-plan leak + // detector never fires, so an ambiguous skip would otherwise be silent. + Ambiguous []ReadField } -// IsEmpty reports whether no result field needs decryption. -func (rp *ReadPlan) IsEmpty() bool { return len(rp.Fields) == 0 } +// IsEmpty reports whether the plan has nothing to act on — no field to decrypt +// and none left ambiguous to account for. +func (rp *ReadPlan) IsEmpty() bool { return len(rp.Fields) == 0 && len(rp.Ambiguous) == 0 } // PlanRead maps a query's result columns (learned from RowDescription, in // order) to the fields that may need decryption, given the relations the -// statement reads from. A field is included when exactly one of those +// statement reads from. A field is decrypted when exactly one of those // relations configures it as a PII column or an attribute value column; // non-encrypted rows pass through harmlessly because decryption is -// marker-driven. A statement with no known relation yields an empty plan. +// marker-driven. A column claimed by more than one relation is recorded as +// ambiguous instead. A statement with no known relation yields an empty plan. func (p *Planner) PlanRead(tables []string, columns []string) *ReadPlan { plan := &ReadPlan{} if len(tables) == 0 { @@ -34,7 +47,11 @@ func (p *Planner) PlanRead(tables []string, columns []string) *ReadPlan { } for i, col := range columns { c := strings.ToUpper(col) - if owner, ok := p.columnOwner(tables, c); ok { + owner, ambiguous := p.columnOwner(tables, c) + switch { + case ambiguous: + plan.Ambiguous = append(plan.Ambiguous, ReadField{Index: i, Column: c}) + case owner != "": plan.Fields = append(plan.Fields, ReadField{Index: i, Table: owner, Column: c}) } } @@ -42,23 +59,23 @@ func (p *Planner) PlanRead(tables []string, columns []string) *ReadPlan { } // columnOwner reports which of the statement's relations configures the result -// column as PII. A column claimed by two joined relations is ambiguous: the -// associated data binds a ciphertext to one specific table, so decrypting under -// the wrong one fails the authentication tag and errors the query. Leave those -// alone — the read-path leak detector flags them instead. -func (p *Planner) columnOwner(tables []string, column string) (string, bool) { - owner := "" +// column as PII. It returns (owner, false) when exactly one relation claims it, +// ("", true) when more than one does — an ambiguous column the proxy cannot +// safely decrypt (the associated data binds a ciphertext to one specific table, +// so decrypting under the wrong one fails the authentication tag) — and +// ("", false) when none do. +func (p *Planner) columnOwner(tables []string, column string) (owner string, ambiguous bool) { for _, t := range tables { t = strings.ToUpper(t) if t == "" || !p.isReadablePIIColumn(t, column) { continue } if owner != "" && owner != t { - return "", false + return "", true } owner = t } - return owner, owner != "" + return owner, false } func (p *Planner) isReadablePIIColumn(table, column string) bool { diff --git a/internal/rewrite/read_test.go b/internal/rewrite/read_test.go index b851488..3f5da9a 100644 --- a/internal/rewrite/read_test.go +++ b/internal/rewrite/read_test.go @@ -112,7 +112,8 @@ func TestPlanReadJoinedTables(t *testing.T) { // TestPlanReadAmbiguousColumnSkipped: when two joined relations both configure // the same column, the proxy cannot tell which AAD the value was sealed with. // Guessing would fail the authentication tag and error the whole query, so the -// field is left for the passthrough detector to flag. +// field is not decrypted — but it is carried in plan.Ambiguous (not silently +// dropped) so DecryptDataRow can count and warn on any envelope it still holds. func TestPlanReadAmbiguousColumnSkipped(t *testing.T) { t.Parallel() @@ -122,11 +123,18 @@ func TestPlanReadAmbiguousColumnSkipped(t *testing.T) { fs.SetColumn("FED_USER_ENTITY", "EMAIL", rule) p := NewPlanner(fs) - if plan := p.PlanRead([]string{"user_entity", "fed_user_entity"}, []string{"email"}); !plan.IsEmpty() { - t.Fatalf("ambiguous column planned anyway: %+v", plan.Fields) + plan := p.PlanRead([]string{"user_entity", "fed_user_entity"}, []string{"email"}) + if len(plan.Fields) != 0 { + t.Fatalf("ambiguous column decrypted anyway: %+v", plan.Fields) + } + if len(plan.Ambiguous) != 1 || plan.Ambiguous[0].Column != "EMAIL" || plan.Ambiguous[0].Index != 0 { + t.Fatalf("ambiguous column not carried through the plan: %+v", plan.Ambiguous) + } + if plan.IsEmpty() { + t.Fatal("plan with an ambiguous field reports empty; the leak detector would never see it") } // Unambiguous on its own. - if plan := p.PlanRead([]string{"user_entity"}, []string{"email"}); len(plan.Fields) != 1 { - t.Fatalf("single-table read lost its field: %+v", plan.Fields) + if plan := p.PlanRead([]string{"user_entity"}, []string{"email"}); len(plan.Fields) != 1 || len(plan.Ambiguous) != 0 { + t.Fatalf("single-table read lost its field or misclassified it: %+v", plan) } } diff --git a/internal/wire/joined_read_test.go b/internal/wire/joined_read_test.go index 646b279..f5c4bfa 100644 --- a/internal/wire/joined_read_test.go +++ b/internal/wire/joined_read_test.go @@ -1,11 +1,13 @@ package wire import ( + "strings" "testing" "github.com/jackc/pgproto3/v2" "github.com/prometheus/client_golang/prometheus/testutil" + "github.com/cozystack/keycloak-kms-proxy/internal/config" "github.com/cozystack/keycloak-kms-proxy/internal/crypto" "github.com/cozystack/keycloak-kms-proxy/internal/observe" "github.com/cozystack/keycloak-kms-proxy/internal/rewrite" @@ -154,9 +156,14 @@ func TestJoinedSelectReuseDecrypts(t *testing.T) { // TestJoinedSelectLeavesNoCiphertextPassthrough asserts the incident counter // stays flat: it is the signal production alerts on, and it was the only // evidence the group-members read was leaking. +// +// Deliberately NOT t.Parallel(): it Reset()s the process-global +// kkp_ciphertext_passthrough_total counter and reads kkp_unrecognized_pii_sql_total, +// so running alongside another test that touches those metrics would race. +// Go runs the non-parallel tests one at a time before any parallel test +// resumes, which keeps these globals to this test for its duration. func TestJoinedSelectLeavesNoCiphertextPassthrough(t *testing.T) { observe.CiphertextPassthrough.Reset() - observe.UnrecognizedPIISQL.Add(0) before := testutil.ToFloat64(observe.UnrecognizedPIISQL) s := newEncryptingSession(t) @@ -179,3 +186,60 @@ func TestJoinedSelectLeavesNoCiphertextPassthrough(t *testing.T) { t.Errorf("joined PII SELECT still counted as unrecognised %v times", got) } } + +// ambiguousJoinSelect joins two relations that both configure EMAIL as PII and +// selects EMAIL (ambiguous — the proxy cannot know which relation's associated +// data sealed it) alongside USERNAME (owned only by USER_ENTITY, so it still +// decrypts). Before the fix the ambiguous EMAIL was dropped from a non-empty +// plan and left the proxy as a raw envelope with nothing counted or logged. +const ambiguousJoinSelect = "select ue.USERNAME,ue.EMAIL " + + "from USER_ENTITY ue join FED_USER_ENTITY fue on fue.ID=ue.ID" + +// TestJoinedAmbiguousColumnCountsPassthrough is the finding-1 regression: a join +// carrying one ambiguous and one unambiguous PII column must still increment +// kkp_ciphertext_passthrough_total for the ambiguous skip, and must decrypt the +// unambiguous column. +// +// Deliberately NOT t.Parallel(): like TestJoinedSelectLeavesNoCiphertextPassthrough +// it Reset()s the process-global kkp_ciphertext_passthrough_total counter. +func TestJoinedAmbiguousColumnCountsPassthrough(t *testing.T) { + observe.CiphertextPassthrough.Reset() + + fs := config.New() + fs.SetColumn("USER_ENTITY", "USERNAME", config.Rule{Scheme: crypto.SchemeDeterministic}) + fs.SetColumn("USER_ENTITY", "EMAIL", config.Rule{Scheme: crypto.SchemeNonDeterministic}) + fs.SetColumn("FED_USER_ENTITY", "EMAIL", config.Rule{Scheme: crypto.SchemeNonDeterministic}) + s := NewSession(rewrite.NewPlanner(fs), newTestCipher(t)) + + if err := s.OnParse(&pgproto3.Parse{Name: "S_amb", Query: ambiguousJoinSelect}); err != nil { + t.Fatalf("OnParse: %v", err) + } + s.OnDescribe(&pgproto3.Describe{ObjectType: 'S', Name: "S_amb"}) + s.OnBind(&pgproto3.Bind{DestinationPortal: "C_1", PreparedStatement: "S_amb"}) + s.OnExecute(&pgproto3.Execute{Portal: "C_1"}) + s.OnSync() + s.OnRowDescription(rowDescriptionFor([]string{"username", "email"})) + + dr := &pgproto3.DataRow{Values: [][]byte{ + storedValue(t, s, crypto.SchemeDeterministic, "USER_ENTITY", "USERNAME", "ivy"), + // The ambiguous EMAIL is a genuine envelope; which relation sealed it is + // exactly what the proxy cannot know, so it stays raw in the output. + storedValue(t, s, crypto.SchemeNonDeterministic, "USER_ENTITY", "EMAIL", "ivy@example.test"), + }} + if err := s.DecryptDataRow(dr); err != nil { + t.Fatalf("DecryptDataRow: %v", err) + } + + // Unambiguous column decrypts. + if got := string(dr.Values[0]); got != "ivy" { + t.Errorf("username not decrypted on ambiguous join: got %q, want %q", got, "ivy") + } + // Ambiguous column is left as a raw envelope, not guessed at. + if got := string(dr.Values[1]); !strings.HasPrefix(got, "$KKP$") { + t.Errorf("ambiguous email should stay a raw envelope, got %q", got) + } + // The incident counter fires exactly once for the ambiguous skip. + if leaked := testutil.ToFloat64(observe.CiphertextPassthrough.WithLabelValues("ambiguous-column")); leaked != 1 { + t.Errorf("kkp_ciphertext_passthrough_total{reason=\"ambiguous-column\"} = %v, want 1", leaked) + } +} diff --git a/internal/wire/transform.go b/internal/wire/transform.go index eee6e74..6d35cb9 100644 --- a/internal/wire/transform.go +++ b/internal/wire/transform.go @@ -230,6 +230,16 @@ func (s *Session) encryptParam(b *pgproto3.Bind, pr rewrite.ParamRule) error { return nil // SQL NULL. } + // Fail loud instead of double-encrypting: a value that already parses as a + // valid ciphertext envelope is a leaked ciphertext being written back + // (e.g. after a read-path passthrough window), not user plaintext. Keycloak + // never sends a well-formed $KKP$ envelope as a PII value, so refusing here + // cannot break the normal encrypt path. + if _, ok, _ := crypto.Parse(string(value)); ok { + return fmt.Errorf("wire: refusing to encrypt $%d (%s.%s): value already carries a ciphertext envelope (double-encryption; a leaked ciphertext written back)", + pr.Param, pr.Table, pr.Column) + } + plaintext := value if pr.Rule.LowercaseNormalize { plaintext = []byte(crypto.NormalizeLowercase(string(value))) @@ -269,6 +279,17 @@ func (s *Session) DecryptDataRow(dr *pgproto3.DataRow) error { flagCiphertextLeak(dr, "empty-plan") return nil } + + // A column claimed by more than one joined relation is ambiguous: the proxy + // cannot know which relation's associated data sealed it, so it is left + // undecrypted. A non-empty plan (some other column resolved) means the + // whole-row empty-plan detector above never sees it, so count and warn here + // — an ambiguous-skip leak must never be silent. + flagAmbiguousPassthrough(dr, plan.Ambiguous) + + if len(plan.Fields) == 0 { + return nil // only ambiguous fields: nothing to decrypt. + } if s.cipher == nil { return errNoCipher } @@ -318,6 +339,29 @@ func flagCiphertextLeak(dr *pgproto3.DataRow, reason string) { } } +// flagAmbiguousPassthrough counts and warns when an ambiguous PII column — one +// claimed by more than one joined relation, so the proxy cannot resolve which +// associated data sealed it — reaches the client still carrying a ciphertext +// envelope. A non-empty read plan hides this from the whole-row empty-plan +// detector, so it is accounted for explicitly. Counts once per DataRow, matching +// kkp_ciphertext_passthrough_total's per-row unit. +func flagAmbiguousPassthrough(dr *pgproto3.DataRow, ambiguous []rewrite.ReadField) { + for _, f := range ambiguous { + if f.Index < 0 || f.Index >= len(dr.Values) { + continue + } + v := dr.Values[f.Index] + if v == nil { + continue + } + if _, ok, err := crypto.Parse(string(v)); ok || err != nil { + observe.CiphertextPassthrough.WithLabelValues("ambiguous-column").Inc() + log.Printf("kkp: WARN ciphertext passed through undecrypted (ambiguous-column) — column %q is claimed by multiple joined relations, associated data unknown", f.Column) + return + } + } +} + // classifyErr buckets decrypt errors into low-cardinality reasons for the // kkp_decrypt_failures_total metric. Refine as new failure modes appear. func classifyErr(err error) string { diff --git a/internal/wire/transform_test.go b/internal/wire/transform_test.go index 73c26f5..a81dfcd 100644 --- a/internal/wire/transform_test.go +++ b/internal/wire/transform_test.go @@ -227,6 +227,28 @@ func TestEncryptBindLikeEscapedWildcardsTreatedLiteral(t *testing.T) { } } +// TestEncryptBindRefusesDoubleEncryption: a value that already parses as a +// ciphertext envelope is a leaked ciphertext being written back, not user +// plaintext. Encrypting it again would produce a double-encrypted row, so the +// write path fails loud instead. +func TestEncryptBindRefusesDoubleEncryption(t *testing.T) { + t.Parallel() + + s := newEncryptingSession(t) + if err := s.OnParse(&pgproto3.Parse{Name: "ins", Query: "INSERT INTO user_entity (email) VALUES ($1)"}); err != nil { + t.Fatalf("OnParse: %v", err) + } + // A genuine envelope, as a leaked ciphertext read back by Keycloak would be. + envelope, err := s.cipher.Encrypt(0 /*deterministic*/, []byte("dana@example.com"), rewrite.AAD("USER_ENTITY", "EMAIL")) + if err != nil { + t.Fatalf("Encrypt: %v", err) + } + b := &pgproto3.Bind{PreparedStatement: "ins", Parameters: [][]byte{[]byte(envelope)}} + if err := s.EncryptBind(b); err == nil { + t.Fatal("EncryptBind accepted an already-encrypted value; double-encryption not refused") + } +} + func TestEncryptBindNullPassthrough(t *testing.T) { t.Parallel() From 3820b6bd416ae0dde52e3874eaa7333671de9520 Mon Sep 17 00:00:00 2001 From: Andrei Kvapil Date: Fri, 28 Aug 2026 12:24:08 +0200 Subject: [PATCH 4/4] test(rewrite): parse-tree conformance oracle, CTE resolution, corpus dedup Harden the analyser and its runtime-SQL conformance gate against the shapes Kirill flagged: - FromTables now resolves CTE references to their underlying relations instead of reporting the CTE alias as a base table (WITH m AS (...) ... FROM m); a WITH RECURSIVE self-reference terminates. - The conformance gate derives the expected relation set from the parse tree via an independent walker rather than a whitespace scan. This fixes the comma-join false negative (r,USER_ENTITY), honours the WHERE sub-query exclusion the doc promised, and cannot be fooled by a string literal containing "from". - Drop a byte-identical duplicate FEDERATED_IDENTITY line from the golden corpus (the generator dedups on (Kind, Table, SQL), so the file must be reproducible) and add TestRuntimeCorpusUnique to guard against a repeat. Doc fixes: Analysis.Table now states a comma join keeps it non-empty naming one of several relations; targetTables no longer over-promises for UPDATE...FROM / DELETE...USING; the test-file header enumerates the new coverage tests; the no-op UnrecognizedPIISQL.Add(0) is removed and the metric-mutating passthrough test documents why it must not run parallel. Signed-off-by: Andrei Kvapil --- internal/rewrite/analyze.go | 132 +++++++++++++++------ internal/rewrite/fromtables_test.go | 46 ++++++++ internal/rewrite/runtime_corpus_test.go | 142 ++++++++++++++++++++--- testdata/keycloak/26.0.0/runtime-sql.txt | 1 - 4 files changed, 267 insertions(+), 54 deletions(-) diff --git a/internal/rewrite/analyze.go b/internal/rewrite/analyze.go index 4bb54c8..c4c2ba5 100644 --- a/internal/rewrite/analyze.go +++ b/internal/rewrite/analyze.go @@ -53,19 +53,22 @@ 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 — notably when the FROM - // clause is a join, whose first item is the join expression rather than a - // range variable. + // 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; 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. + // 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. @@ -188,8 +191,12 @@ func analyzeSelect(sel *pg.SelectStmt) *Analysis { return a } -// targetTables wraps a write statement's target relation as its (single) read -// source, so a RETURNING result set decrypts like any other read. +// 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 @@ -198,48 +205,101 @@ func targetTables(table string) []string { } // selectTables collects every base relation a SELECT reads from, in the order -// they appear, without duplicates. +// 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 { - var tables []string - seen := make(map[string]bool) - collectSelectTables(sel, &tables, seen) - return tables + c := &tableCollector{seen: make(map[string]bool), cteActive: make(map[string]bool)} + c.walkSelect(sel, nil) + return c.tables } -// collectSelectTables walks a SELECT and its set-operation arms (UNION and -// friends), descending into every FROM item. -func collectSelectTables(sel *pg.SelectStmt, tables *[]string, seen map[string]bool) { +// 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 } - collectSelectTables(sel.GetLarg(), tables, seen) - collectSelectTables(sel.GetRarg(), tables, seen) + ctes = extendCTEScope(ctes, sel.GetWithClause()) + c.walkSelect(sel.GetLarg(), ctes) + c.walkSelect(sel.GetRarg(), ctes) for _, item := range sel.GetFromClause() { - collectFromItem(item, tables, seen) + c.walkFromItem(item, ctes) } } -// collectFromItem walks one FROM item: a range variable is a base relation, a -// join expression has two sides, and a sub-select carries its own FROM clause. -func collectFromItem(n *pg.Node, tables *[]string, seen map[string]bool) { +// 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: - name := n.GetRangeVar().GetRelname() - if name == "" || seen[strings.ToUpper(name)] { - return - } - seen[strings.ToUpper(name)] = true - *tables = append(*tables, name) + c.addRelation(n.GetRangeVar().GetRelname(), ctes) case n.GetJoinExpr() != nil: je := n.GetJoinExpr() - collectFromItem(je.GetLarg(), tables, seen) - collectFromItem(je.GetRarg(), tables, seen) + c.walkFromItem(je.GetLarg(), ctes) + c.walkFromItem(je.GetRarg(), ctes) case n.GetRangeSubselect() != nil: - collectSelectTables(n.GetRangeSubselect().GetSubquery().GetSelectStmt(), tables, seen) + 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 { diff --git a/internal/rewrite/fromtables_test.go b/internal/rewrite/fromtables_test.go index aeff0e5..d145662 100644 --- a/internal/rewrite/fromtables_test.go +++ b/internal/rewrite/fromtables_test.go @@ -97,6 +97,52 @@ func TestAnalyzeFromTablesDeduplicatesSelfJoin(t *testing.T) { } } +// 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() diff --git a/internal/rewrite/runtime_corpus_test.go b/internal/rewrite/runtime_corpus_test.go index 0a62e71..10eaa25 100644 --- a/internal/rewrite/runtime_corpus_test.go +++ b/internal/rewrite/runtime_corpus_test.go @@ -12,7 +12,13 @@ // target fails the test instead of silently falling into passthrough); // - for PII-touching SQL specifically, the proxy either built a write // plan (INSERT/UPDATE) or will build a read plan when the RowDescription -// arrives (i.e. `Table` is non-empty). +// arrives (i.e. `Table` is non-empty); +// - every PII relation a SELECT reads (whatever the FROM shape — join, +// comma join, sub-select, set operation, CTE) appears in `FromTables` +// (TestRuntimeCorpusFromTableCoverage), so a joined read cannot slip past +// the single-table gate above and leak ciphertext; +// - the corpus itself is a reproducible set: no (Kind, Table, SQL) tuple +// repeats (TestRuntimeCorpusUnique), matching what the generator emits. // // On a Keycloak upgrade: regenerate the golden via // `examples/encryption-demo/`-style capture, review the diff in the PR. @@ -23,6 +29,8 @@ import ( "os" "strings" "testing" + + pg "github.com/pganalyze/pg_query_go/v6" ) const runtimeCorpusPath = "../../testdata/keycloak/26.0.0/runtime-sql.txt" @@ -180,12 +188,14 @@ func TestRuntimeCorpusFromTableCoverage(t *testing.T) { if len(a.FromTables) > 1 { joined++ } - for tbl := range piiTables { - if !readsFrom(c.SQL, tbl) || containsFold(a.FromTables, tbl) { + for _, rel := range expectedReadRelations(t, c.SQL) { + if _, ok := piiTables[strings.ToLower(rel)]; !ok { continue } - t.Errorf("SELECT reads %s but it is missing from FromTables=%v (silent-passthrough hazard)\n sql=%s", - tbl, a.FromTables, truncateSQL(c.SQL, 200)) + if !containsFold(a.FromTables, rel) { + t.Errorf("SELECT reads PII relation %s but it is missing from FromTables=%v (silent-passthrough hazard)\n sql=%s", + rel, a.FromTables, truncateSQL(c.SQL, 200)) + } } } if joined == 0 { @@ -194,22 +204,120 @@ func TestRuntimeCorpusFromTableCoverage(t *testing.T) { t.Logf("runtime corpus covers %d multi-relation SELECTs", joined) } -// readsFrom reports whether the SQL names the relation right after a FROM or -// JOIN keyword — the positions that put it in the result set's scope. A -// mention inside a WHERE sub-query does not count: those rows never reach the -// client, so they need no decrypt plan. -func readsFrom(sql, table string) bool { - fields := strings.Fields(strings.ToLower(sql)) - for i := 1; i < len(fields); i++ { - prev := fields[i-1] - if prev != "from" && prev != "join" { +// TestRuntimeCorpusUnique — the golden corpus must be reproducible by its +// generator (cmd/capture-keycloak-sql), which dedups statements into a +// map[stmt]struct{} keyed on (Kind, Table, SQL) and writes the sorted set. A +// repeated tuple in the file could therefore never be regenerated, so guard +// against one creeping back in on the next capture/merge. +func TestRuntimeCorpusUnique(t *testing.T) { + t.Parallel() + seen := make(map[runtimeCase]int) + for i, c := range loadRuntimeCorpus(t) { + if first, dup := seen[c]; dup { + t.Errorf("duplicate corpus entry (lines %d and %d): %s %s\n sql=%s", + first+1, i+1, c.Kind, c.Table, truncateSQL(c.SQL, 200)) continue } - if strings.TrimLeft(fields[i], "(") == strings.ToLower(table) { - return true + seen[c] = i + } +} + +// expectedReadRelations parses a SELECT and returns the base relations in its +// result scope, independently of analyze.go: every relation named in a FROM +// clause (walking joins, comma joins, FROM sub-selects and set-operation arms, +// resolving CTE references to their underlying relations), and nothing that +// appears only in a WHERE sub-query — those rows never reach the client, so +// they need no decrypt plan. It is deliberately a second implementation of the +// walk in analyze.go, so this gate cross-checks FromTables rather than trusting +// the code it is testing; a string literal that merely contains the word "from" +// is a constant node, never a relation, so it cannot produce a false positive. +func expectedReadRelations(t *testing.T, sql string) []string { + t.Helper() + res, err := pg.Parse(sql) + if err != nil { + t.Fatalf("oracle parse %q: %v", sql, err) + } + o := &readOracle{seen: map[string]bool{}, active: map[string]bool{}} + for _, raw := range res.GetStmts() { + if sel := raw.GetStmt().GetSelectStmt(); sel != nil { + o.walkSelect(sel, nil) } } - return false + return o.rels +} + +// readOracle is the conformance test's independent relation collector. seen +// deduplicates; active guards a WITH RECURSIVE cycle. +type readOracle struct { + rels []string + seen map[string]bool + active map[string]bool +} + +func (o *readOracle) walkSelect(sel *pg.SelectStmt, ctes map[string]*pg.SelectStmt) { + if sel == nil { + return + } + ctes = oracleCTEScope(ctes, sel.GetWithClause()) + o.walkSelect(sel.GetLarg(), ctes) + o.walkSelect(sel.GetRarg(), ctes) + for _, item := range sel.GetFromClause() { + o.walkFrom(item, ctes) + } +} + +func (o *readOracle) walkFrom(n *pg.Node, ctes map[string]*pg.SelectStmt) { + if n == nil { + return + } + switch { + case n.GetRangeVar() != nil: + o.add(n.GetRangeVar().GetRelname(), ctes) + case n.GetJoinExpr() != nil: + o.walkFrom(n.GetJoinExpr().GetLarg(), ctes) + o.walkFrom(n.GetJoinExpr().GetRarg(), ctes) + case n.GetRangeSubselect() != nil: + o.walkSelect(n.GetRangeSubselect().GetSubquery().GetSelectStmt(), ctes) + } +} + +func (o *readOracle) add(name string, ctes map[string]*pg.SelectStmt) { + if name == "" { + return + } + key := strings.ToUpper(name) + if q, ok := ctes[key]; ok { + if o.active[key] { + return + } + o.active[key] = true + o.walkSelect(q, ctes) + delete(o.active, key) + return + } + if o.seen[key] { + return + } + o.seen[key] = true + o.rels = append(o.rels, name) +} + +func oracleCTEScope(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() { + if cte := node.GetCommonTableExpr(); cte != nil { + if q := cte.GetCtequery().GetSelectStmt(); q != nil { + scope[strings.ToUpper(cte.GetCtename())] = q + } + } + } + return scope } func containsFold(list []string, want string) bool { diff --git a/testdata/keycloak/26.0.0/runtime-sql.txt b/testdata/keycloak/26.0.0/runtime-sql.txt index 74a7249..992fb52 100644 --- a/testdata/keycloak/26.0.0/runtime-sql.txt +++ b/testdata/keycloak/26.0.0/runtime-sql.txt @@ -27,7 +27,6 @@ SELECT credential select c1_0.USER_ID,c1_0.ID,c1_0.CREATED_DATE,c1_0.CREDENTIAL_ SELECT credential select ce1_0.ID,ce1_0.CREATED_DATE,ce1_0.CREDENTIAL_DATA,ce1_0.PRIORITY,ce1_0.SALT,ce1_0.SECRET_DATA,ce1_0.TYPE,ce1_0.USER_ID,ce1_0.USER_LABEL from CREDENTIAL ce1_0 where ce1_0.USER_ID=$1 order by ce1_0.PRIORITY SELECT default_client_scope select dcsrme1_0.SCOPE_ID from DEFAULT_CLIENT_SCOPE dcsrme1_0 where dcsrme1_0.REALM_ID=$1 and dcsrme1_0.DEFAULT_SCOPE=$2 SELECT federated_identity select fie1_0.IDENTITY_PROVIDER,fie1_0.USER_ID,fie1_0.REALM_ID,fie1_0.TOKEN,fie1_0.FEDERATED_USER_ID,fie1_0.FEDERATED_USERNAME from FEDERATED_IDENTITY fie1_0 where fie1_0.USER_ID=$1 -SELECT federated_identity select fie1_0.IDENTITY_PROVIDER,fie1_0.USER_ID,fie1_0.REALM_ID,fie1_0.TOKEN,fie1_0.FEDERATED_USER_ID,fie1_0.FEDERATED_USERNAME from FEDERATED_IDENTITY fie1_0 where fie1_0.USER_ID=$1 SELECT identity_provider select count(ipe2_0.INTERNAL_ID) from IDENTITY_PROVIDER ipe1_0,IDENTITY_PROVIDER ipe2_0 where ipe1_0.REALM_ID=$1 SELECT keycloak_role select re1_0.ID,re1_0.CLIENT,re1_0.CLIENT_REALM_CONSTRAINT,re1_0.CLIENT_ROLE,re1_0.DESCRIPTION,re1_0.NAME,re1_0.REALM_ID from KEYCLOAK_ROLE re1_0 where re1_0.ID=$1 SELECT offline_client_session select pcse1_0.CLIENT_ID,pcse1_0.CLIENT_STORAGE_PROVIDER,pcse1_0.EXTERNAL_CLIENT_ID,pcse1_0.OFFLINE_FLAG,pcse1_0.USER_SESSION_ID,pcse1_0.DATA,pcse1_0.TIMESTAMP,pcse1_0.VERSION from OFFLINE_CLIENT_SESSION pcse1_0 where (pcse1_0.CLIENT_ID,pcse1_0.CLIENT_STORAGE_PROVIDER,pcse1_0.EXTERNAL_CLIENT_ID,pcse1_0.OFFLINE_FLAG,pcse1_0.USER_SESSION_ID) in (($1,$2,$3,$4,$5))