Kubernetes Secret Exposure Sentinel

id: kubernetes-secret-exposure-sentinel
namespace: company.team
description: |
  List Deployments in a namespace, flag any whose pod template injects a
  Secret through a plain environment variable (secretKeyRef) rather than a
  mounted volume, alert Slack on breach, and conditionally annotate the
  flagged Deployments for review. auto_remediate and dry_run are two
  independent gates; no container spec, env var, or Secret is ever changed.

triggers:
  - id: periodic_exposure_check
    type: io.kestra.plugin.core.trigger.Schedule
    description: Check every hour. Shipped disabled; enable it once k8s_namespace
      points at a real cluster and a manual run has been verified. Always fires
      in safe mode (auto_remediate false, dry_run true) regardless of this
      flow's own defaults.
    cron: "0 * * * *"
    disabled: true
    inputs:
      auto_remediate: "false"
      dry_run: "true"

inputs:
  - id: k8s_namespace
    type: STRING
    displayName: "Kubernetes Namespace"
    defaults: "default"
    description: Namespace whose Deployments are audited for Secret-as-env-var exposure.

  - id: max_env_secret_refs
    type: INT
    displayName: "Max Allowed Exposed Deployments"
    defaults: 0
    description: Breach when the count of Deployments injecting a Secret via a plain
      env var exceeds this number. Default 0 means any such Deployment at all is
      a breach.

  - id: auto_remediate
    type: SELECT
    displayName: "Auto Remediate"
    defaults: "false"
    values:
      - "false"
      - "true"
    description: If "true" and exposed Deployments are found, annotate each one with
      a review flag via kubectl Patch (still gated by dry_run below). If
      "false", the risk is only reported.

  - id: dry_run
    type: SELECT
    displayName: "Dry Run"
    defaults: "true"
    values:
      - "true"
      - "false"
    description: If "true", remediation is only logged, no annotation is applied.
      Must be explicitly "false", together with auto_remediate "true", for the
      flag to actually be written.

tasks:
  - id: list_deployments
    type: io.kestra.plugin.kubernetes.kubectl.Get
    description: Fetches every Deployment in k8s_namespace with fetchType STORE,
      which returns a uri pointing to the raw kubectl JSON (a DeploymentList,
      carrying the full spec.template.spec for every Deployment) - verified from
      the plugin's own output schema. FETCH/FETCH_ONE mode only return
      metadataItems/statusItems, not the pod template spec this audit needs.
    connection:
      masterUrl: "{{ secret('K8S_MASTER_URL') }}"
      oauthToken: "{{ secret('K8S_TOKEN') }}"
    namespace: "{{ inputs.k8s_namespace }}"
    resourceType: deployments
    fetchType: STORE

  - id: evaluate_secret_exposure
    type: io.kestra.plugin.scripts.python.Script
    description: Reads the raw DeploymentList and flags any Deployment whose pod
      template has a container with env[].valueFrom.secretKeyRef set - a Secret
      injected as a plain environment variable, which is more easily leaked
      through kubectl describe, process dumps, or verbose logging than a Secret
      mounted as a volume. There is no structured policy-evaluation task in this
      plugin, so the check runs here against the raw spec the Get task actually
      returns.
    inputFiles:
      deployments.json: "{{ outputs.list_deployments.uri }}"
    script: |
      import json
      from kestra import Kestra

      with open("deployments.json") as f:
          data = json.load(f)

      flagged = set()
      for dep in data.get("items", []):
          name = dep.get("metadata", {}).get("name")
          containers = dep.get("spec", {}).get("template", {}).get("spec", {}).get("containers", []) or []
          for c in containers:
              for env in (c.get("env") or []):
                  value_from = env.get("valueFrom") or {}
                  if value_from.get("secretKeyRef"):
                      flagged.add(name)
                      break

      flagged_deployments = sorted(flagged)

      Kestra.outputs({
          "flagged_count": len(flagged_deployments),
          "flagged_deployments": flagged_deployments,
      })

  - id: evaluate_exposure_gate
    type: io.kestra.plugin.core.flow.If
    description: Breach when at least one Deployment injects a Secret through a
      plain env var.
    condition: "{{ outputs.evaluate_secret_exposure.vars.flagged_count >
      inputs.max_env_secret_refs }}"
    then:
      - id: route_remediation
        type: io.kestra.plugin.core.flow.Switch
        description: auto_remediate only decides whether remediation is considered at
          all; dry_run, checked separately below, decides whether the annotation
          is actually written.
        value: "{{ inputs.auto_remediate }}"
        cases:
          "true":
            - id: check_dry_run
              type: io.kestra.plugin.core.flow.If
              condition: "{{ inputs.dry_run == \"false\" }}"
              then:
                - id: flag_exposed_deployments
                  type: io.kestra.plugin.core.flow.Loop
                  description: Annotate each flagged Deployment one at a time, so one failed patch
                    never blocks the rest of the batch.
                  values: "{{ outputs.evaluate_secret_exposure.vars.flagged_deployments }}"
                  concurrencyLimit: 4
                  tasks:
                    - id: annotate_exposed_deployment
                      type: io.kestra.plugin.kubernetes.kubectl.Patch
                      description: A JSON_MERGE patch adding only an annotation, matching this
                        plugin's own documented Secret-rotation example. It
                        never touches spec.template, env, or any Secret - it
                        only marks the Deployment for human review.
                      connection:
                        masterUrl: "{{ secret('K8S_MASTER_URL') }}"
                        oauthToken: "{{ secret('K8S_TOKEN') }}"
                      namespace: "{{ inputs.k8s_namespace }}"
                      resourceType: deployments
                      resourceName: "{{ item.value }}"
                      patchStrategy: JSON_MERGE
                      patch: |
                        {
                          "metadata": {
                            "annotations": {
                              "security.kestra.io/secret-env-exposure": "flagged"
                            }
                          }
                        }
                - id: log_remediation_applied
                  type: io.kestra.plugin.core.log.Log
                  message: "Annotated {{ outputs.evaluate_secret_exposure.vars.flagged_count }}
                    Deployment(s) in {{ inputs.k8s_namespace }} with
                    security.kestra.io/secret-env-exposure=flagged."
              else:
                - id: log_dry_run_remediation
                  type: io.kestra.plugin.core.log.Log
                  message: "DRY RUN: would annotate {{
                    outputs.evaluate_secret_exposure.vars.flagged_count }}
                    Deployment(s) in {{ inputs.k8s_namespace }} with
                    security.kestra.io/secret-env-exposure=flagged (dry_run is
                    true). No change made."
          "false":
            - id: log_remediation_disabled
              type: io.kestra.plugin.core.log.Log
              message: "auto_remediate is false: {{
                outputs.evaluate_secret_exposure.vars.flagged_count }}
                Deployment(s) in {{ inputs.k8s_namespace }} inject a Secret via
                a plain env var, but no annotation was attempted."
      - id: alert_secret_exposure_risk
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Always alert on any risk, independent of whether it was remediated.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "Kubernetes secret exposure sentinel: {{ outputs.evaluate_secret_exposure.vars.flagged_count }} Deployment(s) in namespace {{ inputs.k8s_namespace }} inject a Secret via a plain environment variable instead of a mounted volume. auto_remediate: {{ inputs.auto_remediate }}. dry_run: {{ inputs.dry_run }}. Remediated: {{ inputs.auto_remediate == \"true\" and inputs.dry_run == \"false\" }}. Execution {{ execution.id }}."
          }
    else:
      - id: log_exposure_healthy
        type: io.kestra.plugin.core.log.Log
        message: "{{ inputs.k8s_namespace }} is healthy: 0 Deployment(s) inject a Secret
          via a plain environment variable."

  - id: log_audit_complete
    type: io.kestra.plugin.core.log.Log
    description: Final status line for every run, independent of which branch fired,
      so the execution history always shows the headline numbers.
    message: "Secret exposure audit complete for {{ inputs.k8s_namespace }}.
      flagged_count={{ outputs.evaluate_secret_exposure.vars.flagged_count }}.
      threshold={{ inputs.max_env_secret_refs }}. auto_remediate={{
      inputs.auto_remediate }}. dry_run={{ inputs.dry_run }}."

errors:
  - id: alert_execution_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert when the flow itself fails outside the handled branches, for
      example an unreachable API server or a token without rights to list
      Deployments.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "Kubernetes secret exposure sentinel FAILED in execution {{ execution.id }} for namespace {{ inputs.k8s_namespace }}. The exposure audit did not complete; investigate before trusting the last known exposure status."
      }