guests/static-scan/src/checks.rs
// SPDX-License-Identifier: AGPL-3.0-only
//! The five checks, each a thin adapter over modgunn's own code (nothing is
//! judged twice). Heimdall is a defensive sandbox. Every path that cannot
//! judge returns an `unknown` reason; none of them falls through to clean.
use heimdall_scan::report::unknown;
use heimdall_scan::{purl::pep503, Check, CheckResult, Finding, Purl, Severity, UnknownReason};
use modgunn::core::ArtifactRef;
use modgunn::scan::Advisory;
use modgunn::scip::{typosquat_findings, DepNode, WorldSnapshot};
use crate::archive::{Kind, Walk};
use crate::params::Origin;
pub fn unknown(code: &str, detail: impl Into<String>) -> Option<UnknownReason> {
Some(UnknownReason {
code: code.into(),
detail: detail.into(),
})
}
fn severity(s: modgunn::core::Severity) -> Severity {
use modgunn::core::Severity as M;
match s {
M::Info => Severity::Info,
M::Low => Severity::Low,
M::Medium => Severity::Medium,
M::High => Severity::High,
M::Critical => Severity::Critical,
}
}
fn from_modgunn(f: modgunn::core::Finding, path: Option<&str>) -> Finding {
Finding {
id: f.id,
severity: severity(f.severity),
summary: f.summary,
path: path.map(str::to_owned),
}
}
fn from_advisory(a: &Advisory) -> Finding {
Finding {
id: a.id.clone(),
severity: severity(a.severity),
summary: a.summary.clone(),
path: None,
}
}
/// The finding id of executable lines in a wheel's `.pth` file.
pub const PTH_EXEC: &str = "HEIMDALL-SCAN-WHEEL-PTH-EXEC";
// ── behavior ────────────────────────────────────────────────────────────────
pub fn behavior(purl: &Purl, walk: &Walk) -> CheckResult {
let mut findings = walk.findings.clone();
let why = if let Some(u) = &walk.unreadable {
unknown(unknown::ARCHIVE_UNREADABLE, u.clone())
} else {
let why = match purl.ecosystem() {
"npm" => npm(walk, &mut findings),
"pypi" => pypi(walk, &mut findings),
eco => unknown(
unknown::UNSUPPORTED_ECOSYSTEM,
format!("no install-behavior reader for {eco} packages"),
),
};
// An incomplete walk can never be clean, whatever was read; the cap
// is also why a manifest may be missing, so it is the reason given.
match &walk.over_limits {
Some(o) => unknown(unknown::ARCHIVE_OVER_LIMITS, o.clone()),
None => why,
}
};
CheckResult::new(Check::Behavior, findings, why)
}
fn text(k: &crate::archive::Kept) -> Result<&str, Option<UnknownReason>> {
std::str::from_utf8(&k.bytes)
.map_err(|_| unknown(unknown::NO_MANIFEST, format!("{} is not UTF-8", k.name)))
}
fn npm(walk: &Walk, findings: &mut Vec<Finding>) -> Option<UnknownReason> {
let manifests: Vec<_> = walk
.kept
.iter()
.filter(|k| k.name.ends_with("/package.json"))
.collect();
if manifests.is_empty() {
return unknown(unknown::NO_MANIFEST, "no <dir>/package.json in the archive");
}
// Every copy is judged: npm reads `package/package.json`, a scanner must
// not be steered to a decoy.
for k in manifests {
let t = match text(k) {
Ok(t) => t,
Err(u) => return u,
};
match modgunn::behavior::parse_npm_package(t) {
Ok(b) => findings.extend(
b.findings()
.into_iter()
.map(|f| from_modgunn(f, Some(&k.name))),
),
Err(e) => {
return unknown(
unknown::NO_MANIFEST,
format!("{} is not a package.json: {e}", k.name),
)
}
}
}
None
}
fn is_wheel(walk: &Walk) -> bool {
walk.kind == Some(Kind::Zip)
&& walk
.kept
.iter()
.any(|k| k.name.ends_with("/WHEEL") && k.name.contains(".dist-info/"))
}
fn pypi(walk: &Walk, findings: &mut Vec<Finding>) -> Option<UnknownReason> {
if is_wheel(walk) {
// A wheel runs no install code; its one startup hook is a top-level
// `.pth` file, whose `import` lines run at every interpreter start.
for k in walk.kept.iter().filter(|k| !k.name.contains('/')) {
let t = match text(k) {
Ok(t) => t,
Err(u) => return u,
};
let exec: Vec<&str> = t
.lines()
.filter(|l| l.starts_with("import ") || l.starts_with("import\t"))
.collect();
if exec.is_empty() {
continue;
}
findings.push(Finding {
id: PTH_EXEC.into(),
severity: Severity::Medium,
summary: format!(
"the wheel installs a .pth file with {} line(s) that run at every Python start",
exec.len()
),
path: Some(k.name.clone()),
});
let b = modgunn::behavior::parse_pypi_setup(&exec.join("\n"));
findings.extend(
b.findings()
.into_iter()
.filter(|f| f.id != "MODGUNN-BEHAVIOR-INSTALL-HOOK")
.map(|f| from_modgunn(f, Some(&k.name))),
);
}
return None;
}
let mut read = 0;
for k in &walk.kept {
let file = k.name.rsplit('/').next().unwrap_or("");
if !k.name.contains('/') {
continue;
}
let parse: fn(&str) -> modgunn::behavior::PackageBehavior = match file {
"setup.py" => modgunn::behavior::parse_pypi_setup,
"pyproject.toml" => modgunn::behavior::parse_pyproject_toml,
"setup.cfg" => modgunn::behavior::parse_setup_cfg,
_ => continue,
};
let t = match text(k) {
Ok(t) => t,
Err(u) => return u,
};
read += 1;
findings.extend(
parse(t)
.findings()
.into_iter()
.map(|f| from_modgunn(f, Some(&k.name))),
);
}
if read == 0 {
return unknown(
unknown::NO_MANIFEST,
"no setup.py, pyproject.toml or setup.cfg in the sdist, and no wheel metadata",
);
}
None
}
// ── typosquat + dependency confusion ───────────────────────────────────────
/// A name list: UTF-8, one per line, `#` comments and blank lines skipped.
fn lines(b: &[u8]) -> Result<Vec<&str>, Option<UnknownReason>> {
let t = std::str::from_utf8(b)
.map_err(|_| unknown(unknown::DATA_INVALID, "the name list is not UTF-8"))?;
Ok(t.lines()
.map(str::trim)
.filter(|l| !l.is_empty() && !l.starts_with('#'))
.collect())
}
fn snapshot(purl: &Purl, name: &str) -> WorldSnapshot {
let mut s = WorldSnapshot::new("heimdall-static-scan");
s.dep_nodes = vec![DepNode::new(
purl.ecosystem(),
name,
purl.version.as_deref().unwrap_or(""),
)];
s
}
pub fn typosquat(purl: &Purl, data: Option<&[u8]>) -> CheckResult {
let r = |f, u| CheckResult::new(Check::Typosquat, f, u);
let Some(data) = data else {
return r(
vec![],
unknown(unknown::DATA_MISSING, "no popular-names snapshot"),
);
};
let names = match lines(data) {
Ok(n) => n,
Err(u) => return r(vec![], u),
};
if names.is_empty() {
return r(
vec![],
unknown(unknown::DATA_INVALID, "the popular-names snapshot is empty"),
);
}
let pypi = purl.ecosystem() == "pypi";
let norm: Vec<String> = names
.iter()
.map(|n| if pypi { pep503(n) } else { (*n).to_owned() })
.collect();
let popular: Vec<&str> = norm.iter().map(String::as_str).collect();
let name = purl.package_name();
let f = typosquat_findings(&snapshot(purl, &name), &popular, &[]);
r(f.into_iter().map(|f| from_modgunn(f, None)).collect(), None)
}
pub fn dependency_confusion(
purl: &Purl,
data: Option<&[u8]>,
origin: Option<Origin>,
) -> CheckResult {
let r = |f, u| CheckResult::new(Check::DependencyConfusion, f, u);
let Some(data) = data else {
return r(
vec![],
unknown(unknown::DATA_MISSING, "no internal-namespaces snapshot"),
);
};
let prefixes = match lines(data) {
Ok(n) => n,
Err(u) => return r(vec![], u),
};
match origin {
None => r(
vec![],
unknown(
unknown::ORIGIN_MISSING,
"the job did not say where the package was resolved (`origin`)",
),
),
Some(Origin::Internal) => r(vec![], None),
Some(Origin::Public) => {
let name = purl.package_name();
let f = typosquat_findings(&snapshot(purl, &name), &[], &prefixes);
r(f.into_iter().map(|f| from_modgunn(f, None)).collect(), None)
}
}
}
// ── registry + malicious ───────────────────────────────────────────────────
fn artifact(purl: &Purl, sha256: &str) -> ArtifactRef {
ArtifactRef {
ecosystem: purl.ecosystem().to_owned(),
name: purl.package_name(),
version: purl.version.clone().unwrap_or_default(),
blob_sha256: sha256.to_owned(),
}
}
/// Registry names compare as the registry does: PyPI by PEP 503, crates.io
/// ignoring case and `-`/`_`, others exactly.
fn same_name(eco: &str, a: &str, b: &str) -> bool {
match eco {
"pypi" => pep503(a) == pep503(b),
"cargo" => {
let f = |s: &str| s.to_ascii_lowercase().replace('_', "-");
f(a) == f(b)
}
_ => a == b,
}
}
#[derive(serde::Deserialize)]
struct CratesDoc {
#[serde(rename = "crate")]
krate: Option<Named>,
versions: Vec<CratesVersion>,
}
#[derive(serde::Deserialize)]
struct Named {
name: String,
}
#[derive(serde::Deserialize)]
struct CratesVersion {
num: String,
#[serde(rename = "crate", default)]
crate_name: String,
}
#[derive(serde::Deserialize)]
struct NpmDoc {
name: String,
versions: std::collections::BTreeMap<String, serde_json::Value>,
}
#[derive(serde::Deserialize)]
struct PypiDoc {
info: Named,
releases: std::collections::BTreeMap<String, serde_json::Value>,
}
/// Which package the registry document is about, and its versions (`None`
/// for go.mod, which lists only retractions).
fn identity(eco: &str, t: &str) -> Result<(String, Option<Vec<String>>), String> {
let bad = |e: serde_json::Error| format!("the registry document does not read: {e}");
match eco {
"cargo" => {
let d: CratesDoc = serde_json::from_str(t).map_err(bad)?;
let name = d
.krate
.map(|k| k.name)
.or_else(|| d.versions.first().map(|v| v.crate_name.clone()))
.unwrap_or_default();
Ok((name, Some(d.versions.into_iter().map(|v| v.num).collect())))
}
"npm" => {
let d: NpmDoc = serde_json::from_str(t).map_err(bad)?;
Ok((d.name, Some(d.versions.into_keys().collect())))
}
"pypi" => {
let d: PypiDoc = serde_json::from_str(t).map_err(bad)?;
Ok((d.info.name, Some(d.releases.into_keys().collect())))
}
"go" => {
let m = t
.lines()
.map(str::trim)
.find_map(|l| l.strip_prefix("module "))
.map(|m| m.trim().trim_matches('"').to_owned())
.ok_or("the go.mod has no module line")?;
Ok((m, None))
}
_ => unreachable!("checked by the caller"),
}
}
pub fn registry(purl: &Purl, data: Option<&[u8]>, sha256: &str) -> CheckResult {
let r = |f, u| CheckResult::new(Check::Registry, f, u);
let eco = purl.ecosystem();
if !matches!(eco, "cargo" | "npm" | "pypi" | "go") {
return r(
vec![],
unknown(
unknown::UNSUPPORTED_ECOSYSTEM,
format!("no registry reader for {eco}"),
),
);
}
let Some(version) = purl.version.as_deref() else {
return r(
vec![],
unknown(unknown::NO_VERSION, "the purl has no version"),
);
};
let Some(data) = data else {
return r(
vec![],
unknown(unknown::DATA_MISSING, "no registry snapshot"),
);
};
let Ok(t) = std::str::from_utf8(data) else {
return r(
vec![],
unknown(unknown::DATA_INVALID, "the registry document is not UTF-8"),
);
};
let name = purl.package_name();
match identity(eco, t) {
Err(e) => return r(vec![], unknown(unknown::DATA_INVALID, e)),
Ok((doc_name, versions)) => {
if !same_name(eco, &doc_name, &name) {
return r(
vec![],
unknown(
unknown::DATA_INVALID,
format!("the registry document is about {doc_name:?}, not {name:?}"),
),
);
}
if let Some(vs) = versions {
if !vs.iter().any(|v| v == version) {
return r(
vec![],
unknown(
unknown::DATA_INVALID,
format!("version {version} is not in the registry document (stale snapshot?)"),
),
);
}
}
}
}
let advs = match eco {
"cargo" => modgunn::registry::parse_crates_yanks(t),
"npm" => modgunn::registry::parse_npm_deprecations(t),
"pypi" => modgunn::registry::parse_pypi_yanks(t),
_ => Ok(modgunn::registry::parse_go_retractions(t)),
};
let advs = match advs {
Ok(a) => a,
Err(e) => {
return r(
vec![],
unknown(
unknown::DATA_INVALID,
format!("the registry document does not read: {e}"),
),
)
}
};
let a = artifact(purl, sha256);
r(
advs.iter()
.filter(|adv| adv.matches(&a))
.map(from_advisory)
.collect(),
None,
)
}
pub fn malicious(purl: &Purl, data: Option<&[u8]>, sha256: &str) -> CheckResult {
let r = |f, u| CheckResult::new(Check::Malicious, f, u);
let Some(data) = data else {
return r(
vec![],
unknown(unknown::DATA_MISSING, "no malicious-package feed"),
);
};
let Ok(t) = std::str::from_utf8(data) else {
return r(
vec![],
unknown(unknown::DATA_INVALID, "the feed is not UTF-8"),
);
};
let advs = if matches!(serde_json::from_str::<Vec<serde_json::Value>>(t), Ok(v) if v.is_empty())
{
Vec::new()
} else {
match modgunn::feeds::parse_malicious(t) {
Ok(a) => a,
Err(e) => {
return r(
vec![],
unknown(
unknown::DATA_INVALID,
format!("the feed does not read: {e}"),
),
)
}
}
};
let a = artifact(purl, sha256);
let hits: Vec<Finding> = match purl.version {
Some(_) => advs
.iter()
.filter(|adv| adv.matches(&a))
.map(from_advisory)
.collect(),
// Without a version only whole-package records can match.
None => advs
.iter()
.filter(|adv| {
adv.affected_versions.is_empty() && adv.ranges.is_empty() && adv.matches(&a)
})
.map(from_advisory)
.collect(),
};
let why = if purl.version.is_none() {
unknown(
unknown::NO_VERSION,
"the purl has no version; only whole-package records were matched",
)
} else {
None
};
r(hits, why)
}