diff --git a/.envrc b/.envrc new file mode 100644 index 0000000..3550a30 --- /dev/null +++ b/.envrc @@ -0,0 +1 @@ +use flake diff --git a/.forgejo/workflows/ci.yml b/.forgejo/workflows/ci.yml index 70e6c39..970eece 100644 --- a/.forgejo/workflows/ci.yml +++ b/.forgejo/workflows/ci.yml @@ -7,16 +7,12 @@ on: env: CARGO_TERM_COLOR: always - NIX_CONFIG: experimental-features = nix-command flakes jobs: checks: name: checks runs-on: nixos steps: - - name: Install node - run: nix profile add nixpkgs#nodejs_24 - - name: Checkout uses: https://data.forgejo.org/actions/checkout@v7 @@ -33,8 +29,6 @@ jobs: cargo machete pnpm install --frozen-lockfile - pnpm add --save-dev --save-exact --workspace-root jscpd-linux-x64-gnu@5.0.14 - pnpm run dup RUSTFLAGS="--cfg web_sys_unstable_apis" wasm-pack test --node wasm RUSTFLAGS="--cfg web_sys_unstable_apis" wasm-pack build wasm --target web diff --git a/.forgejo/workflows/release.yml b/.forgejo/workflows/release.yml index 05bc6e2..5996da0 100644 --- a/.forgejo/workflows/release.yml +++ b/.forgejo/workflows/release.yml @@ -14,16 +14,10 @@ on: required: true type: string -env: - NIX_CONFIG: experimental-features = nix-command flakes - jobs: release: runs-on: nixos steps: - - name: Install node & bun - run: nix profile add nixpkgs#nodejs_24 nixpkgs#bun - - name: Check out repo uses: https://data.forgejo.org/actions/checkout@v7 with: @@ -32,9 +26,6 @@ jobs: - name: Install dependencies run: bun install - - name: Install cc linker, sed & jq - run: nix profile add nixpkgs#stdenv.cc nixpkgs#gnused nixpkgs#jq - - name: Build all run: bun build:all diff --git a/.gitignore b/.gitignore index 6dff05e..f600624 100644 --- a/.gitignore +++ b/.gitignore @@ -6,3 +6,4 @@ dist/ *.tgz wasm/pkg/ web_client/ +.direnv diff --git a/Cargo.lock b/Cargo.lock index 88e9b74..8e8f6ba 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -256,9 +256,9 @@ dependencies = [ [[package]] name = "chacha20" -version = "0.10.1" +version = "0.10.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81" +checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06" dependencies = [ "cfg-if", "cpufeatures 0.3.0", @@ -1326,9 +1326,6 @@ dependencies = [ "mtp-transport", "mtp-type-map", "mtp-webserver", - "rand", - "rcgen", - "tokio", ] [[package]] @@ -1363,7 +1360,6 @@ name = "mtp-common" version = "0.3.0" dependencies = [ "quinn", - "rustls", "thiserror 2.0.20", "wtransport", ] @@ -1804,7 +1800,7 @@ version = "0.10.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80" dependencies = [ - "chacha20 0.10.1", + "chacha20 0.10.2", "getrandom 0.4.3", "rand_core 0.10.1", ] diff --git a/Cargo.toml b/Cargo.toml index ee1c93a..2c1b1ec 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -113,10 +113,5 @@ tls = ["crypto", "mtp-crypto?/tls"] # Requires MTP_INSECURE_TLS=1 at runtime. insecure-tls = ["dep:mtp-transport", "mtp-transport?/insecure-tls"] -[dev-dependencies] -tokio = { version = "1", features = ["full"] } -rcgen = "0.14" -rand = "0.10.1" - [package.metadata.cargo-machete] ignored = ["mtp-transport"] diff --git a/client/src/pipe.rs b/client/src/pipe.rs index 8e839dd..136fe01 100644 --- a/client/src/pipe.rs +++ b/client/src/pipe.rs @@ -78,9 +78,41 @@ pub struct PipeRequest { pub(crate) pipe_id: u32, pub(crate) description: String, pub(crate) sender: Sender, + pub(crate) receiver: Receiver, pub(crate) dispatcher: Arc, } +#[cfg(feature = "pipes")] +struct ExpectedPipeGuard { + receiver: Receiver, + pipe_id: u32, + armed: bool, +} + +#[cfg(feature = "pipes")] +impl ExpectedPipeGuard { + fn new(receiver: Receiver, pipe_id: u32) -> Self { + Self { + receiver, + pipe_id, + armed: true, + } + } + + fn disarm(&mut self) { + self.armed = false; + } +} + +#[cfg(feature = "pipes")] +impl Drop for ExpectedPipeGuard { + fn drop(&mut self) { + if self.armed { + self.receiver.cancel_expected_pipe(self.pipe_id); + } + } +} + #[cfg(feature = "pipes")] impl PipeRequest { pub fn id(&self) -> u32 { @@ -92,6 +124,10 @@ impl PipeRequest { } pub async fn accept(self) -> Result { + self.receiver + .expect_pipe(self.pipe_id) + .map_err(PipeError::from)?; + let mut expected_pipe = ExpectedPipeGuard::new(self.receiver.clone(), self.pipe_id); let (pipe_tx, pipe_rx) = tokio::sync::oneshot::channel(); { let mut pending = self.dispatcher.pending_pipes.lock().await; @@ -115,7 +151,10 @@ impl PipeRequest { let timeout = self.dispatcher.policy.read_timeout; match tokio::time::timeout(timeout, pipe_rx).await { - Ok(Ok(reader)) => Ok(reader), + Ok(Ok(reader)) => { + expected_pipe.disarm(); + Ok(reader) + } Ok(Err(_)) => { self.dispatcher .pending_pipes @@ -413,6 +452,7 @@ pub(crate) async fn run_dispatcher( pipe_id, description, sender: sender.clone(), + receiver: receiver.clone(), dispatcher: dispatcher.clone(), }; let _ = pipe_req_tx.send(req).await; diff --git a/codec/Cargo.lock b/codec/Cargo.lock index 5ed6f53..a898cf4 100644 --- a/codec/Cargo.lock +++ b/codec/Cargo.lock @@ -187,9 +187,9 @@ dependencies = [ [[package]] name = "chacha20" -version = "0.10.1" +version = "0.10.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81" +checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06" dependencies = [ "cfg-if", "cpufeatures 0.3.0", @@ -1151,7 +1151,7 @@ version = "0.10.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80" dependencies = [ - "chacha20 0.10.1", + "chacha20 0.10.2", "getrandom 0.4.3", "rand_core 0.10.1", ] diff --git a/common/Cargo.lock b/common/Cargo.lock index 338d35d..8d0f18c 100644 --- a/common/Cargo.lock +++ b/common/Cargo.lock @@ -139,9 +139,9 @@ checksum = "f079e83a288787bcd14a6aea84cee5c87a67c5a3e660c30f557a3d24761b3527" [[package]] name = "chacha20" -version = "0.10.1" +version = "0.10.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81" +checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06" dependencies = [ "cfg-if", "cpufeatures", diff --git a/common/Cargo.toml b/common/Cargo.toml index 24d4777..5fae68e 100644 --- a/common/Cargo.toml +++ b/common/Cargo.toml @@ -15,7 +15,6 @@ wtransport = { version = "0.7.1", default-features = false, features = [ "quinn", "self-signed", ] } -rustls = { version = "0.23.41" } quinn = { version = "0.11.11", default-features = false, features = [ "rustls-aws-lc-rs", "rustls", diff --git a/common/src/lib.rs b/common/src/lib.rs index 1a7fe66..71e87f2 100644 --- a/common/src/lib.rs +++ b/common/src/lib.rs @@ -164,6 +164,9 @@ pub enum CommunicationError { #[error("Stream Error")] StreamError, + #[error("Stream failed after delivery may have started")] + DeliveryUnknown, + #[error("Stream Error: {0}")] #[cfg(not(target_arch = "wasm32"))] StreamWriteError(#[from] wtransport::error::StreamWriteError), @@ -182,6 +185,38 @@ pub enum CommunicationError { Other(String), } +/// How the protocol layer should handle the first frame on a receive stream. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum FirstFrameDisposition { + Message, + Pipe(u32), +} + +/// Classify a first frame without tying the decision to a WebTransport backend. +/// +/// `PipeRequest` is used both as a control message and as the header of the raw +/// stream opened after that request is accepted. Only the protocol layer knows +/// which raw stream IDs are currently expected. +pub fn classify_first_frame( + is_pipe_request: bool, + pipe_id: Option, + pipe_is_expected: bool, +) -> Result { + if !is_pipe_request { + return Ok(FirstFrameDisposition::Message); + } + + let pipe_id = pipe_id.filter(|id| *id != 0).ok_or_else(|| { + CommunicationError::Other("PipeRequest frame must contain a non-zero id".into()) + })?; + + if pipe_is_expected { + Ok(FirstFrameDisposition::Pipe(pipe_id)) + } else { + Ok(FirstFrameDisposition::Message) + } +} + // ---- manual PartialEq (quinn / wtransport types don't impl PartialEq) ---- impl PartialEq for CommunicationError { @@ -212,6 +247,7 @@ impl PartialEq for CommunicationError { (Self::ReadExactError(_), Self::ReadExactError(_)) => true, (Self::StreamClosed, Self::StreamClosed) => true, (Self::StreamError, Self::StreamError) => true, + (Self::DeliveryUnknown, Self::DeliveryUnknown) => true, #[cfg(not(target_arch = "wasm32"))] (Self::StreamWriteError(_), Self::StreamWriteError(_)) => true, #[cfg(not(target_arch = "wasm32"))] diff --git a/crypto/Cargo.lock b/crypto/Cargo.lock index e536096..5b86012 100644 --- a/crypto/Cargo.lock +++ b/crypto/Cargo.lock @@ -181,9 +181,9 @@ dependencies = [ [[package]] name = "chacha20" -version = "0.10.1" +version = "0.10.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81" +checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06" dependencies = [ "cfg-if", "cpufeatures 0.3.0", @@ -853,7 +853,7 @@ version = "0.10.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80" dependencies = [ - "chacha20 0.10.1", + "chacha20 0.10.2", "getrandom 0.4.3", "rand_core 0.10.1", ] diff --git a/docs/WASM-CLIENT.md b/docs/WASM-CLIENT.md index 02be900..8a56bbd 100644 --- a/docs/WASM-CLIENT.md +++ b/docs/WASM-CLIENT.md @@ -104,6 +104,9 @@ if (!MTPClient.isSupported()) { | `requestTimeoutMs` | 30 seconds | Default `request()` timeout. | | `pings` | `false` | Protocol pings, or an object with `intervalMs`. | | `logger` | No-op | Receives SDK state and error events. | +| `schemas` | None | Client-wide request and response schema registry. | +| `throwProtocolErrors` | `false` | Reject requests whose correlated response is an `Error*` frame. | +| `onValidationError` | No-op | Receives subscription validation failures. | | `sessionStorage` | In-memory | E2EE session state storage. | | `encryptedSecretProvider` | In-memory | Independent caller-managed encrypted secret storage. | | `defaultSignatureVerificationPolicy` | `"ed25519"` | Receiver policy for protected signatures. | @@ -471,6 +474,60 @@ const unsubscribe = client.subscribe("SomeType", (message) => { unsubscribe(); ``` +### Zod request and response schemas + +Applications can provide their request and response schemas once when creating +the client. MTP uses `parseAsync`, so synchronous schemas, async refinements, +defaults, coercions, and transforms all work. MTP has no runtime dependency on +Zod; the application supplies its preferred Zod version. + +```typescript +import { z } from "zod"; +import { MTPClient, MTPValidationError } from "mtp"; + +const schemas = { + GetUser: { + request: z.object({ UserId: z.number().int().positive() }), + response: z.object({ + UserId: z.number().int().positive(), + Display: z.string(), + }), + }, +}; + +const client = await MTPClient.create({ + url, + schemas, + throwProtocolErrors: true, + onValidationError(error) { + console.error(error.messageType, error.cause); + }, +}); + +const response = await client.request("GetUser", { UserId: 42 }); +console.log(response.data.Display); +``` + +Request schemas run before frame encoding and transmission. Their transformed +output is sent. Response schemas run after request correlation, and their +transformed output replaces `frame.data`; `frame.raw`, when present, remains the +original wire frame. Invalid requests and responses reject with +`MTPValidationError`. Invalid subscription messages do not reach the handler +and are reported through `onValidationError`. + +`throwProtocolErrors: true` converts correlated `Error*` frames into +`MTPProtocolError`. It defaults to `false` for compatibility. + +`MTPProxyConnection` applies the same schema registry to another TypeScript +request/subscription transport, such as a Tauri command and event proxy: + +```typescript +const connection = new MTPProxyConnection(adapter, { + schemas, + throwProtocolErrors: true, +}); +``` + Protocol ping behavior is defined in [Protocol Reference](PROTOCOL-REFERENCE.md#protocol-keepalive). The SDK configuration is: ```typescript diff --git a/example/Cargo.lock b/example/Cargo.lock index 8d2fb30..f81b014 100644 --- a/example/Cargo.lock +++ b/example/Cargo.lock @@ -225,9 +225,9 @@ dependencies = [ [[package]] name = "chacha20" -version = "0.10.1" +version = "0.10.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81" +checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06" dependencies = [ "cfg-if", "cpufeatures 0.3.0", @@ -1305,7 +1305,6 @@ name = "mtp-common" version = "0.3.0" dependencies = [ "quinn", - "rustls", "thiserror 2.0.20", "wtransport", ] @@ -1702,7 +1701,7 @@ version = "0.10.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80" dependencies = [ - "chacha20 0.10.1", + "chacha20 0.10.2", "getrandom 0.4.3", "rand_core 0.10.1", ] diff --git a/example/server/src/main.rs b/example/server/src/main.rs index 17b4164..08969f4 100644 --- a/example/server/src/main.rs +++ b/example/server/src/main.rs @@ -38,6 +38,7 @@ async fn handle_pipe_loopback( conn: &mtp::webserver::WebMTPConnection, request: mtp::host::PipeRequest< mtp::webserver::WebMtpSender, + mtp::webserver::WebMtpReceiver, mtp::webserver::H3TransportReceiver, >, ) -> Result> { diff --git a/flake.nix b/flake.nix index 7951305..0a40973 100644 --- a/flake.nix +++ b/flake.nix @@ -5,26 +5,30 @@ rust-overlay.url = "github:oxalica/rust-overlay"; }; - outputs = { - self, - nixpkgs, - rust-overlay, - }: let - systems = [ - "aarch64-darwin" - "aarch64-linux" - "x86_64-darwin" - "x86_64-linux" - ]; - eachSystem = f: - nixpkgs.lib.foldl' nixpkgs.lib.recursiveUpdate {} ( - map (system: nixpkgs.lib.mapAttrs (_: value: {${system} = value;}) (f system)) systems - ); - in + outputs = + { + self, + nixpkgs, + rust-overlay, + }: + let + systems = [ + "aarch64-darwin" + "aarch64-linux" + "x86_64-darwin" + "x86_64-linux" + ]; + eachSystem = + f: + nixpkgs.lib.foldl' nixpkgs.lib.recursiveUpdate { } ( + map (system: nixpkgs.lib.mapAttrs (_: value: { ${system} = value; }) (f system)) systems + ); + in eachSystem ( - system: let - overlays = [rust-overlay.overlays.default]; - pkgs = import nixpkgs {inherit system overlays;}; + system: + let + overlays = [ rust-overlay.overlays.default ]; + pkgs = import nixpkgs { inherit system overlays; }; rustToolchain = pkgs.rust-bin.stable.latest.default.override { extensions = [ @@ -32,12 +36,12 @@ "clippy" "rustfmt" ]; - targets = ["wasm32-unknown-unknown"]; + targets = [ "wasm32-unknown-unknown" ]; }; clippyCheck = pkgs.writeShellApplication { name = "mtp-clippy"; - runtimeInputs = [rustToolchain]; + runtimeInputs = [ rustToolchain ]; text = '' export MTP_TYPE_MAPS="''${MTP_TYPE_MAPS:-$PWD/example/type-maps.yaml}" cargo clippy --workspace --exclude mtp-wasm --all-targets --all-features -- -D warnings -W unreachable-pub @@ -46,7 +50,7 @@ macheteCheck = pkgs.writeShellApplication { name = "mtp-machete"; - runtimeInputs = [pkgs.cargo-machete]; + runtimeInputs = [ pkgs.cargo-machete ]; text = '' cargo machete "$@" ''; @@ -54,7 +58,15 @@ buildAll = pkgs.writeShellApplication { name = "mtp-build-all"; - runtimeInputs = [rustToolchain pkgs.cargo-deny pkgs.wasm-pack pkgs.pnpm pkgs.coreutils clippyCheck macheteCheck]; + runtimeInputs = [ + rustToolchain + pkgs.cargo-deny + pkgs.wasm-pack + pkgs.pnpm + pkgs.coreutils + clippyCheck + macheteCheck + ]; text = '' export MTP_TYPE_MAPS="''${MTP_TYPE_MAPS:-$PWD/example/type-maps.yaml}" @@ -66,7 +78,6 @@ cargo check --manifest-path example/Cargo.toml --workspace --all-targets --all-features mtp-clippy mtp-machete - pnpm run dup pnpm run build RUSTFLAGS="--cfg web_sys_unstable_apis" wasm-pack test --node wasm pnpm run test:e2e @@ -79,13 +90,17 @@ healthCheck = pkgs.writeShellApplication { name = "mtp-health"; - runtimeInputs = [clippyCheck macheteCheck]; + runtimeInputs = [ + clippyCheck + macheteCheck + ]; text = '' mtp-clippy mtp-machete ''; }; - in { + in + { devShells = { default = pkgs.mkShell { name = "mtp-dev"; diff --git a/host/src/config.rs b/host/src/config.rs index 0cef52a..99d62a2 100644 --- a/host/src/config.rs +++ b/host/src/config.rs @@ -251,6 +251,8 @@ pub struct HostConfig { #[cfg(feature = "crypto")] pub authentication_policy: AuthenticationPolicy, #[cfg(feature = "crypto")] + authentication_policy_explicit: bool, + #[cfg(feature = "crypto")] pub auth_timeout: Duration, #[cfg(feature = "crypto")] pub require_pq: bool, @@ -288,6 +290,8 @@ impl HostConfig { #[cfg(feature = "crypto")] authentication_policy: AuthenticationPolicy::Unauthenticated, #[cfg(feature = "crypto")] + authentication_policy_explicit: false, + #[cfg(feature = "crypto")] auth_timeout: Duration::from_secs(30), #[cfg(feature = "crypto")] require_pq: true, @@ -341,7 +345,9 @@ impl HostConfig { get_existing_client: GetExistingClient, complete_register: CompleteRegister, ) -> Self { - self.authentication_policy = AuthenticationPolicy::ForceAuthentication; + if !self.authentication_policy_explicit { + self.authentication_policy = AuthenticationPolicy::ForceAuthentication; + } self.host_keyring = host_keyring; self.get_existing_client = Box::new(get_existing_client); self.complete_register = Box::new(complete_register); @@ -351,6 +357,7 @@ impl HostConfig { #[cfg(feature = "crypto")] pub fn with_authentication_policy(mut self, policy: AuthenticationPolicy) -> Self { self.authentication_policy = policy; + self.authentication_policy_explicit = true; self } @@ -441,4 +448,78 @@ mod tests { .expect("repeated registration decision") ); } + + fn test_keyring() -> mtp_crypto::Keyring { + mtp_crypto::Keyring::new( + mtp_crypto::KemPublicKey::new(Vec::new()), + mtp_crypto::KemPrivateKey::new(Vec::new()), + mtp_crypto::SignaturePqPublicKey::new(Vec::new()), + mtp_crypto::SignaturePqPrivateKey::new(Vec::new()), + mtp_crypto::SignaturePublicKey::new(Vec::new()), + mtp_crypto::SignaturePrivateKey::new(Vec::new()), + ) + } + + fn test_get_existing_client() -> GetExistingClient { + Box::new(|_, _| Box::pin(async { None })) + } + + fn test_complete_register() -> CompleteRegister { + Box::new(|_, _| Box::pin(async { 1 })) + } + + fn test_config() -> HostConfig { + HostConfig::new( + IpAddr::V4(std::net::Ipv4Addr::LOCALHOST), + 4433, + Vec::new(), + Vec::new(), + ) + } + + #[test] + fn with_authentication_defaults_to_force_authentication() { + let config = test_config().with_authentication( + test_keyring(), + test_get_existing_client(), + test_complete_register(), + ); + + assert_eq!( + config.authentication_policy, + AuthenticationPolicy::ForceAuthentication + ); + } + + #[test] + fn explicit_authentication_policy_before_with_authentication_is_preserved() { + let config = test_config() + .with_authentication_policy(AuthenticationPolicy::AllowAuthentication) + .with_authentication( + test_keyring(), + test_get_existing_client(), + test_complete_register(), + ); + + assert_eq!( + config.authentication_policy, + AuthenticationPolicy::AllowAuthentication + ); + } + + #[test] + fn explicit_authentication_policy_after_with_authentication_is_preserved() { + let config = test_config() + .with_authentication( + test_keyring(), + test_get_existing_client(), + test_complete_register(), + ) + .with_authentication_policy(AuthenticationPolicy::AllowAuthentication); + + assert_eq!( + config.authentication_policy, + AuthenticationPolicy::AllowAuthentication + ); + } } diff --git a/host/src/connection.rs b/host/src/connection.rs index 71c60db..7871d55 100644 --- a/host/src/connection.rs +++ b/host/src/connection.rs @@ -73,7 +73,7 @@ pub struct MTPConnection< #[cfg(feature = "pipes")] pub(crate) app_rx: Mutex>>, #[cfg(feature = "pipes")] - pub(crate) pipe_req_rx: Mutex>>, + pub(crate) pipe_req_rx: Mutex>>, #[cfg(feature = "pipes")] pub(crate) pipe_dispatcher: Arc>, #[cfg(not(feature = "pipes"))] @@ -381,7 +381,7 @@ where }) } - pub async fn receive_pipe(&self) -> Result, CommunicationError> { + pub async fn receive_pipe(&self) -> Result, CommunicationError> { self.pipe_req_rx .lock() .await diff --git a/host/src/engine.rs b/host/src/engine.rs old mode 100644 new mode 100755 index 38e8422..08ca87f --- a/host/src/engine.rs +++ b/host/src/engine.rs @@ -217,6 +217,12 @@ impl HandshakeEngine { _authentication_context: &AuthenticationContext, ) -> Result { let mut first_msg = receiver.receive().await.map_err(AcceptError::Receive)?; + tracing::debug!( + message_type = ?first_msg.get_type(), + version = ?first_msg.get_str(DataType::Version), + client_id = ?first_msg.get_data(DataType::Id), + "received MTP opening message" + ); let version_str = match first_msg.get_data(DataType::Version) { Some(DataValue::Str(s)) => s.clone(), @@ -271,6 +277,11 @@ impl HandshakeEngine { return Err(AcceptError::UnsupportedVersion(client_version)); } }; + tracing::debug!( + client_version = %client_version, + negotiated_version = %negotiated, + "MTP protocol version negotiated" + ); let codec = VersionedCodec::for_version(self.registry.clone(), negotiated.clone()) .ok_or_else(|| AcceptError::UnsupportedVersion(negotiated.clone()))?; @@ -298,6 +309,12 @@ impl HandshakeEngine { ) || registration || first_msg.get_data(DataType::PublicKeys).is_some() || claimed_client_id.is_some_and(|client_id| client_id != 0); + tracing::info!( + claimed_client_id = ?claimed_client_id, + registration, + authentication_requested, + "classified MTP opening authentication mode" + ); if authentication_requested { let attempt = crate::config::AuthenticationAttempt { peer_network_identity: _authentication_context.peer_network_identity.clone(), @@ -555,6 +572,7 @@ impl HandshakeEngine { } // Unknown or zero ID: fall back to guest + tracing::info!("allocating MTP guest identity"); let guest_id_lease = match self.assign_guest_id().await { Ok(lease) => lease, Err(error) => { @@ -563,6 +581,7 @@ impl HandshakeEngine { } }; let guest_id = guest_id_lease.id; + tracing::info!(guest_id, "allocated MTP guest identity"); send_accepted_generic(sender, &negotiated, tm, Some(guest_id)) .await .map_err(AcceptError::Send)?; @@ -1105,6 +1124,12 @@ async fn send_rejection_generic( .add_typed_default(DataType::Connected, DataValue::BoolFalse) .add_typed_default(DataType::ErrorMessage, DataValue::Str(reason.to_string())), }; + tracing::debug!( + reason = %reason, + response_type = ?response.get_type(), + has_version = response.get_data(DataType::Version).is_some(), + "sending MTP handshake rejection" + ); let _ = sender.send(&response).await; } @@ -1138,6 +1163,11 @@ async fn send_accepted_generic( if let Some(id) = assigned_id { response = response.add_typed_default(DataType::Id, DataValue::UnsignedNumber(id as u128)); } + tracing::debug!( + version = %version, + assigned_id = ?assigned_id, + "sending accepted MTP handshake response" + ); sender.send(&response).await?; sender.finish_stream().await } diff --git a/host/src/pipe.rs b/host/src/pipe.rs index eae1383..192e3d7 100644 --- a/host/src/pipe.rs +++ b/host/src/pipe.rs @@ -27,6 +27,10 @@ pub trait PipeReceiver

: Clone + Send + Sync + 'static where P: tokio::io::AsyncRead + Send + Unpin + 'static, { + fn expect_pipe(&self, pipe_id: u32) -> Result<(), CommunicationError>; + + fn cancel_expected_pipe(&self, pipe_id: u32); + fn receive_pipe_event( &self, ) -> impl std::future::Future, CommunicationError>> + Send; @@ -52,6 +56,14 @@ impl PipeSender for mtp_transport::Sender { } impl PipeReceiver for mtp_transport::Receiver { + fn expect_pipe(&self, pipe_id: u32) -> Result<(), CommunicationError> { + self.expect_pipe(pipe_id) + } + + fn cancel_expected_pipe(&self, pipe_id: u32) { + self.cancel_expected_pipe(pipe_id); + } + async fn receive_pipe_event( &self, ) -> Result, CommunicationError> { @@ -87,6 +99,14 @@ where C: mtp_transport::TransportConnection, C::RecvStream: tokio::io::AsyncRead + Send + Unpin + 'static, { + fn expect_pipe(&self, pipe_id: u32) -> Result<(), CommunicationError> { + self.expect_pipe(pipe_id) + } + + fn cancel_expected_pipe(&self, pipe_id: u32) { + self.cancel_expected_pipe(pipe_id); + } + async fn receive_pipe_event( &self, ) -> Result, CommunicationError> { @@ -152,16 +172,60 @@ where } } -pub struct PipeRequest { +pub struct PipeRequest { pub(crate) pipe_id: u32, pub(crate) description: String, pub(crate) sender: S, + pub(crate) receiver: R, pub(crate) dispatcher: Arc>, } -impl PipeRequest +struct ExpectedPipeGuard +where + R: PipeReceiver

, + P: tokio::io::AsyncRead + Send + Unpin + 'static, +{ + receiver: R, + pipe_id: u32, + armed: bool, + _stream: std::marker::PhantomData

, +} + +impl ExpectedPipeGuard +where + R: PipeReceiver

, + P: tokio::io::AsyncRead + Send + Unpin + 'static, +{ + fn new(receiver: R, pipe_id: u32) -> Self { + Self { + receiver, + pipe_id, + armed: true, + _stream: std::marker::PhantomData, + } + } + + fn disarm(&mut self) { + self.armed = false; + } +} + +impl Drop for ExpectedPipeGuard +where + R: PipeReceiver

, + P: tokio::io::AsyncRead + Send + Unpin + 'static, +{ + fn drop(&mut self) { + if self.armed { + self.receiver.cancel_expected_pipe(self.pipe_id); + } + } +} + +impl PipeRequest where S: PipeSender, + R: PipeReceiver

, P: tokio::io::AsyncRead + Send + Unpin + 'static, { pub fn id(&self) -> u32 { @@ -173,6 +237,10 @@ where } pub async fn accept(self) -> Result, PipeError> { + self.receiver + .expect_pipe(self.pipe_id) + .map_err(PipeError::from)?; + let mut expected_pipe = ExpectedPipeGuard::::new(self.receiver.clone(), self.pipe_id); let (pipe_tx, pipe_rx) = tokio::sync::oneshot::channel(); self.dispatcher .pending_pipes @@ -196,7 +264,10 @@ where } match tokio::time::timeout(self.dispatcher.policy.read_timeout, pipe_rx).await { - Ok(Ok(reader)) => Ok(reader), + Ok(Ok(reader)) => { + expected_pipe.disarm(); + Ok(reader) + } Ok(Err(_)) => { self.dispatcher .pending_pipes @@ -361,7 +432,7 @@ pub(crate) async fn run_dispatcher( receiver: R, sender: S, app_tx: mpsc::Sender>, - pipe_req_tx: mpsc::Sender>, + pipe_req_tx: mpsc::Sender>, dispatcher: Arc>, ) where S: PipeSender, @@ -388,6 +459,7 @@ pub(crate) async fn run_dispatcher( .unwrap_or("") .to_owned(), sender: sender.clone(), + receiver: receiver.clone(), dispatcher: dispatcher.clone(), }; let _ = pipe_req_tx.send(request).await; @@ -446,6 +518,7 @@ pub(crate) async fn run_dispatcher( pipe_id, description: reader.description().to_owned(), sender: sender.clone(), + receiver: receiver.clone(), dispatcher: dispatcher.clone(), }; let _ = pipe_req_tx.send(request).await; diff --git a/mtp-webserver/src/h3.rs b/mtp-webserver/src/h3.rs index c09379c..abb17aa 100644 --- a/mtp-webserver/src/h3.rs +++ b/mtp-webserver/src/h3.rs @@ -143,6 +143,11 @@ pub(crate) async fn run_driver( return; } }; + tracing::debug!( + remote = %remote_addr, + session_id = ?session.session_id(), + "accepted WebTransport MTP session" + ); tokio::spawn(run_session_requests( session.clone(), router.clone(), diff --git a/mtp-webserver/src/transport.rs b/mtp-webserver/src/transport.rs index a484f26..9dcce5f 100644 --- a/mtp-webserver/src/transport.rs +++ b/mtp-webserver/src/transport.rs @@ -32,6 +32,8 @@ pub struct H3TransportSender { pub struct H3TransportReceiver { stream: H3RecvStream, + quinn: quinn::Connection, + read_exact_calls: u64, } impl H3TransportConnection { @@ -55,14 +57,14 @@ impl TransportSendStream for H3TransportSender { self.stream .write_all(buf) .await - .map_err(|_| CommunicationError::StreamError)?; + .map_err(|_| CommunicationError::DeliveryUnknown)?; // Control/authentication frames use a persistent stream. h3 keeps // those writes buffered until flushed; without this the peer can wait // for the challenge while the server waits for its proof. self.stream .flush() .await - .map_err(|_| CommunicationError::StreamError) + .map_err(|_| CommunicationError::DeliveryUnknown) } async fn finish(&mut self) -> Result<(), CommunicationError> { @@ -71,23 +73,53 @@ impl TransportSendStream for H3TransportSender { .await .map_err(|_| CommunicationError::StreamError) } + + fn reset(&mut self, code: u32) -> Result<(), CommunicationError> { + h3::quic::SendStream::reset(&mut self.stream, code as u64); + Ok(()) + } } #[async_trait::async_trait] impl TransportRecvStream for H3TransportReceiver { async fn read_exact(&mut self, buf: &mut [u8]) -> Result<(), CommunicationError> { + let first_read = self.read_exact_calls == 0; + self.read_exact_calls += 1; self.stream .read_exact(buf) .await - .map(|_| ()) + .map(|_| { + if first_read { + tracing::debug!( + remote = %self.quinn.remote_address(), + bytes = buf.len(), + header = ?buf, + "received first bytes from WebTransport MTP stream" + ); + } + }) .map_err(|error| { - if error.kind() == std::io::ErrorKind::UnexpectedEof { - // Browser control frames are sent on one-frame uni streams. - // Reaching FIN while looking for another frame is normal. + if error.kind() == std::io::ErrorKind::UnexpectedEof + || self.quinn.close_reason().is_some() + { + /* + * Reaching FIN, or losing the enclosing QUIC connection, + * is a normal stream-closure path. Do not turn it into a + * frame-header failure and close the connection again. + */ return CommunicationError::StreamClosed; } - error!("[mtp-webserver] receive stream read_exact failed ({} bytes): {error}", buf.len()); - tracing::warn!(len = buf.len(), %error, "WebTransport receive stream read_exact failed"); + error!( + "[mtp-webserver] receive stream read_exact failed ({} bytes): {error}", + buf.len() + ); + tracing::warn!( + remote = %self.quinn.remote_address(), + first_read, + len = buf.len(), + %error, + "WebTransport receive stream read_exact failed" + ); CommunicationError::StreamError }) } @@ -101,6 +133,9 @@ impl TransportRecvStream for H3TransportReceiver { Ok(Some(buf)) } Err(error) => { + if self.quinn.close_reason().is_some() { + return Err(CommunicationError::StreamClosed); + } error!( "[mtp-webserver] receive stream read failed (max {} bytes): {error}", max @@ -110,6 +145,11 @@ impl TransportRecvStream for H3TransportReceiver { } } } + + fn stop(mut self, code: u32) -> Result<(), CommunicationError> { + h3::quic::RecvStream::stop_sending(&mut self.stream, code as u64); + Ok(()) + } } impl tokio::io::AsyncWrite for H3TransportSender { @@ -167,10 +207,27 @@ impl TransportConnection for H3TransportConnection { loop { match self.session.accept_uni().await { Ok(Some((id, stream))) if id == self.session.session_id() => { - return Ok(H3TransportReceiver { stream }); + let stream_id = h3::quic::RecvStream::recv_id(&stream); + tracing::debug!( + remote = %self.quinn.remote_address(), + session_id = ?self.session.session_id(), + stream_id = ?stream_id, + "accepted WebTransport MTP receive stream" + ); + return Ok(H3TransportReceiver { + stream, + quinn: self.quinn.clone(), + read_exact_calls: 0, + }); } - Ok(Some(_)) => { + Ok(Some((stream_session_id, _stream))) => { consecutive_errors = 0; + tracing::debug!( + remote = %self.quinn.remote_address(), + session_id = ?self.session.session_id(), + stream_session_id = ?stream_session_id, + "ignored WebTransport receive stream belonging to another session" + ); continue; } Ok(None) => return Err(CommunicationError::StreamClosed), @@ -306,7 +363,18 @@ async fn accept_web_connection_inner( connection_id, }, ) - .await?; + .await; + #[cfg(feature = "crypto")] + if let Err(error) = &result { + tracing::warn!( + remote = %remote_addr, + connection_id, + %error, + "WebTransport MTP handshake failed" + ); + } + #[cfg(feature = "crypto")] + let result = result?; #[cfg(not(feature = "crypto"))] let result = engine.accept(&sender, &receiver).await?; diff --git a/package.json b/package.json index dd4e7a1..72e0b12 100644 --- a/package.json +++ b/package.json @@ -61,7 +61,6 @@ "pack": "pnpm run release:web", "release:web": "node create-web-release.mjs", "build:all": "nix run .#build-all", - "dup": "jscpd --pattern '**/*.{rs,ts}' --ignore 'target/**' --ignore 'wasm/pkg/**' --min-lines 8 --min-tokens 80 --threshold 4 --reporters console --no-tips .", "test:e2e": "tsc && node test/e2ee.mjs", "test:secrets": "tsc && node --test --test-isolation=none test/encrypted-secret.mjs", "test:wasm-init": "tsc && node --test test/wasm-init.mjs", diff --git a/src/sdk/client.ts b/src/sdk/client.ts index d3c31be..e0307cd 100644 --- a/src/sdk/client.ts +++ b/src/sdk/client.ts @@ -10,6 +10,15 @@ import * as bindings from "mtp/raw"; import { unixTimeMillis, utf8Encode } from "./utils.js"; import type * as RawBindings from "../raw/index"; import type { MTPCommunicationType } from "../type-map/index"; +import { MTPProtocol } from "./schema.js"; +import type { + MTPMessageType, + MTPFrame, + MTPNoSchemas, + MTPRequestData, + MTPResponseFrame, + MTPSchemaRegistry, +} from "./schema.js"; import type { MTPSessionStorage, MTPSessionState } from "./session"; import { MTPSessionManager } from "./session.js"; import { @@ -249,7 +258,9 @@ export interface MTPPublicKeyBundleKeys { sigClPublicKey: Uint8Array; } -export interface MTPClientOptions { +export interface MTPClientOptions< + Registry extends MTPSchemaRegistry = MTPNoSchemas, +> { url: string; descriptor?: string; hostPublicKey?: MTPKeyMaterialInput; @@ -282,6 +293,12 @@ export interface MTPClientOptions { securityProfile?: MTPSecurityProfile; /** One receive resource policy shared by frame and protected-value opening. */ receiveLimits?: MTPReceiveLimits; + /** Application request and response schemas, keyed by communication type. */ + schemas?: Registry; + /** Reject `request()` when the correlated response is an `Error*` frame. */ + throwProtocolErrors?: boolean; + /** Receives subscription validation failures. Request failures reject normally. */ + onValidationError?: (error: import("./schema.js").MTPValidationError) => void; } export interface MTPSecurityProfile { @@ -562,8 +579,8 @@ export interface MTPAcceptEncryptedPipeOptions { signaturePolicy?: MTPSignatureVerificationPolicy; } -type NormalizedMTPClientOptions = Omit< - MTPClientOptions, +type NormalizedMTPClientOptions = Omit< + MTPClientOptions, "hostPublicKey" | "receiveLimits" > & { hostPublicKey?: Uint8Array; @@ -914,14 +931,34 @@ function validateOptions(options) { ) { throw new TypeError("requestTimeoutMs must be a positive safe integer"); } + if (options.schemas != null) { + if (typeof options.schemas !== "object" || Array.isArray(options.schemas)) { + throw new TypeError("schemas must be an object"); + } + for (const [type, pair] of Object.entries(options.schemas)) { + if ( + !pair || + typeof pair !== "object" || + typeof (pair as { request?: { parseAsync?: unknown } }).request + ?.parseAsync !== "function" || + typeof (pair as { response?: { parseAsync?: unknown } }).response + ?.parseAsync !== "function" + ) { + throw new TypeError( + `schemas.${type} must contain request and response schemas with parseAsync()`, + ); + } + } + } } -export class MTPClient { +export class MTPClient { static readonly crypto = crypto; static readonly codec = codec; #credentials: InternalCredentials | null; - #options: NormalizedMTPClientOptions; + #options: NormalizedMTPClientOptions; + readonly #protocol: MTPProtocol | undefined; readonly #protectedReplayGuard = new InMemoryReplayGuard(); readonly #relayReplayGuard = new InMemoryReplayGuard(); readonly raw: MTPRaw; @@ -934,10 +971,17 @@ export class MTPClient { readonly encryptedSecretProvider: MTPEncryptedSecretProvider; private constructor( - options: NormalizedMTPClientOptions, + options: NormalizedMTPClientOptions, client: RawBindings.WasmClient, ) { this.#options = options; + this.#protocol = options.schemas + ? new MTPProtocol({ + schemas: options.schemas, + throwProtocolErrors: options.throwProtocolErrors, + onValidationError: options.onValidationError, + }) + : undefined; this.#credentials = deserializeCredentials(options.credentials); this.raw = { client, bindings }; this.encryptedSecretProvider = @@ -947,7 +991,11 @@ export class MTPClient { ); } - static async create(options: MTPClientOptions): Promise { + static async create< + const Registry extends MTPSchemaRegistry = MTPNoSchemas, + >( + options: MTPClientOptions, + ): Promise> { validateOptions(options); await MTPClient.init(options.wasm); @@ -968,7 +1016,7 @@ export class MTPClient { securityProfile: resolveSecurityProfile(options), }; - let sdk: MTPClient | undefined; + let sdk: MTPClient | undefined; const client = new WasmClient( (state) => emit(normalizedOptions.logger, { @@ -1004,7 +1052,7 @@ export class MTPClient { setReceiveLimits.call(rawClient, normalizedOptions.receiveLimits); } - sdk = new MTPClient(normalizedOptions, client); + sdk = new MTPClient(normalizedOptions, client); await sdk.#loadStoredCredentials(); if (!sdk.#credentials) { sdk.#credentials = { @@ -1238,6 +1286,35 @@ export class MTPClient { }; } + async #parseRequestData( + type: MTPCommunicationType, + data: unknown, + ): Promise> { + if (!this.#protocol || !this.#protocol.schemas[type]) { + return (data ?? {}) as Record; + } + const parsed = await this.#protocol.parseRequest( + type as MTPMessageType, + data as never, + ); + return (parsed ?? {}) as Record; + } + + async #parseResponseData( + requestedType: MTPCommunicationType, + frame: ParsedFrame, + phase: "response" | "subscription" = "response", + ): Promise> { + if (!this.#protocol || !this.#protocol.schemas[requestedType]) { + return frame; + } + return await this.#protocol.parseResponse( + requestedType as MTPMessageType, + frame, + phase, + ); + } + #buildFrame(typeOrFrame, data, options) { if (typeOrFrame instanceof Uint8Array) { if ( @@ -1279,6 +1356,11 @@ export class MTPClient { } async send(message: Uint8Array): Promise; + async send>( + type: Type, + data?: MTPRequestData, + options?: MTPSendOptions, + ): Promise; async send( type: MTPCommunicationType, data: Record, @@ -1286,10 +1368,14 @@ export class MTPClient { ): Promise; async send( typeOrFrame: Uint8Array | MTPCommunicationType, - data?: Record, + data?: unknown, options?: MTPSendOptions, ): Promise { - const message = this.#buildFrame(typeOrFrame, data, options); + const parsedData = + typeof typeOrFrame === "string" + ? await this.#parseRequestData(typeOrFrame, data) + : data; + const message = this.#buildFrame(typeOrFrame, parsedData, options); try { const frame = this.raw.bindings.parse_frame(message); @@ -1345,6 +1431,11 @@ export class MTPClient { data?: never, options?: MTPRequestOptions, ): Promise; + async request>( + type: Type, + data?: MTPRequestData, + options?: MTPRequestOptions, + ): Promise>; async request( type: MTPCommunicationType, data: Record, @@ -1352,15 +1443,19 @@ export class MTPClient { ): Promise; async request( typeOrFrame: Uint8Array | MTPCommunicationType, - data?: Record, + data?: unknown, options: MTPRequestOptions = {}, - ): Promise { + ): Promise> { const timeoutMs = options.timeoutMs ?? this.#options.requestTimeoutMs ?? 30_000; if (!Number.isSafeInteger(timeoutMs) || timeoutMs <= 0) { throw new TypeError("request timeoutMs must be a positive safe integer"); } - const frame = this.#buildFrame(typeOrFrame, data, options); + const parsedData = + typeof typeOrFrame === "string" + ? await this.#parseRequestData(typeOrFrame, data) + : data; + const frame = this.#buildFrame(typeOrFrame, parsedData, options); try { const parsed = this.raw.bindings.parse_frame(frame); emit( @@ -1391,16 +1486,29 @@ export class MTPClient { // The WASM client owns request expiry and its late-response tombstones. // Keeping a second Promise timer here can reject the SDK call while the // protocol request is still allowed to complete successfully. - return await this.raw.client.request( + const response = await this.raw.client.request( frame, options.responseType ?? null, timeoutMs, ); + return typeof typeOrFrame === "string" + ? await this.#parseResponseData(typeOrFrame, response) + : response; } + subscribe>( + type: Type, + handler: ( + message: MTPResponseFrame, + ) => void | Promise, + ): Unsubscribe; subscribe( type: MTPCommunicationType, - handler: (message: ParsedFrame) => void, + handler: (message: ParsedFrame) => void | Promise, + ): Unsubscribe; + subscribe( + type: MTPCommunicationType, + handler: (message: any) => void | Promise, ): Unsubscribe { if (typeof type !== "string" || !type) { throw new TypeError("subscription type must be a non-empty string"); @@ -1408,8 +1516,25 @@ export class MTPClient { if (typeof handler !== "function") { throw new TypeError("subscription handler must be a function"); } - const id = this.raw.client.subscribe(type, handler); - return () => this.raw.client.unsubscribe(id); + let active = true; + const id = this.raw.client.subscribe(type, (message) => { + if (!this.#protocol || !this.#protocol.schemas[type]) { + void handler(message); + return; + } + void this.#parseResponseData(type, message, "subscription").then( + (parsed) => { + if (active) void handler(parsed); + }, + (error) => { + this.#protocol?.reportValidationError(error); + }, + ); + }); + return () => { + active = false; + this.raw.client.unsubscribe(id); + }; } #handleFrame(frame) { diff --git a/src/sdk/index.ts b/src/sdk/index.ts index e93ca7e..5e3a1e9 100644 --- a/src/sdk/index.ts +++ b/src/sdk/index.ts @@ -5,3 +5,4 @@ * keeps the package's historical exports stable. */ export * from "./client.js"; +export * from "./schema.js"; diff --git a/src/sdk/schema.ts b/src/sdk/schema.ts new file mode 100644 index 0000000..9b9aaf6 --- /dev/null +++ b/src/sdk/schema.ts @@ -0,0 +1,234 @@ +import type { MTPRequestOptions, ParsedFrame, Unsubscribe } from "./client.js"; +import type { MTPCommunicationType } from "../type-map/index.js"; + +export interface MTPSchema { + readonly _input: Input; + readonly _output: Output; + parseAsync(value: unknown): Promise; +} + +export interface MTPSchemaPair< + Request extends MTPSchema = MTPSchema, + Response extends MTPSchema = MTPSchema, +> { + request: Request; + response: Response; +} + +export type MTPSchemaRegistry = Record; +export type MTPNoSchemas = Record; + +export type MTPSchemaInput = Schema["_input"]; +export type MTPSchemaOutput = Schema["_output"]; +export type MTPMessageType = + keyof Registry & string; + +export type MTPFrame = { + id?: number; + type: string; + data: Data; + sender?: ParsedFrame["sender"]; + receiver?: ParsedFrame["receiver"]; + raw?: ParsedFrame["raw"]; +}; + +export type MTPTypedFrame = MTPFrame; + +export type MTPResponseFrame< + Registry extends MTPSchemaRegistry, + Type extends MTPMessageType, +> = MTPTypedFrame>; + +export type MTPRequestData< + Registry extends MTPSchemaRegistry, + Type extends MTPMessageType, +> = MTPSchemaInput; + +export type MTPRequestFunction = < + Type extends MTPMessageType, +>( + type: Type, + data?: MTPRequestData, + options?: MTPRequestOptions, +) => Promise>; + +export type MTPSubscriptionFunction = < + Type extends MTPMessageType, +>( + type: Type, + handler: (message: MTPResponseFrame) => void | Promise, +) => Unsubscribe; + +export class MTPValidationError extends Error { + readonly phase: "request" | "response" | "subscription"; + readonly messageType: string; + readonly frame?: MTPFrame; + + constructor( + phase: MTPValidationError["phase"], + messageType: string, + cause: unknown, + frame?: MTPFrame, + ) { + super(`${phase} validation failed for ${messageType}`, { cause }); + this.name = "MTPValidationError"; + this.phase = phase; + this.messageType = messageType; + this.frame = frame; + } +} + +export class MTPProtocolError extends Error { + readonly type: string; + readonly id: number | undefined; + readonly communicationType: string; + readonly requestId: number | undefined; + readonly errorType: string | undefined; + readonly frame: MTPFrame; + + constructor(frame: MTPFrame) { + const errorType = + frame.data && + typeof frame.data === "object" && + !Array.isArray(frame.data) && + typeof (frame.data as Record).ErrorType === "string" + ? ((frame.data as Record).ErrorType as string) + : undefined; + super(errorType ? `${frame.type}: ${errorType}` : frame.type); + this.name = "MTPProtocolError"; + this.type = frame.type; + this.id = frame.id; + this.communicationType = frame.type; + this.requestId = frame.id; + this.errorType = errorType; + this.frame = frame; + } +} + +export interface MTPProtocolOptions { + schemas: Registry; + throwProtocolErrors?: boolean; + onValidationError?: (error: MTPValidationError) => void; +} + +function isErrorFrame(frame: MTPFrame): boolean { + return frame.type.startsWith("Error"); +} + +export class MTPProtocol { + readonly schemas: Registry; + readonly #throwProtocolErrors: boolean; + readonly #onValidationError: + | ((error: MTPValidationError) => void) + | undefined; + + constructor(options: MTPProtocolOptions) { + this.schemas = options.schemas; + this.#throwProtocolErrors = options.throwProtocolErrors ?? false; + this.#onValidationError = options.onValidationError; + } + + async parseRequest>( + type: Type, + data: MTPRequestData | undefined, + ): Promise> { + try { + return await this.schemas[type].request.parseAsync(data); + } catch (error) { + throw new MTPValidationError("request", type, error); + } + } + + async parseResponse>( + requestedType: Type, + frame: MTPFrame, + phase: "response" | "subscription" = "response", + ): Promise> { + if (isErrorFrame(frame)) { + if (phase === "response" && this.#throwProtocolErrors) { + throw new MTPProtocolError(frame); + } + return frame as MTPResponseFrame; + } + + const schema = + this.schemas[frame.type]?.response ?? + this.schemas[requestedType].response; + try { + const data = await schema.parseAsync(frame.data); + return { ...frame, data } as MTPResponseFrame; + } catch (error) { + throw new MTPValidationError( + phase, + frame.type || requestedType, + error, + frame, + ); + } + } + + reportValidationError(error: unknown): void { + if (error instanceof MTPValidationError) { + this.#onValidationError?.(error); + } + } +} + +export interface MTPProxyAdapter { + request( + type: MTPCommunicationType, + data: Record, + options?: MTPRequestOptions, + ): Promise; + subscribe( + type: MTPCommunicationType, + handler: (message: MTPFrame) => void, + ): Unsubscribe; +} + +export class MTPProxyConnection { + readonly #adapter: MTPProxyAdapter; + readonly #protocol: MTPProtocol; + + constructor(adapter: MTPProxyAdapter, options: MTPProtocolOptions) { + this.#adapter = adapter; + this.#protocol = new MTPProtocol(options); + } + + async request>( + type: Type, + data?: MTPRequestData, + options?: MTPRequestOptions, + ): Promise> { + const parsed = await this.#protocol.parseRequest(type, data); + const response = await this.#adapter.request( + type, + (parsed ?? {}) as Record, + options, + ); + return await this.#protocol.parseResponse(type, response); + } + + subscribe>( + type: Type, + handler: ( + message: MTPResponseFrame, + ) => void | Promise, + ): Unsubscribe { + let active = true; + const unsubscribe = this.#adapter.subscribe(type, (message) => { + void this.#protocol.parseResponse(type, message, "subscription").then( + (parsed) => { + if (active) void handler(parsed); + }, + (error) => { + this.#protocol.reportValidationError(error); + }, + ); + }); + return () => { + active = false; + unsubscribe(); + }; + } +} diff --git a/transport/src/connection.rs b/transport/src/connection.rs index dece604..7db09e8 100644 --- a/transport/src/connection.rs +++ b/transport/src/connection.rs @@ -4,6 +4,10 @@ use crate::framing::RetryClassifier; use crate::pipe::PipeReader; use mtp_codec::{CommunicationValue, DecodeError, DecodeLimits, EncodeLimits, TypeMap}; use mtp_common::CommunicationError; +#[cfg(feature = "pipes")] +use mtp_common::{FirstFrameDisposition, classify_first_frame}; +#[cfg(feature = "pipes")] +use std::collections::HashSet; use std::ops::Deref; use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; @@ -288,15 +292,15 @@ impl Sender { Ok(Ok(())) => Ok(()), Ok(Err(wtransport::error::StreamWriteError::Stopped(code))) => { warn!("[Sender] write failed: peer sent STOP_SENDING (error code {code})"); - Err(CommunicationError::StreamClosed) + Err(CommunicationError::DeliveryUnknown) } Ok(Err(other)) => { warn!("[Sender] write failed: {other}"); - Err(CommunicationError::StreamError) + Err(CommunicationError::DeliveryUnknown) } Err(_) => { warn!("[Sender] write timed out (len={})", bytes.len()); - Err(CommunicationError::StreamError) + Err(CommunicationError::DeliveryUnknown) } } } @@ -398,15 +402,15 @@ impl Sender { Ok(Ok(())) => Ok(()), Ok(Err(wtransport::error::StreamWriteError::Stopped(code))) => { warn!("[Sender] finish failed: peer sent STOP_SENDING (error code {code})"); - Err(CommunicationError::StreamClosed) + Err(CommunicationError::DeliveryUnknown) } Ok(Err(other)) => { warn!("[Sender] finish failed: {other}"); - Err(CommunicationError::StreamError) + Err(CommunicationError::DeliveryUnknown) } Err(_) => { warn!("[Sender] finish timed out"); - Err(CommunicationError::StreamError) + Err(CommunicationError::DeliveryUnknown) } } } @@ -745,6 +749,8 @@ struct ReceiverInner { max_message_size: Arc, type_map: Arc>, decode_rejections: Arc, + #[cfg(feature = "pipes")] + expected_pipes: Arc>>, } impl Clone for Receiver { @@ -831,6 +837,10 @@ impl Receiver { let accept_type_map = type_map.clone(); let decode_rejections = Arc::new(DecodeRejectionCounters::default()); let accept_decode_rejections = decode_rejections.clone(); + #[cfg(feature = "pipes")] + let expected_pipes = Arc::new(std::sync::Mutex::new(HashSet::new())); + #[cfg(feature = "pipes")] + let accept_expected_pipes = expected_pipes.clone(); let stream_limit = Arc::new(Semaphore::new(policy.max_concurrent_stream_tasks.max(1))); let accept_stream_limit = stream_limit.clone(); debug!( @@ -900,6 +910,8 @@ impl Receiver { let stream_max_message_size = accept_max_message_size.clone(); let stream_type_map = accept_type_map.clone(); let stream_decode_rejections = accept_decode_rejections.clone(); + #[cfg(feature = "pipes")] + let stream_expected_pipes = accept_expected_pipes.clone(); tokio::spawn(async move { let _permit = permit; @@ -935,38 +947,54 @@ impl Receiver { #[cfg(feature = "pipes")] { - if msg.is_type(mtp_codec::CommunicationType::PipeRequest) - && frame_count == 1 - { - let Some(pipe_id) = msg.id().filter(|id| *id != 0) else { - let error = CommunicationError::Other( - "PipeRequest frame must contain a non-zero id".into(), - ); - let _ = msg_tx_stream.send(Err(error.clone())).await; - stream_handle.close(Some(error)); + if frame_count == 1 { + let is_pipe_request = msg.is_type( + mtp_codec::CommunicationType::PipeRequest, + ); + let pipe_id = msg.id().filter(|id| *id != 0); + let pipe_is_expected = is_pipe_request && pipe_id.is_some_and(|pipe_id| { + stream_expected_pipes + .lock() + .is_ok_and(|mut expected| expected.remove(&pipe_id)) + }); + let disposition = match classify_first_frame( + is_pipe_request, + msg.id(), + pipe_is_expected, + ) { + Ok(disposition) => disposition, + Err(error) => { + let _ = msg_tx_stream + .send(Err(error.clone())) + .await; + stream_handle.close(Some(error)); + break; + } + }; + + if let FirstFrameDisposition::Pipe(pipe_id) = disposition { + let description = msg + .get_str(mtp_codec::DataType::Description) + .unwrap_or("") + .to_string(); + + let pipe_reader = crate::pipe::PipeReader { + stream: s, + description, + pipe_id, + }; + + if pipe_tx_stream + .send(pipe_reader) + .await + .is_err() + { + stream_handle.close(Some( + CommunicationError::StreamClosed, + )); + } break; - }; - let description = msg - .get_str(mtp_codec::DataType::Description) - .unwrap_or("") - .to_string(); - - let pipe_reader = crate::pipe::PipeReader { - stream: s, - description, - pipe_id, - }; - - if pipe_tx_stream - .send(pipe_reader) - .await - .is_err() - { - stream_handle.close(Some( - CommunicationError::StreamClosed, - )); } - break; } } @@ -1111,6 +1139,8 @@ impl Receiver { max_message_size, type_map, decode_rejections, + #[cfg(feature = "pipes")] + expected_pipes, }), } } @@ -1127,6 +1157,26 @@ impl Receiver { *self.inner.type_map.write().await = type_map.clone(); } + #[cfg(feature = "pipes")] + pub fn expect_pipe(&self, pipe_id: u32) -> Result<(), CommunicationError> { + if pipe_id == 0 { + return Err(CommunicationError::Other("pipe id must be non-zero".into())); + } + self.inner + .expected_pipes + .lock() + .map_err(|_| CommunicationError::Other("expected pipe state is unavailable".into()))? + .insert(pipe_id); + Ok(()) + } + + #[cfg(feature = "pipes")] + pub fn cancel_expected_pipe(&self, pipe_id: u32) { + if let Ok(mut expected) = self.inner.expected_pipes.lock() { + expected.remove(&pipe_id); + } + } + /// Return local counts for frames rejected by the structured decoder. /// /// These counters are intentionally local-only; peers continue to receive diff --git a/transport/src/framing.rs b/transport/src/framing.rs index 9fa1d80..ac4e005 100644 --- a/transport/src/framing.rs +++ b/transport/src/framing.rs @@ -82,6 +82,10 @@ mod tests { async fn finish(&mut self) -> Result<(), CommunicationError> { Ok(()) } + + fn reset(&mut self, _code: u32) -> Result<(), CommunicationError> { + Ok(()) + } } #[tokio::test] diff --git a/transport/src/generic_connection.rs b/transport/src/generic_connection.rs index 23f3fe9..4361855 100644 --- a/transport/src/generic_connection.rs +++ b/transport/src/generic_connection.rs @@ -9,9 +9,13 @@ use crate::{ connection::{DecodeRejectionCounters, RuntimePolicy, classify_decode_error}, framing::{RetryClassifier, write_frame}, }; -use mtp_codec::{CommunicationValue, DecodeLimits, TypeMap}; -use mtp_common::CommunicationError; +use mtp_codec::{CommunicationValue, DataType, DecodeLimits, TypeMap}; +use mtp_common::{CommunicationError, FirstFrameDisposition, classify_first_frame}; +#[cfg(feature = "pipes")] +use std::collections::HashSet; use std::sync::Arc; +#[cfg(feature = "pipes")] +use std::sync::Mutex as StdMutex; use std::sync::atomic::{AtomicU64, Ordering}; use tokio::sync::{Mutex, Notify, RwLock, Semaphore, mpsc}; use tokio::time::{Instant, timeout, timeout_at}; @@ -67,6 +71,15 @@ impl GenericSender { if self.connection.close_reason().is_some() { return Err(CommunicationError::StreamClosed); } + if let Some(version) = value.get_str(DataType::Version) { + tracing::debug!( + message_type = ?value.get_type(), + version, + connected = ?value.get_data(DataType::Connected), + client_id = ?value.get_data(DataType::Id), + "sending MTP handshake response frame" + ); + } match self.policy.send_mode { crate::SendMode::SingleStreamPerMessage => { let mut stream = self.open().await?; @@ -76,9 +89,10 @@ impl GenericSender { ) .await .map_err(|_| CommunicationError::StreamError)??; - timeout(self.policy.write_timeout, stream.finish()) - .await - .map_err(|_| CommunicationError::StreamError)? + match timeout(self.policy.write_timeout, stream.finish()).await { + Ok(Ok(())) => Ok(()), + Ok(Err(_)) | Err(_) => Err(CommunicationError::DeliveryUnknown), + } } crate::SendMode::PersistentStream => { let mut stream = self.persistent.lock().await; @@ -195,6 +209,8 @@ pub struct GenericReceiver { type_map: Arc>, queue_notify: Arc, decode_rejections: Arc, + #[cfg(feature = "pipes")] + expected_pipes: Arc>>, _accept_task: Arc>, } @@ -210,6 +226,8 @@ impl Clone for GenericReceiver { type_map: self.type_map.clone(), queue_notify: self.queue_notify.clone(), decode_rejections: self.decode_rejections.clone(), + #[cfg(feature = "pipes")] + expected_pipes: self.expected_pipes.clone(), _accept_task: self._accept_task.clone(), } } @@ -245,6 +263,10 @@ impl GenericReceiver { let task_queue_notify = queue_notify.clone(); let decode_rejections = Arc::new(DecodeRejectionCounters::default()); let task_decode_rejections = decode_rejections.clone(); + #[cfg(feature = "pipes")] + let expected_pipes = Arc::new(StdMutex::new(HashSet::new())); + #[cfg(feature = "pipes")] + let task_expected_pipes = expected_pipes.clone(); let task_accept_task_tx = tx.clone(); #[cfg(feature = "pipes")] let task_accept_task_pipe_tx = pipe_tx.clone(); @@ -303,6 +325,8 @@ impl GenericReceiver { let connection = task_connection.clone(); let type_map = task_type_map.clone(); let decode_rejections = task_decode_rejections.clone(); + #[cfg(feature = "pipes")] + let expected_pipes = task_expected_pipes.clone(); tokio::spawn(async move { let _permit = permit; let mut stream = stream; @@ -333,6 +357,18 @@ impl GenericReceiver { break 'stream; } Err(_) => { + if frames == 0 { + tracing::warn!( + timeout = ?policy.read_timeout, + "MTP receive stream timed out before its first complete frame" + ); + } else { + tracing::debug!( + frames, + timeout = ?policy.read_timeout, + "MTP receive stream idle timeout" + ); + } break; } } @@ -376,6 +412,9 @@ impl GenericReceiver { ) .await; if !matches!(&body_read, Ok(Ok(()))) { + if matches!(&body_read, Ok(Err(CommunicationError::StreamClosed))) { + break 'stream; + } tracing::warn!( pipe_chunk_len = chunk_len, ?body_read, @@ -408,42 +447,62 @@ impl GenericReceiver { break; } }; + tracing::debug!( + frames, + frame_len, + message_type = ?message.get_type(), + "decoded MTP receive frame" + ); let negotiated_type_map = type_map.read().await.clone(); message.set_type_map(&negotiated_type_map); #[cfg(feature = "pipes")] { - if message.is_type(mtp_codec::CommunicationType::PipeRequest) - && frames == 1 - { - let Some(pipe_id) = message.id().filter(|id| *id != 0) else { - let error = CommunicationError::Other( - "PipeRequest frame must contain a non-zero id".into(), - ); - let _ = tx.send(Err(error.clone())).await; - connection.close( - policy.application_close_code, - b"pipe request missing id", - ); - break; - }; - let description = message - .get_str(mtp_codec::DataType::Description) - .unwrap_or("") - .to_string(); - - let pipe_reader = PipeReader { - stream, - description, - pipe_id, + if frames == 1 { + let is_pipe_request = + message.is_type(mtp_codec::CommunicationType::PipeRequest); + let pipe_id = message.id().filter(|id| *id != 0); + let pipe_is_expected = is_pipe_request + && pipe_id.is_some_and(|pipe_id| { + expected_pipes + .lock() + .is_ok_and(|mut expected| expected.remove(&pipe_id)) + }); + let disposition = match classify_first_frame( + is_pipe_request, + message.id(), + pipe_is_expected, + ) { + Ok(disposition) => disposition, + Err(error) => { + let _ = tx.send(Err(error.clone())).await; + connection.close( + policy.application_close_code, + b"pipe request missing id", + ); + break; + } }; - tracing::debug!(pipe_id, description = %pipe_reader.description, "classified incoming pipe stream"); + if let FirstFrameDisposition::Pipe(pipe_id) = disposition { + let description = message + .get_str(mtp_codec::DataType::Description) + .unwrap_or("") + .to_string(); - if pipe_tx.send(pipe_reader).await.is_err() { - break; + let pipe_reader = PipeReader { + stream, + description, + pipe_id, + }; + + tracing::debug!(pipe_id, description = %pipe_reader.description, "classified incoming pipe stream"); + + if pipe_tx.send(pipe_reader).await.is_err() { + break; + } + return; } - return; } } @@ -487,6 +546,8 @@ impl GenericReceiver { type_map, queue_notify, decode_rejections, + #[cfg(feature = "pipes")] + expected_pipes, _accept_task: Arc::new(accept_task), } } @@ -494,6 +555,25 @@ impl GenericReceiver { *self.ping_sender.write().await = Some(sender); } + #[cfg(feature = "pipes")] + pub fn expect_pipe(&self, pipe_id: u32) -> Result<(), CommunicationError> { + if pipe_id == 0 { + return Err(CommunicationError::Other("pipe id must be non-zero".into())); + } + self.expected_pipes + .lock() + .map_err(|_| CommunicationError::Other("expected pipe state is unavailable".into()))? + .insert(pipe_id); + Ok(()) + } + + #[cfg(feature = "pipes")] + pub fn cancel_expected_pipe(&self, pipe_id: u32) { + if let Ok(mut expected) = self.expected_pipes.lock() { + expected.remove(&pipe_id); + } + } + /// Switch from the handshake frame limit to the application frame limit. pub fn set_max_message_size(&self, max_message_size: u64) { self.max_message_size diff --git a/transport/src/transport_traits.rs b/transport/src/transport_traits.rs index 63c0268..70395af 100644 --- a/transport/src/transport_traits.rs +++ b/transport/src/transport_traits.rs @@ -17,6 +17,7 @@ use mtp_common::CommunicationError; pub trait TransportSendStream: tokio::io::AsyncWrite + Send + Sync { async fn write_all(&mut self, buf: &[u8]) -> Result<(), CommunicationError>; async fn finish(&mut self) -> Result<(), CommunicationError>; + fn reset(&mut self, code: u32) -> Result<(), CommunicationError>; } /// A readable unidirectional stream suitable for MTP frames. @@ -28,6 +29,9 @@ pub trait TransportSendStream: tokio::io::AsyncWrite + Send + Sync { pub trait TransportRecvStream: tokio::io::AsyncRead + Send + Sync { async fn read_exact(&mut self, buf: &mut [u8]) -> Result<(), CommunicationError>; async fn read_chunk(&mut self, max: usize) -> Result>, CommunicationError>; + fn stop(self, code: u32) -> Result<(), CommunicationError> + where + Self: Sized; } /// A QUIC/WebTransport connection that provides MTP's unidirectional streams. @@ -47,7 +51,7 @@ impl TransportSendStream for wtransport::SendStream { async fn write_all(&mut self, buf: &[u8]) -> Result<(), CommunicationError> { wtransport::SendStream::write_all(self, buf) .await - .map_err(|_| CommunicationError::StreamError) + .map_err(|_| CommunicationError::DeliveryUnknown) } async fn finish(&mut self) -> Result<(), CommunicationError> { @@ -55,14 +59,23 @@ impl TransportSendStream for wtransport::SendStream { .await .map_err(|_| CommunicationError::StreamError) } + + fn reset(&mut self, code: u32) -> Result<(), CommunicationError> { + wtransport::SendStream::reset(self, wtransport::VarInt::from_u32(code)) + .map_err(|_| CommunicationError::StreamClosed) + } } #[async_trait] impl TransportRecvStream for wtransport::RecvStream { async fn read_exact(&mut self, buf: &mut [u8]) -> Result<(), CommunicationError> { - wtransport::RecvStream::read_exact(self, buf) - .await - .map_err(|_| CommunicationError::StreamError) + match wtransport::RecvStream::read_exact(self, buf).await { + Ok(()) => Ok(()), + Err(wtransport::error::StreamReadExactError::FinishedEarly(0)) => { + Err(CommunicationError::StreamClosed) + } + Err(_) => Err(CommunicationError::StreamError), + } } async fn read_chunk(&mut self, max: usize) -> Result>, CommunicationError> { @@ -76,6 +89,11 @@ impl TransportRecvStream for wtransport::RecvStream { Err(_) => Err(CommunicationError::StreamError), } } + + fn stop(self, code: u32) -> Result<(), CommunicationError> { + wtransport::RecvStream::stop(self, wtransport::VarInt::from_u32(code)); + Ok(()) + } } #[async_trait] diff --git a/transport/tests/generic_pipe.rs b/transport/tests/generic_pipe.rs index 81a875f..7dfb538 100644 --- a/transport/tests/generic_pipe.rs +++ b/transport/tests/generic_pipe.rs @@ -4,7 +4,7 @@ use async_trait::async_trait; use mtp_codec::CommunicationValue; use mtp_common::CommunicationError; use mtp_transport::{ - GenericReceiver, GenericSender, Policy, TransportConnection, TransportEvent, + GenericReceiver, GenericSender, Policy, SendMode, TransportConnection, TransportEvent, TransportRecvStream, TransportSendStream, }; use std::sync::Arc; @@ -53,6 +53,10 @@ impl TransportSendStream for MockSendStream { .await .map_err(|_| CommunicationError::StreamError) } + + fn reset(&mut self, _code: u32) -> Result<(), CommunicationError> { + Ok(()) + } } struct MockRecvStream { @@ -89,6 +93,10 @@ impl TransportRecvStream for MockRecvStream { Err(_) => Err(CommunicationError::StreamError), } } + + fn stop(self, _code: u32) -> Result<(), CommunicationError> { + Ok(()) + } } #[derive(Clone)] @@ -157,6 +165,7 @@ async fn test_open_pipe_and_receive_reader() -> Result<(), Box Result<(), Box let sender = GenericSender::new(conn_a, policy.clone()); let receiver = GenericReceiver::new(conn_b, policy); + receiver.expect_pipe(1)?; let mut pipe_writer = sender.open_pipe(1, "data-pipe").await?; let data = b"hello through the pipe"; @@ -195,6 +205,7 @@ async fn test_pipe_large_payload() -> Result<(), Box> { let sender = GenericSender::new(conn_a, policy.clone()); let receiver = GenericReceiver::new(conn_b, policy); + receiver.expect_pipe(7)?; let mut pipe_writer = sender.open_pipe(7, "big-pipe").await?; let data: Vec = (0..256 * 1024).map(|i| (i % 256) as u8).collect(); @@ -223,6 +234,7 @@ async fn test_receive_event_dispatches_pipe() -> Result<(), Box Result<(), Box Result<(), Box> { let (conn_a, conn_b) = mock_connected_pair().await; - let policy = Arc::new(Policy::default()); + let policy = Arc::new(Policy::default().with_send_mode(SendMode::SingleStreamPerMessage)); let sender = GenericSender::new(conn_a, policy.clone()); let receiver = GenericReceiver::new(conn_b, policy); - let msg = CommunicationValue::new(mtp_codec::CommunicationType::BadRequest); - sender.send(&msg).await?; - - let _pipe_writer = sender.open_pipe(1, "mixed-pipe").await?; + let request = CommunicationValue::new(mtp_codec::CommunicationType::PipeRequest) + .with_id(1) + .add_typed_default( + mtp_codec::DataType::Description, + mtp_codec::DataValue::Str("mixed-pipe".into()), + ); + sender.send(&request).await?; let received = receiver.receive().await?; - assert_eq!( - received.get_type(), - mtp_codec::CommunicationType::BadRequest - .try_to_id(&mtp_codec::TypeMap::latest()) - .unwrap() - ); + assert!(received.is_type(mtp_codec::CommunicationType::PipeRequest)); + + receiver.expect_pipe(1)?; + let _pipe_writer = sender.open_pipe(1, "mixed-pipe").await?; let pipe_reader = receiver.receive_pipe().await?; assert_eq!(pipe_reader.pipe_id(), 1); @@ -291,6 +304,8 @@ async fn test_multiple_pipes() -> Result<(), Box> { let sender = GenericSender::new(conn_a, policy.clone()); let receiver = GenericReceiver::new(conn_b, policy); + receiver.expect_pipe(10)?; + receiver.expect_pipe(20)?; let mut pw1 = sender.open_pipe(10, "first").await?; let mut pw2 = sender.open_pipe(20, "second").await?; diff --git a/type-map/Cargo.toml b/type-map/Cargo.toml index ad811ae..1cacaa0 100644 --- a/type-map/Cargo.toml +++ b/type-map/Cargo.toml @@ -16,3 +16,7 @@ pipes = [] [build-dependencies] serde = { version = "1", features = ["derive"] } serde_yaml = "0.9" + +[package.metadata.cargo-machete] +# cargo-machete does not inspect build.rs, where both build dependencies are used. +ignored = ["serde", "serde_yaml"] diff --git a/wasm/src/client/authentication.rs b/wasm/src/client/authentication.rs index 45cec8d..73bfc9f 100644 --- a/wasm/src/client/authentication.rs +++ b/wasm/src/client/authentication.rs @@ -8,6 +8,14 @@ use crate::config::ConnectionConfig; use crate::error::js_error; use crate::transport::WasmTransport; +fn server_rejection_message(outcome: &CommunicationValue) -> Option<&str> { + (outcome.get_data(DataType::Connected) == Some(&DataValue::BoolFalse)).then(|| { + outcome + .get_str(DataType::ErrorMessage) + .unwrap_or("host rejected the connection") + }) +} + #[wasm_bindgen] #[allow(deprecated)] impl WasmClient { @@ -79,10 +87,31 @@ impl WasmClient { .unwrap_or("host does not support this protocol version"), )); } + + // Generic host rejections are IdentificationResponse frames with + // Connected=false. They intentionally do not carry a negotiated + // Version because negotiation never completed. Check this before + // reading Version, otherwise a useful server error such as an + // authentication timeout is reported as the misleading + // "host omitted a valid negotiated protocol version". + if let Some(message) = server_rejection_message(&outcome) { + self.set_state_if_current(generation, ConnectionState::Disconnected); + return Err(js_error(message)); + } + + let missing_version = || { + js_error(format!( + "host omitted a valid negotiated protocol version (response_type={:?}, connected={:?}, frame_len={})", + outcome.get_type(), + outcome.get_data(DataType::Connected), + outcome_bytes.len(), + )) + }; + let negotiated_version = match outcome.get_data(DataType::Version) { Some(DataValue::Str(version)) => mtp_codec::Version::parse(version) - .ok_or_else(|| js_error("host omitted a valid negotiated protocol version"))?, - _ => return Err(js_error("host omitted a valid negotiated protocol version")), + .ok_or_else(|| missing_version())?, + _ => return Err(missing_version()), }; if negotiated_version != PROTOCOL_VERSION { return Err(js_error( @@ -628,3 +657,24 @@ impl WasmClient { Ok(server_challenge) } } + +#[cfg(test)] +mod tests { + use super::server_rejection_message; + use mtp_codec::{CommunicationType, CommunicationValue, DataType, DataValue}; + + #[test] + fn reports_rejection_reason_without_a_negotiated_version() { + let response = CommunicationValue::new(CommunicationType::IdentificationResponse) + .add_typed_default(DataType::Connected, DataValue::BoolFalse) + .add_typed_default( + DataType::ErrorMessage, + DataValue::Str("authentication handshake timed out".into()), + ); + + assert_eq!( + server_rejection_message(&response), + Some("authentication handshake timed out") + ); + } +} diff --git a/wasm/src/client/receive.rs b/wasm/src/client/receive.rs index 35ceb02..51a48bf 100644 --- a/wasm/src/client/receive.rs +++ b/wasm/src/client/receive.rs @@ -179,6 +179,7 @@ impl WasmClient { let expired_pipe_creations = self.expired_pipe_creations.clone(); let pending_pipes = self.pending_pipes.clone(); let loop_pending_pipes = pending_pipes.clone(); + let expected_pending_pipes = pending_pipes.clone(); let on_pipe_request = self.on_pipe_request.clone(); let loop_pipe_creations = pending_pipe_creations.clone(); let loop_expired_pipe_creations = expired_pipe_creations.clone(); @@ -293,6 +294,12 @@ impl WasmClient { let _ = entry.sender.send(Ok(pipe_reader)); } }, + move |pipe_id| { + expected_pending_pipes + .borrow() + .get(&pipe_id) + .is_some_and(|entry| entry.generation == loop_generation) + }, ) .await; if connection_generation.get() != generation { diff --git a/wasm/src/pipe.rs b/wasm/src/pipe.rs index efa4f07..88783ae 100644 --- a/wasm/src/pipe.rs +++ b/wasm/src/pipe.rs @@ -1,9 +1,6 @@ -use wasm_bindgen::JsCast; use wasm_bindgen::prelude::*; -use wasm_bindgen_futures::JsFuture; -use crate::error::js_error; -use crate::transport::release_writer_lock; +use crate::transport::{BrowserRecvStream, BrowserSendStream, log_stream_error_code}; #[wasm_bindgen(typescript_custom_section)] const PIPE_TS: &str = r#" @@ -23,54 +20,41 @@ export interface PipeReader { #[wasm_bindgen] pub struct PipeWriter { - writer: JsValue, + stream: BrowserSendStream, pipe_id: u32, } impl PipeWriter { - pub fn new(writer: JsValue, pipe_id: u32) -> Self { - Self { writer, pipe_id } + pub(crate) fn new(stream: BrowserSendStream, pipe_id: u32) -> Self { + Self { stream, pipe_id } + } +} + +impl Drop for PipeWriter { + fn drop(&mut self) { + self.stream.release(); } } #[wasm_bindgen] impl PipeWriter { pub async fn write(&mut self, data: &[u8]) -> Result<(), JsValue> { - let chunk = js_sys::Uint8Array::from(data); - let write_fn = js_sys::Reflect::get(&self.writer, &JsValue::from_str("write")) - .map_err(|_| js_error("missing write"))? - .dyn_into::() - .map_err(|_| js_error("write not a function"))?; - let write_promise = write_fn - .call1(&self.writer, &chunk) - .map_err(|e| js_error(format!("write failed: {:?}", e)))?; - JsFuture::from(write_promise.unchecked_into::()).await?; - Ok(()) + self.stream.write_all(data).await } - pub async fn close(self) -> Result<(), JsValue> { - let close_fn = js_sys::Reflect::get(&self.writer, &JsValue::from_str("close")) - .map_err(|_| js_error("missing close"))? - .dyn_into::() - .map_err(|_| js_error("close not a function"))?; - let close_promise = close_fn - .call0(&self.writer) - .map_err(|e| js_error(format!("close failed: {:?}", e)))?; - if let Err(e) = JsFuture::from(close_promise.unchecked_into::()).await { - crate::transport::log_stream_error_code(&e, "pipe writer close"); + pub async fn close(mut self) -> Result<(), JsValue> { + let result = self.stream.finish().await; + if let Err(error) = &result { + log_stream_error_code(error, "pipe writer close"); } - release_writer_lock(&self.writer); - Ok(()) + self.stream.release(); + result } pub fn abort(&mut self) -> Result<(), JsValue> { - let abort_fn = js_sys::Reflect::get(&self.writer, &JsValue::from_str("abort")) - .map_err(|_| js_error("missing abort"))? - .dyn_into::() - .map_err(|_| js_error("abort not a function"))?; - let _ = abort_fn.call0(&self.writer); - release_writer_lock(&self.writer); - Ok(()) + let result = self.stream.reset(0); + self.stream.release(); + result } pub fn pipe_id(&self) -> u32 { @@ -80,19 +64,26 @@ impl PipeWriter { #[wasm_bindgen] pub struct PipeReader { - reader: JsValue, + stream: BrowserRecvStream, description: String, pipe_id: u32, pending: Vec, + finished: bool, } impl PipeReader { - pub fn new(reader: JsValue, pipe_id: u32, description: String, pending: Vec) -> Self { + pub(crate) fn new( + stream: BrowserRecvStream, + pipe_id: u32, + description: String, + pending: Vec, + ) -> Self { Self { - reader, + stream, pipe_id, description, pending, + finished: false, } } } @@ -105,27 +96,18 @@ impl PipeReader { return Ok(js_sys::Uint8Array::from(&data[..]).into()); } - let read_fn = js_sys::Reflect::get(&self.reader, &JsValue::from_str("read")) - .map_err(|_| js_error("missing read"))? - .dyn_into::() - .map_err(|_| js_error("read not a function"))?; - let promise = read_fn - .call0(&self.reader) - .map_err(|_| js_error("read call failed"))? - .unchecked_into::(); - let result = JsFuture::from(promise).await?; - - let done = js_sys::Reflect::get(&result, &JsValue::from_str("done")) - .ok() - .and_then(|v| v.as_bool()) - .unwrap_or(true); - if done { + if self.finished { return Ok(JsValue::NULL); } - let value = js_sys::Reflect::get(&result, &JsValue::from_str("value")) - .map_err(|_| js_error("missing value"))?; - Ok(js_sys::Uint8Array::new(&value).into()) + match self.stream.read_chunk().await? { + Some(value) => Ok(js_sys::Uint8Array::from(&value[..]).into()), + None => { + self.stream.release(); + self.finished = true; + Ok(JsValue::NULL) + } + } } pub fn pipe_id(&self) -> u32 { @@ -136,3 +118,9 @@ impl PipeReader { self.description.clone() } } + +impl Drop for PipeReader { + fn drop(&mut self) { + self.stream.release(); + } +} diff --git a/wasm/src/transport.rs b/wasm/src/transport.rs index a4deea6..b7ffebe 100644 --- a/wasm/src/transport.rs +++ b/wasm/src/transport.rs @@ -9,6 +9,7 @@ use wasm_bindgen_futures::JsFuture; use crate::error::js_error; use crate::frame::parse_frame_value_with_limits; use mtp_codec::{DecodeLimits, EncodeLimits, TypeMap}; +use mtp_common::{FirstFrameDisposition, classify_first_frame}; const CLOSE_FRAME_LEN: u32 = u32::MAX; @@ -25,12 +26,6 @@ pub(crate) fn log_stream_error_code(error: &JsValue, context: &str) { let stream_error_code = js_sys::Reflect::get(error, &JsValue::from_str("streamErrorCode")) .ok() .and_then(|v| v.as_f64()); - if matches!(stream_error_code, Some(0.0)) { - // WebTransport reports peer-driven stream shutdown as code 0 in this - // environment. For one-frame handshake streams, that is expected and - // should not be surfaced as a warning. - return; - } let message = error .as_string() .or_else(|| { @@ -76,6 +71,233 @@ fn resolve_stream_readable(recv_stream: &JsValue) -> Result { } } +#[derive(Clone)] +struct BrowserConnection { + inner: JsValue, + incoming_reader: Rc>>, +} + +pub(crate) struct BrowserSendStream { + writer: JsValue, +} + +pub(crate) struct BrowserRecvStream { + reader: JsValue, +} + +impl BrowserConnection { + async fn connect(url: &str, cert_hashes: Option>) -> Result { + let constructor = + js_sys::Reflect::get(&js_sys::global(), &JsValue::from_str("WebTransport"))? + .dyn_into::() + .map_err(|_| js_error("WebTransport not available"))?; + let args = js_sys::Array::new(); + args.push(&JsValue::from_str(url)); + + if let Some(hashes) = cert_hashes { + let webtransport_hashes = js_sys::Array::new(); + for hash in hashes { + let (algorithm, value) = hash.split_once(':').unwrap_or(("sha-256", hash.as_str())); + if let Ok(value) = hex::decode(value) { + let entry = js_sys::Object::new(); + js_sys::Reflect::set( + &entry, + &JsValue::from_str("algorithm"), + &JsValue::from_str(algorithm), + )?; + js_sys::Reflect::set( + &entry, + &JsValue::from_str("value"), + &js_sys::Uint8Array::from(&value[..]), + )?; + webtransport_hashes.push(&entry); + } + } + if webtransport_hashes.length() > 0 { + let options = js_sys::Object::new(); + js_sys::Reflect::set( + &options, + &JsValue::from_str("serverCertificateHashes"), + &webtransport_hashes, + )?; + args.push(&options); + } + } + + let inner = js_sys::Reflect::construct(&constructor, &args)?; + let ready = js_sys::Reflect::get(&inner, &JsValue::from_str("ready"))? + .dyn_into::() + .map_err(|_| js_error("WebTransport.ready is not a Promise"))?; + JsFuture::from(ready) + .await + .map_err(|error| js_error(format!("WebTransport ready failed: {error:?}")))?; + Ok(Self { + inner, + incoming_reader: Rc::new(RefCell::new(None)), + }) + } + + async fn open_uni(&self) -> Result { + let create_stream = js_sys::Reflect::get( + &self.inner, + &JsValue::from_str("createUnidirectionalStream"), + )? + .dyn_into::() + .map_err(|_| js_error("createUnidirectionalStream not a function"))?; + let stream_promise = create_stream + .call0(&self.inner)? + .dyn_into::() + .map_err(|_| js_error("createUnidirectionalStream did not return a Promise"))?; + let stream = JsFuture::from(stream_promise).await?; + let writable = resolve_stream_writable(&stream)?; + let writer = js_sys::Reflect::get(&writable, &JsValue::from_str("getWriter")) + .map_err(|_| js_error("missing getWriter"))? + .dyn_into::() + .map_err(|_| js_error("getWriter not a function"))? + .call0(&writable) + .map_err(|_| js_error("getWriter call failed"))?; + Ok(BrowserSendStream { writer }) + } + + async fn accept_uni(&self) -> Result, JsValue> { + let streams_reader = if let Some(reader) = self.incoming_reader.borrow().clone() { + reader + } else { + let incoming = js_sys::Reflect::get( + &self.inner, + &JsValue::from_str("incomingUnidirectionalStreams"), + )?; + let reader = js_sys::Reflect::get(&incoming, &JsValue::from_str("getReader")) + .map_err(|_| js_error("missing getReader"))? + .dyn_into::() + .map_err(|_| js_error("getReader not a function"))? + .call0(&incoming) + .map_err(|_| js_error("getReader call failed"))?; + *self.incoming_reader.borrow_mut() = Some(reader.clone()); + reader + }; + + let read = js_sys::Reflect::get(&streams_reader, &JsValue::from_str("read")) + .map_err(|_| js_error("missing read"))? + .dyn_into::() + .map_err(|_| js_error("read not a function"))?; + let promise = read + .call0(&streams_reader) + .map_err(|_| js_error("read call failed"))? + .unchecked_into::(); + let result = JsFuture::from(promise).await.map_err(|error| { + log_stream_error_code(&error, "accept_uni"); + js_error(format!("accept stream failed: {error:?}")) + })?; + if js_sys::Reflect::get(&result, &JsValue::from_str("done")) + .ok() + .and_then(|value| value.as_bool()) + .unwrap_or(false) + { + return Ok(None); + } + + let stream = js_sys::Reflect::get(&result, &JsValue::from_str("value")) + .map_err(|_| js_error("missing value"))?; + let readable = resolve_stream_readable(&stream)?; + let reader = js_sys::Reflect::get(&readable, &JsValue::from_str("getReader")) + .map_err(|_| js_error("missing stream getReader"))? + .dyn_into::() + .map_err(|_| js_error("stream getReader not a function"))? + .call0(&readable) + .map_err(|_| js_error("stream getReader call failed"))?; + Ok(Some(BrowserRecvStream { reader })) + } + + fn close(&self) { + if let Some(reader) = self.incoming_reader.borrow_mut().take() { + release_reader_lock(&reader); + } + if let Ok(close) = js_sys::Reflect::get(&self.inner, &JsValue::from_str("close")) + .and_then(|value| value.dyn_into::()) + { + let _ = close.call1(&self.inner, &js_sys::Object::new()); + } + } +} + +impl BrowserSendStream { + pub(crate) async fn write_all(&mut self, bytes: &[u8]) -> Result<(), JsValue> { + let write = js_sys::Reflect::get(&self.writer, &JsValue::from_str("write")) + .map_err(|_| js_error("missing write"))? + .dyn_into::() + .map_err(|_| js_error("write not a function"))?; + let promise = write + .call1(&self.writer, &js_sys::Uint8Array::from(bytes)) + .map_err(|error| js_error(format!("write failed: {error:?}")))? + .unchecked_into::(); + JsFuture::from(promise).await.map(|_| ()) + } + + pub(crate) async fn finish(&mut self) -> Result<(), JsValue> { + let close = js_sys::Reflect::get(&self.writer, &JsValue::from_str("close")) + .map_err(|_| js_error("missing close"))? + .dyn_into::() + .map_err(|_| js_error("close not a function"))?; + let promise = close + .call0(&self.writer) + .map_err(|error| js_error(format!("close failed: {error:?}")))? + .unchecked_into::(); + JsFuture::from(promise).await.map(|_| ()) + } + + pub(crate) fn reset(&mut self, code: u32) -> Result<(), JsValue> { + let abort = js_sys::Reflect::get(&self.writer, &JsValue::from_str("abort")) + .map_err(|_| js_error("missing abort"))? + .dyn_into::() + .map_err(|_| js_error("abort not a function"))?; + let _ = abort.call1(&self.writer, &JsValue::from_f64(code as f64))?; + Ok(()) + } + + pub(crate) fn release(&self) { + release_writer_lock(&self.writer); + } +} + +impl BrowserRecvStream { + pub(crate) async fn read_chunk(&mut self) -> Result>, JsValue> { + let read = js_sys::Reflect::get(&self.reader, &JsValue::from_str("read")) + .map_err(|_| js_error("missing read"))? + .dyn_into::() + .map_err(|_| js_error("read not a function"))?; + let promise = read + .call0(&self.reader) + .map_err(|_| js_error("read call failed"))? + .unchecked_into::(); + let result = JsFuture::from(promise).await?; + if js_sys::Reflect::get(&result, &JsValue::from_str("done")) + .ok() + .and_then(|value| value.as_bool()) + .unwrap_or(true) + { + return Ok(None); + } + let value = js_sys::Reflect::get(&result, &JsValue::from_str("value")) + .map_err(|_| js_error("missing value"))?; + Ok(Some(js_sys::Uint8Array::new(&value).to_vec())) + } + + #[allow(dead_code)] + pub(crate) fn stop(self, code: u32) -> Result<(), JsValue> { + let cancel = js_sys::Reflect::get(&self.reader, &JsValue::from_str("cancel")) + .map_err(|_| js_error("missing cancel"))? + .dyn_into::() + .map_err(|_| js_error("cancel not a function"))?; + let _ = cancel.call1(&self.reader, &JsValue::from_f64(code as f64))?; + Ok(()) + } + + pub(crate) fn release(&self) { + release_reader_lock(&self.reader); + } +} + /// Releases a writer's lock so an abandoned writer isn't treated as an abort (which sends STOP_SENDING). pub(crate) fn release_writer_lock(writer: &JsValue) { if let Ok(release) = js_sys::Reflect::get(writer, &JsValue::from_str("releaseLock")) @@ -118,18 +340,14 @@ enum FrameOutcome { */ #[derive(Clone)] pub struct WasmTransport { - inner: JsValue, + connection: BrowserConnection, max_message_size: u32, - /// Reader over `incoming_unidirectional_streams()` (a singleton stream of streams). - streams_reader: Rc>>, - /// Reader over the host's current uni-directional stream, if one is open. - stream_reader: Rc>>, + /// Current incoming unidirectional stream, shared across handshake and receive loops. + stream_reader: Rc>>, /// Bytes already read from the current stream but not yet consumed as a frame. buffer: Rc>>, /// Set to `true` when `open_next_stream` succeeds; cleared after the first frame is parsed. new_stream_frame: Rc>, - /// A single ordered browser send stream shared by all cloned transports. - outgoing_writer: Rc>>, /// Serializes stream creation and writes across concurrent callers. send_lock: Rc>, type_map: Rc>, @@ -151,65 +369,17 @@ impl WasmTransport { max_message_size: u32, configured_limits: Option, ) -> Result { - let ctor = js_sys::Reflect::get(&js_sys::global(), &JsValue::from_str("WebTransport"))? - .dyn_into::() - .map_err(|_| js_error("WebTransport not available"))?; - let args = js_sys::Array::new(); - args.push(&JsValue::from_str(url)); - - if let Some(hashes) = cert_hashes { - let wt_hashes = js_sys::Array::new(); - for h in hashes { - let (algo, hex_val) = match h.split_once(':') { - Some((algo, hex_val)) => (algo, hex_val), - None => ("sha-256", h.as_str()), - }; - - if let Ok(bytes) = hex::decode(hex_val) { - let hash = js_sys::Object::new(); - js_sys::Reflect::set( - &hash, - &JsValue::from_str("algorithm"), - &JsValue::from_str(algo), - )?; - js_sys::Reflect::set( - &hash, - &JsValue::from_str("value"), - &js_sys::Uint8Array::from(&bytes[..]), - )?; - wt_hashes.push(&hash); - } - } - if wt_hashes.length() > 0 { - let opts = js_sys::Object::new(); - js_sys::Reflect::set( - &opts, - &JsValue::from_str("serverCertificateHashes"), - &wt_hashes, - )?; - args.push(&opts); - } - }; - - let transport = js_sys::Reflect::construct(&ctor, &args)?; - let ready = js_sys::Reflect::get(&transport, &JsValue::from_str("ready"))? - .dyn_into::() - .map_err(|_| js_error("WebTransport.ready is not a Promise"))?; - JsFuture::from(ready) - .await - .map_err(|e| js_error(format!("WebTransport ready failed: {:?}", e)))?; + let connection = BrowserConnection::connect(url, cert_hashes).await?; let transport_limits = DecodeLimits::for_transport_message_size(max_message_size as u64); let decode_limits = configured_limits .map(|limits| restrict_decode_limits(limits, transport_limits)) .unwrap_or(transport_limits); Ok(Self { - inner: transport, + connection, max_message_size, - streams_reader: Rc::new(RefCell::new(None)), stream_reader: Rc::new(RefCell::new(None)), buffer: Rc::new(RefCell::new(Vec::new())), new_stream_frame: Rc::new(Cell::new(false)), - outgoing_writer: Rc::new(RefCell::new(None)), send_lock: Rc::new(AsyncMutex::new(())), type_map: Rc::new(RefCell::new(TypeMap::latest())), decode_limits: Rc::new(RefCell::new(decode_limits)), @@ -217,7 +387,7 @@ impl WasmTransport { } pub fn inner(&self) -> &JsValue { - &self.inner + &self.connection.inner } pub fn set_type_map(&self, type_map: &TypeMap) { @@ -247,147 +417,61 @@ impl WasmTransport { return Err(js_error("message too large")); } - let writer_val = if let Some(writer) = self.outgoing_writer.borrow().clone() { - writer - } else { - let create_stream = js_sys::Reflect::get( - &self.inner, - &JsValue::from_str("createUnidirectionalStream"), - )? - .dyn_into::() - .map_err(|_| js_error("createUnidirectionalStream not a function"))?; - let stream_promise = create_stream - .call0(&self.inner)? - .dyn_into::() - .map_err(|_| js_error("createUnidirectionalStream did not return a Promise"))?; - let stream = JsFuture::from(stream_promise).await?; - let writable_or_stream = resolve_stream_writable(&stream)?; - let writer = js_sys::Reflect::get(&writable_or_stream, &JsValue::from_str("getWriter")) - .map_err(|_| js_error("missing getWriter"))? - .dyn_into::() - .map_err(|_| js_error("getWriter not a function"))? - .call0(&writable_or_stream) - .map_err(|_| js_error("getWriter call failed"))?; - *self.outgoing_writer.borrow_mut() = Some(writer.clone()); - writer - }; - - let chunk = js_sys::Uint8Array::from(frame); - - let write_fn = js_sys::Reflect::get(&writer_val, &JsValue::from_str("write")) - .map_err(|_| js_error("missing write"))? - .dyn_into::() - .map_err(|_| js_error("write not a function"))?; - let write_promise = write_fn - .call1(&writer_val, &chunk) - .map_err(|e| js_error(format!("write failed: {:?}", e)))?; - if let Err(e) = JsFuture::from(write_promise.unchecked_into::()).await { + // Use one WebTransport uni-stream per MTP frame. Chromium reliably + // publishes a browser-created uni-stream to the peer when it is + // closed; leaving a shared stream open can leave the server waiting + // in accept_uni() until the authentication deadline. The bytes are + // already the canonical MTP self-framed value, so no extra stream + // length prefix is added here. + let mut stream = self.connection.open_uni().await?; + if let Err(e) = stream.write_all(frame).await { log_stream_error_code(&e, "send_frame write"); - self.outgoing_writer.borrow_mut().take(); - release_writer_lock(&writer_val); + stream.release(); return Err(e); } - Ok(()) - } - - /// Get (creating once) the reader over `incoming_unidirectional_streams()`. - fn ensure_streams_reader(&self) -> Result { - if let Some(reader) = self.streams_reader.borrow().clone() { - return Ok(reader); + if let Err(e) = stream.finish().await { + // The frame was already written; do not retry it merely because + // FIN failed, as that would duplicate the MTP frame. + log_stream_error_code(&e, "send_frame close"); } - let incoming = js_sys::Reflect::get( - &self.inner, - &JsValue::from_str("incomingUnidirectionalStreams"), - )?; - let reader = js_sys::Reflect::get(&incoming, &JsValue::from_str("getReader")) - .map_err(|_| js_error("missing getReader"))? - .dyn_into::() - .map_err(|_| js_error("getReader not a function"))? - .call0(&incoming) - .map_err(|_| js_error("getReader call failed"))?; - *self.streams_reader.borrow_mut() = Some(reader.clone()); - Ok(reader) + stream.release(); + + Ok(()) } /// Accept the next incoming uni-directional stream and make it current. /// Returns `false` if the incoming-streams readable has ended. async fn open_next_stream(&self) -> Result { - let streams_reader = self.ensure_streams_reader()?; - - let read_fn = js_sys::Reflect::get(&streams_reader, &JsValue::from_str("read")) - .map_err(|_| js_error("missing read"))? - .dyn_into::() - .map_err(|_| js_error("read not a function"))?; - let promise = read_fn - .call0(&streams_reader) - .map_err(|_| js_error("read call failed"))? - .unchecked_into::(); - let result = match JsFuture::from(promise).await { - Ok(r) => r, - Err(e) => { - log_stream_error_code(&e, "open_next_stream accept"); - return Err(js_error(format!("accept stream failed: {:?}", e))); - } + let Some(stream) = self.connection.accept_uni().await? else { + return Ok(false); }; - let done = js_sys::Reflect::get(&result, &JsValue::from_str("done")) - .ok() - .and_then(|v| v.as_bool()) - .unwrap_or(false); - if done { - return Ok(false); - } - - let recv_stream = js_sys::Reflect::get(&result, &JsValue::from_str("value")) - .map_err(|_| js_error("missing value"))?; - let readable = resolve_stream_readable(&recv_stream)?; - let reader = js_sys::Reflect::get(&readable, &JsValue::from_str("getReader")) - .map_err(|_| js_error("missing stream getReader"))? - .dyn_into::() - .map_err(|_| js_error("stream getReader not a function"))? - .call0(&readable) - .map_err(|_| js_error("stream getReader call failed"))?; - - *self.stream_reader.borrow_mut() = Some(reader); + *self.stream_reader.borrow_mut() = Some(stream); self.new_stream_frame.set(true); Ok(true) } /// Read one chunk from the current stream. `Ok(None)` means the stream ended. async fn read_chunk(&self) -> Result>, JsValue> { - let reader = match self.stream_reader.borrow().clone() { - Some(r) => r, + let mut stream = match self.stream_reader.borrow_mut().take() { + Some(stream) => stream, None => return Ok(None), }; - - let read_fn = js_sys::Reflect::get(&reader, &JsValue::from_str("read")) - .map_err(|_| js_error("missing read"))? - .dyn_into::() - .map_err(|_| js_error("read not a function"))?; - let promise = read_fn - .call0(&reader) - .map_err(|_| js_error("read call failed"))? - .unchecked_into::(); - let result = match JsFuture::from(promise).await { - Ok(r) => r, + let result = match stream.read_chunk().await { + Ok(result) => result, Err(e) => { log_stream_error_code(&e, "read_chunk"); + stream.release(); return Err(js_error(format!("read failed: {:?}", e))); } }; - - let done = js_sys::Reflect::get(&result, &JsValue::from_str("done")) - .ok() - .and_then(|v| v.as_bool()) - .unwrap_or(true); - if done { - return Ok(None); + if result.is_some() { + *self.stream_reader.borrow_mut() = Some(stream); + } else { + stream.release(); } - - let value = js_sys::Reflect::get(&result, &JsValue::from_str("value")) - .map_err(|_| js_error("missing value"))?; - Ok(Some(js_sys::Uint8Array::new(&value).to_vec())) + Ok(result) } /// Try to pull one complete frame out of the buffer without reading more. @@ -449,9 +533,7 @@ impl WasmTransport { } None => { // Stream finished; release the reader's lock to avoid a spurious cancel. - if let Some(reader) = self.stream_reader.borrow_mut().take() { - release_reader_lock(&reader); - } + // `read_chunk` releases the raw stream lock on clean FIN. // A frame is never allowed to span stream boundaries. The // native persistent-stream sender packs frames on one // stream, while the WASM sender uses one stream per frame; @@ -510,15 +592,17 @@ impl WasmTransport { /// Pipe-aware receive loop. Identical to `receive_loop` but detects /// `PipeRequest` as the first frame on a new incoming stream and routes /// the stream to `on_pipe` instead of `on_message`. - pub async fn receive_loop_with_pipes( + pub async fn receive_loop_with_pipes( &self, mut on_message: F, mut on_error: H, mut on_pipe: G, + mut pipe_is_expected: I, ) where F: FnMut(JsValue), G: FnMut(crate::pipe::PipeReader), H: FnMut(JsValue), + I: FnMut(u32) -> bool, { loop { match self.next_frame(self.max_message_size).await { @@ -540,36 +624,44 @@ impl WasmTransport { if is_first { self.new_stream_frame.set(false); - if let Some(comm) = comm.as_ref() - && Some(comm.get_type()) == pipe_request_type - { - let Some(pipe_id) = comm.id().filter(|id| *id != 0) else { - on_error(JsValue::from_str( - "PipeRequest frame must contain a non-zero id", - )); - self.close(); - break; - }; - let description = comm - .get_str(mtp_codec::DataType::Description) - .unwrap_or("") - .to_string(); + if let Some(comm) = comm.as_ref() { + let is_pipe_request = Some(comm.get_type()) == pipe_request_type; + let pipe_id = comm.id().filter(|id| *id != 0); + let is_expected = + is_pipe_request && pipe_id.is_some_and(&mut pipe_is_expected); + let disposition = + match classify_first_frame(is_pipe_request, comm.id(), is_expected) + { + Ok(disposition) => disposition, + Err(error) => { + on_error(JsValue::from_str(&error.to_string())); + self.close(); + break; + } + }; - let pending = { - let mut buf = self.buffer.borrow_mut(); - std::mem::take(&mut *buf) - }; + if let FirstFrameDisposition::Pipe(pipe_id) = disposition { + let description = comm + .get_str(mtp_codec::DataType::Description) + .unwrap_or("") + .to_string(); - if let Some(reader) = self.stream_reader.borrow_mut().take() { - let pipe_reader = crate::pipe::PipeReader::new( - reader, - pipe_id, - description, - pending, - ); - on_pipe(pipe_reader); + let pending = { + let mut buf = self.buffer.borrow_mut(); + std::mem::take(&mut *buf) + }; + + if let Some(reader) = self.stream_reader.borrow_mut().take() { + let pipe_reader = crate::pipe::PipeReader::new( + reader, + pipe_id, + description, + pending, + ); + on_pipe(pipe_reader); + } + continue; } - continue; } } @@ -625,25 +717,7 @@ impl WasmTransport { description: &str, ) -> Result { let _send_guard = self.send_lock.lock().await; - let create_stream = js_sys::Reflect::get( - &self.inner, - &JsValue::from_str("createUnidirectionalStream"), - )? - .dyn_into::() - .map_err(|_| js_error("createUnidirectionalStream not a function"))?; - let stream_promise = create_stream - .call0(&self.inner)? - .dyn_into::() - .map_err(|_| js_error("createUnidirectionalStream did not return a Promise"))?; - let stream = JsFuture::from(stream_promise).await?; - - let writable_or_stream = resolve_stream_writable(&stream)?; - let writer_val = js_sys::Reflect::get(&writable_or_stream, &JsValue::from_str("getWriter")) - .map_err(|_| js_error("missing getWriter"))? - .dyn_into::() - .map_err(|_| js_error("getWriter not a function"))? - .call0(&writable_or_stream) - .map_err(|_| js_error("getWriter call failed"))?; + let mut stream = self.connection.open_uni().await?; let type_map = self.type_map(); let request = mtp_codec::CommunicationValue::new_with_type_map( @@ -659,43 +733,21 @@ impl WasmTransport { .to_bytes() .map_err(|e| js_error(format!("encode failed: {}", e)))?; - let chunk = js_sys::Uint8Array::from(&frame_bytes[..]); - let write_fn = js_sys::Reflect::get(&writer_val, &JsValue::from_str("write")) - .map_err(|_| js_error("missing write"))? - .dyn_into::() - .map_err(|_| js_error("write not a function"))?; - let write_promise = write_fn - .call1(&writer_val, &chunk) - .map_err(|e| js_error(format!("write failed: {:?}", e)))?; - if let Err(e) = JsFuture::from(write_promise.unchecked_into::()).await { + if let Err(e) = stream.write_all(&frame_bytes).await { log_stream_error_code(&e, "open_pipe write"); - release_writer_lock(&writer_val); + stream.release(); return Err(e); } - Ok(crate::pipe::PipeWriter::new(writer_val, pipe_id)) + Ok(crate::pipe::PipeWriter::new(stream, pipe_id)) } pub fn close(&self) { - if let Some(writer) = self.outgoing_writer.borrow_mut().take() { - // The WebTransport session close below terminates the stream. The - // lock must be released first so dropping it is not interpreted as - // an application abort. - release_writer_lock(&writer); - } // Release reader locks before closing so they aren't treated as cancels. if let Some(reader) = self.stream_reader.borrow_mut().take() { - release_reader_lock(&reader); - } - if let Some(reader) = self.streams_reader.borrow_mut().take() { - release_reader_lock(&reader); - } - - if let Ok(close) = js_sys::Reflect::get(&self.inner, &JsValue::from_str("close")) - .and_then(|value| value.dyn_into::()) - { - let _ = close.call1(&self.inner, &js_sys::Object::new()); + reader.release(); } + self.connection.close(); } }