src/federation/mod.rs
//! UPSTREAM-REGISTRY FEDERATION — the *ship* (Skíðblaðnir) that sails between
//! **harbours**.
//!
//! The harbours are the external artifact registries and clouds — **JFrog
//! Artifactory**, **Sonatype Nexus**, **Azure Artifacts** (Azure DevOps feeds)
//! and **AWS CodeArtifact**. This module is the reusable *library expert*: one
//! [`UpstreamRegistry`] trait, one auth enum ([`UpstreamAuth`]), one artifact
//! metadata shape ([`ArtifactMeta`]) and one coordinate→layout mapping
//! ([`upstream_path`]) shared by every harbour impl. A consumer (holger) HIRES
//! the ship to mirror/clone artifacts out of an upstream and into its own store;
//! the consumer owns the orchestration/policy (which feeds, scheduling,
//! dedup-on-store, compression), the ship owns the *mechanism* (auth, list,
//! fetch, mirror).
//!
//! HTTP is **SYNC** via `ureq` (a shared, bounded-timeout agent) — runtime-
//! agnostic, so it is safe inside an async handler (where blocking on `reqwest`
//! would panic) and in plain unit tests, and it stays **pure-Rust** (rustls/ring,
//! no OpenSSL, no `aws-lc-rs`/C/cmake) per the nordisk pure-Rust charter. That is
//! also why the Azure + AWS harbours here speak their documented HTTP endpoints
//! directly with a PAT / bearer authorization token rather than vendoring the
//! async cloud SDKs (which pull a C/cmake toolchain): the auth *token* is minted
//! out-of-band by the operator's `az` / `aws` CLI exactly as npm/pip/cargo
//! consume it — no crypto is hand-rolled here.
#[cfg(feature = "federation")]
use std::io::Read;
#[cfg(feature = "federation")]
use std::sync::OnceLock;
#[cfg(feature = "federation")]
use std::time::Duration;
use anyhow::Result;
#[cfg(feature = "federation")]
use base64::Engine as _;
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
mod artifactory;
#[cfg(feature = "upstream-aws")]
pub mod aws;
#[cfg(feature = "upstream-azure")]
pub mod azure;
mod nexus;
mod oci;
#[cfg(all(test, feature = "federation"))]
pub(crate) mod testutil;
pub use artifactory::ArtifactoryRegistry;
#[cfg(feature = "upstream-aws")]
pub use aws::AwsCodeArtifactRegistry;
#[cfg(feature = "upstream-azure")]
pub use azure::AzureArtifactsRegistry;
pub use nexus::NexusRegistry;
pub use oci::OciDistributionRegistry;
/// Hard cap on a single artifact download (bytes). A malicious/compromised
/// upstream can otherwise stream an unbounded body and OOM the caller.
pub const MAX_DOWNLOAD_BYTES: u64 = 8 * 1024 * 1024 * 1024;
/// How a harbour authenticates. The single source of truth shared by every
/// [`UpstreamRegistry`] impl (Nexus = Basic; Artifactory = Basic/Bearer/API-key;
/// Azure = PAT-as-Basic or AAD Bearer; AWS CodeArtifact = Bearer authorization
/// token).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
pub enum UpstreamAuth {
/// No authentication (anonymous / public read).
#[default]
None,
/// HTTP Basic auth (Nexus, Artifactory, Azure PAT with empty/`pat` user).
Basic { username: String, password: String },
/// Bearer token (Artifactory access token, Azure AAD, AWS CodeArtifact).
Bearer { token: String },
/// Custom API-key header, e.g. Artifactory `X-JFrog-Art-Api-Key`.
ApiKey { header: String, key: String },
/// mTLS client cert — built into the HTTP client, not a per-request header,
/// so it adds no `Authorization` here (treated as no per-request auth).
Mtls { cert_pem: String, key_pem: String },
}
/// Size + checksum + content-type for an upstream artifact. Lets a consumer
/// verify a fetched body before caching and answer index/metadata probes without
/// downloading the body.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct ArtifactMeta {
pub size: Option<u64>,
pub sha256: Option<String>,
pub content_type: Option<String>,
}
/// An artifact coordinate — the harbour-neutral `(namespace?, name, version)`
/// address the layout mapping resolves to a per-registry content path.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ArtifactCoord {
/// Maven group id (`com.example`) or scope; `None` for flat ecosystems.
pub namespace: Option<String>,
pub name: String,
pub version: String,
}
impl ArtifactCoord {
pub fn new(name: impl Into<String>, version: impl Into<String>) -> Self {
Self {
namespace: None,
name: name.into(),
version: version.into(),
}
}
pub fn with_namespace(
namespace: impl Into<String>,
name: impl Into<String>,
version: impl Into<String>,
) -> Self {
Self {
namespace: Some(namespace.into()),
name: name.into(),
version: version.into(),
}
}
}
/// The package ecosystem / layout a harbour serves. Drives [`upstream_path`].
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum RegistryFormat {
Rust,
Maven,
Pip,
Npm,
Nuget,
Gem,
/// OCI Distribution v2 (container images / Helm OCI): `v2/{name}/manifests/{ref}`.
/// Served by [`OciDistributionRegistry`] (GHCR/Harbor/Quay/Docker Hub), which
/// resolves blobs by digest — a content-addressed store, not a file layout.
Oci,
/// Opaque `{name}/{version}` layout — the fallback for formats without a
/// dedicated mapping.
Raw,
}
/// Map an artifact coordinate to the layout path (relative to the repository
/// root) for `fmt`. This is the per-ecosystem translation the registries share;
/// each harbour joins it onto its own content base (`{base}/artifactory/{repo}/…`
/// for Artifactory, `{base}/repository/{repo}/…` for Nexus, …).
pub fn upstream_path(fmt: RegistryFormat, id: &ArtifactCoord) -> String {
match fmt {
RegistryFormat::Rust => format!("{n}/{n}-{v}.crate", n = id.name, v = id.version),
RegistryFormat::Maven => {
let g = id.namespace.as_deref().unwrap_or("").replace('.', "/");
format!("{g}/{a}/{v}/{a}-{v}.jar", a = id.name, v = id.version)
}
RegistryFormat::Pip => {
format!(
"packages/source/{n}/{n}-{v}.tar.gz",
n = id.name,
v = id.version
)
}
RegistryFormat::Npm => format!("{n}/-/{n}-{v}.tgz", n = id.name, v = id.version),
RegistryFormat::Nuget => {
let n = id.name.to_lowercase();
format!("{n}/{v}/{n}.{v}.nupkg", n = n, v = id.version)
}
RegistryFormat::Gem => format!("gems/{n}-{v}.gem", n = id.name, v = id.version),
// OCI Distribution v2 manifest path (repository = name, reference = version).
RegistryFormat::Oci => format!("v2/{n}/manifests/{v}", n = id.name, v = id.version),
RegistryFormat::Raw => format!("{}/{}", id.name, id.version),
}
}
/// A read-oriented remote-cache **upstream harbour**: knows its base URL + auth,
/// can fetch one artifact's bytes (verified against the upstream checksum when
/// advertised), can probe metadata cheaply, and can [`mirror`](UpstreamRegistry::mirror)
/// a set of coordinates into a caller-supplied sink.
pub trait UpstreamRegistry: Send + Sync {
/// The configured harbour name (a routing/identification key).
fn name(&self) -> &str;
/// The upstream base, e.g. `https://nexus.corp` or `https://pkgs.dev.azure.com/org`.
fn base_url(&self) -> &str;
/// Credentials attached to upstream requests.
fn auth(&self) -> &UpstreamAuth;
/// The ecosystem/layout this harbour serves.
fn format(&self) -> RegistryFormat;
/// Enumerate available artifacts. These are content-addressed HTTP registries
/// with no uniform cross-vendor "list everything" API, so enumeration is NOT
/// supported by default — a consumer drives [`mirror`](Self::mirror) with an
/// explicit coordinate set (from a lockfile / index it already resolved). An
/// honest error, never a silent empty list.
fn list(&self) -> Result<Vec<ArtifactCoord>> {
anyhow::bail!(
"{}: enumeration is not supported for this harbour; drive mirror() with an explicit coordinate set",
self.name()
)
}
/// Fetch one artifact's bytes. `Ok(None)` on a clean upstream 404. When the
/// upstream advertises a sha256 the body is verified before it is returned
/// (a cache-poisoning guard; a mismatch is a hard `Err`).
fn fetch(&self, id: &ArtifactCoord) -> Result<Option<Vec<u8>>>;
/// Cheap existence + `size`/`sha256`/`content-type` probe (a HEAD where the
/// harbour supports one). `Ok(None)` when the harbour has no such artifact.
fn metadata(&self, id: &ArtifactCoord) -> Result<Option<ArtifactMeta>>;
/// Existence probe. Default: `metadata(id).is_some()`.
fn exists(&self, id: &ArtifactCoord) -> Result<bool> {
Ok(self.metadata(id)?.is_some())
}
/// Mirror the given coordinates: fetch each and hand the bytes to `sink`
/// (which a consumer uses to write-through / dedup / compress into its own
/// store). Missing artifacts (upstream 404) are skipped. Returns the number
/// of artifacts actually mirrored. This is the ship's primary voyage.
fn mirror(
&self,
ids: &[ArtifactCoord],
sink: &mut dyn FnMut(&ArtifactCoord, Vec<u8>) -> Result<()>,
) -> Result<usize> {
let mut mirrored = 0usize;
for id in ids {
if let Some(bytes) = self.fetch(id)? {
sink(id, bytes)?;
mirrored += 1;
}
}
Ok(mirrored)
}
}
// ---------------------------------------------------------------------------
// Shared SYNC HTTP client (ureq) + auth + verify — used by every harbour impl.
// ---------------------------------------------------------------------------
/// Shared `ureq` agent with bounded connect/read/write timeouts so a hung or
/// black-holed upstream fails fast instead of pinning the calling thread forever.
/// One connection pool, built lazily, shared process-wide.
///
/// ureq 3 (2026-10-02). Its timeouts are deadlines, not ureq 2's per-read socket
/// timeouts: connect 10 s, sending 30 s and waiting for the response head 30 s
/// keep the old bounds. ureq 3 has no idle-read timeout, so the agent carries its
/// own ([`IdleCap`], 2026-10-05): every wait for body bytes is capped at the same
/// 30 s ureq 2 gave a read, so a stream that stops delivering fails in 30 s,
/// while one that keeps moving runs on under the 2 h whole-body deadline — long
/// enough for a healthy [`MAX_DOWNLOAD_BYTES`] transfer, short of "forever".
/// A 4xx/5xx is a response here (`http_status_as_error(false)`), so the OCI 401
/// handshake can read `WWW-Authenticate`; [`status_err`] keeps ureq 2's
/// "≥ 400 is an error" line for every caller. No proxy from the environment and
/// at most 5 redirects, as ureq 2 (without `proxy-from-env`) did.
#[cfg(feature = "federation")]
pub(crate) fn agent() -> &'static ureq::Agent {
static AGENT: OnceLock<ureq::Agent> = OnceLock::new();
AGENT.get_or_init(|| build_agent(Duration::from_secs(10), Duration::from_secs(30)))
}
/// The ONE place the federation agent's policy is written. [`agent`] builds the
/// shared agent through it, and the tests build short-timeout agents through it,
/// so what the tests exercise is the policy that ships.
///
/// `read` bounds sending, waiting for the response head, AND every single wait
/// for more body bytes ([`IdleCap`]) — ureq 2's per-read meaning of the word. The
/// body as a whole is bounded separately, by the 2 h deadline.
#[cfg(feature = "federation")]
pub(crate) fn build_agent(connect: Duration, read: Duration) -> ureq::Agent {
use ureq::unversioned::resolver::DefaultResolver;
use ureq::unversioned::transport::{Connector as _, DefaultConnector};
let config = ureq::Agent::config_builder()
.timeout_connect(Some(connect))
.timeout_send_request(Some(read))
.timeout_send_body(Some(read))
.timeout_recv_response(Some(read))
.timeout_recv_body(Some(Duration::from_secs(2 * 60 * 60)))
.http_status_as_error(false)
.proxy(None)
.max_redirects(5)
.build();
ureq::Agent::with_parts(
config,
DefaultConnector::new().chain(IdleCap(read)),
DefaultResolver::default(),
)
}
/// **The idle bound ureq 3 does not have**, as a connector chained after ureq's
/// own (TCP, or rustls over TCP). It wraps the finished transport and caps every
/// `await_input` wait at the idle bound. ureq hands each wait the time left
/// until the governing deadline (`recv_body`: 2 h), and TCP sets that as the
/// socket read timeout, so a silent peer used to hold the read for up to 2 h; with
/// the cap, a wait that sees no byte in `idle` fails with ureq's
/// `Timeout(RecvBody)`, and a wait that does see one hands the next wait a fresh
/// `idle`. Writes pass through untouched (they have their own 30 s deadlines).
///
/// Built on ureq's `unversioned::transport` API, which ureq exempts from semver:
/// a ureq minor that changes it fails this crate's COMPILE, never its behaviour.
#[cfg(feature = "federation")]
#[derive(Debug)]
pub(crate) struct IdleCap(pub(crate) Duration);
#[cfg(feature = "federation")]
impl ureq::unversioned::transport::Connector<Box<dyn ureq::unversioned::transport::Transport>>
for IdleCap
{
type Out = IdleCapped;
fn connect(
&self,
_details: &ureq::unversioned::transport::ConnectionDetails,
chained: Option<Box<dyn ureq::unversioned::transport::Transport>>,
) -> std::result::Result<Option<IdleCapped>, ureq::Error> {
Ok(chained.map(|inner| IdleCapped { inner, idle: self.0 }))
}
}
/// A transport whose every wait for input is capped at `idle` — see [`IdleCap`].
#[cfg(feature = "federation")]
#[derive(Debug)]
pub(crate) struct IdleCapped {
inner: Box<dyn ureq::unversioned::transport::Transport>,
idle: Duration,
}
#[cfg(feature = "federation")]
impl ureq::unversioned::transport::Transport for IdleCapped {
fn buffers(&mut self) -> &mut dyn ureq::unversioned::transport::Buffers {
self.inner.buffers()
}
fn transmit_output(
&mut self,
amount: usize,
timeout: ureq::unversioned::transport::NextTimeout,
) -> std::result::Result<(), ureq::Error> {
self.inner.transmit_output(amount, timeout)
}
fn await_input(
&mut self,
timeout: ureq::unversioned::transport::NextTimeout,
) -> std::result::Result<bool, ureq::Error> {
let idle = ureq::unversioned::transport::time::Duration::from(self.idle);
// `..timeout`: ureq 3.5 added `per_read`/`per_write` to this (unversioned)
// struct; carrying the rest over compiles against 3.4 and 3.5 alike.
#[allow(clippy::needless_update)]
let next = ureq::unversioned::transport::NextTimeout {
after: timeout.after.min(idle),
reason: timeout.reason,
..timeout
};
self.inner.await_input(next)
}
fn is_open(&mut self) -> bool {
self.inner.is_open()
}
fn is_tls(&self) -> bool {
self.inner.is_tls()
}
}
/// A request on [`agent`] that has not been sent yet.
#[cfg(feature = "federation")]
pub(crate) type Request = ureq::RequestBuilder<ureq::typestate::WithoutBody>;
/// A response from [`agent`].
#[cfg(feature = "federation")]
pub(crate) type Response = ureq::http::Response<ureq::Body>;
/// A response header as text (`None` when absent or not ASCII).
#[cfg(feature = "federation")]
pub(crate) fn header(resp: &Response, name: &str) -> Option<String> {
resp.headers().get(name).and_then(|v| v.to_str().ok()).map(str::to_string)
}
/// The HTTP status when it is an error (≥ 400), as ureq 2's `Error::Status` drew it.
#[cfg(feature = "federation")]
pub(crate) fn status_err(resp: &Response) -> Option<u16> {
let code = resp.status().as_u16();
(code >= 400).then_some(code)
}
/// `true` iff `target` has the same scheme + host (+ port) as `base_url`. SSRF /
/// credential-exfiltration guard: credentials are attached ONLY when the request
/// targets the harbour's own origin, never an arbitrary host a redirect points at.
#[cfg_attr(not(feature = "federation"), allow(dead_code))]
pub(crate) fn same_origin(base_url: &str, target: &str) -> bool {
fn parts(u: &str) -> Option<(String, String, Option<u16>)> {
// scheme://host[:port]/... — minimal, dependency-free origin split.
let (scheme, rest) = u.split_once("://")?;
let authority = rest.split(['/', '?', '#']).next().unwrap_or("");
let authority = authority.rsplit('@').next().unwrap_or(authority); // drop userinfo
let (host, port) = match authority.rsplit_once(':') {
Some((h, p)) if p.chars().all(|c| c.is_ascii_digit()) && !p.is_empty() => {
(h.to_string(), p.parse::<u16>().ok())
}
_ => (authority.to_string(), None),
};
let default_port = match scheme {
"https" => 443,
"http" => 80,
_ => 0,
};
Some((
scheme.to_ascii_lowercase(),
host.to_ascii_lowercase(),
Some(port.unwrap_or(default_port)),
))
}
match (parts(base_url), parts(target)) {
(Some(b), Some(t)) => b == t,
_ => false,
}
}
/// Attach the harbour credential to `req` — but only when `url` is same-origin as
/// `base_url` (so creds never leak cross-origin). `Mtls`/`None` add no header.
#[cfg(feature = "federation")]
pub(crate) fn apply_auth(
base_url: &str,
url: &str,
auth: &UpstreamAuth,
req: Request,
) -> Request {
if !same_origin(base_url, url) {
return req;
}
match auth {
UpstreamAuth::None | UpstreamAuth::Mtls { .. } => req,
UpstreamAuth::Basic { username, password } => {
let token =
base64::engine::general_purpose::STANDARD.encode(format!("{username}:{password}"));
req.header("Authorization", &format!("Basic {token}"))
}
UpstreamAuth::Bearer { token } => req.header("Authorization", &format!("Bearer {token}")),
UpstreamAuth::ApiKey { header, key } => req.header(header.as_str(), key.as_str()),
}
}
/// Blocking authenticated GET. `Ok(None)` on 404, `Ok(Some((bytes, sha256)))` on
/// 200 — where `sha256` is the upstream's advertised `X-Checksum-Sha256` when
/// present. The body read is capped at [`MAX_DOWNLOAD_BYTES`].
#[cfg(feature = "federation")]
pub(crate) fn get_bytes(
base_url: &str,
url: &str,
auth: &UpstreamAuth,
) -> Result<Option<(Vec<u8>, Option<String>)>> {
get_bytes_on(agent(), base_url, url, auth)
}
/// [`get_bytes`] on a caller-supplied agent — the seam the timeout tests drive
/// with a short-timeout agent from [`build_agent`].
#[cfg(feature = "federation")]
pub(crate) fn get_bytes_on(
agent: &ureq::Agent,
base_url: &str,
url: &str,
auth: &UpstreamAuth,
) -> Result<Option<(Vec<u8>, Option<String>)>> {
let req = apply_auth(base_url, url, auth, agent.get(url));
let resp = req.call().map_err(|e| anyhow::anyhow!("upstream {url}: {e}"))?;
match status_err(&resp) {
None => {
let sha256 = header(&resp, "X-Checksum-Sha256");
let mut buf = Vec::new();
resp.into_body()
.into_reader()
.take(MAX_DOWNLOAD_BYTES)
.read_to_end(&mut buf)
.map_err(|e| anyhow::anyhow!("read upstream body {url}: {e}"))?;
Ok(Some((buf, sha256)))
}
Some(404) => Ok(None),
Some(code) => anyhow::bail!("upstream {url}: HTTP {code}"),
}
}
/// Blocking authenticated HEAD → [`ArtifactMeta`]. `Ok(None)` on 404.
#[cfg(feature = "federation")]
pub(crate) fn head_meta(
base_url: &str,
url: &str,
auth: &UpstreamAuth,
) -> Result<Option<ArtifactMeta>> {
let req = apply_auth(base_url, url, auth, agent().head(url));
let resp = req.call().map_err(|e| anyhow::anyhow!("upstream HEAD {url}: {e}"))?;
match status_err(&resp) {
None => Ok(Some(ArtifactMeta {
size: header(&resp, "Content-Length").and_then(|s| s.parse::<u64>().ok()),
sha256: header(&resp, "X-Checksum-Sha256"),
content_type: header(&resp, "Content-Type"),
})),
Some(404) => Ok(None),
Some(code) => anyhow::bail!("upstream HEAD {url}: HTTP {code}"),
}
}
/// Cache-poisoning guard: verify `bytes` against the `expected` upstream sha256
/// (hex, optionally `sha256:`-prefixed, any case/whitespace). `Err` on mismatch.
#[cfg_attr(not(feature = "federation"), allow(dead_code))]
pub(crate) fn verify_sha256(bytes: &[u8], expected: &str) -> Result<()> {
let want = expected
.trim()
.trim_start_matches("sha256:")
.to_ascii_lowercase();
let got = hex::encode(Sha256::digest(bytes));
if got == want {
Ok(())
} else {
anyhow::bail!("upstream sha256 mismatch: expected {want}, computed {got}")
}
}
#[cfg(test)]
mod tests {
#[cfg(feature = "federation")]
use super::testutil::{MockResponse, MockServer};
use super::*;
/// The SHARED `head_meta` (used by Artifactory/Nexus/Azure/AWS) reads all three
/// advertised fields off the HEAD response — size (`Content-Length`), sha256
/// (`X-Checksum-Sha256`) and content-type — now that the mock can emit headers.
/// One test proves the HEAD for every file-path harbour (they share this helper).
#[test]
#[cfg(feature = "federation")]
fn head_meta_reads_size_checksum_and_content_type() {
let server = MockServer::start_with(|_req| {
MockResponse::new(200, b"twelve-bytes".to_vec())
.with_header("X-Checksum-Sha256", "deadbeef")
.with_header("Content-Type", "application/java-archive")
});
let url = format!("{}/artifactory/repo/x", server.base_url);
let meta = head_meta(&server.base_url, &url, &UpstreamAuth::None)
.unwrap()
.expect("meta");
assert_eq!(meta.size, Some(b"twelve-bytes".len() as u64));
assert_eq!(meta.sha256.as_deref(), Some("deadbeef"));
assert_eq!(
meta.content_type.as_deref(),
Some("application/java-archive")
);
assert_eq!(server.requests()[0].method, "HEAD");
crate::functional_status(
"skidbladnir/federation",
"head_meta_reads_checksum_and_type",
meta.sha256.as_deref() == Some("deadbeef"),
"shared HEAD probe surfaces size + X-Checksum-Sha256 + Content-Type",
);
}
/// The SHARED `get_bytes` surfaces the upstream's advertised `X-Checksum-Sha256`
/// (the value each harbour's `fetch` verifies the body against) — the input to the
/// cache-poisoning guard, now assertable end-to-end.
#[test]
#[cfg(feature = "federation")]
fn get_bytes_surfaces_advertised_checksum() {
let body = b"real-bytes".to_vec();
let good = hex::encode(Sha256::digest(&body));
let good_c = good.clone();
let server = MockServer::start_with(move |_req| {
MockResponse::new(200, b"real-bytes".to_vec()).with_header("X-Checksum-Sha256", &good_c)
});
let url = format!("{}/artifactory/repo/x", server.base_url);
let (bytes, sha) = get_bytes(&server.base_url, &url, &UpstreamAuth::None)
.unwrap()
.expect("some bytes");
assert_eq!(bytes, body);
assert_eq!(sha.as_deref(), Some(good.as_str()));
// And that advertised checksum verifies against the body (the guard is happy).
assert!(verify_sha256(&bytes, &sha.unwrap()).is_ok());
}
#[test]
fn upstream_path_maps_each_ecosystem_layout() {
let mvn = ArtifactCoord::with_namespace("com.example", "lib", "2.1");
assert_eq!(
upstream_path(RegistryFormat::Maven, &mvn),
"com/example/lib/2.1/lib-2.1.jar"
);
let crate_id = ArtifactCoord::new("serde", "1.0.0");
assert_eq!(
upstream_path(RegistryFormat::Rust, &crate_id),
"serde/serde-1.0.0.crate"
);
assert_eq!(
upstream_path(RegistryFormat::Npm, &crate_id),
"serde/-/serde-1.0.0.tgz"
);
// OCI Distribution v2 manifest path (repository = name, reference = version).
let oci_id = ArtifactCoord::new("nordisk/holger", "1.2");
assert_eq!(
upstream_path(RegistryFormat::Oci, &oci_id),
"v2/nordisk/holger/manifests/1.2"
);
crate::functional_status(
"skidbladnir/federation",
"upstream_path_layouts",
upstream_path(RegistryFormat::Maven, &mvn) == "com/example/lib/2.1/lib-2.1.jar",
"maven/rust/npm layouts resolve",
);
}
#[test]
fn verify_sha256_matches_normalizes_and_rejects() {
// sha256("hello")
let expected = "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824";
assert!(verify_sha256(b"hello", expected).is_ok());
assert!(
verify_sha256(b"hello", &format!(" sha256:{} ", expected.to_uppercase())).is_ok()
);
assert!(verify_sha256(b"tampered", expected).is_err());
crate::functional_status(
"skidbladnir/federation",
"verify_sha256_guard",
verify_sha256(b"tampered", expected).is_err(),
"tampered bytes are a hard error (cache-poisoning guard)",
);
}
#[test]
fn same_origin_guards_cross_host_and_scheme_and_port() {
assert!(same_origin(
"https://nexus.corp",
"https://nexus.corp/repository/x/a"
));
assert!(same_origin(
"https://nexus.corp:443",
"https://nexus.corp/repository/x"
));
assert!(!same_origin(
"https://nexus.corp",
"https://evil.com/repository/x"
));
assert!(!same_origin(
"https://nexus.corp",
"http://nexus.corp/repository/x"
));
assert!(!same_origin(
"https://nexus.corp",
"https://nexus.corp:8443/x"
));
}
#[test]
#[cfg(feature = "federation")]
fn list_is_honest_unsupported_not_empty() {
let up = NexusRegistry::new(
"nx",
"https://nexus.corp",
"crates-proxy",
UpstreamAuth::None,
RegistryFormat::Rust,
);
assert!(
up.list().is_err(),
"list must be an honest error, never a silent empty vec"
);
}
/// A local upstream that answers ONE request with a 200 whose `Content-Length`
/// promises `promised` bytes, writes `chunks` with `gap` before each, and then
/// holds the socket open for `hold` without closing it. Returns the base URL.
#[cfg(feature = "federation")]
fn body_server(promised: usize, chunks: Vec<Vec<u8>>, gap: Duration, hold: Duration) -> String {
use std::io::Write as _;
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap();
std::thread::spawn(move || {
let Ok((mut sock, _)) = listener.accept() else { return };
let mut scratch = [0u8; 4096];
let _ = sock.read(&mut scratch);
let _ = write!(sock, "HTTP/1.1 200 OK\r\nContent-Length: {promised}\r\n\r\n");
let _ = sock.flush();
for c in chunks {
std::thread::sleep(gap);
if sock.write_all(&c).and_then(|()| sock.flush()).is_err() {
return;
}
}
std::thread::sleep(hold);
});
format!("http://{addr}")
}
/// **A body that stops arriving fails on the idle bound, not on the 2 h
/// whole-body deadline.** The upstream sends the head and 10 of 1000 promised
/// bytes, then goes silent with the socket open. ureq 3's only body bound was
/// `timeout_recv_body` (2 h), so before the idle cap this read sat blocked; the
/// test gives it 5 s against a 1 s idle bound.
#[test]
#[cfg(feature = "federation")]
fn a_stalled_body_fails_on_the_idle_bound_not_the_whole_body_deadline() {
let base = body_server(1000, vec![b"0123456789".to_vec()], Duration::ZERO, Duration::from_secs(30));
let url = format!("{base}/artifactory/repo/x");
let (tx, rx) = std::sync::mpsc::channel();
let started = std::time::Instant::now();
std::thread::spawn(move || {
let agent = build_agent(Duration::from_secs(1), Duration::from_secs(1));
let _ = tx.send(get_bytes_on(&agent, &base, &url, &UpstreamAuth::None).map(|_| ()));
});
let res = rx.recv_timeout(Duration::from_secs(5)).unwrap_or_else(|_| {
panic!(
"the body read was still blocked after {:?} on a stream that stopped after 10 of \
1000 bytes: nothing bounds an idle body but the 2 h whole-body deadline",
started.elapsed()
)
});
let err = res.expect_err("a stalled body must surface an error, never Ok");
let msg = format!("{err:#}");
assert!(msg.contains("read upstream body"), "the error names the body read: {msg}");
assert!(started.elapsed() < Duration::from_secs(5), "took {:?}", started.elapsed());
}
/// **A slow body that keeps moving is NOT cut.** Four bytes with a 1.4 s pause
/// before each: 5.6 s in all, nearly three times the 2 s `read`/idle bound,
/// never more than 70 % of it between two bytes. The bound is per wait, not
/// per body.
///
/// The pause is longer than 1 s on purpose. ureq 3.3.0 let the response-head
/// deadline (`timeout_recv_response`) run on into the body, and once it had
/// passed it handed every later wait 1 s (its `not_zero` floor): an upstream
/// that paused for more than a second after the first 30 s was cut, measured
/// here as `timeout: receive response` at ~2 s. ureq 3.4.1 fixed that
/// ("timeout budgets restarting and applying to later phases", #1194), which
/// is why `Cargo.toml` asks for it.
#[test]
#[cfg(feature = "federation")]
fn a_slow_but_moving_body_is_not_cut_by_the_idle_bound() {
let body = b"slow".to_vec();
let chunks = body.iter().map(|b| vec![*b]).collect();
let base = body_server(body.len(), chunks, Duration::from_millis(1400), Duration::ZERO);
let url = format!("{base}/artifactory/repo/x");
let agent = build_agent(Duration::from_secs(2), Duration::from_secs(2));
let started = std::time::Instant::now();
let (bytes, _) = get_bytes_on(&agent, &base, &url, &UpstreamAuth::None)
.expect("a body that keeps moving completes")
.expect("200");
assert_eq!(bytes, body);
assert!(
started.elapsed() > Duration::from_secs(2),
"the transfer outlived the idle bound (took {:?}), so the bound is idle, not total",
started.elapsed()
);
}
}