A security issue was discovered with Kubernetes that could enable users to send network traffic to…

#!/usr/bin/env python3
# CVE-2021-25740 exposure check for Kubernetes
# Checks for non-system subjects bound to roles that can write Endpoints or EndpointSlices.
# Output: VULNERABLE / PATCHED / UNKNOWN
# Exit codes: 1 vulnerable, 0 patched, 2 unknown

import json
import subprocess
import sys
from typing import Dict, List, Tuple

WRITE_VERBS = {"create", "update", "patch", "delete", "deletecollection", "*"}
TARGETS = [
    ("", "endpoints"),
    ("discovery.k8s.io", "endpointslices"),
]


def run_kubectl(args: List[str]) -> dict:
    cmd = ["kubectl"] + args + ["-o", "json"]
    try:
        p = subprocess.run(cmd, capture_output=True, text=True, check=False)
    except FileNotFoundError:
        print("UNKNOWN: kubectl not found")
        sys.exit(2)
    if p.returncode != 0:
        print(f"UNKNOWN: kubectl failed: {' '.join(cmd)} :: {p.stderr.strip()}")
        sys.exit(2)
    try:
        return json.loads(p.stdout)
    except json.JSONDecodeError:
        print(f"UNKNOWN: invalid JSON from kubectl: {' '.join(cmd)}")
        sys.exit(2)


def rule_is_risky(rule: dict) -> bool:
    api_groups = set(rule.get("apiGroups", []))
    resources = set(rule.get("resources", []))
    verbs = set(rule.get("verbs", []))
    for api_group, resource in TARGETS:
        if (api_group in api_groups or "*" in api_groups) and (resource in resources or "*" in resources):
            if WRITE_VERBS & verbs:
                return True
    return False


def collect_risky_roles(clusterroles: dict, roles: dict) -> Tuple[Dict[str, dict], Dict[Tuple[str, str], dict]]:
    risky_clusterroles = {}
    risky_roles = {}

    for item in clusterroles.get("items", []):
        if any(rule_is_risky(r) for r in item.get("rules", [])):
            risky_clusterroles[item["metadata"]["name"]] = item

    for item in roles.get("items", []):
        if any(rule_is_risky(r) for r in item.get("rules", [])):
            ns = item["metadata"]["namespace"]
            name = item["metadata"]["name"]
            risky_roles[(ns, name)] = item

    return risky_clusterroles, risky_roles


def subject_is_non_system(subject: dict) -> bool:
    kind = subject.get("kind", "")
    name = subject.get("name", "")
    namespace = subject.get("namespace", "")

    if kind == "Group" and name.startswith("system:"):
        return False
    if kind == "User" and name.startswith("system:"):
        return False
    if kind == "ServiceAccount" and namespace == "kube-system":
        return False
    return True


def format_subject(subject: dict) -> str:
    kind = subject.get("kind", "?")
    name = subject.get("name", "?")
    ns = subject.get("namespace")
    return f"{kind}:{ns + '/' if ns else ''}{name}"


def main() -> int:
    # Basic connectivity test
    version = subprocess.run(["kubectl", "version", "--request-timeout=10s"], capture_output=True, text=True)
    if version.returncode != 0:
        print(f"UNKNOWN: cannot talk to cluster: {version.stderr.strip()}")
        return 2

    clusterroles = run_kubectl(["get", "clusterroles"])
    roles = run_kubectl(["get", "roles", "-A"])
    crbs = run_kubectl(["get", "clusterrolebindings"])
    rbs = run_kubectl(["get", "rolebindings", "-A"])

    risky_clusterroles, risky_roles = collect_risky_roles(clusterroles, roles)

    findings = []

    for item in crbs.get("items", []):
        ref = item.get("roleRef", {})
        if ref.get("kind") == "ClusterRole" and ref.get("name") in risky_clusterroles:
            for subj in item.get("subjects", []):
                if subject_is_non_system(subj):
                    findings.append({
                        "binding": f"ClusterRoleBinding/{item['metadata']['name']}",
                        "role": f"ClusterRole/{ref.get('name')}",
                        "subject": format_subject(subj),
                    })

    for item in rbs.get("items", []):
        ns = item["metadata"].get("namespace", "")
        ref = item.get("roleRef", {})
        risky = False
        role_label = ""
        if ref.get("kind") == "ClusterRole" and ref.get("name") in risky_clusterroles:
            risky = True
            role_label = f"ClusterRole/{ref.get('name')}"
        elif ref.get("kind") == "Role" and (ns, ref.get("name")) in risky_roles:
            risky = True
            role_label = f"Role/{ns}/{ref.get('name')}"
        if risky:
            for subj in item.get("subjects", []):
                if subject_is_non_system(subj):
                    findings.append({
                        "binding": f"RoleBinding/{ns}/{item['metadata']['name']}",
                        "role": role_label,
                        "subject": format_subject(subj),
                    })

    if findings:
        print("VULNERABLE: non-system subjects can write Endpoints or EndpointSlices")
        for f in findings[:25]:
            print(f"- {f['subject']} via {f['binding']} -> {f['role']}")
        if len(findings) > 25:
            print(f"- ... {len(findings) - 25} additional risky bindings omitted")
        return 1

    print("PATCHED: no non-system subject bindings to Roles/ClusterRoles with endpoint-object write permissions were found")
    return 0


if __name__ == "__main__":
    sys.exit(main())