General Upgrade, NEW: WebServers, Better Docs

client/Cargo.lock

@@ -4,4 +4,4 @@ version = 4
 
 [[package]]
 name = "client"
-version = "0.1.0"
+version = "0.2.0"

client/Cargo.toml

@@ -1,20 +1,19 @@
 [package]
 name = "mtp-client"
-version = "0.1.0"
+version = "0.2.0"
 edition = "2024"
 
 [dependencies]
-mtp-common = { version = "0.1.0", path = "../common" }
-mtp-codec = { version = "0.1.0", path = "../codec" }
-mtp-transport = { version = "0.1.0", path = "../transport" }
-mtp-crypto = { version = "0.1.0", path = "../crypto", optional = true }
+mtp-common = { version = "0.2.0", path = "../common" }
+mtp-codec = { version = "0.2.0", path = "../codec" }
+mtp-transport = { version = "0.2.0", path = "../transport" }
+mtp-crypto = { version = "0.2.0", path = "../crypto", optional = true }
 rand = "0.8"
 tokio = { version = "1", features = ["rt", "sync", "time"] }
-tracing = "0.1"
 
 [dev-dependencies]
-mtp-host = { version = "0.1.0", path = "../host" }
-mtp-transport = { version = "0.1.0", path = "../transport", features = ["host"] }
+mtp-host = { version = "0.2.0", path = "../host" }
+mtp-transport = { version = "0.2.0", path = "../transport", features = ["host"] }
 rcgen = "0.14"
 
 [features]

client/src/config.rs

@@ -0,0 +1,115 @@
+use tokio::time::Duration;
+
+pub use mtp_transport::Policy;
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum ClientTlsConfig {
+    SystemRoots,
+    PinnedPem(Vec<u8>),
+}
+
+pub struct ClientConfig {
+    pub url: String,
+    pub tls: ClientTlsConfig,
+    pub client_id: u64,
+    pub description: Option<String>,
+    pub policy: Policy,
+    pub ping_interval: Duration,
+    pub ping_jitter: Option<Duration>,
+    pub max_missed_pings: usize,
+    pub ping_timestamp: bool,
+    pub request_timeout: Duration,
+    #[cfg(feature = "crypto")]
+    pub auth_timeout: Duration,
+    #[cfg(feature = "crypto")]
+    pub require_pq: bool,
+}
+
+impl ClientConfig {
+    pub fn new(url: impl Into<String>) -> Self {
+        Self {
+            url: url.into(),
+            tls: ClientTlsConfig::SystemRoots,
+            client_id: 0,
+            description: None,
+            policy: Policy::default(),
+            ping_interval: Duration::ZERO,
+            ping_jitter: None,
+            max_missed_pings: 3,
+            ping_timestamp: true,
+            request_timeout: Duration::from_secs(30),
+            #[cfg(feature = "crypto")]
+            auth_timeout: Duration::from_secs(30),
+            #[cfg(feature = "crypto")]
+            require_pq: true,
+        }
+    }
+
+    pub fn with_tls(mut self, tls: ClientTlsConfig) -> Self {
+        self.tls = tls;
+        self
+    }
+
+    pub fn with_pinned_pem(self, cert_pem: Vec<u8>) -> Self {
+        self.with_tls(ClientTlsConfig::PinnedPem(cert_pem))
+    }
+
+    pub fn with_client_id(mut self, client_id: u64) -> Self {
+        self.client_id = client_id;
+        self
+    }
+
+    pub fn with_description(mut self, description: impl Into<String>) -> Self {
+        self.description = Some(description.into());
+        self
+    }
+
+    pub fn with_policy(mut self, policy: Policy) -> Self {
+        self.policy = policy;
+        self
+    }
+
+    pub fn with_ping_interval(mut self, interval: Duration) -> Self {
+        self.ping_interval = interval;
+        self
+    }
+
+    pub fn with_ping_jitter(mut self, jitter: Option<Duration>) -> Self {
+        self.ping_jitter = jitter;
+        self
+    }
+
+    pub fn with_max_missed_pings(mut self, max_missed_pings: usize) -> Self {
+        self.max_missed_pings = max_missed_pings;
+        self
+    }
+
+    pub fn with_ping_timestamp(mut self, ping_timestamp: bool) -> Self {
+        self.ping_timestamp = ping_timestamp;
+        self
+    }
+
+    pub fn with_request_timeout(mut self, timeout: Duration) -> Self {
+        self.request_timeout = timeout;
+        self
+    }
+
+    #[cfg(feature = "crypto")]
+    pub fn with_auth_timeout(mut self, timeout: Duration) -> Self {
+        self.auth_timeout = timeout;
+        self
+    }
+
+    #[cfg(feature = "crypto")]
+    pub fn with_require_pq(mut self, require_pq: bool) -> Self {
+        self.require_pq = require_pq;
+        self
+    }
+
+    pub(crate) fn server_cert(&self) -> Option<Vec<u8>> {
+        match &self.tls {
+            ClientTlsConfig::SystemRoots => None,
+            ClientTlsConfig::PinnedPem(cert) => Some(cert.clone()),
+        }
+    }
+}

client/src/connection.rs

@@ -0,0 +1,238 @@
+use mtp_codec::{CommunicationValue, Version};
+#[cfg(feature = "pipes")]
+use mtp_codec::{DataType, DataValue};
+use mtp_common::CommunicationError;
+use std::sync::Arc;
+use tokio::sync::{Mutex, mpsc};
+use tokio::time::Duration;
+
+use crate::config::ClientConfig;
+#[cfg(feature = "crypto")]
+use crate::error::AuthState;
+use crate::ping::{PingSession, start_ping_session};
+#[cfg(feature = "pipes")]
+use crate::pipe::PipeRequest;
+use crate::pipe::{PendingRequest, PipeDispatcher, run_dispatcher};
+
+pub struct MTPConnection {
+    pub version: Version,
+    pub sender: mtp_transport::Sender,
+    pub receiver: mtp_transport::Receiver,
+    pub description: Option<String>,
+    pub(crate) ping: Option<PingSession>,
+    pub(crate) app_rx: Mutex<mpsc::Receiver<Result<CommunicationValue, CommunicationError>>>,
+    #[cfg(feature = "pipes")]
+    pub(crate) pipe_req_rx: Mutex<mpsc::Receiver<PipeRequest>>,
+    pub(crate) pipe_dispatcher: Arc<PipeDispatcher>,
+    pub(crate) request_timeout: Duration,
+    pub(crate) _dispatcher_task: tokio::task::JoinHandle<()>,
+    #[cfg(feature = "crypto")]
+    pub auth_state: AuthState,
+    #[cfg(feature = "crypto")]
+    pub client_id: u64,
+}
+
+impl MTPConnection {
+    pub fn get_ping(&self) -> Option<Duration> {
+        self.ping.as_ref().and_then(PingSession::get_ping)
+    }
+
+    pub async fn request(
+        &self,
+        request: &CommunicationValue,
+        expected_response: Option<mtp_codec::CommunicationType>,
+    ) -> Result<CommunicationValue, CommunicationError> {
+        let request_id = request.get_id();
+        if request_id == 0 {
+            return Err(CommunicationError::Other(
+                "request frame must have a non-zero id".into(),
+            ));
+        }
+
+        let (sender, receiver) = tokio::sync::oneshot::channel();
+        let token = Arc::new(());
+        {
+            let mut pending = self.pipe_dispatcher.pending_requests.lock().await;
+            if pending.contains_key(&request_id) {
+                return Err(CommunicationError::Other(format!(
+                    "request id {request_id} is already pending"
+                )));
+            }
+            pending.insert(
+                request_id,
+                PendingRequest {
+                    token: token.clone(),
+                    sender,
+                },
+            );
+        }
+
+        let response = match tokio::time::timeout(self.request_timeout, async {
+            self.sender.send(request).await?;
+            receiver
+                .await
+                .map_err(|_| CommunicationError::StreamClosed)?
+        })
+        .await
+        {
+            Ok(result) => {
+                if result.is_err() {
+                    crate::pipe::remove_pending_request(&self.pipe_dispatcher, request_id, &token)
+                        .await;
+                }
+                result?
+            }
+            Err(_) => {
+                crate::pipe::remove_pending_request(&self.pipe_dispatcher, request_id, &token)
+                    .await;
+                return Err(CommunicationError::Other(format!(
+                    "request {request_id} timed out after {:?}",
+                    self.request_timeout
+                )));
+            }
+        };
+
+        if let Some(expected) = expected_response {
+            let expected_type = expected.try_to_id(&mtp_codec::TypeMap::latest());
+            if Some(response.get_type()) != expected_type {
+                return Err(CommunicationError::Other(format!(
+                    "unexpected response type: expected {:?}, got {:?}; parsed {}",
+                    expected_type,
+                    response.get_type(),
+                    response
+                )));
+            }
+        }
+
+        Ok(response)
+    }
+
+    pub async fn receive(&self) -> Result<CommunicationValue, CommunicationError> {
+        let mut rx = self.app_rx.lock().await;
+        match rx.recv().await {
+            Some(result) => result,
+            None => Err(CommunicationError::StreamClosed),
+        }
+    }
+}
+
+#[cfg(feature = "pipes")]
+impl MTPConnection {
+    pub async fn create_pipe(
+        &self,
+        description: &str,
+    ) -> Result<crate::pipe::PipeHandle, mtp_common::PipeError> {
+        let pipe_id = rand::random::<u32>();
+        let (tx, rx) = tokio::sync::oneshot::channel();
+
+        {
+            let mut pending = self.pipe_dispatcher.pending_creations.lock().await;
+            pending.insert(pipe_id, tx);
+        }
+
+        let request = CommunicationValue::new(mtp_codec::CommunicationType::PipeRequest)
+            .with_id(pipe_id)
+            .add_typed_default(DataType::Description, DataValue::Str(description.into()));
+
+        self.sender
+            .send(&request)
+            .await
+            .map_err(mtp_common::PipeError::from)?;
+
+        Ok(crate::pipe::PipeHandle {
+            pipe_id,
+            description: description.to_string(),
+            sender: self.sender.clone(),
+            response_rx: rx,
+        })
+    }
+
+    pub async fn receive_pipe(&self) -> Result<PipeRequest, CommunicationError> {
+        let mut rx = self.pipe_req_rx.lock().await;
+        match rx.recv().await {
+            Some(req) => Ok(req),
+            None => Err(CommunicationError::StreamClosed),
+        }
+    }
+}
+
+pub(crate) fn connection_from_parts(
+    config: ClientConfig,
+    sender: mtp_transport::Sender,
+    receiver: mtp_transport::Receiver,
+    version: Version,
+    #[cfg(feature = "crypto")] auth_state: AuthState,
+    #[cfg(feature = "crypto")] client_id: u64,
+) -> MTPConnection {
+    let ping = start_ping_session(&config, sender.clone(), &receiver);
+
+    #[cfg(feature = "pipes")]
+    {
+        let (app_tx, app_rx) = mpsc::channel::<Result<CommunicationValue, CommunicationError>>(
+            config.policy.receiver_queue_capacity,
+        );
+        let (pipe_req_tx, pipe_req_rx) =
+            mpsc::channel::<PipeRequest>(config.policy.receiver_queue_capacity);
+
+        let dispatcher = Arc::new(PipeDispatcher {
+            pending_requests: Mutex::new(std::collections::HashMap::new()),
+            pending_creations: Mutex::new(std::collections::HashMap::new()),
+            pending_pipes: Mutex::new(std::collections::HashMap::new()),
+            policy: Arc::new(config.policy),
+        });
+
+        let dispatcher_clone = dispatcher.clone();
+        let sender_clone = sender.clone();
+        let dispatcher_task = tokio::spawn(run_dispatcher(
+            receiver.clone(),
+            sender_clone,
+            app_tx,
+            pipe_req_tx,
+            dispatcher_clone,
+        ));
+
+        MTPConnection {
+            version,
+            sender,
+            receiver,
+            app_rx: Mutex::new(app_rx),
+            pipe_req_rx: Mutex::new(pipe_req_rx),
+            pipe_dispatcher: dispatcher,
+            request_timeout: config.request_timeout,
+            description: config.description,
+            ping,
+            _dispatcher_task: dispatcher_task,
+            #[cfg(feature = "crypto")]
+            auth_state,
+            #[cfg(feature = "crypto")]
+            client_id,
+        }
+    }
+
+    #[cfg(not(feature = "pipes"))]
+    {
+        let (app_tx, app_rx) = mpsc::channel::<Result<CommunicationValue, CommunicationError>>(
+            config.policy.receiver_queue_capacity,
+        );
+        let dispatcher = Arc::new(PipeDispatcher {
+            pending_requests: Mutex::new(std::collections::HashMap::new()),
+        });
+        let task = tokio::spawn(run_dispatcher(receiver.clone(), app_tx, dispatcher.clone()));
+
+        MTPConnection {
+            version,
+            sender,
+            receiver,
+            app_rx: Mutex::new(app_rx),
+            pipe_dispatcher: dispatcher,
+            request_timeout: config.request_timeout,
+            description: config.description,
+            ping,
+            _dispatcher_task: task,
+            #[cfg(feature = "crypto")]
+            auth_state,
+            #[cfg(feature = "crypto")]
+            client_id,
+        }
+    }
+}

client/src/crypto.rs

@@ -0,0 +1,218 @@
+use mtp_codec::{CommunicationValue, DataType, DataValue, CommunicationType, Version};
+use mtp_common::CommunicationError;
+
+pub(crate) fn unexpected_response_type_error(
+    context: &str,
+    expected_type: mtp_codec::CommunicationTypeId,
+    response: &CommunicationValue,
+) -> CommunicationError {
+    CommunicationError::AuthenticationFailed(format!(
+        "unexpected response type during {context}: expected {:?}, got {:?}; parsed {}",
+        expected_type,
+        response.get_type(),
+        response
+    ))
+}
+
+pub(crate) fn verify_host_challenge(
+    challenge: &CommunicationValue,
+    host_pk: &mtp_crypto::PublicKeyBundle,
+    id: u64,
+    server_challenge: u128,
+    require_pq: bool,
+) -> Result<(), CommunicationError> {
+    use mtp_crypto::{auth, verify_ed25519, verify_ml_dsa};
+
+    let sig = match challenge.get_data(DataType::Signature) {
+        DataValue::Bytes(b) => b.clone(),
+        _ => {
+            return Err(CommunicationError::AuthenticationFailed(
+                "Missing host challenge signature".into(),
+            ));
+        }
+    };
+    let pq_sig = match challenge.get_data(DataType::PqSignature) {
+        DataValue::Bytes(b) => b.clone(),
+        _ => vec![],
+    };
+
+    let host_requires_pq = challenge.get_data(DataType::RequirePq) == &DataValue::BoolTrue;
+    if host_requires_pq && host_pk.sig_pq_public_key.as_bytes().is_empty() {
+        return Err(CommunicationError::AuthenticationFailed(
+            "Host requires post-quantum authentication but its PQ public key is absent".into(),
+        ));
+    }
+    if require_pq && pq_sig.is_empty() {
+        return Err(CommunicationError::AuthenticationFailed(
+            "Host challenge is missing the required PQ signature".into(),
+        ));
+    }
+
+    let payload = auth::challenge_payload(id, server_challenge);
+    verify_ed25519(&host_pk.sig_cl_public_key, &payload, &sig).map_err(|_| {
+        CommunicationError::AuthenticationFailed("Host challenge signature invalid".into())
+    })?;
+    if !pq_sig.is_empty() && verify_ml_dsa(&host_pk.sig_pq_public_key, &payload, &pq_sig).is_err() {
+        return Err(CommunicationError::AuthenticationFailed(
+            "Host challenge PQ signature invalid".into(),
+        ));
+    }
+    Ok(())
+}
+
+pub(crate) fn verify_host_final(
+    response: &CommunicationValue,
+    host_pk: &mtp_crypto::PublicKeyBundle,
+    id: u64,
+    client_nonce: u128,
+    server_challenge: u128,
+    require_pq: bool,
+) -> Result<(), CommunicationError> {
+    use mtp_crypto::{auth, verify_ed25519, verify_ml_dsa};
+
+    match response.get_data(DataType::ClientNonce) {
+        DataValue::UnsignedNumber(n) if *n == client_nonce => {}
+        _ => {
+            return Err(CommunicationError::AuthenticationFailed(
+                "Nonce mismatch".into(),
+            ));
+        }
+    }
+
+    let sig = match response.get_data(DataType::Signature) {
+        DataValue::Bytes(b) => b.clone(),
+        _ => {
+            return Err(CommunicationError::AuthenticationFailed(
+                "Missing signature".into(),
+            ));
+        }
+    };
+    let pq_sig = match response.get_data(DataType::PqSignature) {
+        DataValue::Bytes(b) => b.clone(),
+        _ => vec![],
+    };
+    if require_pq && pq_sig.is_empty() {
+        return Err(CommunicationError::AuthenticationFailed(
+            "Host confirmation is missing the required PQ signature".into(),
+        ));
+    }
+
+    let payload = auth::host_final_payload(id, client_nonce, server_challenge);
+    verify_ed25519(&host_pk.sig_cl_public_key, &payload, &sig)
+        .map_err(|_| CommunicationError::AuthenticationFailed("Host signature invalid".into()))?;
+    if !pq_sig.is_empty() && verify_ml_dsa(&host_pk.sig_pq_public_key, &payload, &pq_sig).is_err() {
+        return Err(CommunicationError::AuthenticationFailed(
+            "Host PQ signature invalid".into(),
+        ));
+    }
+    Ok(())
+}
+
+pub(crate) fn check_connected(
+    response: &CommunicationValue,
+    reject_msg: &str,
+) -> Result<(), CommunicationError> {
+    match response.get_data(DataType::Connected) {
+        DataValue::BoolTrue => Ok(()),
+        DataValue::BoolFalse => Err(CommunicationError::AuthenticationFailed(
+            response
+                .get_str(DataType::ErrorMessage)
+                .unwrap_or(reject_msg)
+                .into(),
+        )),
+        _ => Err(CommunicationError::AuthenticationFailed(
+            "Invalid response".into(),
+        )),
+    }
+}
+
+pub(crate) fn negotiated_version(
+    response: &CommunicationValue,
+) -> Result<Version, CommunicationError> {
+    match response.get_data(DataType::Version) {
+        DataValue::Str(version) => Version::parse(version).ok_or_else(|| {
+            CommunicationError::AuthenticationFailed(
+                "Host returned an invalid negotiated protocol version".into(),
+            )
+        }),
+        _ => Err(CommunicationError::AuthenticationFailed(
+            "Host omitted the negotiated protocol version".into(),
+        )),
+    }
+}
+
+pub(crate) fn signed_challenge_response(
+    keys: &mtp_crypto::Keyring,
+    proof_payload: &[u8],
+    client_nonce: u128,
+) -> Result<CommunicationValue, CommunicationError> {
+    use mtp_crypto::{Ed25519Signer, MlDsaSigner, SignatureScheme};
+
+    let signer = Ed25519Signer::new(&keys.sig_cl_secret_key)
+        .map_err(|e| CommunicationError::Other(e.to_string()))?;
+    let signature = signer
+        .sign(proof_payload)
+        .map_err(|e| CommunicationError::Other(e.to_string()))?;
+
+    let mut proof = CommunicationValue::new(CommunicationType::ChallengeResponse)
+        .add_typed_default(
+            DataType::ClientNonce,
+            DataValue::UnsignedNumber(client_nonce),
+        )
+        .add_typed_default(DataType::Signature, DataValue::Bytes(signature));
+
+    if !keys.sig_pq_secret_key.as_bytes().is_empty() {
+        let pq_signer = MlDsaSigner::new(&keys.sig_pq_secret_key, &keys.sig_pq_public_key)
+            .map_err(|e| CommunicationError::Other(e.to_string()))?;
+        let pq_signature = pq_signer
+            .sign(proof_payload)
+            .map_err(|e| CommunicationError::Other(e.to_string()))?;
+        proof = proof.add_typed_default(DataType::PqSignature, DataValue::Bytes(pq_signature));
+    }
+
+    Ok(proof)
+}
+
+pub(crate) async fn receive_verified_challenge(
+    receiver: &mtp_transport::Receiver,
+    tm: &mtp_codec::TypeMap,
+    host_public_key_bundle: &mtp_crypto::PublicKeyBundle,
+    bound_id: u64,
+    context: &str,
+    require_pq: bool,
+    client_has_pq_key: bool,
+) -> Result<u128, CommunicationError> {
+    let challenge = receiver.receive().await?;
+    let expected = CommunicationType::Challenge
+        .try_to_id(tm)
+        .ok_or_else(|| CommunicationError::Other("Challenge is absent from the type map".into()))?;
+    if challenge.get_type() != expected {
+        return Err(unexpected_response_type_error(
+            context, expected, &challenge,
+        ));
+    }
+
+    let server_challenge = match challenge.get_data(DataType::ServerNonce) {
+        DataValue::UnsignedNumber(n) => *n,
+        _ => {
+            return Err(CommunicationError::AuthenticationFailed(
+                "Missing server challenge".into(),
+            ));
+        }
+    };
+    if challenge.get_data(DataType::RequirePq) == &DataValue::BoolTrue && !client_has_pq_key {
+        return Err(CommunicationError::AuthenticationFailed(
+            "Host requires post-quantum authentication but the client PQ key is absent".into(),
+        ));
+    }
+
+    verify_host_challenge(
+        &challenge,
+        host_public_key_bundle,
+        bound_id,
+        server_challenge,
+        require_pq,
+    )?;
+
+    Ok(server_challenge)
+}

client/src/ping.rs

@@ -0,0 +1,102 @@
+use rand::Rng;
+use std::collections::HashMap;
+use std::sync::Arc;
+use tokio::sync::{Mutex, mpsc};
+use tokio::time::{Duration, Instant};
+
+use mtp_codec::{CommunicationValue, DataType, DataValue, CommunicationType};
+use mtp_transport::{Sender, Receiver};
+
+pub(crate) struct PingSession {
+    pub(crate) last_ping: Arc<Mutex<Option<Duration>>>,
+    pub(crate) task: tokio::task::JoinHandle<()>,
+}
+
+impl PingSession {
+    pub(crate) fn get_ping(&self) -> Option<Duration> {
+        self.last_ping.try_lock().ok().and_then(|ping| *ping)
+    }
+}
+
+impl Drop for PingSession {
+    fn drop(&mut self) {
+        self.task.abort();
+    }
+}
+
+pub(crate) fn start_ping_session(
+    config: &crate::config::ClientConfig,
+    sender: Sender,
+    receiver: &Receiver,
+) -> Option<PingSession> {
+    if config.ping_interval.is_zero() {
+        return None;
+    }
+
+    let (pong_tx, mut pong_rx) = mpsc::unbounded_channel();
+    receiver.observe_pongs(pong_tx);
+    let last_ping = Arc::new(Mutex::new(None));
+    let ping_state = last_ping.clone();
+    let interval = config.ping_interval;
+    let ping_jitter = config.ping_jitter;
+    let max_missed_pings = config.max_missed_pings;
+    let ping_timestamp = config.ping_timestamp;
+    let mut close_rx = receiver.handle().subscribe_close();
+
+    let task = tokio::spawn(async move {
+        let mut ticker = tokio::time::interval(interval);
+        ticker.tick().await;
+        let mut pending = HashMap::new();
+
+        loop {
+            tokio::select! {
+                _ = close_rx.changed() => {
+                    if close_rx.borrow().is_some() {
+                        break;
+                    }
+                }
+                _ = ticker.tick() => {
+                    if max_missed_pings > 0 && !pending.is_empty() && pending.len() >= max_missed_pings {
+                        sender.close();
+                        break;
+                    }
+
+                    if let Some(jitter) = ping_jitter && !jitter.is_zero() {
+                        let max_ms = jitter.as_millis() as u64;
+                        let extra = rand::thread_rng().gen_range(0..=max_ms);
+                        tokio::time::sleep(Duration::from_millis(extra)).await;
+                    }
+
+                    let mut ping = CommunicationValue::new(CommunicationType::Ping);
+                    if ping_timestamp {
+                        let sent_at = std::time::SystemTime::now()
+                            .duration_since(std::time::UNIX_EPOCH)
+                            .unwrap_or_default()
+                            .as_millis();
+                        ping = ping.add_typed_default(
+                            DataType::Timestamp,
+                            DataValue::UnsignedNumber(sent_at),
+                        );
+                    }
+                    let id = ping.get_id();
+                    if sender.send(&ping).await.is_err() {
+                        sender.close();
+                        break;
+                    }
+                    pending.insert(id, Instant::now());
+                }
+                pong = pong_rx.recv() => match pong {
+                    Some(pong) => {
+                        if let Some(sent_at) = pending.remove(&pong.get_id()) {
+                            let mut last_ping = ping_state.lock().await;
+                            *last_ping = Some(sent_at.elapsed());
+                        }
+                    }
+                    None => break,
+                },
+            }
+        }
+    });
+
+    Some(PingSession { last_ping, task })
+}

client/src/pipe.rs

@@ -0,0 +1,232 @@
+use mtp_codec::CommunicationValue;
+use mtp_common::CommunicationError;
+use mtp_transport::Receiver;
+use std::collections::HashMap;
+use std::sync::Arc;
+use tokio::sync::{Mutex, mpsc};
+
+#[cfg(feature = "pipes")]
+use mtp_common::PipeError;
+#[cfg(feature = "pipes")]
+use mtp_codec::{DataType, DataValue, CommunicationType};
+#[cfg(feature = "pipes")]
+use mtp_transport::{Policy, Sender};
+
+#[cfg(feature = "pipes")]
+pub struct PipeHandle {
+    pub(crate) pipe_id: u32,
+    pub(crate) description: String,
+    pub(crate) sender: Sender,
+    pub(crate) response_rx: tokio::sync::oneshot::Receiver<Result<bool, PipeError>>,
+}
+
+#[cfg(feature = "pipes")]
+impl PipeHandle {
+    pub fn pipe_id(&self) -> u32 {
+        self.pipe_id
+    }
+
+    pub fn description(&self) -> &str {
+        &self.description
+    }
+
+    pub async fn wait(self) -> Result<Option<mtp_transport::PipeWriter>, PipeError> {
+        match self.response_rx.await {
+            Ok(Ok(true)) => {
+                let writer = self
+                    .sender
+                    .open_pipe(self.pipe_id, &self.description)
+                    .await
+                    .map_err(PipeError::from)?;
+                Ok(Some(writer))
+            }
+            Ok(Ok(false)) => Ok(None),
+            Ok(Err(e)) => Err(e),
+            Err(_) => Err(PipeError::StreamClosed),
+        }
+    }
+}
+
+#[cfg(feature = "pipes")]
+pub struct PipeRequest {
+    pub(crate) pipe_id: u32,
+    pub(crate) description: String,
+    pub(crate) sender: Sender,
+    pub(crate) dispatcher: Arc<PipeDispatcher>,
+}
+
+#[cfg(feature = "pipes")]
+impl PipeRequest {
+    pub fn id(&self) -> u32 {
+        self.pipe_id
+    }
+
+    pub fn description(&self) -> &str {
+        &self.description
+    }
+
+    pub async fn accept(self) -> Result<mtp_transport::PipeReader, PipeError> {
+        let (pipe_tx, pipe_rx) = tokio::sync::oneshot::channel();
+        {
+            let mut pending = self.dispatcher.pending_pipes.lock().await;
+            pending.insert(self.pipe_id, pipe_tx);
+        }
+
+        let resp = CommunicationValue::new(CommunicationType::PipeResponse)
+            .with_id(self.pipe_id)
+            .add_typed_default(DataType::Accepted, DataValue::BoolTrue);
+        self.sender.send(&resp).await.map_err(PipeError::from)?;
+
+        let timeout = self.dispatcher.policy.read_timeout;
+        tokio::time::timeout(timeout, pipe_rx)
+            .await
+            .map_err(|_| PipeError::HandshakeTimeout)?
+            .map_err(|_| PipeError::StreamClosed)
+    }
+
+    pub async fn deny(self) -> Result<(), PipeError> {
+        let resp = CommunicationValue::new(CommunicationType::PipeResponse)
+            .with_id(self.pipe_id)
+            .add_typed_default(DataType::Accepted, DataValue::BoolFalse);
+        self.sender.send(&resp).await.map_err(PipeError::from)?;
+        Ok(())
+    }
+}
+
+pub(crate) struct PendingRequest {
+    pub(crate) token: Arc<()>,
+    pub(crate) sender: tokio::sync::oneshot::Sender<Result<CommunicationValue, CommunicationError>>,
+}
+
+pub(crate) struct PipeDispatcher {
+    pub(crate) pending_requests: Mutex<HashMap<u32, PendingRequest>>,
+    #[cfg(feature = "pipes")]
+    pub(crate) pending_creations: Mutex<HashMap<u32, tokio::sync::oneshot::Sender<Result<bool, PipeError>>>>,
+    #[cfg(feature = "pipes")]
+    pub(crate) pending_pipes: Mutex<HashMap<u32, tokio::sync::oneshot::Sender<mtp_transport::PipeReader>>>,
+    #[cfg(feature = "pipes")]
+    pub(crate) policy: Arc<Policy>,
+}
+
+pub(crate) async fn route_message(
+    msg: CommunicationValue,
+    app_tx: &mpsc::Sender<Result<CommunicationValue, CommunicationError>>,
+    dispatcher: &PipeDispatcher,
+) -> bool {
+    let pending = dispatcher
+        .pending_requests
+        .lock()
+        .await
+        .remove(&msg.get_id());
+    if let Some(tx) = pending {
+        let _ = tx.sender.send(Ok(msg));
+        return true;
+    }
+
+    app_tx.send(Ok(msg)).await.is_ok()
+}
+
+pub(crate) async fn fail_pending_requests(
+    dispatcher: &PipeDispatcher,
+    error: CommunicationError,
+) {
+    let pending = std::mem::take(&mut *dispatcher.pending_requests.lock().await);
+    for (_, pending) in pending {
+        let _ = pending.sender.send(Err(error.clone()));
+    }
+}
+
+pub(crate) async fn remove_pending_request(
+    dispatcher: &PipeDispatcher,
+    request_id: u32,
+    token: &Arc<()>,
+) {
+    let mut pending = dispatcher.pending_requests.lock().await;
+    if pending
+        .get(&request_id)
+        .is_some_and(|entry| Arc::ptr_eq(&entry.token, token))
+    {
+        pending.remove(&request_id);
+    }
+}
+
+#[cfg(feature = "pipes")]
+pub(crate) async fn run_dispatcher(
+    receiver: Receiver,
+    sender: Sender,
+    app_tx: mpsc::Sender<Result<CommunicationValue, CommunicationError>>,
+    pipe_req_tx: mpsc::Sender<PipeRequest>,
+    dispatcher: Arc<PipeDispatcher>,
+) {
+    let pipe_req_type =
+        CommunicationType::PipeRequest.try_to_id(&mtp_codec::TypeMap::latest());
+    let pipe_resp_type =
+        CommunicationType::PipeResponse.try_to_id(&mtp_codec::TypeMap::latest());
+
+    loop {
+        match receiver.receive_event().await {
+            Ok(mtp_transport::TransportEvent::Message(msg)) => {
+                if Some(msg.get_type()) == pipe_req_type {
+                    let pipe_id = msg.get_id();
+                    let description = msg.get_str(DataType::Description).unwrap_or("").to_string();
+                    let req = PipeRequest {
+                        pipe_id,
+                        description,
+                        sender: sender.clone(),
+                        dispatcher: dispatcher.clone(),
+                    };
+                    let _ = pipe_req_tx.send(req).await;
+                    continue;
+                }
+
+                if Some(msg.get_type()) == pipe_resp_type {
+                    let pipe_id = msg.get_id();
+                    let accepted = msg.get_bool(DataType::Accepted).unwrap_or(false);
+                    let mut pending = dispatcher.pending_creations.lock().await;
+                    if let Some(tx) = pending.remove(&pipe_id) {
+                        let _ = tx.send(Ok(accepted));
+                    }
+                    continue;
+                }
+
+                if !route_message(msg, &app_tx, &dispatcher).await {
+                    break;
+                }
+            }
+            Ok(mtp_transport::TransportEvent::Pipe(reader)) => {
+                let pipe_id = reader.pipe_id();
+                let mut pending = dispatcher.pending_pipes.lock().await;
+                if let Some(tx) = pending.remove(&pipe_id) {
+                    let _ = tx.send(reader);
+                }
+            }
+            Err(e) => {
+                fail_pending_requests(&dispatcher, e.clone()).await;
+                let _ = app_tx.send(Err(e)).await;
+                break;
+            }
+        }
+    }
+}
+
+#[cfg(not(feature = "pipes"))]
+pub(crate) async fn run_dispatcher(
+    receiver: Receiver,
+    app_tx: mpsc::Sender<Result<CommunicationValue, CommunicationError>>,
+    dispatcher: Arc<PipeDispatcher>,
+) {
+    loop {
+        match receiver.receive().await {
+            Ok(msg) => {
+                if !route_message(msg, &app_tx, &dispatcher).await {
+                    break;
+                }
+            }
+            Err(e) => {
+                fail_pending_requests(&dispatcher, e.clone()).await;
+                let _ = app_tx.send(Err(e)).await;
+                break;
+            }
+        }
+    }
+}

client/tests/ping.rs

@@ -32,15 +32,15 @@ async fn test_ping_rtt_and_missed_ping_teardown() -> Result<(), Box<dyn std::err
     let (mut host, cert_pem) = start_host(true).await?;
     let url = format!("https://127.0.0.1:{}", host.local_addr().port());
 
-    let client = MTPClient::connect(
+    let client_connect = MTPClient::connect(
         ClientConfig::new(url)
             .with_pinned_pem(cert_pem)
             .with_ping_interval(std::time::Duration::from_millis(25))
             .with_max_missed_pings(3),
-    )
-    .await?;
-
-    let _accepted = host.accept().await?;
+    );
+    let (client, accepted) = tokio::join!(client_connect, host.accept());
+    let client = client?;
+    let _accepted = accepted?;
 
     let ping = tokio::time::timeout(std::time::Duration::from_secs(5), async {
         loop {
@@ -56,15 +56,15 @@ async fn test_ping_rtt_and_missed_ping_teardown() -> Result<(), Box<dyn std::err
 
     let (mut silent_host, silent_cert_pem) = start_host(false).await?;
     let silent_url = format!("https://127.0.0.1:{}", silent_host.local_addr().port());
-    let silent_client = MTPClient::connect(
+    let silent_connect = MTPClient::connect(
         ClientConfig::new(silent_url)
             .with_pinned_pem(silent_cert_pem)
             .with_ping_interval(std::time::Duration::from_millis(25))
             .with_max_missed_pings(2),
-    )
-    .await?;
-
-    let _accepted = silent_host.accept().await?;
+    );
+    let (silent_client, accepted) = tokio::join!(silent_connect, silent_host.accept());
+    let silent_client = silent_client?;
+    let _accepted = accepted?;
 
     let closed = tokio::time::timeout(std::time::Duration::from_secs(5), async {
         loop {