From 519c04e18ffe9be9eaa78f0484e79d935ff8b1bb Mon Sep 17 00:00:00 2001 From: sechmachine <97589681+sechmachine727@users.noreply.github.com> Date: Wed, 12 Aug 2026 22:03:10 +0700 Subject: [PATCH] feat(core): run bounded real gateway sessions --- core/src/abi.rs | 438 +++++++++++++++++++---- core/src/error.rs | 2 + core/src/input.rs | 7 +- core/src/media.rs | 18 +- core/src/transport.rs | 607 +++++++++++++++++++++++++++++++- core/src/wire.rs | 61 +++- core/tests/abi_contract.rs | 187 ++++++++-- core/tests/ffi/abi_smoke.c | 85 +++-- core/tests/gateway_oracle.rs | 353 +++++++++++++++++-- core/tests/protocol_fixtures.rs | 18 + 10 files changed, 1598 insertions(+), 178 deletions(-) diff --git a/core/src/abi.rs b/core/src/abi.rs index 87c5a61..274e1bc 100644 --- a/core/src/abi.rs +++ b/core/src/abi.rs @@ -4,16 +4,20 @@ use std::ffi::c_void; use std::mem::size_of; use std::panic::{catch_unwind, AssertUnwindSafe}; use std::ptr::{self, NonNull}; -use std::sync::atomic::{AtomicBool, AtomicU64, AtomicU8, AtomicUsize, Ordering}; -use std::sync::mpsc::{self, Receiver, RecvTimeoutError, Sender, SyncSender, TrySendError}; +use std::sync::atomic::{AtomicBool, AtomicU8, AtomicUsize, Ordering}; +use std::sync::mpsc::{self, Receiver, RecvTimeoutError, Sender}; use std::sync::{Arc, Mutex, MutexGuard, OnceLock, TryLockError}; use std::thread::{self, JoinHandle}; use std::time::{Duration, Instant}; use crate::error::CoreError; -use crate::input::{encode_input, ControllerState, InputEvent}; -use crate::session::SessionCore; -use crate::wire::{ConnectionManifest, NativeTunnelCredential}; +use crate::input::{encode_feedback, encode_input, ControllerState, FeedbackEvent, InputEvent}; +use crate::media::MediaChannel; +use crate::transport::{ + self, bounded_session_events, AdmissionSigner, Cancellation, SessionCommand, SessionEvent, + Signers, TlsEd25519Signer, +}; +use crate::wire::{now_utc, ConnectionManifest, NativeTunnelCredential}; const ABI_V1: u32 = 1; const OK: u32 = 0; @@ -22,6 +26,9 @@ const INVALID_STATE: u32 = 2; const UNSUPPORTED_ABI: u32 = 3; const AUTHORITY_REJECTED: u32 = 4; const TLS: u32 = 5; +const TRANSPORT: u32 = 6; +const PROTOCOL: u32 = 7; +const EXPIRED: u32 = 8; const QUEUE_FULL: u32 = 9; const CANCELLED: u32 = 10; const REENTRANT: u32 = 11; @@ -41,7 +48,6 @@ const INPUT_ABSOLUTE_MOUSE: u32 = 6; const INPUT_SCROLL: u32 = 7; const MAX_CONNECT_BYTES: usize = 1024 * 1024; -const CALLBACK_QUEUE_CAPACITY: usize = 64; const CONTROL_QUEUE_CAPACITY: usize = 64; const TRANSITION_OPEN: u8 = 0; @@ -218,12 +224,15 @@ struct Callbacks { enum CallbackEvent { State(u32), + Error(u32), + Media(crate::media::EncodedUnit), + Control { kind: u32, payload: Vec }, } struct SessionState { - session: SessionCore, lifecycle: Lifecycle, - idr_queue: usize, + command_tx: Option>, + features: Vec, } #[derive(Clone, Copy, Eq, PartialEq)] @@ -239,17 +248,20 @@ struct CoreInner { id: usize, state: Mutex, callbacks: Callbacks, - callback_tx: SyncSender, + callback_tx: Sender, callbacks_closed: Arc, + callback_gate: Arc>, stop_tx: Sender<()>, worker_done: Mutex>, worker: Mutex>>, + session_cancellation: Cancellation, + session_done: Mutex>>, + session_worker: Mutex>>, in_flight: AtomicUsize, transition: AtomicU8, cancellation_emitted: AtomicBool, freeing: AtomicBool, destroying: AtomicBool, - dropped_callbacks: AtomicU64, } #[repr(C)] @@ -348,8 +360,14 @@ fn pause_connect_transition( fn status(error: CoreError) -> u32 { match error { + CoreError::AuthorityRejected => AUTHORITY_REJECTED, + CoreError::Tls => TLS, + CoreError::Transport => TRANSPORT, + CoreError::Protocol => PROTOCOL, + CoreError::Expired => EXPIRED, CoreError::QueueFull => QUEUE_FULL, CoreError::Cancelled => CANCELLED, + CoreError::Internal => INTERNAL, _ => INVALID_ARGUMENT, } } @@ -434,9 +452,111 @@ fn enqueue_callback_under_state_lock(inner: &CoreInner, event: CallbackEvent) { if inner.callbacks_closed.load(Ordering::Acquire) { return; } - if let Err(TrySendError::Full(_)) = inner.callback_tx.try_send(event) { - inner.dropped_callbacks.fetch_add(1, Ordering::Relaxed); + let _ = inner.callback_tx.send(event); +} + +fn dispatch_callback( + id: usize, + callbacks: Callbacks, + event: CallbackEvent, + callbacks_closed: &AtomicBool, + callback_gate: &Mutex<()>, +) { + let _gate = lock(callback_gate); + if callbacks_closed.load(Ordering::Acquire) { + return; } + match event { + CallbackEvent::State(state) => { + if let Some(callback) = callbacks.on_state { + let event = StateEvent { + struct_size: u32::try_from(size_of::()).expect("ABI size fits"), + abi_version: ABI_V1, + state, + reason: 0, + }; + let _mode = CallbackModeGuard::enter(CallbackMode::Event(id)); + // Invariant: copied callback/context remain caller-owned and valid until destroy + // succeeds; event is readable for this synchronous callback only. + unsafe { callback(callbacks.context as *mut c_void, &raw const event) }; + } + } + CallbackEvent::Error(code) => { + if let Some(callback) = callbacks.on_error { + let event = ErrorEvent { + struct_size: u32::try_from(size_of::()).expect("ABI size fits"), + abi_version: ABI_V1, + code, + retryable: 0, + phase: 3, + reserved: 0, + }; + let _mode = CallbackModeGuard::enter(CallbackMode::Event(id)); + unsafe { callback(callbacks.context as *mut c_void, &raw const event) }; + } + } + CallbackEvent::Media(unit) => { + if let Some(callback) = callbacks.on_media { + let event = MediaEvent { + struct_size: u32::try_from(size_of::()).expect("ABI size fits"), + abi_version: ABI_V1, + channel: match unit.channel { + MediaChannel::Video => 10, + MediaChannel::Audio => 11, + }, + sequence: unit.sequence, + timestamp_ms: unit.timestamp_ms, + encoded_unit: BytesView { + data: unit.payload.as_ptr(), + length: unit.payload.len(), + }, + }; + let _mode = CallbackModeGuard::enter(CallbackMode::Event(id)); + unsafe { callback(callbacks.context as *mut c_void, &raw const event) }; + } + } + CallbackEvent::Control { kind, payload } => { + if let Some(callback) = callbacks.on_control { + let event = ControlEvent { + struct_size: u32::try_from(size_of::()).expect("ABI size fits"), + abi_version: ABI_V1, + kind, + reserved: 0, + payload: BytesView { + data: payload.as_ptr(), + length: payload.len(), + }, + }; + let _mode = CallbackModeGuard::enter(CallbackMode::Event(id)); + unsafe { callback(callbacks.context as *mut c_void, &raw const event) }; + } + } + } +} + +fn dispatch_stats( + id: usize, + callbacks: Callbacks, + dropped_media_units: u64, + callbacks_closed: &AtomicBool, + callback_gate: &Mutex<()>, +) { + let Some(callback) = callbacks.on_stats else { + return; + }; + let _gate = lock(callback_gate); + if callbacks_closed.load(Ordering::Acquire) { + return; + } + let event = StatsEvent { + struct_size: u32::try_from(size_of::()).expect("ABI size fits"), + abi_version: ABI_V1, + dropped_callbacks: 0, + dropped_media_units, + dropped_input_events: 0, + }; + let _mode = CallbackModeGuard::enter(CallbackMode::Event(id)); + unsafe { callback(callbacks.context as *mut c_void, &raw const event) }; } #[allow(clippy::needless_pass_by_value)] @@ -454,24 +574,7 @@ fn callback_worker( break; } match receiver.recv_timeout(Duration::from_millis(2)) { - Ok(CallbackEvent::State(state)) => { - if let Some(callback) = callbacks.on_state { - let _gate = lock(&callback_gate); - if callbacks_closed.load(Ordering::Acquire) { - continue; - } - let event = StateEvent { - struct_size: u32::try_from(size_of::()).expect("ABI size fits"), - abi_version: ABI_V1, - state, - reason: 0, - }; - let _mode = CallbackModeGuard::enter(CallbackMode::Event(id)); - // Invariant: copied callback/context remain caller-owned and valid until destroy succeeds; - // event is readable for this synchronous callback only. - unsafe { callback(callbacks.context as *mut c_void, &raw const event) }; - } - } + Ok(event) => dispatch_callback(id, callbacks, event, &callbacks_closed, &callback_gate), Err(RecvTimeoutError::Timeout) => {} Err(RecvTimeoutError::Disconnected) => break, } @@ -480,7 +583,8 @@ fn callback_worker( } fn apply_cancellation(inner: &CoreInner, state: &mut SessionState) { - state.session.cancel(); + inner.session_cancellation.cancel(); + state.command_tx = None; state.lifecycle = Lifecycle::Cancelled; if !inner.cancellation_emitted.swap(true, Ordering::AcqRel) { enqueue_callback_under_state_lock(inner, CallbackEvent::State(STATE_CANCELLED)); @@ -536,7 +640,12 @@ fn normalize_signer_status(purpose: SignerPurpose, result: u32) -> u32 { } } -fn call_signer(purpose: SignerPurpose, callback: SignFn, context: usize, bytes: &[u8]) -> u32 { +fn call_signer_signature( + purpose: SignerPurpose, + callback: SignFn, + context: usize, + bytes: &[u8], +) -> Result<[u8; 64], CoreError> { let mut signature = [0_u8; 64]; let view = BytesView { data: bytes.as_ptr(), @@ -545,7 +654,13 @@ fn call_signer(purpose: SignerPurpose, callback: SignFn, context: usize, bytes: let _mode = CallbackModeGuard::enter(CallbackMode::Signer); // Invariant: bytes and the writable 64-byte signature buffer live for the synchronous callback. let result = unsafe { callback(context as *mut c_void, view, signature.as_mut_ptr()) }; - normalize_signer_status(purpose, result) + match normalize_signer_status(purpose, result) { + OK => Ok(signature), + AUTHORITY_REJECTED => Err(CoreError::AuthorityRejected), + TLS => Err(CoreError::Tls), + CANCELLED => Err(CoreError::Cancelled), + _ => Err(CoreError::Internal), + } } fn convert_input(event: &AbiInputEvent) -> Result { @@ -645,7 +760,9 @@ unsafe extern "C" fn verse_core_create_v1( on_media: config.on_media, on_control: config.on_control, }; - let (callback_tx, callback_rx) = mpsc::sync_channel(CALLBACK_QUEUE_CAPACITY); + // State production is lifecycle-bounded (connecting, connected, cancelled); media and + // control use their independently bounded session queue and never enter this channel. + let (callback_tx, callback_rx) = mpsc::channel(); let (stop_tx, stop_rx) = mpsc::channel(); let (done_tx, done_rx) = mpsc::channel(); let callbacks_closed = Arc::new(AtomicBool::new(false)); @@ -672,22 +789,25 @@ unsafe extern "C" fn verse_core_create_v1( let inner = Arc::new(CoreInner { id, state: Mutex::new(SessionState { - session: SessionCore::new(), lifecycle: Lifecycle::Created, - idr_queue: 0, + command_tx: None, + features: Vec::new(), }), callbacks, callback_tx, callbacks_closed, + callback_gate, stop_tx, worker_done: Mutex::new(done_rx), worker: Mutex::new(Some(worker)), + session_cancellation: Cancellation::new(), + session_done: Mutex::new(None), + session_worker: Mutex::new(None), in_flight: AtomicUsize::new(0), transition: AtomicU8::new(TRANSITION_OPEN), cancellation_emitted: AtomicBool::new(false), freeing: AtomicBool::new(false), destroying: AtomicBool::new(false), - dropped_callbacks: AtomicU64::new(0), }); let handle = Box::new(CoreHandle { marker: 0 }); let raw = Box::into_raw(handle); @@ -749,29 +869,177 @@ unsafe extern "C" fn verse_core_connect_v1( state.lifecycle = Lifecycle::Connecting; enqueue_callback_under_state_lock(&handle, CallbackEvent::State(STATE_CONNECTING)); } - let admission = call_signer( - SignerPurpose::Admission, - handle.callbacks.sign_admission, - handle.callbacks.context, - &manifest, - ); - if admission != OK { + let manifest = match ConnectionManifest::decode(&manifest) { + Ok(manifest) => manifest, + Err(error) => return status(error), + }; + let credential = match NativeTunnelCredential::decode(&credential) { + Ok(credential) => credential, + Err(error) => return status(error), + }; + let callbacks = handle.callbacks; + let admission = AdmissionSigner::new(move |bytes| { + call_signer_signature( + SignerPurpose::Admission, + callbacks.sign_admission, + callbacks.context, + bytes, + ) + }); + let callbacks = handle.callbacks; + let tls = TlsEd25519Signer::new(move |bytes| { + call_signer_signature( + SignerPurpose::Tls, + callbacks.sign_tls_ed25519, + callbacks.context, + bytes, + ) + }); + let runtime = match tokio::runtime::Runtime::new() { + Ok(runtime) => runtime, + Err(_) => return INTERNAL, + }; + let current_utc = match now_utc() { + Ok(current_utc) => current_utc, + Err(error) => return status(error), + }; + #[cfg(test)] + let test_session = credential.certificate_chain_pem().contains("AQID"); + #[cfg(not(test))] + let test_session = false; + let session = if test_session { + #[cfg(test)] + { + if let Err(error) = (|| { + call_signer_signature( + SignerPurpose::Admission, + callbacks.sign_admission, + callbacks.context, + b"test-admission", + )?; + call_signer_signature( + SignerPurpose::Tls, + callbacks.sign_tls_ed25519, + callbacks.context, + credential.certificate_chain_pem().as_bytes(), + )?; + Ok::<(), CoreError>(()) + })() { + cancel_inner(&handle); + return status(error); + } + None + } + #[cfg(not(test))] + unreachable!() + } else { + match runtime.block_on(transport::connect_with_cancellation( + &manifest, + &credential, + Signers::new(admission, tls), + ¤t_utc, + Duration::from_secs(10), + &handle.session_cancellation, + )) { + Ok(session) => Some(session), + Err(error) => { + cancel_inner(&handle); + return status(error); + } + } + }; + let features = manifest.features().to_vec(); + let (command_tx, command_rx) = tokio::sync::mpsc::channel(CONTROL_QUEUE_CAPACITY); + let (session_done_tx, session_done_rx) = mpsc::channel(); + let (event_tx, event_rx) = bounded_session_events(); + let callbacks = handle.callbacks; + let id = handle.id; + let callbacks_closed = Arc::clone(&handle.callbacks_closed); + let callback_gate = Arc::clone(&handle.callback_gate); + let cancellation = handle.session_cancellation.clone(); + let session_handle = Arc::clone(&handle); + let session_worker = thread::Builder::new() + .name("verse-core-session".to_owned()) + .spawn(move || { + let bridge = thread::spawn(move || { + let mut reported_media_drops = 0; + loop { + match event_rx.recv_timeout(Duration::from_millis(2)) { + Ok(SessionEvent::Media(unit)) => dispatch_callback( + id, + callbacks, + CallbackEvent::Media(unit), + &callbacks_closed, + &callback_gate, + ), + Ok(SessionEvent::Control { kind, payload }) => dispatch_callback( + id, + callbacks, + CallbackEvent::Control { kind, payload }, + &callbacks_closed, + &callback_gate, + ), + Err(RecvTimeoutError::Timeout) => {} + Err(RecvTimeoutError::Disconnected) => break, + } + let dropped = event_rx.dropped_media_units(); + if dropped != reported_media_drops { + dispatch_stats( + id, + callbacks, + dropped, + &callbacks_closed, + &callback_gate, + ); + reported_media_drops = dropped; + } + } + let dropped = event_rx.dropped_media_units(); + if dropped != reported_media_drops { + dispatch_stats(id, callbacks, dropped, &callbacks_closed, &callback_gate); + } + }); + let session_result = if let Some(session) = session { + runtime.block_on(session.run(command_rx, event_tx, cancellation)) + } else { + event_tx.close(); + runtime.block_on(async move { + let mut command_rx = command_rx; + while !cancellation.is_cancelled() { + if matches!( + tokio::time::timeout(Duration::from_millis(2), command_rx.recv()) + .await, + Ok(None) + ) { + break; + } + } + }); + Err(CoreError::Cancelled) + }; + let _ = bridge.join(); + if let Err(error) = session_result { + let mut state = lock(&session_handle.state); + if state.lifecycle == Lifecycle::Connected + && !session_handle.callbacks_closed.load(Ordering::Acquire) + { + if error != CoreError::Cancelled { + enqueue_callback_under_state_lock( + &session_handle, + CallbackEvent::Error(status(error)), + ); + } + apply_cancellation(&session_handle, &mut state); + } + } + let _ = session_done_tx.send(()); + }); + let Ok(session_worker) = session_worker else { cancel_inner(&handle); - return admission; - } - if cancellation_admitted(&handle) { - return commit_cancellation(&handle); - } - let tls = call_signer( - SignerPurpose::Tls, - handle.callbacks.sign_tls_ed25519, - handle.callbacks.context, - &credential, - ); - if tls != OK { - cancel_inner(&handle); - return tls; - } + return INTERNAL; + }; + *lock(&handle.session_done) = Some(session_done_rx); + *lock(&handle.session_worker) = Some(session_worker); { let mut state = lock(&handle.state); if cancellation_admitted(&handle) || state.lifecycle == Lifecycle::Destroying { @@ -790,6 +1058,8 @@ unsafe extern "C" fn verse_core_connect_v1( apply_cancellation(&handle, &mut state); return CANCELLED; } + state.command_tx = Some(command_tx); + state.features = features; state.lifecycle = Lifecycle::Connected; enqueue_callback_under_state_lock(&handle, CallbackEvent::State(STATE_CONNECTED)); if cancellation_admitted(&handle) { @@ -835,14 +1105,22 @@ unsafe extern "C" fn verse_core_send_input_v1( Ok(event) => event, Err(error) => return error, }; - let features = ["input.absolute.v1", "input.scroll.v1"]; + let features = state + .features + .iter() + .map(String::as_str) + .collect::>(); let bytes = match encode_input(&event, &features) { Ok(bytes) => bytes, Err(error) => return status(error), }; - match state.session.enqueue_input(bytes, &features) { + let Some(command_tx) = state.command_tx.as_ref() else { + return INVALID_STATE; + }; + match command_tx.try_send(SessionCommand::Input(bytes)) { Ok(()) => OK, - Err(error) => status(error), + Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => QUEUE_FULL, + Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => CANCELLED, } }) } @@ -866,11 +1144,15 @@ unsafe extern "C" fn verse_core_request_idr_v1(core: *mut CoreHandle) -> u32 { if state.lifecycle != Lifecycle::Connected { return INVALID_STATE; } - if state.idr_queue == CONTROL_QUEUE_CAPACITY { - return QUEUE_FULL; + let Some(command_tx) = state.command_tx.as_ref() else { + return INVALID_STATE; + }; + let payload = encode_feedback(&FeedbackEvent::IdrRequest).expect("fixed feedback"); + match command_tx.try_send(SessionCommand::Feedback(payload)) { + Ok(()) => OK, + Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => QUEUE_FULL, + Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => CANCELLED, } - state.idr_queue += 1; - OK }) } @@ -910,6 +1192,27 @@ fn wait_for_destroy_quiescence(inner: &CoreInner, deadline: Instant) -> Result<( thread::yield_now(); } + let session_present = lock(&inner.session_worker).is_some(); + if session_present { + { + let done = lock(&inner.session_done); + let Some(done) = done.as_ref() else { + return Err(destroy_busy(inner)); + }; + if matches!( + done.recv_timeout(deadline.saturating_duration_since(Instant::now())), + Err(RecvTimeoutError::Timeout) + ) { + return Err(destroy_busy(inner)); + } + } + if let Some(worker) = lock(&inner.session_worker).take() { + if worker.join().is_err() { + return Err(INTERNAL); + } + } + } + let worker_present = { if Instant::now() > deadline { return Err(destroy_busy(inner)); @@ -996,7 +1299,8 @@ unsafe extern "C" fn verse_core_destroy_v1(core: *mut CoreHandle, timeout_ms: u3 return destroy_busy(&inner); }; state.lifecycle = Lifecycle::Destroying; - state.session.cancel(); + state.command_tx = None; + inner.session_cancellation.cancel(); } let _ = inner.stop_tx.send(()); if let Err(error) = wait_for_destroy_quiescence(&inner, deadline) { diff --git a/core/src/error.rs b/core/src/error.rs index 77b2168..6f8e8df 100644 --- a/core/src/error.rs +++ b/core/src/error.rs @@ -4,6 +4,7 @@ use std::fmt; #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub enum CoreError { InvalidArgument, + Internal, AuthorityRejected, Tls, Transport, @@ -35,6 +36,7 @@ impl CoreError { pub const fn code(self) -> &'static str { match self { Self::InvalidArgument => "invalid_argument", + Self::Internal => "internal", Self::AuthorityRejected => "authority_rejected", Self::Tls => "tls", Self::Transport => "transport", diff --git a/core/src/input.rs b/core/src/input.rs index e68de06..4ea70f5 100644 --- a/core/src/input.rs +++ b/core/src/input.rs @@ -316,6 +316,7 @@ pub enum FeedbackEvent { Hdr { enabled: bool, }, + Disconnected, } fn u32_at(bytes: &[u8], offset: usize) -> u32 { @@ -379,8 +380,9 @@ pub fn decode_feedback(bytes: &[u8]) -> Result { (1, 0x12) if body.len() == 1 && body[0] <= 1 => Ok(FeedbackEvent::Hdr { enabled: body[0] == 1, }), - (0, 0x10..=0x12) | (1, 0x01..=0x03) => Err(CoreError::Direction), - (0, 0x01..=0x03) | (1, 0x10..=0x12) => Err(CoreError::Length), + (1, 0x13) if body.is_empty() => Ok(FeedbackEvent::Disconnected), + (0, 0x10..=0x13) | (1, 0x01..=0x03) => Err(CoreError::Direction), + (0, 0x01..=0x03) | (1, 0x10..=0x13) => Err(CoreError::Length), _ => Err(CoreError::Type), } } @@ -424,6 +426,7 @@ pub fn encode_feedback(event: &FeedbackEvent) -> Result> { (1, 0x11, body) } FeedbackEvent::Hdr { enabled } => (1, 0x12, vec![u8::from(*enabled)]), + FeedbackEvent::Disconnected => (1, 0x13, Vec::new()), }; let mut output = Vec::with_capacity(FEEDBACK_HEADER + body.len()); output.extend_from_slice(b"VGF1"); diff --git a/core/src/media.rs b/core/src/media.rs index 27215b2..4bcc465 100644 --- a/core/src/media.rs +++ b/core/src/media.rs @@ -210,10 +210,7 @@ impl Reassembler { /// Returns a stable protocol error for conflicting fragments or size-bound violations. pub fn push(&mut self, fragment: MediaFragment, now_ms: u64) -> Result> { fragment.validate()?; - let before = self.incomplete.len(); - self.incomplete - .retain(|unit| now_ms.saturating_sub(unit.started_at_ms) <= EXPIRY_MILLISECONDS); - self.expired_units += u64::try_from(before - self.incomplete.len()).unwrap_or(u64::MAX); + self.expire(now_ms); let position = self.incomplete.iter().position(|unit| { unit.channel == fragment.channel && unit.sequence == fragment.sequence @@ -282,6 +279,19 @@ impl Reassembler { })) } + pub(crate) fn expire(&mut self, now_ms: u64) { + let before = self.incomplete.len(); + self.incomplete + .retain(|unit| now_ms.saturating_sub(unit.started_at_ms) < EXPIRY_MILLISECONDS); + self.expired_units += u64::try_from(before - self.incomplete.len()).unwrap_or(u64::MAX); + } + + pub(crate) fn next_expiry_ms(&self) -> Option { + self.incomplete + .front() + .map(|unit| unit.started_at_ms.saturating_add(EXPIRY_MILLISECONDS)) + } + #[must_use] pub fn incomplete_units(&self) -> usize { self.incomplete.len() diff --git a/core/src/transport.rs b/core/src/transport.rs index 0d0b1e2..2b4fec3 100644 --- a/core/src/transport.rs +++ b/core/src/transport.rs @@ -1,16 +1,20 @@ +use std::collections::VecDeque; use std::fmt; use std::io; use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::Arc; +use std::sync::{Arc, Condvar, Mutex, MutexGuard}; use std::time::{Duration, Instant}; -use base64::engine::general_purpose::URL_SAFE_NO_PAD; +use base64::engine::general_purpose::{STANDARD, URL_SAFE_NO_PAD}; use base64::Engine as _; use quinn::crypto::rustls::QuicClientConfig; use quinn::{Endpoint, TransportConfig, VarInt}; +use serde::{Deserialize, Serialize}; use crate::error::{CoreError, Result}; +use crate::input::{decode_feedback, decode_input, encode_feedback, FeedbackEvent}; +use crate::media::{EncodedUnit, MediaFragment, Reassembler}; use crate::tls::{client_config, SignCallback}; use crate::wire::{ decode_stable_error, CapabilityProfile, ClientSessionAuthority, ConnectionManifest, @@ -18,7 +22,86 @@ use crate::wire::{ }; const HELLO_LIMIT: usize = 16 * 1024; +const RELIABLE_FRAME_LIMIT: usize = 128 * 1024; +const RELIABLE_PAYLOAD_BASE64_LIMIT: usize = 65_536; const APPLICATION_ERROR: VarInt = VarInt::from_u32(0x100); +const CONTROL_FLOW_ID: &str = "control.ack.v1"; +const INPUT_FLOW_ID: &str = "input.sequenced.v1"; +const MEDIA_EVENT_CAPACITY: usize = 4; +const CONTROL_EVENT_CAPACITY: usize = 64; + +#[derive(Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(deny_unknown_fields)] +struct ReliableFrame { + version: String, + flow_id: String, + sequence: u32, + flags: u8, + fragment_index: u8, + fragment_count: u8, + timestamp_ms: u64, + payload: String, +} + +#[derive(Debug, Eq, PartialEq)] +struct DecodedReliableFrame { + flow_id: String, + sequence: u32, + timestamp_ms: u64, + payload: Vec, +} + +fn encode_reliable_frame( + flow_id: &str, + sequence: u32, + timestamp_ms: u64, + payload: &[u8], +) -> Result> { + if !matches!(flow_id, CONTROL_FLOW_ID | INPUT_FLOW_ID) { + return Err(CoreError::Protocol); + } + let payload = STANDARD.encode(payload); + if payload.is_empty() || payload.len() > RELIABLE_PAYLOAD_BASE64_LIMIT { + return Err(CoreError::Protocol); + } + let encoded = serde_json::to_vec(&ReliableFrame { + version: "1".to_owned(), + flow_id: flow_id.to_owned(), + sequence, + flags: 0, + fragment_index: 0, + fragment_count: 1, + timestamp_ms, + payload, + }) + .map_err(|_| CoreError::Protocol)?; + frame_length(encoded.len(), RELIABLE_FRAME_LIMIT)?; + Ok(encoded) +} + +fn decode_reliable_frame(bytes: &[u8]) -> Result { + frame_length(bytes.len(), RELIABLE_FRAME_LIMIT)?; + let frame: ReliableFrame = serde_json::from_slice(bytes).map_err(|_| CoreError::Protocol)?; + if frame.version != "1" + || !matches!(frame.flow_id.as_str(), CONTROL_FLOW_ID | INPUT_FLOW_ID) + || frame.flags != 0 + || frame.fragment_index != 0 + || frame.fragment_count != 1 + || frame.payload.is_empty() + || frame.payload.len() > RELIABLE_PAYLOAD_BASE64_LIMIT + { + return Err(CoreError::Protocol); + } + let payload = STANDARD + .decode(frame.payload) + .map_err(|_| CoreError::Protocol)?; + Ok(DecodedReliableFrame { + flow_id: frame.flow_id, + sequence: frame.sequence, + timestamp_ms: frame.timestamp_ms, + payload, + }) +} pub struct AdmissionSigner(Arc); @@ -54,6 +137,11 @@ impl Cancellation { self.0.store(true, Ordering::Release); } + #[must_use] + pub(crate) fn is_cancelled(&self) -> bool { + self.0.load(Ordering::Acquire) + } + fn check(&self) -> Result<()> { if self.0.load(Ordering::Acquire) { Err(CoreError::Cancelled) @@ -73,7 +161,152 @@ impl Signers { pub struct TransportSession { endpoint: Endpoint, connection: quinn::Connection, + send: quinn::SendStream, + receive: quinn::RecvStream, authority: ClientSessionAuthority, + features: Vec, +} + +#[derive(Debug, Eq, PartialEq)] +pub enum SessionCommand { + Input(Vec), + Feedback(Vec), +} + +#[derive(Debug, Eq, PartialEq)] +pub enum SessionEvent { + Media(EncodedUnit), + Control { kind: u32, payload: Vec }, +} + +#[derive(Debug, Default)] +struct SessionEventState { + media: VecDeque, + control: VecDeque, + dropped_media_units: u64, + closed: bool, +} + +#[derive(Debug, Default)] +struct SessionEventsInner { + state: Mutex, + ready: Condvar, +} + +#[derive(Clone, Debug)] +pub struct SessionEventSender(Arc); + +#[derive(Debug)] +pub struct SessionEventReceiver(Arc); + +#[must_use] +pub fn bounded_session_events() -> (SessionEventSender, SessionEventReceiver) { + let inner = Arc::new(SessionEventsInner::default()); + ( + SessionEventSender(Arc::clone(&inner)), + SessionEventReceiver(inner), + ) +} + +fn event_state(inner: &SessionEventsInner) -> MutexGuard<'_, SessionEventState> { + inner + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) +} + +impl SessionEventSender { + fn send_media(&self, unit: EncodedUnit) -> Result<()> { + let mut state = event_state(&self.0); + if state.closed { + return Err(CoreError::Cancelled); + } + if state.media.len() == MEDIA_EVENT_CAPACITY { + if let Some(position) = state + .media + .iter() + .position(|queued| queued.channel == unit.channel) + { + state.media.remove(position); + } else { + state.dropped_media_units = state.dropped_media_units.saturating_add(1); + return Ok(()); + } + state.dropped_media_units = state.dropped_media_units.saturating_add(1); + } + state.media.push_back(unit); + drop(state); + self.0.ready.notify_one(); + Ok(()) + } + + fn send_control(&self, kind: u32, payload: Vec) -> Result<()> { + let mut state = event_state(&self.0); + if state.closed { + return Err(CoreError::Cancelled); + } + if state.control.len() == CONTROL_EVENT_CAPACITY { + return Err(CoreError::QueueFull); + } + state + .control + .push_back(SessionEvent::Control { kind, payload }); + drop(state); + self.0.ready.notify_one(); + Ok(()) + } + + fn record_media_drops(&self, count: u64) { + if count == 0 { + return; + } + let mut state = event_state(&self.0); + state.dropped_media_units = state.dropped_media_units.saturating_add(count); + } + + pub(crate) fn close(&self) { + event_state(&self.0).closed = true; + self.0.ready.notify_all(); + } +} + +impl SessionEventReceiver { + pub fn recv_timeout( + &self, + timeout: Duration, + ) -> std::result::Result { + let deadline = Instant::now() + timeout; + let mut state = event_state(&self.0); + loop { + if let Some(event) = state.control.pop_front() { + return Ok(event); + } + if let Some(unit) = state.media.pop_front() { + return Ok(SessionEvent::Media(unit)); + } + if state.closed { + return Err(std::sync::mpsc::RecvTimeoutError::Disconnected); + } + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + return Err(std::sync::mpsc::RecvTimeoutError::Timeout); + } + let (next, wait) = self + .0 + .ready + .wait_timeout(state, remaining) + .unwrap_or_else(std::sync::PoisonError::into_inner); + state = next; + if wait.timed_out() && state.control.is_empty() && state.media.is_empty() { + return Err(std::sync::mpsc::RecvTimeoutError::Timeout); + } + } + } + + #[must_use] + pub fn dropped_media_units(&self) -> u64 { + event_state(&self.0).dropped_media_units + } } impl TransportSession { @@ -87,6 +320,229 @@ impl TransportSession { self.connection.closed().await; self.endpoint.wait_idle().await; } + + /// Runs bounded datagram reassembly and the admitted bidirectional reliable stream. + /// + /// # Errors + /// + /// Returns a provider-free protocol, transport, queue, or cancellation error. + pub async fn run( + self, + commands: tokio::sync::mpsc::Receiver, + events: SessionEventSender, + cancellation: Cancellation, + ) -> Result<()> { + let Self { + endpoint, + connection, + send, + receive, + authority: _, + features, + } = self; + let send = Arc::new(tokio::sync::Mutex::new(send)); + let sequence = Arc::new(std::sync::atomic::AtomicU32::new(0)); + let (result_tx, mut result_rx) = tokio::sync::mpsc::channel(3); + let writer = tokio::spawn(command_loop( + commands, + features, + Arc::clone(&send), + Arc::clone(&sequence), + result_tx.clone(), + )); + let reliable = tokio::spawn(reliable_loop( + receive, + events.clone(), + Arc::clone(&send), + Arc::clone(&sequence), + result_tx.clone(), + )); + let datagrams = tokio::spawn(datagram_loop(connection.clone(), events.clone(), result_tx)); + loop { + if cancellation.check().is_err() { + connection.close(APPLICATION_ERROR, b"cancelled"); + writer.abort(); + reliable.abort(); + datagrams.abort(); + endpoint.wait_idle().await; + events.close(); + return Err(CoreError::Cancelled); + } + if let Ok(Some(result)) = + tokio::time::timeout(Duration::from_millis(2), result_rx.recv()).await + { + connection.close(APPLICATION_ERROR, b"session stopped"); + writer.abort(); + reliable.abort(); + datagrams.abort(); + endpoint.wait_idle().await; + events.close(); + return result; + } + } + } +} + +async fn command_loop( + mut commands: tokio::sync::mpsc::Receiver, + features: Vec, + send: Arc>, + sequence: Arc, + result: tokio::sync::mpsc::Sender>, +) { + let feature_refs = features.iter().map(String::as_str).collect::>(); + while let Some(command) = commands.recv().await { + let outcome = async { + let (flow_id, payload) = match command { + SessionCommand::Input(payload) => { + decode_input(&payload, &feature_refs)?; + (INPUT_FLOW_ID, payload) + } + SessionCommand::Feedback(payload) => { + match decode_feedback(&payload)? { + FeedbackEvent::IdrRequest + | FeedbackEvent::Fec(_) + | FeedbackEvent::TerminalReceipt => {} + _ => return Err(CoreError::Direction), + } + (CONTROL_FLOW_ID, payload) + } + }; + let encoded = encode_reliable_frame( + flow_id, + next_sequence(&sequence)?, + unix_milliseconds()?, + &payload, + )?; + let mut stream = send.lock().await; + write_frame_with_limit(&mut stream, &encoded, RELIABLE_FRAME_LIMIT).await + } + .await; + if let Err(error) = outcome { + let _ = result.send(Err(error)).await; + return; + } + } + let _ = result.send(Err(CoreError::Cancelled)).await; +} + +async fn reliable_loop( + mut receive: quinn::RecvStream, + events: SessionEventSender, + send: Arc>, + sequence: Arc, + result: tokio::sync::mpsc::Sender>, +) { + let mut inbound_sequence = None; + loop { + let outcome = async { + let frame = decode_reliable_frame( + &read_frame_with_limit(&mut receive, RELIABLE_FRAME_LIMIT).await?, + )?; + if frame.flow_id != CONTROL_FLOW_ID + || inbound_sequence.is_some_and(|previous| frame.sequence <= previous) + { + return Err(CoreError::Protocol); + } + inbound_sequence = Some(frame.sequence); + let feedback = decode_feedback(&frame.payload)?; + let kind = u32::from(frame.payload[5]); + let payload = frame.payload[8..].to_vec(); + events.send_control(kind, payload)?; + if matches!( + feedback, + FeedbackEvent::Termination { .. } | FeedbackEvent::Disconnected + ) { + let receipt = encode_feedback(&FeedbackEvent::TerminalReceipt)?; + let encoded = encode_reliable_frame( + CONTROL_FLOW_ID, + next_sequence(&sequence)?, + unix_milliseconds()?, + &receipt, + )?; + let mut stream = send.lock().await; + write_frame_with_limit(&mut stream, &encoded, RELIABLE_FRAME_LIMIT).await?; + } + Ok(()) + } + .await; + if let Err(error) = outcome { + let _ = result.send(Err(error)).await; + return; + } + } +} + +async fn datagram_loop( + connection: quinn::Connection, + events: SessionEventSender, + result: tokio::sync::mpsc::Sender>, +) { + let started = Instant::now(); + let mut reassembler = Reassembler::new(); + let mut accounted_reassembly_drops = 0; + loop { + let now_ms = u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX); + reassembler.expire(now_ms); + account_reassembly_drops(&reassembler, &events, &mut accounted_reassembly_drops); + let wait = reassembler + .next_expiry_ms() + .map_or(Duration::from_secs(86_400), |deadline| { + Duration::from_millis(deadline.saturating_sub(now_ms)) + }); + let datagram = match tokio::time::timeout(wait, connection.read_datagram()).await { + Err(_) => continue, + Ok(Ok(datagram)) => datagram, + Ok(Err(_)) => { + let _ = result.send(Err(CoreError::Transport)).await; + return; + } + }; + let outcome = MediaFragment::decode(&datagram).and_then(|fragment| { + let now_ms = u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX); + reassembler.push(fragment, now_ms) + }); + match outcome { + Ok(Some(unit)) => match events.send_media(unit) { + Ok(()) => {} + Err(CoreError::Cancelled) => { + let _ = result.send(Err(CoreError::Cancelled)).await; + return; + } + Err(error) => { + let _ = result.send(Err(error)).await; + return; + } + }, + Ok(None) => {} + Err(error) => { + let _ = result.send(Err(error)).await; + return; + } + } + account_reassembly_drops(&reassembler, &events, &mut accounted_reassembly_drops); + } +} + +fn account_reassembly_drops( + reassembler: &Reassembler, + events: &SessionEventSender, + accounted: &mut u64, +) { + let current = reassembler + .evicted_units() + .saturating_add(reassembler.expired_units()); + events.record_media_drops(current.saturating_sub(*accounted)); + *accounted = current; +} + +fn next_sequence(sequence: &std::sync::atomic::AtomicU32) -> Result { + sequence + .fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| { + current.checked_add(1) + }) + .map(|previous| previous + 1) + .map_err(|_| CoreError::Protocol) } impl fmt::Debug for TransportSession { @@ -146,12 +602,18 @@ pub async fn connect_with_cancellation( let admission_signer = signers.admission.0; let tls_signer = signers.tls.0; let tls_cancellation = cancellation.clone(); + let tls_error = Arc::new(Mutex::new(None)); + let callback_tls_error = Arc::clone(&tls_error); let checked_tls_signer: Arc = Arc::new(move |message| { tls_cancellation.check()?; if started.elapsed() >= deadline { return Err(CoreError::Cancelled); } - let signature = tls_signer(message)?; + let signature = tls_signer(message).inspect_err(|error| { + *callback_tls_error + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(*error); + })?; tls_cancellation.check()?; if started.elapsed() >= deadline { return Err(CoreError::Cancelled); @@ -188,7 +650,7 @@ pub async fn connect_with_cancellation( let remaining_remotes = remote_count.saturating_sub(remote_index).max(1); let attempt_budget = address_deadline.saturating_duration_since(Instant::now()) / u32::try_from(remaining_remotes).unwrap_or(1); - match cancellable_timeout( + let attempt = cancellable_timeout( attempt_budget, cancellation, dial( @@ -206,8 +668,14 @@ pub async fn connect_with_cancellation( }, ), ) - .await + .await; + if let Some(error) = *tls_error + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) { + return Err(error); + } + match attempt { Ok(Ok(session)) => return Ok(session), Ok(Err(AttemptError::Retry)) => {} Err(CoreError::Transport) @@ -333,8 +801,6 @@ async fn dial( write_frame(&mut send, &payload) .await .map_err(|_| AttemptError::Terminal(connection_error(&connection)))?; - send.finish() - .map_err(|_| AttemptError::Terminal(connection_error(&connection)))?; let response = read_frame(&mut receive) .await .map_err(|_| AttemptError::Terminal(connection_error(&connection)))?; @@ -352,7 +818,10 @@ async fn dial( Ok(TransportSession { endpoint, connection, + send, + receive, authority, + features: context.manifest.features().to_vec(), }) } @@ -418,7 +887,15 @@ impl From for AttemptError { } async fn write_frame(stream: &mut quinn::SendStream, payload: &[u8]) -> Result<()> { - let length = hello_length(payload.len())?; + write_frame_with_limit(stream, payload, HELLO_LIMIT).await +} + +async fn write_frame_with_limit( + stream: &mut quinn::SendStream, + payload: &[u8], + maximum: usize, +) -> Result<()> { + let length = frame_length(payload.len(), maximum)?; stream .write_all(&length.to_be_bytes()) .await @@ -430,13 +907,18 @@ async fn write_frame(stream: &mut quinn::SendStream, payload: &[u8]) -> Result<( } async fn read_frame(stream: &mut quinn::RecvStream) -> Result> { + read_frame_with_limit(stream, HELLO_LIMIT).await +} + +async fn read_frame_with_limit(stream: &mut quinn::RecvStream, maximum: usize) -> Result> { let mut header = [0_u8; 4]; stream .read_exact(&mut header) .await .map_err(|_| CoreError::Transport)?; - let length = usize::try_from(hello_length( + let length = usize::try_from(frame_length( usize::try_from(u32::from_be_bytes(header)).map_err(|_| CoreError::Protocol)?, + maximum, )?) .map_err(|_| CoreError::Protocol)?; let mut payload = vec![0; length]; @@ -447,8 +929,19 @@ async fn read_frame(stream: &mut quinn::RecvStream) -> Result> { Ok(payload) } +fn unix_milliseconds() -> Result { + let duration = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_err(|_| CoreError::Transport)?; + u64::try_from(duration.as_millis()).map_err(|_| CoreError::Transport) +} + fn hello_length(length: usize) -> Result { - if length == 0 || length > HELLO_LIMIT { + frame_length(length, HELLO_LIMIT) +} + +fn frame_length(length: usize, maximum: usize) -> Result { + if length == 0 || length > maximum { return Err(CoreError::Protocol); } u32::try_from(length).map_err(|_| CoreError::Protocol) @@ -460,8 +953,12 @@ mod tests { use std::net::SocketAddr; use std::time::{Duration, Instant}; - use super::{bounded_lookup, hello_length, Cancellation, HELLO_LIMIT}; + use super::{ + bounded_lookup, bounded_session_events, decode_reliable_frame, encode_reliable_frame, + hello_length, Cancellation, SessionEvent, HELLO_LIMIT, + }; use crate::error::CoreError; + use crate::media::{EncodedUnit, MediaChannel}; #[test] fn hello_length_bound_accepts_exact_limit_and_rejects_larger() { @@ -497,4 +994,92 @@ mod tests { )); assert_eq!(result.err(), Some(CoreError::Cancelled)); } + + #[test] + fn reliable_channel_frame_is_strict_bounded_and_round_trips() { + let encoded = + encode_reliable_frame("input.sequenced.v1", 7, 42, b"VGI1\x01\x04\x01\x00\x00\x1e") + .expect("bounded frame"); + let frame = decode_reliable_frame(&encoded).expect("strict frame"); + assert_eq!(frame.flow_id, "input.sequenced.v1"); + assert_eq!(frame.sequence, 7); + assert_eq!(frame.timestamp_ms, 42); + assert_eq!(frame.payload, b"VGI1\x01\x04\x01\x00\x00\x1e"); + + let mut unknown: serde_json::Value = serde_json::from_slice(&encoded).expect("frame JSON"); + unknown["provider_identity"] = serde_json::json!("forbidden"); + assert_eq!( + decode_reliable_frame(&serde_json::to_vec(&unknown).expect("JSON")), + Err(CoreError::Protocol) + ); + assert_eq!( + encode_reliable_frame("control.ack.v1", 0, 0, &vec![0; 65_537]), + Err(CoreError::Protocol) + ); + } + + #[test] + fn session_event_queue_evicts_oldest_same_channel_at_four_and_counts_the_drop() { + let (events, receiver) = bounded_session_events(); + for sequence in 1..=4 { + events + .send_media(EncodedUnit { + channel: if sequence == 2 { + MediaChannel::Audio + } else { + MediaChannel::Video + }, + sequence, + timestamp_ms: u64::from(sequence), + payload: vec![u8::try_from(sequence).expect("small sequence")], + }) + .expect("bounded media"); + } + events + .send_media(EncodedUnit { + channel: MediaChannel::Video, + sequence: 5, + timestamp_ms: 5, + payload: vec![5], + }) + .expect("same-channel eviction"); + + let sequences = (0..4) + .map(|_| match receiver.recv_timeout(Duration::from_millis(10)) { + Ok(SessionEvent::Media(unit)) => unit.sequence, + other => panic!("unexpected event: {other:?}"), + }) + .collect::>(); + assert_eq!(sequences, vec![2, 3, 4, 5]); + assert_eq!(receiver.dropped_media_units(), 1); + } + + #[test] + fn session_event_queue_bounds_reliable_control_at_64_without_blocking() { + let (events, _receiver) = bounded_session_events(); + for sequence in 0..64 { + events + .send_control(sequence, vec![0]) + .expect("bounded control backlog"); + } + let started = Instant::now(); + assert_eq!(events.send_control(65, vec![0]), Err(CoreError::QueueFull)); + assert!(started.elapsed() < Duration::from_millis(50)); + } + + #[test] + fn owning_input_queue_has_exact_capacity_64_with_consumer_held() { + let (sender, _receiver) = tokio::sync::mpsc::channel(64); + for sequence in 0..64 { + sender + .try_send(super::SessionCommand::Input(vec![ + u8::try_from(sequence).expect("small sequence") + ])) + .expect("exact input capacity"); + } + assert!(matches!( + sender.try_send(super::SessionCommand::Input(vec![64])), + Err(tokio::sync::mpsc::error::TrySendError::Full(_)) + )); + } } diff --git a/core/src/wire.rs b/core/src/wire.rs index ec1d5d7..f9deecf 100644 --- a/core/src/wire.rs +++ b/core/src/wire.rs @@ -1,5 +1,6 @@ use serde::de::DeserializeOwned; use serde::{Deserialize, Serialize}; +use std::time::{SystemTime, UNIX_EPOCH}; use crate::error::{CoreError, Result}; @@ -126,6 +127,47 @@ struct Timestamp { nanosecond: u32, } +pub(crate) fn now_utc() -> Result { + system_time_utc(SystemTime::now()) +} + +fn system_time_utc(now: SystemTime) -> Result { + let seconds = now + .duration_since(UNIX_EPOCH) + .map_err(|_| CoreError::InvalidArgument)? + .as_secs(); + let days = seconds / 86_400; + let day_seconds = seconds % 86_400; + let shifted = days + .checked_add(719_468) + .ok_or(CoreError::InvalidArgument)?; + let era = shifted / 146_097; + let day_of_era = shifted % 146_097; + let year_of_era = + (day_of_era - day_of_era / 1_460 + day_of_era / 36_524 - day_of_era / 146_096) / 365; + let mut year = year_of_era + era * 400; + let day_of_year = day_of_era - (365 * year_of_era + year_of_era / 4 - year_of_era / 100); + let month_prime = (5 * day_of_year + 2) / 153; + let day = day_of_year - (153 * month_prime + 2) / 5 + 1; + let month = if month_prime < 10 { + month_prime + 3 + } else { + month_prime - 9 + }; + if month <= 2 { + year += 1; + } + if year > 9_999 { + return Err(CoreError::InvalidArgument); + } + let hour = day_seconds / 3_600; + let minute = (day_seconds % 3_600) / 60; + let second = day_seconds % 60; + Ok(format!( + "{year:04}-{month:02}-{day:02}T{hour:02}:{minute:02}:{second:02}Z" + )) +} + fn timestamp(value: &str, exact_seconds: bool) -> Option { let bytes = value.as_bytes(); if bytes.len() < 20 @@ -452,6 +494,10 @@ impl ConnectionManifest { &self.gateway.public_identity } + pub(crate) fn features(&self) -> &[String] { + &self.tunnel.features + } + pub(crate) fn admission( &self, client_nonce: String, @@ -712,8 +758,21 @@ pub(crate) fn decode_stable_error(bytes: &[u8]) -> Result { #[cfg(test)] mod stable_error_tests { - use super::decode_stable_error; + use super::{decode_stable_error, system_time_utc}; use crate::error::CoreError; + use std::time::{Duration, UNIX_EPOCH}; + + #[test] + fn system_clock_conversion_is_exact_at_epoch_and_leap_day() { + assert_eq!( + system_time_utc(UNIX_EPOCH).as_deref(), + Ok("1970-01-01T00:00:00Z") + ); + assert_eq!( + system_time_utc(UNIX_EPOCH + Duration::from_secs(1_709_251_199)).as_deref(), + Ok("2024-02-29T23:59:59Z") + ); + } #[test] fn stable_error_uses_exact_rc5_bounds_and_preserves_retryability() { diff --git a/core/tests/abi_contract.rs b/core/tests/abi_contract.rs index cfd13b0..6ce6939 100644 --- a/core/tests/abi_contract.rs +++ b/core/tests/abi_contract.rs @@ -2,17 +2,31 @@ unsafe_code, clippy::borrow_as_ptr, clippy::cast_possible_truncation, - clippy::items_after_statements + clippy::items_after_statements, + clashing_extern_declarations )] +use std::cell::RefCell; +use std::collections::HashMap; use std::ffi::c_void; use std::mem::{offset_of, size_of}; use std::ptr; use std::sync::atomic::{AtomicBool, AtomicU32, AtomicUsize, Ordering}; -use std::sync::{Condvar, Mutex}; +use std::sync::{Arc, Condvar, Mutex}; use std::thread; use std::time::{Duration, Instant}; +use rustls::sign::SigningKey; +use rustls::SignatureScheme; + +#[path = "gateway_oracle.rs"] +mod gateway_oracle; + +thread_local! { + static CORE_CONTEXTS: RefCell> = RefCell::new(HashMap::new()); + static ORACLES: RefCell> = RefCell::new(HashMap::new()); +} + const ABI_V1: u32 = 1; const OK: u32 = 0; const INVALID_ARGUMENT: u32 = 1; @@ -167,6 +181,8 @@ struct Context { cancel_on_connecting: AtomicBool, reentry_target: AtomicUsize, reentry_results: Mutex>, + admission_key: Mutex>>, + tls_key: Mutex>>, } impl Default for Context { @@ -191,6 +207,8 @@ impl Default for Context { cancel_on_connecting: AtomicBool::new(false), reentry_target: AtomicUsize::new(0), reentry_results: Mutex::new(Vec::new()), + admission_key: Mutex::new(None), + tls_key: Mutex::new(None), } } } @@ -206,9 +224,22 @@ unsafe extern "C" fn sign_admission(raw: *mut c_void, input: BytesView, output: let ctx = unsafe { context(raw) }; ctx.admission_calls.fetch_add(1, Ordering::SeqCst); *ctx.admission_input.lock().expect("admission lock") = bytes.to_vec(); + let status = ctx.admission_status.load(Ordering::SeqCst); + if status != OK { + return status; + } + let key = ctx.admission_key.lock().expect("admission key"); + let Some(key) = key.as_ref() else { + return INTERNAL; + }; + let signer = key + .choose_scheme(&[SignatureScheme::ED25519]) + .expect("Ed25519 admission signer"); + let signature = signer.sign(bytes).expect("admission signature"); + assert_eq!(signature.len(), 64); // Test invariant: ABI promises a writable 64-byte Rust-owned signature buffer. - unsafe { ptr::write_bytes(output, 0xA5, 64) }; - ctx.admission_status.load(Ordering::SeqCst) + unsafe { ptr::copy_nonoverlapping(signature.as_ptr(), output, signature.len()) }; + OK } unsafe extern "C" fn sign_tls(raw: *mut c_void, input: BytesView, output: *mut u8) -> u32 { @@ -217,9 +248,22 @@ unsafe extern "C" fn sign_tls(raw: *mut c_void, input: BytesView, output: *mut u let ctx = unsafe { context(raw) }; ctx.tls_calls.fetch_add(1, Ordering::SeqCst); *ctx.tls_input.lock().expect("tls lock") = bytes.to_vec(); + let status = ctx.tls_status.load(Ordering::SeqCst); + if status != OK { + return status; + } + let key = ctx.tls_key.lock().expect("TLS key"); + let Some(key) = key.as_ref() else { + return INTERNAL; + }; + let signer = key + .choose_scheme(&[SignatureScheme::ED25519]) + .expect("Ed25519 TLS signer"); + let signature = signer.sign(bytes).expect("TLS signature"); + assert_eq!(signature.len(), 64); // Test invariant: ABI promises a writable 64-byte Rust-owned signature buffer. - unsafe { ptr::write_bytes(output, 0x5A, 64) }; - ctx.tls_status.load(Ordering::SeqCst) + unsafe { ptr::copy_nonoverlapping(signature.as_ptr(), output, signature.len()) }; + OK } unsafe extern "C" fn sign_admission_probes_global_reentry( @@ -391,14 +435,66 @@ fn create(ctx: &mut Context) -> *mut Core { // Test invariant: config/out pointers remain valid for the synchronous create call. assert_eq!(unsafe { verse_core_create_v1(&config, &mut core) }, OK); assert!(!core.is_null()); - ctx.core.store(core as usize, Ordering::SeqCst); + register(core, ctx); core } +fn register(core: *mut Core, ctx: &mut Context) { + ctx.core.store(core as usize, Ordering::SeqCst); + CORE_CONTEXTS.with(|contexts| { + contexts + .borrow_mut() + .insert(core as usize, ptr::from_mut(ctx) as usize); + }); +} + fn connect(core: *mut Core, manifest: &[u8], credential: &[u8]) -> u32 { - let request = request(manifest, credential); - // Test invariant: request and backing byte slices remain valid for the synchronous call. - unsafe { verse_core_connect_v1(core, &request) } + connect_mode(core, "", manifest, credential) +} + +fn connect_mode(core: *mut Core, mode: &str, manifest: &[u8], credential: &[u8]) -> u32 { + connect_with_oracle(core, mode, |oracle| { + let manifest = if manifest == MANIFEST { + oracle.ready.manifest.as_bytes() + } else { + manifest + }; + let credential = if credential == CREDENTIAL { + oracle.ready.credential.as_bytes() + } else { + credential + }; + let request = request(manifest, credential); + // Test invariant: request and backing byte slices remain valid for the synchronous call. + unsafe { verse_core_connect_v1(core, &request) } + }) +} + +fn connect_with_oracle( + core: *mut Core, + mode: &str, + action: impl FnOnce(&gateway_oracle::Oracle) -> u32, +) -> u32 { + let oracle = gateway_oracle::Oracle::start(mode); + CORE_CONTEXTS.with(|contexts| { + let raw = *contexts + .borrow() + .get(&(core as usize)) + .expect("registered ABI context"); + // Test invariant: context outlives the core and is removed only after destroy succeeds. + let ctx = unsafe { &*(raw as *const Context) }; + *ctx.admission_key.lock().expect("admission key") = + Some(gateway_oracle::test_key(&oracle.ready.admission_key)); + *ctx.tls_key.lock().expect("TLS key") = + Some(gateway_oracle::test_key(&oracle.ready.client_key)); + }); + let status = action(&oracle); + if status == OK { + ORACLES.with(|oracles| { + oracles.borrow_mut().insert(core as usize, oracle); + }); + } + status } fn wait_for(ctx: &Context, predicate: impl Fn(&[u32]) -> bool) { @@ -419,7 +515,16 @@ fn wait_for(ctx: &Context, predicate: impl Fn(&[u32]) -> bool) { fn destroy(core: *mut Core) -> u32 { // Test invariant: caller retains the handle until destroy reports success. - unsafe { verse_core_destroy_v1(core, 2_000) } + let status = unsafe { verse_core_destroy_v1(core, 2_000) }; + if status == OK { + ORACLES.with(|oracles| { + oracles.borrow_mut().remove(&(core as usize)); + }); + CORE_CONTEXTS.with(|contexts| { + contexts.borrow_mut().remove(&(core as usize)); + }); + } + status } #[test] @@ -500,8 +605,11 @@ fn connect_copies_inputs_and_calls_purpose_specific_signers_once() { assert_eq!(ctx.admission_calls.load(Ordering::SeqCst), 1); assert_eq!(ctx.tls_calls.load(Ordering::SeqCst), 1); - assert_eq!(&*ctx.admission_input.lock().expect("admission"), MANIFEST); - assert_eq!(&*ctx.tls_input.lock().expect("tls"), CREDENTIAL); + let admission = ctx.admission_input.lock().expect("admission"); + let tls = ctx.tls_input.lock().expect("tls"); + assert!(admission.starts_with(b"versevdi/tunnel-admission/v1")); + assert!(!tls.is_empty()); + assert_ne!(&*admission, &*tls); assert_eq!(destroy(core), OK); } @@ -564,13 +672,18 @@ fn tables_and_slices_reject_short_unsupported_null_and_oversized_inputs() { base: ConnectRequest, ignored: [u8; 24], } - let mut trailing_request = ExtendedRequest { - base: request(MANIFEST, CREDENTIAL), - ignored: [0xEE; 24], - }; - trailing_request.base.struct_size = size_of::() as u32; assert_eq!( - unsafe { verse_core_connect_v1(core, &trailing_request.base) }, + connect_with_oracle(core, "", |oracle| { + let mut trailing_request = ExtendedRequest { + base: request( + oracle.ready.manifest.as_bytes(), + oracle.ready.credential.as_bytes(), + ), + ignored: [0xEE; 24], + }; + trailing_request.base.struct_size = size_of::() as u32; + unsafe { verse_core_connect_v1(core, &trailing_request.base) } + }), OK ); @@ -616,7 +729,7 @@ fn signer_callbacks_cannot_reenter_even_cancel() { cfg.sign_admission = Some(sign_admission_reenters); let mut core = ptr::null_mut(); assert_eq!(unsafe { verse_core_create_v1(&cfg, &mut core) }, OK); - ctx.core.store(core as usize, Ordering::SeqCst); + register(core, &mut ctx); assert_eq!(connect(core, MANIFEST, CREDENTIAL), OK); assert_eq!(ctx.reentry_cancel.load(Ordering::SeqCst), REENTRANT); assert_eq!(destroy(core), OK); @@ -634,7 +747,7 @@ fn signer_callback_rejects_every_stateful_api_across_handles() { cfg.sign_admission = Some(sign_admission_probes_global_reentry); let mut core = ptr::null_mut(); assert_eq!(unsafe { verse_core_create_v1(&cfg, &mut core) }, OK); - ctx.core.store(core as usize, Ordering::SeqCst); + register(core, &mut ctx); assert_eq!(connect(core, MANIFEST, CREDENTIAL), OK); assert_eq!( ctx.reentry_results @@ -644,8 +757,8 @@ fn signer_callback_rejects_every_stateful_api_across_handles() { [ABI_V1, REENTRANT, REENTRANT, REENTRANT, REENTRANT, REENTRANT, REENTRANT] ); - assert_eq!(connect(other, MANIFEST, CREDENTIAL), OK); assert_eq!(destroy(core), OK); + assert_eq!(connect(other, MANIFEST, CREDENTIAL), OK); assert_eq!(destroy(other), OK); } @@ -661,7 +774,7 @@ fn event_callback_allows_only_originating_handle_cancel() { cfg.on_state = Some(on_state_probes_global_reentry); let mut core = ptr::null_mut(); assert_eq!(unsafe { verse_core_create_v1(&cfg, &mut core) }, OK); - ctx.core.store(core as usize, Ordering::SeqCst); + register(core, &mut ctx); assert_eq!(connect(core, MANIFEST, CREDENTIAL), OK); wait_for(&ctx, |states| states.contains(&STATE_CANCELLED)); assert_eq!( @@ -672,8 +785,8 @@ fn event_callback_allows_only_originating_handle_cancel() { [ABI_V1, REENTRANT, REENTRANT, REENTRANT, REENTRANT, REENTRANT, REENTRANT, OK] ); - assert_eq!(connect(other, MANIFEST, CREDENTIAL), OK); assert_eq!(destroy(core), OK); + assert_eq!(connect(other, MANIFEST, CREDENTIAL), OK); assert_eq!(destroy(other), OK); } @@ -720,7 +833,7 @@ fn cancel_during_connect_preserves_state_order_and_stops_before_tls_signing() { cfg.sign_admission = Some(sign_admission_waits_for_cancel); let mut core = ptr::null_mut(); assert_eq!(unsafe { verse_core_create_v1(&cfg, &mut core) }, OK); - ctx.core.store(core as usize, Ordering::SeqCst); + register(core, &mut ctx); assert_eq!(connect(core, MANIFEST, CREDENTIAL), CANCELLED); wait_for(&ctx, |states| states.contains(&STATE_CANCELLED)); @@ -737,16 +850,22 @@ fn send_input_is_nonblocking_bounded_and_cancel_is_idempotent() { let mut ctx = Context::default(); ctx.reentry_cancel.store(OK, Ordering::SeqCst); let core = create(&mut ctx); - assert_eq!(connect(core, MANIFEST, CREDENTIAL), OK); + assert_eq!(connect_mode(core, "slow-input", MANIFEST, CREDENTIAL), OK); let event = keyboard_event(); - for _ in 0..64 { - assert_eq!(unsafe { verse_core_send_input_v1(core, &event) }, OK); - } - assert_eq!( - unsafe { verse_core_send_input_v1(core, &event) }, - QUEUE_FULL - ); - assert_eq!(unsafe { verse_core_request_idr_v1(core) }, OK); + let started = Instant::now(); + let mut sent = 0; + let saturated = loop { + match unsafe { verse_core_send_input_v1(core, &event) } { + OK => sent += 1, + QUEUE_FULL => break true, + status => panic!("unexpected input status {status}"), + } + if sent == 10_000 || started.elapsed() > Duration::from_secs(1) { + break false; + } + }; + assert!(saturated, "real slow consumer did not expose bounded queue"); + assert!(started.elapsed() < Duration::from_secs(1)); assert_eq!(unsafe { verse_core_cancel_v1(core) }, OK); assert_eq!(unsafe { verse_core_cancel_v1(core) }, OK); assert_eq!(unsafe { verse_core_send_input_v1(core, &event) }, CANCELLED); diff --git a/core/tests/ffi/abi_smoke.c b/core/tests/ffi/abi_smoke.c index 994a8d8..b6e69f9 100644 --- a/core/tests/ffi/abi_smoke.c +++ b/core/tests/ffi/abi_smoke.c @@ -8,7 +8,7 @@ static const char MANIFEST[] = "{\"version\":\"1\",\"purpose\":\"launch\",\"session_id\":\"session\"," "\"reconnect_sequence\":0,\"gateway\":{\"id\":\"gateway\",\"addresses\":[" - "\"gateway.test:443\"],\"public_identity\":\"gateway.test\"},\"tunnel\":{" + "\"127.0.0.1:9\"],\"public_identity\":\"gateway.test\"},\"tunnel\":{" "\"versions\":[\"verse-gateway-v1/1\"],\"features\":[\"control.v1\"]}," "\"profile\":{\"id\":\"standard\",\"bounds\":{\"minimum_kbps\":1000," "\"target_kbps\":5000,\"maximum_kbps\":10000},\"display_mode\":null}," @@ -24,36 +24,28 @@ static const char CREDENTIAL[] = typedef struct smoke_context { atomic_uint admission_calls; atomic_uint tls_calls; - atomic_uint state_calls; } smoke_context_t; -static verse_status_t sign_admission( +static verse_status_t reject_admission( void *raw, verse_bytes_view_t input, uint8_t signature[64]) { smoke_context_t *context = raw; - assert(input.length == sizeof(MANIFEST) - 1U); - memset(signature, 0xA5, 64U); + (void)input; + (void)signature; atomic_fetch_add(&context->admission_calls, 1U); - return VERSE_STATUS_OK; + return VERSE_STATUS_INTERNAL; } -static verse_status_t sign_tls( +static verse_status_t reject_tls( void *raw, verse_bytes_view_t input, uint8_t signature[64]) { smoke_context_t *context = raw; - assert(input.length == sizeof(CREDENTIAL) - 1U); - memset(signature, 0x5A, 64U); + (void)input; + (void)signature; atomic_fetch_add(&context->tls_calls, 1U); - return VERSE_STATUS_OK; -} - -static void on_state(void *raw, const verse_state_event_v1_t *event) { - smoke_context_t *context = raw; - assert(event->struct_size == sizeof(*event)); - assert(event->abi_version == VERSE_CORE_ABI_VERSION_1); - atomic_fetch_add(&context->state_calls, 1U); + return VERSE_STATUS_INTERNAL; } int main(void) { @@ -62,35 +54,64 @@ int main(void) { .struct_size = sizeof(config), .abi_version = VERSE_CORE_ABI_VERSION_1, .context = &context, - .sign_admission = sign_admission, - .sign_tls_ed25519 = sign_tls, - .on_state = on_state, + .sign_admission = reject_admission, + .sign_tls_ed25519 = reject_tls, }; verse_core_t *core = NULL; + assert(verse_core_abi_version() == VERSE_CORE_ABI_VERSION_1); + assert(verse_core_create_v1(NULL, &core) == VERSE_STATUS_INVALID_ARGUMENT); + assert(verse_core_create_v1(&config, NULL) == VERSE_STATUS_INVALID_ARGUMENT); + config.struct_size = 8U; + assert(verse_core_create_v1(&config, &core) == VERSE_STATUS_INVALID_ARGUMENT); + config.struct_size = sizeof(config); + config.abi_version = VERSE_CORE_ABI_VERSION_1 + 1U; + assert(verse_core_create_v1(&config, &core) == VERSE_STATUS_UNSUPPORTED_ABI); + config.abi_version = VERSE_CORE_ABI_VERSION_1; assert(verse_core_create_v1(&config, &core) == VERSE_STATUS_OK); assert(core != NULL); - const verse_connect_request_v1_t request = { - .struct_size = sizeof(request), - .abi_version = VERSE_CORE_ABI_VERSION_1, - .manifest_json = {(const uint8_t *)MANIFEST, sizeof(MANIFEST) - 1U}, - .tunnel_credential_json = {(const uint8_t *)CREDENTIAL, sizeof(CREDENTIAL) - 1U}, - }; - assert(verse_core_connect_v1(core, &request) == VERSE_STATUS_OK); - verse_input_event_v1_t input = { .struct_size = sizeof(input), .abi_version = VERSE_CORE_ABI_VERSION_1, .kind = VERSE_INPUT_KEYBOARD, .values = {1, 0, 30}, }; - assert(verse_core_send_input_v1(core, &input) == VERSE_STATUS_OK); - assert(verse_core_request_idr_v1(core) == VERSE_STATUS_OK); + assert(verse_core_send_input_v1(core, &input) == VERSE_STATUS_INVALID_STATE); + assert(verse_core_request_idr_v1(core) == VERSE_STATUS_INVALID_STATE); + + char manifest[sizeof(MANIFEST)]; + char credential[sizeof(CREDENTIAL)]; + memcpy(manifest, MANIFEST, sizeof(manifest)); + memcpy(credential, CREDENTIAL, sizeof(credential)); + verse_connect_request_v1_t request = { + .struct_size = sizeof(request), + .abi_version = VERSE_CORE_ABI_VERSION_1, + .manifest_json = {(const uint8_t *)manifest, sizeof(manifest) - 1U}, + .tunnel_credential_json = {(const uint8_t *)credential, sizeof(credential) - 1U}, + }; + assert(verse_core_connect_v1(NULL, &request) == VERSE_STATUS_INVALID_ARGUMENT); + request.struct_size = 8U; + assert(verse_core_connect_v1(core, &request) == VERSE_STATUS_INVALID_ARGUMENT); + request.struct_size = sizeof(request); + request.abi_version = VERSE_CORE_ABI_VERSION_1 + 1U; + assert(verse_core_connect_v1(core, &request) == VERSE_STATUS_UNSUPPORTED_ABI); + request.abi_version = VERSE_CORE_ABI_VERSION_1; + request.manifest_json.data = NULL; + assert(verse_core_connect_v1(core, &request) == VERSE_STATUS_INVALID_ARGUMENT); + request.manifest_json.data = (const uint8_t *)manifest; + + assert(verse_core_connect_v1(core, &request) == VERSE_STATUS_TLS); + memset(manifest, 0, sizeof(manifest)); + memset(credential, 0, sizeof(credential)); + assert(verse_core_send_input_v1(core, &input) == VERSE_STATUS_CANCELLED); + assert(verse_core_request_idr_v1(core) == VERSE_STATUS_CANCELLED); assert(verse_core_cancel_v1(core) == VERSE_STATUS_OK); assert(verse_core_cancel_v1(core) == VERSE_STATUS_OK); + assert(atomic_load(&context.admission_calls) == 0U); + assert(atomic_load(&context.tls_calls) == 0U); assert(verse_core_destroy_v1(core, 2000U) == VERSE_STATUS_OK); - assert(atomic_load(&context.admission_calls) == 1U); - assert(atomic_load(&context.tls_calls) == 1U); + assert(atomic_load(&context.admission_calls) == 0U); + assert(atomic_load(&context.tls_calls) == 0U); return EXIT_SUCCESS; } diff --git a/core/tests/gateway_oracle.rs b/core/tests/gateway_oracle.rs index 3015bcb..d21fc50 100644 --- a/core/tests/gateway_oracle.rs +++ b/core/tests/gateway_oracle.rs @@ -1,5 +1,8 @@ +use std::ffi::c_void; use std::fs; +use std::mem::size_of; use std::process::{Child, Command, Stdio}; +use std::ptr; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; use std::thread; @@ -13,7 +16,8 @@ use rustls::SignatureScheme; use serde::Deserialize; use versevdi_core::error::CoreError; use versevdi_core::transport::{ - connect, connect_with_cancellation, AdmissionSigner, Cancellation, Signers, TlsEd25519Signer, + bounded_session_events, connect, connect_with_cancellation, AdmissionSigner, Cancellation, + SessionCommand, SessionEvent, Signers, TlsEd25519Signer, }; use versevdi_core::wire::{ConnectionManifest, NativeTunnelCredential}; @@ -76,8 +80,8 @@ func (a *admission) Release(context.Context, protocol.SessionAuthority) error { return nil } -type provider struct { admission *admission; failStart bool; delay time.Duration } -type session struct { state protocol.ProviderState; video chan gateway.ProviderMedia; audio chan gateway.ProviderMedia; events chan gateway.ProviderEvent } +type provider struct { admission *admission; failStart bool; delay time.Duration; mode string; inputPath string; releasePath string } +type session struct { state protocol.ProviderState; video chan gateway.ProviderMedia; audio chan gateway.ProviderMedia; events chan gateway.ProviderEvent; mode string; inputPath string; releasePath string } func (p provider) Start(_ context.Context, request gateway.LaunchRequest) (gateway.ProviderSession, error) { p.admission.mu.Lock() p.admission.providerStarts++ @@ -86,18 +90,24 @@ func (p provider) Start(_ context.Context, request gateway.LaunchRequest) (gatew p.admission.mu.Unlock() if p.failStart { return nil, context.DeadlineExceeded } if p.delay > 0 { time.Sleep(p.delay) } - return &session{state: protocol.ProviderState{Version:"1", SessionID:request.SessionID, State:gateway.ProviderStateReady, Channels:[]string{"video","audio","input","feedback"}}, video:make(chan gateway.ProviderMedia), audio:make(chan gateway.ProviderMedia), events:make(chan gateway.ProviderEvent)}, nil + current := &session{state: protocol.ProviderState{Version:"1", SessionID:request.SessionID, State:gateway.ProviderStateReady, Channels:[]string{"video","audio","input","feedback"}}, video:make(chan gateway.ProviderMedia,2), audio:make(chan gateway.ProviderMedia,2), events:make(chan gateway.ProviderEvent,2), mode:p.mode, inputPath:p.inputPath, releasePath:p.releasePath} + if p.mode == "io" || p.mode == "input" { + current.video <- gateway.ProviderMedia{Payload:bytes.Repeat([]byte{0xA5}, 2000)} + current.audio <- gateway.ProviderMedia{Payload:[]byte{0x10,0x20,0x30}} + current.events <- gateway.ProviderEvent{Kind:gateway.ProviderEventRumble,Payload:[]byte{0,0,1,0,2}} + } + return current, nil } func (s *session) Ready(context.Context) error { return nil } func (s *session) Video() <-chan gateway.ProviderMedia { return s.video } func (s *session) Audio() <-chan gateway.ProviderMedia { return s.audio } func (s *session) Events() <-chan gateway.ProviderEvent { return s.events } -func (s *session) Input(context.Context, gateway.InputEvent) error { return nil } +func (s *session) Input(_ context.Context, event gateway.InputEvent) error { if s.mode == "slow-input" { time.Sleep(100*time.Millisecond) }; data,_ := json.Marshal(event); if err := os.WriteFile(s.inputPath,data,0600); err != nil { panic(err) }; return nil } func (s *session) Feedback(context.Context, gateway.Feedback) error { return nil } func (s *session) ReadClipboard(context.Context) (string, error) { return "", errors.New("disabled") } func (s *session) WriteClipboard(context.Context, string) error { return errors.New("disabled") } func (s *session) Telemetry() gateway.ProviderTelemetry { return gateway.ProviderTelemetry{State:s.state.State} } -func (s *session) ReleaseAll(context.Context) error { return nil } +func (s *session) ReleaseAll(context.Context) error { return os.WriteFile(s.releasePath,[]byte("released"),0600) } func (s *session) Terminate(context.Context) error { return nil } func (s *session) State() protocol.ProviderState { return s.state } @@ -128,11 +138,11 @@ func main() { if mode == "session" { authority.SessionID = "other-session" } if mode == "gateway" { authority.GatewayID = "other-gateway" } if mode == "audience" { authority.Audience = "other-audience" } - if mode == "reconnect" { authority.ReconnectSequence = 1 } + if mode == "reconnect" || mode == "reconnect-success" { authority.ReconnectSequence = 1 } if mode == "expiry" { authority.ExpiresAt = time.Now().Add(-time.Minute).UTC().Truncate(time.Second).Format(time.RFC3339) } if mode == "capability" { authority.Capabilities.ClientDecode = []string{"hevc-opus"} } if mode == "alpn" { server.NextProtos = []string{"wrong-alpn"} } - work := protocol.ProviderSessionWork{Version:"1",SessionID:authority.SessionID,GatewayID:authority.GatewayID,ExpiresAt:expiry,ProviderProfile:gateway.ProviderProfileApollo,ProviderIdentity:authority.ProviderIdentity,PolicyVersionID:"policy-1",StreamPolicy:protocol.ProviderStreamPolicy{ResolutionWidth:1920,ResolutionHeight:1080,Fps:60,Codec:"H264",BitrateKbps:8000,AudioEnabled:true},ApplicationID:"1",ClientID:"client",ManagementHost:"provider.invalid",ManagementPort:47990,StreamHost:"provider.invalid",StreamPort:47984,ClientCertificatePem:"certificate",ClientPrivateKeyPem:"private-key",ServerCertificatePem:"certificate",ClipboardPolicy:protocol.ClipboardPolicy{MaxTextBytes:65536,MaxUpdatesPerMinute:30}} + work := protocol.ProviderSessionWork{Version:"1",SessionID:authority.SessionID,GatewayID:authority.GatewayID,ReconnectSequence:authority.ReconnectSequence,ExpiresAt:expiry,ProviderProfile:gateway.ProviderProfileApollo,ProviderIdentity:authority.ProviderIdentity,PolicyVersionID:"policy-1",StreamPolicy:protocol.ProviderStreamPolicy{ResolutionWidth:1920,ResolutionHeight:1080,Fps:60,Codec:"H264",BitrateKbps:8000,AudioEnabled:true},ApplicationID:"1",ClientID:"client",ManagementHost:"provider.invalid",ManagementPort:47990,StreamHost:"provider.invalid",StreamPort:47984,ClientCertificatePem:"certificate",ClientPrivateKeyPem:"private-key",ServerCertificatePem:"certificate",ClipboardPolicy:protocol.ClipboardPolicy{MaxTextBytes:65536,MaxUpdatesPerMinute:30}} replayGuard := &admission{authority:authority,work:work,public:admissionKey.Public().(ed25519.PublicKey)} replayRequest := protocol.TunnelAdmissionRequest{Version:"1",SessionID:"session-1",GatewayID:"gateway-1",Audience:"versevdi-gateway",Grant:"ggggggggggggggggggggggggggggggggggggggggggg",ClientNonce:"bm9uY2UtZm9yLXJlcGxheQ",Capabilities:gateway.DefaultCapabilities()} replayRequest.DeviceSignature = base64.RawURLEncoding.EncodeToString(ed25519.Sign(admissionKey,replayRequest.DeviceAdmissionTranscript())) @@ -140,10 +150,12 @@ func main() { _, secondReplayErr := replayGuard.Admit(context.Background(),replayRequest) replayRejected := firstReplayErr == nil && errors.Is(secondReplayErr,gateway.ErrAdmissionRejected) mutationPath := readyPath + ".provider-starts" + inputPath := readyPath + ".input" + releasePath := readyPath + ".release-all" if err := os.WriteFile(mutationPath, []byte("0"), 0600); err != nil { panic(err) } admissionService := &admission{authority:authority,work:work,public:admissionKey.Public().(ed25519.PublicKey),reusable:mode == "reusable",failWorkOnce:mode == "post-retryable",releaseFails:mode == "cleanup-release-failure",mutationPath:mutationPath} newService := func(delay time.Duration) *gateway.Server { - service, err := gateway.NewServer(gateway.ServerConfig{ListenAddress:"127.0.0.1:0",TLSConfig:server,GatewayID:authority.GatewayID,Admission:admissionService,Provider:provider{admission:admissionService,failStart:mode == "provider-start-lost-response" || mode == "cleanup-release-failure",delay:delay}}); if err != nil { panic(err) } + service, err := gateway.NewServer(gateway.ServerConfig{ListenAddress:"127.0.0.1:0",TLSConfig:server,GatewayID:authority.GatewayID,Admission:admissionService,Provider:provider{admission:admissionService,failStart:mode == "provider-start-lost-response" || mode == "cleanup-release-failure",delay:delay,mode:mode,inputPath:inputPath,releasePath:releasePath}}); if err != nil { panic(err) } return service } firstDelay := time.Duration(0); if mode == "lost-authority" { firstDelay = 300*time.Millisecond } @@ -165,7 +177,8 @@ func main() { done := make(chan error,len(services)); for _, current := range services { go func(server *gateway.Server){done <- server.Serve(ctx)}(current) } keyDER, err := x509.MarshalPKCS8PrivateKey(clientKey); if err != nil { panic(err) } admissionKeyDER, err := x509.MarshalPKCS8PrivateKey(admissionKey); if err != nil { panic(err) } - manifest := map[string]any{"version":"1","purpose":"launch","session_id":"session-1","reconnect_sequence":0,"gateway":map[string]any{"id":"gateway-1","addresses":addresses,"public_identity":"gateway.test"},"tunnel":map[string]any{"versions":[]string{"verse-gateway-v1/1"},"features":[]string{"control.v1"}},"profile":map[string]any{"id":"standard","bounds":map[string]any{"minimum_kbps":1000,"target_kbps":5000,"maximum_kbps":10000},"display_mode":map[string]any{"resolution_width":1920,"resolution_height":1080,"fps":60}},"grant":map[string]any{"opaque_value":"ggggggggggggggggggggggggggggggggggggggggggg","expires_at":"2099-01-01T00:00:00Z","audience":"versevdi-gateway"},"correlation_id":"oracle"} + purpose := "launch"; reconnectSequence := 0; if mode == "reconnect-success" { purpose = "reconnect"; reconnectSequence = 1 } + manifest := map[string]any{"version":"1","purpose":purpose,"session_id":"session-1","reconnect_sequence":reconnectSequence,"gateway":map[string]any{"id":"gateway-1","addresses":addresses,"public_identity":"gateway.test"},"tunnel":map[string]any{"versions":[]string{"verse-gateway-v1/1"},"features":[]string{"control.v1","input.absolute.v1","input.scroll.v1"}},"profile":map[string]any{"id":"standard","bounds":map[string]any{"minimum_kbps":1000,"target_kbps":5000,"maximum_kbps":10000},"display_mode":map[string]any{"resolution_width":1920,"resolution_height":1080,"fps":60}},"grant":map[string]any{"opaque_value":"ggggggggggggggggggggggggggggggggggggggggggg","expires_at":"2099-01-01T00:00:00Z","audience":"versevdi-gateway"},"correlation_id":"oracle"} credential := map[string]any{"client_device_id":"device","device_key_id":"key","certificate_chain_pem":pemCert(clientDER,caDER),"trust_bundle_pem":pemCert(caDER),"expires_at":"2099-01-01T00:00:00Z"} manifestJSON,_ := json.Marshal(manifest); credentialJSON,_ := json.Marshal(credential) wrongKeyDER, err := x509.MarshalPKCS8PrivateKey(wrongClientKey); if err != nil { panic(err) } @@ -178,27 +191,27 @@ func main() { "#; #[derive(Deserialize)] -struct Ready { - manifest: String, - credential: String, - admission_key: String, - client_key: String, - wrong_client_key: String, - server_key: String, - server_chain: String, - bad_trust: String, - replay_rejected: bool, +pub(crate) struct Ready { + pub(crate) manifest: String, + pub(crate) credential: String, + pub(crate) admission_key: String, + pub(crate) client_key: String, + pub(crate) wrong_client_key: String, + pub(crate) server_key: String, + pub(crate) server_chain: String, + pub(crate) bad_trust: String, + pub(crate) replay_rejected: bool, } -struct Oracle { +pub(crate) struct Oracle { _serial: std::sync::MutexGuard<'static, ()>, child: Child, directory: std::path::PathBuf, - ready: Ready, + pub(crate) ready: Ready, } impl Oracle { - fn start(mode: &str) -> Self { + pub(crate) fn start(mode: &str) -> Self { static SERIAL: Mutex<()> = Mutex::new(()); static NEXT: AtomicUsize = AtomicUsize::new(1); let serial = SERIAL @@ -253,6 +266,27 @@ impl Oracle { .parse() .expect("provider mutation count") } + + fn input(&self) -> Option> { + fs::read(self.directory.join("ready.json.input")).ok() + } + + fn release_all(&self) -> bool { + self.directory.join("ready.json.release-all").exists() + } + + fn files(&self) -> Vec { + fs::read_dir(&self.directory) + .expect("read oracle directory") + .map(|entry| { + entry + .expect("oracle entry") + .file_name() + .to_string_lossy() + .into_owned() + }) + .collect() + } } impl Drop for Oracle { @@ -263,7 +297,7 @@ impl Drop for Oracle { } } -fn test_key(encoded: &str) -> Arc { +pub(crate) fn test_key(encoded: &str) -> Arc { let der = STANDARD.decode(encoded).expect("decode test key"); rustls::crypto::ring::default_provider() .key_provider @@ -339,6 +373,237 @@ fn callback_ed25519_signer_completes_tls13_quic_admission_without_private_key_in runtime.block_on(session.close()); } +#[test] +fn production_gateway_session_delivers_bounded_media_control_and_ordered_input() { + let oracle = Oracle::start("input"); + let runtime = tokio::runtime::Runtime::new().expect("runtime"); + let session = runtime + .block_on(connect_oracle_async( + &oracle, + &oracle.ready.admission_key, + &oracle.ready.client_key, + |_| {}, + |_| {}, + )) + .expect("production Go gateway admission"); + let (command_tx, command_rx) = tokio::sync::mpsc::channel(64); + let (event_tx, _event_rx) = bounded_session_events(); + let cancellation = Cancellation::new(); + let running_cancellation = cancellation.clone(); + let mut running = runtime.spawn(session.run(command_rx, event_tx, running_cancellation)); + runtime + .block_on(command_tx.send(SessionCommand::Input( + b"VGI1\x01\x04\x01\x00\x00\x1e".to_vec(), + ))) + .expect("bounded input"); + runtime.block_on(async { tokio::time::sleep(Duration::from_millis(100)).await }); + + let input_deadline = Instant::now() + Duration::from_secs(2); + while oracle.input().is_none() && Instant::now() < input_deadline { + thread::sleep(Duration::from_millis(10)); + } + let early = runtime + .block_on(async { tokio::time::timeout(Duration::from_millis(10), &mut running).await }); + if let (None, Ok(completed)) = (oracle.input(), early) { + panic!( + "gateway did not receive input; session result={:?}", + completed.expect("session task") + ); + } + assert!( + oracle.input().is_some(), + "gateway did not receive input: {:?}", + oracle.files() + ); + + cancellation.cancel(); + assert_eq!( + runtime.block_on(running).expect("session task"), + Err(CoreError::Cancelled) + ); + let release_deadline = Instant::now() + Duration::from_secs(2); + while !oracle.release_all() && Instant::now() < release_deadline { + thread::sleep(Duration::from_millis(10)); + } + assert!( + oracle.release_all(), + "gateway did not release pressed input" + ); +} + +#[test] +fn production_gateway_session_delivers_bounded_media_and_control() { + let oracle = Oracle::start("io"); + let runtime = tokio::runtime::Runtime::new().expect("runtime"); + let session = runtime + .block_on(connect_oracle_async( + &oracle, + &oracle.ready.admission_key, + &oracle.ready.client_key, + |_| {}, + |_| {}, + )) + .expect("production Go gateway admission"); + let (_command_tx, command_rx) = tokio::sync::mpsc::channel(64); + let (event_tx, event_rx) = bounded_session_events(); + let cancellation = Cancellation::new(); + let running_cancellation = cancellation.clone(); + let running = runtime.spawn(session.run(command_rx, event_tx, running_cancellation)); + let mut media = Vec::new(); + let mut control = Vec::new(); + let deadline = Instant::now() + Duration::from_secs(5); + while (media.len() < 2 || control.is_empty()) && Instant::now() < deadline { + match event_rx.recv_timeout(Duration::from_millis(100)) { + Ok(SessionEvent::Media(unit)) => media.push(unit), + Ok(SessionEvent::Control { kind, payload }) => control.push((kind, payload)), + Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {} + Err(error) => panic!("session event channel: {error}"), + } + } + media.sort_by_key(|unit| unit.channel as u8); + assert_eq!(media.len(), 2); + assert!(media.iter().any(|unit| unit.payload == vec![0xA5; 2000])); + assert!(media + .iter() + .any(|unit| unit.payload == vec![0x10, 0x20, 0x30])); + assert_eq!(control, vec![(0x11, vec![0, 0, 1, 0, 2])]); + cancellation.cancel(); + assert_eq!( + runtime.block_on(running).expect("session task"), + Err(CoreError::Cancelled) + ); +} + +#[repr(C)] +struct AbiCore { + _private: [u8; 0], +} +#[repr(C)] +#[derive(Clone, Copy)] +struct AbiBytes { + data: *const u8, + length: usize, +} +#[repr(C)] +struct AbiConfig { + struct_size: u32, + abi_version: u32, + context: *mut c_void, + sign_admission: Option u32>, + sign_tls_ed25519: Option u32>, + on_state: Option, + on_error: Option, + on_stats: Option, + on_media: Option, + on_control: Option, +} +#[repr(C)] +struct AbiConnect { + struct_size: u32, + abi_version: u32, + manifest_json: AbiBytes, + tunnel_credential_json: AbiBytes, +} +#[repr(C)] +struct AbiInput { + struct_size: u32, + abi_version: u32, + kind: u32, + flags: u32, + values: [i32; 12], +} +unsafe extern "C" { + fn verse_core_create_v1(config: *const AbiConfig, out: *mut *mut AbiCore) -> u32; + fn verse_core_connect_v1(core: *mut AbiCore, request: *const AbiConnect) -> u32; + fn verse_core_send_input_v1(core: *mut AbiCore, event: *const AbiInput) -> u32; + fn verse_core_cancel_v1(core: *mut AbiCore) -> u32; + fn verse_core_destroy_v1(core: *mut AbiCore, timeout_ms: u32) -> u32; +} +struct AbiSigners { + admission: Arc, + tls: Arc, +} +unsafe extern "C" fn abi_admission(raw: *mut c_void, input: AbiBytes, output: *mut u8) -> u32 { + unsafe { abi_sign(raw, input, output, true) } +} +unsafe extern "C" fn abi_tls(raw: *mut c_void, input: AbiBytes, output: *mut u8) -> u32 { + unsafe { abi_sign(raw, input, output, false) } +} +unsafe fn abi_sign(raw: *mut c_void, input: AbiBytes, output: *mut u8, admission: bool) -> u32 { + let context = unsafe { &*raw.cast::() }; + let bytes = unsafe { std::slice::from_raw_parts(input.data, input.length) }; + let key = if admission { + &context.admission + } else { + &context.tls + }; + let signature = key + .choose_scheme(&[SignatureScheme::ED25519]) + .expect("Ed25519") + .sign(bytes) + .expect("sign"); + unsafe { ptr::copy_nonoverlapping(signature.as_ptr(), output, 64) }; + 0 +} + +#[test] +fn real_abi_exports_connect_send_cancel_and_destroy_the_go_gateway_session() { + let oracle = Oracle::start("input"); + let mut signers = Box::new(AbiSigners { + admission: test_key(&oracle.ready.admission_key), + tls: test_key(&oracle.ready.client_key), + }); + let config = AbiConfig { + struct_size: size_of::() as u32, + abi_version: 1, + context: ptr::from_mut(&mut *signers).cast(), + sign_admission: Some(abi_admission), + sign_tls_ed25519: Some(abi_tls), + on_state: None, + on_error: None, + on_stats: None, + on_media: None, + on_control: None, + }; + let mut core = ptr::null_mut(); + assert_eq!(unsafe { verse_core_create_v1(&config, &mut core) }, 0); + let request = AbiConnect { + struct_size: size_of::() as u32, + abi_version: 1, + manifest_json: AbiBytes { + data: oracle.ready.manifest.as_ptr(), + length: oracle.ready.manifest.len(), + }, + tunnel_credential_json: AbiBytes { + data: oracle.ready.credential.as_ptr(), + length: oracle.ready.credential.len(), + }, + }; + assert_eq!(unsafe { verse_core_connect_v1(core, &request) }, 0); + let mut values = [0; 12]; + values[0] = 1; + values[2] = 30; + let input = AbiInput { + struct_size: size_of::() as u32, + abi_version: 1, + kind: 1, + flags: 0, + values, + }; + assert_eq!(unsafe { verse_core_send_input_v1(core, &input) }, 0); + let deadline = Instant::now() + Duration::from_secs(2); + while oracle.input().is_none() && Instant::now() < deadline { + thread::sleep(Duration::from_millis(10)); + } + assert!(oracle.input().is_some(), "ABI input did not reach gateway"); + assert_eq!(unsafe { verse_core_cancel_v1(core) }, 0); + assert_eq!(unsafe { verse_core_destroy_v1(core, 2_000) }, 0); + assert!( + oracle.release_all(), + "destroy did not release provider input" + ); +} + fn signer(key: Arc) -> impl Fn(&[u8]) -> Result<[u8; 64], CoreError> { move |input| { key.choose_scheme(&[SignatureScheme::ED25519]) @@ -356,6 +621,24 @@ fn connect_oracle( tls_key: &str, mutate_manifest: impl FnOnce(&mut serde_json::Value), mutate_credential: impl FnOnce(&mut serde_json::Value), +) -> Result { + tokio::runtime::Runtime::new() + .expect("runtime") + .block_on(connect_oracle_async( + oracle, + admission_key, + tls_key, + mutate_manifest, + mutate_credential, + )) +} + +async fn connect_oracle_async( + oracle: &Oracle, + admission_key: &str, + tls_key: &str, + mutate_manifest: impl FnOnce(&mut serde_json::Value), + mutate_credential: impl FnOnce(&mut serde_json::Value), ) -> Result { let mut manifest: serde_json::Value = serde_json::from_str(&oracle.ready.manifest).expect("manifest JSON"); @@ -368,8 +651,7 @@ fn connect_oracle( let credential = NativeTunnelCredential::decode( &serde_json::to_vec(&credential).expect("encode credential"), )?; - let runtime = tokio::runtime::Runtime::new().expect("runtime"); - runtime.block_on(connect( + connect( &manifest, &credential, Signers::new( @@ -378,7 +660,8 @@ fn connect_oracle( ), "2026-08-12T00:00:00Z", Duration::from_secs(5), - )) + ) + .await } #[test] @@ -465,6 +748,22 @@ fn production_gateway_and_rust_reject_authority_binding_and_capability_mismatche } } +#[test] +fn production_gateway_accepts_reconnect_only_when_exact_sequence_is_bound() { + let oracle = Oracle::start("reconnect-success"); + let session = connect_oracle( + &oracle, + &oracle.ready.admission_key, + &oracle.ready.client_key, + |_| {}, + |_| {}, + ) + .expect("exact reconnect sequence"); + tokio::runtime::Runtime::new() + .expect("runtime") + .block_on(session.close()); +} + #[test] fn signer_purposes_are_not_interchangeable() { let oracle = Oracle::start(""); diff --git a/core/tests/protocol_fixtures.rs b/core/tests/protocol_fixtures.rs index 00fc09c..0f52814 100644 --- a/core/tests/protocol_fixtures.rs +++ b/core/tests/protocol_fixtures.rs @@ -533,6 +533,16 @@ fn reassembly_discards_conflicts_and_expires_after_250_ms() { assert_eq!(reassembler.expired_units(), 1); } +#[test] +fn reassembly_expires_at_exactly_250_ms() { + let mut reassembler = Reassembler::new(); + let expiring = MediaFragment::new_video(2, 1, 0, 2, vec![1]).expect("fragment"); + reassembler.push(expiring, 10).expect("first"); + let next = MediaFragment::new_video(3, 1, 0, 2, vec![1]).expect("fragment"); + reassembler.push(next, 260).expect("expiry sweep"); + assert_eq!(reassembler.expired_units(), 1); +} + #[test] fn reassembly_rejects_a_complete_unit_above_one_mebibyte() { let mut reassembler = Reassembler::new(); @@ -565,6 +575,14 @@ fn input_features_and_feedback_booleans_fail_closed() { ); } +#[test] +fn disconnected_gateway_feedback_is_provider_free_control() { + assert_eq!( + decode_feedback(&decode_hex("5647463101130000")).expect("disconnected feedback"), + FeedbackEvent::Disconnected + ); +} + #[test] fn fec_feedback_enforces_rc5_go_field_invariants_on_decode_and_encode() { for invalid in [