use std::{
fs::{self, File, OpenOptions},
io::Write,
path::{Path, PathBuf},
time::{SystemTime, UNIX_EPOCH},
};
use fs2::FileExt;
use serde::{de::DeserializeOwned, Deserialize, Serialize};
use url::Url;
use crate::fs_ops::replace_file;
use crate::{
apply_directory_patch, default_platform, download_to_file, ensure_available_space,
normalize_managed_paths, read_url_to_string, sha256_bytes, validate_version_upgrade,
verify_directory_manifest_signature_with_keys, verify_file_tree, verify_sha256,
ApplyDirectoryPatchOptions, DirectoryUpdateManifest, DownloadEvent, Error, FileTreeManifest,
HttpHeader, Result,
};
const TRANSACTION_SCHEMA_VERSION: u32 = 1;
const PREPARATION_DISK_RESERVE_BYTES: u64 = 64 * 1024 * 1024;
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PrepareDirectoryUpdateOptions {
pub manifest_url: String,
pub application_id: String,
pub install_root: PathBuf,
pub managed_paths: Vec<String>,
pub current_version: String,
pub expected_version: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub platform: Option<String>,
pub cache_dir: PathBuf,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub hpatchz_path: Option<PathBuf>,
#[serde(default)]
pub signature_public_keys: Vec<String>,
#[serde(default = "default_true")]
pub require_signature: bool,
#[serde(default)]
pub headers: Vec<HttpHeader>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub timeout_secs: Option<u64>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub enum PreparedDirectoryUpdateKind {
Prepared,
AlreadyPrepared,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub enum DirectoryTransactionState {
Preparing,
Ready,
SupervisorStarted,
Committing,
AwaitingHealth,
Committed,
RolledBack,
Failed,
HealthTimeout,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct DirectoryUpdateTransaction {
pub schema_version: u32,
pub transaction_id: String,
pub application_id: String,
pub platform: String,
pub current_version: String,
pub target_version: String,
pub install_root: PathBuf,
pub managed_paths: Vec<String>,
pub source_tree_sha256: String,
pub target_tree_sha256: String,
pub patch_sha256: String,
pub patch_size: u64,
pub full_size: u64,
pub state: DirectoryTransactionState,
pub created_at_ms: u64,
pub updated_at_ms: u64,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub failure_reason: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PreparedDirectoryUpdate {
pub kind: PreparedDirectoryUpdateKind,
pub transaction: DirectoryUpdateTransaction,
pub transaction_path: PathBuf,
pub staging_path: PathBuf,
pub patch_path: PathBuf,
pub bytes_downloaded: u64,
pub saved_bytes: u64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
struct PreparedPointer {
transaction_path: PathBuf,
target_version: String,
target_tree_sha256: String,
}
pub async fn load_directory_update_manifest(
manifest_url: &str,
headers: &[HttpHeader],
timeout_secs: Option<u64>,
signature_public_keys: &[String],
require_signature: bool,
expected_version: Option<&str>,
) -> Result<DirectoryUpdateManifest> {
let manifest_text = read_url_to_string(manifest_url, headers, timeout_secs).await?;
let manifest: DirectoryUpdateManifest = serde_json::from_str(&manifest_text)?;
manifest.validate()?;
if signature_public_keys.is_empty() {
if require_signature {
return Err(Error::SignaturePublicKeyMissing);
}
} else {
verify_directory_manifest_signature_with_keys(&manifest, signature_public_keys)?;
}
if let Some(expected_version) = expected_version {
if manifest.version != expected_version {
return Err(Error::UnexpectedVersion {
expected: expected_version.to_string(),
actual: manifest.version,
});
}
}
if require_signature {
validate_production_transport(manifest_url, &manifest)?;
}
Ok(manifest)
}
pub async fn prepare_directory_update<F>(
options: PrepareDirectoryUpdateOptions,
mut on_event: F,
) -> Result<PreparedDirectoryUpdate>
where
F: FnMut(DownloadEvent),
{
if options.application_id.trim().is_empty() {
return Err(Error::Message("application id is empty".to_string()));
}
if options.current_version.trim().is_empty() || options.expected_version.trim().is_empty() {
return Err(Error::Message(
"current and expected versions are required".to_string(),
));
}
validate_version_upgrade(&options.current_version, &options.expected_version)?;
if !options.install_root.is_dir() {
return Err(Error::Message(format!(
"installation root is not a directory: {}",
options.install_root.display()
)));
}
let managed_paths = normalize_managed_paths(&options.managed_paths)?;
fs::create_dir_all(&options.cache_dir)
.map_err(|error| crate::error::io_path(&options.cache_dir, error))?;
let lock = open_update_lock(&options.cache_dir)?;
lock.lock_exclusive().map_err(Error::Io)?;
let platform = options.platform.clone().unwrap_or_else(default_platform);
let manifest = load_directory_update_manifest(
&options.manifest_url,
&options.headers,
options.timeout_secs,
&options.signature_public_keys,
options.require_signature,
Some(&options.expected_version),
)
.await?;
let release = manifest.platform(&platform)?.clone();
if release.managed_paths != managed_paths {
return Err(Error::Message(format!(
"managed path policy mismatch: manifest={:?}, client={managed_paths:?}",
release.managed_paths
)));
}
let delta = release
.delta_from(&options.current_version)
.cloned()
.ok_or_else(|| Error::NoMatchingDelta {
platform: platform.clone(),
current_version: options.current_version.clone(),
})?;
verify_tree_blocking(
options.install_root.clone(),
managed_paths.clone(),
delta.source.clone(),
)
.await?;
let transaction_id = transaction_id(
&options.application_id,
&platform,
&options.current_version,
&manifest.version,
&delta.source.tree_sha256,
&release.target.tree_sha256,
);
let transactions_root = options.cache_dir.join("transactions");
let transaction_root = transactions_root.join(&transaction_id);
let transaction_path = transaction_root.join("transaction.json");
let manifest_path = transaction_root.join("manifest.json");
let patch_path = transaction_root.join("patch.hdiff");
let staging_path = transaction_root.join("staged");
let result_path = transaction_root.join("result.json");
if transaction_path.is_file() {
let transaction: DirectoryUpdateTransaction = read_json(&transaction_path)?;
if result_path.is_file()
|| matches!(
transaction.state,
DirectoryTransactionState::RolledBack
| DirectoryTransactionState::Failed
| DirectoryTransactionState::HealthTimeout
)
{
return Err(Error::Message(format!(
"previous directory update transaction failed: {}",
transaction.transaction_id
)));
}
if matches!(
transaction.state,
DirectoryTransactionState::Ready | DirectoryTransactionState::SupervisorStarted
) && staging_path.is_dir()
{
verify_tree_blocking(
staging_path.clone(),
managed_paths.clone(),
release.target.clone(),
)
.await?;
return Ok(PreparedDirectoryUpdate {
kind: PreparedDirectoryUpdateKind::AlreadyPrepared,
bytes_downloaded: 0,
saved_bytes: release.full.size.saturating_sub(delta.patch.size),
transaction,
transaction_path,
staging_path,
patch_path,
});
}
}
prepare_transactions_root(&transactions_root)?;
let prepared_pointer = options.cache_dir.join("prepared.json");
if prepared_pointer.is_file() {
fs::remove_file(&prepared_pointer)
.map_err(|error| crate::error::io_path(&prepared_pointer, error))?;
}
let required_cache_bytes = release
.target
.total_size
.checked_add(delta.patch.size)
.and_then(|size| size.checked_add(PREPARATION_DISK_RESERVE_BYTES))
.ok_or_else(|| Error::Message("directory update disk requirement overflow".to_string()))?;
ensure_available_space(&options.cache_dir, required_cache_bytes)?;
fs::create_dir_all(&transaction_root)
.map_err(|error| crate::error::io_path(&transaction_root, error))?;
write_json_atomic(&manifest_path, &manifest)?;
let now = now_millis();
let mut transaction = DirectoryUpdateTransaction {
schema_version: TRANSACTION_SCHEMA_VERSION,
transaction_id,
application_id: options.application_id,
platform,
current_version: options.current_version,
target_version: manifest.version.clone(),
install_root: options.install_root.clone(),
managed_paths: managed_paths.clone(),
source_tree_sha256: delta.source.tree_sha256.clone(),
target_tree_sha256: release.target.tree_sha256.clone(),
patch_sha256: delta.patch.sha256.clone(),
patch_size: delta.patch.size,
full_size: release.full.size,
state: DirectoryTransactionState::Preparing,
created_at_ms: now,
updated_at_ms: now,
failure_reason: None,
};
write_json_atomic(&transaction_path, &transaction)?;
let patch_url = resolve_artifact_url(&options.manifest_url, &delta.patch.url)?;
let stats = download_to_file(
&patch_url,
&patch_path,
&options.headers,
options.timeout_secs,
Some(delta.patch.size),
&mut on_event,
)
.await?;
if stats.bytes_written != delta.patch.size {
return Err(Error::SizeMismatch {
path: patch_path,
expected: delta.patch.size,
actual: stats.bytes_written,
});
}
verify_sha256(&patch_path, &delta.patch.sha256)?;
if staging_path.exists() {
fs::remove_dir_all(&staging_path)
.map_err(|error| crate::error::io_path(&staging_path, error))?;
}
apply_directory_patch_blocking(
options.install_root,
patch_path.clone(),
staging_path.clone(),
managed_paths.clone(),
release.target.clone(),
options.hpatchz_path,
)
.await?;
transaction.state = DirectoryTransactionState::Ready;
transaction.updated_at_ms = now_millis();
write_json_atomic(&transaction_path, &transaction)?;
write_json_atomic(
&options.cache_dir.join("prepared.json"),
&PreparedPointer {
transaction_path: transaction_path.clone(),
target_version: manifest.version,
target_tree_sha256: release.target.tree_sha256,
},
)?;
Ok(PreparedDirectoryUpdate {
kind: PreparedDirectoryUpdateKind::Prepared,
bytes_downloaded: stats.bytes_written,
saved_bytes: release.full.size.saturating_sub(stats.bytes_written),
transaction,
transaction_path,
staging_path,
patch_path,
})
}
pub fn read_directory_transaction(path: impl AsRef<Path>) -> Result<DirectoryUpdateTransaction> {
let transaction: DirectoryUpdateTransaction = read_json(path.as_ref())?;
if transaction.schema_version != TRANSACTION_SCHEMA_VERSION {
return Err(Error::Message(format!(
"unsupported transaction schema version: {}",
transaction.schema_version
)));
}
Ok(transaction)
}
pub fn write_directory_transaction(
path: impl AsRef<Path>,
transaction: &DirectoryUpdateTransaction,
) -> Result<()> {
write_json_atomic(path.as_ref(), transaction)
}
pub fn read_directory_manifest_file(path: impl AsRef<Path>) -> Result<DirectoryUpdateManifest> {
let manifest: DirectoryUpdateManifest = read_json(path.as_ref())?;
manifest.validate()?;
Ok(manifest)
}
pub fn write_json_file_atomic<T: Serialize>(path: impl AsRef<Path>, value: &T) -> Result<()> {
write_json_atomic(path.as_ref(), value)
}
async fn verify_tree_blocking(
root: PathBuf,
managed_paths: Vec<String>,
expected: FileTreeManifest,
) -> Result<()> {
tokio::task::spawn_blocking(move || verify_file_tree(root, &managed_paths, &expected))
.await
.map_err(|error| Error::Message(format!("tree verification task failed: {error}")))??;
Ok(())
}
async fn apply_directory_patch_blocking(
old_path: PathBuf,
patch_path: PathBuf,
output_path: PathBuf,
managed_paths: Vec<String>,
expected_tree: FileTreeManifest,
hpatchz_path: Option<PathBuf>,
) -> Result<()> {
tokio::task::spawn_blocking(move || {
apply_directory_patch(&ApplyDirectoryPatchOptions {
old_path,
patch_path,
output_path,
managed_paths,
expected_tree,
hpatchz_path,
cache_size: Some("64m".to_string()),
parallel_threads: Some(4),
verify_checksums: true,
})
})
.await
.map_err(|error| Error::Message(format!("directory patch task failed: {error}")))??;
Ok(())
}
fn open_update_lock(cache_dir: &Path) -> Result<File> {
OpenOptions::new()
.create(true)
.truncate(false)
.read(true)
.write(true)
.open(cache_dir.join("lock"))
.map_err(Error::Io)
}
fn prepare_transactions_root(path: &Path) -> Result<()> {
if !path.exists() {
return fs::create_dir_all(path).map_err(|error| crate::error::io_path(path, error));
}
for entry in fs::read_dir(path).map_err(|error| crate::error::io_path(path, error))? {
let entry = entry.map_err(|error| crate::error::io_path(path, error))?;
let entry_path = entry.path();
if !entry
.file_type()
.map_err(|error| crate::error::io_path(&entry_path, error))?
.is_dir()
{
return Err(Error::Message(format!(
"unexpected entry in directory update transaction store: {}",
entry_path.display()
)));
}
let transaction_path = entry_path.join("transaction.json");
if transaction_path.is_file() {
let transaction = read_directory_transaction(&transaction_path)?;
if matches!(
transaction.state,
DirectoryTransactionState::SupervisorStarted
| DirectoryTransactionState::Committing
| DirectoryTransactionState::AwaitingHealth
| DirectoryTransactionState::HealthTimeout
) {
return Err(Error::Message(format!(
"another directory update transaction is active: {}",
transaction.transaction_id
)));
}
}
fs::remove_dir_all(&entry_path)
.map_err(|error| crate::error::io_path(&entry_path, error))?;
}
Ok(())
}
fn resolve_artifact_url(manifest_url: &str, artifact_url: &str) -> Result<String> {
if Url::parse(artifact_url).is_ok() {
return Ok(artifact_url.to_string());
}
if let Ok(base) = Url::parse(manifest_url) {
if matches!(base.scheme(), "http" | "https" | "file") {
return Ok(base.join(artifact_url)?.to_string());
}
}
let manifest_path = Path::new(manifest_url);
let parent = manifest_path.parent().unwrap_or_else(|| Path::new("."));
Ok(parent.join(artifact_url).to_string_lossy().to_string())
}
fn validate_production_transport(
manifest_url: &str,
manifest: &DirectoryUpdateManifest,
) -> Result<()> {
require_https(manifest_url, "directory update manifest")?;
for release in manifest.platforms.values() {
let full_url = resolve_artifact_url(manifest_url, &release.full.url)?;
require_https(&full_url, "full updater artifact")?;
for delta in &release.deltas {
let patch_url = resolve_artifact_url(manifest_url, &delta.patch.url)?;
require_https(&patch_url, "directory patch artifact")?;
}
}
Ok(())
}
fn require_https(value: &str, label: &str) -> Result<()> {
let scheme = Url::parse(value)
.map(|url| url.scheme().to_string())
.unwrap_or_else(|_| "local-path".to_string());
if scheme == "https" {
return Ok(());
}
Err(Error::InsecureTransport {
label: label.to_string(),
scheme,
})
}
fn transaction_id(
application_id: &str,
platform: &str,
current_version: &str,
target_version: &str,
source_tree: &str,
target_tree: &str,
) -> String {
let payload = format!(
"{application_id}\0{platform}\0{current_version}\0{target_version}\0{source_tree}\0{target_tree}"
);
let digest = sha256_bytes(payload.as_bytes());
format!(
"{}-to-{}-{}",
sanitize(current_version),
sanitize(target_version),
&digest[..16]
)
}
fn sanitize(value: &str) -> String {
value
.chars()
.map(|character| {
if character.is_ascii_alphanumeric() || matches!(character, '-' | '_' | '.') {
character
} else {
'_'
}
})
.collect()
}
fn read_json<T: DeserializeOwned>(path: &Path) -> Result<T> {
Ok(serde_json::from_slice(
&fs::read(path).map_err(|error| crate::error::io_path(path, error))?,
)?)
}
fn write_json_atomic<T: Serialize>(path: &Path, value: &T) -> Result<()> {
if let Some(parent) = path.parent() {
fs::create_dir_all(parent).map_err(|error| crate::error::io_path(parent, error))?;
}
let temporary = path.with_extension(format!("tmp-{}", std::process::id()));
let bytes = serde_json::to_vec_pretty(value)?;
let mut file =
File::create(&temporary).map_err(|error| crate::error::io_path(&temporary, error))?;
file.write_all(&bytes)
.map_err(|error| crate::error::io_path(&temporary, error))?;
file.sync_all()
.map_err(|error| crate::error::io_path(&temporary, error))?;
replace_file(&temporary, path)
}
fn now_millis() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|duration| duration.as_millis() as u64)
.unwrap_or_default()
}
fn default_true() -> bool {
true
}
#[cfg(test)]
mod tests {
use std::fs;
use tempfile::tempdir;
use super::{
prepare_transactions_root, require_https, transaction_id, write_directory_transaction,
DirectoryTransactionState, DirectoryUpdateTransaction,
};
#[test]
fn transaction_id_is_stable() {
let first = transaction_id("app", "windows-x86_64", "1.0.0", "1.1.0", "a", "b");
let second = transaction_id("app", "windows-x86_64", "1.0.0", "1.1.0", "a", "b");
assert_eq!(first, second);
assert!(first.starts_with("1.0.0-to-1.1.0-"));
}
#[test]
fn production_transport_requires_https() {
assert!(require_https("https://updates.example/latest.json", "manifest").is_ok());
assert!(require_https("http://updates.example/latest.json", "manifest").is_err());
assert!(require_https("C:\\updates\\latest.json", "manifest").is_err());
}
#[test]
fn active_transactions_are_never_pruned() {
let dir = tempdir().unwrap();
let transactions = dir.path().join("transactions");
let active = transactions.join("active");
fs::create_dir_all(&active).unwrap();
let transaction = DirectoryUpdateTransaction {
schema_version: 1,
transaction_id: "active".to_string(),
application_id: "test.app".to_string(),
platform: "windows-x86_64".to_string(),
current_version: "1.0.0".to_string(),
target_version: "1.1.0".to_string(),
install_root: dir.path().join("install"),
managed_paths: vec!["app.exe".to_string()],
source_tree_sha256: "a".repeat(64),
target_tree_sha256: "b".repeat(64),
patch_sha256: "c".repeat(64),
patch_size: 1,
full_size: 2,
state: DirectoryTransactionState::Committing,
created_at_ms: 1,
updated_at_ms: 1,
failure_reason: None,
};
write_directory_transaction(active.join("transaction.json"), &transaction).unwrap();
assert!(prepare_transactions_root(&transactions).is_err());
assert!(active.is_dir());
}
}