Audit a git repository for compromise indicators after GHES incidents. This script is defensive. It cannot prove a repository or server is clean. It flags suspicious signals in a local clone and optional exported log files.

check_repo_compromise.py

"""Audit a git repository for compromise indicators after GHES incidents.

This script is defensive. It cannot prove a repository or server is clean.
It flags suspicious signals in a local clone and optional exported log files.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, 
INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, 
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. 

IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, 
WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, 
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
"""

from __future__ import annotations

import argparse
import json
import subprocess
import sys
from pathlib import Path
from typing import Any

HIGH_RISK_PATH_PATTERNS = (
    ".github/workflows/",
    ".github/actions/",
    ".gitmodules",
    "scripts/",
    ".devcontainer/",
    "Dockerfile",
    "docker-compose",
    "package.json",
    "package-lock.json",
    "pnpm-lock.yaml",
    "yarn.lock",
    "requirements.txt",
    "pyproject.toml",
    "setup.py",
)


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    """Parse command-line arguments for the audit helper."""
    parser = argparse.ArgumentParser(
        description=(
            "Audit a local git repository for indicators of compromise after "
            "GHES incidents such as CVE-2026-3854."
        )
    )
    parser.add_argument(
        "repo",
        nargs="?",
        default=".",
        help="Path to the local git repository to audit. Defaults to the current directory.",
    )
    parser.add_argument(
        "--trusted-author",
        action="append",
        default=[],
        help="Trusted author or committer email. Repeat to allow more than one.",
    )
    parser.add_argument(
        "--recent-commit-limit",
        type=int,
        default=50,
        help="How many recent commits to inspect. Default: 50.",
    )
    parser.add_argument(
        "--baseline-refs",
        type=Path,
        help="Optional JSON file with expected refs, mapping ref names to commit hashes.",
    )
    parser.add_argument(
        "--log-export",
        type=Path,
        help="Optional GHES log export or text file to scan for suspicious push-option patterns.",
    )
    parser.add_argument(
        "--json",
        action="store_true",
        help="Emit the full report as JSON.",
    )
    return parser.parse_args(argv)


def run_git(repo_dir: Path, *args: str) -> str:
    """Run a git command in the target repository and return stdout."""
    result = subprocess.run(
        ["git", *args],
        cwd=repo_dir,
        check=False,
        capture_output=True,
        text=True,
    )
    if result.returncode != 0:
        message = result.stderr.strip() or result.stdout.strip() or "git command failed"
        raise RuntimeError(f"git {' '.join(args)} failed: {message}")
    return result.stdout


def assert_git_repository(repo_dir: Path) -> None:
    """Ensure the target directory is a git work tree."""
    output = run_git(repo_dir, "rev-parse", "--is-inside-work-tree").strip().lower()
    if output != "true":
        raise RuntimeError(f"{repo_dir} is not a git repository")


def load_baseline_refs(baseline_refs_path: Path | None) -> dict[str, str]:
    """Load a baseline ref snapshot from JSON if provided."""
    if baseline_refs_path is None:
        return {}
    data = json.loads(baseline_refs_path.read_text(encoding="utf-8"))
    if not isinstance(data, dict):
        raise RuntimeError(
            "Baseline refs JSON must be an object mapping refs to hashes"
        )
    return {
        str(ref_name): str(object_id)
        for ref_name, object_id in data.items()
        if ref_name and object_id
    }


def list_refs(repo_dir: Path) -> dict[str, str]:
    """Return local branch and tag refs."""
    output = run_git(
        repo_dir,
        "for-each-ref",
        "--format=%(refname)%09%(objectname)",
        "refs/heads",
        "refs/tags",
    )
    refs: dict[str, str] = {}
    for line in output.splitlines():
        if not line.strip():
            continue
        ref_name, object_id = line.split("\t", 1)
        refs[ref_name] = object_id
    return refs


def list_recent_commits(repo_dir: Path, limit: int) -> list[dict[str, Any]]:
    """Return recent commit metadata plus the files each commit changed."""
    if limit <= 0:
        return []

    commit_ids = [
        line.strip()
        for line in run_git(
            repo_dir, "rev-list", f"--max-count={limit}", "HEAD"
        ).splitlines()
        if line.strip()
    ]
    commits: list[dict[str, Any]] = []
    for commit_id in commit_ids:
        metadata = (
            run_git(
                repo_dir,
                "show",
                "--quiet",
                f"--format=%H%x1f%an%x1f%ae%x1f%cn%x1f%ce%x1f%s%x1f%ct",
                commit_id,
            )
            .strip()
            .split("\x1f")
        )
        if len(metadata) != 7:
            continue
        files = [
            line.strip()
            for line in run_git(
                repo_dir,
                "diff-tree",
                "--no-commit-id",
                "--name-only",
                "-r",
                commit_id,
            ).splitlines()
            if line.strip()
        ]
        commits.append(
            {
                "commit": metadata[0],
                "author_name": metadata[1],
                "author_email": metadata[2],
                "committer_name": metadata[3],
                "committer_email": metadata[4],
                "subject": metadata[5],
                "timestamp": int(metadata[6]),
                "files": files,
            }
        )
    return commits


def load_git_status(repo_dir: Path) -> list[str]:
    """Return the porcelain git status lines."""
    output = run_git(repo_dir, "status", "--short")
    return [line for line in output.splitlines() if line.strip()]


def has_high_risk_path(path: str) -> bool:
    """Return true when a changed path touches a sensitive automation surface."""
    normalized = path.replace("\\", "/")
    return any(pattern in normalized for pattern in HIGH_RISK_PATH_PATTERNS)


def scan_log_export(log_export_path: Path | None) -> list[dict[str, Any]]:
    """Scan an exported text log for suspicious push-option/header-injection markers."""
    if log_export_path is None or not log_export_path.is_file():
        return []

    findings: list[dict[str, Any]] = []
    lines = log_export_path.read_text(encoding="utf-8", errors="replace").splitlines()
    suspicious_tokens = (
        "push-option",
        "%0a",
        "%0d",
        "\\n",
        "\\r",
        "x-github-",
        "x-git-",
    )
    for index, line in enumerate(lines, start=1):
        lower_line = line.lower()
        if any(token in lower_line for token in suspicious_tokens):
            findings.append(
                {
                    "kind": "suspicious-log-entry",
                    "severity": "medium",
                    "path": str(log_export_path),
                    "line": index,
                    "message": "Log entry contains push-option or header-injection markers.",
                    "content": line.strip(),
                }
            )
    return findings


def audit_repository(
    repo_dir: Path | str,
    *,
    trusted_authors: set[str] | None = None,
    baseline_refs_path: Path | None = None,
    recent_commit_limit: int = 50,
    log_export_path: Path | None = None,
) -> dict[str, Any]:
    """Audit a repository and return a JSON-serializable report."""
    target_repo = Path(repo_dir).resolve()
    trusted = {
        value.strip().lower() for value in (trusted_authors or set()) if value.strip()
    }

    assert_git_repository(target_repo)
    refs = list_refs(target_repo)
    baseline_refs = load_baseline_refs(baseline_refs_path)
    commits = list_recent_commits(target_repo, recent_commit_limit)
    status_lines = load_git_status(target_repo)

    findings: list[dict[str, Any]] = []

    for ref_name, expected_object in baseline_refs.items():
        actual_object = refs.get(ref_name)
        if actual_object is None:
            findings.append(
                {
                    "kind": "missing-ref",
                    "severity": "high",
                    "ref": ref_name,
                    "expected": expected_object,
                    "message": "Baseline ref is missing from the current repository.",
                }
            )
        elif actual_object != expected_object:
            findings.append(
                {
                    "kind": "ref-drift",
                    "severity": "high",
                    "ref": ref_name,
                    "expected": expected_object,
                    "actual": actual_object,
                    "message": "Ref no longer matches the saved baseline.",
                }
            )

    for commit in commits:
        author_email = commit["author_email"].strip().lower()
        committer_email = commit["committer_email"].strip().lower()
        files = commit["files"]
        if trusted and author_email not in trusted and committer_email not in trusted:
            findings.append(
                {
                    "kind": "unknown-author",
                    "severity": "medium",
                    "commit": commit["commit"],
                    "author_email": commit["author_email"],
                    "committer_email": commit["committer_email"],
                    "message": "Commit author and committer are both outside the trusted allowlist.",
                }
            )
        risky_files = [path for path in files if has_high_risk_path(path)]
        if risky_files:
            findings.append(
                {
                    "kind": "high-risk-path-change",
                    "severity": "high",
                    "commit": commit["commit"],
                    "paths": risky_files,
                    "message": "Commit touched high-risk automation or supply-chain paths.",
                }
            )

    for status_line in status_lines:
        findings.append(
            {
                "kind": "working-tree-change",
                "severity": "low",
                "status": status_line[:2],
                "path": status_line[3:],
                "message": "Working tree is not clean. Review before trusting the repo state.",
            }
        )

    findings.extend(scan_log_export(log_export_path))

    return {
        "repo": str(target_repo),
        "baseline_refs_path": str(baseline_refs_path) if baseline_refs_path else None,
        "log_export_path": str(log_export_path) if log_export_path else None,
        "recent_commit_limit": recent_commit_limit,
        "trusted_authors": sorted(trusted),
        "head": run_git(target_repo, "rev-parse", "HEAD").strip(),
        "current_branch": run_git(target_repo, "branch", "--show-current").strip(),
        "refs": refs,
        "findings": findings,
    }


def print_human_report(report: dict[str, Any]) -> None:
    """Print a concise human-readable report."""
    print(f"Repository: {report['repo']}")
    print(f"Current branch: {report['current_branch'] or '(detached HEAD)'}")
    print(f"HEAD: {report['head']}")
    print()
    if not report["findings"]:
        print("No compromise indicators were found by this local audit.")
        print("This does not prove the GHES server was not exploited.")
        return

    print(f"Findings: {len(report['findings'])}")
    for finding in report["findings"]:
        location = (
            finding.get("commit") or finding.get("ref") or finding.get("path") or "n/a"
        )
        print(f"- [{finding['severity']}] {finding['kind']} :: {location}")
        print(f"  {finding['message']}")


def main(argv: list[str] | None = None) -> int:
    """Run the compromise audit CLI."""
    args = parse_args(argv)
    try:
        report = audit_repository(
            args.repo,
            trusted_authors=set(args.trusted_author),
            baseline_refs_path=args.baseline_refs,
            recent_commit_limit=args.recent_commit_limit,
            log_export_path=args.log_export,
        )
    except Exception as exc:
        print(f"Audit failed: {exc}", file=sys.stderr)
        return 2

    if args.json:
        print(json.dumps(report, indent=2, sort_keys=True))
    else:
        print_human_report(report)
    return 1 if report["findings"] else 0


if __name__ == "__main__":
    raise SystemExit(main())