175 lines
7 KiB
Python
Executable file
175 lines
7 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""Fail on RustSec findings reachable from nigig-site production dependencies.
|
|
|
|
Cargo.lock is workspace-wide, so cargo-audit can report packages that are not in
|
|
this application's normal/build graph. This script consumes unmodified
|
|
`cargo metadata` and `cargo audit --json` output, traverses every non-dev edge
|
|
for all target predicates, and makes that containment explicit rather than
|
|
ignoring advisory IDs globally.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from collections import deque
|
|
from pathlib import Path
|
|
from typing import Any, NoReturn
|
|
|
|
|
|
def fail(message: str) -> NoReturn:
|
|
print(f"ERROR: {message}", file=sys.stderr)
|
|
raise SystemExit(1)
|
|
|
|
|
|
def load_json(path: str) -> dict[str, Any]:
|
|
try:
|
|
value = json.loads(Path(path).read_text())
|
|
except (OSError, UnicodeError, json.JSONDecodeError) as error:
|
|
fail(f"cannot parse {path}: {error}")
|
|
if not isinstance(value, dict):
|
|
fail(f"expected a JSON object in {path}")
|
|
return value
|
|
|
|
|
|
def main() -> None:
|
|
if len(sys.argv) != 4:
|
|
fail("usage: audit-production-deps.py METADATA.json AUDIT.json REPORT.txt")
|
|
metadata = load_json(sys.argv[1])
|
|
audit = load_json(sys.argv[2])
|
|
|
|
packages = metadata.get("packages")
|
|
resolve = metadata.get("resolve")
|
|
if not isinstance(packages, list) or not isinstance(resolve, dict):
|
|
fail("cargo metadata is missing packages/resolve")
|
|
nodes = resolve.get("nodes")
|
|
if not isinstance(nodes, list):
|
|
fail("cargo metadata is missing resolve.nodes")
|
|
|
|
roots = [
|
|
package
|
|
for package in packages
|
|
if package.get("name") == "nigig-site"
|
|
and str(package.get("manifest_path", "")).replace("\\", "/").endswith(
|
|
"/crates/apps/nigig-site/Cargo.toml"
|
|
)
|
|
]
|
|
if len(roots) != 1:
|
|
fail(f"expected exactly one nigig-site package, found {len(roots)}")
|
|
root_id = roots[0].get("id")
|
|
node_by_id = {node.get("id"): node for node in nodes}
|
|
package_by_id = {package.get("id"): package for package in packages}
|
|
if root_id not in node_by_id or root_id not in package_by_id:
|
|
fail("nigig-site is absent from the resolved graph")
|
|
|
|
reachable: set[str] = set()
|
|
queue: deque[str] = deque([root_id])
|
|
while queue:
|
|
package_id = queue.popleft()
|
|
if package_id in reachable:
|
|
continue
|
|
reachable.add(package_id)
|
|
node = node_by_id.get(package_id)
|
|
if not isinstance(node, dict):
|
|
fail(f"resolved node is missing for {package_id}")
|
|
dependencies = node.get("deps")
|
|
if not isinstance(dependencies, list):
|
|
fail(f"resolved dependencies are malformed for {package_id}")
|
|
for dependency in dependencies:
|
|
kinds = dependency.get("dep_kinds")
|
|
if not isinstance(kinds, list) or not kinds:
|
|
fail(f"dependency kinds are missing for {dependency!r}")
|
|
# `kind: null` is a normal dependency. Build dependencies execute in
|
|
# the production build trust boundary. Dev-only edges are excluded.
|
|
if not any(kind.get("kind") in (None, "build") for kind in kinds):
|
|
continue
|
|
child = dependency.get("pkg")
|
|
if child not in node_by_id:
|
|
fail(f"resolved child node is missing for {child}")
|
|
queue.append(child)
|
|
|
|
reachable_versions = {
|
|
(str(package_by_id[item].get("name")), str(package_by_id[item].get("version")))
|
|
for item in reachable
|
|
}
|
|
|
|
vulnerabilities = audit.get("vulnerabilities")
|
|
warnings = audit.get("warnings")
|
|
database = audit.get("database")
|
|
settings = audit.get("settings")
|
|
if not isinstance(vulnerabilities, dict) or not isinstance(warnings, dict):
|
|
fail("cargo-audit JSON is missing vulnerabilities/warnings")
|
|
if not isinstance(database, dict) or not database.get("last-commit"):
|
|
fail("cargo-audit JSON is missing database provenance")
|
|
if not isinstance(settings, dict):
|
|
fail("cargo-audit JSON is missing settings")
|
|
if settings.get("ignore") != []:
|
|
fail("cargo-audit advisory ignores are forbidden")
|
|
if settings.get("target_arch") != [] or settings.get("target_os") != []:
|
|
fail("cargo-audit target filters are forbidden")
|
|
informational = settings.get("informational_warnings")
|
|
if not isinstance(informational, list) or not {
|
|
"unmaintained",
|
|
"unsound",
|
|
"notice",
|
|
}.issubset(informational):
|
|
fail("cargo-audit informational warnings are not fully enabled")
|
|
vulnerability_list = vulnerabilities.get("list")
|
|
if not isinstance(vulnerability_list, list):
|
|
fail("cargo-audit vulnerability list is malformed")
|
|
if vulnerabilities.get("count") != len(vulnerability_list):
|
|
fail("cargo-audit vulnerability count does not match its list")
|
|
if vulnerabilities.get("found") is not bool(vulnerability_list):
|
|
fail("cargo-audit vulnerability flag does not match its list")
|
|
|
|
findings: list[tuple[str, str, str, str]] = []
|
|
excluded: list[tuple[str, str, str, str]] = []
|
|
|
|
def classify(kind: str, item: dict[str, Any]) -> None:
|
|
package = item.get("package")
|
|
advisory = item.get("advisory")
|
|
if not isinstance(package, dict) or not isinstance(advisory, dict):
|
|
fail(f"malformed cargo-audit {kind} item")
|
|
name = str(package.get("name"))
|
|
version = str(package.get("version"))
|
|
advisory_id = str(advisory.get("id"))
|
|
record = (kind, advisory_id, name, version)
|
|
if (name, version) in reachable_versions:
|
|
findings.append(record)
|
|
else:
|
|
excluded.append(record)
|
|
|
|
for item in vulnerability_list:
|
|
if not isinstance(item, dict):
|
|
fail("malformed cargo-audit vulnerability item")
|
|
classify("vulnerability", item)
|
|
for warning_kind, items in warnings.items():
|
|
if not isinstance(items, list):
|
|
fail(f"cargo-audit warning list is malformed: {warning_kind}")
|
|
for item in items:
|
|
if not isinstance(item, dict):
|
|
fail(f"malformed cargo-audit warning item: {warning_kind}")
|
|
classify(f"warning:{warning_kind}", item)
|
|
|
|
report_lines = [
|
|
"nigig-site production RustSec containment report",
|
|
f"reachable normal/build packages (all target predicates): {len(reachable)}",
|
|
f"workspace findings reported by cargo-audit: {len(findings) + len(excluded)}",
|
|
f"reachable findings: {len(findings)}",
|
|
f"excluded dev/unrelated-workspace findings: {len(excluded)}",
|
|
"",
|
|
]
|
|
for prefix, records in (("REACHABLE", findings), ("EXCLUDED", excluded)):
|
|
for kind, advisory_id, name, version in sorted(records):
|
|
report_lines.append(
|
|
f"{prefix} {kind} {advisory_id} {name} {version}"
|
|
)
|
|
report = "\n".join(report_lines) + "\n"
|
|
Path(sys.argv[3]).write_text(report)
|
|
print(report, end="")
|
|
if findings:
|
|
fail("RustSec findings are reachable from nigig-site production dependencies")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|