diff --git a/src-tauri/src/config.rs b/src-tauri/src/config.rs index e0c68d048..ac33dc66c 100644 --- a/src-tauri/src/config.rs +++ b/src-tauri/src/config.rs @@ -309,6 +309,8 @@ pub(crate) fn atomic_write_durable( data: &[u8], new_file_mode: Option, ) -> Result<(), AppError> { + #[cfg(not(unix))] + let _ = new_file_mode; if let Some(parent) = path.parent() { fs::create_dir_all(parent).map_err(|e| AppError::io(parent, e))?; } diff --git a/src-tauri/src/pi_config/gateway.rs b/src-tauri/src/pi_config/gateway.rs index 724a2b992..8e96e05f5 100644 --- a/src-tauri/src/pi_config/gateway.rs +++ b/src-tauri/src/pi_config/gateway.rs @@ -281,13 +281,13 @@ impl CandidateHeaderPlan { }); return Err(reasons); }; - let endpoint = match Url::parse(&model.base_url) { - Ok(endpoint) if valid_gateway_endpoint(&endpoint) => endpoint, - _ => { - reasons.push(PiGatewayReason { - code: PiGatewayReasonCode::InvalidEndpoint, - json_pointer: format!("/models/{model_index}/baseUrl"), - }); + let endpoint = match parse_pi_gateway_endpoint( + &model.base_url, + format!("/models/{model_index}/baseUrl"), + ) { + Ok(endpoint) => endpoint, + Err(reason) => { + reasons.push(reason); return Err(reasons); } }; @@ -373,16 +373,7 @@ impl CandidateHeaderPlan { } pub(crate) fn with_endpoint(&self, endpoint: &str) -> Result { - let endpoint = Url::parse(endpoint).map_err(|_| PiGatewayReason { - code: PiGatewayReasonCode::InvalidEndpoint, - json_pointer: "/customEndpoints".to_string(), - })?; - if !valid_gateway_endpoint(&endpoint) { - return Err(PiGatewayReason { - code: PiGatewayReasonCode::InvalidEndpoint, - json_pointer: "/customEndpoints".to_string(), - }); - } + let endpoint = parse_pi_gateway_endpoint(endpoint, "/customEndpoints")?; let mut candidate = self.clone(); candidate.endpoint = endpoint; Ok(candidate) @@ -780,11 +771,33 @@ fn parse_transport_header_value(value: &str) -> Option { HeaderValue::from_str(value).ok() } -fn valid_gateway_endpoint(endpoint: &Url) -> bool { - matches!(endpoint.scheme(), "http" | "https") +/// Parse the one endpoint domain accepted by Pi's gateway. +/// +/// Control-plane endpoint mutations and runtime candidate construction both +/// use this function so a value cannot be accepted for storage and rejected +/// only after a request starts. Pi-native base URLs may remain visible as +/// direct-only diagnostics; this validator governs gateway-owned routes. +pub(crate) fn parse_pi_gateway_endpoint( + value: &str, + json_pointer: impl Into, +) -> Result { + let json_pointer = json_pointer.into(); + let endpoint = Url::parse(value).map_err(|_| PiGatewayReason { + code: PiGatewayReasonCode::InvalidEndpoint, + json_pointer: json_pointer.clone(), + })?; + if matches!(endpoint.scheme(), "http" | "https") && endpoint.host().is_some() && endpoint.username().is_empty() && endpoint.password().is_none() + { + Ok(endpoint) + } else { + Err(PiGatewayReason { + code: PiGatewayReasonCode::InvalidEndpoint, + json_pointer, + }) + } } #[cfg(test)] diff --git a/src-tauri/src/pi_config/shared_file.rs b/src-tauri/src/pi_config/shared_file.rs index 12f117e88..362f404ce 100644 --- a/src-tauri/src/pi_config/shared_file.rs +++ b/src-tauri/src/pi_config/shared_file.rs @@ -29,6 +29,9 @@ static BEFORE_ROLLBACK_EXCHANGE: LazyLock> = #[cfg(test)] static FAIL_NEXT_ROLLBACK_RESTORE: LazyLock>> = LazyLock::new(|| Mutex::new(HashMap::new())); +#[cfg(test)] +static FAIL_AFTER_ATOMIC_ROLLBACK_SWAP: LazyLock>> = + LazyLock::new(|| Mutex::new(HashMap::new())); #[cfg(all(test, unix))] static FAIL_NEXT_PARENT_SYNC: LazyLock>> = LazyLock::new(|| Mutex::new(HashMap::new())); @@ -36,7 +39,30 @@ static FAIL_NEXT_PARENT_SYNC: LazyLock>> = #[derive(Debug)] struct StagedReplacement { path: PathBuf, - identity: Metadata, + identity: FileIdentity, +} + +/// Stable-enough namespace identity for conditional cleanup. +/// +/// Installed-file classification pairs it with exact bytes; cleanup at a +/// private random path may use identity alone. Windows obtains the value from +/// an open handle instead of unstable `MetadataExt` APIs, keeping the pinned +/// Rust toolchain buildable without weakening identity to timestamps or +/// content alone. +#[derive(Debug, Clone, PartialEq, Eq)] +struct FileIdentity { + #[cfg(unix)] + device: u64, + #[cfg(unix)] + inode: u64, + #[cfg(windows)] + volume: u32, + #[cfg(windows)] + index: u64, + #[cfg(not(any(unix, windows)))] + len: u64, + #[cfg(not(any(unix, windows)))] + modified: Option, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -323,7 +349,6 @@ fn replace_existing_if_equal( }; } - run_before_rollback_exchange_hook(path)?; match rollback_installed_file( path, &staged.identity, @@ -434,7 +459,7 @@ enum InstalledRollback { /// overwritten. fn rollback_installed_file( path: &Path, - installed_identity: &Metadata, + installed_identity: &FileIdentity, installed_bytes: &[u8], displaced: Option<&Path>, parent: &Path, @@ -446,11 +471,27 @@ fn rollback_installed_file( // actual no-replace operation is part of the production guarantee. log::warn!("retrying {label} rollback after transient failure: {error}"); } + if let Some(displaced) = displaced { + return rollback_replacement_atomically( + path, + installed_identity, + installed_bytes, + displaced, + parent, + max_bytes, + label, + ); + } + + // A failed create has no before-image to exchange back into place. Isolate + // the canonical entry and identify it by file-id plus exact bytes before + // deleting it. If another writer won, restore that writer instead. let rejected = sibling_temp_path(path, "rejected"); match rename_noreplace(path, &rejected) { Ok(()) => {} Err(error) if error.kind() == std::io::ErrorKind::NotFound => { - return restore_displaced_without_overwrite(displaced, path, parent, label); + sync_parent(parent)?; + return Ok(InstalledRollback::Superseded); } Err(error) => { return Err(rename_error( @@ -462,9 +503,9 @@ fn rollback_installed_file( } } - let rejected_metadata = - fs::symlink_metadata(&rejected).map_err(|error| AppError::io(&rejected, error))?; - let rejected_is_installed = same_file(installed_identity, &rejected_metadata) + let rejected_identity = + file_identity_from_path(&rejected).map_err(|error| AppError::io(&rejected, error))?; + let rejected_is_installed = installed_identity == &rejected_identity && read_regular_bytes(&rejected, max_bytes, label) .ok() .flatten() @@ -487,38 +528,134 @@ fn rollback_installed_file( return Ok(restored); } - let restored = restore_displaced_without_overwrite(displaced, path, parent, label)?; + sync_parent(parent)?; remove_file_if_identity(&rejected, installed_identity)?; sync_parent(parent)?; - Ok(restored) + Ok(InstalledRollback::Restored) } -fn restore_displaced_without_overwrite( - displaced: Option<&Path>, +/// Restore a replacement with one canonical-preserving namespace operation. +/// +/// The displaced before-image becomes canonical atomically and the value which +/// occupied the canonical path moves to a private recovery path. If that value +/// is not our staged file, a second external writer won; atomically put it back +/// and retain the older external value as a recovery artifact. A crash at any +/// point leaves a real file at the canonical path rather than a gap between two +/// renames. +fn rollback_replacement_atomically( path: &Path, + installed_identity: &FileIdentity, + installed_bytes: &[u8], + displaced: &Path, parent: &Path, + max_bytes: u64, label: &str, ) -> Result { - let Some(displaced) = displaced else { + if path_is_missing(path) { + // An external delete superseded the failed publication. Do not + // resurrect the older displaced value; retain it for reconciliation. sync_parent(parent)?; - return Ok(InstalledRollback::Restored); - }; - match rename_noreplace(displaced, path) { - Ok(()) => { - sync_parent(parent)?; - Ok(InstalledRollback::Restored) - } - Err(error) if is_destination_exists(&error) => { - sync_parent(parent)?; - Ok(InstalledRollback::Superseded) - } - Err(error) => Err(rename_error( - &format!("restore displaced {label}"), - displaced, - path, - error, - )), + return Ok(InstalledRollback::Superseded); } + if !path_is_installed(path, installed_identity, installed_bytes, max_bytes, label) { + // A newer external value is already canonical. Leave it there; the + // older displaced value is already a recovery artifact. The identity + // check cannot be made atomic with an uncooperative writer, but it + // removes the broad two-exchange window from the normal supersession + // path. + sync_parent(parent)?; + return Ok(InstalledRollback::Superseded); + } + run_before_rollback_exchange_hook(path)?; + // Namespace identity, rather than readability or content type, is the + // rollback witness. A symlink/directory which appeared in the commit + // window is unsafe to consume, but it still belongs back at the canonical + // path instead of being overwritten by our staged regular file. + let displaced_identity = + file_identity_from_path(displaced).map_err(|error| AppError::io(displaced, error))?; + + let swapped_out = match restore_displaced_atomically(displaced, path) { + Ok(swapped_out) => swapped_out, + Err(error) + if error.kind() == std::io::ErrorKind::NotFound + && path_is_missing(path) + && !path_is_missing(displaced) => + { + sync_parent(parent)?; + return Ok(InstalledRollback::Superseded); + } + Err(error) => { + return Err(rename_error( + "atomically restore displaced file", + displaced, + path, + error, + )); + } + }; + // Make the canonical-preserving recovery durable before inspecting or + // cleaning either recovery name. A crash after this barrier may require + // reconciliation, but it cannot replay a canonical-path gap. + sync_parent(parent)?; + + #[cfg(test)] + fail_after_atomic_rollback_swap_for_test(path)?; + + let swapped_is_installed = path_is_installed( + &swapped_out, + installed_identity, + installed_bytes, + max_bytes, + label, + ); + if swapped_is_installed { + remove_file_if_identity(&swapped_out, installed_identity)?; + sync_parent(parent)?; + return Ok(if path_has_identity(path, &displaced_identity) { + InstalledRollback::Restored + } else { + // The rollback itself succeeded, then an external writer changed + // or deleted the canonical value. That newer state is authority. + InstalledRollback::Superseded + }); + } + + // The canonical path changed again after our original exchange. The first + // atomic restore placed the older displaced value at the canonical path + // and preserved the newer value at `swapped_out`. Put that newer value + // back with another atomic operation; even a crash between the two swaps + // leaves a valid external value at the canonical path. + if !path_has_identity(path, &displaced_identity) { + sync_parent(parent)?; + return Ok(InstalledRollback::Superseded); + } + let older_recovery = match restore_displaced_atomically(&swapped_out, path) { + Ok(older_recovery) => older_recovery, + Err(error) + if error.kind() == std::io::ErrorKind::NotFound + && path_is_missing(path) + && !path_is_missing(&swapped_out) => + { + sync_parent(parent)?; + return Ok(InstalledRollback::Superseded); + } + Err(error) => { + let _ = sync_parent(parent); + return Err(AppError::Config(format!( + "{label} preserved a newer external value at {} but could not atomically \ + restore it to {}: {error}", + swapped_out.display(), + path.display() + ))); + } + }; + sync_parent(parent)?; + log::warn!( + "{label} changed again during rollback; an external value remains canonical and another \ + external value is preserved at {}", + older_recovery.display() + ); + Ok(InstalledRollback::Superseded) } fn restore_quarantined_file( @@ -550,20 +687,19 @@ fn restore_quarantined_file( fn path_is_installed( path: &Path, - expected_identity: &Metadata, + expected_identity: &FileIdentity, expected_bytes: &[u8], max_bytes: u64, label: &str, ) -> bool { - fs::symlink_metadata(path).ok().is_some_and(|actual| { - actual.file_type().is_file() - && same_file(expected_identity, &actual) - && read_regular_bytes(path, max_bytes, label) - .ok() - .flatten() - .as_deref() - == Some(expected_bytes) - }) + file_identity_from_path(path) + .ok() + .is_some_and(|actual| &actual == expected_identity) + && read_regular_bytes(path, max_bytes, label) + .ok() + .flatten() + .as_deref() + == Some(expected_bytes) } fn path_is_missing(path: &Path) -> bool { @@ -573,13 +709,21 @@ fn path_is_missing(path: &Path) -> bool { ) } -fn remove_file_if_identity(path: &Path, expected: &Metadata) -> Result { +fn path_has_identity(path: &Path, expected: &FileIdentity) -> bool { + file_identity_from_path(path) + .ok() + .is_some_and(|actual| &actual == expected) +} + +fn remove_file_if_identity(path: &Path, expected: &FileIdentity) -> Result { let actual = match fs::symlink_metadata(path) { Ok(actual) => actual, Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(false), Err(error) => return Err(AppError::io(path, error)), }; - if !actual.file_type().is_file() || !same_file(expected, &actual) { + if !actual.file_type().is_file() + || file_identity_from_path(path).map_err(|error| AppError::io(path, error))? != *expected + { return Ok(false); } fs::remove_file(path).map_err(|error| AppError::io(path, error))?; @@ -629,6 +773,8 @@ fn stage_replacement( preserve_mode: bool, new_file_mode: Option, ) -> Result { + #[cfg(not(unix))] + let _ = (preserve_mode, new_file_mode); let staged = sibling_temp_path(path, "cas"); let mut options = OpenOptions::new(); options.create_new(true).write(true); @@ -665,9 +811,7 @@ fn stage_replacement( file.flush() .and_then(|_| file.sync_all()) .map_err(|error| AppError::io(&staged, error))?; - let identity = file - .metadata() - .map_err(|error| AppError::io(&staged, error))?; + let identity = file_identity_from_file(&file).map_err(|error| AppError::io(&staged, error))?; drop(file); Ok(StagedReplacement { path: staged, @@ -843,16 +987,16 @@ fn rename_noreplace(source: &Path, destination: &Path) -> std::io::Result<()> { #[cfg(windows)] fn replace_file_with_backup(path: &Path, replacement: &Path, backup: &Path) -> std::io::Result<()> { use windows_sys::Win32::Storage::FileSystem::{ReplaceFileW, REPLACEFILE_WRITE_THROUGH}; - let path = wide_path(path); - let replacement = wide_path(replacement); - let backup = wide_path(backup); + let path_wide = wide_path(path); + let replacement_wide = wide_path(replacement); + let backup_wide = wide_path(backup); // SAFETY: all buffers are NUL-terminated and remain alive during the // synchronous Win32 call. let result = unsafe { ReplaceFileW( - path.as_ptr(), - replacement.as_ptr(), - backup.as_ptr(), + path_wide.as_ptr(), + replacement_wide.as_ptr(), + backup_wide.as_ptr(), REPLACEFILE_WRITE_THROUGH, std::ptr::null(), std::ptr::null(), @@ -861,7 +1005,43 @@ fn replace_file_with_backup(path: &Path, replacement: &Path, backup: &Path) -> s if result != 0 { Ok(()) } else { - Err(std::io::Error::last_os_error()) + let error = std::io::Error::last_os_error(); + if error.raw_os_error() == Some(1177) { + // ERROR_UNABLE_TO_MOVE_REPLACEMENT_2 is a documented partial + // success: `path` moved to `backup`, `replacement` kept its name, + // and the canonical name may be absent. Restore the backup without + // overwriting a concurrent writer before surfacing the failure. + return Err(recover_partial_replace_backup(path, backup, error)); + } + Err(error) + } +} + +#[cfg(windows)] +fn recover_partial_replace_backup( + path: &Path, + backup: &Path, + replace_error: std::io::Error, +) -> std::io::Error { + match rename_noreplace(backup, path) { + Ok(()) => std::io::Error::new( + replace_error.kind(), + format!("{replace_error}; restored the partial backup to the canonical path"), + ), + Err(recovery_error) if is_destination_exists(&recovery_error) => std::io::Error::new( + replace_error.kind(), + format!( + "{replace_error}; a concurrent canonical value won and the partial backup remains at {}", + backup.display() + ), + ), + Err(recovery_error) => std::io::Error::new( + replace_error.kind(), + format!( + "{replace_error}; the partial backup remains at {} and could not be restored: {recovery_error}", + backup.display() + ), + ), } } @@ -871,6 +1051,12 @@ fn install_over_existing(staged: &Path, path: &Path) -> std::io::Result Ok(staged.to_path_buf()) } +#[cfg(any(target_os = "linux", target_os = "macos"))] +fn restore_displaced_atomically(displaced: &Path, path: &Path) -> std::io::Result { + exchange_paths(displaced, path)?; + Ok(displaced.to_path_buf()) +} + #[cfg(windows)] fn install_over_existing(staged: &Path, path: &Path) -> std::io::Result { let backup = sibling_temp_path(path, "displaced"); @@ -878,6 +1064,13 @@ fn install_over_existing(staged: &Path, path: &Path) -> std::io::Result Ok(backup) } +#[cfg(windows)] +fn restore_displaced_atomically(displaced: &Path, path: &Path) -> std::io::Result { + let backup = sibling_temp_path(path, "rejected"); + replace_file_with_backup(path, displaced, &backup)?; + Ok(backup) +} + #[cfg(not(any(target_os = "linux", target_os = "macos", windows)))] fn rename_noreplace(_source: &Path, _destination: &Path) -> std::io::Result<()> { Err(std::io::Error::new( @@ -894,6 +1087,14 @@ fn install_over_existing(_staged: &Path, _path: &Path) -> std::io::Result std::io::Result { + Err(std::io::Error::new( + std::io::ErrorKind::Unsupported, + "atomic file recovery is unsupported on this platform", + )) +} + fn is_destination_exists(error: &std::io::Error) -> bool { error.kind() == std::io::ErrorKind::AlreadyExists || matches!(error.raw_os_error(), Some(libc::EEXIST)) @@ -989,6 +1190,34 @@ pub(crate) fn fail_next_rollback_restore_for_test(path: &Path) { .or_insert(1); } +#[cfg(test)] +fn fail_next_after_atomic_rollback_swap_for_test(path: &Path) { + let mut failures = FAIL_AFTER_ATOMIC_ROLLBACK_SWAP + .lock() + .expect("atomic rollback-swap test hook lock"); + failures + .entry(path.to_path_buf()) + .and_modify(|remaining| *remaining = remaining.saturating_add(1)) + .or_insert(1); +} + +#[cfg(test)] +fn fail_after_atomic_rollback_swap_for_test(path: &Path) -> Result<(), AppError> { + let mut failures = FAIL_AFTER_ATOMIC_ROLLBACK_SWAP.lock().map_err(|error| { + AppError::Config(format!("Pi atomic rollback-swap hook is poisoned: {error}")) + })?; + let Some(remaining) = failures.get_mut(path) else { + return Ok(()); + }; + *remaining = remaining.saturating_sub(1); + if *remaining == 0 { + failures.remove(path); + } + Err(AppError::Config( + "injected stop after canonical-preserving rollback swap".to_string(), + )) +} + #[cfg(test)] fn run_before_rollback_restore_hook(path: &Path) -> std::io::Result<()> { let mut failures = FAIL_NEXT_ROLLBACK_RESTORE @@ -1074,29 +1303,81 @@ fn open_read_only(path: &Path) -> std::io::Result { } #[cfg(unix)] -fn same_file(left: &Metadata, right: &Metadata) -> bool { +fn file_identity_from_metadata(metadata: &Metadata) -> FileIdentity { use std::os::unix::fs::MetadataExt; - left.dev() == right.dev() && left.ino() == right.ino() + FileIdentity { + device: metadata.dev(), + inode: metadata.ino(), + } +} + +#[cfg(unix)] +fn file_identity_from_file(file: &File) -> std::io::Result { + file.metadata() + .map(|metadata| file_identity_from_metadata(&metadata)) +} + +#[cfg(unix)] +fn file_identity_from_path(path: &Path) -> std::io::Result { + fs::symlink_metadata(path).map(|metadata| file_identity_from_metadata(&metadata)) } #[cfg(windows)] -fn same_file(left: &Metadata, right: &Metadata) -> bool { - use std::os::windows::fs::MetadataExt; - matches!( - ( - left.volume_serial_number(), - left.file_index(), - right.volume_serial_number(), - right.file_index(), - ), - (Some(left_volume), Some(left_index), Some(right_volume), Some(right_index)) - if left_volume == right_volume && left_index == right_index - ) +fn file_identity_from_file(file: &File) -> std::io::Result { + use std::mem::MaybeUninit; + use std::os::windows::io::AsRawHandle; + use windows_sys::Win32::Storage::FileSystem::{ + GetFileInformationByHandle, BY_HANDLE_FILE_INFORMATION, + }; + + let mut information = MaybeUninit::::zeroed(); + // SAFETY: `file` owns a valid handle for the duration of the synchronous + // call and `information` points to writable, correctly sized storage. + let result = + unsafe { GetFileInformationByHandle(file.as_raw_handle(), information.as_mut_ptr()) }; + if result == 0 { + return Err(std::io::Error::last_os_error()); + } + // SAFETY: a successful GetFileInformationByHandle initializes the entire + // BY_HANDLE_FILE_INFORMATION structure. + let information = unsafe { information.assume_init() }; + Ok(FileIdentity { + volume: information.dwVolumeSerialNumber, + index: ((information.nFileIndexHigh as u64) << 32) | information.nFileIndexLow as u64, + }) +} + +#[cfg(windows)] +fn file_identity_from_path(path: &Path) -> std::io::Result { + use std::os::windows::fs::OpenOptionsExt; + use windows_sys::Win32::Storage::FileSystem::{ + FILE_FLAG_BACKUP_SEMANTICS, FILE_FLAG_OPEN_REPARSE_POINT, + }; + + let file = OpenOptions::new() + .read(true) + .custom_flags(FILE_FLAG_BACKUP_SEMANTICS | FILE_FLAG_OPEN_REPARSE_POINT) + .open(path)?; + file_identity_from_file(&file) } #[cfg(not(any(unix, windows)))] -fn same_file(left: &Metadata, right: &Metadata) -> bool { - left.len() == right.len() && left.modified().ok() == right.modified().ok() +fn file_identity_from_metadata(metadata: &Metadata) -> FileIdentity { + FileIdentity { + len: metadata.len(), + modified: metadata.modified().ok(), + } +} + +#[cfg(not(any(unix, windows)))] +fn file_identity_from_file(file: &File) -> std::io::Result { + file.metadata() + .map(|metadata| file_identity_from_metadata(&metadata)) +} + +#[cfg(not(any(unix, windows)))] +fn file_identity_from_path(path: &Path) -> std::io::Result { + fs::symlink_metadata(path).map(|metadata| file_identity_from_metadata(&metadata)) } #[cfg(windows)] @@ -1147,17 +1428,27 @@ fn read_regular_bytes( } let mut file = open_read_only(path).map_err(|error| AppError::io(path, error))?; let opened = file.metadata().map_err(|error| AppError::io(path, error))?; + let opened_identity = + file_identity_from_file(&file).map_err(|error| AppError::io(path, error))?; let bytes = read_limited(Read::by_ref(&mut file).take(max_bytes + 1), path, max_bytes)?; let completed = file.metadata().map_err(|error| AppError::io(path, error))?; + let completed_identity = + file_identity_from_file(&file).map_err(|error| AppError::io(path, error))?; let current = fs::symlink_metadata(path).map_err(|error| AppError::io(path, error))?; + if !current.file_type().is_file() || is_reparse_point(¤t) { + return Err(AppError::Conflict(format!( + "{label} changed during read: {}", + path.display() + ))); + } + let current_identity = + file_identity_from_path(path).map_err(|error| AppError::io(path, error))?; if !opened.file_type().is_file() || is_reparse_point(&opened) || !completed.file_type().is_file() || is_reparse_point(&completed) - || !current.file_type().is_file() - || is_reparse_point(¤t) - || !same_file(&opened, &completed) - || !same_file(&completed, ¤t) + || opened_identity != completed_identity + || completed_identity != current_identity || opened.len() != bytes.len() as u64 || completed.len() != bytes.len() as u64 || current.len() != bytes.len() as u64 @@ -1247,6 +1538,158 @@ mod tests { ); } + #[test] + fn atomic_rollback_stop_keeps_a_canonical_external_file() { + let temp = tempfile::tempdir().expect("tempdir"); + let path = temp.path().join("shared.md"); + fs::write(&path, b"observed").expect("seed"); + let observed = read_shared_file(&path, 1024, "test").expect("snapshot"); + + replace_before_next_compare_exchange(&path, b"external-a"); + replace_before_next_rollback_exchange(&path, b"external-b"); + fail_next_after_atomic_rollback_swap_for_test(&path); + replace_shared_file(&path, &observed.revision, b"ours", 1024, None, "test") + .expect_err("the injected stop interrupts rollback cleanup"); + + assert_eq!( + fs::read(&path).expect("canonical path remains present"), + b"external-a", + "the first atomic recovery step must never create a canonical-path gap" + ); + assert!( + fs::read_dir(temp.path()) + .expect("recovery directory") + .filter_map(Result::ok) + .any(|entry| fs::read(entry.path()).ok().as_deref() == Some(b"external-b")), + "the newer external value must remain recoverable if cleanup never runs" + ); + } + + #[test] + fn external_delete_wins_over_replacement_rollback() { + let temp = tempfile::tempdir().expect("tempdir"); + let path = temp.path().join("shared.md"); + let staged = temp.path().join("staged"); + let displaced = temp.path().join("displaced"); + fs::write(&staged, b"ours").expect("staged witness"); + fs::write(&displaced, b"external-before").expect("displaced value"); + let identity = file_identity_from_path(&staged).expect("staged identity"); + + let outcome = rollback_replacement_atomically( + &path, + &identity, + b"ours", + &displaced, + temp.path(), + 1024, + "test", + ) + .expect("external deletion is a safe superseding state"); + + assert_eq!(outcome, InstalledRollback::Superseded); + assert!( + !path.exists(), + "rollback must not resurrect the deleted path" + ); + assert_eq!( + fs::read(&displaced).expect("older value retained"), + b"external-before" + ); + } + + #[test] + fn visible_external_winner_is_not_exchanged_during_rollback() { + let temp = tempfile::tempdir().expect("tempdir"); + let path = temp.path().join("shared.md"); + let installed = temp.path().join("installed-witness"); + let displaced = temp.path().join("displaced"); + fs::write(&path, b"external-newer").expect("canonical external winner"); + fs::write(&installed, b"ours").expect("installed witness"); + fs::write(&displaced, b"external-older").expect("older displaced value"); + let identity = file_identity_from_path(&installed).expect("installed identity"); + + let outcome = rollback_replacement_atomically( + &path, + &identity, + b"ours", + &displaced, + temp.path(), + 1024, + "test", + ) + .expect("visible external authority is a safe superseding state"); + + assert_eq!(outcome, InstalledRollback::Superseded); + assert_eq!( + fs::read(&path).expect("newer value stays canonical"), + b"external-newer" + ); + assert_eq!( + fs::read(&displaced).expect("older value stays recoverable"), + b"external-older" + ); + } + + #[cfg(unix)] + #[test] + fn unsafe_displaced_entry_is_restored_by_identity_without_being_followed() { + use std::os::unix::fs::symlink; + + let temp = tempfile::tempdir().expect("tempdir"); + let path = temp.path().join("shared.md"); + let displaced = temp.path().join("displaced"); + let target = temp.path().join("target"); + fs::write(&path, b"ours").expect("installed value"); + fs::write(&target, b"external-target").expect("external target"); + symlink(&target, &displaced).expect("unsafe displaced entry"); + let identity = file_identity_from_path(&path).expect("installed identity"); + + let outcome = rollback_replacement_atomically( + &path, + &identity, + b"ours", + &displaced, + temp.path(), + 1024, + "test", + ) + .expect("namespace identity is sufficient to restore an unsafe entry"); + + assert_eq!(outcome, InstalledRollback::Restored); + assert!( + fs::symlink_metadata(&path) + .expect("canonical entry") + .file_type() + .is_symlink(), + "the external namespace entry must be restored instead of consumed" + ); + assert_eq!(fs::read_link(&path).expect("symlink target"), target); + assert_eq!( + fs::read(&target).expect("target remains untouched"), + b"external-target" + ); + assert!(!displaced.exists(), "our rejected file must be removed"); + } + + #[cfg(windows)] + #[test] + fn windows_partial_replace_restores_backup_to_missing_canonical() { + let temp = tempfile::tempdir().expect("tempdir"); + let path = temp.path().join("shared.md"); + let backup = temp.path().join("backup"); + fs::write(&backup, b"external-before").expect("partial backup"); + + let error = + recover_partial_replace_backup(&path, &backup, std::io::Error::from_raw_os_error(1177)); + + assert!(error.to_string().contains("restored")); + assert_eq!( + fs::read(&path).expect("canonical restored"), + b"external-before" + ); + assert!(!backup.exists()); + } + #[cfg(unix)] #[test] fn create_sync_failure_returns_error_only_after_removing_its_file_identity() { diff --git a/src-tauri/src/proxy/pi_runtime.rs b/src-tauri/src/proxy/pi_runtime.rs index 385d70399..3558c3c84 100644 --- a/src-tauri/src/proxy/pi_runtime.rs +++ b/src-tauri/src/proxy/pi_runtime.rs @@ -40,7 +40,7 @@ struct PiRuntimeModel { family: PiGatewayApiFamily, wire_profile: Vec, plan: CandidateHeaderPlan, - endpoints: Vec, + custom_endpoint_plans: Vec, } #[derive(Debug, Clone)] @@ -164,7 +164,7 @@ impl PiRuntimeSnapshot { )) })?; - let mut candidates = expand_model_attempts(primary, false)?; + let mut candidates = expand_model_attempts(primary, false); if self.app_config.auto_failover_enabled { for provider_id in &self.failover_ids { if provider_id == &binding.provider_id { @@ -183,7 +183,7 @@ impl PiRuntimeSnapshot { { continue; } - candidates.extend(expand_model_attempts(candidate, true)?); + candidates.extend(expand_model_attempts(candidate, true)); } } @@ -195,22 +195,11 @@ impl PiRuntimeSnapshot { } } -fn expand_model_attempts( - model: &PiRuntimeModel, - is_failover: bool, -) -> Result, AppError> { - let mut plans = Vec::with_capacity(model.endpoints.len().saturating_add(1)); +fn expand_model_attempts(model: &PiRuntimeModel, is_failover: bool) -> Vec { + let mut plans = Vec::with_capacity(model.custom_endpoint_plans.len().saturating_add(1)); plans.push(model.plan.clone()); - for endpoint in &model.endpoints { - let plan = model.plan.with_endpoint(endpoint).map_err(gateway_reason)?; - if !plans - .iter() - .any(|existing| existing.endpoint() == plan.endpoint()) - { - plans.push(plan); - } - } - Ok(plans + plans.extend(model.custom_endpoint_plans.iter().cloned()); + plans .into_iter() .map(|plan| PiRequestCandidate { provider_id: model.provider_id.clone(), @@ -219,7 +208,7 @@ fn expand_model_attempts( plan, is_failover, }) - .collect()) + .collect() } impl PiRequestCandidate { @@ -485,16 +474,52 @@ fn runtime_model( plan: CandidateHeaderPlan, endpoints: Vec, ) -> Result { + let custom_endpoint_plans = + build_custom_endpoint_plans(&plan, endpoints, &aggregate.provider.id); Ok(PiRuntimeModel { provider_id: aggregate.provider.id.clone(), provider_name: aggregate.provider.name.clone(), family, wire_profile: canonical_wire_profile(model)?, plan, - endpoints, + custom_endpoint_plans, }) } +fn build_custom_endpoint_plans( + primary: &CandidateHeaderPlan, + endpoints: Vec, + provider_id: &str, +) -> Vec { + let mut plans = Vec::::with_capacity(endpoints.len()); + for endpoint in endpoints { + match primary.with_endpoint(&endpoint) { + Ok(candidate) + if candidate.endpoint() != primary.endpoint() + && !plans + .iter() + .any(|existing| existing.endpoint() == candidate.endpoint()) => + { + plans.push(candidate); + } + Ok(_) => {} + Err(reason) => { + // The write boundary rejects these values. Keeping this + // defensive compatibility path prevents an old/corrupt + // auxiliary endpoint from disabling the valid primary route. + // Never log the URL: it may contain the very userinfo which + // caused rejection. + log::warn!( + "ignoring invalid persisted Pi custom endpoint for provider \ + '{provider_id}': {:?}", + reason.code + ); + } + } + } + plans +} + fn canonical_wire_profile(model: &PiComposedNativeModel) -> Result, AppError> { let mut profile = Map::new(); profile.insert("reasoning".to_string(), Value::Bool(model.reasoning)); @@ -1091,7 +1116,7 @@ impl CommandTree { "failed to assign Pi config command to its job: {error}" )); } - return Ok(Self { job }); + Ok(Self { job }) } } #[cfg(not(windows))] @@ -1108,6 +1133,7 @@ impl CommandTree { } #[cfg(windows)] unsafe { + let _ = child; let _ = windows_sys::Win32::System::JobObjects::TerminateJobObject(self.job, 1); } #[cfg(not(any(unix, windows)))] @@ -1204,6 +1230,61 @@ mod tests { assert!(debug.contains("absent")); } + #[test] + fn invalid_persisted_custom_endpoint_is_quarantined_without_losing_primary_route() { + let config: PiManagedProviderConfig = serde_json::from_value(json!({ + "name": "Provider", + "api": "openai-responses", + "baseUrl": "https://primary.example/v1", + "apiKey": "literal", + "models": [{"id": "model-a"}] + })) + .expect("managed config"); + let composition = + compose_managed_pi_provider("provider", &config).expect("compose provider"); + let primary = assess_composition_for_runtime(&composition) + .plans + .into_iter() + .next() + .expect("primary plan"); + let custom_endpoint_plans = build_custom_endpoint_plans( + &primary, + vec![ + "https://user:secret@invalid.example/v1".to_string(), + "https://mirror.example/v1".to_string(), + "https://mirror.example/v1".to_string(), + "https://primary.example/v1".to_string(), + ], + "provider", + ); + assert_eq!( + custom_endpoint_plans + .iter() + .map(|plan| plan.endpoint().as_str()) + .collect::>(), + ["https://mirror.example/v1"] + ); + + let model = PiRuntimeModel { + provider_id: "provider".to_string(), + provider_name: "Provider".to_string(), + family: PiGatewayApiFamily::OpenAiResponses, + wire_profile: Vec::new(), + plan: primary, + custom_endpoint_plans, + }; + let attempts = expand_model_attempts(&model, false); + assert_eq!(attempts.len(), 2); + assert_eq!( + attempts[0].plan.endpoint().as_str(), + "https://primary.example/v1" + ); + assert_eq!( + attempts[1].plan.endpoint().as_str(), + "https://mirror.example/v1" + ); + } + #[test] fn environment_resolution_matches_vendored_transport_oracle() { std::env::set_var("PI_RUNTIME_TEST_VALUE", "environment-secret"); diff --git a/src-tauri/src/services/pi_catalog.rs b/src-tauri/src/services/pi_catalog.rs index 8f9fb3a4a..c5d43883d 100644 --- a/src-tauri/src/services/pi_catalog.rs +++ b/src-tauri/src/services/pi_catalog.rs @@ -14,6 +14,7 @@ use crate::pi_config::document::{ apply_pi_provider_patch_with_receipt, snapshot_pi_provider_values, PiProviderPatchReceipt, PiProviderValuesSnapshot, }; +use crate::pi_config::gateway::parse_pi_gateway_endpoint; use crate::pi_config::model::{ effective_pi_model, validate_pi_managed_provider, PiManagedProviderConfig, PiManagementStatus, }; @@ -623,6 +624,7 @@ impl PiCatalogCoordinator { let catalog_was_empty = state.db.get_all_providers(PI_APP)?.is_empty(); let provider_key = non_empty_native_key(&provider_key)?; let config = managed_config(&input)?; + validate_pi_initial_endpoints(&input)?; let provider_id = input.id.clone(); if state.db.get_pi_projection_for_key(provider_key)?.is_some() { return Err(AppError::Conflict(format!( @@ -822,7 +824,7 @@ impl PiCatalogCoordinator { url: &str, ) -> Result { ensure_managed_provider(state, provider_id)?; - let normalized = normalize_endpoint(url)?; + let normalized = normalize_gateway_endpoint_for_write(url)?; state.db.add_provider_endpoint( &ProviderKey::new(PI_APP, provider_id)?, NewEndpoint::now(normalized)?, @@ -836,7 +838,7 @@ impl PiCatalogCoordinator { url: &str, ) -> Result { ensure_managed_provider(state, provider_id)?; - let normalized = normalize_endpoint(url)?; + let normalized = normalize_endpoint_key(url)?; state .db .remove_provider_endpoint(&ProviderKey::new(PI_APP, provider_id)?, &normalized)?; @@ -1044,23 +1046,35 @@ fn non_empty_native_key(value: &str) -> Result<&str, AppError> { } } -fn normalize_endpoint(value: &str) -> Result { +fn normalize_endpoint_key(value: &str) -> Result { let normalized = value.trim().trim_end_matches('/').to_string(); if normalized.is_empty() { return Err(AppError::InvalidInput( "Pi endpoint URL cannot be empty".to_string(), )); } - let parsed = url::Url::parse(&normalized) - .map_err(|error| AppError::InvalidInput(format!("invalid Pi endpoint URL: {error}")))?; - if !matches!(parsed.scheme(), "http" | "https") || parsed.host().is_none() { - return Err(AppError::InvalidInput( - "Pi endpoint must be an absolute HTTP(S) URL".to_string(), - )); - } Ok(normalized) } +fn normalize_gateway_endpoint_for_write(value: &str) -> Result { + let normalized = normalize_endpoint_key(value)?; + parse_pi_gateway_endpoint(&normalized, "/customEndpoints").map_err(|_| { + AppError::InvalidInput( + "Pi endpoint must be an absolute HTTP(S) URL without embedded credentials".to_string(), + ) + })?; + Ok(normalized) +} + +fn validate_pi_initial_endpoints(input: &ProviderMutationInput) -> Result<(), AppError> { + if let Some(meta) = input.meta.as_ref() { + for endpoint in meta.custom_endpoints.values() { + normalize_gateway_endpoint_for_write(&endpoint.url)?; + } + } + Ok(()) +} + fn compensate_created_provider( state: &AppState, provider_id: &str, @@ -1391,6 +1405,92 @@ mod tests { Ok(()) } + #[test] + #[serial_test::serial] + fn pi_endpoint_writes_reuse_gateway_validation_without_partial_state() -> Result<(), AppError> { + let temp = tempfile::tempdir().expect("tempdir"); + let _home = TestHome::install(temp.path())?; + let pi_dir = temp.path().join("pi-agent"); + configure_pi_directory(&pi_dir)?; + let state = AppState::new(Arc::new(Database::memory()?)); + let bad_url = "https://user:secret@mirror.example/v1"; + + let mut invalid_initial = managed_input("invalid-initial", "https://primary.example/v1"); + invalid_initial.meta = Some(crate::provider::ProviderMeta { + custom_endpoints: std::collections::HashMap::from([( + bad_url.to_string(), + crate::settings::CustomEndpoint { + url: bad_url.to_string(), + added_at: Some(1), + last_used: None, + }, + )]), + ..Default::default() + }); + let error = PiCatalogCoordinator::apply( + &state, + PiCatalogMutation::CreateProvider { + input: invalid_initial, + provider_key: "invalid-initial".to_string(), + activate_if_first: false, + }, + ) + .expect_err("initial gateway endpoints must be validated before create"); + assert!(matches!(error, AppError::InvalidInput(_))); + assert!(!error.to_string().contains("secret")); + assert!(state + .db + .get_provider_aggregate(PI_APP, "invalid-initial")? + .is_none()); + + PiCatalogCoordinator::apply( + &state, + PiCatalogMutation::CreateProvider { + input: managed_input("endpoint-owner", "https://primary.example/v1"), + provider_key: "endpoint-owner".to_string(), + activate_if_first: false, + }, + )?; + let error = PiCatalogCoordinator::apply( + &state, + PiCatalogMutation::AddEndpoint { + provider_id: "endpoint-owner".to_string(), + url: bad_url.to_string(), + }, + ) + .expect_err("an unusable endpoint must not be persisted"); + assert!(matches!(error, AppError::InvalidInput(_))); + assert!(!error.to_string().contains("secret")); + assert!(state + .db + .get_provider_aggregate(PI_APP, "endpoint-owner")? + .expect("provider remains") + .endpoints + .is_empty()); + + // A database restored from an older build may already contain this + // value. Runtime quarantines it, while deletion remains total over the + // persisted endpoint domain so the user can repair the row. + let key = ProviderKey::new(PI_APP, "endpoint-owner")?; + state + .db + .add_provider_endpoint(&key, NewEndpoint::new(bad_url, Some(2), None)?)?; + PiCatalogCoordinator::apply( + &state, + PiCatalogMutation::RemoveEndpoint { + provider_id: "endpoint-owner".to_string(), + url: bad_url.to_string(), + }, + )?; + assert!(state + .db + .get_provider_aggregate(PI_APP, "endpoint-owner")? + .expect("provider remains") + .endpoints + .is_empty()); + Ok(()) + } + #[test] #[serial_test::serial] fn native_import_revalidates_exact_values_before_claiming_success() -> Result<(), AppError> { diff --git a/src-tauri/src/services/provider/mod.rs b/src-tauri/src/services/provider/mod.rs index c501e6649..7d4879c1c 100644 --- a/src-tauri/src/services/provider/mod.rs +++ b/src-tauri/src/services/provider/mod.rs @@ -2762,8 +2762,13 @@ requires_openai_auth = true ..Default::default() }); - ProviderService::update(&state, AppType::ClaudeDesktop, None, updated.clone()) - .expect("update current provider"); + ProviderService::update( + &state, + AppType::ClaudeDesktop, + None, + provider_to_mutation_input(updated.clone()), + ) + .expect("update current provider"); let backup = db .get_live_backup("claude-desktop")