diff --git a/CHANGELOG.md b/CHANGELOG.md index b5d031b43..a812a25e8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,7 @@ All notable changes to OriginWeave are documented in this file. The format follo ### Added - Bounded RFC 6455 frame transport on the established WebDriver BiDi stream: client text frames require a caller-supplied fresh mask key and are masked on the wire, server frames are required to be unmasked, reserved bits/opcodes and nonminimal lengths fail closed, and each frame is limited by payload and monotonic-I/O ceilings; this remains frame transport only and does not assemble BiDi messages or grant browser/Agent authority. +- Bounded WebDriver BiDi `browsingContext.locateNodes` exchange over the established peer-verified WebSocket stream: the exact consumed command is written as one masked text frame, one end-to-end deadline and a 64-frame Ping/Pong budget bound the exchange, each Pong requires a fresh caller-supplied masking key, and only one final bounded text response may pass raw-document admission, exact command correlation, and node-result admission; fragmentation, binary/continuation/close shapes, exhausted entropy, and over-budget control traffic fail closed without granting browser, origin, policy, typed-input, or Agent authority. - Bounded RFC 6455 WebDriver BiDi opening-response validation on the exact peer-verified stream: it admits only HTTP/1.1 `101`, case-insensitive `Upgrade`/`Connection` tokens, and the client-key-correlated `Sec-WebSocket-Accept` value within monotonic time and header-size ceilings; it restores blocking mode and still does not implement WebSocket frames or grant browser/Agent authority. - Bounded WebDriver BiDi loopback TCP transport that consumes one exact no-DNS connect target, retries only explicitly recoverable local transport failures within repository timeout and attempt ceilings, exposes the stream only after operating-system peer inspection and exact peer verification, supports a consuming handoff of the original stream with typed credential-free peer/session/TLS and bounded-attempt evidence, preserves typed causal errors, and performs no DNS, proxy/PAC, process authentication, TLS, WebSocket, BiDi message, browser-action, or Agent-authority step. - Exact WebDriver BiDi socket-peer verification that consumes an approved no-DNS connect target, requires the observed IP address and port to match exactly, preserves the TLS requirement and exact correlated session id, and remains inert metadata that does not authenticate an OS process, does not negotiate TLS, perform a WebSocket handshake, or grant Agent authority. @@ -55,6 +56,19 @@ All notable changes to OriginWeave are documented in this file. The format follo ### Changed +- Made WebDriver BiDi response-fragment admission reject an already oversized + buffer without subtraction underflow or a production panic. + +- Restored exact locateNodes-stack coverage without introducing an uncovered + test-only pattern-guard branch by exercising the deadline callback + without weakening its no-late-entropy assertion and removing an unreachable + private legacy handshake `Debug` implementation superseded by the redacting + wrapper. +- Removed a redundant 20-microsecond loopback deadline assertion whose error + variant depended on host scheduling; deterministic unit-clock tests retain + the shrinking end-to-end budget contract. +- Kept the no-late-entropy deadline gate as one result chain so exact coverage + does not require recreating an expired-clock race through a real socket. - Separated logical origin authority from resolved network destination authority; an origin grant no longer implies permission to connect to every resolver result. - Separated resolved-address authorization from direct transport evidence; an approved IP now becomes a usable stream only after the operating system reports the exact requested IP and port. - Separated exact TCP peer proof from authenticated TLS service identity; an observed peer becomes an authenticated HTTPS stream only after explicit-root, fixed-time, SAN-bound WebPKI verification over that same stream. diff --git a/crates/originweave-network/src/lib.rs b/crates/originweave-network/src/lib.rs index eafdac09d..59902999d 100644 --- a/crates/originweave-network/src/lib.rs +++ b/crates/originweave-network/src/lib.rs @@ -6,8 +6,9 @@ //! It also bridges a session-correlated WebDriver BiDi loopback target from //! `originweave-core` into one bounded exact TCP connection, binds an RFC 6455 //! opening request to that verified plain stream, and can write that exact request -//! under one bounded deadline and validate its bounded RFC 6455 opening response -//! and one bounded frame at a time without granting browser, WebSocket, TLS, +//! under one bounded deadline, validate its bounded RFC 6455 opening response, +//! carry one bounded frame at a time, and bind one exact `locateNodes` command to +//! its bounded correlated response without granting browser, WebSocket, TLS, //! policy, or Agent authority. #![forbid(unsafe_code)] @@ -15,6 +16,11 @@ mod connection; mod webdriver_bidi_connection; +mod webdriver_bidi_locate_nodes_exchange; +#[cfg(test)] +mod webdriver_bidi_locate_nodes_exchange_transport_failure_tests; +#[cfg(test)] +mod webdriver_bidi_locate_nodes_exchange_unit_coverage_tests; mod webdriver_bidi_websocket_control; #[cfg(test)] #[allow(clippy::expect_used)] @@ -36,6 +42,10 @@ pub use webdriver_bidi_connection::{ WebDriverBiDiTcpConnection, WebDriverBiDiTcpConnectionError, WebDriverBiDiTcpConnectionEvidence, WebDriverBiDiTcpConnectionPlan, }; +pub use webdriver_bidi_locate_nodes_exchange::{ + MAX_WEBDRIVER_BIDI_CONTROL_FRAMES_PER_EXCHANGE, + MAX_WEBDRIVER_BIDI_RESPONSE_FRAGMENTS_PER_EXCHANGE, WebDriverBiDiLocateNodesExchangeError, +}; pub use webdriver_bidi_websocket_handshake::{ WebDriverBiDiWebSocketEstablished, WebDriverBiDiWebSocketHandshakePlan, WebDriverBiDiWebSocketOpeningRequestSent, diff --git a/crates/originweave-network/src/webdriver_bidi_locate_nodes_exchange.rs b/crates/originweave-network/src/webdriver_bidi_locate_nodes_exchange.rs new file mode 100644 index 000000000..95c5f9a78 --- /dev/null +++ b/crates/originweave-network/src/webdriver_bidi_locate_nodes_exchange.rs @@ -0,0 +1,649 @@ +use std::{ + error::Error, + fmt, + time::{Duration, Instant}, +}; + +use originweave_core::{ + BoundedWebDriverBiDiResponseDocument, BrowserAuthorityRegistry, + BrowserContextOriginEpochDispatchTarget, MAX_WEBDRIVER_BIDI_RESPONSE_DOCUMENT_BYTES, + ObservedNodeHandle, ValidatedBrowserProtocolUse, ValidatedWebDriverBiDiLocateNodesResult, + WebDriverBiDiLocateNodesCommand, WebDriverBiDiLocateNodesResponseDocumentError, + WebDriverBiDiResponseDocumentAdmissionError, +}; + +use crate::{ + MAX_WEBSOCKET_FRAME_TIMEOUT, WebDriverBiDiWebSocketEstablished, + WebDriverBiDiWebSocketFrameError, WebDriverBiDiWebSocketMaskKey, +}; + +/// Maximum number of valid RFC 6455 Ping/Pong control frames one `locateNodes` exchange will process. +/// +/// RFC 6455 permits control frames to be interleaved with data frames; this OriginWeave-owned +/// resource budget prevents a peer from turning that permission into an unbounded control-frame loop +/// before the correlated BiDi response arrives. The end-to-end exchange deadline remains an +/// independent wall-clock bound. +pub const MAX_WEBDRIVER_BIDI_CONTROL_FRAMES_PER_EXCHANGE: usize = 64; + +/// Maximum number of data fragments accepted for one `locateNodes` response message. +/// +/// RFC 6455 permits a text message to be split into an arbitrary number of continuation frames, +/// including empty fragments. This OriginWeave product-safety budget prevents a peer from turning +/// bounded response bytes into unbounded frame-processing work before the end-to-end deadline. +pub const MAX_WEBDRIVER_BIDI_RESPONSE_FRAGMENTS_PER_EXCHANGE: usize = 256; + +/// Fail-closed failures while exchanging one bounded WebDriver BiDi `locateNodes` command. +/// +/// Every variant preserves the first causal boundary. Frame I/O retains the existing bounded +/// WebSocket error, raw response bytes must pass the core pre-parser admission contract, and the +/// admitted document must correlate to the exact consumed command before result nodes are returned. +/// Protocol-shape, resource-budget, exhausted-deadline, missing caller entropy, and adjacent client +/// masking-key reuse refusals have no nested source because none masks an underlying I/O or parser +/// failure. +#[derive(Debug)] +pub enum WebDriverBiDiLocateNodesExchangeError { + /// Bounded WebSocket frame write or read failed. + Frame(WebDriverBiDiWebSocketFrameError), + /// The single end-to-end exchange deadline was exhausted before the next operation could proceed. + ExchangeDeadlineExceeded { + /// Original caller-supplied deadline budget for the complete exchange. + exchange_timeout: Duration, + }, + /// The peer exceeded the local resource budget for interleaved Ping/Pong frames. + ControlFrameLimitExceeded { + /// Maximum number of control frames admitted for one exchange. + maximum_control_frames: usize, + }, + /// The peer exceeded the local resource budget for response-message data fragments. + ResponseFragmentLimitExceeded { + /// Maximum number of response-message data fragments admitted for one exchange. + maximum_fragments: usize, + }, + /// A server Ping required a fresh client masking key, but the caller supplied none. + PongMaskingKeyUnavailable, + /// A caller supplied the same Pong masking key as the immediately preceding client frame. + PongMaskingKeyReused, + /// The returned frame could not continue the one admissible text response message. + UnexpectedResponseFrame { + /// Whether the returned frame carried the RFC 6455 FIN bit. + fin: bool, + /// Exact returned RFC 6455 opcode. + opcode: u8, + }, + /// The exact response-message payload failed bounded raw-document admission. + ResponseDocument(WebDriverBiDiResponseDocumentAdmissionError), + /// The admitted response document failed parsing, exact correlation, or node admission. + LocateNodesResponse(WebDriverBiDiLocateNodesResponseDocumentError), +} + +impl fmt::Display for WebDriverBiDiLocateNodesExchangeError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Frame(error) => write!( + formatter, + "WebDriver BiDi locateNodes WebSocket frame exchange failed: {error}" + ), + Self::ExchangeDeadlineExceeded { exchange_timeout } => write!( + formatter, + "WebDriver BiDi locateNodes exchange exhausted its {exchange_timeout:?} end-to-end deadline before the next operation" + ), + Self::ControlFrameLimitExceeded { + maximum_control_frames, + } => write!( + formatter, + "WebDriver BiDi locateNodes exchange exceeded the maximum {maximum_control_frames} interleaved control frames" + ), + Self::ResponseFragmentLimitExceeded { maximum_fragments } => write!( + formatter, + "WebDriver BiDi locateNodes exchange exceeded the maximum {maximum_fragments} response-message data fragments" + ), + Self::PongMaskingKeyUnavailable => formatter.write_str( + "WebDriver BiDi locateNodes exchange received Ping without a fresh caller-supplied Pong masking key", + ), + Self::PongMaskingKeyReused => formatter.write_str( + "WebDriver BiDi locateNodes exchange refused a Pong masking key matching the immediately preceding client frame", + ), + Self::UnexpectedResponseFrame { fin, opcode } => write!( + formatter, + "WebDriver BiDi locateNodes exchange requires control handling or one bounded text response message; received fin={fin}, opcode=0x{opcode:02x}" + ), + Self::ResponseDocument(error) => write!( + formatter, + "WebDriver BiDi locateNodes response message failed raw-document admission: {error}" + ), + Self::LocateNodesResponse(error) => write!( + formatter, + "WebDriver BiDi locateNodes response document failed exact wire admission: {error}" + ), + } + } +} + +impl Error for WebDriverBiDiLocateNodesExchangeError { + fn source(&self) -> Option<&(dyn Error + 'static)> { + match self { + Self::Frame(error) => Some(error), + Self::ResponseDocument(error) => Some(error), + Self::LocateNodesResponse(error) => Some(error), + Self::ExchangeDeadlineExceeded { .. } + | Self::ControlFrameLimitExceeded { .. } + | Self::ResponseFragmentLimitExceeded { .. } + | Self::PongMaskingKeyUnavailable + | Self::PongMaskingKeyReused + | Self::UnexpectedResponseFrame { .. } => None, + } + } +} + +fn remaining_exchange_budget( + exchange_timeout: Duration, + elapsed: Duration, +) -> Result { + match exchange_timeout.checked_sub(elapsed) { + Some(remaining) if !remaining.is_zero() => Ok(remaining), + Some(_) | None => Err( + WebDriverBiDiLocateNodesExchangeError::ExchangeDeadlineExceeded { exchange_timeout }, + ), + } +} + +fn remaining_frame_operation_budget( + exchange_timeout: Duration, + elapsed: Duration, +) -> Result { + remaining_exchange_budget(exchange_timeout, elapsed) + .map(|remaining| remaining.min(MAX_WEBSOCKET_FRAME_TIMEOUT)) +} + +fn next_pong_masking_key( + next_key: &mut dyn FnMut() -> Option, +) -> Result { + next_key().ok_or(WebDriverBiDiLocateNodesExchangeError::PongMaskingKeyUnavailable) +} + +fn next_pong_masking_key_before_deadline( + next_key: &mut dyn FnMut() -> Option, + exchange_timeout: Duration, + elapsed: Duration, +) -> Result { + remaining_frame_operation_budget(exchange_timeout, elapsed) + .and_then(|_| next_pong_masking_key(next_key)) +} + +fn map_established_frame_result( + result: Result, +) -> Result { + result.map_err(WebDriverBiDiLocateNodesExchangeError::Frame) +} + +fn read_frame_with_exchange_budget( + exchange_timeout: Duration, + elapsed: Duration, + read_frame: impl FnOnce(Duration) -> Result, +) -> Result { + let remaining_timeout = remaining_frame_operation_budget(exchange_timeout, elapsed)?; + read_frame(remaining_timeout).map_err(WebDriverBiDiLocateNodesExchangeError::Frame) +} + +fn admit_response_fragment( + response_fragment_count: &mut usize, +) -> Result<(), WebDriverBiDiLocateNodesExchangeError> { + if *response_fragment_count == MAX_WEBDRIVER_BIDI_RESPONSE_FRAGMENTS_PER_EXCHANGE { + return Err( + WebDriverBiDiLocateNodesExchangeError::ResponseFragmentLimitExceeded { + maximum_fragments: MAX_WEBDRIVER_BIDI_RESPONSE_FRAGMENTS_PER_EXCHANGE, + }, + ); + } + *response_fragment_count += 1; + Ok(()) +} + +fn append_response_fragment( + response_message: &mut Vec, + payload: &[u8], +) -> Result<(), WebDriverBiDiLocateNodesExchangeError> { + if response_message.len().saturating_add(payload.len()) + > MAX_WEBDRIVER_BIDI_RESPONSE_DOCUMENT_BYTES + { + return Err(WebDriverBiDiLocateNodesExchangeError::ResponseDocument( + WebDriverBiDiResponseDocumentAdmissionError::DocumentTooLarge, + )); + } + response_message.extend_from_slice(payload); + Ok(()) +} + +fn admit_response_payload( + command: WebDriverBiDiLocateNodesCommand, + payload: &[u8], +) -> Result { + let document = BoundedWebDriverBiDiResponseDocument::from_utf8_bytes(payload) + .map_err(WebDriverBiDiLocateNodesExchangeError::ResponseDocument)?; + command + .admit_response_document_nodes(document) + .map_err(WebDriverBiDiLocateNodesExchangeError::LocateNodesResponse) +} + +impl WebDriverBiDiWebSocketEstablished { + /// Exchange one exact bounded `browsingContext.locateNodes` command on this verified stream. + /// + /// The command is serialized by the reviewed core boundary and written as one masked client + /// text frame using `command_masking_key`. Valid server Ping frames are answered with a masked + /// Pong carrying the exact Ping application data, while unsolicited valid Pong frames are + /// consumed without changing BiDi state. Each Ping obtains a caller-supplied client mask from + /// `next_pong_key` only after a positive remaining-budget check; exhausting that caller-owned + /// entropy source or repeating the immediately preceding successful client-frame key fails + /// closed before another client frame is emitted. A later random collision after a different + /// client key remains admissible; the caller is responsible for deriving every key independently + /// from a strong unpredictable entropy source. Callback time is charged by a second deadline + /// check before the Pong write, and the adjacent-key guard does not claim to prove cryptographic + /// unpredictability. + /// + /// RFC 6455 text-message fragmentation is reassembled only for one response message at a time. + /// A non-final text frame starts that message, continuation frames extend it in order, and a final + /// continuation completes it. Ping/Pong control frames remain admissible between fragments. The + /// total assembled response is capped by [`MAX_WEBDRIVER_BIDI_RESPONSE_DOCUMENT_BYTES`] before + /// allocation can grow beyond the existing pre-parser budget, and at most + /// [`MAX_WEBDRIVER_BIDI_RESPONSE_FRAGMENTS_PER_EXCHANGE`] accepted data fragments may compose the + /// message, so empty continuation frames cannot create unbounded processing work. Orphan + /// continuations, a second data message before completion, binary/Close/reserved shapes, and + /// malformed frame sequences fail closed and consume the transport state. + /// + /// `exchange_timeout` is one end-to-end budget for every command write, control-frame read/write, + /// response-fragment read, and response read. Elapsed time is subtracted before every operation, + /// including the initial command write, and the budget is never reset. Each individual frame + /// operation is additionally capped at the established frame timeout ceiling, so a longer + /// end-to-end exchange budget remains valid without widening the per-operation I/O bound. The + /// underlying frame boundary independently caps each frame at its existing size ceiling. In + /// addition, at most [`MAX_WEBDRIVER_BIDI_CONTROL_FRAMES_PER_EXCHANGE`] valid Ping/Pong frames are + /// processed before the exchange fails closed, so RFC 6455 control-frame interleaving cannot + /// create an unbounded iteration budget even when the wall-clock deadline has not yet expired. + /// Any failure consumes this transport state and yields no reusable WebSocket stream, preventing + /// a partially written/read protocol state from becoming later authority. + /// + /// The final complete text message passes the existing bounded UTF-8/document admission, + /// complete WebDriver BiDi response parser, exact command-id correlation, and wire-derived node + /// admission. Success returns the same exact peer-verified WebSocket stream plus untrusted + /// normalized node evidence. It does not authenticate Chromium/ChromeDriver process provenance, + /// prove current OriginWeave session/context/origin/document authority, authorize policy or typed + /// input, mint node handles, execute a browser action, or prove a post-condition. + pub fn exchange_locate_nodes( + self, + command: WebDriverBiDiLocateNodesCommand, + command_masking_key: WebDriverBiDiWebSocketMaskKey, + next_pong_key: &mut dyn FnMut() -> Option, + exchange_timeout: Duration, + ) -> Result< + (Self, ValidatedWebDriverBiDiLocateNodesResult), + WebDriverBiDiLocateNodesExchangeError, + > { + let started_at = Instant::now(); + let write_timeout = + remaining_frame_operation_budget(exchange_timeout, started_at.elapsed())?; + let mut established = map_established_frame_result(self.write_text_frame( + command.as_json(), + command_masking_key, + write_timeout, + ))?; + let mut control_frame_count = 0_usize; + let mut previous_client_masking_key = command_masking_key; + let mut response_fragment_count = 0_usize; + let mut response_message = Vec::new(); + let mut assembling_text_response = false; + + loop { + let (next_established, frame) = read_frame_with_exchange_budget( + exchange_timeout, + started_at.elapsed(), + |remaining_timeout| established.read_frame(remaining_timeout), + )?; + established = next_established; + let opcode = frame.opcode(); + + if matches!(opcode, 0x9 | 0xa) { + if control_frame_count == MAX_WEBDRIVER_BIDI_CONTROL_FRAMES_PER_EXCHANGE { + return Err( + WebDriverBiDiLocateNodesExchangeError::ControlFrameLimitExceeded { + maximum_control_frames: MAX_WEBDRIVER_BIDI_CONTROL_FRAMES_PER_EXCHANGE, + }, + ); + } + control_frame_count += 1; + } + + match opcode { + 0x9 => { + let masking_key = next_pong_masking_key_before_deadline( + next_pong_key, + exchange_timeout, + started_at.elapsed(), + )?; + if previous_client_masking_key == masking_key { + return Err(WebDriverBiDiLocateNodesExchangeError::PongMaskingKeyReused); + } + let remaining_timeout = + remaining_frame_operation_budget(exchange_timeout, started_at.elapsed())?; + established = map_established_frame_result(established.write_pong_frame( + frame.payload(), + masking_key, + remaining_timeout, + ))?; + previous_client_masking_key = masking_key; + } + 0xa => {} + 0x1 if !assembling_text_response && frame.fin() => { + let result = admit_response_payload(command, frame.payload())?; + return Ok((established, result)); + } + 0x1 if !assembling_text_response => { + response_fragment_count = 1; + append_response_fragment(&mut response_message, frame.payload())?; + assembling_text_response = true; + } + 0x0 if assembling_text_response => { + admit_response_fragment(&mut response_fragment_count)?; + append_response_fragment(&mut response_message, frame.payload())?; + if frame.fin() { + let result = admit_response_payload(command, &response_message)?; + return Ok((established, result)); + } + } + _ => { + return Err( + WebDriverBiDiLocateNodesExchangeError::UnexpectedResponseFrame { + fin: frame.fin(), + opcode, + }, + ); + } + } + } + } + + /// Exchange `locateNodes` and bind the exact wire-derived nodes to current browser authority. + /// + /// This is the live transport composition boundary for semantic node observation. The bounded + /// command is exchanged on the already peer-verified WebSocket using [`Self::exchange_locate_nodes`]. + /// Only after exact wire parsing and command correlation succeed does the method revalidate the + /// reviewed WebDriver BiDi `SemanticObservation` proof and exact current + /// session/context/origin/document epoch carried together in `authority` through + /// [`ValidatedWebDriverBiDiLocateNodesResult::bind_current_nodes`]. No raw node identifier can be + /// substituted between the wire response and authority binding. + /// + /// A binding failure consumes this transport result and returns no reusable stream or node + /// handle, so a navigation or authority change observed after command construction cannot be + /// converted into stale node authority. Success returns only current [`ObservedNodeHandle`] + /// values together with the same established peer-verified stream. It still does not authorize + /// typed input, execute an action, or prove a post-condition. + pub fn exchange_locate_nodes_and_bind_current_nodes( + self, + command: WebDriverBiDiLocateNodesCommand, + command_masking_key: WebDriverBiDiWebSocketMaskKey, + next_pong_key: &mut dyn FnMut() -> Option, + exchange_timeout: Duration, + authority: ( + ValidatedBrowserProtocolUse, + BrowserContextOriginEpochDispatchTarget<'_>, + ), + authority_registry: &mut BrowserAuthorityRegistry, + ) -> Result<(Self, Vec), WebDriverBiDiLocateNodesExchangeError> { + let (validated, target) = authority; + let (established, result) = self.exchange_locate_nodes( + command, + command_masking_key, + next_pong_key, + exchange_timeout, + )?; + let handles = match result.bind_current_nodes(validated, authority_registry, target) { + Ok(handles) => handles, + Err(error) => { + return Err(WebDriverBiDiLocateNodesExchangeError::LocateNodesResponse( + WebDriverBiDiLocateNodesResponseDocumentError::NodeBinding(error), + )); + } + }; + Ok((established, handles)) + } +} + +#[cfg(test)] +mod tests { + use std::{error::Error as _, time::Duration}; + + use originweave_core::{ + MAX_WEBDRIVER_BIDI_RESPONSE_DOCUMENT_BYTES, WebDriverBiDiLocateNodesResponseDocumentError, + WebDriverBiDiResponseDocumentAdmissionError, + }; + + use crate::{MAX_WEBSOCKET_FRAME_TIMEOUT, WebDriverBiDiWebSocketFrameError}; + + use super::{ + MAX_WEBDRIVER_BIDI_CONTROL_FRAMES_PER_EXCHANGE, + MAX_WEBDRIVER_BIDI_RESPONSE_FRAGMENTS_PER_EXCHANGE, WebDriverBiDiLocateNodesExchangeError, + append_response_fragment, next_pong_masking_key, next_pong_masking_key_before_deadline, + read_frame_with_exchange_budget, remaining_exchange_budget, + remaining_frame_operation_budget, + }; + + #[test] + fn exchange_budget_consumes_elapsed_time_instead_of_resetting() { + let total = Duration::from_millis(500); + assert_eq!( + format!( + "{:?}", + remaining_exchange_budget(total, Duration::from_millis(175)) + ), + "Ok(325ms)" + ); + assert_eq!( + format!("{:?}", remaining_exchange_budget(total, total)), + "Err(ExchangeDeadlineExceeded { exchange_timeout: 500ms })" + ); + assert_eq!( + format!( + "{:?}", + remaining_exchange_budget(total, Duration::from_millis(501)) + ), + "Err(ExchangeDeadlineExceeded { exchange_timeout: 500ms })" + ); + } + + #[test] + fn exchange_budget_caps_each_frame_operation_without_resetting_total_time() { + assert_eq!( + remaining_frame_operation_budget(Duration::from_secs(6), Duration::ZERO).ok(), + Some(MAX_WEBSOCKET_FRAME_TIMEOUT) + ); + assert_eq!( + remaining_frame_operation_budget(Duration::from_secs(6), Duration::from_secs(2)).ok(), + Some(Duration::from_secs(4)) + ); + assert!( + remaining_frame_operation_budget(Duration::from_secs(6), Duration::from_secs(6)) + .is_err() + ); + } + + #[test] + fn expired_exchange_budget_refuses_frame_read_before_io() { + use std::cell::Cell; + + let exchange_timeout = Duration::from_millis(500); + let read_count = Cell::new(0_usize); + let read_frame = |remaining_timeout| { + read_count.set(read_count.get() + 1); + Ok::(remaining_timeout) + }; + let available = read_frame_with_exchange_budget( + exchange_timeout, + Duration::from_millis(100), + read_frame, + ); + assert_eq!(available.ok(), Some(Duration::from_millis(400))); + assert_eq!(read_count.get(), 1); + + let expired = + read_frame_with_exchange_budget(exchange_timeout, exchange_timeout, read_frame); + assert_eq!( + format!("{expired:?}"), + "Err(ExchangeDeadlineExceeded { exchange_timeout: 500ms })" + ); + assert_eq!(read_count.get(), 1); + + let frame_error = + read_frame_with_exchange_budget(exchange_timeout, Duration::from_millis(100), |_| { + Err::( + WebDriverBiDiWebSocketFrameError::InvalidFrameTimeout { + frame_timeout: Duration::ZERO, + maximum_timeout: MAX_WEBSOCKET_FRAME_TIMEOUT, + }, + ) + }); + assert!(frame_error.is_err()); + } + + #[test] + fn response_fragment_buffer_never_exceeds_document_budget() { + let mut response = vec![0_u8; MAX_WEBDRIVER_BIDI_RESPONSE_DOCUMENT_BYTES - 1]; + assert!(append_response_fragment(&mut response, b"x").is_ok()); + assert_eq!(response.len(), MAX_WEBDRIVER_BIDI_RESPONSE_DOCUMENT_BYTES); + + assert_eq!( + format!("{:?}", append_response_fragment(&mut response, b"y")), + "Err(ResponseDocument(DocumentTooLarge))" + ); + assert_eq!(response.len(), MAX_WEBDRIVER_BIDI_RESPONSE_DOCUMENT_BYTES); + } + + #[test] + fn response_fragment_rejects_an_already_oversized_buffer_without_panicking() { + let mut response = vec![0_u8; MAX_WEBDRIVER_BIDI_RESPONSE_DOCUMENT_BYTES + 1]; + assert!(append_response_fragment(&mut response, &[]).is_err()); + } + + #[test] + fn pong_masking_key_source_fails_closed_when_entropy_is_unavailable() { + let expected = crate::WebDriverBiDiWebSocketMaskKey::new([1, 2, 3, 4]); + let mut available = || Some(expected); + assert_eq!(next_pong_masking_key(&mut available).ok(), Some(expected)); + + let mut unavailable = || None; + assert_eq!( + format!("{:?}", next_pong_masking_key(&mut unavailable)), + "Err(PongMaskingKeyUnavailable)" + ); + } + + #[test] + fn pong_entropy_is_not_drawn_after_exchange_deadline() { + use std::cell::Cell; + + let expected = crate::WebDriverBiDiWebSocketMaskKey::new([0x11, 0x22, 0x33, 0x44]); + let draw_count = Cell::new(0_usize); + let mut next = || { + draw_count.set(draw_count.get() + 1); + Some(expected) + }; + assert_eq!(next(), Some(expected)); + draw_count.set(0); + let deadline_result = next_pong_masking_key_before_deadline( + &mut next, + Duration::from_millis(500), + Duration::from_millis(500), + ); + assert_eq!( + format!("{deadline_result:?}"), + "Err(ExchangeDeadlineExceeded { exchange_timeout: 500ms })" + ); + assert_eq!(draw_count.get(), 0); + + let mut available = || Some(expected); + assert_eq!( + next_pong_masking_key_before_deadline( + &mut available, + Duration::from_millis(500), + Duration::from_millis(100), + ) + .ok(), + Some(expected) + ); + } + + #[test] + fn exchange_errors_preserve_typed_sources_and_protocol_shape() { + let frame = WebDriverBiDiLocateNodesExchangeError::Frame( + WebDriverBiDiWebSocketFrameError::InvalidFrameTimeout { + frame_timeout: Duration::ZERO, + maximum_timeout: MAX_WEBSOCKET_FRAME_TIMEOUT, + }, + ); + assert!(frame.source().is_some()); + assert!( + frame + .to_string() + .contains("WebSocket frame exchange failed") + ); + + let deadline = WebDriverBiDiLocateNodesExchangeError::ExchangeDeadlineExceeded { + exchange_timeout: Duration::from_millis(500), + }; + assert!(deadline.source().is_none()); + assert!(deadline.to_string().contains("end-to-end deadline")); + + let control_limit = WebDriverBiDiLocateNodesExchangeError::ControlFrameLimitExceeded { + maximum_control_frames: MAX_WEBDRIVER_BIDI_CONTROL_FRAMES_PER_EXCHANGE, + }; + assert!(control_limit.source().is_none()); + assert!( + control_limit + .to_string() + .contains("maximum 64 interleaved control frames") + ); + + let fragment_limit = WebDriverBiDiLocateNodesExchangeError::ResponseFragmentLimitExceeded { + maximum_fragments: MAX_WEBDRIVER_BIDI_RESPONSE_FRAGMENTS_PER_EXCHANGE, + }; + assert!(fragment_limit.source().is_none()); + assert!( + fragment_limit + .to_string() + .contains("maximum 256 response-message data fragments") + ); + + let missing_mask = WebDriverBiDiLocateNodesExchangeError::PongMaskingKeyUnavailable; + assert!(missing_mask.source().is_none()); + assert!( + missing_mask + .to_string() + .contains("fresh caller-supplied Pong masking key") + ); + + let reused_mask = WebDriverBiDiLocateNodesExchangeError::PongMaskingKeyReused; + assert!(reused_mask.source().is_none()); + assert!( + reused_mask + .to_string() + .contains("Pong masking key matching the immediately preceding client frame") + ); + + let shape = WebDriverBiDiLocateNodesExchangeError::UnexpectedResponseFrame { + fin: false, + opcode: 0x2, + }; + assert!(shape.source().is_none()); + assert!(shape.to_string().contains("fin=false, opcode=0x02")); + + let document = WebDriverBiDiLocateNodesExchangeError::ResponseDocument( + WebDriverBiDiResponseDocumentAdmissionError::InvalidUtf8, + ); + assert!(document.source().is_some()); + assert!(document.to_string().contains("raw-document admission")); + + let response = WebDriverBiDiLocateNodesExchangeError::LocateNodesResponse( + WebDriverBiDiLocateNodesResponseDocumentError::MissingResultNodes, + ); + assert!(response.source().is_some()); + assert!(response.to_string().contains("exact wire admission")); + } +} diff --git a/crates/originweave-network/src/webdriver_bidi_locate_nodes_exchange_transport_failure_tests.rs b/crates/originweave-network/src/webdriver_bidi_locate_nodes_exchange_transport_failure_tests.rs new file mode 100644 index 000000000..57041b676 --- /dev/null +++ b/crates/originweave-network/src/webdriver_bidi_locate_nodes_exchange_transport_failure_tests.rs @@ -0,0 +1,367 @@ +use std::{ + error::Error, + io::{self, Read, Write}, + net::{Shutdown, TcpListener, TcpStream}, + thread, + time::Duration, +}; + +use originweave_core::{ + BrowserAuthorityRegistry, BrowserContextDispatchTarget, BrowserContextOriginDispatchTarget, + BrowserContextOriginEpochDispatchTarget, BrowserProtocolAdapterDescriptor, + BrowserProtocolCapability, BrowserProtocolKind, Origin, OriginWeaveProtocolVersion, + ValidatedBrowserProtocolUse, WebDriverBiDiAccessibilityQuery, WebDriverBiDiLocateNodesCommand, + WebDriverBiDiWebSocketEndpoint, +}; + +use crate::{ + WebDriverBiDiLocateNodesExchangeError, WebDriverBiDiTcpConnection, + WebDriverBiDiTcpConnectionPlan, WebDriverBiDiWebSocketClientKey, + WebDriverBiDiWebSocketFrameError, WebDriverBiDiWebSocketHandshakePlan, + WebDriverBiDiWebSocketMaskKey, +}; + +const SESSION_ID: &str = "01234567-89ab-cdef-0123-456789abcdef"; +const RFC6455_SAMPLE_KEY: &str = "dGhlIHNhbXBsZSBub25jZQ=="; +const ORIGINWEAVE_PROTOCOL_VERSION: OriginWeaveProtocolVersion = + OriginWeaveProtocolVersion::new(0, 1); +const ADAPTER_VERSION: &str = "originweave-bidi-v1"; +const PROTOCOL_REVISION: &str = "webdriver-bidi-wd-2026-06-01"; +const BROWSER_REVISION: &str = "chromium-r1639810"; + +fn connect(endpoint: &str) -> Result> { + let admitted = WebDriverBiDiWebSocketEndpoint::new(endpoint)?; + let correlated = admitted.correlate_session_id(SESSION_ID)?; + let target = correlated.into_explicit_connect_target()?; + let plan = WebDriverBiDiTcpConnectionPlan::new(target, Duration::from_secs(1), 1)?; + Ok(plan.connect()?) +} + +fn read_opening_request(stream: &mut TcpStream) -> io::Result<()> { + stream.set_read_timeout(Some(Duration::from_secs(2)))?; + let mut request = Vec::new(); + let mut buffer = [0_u8; 512]; + while !request.ends_with(b"\r\n\r\n") { + let count = stream.read(&mut buffer)?; + if count == 0 { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "client opening request ended before the header terminator", + )); + } + request.extend_from_slice(&buffer[..count]); + } + Ok(()) +} + +fn read_masked_client_text_frame(stream: &mut TcpStream) -> io::Result<()> { + stream.set_read_timeout(Some(Duration::from_secs(2)))?; + let mut header = [0_u8; 2]; + stream.read_exact(&mut header)?; + if header[0] != 0x81 || header[1] & 0x80 == 0 { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "client command was not one final masked text frame", + )); + } + + let payload_length = match header[1] & 0x7f { + value @ 0..=125 => u64::from(value), + 126 => { + let mut extended = [0_u8; 2]; + stream.read_exact(&mut extended)?; + u64::from(u16::from_be_bytes(extended)) + } + _ => { + let mut extended = [0_u8; 8]; + stream.read_exact(&mut extended)?; + u64::from_be_bytes(extended) + } + }; + + let mut mask = [0_u8; 4]; + stream.read_exact(&mut mask)?; + let mut remaining = usize::try_from(payload_length).map_err(|_| { + io::Error::new( + io::ErrorKind::InvalidData, + "client command length cannot fit this test process", + ) + })?; + let mut buffer = [0_u8; 512]; + while remaining != 0 { + let chunk = remaining.min(buffer.len()); + stream.read_exact(&mut buffer[..chunk])?; + remaining -= chunk; + } + Ok(()) +} + +fn locate_nodes_command() -> Result> { + let query = WebDriverBiDiAccessibilityQuery::new(Some("button"), Some("Checkout"), 2)?; + Ok(WebDriverBiDiLocateNodesCommand::new( + 7, + "top-level-context", + &query, + )?) +} + +fn controlled_origin() -> Result> { + Origin::parse("https://app.example").map_err(|_error| "valid controlled fixture origin".into()) +} + +fn semantic_observation_proof() -> Result> { + let descriptor = BrowserProtocolAdapterDescriptor::new( + BrowserProtocolKind::WebDriverBiDi, + ORIGINWEAVE_PROTOCOL_VERSION, + ADAPTER_VERSION, + PROTOCOL_REVISION, + BROWSER_REVISION, + &[BrowserProtocolCapability::SemanticObservation], + )?; + Ok(descriptor.validate_use( + ORIGINWEAVE_PROTOCOL_VERSION, + BrowserProtocolKind::WebDriverBiDi, + ADAPTER_VERSION, + PROTOCOL_REVISION, + BROWSER_REVISION, + BrowserProtocolCapability::SemanticObservation, + )?) +} + +#[test] +fn locate_nodes_exchange_preserves_initial_command_write_failure() -> Result<(), Box> { + let listener = TcpListener::bind(("127.0.0.1", 0))?; + let local_addr = listener.local_addr()?; + let server = thread::spawn(move || -> io::Result<()> { + let (mut stream, _) = listener.accept()?; + read_opening_request(&mut stream)?; + stream.write_all( + b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=\r\n\r\n", + )?; + stream.set_read_timeout(Some(Duration::from_secs(2)))?; + let mut byte = [0_u8; 1]; + let count = stream.read(&mut byte)?; + if count != 0 { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "closed client write half still emitted a locateNodes command frame", + )); + } + Ok(()) + }); + + let endpoint = format!("ws://{local_addr}/session/{SESSION_ID}"); + let key = WebDriverBiDiWebSocketClientKey::new(RFC6455_SAMPLE_KEY)?; + let plan = WebDriverBiDiWebSocketHandshakePlan::new(connect(&endpoint)?, key)?; + let written = plan.write_opening_request(Duration::from_millis(500))?; + let established = written.read_opening_response(Duration::from_millis(500))?; + let shutdown_stream = established.try_clone_stream_for_test()?; + shutdown_stream.shutdown(Shutdown::Write)?; + let exchanged = established.exchange_locate_nodes( + locate_nodes_command()?, + WebDriverBiDiWebSocketMaskKey::new([0x11, 0x22, 0x33, 0x44]), + &mut || None, + Duration::from_millis(500), + ); + + let server_result = server + .join() + .map_err(|_| io::Error::other("initial write failure test server panicked"))?; + assert!(server_result.is_ok(), "{server_result:?}"); + + let error = exchanged.err().ok_or_else(|| { + io::Error::other("locateNodes exchange unexpectedly survived a closed client write half") + })?; + assert!( + matches!( + &error, + WebDriverBiDiLocateNodesExchangeError::Frame( + WebDriverBiDiWebSocketFrameError::FrameWriteFailed { .. } + ) + ), + "{error:?}" + ); + assert!(error.source().is_some()); + Ok(()) +} + +#[test] +fn locate_nodes_exchange_preserves_pong_write_failure_after_ping() -> Result<(), Box> { + let listener = TcpListener::bind(("127.0.0.1", 0))?; + let local_addr = listener.local_addr()?; + let server = thread::spawn(move || -> io::Result<()> { + let (mut stream, _) = listener.accept()?; + read_opening_request(&mut stream)?; + stream.write_all( + b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=\r\n\r\n", + )?; + read_masked_client_text_frame(&mut stream)?; + stream.write_all(&[0x89, 0])?; + thread::sleep(Duration::from_millis(250)); + Ok(()) + }); + + let endpoint = format!("ws://{local_addr}/session/{SESSION_ID}"); + let key = WebDriverBiDiWebSocketClientKey::new(RFC6455_SAMPLE_KEY)?; + let plan = WebDriverBiDiWebSocketHandshakePlan::new(connect(&endpoint)?, key)?; + let written = plan.write_opening_request(Duration::from_millis(500))?; + let established = written.read_opening_response(Duration::from_millis(500))?; + let shutdown_stream = established.try_clone_stream_for_test()?; + let pong_key = WebDriverBiDiWebSocketMaskKey::new([0x51, 0x52, 0x53, 0x54]); + let exchanged = established.exchange_locate_nodes( + locate_nodes_command()?, + WebDriverBiDiWebSocketMaskKey::new([0x11, 0x22, 0x33, 0x44]), + &mut || { + let shutdown = shutdown_stream.shutdown(Shutdown::Write); + assert!(shutdown.is_ok(), "{shutdown:?}"); + Some(pong_key) + }, + Duration::from_millis(500), + ); + + let server_result = server + .join() + .map_err(|_| io::Error::other("Pong write failure test server panicked"))?; + assert!(server_result.is_ok(), "{server_result:?}"); + + let error = exchanged.err().ok_or_else(|| { + io::Error::other("locateNodes exchange unexpectedly survived a closed client write half") + })?; + assert!( + matches!( + &error, + WebDriverBiDiLocateNodesExchangeError::Frame( + WebDriverBiDiWebSocketFrameError::FrameWriteFailed { .. } + ) + ), + "{error:?}" + ); + assert!(error.source().is_some()); + Ok(()) +} + +#[test] +fn zero_exchange_timeout_fails_at_exchange_boundary_before_frame_write() +-> Result<(), Box> { + let listener = TcpListener::bind(("127.0.0.1", 0))?; + let local_addr = listener.local_addr()?; + let server = thread::spawn(move || -> io::Result<()> { + let (mut stream, _) = listener.accept()?; + read_opening_request(&mut stream)?; + stream.write_all( + b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=\r\n\r\n", + )?; + stream.set_read_timeout(Some(Duration::from_secs(2)))?; + let mut byte = [0_u8; 1]; + let count = stream.read(&mut byte)?; + if count != 0 { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "zero-budget locateNodes exchange wrote a client frame", + )); + } + Ok(()) + }); + + let endpoint = format!("ws://{local_addr}/session/{SESSION_ID}"); + let key = WebDriverBiDiWebSocketClientKey::new(RFC6455_SAMPLE_KEY)?; + let plan = WebDriverBiDiWebSocketHandshakePlan::new(connect(&endpoint)?, key)?; + let written = plan.write_opening_request(Duration::from_millis(500))?; + let established = written.read_opening_response(Duration::from_millis(500))?; + let exchanged = established.exchange_locate_nodes( + locate_nodes_command()?, + WebDriverBiDiWebSocketMaskKey::new([0x11, 0x22, 0x33, 0x44]), + &mut || None, + Duration::ZERO, + ); + + let error = exchanged.err().ok_or_else(|| { + io::Error::other("zero-budget locateNodes exchange unexpectedly succeeded") + })?; + assert!( + matches!( + &error, + WebDriverBiDiLocateNodesExchangeError::ExchangeDeadlineExceeded { + exchange_timeout + } if exchange_timeout.is_zero() + ), + "{error:?}" + ); + + let server_result = server + .join() + .map_err(|_| io::Error::other("zero-budget exchange test server panicked"))?; + assert!(server_result.is_ok(), "{server_result:?}"); + Ok(()) +} + +#[test] +fn zero_exchange_timeout_binding_wrapper_executes_in_unit_crate_before_frame_write() +-> Result<(), Box> { + let listener = TcpListener::bind(("127.0.0.1", 0))?; + let local_addr = listener.local_addr()?; + let server = thread::spawn(move || -> io::Result<()> { + let (mut stream, _) = listener.accept()?; + read_opening_request(&mut stream)?; + stream.write_all( + b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=\r\n\r\n", + )?; + stream.set_read_timeout(Some(Duration::from_secs(2)))?; + let mut byte = [0_u8; 1]; + let count = stream.read(&mut byte)?; + if count != 0 { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "zero-budget binding wrapper wrote a locateNodes client frame", + )); + } + Ok(()) + }); + + let endpoint = format!("ws://{local_addr}/session/{SESSION_ID}"); + let key = WebDriverBiDiWebSocketClientKey::new(RFC6455_SAMPLE_KEY)?; + let plan = WebDriverBiDiWebSocketHandshakePlan::new(connect(&endpoint)?, key)?; + let written = plan.write_opening_request(Duration::from_millis(500))?; + let established = written.read_opening_response(Duration::from_millis(500))?; + + let mut registry = BrowserAuthorityRegistry::new(); + let origin = controlled_origin()?; + let session = registry.register_session(SESSION_ID)?; + let context = registry.register_context(session, "top-level-context")?; + let epoch = registry.bind_context_origin(session, context, &origin)?; + let target = BrowserContextOriginEpochDispatchTarget::new( + BrowserContextOriginDispatchTarget::new( + BrowserContextDispatchTarget::new(session, context), + &origin, + ), + epoch, + ); + + let exchanged = established.exchange_locate_nodes_and_bind_current_nodes( + locate_nodes_command()?, + WebDriverBiDiWebSocketMaskKey::new([0x11, 0x22, 0x33, 0x44]), + &mut || None, + Duration::ZERO, + (semantic_observation_proof()?, target), + &mut registry, + ); + + let error = exchanged.err().ok_or_else(|| { + io::Error::other("zero-budget locateNodes binding wrapper unexpectedly succeeded") + })?; + assert!( + matches!( + &error, + WebDriverBiDiLocateNodesExchangeError::ExchangeDeadlineExceeded { + exchange_timeout + } if exchange_timeout.is_zero() + ), + "{error:?}" + ); + + let server_result = server + .join() + .map_err(|_| io::Error::other("zero-budget binding-wrapper test server panicked"))?; + assert!(server_result.is_ok(), "{server_result:?}"); + Ok(()) +} diff --git a/crates/originweave-network/src/webdriver_bidi_locate_nodes_exchange_unit_coverage_tests.rs b/crates/originweave-network/src/webdriver_bidi_locate_nodes_exchange_unit_coverage_tests.rs new file mode 100644 index 000000000..bda5c2b98 --- /dev/null +++ b/crates/originweave-network/src/webdriver_bidi_locate_nodes_exchange_unit_coverage_tests.rs @@ -0,0 +1,259 @@ +use std::{ + error::Error, + io::{self, Read, Write}, + net::{TcpListener, TcpStream}, + thread, + time::Duration, +}; + +use originweave_core::{ + BrowserAuthorityRegistry, BrowserContextDispatchTarget, BrowserContextOriginDispatchTarget, + BrowserContextOriginEpochDispatchTarget, BrowserProtocolAdapterDescriptor, + BrowserProtocolCapability, BrowserProtocolKind, Origin, OriginWeaveProtocolVersion, + ValidatedBrowserProtocolUse, WebDriverBiDiAccessibilityQuery, WebDriverBiDiLocateNodesCommand, + WebDriverBiDiWebSocketEndpoint, +}; + +use crate::{ + WebDriverBiDiTcpConnectionPlan, WebDriverBiDiWebSocketClientKey, + WebDriverBiDiWebSocketHandshakePlan, WebDriverBiDiWebSocketMaskKey, +}; + +const SESSION_ID: &str = "01234567-89ab-cdef-0123-456789abcdef"; +const RFC6455_SAMPLE_KEY: &str = "dGhlIHNhbXBsZSBub25jZQ=="; +const RESPONSE_DOCUMENT: &str = + r#"{"type":"success","id":7,"result":{"nodes":[{"type":"node","sharedId":"shared-1"}]}}"#; +const ORIGINWEAVE_PROTOCOL_VERSION: OriginWeaveProtocolVersion = + OriginWeaveProtocolVersion::new(0, 1); +const ADAPTER_VERSION: &str = "originweave-bidi-v1"; +const PROTOCOL_REVISION: &str = "webdriver-bidi-wd-2026-06-01"; +const BROWSER_REVISION: &str = "chromium-r1639810"; + +fn read_opening_request(stream: &mut TcpStream) -> io::Result<()> { + stream.set_read_timeout(Some(Duration::from_secs(2)))?; + let mut request = Vec::new(); + let mut buffer = [0_u8; 512]; + while !request.ends_with(b"\r\n\r\n") { + let count = stream.read(&mut buffer)?; + if count == 0 { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "client opening request ended before the header terminator", + )); + } + request.extend_from_slice(&buffer[..count]); + } + Ok(()) +} + +fn read_client_text_frame(stream: &mut TcpStream) -> io::Result<()> { + stream.set_read_timeout(Some(Duration::from_secs(2)))?; + let mut header = [0_u8; 2]; + stream.read_exact(&mut header)?; + if header[0] != 0x81 || header[1] & 0x80 == 0 { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "expected one masked final client text frame", + )); + } + let payload_length = match header[1] & 0x7f { + value @ 0..=125 => usize::from(value), + 126 => { + let mut extended = [0_u8; 2]; + stream.read_exact(&mut extended)?; + usize::from(u16::from_be_bytes(extended)) + } + 127 => { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "test fixture rejects 64-bit client frame lengths", + )); + } + _ => unreachable!("7-bit WebSocket payload marker"), + }; + let mut mask = [0_u8; 4]; + stream.read_exact(&mut mask)?; + let mut payload = vec![0_u8; payload_length]; + stream.read_exact(&mut payload)?; + Ok(()) +} + +fn write_opening_response(stream: &mut TcpStream) -> io::Result<()> { + stream.write_all( + b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=\r\n\r\n", + ) +} + +fn write_short_server_frame( + stream: &mut TcpStream, + first_byte: u8, + payload: &[u8], +) -> io::Result<()> { + let payload_length = u8::try_from(payload.len()).map_err(|_| { + io::Error::new( + io::ErrorKind::InvalidData, + "unit coverage server frame exceeds short-frame limit", + ) + })?; + if payload_length > 125 { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "unit coverage server frame exceeds short-frame limit", + )); + } + stream.write_all(&[first_byte, payload_length])?; + stream.write_all(payload) +} + +fn locate_nodes_command() -> Result> { + let query = WebDriverBiDiAccessibilityQuery::new(Some("button"), Some("Checkout"), 2)?; + Ok(WebDriverBiDiLocateNodesCommand::new( + 7, + "top-level-context", + &query, + )?) +} + +fn controlled_origin() -> Result> { + Origin::parse("https://app.example").map_err(|_error| "valid controlled fixture origin".into()) +} + +fn semantic_observation_proof() -> Result> { + let descriptor = BrowserProtocolAdapterDescriptor::new( + BrowserProtocolKind::WebDriverBiDi, + ORIGINWEAVE_PROTOCOL_VERSION, + ADAPTER_VERSION, + PROTOCOL_REVISION, + BROWSER_REVISION, + &[BrowserProtocolCapability::SemanticObservation], + )?; + Ok(descriptor.validate_use( + ORIGINWEAVE_PROTOCOL_VERSION, + BrowserProtocolKind::WebDriverBiDi, + ADAPTER_VERSION, + PROTOCOL_REVISION, + BROWSER_REVISION, + BrowserProtocolCapability::SemanticObservation, + )?) +} + +fn establish_client( + local_addr: std::net::SocketAddr, +) -> Result> { + let endpoint = format!("ws://{local_addr}/session/{SESSION_ID}"); + let admitted = WebDriverBiDiWebSocketEndpoint::new(&endpoint)?; + let correlated = admitted.correlate_session_id(SESSION_ID)?; + let target = correlated.into_explicit_connect_target()?; + let connection = + WebDriverBiDiTcpConnectionPlan::new(target, Duration::from_secs(1), 1)?.connect()?; + let key = WebDriverBiDiWebSocketClientKey::new(RFC6455_SAMPLE_KEY)?; + let plan = WebDriverBiDiWebSocketHandshakePlan::new(connection, key)?; + let written = plan.write_opening_request(Duration::from_millis(500))?; + Ok(written.read_opening_response(Duration::from_millis(500))?) +} + +#[test] +fn binding_wrapper_success_path_executes_in_library_unit_crate() -> Result<(), Box> { + let listener = TcpListener::bind(("127.0.0.1", 0))?; + let local_addr = listener.local_addr()?; + let server = thread::spawn(move || -> io::Result<()> { + let (mut stream, _) = listener.accept()?; + read_opening_request(&mut stream)?; + write_opening_response(&mut stream)?; + read_client_text_frame(&mut stream)?; + write_short_server_frame(&mut stream, 0x81, RESPONSE_DOCUMENT.as_bytes()) + }); + + let established = establish_client(local_addr)?; + + let mut registry = BrowserAuthorityRegistry::new(); + let origin = controlled_origin()?; + let session = registry.register_session(SESSION_ID)?; + let context = registry.register_context(session, "top-level-context")?; + let epoch = registry.bind_context_origin(session, context, &origin)?; + let target = BrowserContextOriginEpochDispatchTarget::new( + BrowserContextOriginDispatchTarget::new( + BrowserContextDispatchTarget::new(session, context), + &origin, + ), + epoch, + ); + + let (_established, handles) = established.exchange_locate_nodes_and_bind_current_nodes( + locate_nodes_command()?, + WebDriverBiDiWebSocketMaskKey::new([0x11, 0x22, 0x33, 0x44]), + &mut || None, + Duration::from_millis(500), + (semantic_observation_proof()?, target), + &mut registry, + )?; + assert_eq!(handles.len(), 1); + assert_eq!(handles[0].origin(), &origin); + + server + .join() + .map_err(|_| io::Error::other("unit coverage test server panicked"))??; + Ok(()) +} + +#[test] +fn fragmented_response_executes_nonfinal_text_arm_in_library_unit_crate() +-> Result<(), Box> { + let listener = TcpListener::bind(("127.0.0.1", 0))?; + let local_addr = listener.local_addr()?; + let server = thread::spawn(move || -> io::Result<()> { + let (mut stream, _) = listener.accept()?; + read_opening_request(&mut stream)?; + write_opening_response(&mut stream)?; + read_client_text_frame(&mut stream)?; + let response = RESPONSE_DOCUMENT.as_bytes(); + write_short_server_frame(&mut stream, 0x01, &response[..1])?; + write_short_server_frame(&mut stream, 0x80, &response[1..]) + }); + + let established = establish_client(local_addr)?; + let (_established, result) = established.exchange_locate_nodes( + locate_nodes_command()?, + WebDriverBiDiWebSocketMaskKey::new([0x21, 0x22, 0x23, 0x24]), + &mut || None, + Duration::from_millis(500), + )?; + assert_eq!(result.nodes().len(), 1); + + server + .join() + .map_err(|_| io::Error::other("fragmented unit coverage test server panicked"))??; + Ok(()) +} + +#[test] +fn second_text_frame_during_fragmentation_executes_guard_denial_in_library_unit_crate() +-> Result<(), Box> { + let listener = TcpListener::bind(("127.0.0.1", 0))?; + let local_addr = listener.local_addr()?; + let server = thread::spawn(move || -> io::Result<()> { + let (mut stream, _) = listener.accept()?; + read_opening_request(&mut stream)?; + write_opening_response(&mut stream)?; + read_client_text_frame(&mut stream)?; + write_short_server_frame(&mut stream, 0x01, b"{")?; + write_short_server_frame(&mut stream, 0x81, b"x") + }); + + let established = establish_client(local_addr)?; + let exchange = established.exchange_locate_nodes( + locate_nodes_command()?, + WebDriverBiDiWebSocketMaskKey::new([0x31, 0x32, 0x33, 0x34]), + &mut || None, + Duration::from_millis(500), + ); + assert_eq!( + format!("{exchange:?}"), + "Err(UnexpectedResponseFrame { fin: true, opcode: 1 })" + ); + + server + .join() + .map_err(|_| io::Error::other("second-text unit coverage test server panicked"))??; + Ok(()) +} diff --git a/crates/originweave-network/src/webdriver_bidi_websocket_handshake.rs b/crates/originweave-network/src/webdriver_bidi_websocket_handshake.rs index 5c349ba19..37c37e25a 100644 --- a/crates/originweave-network/src/webdriver_bidi_websocket_handshake.rs +++ b/crates/originweave-network/src/webdriver_bidi_websocket_handshake.rs @@ -150,17 +150,6 @@ pub struct WebDriverBiDiWebSocketHandshakePlan { request: Vec, } -impl fmt::Debug for WebDriverBiDiWebSocketHandshakePlan { - fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { - formatter - .debug_struct("WebDriverBiDiWebSocketHandshakePlan") - .field("verified_peer", self.connection.verified_peer()) - .field("client_key", &"") - .field("request_byte_count", &self.request.len()) - .finish() - } -} - impl WebDriverBiDiWebSocketHandshakePlan { /// Bind one canonical opening request to an already-verified plain BiDi TCP connection. pub fn new( diff --git a/crates/originweave-network/src/webdriver_bidi_websocket_validated.rs b/crates/originweave-network/src/webdriver_bidi_websocket_validated.rs index e9039c89e..25f76a566 100644 --- a/crates/originweave-network/src/webdriver_bidi_websocket_validated.rs +++ b/crates/originweave-network/src/webdriver_bidi_websocket_validated.rs @@ -203,6 +203,12 @@ impl WebDriverBiDiWebSocketEstablished { self.raw.write_timeout() } + /// Clone the exact underlying stream for crate-internal fault-injection tests only. + #[cfg(test)] + pub(crate) fn try_clone_stream_for_test(&self) -> std::io::Result { + self.raw.stream.try_clone() + } + /// Write one unfragmented, masked UTF-8 text frame on this verified stream. /// /// The caller-supplied masking key must come from an approved strong randomness source. The diff --git a/crates/originweave-network/tests/webdriver_bidi_locate_nodes_binding_exchange_failure.rs b/crates/originweave-network/tests/webdriver_bidi_locate_nodes_binding_exchange_failure.rs new file mode 100644 index 000000000..c98a22521 --- /dev/null +++ b/crates/originweave-network/tests/webdriver_bidi_locate_nodes_binding_exchange_failure.rs @@ -0,0 +1,262 @@ +use std::{ + error::Error, + io::{self, Read, Write}, + net::{TcpListener, TcpStream}, + thread, + time::Duration, +}; + +use originweave_core::{ + BrowserAuthorityRegistry, BrowserContextDispatchTarget, BrowserContextOriginDispatchTarget, + BrowserContextOriginEpochDispatchTarget, BrowserProtocolAdapterDescriptor, + BrowserProtocolCapability, BrowserProtocolKind, Origin, OriginWeaveProtocolVersion, + ValidatedBrowserProtocolUse, WebDriverBiDiAccessibilityQuery, WebDriverBiDiLocateNodesCommand, + WebDriverBiDiWebSocketEndpoint, +}; +use originweave_network::{ + WebDriverBiDiLocateNodesExchangeError, WebDriverBiDiTcpConnectionPlan, + WebDriverBiDiWebSocketClientKey, WebDriverBiDiWebSocketFrameError, + WebDriverBiDiWebSocketHandshakePlan, WebDriverBiDiWebSocketMaskKey, +}; + +const SESSION_ID: &str = "01234567-89ab-cdef-0123-456789abcdef"; +const RFC6455_SAMPLE_KEY: &str = "dGhlIHNhbXBsZSBub25jZQ=="; +const REUSED_MASK_REASON: &str = + "client masking key was reused for consecutive frames on this established WebSocket"; +const ORIGINWEAVE_PROTOCOL_VERSION: OriginWeaveProtocolVersion = + OriginWeaveProtocolVersion::new(0, 1); +const ADAPTER_VERSION: &str = "originweave-bidi-v1"; +const PROTOCOL_REVISION: &str = "webdriver-bidi-wd-2026-06-01"; +const BROWSER_REVISION: &str = "chromium-r1639810"; + +type UnexpectedBinaryServer = thread::JoinHandle>>; +type EstablishedWithUnexpectedBinaryServer = ( + originweave_network::WebDriverBiDiWebSocketEstablished, + UnexpectedBinaryServer, +); + +fn connect( + endpoint: &str, +) -> Result> { + let admitted = WebDriverBiDiWebSocketEndpoint::new(endpoint)?; + let correlated = admitted.correlate_session_id(SESSION_ID)?; + let target = correlated.into_explicit_connect_target()?; + let plan = WebDriverBiDiTcpConnectionPlan::new(target, Duration::from_secs(1), 1)?; + Ok(plan.connect()?) +} + +fn read_opening_request(stream: &mut TcpStream) -> io::Result> { + stream.set_read_timeout(Some(Duration::from_secs(2)))?; + let mut request = Vec::new(); + let mut buffer = [0_u8; 512]; + while !request.ends_with(b"\r\n\r\n") { + let count = stream.read(&mut buffer)?; + if count == 0 { + break; + } + request.extend_from_slice(&buffer[..count]); + } + Ok(request) +} + +fn read_client_text_frame(stream: &mut TcpStream) -> io::Result> { + stream.set_read_timeout(Some(Duration::from_secs(2)))?; + let mut header = [0_u8; 2]; + stream.read_exact(&mut header)?; + if header[0] != 0x81 || header[1] & 0x80 == 0 { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "expected one masked final client text frame", + )); + } + let payload_length = match header[1] & 0x7f { + value @ 0..=125 => usize::from(value), + 126 => { + let mut extended = [0_u8; 2]; + stream.read_exact(&mut extended)?; + usize::from(u16::from_be_bytes(extended)) + } + 127 => { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "test fixture rejects 64-bit client frame lengths", + )); + } + _ => unreachable!("7-bit WebSocket payload marker"), + }; + let mut mask = [0_u8; 4]; + stream.read_exact(&mut mask)?; + let mut payload = vec![0_u8; payload_length]; + stream.read_exact(&mut payload)?; + for (index, byte) in payload.iter_mut().enumerate() { + *byte ^= mask[index % mask.len()]; + } + Ok(payload) +} + +fn require_peer_closed_without_another_frame(stream: &mut TcpStream) -> io::Result<()> { + stream.set_read_timeout(Some(Duration::from_secs(2)))?; + let mut byte = [0_u8; 1]; + match stream.read(&mut byte) { + Ok(0) => Ok(()), + Ok(_) => Err(io::Error::new( + io::ErrorKind::InvalidData, + "client emitted a second frame after fail-closed locateNodes command rejection", + )), + Err(error) => Err(io::Error::new( + error.kind(), + format!("client did not close after fail-closed locateNodes rejection: {error}"), + )), + } +} + +fn establish_with_unexpected_binary_response() +-> Result> { + let listener = TcpListener::bind(("127.0.0.1", 0))?; + let local_addr = listener.local_addr()?; + let server = thread::spawn(move || -> io::Result> { + let (mut stream, _) = listener.accept()?; + let request = read_opening_request(&mut stream)?; + if !request.ends_with(b"\r\n\r\n") { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "client opening request was incomplete", + )); + } + stream.write_all( + b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=\r\n\r\n", + )?; + let command = read_client_text_frame(&mut stream)?; + stream.write_all(&[0x82, 0x00])?; + Ok(command) + }); + + let endpoint = format!("ws://{local_addr}/session/{SESSION_ID}"); + let key = WebDriverBiDiWebSocketClientKey::new(RFC6455_SAMPLE_KEY)?; + let plan = WebDriverBiDiWebSocketHandshakePlan::new(connect(&endpoint)?, key)?; + let written = plan.write_opening_request(Duration::from_millis(500))?; + let established = written.read_opening_response(Duration::from_millis(500))?; + Ok((established, server)) +} + +fn locate_nodes_command() -> Result> { + let query = WebDriverBiDiAccessibilityQuery::new(Some("button"), Some("Checkout"), 2)?; + Ok(WebDriverBiDiLocateNodesCommand::new( + 7, + "top-level-context", + &query, + )?) +} + +fn controlled_origin() -> Result> { + Origin::parse("https://app.example").map_err(|_error| "valid controlled fixture origin".into()) +} + +fn semantic_observation_proof() -> Result> { + let descriptor = BrowserProtocolAdapterDescriptor::new( + BrowserProtocolKind::WebDriverBiDi, + ORIGINWEAVE_PROTOCOL_VERSION, + ADAPTER_VERSION, + PROTOCOL_REVISION, + BROWSER_REVISION, + &[BrowserProtocolCapability::SemanticObservation], + )?; + Ok(descriptor.validate_use( + ORIGINWEAVE_PROTOCOL_VERSION, + BrowserProtocolKind::WebDriverBiDi, + ADAPTER_VERSION, + PROTOCOL_REVISION, + BROWSER_REVISION, + BrowserProtocolCapability::SemanticObservation, + )?) +} + +#[test] +fn locate_nodes_command_mask_reuse_fails_before_second_wire_write() -> Result<(), Box> { + let listener = TcpListener::bind(("127.0.0.1", 0))?; + let local_addr = listener.local_addr()?; + let server = thread::spawn(move || -> io::Result> { + let (mut stream, _) = listener.accept()?; + let request = read_opening_request(&mut stream)?; + if !request.ends_with(b"\r\n\r\n") { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "client opening request was incomplete", + )); + } + stream.write_all( + b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=\r\n\r\n", + )?; + let first_frame = read_client_text_frame(&mut stream)?; + require_peer_closed_without_another_frame(&mut stream)?; + Ok(first_frame) + }); + + let endpoint = format!("ws://{local_addr}/session/{SESSION_ID}"); + let key = WebDriverBiDiWebSocketClientKey::new(RFC6455_SAMPLE_KEY)?; + let plan = WebDriverBiDiWebSocketHandshakePlan::new(connect(&endpoint)?, key)?; + let written = plan.write_opening_request(Duration::from_millis(500))?; + let established = written.read_opening_response(Duration::from_millis(500))?; + let reused_mask = WebDriverBiDiWebSocketMaskKey::new([0x11, 0x22, 0x33, 0x44]); + let established = + established.write_text_frame("coverage-primer", reused_mask, Duration::from_millis(500))?; + + let error = established.exchange_locate_nodes( + locate_nodes_command()?, + reused_mask, + &mut || None, + Duration::from_millis(500), + ); + assert!(matches!( + error, + Err(WebDriverBiDiLocateNodesExchangeError::Frame( + WebDriverBiDiWebSocketFrameError::MalformedFrame { + reason: REUSED_MASK_REASON, + } + )) + )); + + let first_frame = server.join().map_err(|_| "test server panicked")??; + assert_eq!(first_frame, b"coverage-primer"); + Ok(()) +} + +#[test] +fn live_binding_wrapper_fails_closed_when_wire_exchange_fails_before_binding() +-> Result<(), Box> { + let (established, server) = establish_with_unexpected_binary_response()?; + let mut registry = BrowserAuthorityRegistry::new(); + let origin = controlled_origin()?; + let session = registry.register_session(SESSION_ID)?; + let context = registry.register_context(session, "top-level-context")?; + let epoch = registry.bind_context_origin(session, context, &origin)?; + let target = BrowserContextOriginEpochDispatchTarget::new( + BrowserContextOriginDispatchTarget::new( + BrowserContextDispatchTarget::new(session, context), + &origin, + ), + epoch, + ); + + let error = established.exchange_locate_nodes_and_bind_current_nodes( + locate_nodes_command()?, + WebDriverBiDiWebSocketMaskKey::new([0x11, 0x22, 0x33, 0x44]), + &mut || None, + Duration::from_millis(500), + (semantic_observation_proof()?, target), + &mut registry, + ); + + assert!(matches!( + error, + Err( + WebDriverBiDiLocateNodesExchangeError::UnexpectedResponseFrame { + fin: true, + opcode: 0x2, + } + ) + )); + let command = server.join().map_err(|_| "test server panicked")??; + assert_eq!(command, locate_nodes_command()?.as_json().as_bytes()); + Ok(()) +} diff --git a/crates/originweave-network/tests/webdriver_bidi_locate_nodes_current_authority.rs b/crates/originweave-network/tests/webdriver_bidi_locate_nodes_current_authority.rs new file mode 100644 index 000000000..2c91dd6bc --- /dev/null +++ b/crates/originweave-network/tests/webdriver_bidi_locate_nodes_current_authority.rs @@ -0,0 +1,264 @@ +use std::{ + error::Error, + io::{self, Read, Write}, + net::{SocketAddr, TcpListener, TcpStream}, + thread, + time::Duration, +}; + +use originweave_core::{ + BrowserAuthorityRegistry, BrowserContextDispatchTarget, BrowserContextOriginDispatchTarget, + BrowserContextOriginEpochDispatchTarget, BrowserProtocolAdapterDescriptor, + BrowserProtocolCapability, BrowserProtocolKind, BrowserSessionId, BrowsingContextId, Origin, + OriginWeaveProtocolVersion, ValidatedBrowserProtocolUse, WebDriverBiDiAccessibilityQuery, + WebDriverBiDiLocateNodesAdmissionError, WebDriverBiDiLocateNodesCommand, + WebDriverBiDiLocateNodesResponseDocumentError, WebDriverBiDiWebSocketEndpoint, +}; +use originweave_network::{ + WebDriverBiDiLocateNodesExchangeError, WebDriverBiDiTcpConnectionPlan, + WebDriverBiDiWebSocketClientKey, WebDriverBiDiWebSocketEstablished, + WebDriverBiDiWebSocketHandshakePlan, WebDriverBiDiWebSocketMaskKey, +}; + +const SESSION_ID: &str = "01234567-89ab-cdef-0123-456789abcdef"; +const RFC6455_SAMPLE_KEY: &str = "dGhlIHNhbXBsZSBub25jZQ=="; +const RESPONSE_DOCUMENT: &str = + r#"{"type":"success","id":7,"result":{"nodes":[{"type":"node","sharedId":"shared-1"}]}}"#; +const ORIGINWEAVE_PROTOCOL_VERSION: OriginWeaveProtocolVersion = + OriginWeaveProtocolVersion::new(0, 1); +const ADAPTER_VERSION: &str = "originweave-bidi-v1"; +const PROTOCOL_REVISION: &str = "webdriver-bidi-wd-2026-06-01"; +const BROWSER_REVISION: &str = "chromium-r1639810"; + +type ServerHandle = thread::JoinHandle>>; +type EstablishedFixture = + Result<(SocketAddr, WebDriverBiDiWebSocketEstablished, ServerHandle), Box>; + +fn connect( + endpoint: &str, +) -> Result> { + let admitted = WebDriverBiDiWebSocketEndpoint::new(endpoint)?; + let correlated = admitted.correlate_session_id(SESSION_ID)?; + let target = correlated.into_explicit_connect_target()?; + let plan = WebDriverBiDiTcpConnectionPlan::new(target, Duration::from_secs(1), 1)?; + Ok(plan.connect()?) +} + +fn read_opening_request(stream: &mut TcpStream) -> io::Result> { + stream.set_read_timeout(Some(Duration::from_secs(2)))?; + let mut request = Vec::new(); + let mut buffer = [0_u8; 512]; + while !request.ends_with(b"\r\n\r\n") { + let count = stream.read(&mut buffer)?; + if count == 0 { + break; + } + request.extend_from_slice(&buffer[..count]); + } + Ok(request) +} + +fn read_client_text_frame(stream: &mut TcpStream) -> io::Result> { + stream.set_read_timeout(Some(Duration::from_secs(2)))?; + let mut header = [0_u8; 2]; + stream.read_exact(&mut header)?; + if header[0] != 0x81 || header[1] & 0x80 == 0 { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "expected one masked final client text frame", + )); + } + let payload_length = match header[1] & 0x7f { + value @ 0..=125 => usize::from(value), + 126 => { + let mut extended = [0_u8; 2]; + stream.read_exact(&mut extended)?; + usize::from(u16::from_be_bytes(extended)) + } + 127 => { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "test fixture rejects 64-bit client frame lengths", + )); + } + _ => unreachable!("7-bit WebSocket payload marker"), + }; + let mut mask = [0_u8; 4]; + stream.read_exact(&mut mask)?; + let mut payload = vec![0_u8; payload_length]; + stream.read_exact(&mut payload)?; + for (index, byte) in payload.iter_mut().enumerate() { + *byte ^= mask[index % mask.len()]; + } + Ok(payload) +} + +fn establish() -> EstablishedFixture { + let listener = TcpListener::bind(("127.0.0.1", 0))?; + let local_addr = listener.local_addr()?; + let server = thread::spawn(move || -> io::Result> { + let (mut stream, _) = listener.accept()?; + let request = read_opening_request(&mut stream)?; + if !request.ends_with(b"\r\n\r\n") { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "client opening request was incomplete", + )); + } + stream.write_all( + b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=\r\n\r\n", + )?; + let command = read_client_text_frame(&mut stream)?; + let response = RESPONSE_DOCUMENT.as_bytes(); + let response_length = u8::try_from(response.len()).map_err(|_| { + io::Error::new( + io::ErrorKind::InvalidData, + "test response exceeds short frame", + ) + })?; + if response_length > 125 { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "test response exceeds short frame", + )); + } + stream.write_all(&[0x81, response_length])?; + stream.write_all(response)?; + Ok(command) + }); + + let endpoint = format!("ws://{local_addr}/session/{SESSION_ID}"); + let key = WebDriverBiDiWebSocketClientKey::new(RFC6455_SAMPLE_KEY)?; + let plan = WebDriverBiDiWebSocketHandshakePlan::new(connect(&endpoint)?, key)?; + let written = plan.write_opening_request(Duration::from_millis(500))?; + let established = written.read_opening_response(Duration::from_millis(500))?; + Ok((local_addr, established, server)) +} + +fn locate_nodes_command() -> Result> { + let query = WebDriverBiDiAccessibilityQuery::new(Some("button"), Some("Checkout"), 2)?; + Ok(WebDriverBiDiLocateNodesCommand::new( + 7, + "top-level-context", + &query, + )?) +} + +fn controlled_origin() -> Result> { + Origin::parse("https://app.example").map_err(|_error| "valid controlled fixture origin".into()) +} + +fn current_target<'a>( + registry: &mut BrowserAuthorityRegistry, + origin: &'a Origin, +) -> Result< + ( + BrowserContextOriginEpochDispatchTarget<'a>, + BrowserSessionId, + BrowsingContextId, + ), + Box, +> { + let session = registry.register_session(SESSION_ID)?; + let context = registry.register_context(session, "top-level-context")?; + let epoch = registry.bind_context_origin(session, context, origin)?; + Ok(( + BrowserContextOriginEpochDispatchTarget::new( + BrowserContextOriginDispatchTarget::new( + BrowserContextDispatchTarget::new(session, context), + origin, + ), + epoch, + ), + session, + context, + )) +} + +fn semantic_observation_proof() -> Result> { + let descriptor = BrowserProtocolAdapterDescriptor::new( + BrowserProtocolKind::WebDriverBiDi, + ORIGINWEAVE_PROTOCOL_VERSION, + ADAPTER_VERSION, + PROTOCOL_REVISION, + BROWSER_REVISION, + &[BrowserProtocolCapability::SemanticObservation], + )?; + Ok(descriptor.validate_use( + ORIGINWEAVE_PROTOCOL_VERSION, + BrowserProtocolKind::WebDriverBiDi, + ADAPTER_VERSION, + PROTOCOL_REVISION, + BROWSER_REVISION, + BrowserProtocolCapability::SemanticObservation, + )?) +} + +#[test] +fn live_websocket_locate_nodes_exchange_binds_wire_nodes_to_current_authority() +-> Result<(), Box> { + let (local_addr, established, server) = establish()?; + let mut registry = BrowserAuthorityRegistry::new(); + let origin = controlled_origin()?; + let (target, _session, _context) = current_target(&mut registry, &origin)?; + let expected_epoch = target.expected_epoch(); + + let (established, handles) = established.exchange_locate_nodes_and_bind_current_nodes( + locate_nodes_command()?, + WebDriverBiDiWebSocketMaskKey::new([0x11, 0x22, 0x33, 0x44]), + &mut || None, + Duration::from_millis(500), + (semantic_observation_proof()?, target), + &mut registry, + )?; + + assert_eq!(handles.len(), 1); + assert_eq!(handles[0].origin(), &origin); + assert_eq!(handles[0].document_epoch(), expected_epoch); + assert_eq!( + established + .transport_evidence() + .verified_peer() + .socket_addr(), + local_addr + ); + let command = server.join().map_err(|_| "test server panicked")??; + assert_eq!(command, locate_nodes_command()?.as_json().as_bytes()); + Ok(()) +} + +#[test] +fn live_exchange_fails_closed_when_document_epoch_changed_before_wire_node_binding() +-> Result<(), Box> { + let (_local_addr, established, server) = establish()?; + let mut registry = BrowserAuthorityRegistry::new(); + let origin = controlled_origin()?; + let (target, session, context) = current_target(&mut registry, &origin)?; + let stale_epoch = target.expected_epoch(); + let current_epoch = registry.advance_document(context)?; + registry.bind_context_origin(session, context, &origin)?; + + let error = established.exchange_locate_nodes_and_bind_current_nodes( + locate_nodes_command()?, + WebDriverBiDiWebSocketMaskKey::new([0x11, 0x22, 0x33, 0x44]), + &mut || None, + Duration::from_millis(500), + (semantic_observation_proof()?, target), + &mut registry, + ); + + assert!(matches!( + error, + Err(WebDriverBiDiLocateNodesExchangeError::LocateNodesResponse( + WebDriverBiDiLocateNodesResponseDocumentError::NodeBinding( + WebDriverBiDiLocateNodesAdmissionError::DocumentEpochMismatch { + expected, + current, + }, + ), + )) if expected == stale_epoch && current == current_epoch + )); + let command = server.join().map_err(|_| "test server panicked")??; + assert_eq!(command, locate_nodes_command()?.as_json().as_bytes()); + Ok(()) +} diff --git a/crates/originweave-network/tests/webdriver_bidi_locate_nodes_exchange_error_contract.rs b/crates/originweave-network/tests/webdriver_bidi_locate_nodes_exchange_error_contract.rs new file mode 100644 index 000000000..f2ed6acfd --- /dev/null +++ b/crates/originweave-network/tests/webdriver_bidi_locate_nodes_exchange_error_contract.rs @@ -0,0 +1,44 @@ +use std::{error::Error as _, time::Duration}; + +use originweave_core::{ + WebDriverBiDiLocateNodesResponseDocumentError, WebDriverBiDiResponseDocumentAdmissionError, +}; +use originweave_network::{ + MAX_WEBSOCKET_FRAME_TIMEOUT, WebDriverBiDiLocateNodesExchangeError, + WebDriverBiDiWebSocketFrameError, +}; + +#[test] +fn downstream_callers_observe_exact_exchange_error_sources() { + let frame = WebDriverBiDiLocateNodesExchangeError::Frame( + WebDriverBiDiWebSocketFrameError::InvalidFrameTimeout { + frame_timeout: Duration::ZERO, + maximum_timeout: MAX_WEBSOCKET_FRAME_TIMEOUT, + }, + ); + assert!(frame.source().is_some()); + + let document = WebDriverBiDiLocateNodesExchangeError::ResponseDocument( + WebDriverBiDiResponseDocumentAdmissionError::InvalidUtf8, + ); + assert!(document.source().is_some()); + + let response = WebDriverBiDiLocateNodesExchangeError::LocateNodesResponse( + WebDriverBiDiLocateNodesResponseDocumentError::MissingResultNodes, + ); + assert!(response.source().is_some()); + + let source_free_errors = [ + WebDriverBiDiLocateNodesExchangeError::ExchangeDeadlineExceeded { + exchange_timeout: Duration::from_millis(500), + }, + WebDriverBiDiLocateNodesExchangeError::PongMaskingKeyUnavailable, + WebDriverBiDiLocateNodesExchangeError::UnexpectedResponseFrame { + fin: false, + opcode: 0x2, + }, + ]; + for error in source_free_errors { + assert!(error.source().is_none(), "{error:?}"); + } +} diff --git a/crates/originweave-network/tests/webdriver_bidi_locate_nodes_exchange_long_budget.rs b/crates/originweave-network/tests/webdriver_bidi_locate_nodes_exchange_long_budget.rs new file mode 100644 index 000000000..84c09eed6 --- /dev/null +++ b/crates/originweave-network/tests/webdriver_bidi_locate_nodes_exchange_long_budget.rs @@ -0,0 +1,160 @@ +use std::{ + io::{self, Read, Write}, + net::TcpListener, + thread, + time::Duration, +}; + +use originweave_core::{ + WebDriverBiDiAccessibilityQuery, WebDriverBiDiLocateNodesCommand, + WebDriverBiDiWebSocketEndpoint, +}; +use originweave_network::{ + WebDriverBiDiTcpConnectionPlan, WebDriverBiDiWebSocketClientKey, + WebDriverBiDiWebSocketHandshakePlan, WebDriverBiDiWebSocketMaskKey, +}; + +const SESSION_ID: &str = "01234567-89ab-cdef-0123-456789abcdef"; +const RFC6455_SAMPLE_KEY: &str = "dGhlIHNhbXBsZSBub25jZQ=="; +const RESPONSE_DOCUMENT: &str = + r#"{"type":"success","id":7,"result":{"nodes":[{"type":"node","sharedId":"shared-1"}]}}"#; + +fn read_client_text_frame(stream: &mut impl Read) -> io::Result<()> { + let mut header = [0_u8; 2]; + stream.read_exact(&mut header)?; + assert_eq!(header[0], 0x81); + assert_ne!(header[1] & 0x80, 0); + + let payload_marker = header[1] & 0x7f; + let payload_length = if payload_marker <= 125 { + usize::from(payload_marker) + } else if payload_marker == 126 { + let mut extended = [0_u8; 2]; + stream.read_exact(&mut extended)?; + usize::from(u16::from_be_bytes(extended)) + } else { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "test command unexpectedly used a 64-bit WebSocket payload length", + )); + }; + + let mut mask = [0_u8; 4]; + stream.read_exact(&mut mask)?; + let mut payload = vec![0_u8; payload_length]; + stream.read_exact(&mut payload)?; + Ok(()) +} + +#[test] +fn exchange_budget_above_per_frame_ceiling_remains_a_valid_end_to_end_budget() { + let listener = TcpListener::bind(("127.0.0.1", 0)); + assert!(listener.is_ok(), "{listener:?}"); + let Ok(listener) = listener else { + return; + }; + let address = listener.local_addr(); + assert!(address.is_ok(), "{address:?}"); + let Ok(address) = address else { + return; + }; + let server = thread::spawn(move || -> io::Result<()> { + let (mut stream, _) = listener.accept()?; + stream.set_read_timeout(Some(Duration::from_secs(1)))?; + + let mut opening = Vec::new(); + let mut byte = [0_u8; 1]; + while !opening.ends_with(b"\r\n\r\n") { + stream.read_exact(&mut byte)?; + opening.push(byte[0]); + } + stream.write_all( + b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=\r\n\r\n", + )?; + + read_client_text_frame(&mut stream)?; + + let response = RESPONSE_DOCUMENT.as_bytes(); + let response_length = u8::try_from(response.len()).map_err(|_| { + io::Error::new( + io::ErrorKind::InvalidData, + "test response unexpectedly exceeded short-frame length", + ) + })?; + stream.write_all(&[0x81, response_length])?; + stream.write_all(response)?; + Ok(()) + }); + + let endpoint = + WebDriverBiDiWebSocketEndpoint::new(&format!("ws://{address}/session/{SESSION_ID}")); + assert!(endpoint.is_ok(), "{endpoint:?}"); + let Ok(endpoint) = endpoint else { + return; + }; + let correlated = endpoint.correlate_session_id(SESSION_ID); + assert!(correlated.is_ok(), "{correlated:?}"); + let Ok(correlated) = correlated else { + return; + }; + let target = correlated.into_explicit_connect_target(); + assert!(target.is_ok(), "{target:?}"); + let Ok(target) = target else { + return; + }; + let connection_plan = WebDriverBiDiTcpConnectionPlan::new(target, Duration::from_secs(1), 1); + assert!(connection_plan.is_ok(), "{connection_plan:?}"); + let Ok(connection_plan) = connection_plan else { + return; + }; + let connection = connection_plan.connect(); + assert!(connection.is_ok(), "{connection:?}"); + let Ok(connection) = connection else { + return; + }; + let key = WebDriverBiDiWebSocketClientKey::new(RFC6455_SAMPLE_KEY); + assert!(key.is_ok(), "{key:?}"); + let Ok(key) = key else { + return; + }; + let handshake = WebDriverBiDiWebSocketHandshakePlan::new(connection, key); + assert!(handshake.is_ok(), "{handshake:?}"); + let Ok(handshake) = handshake else { + return; + }; + let opening = handshake.write_opening_request(Duration::from_millis(500)); + assert!(opening.is_ok(), "{opening:?}"); + let Ok(opening) = opening else { + return; + }; + let established = opening.read_opening_response(Duration::from_millis(500)); + assert!(established.is_ok(), "{established:?}"); + let Ok(established) = established else { + return; + }; + + let query = WebDriverBiDiAccessibilityQuery::new(Some("button"), Some("Checkout"), 2); + assert!(query.is_ok(), "{query:?}"); + let Ok(query) = query else { + return; + }; + let command = WebDriverBiDiLocateNodesCommand::new(7, "top-level-context", &query); + assert!(command.is_ok(), "{command:?}"); + let Ok(command) = command else { + return; + }; + + let exchanged = established.exchange_locate_nodes( + command, + WebDriverBiDiWebSocketMaskKey::new([0x11, 0x22, 0x33, 0x44]), + &mut || None, + Duration::from_secs(6), + ); + assert!(exchanged.is_ok(), "{exchanged:?}"); + + let server_result = server.join(); + assert!(server_result.is_ok(), "{server_result:?}"); + if let Ok(server_io) = server_result { + assert!(server_io.is_ok(), "{server_io:?}"); + } +} diff --git a/crates/originweave-network/tests/webdriver_bidi_locate_nodes_fragmentation.rs b/crates/originweave-network/tests/webdriver_bidi_locate_nodes_fragmentation.rs new file mode 100644 index 000000000..b1c98b61a --- /dev/null +++ b/crates/originweave-network/tests/webdriver_bidi_locate_nodes_fragmentation.rs @@ -0,0 +1,255 @@ +use std::{ + error::Error, + io::{self, Read, Write}, + net::{TcpListener, TcpStream}, + thread, + time::Duration, +}; + +use originweave_core::{ + WebDriverBiDiAccessibilityQuery, WebDriverBiDiLocateNodesCommand, + WebDriverBiDiWebSocketEndpoint, +}; +use originweave_network::{ + WebDriverBiDiTcpConnectionPlan, WebDriverBiDiWebSocketClientKey, + WebDriverBiDiWebSocketHandshakePlan, WebDriverBiDiWebSocketMaskKey, +}; + +const SESSION_ID: &str = "01234567-89ab-cdef-0123-456789abcdef"; +const RFC6455_SAMPLE_KEY: &str = "dGhlIHNhbXBsZSBub25jZQ=="; +const RESPONSE_DOCUMENT: &str = + r#"{"type":"success","id":7,"result":{"nodes":[{"type":"node","sharedId":"shared-1"}]}}"#; +const PING_PAYLOAD: &[u8] = b"ping-between-fragments"; + +type EstablishedFragmentServer = ( + originweave_network::WebDriverBiDiWebSocketEstablished, + thread::JoinHandle>, +); + +fn connect( + endpoint: &str, +) -> Result> { + let admitted = WebDriverBiDiWebSocketEndpoint::new(endpoint)?; + let correlated = admitted.correlate_session_id(SESSION_ID)?; + let target = correlated.into_explicit_connect_target()?; + let plan = WebDriverBiDiTcpConnectionPlan::new(target, Duration::from_secs(1), 1)?; + Ok(plan.connect()?) +} + +fn read_opening_request(stream: &mut TcpStream) -> io::Result<()> { + stream.set_read_timeout(Some(Duration::from_secs(2)))?; + let mut request = Vec::new(); + let mut buffer = [0_u8; 512]; + while !request.ends_with(b"\r\n\r\n") { + let count = stream.read(&mut buffer)?; + if count == 0 { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "client opening request ended before the header terminator", + )); + } + request.extend_from_slice(&buffer[..count]); + } + Ok(()) +} + +fn read_masked_client_frame( + stream: &mut TcpStream, + expected_first_byte: u8, +) -> io::Result> { + stream.set_read_timeout(Some(Duration::from_secs(2)))?; + let mut header = [0_u8; 2]; + stream.read_exact(&mut header)?; + if header[0] != expected_first_byte || header[1] & 0x80 == 0 { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "client frame did not use the expected final masked opcode", + )); + } + let payload_length = match header[1] & 0x7f { + value @ 0..=125 => usize::from(value), + 126 => { + let mut extended = [0_u8; 2]; + stream.read_exact(&mut extended)?; + usize::from(u16::from_be_bytes(extended)) + } + 127 => { + let mut extended = [0_u8; 8]; + stream.read_exact(&mut extended)?; + let payload_length = u64::from_be_bytes(extended); + usize::try_from(payload_length).map_err(|_| { + io::Error::new( + io::ErrorKind::InvalidData, + "client frame length did not fit usize", + ) + })? + } + marker => { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!("unexpected WebSocket length marker {marker}"), + )); + } + }; + let mut mask = [0_u8; 4]; + stream.read_exact(&mut mask)?; + let mut payload = vec![0_u8; payload_length]; + stream.read_exact(&mut payload)?; + for (index, byte) in payload.iter_mut().enumerate() { + *byte ^= mask[index % mask.len()]; + } + Ok(payload) +} + +fn read_masked_client_text_frame(stream: &mut TcpStream) -> io::Result> { + read_masked_client_frame(stream, 0x81) +} + +fn locate_nodes_command() -> Result> { + let query = WebDriverBiDiAccessibilityQuery::new(Some("button"), Some("Checkout"), 2)?; + Ok(WebDriverBiDiLocateNodesCommand::new( + 7, + "top-level-context", + &query, + )?) +} + +fn response_fragments() -> io::Result<(Vec, Vec)> { + let response = RESPONSE_DOCUMENT.as_bytes(); + let split = response.len() / 2; + let first = &response[..split]; + let second = &response[split..]; + let first_length = u8::try_from(first.len()).map_err(|_| { + io::Error::new( + io::ErrorKind::InvalidData, + "first response fragment exceeded one-byte test length", + ) + })?; + let second_length = u8::try_from(second.len()).map_err(|_| { + io::Error::new( + io::ErrorKind::InvalidData, + "second response fragment exceeded one-byte test length", + ) + })?; + + let mut first_frame = Vec::with_capacity(first.len() + 2); + first_frame.extend_from_slice(&[0x01, first_length]); + first_frame.extend_from_slice(first); + let mut second_frame = Vec::with_capacity(second.len() + 2); + second_frame.extend_from_slice(&[0x80, second_length]); + second_frame.extend_from_slice(second); + Ok((first_frame, second_frame)) +} + +fn establish_with_fragmented_response() -> Result> { + let listener = TcpListener::bind(("127.0.0.1", 0))?; + let local_addr = listener.local_addr()?; + let server = thread::spawn(move || -> io::Result<()> { + let (mut stream, _) = listener.accept()?; + read_opening_request(&mut stream)?; + stream.write_all( + b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=\r\n\r\n", + )?; + let _command = read_masked_client_text_frame(&mut stream)?; + let (first_frame, second_frame) = response_fragments()?; + stream.write_all(&first_frame)?; + stream.write_all(&second_frame) + }); + + let endpoint = format!("ws://{local_addr}/session/{SESSION_ID}"); + let key = WebDriverBiDiWebSocketClientKey::new(RFC6455_SAMPLE_KEY)?; + let plan = WebDriverBiDiWebSocketHandshakePlan::new(connect(&endpoint)?, key)?; + let written = plan.write_opening_request(Duration::from_millis(500))?; + let established = written.read_opening_response(Duration::from_millis(500))?; + Ok((established, server)) +} + +fn establish_with_ping_between_fragments() -> Result> { + let listener = TcpListener::bind(("127.0.0.1", 0))?; + let local_addr = listener.local_addr()?; + let server = thread::spawn(move || -> io::Result<()> { + let (mut stream, _) = listener.accept()?; + read_opening_request(&mut stream)?; + stream.write_all( + b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=\r\n\r\n", + )?; + let _command = read_masked_client_text_frame(&mut stream)?; + let (first_frame, second_frame) = response_fragments()?; + stream.write_all(&first_frame)?; + let ping_length = u8::try_from(PING_PAYLOAD.len()).map_err(|_| { + io::Error::new( + io::ErrorKind::InvalidData, + "test Ping payload exceeded 125 bytes", + ) + })?; + stream.write_all(&[0x89, ping_length])?; + stream.write_all(PING_PAYLOAD)?; + + let pong_payload = read_masked_client_frame(&mut stream, 0x8a)?; + if pong_payload != PING_PAYLOAD { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "client Pong did not echo the interleaved Ping payload", + )); + } + stream.write_all(&second_frame) + }); + + let endpoint = format!("ws://{local_addr}/session/{SESSION_ID}"); + let key = WebDriverBiDiWebSocketClientKey::new(RFC6455_SAMPLE_KEY)?; + let plan = WebDriverBiDiWebSocketHandshakePlan::new(connect(&endpoint)?, key)?; + let written = plan.write_opening_request(Duration::from_millis(500))?; + let established = written.read_opening_response(Duration::from_millis(500))?; + Ok((established, server)) +} + +fn assert_successful_exchange( + exchanged: Result< + ( + originweave_network::WebDriverBiDiWebSocketEstablished, + originweave_core::ValidatedWebDriverBiDiLocateNodesResult, + ), + originweave_network::WebDriverBiDiLocateNodesExchangeError, + >, + server: thread::JoinHandle>, +) -> Result<(), Box> { + let server_result = server + .join() + .map_err(|_| io::Error::other("fragmentation regression server panicked"))?; + assert!(server_result.is_ok(), "{server_result:?}"); + assert!(exchanged.is_ok(), "{exchanged:?}"); + + let (_, result) = exchanged?; + assert_eq!(result.command_id(), 7); + assert_eq!(result.nodes().len(), 1); + assert_eq!(result.nodes()[0].shared_id(), "shared-1"); + Ok(()) +} + +#[test] +fn locate_nodes_exchange_reassembles_fragmented_text_response() -> Result<(), Box> { + let (established, server) = establish_with_fragmented_response()?; + let command = locate_nodes_command()?; + let mut no_pong_keys = || None; + let exchanged = established.exchange_locate_nodes( + command, + WebDriverBiDiWebSocketMaskKey::new([0x11, 0x22, 0x33, 0x44]), + &mut no_pong_keys, + Duration::from_millis(500), + ); + assert_successful_exchange(exchanged, server) +} + +#[test] +fn locate_nodes_exchange_handles_ping_between_response_fragments() -> Result<(), Box> { + let (established, server) = establish_with_ping_between_fragments()?; + let command = locate_nodes_command()?; + let mut pong_keys = || Some(WebDriverBiDiWebSocketMaskKey::new([0x55, 0x66, 0x77, 0x88])); + let exchanged = established.exchange_locate_nodes( + command, + WebDriverBiDiWebSocketMaskKey::new([0x11, 0x22, 0x33, 0x44]), + &mut pong_keys, + Duration::from_millis(500), + ); + assert_successful_exchange(exchanged, server) +} diff --git a/crates/originweave-network/tests/webdriver_bidi_locate_nodes_fragmentation_boundaries.rs b/crates/originweave-network/tests/webdriver_bidi_locate_nodes_fragmentation_boundaries.rs new file mode 100644 index 000000000..eedd13fd8 --- /dev/null +++ b/crates/originweave-network/tests/webdriver_bidi_locate_nodes_fragmentation_boundaries.rs @@ -0,0 +1,305 @@ +use std::{ + error::Error, + io::{self, Read, Write}, + net::{TcpListener, TcpStream}, + thread, + time::Duration, +}; + +use originweave_core::{ + MAX_WEBDRIVER_BIDI_RESPONSE_DOCUMENT_BYTES, WebDriverBiDiAccessibilityQuery, + WebDriverBiDiLocateNodesCommand, WebDriverBiDiResponseDocumentAdmissionError, + WebDriverBiDiWebSocketEndpoint, +}; +use originweave_network::{ + MAX_WEBDRIVER_BIDI_RESPONSE_FRAGMENTS_PER_EXCHANGE, WebDriverBiDiLocateNodesExchangeError, + WebDriverBiDiTcpConnectionPlan, WebDriverBiDiWebSocketClientKey, + WebDriverBiDiWebSocketEstablished, WebDriverBiDiWebSocketHandshakePlan, + WebDriverBiDiWebSocketMaskKey, +}; + +const SESSION_ID: &str = "01234567-89ab-cdef-0123-456789abcdef"; +const RFC6455_SAMPLE_KEY: &str = "dGhlIHNhbXBsZSBub25jZQ=="; +const RESPONSE_DOCUMENT: &str = + r#"{"type":"success","id":7,"result":{"nodes":[{"type":"node","sharedId":"shared-1"}]}}"#; + +type EstablishedFrameServer = ( + WebDriverBiDiWebSocketEstablished, + thread::JoinHandle>, +); + +fn read_opening_request(stream: &mut TcpStream) -> io::Result<()> { + stream.set_read_timeout(Some(Duration::from_secs(2)))?; + let mut request = Vec::new(); + let mut buffer = [0_u8; 512]; + while !request.ends_with(b"\r\n\r\n") { + let count = stream.read(&mut buffer)?; + if count == 0 { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "client opening request ended before the header terminator", + )); + } + request.extend_from_slice(&buffer[..count]); + } + Ok(()) +} + +fn read_masked_client_text_frame(stream: &mut TcpStream) -> io::Result<()> { + stream.set_read_timeout(Some(Duration::from_secs(2)))?; + let mut header = [0_u8; 2]; + stream.read_exact(&mut header)?; + if header[0] != 0x81 || header[1] & 0x80 == 0 { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "expected one final masked client text frame", + )); + } + + let payload_length = match header[1] & 0x7f { + value @ 0..=125 => usize::from(value), + 126 => { + let mut extended = [0_u8; 2]; + stream.read_exact(&mut extended)?; + usize::from(u16::from_be_bytes(extended)) + } + 127 => { + let mut extended = [0_u8; 8]; + stream.read_exact(&mut extended)?; + usize::try_from(u64::from_be_bytes(extended)).map_err(|_| { + io::Error::new( + io::ErrorKind::InvalidData, + "client frame payload length did not fit usize", + ) + })? + } + marker => { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!("invalid client frame payload-length marker {marker}"), + )); + } + }; + + let mut mask = [0_u8; 4]; + stream.read_exact(&mut mask)?; + let mut payload = vec![0_u8; payload_length]; + stream.read_exact(&mut payload)?; + Ok(()) +} + +fn server_frame(fin: bool, opcode: u8, payload: &[u8]) -> io::Result> { + let first_byte = if fin { 0x80 | opcode } else { opcode }; + let mut frame = Vec::with_capacity(payload.len() + 10); + frame.push(first_byte); + + if payload.len() <= 125 { + frame.push(u8::try_from(payload.len()).map_err(|_| { + io::Error::new( + io::ErrorKind::InvalidData, + "short server frame length did not fit u8", + ) + })?); + } else if payload.len() <= usize::from(u16::MAX) { + frame.push(126); + frame.extend_from_slice( + &u16::try_from(payload.len()) + .map_err(|_| { + io::Error::new( + io::ErrorKind::InvalidData, + "medium server frame length did not fit u16", + ) + })? + .to_be_bytes(), + ); + } else { + frame.push(127); + frame.extend_from_slice( + &u64::try_from(payload.len()) + .map_err(|_| { + io::Error::new( + io::ErrorKind::InvalidData, + "large server frame length did not fit u64", + ) + })? + .to_be_bytes(), + ); + } + + frame.extend_from_slice(payload); + Ok(frame) +} + +fn establish_with_frames(frames: Vec>) -> Result> { + let listener = TcpListener::bind(("127.0.0.1", 0))?; + let local_addr = listener.local_addr()?; + let server = thread::spawn(move || -> io::Result<()> { + let (mut stream, _) = listener.accept()?; + read_opening_request(&mut stream)?; + stream.write_all( + b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=\r\n\r\n", + )?; + read_masked_client_text_frame(&mut stream)?; + for frame in frames { + stream.write_all(&frame)?; + } + Ok(()) + }); + + let endpoint = format!("ws://{local_addr}/session/{SESSION_ID}"); + let admitted = WebDriverBiDiWebSocketEndpoint::new(&endpoint)?; + let correlated = admitted.correlate_session_id(SESSION_ID)?; + let target = correlated.into_explicit_connect_target()?; + let connection = + WebDriverBiDiTcpConnectionPlan::new(target, Duration::from_secs(1), 1)?.connect()?; + let key = WebDriverBiDiWebSocketClientKey::new(RFC6455_SAMPLE_KEY)?; + let plan = WebDriverBiDiWebSocketHandshakePlan::new(connection, key)?; + let written = plan.write_opening_request(Duration::from_millis(500))?; + let established = written.read_opening_response(Duration::from_millis(500))?; + Ok((established, server)) +} + +fn locate_nodes_command() -> Result> { + let query = WebDriverBiDiAccessibilityQuery::new(Some("button"), Some("Checkout"), 2)?; + Ok(WebDriverBiDiLocateNodesCommand::new( + 7, + "top-level-context", + &query, + )?) +} + +fn join_server(server: thread::JoinHandle>) -> Result<(), Box> { + server + .join() + .map_err(|_| io::Error::other("fragmentation boundary server panicked"))??; + Ok(()) +} + +#[test] +fn three_fragment_response_reassembles() -> Result<(), Box> { + let response = RESPONSE_DOCUMENT.as_bytes(); + let first_end = response.len() / 3; + let second_end = first_end * 2; + let frames = vec![ + server_frame(false, 0x1, &response[..first_end])?, + server_frame(false, 0x0, &response[first_end..second_end])?, + server_frame(true, 0x0, &response[second_end..])?, + ]; + let (established, server) = establish_with_frames(frames)?; + let mut no_pong_keys = || None; + let exchange = established.exchange_locate_nodes( + locate_nodes_command()?, + WebDriverBiDiWebSocketMaskKey::new([0x11, 0x22, 0x33, 0x44]), + &mut no_pong_keys, + Duration::from_millis(500), + ); + + join_server(server)?; + let (_, result) = exchange?; + assert_eq!(result.command_id(), 7); + assert_eq!(result.nodes().len(), 1); + Ok(()) +} + +#[test] +fn oversized_initial_fragment_fails_closed() -> Result<(), Box> { + let payload = vec![b'{'; MAX_WEBDRIVER_BIDI_RESPONSE_DOCUMENT_BYTES + 1]; + let frames = vec![server_frame(false, 0x1, &payload)?]; + let (established, server) = establish_with_frames(frames)?; + let mut no_pong_keys = || None; + let exchange = established.exchange_locate_nodes( + locate_nodes_command()?, + WebDriverBiDiWebSocketMaskKey::new([0x11, 0x22, 0x33, 0x44]), + &mut no_pong_keys, + Duration::from_millis(500), + ); + + join_server(server)?; + assert!(matches!( + exchange, + Err(WebDriverBiDiLocateNodesExchangeError::ResponseDocument( + WebDriverBiDiResponseDocumentAdmissionError::DocumentTooLarge + )) + )); + Ok(()) +} + +#[test] +fn fragmented_response_over_budget_fails_closed() -> Result<(), Box> { + let first_payload = vec![b'{'; MAX_WEBDRIVER_BIDI_RESPONSE_DOCUMENT_BYTES]; + let frames = vec![ + server_frame(false, 0x1, &first_payload)?, + server_frame(true, 0x0, b"}")?, + ]; + let (established, server) = establish_with_frames(frames)?; + let mut no_pong_keys = || None; + let exchange = established.exchange_locate_nodes( + locate_nodes_command()?, + WebDriverBiDiWebSocketMaskKey::new([0x11, 0x22, 0x33, 0x44]), + &mut no_pong_keys, + Duration::from_millis(500), + ); + + join_server(server)?; + assert!(matches!( + exchange, + Err(WebDriverBiDiLocateNodesExchangeError::ResponseDocument( + WebDriverBiDiResponseDocumentAdmissionError::DocumentTooLarge + )) + )); + Ok(()) +} + +#[test] +fn fragmented_invalid_utf8_fails_closed() -> Result<(), Box> { + let frames = vec![ + server_frame(false, 0x1, b"{")?, + server_frame(true, 0x0, &[0xff, b'}'])?, + ]; + let (established, server) = establish_with_frames(frames)?; + let mut no_pong_keys = || None; + let exchange = established.exchange_locate_nodes( + locate_nodes_command()?, + WebDriverBiDiWebSocketMaskKey::new([0x11, 0x22, 0x33, 0x44]), + &mut no_pong_keys, + Duration::from_millis(500), + ); + + join_server(server)?; + assert!(matches!( + exchange, + Err(WebDriverBiDiLocateNodesExchangeError::ResponseDocument( + WebDriverBiDiResponseDocumentAdmissionError::InvalidUtf8 + )) + )); + Ok(()) +} + +#[test] +fn excessive_zero_length_fragments_fail_closed() -> Result<(), Box> { + let mut frames = Vec::with_capacity(MAX_WEBDRIVER_BIDI_RESPONSE_FRAGMENTS_PER_EXCHANGE + 1); + frames.push(server_frame(false, 0x1, b"")?); + for _ in 1..=MAX_WEBDRIVER_BIDI_RESPONSE_FRAGMENTS_PER_EXCHANGE { + frames.push(server_frame(false, 0x0, b"")?); + } + + let (established, server) = establish_with_frames(frames)?; + let mut no_pong_keys = || None; + let exchange = established.exchange_locate_nodes( + locate_nodes_command()?, + WebDriverBiDiWebSocketMaskKey::new([0x11, 0x22, 0x33, 0x44]), + &mut no_pong_keys, + Duration::from_secs(2), + ); + + join_server(server)?; + assert!(matches!( + exchange, + Err( + WebDriverBiDiLocateNodesExchangeError::ResponseFragmentLimitExceeded { + maximum_fragments: MAX_WEBDRIVER_BIDI_RESPONSE_FRAGMENTS_PER_EXCHANGE, + } + ) + )); + Ok(()) +} diff --git a/crates/originweave-network/tests/webdriver_bidi_locate_nodes_fragmented_second_text.rs b/crates/originweave-network/tests/webdriver_bidi_locate_nodes_fragmented_second_text.rs new file mode 100644 index 000000000..15df8cad0 --- /dev/null +++ b/crates/originweave-network/tests/webdriver_bidi_locate_nodes_fragmented_second_text.rs @@ -0,0 +1,141 @@ +use std::{ + error::Error, + io::{self, Read, Write}, + net::{TcpListener, TcpStream}, + thread, + time::Duration, +}; + +use originweave_core::{ + WebDriverBiDiAccessibilityQuery, WebDriverBiDiLocateNodesCommand, + WebDriverBiDiWebSocketEndpoint, +}; +use originweave_network::{ + WebDriverBiDiLocateNodesExchangeError, WebDriverBiDiTcpConnectionPlan, + WebDriverBiDiWebSocketClientKey, WebDriverBiDiWebSocketHandshakePlan, + WebDriverBiDiWebSocketMaskKey, +}; + +const SESSION_ID: &str = "01234567-89ab-cdef-0123-456789abcdef"; +const RFC6455_SAMPLE_KEY: &str = "dGhlIHNhbXBsZSBub25jZQ=="; + +fn read_opening_request(stream: &mut TcpStream) -> io::Result<()> { + stream.set_read_timeout(Some(Duration::from_secs(2)))?; + let mut request = Vec::new(); + let mut buffer = [0_u8; 512]; + while !request.ends_with(b"\r\n\r\n") { + let count = stream.read(&mut buffer)?; + if count == 0 { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "client opening request ended before the header terminator", + )); + } + request.extend_from_slice(&buffer[..count]); + } + Ok(()) +} + +fn read_masked_client_text_frame(stream: &mut TcpStream) -> io::Result<()> { + stream.set_read_timeout(Some(Duration::from_secs(2)))?; + let mut header = [0_u8; 2]; + stream.read_exact(&mut header)?; + if header[0] != 0x81 || header[1] & 0x80 == 0 { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "expected one final masked client text frame", + )); + } + + let payload_length = match header[1] & 0x7f { + value @ 0..=125 => usize::from(value), + 126 => { + let mut extended = [0_u8; 2]; + stream.read_exact(&mut extended)?; + usize::from(u16::from_be_bytes(extended)) + } + 127 => { + let mut extended = [0_u8; 8]; + stream.read_exact(&mut extended)?; + usize::try_from(u64::from_be_bytes(extended)).map_err(|_| { + io::Error::new( + io::ErrorKind::InvalidData, + "client frame payload length did not fit usize", + ) + })? + } + marker => { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!("invalid client frame payload-length marker {marker}"), + )); + } + }; + + let mut mask = [0_u8; 4]; + stream.read_exact(&mut mask)?; + let mut payload = vec![0_u8; payload_length]; + stream.read_exact(&mut payload)?; + Ok(()) +} + +fn locate_nodes_command() -> Result> { + let query = WebDriverBiDiAccessibilityQuery::new(Some("button"), Some("Checkout"), 2)?; + Ok(WebDriverBiDiLocateNodesCommand::new( + 7, + "top-level-context", + &query, + )?) +} + +#[test] +fn second_text_message_during_fragmentation_fails_closed() -> Result<(), Box> { + let listener = TcpListener::bind(("127.0.0.1", 0))?; + let local_addr = listener.local_addr()?; + let server = thread::spawn(move || -> io::Result<()> { + let (mut stream, _) = listener.accept()?; + read_opening_request(&mut stream)?; + stream.write_all( + b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=\r\n\r\n", + )?; + read_masked_client_text_frame(&mut stream)?; + stream.write_all(&[0x01, 0x01, b'{'])?; + stream.write_all(&[0x81, 0x01, b'x']) + }); + + let endpoint = format!("ws://{local_addr}/session/{SESSION_ID}"); + let admitted = WebDriverBiDiWebSocketEndpoint::new(&endpoint)?; + let correlated = admitted.correlate_session_id(SESSION_ID)?; + let target = correlated.into_explicit_connect_target()?; + let connection = + WebDriverBiDiTcpConnectionPlan::new(target, Duration::from_secs(1), 1)?.connect()?; + let key = WebDriverBiDiWebSocketClientKey::new(RFC6455_SAMPLE_KEY)?; + let plan = WebDriverBiDiWebSocketHandshakePlan::new(connection, key)?; + let written = plan.write_opening_request(Duration::from_millis(500))?; + let established = written.read_opening_response(Duration::from_millis(500))?; + + let exchange = established.exchange_locate_nodes( + locate_nodes_command()?, + WebDriverBiDiWebSocketMaskKey::new([0x31, 0x32, 0x33, 0x34]), + &mut || None, + Duration::from_millis(500), + ); + + server + .join() + .map_err(|_| io::Error::other("fragmented second-text server panicked"))??; + assert_eq!( + format!("{exchange:?}"), + "Err(UnexpectedResponseFrame { fin: true, opcode: 1 })" + ); + assert!(matches!( + exchange, + Err( + WebDriverBiDiLocateNodesExchangeError::UnexpectedResponseFrame { + fin: true, + opcode: 0x1 + } + ) + )); + Ok(()) +} diff --git a/crates/originweave-network/tests/webdriver_bidi_locate_nodes_masking_key_freshness.rs b/crates/originweave-network/tests/webdriver_bidi_locate_nodes_masking_key_freshness.rs new file mode 100644 index 000000000..cb533c86e --- /dev/null +++ b/crates/originweave-network/tests/webdriver_bidi_locate_nodes_masking_key_freshness.rs @@ -0,0 +1,236 @@ +use std::{ + error::Error, + io::{self, Read, Write}, + net::{TcpListener, TcpStream}, + thread, + time::Duration, +}; + +use originweave_core::{ + WebDriverBiDiAccessibilityQuery, WebDriverBiDiLocateNodesCommand, + WebDriverBiDiWebSocketEndpoint, +}; +use originweave_network::{ + WebDriverBiDiTcpConnectionPlan, WebDriverBiDiWebSocketClientKey, + WebDriverBiDiWebSocketHandshakePlan, WebDriverBiDiWebSocketMaskKey, +}; + +const SESSION_ID: &str = "01234567-89ab-cdef-0123-456789abcdef"; +const RFC6455_SAMPLE_KEY: &str = "dGhlIHNhbXBsZSBub25jZQ=="; +const RESPONSE_DOCUMENT: &str = + r#"{"type":"success","id":7,"result":{"nodes":[{"type":"node","sharedId":"shared-1"}]}}"#; +const PING_PAYLOAD: &[u8] = b"fresh-mask"; +const COMMAND_MASK: WebDriverBiDiWebSocketMaskKey = + WebDriverBiDiWebSocketMaskKey::new([0x11, 0x22, 0x33, 0x44]); +const PONG_MASK: WebDriverBiDiWebSocketMaskKey = + WebDriverBiDiWebSocketMaskKey::new([0x51, 0x52, 0x53, 0x54]); +const REUSED_MASK_ERROR: &str = "WebDriver BiDi locateNodes exchange refused a Pong masking key matching the immediately preceding client frame"; + +type EstablishedServer = ( + originweave_network::WebDriverBiDiWebSocketEstablished, + thread::JoinHandle>, +); + +fn connect( + endpoint: &str, +) -> Result> { + let admitted = WebDriverBiDiWebSocketEndpoint::new(endpoint)?; + let correlated = admitted.correlate_session_id(SESSION_ID)?; + let target = correlated.into_explicit_connect_target()?; + let plan = WebDriverBiDiTcpConnectionPlan::new(target, Duration::from_secs(1), 1)?; + Ok(plan.connect()?) +} + +fn read_opening_request(stream: &mut TcpStream) -> io::Result<()> { + stream.set_read_timeout(Some(Duration::from_secs(2)))?; + let mut request = Vec::new(); + let mut buffer = [0_u8; 512]; + while !request.ends_with(b"\r\n\r\n") { + let count = stream.read(&mut buffer)?; + if count == 0 { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "client opening request ended before the header terminator", + )); + } + request.extend_from_slice(&buffer[..count]); + } + Ok(()) +} + +fn read_masked_client_frame(stream: &mut TcpStream, expected_first_byte: u8) -> io::Result<()> { + stream.set_read_timeout(Some(Duration::from_secs(2)))?; + let mut header = [0_u8; 2]; + stream.read_exact(&mut header)?; + if header[0] != expected_first_byte || header[1] & 0x80 == 0 { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "client frame did not have the expected final opcode and masking bit", + )); + } + let payload_length = match header[1] & 0x7f { + value @ 0..=125 => usize::from(value), + 126 => { + let mut extended = [0_u8; 2]; + stream.read_exact(&mut extended)?; + usize::from(u16::from_be_bytes(extended)) + } + 127 => { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "test fixture does not admit 64-bit client frame lengths", + )); + } + _ => unreachable!("7-bit WebSocket payload marker"), + }; + let mut mask = [0_u8; 4]; + stream.read_exact(&mut mask)?; + let mut payload = vec![0_u8; payload_length]; + stream.read_exact(&mut payload)?; + Ok(()) +} + +fn write_ping(stream: &mut TcpStream) -> io::Result<()> { + let payload_length = u8::try_from(PING_PAYLOAD.len()).map_err(|_| { + io::Error::new( + io::ErrorKind::InvalidData, + "test Ping payload exceeded one-byte length", + ) + })?; + stream.write_all(&[0x89, payload_length])?; + stream.write_all(PING_PAYLOAD) +} + +fn write_response(stream: &mut TcpStream) -> io::Result<()> { + let payload_length = u8::try_from(RESPONSE_DOCUMENT.len()).map_err(|_| { + io::Error::new( + io::ErrorKind::InvalidData, + "test response exceeded one-byte length", + ) + })?; + stream.write_all(&[0x81, payload_length])?; + stream.write_all(RESPONSE_DOCUMENT.as_bytes()) +} + +fn locate_nodes_command() -> Result> { + let query = WebDriverBiDiAccessibilityQuery::new(Some("button"), Some("Checkout"), 2)?; + Ok(WebDriverBiDiLocateNodesCommand::new( + 7, + "top-level-context", + &query, + )?) +} + +fn establish_with_ping_sequence( + read_first_pong: bool, +) -> Result> { + let listener = TcpListener::bind(("127.0.0.1", 0))?; + let local_addr = listener.local_addr()?; + let server = thread::spawn(move || -> io::Result<()> { + let (mut stream, _) = listener.accept()?; + read_opening_request(&mut stream)?; + stream.write_all( + b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=\r\n\r\n", + )?; + read_masked_client_frame(&mut stream, 0x81)?; + write_ping(&mut stream)?; + if read_first_pong { + read_masked_client_frame(&mut stream, 0x8a)?; + write_ping(&mut stream)?; + } + thread::sleep(Duration::from_millis(150)); + Ok(()) + }); + + let endpoint = format!("ws://{local_addr}/session/{SESSION_ID}"); + let client_key = WebDriverBiDiWebSocketClientKey::new(RFC6455_SAMPLE_KEY)?; + let plan = WebDriverBiDiWebSocketHandshakePlan::new(connect(&endpoint)?, client_key)?; + let written = plan.write_opening_request(Duration::from_millis(500))?; + let established = written.read_opening_response(Duration::from_millis(500))?; + Ok((established, server)) +} + +fn join_server(server: thread::JoinHandle>) -> Result<(), Box> { + let result = server + .join() + .map_err(|_| io::Error::other("masking-key freshness test server panicked"))?; + Ok(result?) +} + +#[test] +fn locate_nodes_exchange_rejects_pong_mask_reused_from_command_frame() -> Result<(), Box> +{ + let (established, server) = establish_with_ping_sequence(false)?; + let exchanged = established.exchange_locate_nodes( + locate_nodes_command()?, + COMMAND_MASK, + &mut || Some(COMMAND_MASK), + Duration::from_millis(500), + ); + + let error = exchanged.err().ok_or_else(|| { + io::Error::other("reusing the command masking key for Pong unexpectedly succeeded") + })?; + assert_eq!(error.to_string(), REUSED_MASK_ERROR); + join_server(server) +} + +#[test] +fn locate_nodes_exchange_rejects_pong_mask_reused_from_prior_pong() -> Result<(), Box> { + let (established, server) = establish_with_ping_sequence(true)?; + let mut keys = [PONG_MASK, PONG_MASK].into_iter(); + let exchanged = established.exchange_locate_nodes( + locate_nodes_command()?, + COMMAND_MASK, + &mut || keys.next(), + Duration::from_millis(500), + ); + + let error = exchanged.err().ok_or_else(|| { + io::Error::other("reusing a prior Pong masking key unexpectedly succeeded") + })?; + assert_eq!(error.to_string(), REUSED_MASK_ERROR); + join_server(server) +} + +#[test] +fn locate_nodes_exchange_allows_non_adjacent_random_mask_collision() -> Result<(), Box> { + let listener = TcpListener::bind(("127.0.0.1", 0))?; + let local_addr = listener.local_addr()?; + let server = thread::spawn(move || -> io::Result<()> { + let (mut stream, _) = listener.accept()?; + read_opening_request(&mut stream)?; + stream.write_all( + b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=\r\n\r\n", + )?; + read_masked_client_frame(&mut stream, 0x81)?; + write_ping(&mut stream)?; + read_masked_client_frame(&mut stream, 0x8a)?; + write_ping(&mut stream)?; + read_masked_client_frame(&mut stream, 0x8a)?; + write_response(&mut stream) + }); + + let endpoint = format!("ws://{local_addr}/session/{SESSION_ID}"); + let client_key = WebDriverBiDiWebSocketClientKey::new(RFC6455_SAMPLE_KEY)?; + let plan = WebDriverBiDiWebSocketHandshakePlan::new(connect(&endpoint)?, client_key)?; + let written = plan.write_opening_request(Duration::from_millis(500))?; + let established = written.read_opening_response(Duration::from_millis(500))?; + let mut keys = [PONG_MASK, COMMAND_MASK].into_iter(); + let exchanged = established.exchange_locate_nodes( + locate_nodes_command()?, + COMMAND_MASK, + &mut || keys.next(), + Duration::from_millis(500), + ); + + let server_result = server + .join() + .map_err(|_| io::Error::other("non-adjacent masking-key collision server panicked"))?; + assert!(server_result.is_ok(), "{server_result:?}"); + let (_, result) = exchanged?; + assert_eq!(result.command_id(), 7); + assert_eq!(result.nodes().len(), 1); + assert_eq!(result.nodes()[0].shared_id(), "shared-1"); + Ok(()) +} diff --git a/crates/originweave-network/tests/webdriver_bidi_locate_nodes_ping_interleave.rs b/crates/originweave-network/tests/webdriver_bidi_locate_nodes_ping_interleave.rs new file mode 100644 index 000000000..3e26f30b7 --- /dev/null +++ b/crates/originweave-network/tests/webdriver_bidi_locate_nodes_ping_interleave.rs @@ -0,0 +1,319 @@ +use std::{ + error::Error, + io::{self, Read, Write}, + net::{TcpListener, TcpStream}, + thread, + time::Duration, +}; + +use originweave_core::{ + WebDriverBiDiAccessibilityQuery, WebDriverBiDiLocateNodesCommand, + WebDriverBiDiWebSocketEndpoint, +}; +use originweave_network::{ + MAX_WEBDRIVER_BIDI_CONTROL_FRAMES_PER_EXCHANGE, WebDriverBiDiLocateNodesExchangeError, + WebDriverBiDiTcpConnectionPlan, WebDriverBiDiWebSocketClientKey, + WebDriverBiDiWebSocketHandshakePlan, WebDriverBiDiWebSocketMaskKey, +}; + +const SESSION_ID: &str = "01234567-89ab-cdef-0123-456789abcdef"; +const RFC6455_SAMPLE_KEY: &str = "dGhlIHNhbXBsZSBub25jZQ=="; +const RESPONSE_DOCUMENT: &str = + r#"{"type":"success","id":7,"result":{"nodes":[{"type":"node","sharedId":"shared-1"}]}}"#; +const PING_PAYLOAD: &[u8] = b"keepalive"; + +type EstablishedPingServer = ( + originweave_network::WebDriverBiDiWebSocketEstablished, + thread::JoinHandle>, +); + +fn connect( + endpoint: &str, +) -> Result> { + let admitted = WebDriverBiDiWebSocketEndpoint::new(endpoint)?; + let correlated = admitted.correlate_session_id(SESSION_ID)?; + let target = correlated.into_explicit_connect_target()?; + let plan = WebDriverBiDiTcpConnectionPlan::new(target, Duration::from_secs(1), 1)?; + Ok(plan.connect()?) +} + +fn read_opening_request(stream: &mut TcpStream) -> io::Result<()> { + stream.set_read_timeout(Some(Duration::from_secs(2)))?; + let mut request = Vec::new(); + let mut buffer = [0_u8; 512]; + while !request.ends_with(b"\r\n\r\n") { + let count = stream.read(&mut buffer)?; + if count == 0 { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "client opening request ended before the header terminator", + )); + } + request.extend_from_slice(&buffer[..count]); + } + Ok(()) +} + +fn read_masked_client_frame( + stream: &mut TcpStream, + expected_first_byte: u8, +) -> io::Result> { + stream.set_read_timeout(Some(Duration::from_secs(2)))?; + let mut header = [0_u8; 2]; + stream.read_exact(&mut header)?; + if header[0] != expected_first_byte || header[1] & 0x80 == 0 { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "client frame did not have the expected final opcode and masking bit", + )); + } + let payload_length = match header[1] & 0x7f { + value @ 0..=125 => usize::from(value), + 126 => { + let mut extended = [0_u8; 2]; + stream.read_exact(&mut extended)?; + usize::from(u16::from_be_bytes(extended)) + } + 127 => { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "test fixture does not admit 64-bit client frame lengths", + )); + } + _ => unreachable!("7-bit WebSocket payload marker"), + }; + if expected_first_byte == 0x8a && payload_length > 125 { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "Pong payload exceeded the RFC 6455 control-frame bound", + )); + } + let mut mask = [0_u8; 4]; + stream.read_exact(&mut mask)?; + let mut payload = vec![0_u8; payload_length]; + stream.read_exact(&mut payload)?; + for (index, byte) in payload.iter_mut().enumerate() { + *byte ^= mask[index % mask.len()]; + } + Ok(payload) +} + +fn locate_nodes_command() -> Result> { + let query = WebDriverBiDiAccessibilityQuery::new(Some("button"), Some("Checkout"), 2)?; + Ok(WebDriverBiDiLocateNodesCommand::new( + 7, + "top-level-context", + &query, + )?) +} + +fn establish_with_ping(keep_open: Duration) -> Result> { + let listener = TcpListener::bind(("127.0.0.1", 0))?; + let local_addr = listener.local_addr()?; + let server = thread::spawn(move || -> io::Result<()> { + let (mut stream, _) = listener.accept()?; + read_opening_request(&mut stream)?; + stream.write_all( + b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=\r\n\r\n", + )?; + let _command = read_masked_client_frame(&mut stream, 0x81)?; + let ping_length = u8::try_from(PING_PAYLOAD.len()).map_err(|_| { + io::Error::new( + io::ErrorKind::InvalidData, + "test Ping payload exceeded one-byte length", + ) + })?; + stream.write_all(&[0x89, ping_length])?; + stream.write_all(PING_PAYLOAD)?; + thread::sleep(keep_open); + Ok(()) + }); + + let endpoint = format!("ws://{local_addr}/session/{SESSION_ID}"); + let key = WebDriverBiDiWebSocketClientKey::new(RFC6455_SAMPLE_KEY)?; + let plan = WebDriverBiDiWebSocketHandshakePlan::new(connect(&endpoint)?, key)?; + let written = plan.write_opening_request(Duration::from_millis(500))?; + let established = written.read_opening_response(Duration::from_millis(500))?; + Ok((established, server)) +} + +fn join_ping_server(server: thread::JoinHandle>) -> Result<(), Box> { + let result = server + .join() + .map_err(|_| io::Error::other("Ping failure test server panicked"))?; + Ok(result?) +} + +#[test] +fn locate_nodes_exchange_answers_ping_and_ignores_unsolicited_pong_before_response() +-> Result<(), Box> { + let listener = TcpListener::bind(("127.0.0.1", 0))?; + let local_addr = listener.local_addr()?; + let server = thread::spawn(move || -> io::Result<(Vec, Vec)> { + let (mut stream, _) = listener.accept()?; + read_opening_request(&mut stream)?; + stream.write_all( + b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=\r\n\r\n", + )?; + let command = read_masked_client_frame(&mut stream, 0x81)?; + let ping_length = u8::try_from(PING_PAYLOAD.len()).map_err(|_| { + io::Error::new( + io::ErrorKind::InvalidData, + "test Ping payload exceeded one-byte length", + ) + })?; + stream.write_all(&[0x89, ping_length])?; + stream.write_all(PING_PAYLOAD)?; + let pong = read_masked_client_frame(&mut stream, 0x8a)?; + stream.write_all(&[0x8a, 0])?; + let response_length = u8::try_from(RESPONSE_DOCUMENT.len()).map_err(|_| { + io::Error::new( + io::ErrorKind::InvalidData, + "test response exceeded one-byte length", + ) + })?; + stream.write_all(&[0x81, response_length])?; + stream.write_all(RESPONSE_DOCUMENT.as_bytes())?; + Ok((command, pong)) + }); + + let endpoint = format!("ws://{local_addr}/session/{SESSION_ID}"); + let key = WebDriverBiDiWebSocketClientKey::new(RFC6455_SAMPLE_KEY)?; + let plan = WebDriverBiDiWebSocketHandshakePlan::new(connect(&endpoint)?, key)?; + let written = plan.write_opening_request(Duration::from_millis(500))?; + let established = written.read_opening_response(Duration::from_millis(500))?; + let command = locate_nodes_command()?; + let expected_command = command.as_json().as_bytes().to_vec(); + let mut pong_keys = [WebDriverBiDiWebSocketMaskKey::new([0x51, 0x52, 0x53, 0x54])].into_iter(); + let exchanged = established.exchange_locate_nodes( + command, + WebDriverBiDiWebSocketMaskKey::new([0x11, 0x22, 0x33, 0x44]), + &mut || pong_keys.next(), + Duration::from_millis(500), + ); + + let server_result = server + .join() + .map_err(|_| io::Error::other("interleaved control-frame test server panicked"))?; + assert!(exchanged.is_ok(), "{exchanged:?}"); + assert!(server_result.is_ok(), "{server_result:?}"); + let (received_command, received_pong) = server_result?; + assert_eq!(received_command, expected_command); + assert_eq!(received_pong, PING_PAYLOAD); + + let (established, result) = exchanged?; + assert_eq!(result.command_id(), 7); + assert_eq!(result.nodes().len(), 1); + assert_eq!(result.nodes()[0].shared_id(), "shared-1"); + assert_eq!( + established + .transport_evidence() + .verified_peer() + .socket_addr(), + local_addr + ); + Ok(()) +} + +#[test] +fn locate_nodes_exchange_fails_closed_when_ping_entropy_is_unavailable() +-> Result<(), Box> { + let (established, server) = establish_with_ping(Duration::from_millis(100))?; + let exchanged = established.exchange_locate_nodes( + locate_nodes_command()?, + WebDriverBiDiWebSocketMaskKey::new([0x11, 0x22, 0x33, 0x44]), + &mut || None, + Duration::from_millis(500), + ); + + let error = exchanged + .err() + .ok_or_else(|| io::Error::other("Ping without masking entropy unexpectedly succeeded"))?; + assert_eq!( + error.to_string(), + "WebDriver BiDi locateNodes exchange received Ping without a fresh caller-supplied Pong masking key" + ); + join_ping_server(server) +} + +#[test] +fn locate_nodes_exchange_charges_ping_callback_time_to_exchange_deadline() +-> Result<(), Box> { + let (established, server) = establish_with_ping(Duration::from_millis(650))?; + let pong_key = WebDriverBiDiWebSocketMaskKey::new([0x51, 0x52, 0x53, 0x54]); + let exchanged = established.exchange_locate_nodes( + locate_nodes_command()?, + WebDriverBiDiWebSocketMaskKey::new([0x11, 0x22, 0x33, 0x44]), + &mut || { + thread::sleep(Duration::from_millis(550)); + Some(pong_key) + }, + Duration::from_millis(500), + ); + + let error = exchanged.err().ok_or_else(|| { + io::Error::other("slow Ping callback unexpectedly reset the exchange deadline") + })?; + assert_eq!( + error.to_string(), + "WebDriver BiDi locateNodes exchange exhausted its 500ms end-to-end deadline before the next operation" + ); + join_ping_server(server) +} + +#[test] +fn locate_nodes_exchange_bounds_valid_interleaved_control_frames() -> Result<(), Box> { + let listener = TcpListener::bind(("127.0.0.1", 0))?; + let local_addr = listener.local_addr()?; + let server = thread::spawn(move || -> io::Result<()> { + let (mut stream, _) = listener.accept()?; + read_opening_request(&mut stream)?; + stream.write_all( + b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=\r\n\r\n", + )?; + let _command = read_masked_client_frame(&mut stream, 0x81)?; + + let mut frames = Vec::with_capacity( + (MAX_WEBDRIVER_BIDI_CONTROL_FRAMES_PER_EXCHANGE + 1) * 2 + RESPONSE_DOCUMENT.len() + 2, + ); + for _ in 0..=MAX_WEBDRIVER_BIDI_CONTROL_FRAMES_PER_EXCHANGE { + frames.extend_from_slice(&[0x8a, 0]); + } + let response_length = u8::try_from(RESPONSE_DOCUMENT.len()).map_err(|_| { + io::Error::new( + io::ErrorKind::InvalidData, + "test response exceeded one-byte length", + ) + })?; + frames.extend_from_slice(&[0x81, response_length]); + frames.extend_from_slice(RESPONSE_DOCUMENT.as_bytes()); + stream.write_all(&frames) + }); + + let endpoint = format!("ws://{local_addr}/session/{SESSION_ID}"); + let key = WebDriverBiDiWebSocketClientKey::new(RFC6455_SAMPLE_KEY)?; + let plan = WebDriverBiDiWebSocketHandshakePlan::new(connect(&endpoint)?, key)?; + let written = plan.write_opening_request(Duration::from_millis(500))?; + let established = written.read_opening_response(Duration::from_millis(500))?; + let exchanged = established.exchange_locate_nodes( + locate_nodes_command()?, + WebDriverBiDiWebSocketMaskKey::new([0x11, 0x22, 0x33, 0x44]), + &mut || None, + Duration::from_millis(500), + ); + + let server_result = server + .join() + .map_err(|_| io::Error::other("control-frame limit test server panicked"))?; + assert!(server_result.is_ok(), "{server_result:?}"); + let error = exchanged + .err() + .ok_or_else(|| io::Error::other("control-frame flood unexpectedly reached response"))?; + assert!(matches!( + error, + WebDriverBiDiLocateNodesExchangeError::ControlFrameLimitExceeded { + maximum_control_frames, + } if maximum_control_frames == MAX_WEBDRIVER_BIDI_CONTROL_FRAMES_PER_EXCHANGE + )); + Ok(()) +} diff --git a/crates/originweave-network/tests/webdriver_bidi_locate_nodes_pong_write_failure.rs b/crates/originweave-network/tests/webdriver_bidi_locate_nodes_pong_write_failure.rs new file mode 100644 index 000000000..ec5f8a915 --- /dev/null +++ b/crates/originweave-network/tests/webdriver_bidi_locate_nodes_pong_write_failure.rs @@ -0,0 +1,153 @@ +use std::{ + error::Error, + io::{self, Read, Write}, + net::{Shutdown, TcpListener, TcpStream}, + thread, + time::Duration, +}; + +use originweave_core::{ + WebDriverBiDiAccessibilityQuery, WebDriverBiDiLocateNodesCommand, + WebDriverBiDiWebSocketEndpoint, +}; +use originweave_network::{ + WebDriverBiDiLocateNodesExchangeError, WebDriverBiDiTcpConnection, + WebDriverBiDiTcpConnectionPlan, WebDriverBiDiWebSocketClientKey, + WebDriverBiDiWebSocketFrameError, WebDriverBiDiWebSocketHandshakePlan, + WebDriverBiDiWebSocketMaskKey, +}; + +const SESSION_ID: &str = "01234567-89ab-cdef-0123-456789abcdef"; +const RFC6455_SAMPLE_KEY: &str = "dGhlIHNhbXBsZSBub25jZQ=="; + +fn connect(endpoint: &str) -> Result> { + let admitted = WebDriverBiDiWebSocketEndpoint::new(endpoint)?; + let correlated = admitted.correlate_session_id(SESSION_ID)?; + let target = correlated.into_explicit_connect_target()?; + let plan = WebDriverBiDiTcpConnectionPlan::new(target, Duration::from_secs(1), 1)?; + Ok(plan.connect()?) +} + +fn read_opening_request(stream: &mut TcpStream) -> io::Result<()> { + stream.set_read_timeout(Some(Duration::from_secs(2)))?; + let mut request = Vec::new(); + let mut buffer = [0_u8; 512]; + while !request.ends_with(b"\r\n\r\n") { + let count = stream.read(&mut buffer)?; + if count == 0 { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "client opening request ended before the header terminator", + )); + } + request.extend_from_slice(&buffer[..count]); + } + Ok(()) +} + +fn read_masked_client_text_frame(stream: &mut TcpStream) -> io::Result<()> { + stream.set_read_timeout(Some(Duration::from_secs(2)))?; + let mut header = [0_u8; 2]; + stream.read_exact(&mut header)?; + if header[0] != 0x81 || header[1] & 0x80 == 0 { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "client command was not one final masked text frame", + )); + } + + let payload_length = match header[1] & 0x7f { + value @ 0..=125 => u64::from(value), + 126 => { + let mut extended = [0_u8; 2]; + stream.read_exact(&mut extended)?; + u64::from(u16::from_be_bytes(extended)) + } + _ => { + let mut extended = [0_u8; 8]; + stream.read_exact(&mut extended)?; + u64::from_be_bytes(extended) + } + }; + + let mut mask = [0_u8; 4]; + stream.read_exact(&mut mask)?; + let mut remaining = usize::try_from(payload_length).map_err(|_| { + io::Error::new( + io::ErrorKind::InvalidData, + "client command length cannot fit this test process", + ) + })?; + let mut buffer = [0_u8; 512]; + while remaining != 0 { + let chunk = remaining.min(buffer.len()); + stream.read_exact(&mut buffer[..chunk])?; + remaining -= chunk; + } + Ok(()) +} + +fn locate_nodes_command() -> Result> { + let query = WebDriverBiDiAccessibilityQuery::new(Some("button"), Some("Checkout"), 2)?; + Ok(WebDriverBiDiLocateNodesCommand::new( + 7, + "top-level-context", + &query, + )?) +} + +#[test] +fn locate_nodes_exchange_preserves_pong_write_failure_after_ping() -> Result<(), Box> { + let listener = TcpListener::bind(("127.0.0.1", 0))?; + let local_addr = listener.local_addr()?; + let server = thread::spawn(move || -> io::Result<()> { + let (mut stream, _) = listener.accept()?; + read_opening_request(&mut stream)?; + stream.write_all( + b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=\r\n\r\n", + )?; + read_masked_client_text_frame(&mut stream)?; + stream.write_all(&[0x89, 0])?; + thread::sleep(Duration::from_millis(250)); + Ok(()) + }); + + let endpoint = format!("ws://{local_addr}/session/{SESSION_ID}"); + let key = WebDriverBiDiWebSocketClientKey::new(RFC6455_SAMPLE_KEY)?; + let connection = connect(&endpoint)?; + let shutdown_stream = connection.stream().try_clone()?; + let plan = WebDriverBiDiWebSocketHandshakePlan::new(connection, key)?; + let written = plan.write_opening_request(Duration::from_millis(500))?; + let established = written.read_opening_response(Duration::from_millis(500))?; + let pong_key = WebDriverBiDiWebSocketMaskKey::new([0x51, 0x52, 0x53, 0x54]); + let exchanged = established.exchange_locate_nodes( + locate_nodes_command()?, + WebDriverBiDiWebSocketMaskKey::new([0x11, 0x22, 0x33, 0x44]), + &mut || { + let shutdown = shutdown_stream.shutdown(Shutdown::Write); + assert!(shutdown.is_ok(), "{shutdown:?}"); + Some(pong_key) + }, + Duration::from_millis(500), + ); + + let server_result = server + .join() + .map_err(|_| io::Error::other("Pong write failure test server panicked"))?; + assert!(server_result.is_ok(), "{server_result:?}"); + + let error = exchanged.err().ok_or_else(|| { + io::Error::other("locateNodes exchange unexpectedly survived a closed client write half") + })?; + assert!( + matches!( + &error, + WebDriverBiDiLocateNodesExchangeError::Frame( + WebDriverBiDiWebSocketFrameError::FrameWriteFailed { .. } + ) + ), + "{error:?}" + ); + assert!(error.source().is_some()); + Ok(()) +} diff --git a/crates/originweave-network/tests/webdriver_bidi_websocket_locate_nodes_exchange.rs b/crates/originweave-network/tests/webdriver_bidi_websocket_locate_nodes_exchange.rs new file mode 100644 index 000000000..d04ff0eb0 --- /dev/null +++ b/crates/originweave-network/tests/webdriver_bidi_websocket_locate_nodes_exchange.rs @@ -0,0 +1,345 @@ +use std::{ + error::Error, + io::{self, Read, Write}, + net::{SocketAddr, TcpListener, TcpStream}, + thread, + time::Duration, +}; + +use originweave_core::{ + WebDriverBiDiAccessibilityQuery, WebDriverBiDiLocateNodesCommand, + WebDriverBiDiWebSocketEndpoint, +}; +use originweave_network::{ + WebDriverBiDiLocateNodesExchangeError, WebDriverBiDiTcpConnectionPlan, + WebDriverBiDiWebSocketClientKey, WebDriverBiDiWebSocketEstablished, + WebDriverBiDiWebSocketHandshakePlan, WebDriverBiDiWebSocketMaskKey, +}; + +const SESSION_ID: &str = "01234567-89ab-cdef-0123-456789abcdef"; +const RFC6455_SAMPLE_KEY: &str = "dGhlIHNhbXBsZSBub25jZQ=="; +const RESPONSE_DOCUMENT: &str = + r#"{"type":"success","id":7,"result":{"nodes":[{"type":"node","sharedId":"shared-1"}]}}"#; +const MISMATCHED_RESPONSE_DOCUMENT: &str = r#"{"type":"success","id":8,"result":{"nodes":[]}}"#; + +type ServerHandle = thread::JoinHandle>>; +type EstablishedFixture = + Result<(SocketAddr, WebDriverBiDiWebSocketEstablished, ServerHandle), Box>; + +fn connect(endpoint: &str) -> originweave_network::WebDriverBiDiTcpConnection { + let admitted = WebDriverBiDiWebSocketEndpoint::new(endpoint); + assert!(admitted.is_ok(), "{admitted:?}"); + let Ok(admitted) = admitted else { + unreachable!("asserted valid endpoint") + }; + let correlated = admitted.correlate_session_id(SESSION_ID); + assert!(correlated.is_ok(), "{correlated:?}"); + let Ok(correlated) = correlated else { + unreachable!("asserted correlated endpoint") + }; + let target = correlated.into_explicit_connect_target(); + assert!(target.is_ok(), "{target:?}"); + let Ok(target) = target else { + unreachable!("asserted explicit target") + }; + let plan = WebDriverBiDiTcpConnectionPlan::new(target, Duration::from_secs(1), 1); + assert!(plan.is_ok(), "{plan:?}"); + let Ok(plan) = plan else { + unreachable!("asserted connection plan") + }; + let connection = plan.connect(); + assert!(connection.is_ok(), "{connection:?}"); + let Ok(connection) = connection else { + unreachable!("asserted loopback connection") + }; + connection +} + +fn read_opening_request(stream: &mut TcpStream) -> io::Result> { + stream.set_read_timeout(Some(Duration::from_secs(2)))?; + let mut request = Vec::new(); + let mut buffer = [0_u8; 512]; + while !request.ends_with(b"\r\n\r\n") { + let count = stream.read(&mut buffer)?; + if count == 0 { + break; + } + request.extend_from_slice(&buffer[..count]); + } + Ok(request) +} + +fn read_client_text_frame(stream: &mut TcpStream) -> io::Result> { + stream.set_read_timeout(Some(Duration::from_secs(2)))?; + let mut header = [0_u8; 2]; + stream.read_exact(&mut header)?; + assert_eq!(header[0], 0x81); + assert_ne!(header[1] & 0x80, 0); + + let payload_length = match header[1] & 0x7f { + value @ 0..=125 => usize::from(value), + 126 => { + let mut extended = [0_u8; 2]; + stream.read_exact(&mut extended)?; + usize::from(u16::from_be_bytes(extended)) + } + 127 => { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "test fixture does not admit 64-bit client frame lengths", + )); + } + _ => unreachable!("7-bit WebSocket payload marker"), + }; + + let mut mask = [0_u8; 4]; + stream.read_exact(&mut mask)?; + let mut payload = vec![0_u8; payload_length]; + stream.read_exact(&mut payload)?; + for (index, byte) in payload.iter_mut().enumerate() { + *byte ^= mask[index % mask.len()]; + } + Ok(payload) +} + +fn server_frame(first_byte: u8, payload: &[u8]) -> io::Result> { + let payload_length = u8::try_from(payload.len()).map_err(|_| { + io::Error::new( + io::ErrorKind::InvalidData, + "test response must fit one short WebSocket text frame", + ) + })?; + if payload_length > 125 { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "test response must fit one short WebSocket text frame", + )); + } + let mut frame = vec![first_byte, payload_length]; + frame.extend_from_slice(payload); + Ok(frame) +} + +fn establish_with_server_frame(response_frame: &[u8]) -> EstablishedFixture { + let listener = TcpListener::bind(("127.0.0.1", 0))?; + let local_addr = listener.local_addr()?; + let response_frame = response_frame.to_vec(); + let server = thread::spawn(move || -> io::Result> { + let (mut stream, _) = listener.accept()?; + let request = read_opening_request(&mut stream)?; + if !request.ends_with(b"\r\n\r\n") { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "client opening request was incomplete", + )); + } + stream.write_all( + b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=\r\n\r\n", + )?; + let command = read_client_text_frame(&mut stream)?; + stream.write_all(&response_frame)?; + Ok(command) + }); + + let endpoint = format!("ws://{local_addr}/session/{SESSION_ID}"); + let key = WebDriverBiDiWebSocketClientKey::new(RFC6455_SAMPLE_KEY)?; + let plan = WebDriverBiDiWebSocketHandshakePlan::new(connect(&endpoint), key)?; + let written = plan.write_opening_request(Duration::from_millis(500))?; + let established = written.read_opening_response(Duration::from_millis(500))?; + Ok((local_addr, established, server)) +} + +fn locate_nodes_command() -> WebDriverBiDiLocateNodesCommand { + let name = "x".repeat(512); + let query = WebDriverBiDiAccessibilityQuery::new(Some("button"), Some(&name), 2); + assert!(query.is_ok(), "{query:?}"); + let Ok(query) = query else { + unreachable!("asserted valid test query") + }; + let command = WebDriverBiDiLocateNodesCommand::new(7, "top-level-context", &query); + assert!(command.is_ok(), "{command:?}"); + let Ok(command) = command else { + unreachable!("asserted valid test command") + }; + command +} + +fn exchange_error( + response_frame: &[u8], + frame_timeout: Duration, + server_must_receive_command: bool, +) -> WebDriverBiDiLocateNodesExchangeError { + let fixture = establish_with_server_frame(response_frame); + assert!(fixture.is_ok(), "{fixture:?}"); + let Ok((_, established, server)) = fixture else { + unreachable!("asserted valid test exchange fixture") + }; + let error = established.exchange_locate_nodes( + locate_nodes_command(), + WebDriverBiDiWebSocketMaskKey::new([0x11, 0x22, 0x33, 0x44]), + &mut || None, + frame_timeout, + ); + assert!(error.is_err(), "{error:?}"); + let Err(error) = error else { + unreachable!("asserted failing test exchange") + }; + let server_result = server.join(); + assert!(server_result.is_ok(), "{server_result:?}"); + let Ok(server_result) = server_result else { + unreachable!("asserted joined test server") + }; + assert_eq!( + server_result.is_ok(), + server_must_receive_command, + "test server command receipt did not match the exchange boundary" + ); + error +} + +#[test] +fn established_stream_exchanges_exact_locate_nodes_command_and_correlates_wire_result() { + let response_frame = server_frame(0x81, RESPONSE_DOCUMENT.as_bytes()); + assert!(response_frame.is_ok(), "{response_frame:?}"); + let Ok(response_frame) = response_frame else { + return; + }; + let fixture = establish_with_server_frame(&response_frame); + assert!(fixture.is_ok(), "{fixture:?}"); + let Ok((local_addr, established, server)) = fixture else { + return; + }; + + let query = WebDriverBiDiAccessibilityQuery::new(Some("button"), Some("Checkout"), 2); + assert!(query.is_ok(), "{query:?}"); + let Ok(query) = query else { + return; + }; + let command = WebDriverBiDiLocateNodesCommand::new(7, "top-level-context", &query); + assert!(command.is_ok(), "{command:?}"); + let Ok(command) = command else { + return; + }; + let expected_command = command.as_json().as_bytes().to_vec(); + + let exchanged = established.exchange_locate_nodes( + command, + WebDriverBiDiWebSocketMaskKey::new([0x11, 0x22, 0x33, 0x44]), + &mut || None, + Duration::from_millis(500), + ); + assert!(exchanged.is_ok(), "{exchanged:?}"); + let Ok((established, result)) = exchanged else { + return; + }; + + assert_eq!(result.command_id(), 7); + assert_eq!(result.browsing_context(), "top-level-context"); + assert_eq!(result.max_node_count(), 2); + assert_eq!(result.nodes().len(), 1); + assert_eq!(result.nodes()[0].shared_id(), "shared-1"); + assert_eq!( + established + .transport_evidence() + .verified_peer() + .socket_addr(), + local_addr + ); + assert_eq!( + established + .transport_evidence() + .verified_peer() + .session_id(), + SESSION_ID + ); + + let server_result = server.join(); + assert!(server_result.is_ok(), "{server_result:?}"); + if let Ok(command_result) = server_result { + assert!(command_result.is_ok(), "{command_result:?}"); + if let Ok(actual_command) = command_result { + assert_eq!(actual_command, expected_command); + } + } +} + +#[test] +fn exchange_rejects_binary_or_orphan_continuation_response_frames() { + for (first_byte, expected_fin, expected_opcode) in + [(0x82_u8, true, 0x02_u8), (0x80_u8, true, 0x00_u8)] + { + let response_frame = server_frame(first_byte, &[]); + assert!(response_frame.is_ok(), "{response_frame:?}"); + let Ok(response_frame) = response_frame else { + return; + }; + let fixture = establish_with_server_frame(&response_frame); + assert!(fixture.is_ok(), "{fixture:?}"); + let Ok((_, established, server)) = fixture else { + return; + }; + + let error = established.exchange_locate_nodes( + locate_nodes_command(), + WebDriverBiDiWebSocketMaskKey::new([0x11, 0x22, 0x33, 0x44]), + &mut || None, + Duration::from_millis(500), + ); + assert!(error.is_err(), "{error:?}"); + let Err(error) = error else { + unreachable!("asserted invalid response frame failure") + }; + assert!(matches!( + error, + WebDriverBiDiLocateNodesExchangeError::UnexpectedResponseFrame { + fin, + opcode, + } if fin == expected_fin && opcode == expected_opcode + )); + + let server_result = server.join(); + assert!(server_result.is_ok(), "{server_result:?}"); + if let Ok(command_result) = server_result { + assert!(command_result.is_ok(), "{command_result:?}"); + } + } +} + +#[test] +fn exchange_preserves_frame_document_and_response_admission_boundaries() { + let write_error = exchange_error(&[], Duration::ZERO, false); + assert!(matches!( + write_error, + WebDriverBiDiLocateNodesExchangeError::ExchangeDeadlineExceeded { + exchange_timeout + } if exchange_timeout.is_zero() + )); + + let read_error = exchange_error(&[], Duration::from_millis(500), true); + assert!(matches!( + read_error, + WebDriverBiDiLocateNodesExchangeError::Frame(_) + )); + + let invalid_utf8_frame = server_frame(0x81, &[0xff]); + assert!(invalid_utf8_frame.is_ok(), "{invalid_utf8_frame:?}"); + let Ok(invalid_utf8_frame) = invalid_utf8_frame else { + return; + }; + let document_error = exchange_error(&invalid_utf8_frame, Duration::from_millis(500), true); + assert!(matches!( + document_error, + WebDriverBiDiLocateNodesExchangeError::ResponseDocument(_) + )); + + let mismatched_frame = server_frame(0x81, MISMATCHED_RESPONSE_DOCUMENT.as_bytes()); + assert!(mismatched_frame.is_ok(), "{mismatched_frame:?}"); + let Ok(mismatched_frame) = mismatched_frame else { + return; + }; + let response_error = exchange_error(&mismatched_frame, Duration::from_millis(500), true); + assert!(matches!( + response_error, + WebDriverBiDiLocateNodesExchangeError::LocateNodesResponse(_) + )); +} diff --git a/tests/test_webdriver_bidi_locate_nodes_coverage_shape.py b/tests/test_webdriver_bidi_locate_nodes_coverage_shape.py new file mode 100644 index 000000000..2b5b4d815 --- /dev/null +++ b/tests/test_webdriver_bidi_locate_nodes_coverage_shape.py @@ -0,0 +1,56 @@ +"""Coverage-shape contract for the bounded BiDi locateNodes exchange boundary. + +The production coverage gate is exact across functions, lines, regions, and branches. +Keeping the caller-supplied Pong entropy callback generic monomorphizes the whole +exchange function per closure type, which creates synthetic per-instantiation +coverage holes despite exercising the protocol paths. The callback is therefore a +borrowed trait object at this boundary: its behavior remains stateful and caller +owned without multiplying production coverage regions. + +The final node-binding error conversion must also avoid generating a separate closure +or helper function. The library is linked into both unit and integration-test +harnesses, and either form can acquire an uncovered duplicate instantiation even when +the real fail-closed binding path is exercised. A direct match inside the already +exercised public wrapper keeps the typed error conversion at the causal boundary +without adding another production symbol for coverage to duplicate. +""" + +from __future__ import annotations + +import pathlib +import unittest + +ROOT = pathlib.Path(__file__).resolve().parents[1] +SOURCE = ( + ROOT + / "crates" + / "originweave-network" + / "src" + / "webdriver_bidi_locate_nodes_exchange.rs" +) + + +class WebDriverBiDiLocateNodesCoverageShapeTests(unittest.TestCase): + """Prevent monomorphization artifacts from invalidating exact coverage evidence.""" + + def test_pong_entropy_callback_is_non_generic_at_exchange_boundary(self) -> None: + source = SOURCE.read_text(encoding="utf-8") + self.assertIn( + "next_pong_key: &mut dyn FnMut() -> Option", + source, + ) + self.assertNotIn( + "next_pong_key: impl FnMut() -> Option", + source, + ) + + def test_node_binding_error_conversion_stays_inside_exercised_wrapper(self) -> None: + source = SOURCE.read_text(encoding="utf-8") + self.assertIn("let handles = match result.bind_current_nodes(", source) + self.assertNotIn("fn map_node_binding_error(", source) + self.assertNotIn(".map_err(map_node_binding_error)?;", source) + self.assertNotIn(".map_err(|error| {", source) + + +if __name__ == "__main__": + unittest.main()