mirror of
https://gitlab.kube.huskypup.net/Scooby/Homelabv4.git
synced 2026-08-21 05:26:49 +00:00
246 lines
9.9 KiB
YAML
246 lines
9.9 KiB
YAML
---
|
|
# 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: {}
|