cesnimda
200 lines
8.1 KiB
Python
200 lines
8.1 KiB
Python
#!/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())
|