fix(iam): delete what a principal or account owns with it

crates/server/src/console/pages/policy_pages.rs

@@ -326,17 +326,29 @@ async fn put_policy(
         .into_response();
     }
 
-    match state
-        .catalog_store
-        .put_policy(
-            account_id,
-            principal_type,
-            principal_name,
-            &form.policy_name,
-            &doc,
-        )
-        .await
+    let stored = match crate::management::require_principal(
+        &*state.catalog_store,
+        account_id,
+        principal_type,
+        principal_name,
+    )
+    .await
     {
+        Ok(()) => {
+            state
+                .catalog_store
+                .put_policy(
+                    account_id,
+                    principal_type,
+                    principal_name,
+                    &form.policy_name,
+                    &doc,
+                )
+                .await
+        }
+        Err(e) => Err(e),
+    };
+    match stored {
         Ok(()) => {
             invalidate_policy_caches(state, account_id, principal_type, principal_name).await;
             Redirect::to(redirect_url).into_response()

crates/server/src/management/iam_policy.rs

@@ -23,6 +23,43 @@ struct PolicyEntry {
     created_at: String,
 }
 
+/// `NotFound` unless the principal exists, as IAM answers `NoSuchEntity` for a
+/// policy or boundary put on a principal it does not have.
+///
+/// Inline policies and boundaries are stored by principal name, so a put on a
+/// name with no principal would wait there and belong to the next principal
+/// created with that name. Its create removes such rows too, which covers a
+/// put that races a delete.
+pub(crate) async fn require_principal(
+    store: &dyn extenddb_storage::CatalogStore,
+    account_id: &str,
+    principal_type: &str,
+    principal_name: &str,
+) -> Result<(), extenddb_storage::management_store::OpError> {
+    let exists = match principal_type {
+        "user" => store
+            .get_user_detail(account_id, principal_name)
+            .await?
+            .is_some(),
+        "group" => store
+            .get_group_detail(account_id, principal_name)
+            .await?
+            .is_some(),
+        "role" => store
+            .get_role_detail(account_id, principal_name)
+            .await?
+            .is_some(),
+        other => unreachable!("principal type {other}"),
+    };
+    if exists {
+        Ok(())
+    } else {
+        Err(extenddb_storage::management_store::OpError::NotFound(
+            format!("IAM {principal_type} not found"),
+        ))
+    }
+}
+
 pub async fn put_user_policy(
     State(state): State<Arc<ManagementState>>,
     headers: HeaderMap,
@@ -211,6 +248,17 @@ async fn put_policy(
         }
     }
 
+    if let Err(e) = require_principal(
+        &*state.catalog_store,
+        account_id,
+        principal_type,
+        principal_name,
+    )
+    .await
+    {
+        return op_err_to_response(OpError::from_storage(e));
+    }
+
     match state
         .catalog_store
         .put_policy(

crates/server/src/management/mod.rs

@@ -21,6 +21,7 @@ mod cache_metrics;
 pub(crate) mod crypto;
 mod iam_group;
 mod iam_policy;
+pub(crate) use iam_policy::require_principal;
 mod iam_role;
 mod iam_user;
 mod iam_user_self;

crates/server/src/management/permissions_boundary.rs

@@ -137,6 +137,17 @@ async fn set_boundary(
         }
     }
 
+    if let Err(e) = super::iam_policy::require_principal(
+        &*state.catalog_store,
+        account_id,
+        principal_type,
+        principal_name,
+    )
+    .await
+    {
+        return op_err_to_response(OpError::from_storage(e));
+    }
+
     let result = match principal_type {
         "user" => {
             state

crates/storage-cassandra/src/catalog_store.rs

@@ -240,6 +240,27 @@ impl CassandraCatalogStore {
         Ok(!rows.is_empty())
     }
 
+    /// [`Self::account_exists`] read at `LOCAL_QUORUM`, for the console's
+    /// quorum reads: an account delete, committed at quorum, is seen.
+    ///
+    /// Before account deletes removed them, a deleted account's users stayed
+    /// in the catalog; the console treats a user whose account is gone as
+    /// absent.
+    pub(crate) async fn account_exists_quorum(&self, account_id: &str) -> Result<bool, OpError> {
+        let query = format!(
+            "SELECT account_id FROM {}.accounts WHERE account_id = ?",
+            self.catalog_keyspace()
+        );
+        let rows = crate::cassandra_util::query_rows_quorum(
+            &self.session,
+            &query,
+            cdrs_tokio::query_values!(account_id),
+            "account_exists_quorum",
+        )
+        .await?;
+        Ok(!rows.is_empty())
+    }
+
     /// Check if a user exists. Used to emulate foreign key checks.
     pub(crate) async fn user_exists(
         &self,

crates/storage-cassandra/src/credential_store.rs

@@ -51,6 +51,25 @@ impl CassandraCredentialStore {
     fn catalog_keyspace(&self) -> String {
         format!("{}_catalog", self.keyspace_prefix)
     }
+
+    /// Whether `account_id` has a row in `accounts`. Before account deletes
+    /// removed them, a deleted account's access keys and role sessions stayed
+    /// in the catalog; a credential of an account that no longer exists is
+    /// unknown.
+    async fn account_exists(&self, account_id: &str) -> Result<bool, DynamoDbError> {
+        let query = format!(
+            "SELECT account_id FROM {}.accounts WHERE account_id = ?",
+            self.catalog_keyspace()
+        );
+        crate::cassandra_util::query_rows::<DynamoDbError>(
+            &self.session,
+            &query,
+            cdrs_tokio::query_values!(account_id),
+            "credential account_exists",
+        )
+        .await
+        .map(|rows| !rows.is_empty())
+    }
 }
 
 #[async_trait]
@@ -125,6 +144,9 @@ impl CassandraCredentialStore {
             tracing::error!("Failed to parse is_active: {e}");
             DynamoDbError::InternalServerError("Internal error during authentication".to_owned())
         })?;
+        if !self.account_exists(&account_id).await? {
+            return Ok(None);
+        }
 
         let secret_key =
             decrypt_secret(&encrypted, &self.encryption_key, access_key_id).map_err(|e| {
@@ -206,6 +228,9 @@ impl CassandraCredentialStore {
             tracing::error!("Failed to parse session_token: {e}");
             DynamoDbError::InternalServerError("Internal error during authentication".to_owned())
         })?;
+        if !self.account_exists(&account_id).await? {
+            return Ok(None);
+        }
 
         let expires_at_ms: i64 = row.get_r_by_name("expires_at").map_err(|e| {
             tracing::error!("Failed to parse expires_at: {e}");

crates/storage-cassandra/src/management_store/accounts.rs

@@ -59,7 +59,23 @@ impl CassandraCatalogStore {
             return Err(OpError::AlreadyExists("Account already exists".to_owned()));
         }
 
-        Ok(())
+        // An account deleted before deletes removed them could have left its
+        // IAM rows under this id. If they cannot be removed, remove the new
+        // `accounts` row again, so that a retry is not refused while they
+        // stay; the retry finds the keyspace complete.
+        let cleared = self.delete_account_iam(account_id).await;
+        if cleared.is_err()
+            && let Err(e) = crate::cassandra_util::apply_lwt::<OpError>(
+                self.session(),
+                &format!("DELETE FROM {catalog_keyspace}.accounts WHERE account_id = ? IF EXISTS"),
+                cdrs_tokio::query_values!(account_id),
+                "create_account undo",
+            )
+            .await
+        {
+            tracing::error!("create_account: remove the new account after a failed clear: {e:?}");
+        }
+        cleared
     }
 
     pub(crate) async fn delete_account_impl(&self, account_id: &str) -> OpResult<()> {
@@ -89,6 +105,12 @@ impl CassandraCatalogStore {
             ));
         }
 
+        // The account's users, groups, roles and everything they own go
+        // first, as the foreign keys to `accounts` remove them on SQLite and
+        // PostgreSQL. Left behind, a user could still sign in, and an account
+        // created later with the same id would get them all.
+        self.delete_account_iam(account_id).await?;
+
         // Backups outlive their source table, but cannot outlive the account
         // keyspace that stores their payload. Remove every denormalized and
         // authoritative catalog row before dropping that keyspace.
@@ -142,7 +164,8 @@ impl CassandraCatalogStore {
         )
         .await?;
 
-        crate::cassandra_util::apply_lwt(
+        // Quorum commit, so the console's quorum reads see the account gone.
+        crate::cassandra_util::apply_lwt_quorum(
             self.session(),
             &format!("DELETE FROM {catalog_keyspace}.accounts WHERE account_id = ? IF EXISTS"),
             cdrs_tokio::query_values!(account_id),