MinIO Object Policy Compliance Gate

id: minio-object-policy-compliance-gate
namespace: company.team
description: |
  List a MinIO bucket, flag objects that are larger than a configured size
  limit or that sit outside an allowed prefix allowlist, alert Slack on any
  violation, and conditionally quarantine the offending objects by moving
  them under a quarantine prefix. auto_remediate and dry_run are two
  independent gates; nothing is ever deleted, only moved.

triggers:
  - id: periodic_policy_check
    type: io.kestra.plugin.core.trigger.Schedule
    description: Check every 6 hours. Shipped disabled; enable it once
      minio_endpoint/bucket point at a real instance and a manual run has been
      verified. Always fires in safe mode (auto_remediate false, dry_run true)
      regardless of this flow's own defaults.
    cron: "0 */6 * * *"
    disabled: true
    inputs:
      auto_remediate: "false"
      dry_run: "true"

inputs:
  - id: minio_endpoint
    type: STRING
    displayName: "MinIO Endpoint"
    defaults: "http://localhost:9000"
    description: URL to the MinIO endpoint, passed as the endpoint property on every
      io.kestra.plugin.minio task below.

  - id: bucket
    type: STRING
    displayName: "Bucket"
    defaults: "data-lake"
    description: Bucket audited for size and allowed-prefix policy violations.

  - id: audit_prefix
    type: STRING
    displayName: "Audit Prefix"
    defaults: ""
    description: Restrict the audit to object keys under this prefix. Leave empty to
      audit the whole bucket.

  - id: allowed_prefixes
    type: STRING
    displayName: "Allowed Prefixes (comma-separated)"
    defaults: "approved/,incoming/"
    description: Object keys that do not start with any of these prefixes are
      flagged as unauthorized-location violations.

  - id: max_object_size_mb
    type: INT
    displayName: "Max Allowed Object Size (MB)"
    defaults: 5120
    description: Objects larger than this size in megabytes (MB) are flagged as
      oversized-object violations, for example an unexpected bulk export landing
      somewhere it should not.

  - id: quarantine_prefix
    type: STRING
    displayName: "Quarantine Prefix"
    defaults: "quarantine/"
    description: Destination prefix violating objects are moved under when
      remediation runs. Objects are moved, never deleted outright.

  - id: auto_remediate
    type: SELECT
    displayName: "Auto Remediate"
    defaults: "false"
    values:
      - "false"
      - "true"
    description: If "true" and violations are found, move the violating objects into
      quarantine_prefix (still gated by dry_run below). If "false", violations
      are only reported.

  - id: dry_run
    type: SELECT
    displayName: "Dry Run"
    defaults: "true"
    values:
      - "true"
      - "false"
    description: If "true", remediation is only logged, no object is moved. Must be
      explicitly "false", together with auto_remediate "true", for objects to
      actually be quarantined.

tasks:
  - id: list_bucket_objects
    type: io.kestra.plugin.minio.List
    description: Lists every object under audit_prefix. Each returned object carries
      key, size, lastModified, etag, and owner - verified from the plugin's own
      output schema, there is no encryption or lifecycle-policy field available
      from this or any other task in this plugin.
    endpoint: "{{ inputs.minio_endpoint }}"
    accessKeyId: "{{ secret('MINIO_ACCESS_KEY_ID') }}"
    secretKeyId: "{{ secret('MINIO_SECRET_KEY_ID') }}"
    bucket: "{{ inputs.bucket }}"
    prefix: "{{ inputs.audit_prefix }}"
    recursive: true
    filter: FILES

  - id: evaluate_object_policy
    type: io.kestra.plugin.scripts.python.Script
    description: Reads the listing from list_bucket_objects and classifies each
      object as oversized, outside the allowed prefixes, both, or compliant.
      There is no structured policy-evaluation task in this plugin, so the
      comparison runs here against the key/size fields the List task actually
      returns.
    inputFiles:
      objects.json: "{{ outputs.list_bucket_objects | json }}"
    script: |
      import json
      from kestra import Kestra

      with open("objects.json") as f:
          listed = json.load(f)

      allowed_prefixes = [p.strip() for p in "{{ inputs.allowed_prefixes }}".split(",") if p.strip()]
      max_size_bytes = int("{{ inputs.max_object_size_mb }}") * 1024 * 1024

      oversized_keys = []
      unauthorized_location_keys = []

      for obj in listed.get("objects", []):
          key = obj.get("key")
          size = obj.get("size") or 0
          if size > max_size_bytes:
              oversized_keys.append(key)
          if allowed_prefixes and not any(key.startswith(p) for p in allowed_prefixes):
              unauthorized_location_keys.append(key)

      violation_keys = sorted(set(oversized_keys) | set(unauthorized_location_keys))

      Kestra.outputs({
          "violation_count": len(violation_keys),
          "violation_keys": violation_keys,
          "oversized_keys": oversized_keys,
          "unauthorized_location_keys": unauthorized_location_keys,
      })

  - id: evaluate_policy_gate
    type: io.kestra.plugin.core.flow.If
    description: Breach when at least one object violates the size or allowed-prefix policy.
    condition: "{{ outputs.evaluate_object_policy.vars.violation_count > 0 }}"
    then:
      - id: route_remediation
        type: io.kestra.plugin.core.flow.Switch
        description: auto_remediate only decides whether remediation is considered at
          all; dry_run, checked separately below, decides whether objects are
          actually moved.
        value: "{{ inputs.auto_remediate }}"
        cases:
          "true":
            - id: check_dry_run
              type: io.kestra.plugin.core.flow.If
              condition: "{{ inputs.dry_run == \"false\" }}"
              then:
                - id: quarantine_violating_objects
                  type: io.kestra.plugin.core.flow.Loop
                  description: Move each violating object into quarantine_prefix one at a time, so
                    one failed copy never blocks the rest of the batch.
                  values: "{{ outputs.evaluate_object_policy.vars.violation_keys }}"
                  concurrencyLimit: 4
                  tasks:
                    - id: quarantine_object
                      type: io.kestra.plugin.minio.Copy
                      description: delete true removes the source key once the copy succeeds, turning
                        this into a move into quarantine_prefix rather than a
                        duplicate.
                      endpoint: "{{ inputs.minio_endpoint }}"
                      accessKeyId: "{{ secret('MINIO_ACCESS_KEY_ID') }}"
                      secretKeyId: "{{ secret('MINIO_SECRET_KEY_ID') }}"
                      from:
                        bucket: "{{ inputs.bucket }}"
                        key: "{{ item.value }}"
                      to:
                        bucket: "{{ inputs.bucket }}"
                        key: "{{ inputs.quarantine_prefix }}{{ item.value }}"
                      delete: true
                - id: log_remediation_applied
                  type: io.kestra.plugin.core.log.Log
                  message: "Quarantined {{ outputs.evaluate_object_policy.vars.violation_count }}
                    object(s) from {{ inputs.bucket }} under {{
                    inputs.quarantine_prefix }}."
              else:
                - id: log_dry_run_remediation
                  type: io.kestra.plugin.core.log.Log
                  message: "DRY RUN: would quarantine {{
                    outputs.evaluate_object_policy.vars.violation_count }}
                    object(s) from {{ inputs.bucket }} under {{
                    inputs.quarantine_prefix }} (dry_run is true). No change
                    made."
          "false":
            - id: log_remediation_disabled
              type: io.kestra.plugin.core.log.Log
              message: "auto_remediate is false: {{
                outputs.evaluate_object_policy.vars.violation_count }} object(s)
                in {{ inputs.bucket }} violate policy, but no quarantine was
                attempted."
      - id: alert_policy_violation
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Always alert on any violation, independent of whether it was
          remediated.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "MinIO object policy gate: {{ outputs.evaluate_object_policy.vars.violation_count }} object(s) in bucket {{ inputs.bucket }} violate policy ({{ outputs.evaluate_object_policy.vars.oversized_keys | length }} oversized, {{ outputs.evaluate_object_policy.vars.unauthorized_location_keys | length }} outside allowed prefixes). auto_remediate: {{ inputs.auto_remediate }}. dry_run: {{ inputs.dry_run }}. Remediated: {{ inputs.auto_remediate == \"true\" and inputs.dry_run == \"false\" }}. Execution {{ execution.id }}."
          }
    else:
      - id: log_policy_healthy
        type: io.kestra.plugin.core.log.Log
        message: "{{ inputs.bucket }} is compliant: 0 object(s) violate the size or
          allowed-prefix policy."

  - id: log_audit_complete
    type: io.kestra.plugin.core.log.Log
    description: Final status line for every run, independent of which branch fired,
      so the execution history always shows the headline numbers.
    message: "Policy audit complete for {{ inputs.bucket }}. violation_count={{
      outputs.evaluate_object_policy.vars.violation_count }}. oversized={{
      outputs.evaluate_object_policy.vars.oversized_keys | length }}.
      unauthorized_location={{
      outputs.evaluate_object_policy.vars.unauthorized_location_keys | length
      }}. auto_remediate={{ inputs.auto_remediate }}. dry_run={{ inputs.dry_run
      }}."

errors:
  - id: alert_execution_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert when the flow itself fails outside the handled branches, for
      example an unreachable MinIO endpoint.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "MinIO object policy gate FAILED in execution {{ execution.id }} for bucket {{ inputs.bucket }}. The policy audit did not complete; investigate before trusting the last known compliance status."
      }