diff --git a/crates/originweave-network/src/lib.rs b/crates/originweave-network/src/lib.rs index 1e94f6075..f34660d65 100644 --- a/crates/originweave-network/src/lib.rs +++ b/crates/originweave-network/src/lib.rs @@ -9,16 +9,17 @@ //! unmasked server-frame reads, assembles bounded WebDriver BiDi text messages, //! classifies complete local-end JSON envelopes, tracks bounded command-response //! correlation, transports a narrowly typed pointer click, admits its typed -//! correlated protocol acknowledgment, admits a bounded navigation-committed -//! post-condition observation for one exact registered context and URL, rotates -//! the matched context's document epoch only from an exact caller-captured -//! pre-action epoch, derives and binds the committed HTTP(S) URL's canonical -//! origin to that newly advanced document, sends narrowly typed `session.status` -//! and `session.end` commands, admits typed correlated status and end responses, -//! observes bounded peer Close or clean-EOF transport cessation, and keeps -//! protocol/transport evidence separate from explicit operational teardown -//! observations without exposing generic JSON bodies or granting browser, TLS, -//! policy, secret, process, profile, or Agent authority. +//! correlated protocol acknowledgment, sends a context-bound subscription for +//! committed-navigation events, retains its typed bounded correlated subscription +//! identifier, admits a bounded navigation-committed post-condition observation +//! for one exact registered context and URL, rotates the matched context's document +//! epoch only from an exact caller-captured pre-action epoch, derives and binds the +//! committed HTTP(S) URL's canonical origin to that newly advanced document, sends +//! narrowly typed `session.status` and `session.end` commands, admits typed +//! correlated status and end responses, observes bounded peer Close or clean-EOF +//! transport cessation, and keeps protocol/transport evidence separate from +//! explicit operational teardown observations without exposing generic JSON bodies +//! or granting browser, TLS, policy, secret, process, profile, or Agent authority. #![forbid(unsafe_code)] #![deny(missing_docs)] @@ -28,6 +29,8 @@ mod webdriver_bidi_command_correlation; mod webdriver_bidi_connection; mod webdriver_bidi_json_envelope; mod webdriver_bidi_navigation_committed_postcondition; +mod webdriver_bidi_navigation_committed_subscription; +mod webdriver_bidi_navigation_committed_subscription_response; mod webdriver_bidi_navigation_document_advance; mod webdriver_bidi_navigation_document_origin; mod webdriver_bidi_pointer_click_response; @@ -69,6 +72,15 @@ pub use webdriver_bidi_navigation_committed_postcondition::{ WebDriverBiDiNavigationCommittedObservationError, WebDriverBiDiNavigationCommittedProjectionError, }; +pub use webdriver_bidi_navigation_committed_subscription::{ + WebDriverBiDiNavigationCommittedSubscriptionCommand, + WebDriverBiDiNavigationCommittedSubscriptionCommandError, +}; +pub use webdriver_bidi_navigation_committed_subscription_response::{ + MAX_WEBDRIVER_BIDI_SUBSCRIPTION_IDENTIFIER_BYTES, + WebDriverBiDiNavigationCommittedSubscriptionResponseError, + WebDriverBiDiNavigationCommittedSubscriptionResult, +}; pub use webdriver_bidi_navigation_document_advance::{ WebDriverBiDiNavigationCommittedDocumentAdvance, WebDriverBiDiNavigationCommittedDocumentAdvanceError, diff --git a/crates/originweave-network/src/webdriver_bidi_navigation_committed_subscription.rs b/crates/originweave-network/src/webdriver_bidi_navigation_committed_subscription.rs new file mode 100644 index 000000000..fae6b9bd0 --- /dev/null +++ b/crates/originweave-network/src/webdriver_bidi_navigation_committed_subscription.rs @@ -0,0 +1,223 @@ +use std::{error::Error, fmt, time::Duration}; + +use originweave_core::{ + BrowserAuthorityRegistry, BrowserRegistryError, BrowserSessionId, BrowsingContextId, +}; + +use crate::{ + MAX_WEBDRIVER_BIDI_JS_UINT, WEBDRIVER_BIDI_NAVIGATION_COMMITTED_METHOD, + WebDriverBiDiCommandCorrelation, WebDriverBiDiCommandCorrelationError, + WebDriverBiDiWebSocketEstablished, WebDriverBiDiWebSocketFrameError, + WebDriverBiDiWebSocketMaskKey, +}; + +const SESSION_SUBSCRIBE_METHOD: &str = "session.subscribe"; + +/// One context-scoped subscription for the committed-navigation WebDriver BiDi event. +/// +/// This command is deliberately narrower than the protocol's generic `session.subscribe` surface: +/// it can request only `browsingContext.navigationCommitted`, for one external context that already +/// maps to the exact supplied OriginWeave session/context pair. It does not expose arbitrary event +/// names, global subscriptions, user-context subscriptions, generic JSON, or arbitrary method +/// dispatch. Successful construction or transport does not authenticate Chromium, authorize a +/// navigation, grant destination or policy authority, or make later event data reusable Agent +/// authority. +pub struct WebDriverBiDiNavigationCommittedSubscriptionCommand { + command_id: u64, + browser_session: BrowserSessionId, + browsing_context: BrowsingContextId, + external_context: String, +} + +impl WebDriverBiDiNavigationCommittedSubscriptionCommand { + /// Construct one bounded context-scoped committed-navigation subscription command. + /// + /// The external protocol identifier must already name the exact registered OriginWeave + /// session/context pair. No registry state is created as a side effect of untrusted adapter text. + pub fn new( + command_id: u64, + registry: &BrowserAuthorityRegistry, + browser_session: BrowserSessionId, + browsing_context: BrowsingContextId, + external_context: &str, + ) -> Result { + if command_id > MAX_WEBDRIVER_BIDI_JS_UINT { + return Err( + WebDriverBiDiNavigationCommittedSubscriptionCommandError::CommandIdOutOfRange { + command_id, + maximum_command_id: MAX_WEBDRIVER_BIDI_JS_UINT, + }, + ); + } + require_registered_context( + registry, + browser_session, + browsing_context, + external_context, + )?; + Ok(Self { + command_id, + browser_session, + browsing_context, + external_context: external_context.to_owned(), + }) + } + + /// Return the exact local correlation identifier serialized by this command. + #[must_use] + pub const fn command_id(&self) -> u64 { + self.command_id + } + + /// Return the exact registered OriginWeave browser session bound during construction. + #[must_use] + pub const fn browser_session(&self) -> BrowserSessionId { + self.browser_session + } + + /// Return the exact registered OriginWeave browsing context bound during construction. + #[must_use] + pub const fn browsing_context(&self) -> BrowsingContextId { + self.browsing_context + } + + /// Borrow the exact external WebDriver BiDi context identifier serialized by this command. + #[must_use] + pub fn external_context(&self) -> &str { + &self.external_context + } + + /// Revalidate, register, and write this exact subscription on an established verified BiDi stream. + /// + /// Context binding is revalidated immediately before command correlation and network I/O so a + /// command retained across registry retirement cannot subscribe a stale or replacement context. + /// Correlation registration then occurs before the first possible remote side effect. A binding + /// or correlation failure therefore writes nothing. After successful registration, a frame-write + /// failure consumes the transport and intentionally leaves the identifier outstanding because a + /// partial or fully emitted frame has ambiguous remote effect. + pub fn send( + self, + registry: &BrowserAuthorityRegistry, + established: WebDriverBiDiWebSocketEstablished, + correlation: &mut WebDriverBiDiCommandCorrelation, + masking_key: WebDriverBiDiWebSocketMaskKey, + frame_timeout: Duration, + ) -> Result< + WebDriverBiDiWebSocketEstablished, + WebDriverBiDiNavigationCommittedSubscriptionCommandError, + > { + require_registered_context( + registry, + self.browser_session, + self.browsing_context, + &self.external_context, + )?; + correlation + .register_command(self.command_id) + .map_err(|source| { + WebDriverBiDiNavigationCommittedSubscriptionCommandError::Correlation { source } + })?; + let message = self.serialized(); + established + .write_text_frame(&message, masking_key, frame_timeout) + .map_err(|source| { + WebDriverBiDiNavigationCommittedSubscriptionCommandError::FrameWrite { source } + }) + } + + fn serialized(&self) -> String { + let mut message = format!( + "{{\"id\":{},\"method\":\"{SESSION_SUBSCRIBE_METHOD}\",\"params\":{{\"events\":[\"{WEBDRIVER_BIDI_NAVIGATION_COMMITTED_METHOD}\"],\"contexts\":[", + self.command_id + ); + push_json_string(&mut message, &self.external_context); + message.push_str("]}}"); + message + } +} + +fn require_registered_context( + registry: &BrowserAuthorityRegistry, + browser_session: BrowserSessionId, + browsing_context: BrowsingContextId, + external_context: &str, +) -> Result<(), WebDriverBiDiNavigationCommittedSubscriptionCommandError> { + registry + .require_registered_context_external_identifier( + browser_session, + browsing_context, + external_context, + ) + .map_err(|source| { + WebDriverBiDiNavigationCommittedSubscriptionCommandError::ContextBinding { source } + }) +} + +fn push_json_string(target: &mut String, value: &str) { + target.push('"'); + for character in value.chars() { + match character { + '"' => target.push_str("\\\""), + '\\' => target.push_str("\\\\"), + _ => target.push(character), + } + } + target.push('"'); +} + +/// Fail-closed failures while constructing or sending one typed committed-navigation subscription. +#[derive(Debug)] +pub enum WebDriverBiDiNavigationCommittedSubscriptionCommandError { + /// The requested command identifier is outside WebDriver BiDi's `js-uint` range. + CommandIdOutOfRange { + /// Rejected command identifier. + command_id: u64, + /// Largest JavaScript-safe identifier admitted by this boundary. + maximum_command_id: u64, + }, + /// The external protocol context does not name the exact registered OriginWeave context. + ContextBinding { + /// Exact typed browser-registry authority failure. + source: BrowserRegistryError, + }, + /// The bounded local correlation registry rejected the command before network I/O. + Correlation { + /// Exact typed correlation failure. + source: WebDriverBiDiCommandCorrelationError, + }, + /// Writing the already-registered command frame failed and the transport is not reusable. + FrameWrite { + /// Exact typed bounded WebSocket frame-write failure. + source: WebDriverBiDiWebSocketFrameError, + }, +} + +impl fmt::Display for WebDriverBiDiNavigationCommittedSubscriptionCommandError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::CommandIdOutOfRange { .. } => formatter.write_str( + "WebDriver BiDi navigation subscription command id is outside the js-uint range", + ), + Self::ContextBinding { .. } => formatter.write_str( + "WebDriver BiDi navigation subscription context does not match registered authority", + ), + Self::Correlation { .. } => formatter.write_str( + "WebDriver BiDi navigation subscription command correlation was rejected", + ), + Self::FrameWrite { .. } => formatter.write_str( + "WebDriver BiDi navigation subscription command frame write failed", + ), + } + } +} + +impl Error for WebDriverBiDiNavigationCommittedSubscriptionCommandError { + fn source(&self) -> Option<&(dyn Error + 'static)> { + match self { + Self::CommandIdOutOfRange { .. } => None, + Self::ContextBinding { source } => Some(source), + Self::Correlation { source } => Some(source), + Self::FrameWrite { source } => Some(source), + } + } +} diff --git a/crates/originweave-network/src/webdriver_bidi_navigation_committed_subscription_response.rs b/crates/originweave-network/src/webdriver_bidi_navigation_committed_subscription_response.rs new file mode 100644 index 000000000..961ea8ec7 --- /dev/null +++ b/crates/originweave-network/src/webdriver_bidi_navigation_committed_subscription_response.rs @@ -0,0 +1,756 @@ +use std::{error::Error, fmt}; + +use crate::{ + WebDriverBiDiCommandCorrelation, WebDriverBiDiCommandCorrelationError, + WebDriverBiDiJsonEnvelope, WebDriverBiDiJsonEnvelopeError, WebDriverBiDiJsonEnvelopeKind, + WebDriverBiDiWebSocketTextMessage, +}; + +/// Maximum decoded UTF-8 bytes retained from a WebDriver BiDi `session.Subscription` identifier. +/// +/// WebDriver BiDi defines the identifier as opaque text without a protocol size ceiling. OriginWeave +/// therefore applies a reviewed local retention bound while preserving the identifier byte-for-byte +/// for later typed subscription lifecycle work. The value is never included in `Debug` output. +pub const MAX_WEBDRIVER_BIDI_SUBSCRIPTION_IDENTIFIER_BYTES: usize = 4_096; + +/// Typed, correlated successful result of one context-scoped WebDriver BiDi `session.subscribe`. +/// +/// This value retains only the exact correlated command id and the bounded opaque subscription +/// identifier returned by the remote end. It does not expose a generic JSON result, grant event, +/// browser, policy, origin, secret, or Agent authority, or prove that any subscribed event has fired. +#[derive(Eq, PartialEq)] +pub struct WebDriverBiDiNavigationCommittedSubscriptionResult { + command_id: u64, + subscription_id: String, +} + +impl fmt::Debug for WebDriverBiDiNavigationCommittedSubscriptionResult { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("WebDriverBiDiNavigationCommittedSubscriptionResult") + .field("command_id", &self.command_id) + .field("subscription_id_len", &self.subscription_id.len()) + .finish() + } +} + +impl WebDriverBiDiNavigationCommittedSubscriptionResult { + /// Parse one bounded local-end message and consume its exact outstanding command on success. + /// + /// Common WebDriver BiDi envelope validation runs first. A successful envelope then undergoes + /// command-specific projection of the required `result.subscription` text before correlation is + /// consumed, so malformed or ambiguous success bodies cannot silently retire a command id. A + /// correlatable protocol-error response consumes its matching id and returns a typed remote + /// failure retaining only the protocol error code. Events, null-id errors, malformed envelopes, + /// and unknown ids fail closed without consuming unrelated outstanding correlation state. + pub fn parse_and_correlate( + message: &WebDriverBiDiWebSocketTextMessage, + correlation: &mut WebDriverBiDiCommandCorrelation, + ) -> Result { + let envelope = WebDriverBiDiJsonEnvelope::parse(message).map_err(|source| { + WebDriverBiDiNavigationCommittedSubscriptionResponseError::Envelope { source } + })?; + + match envelope.kind() { + WebDriverBiDiJsonEnvelopeKind::Success => { + let projected = SubscriptionProjection::parse(message.as_str())?; + let completed = correlation + .correlate_response(&envelope) + .map_err(|source| { + WebDriverBiDiNavigationCommittedSubscriptionResponseError::Correlation { + source, + } + })?; + Ok(Self { + command_id: completed.command_id(), + subscription_id: projected.subscription_id, + }) + } + WebDriverBiDiJsonEnvelopeKind::Error => { + retain_validated_error_code(envelope.error_code()).and_then(|error_code| { + let completed = correlation.correlate_response(&envelope).map_err(|source| { + WebDriverBiDiNavigationCommittedSubscriptionResponseError::Correlation { + source, + } + })?; + Err( + WebDriverBiDiNavigationCommittedSubscriptionResponseError::RemoteProtocolError { + command_id: completed.command_id(), + error_code, + }, + ) + }) + } + WebDriverBiDiJsonEnvelopeKind::Event => Err( + WebDriverBiDiNavigationCommittedSubscriptionResponseError::Correlation { + source: WebDriverBiDiCommandCorrelationError::EventIsNotResponse, + }, + ), + } + } + + /// Return the exact local command identifier consumed by this result. + #[must_use] + pub const fn command_id(&self) -> u64 { + self.command_id + } + + /// Borrow the bounded opaque subscription identifier returned by the remote end. + #[must_use] + pub fn subscription_id(&self) -> &str { + &self.subscription_id + } +} + +/// Fail-closed failures while admitting one typed WebDriver BiDi `session.subscribe` response. +#[derive(Debug, Eq, PartialEq)] +pub enum WebDriverBiDiNavigationCommittedSubscriptionResponseError { + /// Common local-end JSON envelope validation failed. + Envelope { + /// Exact common-envelope validation failure. + source: WebDriverBiDiJsonEnvelopeError, + }, + /// The successful result object omits the required `subscription` member. + MissingSubscription, + /// The successful result object's `subscription` member is not JSON text. + InvalidSubscription, + /// The successful result repeats the `subscription` member and is ambiguous. + DuplicateSubscription, + /// The decoded subscription identifier exceeds the reviewed local retention bound. + SubscriptionTooLarge { + /// Maximum decoded identifier length admitted in bytes. + maximum_bytes: usize, + }, + /// A validated success envelope could not be projected through the command-specific parser. + InvalidResultProjection, + /// Exact command-response correlation failed without consuming unrelated state. + Correlation { + /// Exact typed correlation failure. + source: WebDriverBiDiCommandCorrelationError, + }, + /// The remote end returned a correlatable WebDriver BiDi protocol error for this command. + RemoteProtocolError { + /// Exact local command identifier consumed by the protocol-error response. + command_id: u64, + /// Protocol error code retained from the already validated common envelope. + error_code: String, + }, +} + +impl fmt::Display for WebDriverBiDiNavigationCommittedSubscriptionResponseError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Envelope { .. } => { + formatter.write_str("WebDriver BiDi session.subscribe envelope is invalid") + } + Self::MissingSubscription => formatter + .write_str("WebDriver BiDi session.subscribe result is missing subscription"), + Self::InvalidSubscription => formatter + .write_str("WebDriver BiDi session.subscribe result subscription is invalid"), + Self::DuplicateSubscription => formatter.write_str( + "WebDriver BiDi session.subscribe result contains duplicate subscription", + ), + Self::SubscriptionTooLarge { .. } => formatter.write_str( + "WebDriver BiDi session.subscribe result subscription exceeds the size bound", + ), + Self::InvalidResultProjection => { + formatter.write_str("WebDriver BiDi session.subscribe result projection is invalid") + } + Self::Correlation { .. } => { + formatter.write_str("WebDriver BiDi session.subscribe response correlation failed") + } + Self::RemoteProtocolError { .. } => { + formatter.write_str("WebDriver BiDi session.subscribe returned a protocol error") + } + } + } +} + +impl Error for WebDriverBiDiNavigationCommittedSubscriptionResponseError { + fn source(&self) -> Option<&(dyn Error + 'static)> { + match self { + Self::Envelope { source } => Some(source), + Self::Correlation { source } => Some(source), + Self::MissingSubscription + | Self::InvalidSubscription + | Self::DuplicateSubscription + | Self::SubscriptionTooLarge { .. } + | Self::InvalidResultProjection + | Self::RemoteProtocolError { .. } => None, + } + } +} + +fn retain_validated_error_code( + error_code: Option<&str>, +) -> Result { + error_code.map(str::to_owned).ok_or( + WebDriverBiDiNavigationCommittedSubscriptionResponseError::Envelope { + source: WebDriverBiDiJsonEnvelopeError::MissingRequiredMember { member: "error" }, + }, + ) +} + +struct SubscriptionProjection { + subscription_id: String, +} + +impl SubscriptionProjection { + fn parse( + text: &str, + ) -> Result { + let mut cursor = ProjectionCursor::new(text); + cursor.skip_whitespace(); + if !cursor.consume_byte(b'{') { + return Err( + WebDriverBiDiNavigationCommittedSubscriptionResponseError::InvalidResultProjection, + ); + } + cursor.skip_whitespace(); + if cursor.consume_byte(b'}') { + return Err( + WebDriverBiDiNavigationCommittedSubscriptionResponseError::InvalidResultProjection, + ); + } + + loop { + cursor.skip_whitespace(); + let key = cursor.parse_string().ok_or( + WebDriverBiDiNavigationCommittedSubscriptionResponseError::InvalidResultProjection, + )?; + cursor.skip_whitespace(); + if !cursor.consume_byte(b':') { + return Err( + WebDriverBiDiNavigationCommittedSubscriptionResponseError::InvalidResultProjection, + ); + } + cursor.skip_whitespace(); + if key == "result" { + return cursor.parse_result_object(); + } + if !cursor.skip_value() { + return Err( + WebDriverBiDiNavigationCommittedSubscriptionResponseError::InvalidResultProjection, + ); + } + cursor.skip_whitespace(); + if cursor.consume_byte(b'}') { + return Err( + WebDriverBiDiNavigationCommittedSubscriptionResponseError::InvalidResultProjection, + ); + } + if !cursor.consume_byte(b',') { + return Err( + WebDriverBiDiNavigationCommittedSubscriptionResponseError::InvalidResultProjection, + ); + } + } + } +} + +struct ProjectionCursor<'a> { + input: &'a str, + index: usize, +} + +impl<'a> ProjectionCursor<'a> { + const fn new(input: &'a str) -> Self { + Self { input, index: 0 } + } + + fn current_byte(&self) -> Option { + self.input.as_bytes().get(self.index).copied() + } + + fn consume_byte(&mut self, expected: u8) -> bool { + if self.current_byte() == Some(expected) { + self.index += 1; + true + } else { + false + } + } + + fn skip_whitespace(&mut self) { + while matches!(self.current_byte(), Some(b' ' | b'\t' | b'\n' | b'\r')) { + self.index += 1; + } + } + + fn parse_result_object( + &mut self, + ) -> Result + { + if !self.consume_byte(b'{') { + return Err( + WebDriverBiDiNavigationCommittedSubscriptionResponseError::InvalidResultProjection, + ); + } + self.skip_whitespace(); + let mut subscription_id = None; + if self.consume_byte(b'}') { + return Err( + WebDriverBiDiNavigationCommittedSubscriptionResponseError::MissingSubscription, + ); + } + + loop { + self.skip_whitespace(); + let key = self.parse_string().ok_or( + WebDriverBiDiNavigationCommittedSubscriptionResponseError::InvalidResultProjection, + )?; + self.skip_whitespace(); + if !self.consume_byte(b':') { + return Err( + WebDriverBiDiNavigationCommittedSubscriptionResponseError::InvalidResultProjection, + ); + } + self.skip_whitespace(); + if key == "subscription" { + if subscription_id.is_some() { + return Err( + WebDriverBiDiNavigationCommittedSubscriptionResponseError::DuplicateSubscription, + ); + } + let parsed = self.parse_string().ok_or( + WebDriverBiDiNavigationCommittedSubscriptionResponseError::InvalidSubscription, + )?; + if parsed.len() > MAX_WEBDRIVER_BIDI_SUBSCRIPTION_IDENTIFIER_BYTES { + return Err( + WebDriverBiDiNavigationCommittedSubscriptionResponseError::SubscriptionTooLarge { + maximum_bytes: MAX_WEBDRIVER_BIDI_SUBSCRIPTION_IDENTIFIER_BYTES, + }, + ); + } + subscription_id = Some(parsed); + } else if !self.skip_value() { + return Err( + WebDriverBiDiNavigationCommittedSubscriptionResponseError::InvalidResultProjection, + ); + } + self.skip_whitespace(); + if self.consume_byte(b'}') { + break; + } + if !self.consume_byte(b',') { + return Err( + WebDriverBiDiNavigationCommittedSubscriptionResponseError::InvalidResultProjection, + ); + } + } + + Ok(SubscriptionProjection { + subscription_id: subscription_id.ok_or( + WebDriverBiDiNavigationCommittedSubscriptionResponseError::MissingSubscription, + )?, + }) + } + + fn consume_literal(&mut self, literal: &[u8]) -> bool { + let end = self.index.saturating_add(literal.len()); + if self.input.as_bytes().get(self.index..end) == Some(literal) { + self.index = end; + true + } else { + false + } + } + + fn skip_value(&mut self) -> bool { + self.skip_whitespace(); + match self.current_byte() { + Some(b'"') => self.parse_string().is_some(), + Some(b'{') => self.skip_object(), + Some(b'[') => self.skip_array(), + Some(b't') => self.consume_literal(b"true"), + Some(b'f') => self.consume_literal(b"false"), + Some(b'n') => self.consume_literal(b"null"), + Some(b'-' | b'0'..=b'9') => self.skip_number(), + _ => false, + } + } + + fn skip_object(&mut self) -> bool { + if !self.consume_byte(b'{') { + return false; + } + self.skip_whitespace(); + if self.consume_byte(b'}') { + return true; + } + loop { + self.skip_whitespace(); + if self.parse_string().is_none() { + return false; + } + self.skip_whitespace(); + if !self.consume_byte(b':') { + return false; + } + if !self.skip_value() { + return false; + } + self.skip_whitespace(); + if self.consume_byte(b'}') { + return true; + } + if !self.consume_byte(b',') { + return false; + } + } + } + + fn skip_array(&mut self) -> bool { + if !self.consume_byte(b'[') { + return false; + } + self.skip_whitespace(); + if self.consume_byte(b']') { + return true; + } + loop { + if !self.skip_value() { + return false; + } + self.skip_whitespace(); + if self.consume_byte(b']') { + return true; + } + if !self.consume_byte(b',') { + return false; + } + } + } + + fn skip_number(&mut self) -> bool { + let start = self.index; + while matches!( + self.current_byte(), + Some(b'-' | b'+' | b'.' | b'e' | b'E' | b'0'..=b'9') + ) { + self.index += 1; + } + self.index > start + } + + fn parse_string(&mut self) -> Option { + if !self.consume_byte(b'"') { + return None; + } + let mut output = String::new(); + loop { + let byte = self.current_byte()?; + match byte { + b'"' => { + self.index += 1; + return Some(output); + } + b'\\' => { + self.index += 1; + if !self.parse_escape(&mut output) { + return None; + } + } + 0x00..=0x1f => return None, + _ if byte.is_ascii() => { + output.push(char::from(byte)); + self.index += 1; + } + _ => { + let width = byte.leading_ones() as usize; + let end = self.index + width; + output.push_str(&self.input[self.index..end]); + self.index = end; + } + } + } + } + + fn parse_escape(&mut self, output: &mut String) -> bool { + let Some(escape) = self.current_byte() else { + return false; + }; + self.index += 1; + match escape { + b'"' => output.push('"'), + b'\\' => output.push('\\'), + b'/' => output.push('/'), + b'b' => output.push('\u{0008}'), + b'f' => output.push('\u{000c}'), + b'n' => output.push('\n'), + b'r' => output.push('\r'), + b't' => output.push('\t'), + b'u' => return self.parse_unicode_escape(output), + _ => return false, + } + true + } + + fn parse_unicode_escape(&mut self, output: &mut String) -> bool { + let Some(first) = self.parse_hex_u16() else { + return false; + }; + if (0xd800..=0xdbff).contains(&first) { + if !self.consume_byte(b'\\') || !self.consume_byte(b'u') { + return false; + } + let Some(second) = self.parse_hex_u16() else { + return false; + }; + if !(0xdc00..=0xdfff).contains(&second) { + return false; + } + output.push_str(&String::from_utf16_lossy(&[first, second])); + true + } else if (0xdc00..=0xdfff).contains(&first) { + false + } else { + output.push_str(&String::from_utf16_lossy(&[first])); + true + } + } + + fn parse_hex_u16(&mut self) -> Option { + let mut value = 0_u16; + for _ in 0..4 { + let byte = self.current_byte()?; + let digit = match byte { + b'0'..=b'9' => u16::from(byte - b'0'), + b'a'..=b'f' => u16::from(byte - b'a' + 10), + b'A'..=b'F' => u16::from(byte - b'A' + 10), + _ => return None, + }; + value = (value << 4) | digit; + self.index += 1; + } + Some(value) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn projection_accepts_subscription_unknown_metadata_and_escaped_keys() { + let projected = SubscriptionProjection::parse( + r#"{"meta":[null,true,false,1,-2.5e+3,{"nested":"value"}],"re\u0073ult":{"extra":{},"sub\u0073cription":"sub-\ud83d\ude80"}}"#, + ); + assert!(projected.is_ok()); + assert_eq!( + projected.ok().map(|value| value.subscription_id), + Some("sub-🚀".to_owned()) + ); + + let direct_utf8 = SubscriptionProjection::parse( + "\n\t { \r\n \"result\" : { \"subscription\" : \"구독-a\" } }", + ); + assert_eq!( + direct_utf8.ok().map(|value| value.subscription_id), + Some("구독-a".to_owned()) + ); + } + + #[test] + fn projection_rejects_missing_invalid_duplicate_and_oversized_subscription() { + let cases = [ + (r#"{"result":{}}"#.to_owned(), "missing"), + ( + r#"{"result":{"extra":1}}"#.to_owned(), + "missing after metadata", + ), + (r#"{"result":{"subscription":false}}"#.to_owned(), "invalid"), + ( + r#"{"result":{"subscription":"a","subscription":"b"}}"#.to_owned(), + "duplicate", + ), + ( + format!( + "{{\"result\":{{\"subscription\":\"{}\"}}}}", + "x".repeat(MAX_WEBDRIVER_BIDI_SUBSCRIPTION_IDENTIFIER_BYTES + 1) + ), + "oversized", + ), + (r#"{"x":1}"#.to_owned(), "missing result"), + ]; + + for (document, label) in cases { + assert!(SubscriptionProjection::parse(&document).is_err(), "{label}"); + } + } + + #[test] + fn projection_cursor_rejects_malformed_private_inputs_without_panicking() { + let malformed = [ + "", + "[]", + "{}", + r#"{"x":}"#, + r#"{"x" 1}"#, + r#"{"x":1 ?}"#, + r#"{?}"#, + r#"{"result":[]}"#, + r#"{"result":{?}}"#, + r#"{"result":{"subscription" "x"}}"#, + r#"{"result":{"subscription":"x" "extra":1}}"#, + r#"{"result":{"subscription":"x","extra":?}}"#, + r#"{"result":{"subscription":"\uD800"}}"#, + r#"{"result":{"subscription":"\q"}}"#, + ]; + for document in malformed { + assert!( + SubscriptionProjection::parse(document).is_err(), + "{document}" + ); + } + } + + #[test] + fn projection_cursor_defensive_helpers_cover_hostile_dispatch_edges() { + let mut object = ProjectionCursor::new("[]"); + assert!(!object.skip_object()); + let mut object = ProjectionCursor::new("{}"); + assert!(object.skip_object()); + let mut object = ProjectionCursor::new("{?}"); + assert!(!object.skip_object()); + let mut object = ProjectionCursor::new(r#"{"x" 1}"#); + assert!(!object.skip_object()); + let mut object = ProjectionCursor::new(r#"{"x":?}"#); + assert!(!object.skip_object()); + let mut object = ProjectionCursor::new(r#"{"x":1 ?}"#); + assert!(!object.skip_object()); + let mut object = ProjectionCursor::new(r#"{"x":1,"y":2}"#); + assert!(object.skip_object()); + + let mut array = ProjectionCursor::new("{}"); + assert!(!array.skip_array()); + let mut array = ProjectionCursor::new("[]"); + assert!(array.skip_array()); + let mut array = ProjectionCursor::new("[?]"); + assert!(!array.skip_array()); + let mut array = ProjectionCursor::new("[1 ?]"); + assert!(!array.skip_array()); + + for document in [r#""x""#, "{}", "[]", "true", "false", "null", "-2.5e+3"] { + let mut value = ProjectionCursor::new(document); + assert!(value.skip_value(), "{document}"); + } + let mut value = ProjectionCursor::new("?"); + assert!(!value.skip_value()); + let mut literal = ProjectionCursor::new("tru?"); + assert!(!literal.consume_literal(b"true")); + + let mut number = ProjectionCursor::new("x"); + assert!(!number.skip_number()); + let mut number = ProjectionCursor::new("+1"); + assert!(number.skip_number()); + + let mut string = ProjectionCursor::new("x"); + assert!(string.parse_string().is_none()); + let mut string = ProjectionCursor::new("\"unterminated"); + assert!(string.parse_string().is_none()); + let mut string = ProjectionCursor::new("\"\u{0001}\""); + assert!(string.parse_string().is_none()); + let mut string = ProjectionCursor::new("\"é\""); + assert_eq!(string.parse_string().as_deref(), Some("é")); + + let mut output = String::new(); + let mut escape = ProjectionCursor::new(""); + assert!(!escape.parse_escape(&mut output)); + for sequence in ["\"", "\\", "/", "b", "f", "n", "r", "t"] { + let mut output = String::new(); + let mut escape = ProjectionCursor::new(sequence); + assert!(escape.parse_escape(&mut output), "{sequence:?}"); + } + let mut output = String::new(); + let mut escape = ProjectionCursor::new("q"); + assert!(!escape.parse_escape(&mut output)); + + for sequence in ["0000", "aBcD", "Ff09"] { + let mut hex = ProjectionCursor::new(sequence); + assert!(hex.parse_hex_u16().is_some()); + } + let mut hex = ProjectionCursor::new("xyz1"); + assert!(hex.parse_hex_u16().is_none()); + let mut hex = ProjectionCursor::new("0"); + assert!(hex.parse_hex_u16().is_none()); + + let unicode_cases = [ + ("0041", true), + ("d83d\\ude80", true), + ("d83d", false), + ("d83d\\x", false), + ("d83d\\u0", false), + ("d83d\\u0041", false), + ("dc00", false), + ("zzzz", false), + ]; + for (sequence, expected) in unicode_cases { + let mut output = String::new(); + let mut unicode = ProjectionCursor::new(sequence); + assert_eq!(unicode.parse_unicode_escape(&mut output), expected); + } + } + + #[test] + fn response_errors_have_stable_messages_and_typed_sources() { + let envelope = WebDriverBiDiNavigationCommittedSubscriptionResponseError::Envelope { + source: WebDriverBiDiJsonEnvelopeError::InvalidJson, + }; + assert_eq!( + envelope.to_string(), + "WebDriver BiDi session.subscribe envelope is invalid" + ); + assert!(envelope.source().is_some()); + + let correlation = WebDriverBiDiNavigationCommittedSubscriptionResponseError::Correlation { + source: WebDriverBiDiCommandCorrelationError::CommandNotOutstanding, + }; + assert_eq!( + correlation.to_string(), + "WebDriver BiDi session.subscribe response correlation failed" + ); + assert!(correlation.source().is_some()); + + let source_free = [ + WebDriverBiDiNavigationCommittedSubscriptionResponseError::MissingSubscription, + WebDriverBiDiNavigationCommittedSubscriptionResponseError::InvalidSubscription, + WebDriverBiDiNavigationCommittedSubscriptionResponseError::DuplicateSubscription, + WebDriverBiDiNavigationCommittedSubscriptionResponseError::SubscriptionTooLarge { + maximum_bytes: MAX_WEBDRIVER_BIDI_SUBSCRIPTION_IDENTIFIER_BYTES, + }, + WebDriverBiDiNavigationCommittedSubscriptionResponseError::InvalidResultProjection, + WebDriverBiDiNavigationCommittedSubscriptionResponseError::RemoteProtocolError { + command_id: 7, + error_code: "invalid argument".to_owned(), + }, + ]; + for error in source_free { + assert!(!error.to_string().is_empty()); + assert!(error.source().is_none()); + } + } + + #[test] + fn result_debug_redacts_opaque_subscription_identifier() { + let result = WebDriverBiDiNavigationCommittedSubscriptionResult { + command_id: 7, + subscription_id: "sensitive-subscription".to_owned(), + }; + let debug = format!("{result:?}"); + assert!(debug.contains("command_id")); + assert!(debug.contains("subscription_id_len")); + assert!(!debug.contains("sensitive-subscription")); + assert_eq!(result.command_id(), 7); + assert_eq!(result.subscription_id(), "sensitive-subscription"); + } + + #[test] + fn retain_error_code_fails_closed_when_common_invariant_is_absent() { + assert_eq!( + retain_validated_error_code(Some("invalid argument")).as_deref(), + Ok("invalid argument") + ); + assert!(retain_validated_error_code(None).is_err()); + } +} diff --git a/crates/originweave-network/tests/webdriver_bidi_navigation_committed_subscription.rs b/crates/originweave-network/tests/webdriver_bidi_navigation_committed_subscription.rs new file mode 100644 index 000000000..4b5a7ad39 --- /dev/null +++ b/crates/originweave-network/tests/webdriver_bidi_navigation_committed_subscription.rs @@ -0,0 +1,218 @@ +use std::{ + error::Error, + io::{self, Read, Write}, + net::{TcpListener, TcpStream}, + thread, + time::Duration, +}; + +use originweave_core::{BrowserAuthorityRegistry, WebDriverBiDiWebSocketEndpoint}; +use originweave_network::{ + MAX_WEBDRIVER_BIDI_JS_UINT, WebDriverBiDiCommandCorrelation, + WebDriverBiDiNavigationCommittedSubscriptionCommand, + WebDriverBiDiNavigationCommittedSubscriptionResult, WebDriverBiDiTcpConnectionPlan, + WebDriverBiDiWebSocketClientKey, WebDriverBiDiWebSocketHandshakePlan, + WebDriverBiDiWebSocketMaskKey, WebDriverBiDiWebSocketMessageAssembler, + WebDriverBiDiWebSocketMessageAssembly, +}; + +const SESSION_ID: &str = "01234567-89ab-cdef-0123-456789abcdef"; +const CONTEXT_ID: &str = "context-\"a\\b"; +const RFC6455_SAMPLE_KEY: &str = "dGhlIHNhbXBsZSBub25jZQ=="; +const OPENING_RESPONSE: &[u8] = b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=\r\n\r\n"; +const SUBSCRIBE_RESPONSE: &[u8] = + r#"{"type":"success","id":7,"result":{"subscription":"subscription-a","vendorNote":"café"}}"# + .as_bytes(); + +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_text_frame(stream: &mut TcpStream) -> io::Result> { + 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 length = match header[1] & 0x7f { + length @ 0..=125 => usize::from(length), + 126 => { + let mut extended = [0_u8; 2]; + stream.read_exact(&mut extended)?; + usize::from(u16::from_be_bytes(extended)) + } + _ => { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "test command unexpectedly required 64-bit framing", + )); + } + }; + let mut mask = [0_u8; 4]; + stream.read_exact(&mut mask)?; + let mut payload = vec![0_u8; length]; + stream.read_exact(&mut payload)?; + for (index, byte) in payload.iter_mut().enumerate() { + *byte ^= mask[index % mask.len()]; + } + Ok(payload) +} + +#[test] +fn navigation_committed_subscription_round_trips_on_the_registered_context() +-> 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(OPENING_RESPONSE)?; + let command = read_masked_text_frame(&mut stream)?; + if command + != br#"{"id":7,"method":"session.subscribe","params":{"events":["browsingContext.navigationCommitted"],"contexts":["context-\"a\\b"]}}"# + { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!( + "unexpected session.subscribe command: {}", + String::from_utf8_lossy(&command) + ), + )); + } + stream.write_all(&[0x81, SUBSCRIBE_RESPONSE.len() as u8])?; + stream.write_all(SUBSCRIBE_RESPONSE) + }); + + let mut registry = BrowserAuthorityRegistry::new(); + let session = registry.register_session(SESSION_ID)?; + let context = registry.register_context(session, CONTEXT_ID)?; + + let endpoint = format!("ws://{local_addr}/session/{SESSION_ID}"); + let target = WebDriverBiDiWebSocketEndpoint::new(&endpoint)? + .correlate_session_id(SESSION_ID)? + .into_explicit_connect_target()?; + let connection = + WebDriverBiDiTcpConnectionPlan::new(target, Duration::from_secs(1), 1)?.connect()?; + let key = WebDriverBiDiWebSocketClientKey::new(RFC6455_SAMPLE_KEY)?; + let established = WebDriverBiDiWebSocketHandshakePlan::new(connection, key)? + .write_opening_request(Duration::from_millis(500))? + .read_opening_response(Duration::from_millis(500))?; + + let mut correlation = WebDriverBiDiCommandCorrelation::new(); + let command = WebDriverBiDiNavigationCommittedSubscriptionCommand::new( + 7, ®istry, session, context, CONTEXT_ID, + )?; + assert_eq!(command.command_id(), 7); + assert_eq!(command.browser_session(), session); + assert_eq!(command.browsing_context(), context); + assert_eq!(command.external_context(), CONTEXT_ID); + let established = command.send( + ®istry, + established, + &mut correlation, + WebDriverBiDiWebSocketMaskKey::new([1, 2, 3, 4]), + Duration::from_millis(500), + )?; + assert_eq!(correlation.outstanding_count(), 1); + + let (_established, frame) = established.read_frame(Duration::from_millis(500))?; + let mut assembler = WebDriverBiDiWebSocketMessageAssembler::new(); + let text = match assembler.push_frame(frame)? { + WebDriverBiDiWebSocketMessageAssembly::Text(text) => text, + other => { + return Err(io::Error::other(format!( + "session.subscribe response produced unexpected assembly state: {other:?}" + )) + .into()); + } + }; + let result = WebDriverBiDiNavigationCommittedSubscriptionResult::parse_and_correlate( + &text, + &mut correlation, + )?; + assert_eq!(result.command_id(), 7); + assert_eq!(result.subscription_id(), "subscription-a"); + assert_eq!(correlation.outstanding_count(), 0); + + server + .join() + .map_err(|_| io::Error::other("session.subscribe command test server panicked"))??; + Ok(()) +} + +#[test] +fn subscription_constructor_rejects_out_of_range_command_id_without_source() +-> Result<(), Box> { + let mut registry = BrowserAuthorityRegistry::new(); + let session = registry.register_session(SESSION_ID)?; + let context = registry.register_context(session, CONTEXT_ID)?; + + let result = WebDriverBiDiNavigationCommittedSubscriptionCommand::new( + MAX_WEBDRIVER_BIDI_JS_UINT + 1, + ®istry, + session, + context, + CONTEXT_ID, + ); + let error = match result { + Ok(_) => { + return Err(io::Error::other( + "out-of-range session.subscribe command id was unexpectedly accepted", + ) + .into()); + } + Err(error) => error, + }; + assert_eq!( + error.to_string(), + "WebDriver BiDi navigation subscription command id is outside the js-uint range" + ); + assert!(error.source().is_none()); + Ok(()) +} + +#[test] +fn subscription_constructor_rejects_mismatched_registered_context() -> Result<(), Box> { + let mut registry = BrowserAuthorityRegistry::new(); + let session = registry.register_session(SESSION_ID)?; + let context = registry.register_context(session, CONTEXT_ID)?; + + let result = WebDriverBiDiNavigationCommittedSubscriptionCommand::new( + 7, + ®istry, + session, + context, + "different-context", + ); + let error = match result { + Ok(_) => { + return Err(io::Error::other( + "mismatched session.subscribe context was unexpectedly accepted", + ) + .into()); + } + Err(error) => error, + }; + assert_eq!( + error.to_string(), + "WebDriver BiDi navigation subscription context does not match registered authority" + ); + assert!(error.source().is_some()); + Ok(()) +} diff --git a/crates/originweave-network/tests/webdriver_bidi_navigation_committed_subscription_failures.rs b/crates/originweave-network/tests/webdriver_bidi_navigation_committed_subscription_failures.rs new file mode 100644 index 000000000..2d85a62d8 --- /dev/null +++ b/crates/originweave-network/tests/webdriver_bidi_navigation_committed_subscription_failures.rs @@ -0,0 +1,223 @@ +use std::{ + error::Error, + io::{self, Read, Write}, + net::{SocketAddr, TcpListener, TcpStream}, + thread, + time::Duration, +}; + +use originweave_core::{BrowserAuthorityRegistry, WebDriverBiDiWebSocketEndpoint}; +use originweave_network::{ + WebDriverBiDiCommandCorrelation, WebDriverBiDiNavigationCommittedSubscriptionCommand, + WebDriverBiDiTcpConnectionPlan, WebDriverBiDiWebSocketClientKey, + WebDriverBiDiWebSocketEstablished, WebDriverBiDiWebSocketHandshakePlan, + WebDriverBiDiWebSocketMaskKey, +}; + +const SESSION_ID: &str = "01234567-89ab-cdef-0123-456789abcdef"; +const RFC6455_SAMPLE_KEY: &str = "dGhlIHNhbXBsZSBub25jZQ=="; +const OPENING_RESPONSE: &[u8] = b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=\r\n\r\n"; + +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 require_no_client_command(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, + "subscription command was written despite local rejection", + )), + Err(source) + if matches!( + source.kind(), + io::ErrorKind::ConnectionReset | io::ErrorKind::ConnectionAborted + ) => + { + Ok(()) + } + Err(source) => Err(source), + } +} + +fn spawn_no_command_server(listener: TcpListener) -> thread::JoinHandle> { + thread::spawn(move || { + let (mut stream, _) = listener.accept()?; + read_opening_request(&mut stream)?; + stream.write_all(OPENING_RESPONSE)?; + require_no_client_command(&mut stream) + }) +} + +fn establish_websocket( + local_addr: SocketAddr, +) -> Result> { + let endpoint = format!("ws://{local_addr}/session/{SESSION_ID}"); + let target = WebDriverBiDiWebSocketEndpoint::new(&endpoint)? + .correlate_session_id(SESSION_ID)? + .into_explicit_connect_target()?; + let connection = + WebDriverBiDiTcpConnectionPlan::new(target, Duration::from_secs(1), 1)?.connect()?; + let key = WebDriverBiDiWebSocketClientKey::new(RFC6455_SAMPLE_KEY)?; + Ok(WebDriverBiDiWebSocketHandshakePlan::new(connection, key)? + .write_opening_request(Duration::from_millis(500))? + .read_opening_response(Duration::from_millis(500))?) +} + +#[test] +fn retired_context_is_rejected_before_correlation_or_command_write() -> Result<(), Box> { + let listener = TcpListener::bind(("127.0.0.1", 0))?; + let local_addr = listener.local_addr()?; + let server = spawn_no_command_server(listener); + + let mut registry = BrowserAuthorityRegistry::new(); + let session = registry.register_session(SESSION_ID)?; + let context = registry.register_context(session, "context-a")?; + let command = WebDriverBiDiNavigationCommittedSubscriptionCommand::new( + 7, + ®istry, + session, + context, + "context-a", + )?; + let established = establish_websocket(local_addr)?; + registry.remove_context(context)?; + + let mut correlation = WebDriverBiDiCommandCorrelation::new(); + let result = command.send( + ®istry, + established, + &mut correlation, + WebDriverBiDiWebSocketMaskKey::new([1, 2, 3, 4]), + Duration::from_millis(500), + ); + let error = match result { + Ok(_) => { + return Err(io::Error::other("retired context unexpectedly sent subscription").into()); + } + Err(error) => error, + }; + assert_eq!( + error.to_string(), + "WebDriver BiDi navigation subscription context does not match registered authority" + ); + assert!(error.source().is_some()); + assert_eq!(correlation.outstanding_count(), 0); + + server + .join() + .map_err(|_| io::Error::other("retired-context test server panicked"))??; + Ok(()) +} + +#[test] +fn duplicate_command_id_is_rejected_before_command_write() -> Result<(), Box> { + let listener = TcpListener::bind(("127.0.0.1", 0))?; + let local_addr = listener.local_addr()?; + let server = spawn_no_command_server(listener); + + let mut registry = BrowserAuthorityRegistry::new(); + let session = registry.register_session(SESSION_ID)?; + let context = registry.register_context(session, "context-a")?; + let command = WebDriverBiDiNavigationCommittedSubscriptionCommand::new( + 7, + ®istry, + session, + context, + "context-a", + )?; + let established = establish_websocket(local_addr)?; + + let mut correlation = WebDriverBiDiCommandCorrelation::new(); + correlation.register_command(7)?; + let result = command.send( + ®istry, + established, + &mut correlation, + WebDriverBiDiWebSocketMaskKey::new([1, 2, 3, 4]), + Duration::from_millis(500), + ); + let error = match result { + Ok(_) => { + return Err( + io::Error::other("duplicate command id unexpectedly sent subscription").into(), + ); + } + Err(error) => error, + }; + assert_eq!( + error.to_string(), + "WebDriver BiDi navigation subscription command correlation was rejected" + ); + assert!(error.source().is_some()); + assert_eq!(correlation.outstanding_count(), 1); + + server + .join() + .map_err(|_| io::Error::other("duplicate-command test server panicked"))??; + Ok(()) +} + +#[test] +fn invalid_frame_timeout_consumes_transport_and_retains_correlation() -> Result<(), Box> +{ + let listener = TcpListener::bind(("127.0.0.1", 0))?; + let local_addr = listener.local_addr()?; + let server = spawn_no_command_server(listener); + + let mut registry = BrowserAuthorityRegistry::new(); + let session = registry.register_session(SESSION_ID)?; + let context = registry.register_context(session, "context-a")?; + let command = WebDriverBiDiNavigationCommittedSubscriptionCommand::new( + 7, + ®istry, + session, + context, + "context-a", + )?; + let established = establish_websocket(local_addr)?; + + let mut correlation = WebDriverBiDiCommandCorrelation::new(); + let result = command.send( + ®istry, + established, + &mut correlation, + WebDriverBiDiWebSocketMaskKey::new([1, 2, 3, 4]), + Duration::ZERO, + ); + let error = match result { + Ok(_) => { + return Err( + io::Error::other("zero frame timeout unexpectedly sent subscription").into(), + ); + } + Err(error) => error, + }; + assert_eq!( + error.to_string(), + "WebDriver BiDi navigation subscription command frame write failed" + ); + assert!(error.source().is_some()); + assert_eq!(correlation.outstanding_count(), 1); + + server + .join() + .map_err(|_| io::Error::other("frame-write test server panicked"))??; + Ok(()) +} diff --git a/crates/originweave-network/tests/webdriver_bidi_navigation_committed_subscription_response_failures.rs b/crates/originweave-network/tests/webdriver_bidi_navigation_committed_subscription_response_failures.rs new file mode 100644 index 000000000..82dea5dfe --- /dev/null +++ b/crates/originweave-network/tests/webdriver_bidi_navigation_committed_subscription_response_failures.rs @@ -0,0 +1,212 @@ +use std::{ + error::Error, + io::{self, Read, Write}, + net::{TcpListener, TcpStream}, + thread, + time::Duration, +}; + +use originweave_core::WebDriverBiDiWebSocketEndpoint; +use originweave_network::{ + WebDriverBiDiCommandCorrelation, WebDriverBiDiCommandCorrelationError, + WebDriverBiDiJsonEnvelopeError, WebDriverBiDiNavigationCommittedSubscriptionResponseError, + WebDriverBiDiNavigationCommittedSubscriptionResult, WebDriverBiDiTcpConnectionPlan, + WebDriverBiDiWebSocketClientKey, WebDriverBiDiWebSocketHandshakePlan, + WebDriverBiDiWebSocketMessageAssembler, WebDriverBiDiWebSocketMessageAssembly, + WebDriverBiDiWebSocketTextMessage, +}; + +const SESSION_ID: &str = "01234567-89ab-cdef-0123-456789abcdef"; +const RFC6455_SAMPLE_KEY: &str = "dGhlIHNhbXBsZSBub25jZQ=="; +const OPENING_RESPONSE: &[u8] = b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=\r\n\r\n"; +const MALFORMED_SUCCESS_RESPONSE: &[u8] = br#"{"type":"success","id":7,"result":"#; +const MISSING_SUBSCRIPTION_RESPONSE: &[u8] = br#"{"type":"success","id":7,"result":{"extra":1}}"#; +const UNKNOWN_SUCCESS_RESPONSE: &[u8] = + br#"{"type":"success","id":8,"result":{"subscription":"subscription-b"}}"#; +const MATCHED_ERROR_RESPONSE: &[u8] = + br#"{"type":"error","id":7,"error":"invalid argument","message":"denied"}"#; +const UNKNOWN_ERROR_RESPONSE: &[u8] = + br#"{"type":"error","id":8,"error":"invalid argument","message":"denied"}"#; +const NAVIGATION_EVENT: &[u8] = + br#"{"type":"event","method":"browsingContext.navigationCommitted","params":{}}"#; + +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 write_unmasked_text_frame(stream: &mut TcpStream, document: &[u8]) -> io::Result<()> { + if document.len() <= 125 { + let length = u8::try_from(document.len()).map_err(|_| { + io::Error::new(io::ErrorKind::InvalidData, "short frame length exceeds u8") + })?; + stream.write_all(&[0x81, length])?; + } else { + let length = u16::try_from(document.len()).map_err(|_| { + io::Error::new( + io::ErrorKind::InvalidData, + "test JSON document exceeds two-byte frame length", + ) + })?; + stream.write_all(&[0x81, 126])?; + stream.write_all(&length.to_be_bytes())?; + } + stream.write_all(document) +} + +fn read_text_over_loopback( + document: &'static [u8], +) -> 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(OPENING_RESPONSE)?; + write_unmasked_text_frame(&mut stream, document) + }); + + let endpoint = format!("ws://{local_addr}/session/{SESSION_ID}"); + let target = WebDriverBiDiWebSocketEndpoint::new(&endpoint)? + .correlate_session_id(SESSION_ID)? + .into_explicit_connect_target()?; + let connection = + WebDriverBiDiTcpConnectionPlan::new(target, Duration::from_secs(1), 1)?.connect()?; + let key = WebDriverBiDiWebSocketClientKey::new(RFC6455_SAMPLE_KEY)?; + let established = WebDriverBiDiWebSocketHandshakePlan::new(connection, key)? + .write_opening_request(Duration::from_millis(500))? + .read_opening_response(Duration::from_millis(500))?; + let (_established, frame) = established.read_frame(Duration::from_millis(500))?; + + let mut assembler = WebDriverBiDiWebSocketMessageAssembler::new(); + let text = match assembler.push_frame(frame)? { + WebDriverBiDiWebSocketMessageAssembly::Text(text) => text, + other => { + return Err(io::Error::other(format!( + "subscription response produced unexpected assembly state: {other:?}" + )) + .into()); + } + }; + + server + .join() + .map_err(|_| io::Error::other("subscription response test server panicked"))??; + Ok(text) +} + +#[test] +fn malformed_and_invalid_success_responses_preserve_outstanding_correlation() +-> Result<(), Box> { + let mut correlation = WebDriverBiDiCommandCorrelation::new(); + correlation.register_command(7)?; + + let malformed = read_text_over_loopback(MALFORMED_SUCCESS_RESPONSE)?; + assert_eq!( + WebDriverBiDiNavigationCommittedSubscriptionResult::parse_and_correlate( + &malformed, + &mut correlation, + ), + Err( + WebDriverBiDiNavigationCommittedSubscriptionResponseError::Envelope { + source: WebDriverBiDiJsonEnvelopeError::InvalidJson, + } + ) + ); + assert_eq!(correlation.outstanding_count(), 1); + + let missing = read_text_over_loopback(MISSING_SUBSCRIPTION_RESPONSE)?; + assert_eq!( + WebDriverBiDiNavigationCommittedSubscriptionResult::parse_and_correlate( + &missing, + &mut correlation, + ), + Err(WebDriverBiDiNavigationCommittedSubscriptionResponseError::MissingSubscription) + ); + assert_eq!(correlation.outstanding_count(), 1); + + let unknown = read_text_over_loopback(UNKNOWN_SUCCESS_RESPONSE)?; + assert_eq!( + WebDriverBiDiNavigationCommittedSubscriptionResult::parse_and_correlate( + &unknown, + &mut correlation, + ), + Err( + WebDriverBiDiNavigationCommittedSubscriptionResponseError::Correlation { + source: WebDriverBiDiCommandCorrelationError::CommandNotOutstanding, + } + ) + ); + assert_eq!(correlation.outstanding_count(), 1); + Ok(()) +} + +#[test] +fn protocol_error_consumes_only_its_exact_outstanding_command() -> Result<(), Box> { + let mut correlation = WebDriverBiDiCommandCorrelation::new(); + correlation.register_command(7)?; + + let unknown = read_text_over_loopback(UNKNOWN_ERROR_RESPONSE)?; + assert_eq!( + WebDriverBiDiNavigationCommittedSubscriptionResult::parse_and_correlate( + &unknown, + &mut correlation, + ), + Err( + WebDriverBiDiNavigationCommittedSubscriptionResponseError::Correlation { + source: WebDriverBiDiCommandCorrelationError::CommandNotOutstanding, + } + ) + ); + assert_eq!(correlation.outstanding_count(), 1); + + let matched = read_text_over_loopback(MATCHED_ERROR_RESPONSE)?; + assert_eq!( + WebDriverBiDiNavigationCommittedSubscriptionResult::parse_and_correlate( + &matched, + &mut correlation, + ), + Err( + WebDriverBiDiNavigationCommittedSubscriptionResponseError::RemoteProtocolError { + command_id: 7, + error_code: "invalid argument".to_owned(), + } + ) + ); + assert_eq!(correlation.outstanding_count(), 0); + Ok(()) +} + +#[test] +fn event_response_is_rejected_without_consuming_outstanding_command() -> Result<(), Box> +{ + let mut correlation = WebDriverBiDiCommandCorrelation::new(); + correlation.register_command(7)?; + let event = read_text_over_loopback(NAVIGATION_EVENT)?; + + assert_eq!( + WebDriverBiDiNavigationCommittedSubscriptionResult::parse_and_correlate( + &event, + &mut correlation, + ), + Err( + WebDriverBiDiNavigationCommittedSubscriptionResponseError::Correlation { + source: WebDriverBiDiCommandCorrelationError::EventIsNotResponse, + } + ) + ); + assert_eq!(correlation.outstanding_count(), 1); + Ok(()) +}