server/repository/file-oci/src/lib.rs
//! Disk-backed, writable OCI registry (Distribution Spec v2).
//!
//! This is holger's push target for the modern Helm flow
//! (`helm push chart.tgz oci://host/{repo}` and `helm pull oci://…`) as well as
//! any OCI artifact. Unlike the bespoke ChartMuseum HTTP API it replaces, the
//! OCI protocol is standardized, content-addressed (sha256, integrity built in),
//! and treats every blob as opaque bytes — so the registry never decompresses or
//! parses an uploaded chart, removing that attack surface entirely.
//!
//! Routes (all under the endpoint's repository name; the gateway routes
//! `/v2/{repo}/…` to this backend by the first name segment):
//!
//! GET /v2/ version probe
//! GET|HEAD /v2/{name}/manifests/{ref} manifest by tag or digest
//! GET|HEAD /v2/{name}/blobs/{digest} blob by digest
//! GET /v2/{name}/tags/list tag listing
//! POST /v2/{name}/blobs/uploads/[?digest=] begin (or monolithic) upload
//! PATCH /v2/{name}/blobs/uploads/{uuid} append a chunk
//! PUT /v2/{name}/blobs/uploads/{uuid}?digest=… finalize a blob
//! PUT /v2/{name}/manifests/{ref} store a manifest
//! GET|HEAD /v2/{name}/referrers/{digest}[?artifactType=] OCI 1.1 referrers
//!
//! **Referrers** (OCI Distribution 1.1, lane HOL-REFERRERS): a manifest that names a
//! `subject` (a cosign bundle or notation signature, an SBOM `oras attach`ed to an
//! image) is kept like any other, its push answers `OCI-Subject`, and its descriptor
//! is filed under the subject, so `GET …/referrers/<digest>` lists it. The subject
//! need not be in the registry. The tag scheme older cosign uses
//! (`sha256-<hex>.sig`, `.att`, `.sbom`) is plain tags.
//!
//! Disk layout (content-addressed; blobs shared across images):
//! {base_dir}/blobs/sha256/{hex}
//! {base_dir}/uploads/{id}
//! {base_dir}/manifests/{image}/tags/{tag}
//! {base_dir}/manifests/{image}/sha256/{hex}
//! {base_dir}/referrers/{image}/{subject-hex}/{hex} one referrer's descriptor (JSON)
use std::collections::HashMap;
use std::fs;
use std::io::{Read, Write};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, SystemTime};
use anyhow::{anyhow, bail, Context, Result};
use traits::{sha256_hex, ArtifactEntry, ArtifactFormat, ArtifactId, Caps, RepositoryBackendTrait};
const MANIFEST_MEDIA_TYPE: &str = "application/vnd.oci.image.manifest.v1+json";
/// The referrers API answers an image index.
const INDEX_MEDIA_TYPE: &str = "application/vnd.oci.image.index.v1+json";
/// The largest manifest `list` will read in order to publish its digest.
///
/// A listing must not become a read of the store, so the read that answers
/// `ArtifactEntry::sha256` is capped. 4 MiB is the OCI distribution spec's own
/// stated limit for a manifest; a stored file above it is either not a manifest
/// or is something a listing has no business streaming, and its digest is left
/// unknown instead.
const MAX_DIGESTED_MANIFEST: u64 = 4 * 1024 * 1024;
/// Default ceiling on a single uploaded blob.
///
/// The push path is writable by anyone the deployment lets write, and `PATCH
/// .../blobs/uploads/{id}` appended without any ceiling at all: an upload could be
/// grown until the disk filled, and finalize then read the whole accumulated file
/// into memory. One generous OCI layer — Helm charts are KiB-to-MiB — with the
/// ceiling enforced on every append, so overshoot costs one refused chunk instead
/// of the host. Tune per repository with [`OciRepoFile::with_max_blob_bytes`].
pub const DEFAULT_MAX_BLOB_BYTES: u64 = 1024 * 1024 * 1024;
/// How long an in-flight upload may sit untouched before the next `POST
/// .../blobs/uploads/` reclaims its staging file. A chunked push touches the file
/// on every `PATCH`, so only sessions a client walked away from age out.
const UPLOAD_SESSION_TTL: Duration = Duration::from_secs(24 * 60 * 60);
pub struct OciRepoFile {
pub name: String,
pub base_dir: PathBuf,
/// Monotonic source of temp-file names for [`OciRepoFile::write_atomic`].
/// NOT a source of upload ids — those must be unguessable (see
/// [`OciRepoFile::fresh_upload_id`]) — and the temp file is renamed onto its
/// target within the call.
upload_seq: AtomicU64,
/// Ceiling on one uploaded blob; see [`DEFAULT_MAX_BLOB_BYTES`].
max_blob_bytes: u64,
/// Per-session digest state, fed as chunks land; see [`OciRepoFile::hash_state`].
/// One lock per session (stinky 30): the map's lock is held only to find a
/// session, never while its bytes are hashed, so concurrent pushes of
/// different layers hash in parallel instead of queueing on one mutex.
in_flight: Mutex<HashMap<String, Arc<Mutex<UploadHash>>>>,
}
/// The sha256 of an upload session so far, and how many staged bytes it covers.
struct UploadHash {
hasher: nornir_hash::Sha256Hasher,
hashed: u64,
}
/// The bounded buffer a streamed chunk moves through on its way to the staging
/// file — the whole of what a PATCH costs this process in memory.
const STREAM_BUF_BYTES: usize = 1024 * 1024;
impl OciRepoFile {
pub fn new(name: String, base_dir: PathBuf) -> Result<Self> {
for sub in ["blobs/sha256", "uploads", "manifests", "referrers"] {
fs::create_dir_all(base_dir.join(sub))
.with_context(|| format!("creating {}", base_dir.join(sub).display()))?;
}
Ok(Self {
name,
base_dir,
upload_seq: AtomicU64::new(0),
max_blob_bytes: DEFAULT_MAX_BLOB_BYTES,
in_flight: Mutex::new(HashMap::new()),
})
}
/// Override the per-blob upload ceiling (default [`DEFAULT_MAX_BLOB_BYTES`]).
pub fn with_max_blob_bytes(mut self, max_blob_bytes: u64) -> Self {
self.max_blob_bytes = max_blob_bytes;
self
}
fn blob_path(&self, hex: &str) -> PathBuf {
self.base_dir.join("blobs/sha256").join(hex)
}
fn has_blob(&self, hex: &str) -> bool {
self.blob_path(hex).is_file()
}
/// The manifest file a manifest coordinate (`image:tag` or `image@sha256:…`) names.
fn manifest_path_of(&self, id: &ArtifactId) -> Option<PathBuf> {
let image = match &id.namespace {
Some(ns) => format!("{ns}/{}", id.name),
None => id.name.clone(),
};
if validate_image(&image).is_err() || check_component(&id.version).is_err() {
return None;
}
Some(match parse_sha256(&id.version) {
Some(hex) => self.manifest_digest_path(&image, &hex),
None => self.manifest_tag_path(&image, &id.version),
})
}
fn manifest_tag_path(&self, image: &str, tag: &str) -> PathBuf {
self.base_dir
.join("manifests")
.join(image)
.join("tags")
.join(tag)
}
fn manifest_digest_path(&self, image: &str, hex: &str) -> PathBuf {
self.base_dir
.join("manifests")
.join(image)
.join("sha256")
.join(hex)
}
/// Where the descriptors of `image`'s manifests whose subject is `subject_hex` live.
fn referrers_dir(&self, image: &str, subject_hex: &str) -> PathBuf {
self.base_dir
.join("referrers")
.join(image)
.join(subject_hex)
}
/// Walk `manifests/` and return every stored `(image, tag)` pair. An image is
/// any directory holding a `tags/` subdir (image names may contain slashes,
/// e.g. `demo/alpine-mini`), and each file under `tags/` is one pushed tag.
/// Read-only filesystem walk; missing/partial trees yield fewer rows, never
/// an error.
fn list_image_tags(&self) -> Vec<(String, String)> {
let root = self.base_dir.join("manifests");
let mut out = Vec::new();
let mut stack = vec![root.clone()];
while let Some(dir) = stack.pop() {
let Ok(entries) = fs::read_dir(&dir) else {
continue;
};
for entry in entries.flatten() {
let path = entry.path();
if !path.is_dir() {
continue;
}
if entry.file_name() == "tags" {
// `dir` is the image dir; image = its path relative to root.
let Ok(rel) = dir.strip_prefix(&root) else {
continue;
};
let image = rel.to_string_lossy().replace('\\', "/");
if let Ok(tags) = fs::read_dir(&path) {
for tag in tags.flatten() {
if tag.path().is_file() {
out.push((
image.clone(),
tag.file_name().to_string_lossy().into_owned(),
));
}
}
}
} else if entry.file_name() != "sha256" {
// Descend; skip `sha256/` (digest store, not image dirs).
stack.push(path);
}
}
}
out.sort();
out
}
/// Write `data` to a fresh temp file in the same dir then atomically rename
/// onto `target`, creating parent dirs as needed.
fn write_atomic(&self, target: &Path, data: &[u8]) -> Result<()> {
let parent = target
.parent()
.ok_or_else(|| anyhow!("no parent for {}", target.display()))?;
fs::create_dir_all(parent)?;
let seq = self.upload_seq.fetch_add(1, Ordering::Relaxed);
let tmp = parent.join(format!(".tmp-{seq}"));
fs::write(&tmp, data)?;
if let Err(e) = fs::rename(&tmp, target) {
let _ = fs::remove_file(&tmp);
return Err(e).with_context(|| format!("publishing {}", target.display()));
}
Ok(())
}
// ── blob uploads ────────────────────────────────────────────────────
/// A fresh, UNGUESSABLE upload-session id.
///
/// A process counter (the previous source) made every in-flight session id
/// predictable from a single observed one — `0`, `1`, `2`, … from process
/// start — so anyone could `PATCH` bytes into, or `PUT` a digest onto, a
/// session someone else opened. 128 bits from the OS CSPRNG; if that is
/// unreadable, the OS-seeded `RandomState` keys plus the clock, the pid and
/// the counter are mixed through SHA-256 instead (never the counter alone).
fn fresh_upload_id(&self) -> String {
let seq = self.upload_seq.fetch_add(1, Ordering::Relaxed);
let mut bytes = [0u8; 16];
if fs::File::open("/dev/urandom")
.and_then(|mut f| f.read_exact(&mut bytes))
.is_ok()
{
return bytes.iter().map(|b| format!("{b:02x}")).collect();
}
use std::hash::{BuildHasher, Hasher};
let mut hasher = std::collections::hash_map::RandomState::new().build_hasher();
hasher.write_u64(seq);
hasher.write_u32(std::process::id());
let nanos = SystemTime::now()
.duration_since(SystemTime::UNIX_EPOCH)
.map(|d| d.as_nanos())
.unwrap_or(0);
let mut seed = Vec::with_capacity(32);
seed.extend_from_slice(&hasher.finish().to_le_bytes());
seed.extend_from_slice(&nanos.to_le_bytes());
seed.extend_from_slice(&seq.to_le_bytes());
sha256_hex(&seed)[..32].to_string()
}
/// Reclaim staging files for uploads nobody finished. Without this an
/// abandoned `POST .../blobs/uploads/` left its file on disk forever, so an
/// unauthenticated client could fill the filesystem with sessions it never
/// intends to complete. Amortized onto the next session start; best-effort, so
/// an unreadable entry is skipped rather than failing the request.
fn sweep_abandoned_uploads(&self) {
let dir = self.base_dir.join("uploads");
let Ok(entries) = fs::read_dir(&dir) else {
return;
};
let now = SystemTime::now();
for entry in entries.flatten() {
let path = entry.path();
if !path.is_file() {
continue;
}
let stale = entry
.metadata()
.and_then(|m| m.modified())
.map(|m| now.duration_since(m).unwrap_or_default() > UPLOAD_SESSION_TTL)
.unwrap_or(false);
if stale {
log::info!(
"oci repo '{}': reclaiming abandoned upload {}",
self.name,
path.display()
);
let _ = fs::remove_file(&path);
}
}
}
fn begin_upload(&self, name: &str, query: &str) -> Result<HttpResp> {
// Monolithic form: POST .../uploads/?digest=sha256:… with the whole blob.
if let Some(digest) = query_param(query, "digest") {
// Body carried by the POST is handled by the caller path; here we
// only reach the no-body case, so create a session instead.
let _ = digest;
}
self.sweep_abandoned_uploads();
let id = self.fresh_upload_id();
// Touch the staging file so PATCH/PUT can append.
fs::write(self.base_dir.join("uploads").join(&id), [])?;
self.hash_state().insert(
id.clone(),
Arc::new(Mutex::new(UploadHash {
hasher: nornir_hash::Sha256Hasher::new(),
hashed: 0,
})),
);
Ok(HttpResp::new(202)
.header("Location", &format!("/v2/{name}/blobs/uploads/{id}"))
.header("Docker-Upload-UUID", &id)
.header("Range", "0-0"))
}
/// 413 + drop the session: an over-ceiling upload must not be retryable into
/// the same staging file, and its bytes are reclaimed immediately.
fn refuse_oversize(&self, staging: &Path) -> HttpResp {
let _ = fs::remove_file(staging);
if let Some(id) = staging.file_name().and_then(|s| s.to_str()) {
self.hash_state().remove(id);
}
log::warn!(
"oci repo '{}': upload refused — blob exceeds {} bytes",
self.name,
self.max_blob_bytes
);
HttpResp::new(413).body(b"blob exceeds maximum size".to_vec())
}
/// The digest state of an in-flight upload session, keyed by upload id —
/// fed per chunk as the bytes are appended (stinky-01 section 2.4: ONE
/// write, ONE hash pass). `hashed` is how many bytes the hasher has seen;
/// it must equal the staging file's length for the state to be trusted, so
/// a session this process did not see from the start (a restart mid-push)
/// falls back to hashing the staged file once at finalize.
fn hash_state(&self) -> std::sync::MutexGuard<'_, HashMap<String, Arc<Mutex<UploadHash>>>> {
self.in_flight.lock().unwrap_or_else(|e| e.into_inner())
}
/// Feed `bytes` (just appended, bringing the staging file to `now_len`) to
/// the session's hasher, or forget the session if the hasher is out of step
/// with the file.
fn hash_appended(&self, id: &str, bytes: &[u8], now_len: u64) {
let Some(session) = self.hash_state().get(id).cloned() else {
return;
};
let mut h = session.lock().unwrap_or_else(|e| e.into_inner());
if h.hashed + bytes.len() as u64 == now_len {
h.hasher.update(bytes);
h.hashed = now_len;
} else {
drop(h);
self.hash_state().remove(id);
}
}
/// Append one chunk from a reader — the streaming PATCH: the chunk goes
/// from the wire to the staging file through one bounded buffer and past
/// the session's hasher, and is never whole in memory. The ceiling is
/// enforced per buffer, before the bytes are written. A read error is the
/// door refusing the upload: the file is cut back to where this request
/// found it and the session's digest state is dropped.
fn append_stream(
&self,
staging: &Path,
id: &str,
body: &mut dyn Read,
) -> Result<Option<HttpResp>> {
let start_len = fs::metadata(staging)?.len();
let mut f = fs::OpenOptions::new().append(true).open(staging)?;
let mut buf = vec![0u8; STREAM_BUF_BYTES];
let mut len = start_len;
loop {
let n = match body.read(&mut buf) {
Ok(0) => break,
Ok(n) => n,
Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
Err(e) => {
drop(f);
let _ = fs::OpenOptions::new()
.write(true)
.open(staging)
.and_then(|f| f.set_len(start_len));
self.hash_state().remove(id);
return Err(e.into());
}
};
if len.saturating_add(n as u64) > self.max_blob_bytes {
drop(f);
self.hash_state().remove(id);
return Ok(Some(self.refuse_oversize(staging)));
}
f.write_all(&buf[..n])?;
len += n as u64;
self.hash_appended(id, &buf[..n], len);
}
Ok(None)
}
fn append_chunk(&self, name: &str, id: &str, body: &[u8]) -> Result<HttpResp> {
self.append_chunk_from(name, id, &mut std::io::Cursor::new(body))
}
/// The streaming PATCH: see [`OciRepoFile::append_stream`].
fn append_chunk_from(&self, name: &str, id: &str, body: &mut dyn Read) -> Result<HttpResp> {
check_component(id)?;
let staging = self.base_dir.join("uploads").join(id);
if !staging.is_file() {
return Ok(HttpResp::new(404).body(b"upload unknown".to_vec()));
}
if let Some(refused) = self.append_stream(&staging, id, body)? {
return Ok(refused);
}
let size = fs::metadata(&staging)?.len();
Ok(HttpResp::new(202)
.header("Location", &format!("/v2/{name}/blobs/uploads/{id}"))
.header("Docker-Upload-UUID", id)
.header("Range", &format!("0-{}", size.saturating_sub(1))))
}
fn finalize_upload(&self, name: &str, id: &str, query: &str, body: &[u8]) -> Result<HttpResp> {
self.finalize_upload_from(name, id, query, &mut std::io::Cursor::new(body))
}
/// Finalize: the trailing chunk (if any) is appended like every other, then
/// the digest is what the session's hasher says — no read-back — and the
/// staged file is RENAMED onto its content address, not copied. The bytes
/// of a layer are therefore written once and hashed once, whatever its
/// size. A session whose hasher this process did not see from the start
/// pays one read of the staged file, as before.
fn finalize_upload_from(
&self,
name: &str,
id: &str,
query: &str,
body: &mut dyn Read,
) -> Result<HttpResp> {
check_component(id)?;
let staging = self.base_dir.join("uploads").join(id);
if !staging.is_file() {
return Ok(HttpResp::new(404).body(b"upload unknown".to_vec()));
}
let expected = query_param(query, "digest")
.ok_or_else(|| anyhow!("missing digest on upload finalize"))?;
let hex = parse_sha256(&expected).ok_or_else(|| anyhow!("malformed digest {expected}"))?;
if let Some(refused) = self.append_stream(&staging, id, body)? {
return Ok(refused);
}
let len = fs::metadata(&staging)?.len();
let session = self.hash_state().remove(id);
let in_step = session.and_then(|s| {
let mut h = s.lock().unwrap_or_else(|e| e.into_inner());
let hasher = (h.hashed == len)
.then(|| std::mem::replace(&mut h.hasher, nornir_hash::Sha256Hasher::new()));
drop(h);
hasher
});
let actual = match in_step {
Some(hasher) => hasher.finalize_hex(),
None => nornir_hash::sha256_file(&staging)?,
};
if actual != hex {
let _ = fs::remove_file(&staging);
return Ok(HttpResp::new(400).body(b"digest mismatch".to_vec()));
}
let target = self.blob_path(&hex);
if let Some(parent) = target.parent() {
fs::create_dir_all(parent)?;
}
if let Err(e) = fs::rename(&staging, &target) {
let _ = fs::remove_file(&staging);
return Err(e).with_context(|| format!("publishing {}", target.display()));
}
log::info!("oci repo '{}': stored blob sha256:{}", self.name, hex);
Ok(HttpResp::new(201)
.header("Location", &format!("/v2/{name}/blobs/sha256:{hex}"))
.header("Docker-Content-Digest", &format!("sha256:{hex}")))
}
fn store_monolithic_blob(&self, name: &str, digest: &str, body: &[u8]) -> Result<HttpResp> {
if body.len() as u64 > self.max_blob_bytes {
log::warn!(
"oci repo '{}': monolithic upload refused — blob exceeds {} bytes",
self.name,
self.max_blob_bytes
);
return Ok(HttpResp::new(413).body(b"blob exceeds maximum size".to_vec()));
}
let hex = parse_sha256(digest).ok_or_else(|| anyhow!("malformed digest {digest}"))?;
if sha256_hex(body) != hex {
return Ok(HttpResp::new(400).body(b"digest mismatch".to_vec()));
}
self.write_atomic(&self.blob_path(&hex), body)?;
Ok(HttpResp::new(201)
.header("Location", &format!("/v2/{name}/blobs/sha256:{hex}"))
.header("Docker-Content-Digest", &format!("sha256:{hex}")))
}
// ── manifests ───────────────────────────────────────────────────────
fn put_manifest(&self, image: &str, reference: &str, body: &[u8]) -> Result<HttpResp> {
validate_image(image)?;
check_component(reference)?;
let Ok(manifest) = serde_json::from_slice::<serde_json::Value>(body) else {
return Ok(HttpResp::new(400).body(b"manifest invalid: not JSON".to_vec()));
};
// Every blob the manifest references (config + layers) must already be
// uploaded — reject dangling manifests. The `subject` is a MANIFEST, and
// OCI 1.1 lets it be absent from the registry: it is never required.
for digest in blob_digests(&manifest) {
if !self.has_blob(&digest) {
return Ok(HttpResp::new(400).body(b"manifest references unknown blob".to_vec()));
}
}
// An index names manifests, which this image must already hold.
for digest in index_digests(&manifest) {
if !self.manifest_digest_path(image, &digest).is_file() {
return Ok(HttpResp::new(400).body(b"index references unknown manifest".to_vec()));
}
}
let hex = sha256_hex(body);
self.write_atomic(&self.manifest_digest_path(image, &hex), body)?;
// A digest reference need not also be a tag.
if parse_sha256(reference).is_none() {
self.write_atomic(&self.manifest_tag_path(image, reference), body)?;
}
let mut resp = HttpResp::new(201)
.header(
"Location",
&format!("/v2/{}/{image}/manifests/{reference}", self.name),
)
.header("Docker-Content-Digest", &format!("sha256:{hex}"));
// A referrer: its descriptor is filed under its subject, and the push says so
// (`OCI-Subject`), which tells the client this registry keeps the referrers
// API and no fallback tag is needed.
if let Some(subject) = manifest
.pointer("/subject/digest")
.and_then(|d| d.as_str())
.and_then(parse_sha256)
{
let desc = descriptor_of(&manifest, body);
self.write_atomic(
&self.referrers_dir(image, &subject).join(&hex),
desc.to_string().as_bytes(),
)?;
resp = resp.header("OCI-Subject", &format!("sha256:{subject}"));
}
log::info!(
"oci repo '{}': stored manifest {image}:{reference}",
self.name
);
Ok(resp)
}
/// **`GET /v2/{name}/referrers/{digest}`**: an image index of the descriptors of
/// every manifest of `image` whose subject is `digest`, filtered by
/// `artifactType` when asked (and `OCI-Filters-Applied` says so). A digest with
/// none is an empty index; a malformed one is 400.
fn referrers(&self, image: &str, digest: &str, query: &str) -> Result<HttpResp> {
validate_image(image)?;
let Some(hex) = parse_sha256(digest) else {
return Ok(HttpResp::new(400)
.header("Content-Type", "application/json")
.body(
br#"{"errors":[{"code":"DIGEST_INVALID","message":"not a sha256 digest"}]}"#
.to_vec(),
));
};
let filter = query_param(query, "artifactType").filter(|t| !t.is_empty());
let mut found: Vec<(String, serde_json::Value)> = Vec::new();
if let Ok(rd) = fs::read_dir(self.referrers_dir(image, &hex)) {
for e in rd.flatten() {
let name = e.file_name().to_string_lossy().into_owned();
if name.starts_with(".tmp-") {
continue;
}
let Some(desc) = fs::read(e.path())
.ok()
.and_then(|b| serde_json::from_slice::<serde_json::Value>(&b).ok())
else {
continue;
};
if let Some(t) = &filter {
if desc.get("artifactType").and_then(|a| a.as_str()) != Some(t.as_str()) {
continue;
}
}
found.push((name, desc));
}
}
found.sort_by(|a, b| a.0.cmp(&b.0));
let index = referrers_index(found.into_iter().map(|(_, d)| d).collect());
let mut resp = HttpResp::new(200).header("Content-Type", INDEX_MEDIA_TYPE);
if filter.is_some() {
resp = resp.header("OCI-Filters-Applied", "artifactType");
}
Ok(resp.body(index))
}
fn get_manifest(&self, image: &str, reference: &str, head: bool) -> Result<HttpResp> {
validate_image(image)?;
check_component(reference)?;
let path = match parse_sha256(reference) {
Some(hex) => self.manifest_digest_path(image, &hex),
None => self.manifest_tag_path(image, reference),
};
match fs::read(&path) {
Ok(data) => {
let hex = sha256_hex(&data);
let resp = HttpResp::new(200)
.header("Content-Type", &manifest_media_type(&data))
.header("Docker-Content-Digest", &format!("sha256:{hex}"))
.header("Content-Length", &data.len().to_string());
Ok(if head { resp } else { resp.body(data) })
}
Err(_) => Ok(HttpResp::new(404).body(b"manifest unknown".to_vec())),
}
}
fn get_blob(&self, digest: &str, head: bool) -> Result<HttpResp> {
let hex = match parse_sha256(digest) {
Some(h) => h,
None => return Ok(HttpResp::new(400).body(b"malformed digest".to_vec())),
};
let resp = |len: u64| {
HttpResp::new(200)
.header("Content-Type", "application/octet-stream")
.header("Docker-Content-Digest", &format!("sha256:{hex}"))
.header("Content-Length", &len.to_string())
};
// A HEAD asks the size: the file's metadata answers it, the blob is not read
// (HOL-HEAD-1; it read the whole layer before).
if head {
return Ok(match fs::metadata(self.blob_path(&hex)) {
Ok(m) if m.is_file() => resp(m.len()),
_ => HttpResp::new(404).body(b"blob unknown".to_vec()).head(),
});
}
match fs::read(self.blob_path(&hex)) {
Ok(data) => Ok(resp(data.len() as u64).body(data)),
Err(_) => Ok(HttpResp::new(404).body(b"blob unknown".to_vec())),
}
}
fn tags_list(&self, image: &str) -> Result<HttpResp> {
validate_image(image)?;
let dir = self.base_dir.join("manifests").join(image).join("tags");
// M11: distinguish "unknown repository" from "known but empty". A
// missing manifests/tags dir means the repo isn't known — the OCI
// distribution spec requires 404 NAME_UNKNOWN, not 200 + an empty tag
// list (which masks the difference). Other read errors are surfaced
// (propagated) rather than swallowed into an empty list.
let read = match fs::read_dir(&dir) {
Ok(rd) => rd,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
return Ok(HttpResp::new(404)
.header("Content-Type", "application/json")
.body(name_unknown_body(image)));
}
Err(e) => return Err(e.into()),
};
let mut tags: Vec<String> = Vec::new();
for entry in read {
let entry = entry?; // surface read errors instead of swallowing them
if let Some(s) = entry.file_name().to_str() {
tags.push(s.to_owned());
}
}
tags.sort();
let list = tags
.iter()
.map(|t| format!("{t:?}"))
.collect::<Vec<_>>()
.join(",");
let json = format!(r#"{{"name":"{image}","tags":[{list}]}}"#);
Ok(HttpResp::new(200)
.header("Content-Type", "application/json")
.body(json.into_bytes()))
}
}
impl RepositoryBackendTrait for OciRepoFile {
fn name(&self) -> &str {
&self.name
}
fn format(&self) -> ArtifactFormat {
ArtifactFormat::Docker
}
fn is_writable(&self) -> bool {
true
}
/// A real listing, and uploads that stream into the staging file.
/// Streamed reads and a path mapping for the content-addressed blobs (HOL-PERF-3): a
/// layer streams from its file. A manifest path maps to nothing on purpose: the
/// whole-body door answers it, with the manifest's media type and digest headers.
fn caps(&self) -> Caps {
Caps {
listing: true,
streaming_uploads: true,
streaming_reads: true,
path_mapping: true,
..Caps::NONE
}
}
/// `/v2/<name>/blobs/sha256:<hex>` names the blob; every other path names nothing
/// streamable (the whole-body door answers it).
fn coordinate_for_path(&self, suburl: &str) -> anyhow::Result<Option<ArtifactId>> {
let path = suburl
.split('?')
.next()
.unwrap_or(suburl)
.trim_start_matches('/');
let segs: Vec<&str> = path.split('/').filter(|s| !s.is_empty()).collect();
if segs.first() != Some(&"v2") {
return Ok(None);
}
let rest = &segs[1..];
let rest = rest.strip_prefix(&[self.name.as_str()]).unwrap_or(rest);
Ok(match rest {
[.., "blobs", digest] if rest.len() >= 2 => {
parse_sha256(digest).map(|hex| blob_coordinate(&hex))
}
_ => None,
})
}
/// A blob coordinate streams the blob's file; a manifest coordinate streams the
/// manifest `fetch` reads.
fn fetch_chunks(
&self,
id: &ArtifactId,
chunk_bytes: usize,
) -> anyhow::Result<Option<traits::ChunkIter>> {
let path = match blob_of(id) {
Some(hex) => self.blob_path(&hex),
None => match self.manifest_path_of(id) {
Some(p) => p,
None => return Ok(None),
},
};
match fs::File::open(&path) {
Ok(f) if f.metadata().map(|m| m.is_file()).unwrap_or(false) => Ok(Some(
traits::backend::file_chunks(f, chunk_bytes.clamp(1, STREAM_BUF_BYTES)),
)),
Ok(_) => Ok(None),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
Err(e) => Err(e.into()),
}
}
/// Browse: one entry per pushed `image:tag`. The image's last path segment is
/// the `name`, any leading segments become the `namespace` (matching how
/// `fetch`/`put` split the image), and the tag is the `version`. `size_bytes`
/// is the stored manifest's byte length. `limit == 0` means "no cap" (callers
/// pass 0 for "all"); a positive `limit` caps the number of entries.
fn list(&self, name_filter: Option<&str>, limit: usize) -> anyhow::Result<Vec<ArtifactEntry>> {
let mut out = Vec::new();
for (image, tag) in self.list_image_tags() {
if let Some(f) = name_filter {
if !image.contains(f) {
continue;
}
}
let (namespace, name) = match image.rsplit_once('/') {
Some((ns, n)) => (Some(ns.to_string()), n.to_string()),
None => (None, image.clone()),
};
let tag_path = self.manifest_tag_path(&image, &tag);
let meta = fs::metadata(&tag_path).ok();
let size_bytes = meta.as_ref().map(|m| m.len() as i64).unwrap_or(0);
// ── The one backend that can hand out a REAL digest cheaply ──
//
// The listed artifact IS the manifest — `size_bytes` above is the
// manifest's length, not the image's — and OCI is content-addressed:
// `put_manifest` computes `sha256_hex(body)` and stores the same
// bytes twice, under `manifests/{image}/sha256/{hex}` and under the
// tag. So the tag file's SHA-256 is the `Docker-Content-Digest` a
// registry client already sees, and it is the honest answer here.
//
// It costs a read, so the read is BOUNDED. A manifest is kilobytes
// (the OCI spec's own guidance is 4 MiB; real ones are far under
// it), and anything above MAX_DIGESTED_MANIFEST is left unknown
// rather than allowed to turn a listing into a long read. That
// bound is what keeps this from being the defect it is fixing.
let sha256 = match meta.as_ref().map(|m| m.len()) {
Some(len) if len <= MAX_DIGESTED_MANIFEST => {
fs::read(&tag_path).ok().map(|b| sha256_hex(&b))
}
_ => None,
};
out.push(
ArtifactEntry::unknown_provenance(
ArtifactId {
namespace,
name,
version: tag,
},
size_bytes,
MANIFEST_MEDIA_TYPE,
)
.with_published_system_time(meta.and_then(|m| m.modified().ok()))
.with_sha256(sha256),
);
// `limit == 0` means "no cap" (callers pass 0 for "all"); only break
// once a positive limit is reached. Matches the znippy backends.
if limit != 0 && out.len() >= limit {
break;
}
}
Ok(out)
}
fn fetch(&self, id: &ArtifactId) -> anyhow::Result<Option<traits::Bytes>> {
if let Some(hex) = blob_of(id) {
return Ok(fs::read(self.blob_path(&hex)).ok().map(traits::Bytes::from));
}
let Some(path) = self.manifest_path_of(id) else {
return Ok(None);
};
Ok(fs::read(path).ok().map(traits::Bytes::from))
}
fn put(&self, id: &ArtifactId, data: &[u8]) -> anyhow::Result<()> {
// A blob coordinate stores the blob, its digest checked.
if let Some(hex) = blob_of(id) {
if sha256_hex(data) != hex {
bail!("blob digest mismatch: the bytes are not sha256:{hex}");
}
return self.write_atomic(&self.blob_path(&hex), data);
}
let image = match &id.namespace {
Some(ns) => format!("{ns}/{}", id.name),
None => id.name.clone(),
};
match self.put_manifest(&image, &id.version, data)?.status {
201 => Ok(()),
_ => bail!("manifest rejected (referenced blobs must be uploaded first)"),
}
}
fn handle_http2_request(
&self,
method: &str,
suburl: &str,
body: &[u8],
) -> anyhow::Result<(u16, Vec<(String, String)>, Vec<u8>)> {
let resp = self.dispatch(method, suburl, body).unwrap_or_else(|e| {
// Never leak internal detail to the client; log it instead.
log::warn!("oci repo '{}': {} {} -> {e:#}", self.name, method, suburl);
HttpResp::new(400).body(b"invalid request".to_vec())
});
Ok((resp.status, resp.headers, resp.body))
}
/// A manifest put from a reader: collected (manifests are kilobytes) and `put`.
fn put_from_reader(
&self,
id: &ArtifactId,
body: &mut dyn Read,
size_hint: Option<u64>,
) -> anyhow::Result<()> {
let buf = traits::collect_body(body, size_hint)?;
self.put(id, &buf)
}
/// A `PATCH`/`PUT` on an upload session streams its body straight into the
/// staging file (and past the session's hasher); every other write —
/// a manifest, a monolithic blob — is collected and dispatched as before.
/// Errors are answered the way `handle_http2_request` answers them: the
/// door has already decided a refusal of its own before it reads this.
fn handle_http2_upload(
&self,
method: &str,
suburl: &str,
body: &mut dyn Read,
size_hint: Option<u64>,
) -> anyhow::Result<Option<(u16, Vec<(String, String)>, Vec<u8>)>> {
let session = self.upload_session(suburl);
let resp = match (method, session) {
("PATCH", Some((client_name, id, _))) => {
self.append_chunk_from(&client_name, &id, body)
}
("PUT", Some((client_name, id, query))) => {
self.finalize_upload_from(&client_name, &id, &query, body)
}
_ => {
let buf = traits::collect_body(body, size_hint)?;
return self.handle_http2_request(method, suburl, &buf).map(Some);
}
};
let resp = resp.unwrap_or_else(|e| {
log::warn!("oci repo '{}': {} {} -> {e:#}", self.name, method, suburl);
HttpResp::new(400).body(b"invalid request".to_vec())
});
Ok(Some((resp.status, resp.headers, resp.body)))
}
}
impl OciRepoFile {
/// `(client-facing name, upload id, query)` when `suburl` addresses an
/// upload session (`/v2/{name…}/blobs/uploads/{id}`), else `None`. The same
/// parse `dispatch` does, so the streaming door and the collecting door
/// agree on what a session path is.
fn upload_session(&self, suburl: &str) -> Option<(String, String, String)> {
let (path, query) = match suburl.split_once('?') {
Some((p, q)) => (p, q),
None => (suburl, ""),
};
let segs: Vec<&str> = path
.trim_start_matches('/')
.split('/')
.filter(|s| !s.is_empty())
.collect();
if segs.first() != Some(&"v2") || segs.len() < 4 {
return None;
}
let rest = &segs[1..];
let rest = rest.strip_prefix(&[self.name.as_str()]).unwrap_or(rest);
let mi = rest.iter().position(|s| *s == "blobs")?;
let tail = &rest[mi..];
let [_, uploads, id] = tail else {
return None;
};
if *uploads != "uploads" || check_component(id).is_err() {
return None;
}
let image = rest[..mi].join("/");
let client_name = if image.is_empty() {
self.name.clone()
} else {
format!("{}/{}", self.name, image)
};
Some((client_name, id.to_string(), query.to_string()))
}
fn dispatch(&self, method: &str, suburl: &str, body: &[u8]) -> Result<HttpResp> {
let (path, query) = match suburl.split_once('?') {
Some((p, q)) => (p, q),
None => (suburl, ""),
};
let path = path.trim_start_matches('/');
let segs: Vec<&str> = path.split('/').filter(|s| !s.is_empty()).collect();
if segs.first() != Some(&"v2") {
return Ok(HttpResp::new(404).body(b"not found".to_vec()));
}
// `/v2/` version probe (also handled by the gateway, kept for safety).
if segs.len() == 1 {
return Ok(HttpResp::new(200)
.header("Docker-Distribution-Api-Version", "registry/2.0")
.body(b"{}".to_vec()));
}
// Strip `v2` and our own repo-name prefix, leaving `{image…}/{action}/…`.
let rest = &segs[1..];
let rest = rest.strip_prefix(&[self.name.as_str()]).unwrap_or(rest);
// Locate the action marker; everything before it is the image name.
let marker = rest
.iter()
.position(|s| *s == "manifests" || *s == "blobs" || *s == "tags" || *s == "referrers");
let Some(mi) = marker else {
return Ok(HttpResp::new(404).body(b"not found".to_vec()));
};
let image = rest[..mi].join("/");
let tail = &rest[mi..];
// Self-referential URLs (Location/etc.) must carry the full client-facing
// repository name = `{our repo}/{image}` so the gateway routes follow-ups
// back to this backend.
let client_name = if image.is_empty() {
self.name.clone()
} else {
format!("{}/{}", self.name, image)
};
match (method, tail) {
("GET", ["tags", "list"]) => self.tags_list(&image),
("HEAD", ["tags", "list"]) => self.tags_list(&image).map(HttpResp::head),
("GET", ["manifests", reference]) => self.get_manifest(&image, reference, false),
("HEAD", ["manifests", reference]) => self.get_manifest(&image, reference, true),
("PUT", ["manifests", reference]) => self.put_manifest(&image, reference, body),
("GET", ["referrers", digest]) => self.referrers(&image, digest, query),
("HEAD", ["referrers", digest]) => {
self.referrers(&image, digest, query).map(HttpResp::head)
}
("GET", ["blobs", digest]) => self.get_blob(digest, false),
("HEAD", ["blobs", digest]) => self.get_blob(digest, true),
("POST", ["blobs", "uploads"]) | ("POST", ["blobs", "uploads", ""]) => {
// Monolithic upload when a digest + body are provided up front.
match (query_param(query, "digest"), body.is_empty()) {
(Some(d), false) => self.store_monolithic_blob(&client_name, &d, body),
_ => self.begin_upload(&client_name, query),
}
}
("PATCH", ["blobs", "uploads", id]) => self.append_chunk(&client_name, id, body),
("PUT", ["blobs", "uploads", id]) => {
self.finalize_upload(&client_name, id, query, body)
}
_ => Ok(HttpResp::new(404).body(b"not found".to_vec())),
}
}
}
// ── small HTTP response builder ─────────────────────────────────────────
struct HttpResp {
status: u16,
headers: Vec<(String, String)>,
body: Vec<u8>,
}
impl HttpResp {
fn new(status: u16) -> Self {
Self {
status,
headers: Vec::new(),
body: Vec::new(),
}
}
fn header(mut self, k: &str, v: &str) -> Self {
self.headers.push((k.to_string(), v.to_string()));
self
}
fn body(mut self, body: Vec<u8>) -> Self {
self.body = body;
self
}
/// This answer for a HEAD: its headers and length, no body (HOL-HEAD-1).
fn head(self) -> Self {
let (status, headers, body) =
traits::backend::head_of((self.status, self.headers, self.body));
Self {
status,
headers,
body,
}
}
}
// ── helpers ─────────────────────────────────────────────────────────────
/// The namespace of a blob coordinate. No image is named so: an OCI path component
/// never starts with `_`, and a manifest coordinate always carries a reference.
const BLOB_NAMESPACE: &str = "_blob";
/// The coordinate of the blob `sha256:<hex>`: `_blob/<hex>`, no version.
fn blob_coordinate(hex: &str) -> ArtifactId {
ArtifactId {
namespace: Some(BLOB_NAMESPACE.to_string()),
name: hex.to_string(),
version: String::new(),
}
}
/// The blob a coordinate names, when it is a blob coordinate.
fn blob_of(id: &ArtifactId) -> Option<String> {
(id.namespace.as_deref() == Some(BLOB_NAMESPACE) && id.version.is_empty())
.then(|| parse_sha256(&format!("sha256:{}", id.name)))
.flatten()
}
/// Parse a `sha256:<64-hex>` reference, returning the lowercase hex on success.
fn parse_sha256(s: &str) -> Option<String> {
let hex = s.strip_prefix("sha256:")?;
if hex.len() == 64 && hex.bytes().all(|b| b.is_ascii_hexdigit()) {
Some(hex.to_ascii_lowercase())
} else {
None
}
}
fn query_param(query: &str, key: &str) -> Option<String> {
query.split('&').find_map(|kv| {
let (k, v) = kv.split_once('=')?;
// Clients percent-encode the `:` in `sha256:…` as `%3A`.
(k == key).then(|| percent_decode(v))
})
}
/// Minimal percent-decoding for query-parameter values.
fn percent_decode(s: &str) -> String {
let bytes = s.as_bytes();
let mut out = Vec::with_capacity(bytes.len());
let mut i = 0;
while i < bytes.len() {
if bytes[i] == b'%' && i + 2 < bytes.len() {
if let (Some(h), Some(l)) = (hex_val(bytes[i + 1]), hex_val(bytes[i + 2])) {
out.push(h * 16 + l);
i += 3;
continue;
}
}
out.push(bytes[i]);
i += 1;
}
String::from_utf8_lossy(&out).into_owned()
}
fn hex_val(b: u8) -> Option<u8> {
match b {
b'0'..=b'9' => Some(b - b'0'),
b'a'..=b'f' => Some(b - b'a' + 10),
b'A'..=b'F' => Some(b - b'A' + 10),
_ => None,
}
}
/// Reject a single path component that could escape the storage root.
fn check_component(s: &str) -> Result<()> {
if s.is_empty() || s == "." || s == ".." || s.contains('/') || s.contains('\\') {
bail!("invalid path component {s:?}");
}
Ok(())
}
/// Validate an OCI image name (may contain `/`), guarding every segment.
fn validate_image(image: &str) -> Result<()> {
if image.is_empty() {
bail!("empty image name");
}
for seg in image.split('/') {
check_component(seg)?;
}
Ok(())
}
/// OCI-distribution-spec error body for an unknown repository (404). `image`
/// is already validated by [`validate_image`] (`check_component` rejects
/// quotes/control chars), so direct interpolation is safe.
fn name_unknown_body(image: &str) -> Vec<u8> {
format!(
r#"{{"errors":[{{"code":"NAME_UNKNOWN","message":"repository name not known to registry","detail":{{"name":"{image}"}}}}]}}"#
)
.into_bytes()
}
/// The BLOBS a manifest references: its config and its layers (lowercase hex).
/// Its `subject` (a manifest, which may live anywhere or nowhere yet) and an
/// index's `manifests` are not blobs.
fn blob_digests(manifest: &serde_json::Value) -> Vec<String> {
let config = manifest.pointer("/config/digest").and_then(|d| d.as_str());
let layers = manifest
.get("layers")
.and_then(|l| l.as_array())
.into_iter()
.flatten()
.filter_map(|l| l.get("digest").and_then(|d| d.as_str()));
config
.into_iter()
.chain(layers)
.filter_map(parse_sha256)
.collect()
}
/// The manifests an image index names (lowercase hex).
fn index_digests(manifest: &serde_json::Value) -> Vec<String> {
manifest
.get("manifests")
.and_then(|m| m.as_array())
.into_iter()
.flatten()
.filter_map(|m| {
m.get("digest")
.and_then(|d| d.as_str())
.and_then(parse_sha256)
})
.collect()
}
/// [`blob_digests`] of raw manifest bytes; none for bytes that are not JSON.
#[cfg(test)]
fn referenced_digests(manifest: &[u8]) -> Vec<String> {
serde_json::from_slice::<serde_json::Value>(manifest)
.map(|v| blob_digests(&v))
.unwrap_or_default()
}
/// The `mediaType` a manifest declares about itself (an OCI manifest when it says none).
fn manifest_media_type(body: &[u8]) -> String {
serde_json::from_slice::<serde_json::Value>(body)
.ok()
.and_then(|v| {
v.get("mediaType")
.and_then(|m| m.as_str())
.map(str::to_string)
})
.unwrap_or_else(|| MANIFEST_MEDIA_TYPE.to_string())
}
/// **The descriptor a referrers index lists for one manifest** (OCI 1.1): its media
/// type, digest and size, its `artifactType` (the manifest's own, else its
/// config's media type) and its annotations.
pub fn descriptor_of(manifest: &serde_json::Value, bytes: &[u8]) -> serde_json::Value {
let mut d = serde_json::json!({
"mediaType": manifest.get("mediaType").and_then(|m| m.as_str()).unwrap_or(MANIFEST_MEDIA_TYPE),
"digest": format!("sha256:{}", sha256_hex(bytes)),
"size": bytes.len(),
});
let artifact_type = manifest
.get("artifactType")
.and_then(|a| a.as_str())
.or_else(|| {
manifest
.pointer("/config/mediaType")
.and_then(|m| m.as_str())
});
if let Some(t) = artifact_type {
d["artifactType"] = serde_json::Value::String(t.to_string());
}
if let Some(a) = manifest
.get("annotations")
.filter(|a| a.as_object().is_some_and(|o| !o.is_empty()))
{
d["annotations"] = a.clone();
}
d
}
/// **A referrers answer**: the image index of `descriptors`.
pub fn referrers_index(descriptors: Vec<serde_json::Value>) -> Vec<u8> {
serde_json::json!({ "schemaVersion": 2, "mediaType": INDEX_MEDIA_TYPE, "manifests": descriptors })
.to_string()
.into_bytes()
}
#[cfg(test)]
mod tests {
use super::*;
#[cfg(feature = "testmatrix")]
use traits::test_support::fstatus;
fn repo() -> (tempfile::TempDir, OciRepoFile) {
let dir = tempfile::tempdir().unwrap();
let repo = OciRepoFile::new("charts".into(), dir.path().to_path_buf()).unwrap();
(dir, repo)
}
fn call(
r: &OciRepoFile,
method: &str,
url: &str,
body: &[u8],
) -> (u16, Vec<(String, String)>, Vec<u8>) {
r.handle_http2_request(method, url, body).unwrap()
}
fn header<'a>(h: &'a [(String, String)], k: &str) -> Option<&'a str> {
h.iter().find(|(hk, _)| hk == k).map(|(_, v)| v.as_str())
}
#[test]
fn tags_list_unknown_repo_returns_404_name_unknown() {
let (_d, r) = repo();
// No manifests dir exists for this image → must be 404 NAME_UNKNOWN,
// not 200 + an empty tag list (M11).
let (s, h, body) = call(&r, "GET", "/v2/charts/nope/tags/list", &[]);
assert_eq!(
s,
404,
"unknown repo must 404, body={:?}",
String::from_utf8_lossy(&body)
);
assert_eq!(header(&h, "Content-Type"), Some("application/json"));
let text = String::from_utf8_lossy(&body);
assert!(
text.contains("NAME_UNKNOWN"),
"expected NAME_UNKNOWN, got: {text}"
);
}
#[test]
fn version_probe() {
let (_d, r) = repo();
let (s, h, _) = call(&r, "GET", "/v2/", &[]);
assert_eq!(s, 200);
let api = header(&h, "Docker-Distribution-Api-Version");
assert_eq!(api, Some("registry/2.0"));
#[cfg(feature = "testmatrix")]
fstatus(
"file-oci",
"version_probe",
s == 200 && api == Some("registry/2.0"),
&format!("GET /v2/ -> {s}, api={api:?}"),
);
}
#[test]
fn full_push_pull_roundtrip_session_upload() {
let (_d, r) = repo();
let blob = b"chart-tarball-bytes";
let digest = format!("sha256:{}", sha256_hex(blob));
// begin upload
let (s, h, _) = call(&r, "POST", "/v2/charts/mychart/blobs/uploads/", &[]);
assert_eq!(s, 202);
let loc = header(&h, "Location").unwrap().to_string();
assert!(header(&h, "Docker-Upload-UUID").is_some());
// finalize with the blob + digest on the PUT
let (s, h, _) = call(&r, "PUT", &format!("{loc}?digest={digest}"), blob);
assert_eq!(s, 201);
assert_eq!(header(&h, "Docker-Content-Digest"), Some(digest.as_str()));
// blob is now retrievable + HEADable
let (s, h, got) = call(
&r,
"GET",
&format!("/v2/charts/mychart/blobs/{digest}"),
&[],
);
assert_eq!(s, 200);
assert_eq!(got, blob);
assert_eq!(
header(&h, "Content-Length"),
Some(blob.len().to_string().as_str())
);
let (s, _, body) = call(
&r,
"HEAD",
&format!("/v2/charts/mychart/blobs/{digest}"),
&[],
);
assert_eq!(s, 200);
assert!(body.is_empty());
// a manifest referencing that blob
let manifest = format!(
r#"{{"schemaVersion":2,"config":{{"mediaType":"application/vnd.cncf.helm.config.v1+json","digest":"{digest}"}},"layers":[{{"mediaType":"application/vnd.cncf.helm.chart.content.v1.tar+gzip","digest":"{digest}"}}]}}"#
);
let (s, h, _) = call(
&r,
"PUT",
"/v2/charts/mychart/manifests/0.1.0",
manifest.as_bytes(),
);
assert_eq!(s, 201);
let mdigest = header(&h, "Docker-Content-Digest").unwrap().to_string();
// pull by tag and by digest
let (s, h, got) = call(&r, "GET", "/v2/charts/mychart/manifests/0.1.0", &[]);
assert_eq!(s, 200);
assert_eq!(got, manifest.as_bytes());
assert_eq!(header(&h, "Content-Type"), Some(MANIFEST_MEDIA_TYPE));
let (s, _, got) = call(
&r,
"GET",
&format!("/v2/charts/mychart/manifests/{mdigest}"),
&[],
);
assert_eq!(s, 200);
assert_eq!(got, manifest.as_bytes());
// tags list
let (s, _, got) = call(&r, "GET", "/v2/charts/mychart/tags/list", &[]);
assert_eq!(s, 200);
let tags = String::from_utf8(got).unwrap();
assert!(tags.contains("\"0.1.0\""));
#[cfg(feature = "testmatrix")]
fstatus(
"file-oci",
"push_pull_roundtrip",
s == 200 && tags.contains("\"0.1.0\""),
&format!(
"blob+manifest pushed, pulled by tag+digest, tags list has 0.1.0={}",
tags.contains("\"0.1.0\"")
),
);
}
fn r_ref(r: &OciRepoFile) -> &dyn RepositoryBackendTrait {
r
}
/// Browse: `list()` surfaces every pushed `image:tag`, including
/// slash-namespaced image names (`demo/app`).
fn push_manifest(r: &OciRepoFile, image: &str, tag: &str) {
let blob = format!("blob-for-{image}-{tag}").into_bytes();
let digest = format!("sha256:{}", sha256_hex(&blob));
let (_s, h, _) = call(r, "POST", &format!("/v2/{image}/blobs/uploads/"), &[]);
let loc = header(&h, "Location").unwrap().to_string();
let (s, _, _) = call(r, "PUT", &format!("{loc}?digest={digest}"), &blob);
assert_eq!(s, 201);
let manifest = format!(
r#"{{"schemaVersion":2,"config":{{"mediaType":"application/vnd.oci.image.config.v1+json","digest":"{digest}"}},"layers":[]}}"#
);
let (s, _, _) = call(
r,
"PUT",
&format!("/v2/{image}/manifests/{tag}"),
manifest.as_bytes(),
);
assert_eq!(s, 201);
}
/// Push a manifest whose referenced blob is `blob_len` bytes, so a store
/// can be given real weight without making its manifests bigger.
fn push_manifest_with_blob(r: &OciRepoFile, image: &str, tag: &str, blob_len: usize) {
let mut blob = vec![b'0'; blob_len];
// Distinct content per tag, so the blobs are not one deduplicated file.
let seed = format!("{image}:{tag}").into_bytes();
blob[..seed.len().min(blob_len)].copy_from_slice(&seed[..seed.len().min(blob_len)]);
let digest = format!("sha256:{}", sha256_hex(&blob));
let (_s, h, _) = call(r, "POST", &format!("/v2/{image}/blobs/uploads/"), &[]);
let loc = header(&h, "Location").unwrap().to_string();
let (s, _, _) = call(r, "PUT", &format!("{loc}?digest={digest}"), &blob);
assert_eq!(s, 201);
let manifest = format!(
r#"{{"schemaVersion":2,"config":{{"mediaType":"application/vnd.oci.image.config.v1+json","digest":"{digest}"}},"layers":[]}}"#
);
let (s, _, _) = call(
r,
"PUT",
&format!("/v2/{image}/manifests/{tag}"),
manifest.as_bytes(),
);
assert_eq!(s, 201);
}
/// ★ **THE COST, COUNTED: a listing reads manifests and not the store.**
///
/// This is the one backend whose listing got more expensive — it reads each
/// tag's manifest in order to publish a digest — so the cost is counted
/// here rather than argued about. The warning it guards against is precise:
/// a digest per entry must not turn a cheap listing into a full read of
/// every artifact.
///
/// What is counted, per entry:
/// BEFORE one `metadata` call (for `size_bytes`), 0 bytes read.
/// AFTER the same one `metadata` call, plus one bounded `read` of the
/// manifest — whose length IS `size_bytes`, so the bytes read by
/// the whole listing are exactly `sum(size_bytes)`.
///
/// The blobs — which is where an OCI store's weight actually is — are never
/// opened. This asserts that as a ratio rather than a feeling: the listing
/// reads under a hundredth of the store here, and the assertion fails the
/// moment somebody makes the listing walk blobs or layers.
#[test]
fn the_listing_reads_manifests_and_not_the_store_counted() {
const IMAGES: usize = 200;
const BLOB_LEN: usize = 64 * 1024;
let (_d, r) = repo();
for i in 0..IMAGES {
push_manifest_with_blob(&r, &format!("demo/app-{i}"), "1.0", BLOB_LEN);
}
let entries = r.list(None, 0).unwrap();
assert_eq!(entries.len(), IMAGES);
assert!(
entries.iter().all(|e| e.sha256.is_some()),
"every entry must carry a digest, or the cost below buys nothing"
);
// `size_bytes` is the manifest length, so this IS the number of bytes
// the digest step read. Nothing else on the listing path reads.
let manifest_bytes: i64 = entries.iter().map(|e| e.size_bytes).sum();
let blob_bytes = (IMAGES * BLOB_LEN) as i64;
assert!(
manifest_bytes * 100 < blob_bytes,
"the listing read {manifest_bytes} B of manifests against {blob_bytes} B of blobs — \
that is no longer a bounded read of the index, it is a read of the store"
);
// And each of those reads is individually capped, so one hostile object
// cannot make the listing long even if the ratio held.
assert!(
entries
.iter()
.all(|e| (e.size_bytes as u64) <= MAX_DIGESTED_MANIFEST),
"an entry past the cap must have been left unknown, not read"
);
println!(
"file-oci listing cost, COUNTED: {} entries, {manifest_bytes} B of manifests read \
({} B/entry), {blob_bytes} B of blobs in the store and none of them opened; \
syscalls per entry before = 1 metadata, after = 1 metadata + 1 bounded read",
entries.len(),
manifest_bytes / entries.len() as i64,
);
}
/// ★ **An OCI listing publishes a REAL digest, and it is the same digest a
/// registry client is handed.**
///
/// This is the backend that can answer both fields cheaply, and the reason
/// is not generosity: the listed artifact IS the manifest, OCI is
/// content-addressed, and `put_manifest` stores the same bytes under
/// `manifests/{image}/sha256/{hex}` and under the tag. So the digest in the
/// listing is the one in `Docker-Content-Digest`, and a holger → holger
/// mirror of an OCI repository reaches `unchanged` with no policy written.
///
/// The two are asserted to be EQUAL rather than each merely present — two
/// plausible digests computed over different bytes is exactly the failure
/// that a mirror would turn into a permanent re-fetch.
#[test]
fn an_oci_listing_publishes_the_same_digest_the_registry_door_does() {
let (_d, r) = repo();
push_manifest(&r, "demo/alpine-mini", "1.0");
let entries = r.list(None, 100).unwrap();
assert_eq!(entries.len(), 1);
let e = &entries[0];
let listed = e
.sha256
.clone()
.expect("an OCI listing can prove its manifest");
assert_eq!(
listed.len(),
64,
"64 lowercase hex, no `sha256:` prefix: {listed}"
);
// What the registry door says about the very same manifest.
let (status, headers, _) = call(&r, "GET", "/v2/demo/alpine-mini/manifests/1.0", b"");
assert_eq!(status, 200);
let served = headers
.iter()
.find(|(k, _)| k.eq_ignore_ascii_case("Docker-Content-Digest"))
.map(|(_, v)| v.clone())
.expect("the door publishes a content digest");
assert_eq!(
served,
format!("sha256:{listed}"),
"the listing and the door must name the same bytes"
);
let ms = e
.published_unix_ms
.expect("a stored manifest has a modification time");
assert!(ms > 0, "never a fabricated zero");
}
/// The digest read is BOUNDED, and the bound is what stops a listing
/// becoming a read of the store. A stored file above the cap is left
/// unknown rather than streamed.
#[test]
fn a_manifest_above_the_cap_is_left_unknown_rather_than_read() {
assert_eq!(
MAX_DIGESTED_MANIFEST,
4 * 1024 * 1024,
"the OCI spec's own manifest bound"
);
let (_d, r) = repo();
push_manifest(&r, "demo/alpine-mini", "1.0");
// Overwrite the tag file with something past the cap, in place, so the
// listing sees a stored object it must refuse to hash.
let path = r.manifest_tag_path("demo/alpine-mini", "1.0");
fs::write(&path, vec![b'x'; (MAX_DIGESTED_MANIFEST + 1) as usize]).unwrap();
let entries = r.list(None, 100).unwrap();
assert_eq!(entries.len(), 1);
assert_eq!(
entries[0].sha256, None,
"past the cap the digest is unknown — a listing must not stream an unbounded file"
);
assert!(
entries[0].published_unix_ms.is_some(),
"the time is still free, and is still answered"
);
}
#[test]
fn list_surfaces_pushed_images_and_caps_match_methods() {
let (_d, r) = repo();
assert!(
r.list(None, 100).unwrap().is_empty(),
"empty repo lists nothing"
);
traits::backend::conformance::check(r_ref(&r)).unwrap();
push_manifest(&r, "demo/alpine-mini", "1.0");
push_manifest(&r, "demo/alpine-mini", "1.1");
push_manifest(&r, "toplevel", "latest");
let entries = r.list(None, 100).unwrap();
assert_eq!(
entries.len(),
3,
"two tags of demo/alpine-mini + one toplevel: {entries:?}"
);
// slash-namespaced image → namespace="demo", name="alpine-mini"
let mini: Vec<_> = entries
.iter()
.filter(|e| e.id.namespace.as_deref() == Some("demo") && e.id.name == "alpine-mini")
.map(|e| e.id.version.as_str())
.collect();
assert_eq!(mini, vec!["1.0", "1.1"], "both tags surfaced, sorted");
// single-segment image → no namespace
assert!(entries.iter().any(|e| e.id.namespace.is_none()
&& e.id.name == "toplevel"
&& e.id.version == "latest"));
// size + media type are populated from the stored manifest
let any = &entries[0];
assert!(any.size_bytes > 0, "manifest byte size populated");
assert_eq!(any.content_type, MANIFEST_MEDIA_TYPE);
// name_filter narrows to the matching image
assert_eq!(r.list(Some("toplevel"), 100).unwrap().len(), 1);
// Not an archive: the accessor says so instead of answering.
assert!(traits::is_unsupported(&r.archive().unwrap_err()));
traits::backend::conformance::check(r_ref(&r)).unwrap();
#[cfg(feature = "testmatrix")]
fstatus(
"file-oci",
"browse_list",
entries.len() == 3,
&format!("list() surfaced {} pushed image:tag refs", entries.len()),
);
}
/// Regression: `limit == 0` means "all" (the callers' convention, e.g.
/// `object::paginate_artifacts`), not "at most zero / one". A prior bug broke
/// after the first entry, so Helm/OCI Browse only ever showed one image.
#[test]
fn list_limit_zero_returns_all() {
let (_d, r) = repo();
push_manifest(&r, "demo/alpine-mini", "1.0");
push_manifest(&r, "demo/alpine-mini", "1.1");
push_manifest(&r, "toplevel", "latest");
let all = r.list(None, 0).unwrap();
assert_eq!(
all.len(),
3,
"limit=0 must return every entry, not one: {all:?}"
);
// A positive limit still caps.
assert_eq!(
r.list(None, 2).unwrap().len(),
2,
"positive limit caps the listing"
);
#[cfg(feature = "testmatrix")]
fstatus(
"file-oci",
"list_limit_zero_returns_all",
all.len() == 3,
&format!("limit=0 surfaced {} entries", all.len()),
);
}
/// Blob uploads must have a ceiling, enforced BEFORE the bytes are written.
/// Uncapped, `PATCH .../blobs/uploads/{id}` let an unauthenticated client grow
/// one staging file until the disk filled — and finalize then read the whole
/// accumulated file into memory.
#[test]
fn blob_upload_is_capped() {
let dir = tempfile::tempdir().unwrap();
// A tiny ceiling so the test proves the rule, not the number.
let r = OciRepoFile::new("charts".into(), dir.path().to_path_buf())
.unwrap()
.with_max_blob_bytes(64);
let (_s, h, _) = call(&r, "POST", "/v2/charts/x/blobs/uploads/", &[]);
let loc = header(&h, "Location").unwrap().to_string();
let id = loc.rsplit('/').next().unwrap().to_string();
let staging = dir.path().join("uploads").join(&id);
// A chunk inside the ceiling is accepted.
let (s, _h, _) = call(&r, "PATCH", &loc, &[b'a'; 40]);
assert_eq!(s, 202, "a chunk within the ceiling must be accepted");
assert_eq!(fs::metadata(&staging).unwrap().len(), 40);
// The chunk that would cross it is refused — and NOT written.
let (s, _h, _) = call(&r, "PATCH", &loc, &[b'b'; 40]);
assert_eq!(
s, 413,
"a chunk crossing the ceiling must be refused with 413"
);
assert!(
!staging.is_file(),
"the over-ceiling session's staging bytes must be reclaimed, not left on disk"
);
// A single monolithic POST over the ceiling is refused too.
let big = vec![b'c'; 200];
let digest = format!("sha256:{}", sha256_hex(&big));
let (s, _h, _) = call(
&r,
"POST",
&format!("/v2/charts/x/blobs/uploads/?digest={digest}"),
&big,
);
assert_eq!(s, 413, "a monolithic blob over the ceiling must be refused");
let (s, _h, _) = call(&r, "GET", &format!("/v2/charts/x/blobs/{digest}"), &[]);
assert_eq!(s, 404, "a refused blob must not be stored");
// And a PUT that tries to smuggle the overshoot in as the final chunk.
let (_s, h, _) = call(&r, "POST", "/v2/charts/x/blobs/uploads/", &[]);
let loc2 = header(&h, "Location").unwrap().to_string();
let (s, _h, _) = call(&r, "PATCH", &loc2, &[b'd'; 60]);
assert_eq!(s, 202);
let tail = vec![b'e'; 60];
let (s, _h, _) = call(
&r,
"PUT",
&format!("{loc2}?digest=sha256:{}", "0".repeat(64)),
&tail,
);
assert_eq!(s, 413, "the final chunk on PUT is under the same ceiling");
#[cfg(feature = "testmatrix")]
fstatus(
"file-oci",
"blob_upload_capped",
s == 413,
"PATCH/PUT/monolithic blob uploads are bounded by max_blob_bytes; 413 + staging reclaimed",
);
}
/// Upload-session ids must be unguessable. From a process counter they were
/// `0`, `1`, `2`, … — observe one and you can `PATCH` into (or `PUT` a digest
/// onto) every other client's in-flight session.
#[test]
fn upload_session_ids_are_unpredictable() {
let (_d, r) = repo();
let mut ids = Vec::new();
for _ in 0..8 {
let (_s, h, _) = call(&r, "POST", "/v2/charts/x/blobs/uploads/", &[]);
ids.push(header(&h, "Docker-Upload-UUID").unwrap().to_string());
}
// All distinct…
let unique: std::collections::HashSet<&String> = ids.iter().collect();
assert_eq!(
unique.len(),
ids.len(),
"session ids must be distinct: {ids:?}"
);
// …none of them the counter value an attacker would guess…
for (i, id) in ids.iter().enumerate() {
assert_ne!(
id,
&format!("{i:032x}"),
"session id must not be a counter: {ids:?}"
);
let as_int = u128::from_str_radix(id, 16).expect("32 hex chars");
assert!(
as_int > u128::from(u32::MAX),
"session id must span the whole space, not small integers: {id}"
);
}
// …and successive ids are not adjacent (a counter's are).
let a = u128::from_str_radix(&ids[0], 16).unwrap();
let b = u128::from_str_radix(&ids[1], 16).unwrap();
assert!(
a.abs_diff(b) > 1,
"successive session ids must not be adjacent: {ids:?}"
);
#[cfg(feature = "testmatrix")]
fstatus(
"file-oci",
"upload_ids_unpredictable",
unique.len() == ids.len(),
"blob-upload session ids are 128-bit CSPRNG values, not a process counter",
);
}
/// Abandoned upload sessions are reclaimed. Without the sweep, every
/// `POST .../blobs/uploads/` nobody finished left its staging file on disk
/// forever.
#[test]
fn abandoned_upload_sessions_are_swept() {
let (_d, r) = repo();
// An upload that was started and walked away from, aged past the TTL.
let abandoned = r.base_dir.join("uploads").join("abandoned-session");
fs::write(&abandoned, b"half an upload").unwrap();
let old = SystemTime::now() - (UPLOAD_SESSION_TTL + Duration::from_secs(60));
fs::File::options()
.write(true)
.open(&abandoned)
.unwrap()
.set_times(fs::FileTimes::new().set_modified(old))
.unwrap();
// A session still being worked on (fresh mtime) must survive.
let (_s, h, _) = call(&r, "POST", "/v2/charts/x/blobs/uploads/", &[]);
let live_id = header(&h, "Docker-Upload-UUID").unwrap().to_string();
assert!(
!abandoned.is_file(),
"an upload untouched for longer than the TTL must be reclaimed"
);
assert!(
r.base_dir.join("uploads").join(&live_id).is_file(),
"the in-flight session must survive the sweep"
);
#[cfg(feature = "testmatrix")]
fstatus(
"file-oci",
"abandoned_uploads_swept",
!abandoned.is_file(),
"staging files untouched past UPLOAD_SESSION_TTL are reclaimed on the next upload start",
);
}
#[test]
fn percent_encoded_digest_on_finalize() {
let (_d, r) = repo();
let blob = b"helm-style-upload";
let hex = sha256_hex(blob);
// Clients send `digest=sha256%3A<hex>` (encoded ':').
let (_s, h, _) = call(&r, "POST", "/v2/charts/x/blobs/uploads/", &[]);
let loc = header(&h, "Location").unwrap().to_string();
let (s, _, _) = call(&r, "PUT", &format!("{loc}?digest=sha256%3A{hex}"), blob);
assert_eq!(s, 201);
let (s, _, got) = call(&r, "GET", &format!("/v2/charts/x/blobs/sha256:{hex}"), &[]);
assert_eq!(s, 200);
assert_eq!(got, blob);
}
#[test]
fn monolithic_upload() {
let (_d, r) = repo();
let blob = b"abc";
let digest = format!("sha256:{}", sha256_hex(blob));
let (s, _, _) = call(
&r,
"POST",
&format!("/v2/charts/x/blobs/uploads/?digest={digest}"),
blob,
);
assert_eq!(s, 201);
let (s, _, got) = call(&r, "GET", &format!("/v2/charts/x/blobs/{digest}"), &[]);
assert_eq!(s, 200);
assert_eq!(got, blob);
}
#[test]
fn chunked_upload_via_patch() {
let (_d, r) = repo();
let blob = b"hello world chunked";
let digest = format!("sha256:{}", sha256_hex(blob));
let (_s, h, _) = call(&r, "POST", "/v2/charts/x/blobs/uploads/", &[]);
let loc = header(&h, "Location").unwrap().to_string();
let (s, _, _) = call(&r, "PATCH", &loc, &blob[..5]);
assert_eq!(s, 202);
let (s, _, _) = call(&r, "PUT", &format!("{loc}?digest={digest}"), &blob[5..]);
assert_eq!(s, 201);
let (s, _, got) = call(&r, "GET", &format!("/v2/charts/x/blobs/{digest}"), &[]);
assert_eq!(s, 200);
assert_eq!(got, blob);
}
#[test]
fn digest_mismatch_rejected() {
let (_d, r) = repo();
let wrong = format!("sha256:{}", "0".repeat(64));
let (_s, h, _) = call(&r, "POST", "/v2/charts/x/blobs/uploads/", &[]);
let loc = header(&h, "Location").unwrap().to_string();
let (s, _, _) = call(&r, "PUT", &format!("{loc}?digest={wrong}"), b"data");
assert_eq!(s, 400);
#[cfg(feature = "testmatrix")]
fstatus(
"file-oci",
"digest_mismatch_rejected",
s == 400,
&format!("wrong digest on finalize -> status {s}"),
);
}
#[test]
fn manifest_with_unknown_blob_rejected() {
let (_d, r) = repo();
let missing = format!("sha256:{}", "a".repeat(64));
let manifest = format!(r#"{{"config":{{"digest":"{missing}"}},"layers":[]}}"#);
let (s, _, _) = call(
&r,
"PUT",
"/v2/charts/x/manifests/1.0.0",
manifest.as_bytes(),
);
assert_eq!(s, 400);
}
#[test]
fn unknown_blob_and_manifest_404() {
let (_d, r) = repo();
let d = format!("sha256:{}", "b".repeat(64));
assert_eq!(
call(&r, "GET", &format!("/v2/charts/x/blobs/{d}"), &[]).0,
404
);
assert_eq!(call(&r, "GET", "/v2/charts/x/manifests/nope", &[]).0, 404);
}
#[test]
fn malformed_digest_400() {
let (_d, r) = repo();
assert_eq!(
call(&r, "GET", "/v2/charts/x/blobs/sha256:short", &[]).0,
400
);
}
#[test]
fn traversal_in_image_or_reference_blocked() {
let (_d, r) = repo();
// `..` as an image segment must not escape; resolves to 404/400, never a write outside base.
let (s, _, _) = call(&r, "GET", "/v2/charts/../../etc/manifests/x", &[]);
assert!(s == 404 || s == 400);
let img_err = validate_image("../escape").is_err();
let comp_err = check_component("..").is_err();
assert!(img_err);
assert!(comp_err);
#[cfg(feature = "testmatrix")]
fstatus(
"file-oci",
"path_traversal_blocked",
(s == 404 || s == 400) && img_err && comp_err,
&format!("traversal GET -> {s}, validate_image err={img_err}, check_component err={comp_err}"),
);
}
#[test]
fn referenced_digests_scan() {
let d = format!("sha256:{}", "c".repeat(64));
let m = format!(r#"{{"config":{{"digest":"{d}"}},"layers":[{{"digest":"{d}"}}]}}"#);
let found = referenced_digests(m.as_bytes());
assert_eq!(found.len(), 2);
assert!(found.iter().all(|h| h == &"c".repeat(64)));
}
#[test]
fn put_via_trait_requires_blobs() {
let (_d, r) = repo();
let id = ArtifactId {
namespace: None,
name: "mychart".into(),
version: "1.0.0".into(),
};
// No blobs uploaded → manifest referencing one is rejected.
let bad = format!(r#"{{"config":{{"digest":"sha256:{}"}}}}"#, "d".repeat(64));
assert!(r.put(&id, bad.as_bytes()).is_err());
// A manifest with no blob references is accepted and fetchable by tag.
let ok = br#"{"schemaVersion":2,"layers":[]}"#;
r.put(&id, ok).unwrap();
let fetched = r.fetch(&id).unwrap().unwrap();
assert_eq!(&fetched[..], &ok[..]);
#[cfg(feature = "testmatrix")]
{
let bad_rejected = r
.put(
&id,
format!(r#"{{"config":{{"digest":"sha256:{}"}}}}"#, "e".repeat(64)).as_bytes(),
)
.is_err();
fstatus(
"file-oci",
"put_via_trait_requires_blobs",
fetched == ok && bad_rejected,
&format!("no-blob manifest accepted+fetched={}, missing-blob manifest rejected={bad_rejected}", fetched == ok),
);
}
}
/// **A blob streams from its file** (HOL-PERF-3). `GET /v2/<name>/blobs/<digest>` read
/// the whole layer into memory (a 64 MiB layer was 64 MiB per pull). The backend now
/// maps a blob path to its blob and streams it in pieces; a manifest path maps to
/// nothing, so the whole-body door answers it with the manifest's own headers.
#[test]
fn a_blob_is_streamed_from_its_file_and_mapped_by_its_path() {
let (_d, r) = repo();
let blob: Vec<u8> = (0..3 * 1024 * 1024 + 17).map(|i| (i % 251) as u8).collect();
let digest = format!("sha256:{}", sha256_hex(&blob));
let (s, _, _) = call(
&r,
"POST",
&format!("/v2/charts/img/blobs/uploads/?digest={digest}"),
&blob,
);
assert_eq!(s, 201);
let caps = r.caps();
assert!(caps.streaming_reads && caps.path_mapping, "{caps:?}");
let id = r_ref(&r)
.coordinate(&format!("/v2/charts/img/blobs/{digest}"))
.expect("a blob path names its blob");
let chunks = r
.fetch_chunks(&id, 1 << 20)
.unwrap()
.expect("the blob is held");
assert_eq!(chunks.len, Some(blob.len() as u64));
let mut got = Vec::new();
let mut pieces = 0;
for c in chunks {
let c = c.unwrap();
assert!(c.len() <= 1 << 20, "a piece is at most one piece");
got.extend_from_slice(&c);
pieces += 1;
}
assert_eq!((got == blob, pieces), (true, 4));
assert_eq!(
r_ref(&r).coordinate("/v2/charts/img/manifests/1.0"),
None,
"a manifest is the whole-body door's"
);
let missing = r_ref(&r)
.coordinate(&format!("/v2/charts/img/blobs/sha256:{}", "0".repeat(64)))
.unwrap();
assert!(
r.fetch_chunks(&missing, 1 << 20).unwrap().is_none(),
"a blob not held is not streamed"
);
traits::backend::conformance::check(r_ref(&r)).unwrap();
}
}
/// **OCI Distribution 1.1: the referrers API and the attached artifacts** (lane
/// HOL-REFERRERS, owner 2026-10-30: a customer's `cosign verify`, `notation verify`
/// or admission controller verifies an image pulled from Holger, through Holger).
#[cfg(test)]
mod referrers_tests {
use super::*;
const INDEX_MT: &str = "application/vnd.oci.image.index.v1+json";
const EMPTY_MT: &str = "application/vnd.oci.empty.v1+json";
fn repo() -> (tempfile::TempDir, OciRepoFile) {
let dir = tempfile::tempdir().unwrap();
let repo = OciRepoFile::new("oci".into(), dir.path().to_path_buf()).unwrap();
(dir, repo)
}
fn call(
r: &OciRepoFile,
method: &str,
url: &str,
body: &[u8],
) -> (u16, Vec<(String, String)>, Vec<u8>) {
r.handle_http2_request(method, url, body).unwrap()
}
fn header<'a>(h: &'a [(String, String)], k: &str) -> Option<&'a str> {
h.iter()
.find(|(hk, _)| hk.eq_ignore_ascii_case(k))
.map(|(_, v)| v.as_str())
}
fn digest(b: &[u8]) -> String {
format!("sha256:{}", sha256_hex(b))
}
fn blob(r: &OciRepoFile, image: &str, b: &[u8]) -> String {
let d = digest(b);
let (s, _, _) = call(
r,
"POST",
&format!("/v2/oci/{image}/blobs/uploads/?digest={d}"),
b,
);
assert_eq!(s, 201, "blob upload");
d
}
/// An image: one config, one layer. `(manifest digest, size)`.
fn image(r: &OciRepoFile, image: &str, tag: &str) -> (String, usize) {
// The config names the tag, so two tags are two images.
let config = format!(r#"{{"architecture":"amd64","os":"linux","tag":"{tag}"}}"#);
let c = blob(r, image, config.as_bytes());
let l = blob(r, image, b"a layer");
let m = format!(
r#"{{"schemaVersion":2,"mediaType":"application/vnd.oci.image.manifest.v1+json","config":{{"mediaType":"application/vnd.oci.image.config.v1+json","digest":"{c}","size":{}}},"layers":[{{"mediaType":"application/vnd.oci.image.layer.v1.tar","digest":"{l}","size":7}}]}}"#,
config.len()
);
let (s, _, b) = call(
r,
"PUT",
&format!("/v2/oci/{image}/manifests/{tag}"),
m.as_bytes(),
);
assert_eq!(s, 201, "{}", String::from_utf8_lossy(&b));
(digest(m.as_bytes()), m.len())
}
/// An artifact manifest attached to `subject` (what `oras attach`, `notation sign`
/// and cosign's new bundle format push): the empty config, one layer, `subject`.
fn attach(
r: &OciRepoFile,
image: &str,
subject: &(String, usize),
artifact_type: &str,
layer: &[u8],
annotations: &str,
) -> (u16, Vec<(String, String)>, String) {
let e = blob(r, image, b"{}");
let l = blob(r, image, layer);
let m = format!(
r#"{{"schemaVersion":2,"mediaType":"application/vnd.oci.image.manifest.v1+json","artifactType":"{artifact_type}","config":{{"mediaType":"{EMPTY_MT}","digest":"{e}","size":2}},"layers":[{{"mediaType":"{artifact_type}","digest":"{l}","size":{}}}],"subject":{{"mediaType":"application/vnd.oci.image.manifest.v1+json","digest":"{}","size":{}}},"annotations":{{{annotations}}}}}"#,
layer.len(),
subject.0,
subject.1
);
let d = digest(m.as_bytes());
let (s, h, _) = call(
r,
"PUT",
&format!("/v2/oci/{image}/manifests/{d}"),
m.as_bytes(),
);
(s, h, d)
}
fn referrers(
r: &OciRepoFile,
image: &str,
subject: &str,
query: &str,
) -> (u16, Vec<(String, String)>, serde_json::Value) {
let (s, h, b) = call(
r,
"GET",
&format!("/v2/oci/{image}/referrers/{subject}{query}"),
&[],
);
let v = serde_json::from_slice(&b).unwrap_or(serde_json::Value::Null);
(s, h, v)
}
#[test]
fn a_manifest_with_a_subject_is_kept_and_the_push_answers_oci_subject() {
let (_d, r) = repo();
let img = image(&r, "demo", "1.0");
// The subject is a MANIFEST, never a blob of this repository: the push must
// not be refused as "references unknown blob".
let (s, h, d) = attach(
&r,
"demo",
&img,
"application/vnd.cncf.notary.signature",
b"a jws",
"",
);
assert_eq!(s, 201, "an artifact with a subject is accepted");
assert_eq!(
header(&h, "OCI-Subject"),
Some(img.0.as_str()),
"the push tells the client the registry keeps referrers: {h:?}"
);
assert_eq!(header(&h, "Docker-Content-Digest"), Some(d.as_str()));
// A subject that is not (yet) in the registry is accepted too (OCI 1.1).
let ghost = (format!("sha256:{}", "e".repeat(64)), 10);
let (s, _, _) = attach(&r, "demo", &ghost, "application/spdx+json", b"{}", "");
assert_eq!(s, 201);
}
#[test]
fn the_referrers_api_lists_every_manifest_whose_subject_is_the_digest() {
let (_d, r) = repo();
let img = image(&r, "demo", "1.0");
let other = image(&r, "demo", "2.0");
let (_, _, sig) = attach(
&r,
"demo",
&img,
"application/vnd.cncf.notary.signature",
b"jws",
r#""io.cncf.notary.x509chain.thumbprint#S256":"[\"ab\"]""#,
);
let (_, _, sbom) = attach(
&r,
"demo",
&img,
"application/spdx+json",
br#"{"spdxVersion":"SPDX-2.3"}"#,
r#""org.opencontainers.image.created":"2026-10-30T00:00:00Z""#,
);
let (_, _, _not_this) = attach(&r, "demo", &other, "application/spdx+json", b"other", "");
let (s, h, v) = referrers(&r, "demo", &img.0, "");
assert_eq!(s, 200);
assert_eq!(header(&h, "Content-Type"), Some(INDEX_MT));
assert_eq!(v["schemaVersion"], 2);
assert_eq!(v["mediaType"], INDEX_MT);
let m = v["manifests"].as_array().expect("manifests");
let digests: std::collections::BTreeSet<&str> =
m.iter().filter_map(|d| d["digest"].as_str()).collect();
assert_eq!(
digests,
[sig.as_str(), sbom.as_str()].into_iter().collect(),
"{v}"
);
let one = m.iter().find(|d| d["digest"] == sbom.as_str()).unwrap();
assert_eq!(
one["mediaType"],
"application/vnd.oci.image.manifest.v1+json"
);
assert_eq!(one["artifactType"], "application/spdx+json");
assert!(one["size"].as_u64().unwrap() > 0);
assert_eq!(
one["annotations"]["org.opencontainers.image.created"], "2026-10-30T00:00:00Z",
"the manifest's annotations ride in the descriptor"
);
// Each listed referrer is fetchable by its digest, as the client will.
let (s, h, b) = call(&r, "GET", &format!("/v2/oci/demo/manifests/{sbom}"), &[]);
assert_eq!(s, 200);
assert_eq!(digest(&b), sbom);
assert_eq!(
header(&h, "Content-Type"),
Some("application/vnd.oci.image.manifest.v1+json")
);
}
#[test]
fn an_artifact_type_filter_is_applied_and_said() {
let (_d, r) = repo();
let img = image(&r, "demo", "1.0");
attach(
&r,
"demo",
&img,
"application/vnd.cncf.notary.signature",
b"jws",
"",
);
let (_, _, sbom) = attach(&r, "demo", &img, "application/spdx+json", b"{}", "");
let (s, h, v) = referrers(
&r,
"demo",
&img.0,
"?artifactType=application%2Fspdx%2Bjson",
);
assert_eq!(s, 200);
assert_eq!(header(&h, "OCI-Filters-Applied"), Some("artifactType"));
let m = v["manifests"].as_array().unwrap();
assert_eq!(m.len(), 1, "{v}");
assert_eq!(m[0]["digest"], sbom.as_str());
// No filter asked, none said.
let (_, h, _) = referrers(&r, "demo", &img.0, "");
assert_eq!(header(&h, "OCI-Filters-Applied"), None);
}
#[test]
fn a_digest_with_no_referrers_is_an_empty_index_and_a_malformed_one_is_400() {
let (_d, r) = repo();
let img = image(&r, "demo", "1.0");
let (s, _, v) = referrers(&r, "demo", &img.0, "");
assert_eq!(s, 200);
assert_eq!(v["manifests"], serde_json::json!([]));
let (s, _, v) = referrers(
&r,
"nothing-here",
&format!("sha256:{}", "a".repeat(64)),
"",
);
assert_eq!(
s, 200,
"an unknown subject is an empty index, as OCI 1.1 allows"
);
assert_eq!(v["manifests"], serde_json::json!([]));
let (s, _, _) = call(&r, "GET", "/v2/oci/demo/referrers/sha256:zz", &[]);
assert_eq!(s, 400);
}
#[test]
fn a_manifest_is_served_with_its_own_media_type_and_an_index_needs_its_manifests() {
let (_d, r) = repo();
let a = image(&r, "multi", "amd64");
let idx = format!(
r#"{{"schemaVersion":2,"mediaType":"{INDEX_MT}","manifests":[{{"mediaType":"application/vnd.oci.image.manifest.v1+json","digest":"{}","size":{},"platform":{{"os":"linux","architecture":"amd64"}}}}]}}"#,
a.0, a.1
);
let (s, _, b) = call(&r, "PUT", "/v2/oci/multi/manifests/latest", idx.as_bytes());
assert_eq!(
s,
201,
"an index of manifests this image holds is accepted: {}",
String::from_utf8_lossy(&b)
);
let (s, h, _) = call(&r, "GET", "/v2/oci/multi/manifests/latest", &[]);
assert_eq!(s, 200);
assert_eq!(header(&h, "Content-Type"), Some(INDEX_MT));
let dangling = format!(
r#"{{"schemaVersion":2,"mediaType":"{INDEX_MT}","manifests":[{{"mediaType":"application/vnd.oci.image.manifest.v1+json","digest":"sha256:{}","size":1}}]}}"#,
"f".repeat(64)
);
let (s, _, _) = call(
&r,
"PUT",
"/v2/oci/multi/manifests/broken",
dangling.as_bytes(),
);
assert_eq!(
s, 400,
"an index naming a manifest this image does not hold is refused"
);
}
/// The tag scheme older cosign uses (`sha256-<hex>.sig`, `.att`, `.sbom`): plain
/// tags, kept and listed like any other.
#[test]
fn the_legacy_cosign_tags_are_kept_and_listed() {
let (_d, r) = repo();
let img = image(&r, "demo", "1.0");
let sig_layer = blob(&r, "demo", br#"{"critical":{}}"#);
let cfg = blob(
&r,
"demo",
br#"{"architecture":"","os":"","config":{},"rootfs":{"type":"layers","diff_ids":[]}}"#,
);
let m = format!(
r#"{{"schemaVersion":2,"mediaType":"application/vnd.oci.image.manifest.v1+json","config":{{"mediaType":"application/vnd.oci.image.config.v1+json","digest":"{cfg}","size":1}},"layers":[{{"mediaType":"application/vnd.dev.cosign.simplesigning.v1+json","digest":"{sig_layer}","size":15,"annotations":{{"dev.cosignproject.cosign/signature":"MEUC"}}}}]}}"#
);
let tag = format!("sha256-{}.sig", &img.0[7..]);
let (s, _, _) = call(
&r,
"PUT",
&format!("/v2/oci/demo/manifests/{tag}"),
m.as_bytes(),
);
assert_eq!(s, 201);
let (s, _, b) = call(&r, "GET", "/v2/oci/demo/tags/list", &[]);
assert_eq!(s, 200);
assert!(
String::from_utf8_lossy(&b).contains(&tag),
"{}",
String::from_utf8_lossy(&b)
);
let (s, _, got) = call(&r, "GET", &format!("/v2/oci/demo/manifests/{tag}"), &[]);
assert_eq!((s, got), (200, m.into_bytes()));
}
}