PipAuditWatch - Python dependency CVE watcher

id: pipauditwatch
namespace: company.team
description: Scans Python requirements for known CVEs with pip-audit and alerts on Slack.

labels:
  project: pipauditwatch
  team: supply-chain

inputs:
  - id: requirements
    type: STRING
    displayName: "Requirements content"
    description: paste your requirements.txt content here
    defaults: |
      gunicorn==23.0.0
      requests==2.34.2
      urllib3==1.26.5
  - id: slack_webhook
    type: STRING
    displayName: "Slack Webhook URL"
    description: Incoming webhook URL of the Slack channel for alerts

triggers:
  - id: schedule
    type: io.kestra.plugin.core.trigger.Schedule
    description: scans dependencies every morning
    cron: "0 8 * * *"
    timezone: UTC
    inputs:
      requirements: |
        gunicorn==23.0.0
        requests==2.34.2
        urllib3==1.26.5
      slack_webhook: "{{ secret('SLACK_WEBHOOK') }}"

tasks:
  - id: find_vulns
    type: io.kestra.plugin.scripts.python.Script
    description: runs pip-audit and collects new vulnerabilities
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
      user: "0"
    containerImage: python:3.11
    dependencies:
      - kestra
      - pip-audit
    inputFiles:
      requirements.txt: "{{ inputs.requirements }}"
    outputFiles:
      - audit.json
    script: |
      import json
      import subprocess
      import sys
      from kestra import Kestra

      # pip-audit exits 1 when it finds anything; the JSON is still written.
      # called as a module because the pip-audit CLI binary is not on PATH
      # in Kestra's python runner, only the installed package is.
      result = subprocess.run(
          [sys.executable, '-m', 'pip_audit', '--requirement', 'requirements.txt', '--format', 'json', '-o', 'audit.json'],
          capture_output=True,
          text=True,
      )
      logger = Kestra.logger()
      logger.info(f"pip-audit exit code: {result.returncode}")
      if result.stderr.strip():
          logger.info(f"pip-audit stderr tail: {result.stderr.strip()[-2000:]}")

      try:
          with open('audit.json') as f:
              report = json.load(f)
      except (FileNotFoundError, json.JSONDecodeError) as e:
          # a crashed scanner must fail loudly, never report a fake "clean"
          raise RuntimeError(
              f"pip-audit produced no usable report ({type(e).__name__}). "
              f"stderr tail: {result.stderr.strip()[-1000:]}"
          )

      vulns = []
      # pip-audit 1.x nests findings under "vulnerabilities", 2.x under "dependencies"
      dep_entries = report.get('vulnerabilities') or report.get('dependencies') or []
      for dep in dep_entries:
          pkg = dep.get('name', '?')
          ver = dep.get('version', '?')
          for v in dep.get('vulns', []) or []:
              vid = v.get('id', '?')
              aliases = [a for a in (v.get('aliases') or []) if a != vid]
              fixes = v.get('fix_versions') or []
              vulns.append({
                  'id': f"{pkg}@{ver}:{vid}",
                  'package': pkg,
                  'version': ver,
                  'vid': vid,
                  'aliases': aliases,
                  'fixes': fixes,
              })

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

      new = [v for v in vulns if v['id'] not in already]

      lines = ["*PipAuditWatch - vulnerable Python dependencies*", ""]
      for v in new:
          alias_txt = f" ({', '.join(v['aliases'])})" if v['aliases'] else ""
          fix_txt = f"fixed in {', '.join(v['fixes'])}" if v['fixes'] else "no fixed version published"
          lines.append(f"- *{v['package']} {v['version']}*: {v['vid']}{alias_txt} - {fix_txt}")
      lines.append("")
      lines.append("bump the package and the finding drops out on the next run.")
      alert_text = "\n".join(lines)

      Kestra.outputs({
          'new_vulns': new,
          'count': len(new),
          'alert_text': alert_text,
          'new_state_json': json.dumps(sorted(v['id'] for v in vulns)),
      })

  - id: maybe_alert
    type: io.kestra.plugin.core.flow.If
    description: alerts only about new vulnerabilities
    condition: "{{ outputs.find_vulns.vars.count > 0 }}"
    then:
      - id: send_alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: posts the vulnerability alert to Slack
        url: "{{ inputs.slack_webhook }}"
        messageText: "{{ outputs.find_vulns.vars.alert_text }}"

  - id: save_state
    type: io.kestra.plugin.core.kv.Set
    description: remembers reported CVEs, so fixed ones alert again if reintroduced
    key: "pipauditwatch-state"
    value: "{{ outputs.find_vulns.vars.new_state_json }}"
    overwrite: true