--- # Checkov & Regula IaC Scanner CronJob # Scans Kubernetes manifests daily for security misconfigurations # Results stored in ConfigMap for Grafana dashboard consumption apiVersion: v1 kind: Namespace metadata: name: checkov labels: app: checkov-scanner --- apiVersion: v1 kind: ServiceAccount metadata: name: checkov-scanner namespace: checkov --- apiVersion: rbac.authorization.k8s.io/v1 kind: ClusterRole metadata: name: checkov-scanner rules: # Read all resources for scanning - apiGroups: [""] resources: ["pods", "services", "configmaps", "secrets", "namespaces", "serviceaccounts"] verbs: ["get", "list"] - apiGroups: ["apps"] resources: ["deployments", "statefulsets", "daemonsets", "replicasets"] verbs: ["get", "list"] - apiGroups: ["networking.k8s.io"] resources: ["ingresses", "networkpolicies"] verbs: ["get", "list"] - apiGroups: ["rbac.authorization.k8s.io"] resources: ["roles", "rolebindings", "clusterroles", "clusterrolebindings"] verbs: ["get", "list"] # Write scan results - apiGroups: [""] resources: ["configmaps"] verbs: ["create"] - apiGroups: [""] resources: ["configmaps"] verbs: ["update", "patch"] resourceNames: ["checkov-scan-results"] --- apiVersion: rbac.authorization.k8s.io/v1 kind: ClusterRoleBinding metadata: name: checkov-scanner roleRef: apiGroup: rbac.authorization.k8s.io kind: ClusterRole name: checkov-scanner subjects: - kind: ServiceAccount name: checkov-scanner namespace: checkov --- apiVersion: batch/v1 kind: CronJob metadata: name: checkov-scanner namespace: checkov labels: app: checkov-scanner spec: schedule: "0 3 * * *" # Daily at 3 AM concurrencyPolicy: Forbid successfulJobsHistoryLimit: 3 failedJobsHistoryLimit: 3 jobTemplate: spec: backoffLimit: 1 activeDeadlineSeconds: 1800 # 30 min timeout template: metadata: labels: app: checkov-scanner spec: serviceAccountName: checkov-scanner restartPolicy: Never containers: - name: scanner image: docker.io/bridgecrew/checkov:3.2.334 securityContext: runAsUser: 10000 runAsGroup: 10000 runAsNonRoot: true allowPrivilegeEscalation: false readOnlyRootFilesystem: true capabilities: drop: ["ALL"] seccompProfile: type: RuntimeDefault volumeMounts: - name: tmp mountPath: /tmp command: - python3 - -c - | import subprocess, json, os, ssl, urllib.request, datetime print("=== Checkov Kubernetes Security Scan ===") print(f"Scan started at: {datetime.datetime.utcnow().isoformat()}Z") # Fetch Kubernetes resources as YAML files for scanning token_path = "/var/run/secrets/kubernetes.io/serviceaccount/token" ca_path = "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt" api = "https://kubernetes.default.svc" with open(token_path) as f: token = f.read().strip() ctx = ssl.create_default_context(cafile=ca_path) headers = {"Authorization": f"Bearer {token}"} os.makedirs("/tmp/k8s-manifests", exist_ok=True) # Fetch deployments, statefulsets, daemonsets, pods, services # Focus on workload resources (skip pods/configmaps to reduce memory) resources = [ ("apps/v1", "deployments"), ("apps/v1", "statefulsets"), ("apps/v1", "daemonsets"), ("v1", "services"), ("networking.k8s.io/v1", "ingresses"), ] # Map API group to kind names kind_map = { "deployments": ("Deployment", "apps/v1"), "statefulsets": ("StatefulSet", "apps/v1"), "daemonsets": ("DaemonSet", "apps/v1"), "pods": ("Pod", "v1"), "services": ("Service", "v1"), "configmaps": ("ConfigMap", "v1"), "ingresses": ("Ingress", "networking.k8s.io/v1"), } for api_ver, kind in resources: url = f"{api}/apis/{api_ver}/{kind}" if "/" in api_ver and api_ver != "v1" else f"{api}/api/{api_ver}/{kind}" req = urllib.request.Request(url, headers=headers) try: with urllib.request.urlopen(req, context=ctx) as resp: data = json.loads(resp.read()) kind_name, api_version = kind_map.get(kind, (kind.title(), api_ver)) for item in data.get("items", []): # Ensure apiVersion and kind are set (required by checkov) item["apiVersion"] = api_version item["kind"] = kind_name # Remove status (not part of manifest) item.pop("status", None) ns = item.get("metadata", {}).get("namespace", "default") name = item.get("metadata", {}).get("name", "unknown") fname = f"/tmp/k8s-manifests/{kind}-{ns}-{name}.json" with open(fname, "w") as f: json.dump(item, f) except Exception as e: print(f"Warning: failed to fetch {kind}: {e}") file_count = len(os.listdir("/tmp/k8s-manifests")) print(f"Fetched {file_count} Kubernetes resources") # Run checkov scan on fetched manifests print("Scanning resources with Checkov...") result = subprocess.run( ["checkov", "--framework", "kubernetes", "-d", "/tmp/k8s-manifests", "--output", "json", "--output-file", "/tmp/checkov-out", "--soft-fail", "--quiet"], capture_output=True, text=True ) # Parse results (checkov creates dir with results_json.json inside) passed = failed = skipped = 0 results_file = "/tmp/checkov-out/results_json.json" if not os.path.isfile(results_file): results_file = "/tmp/checkov-out" try: with open(results_file) as f: data = json.load(f) if isinstance(data, list): for entry in data: s = entry.get("summary", {}) passed += s.get("passed", 0) failed += s.get("failed", 0) skipped += s.get("skipped", 0) else: s = data.get("summary", {}) passed = s.get("passed", 0) failed = s.get("failed", 0) skipped = s.get("skipped", 0) except Exception as e: print(f"Warning: failed to parse results: {e}") print(f"\n=== Scan Summary ===") print(f"Passed: {passed}") print(f"Failed: {failed}") print(f"Skipped: {skipped}") scan_time = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ") print(f"Scan completed at: {scan_time}") # Store results in ConfigMap via Kubernetes API cm = json.dumps({ "apiVersion": "v1", "kind": "ConfigMap", "metadata": { "name": "checkov-scan-results", "namespace": "checkov", "labels": {"app": "checkov-scanner"} }, "data": { "last-scan": scan_time, "passed": str(passed), "failed": str(failed), "skipped": str(skipped), } }).encode() # Try patch, then create try: req = urllib.request.Request( f"{api}/api/v1/namespaces/checkov/configmaps/checkov-scan-results", data=cm, headers={**headers, "Content-Type": "application/merge-patch+json"}, method="PATCH") urllib.request.urlopen(req, context=ctx) print("Updated ConfigMap checkov-scan-results") except urllib.error.HTTPError as e: if e.code == 404: req = urllib.request.Request( f"{api}/api/v1/namespaces/checkov/configmaps", data=cm, headers={**headers, "Content-Type": "application/json"}, method="POST") urllib.request.urlopen(req, context=ctx) print("Created ConfigMap checkov-scan-results") else: print(f"Warning: failed to store results: {e}") resources: limits: memory: 1Gi cpu: 500m requests: memory: 512Mi cpu: 100m volumes: - name: tmp emptyDir: {}