fix(tenant): one canonical tenant-id validator, enforced at every ingress (#385)

crates/persistence/src/backends/mongodb/storage.rs

@@ -1507,6 +1507,11 @@ impl ResourceStorage for MongoBackend {
         id: &str,
         display_name: Option<&str>,
     ) -> StorageResult<crate::core::TenantRecord> {
+        // Backstop for the canonical tenant-id contract (issue #385). MongoDB
+        // matches `tenant_id` as an exact BSON string under the default
+        // (case-sensitive) collation, so it has no derivation to protect — this
+        // keeps the precondition uniform across every implementation.
+        self.ensure_canonical_tenant_id(id)?;
         let db = self.get_database().await?;
         let tenants = db.collection::<Document>(MongoBackend::TENANTS_COLLECTION);
         // RFC 3339 string, matching the SQLite registry's `created_at` format so

crates/persistence/src/backends/postgres/storage.rs

@@ -777,6 +777,11 @@ impl ResourceStorage for PostgresBackend {
         id: &str,
         display_name: Option<&str>,
     ) -> StorageResult<crate::core::TenantRecord> {
+        // Backstop for the canonical tenant-id contract (issue #385). As with
+        // SQLite, PostgreSQL keys tenants by an exact-match, case-sensitive
+        // `tenant_id` column and has no derivation to protect — this keeps the
+        // precondition uniform across every implementation.
+        self.ensure_canonical_tenant_id(id)?;
         let client = self.get_client().await?;
         // Plain INSERT so a duplicate id surfaces as a constraint error; the
         // admin handler pre-checks existence and returns 409, so reaching here

crates/persistence/src/backends/s3/keyspace.rs

@@ -304,11 +304,14 @@ impl S3Keyspace {
     /// `{digest}.json` leaf — and `purge_tenant_data` sweeps only those
     /// sub-prefixes, so it cannot delete a settings object either.
     ///
-    /// Do **not** weaken this to "tenant IDs cannot contain `.`". That is true of
-    /// the routing validators (`is_valid_tenant_id`), but *not* of
-    /// `admin_tenants::validate_tenant_id`, which permits `.` and `/`, nor of the
-    /// JWT tenant extractor, which validates nothing. The name is dotted to keep
-    /// it unroutable, but the safety of this namespace must not depend on that.
+    /// Do **not** weaken this to "tenant IDs cannot contain `.`". That was never
+    /// reliably true, and since issue #385 it is plainly false: the canonical
+    /// charset ([`TenantId::parse`](crate::tenant::TenantId::parse)) permits `.`
+    /// on every ingress, so `_system.user-settings` is a *well-formed* tenant id.
+    /// It is refused only because it is listed in
+    /// [`RESERVED_TENANT_SEGMENTS`](crate::tenant::RESERVED_TENANT_SEGMENTS) —
+    /// and that list is a guardrail, not this namespace's safety proof.
+    ///
     /// In particular, a future change that widened a tenant purge to sweep the
     /// whole tenant prefix would break the structural argument above and must
     /// exclude this namespace explicitly.
@@ -377,9 +380,12 @@ fn sanitize(value: &str) -> String {
 /// tenant is worth more here than fixed-length keys. Ids reaching this point are
 /// length-capped upstream, so S3's 1024-byte key limit is not a concern.
 ///
-/// Only `/` is reachable through the admin API's charset — `\` and space are
-/// already rejected there — so in practice this changes the key of exactly the
-/// hierarchical ids that were previously colliding.
+/// Only `/` is reachable through the canonical charset
+/// ([`TenantId::parse`](crate::tenant::TenantId::parse)) — `\`, space, and `%`
+/// are rejected on every ingress since issue #385 — so in practice this changes
+/// the key of exactly the hierarchical ids that were previously colliding. The
+/// `\`/space/`%` arms are kept anyway: this function's contract is to be
+/// injective for *any* input, so it does not inherit the validator's bounds.
 fn registry_object_id(tenant_id: &str) -> String {
     let mut out = String::with_capacity(tenant_id.len());
     for c in tenant_id.chars() {
@@ -398,6 +404,51 @@ fn registry_object_id(tenant_id: &str) -> String {
 mod tests {
     use super::*;
 
+    /// Demonstrates the collision the canonical validator's per-segment reserved
+    /// list exists to prevent (issue #385), and pins that the validator prevents
+    /// it.
+    ///
+    /// `with_tenant_prefix` does not escape, so tenant `acme/resources` nests its
+    /// data at `acme/resources/resources/…` — *inside* the `acme/resources/`
+    /// prefix that tenant `acme` lists and that `purge_tenant_data("acme")`
+    /// deletes. This is a cross-tenant destructive collision, and it survives
+    /// entirely at the keyspace level: nothing here can tell the two apart.
+    ///
+    /// The fix is upstream — such an id can no longer be constructed — so this
+    /// test asserts both halves: the keyspace really does overlap, and
+    /// `TenantId::parse` really does refuse the id. If someone ever removes the
+    /// reserved segment, the second assertion fails and points at the first.
+    #[test]
+    fn hierarchical_id_naming_a_keyspace_namespace_would_overlap_its_parent() {
+        use crate::tenant::TenantId;
+
+        let base = S3Keyspace::new(None);
+        let parent = base.with_tenant_prefix("acme");
+        let nested = base.with_tenant_prefix("acme/resources");
+
+        // The overlap is real: everything the nested tenant writes sits under the
+        // prefix the parent sweeps.
+        let parent_scan = parent.resources_prefix();
+        assert!(
+            nested.resources_prefix().starts_with(&parent_scan),
+            "expected {} to sit inside {parent_scan}",
+            nested.resources_prefix()
+        );
+        assert!(
+            nested.history_root_prefix().starts_with(&parent_scan),
+            "expected {} to sit inside {parent_scan}",
+            nested.history_root_prefix()
+        );
+
+        // Which is why the id is unconstructible.
+        assert!(
+            TenantId::parse("acme/resources").is_err(),
+            "the reserved segment is what keeps the overlap unreachable"
+        );
+        // The parent itself stays perfectly valid.
+        assert!(TenantId::parse("acme").is_ok());
+    }
+
     /// The registry key establishes tenant identity: ids that a lossy sanitiser
     /// would collapse together must map to distinct objects. A collision here
     /// lets one tenant read, overwrite, and deregister another's record.

crates/persistence/src/backends/s3/storage.rs

@@ -1057,6 +1057,13 @@ impl ResourceStorage for S3Backend {
         id: &str,
         display_name: Option<&str>,
     ) -> StorageResult<crate::core::TenantRecord> {
+        // Backstop for the canonical tenant-id contract (issue #385). S3 is the
+        // backend with the most to lose here: `with_tenant_prefix` gives `/`
+        // structural meaning, so tenant `a/resources` would store under
+        // `a/resources/…` — inside the prefix tenant `a` lists and
+        // `purge_tenant_data("a")` deletes. `TenantId::parse` rejects the
+        // reserved segment, making that shape unconstructible.
+        self.ensure_canonical_tenant_id(id)?;
         let location = self
             .registry_location()
             .ok_or_else(|| self.tenant_registry_unsupported())?;

crates/persistence/src/backends/sqlite/storage.rs

@@ -834,6 +834,12 @@ impl ResourceStorage for SqliteBackend {
         id: &str,
         display_name: Option<&str>,
     ) -> StorageResult<crate::core::TenantRecord> {
+        // Backstop for the canonical tenant-id contract (issue #385). SQLite
+        // keys tenants by an exact-match `tenant_id` column, so it has no
+        // collision of its own to defend against — this guards the *registry*
+        // from minting an id the other backends could not keep distinct, and
+        // keeps the precondition uniform across every implementation.
+        self.ensure_canonical_tenant_id(id)?;
         let conn = self.get_connection()?;
         // Plain INSERT so a duplicate id surfaces as a constraint error; the
         // admin handler pre-checks existence and returns 409, so reaching here

crates/persistence/src/composite/storage.rs

@@ -1037,6 +1037,12 @@ impl ResourceStorage for CompositeStorage {
         id: &str,
         display_name: Option<&str>,
     ) -> StorageResult<crate::core::TenantRecord> {
+        // Checked here as well as in the primary (issue #385). The composite is
+        // what a search-backed deployment actually calls, and its secondaries —
+        // Elasticsearch above all — derive index names from the tenant id. The
+        // primary's own check would still catch this, but only after the
+        // composite had already accepted the id as legitimate.
+        self.ensure_canonical_tenant_id(id)?;
         self.primary.register_tenant(id, display_name).await
     }
 

crates/persistence/src/core/storage.rs

@@ -778,9 +778,36 @@ pub trait ResourceStorage: Send + Sync {
         Ok(None)
     }
 
+    /// Rejects a tenant id that does not satisfy the canonical definition.
+    ///
+    /// Every implementation of [`register_tenant`](Self::register_tenant) calls
+    /// this first. It is a **backstop**, not the primary control: ids are
+    /// validated at each REST ingress (header, URL prefix, JWT claim, admin
+    /// body). Its job is to make the storage-layer precondition documented in
+    /// [`crate::tenant`] true by construction, so a future ingress — or an
+    /// embedder using this crate directly — cannot mint a tenant that skipped
+    /// validation. Issue #385.
+    ///
+    /// Deliberately **not** applied to `deregister_tenant`, `purge_tenant_data`,
+    /// or any read path. Those are how an operator cleans up a non-canonical
+    /// tenant that predates this validator; rejecting there would make bad data
+    /// permanently undeletable. Only the mint point is guarded.
+    fn ensure_canonical_tenant_id(&self, id: &str) -> StorageResult<()> {
+        crate::tenant::TenantId::parse(id).map(|_| ()).map_err(|e| {
+            StorageError::Tenant(crate::error::TenantError::NonCanonicalTenantId {
+                tenant_id: id.to_string(),
+                reason: e,
+            })
+        })
+    }
+
     /// Registers (provisions) a new tenant, stamping `created_at` with the
     /// current time. Returns [`StorageError`] if the tenant is already
     /// registered. Default: unsupported-capability error.
+    ///
+    /// Implementations must call
+    /// [`ensure_canonical_tenant_id`](Self::ensure_canonical_tenant_id) before
+    /// writing anything.
     async fn register_tenant(
         &self,
         _id: &str,

crates/persistence/src/error.rs

@@ -8,7 +8,7 @@ use std::fmt;
 
 use thiserror::Error;
 
-use crate::tenant::TenantId;
+use crate::tenant::{TenantId, TenantIdError};
 
 /// The primary error type for all storage operations.
 ///
@@ -168,6 +168,19 @@ pub enum TenantError {
         tenant_id: TenantId,
     },
 
+    /// A tenant id offered for provisioning is not canonical.
+    ///
+    /// Distinct from [`InvalidTenant`](Self::InvalidTenant) because it carries
+    /// the *reason*: this reaches an operator through the admin API, where "the
+    /// id is 71 bytes" is actionable and "invalid tenant" is not.
+    #[error("tenant id '{tenant_id}' is not valid: {reason}")]
+    NonCanonicalTenantId {
+        /// The rejected identifier, as supplied.
+        tenant_id: String,
+        /// Why [`TenantId::parse`](crate::tenant::TenantId::parse) rejected it.
+        reason: TenantIdError,
+    },
+
     /// Tenant is suspended and cannot perform operations.
     #[error("tenant suspended: {tenant_id}")]
     TenantSuspended {