diff --git a/.github/workflows/aibom.yml b/.github/workflows/aibom.yml index 427d36c..9b3737d 100644 --- a/.github/workflows/aibom.yml +++ b/.github/workflows/aibom.yml @@ -1,17 +1,55 @@ -name: AIBOM Drift Gate +name: AIBOM Drift Gate + GitHub Scanner + on: pull_request: + schedule: + - cron: "0 4 * * 1" + workflow_dispatch: + jobs: - aibom: + drift-gate: + if: github.event_name == 'pull_request' runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - uses: actions/setup-python@v5 with: python-version: "3.11" + cache: "pip" - run: pip install -e . - - run: aibom generate . -o new_aibom.json + - run: aibom generate . -o new_aibom.json --profile ai-bom-like - run: | if [ -f .aibom/baseline.json ]; then aibom diff .aibom/baseline.json new_aibom.json --fail-on new-model,new-tool,new-external-provider fi + - uses: actions/upload-artifact@v4 + with: + name: aibom-pr-scan + path: | + new_aibom.json + new_aibom_ai_profile.json + + scheduled-github-scan: + if: github.event_name != 'pull_request' + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.11" + cache: "pip" + - run: pip install -e . + - name: Run multi-repo GitHub scan + env: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + run: | + aibom scan-github \ + --repos-file examples/github_repo_samples/repos.txt \ + --output-dir github-scan-out \ + --profile ai-bom-like \ + --max-repos 10 \ + --timeout-sec 240 + - uses: actions/upload-artifact@v4 + with: + name: aibom-github-scan + path: github-scan-out diff --git a/README.md b/README.md index 5d8883f..d289655 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # AIBOM (Living AI Bill of Materials) -Standards-first, CI-native AIBOM generator for Python/LangChain projects with audit evidence bundling, drift detection, and heuristic risk overlay. +Standards-first, CI-native AIBOM generator for Python/LangChain/JS-TS/Java/Go/.NET projects with SPDX/CycloneDX/SARIF/VEX exports, drift gates, and attestation workflows. ## Install @@ -9,111 +9,84 @@ pip install -e . pip install -r requirements.txt ``` -## CLI - -```bash -aibom --version -aibom --help -``` - -Commands: -- `aibom generate` -- `aibom validate` -- `aibom export` -- `aibom diff` -- `aibom bundle` -- `aibom risk` - -## Usage - -### Generate AIBOM +## CLI quickstart ```bash aibom generate . -o AI_BOM.json -# generation fails fast if JSON Schema validation fails -``` - -Optional prompt-content collection (default is metadata-only): - -```bash -aibom generate . -o AI_BOM.json --include-prompts -``` - -### Audit mode (end-to-end) - -```bash -aibom generate . -o AI_BOM.json --audit-mode --bundle-out evidence.zip +aibom summarize --input AI_BOM.json ``` -### Validation - -```bash -aibom generate --audit-mode --out AI_BOM.json -aibom validate AI_BOM.json -``` +## ai-bom-like compatibility profile -`aibom generate` fails closed before writing output if schema validation fails. Validation errors include JSON-pointer-like paths to the failing field (for example `/metadata/generated_at`). - -### Standards Output +AIBOM keeps the canonical `AI_BOM.json` schema stable by default. For ai-bom-style ergonomics, use `--profile ai-bom-like`. ```bash -aibom export --input AI_BOM.json --format spdx-json -o SPDX.json -aibom export --input AI_BOM.json --format cyclonedx-json -o CYCLONEDX.json -aibom export --input AI_BOM.json --format sarif-json -o FINDINGS.sarif.json -aibom export --input AI_BOM.json --format vex-json -o ADVISORIES.vex.json +aibom generate . -o AI_BOM.json --profile ai-bom-like +# writes AI_BOM.json + AI_BOM_ai_profile.json and prints a concise terminal summary ``` -Internal → SPDX/CycloneDX mapping (extended): -- `models[].type` -> `packages[].name` -- `models[].model` -> `packages[].versionInfo` -- `tools[].name` -> `packages[].name` -- `datasets[].type` -> `packages[].name` -- `risk_findings[]` -> SPDX package advisory refs / CycloneDX `vulnerabilities[]` -- detector metadata (`scan_findings[].confidence`, `severity`, `source_type`) -> external refs/properties -- model `provenance` + `lineage` -> standards-compatible external refs/properties - -### Drift detection - -```bash -aibom diff .aibom/baseline.json AI_BOM.json --fail-on new-model,new-tool,new-external-provider -``` +This profile adds a companion presentation JSON with: +- executive summary counts +- grouped AI assets +- risk highlights +- provenance/compliance rollup +- detector coverage stats -### Evidence bundle +## GitHub scanner quickstart ```bash -aibom bundle --input AI_BOM.json --out evidence.zip --baseline .aibom/baseline.json +aibom scan-github \ + --repo openai/openai-quickstart-python \ + --output-dir out \ + --profile ai-bom-like ``` -Bundle contains: -- `AIBOM.json` -- `SPDX.json` -- `DIFF.json` (if baseline exists) -- `MANIFEST.json` (SHA256s) -- `ENVIRONMENT.json` -- `COMPLIANCE_MAPPING.md` - -### Risk summary +Multi-repo scan: ```bash -aibom risk --input AI_BOM.json +aibom scan-github \ + --repos-file repos.txt \ + --output-dir out \ + --max-repos 20 \ + --timeout-sec 240 \ + --fail-on new-model,new-tool,new-external-provider \ + --max-high-risk 0 \ + --max-unsupported 0 ``` -## For Auditors - -See [`docs/FOR_AUDITORS.md`](docs/FOR_AUDITORS.md) for verification procedure, manifest validation, and reproducibility notes. +Output layout: +- `out//AI_BOM.json` +- `out//AI_BOM_ai_profile.json` (when `--profile ai-bom-like`) +- `out/SUMMARY.md` +- `out/summary.json` -## SOC Deployment Guide +## Core commands -See [`docs/SOC_DEPLOYMENT_GUIDE.md`](docs/SOC_DEPLOYMENT_GUIDE.md) for CI/CD integration and drift gate rollout. +- `aibom generate` +- `aibom scan-github` +- `aibom summarize` +- `aibom validate` +- `aibom export` +- `aibom diff` +- `aibom bundle` +- `aibom attest` +- `aibom risk` -## Compliance Mapping +## Compatibility and migration notes -See [`docs/COMPLIANCE_MAPPING.md`](docs/COMPLIANCE_MAPPING.md). This is a starter mapping only, not legal advice. +- `generate`, `validate`, `export`, `diff`, `bundle`, `attest`, and `risk` remain functional and backward compatible. +- New `scan-github` and `summarize` commands are additive. +- ai-bom-like output is opt-in (`--profile ai-bom-like`) to avoid schema-breaking changes to canonical AIBOM consumers. +- `scan-github` returns nonzero when any repo scan errors or configured gates fail, while still producing aggregate summary files for partial failures. -## Example outputs for known repositories +## Documentation -See [`examples/github_repo_samples/README.md`](examples/github_repo_samples/README.md) for sample output files and a script that scans smaller well-known GitHub AI repositories, with direct links to each scanned repo. +- [GitHub scanner guide](docs/GITHUB_SCANNER_GUIDE.md) +- [For auditors](docs/FOR_AUDITORS.md) +- [SOC deployment guide](docs/SOC_DEPLOYMENT_GUIDE.md) +- [Compliance mapping](docs/COMPLIANCE_MAPPING.md) -## LangChain demo +## Examples -See [`examples/langchain_demo/README.md`](examples/langchain_demo/README.md). +- [`examples/github_repo_samples/`](examples/github_repo_samples/) +- [`examples/langchain_demo/`](examples/langchain_demo/) diff --git a/aibom/bundle.py b/aibom/bundle.py index 81a0f6f..49ec75b 100644 --- a/aibom/bundle.py +++ b/aibom/bundle.py @@ -18,7 +18,6 @@ sha256_bytes, stable_json, validate_safe_path, - PathSecurityError, ) @@ -91,9 +90,7 @@ def _parse_openssl_time(value: str) -> datetime: def _certificate_sans(cert_path: Path) -> list[str]: # Validate certificate path before passing to openssl safe_cert_path = validate_safe_path(cert_path, must_exist=True, must_be_file=True) - ext = _openssl( - ["x509", "-in", str(safe_cert_path), "-noout", "-ext", "subjectAltName"] - ).stdout + ext = _openssl(["x509", "-in", str(safe_cert_path), "-noout", "-ext", "subjectAltName"]).stdout return re.findall(r"DNS:([^,\n]+)", ext) @@ -106,7 +103,7 @@ def _verify_chain( ) -> None: # Validate signing certificate path safe_signing_cert = validate_safe_path(signing_cert, must_exist=True, must_be_file=True) - + if not ca_bundle and not trusted_roots: return @@ -215,7 +212,7 @@ def sign_bundle( safe_bundle_path = validate_safe_path(bundle_path, must_exist=True, must_be_file=True) safe_signing_key = validate_safe_path(signing_key, must_exist=True, must_be_file=True) safe_signing_cert = validate_safe_path(signing_cert, must_exist=True, must_be_file=True) - + signature_path = signature_path or bundle_path.with_suffix(bundle_path.suffix + ".sig") provenance_path = provenance_path or bundle_path.with_name("provenance.json") @@ -247,7 +244,7 @@ def sign_bundle( "sha256": sha256_bytes(signature_path.read_bytes()), "algorithm": "RSA-SHA256", }, - "certificate": _cert_metadata(signing_cert), + "certificate": _cert_metadata(safe_signing_cert), "policy_evaluation": { "status": "not_evaluated", "checks": {}, @@ -273,13 +270,13 @@ def verify_bundle_signature( safe_bundle_path = validate_safe_path(bundle_path, must_exist=True, must_be_file=True) safe_signature_path = validate_safe_path(signature_path, must_exist=True, must_be_file=True) safe_signing_cert = validate_safe_path(signing_cert, must_exist=True, must_be_file=True) - + # Validate optional paths safe_ca_bundle: Path | None = None safe_crl_file: Path | None = None safe_trusted_roots: list[Path] | None = None safe_provenance_path: Path | None = None - + if ca_bundle is not None: safe_ca_bundle = validate_safe_path(ca_bundle, must_exist=True, must_be_file=True) if crl_file is not None: @@ -289,16 +286,22 @@ def verify_bundle_signature( validate_safe_path(p, must_exist=True, must_be_file=True) for p in trusted_roots ] if provenance_path is not None and provenance_path.exists(): - safe_provenance_path = validate_safe_path(provenance_path, must_exist=True, must_be_file=True) - + safe_provenance_path = validate_safe_path( + provenance_path, must_exist=True, must_be_file=True + ) + _enforce_validity_window(safe_signing_cert) - _verify_chain(safe_signing_cert, safe_ca_bundle, safe_trusted_roots, safe_crl_file, revocation_policy) + _verify_chain( + safe_signing_cert, safe_ca_bundle, safe_trusted_roots, safe_crl_file, revocation_policy + ) policy_checks: dict[str, Any] = { "certificate_validity": {"status": "passed"}, "certificate_chain": { "status": "passed" if (safe_ca_bundle or safe_trusted_roots) else "skipped", - "reason": "no trust anchors provided" if not (safe_ca_bundle or safe_trusted_roots) else None, + "reason": ( + "no trust anchors provided" if not (safe_ca_bundle or safe_trusted_roots) else None + ), }, } diff --git a/aibom/cli.py b/aibom/cli.py index 938e821..c436bae 100644 --- a/aibom/cli.py +++ b/aibom/cli.py @@ -10,8 +10,15 @@ from aibom.bundle import create_bundle, sign_bundle, verify_bundle_signature from aibom.diffing import diff_aibom, gate_failures, trend_diff_aibom from aibom.exporters import export_cyclonedx, export_sarif, export_spdx, export_vex +from aibom.github_scan import _load_repos, scan_github_repos +from aibom.presentation import ( + build_ai_bom_like_profile, + profile_json_dumps, + render_text_summary, +) from aibom.risk.heuristics import generate_risk_findings from aibom.storage import load_json, list_run_history, persist_periodic_snapshot, persist_run +from aibom.utils import stable_json from aibom.validation import AIBOMValidationException, validate_aibom @@ -19,7 +26,7 @@ def _write_json(path: Path, data: dict) -> None: - path.write_text(json.dumps(data, indent=2, sort_keys=True), encoding="utf-8") + path.write_text(stable_json(data), encoding="utf-8") def _parse_allowlist(args: argparse.Namespace) -> dict[str, list[str]] | None: @@ -33,9 +40,7 @@ def _parse_allowlist(args: argparse.Namespace) -> dict[str, list[str]] | None: return policy or None -def cmd_generate(args: argparse.Namespace) -> int: - target = Path(args.target).resolve() - out = Path(args.output).resolve() +def _enforce_prompt_ack(args: argparse.Namespace) -> int | None: if args.include_prompts and not args.acknowledge_prompt_exposure_risk: print( "ERROR: --include-prompts is high risk and requires --acknowledge-prompt-exposure-risk.", @@ -48,7 +53,16 @@ def cmd_generate(args: argparse.Namespace) -> int: "WARNING: Including prompts may expose sensitive business logic or secrets in templates.", file=sys.stderr, ) + return None + + +def cmd_generate(args: argparse.Namespace) -> int: + ack_error = _enforce_prompt_ack(args) + if ack_error is not None: + return ack_error + target = Path(args.target).resolve() + out = Path(args.output).resolve() risk_policy_path = Path(args.risk_policy).resolve() if args.risk_policy else None aibom = generate_aibom( @@ -75,6 +89,13 @@ def cmd_generate(args: argparse.Namespace) -> int: return 2 _write_json(out, aibom) + + if args.profile == "ai-bom-like": + profile_out = out.with_name(f"{out.stem}_ai_profile.json") + profile_doc = build_ai_bom_like_profile(aibom) + profile_out.write_text(profile_json_dumps(profile_doc), encoding="utf-8") + print(render_text_summary(aibom)) + persist_run(target, aibom) if args.audit_mode: @@ -232,14 +253,92 @@ def cmd_risk(args: argparse.Namespace) -> int: return 0 +def cmd_summarize(args: argparse.Namespace) -> int: + src = load_json(Path(args.input)) + summary = render_text_summary(src) + if args.json: + print( + json.dumps( + { + "summary": summary, + "counts": { + "models": len(src.get("models", [])), + "tools": len(src.get("tools", [])), + "datasets": len(src.get("datasets", [])), + "frameworks": len(src.get("frameworks", [])), + "prompts": len(src.get("prompts", [])), + }, + }, + indent=2, + sort_keys=True, + ) + ) + else: + print(summary) + return 0 + + +def cmd_scan_github(args: argparse.Namespace) -> int: + ack_error = _enforce_prompt_ack(args) + if ack_error is not None: + return ack_error + + repos = _load_repos(args.repo or [], args.repos_file) + if not repos: + print("ERROR: provide at least one --repo or --repos-file", file=sys.stderr) + return 2 + + risk_policy_path = Path(args.risk_policy).resolve() if args.risk_policy else None + records, exit_code = scan_github_repos( + repos=repos, + output_dir=Path(args.output_dir).resolve(), + branch=args.branch, + depth=args.depth, + token_env=args.token_env, + max_repos=args.max_repos, + timeout_sec=args.timeout_sec, + include_prompts=args.include_prompts, + include_runtime_manifests=args.include_runtime_manifests, + redaction_policy=args.redaction_policy, + risk_policy_path=risk_policy_path, + profile=args.profile, + fail_on=args.fail_on, + max_high_risk=args.max_high_risk, + max_unsupported=args.max_unsupported, + baseline_file=Path(args.baseline) if args.baseline else None, + ) + + payload = {"records": [record.__dict__ for record in records], "exit_code": exit_code} + if args.json: + print(json.dumps(payload, indent=2, sort_keys=True)) + else: + for record in records: + print( + f"[{record.status}] {record.repo} models={record.counts.get('models', 0)} " + f"tools={record.counts.get('tools', 0)} high+risk={record.counts.get('high_or_critical_risks', 0)} " + f"unsupported={record.counts.get('unsupported_artifacts', 0)} verdict={record.gate_verdict}" + ) + if record.error: + print(f" error: {record.error}") + return exit_code + + def build_parser() -> argparse.ArgumentParser: - parser = argparse.ArgumentParser(prog="aibom", description="Living AIBOM generator") + parser = argparse.ArgumentParser( + prog="aibom", + description="Standards-first AIBOM generator with drift gates, attestations, and GitHub scanning.", + ) parser.add_argument("--version", action="version", version=f"aibom {__version__}") sub = parser.add_subparsers(dest="command", required=True) - gen = sub.add_parser("generate") + gen = sub.add_parser( + "generate", + help="Scan a local repository and produce canonical AIBOM JSON.", + description="Generate canonical AIBOM JSON from a source tree. Use --profile ai-bom-like for a companion presentation JSON.", + ) gen.add_argument("target", nargs="?", default=".") gen.add_argument("-o", "--output", default="AI_BOM.json") + gen.add_argument("--profile", choices=["canonical", "ai-bom-like"], default="canonical") gen.add_argument("--include-prompts", action="store_true") gen.add_argument( "--acknowledge-prompt-exposure-risk", @@ -263,6 +362,47 @@ def build_parser() -> argparse.ArgumentParser: ) gen.set_defaults(func=cmd_generate) + gh = sub.add_parser( + "scan-github", + help="Scan one or many GitHub repositories and aggregate outputs.", + description="Clone repositories into a temp workspace, run AIBOM generation, and emit per-repo outputs plus summary files.", + epilog="Example: aibom scan-github --repo openai/openai-quickstart-python --output-dir out --profile ai-bom-like", + ) + gh.add_argument( + "--repo", + action="append", + help="GitHub repo in owner/name format. Repeat for multi-repo scans.", + ) + gh.add_argument("--repos-file", help="File containing owner/name repos, one per line.") + gh.add_argument("--output-dir", default="github_scan_out") + gh.add_argument("--branch") + gh.add_argument("--depth", type=int, default=1) + gh.add_argument("--token-env", default="GITHUB_TOKEN") + gh.add_argument("--max-repos", type=int) + gh.add_argument("--timeout-sec", type=int, default=180) + gh.add_argument("--profile", choices=["canonical", "ai-bom-like"], default="canonical") + gh.add_argument("--include-prompts", action="store_true") + gh.add_argument("--acknowledge-prompt-exposure-risk", action="store_true") + gh.add_argument("--include-runtime-manifests", action="store_true") + gh.add_argument("--redaction-policy", choices=["strict", "default", "off"], default="strict") + gh.add_argument("--risk-policy") + gh.add_argument("--baseline", help="Optional baseline AIBOM for --fail-on drift checks.") + gh.add_argument( + "--fail-on", help="Comma-separated drift gates: new-model,new-tool,new-external-provider" + ) + gh.add_argument( + "--max-high-risk", type=int, help="Fail a repo when high/critical risks exceed this number." + ) + gh.add_argument( + "--max-unsupported", + type=int, + help="Fail a repo when unsupported artifacts exceed this number.", + ) + gh.add_argument( + "--json", action="store_true", help="Emit machine-readable scan status JSON to stdout." + ) + gh.set_defaults(func=cmd_scan_github) + pscan = sub.add_parser("periodic-scan") pscan.add_argument("target", nargs="?", default=".") pscan.add_argument("-o", "--output", default="periodic_scan.json") @@ -277,6 +417,14 @@ def build_parser() -> argparse.ArgumentParser: ) pscan.set_defaults(func=cmd_periodic_scan) + s = sub.add_parser( + "summarize", + help="Print ai-bom-like summary text from an existing AIBOM JSON.", + ) + s.add_argument("--input", required=True) + s.add_argument("--json", action="store_true", help="Emit summary payload as JSON.") + s.set_defaults(func=cmd_summarize) + v = sub.add_parser("validate") v.add_argument("input") v.set_defaults(func=cmd_validate) diff --git a/aibom/github_scan.py b/aibom/github_scan.py new file mode 100644 index 0000000..fb3972a --- /dev/null +++ b/aibom/github_scan.py @@ -0,0 +1,225 @@ +from __future__ import annotations + +import os +import shutil +import subprocess +import tempfile +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Any + +from aibom.analyzer import generate_aibom +from aibom.diffing import diff_aibom, gate_failures +from aibom.presentation import ( + build_ai_bom_like_profile, + profile_json_dumps, + render_markdown_summary, +) +from aibom.storage import load_json +from aibom.utils import stable_json, validate_safe_path +from aibom.validation import validate_aibom + + +@dataclass +class RepoScanRecord: + repo: str + status: str + output_json: str + output_profile_json: str | None + counts: dict[str, int] + gate_verdict: str + gate_failures: list[str] + error: str | None = None + + +def _repo_slug(repo: str) -> str: + return repo.replace("/", "__") + + +def _clone_repo( + repo: str, + dest: Path, + branch: str | None, + depth: int, + token: str | None, + timeout_sec: int, +) -> None: + url = f"https://github.com/{repo}.git" + if token: + url = f"https://x-access-token:{token}@github.com/{repo}.git" + + cmd = ["git", "clone", "--depth", str(depth)] + if branch: + cmd.extend(["--branch", branch]) + cmd.extend([url, str(dest)]) + + subprocess.run(cmd, check=True, capture_output=True, text=True, timeout=timeout_sec) + + +def _count_summary(aibom: dict[str, Any]) -> dict[str, int]: + return { + "models": len(aibom.get("models", [])), + "tools": len(aibom.get("tools", [])), + "datasets": len(aibom.get("datasets", [])), + "frameworks": len(aibom.get("frameworks", [])), + "prompts": len(aibom.get("prompts", [])), + "unsupported_artifacts": len(aibom.get("unsupported_artifacts", [])), + "high_or_critical_risks": sum( + 1 + for item in aibom.get("risk_findings", []) + if str(item.get("severity", "")).lower() in {"high", "critical"} + ), + } + + +def _load_repos(args_repos: list[str], repos_file: str | None) -> list[str]: + repos = list(args_repos) + if repos_file: + file_path = validate_safe_path(Path(repos_file), must_exist=True, must_be_file=True) + for line in file_path.read_text(encoding="utf-8").splitlines(): + candidate = line.strip() + if not candidate or candidate.startswith("#"): + continue + repos.append(candidate) + seen: set[str] = set() + unique: list[str] = [] + for repo in repos: + if repo not in seen: + seen.add(repo) + unique.append(repo) + return unique + + +def scan_github_repos( + repos: list[str], + output_dir: Path, + branch: str | None = None, + depth: int = 1, + token_env: str = "GITHUB_TOKEN", + max_repos: int | None = None, + timeout_sec: int = 180, + include_prompts: bool = False, + include_runtime_manifests: bool = False, + redaction_policy: str = "strict", + risk_policy_path: Path | None = None, + profile: str = "canonical", + fail_on: str | None = None, + max_high_risk: int | None = None, + max_unsupported: int | None = None, + baseline_file: Path | None = None, +) -> tuple[list[RepoScanRecord], int]: + output_dir = validate_safe_path(output_dir, must_exist=False) + output_dir.mkdir(parents=True, exist_ok=True) + + token = os.getenv(token_env) + selected_repos = repos[:max_repos] if max_repos is not None else repos + records: list[RepoScanRecord] = [] + global_failures = 0 + + baseline_doc = load_json(baseline_file) if baseline_file and baseline_file.exists() else None + fail_on_set = set(filter(None, (fail_on or "").split(","))) + + for repo in selected_repos: + repo_dir = output_dir / _repo_slug(repo) + repo_dir.mkdir(parents=True, exist_ok=True) + canonical_output = repo_dir / "AI_BOM.json" + profile_output = repo_dir / "AI_BOM_ai_profile.json" + + try: + with tempfile.TemporaryDirectory(prefix="aibom-gh-") as temp_dir: + clone_dest = Path(temp_dir) / "repo" + _clone_repo( + repo=repo, + dest=clone_dest, + branch=branch, + depth=depth, + token=token, + timeout_sec=timeout_sec, + ) + aibom = generate_aibom( + clone_dest, + include_prompts=include_prompts, + include_runtime_manifests=include_runtime_manifests, + redaction_policy=redaction_policy, + risk_policy_path=risk_policy_path, + ) + validate_aibom(aibom) + canonical_output.write_text(stable_json(aibom), encoding="utf-8") + + profile_path_str: str | None = None + if profile == "ai-bom-like": + ai_profile = build_ai_bom_like_profile(aibom) + profile_output.write_text(profile_json_dumps(ai_profile), encoding="utf-8") + profile_path_str = str(profile_output.relative_to(output_dir)) + + failures: list[str] = [] + if baseline_doc is not None: + failures.extend(gate_failures(diff_aibom(baseline_doc, aibom), fail_on_set)) + if ( + max_high_risk is not None + and _count_summary(aibom)["high_or_critical_risks"] > max_high_risk + ): + failures.append("max-high-risk") + if ( + max_unsupported is not None + and _count_summary(aibom)["unsupported_artifacts"] > max_unsupported + ): + failures.append("max-unsupported") + + gate_verdict = "pass" if not failures else "fail" + if failures: + global_failures += 1 + + records.append( + RepoScanRecord( + repo=repo, + status="ok", + output_json=str(canonical_output.relative_to(output_dir)), + output_profile_json=profile_path_str, + counts=_count_summary(aibom), + gate_verdict=gate_verdict, + gate_failures=sorted(set(failures)), + ) + ) + except Exception as exc: + global_failures += 1 + if repo_dir.exists() and not any(repo_dir.iterdir()): + shutil.rmtree(repo_dir) + records.append( + RepoScanRecord( + repo=repo, + status="error", + output_json="", + output_profile_json=None, + counts={ + "models": 0, + "tools": 0, + "datasets": 0, + "frameworks": 0, + "prompts": 0, + "unsupported_artifacts": 0, + "high_or_critical_risks": 0, + }, + gate_verdict="fail", + gate_failures=["scan-error"], + error=str(exc), + ) + ) + + summary = { + "profile": profile, + "total_repositories": len(selected_repos), + "failed_repositories": global_failures, + "records": [asdict(record) for record in records], + } + (output_dir / "summary.json").write_text(stable_json(summary), encoding="utf-8") + (output_dir / "SUMMARY.md").write_text( + render_markdown_summary(summary["records"]), + encoding="utf-8", + ) + + exit_code = 2 if global_failures else 0 + return records, exit_code + + +__all__ = ["scan_github_repos", "_load_repos"] diff --git a/aibom/presentation.py b/aibom/presentation.py new file mode 100644 index 0000000..5c19c94 --- /dev/null +++ b/aibom/presentation.py @@ -0,0 +1,212 @@ +from __future__ import annotations + +import json +from typing import Any + + +DEFAULT_TOP_RISKS = 3 + + +def _severity_rank(severity: str) -> int: + order = {"critical": 0, "high": 1, "medium": 2, "low": 3} + return order.get(str(severity).lower(), 4) + + +def build_ai_bom_like_profile(aibom: dict[str, Any]) -> dict[str, Any]: + """Build a companion presentation profile without changing canonical schema output.""" + risk_findings = list(aibom.get("risk_findings", [])) + sorted_risks = sorted( + risk_findings, + key=lambda item: ( + _severity_rank(str(item.get("severity", ""))), + -float(item.get("score", 0.0) or 0.0), + str(item.get("title", item.get("id", ""))), + ), + ) + + coverage = aibom.get("coverage_summary", {}) + detectors = list(coverage.get("detectors", [])) + detectors_sorted = sorted(detectors, key=lambda item: item.get("source_type", "")) + + return { + "profile": "ai-bom-like", + "schema_version": aibom.get("schema_version", "1.0"), + "metadata": { + "generated_at": aibom.get("metadata", {}).get("generated_at", ""), + "git_sha": aibom.get("metadata", {}).get("git_sha", "unknown"), + "artifact_sha256": aibom.get("metadata", {}).get("artifact_sha256", ""), + }, + "executive_summary": { + "models": len(aibom.get("models", [])), + "tools": len(aibom.get("tools", [])), + "datasets": len(aibom.get("datasets", [])), + "frameworks": len(aibom.get("frameworks", [])), + "prompts": len(aibom.get("prompts", [])), + "risk_findings": len(risk_findings), + "high_or_critical_risks": sum( + 1 + for item in risk_findings + if str(item.get("severity", "")).lower() in {"high", "critical"} + ), + "unsupported_artifacts": len(aibom.get("unsupported_artifacts", [])), + }, + "ai_assets": { + "models": sorted( + [ + { + "type": item.get("type", "unknown"), + "model": item.get("model", "unknown"), + "source_file": item.get("source_file", "unknown"), + "provider": item.get("provenance", {}).get("provider_endpoint", "unknown"), + } + for item in aibom.get("models", []) + ], + key=lambda item: (item["type"], item["model"], item["source_file"]), + ), + "tools": sorted( + [ + { + "name": item.get("name", "unknown"), + "source_file": item.get("source_file", "unknown"), + } + for item in aibom.get("tools", []) + ], + key=lambda item: (item["name"], item["source_file"]), + ), + "datasets": sorted( + [ + { + "type": item.get("type", "unknown"), + "source_file": item.get("source_file", "unknown"), + } + for item in aibom.get("datasets", []) + ], + key=lambda item: (item["type"], item["source_file"]), + ), + "frameworks": sorted(item.get("name", "") for item in aibom.get("frameworks", [])), + }, + "risk_highlights": [ + { + "id": item.get("id", ""), + "title": item.get("title", ""), + "severity": item.get("severity", ""), + "score": item.get("score", 0.0), + "rule_id": item.get("rule_id", ""), + } + for item in sorted_risks[:DEFAULT_TOP_RISKS] + ], + "provenance_and_compliance": { + "runtime_context": aibom.get("runtime_context", {}), + "risk_policy": aibom.get("risk_policy", {}), + "source_types": aibom.get("source_types", []), + }, + "detector_coverage": { + "unsupported_total": int(coverage.get("unsupported_total", 0) or 0), + "detectors": [ + { + "source_type": item.get("source_type", "unknown"), + "files_scanned": int(item.get("files_scanned", 0) or 0), + "findings": int(item.get("findings", 0) or 0), + } + for item in detectors_sorted + ], + }, + } + + +def render_text_summary( + aibom: dict[str, Any], + drift_failures: list[str] | None = None, + max_risks: int = DEFAULT_TOP_RISKS, +) -> str: + risk_findings = list(aibom.get("risk_findings", [])) + high_or_critical = [ + item + for item in risk_findings + if str(item.get("severity", "")).lower() in {"high", "critical"} + ] + top_risks = sorted( + risk_findings, + key=lambda item: ( + _severity_rank(str(item.get("severity", ""))), + -float(item.get("score", 0.0) or 0.0), + ), + )[:max_risks] + + coverage = aibom.get("coverage_summary", {}) + detectors = sorted( + list(coverage.get("detectors", [])), + key=lambda item: item.get("source_type", ""), + ) + + lines = [ + "AIBOM scan summary", + "-" * 60, + "Counts", + " category count", + f" models {len(aibom.get('models', []))}", + f" tools {len(aibom.get('tools', []))}", + f" datasets {len(aibom.get('datasets', []))}", + f" frameworks {len(aibom.get('frameworks', []))}", + f" prompts {len(aibom.get('prompts', []))}", + f" unsupported {len(aibom.get('unsupported_artifacts', []))}", + f" risks(high+) {len(high_or_critical)}", + "", + "Top risks", + ] + + if not top_risks: + lines.append(" - none") + else: + for item in top_risks: + lines.append( + f" - [{item.get('severity', 'unknown')}] {item.get('title', item.get('id', 'finding'))}" + ) + + lines.extend(["", "Coverage"]) + if not detectors: + lines.append(" - no detector coverage metadata") + else: + for item in detectors: + lines.append( + f" - {item.get('source_type', 'unknown')}: files={item.get('files_scanned', 0)} findings={item.get('findings', 0)}" + ) + + verdict = "pass" + if drift_failures: + verdict = f"fail ({', '.join(sorted(set(drift_failures)))})" + lines.extend(["", f"Drift/gate verdict: {verdict}"]) + return "\n".join(lines) + + +def render_markdown_summary(records: list[dict[str, Any]]) -> str: + lines = [ + "# AIBOM GitHub Scan Summary", + "", + "| Repository | Status | Models | Tools | High+ Risks | Unsupported | Gate Verdict |", + "|---|---:|---:|---:|---:|---:|---|", + ] + for record in records: + lines.append( + "| {repo} | {status} | {models} | {tools} | {risks} | {unsupported} | {verdict} |".format( + repo=record.get("repo", "unknown"), + status=record.get("status", "error"), + models=record.get("counts", {}).get("models", 0), + tools=record.get("counts", {}).get("tools", 0), + risks=record.get("counts", {}).get("high_or_critical_risks", 0), + unsupported=record.get("counts", {}).get("unsupported_artifacts", 0), + verdict=record.get("gate_verdict", "fail"), + ) + ) + + failed = [item["repo"] for item in records if item.get("status") != "ok"] + lines.extend(["", "## Aggregate"]) + lines.append(f"- scanned: {len(records)}") + lines.append(f"- failed: {len(failed)}") + if failed: + lines.append(f"- failed_repositories: {', '.join(sorted(failed))}") + return "\n".join(lines) + "\n" + + +def profile_json_dumps(data: dict[str, Any]) -> str: + return json.dumps(data, indent=2, sort_keys=True) diff --git a/aibom/utils.py b/aibom/utils.py index bc1954d..5f5e78a 100644 --- a/aibom/utils.py +++ b/aibom/utils.py @@ -5,7 +5,6 @@ import logging import os import platform -import re import subprocess import sys from datetime import datetime, timezone @@ -34,7 +33,7 @@ def validate_safe_path( ) -> Path: """ Validate that a path is safe to use in subprocess calls. - + Args: path: The path to validate must_exist: Whether the path must exist @@ -42,30 +41,28 @@ def validate_safe_path( must_be_dir: Whether the path must be a directory allow_symlinks: Whether symlinks are allowed base_dir: If provided, path must resolve to be within this directory - + Returns: The resolved, absolute path - + Raises: PathSecurityError: If the path fails security validation """ # Convert to Path if string path_obj = Path(path) - + # Check for shell metacharacters in the path string path_str = str(path_obj) if any(c in _SHELL_METACHARACTERS for c in path_str): bad_chars = [c for c in path_str if c in _SHELL_METACHARACTERS] - raise PathSecurityError( - f"Path contains shell metacharacters: {bad_chars[:5]}" - ) - + raise PathSecurityError(f"Path contains shell metacharacters: {bad_chars[:5]}") + # Convert to absolute path and resolve symlinks try: abs_path = path_obj.resolve(strict=False) except (OSError, ValueError) as e: raise PathSecurityError(f"Cannot resolve path: {e}") - + # Check if path is within base_dir (path traversal protection) if base_dir is not None: base_resolved = base_dir.resolve() @@ -75,26 +72,26 @@ def validate_safe_path( raise PathSecurityError( f"Path {abs_path} is outside allowed base directory {base_resolved}" ) - + # Check existence if must_exist and not abs_path.exists(): raise PathSecurityError(f"Path does not exist: {abs_path}") - + # Check if it's a symlink if not allow_symlinks and abs_path.is_symlink(): raise PathSecurityError(f"Symlinks are not allowed: {abs_path}") - + # Check file type if must_be_file and must_exist and not abs_path.is_file(): raise PathSecurityError(f"Path is not a file: {abs_path}") - + if must_be_dir and must_exist and not abs_path.is_dir(): raise PathSecurityError(f"Path is not a directory: {abs_path}") - + # Additional check: ensure the path doesn't contain null bytes if b"\x00" in os.fsencode(abs_path): raise PathSecurityError("Path contains null bytes") - + return abs_path @@ -119,7 +116,7 @@ def git_sha(cwd: Path) -> str: try: # Validate the working directory path safe_cwd = validate_safe_path(cwd, must_exist=True, must_be_dir=True) - + out = subprocess.check_output( ["git", "rev-parse", "HEAD"], cwd=str(safe_cwd), diff --git a/docs/GITHUB_SCANNER_GUIDE.md b/docs/GITHUB_SCANNER_GUIDE.md new file mode 100644 index 0000000..518b514 --- /dev/null +++ b/docs/GITHUB_SCANNER_GUIDE.md @@ -0,0 +1,54 @@ +# GitHub Scanner Guide + +## Overview + +`aibom scan-github` clones one or many GitHub repositories into a temporary workspace, runs the normal AIBOM generation pipeline, and writes per-repository outputs plus aggregate summaries. + +## Authentication + +By default the scanner reads `GITHUB_TOKEN` from the environment. + +```bash +export GITHUB_TOKEN=ghp_xxx +aibom scan-github --repo owner/name --output-dir out +``` + +Use a custom env var with `--token-env`: + +```bash +aibom scan-github --repo owner/name --token-env AIBOM_GH_TOKEN --output-dir out +``` + +## Rate limits and reliability + +- Prefer authenticated requests for larger scans. +- Use `--max-repos` to cap scope. +- Use `--timeout-sec` to avoid hanging clones. +- Partial failures are aggregated: failed repositories do not stop subsequent scans. + +## Drift/risk gates + +```bash +aibom scan-github \ + --repos-file repos.txt \ + --output-dir out \ + --fail-on new-model,new-tool,new-external-provider \ + --max-high-risk 0 \ + --max-unsupported 0 +``` + +A nonzero exit code is returned when any repository fails scan or fails a configured gate. + +## Output files + +- Per repo: `AI_BOM.json` +- Optional profile: `AI_BOM_ai_profile.json` (with `--profile ai-bom-like`) +- Aggregate machine output: `summary.json` +- Aggregate human output: `SUMMARY.md` + +## CI example + +See `.github/workflows/aibom.yml` for: +- pull request drift gate +- scheduled multi-repo scans +- artifact upload for scan outputs and summaries diff --git a/examples/github_repo_samples/repos.txt b/examples/github_repo_samples/repos.txt new file mode 100644 index 0000000..aa934aa --- /dev/null +++ b/examples/github_repo_samples/repos.txt @@ -0,0 +1,2 @@ +openai/openai-quickstart-python +langchain-ai/langchain diff --git a/tests/fixtures/golden_ai_profile.json b/tests/fixtures/golden_ai_profile.json new file mode 100644 index 0000000..3e47c40 --- /dev/null +++ b/tests/fixtures/golden_ai_profile.json @@ -0,0 +1,69 @@ +{ + "ai_assets": { + "datasets": [ + { + "source_file": "app.py", + "type": "FAISS.from_texts" + } + ], + "frameworks": [ + "langchain" + ], + "models": [ + { + "model": "gpt-4o-mini", + "provider": "https://api.openai.com/v1", + "source_file": "app.py", + "type": "ChatOpenAI" + } + ], + "tools": [ + { + "name": "initialize_agent", + "source_file": "app.py" + } + ] + }, + "detector_coverage": { + "detectors": [], + "unsupported_total": 0 + }, + "executive_summary": { + "datasets": 1, + "frameworks": 1, + "high_or_critical_risks": 1, + "models": 1, + "prompts": 0, + "risk_findings": 2, + "tools": 1, + "unsupported_artifacts": 0 + }, + "metadata": { + "artifact_sha256": "1111111111111111111111111111111111111111111111111111111111111111", + "generated_at": "20250101T010203Z", + "git_sha": "abc123def456" + }, + "profile": "ai-bom-like", + "provenance_and_compliance": { + "risk_policy": {}, + "runtime_context": {}, + "source_types": [] + }, + "risk_highlights": [ + { + "id": "exfil-surface:initialize_agent:app.py", + "rule_id": "exfil-surface", + "score": 0.0, + "severity": "high", + "title": "" + }, + { + "id": "third-party-provider:ChatOpenAI:app.py", + "rule_id": "third-party-provider", + "score": 0.0, + "severity": "medium", + "title": "" + } + ], + "schema_version": "1.0" +} \ No newline at end of file diff --git a/tests/fixtures/sample_project/alias_wrappers.py b/tests/fixtures/sample_project/alias_wrappers.py index 53c2c09..2a88509 100644 --- a/tests/fixtures/sample_project/alias_wrappers.py +++ b/tests/fixtures/sample_project/alias_wrappers.py @@ -3,6 +3,7 @@ PrimaryModel = ChatModel + def model_factory(model_name: str): return PrimaryModel(model=model_name) diff --git a/tests/test_cli.py b/tests/test_cli.py index 254b009..cf8beef 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -932,3 +932,57 @@ def test_risk_policy_allowlist_suppression_with_audit_trace(tmp_path: Path) -> N and s["reason"] == "approved-external-provider" for s in doc["risk_policy"]["suppressed"] ) + + +def test_cli_summarize_outputs_text_and_json(tmp_path: Path) -> None: + doc = generate_aibom(_fixture_project()) + input_path = tmp_path / "in.json" + input_path.write_text(json.dumps(doc), encoding="utf-8") + + text_proc = subprocess.run( + [sys.executable, "-m", "aibom.cli", "summarize", "--input", str(input_path)], + capture_output=True, + text=True, + check=False, + ) + assert text_proc.returncode == 0 + assert "AIBOM scan summary" in text_proc.stdout + + json_proc = subprocess.run( + [sys.executable, "-m", "aibom.cli", "summarize", "--input", str(input_path), "--json"], + capture_output=True, + text=True, + check=False, + ) + assert json_proc.returncode == 0 + payload = json.loads(json_proc.stdout) + assert payload["counts"]["models"] >= 1 + + +def test_cmd_scan_github_requires_repo_input() -> None: + from aibom import cli as cli_module + + rc = cli_module.cmd_scan_github( + argparse.Namespace( + repo=[], + repos_file=None, + output_dir="out", + branch=None, + depth=1, + token_env="GITHUB_TOKEN", + max_repos=None, + timeout_sec=30, + profile="canonical", + include_prompts=False, + acknowledge_prompt_exposure_risk=False, + include_runtime_manifests=False, + redaction_policy="strict", + risk_policy=None, + baseline=None, + fail_on=None, + max_high_risk=None, + max_unsupported=None, + json=False, + ) + ) + assert rc == 2 diff --git a/tests/test_exporters.py b/tests/test_exporters.py index 639496f..f7960a1 100644 --- a/tests/test_exporters.py +++ b/tests/test_exporters.py @@ -4,6 +4,7 @@ from pathlib import Path from aibom.exporters import export_cyclonedx, export_sarif, export_spdx, export_vex +from aibom.presentation import build_ai_bom_like_profile FIXTURES = Path(__file__).parent / "fixtures" @@ -69,3 +70,10 @@ def test_export_vex_matches_golden_fixture() -> None: assert actual == expected assert actual["@context"].startswith("https://openvex.dev") assert actual["statements"] + + +def test_ai_bom_like_profile_deterministic_rendering() -> None: + aibom_doc = _load_fixture("export_input_aibom.json") + first = build_ai_bom_like_profile(aibom_doc) + second = build_ai_bom_like_profile(aibom_doc) + assert first == second diff --git a/tests/test_github_scan.py b/tests/test_github_scan.py new file mode 100644 index 0000000..3882208 --- /dev/null +++ b/tests/test_github_scan.py @@ -0,0 +1,52 @@ +from __future__ import annotations + +import json +from pathlib import Path + +from aibom.github_scan import _load_repos, scan_github_repos + + +FIXTURE = Path(__file__).parent / "fixtures" / "sample_project" + + +def test_load_repos_dedupes_and_reads_file(tmp_path: Path) -> None: + repos_file = tmp_path / "repos.txt" + repos_file.write_text("owner/a\n# comment\nowner/b\nowner/a\n", encoding="utf-8") + + repos = _load_repos(["owner/c", "owner/b"], str(repos_file)) + + assert repos == ["owner/c", "owner/b", "owner/a"] + + +def test_scan_github_repos_generates_summary_with_partial_failures( + tmp_path: Path, monkeypatch +) -> None: + from aibom import github_scan as mod + + def fake_clone(repo: str, dest: Path, **_kwargs: object) -> None: + if repo == "bad/repo": + raise RuntimeError("clone failed") + dest.mkdir(parents=True, exist_ok=True) + (dest / "app.py").write_text( + FIXTURE.joinpath("app.py").read_text(encoding="utf-8"), encoding="utf-8" + ) + + monkeypatch.setattr(mod, "_clone_repo", fake_clone) + + records, exit_code = scan_github_repos( + repos=["good/repo", "bad/repo"], + output_dir=tmp_path / "out", + profile="ai-bom-like", + max_high_risk=0, + ) + + assert exit_code == 2 + assert len(records) == 2 + assert any(record.status == "ok" for record in records) + assert any(record.status == "error" for record in records) + + summary_json = json.loads((tmp_path / "out" / "summary.json").read_text(encoding="utf-8")) + assert summary_json["total_repositories"] == 2 + assert (tmp_path / "out" / "SUMMARY.md").exists() + assert (tmp_path / "out" / "good__repo" / "AI_BOM.json").exists() + assert (tmp_path / "out" / "good__repo" / "AI_BOM_ai_profile.json").exists() diff --git a/tests/test_presentation.py b/tests/test_presentation.py new file mode 100644 index 0000000..49b9684 --- /dev/null +++ b/tests/test_presentation.py @@ -0,0 +1,67 @@ +from __future__ import annotations + +import json +from pathlib import Path + +from aibom.presentation import ( + build_ai_bom_like_profile, + render_markdown_summary, + render_text_summary, +) + + +FIXTURES = Path(__file__).parent / "fixtures" + + +def _load_fixture(name: str) -> dict: + return json.loads((FIXTURES / name).read_text(encoding="utf-8")) + + +def test_ai_bom_like_profile_matches_snapshot() -> None: + aibom_doc = _load_fixture("export_input_aibom.json") + expected = _load_fixture("golden_ai_profile.json") + + assert build_ai_bom_like_profile(aibom_doc) == expected + + +def test_text_summary_contains_core_sections() -> None: + aibom_doc = _load_fixture("export_input_aibom.json") + summary = render_text_summary(aibom_doc) + + assert "AIBOM scan summary" in summary + assert "Top risks" in summary + assert "Coverage" in summary + assert "Drift/gate verdict: pass" in summary + + +def test_markdown_summary_table_rendering() -> None: + markdown = render_markdown_summary( + [ + { + "repo": "octo/demo", + "status": "ok", + "counts": { + "models": 2, + "tools": 1, + "high_or_critical_risks": 1, + "unsupported_artifacts": 0, + }, + "gate_verdict": "pass", + }, + { + "repo": "octo/bad", + "status": "error", + "counts": { + "models": 0, + "tools": 0, + "high_or_critical_risks": 0, + "unsupported_artifacts": 0, + }, + "gate_verdict": "fail", + }, + ] + ) + + assert "| Repository | Status |" in markdown + assert "octo/demo" in markdown + assert "failed: 1" in markdown