CertWatch - Kubernetes TLS certificate expiry watcher

id: certwatch
namespace: company.team
description: Watches TLS certificates in a namespace and alerts on Slack before they expire.

labels:
  project: certwatch
  team: platform

inputs:
  - id: namespace
    type: STRING
    displayName: "Namespace"
    description: the namespace whose TLS secrets to watch
    defaults: default
  - id: expiry_days
    type: INT
    displayName: "Expiry threshold (days)"
    description: a certificate expiring within this many days counts as troubled
    defaults: 14
  - id: slack_webhook
    type: STRING
    displayName: "Slack Webhook URL"
    description: Incoming webhook URL of the Slack channel for the alerts

triggers:
  - id: schedule
    type: io.kestra.plugin.core.trigger.Schedule
    description: checks certificate expiry every morning
    cron: "0 8 * * *"
    timezone: UTC
    inputs:
      namespace: default
      expiry_days: 14
      slack_webhook: "{{ secret('SLACK_WEBHOOK') }}"

tasks:
  - id: list_secrets
    type: io.kestra.plugin.scripts.shell.Commands
    description: lists all secrets in the namespace as JSON
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
      user: "0"
    containerImage: bitnami/kubectl
    inputFiles:
      client.crt: "{{ secret('CLIENT_CRT') }}"
      client.key: "{{ secret('CLIENT_KEY') }}"
    commands:
      - kubectl get secrets -n {{ inputs.namespace }} -o json
        --client-certificate=client.crt --client-key=client.key --server={{
        secret('MASTERURL') }} --insecure-skip-tls-verify=true > secrets.json
    outputFiles:
      - secrets.json

  - id: find_trouble
    type: io.kestra.plugin.scripts.python.Script
    description: finds certificates expiring soon, drops ones already alerted on
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
      user: "0"
    containerImage: python:3.11
    dependencies:
      - kestra
    script: |
      import base64
      import json
      import subprocess
      import tempfile
      from datetime import datetime, timezone
      from kestra import Kestra

      threshold = int("""{{ inputs.expiry_days }}""")
      namespace = """{{ inputs.namespace }}"""

      with open("""{{ outputs.list_secrets.outputFiles['secrets.json'] }}""") as f:
          secrets = json.load(f)['items']

      # names we already alerted on, from the KV store (null on first run -> empty)
      raw_state = """{{ kv('certwatch-state-' ~ inputs.namespace, errorOnMissing=false) }}"""
      already = set(json.loads(raw_state) if raw_state.strip() else [])

      def days_left(pem_bytes):
          # ask openssl for the expiry date, it is on the image already
          with tempfile.NamedTemporaryFile(suffix='.crt') as tmp:
              tmp.write(pem_bytes)
              tmp.flush()
              out = subprocess.run(
                  ['openssl', 'x509', '-enddate', '-noout', '-in', tmp.name],
                  capture_output=True, text=True, check=True,
              ).stdout.strip()
          # out looks like: notAfter=Nov  1 12:00:00 2026 GMT
          expiry = datetime.strptime(out.split('=', 1)[1], '%b %d %H:%M:%S %Y %Z')
          expiry = expiry.replace(tzinfo=timezone.utc)
          return (expiry - datetime.now(timezone.utc)).days

      troubled = []
      for secret in secrets:
          if secret.get('type') != 'kubernetes.io/tls':
              continue  # only TLS secrets carry certificates
          name = secret['metadata']['name']
          pem = base64.b64decode(secret['data']['tls.crt'])
          left = days_left(pem)
          if left <= threshold:
              troubled.append({'name': name, 'days_left': left})

      troubled_names = set(t['name'] for t in troubled)
      new_trouble = [t for t in troubled if t['name'] not in already]
      # the new state is simply who is troubled right now;
      # renewed certs drop out on their own, so a future expiry alerts again

      lines = [f"*CertWatch - expiring certificates in '{namespace}'*", ""]
      for t in new_trouble:
          lines.append(f"- *{t['name']}*: expires in {t['days_left']} days (threshold {threshold})")
          lines.append(f"  inspect: `kubectl describe secret {t['name']} -n {namespace}`")
      lines.append("")
      alert_text = "\n".join(lines)

      Kestra.outputs({
          'troubled': new_trouble,
          'count': len(new_trouble),
          'new_state_json': json.dumps(sorted(troubled_names)),
          'alert_text': alert_text,
      })

  - id: maybe_alert
    type: io.kestra.plugin.core.flow.If
    description: only alerts when there is something new to report
    condition: "{{ outputs.find_trouble.vars.count > 0 }}"
    then:
      - id: send_alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: posts the expiring certificates to Slack
        url: "{{ inputs.slack_webhook }}"
        messageText: "{{ outputs.find_trouble.vars.alert_text }}"

  - id: save_state
    type: io.kestra.plugin.core.kv.Set
    description: remembers who is troubled now, so renewed certs alert again next time
    key: "certwatch-state-{{ inputs.namespace }}"
    value: "{{ outputs.find_trouble.vars.new_state_json }}"
    overwrite: true