feat: implement server-side signing via hashicorp vault service

Cargo.toml

@@ -23,7 +23,7 @@ futures-util = "0.3.31"
 gasket = { git = "https://github.com/construkts/gasket-rs.git", features = ["derive"] }
 hex = "0.4.3"
 itertools = "0.14.0"
-pallas = { git = "https://github.com/txpipe/pallas.git", features = ["phase-two"] }
+pallas = { git = "https://github.com/txpipe/pallas.git", features = ["phase2", "wallet"] }
 protoc-wkt = "1.0.0"
 serde = { version = "1.0.217", features = ["derive"] }
 thiserror = "2.0.11"
@@ -39,3 +39,5 @@ async-stream = "0.3.6"
 tokio-stream = "0.1.17"
 rand = "0.9.0"
 prost = "0.13.5"
+vaultrs = "0.7.4"
+bip39 = "2.1.0"

src/ledger/u5c/mod.rs

@@ -1,10 +1,14 @@
-use std::{collections::HashMap, pin::Pin, str::FromStr, vec};
+use std::{
+    collections::{BTreeMap, HashMap},
+    pin::Pin,
+    str::FromStr,
+    vec,
+};
 
 use anyhow::bail;
 use async_stream::stream;
 use futures::{Stream, TryStreamExt};
 use pallas::{
-    codec::utils::KeyValuePairs,
     crypto::hash::Hash,
     interop::utxorpc::spec::{
         cardano::{CostModel, Tx},
@@ -322,7 +326,7 @@ fn map_conway_pparams(pparams: &Params) -> ConwayProtParams {
                             .unwrap_or(CostModel { values: vec![] })
                             .values,
                     ),
-                    unknown: KeyValuePairs::from(vec![]),
+                    unknown: BTreeMap::new(),
                 },
                 execution_costs: ExUnitPrices {
                     mem_price: primitives::RationalNumber {

src/main.rs

@@ -16,6 +16,7 @@ mod network;
 mod pipeline;
 mod queue;
 mod server;
+mod signing;
 mod storage;
 mod validation;
 
@@ -56,7 +57,7 @@ async fn main() -> Result<()> {
         Arc::clone(&cursor_storage),
     );
     let server = server::run(
-        config.server,
+        config,
         u5c_data_adapter.clone(),
         Arc::clone(&tx_storage),
         Arc::clone(&tx_chaining),
@@ -76,6 +77,7 @@ struct Config {
     #[serde(default)]
     queues: HashSet<queue::Config>,
     u5c: ledger::u5c::Config,
+    signing: signing::Config,
 }
 
 impl Config {

src/pipeline/ingest.rs

@@ -7,7 +7,9 @@ use tokio::time::sleep;
 use tracing::info;
 
 use super::CAP;
+use crate::signing::SigningAdapter;
 use crate::validation::{evaluate_tx, validate_tx};
+use crate::Config;
 use crate::{
     ledger::u5c::U5cDataAdapter,
     queue::priority::Priority,
@@ -20,6 +22,8 @@ pub struct Stage {
     storage: Arc<SqliteTransaction>,
     priority: Arc<Priority>,
     u5c_adapter: Arc<dyn U5cDataAdapter>,
+    secret_adapter: Arc<dyn SigningAdapter>,
+    config: Config,
     pub output: OutputPort<Vec<u8>>,
 }
 
@@ -28,11 +32,15 @@ impl Stage {
         storage: Arc<SqliteTransaction>,
         priority: Arc<Priority>,
         u5c_adapter: Arc<dyn U5cDataAdapter>,
+        secret_adapter: Arc<dyn SigningAdapter>,
+        config: Config,
     ) -> Self {
         Self {
             storage,
             priority,
             u5c_adapter,
+            secret_adapter,
+            config,
             output: Default::default(),
         }
     }
@@ -78,6 +86,20 @@ impl gasket::framework::Worker<Stage> for Worker {
     ) -> Result<(), WorkerError> {
         for tx in unit {
             let mut tx = tx.clone();
+
+            let should_sign = stage
+                .config
+                .queues
+                .get(&tx.queue)
+                .map(|config| config.server_signing)
+                .unwrap_or(false);
+
+            if should_sign {
+                info!("Signing transaction {} with server key", tx.id);
+                tx.raw = stage.secret_adapter.sign(tx.raw).await.or_retry()?;
+                info!("Transaction {} signed successfully", tx.id);
+            }
+
             let metx = MultiEraTx::decode(&tx.raw).map_err(|_| WorkerError::Recv)?;
 
             if let Err(e) = validate_tx(&metx, stage.u5c_adapter.clone()).await {
@@ -100,7 +122,6 @@ impl gasket::framework::Worker<Stage> for Worker {
                 info!("Failed to broadcast transaction: {}", e);
             } else {
                 info!("Transaction {} broadcasted to receivers", tx.id);
-
                 let tip = stage.u5c_adapter.fetch_tip().await.or_retry()?;
                 tx.status = TransactionStatus::InFlight;
                 tx.slot = Some(tip.0);
@@ -114,12 +135,12 @@ impl gasket::framework::Worker<Stage> for Worker {
 
 #[cfg(test)]
 mod ingest_tests {
-    use std::collections::HashMap;
+    use std::collections::{BTreeMap, HashMap};
     use std::str::FromStr;
     use std::sync::Arc;
 
     use anyhow::Ok;
-    use pallas::codec::utils::KeyValuePairs;
+
     use pallas::crypto::hash::Hash;
     use pallas::ledger::primitives::conway::{
         CostModels, DRepVotingThresholds, PoolVotingThresholds,
@@ -420,7 +441,7 @@ mod ingest_tests {
                     20744, 32, 25933, 32, 24623, 32, 43053543, 10, 53384111, 14333, 10, 43574283,
                     26308, 10,
                 ]),
-                unknown: KeyValuePairs::from(vec![]),
+                unknown: BTreeMap::new(),
             },
             execution_costs: ExUnitPrices {
                 mem_price: RationalNumber {

src/pipeline/mod.rs

@@ -10,6 +10,7 @@ use crate::{
     },
     network::peer_manager::PeerManager,
     queue::priority::Priority,
+    signing::hashicorp::HashicorpVaultClient,
     storage::{
         sqlite::{SqliteCursor, SqliteTransaction},
         Cursor,
@@ -46,12 +47,15 @@ pub async fn run(
 
     let peer_manager = Arc::new(peer_manager);
 
-    let priority = Arc::new(Priority::new(tx_storage.clone(), config.queues));
+    let priority = Arc::new(Priority::new(tx_storage.clone(), config.queues.clone()));
+    let secret_adapter = Arc::new(HashicorpVaultClient::new(config.signing.clone())?);
 
     let mut ingest = ingest::Stage::new(
         tx_storage.clone(),
         priority.clone(),
         u5c_data_adapter.clone(),
+        secret_adapter,
+        config.clone(),
     );
 
     ingest.output.connect(sender);

src/queue/chaining.rs

@@ -176,6 +176,7 @@ mod chaining_tests {
             name: "banana".into(),
             weight: 1,
             chained: true,
+            server_signing: false,
         }]);
 
         let chaining = TxChaining::new(Arc::clone(&storage), queues);
@@ -206,6 +207,7 @@ mod chaining_tests {
             name: "banana".into(),
             weight: 1,
             chained: true,
+            server_signing: false,
         }]);
 
         let chaining = TxChaining::new(Arc::clone(&storage), queues);
@@ -253,11 +255,13 @@ mod chaining_tests {
                 name: "banana".into(),
                 weight: 1,
                 chained: true,
+                server_signing: false,
             },
             Config {
                 name: "orange".into(),
                 weight: 1,
                 chained: true,
+                server_signing: false,
             },
         ]);
 
@@ -305,6 +309,7 @@ mod chaining_tests {
             name: "banana".into(),
             weight: 1,
             chained: true,
+            server_signing: false,
         }]);
 
         let chaining = TxChaining::new(Arc::clone(&storage), queues);
@@ -340,6 +345,7 @@ mod chaining_tests {
             name: "banana".into(),
             weight: 1,
             chained: true,
+            server_signing: false,
         }]);
 
         let chaining = TxChaining::new(Arc::clone(&storage), queues);
@@ -356,6 +362,7 @@ mod chaining_tests {
             name: "banana".into(),
             weight: 1,
             chained: false,
+            server_signing: false,
         }]);
 
         let chaining = TxChaining::new(Arc::clone(&storage), queues);
@@ -372,6 +379,7 @@ mod chaining_tests {
             name: "banana".into(),
             weight: 1,
             chained: true,
+            server_signing: false,
         }]);
 
         let chaining = TxChaining::new(Arc::clone(&storage), queues);
@@ -406,6 +414,7 @@ mod chaining_tests {
             name: "banana".into(),
             weight: 1,
             chained: true,
+            server_signing: false,
         }]);
 
         let chaining = TxChaining::new(Arc::clone(&storage), queues);

src/queue/mod.rs

@@ -15,13 +15,15 @@ pub struct Config {
     pub weight: u8,
     #[serde(default)]
     pub chained: bool,
+    pub server_signing: bool,
 }
 impl Default for Config {
     fn default() -> Self {
         Self {
             name: DEFAULT_QUEUE.into(),
             weight: DEFAULT_WEIGHT,
             chained: false,
+            server_signing: false,
         }
     }
 }

src/queue/priority.rs

@@ -113,16 +113,19 @@ mod priority_tests {
                     name: "default".into(),
                     weight: 1,
                     chained: false,
+                    server_signing: false,
                 },
                 Config {
                     name: "banana".into(),
                     weight: 2,
                     chained: false,
+                    server_signing: false,
                 },
                 Config {
                     name: "orange".into(),
                     weight: 1,
                     chained: false,
+                    server_signing: false,
                 },
             ]),
         );
@@ -155,6 +158,7 @@ mod priority_tests {
                 name: "default".into(),
                 weight: 1,
                 chained: false,
+                server_signing: false,
             }]),
         );
 

src/server/mod.rs

@@ -7,14 +7,14 @@ use tonic::transport::Server;
 use tracing::{error, info};
 
 use crate::{
-    ledger::u5c::U5cDataAdapter, queue::chaining::TxChaining,
-    storage::sqlite::SqliteTransaction,
+    ledger::u5c::U5cDataAdapter, queue::chaining::TxChaining, storage::sqlite::SqliteTransaction,
+    Config as BorosConfig,
 };
 
 mod submit;
 
 pub async fn run(
-    config: Config,
+    config: BorosConfig,
     u5c_adapter: Arc<dyn U5cDataAdapter>,
     tx_storage: Arc<SqliteTransaction>,
     tx_chaining: Arc<TxChaining>,
@@ -30,19 +30,20 @@ pub async fn run(
             Arc::clone(&tx_storage),
             Arc::clone(&tx_chaining),
             Arc::clone(&u5c_adapter),
+            config.queues,
         );
         let submit_service =
             spec::submit::submit_service_server::SubmitServiceServer::new(submit_service);
 
         info!(
-            address = config.listen_address.to_string(),
+            address = config.server.listen_address.to_string(),
             "GRPC server running"
         );
 
         let result = Server::builder()
             .add_service(reflection)
             .add_service(submit_service)
-            .serve(config.listen_address)
+            .serve(config.server.listen_address)
             .await;
 
         if let Err(error) = result {