#!/usr/bin/env python3
"""Agent for managing GitHub Advanced Security code scanning with CodeQL."""

import json
import argparse
import subprocess
from datetime import datetime
from collections import Counter


def gh_api(endpoint, method="GET"):
    """Call GitHub API via gh CLI."""
    cmd = ["gh", "api", endpoint, "--method", method]
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
    if result.returncode != 0:
        return {"error": result.stderr.strip()}
    try:
        return json.loads(result.stdout) if result.stdout.strip() else {}
    except json.JSONDecodeError:
        return {"raw": result.stdout.strip()}


def get_code_scanning_alerts(owner, repo, state="open"):
    """Get code scanning alerts for a repository."""
    alerts = gh_api(f"/repos/{owner}/{repo}/code-scanning/alerts?state={state}&per_page=100")
    if isinstance(alerts, dict) and "error" in alerts:
        return alerts
    return alerts if isinstance(alerts, list) else []


def get_secret_scanning_alerts(owner, repo, state="open"):
    """Get secret scanning alerts for a repository."""
    alerts = gh_api(f"/repos/{owner}/{repo}/secret-scanning/alerts?state={state}&per_page=100")
    if isinstance(alerts, dict) and "error" in alerts:
        return alerts
    return alerts if isinstance(alerts, list) else []


def get_dependabot_alerts(owner, repo, state="open"):
    """Get Dependabot alerts for a repository."""
    alerts = gh_api(f"/repos/{owner}/{repo}/dependabot/alerts?state={state}&per_page=100")
    if isinstance(alerts, dict) and "error" in alerts:
        return alerts
    return alerts if isinstance(alerts, list) else []


def analyze_code_scanning_alerts(alerts):
    """Analyze code scanning alerts and produce summary."""
    if not isinstance(alerts, list):
        return {"error": "No alerts data"}
    by_severity = Counter()
    by_rule = Counter()
    by_tool = Counter()
    critical_alerts = []
    for alert in alerts:
        rule = alert.get("rule", {})
        severity = rule.get("security_severity_level", rule.get("severity", "unknown"))
        by_severity[severity] += 1
        by_rule[rule.get("id", "unknown")] += 1
        tool = alert.get("tool", {}).get("name", "unknown")
        by_tool[tool] += 1
        if severity in ("critical", "high"):
            critical_alerts.append({
                "number": alert.get("number"),
                "rule": rule.get("id", ""),
                "description": rule.get("description", "")[:120],
                "severity": severity,
                "state": alert.get("state", ""),
                "created_at": alert.get("created_at", ""),
                "html_url": alert.get("html_url", ""),
            })
    return {
        "total_alerts": len(alerts),
        "by_severity": dict(by_severity),
        "by_rule": dict(by_rule.most_common(10)),
        "by_tool": dict(by_tool),
        "critical_and_high": critical_alerts[:20],
    }


def analyze_secret_alerts(alerts):
    """Analyze secret scanning alerts."""
    if not isinstance(alerts, list):
        return {"error": "No alerts data"}
    by_type = Counter()
    for alert in alerts:
        by_type[alert.get("secret_type_display_name", alert.get("secret_type", "unknown"))] += 1
    return {
        "total_secrets": len(alerts),
        "by_type": dict(by_type),
        "alerts": [
            {"number": a.get("number"), "type": a.get("secret_type_display_name", ""),
             "state": a.get("state", ""), "created_at": a.get("created_at", "")}
            for a in alerts[:20]
        ],
    }


def analyze_dependabot_alerts(alerts):
    """Analyze Dependabot vulnerability alerts."""
    if not isinstance(alerts, list):
        return {"error": "No alerts data"}
    by_severity = Counter()
    by_ecosystem = Counter()
    for alert in alerts:
        vuln = alert.get("security_vulnerability", alert.get("security_advisory", {}))
        severity = vuln.get("severity", alert.get("severity", "unknown"))
        by_severity[severity] += 1
        dep = alert.get("dependency", {})
        pkg = dep.get("package", {})
        by_ecosystem[pkg.get("ecosystem", "unknown")] += 1
    return {
        "total_alerts": len(alerts),
        "by_severity": dict(by_severity),
        "by_ecosystem": dict(by_ecosystem),
    }


def generate_codeql_workflow(languages, query_suite="security-extended"):
    """Generate a CodeQL analysis GitHub Actions workflow."""
    lang_list = ", ".join(f"'{l}'" for l in languages)
    return f"""name: CodeQL Analysis
on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main]
  schedule:
    - cron: '30 2 * * 1'
jobs:
  analyze:
    name: Analyze (${{{{ matrix.language }}}})
    runs-on: ubuntu-latest
    permissions:
      security-events: write
      contents: read
      actions: read
    strategy:
      fail-fast: false
      matrix:
        language: [{lang_list}]
    steps:
      - uses: actions/checkout@v4
      - uses: github/codeql-action/init@v3
        with:
          languages: ${{{{ matrix.language }}}}
          queries: +{query_suite}
      - uses: github/codeql-action/autobuild@v3
      - uses: github/codeql-action/analyze@v3
        with:
          category: "/language:${{{{ matrix.language }}}}"
"""


def full_security_audit(owner, repo):
    """Run full GHAS security audit for a repository."""
    code_alerts = get_code_scanning_alerts(owner, repo)
    secret_alerts = get_secret_scanning_alerts(owner, repo)
    dependabot_alerts = get_dependabot_alerts(owner, repo)
    return {
        "code_scanning": analyze_code_scanning_alerts(code_alerts),
        "secret_scanning": analyze_secret_alerts(secret_alerts),
        "dependabot": analyze_dependabot_alerts(dependabot_alerts),
    }


def main():
    parser = argparse.ArgumentParser(description="GitHub Advanced Security Agent")
    parser.add_argument("--owner", help="Repository owner")
    parser.add_argument("--repo", help="Repository name")
    parser.add_argument("--action", choices=["audit", "code-alerts", "secrets",
                                              "dependabot", "gen-workflow"],
                        default="audit")
    parser.add_argument("--languages", nargs="+", default=["python", "javascript-typescript"])
    parser.add_argument("--output", default="ghas_report.json")
    args = parser.parse_args()

    report = {"generated_at": datetime.utcnow().isoformat(), "results": {}}

    if args.action == "audit" and args.owner and args.repo:
        results = full_security_audit(args.owner, args.repo)
        report["results"] = results
        cs = results["code_scanning"]
        print(f"[+] Code scanning: {cs.get('total_alerts', 0)} alerts")
        print(f"[+] Secrets: {results['secret_scanning'].get('total_secrets', 0)}")
        print(f"[+] Dependabot: {results['dependabot'].get('total_alerts', 0)}")

    elif args.action == "code-alerts" and args.owner and args.repo:
        alerts = get_code_scanning_alerts(args.owner, args.repo)
        analysis = analyze_code_scanning_alerts(alerts)
        report["results"]["code_scanning"] = analysis
        print(f"[+] {analysis.get('total_alerts', 0)} code scanning alerts")

    elif args.action == "secrets" and args.owner and args.repo:
        alerts = get_secret_scanning_alerts(args.owner, args.repo)
        analysis = analyze_secret_alerts(alerts)
        report["results"]["secret_scanning"] = analysis
        print(f"[+] {analysis.get('total_secrets', 0)} secret alerts")

    elif args.action == "gen-workflow":
        workflow = generate_codeql_workflow(args.languages)
        report["results"]["workflow"] = workflow
        print("[+] CodeQL workflow generated")

    with open(args.output, "w") as f:
        json.dump(report, f, indent=2, default=str)
    print(f"[+] Report saved to {args.output}")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
GitHub Advanced Security Code Scanning Alert Management

Uses the GitHub REST API to query, triage, and report on CodeQL
code scanning alerts across an organization's repositories.
"""

import json
import os
import sys
import urllib.request
import urllib.error
from datetime import datetime, timedelta
from collections import defaultdict


def get_github_headers(token: str) -> dict:
    return {
        "Authorization": f"Bearer {token}",
        "Accept": "application/vnd.github+json",
        "X-GitHub-Api-Version": "2022-11-28",
    }


def github_api_get(url: str, token: str) -> list | dict:
    headers = get_github_headers(token)
    results = []
    page = 1
    while True:
        paginated_url = f"{url}{'&' if '?' in url else '?'}per_page=100&page={page}"
        req = urllib.request.Request(paginated_url, headers=headers)
        try:
            with urllib.request.urlopen(req) as resp:
                data = json.loads(resp.read().decode())
                if isinstance(data, list):
                    if not data:
                        break
                    results.extend(data)
                    page += 1
                else:
                    return data
        except urllib.error.HTTPError as e:
            print(f"HTTP {e.code} for {paginated_url}: {e.read().decode()}")
            break
    return results


def list_org_repos(org: str, token: str) -> list:
    url = f"https://api.github.com/orgs/{org}/repos?type=all"
    repos = github_api_get(url, token)
    return [r["full_name"] for r in repos if isinstance(r, dict)]


def get_code_scanning_alerts(repo: str, token: str, state: str = "open") -> list:
    url = f"https://api.github.com/repos/{repo}/code-scanning/alerts?state={state}"
    return github_api_get(url, token)


def categorize_alerts(alerts: list) -> dict:
    categories = defaultdict(lambda: defaultdict(int))
    for alert in alerts:
        rule = alert.get("rule", {})
        severity = rule.get("security_severity_level", "unknown")
        cwe_tags = [t for t in rule.get("tags", []) if t.startswith("cwe-")]
        tool_name = alert.get("tool", {}).get("name", "unknown")
        categories["by_severity"][severity] += 1
        categories["by_tool"][tool_name] += 1
        for cwe in cwe_tags:
            categories["by_cwe"][cwe] += 1
    return dict(categories)


def calculate_mttr(alerts: list) -> dict:
    resolved = [a for a in alerts if a.get("state") == "fixed"]
    if not resolved:
        return {"total_resolved": 0, "avg_mttr_hours": None}

    durations = []
    for alert in resolved:
        created = datetime.fromisoformat(alert["created_at"].replace("Z", "+00:00"))
        fixed = datetime.fromisoformat(
            alert.get("fixed_at", alert.get("dismissed_at", "")).replace("Z", "+00:00")
        )
        durations.append((fixed - created).total_seconds() / 3600)

    return {
        "total_resolved": len(resolved),
        "avg_mttr_hours": round(sum(durations) / len(durations), 1),
        "min_mttr_hours": round(min(durations), 1),
        "max_mttr_hours": round(max(durations), 1),
    }


def generate_org_report(org: str, token: str) -> dict:
    repos = list_org_repos(org, token)
    report = {
        "organization": org,
        "generated_at": datetime.utcnow().isoformat() + "Z",
        "total_repositories": len(repos),
        "repositories_with_scanning": 0,
        "total_open_alerts": 0,
        "severity_summary": defaultdict(int),
        "top_cwes": defaultdict(int),
        "repo_details": [],
    }

    for repo in repos:
        alerts = get_code_scanning_alerts(repo, token, state="open")
        if not alerts:
            continue

        report["repositories_with_scanning"] += 1
        report["total_open_alerts"] += len(alerts)
        categories = categorize_alerts(alerts)

        for sev, count in categories.get("by_severity", {}).items():
            report["severity_summary"][sev] += count
        for cwe, count in categories.get("by_cwe", {}).items():
            report["top_cwes"][cwe] += count

        closed_alerts = get_code_scanning_alerts(repo, token, state="fixed")
        mttr = calculate_mttr(closed_alerts)

        report["repo_details"].append(
            {
                "repository": repo,
                "open_alerts": len(alerts),
                "severity_breakdown": dict(categories.get("by_severity", {})),
                "mttr": mttr,
            }
        )

    report["severity_summary"] = dict(report["severity_summary"])
    top_cwes_sorted = sorted(report["top_cwes"].items(), key=lambda x: x[1], reverse=True)[:10]
    report["top_cwes"] = dict(top_cwes_sorted)
    return report


def print_report(report: dict) -> None:
    print(f"\n{'='*60}")
    print(f"GHAS Code Scanning Report: {report['organization']}")
    print(f"Generated: {report['generated_at']}")
    print(f"{'='*60}")
    print(f"Total repositories: {report['total_repositories']}")
    print(f"Repositories with scanning enabled: {report['repositories_with_scanning']}")
    coverage = (
        report["repositories_with_scanning"] / report["total_repositories"] * 100
        if report["total_repositories"] > 0
        else 0
    )
    print(f"Coverage: {coverage:.1f}%")
    print(f"Total open alerts: {report['total_open_alerts']}")
    print(f"\nSeverity Summary:")
    for sev in ["critical", "high", "medium", "low", "unknown"]:
        count = report["severity_summary"].get(sev, 0)
        if count > 0:
            print(f"  {sev.upper():12s}: {count}")
    print(f"\nTop CWEs:")
    for cwe, count in report.get("top_cwes", {}).items():
        print(f"  {cwe:15s}: {count}")
    print(f"\nRepository Details:")
    for repo in sorted(report["repo_details"], key=lambda r: r["open_alerts"], reverse=True):
        mttr_str = (
            f"{repo['mttr']['avg_mttr_hours']}h" if repo["mttr"]["avg_mttr_hours"] else "N/A"
        )
        print(f"  {repo['repository']:40s} | Open: {repo['open_alerts']:4d} | Avg MTTR: {mttr_str}")


def main():
    token = os.environ.get("GITHUB_TOKEN")
    if not token:
        print("Error: GITHUB_TOKEN environment variable is required")
        sys.exit(1)

    org = os.environ.get("GITHUB_ORG")
    if not org:
        print("Error: GITHUB_ORG environment variable is required")
        sys.exit(1)

    print(f"Fetching code scanning data for organization: {org}")
    report = generate_org_report(org, token)
    print_report(report)

    output_file = f"ghas_report_{org}_{datetime.utcnow().strftime('%Y%m%d')}.json"
    with open(output_file, "w") as f:
        json.dump(report, f, indent=2, default=str)
    print(f"\nDetailed report saved to: {output_file}")


if __name__ == "__main__":
    main()