id: minio-bucket-default-encryption-enforce-guard
namespace: company.team
description: |
A bucket with no default server side encryption stores every new object as plaintext on disk.
Nothing errors. Every client still reads and writes normally. The gap shows up in an audit. It
shows up again the week after somebody walks off with a drive. Turning the default on is one API call that nobody makes
until it is already a finding.
This flow finds the gap and closes it. The MinIO plugin inventories the bucket, the AWS CLI reads
`get-bucket-encryption`, then the run classifies the bucket as encrypted with SSE-S3, encrypted
with a KMS key, unencrypted or protected. Read-only is the default. With `auto_remediate` on and
`dry_run` off it pauses for an operator, applies an SSE-S3 default, writes one small probe object,
reads the probe back to confirm the default actually applies to a new write, deletes the probe,
then fails the run if the configuration is not on the bucket afterwards.
Be clear about what this buys, because the name "default encryption" oversells itself. It is a
DEFAULT. It applies to objects written after it lands. Every object already in the bucket stays
exactly as it was stored, in plaintext. This flow does not rewrite them. So the report counts
them and names them before anybody approves anything. Closing that half is a copy of each object
over itself, which rewrites real data, so it stays out of here on purpose.
variables:
canary_key: ".kestra-encryption-probe/{{ execution.id }}.txt"
inputs:
- id: dry_run
type: BOOL
defaults: true
description: >-
When true the flow prints the configuration it would apply then writes
nothing. This is the default so a first run cannot change a bucket.
- id: auto_remediate
type: BOOL
defaults: false
description: >-
When true (and dry_run is false) the flow applies the default after an
operator approves the pause. Both switches have to be set, so either one
alone is still safe mode.
- id: bucket
type: STRING
description: The bucket to assess. One bucket per run, so one approval covers
one blast radius.
- 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 in the inventory listing. The inventory sizes the
plaintext that default encryption will not fix, so on a very large bucket
the report says the count is a floor.
- 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:GetEncryptionConfiguration plus, for
the write, s3:PutEncryptionConfiguration, s3:PutObject and s3:DeleteObject
for the probe object.
- 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
encryption task, so the configuration read, the configuration write plus
the probe head request 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_encryption_audit
type: io.kestra.plugin.core.trigger.Schedule
description: >-
Weekly read-only audit. Safe mode is forced here, so a scheduled run
reports the gap without touching the bucket.
cron: "0 6 * * 1"
disabled: true
inputs:
bucket: my-bucket
dry_run: true
auto_remediate: false
tasks:
- id: inventory_objects
type: io.kestra.plugin.minio.List
description: >-
Every current object in the bucket. This listing is the number the whole
run turns on: these are the objects a default encryption rule will NOT
protect. The oldest key here becomes the plaintext baseline the
verification reads back.
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: assess_encryption
type: io.kestra.plugin.scripts.shell.Commands
description: >-
Read the bucket's encryption configuration, join it to the inventory,
classify the bucket plus write the exact document the remediation would
apply. The unset configuration answers with one specific error code, so
that code is the finding. Any other read failure fails this task instead
of reporting a clean bucket.
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 }}"
PROTECTED: "{{ inputs.protected_buckets }}"
MAX_KEYS: "{{ inputs.max_keys }}"
inputFiles:
objects.json: "{{ outputs.inventory_objects.objects | toJson }}"
outputFiles:
- encryption.json
- canary.txt
commands:
- |
set -eu
set +e
aws --endpoint-url "$ENDPOINT" s3api get-bucket-encryption --bucket "$BUCKET" > current.json 2> current.err
ENC_RC=$?
set -e
export ENC_RC
cat > assess.py <<'PY'
import json, os, sys
bucket = os.environ["BUCKET"]
max_keys = int(os.environ["MAX_KEYS"])
protected = [b.strip() for b in os.environ.get("PROTECTED", "").split(",") if b.strip()]
enc_rc = int(os.environ["ENC_RC"])
unset_codes = (
"ServerSideEncryptionConfigurationNotFoundError",
"NoSuchEncryptionConfiguration",
"NoSuchServerSideEncryptionConfiguration",
)
def load(path):
with open(path) as fh:
raw = fh.read().strip()
return json.loads(raw) if raw else []
objects = load("objects.json") or []
err = open("current.err").read()
current = {}
seen_code = ""
if enc_rc == 0:
current = load("current.json") or {}
else:
hit = [c for c in unset_codes if c in err]
if not hit:
sys.stderr.write("encryption read failed: " + err.strip()[:400] + "\n")
sys.exit(1)
seen_code = hit[0]
rules = ((current.get("ServerSideEncryptionConfiguration") or {}).get("Rules") or [])
def algorithm(rule):
return ((rule.get("ApplyServerSideEncryptionByDefault") or {})
.get("SSEAlgorithm") or "")
algorithms = [a for a in (algorithm(r) for r in rules) if a]
kms_keys = [((r.get("ApplyServerSideEncryptionByDefault") or {}).get("KMSMasterKeyID") or "")
for r in rules]
kms_keys = [k for k in kms_keys if k]
if bucket in protected:
status = "PROTECTED"
elif not rules:
status = "UNENCRYPTED"
elif "aws:kms" in algorithms:
status = "ENCRYPTED_KMS"
elif "AES256" in algorithms:
status = "ENCRYPTED_SSE_S3"
else:
status = "ENCRYPTED_OTHER"
needs_rule = "yes" if status == "UNENCRYPTED" else "no"
by_age = sorted(objects, key=lambda o: (o.get("lastModified") or ""))
baseline = (by_age[0].get("key") if by_age else "")
prefixes = sorted({(o.get("key") or "").split("/")[0] + "/"
for o in objects if "/" in (o.get("key") or "")})
counts = {
"status": status,
"needs_rule": needs_rule,
"algorithm": (algorithms[0] if algorithms else "none"),
"existing_rules": len(rules),
"unset_error_code": seen_code,
"total_objects": len(objects),
"total_bytes": sum(int(o.get("size") or 0) for o in objects),
"baseline_key": baseline,
"hit_ceiling": "yes" if len(objects) >= max_keys else "no",
}
rule = {"ApplyServerSideEncryptionByDefault": {"SSEAlgorithm": "AES256"}}
proposed = {"Rules": rules + [rule]} if not rules else {"Rules": rules}
with open("encryption.json", "w") as fh:
json.dump(proposed, fh)
with open("canary.txt", "w") as fh:
fh.write("kestra default encryption probe for " + bucket + "\n")
print("bucket " + bucket + " verdict " + status
+ " default algorithm " + counts["algorithm"])
if seen_code:
print("the bucket answered " + seen_code
+ ", which is how S3 says no default encryption is configured")
print("existing default rules " + str(counts["existing_rules"]))
if kms_keys:
print("existing KMS key reference " + kms_keys[0]
+ ", stronger than SSE-S3, so this flow leaves it alone")
print("objects already stored " + str(counts["total_objects"])
+ " holding " + str(counts["total_bytes"]) + " bytes")
if status == "UNENCRYPTED":
print("those " + str(counts["total_objects"])
+ " objects are on disk as plaintext and a default rule does NOT touch them")
print("prefixes carrying them: " + (", ".join(prefixes) or "the bucket root"))
print("oldest key, read back later as the plaintext baseline: "
+ (baseline or "the bucket is empty"))
if counts["hit_ceiling"] == "yes":
print("the listing hit the max_keys ceiling of " + str(max_keys)
+ " so that object count is a floor, not a total")
print("the configuration this run would apply:")
print(json.dumps(proposed, indent=2))
print("a put of this document REPLACES the bucket's whole encryption configuration, which is"
" why the flow writes only when there is no configuration to replace")
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_encryption.vars.status }} algorithm={{
outputs.assess_encryption.vars.algorithm }} existing_rules={{
outputs.assess_encryption.vars.existing_rules }} stored_objects={{
outputs.assess_encryption.vars.total_objects }} needs_rule={{
outputs.assess_encryption.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 has no encryption configuration,
auto_remediate is on plus dry_run is off. A protected bucket, an SSE-S3
bucket and a KMS bucket all fail the first test, so none of them can be
written by this flow.
condition: "{{ inputs.auto_remediate and inputs.dry_run == false and
outputs.assess_encryption.vars.needs_rule == 'yes' }}"
then:
- id: approve_encryption
type: io.kestra.plugin.core.flow.Pause
description: >-
Hold for an operator. The log above already carries the configuration
document, the object count that stays plaintext plus the key that will
be read back to show it, so the approval is made against real numbers.
onResume:
- id: approved
type: BOOL
defaults: false
description: >-
Set true to apply the SSE-S3 default. Anything else, including
resuming with no answer, leaves the bucket untouched.
- id: apply_gate
type: io.kestra.plugin.core.flow.If
description: Only an explicit true applies the configuration.
condition: "{{ (outputs.approve_encryption.onResume.approved ?? false) }}"
then:
- id: apply_encryption
type: io.kestra.plugin.scripts.shell.Commands
description: >-
Apply the SSE-S3 default. The exit status is recorded here but it
decides nothing. A refused bucket configuration write can report
success, so the read back is the gate.
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:
encryption.json: "{{ outputs.assess_encryption.outputFiles['encryption.json'] }}"
commands:
- |
set -eu
echo "applying this encryption configuration to $BUCKET"
cat encryption.json
echo ""
set +e
aws --endpoint-url "$ENDPOINT" s3api put-bucket-encryption \
--bucket "$BUCKET" \
--server-side-encryption-configuration file://encryption.json > put.out 2> put.err
PUT_RC=$?
set -e
echo "put exit status $PUT_RC"
head -c 400 put.err || true
echo ""
echo "::{\"outputs\":{\"put_rc\":\"$PUT_RC\"}}::"
echo "put_rc=$PUT_RC recorded, the read back is the gate"
- id: write_probe_object
type: io.kestra.plugin.minio.Upload
description: >-
Write one small object AFTER the configuration landed. A default
only shows up on new writes, so writing something is the only
honest way to test it. The verification deletes this object again.
endpoint: "{{ inputs.minio_endpoint }}"
accessKeyId: "{{ inputs.minio_access_key }}"
secretKeyId: "{{ inputs.minio_secret_key }}"
bucket: "{{ inputs.bucket }}"
key: "{{ render(vars.canary_key) }}"
from: "{{ outputs.assess_encryption.outputFiles['canary.txt'] }}"
contentType: text/plain
- id: reverify_encryption
type: io.kestra.plugin.scripts.shell.Commands
description: >-
Read the configuration back, head the probe object to prove a new
write now carries SSE-S3, head the oldest pre-existing object to
show it still does not, delete the probe then fail the run unless
the default is on the bucket. 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 }}"
CANARY_KEY: "{{ render(vars.canary_key) }}"
BASELINE_KEY: "{{ outputs.assess_encryption.vars.baseline_key }}"
PUT_RC: "{{ outputs.apply_encryption.vars.put_rc }}"
commands:
- |
set -eu
set +e
aws --endpoint-url "$ENDPOINT" s3api get-bucket-encryption --bucket "$BUCKET" > after.json 2> after.err
GET_RC=$?
aws --endpoint-url "$ENDPOINT" s3api head-object --bucket "$BUCKET" \
--key "$CANARY_KEY" > canary.json 2> canary.err
CANARY_RC=$?
if [ -n "$BASELINE_KEY" ]; then
aws --endpoint-url "$ENDPOINT" s3api head-object --bucket "$BUCKET" \
--key "$BASELINE_KEY" > baseline.json 2> baseline.err
BASELINE_RC=$?
else
BASELINE_RC=99
echo '{}' > baseline.json
fi
aws --endpoint-url "$ENDPOINT" s3api delete-object --bucket "$BUCKET" \
--key "$CANARY_KEY" > /dev/null 2> delete.err
DELETE_RC=$?
set -e
export GET_RC CANARY_RC BASELINE_RC DELETE_RC
cat > verify.py <<'PY'
import json, os, sys
put_rc = os.environ.get("PUT_RC", "unknown")
get_rc = int(os.environ["GET_RC"])
canary_rc = int(os.environ["CANARY_RC"])
baseline_rc = int(os.environ["BASELINE_RC"])
delete_rc = int(os.environ["DELETE_RC"])
canary_key = os.environ["CANARY_KEY"]
baseline_key = os.environ.get("BASELINE_KEY", "")
def load(path):
with open(path) as fh:
raw = fh.read().strip()
return json.loads(raw) if raw else {}
if delete_rc != 0:
print("WARNING the probe object " + canary_key
+ " could not be deleted, remove it by hand")
print(open("delete.err").read().strip()[:200])
else:
print("probe object " + canary_key + " deleted")
if get_rc != 0:
print("read back failed after a put that exited " + put_rc)
print(open("after.err").read().strip()[:400])
sys.exit(1)
after = load("after.json")
print("configuration now on the bucket:")
print(json.dumps(after, indent=2))
rules = ((after.get("ServerSideEncryptionConfiguration") or {}).get("Rules") or [])
applied = [((r.get("ApplyServerSideEncryptionByDefault") or {}).get("SSEAlgorithm")
or "") for r in rules]
print("default rules on the bucket " + str(len(rules))
+ " algorithms " + (", ".join(a for a in applied if a) or "none"))
canary_sse = ""
if canary_rc == 0:
canary_sse = load("canary.json").get("ServerSideEncryption") or ""
print("the object written after the change reports ServerSideEncryption "
+ (canary_sse or "nothing at all"))
else:
print("the probe object could not be read back: "
+ open("canary.err").read().strip()[:200])
if baseline_rc == 0:
old_sse = load("baseline.json").get("ServerSideEncryption") or ""
print("the pre-existing object " + baseline_key
+ " reports ServerSideEncryption " + (old_sse or "nothing at all"))
if not old_sse:
print("that is the limit of this fix, stated plainly: the default protects"
" new writes only. Objects already in the bucket stay as they were"
" stored. Copy each one over itself to re-encrypt it.")
elif baseline_key:
print("the pre-existing baseline object could not be read: "
+ open("baseline.err").read().strip()[:200])
else:
print("the bucket held no objects, so there is no plaintext baseline to show")
if "AES256" not in applied:
print("no SSE-S3 default landed on the bucket, 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)
if canary_rc != 0 or canary_sse != "AES256":
print("the configuration is there but a fresh write did not come back encrypted,"
" so the default is not being applied and this run failed")
sys.exit(1)
print("verified: the bucket carries an SSE-S3 default plus a new write came back"
" encrypted with AES256")
PY
python3 verify.py
else:
- id: log_declined
type: io.kestra.plugin.core.log.Log
description: The operator declined or resumed with no answer, so nothing was
written.
level: WARN
message: >-
approval not granted for {{ inputs.bucket }}, no encryption
default applied, the bucket is still at verdict {{
outputs.assess_encryption.vars.status }} with {{
outputs.assess_encryption.vars.total_objects }} objects stored as
plaintext
else:
- id: log_no_action
type: io.kestra.plugin.core.log.Log
description: >-
Why this run wrote nothing. An SSE-S3 bucket, a KMS bucket, a
protected bucket plus any safe-mode run all land here.
message: >-
no write for {{ inputs.bucket }}, verdict {{
outputs.assess_encryption.vars.status }}, algorithm {{
outputs.assess_encryption.vars.algorithm }}, needs_rule {{
outputs.assess_encryption.vars.needs_rule }}, dry_run {{
inputs.dry_run }}, auto_remediate {{ inputs.auto_remediate }}. The
configuration 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 default encryption guard failed on bucket {{ inputs.bucket }}. Flow {{ flow.namespace }}.{{ flow.id }} execution {{ execution.id }}. Either the bucket could not be read or no SSE-S3 default was on it after the write. The bucket is unchanged unless the read back says otherwise."
}
