diff --git a/Cargo.lock b/Cargo.lock index 29ed496..30f4e18 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -316,6 +316,7 @@ dependencies = [ "flate2", "futures", "hex", + "libc", "once_cell", "opentelemetry", "opentelemetry-otlp", diff --git a/Cargo.toml b/Cargo.toml index 6c53195..ac0437e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -70,6 +70,9 @@ once_cell = "1.19" # Temporary layer storage while preparing Docker save archives tempfile = "=3.10.1" +# Atomic, no-replace directory publication on Linux and macOS. +libc = "0.2" + [profile.release] lto = true codegen-units = 1 diff --git a/README.md b/README.md index 67370d0..b804dce 100644 --- a/README.md +++ b/README.md @@ -43,6 +43,91 @@ Docker HTTP Passthrough turns a simple HTTP GET into a Docker save-format tar ar The passthrough service does not use the local Docker daemon to pull images. It talks directly to the registry API. +## Verified preparation command + +The separate `prepare` command acquires an image by SHA-256 digest and explicit +platform, verifies its content, and writes a retained Docker-save archive. It +does not start the HTTP server. Running without arguments preserves the existing +HTTP workflow described below. + +```bash +docker-http-passthrough prepare \ + --registry https://registry.example.com \ + --repository vendor/agent \ + --digest sha256:<64-lowercase-hex-digits> \ + --platform linux/amd64 \ + --output-dir ./prepared-agent +``` + +The destination must not exist and its parent must exist. On Linux and macOS, +the completed directory is published with an atomic no-replace rename. Errors +before publication remove temporary files and expose no completed destination. +A filesystem durability error after publication can leave the complete output +directory present; inspect its checksums and reuse it rather than overwriting it. +A hard process kill +can leave a hidden `.dhp-prepare-*` temporary directory; remove abandoned staging +directories only after confirming no preparation is using them. Existing output +directories are never overwritten, including on retries. + +The command verifies raw index/manifest/config/layer digests and descriptor +sizes, exact platform selection, config architecture, layer count and each +uncompressed diff ID. It supports single manifests and one-level indexes, with +plain or gzip layers. Nested indexes, ambiguous/missing platforms, foreign or +zstd layers, and tags as preparation inputs are rejected. A successful conversion +does not establish that the application or package is compatible with a device. + +Output files: + +| File | Meaning | +| --- | --- | +| `package.tar` | Deterministic Docker-save archive with a valid digest-derived internal tag | +| `descriptor.json` | Draft `apphosting.cisco.com/v1alpha1` descriptor; final package digest and byte count, source identity, producer binary digest and recipe digest | +| `provenance.json` | Unsigned in-toto statement binding the source, converter and verification results to the package | +| `recipe.json` | Exact conversion recipe bytes used by `producer.recipeDigest` | +| `checksums.json` | SHA-256 values of the package and metadata files, including final JSON newlines | + +The descriptor follows [the vendored draft schema](schemas/app-hosting-artifact-v1alpha1.schema.json). +The archive is deterministic for identical immutable inputs and converter build; +the descriptor's `createdAt` records this preparation. Reuse a successful output +directory when retrying publication. A new preparation is a new production record, +even if its package bytes match. `buildDigest` hashes the actual running executable, +not a claimed version string or a container image manifest. + +These files are **unsigned and unapproved**. The checksums are integrity metadata, +not independent proof of trust. For CVK Verified delivery, a separate pipeline +signs the exact descriptor bytes with an accepted producer identity (for example +using Cosign) and publishes unchanged package/metadata bytes. Supplier-signature +verification, vulnerability scanning and separately signing provenance are optional +pipeline policies, not requirements imposed by this command. Deployment approval, +revocation and vulnerability policy belong to a separate governed lifecycle. Device +compatibility, IOx package signing and CVK installation remain separate steps. Do +not present unsigned output as an authenticated or approved release. + +Preparation uses HTTPS only. `--registry` defaults to `REGISTRY_URL`, then Docker +Hub; credentials come from `REGISTRY_USERNAME`/`REGISTRY_PASSWORD` or +`REGISTRY_BEARER_TOKEN` (`REGISTRY_TOKEN` alias). Those credentials apply only to +the explicitly selected registry and its accepted token endpoint for this local +invocation. Cross-origin bearer challenges require an exact `--auth-url` (or +`REGISTRY_AUTH_URL`); Docker Hub's token endpoint is built in. Token scopes are +restricted to the requested repository's pull permission. Manifest and token +redirects fail closed. Blob redirects require `--blob-origin https://allowed-host` +for each additional origin and never receive registry credentials. Configure CDN +origins actually used by the trusted registry; they are not discovered or trusted +automatically. Endpoints cannot contain credentials, queries or fragments. + +`--max-bytes` defaults to 2 GiB, has a 64 GiB ceiling, and separately bounds compressed layer bytes, +expanded layer bytes and the final archive. Metadata is limited to 4 MiB, indexes +to 1,024 entries and images to 128 layers. Peak temporary storage can approach +three times the byte budget plus bounded metadata. `--timeout-seconds` defaults +to 1,800 for the whole command; individual network requests are capped at 120 +seconds with a 15-second connection timeout. The command prepares one image at a +time; operators must bound parallel CI jobs and available filesystem capacity. +The existing HTTP routes do not inherit these preparation guarantees or limits. + +For a private registry CA, `prepare --ca-file CA.pem` adds one PEM root to the preparation client's normal TLS trust. Certificate and hostname verification remain enabled; the HTTP server defaults are unchanged. + +Run `docker-http-passthrough prepare --help` for the command syntax. + ## Choose a Download Mode Pick the route based on where the registry details and credentials should come from. diff --git a/schemas/app-hosting-artifact-v1alpha1.schema.json b/schemas/app-hosting-artifact-v1alpha1.schema.json new file mode 100644 index 0000000..f256dee --- /dev/null +++ b/schemas/app-hosting-artifact-v1alpha1.schema.json @@ -0,0 +1,193 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "$id": "urn:cisco:app-hosting:artifact:v1alpha1", + "title": "Draft App Hosting Artifact Descriptor", + "description": "Proposed contract only. Sign and hash exact published UTF-8 bytes. Runtime semantics and security validation are specified in docs/app-image-lifecycle-plan.md.", + "type": "object", + "additionalProperties": false, + "required": [ + "apiVersion", + "kind", + "source", + "package", + "producer", + "createdAt", + "provenance" + ], + "properties": { + "apiVersion": { + "const": "apphosting.cisco.com/v1alpha1" + }, + "kind": { + "const": "AppHostingArtifact" + }, + "source": { + "type": "object", + "additionalProperties": false, + "required": [ + "registry", + "repository", + "requestedDigest", + "manifestDigest", + "configDigest", + "platform" + ], + "properties": { + "registry": { + "type": "string", + "minLength": 1, + "maxLength": 253 + }, + "repository": { + "type": "string", + "minLength": 1, + "maxLength": 255 + }, + "releaseLabel": { + "type": "string", + "minLength": 1, + "maxLength": 256 + }, + "requestedDigest": { + "type": "string", + "pattern": "^sha256:[a-f0-9]{64}$" + }, + "indexDigest": { + "type": "string", + "pattern": "^sha256:[a-f0-9]{64}$" + }, + "manifestDigest": { + "type": "string", + "pattern": "^sha256:[a-f0-9]{64}$" + }, + "configDigest": { + "type": "string", + "pattern": "^sha256:[a-f0-9]{64}$" + }, + "platform": { + "type": "object", + "additionalProperties": false, + "required": [ + "os", + "architecture" + ], + "properties": { + "os": { + "const": "linux" + }, + "architecture": { + "type": "string", + "pattern": "^[a-z0-9][a-z0-9_-]{0,31}$" + }, + "variant": { + "type": "string", + "pattern": "^[a-zA-Z0-9._-]{1,32}$" + } + } + } + } + }, + "package": { + "type": "object", + "additionalProperties": false, + "required": [ + "format", + "mediaType", + "digest", + "sizeBytes" + ], + "properties": { + "format": { + "enum": [ + "docker-save-v1", + "iox-package-v1" + ] + }, + "mediaType": { + "const": "application/x-tar" + }, + "digest": { + "type": "string", + "pattern": "^sha256:[a-f0-9]{64}$" + }, + "sizeBytes": { + "type": "integer", + "minimum": 1, + "maximum": 9007199254740991 + } + } + }, + "producer": { + "type": "object", + "additionalProperties": false, + "required": [ + "name", + "version", + "buildDigest", + "recipeDigest" + ], + "properties": { + "name": { + "type": "string", + "minLength": 1, + "maxLength": 256 + }, + "version": { + "type": "string", + "minLength": 1, + "maxLength": 256 + }, + "buildDigest": { + "type": "string", + "pattern": "^sha256:[a-f0-9]{64}$" + }, + "recipeDigest": { + "type": "string", + "pattern": "^sha256:[a-f0-9]{64}$" + } + } + }, + "createdAt": { + "type": "string", + "format": "date-time" + }, + "provenance": { + "type": "object", + "additionalProperties": false, + "required": [ + "digest", + "mediaType" + ], + "properties": { + "digest": { + "type": "string", + "pattern": "^sha256:[a-f0-9]{64}$" + }, + "mediaType": { + "type": "string", + "minLength": 1, + "maxLength": 256 + } + } + }, + "sbom": { + "type": "object", + "additionalProperties": false, + "required": [ + "digest", + "mediaType" + ], + "properties": { + "digest": { + "type": "string", + "pattern": "^sha256:[a-f0-9]{64}$" + }, + "mediaType": { + "type": "string", + "minLength": 1, + "maxLength": 256 + } + } + } + } +} diff --git a/src/main.rs b/src/main.rs index eee1d15..8388a87 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,6 +1,7 @@ // SPDX-License-Identifier: Apache-2.0 // Copyright 2026 Cisco Systems, Inc. +mod prepare; mod registry; mod tar_builder; mod telemetry; @@ -82,6 +83,10 @@ struct DownloadRouteContext { #[tokio::main] async fn main() -> Result<()> { + let args: Vec = std::env::args().skip(1).collect(); + if !args.is_empty() { + return prepare::run_cli(&args).await; + } // Initialize logging let subscriber = FmtSubscriber::builder() .with_env_filter( diff --git a/src/prepare.rs b/src/prepare.rs new file mode 100644 index 0000000..d18e602 --- /dev/null +++ b/src/prepare.rs @@ -0,0 +1,692 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright 2026 Cisco Systems, Inc. + +//! Local, unsigned artifact preparation. This module does not approve, sign, +//! publish remotely, scan, or install an application. + +use crate::registry::{ + verified::{decode_json, digest_bytes, validate_digest, validate_repository, VerifiedRegistry}, + Manifest, +}; +use anyhow::{anyhow, bail, ensure, Context, Result}; +use flate2::read::MultiGzDecoder; +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; +use sha2::{Digest, Sha256}; +use std::{ + collections::HashSet, + fs::{self, File}, + io::{Read, Write}, + path::{Path, PathBuf}, + time::{Duration, Instant, SystemTime, UNIX_EPOCH}, +}; +use tar::{Builder, Header}; + +const RECIPE: &[u8] = b"{\"format\":\"docker-save-v1\",\"revision\":1,\"repoTag\":\"sha256-\",\"tarMetadata\":\"uid=0,gid=0,mtime=0,mode=0644\",\"layers\":\"ordered-uncompressed-tar\"}\n"; +const MAX_LAYERS: usize = 128; +const HELP: &str = "Usage: docker-http-passthrough prepare --repository PATH --digest sha256:HEX --platform linux/ARCH[/VARIANT] --output-dir NEW_DIRECTORY [--registry https://HOST] [--auth-url https://HOST/PATH] [--ca-file CA.pem] [--blob-origin https://HOST] [--max-bytes N] [--timeout-seconds N]\n\nProduces package.tar, descriptor.json, provenance.json, recipe.json and checksums.json. Output is unsigned and unapproved. No arguments starts the unchanged HTTP server. Registry defaults to REGISTRY_URL or Docker Hub. Credentials use REGISTRY_USERNAME/REGISTRY_PASSWORD or REGISTRY_BEARER_TOKEN. Blob redirect origins must be explicitly allowlisted.\n"; + +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +pub struct Platform { + os: String, + architecture: String, + #[serde(skip_serializing_if = "Option::is_none")] + variant: Option, +} + +impl Platform { + fn parse(value: &str) -> Result { + let parts: Vec<_> = value.split('/').collect(); + ensure!( + (parts.len() == 2 || parts.len() == 3) && parts[0] == "linux", + "platform must be linux/architecture[/variant]" + ); + ensure!( + !parts[1].is_empty() + && parts[1].len() <= 32 + && parts[1].as_bytes()[0].is_ascii_alphanumeric() + && parts[1] + .bytes() + .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b"_-".contains(&b)), + "invalid platform architecture" + ); + let variant = parts.get(2).map(|s| s.to_string()); + if let Some(v) = &variant { + ensure!( + !v.is_empty() + && v.len() <= 32 + && v.bytes() + .all(|b| b.is_ascii_alphanumeric() || b"._-".contains(&b)), + "invalid platform variant" + ); + } + Ok(Self { + os: "linux".into(), + architecture: parts[1].into(), + variant, + }) + } + + fn matches_json(&self, value: &Value) -> bool { + value.get("os").and_then(Value::as_str) == Some(self.os.as_str()) + && value.get("architecture").and_then(Value::as_str) == Some(self.architecture.as_str()) + && value.get("variant").and_then(Value::as_str) == self.variant.as_deref() + } +} + +struct Options { + repository: String, + digest: String, + platform: Platform, + output: PathBuf, + registry: String, + auth_url: Option, + ca_file: Option, + blob_origins: Vec, + max_bytes: u64, + timeout: Duration, +} + +impl Options { + fn parse(args: &[String]) -> Result { + ensure!( + args.first().map(String::as_str) == Some("prepare"), + "unknown command; use --help" + ); + let mut values = std::collections::HashMap::new(); + let mut blob_origins = Vec::new(); + let mut args = args[1..].iter(); + while let Some(flag) = args.next() { + ensure!( + [ + "--repository", + "--digest", + "--platform", + "--output-dir", + "--registry", + "--auth-url", + "--ca-file", + "--blob-origin", + "--max-bytes", + "--timeout-seconds" + ] + .contains(&flag.as_str()), + "unknown preparation option" + ); + let value = args + .next() + .ok_or_else(|| anyhow!("missing value for {flag}"))?; + if flag == "--blob-origin" { + blob_origins.push(value.clone()); + } else { + ensure!( + values.insert(flag.as_str(), value.clone()).is_none(), + "duplicate option {flag}" + ); + } + } + let required = |key| { + values + .get(key) + .cloned() + .ok_or_else(|| anyhow!("missing {key}")) + }; + let repository = required("--repository")?; + validate_repository(&repository)?; + let digest = required("--digest")?; + validate_digest(&digest)?; + let platform = Platform::parse(&required("--platform")?)?; + let max_bytes = values + .get("--max-bytes") + .map(|s| s.parse::()) + .transpose()? + .unwrap_or(2 * 1024 * 1024 * 1024); + ensure!( + (1024..=1_u64 << 36).contains(&max_bytes), + "max-bytes must be between 1024 and 68719476736" + ); + let seconds = values + .get("--timeout-seconds") + .map(|s| s.parse::()) + .transpose()? + .unwrap_or(1800); + ensure!( + (1..=86400).contains(&seconds), + "timeout-seconds must be between 1 and 86400" + ); + Ok(Self { + repository, + digest, + platform, + output: required("--output-dir")?.into(), + registry: values + .remove("--registry") + .or_else(|| std::env::var("REGISTRY_URL").ok()) + .unwrap_or_else(|| "https://registry-1.docker.io".into()), + auth_url: values.remove("--auth-url").or_else(|| { + std::env::var("REGISTRY_AUTH_URL") + .ok() + .filter(|s| !s.is_empty()) + }), + ca_file: values.remove("--ca-file").map(PathBuf::from), + blob_origins, + max_bytes, + timeout: Duration::from_secs(seconds), + }) + } +} + +pub async fn run_cli(args: &[String]) -> Result<()> { + if args == ["--help"] || args == ["-h"] || args == ["prepare", "--help"] { + print!("{HELP}"); + return Ok(()); + } + // Load the same environment file as server mode, before reading options. + let _ = dotenvy::dotenv(); + let options = Options::parse(args)?; + let ca = options + .ca_file + .as_ref() + .map(|path| -> Result> { + let meta = std::fs::metadata(path).context("read registry CA file metadata")?; + ensure!( + meta.is_file() && meta.len() > 0 && meta.len() <= 1024 * 1024, + "registry CA file must be a regular file of at most 1 MiB" + ); + let mut raw = Vec::new(); + std::io::Read::take(std::fs::File::open(path)?, 1024 * 1024 + 1) + .read_to_end(&mut raw)?; + ensure!( + raw.len() <= 1024 * 1024, + "registry CA file exceeds size limit" + ); + Ok(raw) + }) + .transpose()?; + let registry = if let Some(ca) = ca.as_ref() { + VerifiedRegistry::new_with_ca( + &options.registry, + options.auth_url.as_deref(), + &options.blob_origins, + options.timeout.min(Duration::from_secs(120)), + ca, + ) + } else { + VerifiedRegistry::new( + &options.registry, + options.auth_url.as_deref(), + &options.blob_origins, + options.timeout.min(Duration::from_secs(120)), + ) + }? + .with_environment_credentials(); + let descriptor = tokio::time::timeout(options.timeout, prepare(&options, registry)) + .await + .context("preparation exceeded total timeout")??; + println!("{}", serde_json::to_string(&descriptor)?); + Ok(()) +} + +struct SelectedManifest { + bytes: Vec, + digest: String, + index: Option, +} + +async fn select_manifest( + registry: &VerifiedRegistry, + options: &Options, +) -> Result { + let bytes = registry + .metadata(&options.repository, &options.digest, None, true) + .await?; + let value: Value = decode_json(&bytes)?; + ensure!( + value.get("schemaVersion").and_then(Value::as_u64) == Some(2), + "unsupported manifest schema version" + ); + let media_type = value.get("mediaType").and_then(Value::as_str).unwrap_or(""); + if matches!( + media_type, + "application/vnd.oci.image.index.v1+json" + | "application/vnd.docker.distribution.manifest.list.v2+json" + ) { + let entries = value + .get("manifests") + .and_then(Value::as_array) + .ok_or_else(|| anyhow!("index has no manifest descriptors"))?; + ensure!(entries.len() <= 1024, "index has too many entries"); + let selected: Vec<_> = entries + .iter() + .filter(|e| { + e.get("platform") + .is_some_and(|p| options.platform.matches_json(p)) + }) + .collect(); + ensure!( + selected.len() == 1, + "index must contain exactly one matching platform" + ); + let selected = selected[0]; + ensure!( + is_manifest_type( + selected + .get("mediaType") + .and_then(Value::as_str) + .unwrap_or("") + ), + "nested indexes or unsupported child media types are not supported" + ); + let digest = selected + .get("digest") + .and_then(Value::as_str) + .ok_or_else(|| anyhow!("child manifest has no digest"))?; + let size = selected + .get("size") + .and_then(Value::as_u64) + .ok_or_else(|| anyhow!("child manifest has no size"))?; + let bytes = registry + .metadata(&options.repository, digest, Some(size), true) + .await?; + Ok(SelectedManifest { + bytes, + digest: digest.into(), + index: Some(options.digest.clone()), + }) + } else { + ensure!( + is_manifest_type(media_type), + "unsupported manifest media type" + ); + Ok(SelectedManifest { + bytes, + digest: options.digest.clone(), + index: None, + }) + } +} + +fn is_manifest_type(media_type: &str) -> bool { + matches!( + media_type, + "application/vnd.oci.image.manifest.v1+json" + | "application/vnd.docker.distribution.manifest.v2+json" + ) +} + +async fn prepare(options: &Options, registry: VerifiedRegistry) -> Result { + ensure!( + fs::symlink_metadata(&options.output) + .is_err_and(|e| e.kind() == std::io::ErrorKind::NotFound), + "output directory already exists or is inaccessible" + ); + let parent = options + .output + .parent() + .filter(|p| !p.as_os_str().is_empty()) + .unwrap_or_else(|| Path::new(".")); + ensure!( + parent.is_dir(), + "output parent directory must already exist" + ); + let staging = tempfile::Builder::new() + .prefix(".dhp-prepare-") + .tempdir_in(parent)?; + let deadline = Instant::now() + options.timeout; + let selected = select_manifest(®istry, options).await?; + let manifest: Manifest = decode_json(&selected.bytes)?; + ensure!( + manifest.schema_version == 2 + && manifest.manifests.is_none() + && is_manifest_type(manifest.media_type.as_deref().unwrap_or("")), + "unsupported selected manifest" + ); + let config = manifest + .config + .ok_or_else(|| anyhow!("manifest has no config"))?; + ensure!( + matches!( + config.media_type.as_str(), + "application/vnd.oci.image.config.v1+json" + | "application/vnd.docker.container.image.v1+json" + ), + "unsupported config media type" + ); + let layers = manifest + .layers + .ok_or_else(|| anyhow!("manifest has no layers"))?; + ensure!( + layers.len() <= MAX_LAYERS, + "image exceeds layer count limit" + ); + let config_bytes = registry + .metadata( + &options.repository, + &config.digest, + Some(config.size), + false, + ) + .await?; + let config_json: Value = decode_json(&config_bytes)?; + ensure!( + options.platform.matches_json(&config_json), + "image config platform does not match requested platform" + ); + ensure!( + config_json.pointer("/rootfs/type").and_then(Value::as_str) == Some("layers"), + "unsupported rootfs type" + ); + let diff_ids = config_json + .pointer("/rootfs/diff_ids") + .and_then(Value::as_array) + .ok_or_else(|| anyhow!("config has no rootfs diff IDs"))?; + ensure!( + diff_ids.len() == layers.len(), + "layer count differs from config diff IDs" + ); + let mut compressed_total = 0_u64; + for layer in &layers { + validate_digest(&layer.digest)?; + compressed_total = compressed_total + .checked_add(layer.size) + .ok_or_else(|| anyhow!("compressed size overflow"))?; + ensure!( + compressed_total <= options.max_bytes, + "compressed image exceeds byte budget" + ); + layer_encoding(&layer.media_type)?; + } + for diff_id in diff_ids { + validate_digest(diff_id.as_str().ok_or_else(|| anyhow!("invalid diff ID"))?)?; + } + + let package_path = staging.path().join("package.tar"); + let writer = LimitedWriter { + file: File::create(&package_path)?, + hash: Sha256::new(), + size: 0, + limit: options.max_bytes, + deadline, + }; + let mut archive = Builder::new(writer); + let config_name = format!("{}.json", &config.digest[7..]); + let repo_tag = format!("{}:sha256-{}", options.repository, &selected.digest[7..]); + let names: Vec<_> = layers + .iter() + .map(|l| format!("{}/layer.tar", &l.digest[7..])) + .collect(); + let archive_manifest = + json!([{ "Config": config_name, "RepoTags": [repo_tag], "Layers": names }]); + append_bytes( + &mut archive, + "manifest.json", + &serde_json::to_vec(&archive_manifest)?, + )?; + append_bytes(&mut archive, &config_name, &config_bytes)?; + let mut expanded_total = 0_u64; + let mut written = HashSet::new(); + for (layer, diff_id) in layers.iter().zip(diff_ids) { + ensure!(Instant::now() < deadline, "preparation deadline exceeded"); + let compressed = registry + .blob( + &options.repository, + &layer.digest, + layer.size, + staging.path(), + options.max_bytes, + ) + .await?; + let (expanded, size) = expand_layer( + compressed.path(), + layer_encoding(&layer.media_type)?, + diff_id.as_str().unwrap(), + staging.path(), + options.max_bytes - expanded_total, + deadline, + )?; + expanded_total += size; + // Repeated descriptors retain their positions in manifest.json but need + // only one tar entry. Every occurrence's expected diff ID is checked. + if written.insert(layer.digest.clone()) { + let base = &layer.digest[7..]; + append_bytes(&mut archive, &format!("{base}/VERSION"), b"1.0")?; + append_bytes( + &mut archive, + &format!("{base}/json"), + &serde_json::to_vec(&json!({"id": base, "created": "1970-01-01T00:00:00Z"}))?, + )?; + let mut header = tar_header(size); + archive.append_data( + &mut header, + format!("{base}/layer.tar"), + File::open(expanded.path())?, + )?; + } + } + archive.finish()?; + let writer = archive.into_inner()?; + writer.file.sync_all()?; + let package_digest = format!("sha256:{:x}", writer.hash.finalize()); + let package_size = writer.size; + let source = json!({ "registry": registry.authority(), "repository": options.repository, "requestedDigest": options.digest, "manifestDigest": selected.digest, "configDigest": config.digest, "platform": options.platform }); + let mut source = source; + if let Some(index) = selected.index { + source["indexDigest"] = index.into(); + } + let build_digest = hash_file(&std::env::current_exe()?)?; + let producer = json!({ "name": "docker-http-passthrough", "version": env!("CARGO_PKG_VERSION"), "buildDigest": build_digest, "recipeDigest": digest_bytes(RECIPE) }); + let provenance = json!({ + "_type": "https://in-toto.io/Statement/v1", + "subject": [{"name": "package.tar", "digest": {"sha256": &package_digest[7..]}}], + "predicateType": "urn:cisco:docker-http-passthrough:verified-preparation:v1", + "predicate": { "source": source, "producer": producer, "verification": { "compressedSizeBytes": compressed_total, "uncompressedSizeBytes": expanded_total, "manifestConfigLayersAndDiffIDs": true } } + }); + let provenance_bytes = json_bytes(&provenance)?; + let descriptor = json!({ + "apiVersion": "apphosting.cisco.com/v1alpha1", "kind": "AppHostingArtifact", + "source": source, + "package": {"format": "docker-save-v1", "mediaType": "application/x-tar", "digest": package_digest, "sizeBytes": package_size}, + "producer": producer, + "createdAt": utc_timestamp(SystemTime::now().duration_since(UNIX_EPOCH)?.as_secs()), + "provenance": {"digest": digest_bytes(&provenance_bytes), "mediaType": "application/vnd.in-toto+json"} + }); + let descriptor_bytes = json_bytes(&descriptor)?; + write_synced(staging.path().join("descriptor.json"), &descriptor_bytes)?; + write_synced(staging.path().join("provenance.json"), &provenance_bytes)?; + write_synced(staging.path().join("recipe.json"), RECIPE)?; + write_synced( + staging.path().join("checksums.json"), + &json_bytes( + &json!({"package.tar": descriptor["package"]["digest"], "descriptor.json": digest_bytes(&descriptor_bytes), "provenance.json": digest_bytes(&provenance_bytes), "recipe.json": digest_bytes(RECIPE)}), + )?, + )?; + File::open(staging.path())?.sync_all()?; + ensure!( + Instant::now() < deadline, + "preparation deadline exceeded before publication" + ); + publish_directory(staging.path(), &options.output)?; + File::open(parent)?.sync_all()?; + Ok(descriptor) +} + +fn layer_encoding(media_type: &str) -> Result { + match media_type { + "application/vnd.oci.image.layer.v1.tar" + | "application/vnd.docker.image.rootfs.diff.tar" => Ok(false), + "application/vnd.oci.image.layer.v1.tar+gzip" + | "application/vnd.docker.image.rootfs.diff.tar.gzip" => Ok(true), + _ => bail!("unsupported layer media type (only tar and gzip tar are supported)"), + } +} + +fn expand_layer( + path: &Path, + gzip: bool, + diff_id: &str, + directory: &Path, + budget: u64, + deadline: Instant, +) -> Result<(tempfile::NamedTempFile, u64)> { + let input = File::open(path)?; + let mut input: Box = if gzip { + Box::new(MultiGzDecoder::new(input)) + } else { + Box::new(input) + }; + let mut output = tempfile::NamedTempFile::new_in(directory)?; + let mut hash = Sha256::new(); + let mut size = 0_u64; + let mut buffer = [0_u8; 64 * 1024]; + loop { + ensure!( + Instant::now() < deadline, + "preparation deadline exceeded during decompression" + ); + let read = input + .read(&mut buffer) + .context("invalid or truncated compressed layer")?; + if read == 0 { + break; + } + size = size + .checked_add(read as u64) + .ok_or_else(|| anyhow!("expanded size overflow"))?; + ensure!(size <= budget, "expanded image exceeds byte budget"); + hash.update(&buffer[..read]); + output.write_all(&buffer[..read])?; + } + ensure!( + format!("sha256:{:x}", hash.finalize()) == diff_id, + "uncompressed layer diff ID mismatch" + ); + output.flush()?; + Ok((output, size)) +} + +struct LimitedWriter { + file: File, + hash: Sha256, + size: u64, + limit: u64, + deadline: Instant, +} +impl Write for LimitedWriter { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + if Instant::now() >= self.deadline { + return Err(std::io::Error::new( + std::io::ErrorKind::TimedOut, + "preparation deadline exceeded", + )); + } + if bytes.len() as u64 > self.limit.saturating_sub(self.size) { + return Err(std::io::Error::other("package exceeds byte budget")); + } + let size = self.file.write(bytes)?; + self.hash.update(&bytes[..size]); + self.size += size as u64; + Ok(size) + } + fn flush(&mut self) -> std::io::Result<()> { + self.file.flush() + } +} + +fn tar_header(size: u64) -> Header { + let mut header = Header::new_gnu(); + header.set_size(size); + header.set_mode(0o644); + header.set_uid(0); + header.set_gid(0); + header.set_mtime(0); + header.set_entry_type(tar::EntryType::Regular); + header.set_cksum(); + header +} +fn append_bytes(archive: &mut Builder, name: &str, bytes: &[u8]) -> Result<()> { + archive.append_data(&mut tar_header(bytes.len() as u64), name, bytes)?; + Ok(()) +} +fn json_bytes(value: &Value) -> Result> { + let mut bytes = serde_json::to_vec_pretty(value)?; + bytes.push(b'\n'); + Ok(bytes) +} +fn write_synced(path: PathBuf, bytes: &[u8]) -> Result<()> { + let mut file = File::create(path)?; + file.write_all(bytes)?; + file.sync_all()?; + Ok(()) +} +fn hash_file(path: &Path) -> Result { + let mut file = File::open(path)?; + let mut hash = Sha256::new(); + let mut buffer = [0; 64 * 1024]; + loop { + let n = file.read(&mut buffer)?; + if n == 0 { + break; + } + hash.update(&buffer[..n]); + } + Ok(format!("sha256:{:x}", hash.finalize())) +} + +fn publish_directory(source: &Path, destination: &Path) -> Result<()> { + #[cfg(any(target_os = "linux", target_os = "macos"))] + { + use std::os::unix::ffi::OsStrExt; + let source = std::ffi::CString::new(source.as_os_str().as_bytes())?; + let destination = std::ffi::CString::new(destination.as_os_str().as_bytes())?; + // Both APIs atomically publish the whole directory and refuse to replace + // any existing destination, including an empty directory or a symlink. + #[cfg(target_os = "linux")] + let result = unsafe { + libc::renameat2( + libc::AT_FDCWD, + source.as_ptr(), + libc::AT_FDCWD, + destination.as_ptr(), + libc::RENAME_NOREPLACE, + ) + }; + #[cfg(target_os = "macos")] + let result = + unsafe { libc::renamex_np(source.as_ptr(), destination.as_ptr(), libc::RENAME_EXCL) }; + if result != 0 { + return Err(std::io::Error::last_os_error()) + .context("atomic publication failed; output was not replaced"); + } + Ok(()) + } + #[cfg(not(any(target_os = "linux", target_os = "macos")))] + { + let _ = (source, destination); + bail!("atomic preparation publication is supported on Linux and macOS only") + } +} + +// Gregorian civil date conversion from Unix days; no shell command or new date +// runtime dependency. The descriptor's timestamp is historical, not approval. +fn utc_timestamp(seconds: u64) -> String { + let days = (seconds / 86400) as i64 + 719468; + let era = days / 146097; + let day_of_era = days - era * 146097; + let year_of_era = + (day_of_era - day_of_era / 1460 + day_of_era / 36524 - day_of_era / 146096) / 365; + let year = year_of_era + era * 400; + let day_of_year = day_of_era - (365 * year_of_era + year_of_era / 4 - year_of_era / 100); + let month_prime = (5 * day_of_year + 2) / 153; + let day = day_of_year - (153 * month_prime + 2) / 5 + 1; + let month = month_prime + if month_prime < 10 { 3 } else { -9 }; + let year = year + i64::from(month <= 2); + format!( + "{year:04}-{month:02}-{day:02}T{:02}:{:02}:{:02}Z", + seconds % 86400 / 3600, + seconds % 3600 / 60, + seconds % 60 + ) +} + +#[cfg(test)] +mod tests; diff --git a/src/prepare/tests.rs b/src/prepare/tests.rs new file mode 100644 index 0000000..bfbc4ce --- /dev/null +++ b/src/prepare/tests.rs @@ -0,0 +1,379 @@ +use super::*; +use axum::{ + extract::{Request, State}, + http::StatusCode, + response::IntoResponse, + Router, +}; +use std::{collections::HashMap, sync::Arc}; +use tokio::task::JoinHandle; + +#[derive(Clone, Copy)] +enum Fault { + None, + Manifest, + Config, + Layer, + LayerSize, + DiffId, + Platform, + Zstd, + IndexSize, + NestedIndex, + DuplicatePlatform, + DuplicateConfig, + DuplicateManifest, +} + +struct Fixture { + origin: String, + digest: String, + task: JoinHandle<()>, +} +impl Drop for Fixture { + fn drop(&mut self) { + self.task.abort(); + } +} + +impl Fixture { + async fn start(fault: Fault, index: bool, gzip: bool) -> Self { + let mut inner_tar = Builder::new(Vec::new()); + inner_tar + .append_data(&mut tar_header(6), "hello.txt", &b"hello\n"[..]) + .unwrap(); + let plain = inner_tar.into_inner().unwrap(); + let diff_id = if matches!(fault, Fault::DiffId) { + digest_bytes(b"wrong layer") + } else { + digest_bytes(&plain) + }; + let layer = if gzip { + let mut encoder = + flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default()); + encoder.write_all(&plain).unwrap(); + encoder.finish().unwrap() + } else { + plain + }; + let config = serde_json::to_vec(&json!({"os": "linux", "architecture": if matches!(fault, Fault::Platform) { "arm64" } else { "amd64" }, "rootfs": {"type": "layers", "diff_ids": [diff_id]}})).unwrap(); + let config = if matches!(fault, Fault::DuplicateConfig) { + [b"{\"architecture\":\"arm64\",".as_slice(), &config[1..]].concat() + } else { + config + }; + let config_digest = digest_bytes(&config); + let layer_digest = digest_bytes(&layer); + let layer_type = if matches!(fault, Fault::Zstd) { + "application/vnd.oci.image.layer.v1.tar+zstd" + } else if gzip { + "application/vnd.oci.image.layer.v1.tar+gzip" + } else { + "application/vnd.oci.image.layer.v1.tar" + }; + let manifest = serde_json::to_vec(&json!({"schemaVersion": 2, "mediaType": "application/vnd.oci.image.manifest.v1+json", "config": {"digest": config_digest, "size": config.len(), "mediaType": "application/vnd.oci.image.config.v1+json"}, "layers": [{"digest": layer_digest, "size": layer.len() + usize::from(matches!(fault, Fault::LayerSize)), "mediaType": layer_type}]})).unwrap(); + let manifest = if matches!(fault, Fault::DuplicateManifest) { + [b"{\"schemaVersion\":1,".as_slice(), &manifest[1..]].concat() + } else { + manifest + }; + let manifest_digest = digest_bytes(&manifest); + let mut objects = HashMap::new(); + objects.insert( + format!("/v2/test/agent/manifests/{manifest_digest}"), + if matches!(fault, Fault::Manifest) { + b"tampered".to_vec() + } else { + manifest.clone() + }, + ); + objects.insert( + format!("/v2/test/agent/blobs/{config_digest}"), + if matches!(fault, Fault::Config) { + let mut changed = config.clone(); + changed[0] ^= 1; + changed + } else { + config + }, + ); + objects.insert( + format!("/v2/test/agent/blobs/{layer_digest}"), + if matches!(fault, Fault::Layer) { + let mut changed = layer.clone(); + changed[0] ^= 1; + changed + } else { + layer + }, + ); + let digest = if index { + let entry = json!({"digest": manifest_digest, "size": manifest.len() + usize::from(matches!(fault, Fault::IndexSize)), "mediaType": if matches!(fault, Fault::NestedIndex) {"application/vnd.oci.image.index.v1+json"} else {"application/vnd.oci.image.manifest.v1+json"}, "platform": {"os": "linux", "architecture": "amd64"}}); + let entries = if matches!(fault, Fault::DuplicatePlatform) { + vec![entry.clone(), entry] + } else { + vec![entry] + }; + let index = serde_json::to_vec(&json!({"schemaVersion": 2, "mediaType": "application/vnd.oci.image.index.v1+json", "manifests": entries})).unwrap(); + let digest = digest_bytes(&index); + objects.insert(format!("/v2/test/agent/manifests/{digest}"), index); + digest + } else { + manifest_digest + }; + async fn serve( + State(objects): State>>>, + request: Request, + ) -> impl IntoResponse { + match objects.get(request.uri().path()) { + Some(bytes) => (StatusCode::OK, bytes.clone()), + None => (StatusCode::NOT_FOUND, Vec::new()), + } + } + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let origin = format!("http://{}", listener.local_addr().unwrap()); + let app = Router::new().fallback(serve).with_state(Arc::new(objects)); + let task = tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + Self { + origin, + digest, + task, + } + } + + fn options(&self, output: &Path) -> Options { + Options { + repository: "test/agent".into(), + digest: self.digest.clone(), + platform: Platform::parse("linux/amd64").unwrap(), + output: output.into(), + registry: self.origin.clone(), + auth_url: None, + ca_file: None, + blob_origins: vec![], + max_bytes: 1024 * 1024, + // Hashing the debug test executable can be slow on a busy CI host. + // Deadline behavior is exercised separately with an expired clock. + timeout: Duration::from_secs(60), + } + } + fn client(&self) -> VerifiedRegistry { + VerifiedRegistry::for_test(&self.origin).unwrap() + } +} + +#[tokio::test] +async fn verified_index_gzip_preparation_is_deterministic_and_descriptor_binds_bytes() { + let registry = Fixture::start(Fault::None, true, true).await; + let temp = tempfile::tempdir().unwrap(); + let output_a = temp.path().join("a"); + let output_b = temp.path().join("b"); + let a = prepare(®istry.options(&output_a), registry.client()) + .await + .unwrap(); + let b = prepare(®istry.options(&output_b), registry.client()) + .await + .unwrap(); + let bytes = fs::read(output_a.join("package.tar")).unwrap(); + assert_eq!(bytes, fs::read(output_b.join("package.tar")).unwrap()); + assert_eq!(a["package"]["digest"], digest_bytes(&bytes)); + assert_eq!(a["package"]["sizeBytes"], bytes.len()); + assert_eq!(a["package"], b["package"]); + assert_eq!(a["source"]["indexDigest"], registry.digest); + assert_eq!( + a["provenance"]["digest"], + digest_bytes(&fs::read(output_a.join("provenance.json")).unwrap()) + ); + assert_eq!( + a["producer"]["recipeDigest"], + digest_bytes(&fs::read(output_a.join("recipe.json")).unwrap()) + ); + assert!(a.get("approved").is_none()); + let checksums: Value = + serde_json::from_slice(&fs::read(output_a.join("checksums.json")).unwrap()).unwrap(); + assert_eq!( + checksums["descriptor.json"], + digest_bytes(&fs::read(output_a.join("descriptor.json")).unwrap()) + ); + let mut tar = tar::Archive::new(bytes.as_slice()); + let mut saw_manifest = false; + for entry in tar.entries().unwrap() { + let mut entry = entry.unwrap(); + assert_eq!(entry.header().mtime().unwrap(), 0); + assert_eq!(entry.header().uid().unwrap(), 0); + if entry.path().unwrap() == Path::new("manifest.json") { + let manifest: Value = serde_json::from_reader(&mut entry).unwrap(); + let repo_tag = manifest[0]["RepoTags"][0].as_str().unwrap(); + assert!(repo_tag.starts_with("test/agent:sha256-")); + assert_eq!(repo_tag.matches(':').count(), 1); + saw_manifest = true; + } + } + assert!(saw_manifest); +} + +#[tokio::test] +async fn corrupt_content_and_unsupported_sources_never_publish_or_leave_scratch() { + for (fault, indexed, message) in [ + (Fault::Manifest, false, "metadata digest mismatch"), + (Fault::Config, false, "metadata digest mismatch"), + (Fault::Layer, false, "layer digest mismatch"), + (Fault::LayerSize, false, "layer size differs"), + (Fault::DiffId, false, "diff ID mismatch"), + (Fault::Platform, false, "platform does not match"), + (Fault::Zstd, false, "unsupported layer media type"), + (Fault::IndexSize, true, "metadata size differs"), + (Fault::NestedIndex, true, "nested indexes"), + (Fault::DuplicateConfig, false, "ambiguous source JSON"), + (Fault::DuplicateManifest, false, "ambiguous source JSON"), + ( + Fault::DuplicatePlatform, + true, + "exactly one matching platform", + ), + ] { + let registry = Fixture::start(fault, indexed, true).await; + let temp = tempfile::tempdir().unwrap(); + let output = temp.path().join("artifact"); + let error = prepare(®istry.options(&output), registry.client()) + .await + .unwrap_err(); + assert!( + error.to_string().contains(message), + "expected {message}, got {error:#}" + ); + assert!(!output.exists()); + assert_eq!(fs::read_dir(temp.path()).unwrap().count(), 0); + } +} + +#[tokio::test] +async fn direct_manifest_and_plain_layer_work_and_existing_outputs_are_untouched() { + let registry = Fixture::start(Fault::None, false, false).await; + let temp = tempfile::tempdir().unwrap(); + let output = temp.path().join("artifact"); + let descriptor = prepare(®istry.options(&output), registry.client()) + .await + .unwrap(); + assert!(descriptor["source"].get("indexDigest").is_none()); + assert_eq!( + descriptor["source"]["requestedDigest"], + descriptor["source"]["manifestDigest"] + ); + let before = fs::read(output.join("descriptor.json")).unwrap(); + assert!(prepare(®istry.options(&output), registry.client()) + .await + .is_err()); + assert_eq!(before, fs::read(output.join("descriptor.json")).unwrap()); +} + +#[tokio::test] +async fn package_budget_and_missing_index_platform_fail_closed() { + let registry = Fixture::start(Fault::None, true, true).await; + let temp = tempfile::tempdir().unwrap(); + let output = temp.path().join("artifact"); + let mut options = registry.options(&output); + options.max_bytes = 1024; + assert!(prepare(&options, registry.client()).await.is_err()); + assert!(!output.exists()); + options.max_bytes = 1024 * 1024; + options.platform = Platform::parse("linux/arm64").unwrap(); + assert!(prepare(&options, registry.client()) + .await + .unwrap_err() + .to_string() + .contains("exactly one")); +} + +#[test] +fn expansion_budget_diff_id_and_deadline_are_enforced() { + let temp = tempfile::tempdir().unwrap(); + let mut encoder = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::best()); + let expanded = vec![0; 1024 * 1024]; + encoder.write_all(&expanded).unwrap(); + let path = temp.path().join("layer.gz"); + fs::write(&path, encoder.finish().unwrap()).unwrap(); + let deadline = Instant::now() + Duration::from_secs(10); + assert!(expand_layer( + &path, + true, + &digest_bytes(&expanded), + temp.path(), + 1024, + deadline + ) + .unwrap_err() + .to_string() + .contains("byte budget")); + assert!(expand_layer( + &path, + true, + &digest_bytes(b"wrong"), + temp.path(), + expanded.len() as u64, + deadline + ) + .unwrap_err() + .to_string() + .contains("diff ID mismatch")); + assert!(expand_layer( + &path, + true, + &digest_bytes(&expanded), + temp.path(), + expanded.len() as u64, + Instant::now() + ) + .is_err()); +} + +#[test] +fn atomic_publish_refuses_existing_directory_and_symlink() { + let temp = tempfile::tempdir().unwrap(); + let source = temp.path().join("source"); + fs::create_dir(&source).unwrap(); + fs::write(source.join("data"), b"new").unwrap(); + let destination = temp.path().join("destination"); + fs::create_dir(&destination).unwrap(); + assert!(publish_directory(&source, &destination).is_err()); + assert!(source.join("data").exists()); + assert_eq!(fs::read_dir(&destination).unwrap().count(), 0); + #[cfg(unix)] + { + let link = temp.path().join("link"); + std::os::unix::fs::symlink(&destination, &link).unwrap(); + assert!(publish_directory(&source, &link).is_err()); + } +} + +#[test] +fn command_inputs_and_timestamp_are_strict() { + for invalid in [ + "agent:latest", + "agent@sha256:abc", + "../agent", + "https://host/image", + "Agent", + "agent?x=1", + "a..b", + "a._b", + "a___b", + ] { + assert!(validate_repository(invalid).is_err()); + } + assert!(validate_repository("team/agent_name.v2").is_ok()); + assert!(validate_repository("team__name/agent--name.v2").is_ok()); + assert!(validate_digest("sha256:abc").is_err()); + assert!(validate_digest(&digest_bytes(b"test")).is_ok()); + assert!(Platform::parse("amd64").is_err()); + assert!(Platform::parse("windows/amd64").is_err()); + assert!(Platform::parse("linux/_amd64").is_err()); + assert!(Platform::parse("linux/-amd64").is_err()); + assert_eq!(utc_timestamp(0), "1970-01-01T00:00:00Z"); + assert_eq!(utc_timestamp(1709251199), "2024-02-29T23:59:59Z"); + assert!( + Options::parse(&["prepare".into(), "--repository".into(), "test/agent".into()]).is_err() + ); +} diff --git a/src/registry.rs b/src/registry.rs index 2b275cd..a0162f4 100644 --- a/src/registry.rs +++ b/src/registry.rs @@ -20,6 +20,8 @@ use std::env; use std::pin::Pin; use tracing::{debug, info}; +pub mod verified; + /// Base URL for the container registry V2 API. /// Default: Docker Hub (registry-1.docker.io) static REGISTRY_URL: Lazy = Lazy::new(|| { diff --git a/src/registry/verified.rs b/src/registry/verified.rs new file mode 100644 index 0000000..e34fa3e --- /dev/null +++ b/src/registry/verified.rs @@ -0,0 +1,757 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright 2026 Cisco Systems, Inc. + +//! Digest-verified acquisition for the prepare command. The legacy HTTP routes +//! intentionally keep their existing client and authentication behavior. + +use super::{parse_authenticate_header, AuthResponse, RegistryCredentials}; +use anyhow::{anyhow, bail, ensure, Context, Result}; +use futures::StreamExt; +use reqwest::{header, Client, Response, Url}; +use sha2::{Digest, Sha256}; +use std::{io::Write, path::Path, time::Duration}; +use tempfile::NamedTempFile; + +const ACCEPT: &str = "application/vnd.oci.image.index.v1+json, application/vnd.oci.image.manifest.v1+json, application/vnd.docker.distribution.manifest.list.v2+json, application/vnd.docker.distribution.manifest.v2+json"; +pub const METADATA_LIMIT: u64 = 4 * 1024 * 1024; + +pub fn validate_digest(value: &str) -> Result<()> { + let hex = value.strip_prefix("sha256:").unwrap_or_default(); + ensure!( + hex.len() == 64 + && hex + .bytes() + .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b)), + "expected lowercase sha256:<64 hex digits>" + ); + Ok(()) +} + +pub fn digest_bytes(bytes: &[u8]) -> String { + format!("sha256:{:x}", Sha256::digest(bytes)) +} + +pub fn validate_repository(value: &str) -> Result<()> { + ensure!( + !value.is_empty() && value.len() <= 255, + "invalid repository length" + ); + for component in value.split('/') { + let bytes = component.as_bytes(); + let alphanumeric = |b: u8| b.is_ascii_lowercase() || b.is_ascii_digit(); + let mut pos = 0; + loop { + let start = pos; + while pos < bytes.len() && alphanumeric(bytes[pos]) { + pos += 1; + } + ensure!(pos > start, "invalid canonical repository component"); + if pos == bytes.len() { + break; + } + match bytes[pos] { + b'.' => pos += 1, + b'_' => { pos += 1; if pos < bytes.len() && bytes[pos] == b'_' { pos += 1; } }, + b'-' => { while pos < bytes.len() && bytes[pos] == b'-' { pos += 1; } }, + _ => bail!("repository must be a lowercase canonical path without a tag, digest, URL or credentials"), + } + } + } + Ok(()) +} + +fn endpoint(value: &str, allow_test_http: bool) -> Result { + let url = Url::parse(value).map_err(|_| anyhow!("invalid registry/auth endpoint"))?; + ensure!( + url.scheme() == "https" + || (allow_test_http + && url.scheme() == "http" + && matches!(url.host_str(), Some("127.0.0.1") | Some("localhost"))), + "preparation requires HTTPS endpoints" + ); + ensure!( + url.host_str().is_some() + && url.username().is_empty() + && url.password().is_none() + && url.query().is_none() + && url.fragment().is_none(), + "endpoint must not contain credentials, query or fragment" + ); + Ok(url) +} + +pub struct VerifiedRegistry { + client: Client, + origin: Url, + token_endpoint: Option, + blob_origins: Vec, + credentials: Option, + bearer: Option, + allow_test_http: bool, +} + +impl VerifiedRegistry { + pub fn new( + registry: &str, + token_endpoint: Option<&str>, + blob_origins: &[String], + timeout: Duration, + ) -> Result { + Self::build(registry, token_endpoint, blob_origins, timeout, false, None) + } + + pub fn new_with_ca( + registry: &str, + token_endpoint: Option<&str>, + blob_origins: &[String], + timeout: Duration, + ca_pem: &[u8], + ) -> Result { + Self::build( + registry, + token_endpoint, + blob_origins, + timeout, + false, + Some(ca_pem), + ) + } + + fn build( + registry: &str, + token_endpoint: Option<&str>, + blob_origins: &[String], + timeout: Duration, + allow_test_http: bool, + ca_pem: Option<&[u8]>, + ) -> Result { + let registry = match registry { + "https://docker.io" | "https://index.docker.io" => "https://registry-1.docker.io", + other => other, + }; + let origin = endpoint(registry, allow_test_http)?; + ensure!( + origin.path() == "/", + "registry must be an origin without a path" + ); + ensure!( + origin.origin().ascii_serialization().len() - origin.scheme().len() - 3 <= 253, + "registry authority exceeds descriptor length limit" + ); + let default_token = (origin.as_str() == "https://registry-1.docker.io/") + .then_some("https://auth.docker.io/token"); + let token_endpoint = token_endpoint + .or(default_token) + .map(|s| endpoint(s, allow_test_http)) + .transpose()?; + let blob_origins: Vec<_> = blob_origins + .iter() + .map(|s| endpoint(s, allow_test_http)) + .collect::>()?; + ensure!( + blob_origins.iter().all(|u| u.path() == "/"), + "blob redirect allowlist entries must be origins" + ); + let mut client = Client::builder() + .redirect(reqwest::redirect::Policy::none()) + .connect_timeout(Duration::from_secs(15)) + .timeout(timeout); + if let Some(pem) = ca_pem { + ensure!( + !pem.is_empty() && pem.len() <= 1024 * 1024, + "registry CA file must contain at most 1 MiB" + ); + let certificates = reqwest::Certificate::from_pem_bundle(pem) + .context("invalid registry CA certificate")?; + ensure!( + certificates.len() == 1, + "registry CA file must contain exactly one PEM certificate" + ); + client = client.add_root_certificate(certificates.into_iter().next().unwrap()); + } + Ok(Self { + client: client.build()?, + origin, + token_endpoint, + blob_origins, + credentials: None, + bearer: None, + allow_test_http, + }) + } + + pub fn with_environment_credentials(mut self) -> Self { + self.credentials = super::env_value("REGISTRY_USERNAME").map(|username| { + RegistryCredentials::new(username, super::env_value("REGISTRY_PASSWORD")) + }); + self.bearer = super::env_value("REGISTRY_BEARER_TOKEN") + .or_else(|| super::env_value("REGISTRY_TOKEN")); + self + } + + pub fn authority(&self) -> String { + self.origin + .origin() + .ascii_serialization() + .split_once("://") + .unwrap() + .1 + .to_string() + } + + async fn request(&self, repository: &str, kind: &str, digest: &str) -> Result { + validate_repository(repository)?; + validate_digest(digest)?; + let url = self + .origin + .join(&format!("v2/{repository}/{kind}/{digest}"))?; + let mut request = self.client.get(url.clone()).header(header::ACCEPT, ACCEPT); + if let Some(token) = &self.bearer { + request = request.bearer_auth(token); + } else if let Some(credentials) = &self.credentials { + request = request.basic_auth(&credentials.username, credentials.password.as_deref()); + } + let mut response = request + .send() + .await + .map_err(|_| anyhow!("registry request failed or timed out"))?; + if response.status() == reqwest::StatusCode::UNAUTHORIZED { + let challenge = response + .headers() + .get(header::WWW_AUTHENTICATE) + .and_then(|h| h.to_str().ok()) + .and_then(parse_authenticate_header) + .ok_or_else(|| { + anyhow!("registry authentication challenge missing or unsupported") + })?; + ensure!( + challenge.scheme == "bearer", + "registry rejected configured credentials" + ); + let realm = endpoint( + challenge + .realm + .as_deref() + .ok_or_else(|| anyhow!("token challenge has no realm"))?, + self.allow_test_http, + )?; + match &self.token_endpoint { + Some(allowed) => ensure!( + realm == *allowed, + "registry requested an unapproved token endpoint" + ), + None => ensure!( + realm.origin() == self.origin.origin(), + "cross-origin token endpoint requires explicit --auth-url" + ), + } + let mut token_url = realm; + { + let mut query = token_url.query_pairs_mut(); + if let Some(service) = challenge.service { + query.append_pair("service", &service); + } + // Never accept broader registry-supplied scopes. + query.append_pair("scope", &format!("repository:{repository}:pull")); + } + let mut token_request = self.client.get(token_url); + if let Some(credentials) = &self.credentials { + token_request = token_request + .basic_auth(&credentials.username, credentials.password.as_deref()); + } + let response_token = token_request + .send() + .await + .map_err(|_| anyhow!("token request failed or timed out"))?; + ensure!( + response_token.status().is_success(), + "token endpoint rejected request (status {})", + response_token.status() + ); + let token: AuthResponse = + serde_json::from_slice(&read_bounded(response_token, 64 * 1024).await?) + .context("invalid token response")?; + response = self + .client + .get(url) + .header(header::ACCEPT, ACCEPT) + .bearer_auth(token.into_token()?) + .send() + .await + .map_err(|_| anyhow!("authenticated registry request failed or timed out"))?; + } + for _ in 0..3 { + if !response.status().is_redirection() { + break; + } + ensure!(kind == "blobs", "manifest redirects are not supported"); + let location = response + .headers() + .get(header::LOCATION) + .and_then(|h| h.to_str().ok()) + .ok_or_else(|| anyhow!("blob redirect has no location"))?; + let target = response + .url() + .join(location) + .map_err(|_| anyhow!("invalid blob redirect"))?; + ensure!( + target.username().is_empty() + && target.password().is_none() + && target.fragment().is_none(), + "invalid blob redirect authority" + ); + ensure!( + target.scheme() == "https" + || (self.allow_test_http && target.origin() == self.origin.origin()), + "blob redirect would downgrade HTTPS" + ); + ensure!( + target.origin() == self.origin.origin() + || self + .blob_origins + .iter() + .any(|u| u.origin() == target.origin()), + "blob redirect origin is not allowlisted; configure --blob-origin" + ); + // Signed CDN query strings are allowed, but never logged. No registry + // credentials are attached to any redirect request. + response = self + .client + .get(target) + .send() + .await + .map_err(|_| anyhow!("blob redirect request failed or timed out"))?; + } + ensure!( + response.status().is_success(), + "registry request failed (status {})", + response.status() + ); + Ok(response) + } + + pub async fn metadata( + &self, + repository: &str, + digest: &str, + expected_size: Option, + manifest: bool, + ) -> Result> { + if let Some(size) = expected_size { + ensure!(size <= METADATA_LIMIT, "metadata exceeds configured limit"); + } + let response = self + .request( + repository, + if manifest { "manifests" } else { "blobs" }, + digest, + ) + .await?; + let bytes = read_bounded(response, expected_size.unwrap_or(METADATA_LIMIT)).await?; + verify_bytes(&bytes, digest, expected_size)?; + Ok(bytes) + } + + pub async fn blob( + &self, + repository: &str, + digest: &str, + size: u64, + directory: &Path, + limit: u64, + ) -> Result { + ensure!( + size <= limit, + "compressed layer exceeds remaining byte budget" + ); + let mut response = self + .request(repository, "blobs", digest) + .await? + .bytes_stream(); + let mut output = NamedTempFile::new_in(directory)?; + let mut hash = Sha256::new(); + let mut received = 0_u64; + while let Some(chunk) = response.next().await { + let chunk = chunk.map_err(|_| anyhow!("layer download failed or timed out"))?; + received = received + .checked_add(chunk.len() as u64) + .ok_or_else(|| anyhow!("layer size overflow"))?; + ensure!(received <= size, "layer exceeds descriptor size"); + hash.update(&chunk); + output.write_all(&chunk)?; + } + ensure!(received == size, "layer size differs from descriptor"); + ensure!( + format!("sha256:{:x}", hash.finalize()) == digest, + "layer digest mismatch" + ); + output.flush()?; + Ok(output) + } + + #[cfg(test)] + pub fn for_test(registry: &str) -> Result { + Self::build(registry, None, &[], Duration::from_secs(2), true, None) + } +} + +async fn read_bounded(response: Response, limit: u64) -> Result> { + if response.content_length().is_some_and(|size| size > limit) { + bail!("metadata response exceeds byte limit"); + } + let mut bytes = Vec::new(); + let mut stream = response.bytes_stream(); + while let Some(chunk) = stream.next().await { + let chunk = chunk.map_err(|_| anyhow!("metadata download failed or timed out"))?; + ensure!( + bytes.len() as u64 + chunk.len() as u64 <= limit, + "metadata response exceeds byte limit" + ); + bytes.extend_from_slice(&chunk); + } + Ok(bytes) +} + +pub fn verify_bytes(bytes: &[u8], digest: &str, size: Option) -> Result<()> { + validate_digest(digest)?; + ensure!( + size.is_none_or(|size| size == bytes.len() as u64), + "metadata size differs from descriptor" + ); + ensure!(digest_bytes(bytes) == digest, "metadata digest mismatch"); + Ok(()) +} + +/// Reject ambiguous JSON before model decoding: the raw config will also be +/// parsed by the downstream container runtime. +pub fn decode_json(bytes: &[u8]) -> Result { + struct UniqueKeys; + impl<'de> serde::Deserialize<'de> for UniqueKeys { + fn deserialize>( + deserializer: D, + ) -> std::result::Result { + struct Visitor; + impl<'de> serde::de::Visitor<'de> for Visitor { + type Value = UniqueKeys; + fn expecting(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("JSON without duplicate object keys") + } + fn visit_map>( + self, + mut map: A, + ) -> std::result::Result { + let mut keys = std::collections::HashSet::new(); + while let Some(key) = map.next_key::()? { + if !keys.insert(key) { + return Err(serde::de::Error::custom("duplicate JSON object key")); + } + map.next_value::()?; + } + Ok(UniqueKeys) + } + fn visit_seq>( + self, + mut seq: A, + ) -> std::result::Result { + while seq.next_element::()?.is_some() {} + Ok(UniqueKeys) + } + fn visit_bool( + self, + _: bool, + ) -> std::result::Result { + Ok(UniqueKeys) + } + fn visit_i64( + self, + _: i64, + ) -> std::result::Result { + Ok(UniqueKeys) + } + fn visit_u64( + self, + _: u64, + ) -> std::result::Result { + Ok(UniqueKeys) + } + fn visit_f64( + self, + _: f64, + ) -> std::result::Result { + Ok(UniqueKeys) + } + fn visit_str( + self, + _: &str, + ) -> std::result::Result { + Ok(UniqueKeys) + } + fn visit_unit(self) -> std::result::Result { + Ok(UniqueKeys) + } + } + deserializer.deserialize_any(Visitor) + } + } + serde_json::from_slice::(bytes).context("invalid or ambiguous source JSON")?; + Ok(serde_json::from_slice(bytes)?) +} + +#[cfg(test)] +mod tests { + use super::*; + use axum::{ + extract::Request, + http::StatusCode, + response::{IntoResponse, Response as AxumResponse}, + Router, + }; + use std::sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, Mutex, + }; + + struct Server { + origin: String, + task: tokio::task::JoinHandle<()>, + } + impl Drop for Server { + fn drop(&mut self) { + self.task.abort(); + } + } + async fn server(handler: F) -> Server + where + F: Fn(Request) -> AxumResponse + Clone + Send + Sync + 'static, + { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let origin = format!("http://{}", listener.local_addr().unwrap()); + let app = Router::new().fallback(move |r: Request| { + let response = handler(r); + async move { response } + }); + let task = tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + Server { origin, task } + } + + #[test] + fn verification_rejects_same_size_tampering_and_unsupported_digests() { + let digest = digest_bytes(b"abcd"); + assert!(verify_bytes(b"abcd", &digest, Some(4)).is_ok()); + assert!(verify_bytes(b"abce", &digest, Some(4)) + .unwrap_err() + .to_string() + .contains("digest mismatch")); + assert!(verify_bytes(b"abcd", &digest, Some(5)).is_err()); + assert!(verify_bytes(b"abcd", "sha512:abcd", None).is_err()); + } + + #[test] + fn endpoint_validation_rejects_plaintext_userinfo_and_query() { + for value in [ + "http://registry.example.com", + "https://user:password@registry.example.com", + "https://registry.example.com?token=secret", + "https://registry.example.com/path", + "https://registry.example.com#fragment", + ] { + assert!(VerifiedRegistry::new(value, None, &[], Duration::from_secs(1)).is_err()); + } + assert!(VerifiedRegistry::new( + "https://registry.example.com", + None, + &[], + Duration::from_secs(1) + ) + .is_ok()); + } + + #[tokio::test] + async fn unapproved_token_realm_is_never_contacted() { + let contacted = Arc::new(AtomicUsize::new(0)); + let hits = Arc::clone(&contacted); + let token_server = server(move |_| { + hits.fetch_add(1, Ordering::SeqCst); + "unexpected".into_response() + }) + .await; + let realm = format!("{}/token", token_server.origin); + let registry = server(move |_| { + ( + StatusCode::UNAUTHORIZED, + [("www-authenticate", format!("Bearer realm=\"{realm}\""))], + "", + ) + .into_response() + }) + .await; + let mut client = VerifiedRegistry::for_test(®istry.origin).unwrap(); + client.credentials = Some(RegistryCredentials::new( + "user".into(), + Some("secret".into()), + )); + let error = client + .metadata("test/agent", &digest_bytes(b"data"), None, true) + .await + .unwrap_err(); + assert!(error.to_string().contains("cross-origin token endpoint")); + assert_eq!(contacted.load(Ordering::SeqCst), 0); + } + + #[tokio::test] + async fn allowed_token_endpoint_gets_only_requested_pull_scope() { + let token_requests = Arc::new(Mutex::new(Vec::new())); + let records = Arc::clone(&token_requests); + let token_server = server(move |request| { + records.lock().unwrap().push(( + request.uri().to_string(), + request + .headers() + .get("authorization") + .unwrap() + .to_str() + .unwrap() + .to_string(), + )); + "{\"token\":\"issued-token\"}".into_response() + }) + .await; + let realm = format!("{}/token", token_server.origin); + let challenge_realm = realm.clone(); + let registry = server(move |request| { + if request.headers().get("authorization").is_some_and(|v| v == "Bearer issued-token") { return "data".into_response(); } + (StatusCode::UNAUTHORIZED, [("www-authenticate", format!("Bearer realm=\"{challenge_realm}\",service=\"test\",scope=\"repository:other/private:pull,push\""))], "").into_response() + }).await; + let mut client = VerifiedRegistry::build( + ®istry.origin, + Some(&realm), + &[], + Duration::from_secs(2), + true, + None, + ) + .unwrap(); + client.credentials = Some(RegistryCredentials::new( + "user".into(), + Some("secret".into()), + )); + assert_eq!( + client + .metadata("test/agent", &digest_bytes(b"data"), Some(4), true) + .await + .unwrap(), + b"data" + ); + let records = token_requests.lock().unwrap(); + assert_eq!(records.len(), 1); + let query = Url::parse(&format!("http://test{}", records[0].0)).unwrap(); + assert_eq!( + query + .query_pairs() + .find(|(key, _)| key == "scope") + .unwrap() + .1, + "repository:test/agent:pull" + ); + assert!(records[0].1.starts_with("Basic ")); + } + + #[tokio::test] + async fn disallowed_blob_redirect_is_never_contacted() { + let contacted = Arc::new(AtomicUsize::new(0)); + let hits = Arc::clone(&contacted); + let blob_server = server(move |_| { + hits.fetch_add(1, Ordering::SeqCst); + "data".into_response() + }) + .await; + let target = format!("{}/blob?signature=secret", blob_server.origin); + let registry = server(move |_| { + ( + StatusCode::TEMPORARY_REDIRECT, + [("location", target.clone())], + "", + ) + .into_response() + }) + .await; + let client = VerifiedRegistry::for_test(®istry.origin).unwrap(); + assert!(client + .metadata("test/agent", &digest_bytes(b"data"), Some(4), false) + .await + .is_err()); + assert_eq!(contacted.load(Ordering::SeqCst), 0); + } + + #[tokio::test] + async fn same_origin_blob_redirect_strips_registry_credentials() { + let observed = Arc::new(Mutex::new(Vec::new())); + let records = Arc::clone(&observed); + let registry = server(move |request| { + records.lock().unwrap().push(( + request.uri().path().to_string(), + request + .headers() + .get("authorization") + .map(|v| v.to_str().unwrap().to_string()), + )); + if request.uri().path() == "/download" { + return "data".into_response(); + } + ( + StatusCode::TEMPORARY_REDIRECT, + [("location", "/download?signature=secret")], + "", + ) + .into_response() + }) + .await; + let mut client = VerifiedRegistry::for_test(®istry.origin).unwrap(); + client.credentials = Some(RegistryCredentials::new( + "user".into(), + Some("secret".into()), + )); + assert_eq!( + client + .metadata("test/agent", &digest_bytes(b"data"), Some(4), false) + .await + .unwrap(), + b"data" + ); + let records = observed.lock().unwrap(); + assert_eq!(records.len(), 2); + assert!(records[0].1.as_deref().unwrap().starts_with("Basic ")); + assert!(records[1].1.is_none()); + } +} + +#[cfg(test)] +#[test] +fn custom_registry_ca_rejects_invalid_material_and_plaintext_origin() { + assert!(VerifiedRegistry::new_with_ca( + "https://registry.example.test", + None, + &[], + Duration::from_secs(2), + b"not a certificate" + ) + .is_err()); + assert!(VerifiedRegistry::new_with_ca( + "https://registry.example.test", + None, + &[], + Duration::from_secs(2), + b"" + ) + .is_err()); + assert!(VerifiedRegistry::new_with_ca( + "http://127.0.0.1", + None, + &[], + Duration::from_secs(2), + b"not a certificate" + ) + .is_err()); +}