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

Merge branch 'master' of ssh://git.methanium.net/methanium/mtp
All checks were successful
CI / checks (push) Successful in 5m56s
Details

Browse source
This commit is contained in:
Alex 2026-07-28 18:50:04 +02:00
commit 590810ce59
Signed by: alex
SSH key fingerprint: SHA256:D1+Ub8o0v4K5y1JNivW8IxEOelqLSvPmUzBbDIoZkRQ
10 changed files with 195 additions and 158 deletions
Download patch file Download diff file Expand all files Collapse all files

34
host/src/connection.rs
View file

@ -3,6 +3,7 @@ use mtp_codec::{CommunicationType, DataType, DataValue};
use mtp_codec::{CommunicationValue, Version, registry::VersionedCodec}; use mtp_codec::{CommunicationValue, Version, registry::VersionedCodec};
use mtp_common::CommunicationError; use mtp_common::CommunicationError;
use std::net::SocketAddr; use std::net::SocketAddr;
#[cfg(feature = "pipes")]
use std::sync::Arc; use std::sync::Arc;
#[cfg(feature = "pipes")] #[cfg(feature = "pipes")]
use tokio::sync::{Mutex, mpsc}; use tokio::sync::{Mutex, mpsc};
@ -11,6 +12,7 @@ use tokio::sync::{Mutex, mpsc};
use crate::error::random_client_id; use crate::error::random_client_id;
#[cfg(feature = "pipes")] #[cfg(feature = "pipes")]
use crate::pipe::{PipeDispatcher, PipeReceiver, PipeRequest, PipeSender, run_dispatcher}; use crate::pipe::{PipeDispatcher, PipeReceiver, PipeRequest, PipeSender, run_dispatcher};
#[cfg(feature = "pipes")]
use mtp_transport::Policy; use mtp_transport::Policy;
mod connection_capability { mod connection_capability {
@ -161,6 +163,9 @@ where
} }
/// Construct an MTP connection with an explicit policy for pipe dispatch. /// Construct an MTP connection with an explicit policy for pipe dispatch.
// The shared transport constructor keeps its argument order aligned with
// `from_transport_parts_with_remote_addr`; policy is required only here.
#[allow(clippy::too_many_arguments)]
pub fn from_transport_parts_with_policy( pub fn from_transport_parts_with_policy(
version: Version, version: Version,
codec: VersionedCodec, codec: VersionedCodec,
@ -255,35 +260,6 @@ impl<S, R, P> MTPConnection<S, R, P> {
client_public_key: None, client_public_key: None,
} }
} }
pub fn from_transport_parts_with_policy(
version: Version,
codec: VersionedCodec,
sender: S,
receiver: R,
path: String,
description: Option<String>,
remote_addr: Option<SocketAddr>,
_policy: Arc<Policy>,
) -> Self {
Self {
version,
codec,
sender,
receiver,
path,
remote_addr,
description,
_pipe_stream: std::marker::PhantomData,
_dispatcher_task: tokio::spawn(async {}),
#[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"))] #[cfg(not(feature = "pipes"))]

59
host/src/engine.rs
View file

@ -33,8 +33,7 @@ pub trait HandshakeSender: Send + Sync {
pub trait HandshakeReceiver: Send + Sync { pub trait HandshakeReceiver: Send + Sync {
fn receive( fn receive(
&self, &self,
) -> impl std::future::Future<Output = Result<CommunicationValue, CommunicationError>> ) -> impl std::future::Future<Output = Result<CommunicationValue, CommunicationError>> + Send;
+ Send;
} }
/// The result of a successful handshake, containing everything needed to /// The result of a successful handshake, containing everything needed to
@ -136,7 +135,10 @@ impl HandshakeEngine {
} }
}; };
let negotiated = match self.registry.negotiate(std::slice::from_ref(&client_version)) { let negotiated = match self
.registry
.negotiate(std::slice::from_ref(&client_version))
{
Some(v) => v, Some(v) => v,
None => { None => {
send_rejection_generic( send_rejection_generic(
@ -300,24 +302,23 @@ impl HandshakeEngine {
_ => 0, _ => 0,
}; };
if cid > 0 { if cid > 0
if let Some(bundle) = && let Some(bundle) =
(self.config.get_existing_client)(cid, description.clone()).await (self.config.get_existing_client)(cid, description.clone()).await
{ {
return self return self
.complete_auth_handshake( .complete_auth_handshake(
sender, sender,
receiver, receiver,
Flow::Login { id: cid, bundle }, Flow::Login { id: cid, bundle },
CommunicationType::IdentificationResponse, CommunicationType::IdentificationResponse,
&negotiated, &negotiated,
&codec, &codec,
description, description,
version_str, version_str,
client_version, client_version,
) )
.await; .await;
}
} }
// Unknown or zero ID: fall back to guest // Unknown or zero ID: fall back to guest
@ -634,8 +635,8 @@ impl HandshakeEngine {
DataValue::UnsignedNumber(client_nonce), DataValue::UnsignedNumber(client_nonce),
) )
.add_typed_default(DataType::Signature, DataValue::Bytes(host_sig)); .add_typed_default(DataType::Signature, DataValue::Bytes(host_sig));
response = response response =
.add_typed_default(DataType::Version, DataValue::Str(negotiated.to_string())); response.add_typed_default(DataType::Version, DataValue::Str(negotiated.to_string()));
if pq_enabled { if pq_enabled {
response = response =
response.add_typed_default(DataType::PqSignature, DataValue::Bytes(host_pq_sig)); response.add_typed_default(DataType::PqSignature, DataValue::Bytes(host_pq_sig));
@ -722,19 +723,15 @@ fn extract_register_bundle(
msg: &CommunicationValue, msg: &CommunicationValue,
) -> Result<mtp_crypto::PublicKeyBundle, AcceptError> { ) -> Result<mtp_crypto::PublicKeyBundle, AcceptError> {
match msg.get_data(DataType::PublicKeys) { match msg.get_data(DataType::PublicKeys) {
DataValue::Bytes(b) => mtp_crypto::PublicKeyBundle::from_bytes(b).map_err(|_| { DataValue::Bytes(b) => mtp_crypto::PublicKeyBundle::from_bytes(b)
AcceptError::AuthenticationFailed("invalid public key bundle".into()) .map_err(|_| AcceptError::AuthenticationFailed("invalid public key bundle".into())),
}),
_ => Err(AcceptError::AuthenticationFailed( _ => Err(AcceptError::AuthenticationFailed(
"missing public keys".into(), "missing public keys".into(),
)), )),
} }
} }
async fn send_rejection_generic<S: HandshakeSender>( async fn send_rejection_generic<S: HandshakeSender>(sender: &S, reason: RejectionReason) {
sender: &S,
reason: RejectionReason,
) {
let response = match &reason { let response = match &reason {
RejectionReason::BadVersion { supported_versions } => { RejectionReason::BadVersion { supported_versions } => {
CommunicationValue::new(CommunicationType::ErrorBadVersion) CommunicationValue::new(CommunicationType::ErrorBadVersion)
@ -815,7 +812,9 @@ impl<C: mtp_transport::TransportConnection> HandshakeSender for mtp_transport::G
} }
} }
impl<C: mtp_transport::TransportConnection> HandshakeReceiver for mtp_transport::GenericReceiver<C> { impl<C: mtp_transport::TransportConnection> HandshakeReceiver
for mtp_transport::GenericReceiver<C>
{
fn receive( fn receive(
&self, &self,
) -> impl std::future::Future<Output = Result<CommunicationValue, CommunicationError>> + Send ) -> impl std::future::Future<Output = Result<CommunicationValue, CommunicationError>> + Send

11
host/src/handshake.rs
View file

@ -1,7 +1,10 @@
#[cfg(not(feature = "crypto"))]
use mtp_codec::{Version, registry::{Registry, VersionedCodec}};
#[cfg(feature = "crypto")] #[cfg(feature = "crypto")]
use mtp_codec::registry::Registry; use mtp_codec::registry::Registry;
#[cfg(not(feature = "crypto"))]
use mtp_codec::{
Version,
registry::{Registry, VersionedCodec},
};
use mtp_transport::{Receiver, Sender}; use mtp_transport::{Receiver, Sender};
use std::sync::Arc; use std::sync::Arc;
use std::time::Instant; use std::time::Instant;
@ -136,9 +139,7 @@ impl HandshakeContext {
#[cfg(feature = "crypto")] #[cfg(feature = "crypto")]
{ {
Ok(Some(self.connection_from_handshake_result( Ok(Some(self.connection_from_handshake_result(
sender, sender, receiver, result,
receiver,
result,
))) )))
} }
#[cfg(not(feature = "crypto"))] #[cfg(not(feature = "crypto"))]

2
host/src/lib.rs
View file

@ -11,7 +11,7 @@ pub use MTPHost as Host;
pub use config::HostConfig; pub use config::HostConfig;
pub use config::Policy; pub use config::Policy;
pub use connection::{MTPConnection, MtpReceiverLike, MtpSenderLike}; pub use connection::{MTPConnection, MtpReceiverLike, MtpSenderLike};
pub use engine::{HandshakeEngine, HandshakeResult, HandshakeReceiver, HandshakeSender}; pub use engine::{HandshakeEngine, HandshakeReceiver, HandshakeResult, HandshakeSender};
pub use error::AcceptError; pub use error::AcceptError;
pub use handshake::MTPHost; pub use handshake::MTPHost;
pub use mtp_transport::Receiver; pub use mtp_transport::Receiver;

8
mtp-webserver/src/server.rs
View file

@ -352,9 +352,11 @@ fn build_endpoint(
transport.keep_alive_interval(Some(keep_alive)); transport.keep_alive_interval(Some(keep_alive));
} }
if let Some(idle_timeout) = config.policy.max_idle_timeout { if let Some(idle_timeout) = config.policy.max_idle_timeout {
transport.max_idle_timeout(Some(idle_timeout.try_into().map_err( transport.max_idle_timeout(Some(
|error| CommunicationError::Other(format!("{error}")), idle_timeout
)?)); .try_into()
.map_err(|error| CommunicationError::Other(format!("{error}")))?,
));
} }
Arc::new(transport) Arc::new(transport)
}); });

143
mtp-webserver/src/transport.rs
View file

@ -267,17 +267,13 @@ pub(crate) async fn accept_web_connection(
send_pongs: bool, send_pongs: bool,
policy: Policy, policy: Policy,
host_config: Arc<HostConfig>, host_config: Arc<HostConfig>,
auth_semaphore: Arc<tokio::sync::Semaphore>, _auth_semaphore: Arc<tokio::sync::Semaphore>,
) -> Result<WebMTPConnection, AcceptError> { ) -> Result<WebMTPConnection, AcceptError> {
#[cfg(feature = "crypto")] #[cfg(feature = "crypto")]
{ {
let permit = auth_semaphore let permit = _auth_semaphore.clone().acquire_owned().await.map_err(|_| {
.clone() AcceptError::AuthenticationFailed("authentication service stopped".into())
.acquire_owned() })?;
.await
.map_err(|_| {
AcceptError::AuthenticationFailed("authentication service stopped".into())
})?;
let result = tokio::time::timeout( let result = tokio::time::timeout(
host_config.auth_timeout, host_config.auth_timeout,
accept_web_connection_inner(session, path, quinn, send_pongs, policy, host_config), accept_web_connection_inner(session, path, quinn, send_pongs, policy, host_config),
@ -327,8 +323,20 @@ async fn accept_web_connection_inner(
if send_pongs { if send_pongs {
receiver.respond_to_pings(sender.clone()).await; receiver.respond_to_pings(sender.clone()).await;
} }
#[cfg(feature = "pipes")]
let connection: WebMTPConnection = mtp_host::MTPConnection::from_transport_parts_with_policy(
negotiated,
codec,
sender,
receiver,
path,
description.clone(),
Some(remote_addr),
policy,
);
#[cfg(not(feature = "pipes"))]
let connection: WebMTPConnection = let connection: WebMTPConnection =
mtp_host::MTPConnection::from_transport_parts_with_policy( mtp_host::MTPConnection::from_transport_parts_with_remote_addr(
negotiated, negotiated,
codec, codec,
sender, sender,
@ -336,12 +344,11 @@ async fn accept_web_connection_inner(
path, path,
description.clone(), description.clone(),
Some(remote_addr), Some(remote_addr),
policy,
); );
#[cfg(not(feature = "crypto"))] #[cfg(not(feature = "crypto"))]
{ {
connection.receiver.set_max_message_size(max_message_size); connection.receiver.set_max_message_size(max_message_size);
return Ok(connection); Ok(connection)
} }
#[cfg(feature = "crypto")] #[cfg(feature = "crypto")]
let mut connection = connection; let mut connection = connection;
@ -364,13 +371,14 @@ async fn accept_web_connection_inner(
// Unauthenticated: send accepted response with guest ID (or ID 0) // Unauthenticated: send accepted response with guest ID (or ID 0)
if !is_allow_auth && !is_force_auth { if !is_allow_auth && !is_force_auth {
let response = mtp_codec::CommunicationValue::new(CommunicationType::IdentificationResponse) let response =
.add_typed_default(DataType::Connected, DataValue::BoolTrue) mtp_codec::CommunicationValue::new(CommunicationType::IdentificationResponse)
.add_typed_default( .add_typed_default(DataType::Connected, DataValue::BoolTrue)
DataType::Version, .add_typed_default(
DataValue::Str(connection.version.to_string()), DataType::Version,
) DataValue::Str(connection.version.to_string()),
.add_typed_default(DataType::Id, DataValue::UnsignedNumber(0)); )
.add_typed_default(DataType::Id, DataValue::UnsignedNumber(0));
connection connection
.sender .sender
.send(&response) .send(&response)
@ -405,16 +413,31 @@ async fn accept_web_connection_inner(
if let Some(bundle) = if let Some(bundle) =
(_host_config.get_existing_client)(id, description.clone()).await (_host_config.get_existing_client)(id, description.clone()).await
{ {
(id, Some(bundle), CommunicationType::IdentificationResponse, false) (
id,
Some(bundle),
CommunicationType::IdentificationResponse,
false,
)
} else { } else {
// Unknown client: fall back to guest // Unknown client: fall back to guest
let guest_id = H3TransportConnection::assign_guest_id(&_host_config).await?; let guest_id = H3TransportConnection::assign_guest_id(&_host_config).await?;
(guest_id, None, CommunicationType::IdentificationResponse, true) (
guest_id,
None,
CommunicationType::IdentificationResponse,
true,
)
} }
} else { } else {
// ID zero or missing: fall back to guest // ID zero or missing: fall back to guest
let guest_id = H3TransportConnection::assign_guest_id(&_host_config).await?; let guest_id = H3TransportConnection::assign_guest_id(&_host_config).await?;
(guest_id, None, CommunicationType::IdentificationResponse, true) (
guest_id,
None,
CommunicationType::IdentificationResponse,
true,
)
} }
} else if first_type_opt == reg_type { } else if first_type_opt == reg_type {
// Registration: always authenticate (both AllowAuth and ForceAuth) // Registration: always authenticate (both AllowAuth and ForceAuth)
@ -442,7 +465,12 @@ async fn accept_web_connection_inner(
let bundle = (_host_config.get_existing_client)(id, description.clone()) let bundle = (_host_config.get_existing_client)(id, description.clone())
.await .await
.ok_or_else(|| AcceptError::AuthenticationFailed("unknown client id".into()))?; .ok_or_else(|| AcceptError::AuthenticationFailed("unknown client id".into()))?;
(id, Some(bundle), CommunicationType::IdentificationResponse, false) (
id,
Some(bundle),
CommunicationType::IdentificationResponse,
false,
)
} else { } else {
return Err(AcceptError::AuthenticationFailed( return Err(AcceptError::AuthenticationFailed(
"unexpected authentication message".into(), "unexpected authentication message".into(),
@ -453,10 +481,14 @@ async fn accept_web_connection_inner(
// Guest path: skip challenge/response, send accepted with guest ID // Guest path: skip challenge/response, send accepted with guest ID
if is_guest { if is_guest {
let guest_id = client_id; let guest_id = client_id;
let response = mtp_codec::CommunicationValue::new(CommunicationType::IdentificationResponse) let response =
.add_typed_default(DataType::Connected, DataValue::BoolTrue) mtp_codec::CommunicationValue::new(CommunicationType::IdentificationResponse)
.add_typed_default(DataType::Version, DataValue::Str(connection.version.to_string())) .add_typed_default(DataType::Connected, DataValue::BoolTrue)
.add_typed_default(DataType::Id, DataValue::UnsignedNumber(guest_id as u128)); .add_typed_default(
DataType::Version,
DataValue::Str(connection.version.to_string()),
)
.add_typed_default(DataType::Id, DataValue::UnsignedNumber(guest_id as u128));
connection connection
.sender .sender
.send(&response) .send(&response)
@ -513,14 +545,11 @@ async fn accept_web_connection_inner(
let host_config = _host_config.clone(); let host_config = _host_config.clone();
let pq_signer = host_pq_signer.clone(); let pq_signer = host_pq_signer.clone();
async move { async move {
let signer = let signer = Ed25519Signer::new(&host_config.host_keyring.sig_cl_secret_key)
Ed25519Signer::new(&host_config.host_keyring.sig_cl_secret_key) .map_err(|e| AcceptError::AuthenticationFailed(e.to_string()))?;
.map_err(|e| AcceptError::AuthenticationFailed(e.to_string()))?;
if let Some(pq_signer) = pq_signer { if let Some(pq_signer) = pq_signer {
mtp_crypto::sign_parallel::sign_dual_parallel_shared_pq( mtp_crypto::sign_parallel::sign_dual_parallel_shared_pq(
signer, signer, pq_signer, payload,
pq_signer,
payload,
) )
.await .await
.map_err(|e| AcceptError::AuthenticationFailed(e.to_string())) .map_err(|e| AcceptError::AuthenticationFailed(e.to_string()))
@ -535,21 +564,20 @@ async fn accept_web_connection_inner(
let sign_challenge_started = Instant::now(); let sign_challenge_started = Instant::now();
let (sig, pq_sig) = host_sign(auth::challenge_payload(client_id, server_challenge)).await?; let (sig, pq_sig) = host_sign(auth::challenge_payload(client_id, server_challenge)).await?;
tracing::debug!(elapsed = ?sign_challenge_started.elapsed(), "web authentication handshake: sign challenge"); tracing::debug!(elapsed = ?sign_challenge_started.elapsed(), "web authentication handshake: sign challenge");
let mut challenge = let mut challenge = mtp_codec::CommunicationValue::new(CommunicationType::Challenge)
mtp_codec::CommunicationValue::new(CommunicationType::Challenge) .add_typed_default(
.add_typed_default( DataType::ServerNonce,
DataType::ServerNonce, DataValue::UnsignedNumber(server_challenge),
DataValue::UnsignedNumber(server_challenge), )
) .add_typed_default(DataType::Signature, DataValue::Bytes(sig))
.add_typed_default(DataType::Signature, DataValue::Bytes(sig)) .add_typed_default(
.add_typed_default( DataType::RequirePq,
DataType::RequirePq, if _host_config.require_pq {
if _host_config.require_pq { DataValue::BoolTrue
DataValue::BoolTrue } else {
} else { DataValue::BoolFalse
DataValue::BoolFalse },
}, );
);
if pq_enabled { if pq_enabled {
challenge = challenge =
challenge.add_typed_default(DataType::PqSignature, DataValue::Bytes(pq_sig)); challenge.add_typed_default(DataType::PqSignature, DataValue::Bytes(pq_sig));
@ -579,8 +607,7 @@ async fn accept_web_connection_inner(
} }
}; };
tracing::debug!(elapsed = ?receive_proof_started.elapsed(), "web authentication handshake: receive client proof"); tracing::debug!(elapsed = ?receive_proof_started.elapsed(), "web authentication handshake: receive client proof");
if Some(proof.get_type()) != CommunicationType::ChallengeResponse.try_to_id(&tm) if Some(proof.get_type()) != CommunicationType::ChallengeResponse.try_to_id(&tm) {
{
return Err(AcceptError::AuthenticationFailed( return Err(AcceptError::AuthenticationFailed(
"missing challenge response".into(), "missing challenge response".into(),
)); ));
@ -605,11 +632,7 @@ async fn accept_web_connection_inner(
DataValue::Bytes(bytes) => bytes.as_slice(), DataValue::Bytes(bytes) => bytes.as_slice(),
_ => &[], _ => &[],
}; };
let payload = if first.get_type() let payload = if first.get_type() == CommunicationType::Register.try_to_id(&tm).unwrap() {
== CommunicationType::Register
.try_to_id(&tm)
.unwrap()
{
auth::register_proof_payload( auth::register_proof_payload(
&version.to_string(), &version.to_string(),
&client_bundle.as_bytes(), &client_bundle.as_bytes(),
@ -643,9 +666,13 @@ async fn accept_web_connection_inner(
tracing::debug!(elapsed = ?verify_proof_started.elapsed(), "web authentication handshake: verify client proof"); tracing::debug!(elapsed = ?verify_proof_started.elapsed(), "web authentication handshake: verify client proof");
if !proof_ok { if !proof_ok {
let rejection = mtp_codec::CommunicationValue::new(CommunicationType::IdentificationResponse) let rejection =
.add_typed_default(DataType::Connected, DataValue::BoolFalse) mtp_codec::CommunicationValue::new(CommunicationType::IdentificationResponse)
.add_typed_default(DataType::ErrorMessage, DataValue::Str("client proof signature invalid".into())); .add_typed_default(DataType::Connected, DataValue::BoolFalse)
.add_typed_default(
DataType::ErrorMessage,
DataValue::Str("client proof signature invalid".into()),
);
let _ = connection.sender.send(&rejection).await; let _ = connection.sender.send(&rejection).await;
connection.sender.close(); connection.sender.close();
return Err(AcceptError::AuthenticationFailed( return Err(AcceptError::AuthenticationFailed(
@ -698,5 +725,3 @@ async fn accept_web_connection_inner(
Ok(connection) Ok(connection)
} }
} }

4
src/sdk/index.ts
View file

@ -857,6 +857,10 @@ export class MTPClient {
return this.raw.client.state; return this.raw.client.state;
} }
get pingMs(): number | null {
return this.raw.client.ping_ms ?? null;
}
async #loadStoredCredentials() { async #loadStoredCredentials() {
if (this.#credentials || !this.#options.storage) { if (this.#credentials || !this.#options.storage) {
return; return;

48
transport/src/generic_connection.rs
View file

@ -198,7 +198,9 @@ impl<C: TransportConnection> GenericReceiver<C> {
let (pipe_tx, pipe_rx) = mpsc::channel(policy.receiver_queue_capacity); let (pipe_tx, pipe_rx) = mpsc::channel(policy.receiver_queue_capacity);
let ping_sender: Arc<RwLock<Option<GenericSender<C>>>> = Arc::new(RwLock::new(None)); let ping_sender: Arc<RwLock<Option<GenericSender<C>>>> = Arc::new(RwLock::new(None));
let max_message_size = Arc::new(AtomicU64::new( let max_message_size = Arc::new(AtomicU64::new(
policy.handshake_max_message_size.min(policy.max_message_size), policy
.handshake_max_message_size
.min(policy.max_message_size),
)); ));
let task_ping_sender = ping_sender.clone(); let task_ping_sender = ping_sender.clone();
let task_connection = connection.clone(); let task_connection = connection.clone();
@ -214,8 +216,8 @@ impl<C: TransportConnection> GenericReceiver<C> {
loop { loop {
// Backpressure: stop accepting new streams if the output queue is full. // Backpressure: stop accepting new streams if the output queue is full.
#[cfg(feature = "pipes")] #[cfg(feature = "pipes")]
let cap_full = task_accept_task_tx.capacity() == 0 let cap_full =
|| task_accept_task_pipe_tx.capacity() == 0; task_accept_task_tx.capacity() == 0 || task_accept_task_pipe_tx.capacity() == 0;
#[cfg(not(feature = "pipes"))] #[cfg(not(feature = "pipes"))]
let cap_full = task_accept_task_tx.capacity() == 0; let cap_full = task_accept_task_tx.capacity() == 0;
@ -269,10 +271,7 @@ impl<C: TransportConnection> GenericReceiver<C> {
{ {
let close_error = CommunicationError::StreamError; let close_error = CommunicationError::StreamError;
let _ = tx.send(Err(close_error.clone())).await; let _ = tx.send(Err(close_error.clone())).await;
connection.close( connection.close(policy.application_close_code, b"max frames exceeded");
policy.application_close_code,
b"max frames exceeded",
);
break; break;
} }
let mut len = [0; 4]; let mut len = [0; 4];
@ -291,12 +290,12 @@ impl<C: TransportConnection> GenericReceiver<C> {
break; break;
} }
Err(_) => { Err(_) => {
tracing::warn!("MTP receive stream timed out while reading frame header"); tracing::warn!(
let _ = tx.send(Err(CommunicationError::StreamError)).await; "MTP receive stream timed out while reading frame header"
connection.close(
policy.application_close_code,
b"frame header timeout",
); );
let _ = tx.send(Err(CommunicationError::StreamError)).await;
connection
.close(policy.application_close_code, b"frame header timeout");
break; break;
} }
} }
@ -308,10 +307,7 @@ impl<C: TransportConnection> GenericReceiver<C> {
if len as u64 > frame_limit { if len as u64 > frame_limit {
tracing::warn!(len, "MTP receive stream frame is too large"); tracing::warn!(len, "MTP receive stream frame is too large");
let _ = tx.send(Err(CommunicationError::MessageTooLarge)).await; let _ = tx.send(Err(CommunicationError::MessageTooLarge)).await;
connection.close( connection.close(policy.application_close_code, b"frame too large");
policy.application_close_code,
b"frame too large",
);
break; break;
} }
let target_len = len as usize; let target_len = len as usize;
@ -322,10 +318,8 @@ impl<C: TransportConnection> GenericReceiver<C> {
"MTP receive stream could not reserve frame body" "MTP receive stream could not reserve frame body"
); );
let _ = tx.send(Err(CommunicationError::MessageTooLarge)).await; let _ = tx.send(Err(CommunicationError::MessageTooLarge)).await;
connection.close( connection
policy.application_close_code, .close(policy.application_close_code, b"frame allocation failed");
b"frame allocation failed",
);
break; break;
} }
while body.len() < target_len { while body.len() < target_len {
@ -345,10 +339,8 @@ impl<C: TransportConnection> GenericReceiver<C> {
"MTP receive stream failed while reading frame body" "MTP receive stream failed while reading frame body"
); );
let _ = tx.send(Err(CommunicationError::StreamError)).await; let _ = tx.send(Err(CommunicationError::StreamError)).await;
connection.close( connection
policy.application_close_code, .close(policy.application_close_code, b"frame body read error");
b"frame body read error",
);
break; break;
} }
body.extend_from_slice(&chunk[..chunk_len]); body.extend_from_slice(&chunk[..chunk_len]);
@ -361,12 +353,10 @@ impl<C: TransportConnection> GenericReceiver<C> {
Ok(message) => message, Ok(message) => message,
Err(_) => { Err(_) => {
tracing::warn!("MTP receive stream contained an invalid frame"); tracing::warn!("MTP receive stream contained an invalid frame");
let _ = tx.send(Err(CommunicationError::ParseCommunicationValue)) let _ = tx
.send(Err(CommunicationError::ParseCommunicationValue))
.await; .await;
connection.close( connection.close(policy.application_close_code, b"invalid frame");
policy.application_close_code,
b"invalid frame",
);
break; break;
} }
}; };

43
wasm/src/client.rs
View file

@ -44,9 +44,19 @@ fn route_incoming_frame(
on_message: &js_sys::Function, on_message: &js_sys::Function,
subscriptions: &Rc<RefCell<HashMap<u32, (String, js_sys::Function)>>>, subscriptions: &Rc<RefCell<HashMap<u32, (String, js_sys::Function)>>>,
pending_requests: &Rc<RefCell<HashMap<u32, PendingRequest>>>, pending_requests: &Rc<RefCell<HashMap<u32, PendingRequest>>>,
pending_pings: &Rc<RefCell<HashMap<u32, f64>>>,
ping_ms: &Rc<Cell<Option<f64>>>,
) { ) {
let message_type = frame_type(frame); let message_type = frame_type(frame);
if message_type.as_deref() == Some("Pong") {
if let Some(sent_at) = frame_id(frame).and_then(|id| pending_pings.borrow_mut().remove(&id))
{
ping_ms.set(Some(js_sys::Date::now() - sent_at));
}
return;
}
if let Some(request_id) = frame_id(frame) { if let Some(request_id) = frame_id(frame) {
let pending = pending_requests.borrow_mut().remove(&request_id); let pending = pending_requests.borrow_mut().remove(&request_id);
if let Some(pending) = pending { if let Some(pending) = pending {
@ -149,6 +159,8 @@ pub struct WasmClient {
next_subscription_id: Rc<Cell<u32>>, next_subscription_id: Rc<Cell<u32>>,
pending_requests: Rc<RefCell<HashMap<u32, PendingRequest>>>, pending_requests: Rc<RefCell<HashMap<u32, PendingRequest>>>,
ping_timer: Rc<RefCell<Option<PingTimer>>>, ping_timer: Rc<RefCell<Option<PingTimer>>>,
pending_pings: Rc<RefCell<HashMap<u32, f64>>>,
ping_ms: Rc<Cell<Option<f64>>>,
pending_pipe_creations: Rc<RefCell<HashMap<u32, oneshot::Sender<Result<bool, JsValue>>>>>, pending_pipe_creations: Rc<RefCell<HashMap<u32, oneshot::Sender<Result<bool, JsValue>>>>>,
pending_pipes: Rc<RefCell<HashMap<u32, oneshot::Sender<PipeReader>>>>, pending_pipes: Rc<RefCell<HashMap<u32, oneshot::Sender<PipeReader>>>>,
on_pipe_request: Rc<RefCell<Option<js_sys::Function>>>, on_pipe_request: Rc<RefCell<Option<js_sys::Function>>>,
@ -184,6 +196,8 @@ impl WasmClient {
next_subscription_id: Rc::new(Cell::new(1)), next_subscription_id: Rc::new(Cell::new(1)),
pending_requests: Rc::new(RefCell::new(HashMap::new())), pending_requests: Rc::new(RefCell::new(HashMap::new())),
ping_timer: Rc::new(RefCell::new(None)), ping_timer: Rc::new(RefCell::new(None)),
pending_pings: Rc::new(RefCell::new(HashMap::new())),
ping_ms: Rc::new(Cell::new(None)),
pending_pipe_creations: Rc::new(RefCell::new(HashMap::new())), pending_pipe_creations: Rc::new(RefCell::new(HashMap::new())),
pending_pipes: Rc::new(RefCell::new(HashMap::new())), pending_pipes: Rc::new(RefCell::new(HashMap::new())),
on_pipe_request: Rc::new(RefCell::new(None)), on_pipe_request: Rc::new(RefCell::new(None)),
@ -200,6 +214,11 @@ impl WasmClient {
self.state.get() as u8 self.state.get() as u8
} }
#[wasm_bindgen(getter)]
pub fn ping_ms(&self) -> Option<f64> {
self.ping_ms.get()
}
#[wasm_bindgen] #[wasm_bindgen]
pub async fn connect(&self, config: &ConnectionConfig) -> Result<(), JsValue> { pub async fn connect(&self, config: &ConnectionConfig) -> Result<(), JsValue> {
self.set_state(ConnectionState::Connecting); self.set_state(ConnectionState::Connecting);
@ -582,11 +601,17 @@ impl WasmClient {
}; };
let interval_ms = interval_ms.max(1_000) as i32; let interval_ms = interval_ms.max(1_000) as i32;
let on_error = self.on_error.clone(); let on_error = self.on_error.clone();
let pending_pings = self.pending_pings.clone();
let closure = Closure::wrap(Box::new(move || { let closure = Closure::wrap(Box::new(move || {
let transport = transport.clone(); let transport = transport.clone();
let on_error = on_error.clone(); let on_error = on_error.clone();
let pending_pings = pending_pings.clone();
wasm_bindgen_futures::spawn_local(async move { wasm_bindgen_futures::spawn_local(async move {
let timestamp = js_sys::Date::now() as u64; let sent_at = js_sys::Date::now();
pending_pings
.borrow_mut()
.retain(|_, pending_at| sent_at - *pending_at < interval_ms as f64 * 3.0);
let timestamp = sent_at as u64;
let frame = CommunicationValue::new(CommunicationType::Ping) let frame = CommunicationValue::new(CommunicationType::Ping)
.add_typed_default( .add_typed_default(
DataType::Description, DataType::Description,
@ -596,12 +621,16 @@ impl WasmClient {
DataType::Timestamp, DataType::Timestamp,
DataValue::UnsignedNumber(timestamp as u128), DataValue::UnsignedNumber(timestamp as u128),
) )
.with_sender(client_id) .with_sender(client_id);
let ping_id = frame.get_id();
let frame = frame
.to_bytes() .to_bytes()
.map_err(|e| js_error(format!("encode ping failed: {}", e))); .map_err(|e| js_error(format!("encode ping failed: {}", e)));
match frame { match frame {
Ok(frame) => { Ok(frame) => {
pending_pings.borrow_mut().insert(ping_id, sent_at);
if let Err(error) = transport.send_frame(&frame).await { if let Err(error) = transport.send_frame(&frame).await {
pending_pings.borrow_mut().remove(&ping_id);
let _ = on_error.call1(&JsValue::NULL, &error); let _ = on_error.call1(&JsValue::NULL, &error);
} }
} }
@ -629,6 +658,8 @@ impl WasmClient {
#[wasm_bindgen] #[wasm_bindgen]
pub fn stop_protocol_pings(&self) { pub fn stop_protocol_pings(&self) {
self.pending_pings.borrow_mut().clear();
self.ping_ms.set(None);
let Some(timer) = self.ping_timer.borrow_mut().take() else { let Some(timer) = self.ping_timer.borrow_mut().take() else {
return; return;
}; };
@ -741,6 +772,10 @@ impl WasmClient {
let pending_requests = self.pending_requests.clone(); let pending_requests = self.pending_requests.clone();
let loop_pending_requests = pending_requests.clone(); let loop_pending_requests = pending_requests.clone();
let ping_timer = self.ping_timer.clone(); let ping_timer = self.ping_timer.clone();
let pending_pings = self.pending_pings.clone();
let loop_pending_pings = pending_pings.clone();
let ping_ms = self.ping_ms.clone();
let loop_ping_ms = ping_ms.clone();
let pending_pipe_creations = self.pending_pipe_creations.clone(); let pending_pipe_creations = self.pending_pipe_creations.clone();
let pending_pipes = self.pending_pipes.clone(); let pending_pipes = self.pending_pipes.clone();
let on_pipe_request = self.on_pipe_request.clone(); let on_pipe_request = self.on_pipe_request.clone();
@ -808,6 +843,8 @@ impl WasmClient {
&on_msg, &on_msg,
&subscriptions, &subscriptions,
&loop_pending_requests, &loop_pending_requests,
&loop_pending_pings,
&loop_ping_ms,
); );
}, },
on_err.clone(), on_err.clone(),
@ -822,6 +859,8 @@ impl WasmClient {
.await; .await;
state.set(ConnectionState::Disconnected); state.set(ConnectionState::Disconnected);
stop_ping_timer(&ping_timer); stop_ping_timer(&ping_timer);
pending_pings.borrow_mut().clear();
ping_ms.set(None);
reject_pending_requests(&pending_requests, "disconnected"); reject_pending_requests(&pending_requests, "disconnected");
client_pipe::reject_pending_pipe_creations(&pending_pipe_creations, "disconnected"); client_pipe::reject_pending_pipe_creations(&pending_pipe_creations, "disconnected");
}); });

1
wasm/types/mtp_wasm.d.ts vendored
View file

@ -121,6 +121,7 @@ export class WasmClient implements DisposableWasmObject {
subscribe(message_type: string, callback: MessageCallback): number; subscribe(message_type: string, callback: MessageCallback): number;
unsubscribe(id: number): boolean; unsubscribe(id: number): boolean;
static is_supported(): boolean; static is_supported(): boolean;
readonly ping_ms: number | undefined;
readonly state: ConnectionState; readonly state: ConnectionState;
} }