fix(s3,provisioning): close Red re-review findings — control-plane forge + injective org slug

Red re-review (0 critical, 1 high, 1 med, 1 low): the s3 data-plane HIGH was
confirmed CLOSED, but Red found the same forge open on the provisioning control
plane (worse: destroy DB + credential exfil), proved my org-fold dispute WRONG
with a reachable cross-tenant collision, and asked to lock the fix's dependency.

- [HIGH] provisioning.tenant() now requires ctx.User() (provisioning.go:454) —
  same gate as the s3 fix. Without it, an in-cluster caller could forge
  'X-Org-Id: victim' with NO bearer and POST /v1/sql (allocate a DB in the
  victim's namespace + receive its connection string + password), DELETE
  /v1/sql/:name (destroy the victim's DB), or enumerate resources. SanitizeIdentity
  restores a forged X-Org-Id on the no-principal Phase-1 path but strips X-User-Id;
  gating on it refuses only the anonymous forge. Test:
  TestForgedOrgWithoutPrincipalRefused (forged org + no principal -> 403 across
  POST/DELETE/GET, provisioner never runs).
- [MED, dispute WITHDRAWN — Red was right] provisioning.sanitizeOrg is now
  INJECTIVE (provisioning.go:468): identity on a clean [a-z0-9-] slug, else the
  fold + '-'+16hex SHA-256(raw owner) — mirroring iam/object/orgdb.go:orgSlug.
  The old lossy fold collapsed 'Acme'/'acme' and 'team.a'/'team-a' onto one slug,
  and since the whole tenant->bucket/DB namespace hashes THAT slug, two distinct
  orgs shared one physical namespace (reachable: the IAM org name is a varchar
  with no shape validator, so a fold-sibling is registerable + mints a valid
  token). Tests: TestSanitizeOrgInjective (the exact collisions no longer collide,
  incl. derived orgHash) + updated TestSanitizeOrg.
- [LOW] locked the s3/provisioning fixes' cross-file dependency:
  TestSanitizeIdentity_AnonPathHasNoUserId asserts a client-forged X-User-Id does
  NOT survive the anon path (ctx.User()=="") while X-Org-Id does — so a future
  refactor that restored X-User-Id fails this test first.

All fold consumers (orgHash -> SQL/KV/CH/S3 physical namespaces) inherit the
injective slug through the ONE sanitizeOrg. go test green (provisioning + s3 +
s3admin + root identity); cmd/cloud builds; gofmt/vet clean; zero go.sum drift.

clients/provisioning/billing_test.go

@@ -96,6 +96,7 @@ func postCreate(t *testing.T, s *svc, kind, org, name string) *http.Response {
 	req.Header.Set("Content-Type", "application/json")
 	if org != "" {
 		req.Header.Set("X-Org-Id", org)
+		req.Header.Set("X-User-Id", "u-"+org) // validated principal (tenant() gates on X-User-Id)
 	}
 	resp, err := app.Fiber().Test(req)
 	if err != nil {

clients/provisioning/provisioning.go

@@ -445,13 +445,27 @@ func (s *svc) drop(kind string) zip.Handler {
 // tenant resolves the org for a request. Empty org is allowed only for admins,
 // who are bucketed under the literal "admin" org.
 //
-// Trusting the gateway-minted X-User-IsAdmin claim (c.IsAdmin()) is acceptable
-// here: the blast radius of a forged claim is bounded to the single literal
-// "admin" org bucket. An admin still gets a distinct physical namespace
-// ("o"<hash("admin")>_…) and cannot name into any real tenant's resources. The
-// gateway strips client-supplied identity headers and only sets this claim on
-// the JWT-validated path (HIP-0026), so it cannot be spoofed from the edge.
+// REQUIRES A VALIDATED PRINCIPAL (RED HIGH). SanitizeIdentity strips X-User-Id on
+// ingress and re-sets it ONLY for a validated bearer/cookie; on the no-principal
+// "Phase-1 data path" it RESTORES the client's raw X-Org-Id but leaves X-User-Id
+// empty. This control plane ALLOCATES and DESTROYS real backend resources and
+// returns generated credentials, so trusting X-Org-Id alone let an in-cluster
+// caller (a co-namespace pod within the cloud-api NetworkPolicy) forge
+// `X-Org-Id: victim` with NO bearer and provision a DB in the victim's namespace
+// (receiving its connection string + password), destroy the victim's database, or
+// enumerate its resources — strictly worse than a data read. Gating on c.User()
+// (X-User-Id) refuses ONLY that anonymous-forge path: every legitimate caller
+// arrives through the console/gateway with a validated principal, so no real
+// client breaks.
+//
+// The X-User-IsAdmin claim is likewise only trustworthy under a validated
+// principal — SanitizeIdentity sets it only for a JWT-verified global admin
+// (HIP-0026) — and even then reaches only the literal "admin" org's own physical
+// namespace, never a real tenant's.
 func tenant(c *zip.Ctx) (string, bool) {
+	if c.User() == "" {
+		return "", false // no validated principal — refuse the forgeable data path
+	}
 	org := sanitizeOrg(c.Org())
 	if org != "" {
 		return org, true
@@ -462,13 +476,39 @@ func tenant(c *zip.Ctx) (string, bool) {
 	return "", false
 }
 
-// sanitizeOrg reduces a gateway org id to a lowercase [a-z0-9-] slug, capped at
-// 32 chars. Defense in depth: org comes from the JWT via the gateway, but it
-// still flows into physical identifiers.
+// sanitizeOrg reduces a gateway org id to a lowercase [a-z0-9-] slug that is
+// INJECTIVE in the raw owner: it is the identity on an owner already shaped like
+// a DNS-1123 label, otherwise the folded slug is disambiguated with "-" + the
+// first 16 hex of SHA-256(raw owner). Without the suffix the fold was lossy —
+// ToLower + every non-[a-z0-9-]→"-" + a 32-char truncation collapse distinct
+// owners (`Acme`/`acme`, `team.a`/`team-a`) onto one slug, and since the whole
+// tenant→bucket/DB namespace hashes THIS slug (orgHash, physicalName), that was a
+// cross-tenant collision: two orgs sharing one physical namespace (RED MED, proven
+// reachable because the IAM org name is a varchar with no shape validator, so a
+// fold-sibling of a target org is registerable and mints a valid token). Uses the
+// SAME disambiguation STRATEGY as iam/object/orgdb.go:orgSlug (fold + SHA-256
+// suffix) — not byte-identical (this caps the identity branch at 32 chars and
+// truncates the fold, since the slug keys cloud's own bucket/DB names, which stay
+// consistent across allocate + operate; IAM's orgSlug keys only its per-org
+// SQLite files). The org still flows into physical identifiers, so this is the
+// isolation boundary.
+//
+// The identity fast-path is withheld from a clean slug that ITSELF looks like a
+// suffixed output (`<label>-<16 lowercase hex>`): such a slug is ambiguous with a
+// folded owner's disambiguation, so it too is re-suffixed — otherwise a squatted
+// org literally named "foo-<sha256(Foo)[:8]>" would alias non-slug owner "Foo".
+// Real org names essentially never end in "-"+16-hex, so this shifts nothing real
+// while closing the alias class completely (every non-identity output carries a
+// hash of its OWN raw bytes, so two distinct raws collide only on a full 64-bit
+// SHA-256 prefix collision).
 func sanitizeOrg(s string) string {
-	s = strings.ToLower(strings.TrimSpace(s))
+	raw := strings.TrimSpace(s)
+	lower := strings.ToLower(raw)
+	if isDNSLabel(lower) && lower == raw && len(lower) <= 32 && !looksSuffixed(lower) {
+		return lower // already a clean, unambiguous slug: identity, no suffix needed
+	}
 	var b strings.Builder
-	for _, r := range s {
+	for _, r := range lower {
 		switch {
 		case r >= 'a' && r <= 'z', r >= '0' && r <= '9', r == '-':
 			b.WriteRune(r)
@@ -476,11 +516,53 @@ func sanitizeOrg(s string) string {
 			b.WriteRune('-')
 		}
 	}
-	out := strings.Trim(b.String(), "-")
-	if len(out) > 32 {
-		out = strings.Trim(out[:32], "-")
+	folded := strings.Trim(b.String(), "-")
+	if len(folded) > 32 {
+		folded = strings.Trim(folded[:32], "-")
 	}
-	return out
+	// Disambiguate with a 64-bit hash of the RAW owner so distinct owners that
+	// fold together stay distinct. Empty raw → "" (an unresolvable org; the caller
+	// gates on non-empty). The suffix is derived from raw, not the fold, so
+	// collision would need a SHA-256 collision on the exact owner bytes.
+	if raw == "" {
+		return ""
+	}
+	sum := sha256.Sum256([]byte(raw))
+	return folded + "-" + hex.EncodeToString(sum[:8])
+}
+
+// isDNSLabel reports whether s is a non-empty [a-z0-9-] string that is not "."
+// or "..". Such an owner needs no disambiguation (sanitizeOrg is the identity on
+// it), matching iam/object/orgdb.go's validateOrgSlug.
+func isDNSLabel(s string) bool {
+	if s == "" || s == "." || s == ".." {
+		return false
+	}
+	for _, r := range s {
+		switch {
+		case r >= 'a' && r <= 'z', r >= '0' && r <= '9', r == '-':
+		default:
+			return false
+		}
+	}
+	return true
+}
+
+// looksSuffixed reports whether s ends in "-" + exactly 16 lowercase-hex chars —
+// the shape of sanitizeOrg's own disambiguation suffix. A clean slug of this
+// shape is denied the identity fast-path (and re-suffixed) so it can never alias
+// a folded non-slug owner's output. This is the ONLY collision class the identity
+// fast-path could otherwise admit.
+func looksSuffixed(s string) bool {
+	if len(s) < 17 || s[len(s)-17] != '-' {
+		return false
+	}
+	for _, r := range s[len(s)-16:] {
+		if !((r >= '0' && r <= '9') || (r >= 'a' && r <= 'f')) {
+			return false
+		}
+	}
+	return true
 }
 
 // orgHash returns a fixed-width, collision-resistant tag for an org slug: the

clients/provisioning/provisioning_test.go

@@ -148,7 +148,7 @@ func TestStore_PhysicalNameConflict(t *testing.T) {
 	// Distinct identity (different name + id), but force a physical collision.
 	b := sampleResource("orders")
 	b.ID, b.Name = "rs_other", "events" // satisfies UNIQUE(org,kind,name)
-	b.PhysicalName = a.PhysicalName      // but collides on physical_name
+	b.PhysicalName = a.PhysicalName     // but collides on physical_name
 	if exists, err := s.PhysicalExists(ctx, b.PhysicalName); err != nil || !exists {
 		t.Fatalf("PhysicalExists = (%v,%v), want (true,nil)", exists, err)
 	}
@@ -172,20 +172,85 @@ func TestNameValidation(t *testing.T) {
 	}
 }
 
+// TestSanitizeOrg: a clean DNS-1123-label owner maps to itself (identity, no
+// suffix); a non-slug owner folds to [a-z0-9-] AND carries a 16-hex SHA-256
+// disambiguation suffix. The exported IAM owner claim is already a clean slug,
+// so the common path is the identity — the suffix exists only to keep distinct
+// non-slug owners distinct.
 func TestSanitizeOrg(t *testing.T) {
-	cases := map[string]string{
-		"acme":                  "acme",
-		"Acme Corp":             "acme-corp",
-		" hanzo ":               "hanzo",
-		"a@b.c":                 "a-b-c",
-		"--weird--":             "weird",
-		strings.Repeat("z", 50): strings.Repeat("z", 32),
+	// Clean slugs (already [a-z0-9-] after trim): identity, no suffix. Trimming
+	// surrounding whitespace is not a meaningful distinction, so " hanzo " and
+	// "hanzo" intentionally coincide (both trim to the same clean slug).
+	identity := map[string]string{
+		"acme": "acme", "hanzo": "hanzo", "my-org": "my-org", "org123": "org123",
+		"a-b-c-d": "a-b-c-d", " hanzo ": "hanzo", "--weird--": "--weird--",
 	}
-	for in, want := range cases {
+	for in, want := range identity {
 		if got := sanitizeOrg(in); got != want {
-			t.Errorf("sanitizeOrg(%q) = %q, want %q", in, got, want)
+			t.Errorf("sanitizeOrg(%q) = %q, want identity %q", in, got, want)
 		}
 	}
+	// Owners with chars OUTSIDE [a-z0-9-] (uppercase, '.', '@', space) fold + get a
+	// "-"+16hex suffix, so the result is NOT the bare fold (that bare fold is the
+	// collision target) and stays a DNS-safe slug.
+	for _, in := range []string{"Acme Corp", "a@b.c", "team.a", "Widgets"} {
+		got := sanitizeOrg(in)
+		if len(got) < 17 || got[len(got)-17] != '-' {
+			t.Errorf("sanitizeOrg(%q) = %q, want a folded slug + '-'+16hex suffix", in, got)
+		}
+		for _, r := range got {
+			if !(r >= 'a' && r <= 'z' || r >= '0' && r <= '9' || r == '-') {
+				t.Errorf("sanitizeOrg(%q) = %q has unsafe char %q", in, got, r)
+			}
+		}
+	}
+}
+
+// TestSanitizeOrgInjective is the RED cross-tenant regression: distinct raw
+// owners that USED to fold to the same slug (and thus shared one physical
+// bucket/DB namespace) must now map to DIFFERENT slugs. Proven for the exact
+// collisions RED found: Acme/acme (case) and team.a/team-a (separator).
+func TestSanitizeOrgInjective(t *testing.T) {
+	collisionPairs := [][2]string{
+		{"Acme", "acme"},
+		{"team.a", "team-a"},
+		{"a.b", "a-b"},
+		{"Foo_Bar", "foo-bar"},
+		{"WIDGETS", "widgets"},
+	}
+	for _, p := range collisionPairs {
+		x, y := sanitizeOrg(p[0]), sanitizeOrg(p[1])
+		if x == y {
+			t.Errorf("sanitizeOrg fold collision: %q and %q both → %q (cross-tenant namespace share!)", p[0], p[1], x)
+		}
+		// And the derived physical namespace (what actually keys buckets/DBs) is
+		// distinct too — the property that matters for isolation.
+		if orgHash(x) == orgHash(y) {
+			t.Errorf("orgHash collision after sanitize: %q/%q share a physical namespace", p[0], p[1])
+		}
+	}
+	// A clean lowercase slug is unaffected (no false-splitting of legit owners).
+	if sanitizeOrg("acme") != "acme" {
+		t.Error("a clean slug must remain identity")
+	}
+
+	// ALIAS class: a clean slug that LOOKS like a suffixed output
+	// ("foo-<16hex>") must NOT alias the sanitized output of a non-slug owner that
+	// folds to "foo" and hashes to that suffix. The suffixed-looking slug is denied
+	// identity and re-suffixed, so a squatted "foo-<sha256(Foo)[:8]>" can never
+	// collide with "Foo".
+	nonSlug := "Foo"
+	foldedOut := sanitizeOrg(nonSlug) // "foo-<hash(Foo)[:8]>"
+	if sanitizeOrg(foldedOut) == foldedOut {
+		t.Errorf("suffix-looking slug %q kept identity — aliases the non-slug output of %q", foldedOut, nonSlug)
+	}
+	if sanitizeOrg(foldedOut) == sanitizeOrg(nonSlug) {
+		t.Errorf("alias collision: squatted %q and non-slug %q map to the same slug", foldedOut, nonSlug)
+	}
+	// A normal slug that does NOT look suffixed keeps identity.
+	if sanitizeOrg("my-team-42") != "my-team-42" {
+		t.Error("a normal slug must keep identity (not over-suffixed)")
+	}
 }
 
 // TestPhysicalNameInjective is the regression test for the cross-tenant
@@ -265,6 +330,48 @@ func TestCreateOrgGate(t *testing.T) {
 	}
 }
 
+// TestForgedOrgWithoutPrincipalRefused (RED HIGH): a caller that forges
+// X-Org-Id but presents NO validated principal (no X-User-Id) — exactly the
+// SanitizeIdentity "Phase-1 data path" residual an in-cluster pod could send with
+// no bearer — must be refused 403 and provision NOTHING. Without the ctx.User()
+// gate this forged request would allocate a DB in the victim's namespace and
+// return its connection string + generated password, or (on DELETE) destroy it.
+func TestForgedOrgWithoutPrincipalRefused(t *testing.T) {
+	s, mp := newTestSvc(t, "sql")
+	app := zip.New(zip.Config{DisableStartupMessage: true})
+	app.Post("/v1/sql", s.create("sql"))
+	app.Delete("/v1/sql/:name", s.drop("sql"))
+	app.Get("/v1/sql", s.list("sql"))
+
+	for _, tc := range []struct{ method, path, body string }{
+		{"POST", "/v1/sql", `{"name":"orders"}`},
+		{"DELETE", "/v1/sql/orders", ""},
+		{"GET", "/v1/sql", ""},
+	} {
+		var rdr io.Reader
+		if tc.body != "" {
+			rdr = strings.NewReader(tc.body)
+		}
+		req, _ := http.NewRequest(tc.method, tc.path, rdr)
+		if tc.body != "" {
+			req.Header.Set("Content-Type", "application/json")
+		}
+		// Forge the victim's org WITHOUT any validated principal (no X-User-Id).
+		req.Header.Set("X-Org-Id", "victim-org")
+		resp, err := app.Fiber().Test(req)
+		if err != nil {
+			t.Fatalf("%s %s: %v", tc.method, tc.path, err)
+		}
+		if resp.StatusCode != http.StatusForbidden {
+			body, _ := io.ReadAll(resp.Body)
+			t.Errorf("%s %s (forged org, no principal) = %d body=%s, want 403 — control-plane forge NOT closed!", tc.method, tc.path, resp.StatusCode, body)
+		}
+	}
+	if mp.created != 0 || mp.dropped != 0 {
+		t.Fatalf("provisioner ran (created=%d dropped=%d) for forged requests, want 0/0 — resources touched in victim's namespace!", mp.created, mp.dropped)
+	}
+}
+
 // TestCreateKMSDegradePersistsNoPlaintext: with KMS unconfigured, create returns
 // the generated password ONCE and persists NO plaintext (stored row carries an
 // empty secret_ref and the password appears in no stored column).
@@ -279,6 +386,7 @@ func TestCreateKMSDegradePersistsNoPlaintext(t *testing.T) {
 	req, _ := http.NewRequest("POST", "/v1/kv", strings.NewReader(`{"name":"cache"}`))
 	req.Header.Set("Content-Type", "application/json")
 	req.Header.Set("X-Org-Id", "acme")
+	req.Header.Set("X-User-Id", "u-acme") // validated principal (tenant() gates on X-User-Id)
 	resp, err := app.Fiber().Test(req)
 	if err != nil {
 		t.Fatalf("Test: %v", err)

middleware_identity_test.go

@@ -264,6 +264,29 @@ func TestSanitizeIdentity_NilValidatorStillStripsAdmin(t *testing.T) {
 	}
 }
 
+// TestSanitizeIdentity_AnonPathHasNoUserId locks the invariant the s3 + provisioning
+// data-plane fixes depend on (RED): on the no-principal path, SanitizeIdentity
+// RESTORES a forged X-Org-Id (Phase-1 passthrough) but a forged X-User-Id does
+// NOT survive — X-User-Id is in authorityHeaders and only re-set for a validated
+// principal. So ctx.User()=="" is the reliable "no validated principal" signal
+// those subsystems gate on. If a refactor ever restored X-User-Id here, the
+// gate would silently reopen — this test fails first.
+func TestSanitizeIdentity_AnonPathHasNoUserId(t *testing.T) {
+	// nil validator = the no-principal path for ANY request (no JWKS wired), which
+	// is the same header-restore branch a bad/absent bearer takes.
+	app, got := newIdentityApp(t, nil)
+	probe(t, app, func(r *http.Request) {
+		r.Header.Set("X-Org-Id", "victim")
+		r.Header.Set("X-User-Id", "forged-user") // client-supplied — must be stripped
+	})
+	if got.user != "" {
+		t.Fatalf("User() = %q on the anon path, want \"\" — a client-forged X-User-Id survived, reopening the data-plane forge!", got.user)
+	}
+	if got.org != "victim" {
+		t.Errorf("Org() = %q, want %q (Phase-1 org passthrough is intentional; the fix is that User() is empty)", got.org, "victim")
+	}
+}
+
 // Focused validator unit tests: issuer, audience, expiry, signature, and a
 // missing issuer are all enforced.
 func TestIdentityValidator(t *testing.T) {