feat(rust): switching key negotiation to a double secure channel

implementations/rust/ockam/ockam_api/Cargo.toml

@@ -112,7 +112,6 @@ tracing-error = "0.2.0"
 tracing-opentelemetry = "0.27.0"
 tracing-subscriber = { version = "0.3", features = ["json"] }
 url = "2.5.2"
-zeroize = { version = "1.8.1", features = ["zeroize_derive"] }
 
 ockam_multiaddr = { path = "../ockam_multiaddr", version = "0.66.0", features = ["cbor", "serde"] }
 ockam_transport_core = { path = "../ockam_transport_core", version = "^0.99.0" }

implementations/rust/ockam/ockam_api/src/kafka/key_exchange/controller.rs

@@ -266,7 +266,6 @@ mod test {
     pub fn rekey_rotation() -> ockam_core::Result<()> {
         let runtime = Arc::new(Runtime::new().unwrap());
         let runtime_cloned = runtime.clone();
-        std::env::set_var("OCKAM_LOGGING", "false");
 
         runtime_cloned.block_on(async move {
             let test_body = async move {
@@ -297,24 +296,34 @@ mod test {
                     .get_flow_control_with_spawner(&DefaultAddress::SECURE_CHANNEL_LISTENER.into())
                     .unwrap();
 
+                let test_clock = TestClock::new(0);
+
                 KafkaKeyExchangeListener::create(
+                    test_clock.clone(),
                     &consumer_node.context,
                     consumer_node
                         .node_manager
                         .secure_channels
                         .vault()
                         .encryption_at_rest_vault,
-                    Duration::from_secs(60),
-                    Duration::from_secs(60),
-                    Duration::from_secs(60),
+                    consumer_node
+                        .node_manager
+                        .secure_channels
+                        .vault()
+                        .secure_channel_vault,
+                    consumer_node
+                        .node_manager
+                        .secure_channels
+                        .secure_channel_registry(),
+                    Duration::from_secs(5 * 60),  //rotation
+                    Duration::from_secs(10 * 60), //validity
+                    Duration::from_secs(60),      //rekey
                     &consumer_secure_channel_listener_flow_control_id,
                     AllowAll,
                     AllowAll,
                 )
                 .await?;
 
-                let test_clock = TestClock::new(0);
-
                 let destination = consumer_node.listen_address().await.multi_addr().unwrap();
                 let producer_secure_channel_controller = create_secure_channel_controller(
                     test_clock.clone(),
@@ -355,7 +364,10 @@ mod test {
                     .await?;
 
                 assert_eq!(third_key.rekey_counter, 1);
-                assert_eq!(first_key.secret_key_handle, third_key.secret_key_handle);
+                assert_eq!(
+                    first_key.key_identifier_for_consumer,
+                    third_key.key_identifier_for_consumer
+                );
 
                 // 04:00 - yet another rekey should happen, but no rotation
                 test_clock.add_seconds(60 * 3);
@@ -365,7 +377,10 @@ mod test {
                     .await?;
 
                 assert_eq!(fourth_key.rekey_counter, 2);
-                assert_eq!(first_key.secret_key_handle, fourth_key.secret_key_handle);
+                assert_eq!(
+                    first_key.key_identifier_for_consumer,
+                    fourth_key.key_identifier_for_consumer
+                );
 
                 // 05:00 - the default duration of the key is 10 minutes,
                 // but the rotation should happen after 5 minutes
@@ -375,7 +390,10 @@ mod test {
                     .get_or_exchange_key(&mut producer_node.context, "topic_name")
                     .await?;
 
-                assert_ne!(third_key.secret_key_handle, fifth_key.secret_key_handle);
+                assert_ne!(
+                    third_key.key_identifier_for_consumer,
+                    fifth_key.key_identifier_for_consumer
+                );
                 assert_eq!(fifth_key.rekey_counter, 0);
 
                 // Now let's simulate a failure to rekey by shutting down the consumer
@@ -389,7 +407,10 @@ mod test {
                     .await?;
 
                 assert_eq!(sixth_key.rekey_counter, 1);
-                assert_eq!(fifth_key.secret_key_handle, sixth_key.secret_key_handle);
+                assert_eq!(
+                    fifth_key.key_identifier_for_consumer,
+                    sixth_key.key_identifier_for_consumer
+                );
 
                 // 10:00 - Rotation fails, but the existing key is still valid
                 // and needs to be rekeyed
@@ -400,7 +421,10 @@ mod test {
                     .await?;
 
                 assert_eq!(seventh_key.rekey_counter, 2);
-                assert_eq!(fifth_key.secret_key_handle, seventh_key.secret_key_handle);
+                assert_eq!(
+                    fifth_key.key_identifier_for_consumer,
+                    seventh_key.key_identifier_for_consumer
+                );
 
                 // 15:00 - Rotation fails, and the existing key is no longer valid
                 test_clock.add_seconds(60 * 5);

implementations/rust/ockam/ockam_api/src/kafka/key_exchange/listener.rs

@@ -1,37 +1,42 @@
 use crate::DefaultAddress;
 use minicbor::{CborLen, Decode, Encode};
-use ockam::identity::TimestampInSeconds;
+use ockam::identity::{
+    SecureChannelApiRequest, SecureChannelApiResponse, SecureChannelRegistry, TimestampInSeconds,
+};
+use ockam_core::compat::clock::Clock;
 use ockam_core::flow_control::FlowControlId;
 use ockam_core::{
-    async_trait, Address, Decodable, Encodable, Encoded, IncomingAccessControl, Message,
+    async_trait, route, Address, Decodable, Encodable, Encoded, IncomingAccessControl, Message,
     OutgoingAccessControl, Routed, Worker,
 };
 use ockam_node::{Context, WorkerBuilder};
-use ockam_vault::VaultForEncryptionAtRest;
-use rand::Rng;
+use ockam_vault::{VaultForEncryptionAtRest, VaultForSecureChannels};
 use std::sync::Arc;
 use std::time::Duration;
 
 pub(crate) struct KafkaKeyExchangeListener {
     encryption_at_rest: Arc<dyn VaultForEncryptionAtRest>,
+    secure_channel_vault: Arc<dyn VaultForSecureChannels>,
+    secure_channel_registry: SecureChannelRegistry,
     rekey_period: Duration,
     key_validity: Duration,
     key_rotation: Duration,
+    clock: Box<dyn Clock>,
 }
 
 #[derive(Debug, CborLen, Encode, Decode)]
 #[rustfmt::skip]
 pub(crate) struct KeyExchangeRequest {
+    #[n(1)] pub local_decryptor_address: Address,
 }
 
 #[derive(Debug, CborLen, Encode, Decode)]
 #[rustfmt::skip]
 pub(crate) struct KeyExchangeResponse {
     #[n(0)] pub key_identifier_for_consumer: Vec<u8>,
-    #[n(1)] pub secret_key: [u8; 32],
-    #[n(2)] pub valid_until: TimestampInSeconds,
-    #[n(3)] pub rotate_after: TimestampInSeconds,
-    #[n(4)] pub rekey_period: Duration,
+    #[n(1)] pub valid_until: TimestampInSeconds,
+    #[n(2)] pub rotate_after: TimestampInSeconds,
+    #[n(3)] pub rekey_period: Duration,
 }
 
 impl Encodable for KeyExchangeRequest {
@@ -70,14 +75,43 @@ impl Worker for KafkaKeyExchangeListener {
         context: &mut Self::Context,
         message: Routed<Self::Message>,
     ) -> ockam_core::Result<()> {
-        let mut secret_key = [0u8; 32];
-        rand::thread_rng().fill(&mut secret_key[..]);
-        let handle = self
-            .encryption_at_rest
-            .import_aead_key(secret_key.to_vec())
-            .await?;
+        let request: KeyExchangeRequest = minicbor::decode(message.payload())?;
+        let local_decryptor = Address::from_string(request.local_decryptor_address);
+
+        let entry = self
+            .secure_channel_registry
+            .get_channel_by_decryptor_address(&local_decryptor);
+        let handle = match entry {
+            None => {
+                warn!("No secure channel found for local decryptor {local_decryptor}",);
+                return Ok(());
+            }
+            Some(entry) => {
+                let response: SecureChannelApiResponse = context
+                    .send_and_receive(
+                        route![entry.decryptor_api_address().clone()],
+                        SecureChannelApiRequest::ExtractKey,
+                    )
+                    .await?;
+
+                let key_identifier = match response {
+                    SecureChannelApiResponse::Ok(key_identifier) => key_identifier,
+                    SecureChannelApiResponse::Err(error) => {
+                        error!("Error extracting key: {error}");
+                        return Ok(());
+                    }
+                };
+
+                let secret = self
+                    .secure_channel_vault
+                    .export_rekey(&key_identifier)
+                    .await?;
+
+                self.encryption_at_rest.import_aead_key(secret).await?
+            }
+        };
 
-        let now = TimestampInSeconds(ockam_core::compat::time::now()?);
+        let now = TimestampInSeconds(self.clock.now()?);
         let valid_until = now + self.key_validity;
         let rotate_after = now + self.key_rotation;
 
@@ -86,7 +120,6 @@ impl Worker for KafkaKeyExchangeListener {
                 message.return_route().clone(),
                 KeyExchangeResponse {
                     key_identifier_for_consumer: handle.into_vec(),
-                    secret_key,
                     valid_until,
                     rotate_after,
                     rekey_period: self.rekey_period,
@@ -101,8 +134,11 @@ impl Worker for KafkaKeyExchangeListener {
 impl KafkaKeyExchangeListener {
     #[allow(clippy::too_many_arguments)]
     pub async fn create(
+        clock: impl Clock,
         context: &Context,
         encryption_at_rest: Arc<dyn VaultForEncryptionAtRest>,
+        secure_channel_vault: Arc<dyn VaultForSecureChannels>,
+        secure_channel_registry: SecureChannelRegistry,
         key_rotation: Duration,
         key_validity: Duration,
         rekey_period: Duration,
@@ -117,9 +153,12 @@ impl KafkaKeyExchangeListener {
 
         WorkerBuilder::new(KafkaKeyExchangeListener {
             encryption_at_rest,
+            secure_channel_vault,
             key_rotation,
             key_validity,
             rekey_period,
+            secure_channel_registry,
+            clock: Box::new(clock),
         })
         .with_address(address)
         .with_incoming_access_control(incoming_access_control)

implementations/rust/ockam/ockam_api/src/kafka/key_exchange/secure_channels.rs

@@ -5,9 +5,9 @@ use crate::kafka::key_exchange::controller::{
 use crate::kafka::key_exchange::listener::{KeyExchangeRequest, KeyExchangeResponse};
 use crate::kafka::ConsumerResolution;
 use crate::DefaultAddress;
-use ockam::identity::TimestampInSeconds;
+use ockam::identity::{SecureChannelApiRequest, SecureChannelApiResponse, TimestampInSeconds};
 use ockam_core::errcode::{Kind, Origin};
-use ockam_core::{Error, Result};
+use ockam_core::{route, Error, Result};
 use ockam_multiaddr::proto::{Secure, Service};
 use ockam_multiaddr::MultiAddr;
 use ockam_node::{Context, MessageSendReceiveOptions};
@@ -35,6 +35,72 @@ impl KafkaKeyExchangeControllerImpl {
         destination.push_back(Secure::new(DefaultAddress::SECURE_CHANNEL_LISTENER))?;
         destination.push_back(Service::new(DefaultAddress::KAFKA_CUSTODIAN))?;
         if let Some(node_manager) = inner.node_manager.upgrade() {
+            // create a second secure channel to be used for key exchange
+            let (aead_secret_key_handle, their_decryptor_address) = {
+                let key_exchange_connection = node_manager
+                    .make_connection(context, &destination, node_manager.identifier(), None, None)
+                    .await?;
+
+                let encryptor_address = key_exchange_connection
+                    .secure_channel_encryptors
+                    .first()
+                    .expect("encryptor should be present");
+
+                let entry = node_manager
+                    .secure_channels
+                    .secure_channel_registry()
+                    .get_channel_by_encryptor_address(encryptor_address)
+                    .expect("channel should be present");
+
+                if !inner
+                    .consumer_policy_access_control
+                    .is_identity_authorized(entry.their_id())
+                    .await?
+                {
+                    key_exchange_connection
+                        .close(context, &node_manager)
+                        .await?;
+                    return Err(Error::new(
+                        Origin::Channel,
+                        Kind::Invalid,
+                        "Consumer is not authorized to use the secure channel",
+                    ));
+                }
+
+                let response: SecureChannelApiResponse = context
+                    .send_and_receive(
+                        route![entry.encryptor_api_address().clone()],
+                        SecureChannelApiRequest::ExtractKey,
+                    )
+                    .await?;
+
+                match response {
+                    SecureChannelApiResponse::Ok(secret_handle) => {
+                        let secret = node_manager
+                            .secure_channels
+                            .vault()
+                            .secure_channel_vault
+                            .export_rekey(&secret_handle)
+                            .await?;
+
+                        key_exchange_connection
+                            .close(context, &node_manager)
+                            .await?;
+
+                        (
+                            self.encryption_at_rest.import_aead_key(secret).await?,
+                            entry.their_decryptor_address(),
+                        )
+                    }
+                    SecureChannelApiResponse::Err(error) => {
+                        key_exchange_connection
+                            .close(context, &node_manager)
+                            .await?;
+                        return Err(error);
+                    }
+                }
+            };
+
             let connection = node_manager
                 .make_connection(context, &destination, node_manager.identifier(), None, None)
                 .await?;
@@ -52,14 +118,17 @@ impl KafkaKeyExchangeControllerImpl {
 
             let route = connection.route()?;
             let response: KeyExchangeResponse = context
-                .send_and_receive_extended(route, KeyExchangeRequest {}, send_and_receive_options)
+                .send_and_receive_extended(
+                    route,
+                    KeyExchangeRequest {
+                        local_decryptor_address: their_decryptor_address,
+                    },
+                    send_and_receive_options,
+                )
                 .await?
                 .into_body()?;
 
-            let aead_secret_key_handle = self
-                .encryption_at_rest
-                .import_aead_key(response.secret_key.to_vec())
-                .await?;
+            connection.close(context, &node_manager).await?;
 
             Ok(ExchangedKey {
                 secret_key_handler: aead_secret_key_handle,

implementations/rust/ockam/ockam_api/src/nodes/service/kafka_services.rs

@@ -23,6 +23,7 @@ use ockam::{Address, Context, Result};
 use ockam_abac::PolicyExpression;
 use ockam_abac::{Action, Resource, ResourceType};
 use ockam_core::api::{Error, Response};
+use ockam_core::compat::clock::ProductionClock;
 use ockam_core::compat::rand::random_string;
 use ockam_core::flow_control::FlowControls;
 use ockam_core::route;
@@ -197,8 +198,11 @@ impl InMemoryNode {
 
         // TODO: remove key exchange from inlets
         KafkaKeyExchangeListener::create(
+            ProductionClock,
             context,
             vault.encryption_at_rest_vault,
+            self.secure_channels.vault().secure_channel_vault,
+            self.secure_channels.secure_channel_registry(),
             std::time::Duration::from_secs(60 * 60 * 24),
             std::time::Duration::from_secs(60 * 60 * 30),
             std::time::Duration::from_secs(60 * 60),
@@ -395,8 +399,11 @@ impl InMemoryNode {
             .await?;
 
         KafkaKeyExchangeListener::create(
+            ProductionClock,
             context,
             vault.encryption_at_rest_vault,
+            self.secure_channels.vault().secure_channel_vault,
+            self.secure_channels.secure_channel_registry(),
             request.key_rotation,
             request.key_validity,
             request.rekey_period,