Scheduled Container Registry Vulnerability Audit with Trivy and Slack Alerts

id: container-registry-trivy-cve-scan
namespace: company.security

inputs:
  - id: image
    type: STRING
    displayName: Container Image
    description: Full container image reference to audit (e.g.,
      docker.io/library/python:3.11-slim or ghcr.io/org/repo:latest).
    defaults: python:3.11-slim

  - id: critical_threshold
    type: INT
    displayName: Critical CVE Threshold
    description: Maximum acceptable CRITICAL severity vulnerabilities before
      triggering an alert.
    defaults: 0

  - id: high_threshold
    type: INT
    displayName: High CVE Threshold
    description: Maximum acceptable HIGH severity vulnerabilities before triggering
      an alert.
    defaults: 5

  - id: fail_on_critical
    type: BOOL
    displayName: Fail on Critical CVEs
    description: Terminate the execution with a FAILED status if critical
      vulnerabilities exceed threshold.
    defaults: false

  - id: notify_slack
    type: BOOL
    displayName: Notify Slack
    description: Send a summary digest to Slack when vulnerabilities exceed
      thresholds. Requires SLACK_WEBHOOK_URL secret.
    defaults: true

concurrency:
  limit: 1

triggers:
  - id: weekly_security_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: Run weekly vulnerability audit of production images every Monday at
      08:00 UTC.
    cron: "0 8 * * 1"

tasks:
  - id: scan_image
    type: io.kestra.plugin.scripts.shell.Commands
    description: Scan container image for HIGH and CRITICAL vulnerabilities using
      Aqua Security Trivy and output a JSON report.
    containerImage: aquasec/trivy:latest
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    env:
      TRIVY_USERNAME: "{{ secret('REGISTRY_USERNAME') }}"
      TRIVY_PASSWORD: "{{ secret('REGISTRY_PASSWORD') }}"
    outputFiles:
      - trivy-report.json
    commands:
      - |
        set -euo pipefail
        IMAGE="{{ inputs.image }}"
        echo "Starting Trivy vulnerability scan for image: ${IMAGE}"

        # Run trivy scan in non-blocking mode to capture complete JSON report
        trivy image \
          --format json \
          --output trivy-report.json \
          --severity CRITICAL,HIGH \
          --quiet \
          "${IMAGE}" || {
            EXIT_CODE=$?
            echo "Trivy scanner exited with code ${EXIT_CODE}"
            if [ ! -s trivy-report.json ]; then
              echo "Failed to generate Trivy report!" >&2
              exit "${EXIT_CODE}"
            fi
          }

        echo "Trivy scan completed. Report size: $(wc -c < trivy-report.json) bytes"

  - id: parse_report
    type: io.kestra.plugin.scripts.python.Script
    description: Parse the Trivy JSON report, aggregate vulnerability counts by
      severity, identify fixable CVEs, and generate an executive summary.
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    containerImage: python:3.11-slim
    inputFiles:
      trivy-report.json: "{{ outputs.scan_image.outputFiles['trivy-report.json'] }}"
    beforeCommands:
      - pip install --no-cache-dir kestra
    script: |
      import json
      from kestra import Kestra

      with open("trivy-report.json", "r") as f:
          data = json.load(f)

      critical_count = 0
      high_count = 0
      fixable_count = 0
      findings = []

      results = data.get("Results", [])
      for result in results:
          target = result.get("Target", "Unknown")
          vulnerabilities = result.get("Vulnerabilities") or []
          for vuln in vulnerabilities:
              sev = vuln.get("Severity", "UNKNOWN").upper()
              vuln_id = vuln.get("VulnerabilityID", "N/A")
              pkg_name = vuln.get("PkgName", "N/A")
              installed_ver = vuln.get("InstalledVersion", "N/A")
              fixed_ver = vuln.get("FixedVersion")

              if sev == "CRITICAL":
                  critical_count += 1
              elif sev == "HIGH":
                  high_count += 1

              if fixed_ver:
                  fixable_count += 1

              findings.append({
                  "id": vuln_id,
                  "severity": sev,
                  "package": pkg_name,
                  "installed": installed_ver,
                  "fixed": fixed_ver or "No fix available",
                  "target": target
              })

      # Sort findings: CRITICAL first, then HIGH
      severity_order = {"CRITICAL": 0, "HIGH": 1}
      findings.sort(key=lambda x: severity_order.get(x["severity"], 2))

      # Build concise digest summary of top findings
      top_findings = findings[:10]
      summary_lines = [
          f"Image: {{ inputs.image }}",
          f"Total Vulnerabilities: {len(findings)} (Critical: {critical_count}, High: {high_count})",
          f"Actionable Fixable CVEs: {fixable_count}",
          "",
          "Top Findings:"
      ]

      for f_item in top_findings:
          summary_lines.append(
              f"- [{f_item['severity']}] {f_item['package']} ({f_item['installed']}) -> {f_item['id']} [Fixed: {f_item['fixed']}]"
          )

      if len(findings) > 10:
          summary_lines.append(f"... and {len(findings) - 10} more vulnerabilities.")

      digest_text = "\n".join(summary_lines)
      print(digest_text)

      Kestra.outputs({
          "total_count": len(findings),
          "critical_count": critical_count,
          "high_count": high_count,
          "fixable_count": fixable_count,
          "digest": digest_text
      })

  - id: evaluate_vulnerabilities
    type: io.kestra.plugin.core.flow.If
    description: Check if detected vulnerabilities exceed defined critical or high
      thresholds.
    condition: >-
      {{ (outputs.parse_report.vars.critical_count | number) >
      (inputs.critical_threshold | number) or
      (outputs.parse_report.vars.high_count | number) > (inputs.high_threshold |
      number) }}
    then:
      - id: notify_slack_alert
        type: io.kestra.plugin.core.flow.If
        description: Post vulnerability alert digest to Slack if notifications are enabled.
        condition: "{{ inputs.notify_slack }}"
        then:
          - id: slack_alert
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: Send structured vulnerability card to Slack security channel.
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {
                "text": "Security Alert: Container Vulnerability Audit Threshold Exceeded for `{{ inputs.image }}`!\n• Critical CVEs: {{ outputs.parse_report.vars.critical_count }} (threshold: {{ inputs.critical_threshold }})\n• High CVEs: {{ outputs.parse_report.vars.high_count }} (threshold: {{ inputs.high_threshold }})\n• Fixable CVEs: {{ outputs.parse_report.vars.fixable_count }}\n\n```\n{{ outputs.parse_report.vars.digest }}\n```\nExecution: {{ execution.id }}"
              }

      - id: check_fail_condition
        type: io.kestra.plugin.core.flow.If
        description: Conditionally terminate execution with failure status if
          fail_on_critical is set.
        condition: "{{ inputs.fail_on_critical and
          (outputs.parse_report.vars.critical_count | number) >
          (inputs.critical_threshold | number) }}"
        then:
          - id: fail_execution
            type: io.kestra.plugin.core.execution.Fail
            description: Fail execution due to unmitigated critical vulnerabilities.
            errorMessage: "Audit failed: Container image {{ inputs.image }} contains {{
              outputs.parse_report.vars.critical_count }} CRITICAL CVEs,
              exceeding allowed threshold of {{ inputs.critical_threshold }}."
    else:
      - id: log_clean_audit
        type: io.kestra.plugin.core.log.Log
        description: Record compliant audit status in execution logs.
        message: >-
          Container image audit PASSED for '{{ inputs.image }}'. Critical CVEs:
          {{ outputs.parse_report.vars.critical_count }}, High CVEs: {{
          outputs.parse_report.vars.high_count }}, Fixable: {{
          outputs.parse_report.vars.fixable_count }}. All vulnerabilities within
          acceptable thresholds.

errors:
  - id: on_scan_error_alert
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert the team if the vulnerability scanner fails to pull or
      analyze the container image.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "Container audit error in flow `{{ flow.id }}`: failed to scan image `{{ inputs.image }}`. Check registry credentials and worker network connectivity. Execution {{ execution.id }}."
      }

outputs:
  - id: total_vulnerabilities
    type: INT
    description: Total number of HIGH and CRITICAL vulnerabilities detected.
    value: "{{ outputs.parse_report.vars.total_count }}"
  - id: critical_count
    type: INT
    description: Count of CRITICAL severity vulnerabilities found.
    value: "{{ outputs.parse_report.vars.critical_count }}"
  - id: high_count
    type: INT
    description: Count of HIGH severity vulnerabilities found.
    value: "{{ outputs.parse_report.vars.high_count }}"
  - id: fixable_count
    type: INT
    description: Number of vulnerabilities that have an available vendor patch.
    value: "{{ outputs.parse_report.vars.fixable_count }}"
  - id: scan_digest
    type: STRING
    description: Formatted summary digest of the top findings.
    value: "{{ outputs.parse_report.vars.digest }}"