server/lib/src/pull_intake/depset.rs
//! **A pull's dependency closure, approved as ONE set** (lane HOL-DEPSET-APPROVE).
//!
//! A pull of one package ([`super::Want`]) brings its whole dependency closure into the
//! intake chain at once (Maven: [`super::pom::closure`]; npm, cargo and Python from their
//! registries' metadata, [`super::closures`]). Every member is fetched, held against its
//! digests and walked through the chain like any pull: the inspector, the malware engines,
//! the scanners and provenance, one artifact at a time. The set is then offered to the
//! approvers as ONE item:
//!
//! * **approved as a whole**: an approver's set approval casts that approver's vote on every
//! member waiting at POLICY_CHECKED. The two-person rule is the chain's own and is
//! unchanged: each member needs its distinct approvers, never its importer;
//! * **refused as a whole**: a refusal blocks every member still waiting;
//! * **a member with a refusal reason** (a CVE, malware, a licence, provenance: the chain
//! BLOCKED it, or the door refused it) refuses the set, named;
//! * **fail closed**: a member that could not be resolved or scanned (or a set still
//! resolving) holds the set, named; nothing in it is approved.
//!
//! The record is `<prod_root>/lineage/pull-intake/depsets.jsonl` (append-only; the newest
//! record of an id wins). Holger serves the sets at `GET /-/intake/sets`, one at
//! `GET /-/intake/set?id=`, and the verbs at `POST /-/intake/set/approve` and
//! `POST /-/intake/set/refuse` (`?id=&reason=`); holger-ui renders them.
use serde::{Deserialize, Serialize};
use traits::ArtifactId;
use super::{PullIntake, Want};
use crate::auth::{AuthConfig, AuthIdentity};
use crate::promote::duties::{self, Verb};
use crate::promote::intake::{
ApproveOutcome, Approver, ArtifactState, IntakeChain, TransitionRequest,
};
/// The record of the sets.
const DEPSETS: &str = "depsets.jsonl";
/// One member of a set: a coordinate in the chain.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct DepSetMember {
/// The chain coordinate (`central/org/apache/…/x-1.0.jar`).
pub name: String,
/// `requested`, `parent`, `bom`, `descriptor`, `dependency`.
pub role: String,
/// What needed it (`group:artifact:version`, `package@version`).
#[serde(default)]
pub via: String,
/// The door's refusal when the member never entered the chain.
#[serde(default)]
pub refused: Option<String>,
}
/// **One set**, as recorded when its pull ran.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct DepSetRecord {
pub id: String,
pub source: String,
/// The source's kind (`maven`, `npm`, `crates`, `python`).
pub ecosystem: String,
/// What was asked for, in the ecosystem's words.
pub root: String,
pub want: Want,
pub at: i64,
pub members: Vec<DepSetMember>,
/// `(coordinate, why)`: what could not be resolved. Each holds the set.
#[serde(default)]
pub unresolved: Vec<(String, String)>,
/// What the pull says beside the set (an optional dependency left out, a version with
/// no metadata to read its dependencies from). Notes never hold a set.
#[serde(default)]
pub notes: Vec<String>,
}
/// Where a set stands.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum SetStatus {
/// The pull is still fetching and scanning members.
Resolving,
/// A member could not be resolved or scanned: nothing is approved (fail closed).
Held,
/// A member carries a refusal reason: the set is refused.
Refused,
/// Every member is scanned and policy-checked; the set waits for its approvers.
AwaitingApproval,
/// Every member is approved; the seal and promotion run.
Approved,
/// Every member is AVAILABLE.
Available,
}
impl SetStatus {
pub fn name(self) -> &'static str {
match self {
SetStatus::Resolving => "resolving",
SetStatus::Held => "held",
SetStatus::Refused => "refused",
SetStatus::AwaitingApproval => "awaiting-approval",
SetStatus::Approved => "approved",
SetStatus::Available => "available",
}
}
}
/// One member as an approver is shown it.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct MemberView {
pub name: String,
pub role: String,
pub via: String,
/// The chain state (`policy-checked` …), or `not-in-chain`.
pub state: String,
pub sha256: String,
/// The votes cast on its current bytes.
pub approvals: u32,
pub approvers: Vec<String>,
/// Why it is refused or held, when it is.
pub reason: String,
/// Advisories recorded as a warning (a Maven POM's).
pub warnings: Vec<String>,
}
/// **One set as an approver is shown it.**
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct SetView {
pub id: String,
/// What a vote names (SEC-REVIEW D4): 16 hex of the sha256 over the members and their
/// sha256 as shown, so a re-pull that changes them changes it.
pub revision: String,
pub source: String,
pub ecosystem: String,
pub root: String,
pub at: i64,
pub status: SetStatus,
/// The words for the status: what it waits for, or why it is refused or held.
pub summary: String,
pub members: Vec<MemberView>,
/// `(member or coordinate, why)`: the members that refuse the set.
pub blockers: Vec<(String, String)>,
/// `(member or coordinate, why)`: what holds the set.
pub holds: Vec<(String, String)>,
pub approvals_required: u32,
/// The approvers who voted on EVERY member still waiting.
pub approvers: Vec<String>,
/// How many members wait at POLICY_CHECKED.
pub awaiting: usize,
pub available: usize,
pub notes: Vec<String>,
}
/// What a set verb did.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct SetAnswer {
pub id: String,
pub status: SetStatus,
/// Members this call voted on (approve) or blocked (refuse).
pub moved: usize,
pub message: String,
pub set: SetView,
}
/// A set verb refused: the HTTP status and the tagged words.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SetError {
pub status: u16,
pub message: String,
}
impl std::fmt::Display for SetError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(&self.message)
}
}
impl std::error::Error for SetError {}
impl SetError {
fn of(e: crate::promote::intake::IntakeError) -> Self {
Self {
status: e.status(),
message: e.tagged(),
}
}
}
/// A set's id: 16 hex of the sha256 of its source and want.
pub fn set_id(source: &str, want: &Want) -> String {
let d =
crate::attest::ArtifactDigest::of_sha256(format!("{source}\n{}", want.key()).as_bytes());
d.hex()[..16].to_string()
}
fn id_of(name: &str) -> ArtifactId {
ArtifactId {
namespace: None,
name: name.to_string(),
version: String::new(),
}
}
/// `policy-checked`, `available` …: the state as the sets name it.
fn state_name(s: ArtifactState) -> String {
s.name().to_ascii_lowercase().replace('_', "-")
}
/// What a set's members' drives say: `(name, line, refused at the door)` → members.
pub(super) fn members_of(
drives: &[(String, String, bool)],
roles: &dyn Fn(&str) -> (String, String),
) -> Vec<DepSetMember> {
drives
.iter()
.map(|(name, line, refused)| {
let (role, via) = roles(name);
DepSetMember {
name: name.clone(),
role,
via,
refused: refused.then(|| line.clone()),
}
})
.collect()
}
impl PullIntake {
/// Record a set (its pull ran). The newest record of an id wins.
pub(super) fn record_set(&self, rec: &DepSetRecord) {
if let Err(e) = self.append(DEPSETS, rec) {
log::warn!(
"pull-through: the dependency set {} could not be recorded: {e}",
rec.id
);
}
}
/// Every set, the newest record of each id, newest first.
pub fn set_records(&self) -> Vec<DepSetRecord> {
let mut by_id: std::collections::BTreeMap<String, DepSetRecord> =
std::collections::BTreeMap::new();
for r in self.read::<DepSetRecord>(DEPSETS) {
by_id.insert(r.id.clone(), r);
}
let mut out: Vec<DepSetRecord> = by_id.into_values().collect();
out.sort_by(|a, b| b.at.cmp(&a.at).then(a.id.cmp(&b.id)));
out
}
/// The newest record of set `id`.
pub fn set_record(&self, id: &str) -> Option<DepSetRecord> {
self.read::<DepSetRecord>(DEPSETS)
.into_iter()
.rev()
.find(|r| r.id == id)
}
/// The advisories a member's scan recorded as a warning.
fn warnings_of(chain: &IntakeChain, sha256: &str) -> Vec<String> {
let Some(p) = chain.scanner() else {
return Vec::new();
};
let ev = match p.stored(sha256) {
Ok(crate::scanner::Stored::Fresh { evidence, .. })
| Ok(crate::scanner::Stored::Stale { evidence, .. }) => evidence,
_ => return Vec::new(),
};
ev.records_of(crate::scanner::ScannerKind::Vulnerability)
.filter(|r| r.verdict == crate::scanner::ScanVerdict::Warn)
.flat_map(|r| {
r.findings
.iter()
.map(|f| format!("{} ({})", f.id, format!("{:?}", f.severity).to_lowercase()))
})
.collect()
}
/// **Where a set stands**, from its members' states now.
pub fn set_view(&self, chain: &IntakeChain, rec: &DepSetRecord) -> SetView {
let mut members = Vec::new();
let mut blockers = Vec::new();
let mut holds: Vec<(String, String)> = rec
.unresolved
.iter()
.map(|(c, why)| (c.clone(), format!("not resolved: {why}")))
.collect();
let (mut awaiting, mut available, mut approved) = (0usize, 0usize, 0usize);
let mut common: Option<std::collections::BTreeSet<String>> = None;
for m in &rec.members {
let id = id_of(&m.name);
let mut view = MemberView {
name: m.name.clone(),
role: m.role.clone(),
via: m.via.clone(),
state: "not-in-chain".into(),
sha256: String::new(),
approvals: 0,
approvers: Vec::new(),
reason: String::new(),
warnings: Vec::new(),
};
match chain.status(&id) {
None => {
match m
.refused
.clone()
.or_else(|| self.refused_at_door(&rec.source, &m.name))
{
Some(why) => {
view.reason = why.clone();
blockers.push((m.name.clone(), why));
}
None => {
view.reason =
"not in the intake chain: it was not fetched or not scanned yet"
.into();
holds.push((m.name.clone(), view.reason.clone()));
}
}
}
Some(st) => {
view.state = state_name(st.state);
view.sha256 = st.sha256.clone();
view.warnings = Self::warnings_of(chain, &st.sha256);
let votes = chain.pending_approvals(&id);
view.approvals = votes.len() as u32;
view.approvers = votes.iter().map(|v| v.approver.clone()).collect();
match st.state {
ArtifactState::Blocked => {
view.reason = super::last_reason(chain, &id);
blockers.push((m.name.clone(), view.reason.clone()));
}
ArtifactState::Quarantined if chain.invalidation_of(&id).is_some() => {
view.reason = chain.invalidation_of(&id).unwrap_or_default();
blockers.push((m.name.clone(), view.reason.clone()));
}
ArtifactState::PolicyChecked => {
awaiting += 1;
let mine: std::collections::BTreeSet<String> =
view.approvers.iter().cloned().collect();
common = Some(match common {
None => mine,
Some(c) => c.intersection(&mine).cloned().collect(),
});
}
ArtifactState::Available => available += 1,
ArtifactState::Approved
| ArtifactState::Sealed
| ArtifactState::Promoted => approved += 1,
s => {
view.reason = format!(
"stands at {} in the intake chain, not scanned and policy-checked yet",
state_name(s)
);
holds.push((m.name.clone(), view.reason.clone()));
}
}
}
}
members.push(view);
}
let resolving = self.open_job(&rec.source, &rec.want).is_some();
let n = rec.members.len();
let status = if !blockers.is_empty() {
SetStatus::Refused
} else if resolving {
SetStatus::Resolving
} else if !holds.is_empty() || n == 0 {
SetStatus::Held
} else if available == n {
SetStatus::Available
} else if awaiting > 0 {
SetStatus::AwaitingApproval
} else {
SetStatus::Approved
};
let need = chain.approvals_required();
let approvers: Vec<String> = common.unwrap_or_default().into_iter().collect();
let summary = match status {
SetStatus::Refused => format!(
"refused: {} of {n} member(s) carry a refusal reason ({})",
blockers.len(),
blockers
.iter()
.take(3)
.map(|(m, _)| m.as_str())
.collect::<Vec<_>>()
.join(", ")
),
SetStatus::Resolving => format!("resolving: {n} member(s) so far"),
SetStatus::Held if n == 0 => "held: the pull brought in nothing".to_string(),
SetStatus::Held => format!("held: {} member(s) not resolved or not scanned", holds.len()),
SetStatus::AwaitingApproval => format!(
"{awaiting} of {n} member(s) wait for approval; {} of {need} approver(s) approved the whole set",
approvers.len()
),
SetStatus::Approved => format!("approved: {approved} member(s) being sealed and promoted"),
SetStatus::Available => format!("available: all {n} member(s)"),
};
let revision = revision_of(&members);
SetView {
id: rec.id.clone(),
revision,
source: rec.source.clone(),
ecosystem: rec.ecosystem.clone(),
root: rec.root.clone(),
at: rec.at,
status,
summary,
members,
blockers,
holds,
approvals_required: need,
approvers,
awaiting,
available,
notes: rec.notes.clone(),
}
}
/// May the caller read every member of the set?
fn readable(
chain: &IntakeChain,
auth: &AuthConfig,
identity: Option<&AuthIdentity>,
rec: &DepSetRecord,
) -> bool {
let held = duties::duties_of(auth, identity);
let principal = duties::principal_of(identity);
rec.members.iter().all(|m| {
let state = chain
.status(&id_of(&m.name))
.map(|s| s.state)
.unwrap_or(ArtifactState::Untrusted);
duties::authorize(&held, &principal, &m.name, Verb::Read(state)).is_ok()
})
}
/// **The sets the caller may read**, newest first. `status`: only those in it.
pub fn sets_as(
&self,
chain: &IntakeChain,
auth: &AuthConfig,
identity: Option<&AuthIdentity>,
status: Option<&str>,
) -> Vec<SetView> {
self.set_records()
.into_iter()
.filter(|r| Self::readable(chain, auth, identity, r))
.map(|r| self.set_view(chain, &r))
.filter(|v| status.is_none_or(|s| s.is_empty() || v.status.name() == s))
.collect()
}
/// **One set**, when the caller may read it.
pub fn set_as(
&self,
chain: &IntakeChain,
auth: &AuthConfig,
identity: Option<&AuthIdentity>,
id: &str,
) -> Result<SetView, SetError> {
let rec = self.set_record(id).ok_or_else(|| unknown(id))?;
if !Self::readable(chain, auth, identity, &rec) {
return Err(unknown(id));
}
Ok(self.set_view(chain, &rec))
}
/// **Approve a set as a whole.** Refused by name when a member carries a refusal
/// reason, held when one is not resolved or scanned (nothing is voted). Otherwise the
/// caller's vote is cast on every member waiting at POLICY_CHECKED (a member the caller
/// voted on already keeps that vote), through the chain's own two-person rule; the
/// members that reach APPROVED are walked on to AVAILABLE.
#[allow(clippy::too_many_arguments)]
pub fn approve_set(
&self,
chain: &IntakeChain,
auth: &AuthConfig,
identity: Option<&AuthIdentity>,
id: &str,
revision: &str,
reason: &str,
source: &str,
) -> Result<SetAnswer, SetError> {
let rec = self.set_record(id).ok_or_else(|| unknown(id))?;
let principal = duties::principal_of(identity);
let held = duties::duties_of(auth, identity);
let label = format!("dependency set {id} ({})", rec.root);
duties::authorize(&held, &principal, &label, Verb::Approve)
.map_err(|e| SetError::of(crate::promote::intake::IntakeError::Duty(e)))?;
let view = self.set_view(chain, &rec);
match view.status {
SetStatus::Refused => {
return Err(SetError {
status: 409,
message: holger_errcode::DEPSET_REFUSED.tag(format!(
"{label} cannot be approved: {}",
named(&view.blockers)
)),
})
}
SetStatus::Resolving | SetStatus::Held => {
return Err(SetError {
status: 409,
message: holger_errcode::DEPSET_HELD.tag(format!(
"{label} is {}: nothing in it is approved until every member is resolved and scanned{}",
view.status.name(),
if view.holds.is_empty() {
String::new()
} else {
format!(" ({})", named(&view.holds))
}
)),
})
}
_ => {}
}
current(&view, &label, revision)?;
let approver = Approver::from_identity(identity);
let why = if reason.trim().is_empty() {
format!("approved as part of {label} at revision {}", view.revision)
} else {
format!(
"{} (as part of {label} at revision {})",
reason.trim(),
view.revision
)
};
let mut moved = 0;
let mut errors = Vec::new();
for m in &rec.members {
let mid = id_of(&m.name);
if chain.state(&mid) != ArtifactState::PolicyChecked {
continue;
}
if chain
.pending_approvals(&mid)
.iter()
.any(|v| v.approver == approver.principal)
{
continue;
}
let req = TransitionRequest::new(mid.clone(), ArtifactState::Approved, &principal)
.with_reason(why.clone())
.from_source(source);
match chain.approve_or_advance(req, &approver) {
Ok(ApproveOutcome::Pending { .. }) | Ok(ApproveOutcome::Approved { .. }) => {
moved += 1
}
// The first member refuses the caller (the importer, an anonymous caller):
// nothing else is tried, nothing was cast.
Err(e) if moved == 0 => return Err(SetError::of(e)),
Err(e) => errors.push(format!("{}: {}", m.name, e.tagged())),
}
}
// The members that reached APPROVED are sealed and promoted now.
if let Some(src) = self.source(&rec.source) {
let ids: Vec<ArtifactId> = rec.members.iter().map(|m| id_of(&m.name)).collect();
let _ = self.settle(chain, src, &ids);
}
let set = self.set_view(chain, &rec);
let mut message = format!(
"{principal} approved {label}: {moved} vote(s) cast; {}",
set.summary
);
if !errors.is_empty() {
message.push_str(&format!(
"; refused on {}: {}",
errors.len(),
errors.join("; ")
));
}
Ok(SetAnswer {
id: id.to_string(),
status: set.status,
moved,
message,
set,
})
}
/// **Refuse a set as a whole**: every member still waiting at POLICY_CHECKED is
/// BLOCKED with the reason. AVAILABLE members (another set's, approved before) stay.
#[allow(clippy::too_many_arguments)]
pub fn refuse_set(
&self,
chain: &IntakeChain,
auth: &AuthConfig,
identity: Option<&AuthIdentity>,
id: &str,
revision: &str,
reason: &str,
source: &str,
) -> Result<SetAnswer, SetError> {
let rec = self.set_record(id).ok_or_else(|| unknown(id))?;
let principal = duties::principal_of(identity);
let held = duties::duties_of(auth, identity);
let label = format!("dependency set {id} ({})", rec.root);
duties::authorize(&held, &principal, &label, Verb::Block)
.map_err(|e| SetError::of(crate::promote::intake::IntakeError::Duty(e)))?;
if principal == duties::ANONYMOUS {
return Err(SetError::of(crate::promote::intake::IntakeError::Duty(
duties::DutyError::AnonymousApprover { artifact: label },
)));
}
current(&self.set_view(chain, &rec), &label, revision)?;
let why = format!(
"refused as part of {label} at revision {} by {principal}{}",
revision.trim(),
if reason.trim().is_empty() {
String::new()
} else {
format!(": {}", reason.trim())
}
);
let approver = Approver::from_identity(identity);
let mut moved = 0;
for m in &rec.members {
let mid = id_of(&m.name);
if chain.state(&mid) != ArtifactState::PolicyChecked {
continue;
}
let req = TransitionRequest::new(mid, ArtifactState::Blocked, &principal)
.with_reason(why.clone())
.from_source(source);
chain
.approve_or_advance(req, &approver)
.map_err(SetError::of)?;
moved += 1;
}
let set = self.set_view(chain, &rec);
Ok(SetAnswer {
id: id.to_string(),
status: set.status,
moved,
message: format!("{principal} refused {label}: {moved} member(s) blocked"),
set,
})
}
}
/// A set's revision (D4): 16 hex of the sha256 over `name sha256` of each member, by name.
fn revision_of(members: &[MemberView]) -> String {
let mut lines: Vec<String> = members
.iter()
.map(|m| format!("{} {}\n", m.name, m.sha256))
.collect();
lines.sort();
crate::attest::ArtifactDigest::of_sha256(lines.concat().as_bytes()).hex()[..16].to_string()
}
/// `Ok` when the vote names the revision the set has now (D4); a vote naming none or another
/// is refused (409, `holger-depset-4`) before anything is cast.
fn current(view: &SetView, label: &str, revision: &str) -> Result<(), SetError> {
if revision.trim() == view.revision {
return Ok(());
}
Err(SetError {
status: 409,
message: holger_errcode::DEPSET_STALE.tag(format!(
"{label} is at revision {} and the vote names {}: read the set again and judge what it shows",
view.revision,
if revision.trim().is_empty() {
"none".to_string()
} else {
format!("'{}'", revision.trim())
}
)),
})
}
fn unknown(id: &str) -> SetError {
SetError {
status: 404,
message: holger_errcode::DEPSET_UNKNOWN
.tag(format!("no dependency set '{id}' is readable here")),
}
}
fn named(list: &[(String, String)]) -> String {
let mut s: Vec<String> = list
.iter()
.take(5)
.map(|(m, why)| format!("{m}: {why}"))
.collect();
if list.len() > 5 {
s.push(format!("and {} more", list.len() - 5));
}
s.join("; ")
}