Monitoring Kyverno policy violations with Wazuh

Kubernetes workloads introduce security risks when deployed with excessive privileges, unsafe configurations, or settings that do not meet organizational requirements. Admission controls evaluate resource requests before Kubernetes creates or updates resources. These policy decisions provide useful security telemetry by identifying non-compliant workloads, rejected deployment attempts, and configuration changes.

Kyverno is a Kubernetes-native policy engine that validates resource requests against declarative policies. In Audit mode, Kyverno admits non-compliant resources and records the policy violations in policy reports. When enforcement is enabled, Kyverno rejects non-compliant requests. Kyverno writes Kubernetes events in both Audit and Deny modes, and this integration alerts only on requests that a Deny policy rejects. By default, policy reports and events stay within the Kubernetes cluster, making them invisible to security teams.

Wazuh is an open source security platform that collects and analyzes security telemetry from monitored endpoints. This blog post uses a custom Python collector to retrieve Kyverno policy reports, Kubernetes events, policy settings, and exception declarations. Wazuh analyzes the collector’s output to alert on policy violations, enforcement downgrades, policy removals, and exception declarations. Each policy finding includes policy, resource, namespace, and current enforcement context. Wazuh file integrity monitoring also detects changes to local policy manifests.

Infrastructure

We use the following infrastructure to demonstrate monitoring Kyverno policy violations:

• A pre-built, ready-to-use Wazuh OVA 4.14.8, which includes the Wazuh central components (Wazuh server, Wazuh indexer, and Wazuh dashboard). Follow this guide to download and set up the Wazuh virtual machine.
• An Ubuntu 24.04 endpoint with the following installed:
  • The Wazuh agent 4.14.8 installed and enrolled in the Wazuh server. 
  • A self-managed Kubernetes cluster. We use K3s.

Kyverno policy violation reporting

A Kyverno ValidatingPolicy evaluates Kubernetes admission requests against the conditions defined in its validations field. The matchConstraints field specifies which resources the policy evaluates.

When a validation fails, the validationActions field determines what Kyverno does. An Audit action allows the resource to be created and records the violation in a PolicyReport. Kyverno uses ClusterPolicyReport for cluster-scoped resources.

A Deny action rejects the admission request before Kubernetes creates the resource. Because the resource does not exist, the denied request does not appear as a failed PolicyReport result. Kyverno instead creates a Kubernetes Event with the reason PolicyViolation.

Kyverno also evaluates existing resources when evaluation.background.enabled is set to true. This lets policy reports show resources that currently violate a policy even if they existed before the policy changed. The collector therefore reads both policy reports and PolicyViolation events to capture existing violations and rejected admission attempts.

Architecture

The collector reads Kyverno findings from PolicyReport and ClusterPolicyReport objects and PolicyViolation events, along with the live policy settings and PolicyException objects. It reads each source through the Kubernetes API and checks each policy’s validationActions field to classify a finding as Audit or Deny.

The collector runs every 60 seconds and writes one JSON record per finding to /var/log/kyverno/kyverno-wazuh.log. It writes a record only when a failing result is new, recurrent, or is still failing 24 hours later. It stores processed findings in /var/lib/kyverno-wazuh/seen.json and the previous policy settings in /var/lib/kyverno-wazuh/policies.json, so it doesn’t report the same result on every run.

The Wazuh agent reads the JSON log file and forwards the records to the Wazuh server. The JSON decoder and custom rules generate alerts, while file integrity monitoring separately tracks changes in the /opt/kyverno-policies directory.

Configuration

Ubuntu endpoint

Run the following tests on the Ubuntu endpoint. Each test builds on the resources that the previous tests create. The collector runs every 60 seconds, so allow up to a minute before you look for each alert on the Wazuh dashboard.

Install Kyverno

1. Install Kyverno v1.19.1:

# export KUBECONFIG=/etc/rancher/k3s/k3s.yaml
# kubectl create -f https://github.com/kyverno/kyverno/releases/download/v1.19.1/install.yaml

2. Confirm that the Kyverno controllers are running:

# kubectl -n kyverno get pods

NAME                                             READY   STATUS    RESTARTS   AGE
kyverno-admission-controller-687f76f7f4-cbc5b    1/1     Running   0          2m21s
kyverno-background-controller-66bb47c44c-6f2j8   1/1     Running   0          2m21s
kyverno-cleanup-controller-6d5587df98-gcmtn      1/1     Running   0          2m21s
kyverno-reports-controller-7db7dcdd79-pdl7n      1/1     Running   0          2m21s

Create Kyverno policies

1. Create the following directories in the /opt/ directory:

• kyverno-policies for the Kyverno policy manifests.
• kyverno-wazuh for the integration manifests.

# mkdir -p /opt/kyverno-policies /opt/kyverno-wazuh

2. Set root as the owner and assign 0755 permissions. These permissions allow all users to read and execute, while only root can write to the directories:

# chown root:root /opt/kyverno-policies /opt/kyverno-wazuh
# chmod 0755 /opt/kyverno-policies /opt/kyverno-wazuh

3. Create a file /opt/kyverno-wazuh/kyverno-ephemeralcontainers-rbac.yaml with the following role-based access control (RBAC) configuration to grant Kyverno read access to pods/ephemeralcontainers for policy reporting:

apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: kyverno:ephemeralcontainers-reader
  labels:
    rbac.kyverno.io/aggregate-to-reports-controller: "true"
    rbac.kyverno.io/aggregate-to-background-controller: "true"
    rbac.kyverno.io/aggregate-to-admission-controller: "true"
rules:
  - apiGroups: [""]
    resources: ["pods/ephemeralcontainers"]
    verbs: ["get", "list", "watch"]

4. Apply the RBAC manifest:

# kubectl apply -f /opt/kyverno-wazuh/kyverno-ephemeralcontainers-rbac.yaml

5. Create the file /opt/kyverno-policies/disallow-privileged-containers.yaml with the following policy. This policy checks regular, init, and ephemeral containers for privileged mode. It runs in Audit mode, allowing the request while reporting containers configured with securityContext.privileged: true.

apiVersion: policies.kyverno.io/v1
kind: ValidatingPolicy
metadata:
  name: disallow-privileged-containers
  annotations:
    policies.kyverno.io/title: Disallow Privileged Containers
    policies.kyverno.io/category: Pod Security Standards (Baseline)
    policies.kyverno.io/severity: high
spec:
  validationActions:
    - Audit
  evaluation:
    background:
      enabled: true
  matchConstraints:
    namespaceSelector:
      matchExpressions:
        - key: kubernetes.io/metadata.name
          operator: NotIn
          values:
            - kube-system
            - kube-public
            - kube-node-lease
            - kyverno
    resourceRules:
      - apiGroups: [""]
        apiVersions: ["v1"]
        operations: ["CREATE", "UPDATE"]
        resources: ["pods", "pods/ephemeralcontainers"]
  validations:
    - message: >-
        Privileged mode is disallowed. The fields
        spec.containers[*].securityContext.privileged must be unset or set to false.
      expression: >-
        object.spec.containers.all(c,
          !has(c.securityContext) || !has(c.securityContext.privileged) ||
          c.securityContext.privileged == false) &&
        (!has(object.spec.initContainers) || object.spec.initContainers.all(c,
          !has(c.securityContext) || !has(c.securityContext.privileged) ||
          c.securityContext.privileged == false)) &&
        (!has(object.spec.ephemeralContainers) || object.spec.ephemeralContainers.all(c,
          !has(c.securityContext) || !has(c.securityContext.privileged) ||
          c.securityContext.privileged == false))

6. Create the file /opt/kyverno-policies/restrict-image-registries.yaml with the following policy. This policy requires regular, init, and ephemeral containers to use images from registry.k8s.io. It runs in Deny mode and rejects matching admission requests containing images from other registries.

apiVersion: policies.kyverno.io/v1
kind: ValidatingPolicy
metadata:
  name: restrict-image-registries
  annotations:
    policies.kyverno.io/title: Restrict Image Registries
    policies.kyverno.io/category: Supply Chain Security
    policies.kyverno.io/severity: high
spec:
  validationActions:
    - Deny
  evaluation:
    background:
      enabled: true
  matchConstraints:
    namespaceSelector:
      matchExpressions:
        - key: kubernetes.io/metadata.name
          operator: NotIn
          values:
            - kube-system
            - kube-public
            - kube-node-lease
            - kyverno
    resourceRules:
      - apiGroups: [""]
        apiVersions: ["v1"]
        operations: ["CREATE", "UPDATE"]
        resources: ["pods", "pods/ephemeralcontainers"]
  validations:
    - message: >-
        Container images must be pulled from the approved registry registry.k8s.io.
      expression: >-
        object.spec.containers.all(c, c.image.startsWith('registry.k8s.io/')) &&
        (!has(object.spec.initContainers) ||
          object.spec.initContainers.all(c, c.image.startsWith('registry.k8s.io/'))) &&
        (!has(object.spec.ephemeralContainers) ||
          object.spec.ephemeralContainers.all(c, c.image.startsWith('registry.k8s.io/')))

7. Create the file /opt/kyverno-policies/require-namespace-owner-label.yaml with the following policy. This policy checks that namespaces have an owner label identifying the responsible team. It runs in Audit mode, allowing namespace creation while reporting missing labels. The policy excludes kube-system, kube-public, kube-node-lease, kyverno, and the default namespace.

apiVersion: policies.kyverno.io/v1
kind: ValidatingPolicy
metadata:
  name: require-namespace-owner-label
  annotations:
    policies.kyverno.io/title: Require Namespace Owner Label
    policies.kyverno.io/category: Namespace Governance
    policies.kyverno.io/severity: medium
spec:
  validationActions:
    - Audit
  evaluation:
    background:
      enabled: true
  matchConditions:
    - name: exclude-system-namespaces
      expression: >-
        !(object.metadata.name in
          ['kube-system', 'kube-public', 'kube-node-lease', 'kyverno', 'default'])
  matchConstraints:
    resourceRules:
      - apiGroups: [""]
        apiVersions: ["v1"]
        operations: ["CREATE", "UPDATE"]
        resources: ["namespaces"]
  validations:
    - message: >-
        Namespaces must carry an owner label identifying the responsible team.
      expression: "'owner' in object.metadata.?labels.orValue({})"

8. Apply the policies:

# kubectl apply -f /opt/kyverno-policies/

validatingpolicy.policies.kyverno.io/disallow-privileged-containers created
validatingpolicy.policies.kyverno.io/require-namespace-owner-label created
validatingpolicy.policies.kyverno.io/restrict-image-registries created

9. Confirm that the Kyverno policies are ready:

# kubectl get validatingpolicy

NAME                             AGE   READY
disallow-privileged-containers   2s    true
require-namespace-owner-label    2s    true
restrict-image-registries        2s    true

Configure the collector

1. Create a manifest /opt/kyverno-wazuh/rbac.yaml to provision a dedicated service account with read-only access to the Kubernetes resources the collector needs:

apiVersion: v1
kind: ServiceAccount
metadata:
  name: kyverno-wazuh-collector
  namespace: kyverno
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: kyverno-wazuh-collector
rules:
  - apiGroups: ["wgpolicyk8s.io"]
    resources: ["policyreports", "clusterpolicyreports"]
    verbs: ["get", "list"]
  - apiGroups: ["policies.kyverno.io"]
    resources:
      - validatingpolicies
      - namespacedvalidatingpolicies
      - policyexceptions
    verbs: ["get", "list"]
  - apiGroups: ["kyverno.io"]
    resources:
      - clusterpolicies
      - policies
      - policyexceptions
    verbs: ["get", "list"]
  - apiGroups: [""]
    resources: ["events"]
    verbs: ["get", "list"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: kyverno-wazuh-collector
roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: ClusterRole
  name: kyverno-wazuh-collector
subjects:
  - kind: ServiceAccount
    name: kyverno-wazuh-collector
    namespace: kyverno
---
apiVersion: v1
kind: Secret
metadata:
  name: kyverno-wazuh-collector-token
  namespace: kyverno
  annotations:
    kubernetes.io/service-account.name: kyverno-wazuh-collector
type: kubernetes.io/service-account-token

2. Apply the /opt/kyverno-wazuh/rbac.yaml manifest:

# kubectl apply -f /opt/kyverno-wazuh/rbac.yaml

3. Create a dedicated system account to run the collector without root privileges, then create its working directories:

# useradd --system --no-create-home --shell /usr/sbin/nologin kyverno-wazuh 

4. Create the following directories:

• /etc/kyverno-wazuh for the kubeconfig file containing the Kubernetes service account token. 
• /var/lib/kyverno-wazuh for seen.json and policies.json, which track collected findings and previous policy settings. 
• /var/log/kyverno for the collector’s output file, kyverno-wazuh.log. 

# mkdir -p /etc/kyverno-wazuh /var/lib/kyverno-wazuh /var/log/kyverno

5. Set kyverno-wazuh as the owner and assign the directory permissions:

• 0700 gives the owner read, write, and execute permissions to the credentials directory.
• 0750 gives the owner read, write, and execute permissions, and the group read and execute permissions for the state directory.
• 0755 gives the owner read, write, and execute permissions, and all other users read and execute permissions for the log directory.

# chown kyverno-wazuh:kyverno-wazuh /etc/kyverno-wazuh /var/lib/kyverno-wazuh /var/log/kyverno
# chmod 0700 /etc/kyverno-wazuh
# chmod 0750 /var/lib/kyverno-wazuh
# chmod 0755 /var/log/kyverno 

6. Run the following commands to build a kubeconfig from the service account token:

# TOKEN=$(kubectl -n kyverno get secret kyverno-wazuh-collector-token -o jsonpath='{.data.token}' | base64 -d)
# CA=$(kubectl -n kyverno get secret kyverno-wazuh-collector-token -o jsonpath='{.data.ca\.crt}')
# cat > /etc/kyverno-wazuh/kubeconfig <<EOF
apiVersion: v1
kind: Config
clusters:
  - name: local
    cluster:
      server: https://127.0.0.1:6443
      certificate-authority-data: ${CA}
users:
  - name: kyverno-wazuh-collector
    user:
      token: ${TOKEN}
contexts:
  - name: default
    context:
      cluster: local
      user: kyverno-wazuh-collector
current-context: default
EOF

7. Set kyverno-wazuh as the owner and group of the kubeconfig file:

# chown kyverno-wazuh:kyverno-wazuh /etc/kyverno-wazuh/kubeconfig

8. Set 0600 permissions so only the owner can read and modify the file containing the Kubernetes credentials:

# chmod 0600 /etc/kyverno-wazuh/kubeconfig

9. Create the collector file /usr/local/bin/kyverno-wazuh-collector.py with the following content. The script collects Kyverno policy reports, Kubernetes events, policy changes, and exception declarations through the Kubernetes API. It writes each finding as a JSON record in /var/log/kyverno/kyverno-wazuh.log for the Wazuh agent to collect.

#!/usr/bin/env python3

import hashlib
import json
from datetime import datetime
import os
import re
import subprocess
import sys
import time

KUBECONFIG = os.environ.get("KUBECONFIG", "/etc/kyverno-wazuh/kubeconfig")
KUBECTL = os.environ.get("KUBECTL", "/usr/local/bin/kubectl")
OUT_LOG = os.environ.get("KYVERNO_OUT_LOG", "/var/log/kyverno/kyverno-wazuh.log")
STATE_FILE = os.environ.get("KYVERNO_STATE_FILE", "/var/lib/kyverno-wazuh/seen.json")
POLICY_STATE_FILE = os.environ.get("KYVERNO_POLICY_STATE", "/var/lib/kyverno-wazuh/policies.json")
REMINDER_AFTER = 86400

CEL_POLICY_KINDS = ("validatingpolicies", "namespacedvalidatingpolicies")
CEL_KINDS = ("ValidatingPolicy", "NamespacedValidatingPolicy")
LEGACY_KINDS = ("ClusterPolicy", "Policy")
REPORT_SOURCE_KINDS = {
    "KyvernoValidatingPolicy": CEL_KINDS,
    "KyvernoPolicy": LEGACY_KINDS,
    "kyverno": LEGACY_KINDS,
}
FAILING_RESULTS = ("fail", "error")
LEGACY_POLICY_KINDS = ("clusterpolicies", "policies")
EXCEPTION_KINDS = ("policyexceptions.policies.kyverno.io", "policyexceptions.kyverno.io")
POLICY_EVENT_KINDS = ("ClusterPolicy", "Policy", "ValidatingPolicy", "NamespacedValidatingPolicy")
ADMISSION_COMPONENT = "kyverno-admission"
ACTION_LABEL = {"enforce": "Deny", "audit": "Audit"}

def action_label(mode):
    return ACTION_LABEL.get(mode, mode or "unknown")

ABSENT_TYPE_MARKERS = ("doesn't have a resource type", "could not find the requested resource")

EVENT_RE = re.compile(
    r"^(?P<kind>\S+)\s+(?:(?P<namespace>[^/\s]+)/)?(?P<name>\S+?):\s+"
    r"(?:\[(?P<rule>[^\]]+)\]\s+)?(?P<result>\w+)(?:\s+\(blocked\))?;\s*(?P<message>.*)$"
)

def kubectl_json(*args, optional=False):
    env = dict(os.environ, KUBECONFIG=KUBECONFIG)
    proc = subprocess.run(
        [KUBECTL, *args, "-o", "json"],
        capture_output=True, text=True, env=env, timeout=60,
    )
    if proc.returncode != 0:
        stderr = proc.stderr.strip()
        if optional and any(marker in stderr for marker in ABSENT_TYPE_MARKERS):
            return {"items": []}
        print("kubectl %s failed: %s" % (" ".join(args), stderr), file=sys.stderr)
        sys.exit(1)
    return json.loads(proc.stdout)

def policy_id(kind, namespace, name):
    return "%s|%s|%s" % (kind or "", namespace or "", name or "")

def split_policy_ref(ref):
    if ref and "/" in ref:
        ns, _, name = ref.partition("/")
        return ns, name
    return "", ref or ""

def policy_inventory():
    inventory = {}
    for kind in CEL_POLICY_KINDS:
        for item in kubectl_json("get", kind, "-A", optional=True).get("items", []):
            meta, spec = item.get("metadata") or {}, item.get("spec") or {}
            actions = spec.get("validationActions") or []
            admission = ((spec.get("evaluation") or {}).get("admission") or {})
            inventory[policy_id(item.get("kind"), meta.get("namespace"), meta.get("name"))] = {
                "kind": item.get("kind"),
                "namespace": meta.get("namespace") or "",
                "name": meta.get("name"),
                "uid": meta.get("uid"),
                "mode": "enforce" if "Deny" in actions else "audit",
                "admission": admission.get("enabled", True),
            }
    for kind in LEGACY_POLICY_KINDS:
        for item in kubectl_json("get", kind, "-A", optional=True).get("items", []):
            meta, spec = item.get("metadata") or {}, item.get("spec") or {}
            actions = [(r.get("validate") or {}).get("failureAction")
                       for r in spec.get("rules") or []]
            actions = [a.lower() for a in actions if a]
            fallback = spec.get("validationFailureAction")
            if not actions and fallback:
                actions = [fallback.lower()]
            if not actions:
                continue
            inventory[policy_id(item.get("kind"), meta.get("namespace"), meta.get("name"))] = {
                "kind": item.get("kind"),
                "namespace": meta.get("namespace") or "",
                "name": meta.get("name"),
                "uid": meta.get("uid"),
                "mode": "enforce" if "enforce" in actions else "audit",
                "admission": True,
            }
    return inventory

def lookup_policy(inventory, name, namespace="", kinds=()):
    hits = [inventory[pid] for pid in
            (policy_id(k, namespace, name) for k in kinds) if pid in inventory]
    return hits[0] if len(hits) == 1 else None

def load_json(path, default):
    try:
        with open(path) as fh:
            return json.load(fh)
    except (OSError, ValueError):
        return default

def save_json(path, data):
    os.makedirs(os.path.dirname(path), exist_ok=True)
    tmp = path + ".tmp"
    with open(tmp, "w") as fh:
        json.dump(data, fh)
    os.replace(tmp, path)

def emit(records):
    if not records:
        return
    os.makedirs(os.path.dirname(OUT_LOG), exist_ok=True)
    with open(OUT_LOG, "a") as fh:
        for rec in records:
            fh.write(json.dumps({"kyverno": rec}, sort_keys=True) + "\n")

def is_expired(expires, now):
    if not expires:
        return False
    try:
        return datetime.fromisoformat(expires).timestamp() < now
    except ValueError:
        print("unparsable expiresAt: %s" % expires, file=sys.stderr)
        return False

def digest(*parts):
    return hashlib.sha256("|".join(parts).encode()).hexdigest()[:32]

def lifecycle_for(state, key, result, now):
    previous = state.get(key)
    if previous is None:
        state[key] = {"result": result, "t": now}
        return "new" if result in FAILING_RESULTS else None
    was = previous.get("result")
    previous["result"] = result
    if result not in FAILING_RESULTS:
        return None
    if was != result:
        previous["t"] = now
        return "recurrence"
    if now - previous.get("t", 0) >= REMINDER_AFTER:
        previous["t"] = now
        return "reminder"
    return None

def prune_state(state, touched, now):
    return {k: v for k, v in state.items()
            if k in touched or now - v.get("t", 0) < REMINDER_AFTER * 2}

def result_timestamp(res):
    ts = res.get("timestamp") or {}
    seconds = ts.get("seconds")
    if not seconds:
        return ""
    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(int(seconds)))

def collect_reports(state, touched, now, inventory):
    records = []
    for kind in ("policyreport", "clusterpolicyreport"):
        for item in kubectl_json("get", kind, "-A", optional=True).get("items", []):
            scope = item.get("scope") or {}
            meta = item.get("metadata") or {}
            for res in item.get("results") or []:
                policy_ns, policy_name = split_policy_ref(res.get("policy"))
                kinds = REPORT_SOURCE_KINDS.get(res.get("source"))
                if kinds is None:
                    print("unsupported report source: %s" % res.get("source"), file=sys.stderr)
                    continue
                entry = lookup_policy(inventory, policy_name, policy_ns, kinds)
                if entry is None:
                    continue
                mode = entry["mode"]
                key = digest(kind, scope.get("uid", meta.get("name", "")),
                             entry["kind"], entry["namespace"], entry["name"],
                             entry.get("uid") or "", res.get("rule") or "", mode)
                touched.add(key)
                stage = lifecycle_for(state, key, res.get("result"), now)
                if stage is None:
                    continue
                records.append({
                    "source": kind,
                    "event_type": "existing_resource_violation" if mode == "enforce" else "violation",
                    "lifecycle": stage,
                    "enforcement": action_label(mode),
                    "policy": policy_name,
                    "policy_kind": entry["kind"],
                    "policy_namespace": policy_ns or entry["namespace"],
                    "rule": res.get("rule"),
                    "result": res.get("result"),
                    "severity": res.get("severity"),
                    "category": res.get("category"),
                    "policy_source": res.get("source"),
                    "resource_kind": scope.get("kind"),
                    "resource_name": scope.get("name"),
                    "resource_namespace": scope.get("namespace") or "cluster-scoped",
                    "resource_uid": scope.get("uid"),
                    "report_name": meta.get("name"),
                    "source_timestamp": result_timestamp(res),
                    "collected_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(now)),
                    "message": res.get("message"),
                })
    return records

def collect_events(state, touched, now, inventory):
    records = []
    for item in kubectl_json("get", "events", "-A",
                             "--field-selector", "reason=PolicyViolation").get("items", []):
        involved = item.get("involvedObject") or {}
        if involved.get("kind") not in POLICY_EVENT_KINDS:
            continue
        meta = item.get("metadata") or {}
        parsed = EVENT_RE.match(item.get("message", ""))
        if not parsed:
            print("unparsed PolicyViolation event: %s" % meta.get("uid"), file=sys.stderr)
            continue
        entry = lookup_policy(inventory, involved.get("name"),
                              involved.get("namespace") or "", (involved.get("kind"),))
        if entry and involved.get("uid") and entry.get("uid") != involved["uid"]:
            entry = None
        mode = entry["mode"] if entry else "unknown"
        component = item.get("reportingComponent") or (item.get("source") or {}).get("component")
        if component != ADMISSION_COMPONENT:
            continue
        key = digest("event", meta.get("uid", ""), str(item.get("count", 1)))
        touched.add(key)
        if key in state:
            continue
        state[key] = {"result": "fail", "t": now}
        if entry is None:
            print("admission event for unresolved policy %s/%s, not classified: %s"
                  % (involved.get("namespace") or "-", involved.get("name"), meta.get("uid")),
                  file=sys.stderr)
            continue
        if mode != "enforce":
            continue
        records.append({
            "source": "event",
            "event_type": "rejected",
            "lifecycle": "new",
            "enforcement": action_label(mode),
            "policy": involved.get("name"),
            "policy_kind": involved.get("kind"),
            "policy_namespace": involved.get("namespace") or "",
            "rule": parsed.group("rule"),
            "result": parsed.group("result"),
            "resource_kind": parsed.group("kind"),
            "resource_name": parsed.group("name"),
            "resource_namespace": parsed.group("namespace") or "cluster-scoped",
            "event_uid": meta.get("uid"),
            "event_count": item.get("count", 1),
            "reporting_component": component,
            "source_timestamp": item.get("lastTimestamp") or item.get("eventTime") or "",
            "collected_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(now)),
            "message": parsed.group("message"),
        })
    return records

def collect_policy_changes(previous, current, now):
    if previous is None:
        return []
    stamp = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(now))
    records = []
    for pid, entry in sorted(current.items()):
        was = previous.get(pid)
        if not was:
            continue
        base = {
            "source": "policy_state",
            "lifecycle": "new",
            "policy": entry["name"],
            "policy_kind": entry["kind"],
            "policy_namespace": entry["namespace"],
            "collected_at": stamp,
        }
        if was.get("mode") == "enforce" and entry["mode"] != "enforce":
            records.append(dict(base,
                                event_type="policy_downgraded",
                                enforcement=action_label(entry["mode"]),
                                previous_enforcement=action_label(was.get("mode")),
                                message="Policy %s no longer denies admission requests."
                                        % entry["name"]))
        if was.get("uid") and entry.get("uid") and was["uid"] != entry["uid"]:
            records.append(dict(base,
                                event_type="policy_replaced",
                                enforcement=action_label(entry["mode"]),
                                previous_enforcement=action_label(was.get("mode")),
                                message="Policy %s was replaced by a new object with the same "
                                        "name." % entry["name"]))
        if was.get("admission", True) and not entry["admission"]:
            records.append(dict(base,
                                event_type="policy_admission_disabled",
                                enforcement=action_label(entry["mode"]),
                                previous_enforcement=action_label(was.get("mode")),
                                message="Policy %s no longer evaluates admission requests."
                                        % entry["name"]))
    for pid, was in sorted(previous.items()):
        if pid not in current:
            records.append({
                "source": "policy_state",
                "event_type": "policy_removed",
                "lifecycle": "new",
                "enforcement": "none",
                "previous_enforcement": action_label(was.get("mode")),
                "policy": was.get("name"),
                "policy_kind": was.get("kind"),
                "policy_namespace": was.get("namespace", ""),
                "collected_at": stamp,
                "message": "Policy %s is no longer present in the cluster." % was.get("name"),
            })
    return records

def collect_exceptions(state, touched, now):
    records = []
    stamp = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(now))
    for kind in EXCEPTION_KINDS:
        for item in kubectl_json("get", kind, "-A", optional=True).get("items", []):
            meta, spec = item.get("metadata") or {}, item.get("spec") or {}
            refs = [r.get("name") for r in spec.get("policyRefs") or [] if r.get("name")]
            for legacy in spec.get("exceptions") or []:
                if legacy.get("policyName"):
                    refs.append(legacy["policyName"])
            expires = spec.get("expiresAt") or ""
            expired = is_expired(expires, now)
            key = digest("exception", meta.get("uid", ""),
                         str(meta.get("generation", "")), "expired" if expired else "active")
            touched.add(key)
            if key in state:
                continue
            state[key] = {"result": "fail", "t": now}
            policies = ",".join(sorted(set(refs))) or "unspecified"
            records.append({
                "source": "policyexception",
                "event_type": "exception_expired" if expired else "exception_declared",
                "lifecycle": "new",
                "exception_name": meta.get("name"),
                "exception_kind": item.get("kind"),
                "exception_namespace": meta.get("namespace") or "cluster-scoped",
                "policy": policies,
                "expires_at": expires or "never",
                "collected_at": stamp,
                "message": "PolicyException %s declares an exemption from %s."
                           % (meta.get("name"), policies),
            })
    return records

def main():
    now = time.time()
    state = load_json(STATE_FILE, {})
    touched = set()
    inventory = policy_inventory()
    records = (collect_policy_changes(load_json(POLICY_STATE_FILE, None), inventory, now)
               + collect_reports(state, touched, now, inventory)
               + collect_events(state, touched, now, inventory)
               + collect_exceptions(state, touched, now))
    emit(records)
    save_json(STATE_FILE, prune_state(state, touched, now))
    save_json(POLICY_STATE_FILE, inventory)
    print("emitted %d record(s)" % len(records))

if __name__ == "__main__":
    main()

10. Set the ownership and permissions on the collector script:

# chown root:root /usr/local/bin/kyverno-wazuh-collector.py
# chmod 0755 /usr/local/bin/kyverno-wazuh-collector.py

11. Create the file /etc/systemd/system/kyverno-wazuh-collector.service with the following content. This service configures systemd to run the collector using the dedicated system account and Kubernetes credentials: 

[Unit]
Description=Kyverno to Wazuh policy telemetry collector
After=k3s.service

[Service]
Type=oneshot
User=kyverno-wazuh
Group=kyverno-wazuh
Environment=KUBECONFIG=/etc/kyverno-wazuh/kubeconfig
Environment=HOME=/var/lib/kyverno-wazuh
ExecStart=/usr/local/bin/kyverno-wazuh-collector.py

12. Create the file /etc/systemd/system/kyverno-wazuh-collector.timer with the following content. This schedules the collector to run every 60 seconds:

Unit]
Description=Run the Kyverno to Wazuh collector every 60 seconds

[Timer]
OnBootSec=60
OnUnitActiveSec=60
AccuracySec=5s
Unit=kyverno-wazuh-collector.service

[Install]
WantedBy=timers.target

13. Create the collector log file. This file stores Kyverno findings for the Wazuh agent to collect.

# install -o kyverno-wazuh -g kyverno-wazuh -m 0644 /dev/null /var/log/kyverno/kyverno-wazuh.log

14. Reload systemd and start the kyverno-wazuh-collector service:

# systemctl daemon-reload
# systemctl start kyverno-wazuh-collector.service
# systemctl enable --now kyverno-wazuh-collector.timer

15. Confirm that the collector runs and writes records:

# systemctl status kyverno-wazuh-collector.service --no-pager
# journalctl -u kyverno-wazuh-collector.service -n 20 --no-pager

Configure the Wazuh agent

1. Append the following block to the /var/ossec/etc/ossec.conf file to monitor the collector logs:

<ossec_config>
  <localfile>
    <log_format>json</log_format>
    <location>/var/log/kyverno/kyverno-wazuh.log</location>
  </localfile>
</ossec_config>

2. Add the following setting inside the existing <syscheck> block in /var/ossec/etc/ossec.conf to monitor the /opt/kyverno-policies policy directory in realtime:

<directories check_all="yes" realtime="yes" report_changes="yes">/opt/kyverno-policies</directories>

3. Restart the Wazuh agent to apply the configuration changes:

# systemctl restart wazuh-agent

# systemctl restart wazuh-agent

Wazuh server

Perform the following steps on the Wazuh dashboard:

1. Navigate to Server management > Rules, then click + Add new rules file.
2. Copy and paste the rules below and name the file kyverno_rules.xml. Click Save, then Reload to apply the changes: 

<group name="kyverno,kubernetes,">
  <rule id="101900" level="0">
    <decoded_as>json</decoded_as>
    <field name="kyverno.source">\.+</field>
    <description>Kyverno policy telemetry event.</description>
  </rule>
  <rule id="101901" level="6">
    <if_sid>101900</if_sid>
    <field name="kyverno.source">^policyreport$</field>
    <field name="kyverno.event_type">^violation$</field>
    <field name="kyverno.result">^fail$</field>
    <field name="kyverno.lifecycle" type="pcre2">^(new|recurrence)$</field>
    <description>Kyverno Audit policy violation: policy $(kyverno.policy) failed on $(kyverno.resource_kind) $(kyverno.resource_name) in namespace $(kyverno.resource_namespace). Finding is $(kyverno.lifecycle).</description>
    <group>kyverno_audit_violation,</group>
  </rule>
  <rule id="101902" level="10">
    <if_sid>101900</if_sid>
    <field name="kyverno.event_type">^rejected$</field>
    <field name="kyverno.result">^fail$</field>
    <description>Kyverno admission request refused by Deny policy $(kyverno.policy): $(kyverno.resource_kind) $(kyverno.resource_name) in $(kyverno.resource_namespace).</description>
    <group>kyverno_admission_refused,</group>
  </rule>
  <rule id="101903" level="10" frequency="5" timeframe="300">
    <if_matched_sid>101901</if_matched_sid>
    <same_field>kyverno.resource_namespace</same_field>
    <description>Kyverno repeat offender: 5 or more new or recurring policy violations received for namespace $(kyverno.resource_namespace) within 5 minutes.</description>
    <group>kyverno_repeat_offender,</group>
  </rule>
  <rule id="101904" level="10">
    <if_sid>550,553,554</if_sid>
    <field name="file">^/opt/kyverno-policies</field>
    <description>Kyverno policy file $(file) was added, modified, or deleted.</description>
    <group>syscheck,kyverno_policy_change,</group>
  </rule>
  <rule id="101905" level="12">
    <if_sid>101900</if_sid>
    <field name="kyverno.event_type">^existing_resource_violation$</field>
    <field name="kyverno.result">^fail$</field>
    <field name="kyverno.lifecycle" type="pcre2">^(new|recurrence)$</field>
    <description>Kyverno Deny policy $(kyverno.policy) has a failed report for existing $(kyverno.resource_kind) $(kyverno.resource_name) in $(kyverno.resource_namespace).</description>
    <group>kyverno_existing_violation,</group>
  </rule>
  <rule id="101906" level="10">
    <if_sid>101902</if_sid>
    <field name="kyverno.policy">^restrict-image-registries$</field>
    <field name="kyverno.result">^fail$</field>
    <description>Registry policy validation failed for $(kyverno.resource_kind) $(kyverno.resource_name) in $(kyverno.resource_namespace).</description>
    <mitre>
      <id>T1610</id>
    </mitre>
    <group>kyverno_untrusted_image,</group>
  </rule>
  <rule id="101907" level="12">
    <if_sid>101900</if_sid>
    <field name="kyverno.event_type">^policy_downgraded$</field>
    <description>Kyverno policy $(kyverno.policy) was downgraded from $(kyverno.previous_enforcement) to $(kyverno.enforcement). It no longer refuses admission requests.</description>
    <mitre>
      <id>T1562.001</id>
    </mitre>
    <group>kyverno_enforcement_weakened,</group>
  </rule>
  <rule id="101908" level="12">
    <if_sid>101900</if_sid>
    <field name="kyverno.event_type">^policy_removed$</field>
    <description>Kyverno policy $(kyverno.policy) was removed from the cluster.</description>
    <mitre>
      <id>T1562.001</id>
    </mitre>
    <group>kyverno_enforcement_weakened,</group>
  </rule>
  <rule id="101909" level="10">
    <if_sid>101900</if_sid>
    <field name="kyverno.event_type">^exception_declared$</field>
    <description>Kyverno PolicyException $(kyverno.exception_name) in namespace $(kyverno.exception_namespace) declares an exemption from policy $(kyverno.policy), expiring $(kyverno.expires_at).</description>
    <group>kyverno_policy_exception,</group>
  </rule>
  <rule id="101910" level="6">
    <if_sid>101900</if_sid>
    <field name="kyverno.source">^clusterpolicyreport$</field>
    <field name="kyverno.event_type">^violation$</field>
    <field name="kyverno.result">^fail$</field>
    <field name="kyverno.lifecycle" type="pcre2">^(new|recurrence)$</field>
    <description>Kyverno Audit policy violation on a cluster-scoped resource: policy $(kyverno.policy) failed on $(kyverno.resource_kind) $(kyverno.resource_name).</description>
    <group>kyverno_cluster_scoped,</group>
  </rule>
  <rule id="101911" level="12">
    <if_sid>101900</if_sid>
    <field name="kyverno.event_type">^policy_admission_disabled$</field>
    <description>Kyverno policy $(kyverno.policy) stopped evaluating admission requests while its action stayed $(kyverno.enforcement).</description>
    <mitre>
      <id>T1562.001</id>
    </mitre>
    <group>kyverno_enforcement_weakened,</group>
  </rule>
  <rule id="101912" level="3">
    <if_sid>101900</if_sid>
    <field name="kyverno.event_type">^exception_expired$</field>
    <description>Kyverno PolicyException $(kyverno.exception_name) in namespace $(kyverno.exception_namespace) expired at $(kyverno.expires_at).</description>
    <group>kyverno_policy_exception,</group>
  </rule>
  <rule id="101913" level="3">
    <if_sid>101900</if_sid>
    <field name="kyverno.lifecycle">^reminder$</field>
    <description>Kyverno finding still unresolved after 24 hours: policy $(kyverno.policy) on $(kyverno.resource_kind) $(kyverno.resource_name).</description>
    <group>kyverno_reminder,</group>
  </rule>
  <rule id="101914" level="10">
    <if_sid>101900</if_sid>
    <field name="kyverno.result">^error$</field>
    <field name="kyverno.lifecycle" type="pcre2">^(new|recurrence)$</field>
    <description>Kyverno could not complete policy evaluation for $(kyverno.resource_kind) $(kyverno.resource_name) in $(kyverno.resource_namespace) under policy $(kyverno.policy).</description>
    <group>kyverno_evaluation_error,</group>
  </rule>
  <rule id="101915" level="10">
    <if_sid>101900</if_sid>
    <field name="kyverno.event_type">^policy_replaced$</field>
    <description>Kyverno policy $(kyverno.policy) was deleted and recreated with the same name.</description>
    <group>kyverno_enforcement_weakened,</group>
  </rule>
</group>

Where:

• Rule 101900 groups all Kyverno records and serves as the base rule.
• Rule 101901 is triggered when a new or recurring Audit policy violation is reported for a namespaced resource.
• Rule 101902  is triggered when Kyverno rejects an admission request under a Deny policy.
• Rule 101903 is triggered when rule 101901 fires five times for the same namespace within five minutes.
• Rule 101904  is triggered when a file under /opt/kyverno-policies is added, modified, or deleted.
• Rule 101905  is triggered when a violation is reported for an existing resource under a policy configured with Deny.
• Rule 101906 is triggered when the restrict-image-registries policy rejects an admission request.
• Rule 101907  is triggered when a policy is downgraded from Deny to Audit.
• Rule 101908 is triggered when a policy is removed from the cluster.
• Rule 101909 is triggered when a PolicyException exempts resources from a policy.
• Rule 101910 is triggered when an Audit policy violation is reported for a cluster-scoped resource.
• Rule 101911 is triggered when admission evaluation is disabled for a policy.
• Rule 101912 is triggered when a PolicyException has expired.
• Rule 101913 is triggered when a failing finding is still unresolved 24 hours after the collector first reported it.
• Rule 101914 is triggered when Kyverno reports a policy evaluation error.
• Rule 101915 is triggered when a policy is deleted and recreated with the same name.

Testing 

Perform the following tests on the Ubuntu endpoint to validate the integration.

Cluster-scoped policy violation

The require-namespace-owner-label policy runs in Audit mode and requires an owner label to identify the team responsible for each namespace. Namespaces without this label are allowed, but Kyverno records the violations in ClusterPolicyReport objects.

Create namespaces billing, dev-apps, and payments without an owner label:

# kubectl create ns billing
# kubectl create ns dev-apps
# kubectl create ns payments

This generates three alerts for rule 101910, one for each namespace. 

Audit policy violation

The disallow-privileged-containers policy runs in Audit mode and checks whether containers have the privileged mode enabled. Kyverno allows pods that fail this check and records the violations in policy reports.

1. Create a privileged pod on the Kubernetes endpoint:

# kubectl apply -f - <<EOF
apiVersion: v1
kind: Pod
metadata:
  name: privileged-app
  namespace: dev-apps
spec:
  containers:
    - name: app
      image: registry.k8s.io/pause:3.9
      securityContext:
        privileged: true
EOF

pod/privileged-app created

This generates an alert for rule 101901.

Repeated namespace violations

Multiple policy findings in one namespace can indicate a shared workload configuration that needs review. Wazuh correlates new or recurring Audit report findings for the same namespace and generates an alert when it receives five or more alerts within five minutes.

1. Create five privileged pods in the payments namespace:

# for i in 1 2 3 4 5; do
kubectl apply -f - <<EOF
apiVersion: v1
kind: Pod
metadata:
  name: legacy-batch-$i
  namespace: payments
spec:
  containers:
    - name: app
      image: registry.k8s.io/pause:3.9
      securityContext:
        privileged: true
EOF
done

This generates four alerts for rule 101901 and an alert for rule 101903.

2. Confirm that the pods are running:

# kubectl get pods -n payments

NAME             READY   STATUS    RESTARTS   AGE
legacy-batch-1   1/1     Running   0          8s
legacy-batch-2   1/1     Running   0          7s
legacy-batch-3   1/1     Running   0          5s
legacy-batch-4   1/1     Running   0          3s
legacy-batch-5   1/1     Running   0          2s

Denied admission request

The restrict-image-registries policy runs in Deny mode and permits container images only from registry.k8s.io. Kyverno rejects pod creation requests that reference other registries and records the failed evaluations in Kubernetes events.

1. Create a manifest for a pod that uses an unapproved container registry:

# cat > /tmp/untrusted-registry-pod.yaml <<EOF
apiVersion: v1
kind: Pod
metadata:
  name: untrusted-registry-app
  namespace: payments
spec:
  containers:
    - name: app
      image: docker.io/library/nginx:1.27
EOF

2. Apply the manifest:

# kubectl apply -f /tmp/untrusted-registry-pod.yaml

Kyverno rejects the admission request:

Error from server: error when creating "/tmp/untrusted-registry-pod.yaml": admission webhook "vpol.validate.kyverno.svc-fail" denied the request: Policy restrict-image-registries failed: Container images must be pulled from the approved registry registry.k8s.io.

This generates an alert for rule 101906.

Policy switched to Deny

Changing the disallow-privileged-containers policy from Audit to Deny affects subsequent admission requests without evicting existing pods. Existing privileged pods remain visible through failed policy reports, while new privileged pod creation requests are rejected.

1. Change the disallow-privileged-containers policy to Deny:

# kubectl patch validatingpolicy disallow-privileged-containers --type=json \
  -p='[{"op":"replace","path":"/spec/validationActions","value":["Deny"]}]'

This generates six alerts for rule 101905, one for each privileged pod that is still running.

2. Create a privileged pod:

# kubectl apply -f - <<EOF
apiVersion: v1
kind: Pod
metadata:
  name: privileged-retry
  namespace: dev-apps
spec:
  containers:
    - name: app
      image: registry.k8s.io/pause:3.9
      securityContext:
        privileged: true
EOF

Error from server: error when creating "STDIN": admission webhook "vpol.validate.kyverno.svc-fail" denied the request: Policy disallow-privileged-containers failed: Privileged mode is disallowed. The fields spec.containers[*].securityContext.privileged must be unset or set to false.

This generates an alert for rule 101902.

PolicyException declaration

A PolicyException declares an exemption for selected resources from a policy. Monitoring these declarations helps security teams review changes that could weaken enforcement.

Create an exception declaration targeting the batch pods, against the privileged-container policy:

# kubectl apply -f - <<EOF
apiVersion: policies.kyverno.io/v1
kind: PolicyException
metadata:
  name: legacy-batch-exemption
  namespace: payments
spec:
  policyRefs:
    - name: disallow-privileged-containers
      kind: ValidatingPolicy
  matchConditions:
    - name: exempt-legacy-batch
      expression: "object.metadata.name.startsWith('legacy-batch-')"
EOF

Warning: PolicyException resources would not be processed until it is enabled.
policyexception.policies.kyverno.io/legacy-batch-exemption created

This generates an alert for rule 101909.

Weakened and removed policy

A local policy manifest change does not affect the policy running in Kubernetes until you apply the manifest. Wazuh file integrity monitoring (FIM) detects edits to the local manifests. The collector detects changes to the live policies by comparing each policy’s settings with its previous run.

1. Downgrade the restrict-image-registries policy from Deny to Audit in the manifest:

# sed -i 's/- Deny/- Audit/' /opt/kyverno-policies/restrict-image-registries.yaml

This generates an alert for rule 101904.

2. Apply the edited manifest:

# kubectl apply -f /opt/kyverno-policies/restrict-image-registries.yaml
# systemctl start kyverno-wazuh-collector.service
# kubectl get validatingpolicy restrict-image-registries -o jsonpath='{.spec.validationActions}'

This generates an alert for rule 101907.

3. Remove the policy from the cluster:

# kubectl delete validatingpolicy restrict-image-registries

This generates an alert for rule 101908.

Disabled admission evaluation

A policy keeps its validation action when admission evaluation is disabled, so the action alone doesn’t show the change. The collector reads the admission setting and alerts when it is turned off.

1. Disable admission evaluation for the require-namespace-owner-label policy:

# kubectl patch validatingpolicy require-namespace-owner-label --type=json \
  -p='[{"op":"add","path":"/spec/evaluation/admission","value":{"enabled":false}}]'
# systemctl start kyverno-wazuh-collector.service

The policy remains configured with Audit, but it no longer evaluates admission requests.

This generates an alert for rule 101911.

2. Restore the admission evaluation before continuing:

# kubectl patch validatingpolicy require-namespace-owner-label --type=json \
-p='[{"op":"replace","path":"/spec/evaluation/admission/enabled","value":true}]'
# systemctl start kyverno-wazuh-collector.service

Expired policy exception

A PolicyException can include an expiry time to limit the duration of a declared exemption. The collector reports the declaration when it first discovers it and generates a separate alert after its expiry time passes.

1. Create a PolicyException object that expires in two minutes:

# EXPIRY=$(date -u -d '+2 minutes' +%Y-%m-%dT%H:%M:%SZ)
# kubectl apply -f - <<EOF
apiVersion: policies.kyverno.io/v1
kind: PolicyException
metadata:
  name: temporary-batch-exemption
  namespace: payments
spec:
  expiresAt: "${EXPIRY}"
  policyRefs:
    - name: disallow-privileged-containers
      kind: ValidatingPolicy
  matchConditions:
    - name: exempt-temporary-batch
      expression: "object.metadata.name.startsWith('temp-batch-')"
EOF

2. Wait 150 seconds for the exception to expire, then run the collector. The collector detects the expiry only while the PolicyException object exists, so it must run before you delete the object:

# sleep 150
# systemctl start kyverno-wazuh-collector.service

This generates an alert for rule 101912.

3. Remove the temporary PolicyException object:

# kubectl delete policyexception.policies.kyverno.io temporary-batch-exemption -n payments

Policy evaluation error

A policy expression can produce an error result when it accesses a field that is absent from a resource. This indicates that Kyverno could not complete the evaluation. Wazuh reports evaluation errors separately from policy violations.

1. Create a policy that reads the label without checking whether it exists:

# kubectl apply -f - <<EOF
apiVersion: policies.kyverno.io/v1
kind: ValidatingPolicy
metadata:
  name: require-team-label
spec:
  validationActions:
    - Audit
  evaluation:
    background:
      enabled: true
  matchConstraints:
    namespaceSelector:
      matchExpressions:
        - key: kubernetes.io/metadata.name
          operator: In
          values:
            - dev-apps
    resourceRules:
      - apiGroups: [""]
        apiVersions: ["v1"]
        operations: ["CREATE", "UPDATE"]
        resources: ["pods"]
  validations:
    - message: >-
        Pods must carry a team label identifying the owning team.
      expression: "object.metadata.labels['team'] == 'platform'"
EOF

2. Wait about 30 seconds for the background scan to finish, then inspect the reports for the pods in the dev-apps namespace:

# kubectl get policyreport -n dev-apps -o jsonpath='{range .items[*].results[*]}{.policy}{"  "}{.result}{"  "}{.message}{"\n"}{end}'

disallow-privileged-containers  fail  Privileged mode is disallowed. The fields spec.containers[*].securityContext.privileged must be unset or set to false.
require-team-label  error  error: no such key: labels

This generates an alert for rule 101914.

Policy replaced by a new object

Deleting a policy and recreating it under the same name leaves the name in place while the object behind it changes. The collector compares object UIDs between runs, so the replacement is visible even though the policy list looks unchanged.

1. Stop the timer so both commands fall between the same two collector runs:

# systemctl stop kyverno-wazuh-collector.timer

2. Delete the require-namespace-owner-label policy and recreate it:

# kubectl delete validatingpolicy require-namespace-owner-label
# kubectl apply -f /opt/kyverno-policies/require-namespace-owner-label.yaml

3. Start the timer and collector service:

# systemctl start kyverno-wazuh-collector.service
# systemctl start kyverno-wazuh-collector.timer

This generates an alert for rule 101915. It also generates two alerts for rule 101910, for the dev-apps and payments namespaces, because the collector tracks each finding per policy object and reports findings under the recreated policy as new. If the collector runs between the delete and the apply, it reports rule 101908 instead, because the policy is absent at that moment.

Visualizing the alerts

Perform the following steps on the Wazuh dashboard to view the generated alerts:

1. Navigate to Threat intelligence > Threat Hunting > Events.
2. Search for rule.groups:kyverno.

Conclusion

Kyverno provides policy enforcement and compliance visibility for Kubernetes workloads. PolicyReports identify existing resources that violate configured policies, including violations detected through background evaluation. Events record admission requests that Kyverno rejects before Kubernetes creates the requested resource.

This integration uses a custom Python collector to retrieve Kyverno policy findings and configuration changes. Wazuh correlates Kyverno findings with policy configuration changes to provide centralized visibility into policy enforcement. The integration also identifies changes that can weaken enforcement, including policy downgrades, policy removals, and declared exceptions.

Wazuh file integrity monitoring complements this visibility by detecting additions, modifications, and deletions of local Kyverno policy manifests. Together, these capabilities provide a centralized view of policy violations, rejected deployments, and changes to the policies governing Kubernetes workloads.

Wazuh is a free and open source security platform. Check out our documentation to learn more, and join our community for help and discussion.

References

• Kyverno policy reports
• Kyverno ValidatingPolicy
• Wazuh custom rules
• Wazuh log data collection
• Wazuh file integrity monitoring
• MITRE ATT&CK T1610: Deploy Container