Build a Secure Webhook Ingest

Introduction

30 Second Summary

A meeting platform can send the same update twice. Without a careful checkpoint, a duplicate or altered update can enter the rest of a system as if it were trustworthy.

In this project, you will build a secure webhook admission boundary that places synthetic Zoom meeting events onto an encrypted Amazon SQS FIFO queue. You will preserve raw bytes for HMAC verification before deploying the receiver with AWS CDK.

What You'll Build

Your finished demo turns two successful deliveries of the same signed synthetic meeting event into one normalized FIFO message.

By the end of this project, you'll have:

• A secure admission path that returns 202 for a valid synthetic meeting event. Invalid signatures stop before workflow decisions.
• A versioned event envelope for normalized meeting data. Its FIFO message group keeps each meeting in order.
• A repeatable deployment and teardown that leaves the $20 budget guardrail in place after the ingest stack is removed. Before-and-after inventories prove what remains.
• Secret Mission: Add executable boundary tests proving that requests aged 299 and 300 seconds pass while a request aged 301 seconds fails.

Are there any prerequisites?

You need an AWS account with permission to deploy CDK stacks and budgets. Your WSL2 environment also needs Python 3.12 or later plus Go 1.23 or later.

Before We Start

This module accepts signed synthetic Zoom-compatible RTMS webhook deliveries and emits normalized messages to a FIFO queue. The downstream capture worker stays out of scope so you can focus on building a secure admission boundary.

Validate and Pin the WSL2 Toolchain

Correct webhook code can still fail when the WSL2 clock drifts. An expired role session can make valid AWS calls look broken.

A mismatched AWS CDK toolchain can produce misleading deployment failures. This step pins the repository dependencies before one checker proves the local environment is ready.

In this step, get ready to:

• Check that Python and Go meet the required versions.
• Create the repository with pinned Python and local CDK dependencies.
• Record your predictions before validating the complete WSL2 environment.

Confirm Python and Go versions

The repository setup depends on Python 3.12 or later. Future capture-worker modules also depend on Go 1.23 or later.

• Start your WSL2 distribution from Windows Terminal.
• Check the Python version by running this command:

python3 --version

What does this command check?

The --version option prints the Python version selected by python3. The repository setup uses that interpreter to create .venv.

✔️ I see Python 3.12 or higher

Python meets the project requirement. Continue to the Go check.

ⓧ I see an older Python version

Your current Python interpreter is below the required version. Install Python 3.12 or later before creating the virtual environment.

• Follow the official Python guidance for Unix platforms.
• Install a Python release at version 3.12 or later.
• Close the current WSL2 terminal.
• Start a fresh WSL2 terminal.
• Run the Python version check above again.

ⓧ Python command not found

Python is unavailable inside your WSL2 distribution. Install it in the Linux environment before continuing.

• Follow the official Python guidance for Unix platforms.
• Install Python at version 3.12 or later.
• Close the current WSL2 terminal.
• Start a fresh WSL2 terminal.
• Run the Python version check above again.

• Check the Go version by running this command:

go version

What does this command check?

The go version command prints the active Go toolchain. The version must be 1.23 or later.

✔️ I see Go 1.23 or higher

Go meets the project requirement. Your two unconfirmed language runtimes are ready.

ⓧ I see an older Go version

Your active Go toolchain is below the required version. Replace it with a current Linux release.

• Follow the official Go installation guide.
• Install Go at version 1.23 or later.
• Restart your WSL2 terminal after updating the toolchain.
• Run the Go version check above again.

ⓧ Go command not found

Go is unavailable inside your WSL2 distribution. The official Linux installer adds the toolchain to the Linux environment.

• Follow the official Go installation guide.
• Install Go at version 1.23 or later.
• Restart your WSL2 terminal after installation.
• Run the Go version check above again.

Still seeing the wrong version?

A reopened WSL2 terminal reloads your executable search path. Confirm that you installed the runtime inside WSL2 instead of only installing its Windows version.

Ask for help with the runtime that still resolves incorrectly: Help me diagnose why WSL2 is selecting the wrong Python or Go version.

Create the pinned repository

Linux tools perform best when their project files stay in the WSL2 Linux filesystem. A home-directory repository also keeps the virtual environment away from /mnt/c.

• Create the repository folders under your WSL2 home directory by running these commands:

cd ~
mkdir meeting-capture-adapter
cd meeting-capture-adapter
mkdir scripts
mkdir docs

What do these commands create?

The commands create meeting-capture-adapter inside your Linux home directory. The scripts folder holds executable project utilities.

The docs folder holds decisions and predictions. This structure stays outside the mounted Windows drive.

• Confirm the repository location by running this command:

pwd

What should the path show?

The path should end with /meeting-capture-adapter. A path without /mnt/c confirms the repository is stored in the WSL2 Linux filesystem.

Did the folder command fail?

A folder with the same name may already exist. Inspect that folder before deleting or replacing anything.

Ask for help without removing existing files: Help me safely inspect an existing meeting-capture-adapter folder in WSL2.

Exact dependency pins make every learner install the same CDK library surface. The local CDK CLI pin also prevents a global CLI update from silently changing this project.

• Create requirements.txt inside meeting-capture-adapter with this content:

aws-cdk-lib==2.273.0
constructs==10.8.1
boto3==1.43.114

What do these pins control?

• The aws-cdk-lib pin fixes the Python infrastructure library at version 2.273.0.
• The constructs pin fixes the construct programming model at version 10.8.1.
• The boto3 pin fixes the AWS SDK for Python at version 1.43.114.

• Save requirements.txt.
• Create package.json inside meeting-capture-adapter with this content:

{
  "private": true,
  "devDependencies": {
    "aws-cdk": "2.1145.0"
  }
}

Why use a local CDK dependency?

The local aws-cdk package pins the CLI at version 2.1145.0. The Makefile later invokes that local copy through npx cdk.

• Save package.json.
• Confirm both dependency manifests exist by running this command:

ls

What should be listed?

You should see requirements.txt with package.json. You should also see the scripts and docs folders.

• Create cdk.json inside meeting-capture-adapter with this content:

{
  "app": "python3 app.py"
}

What does this configuration declare?

The app setting tells CDK how to run the Python infrastructure application. A later step creates app.py.

• Save cdk.json.
• Create Makefile inside meeting-capture-adapter with this first section:

AWS_PROFILE ?= default
REGION := us-east-1
PYTHON := .venv/bin/python
CDK := npx cdk

.PHONY: setup check package test diff deploy invoke inventory destroy destroy-guardrail

setup:
	python3 -m venv .venv
	$(PYTHON) -m pip install -r requirements.txt
	npm install

check:
	$(PYTHON) scripts/check_prereqs.py --profile $(AWS_PROFILE)

package:
	rm -rf build/receiver
	mkdir -p build/receiver
	$(PYTHON) -m pip install -r src/receiver/requirements.txt --target build/receiver
	cp src/receiver/handler.py build/receiver/handler.py

test:
	PYTHONPATH=src/receiver $(PYTHON) -m unittest discover -s tests -v

What does this Makefile section do?

• The shared variables keep the AWS profile with the region in one place.
• The setup target creates .venv before installing the pinned Python and npm dependencies.
• The check target runs the prerequisite checker with the selected profile.
• The package and test targets prepare the receiver code used in later steps.

• Save the first section of Makefile.
• Confirm the new files are present by running this command:

ls

What should be listed now?

You should now see Makefile with cdk.json. The two dependency manifests remain in the same folder.

• Add the remaining targets below the test target in Makefile by copying this section:

diff: package test
	BUDGET_EMAIL=$(BUDGET_EMAIL) $(CDK) diff --profile $(AWS_PROFILE)

deploy: package test
	test -n "$(BUDGET_EMAIL)" || (echo "Set BUDGET_EMAIL" && exit 1)
	BUDGET_EMAIL=$(BUDGET_EMAIL) $(CDK) deploy MeetingCaptureGuardrailStack MeetingCaptureIngestStack --profile $(AWS_PROFILE) --require-approval never --outputs-file cdk-outputs.json

invoke:
	$(PYTHON) tools/send_synthetic.py --profile $(AWS_PROFILE)

inventory:
	-aws cloudformation list-stack-resources --stack-name MeetingCaptureIngestStack --profile $(AWS_PROFILE) --region $(REGION)
	aws sqs list-queues --profile $(AWS_PROFILE) --region $(REGION)
	aws apigatewayv2 get-apis --profile $(AWS_PROFILE) --region $(REGION)
	aws lambda list-functions --profile $(AWS_PROFILE) --region $(REGION)
	aws logs describe-log-groups --profile $(AWS_PROFILE) --region $(REGION)
	aws secretsmanager list-secrets --profile $(AWS_PROFILE) --region $(REGION)

destroy:
	BUDGET_EMAIL=$(BUDGET_EMAIL) $(CDK) destroy MeetingCaptureIngestStack --profile $(AWS_PROFILE)

destroy-guardrail:
	BUDGET_EMAIL=$(BUDGET_EMAIL) $(CDK) destroy MeetingCaptureGuardrailStack --profile $(AWS_PROFILE)

Why define later targets now?

These targets give every later module action one repeatable entry point. The current step uses only setup with check.

The deployment and cleanup targets already share the same profile variable. That consistency reduces command drift later.

• Save the completed Makefile.

Does the Makefile look misaligned?

Each command under a Make target must begin with a tab character. Spaces at the start of those command lines can prevent Make from reading the file.

Ask for help checking the indentation: Help me find incorrect command indentation in my Makefile.

Build the checker and record predictions

The checker turns a collection of tool versions into one pass or fail gate. It also verifies the temporary identity supplied by your assume-role profile through AWS Security Token Service.

• Create scripts/check_prereqs.py with this first section:

import argparse
import email.utils
import json
import re
import subprocess
import sys
import time
import urllib.request

REGION = "us-east-1"
CLOCK_URL = "https://aws.amazon.com/sqs/pricing/"


def run(command):
    result = subprocess.run(command, check=True, capture_output=True, text=True)
    return (result.stdout or result.stderr).strip()


def check_version(name, command, pattern, minimum=None):
    output = run(command)
    match = re.search(pattern, output)
    if not match:
        raise RuntimeError(f"Could not parse {name} version from: {output}")
    version = tuple(int(part) for part in match.group(1).split("."))
    if minimum and version < minimum:
        raise RuntimeError(f"{name} {version} is below required {minimum}")
    print(f"PASS {name}: {output.splitlines()[0]}")

What does this checker section do?

• The constants keep the AWS region with the HTTPS clock source explicit.
• The run() helper captures output from each local command.
• The check_version() helper extracts a numeric version before comparing it with the required minimum.
• Every successful version check prints a line beginning with PASS.

• Save scripts/check_prereqs.py.
• Confirm the checker file exists by running this command:

ls scripts

What should be listed?

You should see check_prereqs.py inside the scripts folder. This confirms the file is stored in the repository.

• Add the clock-drift function below check_version() in scripts/check_prereqs.py by copying this code:

def check_clock():
    request = urllib.request.Request(CLOCK_URL, method="HEAD")
    with urllib.request.urlopen(request, timeout=10) as response:
        remote = email.utils.parsedate_to_datetime(response.headers["Date"]).timestamp()
    drift = abs(time.time() - remote)
    if drift >= 5:
        raise RuntimeError(
            f"clock drift is {drift:.1f}s. Run 'wsl --shutdown' in PowerShell, reopen WSL2, and retry"
        )
    print(f"PASS clock drift: {drift:.1f}s")

How does the clock check work?

The function compares your WSL2 clock with the Date header returned by an HTTPS request. It stops the check when the difference reaches five seconds.

Webhook signatures include a timestamp. Detecting drift now prevents a correct signature from appearing stale later.

• Save scripts/check_prereqs.py.
• Add the main checker below check_clock() by copying this code:

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--profile", required=True)
    args = parser.parse_args()

    check_version("Python", [sys.executable, "--version"], r"Python (\d+\.\d+\.\d+)", (3, 12, 0))
    check_version("Go", ["go", "version"], r"go(\d+\.\d+(?:\.\d+)?)", (1, 23))
    check_version("AWS CLI", ["aws", "--version"], r"aws-cli/(\d+\.\d+\.\d+)", (2, 0, 0))
    check_version("Node.js", ["node", "--version"], r"v(\d+\.\d+\.\d+)", (18, 0, 0))
    check_version("CDK global", ["cdk", "--version"], r"(\d+\.\d+\.\d+)")
    check_version("CDK local", ["npx", "cdk", "--version"], r"(\d+\.\d+\.\d+)")
    for name, command in (
        ("Docker", ["docker", "--version"]),
        ("Make", ["make", "--version"]),
        ("Git", ["git", "--version"]),
    ):
        print(f"PASS {name}: {run(command).splitlines()[0]}")

    identity = json.loads(run([
        "aws", "sts", "get-caller-identity", "--profile", args.profile, "--region", REGION
    ]))
    print(f"PASS AWS identity: {identity['Arn']} in {identity['Account']} ({REGION})")
    check_clock()
    print("PASS")


if __name__ == "__main__":
    main()

What does the main checker verify?

The version checks cover Python, Go, the AWS CLI, Node.js, global CDK, local CDK, Docker, GNU Make, and Git.

The identity call proves that the supplied profile can obtain temporary credentials in us-east-1. The final clock check must also pass before the script prints its last PASS.

• Save the completed scripts/check_prereqs.py.
• Confirm the completed checker remains in the scripts folder by running this command:

ls scripts

What does this confirm?

The output should still include check_prereqs.py. The next command runs that saved file through the Make target.

Set your expectations first

Predictions give you a baseline to compare with the signature test and deployed behavior. Record your current expectation before later results influence it.

• Signature trap: whether parsing and reserializing signed JSON changes the signed bytes.
• Duplicate delivery: how many HTTP acknowledgements and FIFO messages two identical deliveries produce.
• HTTP result: which status code a valid delivery returns.
• Queue result: how many messages the receive operation retrieves.
• Teardown result: which ingest resources disappear and which guardrail remains.

• Create docs/predictions.md in your WSL-connected editor.
• Write one short prediction for each prompt above in docs/predictions.md.
• Save docs/predictions.md.
• Confirm the predictions file exists by running this command:

ls docs

What should be listed?

You should see predictions.md inside the docs folder. Your predictions now form a record you can revisit after each experiment.

Your checker needs the exact assume-role profile configured in WSL2. Capture that profile name here so later Make targets reuse it consistently.

• Record your assume-role profile name here: your-assume-role-profile.

No module resources are created

The final command creates local dependency folders only. Its AWS activity is limited to an STS identity request.

The clock check makes an HTTPS metadata request. No module AWS resources exist after this step.

The first setup can take several minutes while Python and npm download the pinned packages. A quiet terminal during a download does not mean the process has stopped.

Before you run the complete gate, do you expect every tool and session check to pass on the first attempt?

• Create .venv before installing the pinned dependencies and running every prerequisite check with this command:

make setup check AWS_PROFILE=[[AWS_PROFILE="your-assume-role-profile"]]

What does this command prove?

The setup target creates .venv before installing the versions pinned in requirements.txt and package.json.

The check target prints each tool version. It also prints the active STS identity with clock drift below five seconds.

You should see a PASS line for every required tool. The output should finish with the AWS identity in us-east-1, clock drift below five seconds, and a final PASS.

That is the environment gate cleared. Your repository now has pinned dependencies with a verified temporary AWS identity.

Did the prerequisite gate stop?

A tool failure names the version that failed or the executable that could not run. Update that tool inside WSL2 before rerunning the same Make command.

An identity failure usually means the assume-role session needs refreshing. Reauthenticate the profile before rerunning the check.

• Press the Windows key to open search if the clock check fails.
• Type PowerShell into the search field.
• Press Enter to start PowerShell.
• Shut down the WSL2 utility virtual machine by running this command:

wsl --shutdown

This command stops every running WSL2 distribution. Reopen your distribution before rerunning the Make check.

Ask for help with the failed gate: Help me diagnose the prerequisite check that stopped before PASS.

✔️ Awesome, I've got everything!

Your pinned dependencies are installed. Double check that docs/predictions.md is saved before moving on.

ⓧ I'd like to double check the full code

Compare each repository file with these cumulative references.

aws-cdk-lib==2.273.0
constructs==10.8.1
boto3==1.43.114

Requirements reference

This file contains the three exact Python dependency pins used by make setup.

{
  "private": true,
  "devDependencies": {
    "aws-cdk": "2.1145.0"
  }
}

Package reference

This file installs the pinned local CDK CLI as a development dependency.

{
  "app": "python3 app.py"
}

CDK configuration reference

This configuration defines the command that CDK uses for the infrastructure application.

AWS_PROFILE ?= default
REGION := us-east-1
PYTHON := .venv/bin/python
CDK := npx cdk

.PHONY: setup check package test diff deploy invoke inventory destroy destroy-guardrail

setup:
	python3 -m venv .venv
	$(PYTHON) -m pip install -r requirements.txt
	npm install

check:
	$(PYTHON) scripts/check_prereqs.py --profile $(AWS_PROFILE)

package:
	rm -rf build/receiver
	mkdir -p build/receiver
	$(PYTHON) -m pip install -r src/receiver/requirements.txt --target build/receiver
	cp src/receiver/handler.py build/receiver/handler.py

test:
	PYTHONPATH=src/receiver $(PYTHON) -m unittest discover -s tests -v

diff: package test
	BUDGET_EMAIL=$(BUDGET_EMAIL) $(CDK) diff --profile $(AWS_PROFILE)

deploy: package test
	test -n "$(BUDGET_EMAIL)" || (echo "Set BUDGET_EMAIL" && exit 1)
	BUDGET_EMAIL=$(BUDGET_EMAIL) $(CDK) deploy MeetingCaptureGuardrailStack MeetingCaptureIngestStack --profile $(AWS_PROFILE) --require-approval never --outputs-file cdk-outputs.json

invoke:
	$(PYTHON) tools/send_synthetic.py --profile $(AWS_PROFILE)

inventory:
	-aws cloudformation list-stack-resources --stack-name MeetingCaptureIngestStack --profile $(AWS_PROFILE) --region $(REGION)
	aws sqs list-queues --profile $(AWS_PROFILE) --region $(REGION)
	aws apigatewayv2 get-apis --profile $(AWS_PROFILE) --region $(REGION)
	aws lambda list-functions --profile $(AWS_PROFILE) --region $(REGION)
	aws logs describe-log-groups --profile $(AWS_PROFILE) --region $(REGION)
	aws secretsmanager list-secrets --profile $(AWS_PROFILE) --region $(REGION)

destroy:
	BUDGET_EMAIL=$(BUDGET_EMAIL) $(CDK) destroy MeetingCaptureIngestStack --profile $(AWS_PROFILE)

destroy-guardrail:
	BUDGET_EMAIL=$(BUDGET_EMAIL) $(CDK) destroy MeetingCaptureGuardrailStack --profile $(AWS_PROFILE)

Makefile reference

This full reference includes every target used across the module. The current step relies on setup with check.

import argparse
import email.utils
import json
import re
import subprocess
import sys
import time
import urllib.request

REGION = "us-east-1"
CLOCK_URL = "https://aws.amazon.com/sqs/pricing/"


def run(command):
    result = subprocess.run(command, check=True, capture_output=True, text=True)
    return (result.stdout or result.stderr).strip()


def check_version(name, command, pattern, minimum=None):
    output = run(command)
    match = re.search(pattern, output)
    if not match:
        raise RuntimeError(f"Could not parse {name} version from: {output}")
    version = tuple(int(part) for part in match.group(1).split("."))
    if minimum and version < minimum:
        raise RuntimeError(f"{name} {version} is below required {minimum}")
    print(f"PASS {name}: {output.splitlines()[0]}")


def check_clock():
    request = urllib.request.Request(CLOCK_URL, method="HEAD")
    with urllib.request.urlopen(request, timeout=10) as response:
        remote = email.utils.parsedate_to_datetime(response.headers["Date"]).timestamp()
    drift = abs(time.time() - remote)
    if drift >= 5:
        raise RuntimeError(
            f"clock drift is {drift:.1f}s. Run 'wsl --shutdown' in PowerShell, reopen WSL2, and retry"
        )
    print(f"PASS clock drift: {drift:.1f}s")


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--profile", required=True)
    args = parser.parse_args()

    check_version("Python", [sys.executable, "--version"], r"Python (\d+\.\d+\.\d+)", (3, 12, 0))
    check_version("Go", ["go", "version"], r"go(\d+\.\d+(?:\.\d+)?)", (1, 23))
    check_version("AWS CLI", ["aws", "--version"], r"aws-cli/(\d+\.\d+\.\d+)", (2, 0, 0))
    check_version("Node.js", ["node", "--version"], r"v(\d+\.\d+\.\d+)", (18, 0, 0))
    check_version("CDK global", ["cdk", "--version"], r"(\d+\.\d+\.\d+)")
    check_version("CDK local", ["npx", "cdk", "--version"], r"(\d+\.\d+\.\d+)")
    for name, command in (
        ("Docker", ["docker", "--version"]),
        ("Make", ["make", "--version"]),
        ("Git", ["git", "--version"]),
    ):
        print(f"PASS {name}: {run(command).splitlines()[0]}")

    identity = json.loads(run([
        "aws", "sts", "get-caller-identity", "--profile", args.profile, "--region", REGION
    ]))
    print(f"PASS AWS identity: {identity['Arn']} in {identity['Account']} ({REGION})")
    check_clock()
    print("PASS")


if __name__ == "__main__":
    main()

Prerequisite checker reference

This full reference verifies every required tool before checking the STS identity and HTTPS clock drift.

Your docs/predictions.md file should cover these five outcomes:

• Parsing and reserializing a signed JSON delivery changes its signed bytes.
• Two identical signed deliveries should receive two HTTP acknowledgements but yield one FIFO message during SQS deduplication.
• A valid delivery should return 202.
• The receive operation should retrieve one message.
• Teardown should remove ingest resources while retaining the budget guardrail.

Your toolchain is pinned with the clock and temporary AWS identity verified. Next, you will define the event contract before triggering the planned raw-signature mismatch.

Define the Contract and Trigger the Signature Trap

Your validated WSL2 toolchain gives you a reliable foundation for the webhook admission boundary. The next risk lives inside the payload itself.

Webhook signatures protect exact byte sequences. JSON can preserve the same values while changing whitespace.

The contract fixes the envelope shape. A focused local test then exposes what reserialization does to signed bytes.

In this step, get ready to:

• Define the versioned ingress envelope.
• Record the two architecture decisions.
• Trigger the JSON signature trap with synthetic data.

This Step Stays Local

Every check in this step runs inside your validated repository. No module AWS resources are created.

Define the ingress envelope

A versioned JSON Schema records the fields every admitted event must carry. This gives future producers plus consumers one shared contract.

• Create schemas/ingress-envelope.schema.json inside your repository by adding the following code:

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "$id": "urn:meeting-capture-adapter:ingress-envelope:1.0",
  "title": "Meeting Capture Ingress Envelope 1.0",
  "type": "object",
  "additionalProperties": false,
  "required": [
    "schema_version",
    "source",
    "tenant_key",
    "event_type",
    "event_ts",
    "received_at",
    "meeting_key",
    "stream_id",
    "source_event"
  ],
  "properties": {
    "schema_version": {"const": "1.0"},
    "source": {"const": "zoom"},
    "tenant_key": {"type": "string", "minLength": 1},
    "event_type": {"type": "string", "minLength": 1},
    "event_ts": {"type": "integer"},
    "received_at": {"type": "string", "format": "date-time"},
    "meeting_key": {"type": "string", "minLength": 1},
    "stream_id": {"type": "string", "minLength": 1},
    "source_event": {"type": "object"}
  }
}

What Does This Schema Define?

• The required array makes every envelope field mandatory.
• The schema_version value fixes this contract at 1.0.
• The source value identifies synthetic Zoom-compatible input.
• The source_event object preserves the original event for downstream processing.
• The additionalProperties setting prevents undocumented top-level fields from slipping into the envelope.

• Save schemas/ingress-envelope.schema.json.
• Confirm the file appears inside the schemas folder in your editor's file explorer.

You should see one schema file with nine required envelope fields. The title should identify version 1.0.

Does the schema show a syntax problem?

Check the commas after each property. Confirm that every opening brace has a matching closing brace.

Use this prompt for a focused comparison: Help me find the JSON syntax difference in my ingress envelope schema.

✔️ Awesome, I've got everything!

Your versioned ingress envelope is saved with every required field.

ⓧ I'd like to double check the full code

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "$id": "urn:meeting-capture-adapter:ingress-envelope:1.0",
  "title": "Meeting Capture Ingress Envelope 1.0",
  "type": "object",
  "additionalProperties": false,
  "required": [
    "schema_version",
    "source",
    "tenant_key",
    "event_type",
    "event_ts",
    "received_at",
    "meeting_key",
    "stream_id",
    "source_event"
  ],
  "properties": {
    "schema_version": {"const": "1.0"},
    "source": {"const": "zoom"},
    "tenant_key": {"type": "string", "minLength": 1},
    "event_type": {"type": "string", "minLength": 1},
    "event_ts": {"type": "integer"},
    "received_at": {"type": "string", "format": "date-time"},
    "meeting_key": {"type": "string", "minLength": 1},
    "stream_id": {"type": "string", "minLength": 1},
    "source_event": {"type": "object"}
  }
}

Record the architecture decisions

An architecture decision record captures a choice plus the conditions that could change it. This keeps today's boundary understandable when the worker arrives later.

What Should the ADRs Capture?

ADR-001 chooses Python for AWS Lambda admission. It reserves Go for future capture workers.

ADR-001 compares all-Python with all-Go. Its reversibility conditions depend on worker requirements plus operational constraints.

ADR-002 chooses Amazon API Gateway HTTP API for webhook admission. It compares this choice with a REST API plus a Lambda Function URL.

ADR-002 can be revisited when authentication requirements change. Routing requirements or observability requirements can also justify a new decision.

• Create docs/ADR-001-python-admission-go-workers.md in your editor using the first decision record described above.
• Save docs/ADR-001-python-admission-go-workers.md.

You should see ADR-001 under the docs folder. Its decision should name Python admission plus future Go workers.

• Create docs/ADR-002-http-api.md in your editor using the second decision record described above.
• Save docs/ADR-002-http-api.md.

You should now see both ADR files under docs. Each record should contain a decision plus alternatives plus reversibility conditions.

Trigger the signature trap

An HMAC turns request bytes into a digest with a shared secret. Verification succeeds only when the receiver reproduces the same byte sequence.

The local experiment signs a compact synthetic fixture. It then parses that fixture before producing a whitespace-formatted copy.

• Create src/receiver/handler.py inside your repository by adding this local signature verifier:

import hashlib
import hmac
import json
import os
import time


def verify_signature(body, timestamp, signature, secret, now=None):
    if not timestamp or not signature:
        return False
    try:
        timestamp_value = int(timestamp)
    except ValueError:
        return False
    current = int(time.time() if now is None else now)
    maximum_age = int(os.environ.get("MAX_SIGNATURE_AGE_SECONDS", "300"))
    if abs(current - timestamp_value) > maximum_age:
        return False
    message = b"v0:" + timestamp.encode("utf-8") + b":" + body
    digest = hmac.new(secret.encode("utf-8"), message, hashlib.sha256).hexdigest()
    return hmac.compare_digest(f"v0={digest}", signature)

What Does This Verifier Do?

• The verifier rejects requests without a timestamp or signature.
• The timestamp check applies the adapter's 300 second freshness policy.
• The signed message combines the version prefix with the timestamp plus the supplied body bytes.
• The constant-time comparison checks the calculated digest against the supplied signature.

• Save src/receiver/handler.py.
• Confirm your editor's outline lists verify_signature.

The receiver module now exposes one signature verifier. The focused test can feed either original bytes or reconstructed bytes into it.

Can't find the verifier from your test?

Confirm that handler.py sits inside src/receiver. Check that the function name is exactly verify_signature.

Use this prompt to inspect the file path plus import: Help me diagnose why tests/test_handler.py cannot import src/receiver/handler.py.

✔️ Awesome, I've got everything!

Your local verifier is saved in the receiver module.

ⓧ I'd like to double check the full code

import hashlib
import hmac
import json
import os
import time


def verify_signature(body, timestamp, signature, secret, now=None):
    if not timestamp or not signature:
        return False
    try:
        timestamp_value = int(timestamp)
    except ValueError:
        return False
    current = int(time.time() if now is None else now)
    maximum_age = int(os.environ.get("MAX_SIGNATURE_AGE_SECONDS", "300"))
    if abs(current - timestamp_value) > maximum_age:
        return False
    message = b"v0:" + timestamp.encode("utf-8") + b":" + body
    digest = hmac.new(secret.encode("utf-8"), message, hashlib.sha256).hexdigest()
    return hmac.compare_digest(f"v0={digest}", signature)

The test needs a stable synthetic event plus a helper that signs its compact representation. These helpers keep the experiment repeatable without using live meeting data.

• Create tests/test_handler.py inside your repository by adding the synthetic fixture foundation below:

import hashlib
import hmac
import json
import sys
import time
import unittest
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parents[1] / "src" / "receiver"))
import handler

SECRET = "synthetic-test-secret"


def body_text():
    return json.dumps({
        "event": "meeting.rtms_started",
        "event_ts": 1760486400000,
        "payload": {
            "account_id": "synthetic-account",
            "meeting_uuid": "synthetic-meeting",
            "rtms_stream_id": "synthetic-stream",
            "server_urls": "synthetic-only"
        }
    }, separators=(",", ":"))

What Does This Fixture Represent?

• The path insertion lets the test import handler.py from src/receiver.
• The shared secret is synthetic test data.
• The fixture follows the RTMS-shaped event fields used by this module.
• The compact separators remove optional whitespace from the fixture.

• Save tests/test_handler.py.
• Confirm your editor's outline lists body_text.

Your test file now produces one deterministic compact body. Its event values remain synthetic.

Does the fixture show a syntax problem?

Check the closing braces around payload. Confirm that the final json.dumps call includes its closing parenthesis.

Use this prompt for a fixture-only review: Help me find the syntax issue in my synthetic body_text fixture.

• Add the signature helper below body_text in tests/test_handler.py by copying this code:

def signature(body, timestamp):
    message = f"v0:{timestamp}:{body}".encode()
    return "v0=" + hmac.new(SECRET.encode(), message, hashlib.sha256).hexdigest()

What Does the Signature Helper Do?

The helper signs the compact body with the synthetic secret. The timestamp becomes part of the signed message.

• Save tests/test_handler.py.
• Confirm your editor's outline now lists signature.

The test module can now generate a signature tied to one exact string representation.

Is the signature helper missing from the outline?

Check that signature starts at the left edge of the file. Confirm that the helper sits below the complete body_text function.

Use this prompt to check the function structure: Help me fix the placement of my signature helper in tests/test_handler.py.

The final test deliberately reconstructs the compact fixture with indentation. It asks the verifier to compare that reconstructed body against the signature created from the compact body.

• Add the focused test below signature in tests/test_handler.py by copying this code:

class HandlerTests(unittest.TestCase):
    def test_reserialization_is_not_signature_safe(self):
        compact = body_text()
        timestamp = str(int(time.time()))
        pretty = json.dumps(json.loads(compact), indent=2)
        self.assertFalse(handler.verify_signature(
            pretty.encode(), timestamp, signature(compact, timestamp), SECRET, now=int(timestamp)
        ))

What Does This Test Compare?

• The compact value holds the source string that gets signed.
• The pretty value holds the parsed plus indented reconstruction.
• The assertion checks whether the original signature validates against the reconstructed bytes.

• Save tests/test_handler.py.

Before you run the focused test, do you expect whitespace-formatted JSON to preserve the signature created from compact JSON?

• Run the local test suite from your repository by running:

make test

What Did the Test Prove?

The focused test passes because the verifier rejects the reconstructed body. The JSON values stayed equivalent while the signed bytes changed.

This is the intended shortfall. A receiver that parses before verification loses the byte sequence authenticated by the sender.

You have caught the exact byte-level trap before deploying anything. The repository now contains a reproducible test that protects the lesson.

Does the focused test fail to run?

Confirm that your virtual environment still contains the pinned dependencies from Step 1. Check that handler.py lives under src/receiver.

Compare the function names in both files. The test must call handler.verify_signature.

Use this prompt with your terminal output: Help me diagnose why the focused signature reserialization test will not run.

✔️ Awesome, I've got everything!

Your focused test passes while proving that reserialized JSON fails signature verification.

ⓧ I'd like to double check the full code

import hashlib
import hmac
import json
import sys
import time
import unittest
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parents[1] / "src" / "receiver"))
import handler

SECRET = "synthetic-test-secret"


def body_text():
    return json.dumps({
        "event": "meeting.rtms_started",
        "event_ts": 1760486400000,
        "payload": {
            "account_id": "synthetic-account",
            "meeting_uuid": "synthetic-meeting",
            "rtms_stream_id": "synthetic-stream",
            "server_urls": "synthetic-only"
        }
    }, separators=(",", ":"))


def signature(body, timestamp):
    message = f"v0:{timestamp}:{body}".encode()
    return "v0=" + hmac.new(SECRET.encode(), message, hashlib.sha256).hexdigest()


class HandlerTests(unittest.TestCase):
    def test_reserialization_is_not_signature_safe(self):
        compact = body_text()
        timestamp = str(int(time.time()))
        pretty = json.dumps(json.loads(compact), indent=2)
        self.assertFalse(handler.verify_signature(
            pretty.encode(), timestamp, signature(compact, timestamp), SECRET, now=int(timestamp)
        ))

Your contract plus decision records now define the admission boundary. Next up, you'll verify the untouched request body before parsing or side effects.

Verify Raw Bytes and Build the Receiver

Your previous test proved that equivalent JSON can produce different signed bytes. The deliberately naive receiver exposed that gap before any cloud resources existed.

The fix belongs at the trust boundary. Your AWS Lambda receiver must verify the exact body from the API Gateway HTTP API before parsing events or triggering side effects.

In this step, get ready to:

• Extract raw UTF-8 or base64-decoded request bytes.
• Verify signatures before normalizing events for the FIFO queue.
• Prove the receiver behavior with synthetic tests and fake AWS clients.

Verify the exact request bytes

The receiver needs raw bytes for HMAC-SHA256 verification. It also needs cached clients for Amazon SQS and AWS Secrets Manager so warm invocations can reuse them.

• In src/receiver/handler.py, remove the deliberately naive verification experiment.
• Add the imports and cached client helper by pasting this code:

import base64
import hashlib
import hmac
import json
import os
import time
from datetime import datetime, timezone

import boto3

_sqs_client = None
_secrets_client = None
_secret_cache = None


def _client(service_name):
    global _sqs_client, _secrets_client
    if service_name == "sqs":
        if _sqs_client is None:
            _sqs_client = boto3.client("sqs")
        return _sqs_client
    if _secrets_client is None:
        _secrets_client = boto3.client("secretsmanager")
    return _secrets_client

What does this code do?

• The standard-library imports handle byte decoding, hashing, signatures, JSON parsing, environment values, timestamps, and UTC receive times.
• Boto3 creates the service clients used by the receiver.
• The module-level client variables survive warm Lambda invocations. This avoids rebuilding the same clients for every accepted webhook.

• Save src/receiver/handler.py.
• Check your editor diagnostics. This section should show no syntax errors.

Seeing import or syntax problems?

Check that every import sits at the top of src/receiver/handler.py. Confirm that the body of _client uses four spaces of indentation.

Ask for help with the client scaffold:

The webhook secret is fetched only when the cache is empty. Later invocations reuse the cached value instead of making another Secrets Manager request.

• Add the secret cache helper below _client by pasting this code:

def _secret():
    global _secret_cache
    if _secret_cache is None:
        response = _client("secretsmanager").get_secret_value(
            SecretId=os.environ["WEBHOOK_SECRET_ARN"]
        )
        _secret_cache = response["SecretString"]
    return _secret_cache

How does the secret cache work?

• The first call reads the secret identified by WEBHOOK_SECRET_ARN.
• The helper stores the returned SecretString in memory.
• Later calls return the cached string without logging or exposing it.

• Save src/receiver/handler.py.
• Check that _secret appears at the left edge of the file with its body indented beneath it.

Is the cache helper highlighted as invalid?

Make sure _secret begins after the complete _client function. Check that the environment key matches WEBHOOK_SECRET_ARN exactly.

Ask for help with the cache helper:

API Gateway responses need a consistent shape. Incoming bodies also need one extraction path that produces bytes for both plain and base64-encoded requests.

• Add the response and raw-body helpers below _secret by pasting this code:

def _response(status_code, payload):
    return {
        "statusCode": status_code,
        "headers": {"content-type": "application/json"},
        "body": json.dumps(payload, separators=(",", ":")),
        "isBase64Encoded": False,
    }


def raw_body(event):
    body = event.get("body") or ""
    if event.get("isBase64Encoded", False):
        return base64.b64decode(body)
    return body.encode("utf-8")

Why keep the body as bytes?

• The response helper produces compact JSON for every HTTP result.
• The raw-body helper decodes base64 when API Gateway marks the body as encoded.
• Plain text bodies become UTF-8 bytes without parsing or reserialization. Those exact bytes remain available for signature verification.

• Save src/receiver/handler.py.
• Check your editor diagnostics. Both helpers should show no syntax errors.

Is raw-body extraction failing?

Check that base64.b64decode receives the original body. Confirm that the plain path ends with body.encode("utf-8").

Ask for help with body extraction:

Signature verification combines the timestamp with the untouched body bytes. The replay check rejects requests outside the adapter-owned five-minute window before any queue operation occurs.

• Add the signature verifier below raw_body by pasting this code:

def verify_signature(body, timestamp, signature, secret, now=None):
    if not timestamp or not signature:
        return False
    try:
        timestamp_value = int(timestamp)
    except ValueError:
        return False
    current = int(time.time() if now is None else now)
    maximum_age = int(os.environ.get("MAX_SIGNATURE_AGE_SECONDS", "300"))
    if abs(current - timestamp_value) > maximum_age:
        return False
    message = b"v0:" + timestamp.encode("utf-8") + b":" + body
    digest = hmac.new(secret.encode("utf-8"), message, hashlib.sha256).hexdigest()
    return hmac.compare_digest(f"v0={digest}", signature)

What does the verifier protect?

• Missing or malformed signature metadata fails verification immediately.
• MAX_SIGNATURE_AGE_SECONDS defaults to 300 seconds. This is the adapter's replay policy.
• The signed message preserves the required v0:<timestamp>:<body> structure.
• hmac.compare_digest performs the final constant-time comparison.

• Save src/receiver/handler.py.
• Check that your editor reports no syntax errors in verify_signature.

Does the verifier reject every request?

Check the byte separators around the timestamp. Confirm that the expected signature begins with v0= before comparison.

Ask for help with signature verification:

Normalize and queue verified events

A verified RTMS event still needs a stable envelope before it can enter the queue. The envelope carries the contract fields defined in schemas/ingress-envelope.schema.json.

• Add the normalization helper below verify_signature by pasting this code:

def normalize(document):
    payload = document["payload"]
    return {
        "schema_version": "1.0",
        "source": "zoom",
        "tenant_key": payload["account_id"],
        "event_type": document["event"],
        "event_ts": document["event_ts"],
        "received_at": datetime.now(timezone.utc).isoformat(),
        "meeting_key": payload["meeting_uuid"],
        "stream_id": payload["rtms_stream_id"],
        "source_event": document,
    }

How does normalization help?

• The normalized object follows ingress envelope version 1.0.
• account_id becomes the tenant key used by later modules.
• meeting_uuid becomes the meeting key used to preserve per-meeting order.
• The original synthetic event remains available under source_event.

• Save src/receiver/handler.py.
• Compare the returned keys with schemas/ingress-envelope.schema.json. Every required property should have a matching value.

Are envelope fields missing?

Check that each source lookup matches the synthetic event shape. Confirm that account_id, meeting_uuid, and rtms_stream_id come from payload.

Ask for help with normalization:

The FIFO queue needs an explicit deduplication identifier. A fingerprint built from stable event identity fields gives duplicate deliveries the same value.

• Add the event fingerprint helper below normalize by pasting this code:

def event_fingerprint(envelope):
    fields = (
        envelope["tenant_key"],
        envelope["event_type"],
        envelope["meeting_key"],
        envelope["stream_id"],
        str(envelope["event_ts"]),
    )
    return hashlib.sha256("|".join(fields).encode("utf-8")).hexdigest()

Why fingerprint these fields?

The fingerprint combines the tenant, event type, meeting, stream, and event timestamp. Repeated delivery of the same logical event produces the same SHA-256 value.

• Save src/receiver/handler.py.
• Check that event_fingerprint returns the hexadecimal digest from the joined fields.

Is the fingerprint changing unexpectedly?

Check the field order inside the tuple. Confirm that the separator is | and that event_ts becomes a string before joining.

Ask for help with fingerprint stability:

The handler now needs to enforce the trust boundary. URL validation follows its own challenge-response path before signed RTMS events continue to normalization.

• Start lambda_handler below event_fingerprint by pasting this code:

def lambda_handler(event, context):
    del context
    body = raw_body(event)
    try:
        document = json.loads(body)
    except (json.JSONDecodeError, UnicodeDecodeError):
        return _response(400, {"error": "invalid_json"})

    if document.get("event") == "endpoint.url_validation":
        plain_token = document.get("payload", {}).get("plainToken")
        if not plain_token:
            return _response(400, {"error": "missing_plain_token"})
        encrypted = hmac.new(
            _secret().encode("utf-8"),
            plain_token.encode("utf-8"),
            hashlib.sha256,
        ).hexdigest()
        return _response(200, {"plainToken": plain_token, "encryptedToken": encrypted})

What happens at the handler entrance?

• The handler extracts the body before parsing it.
• Malformed JSON receives a 400 response without reaching the queue.
• The endpoint.url_validation path returns the plain token with its encrypted token.
• The challenge response uses the cached webhook secret without printing it.

• Save src/receiver/handler.py.
• Check your editor diagnostics. The handler should remain syntactically valid at this point.

Is the validation branch malformed?

Check that plainToken comes from the nested payload. Confirm that the HMAC uses the secret as its key.

Ask for help with endpoint validation:

Every other event must pass signature verification before normalization. Lowercasing the incoming header names gives the verifier one consistent lookup path.

• Continue lambda_handler directly below the endpoint validation response by pasting this code:

    headers = {key.lower(): value for key, value in event.get("headers", {}).items()}
    if not verify_signature(
        body,
        headers.get("x-zm-request-timestamp", ""),
        headers.get("x-zm-signature", ""),
        _secret(),
    ):
        return _response(401, {"error": "invalid_signature"})

    try:
        envelope = normalize(document)
    except (KeyError, TypeError):
        return _response(400, {"error": "invalid_rtms_event"})

Where is the trust boundary?

• The verifier receives the original body bytes with x-zm-request-timestamp and x-zm-signature.
• Failed verification returns 401 before normalization or queue access.
• Missing RTMS fields return 400 without producing a message.

• Save src/receiver/handler.py.
• Check that this continuation remains indented inside lambda_handler.

Is the continuation outside the handler?

Every line in this snippet begins with four spaces because it belongs inside lambda_handler. Keep the nested verifier arguments indented one level further.

Ask for help with the trust boundary:

Only a verified envelope reaches the queue send. The log records identifiers needed for operations without recording the source payload or webhook secret.

• Finish lambda_handler directly below the normalization error response by pasting this code:

    deduplication_id = event_fingerprint(envelope)
    message_body = json.dumps(envelope, separators=(",", ":"), sort_keys=True)
    _client("sqs").send_message(
        QueueUrl=os.environ["QUEUE_URL"],
        MessageBody=message_body,
        MessageGroupId=envelope["meeting_key"],
        MessageDeduplicationId=deduplication_id,
    )
    print(json.dumps({
        "status": "accepted",
        "event_type": envelope["event_type"],
        "tenant_key": envelope["tenant_key"],
        "meeting_key": envelope["meeting_key"],
        "deduplication_id": deduplication_id,
    }, separators=(",", ":")))
    return _response(202, {"status": "accepted", "deduplication_id": deduplication_id})

How does the FIFO send work?

• QUEUE_URL identifies the destination queue.
• MessageGroupId uses meeting_key so events from one meeting retain their order.
• MessageDeduplicationId uses the stable event fingerprint.
• The structured log contains event identifiers without the payload or secret.
• A successful queue send returns 202 with the deduplication identifier.

• Save src/receiver/handler.py.
• Check your editor diagnostics. The complete file should show no syntax errors.

Is the queue section reporting an indentation error?

Keep the entire queue send, structured log, and final response inside lambda_handler. Confirm that each dictionary closes before the next statement begins.

Ask for help with the queue section:

The receiver now fails closed

Invalid input cannot reach normalization, logs, or Amazon SQS. The receiver only reads its secret and sends verified envelopes, which defines the permissions the deployment must grant.

✔️ Awesome, I've got everything!

Your receiver now verifies raw bytes before it normalizes or queues an event.

ⓧ I'd like to double check the full code

import base64
import hashlib
import hmac
import json
import os
import time
from datetime import datetime, timezone

import boto3

_sqs_client = None
_secrets_client = None
_secret_cache = None


def _client(service_name):
    global _sqs_client, _secrets_client
    if service_name == "sqs":
        if _sqs_client is None:
            _sqs_client = boto3.client("sqs")
        return _sqs_client
    if _secrets_client is None:
        _secrets_client = boto3.client("secretsmanager")
    return _secrets_client


def _secret():
    global _secret_cache
    if _secret_cache is None:
        response = _client("secretsmanager").get_secret_value(
            SecretId=os.environ["WEBHOOK_SECRET_ARN"]
        )
        _secret_cache = response["SecretString"]
    return _secret_cache


def _response(status_code, payload):
    return {
        "statusCode": status_code,
        "headers": {"content-type": "application/json"},
        "body": json.dumps(payload, separators=(",", ":")),
        "isBase64Encoded": False,
    }


def raw_body(event):
    body = event.get("body") or ""
    if event.get("isBase64Encoded", False):
        return base64.b64decode(body)
    return body.encode("utf-8")


def verify_signature(body, timestamp, signature, secret, now=None):
    if not timestamp or not signature:
        return False
    try:
        timestamp_value = int(timestamp)
    except ValueError:
        return False
    current = int(time.time() if now is None else now)
    maximum_age = int(os.environ.get("MAX_SIGNATURE_AGE_SECONDS", "300"))
    if abs(current - timestamp_value) > maximum_age:
        return False
    message = b"v0:" + timestamp.encode("utf-8") + b":" + body
    digest = hmac.new(secret.encode("utf-8"), message, hashlib.sha256).hexdigest()
    return hmac.compare_digest(f"v0={digest}", signature)


def normalize(document):
    payload = document["payload"]
    return {
        "schema_version": "1.0",
        "source": "zoom",
        "tenant_key": payload["account_id"],
        "event_type": document["event"],
        "event_ts": document["event_ts"],
        "received_at": datetime.now(timezone.utc).isoformat(),
        "meeting_key": payload["meeting_uuid"],
        "stream_id": payload["rtms_stream_id"],
        "source_event": document,
    }


def event_fingerprint(envelope):
    fields = (
        envelope["tenant_key"],
        envelope["event_type"],
        envelope["meeting_key"],
        envelope["stream_id"],
        str(envelope["event_ts"]),
    )
    return hashlib.sha256("|".join(fields).encode("utf-8")).hexdigest()


def lambda_handler(event, context):
    del context
    body = raw_body(event)
    try:
        document = json.loads(body)
    except (json.JSONDecodeError, UnicodeDecodeError):
        return _response(400, {"error": "invalid_json"})

    if document.get("event") == "endpoint.url_validation":
        plain_token = document.get("payload", {}).get("plainToken")
        if not plain_token:
            return _response(400, {"error": "missing_plain_token"})
        encrypted = hmac.new(
            _secret().encode("utf-8"),
            plain_token.encode("utf-8"),
            hashlib.sha256,
        ).hexdigest()
        return _response(200, {"plainToken": plain_token, "encryptedToken": encrypted})

    headers = {key.lower(): value for key, value in event.get("headers", {}).items()}
    if not verify_signature(
        body,
        headers.get("x-zm-request-timestamp", ""),
        headers.get("x-zm-signature", ""),
        _secret(),
    ):
        return _response(401, {"error": "invalid_signature"})

    try:
        envelope = normalize(document)
    except (KeyError, TypeError):
        return _response(400, {"error": "invalid_rtms_event"})

    deduplication_id = event_fingerprint(envelope)
    message_body = json.dumps(envelope, separators=(",", ":"), sort_keys=True)
    _client("sqs").send_message(
        QueueUrl=os.environ["QUEUE_URL"],
        MessageBody=message_body,
        MessageGroupId=envelope["meeting_key"],
        MessageDeduplicationId=deduplication_id,
    )
    print(json.dumps({
        "status": "accepted",
        "event_type": envelope["event_type"],
        "tenant_key": envelope["tenant_key"],
        "meeting_key": envelope["meeting_key"],
        "deduplication_id": deduplication_id,
    }, separators=(",", ":")))
    return _response(202, {"status": "accepted", "deduplication_id": deduplication_id})

How to compare this file

This reference shows the complete receiver after every chunk above. Function order and identifier spelling should match your saved file.

Prove the receiver with synthetic tests

Local fakes let the tests inspect secret retrieval and queue parameters without creating AWS resources. The test secret and RTMS event remain synthetic.

• Create src/receiver/requirements.txt inside the existing src/receiver folder.
• Pin the receiver dependency by pasting this line:

boto3==1.43.114

Why pin the receiver dependency?

The deployment package receives the same Boto3 version used by the repository. The exact pin keeps local tests and packaged code aligned.

• Save src/receiver/requirements.txt.
• Confirm that the file contains exactly one dependency line.
• Create the meeting_capture_adapter folder at the top level of the repository.
• Create an empty meeting_capture_adapter/__init__.py file inside that folder.
• Confirm that meeting_capture_adapter/__init__.py appears in your editor's file sidebar.

Is the receiver requirement in the wrong folder?

Check that the dependency file sits beside src/receiver/handler.py. Keep the top-level requirements.txt unchanged.

Ask for help with the package layout:

The test scaffold replaces live AWS clients with in-memory fakes. This makes each assertion repeatable without credentials, queues, or secrets.

• In tests/test_handler.py, replace the previous focused experiment with this test scaffold:

import base64
import hashlib
import hmac
import json
import os
import sys
import time
import unittest
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parents[1] / "src" / "receiver"))
import handler

SECRET = "synthetic-test-secret"


class FakeSecrets:
    def get_secret_value(self, **kwargs):
        self.request = kwargs
        return {"SecretString": SECRET}


class FakeSqs:
    def __init__(self):
        self.messages = []

    def send_message(self, **kwargs):
        self.messages.append(kwargs)
        return {"MessageId": "synthetic-message", "SequenceNumber": "1"}

What do the fake clients capture?

• FakeSecrets returns the synthetic secret and records the request arguments.
• FakeSqs stores every message request in memory.
• The tests can inspect FIFO fields without contacting AWS.

• Save tests/test_handler.py.
• Check your editor diagnostics. The imports and fake classes should show no syntax errors.

Can the tests import the receiver?

Check that the inserted path points to src/receiver from the repository's tests folder. Confirm that import handler comes after the path insertion.

Ask for help with the test import:

The fixture preserves one compact byte representation for signing. Its fields match the synthetic RTMS event that the receiver normalizes.

• Add the synthetic body and signature helpers below FakeSqs by pasting this code:

def body_text():
    return json.dumps({
        "event": "meeting.rtms_started",
        "event_ts": 1760486400000,
        "payload": {
            "account_id": "synthetic-account",
            "meeting_uuid": "synthetic-meeting",
            "rtms_stream_id": "synthetic-stream",
            "server_urls": "synthetic-only"
        }
    }, separators=(",", ":"))


def signature(body, timestamp):
    message = f"v0:{timestamp}:{body}".encode()
    return "v0=" + hmac.new(SECRET.encode(), message, hashlib.sha256).hexdigest()

Why sign one compact fixture?

The signature helper hashes the exact compact string returned by body_text. A test can later change the whitespace while preserving the same JSON values.

• Save tests/test_handler.py.
• Check that body_text closes before signature begins.

Is the fixture producing invalid JSON?

Check every brace around the payload. Confirm that separators=(",", ":") sits in the json.dumps call.

Ask for help with the fixture:

Each test resets the environment and fake clients. This prevents cached state from one test changing another test's result.

• Add the test class setup and signed-event builder below signature by pasting this code:

class HandlerTests(unittest.TestCase):
    def setUp(self):
        os.environ["QUEUE_URL"] = "synthetic-queue-url"
        os.environ["WEBHOOK_SECRET_ARN"] = "synthetic-secret-arn"
        os.environ["MAX_SIGNATURE_AGE_SECONDS"] = "300"
        handler._secret_cache = None
        handler._secrets_client = FakeSecrets()
        handler._sqs_client = FakeSqs()

    def signed_event(self, body=None, encoded=False):
        body = body or body_text()
        timestamp = str(int(time.time()))
        event_body = base64.b64encode(body.encode()).decode() if encoded else body
        return {
            "version": "2.0",
            "headers": {
                "x-zm-request-timestamp": timestamp,
                "x-zm-signature": signature(body, timestamp),
            },
            "body": event_body,
            "isBase64Encoded": encoded,
        }

How does each test stay isolated?

• setUp supplies synthetic environment values before every test.
• The cached secret resets so each test starts from the same state.
• signed_event can return a plain or base64-encoded API Gateway event.

• Save tests/test_handler.py.
• Check that both methods remain indented inside HandlerTests.

Are the test methods outside the class?

Keep setUp and signed_event indented four spaces beneath HandlerTests. Their method bodies need one additional indentation level.

Ask for help with the test class:

The first tests exercise the successful path, the base64 path, and the rejection path. The fake queue exposes exactly what the handler attempted to send.

• Add these three methods inside HandlerTests below signed_event by pasting this code:

    def test_accepts_signed_event_and_sets_fifo_fields(self):
        response = handler.lambda_handler(self.signed_event(), None)
        self.assertEqual(202, response["statusCode"])
        queued = handler._sqs_client.messages[0]
        self.assertEqual("synthetic-meeting", queued["MessageGroupId"])
        self.assertEqual(64, len(queued["MessageDeduplicationId"]))

    def test_accepts_base64_body(self):
        response = handler.lambda_handler(self.signed_event(encoded=True), None)
        self.assertEqual(202, response["statusCode"])

    def test_rejects_invalid_signature(self):
        event = self.signed_event()
        event["headers"]["x-zm-signature"] = "v0=invalid"
        self.assertEqual(401, handler.lambda_handler(event, None)["statusCode"])

What do these tests prove?

• A valid signed event returns 202 after normalization and queueing.
• The FIFO group uses synthetic-meeting and the deduplication identifier has a SHA-256 digest length.
• A base64-encoded request reaches the same accepted path.
• An invalid signature returns 401 without entering the accepted path.

• Save tests/test_handler.py.
• Check that all three test methods align with setUp inside the class.

Are the first receiver tests failing?

Check that handler._sqs_client points to FakeSqs in setUp. Confirm that the valid event's signature is calculated from the same body string.

Ask for help with the receiver tests:

• Run the current receiver tests by running this command:

make test

What does this check prove?

The test target runs the receiver suite inside the repository's virtual environment. At this point, the three completed receiver tests should pass.

You should see three tests pass. The accepted-event log should contain identifiers without the source payload or synthetic secret.

Did one of the first three tests fail?

Read the failing test name first. Compare that method with the matching receiver path before changing unrelated helpers.

Ask for help with the failing test:

Two final tests preserve the lesson from the signature trap and prove the URL validation response. Together they cover both webhook admission paths.

• Add the remaining methods inside HandlerTests below the invalid-signature test.
• Add the test runner after the class by pasting this code:

    def test_reserialization_is_not_signature_safe(self):
        compact = body_text()
        timestamp = str(int(time.time()))
        pretty = json.dumps(json.loads(compact), indent=2)
        self.assertFalse(handler.verify_signature(
            pretty.encode(), timestamp, signature(compact, timestamp), SECRET, now=int(timestamp)
        ))

    def test_endpoint_validation(self):
        body = json.dumps({
            "event": "endpoint.url_validation",
            "event_ts": 1760486400000,
            "payload": {"plainToken": "synthetic-token"},
        }, separators=(",", ":"))
        response = handler.lambda_handler({"body": body, "headers": {}, "isBase64Encoded": False}, None)
        payload = json.loads(response["body"])
        expected = hmac.new(SECRET.encode(), b"synthetic-token", hashlib.sha256).hexdigest()
        self.assertEqual({"plainToken": "synthetic-token", "encryptedToken": expected}, payload)


if __name__ == "__main__":
    unittest.main()

What do the final tests protect?

• The reserialization test signs the compact body before verifying a pretty-printed version.
• The false result preserves evidence that signature checks depend on bytes.
• The endpoint validation test calculates the expected encrypted token independently.
• The final assertion checks both response fields.

• Save tests/test_handler.py.
• Check that the final test runner begins at the left edge after HandlerTests ends.

Are the final tests missing from the run?

Keep both method names prefixed with test_ inside HandlerTests. Place the final test runner outside the class.

Ask for help with test discovery:

Before you run the complete suite, do you expect the pretty-printed body to pass the signature check or fail it?

• Run the complete local receiver suite by running this command:

make test

What does the complete suite check?

The suite covers valid signatures, invalid signatures, base64 bodies, normalization, FIFO parameters, URL validation, and the reserialization trap. It uses fake AWS clients, so this command creates no module AWS resources.

You should see all five tests pass. The planned whitespace mismatch remains a passing assertion because the reconstructed bytes fail verification.

Is the complete suite failing?

Use the first failing test name to choose the receiver function to inspect. Check raw-body extraction for the base64 test, signature construction for authentication tests, and indentation for missing tests.

Ask for help with the complete suite:

That is the trust boundary working. Verified bytes now become a normalized FIFO message while invalid input stops before side effects.

✔️ Awesome, I've got everything!

Your synthetic receiver suite passes without creating any AWS resources.

ⓧ I'd like to double check the full code

import base64
import hashlib
import hmac
import json
import os
import sys
import time
import unittest
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parents[1] / "src" / "receiver"))
import handler

SECRET = "synthetic-test-secret"


class FakeSecrets:
    def get_secret_value(self, **kwargs):
        self.request = kwargs
        return {"SecretString": SECRET}


class FakeSqs:
    def __init__(self):
        self.messages = []

    def send_message(self, **kwargs):
        self.messages.append(kwargs)
        return {"MessageId": "synthetic-message", "SequenceNumber": "1"}


def body_text():
    return json.dumps({
        "event": "meeting.rtms_started",
        "event_ts": 1760486400000,
        "payload": {
            "account_id": "synthetic-account",
            "meeting_uuid": "synthetic-meeting",
            "rtms_stream_id": "synthetic-stream",
            "server_urls": "synthetic-only"
        }
    }, separators=(",", ":"))


def signature(body, timestamp):
    message = f"v0:{timestamp}:{body}".encode()
    return "v0=" + hmac.new(SECRET.encode(), message, hashlib.sha256).hexdigest()


class HandlerTests(unittest.TestCase):
    def setUp(self):
        os.environ["QUEUE_URL"] = "synthetic-queue-url"
        os.environ["WEBHOOK_SECRET_ARN"] = "synthetic-secret-arn"
        os.environ["MAX_SIGNATURE_AGE_SECONDS"] = "300"
        handler._secret_cache = None
        handler._secrets_client = FakeSecrets()
        handler._sqs_client = FakeSqs()

    def signed_event(self, body=None, encoded=False):
        body = body or body_text()
        timestamp = str(int(time.time()))
        event_body = base64.b64encode(body.encode()).decode() if encoded else body
        return {
            "version": "2.0",
            "headers": {
                "x-zm-request-timestamp": timestamp,
                "x-zm-signature": signature(body, timestamp),
            },
            "body": event_body,
            "isBase64Encoded": encoded,
        }

    def test_accepts_signed_event_and_sets_fifo_fields(self):
        response = handler.lambda_handler(self.signed_event(), None)
        self.assertEqual(202, response["statusCode"])
        queued = handler._sqs_client.messages[0]
        self.assertEqual("synthetic-meeting", queued["MessageGroupId"])
        self.assertEqual(64, len(queued["MessageDeduplicationId"]))

    def test_accepts_base64_body(self):
        response = handler.lambda_handler(self.signed_event(encoded=True), None)
        self.assertEqual(202, response["statusCode"])

    def test_rejects_invalid_signature(self):
        event = self.signed_event()
        event["headers"]["x-zm-signature"] = "v0=invalid"
        self.assertEqual(401, handler.lambda_handler(event, None)["statusCode"])

    def test_reserialization_is_not_signature_safe(self):
        compact = body_text()
        timestamp = str(int(time.time()))
        pretty = json.dumps(json.loads(compact), indent=2)
        self.assertFalse(handler.verify_signature(
            pretty.encode(), timestamp, signature(compact, timestamp), SECRET, now=int(timestamp)
        ))

    def test_endpoint_validation(self):
        body = json.dumps({
            "event": "endpoint.url_validation",
            "event_ts": 1760486400000,
            "payload": {"plainToken": "synthetic-token"},
        }, separators=(",", ":"))
        response = handler.lambda_handler({"body": body, "headers": {}, "isBase64Encoded": False}, None)
        payload = json.loads(response["body"])
        expected = hmac.new(SECRET.encode(), b"synthetic-token", hashlib.sha256).hexdigest()
        self.assertEqual({"plainToken": "synthetic-token", "encryptedToken": expected}, payload)


if __name__ == "__main__":
    unittest.main()

How to compare this file

This reference shows the complete synthetic test suite for the receiver. The fake clients, fixture values, and five test methods should match your saved file.

Your receiver now preserves signed bytes, rejects untrusted requests, and emits a versioned FIFO envelope. Next, you will make the queue settings concrete before deploying the infrastructure with CDK.

Deploy the Guardrail and Ingest Stack

Your receiver now enforces the trust boundary in local tests. A cloud deployment introduces queues, permissions, public routing, logging, secrets, and cost controls that must match that boundary.

In this step, you will use AWS CDK v2 to deploy the admission path. First, you will explore temporary Amazon SQS FIFO queues so the source queue and dead-letter queue settings become concrete before infrastructure as code takes over.

In this step, get ready to:

• Inspect temporary FIFO source queue and DLQ settings.
• Define the persistent guardrail stack plus the ephemeral ingest stack in CDK.
• Deploy both stacks in us-east-1.

Explore FIFO queue settings

A FIFO source queue preserves order within each message group. A DLQ isolates messages that repeatedly fail processing. Creating both manually makes their relationship visible before CDK creates the final versions.

• Open the AWS Console in your browser.
• Sign in through the account path associated with your assume-role profile.
• Complete your normal authentication prompt if one appears.
• Select the us-east-1 region.
• Open Amazon SQS from the console search bar.
• Click Create queue.
• Select FIFO under Type.
• Enter meeting-capture-ingress-dlq.fifo as the queue name.

Why does the name end in .fifo?

The .fifo suffix identifies an SQS FIFO queue. The queue type and suffix must agree.

• Leave Content-based deduplication disabled.
• Keep server-side encryption enabled.
• Set the message retention period to four days.
• Click Create queue.

You should see meeting-capture-ingress-dlq.fifo listed as a FIFO queue. This queue now provides a temporary destination for repeatedly failed messages.

• Return to the Amazon SQS queue list.
• Click Create queue.
• Select FIFO under Type.
• Enter meeting-capture-ingress.fifo as the queue name.
• Leave Content-based deduplication disabled.
• Keep server-side encryption enabled.
• Set the message retention period to four days.
• Set the visibility timeout to 30 seconds.

Why disable content-based deduplication?

The receiver calculates a SHA-256 event fingerprint. It sends that fingerprint as the explicit message deduplication ID.

Explicit IDs make the producer's deduplication decision visible. They also keep the policy independent from the serialized message body.

• Enable the Dead-letter queue setting.
• Select meeting-capture-ingress-dlq.fifo as the DLQ.
• Set the maximum receive count to 3.
• Review the Redrive allow policy setting.
• Click Create queue.

You should see both temporary queues in the queue list. The source queue should point to the FIFO DLQ after three unsuccessful receives.

• Select meeting-capture-ingress.fifo from the queue list.
• Click Delete.
• Enter delete in the Delete queue dialog.
• Click Delete in the dialog.
• Select meeting-capture-ingress-dlq.fifo from the queue list.
• Click Delete.
• Enter delete in the Delete queue dialog.
• Click Delete in the dialog.

That manual detour has done its job. Both temporary queues are gone, so the CDK deployment can create the managed versions without name collisions or configuration drift.

Still see a temporary queue?

Refresh the queue list. Confirm that you deleted the source queue before deleting its DLQ.

If a queue remains, help me remove the temporary FIFO queues safely.

Define the guardrail and ingest stacks

The CDK application separates persistent cost monitoring from ephemeral ingestion infrastructure. This lets the next step delete the ingest boundary while preserving its monthly budget guardrail.

• Create stacks.py inside the meeting_capture_adapter directory in your repository editor.
• Add the CDK imports to meeting_capture_adapter/stacks.py by pasting this code:

from aws_cdk import CfnOutput, Duration, RemovalPolicy, Stack
from aws_cdk import aws_apigatewayv2 as apigwv2
from aws_cdk import aws_budgets as budgets
from aws_cdk import aws_lambda as lambda_
from aws_cdk import aws_logs as logs
from aws_cdk import aws_secretsmanager as secretsmanager
from aws_cdk import aws_sqs as sqs
from aws_cdk.aws_apigatewayv2_integrations import HttpLambdaIntegration
from constructs import Construct

What do these imports provide?

• The CDK modules expose the budget, queue, secret, log, Lambda, and HTTP API constructs used by the stacks.
• The Construct base type lets each stack join the CDK construct tree.
• The output and removal-policy types control deployed values plus teardown behavior.

• Save meeting_capture_adapter/stacks.py.
• Confirm that your editor shows no unresolved import diagnostics.

See unresolved CDK imports?

Confirm that your editor uses the repository's virtual environment. Check that the pinned dependencies from requirements.txt remain installed.

If the imports still fail, help me diagnose the CDK imports in stacks.py.

The persistent stack uses AWS Budgets to monitor monthly cost. Its three notifications report actual spend at absolute thresholds below the 20 USD limit.

• Add GuardrailStack below the imports in meeting_capture_adapter/stacks.py by pasting this code:

class GuardrailStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, *, budget_email: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)
        subscriber = budgets.CfnBudget.SubscriberProperty(
            address=budget_email,
            subscription_type="EMAIL",
        )
        notifications = [
            budgets.CfnBudget.NotificationWithSubscribersProperty(
                notification=budgets.CfnBudget.NotificationProperty(
                    comparison_operator="GREATER_THAN",
                    notification_type="ACTUAL",
                    threshold=amount,
                    threshold_type="ABSOLUTE_VALUE",
                ),
                subscribers=[subscriber],
            )
            for amount in (5, 10, 15)
        ]
        budgets.CfnBudget(
            self,
            "MonthlyBudget",
            budget=budgets.CfnBudget.BudgetDataProperty(
                budget_name="meeting-capture-adapter-monthly",
                budget_type="COST",
                time_unit="MONTHLY",
                budget_limit=budgets.CfnBudget.SpendProperty(amount=20, unit="USD"),
            ),
            notifications_with_subscribers=notifications,
        )

What does the guardrail stack do?

• The subscriber sends budget notifications to the email supplied during deployment.
• The notification list creates actual-spend thresholds at 5, 10, and 15 USD.
• The budget tracks monthly cost with a 20 USD limit.

• Save meeting_capture_adapter/stacks.py.
• Confirm that GuardrailStack remains aligned at the left edge of the file.

Does the budget class look nested?

Check the indentation immediately before class GuardrailStack. The class must begin at the left edge.

If the structure is unclear, help me check the GuardrailStack indentation.

The ingest stack begins with two encrypted FIFO queues. The source queue explicitly disables content-based deduplication because the receiver supplies its own event fingerprint.

• Add the FIFO queue definitions below GuardrailStack in meeting_capture_adapter/stacks.py by pasting this code:

class IngestStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        dlq = sqs.Queue(
            self,
            "IngressDlq",
            queue_name="meeting-capture-ingress-dlq.fifo",
            fifo=True,
            enforce_ssl=True,
            retention_period=Duration.days(4),
            removal_policy=RemovalPolicy.DESTROY,
        )
        queue = sqs.Queue(
            self,
            "IngressQueue",
            queue_name="meeting-capture-ingress.fifo",
            fifo=True,
            content_based_deduplication=False,
            enforce_ssl=True,
            dead_letter_queue=sqs.DeadLetterQueue(max_receive_count=3, queue=dlq),
            retention_period=Duration.days(4),
            visibility_timeout=Duration.seconds(30),
            removal_policy=RemovalPolicy.DESTROY,
        )

How are the queues configured?

• Both queue names end with .fifo.
• Both queues require secure transport.
• The source queue moves a repeatedly received message to the DLQ after three receives.
• The source queue uses a 30-second visibility timeout plus four-day retention.

• Save meeting_capture_adapter/stacks.py.
• Confirm that both queue definitions remain inside IngestStack.__init__.

Are the queue definitions misaligned?

Check that dlq and queue have the same indentation. Confirm that both remain below the super().__init__ call.

If your editor reports a syntax problem, help me compare the FIFO queue definitions.

The receiver needs a generated secret plus a one-day log group. AWS Secrets Manager owns the secret lifecycle while Amazon CloudWatch Logs limits retained receiver logs.

• Add the secret and log group below the queue definitions by pasting this code:

        secret = secretsmanager.Secret(
            self,
            "SyntheticWebhookSecret",
            description="Synthetic webhook secret for Meeting Capture Adapter Module 1",
            removal_policy=RemovalPolicy.DESTROY,
        )
        log_group = logs.LogGroup(
            self,
            "ReceiverLogs",
            log_group_name="/aws/lambda/meeting-capture-webhook-receiver",
            retention=logs.RetentionDays.ONE_DAY,
            removal_policy=RemovalPolicy.DESTROY,
        )

Why manage these resources explicitly?

The secret receives a generated value because no secret text appears in the repository. The log group uses one-day retention to limit stored test logs.

Both resources use RemovalPolicy.DESTROY so deleting the ingest stack removes its ephemeral state.

• Save meeting_capture_adapter/stacks.py.
• Confirm that the secret description matches Synthetic webhook secret for Meeting Capture Adapter Module 1.

Is the log group outside IngestStack?

Compare the indentation of secret and log_group with the earlier queue definition. All three belong inside the same initializer.

If the nesting is unclear, help me place the secret and log group correctly.

The AWS Lambda construct packages the receiver from build/receiver. Its environment contains resource references plus the adapter-owned replay limit.

• Add the Lambda definition plus least-privilege grants below the log group by pasting this code:

        receiver = lambda_.Function(
            self,
            "WebhookReceiver",
            function_name="meeting-capture-webhook-receiver",
            runtime=lambda_.Runtime.PYTHON_3_14,
            handler="handler.lambda_handler",
            code=lambda_.Code.from_asset("build/receiver"),
            memory_size=128,
            timeout=Duration.seconds(5),
            log_group=log_group,
            environment={
                "QUEUE_URL": queue.queue_url,
                "WEBHOOK_SECRET_ARN": secret.secret_arn,
                "MAX_SIGNATURE_AGE_SECONDS": "300",
            },
        )
        queue.grant_send_messages(receiver)
        secret.grant_read(receiver)

What reaches the Lambda?

• The runtime uses Python 3.14 with 128 MB memory.
• The timeout limits an invocation to five seconds.
• The environment passes resource identifiers instead of secret values.
• The grants allow queue sends plus secret reads.

• Save meeting_capture_adapter/stacks.py.
• Confirm that no literal secret value appears in the Lambda environment.

Does the Lambda reference the wrong path?

Confirm that the asset path is build/receiver. Confirm that the handler is handler.lambda_handler.

If either reference differs, help me check the Lambda asset and handler configuration.

An Amazon API Gateway HTTP API provides the public ingress boundary. The stack exports the values needed for synthetic invocation plus later inventory checks.

• Finish IngestStack by adding the integration, route, and outputs below the grants:

        integration = HttpLambdaIntegration("ReceiverIntegration", receiver)
        api = apigwv2.HttpApi(
            self,
            "IngressApi",
            api_name="meeting-capture-ingress",
            description="Synthetic Zoom-compatible webhook ingress",
        )
        api.add_routes(
            path="/webhooks/zoom",
            methods=[apigwv2.HttpMethod.POST],
            integration=integration,
        )

        CfnOutput(self, "ApiUrl", value=f"{api.api_endpoint}/webhooks/zoom")
        CfnOutput(self, "QueueUrl", value=queue.queue_url)
        CfnOutput(self, "SecretArn", value=secret.secret_arn)

How does the public route connect?

• The Lambda integration directs HTTP API requests to the receiver.
• The API exposes POST requests at /webhooks/zoom.
• The outputs publish the API URL, queue URL, and secret ARN without publishing the secret value.

• Save meeting_capture_adapter/stacks.py.
• Confirm that the final three output lines remain inside IngestStack.__init__.

Is the route missing from the stack?

Confirm that api.add_routes uses /webhooks/zoom plus the Lambda integration. Check that the method remains POST.

If the route structure differs, help me compare the HTTP API section.

✔️ Awesome, stacks.py is complete

Great. Save meeting_capture_adapter/stacks.py before creating the CDK application entry point.

ⓧ I'd like to double check the full code

Compare your file with the complete meeting_capture_adapter/stacks.py below.

from aws_cdk import CfnOutput, Duration, RemovalPolicy, Stack
from aws_cdk import aws_apigatewayv2 as apigwv2
from aws_cdk import aws_budgets as budgets
from aws_cdk import aws_lambda as lambda_
from aws_cdk import aws_logs as logs
from aws_cdk import aws_secretsmanager as secretsmanager
from aws_cdk import aws_sqs as sqs
from aws_cdk.aws_apigatewayv2_integrations import HttpLambdaIntegration
from constructs import Construct


class GuardrailStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, *, budget_email: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)
        subscriber = budgets.CfnBudget.SubscriberProperty(
            address=budget_email,
            subscription_type="EMAIL",
        )
        notifications = [
            budgets.CfnBudget.NotificationWithSubscribersProperty(
                notification=budgets.CfnBudget.NotificationProperty(
                    comparison_operator="GREATER_THAN",
                    notification_type="ACTUAL",
                    threshold=amount,
                    threshold_type="ABSOLUTE_VALUE",
                ),
                subscribers=[subscriber],
            )
            for amount in (5, 10, 15)
        ]
        budgets.CfnBudget(
            self,
            "MonthlyBudget",
            budget=budgets.CfnBudget.BudgetDataProperty(
                budget_name="meeting-capture-adapter-monthly",
                budget_type="COST",
                time_unit="MONTHLY",
                budget_limit=budgets.CfnBudget.SpendProperty(amount=20, unit="USD"),
            ),
            notifications_with_subscribers=notifications,
        )


class IngestStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        dlq = sqs.Queue(
            self,
            "IngressDlq",
            queue_name="meeting-capture-ingress-dlq.fifo",
            fifo=True,
            enforce_ssl=True,
            retention_period=Duration.days(4),
            removal_policy=RemovalPolicy.DESTROY,
        )
        queue = sqs.Queue(
            self,
            "IngressQueue",
            queue_name="meeting-capture-ingress.fifo",
            fifo=True,
            content_based_deduplication=False,
            enforce_ssl=True,
            dead_letter_queue=sqs.DeadLetterQueue(max_receive_count=3, queue=dlq),
            retention_period=Duration.days(4),
            visibility_timeout=Duration.seconds(30),
            removal_policy=RemovalPolicy.DESTROY,
        )
        secret = secretsmanager.Secret(
            self,
            "SyntheticWebhookSecret",
            description="Synthetic webhook secret for Meeting Capture Adapter Module 1",
            removal_policy=RemovalPolicy.DESTROY,
        )
        log_group = logs.LogGroup(
            self,
            "ReceiverLogs",
            log_group_name="/aws/lambda/meeting-capture-webhook-receiver",
            retention=logs.RetentionDays.ONE_DAY,
            removal_policy=RemovalPolicy.DESTROY,
        )
        receiver = lambda_.Function(
            self,
            "WebhookReceiver",
            function_name="meeting-capture-webhook-receiver",
            runtime=lambda_.Runtime.PYTHON_3_14,
            handler="handler.lambda_handler",
            code=lambda_.Code.from_asset("build/receiver"),
            memory_size=128,
            timeout=Duration.seconds(5),
            log_group=log_group,
            environment={
                "QUEUE_URL": queue.queue_url,
                "WEBHOOK_SECRET_ARN": secret.secret_arn,
                "MAX_SIGNATURE_AGE_SECONDS": "300",
            },
        )
        queue.grant_send_messages(receiver)
        secret.grant_read(receiver)

        integration = HttpLambdaIntegration("ReceiverIntegration", receiver)
        api = apigwv2.HttpApi(
            self,
            "IngressApi",
            api_name="meeting-capture-ingress",
            description="Synthetic Zoom-compatible webhook ingress",
        )
        api.add_routes(
            path="/webhooks/zoom",
            methods=[apigwv2.HttpMethod.POST],
            integration=integration,
        )

        CfnOutput(self, "ApiUrl", value=f"{api.api_endpoint}/webhooks/zoom")
        CfnOutput(self, "QueueUrl", value=queue.queue_url)
        CfnOutput(self, "SecretArn", value=secret.secret_arn)

The application entry point fixes both stacks to us-east-1. It also refuses to synthesize without a budget notification email.

• Create app.py at the top level of your repository in your editor.
• Add the CDK application entry point to app.py by pasting this code:

import os

from aws_cdk import App, Environment

from meeting_capture_adapter.stacks import GuardrailStack, IngestStack

REGION = "us-east-1"
budget_email = os.environ.get("BUDGET_EMAIL")
if not budget_email:
    raise ValueError("Set BUDGET_EMAIL before synth or deploy")

app = App()
environment = Environment(region=REGION)
GuardrailStack(app, "MeetingCaptureGuardrailStack", budget_email=budget_email, env=environment)
IngestStack(app, "MeetingCaptureIngestStack", env=environment)
app.synth()

How does the application assemble the stacks?

• The region constant keeps both stacks in us-east-1.
• The budget email check stops synthesis when the notification destination is missing.
• The stack IDs become MeetingCaptureGuardrailStack and MeetingCaptureIngestStack.
• The final synthesis call turns the constructs into deployable CloudFormation templates.

• Save app.py.
• Confirm that app.py sits beside cdk.json at the top level of the repository.

Can CDK find the application?

Confirm that the file is named app.py. Confirm that cdk.json still points to python3 app.py.

If the entry point is not found, help me check the CDK application path.

✔️ Awesome, app.py is complete

Great. Save app.py before reviewing the infrastructure diff.

ⓧ I'd like to double check the full code

Compare your file with the complete app.py below.

import os

from aws_cdk import App, Environment

from meeting_capture_adapter.stacks import GuardrailStack, IngestStack

REGION = "us-east-1"
budget_email = os.environ.get("BUDGET_EMAIL")
if not budget_email:
    raise ValueError("Set BUDGET_EMAIL before synth or deploy")

app = App()
environment = Environment(region=REGION)
GuardrailStack(app, "MeetingCaptureGuardrailStack", budget_email=budget_email, env=environment)
IngestStack(app, "MeetingCaptureIngestStack", env=environment)
app.synth()

Review and deploy the stacks

A CDK diff previews the infrastructure changes before AWS creates anything. This is your chance to confirm that the template contains the intended guardrail and ingress boundary.

• Replace your-assume-role-profile with your validated assume-role profile in the command below.
• Replace you@example.com with the email address that should receive budget notifications.
• Review the synthesized infrastructure by running this command in the repository terminal:

make diff AWS_PROFILE=your-assume-role-profile BUDGET_EMAIL=you@example.com

What does the diff target check?

The target packages the receiver before running the local tests. It then asks CDK to compare both synthesized stacks with the current AWS account.

Review the additions for the monthly budget, FIFO queue pair, generated secret, log group, Lambda, HTTP API, IAM grants, and stack outputs.

The terminal should show additions for both stacks. It should not show unexpected deletion of shared CDK bootstrap infrastructure.

Does the diff fail?

Confirm that your temporary AWS role session remains valid. Check that the profile name matches the one that passed the prerequisite check.

If the failure points to packaging, confirm that Docker remains available from WSL2. For focused help, help me diagnose my CDK diff failure.

Know what this deployment creates

This action creates a persistent Secrets Manager secret that costs about $0.00056 per hour until the ingest stack is deleted. The prescribed run remains very small.

The 20 USD budget sends alerts. It does not guarantee that AWS stops spending at that amount.

Deploying cloud infrastructure can feel risky. The ingest resources use destroy policies, and the next step removes them after the behavioral proof.

• Deploy both stacks by running this command with your profile and email values:

make deploy AWS_PROFILE=your-assume-role-profile BUDGET_EMAIL=you@example.com

What does the deploy target do?

The target packages the receiver and reruns the local tests. CDK then deploys MeetingCaptureGuardrailStack plus MeetingCaptureIngestStack.

The deployment writes the ingest outputs to cdk-outputs.json for the synthetic invocation in the next step.

Expect the deployment to take several minutes while CloudFormation creates both stacks. A quiet terminal during resource creation does not mean the process has stalled.

• Wait for the deployment command to return to the terminal prompt.
• Open cdk-outputs.json in your repository editor.
• Confirm that MeetingCaptureIngestStack contains ApiUrl.
• Confirm that the same output object contains QueueUrl.
• Confirm that the same output object contains SecretArn.
• Leave the generated secret value unopened.

Did the deployment stop?

Read the first failed resource in the deployment output. Confirm that the assume-role session has permission to deploy CDK stacks plus budgets.

If the command reports a resource-name conflict, return to Amazon SQS and confirm that both temporary manual queues were deleted. For guided diagnosis, help me troubleshoot the failed CDK deployment.

Before you refresh the stack list, which two stack names do you expect to see in the completed state?

• Return to the AWS Console.
• Open CloudFormation in us-east-1.
• Refresh the stack list.
• Confirm that MeetingCaptureGuardrailStack shows CREATE_COMPLETE.
• Confirm that MeetingCaptureIngestStack shows CREATE_COMPLETE.

You should see both stacks in the completed state. That proves CloudFormation created the persistent guardrail plus the ephemeral ingress boundary in the intended region.

• Open the resource list for MeetingCaptureGuardrailStack.
• Confirm that its budget is named meeting-capture-adapter-monthly.
• Open the resource list for MeetingCaptureIngestStack.
• Confirm that the HTTP API is named meeting-capture-ingress.
• Confirm that the Lambda is named meeting-capture-webhook-receiver.
• Confirm that both FIFO queues appear in the resource list.
• Confirm that the log group is named /aws/lambda/meeting-capture-webhook-receiver.

The deployment is complete. Your public route now leads to a receiver with restricted queue-send and secret-read access.

Your admission boundary is live with its cost guardrail in place. Next, you will send duplicate signed events, inspect the resulting FIFO message, capture the inventory, and remove the ephemeral ingest stack.

Prove Deduplication and Tear Down

Your deployed admission boundary has reached CREATE_COMPLETE. That status proves AWS created the resources successfully.

Behavior and cleanup still need evidence. You will test duplicate delivery through Amazon SQS FIFO before using AWS CDK to remove the ingest stack.

In this step, get ready to:

• Send the same signed synthetic event twice.
• Capture resource inventories before teardown.
• Destroy the ingest stack while preserving the budget guardrail.

Check the proof harness

The repository from earlier contains a synthetic sender that reads the deployed outputs. It keeps one compact JSON body unchanged across both requests.

The existing Makefile keeps invocation and inventory separate from the two stack deletion paths. This separation protects the persistent budget from the ingest teardown.

• Compare the existing proof harness with the complete references below.

✔️ Awesome, I've got everything!

Your synthetic sender and Make targets are ready for the deployed proof.

ⓧ I'd like to double check the full code

• Compare your existing tools/send_synthetic.py with this complete reference:

import argparse
import hashlib
import hmac
import json
import time
import urllib.request

import boto3

STACK = "MeetingCaptureIngestStack"
REGION = "us-east-1"


def post(url, body, secret):
    timestamp = str(int(time.time()))
    message = f"v0:{timestamp}:{body}".encode()
    signature = "v0=" + hmac.new(secret.encode(), message, hashlib.sha256).hexdigest()
    request = urllib.request.Request(
        url,
        data=body.encode(),
        method="POST",
        headers={
            "content-type": "application/json",
            "x-zm-request-timestamp": timestamp,
            "x-zm-signature": signature,
        },
    )
    started = time.perf_counter()
    with urllib.request.urlopen(request, timeout=10) as response:
        payload = response.read().decode()
        elapsed_ms = (time.perf_counter() - started) * 1000
        return response.status, elapsed_ms, payload


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--profile", required=True)
    parser.add_argument("--outputs", default="cdk-outputs.json")
    args = parser.parse_args()

    with open(args.outputs, encoding="utf-8") as file:
        outputs = json.load(file)[STACK]
    session = boto3.Session(profile_name=args.profile, region_name=REGION)
    secret = session.client("secretsmanager").get_secret_value(
        SecretId=outputs["SecretArn"]
    )["SecretString"]
    body = json.dumps({
        "event": "meeting.rtms_started",
        "event_ts": int(time.time() * 1000),
        "payload": {
            "account_id": "synthetic-account",
            "meeting_uuid": "synthetic-meeting",
            "rtms_stream_id": "synthetic-stream",
            "server_urls": "synthetic-only"
        }
    }, separators=(",", ":"))

    for attempt in (1, 2):
        status, elapsed, payload = post(outputs["ApiUrl"], body, secret)
        print(f"delivery {attempt}: status={status} latency_ms={elapsed:.1f} body={payload}")

    sqs = session.client("sqs")
    received = sqs.receive_message(
        QueueUrl=outputs["QueueUrl"],
        MaxNumberOfMessages=10,
        WaitTimeSeconds=5,
    ).get("Messages", [])
    print(f"retrieved_messages={len(received)}")
    if len(received) != 1:
        raise RuntimeError(f"Expected one deduplicated message, found {len(received)}")
    print(json.dumps(json.loads(received[0]["Body"]), indent=2))
    sqs.delete_message(
        QueueUrl=outputs["QueueUrl"],
        ReceiptHandle=received[0]["ReceiptHandle"],
    )


if __name__ == "__main__":
    main()

• Compare your existing Makefile with this complete reference:

AWS_PROFILE ?= default
REGION := us-east-1
PYTHON := .venv/bin/python
CDK := npx cdk

.PHONY: setup check package test diff deploy invoke inventory destroy destroy-guardrail

setup:
	python3 -m venv .venv
	$(PYTHON) -m pip install -r requirements.txt
	npm install

check:
	$(PYTHON) scripts/check_prereqs.py --profile $(AWS_PROFILE)

package:
	rm -rf build/receiver
	mkdir -p build/receiver
	$(PYTHON) -m pip install -r src/receiver/requirements.txt --target build/receiver
	cp src/receiver/handler.py build/receiver/handler.py

test:
	PYTHONPATH=src/receiver $(PYTHON) -m unittest discover -s tests -v

diff: package test
	BUDGET_EMAIL=$(BUDGET_EMAIL) $(CDK) diff --profile $(AWS_PROFILE)

deploy: package test
	test -n "$(BUDGET_EMAIL)" || (echo "Set BUDGET_EMAIL" && exit 1)
	BUDGET_EMAIL=$(BUDGET_EMAIL) $(CDK) deploy MeetingCaptureGuardrailStack MeetingCaptureIngestStack --profile $(AWS_PROFILE) --require-approval never --outputs-file cdk-outputs.json

invoke:
	$(PYTHON) tools/send_synthetic.py --profile $(AWS_PROFILE)

inventory:
	-aws cloudformation list-stack-resources --stack-name MeetingCaptureIngestStack --profile $(AWS_PROFILE) --region $(REGION)
	aws sqs list-queues --profile $(AWS_PROFILE) --region $(REGION)
	aws apigatewayv2 get-apis --profile $(AWS_PROFILE) --region $(REGION)
	aws lambda list-functions --profile $(AWS_PROFILE) --region $(REGION)
	aws logs describe-log-groups --profile $(AWS_PROFILE) --region $(REGION)
	aws secretsmanager list-secrets --profile $(AWS_PROFILE) --region $(REGION)

destroy:
	BUDGET_EMAIL=$(BUDGET_EMAIL) $(CDK) destroy MeetingCaptureIngestStack --profile $(AWS_PROFILE)

destroy-guardrail:
	BUDGET_EMAIL=$(BUDGET_EMAIL) $(CDK) destroy MeetingCaptureGuardrailStack --profile $(AWS_PROFILE)

How does the proof harness work?

• The sender reads ApiUrl. It also reads QueueUrl and SecretArn from cdk-outputs.json.
• The selected profile retrieves the generated secret without printing it.
• The post() function signs the exact compact body.
• The loop posts the unchanged body twice.
• The queue check retrieves up to ten messages.
• The final call deletes the retrieved message with its receipt handle.

Invoke duplicates and destroy the ingest stack

A behavior proof needs the deployed route and queue to participate in the same run. The command below invokes the sender before it inventories and removes the resources.

This deletion is intentional

Deleting the ingest stack removes the admission path and its ephemeral resources. This includes the API and receiver.

The source queue and DLQ are also removed. The log group and generated secret follow the stack removal policy.

MeetingCaptureGuardrailStack stays deployed. Its USD 20 monthly budget continues protecting later modules.

Before you run this, how many HTTP acknowledgements and retrievable queue messages do you expect from two identical deliveries?

• Replace your-assume-role-profile with your existing assume-role profile name when you type the command.
• Run the deployed proof and capture the pre-destroy inventories with this command:
• Approve the CDK deletion only after it names MeetingCaptureIngestStack.

make invoke inventory destroy AWS_PROFILE=your-assume-role-profile

What does this command do?

• The invoke target retrieves the generated secret through AWS Secrets Manager.
• The sender posts the identical signed event twice.
• The sender receives the available FIFO messages.
• The sender prints the normalized non-secret envelope.
• The inventory target lists the ingest stack through AWS CloudFormation.
• The inventory also lists queues and APIs. It lists functions and log groups.
• The inventory finishes by listing secrets.
• The destroy target removes only MeetingCaptureIngestStack.

You will see two delivery lines with status 202. You will then see retrieved_messages=1 followed by the normalized envelope.

The sender deletes that message from meeting-capture-ingress.fifo. CDK then removes the ingest stack.

That is the behavior proof complete. Your deployed boundary acknowledged both deliveries while the FIFO queue exposed one message.

Did the proof chain stop?

• Check that your assume-role session remains active in WSL2.
• Check that cdk-outputs.json still contains the MeetingCaptureIngestStack outputs.
• Check that BUDGET_EMAIL remains available for CDK application synthesis.

Use this if the chain still stops: Help me diagnose the first failing target.

Verify cleanup and evaluate the boundary

The first inventory captured the resources before deletion. A second inventory now checks the account after the ingest stack is gone.

Before you run the final check, which named ingest resources do you expect to find?

• Capture the post-destroy inventory by running:

make inventory

What does the final inventory check?

The target checks the former stack in us-east-1. It also checks the regional service inventories.

Your account may contain unrelated resources. Success means the named Meeting Capture ingest resources are absent.

You should no longer find MeetingCaptureIngestStack or meeting-capture-ingress. You should also no longer find meeting-capture-webhook-receiver.

The source queue and meeting-capture-ingress-dlq.fifo should be absent. The generated secret and one-day log group should also be absent.

• Use the signed-in AWS Console session from earlier to view CloudFormation in us-east-1.
• Confirm that MeetingCaptureGuardrailStack remains CREATE_COMPLETE.

The ephemeral admission path is gone. The persistent monthly budget remains available for the next module.

Does the inventory mention a missing stack?

The CloudFormation lookup can report that the deleted ingest stack is missing. The leading hyphen in the Makefile allows the remaining inventory commands to continue.

Use this if a named ingest resource still appears: Help me investigate a resource that remained after teardown.

Reconcile the module estimate

• Ten HTTP API calls at $1.00 per million cost $0.000010.
• Ten Lambda requests at $0.20 per million cost $0.000002.
• Ten executions at 128 MB for 100 ms use 0.125 GB-seconds.
• That duration costs about $0.00000208 at $0.0000166667 per GB-second.
• One secret stored for one hour costs about $0.00056.
• Ten secret API calls at $0.05 per 10,000 calls cost $0.00005.
• The prescribed SQS traffic costs $0 while it remains inside the monthly one-million-request allowance.
• The prescribed log traffic costs $0 while it remains inside the monthly 5 GB allowance.
• The estimated module subtotal is about $0.000624 under those stated allowances.
• Shared CDK bootstrap storage is excluded because it predates this stack.

What did this proof establish?

The two 202 responses show that both requests reached the admission boundary. The one queue message shows producer-side suppression during the five-minute FIFO deduplication interval.

Future consumers still need idempotency for end-to-end processing. This test covers the producer-side deduplication scope.

The trust boundary begins with the raw request bytes. Signature verification and replay checks must finish before parsing drives workflow decisions or side effects.

A Secrets Manager outage must produce a retryable server response that fails closed. Warm-cache reuse reduces secret lookups without admitting unverified traffic.

Test your design reasoning

• Explain why two 202 responses plus one FIFO message still require future consumer idempotency.
• Identify the operations that belong before parsing or side effects.
• Describe the endpoint behavior required during a Secrets Manager outage.

You have proved duplicate suppression and captured teardown evidence. The cost guardrail remains while every named ingest resource is gone.

Secret mission

Defend the Replay Boundary

A valid signature can still be reused after its original delivery. Add deterministic boundary tests for the adapter's five-minute freshness policy. Document who owns the policy and what evidence could justify changing it.

Clean Up Your Resources

Clean Up Your Resources

Your ingest stack cleanup is complete. Choose whether to keep the free AWS Budgets guardrail, stay paused between modules, or delete the guardrail when the series ends.

Resources you used:

• The deployed MeetingCaptureGuardrailStack contains the meeting-capture-adapter-monthly monthly budget. It sends actual-cost notifications at USD 5, USD 10, and USD 15.
• The MeetingCaptureIngestStack is deleted. Its CloudFormation resources are absent from the post-destroy inventory.
• The Amazon API Gateway HTTP API named meeting-capture-ingress is deleted.
• The AWS Lambda function named meeting-capture-webhook-receiver is deleted with its role.
• The Amazon SQS FIFO queues named meeting-capture-ingress.fifo and meeting-capture-ingress-dlq.fifo are deleted.
• The CloudWatch Logs group named /aws/lambda/meeting-capture-webhook-receiver is deleted.
• The generated Secrets Manager secret named SyntheticWebhookSecret is deleted.

Shared Bootstrap Exception

The existing AWS CDK bootstrap stack and staging bucket predate this module. They stay in place because other CDK applications may rely on them.

Your inventory records this shared infrastructure as an explicit teardown exception.

Keep everything running

No action needed. Choose this if you plan to continue the Meeting Capture Adapter series with the current cost guardrail.

• Retain MeetingCaptureGuardrailStack with its USD 20 monthly budget.
• Keep the repository and local tests available for the next module.
• Leave the shared CDK bootstrap infrastructure untouched.

Pause - I'll come back to this later

Your cloud environment already matches the recommended state between modules. The ingest stack is gone while the free guardrail remains.

• Leave MeetingCaptureGuardrailStack deployed to keep the monthly budget active.
• Retain the post-destroy inventories as cleanup evidence.
• Keep the shared CDK bootstrap stack available for future deployments.

Delete - I don't want to use this again

Deleting the guardrail is final for this series. Your repository and shared bootstrap infrastructure stay intact.

• Destroy MeetingCaptureGuardrailStack from the repository by running this command in the WSL2 terminal:
• Confirm the deletion when AWS CDK asks for approval.

make destroy-guardrail AWS_PROFILE=[[AWS_PROFILE="your-assume-role-profile"]] BUDGET_EMAIL=[[BUDGET_EMAIL="you@example.com"]]

What Does This Command Do?

• The destroy-guardrail target selects only MeetingCaptureGuardrailStack for deletion.
• The AWS_PROFILE value uses the same assume-role profile from deployment.
• The BUDGET_EMAIL value lets the CDK app synthesize before deletion.

• Use the region selector in the AWS console top navigation to select US East (N. Virginia), us-east-1.
• Open AWS CloudFormation from the AWS console search bar.
• Confirm that MeetingCaptureGuardrailStack no longer appears among the active stacks.
• Open AWS Budgets from the AWS console search bar.
• Confirm that meeting-capture-adapter-monthly no longer appears in the budget list.
• Review the post-destroy inventory from Step 5 to confirm that the named ingest resources remain absent.

Guardrail Stack Still Present?

• Refresh the temporary assume-role session using the same process from Step 1.
• Rerun the project target with the same profile and budget email.
• Help me debug why MeetingCaptureGuardrailStack still exists after I ran the destroy-guardrail target.

Once both entries are absent, the final persistent module resource is gone. Your local implementation evidence remains available.

Nice Work!

Nice Work!

Excellent work! You built a secure webhook admission boundary for signed synthetic RTMS events. Your finished system preserves raw request bytes before placing normalized events on an encrypted FIFO queue.

What you learned:

• Build a versioned webhook contract for normalized synthetic RTMS events. Protect its trust boundary with raw-body HMAC verification before parsing or side effects.
• Deploy a cost-bounded admission path with AWS CDK in us-east-1. Keep a persistent $20 budget guardrail through AWS Budgets after removing the ingest resources.
• Prove duplicate suppression in Amazon SQS FIFO by observing two 202 acknowledgements with one retrievable message. Verify clean teardown through before-and-after resource inventories.
• Secret Mission: Add replay-window boundary tests for ages 299, 300, and 301 seconds. Document the five-minute window as an adapter-owned policy in docs/ADR-003-replay-window-policy.md.

Ready to quiz yourself?