From 188caf56cc994910bd05ea437ee45bd45d3e3027 Mon Sep 17 00:00:00 2001 From: Alex Emmet <111742636+Alex-Emmet@users.noreply.github.com> Date: Fri, 14 Aug 2026 14:39:09 +0200 Subject: [PATCH] [Fix] Clean --- Cargo.lock | 1 - client/src/pipe.rs | 9 +- codec/src/lib.rs | 4 +- codec/src/protected.rs | 172 ++++++++++++++------------------- example/Cargo.lock | 1 - example/server/src/handlers.rs | 12 ++- host/src/engine.rs | 28 +++--- mtp-webserver/Cargo.toml | 3 +- mtp-webserver/src/transport.rs | 2 + transport/src/connection.rs | 23 ++--- transport/tests/integration.rs | 12 ++- wasm/src/protected.rs | 14 +-- 12 files changed, 124 insertions(+), 157 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 93d18c9..a6c3d32 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1485,7 +1485,6 @@ dependencies = [ "mtp-host", "mtp-transport", "quinn", - "rand", "rcgen", "rustls", "thiserror 2.0.20", diff --git a/client/src/pipe.rs b/client/src/pipe.rs index 4dc590a..f9944e8 100644 --- a/client/src/pipe.rs +++ b/client/src/pipe.rs @@ -200,14 +200,13 @@ pub(crate) async fn expire_pending_request( let mut expired = dispatcher.expired_requests.lock().await; let now = Instant::now(); expired.retain(|_, expires_at| *expires_at > now); - if expired.len() >= MAX_EXPIRED_REQUEST_TOMBSTONES { - if let Some(oldest) = expired + if expired.len() >= MAX_EXPIRED_REQUEST_TOMBSTONES + && let Some(oldest) = expired .iter() .min_by_key(|(_, expires_at)| **expires_at) .map(|(id, _)| *id) - { - expired.remove(&oldest); - } + { + expired.remove(&oldest); } expired.insert(request_id, now + EXPIRED_REQUEST_TOMBSTONE_TTL); } diff --git a/codec/src/lib.rs b/codec/src/lib.rs index 7249251..c910cd8 100644 --- a/codec/src/lib.rs +++ b/codec/src/lib.rs @@ -16,8 +16,8 @@ pub use mtp_common::{CodecError, TimeError, unix_time_millis}; #[cfg(feature = "crypto")] pub use protected::{ CURRENT_PROTECTED_VERSION, InMemoryReplayGuard, ProtectedError, ProtectedMessageBuilder, - ReplayError, ReplayGuard, VerifiedProtectedMessage, open_protected, open_protected_with, - open_protected_with_keys, protected_claimed_signer_id, + ProtectedOpenOptions, ReplayError, ReplayGuard, VerifiedProtectedMessage, open_protected, + open_protected_with, open_protected_with_keys, protected_claimed_signer_id, }; #[cfg(feature = "crypto")] pub use relay::{ diff --git a/codec/src/protected.rs b/codec/src/protected.rs index f423ec9..e68689f 100644 --- a/codec/src/protected.rs +++ b/codec/src/protected.rs @@ -247,6 +247,35 @@ pub struct VerifiedProtectedMessage { pub matched_signer_key_index: usize, } +/// Options that control verification of a direct protected message. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub struct ProtectedOpenOptions { + /// Require the protected frame to be addressed to this receiver when set. + pub expected_receiver_id: Option, + /// Purpose used to verify the protected envelope signature. + pub signature_purpose: ProtectionPurpose, + /// Purpose used to decrypt the protected envelope. + pub encryption_purpose: ProtectionPurpose, + /// Signature algorithms accepted by the receiver. + pub policy: ProtectionPolicy, +} + +impl ProtectedOpenOptions { + pub const fn new( + expected_receiver_id: Option, + signature_purpose: ProtectionPurpose, + encryption_purpose: ProtectionPurpose, + policy: ProtectionPolicy, + ) -> Self { + Self { + expected_receiver_id, + signature_purpose, + encryption_purpose, + policy, + } + } +} + fn protected_field_id( data_type: DataType, type_map: &TypeMap, @@ -385,10 +414,7 @@ pub fn open_protected_with( keyrings: &[&Keyring], expected_signer_id: Option, resolve_signer_keys: F, - expected_receiver_id: Option, - signature_purpose: ProtectionPurpose, - encryption_purpose: ProtectionPurpose, - policy: ProtectionPolicy, + options: ProtectedOpenOptions, replay_guard: Option<&mut dyn ReplayGuard>, ) -> Result where @@ -396,7 +422,7 @@ where { validate_protected_frame(frame)?; let type_map = frame.type_map().cloned().unwrap_or_else(TypeMap::latest); - let decrypted = decrypt_protected_payload(frame, keyrings, encryption_purpose)?; + let decrypted = decrypt_protected_payload(frame, keyrings, options.encryption_purpose)?; let signed = decrypted .as_signed() .ok_or(ProtectedError::PayloadNotSigned)?; @@ -411,16 +437,7 @@ where } let signer_keys = resolve_signer_keys(signed.signer_id) .ok_or(ProtectionError::SignerKeyNotFound(signed.signer_id))?; - open_decrypted_protected( - frame, - type_map, - signed, - &signer_keys, - expected_receiver_id, - signature_purpose, - policy, - replay_guard, - ) + open_decrypted_protected(frame, type_map, signed, &signer_keys, options, replay_guard) } /// Open a direct protected message against already resolved trusted signer @@ -431,14 +448,11 @@ pub fn open_protected_with_keys( keyrings: &[&Keyring], expected_signer_id: u64, signer_public_keys: &[PublicKeyBundle], - expected_receiver_id: Option, - signature_purpose: ProtectionPurpose, - encryption_purpose: ProtectionPurpose, - policy: ProtectionPolicy, + options: ProtectedOpenOptions, replay_guard: Option<&mut dyn ReplayGuard>, ) -> Result { let type_map = validate_protected_frame(frame)?; - let decrypted = decrypt_protected_payload(frame, keyrings, encryption_purpose)?; + let decrypted = decrypt_protected_payload(frame, keyrings, options.encryption_purpose)?; let signed = decrypted .as_signed() .ok_or(ProtectedError::PayloadNotSigned)?; @@ -454,9 +468,7 @@ pub fn open_protected_with_keys( type_map, signed, signer_public_keys, - expected_receiver_id, - signature_purpose, - policy, + options, replay_guard, ) } @@ -468,21 +480,15 @@ pub fn open_protected( keyring: &Keyring, expected_signer_id: u64, signer_public_key: &PublicKeyBundle, - expected_receiver_id: Option, - signature_purpose: ProtectionPurpose, - encryption_purpose: ProtectionPurpose, - policy: ProtectionPolicy, + options: ProtectedOpenOptions, replay_guard: Option<&mut dyn ReplayGuard>, ) -> Result { open_protected_with_keys( frame, std::slice::from_ref(&keyring), expected_signer_id, - std::slice::from_ref(&signer_public_key), - expected_receiver_id, - signature_purpose, - encryption_purpose, - policy, + std::slice::from_ref(signer_public_key), + options, replay_guard, ) } @@ -492,19 +498,20 @@ fn open_decrypted_protected( type_map: TypeMap, signed: &crate::SignedValue, signer_public_keys: &[PublicKeyBundle], - expected_receiver_id: Option, - signature_purpose: ProtectionPurpose, - policy: ProtectionPolicy, + options: ProtectedOpenOptions, mut replay_guard: Option<&mut dyn ReplayGuard>, ) -> Result { let matched_signer_key_index = signed.verify_with_key_history_index( signed.signer_id, signer_public_keys, - signature_purpose, - policy, + options.signature_purpose, + options.policy, )?; let receiver_id = frame.receiver().ok_or(ProtectedError::MissingReceiver)?; - if expected_receiver_id.is_some_and(|expected| expected != receiver_id) { + if options + .expected_receiver_id + .is_some_and(|expected| expected != receiver_id) + { return Err(ProtectedError::ExpectedReceiverMismatch); } if frame @@ -568,6 +575,15 @@ mod tests { const SIGNATURE_PURPOSE: ProtectionPurpose = ProtectionPurpose(0x40); const ENCRYPTION_PURPOSE: ProtectionPurpose = ProtectionPurpose(0x41); + fn open_options(expected_receiver_id: Option) -> ProtectedOpenOptions { + ProtectedOpenOptions::new( + expected_receiver_id, + SIGNATURE_PURPOSE, + ENCRYPTION_PURPOSE, + ProtectionPolicy::from(crate::SignaturePolicy::Ed25519), + ) + } + #[derive(Default)] struct RecordingReplayGuard { created_at: Option, @@ -720,10 +736,7 @@ mod tests { &recipient, 7, &sender.public_key_bundle(), - Some(42), - SIGNATURE_PURPOSE, - ENCRYPTION_PURPOSE, - ProtectionPolicy::from(crate::SignaturePolicy::Ed25519), + open_options(Some(42)), Some(&mut guard), ) .expect("protected message should open"); @@ -737,10 +750,7 @@ mod tests { &recipient, 7, &sender.public_key_bundle(), - Some(42), - SIGNATURE_PURPOSE, - ENCRYPTION_PURPOSE, - ProtectionPolicy::from(crate::SignaturePolicy::Ed25519), + open_options(Some(42)), Some(&mut guard), ), Err(ProtectedError::Replay) @@ -777,10 +787,7 @@ mod tests { &recipient, 7, &sender.public_key_bundle(), - Some(42), - SIGNATURE_PURPOSE, - ENCRYPTION_PURPOSE, - ProtectionPolicy::from(crate::SignaturePolicy::Ed25519), + open_options(Some(42)), None, ) .expect("outer fields should verify"); @@ -813,10 +820,7 @@ mod tests { &recipient, 7, &sender.public_key_bundle(), - Some(42), - SIGNATURE_PURPOSE, - ENCRYPTION_PURPOSE, - ProtectionPolicy::from(crate::SignaturePolicy::Ed25519), + open_options(Some(42)), None, ), Err(ProtectedError::MissingProtectedVersion) @@ -844,10 +848,7 @@ mod tests { &recipient, 7, &sender.public_key_bundle(), - Some(42), - SIGNATURE_PURPOSE, - ENCRYPTION_PURPOSE, - ProtectionPolicy::from(crate::SignaturePolicy::Ed25519), + open_options(Some(42)), None, ), Err(ProtectedError::UnsupportedProtectedVersion(2)) @@ -897,10 +898,7 @@ mod tests { &recipient, 7, &sender.public_key_bundle(), - None, - SIGNATURE_PURPOSE, - ENCRYPTION_PURPOSE, - ProtectionPolicy::from(crate::SignaturePolicy::Ed25519), + open_options(None), None, ), Err(ProtectedError::MessageTypeMismatch) @@ -913,10 +911,7 @@ mod tests { &recipient, 7, &sender.public_key_bundle(), - None, - SIGNATURE_PURPOSE, - ENCRYPTION_PURPOSE, - ProtectionPolicy::from(crate::SignaturePolicy::Ed25519), + open_options(None), None, ), Err(ProtectedError::FinalRecipientMismatch) @@ -944,10 +939,7 @@ mod tests { &recipient, 7, &sender.public_key_bundle(), - None, - SIGNATURE_PURPOSE, - ENCRYPTION_PURPOSE, - ProtectionPolicy::from(crate::SignaturePolicy::Ed25519), + open_options(None), None, ), Err(ProtectedError::FinalRecipientMismatch) @@ -959,10 +951,7 @@ mod tests { &recipient, 7, &sender.public_key_bundle(), - Some(43), - SIGNATURE_PURPOSE, - ENCRYPTION_PURPOSE, - ProtectionPolicy::from(crate::SignaturePolicy::Ed25519), + open_options(Some(43)), None, ), Err(ProtectedError::ExpectedReceiverMismatch) @@ -994,10 +983,7 @@ mod tests { &recipient, 7, &sender.public_key_bundle(), - Some(42), - SIGNATURE_PURPOSE, - ENCRYPTION_PURPOSE, - ProtectionPolicy::from(crate::SignaturePolicy::Ed25519), + open_options(Some(42)), None, ) .expect("matching exposed sender"); @@ -1007,10 +993,7 @@ mod tests { &recipient, 7, &sender.public_key_bundle(), - Some(42), - SIGNATURE_PURPOSE, - ENCRYPTION_PURPOSE, - ProtectionPolicy::from(crate::SignaturePolicy::Ed25519), + open_options(Some(42)), None, ), Err(ProtectedError::SenderMismatch) @@ -1049,10 +1032,7 @@ mod tests { &recipient, 7, &sender.public_key_bundle(), - Some(42), - SIGNATURE_PURPOSE, - ENCRYPTION_PURPOSE, - ProtectionPolicy::from(crate::SignaturePolicy::Ed25519), + open_options(Some(42)), None, ), Err(ProtectedError::PayloadNotEncrypted) @@ -1066,10 +1046,7 @@ mod tests { &recipient, 7, &sender.public_key_bundle(), - Some(42), - SIGNATURE_PURPOSE, - ENCRYPTION_PURPOSE, - ProtectionPolicy::from(crate::SignaturePolicy::Ed25519), + open_options(Some(42)), None, ), Err(ProtectedError::PayloadNotSigned) @@ -1090,10 +1067,7 @@ mod tests { resolver_calls += 1; Some(vec![sender.public_key_bundle()]) }, - Some(42), - SIGNATURE_PURPOSE, - ENCRYPTION_PURPOSE, - ProtectionPolicy::from(crate::SignaturePolicy::Ed25519), + open_options(Some(42)), None, ); assert!(matches!( @@ -1138,10 +1112,7 @@ mod tests { current_sender.public_key_bundle(), old_sender.public_key_bundle(), ], - Some(42), - SIGNATURE_PURPOSE, - ENCRYPTION_PURPOSE, - ProtectionPolicy::from(crate::SignaturePolicy::Ed25519), + open_options(Some(42)), None, ) .expect("key history should open"); @@ -1167,10 +1138,7 @@ mod tests { &recipient, 7, &sender.public_key_bundle(), - Some(42), - SIGNATURE_PURPOSE, - ENCRYPTION_PURPOSE, - ProtectionPolicy::from(crate::SignaturePolicy::Ed25519), + open_options(Some(42)), None, ) .expect("arbitrary application value should open"); diff --git a/example/Cargo.lock b/example/Cargo.lock index 1c7fa26..1c15380 100644 --- a/example/Cargo.lock +++ b/example/Cargo.lock @@ -1404,7 +1404,6 @@ dependencies = [ "mtp-host", "mtp-transport", "quinn", - "rand", "rustls", "thiserror 2.0.20", "tokio", diff --git a/example/server/src/handlers.rs b/example/server/src/handlers.rs index 65dfedd..1e6320b 100644 --- a/example/server/src/handlers.rs +++ b/example/server/src/handlers.rs @@ -2,7 +2,7 @@ use std::collections::HashMap; use mtp::codec::{ CommunicationType, CommunicationValue, DataType, DataTypeId, DataValue, InMemoryReplayGuard, - ProtectionPolicy, ProtectionPurpose, SignaturePolicy, TypeMap, + ProtectedOpenOptions, ProtectionPolicy, ProtectionPurpose, SignaturePolicy, TypeMap, forward_relay_frame, open_protected_with, open_relay_content, open_relay_metadata_with, }; use mtp::crypto::{Keyring, PublicKeyBundle}; @@ -70,10 +70,12 @@ fn process_direct_protected( std::slice::from_ref(&host_keyring), None, |signer_id| resolve_signer_key(signer_id, registered_clients).map(|key| vec![key]), - Some(DIRECT_DESTINATION_ID), - ProtectionPurpose::from(DIRECT_SIGNATURE_PURPOSE), - ProtectionPurpose::from(DIRECT_ENCRYPTION_PURPOSE), - SIGNATURE_POLICY, + ProtectedOpenOptions::new( + Some(DIRECT_DESTINATION_ID), + ProtectionPurpose::from(DIRECT_SIGNATURE_PURPOSE), + ProtectionPurpose::from(DIRECT_ENCRYPTION_PURPOSE), + SIGNATURE_POLICY, + ), Some(accepted_messages), ) .map_err(|e| format!("direct protected message could not be authenticated: {e}"))?; diff --git a/host/src/engine.rs b/host/src/engine.rs index 078c2f1..cd16934 100644 --- a/host/src/engine.rs +++ b/host/src/engine.rs @@ -344,7 +344,7 @@ impl HandshakeEngine { // Authenticated clients include PublicKeys in Identification as an // intent marker; this avoids acknowledging the opening as a guest // connection and leaving the client waiting for a Challenge. - if Some(first_msg.get_type()) == CommunicationType::Register.try_to_id(&tm) + if Some(first_msg.get_type()) == CommunicationType::Register.try_to_id(tm) || first_msg.get_data(DataType::PublicKeys).is_some() { send_rejection_generic( @@ -400,7 +400,7 @@ impl HandshakeEngine { let tm = codec.type_map(); // Register frames always go through full authentication - if Some(first_msg.get_type()) == CommunicationType::Register.try_to_id(&tm) { + if Some(first_msg.get_type()) == CommunicationType::Register.try_to_id(tm) { let bundle = match extract_register_bundle(&first_msg) { Ok(bundle) => bundle, Err(error) => { @@ -425,7 +425,7 @@ impl HandshakeEngine { } // Identification: try lookup, fall back to guest - if Some(first_msg.get_type()) == CommunicationType::Identification.try_to_id(&tm) { + if Some(first_msg.get_type()) == CommunicationType::Identification.try_to_id(tm) { let cid = match first_msg.get_data(DataType::Id) { Some(DataValue::UnsignedNumber(n)) => u64::try_from(*n).unwrap_or(0), _ => 0, @@ -504,7 +504,7 @@ impl HandshakeEngine { let tm = codec.type_map(); let (flow, response_type) = if Some(first_msg.get_type()) - == CommunicationType::Identification.try_to_id(&tm) + == CommunicationType::Identification.try_to_id(tm) { let cid = match first_msg.get_data(DataType::Id) { Some(DataValue::UnsignedNumber(n)) => match u64::try_from(*n) { @@ -545,7 +545,7 @@ impl HandshakeEngine { Flow::Login { id: cid, bundle }, CommunicationType::IdentificationResponse, ) - } else if Some(first_msg.get_type()) == CommunicationType::Register.try_to_id(&tm) { + } else if Some(first_msg.get_type()) == CommunicationType::Register.try_to_id(tm) { let bundle = match extract_register_bundle(&first_msg) { Ok(bundle) => bundle, Err(error) => { @@ -710,7 +710,7 @@ impl HandshakeEngine { sender.close(); AcceptError::Receive(e) })?; - if Some(proof.get_type()) != CommunicationType::ChallengeResponse.try_to_id(&tm) { + if Some(proof.get_type()) != CommunicationType::ChallengeResponse.try_to_id(tm) { let error = AcceptError::AuthenticationFailed("missing challenge response".into()); reject_error_generic(sender, &error, tm).await; return Err(error); @@ -1041,8 +1041,8 @@ impl HandshakeSender for mtp_transport::Sender { ) -> impl std::future::Future> + Send { mtp_transport::Sender::finish_stream(self) } - fn set_type_map(&self, type_map: &TypeMap) -> impl std::future::Future + Send { - async move { self.set_type_map(type_map).await } + async fn set_type_map(&self, type_map: &TypeMap) { + self.set_type_map(type_map).await; } fn close(&self) { let sender = self.clone(); @@ -1058,8 +1058,8 @@ impl HandshakeReceiver for mtp_transport::Receiver { mtp_transport::Receiver::receive(self) } - fn set_type_map(&self, type_map: &TypeMap) -> impl std::future::Future + Send { - async move { self.set_type_map(type_map).await } + async fn set_type_map(&self, type_map: &TypeMap) { + self.set_type_map(type_map).await; } } @@ -1075,8 +1075,8 @@ impl HandshakeSender for mtp_transport::G ) -> impl std::future::Future> + Send { mtp_transport::GenericSender::finish_stream(self) } - fn set_type_map(&self, type_map: &TypeMap) -> impl std::future::Future + Send { - async move { self.set_type_map(type_map).await } + async fn set_type_map(&self, type_map: &TypeMap) { + self.set_type_map(type_map).await; } fn close(&self) { mtp_transport::GenericSender::close(self); @@ -1093,8 +1093,8 @@ impl HandshakeReceiver mtp_transport::GenericReceiver::receive(self) } - fn set_type_map(&self, type_map: &TypeMap) -> impl std::future::Future + Send { - async move { self.set_type_map(type_map).await } + async fn set_type_map(&self, type_map: &TypeMap) { + self.set_type_map(type_map).await; } } diff --git a/mtp-webserver/Cargo.toml b/mtp-webserver/Cargo.toml index 88f7bb1..6b5ef4e 100644 --- a/mtp-webserver/Cargo.toml +++ b/mtp-webserver/Cargo.toml @@ -25,7 +25,6 @@ rustls = "0.23" tracing = "0.1" thiserror = "2" async-trait = "0.1" -rand = { version = "0.10.1", optional = true } [dev-dependencies] rcgen = "0.14" @@ -33,5 +32,5 @@ hyper = { version = "1", features = ["client", "http2"] } [features] default = [] -crypto = ["mtp-host/crypto", "dep:rand"] +crypto = ["mtp-host/crypto"] pipes = ["mtp-host/pipes", "mtp-transport/pipes"] diff --git a/mtp-webserver/src/transport.rs b/mtp-webserver/src/transport.rs index 3834f64..5c81765 100644 --- a/mtp-webserver/src/transport.rs +++ b/mtp-webserver/src/transport.rs @@ -220,6 +220,7 @@ pub type WebMtpReceiver = GenericReceiver; pub type WebMTPConnection = mtp_host::MTPConnection; +#[allow(clippy::too_many_arguments)] pub(crate) async fn accept_web_connection( session: Arc, path: String, @@ -268,6 +269,7 @@ pub(crate) async fn accept_web_connection( .await } +#[allow(clippy::too_many_arguments)] async fn accept_web_connection_inner( session: Arc, path: String, diff --git a/transport/src/connection.rs b/transport/src/connection.rs index 7f0b5d6..c21bd67 100644 --- a/transport/src/connection.rs +++ b/transport/src/connection.rs @@ -1078,18 +1078,12 @@ impl Receiver { #[instrument(skip(self), level = "trace")] pub async fn receive(&self) -> Result { let mut close_rx = self.inner.handle.subscribe_close(); - if close_rx.borrow().is_some() { - return Err(self - .inner - .handle - .close_reason() - .unwrap_or(CommunicationError::StreamClosed)); - } #[cfg(feature = "pipes")] { let mut rx = self.inner.msg_rx.lock().await; let result = tokio::select! { + biased; message = rx.recv() => message, _ = close_rx.changed() => return Err(close_rx .borrow() @@ -1113,6 +1107,7 @@ impl Receiver { { let mut rx = self.inner.rx.lock().await; let result = tokio::select! { + biased; message = rx.recv() => message, _ = close_rx.changed() => return Err(close_rx .borrow() @@ -1136,17 +1131,11 @@ impl Receiver { #[cfg(feature = "pipes")] #[instrument(skip(self), level = "trace")] pub async fn receive_event(&self) -> Result { - if self.inner.handle.is_closed() { - return Err(self - .inner - .handle - .close_reason() - .unwrap_or(CommunicationError::StreamClosed)); - } - + let mut close_rx = self.inner.handle.subscribe_close(); let mut msg_rx = self.inner.msg_rx.lock().await; let mut pipe_rx = self.inner.pipe_rx.lock().await; tokio::select! { + biased; msg = msg_rx.recv() => { match msg { Some(Ok(val)) => { @@ -1174,6 +1163,10 @@ impl Receiver { .unwrap_or(CommunicationError::StreamClosed)), } } + _ = close_rx.changed() => Err(close_rx + .borrow() + .clone() + .unwrap_or(CommunicationError::StreamClosed)), } } diff --git a/transport/tests/integration.rs b/transport/tests/integration.rs index 7028528..3154747 100644 --- a/transport/tests/integration.rs +++ b/transport/tests/integration.rs @@ -342,13 +342,17 @@ async fn test_max_frames_per_stream_enforced() -> Result<(), Box