server/lib/src/pull_intake/ecosystems.rs
//! **npm, Maven and cargo on the pull-through seam**: what a request for each is
//! told, and the pull job's body for each. The same seam as pip and OCI: fetch,
//! [`IntakeChain::ingest_pulled`] as `pull-through:<source>`, the walk (scanners
//! and provenance at SCANNED, policy, approval), served only once AVAILABLE.
use super::{apk, crates, maven, now_secs, npm, Ask, PullIntake, RemoteKind, SourceRt, Want};
use crate::promote::intake::{ArtifactState, IntakeChain};
use serde::{Deserialize, Serialize};
use traits::ArtifactId;
use crate::lock::LockOrRecover as _;
/// **What an npm registry listed for one package** at its last pull (lane HOL-FIELD-TEST, the
/// npm half of P2): each version's manifest fields npm resolves a tree from, and its integrity.
/// The packument lists the versions that are not AVAILABLE here as deprecated, so an exact pin
/// can ask for one.
#[derive(Debug, Clone, Serialize, Deserialize)]
struct NpmListing {
source: String,
package: String,
at: i64,
versions: Vec<(String, serde_json::Value, String)>,
}
/// **A Maven plugin group's plugin list** as its source published it (lane HOL-FIELD-TEST P13).
#[derive(Debug, Clone, Serialize, Deserialize)]
struct MavenPluginGroup {
source: String,
group: String,
at: i64,
plugins: Vec<(String, String, String)>,
}
/// The record of the plugin groups the Maven sources listed ([`MavenPluginGroup`]).
const MAVEN_PLUGIN_GROUPS: &str = "maven-plugin-groups.jsonl";
/// **What a cargo index listed for one crate** at its last pull (lane HOL-FIELD-TEST, the cargo
/// half of P2): each unyanked version's line as the registry wrote it. Holger's index names the
/// ones that are not AVAILABLE here yanked, so a lock file can still take one.
#[derive(Debug, Clone, Serialize, Deserialize)]
struct CrateListing {
source: String,
name: String,
at: i64,
lines: Vec<(String, String)>,
}
/// The record of the crate indexes the cargo sources listed ([`CrateListing`]).
const CRATE_LISTINGS: &str = "crates-listings.jsonl";
/// A crate as its source names it: `(source, lowercase name)`.
type CrateKey = (String, String);
/// **The cargo records, read from disk once and kept by crate** (HOL-DOGFOOD-8). Every
/// request for a crate's index file used to read and parse the whole of
/// `crates-listings.jsonl` and `crates-index.jsonl`, once per request, a thousand requests at
/// once under `cargo fetch`. Now each file is read once, the first time it is asked, and kept
/// as the newest listing and the pulled lines of each crate; every append updates it.
#[derive(Default)]
pub(super) struct CrateRecords {
listings:
std::sync::Mutex<Option<std::collections::HashMap<CrateKey, std::sync::Arc<CrateListing>>>>,
lines: std::sync::Mutex<Option<std::collections::HashMap<CrateKey, Vec<CrateLine>>>>,
}
/// The record of the packuments the npm sources listed ([`NpmListing`]).
const NPM_LISTINGS: &str = "npm-listings.jsonl";
/// The manifest fields of a listed version npm needs to resolve and install it.
const NPM_MANIFEST_FIELDS: [&str; 13] = [
"dependencies",
"optionalDependencies",
"peerDependencies",
"peerDependenciesMeta",
"bundleDependencies",
"bundledDependencies",
"bin",
"engines",
"os",
"cpu",
"directories",
"hasInstallScript",
"license",
];
/// One crate's index line as the source answered it when the crate was pulled.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
struct CrateLine {
source: String,
name: String,
version: String,
line: String,
}
/// One APK package's stanza as the signed upstream index wrote it when it was pulled.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
struct ApkLine {
source: String,
arch: String,
name: String,
version: String,
stanza: String,
}
/// What a Maven path is served as.
pub fn maven_ctype(file: &str) -> &'static str {
if file.ends_with(".pom") || file.ends_with(".xml") {
"text/xml"
} else if file.ends_with(".jar") {
"application/java-archive"
} else {
"application/octet-stream"
}
}
impl PullIntake {
fn source_of(&self, source: &str, kind: RemoteKind) -> Result<&SourceRt, Ask> {
self.source(source)
.filter(|s| s.cfg.kind == kind)
.ok_or_else(|| Ask::Unknown {
why: format!("'{source}' is no {} source", kind.name()),
})
}
/// The coordinates this source has brought in that `f` keeps.
fn ids_where(&self, chain: &IntakeChain, f: impl Fn(&ArtifactId) -> bool) -> Vec<ArtifactId> {
chain
.all()
.into_iter()
.map(|(id, _)| id)
.filter(|id| id.namespace.is_none() && id.version.is_empty() && f(id))
.collect()
}
/// **Where `want` stands**, and a pull when nothing is on the way: the
/// approved coordinates are walked on (settled), an open job is "fetching",
/// and the source is asked when it never was, when a failure is old, or (with
/// `refresh`) when an AVAILABLE answer is past its refresh age.
pub(super) fn ask_want(
&self,
chain: &IntakeChain,
src: &SourceRt,
want: &Want,
ids: &[ArtifactId],
refresh: bool,
) -> Ask {
let name = &src.cfg.name;
let settled = self.settle(chain, src, ids);
if settled
== (Ask::Unknown {
why: "nothing was pulled".into(),
})
{
if let Some(why) = ids
.iter()
.find_map(|id| self.refused_at_door(name, &id.name))
{
return Ask::Refused { why };
}
}
if let Some(job) = self.open_job(name, want) {
return settled.best(Ask::Fetching {
retry_after_secs: self.retry_after_secs,
job,
});
}
let last = self.last_request(name, want);
let stale = last
.as_ref()
.is_none_or(|r| now_secs() - r.at >= src.refresh_secs());
match &settled {
// In the chain already: it is walked there, never pulled again.
Ask::Held { .. } | Ask::Refused { .. } => return settled,
Ask::Ready => {
if refresh && stale {
let _ = self.enqueue(name, want);
}
return Ask::Ready;
}
_ => {}
}
if last.is_none()
|| last
.as_ref()
.is_some_and(|r| r.outcome == "failed" && stale)
{
return self.enqueue(name, want);
}
match last {
Some(r) if r.outcome == "refused" => Ask::Refused { why: r.why },
Some(r) if r.outcome != "fetched" => Ask::Unknown { why: r.why },
_ => settled,
}
}
/// Run `f`; when it says a pull is in flight and the source waits
/// (`wait_secs`), wait for that job (bounded) and ask once more.
fn waited<T>(&self, src: &SourceRt, f: impl Fn() -> Result<T, Ask>) -> Result<T, Ask> {
let first = f();
let wait = src.fetch_wait();
if let (Err(Ask::Fetching { job, .. }), false) = (&first, wait.is_zero()) {
if let Some(q) = self.queue.get() {
let _ = q.await_completion(&crate::jobs::JobId(job.clone()), wait);
return f();
}
}
first
}
/// The sealed bytes of `id` once it is AVAILABLE; otherwise where `want` stands.
fn serve_or_ask(
&self,
chain: &IntakeChain,
src: &SourceRt,
id: &ArtifactId,
want: &Want,
) -> Result<Vec<u8>, Ask> {
if chain.state(id) == ArtifactState::Available {
if let Ok(Some(b)) = chain.serve(id) {
return Ok(b);
}
}
match self.ask_want(chain, src, want, std::slice::from_ref(id), false) {
Ask::Ready => chain.serve(id).ok().flatten().ok_or_else(|| Ask::Unknown {
why: format!("{} is not AVAILABLE", id.name),
}),
a => Err(a),
}
}
fn available(chain: &IntakeChain, ids: &[ArtifactId]) -> Vec<ArtifactId> {
ids.iter()
.filter(|id| chain.state(id) == ArtifactState::Available)
.cloned()
.collect()
}
// ── npm ──
/// **A packument was asked for**: npm's document over the AVAILABLE versions
/// (each its sealed tarball's own `package.json`), tarball links under
/// `tarball_base` (`<public base>/<production>/<source>`).
pub fn npm_packument(
&self,
chain: &IntakeChain,
source: &str,
package: &str,
tarball_base: &str,
) -> Result<Vec<u8>, Ask> {
let src = self.source_of(source, RemoteKind::Npm)?;
self.waited(src, || {
self.npm_packument_once(chain, source, package, tarball_base)
})
}
fn npm_packument_once(
&self,
chain: &IntakeChain,
source: &str,
package: &str,
tarball_base: &str,
) -> Result<Vec<u8>, Ask> {
let src = self.source_of(source, RemoteKind::Npm)?;
if !crate::promote::npm::valid_package_name(package) {
return Err(Ask::Unknown {
why: format!("'{package}' is not an npm package name"),
});
}
let ids = self.ids_where(chain, |id| {
npm::coords_of(source, id).is_some_and(|(p, _)| p == package)
});
let want = Want::NpmPackage {
package: package.into(),
};
let ask = self.ask_want(chain, src, &want, &ids, true);
let mut versions = Vec::new();
for id in Self::available(chain, &ids) {
let Some((_, version)) = npm::coords_of(source, &id) else {
continue;
};
let Some(st) = chain.status(&id) else {
continue;
};
let cached = self.npm_meta.lock_or_recover().get(&st.sha256).cloned();
let meta = match cached {
Some(m) => m,
// The version's index row (lane HOL-PERF-2): never the sealed tarball.
None => {
match chain
.index_row(&id)
.as_deref()
.and_then(crate::promote::npm::meta_of_row)
{
Some(m) => {
self.npm_meta
.lock_or_recover()
.insert(st.sha256.clone(), m.clone());
m
}
None => {
log::warn!("pull-through {source}: {package}@{version} has no index row; it is left out of its packument");
continue;
}
}
}
};
versions.push((version, meta));
}
let held = self.npm_held(chain, source, package);
if versions.is_empty() && held.is_empty() {
return Err(ask);
}
// `latest` and every tag name AVAILABLE versions only; the held ones ride beside them.
let set = versions.iter().map(|(v, _)| v.clone()).collect();
let tags = crate::promote::npm::effective_tags(&[], &set);
versions.extend(held);
versions.sort_by(|a, b| crate::promote::npm::compare_versions(&a.0, &b.0));
Ok(
crate::promote::npm::packument(package, &versions, &tags, tarball_base)
.to_string()
.into_bytes(),
)
}
/// Keep what the registry listed for `package` ([`NpmListing`]): written before a version
/// is driven, so a client's retry already sees the packument.
pub(super) fn remember_npm_listing(
&self,
source: &str,
package: &str,
doc: &serde_json::Value,
) {
let Some(all) = doc.get("versions").and_then(|v| v.as_object()) else {
return;
};
let versions = all
.iter()
.filter_map(|(version, v)| {
let integrity = v.pointer("/dist/integrity")?.as_str()?.to_string();
let mut manifest = serde_json::Map::new();
for field in NPM_MANIFEST_FIELDS {
if let Some(x) = v.get(field) {
manifest.insert(field.into(), x.clone());
}
}
Some((
version.clone(),
serde_json::Value::Object(manifest),
integrity,
))
})
.collect();
let listing = NpmListing {
source: source.into(),
package: package.into(),
at: now_secs(),
versions,
};
if let Err(e) = self.append(NPM_LISTINGS, &listing) {
log::warn!("pull-through: the npm listing could not be written: {e}");
}
}
/// **The versions the registry lists for `package` that are not AVAILABLE here**, as the
/// packument names them: deprecated with the reason, so npm keeps a range to the approved
/// versions and an exact pin asks for the tarball, which pulls it ([`Self::npm_tarball`]).
/// Refused, invalidated and door-refused versions are never offered.
fn npm_held(
&self,
chain: &IntakeChain,
source: &str,
package: &str,
) -> Vec<(String, crate::promote::npm::VersionMeta)> {
let Some(listing) = self
.read::<NpmListing>(NPM_LISTINGS)
.into_iter()
.rev()
.find(|l| l.source == source && l.package == package)
else {
return Vec::new();
};
let mut out = Vec::new();
for (version, manifest, integrity) in listing.versions {
let id = npm::id(source, package, &version);
let reason = match chain.status(&id).map(|_| chain.state(&id)) {
None if self.refused_at_door(source, &id.name).is_some() => continue,
None => "not in Holger yet: installing this exact version pulls it into Holger's intake chain, where it waits for approval".to_string(),
Some(ArtifactState::Available | ArtifactState::Blocked | ArtifactState::Quarantined) => continue,
Some(state) => format!(
"held by Holger's intake chain ({}), waiting for approval",
serde_json::to_value(state)
.ok()
.and_then(|v| v.as_str().map(str::to_string))
.unwrap_or_default()
),
};
let mut manifest = manifest;
if let Some(m) = manifest.as_object_mut() {
m.insert("deprecated".into(), reason.into());
}
out.push((
version,
crate::promote::npm::VersionMeta {
manifest,
integrity,
},
));
}
out
}
/// **A tarball was asked for** (`<package>/-/<file>.tgz`).
pub fn npm_tarball(
&self,
chain: &IntakeChain,
source: &str,
package: &str,
file: &str,
) -> Result<Vec<u8>, Ask> {
let src = self.source_of(source, RemoteKind::Npm)?;
self.waited(src, || self.npm_tarball_once(chain, source, package, file))
}
fn npm_tarball_once(
&self,
chain: &IntakeChain,
source: &str,
package: &str,
file: &str,
) -> Result<Vec<u8>, Ask> {
let src = self.source_of(source, RemoteKind::Npm)?;
let id = ArtifactId {
namespace: None,
name: format!("{source}/{package}/-/{file}"),
version: String::new(),
};
let Some((p, version)) = npm::coords_of(source, &id).filter(|(p, _)| p == package) else {
return Err(Ask::Unknown {
why: format!("'{file}' is not a tarball of {package}"),
});
};
self.serve_or_ask(
chain,
src,
&id,
&Want::NpmVersion {
package: p,
version,
},
)
}
// ── Maven ──
/// **A path below a Maven source**: a file, a checksum of one, or
/// `maven-metadata.xml` over the AVAILABLE versions. `(content type, body)`.
pub fn maven_read(
&self,
chain: &IntakeChain,
source: &str,
path: &maven::MavenPath,
) -> Result<(&'static str, Vec<u8>), Ask> {
let src = self.source_of(source, RemoteKind::Maven)?;
self.waited(src, || self.maven_read_once(chain, source, path))
}
fn maven_read_once(
&self,
chain: &IntakeChain,
source: &str,
path: &maven::MavenPath,
) -> Result<(&'static str, Vec<u8>), Ask> {
let src = self.source_of(source, RemoteKind::Maven)?;
match path {
maven::MavenPath::File {
group,
artifact,
version,
file,
} => {
let id = maven::id(source, &path.path());
let want = if maven::version_files(artifact, version).contains(file) {
Want::MavenVersion {
group: group.clone(),
artifact: artifact.clone(),
version: version.clone(),
}
} else {
Want::MavenFile { path: path.path() }
};
self.serve_or_ask(chain, src, &id, &want)
.map(|b| (maven_ctype(file), b))
}
maven::MavenPath::Metadata { group, artifact } => {
let prefix = format!("{source}/{group}/{artifact}/");
let ids = self.ids_where(chain, |id| {
id.name
.strip_prefix(&prefix)
.is_some_and(|r| r.split('/').count() == 2)
});
let want = Want::MavenLatest {
group: group.clone(),
artifact: artifact.clone(),
};
let ask = self.ask_want(chain, src, &want, &ids, false);
let versions: Vec<String> = Self::available(chain, &ids)
.iter()
.filter_map(|id| {
id.name
.strip_prefix(&prefix)?
.split('/')
.next()
.map(str::to_string)
})
.collect();
if versions.is_empty() {
// P13: a plugin group's metadata, as its source listed it.
let path = format!("{group}/{artifact}");
if let Some(g) = self
.read::<MavenPluginGroup>(MAVEN_PLUGIN_GROUPS)
.into_iter()
.rev()
.find(|g| g.source == source && g.group == path)
{
return Ok(("text/xml", maven::plugins_xml(&g.plugins).into_bytes()));
}
return Err(ask);
}
Ok((
"text/xml",
maven::metadata_xml(group, artifact, versions).into_bytes(),
))
}
maven::MavenPath::Checksum { of, algo } => {
let (_, body) = self.maven_read_once(chain, source, of)?;
Ok((
"text/plain",
maven::digest(algo, &body).unwrap_or_default().into_bytes(),
))
}
}
}
// ── cargo ──
fn crate_lines(&self, source: &str, name: &str) -> Vec<CrateLine> {
let mut m = self.crate_records.lines.lock_or_recover();
let m = m.get_or_insert_with(|| {
let mut by: std::collections::HashMap<CrateKey, Vec<CrateLine>> = Default::default();
for l in self.read::<CrateLine>("crates-index.jsonl") {
by.entry((l.source.clone(), l.name.clone()))
.or_default()
.push(l);
}
by
});
m.get(&(source.to_string(), name.to_string()))
.cloned()
.unwrap_or_default()
}
/// The newest listing of a crate (HOL-DOGFOOD-8: from memory, the file read once).
fn crate_listing(&self, source: &str, name: &str) -> Option<std::sync::Arc<CrateListing>> {
let mut m = self.crate_records.listings.lock_or_recover();
let m = m.get_or_insert_with(|| {
let mut by: std::collections::HashMap<CrateKey, std::sync::Arc<CrateListing>> =
Default::default();
for l in self.read::<CrateListing>(CRATE_LISTINGS) {
by.insert((l.source.clone(), l.name.clone()), std::sync::Arc::new(l));
}
by
});
m.get(&(source.to_string(), name.to_ascii_lowercase()))
.cloned()
}
/// **A crate's index file was asked for**: the AVAILABLE versions' lines.
pub fn crate_index(
&self,
chain: &IntakeChain,
source: &str,
name: &str,
) -> Result<Vec<u8>, Ask> {
let src = self.source_of(source, RemoteKind::Crates)?;
self.waited(src, || self.crate_index_once(chain, source, name))
}
fn crate_index_once(
&self,
chain: &IntakeChain,
source: &str,
name: &str,
) -> Result<Vec<u8>, Ask> {
let src = self.source_of(source, RemoteKind::Crates)?;
let ids = self.ids_where(chain, |id| {
crates::coords_of(source, id).is_some_and(|(n, _)| n == name)
});
let ask = self.ask_want(
chain,
src,
&Want::CrateName { name: name.into() },
&ids,
true,
);
let lines = self.crate_lines(source, name);
let mut out = String::new();
let mut seen = std::collections::BTreeSet::new();
for id in Self::available(chain, &ids) {
let Some((_, version)) = crates::coords_of(source, &id) else {
continue;
};
let Some(st) = chain.status(&id) else {
continue;
};
// The line, with the sealed bytes' digest: the one cargo checks. A serve-only
// source (a downstream) reads it from the sealed `.crate` itself.
let line = match lines.iter().rev().find(|l| l.version == version) {
Some(l) => l.line.clone(),
None if src.cfg.serves_only() => {
match self.derived_line(chain, &id, name, &version, &st.sha256) {
Some(l) => l,
None => continue,
}
}
None => continue,
};
let Ok(mut v) = serde_json::from_str::<serde_json::Value>(&line) else {
continue;
};
if v.get("cksum").and_then(|c| c.as_str()) != Some(st.sha256.as_str()) {
log::warn!("pull-through {source}: {name} {version}'s index line names another cksum than its sealed bytes; left out");
continue;
}
v["yanked"] = serde_json::Value::Bool(false);
if seen.insert(version) {
out.push_str(&v.to_string());
out.push('\n');
}
}
// The registry's other versions, YANKED: cargo resolves no new requirement onto one, a
// lock file that names it takes it, and its `.crate` pulls exactly that version.
if let Some(listing) = self.crate_listing(source, name) {
for (version, raw) in listing.lines.iter().cloned() {
if seen.contains(&version) {
continue;
}
let id = crates::id(source, name, &version);
match chain.status(&id).map(|_| chain.state(&id)) {
None if self.refused_at_door(source, &id.name).is_some() => continue,
Some(
ArtifactState::Available
| ArtifactState::Blocked
| ArtifactState::Quarantined,
) => continue,
_ => {}
}
let Ok(mut v) = serde_json::from_str::<serde_json::Value>(&raw) else {
continue;
};
v["yanked"] = serde_json::Value::Bool(true);
if seen.insert(version) {
out.push_str(&v.to_string());
out.push('\n');
}
}
}
if out.is_empty() {
return Err(ask);
}
Ok(out.into_bytes())
}
/// The index line of a sealed `.crate`, read from its own `Cargo.toml` once (by sha256).
fn derived_line(
&self,
chain: &IntakeChain,
id: &ArtifactId,
name: &str,
version: &str,
sha256: &str,
) -> Option<String> {
if let Some(l) = self.derived_lines.lock_or_recover().get(sha256) {
return Some(l.clone());
}
let bytes = chain.serve(id).ok().flatten()?;
match crates::line_from_crate(&bytes, name, version) {
Ok(l) => {
self.derived_lines
.lock_or_recover()
.insert(sha256.to_string(), l.clone());
Some(l)
}
Err(e) => {
log::warn!("crates source: {name} {version}'s index line could not be read from its sealed bytes: {e:#}");
None
}
}
}
/// **A `.crate` was asked for.**
pub fn crate_file(
&self,
chain: &IntakeChain,
source: &str,
name: &str,
version: &str,
) -> Result<Vec<u8>, Ask> {
let src = self.source_of(source, RemoteKind::Crates)?;
self.waited(src, || self.crate_file_once(chain, source, name, version))
}
fn crate_file_once(
&self,
chain: &IntakeChain,
source: &str,
name: &str,
version: &str,
) -> Result<Vec<u8>, Ask> {
let src = self.source_of(source, RemoteKind::Crates)?;
let id = crates::id(source, name, version);
self.serve_or_ask(
chain,
src,
&id,
&Want::CrateVersion {
name: name.to_ascii_lowercase(),
version: version.into(),
},
)
}
// ── the job ──
/// The pull job's body for the npm, Maven, cargo, APK and Go wants.
pub(super) fn run_ecosystem(
&self,
chain: &IntakeChain,
src: &SourceRt,
want: &Want,
) -> anyhow::Result<String> {
match want {
Want::NpmPackage { package } => self.run_npm(chain, src, package, None, want),
Want::NpmVersion { package, version } => {
self.run_npm(chain, src, package, Some(version), want)
}
Want::MavenVersion {
group,
artifact,
version,
} => self.run_maven_version(chain, src, group, artifact, version, want),
Want::MavenFile { path } => {
let id = maven::id(&src.cfg.name, path);
match self.drive_maven(chain, src, &id, path)? {
Some(d) => Ok(self.record_drives(&src.cfg.name, want, vec![d])),
None => Ok(self.unknown(
src,
want,
format!("source '{}' has no {path}", src.cfg.name),
)),
}
}
Want::MavenLatest { group, artifact } => {
let text = maven::metadata_text(
self.http.as_ref(),
&src.cfg.url,
group,
artifact,
src.auth(),
)?;
// SEC-D3: the metadata is read in a job; the server takes its answer only.
let meta: Option<super::manifest_job::MetadataAnswer> = match &text {
Some(t) => Some(super::manifest_job::run_job(
chain,
&self.root,
super::manifest_job::MAVEN_METADATA,
&[(super::manifest_job::METADATA_XML.into(), t.as_bytes())],
)?),
None => None,
};
// P13: a GROUP's metadata (`org/apache/maven/plugins/`) lists plugin prefixes;
// the list is kept for the door, names only.
let plugins = meta.as_ref().map(|m| m.plugins.clone()).unwrap_or_default();
if !plugins.is_empty() {
let kept = MavenPluginGroup {
source: src.cfg.name.clone(),
group: format!("{group}/{artifact}"),
at: super::now_secs(),
plugins,
};
if let Err(e) = self.append(MAVEN_PLUGIN_GROUPS, &kept) {
log::warn!("pull-through: the plugin group could not be written: {e}");
}
}
match meta.and_then(|m| m.release) {
Some(v) => self.run_maven_version(chain, src, group, artifact, &v, want),
None => Ok(self.unknown(
src,
want,
format!(
"source '{}' publishes no release of {group}/{artifact}",
src.cfg.name
),
)),
}
}
Want::CrateName { name } => self.run_crate(chain, src, name, None, want),
Want::CrateVersion { name, version } => {
self.run_crate(chain, src, name, Some(version), want)
}
Want::ApkName { arch, name } => self.run_apk(chain, src, arch, name, None, want),
Want::ApkFile { arch, file } => match apk::parse_file(file) {
Some((name, version)) => {
self.run_apk(chain, src, arch, &name, Some(&version), want)
}
None => anyhow::bail!(crate::jobs::JobPermanent(format!(
"'{file}' is not an APK file name"
))),
},
Want::GoVersion { module, version } => self.run_go(chain, src, module, version, want),
_ => anyhow::bail!(crate::jobs::JobPermanent(format!(
"source '{}' does not pull {}",
src.cfg.name,
want.key()
))),
}
}
pub(super) fn unknown(&self, src: &SourceRt, want: &Want, why: String) -> String {
self.record(&src.cfg.name, want, "unknown", Vec::new(), &why);
why
}
/// INGEST (when not in the chain yet) and walk. `(name, line, refused at the door)`.
pub(super) fn drive(
&self,
chain: &IntakeChain,
src: &SourceRt,
id: &ArtifactId,
fetch: impl FnOnce() -> anyhow::Result<(Vec<u8>, Vec<super::provenance::Material>)>,
) -> anyhow::Result<(String, String, bool)> {
if chain.status(id).is_none() {
let (bytes, materials) = fetch()?;
if let Some(line) = self.ingest(chain, src, id, &bytes, &materials)? {
return Ok((id.name.clone(), line, true));
}
}
Ok((id.name.clone(), self.walk(chain, src, id, None)?, false))
}
fn run_npm(
&self,
chain: &IntakeChain,
src: &SourceRt,
package: &str,
version: Option<&str>,
want: &Want,
) -> anyhow::Result<String> {
let Some(doc) = npm::packument(self.http.as_ref(), &src.cfg.url, package, src.auth())?
else {
return Ok(self.unknown(
src,
want,
format!("source '{}' does not publish '{package}'", src.cfg.name),
));
};
let page = format!(
"{}/{}",
src.cfg.url.trim_end_matches('/'),
npm::url_name(package)
);
self.remember_npm_listing(&src.cfg.name, package, &doc);
let Some(v) = version.map(str::to_string).or_else(|| npm::latest(&doc)) else {
return Ok(self.unknown(
src,
want,
format!(
"source '{}' names no `latest` for '{package}'",
src.cfg.name
),
));
};
let Some(rv) = npm::version(&doc, &page, package, &v) else {
return Ok(self.unknown(
src,
want,
format!("source '{}' does not publish {package}@{v}", src.cfg.name),
));
};
let _ = rv;
let w = self.npm_walk(chain, src, package, &v, doc)?;
Ok(self.record_walk(src, want, format!("{package}@{v}"), w))
}
fn drive_maven(
&self,
chain: &IntakeChain,
src: &SourceRt,
id: &ArtifactId,
path: &str,
) -> anyhow::Result<Option<(String, String, bool)>> {
if chain.status(id).is_some() {
return Ok(Some((
id.name.clone(),
self.walk(chain, src, id, None)?,
false,
)));
}
let Some(fetched) = maven::fetch_file(
self.http.as_ref(),
&src.cfg.url,
path,
src.auth(),
src.cap(),
)?
else {
return Ok(None);
};
let mut once = Some(fetched);
self.drive(chain, src, id, || {
once.take().ok_or_else(|| anyhow::anyhow!("fetched twice"))
})
.map(Some)
}
/// **A Maven version and its whole closure** (lane HOL-DEPSET-APPROVE): every POM Maven
/// reads to use it (parents, BOMs, every version the graph visits) and the artifact of
/// every mediation winner ([`super::pom::closure`]), each fetched against its published
/// checksums and driven through the chain, then recorded as ONE dependency set.
fn run_maven_version(
&self,
chain: &IntakeChain,
src: &SourceRt,
group: &str,
artifact: &str,
version: &str,
want: &Want,
) -> anyhow::Result<String> {
let path = format!("{group}/{artifact}/{version}/{artifact}-{version}.pom");
match maven::parse(&path) {
Some(maven::MavenPath::File { .. }) => {}
_ => anyhow::bail!(crate::jobs::JobPermanent(format!(
"'{path}' is not a Maven path Holger pulls"
))),
}
let root = super::pom::Gav::new(&group.replace('/', "."), artifact, version);
let fetched = MavenFetched {
pull: self,
src,
files: std::cell::RefCell::new(std::collections::HashMap::new()),
};
// SEC-D3 (lane HOL-312-HARDEN): the POMs and the metadata are parsed in jobs, round by
// round; this process only fetches the bytes each round names.
let closure = super::manifest_job::closure_in_jobs(
chain,
&self.root,
&root,
&|path| fetched.pom_bytes(path),
&|key| fetched.metadata_bytes(key),
)?;
if closure.files.is_empty() {
return Ok(self.unknown(
src,
want,
format!(
"source '{}' has no {group}/{artifact}/{version}{}",
src.cfg.name,
closure
.unresolved
.first()
.map(|(_, why)| format!(" ({why})"))
.unwrap_or_default()
),
));
}
let mut drives = Vec::new();
let mut unresolved = closure.unresolved.clone();
let mut roles = std::collections::HashMap::new();
for f in &closure.files {
// D2: only a Maven path of plain segments is fetched (nothing a CDN would read as
// another file, nothing named after what it is not).
let plain = f.path.split('/').all(|s| {
!s.is_empty() && s != "." && s != ".." && !s.contains(['?', '#', '\\', '%'])
});
if !plain || maven::parse(&f.path).is_none() {
unresolved.push((f.path.clone(), "not a Maven path segment by segment".into()));
continue;
}
let id = maven::id(&src.cfg.name, &f.path);
roles.insert(id.name.clone(), (f.role.name().to_string(), f.via.clone()));
let kept = fetched.files.borrow_mut().remove(&f.path);
let d = match kept {
Some(bytes) => Some(self.drive(chain, src, &id, || Ok(bytes))?),
None => self.drive_maven(chain, src, &id, &f.path)?,
};
match d {
Some(d) => drives.push(d),
None => unresolved.push((
f.path.clone(),
format!("source '{}' does not publish it", src.cfg.name),
)),
}
}
let rec = super::depset::DepSetRecord {
id: super::depset::set_id(&src.cfg.name, want),
source: src.cfg.name.clone(),
ecosystem: src.cfg.kind.name().into(),
root: root.to_string(),
want: want.clone(),
at: super::now_secs(),
members: super::depset::members_of(&drives, &|name: &str| {
roles.get(name).cloned().unwrap_or_default()
}),
unresolved,
notes: Vec::new(),
};
self.record_set(&rec);
Ok(self.record_drives(&src.cfg.name, want, drives))
}
/// Keep what the registry listed for crate `name` ([`CrateListing`]).
pub(super) fn remember_listing_crate(
&self,
source: &str,
name: &str,
lines: &[crates::IndexLine],
) {
let listing = CrateListing {
source: source.to_string(),
name: name.to_ascii_lowercase(),
at: super::now_secs(),
lines: lines
.iter()
.filter(|l| !l.yanked)
.map(|l| (l.vers.clone(), l.raw.to_string()))
.collect(),
};
if let Err(e) = self.append(CRATE_LISTINGS, &listing) {
log::warn!("pull-through: the crate listing could not be written: {e}");
return;
}
if let Some(m) = self.crate_records.listings.lock_or_recover().as_mut() {
m.insert(
(listing.source.clone(), listing.name.clone()),
std::sync::Arc::new(listing),
);
}
}
/// **A crate's index file, through the ONE shared index** (HOL-DOGFOOD-8): read from the
/// upstream once per [`crates::SHARED_INDEX_TTL`] for every pull and closure walk, and its
/// listing recorded once per read (it was once per walk and per edge).
pub(super) fn crate_index_lines(
&self,
src: &SourceRt,
name: &str,
) -> anyhow::Result<Option<crates::Lines>> {
let (http, url, auth) = (self.http.as_ref(), src.cfg.url.as_str(), src.auth());
let (lines, fresh) = self
.crate_index
.get(&src.cfg.name, name, || crates::index(http, url, name, auth))?;
if fresh {
if let Some(l) = &lines {
self.remember_listing_crate(&src.cfg.name, name, l);
}
}
Ok(lines)
}
/// INGEST and walk one crate version, its index line kept for Holger's index.
pub(super) fn drive_crate_line(
&self,
chain: &IntakeChain,
src: &SourceRt,
line: &crates::IndexLine,
) -> anyhow::Result<(String, String, bool)> {
let id = crates::id(&src.cfg.name, &line.name, &line.vers);
if chain.status(&id).is_none() && line.yanked {
return Ok((
id.name.clone(),
format!("{}: yanked upstream, not pulled", id.name),
true,
));
}
let (http, auth, cap) = (self.http.as_ref(), src.auth(), src.cap());
let fresh = chain.status(&id).is_none();
let d = self.drive(chain, src, &id, || {
crates::fetch_crate(http, &src.cfg.url, line, auth, cap).map(|b| (b, Vec::new()))
})?;
if fresh && !d.2 {
let rec = CrateLine {
source: src.cfg.name.clone(),
name: line.name.to_ascii_lowercase(),
version: line.vers.clone(),
line: line.raw.to_string(),
};
if let Err(e) = self.append("crates-index.jsonl", &rec) {
log::warn!("pull-through: the crate index line could not be recorded: {e}");
} else if let Some(m) = self.crate_records.lines.lock_or_recover().as_mut() {
m.entry((rec.source.clone(), rec.name.clone()))
.or_default()
.push(rec);
}
}
Ok(d)
}
/// A crate version and its closure (normal and build dependencies), as ONE set.
fn run_crate(
&self,
chain: &IntakeChain,
src: &SourceRt,
name: &str,
version: Option<&str>,
want: &Want,
) -> anyhow::Result<String> {
// What the registry lists, for the index's yanked lines (the cargo half of P2): recorded
// by the shared index, once per upstream read.
let Some(lines) = self.crate_index_lines(src, name)? else {
return Ok(self.unknown(
src,
want,
format!("source '{}' does not publish crate '{name}'", src.cfg.name),
));
};
let line = match version {
Some(v) => lines.iter().position(|l| l.vers == v),
None => {
crates::newest(&lines).and_then(|n| lines.iter().position(|l| l.vers == n.vers))
}
};
let Some(at) = line else {
return Ok(self.unknown(
src,
want,
format!(
"source '{}' publishes no {} of crate '{name}'",
src.cfg.name,
version.unwrap_or("live release")
),
));
};
let line = &lines[at];
let id = crates::id(&src.cfg.name, &line.name, &line.vers);
if chain.status(&id).is_none() && line.yanked {
let d = self.drive_crate_line(chain, src, line)?;
return Ok(self.record_drives(&src.cfg.name, want, vec![d]));
}
let root = format!("{}@{}", line.name, line.vers);
let w = self.crate_walk(chain, src, std::sync::Arc::clone(&lines), at)?;
Ok(self.record_walk(src, want, root, w))
}
// ── APK ──
fn apk_lines(&self, source: &str, arch: &str) -> Vec<ApkLine> {
self.read::<ApkLine>("apk-index.jsonl")
.into_iter()
.filter(|l| l.source == source && l.arch == arch)
.collect()
}
fn apk_ids(
&self,
chain: &IntakeChain,
source: &str,
f: impl Fn(&str, &str, &str) -> bool,
) -> Vec<ArtifactId> {
self.ids_where(chain, |id| {
apk::coords_of(source, id).is_some_and(|(a, n, v)| f(&a, &n, &v))
})
}
/// **Holger's `APKINDEX.tar.gz` for one arch of an APK source**: the AVAILABLE
/// packages only, each with the upstream's stanza, signed with Holger's key.
/// Approved packages are walked on to AVAILABLE first.
pub fn apk_index(&self, chain: &IntakeChain, source: &str, arch: &str) -> Result<Vec<u8>, Ask> {
let src = self.source_of(source, RemoteKind::Apk)?;
if !apk::valid_arch(arch) {
return Err(Ask::Unknown {
why: format!("'{arch}' is not an APK arch"),
});
}
let ids = self.apk_ids(chain, source, |a, _, _| a == arch);
for id in &ids {
let _ = self.settle(chain, src, std::slice::from_ref(id));
}
let lines = self.apk_lines(source, arch);
let mut stanzas = Vec::new();
let mut seen = std::collections::BTreeSet::new();
for id in Self::available(chain, &ids) {
let Some((_, name, version)) = apk::coords_of(source, &id) else {
continue;
};
let Some(l) = lines
.iter()
.rev()
.find(|l| l.name == name && l.version == version)
else {
log::warn!("pull-through {source}: {name} {version} is AVAILABLE with no recorded index stanza; left out");
continue;
};
if seen.insert((name, version)) {
stanzas.push(l.stanza.clone());
}
}
let signer = self.apk_signer().map_err(|e| Ask::Unknown {
why: format!("{e:#}"),
})?;
signer
.index(
&format!("holger {source} {arch}: the AVAILABLE packages"),
&stanzas,
)
.map_err(|e| Ask::Unknown {
why: format!("the index could not be written: {e:#}"),
})
}
/// **`<arch>/-/pull/<name>`**: the newest version of `name` and its closure,
/// pulled into the chain. `Ok(line)` once that version is AVAILABLE.
pub fn apk_pull(
&self,
chain: &IntakeChain,
source: &str,
arch: &str,
name: &str,
) -> Result<String, Ask> {
let src = self.source_of(source, RemoteKind::Apk)?;
self.waited(src, || self.apk_pull_once(chain, source, arch, name))
}
fn apk_pull_once(
&self,
chain: &IntakeChain,
source: &str,
arch: &str,
name: &str,
) -> Result<String, Ask> {
let src = self.source_of(source, RemoteKind::Apk)?;
if !apk::valid_arch(arch) || !apk::valid_name(name) {
return Err(Ask::Unknown {
why: format!("'{arch}/{name}' is not an APK arch and package name"),
});
}
let ids = self.apk_ids(chain, source, |a, n, _| a == arch && n == name);
let want = Want::ApkName {
arch: arch.into(),
name: name.into(),
};
// Where the closure the last pull brought in stands, one coordinate a line: what
// an approver walks (and `apk add` needs AVAILABLE, or installed already).
let closure = || -> String {
let Some(r) = self.last_request(source, &want) else {
return String::new();
};
r.names
.iter()
.map(|n| {
let id = ArtifactId {
namespace: None,
name: n.clone(),
version: String::new(),
};
format!("\n {n}: {}", chain.state(&id))
})
.collect()
};
match self.ask_want(chain, src, &want, &ids, false) {
Ask::Ready => {
let mut v: Vec<String> = Self::available(chain, &ids)
.iter()
.filter_map(|id| apk::coords_of(source, id).map(|c| c.2))
.collect();
v.sort_by(|a, b| apk::compare_versions(a, b));
Ok(format!(
"{name} {} is AVAILABLE from '{source}' ({arch}); the pull brought in:{}\n",
v.join(", "),
closure()
))
}
Ask::Held { state, detail } => Err(Ask::Held {
state,
detail: format!("{detail}; the pull brought in:{}", closure()),
}),
a => Err(a),
}
}
/// **A `.apk` was asked for.**
pub fn apk_file(
&self,
chain: &IntakeChain,
source: &str,
arch: &str,
file: &str,
) -> Result<Vec<u8>, Ask> {
let src = self.source_of(source, RemoteKind::Apk)?;
self.waited(src, || self.apk_file_once(chain, source, arch, file))
}
fn apk_file_once(
&self,
chain: &IntakeChain,
source: &str,
arch: &str,
file: &str,
) -> Result<Vec<u8>, Ask> {
let src = self.source_of(source, RemoteKind::Apk)?;
let Some((name, version)) = apk::parse_file(file).filter(|_| apk::valid_arch(arch)) else {
return Err(Ask::Unknown {
why: format!("'{arch}/{file}' is not an APK file"),
});
};
let id = apk::id(source, arch, &name, &version);
self.serve_or_ask(
chain,
src,
&id,
&Want::ApkFile {
arch: arch.into(),
file: file.into(),
},
)
}
/// The pull of an APK name (its newest version and closure) or of one version.
fn run_apk(
&self,
chain: &IntakeChain,
src: &SourceRt,
arch: &str,
name: &str,
version: Option<&str>,
want: &Want,
) -> anyhow::Result<String> {
let source = &src.cfg.name;
if !apk::valid_arch(arch) || !apk::valid_name(name) {
anyhow::bail!(crate::jobs::JobPermanent(format!(
"'{arch}/{name}' is not an APK arch and package name"
)));
}
// The keys this source's index may be signed with: the pinned ones, narrowed by its gate.
let keys: Vec<_> = self
.apk_keys()
.into_iter()
.filter(|k| src.demand.apk_keys.is_empty() || src.demand.apk_keys.contains(&k.name))
.collect();
let (raw, idx) =
match apk::fetch_index(self.http.as_ref(), &src.cfg.url, arch, src.auth(), &keys)? {
apk::UpstreamIndex::Verified(raw, idx) => (raw, idx),
apk::UpstreamIndex::Missing => {
return Ok(self.unknown(
src,
want,
format!("source '{source}' has no index for {arch}"),
))
}
apk::UpstreamIndex::Refused(why) => {
let what = apk::id(source, arch, name, version.unwrap_or("0-r0")).name;
self.record(source, want, "refused", vec![what], &why);
return Ok(why);
}
};
let root = match version {
Some(v) => idx.entries.iter().find(|e| {
e.name == name
&& e.version == v
&& (e.arch.is_empty() || e.arch == arch || e.arch == "noarch")
}),
None => apk::newest(&idx, arch, name),
};
let Some(root) = root else {
return Ok(self.unknown(
src,
want,
format!(
"source '{source}' publishes no {name}{} for {arch}",
version.map(|v| format!(" {v}")).unwrap_or_default()
),
));
};
// A file request pulls that file; a name request its closure.
let entries = if version.is_some() {
vec![root]
} else {
apk::closure(&idx, arch, root)
};
let material = [super::provenance::Material {
name: "APKINDEX.tar.gz".into(),
bytes: raw,
}];
let mut drives = Vec::new();
for (i, e) in entries.iter().enumerate() {
let id = apk::id(source, arch, &e.name, &e.version);
if chain.status(&id).is_some() {
drives.push((id.name.clone(), self.walk(chain, src, &id, None)?, false));
continue;
}
let fetched = match apk::fetch_package(
self.http.as_ref(),
&src.cfg.url,
arch,
e,
&idx,
src.auth(),
src.cap(),
) {
Ok(f) => f,
Err(err) if i == 0 => return Err(err),
Err(err) => {
log::warn!(
"pull-through {source}: {} (needed by {name}) could not be pulled: {err:#}",
id.name
);
continue;
}
};
match fetched {
apk::Package::Verified(bytes) => {
let d = self.drive(chain, src, &id, || Ok((bytes, material.to_vec())))?;
if !d.2 {
let rec = ApkLine {
source: source.clone(),
arch: arch.into(),
name: e.name.clone(),
version: e.version.clone(),
stanza: e.stanza.clone(),
};
if let Err(err) = self.append("apk-index.jsonl", &rec) {
log::warn!(
"pull-through: the apk index stanza could not be recorded: {err}"
);
}
}
drives.push(d);
}
apk::Package::Refused(why) => drives.push((id.name.clone(), why, true)),
apk::Package::Missing if i == 0 => {
return Ok(self.unknown(
src,
want,
format!(
"source '{source}' names {} in its index and does not serve it",
id.name
),
));
}
apk::Package::Missing => {
log::info!(
"pull-through {source}: {} (needed by {name}) is not served by the source",
id.name
);
}
}
}
// Only the asked-for package decides the answer; its closure rides along.
let root_id = apk::id(source, arch, &root.name, &root.version).name;
if drives.first().is_some_and(|d| d.0 == root_id && d.2) {
let why = drives[0].1.clone();
self.record(source, want, "refused", vec![root_id], &why);
return Ok(why);
}
Ok(self.record_drives(source, want, drives))
}
}
/// The POMs of a Maven source as the closure reads them: each fetched once against its
/// published checksums, its bytes and material kept for the drive that ingests it.
struct MavenFetched<'a> {
pull: &'a PullIntake,
src: &'a SourceRt,
#[allow(clippy::type_complexity)]
files: std::cell::RefCell<
std::collections::HashMap<String, (Vec<u8>, Vec<super::provenance::Material>)>,
>,
}
impl MavenFetched<'_> {
/// The bytes of the POM at `path`, fetched once against its published checksums (never
/// parsed here: the closure's jobs read them).
fn pom_bytes(&self, path: &str) -> anyhow::Result<Option<Vec<u8>>> {
if let Some((b, _)) = self.files.borrow().get(path) {
return Ok(Some(b.clone()));
}
// D3: a POM is fetched under the POM cap, never the source's artifact cap (4 GiB by
// default).
let got = maven::fetch_file(
self.pull.http.as_ref(),
&self.src.cfg.url,
path,
self.src.auth(),
self.src.cap().min(super::pom::MAX_POM_BYTES),
)?;
Ok(got.map(|(b, m)| {
self.files
.borrow_mut()
.insert(path.to_string(), (b.clone(), m));
b
}))
}
/// The bytes of `<group path>/<artifact>/maven-metadata.xml` (`key` is
/// `<group path>/<artifact>`), never parsed here.
fn metadata_bytes(&self, key: &str) -> anyhow::Result<Option<Vec<u8>>> {
let Some((group_path, artifact)) = key.rsplit_once('/') else {
return Ok(None);
};
Ok(maven::metadata_text(
self.pull.http.as_ref(),
&self.src.cfg.url,
group_path,
artifact,
self.src.auth(),
)?
.map(String::into_bytes))
}
}