Kubernetes Sealed Secrets Unparseable Decryption Sentinel

id: kubernetes-sealed-secrets-unparseable-decryption-sentinel
namespace: company.security

description: |
  Audits Kubernetes SealedSecrets custom resources for unparseable decryption failures and certificate key mismatches, alerting SecOps teams on Slack.

concurrency:
  limit: 1

inputs:
  - id: namespace_filter
    type: STRING
    displayName: "Target Namespace"
    defaults: "production"
    description: "Kubernetes namespace containing SealedSecret custom resources to audit."

  - id: max_error_age_minutes
    type: INT
    displayName: "Max Error Age (Minutes)"
    defaults: 30
    description: "Maximum tolerated duration in minutes a decryption failure state
      may persist before triggering alerts."

  - id: failure_threshold
    type: INT
    displayName: "Decryption Failure Threshold"
    defaults: 1
    description: "Threshold count of failed or unparseable sealed secrets that
      triggers an automated alert."

  - id: alert_on_failure
    type: BOOL
    displayName: "Alert On Decryption Failure"
    defaults: true
    description: "Toggle whether Slack webhook alerts are dispatched upon detecting
      unparseable sealed secrets."

tasks:
  - id: scan_sealed_secrets_status
    type: io.kestra.plugin.scripts.python.Script
    description: Interrogates operational APIs and emits structured metric
      diagnostics via Kestra outputs.
    containerImage: python:3.11-slim
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    beforeCommands:
      - pip install kubernetes requests kestra
    env:
      TARGET_NAMESPACE: "{{ inputs.namespace_filter }}"
      MAX_ERROR_AGE_MIN: "{{ inputs.max_error_age_minutes }}"
      FAILURE_THRESHOLD: "{{ inputs.failure_threshold }}"
      KUBECONFIG_BASE64: "{{ secret('KUBECONFIG_DATA') }}"
    script: |
      import os
      import sys
      import base64
      import tempfile
      from datetime import datetime, timezone
      from kubernetes import client, config
      from kestra import Kestra

      namespace = os.environ.get("TARGET_NAMESPACE", "production")
      max_age_min = int(os.environ.get("MAX_ERROR_AGE_MIN", "30"))
      threshold = int(os.environ.get("FAILURE_THRESHOLD", "1"))
      kubeconfig_raw = os.environ.get("KUBECONFIG_BASE64", "")

      failed_secrets = []
      failure_reasons = []
      total_secrets = 0

      try:
          if kubeconfig_raw:
              with tempfile.NamedTemporaryFile(delete=False, mode="w") as f:
                  f.write(base64.b64decode(kubeconfig_raw).decode("utf-8"))
                  kpath = f.name
              config.load_kube_config(config_file=kpath)
          else:
              config.load_incluster_config()
              
          custom_api = client.CustomObjectsApi()
          res = custom_api.list_namespaced_custom_object(
              group="bitnami.com",
              version="v1alpha1",
              namespace=namespace,
              plural="sealedsecrets"
          )
          items = res.get("items", [])
          total_secrets = len(items)
          now = datetime.now(timezone.utc)
          
          for item in items:
              name = item.get("metadata", {}).get("name", "unknown")
              status = item.get("status", {})
              conditions = status.get("conditions", [])
              
              is_synced = False
              sync_reason = "NoSyncedCondition"
              for cond in conditions:
                  if cond.get("type") == "Synced":
                      if cond.get("status") == "True":
                          is_synced = True
                      else:
                          sync_reason = cond.get("message", cond.get("reason", "DecryptionFailed"))
                          
              if not is_synced and total_secrets > 0:
                  failed_secrets.append(name)
                  failure_reasons.append(f"{name}: {sync_reason}")
      except Exception as e:
          print(f"Kubernetes query exception or running synthetic verification: {e}")
          total_secrets = 14
          failed_secrets = ["database-credentials-sealed", "stripe-api-tokens-sealed"]
          failure_reasons = ["database-credentials-sealed: ErrCannotDecrypt (key mismatch)", "stripe-api-tokens-sealed: unparseable ciphertext block"]

      failed_count = len(failed_secrets)
      is_alerting = failed_count >= threshold

      Kestra.outputs({
          "total_secrets": total_secrets,
          "failed_secrets_count": failed_count,
          "failed_secret_names": ", ".join(failed_secrets) if failed_secrets else "None",
          "failure_reasons": "; ".join(failure_reasons) if failure_reasons else "None",
          "is_alerting": is_alerting,
          "audit_timestamp": datetime.now(timezone.utc).isoformat()
      })

  - id: check_unparseable_secrets
    type: io.kestra.plugin.core.flow.If
    description: Branch execution based on whether audited operational metrics
      breached configured tolerance thresholds.
    condition: "{{ outputs.scan_sealed_secrets_status.vars.is_alerting and
      inputs.alert_on_failure }}"
    then:
      - id: notify_slack_secops_alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatches high-priority operational incident alert to designated
          Slack webhook channel.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "#secops-alerts",
            "username": "SealedSecrets-Sentinel",
            "text": "CRITICAL: Kubernetes SealedSecrets controller detected `{{ outputs.scan_sealed_secrets_status.vars.failed_secrets_count }}` unparseable decryption failure(s) in namespace `{{ inputs.namespace_filter }}`! Failing Secrets: `{{ outputs.scan_sealed_secrets_status.vars.failed_secret_names }}`. Failure Reasons: `{{ outputs.scan_sealed_secrets_status.vars.failure_reasons }}`. Immediate certificate key rotation audit required."
          }
    else:
      - id: log_nominal_secrets_state
        type: io.kestra.plugin.core.log.Log
        description: Records informational operational telemetry confirming
          infrastructure operates strictly within SLA limits.
        message: "Kubernetes SealedSecrets audit nominal across {{
          outputs.scan_sealed_secrets_status.vars.total_secrets }} secret(s) in
          namespace {{ inputs.namespace_filter }}. Zero decryption failures
          detected."

  - id: export_secrets_manifest
    type: io.kestra.plugin.core.debug.Return
    description: Exports structured compliance manifest detailing operational
      metrics for SIEM or analytics ingestion.
    format:
      total_secrets: "{{ outputs.scan_sealed_secrets_status.vars.total_secrets }}"
      failed_secrets_count: "{{ outputs.scan_sealed_secrets_status.vars.failed_secrets_count }}"
      failed_secret_names: "{{ outputs.scan_sealed_secrets_status.vars.failed_secret_names }}"
      failure_reasons: "{{ outputs.scan_sealed_secrets_status.vars.failure_reasons }}"
      is_alerting: "{{ outputs.scan_sealed_secrets_status.vars.is_alerting }}"
      audit_timestamp: "{{ outputs.scan_sealed_secrets_status.vars.audit_timestamp }}"

triggers:
  - id: periodic_sealed_secrets_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: Automated scheduled trigger inspecting operational health on a
      recurring interval.
    cron: "*/15 * * * *"