--- name: Certyo + Amazon Web Services Integration version: 1.0.0 description: Generate AWS → Certyo integration code with EventBridge, Lambda, Step Functions, and working examples api_base: https://www.certyos.com auth: X-API-Key header last_updated: 2026-04-14 --- # Certyo + Amazon Web Services Integration Skill This skill generates production-ready AWS integrations that ingest records into Certyo's blockchain-anchored authenticity platform. It uses EventBridge API Destinations for simple forwarding, Lambda for transformation, Step Functions for verification workflows, and CDK for infrastructure as code. ## Certyo API Reference ### Endpoints | Method | Path | Description | Response | |--------|------|-------------|----------| | `POST` | `/api/v1/records` | Ingest a single record | `202 Accepted` | | `POST` | `/api/v1/records/bulk` | Ingest up to 1000 records | `202 Accepted` | | `POST` | `/api/v1/verify/record` | Verify blockchain anchoring | `200 OK` | | `GET` | `/api/v1/records` | Query records | `200 OK` | ### Authentication All requests require the `X-API-Key` header with a valid Certyo API key. ### Record Payload ```json { "tenantId": "string (required)", "database": "string (required)", "collection": "string (required)", "recordId": "string (required)", "recordPayload": { "...any JSON (required)" }, "clientId": "string (optional)", "recordVersion": "string (optional, default '1')", "operationType": "upsert|insert|update|delete (optional)", "sourceTimestamp": "ISO 8601 (optional)", "idempotencyKey": "string (optional)" } ``` ### Response (202 Accepted) ```json { "recordId": "string", "recordHash": "string (SHA-256)", "tenantId": "string", "acceptedAt": "ISO 8601", "idempotencyReplayed": false } ``` ### Pipeline Record ingestion flows through: Kafka accumulation (1000 records or 60s flush) then Merkle tree computation then IPFS pin then Polygon anchor (~60-90s end to end). ## Integration Pattern ### Simple Forwarding (No Lambda) ``` EventBridge Rule → Input Transformer → API Destination → Certyo API ``` Use this when the source event structure can be mapped to the Certyo payload using EventBridge Input Transformers alone. This is the cheapest pattern (no Lambda invocations). ### Transform and Forward (Lambda) ``` EventBridge Rule → Lambda (transform) → Certyo API | SQS DLQ (failures) ``` Use this when the source event needs complex transformation, enrichment from other services, or conditional logic before sending to Certyo. ### Verification Workflow (Step Functions) ``` Step Functions (Standard Workflow): Ingest → Wait 90s → Verify → Branch (verified/retry/fail) ``` Use Standard Workflows for verification because they support the Wait state (up to 1 year). Use Express Workflows for high-volume short-duration ingestion. ### AWS Services Used | Service | Role | Cost Model | |---------|------|------------| | **EventBridge** | Event routing, rules, input transformation | $1/million events | | **API Destinations** | Direct HTTP forwarding to Certyo (no Lambda) | Included in EventBridge | | **Lambda** | Complex transformation, enrichment | Per-invocation + duration | | **Step Functions** | Orchestration, verification with Wait state | Per state transition | | **Secrets Manager** | Certyo API key storage | $0.40/secret/month | | **SQS** | Dead-letter queue for failed events | $0.40/million requests | | **API Gateway** | Optional: expose Certyo proxy to internal consumers | Per-request | ## Authentication ### EventBridge Connection with API Key EventBridge Connections store the Certyo API key and inject it as the `X-API-Key` header on every API Destination invocation. The key is encrypted at rest and managed by EventBridge. ### Lambda uses Secrets Manager Lambda functions retrieve the Certyo API key from Secrets Manager at cold start and cache it for the lifetime of the execution context. Use IAM roles to grant `secretsmanager:GetSecretValue` permission. ### IAM Roles Every component uses a dedicated IAM role with least-privilege permissions. Lambda roles get Secrets Manager read access and CloudWatch Logs write access. EventBridge roles get API Destination invocation access. ## Code Examples ### Python Lambda: S3 Trigger to Certyo Ingestion **`lambda/certyo_s3_ingest.py`**: ```python """ Lambda function triggered by S3 PutObject events. Reads the uploaded object metadata and ingests a record into Certyo. """ import hashlib import json import os import urllib.request import urllib.error from datetime import datetime, timezone import boto3 from botocore.exceptions import ClientError # Cache API key across warm invocations _api_key_cache = None CERTYO_BASE_URL = os.environ.get("CERTYO_BASE_URL", "https://www.certyos.com") CERTYO_TENANT_ID = os.environ["CERTYO_TENANT_ID"] SECRET_ARN = os.environ["CERTYO_API_KEY_SECRET_ARN"] def get_api_key() -> str: """Retrieve Certyo API key from Secrets Manager (cached).""" global _api_key_cache if _api_key_cache is not None: return _api_key_cache client = boto3.client("secretsmanager") try: response = client.get_secret_value(SecretId=SECRET_ARN) _api_key_cache = response["SecretString"] return _api_key_cache except ClientError as e: raise RuntimeError(f"Failed to retrieve Certyo API key: {e}") from e def build_certyo_payload(bucket: str, key: str, s3_event: dict) -> dict: """Transform S3 event into Certyo record payload.""" size = s3_event.get("object", {}).get("size", 0) etag = s3_event.get("object", {}).get("eTag", "") event_time = s3_event.get("eventTime", datetime.now(timezone.utc).isoformat()) record_id = hashlib.sha256(f"{bucket}/{key}".encode()).hexdigest()[:32] return { "tenantId": CERTYO_TENANT_ID, "database": "s3", "collection": bucket, "recordId": record_id, "recordPayload": { "bucket": bucket, "key": key, "size": size, "etag": etag, "eventName": s3_event.get("eventName", "PutObject"), "sourceIPAddress": s3_event.get("requestParameters", {}).get( "sourceIPAddress", "unknown" ), }, "operationType": "insert", "sourceTimestamp": event_time, "idempotencyKey": f"s3:{bucket}:{key}:{etag}", } def post_to_certyo(payload: dict) -> dict: """Send record to Certyo API.""" api_key = get_api_key() data = json.dumps(payload).encode("utf-8") req = urllib.request.Request( f"{CERTYO_BASE_URL}/api/v1/records", data=data, headers={ "X-API-Key": api_key, "Content-Type": "application/json", }, method="POST", ) try: with urllib.request.urlopen(req, timeout=30) as response: body = json.loads(response.read().decode("utf-8")) print( f"Record ingested. recordId={body.get('recordId')} " f"hash={body.get('recordHash')} " f"status={response.status}" ) return body except urllib.error.HTTPError as e: error_body = e.read().decode("utf-8") if e.fp else "no body" print(f"Certyo API error. status={e.code} body={error_body}") raise def handler(event, context): """Lambda handler for S3 event notifications.""" results = [] for record in event.get("Records", []): s3_info = record.get("s3", {}) bucket = s3_info.get("bucket", {}).get("name", "") key = s3_info.get("object", {}).get("key", "") if not bucket or not key: print(f"Skipping record with missing bucket/key: {record}") continue print(f"Processing s3://{bucket}/{key}") payload = build_certyo_payload(bucket, key, record) result = post_to_certyo(payload) results.append(result) return { "statusCode": 200, "body": json.dumps( {"processed": len(results), "results": results} ), } ``` ### CDK TypeScript: EventBridge + API Destination + DLQ **`lib/certyo-eventbridge-stack.ts`**: ```typescript import * as cdk from "aws-cdk-lib"; import * as events from "aws-cdk-lib/aws-events"; import * as targets from "aws-cdk-lib/aws-events-targets"; import * as sqs from "aws-cdk-lib/aws-sqs"; import * as secretsmanager from "aws-cdk-lib/aws-secretsmanager"; import * as iam from "aws-cdk-lib/aws-iam"; import { Construct } from "constructs"; export interface CertyoEventBridgeStackProps extends cdk.StackProps { certyoTenantId: string; certyoApiKeySecretArn?: string; } export class CertyoEventBridgeStack extends cdk.Stack { public readonly connection: events.Connection; public readonly apiDestination: events.ApiDestination; constructor( scope: Construct, id: string, props: CertyoEventBridgeStackProps ) { super(scope, id, props); // Dead-letter queue for failed deliveries const dlq = new sqs.Queue(this, "CertyoDLQ", { queueName: "certyo-eventbridge-dlq", retentionPeriod: cdk.Duration.days(14), encryption: sqs.QueueEncryption.SQS_MANAGED, }); // Secrets Manager secret for Certyo API key const apiKeySecret = props.certyoApiKeySecretArn ? secretsmanager.Secret.fromSecretCompleteArn( this, "CertyoApiKey", props.certyoApiKeySecretArn ) : new secretsmanager.Secret(this, "CertyoApiKey", { secretName: "certyo/api-key", description: "Certyo platform API key for record ingestion", }); // EventBridge Connection with API Key auth this.connection = new events.Connection(this, "CertyoConnection", { connectionName: "certyo-api-connection", description: "Connection to Certyo blockchain platform", authorization: events.Authorization.apiKey( "X-API-Key", cdk.SecretValue.secretsManager(apiKeySecret.secretArn) ), }); // API Destination targeting Certyo records endpoint this.apiDestination = new events.ApiDestination( this, "CertyoApiDestination", { apiDestinationName: "certyo-ingest-records", connection: this.connection, endpoint: "https://www.certyos.com/api/v1/records", httpMethod: events.HttpMethod.POST, rateLimitPerSecond: 100, description: "Certyo record ingestion API destination", } ); // EventBridge rule: match application events and forward to Certyo const ingestRule = new events.Rule(this, "CertyoIngestRule", { ruleName: "certyo-ingest-records", description: "Forward application events to Certyo for blockchain anchoring", eventPattern: { source: ["myapp.records"], detailType: ["RecordCreated", "RecordUpdated"], }, }); // Input transformer: map source event to Certyo payload ingestRule.addTarget( new targets.ApiDestination(this.apiDestination, { deadLetterQueue: dlq, retryAttempts: 185, // ~24h coverage with exponential backoff event: events.RuleTargetInput.fromObject({ tenantId: props.certyoTenantId, database: events.EventField.fromPath("$.source"), collection: events.EventField.fromPath("$.detail-type"), recordId: events.EventField.fromPath("$.detail.id"), recordPayload: events.EventField.fromPath("$.detail"), operationType: "upsert", sourceTimestamp: events.EventField.fromPath("$.time"), idempotencyKey: events.EventField.fromPath("$.id"), }), }) ); // Rule for bulk ingestion (batched events) const bulkRule = new events.Rule(this, "CertyoBulkRule", { ruleName: "certyo-bulk-ingest", description: "Forward bulk record events to Certyo bulk endpoint", eventPattern: { source: ["myapp.records"], detailType: ["RecordBatchCreated"], }, }); const bulkApiDestination = new events.ApiDestination( this, "CertyoBulkApiDestination", { apiDestinationName: "certyo-ingest-records-bulk", connection: this.connection, endpoint: "https://www.certyos.com/api/v1/records/bulk", httpMethod: events.HttpMethod.POST, rateLimitPerSecond: 10, } ); bulkRule.addTarget( new targets.ApiDestination(bulkApiDestination, { deadLetterQueue: dlq, retryAttempts: 185, event: events.RuleTargetInput.fromEventPath("$.detail.records"), }) ); // CloudWatch alarm on DLQ depth const dlqAlarm = dlq .metricApproximateNumberOfMessagesVisible() .createAlarm(this, "CertyoDLQAlarm", { alarmName: "certyo-dlq-messages", threshold: 1, evaluationPeriods: 1, treatMissingData: cdk.aws_cloudwatch.TreatMissingData.NOT_BREACHING, }); // Outputs new cdk.CfnOutput(this, "DLQUrl", { value: dlq.queueUrl }); new cdk.CfnOutput(this, "ConnectionArn", { value: this.connection.connectionArn, }); new cdk.CfnOutput(this, "ApiDestinationArn", { value: this.apiDestination.apiDestinationArn, }); } } ``` ### Step Functions ASL: Ingest, Wait, Verify, Branch **`step-functions/certyo-verify-workflow.asl.json`**: ```json { "Comment": "Certyo record ingestion and verification workflow. Ingests a record, waits for blockchain anchoring, then verifies.", "StartAt": "IngestRecord", "States": { "IngestRecord": { "Type": "Task", "Resource": "arn:aws:states:::http:invoke", "Parameters": { "ApiEndpoint": "https://www.certyos.com/api/v1/records", "Method": "POST", "Authentication": { "ConnectionArn.$": "$.connectionArn" }, "RequestBody": { "tenantId.$": "$.tenantId", "database.$": "$.database", "collection.$": "$.collection", "recordId.$": "$.recordId", "recordPayload.$": "$.recordPayload", "operationType.$": "$.operationType", "sourceTimestamp.$": "$.sourceTimestamp", "idempotencyKey.$": "$.idempotencyKey" }, "Headers": { "Content-Type": "application/json" } }, "ResultPath": "$.ingestResult", "ResultSelector": { "recordId.$": "$.ResponseBody.recordId", "recordHash.$": "$.ResponseBody.recordHash", "acceptedAt.$": "$.ResponseBody.acceptedAt", "statusCode.$": "$.StatusCode" }, "Retry": [ { "ErrorEquals": ["States.Http5xx", "States.Http429"], "IntervalSeconds": 10, "MaxAttempts": 5, "BackoffRate": 2.0 } ], "Catch": [ { "ErrorEquals": ["States.ALL"], "ResultPath": "$.error", "Next": "IngestionFailed" } ], "Next": "WaitForAnchoring" }, "WaitForAnchoring": { "Type": "Wait", "Seconds": 90, "Comment": "Wait 90 seconds for Kafka accumulation + Merkle tree + IPFS + Polygon anchoring", "Next": "VerifyRecord" }, "VerifyRecord": { "Type": "Task", "Resource": "arn:aws:states:::http:invoke", "Parameters": { "ApiEndpoint": "https://www.certyos.com/api/v1/verify/record", "Method": "POST", "Authentication": { "ConnectionArn.$": "$.connectionArn" }, "RequestBody": { "tenantId.$": "$.tenantId", "recordId.$": "$.ingestResult.recordId", "recordHash.$": "$.ingestResult.recordHash" }, "Headers": { "Content-Type": "application/json" } }, "ResultPath": "$.verifyResult", "ResultSelector": { "verified.$": "$.ResponseBody.verified", "merkleRoot.$": "$.ResponseBody.merkleRoot", "transactionHash.$": "$.ResponseBody.transactionHash" }, "Retry": [ { "ErrorEquals": ["States.Http5xx"], "IntervalSeconds": 15, "MaxAttempts": 3, "BackoffRate": 2.0 } ], "Catch": [ { "ErrorEquals": ["States.ALL"], "ResultPath": "$.error", "Next": "VerificationFailed" } ], "Next": "CheckVerification" }, "CheckVerification": { "Type": "Choice", "Choices": [ { "Variable": "$.verifyResult.verified", "BooleanEquals": true, "Next": "RecordVerified" } ], "Default": "IncrementRetryCount" }, "IncrementRetryCount": { "Type": "Pass", "Parameters": { "retryCount.$": "States.MathAdd($.retryCount, 1)" }, "ResultPath": "$.retryState", "Next": "CheckRetryLimit" }, "CheckRetryLimit": { "Type": "Choice", "Choices": [ { "Variable": "$.retryState.retryCount", "NumericGreaterThan": 10, "Next": "VerificationTimeout" } ], "Default": "WaitBeforeRetry" }, "WaitBeforeRetry": { "Type": "Wait", "Seconds": 30, "Next": "VerifyRecord" }, "RecordVerified": { "Type": "Succeed", "Comment": "Record successfully verified on Polygon blockchain" }, "VerificationTimeout": { "Type": "Fail", "Error": "VerificationTimeout", "Cause": "Record was not verified after 10 retry attempts (~7 minutes)" }, "IngestionFailed": { "Type": "Fail", "Error": "IngestionFailed", "Cause": "Failed to ingest record into Certyo after retries" }, "VerificationFailed": { "Type": "Fail", "Error": "VerificationFailed", "Cause": "Verification API call failed after retries" } } } ``` ### AWS CLI: Full Pipeline Setup ```bash #!/bin/bash # Deploy Certyo integration pipeline on AWS # Prerequisites: AWS CLI configured, appropriate IAM permissions set -euo pipefail REGION="${AWS_REGION:-us-east-1}" ACCOUNT_ID=$(aws sts get-caller-identity --query Account --output text) CERTYO_API_KEY="${CERTYO_API_KEY:?Set CERTYO_API_KEY environment variable}" CERTYO_TENANT_ID="${CERTYO_TENANT_ID:?Set CERTYO_TENANT_ID environment variable}" echo "Deploying Certyo integration pipeline in ${REGION}..." # 1. Store API key in Secrets Manager echo "Creating Secrets Manager secret..." aws secretsmanager create-secret \ --name "certyo/api-key" \ --description "Certyo platform API key" \ --secret-string "${CERTYO_API_KEY}" \ --region "${REGION}" 2>/dev/null || \ aws secretsmanager update-secret \ --secret-id "certyo/api-key" \ --secret-string "${CERTYO_API_KEY}" \ --region "${REGION}" SECRET_ARN=$(aws secretsmanager describe-secret \ --secret-id "certyo/api-key" \ --query ARN --output text \ --region "${REGION}") echo "Secret ARN: ${SECRET_ARN}" # 2. Create EventBridge Connection echo "Creating EventBridge connection..." CONNECTION_ARN=$(aws events create-connection \ --name "certyo-api-connection" \ --authorization-type "API_KEY" \ --auth-parameters "{ \"ApiKeyAuthParameters\": { \"ApiKeyName\": \"X-API-Key\", \"ApiKeyValue\": \"${CERTYO_API_KEY}\" } }" \ --region "${REGION}" \ --query ConnectionArn --output text) echo "Connection ARN: ${CONNECTION_ARN}" # 3. Create API Destination echo "Creating API Destination..." API_DEST_ARN=$(aws events create-api-destination \ --name "certyo-ingest-records" \ --connection-arn "${CONNECTION_ARN}" \ --invocation-endpoint "https://www.certyos.com/api/v1/records" \ --http-method "POST" \ --invocation-rate-limit-per-second 100 \ --region "${REGION}" \ --query ApiDestinationArn --output text) echo "API Destination ARN: ${API_DEST_ARN}" # 4. Create SQS DLQ echo "Creating DLQ..." DLQ_URL=$(aws sqs create-queue \ --queue-name "certyo-eventbridge-dlq" \ --attributes '{ "MessageRetentionPeriod": "1209600", "SqsManagedSseEnabled": "true" }' \ --region "${REGION}" \ --query QueueUrl --output text) DLQ_ARN=$(aws sqs get-queue-attributes \ --queue-url "${DLQ_URL}" \ --attribute-names QueueArn \ --query Attributes.QueueArn --output text \ --region "${REGION}") echo "DLQ ARN: ${DLQ_ARN}" # 5. Create IAM role for EventBridge to invoke API Destination echo "Creating IAM role for EventBridge..." TRUST_POLICY='{ "Version": "2012-10-17", "Statement": [{ "Effect": "Allow", "Principal": {"Service": "events.amazonaws.com"}, "Action": "sts:AssumeRole" }] }' ROLE_ARN=$(aws iam create-role \ --role-name "certyo-eventbridge-role" \ --assume-role-policy-document "${TRUST_POLICY}" \ --query Role.Arn --output text 2>/dev/null || \ aws iam get-role \ --role-name "certyo-eventbridge-role" \ --query Role.Arn --output text) # Attach permissions for API Destination and DLQ aws iam put-role-policy \ --role-name "certyo-eventbridge-role" \ --policy-name "certyo-api-destination-policy" \ --policy-document "{ \"Version\": \"2012-10-17\", \"Statement\": [ { \"Effect\": \"Allow\", \"Action\": [\"events:InvokeApiDestination\"], \"Resource\": [\"${API_DEST_ARN}\"] }, { \"Effect\": \"Allow\", \"Action\": [\"sqs:SendMessage\"], \"Resource\": [\"${DLQ_ARN}\"] } ] }" echo "IAM Role ARN: ${ROLE_ARN}" # 6. Create EventBridge rule with input transformer echo "Creating EventBridge rule..." aws events put-rule \ --name "certyo-ingest-records" \ --event-pattern '{ "source": ["myapp.records"], "detail-type": ["RecordCreated", "RecordUpdated"] }' \ --state "ENABLED" \ --description "Forward record events to Certyo for blockchain anchoring" \ --region "${REGION}" # Add API Destination target with input transformer aws events put-targets \ --rule "certyo-ingest-records" \ --targets "[{ \"Id\": \"certyo-api-destination\", \"Arn\": \"${API_DEST_ARN}\", \"RoleArn\": \"${ROLE_ARN}\", \"InputTransformer\": { \"InputPathsMap\": { \"id\": \"$.detail.id\", \"source\": \"$.source\", \"detailType\": \"$.detail-type\", \"detail\": \"$.detail\", \"time\": \"$.time\", \"eventId\": \"$.id\" }, \"InputTemplate\": \"{\\\"tenantId\\\": \\\"${CERTYO_TENANT_ID}\\\", \\\"database\\\": , \\\"collection\\\": , \\\"recordId\\\": , \\\"recordPayload\\\": , \\\"operationType\\\": \\\"upsert\\\", \\\"sourceTimestamp\\\":