Docker Registry Image Tag Immutability Guard

id: docker-registry-image-tag-immutability-guard
namespace: company.team
description: |
  Container image tag immutability and supply-chain security auditor. Scans Kubernetes container image
  references to flag mutable tags such as :latest or unpinned digests, alerting DevSecOps teams on Slack.

concurrency:
  limit: 1

inputs:
  - id: container_images_list
    type: STRING
    displayName: Container Images List
    defaults: "registry.company.com/payment-service:latest,registry.company.com/aut\
      h-service:v2.1.0@sha256:4b227777d4dd1fc61c6f884f48e18073,registry.company\
      .com/web-frontend:staging"
    description: "Comma-separated list of container image references extracted from
      Kubernetes manifests or deployment manifests."

  - id: disallow_latest_tag
    type: BOOL
    displayName: Disallow :latest Tag
    defaults: true
    description: "Whether to treat the :latest tag or missing tag as an immediate
      security violation."

  - id: require_sha256_digest
    type: BOOL
    displayName: Require SHA256 Digest
    defaults: false
    description: "When true, mandates that every container image reference must
      include an immutable @sha256: digest."

  - id: slack_channel
    type: STRING
    displayName: Slack Channel
    defaults: "#security-alerts"
    description: "Slack channel destination for container image immutability alerts."

triggers:
  - id: daily_image_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: "Audits container image tag immutability daily at 07:00 UTC."
    cron: "0 7 * * *"

tasks:
  - id: audit_image_tags
    type: io.kestra.plugin.scripts.python.Script
    description: "Parses container image URIs to verify tag immutability and digest
      pinning compliance."
    containerImage: python:3.11-slim
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    outputFiles:
      - image_immutability_report.json
    script: |
      import re
      import json
      from kestra import Kestra

      images_raw = "{{ inputs.container_images_list }}"
      disallow_latest = "{{ inputs.disallow_latest_tag }}" == "true"
      require_digest = "{{ inputs.require_sha256_digest }}" == "true"

      image_list = [img.strip() for img in images_raw.split(",") if img.strip()]

      violations = []
      compliant_images = []

      # Regex pattern for image parsing: [registry/][namespace/]repository[:tag][@sha256:digest]
      digest_regex = re.compile(r"@sha256:[a-fA-F0-9]{64}")

      for img in image_list:
          reasons = []
          has_digest = bool(digest_regex.search(img))

          # Strip digest to inspect tag
          img_no_digest = img.split("@")[0] if "@" in img else img
          parts = img_no_digest.split(":")

          if len(parts) == 1:
              tag = "latest"  # Docker default when tag is omitted
              is_default_latest = True
          else:
              tag = parts[-1]
              is_default_latest = False

          if disallow_latest and (tag == "latest" or is_default_latest):
              reasons.append("Uses mutable 'latest' tag or omitted tag")

          if tag in ["staging", "dev", "master", "main"]:
              reasons.append(f"Uses branch-tracking mutable tag '{tag}'")

          if require_digest and not has_digest:
              reasons.append("Missing required immutable @sha256: digest pinning")

          if reasons:
              violations.append({
                  "image": img,
                  "detected_tag": tag,
                  "has_digest": has_digest,
                  "reasons": reasons
              })
          else:
              compliant_images.append({
                  "image": img,
                  "detected_tag": tag,
                  "has_digest": has_digest
              })

      has_violations = len(violations) > 0
      top_violating = violations[0]["image"] if violations else "none"
      top_reason = violations[0]["reasons"][0] if violations else "none"

      report = {
          "total_scanned": len(image_list),
          "violations_count": len(violations),
          "compliant_count": len(compliant_images),
          "violations": violations,
          "compliant_images": compliant_images
      }

      with open("image_immutability_report.json", "w") as f:
          json.dump(report, f, indent=2)

      Kestra.outputs({
          "has_violations": has_violations,
          "violations_count": len(violations),
          "top_violating_image": top_violating,
          "top_reason": top_reason,
          "total_scanned": len(image_list)
      })

  - id: check_for_violations
    type: io.kestra.plugin.core.flow.If
    description: "Evaluates whether container image tag immutability violations were
      detected."
    condition: "{{ outputs.audit_image_tags.vars.has_violations }}"
    then:
      - id: send_slack_immutability_alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: "Dispatches high-priority Slack notification for mutable container
          image references."
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "{{ inputs.slack_channel }}",
            "text": "*SECURITY ALERT: Mutable Container Image References Detected*\n• *Violations Count*: {{ outputs.audit_image_tags.vars.violations_count }} of {{ outputs.audit_image_tags.vars.total_scanned }} images\n• *Primary Offender*: `{{ outputs.audit_image_tags.vars.top_violating_image }}`\n• *Violation Details*: {{ outputs.audit_image_tags.vars.top_reason }}\n• *Operational Impact*: Mutable tags such as `:latest` or unpinned digests allow unintended upstream image updates, breaking build reproducibility and exposing clusters to container supply chain attacks.\n• *Recommended Remediation*: Pin deployment manifests to immutable cryptographic digests (`image@sha256:...`) or strict immutable release tags."
          }

    else:
      - id: log_compliant_status
        type: io.kestra.plugin.core.log.Log
        description: "Logs confirmation that all scanned container images comply with
          immutability standards."
        message: "Container image tag immutability audit passed. All {{
          outputs.audit_image_tags.vars.total_scanned }} scanned images satisfy
          tag pinning policies."

  - id: export_audit_manifest
    type: io.kestra.plugin.core.debug.Return
    description: "Outputs structured JSON container security audit manifest."
    format: |
      {
        "pipeline": "docker-registry-image-tag-immutability-guard",
        "audited_at": "{{ execution.startDate }}",
        "total_scanned": {{ outputs.audit_image_tags.vars.total_scanned }},
        "violations_count": {{ outputs.audit_image_tags.vars.violations_count }},
        "status": "{{ outputs.audit_image_tags.vars.has_violations ? 'ACTION_REQUIRED' : 'COMPLIANT' }}"
      }