#!/usr/bin/env python3
"""Kubernetes RBAC hardening audit agent.

Audits Kubernetes Role-Based Access Control configuration for security
weaknesses including overly permissive ClusterRoles, wildcard permissions,
privilege escalation paths, and service account misconfigurations.
"""
import argparse
import json
import os
import subprocess
import sys
from datetime import datetime, timezone


def run_kubectl(args_list, timeout=60):
    """Execute kubectl and return parsed JSON."""
    cmd = ["kubectl"] + args_list + ["-o", "json"]
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
    if result.returncode != 0:
        return None
    try:
        return json.loads(result.stdout)
    except json.JSONDecodeError:
        return None


def audit_cluster_roles():
    """Audit ClusterRoles for dangerous permissions."""
    findings = []
    data = run_kubectl(["get", "clusterroles"])
    if not data:
        return findings

    dangerous_verbs = {"*", "create", "update", "patch", "delete"}
    dangerous_resources = {"secrets", "pods/exec", "clusterroles", "clusterrolebindings",
                           "roles", "rolebindings", "serviceaccounts", "nodes"}
    escalation_resources = {"clusterroles", "clusterrolebindings", "roles", "rolebindings"}

    for role in data.get("items", []):
        name = role.get("metadata", {}).get("name", "")
        if name.startswith("system:"):
            continue  # Skip system roles

        for rule in role.get("rules", []):
            verbs = set(rule.get("verbs", []))
            resources = set(rule.get("resources", []))
            api_groups = rule.get("apiGroups", [])

            # Wildcard everything
            if "*" in verbs and "*" in resources:
                findings.append({
                    "type": "wildcard_all", "role": name, "kind": "ClusterRole",
                    "severity": "CRITICAL",
                    "detail": "Full wildcard access (verbs: *, resources: *)",
                    "recommendation": "Replace with specific verbs and resources",
                })

            # Secrets access
            if resources & {"secrets", "*"} and verbs & {"get", "list", "watch", "*"}:
                findings.append({
                    "type": "secrets_access", "role": name, "kind": "ClusterRole",
                    "severity": "HIGH",
                    "detail": f"Can read secrets (verbs: {verbs & {'get', 'list', 'watch', '*'}})",
                })

            # Pod exec
            if "pods/exec" in resources or ("pods" in resources and "create" in verbs):
                findings.append({
                    "type": "pod_exec", "role": name, "kind": "ClusterRole",
                    "severity": "HIGH",
                    "detail": "Can exec into pods (container escape risk)",
                })

            # Privilege escalation via RBAC modification
            if resources & escalation_resources and verbs & {"create", "update", "patch", "*"}:
                findings.append({
                    "type": "rbac_escalation", "role": name, "kind": "ClusterRole",
                    "severity": "CRITICAL",
                    "detail": f"Can modify RBAC resources: {resources & escalation_resources}",
                })

            # Node access
            if "nodes" in resources and verbs & {"get", "list", "proxy", "*"}:
                findings.append({
                    "type": "node_access", "role": name, "kind": "ClusterRole",
                    "severity": "MEDIUM",
                    "detail": "Can access node resources",
                })

    return findings


def audit_cluster_role_bindings():
    """Audit ClusterRoleBindings for overly broad subject assignments."""
    findings = []
    data = run_kubectl(["get", "clusterrolebindings"])
    if not data:
        return findings

    for binding in data.get("items", []):
        name = binding.get("metadata", {}).get("name", "")
        if name.startswith("system:"):
            continue

        role_ref = binding.get("roleRef", {})
        role_name = role_ref.get("name", "")
        subjects = binding.get("subjects", [])

        for subject in subjects:
            kind = subject.get("kind", "")
            subj_name = subject.get("name", "")
            namespace = subject.get("namespace", "")

            # Cluster-admin binding
            if role_name == "cluster-admin":
                findings.append({
                    "type": "cluster_admin_binding",
                    "binding": name, "subject": f"{kind}/{subj_name}",
                    "severity": "CRITICAL",
                    "detail": f"cluster-admin bound to {kind} '{subj_name}'",
                })

            # Group bindings to all authenticated/unauthenticated
            if kind == "Group" and subj_name in ("system:authenticated", "system:unauthenticated"):
                findings.append({
                    "type": "broad_group_binding",
                    "binding": name, "subject": subj_name,
                    "severity": "CRITICAL" if subj_name == "system:unauthenticated" else "HIGH",
                    "detail": f"Role '{role_name}' bound to group '{subj_name}'",
                })

            # Default service account bindings
            if kind == "ServiceAccount" and subj_name == "default":
                findings.append({
                    "type": "default_sa_binding",
                    "binding": name, "subject": f"default SA in {namespace}",
                    "severity": "MEDIUM",
                    "detail": f"Role '{role_name}' bound to default service account",
                })

    return findings


def audit_service_accounts(namespace=None):
    """Audit service accounts for misconfigurations."""
    findings = []
    cmd = ["get", "serviceaccounts", "--all-namespaces"] if not namespace else ["get", "serviceaccounts", "-n", namespace]
    data = run_kubectl(cmd)
    if not data:
        return findings

    for sa in data.get("items", []):
        name = sa.get("metadata", {}).get("name", "")
        ns = sa.get("metadata", {}).get("namespace", "")
        automount = sa.get("automountServiceAccountToken", None)

        if name == "default" and automount is not False:
            findings.append({
                "type": "default_sa_automount",
                "namespace": ns, "service_account": name,
                "severity": "MEDIUM",
                "detail": f"Default SA in '{ns}' has automountServiceAccountToken enabled",
                "recommendation": "Set automountServiceAccountToken: false on default SA",
            })

        secrets = sa.get("secrets", [])
        if len(secrets) > 1:
            findings.append({
                "type": "sa_multiple_secrets",
                "namespace": ns, "service_account": name,
                "severity": "LOW",
                "detail": f"SA has {len(secrets)} token secrets",
            })

    return findings


def format_summary(role_findings, binding_findings, sa_findings):
    """Print RBAC audit summary."""
    all_findings = role_findings + binding_findings + sa_findings
    print(f"\n{'='*60}")
    print(f"  Kubernetes RBAC Hardening Audit")
    print(f"{'='*60}")
    print(f"  ClusterRole Issues    : {len(role_findings)}")
    print(f"  Binding Issues        : {len(binding_findings)}")
    print(f"  ServiceAccount Issues : {len(sa_findings)}")
    print(f"  Total Findings        : {len(all_findings)}")

    severity_counts = {}
    for f in all_findings:
        sev = f.get("severity", "INFO")
        severity_counts[sev] = severity_counts.get(sev, 0) + 1

    print(f"\n  By Severity:")
    for sev in ["CRITICAL", "HIGH", "MEDIUM", "LOW"]:
        count = severity_counts.get(sev, 0)
        if count:
            print(f"    {sev:10s}: {count}")

    if all_findings:
        print(f"\n  Top Findings:")
        for f in sorted(all_findings, key=lambda x: {"CRITICAL": 0, "HIGH": 1, "MEDIUM": 2}.get(x["severity"], 9))[:15]:
            print(f"    [{f['severity']:8s}] {f['type']:25s} | {f.get('detail', '')[:50]}")

    return severity_counts


def main():
    parser = argparse.ArgumentParser(description="Kubernetes RBAC hardening audit agent")
    parser.add_argument("--namespace", "-n", help="Specific namespace to audit")
    parser.add_argument("--kubeconfig", help="Path to kubeconfig")
    parser.add_argument("--skip-roles", action="store_true")
    parser.add_argument("--skip-bindings", action="store_true")
    parser.add_argument("--skip-sa", action="store_true")
    parser.add_argument("--output", "-o", help="Output JSON report")
    parser.add_argument("--verbose", "-v", action="store_true")
    args = parser.parse_args()

    if args.kubeconfig:
        os.environ["KUBECONFIG"] = args.kubeconfig

    role_findings = [] if args.skip_roles else audit_cluster_roles()
    binding_findings = [] if args.skip_bindings else audit_cluster_role_bindings()
    sa_findings = [] if args.skip_sa else audit_service_accounts(args.namespace)

    severity_counts = format_summary(role_findings, binding_findings, sa_findings)

    report = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "tool": "K8s RBAC Audit",
        "role_findings": role_findings,
        "binding_findings": binding_findings,
        "sa_findings": sa_findings,
        "severity_counts": severity_counts,
        "risk_level": (
            "CRITICAL" if severity_counts.get("CRITICAL", 0) > 0
            else "HIGH" if severity_counts.get("HIGH", 0) > 0
            else "MEDIUM" if severity_counts.get("MEDIUM", 0) > 0
            else "LOW"
        ),
    }

    if args.output:
        with open(args.output, "w") as f:
            json.dump(report, f, indent=2)
        print(f"\n[+] Report saved to {args.output}")
    elif args.verbose:
        print(json.dumps(report, indent=2))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Kubernetes RBAC Audit and Hardening Tool

Audits RBAC configurations to identify overprivileged accounts,
cluster-admin sprawl, default service account usage, and
generates hardening recommendations.
"""

import json
import subprocess
import sys
import argparse
from datetime import datetime
from collections import defaultdict


DANGEROUS_VERBS = {"*", "create", "update", "patch", "delete"}
DANGEROUS_RESOURCES = {
    "secrets", "pods/exec", "clusterroles", "clusterrolebindings",
    "roles", "rolebindings", "serviceaccounts/token", "nodes/proxy"
}
DANGEROUS_API_GROUPS = {"*"}


def run_kubectl(args: list[str]) -> str:
    try:
        result = subprocess.run(
            ["kubectl"] + args, capture_output=True, text=True, timeout=30
        )
        return result.stdout.strip()
    except (subprocess.TimeoutExpired, FileNotFoundError):
        return ""


def get_cluster_role_bindings() -> list[dict]:
    output = run_kubectl(["get", "clusterrolebindings", "-o", "json"])
    if not output:
        return []
    try:
        return json.loads(output).get("items", [])
    except json.JSONDecodeError:
        return []


def get_role_bindings() -> list[dict]:
    output = run_kubectl(["get", "rolebindings", "--all-namespaces", "-o", "json"])
    if not output:
        return []
    try:
        return json.loads(output).get("items", [])
    except json.JSONDecodeError:
        return []


def get_cluster_roles() -> dict:
    output = run_kubectl(["get", "clusterroles", "-o", "json"])
    if not output:
        return {}
    try:
        items = json.loads(output).get("items", [])
        return {item["metadata"]["name"]: item.get("rules", []) for item in items}
    except json.JSONDecodeError:
        return {}


def audit_cluster_admin_bindings(crbs: list[dict]) -> list[dict]:
    findings = []
    for crb in crbs:
        if crb.get("roleRef", {}).get("name") == "cluster-admin":
            for subject in crb.get("subjects", []):
                findings.append({
                    "severity": "CRITICAL",
                    "type": "cluster_admin_binding",
                    "binding": crb["metadata"]["name"],
                    "subject_kind": subject.get("kind", ""),
                    "subject_name": subject.get("name", ""),
                    "subject_namespace": subject.get("namespace", ""),
                    "description": f"cluster-admin bound to {subject.get('kind', '')}/{subject.get('name', '')}"
                })
    return findings


def audit_wildcard_permissions(roles: dict) -> list[dict]:
    findings = []
    for role_name, rules in roles.items():
        for rule in rules:
            verbs = rule.get("verbs", [])
            resources = rule.get("resources", [])
            api_groups = rule.get("apiGroups", [])

            if "*" in verbs and "*" in resources:
                findings.append({
                    "severity": "HIGH",
                    "type": "wildcard_permissions",
                    "role": role_name,
                    "description": f"ClusterRole {role_name} has wildcard verbs and resources"
                })
            elif "*" in verbs:
                findings.append({
                    "severity": "MEDIUM",
                    "type": "wildcard_verbs",
                    "role": role_name,
                    "resources": resources,
                    "description": f"ClusterRole {role_name} has wildcard verbs on {resources}"
                })
    return findings


def audit_dangerous_permissions(roles: dict) -> list[dict]:
    findings = []
    for role_name, rules in roles.items():
        for rule in rules:
            verbs = set(rule.get("verbs", []))
            resources = set(rule.get("resources", []))
            dangerous_matches = resources.intersection(DANGEROUS_RESOURCES)
            has_dangerous_verbs = verbs.intersection(DANGEROUS_VERBS)

            if dangerous_matches and has_dangerous_verbs:
                findings.append({
                    "severity": "HIGH",
                    "type": "dangerous_permission",
                    "role": role_name,
                    "resources": list(dangerous_matches),
                    "verbs": list(has_dangerous_verbs),
                    "description": f"ClusterRole {role_name} grants {list(has_dangerous_verbs)} on {list(dangerous_matches)}"
                })
    return findings


def audit_default_service_accounts(rbs: list[dict], crbs: list[dict]) -> list[dict]:
    findings = []
    for binding in rbs + crbs:
        for subject in binding.get("subjects", []):
            if subject.get("kind") == "ServiceAccount" and subject.get("name") == "default":
                findings.append({
                    "severity": "MEDIUM",
                    "type": "default_sa_binding",
                    "binding": binding["metadata"]["name"],
                    "namespace": subject.get("namespace", "N/A"),
                    "role": binding.get("roleRef", {}).get("name", ""),
                    "description": f"Default service account in {subject.get('namespace', 'N/A')} has role binding"
                })
    return findings


def generate_report(all_findings: list[dict], output_format: str = "text") -> str:
    critical = [f for f in all_findings if f["severity"] == "CRITICAL"]
    high = [f for f in all_findings if f["severity"] == "HIGH"]
    medium = [f for f in all_findings if f["severity"] == "MEDIUM"]

    if output_format == "json":
        return json.dumps({
            "timestamp": datetime.utcnow().isoformat(),
            "summary": {"critical": len(critical), "high": len(high), "medium": len(medium)},
            "findings": all_findings
        }, indent=2)

    lines = ["=" * 70, "KUBERNETES RBAC HARDENING AUDIT REPORT",
             f"Generated: {datetime.utcnow().isoformat()}", "=" * 70]
    lines.append(f"\nFindings: {len(critical)} Critical, {len(high)} High, {len(medium)} Medium")

    for sev, items in [("CRITICAL", critical), ("HIGH", high), ("MEDIUM", medium)]:
        if items:
            lines.append(f"\n## {sev}")
            for f in items:
                lines.append(f"  [{f['type']}] {f['description']}")

    lines.append("\n" + "=" * 70)
    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(description="Kubernetes RBAC Audit Tool")
    parser.add_argument("--format", choices=["text", "json"], default="text")
    args = parser.parse_args()

    crbs = get_cluster_role_bindings()
    rbs = get_role_bindings()
    roles = get_cluster_roles()

    all_findings = []
    all_findings.extend(audit_cluster_admin_bindings(crbs))
    all_findings.extend(audit_wildcard_permissions(roles))
    all_findings.extend(audit_dangerous_permissions(roles))
    all_findings.extend(audit_default_service_accounts(rbs, crbs))

    print(generate_report(all_findings, args.format))
    sys.exit(1 if any(f["severity"] == "CRITICAL" for f in all_findings) else 0)


if __name__ == "__main__":
    main()