use std::cell::{OnceCell, RefCell};
use std::path::PathBuf;
use crate::source::{LayerRef, LayerSource, SourceError};
pub const LAYER_ARTIFACT_TYPE: &str = "application/vnd.pulseengine.varve.layer.v1+json";
pub const ANN_ROLE: &str = "eu.pulseengine.varve.role";
pub const ROLE_ENVELOPE: &str = "envelope";
pub const ROLE_PAYLOAD: &str = "payload";
pub const ROLE_LINE_STATUS: &str = "line-status";
pub const ROLE_LINE_INDEX: &str = "line-index";
pub const ROLE_ATTESTATION_STATEMENT: &str = "attestation-statement";
pub const ROLE_ATTESTATION_BYTES: &str = "attestation-bytes";
pub const CREDENTIAL_ENV: &str = "VARVE_REGISTRY_AUTH";
pub const MANIFEST_ACCEPT: &str = "application/vnd.oci.image.manifest.v1+json, \
application/vnd.docker.distribution.manifest.v2+json, \
application/vnd.oci.image.index.v1+json, \
application/vnd.docker.distribution.manifest.list.v2+json";
const TAGS_PAGE_SIZE: u32 = 100;
const MAX_TAG_PAGES: usize = 64;
const MAX_BODY_BYTES: u64 = 8 * 1024 * 1024 * 1024;
const MAX_TOKEN_BYTES: u64 = 1024 * 1024;
fn layers_of_line(tags: Vec<String>, line: &str) -> Vec<String> {
tags.into_iter()
.filter(|tag| {
tag.parse::<crate::layer::LayerId>()
.is_ok_and(|id| id.line().to_string() == line)
})
.collect()
}
fn layer_digest_for_role(manifest: &serde_json::Value, role: &str) -> Option<String> {
manifest["layers"]
.as_array()?
.iter()
.find(|l| l["annotations"][ANN_ROLE] == role)
.and_then(|l| l["digest"].as_str())
.map(str::to_string)
}
const B64: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
fn base64_encode(input: &[u8]) -> String {
let mut out = String::with_capacity(input.len().div_ceil(3) * 4);
for chunk in input.chunks(3) {
let b0 = chunk[0] as u32;
let b1 = *chunk.get(1).unwrap_or(&0) as u32;
let b2 = *chunk.get(2).unwrap_or(&0) as u32;
let n = (b0 << 16) | (b1 << 8) | b2;
out.push(B64[(n >> 18) as usize & 63] as char);
out.push(B64[(n >> 12) as usize & 63] as char);
out.push(if chunk.len() > 1 {
B64[(n >> 6) as usize & 63] as char
} else {
'='
});
out.push(if chunk.len() > 2 {
B64[n as usize & 63] as char
} else {
'='
});
}
out
}
fn base64_decode(input: &str) -> Option<Vec<u8>> {
let mut acc: u32 = 0;
let mut bits: u32 = 0;
let mut out = Vec::with_capacity(input.len() / 4 * 3);
for c in input.bytes() {
let v = match c {
b'A'..=b'Z' => c - b'A',
b'a'..=b'z' => c - b'a' + 26,
b'0'..=b'9' => c - b'0' + 52,
b'+' => 62,
b'/' => 63,
b'=' | b'\n' | b'\r' | b' ' | b'\t' => continue,
_ => return None,
} as u32;
acc = ((acc << 6) | v) & 0x3_FFFF;
bits += 6;
if bits >= 8 {
bits -= 8;
out.push((acc >> bits) as u8);
}
}
Some(out)
}
#[derive(Clone, PartialEq, Eq)]
struct Credential {
username: String,
password: String,
origin: String,
}
impl std::fmt::Debug for Credential {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Credential")
.field("origin", &self.origin)
.field("username", &"<redacted>")
.field("password", &"<redacted>")
.finish()
}
}
impl Credential {
fn basic_header(&self) -> String {
format!(
"Basic {}",
base64_encode(format!("{}:{}", self.username, self.password).as_bytes())
)
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
enum CredentialLookup {
Found(Credential),
HelperOnly {
helper: String,
origin: String,
},
Malformed {
origin: String,
},
Absent,
}
fn first_usable(lookups: Vec<CredentialLookup>) -> CredentialLookup {
let mut explanation = CredentialLookup::Absent;
for lookup in lookups {
match lookup {
CredentialLookup::Found(_) => return lookup,
CredentialLookup::Absent => {}
other => {
if matches!(explanation, CredentialLookup::Absent) {
explanation = other;
}
}
}
}
explanation
}
fn credential_from_env_value(value: &str) -> CredentialLookup {
let origin = format!("${CREDENTIAL_ENV}");
let value = value.trim_end_matches(['\n', '\r']);
if value.is_empty() {
return CredentialLookup::Absent;
}
match value.split_once(':') {
Some((username, password)) if !username.is_empty() => CredentialLookup::Found(Credential {
username: username.to_string(),
password: password.to_string(),
origin,
}),
_ => CredentialLookup::Malformed { origin },
}
}
fn decode_basic_auth(encoded: &str) -> Option<(String, String)> {
let decoded = base64_decode(encoded.trim())?;
let text = String::from_utf8(decoded).ok()?;
let (username, password) = text.split_once(':')?;
if username.is_empty() {
return None;
}
Some((username.to_string(), password.to_string()))
}
fn registry_key_matches(key: &str, registry: &str) -> bool {
fn host(s: &str) -> String {
let s = s
.strip_prefix("https://")
.or_else(|| s.strip_prefix("http://"))
.unwrap_or(s);
s.split('/').next().unwrap_or(s).to_ascii_lowercase()
}
const HUB: [&str; 3] = ["docker.io", "index.docker.io", "registry-1.docker.io"];
let (key, registry) = (host(key), host(registry));
key == registry || (HUB.contains(&key.as_str()) && HUB.contains(®istry.as_str()))
}
fn credential_from_docker_config(
config: &serde_json::Value,
registry: &str,
origin: &str,
) -> CredentialLookup {
if let Some(auths) = config["auths"].as_object()
&& let Some((_, entry)) = auths
.iter()
.find(|(k, _)| registry_key_matches(k, registry))
{
if let Some(auth) = entry["auth"].as_str().filter(|a| !a.is_empty()) {
return match decode_basic_auth(auth) {
Some((username, password)) => CredentialLookup::Found(Credential {
username,
password,
origin: origin.to_string(),
}),
None => CredentialLookup::Malformed {
origin: origin.to_string(),
},
};
}
if let (Some(username), Some(password)) =
(entry["username"].as_str(), entry["password"].as_str())
&& !username.is_empty()
{
return CredentialLookup::Found(Credential {
username: username.to_string(),
password: password.to_string(),
origin: origin.to_string(),
});
}
}
if let Some(helpers) = config["credHelpers"].as_object()
&& let Some((_, helper)) = helpers
.iter()
.find(|(k, _)| registry_key_matches(k, registry))
&& let Some(helper) = helper.as_str().filter(|h| !h.is_empty())
{
return CredentialLookup::HelperOnly {
helper: helper.to_string(),
origin: origin.to_string(),
};
}
if let Some(store) = config["credsStore"].as_str().filter(|s| !s.is_empty()) {
return CredentialLookup::HelperOnly {
helper: store.to_string(),
origin: origin.to_string(),
};
}
CredentialLookup::Absent
}
fn credential_config_paths() -> Vec<PathBuf> {
let dir = |var: &str, tail: &str| -> Option<PathBuf> {
let value = std::env::var(var).ok()?;
if value.is_empty() {
return None;
}
Some(PathBuf::from(value).join(tail))
};
[
dir("DOCKER_CONFIG", "config.json"),
dir("HOME", ".docker/config.json"),
dir("XDG_RUNTIME_DIR", "containers/auth.json"),
]
.into_iter()
.flatten()
.collect()
}
fn lookups_from_paths(paths: &[PathBuf], registry: &str) -> Vec<CredentialLookup> {
paths
.iter()
.filter_map(|path| {
let text = std::fs::read_to_string(path).ok()?;
let json = serde_json::from_str::<serde_json::Value>(&text).ok()?;
Some(credential_from_docker_config(
&json,
registry,
&path.display().to_string(),
))
})
.collect()
}
fn resolve_credential(registry: &str) -> CredentialLookup {
let mut lookups = Vec::new();
if let Ok(value) = std::env::var(CREDENTIAL_ENV) {
lookups.push(credential_from_env_value(&value));
}
lookups.extend(lookups_from_paths(&credential_config_paths(), registry));
first_usable(lookups)
}
fn credential_advice(lookup: &CredentialLookup, registry: &str, repository: &str) -> String {
match lookup {
CredentialLookup::Found(credential) => format!(
"varve sent the credential from {} and the registry rejected it. Check that the \
username is right and that it may pull {repository}.",
credential.origin
),
CredentialLookup::HelperOnly { helper, origin } => format!(
"varve offered no credential: {origin} delegates {registry} to the credential helper \
'{helper}', and varve does not execute credential helpers — sourcing a secret by \
running a PATH-resolved binary is exactly the trust varve refuses (REQ-SHADOW-001). \
Supply it directly instead: {CREDENTIAL_ENV}='<username>:<password>' (for ECR: \
{CREDENTIAL_ENV}=\"AWS:$(aws ecr get-login-password --region <region>)\")."
),
CredentialLookup::Malformed { origin } => format!(
"varve offered no credential: {origin} is set but is not a `username:password` pair. \
(varve does not log the value.)"
),
CredentialLookup::Absent => format!(
"varve offered no credential: set {CREDENTIAL_ENV}='<username>:<password>', or \
`docker login {registry}` so the credential lands in the `auths` section of \
~/.docker/config.json — varve reads `auths`, and does not run credential helpers."
),
}
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
struct BearerChallenge {
realm: Option<String>,
service: Option<String>,
scope: Option<String>,
}
fn parse_bearer_challenge(header: &str) -> Option<BearerChallenge> {
let header = header.trim();
let (scheme, params) = match header.split_once(char::is_whitespace) {
Some((scheme, params)) => (scheme, params),
None => (header, ""),
};
if !scheme.eq_ignore_ascii_case("Bearer") {
return None;
}
let chars: Vec<char> = params.chars().collect();
let mut challenge = BearerChallenge::default();
let mut i = 0;
while i < chars.len() {
while i < chars.len() && (chars[i] == ',' || chars[i].is_whitespace()) {
i += 1;
}
let key_start = i;
while i < chars.len() && chars[i] != '=' && chars[i] != ',' {
i += 1;
}
if i >= chars.len() || chars[i] != '=' {
break;
}
let key = chars[key_start..i]
.iter()
.collect::<String>()
.trim()
.to_ascii_lowercase();
i += 1;
let value = if chars.get(i) == Some(&'"') {
i += 1;
let mut value = String::new();
while i < chars.len() {
if chars[i] == '\\' && i + 1 < chars.len() {
value.push(chars[i + 1]);
i += 2;
continue;
}
if chars[i] == '"' {
i += 1;
break;
}
value.push(chars[i]);
i += 1;
}
value
} else {
let value_start = i;
while i < chars.len() && chars[i] != ',' {
i += 1;
}
chars[value_start..i]
.iter()
.collect::<String>()
.trim()
.to_string()
};
match key.as_str() {
"realm" => challenge.realm = Some(value),
"service" => challenge.service = Some(value),
"scope" => challenge.scope = Some(value),
_ => {}
}
}
Some(challenge)
}
fn percent_encode(value: &str) -> String {
let mut out = String::with_capacity(value.len());
for b in value.bytes() {
match b {
b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
out.push(b as char)
}
_ => out.push_str(&format!("%{b:02X}")),
}
}
out
}
fn token_url(challenge: &BearerChallenge, default_scope: &str) -> Option<String> {
let realm = challenge.realm.as_deref()?.trim();
if realm.is_empty() {
return None;
}
let mut query = Vec::new();
if let Some(service) = challenge.service.as_deref().filter(|s| !s.is_empty()) {
query.push(format!("service={}", percent_encode(service)));
}
let scope = challenge
.scope
.as_deref()
.filter(|s| !s.is_empty())
.unwrap_or(default_scope);
query.push(format!("scope={}", percent_encode(scope)));
let separator = if realm.contains('?') { '&' } else { '?' };
Some(format!("{realm}{separator}{}", query.join("&")))
}
fn realm_is_acceptable(realm: &str, reference_scheme: &str) -> bool {
if reference_scheme == "https" {
realm.starts_with("https://")
} else {
realm.starts_with("http://") || realm.starts_with("https://")
}
}
fn token_from_body(body: &str) -> Option<String> {
let json: serde_json::Value = serde_json::from_str(body).ok()?;
json["token"]
.as_str()
.or_else(|| json["access_token"].as_str())
.filter(|t| !t.is_empty())
.map(str::to_string)
}
fn origin_of(url: &str) -> Option<String> {
let scheme_end = url.find("://")?;
let after = &url[scheme_end + 3..];
let authority_end = after.find('/').unwrap_or(after.len());
Some(url[..scheme_end + 3 + authority_end].to_ascii_lowercase())
}
fn resolve_next_url(base: &str, target: &str) -> Option<String> {
let origin = origin_of(base)?;
let absolute = if target.contains("://") {
target.to_string()
} else if let Some(path) = target.strip_prefix('/') {
format!("{origin}/{path}")
} else {
let path_base = base.split(['?', '#']).next().unwrap_or(base);
let cut = path_base.rfind('/')?;
format!("{}/{target}", &path_base[..cut])
};
(origin_of(&absolute)? == origin).then_some(absolute)
}
fn parse_link_next(link: &str, current: &str) -> Option<String> {
let mut segments = Vec::new();
let mut current_segment = String::new();
let mut depth = 0i32;
for c in link.chars() {
match c {
'<' => {
depth += 1;
current_segment.push(c);
}
'>' => {
depth -= 1;
current_segment.push(c);
}
',' if depth == 0 => segments.push(std::mem::take(&mut current_segment)),
_ => current_segment.push(c),
}
}
segments.push(current_segment);
for segment in segments {
let segment = segment.trim();
let Some(open) = segment.find('<') else {
continue;
};
let Some(close) = segment[open..].find('>').map(|i| open + i) else {
continue;
};
let is_next = segment[close + 1..].split(';').any(|param| {
param
.split_once('=')
.is_some_and(|(k, v)| k.trim().eq_ignore_ascii_case("rel") && rel_is_next(v))
});
if is_next {
return resolve_next_url(current, segment[open + 1..close].trim());
}
}
None
}
fn rel_is_next(value: &str) -> bool {
value
.trim()
.trim_matches('"')
.split_whitespace()
.any(|r| r.eq_ignore_ascii_case("next"))
}
fn tags_first_page_url(base: &str) -> String {
format!("{base}/tags/list?n={TAGS_PAGE_SIZE}")
}
fn tags_from_page(bytes: &[u8]) -> Result<Vec<String>, SourceError> {
let json: serde_json::Value = serde_json::from_slice(bytes)
.map_err(|e| SourceError::Transport(format!("tags/list: {e}")))?;
Ok(json["tags"]
.as_array()
.map(|tags| {
tags.iter()
.filter_map(|t| t.as_str().map(str::to_string))
.collect()
})
.unwrap_or_default())
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RegistryRef {
pub registry: String,
pub repository: String,
pub scheme: String,
}
impl RegistryRef {
pub fn parse(reference: &str) -> Result<Self, SourceError> {
let (scheme, rest) = if let Some(rest) = reference.strip_prefix("oci://") {
("https", rest)
} else if let Some(rest) = reference.strip_prefix("oci+http://") {
("http", rest)
} else {
return Err(SourceError::Transport(format!(
"'{reference}' is not an oci:// reference"
)));
};
let (registry, repository) = rest.split_once('/').ok_or_else(|| {
SourceError::Transport(format!("'{reference}' has no repository path"))
})?;
if registry.is_empty() || repository.is_empty() {
return Err(SourceError::Transport(format!(
"'{reference}' has an empty registry or repository"
)));
}
Ok(RegistryRef {
registry: registry.to_string(),
repository: repository.trim_end_matches('/').to_string(),
scheme: scheme.to_string(),
})
}
}
fn agent_config() -> ureq::config::Config {
ureq::Agent::config_builder()
.redirect_auth_headers(ureq::config::RedirectAuthHeaders::Never)
.http_status_as_error(false)
.build()
}
struct Fetched {
status: u16,
bytes: Vec<u8>,
link: Option<String>,
challenge: Option<String>,
}
pub struct RegistrySource {
reference: RegistryRef,
agent: ureq::Agent,
token: RefCell<Option<String>>,
credential: OnceCell<CredentialLookup>,
}
impl std::fmt::Debug for RegistrySource {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("RegistrySource")
.field("reference", &self.reference)
.field(
"token",
&self
.token
.borrow()
.as_ref()
.map(|_| "<redacted bearer token>"),
)
.field("credential", &self.credential)
.finish()
}
}
impl RegistrySource {
pub fn new(reference: RegistryRef) -> Self {
RegistrySource {
reference,
agent: ureq::Agent::new_with_config(agent_config()),
token: RefCell::new(None),
credential: OnceCell::new(),
}
}
pub fn parse(reference: &str) -> Result<Self, SourceError> {
Ok(Self::new(RegistryRef::parse(reference)?))
}
pub fn with_credential(self, username: &str, password: &str) -> Self {
let _ = self.credential.set(CredentialLookup::Found(Credential {
username: username.to_string(),
password: password.to_string(),
origin: "the credential supplied to RegistrySource::with_credential".to_string(),
}));
self
}
fn credential(&self) -> &CredentialLookup {
self.credential
.get_or_init(|| resolve_credential(&self.reference.registry))
}
fn base(&self) -> String {
format!(
"{}://{}/v2/{}",
self.reference.scheme, self.reference.registry, self.reference.repository
)
}
fn send(&self, url: &str, accept: &str, token: Option<&str>) -> Result<Fetched, SourceError> {
let mut request = self.agent.get(url).header("Accept", accept);
if let Some(token) = token {
request = request.header("Authorization", &format!("Bearer {token}"));
}
let mut response = request
.call()
.map_err(|e| SourceError::Transport(e.to_string()))?;
let status = response.status().as_u16();
let header = |name: &str| {
response
.headers()
.get(name)
.and_then(|v| v.to_str().ok())
.map(str::to_string)
};
let link = header("link");
let challenge = header("www-authenticate");
let bytes = response
.body_mut()
.with_config()
.limit(MAX_BODY_BYTES)
.read_to_vec()
.map_err(|e| SourceError::Transport(e.to_string()))?;
Ok(Fetched {
status,
bytes,
link,
challenge,
})
}
fn obtain_token(&self, challenge: &BearerChallenge) -> Result<String, SourceError> {
let default_scope = format!("repository:{}:pull", self.reference.repository);
let url = token_url(challenge, &default_scope).ok_or_else(|| {
SourceError::Transport(format!(
"{} demanded authentication but its WWW-Authenticate challenge names no realm, \
so varve has no token endpoint to ask",
self.reference.registry
))
})?;
if !realm_is_acceptable(&url, &self.reference.scheme) {
return Err(SourceError::Transport(format!(
"{} is an https registry but points its token realm at {url}; varve will not \
send a credential over cleartext",
self.reference.registry
)));
}
let mut request = self.agent.get(&url).header("Accept", "application/json");
if let CredentialLookup::Found(credential) = self.credential() {
request = request.header("Authorization", &credential.basic_header());
}
let mut response = request
.call()
.map_err(|e| SourceError::Transport(format!("token request to {url} failed: {e}")))?;
let status = response.status().as_u16();
if status == 401 || status == 403 {
return Err(self.auth_error(&format!("the token endpoint {url}"), status));
}
if !(200..300).contains(&status) {
return Err(SourceError::Transport(format!(
"token endpoint {url} returned HTTP {status}"
)));
}
let body = response
.body_mut()
.with_config()
.limit(MAX_TOKEN_BYTES)
.read_to_string()
.map_err(|e| SourceError::Transport(format!("token response: {e}")))?;
token_from_body(&body).ok_or_else(|| {
SourceError::Transport(format!(
"token endpoint {url} answered HTTP {status} with no `token` field"
))
})
}
fn auth_error(&self, what: &str, status: u16) -> SourceError {
SourceError::Transport(format!(
"{} refused access to {} at {what} (HTTP {status}). {}",
self.reference.registry,
self.reference.repository,
credential_advice(
self.credential(),
&self.reference.registry,
&self.reference.repository
)
))
}
fn fetch(&self, url: &str, accept: &str) -> Result<Fetched, SourceError> {
let cached = self.token.borrow().clone();
let first = self.send(url, accept, cached.as_deref())?;
if first.status != 401 {
return Ok(first);
}
let challenge = first
.challenge
.as_deref()
.and_then(parse_bearer_challenge)
.ok_or_else(|| {
SourceError::Transport(format!(
"{} answered HTTP 401 for {url} with no Bearer challenge varve could parse \
({}), so there is no token endpoint to ask. {}",
self.reference.registry,
match &first.challenge {
Some(header) => format!("WWW-Authenticate: {header}"),
None => "no WWW-Authenticate header".to_string(),
},
credential_advice(
self.credential(),
&self.reference.registry,
&self.reference.repository
)
))
})?;
let token = self.obtain_token(&challenge)?;
*self.token.borrow_mut() = Some(token.clone());
let second = self.send(url, accept, Some(&token))?;
if second.status == 401 {
return Err(self.auth_error(url, second.status));
}
Ok(second)
}
fn get_checked(&self, url: &str, accept: &str) -> Result<Fetched, SourceError> {
let fetched = self.fetch(url, accept)?;
match fetched.status {
200..=299 => Ok(fetched),
404 => Err(SourceError::NotFound(url.to_string())),
status => Err(SourceError::Transport(format!(
"{url} returned HTTP {status}"
))),
}
}
fn get(&self, url: &str, accept: &str) -> Result<Vec<u8>, SourceError> {
Ok(self.get_checked(url, accept)?.bytes)
}
fn artifact_manifest_for_tag(&self, tag: &str) -> Result<serde_json::Value, SourceError> {
let manifest_bytes =
self.get(&format!("{}/manifests/{tag}", self.base()), MANIFEST_ACCEPT)?;
serde_json::from_slice(&manifest_bytes)
.map_err(|e| SourceError::Transport(format!("artifact manifest: {e}")))
}
fn envelope_for_tag(&self, tag: &str) -> Result<Vec<u8>, SourceError> {
let manifest = self.artifact_manifest_for_tag(tag)?;
let envelope_digest = layer_digest_for_role(&manifest, ROLE_ENVELOPE).ok_or_else(|| {
SourceError::NotFound(format!("tag {tag} carries no varve envelope layer"))
})?;
self.fetch_blob(&envelope_digest)
}
fn line_status_for_tag(&self, tag: &str) -> Result<Option<Vec<u8>>, SourceError> {
let manifest = self.artifact_manifest_for_tag(tag)?;
match layer_digest_for_role(&manifest, ROLE_LINE_STATUS) {
Some(digest) => self.fetch_blob(&digest).map(Some),
None => Ok(None),
}
}
fn line_index_for_tag(&self, tag: &str) -> Result<Option<Vec<u8>>, SourceError> {
let manifest = match self.artifact_manifest_for_tag(tag) {
Ok(manifest) => manifest,
Err(SourceError::NotFound(_)) => return Ok(None),
Err(e) => return Err(e),
};
match layer_digest_for_role(&manifest, ROLE_LINE_INDEX) {
Some(digest) => self.fetch_blob(&digest).map(Some),
None => Ok(None),
}
}
fn attestations_for_tag(
&self,
tag: &str,
) -> Result<Vec<crate::attestcarry::CarriedAttestation>, SourceError> {
let manifest = self.artifact_manifest_for_tag(tag)?;
let Some(layers) = manifest["layers"].as_array() else {
return Ok(Vec::new());
};
let mut out = Vec::new();
for l in layers
.iter()
.filter(|l| l["annotations"][ANN_ROLE] == ROLE_ATTESTATION_STATEMENT)
{
let Some(st_digest) = l["digest"].as_str() else {
continue;
};
let bytes_digest = layers
.iter()
.find(|b| {
b["annotations"][ANN_ROLE] == ROLE_ATTESTATION_BYTES
&& b["annotations"][crate::attestcarry::ANN_STATEMENT] == *st_digest
})
.and_then(|b| b["digest"].as_str())
.ok_or_else(|| {
SourceError::NotFound(format!(
"tag {tag} carries attestation statement {st_digest} but the manifest \
references no bytes for it — the claim travelled and the evidence \
did not"
))
})?;
out.push(crate::attestcarry::CarriedAttestation {
statement_digest: st_digest.to_string(),
statement: self.fetch_blob(st_digest)?,
bytes: self.fetch_blob(bytes_digest)?,
});
}
out.sort_by(|a, b| a.statement_digest.cmp(&b.statement_digest));
Ok(out)
}
fn tags(&self) -> Result<Vec<String>, SourceError> {
let mut url = tags_first_page_url(&self.base());
let mut out = Vec::new();
for _ in 0..MAX_TAG_PAGES {
let page = self.get_checked(&url, "application/json")?;
out.extend(tags_from_page(&page.bytes)?);
let next = page
.link
.as_deref()
.and_then(|link| parse_link_next(link, &url));
let Some(next) = next else {
return Ok(out);
};
if next == url {
return Err(SourceError::Transport(format!(
"{url} answered with a Link rel=\"next\" pointing at the page it came from; \
refusing to loop"
)));
}
url = next;
}
Err(SourceError::Transport(format!(
"{}/tags/list was still handing out `Link: rel=\"next\"` after {MAX_TAG_PAGES} pages. \
varve stops rather than looping, and refuses to answer from a partial tag list — a \
short list would silently turn a digest pin into 'not found'.",
self.base()
)))
}
}
impl LayerSource for RegistrySource {
fn fetch_manifest(&self, layer: &LayerRef) -> Result<Vec<u8>, SourceError> {
match layer {
LayerRef::Name(id) => self.envelope_for_tag(&id.to_string()),
LayerRef::Digest(digest) => {
for tag in self.tags()? {
if let Ok(envelope) = self.envelope_for_tag(&tag)
&& let Ok(text) = std::str::from_utf8(&envelope)
&& let Ok(env) = wsc::dsse::DsseEnvelope::from_json(text)
&& let Ok(payload) = env.payload_bytes()
&& &crate::store::manifest_digest(&payload) == digest
{
return Ok(envelope);
}
}
Err(SourceError::NotFound(digest.clone()))
}
}
}
fn fetch_line_status(&self, layer: &LayerRef) -> Result<Option<Vec<u8>>, SourceError> {
match layer {
LayerRef::Name(id) => self.line_status_for_tag(&id.to_string()),
LayerRef::Digest(digest) => {
for tag in self.tags()? {
if let Ok(envelope) = self.envelope_for_tag(&tag)
&& let Ok(text) = std::str::from_utf8(&envelope)
&& let Ok(env) = wsc::dsse::DsseEnvelope::from_json(text)
&& let Ok(payload) = env.payload_bytes()
&& &crate::store::manifest_digest(&payload) == digest
{
return self.line_status_for_tag(&tag);
}
}
Ok(None)
}
}
}
fn fetch_line_index(&self, line: &str) -> Result<Option<Vec<u8>>, SourceError> {
self.line_index_for_tag(&crate::lineindex::index_tag(line))
}
fn served_layers(&self, line: &str) -> Result<Option<Vec<String>>, SourceError> {
Ok(Some(layers_of_line(self.tags()?, line)))
}
fn fetch_attestations(
&self,
layer: &LayerRef,
) -> Result<Vec<crate::attestcarry::CarriedAttestation>, SourceError> {
match layer {
LayerRef::Name(id) => self.attestations_for_tag(&id.to_string()),
LayerRef::Digest(digest) => {
for tag in self.tags()? {
if let Ok(envelope) = self.envelope_for_tag(&tag)
&& let Ok(text) = std::str::from_utf8(&envelope)
&& let Ok(env) = wsc::dsse::DsseEnvelope::from_json(text)
&& let Ok(payload) = env.payload_bytes()
&& &crate::store::manifest_digest(&payload) == digest
{
return self.attestations_for_tag(&tag);
}
}
Ok(Vec::new())
}
}
}
fn fetch_blob(&self, digest: &str) -> Result<Vec<u8>, SourceError> {
let bytes = self.get(
&format!("{}/blobs/{digest}", self.base()),
"application/octet-stream",
)?;
if crate::store::manifest_digest(&bytes) != digest {
return Err(SourceError::Transport(format!(
"registry returned wrong bytes for {digest}"
)));
}
Ok(bytes)
}
}
#[cfg(test)]
mod tests {
use super::*;
const SECRET: &str = "s3cr3t-do-not-log";
#[test]
fn oci_references_parse_and_bad_ones_are_refused() {
let r = RegistryRef::parse("oci://ghcr.io/pulseengine/layers").unwrap();
assert_eq!(r.registry, "ghcr.io");
assert_eq!(r.repository, "pulseengine/layers");
assert_eq!(r.scheme, "https");
let t = RegistryRef::parse("oci+http://127.0.0.1:5000/test/repo").unwrap();
assert_eq!(t.scheme, "http");
assert_eq!(t.registry, "127.0.0.1:5000");
for bad in [
"https://ghcr.io/x",
"oci://",
"oci://hostonly",
"oci://host/",
] {
assert!(RegistryRef::parse(bad).is_err(), "{bad} must not parse");
}
}
#[test]
fn a_role_annotated_layer_digest_is_found_and_absence_is_none() {
let manifest = serde_json::json!({
"layers": [
{"digest": "sha256:aaa", "annotations": {ANN_ROLE: ROLE_ENVELOPE}},
{"digest": "sha256:bbb", "annotations": {ANN_ROLE: ROLE_PAYLOAD}},
{"digest": "sha256:ccc", "annotations": {ANN_ROLE: ROLE_LINE_STATUS}},
]
});
assert_eq!(
layer_digest_for_role(&manifest, ROLE_LINE_STATUS),
Some("sha256:ccc".to_string()),
"the baseline line-status layer must be found by its role"
);
assert_eq!(
layer_digest_for_role(&manifest, ROLE_ENVELOPE),
Some("sha256:aaa".to_string())
);
let bare = serde_json::json!({
"layers": [{"digest": "sha256:aaa", "annotations": {ANN_ROLE: ROLE_ENVELOPE}}]
});
assert_eq!(layer_digest_for_role(&bare, ROLE_LINE_STATUS), None);
assert_ne!(ROLE_LINE_INDEX, ROLE_LINE_STATUS);
assert_eq!(layer_digest_for_role(&manifest, ROLE_LINE_INDEX), None);
let indexed = serde_json::json!({
"layers": [
{"digest": "sha256:ccc", "annotations": {ANN_ROLE: ROLE_LINE_STATUS}},
{"digest": "sha256:ddd", "annotations": {ANN_ROLE: ROLE_LINE_INDEX}},
]
});
assert_eq!(
layer_digest_for_role(&indexed, ROLE_LINE_INDEX),
Some("sha256:ddd".to_string())
);
}
#[test]
fn a_registrys_listing_for_a_line_is_that_lines_layers_and_nothing_else() {
let tags = vec![
"2026.08.0".to_string(),
"2026.08.10".to_string(),
"2026.09.0".to_string(), "line-index-2026.08".to_string(), "latest".to_string(), "2026.08.01".to_string(), "2026.08".to_string(), ];
assert_eq!(
layers_of_line(tags.clone(), "2026.08"),
vec!["2026.08.0".to_string(), "2026.08.10".to_string()],
);
assert_eq!(
layers_of_line(tags, "2026.09"),
vec!["2026.09.0".to_string()]
);
assert!(layers_of_line(vec!["latest".to_string()], "2026.08").is_empty());
}
#[test]
fn a_bearer_challenge_yields_realm_service_and_scope() {
let c = parse_bearer_challenge(
r#"Bearer realm="https://auth.example.test/token",service="registry.example.test",scope="repository:org/repo:pull""#,
)
.expect("a Bearer challenge must parse");
assert_eq!(c.realm.as_deref(), Some("https://auth.example.test/token"));
assert_eq!(c.service.as_deref(), Some("registry.example.test"));
assert_eq!(c.scope.as_deref(), Some("repository:org/repo:pull"));
let c =
parse_bearer_challenge(r#"Bearer realm="https://a/t",scope="repository:x:pull,push""#)
.unwrap();
assert_eq!(
c.scope.as_deref(),
Some("repository:x:pull,push"),
"a quoted scope must survive its own commas"
);
let c = parse_bearer_challenge("bearer realm=https://a/t, service=reg").unwrap();
assert_eq!(c.realm.as_deref(), Some("https://a/t"));
assert_eq!(c.service.as_deref(), Some("reg"));
assert_eq!(parse_bearer_challenge(r#"Basic realm="x""#), None);
assert_eq!(
parse_bearer_challenge("Bearer"),
Some(BearerChallenge::default())
);
}
#[test]
fn the_token_url_comes_from_the_realm_the_registry_named() {
let c = parse_bearer_challenge(
r#"Bearer realm="https://auth.example.test/v1/token",service="reg.example.test""#,
)
.unwrap();
let url = token_url(&c, "repository:fallback:pull").unwrap();
assert!(
url.starts_with("https://auth.example.test/v1/token?"),
"the realm decides the endpoint, not a hardcoded /token: {url}"
);
assert!(url.contains("service=reg.example.test"), "{url}");
assert!(
url.contains("scope=repository%3Afallback%3Apull"),
"an absent scope falls back to a pull scope for the repository: {url}"
);
let c = parse_bearer_challenge(r#"Bearer realm="https://gl.test/jwt/auth?x=1""#).unwrap();
let url = token_url(&c, "repository:r:pull").unwrap();
assert!(url.starts_with("https://gl.test/jwt/auth?x=1&"), "{url}");
assert_eq!(url.matches('?').count(), 1, "{url}");
assert_eq!(token_url(&BearerChallenge::default(), "s"), None);
assert_eq!(
token_url(
&BearerChallenge {
realm: Some(" ".into()),
..Default::default()
},
"s"
),
None
);
}
#[test]
fn an_https_registry_may_not_redirect_its_token_realm_to_cleartext() {
assert!(realm_is_acceptable(
"https://auth.example.test/token",
"https"
));
assert!(
!realm_is_acceptable("http://auth.example.test/token", "https"),
"an https registry must not talk varve into posting Basic over http"
);
assert!(realm_is_acceptable("http://127.0.0.1:5000/token", "http"));
assert!(realm_is_acceptable("https://127.0.0.1:5000/token", "http"));
assert!(!realm_is_acceptable("ftp://x/token", "http"));
}
#[test]
fn a_token_response_is_read_from_either_spelling() {
assert_eq!(
token_from_body(r#"{"token":"abc"}"#).as_deref(),
Some("abc")
);
assert_eq!(
token_from_body(r#"{"access_token":"xyz"}"#).as_deref(),
Some("xyz"),
"the OAuth2 spelling several registries answer with"
);
assert_eq!(token_from_body(r#"{"token":""}"#), None);
assert_eq!(token_from_body(r#"{"nope":1}"#), None);
assert_eq!(token_from_body("not json"), None);
}
#[test]
fn base64_round_trips_and_decodes_a_docker_auth_field() {
for input in [
"".as_bytes(),
b"a",
b"ab",
b"abc",
b"user:pass",
b"\x00\xff\xfe\x01",
] {
assert_eq!(
base64_decode(&base64_encode(input)).as_deref(),
Some(input),
"round trip"
);
}
assert_eq!(base64_encode(b"user:pass"), "dXNlcjpwYXNz");
assert_eq!(
decode_basic_auth("dXNlcjpwYXNz"),
Some(("user".to_string(), "pass".to_string()))
);
assert_eq!(
decode_basic_auth("dXNlcjpwYXNz\n"),
Some(("user".to_string(), "pass".to_string()))
);
assert_eq!(
decode_basic_auth(&base64_encode(b"user:a:b")),
Some(("user".to_string(), "a:b".to_string()))
);
assert_eq!(base64_decode("not base64!"), None);
assert_eq!(decode_basic_auth(&base64_encode(b"nocolon")), None);
assert_eq!(decode_basic_auth(&base64_encode(b":onlypass")), None);
}
#[test]
fn a_docker_config_auths_entry_becomes_a_credential() {
let config = serde_json::json!({
"auths": {
"ghcr.io": { "auth": base64_encode(format!("alice:{SECRET}").as_bytes()) }
}
});
match credential_from_docker_config(&config, "ghcr.io", "/cfg") {
CredentialLookup::Found(c) => {
assert_eq!(c.username, "alice");
assert_eq!(c.password, SECRET);
assert_eq!(c.origin, "/cfg");
}
other => panic!("expected a credential, got {other:?}"),
}
let config = serde_json::json!({
"auths": { "https://index.docker.io/v1/": { "auth": base64_encode(b"bob:pw") } }
});
assert!(matches!(
credential_from_docker_config(&config, "registry-1.docker.io", "/cfg"),
CredentialLookup::Found(_)
));
let config = serde_json::json!({
"auths": { "reg.test": { "username": "carol", "password": SECRET } }
});
match credential_from_docker_config(&config, "reg.test", "/cfg") {
CredentialLookup::Found(c) => assert_eq!(c.username, "carol"),
other => panic!("expected a credential, got {other:?}"),
}
assert_eq!(
credential_from_docker_config(&config, "other.test", "/cfg"),
CredentialLookup::Absent
);
let config = serde_json::json!({ "auths": { "reg.test": { "auth": "%%%" } } });
assert!(matches!(
credential_from_docker_config(&config, "reg.test", "/cfg"),
CredentialLookup::Malformed { .. }
));
}
#[test]
fn a_credential_helper_is_named_and_never_run() {
let config = serde_json::json!({ "credsStore": "osxkeychain" });
assert_eq!(
credential_from_docker_config(&config, "ghcr.io", "~/.docker/config.json"),
CredentialLookup::HelperOnly {
helper: "osxkeychain".to_string(),
origin: "~/.docker/config.json".to_string()
},
"a credsStore-only config must be reported, not executed"
);
let config = serde_json::json!({ "credHelpers": { "ghcr.io": "ghcr-login" } });
assert_eq!(
credential_from_docker_config(&config, "ghcr.io", "/cfg"),
CredentialLookup::HelperOnly {
helper: "ghcr-login".to_string(),
origin: "/cfg".to_string()
}
);
let config = serde_json::json!({ "credHelpers": { "other.test": "h" } });
assert_eq!(
credential_from_docker_config(&config, "ghcr.io", "/cfg"),
CredentialLookup::Absent
);
let config = serde_json::json!({
"credsStore": "osxkeychain",
"auths": { "ghcr.io": { "auth": base64_encode(b"alice:pw") } }
});
assert!(matches!(
credential_from_docker_config(&config, "ghcr.io", "/cfg"),
CredentialLookup::Found(_)
));
}
#[test]
fn the_environment_variable_is_a_username_colon_password_pair() {
match credential_from_env_value(&format!("alice:{SECRET}")) {
CredentialLookup::Found(c) => {
assert_eq!(c.username, "alice");
assert_eq!(c.password, SECRET);
assert_eq!(c.origin, "$VARVE_REGISTRY_AUTH");
}
other => panic!("expected a credential, got {other:?}"),
}
match credential_from_env_value("AWS:token-value\n") {
CredentialLookup::Found(c) => assert_eq!(c.password, "token-value"),
other => panic!("expected a credential, got {other:?}"),
}
assert_eq!(credential_from_env_value(""), CredentialLookup::Absent);
assert!(matches!(
credential_from_env_value("no-colon-here"),
CredentialLookup::Malformed { .. }
));
assert!(matches!(
credential_from_env_value(":only-password"),
CredentialLookup::Malformed { .. }
));
}
#[test]
fn precedence_prefers_a_real_credential_and_otherwise_keeps_the_explanation() {
let found = CredentialLookup::Found(Credential {
username: "a".into(),
password: "b".into(),
origin: "second".into(),
});
let helper = CredentialLookup::HelperOnly {
helper: "h".into(),
origin: "first".into(),
};
assert_eq!(
first_usable(vec![helper.clone(), found.clone()]),
found,
"a usable credential wins wherever it is found"
);
let first_found = CredentialLookup::Found(Credential {
username: "z".into(),
password: "b".into(),
origin: "first".into(),
});
assert_eq!(
first_usable(vec![first_found.clone(), found.clone()]),
first_found
);
assert_eq!(
first_usable(vec![CredentialLookup::Absent, helper.clone()]),
helper
);
assert_eq!(first_usable(vec![]), CredentialLookup::Absent);
}
#[test]
fn config_files_are_read_in_order_and_a_broken_one_is_skipped() {
let tmp = tempfile::tempdir().unwrap();
let broken = tmp.path().join("broken.json");
std::fs::write(&broken, "{ not json").unwrap();
let good = tmp.path().join("good.json");
std::fs::write(
&good,
serde_json::to_vec(&serde_json::json!({
"auths": { "reg.test": { "auth": base64_encode(format!("dave:{SECRET}").as_bytes()) } }
}))
.unwrap(),
)
.unwrap();
let missing = tmp.path().join("absent.json");
let lookups = lookups_from_paths(&[missing, broken, good], "reg.test");
assert_eq!(
lookups.len(),
1,
"a missing and an unparseable config contribute nothing, they do not fail the pull"
);
match first_usable(lookups) {
CredentialLookup::Found(c) => assert_eq!(c.username, "dave"),
other => panic!("expected the good config's credential, got {other:?}"),
}
}
#[test]
fn a_credential_never_reaches_a_debug_line_or_an_error_message() {
let credential = Credential {
username: "alice".into(),
password: SECRET.into(),
origin: "/home/u/.docker/config.json".into(),
};
let debug = format!("{credential:?}");
assert!(
!debug.contains(SECRET),
"Debug leaked the password: {debug}"
);
assert!(
!debug.contains("alice"),
"Debug leaked the username: {debug}"
);
assert!(debug.contains("/home/u/.docker/config.json"), "{debug}");
let lookup = CredentialLookup::Found(credential.clone());
let debug = format!("{lookup:?}");
assert!(!debug.contains(SECRET), "{debug}");
let advice = credential_advice(&lookup, "ghcr.io", "org/repo");
assert!(!advice.contains(SECRET), "advice leaked the password");
assert!(
advice.contains("/home/u/.docker/config.json"),
"the advice must name where the rejected credential came from: {advice}"
);
assert_eq!(
credential.basic_header(),
format!(
"Basic {}",
base64_encode(format!("alice:{SECRET}").as_bytes())
)
);
let source = RegistrySource::parse("oci://ghcr.io/org/repo")
.unwrap()
.with_credential("alice", SECRET);
*source.token.borrow_mut() = Some("issued-bearer-token".to_string());
let debug = format!("{source:?}");
assert!(
!debug.contains("issued-bearer-token"),
"RegistrySource Debug leaked the bearer token: {debug}"
);
assert!(
debug.contains("ghcr.io"),
"the reference is not a secret and must stay legible: {debug}"
);
assert!(
!debug.contains(SECRET),
"RegistrySource Debug leaked the password: {debug}"
);
}
#[test]
fn a_refusal_distinguishes_no_credential_from_a_rejected_one() {
let rejected = credential_advice(
&CredentialLookup::Found(Credential {
username: "alice".into(),
password: SECRET.into(),
origin: "$VARVE_REGISTRY_AUTH".into(),
}),
"ghcr.io",
"org/repo",
);
assert!(
rejected.contains("rejected it"),
"a rejected credential must be named as rejected: {rejected}"
);
assert!(!rejected.contains("offered no credential"), "{rejected}");
for lookup in [
CredentialLookup::Absent,
CredentialLookup::Malformed {
origin: "$VARVE_REGISTRY_AUTH".into(),
},
CredentialLookup::HelperOnly {
helper: "osxkeychain".into(),
origin: "~/.docker/config.json".into(),
},
] {
let advice = credential_advice(&lookup, "ghcr.io", "org/repo");
assert!(
advice.contains("offered no credential"),
"{lookup:?} must be reported as having offered nothing: {advice}"
);
assert!(
advice.contains(CREDENTIAL_ENV),
"every no-credential message must name the fix: {advice}"
);
}
let advice = credential_advice(
&CredentialLookup::HelperOnly {
helper: "osxkeychain".into(),
origin: "~/.docker/config.json".into(),
},
"ghcr.io",
"org/repo",
);
assert!(advice.contains("osxkeychain"), "{advice}");
assert!(
advice.contains("does not execute credential helpers"),
"{advice}"
);
assert!(advice.contains("REQ-SHADOW-001"), "{advice}");
}
#[test]
fn a_link_header_names_the_next_page_and_only_within_the_origin() {
let current = "https://reg.test/v2/org/repo/tags/list?n=100";
assert_eq!(
parse_link_next(
r#"</v2/org/repo/tags/list?n=100&last=2026.08.9>; rel="next""#,
current
)
.as_deref(),
Some("https://reg.test/v2/org/repo/tags/list?n=100&last=2026.08.9")
);
assert_eq!(
parse_link_next(
r#"</v2/a?x=1>; rel=prev, </v2/b?x=2>; type="text"; rel="next""#,
current
)
.as_deref(),
Some("https://reg.test/v2/b?x=2")
);
assert_eq!(
parse_link_next(r#"<https://reg.test/v2/next>; rel="next""#, current).as_deref(),
Some("https://reg.test/v2/next")
);
assert_eq!(
parse_link_next(r#"<https://evil.test/v2/next>; rel="next""#, current),
None,
"a rel=next pointing off-origin must not be followed"
);
assert_eq!(parse_link_next(r#"</v2/a>; rel="prev""#, current), None);
assert_eq!(parse_link_next("", current), None);
assert!(parse_link_next(r#"</v2/a>; rel="prev next""#, current).is_some());
}
#[test]
fn a_tags_page_is_parsed_and_a_broken_one_is_not_an_empty_repository() {
assert_eq!(
tags_from_page(br#"{"name":"r","tags":["a","b"]}"#).unwrap(),
vec!["a".to_string(), "b".to_string()]
);
assert_eq!(
tags_from_page(br#"{"name":"r","tags":null}"#).unwrap(),
Vec::<String>::new()
);
assert!(tags_from_page(b"<html>502</html>").is_err());
}
#[test]
fn the_first_tags_page_asks_the_registry_to_paginate() {
let url = tags_first_page_url("https://reg.test/v2/org/repo");
assert_eq!(
url,
format!("https://reg.test/v2/org/repo/tags/list?n={TAGS_PAGE_SIZE}")
);
assert!(
url.contains("?n="),
"without ?n= a registry may answer one implementation-defined page and \
the client never learns there was more: {url}"
);
assert_eq!(MAX_TAG_PAGES, 64);
}
#[test]
fn the_manifest_accept_header_offers_the_docker_type_as_well_as_the_oci_one() {
assert!(
MANIFEST_ACCEPT.contains("application/vnd.oci.image.manifest.v1+json"),
"{MANIFEST_ACCEPT}"
);
assert!(
MANIFEST_ACCEPT.contains("application/vnd.docker.distribution.manifest.v2+json"),
"a registry serving only the Docker type is unreachable without this: \
{MANIFEST_ACCEPT}"
);
}
#[test]
fn the_agent_never_carries_authorization_across_a_redirect() {
let config = agent_config();
assert_eq!(
config.redirect_auth_headers(),
ureq::config::RedirectAuthHeaders::Never,
"blob fetches redirect to CDNs; the credential must not go with them"
);
assert!(
!config.http_status_as_error(),
"a 401 must arrive as a response so its WWW-Authenticate challenge can be read"
);
}
#[test]
fn percent_encoding_escapes_what_a_scope_contains() {
assert_eq!(
percent_encode("repository:org/repo:pull"),
"repository%3Aorg%2Frepo%3Apull"
);
assert_eq!(percent_encode("a-b_c.d~e"), "a-b_c.d~e");
assert_eq!(percent_encode("a b"), "a%20b");
}
}