methanium/mtp
Watch
3
0
Fork
You've already forked mtp
 0
Code Issues Pull requests 3 Projects Releases 22 Packages Wiki Activity Actions

Brought Example up to spec
Some checks failed
CI / checks (push) Failing after 2m25s
Details

Browse source
This commit is contained in:
Alex Emmet 2026-07-19 00:29:24 +02:00
commit eaae9b0d13
41 changed files with 1695 additions and 655 deletions
Download patch file Download diff file Expand all files Collapse all files

135
host/src/connection.rs
View file

@ -1,3 +1,5 @@
#[cfg(feature = "pipes")]
use mtp_codec::{CommunicationType, DataType, DataValue};
use mtp_codec::{CommunicationValue, Version, registry::VersionedCodec};
use mtp_common::CommunicationError;
#[cfg(feature = "pipes")]
@ -8,7 +10,9 @@ use tokio::sync::{Mutex, mpsc};
#[cfg(feature = "crypto")]
use crate::error::random_client_id;
#[cfg(feature = "pipes")]
use crate::pipe::{PipeDispatcher, PipeRequest};
use crate::pipe::{PipeDispatcher, PipeReceiver, PipeRequest, PipeSender, run_dispatcher};
#[cfg(feature = "pipes")]
use mtp_transport::Policy;
mod connection_capability {
pub trait Sealed {}
@ -44,7 +48,11 @@ impl<C: mtp_transport::TransportConnection> MtpReceiverLike for mtp_transport::G
}
}
pub struct MTPConnection<S = mtp_transport::Sender, R = mtp_transport::Receiver> {
pub struct MTPConnection<
S = mtp_transport::Sender,
R = mtp_transport::Receiver,
P = wtransport::RecvStream,
> {
pub version: Version,
pub codec: VersionedCodec,
pub sender: S,
@ -58,9 +66,11 @@ pub struct MTPConnection<S = mtp_transport::Sender, R = mtp_transport::Receiver>
#[cfg(feature = "pipes")]
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_req_rx: Mutex<mpsc::Receiver<PipeRequest<S, P>>>,
#[cfg(feature = "pipes")]
pub(crate) pipe_dispatcher: Arc<PipeDispatcher>,
pub(crate) pipe_dispatcher: Arc<PipeDispatcher<P>>,
#[cfg(not(feature = "pipes"))]
pub(crate) _pipe_stream: std::marker::PhantomData<P>,
pub description: Option<String>,
pub(crate) _dispatcher_task: tokio::task::JoinHandle<()>,
#[cfg(feature = "crypto")]
@ -71,7 +81,13 @@ pub struct MTPConnection<S = mtp_transport::Sender, R = mtp_transport::Receiver>
pub client_public_key: Option<mtp_crypto::PublicKeyBundle>,
}
impl<S, R> MTPConnection<S, R> {
#[cfg(feature = "pipes")]
impl<S, R, P> MTPConnection<S, R, P>
where
S: PipeSender,
R: PipeReceiver<P>,
P: tokio::io::AsyncRead + Send + Unpin + 'static,
{
/// Construct an MTP connection from an alternative transport backend.
///
/// Native `MTPHost` users continue to receive the default
@ -86,25 +102,60 @@ impl<S, R> MTPConnection<S, R> {
path: String,
description: Option<String>,
) -> Self {
#[cfg(feature = "pipes")]
let (_, app_rx) = mpsc::channel::<Result<CommunicationValue, CommunicationError>>(1);
#[cfg(feature = "pipes")]
let (_, pipe_req_rx) = mpsc::channel::<PipeRequest>(1);
#[cfg(feature = "pipes")]
let dispatcher = Arc::new(PipeDispatcher::default_for_external());
let policy = Arc::new(Policy::default());
let (app_tx, app_rx) = mpsc::channel(policy.receiver_queue_capacity);
let (pipe_req_tx, pipe_req_rx) = mpsc::channel(policy.receiver_queue_capacity);
let dispatcher = Arc::new(PipeDispatcher {
pending_creations: Mutex::new(std::collections::HashMap::new()),
pending_pipes: Mutex::new(std::collections::HashMap::new()),
policy,
});
let task = tokio::spawn(run_dispatcher(
receiver.clone(),
sender.clone(),
app_tx,
pipe_req_tx,
dispatcher.clone(),
));
Self {
version,
codec,
sender,
receiver,
path,
#[cfg(feature = "pipes")]
app_rx: Mutex::new(app_rx),
#[cfg(feature = "pipes")]
pipe_req_rx: Mutex::new(pipe_req_rx),
#[cfg(feature = "pipes")]
pipe_dispatcher: dispatcher,
description,
_dispatcher_task: task,
#[cfg(feature = "crypto")]
auth_state: crate::error::AuthState::Unauthenticated,
#[cfg(feature = "crypto")]
client_id: random_client_id(),
#[cfg(feature = "crypto")]
client_public_key: None,
}
}
}
#[cfg(not(feature = "pipes"))]
impl<S, R, P> MTPConnection<S, R, P> {
pub fn from_transport_parts(
version: Version,
codec: VersionedCodec,
sender: S,
receiver: R,
path: String,
description: Option<String>,
) -> Self {
Self {
version,
codec,
sender,
receiver,
path,
description,
_pipe_stream: std::marker::PhantomData,
_dispatcher_task: tokio::spawn(async {}),
#[cfg(feature = "crypto")]
auth_state: crate::error::AuthState::Unauthenticated,
@ -124,3 +175,59 @@ impl<S: MtpSenderLike, R: MtpReceiverLike> MTPConnection<S, R> {
Ok(message)
}
}
#[cfg(feature = "pipes")]
impl<S, R, P> MTPConnection<S, R, P>
where
S: PipeSender,
P: tokio::io::AsyncRead + Send + Unpin + 'static,
{
pub async fn receive(&self) -> Result<CommunicationValue, CommunicationError> {
let mut rx = self.app_rx.lock().await;
match rx.recv().await {
Some(Ok(mut message)) => {
message.set_type_map(self.codec.type_map());
Ok(message)
}
Some(Err(error)) => Err(error),
None => Err(CommunicationError::StreamClosed),
}
}
pub async fn create_pipe(
&self,
description: &str,
) -> Result<crate::pipe::PipeHandle<S>, mtp_common::PipeError> {
let pipe_id = rand::random::<u32>();
let (response_tx, response_rx) = tokio::sync::oneshot::channel();
self.pipe_dispatcher
.pending_creations
.lock()
.await
.insert(pipe_id, response_tx);
let request = CommunicationValue::new(CommunicationType::PipeRequest)
.with_id(pipe_id)
.add_typed_default(DataType::Description, DataValue::Str(description.into()));
self.sender
.send_pipe_message(&request)
.await
.map_err(mtp_common::PipeError::from)?;
Ok(crate::pipe::PipeHandle {
pipe_id,
description: description.to_owned(),
sender: self.sender.clone(),
response_rx,
})
}
pub async fn receive_pipe(&self) -> Result<PipeRequest<S, P>, CommunicationError> {
self.pipe_req_rx
.lock()
.await
.recv()
.await
.ok_or(CommunicationError::StreamClosed)
}
}

250
host/src/handshake.rs
View file

@ -1,4 +1,4 @@
#[cfg(any(feature = "crypto", feature = "pipes"))]
#[cfg(feature = "crypto")]
use mtp_codec::{CommunicationType, CommunicationValue};
use mtp_codec::{
DataType, DataValue, Version,
@ -7,6 +7,9 @@ use mtp_codec::{
use mtp_common::RejectionReason;
use mtp_transport::{Receiver, Sender};
use std::sync::Arc;
use std::time::Instant;
#[cfg(feature = "crypto")]
use std::time::Instant;
#[cfg(feature = "pipes")]
use tokio::sync::mpsc;
@ -18,9 +21,9 @@ use crate::connection::MTPConnection;
use crate::error::AuthState;
use crate::error::{AcceptError, extract_version, send_accepted, send_rejection};
#[cfg(feature = "pipes")]
use crate::pipe::run_dispatcher;
use crate::pipe::PipeDispatcher;
#[cfg(feature = "pipes")]
use crate::pipe::{PipeDispatcher, PipeRequest};
use crate::pipe::run_dispatcher;
pub struct MTPHost {
pub(crate) transport: mtp_transport::Host,
@ -71,11 +74,16 @@ impl MTPHost {
}
if self.handshakes.is_empty() {
let incoming_started = Instant::now();
match self.transport.next().await {
Some((sender, receiver)) => {
tracing::debug!(elapsed = ?incoming_started.elapsed(), "host accept loop: dispatch authentication handshake");
let context = self.context.clone();
self.handshakes.spawn(async move {
context.accept_pair_timed(sender, receiver).await
let handshake_started = Instant::now();
let result = context.accept_pair_timed(sender, receiver).await;
tracing::debug!(elapsed = ?handshake_started.elapsed(), success = result.is_ok(), "host accept loop: authentication handshake finished");
result
});
continue;
}
@ -99,9 +107,15 @@ impl MTPHost {
incoming = self.transport.next() => {
match incoming {
Some((sender, receiver)) => {
tracing::debug!("host accept loop: dispatch authentication handshake");
let context = self.context.clone();
self.handshakes
.spawn(async move { context.accept_pair_timed(sender, receiver).await });
.spawn(async move {
let handshake_started = Instant::now();
let result = context.accept_pair_timed(sender, receiver).await;
tracing::debug!(elapsed = ?handshake_started.elapsed(), success = result.is_ok(), "host accept loop: authentication handshake finished");
result
});
}
None => self.transport_closed = true,
}
@ -172,7 +186,7 @@ impl HandshakeContext {
},
)
.await;
sender.close();
sender.close().await;
return Err(AcceptError::AuthenticationFailed(
"authentication not allowed on this host".into(),
));
@ -187,7 +201,7 @@ impl HandshakeContext {
},
)
.await;
sender.close();
sender.close().await;
return Err(AcceptError::MissingVersion);
}
};
@ -208,7 +222,7 @@ impl HandshakeContext {
},
)
.await;
sender.close();
sender.close().await;
return Err(AcceptError::UnsupportedVersion(client_version));
}
};
@ -256,7 +270,7 @@ impl HandshakeContext {
},
)
.await;
sender.close();
sender.close().await;
return Err(AcceptError::MissingVersion);
}
};
@ -273,7 +287,7 @@ impl HandshakeContext {
},
)
.await;
sender.close();
sender.close().await;
return Err(AcceptError::UnsupportedVersion(client_version));
}
};
@ -359,14 +373,6 @@ impl HandshakeContext {
receiver.respond_to_pings(sender.clone());
}
#[cfg(feature = "pipes")]
let (_, app_rx) = mpsc::channel::<
Result<mtp_codec::CommunicationValue, mtp_common::CommunicationError>,
>(1);
#[cfg(feature = "pipes")]
let (_, pipe_req_rx) = mpsc::channel::<PipeRequest>(1);
#[cfg(feature = "pipes")]
let dispatcher = Arc::new(PipeDispatcher);
let task = tokio::spawn(async {});
MTPConnection {
@ -375,12 +381,7 @@ impl HandshakeContext {
sender,
receiver,
path: "/".to_string(),
#[cfg(feature = "pipes")]
app_rx: tokio::sync::Mutex::new(app_rx),
#[cfg(feature = "pipes")]
pipe_req_rx: tokio::sync::Mutex::new(pipe_req_rx),
#[cfg(feature = "pipes")]
pipe_dispatcher: dispatcher,
_pipe_stream: std::marker::PhantomData,
description,
_dispatcher_task: task,
}
@ -451,11 +452,6 @@ impl HandshakeContext {
receiver.respond_to_pings(sender.clone());
}
let (_, app_rx) = mpsc::channel::<
Result<mtp_codec::CommunicationValue, mtp_common::CommunicationError>,
>(1);
let (_, pipe_req_rx) = mpsc::channel::<PipeRequest>(1);
let dispatcher = Arc::new(PipeDispatcher);
let task = tokio::spawn(async {});
MTPConnection {
@ -464,9 +460,7 @@ impl HandshakeContext {
sender,
receiver,
path: "/".to_string(),
app_rx: tokio::sync::Mutex::new(app_rx),
pipe_req_rx: tokio::sync::Mutex::new(pipe_req_rx),
pipe_dispatcher: dispatcher,
_pipe_stream: std::marker::PhantomData,
description,
_dispatcher_task: task,
auth_state,
@ -477,60 +471,6 @@ impl HandshakeContext {
}
}
#[cfg(feature = "pipes")]
impl MTPConnection {
pub async fn receive(
&self,
) -> Result<mtp_codec::CommunicationValue, mtp_common::CommunicationError> {
let mut rx = self.app_rx.lock().await;
match rx.recv().await {
Some(Ok(mut message)) => {
message.set_type_map(self.codec.type_map());
Ok(message)
}
Some(Err(error)) => Err(error),
None => Err(mtp_common::CommunicationError::StreamClosed),
}
}
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(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, mtp_common::CommunicationError> {
let mut rx = self.pipe_req_rx.lock().await;
match rx.recv().await {
Some(req) => Ok(req),
None => Err(mtp_common::CommunicationError::StreamClosed),
}
}
}
#[cfg(feature = "crypto")]
enum Flow {
Login {
@ -581,21 +521,21 @@ impl HandshakeContext {
let hello = match receiver.receive().await {
Ok(m) => m,
Err(e) => {
sender.close();
sender.close().await;
return Err(AcceptError::Receive(e));
}
};
let version_str = match hello.get_data(DataType::Version) {
DataValue::Str(s) => s.clone(),
_ => {
sender.close();
sender.close().await;
return Err(AcceptError::MissingVersion);
}
};
let client_version = match Version::parse(&version_str) {
Some(v) => v,
None => {
sender.close();
sender.close().await;
return Err(AcceptError::MissingVersion);
}
};
@ -611,7 +551,7 @@ impl HandshakeContext {
let cid = match hello.get_data(DataType::Id) {
DataValue::UnsignedNumber(n) => *n as u64,
_ => {
sender.close();
sender.close().await;
return Err(AcceptError::AuthenticationFailed(
"missing client id".into(),
));
@ -628,7 +568,7 @@ impl HandshakeContext {
DataValue::Str("unknown client id".into()),
);
let _ = sender.send(&rejection).await;
sender.close();
sender.close().await;
return Err(AcceptError::AuthenticationFailed(
"unknown client id".into(),
));
@ -644,7 +584,7 @@ impl HandshakeContext {
AcceptError::AuthenticationFailed("invalid public key bundle".into())
})?,
_ => {
sender.close();
sender.close().await;
return Err(AcceptError::AuthenticationFailed(
"missing public keys".into(),
));
@ -656,7 +596,7 @@ impl HandshakeContext {
CommunicationType::RegisterResponse,
)
} else {
sender.close();
sender.close().await;
return Err(AcceptError::AuthenticationFailed(
"unexpected authentication message".into(),
));
@ -685,11 +625,11 @@ impl HandshakeContext {
client_version: Version,
description: Option<String>,
) -> Result<Option<MTPConnection>, AcceptError> {
use mtp_crypto::{
Ed25519Signer, MlDsaSigner, SignatureScheme, auth, verify_ed25519, verify_ml_dsa,
};
use mtp_crypto::{Ed25519Signer, MlDsaSigner, SignatureScheme, auth, verify_ed25519};
let handshake_started = Instant::now();
let tm = mtp_codec::TypeMap::latest();
let negotiate_started = Instant::now();
let negotiated = match self
.registry
.negotiate(std::slice::from_ref(&client_version))
@ -703,10 +643,11 @@ impl HandshakeContext {
},
)
.await;
sender.close();
sender.close().await;
return Err(AcceptError::UnsupportedVersion(client_version));
}
};
tracing::debug!(elapsed = ?negotiate_started.elapsed(), "authentication handshake: version negotiation");
let pq_enabled = !self
.config
.host_keyring
@ -729,30 +670,43 @@ impl HandshakeContext {
},
)
.await;
sender.close();
sender.close().await;
return Err(AcceptError::AuthenticationFailed(
"PQ authentication is required but the host PQ key is absent".into(),
));
}
let host_sign = |payload: &[u8]| -> Result<(Vec<u8>, Vec<u8>), AcceptError> {
let signer = Ed25519Signer::new(&self.config.host_keyring.sig_cl_secret_key)
.map_err(|e| AcceptError::AuthenticationFailed(e.to_string()))?;
let sig = signer
.sign(payload)
.map_err(|e| AcceptError::AuthenticationFailed(e.to_string()))?;
let pq_sig = if pq_enabled {
let pq = MlDsaSigner::new(
let signer_init_started = Instant::now();
let host_pq_signer = if pq_enabled {
Some(Arc::new(
MlDsaSigner::new(
&self.config.host_keyring.sig_pq_secret_key,
&self.config.host_keyring.sig_pq_public_key,
)
.map_err(|e| AcceptError::AuthenticationFailed(e.to_string()))?,
))
} else {
None
};
tracing::debug!(elapsed = ?signer_init_started.elapsed(), "authentication handshake: signer initialization");
let host_sign = |payload: Vec<u8>| async {
let signer = Ed25519Signer::new(&self.config.host_keyring.sig_cl_secret_key)
.map_err(|e| AcceptError::AuthenticationFailed(e.to_string()))?;
pq.sign(payload)
.map_err(|e| AcceptError::AuthenticationFailed(e.to_string()))?
if let Some(pq_signer) = host_pq_signer.as_ref() {
mtp_crypto::sign_parallel::sign_dual_parallel_shared_pq(
signer,
Arc::clone(pq_signer),
payload,
)
.await
.map_err(|e| AcceptError::AuthenticationFailed(e.to_string()))
} else {
Vec::new()
};
Ok((sig, pq_sig))
let sig = signer
.sign(&payload)
.map_err(|e| AcceptError::AuthenticationFailed(e.to_string()))?;
Ok((sig, Vec::new()))
}
};
let challenge_id = match &flow {
@ -761,8 +715,10 @@ impl HandshakeContext {
};
let server_challenge: u128 = rand::random();
let sign_challenge_started = Instant::now();
let (chal_sig, chal_pq_sig) =
host_sign(&auth::challenge_payload(challenge_id, server_challenge))?;
host_sign(auth::challenge_payload(challenge_id, server_challenge)).await?;
tracing::debug!(elapsed = ?sign_challenge_started.elapsed(), "authentication handshake: sign challenge");
let mut challenge_msg = CommunicationValue::new(CommunicationType::Challenge)
.add_typed_default(
@ -782,20 +738,24 @@ impl HandshakeContext {
challenge_msg = challenge_msg
.add_typed_default(DataType::PqSignature, DataValue::Bytes(chal_pq_sig));
}
let send_challenge_started = Instant::now();
if let Err(e) = sender.send(&challenge_msg).await {
sender.close();
sender.close().await;
return Err(AcceptError::Send(e));
}
tracing::debug!(elapsed = ?send_challenge_started.elapsed(), "authentication handshake: send challenge");
let receive_proof_started = Instant::now();
let proof = match receiver.receive().await {
Ok(m) => m,
Err(e) => {
sender.close();
sender.close().await;
return Err(AcceptError::Receive(e));
}
};
tracing::debug!(elapsed = ?receive_proof_started.elapsed(), "authentication handshake: receive client proof");
if Some(proof.get_type()) != CommunicationType::ChallengeResponse.try_to_id(&tm) {
sender.close();
sender.close().await;
return Err(AcceptError::AuthenticationFailed(
"missing challenge response".into(),
));
@ -803,7 +763,7 @@ impl HandshakeContext {
let client_nonce = match proof.get_data(DataType::ClientNonce) {
DataValue::UnsignedNumber(n) => *n,
_ => {
sender.close();
sender.close().await;
return Err(AcceptError::AuthenticationFailed(
"missing client nonce".into(),
));
@ -812,7 +772,7 @@ impl HandshakeContext {
let sig_bytes = match proof.get_data(DataType::Signature) {
DataValue::Bytes(b) => b.clone(),
_ => {
sender.close();
sender.close().await;
return Err(AcceptError::AuthenticationFailed(
"missing challenge signature".into(),
));
@ -837,18 +797,24 @@ impl HandshakeContext {
};
let has_client_pq_key = !bundle.sig_pq_public_key.as_bytes().is_empty();
let pq_ok = if self.config.require_pq {
has_client_pq_key
&& !pq_sig_bytes.is_empty()
&& verify_ml_dsa(&bundle.sig_pq_public_key, &proof_payload, &pq_sig_bytes).is_ok()
let verify_proof_started = Instant::now();
let proof_ok = if pq_sig_bytes.is_empty() {
!self.config.require_pq
&& verify_ed25519(&bundle.sig_cl_public_key, &proof_payload, &sig_bytes).is_ok()
} else if has_client_pq_key {
mtp_crypto::sign_parallel::verify_dual_parallel(
bundle.sig_cl_public_key.clone(),
bundle.sig_pq_public_key.clone(),
proof_payload,
sig_bytes,
pq_sig_bytes,
)
.await
.is_ok()
} else {
pq_sig_bytes.is_empty()
|| (has_client_pq_key
&& verify_ml_dsa(&bundle.sig_pq_public_key, &proof_payload, &pq_sig_bytes)
.is_ok())
false
};
let proof_ok =
verify_ed25519(&bundle.sig_cl_public_key, &proof_payload, &sig_bytes).is_ok() && pq_ok;
tracing::debug!(elapsed = ?verify_proof_started.elapsed(), "authentication handshake: verify client proof");
if !proof_ok {
send_rejection(
@ -858,12 +824,13 @@ impl HandshakeContext {
},
)
.await;
sender.close();
sender.close().await;
return Err(AcceptError::AuthenticationFailed(
"client proof signature invalid".into(),
));
}
let register_started = Instant::now();
let (assigned_id, client_bundle) = match flow {
Flow::Login { id, bundle } => (id, bundle),
Flow::Register { bundle, .. } => {
@ -872,12 +839,16 @@ impl HandshakeContext {
(new_id, bundle)
}
};
tracing::debug!(elapsed = ?register_started.elapsed(), "authentication handshake: registration callback");
let (host_sig, host_pq_sig) = host_sign(&auth::host_final_payload(
let sign_final_started = Instant::now();
let (host_sig, host_pq_sig) = host_sign(auth::host_final_payload(
assigned_id,
client_nonce,
server_challenge,
))?;
))
.await?;
tracing::debug!(elapsed = ?sign_final_started.elapsed(), "authentication handshake: sign final response");
let mut response = CommunicationValue::new(response_type)
.add_typed_default(DataType::Connected, DataValue::BoolTrue)
@ -894,14 +865,17 @@ impl HandshakeContext {
response.add_typed_default(DataType::PqSignature, DataValue::Bytes(host_pq_sig));
}
let send_final_started = Instant::now();
if let Err(e) = sender.send(&response).await {
sender.close();
sender.close().await;
return Err(AcceptError::Send(e));
}
if let Err(e) = sender.finish_stream().await {
sender.close();
sender.close().await;
return Err(AcceptError::Send(e));
}
tracing::debug!(elapsed = ?send_final_started.elapsed(), "authentication handshake: send final response");
tracing::debug!(elapsed = ?handshake_started.elapsed(), "authentication handshake: complete");
let codec = match VersionedCodec::for_version(self.registry.clone(), negotiated.clone()) {
Some(codec) => codec,
@ -939,14 +913,14 @@ impl HandshakeContext {
let version_str = match hello.get_data(DataType::Version) {
DataValue::Str(s) => s.clone(),
_ => {
sender.close();
sender.close().await;
return Err(AcceptError::MissingVersion);
}
};
let client_version = match Version::parse(&version_str) {
Some(v) => v,
None => {
sender.close();
sender.close().await;
return Err(AcceptError::MissingVersion);
}
};
@ -959,16 +933,16 @@ impl HandshakeContext {
if Some(hello.get_type()) == CommunicationType::Register.try_to_id(&tm) {
let bundle = match hello.get_data(DataType::PublicKeys) {
DataValue::Bytes(b) => PublicKeyBundle::from_bytes(b).map_err(|_| {
sender.close();
AcceptError::AuthenticationFailed("invalid public key bundle".into())
})?,
_ => {
sender.close();
sender.close().await;
return Err(AcceptError::AuthenticationFailed(
"missing public keys".into(),
));
}
};
sender.close().await;
let pk_bytes = bundle.as_bytes();
return self
.complete_auth_handshake(
@ -1019,7 +993,7 @@ impl HandshakeContext {
},
)
.await;
sender.close();
sender.close().await;
return Err(AcceptError::UnsupportedVersion(client_version));
}
};
@ -1046,7 +1020,7 @@ impl HandshakeContext {
)));
}
sender.close();
sender.close().await;
Err(AcceptError::AuthenticationFailed(
"unexpected message type".into(),
))

9
host/src/lib.rs
View file

@ -20,6 +20,8 @@ pub use mtp_transport::Sender;
pub use mtp_common::PipeError;
#[cfg(feature = "pipes")]
pub use mtp_transport::PipeWriter;
#[cfg(feature = "pipes")]
pub use pipe::PipeRequest;
pub use mtp_codec::registry::Registry;
@ -31,11 +33,15 @@ pub use error::AuthState;
#[cfg(test)]
mod tests {
use super::*;
use mtp_codec::{CommunicationType, DataType, DataValue, registry::VersionedCodec};
#[cfg(not(feature = "pipes"))]
use mtp_codec::registry::VersionedCodec;
use mtp_codec::{CommunicationType, DataType, DataValue};
#[cfg(not(feature = "pipes"))]
#[derive(Clone, Debug, PartialEq, Eq)]
struct AlternateSender;
#[cfg(not(feature = "pipes"))]
#[derive(Clone, Debug, PartialEq, Eq)]
struct AlternateReceiver;
@ -80,6 +86,7 @@ mod tests {
Ok(())
}
#[cfg(not(feature = "pipes"))]
#[tokio::test]
async fn alternative_transports_use_the_shared_connection_type() {
let registry = Registry::builtin();

268
host/src/pipe.rs
View file

@ -1,25 +1,106 @@
use mtp_codec::{CommunicationType, CommunicationValue, DataType, DataValue};
use mtp_common::PipeError;
use mtp_transport::{PipeReader, Policy, Receiver, Sender};
use mtp_common::{CommunicationError, PipeError};
use mtp_transport::{PipeReader, PipeWriter, Policy, TransportEvent};
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::{Mutex, mpsc};
use tracing::debug;
/// The sender operations needed by the transport-independent pipe protocol.
pub trait PipeSender: Clone + Send + Sync + 'static {
type Writer: tokio::io::AsyncWrite + Send + Unpin + 'static;
pub struct PipeHandle {
fn send_pipe_message(
&self,
message: &CommunicationValue,
) -> impl std::future::Future<Output = Result<(), CommunicationError>> + Send;
fn open_pipe_stream(
&self,
pipe_id: u32,
description: &str,
) -> impl std::future::Future<Output = Result<PipeWriter<Self::Writer>, CommunicationError>> + Send;
}
/// The receiver operations needed by the transport-independent pipe protocol.
pub trait PipeReceiver<P>: Clone + Send + Sync + 'static
where
P: tokio::io::AsyncRead + Send + Unpin + 'static,
{
fn receive_pipe_event(
&self,
) -> impl std::future::Future<Output = Result<TransportEvent<P>, CommunicationError>> + Send;
}
impl PipeSender for mtp_transport::Sender {
type Writer = wtransport::SendStream;
async fn send_pipe_message(
&self,
message: &CommunicationValue,
) -> Result<(), CommunicationError> {
self.send(message).await
}
async fn open_pipe_stream(
&self,
pipe_id: u32,
description: &str,
) -> Result<PipeWriter<Self::Writer>, CommunicationError> {
self.open_pipe(pipe_id, description).await
}
}
impl PipeReceiver<wtransport::RecvStream> for mtp_transport::Receiver {
async fn receive_pipe_event(
&self,
) -> Result<TransportEvent<wtransport::RecvStream>, CommunicationError> {
self.receive_event().await
}
}
impl<C> PipeSender for mtp_transport::GenericSender<C>
where
C: mtp_transport::TransportConnection,
C::SendStream: tokio::io::AsyncWrite + Send + Unpin + 'static,
{
type Writer = C::SendStream;
async fn send_pipe_message(
&self,
message: &CommunicationValue,
) -> Result<(), CommunicationError> {
self.send(message).await
}
async fn open_pipe_stream(
&self,
pipe_id: u32,
description: &str,
) -> Result<PipeWriter<Self::Writer>, CommunicationError> {
self.open_pipe(pipe_id, description).await
}
}
impl<C> PipeReceiver<C::RecvStream> for mtp_transport::GenericReceiver<C>
where
C: mtp_transport::TransportConnection,
C::RecvStream: tokio::io::AsyncRead + Send + Unpin + 'static,
{
async fn receive_pipe_event(
&self,
) -> Result<TransportEvent<C::RecvStream>, CommunicationError> {
self.receive_event().await
}
}
pub struct PipeHandle<S: PipeSender> {
pub(crate) pipe_id: u32,
pub(crate) description: String,
pub(crate) sender: Sender,
pub(crate) sender: S,
pub(crate) response_rx: tokio::sync::oneshot::Receiver<Result<bool, PipeError>>,
}
impl PipeHandle {
impl<S: PipeSender> PipeHandle<S> {
pub fn pipe_id(&self) -> u32 {
self.pipe_id
}
@ -28,31 +109,33 @@ impl PipeHandle {
&self.description
}
pub async fn wait(self) -> Result<Option<mtp_transport::PipeWriter>, PipeError> {
pub async fn wait(self) -> Result<Option<PipeWriter<S::Writer>>, 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(true)) => self
.sender
.open_pipe_stream(self.pipe_id, &self.description)
.await
.map(Some)
.map_err(PipeError::from),
Ok(Ok(false)) => Ok(None),
Ok(Err(e)) => Err(e),
Ok(Err(error)) => Err(error),
Err(_) => Err(PipeError::StreamClosed),
}
}
}
pub struct PipeRequest {
pub struct PipeRequest<S, P> {
pub(crate) pipe_id: u32,
pub(crate) description: String,
pub(crate) sender: Sender,
pub(crate) dispatcher: Arc<PipeDispatcher>,
pub(crate) sender: S,
pub(crate) dispatcher: Arc<PipeDispatcher<P>>,
}
impl PipeRequest {
impl<S, P> PipeRequest<S, P>
where
S: PipeSender,
P: tokio::io::AsyncRead + Send + Unpin + 'static,
{
pub fn id(&self) -> u32 {
self.pipe_id
}
@ -61,133 +144,106 @@ impl PipeRequest {
&self.description
}
pub async fn accept(self) -> Result<PipeReader, PipeError> {
pub async fn accept(self) -> Result<PipeReader<P>, 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);
}
self.dispatcher
.pending_pipes
.lock()
.await
.insert(self.pipe_id, pipe_tx);
let resp = CommunicationValue::new(CommunicationType::PipeResponse)
let response = 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)?;
self.sender
.send_pipe_message(&response)
.await
.map_err(PipeError::from)?;
let timeout = self.dispatcher.policy.read_timeout;
tokio::time::timeout(timeout, pipe_rx)
tokio::time::timeout(self.dispatcher.policy.read_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)
let response = 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(())
self.sender
.send_pipe_message(&response)
.await
.map_err(PipeError::from)
}
}
pub(crate) struct PipeDispatcher {
pub(crate) struct PipeDispatcher<P> {
pub(crate) pending_creations:
Mutex<HashMap<u32, tokio::sync::oneshot::Sender<Result<bool, PipeError>>>>,
pub(crate) pending_pipes: Mutex<HashMap<u32, tokio::sync::oneshot::Sender<PipeReader>>>,
pub(crate) pending_pipes: Mutex<HashMap<u32, tokio::sync::oneshot::Sender<PipeReader<P>>>>,
pub(crate) policy: Arc<Policy>,
}
impl PipeDispatcher {
pub(crate) fn default_for_external() -> Self {
Self {
pending_creations: Mutex::new(HashMap::new()),
pending_pipes: Mutex::new(HashMap::new()),
policy: Arc::new(Policy::default()),
}
}
}
pub(crate) async fn run_dispatcher(
receiver: Receiver,
sender: Sender,
app_tx: mpsc::Sender<Result<CommunicationValue, mtp_common::CommunicationError>>,
pipe_req_tx: mpsc::Sender<PipeRequest>,
dispatcher: Arc<PipeDispatcher>,
) {
pub(crate) async fn run_dispatcher<S, R, P>(
receiver: R,
sender: S,
app_tx: mpsc::Sender<Result<CommunicationValue, CommunicationError>>,
pipe_req_tx: mpsc::Sender<PipeRequest<S, P>>,
dispatcher: Arc<PipeDispatcher<P>>,
) where
S: PipeSender,
R: PipeReceiver<P>,
P: tokio::io::AsyncRead + Send + Unpin + 'static,
{
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)) => {
debug!(
target = "mtp.host",
message_type = ?msg.get_type(),
message_id = msg.get_id(),
"dispatcher received message"
);
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();
debug!(
target = "mtp.host",
pipe_id, description, "dispatcher classified pipe request"
);
let req = PipeRequest {
pipe_id,
description,
match receiver.receive_pipe_event().await {
Ok(TransportEvent::Message(message)) => {
if Some(message.get_type()) == pipe_req_type {
let request = PipeRequest {
pipe_id: message.get_id(),
description: message
.get_str(DataType::Description)
.unwrap_or("")
.to_owned(),
sender: sender.clone(),
dispatcher: dispatcher.clone(),
};
let _ = pipe_req_tx.send(req).await;
let _ = pipe_req_tx.send(request).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);
debug!(
target = "mtp.host",
pipe_id, accepted, "dispatcher classified pipe response"
);
if Some(message.get_type()) == pipe_resp_type {
let mut pending = dispatcher.pending_creations.lock().await;
if let Some(tx) = pending.remove(&pipe_id) {
let _ = tx.send(Ok(accepted));
if let Some(reply) = pending.remove(&message.get_id()) {
let _ =
reply.send(Ok(message.get_bool(DataType::Accepted).unwrap_or(false)));
}
continue;
}
if app_tx.send(Ok(msg)).await.is_err() {
if app_tx.send(Ok(message)).await.is_err() {
break;
}
}
Ok(mtp_transport::TransportEvent::Pipe(reader)) => {
Ok(TransportEvent::Pipe(reader)) => {
let pipe_id = reader.pipe_id();
debug!(
target = "mtp.host",
pipe_id,
description = reader.description(),
"dispatcher received pipe stream"
);
let mut pending = dispatcher.pending_pipes.lock().await;
if let Some(tx) = pending.remove(&pipe_id) {
let _ = tx.send(reader);
if let Some(reply) = pending.remove(&pipe_id) {
let _ = reply.send(reader);
continue;
}
debug!(
target = "mtp.host",
pipe_id, "dispatcher treating pipe stream as pipe request"
);
let req = PipeRequest {
drop(pending);
let request = PipeRequest {
pipe_id,
description: reader.description().to_string(),
description: reader.description().to_owned(),
sender: sender.clone(),
dispatcher: dispatcher.clone(),
};
let _ = pipe_req_tx.send(req).await;
let _ = pipe_req_tx.send(request).await;
}
Err(e) => {
if app_tx.send(Err(e)).await.is_err() {
Err(error) => {
if app_tx.send(Err(error)).await.is_err() {
break;
}
}