refactor(cat-gateway): Apply RBAC validation logic from catalyst-libs

catalyst-gateway/bin/Cargo.toml

@@ -16,7 +16,7 @@ workspace = true
 
 [dependencies]
 cardano-chain-follower = { version = "0.0.15", git = "https://github.com/input-output-hk/catalyst-libs.git", tag = "cardano-chain-follower/v0.0.15" }
-rbac-registration = { version = "0.0.11", git = "https://github.com/input-output-hk/catalyst-libs.git", tag = "rbac-registration/v0.0.11" }
+rbac-registration = { version = "0.0.11", git = "https://github.com/input-output-hk/catalyst-libs.git", rev = "81e7f42a5bb50ff64f49cc1510b823301fb5dae1" }
 catalyst-signed-doc = { version = "0.0.8", git = "https://github.com/input-output-hk/catalyst-libs.git", tag = "catalyst-signed-doc/v0.0.8" }
 catalyst-signed-doc-v1 = { package = "catalyst-signed-doc", version = "0.0.4", git = "https://github.com/input-output-hk/catalyst-libs.git", tag = "catalyst-signed-doc/v.0.0.4" }
 c509-certificate = { version = "0.0.3", git = "https://github.com/input-output-hk/catalyst-libs.git", tag = "c509-certificate-v0.0.3" }

catalyst-gateway/bin/src/db/index/block/rbac509/mod.rs

@@ -13,7 +13,7 @@ use std::{
 
 use anyhow::{Context, Result};
 use cardano_chain_follower::{hashes::TransactionId, MultiEraBlock, Slot, TxnIndex};
-use rbac_registration::cardano::cip509::Cip509;
+use rbac_registration::{cardano::cip509::Cip509, registration::cardano::validation::RbacValidationError};
 use scylla::client::session::Session;
 use tokio::sync::watch;
 use tracing::{debug, error};
@@ -25,8 +25,7 @@ use crate::{
     },
     metrics::caches::rbac::{inc_index_sync, inc_invalid_rbac_reg_count},
     rbac::{
-        validate_rbac_registration, RbacBlockIndexingContext, RbacValidationError,
-        RbacValidationSuccess,
+        validate_rbac_registration, RbacBlockIndexingContext,
     },
     settings::cassandra_db::EnvVars,
 };
@@ -134,13 +133,13 @@ impl Rbac509InsertQuery {
             // Write updates to the database. There can be multiple updates in one registration
             // because a new chain can take ownership of stake addresses of the existing chains and
             // in that case we want to record changes to all those chains as well as the new one.
-            Ok(RbacValidationSuccess {
-                catalyst_id,
-                stake_addresses,
-                public_keys,
-                modified_chains,
-                purpose,
-            }) => {
+            Ok(cip509) => {
+                let catalyst_id = cip509.catalyst_id().context("Cip509 error: cannot read Catalyst ID")?;
+                let stake_addresses = cip509.stake_addresses().clone();
+                let public_keys = cip509.public_keys().clone();
+                let modified_chains = cip509.modified_chains().clone();
+                let purpose = cip509.purpose();
+
                 // Record the transaction identifier (hash) of a new registration.
                 self.catalyst_id_for_txn_id
                     .push(insert_catalyst_id_for_txn_id::Params::new(
@@ -171,7 +170,7 @@ impl Rbac509InsertQuery {
                 }
                 // Update the chain this registration belongs to.
                 self.registrations.push(insert_rbac509::Params::new(
-                    catalyst_id,
+                    catalyst_id.clone(),
                     txn_hash,
                     slot,
                     index,

catalyst-gateway/bin/src/rbac/indexing_context.rs

@@ -2,11 +2,23 @@
 
 use std::collections::{HashMap, HashSet};
 
+use anyhow::Context;
 use cardano_chain_follower::{hashes::TransactionId, Slot, StakeAddress, TxnIndex};
 use catalyst_types::catalyst_id::CatalystId;
 use ed25519_dalek::VerifyingKey;
+use futures::StreamExt;
+use rbac_registration::providers::RbacRegistrationProvider;
 
-use crate::db::index::queries::rbac::get_rbac_registrations::Query as RbacQuery;
+use crate::{
+    db::index::{
+        queries::rbac::get_rbac_registrations::Query as RbacQuery, session::CassandraSession,
+    },
+    rbac::{
+        chains_cache::cached_persistent_rbac_chain,
+        get_chain::{apply_regs, build_rbac_chain, persistent_rbac_chain},
+        latest_rbac_chain,
+    },
+};
 
 /// A RBAC context used during indexing.
 ///
@@ -160,3 +172,165 @@ impl RbacBlockIndexingContext {
         self.registrations.get(id).map(Vec::as_slice)
     }
 }
+
+impl RbacRegistrationProvider for RbacBlockIndexingContext {
+    async fn chain(
+        &self,
+        id: CatalystId,
+        is_persistent: bool,
+    ) -> anyhow::Result<Option<rbac_registration::registration::cardano::RegistrationChain>> {
+        let chain = if is_persistent {
+            persistent_rbac_chain(&id).await?
+        } else {
+            latest_rbac_chain(&id).await?.map(|i| i.chain)
+        };
+
+        // Apply additional registrations from context if any.
+        if let Some(regs) = self.find_registrations(&id) {
+            let regs = regs.iter().cloned();
+            match chain {
+                Some(c) => apply_regs(c, regs).await.map(Some),
+                None => build_rbac_chain(regs).await,
+            }
+        } else {
+            Ok(chain)
+        }
+    }
+
+    async fn catalyst_id_from_txn_id(
+        &self,
+        txn_id: TransactionId,
+        is_persistent: bool,
+    ) -> anyhow::Result<Option<CatalystId>> {
+        use crate::db::index::queries::rbac::get_catalyst_id_from_transaction_id::Query;
+
+        // Check the context first.
+        if let Some(catalyst_id) = self.find_transaction(&txn_id) {
+            return Ok(Some(catalyst_id.to_owned()));
+        }
+
+        // Then try to find in the persistent database.
+        let session =
+            CassandraSession::get(true).context("Failed to get Cassandra persistent session")?;
+        if let Some(id) = Query::get(&session, txn_id).await? {
+            return Ok(Some(id));
+        }
+
+        // Conditionally check the volatile database.
+        if !is_persistent {
+            let session =
+                CassandraSession::get(false).context("Failed to get Cassandra volatile session")?;
+            return Query::get(&session, txn_id).await;
+        }
+
+        Ok(None)
+    }
+
+    async fn catalyst_id_from_stake_address(
+        &self,
+        address: &StakeAddress,
+        is_persistent: bool,
+    ) -> anyhow::Result<Option<CatalystId>> {
+        use crate::db::index::queries::rbac::get_catalyst_id_from_stake_address::Query;
+
+        // Check the context first.
+        if let Some(catalyst_id) = self.find_address(address) {
+            return Ok(Some(catalyst_id.to_owned()));
+        }
+
+        // Then try to find in the persistent database.
+        let session =
+            CassandraSession::get(true).context("Failed to get Cassandra persistent session")?;
+        if let Some(id) = Query::latest(&session, address).await? {
+            return Ok(Some(id));
+        }
+
+        // Conditionally check the volatile database.
+        if !is_persistent {
+            let session =
+                CassandraSession::get(false).context("Failed to get Cassandra volatile session")?;
+            return Query::latest(&session, address).await;
+        }
+
+        Ok(None)
+    }
+
+    async fn catalyst_id_from_public_key(
+        &self,
+        key: VerifyingKey,
+        is_persistent: bool,
+    ) -> anyhow::Result<Option<CatalystId>> {
+        use crate::db::index::queries::rbac::get_catalyst_id_from_public_key::Query;
+
+        // Check the context first.
+        if let Some(catalyst_id) = self.find_public_key(&key) {
+            return Ok(Some(catalyst_id.to_owned()));
+        }
+
+        // Then try to find in the persistent database.
+        let session =
+            CassandraSession::get(true).context("Failed to get Cassandra persistent session")?;
+        if let Some(id) = Query::get(&session, key).await? {
+            return Ok(Some(id));
+        }
+
+        // Conditionally check the volatile database.
+        if !is_persistent {
+            let session =
+                CassandraSession::get(false).context("Failed to get Cassandra volatile session")?;
+            return Query::get(&session, key).await;
+        }
+
+        Ok(None)
+    }
+
+    async fn is_chain_known(
+        &self,
+        id: CatalystId,
+        is_persistent: bool,
+    ) -> anyhow::Result<bool> {
+        if self.find_registrations(&id).is_some() {
+            return Ok(true);
+        }
+
+        let session =
+            CassandraSession::get(true).context("Failed to get Cassandra persistent session")?;
+
+        // We only cache persistent chains, so it is ok to check the cache regardless of the
+        // `is_persistent` parameter value.
+        if cached_persistent_rbac_chain(&session, &id).is_some() {
+            return Ok(true);
+        }
+
+        if is_cat_id_known(&session, &id).await? {
+            return Ok(true);
+        }
+
+        // Conditionally check the volatile database.
+        if !is_persistent {
+            let session =
+                CassandraSession::get(false).context("Failed to get Cassandra volatile session")?;
+            if is_cat_id_known(&session, &id).await? {
+                return Ok(true);
+            }
+        }
+
+        Ok(false)
+    }
+}
+
+/// Returns `true` if there is at least one registration with the given Catalyst ID.
+async fn is_cat_id_known(
+    session: &CassandraSession,
+    id: &CatalystId,
+) -> anyhow::Result<bool> {
+    use crate::db::index::queries::rbac::get_rbac_registrations::{Query, QueryParams};
+
+    Ok(Query::execute(session, QueryParams {
+        catalyst_id: id.clone().into(),
+    })
+    .await?
+    .next()
+    .await
+    .is_some())
+}

catalyst-gateway/bin/src/rbac/mod.rs

@@ -5,10 +5,8 @@ mod chains_cache;
 mod get_chain;
 mod indexing_context;
 mod validation;
-mod validation_result;
 
 pub use chain_info::ChainInfo;
 pub use get_chain::latest_rbac_chain;
 pub use indexing_context::RbacBlockIndexingContext;
 pub use validation::validate_rbac_registration;
-pub use validation_result::{RbacValidationError, RbacValidationResult, RbacValidationSuccess};