diff --git a/CHANGELOG.md b/CHANGELOG.md
index ad685417..9aeccad6 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,6 +9,7 @@ OpenShield uses [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ### Added
 
+- Fifteen enterprise AKS and Kubernetes workload rules covering API restrictions, network policy, Defender, secrets, workload isolation, RBAC, and image trust
 - Azure Network Layer Assurance API with 20-domain coverage, network-rule classification, and authoritative IP forwarding and direct Internet route checks
 - Azure Resource Graph inventory snapshots as the first OpenShield Evidence Graph foundation
 - Azure Data Link Layer Assurance API with LLC and MAC coverage plus ExpressRoute Direct MACsec checks
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 9233250a..6897f530 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -176,7 +176,7 @@ Use the format: `AZ-[CATEGORY]-[NUMBER]`
 | Database | DB | AZ-DB-001 |
 | Compute | CMP | AZ-CMP-001 |
 | Key Vault | KV | AZ-KV-001 |
-| Kubernetes (AKS) | AKS | AZ-AKS-001 |
+| Kubernetes (AKS) | AKS | AZ-AKS-001 to AZ-AKS-021 |
 
 Check existing rules before picking a number to avoid clashes.
 
@@ -201,6 +201,7 @@ Use the existing wrapper methods in `scanner/azure_client.py` rather than constr
 | `azure_client.get_sql_server_auditing_policy(resource_group, server_name)` | ServerBlobAuditingPolicy or None |
 | `azure_client.get_key_vaults()` | List of Key Vault objects |
 | `azure_client.get_managed_clusters()` | List of AKS ManagedCluster objects, or `None` on API failure |
+| `azure_client.get_aks_security_posture()` | Cached ARM, Defender, and Kubernetes evidence, or `None` on AKS inventory failure |
 | `azure_client.get_applications()` | Paginated App Registration dictionaries, or `None` on Graph failure |
 | `azure_client.get_managed_identity_service_principals()` | Managed Identity service principals, or `None` on Graph failure |
 | `azure_client.get_subscription_role_assignments()` | Subscription RBAC assignments, or `None` on API failure |
diff --git a/README.md b/README.md
index b064c1e0..d3156e99 100644
--- a/README.md
+++ b/README.md
@@ -59,10 +59,10 @@ Findings map to NIST FIPS 203 (ML-KEM), FIPS 204 (ML-DSA), and FIPS 205 (SLH-DSA
 
 | Feature | Description |
 |---|---|
-| **Misconfiguration Scanner** | Runs 80 Azure security rules across storage, network, identity, database, compute, Key Vault, AKS, post-quantum cryptography, backup, serverless, private endpoint, and supply chain posture |
+| **Misconfiguration Scanner** | Runs 95 Azure security rules across storage, network, identity, database, compute, Key Vault, AKS, Kubernetes workloads, post-quantum cryptography, backup, serverless, private endpoint, and supply chain posture |
 | **Compliance Mapper** | Maps findings to CIS Benchmarks, NIST CSF, ISO 27001, and SOC 2 framework JSON files |
 | **Scan History API** | Stores scans and findings in PostgreSQL and exposes findings, score, scan history, compliance posture, drift, and resource inventory over REST |
-| **Remediation Playbooks** | Every rule ships with a matching Azure CLI remediation script (80 playbooks) |
+| **Remediation Playbooks** | Every documented rule ships with a matching review-gated remediation script (95 playbooks) |
 | **Security Dashboard** | Full React dashboard deployed on Vercel - live monitoring, findings, compliance, drift, prioritization, and AI-layer views |
 | **Project Website** | Documentation and reference site at [openshield-website.vercel.app](https://openshield-website.vercel.app) - blog, rules gallery, docs, roadmap, releases, and interactive playground |
 | **Sentinel Integration** | Normalises findings and pushes them into Microsoft Sentinel via a Log Analytics custom table and KQL analytics rules |
@@ -104,11 +104,11 @@ Project policies and assurance evidence:
 flowchart TD
     A["React Dashboard\nVercel · Live"]
     B["Flask REST API\nJWT · CORS · Blueprints"]
-    C["Scanner Engine\n80 Python rules"]
+    C["Scanner Engine\n95 Python rules"]
     D["Azure Subscription\nScanned via Azure SDK + Graph"]
     E["Compliance Framework JSON\nCIS · NIST · ISO 27001 · SOC 2"]
     F["PostgreSQL Database\nFindings · Scans"]
-    G["Azure CLI Playbooks\n80 remediation scripts"]
+    G["Azure CLI Playbooks\n95 remediation scripts"]
     H["sentinel/ingest.py\nNormalise + HMAC upload"]
     I["Microsoft Sentinel\nOpenShieldFindings_CL · KQL rules"]
 
diff --git a/compliance/frameworks/cis_azure_benchmark.json b/compliance/frameworks/cis_azure_benchmark.json
index 10bd80ad..175c6298 100644
--- a/compliance/frameworks/cis_azure_benchmark.json
+++ b/compliance/frameworks/cis_azure_benchmark.json
@@ -263,6 +263,81 @@
       "control_name": "AKS node OS upgrade baseline (not mapped in CIS Azure Foundations 2.0.0)",
       "description": "Microsoft recommends a managed node OS upgrade channel for timely security patches. This check has no direct control in the repository's CIS Azure Foundations 2.0.0 benchmark."
     },
+    "AZ-AKS-007": {
+      "control_id": "N/A-AKS-007",
+      "control_name": "AKS API authorized IP baseline (not mapped in CIS Azure Foundations 2.0.0)",
+      "description": "Restricting a public AKS API server to approved administrator networks has no direct control in the repository's CIS Azure Foundations 2.0.0 benchmark."
+    },
+    "AZ-AKS-008": {
+      "control_id": "N/A-AKS-008",
+      "control_name": "AKS network policy engine baseline (not mapped in CIS Azure Foundations 2.0.0)",
+      "description": "Enabling a Kubernetes network policy engine has no direct control in the repository's CIS Azure Foundations 2.0.0 benchmark."
+    },
+    "AZ-AKS-009": {
+      "control_id": "N/A-AKS-009",
+      "control_name": "Kubernetes namespace NetworkPolicy baseline (not mapped in CIS Azure Foundations 2.0.0)",
+      "description": "Requiring NetworkPolicy objects for application namespaces is a Kubernetes data-plane control outside the Azure Foundations benchmark."
+    },
+    "AZ-AKS-010": {
+      "control_id": "2.1.8",
+      "control_name": "Ensure that Microsoft Defender for Containers is set to On",
+      "description": "The Containers pricing plan must use the Standard tier to provide Microsoft Defender for Containers protection."
+    },
+    "AZ-AKS-011": {
+      "control_id": "N/A-AKS-011",
+      "control_name": "AKS Key Vault or KMS secret protection baseline (not directly mapped)",
+      "description": "Key Vault CSI and KMS-backed Kubernetes secret protection have no single direct control in CIS Azure Foundations 2.0.0."
+    },
+    "AZ-AKS-012": {
+      "control_id": "N/A-AKS-012",
+      "control_name": "Secrets Store CSI rotation baseline (not directly mapped)",
+      "description": "Automatic rotation of CSI-mounted secrets has no direct control in CIS Azure Foundations 2.0.0."
+    },
+    "AZ-AKS-013": {
+      "control_id": "N/A-AKS-013",
+      "control_name": "Privileged container baseline (Kubernetes data-plane control)",
+      "description": "Privileged container restrictions are Kubernetes data-plane controls outside CIS Azure Foundations 2.0.0."
+    },
+    "AZ-AKS-014": {
+      "control_id": "N/A-AKS-014",
+      "control_name": "Host network baseline (Kubernetes data-plane control)",
+      "description": "Host network restrictions are Kubernetes data-plane controls outside CIS Azure Foundations 2.0.0."
+    },
+    "AZ-AKS-015": {
+      "control_id": "N/A-AKS-015",
+      "control_name": "Host PID baseline (Kubernetes data-plane control)",
+      "description": "Host PID namespace restrictions are Kubernetes data-plane controls outside CIS Azure Foundations 2.0.0."
+    },
+    "AZ-AKS-016": {
+      "control_id": "N/A-AKS-016",
+      "control_name": "Host IPC baseline (Kubernetes data-plane control)",
+      "description": "Host IPC namespace restrictions are Kubernetes data-plane controls outside CIS Azure Foundations 2.0.0."
+    },
+    "AZ-AKS-017": {
+      "control_id": "N/A-AKS-017",
+      "control_name": "HostPath volume baseline (Kubernetes data-plane control)",
+      "description": "HostPath volume restrictions are Kubernetes data-plane controls outside CIS Azure Foundations 2.0.0."
+    },
+    "AZ-AKS-018": {
+      "control_id": "N/A-AKS-018",
+      "control_name": "Kubernetes cluster-admin assignment baseline (not directly mapped)",
+      "description": "Kubernetes ClusterRoleBinding scope is outside CIS Azure Foundations 2.0.0."
+    },
+    "AZ-AKS-019": {
+      "control_id": "N/A-AKS-019",
+      "control_name": "Trusted container registry baseline (not directly mapped)",
+      "description": "Kubernetes workload registry trust policy has no direct control in CIS Azure Foundations 2.0.0."
+    },
+    "AZ-AKS-020": {
+      "control_id": "N/A-AKS-020",
+      "control_name": "Latest image tag baseline (not directly mapped)",
+      "description": "Kubernetes image tag immutability has no direct control in CIS Azure Foundations 2.0.0."
+    },
+    "AZ-AKS-021": {
+      "control_id": "N/A-AKS-021",
+      "control_name": "Digest-pinned image baseline (not directly mapped)",
+      "description": "Kubernetes image digest pinning has no direct control in CIS Azure Foundations 2.0.0."
+    },
     "AZ-IDN-010": {
       "control_id": "N/A-IDN-010",
       "control_name": "App Registration ownership (not mapped in CIS Azure Foundations 2.0.0)",
diff --git a/compliance/frameworks/iso27001.json b/compliance/frameworks/iso27001.json
index 9044730f..9312c4bb 100644
--- a/compliance/frameworks/iso27001.json
+++ b/compliance/frameworks/iso27001.json
@@ -263,6 +263,81 @@
       "control_name": "Management of technical vulnerabilities",
       "description": "Automatic AKS node OS upgrades help deploy tested security patches within a managed maintenance process."
     },
+    "AZ-AKS-007": {
+      "control_id": "A.13.1.1",
+      "control_name": "Network controls",
+      "description": "Authorized IP ranges constrain public AKS API access to approved networks."
+    },
+    "AZ-AKS-008": {
+      "control_id": "A.13.1.3",
+      "control_name": "Segregation in networks",
+      "description": "A Kubernetes network policy engine enables workload network segregation."
+    },
+    "AZ-AKS-009": {
+      "control_id": "A.13.1.3",
+      "control_name": "Segregation in networks",
+      "description": "Namespace NetworkPolicy objects enforce workload communication boundaries."
+    },
+    "AZ-AKS-010": {
+      "control_id": "A.12.6.1",
+      "control_name": "Management of technical vulnerabilities",
+      "description": "Defender for Containers monitors workload threats and vulnerabilities."
+    },
+    "AZ-AKS-011": {
+      "control_id": "A.10.1.1",
+      "control_name": "Policy on cryptographic controls",
+      "description": "Key Vault or KMS-backed protection applies approved cryptographic controls to secrets."
+    },
+    "AZ-AKS-012": {
+      "control_id": "A.9.4.3",
+      "control_name": "Password management system",
+      "description": "CSI secret rotation maintains mounted credentials through an approved lifecycle."
+    },
+    "AZ-AKS-013": {
+      "control_id": "A.9.4.1",
+      "control_name": "Information access restriction",
+      "description": "Privileged container restrictions preserve the workload and node access boundary."
+    },
+    "AZ-AKS-014": {
+      "control_id": "A.13.1.3",
+      "control_name": "Segregation in networks",
+      "description": "Host network restrictions preserve workload network segregation."
+    },
+    "AZ-AKS-015": {
+      "control_id": "A.9.4.1",
+      "control_name": "Information access restriction",
+      "description": "Host PID restrictions prevent access to node process information."
+    },
+    "AZ-AKS-016": {
+      "control_id": "A.9.4.1",
+      "control_name": "Information access restriction",
+      "description": "Host IPC restrictions prevent access to node communication resources."
+    },
+    "AZ-AKS-017": {
+      "control_id": "A.9.4.1",
+      "control_name": "Information access restriction",
+      "description": "HostPath restrictions prevent direct access to node filesystems."
+    },
+    "AZ-AKS-018": {
+      "control_id": "A.9.2.3",
+      "control_name": "Management of privileged access rights",
+      "description": "A cluster-admin allowlist limits Kubernetes privileged access."
+    },
+    "AZ-AKS-019": {
+      "control_id": "A.14.2.5",
+      "control_name": "Secure system engineering principles",
+      "description": "Trusted image sources preserve the approved software supply boundary."
+    },
+    "AZ-AKS-020": {
+      "control_id": "A.12.1.2",
+      "control_name": "Change management",
+      "description": "Explicit image versions make deployment changes reviewable."
+    },
+    "AZ-AKS-021": {
+      "control_id": "A.14.2.5",
+      "control_name": "Secure system engineering principles",
+      "description": "Digest-pinned images provide immutable deployment inputs."
+    },
     "AZ-IDN-010": {
       "control_id": "A.9.2.1",
       "control_name": "User registration and de-registration",
diff --git a/compliance/frameworks/nist_csf.json b/compliance/frameworks/nist_csf.json
index 7fe766b0..0affb0fe 100644
--- a/compliance/frameworks/nist_csf.json
+++ b/compliance/frameworks/nist_csf.json
@@ -263,6 +263,81 @@
       "control_name": "A vulnerability management plan is developed and implemented",
       "description": "Managed node OS upgrade channels apply tested security updates to reduce exposure to known operating-system vulnerabilities."
     },
+    "AZ-AKS-007": {
+      "control_id": "PR.AC-3",
+      "control_name": "Remote access is managed",
+      "description": "Approved API server IP ranges constrain remote AKS administration paths."
+    },
+    "AZ-AKS-008": {
+      "control_id": "PR.AC-5",
+      "control_name": "Network integrity is protected",
+      "description": "A network policy engine provides enforcement for Kubernetes workload traffic boundaries."
+    },
+    "AZ-AKS-009": {
+      "control_id": "PR.AC-5",
+      "control_name": "Network integrity is protected",
+      "description": "Namespace NetworkPolicy objects define and enforce allowed workload communication."
+    },
+    "AZ-AKS-010": {
+      "control_id": "DE.CM-8",
+      "control_name": "Vulnerability scans are performed",
+      "description": "Defender for Containers supplies managed workload vulnerability and threat monitoring."
+    },
+    "AZ-AKS-011": {
+      "control_id": "PR.DS-1",
+      "control_name": "Data at rest is protected",
+      "description": "Key Vault CSI or KMS-backed protection reduces exposure of Kubernetes secret material."
+    },
+    "AZ-AKS-012": {
+      "control_id": "PR.AC-1",
+      "control_name": "Identities and credentials are managed",
+      "description": "Automatic CSI rotation keeps mounted workload credentials within their approved lifecycle."
+    },
+    "AZ-AKS-013": {
+      "control_id": "PR.AC-4",
+      "control_name": "Access permissions are managed",
+      "description": "Blocking privileged containers prevents workloads from receiving unrestricted node capabilities."
+    },
+    "AZ-AKS-014": {
+      "control_id": "PR.AC-5",
+      "control_name": "Network integrity is protected",
+      "description": "Blocking hostNetwork keeps workload networking inside Kubernetes isolation boundaries."
+    },
+    "AZ-AKS-015": {
+      "control_id": "PR.AC-4",
+      "control_name": "Access permissions are managed",
+      "description": "Blocking hostPID prevents workloads from observing or controlling node processes."
+    },
+    "AZ-AKS-016": {
+      "control_id": "PR.AC-4",
+      "control_name": "Access permissions are managed",
+      "description": "Blocking hostIPC prevents workloads from sharing node inter-process communication resources."
+    },
+    "AZ-AKS-017": {
+      "control_id": "PR.AC-4",
+      "control_name": "Access permissions are managed",
+      "description": "Blocking hostPath prevents direct workload access to the node filesystem."
+    },
+    "AZ-AKS-018": {
+      "control_id": "PR.AC-4",
+      "control_name": "Access permissions are managed",
+      "description": "Restricting cluster-admin bindings enforces least privilege for Kubernetes subjects."
+    },
+    "AZ-AKS-019": {
+      "control_id": "PR.DS-6",
+      "control_name": "Integrity checking mechanisms are used",
+      "description": "Trusted registries provide an approved source boundary for workload images."
+    },
+    "AZ-AKS-020": {
+      "control_id": "PR.IP-1",
+      "control_name": "A baseline configuration is maintained",
+      "description": "Avoiding latest tags makes deployed image versions explicit and reviewable."
+    },
+    "AZ-AKS-021": {
+      "control_id": "PR.DS-6",
+      "control_name": "Integrity checking mechanisms are used",
+      "description": "Digest pinning binds a workload to verified immutable image content."
+    },
     "AZ-IDN-010": {
       "control_id": "PR.AC-4",
       "control_name": "Access permissions and authorizations are managed",
diff --git a/compliance/frameworks/soc2.json b/compliance/frameworks/soc2.json
index 38430730..d51d6171 100644
--- a/compliance/frameworks/soc2.json
+++ b/compliance/frameworks/soc2.json
@@ -263,6 +263,81 @@
       "control_name": "Detects and Monitors Configuration Changes",
       "description": "Managed node OS upgrade channels maintain worker-node security patches through an observable Azure-controlled process."
     },
+    "AZ-AKS-007": {
+      "control_id": "CC6.6",
+      "control_name": "Restricts Access to Information Assets",
+      "description": "Authorized IP ranges restrict AKS API access to approved administration networks."
+    },
+    "AZ-AKS-008": {
+      "control_id": "CC6.6",
+      "control_name": "Restricts Access from Outside Boundaries",
+      "description": "A network policy engine enables workload boundary enforcement."
+    },
+    "AZ-AKS-009": {
+      "control_id": "CC6.6",
+      "control_name": "Restricts Access from Outside Boundaries",
+      "description": "Namespace NetworkPolicy objects enforce workload traffic boundaries."
+    },
+    "AZ-AKS-010": {
+      "control_id": "CC7.1",
+      "control_name": "Detection and Monitoring",
+      "description": "Defender for Containers detects workload threats and vulnerabilities."
+    },
+    "AZ-AKS-011": {
+      "control_id": "CC6.1",
+      "control_name": "Logical Access Security",
+      "description": "Key Vault or KMS-backed secret protection reduces credential exposure."
+    },
+    "AZ-AKS-012": {
+      "control_id": "CC6.1",
+      "control_name": "Logical Access Security",
+      "description": "Automatic CSI rotation maintains workload credential controls."
+    },
+    "AZ-AKS-013": {
+      "control_id": "CC6.1",
+      "control_name": "Logical Access Security",
+      "description": "Privileged container restrictions preserve node isolation."
+    },
+    "AZ-AKS-014": {
+      "control_id": "CC6.6",
+      "control_name": "Restricts Access from Outside Boundaries",
+      "description": "Host network restrictions preserve network isolation."
+    },
+    "AZ-AKS-015": {
+      "control_id": "CC6.1",
+      "control_name": "Logical Access Security",
+      "description": "Host PID restrictions prevent node process access."
+    },
+    "AZ-AKS-016": {
+      "control_id": "CC6.1",
+      "control_name": "Logical Access Security",
+      "description": "Host IPC restrictions prevent node communication access."
+    },
+    "AZ-AKS-017": {
+      "control_id": "CC6.1",
+      "control_name": "Logical Access Security",
+      "description": "HostPath restrictions prevent direct node filesystem access."
+    },
+    "AZ-AKS-018": {
+      "control_id": "CC6.3",
+      "control_name": "Role-Based Access",
+      "description": "A cluster-admin allowlist enforces least privilege."
+    },
+    "AZ-AKS-019": {
+      "control_id": "CC8.1",
+      "control_name": "Change Management",
+      "description": "Trusted registries constrain approved deployment sources."
+    },
+    "AZ-AKS-020": {
+      "control_id": "CC8.1",
+      "control_name": "Change Management",
+      "description": "Explicit image versions make workload changes traceable."
+    },
+    "AZ-AKS-021": {
+      "control_id": "CC8.1",
+      "control_name": "Change Management",
+      "description": "Digest pinning makes deployed image content immutable."
+    },
     "AZ-IDN-010": {
       "control_id": "CC6.2",
       "control_name": "Registers and Authorizes Users",
diff --git a/config/aks-kubeconfig-contexts.example.json b/config/aks-kubeconfig-contexts.example.json
new file mode 100644
index 00000000..50703e1a
--- /dev/null
+++ b/config/aks-kubeconfig-contexts.example.json
@@ -0,0 +1,3 @@
+{
+  "/subscriptions/00000000-0000-0000-0000-000000000000/resourceGroups/platform-prod/providers/Microsoft.ContainerService/managedClusters/aks-prod": "aks-prod-platform-context"
+}
diff --git a/config/aks-security-policy.example.json b/config/aks-security-policy.example.json
new file mode 100644
index 00000000..824bb556
--- /dev/null
+++ b/config/aks-security-policy.example.json
@@ -0,0 +1,7 @@
+{
+  "approved_authorized_ip_ranges": ["203.0.113.0/24"],
+  "trusted_registry_prefixes": ["contoso.azurecr.io/"],
+  "allowed_cluster_admin_subjects": ["Group:aks-platform-admins"],
+  "excluded_namespaces": ["kube-system", "gatekeeper-system"],
+  "require_image_digests": true
+}
diff --git a/docs/aks-security-rules.md b/docs/aks-security-rules.md
index 3a8541b2..15e35de8 100644
--- a/docs/aks-security-rules.md
+++ b/docs/aks-security-rules.md
@@ -1,54 +1,106 @@
-# AKS Security Rules
+# AKS and Kubernetes Security Rules
 
-OpenShield evaluates Azure Kubernetes Service control-plane configuration through
-the Azure Resource Manager API. The scanner does not download kubeconfig files,
-request Kubernetes administrator credentials, or inspect in-cluster workloads.
+OpenShield evaluates Azure Kubernetes Service across three evidence planes:
+
+- Azure Resource Manager for cluster configuration and add-on state.
+- Microsoft Defender for Cloud for the Containers protection plan.
+- The Kubernetes API for namespaces, network policies, workloads, pods, and cluster role bindings.
+
+The original six AKS rules remain control-plane-only. The issue #255 extension adds fifteen independently remediable controls so each domain and subdomain has its own rule, finding, mapping, test coverage, and playbook.
 
 ## Coverage
 
-| Rule | Control |
-|---|---|
-| `AZ-AKS-001` | Private API server endpoint |
-| `AZ-AKS-002` | Local account disablement |
-| `AZ-AKS-003` | Control-plane managed identity |
-| `AZ-AKS-004` | OIDC issuer and Workload Identity |
-| `AZ-AKS-005` | Azure Policy add-on |
-| `AZ-AKS-006` | Managed node OS security upgrades |
+| Domain | Subdomain | Rule | Control |
+|---|---|---|---|
+| Control plane | Endpoint isolation | `AZ-AKS-001` | Private API server endpoint |
+| Identity | Local authentication | `AZ-AKS-002` | Local account disablement |
+| Identity | Cluster identity | `AZ-AKS-003` | Control-plane managed identity |
+| Identity | Workload identity | `AZ-AKS-004` | OIDC issuer and Workload Identity |
+| Governance | Admission policy | `AZ-AKS-005` | Azure Policy add-on |
+| Node security | Patch management | `AZ-AKS-006` | Managed node OS security upgrades |
+| Control plane | Public endpoint restriction | `AZ-AKS-007` | Approved API server IP ranges |
+| Network security | Policy engine | `AZ-AKS-008` | Azure, Calico, or Cilium network policy engine |
+| Network security | Namespace segmentation | `AZ-AKS-009` | NetworkPolicy presence in eligible namespaces |
+| Threat protection | Managed detection | `AZ-AKS-010` | Defender for Containers subscription plan |
+| Secret protection | Encryption and external secrets | `AZ-AKS-011` | Key Vault CSI or Key Vault KMS protection |
+| Secret protection | Credential lifecycle | `AZ-AKS-012` | Secrets Store CSI automatic rotation |
+| Workload isolation | Privilege boundary | `AZ-AKS-013` | Privileged application and init containers |
+| Workload isolation | Network namespace | `AZ-AKS-014` | Host network access |
+| Workload isolation | Process namespace | `AZ-AKS-015` | Host PID access |
+| Workload isolation | IPC namespace | `AZ-AKS-016` | Host IPC access |
+| Workload isolation | Node filesystem | `AZ-AKS-017` | HostPath volumes |
+| Authorization | Cluster privilege | `AZ-AKS-018` | Approved cluster-admin subjects |
+| Software supply chain | Registry trust | `AZ-AKS-019` | Approved image registry prefixes |
+| Software supply chain | Floating release | `AZ-AKS-020` | Latest and implicit latest tags |
+| Software supply chain | Image immutability | `AZ-AKS-021` | SHA-256 digest pinning |
+
+## Organization policy
+
+Rules that require an approved scope or allowlist load a strict JSON policy from `OPENSHIELD_AKS_SECURITY_POLICY`. Start from `config/aks-security-policy.example.json` and keep the deployed policy outside the repository if it contains internal network or registry names.
+
+The policy defines:
+
+- Approved API server CIDR ranges.
+- Trusted registry and repository prefixes.
+- Approved `cluster-admin` Groups and Users in `Kind:name` form.
+- Approved `cluster-admin` ServiceAccounts in `ServiceAccount:namespace:name` form.
+- Namespaces excluded from workload evaluation.
+- Whether image digest pinning is required.
+
+Missing, malformed, or incomplete policy is UNKNOWN. OpenShield does not invent organization-specific defaults.
+
+## Kubernetes credentials
 
-## Required permissions
+Set `OPENSHIELD_AKS_KUBECONFIG` to a read-only kubeconfig. Set `OPENSHIELD_AKS_KUBECONFIG_CONTEXTS` to a strict JSON mapping from each complete AKS resource ID to its kubeconfig context name. Start from `config/aks-kubeconfig-contexts.example.json`.
 
-The scanning identity needs the following Azure Resource Manager action at the
-subscription or relevant resource-group scope:
+Resource IDs are matched case-insensitively. Each context may map to only one AKS resource ID, and every scanned cluster must have an explicit entry. Missing, malformed, unresolved, or ambiguous mappings remain UNKNOWN. This prevents clusters with the same resource-group-scoped name from sharing evidence accidentally.
+
+The scanner never records kubeconfig content, tokens, Kubernetes Secrets, or environment variable values.
+
+The Kubernetes identity needs read-only access equivalent to:
+
+```text
+get,list namespaces
+get,list pods
+get,list deployments,statefulsets,daemonsets.apps
+get,list jobs,cronjobs.batch
+get,list networkpolicies.networking.k8s.io
+get,list clusterrolebindings.rbac.authorization.k8s.io
+```
+
+The Azure identity needs:
 
 ```text
 Microsoft.ContainerService/managedClusters/read
+Microsoft.Security/pricings/read
 ```
 
-Azure's built-in **Reader** role includes this action. No Kubernetes RBAC role,
-cluster credential, Microsoft Graph permission, or data-plane access is needed.
+## Evidence states
+
+The scan engine represents FAIL as a finding and represents PASS, UNKNOWN, and NOT_APPLICABLE as an empty finding list with an explicit log entry.
+
+- `FAIL` requires positive evidence of the unsafe setting.
+- `UNKNOWN` is used for missing policy, missing kubeconfig, stopped or unreachable clusters, authorization errors, malformed identity, and unavailable ARM or Defender evidence.
+- `PARTIAL` preserves positive evidence from reachable namespaces but never treats missing namespaces as compliant.
+- `NOT_APPLICABLE` is used when no AKS cluster, eligible namespace, workload, CSI provider, or relevant object exists.
 
-If Azure denies or fails the inventory request, the AKS accessor returns an
-indeterminate state. Rules skip evaluation and log the failure instead of
-reporting the subscription as compliant.
+Every FAIL includes cluster, namespace, workload, container, image, subject, role, source, collection timestamp, observed value, expected value, required permissions, severity, confidence, and a null UNKNOWN reason.
 
 ## Remediation safety
 
-The matching CLI playbooks validate the selected Azure account and cluster,
-describe operational impact, and require the operator to type `APPLY`. Enabling
-a private endpoint, disabling local credentials, migrating identity, or changing
-node patch behavior should first be tested in a non-production cluster.
+Every rule has a matching `playbooks/cli/fix_az_aks_<id>.sh` file. The playbooks use the same review gate as other enterprise controls. Safe Azure changes are automated only after account, target, and impact confirmation. Kubernetes manifest, RBAC, and network policy changes remain operator-reviewed because a generic automatic patch could interrupt production or remove break-glass access.
 
 ## Compliance note
 
-This repository currently models CIS Microsoft Azure Foundations Benchmark
-2.0.0. That benchmark does not directly identify these six AKS configuration
-checks, so their CIS values are explicitly recorded as `N/A-AKS-*`. The rules
-do not claim CIS Kubernetes Benchmark coverage. NIST CSF 1.1, ISO/IEC 27001:2013,
-and SOC 2 mappings follow the framework versions already used by OpenShield.
+The repository models CIS Microsoft Azure Foundations Benchmark 2.0.0. Only Defender for Containers has a direct control in that benchmark. Kubernetes data-plane controls are explicitly recorded as `N/A-AKS-*` rather than being assigned unsupported CIS Kubernetes Benchmark identifiers. NIST CSF 1.1, ISO/IEC 27001:2013, and SOC 2 mappings follow the framework versions already used by OpenShield.
 
 ## Authoritative references
 
-- [Azure Policy built-ins for AKS](https://learn.microsoft.com/azure/aks/policy-reference)
-- [AKS baseline architecture](https://learn.microsoft.com/azure/architecture/reference-architectures/containers/aks/baseline-aks)
-- [AKS managed-cluster API](https://learn.microsoft.com/rest/api/aks/managed-clusters/get)
-- [AKS node OS automatic upgrades](https://learn.microsoft.com/azure/aks/auto-upgrade-node-os-image)
+- [AKS API server authorized IP ranges](https://learn.microsoft.com/azure/aks/api-server-authorized-ip-ranges)
+- [AKS network policies](https://learn.microsoft.com/azure/aks/use-network-policies)
+- [Defender for Containers](https://learn.microsoft.com/azure/defender-for-cloud/defender-for-containers-introduction)
+- [Azure Key Vault provider for Secrets Store CSI](https://learn.microsoft.com/azure/aks/csi-secrets-store-driver)
+- [AKS Key Management Service](https://learn.microsoft.com/azure/aks/use-kms-etcd-encryption)
+- [Kubernetes Pod Security Standards](https://kubernetes.io/docs/concepts/security/pod-security-standards/)
+- [Kubernetes RBAC](https://kubernetes.io/docs/reference/access-authn-authz/rbac/)
+- [Kubernetes images](https://kubernetes.io/docs/concepts/containers/images/)
diff --git a/docs/architecture.md b/docs/architecture.md
index cbf8a71e..b025b90d 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -2,7 +2,7 @@
 
 ## Overview
 
-OpenShield is a modular, open source Cloud Security Posture Management (CSPM) platform for Azure. It scans your Azure subscription against 80 security rules, maps findings to compliance frameworks (CIS, NIST CSF, ISO 27001, SOC 2), stores results in PostgreSQL, and exposes posture data through a Flask REST API consumed by a live React dashboard.
+OpenShield is a modular, open source Cloud Security Posture Management (CSPM) platform for Azure. It scans Azure and connected Kubernetes evidence against 95 documented security rules, maps findings to compliance frameworks (CIS, NIST CSF, ISO 27001, SOC 2), stores results in PostgreSQL, and exposes posture data through a Flask REST API consumed by a live React dashboard.
 
 ---
 
@@ -43,7 +43,7 @@ OpenShield is a modular, open source Cloud Security Posture Management (CSPM) pl
 ┌───────────▼──────────────────────────────────────────────────────┐
 │                   Rule Modules (scanner/rules/)                   │
 │                                                                   │
-│  80 rule files across Storage, Network, Identity, Database,       │
+│  95 documented rules across Storage, Network, Identity, Database, │
 │  Compute, Key Vault, AKS, post-quantum cryptography, Backup,      │
 │  Serverless, Private Endpoint posture, and Supply Chain           │
 └───────────┬───────────────────────────────────────────────────────┘
@@ -111,7 +111,7 @@ result = engine.run_scan()
 
 ### 4. Current Rule Modules
 
-There are 80 rule files in `scanner/rules/`. See `docs/rules-reference.md` for the full table.
+There are 95 documented controls in `docs/rules-reference.md`, including 21 AKS and Kubernetes controls.
 
 | Category | Count | Rules |
 |---|---|---|
@@ -121,7 +121,7 @@ There are 80 rule files in `scanner/rules/`. See `docs/rules-reference.md` for t
 | Database | 4 | AZ-DB-001 to 004 |
 | Compute | 4 | AZ-CMP-001 to 004 |
 | Key Vault | 5 | AZ-KV-001 to 005 |
-| Kubernetes | 6 | AZ-AKS-001 to 006 |
+| Kubernetes | 21 | AZ-AKS-001 to 021 |
 | Post-quantum | 3 | AZ-PQC-001 to 003 |
 | Backup | 4 | AZ-BAK-001, 002, 004, 006 |
 | Serverless | 5 | AZ-FUNC-001 to 005 |
diff --git a/docs/learn/index.html b/docs/learn/index.html
index 048a79ea..6ba20e2d 100644
--- a/docs/learn/index.html
+++ b/docs/learn/index.html
@@ -841,7 +841,7 @@ <h2 class="section-title">Production-shaped, MVP-friendly architecture</h2>
 
     <section id="rules" class="learn-section" data-search="rules coverage network storage compute database identity key vault post quantum severity high medium low findings">
       <p class="section-kicker">Rule coverage</p>
-      <h2 class="section-title">80 Azure security rules</h2>
+      <h2 class="section-title">95 Azure security rules</h2>
       <p class="section-intro">
         OpenShield currently has 80 dynamic rules. The strongest contributor work improves rule accuracy, reduces false positives,
         strengthens validation, or improves remediation quality.
diff --git a/docs/rules-reference.md b/docs/rules-reference.md
index b9e12775..3e3439a7 100644
--- a/docs/rules-reference.md
+++ b/docs/rules-reference.md
@@ -1,6 +1,6 @@
 # Rules Reference
 
-OpenShield currently ships 80 Azure scan rules. This table is generated from the module-level constants in `scanner/rules/`.
+OpenShield currently ships 95 Azure scan rules. This table is generated from the module-level constants in `scanner/rules/`.
 
 | Rule ID | Name | Severity | Category | CIS | NIST | ISO 27001 |
 |---|---|---|---|---|---|---|
@@ -61,6 +61,21 @@ OpenShield currently ships 80 Azure scan rules. This table is generated from the
 | AZ-AKS-004 | AKS Workload Identity Not Fully Enabled | MEDIUM | Kubernetes | N/A-AKS-004 | PR.AC-4 | A.9.2.3 |
 | AZ-AKS-005 | AKS Azure Policy Add-on Not Enabled | MEDIUM | Kubernetes | N/A-AKS-005 | PR.IP-1 | A.12.1.2 |
 | AZ-AKS-006 | AKS Node OS Automatic Upgrades Disabled | HIGH | Kubernetes | N/A-AKS-006 | PR.IP-12 | A.12.6.1 |
+| AZ-AKS-007 | AKS API Server Lacks Approved IP Restrictions | HIGH | Kubernetes | N/A-AKS-007 | PR.AC-3 | A.13.1.1 |
+| AZ-AKS-008 | AKS Cluster Has No Kubernetes Network Policy | HIGH | Kubernetes | N/A-AKS-008 | PR.AC-5 | A.13.1.3 |
+| AZ-AKS-009 | Kubernetes Namespace Has No NetworkPolicy | HIGH | Kubernetes | N/A-AKS-009 | PR.AC-5 | A.13.1.3 |
+| AZ-AKS-010 | Defender for Containers Protection Disabled | HIGH | Kubernetes | 2.1.8 | DE.CM-8 | A.12.6.1 |
+| AZ-AKS-011 | AKS Secrets Lack Key Vault or KMS-Backed Protection | HIGH | Kubernetes | N/A-AKS-011 | PR.DS-1 | A.10.1.1 |
+| AZ-AKS-012 | Secrets Store CSI Secret Rotation Disabled | MEDIUM | Kubernetes | N/A-AKS-012 | PR.AC-1 | A.9.4.3 |
+| AZ-AKS-013 | Kubernetes Workload Permits Privileged Containers | CRITICAL | Kubernetes | N/A-AKS-013 | PR.AC-4 | A.9.4.1 |
+| AZ-AKS-014 | Kubernetes Workload Uses Host Network | HIGH | Kubernetes | N/A-AKS-014 | PR.AC-5 | A.13.1.3 |
+| AZ-AKS-015 | Kubernetes Workload Uses Host PID Namespace | HIGH | Kubernetes | N/A-AKS-015 | PR.AC-4 | A.9.4.1 |
+| AZ-AKS-016 | Kubernetes Workload Uses Host IPC Namespace | HIGH | Kubernetes | N/A-AKS-016 | PR.AC-4 | A.9.4.1 |
+| AZ-AKS-017 | Kubernetes Workload Uses Unrestricted HostPath Volumes | HIGH | Kubernetes | N/A-AKS-017 | PR.AC-4 | A.9.4.1 |
+| AZ-AKS-018 | Kubernetes Cluster Admin Access Assigned Too Broadly | CRITICAL | Kubernetes | N/A-AKS-018 | PR.AC-4 | A.9.2.3 |
+| AZ-AKS-019 | Kubernetes Workload Uses Untrusted Registry | HIGH | Kubernetes | N/A-AKS-019 | PR.DS-6 | A.14.2.5 |
+| AZ-AKS-020 | Kubernetes Workload Uses Latest Image Tag | HIGH | Kubernetes | N/A-AKS-020 | PR.IP-1 | A.12.1.2 |
+| AZ-AKS-021 | Kubernetes Workload Image Is Not Digest Pinned | MEDIUM | Kubernetes | N/A-AKS-021 | PR.DS-6 | A.14.2.5 |
 | AZ-BAK-001 | Backup Soft Delete Disabled or Below 35 Days | CRITICAL | Backup | N/A-BAK-001 | PR.IP-4 | A.12.3.1 |
 | AZ-BAK-002 | Backup Vault Immutability Disabled | HIGH | Backup | N/A-BAK-002 | PR.IP-4 | A.12.3.1 |
 | AZ-BAK-004 | Backup Multiuser Authorization Missing | HIGH | Backup | N/A-BAK-004 | PR.AC-4 | A.9.2.3 |
diff --git a/playbooks/cli/fix_az_aks_007.sh b/playbooks/cli/fix_az_aks_007.sh
new file mode 100755
index 00000000..1edd6a8d
--- /dev/null
+++ b/playbooks/cli/fix_az_aks_007.sh
@@ -0,0 +1,2 @@
+#!/bin/bash
+exec "$(dirname "$0")/review_aks_security.sh" AZ-AKS-007 "$@"
diff --git a/playbooks/cli/fix_az_aks_008.sh b/playbooks/cli/fix_az_aks_008.sh
new file mode 100755
index 00000000..4982330c
--- /dev/null
+++ b/playbooks/cli/fix_az_aks_008.sh
@@ -0,0 +1,2 @@
+#!/bin/bash
+exec "$(dirname "$0")/review_aks_security.sh" AZ-AKS-008 "$@"
diff --git a/playbooks/cli/fix_az_aks_009.sh b/playbooks/cli/fix_az_aks_009.sh
new file mode 100755
index 00000000..4a8f1c1d
--- /dev/null
+++ b/playbooks/cli/fix_az_aks_009.sh
@@ -0,0 +1,2 @@
+#!/bin/bash
+exec "$(dirname "$0")/review_aks_security.sh" AZ-AKS-009 "$@"
diff --git a/playbooks/cli/fix_az_aks_010.sh b/playbooks/cli/fix_az_aks_010.sh
new file mode 100755
index 00000000..40e9f948
--- /dev/null
+++ b/playbooks/cli/fix_az_aks_010.sh
@@ -0,0 +1,2 @@
+#!/bin/bash
+exec "$(dirname "$0")/review_aks_security.sh" AZ-AKS-010 "$@"
diff --git a/playbooks/cli/fix_az_aks_011.sh b/playbooks/cli/fix_az_aks_011.sh
new file mode 100755
index 00000000..eeb268f5
--- /dev/null
+++ b/playbooks/cli/fix_az_aks_011.sh
@@ -0,0 +1,2 @@
+#!/bin/bash
+exec "$(dirname "$0")/review_aks_security.sh" AZ-AKS-011 "$@"
diff --git a/playbooks/cli/fix_az_aks_012.sh b/playbooks/cli/fix_az_aks_012.sh
new file mode 100755
index 00000000..2ce922f7
--- /dev/null
+++ b/playbooks/cli/fix_az_aks_012.sh
@@ -0,0 +1,2 @@
+#!/bin/bash
+exec "$(dirname "$0")/review_aks_security.sh" AZ-AKS-012 "$@"
diff --git a/playbooks/cli/fix_az_aks_013.sh b/playbooks/cli/fix_az_aks_013.sh
new file mode 100755
index 00000000..60704637
--- /dev/null
+++ b/playbooks/cli/fix_az_aks_013.sh
@@ -0,0 +1,2 @@
+#!/bin/bash
+exec "$(dirname "$0")/review_aks_security.sh" AZ-AKS-013 "$@"
diff --git a/playbooks/cli/fix_az_aks_014.sh b/playbooks/cli/fix_az_aks_014.sh
new file mode 100755
index 00000000..8cff84ff
--- /dev/null
+++ b/playbooks/cli/fix_az_aks_014.sh
@@ -0,0 +1,2 @@
+#!/bin/bash
+exec "$(dirname "$0")/review_aks_security.sh" AZ-AKS-014 "$@"
diff --git a/playbooks/cli/fix_az_aks_015.sh b/playbooks/cli/fix_az_aks_015.sh
new file mode 100755
index 00000000..4d75bfb5
--- /dev/null
+++ b/playbooks/cli/fix_az_aks_015.sh
@@ -0,0 +1,2 @@
+#!/bin/bash
+exec "$(dirname "$0")/review_aks_security.sh" AZ-AKS-015 "$@"
diff --git a/playbooks/cli/fix_az_aks_016.sh b/playbooks/cli/fix_az_aks_016.sh
new file mode 100755
index 00000000..4a99eb41
--- /dev/null
+++ b/playbooks/cli/fix_az_aks_016.sh
@@ -0,0 +1,2 @@
+#!/bin/bash
+exec "$(dirname "$0")/review_aks_security.sh" AZ-AKS-016 "$@"
diff --git a/playbooks/cli/fix_az_aks_017.sh b/playbooks/cli/fix_az_aks_017.sh
new file mode 100755
index 00000000..7daaa9c5
--- /dev/null
+++ b/playbooks/cli/fix_az_aks_017.sh
@@ -0,0 +1,2 @@
+#!/bin/bash
+exec "$(dirname "$0")/review_aks_security.sh" AZ-AKS-017 "$@"
diff --git a/playbooks/cli/fix_az_aks_018.sh b/playbooks/cli/fix_az_aks_018.sh
new file mode 100755
index 00000000..66681414
--- /dev/null
+++ b/playbooks/cli/fix_az_aks_018.sh
@@ -0,0 +1,2 @@
+#!/bin/bash
+exec "$(dirname "$0")/review_aks_security.sh" AZ-AKS-018 "$@"
diff --git a/playbooks/cli/fix_az_aks_019.sh b/playbooks/cli/fix_az_aks_019.sh
new file mode 100755
index 00000000..8438e506
--- /dev/null
+++ b/playbooks/cli/fix_az_aks_019.sh
@@ -0,0 +1,2 @@
+#!/bin/bash
+exec "$(dirname "$0")/review_aks_security.sh" AZ-AKS-019 "$@"
diff --git a/playbooks/cli/fix_az_aks_020.sh b/playbooks/cli/fix_az_aks_020.sh
new file mode 100755
index 00000000..e8fbf793
--- /dev/null
+++ b/playbooks/cli/fix_az_aks_020.sh
@@ -0,0 +1,2 @@
+#!/bin/bash
+exec "$(dirname "$0")/review_aks_security.sh" AZ-AKS-020 "$@"
diff --git a/playbooks/cli/fix_az_aks_021.sh b/playbooks/cli/fix_az_aks_021.sh
new file mode 100755
index 00000000..a324fc34
--- /dev/null
+++ b/playbooks/cli/fix_az_aks_021.sh
@@ -0,0 +1,2 @@
+#!/bin/bash
+exec "$(dirname "$0")/review_aks_security.sh" AZ-AKS-021 "$@"
diff --git a/playbooks/cli/review_aks_security.sh b/playbooks/cli/review_aks_security.sh
new file mode 100755
index 00000000..a459ae3d
--- /dev/null
+++ b/playbooks/cli/review_aks_security.sh
@@ -0,0 +1,43 @@
+#!/bin/bash
+# Shared safety gate for issue #255 AKS and Kubernetes remediations.
+set -euo pipefail
+
+RULE_ID=${1:-}
+TARGET=${2:-}
+VALUE=${3:-}
+if [ -z "$RULE_ID" ] || [ -z "$TARGET" ]; then
+  echo "Usage: $0 <rule-id> <cluster-resource-id-or-kube-context> [approved-value]"
+  exit 1
+fi
+
+echo "Rule: $RULE_ID"
+echo "Target: $TARGET"
+echo "WARNING: AKS control-plane, RBAC, network policy, and workload changes can interrupt production."
+echo "Validate availability, break-glass access, admission policy, and rollback before continuing."
+read -r -p "Type APPLY to confirm that the target and impact were reviewed: " CONFIRM
+[ "$CONFIRM" = "APPLY" ] || { echo "Cancelled."; exit 1; }
+
+case "$RULE_ID" in
+  AZ-AKS-007)
+    [ -n "$VALUE" ] || { echo "Provide approved comma-separated IP ranges as the third argument."; exit 1; }
+    az account show --output none
+    az aks update --ids "$TARGET" --api-server-authorized-ip-ranges "$VALUE"
+    ;;
+  AZ-AKS-010)
+    az account show --output none
+    az security pricing create --name Containers --tier Standard
+    ;;
+  AZ-AKS-011)
+    az account show --output none
+    az aks enable-addons --ids "$TARGET" --addons azure-keyvault-secrets-provider
+    ;;
+  AZ-AKS-012)
+    az account show --output none
+    az aks addon update --ids "$TARGET" --addon azure-keyvault-secrets-provider --enable-secret-rotation
+    ;;
+  *)
+    kubectl --context "$TARGET" auth can-i get pods --all-namespaces
+    echo "Review completed. Apply the rule-specific remediation from the finding after testing the manifest or RBAC change."
+    echo "No Kubernetes object was changed automatically."
+    ;;
+esac
diff --git a/requirements.txt b/requirements.txt
index af4b192b..d82003e7 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -19,6 +19,7 @@ azure-mgmt-securityinsight==1.0.0
 azure-mgmt-loganalytics==14.0.0
 azure-mgmt-dns==8.0.0
 azure-mgmt-containerservice==41.3.0
+kubernetes==36.0.0
 azure-mgmt-recoveryservices==4.1.0
 psycopg2-binary==2.9.9
 python-dotenv==1.2.2
diff --git a/scanner/aks_security.py b/scanner/aks_security.py
new file mode 100644
index 00000000..1a553f8d
--- /dev/null
+++ b/scanner/aks_security.py
@@ -0,0 +1,426 @@
+"""Failure-aware Azure and Kubernetes evidence collection for AKS controls."""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Iterable, Mapping
+
+logger = logging.getLogger(__name__)
+
+KUBECONFIG_ENV_VAR = "OPENSHIELD_AKS_KUBECONFIG"
+KUBECONFIG_CONTEXTS_ENV_VAR = "OPENSHIELD_AKS_KUBECONFIG_CONTEXTS"
+SYSTEM_NAMESPACES = frozenset({"kube-system", "kube-public", "kube-node-lease", "gatekeeper-system"})
+
+
+class KubeconfigContextUnresolvedError(ValueError):
+    """Raised when an explicit resource-ID mapping names no kubeconfig context."""
+
+
+@dataclass(frozen=True)
+class AksClusterEvidence:
+    """One cluster's secret-free control-plane and workload posture."""
+
+    cluster: Any
+    status: str
+    source: str
+    collected_at: str
+    unknown_reason: str | None = None
+    partial_reasons: tuple[str, ...] = field(default_factory=tuple)
+    control_plane: Mapping[str, Any] = field(default_factory=dict)
+    namespaces: tuple[str, ...] = field(default_factory=tuple)
+    network_policy_namespaces: tuple[str, ...] = field(default_factory=tuple)
+    workloads: tuple[Mapping[str, Any], ...] = field(default_factory=tuple)
+    cluster_admin_bindings: tuple[Mapping[str, Any], ...] = field(default_factory=tuple)
+
+
+def _value(item: Any, name: str, default: Any = None) -> Any:
+    if isinstance(item, Mapping):
+        return item.get(name, default)
+    return getattr(item, name, default)
+
+
+def _nested(item: Any, *names: str, default: Any = None) -> Any:
+    current = item
+    for name in names:
+        current = _value(current, name, None)
+        if current is None:
+            return default
+    return current
+
+
+def _enum(value: Any) -> str:
+    return str(getattr(value, "value", value) or "").strip()
+
+
+def _normalize_cluster_id(resource_id: str) -> str:
+    normalized = resource_id.strip().rstrip("/").lower()
+    parts = normalized.split("/")
+    if (
+        len(parts) != 9
+        or parts[0]
+        or parts[1] != "subscriptions"
+        or not parts[2]
+        or parts[3] != "resourcegroups"
+        or not parts[4]
+        or parts[5] != "providers"
+        or parts[6] != "microsoft.containerservice"
+        or parts[7] != "managedclusters"
+        or not parts[8]
+    ):
+        raise ValueError("kubeconfig context keys must be complete AKS resource IDs")
+    return normalized
+
+
+def load_kubeconfig_contexts(path: str | Path) -> dict[str, str]:
+    with Path(path).open(encoding="utf-8") as handle:
+        raw = json.load(handle)
+    if not isinstance(raw, dict):
+        raise ValueError("AKS kubeconfig context mapping must be a JSON object")
+    mapping: dict[str, str] = {}
+    assigned_contexts: set[str] = set()
+    for resource_id, context in raw.items():
+        if not isinstance(resource_id, str) or not isinstance(context, str) or not context.strip():
+            raise ValueError("AKS kubeconfig context mapping must contain non-empty string keys and values")
+        normalized_id = _normalize_cluster_id(resource_id)
+        normalized_context = context.strip()
+        if normalized_id in mapping:
+            raise ValueError("AKS kubeconfig context mapping contains duplicate resource IDs")
+        if normalized_context in assigned_contexts:
+            raise ValueError("AKS kubeconfig contexts must map to only one cluster resource ID")
+        mapping[normalized_id] = normalized_context
+        assigned_contexts.add(normalized_context)
+    return mapping
+
+
+def _container(container: Any) -> dict[str, Any]:
+    security = _value(container, "security_context")
+    capabilities = _nested(security, "capabilities", "add", default=[]) or []
+    secret_references: set[str] = set()
+    for environment in _value(container, "env", []) or []:
+        secret_name = _nested(environment, "value_from", "secret_key_ref", "name")
+        if secret_name:
+            secret_references.add(str(secret_name))
+    for source in _value(container, "env_from", []) or []:
+        secret_name = _nested(source, "secret_ref", "name")
+        if secret_name:
+            secret_references.add(str(secret_name))
+    return {
+        "name": str(_value(container, "name", "") or ""),
+        "image": str(_value(container, "image", "") or ""),
+        "privileged": _value(security, "privileged"),
+        "allow_privilege_escalation": _value(security, "allow_privilege_escalation"),
+        "run_as_non_root": _value(security, "run_as_non_root"),
+        "read_only_root_filesystem": _value(security, "read_only_root_filesystem"),
+        "capabilities_add": tuple(str(item) for item in capabilities),
+        "seccomp_profile": _enum(_nested(security, "seccomp_profile", "type")),
+        "native_secret_references": tuple(sorted(secret_references)),
+    }
+
+
+def _workload(kind: str, item: Any) -> dict[str, Any]:
+    metadata = _value(item, "metadata")
+    spec = _value(item, "spec")
+    pod_spec = spec
+    if kind == "CronJob":
+        pod_spec = _nested(spec, "job_template", "spec", "template", "spec")
+    elif kind != "Pod":
+        pod_spec = _nested(spec, "template", "spec")
+    volumes = _value(pod_spec, "volumes", []) or []
+    host_paths = []
+    native_secret_references: set[str] = set()
+    csi_secret_provider_classes: set[str] = set()
+    for volume in volumes:
+        path = _nested(volume, "host_path", "path")
+        if path:
+            host_paths.append(str(path))
+        secret_name = _nested(volume, "secret", "secret_name")
+        if secret_name:
+            native_secret_references.add(str(secret_name))
+        for source in _nested(volume, "projected", "sources", default=[]) or []:
+            projected_secret_name = _nested(source, "secret", "name")
+            if projected_secret_name:
+                native_secret_references.add(str(projected_secret_name))
+        csi = _value(volume, "csi")
+        if str(_value(csi, "driver", "") or "") == "secrets-store.csi.k8s.io":
+            attributes = _value(csi, "volume_attributes", {}) or {}
+            provider_class = attributes.get("secretProviderClass") if isinstance(attributes, Mapping) else None
+            if provider_class:
+                csi_secret_provider_classes.add(str(provider_class))
+    containers = tuple(_container(container) for container in (_value(pod_spec, "containers", []) or []))
+    init_containers = tuple(_container(container) for container in (_value(pod_spec, "init_containers", []) or []))
+    for container in containers + init_containers:
+        native_secret_references.update(container["native_secret_references"])
+    return {
+        "kind": kind,
+        "namespace": str(_value(metadata, "namespace", "default") or "default"),
+        "name": str(_value(metadata, "name", "") or ""),
+        "service_account": str(_value(pod_spec, "service_account_name", "default") or "default"),
+        "automount_service_account_token": _value(pod_spec, "automount_service_account_token"),
+        "host_network": _value(pod_spec, "host_network"),
+        "host_pid": _value(pod_spec, "host_pid"),
+        "host_ipc": _value(pod_spec, "host_ipc"),
+        "host_paths": tuple(host_paths),
+        "native_secret_references": tuple(sorted(native_secret_references)),
+        "csi_secret_provider_classes": tuple(sorted(csi_secret_provider_classes)),
+        "containers": containers,
+        "init_containers": init_containers,
+    }
+
+
+class AksSecurityCollector:
+    """Collect AKS ARM, Defender, and Kubernetes API evidence without secrets."""
+
+    def __init__(
+        self,
+        credential: Any,
+        subscription_id: str,
+        *,
+        kubeconfig_path: str | None = None,
+        kubeconfig_contexts_path: str | None = None,
+    ) -> None:
+        self.credential = credential
+        self.subscription_id = subscription_id
+        self.kubeconfig_path = kubeconfig_path or os.environ.get(KUBECONFIG_ENV_VAR)
+        self.kubeconfig_contexts_path = kubeconfig_contexts_path or os.environ.get(KUBECONFIG_CONTEXTS_ENV_VAR)
+
+    def collect(self, clusters: Iterable[Any]) -> list[AksClusterEvidence]:
+        defender_enabled = self._defender_for_containers()
+        context_mapping, mapping_error = self._load_context_mapping()
+        return [
+            self._collect_cluster(cluster, defender_enabled, context_mapping, mapping_error) for cluster in clusters
+        ]
+
+    def _load_context_mapping(self) -> tuple[Mapping[str, str], str | None]:
+        if not self.kubeconfig_contexts_path:
+            return {}, "KUBECONFIG_CONTEXTS_NOT_CONFIGURED"
+        if not Path(self.kubeconfig_contexts_path).is_file():
+            return {}, "KUBECONFIG_CONTEXTS_NOT_FOUND"
+        try:
+            return load_kubeconfig_contexts(self.kubeconfig_contexts_path), None
+        except (OSError, ValueError, json.JSONDecodeError) as exc:
+            logger.warning("AKS kubeconfig context mapping is invalid: %s", type(exc).__name__)
+            return {}, "KUBECONFIG_CONTEXTS_INVALID"
+
+    def _defender_for_containers(self) -> bool | None:
+        try:
+            from azure.mgmt.security import SecurityCenter
+
+            pricing = SecurityCenter(self.credential, self.subscription_id).pricings.get("Containers")
+            return _enum(_value(pricing, "pricing_tier")).lower() == "standard"
+        except Exception as exc:
+            logger.warning("Defender for Containers evidence unavailable: %s", type(exc).__name__)
+            return None
+
+    def _control_plane(self, cluster: Any, defender_enabled: bool | None) -> dict[str, Any]:
+        properties = _value(cluster, "properties", cluster)
+        api_profile = _value(properties, "api_server_access_profile")
+        network_profile = _value(properties, "network_profile")
+        security_profile = _value(properties, "security_profile")
+        kms = _value(security_profile, "azure_key_vault_kms")
+        addons = _value(properties, "addon_profiles", {}) or {}
+        csi = addons.get("azureKeyvaultSecretsProvider") if isinstance(addons, Mapping) else None
+        csi_config = _value(csi, "config", {}) or {}
+        power_state = _enum(_nested(properties, "power_state", "code"))
+        provisioning_state = _enum(_value(properties, "provisioning_state"))
+        network_policy = _enum(_value(network_profile, "network_policy"))
+        if network_profile is not None and not network_policy:
+            network_policy = "none"
+        return {
+            "private_cluster_enabled": _value(api_profile, "enable_private_cluster"),
+            "authorized_ip_ranges": tuple(_value(api_profile, "authorized_ip_ranges", []) or []),
+            "network_policy": network_policy,
+            "defender_for_containers_enabled": defender_enabled,
+            "kms_enabled": _value(kms, "enabled", False) if kms is not None else False,
+            "kms_key_id": str(_value(kms, "key_id", "") or ""),
+            "csi_enabled": _value(csi, "enabled", False) if csi is not None else False,
+            "csi_rotation_enabled": str(csi_config.get("enableSecretRotation", "")).lower() == "true",
+            "power_state": power_state,
+            "provisioning_state": provisioning_state,
+        }
+
+    def _collect_cluster(
+        self,
+        cluster: Any,
+        defender_enabled: bool | None,
+        context_mapping: Mapping[str, str],
+        mapping_error: str | None,
+    ) -> AksClusterEvidence:
+        timestamp = datetime.now(timezone.utc).isoformat()
+        control_plane = self._control_plane(cluster, defender_enabled)
+        if control_plane["power_state"].lower() == "stopped":
+            return AksClusterEvidence(
+                cluster, "UNKNOWN", "ARM and Kubernetes API", timestamp, "CLUSTER_STOPPED", control_plane=control_plane
+            )
+        if control_plane["provisioning_state"].lower() in {"canceled", "deleting", "failed"}:
+            return AksClusterEvidence(
+                cluster,
+                "UNKNOWN",
+                "ARM and Kubernetes API",
+                timestamp,
+                "CLUSTER_STATE_UNSUPPORTED",
+                control_plane=control_plane,
+            )
+        if not self.kubeconfig_path:
+            return AksClusterEvidence(
+                cluster,
+                "UNKNOWN",
+                "ARM and Kubernetes API",
+                timestamp,
+                "KUBECONFIG_NOT_CONFIGURED",
+                control_plane=control_plane,
+            )
+        if not Path(self.kubeconfig_path).is_file():
+            return AksClusterEvidence(
+                cluster,
+                "UNKNOWN",
+                "ARM and Kubernetes API",
+                timestamp,
+                "KUBECONFIG_NOT_FOUND",
+                control_plane=control_plane,
+            )
+        if mapping_error:
+            return AksClusterEvidence(
+                cluster,
+                "UNKNOWN",
+                "ARM and Kubernetes API",
+                timestamp,
+                mapping_error,
+                control_plane=control_plane,
+            )
+        try:
+            cluster_id = _normalize_cluster_id(str(_value(cluster, "id", "") or ""))
+        except ValueError:
+            context = None
+        else:
+            context = context_mapping.get(cluster_id)
+        if not context:
+            return AksClusterEvidence(
+                cluster,
+                "UNKNOWN",
+                "ARM and Kubernetes API",
+                timestamp,
+                "KUBECONFIG_CONTEXT_UNRESOLVED",
+                control_plane=control_plane,
+            )
+        try:
+            return self._collect_kubernetes(cluster, context, control_plane, timestamp)
+        except KubeconfigContextUnresolvedError:
+            return AksClusterEvidence(
+                cluster,
+                "UNKNOWN",
+                "ARM and Kubernetes API",
+                timestamp,
+                "KUBECONFIG_CONTEXT_UNRESOLVED",
+                control_plane=control_plane,
+            )
+        except Exception as exc:
+            logger.warning(
+                "Kubernetes evidence unavailable for %s: %s",
+                _value(cluster, "name", "unknown"),
+                type(exc).__name__,
+            )
+            return AksClusterEvidence(
+                cluster,
+                "UNKNOWN",
+                "ARM and Kubernetes API",
+                timestamp,
+                "KUBERNETES_API_UNAVAILABLE",
+                control_plane=control_plane,
+            )
+
+    def _collect_kubernetes(
+        self,
+        cluster: Any,
+        context: str,
+        control_plane: Mapping[str, Any],
+        timestamp: str,
+    ) -> AksClusterEvidence:
+        from kubernetes import client, config
+
+        contexts, _ = config.list_kube_config_contexts(config_file=self.kubeconfig_path)
+        available_contexts = {str(item.get("name", "")) for item in contexts or () if isinstance(item, Mapping)}
+        if context not in available_contexts:
+            raise KubeconfigContextUnresolvedError("mapped kubeconfig context does not exist")
+        config.load_kube_config(config_file=self.kubeconfig_path, context=context)
+        core = client.CoreV1Api()
+        apps = client.AppsV1Api()
+        batch = client.BatchV1Api()
+        networking = client.NetworkingV1Api()
+        rbac = client.RbacAuthorizationV1Api()
+
+        try:
+            namespace_items = core.list_namespace().items
+        except Exception:
+            return AksClusterEvidence(
+                cluster,
+                "UNKNOWN",
+                "ARM and Kubernetes API",
+                timestamp,
+                "NAMESPACE_DISCOVERY_INCOMPLETE",
+                control_plane=control_plane,
+            )
+        namespaces = tuple(
+            str(_nested(item, "metadata", "name", default=""))
+            for item in namespace_items
+            if _nested(item, "metadata", "name")
+        )
+        eligible = tuple(name for name in namespaces if name not in SYSTEM_NAMESPACES)
+        partial: list[str] = []
+        policies: set[str] = set()
+        workloads: list[Mapping[str, Any]] = []
+        for namespace in eligible:
+            try:
+                if networking.list_namespaced_network_policy(namespace).items:
+                    policies.add(namespace)
+                workload_calls = (
+                    ("Deployment", apps.list_namespaced_deployment),
+                    ("StatefulSet", apps.list_namespaced_stateful_set),
+                    ("DaemonSet", apps.list_namespaced_daemon_set),
+                    ("Job", batch.list_namespaced_job),
+                    ("CronJob", batch.list_namespaced_cron_job),
+                    ("Pod", core.list_namespaced_pod),
+                )
+                for kind, call in workload_calls:
+                    for item in call(namespace).items:
+                        if kind == "Pod" and (_nested(item, "metadata", "owner_references", default=[]) or []):
+                            continue
+                        workloads.append(_workload(kind, item))
+            except Exception:
+                partial.append(namespace)
+
+        bindings = []
+        try:
+            for binding in rbac.list_cluster_role_binding().items:
+                if str(_nested(binding, "role_ref", "name", default="")).lower() != "cluster-admin":
+                    continue
+                metadata = _value(binding, "metadata")
+                for subject in _value(binding, "subjects", []) or []:
+                    bindings.append(
+                        {
+                            "binding": str(_value(metadata, "name", "") or ""),
+                            "kind": str(_value(subject, "kind", "") or ""),
+                            "name": str(_value(subject, "name", "") or ""),
+                            "namespace": str(_value(subject, "namespace", "") or ""),
+                        }
+                    )
+        except Exception:
+            partial.append("cluster-rbac")
+
+        status = "PARTIAL" if partial else "COMPLETE"
+        return AksClusterEvidence(
+            cluster=cluster,
+            status=status,
+            source="ARM, Microsoft Defender for Cloud, and Kubernetes API",
+            collected_at=timestamp,
+            partial_reasons=tuple(partial),
+            control_plane=control_plane,
+            namespaces=eligible,
+            network_policy_namespaces=tuple(sorted(policies)),
+            workloads=tuple(workloads),
+            cluster_admin_bindings=tuple(bindings),
+        )
diff --git a/scanner/azure_client.py b/scanner/azure_client.py
index dda5befa..8015a7fc 100644
--- a/scanner/azure_client.py
+++ b/scanner/azure_client.py
@@ -51,6 +51,7 @@ def __init__(self, subscription_id: str, credential: Optional[Any] = None) -> No
         self.subscription_id = subscription_id
         self.credential = credential or DefaultAzureCredential()
         self._managed_clusters_cache: Any = _UNSET
+        self._aks_security_posture_cache: Any = _UNSET
         self._function_apps_cache: Any = _UNSET
         self._private_endpoint_posture_cache: Any = _UNSET
         self._recovery_vaults_cache: Any = _UNSET
@@ -407,6 +408,19 @@ def get_managed_clusters(self) -> Optional[List[Any]]:
 
         return self._managed_clusters_cache
 
+    def get_aks_security_posture(self) -> Optional[List[Any]]:
+        """Collect cached ARM, Defender, and Kubernetes evidence for AKS rules."""
+        if self._aks_security_posture_cache is not _UNSET:
+            return self._aks_security_posture_cache
+        clusters = self.get_managed_clusters()
+        if clusters is None:
+            self._aks_security_posture_cache = None
+            return None
+        from scanner.aks_security import AksSecurityCollector
+
+        self._aks_security_posture_cache = AksSecurityCollector(self.credential, self.subscription_id).collect(clusters)
+        return self._aks_security_posture_cache
+
     # ------------------------------------------------------------------ #
     # Compute                                                               #
     # ------------------------------------------------------------------ #
diff --git a/scanner/rules/_aks_enterprise_common.py b/scanner/rules/_aks_enterprise_common.py
new file mode 100644
index 00000000..5ea219dd
--- /dev/null
+++ b/scanner/rules/_aks_enterprise_common.py
@@ -0,0 +1,492 @@
+"""Shared policy and evaluators for issue #255 AKS enterprise controls."""
+
+from __future__ import annotations
+
+import ipaddress
+import json
+import logging
+import os
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Mapping
+
+from scanner.rules._aks_common import finding, resource_identity
+
+logger = logging.getLogger(__name__)
+POLICY_ENV_VAR = "OPENSHIELD_AKS_SECURITY_POLICY"
+KUBERNETES_PERMISSIONS = [
+    "get,list namespaces",
+    "get,list workloads and pods",
+    "get,list networkpolicies.networking.k8s.io",
+    "get,list clusterrolebindings.rbac.authorization.k8s.io",
+]
+ARM_PERMISSIONS = [
+    "Microsoft.ContainerService/managedClusters/read",
+    "Microsoft.Security/pricings/read",
+]
+
+
+@dataclass(frozen=True)
+class AksSecurityPolicy:
+    approved_authorized_ip_ranges: frozenset[str]
+    trusted_registry_prefixes: tuple[str, ...]
+    allowed_cluster_admin_subjects: frozenset[str]
+    excluded_namespaces: frozenset[str]
+    require_image_digests: bool
+
+
+def value(item: Any, name: str, default: Any = None) -> Any:
+    if isinstance(item, Mapping):
+        return item.get(name, default)
+    return getattr(item, name, default)
+
+
+def canonical_cluster_admin_subject(kind: str, name: str, namespace: str = "") -> str:
+    normalized_kind = kind.strip().lower()
+    normalized_name = name.strip()
+    normalized_namespace = namespace.strip()
+    if not normalized_name:
+        raise ValueError("cluster-admin subject name must not be empty")
+    if normalized_kind == "serviceaccount":
+        if not normalized_namespace:
+            raise ValueError("cluster-admin ServiceAccount subjects must include a namespace")
+        return f"ServiceAccount:{normalized_namespace}:{normalized_name}"
+    if normalized_kind == "group":
+        return f"Group:{normalized_name}"
+    if normalized_kind == "user":
+        return f"User:{normalized_name}"
+    raise ValueError("cluster-admin subjects must be Groups, Users, or ServiceAccounts")
+
+
+def _policy_subject(value: str) -> str:
+    parts = [part.strip() for part in value.split(":")]
+    if len(parts) == 3 and parts[0].lower() == "serviceaccount":
+        return canonical_cluster_admin_subject(parts[0], parts[2], parts[1]).lower()
+    if len(parts) == 2:
+        return canonical_cluster_admin_subject(parts[0], parts[1]).lower()
+    raise ValueError("cluster-admin subjects must use Group:name, User:name, or ServiceAccount:namespace:name")
+
+
+def load_policy(path: str | Path) -> AksSecurityPolicy:
+    with Path(path).open(encoding="utf-8") as handle:
+        raw = json.load(handle)
+    required = {
+        "approved_authorized_ip_ranges",
+        "trusted_registry_prefixes",
+        "allowed_cluster_admin_subjects",
+        "excluded_namespaces",
+        "require_image_digests",
+    }
+    if not isinstance(raw, dict) or set(raw) != required:
+        raise ValueError("AKS security policy has missing or unsupported fields")
+    for key in required - {"require_image_digests"}:
+        if not isinstance(raw[key], list) or any(not isinstance(item, str) or not item.strip() for item in raw[key]):
+            raise ValueError(f"{key} must be a list of non-empty strings")
+    if not isinstance(raw["require_image_digests"], bool):
+        raise ValueError("require_image_digests must be boolean")
+    approved_ranges = set()
+    for item in raw["approved_authorized_ip_ranges"]:
+        approved_ranges.add(str(ipaddress.ip_network(item.strip(), strict=False)))
+    prefixes = tuple(item.strip().lower() for item in raw["trusted_registry_prefixes"])
+    if any("/" not in item for item in prefixes):
+        raise ValueError("trusted registry prefixes must include a registry and repository separator")
+    return AksSecurityPolicy(
+        approved_authorized_ip_ranges=frozenset(approved_ranges),
+        trusted_registry_prefixes=prefixes,
+        allowed_cluster_admin_subjects=frozenset(
+            _policy_subject(item) for item in raw["allowed_cluster_admin_subjects"]
+        ),
+        excluded_namespaces=frozenset(item.strip().lower() for item in raw["excluded_namespaces"]),
+        require_image_digests=raw["require_image_digests"],
+    )
+
+
+def policy_from_env(rule_id: str) -> AksSecurityPolicy | None:
+    path = os.environ.get(POLICY_ENV_VAR)
+    if not path:
+        logger.warning("%s: %s is not set; result is UNKNOWN", rule_id, POLICY_ENV_VAR)
+        return None
+    try:
+        return load_policy(path)
+    except Exception as exc:
+        logger.warning("%s: AKS security policy is invalid: %s; result is UNKNOWN", rule_id, exc)
+        return None
+
+
+def _metadata(
+    evidence: Any,
+    *,
+    namespace: str | None = None,
+    workload: str | None = None,
+    container: str | None = None,
+    image: str | None = None,
+    subject: str | None = None,
+    role: str | None = None,
+    observed: Any = None,
+    expected: Any = None,
+    permissions: list[str] | None = None,
+) -> dict[str, Any]:
+    cluster = value(evidence, "cluster")
+    _, cluster_name = resource_identity(cluster)
+    return {
+        "cluster": cluster_name,
+        "namespace": namespace,
+        "workload": workload,
+        "container": container,
+        "image": image,
+        "subject": subject,
+        "role": role,
+        "evidence_source": value(evidence, "source", "ARM and Kubernetes API"),
+        "evidence_collected_at": value(evidence, "collected_at"),
+        "observed_value": observed,
+        "expected_value": expected,
+        "permissions_required": permissions or ARM_PERMISSIONS + KUBERNETES_PERMISSIONS,
+        "confidence": "HIGH",
+        "unknown_reason": None,
+    }
+
+
+def _emit(module: Mapping[str, Any], evidence: Any, metadata: Mapping[str, Any]) -> dict[str, Any]:
+    cluster = value(evidence, "cluster")
+    complete_metadata = dict(metadata)
+    complete_metadata.update(
+        {
+            "evidence": {
+                "source": metadata.get("evidence_source"),
+                "collected_at": metadata.get("evidence_collected_at"),
+                "observed": metadata.get("observed_value"),
+                "expected": metadata.get("expected_value"),
+            },
+            "remediation": module["REMEDIATION"],
+            "severity": module["SEVERITY"],
+        }
+    )
+    result = finding(
+        cluster,
+        rule_id=module["RULE_ID"],
+        rule_name=module["RULE_NAME"],
+        severity=module["SEVERITY"],
+        category=module["CATEGORY"],
+        description=module["DESCRIPTION"],
+        remediation=module["REMEDIATION"],
+        playbook=module["PLAYBOOK"],
+        frameworks=module["FRAMEWORKS"],
+        metadata=complete_metadata,
+    )
+    namespace = metadata.get("namespace")
+    workload = metadata.get("workload")
+    if namespace and workload:
+        result["resource_id"] = f"{result['resource_id']}/namespaces/{namespace}/workloads/{workload}"
+        result["resource_name"] = f"{namespace}/{workload}"
+        result["resource_type"] = "Kubernetes/workloads"
+    elif namespace:
+        result["resource_id"] = f"{result['resource_id']}/namespaces/{namespace}"
+        result["resource_name"] = namespace
+        result["resource_type"] = "Kubernetes/namespaces"
+    return result
+
+
+def _workloads(evidence: Any, policy: AksSecurityPolicy | None) -> list[Mapping[str, Any]]:
+    result = []
+    for workload in value(evidence, "workloads", ()) or ():
+        namespace = str(value(workload, "namespace", "default") or "default")
+        if policy and namespace.lower() in policy.excluded_namespaces:
+            continue
+        result.append(workload)
+    return result
+
+
+def _normalize_image_registry(image: str) -> str:
+    normalized = image.lower()
+    registry = normalized.split("/", 1)[0]
+    if "/" not in normalized or not ("." in registry or ":" in registry or registry == "localhost"):
+        return f"docker.io/{normalized}"
+    return normalized
+
+
+def _matches_trusted_registry_prefix(image: str, prefix: str) -> bool:
+    """Match a registry/repository prefix without crossing a repository boundary."""
+    if prefix.endswith("/"):
+        return image.startswith(prefix)
+    if image == prefix:
+        return True
+    # Tags, digests, and child repositories are valid continuations; a raw
+    # string continuation such as ``team-evil`` must not inherit trust.
+    return any(image.startswith(f"{prefix}{separator}") for separator in ("/", ":", "@"))
+
+
+def scan_control(azure_client: Any, module: Mapping[str, Any], control: str) -> list[dict[str, Any]]:
+    rule_id = module["RULE_ID"]
+    evidence_items = azure_client.get_aks_security_posture()
+    if evidence_items is None:
+        logger.warning("%s: AKS inventory unavailable; result is UNKNOWN", rule_id)
+        return []
+    if not evidence_items:
+        logger.info("%s: no AKS clusters exist; result is NOT_APPLICABLE", rule_id)
+        return []
+    policy_required = control in {
+        "api_restrictions",
+        "cluster_admin",
+        "untrusted_registry",
+        "mutable_image",
+    }
+    policy = policy_from_env(rule_id) if policy_required else None
+    if policy_required and policy is None:
+        return []
+    approved_networks = (
+        [ipaddress.ip_network(item) for item in policy.approved_authorized_ip_ranges]
+        if control == "api_restrictions" and policy
+        else []
+    )
+    findings: list[dict[str, Any]] = []
+    for evidence in evidence_items:
+        cluster = value(evidence, "cluster")
+        resource_id, resource_name = resource_identity(cluster)
+        if not resource_id or not resource_name:
+            logger.warning("%s: malformed cluster identity; result is UNKNOWN", rule_id)
+            continue
+        cp = value(evidence, "control_plane", {}) or {}
+
+        if control == "api_restrictions":
+            private = value(cp, "private_cluster_enabled")
+            ranges = tuple(value(cp, "authorized_ip_ranges", ()) or ())
+            if private is True:
+                continue
+            if private is None:
+                logger.warning("%s: API exposure is unknown for %s", rule_id, resource_name)
+                continue
+            normalized: set[ipaddress.IPv4Network | ipaddress.IPv6Network] = set()
+            try:
+                normalized = {ipaddress.ip_network(item, strict=False) for item in ranges}
+            except ValueError:
+                logger.warning("%s: malformed authorized IP evidence for %s", rule_id, resource_name)
+                continue
+            if normalized and all(
+                any(actual.version == allowed.version and actual.subnet_of(allowed) for allowed in approved_networks)
+                for actual in normalized
+            ):
+                continue
+            findings.append(
+                _emit(
+                    module,
+                    evidence,
+                    _metadata(
+                        evidence,
+                        observed={"private": False, "authorized_ip_ranges": sorted(str(item) for item in normalized)},
+                        expected="private API server or only approved IP ranges",
+                        permissions=ARM_PERMISSIONS,
+                    ),
+                )
+            )
+            continue
+
+        if control in {"network_policy", "defender", "csi_rotation"}:
+            field = {
+                "network_policy": "network_policy",
+                "defender": "defender_for_containers_enabled",
+                "csi_rotation": "csi_rotation_enabled",
+            }.get(control)
+            if control == "network_policy":
+                observed = str(value(cp, field, "") or "")
+                if observed.lower() in {"azure", "calico", "cilium"}:
+                    continue
+                if not observed:
+                    logger.warning("%s: network policy evidence is unknown for %s", rule_id, resource_name)
+                    continue
+                expected = "Azure, Calico, or Cilium network policy"
+            elif control == "csi_rotation":
+                csi_enabled = value(cp, "csi_enabled")
+                if csi_enabled is not True:
+                    logger.info(
+                        "%s: CSI provider is not enabled for %s; result is NOT_APPLICABLE", rule_id, resource_name
+                    )
+                    continue
+                observed = value(cp, field)
+                if observed is True:
+                    continue
+                if observed is None:
+                    logger.warning("%s: CSI rotation evidence is unknown for %s", rule_id, resource_name)
+                    continue
+                expected = True
+            else:
+                observed = value(cp, field)
+                if observed is True:
+                    continue
+                if observed is None:
+                    logger.warning("%s: Defender evidence is unknown for %s", rule_id, resource_name)
+                    continue
+                expected = True
+            findings.append(
+                _emit(
+                    module,
+                    evidence,
+                    _metadata(evidence, observed=observed, expected=expected, permissions=ARM_PERMISSIONS),
+                )
+            )
+            continue
+
+        status = str(value(evidence, "status", "UNKNOWN") or "UNKNOWN").upper()
+        if status == "UNKNOWN":
+            logger.warning("%s: Kubernetes evidence unavailable for %s; result is UNKNOWN", rule_id, resource_name)
+            continue
+        workloads = _workloads(evidence, policy)
+        if control == "secret_protection":
+            if value(cp, "kms_enabled") is True:
+                continue
+            referenced = [
+                workload for workload in workloads if tuple(value(workload, "native_secret_references", ()) or ())
+            ]
+            if not referenced:
+                logger.info(
+                    "%s: no in-scope references for %s; result is NOT_APPLICABLE",
+                    rule_id,
+                    resource_name,
+                )
+                continue
+            for workload in referenced:
+                findings.append(
+                    _workload_finding(
+                        module,
+                        evidence,
+                        workload,
+                        None,
+                        {
+                            "kms_enabled": False,
+                            "kms_key_id": str(value(cp, "kms_key_id", "") or ""),
+                            "csi_enabled": value(cp, "csi_enabled"),
+                            "native_secret_references": list(value(workload, "native_secret_references", ()) or ()),
+                            "csi_secret_provider_classes": list(
+                                value(workload, "csi_secret_provider_classes", ()) or ()
+                            ),
+                        },
+                        "KMS-backed native Secrets or CSI-only secret references",
+                    )
+                )
+            continue
+        if (
+            control
+            in {
+                "privileged",
+                "host_network",
+                "host_pid",
+                "host_ipc",
+                "host_path",
+                "untrusted_registry",
+                "latest_image",
+                "mutable_image",
+            }
+            and not workloads
+        ):
+            logger.info("%s: no eligible workloads for %s; result is NOT_APPLICABLE", rule_id, resource_name)
+            continue
+
+        if control == "network_policy_namespaces":
+            discovered = set(value(evidence, "namespaces", ()) or ())
+            protected = set(value(evidence, "network_policy_namespaces", ()) or ())
+            failed = set(value(evidence, "partial_reasons", ()) or ())
+            excluded = policy.excluded_namespaces if policy else frozenset()
+            for namespace in sorted(name for name in discovered - protected - failed if name.lower() not in excluded):
+                findings.append(
+                    _emit(
+                        module,
+                        evidence,
+                        _metadata(
+                            evidence,
+                            namespace=namespace,
+                            observed="no NetworkPolicy",
+                            expected="one or more NetworkPolicies",
+                        ),
+                    )
+                )
+        elif control == "privileged":
+            for workload in workloads:
+                containers = tuple(value(workload, "containers", ()) or ()) + tuple(
+                    value(workload, "init_containers", ()) or ()
+                )
+                for container in containers:
+                    if value(container, "privileged") is True:
+                        findings.append(_workload_finding(module, evidence, workload, container, True, False))
+        elif control in {"host_network", "host_pid", "host_ipc"}:
+            for workload in workloads:
+                if value(workload, control) is True:
+                    findings.append(_workload_finding(module, evidence, workload, None, True, False))
+        elif control == "host_path":
+            for workload in workloads:
+                paths = tuple(value(workload, "host_paths", ()) or ())
+                if paths:
+                    findings.append(
+                        _workload_finding(module, evidence, workload, None, list(paths), "no hostPath volumes")
+                    )
+        elif control == "cluster_admin":
+            for binding in value(evidence, "cluster_admin_bindings", ()) or ():
+                try:
+                    subject = canonical_cluster_admin_subject(
+                        str(value(binding, "kind", "") or ""),
+                        str(value(binding, "name", "") or ""),
+                        str(value(binding, "namespace", "") or ""),
+                    )
+                except ValueError:
+                    logger.warning("%s: ambiguous cluster-admin subject evidence for %s", rule_id, resource_name)
+                    continue
+                if subject.lower() in policy.allowed_cluster_admin_subjects:
+                    continue
+                metadata = _metadata(
+                    evidence,
+                    subject=subject,
+                    role="cluster-admin",
+                    observed=value(binding, "binding"),
+                    expected=sorted(policy.allowed_cluster_admin_subjects),
+                )
+                findings.append(_emit(module, evidence, metadata))
+        elif control in {"untrusted_registry", "latest_image", "mutable_image"}:
+            for workload in workloads:
+                containers = tuple(value(workload, "containers", ()) or ()) + tuple(
+                    value(workload, "init_containers", ()) or ()
+                )
+                for container in containers:
+                    image = str(value(container, "image", "") or "")
+                    if not image:
+                        logger.warning("%s: image evidence missing for %s", rule_id, value(workload, "name"))
+                        continue
+                    normalized = _normalize_image_registry(image)
+                    trusted = control != "untrusted_registry" or any(
+                        _matches_trusted_registry_prefix(normalized, prefix)
+                        for prefix in policy.trusted_registry_prefixes
+                    )
+                    digest = "@sha256:" in normalized
+                    latest = normalized.endswith(":latest") or (":" not in normalized.rsplit("/", 1)[-1] and not digest)
+                    violates = {
+                        "untrusted_registry": not trusted,
+                        "latest_image": latest,
+                        "mutable_image": bool(policy and policy.require_image_digests and not digest and not latest),
+                    }[control]
+                    if not violates:
+                        continue
+                    observed = {"trusted_registry": trusted, "latest_tag": latest, "digest_pinned": digest}
+                    expected = {
+                        "untrusted_registry": "approved registry prefix",
+                        "latest_image": "explicit non-latest image reference",
+                        "mutable_image": "sha256 digest-pinned image",
+                    }[control]
+                    findings.append(_workload_finding(module, evidence, workload, container, observed, expected))
+    return findings
+
+
+def _workload_finding(
+    module: Mapping[str, Any], evidence: Any, workload: Mapping[str, Any], container: Any, observed: Any, expected: Any
+) -> dict[str, Any]:
+    namespace = str(value(workload, "namespace", "default") or "default")
+    name = str(value(workload, "name", "") or "")
+    return _emit(
+        module,
+        evidence,
+        _metadata(
+            evidence,
+            namespace=namespace,
+            workload=f"{value(workload, 'kind', 'Workload')}/{name}",
+            container=str(value(container, "name", "") or "") or None,
+            image=str(value(container, "image", "") or "") or None,
+            observed=observed,
+            expected=expected,
+        ),
+    )
diff --git a/scanner/rules/az_aks_007.py b/scanner/rules/az_aks_007.py
new file mode 100644
index 00000000..ee376312
--- /dev/null
+++ b/scanner/rules/az_aks_007.py
@@ -0,0 +1,20 @@
+"""AZ-AKS-007: AKS API server lacks approved access restrictions."""
+
+from typing import Any, Dict, List
+
+from scanner.rules._aks_enterprise_common import scan_control
+
+RULE_ID = "AZ-AKS-007"
+RULE_NAME = "AKS API Server Lacks Approved IP Restrictions"
+SEVERITY = "HIGH"
+CATEGORY = "Kubernetes"
+FRAMEWORKS = {"CIS": "N/A-AKS-007", "NIST": "PR.AC-3", "ISO27001": "A.13.1.1", "SOC2": "CC6.6"}
+DESCRIPTION = "A public AKS API server accepts traffic from sources outside the organization-approved IP ranges."
+REMEDIATION = (
+    "Use a private cluster or restrict API server authorized IP ranges to the approved administration networks."
+)
+PLAYBOOK = "playbooks/cli/fix_az_aks_007.sh"
+
+
+def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]:
+    return scan_control(azure_client, globals(), "api_restrictions")
diff --git a/scanner/rules/az_aks_008.py b/scanner/rules/az_aks_008.py
new file mode 100644
index 00000000..284ed1d7
--- /dev/null
+++ b/scanner/rules/az_aks_008.py
@@ -0,0 +1,18 @@
+"""AZ-AKS-008: AKS network policy is not configured."""
+
+from typing import Any, Dict, List
+
+from scanner.rules._aks_enterprise_common import scan_control
+
+RULE_ID = "AZ-AKS-008"
+RULE_NAME = "AKS Cluster Has No Kubernetes Network Policy"
+SEVERITY = "HIGH"
+CATEGORY = "Kubernetes"
+FRAMEWORKS = {"CIS": "N/A-AKS-008", "NIST": "PR.AC-5", "ISO27001": "A.13.1.3", "SOC2": "CC6.6"}
+DESCRIPTION = "The AKS cluster has no supported network policy engine to enforce workload traffic boundaries."
+REMEDIATION = "Enable Azure, Calico, or Cilium network policy using a tested cluster upgrade or migration plan."
+PLAYBOOK = "playbooks/cli/fix_az_aks_008.sh"
+
+
+def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]:
+    return scan_control(azure_client, globals(), "network_policy")
diff --git a/scanner/rules/az_aks_009.py b/scanner/rules/az_aks_009.py
new file mode 100644
index 00000000..143c085b
--- /dev/null
+++ b/scanner/rules/az_aks_009.py
@@ -0,0 +1,20 @@
+"""AZ-AKS-009: Eligible namespace has no Kubernetes NetworkPolicy."""
+
+from typing import Any, Dict, List
+
+from scanner.rules._aks_enterprise_common import scan_control
+
+RULE_ID = "AZ-AKS-009"
+RULE_NAME = "Kubernetes Namespace Has No NetworkPolicy"
+SEVERITY = "HIGH"
+CATEGORY = "Kubernetes"
+FRAMEWORKS = {"CIS": "N/A-AKS-009", "NIST": "PR.AC-5", "ISO27001": "A.13.1.3", "SOC2": "CC6.6"}
+DESCRIPTION = "An eligible application namespace contains no NetworkPolicy and has no workload traffic boundary."
+REMEDIATION = (
+    "Apply tested default-deny ingress and egress policies, then add explicit workload communication allowances."
+)
+PLAYBOOK = "playbooks/cli/fix_az_aks_009.sh"
+
+
+def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]:
+    return scan_control(azure_client, globals(), "network_policy_namespaces")
diff --git a/scanner/rules/az_aks_010.py b/scanner/rules/az_aks_010.py
new file mode 100644
index 00000000..5779c647
--- /dev/null
+++ b/scanner/rules/az_aks_010.py
@@ -0,0 +1,18 @@
+"""AZ-AKS-010: Defender for Containers is disabled."""
+
+from typing import Any, Dict, List
+
+from scanner.rules._aks_enterprise_common import scan_control
+
+RULE_ID = "AZ-AKS-010"
+RULE_NAME = "Defender for Containers Protection Disabled"
+SEVERITY = "HIGH"
+CATEGORY = "Kubernetes"
+FRAMEWORKS = {"CIS": "2.1.8", "NIST": "DE.CM-8", "ISO27001": "A.12.6.1", "SOC2": "CC7.1"}
+DESCRIPTION = "Microsoft Defender for Containers is disabled, removing managed threat detection for in-scope clusters."
+REMEDIATION = "Enable the Containers Defender plan at subscription scope and validate sensor deployment."
+PLAYBOOK = "playbooks/cli/fix_az_aks_010.sh"
+
+
+def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]:
+    return scan_control(azure_client, globals(), "defender")
diff --git a/scanner/rules/az_aks_011.py b/scanner/rules/az_aks_011.py
new file mode 100644
index 00000000..6e4da141
--- /dev/null
+++ b/scanner/rules/az_aks_011.py
@@ -0,0 +1,20 @@
+"""AZ-AKS-011: AKS secrets lack Key Vault or KMS-backed protection."""
+
+from typing import Any, Dict, List
+
+from scanner.rules._aks_enterprise_common import scan_control
+
+RULE_ID = "AZ-AKS-011"
+RULE_NAME = "AKS Secrets Lack Key Vault or KMS-Backed Protection"
+SEVERITY = "HIGH"
+CATEGORY = "Kubernetes"
+FRAMEWORKS = {"CIS": "N/A-AKS-011", "NIST": "PR.DS-1", "ISO27001": "A.10.1.1", "SOC2": "CC6.1"}
+DESCRIPTION = "A workload references native Kubernetes Secrets while Azure Key Vault KMS encryption is disabled."
+REMEDIATION = (
+    "Enable Key Vault KMS for Kubernetes secrets or the Key Vault Secrets Store CSI provider with workload identity."
+)
+PLAYBOOK = "playbooks/cli/fix_az_aks_011.sh"
+
+
+def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]:
+    return scan_control(azure_client, globals(), "secret_protection")
diff --git a/scanner/rules/az_aks_012.py b/scanner/rules/az_aks_012.py
new file mode 100644
index 00000000..a478f4d9
--- /dev/null
+++ b/scanner/rules/az_aks_012.py
@@ -0,0 +1,20 @@
+"""AZ-AKS-012: Secrets Store CSI rotation is disabled."""
+
+from typing import Any, Dict, List
+
+from scanner.rules._aks_enterprise_common import scan_control
+
+RULE_ID = "AZ-AKS-012"
+RULE_NAME = "Secrets Store CSI Secret Rotation Disabled"
+SEVERITY = "MEDIUM"
+CATEGORY = "Kubernetes"
+FRAMEWORKS = {"CIS": "N/A-AKS-012", "NIST": "PR.AC-1", "ISO27001": "A.9.4.3", "SOC2": "CC6.1"}
+DESCRIPTION = "The enabled Key Vault Secrets Store CSI provider does not automatically rotate mounted secret material."
+REMEDIATION = (
+    "Enable secret rotation for the AKS Key Vault Secrets Store CSI provider and test application reload behavior."
+)
+PLAYBOOK = "playbooks/cli/fix_az_aks_012.sh"
+
+
+def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]:
+    return scan_control(azure_client, globals(), "csi_rotation")
diff --git a/scanner/rules/az_aks_013.py b/scanner/rules/az_aks_013.py
new file mode 100644
index 00000000..3accd01c
--- /dev/null
+++ b/scanner/rules/az_aks_013.py
@@ -0,0 +1,18 @@
+"""AZ-AKS-013: Workload permits privileged containers."""
+
+from typing import Any, Dict, List
+
+from scanner.rules._aks_enterprise_common import scan_control
+
+RULE_ID = "AZ-AKS-013"
+RULE_NAME = "Kubernetes Workload Permits Privileged Containers"
+SEVERITY = "CRITICAL"
+CATEGORY = "Kubernetes"
+FRAMEWORKS = {"CIS": "N/A-AKS-013", "NIST": "PR.AC-4", "ISO27001": "A.9.4.1", "SOC2": "CC6.1"}
+DESCRIPTION = "A workload explicitly enables privileged mode for an application or init container."
+REMEDIATION = "Remove privileged mode and grant only the specific Linux capabilities required by the workload."
+PLAYBOOK = "playbooks/cli/fix_az_aks_013.sh"
+
+
+def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]:
+    return scan_control(azure_client, globals(), "privileged")
diff --git a/scanner/rules/az_aks_014.py b/scanner/rules/az_aks_014.py
new file mode 100644
index 00000000..ee6d6356
--- /dev/null
+++ b/scanner/rules/az_aks_014.py
@@ -0,0 +1,18 @@
+"""AZ-AKS-014: Workload shares the host network namespace."""
+
+from typing import Any, Dict, List
+
+from scanner.rules._aks_enterprise_common import scan_control
+
+RULE_ID = "AZ-AKS-014"
+RULE_NAME = "Kubernetes Workload Uses Host Network"
+SEVERITY = "HIGH"
+CATEGORY = "Kubernetes"
+FRAMEWORKS = {"CIS": "N/A-AKS-014", "NIST": "PR.AC-5", "ISO27001": "A.13.1.3", "SOC2": "CC6.6"}
+DESCRIPTION = "A workload explicitly shares the Kubernetes node network namespace."
+REMEDIATION = "Disable hostNetwork and use Kubernetes Services or an approved network integration."
+PLAYBOOK = "playbooks/cli/fix_az_aks_014.sh"
+
+
+def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]:
+    return scan_control(azure_client, globals(), "host_network")
diff --git a/scanner/rules/az_aks_015.py b/scanner/rules/az_aks_015.py
new file mode 100644
index 00000000..6cceea01
--- /dev/null
+++ b/scanner/rules/az_aks_015.py
@@ -0,0 +1,18 @@
+"""AZ-AKS-015: Workload shares the host PID namespace."""
+
+from typing import Any, Dict, List
+
+from scanner.rules._aks_enterprise_common import scan_control
+
+RULE_ID = "AZ-AKS-015"
+RULE_NAME = "Kubernetes Workload Uses Host PID Namespace"
+SEVERITY = "HIGH"
+CATEGORY = "Kubernetes"
+FRAMEWORKS = {"CIS": "N/A-AKS-015", "NIST": "PR.AC-4", "ISO27001": "A.9.4.1", "SOC2": "CC6.1"}
+DESCRIPTION = "A workload explicitly shares the Kubernetes node process namespace."
+REMEDIATION = "Disable hostPID and use scoped observability or management interfaces instead of node process access."
+PLAYBOOK = "playbooks/cli/fix_az_aks_015.sh"
+
+
+def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]:
+    return scan_control(azure_client, globals(), "host_pid")
diff --git a/scanner/rules/az_aks_016.py b/scanner/rules/az_aks_016.py
new file mode 100644
index 00000000..38acda25
--- /dev/null
+++ b/scanner/rules/az_aks_016.py
@@ -0,0 +1,18 @@
+"""AZ-AKS-016: Workload shares the host IPC namespace."""
+
+from typing import Any, Dict, List
+
+from scanner.rules._aks_enterprise_common import scan_control
+
+RULE_ID = "AZ-AKS-016"
+RULE_NAME = "Kubernetes Workload Uses Host IPC Namespace"
+SEVERITY = "HIGH"
+CATEGORY = "Kubernetes"
+FRAMEWORKS = {"CIS": "N/A-AKS-016", "NIST": "PR.AC-4", "ISO27001": "A.9.4.1", "SOC2": "CC6.1"}
+DESCRIPTION = "A workload explicitly shares the Kubernetes node inter-process communication namespace."
+REMEDIATION = "Disable hostIPC and use scoped application communication mechanisms."
+PLAYBOOK = "playbooks/cli/fix_az_aks_016.sh"
+
+
+def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]:
+    return scan_control(azure_client, globals(), "host_ipc")
diff --git a/scanner/rules/az_aks_017.py b/scanner/rules/az_aks_017.py
new file mode 100644
index 00000000..ba4b78cd
--- /dev/null
+++ b/scanner/rules/az_aks_017.py
@@ -0,0 +1,18 @@
+"""AZ-AKS-017: Workload mounts hostPath volumes."""
+
+from typing import Any, Dict, List
+
+from scanner.rules._aks_enterprise_common import scan_control
+
+RULE_ID = "AZ-AKS-017"
+RULE_NAME = "Kubernetes Workload Uses Unrestricted HostPath Volumes"
+SEVERITY = "HIGH"
+CATEGORY = "Kubernetes"
+FRAMEWORKS = {"CIS": "N/A-AKS-017", "NIST": "PR.AC-4", "ISO27001": "A.9.4.1", "SOC2": "CC6.1"}
+DESCRIPTION = "A workload mounts a path from the Kubernetes node filesystem."
+REMEDIATION = "Replace hostPath with a scoped persistent volume or an approved CSI volume and enforce admission policy."
+PLAYBOOK = "playbooks/cli/fix_az_aks_017.sh"
+
+
+def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]:
+    return scan_control(azure_client, globals(), "host_path")
diff --git a/scanner/rules/az_aks_018.py b/scanner/rules/az_aks_018.py
new file mode 100644
index 00000000..ed372094
--- /dev/null
+++ b/scanner/rules/az_aks_018.py
@@ -0,0 +1,22 @@
+"""AZ-AKS-018: cluster-admin access is assigned too broadly."""
+
+from typing import Any, Dict, List
+
+from scanner.rules._aks_enterprise_common import scan_control
+
+RULE_ID = "AZ-AKS-018"
+RULE_NAME = "Kubernetes Cluster Admin Access Assigned Too Broadly"
+SEVERITY = "CRITICAL"
+CATEGORY = "Kubernetes"
+FRAMEWORKS = {"CIS": "N/A-AKS-018", "NIST": "PR.AC-4", "ISO27001": "A.9.2.3", "SOC2": "CC6.3"}
+DESCRIPTION = (
+    "A ClusterRoleBinding grants cluster-admin to a subject outside the approved platform administrator allowlist."
+)
+REMEDIATION = (
+    "Remove the broad binding and replace it with namespace-scoped, least-privilege roles for approved groups."
+)
+PLAYBOOK = "playbooks/cli/fix_az_aks_018.sh"
+
+
+def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]:
+    return scan_control(azure_client, globals(), "cluster_admin")
diff --git a/scanner/rules/az_aks_019.py b/scanner/rules/az_aks_019.py
new file mode 100644
index 00000000..7fe5b7c4
--- /dev/null
+++ b/scanner/rules/az_aks_019.py
@@ -0,0 +1,18 @@
+"""AZ-AKS-019: Workload uses an untrusted registry."""
+
+from typing import Any, Dict, List
+
+from scanner.rules._aks_enterprise_common import scan_control
+
+RULE_ID = "AZ-AKS-019"
+RULE_NAME = "Kubernetes Workload Uses Untrusted Registry"
+SEVERITY = "HIGH"
+CATEGORY = "Kubernetes"
+FRAMEWORKS = {"CIS": "N/A-AKS-019", "NIST": "PR.DS-6", "ISO27001": "A.14.2.5", "SOC2": "CC8.1"}
+DESCRIPTION = "A workload pulls an image from a registry outside the organization-approved trust prefixes."
+REMEDIATION = "Mirror the image into an approved registry, verify it, and update the workload reference."
+PLAYBOOK = "playbooks/cli/fix_az_aks_019.sh"
+
+
+def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]:
+    return scan_control(azure_client, globals(), "untrusted_registry")
diff --git a/scanner/rules/az_aks_020.py b/scanner/rules/az_aks_020.py
new file mode 100644
index 00000000..fc33a0ba
--- /dev/null
+++ b/scanner/rules/az_aks_020.py
@@ -0,0 +1,18 @@
+"""AZ-AKS-020: Workload uses latest or an implicit latest tag."""
+
+from typing import Any, Dict, List
+
+from scanner.rules._aks_enterprise_common import scan_control
+
+RULE_ID = "AZ-AKS-020"
+RULE_NAME = "Kubernetes Workload Uses Latest Image Tag"
+SEVERITY = "HIGH"
+CATEGORY = "Kubernetes"
+FRAMEWORKS = {"CIS": "N/A-AKS-020", "NIST": "PR.IP-1", "ISO27001": "A.12.1.2", "SOC2": "CC8.1"}
+DESCRIPTION = "A workload uses the latest tag or omits a tag, allowing deployments to change without a manifest change."
+REMEDIATION = "Replace latest or implicit latest with an approved version and preferably a verified sha256 digest."
+PLAYBOOK = "playbooks/cli/fix_az_aks_020.sh"
+
+
+def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]:
+    return scan_control(azure_client, globals(), "latest_image")
diff --git a/scanner/rules/az_aks_021.py b/scanner/rules/az_aks_021.py
new file mode 100644
index 00000000..62731108
--- /dev/null
+++ b/scanner/rules/az_aks_021.py
@@ -0,0 +1,18 @@
+"""AZ-AKS-021: Workload uses a mutable image reference."""
+
+from typing import Any, Dict, List
+
+from scanner.rules._aks_enterprise_common import scan_control
+
+RULE_ID = "AZ-AKS-021"
+RULE_NAME = "Kubernetes Workload Image Is Not Digest Pinned"
+SEVERITY = "MEDIUM"
+CATEGORY = "Kubernetes"
+FRAMEWORKS = {"CIS": "N/A-AKS-021", "NIST": "PR.DS-6", "ISO27001": "A.14.2.5", "SOC2": "CC8.1"}
+DESCRIPTION = "A workload image uses a mutable tag instead of a verified sha256 digest required by policy."
+REMEDIATION = "Resolve the approved image digest and pin the workload image to its sha256 reference."
+PLAYBOOK = "playbooks/cli/fix_az_aks_021.sh"
+
+
+def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]:
+    return scan_control(azure_client, globals(), "mutable_image")
diff --git a/tests/helpers/mock_azure.py b/tests/helpers/mock_azure.py
index 5b8b859f..2422abef 100644
--- a/tests/helpers/mock_azure.py
+++ b/tests/helpers/mock_azure.py
@@ -86,6 +86,7 @@ def __init__(self) -> None:
         self._dns_record_sets: Dict[Tuple[str, str], List[Any]] = {}
         self._web_apps: List[Any] = []
         self._managed_clusters: Optional[List[Any]] = []
+        self._aks_security_posture: Optional[List[Any]] = []
         self._applications: Optional[List[Dict[str, Any]]] = []
         self._managed_identity_principals: Optional[List[Dict[str, Any]]] = []
         self._subscription_role_assignments: Optional[List[Any]] = []
@@ -132,6 +133,14 @@ def set_managed_clusters(self, clusters: Optional[List[Any]]) -> "MockAzureClien
     def get_managed_clusters(self) -> Optional[List[Any]]:
         return self._managed_clusters
 
+    def set_aks_security_posture(self, posture: Optional[List[Any]]) -> "MockAzureClient":
+        """Configure AKS enterprise evidence; ``None`` represents collection failure."""
+        self._aks_security_posture = posture
+        return self
+
+    def get_aks_security_posture(self) -> Optional[List[Any]]:
+        return self._aks_security_posture
+
     def set_applications(self, applications: Optional[List[Dict[str, Any]]]) -> "MockAzureClient":
         self._applications = applications
         return self
diff --git a/tests/test_aks_security_collector.py b/tests/test_aks_security_collector.py
new file mode 100644
index 00000000..565b632d
--- /dev/null
+++ b/tests/test_aks_security_collector.py
@@ -0,0 +1,326 @@
+"""Unit tests for failure-aware AKS and Kubernetes evidence collection."""
+
+import json
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from scanner.aks_security import AksSecurityCollector, load_kubeconfig_contexts
+
+PAYMENTS_CLUSTER_ID = (
+    "/subscriptions/sub/resourceGroups/payments/providers/Microsoft.ContainerService/managedClusters/shared"
+)
+ORDERS_CLUSTER_ID = (
+    "/subscriptions/sub/resourceGroups/orders/providers/Microsoft.ContainerService/managedClusters/shared"
+)
+
+
+def ns(**kwargs):
+    return SimpleNamespace(**kwargs)
+
+
+def cluster(**overrides):
+    values = {
+        "id": "/subscriptions/sub/resourceGroups/rg/providers/Microsoft.ContainerService/managedClusters/aks-1",
+        "name": "aks-1",
+        "properties": ns(
+            api_server_access_profile=ns(enable_private_cluster=False, authorized_ip_ranges=["203.0.113.0/24"]),
+            network_profile=ns(network_policy="cilium"),
+            security_profile=ns(azure_key_vault_kms=ns(enabled=True, key_id="key-id")),
+            addon_profiles={"azureKeyvaultSecretsProvider": ns(enabled=True, config={"enableSecretRotation": "true"})},
+            power_state=ns(code="Running"),
+            provisioning_state="Succeeded",
+        ),
+    }
+    values.update(overrides)
+    return ns(**values)
+
+
+def test_stopped_cluster_is_explicitly_unknown():
+    stopped = cluster()
+    stopped.properties.power_state.code = "Stopped"
+    collector = AksSecurityCollector(MagicMock(), "sub", kubeconfig_path="/unused")
+
+    with patch.object(collector, "_defender_for_containers", return_value=True):
+        result = collector.collect([stopped])[0]
+
+    assert result.status == "UNKNOWN"
+    assert result.unknown_reason == "CLUSTER_STOPPED"
+    assert result.control_plane["network_policy"] == "cilium"
+
+
+def test_unsupported_cluster_state_is_explicitly_unknown():
+    failed = cluster()
+    failed.properties.provisioning_state = "Failed"
+    collector = AksSecurityCollector(MagicMock(), "sub", kubeconfig_path="/unused")
+
+    with patch.object(collector, "_defender_for_containers", return_value=True):
+        result = collector.collect([failed])[0]
+
+    assert result.status == "UNKNOWN"
+    assert result.unknown_reason == "CLUSTER_STATE_UNSUPPORTED"
+
+
+def test_missing_kubeconfig_is_explicitly_unknown():
+    collector = AksSecurityCollector(MagicMock(), "sub", kubeconfig_path="/does/not/exist")
+
+    with patch.object(collector, "_defender_for_containers", return_value=False):
+        result = collector.collect([cluster()])[0]
+
+    assert result.status == "UNKNOWN"
+    assert result.unknown_reason == "KUBECONFIG_NOT_FOUND"
+    assert result.control_plane["defender_for_containers_enabled"] is False
+
+
+@patch("azure.mgmt.security.SecurityCenter")
+def test_defender_for_containers_preserves_success_and_failure(security_type):
+    collector = AksSecurityCollector(MagicMock(), "sub")
+    security_type.return_value.pricings.get.return_value = ns(pricing_tier="Standard")
+    assert collector._defender_for_containers() is True
+
+    security_type.return_value.pricings.get.side_effect = PermissionError("forbidden")
+    assert collector._defender_for_containers() is None
+
+
+@patch("scanner.aks_security.Path.is_file", return_value=True)
+@patch("kubernetes.config.load_kube_config")
+@patch("kubernetes.client.RbacAuthorizationV1Api")
+@patch("kubernetes.client.NetworkingV1Api")
+@patch("kubernetes.client.BatchV1Api")
+@patch("kubernetes.client.AppsV1Api")
+@patch("kubernetes.client.CoreV1Api")
+def test_kubernetes_collection_normalizes_workloads_network_policy_and_rbac(
+    core_type,
+    apps_type,
+    batch_type,
+    networking_type,
+    rbac_type,
+    load_config,
+    _is_file,
+):
+    core = core_type.return_value
+    core.list_namespace.return_value = ns(items=[ns(metadata=ns(name="kube-system")), ns(metadata=ns(name="payments"))])
+    pod_spec = ns(
+        service_account_name="api",
+        automount_service_account_token=False,
+        host_network=True,
+        host_pid=False,
+        host_ipc=False,
+        volumes=[
+            ns(host_path=ns(path="/var/run")),
+            ns(secret=ns(secret_name="database-password")),
+            ns(projected=ns(sources=[ns(secret=ns(name="projected-token"))])),
+            ns(
+                csi=ns(
+                    driver="secrets-store.csi.k8s.io",
+                    volume_attributes={"secretProviderClass": "payments-api"},
+                )
+            ),
+        ],
+        containers=[
+            ns(
+                name="api",
+                image="contoso.azurecr.io/api:latest",
+                security_context=ns(privileged=True, capabilities=ns(add=["NET_ADMIN"])),
+                env=[ns(value_from=ns(secret_key_ref=ns(name="api-token")))],
+                env_from=[],
+            )
+        ],
+        init_containers=[],
+    )
+    deployment = ns(metadata=ns(namespace="payments", name="api"), spec=ns(template=ns(spec=pod_spec)))
+    apps_type.return_value.list_namespaced_deployment.return_value = ns(items=[deployment])
+    apps_type.return_value.list_namespaced_stateful_set.return_value = ns(items=[])
+    apps_type.return_value.list_namespaced_daemon_set.return_value = ns(items=[])
+    batch_type.return_value.list_namespaced_job.return_value = ns(items=[])
+    batch_type.return_value.list_namespaced_cron_job.return_value = ns(items=[])
+    core.list_namespaced_pod.return_value = ns(items=[])
+    networking_type.return_value.list_namespaced_network_policy.return_value = ns(items=[object()])
+    rbac_type.return_value.list_cluster_role_binding.return_value = ns(
+        items=[
+            ns(
+                metadata=ns(name="developers-admin"),
+                role_ref=ns(name="cluster-admin"),
+                subjects=[ns(kind="Group", name="developers", namespace=None)],
+            )
+        ]
+    )
+    collector = AksSecurityCollector(MagicMock(), "sub", kubeconfig_path="/tmp/kubeconfig")
+
+    with patch(
+        "kubernetes.config.list_kube_config_contexts",
+        return_value=([{"name": "rg-payments"}], None),
+    ):
+        result = collector._collect_kubernetes(
+            cluster(),
+            "rg-payments",
+            collector._control_plane(cluster(), True),
+            "timestamp",
+        )
+
+    load_config.assert_called_once_with(config_file="/tmp/kubeconfig", context="rg-payments")
+    assert result.status == "COMPLETE"
+    assert result.namespaces == ("payments",)
+    assert result.network_policy_namespaces == ("payments",)
+    assert result.workloads[0]["host_network"] is True
+    assert result.workloads[0]["host_paths"] == ("/var/run",)
+    assert result.workloads[0]["containers"][0]["privileged"] is True
+    assert result.workloads[0]["native_secret_references"] == (
+        "api-token",
+        "database-password",
+        "projected-token",
+    )
+    assert result.workloads[0]["csi_secret_provider_classes"] == ("payments-api",)
+    assert result.cluster_admin_bindings[0]["name"] == "developers"
+
+
+@patch("scanner.aks_security.Path.is_file", return_value=True)
+@patch("kubernetes.config.load_kube_config")
+@patch("kubernetes.client.RbacAuthorizationV1Api")
+@patch("kubernetes.client.NetworkingV1Api")
+@patch("kubernetes.client.BatchV1Api")
+@patch("kubernetes.client.AppsV1Api")
+@patch("kubernetes.client.CoreV1Api")
+def test_namespace_discovery_failure_is_explicitly_unknown(
+    core_type,
+    _apps_type,
+    _batch_type,
+    _networking_type,
+    _rbac_type,
+    _load_config,
+    _is_file,
+):
+    core_type.return_value.list_namespace.side_effect = PermissionError("forbidden")
+    collector = AksSecurityCollector(MagicMock(), "sub", kubeconfig_path="/tmp/kubeconfig")
+
+    with patch(
+        "kubernetes.config.list_kube_config_contexts",
+        return_value=([{"name": "rg-payments"}], None),
+    ):
+        result = collector._collect_kubernetes(
+            cluster(),
+            "rg-payments",
+            collector._control_plane(cluster(), True),
+            "timestamp",
+        )
+
+    assert result.status == "UNKNOWN"
+    assert result.unknown_reason == "NAMESPACE_DISCOVERY_INCOMPLETE"
+
+
+@patch("scanner.aks_security.Path.is_file", return_value=True)
+@patch("kubernetes.config.load_kube_config", side_effect=PermissionError("unauthorized"))
+@patch("kubernetes.config.list_kube_config_contexts", return_value=([{"name": "rg-payments"}], None))
+@patch(
+    "scanner.aks_security.load_kubeconfig_contexts",
+    return_value={
+        "/subscriptions/sub/resourcegroups/rg/providers/microsoft.containerservice/managedclusters/aks-1": "rg-payments"
+    },
+)
+def test_unauthorized_cluster_is_unknown(_load_contexts, _list_contexts, _load_config, _is_file):
+    collector = AksSecurityCollector(
+        MagicMock(),
+        "sub",
+        kubeconfig_path="/tmp/kubeconfig",
+        kubeconfig_contexts_path="/tmp/contexts.json",
+    )
+
+    with patch.object(collector, "_defender_for_containers", return_value=True):
+        result = collector.collect([cluster()])[0]
+
+    assert result.status == "UNKNOWN"
+    assert result.unknown_reason == "KUBERNETES_API_UNAVAILABLE"
+
+
+def test_duplicate_cluster_names_use_distinct_resource_id_contexts(tmp_path):
+    kubeconfig = tmp_path / "kubeconfig"
+    kubeconfig.write_text("contexts: []", encoding="utf-8")
+    contexts = tmp_path / "contexts.json"
+    contexts.write_text(
+        json.dumps({PAYMENTS_CLUSTER_ID: "payments-context", ORDERS_CLUSTER_ID: "orders-context"}),
+        encoding="utf-8",
+    )
+    collector = AksSecurityCollector(
+        MagicMock(),
+        "sub",
+        kubeconfig_path=str(kubeconfig),
+        kubeconfig_contexts_path=str(contexts),
+    )
+    clusters = [
+        cluster(
+            id=PAYMENTS_CLUSTER_ID,
+            name="shared",
+        ),
+        cluster(
+            id=ORDERS_CLUSTER_ID,
+            name="shared",
+        ),
+    ]
+
+    with (
+        patch.object(collector, "_defender_for_containers", return_value=True),
+        patch.object(collector, "_collect_kubernetes", side_effect=["payments", "orders"]) as collect_kubernetes,
+    ):
+        assert collector.collect(clusters) == ["payments", "orders"]
+
+    assert collect_kubernetes.call_args_list[0].args[1] == "payments-context"
+    assert collect_kubernetes.call_args_list[1].args[1] == "orders-context"
+
+
+def test_unmapped_cluster_context_is_explicitly_unknown(tmp_path):
+    kubeconfig = tmp_path / "kubeconfig"
+    kubeconfig.write_text("contexts: []", encoding="utf-8")
+    contexts = tmp_path / "contexts.json"
+    contexts.write_text("{}", encoding="utf-8")
+    collector = AksSecurityCollector(
+        MagicMock(),
+        "sub",
+        kubeconfig_path=str(kubeconfig),
+        kubeconfig_contexts_path=str(contexts),
+    )
+
+    with patch.object(collector, "_defender_for_containers", return_value=True):
+        result = collector.collect([cluster()])[0]
+
+    assert result.status == "UNKNOWN"
+    assert result.unknown_reason == "KUBECONFIG_CONTEXT_UNRESOLVED"
+
+
+@patch("kubernetes.config.list_kube_config_contexts", return_value=([{"name": "another-context"}], None))
+def test_missing_mapped_kubeconfig_context_is_explicitly_unknown(_list_contexts, tmp_path):
+    kubeconfig = tmp_path / "kubeconfig"
+    kubeconfig.write_text("contexts: []", encoding="utf-8")
+    contexts = tmp_path / "contexts.json"
+    contexts.write_text(
+        json.dumps(
+            {
+                "/subscriptions/sub/resourceGroups/rg/providers/"
+                "Microsoft.ContainerService/managedClusters/aks-1": "expected-context"
+            }
+        ),
+        encoding="utf-8",
+    )
+    collector = AksSecurityCollector(
+        MagicMock(),
+        "sub",
+        kubeconfig_path=str(kubeconfig),
+        kubeconfig_contexts_path=str(contexts),
+    )
+
+    with patch.object(collector, "_defender_for_containers", return_value=True):
+        result = collector.collect([cluster()])[0]
+
+    assert result.status == "UNKNOWN"
+    assert result.unknown_reason == "KUBECONFIG_CONTEXT_UNRESOLVED"
+
+
+def test_context_mapping_rejects_one_context_for_multiple_clusters(tmp_path):
+    contexts = tmp_path / "contexts.json"
+    contexts.write_text(
+        json.dumps({PAYMENTS_CLUSTER_ID: "shared-context", ORDERS_CLUSTER_ID: "shared-context"}),
+        encoding="utf-8",
+    )
+
+    with pytest.raises(ValueError, match="only one cluster resource ID"):
+        load_kubeconfig_contexts(contexts)
diff --git a/tests/test_azure_client_aks.py b/tests/test_azure_client_aks.py
index ad476f23..c2c58d6b 100644
--- a/tests/test_azure_client_aks.py
+++ b/tests/test_azure_client_aks.py
@@ -42,3 +42,24 @@ def test_get_managed_clusters_returns_none_on_unexpected_failure(client_type):
     client = AzureClient("sub-1", credential=MagicMock())
 
     assert client.get_managed_clusters() is None
+
+
+@patch("scanner.aks_security.AksSecurityCollector")
+@patch("scanner.azure_client.ContainerServiceClient")
+def test_get_aks_security_posture_collects_and_caches(client_type, collector_type):
+    clusters = ["cluster-a"]
+    client_type.return_value.managed_clusters.list.return_value = clusters
+    collector_type.return_value.collect.return_value = ["evidence-a"]
+    client = AzureClient("sub-1", credential=MagicMock())
+
+    assert client.get_aks_security_posture() == ["evidence-a"]
+    assert client.get_aks_security_posture() == ["evidence-a"]
+    collector_type.return_value.collect.assert_called_once_with(clusters)
+
+
+@patch("scanner.azure_client.ContainerServiceClient")
+def test_get_aks_security_posture_preserves_inventory_failure(client_type):
+    client_type.return_value.managed_clusters.list.side_effect = RuntimeError("unavailable")
+    client = AzureClient("sub-1", credential=MagicMock())
+
+    assert client.get_aks_security_posture() is None
diff --git a/tests/test_rules_aks_enterprise.py b/tests/test_rules_aks_enterprise.py
new file mode 100644
index 00000000..4c478c46
--- /dev/null
+++ b/tests/test_rules_aks_enterprise.py
@@ -0,0 +1,490 @@
+"""Tests for the issue #255 enterprise AKS and workload rule pack."""
+
+from datetime import datetime, timezone
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import pytest
+
+from scanner.aks_security import AksClusterEvidence
+from scanner.rules import (
+    az_aks_007,
+    az_aks_008,
+    az_aks_009,
+    az_aks_010,
+    az_aks_011,
+    az_aks_012,
+    az_aks_013,
+    az_aks_014,
+    az_aks_015,
+    az_aks_016,
+    az_aks_017,
+    az_aks_018,
+    az_aks_019,
+    az_aks_020,
+    az_aks_021,
+)
+from scanner.rules._aks_enterprise_common import load_policy
+
+RULES = [
+    az_aks_007,
+    az_aks_008,
+    az_aks_009,
+    az_aks_010,
+    az_aks_011,
+    az_aks_012,
+    az_aks_013,
+    az_aks_014,
+    az_aks_015,
+    az_aks_016,
+    az_aks_017,
+    az_aks_018,
+    az_aks_019,
+    az_aks_020,
+    az_aks_021,
+]
+REQUIRED_METADATA = {
+    "cluster",
+    "namespace",
+    "workload",
+    "container",
+    "image",
+    "subject",
+    "role",
+    "evidence",
+    "remediation",
+    "permissions_required",
+    "severity",
+    "confidence",
+    "unknown_reason",
+}
+
+
+def test_every_rule_has_its_own_executable_playbook():
+    for rule in RULES:
+        path = Path(rule.PLAYBOOK)
+        assert path.name == f"fix_{rule.RULE_ID.lower().replace('-', '_')}.sh"
+        assert path.is_file()
+        assert path.stat().st_mode & 0o111
+
+
+def cluster(name="aks-1"):
+    return SimpleNamespace(
+        id=f"/subscriptions/sub/resourceGroups/rg/providers/Microsoft.ContainerService/managedClusters/{name}",
+        name=name,
+    )
+
+
+def container(**overrides):
+    values = {
+        "name": "api",
+        "image": "contoso.azurecr.io/app/api@sha256:" + "a" * 64,
+        "privileged": False,
+        "allow_privilege_escalation": False,
+        "run_as_non_root": True,
+        "read_only_root_filesystem": True,
+        "capabilities_add": (),
+        "seccomp_profile": "RuntimeDefault",
+    }
+    values.update(overrides)
+    return values
+
+
+def workload(**overrides):
+    values = {
+        "kind": "Deployment",
+        "namespace": "payments",
+        "name": "api",
+        "service_account": "api",
+        "automount_service_account_token": False,
+        "host_network": False,
+        "host_pid": False,
+        "host_ipc": False,
+        "host_paths": (),
+        "native_secret_references": (),
+        "csi_secret_provider_classes": ("payments-api",),
+        "containers": (container(),),
+        "init_containers": (),
+    }
+    values.update(overrides)
+    return values
+
+
+def evidence(**overrides):
+    values = {
+        "cluster": cluster(),
+        "status": "COMPLETE",
+        "source": "test evidence",
+        "collected_at": datetime.now(timezone.utc).isoformat(),
+        "control_plane": {
+            "private_cluster_enabled": True,
+            "authorized_ip_ranges": (),
+            "network_policy": "cilium",
+            "defender_for_containers_enabled": True,
+            "kms_enabled": True,
+            "csi_enabled": True,
+            "csi_rotation_enabled": True,
+        },
+        "namespaces": ("payments",),
+        "network_policy_namespaces": ("payments",),
+        "workloads": (workload(),),
+        "cluster_admin_bindings": ({"binding": "platform", "kind": "Group", "name": "aks-platform-admins"},),
+    }
+    values.update(overrides)
+    return AksClusterEvidence(**values)
+
+
+@pytest.fixture
+def policy_file(tmp_path, monkeypatch):
+    path = tmp_path / "aks-policy.json"
+    path.write_text(
+        """{
+  "approved_authorized_ip_ranges": ["203.0.113.0/24"],
+  "trusted_registry_prefixes": ["contoso.azurecr.io/"],
+  "allowed_cluster_admin_subjects": ["Group:aks-platform-admins"],
+  "excluded_namespaces": ["kube-system"],
+  "require_image_digests": true
+}""",
+        encoding="utf-8",
+    )
+    monkeypatch.setenv("OPENSHIELD_AKS_SECURITY_POLICY", str(path))
+    return path
+
+
+@pytest.mark.parametrize("rule", RULES)
+def test_secure_evidence_has_no_findings(rule, policy_file):
+    client = MagicMock()
+    client.get_aks_security_posture.return_value = [evidence()]
+    assert rule.scan(client, "sub") == []
+
+
+@pytest.mark.parametrize("rule", RULES)
+def test_empty_inventory_is_not_applicable(rule, policy_file):
+    client = MagicMock()
+    client.get_aks_security_posture.return_value = []
+    assert rule.scan(client, "sub") == []
+
+
+@pytest.mark.parametrize("rule", RULES)
+def test_inventory_failure_is_unknown(rule, policy_file):
+    client = MagicMock()
+    client.get_aks_security_posture.return_value = None
+    assert rule.scan(client, "sub") == []
+
+
+@pytest.mark.parametrize("rule", RULES[6:])
+def test_unreachable_cluster_never_creates_workload_findings(rule, policy_file):
+    client = MagicMock()
+    client.get_aks_security_posture.return_value = [evidence(status="UNKNOWN", unknown_reason="UNREACHABLE")]
+    assert rule.scan(client, "sub") == []
+
+
+@pytest.mark.parametrize(
+    ("rule", "changes", "expected_metadata"),
+    [
+        (az_aks_007, {"private_cluster_enabled": False, "authorized_ip_ranges": ()}, "observed_value"),
+        (az_aks_008, {"network_policy": "none"}, "observed_value"),
+        (az_aks_010, {"defender_for_containers_enabled": False}, "observed_value"),
+        (az_aks_012, {"csi_enabled": True, "csi_rotation_enabled": False}, "observed_value"),
+    ],
+)
+def test_control_plane_violations_emit_complete_evidence(rule, changes, expected_metadata, policy_file):
+    item = evidence()
+    item.control_plane.update(changes)
+    client = MagicMock()
+    client.get_aks_security_posture.return_value = [item]
+    findings = rule.scan(client, "sub")
+    assert len(findings) == 1
+    assert findings[0]["rule_id"] == rule.RULE_ID
+    assert expected_metadata in findings[0]["metadata"]
+    assert findings[0]["metadata"]["unknown_reason"] is None
+    assert findings[0]["metadata"]["confidence"] == "HIGH"
+    assert REQUIRED_METADATA.issubset(findings[0]["metadata"])
+
+
+@pytest.mark.parametrize(
+    ("rule", "bad_workload", "metadata_key"),
+    [
+        (az_aks_013, workload(containers=(container(privileged=True),)), "container"),
+        (az_aks_014, workload(host_network=True), "workload"),
+        (az_aks_015, workload(host_pid=True), "workload"),
+        (az_aks_016, workload(host_ipc=True), "workload"),
+        (az_aks_017, workload(host_paths=("/var/run",)), "workload"),
+        (az_aks_019, workload(containers=(container(image="docker.io/library/nginx@sha256:" + "b" * 64),)), "image"),
+        (az_aks_020, workload(containers=(container(image="contoso.azurecr.io/app/api:latest"),)), "image"),
+        (az_aks_021, workload(containers=(container(image="contoso.azurecr.io/app/api:1.2.3"),)), "image"),
+    ],
+)
+def test_workload_violations_emit_granular_evidence(rule, bad_workload, metadata_key, policy_file):
+    client = MagicMock()
+    client.get_aks_security_posture.return_value = [evidence(workloads=(bad_workload,))]
+    finding = rule.scan(client, "sub")[0]
+    assert finding["resource_type"] == "Kubernetes/workloads"
+    assert finding["metadata"]["namespace"] == "payments"
+    assert finding["metadata"][metadata_key]
+    assert REQUIRED_METADATA.issubset(finding["metadata"])
+
+
+def test_broad_cluster_admin_and_missing_namespace_policy_are_detected(policy_file):
+    client = MagicMock()
+    client.get_aks_security_posture.return_value = [
+        evidence(
+            namespaces=("payments", "orders"),
+            network_policy_namespaces=("payments",),
+            cluster_admin_bindings=({"binding": "everyone", "kind": "Group", "name": "developers"},),
+        )
+    ]
+    admin = az_aks_018.scan(client, "sub")[0]
+    assert admin["metadata"]["subject"] == "Group:developers"
+    assert admin["metadata"]["role"] == "cluster-admin"
+    namespace = az_aks_009.scan(client, "sub")[0]
+    assert namespace["metadata"]["namespace"] == "orders"
+
+
+def test_service_account_allowlist_is_namespace_scoped(policy_file):
+    policy_file.write_text(
+        """{
+  "approved_authorized_ip_ranges": ["203.0.113.0/24"],
+  "trusted_registry_prefixes": ["contoso.azurecr.io/"],
+  "allowed_cluster_admin_subjects": ["ServiceAccount:platform:builder"],
+  "excluded_namespaces": ["kube-system"],
+  "require_image_digests": true
+}""",
+        encoding="utf-8",
+    )
+    client = MagicMock()
+    client.get_aks_security_posture.return_value = [
+        evidence(
+            cluster_admin_bindings=(
+                {
+                    "binding": "platform-builder",
+                    "kind": "ServiceAccount",
+                    "namespace": "platform",
+                    "name": "builder",
+                },
+                {
+                    "binding": "attacker-builder",
+                    "kind": "ServiceAccount",
+                    "namespace": "attacker",
+                    "name": "builder",
+                },
+            )
+        )
+    ]
+
+    findings = az_aks_018.scan(client, "sub")
+
+    assert len(findings) == 1
+    assert findings[0]["metadata"]["subject"] == "ServiceAccount:attacker:builder"
+
+
+def test_positive_evidence_survives_partial_collection(policy_file):
+    client = MagicMock()
+    client.get_aks_security_posture.return_value = [
+        evidence(status="PARTIAL", partial_reasons=("orders",), workloads=(workload(host_ipc=True),))
+    ]
+    assert len(az_aks_016.scan(client, "sub")) == 1
+
+
+def test_failed_namespace_never_becomes_missing_network_policy_finding(policy_file):
+    client = MagicMock()
+    client.get_aks_security_posture.return_value = [
+        evidence(
+            status="PARTIAL",
+            partial_reasons=("orders",),
+            namespaces=("payments", "orders"),
+            network_policy_namespaces=("payments",),
+        )
+    ]
+    assert az_aks_009.scan(client, "sub") == []
+
+
+def test_authorized_subnet_inside_approved_supernet_is_compliant(policy_file):
+    item = evidence()
+    item.control_plane.update({"private_cluster_enabled": False, "authorized_ip_ranges": ("203.0.113.15/32",)})
+    client = MagicMock()
+    client.get_aks_security_posture.return_value = [item]
+    assert az_aks_007.scan(client, "sub") == []
+
+
+def test_privileged_init_container_is_detected(policy_file):
+    client = MagicMock()
+    client.get_aks_security_posture.return_value = [
+        evidence(workloads=(workload(init_containers=(container(name="setup", privileged=True),)),))
+    ]
+    finding = az_aks_013.scan(client, "sub")[0]
+    assert finding["metadata"]["container"] == "setup"
+
+
+def test_implicit_latest_tag_is_detected(policy_file):
+    client = MagicMock()
+    client.get_aks_security_posture.return_value = [
+        evidence(workloads=(workload(containers=(container(image="contoso.azurecr.io/app/api"),)),))
+    ]
+    assert len(az_aks_020.scan(client, "sub")) == 1
+
+
+def test_disabled_csi_provider_makes_rotation_not_applicable(policy_file):
+    item = evidence()
+    item.control_plane.update({"csi_enabled": False, "csi_rotation_enabled": False})
+    client = MagicMock()
+    client.get_aks_security_posture.return_value = [item]
+    assert az_aks_012.scan(client, "sub") == []
+
+
+def test_native_secret_reference_without_kms_is_detected(policy_file):
+    item = evidence(
+        workloads=(
+            workload(
+                native_secret_references=("database-password",),
+                csi_secret_provider_classes=(),
+            ),
+        )
+    )
+    item.control_plane.update({"kms_enabled": False, "csi_enabled": False})
+    client = MagicMock()
+    client.get_aks_security_posture.return_value = [item]
+
+    finding = az_aks_011.scan(client, "sub")[0]
+
+    assert finding["metadata"]["observed_value"]["native_secret_references"] == ["database-password"]
+
+
+@pytest.mark.parametrize(
+    ("rule", "field"),
+    [
+        (az_aks_007, "private_cluster_enabled"),
+        (az_aks_009, None),
+        (az_aks_010, "defender_for_containers_enabled"),
+    ],
+)
+def test_incomplete_evidence_never_creates_false_findings(rule, field, policy_file):
+    item = evidence(status="UNKNOWN", unknown_reason="INCOMPLETE_DISCOVERY")
+    if field:
+        item.control_plane[field] = None
+    client = MagicMock()
+    client.get_aks_security_posture.return_value = [item]
+    assert rule.scan(client, "sub") == []
+
+
+@pytest.mark.parametrize("rule", (az_aks_007, az_aks_018, az_aks_019, az_aks_021))
+def test_missing_policy_is_unknown_for_policy_driven_rules(rule, monkeypatch):
+    monkeypatch.delenv("OPENSHIELD_AKS_SECURITY_POLICY", raising=False)
+    client = MagicMock()
+    client.get_aks_security_posture.return_value = [evidence()]
+    assert rule.scan(client, "sub") == []
+
+
+@pytest.mark.parametrize(
+    ("rule", "item"),
+    [
+        (
+            az_aks_009,
+            evidence(namespaces=("payments", "orders"), network_policy_namespaces=("payments",)),
+        ),
+        (
+            az_aks_011,
+            evidence(
+                control_plane={"kms_enabled": False},
+                workloads=(workload(native_secret_references=("database-password",)),),
+            ),
+        ),
+        (az_aks_013, evidence(workloads=(workload(containers=(container(privileged=True),)),))),
+        (az_aks_014, evidence(workloads=(workload(host_network=True),))),
+        (az_aks_015, evidence(workloads=(workload(host_pid=True),))),
+        (az_aks_016, evidence(workloads=(workload(host_ipc=True),))),
+        (az_aks_017, evidence(workloads=(workload(host_paths=("/var/run",)),))),
+        (
+            az_aks_020,
+            evidence(workloads=(workload(containers=(container(image="contoso.azurecr.io/app/api:latest"),)),)),
+        ),
+    ],
+)
+def test_policy_independent_workload_rules_run_without_policy(rule, item, monkeypatch):
+    monkeypatch.delenv("OPENSHIELD_AKS_SECURITY_POLICY", raising=False)
+    client = MagicMock()
+    client.get_aks_security_posture.return_value = [item]
+
+    assert len(rule.scan(client, "sub")) == 1
+
+
+@pytest.mark.parametrize("image", ("nginx", "redis:7", "library/nginx:1.29"))
+def test_docker_hub_short_form_images_match_trusted_registry(image, policy_file):
+    policy_file.write_text(
+        """{
+  "approved_authorized_ip_ranges": ["203.0.113.0/24"],
+  "trusted_registry_prefixes": ["docker.io/"],
+  "allowed_cluster_admin_subjects": ["Group:aks-platform-admins"],
+  "excluded_namespaces": ["kube-system"],
+  "require_image_digests": true
+}""",
+        encoding="utf-8",
+    )
+    client = MagicMock()
+    client.get_aks_security_posture.return_value = [
+        evidence(workloads=(workload(containers=(container(image=image),)),))
+    ]
+
+    assert az_aks_019.scan(client, "sub") == []
+
+
+def test_trusted_registry_prefix_does_not_cross_repository_boundary(policy_file):
+    policy_file.write_text(
+        """{
+  "approved_authorized_ip_ranges": ["203.0.113.0/24"],
+  "trusted_registry_prefixes": ["contoso.azurecr.io/team"],
+  "allowed_cluster_admin_subjects": ["Group:aks-platform-admins"],
+  "excluded_namespaces": ["kube-system"],
+  "require_image_digests": true
+}""",
+        encoding="utf-8",
+    )
+    client = MagicMock()
+    client.get_aks_security_posture.return_value = [
+        evidence(
+            workloads=(
+                workload(
+                    containers=(
+                        container(name="trusted", image="contoso.azurecr.io/team/api:1.0"),
+                        container(name="lookalike", image="contoso.azurecr.io/team-evil/api:1.0"),
+                    )
+                ),
+            )
+        )
+    ]
+
+    findings = az_aks_019.scan(client, "sub")
+
+    assert len(findings) == 1
+    assert findings[0]["metadata"]["container"] == "lookalike"
+    assert findings[0]["metadata"]["image"] == "contoso.azurecr.io/team-evil/api:1.0"
+
+
+def test_policy_validation_rejects_unknown_fields(policy_file):
+    assert load_policy(policy_file).require_image_digests is True
+    policy_file.write_text('{"unexpected": true}', encoding="utf-8")
+    with pytest.raises(ValueError, match="missing or unsupported"):
+        load_policy(policy_file)
+
+
+def test_policy_validation_rejects_service_account_without_namespace(policy_file):
+    policy_file.write_text(
+        """{
+  "approved_authorized_ip_ranges": ["203.0.113.0/24"],
+  "trusted_registry_prefixes": ["contoso.azurecr.io/"],
+  "allowed_cluster_admin_subjects": ["ServiceAccount:builder"],
+  "excluded_namespaces": ["kube-system"],
+  "require_image_digests": true
+}""",
+        encoding="utf-8",
+    )
+
+    with pytest.raises(ValueError, match="include a namespace"):
+        load_policy(policy_file)
+
+
+def test_malformed_cluster_identity_is_unknown(policy_file):
+    client = MagicMock()
+    client.get_aks_security_posture.return_value = [evidence(cluster=cluster(name=""))]
+    assert az_aks_013.scan(client, "sub") == []