feat(cat-gateway): RBAC Metadata Indexing (#799)

* feat(catalyst-gateway): add draft schema for RBAC registrations

* wip: schema for RBAC valid registrations

* wip: fix doc comment

* wip: skeleton code for indexing rbac registrations

* fix: RBAC509_registration table name

* fix: rbac509_registration clustering key order

* wip: rbac 509 prepared queries are added to block indexing

* wip: rbac 509 prepare and execute batch queries

* wip: index rbac registrations

* fix: update to use upstream changes

* feat(cat-gateway): add table schemas for chain root indexing.

Add tables for indexing chain root hashes by TX Id, Role 0, and Stake
Address.

* feat(cat-gateway): prepare queries for chain root indexing

* wip(cat-gateway): add caches for chain_root

* chore(utilities/local-scylla): tweaks to justfile

* wip

* fix(cat-gateway): spelling

* chore: update Earthfile to v3.2.13

* wip(cat-gateway): chain-root indexing insert queries

* wip(cat-gateway): use wip branch in catalyst-libs

* wip(cat-gateway): role0 key indexing

* Extract role0 key from certificate
* WIP: Extract stake address from certificate

* chore: update Earthfiles

* feat(catalyst-gateway): index stake addresses from x509 certs

* wip(cat-gateway): index stake addresses from C509 certs

* feat(cat-gateway): index stake addresses from C509 certs

* fix(cat-gateway): bump dependencies, cleanup

* fix(cat-gateway): code format

* fix(cat-gateway): update chain-follower code

* fix(cat-gateway): update schema version for tests

---------

Co-authored-by: Steven Johnson <[email protected]>
Co-authored-by: Steven Johnson <[email protected]>

.config/dictionaries/project.dic

@@ -81,6 +81,7 @@ encryptor
 endfunction
 Eternl
 EUTXO
+extn
 fetchval
 fmtchk
 fmtfix

catalyst-gateway/bin/Cargo.toml

@@ -15,7 +15,8 @@ repository.workspace =  true
 workspace = true
 
 [dependencies]
-cardano-chain-follower = { version = "0.0.2", git = "https://github.com/input-output-hk/catalyst-libs.git", tag = "v0.0.2" }
+cardano-chain-follower = { version = "0.0.3", git = "https://github.com/input-output-hk/catalyst-libs.git", tag = "v0.0.3" }
+c509-certificate = { version = "0.0.3", git = "https://github.com/input-output-hk/catalyst-libs.git", tag = "v0.0.3" }
 
 pallas = { version = "0.30.1", git = "https://github.com/input-output-hk/catalyst-pallas.git", rev = "9b5183c8b90b90fe2cc319d986e933e9518957b3" }
 pallas-traverse = { version = "0.30.1", git = "https://github.com/input-output-hk/catalyst-pallas.git", rev = "9b5183c8b90b90fe2cc319d986e933e9518957b3" }
@@ -61,7 +62,7 @@ strum = { version = "0.26.3", features = ["derive"] }
 strum_macros = "0.26.4"
 openssl = { version = "0.10.66", features = ["vendored"] }
 num-bigint = "0.4.6"
-futures = "0.3.30"
+futures = "0.3.31"
 rand = "0.8.5"
 moka = { version = "0.12.8", features = ["future"] }
 crossbeam-skiplist = "0.1.3"
@@ -90,9 +91,11 @@ rust-embed = "8.5.0"
 num-traits = "0.2.19"
 base64 = "0.22.1"
 dashmap = "6.1.0"
+x509-cert = "0.2.5"
+der-parser = "9.0.0"
 
 [dev-dependencies]
 proptest = "1.5.0"
 
 [build-dependencies]
-build-info-build = "0.0.38"
+build-info-build = "0.0.38"

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

@@ -3,12 +3,14 @@
 
 pub(crate) mod certs;
 pub(crate) mod cip36;
+pub(crate) mod rbac509;
 pub(crate) mod txi;
 pub(crate) mod txo;
 
 use cardano_chain_follower::MultiEraBlock;
 use certs::CertInsertQuery;
 use cip36::Cip36InsertQuery;
+use rbac509::Rbac509InsertQuery;
 use tracing::error;
 use txi::TxiInsertQuery;
 use txo::TxoInsertQuery;
@@ -25,6 +27,7 @@ pub(crate) async fn index_block(block: &MultiEraBlock) -> anyhow::Result<()> {
 
     let mut cert_index = CertInsertQuery::new();
     let mut cip36_index = Cip36InsertQuery::new();
+    let mut rbac509_index = Rbac509InsertQuery::new();
 
     let mut txi_index = TxiInsertQuery::new();
     let mut txo_index = TxoInsertQuery::new();
@@ -52,6 +55,9 @@ pub(crate) async fn index_block(block: &MultiEraBlock) -> anyhow::Result<()> {
 
         // Index the TXOs.
         txo_index.index(txs, slot_no, &txn_hash, txn);
+
+        // Index RBAC 509 inside the transaction.
+        rbac509_index.index(&txn_hash, txn_index, txn, slot_no, block);
     }
 
     // We then execute each batch of data from the block.
@@ -62,6 +68,7 @@ pub(crate) async fn index_block(block: &MultiEraBlock) -> anyhow::Result<()> {
     query_handles.extend(txi_index.execute(&session));
     query_handles.extend(cert_index.execute(&session));
     query_handles.extend(cip36_index.execute(&session));
+    query_handles.extend(rbac509_index.execute(&session));
 
     let mut result: anyhow::Result<()> = Ok(());
 

catalyst-gateway/bin/src/db/index/block/rbac509/cql/insert_chain_root_for_role0_key.cql

@@ -0,0 +1,12 @@
+-- Index of Chain Root For Role0 Key. RBAC 509 registrations.
+INSERT INTO chain_root_for_role0_key (
+    role0_key,
+    slot_no,
+    txn,
+    chain_root
+) VALUES (
+    :role0_key,
+    :slot_no,
+    :txn,
+    :chain_root
+);

catalyst-gateway/bin/src/db/index/block/rbac509/cql/insert_chain_root_for_stake_address.cql

@@ -0,0 +1,12 @@
+-- Index of Chain Root For Stake Address. RBAC 509 registrations.
+INSERT INTO chain_root_for_stake_addr (
+    stake_addr,
+    slot_no,
+    txn,
+    chain_root
+) VALUES (
+    :stake_addr,
+    :slot_no,
+    :txn,
+    :chain_root
+);

catalyst-gateway/bin/src/db/index/block/rbac509/cql/insert_chain_root_for_txn_id.cql

@@ -0,0 +1,8 @@
+-- Index of Chain Root For TX ID. RBAC 509 registrations.
+INSERT INTO chain_root_for_txn_id (
+    transaction_id,
+    chain_root
+) VALUES (
+    :transaction_id,
+    :chain_root
+);

catalyst-gateway/bin/src/db/index/block/rbac509/cql/insert_rbac509.cql

@@ -0,0 +1,16 @@
+-- Index RBAC 509 Registrations (Valid)
+INSERT INTO RBAC509_registration (
+    chain_root,
+    slot_no,
+    txn,
+    transaction_id,
+    purpose,
+    prv_txn_id
+) VALUES (
+    :chain_root,
+    :slot_no,
+    :txn,
+    :transaction_id,
+    :purpose,
+    :prv_txn_id
+);

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

@@ -0,0 +1,71 @@
+//! Index RBAC Chain Root For Role 0 Key Insert Query.
+use std::{fmt::Debug, sync::Arc};
+
+use scylla::{SerializeRow, Session};
+use tracing::error;
+
+use crate::{
+    db::index::queries::{PreparedQueries, SizedBatch},
+    settings::cassandra_db::EnvVars,
+};
+
+/// Index RBAC Chain Root by Role 0 Key
+const INSERT_CHAIN_ROOT_FOR_ROLE0_KEY_QUERY: &str =
+    include_str!("./cql/insert_chain_root_for_role0_key.cql");
+
+/// Insert Chain Root For Role 0 Key Query Parameters
+#[derive(SerializeRow)]
+pub(super) struct Params {
+    /// Role 0 Key Hash. 32 bytes.
+    role0_key: Vec<u8>,
+    /// Block Slot Number
+    slot_no: num_bigint::BigInt,
+    /// Transaction Offset inside the block.
+    txn: i16,
+    /// Chain Root Hash. 32 bytes.
+    chain_root: Vec<u8>,
+}
+
+impl Debug for Params {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("Params")
+            .field("role0_key", &self.role0_key)
+            .field("slot_no", &self.slot_no)
+            .field("txn", &self.txn)
+            .field("chain_root", &self.chain_root)
+            .finish()
+    }
+}
+
+impl Params {
+    /// Create a new record for this transaction.
+    pub(super) fn new(role0_key: &[u8], chain_root: &[u8], slot_no: u64, txn: i16) -> Self {
+        Params {
+            role0_key: role0_key.to_vec(),
+            slot_no: num_bigint::BigInt::from(slot_no),
+            txn,
+            chain_root: chain_root.to_vec(),
+        }
+    }
+
+    /// Prepare Batch of RBAC Registration Index Data Queries
+    pub(super) async fn prepare_batch(
+        session: &Arc<Session>, cfg: &EnvVars,
+    ) -> anyhow::Result<SizedBatch> {
+        let insert_queries = PreparedQueries::prepare_batch(
+            session.clone(),
+            INSERT_CHAIN_ROOT_FOR_ROLE0_KEY_QUERY,
+            cfg,
+            scylla::statement::Consistency::Any,
+            true,
+            false,
+        )
+        .await;
+
+        if let Err(ref error) = insert_queries {
+            error!(error=%error,"Failed to prepare Insert Chain Root For Role 0 Key Query.");
+        };
+
+        insert_queries
+    }
+}

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

@@ -0,0 +1,71 @@
+//! Index RBAC Chain Root For Stake Address Insert Query.
+use std::{fmt::Debug, sync::Arc};
+
+use scylla::{SerializeRow, Session};
+use tracing::error;
+
+use crate::{
+    db::index::queries::{PreparedQueries, SizedBatch},
+    settings::cassandra_db::EnvVars,
+};
+
+/// Index RBAC Chain Root by Stake Address
+const INSERT_CHAIN_ROOT_FOR_STAKE_ADDRESS_QUERY: &str =
+    include_str!("./cql/insert_chain_root_for_stake_address.cql");
+
+/// Insert Chain Root For Stake Address Query Parameters
+#[derive(SerializeRow)]
+pub(super) struct Params {
+    /// Stake Address Hash. 32 bytes.
+    stake_address: Vec<u8>,
+    /// Block Slot Number
+    slot_no: num_bigint::BigInt,
+    /// Transaction Offset inside the block.
+    txn: i16,
+    /// Chain Root Hash. 32 bytes.
+    chain_root: Vec<u8>,
+}
+
+impl Debug for Params {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("Params")
+            .field("stake_address", &self.stake_address)
+            .field("slot_no", &self.slot_no)
+            .field("txn", &self.txn)
+            .field("chain_root", &self.chain_root)
+            .finish()
+    }
+}
+
+impl Params {
+    /// Create a new record for this transaction.
+    pub(super) fn new(stake_address: &[u8], chain_root: &[u8], slot_no: u64, txn: i16) -> Self {
+        Params {
+            stake_address: stake_address.to_vec(),
+            slot_no: num_bigint::BigInt::from(slot_no),
+            txn,
+            chain_root: chain_root.to_vec(),
+        }
+    }
+
+    /// Prepare Batch of RBAC Registration Index Data Queries
+    pub(super) async fn prepare_batch(
+        session: &Arc<Session>, cfg: &EnvVars,
+    ) -> anyhow::Result<SizedBatch> {
+        let insert_queries = PreparedQueries::prepare_batch(
+            session.clone(),
+            INSERT_CHAIN_ROOT_FOR_STAKE_ADDRESS_QUERY,
+            cfg,
+            scylla::statement::Consistency::Any,
+            true,
+            false,
+        )
+        .await;
+
+        if let Err(ref error) = insert_queries {
+            error!(error=%error,"Failed to prepare Insert Chain Root For Stake Address Query.");
+        };
+
+        insert_queries
+    }
+}

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

@@ -0,0 +1,63 @@
+//! Index RBAC Chain Root For Transaction ID Insert Query.
+use std::{fmt::Debug, sync::Arc};
+
+use scylla::{SerializeRow, Session};
+use tracing::error;
+
+use crate::{
+    db::index::queries::{PreparedQueries, SizedBatch},
+    settings::cassandra_db::EnvVars,
+};
+
+/// Index RBAC Chain Root by TX ID.
+const INSERT_CHAIN_ROOT_FOR_TXN_ID_QUERY: &str =
+    include_str!("./cql/insert_chain_root_for_txn_id.cql");
+
+/// Insert Chain Root For Transaction ID Query Parameters
+#[derive(SerializeRow)]
+pub(super) struct Params {
+    /// Transaction ID Hash. 32 bytes.
+    transaction_id: Vec<u8>,
+    /// Chain Root Hash. 32 bytes.
+    chain_root: Vec<u8>,
+}
+
+impl Debug for Params {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("Params")
+            .field("transaction_id", &self.transaction_id)
+            .field("chain_root", &self.chain_root)
+            .finish()
+    }
+}
+
+impl Params {
+    /// Create a new record for this transaction.
+    pub(super) fn new(chain_root: &[u8], transaction_id: &[u8]) -> Self {
+        Params {
+            transaction_id: transaction_id.to_vec(),
+            chain_root: chain_root.to_vec(),
+        }
+    }
+
+    /// Prepare Batch of RBAC Registration Index Data Queries
+    pub(super) async fn prepare_batch(
+        session: &Arc<Session>, cfg: &EnvVars,
+    ) -> anyhow::Result<SizedBatch> {
+        let insert_queries = PreparedQueries::prepare_batch(
+            session.clone(),
+            INSERT_CHAIN_ROOT_FOR_TXN_ID_QUERY,
+            cfg,
+            scylla::statement::Consistency::Any,
+            true,
+            false,
+        )
+        .await;
+
+        if let Err(ref error) = insert_queries {
+            error!(error=%error,"Failed to prepare Insert Chain Root For TXN ID Query.");
+        };
+
+        insert_queries
+    }
+}