native/src/update_manager.rs
Rama main · réplica de lectura del repositorio original.
//! CLI/bootstrap update client for schema-v2 release manifests.
//!
//! This stages verified manager/runtime payloads. Installing them is a separate
//! bootstrap step because Windows cannot replace the running manager executable
//! and macOS runtime payloads are signed app-bundle ZIPs.
use futures_util::StreamExt;
use reqwest::StatusCode;
use reqwest::header::{CONTENT_RANGE, RANGE};
use semver::Version;
use serde::Deserialize;
use sha2::{Digest, Sha256};
use std::collections::BTreeMap;
use std::fs;
use std::io::Read;
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};
use tokio::io::AsyncReadExt;
use tokio::io::AsyncWriteExt;
const SCHEMA_VERSION: u32 = 2;
const MANIFEST_MAX_BYTES: usize = 128 * 1024;
const PAYLOAD_MAX_BYTES: u64 = 1024 * 1024 * 1024;
const PAYLOAD_DOWNLOAD_ATTEMPTS: usize = 4;
const PAYLOAD_STALL_TIMEOUT: Duration = Duration::from_secs(30);
const PAYLOAD_THROUGHPUT_WINDOW: Duration = Duration::from_secs(15);
const PAYLOAD_MIN_BYTES_PER_SECOND: u64 = 64 * 1024;
pub const DEFAULT_BOOTSTRAP_MANIFEST_URL: &str =
"https://github.com/devrajmahar/origin-speak/releases/latest/download/bootstrap-update.json";
const GITHUB_RELEASES_BASE_URL: &str =
"https://github.com/devrajmahar/origin-speak/releases/download/";
#[derive(Clone, Debug, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct BootstrapManifest {
pub schema_version: u32,
pub version: String,
pub platforms: BTreeMap<String, PlatformPayloads>,
}
#[derive(Clone, Debug, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct PlatformPayloads {
pub manager: Payload,
pub runtime: Payload,
}
#[derive(Clone, Debug, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Payload {
pub kind: String,
pub arch: String,
pub path: String,
pub url: String,
pub sha256: String,
}
#[derive(Clone, Debug)]
pub struct AvailableUpdate {
pub version: Version,
pub manager: Payload,
pub runtime: Payload,
}
#[derive(Clone, Debug)]
pub struct StagedUpdate {
pub version: Version,
pub manager_path: PathBuf,
pub manager_sha256: String,
pub runtime_path: PathBuf,
pub runtime_sha256: String,
}
pub async fn check_for_update(
current_version: &str,
manifest_url: Option<&str>,
) -> Result<Option<AvailableUpdate>, String> {
let manifest = fetch_manifest(manifest_url).await?;
select_update(current_version, manifest)
}
pub async fn fetch_release(manifest_url: Option<&str>) -> Result<AvailableUpdate, String> {
let manifest = fetch_manifest(manifest_url).await?;
release_for_platform(manifest)
}
pub async fn fetch_release_for_version(version: &Version) -> Result<AvailableUpdate, String> {
let url = versioned_manifest_url(DEFAULT_BOOTSTRAP_MANIFEST_URL, version)?;
let release = fetch_release(Some(url.as_str())).await?;
if release.version != *version {
return Err(format!(
"versioned bootstrap manifest returned {}, expected {version}",
release.version
));
}
Ok(release)
}
fn versioned_manifest_url(
latest_manifest_url: &str,
version: &Version,
) -> Result<reqwest::Url, String> {
let latest = secure_url(latest_manifest_url, "bootstrap manifest")?;
if latest.as_str() == DEFAULT_BOOTSTRAP_MANIFEST_URL {
let versioned = secure_url(
&format!("{GITHUB_RELEASES_BASE_URL}v{version}/bootstrap-update.json"),
"versioned bootstrap manifest",
)?;
return Ok(versioned);
}
let relative = format!("releases/v{version}/bootstrap-update.json");
let versioned = latest
.join(&relative)
.map_err(|error| format!("build versioned bootstrap manifest URL: {error}"))?;
ensure_secure_url(&versioned, "versioned bootstrap manifest")?;
Ok(versioned)
}
async fn fetch_manifest(manifest_url: Option<&str>) -> Result<BootstrapManifest, String> {
let url = secure_url(
manifest_url.unwrap_or(DEFAULT_BOOTSTRAP_MANIFEST_URL),
"bootstrap manifest",
)?;
let client = reqwest::Client::builder()
.connect_timeout(Duration::from_secs(8))
.timeout(Duration::from_secs(20))
.user_agent(format!(
"OriginSpeak/{} cli-manager",
env!("CARGO_PKG_VERSION")
))
.build()
.map_err(|error| format!("create update client: {error}"))?;
let response = client
.get(url)
.send()
.await
.map_err(|error| format!("fetch bootstrap manifest: {error}"))?;
ensure_secure_url(response.url(), "bootstrap manifest redirect target")?;
let response = response
.error_for_status()
.map_err(|error| format!("fetch bootstrap manifest: {error}"))?;
if response
.content_length()
.is_some_and(|size| size > MANIFEST_MAX_BYTES as u64)
{
return Err("bootstrap manifest exceeds the 128 KiB safety limit".to_string());
}
let bytes = response
.bytes()
.await
.map_err(|error| format!("read bootstrap manifest: {error}"))?;
if bytes.len() > MANIFEST_MAX_BYTES {
return Err("bootstrap manifest exceeds the 128 KiB safety limit".to_string());
}
let manifest: BootstrapManifest = serde_json::from_slice(&bytes)
.map_err(|error| format!("parse bootstrap manifest: {error}"))?;
Ok(manifest)
}
pub async fn stage_update<F>(
update: &AvailableUpdate,
staging_dir: &Path,
mut on_progress: F,
) -> Result<StagedUpdate, String>
where
F: FnMut(&str, u64, Option<u64>),
{
validate_available_update(update)?;
prepare_staging_dir(staging_dir, &update.version).await?;
let result = async {
let client = reqwest::Client::builder()
.connect_timeout(Duration::from_secs(8))
.timeout(Duration::from_secs(60 * 30))
.build()
.map_err(|error| format!("create update client: {error}"))?;
let manager_path =
download_payload(&client, &update.manager, staging_dir, &mut on_progress).await?;
let runtime_path =
download_payload(&client, &update.runtime, staging_dir, &mut on_progress).await?;
Ok(StagedUpdate {
version: update.version.clone(),
manager_path,
manager_sha256: update.manager.sha256.clone(),
runtime_path,
runtime_sha256: update.runtime.sha256.clone(),
})
}
.await;
result
}
pub fn verify_staged_payload(
staging_dir: &Path,
version: &Version,
payload: &Path,
expected_sha256: &str,
) -> Result<(), String> {
validate_sha256(expected_sha256)?;
let staging_metadata = fs::symlink_metadata(staging_dir).map_err(|error| {
format!(
"inspect update staging directory {} before apply: {error}",
staging_dir.display()
)
})?;
if !staging_metadata.is_dir() || is_link_or_reparse_point(&staging_metadata) {
return Err(format!(
"refusing update staging path that is not a real owned directory: {}",
staging_dir.display()
));
}
validate_staging_marker_sync(
&staging_dir.join(crate::management::UPDATE_STAGING_MARKER_FILE),
version,
)?;
if payload.parent() != Some(staging_dir) {
return Err(format!(
"refusing update payload outside the owned staging directory: {}",
payload.display()
));
}
verify_regular_file_sha256(payload, expected_sha256, "staged update payload")
}
pub fn verify_regular_file_sha256(
path: &Path,
expected_sha256: &str,
label: &str,
) -> Result<(), String> {
validate_sha256(expected_sha256)?;
let metadata = fs::symlink_metadata(path)
.map_err(|error| format!("inspect {label} {}: {error}", path.display()))?;
if !metadata.is_file() || is_link_or_reparse_point(&metadata) {
return Err(format!(
"refusing {label} that is not a regular file: {}",
path.display()
));
}
let actual = sha256_file(path)?;
if actual != expected_sha256 {
return Err(format!(
"SHA-256 verification failed for {label} {}",
path.display()
));
}
Ok(())
}
fn validate_staging_marker_sync(marker: &Path, version: &Version) -> Result<(), String> {
let metadata = fs::symlink_metadata(marker).map_err(|_| {
format!(
"refusing unowned update staging directory; marker is missing: {}",
marker.display()
)
})?;
if !metadata.is_file() || is_link_or_reparse_point(&metadata) {
return Err(format!(
"refusing invalid update staging marker: {}",
marker.display()
));
}
let contents = fs::read_to_string(marker)
.map_err(|error| format!("read update staging marker: {error}"))?;
let expected = crate::management::update_staging_marker_contents(&version.to_string());
if contents != expected {
return Err(format!(
"refusing update staging directory with a mismatched ownership marker: {}",
marker.display()
));
}
Ok(())
}
fn sha256_file(path: &Path) -> Result<String, String> {
let mut file = fs::File::open(path)
.map_err(|error| format!("open staged payload {}: {error}", path.display()))?;
let mut hasher = Sha256::new();
let mut buffer = vec![0_u8; 1024 * 1024];
loop {
let read = file
.read(&mut buffer)
.map_err(|error| format!("hash staged payload {}: {error}", path.display()))?;
if read == 0 {
break;
}
hasher.update(&buffer[..read]);
}
Ok(format!("{:x}", hasher.finalize()))
}
fn validate_sha256(sha256: &str) -> Result<(), String> {
if sha256.len() == 64
&& sha256
.bytes()
.all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
{
Ok(())
} else {
Err("update payload SHA-256 must be 64 lowercase hexadecimal characters".to_string())
}
}
#[cfg(target_os = "windows")]
fn is_link_or_reparse_point(metadata: &fs::Metadata) -> bool {
use std::os::windows::fs::MetadataExt;
const FILE_ATTRIBUTE_REPARSE_POINT: u32 = 0x0400;
metadata.file_type().is_symlink()
|| metadata.file_attributes() & FILE_ATTRIBUTE_REPARSE_POINT != 0
}
#[cfg(not(target_os = "windows"))]
fn is_link_or_reparse_point(metadata: &fs::Metadata) -> bool {
metadata.file_type().is_symlink()
}
pub async fn stage_runtime_payload<F>(
release: &AvailableUpdate,
staging_dir: &Path,
mut on_progress: F,
) -> Result<PathBuf, String>
where
F: FnMut(&str, u64, Option<u64>),
{
validate_available_update(release)?;
prepare_staging_dir(staging_dir, &release.version).await?;
let result = async {
let client = reqwest::Client::builder()
.connect_timeout(Duration::from_secs(8))
.timeout(Duration::from_secs(60 * 30))
.build()
.map_err(|error| format!("create update client: {error}"))?;
download_payload(&client, &release.runtime, staging_dir, &mut on_progress).await
}
.await;
result
}
#[cfg(test)]
async fn cleanup_owned_staging_dir(staging_dir: &Path, version: &Version) -> Result<(), String> {
let metadata = match tokio::fs::symlink_metadata(staging_dir).await {
Ok(metadata) => metadata,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()),
Err(error) => {
return Err(format!(
"inspect update staging directory {} before cleanup: {error}",
staging_dir.display()
));
}
};
if !metadata.is_dir() || metadata.file_type().is_symlink() {
return Err(format!(
"refusing to clean update staging path that is not a real directory: {}",
staging_dir.display()
));
}
let marker = staging_dir.join(crate::management::UPDATE_STAGING_MARKER_FILE);
validate_staging_marker(&marker, version).await?;
tokio::fs::remove_dir_all(staging_dir)
.await
.map_err(|error| {
format!(
"remove owned update staging {}: {error}",
staging_dir.display()
)
})
}
async fn prepare_staging_dir(staging_dir: &Path, version: &Version) -> Result<(), String> {
let marker = staging_dir.join(crate::management::UPDATE_STAGING_MARKER_FILE);
match tokio::fs::symlink_metadata(staging_dir).await {
Ok(metadata) => {
if !metadata.is_dir() || metadata.file_type().is_symlink() {
return Err(format!(
"refusing update staging path that is not an owned directory: {}",
staging_dir.display()
));
}
validate_staging_marker(&marker, version).await?;
}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
tokio::fs::create_dir(staging_dir)
.await
.map_err(|error| format!("create update staging directory: {error}"))?;
let contents = crate::management::update_staging_marker_contents(&version.to_string());
let mut options = tokio::fs::OpenOptions::new();
options.write(true).create_new(true);
let claim_result = async {
let mut file = options
.open(&marker)
.await
.map_err(|error| format!("claim update staging directory: {error}"))?;
file.write_all(contents.as_bytes())
.await
.map_err(|error| format!("write update staging marker: {error}"))?;
file.flush()
.await
.map_err(|error| format!("flush update staging marker: {error}"))?;
file.sync_all()
.await
.map_err(|error| format!("sync update staging marker: {error}"))?;
Ok::<(), String>(())
}
.await;
if let Err(error) = claim_result {
let _ = tokio::fs::remove_file(&marker).await;
let _ = tokio::fs::remove_dir(staging_dir).await;
return Err(error);
}
}
Err(error) => {
return Err(format!(
"inspect update staging directory {}: {error}",
staging_dir.display()
));
}
}
Ok(())
}
async fn validate_staging_marker(marker: &Path, version: &Version) -> Result<(), String> {
let metadata = tokio::fs::symlink_metadata(marker).await.map_err(|_| {
format!(
"refusing to reuse unowned update staging directory; marker is missing: {}",
marker.display()
)
})?;
if !metadata.is_file() || metadata.file_type().is_symlink() {
return Err(format!(
"refusing invalid update staging marker: {}",
marker.display()
));
}
let mut file = tokio::fs::File::open(marker)
.await
.map_err(|error| format!("open update staging marker: {error}"))?;
let mut contents = String::new();
file.read_to_string(&mut contents)
.await
.map_err(|error| format!("read update staging marker: {error}"))?;
let expected = crate::management::update_staging_marker_contents(&version.to_string());
if contents != expected {
return Err(format!(
"refusing update staging directory with a mismatched ownership marker: {}",
marker.display()
));
}
Ok(())
}
fn select_update(
current_version: &str,
manifest: BootstrapManifest,
) -> Result<Option<AvailableUpdate>, String> {
if manifest.schema_version != SCHEMA_VERSION {
return Err(format!(
"unsupported bootstrap manifest schema {} (expected {SCHEMA_VERSION})",
manifest.schema_version
));
}
let current = Version::parse(current_version)
.map_err(|error| format!("invalid current version {current_version}: {error}"))?;
let available = Version::parse(&manifest.version)
.map_err(|error| format!("invalid available version {}: {error}", manifest.version))?;
if available <= current {
return Ok(None);
}
release_for_platform(manifest).map(Some)
}
fn release_for_platform(manifest: BootstrapManifest) -> Result<AvailableUpdate, String> {
if manifest.schema_version != SCHEMA_VERSION {
return Err(format!(
"unsupported bootstrap manifest schema {} (expected {SCHEMA_VERSION})",
manifest.schema_version
));
}
let available = Version::parse(&manifest.version)
.map_err(|error| format!("invalid available version {}: {error}", manifest.version))?;
let key = platform_key()?;
let payloads = manifest
.platforms
.get(key)
.ok_or_else(|| format!("manifest has no payloads for {key}"))?;
validate_payload_pair(key, payloads)?;
Ok(AvailableUpdate {
version: available,
manager: payloads.manager.clone(),
runtime: payloads.runtime.clone(),
})
}
fn validate_payload_pair(platform: &str, payloads: &PlatformPayloads) -> Result<(), String> {
let (manager_kind, runtime_kind, arch) = match platform {
"windows-x86_64" => ("cli-manager", "silent-runtime", "x86_64"),
"macos-universal" => ("cli-manager", "app-bundle-zip", "universal"),
_ => return Err(format!("unsupported update platform: {platform}")),
};
validate_payload(&payloads.manager, manager_kind, arch)?;
validate_payload(&payloads.runtime, runtime_kind, arch)?;
Ok(())
}
fn validate_available_update(update: &AvailableUpdate) -> Result<(), String> {
let platform = platform_key()?;
let payloads = PlatformPayloads {
manager: update.manager.clone(),
runtime: update.runtime.clone(),
};
validate_payload_pair(platform, &payloads)
}
fn validate_payload(payload: &Payload, kind: &str, arch: &str) -> Result<(), String> {
if payload.kind != kind || payload.arch != arch {
return Err(format!(
"invalid {} payload kind/architecture",
payload.kind
));
}
validate_payload_shape(payload)
}
async fn download_payload<F>(
client: &reqwest::Client,
payload: &Payload,
staging_dir: &Path,
on_progress: &mut F,
) -> Result<PathBuf, String>
where
F: FnMut(&str, u64, Option<u64>),
{
validate_payload_shape(payload)?;
let url = secure_url(&payload.url, "update payload")?;
let destination = staging_dir.join(&payload.path);
let partial = staging_dir.join(format!("{}.part", payload.path));
if verified_existing_payload(&destination, &payload.sha256).await? {
let size = tokio::fs::metadata(&destination)
.await
.map_err(|error| format!("inspect staged {}: {error}", destination.display()))?
.len();
on_progress(&payload.path, size, Some(size));
return Ok(destination);
}
let mut last_error = String::new();
for attempt in 0..PAYLOAD_DOWNLOAD_ATTEMPTS {
let reconnect_slow_transfer = attempt + 1 < PAYLOAD_DOWNLOAD_ATTEMPTS;
match download_payload_attempt(
client,
payload,
&url,
&partial,
&destination,
reconnect_slow_transfer,
on_progress,
)
.await
{
Ok(path) => return Ok(path),
Err(error) => {
last_error = error;
if attempt + 1 < PAYLOAD_DOWNLOAD_ATTEMPTS {
tokio::time::sleep(Duration::from_secs(1_u64 << attempt)).await;
}
}
}
}
Err(format!(
"download {} failed after {PAYLOAD_DOWNLOAD_ATTEMPTS} attempts: {last_error}",
payload.path
))
}
async fn download_payload_attempt<F>(
client: &reqwest::Client,
payload: &Payload,
url: &reqwest::Url,
partial: &Path,
destination: &Path,
reconnect_slow_transfer: bool,
on_progress: &mut F,
) -> Result<PathBuf, String>
where
F: FnMut(&str, u64, Option<u64>),
{
let mut resume_offset = safe_partial_len(partial).await?;
if resume_offset > PAYLOAD_MAX_BYTES {
tokio::fs::remove_file(partial)
.await
.map_err(|error| format!("remove oversized partial {}: {error}", partial.display()))?;
resume_offset = 0;
}
let mut request = client.get(url.clone());
if resume_offset > 0 {
request = request.header(RANGE, format!("bytes={resume_offset}-"));
}
let response = request
.send()
.await
.map_err(|error| format!("download {}: {error}", payload.path))?;
ensure_secure_url(response.url(), "update payload redirect target")?;
if resume_offset > 0 && response.status() == StatusCode::RANGE_NOT_SATISFIABLE {
if hash_file_async(partial.to_path_buf()).await? == payload.sha256 {
return install_verified_partial(partial, destination, payload).await;
}
tokio::fs::remove_file(partial)
.await
.map_err(|error| format!("remove rejected partial {}: {error}", partial.display()))?;
return Err(format!(
"server rejected the saved range for {}; restarting",
payload.path
));
}
let status = response.status();
let response = response
.error_for_status()
.map_err(|error| format!("download {}: {error}", payload.path))?;
let content_length = response.content_length();
let content_range = response
.headers()
.get(CONTENT_RANGE)
.and_then(|value| value.to_str().ok());
let (append, expected_total) = match status {
StatusCode::PARTIAL_CONTENT => {
let total = parse_content_range(content_range, resume_offset, content_length)?;
(resume_offset > 0, Some(total))
}
StatusCode::OK => {
if resume_offset > 0 {
tokio::fs::remove_file(partial).await.map_err(|error| {
format!("restart partial download {}: {error}", partial.display())
})?;
resume_offset = 0;
}
(false, content_length)
}
_ => {
return Err(format!(
"download {} returned unexpected HTTP {status}",
payload.path
));
}
};
if expected_total.is_some_and(|size| size > PAYLOAD_MAX_BYTES) {
return Err(format!("{} exceeds the 1 GiB safety limit", payload.path));
}
let mut file = open_safe_partial(partial, append).await?;
let mut stream = response.bytes_stream();
let mut downloaded = resume_offset;
let mut throughput_started = Instant::now();
let mut throughput_bytes = downloaded;
on_progress(&payload.path, downloaded, expected_total);
loop {
let next = tokio::time::timeout(PAYLOAD_STALL_TIMEOUT, stream.next())
.await
.map_err(|_| {
format!(
"download {} stalled for {} seconds",
payload.path,
PAYLOAD_STALL_TIMEOUT.as_secs()
)
})?;
let Some(chunk) = next else { break };
let chunk = chunk.map_err(|error| format!("download {}: {error}", payload.path))?;
downloaded = downloaded.saturating_add(chunk.len() as u64);
if downloaded > PAYLOAD_MAX_BYTES
|| expected_total.is_some_and(|expected| downloaded > expected)
{
return Err(format!("{} exceeded its declared size", payload.path));
}
file.write_all(&chunk)
.await
.map_err(|error| format!("write {}: {error}", partial.display()))?;
on_progress(&payload.path, downloaded, expected_total);
let throughput_elapsed = throughput_started.elapsed();
if throughput_elapsed >= PAYLOAD_THROUGHPUT_WINDOW {
let window_bytes = downloaded.saturating_sub(throughput_bytes);
if reconnect_slow_transfer && transfer_is_too_slow(window_bytes, throughput_elapsed) {
file.flush()
.await
.map_err(|error| format!("flush {}: {error}", partial.display()))?;
return Err(format!(
"download {} stayed below {} KiB/s for {} seconds; reconnecting",
payload.path,
PAYLOAD_MIN_BYTES_PER_SECOND / 1024,
throughput_elapsed.as_secs()
));
}
throughput_started = Instant::now();
throughput_bytes = downloaded;
}
}
file.flush()
.await
.map_err(|error| format!("flush {}: {error}", partial.display()))?;
file.sync_all()
.await
.map_err(|error| format!("sync {}: {error}", partial.display()))?;
drop(file);
if downloaded == 0 {
return Err(format!("downloaded {} is empty", payload.path));
}
if expected_total.is_some_and(|expected| downloaded != expected) {
return Err(format!(
"download {} ended at {downloaded} bytes, expected {}",
payload.path,
expected_total.unwrap_or_default()
));
}
if hash_file_async(partial.to_path_buf()).await? != payload.sha256 {
tokio::fs::remove_file(partial)
.await
.map_err(|error| format!("remove corrupt partial {}: {error}", partial.display()))?;
return Err(format!("SHA-256 verification failed for {}", payload.path));
}
install_verified_partial(partial, destination, payload).await
}
fn transfer_is_too_slow(bytes: u64, elapsed: Duration) -> bool {
elapsed >= PAYLOAD_THROUGHPUT_WINDOW
&& (bytes as u128) * 1_000_000_000
< (PAYLOAD_MIN_BYTES_PER_SECOND as u128) * elapsed.as_nanos()
}
async fn verified_existing_payload(path: &Path, expected_sha256: &str) -> Result<bool, String> {
let metadata = match tokio::fs::symlink_metadata(path).await {
Ok(metadata) => metadata,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(false),
Err(error) => {
return Err(format!(
"inspect staged payload {}: {error}",
path.display()
));
}
};
if !metadata.is_file() || is_link_or_reparse_point(&metadata) {
return Err(format!(
"refusing staged payload that is not a regular file: {}",
path.display()
));
}
if metadata.len() == 0 || metadata.len() > PAYLOAD_MAX_BYTES {
tokio::fs::remove_file(path).await.map_err(|error| {
format!("remove invalid staged payload {}: {error}", path.display())
})?;
return Ok(false);
}
if hash_file_async(path.to_path_buf()).await? == expected_sha256 {
Ok(true)
} else {
tokio::fs::remove_file(path).await.map_err(|error| {
format!("remove corrupt staged payload {}: {error}", path.display())
})?;
Ok(false)
}
}
async fn install_verified_partial(
partial: &Path,
destination: &Path,
payload: &Payload,
) -> Result<PathBuf, String> {
match tokio::fs::remove_file(&destination).await {
Ok(()) => {}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
Err(error) => {
return Err(format!(
"remove stale staged payload {}: {error}",
destination.display()
));
}
}
if let Err(error) = tokio::fs::rename(&partial, &destination).await {
return Err(format!("stage {}: {error}", payload.path));
}
Ok(destination.to_path_buf())
}
async fn safe_partial_len(path: &Path) -> Result<u64, String> {
match tokio::fs::symlink_metadata(path).await {
Ok(metadata) => {
if !metadata.is_file() || is_link_or_reparse_point(&metadata) {
return Err(format!(
"refusing non-regular update partial: {}",
path.display()
));
}
Ok(metadata.len())
}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(0),
Err(error) => Err(format!(
"inspect update partial {}: {error}",
path.display()
)),
}
}
async fn open_safe_partial(path: &Path, append: bool) -> Result<tokio::fs::File, String> {
let existing = safe_partial_len(path).await?;
let mut options = tokio::fs::OpenOptions::new();
if append {
if existing == 0 {
return Err(format!(
"resume partial is missing or empty: {}",
path.display()
));
}
options.append(true);
} else {
match tokio::fs::remove_file(path).await {
Ok(()) => {}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
Err(error) => {
return Err(format!("remove stale partial {}: {error}", path.display()));
}
}
options.write(true).create_new(true);
}
options
.open(path)
.await
.map_err(|error| format!("open {}: {error}", path.display()))
}
async fn hash_file_async(path: PathBuf) -> Result<String, String> {
tokio::task::spawn_blocking(move || sha256_file(&path))
.await
.map_err(|error| format!("hash update payload task failed: {error}"))?
}
fn parse_content_range(
content_range: Option<&str>,
expected_start: u64,
content_length: Option<u64>,
) -> Result<u64, String> {
let value = content_range.ok_or_else(|| "resume response omitted Content-Range".to_string())?;
let value = value
.strip_prefix("bytes ")
.ok_or_else(|| format!("invalid Content-Range: {value}"))?;
let (range, total) = value
.split_once('/')
.ok_or_else(|| format!("invalid Content-Range: {value}"))?;
let (start, end) = range
.split_once('-')
.ok_or_else(|| format!("invalid Content-Range: {value}"))?;
let start = start
.parse::<u64>()
.map_err(|_| format!("invalid Content-Range start: {value}"))?;
let end = end
.parse::<u64>()
.map_err(|_| format!("invalid Content-Range end: {value}"))?;
let total = total
.parse::<u64>()
.map_err(|_| format!("invalid Content-Range total: {value}"))?;
if start != expected_start || end < start || total <= end {
return Err(format!("unexpected Content-Range: {value}"));
}
let range_len = end - start + 1;
if content_length.is_some_and(|length| length != range_len) {
return Err(format!("Content-Range length mismatch: {value}"));
}
Ok(total)
}
fn validate_payload_shape(payload: &Payload) -> Result<(), String> {
if payload.path.is_empty()
|| payload.path.contains('/')
|| payload.path.contains('\\')
|| Path::new(&payload.path)
.file_name()
.and_then(|name| name.to_str())
!= Some(payload.path.as_str())
{
return Err("update payload path must be a single filename".to_string());
}
if payload.sha256.len() != 64
|| !payload
.sha256
.bytes()
.all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
{
return Err(
"update payload SHA-256 must be 64 lowercase hexadecimal characters".to_string(),
);
}
secure_url(&payload.url, "update payload")?;
Ok(())
}
fn secure_url(raw: &str, label: &str) -> Result<reqwest::Url, String> {
let url = reqwest::Url::parse(raw).map_err(|error| format!("invalid {label} URL: {error}"))?;
ensure_secure_url(&url, label)?;
Ok(url)
}
fn ensure_secure_url(url: &reqwest::Url, label: &str) -> Result<(), String> {
if url.scheme() != "https" || !url.username().is_empty() || url.password().is_some() {
return Err(format!("{label} URL must be credential-free HTTPS"));
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
fn payload(kind: &str, arch: &str, path: &str) -> Payload {
Payload {
kind: kind.to_string(),
arch: arch.to_string(),
path: path.to_string(),
url: format!("https://example.invalid/{path}"),
sha256: "a".repeat(64),
}
}
#[test]
fn payload_validation_rejects_path_traversal_and_bad_hashes() {
let mut candidate = payload("cli-manager", "x86_64", "../origin.exe");
assert!(validate_payload_shape(&candidate).is_err());
candidate.path = "origin.exe".to_string();
candidate.sha256 = "ABC".to_string();
assert!(validate_payload_shape(&candidate).is_err());
}
#[test]
fn redirect_targets_must_remain_credential_free_https() {
let https = reqwest::Url::parse("https://updates.example/path").unwrap();
assert!(ensure_secure_url(&https, "redirect").is_ok());
let http = reqwest::Url::parse("http://updates.example/path").unwrap();
assert!(ensure_secure_url(&http, "redirect").is_err());
let credentialed = reqwest::Url::parse("https://user@example.com/path").unwrap();
assert!(ensure_secure_url(&credentialed, "redirect").is_err());
}
#[test]
fn resumed_download_requires_an_exact_content_range() {
assert_eq!(
parse_content_range(Some("bytes 1024-2047/4096"), 1024, Some(1024)).unwrap(),
4096
);
assert!(parse_content_range(Some("bytes 0-1023/4096"), 1024, Some(1024)).is_err());
assert!(parse_content_range(Some("bytes 1024-2047/4096"), 1024, Some(512)).is_err());
assert!(parse_content_range(None, 1024, Some(1024)).is_err());
}
#[test]
fn slow_transfer_reconnect_uses_a_sustained_conservative_floor() {
let window = PAYLOAD_THROUGHPUT_WINDOW;
let floor_bytes = PAYLOAD_MIN_BYTES_PER_SECOND * window.as_secs();
assert!(!transfer_is_too_slow(
PAYLOAD_MIN_BYTES_PER_SECOND,
window - Duration::from_secs(1)
));
assert!(transfer_is_too_slow(floor_bytes - 1, window));
assert!(!transfer_is_too_slow(floor_bytes, window));
assert!(!transfer_is_too_slow(floor_bytes * 10, window));
}
#[tokio::test]
async fn regular_partial_is_retained_for_resume() {
let root = std::env::temp_dir().join(format!(
"origin-speak-update-resume-test-{}",
std::process::id()
));
let _ = std::fs::remove_dir_all(&root);
std::fs::create_dir_all(&root).unwrap();
let partial = root.join("payload.part");
std::fs::write(&partial, b"saved bytes").unwrap();
assert_eq!(safe_partial_len(&partial).await.unwrap(), 11);
let file = open_safe_partial(&partial, true).await.unwrap();
drop(file);
assert_eq!(std::fs::read(&partial).unwrap(), b"saved bytes");
std::fs::remove_dir_all(root).unwrap();
}
#[test]
fn default_manifest_uses_github_releases_latest_download() {
assert_eq!(
DEFAULT_BOOTSTRAP_MANIFEST_URL,
"https://github.com/devrajmahar/origin-speak/releases/latest/download/bootstrap-update.json"
);
}
#[test]
fn github_versioned_manifest_is_pinned_to_exact_release_download() {
let version = Version::parse("1.2.3").unwrap();
let url = versioned_manifest_url(DEFAULT_BOOTSTRAP_MANIFEST_URL, &version).unwrap();
assert_eq!(
url.as_str(),
"https://github.com/devrajmahar/origin-speak/releases/download/v1.2.3/bootstrap-update.json"
);
}
#[test]
fn custom_non_github_manifest_keeps_relative_versioned_release_path() {
let version = Version::parse("1.2.3").unwrap();
let url = versioned_manifest_url(
"https://updates.example.invalid/channel/bootstrap-update.json",
&version,
)
.unwrap();
assert_eq!(
url.as_str(),
"https://updates.example.invalid/channel/releases/v1.2.3/bootstrap-update.json"
);
}
#[tokio::test]
async fn update_partial_refuses_symlink() {
#[cfg(unix)]
{
use std::os::unix::fs::symlink;
let root = std::env::temp_dir().join(format!(
"origin-speak-update-partial-test-{}",
std::process::id()
));
let _ = std::fs::remove_dir_all(&root);
std::fs::create_dir_all(&root).unwrap();
let target = root.join("target");
let partial = root.join("payload.part");
std::fs::write(&target, b"keep").unwrap();
symlink(&target, &partial).unwrap();
assert!(safe_partial_len(&partial).await.is_err());
assert_eq!(std::fs::read(&target).unwrap(), b"keep");
std::fs::remove_dir_all(&root).unwrap();
}
}
#[tokio::test]
async fn owned_staging_cleanup_requires_exact_marker() {
let nonce = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos();
let root = std::env::temp_dir().join(format!(
"origin-speak-update-cleanup-test-{}-{nonce}",
std::process::id()
));
let version = Version::parse("1.2.3").unwrap();
let owned = root.join("owned");
let unowned = root.join("unowned");
std::fs::create_dir_all(owned.join("nested")).unwrap();
std::fs::create_dir_all(&unowned).unwrap();
std::fs::write(
owned.join(crate::management::UPDATE_STAGING_MARKER_FILE),
crate::management::update_staging_marker_contents("1.2.3"),
)
.unwrap();
std::fs::write(owned.join("nested/payload.part"), b"partial").unwrap();
std::fs::write(unowned.join("payload.part"), b"keep").unwrap();
cleanup_owned_staging_dir(&owned, &version).await.unwrap();
assert!(!owned.exists());
assert!(cleanup_owned_staging_dir(&unowned, &version).await.is_err());
assert!(unowned.join("payload.part").is_file());
std::fs::remove_dir_all(root).unwrap();
}
#[test]
fn staged_payload_revalidation_requires_hash_and_exact_ownership_marker() {
let nonce = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos();
let root = std::env::temp_dir().join(format!(
"origin-speak-update-revalidate-test-{}-{nonce}",
std::process::id()
));
let version = Version::parse("2.3.4").unwrap();
std::fs::create_dir_all(&root).unwrap();
std::fs::write(
root.join(crate::management::UPDATE_STAGING_MARKER_FILE),
crate::management::update_staging_marker_contents("2.3.4"),
)
.unwrap();
let staged = root.join("origin.exe");
std::fs::write(&staged, b"verified manager").unwrap();
let expected = sha256_file(&staged).unwrap();
verify_staged_payload(&root, &version, &staged, &expected).unwrap();
std::fs::write(&staged, b"tampered manager").unwrap();
assert!(verify_staged_payload(&root, &version, &staged, &expected).is_err());
std::fs::write(&staged, b"verified manager").unwrap();
std::fs::write(
root.join(crate::management::UPDATE_STAGING_MARKER_FILE),
crate::management::update_staging_marker_contents("9.9.9"),
)
.unwrap();
assert!(verify_staged_payload(&root, &version, &staged, &expected).is_err());
std::fs::remove_dir_all(root).unwrap();
}
#[test]
fn installed_payload_hash_verification_detects_post_copy_tampering() {
let root = std::env::temp_dir().join(format!(
"origin-speak-update-installed-hash-test-{}",
std::process::id()
));
let _ = std::fs::remove_dir_all(&root);
std::fs::create_dir_all(&root).unwrap();
let installed = root.join("origin-runtime.exe");
std::fs::write(&installed, b"expected runtime").unwrap();
let expected = sha256_file(&installed).unwrap();
verify_regular_file_sha256(&installed, &expected, "installed runtime").unwrap();
std::fs::write(&installed, b"changed runtime").unwrap();
assert!(verify_regular_file_sha256(&installed, &expected, "installed runtime").is_err());
std::fs::remove_dir_all(root).unwrap();
}
#[test]
fn update_payload_hashing_fits_on_a_small_thread_stack() {
let root = std::env::temp_dir().join(format!(
"origin-speak-update-small-stack-hash-{}",
std::process::id()
));
let _ = std::fs::remove_dir_all(&root);
std::fs::create_dir_all(&root).unwrap();
let payload = root.join("payload.bin");
std::fs::write(&payload, b"abc").unwrap();
let thread_payload = payload.clone();
let digest = std::thread::Builder::new()
.stack_size(128 * 1024)
.spawn(move || sha256_file(&thread_payload))
.expect("spawn small-stack update hashing thread")
.join()
.expect("small-stack update hashing thread panicked")
.expect("hash update payload");
assert_eq!(
digest,
"ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad"
);
std::fs::remove_dir_all(root).unwrap();
}
}
fn platform_key() -> Result<&'static str, String> {
match (std::env::consts::OS, std::env::consts::ARCH) {
("windows", "x86_64") => Ok("windows-x86_64"),
("macos", "x86_64" | "aarch64") => Ok("macos-universal"),
(os, arch) => Err(format!("updates are not published for {os}/{arch}")),
}
}