agent/lib/src/strict.rs
//! Strict transfer mode + provenance receipts.
//!
//! The existing transfer functions in [`crate::operations`] intentionally treat a
//! per-asset failure as a *tally* and return `Ok(())` with failures counted (the
//! documented airgap-dump compatibility contract). This module adds an **opt-in
//! strict mode** *alongside* that behavior — it does not change the existing
//! functions, so best-effort semantics are preserved by construction.
//!
//! In [`TransferMode::Strict`], any failed download, hostile asset path, size
//! mismatch, or missing asset makes the whole operation fail (a non-zero result),
//! a partial archive never appears under the final filename (temp + atomic rename),
//! and a [`TransferReceipt`] — suitable for embedding in a signed update bundle —
//! records every asset's dual digest with credentials stripped from all endpoints.
use std::path::{Path, PathBuf};
use anyhow::{bail, Result};
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use traits::{ConnectorTrait, RemoteAsset};
use crate::operations::is_safe_member_path;
/// 8 GiB per-asset cap (matches the connector download cap).
pub const DEFAULT_MAX_ASSET_BYTES: u64 = 8 * 1024 * 1024 * 1024;
/// Transfer failure policy.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TransferMode {
/// Per-asset failures are tallied; the operation still succeeds (legacy contract).
BestEffort,
/// Any per-asset failure fails the whole operation.
Strict,
}
/// Options for a strict-capable transfer.
pub struct TransferOptions {
pub mode: TransferMode,
pub max_asset_bytes: u64,
pub source_type: String,
/// The configured source endpoint. Credentials are stripped before recording.
pub source_endpoint: String,
/// Where to write the JSON receipt (written even on strict failure when possible).
pub receipt_path: Option<PathBuf>,
}
impl TransferOptions {
pub fn new(
mode: TransferMode,
source_type: impl Into<String>,
source_endpoint: impl Into<String>,
) -> Self {
TransferOptions {
mode,
max_asset_bytes: DEFAULT_MAX_ASSET_BYTES,
source_type: source_type.into(),
source_endpoint: source_endpoint.into(),
receipt_path: None,
}
}
}
/// A safe, credential-free description of one failed asset.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TransferFailure {
pub asset_path: String,
pub reason: String,
}
/// Per-asset provenance record.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AssetRecord {
pub source_path: String,
pub size: u64,
pub sha256: String,
pub blake3: String,
/// Source URI with any credentials removed.
pub source_uri: Option<String>,
}
/// Structured outcome of a transfer.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TransferReport {
pub requested_assets: usize,
pub successful_assets: usize,
pub failed_assets: usize,
pub failures: Vec<TransferFailure>,
pub bytes_transferred: u64,
pub complete: bool,
}
/// A provenance receipt, suitable for embedding as bundle evidence.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TransferReceipt {
pub operation_id: String,
pub mode: String,
pub started_at: DateTime<Utc>,
pub completed_at: DateTime<Utc>,
pub source_type: String,
/// Credentials removed.
pub source_endpoint: String,
pub source_repository: String,
pub destination_type: String,
pub requested_assets: usize,
pub successful_assets: usize,
pub failed_assets: usize,
pub bytes_transferred: u64,
pub assets: Vec<AssetRecord>,
pub failures: Vec<TransferFailure>,
pub final_archive_digest: Option<String>,
pub complete: bool,
}
fn hex(bytes: &[u8]) -> String {
const LUT: &[u8; 16] = b"0123456789abcdef";
let mut s = String::with_capacity(bytes.len() * 2);
for &b in bytes {
s.push(LUT[(b >> 4) as usize] as char);
s.push(LUT[(b & 0xf) as usize] as char);
}
s
}
fn sha256_hex(b: &[u8]) -> String {
hex(&Sha256::digest(b))
}
fn blake3_hex(b: &[u8]) -> String {
blake3::hash(b).to_hex().to_string()
}
/// Remove `user:pass@` userinfo from the authority of a URL, if present.
pub fn strip_credentials(url: &str) -> String {
if let Some(scheme_end) = url.find("://") {
let after = &url[scheme_end + 3..];
let authority_end = after.find('/').unwrap_or(after.len());
if let Some(at) = after[..authority_end].find('@') {
return format!("{}://{}", &url[..scheme_end], &after[at + 1..]);
}
}
url.to_string()
}
/// Pack a directory into `output` (a `.znippy`) atomically. `compress_dir` forces
/// the output name to `<path>.with_extension("znippy")`, so we rename the file it
/// actually wrote into `output`.
fn pack_dir_atomic(dir: &Path, output: &Path) -> Result<()> {
let mut tmp_name = output
.file_name()
.map(|s| s.to_os_string())
.unwrap_or_default();
tmp_name.push(".partial.tmp");
let tmp = output.with_file_name(tmp_name);
let written = tmp.with_extension("znippy");
let _ = std::fs::remove_file(&written);
let dir_pb = dir.to_path_buf();
znippy_compress::compress_dir(&dir_pb, &tmp, false, None, None, None)
.map_err(|e| anyhow::anyhow!("znippy pack failed: {e}"))?;
std::fs::rename(&written, output).map_err(|e| {
let _ = std::fs::remove_file(&written);
anyhow::anyhow!("rename archive into place: {e}")
})?;
Ok(())
}
/// Transfer every asset in `repository` from `connector` into a `.znippy` archive
/// at `output`, honoring [`TransferOptions::mode`]. Lists assets from the connector,
/// then delegates to [`transfer_assets_to_znippy_with_options`].
pub async fn transfer_repo_to_znippy_with_options(
connector: &dyn ConnectorTrait,
repository: &str,
output: &Path,
opts: &TransferOptions,
) -> Result<TransferReport> {
// A listing error is fatal in both modes (matches the existing dump semantics).
let assets = connector.list_assets(repository).await?;
transfer_assets_to_znippy_with_options(connector, repository, &assets, output, opts).await
}
/// Transfer a pre-fetched `assets` list from `connector` into a `.znippy` archive
/// at `output`, honoring [`TransferOptions::mode`]. Returns a structured report. In
/// strict mode an incomplete transfer returns `Err` (after writing the receipt when
/// possible) and leaves no archive at `output`. Used by the CLI, where the asset
/// list comes from a lockfile (world) or a listing (nexus/artifactory/holger).
pub async fn transfer_assets_to_znippy_with_options(
connector: &dyn ConnectorTrait,
repository: &str,
assets: &[RemoteAsset],
output: &Path,
opts: &TransferOptions,
) -> Result<TransferReport> {
let started = Utc::now();
let requested = assets.len();
let staging = tempfile::TempDir::new()?;
let mut records: Vec<AssetRecord> = Vec::new();
let mut failures: Vec<TransferFailure> = Vec::new();
let mut bytes: u64 = 0;
for asset in assets {
if !is_safe_member_path(&asset.path) {
failures.push(TransferFailure {
asset_path: asset.path.clone(),
reason: "unsafe asset path".into(),
});
continue;
}
// Deliberately record a generic reason (never the raw error) so a
// credential-bearing URL cannot leak into the receipt.
let data = match connector.download_asset(asset).await {
Ok(d) => d,
Err(_) => {
failures.push(TransferFailure {
asset_path: asset.path.clone(),
reason: "download failed".into(),
});
continue;
}
};
if data.len() as u64 > opts.max_asset_bytes {
failures.push(TransferFailure {
asset_path: asset.path.clone(),
reason: "asset exceeds size cap".into(),
});
continue;
}
if let Some(advertised) = asset.size {
if advertised != data.len() as u64 {
failures.push(TransferFailure {
asset_path: asset.path.clone(),
reason: "size mismatch versus advertised size".into(),
});
continue;
}
}
let dest = staging.path().join(&asset.path);
if let Some(parent) = dest.parent() {
std::fs::create_dir_all(parent)?;
}
std::fs::write(&dest, &data)?;
bytes += data.len() as u64;
records.push(AssetRecord {
source_path: asset.path.clone(),
size: data.len() as u64,
sha256: sha256_hex(&data),
blake3: blake3_hex(&data),
source_uri: Some(strip_credentials(&asset.download_url)),
});
}
let failed = failures.len();
let complete = failed == 0;
let mode_str = match opts.mode {
TransferMode::BestEffort => "best-effort",
TransferMode::Strict => "strict",
};
let op_id = format!("xfer-{}", started.timestamp_nanos_opt().unwrap_or_default());
// Strict + incomplete: no archive, write the receipt, fail.
if opts.mode == TransferMode::Strict && !complete {
let receipt = TransferReceipt {
operation_id: op_id,
mode: mode_str.into(),
started_at: started,
completed_at: Utc::now(),
source_type: opts.source_type.clone(),
source_endpoint: strip_credentials(&opts.source_endpoint),
source_repository: repository.to_string(),
destination_type: "znippy".into(),
requested_assets: requested,
successful_assets: records.len(),
failed_assets: failed,
bytes_transferred: bytes,
assets: records,
failures: failures.clone(),
final_archive_digest: None,
complete: false,
};
write_receipt(opts, &receipt)?;
bail!(
"strict transfer incomplete: {failed} of {requested} asset(s) failed for repository {repository:?}"
);
}
// Pack the successfully-fetched assets atomically.
pack_dir_atomic(staging.path(), output)?;
let archive_bytes = std::fs::read(output)?;
let final_digest = sha256_hex(&archive_bytes);
let receipt = TransferReceipt {
operation_id: op_id,
mode: mode_str.into(),
started_at: started,
completed_at: Utc::now(),
source_type: opts.source_type.clone(),
source_endpoint: strip_credentials(&opts.source_endpoint),
source_repository: repository.to_string(),
destination_type: "znippy".into(),
requested_assets: requested,
successful_assets: records.len(),
failed_assets: failed,
bytes_transferred: bytes,
assets: records.clone(),
failures: failures.clone(),
final_archive_digest: Some(final_digest),
complete,
};
write_receipt(opts, &receipt)?;
Ok(TransferReport {
requested_assets: requested,
successful_assets: records.len(),
failed_assets: failed,
failures,
bytes_transferred: bytes,
complete,
})
}
fn write_receipt(opts: &TransferOptions, receipt: &TransferReceipt) -> Result<()> {
if let Some(path) = &opts.receipt_path {
let bytes = serde_json::to_vec_pretty(receipt)?;
std::fs::write(path, bytes)?;
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use async_trait::async_trait;
use std::collections::{HashMap, HashSet};
use traits::{RemoteAsset, RemoteRepository};
struct FakeConnector {
assets: Vec<RemoteAsset>,
data: HashMap<String, Vec<u8>>,
fail: HashSet<String>,
}
#[async_trait]
impl ConnectorTrait for FakeConnector {
fn name(&self) -> &str {
"fake"
}
async fn list_repositories(&self) -> Result<Vec<RemoteRepository>> {
Ok(vec![])
}
async fn list_assets(&self, _repo: &str) -> Result<Vec<RemoteAsset>> {
Ok(self.assets.clone())
}
async fn download_asset(&self, asset: &RemoteAsset) -> Result<Vec<u8>> {
if self.fail.contains(&asset.path) {
bail!("simulated download failure");
}
Ok(self.data.get(&asset.path).cloned().unwrap_or_default())
}
async fn upload_asset(&self, _r: &str, _p: &str, _d: &[u8]) -> Result<()> {
Ok(())
}
}
fn asset(path: &str) -> RemoteAsset {
RemoteAsset {
path: path.into(),
download_url: format!("https://user:secret@registry.example/{path}"),
content_type: None,
size: None,
sha256: None,
change_token: None,
}
}
fn two_asset_connector(fail: &[&str]) -> FakeConnector {
let mut data = HashMap::new();
data.insert("a.bin".to_string(), b"aaaa".to_vec());
data.insert("b.bin".to_string(), b"bbbbbb".to_vec());
FakeConnector {
assets: vec![asset("a.bin"), asset("b.bin")],
data,
fail: fail.iter().map(|s| s.to_string()).collect(),
}
}
#[tokio::test]
async fn strict_all_ok_writes_complete_receipt() {
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("intake.znippy");
let receipt = dir.path().join("receipt.json");
let conn = two_asset_connector(&[]);
let mut opts = TransferOptions::new(
TransferMode::Strict,
"nexus",
"https://user:pw@registry.example",
);
opts.receipt_path = Some(receipt.clone());
let report = transfer_repo_to_znippy_with_options(&conn, "repo", &out, &opts)
.await
.unwrap();
assert!(report.complete);
assert_eq!(report.successful_assets, 2);
assert!(out.exists());
let r: TransferReceipt = serde_json::from_slice(&std::fs::read(&receipt).unwrap()).unwrap();
assert_eq!(r.assets.len(), 2);
assert!(r.final_archive_digest.is_some());
assert_eq!(r.bytes_transferred, 10);
}
#[tokio::test]
async fn strict_fails_on_one_failed_download() {
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("intake.znippy");
let receipt = dir.path().join("receipt.json");
let conn = two_asset_connector(&["b.bin"]);
let mut opts =
TransferOptions::new(TransferMode::Strict, "nexus", "https://registry.example");
opts.receipt_path = Some(receipt.clone());
let res = transfer_repo_to_znippy_with_options(&conn, "repo", &out, &opts).await;
assert!(res.is_err(), "strict mode must fail on a failed download");
assert!(
!out.exists(),
"no archive under the final filename on failure"
);
// Receipt still written, marked incomplete.
let r: TransferReceipt = serde_json::from_slice(&std::fs::read(&receipt).unwrap()).unwrap();
assert!(!r.complete);
assert_eq!(r.failed_assets, 1);
}
#[tokio::test]
async fn strict_fails_on_hostile_path() {
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("intake.znippy");
let mut conn = two_asset_connector(&[]);
conn.assets.push(asset("../escape"));
let opts = TransferOptions::new(TransferMode::Strict, "nexus", "https://registry.example");
let res = transfer_repo_to_znippy_with_options(&conn, "repo", &out, &opts).await;
assert!(res.is_err());
assert!(!out.exists());
}
#[tokio::test]
async fn best_effort_tolerates_failure_and_still_packs() {
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("intake.znippy");
let conn = two_asset_connector(&["b.bin"]);
let opts = TransferOptions::new(
TransferMode::BestEffort,
"nexus",
"https://registry.example",
);
let report = transfer_repo_to_znippy_with_options(&conn, "repo", &out, &opts)
.await
.unwrap();
assert!(!report.complete);
assert_eq!(report.failed_assets, 1);
assert_eq!(report.successful_assets, 1);
assert!(
out.exists(),
"best-effort still produces an archive of the good assets"
);
}
#[tokio::test]
async fn receipt_strips_credentials_everywhere() {
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("intake.znippy");
let receipt = dir.path().join("receipt.json");
let conn = two_asset_connector(&[]);
let mut opts = TransferOptions::new(
TransferMode::Strict,
"nexus",
"https://user:secret@registry.example/base",
);
opts.receipt_path = Some(receipt.clone());
transfer_repo_to_znippy_with_options(&conn, "repo", &out, &opts)
.await
.unwrap();
let raw = std::fs::read_to_string(&receipt).unwrap();
assert!(
!raw.contains("secret"),
"no credential may appear in the receipt"
);
assert!(
!raw.contains("user:"),
"no userinfo may appear in the receipt"
);
}
#[test]
fn strip_credentials_removes_userinfo() {
assert_eq!(strip_credentials("https://u:p@h/x"), "https://h/x");
assert_eq!(strip_credentials("https://h/x"), "https://h/x");
assert_eq!(strip_credentials("https://h/x?a=b@c"), "https://h/x?a=b@c");
}
}