server/lib/src/scanner/wheel.rs
//! Python wheels: read a wheel's METADATA, and the wheel SBOM scanner.
//!
//! This is the ONE core-metadata reader. The licence scanner
//! (`scanner::license`) reads a wheel's METADATA and an sdist's PKG-INFO through
//! [`parse_metadata`] and [`read_metadata`]; nothing else parses the format.
//!
//! A wheel is a zip with one `<name>-<version>.dist-info/METADATA` at its top
//! level, written in the core-metadata header format (`Name:`, `Version:`,
//! `License-Expression:`, `Requires-Dist:` …). The zip is read through
//! `holger-zip-entry`, which refuses traversal paths and decompression bombs and
//! never trusts the declared sizes.
//!
//! The SBOM is built through the existing CycloneDX code: modgunn's
//! `to_cyclonedx` via [`crate::sbom::build_cyclonedx_bound`], which binds the
//! document to the wheel's SHA-256. The wheel's declared dependencies are added
//! as CycloneDX components with a `dependencies` entry each. A wheel only
//! declares version ranges, so those components carry the range as a property
//! and no version.
use serde_json::{json, Value};
use super::{FindingSeverity, ScanFinding, ScanOutcome, ScanSubject, Scanner, ScannerKind};
use crate::attest::ArtifactDigest;
use crate::sbom::{build_cyclonedx_bound, SbomComponent, SbomInput};
/// METADATA is small. 1 MiB is far beyond any real one and bounds a hostile one.
pub(crate) const MAX_METADATA_BYTES: usize = 1024 * 1024;
/// The purl type of a Python package.
pub const PURL_TYPE: &str = "pypi";
/// What a wheel says about itself.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct WheelMetadata {
pub name: String,
pub version: String,
/// `License-Expression` (an SPDX expression), else `License`. What the SBOM
/// names.
pub license: Option<String>,
/// `License-Expression` alone (PEP 639). The licence scanner reads it first.
pub license_expression: Option<String>,
/// The free-text `License` field alone. The licence scanner reads it only
/// when it parses as an SPDX expression.
pub license_field: Option<String>,
/// `License ::` trove classifiers.
pub license_classifiers: Vec<String>,
/// Every `Requires-Dist` line, as written.
pub requires_dist: Vec<String>,
}
/// Does the file name look like a wheel?
pub fn is_wheel(file_name: &str) -> bool {
file_name.to_ascii_lowercase().ends_with(".whl")
}
/// Is `member` the top-level `*.dist-info/METADATA`?
fn is_metadata_member(member: &str) -> bool {
let mut parts = member.split('/');
matches!(
(parts.next(), parts.next(), parts.next()),
(Some(dir), Some("METADATA"), None) if dir.ends_with(".dist-info")
)
}
/// Parse core-metadata headers. The body after the first empty line is the
/// long description and is ignored.
pub fn parse_metadata(text: &str) -> anyhow::Result<WheelMetadata> {
let mut fields: Vec<(String, String)> = Vec::new();
for line in text.lines() {
if line.is_empty() {
break;
}
if line.starts_with([' ', '\t']) {
// A continuation line belongs to the previous header.
if let Some((_, v)) = fields.last_mut() {
v.push('\n');
v.push_str(line.trim());
}
continue;
}
if let Some((k, v)) = line.split_once(':') {
fields.push((k.trim().to_ascii_lowercase(), v.trim().to_string()));
}
}
let get = |k: &str| fields.iter().find(|(f, _)| f == k).map(|(_, v)| v.clone());
let all = |k: &str| -> Vec<String> {
fields
.iter()
.filter(|(f, _)| f == k)
.map(|(_, v)| v.clone())
.collect()
};
let name = get("name").filter(|s| !s.is_empty());
let version = get("version").filter(|s| !s.is_empty());
let (Some(name), Some(version)) = (name, version) else {
anyhow::bail!("METADATA has no Name and Version");
};
let stated = |k: &str| get(k).filter(|s| !s.is_empty() && !s.eq_ignore_ascii_case("unknown"));
let license_expression = stated("license-expression");
let license_field = stated("license");
Ok(WheelMetadata {
name,
version,
license: license_expression.clone().or_else(|| license_field.clone()),
license_expression,
license_field,
license_classifiers: all("classifier")
.into_iter()
.filter(|c| c.starts_with("License ::"))
.collect(),
requires_dist: all("requires-dist"),
})
}
/// Read a wheel's METADATA.
pub fn read_metadata(wheel: &[u8]) -> anyhow::Result<WheelMetadata> {
let Some((_, bytes)) =
zip_entry::find_member_capped(wheel, MAX_METADATA_BYTES, is_metadata_member)?
else {
anyhow::bail!("the wheel has no top-level *.dist-info/METADATA");
};
let text = String::from_utf8(bytes).map_err(|_| anyhow::anyhow!("METADATA is not UTF-8"))?;
parse_metadata(&text)
}
/// The distribution name a `Requires-Dist` line names (PEP 508: the name runs up
/// to the first character that is not a letter, digit, `-`, `_` or `.`).
pub fn requirement_name(req: &str) -> Option<&str> {
let req = req.trim();
let end = req
.find(|c: char| !(c.is_ascii_alphanumeric() || c == '-' || c == '_' || c == '.'))
.unwrap_or(req.len());
(end > 0).then(|| &req[..end])
}
/// PEP 503 normalization: lowercase, runs of `-_.` become one `-`.
pub fn normalize_name(name: &str) -> String {
let mut out = String::with_capacity(name.len());
let mut sep = false;
for c in name.chars() {
if c == '-' || c == '_' || c == '.' {
sep = true;
} else {
if sep && !out.is_empty() {
out.push('-');
}
sep = false;
out.push(c.to_ascii_lowercase());
}
}
out
}
/// ★ The CycloneDX SBOM for a wheel, bound to its SHA-256.
pub fn wheel_sbom(meta: &WheelMetadata, file_name: &str, digest: &ArtifactDigest) -> Value {
let input = SbomInput {
subject: file_name.to_string(),
components: vec![SbomComponent {
ecosystem: PURL_TYPE.into(),
name: normalize_name(&meta.name),
version: meta.version.clone(),
license: meta.license.clone(),
}],
vulns: Vec::new(),
};
let bytes = build_cyclonedx_bound(&input, digest);
let mut doc: Value =
serde_json::from_slice(&bytes).expect("the document just emitted re-parses");
// `metadata.component` already carries the `sha256:<hex>` bom-ref; the
// wheel's own component is referenced by its purl, so the two stay unique.
let mut root_ref = String::new();
if let Some(root) = doc.pointer_mut("/components/0") {
root_ref = root["purl"].as_str().unwrap_or_default().to_string();
root["bom-ref"] = json!(root_ref.clone());
root["hashes"] = json!([{ "alg": "SHA-256", "content": digest.hex() }]);
if !meta.license_classifiers.is_empty() {
root["properties"] = json!(meta
.license_classifiers
.iter()
.map(|c| json!({ "name": "python:classifier", "value": c }))
.collect::<Vec<_>>());
}
}
let mut depends_on = Vec::new();
if let Some(components) = doc["components"].as_array_mut() {
for req in &meta.requires_dist {
let Some(name) = requirement_name(req) else {
continue;
};
let name = normalize_name(name);
let bom_ref = format!("pkg:{PURL_TYPE}/{name}");
if depends_on.contains(&bom_ref) {
continue;
}
components.push(json!({
"type": "library",
"name": name,
"bom-ref": bom_ref.clone(),
"purl": bom_ref.clone(),
"scope": if req.contains("extra ==") || req.contains("extra==") { "optional" } else { "required" },
"properties": [{ "name": "python:requires-dist", "value": req }],
}));
depends_on.push(bom_ref);
}
}
doc["dependencies"] = json!([{ "ref": root_ref, "dependsOn": depends_on }]);
doc
}
/// The SBOM generator for Python wheels.
#[derive(Debug, Clone, Copy, Default)]
pub struct WheelSbomScanner;
impl WheelSbomScanner {
pub const ID: &'static str = "holger-wheel-sbom";
}
impl Scanner for WheelSbomScanner {
fn id(&self) -> &str {
Self::ID
}
fn version(&self) -> String {
format!(
"holger-server-lib {}; CycloneDX 1.5",
env!("CARGO_PKG_VERSION")
)
}
fn kind(&self) -> ScannerKind {
ScannerKind::Sbom
}
fn scan(&self, subject: &ScanSubject<'_>) -> anyhow::Result<ScanOutcome> {
if !is_wheel(subject.file_name) {
return Ok(ScanOutcome::pass(format!(
"{:?} is not a wheel; this generator only builds SBOMs for wheels",
subject.file_name
)));
}
let meta = match read_metadata(subject.bytes) {
Ok(m) => m,
Err(e) => {
return Ok(ScanOutcome::error(format!(
"the wheel's METADATA cannot be read: {e:#}"
)))
}
};
let digest = ArtifactDigest::of_sha256(subject.bytes);
let doc = wheel_sbom(&meta, subject.file_name, &digest);
let mut findings = Vec::new();
if meta.license.is_none() && meta.license_classifiers.is_empty() {
findings.push(ScanFinding::new(
"sbom-license-unknown",
FindingSeverity::Info,
"the wheel declares no license",
));
}
let detail = format!(
"CycloneDX SBOM for {} {} with {} declared dependenc(ies)",
meta.name,
meta.version,
meta.requires_dist.len()
);
Ok(ScanOutcome {
findings,
..ScanOutcome::pass(detail)
}
.with_document(doc))
}
}
#[cfg(test)]
mod tests {
use super::*;
const META: &str = "Metadata-Version: 2.4\nName: Demo_Pkg\nVersion: 1.2.0\nLicense-Expression: MIT\nClassifier: License :: OSI Approved :: MIT License\nClassifier: Programming Language :: Python\nRequires-Dist: requests>=2.31\nRequires-Dist: rich; extra == \"cli\"\nDescription-Content-Type: text/plain\n\nName: not-a-header\n";
#[test]
fn metadata_headers_are_read_and_the_body_is_not() {
let m = parse_metadata(META).unwrap();
assert_eq!((m.name.as_str(), m.version.as_str()), ("Demo_Pkg", "1.2.0"));
assert_eq!(m.license.as_deref(), Some("MIT"));
assert_eq!(
m.license_classifiers,
vec!["License :: OSI Approved :: MIT License"]
);
assert_eq!(m.requires_dist.len(), 2);
assert!(parse_metadata("Metadata-Version: 2.1\n").is_err());
}
#[test]
fn names_are_normalized_and_requirements_named() {
assert_eq!(normalize_name("Demo_Pkg"), "demo-pkg");
assert_eq!(normalize_name("zope.interface"), "zope-interface");
assert_eq!(requirement_name("requests>=2.31"), Some("requests"));
assert_eq!(requirement_name("rich ; extra == 'cli'"), Some("rich"));
assert_eq!(requirement_name(">=1"), None);
}
#[test]
fn a_wheel_without_metadata_is_an_error_not_a_pass() {
let whl = zip_entry::fixture::zip(&[("demo/__init__.py", b"")]);
let out = WheelSbomScanner
.scan(&ScanSubject::new("demo-1.0-py3-none-any.whl", "pip", &whl))
.unwrap();
assert_eq!(
out.verdict,
super::super::ScanVerdict::Error,
"{}",
out.detail
);
}
}