fix(events): check stream identity against the stored account

CHANGELOG.md

@@ -18,6 +18,14 @@ is pre-1.0; breaking changes bump the **minor** version per
   schema replaces the built-in one, as at startup. JSON and plain output keep
   their fields, and `schema_count` still counts only the project's schemas;
   the human summary also reports the system schemas.
+- Entity event streams accept password and OAuth login tokens. Since 0.48.0 the
+  live identity check looked accounts up by the token subject (`user:<username>`)
+  rather than the stored username, so every tenant member was refused with 403
+  `Stream authorization is no longer valid.`, and streams of accounts without a
+  tenant chain stayed open after deactivation or role changes. Platform
+  administrators who hold tenant memberships are now checked without tenant
+  scope, matching the tenant-scope middleware. Streams still close and stay
+  refused after deactivation, role changes, or membership removal.
 
 ## [0.49.0] - 2026-10-06
 

crates/schema-forge-acton/src/middleware/tenant_scope.rs

@@ -106,7 +106,7 @@ where
         return next.run(request.into()).await;
     };
 
-    if claims.has_role(PLATFORM_ADMIN_ROLE) {
+    if !is_tenant_scoped(&claims) {
         // Platform admin bypasses tenancy. Pass claims through unmodified.
         return next.run(request.into()).await;
     }
@@ -248,21 +248,37 @@ fn select_active_tenant<'a, B>(
         .ok_or(TenantScopeRefusal::NotInMemberships)
 }
 
-/// Add only the selected membership's role, without changing global account roles.
+/// Whether this middleware scopes the caller to an active tenant.
+///
+/// Platform administrators bypass tenancy: their claims reach the handlers
+/// exactly as login minted them, so their `tenant_chain` is the flat
+/// membership set rather than an active-tenant walk, and no membership role
+/// is projected. Anything that re-derives the per-request claims (the live
+/// event-stream identity check) must apply the same rule.
+pub(crate) fn is_tenant_scoped(claims: &Claims) -> bool {
+    !claims.has_role(PLATFORM_ADMIN_ROLE)
+}
+
+/// Scoped roles the selected membership contributes to the request.
 /// The reserved platform administrator role cannot originate in a membership.
+pub(crate) fn active_membership_roles<'a>(
+    tenant_roles: &'a [schema_forge_backend::user_store::TenantRole],
+    active: &'a TenantRef,
+) -> impl Iterator<Item = &'a str> + 'a {
+    tenant_roles
+        .iter()
+        .filter(move |membership| membership.tenant == *active)
+        .map(|membership| membership.role.as_str())
+        .filter(|role| !role.is_empty() && *role != PLATFORM_ADMIN_ROLE)
+}
+
+/// Add only the selected membership's role, without changing global account roles.
 pub(crate) fn add_active_membership_role(claims: &mut Claims, active: &TenantRef) {
-    let roles: Vec<schema_forge_backend::user_store::TenantRole> =
+    let tenant_roles: Vec<schema_forge_backend::user_store::TenantRole> =
         claims.custom_claim_as("tenant_roles").unwrap_or_default();
-    for membership in roles
-        .iter()
-        .filter(|membership| membership.tenant == *active)
-    {
-        let role = &membership.role;
-        if !role.is_empty()
-            && role != PLATFORM_ADMIN_ROLE
-            && !claims.roles.iter().any(|existing| existing == role)
-        {
-            claims.roles.push(role.clone());
+    for role in active_membership_roles(&tenant_roles, active) {
+        if !claims.roles.iter().any(|existing| existing == role) {
+            claims.roles.push(role.to_owned());
         }
     }
 }

crates/schema-forge-acton/src/routes/auth.rs

@@ -371,6 +371,13 @@ pub(crate) async fn emit_login_result(
     }
 }
 
+/// The stored account a token speaks for. Login mints `sub = "user:<username>"`,
+/// while the auth store, the Cedar principal, and ownership all key on the bare
+/// username. Refresh, `/auth/me`, and entity streams re-read the account here.
+pub(crate) fn account_username(claims: &Claims) -> &str {
+    crate::authz::adapters::user_id_from_sub(&claims.sub)
+}
+
 /// `POST /auth/refresh` — exchange a still-valid bearer token for a fresh
 /// one with a new 1-hour expiry.
 ///
@@ -392,10 +399,7 @@ pub async fn refresh(
         return unauthorized_response();
     };
 
-    let username = claims
-        .username
-        .clone()
-        .unwrap_or_else(|| claims.sub.trim_start_matches("user:").to_string());
+    let username = account_username(&claims).to_owned();
 
     // Re-read the User row on every refresh — no claim copy-forward. A row
     // mutated since the original login (e.g., role change, client_org
@@ -489,10 +493,7 @@ pub async fn me(
         return unauthorized_response();
     };
 
-    let username = claims
-        .username
-        .clone()
-        .unwrap_or_else(|| claims.sub.trim_start_matches("user:").to_string());
+    let username = account_username(&claims).to_owned();
 
     // Re-read live state (roles/active/memberships), same contract as refresh:
     // a grant, revocation, or deactivation since login takes effect here.
@@ -842,6 +843,21 @@ pub fn auth_routes_with_password_login(
 mod tests {
     use super::*;
 
+    #[test]
+    fn account_username_maps_the_login_subject_to_the_stored_account() {
+        let claims = |sub: &str| -> Claims {
+            serde_json::from_value(serde_json::json!({
+                "sub": sub, "roles": [], "perms": [], "exp": 9_999_999_999_u64
+            }))
+            .unwrap()
+        };
+        assert_eq!(
+            account_username(&claims("user:solo@example.gov")),
+            "solo@example.gov"
+        );
+        assert_eq!(account_username(&claims("alice")), "alice");
+    }
+
     #[test]
     fn build_login_claims_sets_subject_and_roles() {
         let mappings = PrincipalClaimMappings::default();

crates/schema-forge-acton/src/routes/events.rs

@@ -1,14 +1,16 @@
 //! Authenticated change streams with live identity checks and canonical read projection.
 use super::{
+    auth::account_username,
     entities,
     query_params::{parse_filter_key, parse_filter_params, FilterOp},
 };
 use crate::{
-    access::{check_schema_access, AccessAction, OptionalClaims, PLATFORM_ADMIN_ROLE},
+    access::{check_schema_access, AccessAction, OptionalClaims},
     config::SchemaForgeConfig,
     error::ForgeError,
     events::{self, CommittedSnapshot, EventsRuntime, Subscribe, Subscription},
     messages::{GetSchema, GetTenantConfig, ReplyChannel},
+    middleware::tenant_scope,
     ForgeActor,
 };
 use acton_service::{
@@ -23,12 +25,17 @@ use axum::{
     response::{IntoResponse, Response},
 };
 use futures::stream;
-use schema_forge_backend::{Entity, TenantRef};
+use schema_forge_backend::{user_store::TenantRole, Entity, TenantRef};
 use schema_forge_core::{
     query::Filter,
     types::{DynamicValue, SchemaDefinition},
 };
-use std::{collections::HashMap, convert::Infallible, sync::Arc, time::Duration};
+use std::{
+    collections::{BTreeSet, HashMap},
+    convert::Infallible,
+    sync::Arc,
+    time::Duration,
+};
 use tokio::sync::oneshot;
 
 fn invalid_filter() -> ForgeError {
@@ -108,59 +115,58 @@ struct StreamState {
     done: bool,
     ticker: tokio::time::Interval,
 }
+/// Roles the request claims must still carry: the account's current global
+/// roles plus, for a tenant-scoped caller, the active membership's current
+/// scoped role, exactly as `tenant_scope` projects them.
+fn expected_roles<'a>(
+    global: &'a [String],
+    tenant_roles: &'a [TenantRole],
+    active: Option<&'a TenantRef>,
+) -> BTreeSet<&'a str> {
+    let mut roles: BTreeSet<&str> = global.iter().map(String::as_str).collect();
+    if let Some(active) = active {
+        roles.extend(tenant_scope::active_membership_roles(tenant_roles, active));
+    }
+    roles
+}
 impl StreamState {
+    /// The active tenant the middleware scoped this request to, if any.
+    /// Platform administrators bypass tenant scope, so their `tenant_chain`
+    /// is the login-time membership set and names no active tenant.
+    fn active_tenant(&self) -> Option<&TenantRef> {
+        self.effective_chain
+            .last()
+            .filter(|_| tenant_scope::is_tenant_scoped(&self.claims))
+    }
     async fn identity_valid(&self) -> bool {
         if self.claims.exp <= chrono::Utc::now().timestamp() {
             return false;
         }
-        match self.runtime.auth_store.get_user(&self.claims.sub).await {
-            Ok(Some(user)) if user.active => {
-                let mut expected = self.claims.clone();
-                expected.roles = user.roles;
-                if let Some(active) = self.effective_chain.last() {
-                    let Ok(roles) = self
-                        .runtime
-                        .auth_store
-                        .list_tenant_roles(&self.claims.sub)
-                        .await
-                    else {
-                        return false;
-                    };
-                    let Ok(value) = serde_json::to_value(roles) else {
-                        return false;
-                    };
-                    expected.custom.insert("tenant_roles".into(), value);
-                    crate::middleware::tenant_scope::add_active_membership_role(
-                        &mut expected,
-                        active,
-                    );
-                }
-                if expected
-                    .roles
-                    .iter()
-                    .collect::<std::collections::BTreeSet<_>>()
-                    != self
-                        .claims
-                        .roles
-                        .iter()
-                        .collect::<std::collections::BTreeSet<_>>()
-                {
-                    return false;
-                }
-            }
+        let username = account_username(&self.claims);
+        let user = match self.runtime.auth_store.get_user(username).await {
+            Ok(Some(user)) if user.active => user,
             Ok(None) if !self.local_account && self.effective_chain.is_empty() => return true,
             _ => return false,
+        };
+        let active = self.active_tenant();
+        let tenant_roles = match active {
+            Some(_) => match self.runtime.auth_store.list_tenant_roles(username).await {
+                Ok(roles) => roles,
+                Err(_) => return false,
+            },
+            None => Vec::new(),
+        };
+        let current: BTreeSet<&str> = self.claims.roles.iter().map(String::as_str).collect();
+        if expected_roles(&user.roles, &tenant_roles, active) != current {
+            return false;
         }
-        let Some(leaf) = self.effective_chain.last() else {
+        let Some(leaf) = active else {
             return true;
         };
-        if self.claims.has_role(PLATFORM_ADMIN_ROLE) {
-            return true;
-        }
         let Ok(memberships) = self
             .runtime
             .auth_store
-            .list_tenant_memberships(&self.claims.sub)
+            .list_tenant_memberships(username)
             .await
         else {
             return false;
@@ -183,15 +189,9 @@ impl StreamState {
         let Some(config) = config else {
             return false;
         };
-        match crate::middleware::tenant_scope::walk_to_root(
-            leaf,
-            &config,
-            self.runtime.entity_store.as_ref(),
-        )
-        .await
-        {
+        match tenant_scope::walk_to_root(leaf, &config, self.runtime.entity_store.as_ref()).await {
             Ok(chain) => chain == self.effective_chain,
-            Err(crate::middleware::tenant_scope::WalkError::EntityMissing { .. }) => {
+            Err(tenant_scope::WalkError::EntityMissing { .. }) => {
                 self.effective_chain == vec![leaf.clone()]
             }
             Err(_) => false,
@@ -334,7 +334,7 @@ pub async fn subscribe(
     let filters = filters(&state, &schema, &claims, &params).await?;
     let local_account = runtime
         .auth_store
-        .get_user(&claims.sub)
+        .get_user(account_username(&claims))
         .await
         .map_err(ForgeError::from)?
         .is_some();
@@ -391,3 +391,64 @@ pub async fn subscribe(
         .insert("x-accel-buffering", "no".parse().expect("static header"));
     Ok(response)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::access::PLATFORM_ADMIN_ROLE;
+
+    fn tenant(id: &str) -> TenantRef {
+        TenantRef {
+            schema: "Organization".into(),
+            entity_id: id.into(),
+        }
+    }
+    fn grant(id: &str, role: &str) -> TenantRole {
+        TenantRole {
+            tenant: tenant(id),
+            role: role.into(),
+        }
+    }
+    fn claims(sub: &str, roles: &[&str]) -> Claims {
+        serde_json::from_value(serde_json::json!({
+            "sub": sub, "roles": roles, "perms": [], "exp": 9_999_999_999_u64
+        }))
+        .unwrap()
+    }
+
+    #[test]
+    fn expected_roles_add_only_the_active_membership_role() {
+        let global = vec!["member".to_owned()];
+        let grants = [
+            grant("alpha", "owner"),
+            grant("beta", "auditor"),
+            grant("alpha", PLATFORM_ADMIN_ROLE),
+        ];
+        let alpha = tenant("alpha");
+        assert_eq!(
+            expected_roles(&global, &grants, Some(&alpha)),
+            BTreeSet::from(["member", "owner"])
+        );
+        assert_eq!(
+            expected_roles(&global, &grants, None),
+            BTreeSet::from(["member"])
+        );
+        let lobby = tenant("lobby");
+        assert_eq!(
+            expected_roles(&global, &grants, Some(&lobby)),
+            BTreeSet::from(["member"])
+        );
+    }
+
+    #[test]
+    fn platform_admins_are_never_tenant_scoped() {
+        assert!(!tenant_scope::is_tenant_scoped(&claims(
+            "user:admin",
+            &[PLATFORM_ADMIN_ROLE]
+        )));
+        assert!(tenant_scope::is_tenant_scoped(&claims(
+            "user:solo",
+            &["member"]
+        )));
+    }
+}