feat: Add SCIM scenarios; Fix findings

crates/core/src/api/api_key_auth.rs

@@ -25,7 +25,7 @@ use axum::extract::{FromRef, FromRequestParts, Path};
 use axum::http::request::Parts;
 use governor::clock::Clock as _;
 use ipnet::IpNet;
-use tracing::warn;
+use tracing::{debug, warn};
 
 use openstack_keystone_core_types::api_key::ApiClientResource;
 use openstack_keystone_core_types::auth::AuthenticationError;
@@ -241,12 +241,25 @@ async fn resolve_verified_api_client(
         // Dummy hash: burn the same Argon2id cost as a real verification
         // to prevent timing-based enumeration of valid lookup hashes
         // (ADR 0021 Invariant 7).
+        debug!(
+            domain_id,
+            lookup_hash = %parsed.lookup_hash,
+            "api_key lookup returned no resource"
+        );
         let _ = crypto::generate_dummy_hash(&cfg).await;
         return Err(AuthenticationError::Unauthorized.into());
     };
 
     let now = chrono::Utc::now().timestamp();
     if !resource.is_active(now) {
+        debug!(
+            domain_id,
+            lookup_hash = %parsed.lookup_hash,
+            enabled = resource.enabled,
+            expires_at = resource.expires_at,
+            now,
+            "api_key resource found but inactive"
+        );
         let _ = crypto::generate_dummy_hash(&cfg).await;
         return Err(AuthenticationError::Unauthorized.into());
     }
@@ -260,6 +273,20 @@ async fn resolve_verified_api_client(
         cfg.trusted_header,
     );
     if !ip_allowed(effective_ip, &resource.allowed_ips) {
+        // Burn the same Argon2id cost as a real verification here too --
+        // otherwise this branch returns near-instantly while the
+        // not-found/inactive/wrong-secret branches all pay the hashing
+        // cost, letting an attacker distinguish "this lookup_hash exists
+        // but my IP is disallowed" from "this lookup_hash doesn't exist"
+        // by response latency alone (ADR 0021 Invariant 7).
+        debug!(
+            domain_id,
+            client_id = %resource.client_id,
+            ?effective_ip,
+            allowed_ips = ?resource.allowed_ips,
+            "api_key resource found but IP not allowed"
+        );
+        let _ = crypto::generate_dummy_hash(&cfg).await;
         return Err(AuthenticationError::Unauthorized.into());
     }
 
@@ -271,6 +298,12 @@ async fn resolve_verified_api_client(
             AuthenticationError::Unauthorized
         })?;
     if !verified {
+        debug!(
+            domain_id,
+            client_id = %resource.client_id,
+            lookup_hash = %parsed.lookup_hash,
+            "api_key resource found, IP allowed, but secret verification failed"
+        );
         return Err(AuthenticationError::Unauthorized.into());
     }
 

crates/storage/src/app.rs

@@ -832,18 +832,20 @@ impl StorageApi for Storage {
                          forwarding get_by_key to leader"
                     );
 
-                    if let Ok(forwarded) = self
+                    // A failed forward means we can no longer prove the
+                    // leader we found is still current (or reachable) --
+                    // falling back to a local read here would silently
+                    // serve possibly-stale data from a follower instead of
+                    // surfacing the inconsistency (this bit us as
+                    // intermittent false-negative auth lookups under load).
+                    // Fail the read; the caller retries.
+                    return self
                         .forwarded_get_by_key(lead_id, lead_node.rpc_addr, key, keyspace)
-                        .await
-                    {
-                        return Ok(forwarded);
-                    }
-                    debug!("forwarded get_by_key failed; falling back to local read");
+                        .await;
                 } else {
-                    debug!(
-                        "ensure_linearizable (ReadIndex) returned ForwardToLeader \
-                         without leader info; falling back to local read"
-                    );
+                    return Err(ApiStoreError::Other(Box::new(StoreError::Other(eyre::eyre!(
+                        "ReadIndex returned ForwardToLeader without leader info"
+                    )))));
                 }
             }
             Err(e) => {
@@ -927,19 +929,15 @@ impl StorageApi for Storage {
                          forwarding prefix to leader"
                     );
 
-                    if let Ok(forwarded) = self
+                    // See `get_by_key`'s equivalent branch: a failed forward
+                    // must not fall back to an unguarded local read.
+                    return self
                         .forwarded_prefix_read(lead_id, lead_node.rpc_addr, prefix, keyspace)
-                        .await
-                    {
-                        return Ok(forwarded);
-                    }
-                    debug!("forwarded prefix failed; falling back to local read");
+                        .await;
                 } else {
-                    // No leader info — fall back to local read (e.g., removed node).
-                    debug!(
-                        "ensure_linearizable (ReadIndex) returned ForwardToLeader \
-                         without leader info; falling back to local read"
-                    );
+                    return Err(ApiStoreError::Other(Box::new(StoreError::Other(eyre::eyre!(
+                        "ReadIndex returned ForwardToLeader without leader info"
+                    )))));
                 }
             }
             Err(e) => {
@@ -1021,18 +1019,21 @@ impl StorageApi for Storage {
                          for prefix_index; forwarding to leader"
                     );
 
-                    if let Ok(forwarded) = self
+                    // See `get_by_key`'s equivalent branch: a failed forward
+                    // must not fall back to an unguarded local read. This
+                    // path backs `get_ruleset_by_source`'s index scan --
+                    // relevant to the intermittent false-negative "mapping
+                    // ruleset not found" 401s observed under concurrent
+                    // load, whose confirmed root cause was the leader-side
+                    // forwarded handler skipping its own leadership/read-
+                    // index re-check (fixed in `storage_service.rs`).
+                    return self
                         .forwarded_prefix_index(lead_id, lead_node.rpc_addr, prefix)
-                        .await
-                    {
-                        return Ok(forwarded);
-                    }
-                    debug!("forwarded prefix_index failed; falling back to local read");
+                        .await;
                 } else {
-                    debug!(
-                        "ensure_linearizable (ReadIndex) returned ForwardToLeader \
-                         without leader info; falling back to local read"
-                    );
+                    return Err(ApiStoreError::Other(Box::new(StoreError::Other(eyre::eyre!(
+                        "ReadIndex returned ForwardToLeader without leader info"
+                    )))));
                 }
             }
             Err(e) => {

crates/storage/src/grpc/storage_service.rs

@@ -14,6 +14,7 @@
 
 use std::sync::Arc;
 
+use openraft::ReadPolicy;
 use tonic::{Request, Response, Status};
 
 use crate::DataTier;
@@ -58,6 +59,23 @@ impl StorageServiceImpl {
             state_machine_store,
         }
     }
+
+    /// Re-confirm this node is still leader with an up-to-date read index
+    /// before serving a forwarded read.
+    ///
+    /// A follower that observed `ForwardToLeader` forwards the read here,
+    /// but leadership can change between that observation and this RPC
+    /// landing -- without re-checking, a former leader would serve a local
+    /// read that is no longer guaranteed linearizable (e.g. a competing
+    /// leader committed writes this node hasn't seen), silently breaking
+    /// read-your-writes for the forwarded caller.
+    async fn ensure_leader_linearizable(&self) -> Result<(), Status> {
+        self.raft_node
+            .ensure_linearizable(ReadPolicy::ReadIndex)
+            .await
+            .map(|_| ())
+            .map_err(|e| Status::unavailable(format!("not linearizable leader: {e}")))
+    }
 }
 
 #[tonic::async_trait]
@@ -98,6 +116,7 @@ impl StorageService for StorageServiceImpl {
         &self,
         request: Request<pb::api::ForwardedGetRequest>,
     ) -> Result<Response<pb::api::ForwardedGetResponse>, Status> {
+        self.ensure_leader_linearizable().await?;
         let req = request.into_inner();
         let key = match std::str::from_utf8(&req.key) {
             Ok(s) => s.to_string(),
@@ -177,6 +196,7 @@ impl StorageService for StorageServiceImpl {
         &self,
         request: Request<pb::api::ForwardedPrefixRequest>,
     ) -> Result<Response<pb::api::ForwardedPrefixResponse>, Status> {
+        self.ensure_leader_linearizable().await?;
         let req = request.into_inner();
 
         let ks = match &req.keyspace {
@@ -256,6 +276,7 @@ impl StorageService for StorageServiceImpl {
         &self,
         request: Request<pb::api::ForwardedPrefixIndexRequest>,
     ) -> Result<Response<pb::api::ForwardedPrefixIndexResponse>, Status> {
+        self.ensure_leader_linearizable().await?;
         let req = request.into_inner();
 
         let items: Vec<_> = self

tests/loadtest/src/main.rs

@@ -60,6 +60,14 @@ use crate::v4::oauth2::{
     jwks as oauth2_jwks, list_clients as oauth2_client_list,
     show_client as oauth2_client_show, well_known as oauth2_well_known,
 };
+use crate::v4::scim_realm::{
+    list as scim_realm_list, show as scim_realm_show, update as scim_realm_update,
+};
+use crate::v4::scim_sync::{
+    provision_api_key as scim_sync_provision_api_key, sync_create as scim_sync_create,
+    sync_delete as scim_sync_delete, sync_list as scim_sync_list, sync_patch as scim_sync_patch,
+    sync_show as scim_sync_show,
+};
 
 /// Per-GooseUser session state shared across transactions.
 pub struct Session {
@@ -104,6 +112,10 @@ async fn main() -> Result<(), GooseError> {
     if let Some(auth_role_id) = &seed_state.auth_role_id {
         v3::role_assignment::set_auth_role_id(auth_role_id.clone());
     }
+    if let Some(scim_realm_provider_id) = &seed_state.scim_realm_provider_id {
+        v4::scim_realm::set_realm_provider_id(scim_realm_provider_id.clone());
+        v4::scim_sync::set_realm_provider_id(scim_realm_provider_id.clone());
+    }
 
     // Default to 45 users so all weighted scenarios get at least 1 user
     // (total weight = 36; 45 ensures proportional coverage).
@@ -278,6 +290,33 @@ async fn main() -> Result<(), GooseError> {
                 .set_weight(3)?
                 .register_transaction(transaction!(oauth2_jwks))
                 .register_transaction(transaction!(oauth2_well_known)),
+        )
+        // Raft-backed SCIM realm reads (ADR 0024): scim-realm-driver-raft.
+        // No per-VU create/delete -- realm is seeded once (no delete endpoint
+        // exists, only per-resource purge), shared read-mostly across VUs.
+        .register_scenario(
+            scenario!("ScimRealmRead")
+                .set_weight(1)?
+                .register_transaction(transaction!(openstack_login).set_on_start())
+                .register_transaction(transaction!(scim_realm_show))
+                .register_transaction(transaction!(scim_realm_list))
+                .register_transaction(transaction!(scim_realm_update)),
+        )
+        // Real SCIM sync simulation (ADR 0021 + 0024): unlike ScimRealmRead
+        // (admin x-auth-token reads), this authenticates via a per-VU
+        // `kscim_...` API Key bearer against `/SCIM/v2`, the actual ingress
+        // path an IdP-side SCIM connector uses -- provision, reconcile
+        // (list/show), attribute drift (PATCH), offboard (DELETE).
+        .register_scenario(
+            scenario!("ScimSync")
+                .set_weight(1)?
+                .register_transaction(transaction!(openstack_login).set_on_start())
+                .register_transaction(transaction!(scim_sync_provision_api_key).set_on_start())
+                .register_transaction(transaction!(scim_sync_create).set_on_start())
+                .register_transaction(transaction!(scim_sync_list))
+                .register_transaction(transaction!(scim_sync_show))
+                .register_transaction(transaction!(scim_sync_patch))
+                .register_transaction(transaction!(scim_sync_delete).set_on_stop()),
         );
 
     attack.execute().await?;