diff --git a/CLAUDE.md b/CLAUDE.md index fc65e541..ce9c9464 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -740,6 +740,18 @@ Invariants for anyone touching this path: + the `__default_*`/`duckgres_*` prefixes, which activation re-creates). It MUST run before replay on every CreateSession in shared-warm mode, and a wipe failure MUST fail the session. +- **The same boundary covers user ATTACHes** (`wipeUserCatalogs`, + `duckdbservice/user_catalogs.go`, runs immediately before the secret wipe): + DuckDB catalogs are instance-global too, so an external catalog attached by + one session (e.g. a postgres_scanner source holding a live, authenticated + upstream connection pool) would otherwise be inherited by the next session + on a hot-idle worker. Every non-internal catalog is detached except the + system-managed reserved set (`ducklake`, `delta`, `memory` — activation / + the pg_catalog compat layer own those). A wipe failure fails the session. + Note the inherited pool is frozen at its creation-time configuration: + postgres_scanner's `pg_pool_max_connections` SET has no set-callback and + only applies at pool creation, so a session inheriting a stale attach cannot + reconfigure its pool — one more reason the attach must not survive. - **Execute-then-persist ordering.** Persist only statements DuckDB accepted; a store failure after a successful exec is an ERROR telling the user the secret will NOT survive the session. Replay failures at session create are @@ -764,7 +776,8 @@ Invariants for anyone touching this path: the original (un-redacted) query so it can classify. - Touching the interception, wipe/replay, or payload shape → update `server/conn_user_secrets_test.go`, `duckdbservice/user_secrets_test.go`, - and the `persistent_user_secret`(+`_isolation`) assertions in + `duckdbservice/user_catalogs_test.go` (the catalog wipe), and the + `persistent_user_secret`(+`_isolation`) / `user_catalog_wipe` assertions in `tests/mw-dev/e2e/harness.sh`. ## Admin Console (VPC-private web UI, `kubernetes` tag) diff --git a/duckdbservice/service.go b/duckdbservice/service.go index 3e695406..e6bf5b0d 100644 --- a/duckdbservice/service.go +++ b/duckdbservice/service.go @@ -1224,11 +1224,14 @@ func (p *SessionPool) CreateSession(username, memoryLimit string, threads int, s // secrets are instance-global, so a hot-idle worker reused by a different // user of the same org would otherwise see the previous user's secrets — // both persistent ones and non-persistent (plain/TEMPORARY CREATE SECRET) - // ones that pass through to the worker. Wipe ALL user secrets first - // (mandatory — this is the cross-user isolation step), then replay this - // user's secrets from the control plane. Replay failures degrade to - // warnings; a wipe failure fails the session because handing user A's - // secrets to user B is not acceptable. + // ones that pass through to the worker. Attached catalogs are likewise + // instance-global, so user ATTACHes are wiped on the same boundary (a + // catalog holds a live upstream connection pool). Detach ALL user catalogs + // and wipe ALL user secrets first (mandatory — this is the cross-user + // isolation step), then replay this user's secrets from the control plane. + // Replay failures degrade to warnings; a wipe failure fails the session + // because handing user A's secrets or attached sources to user B is not + // acceptable. var secretWarnings []string switch { case p.sharedWarmMode && p.maxSessions != 1: @@ -1245,6 +1248,24 @@ func (p *SessionPool) CreateSession(username, memoryLimit string, threads int, s } case p.sharedWarmMode: secretCtx, secretCancel := context.WithTimeout(context.Background(), userSecretOpTimeout) + // Catalog wipe first: an attached catalog (e.g. postgres_scanner) holds + // a live, authenticated upstream connection pool, so it is the same + // cross-user isolation boundary as secrets, one level up. Detaching + // before the secret wipe also closes pools that reference the secrets + // about to be dropped. A failure fails the session, exactly like the + // secret wipe below. + wipedCatalogs, catalogWipeErr := wipeUserCatalogs(secretCtx, conn) + if catalogWipeErr != nil { + secretCancel() + _ = conn.Close() + p.mu.Lock() + p.reserved-- + p.mu.Unlock() + return nil, nil, fmt.Errorf("wipe user catalogs before session start: %w", catalogWipeErr) + } + if len(wipedCatalogs) > 0 { + slog.Info("Detached user catalogs left by previous session.", "user", username, "catalogs", wipedCatalogs) + } wiped, wipeErr := wipeUserSecrets(secretCtx, conn) if wipeErr != nil { secretCancel() diff --git a/duckdbservice/user_catalogs.go b/duckdbservice/user_catalogs.go new file mode 100644 index 00000000..ae1316f8 --- /dev/null +++ b/duckdbservice/user_catalogs.go @@ -0,0 +1,68 @@ +package duckdbservice + +import ( + "context" + "fmt" + "strings" +) + +// reservedCatalogNames are the attached catalogs that must survive the +// session-create wipe: the org's DuckLake/Delta catalogs (activation owns +// them — detaching would break the session's metadata init) and `memory`, +// which carries the pg_catalog compatibility layer (memory.main.*). +// `system`/`temp` are excluded structurally instead: duckdb_databases() +// marks them internal, and DuckDB refuses to detach them. +var reservedCatalogNames = map[string]bool{ + "ducklake": true, + "delta": true, + "memory": true, +} + +// wipeUserCatalogs detaches every user-attached catalog on the shared per-org +// DuckDB instance. DuckDB ATTACH is instance-global and a hot-idle worker is +// reused across sessions of an org, so without this a catalog attached by one +// session (say an external Postgres source with a live, authenticated +// connection pool inside postgres_scanner) is silently inherited by the next +// session — the same cross-user isolation boundary as wipeUserSecrets, one +// level up. (A stale inherited attach also carries its creation-time +// connection-pool configuration: postgres_scanner's pg_pool_max_connections +// SET only applies at pool creation, so the inheriting session cannot +// reconfigure the pool it just inherited.) +// +// A wipe failure fails the session: handing user A's attached sources (and +// their authenticated upstream connections) to user B is not acceptable. +// Returns the names that were detached. +func wipeUserCatalogs(ctx context.Context, h secretDBHandle) ([]string, error) { + rows, err := h.QueryContext(ctx, "SELECT database_name, internal FROM duckdb_databases()") + if err != nil { + return nil, fmt.Errorf("list databases: %w", err) + } + var names []string + for rows.Next() { + var name string + var internal bool + if err := rows.Scan(&name, &internal); err != nil { + _ = rows.Close() + return nil, err + } + if internal || reservedCatalogNames[name] { + continue + } + names = append(names, name) + } + if err := rows.Err(); err != nil { + return nil, err + } + _ = rows.Close() + + var wiped []string + for _, name := range names { + // Catalog names are user-controlled; quote defensively. + quoted := `"` + strings.ReplaceAll(name, `"`, `""`) + `"` + if _, err := h.ExecContext(ctx, "DETACH DATABASE IF EXISTS "+quoted); err != nil { + return wiped, fmt.Errorf("detach catalog %q: %w", name, err) + } + wiped = append(wiped, name) + } + return wiped, nil +} diff --git a/duckdbservice/user_catalogs_test.go b/duckdbservice/user_catalogs_test.go new file mode 100644 index 00000000..c41fc390 --- /dev/null +++ b/duckdbservice/user_catalogs_test.go @@ -0,0 +1,131 @@ +package duckdbservice + +import ( + "context" + "database/sql" + "testing" + "time" + + _ "github.com/duckdb/duckdb-go/v2" + "github.com/posthog/duckgres/server" +) + +func databaseNames(t *testing.T, db *sql.DB) map[string]bool { + t.Helper() + rows, err := db.Query("SELECT database_name FROM duckdb_databases()") + if err != nil { + t.Fatalf("duckdb_databases: %v", err) + } + defer func() { _ = rows.Close() }() + names := map[string]bool{} + for rows.Next() { + var name string + if err := rows.Scan(&name); err != nil { + t.Fatalf("scan: %v", err) + } + names[name] = true + } + return names +} + +// A hot-idle worker is reused across sessions of an org, and DuckDB attached +// catalogs are instance-global: a user-attached catalog (e.g. an external +// Postgres source) would otherwise linger into the next session — leaking the +// previous session's data sources (and, for postgres_scanner, its live pooled +// connections) to whoever gets the worker next. The session-create wipe must +// detach every user catalog while preserving the system-managed ones +// (ducklake, delta, memory, and the internal system/temp databases). +func TestWipeUserCatalogs(t *testing.T) { + db := openSecretTestDB(t) + mustExec := func(q string) { + t.Helper() + if _, err := db.Exec(q); err != nil { + t.Fatalf("exec %q: %v", q, err) + } + } + // User catalogs left behind by a "previous session". One name uses + // identifier-quoting edge cases to prove the wipe quotes safely. + mustExec("ATTACH ':memory:' AS userdb") + mustExec(`ATTACH ':memory:' AS "weird ""quoted"" name"`) + // Name-simulate the reserved system-managed catalogs (the real ones are + // attached by activation with these exact names). + mustExec("ATTACH ':memory:' AS ducklake") + mustExec("ATTACH ':memory:' AS delta") + + wiped, err := wipeUserCatalogs(context.Background(), db) + if err != nil { + t.Fatalf("wipeUserCatalogs: %v", err) + } + if len(wiped) != 2 { + t.Fatalf("expected 2 wiped catalogs, got %v", wiped) + } + names := databaseNames(t, db) + for _, gone := range []string{"userdb", `weird "quoted" name`} { + if names[gone] { + t.Errorf("catalog %q survived the wipe", gone) + } + } + for _, kept := range []string{"ducklake", "delta", "memory", "system", "temp"} { + if !names[kept] { + t.Errorf("reserved/internal catalog %q was wiped", kept) + } + } +} + +// CreateSession on a shared-warm (k8s) worker must detach catalogs attached by +// the previous session of the same worker: DuckDB ATTACH is instance-global +// and the worker is reused, so without the wipe the next session inherits the +// previous session's external sources (and their authenticated pools). +func TestCreateSessionWipesPreviousSessionCatalogs(t *testing.T) { + pool := &SessionPool{ + sessions: make(map[string]*Session), + stopRefresh: make(map[string]func()), + duckLakeSem: make(chan struct{}, 1), + cfg: server.Config{Users: map[string]string{"postgres": "postgres"}}, + startTime: time.Now(), + warmupDone: make(chan struct{}), + sharedWarmMode: true, + maxSessions: 1, // k8s workers run with DUCKGRES_DUCKDB_MAX_SESSIONS=1 + } + close(pool.warmupDone) + pool.createDBPair = func(server.Config, chan struct{}, string, time.Time, string) (*DuckDBPair, error) { + db, err := sql.Open("duckdb", "") + if err != nil { + return nil, err + } + return PairFromMain(db), nil + } + pool.activateDBConnection = func(*sql.DB, server.Config, chan struct{}, string) error { return nil } + if err := pool.activateTenant(ActivationPayload{ + WorkerControlMetadata: server.WorkerControlMetadata{ + OwnerEpoch: 1, + CPInstanceID: "cp-test", + WorkerID: 1, + }, + OrgID: "test-org", + }); err != nil { + t.Fatalf("activateTenant: %v", err) + } + + first, _, err := pool.CreateSession("alice", "", 0, nil) + if err != nil { + t.Fatalf("first CreateSession: %v", err) + } + // Alice's session attaches an external source catalog. + if _, err := pool.warmupDB.Exec("ATTACH ':memory:' AS userdb"); err != nil { + t.Fatalf("attach user catalog: %v", err) + } + if err := pool.DestroySession(first.ID); err != nil { + t.Fatalf("DestroySession: %v", err) + } + + if _, _, err := pool.CreateSession("bob", "", 0, nil); err != nil { + t.Fatalf("second CreateSession: %v", err) + } + if databaseNames(t, pool.warmupDB)["userdb"] { + t.Fatal("user-attached catalog from the previous session survived CreateSession") + } + if !databaseNames(t, pool.warmupDB)["memory"] { + t.Fatal("reserved memory catalog was wiped") + } +} diff --git a/tests/mw-dev/e2e/harness.sh b/tests/mw-dev/e2e/harness.sh index 5bfb09d8..585815c2 100755 --- a/tests/mw-dev/e2e/harness.sh +++ b/tests/mw-dev/e2e/harness.sh @@ -2555,6 +2555,38 @@ persistent_user_secret_isolation() { # org rootpw log "user secret isolation OK on $org ($u2 blind to root's persistent AND temporary secrets; root's stored copy intact)" } +# Cross-session catalog hygiene: DuckDB ATTACH is instance-global, so a catalog +# attached by one session (e.g. an external postgres_scanner source holding a +# live authenticated connection pool) would linger on a hot-idle worker and be +# inherited by the next session unless the session-create wipe detaches it. +# The second connection typically reuses the first's hot-idle worker — the leak +# path — but the invariant holds either way: a fresh session must never see +# another session's attached catalogs, while the system catalogs (ducklake, +# memory) must be preserved. +user_catalog_wipe() { # org password + org="$1"; pw="$2"; cname="e2e_probe_catalog_$$" + log "user catalog wipe on $org" + + # Same-session check: BOTH statements must ride ONE connection, so feed psql + # via stdin (each statement is its own Q message on the same session). Two + # separate `pg` calls would be two sessions — and the second one's + # CreateSession wipe would already have detached the probe. + out="$(printf "ATTACH IF NOT EXISTS ':memory:' AS %s;\nSELECT count(*) FROM duckdb_databases() WHERE database_name = '%s' AND NOT internal;\n" "$cname" "$cname" | \ + PGPASSWORD="$pw" psql \ + "sslmode=require host=$org$SNI_SUFFIX hostaddr=$CP_IP port=5432 user=root dbname=ducklake" \ + -v ON_ERROR_STOP=1 -tA 2>&1)" || fail "user catalog wipe: attach session failed: $out" + [ "$(printf '%s' "$out" | tail -1)" = "1" ] || fail "user catalog wipe: probe catalog not attached in its own session ($out)" + + # ...and gone from the NEXT fresh session (same hot-idle worker or not). + n="$(pg "$org" "$pw" ducklake "SELECT count(*) FROM duckdb_databases() WHERE database_name = '$cname' AND NOT internal")" + [ "$n" = "0" ] || fail "user catalog wipe: probe catalog leaked into a fresh session (count=$n)" + + # The system-managed catalogs must survive the wipe on every session. + n="$(pg "$org" "$pw" ducklake "SELECT count(*) FROM duckdb_databases() WHERE database_name IN ('ducklake','memory')")" + [ "$n" = "2" ] || fail "user catalog wipe: system catalogs missing on a fresh session (count=$n)" + log "user catalog wipe OK on $org (probe attached in-session, detached by next session, ducklake+memory preserved)" +} + # ---- resilience ----------------------------------------------------------- # Worker pod killed mid-life → CP refills and a fresh query succeeds. # Ported from TestK8sWorkerCrashRecovery. @@ -4362,6 +4394,7 @@ lane_cnpg() { # full wire/catalog/concurrency/sizing coverage on the cnpg org httpfs_retry_budget "$CNPG" "$cnpg_pw" # S3-503 retry budget raised per worker (applyHTTPFSRetryBudget) persistent_user_secret "$CNPG" "$cnpg_pw" # after rw_ducklake (org worker hot) persistent_user_secret_isolation "$CNPG" "$cnpg_pw" + user_catalog_wipe "$CNPG" "$cnpg_pw" pipeline_error_recovery "$CNPG" "$cnpg_pw" # after rw_ducklake (table writes proven) server_side_cursors "$CNPG" "$cnpg_pw" cancel_then_reuse_same_session "$CNPG" "$cnpg_pw"