S3 bucket connector

Syncs CSV/Parquet objects from your S3 bucket into canonical events. Object-incremental like the GCS connector; read-only against your bucket.

Prerequisites

An API key with the admin scope.
An S3 bucket + its region, with event-shaped objects under a prefix.
An IAM user (or role) you control.

Create the credential (least privilege)

The connector calls ListObjectsV2 under your prefix (s3:ListBucket), then HeadObject and ranged GetObject on the objects it reads (both s3:GetObject). Create a dedicated IAM user whose policy allows exactly that — no writes, no other buckets, and listing limited to the prefix:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "TesseraListPrefix",
      "Effect": "Allow",
      "Action": "s3:ListBucket",
      "Resource": "arn:aws:s3:::acme-shipment-events",
      "Condition": { "StringLike": { "s3:prefix": ["events/*"] } }
    },
    {
      "Sid": "TesseraReadObjects",
      "Effect": "Allow",
      "Action": "s3:GetObject",
      "Resource": "arn:aws:s3:::acme-shipment-events/events/*"
    }
  ]
}

Without a prefix, drop the Condition and use arn:aws:s3:::acme-shipment-events/* as the object resource. A manifest_object and every key it lists must sit under the prefix. When the objects use SSE-KMS with a customer-managed key, also allow kms:Decrypt on that key. The console's S3 form renders this policy with your bucket and prefix filled in.

Mint an access key for that user. Tessera stores it envelope-encrypted; it is never readable through the API again, and never appears in logs or job arguments.

Create the source

{
  "name": "shipment-events",
  "kind": "s3",
  "config": {
    "bucket": "acme-shipment-events",
    "region": "us-east-1",
    "prefix": "events/",
    "format": "csv"
  },
  "credentials": {
    "access_key_id": "…",
    "secret_access_key": "…"
  }
}

Map columns, test, and run the first sync

POST /sources/{id}/connection-diagnostics runs two checks: list_objects_v2 (stage object) lists up to five objects under the prefix, and get_object (stage authorization) reads one byte of the first one; with an empty prefix, get_object reports warn because object reads are not verified yet. The AWS error code decides a failed check's failure cause: an unknown, inactive or mistyped key is authentication, a policy that does not allow the action is authorization, an unreachable endpoint is network, a missing bucket is not_found and a wrong region is configuration. Details never repeat AWS's message or the access key. The response's checked_at is the server time of the test, which is also recorded in the audit log as source.test_connection. POST /sources/{id}/test-connection runs the same checks and returns only their names, statuses and details.

The logistics mapping — shipment entities from the sample dataset's CSV:

mapping-logistics.json
{
  "version": 1,
  "entity": { "type": { "constant": "shipment" }, "id_column": "shipment_ref" },
  "event_type": { "from_column": "event_type" },
  "timestamp": { "column": "occurred_at" },
  "event_id_column": "event_id",
  "attributes": [
    { "column": "weight_kg" },
    { "column": "carrier" },
    { "column": "location" },
    { "column": "delay_hours" },
    { "column": "delay_reason" }
  ]
}

Download connect/s3/mapping-logistics.json — this exact file runs in CI on every change (hosts shown are for this environment).

01-create-and-sync.sh
#!/usr/bin/env bash
# S3 bucket connector: create the source (vaulted credentials), test, map, sync.
set -euo pipefail

: "${TESSERA_API_BASE:?e.g. https://api.staging.tessera.zone/v1}"
: "${TESSERA_API_KEY:?your tsk_test_… key}"
: "${TESSERA_S3_ACCESS_KEY_ID:?read-only IAM access key id}"
: "${TESSERA_S3_SECRET_ACCESS_KEY:?read-only IAM secret}"
AUTH=(-H "Authorization: Bearer $TESSERA_API_KEY")

# 1. Create the source. Credentials are write-only: envelope-encrypted at
#    rest and never readable through the API again.
SOURCE="$(curl -sS -X POST "$TESSERA_API_BASE/sources" "${AUTH[@]}" \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: docs-s3-source" \
  -d '{
    "name": "shipment-events",
    "kind": "s3",
    "config": {
      "bucket": "acme-shipment-events",
      "region": "us-east-1",
      "prefix": "events/",
      "format": "csv"
    },
    "credentials": {
      "access_key_id": "'"$TESSERA_S3_ACCESS_KEY_ID"'",
      "secret_access_key": "'"$TESSERA_S3_SECRET_ACCESS_KEY"'"
    }
  }')"
SOURCE_ID="$(printf '%s' "$SOURCE" | python3 -c 'import json,sys; print(json.load(sys.stdin)["source_id"])')"
printf '%s' "$SOURCE" | python3 -c '
import json, sys
source = json.load(sys.stdin)
print("source created: %s credentials_set=%s" % (source["kind"], source["credentials_set"]))
'

# 2. Test the connection.
curl -sS -X POST "$TESSERA_API_BASE/sources/$SOURCE_ID/test-connection" "${AUTH[@]}" \
  | python3 -c 'import json,sys; print("test-connection:", json.load(sys.stdin)["status"])'

# 3. Map columns, then run the first sync.
curl -sS -X PUT "$TESSERA_API_BASE/sources/$SOURCE_ID/mapping" "${AUTH[@]}" \
  -H "Content-Type: application/json" --data-binary @mapping-logistics.json > /dev/null

SYNC="$(curl -sS -X POST "$TESSERA_API_BASE/sources/$SOURCE_ID/syncs" "${AUTH[@]}" \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: docs-s3-first-sync" \
  -d '{"mode": "full"}')"
SYNC_ID="$(printf '%s' "$SYNC" | python3 -c 'import json,sys; print(json.load(sys.stdin)["sync_id"])')"
for _ in $(seq 1 300); do
  SYNC="$(curl -sS "$TESSERA_API_BASE/syncs/$SYNC_ID" "${AUTH[@]}")"
  STATUS="$(printf '%s' "$SYNC" | python3 -c 'import json,sys; print(json.load(sys.stdin)["status"])')"
  if [ "$STATUS" = "succeeded" ] || [ "$STATUS" = "failed" ]; then break; fi
  sleep 1
done
printf '%s' "$SYNC" | python3 -c '
import json, sys
sync = json.load(sys.stdin)
print("first sync: %s rows_landed=%s rejected=%s"
      % (sync["status"], sync["rows_landed"], sync["rows_rejected"]))
'

Download connect/s3/01-create-and-sync.sh — this exact file runs in CI on every change (hosts shown are for this environment).

In CI this runs against the local stack's fake bucket (seeded with the logistics sample CSV); the calls are identical against real S3.

Rotate the credential

The source holds one static access key. Tessera cannot use a session token or assume a role, so a temporary key you issue for a person or a job stops working when it expires. To rotate: create a second access key for the same IAM user, PATCH /sources/{id} with both fields in credentials, confirm POST /sources/{id}/connection-diagnostics passes, then deactivate the old key in AWS. A sync already running finishes with the key it started with.

Troubleshooting

Symptom	Cause	Fix
`list_objects_v2` fails with `failure: authentication`	the access key is unknown, inactive or expired, or the secret is mistyped	replace the credentials with an active key
`list_objects_v2` fails with `failure: authorization`	policy missing `s3:ListBucket` on the bucket ARN (or the prefix condition does not match)	both statements above are required
`get_object` fails with `failure: authorization`	`s3:GetObject` not granted on `…/*`, or `kms:Decrypt` missing for SSE-KMS objects	note the `/*` — object ARNs differ from the bucket ARN
`failure: configuration`	the bucket lives in a different region	set the bucket's actual `region`
sync lands 0 objects	wrong prefix	check the prefix
credentials rotated	old key revoked upstream	`PATCH /sources/{id}` with new `credentials` (full replacement)

Prerequisites​

Create the credential (least privilege)​

Create the source​

Map columns, test, and run the first sync​

Rotate the credential​

Troubleshooting​