diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e6ba914..df41685 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -59,7 +59,7 @@ jobs: - uses: Swatinem/rust-cache@v2 - uses: taiki-e/install-action@cargo-llvm-cov - name: Build loader fixtures - run: cargo build --locked --example module_clock --example module_clock_two --example module_wrong_target --no-default-features --features modules,macros + run: cargo build --locked --example module_clock --example module_clock_two --example module_wrong_target --example module_context --example module_context_eager --no-default-features --features modules,macros - name: Enforce 90% line coverage per source file shell: bash run: | @@ -67,7 +67,9 @@ jobs: module=$(find target/debug/examples -maxdepth 1 -type f -name 'libmodule_clock.so' -print -quit) module_two=$(find target/debug/examples -maxdepth 1 -type f -name 'libmodule_clock_two.so' -print -quit) wrong_target=$(find target/debug/examples -maxdepth 1 -type f -name 'libmodule_wrong_target.so' -print -quit) - for artifact in "$module" "$module_two" "$wrong_target"; do + context_module=$(find target/debug/examples -maxdepth 1 -type f -name 'libmodule_context.so' -print -quit) + context_eager_module=$(find target/debug/examples -maxdepth 1 -type f -name 'libmodule_context_eager.so' -print -quit) + for artifact in "$module" "$module_two" "$wrong_target" "$context_module" "$context_eager_module"; do if [ -z "$artifact" ] || [ ! -f "$artifact" ]; then echo "a module loader fixture is missing" >&2 exit 1 @@ -91,7 +93,7 @@ jobs: # process/network branches are not meaningful per-file coverage. The # test-support module is integration-test glue whose platform-specific # dynamic-loader and timeout paths run in the dedicated OS jobs instead. - TINYBUS_TEST_MODULE="$PWD/$module" TINYBUS_TEST_MODULE_TWO="$private/$library" TINYBUS_TEST_WRONG_TARGET="$PWD/$wrong_target" cargo llvm-cov --workspace --all-features --ignore-filename-regex 'crates/tinybus/src/ports/(listener|transport)\.rs$|crates/tinybus/src/module/github\.rs$|crates/tinybus/src/module/(host|mod)\.rs$|crates/tinybus/src/bin/tinybus\.rs$|crates/tinybus/src/connection\.rs$|crates/tinybus/src/test_support\.rs$' --fail-under-file-lines 90 -- --include-ignored + TINYBUS_TEST_MODULE="$PWD/$module" TINYBUS_TEST_MODULE_TWO="$private/$library" TINYBUS_TEST_WRONG_TARGET="$PWD/$wrong_target" TINYBUS_TEST_CONTEXT_MODULE="$PWD/$context_module" TINYBUS_TEST_CONTEXT_EAGER_MODULE="$PWD/$context_eager_module" cargo llvm-cov --workspace --all-features --ignore-filename-regex 'crates/tinybus/src/ports/(listener|transport)\.rs$|crates/tinybus/src/module/github\.rs$|crates/tinybus/src/module/(host|mod)\.rs$|crates/tinybus/src/bin/tinybus\.rs$|crates/tinybus/src/connection\.rs$|crates/tinybus/src/test_support\.rs$' --fail-under-file-lines 90 -- --include-ignored # The bus is meant to be embeddable on every platform OpenHuman ships on. The # socket adapter remains Unix-only, while the in-process module transport @@ -128,7 +130,7 @@ jobs: - uses: dtolnay/rust-toolchain@stable - uses: Swatinem/rust-cache@v2 - name: Build example module - run: cargo build --locked --example module_clock --example module_clock_two --example module_wrong_target --no-default-features --features modules,macros + run: cargo build --locked --example module_clock --example module_clock_two --example module_wrong_target --example module_context --example module_context_eager --no-default-features --features modules,macros - name: Test shared module test support run: cargo test --locked --no-default-features --features modules,macros,test-support test_support::tests - name: Exercise the real loader (Unix) @@ -139,7 +141,9 @@ jobs: module=$(find target/debug/examples -maxdepth 1 -type f \( -name 'libmodule_clock.so' -o -name 'libmodule_clock.dylib' \) -print -quit) module_two=$(find target/debug/examples -maxdepth 1 -type f \( -name 'libmodule_clock_two.so' -o -name 'libmodule_clock_two.dylib' \) -print -quit) wrong_target=$(find target/debug/examples -maxdepth 1 -type f \( -name 'libmodule_wrong_target.so' -o -name 'libmodule_wrong_target.dylib' \) -print -quit) - for artifact in "$module" "$module_two" "$wrong_target"; do + context_module=$(find target/debug/examples -maxdepth 1 -type f \( -name 'libmodule_context.so' -o -name 'libmodule_context.dylib' \) -print -quit) + context_eager_module=$(find target/debug/examples -maxdepth 1 -type f \( -name 'libmodule_context_eager.so' -o -name 'libmodule_context_eager.dylib' \) -print -quit) + for artifact in "$module" "$module_two" "$wrong_target" "$context_module" "$context_eager_module"; do if [ -z "$artifact" ] || [ ! -f "$artifact" ]; then echo "a module example artifact is missing; the loader tests would not exercise it" >&2 exit 1 @@ -156,7 +160,7 @@ jobs: digest="$(shasum -a 256 "$private/$library" | awk '{print $1}')" fi printf '"%s" = "%s"\n' "$library" "$digest" > "$private/modules.toml" - TINYBUS_TEST_MODULE="$PWD/$module" TINYBUS_TEST_MODULE_TWO="$private/$library" TINYBUS_TEST_WRONG_TARGET="$PWD/$wrong_target" cargo test --locked --no-default-features --features modules,macros,test-support -- --ignored + TINYBUS_TEST_MODULE="$PWD/$module" TINYBUS_TEST_MODULE_TWO="$private/$library" TINYBUS_TEST_WRONG_TARGET="$PWD/$wrong_target" TINYBUS_TEST_CONTEXT_MODULE="$PWD/$context_module" TINYBUS_TEST_CONTEXT_EAGER_MODULE="$PWD/$context_eager_module" cargo test --locked --no-default-features --features modules,macros,test-support -- --ignored # Directory admission against this runner's real filesystem: a bundle # in a directory other accounts can write falls back to the verified # release cache, and on macOS the real /Applications is admitted. @@ -190,13 +194,22 @@ jobs: $acl.AddAccessRule($rule) Set-Acl -LiteralPath $private -AclObject $acl Copy-Item target/debug/examples/module_clock.dll $private + Copy-Item target/debug/examples/module_context.dll $private + Copy-Item target/debug/examples/module_context_eager.dll $private + Copy-Item target/debug/examples/module_wrong_target.dll $private + # Admit every intentional fixture by its actual built bytes so later + # checks, including the wrong-target gate, can exercise their contract. + $fixtureAllowlist = @('module_clock.dll', 'module_context.dll', 'module_context_eager.dll', 'module_wrong_target.dll') | ForEach-Object { + $digest = (Get-FileHash -LiteralPath (Join-Path $private $_) -Algorithm SHA256).Hash.ToLowerInvariant() + '"' + $_ + '" = "' + $digest + '"' + } + $fixtureAllowlist | Set-Content -LiteralPath (Join-Path $private modules.toml) $clockTwoDirectory = Join-Path $private module-clock-two New-Item -ItemType Directory -Force $clockTwoDirectory | Out-Null Set-Acl -LiteralPath $clockTwoDirectory -AclObject $acl Copy-Item target/debug/examples/module_clock_two.dll $clockTwoDirectory $clockTwoHash = (Get-FileHash -LiteralPath (Join-Path $clockTwoDirectory module_clock_two.dll) -Algorithm SHA256).Hash.ToLowerInvariant() '"module_clock_two.dll" = "' + $clockTwoHash + '"' | Set-Content -LiteralPath (Join-Path $clockTwoDirectory modules.toml) -NoNewline - Copy-Item target/debug/examples/module_wrong_target.dll $private $mixed = Join-Path $private mixed New-Item -ItemType Directory -Force $mixed | Out-Null Set-Acl -LiteralPath $mixed -AclObject $acl @@ -207,6 +220,8 @@ jobs: $wrongTarget = (Get-ChildItem "$private/module_wrong_target.dll" | Select-Object -First 1).FullName $env:TEMP = $private $env:TMP = $private + $env:TINYBUS_TEST_CONTEXT_MODULE = (Get-Item "$private/module_context.dll").FullName + $env:TINYBUS_TEST_CONTEXT_EAGER_MODULE = (Get-Item "$private/module_context_eager.dll").FullName $env:TINYBUS_TEST_MODULE = $module $env:TINYBUS_TEST_MODULE_TWO = $moduleTwo $env:TINYBUS_TEST_WRONG_TARGET = $wrongTarget diff --git a/Cargo.toml b/Cargo.toml index b56957a..addae4a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -13,6 +13,7 @@ resolver = "3" [workspace.package] version = "0.1.3" edition = "2024" +rust-version = "1.85" license = "GPL-3.0-only" repository = "https://github.com/tinyhumansai/tinybus" diff --git a/crates/tinybus-macros/Cargo.toml b/crates/tinybus-macros/Cargo.toml index 9efec82..ccb7426 100644 --- a/crates/tinybus-macros/Cargo.toml +++ b/crates/tinybus-macros/Cargo.toml @@ -2,6 +2,7 @@ name = "tinybus-macros" version.workspace = true edition.workspace = true +rust-version.workspace = true license.workspace = true repository.workspace = true description = "The #[interface] attribute for tinybus. Not useful on its own; use the `tinybus` crate." diff --git a/crates/tinybus-macros/src/lib.rs b/crates/tinybus-macros/src/lib.rs index 14212f0..2793746 100644 --- a/crates/tinybus-macros/src/lib.rs +++ b/crates/tinybus-macros/src/lib.rs @@ -210,13 +210,15 @@ fn expand( /// Read `name = "…"` off the attribute. fn interface_name(args: &Punctuated) -> syn::Result { for arg in args { - if let Meta::NameValue(nv) = arg - && nv.path.is_ident("name") - && let syn::Expr::Lit(lit) = &nv.value - && let syn::Lit::Str(s) = &lit.lit - { - return Ok(s.value()); + let Meta::NameValue(nv) = arg else { continue }; + if !nv.path.is_ident("name") { + continue; } + let syn::Expr::Lit(lit) = &nv.value else { + continue; + }; + let syn::Lit::Str(s) = &lit.lit else { continue }; + return Ok(s.value()); } Err(Error::new( proc_macro2::Span::call_site(), diff --git a/crates/tinybus-module/Cargo.toml b/crates/tinybus-module/Cargo.toml index 48f4907..6cf7994 100644 --- a/crates/tinybus-module/Cargo.toml +++ b/crates/tinybus-module/Cargo.toml @@ -2,6 +2,7 @@ name = "tinybus-module" version.workspace = true edition.workspace = true +rust-version.workspace = true license.workspace = true repository.workspace = true description = "Module-side runtime and export macro for trusted tinybus cdylibs." diff --git a/crates/tinybus-module/src/lib.rs b/crates/tinybus-module/src/lib.rs index c6cc333..468175b 100644 --- a/crates/tinybus-module/src/lib.rs +++ b/crates/tinybus-module/src/lib.rs @@ -3,11 +3,16 @@ //! Each module owns a Tokio runtime. A statically linked `cdylib` has its own //! Tokio thread-locals, so attempting to borrow the host runtime is both //! incorrect and capable of silently blocking a host worker thread. +//! +//! Caller context depends on the trusted host's optional ABI-v1 routing callback. +//! Its assertion is captured per frame at enqueue time; older host prefixes +//! remain compatible but supply unverified context. Sender headers remain intact +//! for protocol routing and streams even when they carry no caller authority. use std::ffi::c_void; use std::future::Future; use std::panic::{AssertUnwindSafe, catch_unwind}; -use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::{Arc, Mutex as StdMutex, OnceLock}; use std::time::Duration; @@ -164,6 +169,12 @@ unsafe impl Send for HostCalls {} unsafe impl Sync for HostCalls {} impl HostCalls { + fn is_broker_routed(&self) -> bool { + self.0 + .broker_routing + .is_some_and(|routing| unsafe { routing(self.0.host_ctx) } == 1) + } + fn send(&self, bytes: &[u8]) -> i32 { unsafe { (self.0.send)(self.0.host_ctx, bytes.as_ptr(), bytes.len()) } } @@ -244,9 +255,13 @@ impl tracing::field::Visit for LogVisitor { } } +// Private envelope captures trusted-host routing before the frame is queued. +type IncomingFrame = (Vec, bool); + struct ModuleTransport { host: HostCalls, - inbound: Mutex>>, + inbound: Mutex>, + provenance: Arc, detach_on_panic: bool, } @@ -273,11 +288,16 @@ impl Transport for ModuleTransport { async fn recv(&self) -> Result> { let bytes = self.inbound.lock().await.recv().await; - let Some(bytes) = bytes else { + let Some((bytes, brokered)) = bytes else { return Ok(None); }; self.host.wake(); - Ok(Some(serde_json::from_slice(&bytes)?)) + let message = serde_json::from_slice(&bytes)?; + // The connection is the only reader and samples this slot immediately + // after recv. Keep header identity intact for stream ownership/routing; + // only contextual dispatch consumes the captured delivery authority. + self.provenance.store(brokered, Ordering::Release); + Ok(Some(message)) } async fn close(&self) -> Result<()> { @@ -290,8 +310,21 @@ impl Transport for ModuleTransport { } } +fn module_connection(transport: Arc) -> Connection { + if transport.host.0.broker_routing.is_some() { + let provenance = transport.provenance.clone(); + // SAFETY: this private transport has one reader, publishes the queue's + // per-frame proof before returning, and no other writer touches its + // slot. The trusted host asserts routing at deliver, not at receive. + unsafe { Connection::__attach_module_with_provenance(transport, provenance) } + } else { + Connection::attach(transport) + } +} + struct RuntimeState { - inbound: StdMutex>>>, + inbound: StdMutex>>, + host: HostCalls, runtime: StdMutex>, reinitialize: Option, } @@ -307,10 +340,11 @@ unsafe extern "C" fn deliver(ctx: *mut c_void, ptr: *const u8, len: usize) -> i3 return TB_BAD_ARGUMENT; } let state = unsafe { &*(ctx.cast::()) }; + let brokered = state.host.is_broker_routed(); let bytes = unsafe { std::slice::from_raw_parts(ptr, len) }.to_vec(); let sender = state.inbound.lock().expect("module inbound lock").clone(); match sender { - Some(sender) => match sender.try_send(bytes) { + Some(sender) => match sender.try_send((bytes, brokered)) { Ok(()) => TB_OK, Err(mpsc::error::TrySendError::Full(_)) => TB_BACKPRESSURE, Err(mpsc::error::TrySendError::Closed(_)) => TB_CLOSED, @@ -398,6 +432,15 @@ unsafe fn write_module_vtable( /// /// This is public only for [`module_export!`] expansions. Module authors call /// the macro, not this function directly. +/// +/// # Safety +/// The host vtable must meet [`TbHostVtable::read_compatible`]'s requirements. +/// A non-null `out` must be aligned and writable for its declared capacity, +/// with a readable size field; writes are bounded to that capacity. Borrowed +/// configuration bytes must remain readable during initialization. +/// In particular an affirmative routing callback is a trusted host assertion +/// that only real broker deliveries enter `deliver`, not an arbitrary frame +/// injection capability. Artifact attestation does not establish this fact. #[doc(hidden)] pub unsafe fn start_module( host: *const TbHostVtable, @@ -424,6 +467,10 @@ where } /// Start a linked module without replacing the host's process-global hooks. +/// +/// # Safety +/// The caller must uphold the host, output, configuration, and routing contracts +/// of [`start_module`]. #[doc(hidden)] pub unsafe fn start_linked_module( host: *const TbHostVtable, @@ -475,10 +522,10 @@ where } // `size` is the frozen prefix field; do not copy the full vtable until // the host has proved that all v1 fields are present. - if unsafe { host.cast::().read() } < size_of::() as u32 { + let Some(host) = (unsafe { TbHostVtable::read_compatible(host) }) else { return TB_BAD_ARGUMENT; - } - let host = HostCalls(unsafe { *host }); + }; + let host = HostCalls(host); // A cdylib carries its own statically linked `tracing` and `std` state; // this registration is global to the module's copy, not the embedding @@ -531,16 +578,18 @@ where Err(_) => return TB_CLOSED, }; let (inbound_tx, inbound_rx) = mpsc::channel(MODULE_QUEUE_CAPACITY); - let transport = Box::new(ModuleTransport { + let transport = Arc::new(ModuleTransport { host, inbound: Mutex::new(inbound_rx), + provenance: Arc::new(AtomicBool::new(false)), detach_on_panic, }); let connection_slot = Arc::new(StdMutex::new(None)); let task_connection_slot = connection_slot.clone(); runtime.spawn(async move { - let outcome = match Connection::connect(transport).await { + let connection = module_connection(transport); + let outcome = match connection.handshake().await.map(|()| connection) { Ok(connection) => { connection.__set_panic_handler(std::sync::Arc::new(move || { let location = panic_location @@ -581,6 +630,7 @@ where let supports_reinitialize = reinitializer.is_some(); let state = Box::new(RuntimeState { + host, inbound: StdMutex::new(Some(inbound_tx)), runtime: StdMutex::new(Some(runtime)), reinitialize: reinitializer, @@ -600,10 +650,7 @@ unsafe fn parse_config( if host.is_null() { return Err(TB_BAD_ARGUMENT); } - if unsafe { host.cast::().read() } < size_of::() as u32 { - return Err(TB_BAD_ARGUMENT); - } - let host_ref = unsafe { &*host }; + let host_ref = unsafe { TbHostVtable::read_compatible(host) }.ok_or(TB_BAD_ARGUMENT)?; if host_ref.config.len > 1024 * 1024 || (host_ref.config.ptr.is_null() && host_ref.config.len != 0) { @@ -622,6 +669,10 @@ unsafe fn parse_config( /// This retains the original `FnOnce` contract for direct SDK callers. The /// export macro uses [`start_reconfigurable_module`] because a named setup /// function can safely be called again. +/// +/// # Safety +/// The caller must uphold the host, output, configuration, and routing contracts +/// of [`start_module`]. #[doc(hidden)] pub unsafe fn start_module_with_config( host: *const TbHostVtable, @@ -653,6 +704,10 @@ where } /// Initialize a configured module whose setup function may be called again. +/// +/// # Safety +/// The caller must uphold the host, output, configuration, and routing contracts +/// of [`start_module`]. #[doc(hidden)] pub unsafe fn start_reconfigurable_module( host: *const TbHostVtable, @@ -672,6 +727,10 @@ where } /// Start a configurable linked module without process-global hooks. +/// +/// # Safety +/// The caller must uphold the host, output, configuration, and routing contracts +/// of [`start_module`]. #[doc(hidden)] pub unsafe fn start_linked_reconfigurable_module( host: *const TbHostVtable, diff --git a/crates/tinybus-module/src/lib_tests.rs b/crates/tinybus-module/src/lib_tests.rs index 53ce560..663af56 100644 --- a/crates/tinybus-module/src/lib_tests.rs +++ b/crates/tinybus-module/src/lib_tests.rs @@ -73,6 +73,7 @@ fn host(config: &[u8]) -> TbHostVtable { len: config.len(), }, ready: host_ready, + broker_routing: None, } } @@ -88,8 +89,9 @@ fn message() -> Message { #[test] fn a_module_whose_queue_is_full_reports_backpressure_rather_than_blocking_the_broker() { let (sender, _receiver) = mpsc::channel(1); - sender.try_send(vec![1]).unwrap(); + sender.try_send((vec![1], false)).unwrap(); let state = RuntimeState { + host: HostCalls(host(b"{}")), inbound: StdMutex::new(Some(sender)), runtime: StdMutex::new(None), reinitialize: None, @@ -108,6 +110,7 @@ fn a_module_whose_queue_is_full_reports_backpressure_rather_than_blocking_the_br #[test] fn a_frame_over_the_size_cap_is_rejected_rather_than_truncated() { let state = RuntimeState { + host: HostCalls(host(b"{}")), inbound: StdMutex::new(None), runtime: StdMutex::new(None), reinitialize: None, @@ -199,6 +202,7 @@ fn a_manifest_declaration_exports_the_declared_surface_and_dependencies() { #[test] fn a_panicking_shutdown_callback_reports_panicked_not_timed_out() { let state = RuntimeState { + host: HostCalls(host(b"{}")), inbound: StdMutex::new(None), runtime: StdMutex::new(None), reinitialize: None, @@ -222,6 +226,7 @@ fn deliver_and_shutdown_validate_their_arguments_and_closed_state() { TB_BAD_ARGUMENT ); let state = RuntimeState { + host: HostCalls(host(b"{}")), inbound: StdMutex::new(None), runtime: StdMutex::new(None), reinitialize: None, @@ -250,6 +255,7 @@ async fn module_transport_maps_host_results_and_wakes_after_receiving() { let transport = ModuleTransport { host: HostCalls(host(&[])), inbound: Mutex::new(receiver), + provenance: Arc::new(AtomicBool::new(false)), detach_on_panic: true, }; HOST_SEND_CODE.store(TB_BACKPRESSURE, Ordering::Release); @@ -275,7 +281,7 @@ async fn module_transport_maps_host_results_and_wakes_after_receiving() { assert!(HOST_FAULTED.load(Ordering::Acquire)); HOST_WAKES.store(0, Ordering::Release); sender - .send(serde_json::to_vec(&message()).unwrap()) + .send((serde_json::to_vec(&message()).unwrap(), false)) .await .unwrap(); assert_eq!(transport.recv().await.unwrap().unwrap(), message()); @@ -534,7 +540,7 @@ fn configured_startup_refuses_a_host_vtable_it_cannot_trust() { // A host built against an older, smaller descriptor. Reading our // fields out of it would run off the end of what it allocated. let mut truncated = host(b"{}"); - truncated.size = (size_of::() - 1) as u32; + truncated.size = tinybus::module::abi::TB_HOST_VTABLE_BASE_SIZE - 1; assert_eq!( attempt(&truncated), TB_BAD_ARGUMENT, @@ -591,9 +597,10 @@ fn configured_startup_refuses_a_host_vtable_it_cannot_trust() { #[test] fn deliver_reports_closed_when_the_modules_queue_receiver_is_gone() { - let (tx, rx) = mpsc::channel::>(1); + let (tx, rx) = mpsc::channel::<(Vec, bool)>(1); drop(rx); let state = RuntimeState { + host: HostCalls(host(b"{}")), inbound: StdMutex::new(Some(tx)), runtime: StdMutex::new(None), reinitialize: None, @@ -619,6 +626,7 @@ fn reinitialize_validates_its_arguments_and_reports_closed_without_a_handler() { TB_BAD_ARGUMENT ); let state = RuntimeState { + host: HostCalls(host(b"{}")), inbound: StdMutex::new(None), runtime: StdMutex::new(None), reinitialize: None, @@ -713,3 +721,255 @@ fn a_linked_reconfigurable_module_refuses_bad_hosts_and_reports_closed_before_it ); assert_eq!(unsafe { (out.shutdown)(out.module_ctx, 10) }, TB_OK); } + +#[test] +fn an_older_host_prefix_is_copied_without_reading_the_additive_tail() { + let current = host(b"{}"); + let prefix = tinybus::module::abi::TbHostVtableV1Prefix { + size: tinybus::module::abi::TB_HOST_VTABLE_BASE_SIZE, + _reserved: current._reserved, + host_ctx: current.host_ctx, + send: current.send, + wake: current.wake, + log: current.log, + fault: current.fault, + config: current.config, + ready: current.ready, + }; + let ptr = std::ptr::from_ref(&prefix).cast::(); + let normalized = unsafe { TbHostVtable::read_compatible(ptr) }.unwrap(); + assert!(normalized.broker_routing.is_none()); + assert_eq!( + unsafe { parse_config::(ptr) }.unwrap(), + serde_json::json!({}) + ); + assert_eq!( + tinybus::module::abi::TB_HOST_VTABLE_BASE_SIZE as usize, + std::mem::offset_of!(TbHostVtable, broker_routing) + ); +} + +#[test] +fn an_incomplete_host_tail_never_reads_the_callback() { + let mut current = host(b"{}"); + current.size = size_of::() as u32 - 1; + let normalized = unsafe { TbHostVtable::read_compatible(¤t) }.unwrap(); + assert!(normalized.broker_routing.is_none()); + assert!(unsafe { TbHostVtable::read_compatible(std::ptr::null()) }.is_none()); + current.size = tinybus::module::abi::TB_HOST_VTABLE_BASE_SIZE - 1; + assert!(unsafe { TbHostVtable::read_compatible(¤t) }.is_none()); +} + +unsafe extern "C" fn routing_assertion(ctx: *mut c_void) -> u32 { + unsafe { ctx.cast::().read() } +} + +#[test] +fn native_routing_requires_an_explicit_affirmative_host_assertion() { + let mut table = host(b"{}"); + assert!(!HostCalls(table).is_broker_routed()); + table.broker_routing = Some(routing_assertion); + for value in [0u32, 2, u32::MAX] { + table.host_ctx = std::ptr::from_ref(&value).cast_mut().cast(); + assert!(!HostCalls(table).is_broker_routed()); + } + // Counterfeit host callbacks can assert 1: that is why start_module and + // __attach_brokered_module are unsafe trusted-host contracts, never safe + // generic transport authentication APIs. + let affirmative = 1u32; + table.host_ctx = std::ptr::from_ref(&affirmative).cast_mut().cast(); + assert!(HostCalls(table).is_broker_routed()); +} + +unsafe extern "C" fn atomic_routing_assertion(ctx: *mut c_void) -> u32 { + unsafe { &*ctx.cast::() }.load(Ordering::Acquire) as u32 +} + +#[tokio::test] +async fn admission_never_promotes_a_frame_already_queued_without_provenance() { + let _guard = host_state_guard().await; + let admitted = AtomicBool::new(false); + let mut table = host(b"{}"); + table.host_ctx = std::ptr::from_ref(&admitted).cast_mut().cast(); + table.broker_routing = Some(atomic_routing_assertion); + let (sender, receiver) = mpsc::channel(2); + let state = RuntimeState { + host: HostCalls(table), + inbound: StdMutex::new(Some(sender)), + runtime: StdMutex::new(None), + reinitialize: None, + }; + let transport = ModuleTransport { + host: HostCalls(table), + inbound: Mutex::new(receiver), + provenance: Arc::new(AtomicBool::new(false)), + detach_on_panic: false, + }; + let mut frame = message(); + frame.header.sender = Some(":1.99".parse().unwrap()); + let bytes = serde_json::to_vec(&frame).unwrap(); + let enqueue = || unsafe { + deliver( + std::ptr::from_ref(&state).cast_mut().cast(), + bytes.as_ptr(), + bytes.len(), + ) + }; + assert_eq!(enqueue(), TB_OK); + admitted.store(true, Ordering::Release); + assert_eq!(enqueue(), TB_OK); + assert_eq!(transport.recv().await.unwrap().unwrap(), frame); + assert!(!transport.provenance.load(Ordering::Acquire)); + assert_eq!(transport.recv().await.unwrap().unwrap(), frame); + assert!(transport.provenance.load(Ordering::Acquire)); +} + +#[tokio::test] +async fn old_hosts_retain_sender_headers_without_authenticating_them() { + let _guard = host_state_guard().await; + let (sender, receiver) = mpsc::channel(1); + let transport = ModuleTransport { + host: HostCalls(host(b"{}")), + inbound: Mutex::new(receiver), + provenance: Arc::new(AtomicBool::new(false)), + detach_on_panic: false, + }; + let mut frame = message(); + frame.header.sender = Some(":1.99".parse().unwrap()); + sender + .send((serde_json::to_vec(&frame).unwrap(), false)) + .await + .unwrap(); + assert_eq!(transport.recv().await.unwrap().unwrap(), frame); + // No routing callback means startup selects ordinary Connection::attach; + // sender headers remain protocol data, not incoming-call authority. + assert!(transport.host.0.broker_routing.is_none()); +} + +struct UnverifiedContext; +#[async_trait] +impl tinybus::Interface for UnverifiedContext { + fn name(&self) -> tinybus::InterfaceName { + "org.example.Context".parse().unwrap() + } + fn members(&self) -> Vec { + vec!["Inspect".parse().unwrap()] + } + async fn call( + &self, + _: &tinybus::MemberName, + _: serde_json::Value, + ) -> Result { + Err(Error::failed("context required")) + } + async fn call_with_context( + &self, + _: &tinybus::MemberName, + _: serde_json::Value, + context: &tinybus::CallContext, + ) -> Result { + Ok(serde_json::json!(context.authenticated_sender().is_some())) + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn unverified_native_frames_preserve_distinct_stream_owners_and_unverified_context() { + let _guard = host_state_guard().await; + for compatible_callback in [false, true] { + let admitted = AtomicBool::new(false); + let mut table = host(b"{}"); + table.send = capture_host_send; + if compatible_callback { + table.host_ctx = std::ptr::from_ref(&admitted).cast_mut().cast(); + table.broker_routing = Some(atomic_routing_assertion); + } + let (outgoing_tx, mut outgoing_rx) = std::sync::mpsc::sync_channel(2); + *START_OUTGOING + .get_or_init(|| StdMutex::new(None)) + .lock() + .unwrap() = Some(outgoing_tx); + let (sender, receiver) = mpsc::channel(2); + let state = RuntimeState { + host: HostCalls(table), + inbound: StdMutex::new(Some(sender)), + runtime: StdMutex::new(None), + reinitialize: None, + }; + let transport = Arc::new(ModuleTransport { + host: HostCalls(table), + inbound: Mutex::new(receiver), + provenance: Arc::new(AtomicBool::new(false)), + detach_on_panic: false, + }); + let connection = module_connection(transport); + connection + .serve_at("/context".parse().unwrap(), UnverifiedContext) + .await + .unwrap(); + let mut serial = 0; + for (member, args, peer) in [ + ("Open", serde_json::json!([{}]), ":1.1"), + ("Write", serde_json::json!(["s1", 0, "eA=="]), ":1.2"), + ("Write", serde_json::json!(["s1", 0, "eA=="]), ":1.1"), + ("Inspect", serde_json::json!([]), ":1.1"), + ("Inspect", serde_json::json!([]), ":1.2"), + ] { + serial += 1; + let inspect = member == "Inspect"; + let mut frame = Message::method_call( + ":1.99".parse().unwrap(), + if inspect { + "/context" + } else { + tinybus::stream::STREAM_PATH + } + .parse() + .unwrap(), + if inspect { + "org.example.Context" + } else { + tinybus::stream::STREAM_INTERFACE + } + .parse() + .unwrap(), + member.parse().unwrap(), + args, + ); + frame.header.serial = serial; + frame.header.sender = Some(peer.parse().unwrap()); + let bytes = serde_json::to_vec(&frame).unwrap(); + assert_eq!( + unsafe { + deliver( + std::ptr::from_ref(&state).cast_mut().cast(), + bytes.as_ptr(), + bytes.len(), + ) + }, + TB_OK + ); + let (bytes, receiver) = tokio::task::spawn_blocking(move || { + let bytes = outgoing_rx.recv_timeout(Duration::from_secs(5)).unwrap(); + (bytes, outgoing_rx) + }) + .await + .unwrap(); + outgoing_rx = receiver; + let reply: Message = serde_json::from_slice(&bytes).unwrap(); + assert_eq!(reply.header.reply_serial, Some(serial)); + if peer == ":1.2" && member == "Write" { + assert_eq!( + reply.header.error_name.as_deref(), + Some("ai.tinyhumans.tinybus.Error.UnknownStream"), + "wrong peer rejected with callback={compatible_callback}" + ); + } else { + assert!(reply.header.error_name.is_none(), "{reply:?}"); + if inspect { + assert_eq!(reply.body, serde_json::json!(false)); + } + } + } + drop(connection); + } +} diff --git a/crates/tinybus/Cargo.toml b/crates/tinybus/Cargo.toml index a41e882..5d6503d 100644 --- a/crates/tinybus/Cargo.toml +++ b/crates/tinybus/Cargo.toml @@ -2,6 +2,7 @@ name = "tinybus" version.workspace = true edition.workspace = true +rust-version.workspace = true license.workspace = true repository.workspace = true build = "build.rs" @@ -118,3 +119,13 @@ required-features = ["modules"] [dev-dependencies.tinybus-module] workspace = true + +[[example]] +name = "module_context" +crate-type = ["cdylib"] +required-features = ["modules"] + +[[example]] +name = "module_context_eager" +crate-type = ["cdylib"] +required-features = ["modules"] diff --git a/crates/tinybus/examples/module_context.rs b/crates/tinybus/examples/module_context.rs new file mode 100644 index 0000000..16fcb34 --- /dev/null +++ b/crates/tinybus/examples/module_context.rs @@ -0,0 +1,60 @@ +//! Native fixture exercising contextual dispatch and a callback on its own runtime. + +use async_trait::async_trait; +use serde_json::Value; +use tinybus::{CallContext, Connection, Error, Interface, InterfaceName, MemberName, Result}; + +struct ContextService(Connection); + +#[async_trait] +impl Interface for ContextService { + fn name(&self) -> InterfaceName { + "org.example.Context".parse().unwrap() + } + fn members(&self) -> Vec { + vec!["Inspect".parse().unwrap(), "Callback".parse().unwrap()] + } + async fn call(&self, _: &MemberName, _: Value) -> Result { + Err(Error::failed("authenticated incoming context required")) + } + async fn call_with_context( + &self, + member: &MemberName, + args: Value, + context: &CallContext, + ) -> Result { + let sender = context + .authenticated_sender() + .ok_or_else(|| Error::failed("unverified caller"))?; + if args != serde_json::json!([]) { + return Err(Error::failed("expected no arguments")); + } + match member.as_str() { + "Inspect" => Ok(serde_json::json!(sender)), + "Callback" => { + let reply: String = self + .0 + .proxy(sender.as_str(), "/callback", "org.example.Callback")? + .call("Observe", ()) + .await?; + Ok(serde_json::json!([sender, reply])) + } + _ => Err(Error::failed("unknown method")), + } + } +} + +async fn setup(connection: Connection) -> Result<()> { + connection + .serve_at("/context".parse()?, ContextService(connection.clone())) + .await?; + connection.request_name("org.example.Context").await +} + +tinybus_module::module_export! { + setup = setup, + worker_threads = 1, + provides = ["org.example.Context"], + methods = ["Inspect", "Callback"], + signals = [], requires = [], optional = [], lazy = true, +} diff --git a/crates/tinybus/examples/module_context_eager.rs b/crates/tinybus/examples/module_context_eager.rs new file mode 100644 index 0000000..f4a92d3 --- /dev/null +++ b/crates/tinybus/examples/module_context_eager.rs @@ -0,0 +1,109 @@ +//! Native fixture exercising contextual dispatch and a callback on its own runtime. + +use async_trait::async_trait; +use serde_json::Value; +use tinybus::{CallContext, Connection, Error, Interface, InterfaceName, MemberName, Result}; + +struct ContextService(Connection); + +#[async_trait] +impl Interface for ContextService { + fn name(&self) -> InterfaceName { + "org.example.Context".parse().unwrap() + } + fn members(&self) -> Vec { + vec!["Inspect".parse().unwrap(), "Callback".parse().unwrap()] + } + async fn call(&self, _: &MemberName, _: Value) -> Result { + Err(Error::failed("authenticated incoming context required")) + } + async fn call_with_context( + &self, + member: &MemberName, + args: Value, + context: &CallContext, + ) -> Result { + let sender = context + .authenticated_sender() + .ok_or_else(|| Error::failed("unverified caller"))?; + if args != serde_json::json!([]) { + return Err(Error::failed("expected no arguments")); + } + match member.as_str() { + "Inspect" => Ok(serde_json::json!(sender)), + "Callback" => { + let reply: String = self + .0 + .proxy(sender.as_str(), "/callback", "org.example.Callback")? + .call("Observe", ()) + .await?; + Ok(serde_json::json!([sender, reply])) + } + _ => Err(Error::failed("unknown method")), + } + } +} + +async fn setup(connection: Connection) -> Result<()> { + connection + .serve_at("/context".parse()?, ContextService(connection.clone())) + .await?; + connection.request_name("org.example.Context").await +} + +tinybus_module::module_export! { + @plain export = {#[unsafe(no_mangle)]}, + start = eager_start, + setup = setup, + worker_threads = 1, + provides = ["org.example.Context"], + methods = ["Inspect", "Callback"], + signals = [], requires = [], optional = [], lazy = false, +} + +// The initializer cannot return (and therefore cannot be admitted) until the +// independent SDK worker has sent Hello. This forces the eager race without +// assuming which thread the scheduler runs first. +static HELLO: (std::sync::Mutex, std::sync::Condvar) = + (std::sync::Mutex::new(false), std::sync::Condvar::new()); +static SEND: std::sync::OnceLock< + unsafe extern "C" fn(*mut std::ffi::c_void, *const u8, usize) -> i32, +> = std::sync::OnceLock::new(); + +unsafe extern "C" fn observe_send(ctx: *mut std::ffi::c_void, ptr: *const u8, len: usize) -> i32 { + let code = unsafe { SEND.get().unwrap()(ctx, ptr, len) }; + *HELLO.0.lock().unwrap() = true; + HELLO.1.notify_one(); + code +} + +unsafe fn eager_start( + host: *const tinybus::module::abi::TbHostVtable, + out: *mut tinybus::module::abi::TbModuleVtable, + workers: usize, + detach: bool, + setup: F, +) -> i32 +where + F: FnOnce(Connection) -> Fut + Send + 'static, + Fut: std::future::Future> + Send + 'static, +{ + let mut table = unsafe { tinybus::module::abi::TbHostVtable::read_compatible(host) }.unwrap(); + // Eager activation has not attached its private transport to the broker. + assert_eq!(unsafe { table.broker_routing.unwrap()(table.host_ctx) }, 0); + SEND.set(table.send).unwrap(); + table.send = observe_send; + let code = unsafe { tinybus_module::start_module(&table, out, workers, detach, setup) }; + if code == tinybus::module::abi::TB_OK { + let hello = HELLO.0.lock().unwrap(); + let (hello, timeout) = HELLO + .1 + .wait_timeout_while(hello, std::time::Duration::from_secs(10), |sent| !*sent) + .unwrap(); + assert!( + *hello && !timeout.timed_out(), + "SDK must send Hello before admission" + ); + } + code +} diff --git a/crates/tinybus/src/attest.rs b/crates/tinybus/src/attest.rs index 8581695..308600e 100644 --- a/crates/tinybus/src/attest.rs +++ b/crates/tinybus/src/attest.rs @@ -22,6 +22,12 @@ //! as acceptable. The hash is computed by the host over bytes the host read //! itself; nothing a module says about itself participates. //! +//! Admission also binds that record to the module's fixed unique peer. A unique +//! lookup retains the original manifest name and digest even after alias release; +//! extra aliases and successor peers inherit no attestation. Detach removes the +//! record. This recipient check is separate from incoming [`crate::CallContext`]: +//! neither an artifact digest nor a peer name authenticates a human. +//! //! # What it does not establish, and this matters //! //! An in-process module shares the host's address space. It can read host @@ -62,9 +68,12 @@ use crate::name::BusName; /// and checked on every confidential delivery. It carries no path: the hash and /// the name it was verified for are enough to audit the decision, and a /// filesystem layout is not something to publish on a bus. +/// +/// Querying an admitted unique peer returns its original manifest identity; +/// `name` does not become the unique destination used for that query. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct Attestation { - /// The well-known name this artifact was verified *for*. + /// The original well-known name this artifact was verified *for*. /// /// Bound to the name rather than floating free, because "some allowlisted /// artifact is loaded" is not the question a sender is asking. The question diff --git a/crates/tinybus/src/bin/tinybus.rs b/crates/tinybus/src/bin/tinybus.rs index 34fa2d9..903ad76 100644 --- a/crates/tinybus/src/bin/tinybus.rs +++ b/crates/tinybus/src/bin/tinybus.rs @@ -504,10 +504,10 @@ fn resolve_address(explicit: Option) -> Result { if let Some(path) = explicit { return Ok(path); } - if let Ok(dir) = std::env::var("XDG_RUNTIME_DIR") - && !dir.is_empty() - { - return Ok(PathBuf::from(dir).join("tinybus").join("bus")); + if let Ok(dir) = std::env::var("XDG_RUNTIME_DIR") { + if !dir.is_empty() { + return Ok(PathBuf::from(dir).join("tinybus").join("bus")); + } } let uid = unsafe { libc_getuid() }; Ok(PathBuf::from(format!("/tmp/tinybus-{uid}/bus"))) diff --git a/crates/tinybus/src/broker.rs b/crates/tinybus/src/broker.rs index 6ef5282..740ac8e 100644 --- a/crates/tinybus/src/broker.rs +++ b/crates/tinybus/src/broker.rs @@ -118,6 +118,12 @@ impl Broker { .expect("router lock is never held across a panic point") .attach(outbox); + #[cfg(feature = "modules")] + if let Some(module) = crate::transport::downcast::( + transport.as_ref(), + ) { + module.admit_broker(); + } tracing::debug!(peer = %unique, transport = %transport.describe(), "peer attached"); tokio::spawn(writer_task(transport.clone(), inbox)); tokio::spawn(reader_task(self.clone(), transport, id, unique.clone())); @@ -207,29 +213,18 @@ impl Broker { )); } - // A confidential *call* may only go to a well-known name the - // broker has verified an artifact for. A confidential *reply* - // goes back to the unique name the broker itself minted for the - // peer that made the call — that peer already chose to take - // part in the exchange, and unique names are never reused, so - // there is no one else the reply could reach. + // Both destination forms resolve and check admission under + // one router lock. Unique names pin the admitted artifact; + // well-known names require an exact attested identity. + // Confidential replies still return to the caller's unique + // connection, which chose to participate in the exchange. let confidential_call = message.header.confidential && message.header.kind == MessageKind::MethodCall; let target = if confidential_call { - if destination.is_unique() { - // The broker knows *which connection* a unique name is, - // but not what binary is behind it. A sender that needs - // that answer has to address the well-known name. - Err(Error::not_attested( - destination.clone(), - "a confidential call must address a well-known name", - )) - } else { - self.router - .lock() - .expect("router lock") - .resolve_attested(&destination) - } + self.router + .lock() + .expect("router lock") + .resolve_attested(&destination) } else { self.router .lock() @@ -256,13 +251,13 @@ impl Broker { }); let target = target?; // A stopped, disabled, failed or faulted module can keep its - // well-known name for a moment after the host settles its - // state: the peer detaches only once its reader sees the + // unique connection and well-known name briefly after the host + // settles its state: the peer detaches only once its reader sees the // closed transport. A call routed into that window would sit // in a queue nobody drains and the caller would wait out its // whole deadline, so answer from the host's state instead. #[cfg(feature = "modules")] - if message.header.kind == MessageKind::MethodCall && !destination.is_unique() { + if message.header.kind == MessageKind::MethodCall { let control = self .modules .lock() @@ -678,7 +673,15 @@ fn parse_args(member: &MemberName, body: Value) /// Drain one peer's queue onto its transport. async fn writer_task(transport: Arc, mut inbox: mpsc::Receiver) { while let Some(message) = inbox.recv().await { - if let Err(e) = transport.send(message).await { + let result = if let Some(memory) = crate::transport::downcast::< + crate::transport::memory::MemoryTransport, + >(transport.as_ref()) + { + memory.send_brokered(message).await + } else { + transport.send(message).await + }; + if let Err(e) = result { tracing::debug!(error = %e, "peer write failed; dropping the peer"); break; } diff --git a/crates/tinybus/src/broker_tests.rs b/crates/tinybus/src/broker_tests.rs index 8da0d69..72996d5 100644 --- a/crates/tinybus/src/broker_tests.rs +++ b/crates/tinybus/src/broker_tests.rs @@ -487,9 +487,8 @@ async fn a_confidential_call_to_an_unattested_recipient_is_refused() { } #[tokio::test] -async fn a_confidential_call_may_not_address_a_unique_name() { - // The broker knows which connection `:1.n` is, but not what binary is - // behind it, so it cannot answer the question the sender is asking. +async fn a_confidential_call_to_an_unattested_unique_peer_is_refused() { + // An ordinary connection has no admitted artifact, even at its unique name. let (_bus, service, client) = bus().await; let unique = service.unique_name().unwrap(); let proxy = client @@ -563,6 +562,14 @@ async fn a_name_handed_on_to_another_peer_does_not_hand_on_its_attestation() { // the real module released it would inherit the right to be handed // secrets without a single byte having been hashed. let (bus, _broker, service, client) = attested_bus().await; + let fixed = client + .proxy( + service.unique_name().unwrap().as_str(), + VOICE_PATH, + VOICE_NAME, + ) + .unwrap(); + let admitted = fixed.attestation().await.unwrap(); let voice = client.proxy(VOICE_NAME, VOICE_PATH, VOICE_NAME).unwrap(); assert!(voice.attestation().await.unwrap().is_some()); @@ -576,6 +583,43 @@ async fn a_name_handed_on_to_another_peer_does_not_hand_on_its_attestation() { .unwrap(); impostor.request_name(VOICE_NAME).await.unwrap(); + assert_eq!(fixed.attestation().await.unwrap(), admitted); + assert!( + fixed + .call_confidential::("Transcribe", ("/tmp/secret.wav",)) + .await + .is_ok() + ); + let replacement = client + .proxy( + impostor.unique_name().unwrap().as_str(), + VOICE_PATH, + VOICE_NAME, + ) + .unwrap(); + assert_eq!(replacement.attestation().await.unwrap(), None); + assert_eq!( + replacement + .call_confidential::("Transcribe", ("/tmp/secret.wav",)) + .await + .unwrap_err() + .wire_name(), + Error::NOT_ATTESTED + ); + service.request_name("org.example.Extra").await.unwrap(); + let extra = client + .proxy("org.example.Extra", VOICE_PATH, VOICE_NAME) + .unwrap(); + assert_eq!(extra.attestation().await.unwrap(), None); + assert_eq!( + extra + .call_confidential::("Transcribe", ("/tmp/secret.wav",)) + .await + .unwrap_err() + .wire_name(), + Error::NOT_ATTESTED + ); + // The impostor owns the name and answers ordinary calls... assert!( voice @@ -590,6 +634,30 @@ async fn a_name_handed_on_to_another_peer_does_not_hand_on_its_attestation() { .await .unwrap_err(); assert_eq!(error.wire_name(), Error::NOT_ATTESTED); + + let mut changes = client + .add_match( + MatchRule::new() + .signals() + .member(MemberName::new("NameOwnerChanged").unwrap()), + ) + .await + .unwrap(); + service.close().await.unwrap(); + let detached = tokio::time::timeout(Duration::from_secs(5), changes.recv()) + .await + .unwrap() + .unwrap(); + assert_eq!(detached.body[0], "org.example.Extra"); + assert_eq!(fixed.attestation().await.unwrap(), None); + assert_eq!( + fixed + .call_confidential::("Transcribe", ("/tmp/secret.wav",)) + .await + .unwrap_err() + .wire_name(), + "ai.tinyhumans.tinybus.Error.NameHasNoOwner" + ); } #[tokio::test] @@ -816,3 +884,28 @@ async fn module_control_calls_validate_their_arguments_and_answer_for_unknown_mo assert_eq!(error.wire_name(), bad_arguments, "{member}: {error}"); } } + +#[cfg(feature = "modules")] +#[tokio::test] +async fn a_confidential_call_reaches_the_fixed_unique_admitted_peer() { + let (_bus, _broker, service, client) = attested_bus().await; + let proxy = client + .proxy( + service.unique_name().unwrap().as_str(), + VOICE_PATH, + VOICE_NAME, + ) + .unwrap(); + let transcript: String = proxy + .call_confidential("Transcribe", ("/tmp/secret.wav",)) + .await + .unwrap(); + assert_eq!(transcript, "transcript of /tmp/secret.wav"); + assert_eq!( + proxy.attestation().await.unwrap(), + client + .attestation(BusName::new(VOICE_NAME).unwrap()) + .await + .unwrap() + ); +} diff --git a/crates/tinybus/src/connection.rs b/crates/tinybus/src/connection.rs index 747dd14..5aa246a 100644 --- a/crates/tinybus/src/connection.rs +++ b/crates/tinybus/src/connection.rs @@ -27,7 +27,7 @@ use std::future::Future; use std::io::{self, Write}; use std::pin::Pin; use std::sync::Arc; -use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::task::{Context, Poll}; use std::time::Duration; @@ -143,6 +143,8 @@ pub const OUTBOX_CAPACITY: usize = 1024; struct Inner { transport: Arc, + brokered_native: bool, + native_provenance: Option>, /// Everything outbound goes through here and out via the writer task. /// Serialising writes through one task is also what lets `send` be called /// concurrently without interleaving two frames on the wire. @@ -157,7 +159,7 @@ struct Inner { panic_handler: std::sync::RwLock Error + Send + Sync>>>, /// Bulk streams being received. On the connection rather than in the object /// tree because a chunk has to be checked against the header's stamped - /// `sender`, and [`Interface`] deliberately never sees a header. + /// `sender`; interfaces receive only the verified subset in CallContext. streams: StreamRegistry, } @@ -239,11 +241,56 @@ impl Connection { /// Used by the broker for its own side of a peer link, and by tests that /// drive the protocol by hand. Ordinary callers want /// [`Connection::connect`]. + /// + /// This safe constructor grants no native routing authority. Concrete + /// memory links authenticate only deliveries sealed by the broker writer; + /// a public `Transport::send`, wrapper, or successful Hello cannot forge it. pub fn attach(transport: Arc) -> Self { + Self::attach_inner(transport, false, None) + } + + /// Attach the module SDK's broker-proven C transport bridge. + /// + /// # Safety + /// The caller must guarantee every received frame comes exclusively from + /// a real TinyBus broker which overwrites sender on ingress. A successful + /// Hello, artifact attestation, or arbitrary C callback is insufficient. + /// This boundary is an unsafe trusted-host contract, not cryptographic + /// authentication against a malicious native host. + #[doc(hidden)] + pub unsafe fn __attach_brokered_module(transport: Arc) -> Self { + Self::attach_inner(transport, true, None) + } + + /// Attach the SDK bridge with a separately captured proof for each frame. + /// + /// # Safety + /// The transport must have exactly one reader: this connection's dispatch + /// loop. Before returning each received frame it must publish that frame's + /// immutable delivery proof into `provenance`; nobody else may write the + /// slot. True must mean the trusted host asserted exclusive real broker + /// delivery when that frame entered the SDK queue, never later admission. + /// Header identity remains protocol data unless that proof is true. + /// This is a trusted native host contract, not cryptographic authentication. + #[doc(hidden)] + pub unsafe fn __attach_module_with_provenance( + transport: Arc, + provenance: Arc, + ) -> Self { + Self::attach_inner(transport, false, Some(provenance)) + } + + fn attach_inner( + transport: Arc, + brokered_native: bool, + native_provenance: Option>, + ) -> Self { let (signals, _) = broadcast::channel(SIGNAL_BUFFER); let (outbox, outbound) = mpsc::channel(OUTBOX_CAPACITY); let inner = Arc::new(Inner { transport, + brokered_native, + native_provenance, outbox, // Serials start at 1: zero is the "unassigned" value a freshly // built `Message` carries, so it must never be a live serial. @@ -343,6 +390,11 @@ impl Connection { /// What the broker has verified about whoever owns `name`. /// + /// Accepts either an exact well-known name or a fixed unique peer. For an + /// admitted unique peer, the record retains the artifact's original manifest + /// name and digest across alias release; extra aliases remain unattested. + /// A name handover gives the successor no record, and detach removes it. + /// /// The check a sender makes before it commits to handing over a secret. /// `None` means the bus will refuse a confidential message to that name — /// nothing owns it, or the operator never allowlisted an artifact for it. @@ -1119,8 +1171,26 @@ async fn writer_loop(transport: Arc, mut outbound: mpsc::Receiver /// The single reader task. Owns `recv`; never awaits user code inline. async fn dispatch_loop(inner: Arc) { loop { - let message = match inner.transport.recv().await { - Ok(Some(message)) => message, + let delivery = if let Some(memory) = crate::transport::downcast::< + crate::transport::memory::MemoryTransport, + >(inner.transport.as_ref()) + { + memory.recv_with_provenance().await + } else { + inner.transport.recv().await.map(|message| { + message.map(|message| { + // The exclusive reader samples the proof immediately after + // recv, before any await or next recv can replace its slot. + let brokered = inner + .native_provenance + .as_ref() + .map_or(inner.brokered_native, |proof| proof.load(Ordering::Acquire)); + (message, brokered) + }) + }) + }; + let (message, brokered) = match delivery { + Ok(Some(delivery)) => delivery, Ok(None) => break, Err(e) => { tracing::debug!(error = %e, "connection read failed; closing"); @@ -1148,7 +1218,7 @@ async fn dispatch_loop(inner: Arc) { MessageKind::MethodCall => { // Spawned, so a long-running method does not block replies to // calls this connection has outstanding. - tokio::spawn(handle_call(inner.clone(), message)); + tokio::spawn(handle_call(inner.clone(), message, brokered)); } } } @@ -1159,7 +1229,7 @@ async fn dispatch_loop(inner: Arc) { } /// Run one inbound method call and send its reply. -async fn handle_call(inner: Arc, message: Message) { +async fn handle_call(inner: Arc, message: Message, brokered: bool) { let header = message.header.clone(); let panic_handler = inner .panic_handler @@ -1171,12 +1241,12 @@ async fn handle_call(inner: Arc, message: Message) { "a confidential call cannot request a streamed reply", )) } else if let Some(panic_handler) = panic_handler { - match CatchUnwind::new(dispatch(&inner, &header, message.body)).await { + match CatchUnwind::new(dispatch(&inner, &header, message.body, brokered)).await { Ok(result) => result, Err(()) => Err(panic_handler()), } } else { - dispatch(&inner, &header, message.body).await + dispatch(&inner, &header, message.body, brokered).await }; let result = match result { @@ -1282,7 +1352,7 @@ impl Future for CatchUnwind { } } -async fn dispatch(inner: &Inner, header: &Header, body: Value) -> Result { +async fn dispatch(inner: &Inner, header: &Header, body: Value, brokered: bool) -> Result { // Streams are answered before the object tree is consulted, and without the // service having exported anything: bulk transfer is bus plumbing, and a // service that forgot to export it would be a service you cannot send a @@ -1297,10 +1367,23 @@ async fn dispatch(inner: &Inner, header: &Header, body: Value) -> Result else { return Err(Error::protocol("method call is missing an address")); }; - let objects = inner.objects.read().await; - objects - .dispatch_with_confidential(path, interface, member, body, header.confidential) - .await + let context = if brokered { + crate::CallContext::brokered(header.sender.as_ref()) + } else { + crate::CallContext::default() + }; + // Snapshot admission under the lock, then release it before user code: + // callbacks can register/unserve interfaces while another call is parked. + let target = inner.objects.read().await.lookup(path, interface)?; + ObjectTree::invoke( + target, + interface, + member, + body, + header.confidential, + &context, + ) + .await } #[cfg(test)] diff --git a/crates/tinybus/src/connection_tests.rs b/crates/tinybus/src/connection_tests.rs index c820c92..c09a882 100644 --- a/crates/tinybus/src/connection_tests.rs +++ b/crates/tinybus/src/connection_tests.rs @@ -415,3 +415,302 @@ async fn module_management_helpers_forward_their_requests_to_the_broker() { ); assert!(connection.enable_module("missing", true).await.is_err()); } + +struct ContextEcho; +#[async_trait] +impl Interface for ContextEcho { + fn name(&self) -> InterfaceName { + Echo.name() + } + fn members(&self) -> Vec { + vec![MemberName::new("Echo").unwrap()] + } + async fn call(&self, _: &MemberName, _: Value) -> Result { + Err(Error::failed("context override was lost")) + } + async fn call_with_context( + &self, + _: &MemberName, + args: Value, + context: &crate::CallContext, + ) -> Result { + Ok(serde_json::json!([context.authenticated_sender(), args])) + } +} + +#[tokio::test] +async fn brokered_context_authenticates_distinct_senders_and_overwrites_forgery() { + let bus = MemoryBus::new(); + Broker::new().spawn(bus.clone()); + let service = Connection::connect(bus.connect().await.unwrap()) + .await + .unwrap(); + service.request_name("ai.tinyhumans.Test").await.unwrap(); + service + .serve_at(path(), Arc::new(ContextEcho)) + .await + .unwrap(); + let first = Connection::connect(bus.connect().await.unwrap()) + .await + .unwrap(); + let second = Connection::connect(bus.connect().await.unwrap()) + .await + .unwrap(); + assert_ne!(first.unique_name(), second.unique_name()); + for caller in [&first, &second] { + let mut request = call("Echo", serde_json::json!([42])); + request.header.sender = first.unique_name(); + let result = caller.call_raw(request, DEFAULT_TIMEOUT).await.unwrap(); + assert_eq!(result, serde_json::json!([caller.unique_name(), [42]])); + } +} + +#[tokio::test] +async fn direct_header_sender_is_not_authenticated() { + let (a, b) = MemoryTransport::pair(); + let service = Connection::attach(Arc::new(b)); + service.serve_at(path(), ContextEcho).await.unwrap(); + let client = Connection::attach(Arc::new(a)); + let mut request = call("Echo", serde_json::json!([])); + request.header.sender = Some(BusName::new(":1.42").unwrap()); + let result = client.call_raw(request, DEFAULT_TIMEOUT).await.unwrap(); + assert_eq!(result, serde_json::json!([null, []])); +} + +struct RegisterDuringCall(Connection); +#[async_trait] +impl Interface for RegisterDuringCall { + fn name(&self) -> InterfaceName { + Echo.name() + } + fn members(&self) -> Vec { + vec![MemberName::new("Echo").unwrap()] + } + async fn call(&self, _: &MemberName, args: Value) -> Result { + self.0 + .serve_at(ObjectPath::new("/replacement").unwrap(), Echo) + .await?; + Ok(args) + } +} + +#[tokio::test] +async fn inbound_callback_can_register_an_interface_without_deadlock() { + let (client, service) = pair().await; + service + .serve_at(path(), RegisterDuringCall(service.clone())) + .await + .unwrap(); + let result = client + .call_raw( + call("Echo", serde_json::json!([1])), + Duration::from_millis(500), + ) + .await; + assert_eq!(result.unwrap(), serde_json::json!([1])); + service.unserve(&path()).await; +} + +struct CaptureContext(mpsc::UnboundedSender>); +#[async_trait] +impl Interface for CaptureContext { + fn name(&self) -> InterfaceName { + Echo.name() + } + fn members(&self) -> Vec { + vec![MemberName::new("Echo").unwrap()] + } + async fn call(&self, _: &MemberName, _: Value) -> Result { + unreachable!() + } + async fn call_with_context( + &self, + _: &MemberName, + _: Value, + context: &crate::CallContext, + ) -> Result { + self.0 + .send(context.authenticated_sender().cloned()) + .unwrap(); + Ok(Value::Null) + } +} + +#[tokio::test] +async fn retaining_a_broker_endpoint_does_not_allow_authenticated_injection() { + let (peer, broker_end) = MemoryTransport::pair(); + let broker_end = Arc::new(broker_end); + Broker::new().attach(broker_end.clone()); + let service = Connection::connect(Box::new(peer)).await.unwrap(); + let (tx, mut rx) = mpsc::unbounded_channel(); + service.serve_at(path(), CaptureContext(tx)).await.unwrap(); + let mut forged = call("Echo", serde_json::json!([])); + forged.header.sender = Some(BusName::new(":1.42").unwrap()); + broker_end.send(forged).await.unwrap(); + assert_eq!( + tokio::time::timeout(Duration::from_secs(1), rx.recv()) + .await + .unwrap() + .unwrap(), + None + ); +} + +struct LyingTransport { + inner: MemoryTransport, +} +impl LyingTransport { + // This is exactly why provenance does not use a user-overridable as_any. + fn as_any(&self) -> &dyn std::any::Any { + &self.inner + } +} +#[async_trait] +impl Transport for LyingTransport { + async fn send(&self, message: Message) -> Result<()> { + self.inner.send(message).await + } + async fn recv(&self) -> Result> { + let mut message = self.inner.recv().await?; + if let Some(message) = message + .as_mut() + .filter(|message| message.header.kind == MessageKind::MethodCall) + { + message.header.sender = Some(BusName::new(":1.42").unwrap()); + } + Ok(message) + } + async fn close(&self) -> Result<()> { + self.inner.close().await + } + fn describe(&self) -> String { + "memory".into() + } +} + +#[tokio::test] +async fn a_custom_transport_cannot_borrow_its_inner_broker_provenance() { + let bus = MemoryBus::new(); + let broker = Broker::new(); + broker.spawn(bus.clone()); + let (peer, broker_end) = MemoryTransport::pair(); + broker.attach(Arc::new(broker_end)); + let wrapper = LyingTransport { inner: peer }; + assert!(wrapper.as_any().is::()); + let service = Connection::connect(Box::new(wrapper)).await.unwrap(); + let (tx, mut rx) = mpsc::unbounded_channel(); + service.serve_at(path(), CaptureContext(tx)).await.unwrap(); + service.request_name("ai.tinyhumans.Test").await.unwrap(); + let client = Connection::connect(bus.connect().await.unwrap()) + .await + .unwrap(); + client + .send(call("Echo", serde_json::json!([]))) + .await + .unwrap(); + assert_eq!( + tokio::time::timeout(Duration::from_secs(1), rx.recv()) + .await + .unwrap() + .unwrap(), + None + ); +} + +#[tokio::test] +async fn replacing_a_well_known_name_does_not_inherit_the_old_sender_identity() { + let bus = MemoryBus::new(); + Broker::new().spawn(bus.clone()); + let service = Connection::connect(bus.connect().await.unwrap()) + .await + .unwrap(); + service.request_name("ai.tinyhumans.Test").await.unwrap(); + service.serve_at(path(), ContextEcho).await.unwrap(); + let first = Connection::connect(bus.connect().await.unwrap()) + .await + .unwrap(); + first.request_name("ai.tinyhumans.Operator").await.unwrap(); + let old = first.unique_name().unwrap(); + first.release_name("ai.tinyhumans.Operator").await.unwrap(); + first.close().await.unwrap(); + let replacement = Connection::connect(bus.connect().await.unwrap()) + .await + .unwrap(); + replacement + .request_name("ai.tinyhumans.Operator") + .await + .unwrap(); + let current = replacement.unique_name().unwrap(); + assert_ne!(old, current); + assert_eq!( + service.name_owner("ai.tinyhumans.Operator").await.unwrap(), + Some(current.clone()) + ); + let mut request = call("Echo", serde_json::json!([])); + request.header.sender = Some(old); + let result = replacement + .call_raw(request, DEFAULT_TIMEOUT) + .await + .unwrap(); + assert_eq!(result, serde_json::json!([current, []])); +} + +struct ParkedSnapshot { + started: Arc, + release: Arc, +} +#[async_trait] +impl Interface for ParkedSnapshot { + fn name(&self) -> InterfaceName { + Echo.name() + } + fn members(&self) -> Vec { + vec!["Echo".parse().unwrap()] + } + async fn call(&self, _: &MemberName, args: Value) -> Result { + self.started.notify_one(); + self.release.notified().await; + Ok(args) + } +} + +#[tokio::test] +async fn unserve_stops_new_admission_but_preserves_an_already_admitted_call() { + let (client, service) = pair().await; + let started = Arc::new(tokio::sync::Notify::new()); + let release = Arc::new(tokio::sync::Notify::new()); + service + .serve_at( + path(), + ParkedSnapshot { + started: started.clone(), + release: release.clone(), + }, + ) + .await + .unwrap(); + let active_client = client.clone(); + let active = tokio::spawn(async move { + active_client + .call_raw(call("Echo", serde_json::json!([1])), DEFAULT_TIMEOUT) + .await + }); + tokio::time::timeout(Duration::from_secs(1), started.notified()) + .await + .unwrap(); + assert!( + tokio::time::timeout(Duration::from_secs(1), service.unserve(&path())) + .await + .unwrap() + ); + let rejected = client + .call_raw(call("Echo", serde_json::json!([2])), DEFAULT_TIMEOUT) + .await + .unwrap_err(); + assert_eq!( + rejected.wire_name(), + "ai.tinyhumans.tinybus.Error.UnknownObject" + ); + release.notify_one(); + assert_eq!(active.await.unwrap().unwrap(), serde_json::json!([1])); +} diff --git a/crates/tinybus/src/events/subscriber.rs b/crates/tinybus/src/events/subscriber.rs index d8637a6..5011091 100644 --- a/crates/tinybus/src/events/subscriber.rs +++ b/crates/tinybus/src/events/subscriber.rs @@ -162,8 +162,9 @@ pub(crate) fn spawn( continue; }; - if let Some(allowed) = &domains - && !allowed.iter().any(|d| d == event.domain()) + if domains + .as_ref() + .is_some_and(|allowed| !allowed.iter().any(|d| d == event.domain())) { continue; } diff --git a/crates/tinybus/src/lib.rs b/crates/tinybus/src/lib.rs index 0fc78fa..fc8d19b 100644 --- a/crates/tinybus/src/lib.rs +++ b/crates/tinybus/src/lib.rs @@ -105,7 +105,7 @@ pub use crate::ports::{Listener, Transport}; pub use crate::proxy::Proxy; pub use crate::router::MatchRule; pub use crate::secret::{Secret, harden_process}; -pub use crate::service::Interface; +pub use crate::service::{CallContext, Interface}; pub use crate::stream::{ MAX_CHUNK_LEN, StreamDescriptor, StreamLimits, StreamReader, StreamRef, StreamWriter, }; diff --git a/crates/tinybus/src/message/mod.rs b/crates/tinybus/src/message/mod.rs index 78c1388..67355c2 100644 --- a/crates/tinybus/src/message/mod.rs +++ b/crates/tinybus/src/message/mod.rs @@ -59,8 +59,9 @@ pub struct Header { pub reply_serial: Option, /// Stamped by the broker, never trusted from the peer. /// - /// A service authorising a call reads this field; if peers could set it, - /// every authorisation decision on the bus would be forgeable. + /// Caller authorization uses `CallContext::authenticated_sender`, which + /// also verifies delivery provenance. Direct transports can supply arbitrary + /// headers, so this wire field alone is not authenticated authority. #[serde(default, skip_serializing_if = "Option::is_none")] pub sender: Option, /// Who the message is for. `None` on signals, which are broadcast. diff --git a/crates/tinybus/src/module/abi.rs b/crates/tinybus/src/module/abi.rs index fc68daa..b9ffeec 100644 --- a/crates/tinybus/src/module/abi.rs +++ b/crates/tinybus/src/module/abi.rs @@ -1,4 +1,8 @@ //! Stable, C-shaped module ABI. No Rust-layout value crosses this boundary. +//! +//! ABI revision one grows through optional vtable tails. Read the frozen prefix +//! first and check its declared size before reading a tail; an older host has no +//! broker-routing callback and therefore supplies no authenticated call context. use std::ffi::c_void; @@ -101,6 +105,34 @@ impl TbAbiDescriptor { } } +/// Frozen revision-one host vtable prefix, accepted from older hosts. +/// +/// Its layout ends before [`TbHostVtable::broker_routing`]. Older modules can +/// read this prefix from a newer host without interpreting the additive tail. +#[repr(C)] +#[derive(Clone, Copy)] +pub struct TbHostVtableV1Prefix { + /// Size of this vtable, allowing additive growth. + pub size: u32, + /// Reserved and zero in v1. + pub _reserved: u32, + /// Opaque host context, never dereferenced by the module. + pub host_ctx: *mut c_void, + /// Send one complete JSON message frame to the host. + pub send: unsafe extern "C" fn(*mut c_void, *const u8, usize) -> i32, + /// Tell the host the module's inbound queue has room again. + pub wake: unsafe extern "C" fn(*mut c_void), + /// Forward one log record to the host. + pub log: unsafe extern "C" fn(*mut c_void, u32, *const u8, usize), + /// Mark the module failed and detach it. + pub fault: unsafe extern "C" fn(*mut c_void, *const u8, usize), + /// Borrowed JSON configuration. The module must copy it during init and + /// must not retain this pointer. + pub config: TbSlice, + /// Module setup completed and its declared bus surface is ready. + pub ready: unsafe extern "C" fn(*mut c_void), +} + /// Calls from a module into its host. #[repr(C)] #[derive(Clone, Copy)] @@ -124,6 +156,58 @@ pub struct TbHostVtable { pub config: TbSlice, /// Module setup completed and its declared bus surface is ready. pub ready: unsafe extern "C" fn(*mut c_void), + /// Optional ABI-v1 routing assertion, read only when `size` covers the tail. + /// + /// The SDK samples this callback when each frame enters its inbound queue. + /// Exactly 1 asserts exclusive delivery from the actual broker pipeline, + /// which overwrites sender on ingress; other values are unverified. Later + /// admission does not authenticate a frame already queued without proof. + /// Older hosts omit this field and therefore supply unverified context. + /// The SDK trusts this callback under the unsafe module-init host contract; + /// it is not authentication of arbitrary user-provided C vtables. + pub broker_routing: Option u32>, +} + +/// Size required from an older host before reading its frozen prefix. +pub const TB_HOST_VTABLE_BASE_SIZE: u32 = size_of::() as u32; + +impl TbHostVtable { + /// Copy the available host prefix and normalize an absent additive tail. + /// + /// Returns `None` for a null pointer or a size below + /// [`TB_HOST_VTABLE_BASE_SIZE`]. A prefix-only or incomplete-tail host is + /// accepted with `broker_routing: None`; only a full tail is copied. + /// + /// # Safety + /// A non-null `host` must be aligned and readable for its first size field + /// and every byte that size declares. Its callbacks must be valid and + /// thread-safe, and their context must outlive the module runtime. An + /// affirmative routing callback (returning exactly 1) must truthfully + /// guarantee exclusive broker delivery at the time each frame is queued. + #[doc(hidden)] + pub unsafe fn read_compatible(host: *const Self) -> Option { + if host.is_null() || unsafe { host.cast::().read() } < TB_HOST_VTABLE_BASE_SIZE { + return None; + } + let base = unsafe { host.cast::().read() }; + let broker_routing = if base.size as usize >= size_of::() { + unsafe { std::ptr::addr_of!((*host).broker_routing).read() } + } else { + None + }; + Some(Self { + size: base.size, + _reserved: base._reserved, + host_ctx: base.host_ctx, + send: base.send, + wake: base.wake, + log: base.log, + fault: base.fault, + config: base.config, + ready: base.ready, + broker_routing, + }) + } } /// Calls from a host into one initialized module. diff --git a/crates/tinybus/src/module/host.rs b/crates/tinybus/src/module/host.rs index 99d48da..112458d 100644 --- a/crates/tinybus/src/module/host.rs +++ b/crates/tinybus/src/module/host.rs @@ -369,6 +369,10 @@ impl ModuleHost { } /// Load one newly installed module. + /// + /// A verified operator allowlist admits confidential delivery to the exact + /// manifest name and this module's fixed unique peer. It does not attest + /// extra aliases or grant caller authority. See [`crate::attest`]. pub fn load_file(&self, path: impl AsRef) -> Result { self.load_file_with_config(path, serde_json::json!({})) } @@ -482,6 +486,9 @@ impl ModuleHost { /// This is the testable seam beneath `dlopen`: callers must ensure `init` /// and every pointer reachable through it remain valid for the process /// lifetime, exactly as the real loader does by leaking its handle. + /// No digest is supplied by this API: any recipient attestation comes from + /// an allowlist beside `file`, not from the descriptor or callback itself. + /// Broker attachment establishes delivery provenance separately. /// /// # Safety /// @@ -528,7 +535,9 @@ impl ModuleHost { /// /// The executable's digest identifies the admitted code in the ordinary /// attestation response. This is host trust, not a digest of a separate - /// library or a claim made by the module itself. + /// library or a claim made by the module itself. The record applies to the + /// manifest name and fixed unique peer, not additional aliases; recipient + /// admission alone does not grant incoming callers permission. /// /// # Errors /// Returns an error if the host executable cannot be hashed or the module @@ -1582,7 +1591,13 @@ impl ModuleControl for ModuleHostInner { let loaded = self.loaded.lock().expect("module list lock"); let loaded_info = loaded .iter() - .find(|module| &module.info.manifest.bus_name == bus_name) + .find(|module| { + if bus_name.is_unique() { + &module.unique_name == bus_name + } else { + &module.info.manifest.bus_name == bus_name + } + }) .map(LoadedModule::snapshot); drop(loaded); let info = loaded_info.or_else(|| { diff --git a/crates/tinybus/src/module/host_tests.rs b/crates/tinybus/src/module/host_tests.rs index 5c3d03d..485e31d 100644 --- a/crates/tinybus/src/module/host_tests.rs +++ b/crates/tinybus/src/module/host_tests.rs @@ -2246,6 +2246,21 @@ async fn a_module_loaded_from_an_allowlisted_artifact_becomes_an_attested_recipi .expect("an allowlisted module is attested"); assert_eq!(attestation.sha256, hash); assert_eq!(attestation.name, info.manifest.bus_name); + let unique = client + .name_owner(info.manifest.bus_name) + .await + .unwrap() + .unwrap(); + let proxy = client + .proxy( + unique.as_str(), + "/ai/tinyhumans/openhuman/Clock", + "ai.tinyhumans.openhuman.Clock", + ) + .unwrap(); + assert_eq!(proxy.attestation().await.unwrap(), Some(attestation)); + let now: String = proxy.call_confidential("Now", ()).await.unwrap(); + assert!(!now.is_empty()); } #[cfg(unix)] @@ -2348,3 +2363,171 @@ async fn a_module_with_neither_a_pin_nor_an_allowlist_is_loaded_but_never_attest "a module nobody vouched for must not be eligible to receive a secret" ); } + +struct NativeContextCallback; +#[async_trait::async_trait] +impl crate::Interface for NativeContextCallback { + fn name(&self) -> crate::InterfaceName { + "org.example.Callback".parse().unwrap() + } + fn members(&self) -> Vec { + vec!["Observe".parse().unwrap()] + } + async fn call( + &self, + _: &crate::MemberName, + _: serde_json::Value, + ) -> crate::Result { + Err(Error::failed("missing contextual dispatch")) + } + async fn call_with_context( + &self, + _: &crate::MemberName, + args: serde_json::Value, + context: &crate::CallContext, + ) -> crate::Result { + assert_eq!(args, serde_json::json!([])); + Ok(serde_json::json!( + context + .authenticated_sender() + .expect("native callback is broker routed") + )) + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[ignore = "requires TINYBUS_TEST_CONTEXT_MODULE to point at the built cdylib"] +async fn a_real_cdylib_authenticates_incoming_calls_and_its_host_callback() { + authenticates_native_context("TINYBUS_TEST_CONTEXT_MODULE").await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[ignore = "requires TINYBUS_TEST_CONTEXT_EAGER_MODULE to point at the built cdylib"] +async fn an_eager_cdylib_authenticates_after_its_worker_runs_before_admission() { + authenticates_native_context("TINYBUS_TEST_CONTEXT_EAGER_MODULE").await; +} + +async fn authenticates_native_context(variable: &str) { + let artifact = PathBuf::from(std::env::var_os(variable).expect(variable)); + let hash = crate::module::hash::file_hex(std::fs::File::open(&artifact).unwrap()).unwrap(); + #[cfg(unix)] + let (_dir, artifact) = staged_module(&artifact, &hash); + let bus = MemoryBus::new(); + let broker = Broker::new(); + let task = broker.spawn(bus.clone()); + let modules = ModuleHost::new(broker); + let info = modules.load_file(artifact).unwrap(); + let client = Connection::connect(bus.connect().await.unwrap()) + .await + .unwrap(); + client + .serve_at("/callback".parse().unwrap(), NativeContextCallback) + .await + .unwrap(); + let unique = client + .name_owner("org.example.Context") + .await + .unwrap() + .unwrap(); + let proxy = client + .proxy(unique.as_str(), "/context", "org.example.Context") + .unwrap(); + assert_eq!( + proxy.attestation().await.unwrap(), + Some(crate::attest::Attestation { + name: info.manifest.bus_name, + sha256: hash + }) + ); + let sender: crate::BusName = proxy.call_confidential("Inspect", ()).await.unwrap(); + assert_eq!(Some(sender.clone()), client.unique_name()); + let module = client + .name_owner("org.example.Context") + .await + .unwrap() + .unwrap(); + let callback: (crate::BusName, crate::BusName) = + proxy.call_confidential("Callback", ()).await.unwrap(); + assert_eq!(callback, (sender, module)); + task.abort(); +} + +#[tokio::test] +async fn fixed_unique_module_calls_are_refused_in_terminal_states_before_detach() { + let _test_guard = FAKE_MODULE_TEST_LOCK.lock().await; + let bus = MemoryBus::new(); + let broker = Broker::new(); + let broker_task = broker.spawn(bus.clone()); + let host = ModuleHost::new(broker.clone()); + let mut lazy_manifest = manifest(); + lazy_manifest.lazy_init = true; + unsafe { + host.attach_raw( + "clock.so", + TbAbiDescriptor::current("clock", "0.1.0"), + lazy_manifest, + lazy_echo_init, + ) + } + .unwrap(); + let unique = host.inner.loaded.lock().unwrap()[0].unique_name.clone(); + broker.attest_module( + &unique, + crate::attest::Attestation { + name: manifest().bus_name, + sha256: "a".repeat(64), + }, + ); + let client = Connection::connect(bus.connect().await.unwrap()) + .await + .unwrap(); + let proxy = client + .proxy( + unique.as_str(), + "/ai/tinyhumans/module/Clock", + "ai.tinyhumans.module.Clock", + ) + .unwrap(); + for (state, expected) in [ + (ModuleState::Stopped, "stopped"), + (ModuleState::Disabled, "disabled"), + ( + ModuleState::Faulted { + reason: "fixture".into(), + }, + "faulted", + ), + ( + ModuleState::Failed { + reason: "fixture".into(), + }, + "failed", + ), + ] { + // Keep the real transport attached while setting its terminal state: + // this probes the pre-detach window without racing the reader task. + host.inner.loaded.lock().unwrap()[0].info.state = state; + assert!( + host.inner + .unavailable_for(&client.unique_name().unwrap()) + .is_none() + ); + assert!( + matches!(host.inner.unavailable_for(&unique), Some(Error::ModuleUnavailable { state, .. }) if state == expected) + ); + for confidential in [false, true] { + let result = if confidential { + proxy + .call_confidential::("Echo", ("terminal",)) + .await + } else { + proxy.call::("Echo", ("terminal",)).await + }; + assert_eq!( + result.unwrap_err().wire_name(), + "ai.tinyhumans.tinybus.Error.ModuleUnavailable" + ); + } + } + broker_task.abort(); +} diff --git a/crates/tinybus/src/module/mod.rs b/crates/tinybus/src/module/mod.rs index 34eb131..40e1e21 100644 --- a/crates/tinybus/src/module/mod.rs +++ b/crates/tinybus/src/module/mod.rs @@ -26,7 +26,7 @@ mod remembered_hash; #[cfg(feature = "modules")] mod resolve; #[cfg(feature = "modules")] -mod transport; +pub(crate) mod transport; #[cfg(feature = "modules")] mod windows_acl; diff --git a/crates/tinybus/src/module/remembered_hash.rs b/crates/tinybus/src/module/remembered_hash.rs index 81df13f..f4a9e6e 100644 --- a/crates/tinybus/src/module/remembered_hash.rs +++ b/crates/tinybus/src/module/remembered_hash.rs @@ -75,18 +75,18 @@ mod imp { pub(super) fn hex(file: File, age: Duration) -> io::Result<(String, bool)> { let before = identity(&file)?; let eligible = quiet_for(&before, age); - if eligible && let Some(known) = memory().lock().ok().and_then(|m| m.get(&before).cloned()) - { - return Ok((known, true)); + if eligible { + if let Some(known) = memory().lock().ok().and_then(|m| m.get(&before).cloned()) { + return Ok((known, true)); + } } let digest = crate::module::hash::file_hex(&file)?; // Hashed bytes are only attributable to this identity if the file did // not change underneath the read. - if eligible - && identity(&file).is_ok_and(|after| after == before) - && let Ok(mut memory) = memory().lock() - { - memory.insert(before, digest.clone()); + if eligible && identity(&file).is_ok_and(|after| after == before) { + if let Ok(mut memory) = memory().lock() { + memory.insert(before, digest.clone()); + } } Ok((digest, false)) } diff --git a/crates/tinybus/src/module/transport.rs b/crates/tinybus/src/module/transport.rs index 519bdb1..d8b7d3a 100644 --- a/crates/tinybus/src/module/transport.rs +++ b/crates/tinybus/src/module/transport.rs @@ -36,6 +36,7 @@ struct HostContext { init_started: AtomicBool, init_notify: Notify, ready: AtomicBool, + broker_routing: AtomicBool, ready_notify: Notify, inflight: AtomicUsize, } @@ -95,6 +96,10 @@ impl DeferredInitializer { } impl ModuleTransport { + pub(crate) fn admit_broker(&self) { + self.context.broker_routing.store(true, Ordering::Release); + } + pub(crate) fn new(label: String, config: Vec) -> (Arc, TbHostVtable) { let (inbound_tx, inbound_rx) = mpsc::channel(HOST_QUEUE_CAPACITY); let context = Box::leak(Box::new(HostContext { @@ -106,6 +111,7 @@ impl ModuleTransport { init_started: AtomicBool::new(false), init_notify: Notify::new(), ready: AtomicBool::new(false), + broker_routing: AtomicBool::new(false), ready_notify: Notify::new(), inflight: AtomicUsize::new(0), })); @@ -138,6 +144,7 @@ impl ModuleTransport { fault: host_fault, config: config_slice, ready: host_ready, + broker_routing: Some(host_broker_routing), }; (transport, vtable) } @@ -680,6 +687,13 @@ unsafe extern "C" fn host_fault(ctx: *mut c_void, _: *const u8, _: usize) { })); } +unsafe extern "C" fn host_broker_routing(ctx: *mut c_void) -> u32 { + // HostContext is leaked for the lifetime of the vtable; no module-owned + // JSON or transport label can grant this provenance. + unsafe { ctx.cast::().as_ref() } + .is_some_and(|context| context.broker_routing.load(Ordering::Acquire)) as u32 +} + unsafe extern "C" fn host_ready(ctx: *mut c_void) { let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { if let Some(context) = unsafe { ctx.cast::().as_ref() } { diff --git a/crates/tinybus/src/module/transport_tests.rs b/crates/tinybus/src/module/transport_tests.rs index e189afb..8ef7431 100644 --- a/crates/tinybus/src/module/transport_tests.rs +++ b/crates/tinybus/src/module/transport_tests.rs @@ -485,3 +485,42 @@ async fn delivery_maps_a_faulted_transport_a_missing_module_and_module_refusals( Err(Error::ConnectionClosed) )); } + +#[test] +fn native_provenance_requires_actual_broker_admission() { + let (transport, host) = ModuleTransport::new("context".into(), b"{}".to_vec()); + let callback = host + .broker_routing + .expect("additive broker provenance callback"); + assert_eq!(unsafe { callback(host.host_ctx) }, 0); + assert_eq!(unsafe { callback(std::ptr::null_mut()) }, 0); + let runtime = tokio::runtime::Runtime::new().unwrap(); + let _guard = runtime.enter(); + crate::broker::Broker::new().attach(transport); + assert_eq!(unsafe { callback(host.host_ctx) }, 1); +} + +unsafe extern "C" fn legacy_host_prefix_init( + host: *const TbHostVtable, + out: *mut TbModuleVtable, +) -> i32 { + let prefix = unsafe { + host.cast::() + .read() + }; + assert!(prefix.size >= crate::module::abi::TB_HOST_VTABLE_BASE_SIZE); + assert_eq!(prefix.config.len, 2); + // An old module only copies these bytes and never interprets the tail. + unsafe { initialize_ok(host, out) } +} + +#[tokio::test] +async fn a_new_host_keeps_the_frozen_prefix_readable_by_old_modules() { + let (transport, host) = ModuleTransport::new("legacy".into(), b"{}".to_vec()); + let mut module = TbModuleVtable::default(); + assert_eq!( + unsafe { legacy_host_prefix_init(&host, &mut module) }, + TB_OK + ); + transport.initialize(module).unwrap(); +} diff --git a/crates/tinybus/src/ports/transport.rs b/crates/tinybus/src/ports/transport.rs index b610d7c..6336c29 100644 --- a/crates/tinybus/src/ports/transport.rs +++ b/crates/tinybus/src/ports/transport.rs @@ -21,8 +21,11 @@ use crate::error::Result; use crate::message::Message; /// One peer's bidirectional link, framed at the message level. +/// +/// `Any` supplies intrinsic concrete identity for private transport provenance; +/// it does not offer custom transports a hook to claim broker authentication. #[async_trait] -pub trait Transport: Send + Sync + 'static { +pub trait Transport: std::any::Any + Send + Sync + 'static { /// Write one message. Must be safe to call from many tasks at once. async fn send(&self, message: Message) -> Result<()>; diff --git a/crates/tinybus/src/router.rs b/crates/tinybus/src/router.rs index fe67bb5..afebd13 100644 --- a/crates/tinybus/src/router.rs +++ b/crates/tinybus/src/router.rs @@ -93,28 +93,34 @@ impl MatchRule { /// Whether `message` satisfies every set field. pub fn matches(&self, message: &Message) -> bool { let h = &message.header; - if let Some(kind) = self.kind - && kind != h.kind - { + if self.kind.is_some_and(|kind| kind != h.kind) { return false; } - if let Some(sender) = &self.sender - && h.sender.as_ref() != Some(sender) + if self + .sender + .as_ref() + .is_some_and(|sender| h.sender.as_ref() != Some(sender)) { return false; } - if let Some(interface) = &self.interface - && h.interface.as_ref() != Some(interface) + if self + .interface + .as_ref() + .is_some_and(|interface| h.interface.as_ref() != Some(interface)) { return false; } - if let Some(member) = &self.member - && h.member.as_ref() != Some(member) + if self + .member + .as_ref() + .is_some_and(|member| h.member.as_ref() != Some(member)) { return false; } - if let Some(path) = &self.path - && h.path.as_ref() != Some(path) + if self + .path + .as_ref() + .is_some_and(|path| h.path.as_ref() != Some(path)) { return false; } @@ -184,6 +190,9 @@ struct Peer { /// absence of an entry is what refuses a confidential delivery, so an /// ordinary out-of-process peer is ineligible by construction. attestations: HashMap, + /// The artifact admitted for this exact connection. Its original manifest + /// identity survives alias release; unrelated aliases never inherit it. + admitted_artifact: Option, } /// Who is attached, what they are called, and what they want to hear. @@ -221,6 +230,7 @@ impl Router { matches: Vec::new(), manifest: None, attestations: HashMap::new(), + admitted_artifact: None, }, ); self.names.insert(unique.clone(), id); @@ -392,7 +402,10 @@ impl Router { self.peers.get(id).map(|p| p.unique.clone()) } - /// Record what the host verified about the peer owning `name`. + /// Record what the host verified for an exact well-known name. + /// + /// This does not admit an artifact for the peer's unique connection or + /// replace an artifact explicitly admitted by the module host. /// /// Gated with module loading, because that is the only thing that can /// produce an attestation. Without it nothing is ever attested and every @@ -412,8 +425,8 @@ impl Router { } } - /// [`Router::set_attestation`] addressed by the peer's unique name, for the - /// module host, which holds that rather than the internal peer id. + /// Record the module host's verified artifact on its exact unique peer, + /// retaining its manifest name and digest for fixed-peer delivery. #[cfg(feature = "modules")] pub(crate) fn set_attestation_for_unique( &mut self, @@ -421,6 +434,9 @@ impl Router { attestation: Attestation, ) { if let Some(id) = self.names.get(unique).copied() { + if let Some(peer) = self.peers.get_mut(&id) { + peer.admitted_artifact = Some(attestation.clone()); + } self.set_attestation(id, attestation); } } @@ -428,11 +444,17 @@ impl Router { /// What the broker verified about whoever owns `name`, if anything. pub fn attestation_of(&self, name: &BusName) -> Option { let id = self.names.get(name)?; - self.peers.get(id)?.attestations.get(name).cloned() + let peer = self.peers.get(id)?; + if name.is_unique() { + peer.admitted_artifact.clone() + } else { + peer.attestations.get(name).cloned() + } } /// The outbox of whoever owns `destination`, but only if the host has - /// verified that peer's artifact *for that name*. + /// verified that peer's artifact for that well-known identity, or admitted + /// the artifact on that exact unique connection. /// /// The lookup and the check are one operation on purpose. Resolving first /// and checking after would leave a window in which a caller could hold a @@ -447,7 +469,12 @@ impl Router { .peers .get(id) .ok_or_else(|| Error::NameHasNoOwner(destination.clone()))?; - if !peer.attestations.contains_key(destination) { + let attested = if destination.is_unique() { + peer.admitted_artifact.is_some() + } else { + peer.attestations.contains_key(destination) + }; + if !attested { return Err(Error::not_attested( destination.clone(), "only a loaded module with a verified artifact may receive a secret", diff --git a/crates/tinybus/src/router_tests.rs b/crates/tinybus/src/router_tests.rs index 6b64366..2420fae 100644 --- a/crates/tinybus/src/router_tests.rs +++ b/crates/tinybus/src/router_tests.rs @@ -241,3 +241,92 @@ fn a_dead_peers_attestation_does_not_survive_it() { assert_eq!(router.attestation_of(&name), None); assert!(router.resolve_attested(&name).is_err()); } + +#[cfg(feature = "modules")] +#[test] +fn a_name_attestation_does_not_admit_the_unique_connection() { + let mut router = Router::default(); + let (id, unique) = router.attach(outbox()); + let wallet = BusName::new("ai.tinyhumans.openhuman.Wallet").unwrap(); + let voice = BusName::new("ai.tinyhumans.openhuman.Voice").unwrap(); + router.request_name(id, wallet.clone()).unwrap(); + router.request_name(id, voice.clone()).unwrap(); + let verified = attestation(wallet.as_str()); + router.set_attestation(id, verified.clone()); + + assert_eq!(router.attestation_of(&wallet), Some(verified)); + assert!(router.resolve_attested(&wallet).is_ok()); + assert_eq!(router.attestation_of(&voice), None); + assert_eq!( + router.resolve_attested(&voice).unwrap_err().wire_name(), + Error::NOT_ATTESTED + ); + assert_eq!(router.attestation_of(&unique), None); + assert_eq!( + router.resolve_attested(&unique).unwrap_err().wire_name(), + Error::NOT_ATTESTED + ); +} + +#[cfg(feature = "modules")] +#[test] +fn a_name_attestation_does_not_replace_an_explicitly_admitted_artifact() { + let mut router = Router::default(); + let (id, unique) = router.attach(outbox()); + let wallet = BusName::new("ai.tinyhumans.openhuman.Wallet").unwrap(); + let voice = BusName::new("ai.tinyhumans.openhuman.Voice").unwrap(); + router.request_name(id, wallet.clone()).unwrap(); + router.request_name(id, voice.clone()).unwrap(); + let admitted = attestation(wallet.as_str()); + router.set_attestation_for_unique(&unique, admitted.clone()); + let mut named = attestation(voice.as_str()); + named.sha256 = "b".repeat(64); + router.set_attestation(id, named.clone()); + + assert_eq!(router.attestation_of(&unique), Some(admitted.clone())); + assert!(router.resolve_attested(&unique).is_ok()); + assert_eq!(router.attestation_of(&wallet), Some(admitted)); + assert!(router.resolve_attested(&wallet).is_ok()); + assert_eq!(router.attestation_of(&voice), Some(named)); + assert!(router.resolve_attested(&voice).is_ok()); +} + +#[cfg(feature = "modules")] +#[test] +fn a_fixed_unique_peer_retains_its_admitted_artifact_across_alias_handover_until_detach() { + let mut router = Router::default(); + let (tx, mut rx) = tokio::sync::mpsc::channel(4); + let (id, unique) = router.attach(tx); + let name = BusName::new("ai.tinyhumans.openhuman.Wallet").unwrap(); + let alias = BusName::new("ai.tinyhumans.openhuman.Extra").unwrap(); + router.request_name(id, name.clone()).unwrap(); + let admitted = attestation(name.as_str()); + router.set_attestation_for_unique(&unique, admitted.clone()); + assert_eq!(router.attestation_of(&unique), Some(admitted.clone())); + let target = router.resolve_attested(&unique).unwrap(); + target + .try_send(signal("ai.tinyhumans.Test", "Tick", "/")) + .unwrap(); + assert_eq!( + rx.try_recv().unwrap().header.member.unwrap().as_str(), + "Tick" + ); + router.request_name(id, alias.clone()).unwrap(); + assert_eq!(router.attestation_of(&alias), None); + assert!(router.resolve_attested(&alias).is_err()); + router.release_name(id, &name).unwrap(); + let (successor, successor_unique) = router.attach(outbox()); + router.request_name(successor, name.clone()).unwrap(); + assert_eq!(router.attestation_of(&name), None); + assert_eq!(router.attestation_of(&successor_unique), None); + assert!(router.resolve_attested(&name).is_err()); + assert!(router.resolve_attested(&successor_unique).is_err()); + assert_eq!(router.attestation_of(&unique), Some(admitted)); + assert!(router.resolve_attested(&unique).is_ok()); + router.detach(id); + assert_eq!(router.attestation_of(&unique), None); + assert!(matches!( + router.resolve_attested(&unique), + Err(Error::NameHasNoOwner(_)) + )); +} diff --git a/crates/tinybus/src/service/mod.rs b/crates/tinybus/src/service/mod.rs index 88202e5..07ade4c 100644 --- a/crates/tinybus/src/service/mod.rs +++ b/crates/tinybus/src/service/mod.rs @@ -25,15 +25,53 @@ use async_trait::async_trait; use serde_json::Value; use crate::error::Result; -use crate::name::{InterfaceName, MemberName}; +use crate::name::{BusName, InterfaceName, MemberName}; pub use crate::service::tree::ObjectTree; +/// Transport-authenticated facts about one incoming method call. +/// +/// A sender identifies a connection in this broker session, never a human or +/// an artifact. Bind it to trusted initialization separately. This type has no +/// wire representation and no public constructor for authenticated identity. +/// Direct dispatch and transports without proven broker routing are unverified. +/// Memory deliveries carry a private broker proof; custom transports and Unix +/// sockets remain unverified even after Hello. Native SDK deliveries require a +/// trusted host's additive C routing callback. Confidential recipient attestation +/// is separate from incoming sender authentication. Invalidate any binding on +/// disconnect or broker replacement; names are unique only within one session. +/// +/// Authority cannot be deserialized from invocation JSON: +/// ```compile_fail +/// let _: tinybus::CallContext = serde_json::from_str("{}").unwrap(); +/// ``` +#[derive(Debug, Clone, Default)] +pub struct CallContext { + sender: Option, +} + +impl CallContext { + /// The broker-stamped unique sender, only on a proven broker delivery path. + /// + /// Returns `None` for default contexts, unverified deliveries, and missing + /// or well-known sender names. A returned name identifies a peer within + /// this broker session; it does not grant permission for an operation. + pub fn authenticated_sender(&self) -> Option<&BusName> { + self.sender.as_ref() + } + + pub(crate) fn brokered(sender: Option<&BusName>) -> Self { + Self { + sender: sender.filter(|name| name.is_unique()).cloned(), + } + } +} + /// One contract a service implements. /// -/// Implement this by hand only for dynamic interfaces — a bridge that proxies -/// an interface it does not know at compile time. Everything else should use -/// `#[tinybus::interface]`. +/// Use `#[tinybus::interface]` for ordinary typed methods. Implement by hand for +/// dynamic bridges or to authorize incoming calls with `call_with_context`; +/// context never becomes an extra positional argument. #[async_trait] pub trait Interface: Send + Sync + 'static { /// The interface's name, e.g. `ai.tinyhumans.openhuman.Voice`. @@ -64,6 +102,21 @@ pub trait Interface: Send + Sync + 'static { /// because a service that has panicked has an unknown internal state and /// answering the next call from it is worse than being restarted. async fn call(&self, member: &MemberName, args: Value) -> Result; + + /// Run a call with transport-authenticated incoming facts. + /// + /// The default delegates to [`Interface::call`] without inspecting context, + /// preserving legacy implementations and their positional arguments. + /// Override this method to authorize callers; an unverified + /// context must fail closed when caller identity is required. + async fn call_with_context( + &self, + member: &MemberName, + args: Value, + _context: &CallContext, + ) -> Result { + self.call(member, args).await + } } #[async_trait] @@ -83,4 +136,17 @@ impl Interface for std::sync::Arc { async fn call(&self, member: &MemberName, args: Value) -> Result { (**self).call(member, args).await } + + async fn call_with_context( + &self, + member: &MemberName, + args: Value, + context: &CallContext, + ) -> Result { + (**self).call_with_context(member, args, context).await + } } + +#[cfg(test)] +#[path = "mod_tests.rs"] +mod tests; diff --git a/crates/tinybus/src/service/mod_tests.rs b/crates/tinybus/src/service/mod_tests.rs new file mode 100644 index 0000000..3288105 --- /dev/null +++ b/crates/tinybus/src/service/mod_tests.rs @@ -0,0 +1,18 @@ +use super::*; + +#[test] +fn only_a_brokered_unique_sender_is_authenticated() { + assert!(CallContext::default().authenticated_sender().is_none()); + assert!(CallContext::brokered(None).authenticated_sender().is_none()); + let well_known = BusName::new("org.example.Host").unwrap(); + assert!( + CallContext::brokered(Some(&well_known)) + .authenticated_sender() + .is_none() + ); + let unique = BusName::new(":1.42").unwrap(); + assert_eq!( + CallContext::brokered(Some(&unique)).authenticated_sender(), + Some(&unique) + ); +} diff --git a/crates/tinybus/src/service/tree.rs b/crates/tinybus/src/service/tree.rs index dd7178d..4bd8bb6 100644 --- a/crates/tinybus/src/service/tree.rs +++ b/crates/tinybus/src/service/tree.rs @@ -14,7 +14,7 @@ use serde_json::Value; use crate::error::{Error, Result}; use crate::name::{InterfaceName, MemberName, ObjectPath}; -use crate::service::Interface; +use crate::service::{CallContext, Interface}; /// Everything one connection exports. #[derive(Default)] @@ -82,7 +82,7 @@ impl ObjectTree { }) } - /// Resolve and invoke in one step. + /// Resolve and invoke an ordinary call with an unverified default context. pub async fn dispatch( &self, path: &ObjectPath, @@ -96,6 +96,9 @@ impl ObjectTree { /// Resolve and invoke one call while enforcing the delivery's /// confidentiality flag before argument decoding reaches user code. + /// + /// Uses an unverified default context even when `confidential` is true; + /// the flag does not authenticate a caller. pub async fn dispatch_with_confidential( &self, path: &ObjectPath, @@ -103,8 +106,45 @@ impl ObjectTree { member: &MemberName, args: Value, confidential: bool, + ) -> Result { + self.dispatch_with_context( + path, + interface, + member, + args, + confidential, + &CallContext::default(), + ) + .await + } + + /// Resolve and invoke with out-of-band incoming context and delivery flags. + /// + /// Checks the member and its confidentiality requirement before calling + /// [`Interface::call_with_context`]. Passes `context` through unchanged; + /// this method does not derive caller identity from `args` or the flag. + pub async fn dispatch_with_context( + &self, + path: &ObjectPath, + interface: &InterfaceName, + member: &MemberName, + args: Value, + confidential: bool, + context: &CallContext, ) -> Result { let target = self.lookup(path, interface)?; + Self::invoke(target, interface, member, args, confidential, context).await + } + + /// Invoke a resolved snapshot without holding the connection's tree lock. + pub(crate) async fn invoke( + target: Arc, + interface: &InterfaceName, + member: &MemberName, + args: Value, + confidential: bool, + context: &CallContext, + ) -> Result { if !target.members().contains(member) { return Err(Error::UnknownMethod { interface: interface.clone(), @@ -117,7 +157,7 @@ impl ObjectTree { member: member.clone(), }); } - target.call(member, args).await + target.call_with_context(member, args, context).await } } diff --git a/crates/tinybus/src/service/tree_tests.rs b/crates/tinybus/src/service/tree_tests.rs index 14df0df..8a9733b 100644 --- a/crates/tinybus/src/service/tree_tests.rs +++ b/crates/tinybus/src/service/tree_tests.rs @@ -192,3 +192,32 @@ fn one_object_can_carry_several_contracts() { assert!(tree.remove(&path())); assert!(!tree.remove(&path())); } + +#[tokio::test] +async fn direct_context_dispatch_defaults_to_unverified_and_keeps_legacy_arguments() { + let mut tree = ObjectTree::new(); + tree.insert( + path(), + Arc::new(Echo { + name: "ai.tinyhumans.Voice", + tag: "legacy", + confidential: false, + }), + ); + let member = MemberName::new("Echo").unwrap(); + let context = CallContext::default(); + assert_eq!(context.authenticated_sender(), None); + assert_eq!( + tree.dispatch_with_context( + &path(), + &iface("ai.tinyhumans.Voice"), + &member, + serde_json::json!([1, 2]), + false, + &context + ) + .await + .unwrap(), + "legacy" + ); +} diff --git a/crates/tinybus/src/stream/base64.rs b/crates/tinybus/src/stream/base64.rs index 6879f6b..9c17046 100644 --- a/crates/tinybus/src/stream/base64.rs +++ b/crates/tinybus/src/stream/base64.rs @@ -42,7 +42,7 @@ pub fn encode(bytes: &[u8]) -> String { /// definition, and this message travels back to the peer as an error reply. pub fn decode(text: &str) -> Result> { let bytes = text.as_bytes(); - if !bytes.len().is_multiple_of(4) { + if bytes.len() % 4 != 0 { return Err(Error::protocol( "base64 chunk length is not a multiple of four", )); diff --git a/crates/tinybus/src/transport/memory.rs b/crates/tinybus/src/transport/memory.rs index 6274e99..32e8818 100644 --- a/crates/tinybus/src/transport/memory.rs +++ b/crates/tinybus/src/transport/memory.rs @@ -33,16 +33,25 @@ use crate::ports::{Listener, Transport}; /// backpressure rather than a longer queue. pub const CHANNEL_CAPACITY: usize = 256; +struct Delivery { + message: Message, + brokered: bool, +} + /// One end of an in-process link. /// +/// Broker-written deliveries carry private per-frame provenance. Public +/// [`Transport::send`] cannot set it, and public [`Transport::recv`] discards it; +/// replaying a sender header or wrapping this transport grants no call authority. +/// /// The outbound sender lives in an `Option` behind a *std* mutex so that /// [`Transport::close`] can drop it: dropping the sender is what makes the far /// end's `recv` return `None`, and without that a closed in-process link would /// look, to the peer, exactly like an idle one. The lock is a std mutex rather /// than a tokio one precisely so that it is never held across the send await. pub struct MemoryTransport { - outbound: std::sync::Mutex>>, - inbound: Mutex>, + outbound: std::sync::Mutex>>, + inbound: Mutex>, label: String, } @@ -54,7 +63,7 @@ impl MemoryTransport { (Self::new(a_tx, b_rx), Self::new(b_tx, a_rx)) } - fn new(outbound: mpsc::Sender, inbound: mpsc::Receiver) -> Self { + fn new(outbound: mpsc::Sender, inbound: mpsc::Receiver) -> Self { Self { outbound: std::sync::Mutex::new(Some(outbound)), inbound: Mutex::new(inbound), @@ -62,8 +71,30 @@ impl MemoryTransport { } } + // Only the actual broker writer can mint this out-of-band provenance. + pub(crate) async fn send_brokered(&self, message: Message) -> Result<()> { + self.send_delivery(message, true).await + } + + async fn send_delivery(&self, message: Message, brokered: bool) -> Result<()> { + self.sender()? + .send(Delivery { message, brokered }) + .await + .map_err(|_| Error::transport("the peer end of the in-process link was dropped")) + } + + pub(crate) async fn recv_with_provenance(&self) -> Result> { + Ok(self + .inbound + .lock() + .await + .recv() + .await + .map(|delivery| (delivery.message, delivery.brokered))) + } + /// Take a clone of the sender without holding the lock across an await. - fn sender(&self) -> Result> { + fn sender(&self) -> Result> { self.outbound .lock() .expect("the outbound lock is never held across a panic point") @@ -75,17 +106,16 @@ impl MemoryTransport { #[async_trait] impl Transport for MemoryTransport { async fn send(&self, message: Message) -> Result<()> { - self.sender()? - .send(message) - .await - .map_err(|_| Error::transport("the peer end of the in-process link was dropped")) + self.send_delivery(message, false).await } async fn recv(&self) -> Result> { - // Single-reader contract (see the port docs) makes holding this lock - // across the await safe: there is never a second task to block. - let mut inbound = self.inbound.lock().await; - Ok(inbound.recv().await) + // Public receive discards provenance; wrapping/replaying a frame cannot + // carry the private broker delivery proof into a custom transport. + Ok(self + .recv_with_provenance() + .await? + .map(|(message, _)| message)) } async fn close(&self) -> Result<()> { diff --git a/crates/tinybus/src/transport/mod.rs b/crates/tinybus/src/transport/mod.rs index 35ec149..d185546 100644 --- a/crates/tinybus/src/transport/mod.rs +++ b/crates/tinybus/src/transport/mod.rs @@ -9,3 +9,17 @@ pub mod memory; #[cfg(all(feature = "uds", unix))] pub mod unix; + +/// Intrinsic downcast: an overridable `as_any` hook could lie about the concrete +/// receiver while injecting forged frames from another transport. This uses +/// `Any`'s blanket implementation and works before trait upcasting stabilized. +pub(crate) fn downcast( + transport: &dyn crate::ports::Transport, +) -> Option<&T> { + if std::any::Any::type_id(transport) != std::any::TypeId::of::() { + return None; + } + // SAFETY: intrinsic TypeId above proves that the trait object's data pointer + // is a T. The borrow retains exactly the original lifetime. + Some(unsafe { &*(std::ptr::from_ref(transport).cast::()) }) +} diff --git a/docs/modules/attest/README.md b/docs/modules/attest/README.md index a16c1cc..5d36a3a 100644 --- a/docs/modules/attest/README.md +++ b/docs/modules/attest/README.md @@ -26,6 +26,17 @@ digest, and that the operator listed that digest as acceptable. The hash is computed by the host over bytes the host read itself. Nothing a module claims about itself participates. +An admitted module can receive confidential calls at its original well-known +name or its fixed unique peer. The unique lookup returns the same attestation, +including the original manifest name and verified digest. Extra aliases remain +unattested. Releasing the manifest alias does not retarget a pinned unique peer; +a successor inherits no attestation, and detach removes the old peer's record. +Terminal module states refuse unique calls even before transport detach. + +Recipient attestation does not authenticate the caller or grant human authority. +Incoming caller identity still requires broker-proven `CallContext` and a policy +that binds the trusted peer to its permitted operations. + ```toml # modules.toml, beside the artifact — the same file the loader already uses "libwallet.so" = "41edece42d63e8d9bf515a9ba6932e1c20cbc9f5a5d134645adb5db1b9737ea3" @@ -148,8 +159,9 @@ problems with different fixes. artifact *as the wallet*, not as everything that module also answers to. - A confidential **signal** is refused on ingress. A broadcast has no single recipient to attest, so there is nothing the flag could mean. -- A confidential **call** must address a well-known name. A unique name - identifies a connection, not an artifact. +- A confidential **call** must address the attested well-known name or the + unique peer on which the host admitted that artifact. Ordinary unique peers + remain ineligible. - A confidential **reply** inherits the flag and goes back to the caller. A key derivation answers with a key, and a reply that quietly lost the flag would leak on the way back what the call protected on the way out. diff --git a/docs/protocol.md b/docs/protocol.md index 8ce89e1..1645df3 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -77,8 +77,8 @@ A broker that sees it **must**: - refuse a `signal` carrying it, and refuse any message carrying it without a `destination`; -- refuse a `method_call` carrying it unless the destination is a well-known name - owned by a recipient the host has attested, replying +- refuse a `method_call` carrying it unless the destination is the exact + attested well-known name or fixed unique peer of an admitted artifact, replying `ai.tinyhumans.tinybus.Error.NotAttested`; - never deliver the message to a match-rule subscriber, and never log its body.