diff --git a/Cargo.lock b/Cargo.lock index a6c3d32..93d18c9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1485,6 +1485,7 @@ 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 f9944e8..4dc590a 100644 --- a/client/src/pipe.rs +++ b/client/src/pipe.rs @@ -200,13 +200,14 @@ 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 - && let Some(oldest) = expired + if expired.len() >= MAX_EXPIRED_REQUEST_TOMBSTONES { + if 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 c910cd8..7249251 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, - ProtectedOpenOptions, ReplayError, ReplayGuard, VerifiedProtectedMessage, open_protected, - open_protected_with, open_protected_with_keys, protected_claimed_signer_id, + 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 e68689f..f423ec9 100644 --- a/codec/src/protected.rs +++ b/codec/src/protected.rs @@ -247,35 +247,6 @@ 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, @@ -414,7 +385,10 @@ pub fn open_protected_with( keyrings: &[&Keyring], expected_signer_id: Option, resolve_signer_keys: F, - options: ProtectedOpenOptions, + expected_receiver_id: Option, + signature_purpose: ProtectionPurpose, + encryption_purpose: ProtectionPurpose, + policy: ProtectionPolicy, replay_guard: Option<&mut dyn ReplayGuard>, ) -> Result where @@ -422,7 +396,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, options.encryption_purpose)?; + let decrypted = decrypt_protected_payload(frame, keyrings, encryption_purpose)?; let signed = decrypted .as_signed() .ok_or(ProtectedError::PayloadNotSigned)?; @@ -437,7 +411,16 @@ 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, options, replay_guard) + open_decrypted_protected( + frame, + type_map, + signed, + &signer_keys, + expected_receiver_id, + signature_purpose, + policy, + replay_guard, + ) } /// Open a direct protected message against already resolved trusted signer @@ -448,11 +431,14 @@ pub fn open_protected_with_keys( keyrings: &[&Keyring], expected_signer_id: u64, signer_public_keys: &[PublicKeyBundle], - options: ProtectedOpenOptions, + expected_receiver_id: Option, + signature_purpose: ProtectionPurpose, + encryption_purpose: ProtectionPurpose, + policy: ProtectionPolicy, replay_guard: Option<&mut dyn ReplayGuard>, ) -> Result { let type_map = validate_protected_frame(frame)?; - let decrypted = decrypt_protected_payload(frame, keyrings, options.encryption_purpose)?; + let decrypted = decrypt_protected_payload(frame, keyrings, encryption_purpose)?; let signed = decrypted .as_signed() .ok_or(ProtectedError::PayloadNotSigned)?; @@ -468,7 +454,9 @@ pub fn open_protected_with_keys( type_map, signed, signer_public_keys, - options, + expected_receiver_id, + signature_purpose, + policy, replay_guard, ) } @@ -480,15 +468,21 @@ pub fn open_protected( keyring: &Keyring, expected_signer_id: u64, signer_public_key: &PublicKeyBundle, - options: ProtectedOpenOptions, + expected_receiver_id: Option, + signature_purpose: ProtectionPurpose, + encryption_purpose: ProtectionPurpose, + policy: ProtectionPolicy, 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), - options, + std::slice::from_ref(&signer_public_key), + expected_receiver_id, + signature_purpose, + encryption_purpose, + policy, replay_guard, ) } @@ -498,20 +492,19 @@ fn open_decrypted_protected( type_map: TypeMap, signed: &crate::SignedValue, signer_public_keys: &[PublicKeyBundle], - options: ProtectedOpenOptions, + expected_receiver_id: Option, + signature_purpose: ProtectionPurpose, + policy: ProtectionPolicy, 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, - options.signature_purpose, - options.policy, + signature_purpose, + policy, )?; let receiver_id = frame.receiver().ok_or(ProtectedError::MissingReceiver)?; - if options - .expected_receiver_id - .is_some_and(|expected| expected != receiver_id) - { + if expected_receiver_id.is_some_and(|expected| expected != receiver_id) { return Err(ProtectedError::ExpectedReceiverMismatch); } if frame @@ -575,15 +568,6 @@ 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, @@ -736,7 +720,10 @@ mod tests { &recipient, 7, &sender.public_key_bundle(), - open_options(Some(42)), + Some(42), + SIGNATURE_PURPOSE, + ENCRYPTION_PURPOSE, + ProtectionPolicy::from(crate::SignaturePolicy::Ed25519), Some(&mut guard), ) .expect("protected message should open"); @@ -750,7 +737,10 @@ mod tests { &recipient, 7, &sender.public_key_bundle(), - open_options(Some(42)), + Some(42), + SIGNATURE_PURPOSE, + ENCRYPTION_PURPOSE, + ProtectionPolicy::from(crate::SignaturePolicy::Ed25519), Some(&mut guard), ), Err(ProtectedError::Replay) @@ -787,7 +777,10 @@ mod tests { &recipient, 7, &sender.public_key_bundle(), - open_options(Some(42)), + Some(42), + SIGNATURE_PURPOSE, + ENCRYPTION_PURPOSE, + ProtectionPolicy::from(crate::SignaturePolicy::Ed25519), None, ) .expect("outer fields should verify"); @@ -820,7 +813,10 @@ mod tests { &recipient, 7, &sender.public_key_bundle(), - open_options(Some(42)), + Some(42), + SIGNATURE_PURPOSE, + ENCRYPTION_PURPOSE, + ProtectionPolicy::from(crate::SignaturePolicy::Ed25519), None, ), Err(ProtectedError::MissingProtectedVersion) @@ -848,7 +844,10 @@ mod tests { &recipient, 7, &sender.public_key_bundle(), - open_options(Some(42)), + Some(42), + SIGNATURE_PURPOSE, + ENCRYPTION_PURPOSE, + ProtectionPolicy::from(crate::SignaturePolicy::Ed25519), None, ), Err(ProtectedError::UnsupportedProtectedVersion(2)) @@ -898,7 +897,10 @@ mod tests { &recipient, 7, &sender.public_key_bundle(), - open_options(None), + None, + SIGNATURE_PURPOSE, + ENCRYPTION_PURPOSE, + ProtectionPolicy::from(crate::SignaturePolicy::Ed25519), None, ), Err(ProtectedError::MessageTypeMismatch) @@ -911,7 +913,10 @@ mod tests { &recipient, 7, &sender.public_key_bundle(), - open_options(None), + None, + SIGNATURE_PURPOSE, + ENCRYPTION_PURPOSE, + ProtectionPolicy::from(crate::SignaturePolicy::Ed25519), None, ), Err(ProtectedError::FinalRecipientMismatch) @@ -939,7 +944,10 @@ mod tests { &recipient, 7, &sender.public_key_bundle(), - open_options(None), + None, + SIGNATURE_PURPOSE, + ENCRYPTION_PURPOSE, + ProtectionPolicy::from(crate::SignaturePolicy::Ed25519), None, ), Err(ProtectedError::FinalRecipientMismatch) @@ -951,7 +959,10 @@ mod tests { &recipient, 7, &sender.public_key_bundle(), - open_options(Some(43)), + Some(43), + SIGNATURE_PURPOSE, + ENCRYPTION_PURPOSE, + ProtectionPolicy::from(crate::SignaturePolicy::Ed25519), None, ), Err(ProtectedError::ExpectedReceiverMismatch) @@ -983,7 +994,10 @@ mod tests { &recipient, 7, &sender.public_key_bundle(), - open_options(Some(42)), + Some(42), + SIGNATURE_PURPOSE, + ENCRYPTION_PURPOSE, + ProtectionPolicy::from(crate::SignaturePolicy::Ed25519), None, ) .expect("matching exposed sender"); @@ -993,7 +1007,10 @@ mod tests { &recipient, 7, &sender.public_key_bundle(), - open_options(Some(42)), + Some(42), + SIGNATURE_PURPOSE, + ENCRYPTION_PURPOSE, + ProtectionPolicy::from(crate::SignaturePolicy::Ed25519), None, ), Err(ProtectedError::SenderMismatch) @@ -1032,7 +1049,10 @@ mod tests { &recipient, 7, &sender.public_key_bundle(), - open_options(Some(42)), + Some(42), + SIGNATURE_PURPOSE, + ENCRYPTION_PURPOSE, + ProtectionPolicy::from(crate::SignaturePolicy::Ed25519), None, ), Err(ProtectedError::PayloadNotEncrypted) @@ -1046,7 +1066,10 @@ mod tests { &recipient, 7, &sender.public_key_bundle(), - open_options(Some(42)), + Some(42), + SIGNATURE_PURPOSE, + ENCRYPTION_PURPOSE, + ProtectionPolicy::from(crate::SignaturePolicy::Ed25519), None, ), Err(ProtectedError::PayloadNotSigned) @@ -1067,7 +1090,10 @@ mod tests { resolver_calls += 1; Some(vec![sender.public_key_bundle()]) }, - open_options(Some(42)), + Some(42), + SIGNATURE_PURPOSE, + ENCRYPTION_PURPOSE, + ProtectionPolicy::from(crate::SignaturePolicy::Ed25519), None, ); assert!(matches!( @@ -1112,7 +1138,10 @@ mod tests { current_sender.public_key_bundle(), old_sender.public_key_bundle(), ], - open_options(Some(42)), + Some(42), + SIGNATURE_PURPOSE, + ENCRYPTION_PURPOSE, + ProtectionPolicy::from(crate::SignaturePolicy::Ed25519), None, ) .expect("key history should open"); @@ -1138,7 +1167,10 @@ mod tests { &recipient, 7, &sender.public_key_bundle(), - open_options(Some(42)), + Some(42), + SIGNATURE_PURPOSE, + ENCRYPTION_PURPOSE, + ProtectionPolicy::from(crate::SignaturePolicy::Ed25519), None, ) .expect("arbitrary application value should open"); diff --git a/example/Cargo.lock b/example/Cargo.lock index 1c15380..1c7fa26 100644 --- a/example/Cargo.lock +++ b/example/Cargo.lock @@ -1404,6 +1404,7 @@ 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 1e6320b..65dfedd 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, - ProtectedOpenOptions, ProtectionPolicy, ProtectionPurpose, SignaturePolicy, TypeMap, + ProtectionPolicy, ProtectionPurpose, SignaturePolicy, TypeMap, forward_relay_frame, open_protected_with, open_relay_content, open_relay_metadata_with, }; use mtp::crypto::{Keyring, PublicKeyBundle}; @@ -70,12 +70,10 @@ fn process_direct_protected( std::slice::from_ref(&host_keyring), None, |signer_id| resolve_signer_key(signer_id, registered_clients).map(|key| vec![key]), - ProtectedOpenOptions::new( - Some(DIRECT_DESTINATION_ID), - ProtectionPurpose::from(DIRECT_SIGNATURE_PURPOSE), - ProtectionPurpose::from(DIRECT_ENCRYPTION_PURPOSE), - SIGNATURE_POLICY, - ), + 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 cd16934..078c2f1 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) } - async fn set_type_map(&self, type_map: &TypeMap) { - self.set_type_map(type_map).await; + fn set_type_map(&self, type_map: &TypeMap) -> impl std::future::Future + Send { + async move { 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) } - async fn set_type_map(&self, type_map: &TypeMap) { - self.set_type_map(type_map).await; + fn set_type_map(&self, type_map: &TypeMap) -> impl std::future::Future + Send { + async move { 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) } - async fn set_type_map(&self, type_map: &TypeMap) { - self.set_type_map(type_map).await; + fn set_type_map(&self, type_map: &TypeMap) -> impl std::future::Future + Send { + async move { 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) } - async fn set_type_map(&self, type_map: &TypeMap) { - self.set_type_map(type_map).await; + fn set_type_map(&self, type_map: &TypeMap) -> impl std::future::Future + Send { + async move { self.set_type_map(type_map).await } } } diff --git a/mtp-webserver/Cargo.toml b/mtp-webserver/Cargo.toml index 6b5ef4e..88f7bb1 100644 --- a/mtp-webserver/Cargo.toml +++ b/mtp-webserver/Cargo.toml @@ -25,6 +25,7 @@ rustls = "0.23" tracing = "0.1" thiserror = "2" async-trait = "0.1" +rand = { version = "0.10.1", optional = true } [dev-dependencies] rcgen = "0.14" @@ -32,5 +33,5 @@ hyper = { version = "1", features = ["client", "http2"] } [features] default = [] -crypto = ["mtp-host/crypto"] +crypto = ["mtp-host/crypto", "dep:rand"] pipes = ["mtp-host/pipes", "mtp-transport/pipes"] diff --git a/mtp-webserver/src/transport.rs b/mtp-webserver/src/transport.rs index 5c81765..3834f64 100644 --- a/mtp-webserver/src/transport.rs +++ b/mtp-webserver/src/transport.rs @@ -220,7 +220,6 @@ 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, @@ -269,7 +268,6 @@ 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 c21bd67..7f0b5d6 100644 --- a/transport/src/connection.rs +++ b/transport/src/connection.rs @@ -1078,12 +1078,18 @@ 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() @@ -1107,7 +1113,6 @@ 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() @@ -1131,11 +1136,17 @@ impl Receiver { #[cfg(feature = "pipes")] #[instrument(skip(self), level = "trace")] pub async fn receive_event(&self) -> Result { - let mut close_rx = self.inner.handle.subscribe_close(); + if self.inner.handle.is_closed() { + return Err(self + .inner + .handle + .close_reason() + .unwrap_or(CommunicationError::StreamClosed)); + } + 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)) => { @@ -1163,10 +1174,6 @@ 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 3154747..7028528 100644 --- a/transport/tests/integration.rs +++ b/transport/tests/integration.rs @@ -342,17 +342,13 @@ async fn test_max_frames_per_stream_enforced() -> Result<(), Box