feat(p2p/webrtc): authenticate connection after it has been established

Author: binier

node/common/src/service/p2p.rs

@@ -6,6 +6,7 @@ use node::{
     p2p::{
         connection::outgoing::P2pConnectionOutgoingInitOpts,
         identity::{EncryptableType, PublicKey},
+        webrtc::ConnectionAuth,
         PeerId,
     },
 };
@@ -55,6 +56,28 @@ impl webrtc::P2pServiceWebrtc for NodeService {
     ) -> Result<T, ()> {
         self.p2p.sec_key.decrypt(other_pk, encrypted)
     }
+
+    fn auth_encrypt_and_send(
+        &mut self,
+        peer_id: PeerId,
+        other_pub_key: &PublicKey,
+        auth: ConnectionAuth,
+    ) {
+        let encrypted = auth.encrypt(&self.p2p.sec_key, other_pub_key, &mut self.rng);
+        if let Some(peer) = self.peers().get(&peer_id) {
+            let _ = peer
+                .cmd_sender
+                .send(webrtc::PeerCmd::ConnectionAuthorizationSend(encrypted));
+        }
+    }
+
+    fn auth_decrypt(
+        &mut self,
+        other_pub_key: &PublicKey,
+        auth: node::p2p::webrtc::ConnectionAuthEncrypted,
+    ) -> Option<ConnectionAuth> {
+        auth.decrypt(&self.p2p.sec_key, other_pub_key)
+    }
 }
 
 impl webrtc_with_libp2p::P2pServiceWebrtcWithLibp2p for NodeService {

node/src/action_kind.rs

@@ -318,7 +318,8 @@ pub enum ActionKind {
     P2pConnectionIncomingLibp2pReceived,
     P2pConnectionIncomingSuccess,
     P2pConnectionIncomingTimeout,
-    P2pConnectionIncomingEffectfulAnswerSend,
+    P2pConnectionIncomingEffectfulConnectionAuthorizationDecryptAndCheck,
+    P2pConnectionIncomingEffectfulConnectionAuthorizationEncryptAndSend,
     P2pConnectionIncomingEffectfulInit,
     P2pConnectionOutgoingAnswerRecvError,
     P2pConnectionOutgoingAnswerRecvPending,
@@ -338,6 +339,8 @@ pub enum ActionKind {
     P2pConnectionOutgoingSuccess,
     P2pConnectionOutgoingTimeout,
     P2pConnectionOutgoingEffectfulAnswerSet,
+    P2pConnectionOutgoingEffectfulConnectionAuthorizationDecryptAndCheck,
+    P2pConnectionOutgoingEffectfulConnectionAuthorizationEncryptAndSend,
     P2pConnectionOutgoingEffectfulInit,
     P2pConnectionOutgoingEffectfulOfferSend,
     P2pConnectionOutgoingEffectfulRandomInit,
@@ -657,7 +660,7 @@ pub enum ActionKind {
 }
 
 impl ActionKind {
-    pub const COUNT: u16 = 545;
+    pub const COUNT: u16 = 548;
 }
 
 impl std::fmt::Display for ActionKind {
@@ -1871,6 +1874,12 @@ impl ActionKindGet for P2pConnectionOutgoingEffectfulAction {
             Self::Init { .. } => ActionKind::P2pConnectionOutgoingEffectfulInit,
             Self::OfferSend { .. } => ActionKind::P2pConnectionOutgoingEffectfulOfferSend,
             Self::AnswerSet { .. } => ActionKind::P2pConnectionOutgoingEffectfulAnswerSet,
+            Self::ConnectionAuthorizationEncryptAndSend { .. } => {
+                ActionKind::P2pConnectionOutgoingEffectfulConnectionAuthorizationEncryptAndSend
+            }
+            Self::ConnectionAuthorizationDecryptAndCheck { .. } => {
+                ActionKind::P2pConnectionOutgoingEffectfulConnectionAuthorizationDecryptAndCheck
+            }
         }
     }
 }
@@ -1879,7 +1888,12 @@ impl ActionKindGet for P2pConnectionIncomingEffectfulAction {
     fn kind(&self) -> ActionKind {
         match self {
             Self::Init { .. } => ActionKind::P2pConnectionIncomingEffectfulInit,
-            Self::AnswerSend { .. } => ActionKind::P2pConnectionIncomingEffectfulAnswerSend,
+            Self::ConnectionAuthorizationEncryptAndSend { .. } => {
+                ActionKind::P2pConnectionIncomingEffectfulConnectionAuthorizationEncryptAndSend
+            }
+            Self::ConnectionAuthorizationDecryptAndCheck { .. } => {
+                ActionKind::P2pConnectionIncomingEffectfulConnectionAuthorizationDecryptAndCheck
+            }
         }
     }
 }

node/src/event_source/event_source_effects.rs

@@ -184,12 +184,16 @@ pub fn event_source_effects<S: Service>(store: &mut Store<S>, action: EventSourc
                                 error,
                             });
                         }
-                        Ok(_) => {
-                            let _ = store
-                                .dispatch(P2pConnectionOutgoingAction::FinalizeSuccess { peer_id })
-                                || store.dispatch(P2pConnectionIncomingAction::FinalizeSuccess {
+                        Ok(auth) => {
+                            let _ = store.dispatch(P2pConnectionOutgoingAction::FinalizeSuccess {
+                                peer_id,
+                                remote_auth: Some(auth.clone()),
+                            }) || store.dispatch(
+                                P2pConnectionIncomingAction::FinalizeSuccess {
                                     peer_id,
-                                });
+                                    remote_auth: auth.clone(),
+                                },
+                            );
                         }
                     },
                     P2pConnectionEvent::Closed(peer_id) => {

node/testing/src/node/rust/event.rs

@@ -1,5 +1,5 @@
 use node::{
-    p2p::{P2pConnectionEvent, P2pEvent, PeerId},
+    p2p::{webrtc::ConnectionAuthEncrypted, P2pConnectionEvent, P2pEvent, PeerId},
     rpc::{RpcId, RpcRequest},
 };
 use serde::{Deserialize, Serialize};
@@ -10,7 +10,7 @@ pub use node::event_source::Event;
 pub enum NonDeterministicEvent {
     /// Non-deterministic because libp2p kademlia initiates connections
     /// without state machine knowing about it.
-    P2pConnectionFinalized(PeerId, Result<(), String>),
+    P2pConnectionFinalized(PeerId, Result<ConnectionAuthEncrypted, String>),
     P2pConnectionClosed(PeerId),
 
     RpcReadonly(RpcId, Box<RpcRequest>),

node/testing/src/service/mod.rs

@@ -375,6 +375,24 @@ impl P2pServiceWebrtc for NodeTestingService {
     ) -> Result<T, ()> {
         self.real.decrypt(other_pub_key, encrypted)
     }
+
+    fn auth_encrypt_and_send(
+        &mut self,
+        peer_id: PeerId,
+        other_pub_key: &node::p2p::identity::PublicKey,
+        auth: webrtc::ConnectionAuth,
+    ) {
+        self.real
+            .auth_encrypt_and_send(peer_id, other_pub_key, auth)
+    }
+
+    fn auth_decrypt(
+        &mut self,
+        other_pub_key: &node::p2p::identity::PublicKey,
+        auth: webrtc::ConnectionAuthEncrypted,
+    ) -> Option<webrtc::ConnectionAuth> {
+        self.real.auth_decrypt(other_pub_key, auth)
+    }
 }
 
 impl P2pServiceWebrtcWithLibp2p for NodeTestingService {

p2p/src/connection/incoming/p2p_connection_incoming_actions.rs

@@ -49,6 +49,7 @@ pub enum P2pConnectionIncomingAction {
     /// Incoming connection finalized.
     FinalizeSuccess {
         peer_id: PeerId,
+        remote_auth: webrtc::ConnectionAuthEncrypted,
     },
     /// Timeout establishing incoming connection.
     Timeout {
@@ -86,7 +87,7 @@ impl P2pConnectionIncomingAction {
             | Self::AnswerSendSuccess { peer_id }
             | Self::FinalizePending { peer_id }
             | Self::FinalizeError { peer_id, .. }
-            | Self::FinalizeSuccess { peer_id }
+            | Self::FinalizeSuccess { peer_id, .. }
             | Self::Timeout { peer_id }
             | Self::Error { peer_id, .. }
             | Self::Success { peer_id }
@@ -172,7 +173,7 @@ impl redux::EnablingCondition<P2pState> for P2pConnectionIncomingAction {
                     )
                 })
             }
-            P2pConnectionIncomingAction::FinalizeSuccess { peer_id } => {
+            P2pConnectionIncomingAction::FinalizeSuccess { peer_id, .. } => {
                 state.peers.get(peer_id).map_or(false, |peer| {
                     matches!(
                         &peer.status,
@@ -198,6 +199,9 @@ impl redux::EnablingCondition<P2pState> for P2pConnectionIncomingAction {
                         P2pConnectionIncomingError::FinalizeError(_) => {
                             matches!(s, P2pConnectionIncomingState::FinalizePending { .. })
                         }
+                        P2pConnectionIncomingError::ConnectionAuthError => {
+                            matches!(s, P2pConnectionIncomingState::FinalizeSuccess { .. })
+                        }
                         P2pConnectionIncomingError::Timeout => true,
                     },
                     _ => false,

p2p/src/connection/incoming/p2p_connection_incoming_reducer.rs

@@ -174,10 +174,7 @@ impl P2pConnectionIncomingState {
 
                 match signaling {
                     IncomingSignalingMethod::Http => {
-                        dispatcher.push(P2pConnectionIncomingEffectfulAction::AnswerSend {
-                            peer_id,
-                            answer: answer.clone(),
-                        });
+                        // Will respond to the rpc below.
                     }
                     IncomingSignalingMethod::P2p { relay_peer_id } => {
                         dispatcher.push(P2pChannelsSignalingExchangeAction::AnswerSend {
@@ -243,19 +240,25 @@ impl P2pConnectionIncomingState {
                     ..
                 } = state
                 {
+                    let auth = offer.conn_auth(answer);
+                    let other_pub_key = offer.identity_pub_key.clone();
+
                     *state = Self::FinalizePending {
                         time: meta.time(),
                         signaling: *signaling,
                         offer: offer.clone(),
                         answer: answer.clone(),
                         rpc_id: rpc_id.take(),
                     };
+
+                    state_context.into_dispatcher().push(P2pConnectionIncomingEffectfulAction::ConnectionAuthorizationEncryptAndSend { peer_id, other_pub_key, auth });
                 } else {
                     bug_condition!(
                         "Invalid state for `P2pConnectionIncomingAction::FinalizePending`: {:?}",
                         state
                     );
                 }
+
                 Ok(())
             }
             P2pConnectionIncomingAction::FinalizeError { error, .. } => {
@@ -266,35 +269,48 @@ impl P2pConnectionIncomingState {
                 });
                 Ok(())
             }
-            P2pConnectionIncomingAction::FinalizeSuccess { .. } => {
+            P2pConnectionIncomingAction::FinalizeSuccess { remote_auth, .. } => {
                 let state = p2p_state
                     .incoming_peer_connection_mut(&peer_id)
-                    .ok_or_else(|| format!("Invalid state for: {:?}", action))?;
-                if let Self::FinalizePending {
+                    .ok_or_else(|| {
+                        "Invalid state for: P2pConnectionIncomingAction::FinalizeSuccess".to_owned()
+                    })?;
+                let (expected_auth, other_pub_key) = if let Self::FinalizePending {
                     signaling,
                     offer,
                     answer,
                     rpc_id,
                     ..
                 } = state
                 {
+                    let expected_auth = offer.conn_auth(answer);
+                    let other_pub_key = offer.identity_pub_key.clone();
                     *state = Self::FinalizeSuccess {
                         time: meta.time(),
                         signaling: *signaling,
                         offer: offer.clone(),
                         answer: answer.clone(),
                         rpc_id: rpc_id.take(),
                     };
+                    (expected_auth, other_pub_key)
                 } else {
                     bug_condition!(
                         "Invalid state for `P2pConnectionIncomingAction::FinalizeSuccess`: {:?}",
                         state
                     );
                     return Ok(());
-                }
+                };
 
                 let dispatcher = state_context.into_dispatcher();
-                dispatcher.push(P2pConnectionIncomingAction::Success { peer_id });
+
+                dispatcher.push(
+                    P2pConnectionIncomingEffectfulAction::ConnectionAuthorizationDecryptAndCheck {
+                        peer_id,
+                        other_pub_key,
+                        expected_auth,
+                        auth: remote_auth,
+                    },
+                );
                 Ok(())
             }
             P2pConnectionIncomingAction::Timeout { .. } => {

p2p/src/connection/incoming/p2p_connection_incoming_state.rs

@@ -127,6 +127,8 @@ pub enum P2pConnectionIncomingError {
     SdpCreateError(String),
     #[error("finalization error: {0}")]
     FinalizeError(String),
+    #[error("connection authentication failed")]
+    ConnectionAuthError,
     #[error("timeout error")]
     Timeout,
 }

p2p/src/connection/incoming_effectful/p2p_connection_incoming_effectful_actions.rs

@@ -1,6 +1,8 @@
 use crate::{
     connection::{incoming::P2pConnectionIncomingInitOpts, P2pConnectionEffectfulAction},
-    webrtc, P2pState, PeerId,
+    identity::PublicKey,
+    webrtc::{ConnectionAuth, ConnectionAuthEncrypted},
+    P2pState, PeerId,
 };
 use openmina_core::ActionEvent;
 use serde::{Deserialize, Serialize};
@@ -10,9 +12,16 @@ use serde::{Deserialize, Serialize};
 pub enum P2pConnectionIncomingEffectfulAction {
     /// Incoming connection is initialized.
     Init { opts: P2pConnectionIncomingInitOpts },
-    AnswerSend {
+    ConnectionAuthorizationEncryptAndSend {
         peer_id: PeerId,
-        answer: Box<webrtc::Answer>,
+        other_pub_key: PublicKey,
+        auth: ConnectionAuth,
+    },
+    ConnectionAuthorizationDecryptAndCheck {
+        peer_id: PeerId,
+        other_pub_key: PublicKey,
+        expected_auth: ConnectionAuth,
+        auth: ConnectionAuthEncrypted,
     },
 }
 

p2p/src/connection/incoming_effectful/p2p_connection_incoming_effectful_effects.rs

@@ -1,7 +1,10 @@
 use redux::ActionMeta;
 
 use super::P2pConnectionIncomingEffectfulAction;
-use crate::connection::{incoming::P2pConnectionIncomingAction, P2pConnectionService};
+use crate::connection::{
+    incoming::{P2pConnectionIncomingAction, P2pConnectionIncomingError},
+    P2pConnectionService,
+};
 
 impl P2pConnectionIncomingEffectfulAction {
     pub fn effects<Store, S>(self, _meta: &ActionMeta, store: &mut Store)
@@ -15,8 +18,33 @@ impl P2pConnectionIncomingEffectfulAction {
                 store.service().incoming_init(peer_id, *opts.offer);
                 store.dispatch(P2pConnectionIncomingAction::AnswerSdpCreatePending { peer_id });
             }
-            P2pConnectionIncomingEffectfulAction::AnswerSend { peer_id, answer } => {
-                store.service().set_answer(peer_id, *answer);
+            P2pConnectionIncomingEffectfulAction::ConnectionAuthorizationEncryptAndSend {
+                peer_id,
+                other_pub_key,
+                auth,
+            } => {
+                store
+                    .service()
+                    .auth_encrypt_and_send(peer_id, &other_pub_key, auth);
+            }
+            P2pConnectionIncomingEffectfulAction::ConnectionAuthorizationDecryptAndCheck {
+                peer_id,
+                other_pub_key,
+                expected_auth,
+                auth,
+            } => {
+                if store
+                    .service()
+                    .auth_decrypt(&other_pub_key, auth)
+                    .map_or(false, |remote_auth| remote_auth == expected_auth)
+                {
+                    store.dispatch(P2pConnectionIncomingAction::Success { peer_id });
+                } else {
+                    store.dispatch(P2pConnectionIncomingAction::Error {
+                        peer_id,
+                        error: P2pConnectionIncomingError::ConnectionAuthError,
+                    });
+                }
             }
         }
     }