Let every duckgres login authenticate to Trino

CLAUDE.md

@@ -1577,6 +1577,33 @@ password/tenant/catalog changes never propagate.
   `opa.ManagedCatalogPattern`, and the regex literal inside `policy.rego`.
   `TestTrinoCatalogNameMatchesManagedNamePattern` +
   `TestPolicyRegoContainsManagedNamePattern` fail if any one moves alone.
+- **Every duckgres login authenticates to Trino, under `<database_name>.<username>`.**
+  `ListTrinoEnabledOrgs` returns the org's logins in `Users`, and
+  `BuildTrinoAuthFiles` writes one `password.db` line per login with the
+  bcrypt hash **copied through unchanged** — it is the same hash pgwire
+  verifies, so one password works on both engines and nothing is re-hashed or
+  minted. The bare `<database_name>` principal survives alongside them. Three
+  rules that are load-bearing rather than cosmetic: (1) usernames are
+  projected through an **allowlist** (`trinoUsernamePattern`) because
+  duckgres barely validates them and a `:`, `,` or newline would let whoever
+  can create org users append lines to `password.db` — including an admin
+  line; (2) `rejectPrincipalCollisions` now also holds back orgs that derive
+  the same Trino username, since the password file is ONE flat namespace per
+  cell and a duplicate line is a cross-tenant auth bug; (3) the resource-group
+  selector's `orgCaptureRegex` captures only up to the first `.`, or every
+  login gets a private leaf with the full per-tenant limits and an org with
+  ten logins holds ten times its budget.
+- **A project-scoped login (`project_reader` / `project_user`) joins
+  `scope_<org>_team_<id>`, NOT the org group.** The scope group owns the same
+  catalog in `group_catalogs` — so the cross-tenant check is the unchanged
+  check — and carries a `group_scopes` document that narrows it to that
+  team's schemas. Scopes only ever SUBTRACT; keep it that way if the rules are
+  restructured. The scope comes from `OrgUserQueryAccess`, the same derivation
+  the pgwire path uses, so Trino and DuckDB cannot disagree about it, and a
+  scoped row whose scope will not resolve is DROPPED rather than projected
+  unscoped. Scoped logins get **no write authority at all** — `project_user`
+  is read/write on pgwire and read-only here, a narrowing; making it writable
+  means gating writes per-schema, not per-catalog.
 - **The Rego policy is the tenant-isolation boundary.** The cell can assume
   every per-org duckling role, so nothing below OPA stops org A reading org
   B's catalog. Treat `provisioner/opa/policy.rego` as security review.

controlplane/configstore/models.go

@@ -522,6 +522,69 @@ type TrinoEnabledOrg struct {
 	CellID           string
 	RootPasswordHash string                            // bcrypt hash from OrgUser row where Username = "root"
 	State            ManagedWarehouseProvisioningState // current state at read time
+	// Users are the org's own duckgres logins, each of which authenticates
+	// to Trino under TrinoUserPrincipal(Username) with the very same bcrypt
+	// hash it uses at the pgwire handshake. Populated by a second query in
+	// ListTrinoEnabledOrgs, hence `gorm:"-"` -- it is not a column on the
+	// row the outer join scans into.
+	Users []TrinoOrgUser `gorm:"-"`
+}
+
+// TrinoOrgUser is one of an org's duckgres logins, projected into the cell's
+// password file so a person who already has a pgwire credential can use that
+// same credential against Trino instead of sharing the org's root password.
+type TrinoOrgUser struct {
+	Username string
+	// PasswordHash is duckgres_org_users.password, copied through unchanged.
+	// It is bcrypt at cost 10, which Trino's file authenticator accepts as
+	// is (its floor is cost 8), so ONE password works on both engines and no
+	// separate Trino credential is ever minted or stored.
+	PasswordHash string
+	// Scope, when non-nil, restricts the login to one project's schemas.
+	// Mirrors duckgres_org_users.access_mode's project_reader / project_user
+	// modes; nil means an unrestricted org-wide login.
+	//
+	// The value is whatever OrgUserQueryAccess reports for this user, so the
+	// scope Trino enforces and the scope pgwire enforces are the same object
+	// derived by the same code -- see ListTrinoEnabledOrgs.
+	Scope *OrgUserQueryAccess
+	// TeamID is the project a scoped login is bound to, and is what its Trino
+	// group is keyed on. Non-nil exactly when Scope is: the two are set
+	// together and a scoped row with no team is dropped rather than
+	// projected (the table's CHECK constraint already forbids that shape).
+	//
+	// Carried here rather than on OrgUserQueryAccess because that type is the
+	// pgwire session path's policy object and has no reason to grow a field
+	// only the Trino projection reads.
+	TeamID *int64
+}
+
+// TrinoPrincipalSeparator joins an org's principal to one of its usernames to
+// form a cell-wide-unique Trino username.
+//
+// Trino's password file is ONE flat namespace per cell while duckgres keys a
+// login on (org, username) and recovers the org from SNI -- which a Trino
+// login carries no equivalent of. Two orgs may each have an `analyst`, and
+// two identical password-file lines would let one org's user authenticate
+// against the other's entry and land in the other's group. Qualifying the
+// username is what makes the flat namespace safe.
+//
+// `.` is the separator because a valid database_name cannot contain one (see
+// ValidateDatabaseName: a DNS label) and neither can a projectable username
+// (see projectableTrinoUsername), so `<org>.<user>` splits unambiguously and
+// the org prefix stays recoverable -- which the resource-group selector's
+// named capture depends on.
+const TrinoPrincipalSeparator = "."
+
+// TrinoUserPrincipal returns the Trino username for one of the org's duckgres
+// logins: the org's own principal, the separator, then the duckgres username.
+// Returns "" when either half is missing, which callers skip.
+func (o TrinoEnabledOrg) TrinoUserPrincipal(username string) string {
+	principal := o.TrinoPrincipal()
+	if principal == "" || username == "" {
+		return ""
+	}
+	return principal + TrinoPrincipalSeparator + username
 }
 
 // TrinoPrincipal is the tenant's customer-facing identity in Trino: the

controlplane/configstore/trino.go

@@ -3,6 +3,7 @@ package configstore
 import (
 	"errors"
 	"fmt"
+	"log/slog"
 	"time"
 
 	"gorm.io/gorm"
@@ -188,9 +189,15 @@ func (cs *ConfigStore) DisableTrino(orgID string) error {
 }
 
 // ListTrinoEnabledOrgs returns every org with ManagedWarehouseTrino.Enabled
-// = true joined against its `root` OrgUser row. The provisioner needs the
-// bcrypt hash to project the Trino password file, so this is a single join
-// rather than two round-trips.
+// = true joined against its `root` OrgUser row, each carrying the org's full
+// set of projectable logins in Users. The provisioner needs the bcrypt hashes
+// to project the Trino password file.
+//
+// The `root` join stays because database_name alone remains a principal in
+// its own right (TrinoPrincipal) for service-to-service use and for every
+// client configured before per-user logins existed. Users is the ADDITIONAL
+// per-human projection; root therefore appears twice, as `<db>` and as
+// `<db>.root`, and both authenticate against the same hash.
 //
 // Orgs that are Trino-enabled but have no `root` OrgUser are skipped — that
 // shape can't legitimately happen via the provisioning API (CreateOrgUser
@@ -229,9 +236,94 @@ func (cs *ConfigStore) ListTrinoEnabledOrgs() ([]TrinoEnabledOrg, error) {
 	if err != nil {
 		return nil, fmt.Errorf("list trino-enabled orgs: %w", err)
 	}
+	if len(out) == 0 {
+		return out, nil
+	}
+	if err := cs.attachTrinoOrgUsers(out); err != nil {
+		return nil, err
+	}
 	return out, nil
 }
 
+// trinoOrgUserRow is one (org, login) pair from the second listing query.
+type trinoOrgUserRow struct {
+	OrgID      string
+	Username   string
+	Password   string
+	AccessMode string
+	TeamID     *int64
+}
+
+// attachTrinoOrgUsers loads every projectable duckgres login for the listed
+// orgs and hangs it off the matching TrinoEnabledOrg.
+//
+// A second query rather than a widened join: the outer listing is one row per
+// org and the provisioner's per-org steps (catalog, tenant password, state)
+// all key on that shape, so fanning it out to one row per user would make
+// every caller de-duplicate. One extra round trip per reconcile tick is not
+// a cost worth that.
+//
+// Two rows are excluded in SQL, both fail-closed:
+//
+//   - disabled = true. duckgres_org_users.disabled is the per-user kill
+//     switch. Trino learns about a flip only when the projected Secret is
+//     re-read (kubelet sync + the group provider's file.refresh-period), so
+//     a disable takes effect here in up to a couple of minutes rather than
+//     instantly as it does on pgwire. That lag is worth stating wherever the
+//     kill switch is surfaced to operators; it is not a reason to leave the
+//     row out of the exclusion.
+//   - a blank password. There is no hash to project and Trino would reject
+//     the line anyway.
+//
+// Project-scoped logins ARE included, and carry their scope. The scope is
+// read through OrgUserQueryAccess -- the SAME derivation the pgwire session
+// path uses -- so Trino and DuckDB can never disagree about which schemas a
+// project login may read. A scoped row whose scope cannot be resolved is
+// dropped rather than projected unscoped: an unresolvable scope must never
+// silently widen into org-wide access.
+func (cs *ConfigStore) attachTrinoOrgUsers(orgs []TrinoEnabledOrg) error {
+	ids := make([]string, 0, len(orgs))
+	for _, o := range orgs {
+		ids = append(ids, o.OrgID)
+	}
+	var rows []trinoOrgUserRow
+	err := cs.db.Table("duckgres_org_users").
+		Select("org_id, username, password, access_mode, team_id").
+		Where("org_id IN ?", ids).
+		Where("disabled = ?", false).
+		Where("password <> ''").
+		Order("org_id ASC, username ASC").
+		Scan(&rows).Error
+	if err != nil {
+		return fmt.Errorf("list trino org users: %w", err)
+	}
+
+	byOrg := make(map[string][]TrinoOrgUser, len(orgs))
+	for _, r := range rows {
+		u := TrinoOrgUser{Username: r.Username, PasswordHash: r.Password}
+		if IsProjectScopedAccessMode(r.AccessMode) {
+			access, scoped := cs.OrgUserQueryAccess(r.OrgID, r.Username)
+			if !scoped || r.TeamID == nil {
+				// The row says scoped but the snapshot does not agree --
+				// an unloaded or stale snapshot, or a user written since
+				// the last poll. Projecting it now would grant the whole
+				// org catalog to a login that must only see one project.
+				// Drop it; the next tick projects it once both agree.
+				slog.Warn("Trino: skipping project-scoped login whose scope is unresolved.",
+					"org", r.OrgID, "user", r.Username, "access_mode", r.AccessMode)
+				continue
+			}
+			u.Scope = &access
+			u.TeamID = r.TeamID
+		}
+		byOrg[r.OrgID] = append(byOrg[r.OrgID], u)
+	}
+	for i := range orgs {
+		orgs[i].Users = byOrg[orgs[i].OrgID]
+	}
+	return nil
+}
+
 // GetManagedWarehouseTrino reads the Trino row for an org. Returns
 // (nil, nil) when no row exists so callers can distinguish "never
 // configured" from a DB error.

controlplane/provisioner/opa/builder.go

@@ -47,16 +47,22 @@ func NewBuilder() BundleBuilder {
 // activates a deny-everything policy (since no group owns any catalog).
 // That is the correct bootstrap behaviour: until the provisioner pushes
 // a populated GroupCatalogs, all customer queries are denied.
-func (defaultBuilder) BuildBundle(gc GroupCatalogs) ([]byte, error) {
-	data, err := buildDataDocument(gc)
+//
+// gs carries the project scopes that narrow individual groups. A nil or empty
+// GroupScopes means no group is scoped, which is the pre-scopes behaviour: a
+// group reads the whole catalog it owns. Both documents are always emitted so
+// the policy's `data.group_scopes[g]` lookup is undefined-on-missing-key
+// rather than an error on a missing document.
+func (defaultBuilder) BuildBundle(gc GroupCatalogs, gs GroupScopes) ([]byte, error) {
+	data, err := buildDataDocument(gc, gs)
 	if err != nil {
 		return nil, fmt.Errorf("build data document: %w", err)
 	}
 
 	b := bundle.Bundle{
 		Manifest: bundle.Manifest{
 			Revision: bundleRevision,
-			Roots:    &[]string{"trino", "group_catalogs"},
+			Roots:    &[]string{"trino", "group_catalogs", "group_scopes"},
 		},
 		Modules: []bundle.ModuleFile{
 			{
@@ -85,26 +91,30 @@ func (defaultBuilder) BuildBundle(gc GroupCatalogs) ([]byte, error) {
 // stores under data.<root>. We always emit `group_catalogs` even when gc is
 // nil so the policy's `data.group_catalogs[group][catalog]` lookup is
 // well-formed (undefined-on-missing-key, not error-on-missing-document).
-func buildDataDocument(gc GroupCatalogs) (map[string]interface{}, error) {
+func buildDataDocument(gc GroupCatalogs, gs GroupScopes) (map[string]interface{}, error) {
 	// JSON round-trip ensures we emit canonical JSON-decoded types
 	// (map[string]interface{} and bool) regardless of what the caller
 	// passes in. OPA's bundle loader expects these types and treats
 	// concrete map[string]map[string]bool as opaque if it ever leaks
 	// through. Round-tripping is also a stable serialization for tests.
+	if gc == nil {
+		// Marshalling a nil map emits "null"; substitute an empty object so
+		// the policy sees `data.group_catalogs == {}` not `null`.
+		gc = GroupCatalogs{}
+	}
+	if gs == nil {
+		gs = GroupScopes{}
+	}
 	raw, err := json.Marshal(struct {
 		GroupCatalogs GroupCatalogs `json:"group_catalogs"`
-	}{GroupCatalogs: gc})
+		GroupScopes   GroupScopes   `json:"group_scopes"`
+	}{GroupCatalogs: gc, GroupScopes: gs})
 	if err != nil {
-		return nil, fmt.Errorf("marshal group_catalogs: %w", err)
-	}
-	if gc == nil {
-		// Marshalling a nil map emits "null"; substitute an empty object
-		// so the policy sees `data.group_catalogs == {}` not `null`.
-		raw = []byte(`{"group_catalogs":{}}`)
+		return nil, fmt.Errorf("marshal bundle data: %w", err)
 	}
 	var data map[string]interface{}
 	if err := json.Unmarshal(raw, &data); err != nil {
-		return nil, fmt.Errorf("unmarshal group_catalogs: %w", err)
+		return nil, fmt.Errorf("unmarshal bundle data: %w", err)
 	}
 	return data, nil
 }