#!/usr/bin/env python3
"""Heuristically audit GitHub Actions workflows for permission hazards.

Uses only the Python standard library. It does not execute workflow code, parse
secrets, or make network requests. Findings require human review.
"""
from __future__ import annotations

import argparse
import json
import re
from pathlib import Path

MAX_BYTES = 300_000
WRITE_SCOPES = re.compile(r"(?im)^\s*(contents|actions|checks|deployments|issues|packages|pages|pull-requests|repository-projects|security-events|statuses)\s*:\s*write\s*(?:#.*)?$")
ALL_WRITE = re.compile(r"(?im)^\s*permissions\s*:\s*write-all\s*(?:#.*)?$")
PR_TARGET = re.compile(r"(?m)^\s*pull_request_target\s*:")
CHECKOUT = re.compile(r"(?im)^\s*(?:-\s*)?uses\s*:\s*actions/checkout@")
HEAD_REF = re.compile(r"github\.event\.pull_request\.head\.(?:sha|ref)|github\.head_ref", re.I)
PERSIST_CREDENTIALS = re.compile(r"(?im)^\s*persist-credentials\s*:\s*(?:true|yes|on)\s*(?:#.*)?$")
SECRET_REF = re.compile(r"\$\{\{\s*secrets\.[A-Za-z_][A-Za-z0-9_]*\s*}}")
UNPINNED_ACTION = re.compile(r"(?im)^\s*(?:-\s*)?uses\s*:\s*([^\s#]+)@([^\s#]+)")
COMMIT_SHA = re.compile(r"^[0-9a-f]{40}$", re.I)


def line_number(text: str, offset: int) -> int:
    return text.count('\n', 0, offset) + 1


def finding(path: str, line: int, severity: str, rule: str, message: str, remediation: str) -> dict:
    return {
        'path': path, 'line': line, 'severity': severity, 'rule': rule,
        'message': message, 'remediation': remediation,
    }


def audit_text(relative: str, text: str) -> list[dict]:
    findings: list[dict] = []
    pr_target = PR_TARGET.search(text)
    checkout = CHECKOUT.search(text)
    head_ref = HEAD_REF.search(text)

    if pr_target and checkout and head_ref:
        findings.append(finding(
            relative, line_number(text, head_ref.start()), 'critical',
            'pr-target-untrusted-checkout',
            'pull_request_target appears to check out pull-request head code.',
            'Do not execute untrusted PR code with the privileged pull_request_target token; use pull_request or split trusted write work into a separate workflow.',
        ))

    match = ALL_WRITE.search(text)
    if match:
        findings.append(finding(
            relative, line_number(text, match.start()), 'high', 'write-all',
            'Workflow grants write-all permissions.',
            'Replace write-all with the minimum job-scoped permissions required.',
        ))

    for match in WRITE_SCOPES.finditer(text):
        scope = match.group(1)
        findings.append(finding(
            relative, line_number(text, match.start()), 'medium', 'write-permission',
            f'Workflow grants {scope}: write.',
            'Confirm the write operation is necessary and prefer granting it only to the specific trusted job.',
        ))

    if pr_target:
        for match in SECRET_REF.finditer(text):
            findings.append(finding(
                relative, line_number(text, match.start()), 'high', 'secret-on-pr-target',
                'A secret reference appears in a pull_request_target workflow.',
                'Ensure no untrusted pull-request code, arguments, artifacts, or scripts can access or influence this secret-consuming step.',
            ))

    for match in PERSIST_CREDENTIALS.finditer(text):
        findings.append(finding(
            relative, line_number(text, match.start()), 'low', 'persisted-checkout-credentials',
            'Checkout credentials are explicitly persisted.',
            'Set persist-credentials: false unless later git operations require the workflow token.',
        ))

    for match in UNPINNED_ACTION.finditer(text):
        action, ref = match.groups()
        if action.startswith('./') or action.startswith('docker://') or COMMIT_SHA.fullmatch(ref):
            continue
        findings.append(finding(
            relative, line_number(text, match.start()), 'info', 'action-not-sha-pinned',
            f'{action}@{ref} is not pinned to a full commit SHA.',
            'For stronger supply-chain integrity, pin third-party actions to a reviewed full commit SHA and record the release version in a comment.',
        ))

    return findings


def audit(root: Path) -> dict:
    root = root.resolve()
    workflow_dir = root / '.github' / 'workflows'
    files = []
    findings = []
    warnings = []
    if not workflow_dir.is_dir() or workflow_dir.is_symlink():
        return {'schema': 'gha-permission-audit-v1', 'files': files, 'findings': findings, 'warnings': ['No .github/workflows directory found.'], 'heuristic_only': True}

    for path in sorted(workflow_dir.iterdir()):
        if path.is_symlink() or not path.is_file() or path.suffix.lower() not in {'.yml', '.yaml'}:
            continue
        relative = path.relative_to(root).as_posix()
        size = path.stat().st_size
        if size > MAX_BYTES:
            warnings.append(f'Skipped oversized workflow: {relative}')
            continue
        text = path.read_text(encoding='utf-8', errors='replace')
        files.append(relative)
        findings.extend(audit_text(relative, text))

    order = {'critical': 0, 'high': 1, 'medium': 2, 'low': 3, 'info': 4}
    findings.sort(key=lambda item: (order[item['severity']], item['path'], item['line'], item['rule']))
    return {
        'schema': 'gha-permission-audit-v1',
        'files': files,
        'findings': findings,
        'summary': {
            'workflows_scanned': len(files),
            'findings': len(findings),
            'by_severity': {level: sum(x['severity'] == level for x in findings) for level in order},
        },
        'warnings': warnings,
        'heuristic_only': True,
        'notice': 'Static heuristic review only. Confirm each finding against workflow intent and GitHub documentation.',
    }


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('project', nargs='?', default='.', type=Path)
    parser.add_argument('--format', choices=('json', 'text'), default='text')
    args = parser.parse_args()
    report = audit(args.project)
    if args.format == 'json':
        print(json.dumps(report, ensure_ascii=False, indent=2))
        return
    summary = report.get('summary', {'workflows_scanned': 0, 'findings': 0})
    print(f"Workflows scanned: {summary['workflows_scanned']}")
    print(f"Findings: {summary['findings']}")
    for item in report['findings']:
        print(f"[{item['severity'].upper()}] {item['path']}:{item['line']} {item['rule']}")
        print(f"  {item['message']}")
        print(f"  Suggested review: {item['remediation']}")
    for warning in report['warnings']:
        print(f"[WARNING] {warning}")


if __name__ == '__main__':
    main()