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/<repo>/code-scanner/<YYYYMMDDHHMMSS>) 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) <noreply@anthropic.com>
This commit is contained in:
Kevin Veen-Birkenbach
2026-06-28 02:36:46 +02:00
parent e8aed49a4c
commit 13fc8fe885
7 changed files with 329 additions and 0 deletions

View File

@@ -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,
)

View File

@@ -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",

View File

@@ -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}")

View File

@@ -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",

View File

@@ -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

View File

@@ -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/<repo>/code-scanner/<YYYYMMDDHHMMSS>."
),
)
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/<repo>/code-scanner/<YYYYMMDDHHMMSS>)."),
)
parser.add_argument(
"--state",
default=None,
choices=["open", "closed", "dismissed", "fixed"],
help="Only download alerts in this state (default: all states).",
)

View File

@@ -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()