Storage hook on AWS S3 with SQS

Read this if your generated images, audio or video land in an S3 bucket and you want every object marked in place without changing the code that writes it. You will run laidline ingest against an SQS queue fed by S3 notifications, with signing in your own KMS key. Skip if you mark inside your own code (Quickstart) or run on Lambda (AWS Lambda), Cloud Run (Google Cloud Run) or Cloudflare R2 (the last section). Time: about 30 minutes, most of it creating AWS resources. Needs: Python 3.12, pip install "laidline[image,ingest]", an AWS account where you may create S3 buckets, SQS queues and a KMS key, and a trial or registered issuer number from laidline init.

In short

• laidline ingest long-polls an SQS queue that receives S3 ObjectCreated events and marks each object where it lies.
• Media never leaves your account: the worker runs where you run it, with your IAM role and your KMS signing key.
• An already-marked object is skipped, so the write-back echo and re-uploads never double-mark.
• Permanent failures go to a dead-letter queue and transient ones are redelivered up to ingest.max_attempts.
• The same worker reads any S3-compatible store; Cloudflare R2 with Queues is the last section.

What it does

object created under generated/  -->  S3 event notification  -->  SQS queue
        laidline ingest (your container) long-polls the queue
        HEAD  (skip if laidline-asset-id metadata is set)
        GET   -> PixelSeal watermark -> re-encode -> C2PA manifest signed with your KMS key
                 (an upstream manifest, e.g. from Replicate or BFL, is kept as a parentOf ingredient)
        PUT back in place (or under marked/) with laidline-asset-id / laidline-manifest-id metadata,
            content type preserved (audio aliases are canonicalised: audio/x-wav is written back as audio/wav),
            cache-control preserved
        evidence entry  route: ingest:s3_sqs, upstream from the key prefix
        delete the message

Media never leaves your account: the worker runs where you run it, with your IAM role.

On hosts without a persistent disk (ECS without a volume, Lambda, Cloud Run) make evidence.sinks[0] an s3 (Object Lock) or postgres sink; a file sink is for a host with a volume (Evidence).

What you need

• Python 3.12 on the host that runs the worker, and the extras for the media you mark:

  pip install "laidline[image,ingest]"
  

  Add audio and video only if you mark those objects. ingest brings boto3 and prometheus-client.

• An S3 bucket, an SQS queue with a dead-letter queue, an Object Lock bucket for the s3 evidence sink, and an asymmetric KMS key (ECC_NIST_P256, SIGN_VERIFY), all in one region (step 1).

• An IAM role for the worker (the policy under IAM).

• A configuration from laidline init (Quickstart): its watermark and payload_keys blocks and your issuer number go into the worker's configuration.

Setup

1. Create the resources

The commands below use eu-central-1; use your region throughout. Replace <bucket>, <account> and the other placeholders.

Create the media bucket and block public access:

aws s3api create-bucket --bucket <bucket> --region eu-central-1 \
  --create-bucket-configuration LocationConstraint=eu-central-1
aws s3api put-public-access-block --bucket <bucket> --public-access-block-configuration \
  BlockPublicAcls=true,IgnorePublicAcls=true,BlockPublicPolicy=true,RestrictPublicBuckets=true

Create the dead-letter queue and the queue, and note both URLs:

aws sqs create-queue --queue-name <name>-dlq --query QueueUrl --output text
aws sqs create-queue --queue-name <name> --query QueueUrl --output text

Expected output, one URL per command:

https://sqs.eu-central-1.amazonaws.com/<account>/<name>-dlq
https://sqs.eu-central-1.amazonaws.com/<account>/<name>

Give the queue its settings and the policy that lets S3 send for this bucket only. Save this as queue-attributes.json, with the two queue ARNs (aws sqs get-queue-attributes --queue-url <url> --attribute-names QueueArn) filled in:

{
  "Policy": "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Principal\":{\"Service\":\"s3.amazonaws.com\"},\"Action\":\"SQS:SendMessage\",\"Resource\":\"<queue ARN>\",\"Condition\":{\"ArnLike\":{\"aws:SourceArn\":\"arn:aws:s3:*:*:<bucket>\"},\"StringEquals\":{\"aws:SourceAccount\":\"<account>\"}}}]}",
  "VisibilityTimeout": "180",
  "ReceiveMessageWaitTimeSeconds": "20",
  "RedrivePolicy": "{\"deadLetterTargetArn\":\"<DLQ ARN>\",\"maxReceiveCount\":\"5\"}"
}

aws sqs set-queue-attributes --queue-url <queue URL> --attributes file://queue-attributes.json

Point the bucket's s3:ObjectCreated:* events at the queue, filtered on generated/. This call replaces the bucket's whole notification configuration, so merge any entries you already have into the file first. Save as notification.json:

{
  "QueueConfigurations": [
    {
      "Id": "laidline-object-created",
      "QueueArn": "<queue ARN>",
      "Events": ["s3:ObjectCreated:*"],
      "Filter": {"Key": {"FilterRules": [{"Name": "prefix", "Value": "generated/"}]}}
    }
  ]
}

aws s3api put-bucket-notification-configuration --bucket <bucket> --notification-configuration file://notification.json

Create the evidence bucket. Object Lock can be switched on only at creation or on a versioned bucket, and creating the bucket with it turns versioning on:

aws s3api create-bucket --bucket <bucket>-evidence --region eu-central-1 --object-lock-enabled-for-bucket \
  --create-bucket-configuration LocationConstraint=eu-central-1
aws s3api put-public-access-block --bucket <bucket>-evidence --public-access-block-configuration \
  BlockPublicAcls=true,IgnorePublicAcls=true,BlockPublicPolicy=true,RestrictPublicBuckets=true
aws s3api put-object-lock-configuration --bucket <bucket>-evidence --object-lock-configuration \
  '{"ObjectLockEnabled":"Enabled","Rule":{"DefaultRetention":{"Mode":"GOVERNANCE","Days":1}}}'

GOVERNANCE with one day of retention lets a principal with s3:BypassGovernanceRetention remove objects, which suits a trial you will delete. For evidence you must keep, choose COMPLIANCE mode and a retention that matches your obligations: then nobody, the account root included, can delete a version or shorten its retention until it expires. Object Lock blocks destruction, not hiding: a plain DeleteObject still adds a delete marker that hides the latest version from listings. The evidence chain detects that gap and the original version stays retrievable by its version id (Evidence).

Create the signing key:

aws kms create-key --key-spec ECC_NIST_P256 --key-usage SIGN_VERIFY --query KeyMetadata.Arn --output text

Expected output:

arn:aws:kms:<region>:<account>:key/<id>

What just happened: you created the bucket whose uploads get marked, the queue that carries its events (with a dead-letter queue after 5 receives), the Object Lock bucket for the evidence log and the key that signs.

2. Issue a certificate over the KMS public key

For development the development CA is enough:

laidline keys dev-cert --kms-key-id <key ARN> --region eu-central-1 --org "Acme GmbH"

What just happened: Laidline wrote dev-chain.pem and dev-ca.pem; no private key file exists, the key stays in KMS. In production order a certificate over the same key as Certificates and keys describes (laidline keys csr).

3. Configure the worker

Create laidline.ingest.yaml:

system: {name: acme-flux, version: "2026.10"}
signing:
  backend: aws-kms
  aws: {key_id: "arn:aws:kms:eu-central-1:<account>:key/<id>", region: eu-central-1}
  cert_chain_path: ./certs/dev-chain.pem
  tsa: {required: false}
# paste the watermark and payload_keys blocks that laidline init wrote; never hand-write watermark.issuer_id
evidence:
  sinks:
    - {type: s3, bucket: <bucket>-evidence, object_lock: true}
ingest:
  source: s3_sqs
  queue_url: "https://sqs.eu-central-1.amazonaws.com/<account>/<name>"
  dlq_url: "https://sqs.eu-central-1.amazonaws.com/<account>/<name>-dlq"
  prefixes: ["generated/"]
  writeback: in_place
  upstream_patterns: {"generated/replicate/": replicate, "generated/fal/": fal}
  s3: {region: eu-central-1}

The ingest keys are in Configuration, the s3 sink in Configuration. tsa: {required: false} suits a trial; leave the default in production (Certificates and keys).

What just happened: the worker now knows its queue, its dead-letter queue, the prefixes to mark and where the evidence goes.

4. Run the worker

laidline ingest --config laidline.ingest.yaml --max-batches 3

Expected output, one JSON count of outcomes for the whole run (per-object outcomes are INFO lines on stderr); it is empty while the queue is:

{}

--max-batches N stops after N polls; leave it off to run until you stop the worker. A poll returns up to 10 messages, often fewer. The first poll after you create the notification logs sqs: dropping message without object-created records for S3's s3:TestEvent message; that is expected.

What just happened: the worker long-polled the queue three times. With nothing uploaded yet it marked nothing.

Check it worked

aws s3 cp in.png s3://<bucket>/generated/replicate/in.png

Wait a few seconds for the worker, then read the object's metadata:

aws s3api head-object --bucket <bucket> --key generated/replicate/in.png --query Metadata

Expected output:

{
    "laidline-asset-id": "...",
    "laidline-manifest-id": "..."
}

Download the object and run laidline detect on it: it reports marked_by_this_system.

IAM

This is the least-privilege policy for the worker's role. Replace the placeholders; the evidence statements apply when the evidence sink is an s3 sink.

{
  "Version": "2012-10-17",
  "Statement": [
    {"Sid": "ReadWriteObjects", "Effect": "Allow", "Action": ["s3:GetObject", "s3:PutObject"], "Resource": "arn:aws:s3:::<bucket>/*"},
    {"Sid": "ConsumeQueue", "Effect": "Allow", "Action": ["sqs:ReceiveMessage", "sqs:DeleteMessage", "sqs:ChangeMessageVisibility", "sqs:GetQueueAttributes"], "Resource": "arn:aws:sqs:<region>:<account>:<name>"},
    {"Sid": "ForwardToDlq", "Effect": "Allow", "Action": ["sqs:SendMessage"], "Resource": "arn:aws:sqs:<region>:<account>:<name>-dlq"},
    {"Sid": "SignWithKey", "Effect": "Allow", "Action": ["kms:Sign", "kms:GetPublicKey"], "Resource": "arn:aws:kms:<region>:<account>:key/<id>"},
    {"Sid": "EvidenceAppendOnly", "Effect": "Allow", "Action": ["s3:PutObject", "s3:GetObject", "s3:PutObjectRetention"], "Resource": "arn:aws:s3:::<bucket>-evidence/*"},
    {"Sid": "EvidenceList", "Effect": "Allow", "Action": ["s3:ListBucket", "s3:GetBucketObjectLockConfiguration"], "Resource": "arn:aws:s3:::<bucket>-evidence"}
  ]
}

The queue itself needs the resource policy from step 1 (principal s3.amazonaws.com, aws:SourceArn the bucket, aws:SourceAccount your account).

The evidence sink needs ListBucket in production: without it S3 answers a GET of a missing key with 403, so a registry lookup for an unknown asset raises. The policy has no s3:DeleteObject*. Evidence explains each permission. Emptying a GOVERNANCE evidence bucket when you tear a trial down needs s3:ListBucketVersions, s3:GetObjectVersion, s3:DeleteObjectVersion and s3:BypassGovernanceRetention; grant them to the principal that does that, never to the worker's role.

If aws sts get-caller-identity shows the account root, use it for a one-off only and create an IAM user or role from this policy for anything else.

Idempotency, echoes and retries

• Writing back in place fires a second ObjectCreated event. The worker sees laidline-asset-id on HEAD and acks it (skipped_already_marked, reason metadata). With writeback: prefix:marked/ plus a notification prefix filter there is no echo at all.
• A copy that lost its metadata is still recognised: the object's own manifest carries the asset id, and if that id is in this deployment's evidence log the object is skipped (reason manifest). Nothing is ever double-watermarked.
• unsupported_mime, unsupported_format, unsupported_input, undecodable_image, undecodable_audio, undecodable_video, audio_too_long, video_too_long, video_too_large, encode_failed, audio_unavailable, video_unavailable and video_disabled fail immediately (logged failed, forwarded to the DLQ, acked). Anything else (TSA down, KMS throttled, S3 error) is redelivered (visibility reset to 0) until ingest.max_attempts, then failed the same way. SQS's own redrive policy is a second safety net. After fixing the cause, move DLQ messages back with aws sqs start-message-move-task --source-arn <DLQ ARN> --destination-arn <queue ARN>, or re-upload the object.
• Objects over ingest.max_bytes and keys outside ingest.prefixes are acked as skipped_filtered.

Images, audio and video

The default ingest.mime_allowlist is png, jpeg and webp: other objects are acked skipped_filtered with only an INFO line. For audio and video add mime_allowlist: [image/png, image/jpeg, image/webp, audio/wav, video/mp4], install the audio and video extras, and set watermark.video.enabled: true for video.

Video objects (video/mp4 or video/quicktime in ingest.mime_allowlist, with watermark.video.enabled) are marked in place like images, written back as video/mp4. The worker holds the whole object in memory and marks it synchronously (VideoSeal on CPU takes seconds per second of video), so ingest.max_bytes (default 64 MiB) is the video size limit as well: raise it deliberately, keep ingest.visibility_timeout (default 180 s) above the longest mark, and let watermark.video.max_duration_s refuse long videos (video_too_long, not retried, like undecodable_video).

Audio objects (audio/wav, audio/flac, audio/mpeg, audio/mp4, audio/ogg or audio/aac in ingest.mime_allowlist) are AudioSeal-marked and written back in their own container; wav, flac, mp3 and m4a are also signed (Audio). Allowlist aliases such as audio/x-wav are read as their canonical type.

The PixelSeal image weights (1.24 GB download, 1.55 GB on disk) and the AudioSeal and VideoSeal weights download on first use: the first image object waits for PixelSeal (23.7 s for the first ingest process in the measured run, the download plus three polls). Run laidline models fetch in the image to pre-fill them.

Costs

A KMS asymmetric key costs about $1 per month while it exists, plus $0.15 per 10,000 sign requests (kms:GetPublicKey is called once per process start). S3 requests and SQS long-polls are negligible at low volumes. Deleting the key schedules it for deletion (7-day minimum), after which the monthly charge stops.

Limits

• A synchronous in-region S3 evidence write adds about 55 to 77 ms to a mark; run the worker in the bucket's region (Benchmarks).
• The worker holds each object in memory: ingest.max_bytes (64 MiB) bounds images, audio and video alike.
• Delivery is at least once. The metadata and manifest checks make a repeated delivery a skip, not a second mark.

What was proven

Live on 19 Oct 2026 against a real AWS account in eu-central-1: JPEG and PNG images marked in place with metadata and cache-control preserved, the write-back echo skipped, a re-put marked object skipped through its manifest, an upstream-signed object kept as an ingredient, a text file filtered, write-back under marked/, KMS signing that validates and rejects tampering, and Object Lock evidence appends with detect through the S3 registry. A WAV and a 5 s CIF mp4 were marked in place once the allowlist and watermark.video.enabled named them. Audio and video ingest are also covered by tests with an in-memory store.

R2 and other S3-compatible stores

The same worker talks to any S3-compatible store: set ingest.s3.endpoint_url and the store's S3 credentials.

R2 has no S3 Object Lock API, so an s3 evidence sink on R2 must keep object_lock: false. With object_lock: true it fails at start with "... could not be read (...); R2 has no S3 Object Lock API". Keep the evidence log on a different store (S3 with Object Lock, or Postgres) when you need write-once retention.

Cloudflare R2 with Queues

R2 event notifications go to a Cloudflare Queue. laidline ingest runs in your own container (Workers cannot run the markers) and consumes the queue through Cloudflare's HTTP pull API. Objects are read and written through R2's S3 API.

pip install "laidline[image,ingest]"

1. Create the queues and the notification (wrangler 3.x)

wrangler queues create laidline-ingest
wrangler queues create laidline-ingest-dlq
wrangler queues consumer http add laidline-ingest --dead-letter-queue laidline-ingest-dlq --message-retries 5
wrangler r2 bucket notification create <bucket> --event-type object-create --queue laidline-ingest --prefix generated/

What just happened: objects created under generated/ now post a message to laidline-ingest; a message delivered 5 times without an acknowledgement moves to laidline-ingest-dlq.

2. Create the tokens

In the Cloudflare dashboard choose "Create custom token":

1. Queues pull consumer: permission Queues Edit on the account. Use it as CLOUDFLARE_API_TOKEN. A token with R2 access only is refused at the pull with 401 (code 10000).
2. R2 S3 API token: Object Read & Write on the bucket. Use the pair as AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY. You can also derive the pair from an API token: the access key id is the token's id (result.id from /user/tokens/verify) and the secret is the SHA-256 hex of the token value. Export them with AWS_SESSION_TOKEN and AWS_PROFILE unset, or the S3 client picks up those AWS credentials instead.

One account token can do both: give it Workers R2 Storage Edit and Queues Edit, use it as CLOUDFLARE_API_TOKEN and derive the S3 pair from it.

3. Configure the worker

ingest:
  source: r2_queues
  prefixes: ["generated/"]
  r2: {account_id: <account id>, queue_id: <queue id from `wrangler queues list`>, api_token_env: CLOUDFLARE_API_TOKEN}
  s3: {region: auto, endpoint_url: https://<account id>.r2.cloudflarestorage.com}
  wait_seconds: 2

Set ingest.wait_seconds as well: the Queues pull API answers at once and does not long-poll, so it is the sleep after an empty pull (2 worked). Without a sleep the worker pulls in a tight loop into Cloudflare's API limit of 1,200 requests per 5 minutes. Add the signing, watermark, payload_keys and evidence blocks as for S3 above.

4. Run the worker

laidline ingest --config laidline.yaml

Poison messages go to the consumer's dead-letter queue after --message-retries deliveries; the worker also logs failed after ingest.max_attempts.

What just happened: the worker pulls the queue over HTTP, marks each uploaded object through R2's S3 API and writes it back in place. Upload an image under generated/ and check its metadata and laidline detect as in Check it worked.

R2 limits

• Under watermark.device: auto the worker marked on the CPU although the host had a CUDA GPU; marking on the GPU was not run.
• The first event waits for the models to load on the CPU (most of the 28.9 s from upload to write-back in the measured run). Warm-worker latency was not measured.
• Each in-place write-back raises its own PutObject event, which the worker skips by the object's metadata (skipped_already_marked).
• Not run: the dead-letter path after retries, marking on the GPU and multipart uploads.

R2 proven

Run on 21 Oct 2026 against a real R2 bucket and Queue: laidline ingest pulled an uploaded JPEG and marked it in place 28.9 s after the upload, most of it loading the models on the CPU. laidline detect graded it A (C2PA valid and trusted, watermark found), and the queue backlog was 0 with nothing in the dead-letter queue. Pull bodies arrive as plain JSON strings (CF-Content-Type: json), not base64.

Troubleshooting

Reference

• Configuration: every ingest key.
• Evidence: the s3 sink, Object Lock and each permission.
• Certificates and keys: the KMS certificate, CSRs and timestamps.
• AWS Lambda: the same hook as a Lambda function.
• Audio and Video: marking those media.
• Benchmarks: in-region evidence write latency.