Build an S3 public-access guardrail and iterate it on your laptop

You are the security engineer at a fast-moving startup. The engineering team has a few dozen AWS accounts and ships dozens of S3 buckets a week. Every few months, somebody accidentally opens one up: a public ACL on a bucket that should not be readable to the world, a bucket policy that grants Principal: "*", a deletion of the default PublicAccessBlock. You do not want to be the human who finds out about it from a bug bounty report.

The boring AWS pattern is: CloudTrail records every S3 API call, an EventBridge rule matches the seven calls that can widen public access, and a Lambda inspects the bucket and locks down its PublicAccessBlock the moment something looks wrong. The whole thing runs in the background; nobody opens the AWS console. The first time the engineering team learns the guardrail exists is when a colleague's bucket flips private a few seconds after they tried to make it public.

Iterating that on real AWS is awkward: every change to the EventBridge pattern or the Lambda is a Terraform apply, and verifying the pipeline end to end means making a real public bucket in a real account, then either waiting for the auto-fix or having to remember to clean up if you got the rule wrong. This tutorial builds the whole pipeline on LocalEmu in about 18 seconds, lets you trigger the auto-remediation from the AWS CLI without making anything actually public on the internet, and tears down cleanly in 10. 11 Terraform resources, seven integration tests, no AWS account.

What you will have working at the end

From the command line, in order, with the real captured output later on this page:

A. Create a bucket with no PublicAccessBlock and apply a public-read ACL. Wait two seconds. The bucket's PAB is now fully on, the bucket is tagged le-guardrail-source=automated, and the DynamoDB incidents table has a row for the event. You did not invoke the Lambda; EventBridge did it for you.
B. Replay the same CloudTrail event. The handler returns already_processed because the eventID was claimed atomically by a ConditionExpression="attribute_not_exists(event_id)" put. Idempotency without a separate cache.
C. Seven pytest tests pass in under five seconds, covering: happy path, private bucket no-op, dedup, missing bucket, missing bucketName, EventBridge pattern shape, and the full real CloudTrail to EventBridge to Lambda end-to-end flow.

Architecture

user / app  ──S3 API call──▶  S3
                                │ logs every API call
                                ▼
                             CloudTrail
                                │ emits AWS API Call via CloudTrail event
                                ▼
                             EventBridge: rule ── matches 7 specific S3 eventNames
                                │
                                ▼
                             Lambda: remediator
                                │
                ┌───────────────┼───────────────┐
                ▼               ▼               ▼
       DynamoDB: incidents   S3 PutPAB + tag   SNS: alerts
       (dedup by eventID)    (the fix)        (notify humans)

1. The remediator Lambda

The handler does five things in order: dedup, race-check, decide whether the bucket is actually public, remediate, alert. The order matters: dedup runs before the bucket inspection so that a replay of the same CloudTrail event is a clean no-op even if the bucket state changed in the meantime (because, say, we already remediated it). The DynamoDB conditional write is the only mechanism that gives us idempotency; nothing else holds.

# src/handler.py: the guardrail Lambda.
# Triggered by an EventBridge rule matching S3 API calls that can
# widen public access. Inspects the bucket, remediates, dedups, alerts.

INCIDENTS_TABLE  = os.environ["INCIDENTS_TABLE"]
ALERTS_TOPIC_ARN = os.environ["ALERTS_TOPIC_ARN"]

_s3        = boto3.client("s3")
_sns       = boto3.client("sns")
_incidents = boto3.resource("dynamodb").Table(INCIDENTS_TABLE)


def handle(event, _ctx):
    detail     = event.get("detail") or {}
    event_id   = detail.get("eventID") or ""
    event_name = detail.get("eventName") or ""
    bucket     = (detail.get("requestParameters") or {}).get("bucketName")
    if not bucket:
        return {"remediated": False, "reason": "no_bucket"}

    # Dedup: claim the CloudTrail eventID BEFORE any inspection. A replay
    # of the same event is a no-op even if the bucket state has changed.
    if not _claim_event(event_id or f"{event_name}:{bucket}:{int(time.time())}",
                         bucket=bucket, event_name=event_name):
        return {"remediated": False, "reason": "already_processed", "bucket": bucket}

    # Race: the bucket might have been deleted between the CloudTrail event and us.
    try:
        _s3.head_bucket(Bucket=bucket)
    except ClientError as e:
        if e.response["Error"]["Code"] in ("404", "NoSuchBucket", "NotFound"):
            return {"remediated": False, "reason": "bucket_missing", "bucket": bucket}
        raise

    is_public, why = _bucket_is_public(bucket)
    if not is_public:
        return {"remediated": False, "reason": why, "bucket": bucket}

    _remediate(bucket)
    _alert(bucket=bucket, reason=why, event_id=event_id, event_name=event_name)
    return {"remediated": True, "bucket": bucket, "reason": why}

The interesting bit is how the handler decides "is this bucket public?". The rule is the same as on real AWS: a fully-on PublicAccessBlock makes the bucket private regardless of any public ACL or policy, so we check PAB first. Only if PAB is absent or partial do we look at ACL grants and the bucket policy's IsPublic status:

# src/handler.py: how we decide a bucket is public.
# PublicAccessBlock wins. If it is fully on, the bucket is private no
# matter what the ACL or policy says: AWS enforces that.

def _bucket_is_public(bucket: str) -> tuple[bool, str]:
    # 1. Fully-on PAB wins. Bucket is definitively private.
    try:
        cfg = _s3.get_public_access_block(Bucket=bucket)["PublicAccessBlockConfiguration"]
        if all([cfg["BlockPublicAcls"], cfg["IgnorePublicAcls"],
                cfg["BlockPublicPolicy"], cfg["RestrictPublicBuckets"]]):
            return False, "pab_fully_on"
    except ClientError as e:
        if e.response["Error"]["Code"] != "NoSuchPublicAccessBlockConfiguration":
            raise
        # No PAB at all is suspicious; fall through to the ACL / policy checks.

    # 2. Any ACL grant to AllUsers or AuthenticatedUsers means public.
    for g in _s3.get_bucket_acl(Bucket=bucket).get("Grants", []):
        uri = (g.get("Grantee") or {}).get("URI", "")
        if uri.endswith("/AllUsers") or uri.endswith("/AuthenticatedUsers"):
            return True, f"acl_grants_{g['Grantee'].get('Type', 'Group')}"

    # 3. Bucket-policy status (IsPublic from AWS's policy analyzer).
    try:
        if _s3.get_bucket_policy_status(Bucket=bucket).get("PolicyStatus", {}).get("IsPublic"):
            return True, "bucket_policy_is_public"
    except ClientError:
        pass
    return False, "ok"

2. The EventBridge rule

The rule matches AWS API Call events from CloudTrail where the source is S3 and the event name is one of the seven calls that can widen public access. The aws_lambda_permission block grants EventBridge the right to invoke the function, the source_arn constraint scopes that grant down to this one rule:

# terraform/main.tf: the EventBridge rule.
# Matches the seven S3 API calls that can widen public access.

resource "aws_cloudwatch_event_rule" "on_s3_policy_change" {
  name = "${var.prefix}-on-s3-policy-change"
  event_pattern = jsonencode({
    source        = ["aws.s3"]
    "detail-type" = ["AWS API Call via CloudTrail"]
    detail = {
      eventSource = ["s3.amazonaws.com"]
      eventName   = [
        "CreateBucket",
        "PutBucketAcl",
        "PutBucketPolicy",
        "PutBucketPublicAccessBlock",
        "DeletePublicAccessBlock",
        "DeleteBucketPublicAccessBlock",
        "PutBucketWebsite",
      ]
    }
  })
}

resource "aws_cloudwatch_event_target" "to_lambda" {
  rule = aws_cloudwatch_event_rule.on_s3_policy_change.name
  arn  = aws_lambda_function.remediator.arn
}

resource "aws_lambda_permission" "allow_eventbridge" {
  action        = "lambda:InvokeFunction"
  function_name = aws_lambda_function.remediator.function_name
  principal     = "events.amazonaws.com"
  source_arn    = aws_cloudwatch_event_rule.on_s3_policy_change.arn
}

The other three resources in the project are the CloudTrail trail itself (so the events are emitted at all), the IAM role the Lambda runs as, and the SNS topic that _alert() publishes to. The role grants the Lambda exactly the S3 actions it needs to inspect and remediate (head_bucket, get_bucket_acl, get_bucket_policy_status, get_public_access_block, put_public_access_block, put_bucket_tagging), plus DynamoDB put_item on the incidents table and sns:Publish on the topic.

3. Run the scenario on your laptop

Clone the project, start LocalEmu in another terminal, then deploy:

$ git clone https://github.com/localemu/localemu-examples
$ cd localemu-examples/06-s3-guardrail
$ localemu start # in a separate terminal
$ ./scripts/deploy.sh local

Eleven resources apply in about 18 seconds. Most of the wall time is the Lambda container cold-start:

$ ./scripts/deploy.sh local
aws_cloudtrail.trail:                            Creation complete after 1s
aws_dynamodb_table.incidents:                    Creation complete after 0s
aws_sns_topic.alerts:                            Creation complete after 0s
aws_iam_role.lambda_role:                        Creation complete after 0s
aws_iam_role_policy.lambda_policy:               Creation complete after 0s
aws_lambda_function.remediator:                  Creation complete after 5s
aws_cloudwatch_event_rule.on_s3_policy_change:   Creation complete after 0s
aws_cloudwatch_event_target.to_lambda:           Creation complete after 0s
aws_lambda_permission.allow_eventbridge:         Creation complete after 0s

Apply complete! Resources: 11 added, 0 changed, 0 destroyed.

→ deployed to local. outputs:
alerts_topic_arn  = "arn:aws:sns:us-east-1:000000000000:le-guard-alerts"
function_name     = "le-guard-remediator"
incidents_table   = "le-guard-incidents"
rule_name         = "le-guard-on-s3-policy-change"
trail_name        = "le-guard-trail"

real    0m18.278s

Now the interesting part: drive the full scenario without ever invoking the Lambda directly. The AWS CLI calls go through LocalEmu's S3, CloudTrail records them, EventBridge matches the rule, the Lambda fires, and the bucket gets locked down. From your terminal you only see the cause (PutBucketAcl) and the effect (PublicAccessBlock fully on):

$ # 1. Confirm the EventBridge rule is wired to the right CloudTrail events.
$ aws --endpoint-url http://localhost:4566 events list-rules \
       --query 'Rules[?Name==`le-guard-on-s3-policy-change`].[Name,State]' --output table
+--------------------------------+-----------+
|  le-guard-on-s3-policy-change  |  ENABLED  |
+--------------------------------+-----------+


$ # 2. Create a bucket and disable its default PublicAccessBlock so a public
$ #    ACL can actually take effect. The guardrail does not run yet.
$ DEMO=le-guard-demo-$(date +%s)
$ aws --endpoint-url http://localhost:4566 s3 mb s3://$DEMO
$ aws --endpoint-url http://localhost:4566 s3api put-bucket-ownership-controls \
       --bucket $DEMO \
       --ownership-controls 'Rules=[{ObjectOwnership=BucketOwnerPreferred}]'
$ aws --endpoint-url http://localhost:4566 s3api put-public-access-block \
       --bucket $DEMO \
       --public-access-block-configuration 'BlockPublicAcls=false,IgnorePublicAcls=false,BlockPublicPolicy=false,RestrictPublicBuckets=false'
$ aws --endpoint-url http://localhost:4566 s3api get-public-access-block \
       --bucket $DEMO --query 'PublicAccessBlockConfiguration' --output json
{
  "BlockPublicAcls":       false,
  "IgnorePublicAcls":      false,
  "BlockPublicPolicy":     false,
  "RestrictPublicBuckets": false
}


$ # 3. Make the bucket public. CloudTrail records this; EventBridge matches
$ #    the event; the guardrail Lambda fires automatically. NO MANUAL INVOKE.
$ aws --endpoint-url http://localhost:4566 s3api put-bucket-acl \
       --bucket $DEMO --acl public-read


$ # 4. Wait a couple seconds, then look at the bucket's PublicAccessBlock again.
$ #    All four flags should now be true. The guardrail did this on its own.
$ sleep 2
$ aws --endpoint-url http://localhost:4566 s3api get-public-access-block \
       --bucket $DEMO --query 'PublicAccessBlockConfiguration' --output json
{
  "BlockPublicAcls":       true,
  "IgnorePublicAcls":      true,
  "BlockPublicPolicy":     true,
  "RestrictPublicBuckets": true
}


$ # 5. The bucket is tagged so a human can see which fix was automated.
$ aws --endpoint-url http://localhost:4566 s3api get-bucket-tagging \
       --bucket $DEMO --output json
{
  "TagSet": [
    { "Key": "le-guardrail-source", "Value": "automated" }
  ]
}


$ # 6. The incidents table records one row per CloudTrail event that fired.
$ aws --endpoint-url http://localhost:4566 dynamodb scan \
       --table-name le-guard-incidents --output table \
       --query 'Items[*].[event_name.S,event_id.S]' \
       --filter-expression '#b = :v' \
       --expression-attribute-names '{"#b":"bucket"}' \
       --expression-attribute-values '{":v":{"S":"'$DEMO'"}}'
+----------------+--------------------------------------+
|  CreateBucket  | dc2f50ff-a6d7-4c93-b356-17a839289612 |
|  PutBucketAcl  | bfb7a0a1-b22a-4c56-ae72-7ccc892136f2 |
+----------------+--------------------------------------+

# Two rows because the rule matched two CloudTrail events: CreateBucket
# (no remediation needed; the new bucket already had PAB-on by default)
# and PutBucketAcl (the one that did need it).

The same seven scenarios live in tests/test_guardrail.py, including the same full CloudTrail-to-EventBridge end-to-end flow as the last test. The suite runs in under five seconds:

$ ./scripts/test.sh local

============================= test session starts ==============================
platform darwin -- Python 3.13.12, pytest-9.0.3
collected 7 items

tests/test_guardrail.py::test_public_bucket_is_remediated                              PASSED
tests/test_guardrail.py::test_private_bucket_is_no_op                                  PASSED
tests/test_guardrail.py::test_duplicate_event_is_deduped                               PASSED
tests/test_guardrail.py::test_missing_bucket_is_ignored                                PASSED
tests/test_guardrail.py::test_event_without_bucket_name_is_handled                     PASSED
tests/test_guardrail.py::test_eventbridge_rule_matches_the_right_apis                  PASSED
tests/test_guardrail.py::test_real_cloudtrail_eventbridge_flow_remediates_public_bucket PASSED

============================== 7 passed in 4.36s ==============================

Tear it back down:

$ ./scripts/teardown.sh local
Destroy complete! Resources: 11 destroyed.

→ verifying teardown for prefix 'le-guard' on local
  clean: nothing left behind

real    0m10.006s

Deploy 18 seconds, tests 4 seconds, teardown 10 seconds. About thirty seconds round trip per change to the handler, the rule pattern, or the IAM policy.

4. The same code on real AWS

Same scripts, aws instead of local:

$ ./scripts/deploy.sh aws
$ ./scripts/test.sh aws
$ ./scripts/teardown.sh aws

Four places where the same Terraform behaves differently on real AWS, all worth knowing before you point this at a production account:

• CloudTrail-to-EventBridge delivery is not instant on real AWS. Expect 5 to 15 minutes typically, occasionally longer. LocalEmu emits the events synchronously, which is what makes the iteration loop fast but is also why the "wait 2 seconds" line in the demo above is not realistic for production. In production you treat this as eventual remediation, not real-time.
• PublicAccessBlock has been on-by-default for new buckets since April 2023. The explicit PAB-off in the demo is what makes the public ACL actually take effect; without it, the ACL is stored but the bucket stays effectively private. The handler does the right thing either way (its PAB-first check correctly returns "private" for a PAB-on bucket regardless of ACL).
• CloudTrail data events vs management events. The seven API calls in the rule are management events (CreateBucket, PutBucketAcl, and so on), which CloudTrail emits to EventBridge in any region with a trail enabled. Object-level data events (GetObject, PutObject) are a separate, paid stream and are not what this guardrail watches.
• SNS subscription policies. Real AWS requires an SNS topic policy granting events.amazonaws.com the right to publish on its behalf, and any subscriber protocol (HTTPS, SQS, email) needs its own confirm-subscribe step. The template ships the alerts topic without subscribers for the demo; wire up a Slack / PagerDuty / email subscriber before deploying to production.

Broader comparison in LocalEmu vs Real AWS and Known Limitations.

Get the full project

git clone https://github.com/localemu/localemu-examples : the S3 guardrail tutorial lives in 06-s3-guardrail/ with the Terraform, the handler, the seven integration tests, and the deploy / test / teardown scripts that produced every terminal output on this page.

Build an S3 public-access guardrail and iterate it on your laptop

What you will have working at the end

1. The remediator Lambda

2. The EventBridge rule

3. Run the scenario on your laptop

4. The same code on real AWS

Where to go next

Event-driven fan-out

Image pipeline

IAM least-privilege mutation tests

LocalEmu vs Real AWS