From 13fc8fe885575e5bd828a1efc3ff2ee84c0716c9 Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Sun, 28 Jun 2026 02:36:46 +0200 Subject: [PATCH] feat(cli): add code-scanning command to download GitHub scan results pkgmgr code-scanning fetches a repository's GitHub code scanning alerts (and analysis metadata) via the gh CLI and writes alerts.json, a readable summary.md digest, and analyses.json into a timestamped directory (default /tmp//code-scanner/) for offline analysis. The repository is resolved from gh or --repo; authentication is delegated to gh; --state filters by alert state. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/pkgmgr/actions/code_scanning/__init__.py | 161 ++++++++++++++++++ src/pkgmgr/cli/commands/__init__.py | 2 + src/pkgmgr/cli/commands/code_scanning.py | 24 +++ src/pkgmgr/cli/dispatch.py | 5 + src/pkgmgr/cli/parser/__init__.py | 2 + src/pkgmgr/cli/parser/code_scanning_cmd.py | 39 +++++ .../code_scanning/test_code_scanning.py | 96 +++++++++++ 7 files changed, 329 insertions(+) create mode 100644 src/pkgmgr/actions/code_scanning/__init__.py create mode 100644 src/pkgmgr/cli/commands/code_scanning.py create mode 100644 src/pkgmgr/cli/parser/code_scanning_cmd.py create mode 100644 tests/unit/pkgmgr/actions/code_scanning/test_code_scanning.py diff --git a/src/pkgmgr/actions/code_scanning/__init__.py b/src/pkgmgr/actions/code_scanning/__init__.py new file mode 100644 index 0000000..814a4b6 --- /dev/null +++ b/src/pkgmgr/actions/code_scanning/__init__.py @@ -0,0 +1,161 @@ +"""Download GitHub code scanning results via the ``gh`` CLI. + +Fetches all code scanning alerts (and the analysis metadata) for a +repository and writes them, plus a readable digest, into a timestamped +directory so they can be read and analysed offline. Authentication is +delegated to ``gh`` (credentials from its keyring/login). +""" + +from __future__ import annotations + +import json +import os +import shutil +import subprocess +from collections import Counter +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any, List, Optional + + +class CodeScanningError(RuntimeError): + """Raised when code scanning results cannot be downloaded.""" + + +@dataclass +class CodeScanningResult: + repo: str + output_dir: str + alert_count: int + files: List[str] = field(default_factory=list) + + +def _gh(args: List[str]) -> "subprocess.CompletedProcess[str]": + return subprocess.run(["gh", *args], capture_output=True, text=True) + + +def _resolve_repo(repo: Optional[str]) -> str: + if repo: + return repo + proc = _gh(["repo", "view", "--json", "nameWithOwner", "-q", ".nameWithOwner"]) + name = proc.stdout.strip() + if proc.returncode != 0 or not name: + raise CodeScanningError( + "could not resolve the current repository; run inside a GitHub " + "repo or pass --repo OWNER/REPO " + f"({proc.stderr.strip()})" + ) + return name + + +def _fetch_json(endpoint: str, params: Optional[List[str]] = None) -> Any: + args = ["api", endpoint, "--paginate"] + for param in params or []: + args += ["-f", param] + proc = _gh(args) + if proc.returncode != 0: + raise CodeScanningError( + f"gh api {endpoint} failed: {proc.stderr.strip() or 'unknown error'}" + ) + body = proc.stdout.strip() + return json.loads(body) if body else [] + + +def _alert_row(alert: dict) -> str: + rule = alert.get("rule") or {} + instance = alert.get("most_recent_instance") or {} + location = instance.get("location") or {} + message = (instance.get("message") or {}).get("text", "").strip().replace("\n", " ") + severity = rule.get("security_severity_level") or rule.get("severity") or "unknown" + path = location.get("path", "?") + line = location.get("start_line", "?") + state = alert.get("state", "?") + rule_id = rule.get("id", "?") + return f"- [{severity}] {rule_id} — {path}:{line} ({state})\n {message}" + + +def _build_summary(repo: str, generated_at: str, alerts: List[dict]) -> str: + by_severity: Counter = Counter() + by_state: Counter = Counter() + by_rule: Counter = Counter() + for alert in alerts: + rule = alert.get("rule") or {} + by_severity[ + rule.get("security_severity_level") or rule.get("severity") or "unknown" + ] += 1 + by_state[alert.get("state", "unknown")] += 1 + by_rule[rule.get("id", "unknown")] += 1 + + lines = [ + f"# Code scanning summary — {repo}", + "", + f"- Generated: {generated_at}", + f"- Total alerts: {len(alerts)}", + "", + "## By severity", + "", + ] + lines += [f"- {sev}: {count}" for sev, count in by_severity.most_common()] or [ + "- none" + ] + lines += ["", "## By state", ""] + lines += [f"- {state}: {count}" for state, count in by_state.most_common()] or [ + "- none" + ] + lines += ["", "## By rule", ""] + lines += [f"- {rule}: {count}" for rule, count in by_rule.most_common()] or [ + "- none" + ] + lines += ["", "## Alerts", ""] + lines += [_alert_row(alert) for alert in alerts] or ["- none"] + return "\n".join(lines) + "\n" + + +def download_code_scanning( + repo: Optional[str] = None, + output_dir: Optional[str] = None, + state: Optional[str] = None, +) -> CodeScanningResult: + if not shutil.which("gh"): + raise CodeScanningError("the GitHub CLI 'gh' is not installed or not on PATH") + + repo = _resolve_repo(repo) + repo_name = repo.rstrip("/").split("/")[-1] + + if output_dir is None: + timestamp = datetime.now().strftime("%Y%m%d%H%M%S") + output_dir = os.path.join("/tmp", repo_name, "code-scanner", timestamp) + output_dir = os.path.abspath(os.path.expanduser(output_dir)) + os.makedirs(output_dir, exist_ok=True) + generated_at = datetime.now().isoformat(timespec="seconds") + + alerts = _fetch_json( + f"repos/{repo}/code-scanning/alerts", + params=[f"state={state}"] if state else None, + ) + + files: List[str] = [] + + def _write(name: str, content: str) -> None: + path = os.path.join(output_dir, name) + with open(path, "w", encoding="utf-8") as handle: + handle.write(content) + files.append(path) + + _write("alerts.json", json.dumps(alerts, indent=2, ensure_ascii=False) + "\n") + _write("summary.md", _build_summary(repo, generated_at, alerts)) + + try: + analyses = _fetch_json(f"repos/{repo}/code-scanning/analyses") + _write( + "analyses.json", json.dumps(analyses, indent=2, ensure_ascii=False) + "\n" + ) + except CodeScanningError as exc: + _write("analyses.json", json.dumps({"error": str(exc)}, indent=2) + "\n") + + return CodeScanningResult( + repo=repo, + output_dir=output_dir, + alert_count=len(alerts), + files=files, + ) diff --git a/src/pkgmgr/cli/commands/__init__.py b/src/pkgmgr/cli/commands/__init__.py index faf8cde..9bdc8ef 100644 --- a/src/pkgmgr/cli/commands/__init__.py +++ b/src/pkgmgr/cli/commands/__init__.py @@ -1,4 +1,5 @@ from .archive import handle_archive +from .code_scanning import handle_code_scanning from .repos import handle_repos_command from .config import handle_config from .tools import handle_tools_command @@ -12,6 +13,7 @@ from .mirror import handle_mirror_command __all__ = [ "handle_archive", + "handle_code_scanning", "handle_repos_command", "handle_config", "handle_tools_command", diff --git a/src/pkgmgr/cli/commands/code_scanning.py b/src/pkgmgr/cli/commands/code_scanning.py new file mode 100644 index 0000000..367d283 --- /dev/null +++ b/src/pkgmgr/cli/commands/code_scanning.py @@ -0,0 +1,24 @@ +from __future__ import annotations + +import sys + +from pkgmgr.actions.code_scanning import CodeScanningError, download_code_scanning +from pkgmgr.cli.context import CLIContext + + +def handle_code_scanning(args, ctx: CLIContext) -> None: + try: + result = download_code_scanning( + repo=getattr(args, "repo", None), + output_dir=getattr(args, "output", None), + state=getattr(args, "state", None), + ) + except CodeScanningError as exc: + print(f"[ERROR] {exc}", file=sys.stderr) + sys.exit(1) + + print(f"[code-scanning] Repository: {result.repo}") + print(f"[code-scanning] Alerts: {result.alert_count}") + print(f"[code-scanning] Output: {result.output_dir}") + for path in result.files: + print(f" - {path}") diff --git a/src/pkgmgr/cli/dispatch.py b/src/pkgmgr/cli/dispatch.py index 94f54de..4a718d7 100644 --- a/src/pkgmgr/cli/dispatch.py +++ b/src/pkgmgr/cli/dispatch.py @@ -11,6 +11,7 @@ from pkgmgr.core.repository.dir import get_repo_dir from pkgmgr.cli.commands import ( handle_archive, + handle_code_scanning, handle_repos_command, handle_tools_command, handle_release, @@ -65,6 +66,10 @@ def dispatch_command(args, ctx: CLIContext) -> None: handle_archive(args, ctx) return + if args.command == "code-scanning": + handle_code_scanning(args, ctx) + return + commands_with_selection = { "install", "update", diff --git a/src/pkgmgr/cli/parser/__init__.py b/src/pkgmgr/cli/parser/__init__.py index 63307b5..390ca88 100644 --- a/src/pkgmgr/cli/parser/__init__.py +++ b/src/pkgmgr/cli/parser/__init__.py @@ -7,6 +7,7 @@ from pkgmgr.cli.proxy import register_proxy_commands from .archive_cmd import add_archive_subparser from .branch_cmd import add_branch_subparsers from .changelog_cmd import add_changelog_subparser +from .code_scanning_cmd import add_code_scanning_subparser from .common import SortedSubParsersAction from .config_cmd import add_config_subparsers from .install_update import add_install_update_subparsers @@ -67,6 +68,7 @@ def create_parser(description_text: str) -> argparse.ArgumentParser: add_make_subparsers(subparsers) add_mirror_subparsers(subparsers) add_archive_subparser(subparsers) + add_code_scanning_subparser(subparsers) register_proxy_commands(subparsers) return parser diff --git a/src/pkgmgr/cli/parser/code_scanning_cmd.py b/src/pkgmgr/cli/parser/code_scanning_cmd.py new file mode 100644 index 0000000..8f7b820 --- /dev/null +++ b/src/pkgmgr/cli/parser/code_scanning_cmd.py @@ -0,0 +1,39 @@ +from __future__ import annotations + +import argparse + + +def add_code_scanning_subparser(subparsers: argparse._SubParsersAction) -> None: + """Register the code-scanning subcommand. + + Downloads a repository's GitHub code scanning alerts (and analysis + metadata) into a timestamped directory for offline analysis, using the + ``gh`` CLI for authentication. + """ + parser = subparsers.add_parser( + "code-scanning", + help="Download GitHub code scanning results for offline analysis.", + description=( + "Fetch all code scanning alerts (plus analysis metadata) for a " + "repository via the gh CLI and write them, with a readable " + "summary, into a timestamped directory. Defaults to the current " + "repository and /tmp//code-scanner/." + ), + ) + parser.add_argument( + "--repo", + default=None, + help="Target repository as OWNER/REPO (default: current repo via gh).", + ) + parser.add_argument( + "--output", + "-o", + default=None, + help=("Output directory (default: /tmp//code-scanner/)."), + ) + parser.add_argument( + "--state", + default=None, + choices=["open", "closed", "dismissed", "fixed"], + help="Only download alerts in this state (default: all states).", + ) diff --git a/tests/unit/pkgmgr/actions/code_scanning/test_code_scanning.py b/tests/unit/pkgmgr/actions/code_scanning/test_code_scanning.py new file mode 100644 index 0000000..17a3d76 --- /dev/null +++ b/tests/unit/pkgmgr/actions/code_scanning/test_code_scanning.py @@ -0,0 +1,96 @@ +from __future__ import annotations + +import json +import os +import shutil +import tempfile +import unittest +from types import SimpleNamespace +from unittest import mock + +from pkgmgr.actions.code_scanning import ( + CodeScanningError, + download_code_scanning, +) + +_ALERTS = [ + { + "number": 1, + "state": "open", + "html_url": "https://example/1", + "rule": { + "id": "py/sql-injection", + "security_severity_level": "high", + "severity": "error", + "description": "SQL injection", + }, + "most_recent_instance": { + "location": {"path": "app.py", "start_line": 10}, + "message": {"text": "Possible SQL injection"}, + }, + "tool": {"name": "CodeQL"}, + } +] + + +def _cp(stdout: str = "", returncode: int = 0, stderr: str = "") -> SimpleNamespace: + return SimpleNamespace(stdout=stdout, returncode=returncode, stderr=stderr) + + +def _fake_gh(args): + if args[:2] == ["repo", "view"]: + return _cp(stdout="owner/pkgmgr-cs-test\n") + endpoint = args[1] if len(args) > 1 else "" + if endpoint.endswith("code-scanning/alerts"): + return _cp(stdout=json.dumps(_ALERTS)) + if endpoint.endswith("code-scanning/analyses"): + return _cp(stdout=json.dumps([{"id": 42, "created_at": "2026-01-01"}])) + return _cp(returncode=1, stderr=f"unexpected: {args}") + + +class TestDownloadCodeScanning(unittest.TestCase): + def test_writes_alerts_summary_and_analyses(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + with ( + mock.patch( + "pkgmgr.actions.code_scanning.shutil.which", + return_value="/usr/bin/gh", + ), + mock.patch("pkgmgr.actions.code_scanning._gh", side_effect=_fake_gh), + ): + result = download_code_scanning(output_dir=tmp) + + self.assertEqual(result.repo, "owner/pkgmgr-cs-test") + self.assertEqual(result.alert_count, 1) + + with open(os.path.join(tmp, "alerts.json"), encoding="utf-8") as f: + self.assertEqual(json.load(f), _ALERTS) + with open(os.path.join(tmp, "summary.md"), encoding="utf-8") as f: + summary = f.read() + self.assertIn("py/sql-injection", summary) + self.assertIn("app.py:10", summary) + self.assertIn("high", summary) + self.assertTrue(os.path.exists(os.path.join(tmp, "analyses.json"))) + + def test_default_output_dir_uses_repo_name(self) -> None: + with ( + mock.patch( + "pkgmgr.actions.code_scanning.shutil.which", return_value="/usr/bin/gh" + ), + mock.patch("pkgmgr.actions.code_scanning._gh", side_effect=_fake_gh), + ): + result = download_code_scanning() + self.addCleanup(shutil.rmtree, "/tmp/pkgmgr-cs-test", ignore_errors=True) + self.assertTrue( + result.output_dir.startswith("/tmp/pkgmgr-cs-test/code-scanner/"), + msg=result.output_dir, + ) + + def test_errors_when_gh_missing(self) -> None: + with mock.patch("pkgmgr.actions.code_scanning.shutil.which", return_value=None): + with self.assertRaises(CodeScanningError): + download_code_scanning(output_dir="/tmp/should-not-be-created") + + +if __name__ == "__main__": # pragma: no cover + unittest.main()