#!/usr/bin/env python3
"""Kubernetes Network Policy Agent - audits pod-to-pod communication and network policy coverage."""

import json
import argparse
import logging
import subprocess
from datetime import datetime

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)


def kubectl_json(args_list):
    """Execute kubectl command and return JSON output."""
    cmd = ["kubectl"] + args_list + ["-o", "json"]
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
    return json.loads(result.stdout) if result.returncode == 0 else {}


def get_namespaces():
    return kubectl_json(["get", "namespaces"])


def get_network_policies(namespace="--all-namespaces"):
    if namespace == "--all-namespaces":
        return kubectl_json(["get", "networkpolicies", "--all-namespaces"])
    return kubectl_json(["get", "networkpolicies", "-n", namespace])


def get_pods(namespace="--all-namespaces"):
    if namespace == "--all-namespaces":
        return kubectl_json(["get", "pods", "--all-namespaces"])
    return kubectl_json(["get", "pods", "-n", namespace])


def analyze_policy_coverage(policies, pods):
    """Determine which pods are covered by network policies."""
    policy_selectors = []
    for item in policies.get("items", []):
        ns = item.get("metadata", {}).get("namespace", "")
        spec = item.get("spec", {})
        selector = spec.get("podSelector", {}).get("matchLabels", {})
        policy_types = spec.get("policyTypes", [])
        policy_selectors.append({
            "namespace": ns,
            "selector": selector,
            "ingress": "Ingress" in policy_types or spec.get("ingress") is not None,
            "egress": "Egress" in policy_types or spec.get("egress") is not None,
        })
    covered_pods = set()
    uncovered_pods = []
    for pod in pods.get("items", []):
        pod_ns = pod.get("metadata", {}).get("namespace", "")
        pod_name = pod.get("metadata", {}).get("name", "")
        pod_labels = pod.get("metadata", {}).get("labels", {})
        is_covered = False
        for policy in policy_selectors:
            if policy["namespace"] != pod_ns:
                continue
            if not policy["selector"]:
                is_covered = True
                break
            if all(pod_labels.get(k) == v for k, v in policy["selector"].items()):
                is_covered = True
                break
        if is_covered:
            covered_pods.add(f"{pod_ns}/{pod_name}")
        else:
            uncovered_pods.append({"namespace": pod_ns, "pod": pod_name, "labels": pod_labels})
    total = len(pods.get("items", []))
    return {
        "total_pods": total,
        "covered_pods": len(covered_pods),
        "uncovered_pods": len(uncovered_pods),
        "coverage_percent": round(len(covered_pods) / max(total, 1) * 100, 1),
        "uncovered_pod_list": uncovered_pods[:20],
    }


def detect_overly_permissive_policies(policies):
    """Find network policies that allow all traffic."""
    findings = []
    for item in policies.get("items", []):
        name = item.get("metadata", {}).get("name", "")
        ns = item.get("metadata", {}).get("namespace", "")
        spec = item.get("spec", {})
        if not spec.get("podSelector", {}).get("matchLabels"):
            ingress_rules = spec.get("ingress", [])
            for rule in ingress_rules:
                if not rule.get("from"):
                    findings.append({"policy": name, "namespace": ns,
                                   "issue": "allows_all_ingress", "severity": "high"})
            egress_rules = spec.get("egress", [])
            for rule in egress_rules:
                if not rule.get("to"):
                    findings.append({"policy": name, "namespace": ns,
                                   "issue": "allows_all_egress", "severity": "medium"})
    return findings


def analyze_namespace_isolation(policies, namespaces):
    """Check which namespaces have default-deny policies."""
    ns_with_deny = set()
    for item in policies.get("items", []):
        spec = item.get("spec", {})
        if (not spec.get("podSelector", {}).get("matchLabels") and
            not spec.get("ingress") and "Ingress" in spec.get("policyTypes", [])):
            ns_with_deny.add(item.get("metadata", {}).get("namespace", ""))
    all_ns = [ns.get("metadata", {}).get("name", "") for ns in namespaces.get("items", [])]
    system_ns = {"kube-system", "kube-public", "kube-node-lease"}
    user_ns = [ns for ns in all_ns if ns not in system_ns]
    return {
        "total_namespaces": len(user_ns),
        "namespaces_with_default_deny": len(ns_with_deny),
        "isolation_percent": round(len(ns_with_deny) / max(len(user_ns), 1) * 100, 1),
        "unprotected_namespaces": [ns for ns in user_ns if ns not in ns_with_deny],
    }


def generate_report(policies, pods, namespaces):
    coverage = analyze_policy_coverage(policies, pods)
    permissive = detect_overly_permissive_policies(policies)
    isolation = analyze_namespace_isolation(policies, namespaces)
    report = {
        "timestamp": datetime.utcnow().isoformat(),
        "total_network_policies": len(policies.get("items", [])),
        "pod_coverage": coverage,
        "namespace_isolation": isolation,
        "overly_permissive_policies": permissive,
        "total_findings": len(permissive),
    }
    return report


def main():
    parser = argparse.ArgumentParser(description="Kubernetes Network Policy Audit Agent")
    parser.add_argument("--namespace", default="--all-namespaces", help="Namespace to audit")
    parser.add_argument("--output", default="k8s_netpol_report.json")
    args = parser.parse_args()

    policies = get_network_policies(args.namespace)
    pods = get_pods(args.namespace)
    namespaces = get_namespaces()
    report = generate_report(policies, pods, namespaces)
    with open(args.output, "w") as f:
        json.dump(report, f, indent=2, default=str)
    logger.info("K8s NetPol: %.1f%% pod coverage, %.1f%% namespace isolation, %d findings",
                report["pod_coverage"]["coverage_percent"],
                report["namespace_isolation"]["isolation_percent"],
                report["total_findings"])
    print(json.dumps(report, indent=2, default=str))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Kubernetes Network Policy Auditor

Checks for missing network policies, default-deny enforcement,
and identifies namespaces without proper segmentation.
"""

import subprocess
import json
import sys
from dataclasses import dataclass, field


@dataclass
class NetPolFinding:
    namespace: str
    severity: str
    issue: str
    remediation: str


def run_kubectl_json(args: list):
    cmd = ["kubectl"] + args + ["-o", "json"]
    try:
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
        if result.returncode != 0:
            return None
        return json.loads(result.stdout)
    except (subprocess.TimeoutExpired, json.JSONDecodeError, FileNotFoundError):
        return None


def audit_network_policies():
    findings = []
    system_ns = {"kube-system", "kube-public", "kube-node-lease"}

    namespaces = run_kubectl_json(["get", "namespaces"])
    if not namespaces:
        print("[!] Cannot list namespaces")
        return findings

    for ns in namespaces.get("items", []):
        ns_name = ns["metadata"]["name"]
        if ns_name in system_ns:
            continue

        netpols = run_kubectl_json(["get", "networkpolicies", "-n", ns_name])
        policies = netpols.get("items", []) if netpols else []

        if not policies:
            findings.append(NetPolFinding(
                namespace=ns_name, severity="HIGH",
                issue="No NetworkPolicies defined",
                remediation=f"Create default-deny ingress/egress policy in namespace '{ns_name}'"
            ))
            continue

        # Check for default-deny
        has_default_deny_ingress = False
        has_default_deny_egress = False

        for pol in policies:
            spec = pol.get("spec", {})
            pod_selector = spec.get("podSelector", {})
            policy_types = spec.get("policyTypes", [])

            if not pod_selector.get("matchLabels") and not pod_selector.get("matchExpressions"):
                if "Ingress" in policy_types and not spec.get("ingress"):
                    has_default_deny_ingress = True
                if "Egress" in policy_types and not spec.get("egress"):
                    has_default_deny_egress = True

        if not has_default_deny_ingress:
            findings.append(NetPolFinding(
                namespace=ns_name, severity="HIGH",
                issue="Missing default-deny ingress policy",
                remediation="Create NetworkPolicy with empty podSelector and Ingress policyType with no ingress rules"
            ))

        if not has_default_deny_egress:
            findings.append(NetPolFinding(
                namespace=ns_name, severity="MEDIUM",
                issue="Missing default-deny egress policy",
                remediation="Create NetworkPolicy with empty podSelector and Egress policyType with no egress rules"
            ))

    return findings


def main():
    print("[*] Kubernetes Network Policy Auditor\n")
    findings = audit_network_policies()

    print(f"\n{'='*60}")
    print(f"NETWORK POLICY AUDIT REPORT")
    print(f"{'='*60}")
    print(f"Total Findings: {len(findings)}")

    for sev in ["CRITICAL", "HIGH", "MEDIUM", "LOW"]:
        sev_findings = [f for f in findings if f.severity == sev]
        if sev_findings:
            print(f"\n{sev}:")
            for f in sev_findings:
                print(f"  [{f.namespace}] {f.issue}")
                print(f"    Fix: {f.remediation}")

    with open("netpol_audit_report.json", "w") as fh:
        json.dump({"findings": [{"namespace": f.namespace, "severity": f.severity,
                                  "issue": f.issue, "remediation": f.remediation}
                                 for f in findings]}, fh, indent=2)
    print("\n[*] Report saved to netpol_audit_report.json")

    if any(f.severity in ("CRITICAL", "HIGH") for f in findings):
        sys.exit(1)


if __name__ == "__main__":
    main()