feat(trino): let every duckgres login authenticate to Trino

CLAUDE.md

@@ -1589,7 +1589,15 @@ password/tenant/catalog changes never propagate.
   and the coordinator is reachable. `DUCKGRES_TRINO_CLIENT_URL` names the HTTPS
   endpoint clients should dial; when unset, the control plane falls back to its
   coordinator URL and TLS server name. The response contains host, port and the
-  tenant principal, never a password or password hash.
+  tenant principal, never a password or password hash. A client URL whose
+  leading host label is `{database_name}` (registry `client_url` or the env
+  var; `ResolveTrinoClientURL`) advertises each org's own
+  `<database_name>.<domain>` with username `root`: the Trino fork qualifies a
+  login with the org its host names
+  (`http-server.authentication.password.host-qualified-user.domains`), so a
+  user types the same username and host on both engines. Set it only once
+  that host actually reaches Trino — advertising an unrouted host is worse
+  than advertising the shared one.
 - **`Reconcile` order is load-bearing**: cluster secrets → auth files →
   resource groups → OPA bundle → tenant passwords → catalogs, and the
   `globalErr` gate SKIPS the catalog step if any projection failed. A
@@ -1633,6 +1641,40 @@ password/tenant/catalog changes never propagate.
   `opa.ManagedCatalogPattern`, and the regex literal inside `policy.rego`.
   `TestTrinoCatalogNameMatchesManagedNamePattern` +
   `TestPolicyRegoContainsManagedNamePattern` fail if any one moves alone.
+- **A Trino principal resolves to its org by EXACT lookup, never by splitting.**
+  `configstore.NewTrinoPrincipalOwners` indexes the principals the auth files
+  project (bare `<database_name>` → root, `<database_name>.<username>` → that
+  login). Usage metering (`trino_usage_collector.go`, which also passes the
+  duckgres username — not the principal — to the team resolver) and the admin
+  console both use it, so a per-user login's queries are its org's usage and a
+  principal outside the password file is never attributed to anyone.
+- **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/admin/trino.go

@@ -103,15 +103,44 @@ type TrinoOrgStatus struct {
 	QueuedQueries  int `json:"queued_queries"`
 }
 
-func (c TrinoCell) connectionFor(username string) *TrinoConnection {
-	if username == "" {
+// TrinoClientHostPlaceholder, as the leading label of a cell's client URL,
+// stands for each org's database_name: `https://{database_name}.<domain>`
+// gives every org the same host name it uses for pgwire. Trino qualifies a
+// login with the org that host names, so the connection then advertises the
+// org's own duckgres username (root) rather than the bare org principal.
+const TrinoClientHostPlaceholder = "{database_name}"
+
+// ResolveTrinoClientURL substitutes databaseName into a client URL template.
+// A URL without the placeholder is returned unchanged with perOrgHost false.
+// ok is false when the placeholder is anywhere but the leading host label, or
+// databaseName is not a DNS label and so cannot be a host name label.
+func ResolveTrinoClientURL(template, databaseName string) (resolved string, perOrgHost bool, ok bool) {
+	if !strings.Contains(template, TrinoClientHostPlaceholder) {
+		return template, false, true
+	}
+	const prefix = "https://" + TrinoClientHostPlaceholder + "."
+	if !strings.HasPrefix(template, prefix) || strings.Count(template, TrinoClientHostPlaceholder) != 1 {
+		return "", true, false
+	}
+	if configstore.ValidateDatabaseName(databaseName) != nil {
+		return "", true, false
+	}
+	return strings.Replace(template, TrinoClientHostPlaceholder, databaseName, 1), true, true
+}
+
+func (c TrinoCell) connectionFor(principal string) *TrinoConnection {
+	if principal == "" {
 		return nil
 	}
 
 	clientURL := c.ClientURL
 	if clientURL == "" {
 		clientURL = c.CoordinatorURL
 	}
+	clientURL, perOrgHost, ok := ResolveTrinoClientURL(clientURL, principal)
+	if !ok {
+		return nil
+	}
 	parsedClientURL, err := url.Parse(clientURL)
 	if err != nil || parsedClientURL.Scheme != "https" || parsedClientURL.Hostname() == "" {
 		return nil
@@ -130,6 +159,10 @@ func (c TrinoCell) connectionFor(username string) *TrinoConnection {
 		host = c.TLSServerName
 	}
 
+	username := principal
+	if perOrgHost {
+		username = "root"
+	}
 	return &TrinoConnection{Host: host, Port: port, Username: username}
 }
 
@@ -277,7 +310,7 @@ func registerTrinoAPI(r *gin.RouterGroup, api *TrinoAPI) {
 // principalIndex maps Trino principals to org ids, and carries the org
 // rows the handlers annotate with.
 type principalIndex struct {
-	orgByPrincipal map[string]string
+	orgByPrincipal configstore.TrinoPrincipalOwners
 	rows           []configstore.TrinoEnabledOrg
 }
 
@@ -286,7 +319,7 @@ func (a *TrinoAPI) index() (principalIndex, error) {
 	if err != nil {
 		return principalIndex{}, err
 	}
-	idx := principalIndex{orgByPrincipal: make(map[string]string, len(rows)), rows: rows}
+	idx := principalIndex{rows: rows}
 	if a.filterCell {
 		idx.rows = nil
 		for _, row := range rows {
@@ -295,11 +328,7 @@ func (a *TrinoAPI) index() (principalIndex, error) {
 			}
 		}
 	}
-	for _, o := range idx.rows {
-		if p := o.TrinoPrincipal(); p != "" {
-			idx.orgByPrincipal[p] = o.OrgID
-		}
-	}
+	idx.orgByPrincipal = configstore.NewTrinoPrincipalOwners(idx.rows)
 	return idx, nil
 }
 
@@ -327,12 +356,12 @@ func (a *TrinoAPI) liveQueries(ctx context.Context, known ...principalIndex) ([]
 		// The coordinator answered; a config-store blip should degrade the
 		// org column, not the whole live view.
 		slog.Warn("admin: trino org index unavailable, serving unannotated queries", "error", idxErr)
-		return queries, principalIndex{orgByPrincipal: map[string]string{}}, nil
+		return queries, principalIndex{orgByPrincipal: configstore.TrinoPrincipalOwners{}}, nil
 	}
 	out := make([]TrinoQuery, len(queries))
 	copy(out, queries)
 	for i := range out {
-		out[i].Org = idx.orgByPrincipal[out[i].Principal]
+		out[i].Org = idx.orgByPrincipal.OrgID(out[i].Principal)
 	}
 	return out, idx, nil
 }
@@ -496,7 +525,7 @@ func (a *TrinoAPI) handleQueryDetail(c *gin.Context) {
 		return
 	}
 	if idx, idxErr := a.index(); idxErr == nil {
-		q.Org = idx.orgByPrincipal[q.Principal]
+		q.Org = idx.orgByPrincipal.OrgID(q.Principal)
 	}
 	c.JSON(http.StatusOK, q)
 }
@@ -530,7 +559,7 @@ func (a *TrinoAPI) handleKillQuery(c *gin.Context) {
 	targetOrg := ""
 	if q, err := a.client.Query(c.Request.Context(), queryID); err == nil {
 		if idx, idxErr := a.index(); idxErr == nil {
-			targetOrg = idx.orgByPrincipal[q.Principal]
+			targetOrg = idx.orgByPrincipal.OrgID(q.Principal)
 		}
 	}
 

controlplane/admin/trino_test.go

@@ -111,7 +111,8 @@ func trinoTestRouter(api *TrinoAPI, role Role) *gin.Engine {
 func twoOrgTrinoStore() *fakeTrinoOrgStore {
 	return &fakeTrinoOrgStore{
 		orgs: []configstore.TrinoEnabledOrg{
-			{OrgID: "org-a", DatabaseName: "db_a", Tier: "free", CellID: "cell-test", State: configstore.ManagedWarehouseStateReady, RootPasswordHash: "$2a$10$secrethash"},
+			{OrgID: "org-a", DatabaseName: "db_a", Tier: "free", CellID: "cell-test", State: configstore.ManagedWarehouseStateReady, RootPasswordHash: "$2a$10$secrethash",
+				Users: []configstore.TrinoOrgUser{{Username: "analyst", PasswordHash: "$2a$10$analysthash"}}},
 			{OrgID: "org-b", DatabaseName: "db_b", Tier: "scale", CellID: "cell-test", State: configstore.ManagedWarehouseStatePending, RootPasswordHash: "$2a$10$othersecret"},
 		},
 	}
@@ -158,6 +159,10 @@ func TestQueriesAreAnnotatedWithTheOwningOrg(t *testing.T) {
 	coord := &fakeTrinoCoordinator{queries: []TrinoQuery{
 		{QueryID: "q1", State: "RUNNING", Principal: "db_a", ElapsedMS: 100},
 		{QueryID: "q2", State: "RUNNING", Principal: "db_b", ElapsedMS: 200},
+		// A per-user login authenticates as <database_name>.<username>.
+		{QueryID: "q4", State: "RUNNING", Principal: "db_a.analyst", ElapsedMS: 300},
+		// Not in the password file: the org prefix alone must not attribute it.
+		{QueryID: "q5", State: "RUNNING", Principal: "db_a.nobody", ElapsedMS: 400},
 		// A query from a principal that is not a tenant: the provisioner's
 		// own reconcile DDL. It must appear with an EMPTY org rather than
 		// be silently attributed to someone.
@@ -174,7 +179,7 @@ func TestQueriesAreAnnotatedWithTheOwningOrg(t *testing.T) {
 		q := raw.(map[string]any)
 		got[q["query_id"].(string)] = q["org"].(string)
 	}
-	want := map[string]string{"q1": "org-a", "q2": "org-b", "q3": ""}
+	want := map[string]string{"q1": "org-a", "q2": "org-b", "q3": "", "q4": "org-a", "q5": ""}
 	for id, wantOrg := range want {
 		if got[id] != wantOrg {
 			t.Errorf("query %s: org = %q, want %q", id, got[id], wantOrg)
@@ -596,6 +601,51 @@ func TestReadyOrgDetailReturnsTenantClientConnection(t *testing.T) {
 	}
 }
 
+// With a per-org client host, an org connects to Trino at the same
+// <database_name>.<domain> it uses for pgwire, as its own duckgres login:
+// Trino qualifies that login with the org the host names.
+func TestReadyOrgDetailAdvertisesPerOrgClientHost(t *testing.T) {
+	store := &fakeTrinoOrgStore{
+		orgs: []configstore.TrinoEnabledOrg{{OrgID: "org-a", DatabaseName: "tenant-a", CellID: "cell-test", State: configstore.ManagedWarehouseStateReady}},
+		rows: map[string]*configstore.ManagedWarehouseTrino{
+			"org-a": {OrgID: "org-a", Enabled: true, TrinoCellID: "cell-test", State: configstore.ManagedWarehouseStateReady},
+		},
+	}
+	api := NewTrinoAPI(TrinoCell{ID: "cell-test", CoordinatorURL: "https://coordinator.invalid", ClientURL: "https://{database_name}.dw.example.com"}, &fakeTrinoCoordinator{}, store, nil)
+	r := trinoTestRouter(api, RoleViewer)
+
+	code, body := doTrinoJSON(t, r, http.MethodGet, "/api/v1/orgs/org-a/trino", "")
+	if code != http.StatusOK {
+		t.Fatalf("expected 200, got %d", code)
+	}
+	connection := body["status"].(map[string]any)["connection"].(map[string]any)
+	if connection["host"] != "tenant-a.dw.example.com" || connection["port"] != float64(443) || connection["username"] != "root" {
+		t.Errorf("connection = %v, want tenant-a.dw.example.com:443 as root", connection)
+	}
+}
+
+func TestResolveTrinoClientURL(t *testing.T) {
+	for _, tc := range []struct {
+		template, database, want string
+		perOrg, ok               bool
+	}{
+		{"https://trino.example.com", "tenant-a", "https://trino.example.com", false, true},
+		{"https://{database_name}.dw.example.com", "tenant-a", "https://tenant-a.dw.example.com", true, true},
+		{"https://{database_name}.dw.example.com:8443", "tenant-a", "https://tenant-a.dw.example.com:8443", true, true},
+		// A grandfathered database_name that is no DNS label cannot be a host label.
+		{"https://{database_name}.dw.example.com", "db_a", "", true, false},
+		{"https://{database_name}.dw.example.com", "a.b", "", true, false},
+		{"https://gateway.{database_name}.example.com", "tenant-a", "", true, false},
+		{"https://{database_name}.{database_name}.example.com", "tenant-a", "", true, false},
+		{"http://{database_name}.dw.example.com", "tenant-a", "", true, false},
+	} {
+		got, perOrg, ok := ResolveTrinoClientURL(tc.template, tc.database)
+		if got != tc.want || perOrg != tc.perOrg || ok != tc.ok {
+			t.Errorf("ResolveTrinoClientURL(%q, %q) = (%q, %v, %v), want (%q, %v, %v)", tc.template, tc.database, got, perOrg, ok, tc.want, tc.perOrg, tc.ok)
+		}
+	}
+}
+
 func TestTrinoCellAPIAliasPreservesStoredOwnership(t *testing.T) {
 	for _, storedID := range []string{"stored-cell", "another-cell", "legacy", ""} {
 		t.Run("stored="+storedID, func(t *testing.T) {

controlplane/configstore/models.go

@@ -522,6 +522,110 @@ 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
+}
+
+// TrinoPrincipalOwner is the duckgres login a Trino principal authenticated
+// as: the org, and the org user whose password line it matched.
+type TrinoPrincipalOwner struct {
+	OrgID    string
+	Username string
+}
+
+// TrinoPrincipalOwners maps each Trino principal an org projects back to the
+// duckgres login it belongs to. Trino reports only the principal, and every
+// consumer that attributes a query to a tenant (usage metering, the admin
+// console) must see a per-user login as its org's, not as an unknown user.
+//
+// It is built from the SAME principals BuildTrinoAuthFiles projects, by exact
+// match, rather than by splitting on the separator, so a principal that is not
+// in the password file never resolves to an org.
+type TrinoPrincipalOwners map[string]TrinoPrincipalOwner
+
+// NewTrinoPrincipalOwners indexes the projected principals of orgs. The bare
+// org principal authenticates with the root login's hash, so it resolves to
+// root.
+func NewTrinoPrincipalOwners(orgs []TrinoEnabledOrg) TrinoPrincipalOwners {
+	owners := make(TrinoPrincipalOwners, len(orgs))
+	for _, o := range orgs {
+		if p := o.TrinoPrincipal(); p != "" {
+			owners[p] = TrinoPrincipalOwner{OrgID: o.OrgID, Username: "root"}
+		}
+		for _, u := range o.Users {
+			if p := o.TrinoUserPrincipal(u.Username); p != "" {
+				owners[p] = TrinoPrincipalOwner{OrgID: o.OrgID, Username: u.Username}
+			}
+		}
+	}
+	return owners
+}
+
+// OrgID returns the org that owns principal, or "" for an operational or
+// unknown principal.
+func (owners TrinoPrincipalOwners) OrgID(principal string) string {
+	return owners[principal].OrgID
 }
 
 // TrinoPrincipal is the tenant's customer-facing identity in Trino: the