Enforce a MinIO bucket lifecycle expiry and noncurrent version policy

id: minio-bucket-lifecycle-expiry-enforce-guard
namespace: company.team

description: |
  A bucket with no lifecycle configuration never expires anything. Temp and staging prefixes grow
  until somebody notices the bill. On a versioned bucket every overwrite parks its predecessor
  behind the current object where no ordinary listing shows it. Nothing in MinIO or S3 cleans that up
  on its own. Somebody has to install the policy.

  This flow installs it. It inventories one bucket with the MinIO plugin, reads the bucket's
  versioning state and its current lifecycle configuration with the AWS CLI, then classifies the
  bucket as compliant, missing a configuration entirely, missing the object expiry clause or missing
  the NoncurrentVersionExpiration clause. Read-only is the default. With `auto_remediate` on and
  `dry_run` off it pauses for an operator, writes a standard expiry plus noncurrent-version rule,
  then reads the configuration back and fails the run if the rule is not there.

  The inversion that makes this safe to automate is worth being clear about. The flow installs a
  POLICY. It never deletes an object itself. No object disappears while the run is going. What it
  writes is an instruction MinIO acts on later, on its own schedule: current objects under the target
  prefix are deleted once they pass `expire_after_days`. Noncurrent versions anywhere in the
  bucket are deleted once they pass `noncurrent_expire_after_days`. That is real deletion with a real
  consequence, just deferred, which is why the write sits behind an approval and why the run reports
  the size of the first sweep before anybody approves anything.

inputs:
  - id: dry_run
    type: BOOL
    defaults: true
    description: >-
      When true the flow reports the rule it would install and writes nothing.
      This is the default so a first run can never change a bucket.

  - id: auto_remediate
    type: BOOL
    defaults: false
    description: >-
      When true (and dry_run is false) the flow installs the lifecycle rule
      after an operator approves the pause. Both switches have to be set, so one
      of them alone is still safe mode.

  - id: bucket
    type: STRING
    description: The bucket to assess. One bucket per run, so an approval covers one
      blast radius.

  - id: target_prefix
    type: STRING
    defaults: "temp/"
    description: >-
      Key prefix the object expiry clause applies to. Keep it narrow. An empty
      value expires every current object in the bucket once it passes the
      window, which is almost never what anyone wants.

  - id: expire_after_days
    type: INT
    defaults: 30
    description: >-
      Age in days after which MinIO deletes a current object under the target
      prefix. The flow does not delete it, the policy does, on MinIO's own
      schedule.

  - id: noncurrent_expire_after_days
    type: INT
    defaults: 7
    description: >-
      Days a noncurrent version is kept after it stops being current. This is
      the clause that stops hidden version mass accumulating. It applies to the
      whole bucket rather than one prefix. Treat it as the rollback window.

  - id: protected_buckets
    type: STRING
    defaults: ""
    description: >-
      Comma-separated bucket names this flow must never write to, whatever the
      other inputs say. A bucket named here is still inventoried and still
      reported, it is just never remediated.

  - id: max_keys
    type: INT
    defaults: 1000
    description: >-
      Ceiling on keys per listing. The inventory is a sizing read, not a full
      crawl, so a very large bucket reports the first page and the report says
      so.

  - id: minio_endpoint
    type: STRING
    defaults: http://minio.internal:9000
    description: S3 API endpoint. The MinIO plugin and the AWS CLI both connect to it.

  - id: minio_access_key
    type: STRING
    defaults: "{{ secret('MINIO_ACCESS_KEY_ID') }}"
    description: >-
      Access key. Needs s3:ListBucket, s3:ListBucketVersions,
      s3:GetBucketVersioning, s3:GetLifecycleConfiguration and, for the write,
      s3:PutLifecycleConfiguration.

  - id: minio_secret_key
    type: STRING
    defaults: "{{ secret('MINIO_SECRET_KEY_ID') }}"
    description: Secret key for the access key above.

  - id: awscli_image
    type: STRING
    defaults: public.ecr.aws/aws-cli/aws-cli:latest
    description: >-
      Image carrying the AWS CLI plus python3. The MinIO plugin has no lifecycle
      task, so the configuration read and the configuration write go through
      this client.

  - id: docker_network
    type: STRING
    defaults: bridge
    description: >-
      Docker network the client container joins so it can reach the endpoint.
      Set it to your own network when the endpoint is a container name.

triggers:
  - id: weekly_lifecycle_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: >-
      Weekly read-only audit. Safe mode is forced here, so a scheduled run
      reports drift and never installs anything.
    cron: "0 6 * * 1"
    disabled: true
    inputs:
      bucket: my-versioned-bucket
      dry_run: true
      auto_remediate: false

tasks:
  - id: list_bucket_objects
    type: io.kestra.plugin.minio.List
    description: >-
      Current objects in the bucket, the way a normal client sees it. This count
      is the denominator for the version arithmetic below and it carries the
      lastModified of every key, which is what sizes the first expiry sweep.
    endpoint: "{{ inputs.minio_endpoint }}"
    accessKeyId: "{{ inputs.minio_access_key }}"
    secretKeyId: "{{ inputs.minio_secret_key }}"
    bucket: "{{ inputs.bucket }}"
    recursive: true
    includeVersions: false
    filter: FILES
    maxKeys: "{{ inputs.max_keys }}"

  - id: list_version_entries
    type: io.kestra.plugin.minio.List
    description: >-
      The same bucket with version listing on, so every stored version and every
      delete marker comes back as its own entry. The gap against the current
      listing is the noncurrent mass the NoncurrentVersionExpiration clause
      exists to bound.
    endpoint: "{{ inputs.minio_endpoint }}"
    accessKeyId: "{{ inputs.minio_access_key }}"
    secretKeyId: "{{ inputs.minio_secret_key }}"
    bucket: "{{ inputs.bucket }}"
    recursive: true
    includeVersions: true
    filter: FILES
    maxKeys: "{{ inputs.max_keys }}"

  - id: assess_lifecycle
    type: io.kestra.plugin.scripts.shell.Commands
    description: >-
      Read the bucket's versioning state and its current lifecycle
      configuration, join them to both listings, classify the bucket and write
      the exact rule document the remediation would PUT. An unreadable bucket
      fails this task instead of reporting a clean result.
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
      image: "{{ inputs.awscli_image }}"
      networkMode: "{{ inputs.docker_network }}"
      entryPoint: []
    env:
      AWS_ACCESS_KEY_ID: "{{ inputs.minio_access_key }}"
      AWS_SECRET_ACCESS_KEY: "{{ inputs.minio_secret_key }}"
      AWS_DEFAULT_REGION: us-east-1
      AWS_EC2_METADATA_DISABLED: "true"
      ENDPOINT: "{{ inputs.minio_endpoint }}"
      BUCKET: "{{ inputs.bucket }}"
      TARGET_PREFIX: "{{ inputs.target_prefix }}"
      EXPIRE_DAYS: "{{ inputs.expire_after_days }}"
      NONCURRENT_DAYS: "{{ inputs.noncurrent_expire_after_days }}"
      PROTECTED: "{{ inputs.protected_buckets }}"
      MAX_KEYS: "{{ inputs.max_keys }}"
    inputFiles:
      objects.json: "{{ outputs.list_bucket_objects.objects | toJson }}"
      versions.json: "{{ outputs.list_version_entries.objects | toJson }}"
    outputFiles:
      - lifecycle.json
    commands:
      - |
        set -eu
        aws --endpoint-url "$ENDPOINT" s3api get-bucket-versioning --bucket "$BUCKET" > versioning.json
        set +e
        aws --endpoint-url "$ENDPOINT" s3api get-bucket-lifecycle-configuration --bucket "$BUCKET" > current.json 2> current.err
        LC_RC=$?
        set -e
        export LC_RC
        cat > assess.py <<'PY'
        import datetime, json, os, sys

        bucket = os.environ["BUCKET"]
        prefix = os.environ.get("TARGET_PREFIX", "")
        expire_days = int(os.environ["EXPIRE_DAYS"])
        noncurrent_days = int(os.environ["NONCURRENT_DAYS"])
        max_keys = int(os.environ["MAX_KEYS"])
        protected = [b.strip() for b in os.environ.get("PROTECTED", "").split(",") if b.strip()]
        lc_rc = int(os.environ["LC_RC"])

        def load(path):
            with open(path) as fh:
                raw = fh.read().strip()
            return json.loads(raw) if raw else []

        objects = load("objects.json") or []
        versions = load("versions.json") or []
        versioning = (load("versioning.json") or {}).get("Status") or "Off"

        err = open("current.err").read()
        current = {}
        if lc_rc == 0:
            current = load("current.json") or {}
        elif "NoSuchLifecycleConfiguration" not in err:
            sys.stderr.write("lifecycle read failed: " + err.strip()[:400] + "\n")
            sys.exit(1)
        rules = current.get("Rules") or []

        def rule_prefix(rule):
            flt = rule.get("Filter") or {}
            if "Prefix" in flt:
                return flt.get("Prefix") or ""
            nested = flt.get("And") or {}
            if "Prefix" in nested:
                return nested.get("Prefix") or ""
            return rule.get("Prefix") or ""

        enabled = [r for r in rules if (r.get("Status") or "") == "Enabled"]
        covering = [r for r in enabled if prefix.startswith(rule_prefix(r))]
        has_expiry = any(int((r.get("Expiration") or {}).get("Days") or 0) > 0 for r in covering)
        has_noncurrent = any(
            int((r.get("NoncurrentVersionExpiration") or {}).get("NoncurrentDays") or 0) > 0
            for r in enabled
        )

        if bucket in protected:
            status = "PROTECTED"
        elif versioning != "Enabled":
            status = "VERSIONING_OFF"
        elif not rules:
            status = "NO_CONFIG"
        elif has_expiry and has_noncurrent:
            status = "COMPLIANT"
        elif has_noncurrent:
            status = "MISSING_EXPIRATION"
        else:
            status = "MISSING_NONCURRENT"
        needs_rule = "yes" if status in ("NO_CONFIG", "MISSING_EXPIRATION", "MISSING_NONCURRENT") else "no"

        now = datetime.datetime.now(datetime.timezone.utc)

        def age_days(stamp):
            text = (stamp or "").replace("Z", "+00:00")
            try:
                return (now - datetime.datetime.fromisoformat(text)).total_seconds() / 86400.0
            except ValueError:
                return -1.0

        in_prefix = [o for o in objects if (o.get("key") or "").startswith(prefix)]
        ages = [a for a in (age_days(o.get("lastModified")) for o in in_prefix) if a >= 0]
        total_objects = len(objects)
        version_entries = len(versions)
        counts = {
            "status": status,
            "needs_rule": needs_rule,
            "versioning": versioning,
            "total_objects": total_objects,
            "total_bytes": sum(int(o.get("size") or 0) for o in objects),
            "prefix_objects": len(in_prefix),
            "prefix_bytes": sum(int(o.get("size") or 0) for o in in_prefix),
            "version_entries": version_entries,
            "noncurrent_entries": max(version_entries - total_objects, 0),
            "past_window": len([a for a in ages if a >= expire_days]),
            "oldest_prefix_days": round(max(ages), 2) if ages else 0.0,
            "existing_rules": len(rules),
        }

        tops = sorted({(o.get("key") or "").split("/")[0] + "/"
                       for o in objects if "/" in (o.get("key") or "")})
        covered = [t for t in tops if t.startswith(prefix) or prefix == ""]
        uncovered = [t for t in tops if t not in covered]

        rule = {
            "ID": "kestra-expiry-guard",
            "Status": "Enabled",
            "Filter": {"Prefix": prefix},
            "Expiration": {"Days": expire_days},
            "NoncurrentVersionExpiration": {"NoncurrentDays": noncurrent_days},
        }
        kept = [r for r in rules if (r.get("ID") or "") != "kestra-expiry-guard"]
        proposed = {"Rules": kept + [rule]}
        with open("lifecycle.json", "w") as fh:
            json.dump(proposed, fh)

        print("bucket " + bucket + " versioning " + versioning + " verdict " + status)
        print("existing lifecycle rules " + str(counts["existing_rules"])
              + " covering the target prefix " + str(len(covering)))
        print("current objects " + str(counts["total_objects"])
              + " holding " + str(counts["total_bytes"]) + " bytes")
        print("version entries " + str(counts["version_entries"])
              + " of which noncurrent " + str(counts["noncurrent_entries"]))
        print("under prefix '" + prefix + "' objects " + str(counts["prefix_objects"])
              + " holding " + str(counts["prefix_bytes"]) + " bytes")
        print("oldest object under that prefix " + str(counts["oldest_prefix_days"]) + " days")
        print("already past the " + str(expire_days) + " day window, so the first sweep deletes "
              + str(counts["past_window"]) + " current objects")
        print("top level prefixes covered by the proposed rule " + (", ".join(covered) or "none"))
        print("top level prefixes left growing " + (", ".join(uncovered) or "none"))
        if total_objects >= max_keys or version_entries >= max_keys:
            print("a listing hit the max_keys ceiling of " + str(max_keys)
                  + " so these counts are a floor, not a total")
        print("the policy this run would install, applied by MinIO later and never by this flow:")
        print(json.dumps(proposed, indent=2))
        print("::" + json.dumps({"outputs": counts}) + "::")
        PY
        python3 assess.py

  - id: log_assessment
    type: io.kestra.plugin.core.log.Log
    description: One line that survives in the execution log whatever the gate decides next.
    message: >-
      bucket={{ inputs.bucket }} verdict={{ outputs.assess_lifecycle.vars.status
      }} versioning={{ outputs.assess_lifecycle.vars.versioning }} objects={{
      outputs.assess_lifecycle.vars.total_objects }} noncurrent={{
      outputs.assess_lifecycle.vars.noncurrent_entries }} first_sweep={{
      outputs.assess_lifecycle.vars.past_window }} needs_rule={{
      outputs.assess_lifecycle.vars.needs_rule }} dry_run={{ inputs.dry_run }}
      auto_remediate={{ inputs.auto_remediate }}

  - id: remediation_gate
    type: io.kestra.plugin.core.flow.If
    description: >-
      The write happens only when the bucket needs a rule, auto_remediate is on
      and dry_run is off. A protected bucket and a bucket whose versioning is
      not enabled both fail the first test, so neither can be written by this
      flow.
    condition: "{{ inputs.auto_remediate and inputs.dry_run == false and
      outputs.assess_lifecycle.vars.needs_rule == 'yes' }}"
    then:
      - id: approve_policy
        type: io.kestra.plugin.core.flow.Pause
        description: >-
          Hold for an operator. The execution log above already carries the rule
          document and the size of the first sweep, so the approval is made
          against real numbers.
        onResume:
          - id: approved
            type: BOOL
            defaults: false
            description: >-
              Set true to install the lifecycle rule. Anything else, including
              resuming without an answer, leaves the bucket untouched.

      - id: apply_gate
        type: io.kestra.plugin.core.flow.If
        description: Only an explicit true installs the rule.
        condition: "{{ (outputs.approve_policy.onResume.approved ?? false) }}"
        then:
          - id: apply_lifecycle
            type: io.kestra.plugin.scripts.shell.Commands
            description: >-
              PUT the rule document the assessment wrote. A lifecycle PUT
              replaces the whole configuration, so the document carries the
              bucket's existing rules alongside the new one. The exit status is
              recorded and is not the gate. A refused or ignored write can
              report success, so the read-back decides.
            taskRunner:
              type: io.kestra.plugin.scripts.runner.docker.Docker
              image: "{{ inputs.awscli_image }}"
              networkMode: "{{ inputs.docker_network }}"
              entryPoint: []
            env:
              AWS_ACCESS_KEY_ID: "{{ inputs.minio_access_key }}"
              AWS_SECRET_ACCESS_KEY: "{{ inputs.minio_secret_key }}"
              AWS_DEFAULT_REGION: us-east-1
              AWS_EC2_METADATA_DISABLED: "true"
              ENDPOINT: "{{ inputs.minio_endpoint }}"
              BUCKET: "{{ inputs.bucket }}"
            inputFiles:
              lifecycle.json: "{{ outputs.assess_lifecycle.outputFiles['lifecycle.json'] }}"
            commands:
              - |
                set -eu
                echo "installing this policy on $BUCKET"
                cat lifecycle.json
                set +e
                aws --endpoint-url "$ENDPOINT" s3api put-bucket-lifecycle-configuration \
                  --bucket "$BUCKET" --lifecycle-configuration file://lifecycle.json > put.out 2> put.err
                PUT_RC=$?
                set -e
                echo "put exit status $PUT_RC"
                head -c 400 put.err || true
                echo "::{\"outputs\":{\"put_rc\":\"$PUT_RC\"}}::"
                echo "put_rc=$PUT_RC recorded, the read-back is the gate"

          - id: reverify_lifecycle
            type: io.kestra.plugin.scripts.shell.Commands
            description: >-
              Read the configuration back and fail the run unless an enabled
              rule covering the target prefix carries both the object expiry and
              the NoncurrentVersionExpiration the inputs asked for. This is the
              only task allowed to call the remediation done.
            taskRunner:
              type: io.kestra.plugin.scripts.runner.docker.Docker
              image: "{{ inputs.awscli_image }}"
              networkMode: "{{ inputs.docker_network }}"
              entryPoint: []
            env:
              AWS_ACCESS_KEY_ID: "{{ inputs.minio_access_key }}"
              AWS_SECRET_ACCESS_KEY: "{{ inputs.minio_secret_key }}"
              AWS_DEFAULT_REGION: us-east-1
              AWS_EC2_METADATA_DISABLED: "true"
              ENDPOINT: "{{ inputs.minio_endpoint }}"
              BUCKET: "{{ inputs.bucket }}"
              TARGET_PREFIX: "{{ inputs.target_prefix }}"
              EXPIRE_DAYS: "{{ inputs.expire_after_days }}"
              NONCURRENT_DAYS: "{{ inputs.noncurrent_expire_after_days }}"
              PUT_RC: "{{ outputs.apply_lifecycle.vars.put_rc }}"
            commands:
              - |
                set -eu
                set +e
                aws --endpoint-url "$ENDPOINT" s3api get-bucket-lifecycle-configuration --bucket "$BUCKET" > after.json 2> after.err
                GET_RC=$?
                set -e
                export GET_RC
                cat > verify.py <<'PY'
                import json, os, sys

                prefix = os.environ.get("TARGET_PREFIX", "")
                expire_days = int(os.environ["EXPIRE_DAYS"])
                noncurrent_days = int(os.environ["NONCURRENT_DAYS"])
                put_rc = os.environ.get("PUT_RC", "unknown")
                get_rc = int(os.environ["GET_RC"])

                if get_rc != 0:
                    detail = open("after.err").read().strip()[:400]
                    print("read back failed after a put that exited " + put_rc)
                    print(detail)
                    sys.exit(1)

                with open("after.json") as fh:
                    after = json.loads(fh.read() or "{}")
                print("configuration now on the bucket:")
                print(json.dumps(after, indent=2))
                rules = after.get("Rules") or []

                def rule_prefix(rule):
                    flt = rule.get("Filter") or {}
                    if "Prefix" in flt:
                        return flt.get("Prefix") or ""
                    nested = flt.get("And") or {}
                    if "Prefix" in nested:
                        return nested.get("Prefix") or ""
                    return rule.get("Prefix") or ""

                enabled = [r for r in rules if (r.get("Status") or "") == "Enabled"]
                covering = [r for r in enabled if prefix.startswith(rule_prefix(r))]
                expiry_ok = [
                    r for r in covering
                    if int((r.get("Expiration") or {}).get("Days") or 0) == expire_days
                ]
                noncurrent_ok = [
                    r for r in enabled
                    if int((r.get("NoncurrentVersionExpiration") or {}).get("NoncurrentDays") or 0)
                    == noncurrent_days
                ]
                print("enabled rules " + str(len(enabled))
                      + " covering the prefix " + str(len(covering))
                      + " with the expiry asked for " + str(len(expiry_ok))
                      + " with the noncurrent expiry asked for " + str(len(noncurrent_ok)))
                if not expiry_ok or not noncurrent_ok:
                    print("the rule did not land, so this run failed")
                    print("the put exited " + put_rc
                          + ", which tells you nothing on its own, which is why this check exists")
                    sys.exit(1)
                print("rule landed: current objects under '" + prefix + "' now expire at "
                      + str(expire_days) + " days")
                print("noncurrent versions in this bucket now expire at "
                      + str(noncurrent_days) + " days after they stop being current")
                print("MinIO performs those deletions on its own schedule. This flow deleted nothing.")
                PY
                python3 verify.py
        else:
          - id: log_declined
            type: io.kestra.plugin.core.log.Log
            description: The operator declined or resumed without an answer, so nothing was
              written.
            level: WARN
            message: >-
              approval not granted for {{ inputs.bucket }}, no lifecycle rule
              installed, the bucket is still at verdict {{
              outputs.assess_lifecycle.vars.status }}
    else:
      - id: log_no_action
        type: io.kestra.plugin.core.log.Log
        description: >-
          Why this run wrote nothing. A compliant bucket, a protected bucket, a
          bucket whose versioning is not enabled and a safe-mode run all land
          here.
        message: >-
          no write for {{ inputs.bucket }}, verdict {{
          outputs.assess_lifecycle.vars.status }}, needs_rule {{
          outputs.assess_lifecycle.vars.needs_rule }}, dry_run {{ inputs.dry_run
          }}, auto_remediate {{ inputs.auto_remediate }}. The rule document is
          printed above for review.

errors:
  - id: alert_slack
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Post the failure to Slack. The webhook comes from a secret, never
      from the flow text.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "MinIO lifecycle guard failed on bucket {{ inputs.bucket }}. Flow {{ flow.namespace }}.{{ flow.id }} execution {{ execution.id }}. Either the bucket could not be read or the lifecycle rule did not land after the write. The bucket is unchanged unless the read-back says otherwise."
      }