diff --git a/crates/solstone-linux/src/cli.rs b/crates/solstone-linux/src/cli.rs index d7fc667..5f8a508 100644 --- a/crates/solstone-linux/src/cli.rs +++ b/crates/solstone-linux/src/cli.rs @@ -1155,6 +1155,44 @@ mod tests { ); assert_eq!(body["label"], "host-a"); assert_eq!(body["stream_type"], "desktop"); + + for (cli_token, env_token, expected) in [ + (Some("cli-token"), None, "cli-token"), + (None, Some("env-token"), "env-token"), + (Some("cli-wins"), Some("env-loses"), "cli-wins"), + ] { + let token_temp = tempfile::tempdir().unwrap(); + let token_paths = paths(&token_temp); + let mut token_output = Vec::new(); + let mut token_errors = Vec::new(); + assert_eq!( + cmd_setup( + SetupOptions { + server_url: Some(server.url.clone()), + token: cli_token.map(str::to_owned), + stream_name: Some("host-a".into()), + non_interactive: true, + }, + token_paths.clone(), + env_token.map(str::to_owned), + &mut RealRegistrar, + &mut token_output, + &mut token_errors, + ), + 0 + ); + assert_eq!(load_config(token_paths).config.key, expected); + assert!( + String::from_utf8(token_output) + .unwrap() + .contains("Using provided token; skipping registration.") + ); + assert_eq!( + server.requests().len(), + 1, + "token setup must skip registration" + ); + } assert!(peer.requests().is_empty()); runtime.block_on(peer.shutdown()); } diff --git a/crates/solstone-linux/src/lib.rs b/crates/solstone-linux/src/lib.rs index 39a2716..0071e59 100644 --- a/crates/solstone-linux/src/lib.rs +++ b/crates/solstone-linux/src/lib.rs @@ -18,9 +18,7 @@ pub mod matching; pub mod observer; pub mod pipeline; pub mod positions; -#[allow(dead_code)] mod private_file; -#[allow(dead_code)] mod private_link; pub mod recovery; pub mod restore_token; @@ -43,7 +41,6 @@ pub mod video; #[cfg(test)] mod observer_contract_tests; #[cfg(test)] -#[allow(dead_code)] mod private_link_test_peer; #[cfg(test)] mod release_rail_tests; diff --git a/crates/solstone-linux/src/private_file.rs b/crates/solstone-linux/src/private_file.rs index 926be3b..155b3c7 100644 --- a/crates/solstone-linux/src/private_file.rs +++ b/crates/solstone-linux/src/private_file.rs @@ -5,8 +5,8 @@ use std::{ fmt, fs::{self, File}, io::{self, Write}, - os::unix::fs::PermissionsExt, - path::{Path, PathBuf}, + os::fd::AsFd, + path::Path, sync::atomic::{AtomicU64, Ordering}, }; @@ -102,27 +102,48 @@ pub(crate) fn ensure_private_directory(path: &Path) -> Result<(), PrivateFileErr } fn set_and_verify_mode(path: &Path, mode: u32, directory: bool) -> Result<(), PrivateFileError> { - fs::set_permissions(path, fs::Permissions::from_mode(mode)) - .map_err(|error| PrivateFileError::io("target", "chmod", error))?; - let metadata = fs::symlink_metadata(path) - .map_err(|error| PrivateFileError::io("target", "inspect", error))?; + let flags = rustix::fs::OFlags::RDONLY + | rustix::fs::OFlags::CLOEXEC + | rustix::fs::OFlags::NOFOLLOW + | if directory { + rustix::fs::OFlags::DIRECTORY + } else { + rustix::fs::OFlags::empty() + }; + let descriptor = rustix::fs::openat(rustix::fs::CWD, path, flags, rustix::fs::Mode::empty()) + .map_err(|error| { + if error == rustix::io::Errno::LOOP || error == rustix::io::Errno::NOTDIR { + PrivateFileError::InvalidTarget("target") + } else { + PrivateFileError::io("target", "open", error.into()) + } + })?; + let expected = rustix::fs::Mode::from_raw_mode(mode); + rustix::fs::fchmod(&descriptor, expected) + .map_err(|error| PrivateFileError::io("target", "chmod", error.into()))?; + let stat = rustix::fs::fstat(&descriptor) + .map_err(|error| PrivateFileError::io("target", "inspect", error.into()))?; let valid_kind = if directory { - metadata.is_dir() + rustix::fs::FileType::from_raw_mode(stat.st_mode) == rustix::fs::FileType::Directory } else { - metadata.is_file() + rustix::fs::FileType::from_raw_mode(stat.st_mode) == rustix::fs::FileType::RegularFile }; - if metadata.file_type().is_symlink() - || !valid_kind - || metadata.permissions().mode() & 0o777 != mode - { + if !valid_kind || rustix::fs::Mode::from_raw_mode(stat.st_mode) != expected { return Err(PrivateFileError::InvalidTarget("target")); } Ok(()) } pub(crate) fn open_regular_readonly(path: &Path) -> Result { + open_regular_readonly_at(rustix::fs::CWD, path) +} + +fn open_regular_readonly_at( + directory: Fd, + path: &Path, +) -> Result { let descriptor = rustix::fs::openat( - rustix::fs::CWD, + directory, path, rustix::fs::OFlags::RDONLY | rustix::fs::OFlags::CLOEXEC | rustix::fs::OFlags::NOFOLLOW, rustix::fs::Mode::empty(), @@ -156,47 +177,72 @@ pub(crate) fn atomic_write_bytes_with_fault( bytes: &[u8], fault: &dyn DurableWriteFault, ) -> Result<(), PrivateFileError> { - let previous = match fs::symlink_metadata(path) { - Ok(metadata) if metadata.file_type().is_symlink() || !metadata.is_file() => { - return Err(PrivateFileError::InvalidTarget("file")); - } - Ok(_) => Some(fs::read(path).map_err(|error| PrivateFileError::io("file", "read", error))?), - Err(error) if error.kind() == io::ErrorKind::NotFound => None, - Err(error) => return Err(PrivateFileError::io("file", "inspect", error)), - }; let parent = path .parent() .ok_or(PrivateFileError::InvalidTarget("file"))?; let name = path .file_name() - .ok_or(PrivateFileError::InvalidTarget("file"))? - .to_string_lossy(); - let temporary = parent.join(format!( + .ok_or(PrivateFileError::InvalidTarget("file"))?; + let parent_descriptor = open_directory(parent)?; + match open_regular_readonly_at(&parent_descriptor, Path::new(name)) { + Ok(file) => drop(file), + Err(PrivateFileError::Io { + kind: io::ErrorKind::NotFound, + .. + }) => {} + Err(error) => return Err(error), + } + let temporary = format!( ".{name}.{}.{}.tmp", std::process::id(), - TEMP_COUNTER.fetch_add(1, Ordering::Relaxed) - )); - let result = write_temporary(path, &temporary, parent, bytes, fault); + TEMP_COUNTER.fetch_add(1, Ordering::Relaxed), + name = name.to_string_lossy() + ); + let result = write_temporary(&parent_descriptor, name, &temporary, bytes, fault); if result.is_err() { - let _ = fs::remove_file(&temporary); - if let Some(previous) = previous { - let _ = fs::write(path, previous); - } + let _ = rustix::fs::unlinkat( + &parent_descriptor, + temporary.as_str(), + rustix::fs::AtFlags::empty(), + ); + // Before rename the target is untouched; after rename it contains the new + // complete value. Rewriting either value after an error cannot be made safe. } result } +fn open_directory(path: &Path) -> Result { + rustix::fs::openat( + rustix::fs::CWD, + path, + rustix::fs::OFlags::RDONLY + | rustix::fs::OFlags::CLOEXEC + | rustix::fs::OFlags::NOFOLLOW + | rustix::fs::OFlags::DIRECTORY, + rustix::fs::Mode::empty(), + ) + .map(File::from) + .map_err(|error| { + if error == rustix::io::Errno::LOOP || error == rustix::io::Errno::NOTDIR { + PrivateFileError::InvalidTarget("directory") + } else { + PrivateFileError::io("directory", "open", error.into()) + } + }) +} + fn write_temporary( - path: &Path, - temporary: &PathBuf, - parent: &Path, + parent: &File, + name: &std::ffi::OsStr, + temporary: &str, bytes: &[u8], fault: &dyn DurableWriteFault, ) -> Result<(), PrivateFileError> { fault .before(DurableWriteStage::Create) .map_err(|error| PrivateFileError::io("file", "create", error))?; - let descriptor = rustix::fs::open( + let descriptor = rustix::fs::openat( + parent, temporary, rustix::fs::OFlags::CREATE | rustix::fs::OFlags::EXCL @@ -207,8 +253,8 @@ fn write_temporary( ) .map_err(|error| PrivateFileError::io("file", "create", error.into()))?; let mut file = File::from(descriptor); - file.set_permissions(fs::Permissions::from_mode(0o600)) - .map_err(|error| PrivateFileError::io("file", "chmod", error))?; + rustix::fs::fchmod(&file, rustix::fs::Mode::RUSR | rustix::fs::Mode::WUSR) + .map_err(|error| PrivateFileError::io("file", "chmod", error.into()))?; fault .before(DurableWriteStage::Write) .map_err(|error| PrivateFileError::io("file", "write", error))?; @@ -223,19 +269,20 @@ fn write_temporary( fault .before(DurableWriteStage::Rename) .map_err(|error| PrivateFileError::io("file", "rename", error))?; - fs::rename(temporary, path).map_err(|error| PrivateFileError::io("file", "rename", error))?; + rustix::fs::renameat(parent, temporary, parent, name) + .map_err(|error| PrivateFileError::io("file", "rename", error.into()))?; fault .before(DurableWriteStage::DirSync) .map_err(|error| PrivateFileError::io("directory", "fsync", error))?; - File::open(parent) - .and_then(|directory| directory.sync_all()) + parent + .sync_all() .map_err(|error| PrivateFileError::io("directory", "fsync", error)) } #[cfg(test)] mod tests { use super::*; - use std::os::unix::fs::{MetadataExt, symlink}; + use std::os::unix::fs::{MetadataExt, PermissionsExt, symlink}; struct FailAt(DurableWriteStage); impl DurableWriteFault for FailAt { @@ -303,13 +350,12 @@ mod tests { } #[test] - fn every_injected_stage_preserves_previous_complete_file() { + fn every_pre_rename_failure_preserves_previous_complete_file() { for stage in [ DurableWriteStage::Create, DurableWriteStage::Write, DurableWriteStage::Fsync, DurableWriteStage::Rename, - DurableWriteStage::DirSync, ] { let temp = tempfile::tempdir().unwrap(); let path = temp.path().join("state"); @@ -326,6 +372,51 @@ mod tests { } } + #[test] + fn directory_sync_failure_leaves_one_complete_json_value() { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("state"); + let previous = br#"{"value":"previous"}"#; + let next = br#"{"value":"next"}"#; + atomic_write_bytes(&path, previous).unwrap(); + assert!( + atomic_write_bytes_with_fault(&path, next, &FailAt(DurableWriteStage::DirSync)) + .is_err() + ); + let current = fs::read(&path).unwrap(); + assert!(current == previous || current == next); + assert!(serde_json::from_slice::(¤t).is_ok()); + assert!(fs::read_dir(temp.path()).unwrap().all(|entry| { + !entry + .unwrap() + .file_name() + .to_string_lossy() + .ends_with(".tmp") + })); + } + + #[test] + fn every_injected_error_cleans_up_temporary_files() { + for stage in [ + DurableWriteStage::Create, + DurableWriteStage::Write, + DurableWriteStage::Fsync, + DurableWriteStage::Rename, + DurableWriteStage::DirSync, + ] { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("state"); + assert!(atomic_write_bytes_with_fault(&path, b"complete", &FailAt(stage)).is_err()); + assert!(fs::read_dir(temp.path()).unwrap().all(|entry| { + !entry + .unwrap() + .file_name() + .to_string_lossy() + .ends_with(".tmp") + })); + } + } + #[test] fn failed_initial_write_never_leaves_partial_target_or_temporary() { let temp = tempfile::tempdir().unwrap(); diff --git a/crates/solstone-linux/src/private_link.rs b/crates/solstone-linux/src/private_link.rs index b3faf89..bdc3c39 100644 --- a/crates/solstone-linux/src/private_link.rs +++ b/crates/solstone-linux/src/private_link.rs @@ -6,7 +6,6 @@ use std::{ fs::{self, File}, future::Future, io::{self, Read}, - os::unix::fs::PermissionsExt, path::{Path, PathBuf}, pin::Pin, sync::{Arc, Mutex, RwLock}, @@ -26,8 +25,8 @@ use spl_transport::{ }; use crate::private_file::{ - DurableWriteFault, PrivateFileError, atomic_write_bytes, atomic_write_bytes_with_fault, - ensure_private_directory, open_regular_readonly, + DurableWriteFault, NoWriteFault, PrivateFileError, atomic_write_bytes, + atomic_write_bytes_with_fault, ensure_private_directory, open_regular_readonly, }; pub(crate) const CREDENTIALS_FILENAME: &str = "credentials.json"; @@ -68,7 +67,6 @@ pub(crate) enum PrivateIoOperation { Chmod, Lock, Read, - Serialize, Persist, } @@ -242,9 +240,22 @@ impl PrivateStateLock { operation: PrivateIoOperation::Canonicalize, source, })?; - let descriptor = rustix::fs::openat( + let root_descriptor = rustix::fs::openat( rustix::fs::CWD, - canonical_root.join(PRIVATE_STATE_LOCK_FILENAME), + &canonical_root, + rustix::fs::OFlags::RDONLY + | rustix::fs::OFlags::CLOEXEC + | rustix::fs::OFlags::NOFOLLOW + | rustix::fs::OFlags::DIRECTORY, + rustix::fs::Mode::empty(), + ) + .map_err(|source| PrivateStateError::Io { + operation: PrivateIoOperation::Open, + source: source.into(), + })?; + let descriptor = rustix::fs::openat( + &root_descriptor, + PRIVATE_STATE_LOCK_FILENAME, rustix::fs::OFlags::RDWR | rustix::fs::OFlags::CLOEXEC | rustix::fs::OFlags::NOFOLLOW @@ -268,11 +279,13 @@ impl PrivateStateLock { kind: PrivateTargetKind::Lock, }); } - file.set_permissions(fs::Permissions::from_mode(0o600)) - .map_err(|source| PrivateStateError::Io { + rustix::fs::fchmod(&file, rustix::fs::Mode::RUSR | rustix::fs::Mode::WUSR).map_err( + |source| PrivateStateError::Io { operation: PrivateIoOperation::Chmod, - source, - })?; + source: source.into(), + }, + )?; + verify_private_lock(&file)?; match rustix::fs::flock(&file, rustix::fs::FlockOperation::NonBlockingLockExclusive) { Ok(()) => {} Err(rustix::io::Errno::WOULDBLOCK) => return Err(PrivateStateError::LockContended), @@ -294,6 +307,22 @@ impl PrivateStateLock { } } +fn verify_private_lock(file: &File) -> Result<(), PrivateStateError> { + let stat = rustix::fs::fstat(file).map_err(|source| PrivateStateError::Io { + operation: PrivateIoOperation::Inspect, + source: source.into(), + })?; + let expected_mode = rustix::fs::Mode::RUSR | rustix::fs::Mode::WUSR; + if rustix::fs::FileType::from_raw_mode(stat.st_mode) != rustix::fs::FileType::RegularFile + || rustix::fs::Mode::from_raw_mode(stat.st_mode) != expected_mode + { + return Err(PrivateStateError::InvalidTarget { + kind: PrivateTargetKind::Lock, + }); + } + Ok(()) +} + pub(crate) fn read_pair_link(input: R) -> Result { let mut bytes = Vec::new(); input @@ -340,6 +369,8 @@ impl Pairer for SplPairer { } } +// The cutover lode will call this setup root. +#[allow(dead_code)] pub(crate) async fn setup( config_root: &Path, device_label: &str, @@ -348,7 +379,7 @@ pub(crate) async fn setup( setup_with_pairer(&SplPairer, config_root, device_label, input).await } -pub(crate) async fn setup_with_pairer( +async fn setup_with_pairer( pairer: &dyn Pairer, config_root: &Path, device_label: &str, @@ -385,6 +416,8 @@ fn read_private_file( Ok(Some(bytes)) } +// The cutover lode will load credentials through this root. +#[allow(dead_code)] pub(crate) fn load_credential(config_root: &Path) -> Result, PrivateStateError> { let Some(bytes) = read_private_file( &config_root.join(CREDENTIALS_FILENAME), @@ -443,6 +476,7 @@ pub(crate) fn load_observer( Ok(Some(observer)) } +#[cfg(test)] pub(crate) fn persist_observer( config_root: &Path, observer: &ObserverState, @@ -458,7 +492,7 @@ pub(crate) fn persist_observer( } fn contains_invalid_header_value(value: &str) -> bool { - value.bytes().any(|byte| matches!(byte, b'\r' | b'\n' | 0)) + reqwest::header::HeaderValue::from_bytes(value.as_bytes()).is_err() } fn persist_and_publish_observer( @@ -562,10 +596,13 @@ pub(crate) struct PrivateLinkSession { origin: Url, opener: Arc, handle: JournalBridgeHandle, - token_persistence: Option>, + token_persistence: Arc, + state_lock: PrivateStateLock, } impl PrivateLinkSession { + // The cutover lode will issue confined requests through this root. + #[allow(dead_code)] pub(crate) fn request( &self, method: Method, @@ -578,13 +615,11 @@ impl PrivateLinkSession { .timeout(LOOPBACK_REQUEST_TIMEOUT)) } + // The cutover lode will explicitly quiesce sessions through this root. + #[allow(dead_code)] pub(crate) async fn shutdown(self) -> Result<(), PrivateStateError> { let status = self.handle.shutdown_and_wait().await; - if self - .token_persistence - .as_ref() - .is_some_and(|state| state.failed()) - { + if self.token_persistence.failed() { return Err(PrivateStateError::TokenPersistenceFailed); } if status.listener_active || status.active_requests != 0 { @@ -592,6 +627,33 @@ impl PrivateLinkSession { } Ok(()) } + + fn publish_observer(&self, observer: &ObserverState) -> Result<(), PrivateStateError> { + persist_and_publish_observer( + self.state_lock.root(), + observer, + &self.opener, + &NoWriteFault, + ) + } + + #[cfg(test)] + fn publish_observer_with_fault( + &self, + observer: &ObserverState, + fault: &dyn DurableWriteFault, + ) -> Result<(), PrivateStateError> { + persist_and_publish_observer(self.state_lock.root(), observer, &self.opener, fault) + } +} + +// The cutover lode will publish completed registrations through this root. +#[allow(dead_code)] +pub(crate) fn publish_observer_registration( + session: &PrivateLinkSession, + observer: &ObserverState, +) -> Result<(), PrivateStateError> { + session.publish_observer(observer) } struct TokenPersistence { @@ -647,30 +709,48 @@ impl TokenPersistence { } } +// The cutover lode will start runtime sessions through this root. +#[allow(dead_code)] pub(crate) async fn start_private_link_session( + config_root: &Path, credential: Credential, expected_name: &str, ) -> Result { - start_private_link_session_inner(credential, expected_name, None, None).await + start_private_link_session_inner( + config_root, + credential, + expected_name, + Arc::new(NoWriteFault), + SessionTestCapture::default(), + ) + .await +} + +#[derive(Default)] +struct SessionTestCapture { + #[cfg(test)] + capability: Option>>>, } async fn start_private_link_session_inner( + config_root: &Path, credential: Credential, expected_name: &str, - persistence: Option<(PathBuf, Arc)>, - capability_capture: Option<&Mutex>>, + persistence_fault: Arc, + _test_capture: SessionTestCapture, ) -> Result { + let state_lock = PrivateStateLock::acquire(config_root)?; + let config_root = state_lock.root().to_path_buf(); + let credential_instance_id = credential.instance_id.clone(); let endpoint_hosts = credential .endpoints .iter() .map(|endpoint| endpoint.host.clone()) .collect(); - let (token_persistence, hook) = persistence.map_or((None, None), |(root, fault)| { - let (state, hook) = TokenPersistence::new(root, credential.clone(), fault); - (Some(state), Some(hook)) - }); - let transport = - TransportClient::new(credential, hook).map_err(|_| PrivateStateError::BridgeUnavailable)?; + let (token_persistence, hook) = + TokenPersistence::new(config_root.clone(), credential.clone(), persistence_fault); + let transport = TransportClient::new(credential, Some(hook)) + .map_err(|_| PrivateStateError::BridgeUnavailable)?; let opener = Arc::new(PrivateLinkOpener::new(transport, expected_name.to_owned())); let bridge_names = BridgeNames { capability_cookie_name: "solstone_linux_cap".to_owned(), @@ -712,7 +792,8 @@ async fn start_private_link_session_inner( let bootstrap_url = handle .bootstrap_url() .ok_or(PrivateStateError::BootstrapFailed)?; - if let Some(capture) = capability_capture { + #[cfg(test)] + if let Some(capture) = _test_capture.capability { let capability = Url::parse(&bootstrap_url) .ok() .and_then(|url| { @@ -742,12 +823,21 @@ async fn start_private_link_session_inner( handle.begin_shutdown(); return Err(PrivateStateError::BootstrapFailed); } + if let Some(observer) = load_observer( + &config_root, + &credential_instance_id, + expected_name, + &origin, + )? { + opener.set_registered(&observer)?; + } Ok(PrivateLinkSession { client, origin, opener, handle, token_persistence, + state_lock, }) } @@ -760,7 +850,7 @@ mod tests { use std::{ io::Cursor, net::TcpListener, - os::unix::fs::{MetadataExt, symlink}, + os::unix::fs::{MetadataExt, PermissionsExt, symlink}, process::Command, sync::{ Arc, @@ -799,9 +889,22 @@ mod tests { } } - fn assert_load_rejection_keeps_opener_unregistered(state: ObserverState) { + async fn start_peer_session(peer: &PrivateLinkPeer) -> (tempfile::TempDir, PrivateLinkSession) { + let temp = tempfile::tempdir().unwrap(); + let session = start_private_link_session(temp.path(), peer.credential(), "stream") + .await + .unwrap(); + (temp, session) + } + + async fn assert_load_rejection_keeps_opener_unregistered(state: ObserverState) { let temp = tempfile::tempdir().unwrap(); persist_observer(temp.path(), &state).unwrap(); + let peer = PrivateLinkPeer::start().await; + let opener = PrivateLinkOpener::new( + TransportClient::new(peer.credential(), None).unwrap(), + "stream".to_owned(), + ); let loaded = load_observer( temp.path(), "instance", @@ -809,19 +912,22 @@ mod tests { &Url::parse("http://127.0.0.1:1").unwrap(), ) .unwrap(); - assert!(loaded.is_none()); - let headers = proxy_headers_for_auth(&[], &OpenerAuth::Unregistered); + if let Some(observer) = loaded { + opener.set_registered(&observer).unwrap(); + } + let headers = opener.proxy_headers(&[]).unwrap(); assert_eq!( headers, vec![(PROTOCOL_VERSION_HEADER_NAME.to_owned(), "2".to_owned())] ); + peer.shutdown().await; } macro_rules! opener_rejection_test { ($name:ident, $state:expr) => { - #[test] - fn $name() { - assert_load_rejection_keeps_opener_unregistered($state); + #[tokio::test] + async fn $name() { + assert_load_rejection_keeps_opener_unregistered($state).await; } }; } @@ -850,7 +956,7 @@ mod tests { opener_rejection_test!( opener_stays_unregistered_for_unsafe_key, ObserverState { - key: "bad\rkey".into(), + key: "bad\u{1}key".into(), ..observer("/ingest") } ); @@ -890,9 +996,15 @@ mod tests { opener_rejection_test!(opener_stays_unregistered_for_fragment, observer("/a#f")); opener_rejection_test!(opener_stays_unregistered_for_backslash, observer("/a\\b")); - fn assert_redacted_rejection(bytes: &[u8]) { + async fn assert_redacted_setup_rejection(bytes: &[u8]) { let temp = tempfile::tempdir().unwrap(); - let error = read_pair_link(Cursor::new(bytes)).unwrap_err(); + let pairer = FakePairer { + calls: Arc::new(AtomicUsize::new(0)), + result: Some(credential()), + }; + let error = setup_with_pairer(&pairer, temp.path(), "device", Cursor::new(bytes.to_vec())) + .await + .unwrap_err(); let material = String::from_utf8_lossy(bytes); if !material.is_empty() { assert!(!format!("{error}").contains(material.as_ref())); @@ -902,29 +1014,29 @@ mod tests { assert!(!temp.path().join(OBSERVER_FILENAME).exists()); } - #[test] - fn pair_input_empty_is_rejected() { - assert_redacted_rejection(b""); + #[tokio::test] + async fn pair_input_empty_is_rejected() { + assert_redacted_setup_rejection(b"").await; } - #[test] - fn pair_input_invalid_utf8_is_rejected() { - assert_redacted_rejection(b"\xff"); + #[tokio::test] + async fn pair_input_invalid_utf8_is_rejected() { + assert_redacted_setup_rejection(b"\xff").await; } - #[test] - fn pair_input_embedded_whitespace_is_rejected() { - assert_redacted_rejection(b"pair link"); + #[tokio::test] + async fn pair_input_embedded_whitespace_is_rejected() { + assert_redacted_setup_rejection(b"pair link").await; } - #[test] - fn pair_input_leading_whitespace_is_rejected() { - assert_redacted_rejection(b" pair"); + #[tokio::test] + async fn pair_input_leading_whitespace_is_rejected() { + assert_redacted_setup_rejection(b" pair").await; } #[test] fn pair_input_trailing_spaces_and_tabs_are_accepted_after_trim() { assert_eq!(read_pair_link(Cursor::new(b"pair \t")).unwrap(), "pair"); } - #[test] - fn pair_input_multiple_line_endings_are_rejected() { - assert_redacted_rejection(b"pair\nother\n"); + #[tokio::test] + async fn pair_input_multiple_line_endings_are_rejected() { + assert_redacted_setup_rejection(b"pair\nother\n").await; } #[test] fn pair_input_without_terminator_is_accepted() { @@ -945,9 +1057,9 @@ mod tests { 4096 ); } - #[test] - fn pair_input_4097_bytes_is_rejected() { - assert_redacted_rejection(&vec![b'a'; 4097]); + #[tokio::test] + async fn pair_input_4097_bytes_is_rejected() { + assert_redacted_setup_rejection(&vec![b'a'; 4097]).await; } struct FakePairer { @@ -1213,6 +1325,21 @@ mod tests { assert!(!referent.join(PRIVATE_STATE_LOCK_FILENAME).exists()); } + #[test] + fn lock_verification_rejects_unexpected_mode() { + let temp = tempfile::tempdir().unwrap(); + let lock_path = temp.path().join(PRIVATE_STATE_LOCK_FILENAME); + fs::write(&lock_path, b"").unwrap(); + fs::set_permissions(&lock_path, fs::Permissions::from_mode(0o640)).unwrap(); + let file = File::open(lock_path).unwrap(); + assert!(matches!( + verify_private_lock(&file), + Err(PrivateStateError::InvalidTarget { + kind: PrivateTargetKind::Lock + }) + )); + } + struct CountingReader(Arc); impl Read for CountingReader { fn read(&mut self, _buffer: &mut [u8]) -> io::Result { @@ -1246,13 +1373,63 @@ mod tests { } #[tokio::test] - async fn bridge_reuses_one_carrier_across_registration_transition() { + async fn second_runtime_session_is_lock_contended() { + let temp = tempfile::tempdir().unwrap(); + let peer = PrivateLinkPeer::start().await; + let first = start_private_link_session(temp.path(), peer.credential(), "stream") + .await + .unwrap(); + assert!(matches!( + start_private_link_session(temp.path(), peer.credential(), "stream").await, + Err(PrivateStateError::LockContended) + )); + first.shutdown().await.unwrap(); + peer.shutdown().await; + } + + #[tokio::test] + async fn session_restores_durable_observer_into_real_opener() { + let temp = tempfile::tempdir().unwrap(); let peer = PrivateLinkPeer::start().await; peer.enqueue_response(200, b"{}".to_vec()); - peer.enqueue_response(200, b"{}".to_vec()); - let session = start_private_link_session(peer.credential(), "stream") + let durable = ObserverState { + credential_instance_id: peer.credential().instance_id, + ..observer("/ingest") + }; + persist_observer(temp.path(), &durable).unwrap(); + let session = start_private_link_session(temp.path(), peer.credential(), "stream") .await .unwrap(); + session + .request(Method::GET, "/restored") + .unwrap() + .send() + .await + .unwrap(); + let requests = peer.requests(); + assert_eq!(requests.len(), 1); + for name in [ + OBSERVER_HEADER_NAME, + PROTOCOL_VERSION_HEADER_NAME, + "authorization", + ] { + assert!( + requests[0] + .headers + .iter() + .any(|(candidate, _)| candidate.eq_ignore_ascii_case(name)) + ); + } + session.shutdown().await.unwrap(); + peer.shutdown().await; + } + + #[tokio::test] + async fn bridge_reuses_one_carrier_across_registration_transition() { + let peer = PrivateLinkPeer::start().await; + peer.enqueue_response(200, b"{}".to_vec()); + peer.enqueue_response(200, b"{}".to_vec()); + let (_temp, session) = start_peer_session(&peer).await; assert_eq!( session .request(Method::GET, "/unregistered") @@ -1263,10 +1440,7 @@ mod tests { .status(), StatusCode::OK ); - session - .opener - .set_registered(&observer("/app/observer/ingest")) - .unwrap(); + publish_observer_registration(&session, &observer("/app/observer/ingest")).unwrap(); assert_eq!( session .request(Method::GET, "/registered") @@ -1336,9 +1510,7 @@ mod tests { #[tokio::test] async fn bridge_rejects_untrusted_local_authority_and_auth_without_upstream() { let peer = PrivateLinkPeer::start().await; - let session = start_private_link_session(peer.credential(), "stream") - .await - .unwrap(); + let (_temp, session) = start_peer_session(&peer).await; for (name, value) in [ (OBSERVER_HEADER_NAME, "forged"), (PROTOCOL_VERSION_HEADER_NAME, "2"), @@ -1379,9 +1551,7 @@ mod tests { async fn loopback_client_does_not_follow_upstream_redirects() { let peer = PrivateLinkPeer::start().await; peer.enqueue_response(302, Vec::new()); - let session = start_private_link_session(peer.credential(), "stream") - .await - .unwrap(); + let (_temp, session) = start_peer_session(&peer).await; let response = session .request(Method::GET, "/redirect") .unwrap() @@ -1399,9 +1569,7 @@ mod tests { let peer = PrivateLinkPeer::start().await; let body = vec![b'x'; spl_core::mux::INITIAL_WINDOW + 131_072]; peer.enqueue_response(200, body.clone()); - let session = start_private_link_session(peer.credential(), "stream") - .await - .unwrap(); + let (_temp, session) = start_peer_session(&peer).await; let received = session .request(Method::GET, "/large") .unwrap() @@ -1450,9 +1618,7 @@ mod tests { ); let peer = PrivateLinkPeer::start().await; peer.enqueue_response(200, b"{}".to_vec()); - let session = start_private_link_session(peer.credential(), "stream") - .await - .unwrap(); + let (_temp, session) = start_peer_session(&peer).await; assert_eq!( session .request(Method::GET, "/proxy-proof") @@ -1477,7 +1643,7 @@ mod tests { for _ in 0..6 { peer.enqueue_response(200, b"{}".to_vec()); } - let session = start_private_link_session(peer.credential(), "stream") + let session = start_private_link_session(temp.path(), peer.credential(), "stream") .await .unwrap(); let next = ObserverState { @@ -1493,18 +1659,23 @@ mod tests { DurableWriteStage::DirSync, ] { assert!( - persist_and_publish_observer( - temp.path(), - &next, - &session.opener, - &FailStage(stage) - ) - .is_err() - ); - assert_eq!( - fs::read(temp.path().join(OBSERVER_FILENAME)).unwrap(), - prior_bytes + session + .publish_observer_with_fault(&next, &FailStage(stage)) + .is_err() ); + let current_bytes = fs::read(temp.path().join(OBSERVER_FILENAME)).unwrap(); + if stage == DurableWriteStage::DirSync { + assert!( + current_bytes == prior_bytes + || current_bytes == serde_json::to_vec(&next).unwrap() + ); + assert!( + serde_json::from_slice::(¤t_bytes).is_ok(), + "directory-sync failure must leave one complete observer value" + ); + } else { + assert_eq!(current_bytes, prior_bytes); + } let loaded = load_observer( temp.path(), "instance", @@ -1512,7 +1683,11 @@ mod tests { &Url::parse("http://127.0.0.1:1").unwrap(), ) .unwrap(); - assert!(loaded.as_ref() == Some(&prior)); + if stage == DurableWriteStage::DirSync { + assert!(loaded.as_ref() == Some(&prior) || loaded.as_ref() == Some(&next)); + } else { + assert!(loaded.as_ref() == Some(&prior)); + } session .request(Method::GET, "/still-unregistered") .unwrap() @@ -1520,13 +1695,7 @@ mod tests { .await .unwrap(); } - persist_and_publish_observer( - temp.path(), - &next, - &session.opener, - &crate::private_file::NoWriteFault, - ) - .unwrap(); + publish_observer_registration(&session, &next).unwrap(); session .request(Method::GET, "/registered-after-durable") .unwrap() @@ -1596,24 +1765,18 @@ mod tests { let prior_bytes = fs::read(temp.path().join(CREDENTIALS_FILENAME)).unwrap(); let peer = PrivateLinkPeer::start().await; let session = start_private_link_session_inner( + temp.path(), peer.credential(), "stream", - Some(( - temp.path().to_path_buf(), - Arc::new(RecordingFault { - stages: Arc::new(Mutex::new(Vec::new())), - fail: Some(stage), - }), - )), - None, + Arc::new(RecordingFault { + stages: Arc::new(Mutex::new(Vec::new())), + fail: Some(stage), + }), + SessionTestCapture::default(), ) .await .unwrap(); - session - .token_persistence - .as_ref() - .unwrap() - .persist("failed-refresh", 999); + session.token_persistence.persist("failed-refresh", 999); assert_eq!( fs::read(temp.path().join(CREDENTIALS_FILENAME)).unwrap(), prior_bytes @@ -1647,17 +1810,26 @@ mod tests { .to_owned(); let device_token = "device-token-sentinel"; paired.device_token = Some(device_token.into()); - let capability = Mutex::new(None); - let session = start_private_link_session_inner(paired, "stream", None, Some(&capability)) - .await - .unwrap(); - let capability = capability.into_inner().unwrap().unwrap(); + let temp = tempfile::tempdir().unwrap(); + let capability = Arc::new(Mutex::new(None)); + let session = start_private_link_session_inner( + temp.path(), + paired, + "stream", + Arc::new(NoWriteFault), + SessionTestCapture { + capability: Some(capability.clone()), + }, + ) + .await + .unwrap(); + let capability = capability.lock().unwrap().clone().unwrap(); let observer_key = "observer-key-sentinel"; let registered = ObserverState { key: observer_key.into(), ..observer("/ingest") }; - session.opener.set_registered(®istered).unwrap(); + publish_observer_registration(&session, ®istered).unwrap(); let request_debug = format!( "{:?}", session diff --git a/crates/solstone-linux/src/private_link_test_peer.rs b/crates/solstone-linux/src/private_link_test_peer.rs index 30370ea..a2e8e44 100644 --- a/crates/solstone-linux/src/private_link_test_peer.rs +++ b/crates/solstone-linux/src/private_link_test_peer.rs @@ -36,10 +36,7 @@ use tokio_rustls::{TlsAcceptor, server::TlsStream}; #[derive(Clone)] pub(crate) struct PeerRequest { - pub(crate) method: String, - pub(crate) path: String, pub(crate) headers: Vec<(String, String)>, - pub(crate) body: Vec, } #[derive(Clone)] @@ -59,7 +56,6 @@ struct PeerState { responses: Arc>>, requests: Arc>>, accepted: Arc, - uploaded: Arc, } pub(crate) struct PrivateLinkPeer { @@ -76,7 +72,6 @@ impl PrivateLinkPeer { responses: Arc::new(Mutex::new(VecDeque::new())), requests: Arc::new(Mutex::new(Vec::new())), accepted: Arc::new(AtomicUsize::new(0)), - uploaded: Arc::new(AtomicUsize::new(0)), }; let task_state = state.clone(); let task = tokio::spawn(async move { @@ -202,9 +197,6 @@ async fn serve_carrier(mut tls: TlsStream, state: &PeerState) -> io:: requests.entry(frame.stream_id).or_default(); } if frame.flags & FLAG_DATA != 0 { - state - .uploaded - .fetch_add(frame.payload.len(), Ordering::SeqCst); requests .entry(frame.stream_id) .or_default() @@ -308,16 +300,11 @@ fn parse_request(raw: &[u8]) -> Option { let head = std::str::from_utf8(&raw[..split]).ok()?; let mut lines = head.split("\r\n"); let mut request = lines.next()?.split_whitespace(); - let method = request.next()?.to_owned(); - let path = request.next()?.to_owned(); + request.next()?; + request.next()?; let headers = lines .filter_map(|line| line.split_once(':')) .map(|(name, value)| (name.to_owned(), value.trim().to_owned())) .collect(); - Some(PeerRequest { - method, - path, - headers, - body: raw[split + 4..].to_vec(), - }) + Some(PeerRequest { headers }) } diff --git a/crates/solstone-linux/src/upload.rs b/crates/solstone-linux/src/upload.rs index 023b128..8d29b29 100644 --- a/crates/solstone-linux/src/upload.rs +++ b/crates/solstone-linux/src/upload.rs @@ -809,6 +809,44 @@ mod tests { path } + fn find_bytes(haystack: &[u8], needle: &[u8]) -> Option { + haystack + .windows(needle.len()) + .position(|window| window == needle) + } + + type ParsedPart<'a> = (Vec<(&'a str, &'a str)>, &'a [u8]); + + fn parse_multipart<'a>(body: &'a [u8], boundary: &str) -> Vec> { + let delimiter = format!("--{boundary}").into_bytes(); + let next_delimiter = [b"\r\n".as_slice(), delimiter.as_slice()].concat(); + let mut cursor = body; + let mut parts = Vec::new(); + loop { + assert!(cursor.starts_with(&delimiter)); + cursor = &cursor[delimiter.len()..]; + if cursor.starts_with(b"--\r\n") { + assert_eq!(cursor, b"--\r\n"); + return parts; + } + assert!(cursor.starts_with(b"\r\n")); + cursor = &cursor[2..]; + let header_end = find_bytes(cursor, b"\r\n\r\n").unwrap(); + let header_text = std::str::from_utf8(&cursor[..header_end]).unwrap(); + let headers = header_text + .split("\r\n") + .map(|line| { + let (name, value) = line.split_once(": ").unwrap(); + (name, value) + }) + .collect(); + cursor = &cursor[header_end + 4..]; + let body_end = find_bytes(cursor, &next_delimiter).unwrap(); + parts.push((headers, &cursor[..body_end])); + cursor = &cursor[body_end + 2..]; + } + } + // tests/test_upload.py::test_ensure_registered_posts_descriptor_and_persists #[tokio::test] async fn ensure_registered_posts_descriptor_and_persists() { @@ -873,26 +911,82 @@ mod tests { (requests[0].method.as_str(), requests[0].uri.as_str()), ("POST", "/app/observer/register") ); + assert_eq!( + requests[0].headers.get("content-type").unwrap(), + "application/json" + ); + assert!(requests[0].headers.get("authorization").is_none()); + assert_eq!( + serde_json::from_slice::(&requests[0].body).unwrap(), + json!({ + "hostname": "host-a", + "label": "host-a", + "platform": "linux", + "stream_type": "desktop", + "version": "0.1.0", + }) + ); assert_eq!( (requests[1].method.as_str(), requests[1].uri.as_str()), ("POST", "/app/observer/ingest") ); - assert!( + assert_eq!( requests[1] .headers .get("authorization") .unwrap() .to_str() - .unwrap() - .starts_with("Bearer K123456789") + .unwrap(), + "Bearer K123456789" + ); + let content_type = requests[1] + .headers + .get("content-type") + .unwrap() + .to_str() + .unwrap(); + let boundary = content_type + .strip_prefix("multipart/form-data; boundary=") + .filter(|value| { + !value.is_empty() + && value + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || byte == b'-') + }) + .unwrap(); + assert_eq!( + content_type, + format!("multipart/form-data; boundary={boundary}") + ); + let parts = parse_multipart(&requests[1].body, boundary); + assert_eq!(parts.len(), 3); + assert_eq!( + parts[0], + ( + vec![("Content-Disposition", "form-data; name=\"day\"")], + b"20260101".as_slice() + ) + ); + assert_eq!( + parts[1], + ( + vec![("Content-Disposition", "form-data; name=\"segment\"")], + b"120000".as_slice() + ) + ); + assert_eq!( + parts[2], + ( + vec![ + ( + "Content-Disposition", + "form-data; name=\"files\"; filename=\"capture.jsonl\"" + ), + ("Content-Type", "application/octet-stream"), + ], + b"{\"event\":1}\n".as_slice() + ) ); - let upload = String::from_utf8_lossy(&requests[1].body); - assert!(upload.contains("name=\"day\"")); - assert!(upload.contains("20260101")); - assert!(upload.contains("name=\"segment\"")); - assert!(upload.contains("120000")); - assert!(upload.contains("name=\"files\"")); - assert!(upload.contains("{\"event\":1}")); assert_eq!( (requests[2].method.as_str(), requests[2].uri.as_str()), ("GET", "/app/observer/ingest/segments/20260101")