#!/usr/bin/env python3 """Repository-local supply-chain checks with no third-party runtime dependency.""" from __future__ import annotations import argparse import base64 import hashlib import json import re import subprocess import sys from pathlib import Path from typing import Iterable from urllib.parse import quote SECRET_PATTERNS = { "private-key": re.compile(rb"-----BEGIN (?:RSA |EC |OPENSSH |DSA )?PRIVATE KEY-----"), "jwt": re.compile(rb"\beyJ[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\b"), "aws-access-key": re.compile(rb"\b(?:AKIA|ASIA)[A-Z0-9]{16}\b"), "github-token": re.compile(rb"\b(?:gh[pousr]_[A-Za-z0-9]{36,255}|github_pat_[A-Za-z0-9_]{82,255})\b"), "google-api-key": re.compile(rb"\bAIza[A-Za-z0-9_-]{35}\b"), "slack-token": re.compile(rb"\bxox[baprs]-[A-Za-z0-9-]{20,}\b"), "stripe-live-key": re.compile(rb"\bsk_live_[A-Za-z0-9]{16,}\b"), } MAX_SCANNED_FILE_BYTES = 2 * 1024 * 1024 def tracked_files(root: Path) -> list[str]: result = subprocess.run( ["git", "ls-files", "-z"], cwd=root, check=True, capture_output=True, ) return [item.decode("utf-8", errors="surrogateescape") for item in result.stdout.split(b"\0") if item] def scan_files(root: Path, paths: Iterable[str]) -> list[tuple[str, str, int]]: findings: list[tuple[str, str, int]] = [] for relative in paths: path = root / relative if not path.is_file() or path.stat().st_size > MAX_SCANNED_FILE_BYTES: continue data = path.read_bytes() if b"\0" in data: continue for line_number, line in enumerate(data.splitlines(), start=1): for secret_type, pattern in SECRET_PATTERNS.items(): if pattern.search(line): findings.append((secret_type, relative.replace("\\", "/"), line_number)) return findings def _purl(ecosystem: str, name: str, version: str) -> str: if ecosystem == "npm" and name.startswith("@"): namespace, package = name[1:].split("/", 1) return f"pkg:npm/{quote(namespace, safe='')}/{quote(package, safe='')}@{quote(version, safe='')}" return f"pkg:{ecosystem}/{quote(name, safe='')}@{quote(version, safe='')}" def _component(ecosystem: str, name: str, version: str, manifests: set[str], scope: str = "required", hashes: list[dict[str, str]] | None = None) -> dict[str, object]: component: dict[str, object] = { "type": "library", "bom-ref": _purl(ecosystem, name, version), "name": name, "version": version, "scope": scope, "purl": _purl(ecosystem, name, version), "properties": [{"name": "jobjakt:source-manifest", "value": manifest} for manifest in sorted(manifests)], } if hashes: component["hashes"] = hashes return component def _npm_components(root: Path) -> list[dict[str, object]]: relative = "job-tracker-ui/package-lock.json" lock = json.loads((root / relative).read_text(encoding="utf-8")) components: dict[tuple[str, str], dict[str, object]] = {} for package_path, package in lock.get("packages", {}).items(): if not package_path or not package.get("version"): continue name = package.get("name") or package_path.rsplit("node_modules/", 1)[-1] version = str(package["version"]) integrity = str(package.get("integrity", "")) hashes: list[dict[str, str]] = [] if integrity.startswith("sha512-"): try: hashes.append({"alg": "SHA-512", "content": base64.b64decode(integrity[7:]).hex()}) except ValueError: pass components[(name, version)] = _component( "npm", name, version, {relative}, "optional" if package.get("dev") else "required", hashes, ) return list(components.values()) def _nuget_components(root: Path) -> list[dict[str, object]]: manifests = ["JobTrackerApi/packages.lock.json", "JobTrackerApi.Tests/packages.lock.json"] packages: dict[tuple[str, str], set[str]] = {} hashes: dict[tuple[str, str], str] = {} for relative in manifests: lock = json.loads((root / relative).read_text(encoding="utf-8")) for target in lock.get("dependencies", {}).values(): for name, package in target.items(): version = str(package.get("resolved", "")) if not version: continue key = (name, version) packages.setdefault(key, set()).add(relative) if package.get("contentHash"): hashes[key] = str(package["contentHash"]) result = [] for (name, version), sources in packages.items(): package_hashes = [] if (content_hash := hashes.get((name, version))): try: package_hashes.append({"alg": "SHA-512", "content": base64.b64decode(content_hash).hex()}) except ValueError: pass result.append(_component("nuget", name, version, sources, hashes=package_hashes)) return result def _python_components(root: Path) -> list[dict[str, object]]: relative = "tools/summarizer/requirements-linux.lock" pattern = re.compile(r"^([A-Za-z0-9_.-]+)==([^\s\\]+)") packages: dict[tuple[str, str], set[str]] = {} for line in (root / relative).read_text(encoding="utf-8").splitlines(): if match := pattern.match(line): packages.setdefault((match.group(1), match.group(2)), set()).add(relative) return [_component("pypi", name, version, sources) for (name, version), sources in packages.items()] def generate_sbom(root: Path) -> dict[str, object]: lock_paths = [ "job-tracker-ui/package-lock.json", "JobTrackerApi/packages.lock.json", "JobTrackerApi.Tests/packages.lock.json", "tools/summarizer/requirements-linux.lock", ] digest = hashlib.sha256() for relative in lock_paths: digest.update(relative.encode("utf-8")) digest.update((root / relative).read_bytes()) components = _npm_components(root) + _nuget_components(root) + _python_components(root) components.sort(key=lambda item: (str(item["purl"]).lower(), str(item["version"]))) return { "bomFormat": "CycloneDX", "specVersion": "1.5", "serialNumber": f"urn:uuid:{digest.hexdigest()[:8]}-{digest.hexdigest()[8:12]}-5{digest.hexdigest()[13:16]}-a{digest.hexdigest()[17:20]}-{digest.hexdigest()[20:32]}", "version": 1, "metadata": { "component": {"type": "application", "bom-ref": "pkg:generic/jobjakt", "name": "Jobjakt"}, "tools": {"components": [{"type": "application", "name": "scripts/supply-chain.py"}]}, "properties": [{"name": "jobjakt:lock-digest-sha256", "value": digest.hexdigest()}], }, "components": components, } def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("command", choices=("scan-secrets", "generate-sbom")) parser.add_argument("--root", type=Path, default=Path(__file__).resolve().parents[1]) parser.add_argument("--output", type=Path) args = parser.parse_args() root = args.root.resolve() if args.command == "scan-secrets": findings = scan_files(root, tracked_files(root)) for secret_type, relative, line_number in findings: print(f"{secret_type}: {relative}:{line_number}") if findings: print(f"Tracked-secret scan failed with {len(findings)} high-confidence finding(s); values were suppressed.", file=sys.stderr) return 1 print("Tracked-secret scan passed; no high-confidence credential patterns found.") return 0 if not args.output: parser.error("generate-sbom requires --output") output = args.output if args.output.is_absolute() else root / args.output output.parent.mkdir(parents=True, exist_ok=True) output.write_text(json.dumps(generate_sbom(root), indent=2, ensure_ascii=False) + "\n", encoding="utf-8") print(f"CycloneDX SBOM written to {output}.") return 0 if __name__ == "__main__": raise SystemExit(main())