/// Performs the server side TLS hand shake processing
Performs the server side TLS hand shake processing
Performs the server side TLS hand shake processing
async fn tls_server_handshake(
        &self,
        client_type: TransportClientType,
        flow_tag: FlowTag,
        stream: TcpStream,
    ) -> Result<(), TransportErrorCode> {
        let local_addr = Self::sock_addr(stream.local_addr())?;
        let peer_addr = Self::sock_addr(stream.peer_addr())?;
        let allowed_clients = {
            let allowed_clients = self.allowed_clients.read().unwrap().clone();
            match AllowedClients::new_with_nodes(allowed_clients) {
                Err(e) => {
                    self.control_plane_metrics
                        .tcp_server_handshake_failed
                        .with_label_values(&[&flow_tag.to_string()])
                        .inc();
                    warn!(
                        every_n_seconds => 30,
                        self.log,
                        "ControlPlane::tls_server_handshake() no allowed clients: failed local_addr = {:?} node_id = {:?}, peer_addr = {:?}, error = {:?}",
                     local_addr,
                     self.node_id,
                     peer_addr,
                     e);
                    return Err(TransportErrorCode::PeerTlsInfoNotFound);
                }
                Ok(allowed_clients) => allowed_clients,
            }
        };
        let registry_version = *self.registry_version.read().unwrap();
        let ret = tokio::time::timeout(
            Duration::from_secs(TLS_HANDSHAKE_TIMEOUT_SECONDS),
            self.crypto
                .perform_tls_server_handshake(stream, allowed_clients, registry_version),
        )
        .await;
        if ret.is_err() {
            self.control_plane_metrics
                .tcp_server_handshake_failed
                .with_label_values(&[&flow_tag.to_string()])
                .inc();
            warn!(
                every_n_seconds => 30,
                self.log,
                    "ControlPlane::tls_server_handshake() timed out: \
                      local_addr = {:?}, node_id = {:?}, peer_addr = {:?}",
                     local_addr,
                     self.node_id,
                     peer_addr);
            return Err(TransportErrorCode::TimeoutExpired);
        }
        let (tls_stream, authenticated_peer) = match ret.unwrap() {
            Ok((tls_stream, peer_id)) => (tls_stream.split(), peer_id),
            Err(e) => {
                self.control_plane_metrics
                    .tcp_server_handshake_failed
                    .with_label_values(&[&flow_tag.to_string()])
                    .inc();
                warn!(
                    every_n_seconds => 30,
                    self.log,
                    "ControlPlane::tls_server_handshake(): failed local_addr = {:?} \
                     node_id = {:?}, peer_addr = {:?}, error = {:?}",
                    local_addr,
                    self.node_id,
                    peer_addr,
                    e
                );
                return Err(TransportErrorCode::PeerTlsInfoNotFound);
            }
        };
        let (tls_reader, tls_writer) = tls_stream;
        let peer_id = match authenticated_peer {
            AuthenticatedPeer::Node(node_id) => node_id,
            AuthenticatedPeer::Cert(_) => {
                warn!(
                    every_n_seconds => 10,
                    self.log,
                    "ControlPlane::tls_server_handshake(): failed local_addr = {:?} \
                        node_id = {:?}, peer_addr = {:?}, error = cert instead of node id",
                    local_addr,
                    self.node_id,
                    peer_addr
                );
                return Err(TransportErrorCode::PeerTlsInfoNotFound);
            }
        };
        self.process_handshake_result(
            peer_id,
            ConnectionRole::Server,
            client_type,
            flow_tag,
            local_addr,
            peer_addr,
            tls_reader,
            tls_writer,
        )
        .await
        .map(|_| {
            self.control_plane_metrics
                .tcp_server_handshake_success
                .with_label_values(&[&flow_tag.to_string()])
                .inc()
        })
    }
async fn tls_server_handshake ( & self , client_type : TransportClientType , flow_tag : FlowTag , stream : TcpStream , ) -> Result < ( ) , TransportErrorCode > { let local_addr = Self :: sock_addr ( stream . local_addr ( ) ) ? ; let peer_addr = Self :: sock_addr ( stream . peer_addr ( ) ) ? ; let allowed_clients = { let allowed_clients = self . allowed_clients . read ( ) . unwrap ( ) . clone ( ) ; match AllowedClients :: new_with_nodes ( allowed_clients ) { Err ( e ) => { self . control_plane_metrics . tcp_server_handshake_failed . with_label_values ( & [ & flow_tag . to_string ( ) ] ) . inc ( ) ; warn ! ( every_n_seconds 30 self log " " local_addr self node_id peer_addr e ) ; return Err ( TransportErrorCode :: PeerTlsInfoNotFound ) ; } Ok ( allowed_clients ) => allowed_clients , } } ; let registry_version = * self . registry_version . read ( ) . unwrap ( ) ; let ret = tokio :: time :: timeout ( Duration :: from_secs ( TLS_HANDSHAKE_TIMEOUT_SECONDS ) , self . crypto . perform_tls_server_handshake ( stream , allowed_clients , registry_version ) , ) . await ; if ret . is_err ( ) { self . control_plane_metrics . tcp_server_handshake_failed . with_label_values ( & [ & flow_tag . to_string ( ) ] ) . inc ( ) ; warn ! ( every_n_seconds 30 self log " \
 " local_addr self node_id peer_addr ) ; return Err ( TransportErrorCode :: TimeoutExpired ) ; } let ( tls_stream , authenticated_peer ) = match ret . unwrap ( ) { Ok ( ( tls_stream , peer_id ) ) => ( tls_stream . split ( ) , peer_id ) , Err ( e ) => { self . control_plane_metrics . tcp_server_handshake_failed . with_label_values ( & [ & flow_tag . to_string ( ) ] ) . inc ( ) ; warn ! ( every_n_seconds 30 self log " \
 " local_addr self node_id peer_addr e ) ; return Err ( TransportErrorCode :: PeerTlsInfoNotFound ) ; } } ; let ( tls_reader , tls_writer ) = tls_stream ; let peer_id = match authenticated_peer { AuthenticatedPeer :: Node ( node_id ) => node_id , AuthenticatedPeer :: Cert ( _ ) => { warn ! ( every_n_seconds 10 self log " \
 " local_addr self node_id peer_addr ) ; return Err ( TransportErrorCode :: PeerTlsInfoNotFound ) ; } } ; self . process_handshake_result ( peer_id , ConnectionRole :: Server , client_type , flow_tag , local_addr , peer_addr , tls_reader , tls_writer , ) . await . map ( | _ | { self . control_plane_metrics . tcp_server_handshake_success . with_label_values ( & [ & flow_tag . to_string ( ) ] ) . inc ( ) } ) }
Performs the server side TLS hand shake processing
Performs the server side TLS hand shake processing
&self
client_type
TransportClientType
flow_tag
FlowTag
stream
TcpStream