NetworkPolicy Coverage Audits with kubectl

id: networkpolicy-coverage-audit
namespace: company.team
description: |
  Audit Kubernetes NetworkPolicy coverage from a weekly schedule or an
  on-demand webhook. The inventory is collected in the kubectl container and
  falls back to bundled namespace and policy fixtures when no cluster is
  reachable, a python audit computes the namespaces with no default deny
  policy plus every policy allowing 0.0.0.0/0, and when either list is
  non-empty the flow opens a GitHub issue, alerts Slack, and appends ready to
  apply remediation yaml per uncovered namespace before publishing the
  verdict and counts as outputs.

triggers:
  - id: weekly_coverage_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: Weekly NetworkPolicy coverage audit every Monday at 06:00 UTC,
      shipped disabled until an operator points the flow at a reachable cluster
      and switches it on.
    cron: "0 6 * * 1"
    disabled: true
  - id: on_demand_audit
    type: io.kestra.plugin.core.trigger.Webhook
    description: POST here to audit coverage on demand, for example after onboarding
      a namespace or before an external security review.
    key: "{{ secret('NETWORKPOLICY_WEBHOOK_KEY') }}"

inputs:
  - id: cluster_name
    type: STRING
    defaults: prod-eks-cluster
    description: Logical name of the cluster under audit, used in the issue title,
      the Slack message, and the remediation yaml header.
  - id: github_repository
    type: STRING
    defaults: your-org/platform-baselines
    description: Repository in owner/repo form that receives the coverage issue
      whenever a gap is found.

tasks:
  # --- 1. inventory collection ---
  - id: policy_workspace
    type: io.kestra.plugin.core.flow.WorkingDirectory
    description: One shared directory so the inventory files, the audit, the verdict
      gate, and the remediation yaml all read and write the same files.
    tasks:
      - id: collect_inventory
        type: io.kestra.plugin.scripts.shell.Commands
        description: Write the bundled namespace and NetworkPolicy fixtures with
          heredocs, then run best effort kubectl listings so a reachable cluster
          supersedes the fixtures with live inventory.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
          entryPoint: []
        containerImage: bitnami/kubectl:latest
        env:
          CLUSTER_NAME: "{{ inputs.cluster_name }}"
        commands:
          - |
            set -u
            echo "Collecting NetworkPolicy inventory for cluster ${CLUSTER_NAME}"

            cat > namespaces.json <<'JSON'
            {
              "apiVersion": "v1", "kind": "NamespaceList",
              "items": [
                {"metadata": {"name": "payments", "labels": {"tier": "critical", "environment": "prod"}}, "status": {"phase": "Active"}},
                {"metadata": {"name": "storefront", "labels": {"tier": "frontend", "environment": "prod"}}, "status": {"phase": "Active"}},
                {"metadata": {"name": "analytics", "labels": {"tier": "data", "environment": "staging"}}, "status": {"phase": "Active"}},
                {"metadata": {"name": "sandbox", "labels": {"tier": "experimental", "environment": "dev"}}, "status": {"phase": "Active"}}
              ]
            }
            JSON

            cat > networkpolicies.json <<'JSON'
            {
              "apiVersion": "networking.k8s.io/v1", "kind": "NetworkPolicyList",
              "items": [
                {"metadata": {"name": "default-deny", "namespace": "payments", "labels": {"app.kubernetes.io/managed-by": "platform"}}, "spec": {"podSelector": {}, "policyTypes": ["Ingress", "Egress"]}},
                {"metadata": {"name": "default-deny", "namespace": "storefront", "labels": {"app.kubernetes.io/managed-by": "platform"}}, "spec": {"podSelector": {}, "policyTypes": ["Ingress", "Egress"]}},
                {"metadata": {"name": "allow-all-ingress", "namespace": "analytics", "labels": {"app.kubernetes.io/managed-by": "legacy-scripts"}}, "spec": {"podSelector": {}, "policyTypes": ["Ingress"], "ingress": [{"from": [{"ipBlock": {"cidr": "0.0.0.0/0"}}]}]}}
              ]
            }
            JSON

            echo "live NetworkPolicy table from the cluster (best effort):"
            kubectl get networkpolicy -A || true
            kubectl get networkpolicy -A -o json > live-networkpolicies.json || true
            kubectl get namespace -o json > live-namespaces.json || true

            if [ -s live-networkpolicies.json ] && [ -s live-namespaces.json ]; then
              echo "live inventory captured, fixtures superseded"
            else
              rm -f live-networkpolicies.json live-namespaces.json
              echo "no reachable cluster, auditing the bundled fixtures"
            fi

            echo "namespaces fixture $(wc -c < namespaces.json) bytes, policies fixture $(wc -c < networkpolicies.json) bytes"

      # --- 2. coverage audit ---
      - id: audit_policies
        type: io.kestra.plugin.scripts.python.Script
        description: Read the inventory, mark every namespace without a default deny
          policy, flag every policy carrying a 0.0.0.0/0 or ::/0 rule, and emit
          the verdict through the outputs protocol.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        containerImage: python:3.12-slim
        env:
          CLUSTER_NAME: "{{ inputs.cluster_name }}"
        script: |
          import json
          import os

          cluster = os.environ.get("CLUSTER_NAME", "unknown-cluster")

          def load(path):
              if not os.path.exists(path) or os.path.getsize(path) == 0:
                  return None
              try:
                  with open(path) as handle:
                      document = json.load(handle)
              except ValueError:
                  return None
              return document if isinstance(document, dict) else None

          def entries(document):
              raw = (document or {}).get("items")
              if not isinstance(raw, list):
                  return []
              return [entry for entry in raw if isinstance(entry, dict)]

          def pick(primary, fallback):
              live = entries(load(primary))
              return live if live else entries(load(fallback))

          def metadata_of(policy):
              metadata = policy.get("metadata")
              return metadata if isinstance(metadata, dict) else {}

          def opens_everywhere(policy):
              spec = policy.get("spec") if isinstance(policy.get("spec"), dict) else {}
              for direction, peers in (("ingress", "from"), ("egress", "to")):
                  rules = spec.get(direction)
                  for rule in (rules if isinstance(rules, list) else []):
                      if not isinstance(rule, dict):
                          continue
                      for peer in (rule.get(peers) or []):
                          block = peer.get("ipBlock") if isinstance(peer, dict) else None
                          if isinstance(block, dict) and str(block.get("cidr") or "") in ("0.0.0.0/0", "::/0"):
                              return True
              return False

          def is_default_deny(policy):
              spec = policy.get("spec") if isinstance(policy.get("spec"), dict) else {}
              selector = spec.get("podSelector") if isinstance(spec.get("podSelector"), dict) else {}
              if selector.get("matchLabels") or selector.get("matchExpressions"):
                  return False
              policy_types = spec.get("policyTypes")
              policy_types = ["Ingress", "Egress"] if policy_types is None else policy_types
              if not isinstance(policy_types, list):
                  return False
              if "Ingress" not in [str(entry) for entry in policy_types]:
                  return False
              return not opens_everywhere(policy)

          namespace_names = set()
          for entry in pick("live-namespaces.json", "namespaces.json"):
              name = metadata_of(entry).get("name")
              if name:
                  namespace_names.add(str(name))
          namespace_names = sorted(namespace_names)

          policies = pick("live-networkpolicies.json", "networkpolicies.json")
          denying = {str(metadata_of(p).get("namespace") or "") for p in policies if is_default_deny(p)}
          uncovered_ns = [name for name in namespace_names if name not in denying]
          permissive = sorted({str(metadata_of(p).get("namespace") or "") + "/" + str(metadata_of(p).get("name") or "unnamed")
                               for p in policies if opens_everywhere(p)})

          if not namespace_names:
              verdict = "NO_INVENTORY"
          elif uncovered_ns or permissive:
              verdict = "GAPS_FOUND"
          else:
              verdict = "LOCKED_DOWN"

          payload = {"verdict": verdict, "uncovered_ns": uncovered_ns,
                     "permissive": permissive, "namespaces": len(namespace_names)}
          print("cluster=%s namespaces=%d uncovered=%d permissive=%d verdict=%s"
                % (cluster, len(namespace_names), len(uncovered_ns), len(permissive), verdict))
          print("::" + json.dumps({"outputs": payload}) + "::")

  # --- 3. verdict gate ---
  - id: check_gaps
    type: io.kestra.plugin.core.flow.If
    description: Escalate only when at least one namespace lacks a default deny
      policy or at least one policy allows 0.0.0.0/0, a locked down cluster is
      recorded and left alone.
    condition: "{{ (outputs.audit_policies.vars.uncovered_ns | length) > 0 or
      (outputs.audit_policies.vars.permissive | length) > 0 }}"
    then:
      - id: open_coverage_issue
        type: io.kestra.plugin.github.issues.Create
        description: Open a labeled GitHub issue listing the uncovered namespaces and
          the over-permissive policies so the coverage gap is tracked as work.
        jwtToken: "{{ secret('GITHUB_TOKEN') }}"
        repository: "{{ inputs.github_repository }}"
        title: "NetworkPolicy coverage gaps in {{ inputs.cluster_name }}"
        body: |
          ## NetworkPolicy coverage audit

          Cluster **{{ inputs.cluster_name }}** scanned **{{ outputs.audit_policies.vars.namespaces }}** namespace(s).

          Namespaces without a default deny policy: {{ outputs.audit_policies.vars.uncovered_ns | join(', ') }}

          Policies allowing 0.0.0.0/0: {{ outputs.audit_policies.vars.permissive | join(', ') }}

          Execution {{ execution.id }}

          Apply the remediation yaml attached to this execution, then re-run the audit and confirm the verdict returns LOCKED_DOWN before closing this issue.
        labels:
          - kubernetes
          - network-policy
      - id: alert_slack
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Tell the platform channel how many namespaces are uncovered, which
          policies open the world, and that a GitHub issue is waiting.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "NetworkPolicy coverage audit found gaps in {{ inputs.cluster_name }} - {{ outputs.audit_policies.vars.namespaces }} namespace(s) scanned, {{ outputs.audit_policies.vars.uncovered_ns | length }} without a default deny policy, {{ outputs.audit_policies.vars.permissive | length }} allowing 0.0.0.0/0. {{ outputs.audit_policies.vars.uncovered_ns | join(', ') }}. Issue opened in {{ inputs.github_repository }}. Execution {{ execution.id }}."
          }
      - id: append_remediation
        type: io.kestra.plugin.core.flow.Loop
        description: Walk the uncovered namespaces one at a time and append a ready to
          apply remediation yaml block for each into the shared report.
        values: "{{ outputs.audit_policies.vars.uncovered_ns }}"
        concurrencyLimit: 1
        tasks:
          - id: append_namespace_yaml
            type: io.kestra.plugin.scripts.python.Script
            description: Append a default deny NetworkPolicy manifest for the current
              namespace, plus a tightened replacement for any allow-all policy
              living in that namespace.
            taskRunner:
              type: io.kestra.plugin.scripts.runner.docker.Docker
            containerImage: python:3.12-slim
            env:
              TARGET_NAMESPACE: "{{ item }}"
              PERMISSIVE: "{{ outputs.audit_policies.vars.permissive | join(',') }}"
              CLUSTER_NAME: "{{ inputs.cluster_name }}"
            script: |
              import json
              import os

              namespace = os.environ.get("TARGET_NAMESPACE", "").strip()
              cluster = os.environ.get("CLUSTER_NAME", "unknown-cluster")
              refs = [ref.strip() for ref in os.environ.get("PERMISSIVE", "").split(",") if ref.strip()]

              if not namespace:
                  raise SystemExit("TARGET_NAMESPACE is required to append remediation yaml")

              def manifest(name, note, ingress=None):
                  block = ["# cluster " + cluster + ", " + note,
                           "apiVersion: networking.k8s.io/v1",
                           "kind: NetworkPolicy",
                           "metadata:",
                           "  name: " + name,
                           "  namespace: " + namespace,
                           "spec:",
                           "  podSelector: {}",
                           "  policyTypes:",
                           "    - Ingress"]
                  return block + (ingress or []) + [""]

              lines = manifest("default-deny", "namespace " + namespace + " has no default deny NetworkPolicy")
              tightened = []
              for reference in refs:
                  parts = reference.split("/", 1)
                  if len(parts) != 2 or parts[0] != namespace:
                      continue
                  lines += manifest(parts[1], reference + " allows 0.0.0.0/0, replace it with namespace scoped sources",
                                    ["  ingress:",
                                     "    - from:",
                                     "        - namespaceSelector:",
                                     "            matchLabels:",
                                     "              kubernetes.io/metadata.name: " + namespace])
                  tightened.append(reference)

              with open("remediation.yaml", "a") as handle:
                  handle.write("\n".join(lines).rstrip() + "\n")

              print("appended remediation yaml for " + namespace)
              print("::" + json.dumps({"outputs": {"namespace": namespace, "tightened": tightened, "lines": len(lines)}}) + "::")
    else:
      - id: log_locked_down
        type: io.kestra.plugin.core.log.Log
        description: Record the all-clear so the execution history doubles as
          NetworkPolicy audit evidence.
        message: "NetworkPolicy coverage locked down for {{ inputs.cluster_name }} - {{
          outputs.audit_policies.vars.namespaces }} namespace(s) scanned, 0
          uncovered, 0 allowing 0.0.0.0/0, verdict {{
          outputs.audit_policies.vars.verdict }}."

errors:
  - id: notify_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Escalate any failure to the ops channel with the execution url.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {"text": "FAILED: {{ flow.id }} {{ execution.id }}"}

outputs:
  - id: verdict
    type: STRING
    description: LOCKED_DOWN when every audited namespace has a default deny policy
      and nothing allows 0.0.0.0/0, GAPS_FOUND when either is missing,
      NO_INVENTORY when no inventory could be read.
    value: "{{ outputs.audit_policies.vars.verdict }}"
  - id: uncovered_ns
    type: ARRAY
    description: 'Namespaces without a default deny policy, for example
      ["analytics", "sandbox"].'
    value: "{{ outputs.audit_policies.vars.uncovered_ns | toJson }}"
  - id: permissive
    type: ARRAY
    description: 'Policies allowing 0.0.0.0/0, written as namespace/name, for
      example ["analytics/allow-all-ingress"].'
    value: "{{ outputs.audit_policies.vars.permissive | toJson }}"
  - id: namespaces
    type: INT
    description: Number of namespaces present in the audited inventory, for example 4.
    value: "{{ outputs.audit_policies.vars.namespaces }}"

How it works

weekly_coverage_audit (io.kestra.plugin.core.trigger.Schedule, cron 0 6 * * 1) is the primary trigger and ships disabled: true, so nothing runs until an operator switches it on. on_demand_audit (io.kestra.plugin.core.trigger.Webhook, keyed) accepts an on-demand audit from a CI job or a laptop.
policy_workspace (io.kestra.plugin.core.flow.WorkingDirectory) opens one shared filesystem so the inventory, the audit, and the remediation yaml all see the same files.
collect_inventory (io.kestra.plugin.scripts.shell.Commands on the io.kestra.plugin.scripts.runner.docker.Docker runner, image bitnami/kubectl:latest) writes namespaces.json, four labeled namespaces, and networkpolicies.json, two default deny policies plus one allow-all ingress policy on 0.0.0.0/0 plus one namespace with no policy at all, with heredocs. It then runs kubectl get networkpolicy -A and captures the JSON listings with || true so an unreachable cluster never fails the run. When both live files come back non-empty they supersede the fixtures, otherwise they are removed and the bundled fixtures keep the audit deterministic offline.
audit_policies (io.kestra.plugin.scripts.python.Script on python:3.12-slim) reads the inventory, marks every namespace with no default deny policy, flags every policy carrying a 0.0.0.0/0 or ::/0 rule, and emits verdict, uncovered_ns, permissive, and namespaces through the outputs protocol.
check_gaps (io.kestra.plugin.core.flow.If) branches on the two lists. When either is non-empty, open_coverage_issue (io.kestra.plugin.github.issues.Create) files a labeled issue, alert_slack posts the counts, and append_remediation (io.kestra.plugin.core.flow.Loop, concurrencyLimit: 1) walks the uncovered namespaces so append_namespace_yaml appends a ready to apply default deny manifest per namespace, plus a tightened replacement for any allow-all policy in that namespace, into remediation.yaml.
When both lists are empty, log_locked_down (io.kestra.plugin.core.log.Log) records the all-clear, so the execution history doubles as audit evidence. The errors block alerts Slack whenever the inventory or the audit breaks, because a dead collector must never read as a locked down cluster.

What you get

Gateable results: verdict, uncovered_ns, permissive, and namespaces are flow outputs any downstream flow can branch on.
Escalation where engineers already work: a labeled GitHub issue naming every uncovered namespace and every allow-all policy, plus a Slack ping with the counts.
remediation.yaml: one default deny NetworkPolicy manifest per uncovered namespace, with a tightened replacement for each allow-all policy, ready to kubectl apply -f.

Who it's for

Platform and SRE teams that must prove every namespace ships a default deny baseline.
Security engineers chasing allow-all NetworkPolicies that quietly expose workloads to the internet.

Why orchestrate this with Kestra

A kubectl get networkpolicy -A is a one-shot command with no memory, no threshold, and no notification path, and nobody runs it twice. Kestra supplies the disabled-by-default weekly schedule as the safety net, a keyed webhook for on-demand runs, a hermetic kubectl container so the tool version lives in the flow file, a branch that separates escalation from the all-clear, execution history with the inventory attached, and routing into GitHub and Slack from one declarative definition. A cron entry on a bastion host gives you none of that, it only reports whatever the last person remembered to check.

Prerequisites

A worker with Docker access, the inventory runs in bitnami/kubectl:latest and the audit in python:3.12-slim.
A reachable Kubernetes API for live inventory. Without one the flow still completes on the bundled fixtures, which is exactly what makes it safe to run in CI or on a laptop.
A GitHub token that can open issues in github_repository (classic PAT scope repo, or fine-grained Issues: write) and a Slack incoming webhook for escalation and failure alerts.

Secrets

NETWORKPOLICY_WEBHOOK_KEY: key segment guarding the on-demand webhook URL, anyone holding it can trigger an audit.
GITHUB_TOKEN: token used to open the coverage issue.
SLACK_WEBHOOK_URL: Slack incoming webhook URL.

Quick start

Add the three secrets to your Kestra namespace.
Run the flow once with the default inputs and confirm verdict is GAPS_FOUND, uncovered_ns lists analytics and sandbox, permissive lists analytics/allow-all-ingress, and namespaces is 4.
Open remediation.yaml in the working directory of that execution and apply it to a test cluster to watch the verdict return LOCKED_DOWN.
Point github_repository at a repository where issues are welcome, re-run, and confirm the issue and the Slack message arrive, then trigger on_demand_audit at /api/v1/{tenant}/executions/webhook/{namespace}/networkpolicy-coverage-audit/{NETWORKPOLICY_WEBHOOK_KEY} or enable weekly_coverage_audit once a cluster is reachable from the worker.

How to extend

Audit a fleet by fanning the cluster_name input out with a Loop over a cluster list, one execution per cluster.
Add a core.flow.Fail task under check_gaps to turn a coverage gap into a red run for a CD pipeline.
Extend audit_policies with egress-only baselines, or with exemptions for system namespaces such as kube-system, then ship remediation.yaml to S3 as audit evidence.

Audit NetworkPolicy Coverage Across Every Namespace

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Audit NetworkPolicy Coverage Across Every Namespace

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Infrastructure Blueprints

Run Ansible Against NetBox Devices by Role and Site

New Relic APM Apdex Score Degradation Sentinel

Newman (Postman CLI) Automated API Contract Regression Gate

Provision a VM and Register It in NetBox

Netskope Security Alert Volume Sentinel

Nexus Repository Blobstore Storage Space Exhaustion Sentinel

New to Kestra?