From 22ae0b65c30ac03cc55a21025b03b450c7da73b7 Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Mon, 31 Aug 2026 22:18:35 -0400 Subject: [PATCH 01/21] feat: add auth secrets and PII boundaries --- Dockerfile | 2 +- api_router.py | 23 +++++------- auth.py | 84 +++++++++++++++++++++++++++++++++++++++++++ consumer.py | 85 +++++++++++++++----------------------------- event_bus.py | 81 +++++++++++++++++------------------------ main.py | 28 ++------------- pii_policy.py | 47 ++++++++++++++++++++++++ pyproject.toml | 8 ++++- requirements-dev.txt | 3 ++ requirements.txt | 2 ++ runtime_secrets.py | 42 ++++++++++++++++++++++ 11 files changed, 258 insertions(+), 147 deletions(-) create mode 100644 auth.py create mode 100644 pii_policy.py create mode 100644 runtime_secrets.py diff --git a/Dockerfile b/Dockerfile index 4a6f344..1b6e28e 100644 --- a/Dockerfile +++ b/Dockerfile @@ -7,7 +7,7 @@ COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY shared ./shared -COPY api_router.py consumer.py event_bus.py location_store.py main.py models.py utils.py ./ +COPY api_router.py auth.py consumer.py event_bus.py location_store.py main.py models.py pii_policy.py runtime_secrets.py utils.py ./ EXPOSE 8000 diff --git a/api_router.py b/api_router.py index 608c388..dfed2ce 100644 --- a/api_router.py +++ b/api_router.py @@ -1,5 +1,6 @@ -from fastapi import APIRouter, HTTPException +from fastapi import APIRouter, Depends, HTTPException +from auth import require_authenticated_request from utils import get_logger logger = get_logger("DriverLocationAPI") @@ -9,11 +10,9 @@ DRIVER_STORE = None -@router.get("/drivers") +@router.get("/drivers", dependencies=[Depends(require_authenticated_request)]) async def get_all_drivers(): - """ - Returns all active drivers and their current coordinates. - """ + """Return active drivers and current coordinates to authenticated callers.""" if DRIVER_STORE is None: raise HTTPException(500, "Driver store not initialized.") @@ -21,11 +20,9 @@ async def get_all_drivers(): return {"count": len(drivers), "drivers": drivers} -@router.get("/drivers/{driver_id}") +@router.get("/drivers/{driver_id}", dependencies=[Depends(require_authenticated_request)]) async def get_driver(driver_id: str): - """ - Returns location data for a specific driver. - """ + """Return location data for a specific driver to an authenticated caller.""" if DRIVER_STORE is None: raise HTTPException(500, "Driver store not initialized.") @@ -36,11 +33,9 @@ async def get_driver(driver_id: str): return driver -@router.get("/count") +@router.get("/count", dependencies=[Depends(require_authenticated_request)]) async def get_driver_count(): - """ - Quick counter to see how many drivers are active. - """ + """Return the number of active drivers to an authenticated caller.""" if DRIVER_STORE is None: raise HTTPException(500, "Driver store not initialized.") @@ -62,7 +57,7 @@ async def readiness_check(): try: ready = DRIVER_STORE.is_ready() except Exception as exc: - logger.error("Driver store readiness check failed: %s", exc) + logger.error("Driver store readiness check failed") raise HTTPException(503, "Driver store is unavailable.") from exc if not ready: diff --git a/auth.py b/auth.py new file mode 100644 index 0000000..3e9fe5d --- /dev/null +++ b/auth.py @@ -0,0 +1,84 @@ +"""OIDC JWT authentication boundary for sensitive driver-location endpoints.""" + +import os +from functools import lru_cache +from typing import Any + +import jwt +from fastapi import Depends, HTTPException, status +from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer +from jwt import PyJWKClient + +_bearer = HTTPBearer(auto_error=False) + + +def auth_required() -> bool: + """Return whether authentication is enforced for sensitive endpoints.""" + return os.getenv("AUTH_REQUIRED", "false").strip().lower() in {"1", "true", "yes", "on"} + + +@lru_cache(maxsize=4) +def _jwks_client(jwks_url: str) -> PyJWKClient: + return PyJWKClient(jwks_url) + + +def verify_bearer_token(token: str) -> dict[str, Any]: + """Verify an RS256 OIDC/Cognito JWT using issuer JWKS and configured audience.""" + issuer = os.getenv("AUTH_ISSUER", "").rstrip("/") + audience = os.getenv("AUTH_AUDIENCE", "") + token_use = os.getenv("AUTH_TOKEN_USE", "access") + + if not issuer or not audience: + raise RuntimeError("AUTH_ISSUER and AUTH_AUDIENCE are required when AUTH_REQUIRED=true") + + jwks_url = os.getenv("AUTH_JWKS_URL") or f"{issuer}/.well-known/jwks.json" + signing_key = _jwks_client(jwks_url).get_signing_key_from_jwt(token) + + try: + claims = jwt.decode( + token, + signing_key.key, + algorithms=["RS256"], + issuer=issuer, + options={"verify_aud": False}, + ) + except jwt.PyJWTError as exc: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Invalid bearer token.", + headers={"WWW-Authenticate": "Bearer"}, + ) from exc + + claimed_audience = claims.get("aud") or claims.get("client_id") + if claimed_audience != audience: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Bearer token audience is not authorized.", + headers={"WWW-Authenticate": "Bearer"}, + ) + + if token_use and claims.get("token_use") not in {None, token_use}: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Bearer token type is not authorized.", + headers={"WWW-Authenticate": "Bearer"}, + ) + + return claims + + +async def require_authenticated_request( + credentials: HTTPAuthorizationCredentials | None = Depends(_bearer), +) -> dict[str, Any] | None: + """Require a valid bearer token only when AUTH_REQUIRED is enabled.""" + if not auth_required(): + return None + + if credentials is None or credentials.scheme.lower() != "bearer": + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Bearer authentication is required.", + headers={"WWW-Authenticate": "Bearer"}, + ) + + return verify_bearer_token(credentials.credentials) diff --git a/consumer.py b/consumer.py index 37e9986..9281bea 100644 --- a/consumer.py +++ b/consumer.py @@ -1,56 +1,29 @@ -from models import DriverLocationEvent -from utils import get_logger - -logger = get_logger("DriverLocationConsumer") - - -class DriverLocationConsumer: - """ - Listens for incoming driver telemetry updates and stores them in - DriverLocationStore. - - Expected event format: - { - "driver_id": "d123", - "lat": 40.712, - "lon": -74.005, - "timestamp": "...", - "status": "available" - } - """ - - def __init__(self, event_bus, store): - self.event_bus = event_bus - self.store = store - self.logger = logger - - # ------------------------------------------------------------ - # Handle Incoming Driver Location Updates - # ------------------------------------------------------------ - - async def handle_driver_location(self, data: dict): - """ - Convert dictionary → Pydantic model → update store. - """ - - try: - event = DriverLocationEvent(**data) - except (TypeError, ValueError) as e: - self.logger.error(f"Invalid driver location event: {e}") - return - - # Update the driver store - self.store.upsert_driver(event) - - self.logger.info(f"Driver update processed: {event.driver_id} @ ({event.lat}, {event.lon})") - - # ------------------------------------------------------------ - # Subscribe to EventBus Topic - # ------------------------------------------------------------ - - async def start(self): - """ - Begins listening to driver_location_updates topic. - """ - self.logger.info("DriverLocationConsumer listening for driver updates...") - await self.event_bus.subscribe("driver_location_updates", self.handle_driver_location) +from models import DriverLocationEvent +from utils import get_logger + +logger = get_logger("DriverLocationConsumer") + + +class DriverLocationConsumer: + """Validate driver telemetry updates and store accepted events.""" + + def __init__(self, event_bus, store): + self.event_bus = event_bus + self.store = store + self.logger = logger + + async def handle_driver_location(self, data: dict): + """Convert a dictionary to the typed event model and update the store.""" + try: + event = DriverLocationEvent(**data) + except (TypeError, ValueError): + self.logger.error("Invalid driver location event rejected") + return + + self.store.upsert_driver(event) + self.logger.info("Driver update processed: %s", event.driver_id) + + async def start(self): + """Begin listening to the driver_location_updates topic.""" + self.logger.info("DriverLocationConsumer listening for driver updates...") + await self.event_bus.subscribe("driver_location_updates", self.handle_driver_location) diff --git a/event_bus.py b/event_bus.py index d198bba..b1087b6 100644 --- a/event_bus.py +++ b/event_bus.py @@ -1,49 +1,32 @@ -import asyncio -from collections import defaultdict -from collections.abc import Callable -from typing import Any - -from utils import get_logger - -logger = get_logger("EventBus") - - -class EventBus: - """ - A simple in-memory asynchronous event bus that provides: - - publish(topic, message) - - subscribe(topic, callback) - - In production this can be replaced with Kafka, Redis Streams, - NATS, RabbitMQ, AWS SNS/SQS, GCP PubSub, etc. - """ - - def __init__(self): - # topic → list of subscriber callback functions - self.subscribers: dict[str, list[Callable]] = defaultdict(list) - # lock to avoid race conditions when adding/removing subscribers - self.lock = asyncio.Lock() - - async def publish(self, topic: str, message: Any): - """ - Publish an event to a topic. - Invokes all subscribers concurrently. - """ - if topic not in self.subscribers: - logger.warning(f"[EVENT BUS] No subscribers for topic '{topic}'.") - return - - logger.info(f"[EVENT BUS] Publishing to {topic}: {message}") - - callbacks = list(self.subscribers[topic]) - - # Run all subscribers concurrently - await asyncio.gather(*(callback(message) for callback in callbacks)) - - async def subscribe(self, topic: str, callback: Callable): - """ - Register a subscriber callback for a topic. - """ - async with self.lock: - self.subscribers[topic].append(callback) - logger.info(f"[EVENT BUS] Subscriber added for topic '{topic}'.") +import asyncio +from collections import defaultdict +from collections.abc import Callable +from typing import Any + +from utils import get_logger + +logger = get_logger("EventBus") + + +class EventBus: + """Simple in-memory asynchronous publish/subscribe event bus.""" + + def __init__(self): + self.subscribers: dict[str, list[Callable]] = defaultdict(list) + self.lock = asyncio.Lock() + + async def publish(self, topic: str, message: Any): + """Publish an event without logging the payload contents.""" + if topic not in self.subscribers: + logger.warning("[EVENT BUS] No subscribers for topic '%s'.", topic) + return + + logger.info("[EVENT BUS] Publishing event to topic '%s'.", topic) + callbacks = list(self.subscribers[topic]) + await asyncio.gather(*(callback(message) for callback in callbacks)) + + async def subscribe(self, topic: str, callback: Callable): + """Register a subscriber callback for a topic.""" + async with self.lock: + self.subscribers[topic].append(callback) + logger.info("[EVENT BUS] Subscriber added for topic '%s'.", topic) diff --git a/main.py b/main.py index 0de5e83..6d1d7e4 100644 --- a/main.py +++ b/main.py @@ -8,16 +8,13 @@ from consumer import DriverLocationConsumer from event_bus import EventBus from location_store import create_driver_store +from runtime_secrets import resolve_driver_redis_url from utils import get_logger -# ------------------------------------------------------------ -# INITIALIZATION -# ------------------------------------------------------------ - logger = get_logger("DriverLocationMain") event_bus = EventBus() -driver_store = create_driver_store(os.getenv("DRIVER_LOCATION_REDIS_URL")) +driver_store = create_driver_store(resolve_driver_redis_url()) app = FastAPI( title="Driver Location Service", @@ -35,39 +32,18 @@ async def optional_instance_id_header(request, call_next): return response -# ------------------------------------------------------------ -# STARTUP SEQUENCE -# ------------------------------------------------------------ - - @app.on_event("startup") async def startup_event(): - logger.info("Starting Driver Location Service...") - - # Inject store into API router api_router.DRIVER_STORE = driver_store - - # Initialize consumer consumer = DriverLocationConsumer(event_bus=event_bus, store=driver_store) - - # Subscribe to driver location update events await event_bus.subscribe("driver_location_updates", consumer.handle_driver_location) - logger.info("Driver Location Service started successfully.") -# ------------------------------------------------------------ -# ROUTER -# ------------------------------------------------------------ - app.include_router(router, prefix="/driver-location") -# ------------------------------------------------------------ -# LOCAL RUNNER -# ------------------------------------------------------------ - if __name__ == "__main__": import uvicorn diff --git a/pii_policy.py b/pii_policy.py new file mode 100644 index 0000000..3dd4061 --- /dev/null +++ b/pii_policy.py @@ -0,0 +1,47 @@ +"""Explicit direct-PII policy for application and event boundaries.""" + +from typing import Any + +DIRECT_PII_KEYS = frozenset( + { + "email", + "email_address", + "phone", + "phone_number", + "full_name", + "first_name", + "last_name", + "street_address", + "home_address", + "ssn", + "social_security_number", + "card_number", + "cvv", + "bank_account", + "routing_number", + } +) + + +def direct_pii_paths(value: Any, prefix: str = "") -> list[str]: + """Return dotted paths whose key names are classified as direct PII.""" + matches: list[str] = [] + if isinstance(value, dict): + for key, nested in value.items(): + key_text = str(key) + path = f"{prefix}.{key_text}" if prefix else key_text + if key_text.lower() in DIRECT_PII_KEYS: + matches.append(path) + matches.extend(direct_pii_paths(nested, path)) + elif isinstance(value, list): + for index, nested in enumerate(value): + path = f"{prefix}[{index}]" if prefix else f"[{index}]" + matches.extend(direct_pii_paths(nested, path)) + return matches + + +def assert_no_direct_pii(value: Any) -> None: + """Reject payloads containing direct-identifier fields at the event boundary.""" + paths = direct_pii_paths(value) + if paths: + raise ValueError(f"Direct PII fields are not allowed in event payloads: {', '.join(paths)}") diff --git a/pyproject.toml b/pyproject.toml index be9e69f..02b5f0f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -19,6 +19,8 @@ dependencies = [ "python-json-logger>=2,<4", "redis>=5,<6", "prometheus-client>=0.20,<1.0", + "boto3>=1.34,<2.0", + "PyJWT[crypto]>=2.8,<3.0", ] [project.urls] @@ -29,6 +31,7 @@ Changelog = "https://github.com/CoreyLeath-code/Scalable-Event-Driven-Ride-Shari [tool.setuptools] py-modules = [ "api_router", + "auth", "consumer", "event_bus", "kafka_bus", @@ -36,10 +39,12 @@ py-modules = [ "main", "matching_engine", "models", + "pii_policy", "pricing_engine", "producer", "rabbitmq_bus", "redis_bus", + "runtime_secrets", "utils", ] @@ -54,9 +59,10 @@ line-length = 100 pythonpath = ["."] testpaths = [".", "tests"] python_files = ["test_*.py"] -addopts = "-ra -m 'not integration'" +addopts = "-ra -m 'not integration and not aws_integration'" markers = [ "integration: requires Docker-hosted Kafka, Redis, and RabbitMQ services", + "aws_integration: requires a real AWS development environment and VPC network access", ] [tool.coverage.run] diff --git a/requirements-dev.txt b/requirements-dev.txt index a9cf48a..40accf8 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -13,6 +13,9 @@ pytest-asyncio aiokafka>=0.11,<1.0 aio-pika>=9,<10 +# Real Amazon MSK integration harness (manual/VPC runner only) +aws-msk-iam-sasl-signer-python>=1.0,<2.0 + # Type-checking types-requests diff --git a/requirements.txt b/requirements.txt index c7eee1e..82a6dff 100644 --- a/requirements.txt +++ b/requirements.txt @@ -6,3 +6,5 @@ kafka-python>=2.0,<3.0 python-json-logger>=2,<4 redis>=5,<6 prometheus-client>=0.20,<1.0 +boto3>=1.34,<2.0 +PyJWT[crypto]>=2.8,<3.0 diff --git a/runtime_secrets.py b/runtime_secrets.py new file mode 100644 index 0000000..bee383e --- /dev/null +++ b/runtime_secrets.py @@ -0,0 +1,42 @@ +"""Runtime secret resolution with AWS Secrets Manager and local-development fallback.""" + +import json +import os +from typing import Any + + +def _default_secrets_client(): + import boto3 + + return boto3.client("secretsmanager") + + +def load_secret_json(secret_id: str, *, client: Any = None) -> dict[str, Any]: + """Fetch a JSON secret without logging or returning AWS response metadata.""" + secrets_client = client or _default_secrets_client() + response = secrets_client.get_secret_value(SecretId=secret_id) + secret_string = response.get("SecretString") + if not isinstance(secret_string, str) or not secret_string: + raise RuntimeError("Secrets Manager secret must contain a non-empty SecretString") + + try: + payload = json.loads(secret_string) + except json.JSONDecodeError as exc: + raise RuntimeError("Secrets Manager secret must contain valid JSON") from exc + + if not isinstance(payload, dict): + raise RuntimeError("Secrets Manager secret JSON must be an object") + return payload + + +def resolve_driver_redis_url(*, client: Any = None) -> str | None: + """Resolve Redis URL from Secrets Manager when configured, otherwise local env.""" + secret_id = os.getenv("DRIVER_LOCATION_REDIS_SECRET_ID") + if not secret_id: + return os.getenv("DRIVER_LOCATION_REDIS_URL") + + secret = load_secret_json(secret_id, client=client) + redis_url = secret.get("redis_url") + if not isinstance(redis_url, str) or not redis_url.strip(): + raise RuntimeError("Runtime secret requires a non-empty redis_url field") + return redis_url From e869b2e6cd6c798ed9557aea4d3a5f8dd5676053 Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Mon, 31 Aug 2026 22:19:35 -0400 Subject: [PATCH 02/21] test: harden broker and PII boundaries --- kafka_bus.py | 94 +++++++++---------- rabbitmq_bus.py | 84 +++++++++-------- redis_bus.py | 78 +++++++-------- serverless/notification_worker/handler.py | 8 +- serverless/notification_worker/pii_policy.py | 45 +++++++++ serverless/ride_event_processor/handler.py | 15 +-- serverless/ride_event_processor/pii_policy.py | 45 +++++++++ tests/test_aws_security_boundaries.py | 63 +++++++++++++ tests/test_broker_contracts.py | 30 ++++++ tests/test_security_boundaries.py | 68 ++++++++++++++ 10 files changed, 397 insertions(+), 133 deletions(-) create mode 100644 serverless/notification_worker/pii_policy.py create mode 100644 serverless/ride_event_processor/pii_policy.py create mode 100644 tests/test_aws_security_boundaries.py create mode 100644 tests/test_broker_contracts.py create mode 100644 tests/test_security_boundaries.py diff --git a/kafka_bus.py b/kafka_bus.py index 2b6ed2c..e5e00c7 100644 --- a/kafka_bus.py +++ b/kafka_bus.py @@ -1,49 +1,45 @@ -import json - -from aiokafka import AIOKafkaConsumer, AIOKafkaProducer - -from .base import EventBus - - -class KafkaEventBus(EventBus): - """ - Kafka implementation of the EventBus interface. - Uses aiokafka for async publish/subscribe. - """ - - def __init__(self, bootstrap_servers="localhost:9092", group_id="ride-sharing"): - self.bootstrap_servers = bootstrap_servers - self.group_id = group_id - self.producer = None - self.consumer = None - - async def connect(self): - self.producer = AIOKafkaProducer( - bootstrap_servers=self.bootstrap_servers, - value_serializer=lambda v: json.dumps(v).encode("utf-8"), - ) - await self.producer.start() - - async def publish(self, topic: str, message: dict): - if not self.producer: - raise RuntimeError("Kafka producer not initialized. Call connect().") - await self.producer.send_and_wait(topic, message) - - async def subscribe(self, topic: str, handler): - consumer = AIOKafkaConsumer( - topic, - bootstrap_servers=self.bootstrap_servers, - group_id=self.group_id, - value_deserializer=lambda v: json.loads(v.decode("utf-8")), - ) - await consumer.start() - - try: - async for msg in consumer: - await handler(msg.value) - finally: - await consumer.stop() - - async def close(self): - if self.producer: - await self.producer.stop() +import json + +from aiokafka import AIOKafkaConsumer, AIOKafkaProducer + +from event_bus import EventBus + + +class KafkaEventBus(EventBus): + """Kafka implementation using aiokafka for async publish/subscribe.""" + + def __init__(self, bootstrap_servers="localhost:9092", group_id="ride-sharing"): + self.bootstrap_servers = bootstrap_servers + self.group_id = group_id + self.producer = None + + async def connect(self): + self.producer = AIOKafkaProducer( + bootstrap_servers=self.bootstrap_servers, + value_serializer=lambda value: json.dumps(value).encode("utf-8"), + ) + await self.producer.start() + + async def publish(self, topic: str, message: dict): + if not self.producer: + raise RuntimeError("Kafka producer not initialized. Call connect().") + await self.producer.send_and_wait(topic, message) + + async def subscribe(self, topic: str, handler): + consumer = AIOKafkaConsumer( + topic, + bootstrap_servers=self.bootstrap_servers, + group_id=self.group_id, + value_deserializer=lambda value: json.loads(value.decode("utf-8")), + ) + await consumer.start() + try: + async for message in consumer: + await handler(message.value) + finally: + await consumer.stop() + + async def close(self): + if self.producer: + await self.producer.stop() + self.producer = None diff --git a/rabbitmq_bus.py b/rabbitmq_bus.py index 2c5adc3..ce4982e 100644 --- a/rabbitmq_bus.py +++ b/rabbitmq_bus.py @@ -1,39 +1,45 @@ -import json - -import aio_pika - -from .base import EventBus - - -class RabbitMQEventBus(EventBus): - """ - RabbitMQ implementation using aio-pika. - Queue-based, good for job/task style processing. - """ - - def __init__(self, url="amqp://guest:guest@localhost/"): - self.url = url - self.connection = None - self.channel = None - - async def connect(self): - self.connection = await aio_pika.connect_robust(self.url) - self.channel = await self.connection.channel() - - async def publish(self, topic: str, message: dict): - queue = await self.channel.declare_queue(topic, durable=True) - await self.channel.default_exchange.publish( - aio_pika.Message(body=json.dumps(message).encode()), routing_key=queue.name - ) - - async def subscribe(self, topic: str, handler): - queue = await self.channel.declare_queue(topic, durable=True) - - async with queue.iterator() as queue_iter: - async for msg in queue_iter: - async with msg.process(): - await handler(json.loads(msg.body.decode())) - - async def close(self): - if self.connection: - await self.connection.close() +import json + +import aio_pika + +from event_bus import EventBus + + +class RabbitMQEventBus(EventBus): + """RabbitMQ event-bus implementation using aio-pika.""" + + def __init__(self, url="amqp://guest:guest@localhost/"): + self.url = url + self.connection = None + self.channel = None + + async def connect(self): + self.connection = await aio_pika.connect_robust(self.url) + self.channel = await self.connection.channel() + + def _channel(self): + if self.channel is None: + raise RuntimeError("RabbitMQ channel not initialized. Call connect().") + return self.channel + + async def publish(self, topic: str, message: dict): + channel = self._channel() + queue = await channel.declare_queue(topic, durable=True) + await channel.default_exchange.publish( + aio_pika.Message(body=json.dumps(message).encode()), routing_key=queue.name + ) + + async def subscribe(self, topic: str, handler): + channel = self._channel() + queue = await channel.declare_queue(topic, durable=True) + + async with queue.iterator() as queue_iter: + async for message in queue_iter: + async with message.process(): + await handler(json.loads(message.body.decode())) + + async def close(self): + if self.connection: + await self.connection.close() + self.connection = None + self.channel = None diff --git a/redis_bus.py b/redis_bus.py index c2d4d82..3df7a1f 100644 --- a/redis_bus.py +++ b/redis_bus.py @@ -1,37 +1,41 @@ -import json - -import redis.asyncio as redis - -from .base import EventBus - - -class RedisEventBus(EventBus): - """ - Redis Streams based implementation. - Great for lightweight event-driven pipelines. - """ - - def __init__(self, redis_url="redis://localhost:6379"): - self.redis_url = redis_url - self.redis = None - - async def connect(self): - self.redis = redis.from_url(self.redis_url, decode_responses=True) - - async def publish(self, topic: str, message: dict): - await self.redis.xadd(topic, {"data": json.dumps(message)}) - - async def subscribe(self, topic: str, handler): - last_id = "$" - while True: - streams = await self.redis.xread({topic: last_id}, block=5000) - if streams: - _, messages = streams[0] - for msg_id, fields in messages: - last_id = msg_id - data = json.loads(fields["data"]) - await handler(data) - - async def close(self): - if self.redis: - await self.redis.aclose() +import json + +import redis.asyncio as redis + +from event_bus import EventBus + + +class RedisEventBus(EventBus): + """Redis Streams event-bus implementation.""" + + def __init__(self, redis_url="redis://localhost:6379"): + self.redis_url = redis_url + self.redis = None + + async def connect(self): + self.redis = redis.from_url(self.redis_url, decode_responses=True) + + def _client(self): + if self.redis is None: + raise RuntimeError("Redis client not initialized. Call connect().") + return self.redis + + async def publish(self, topic: str, message: dict): + await self._client().xadd(topic, {"data": json.dumps(message)}) + + async def subscribe(self, topic: str, handler): + client = self._client() + last_id = "$" + while True: + streams = await client.xread({topic: last_id}, block=5000) + if streams: + _, messages = streams[0] + for message_id, fields in messages: + last_id = message_id + data = json.loads(fields["data"]) + await handler(data) + + async def close(self): + if self.redis: + await self.redis.aclose() + self.redis = None diff --git a/serverless/notification_worker/handler.py b/serverless/notification_worker/handler.py index 1878503..a166282 100644 --- a/serverless/notification_worker/handler.py +++ b/serverless/notification_worker/handler.py @@ -5,6 +5,11 @@ import os from typing import Any +try: + from .pii_policy import assert_no_direct_pii +except ImportError: # Lambda ZIP imports handler as a top-level module. + from pii_policy import assert_no_direct_pii + LOGGER = logging.getLogger(__name__) LOGGER.setLevel(logging.INFO) @@ -17,7 +22,7 @@ def _default_sns_client(): - import boto3 # type: ignore[import-not-found] + import boto3 return boto3.client("sns") @@ -35,6 +40,7 @@ def _extract_event(body: str) -> dict[str, Any]: if not isinstance(payload, dict): raise TypeError("Notification payload must be a JSON object") + assert_no_direct_pii(payload) return payload diff --git a/serverless/notification_worker/pii_policy.py b/serverless/notification_worker/pii_policy.py new file mode 100644 index 0000000..3a5b5f0 --- /dev/null +++ b/serverless/notification_worker/pii_policy.py @@ -0,0 +1,45 @@ +"""Direct-PII guard bundled with the notification Lambda ZIP.""" + +from typing import Any + +DIRECT_PII_KEYS = frozenset( + { + "email", + "email_address", + "phone", + "phone_number", + "full_name", + "first_name", + "last_name", + "street_address", + "home_address", + "ssn", + "social_security_number", + "card_number", + "cvv", + "bank_account", + "routing_number", + } +) + + +def _paths(value: Any, prefix: str = "") -> list[str]: + matches: list[str] = [] + if isinstance(value, dict): + for key, nested in value.items(): + key_text = str(key) + path = f"{prefix}.{key_text}" if prefix else key_text + if key_text.lower() in DIRECT_PII_KEYS: + matches.append(path) + matches.extend(_paths(nested, path)) + elif isinstance(value, list): + for index, nested in enumerate(value): + path = f"{prefix}[{index}]" if prefix else f"[{index}]" + matches.extend(_paths(nested, path)) + return matches + + +def assert_no_direct_pii(value: Any) -> None: + matches = _paths(value) + if matches: + raise ValueError("Direct PII is not allowed in the asynchronous notification pipeline") diff --git a/serverless/ride_event_processor/handler.py b/serverless/ride_event_processor/handler.py index f9a6fd1..689f83d 100644 --- a/serverless/ride_event_processor/handler.py +++ b/serverless/ride_event_processor/handler.py @@ -6,6 +6,11 @@ import os from typing import Any +try: + from .pii_policy import assert_no_direct_pii +except ImportError: # Lambda ZIP imports handler as a top-level module. + from pii_policy import assert_no_direct_pii + LOGGER = logging.getLogger(__name__) LOGGER.setLevel(logging.INFO) @@ -29,6 +34,7 @@ def _decode_record(record: dict[str, Any]) -> dict[str, Any]: if not isinstance(event_type, str) or not event_type.strip(): raise ValueError("MSK event payload requires event_type or type") + assert_no_direct_pii(payload) return payload @@ -49,7 +55,7 @@ def _build_envelope(record: dict[str, Any], payload: dict[str, Any]) -> dict[str def _default_sqs_client(): - import boto3 # type: ignore[import-not-found] + import boto3 return boto3.client("sqs") @@ -57,12 +63,7 @@ def _default_sqs_client(): def lambda_handler( event: dict[str, Any], _context: Any, *, sqs_client: Any = None ) -> dict[str, int]: - """Forward an MSK Lambda batch to the configured SQS queue. - - The function intentionally fails the invocation on malformed input or an SQS send - failure. Amazon MSK/Lambda can then retry the batch. Consumers must therefore use the - emitted idempotency_key because delivery is at-least-once. - """ + """Forward an MSK Lambda batch to the configured SQS queue.""" queue_url = os.environ.get("PROCESSED_EVENTS_QUEUE_URL") if not queue_url: raise RuntimeError("PROCESSED_EVENTS_QUEUE_URL is required") diff --git a/serverless/ride_event_processor/pii_policy.py b/serverless/ride_event_processor/pii_policy.py new file mode 100644 index 0000000..77b27a1 --- /dev/null +++ b/serverless/ride_event_processor/pii_policy.py @@ -0,0 +1,45 @@ +"""Direct-PII guard bundled with the ride-event Lambda ZIP.""" + +from typing import Any + +DIRECT_PII_KEYS = frozenset( + { + "email", + "email_address", + "phone", + "phone_number", + "full_name", + "first_name", + "last_name", + "street_address", + "home_address", + "ssn", + "social_security_number", + "card_number", + "cvv", + "bank_account", + "routing_number", + } +) + + +def _paths(value: Any, prefix: str = "") -> list[str]: + matches: list[str] = [] + if isinstance(value, dict): + for key, nested in value.items(): + key_text = str(key) + path = f"{prefix}.{key_text}" if prefix else key_text + if key_text.lower() in DIRECT_PII_KEYS: + matches.append(path) + matches.extend(_paths(nested, path)) + elif isinstance(value, list): + for index, nested in enumerate(value): + path = f"{prefix}[{index}]" if prefix else f"[{index}]" + matches.extend(_paths(nested, path)) + return matches + + +def assert_no_direct_pii(value: Any) -> None: + matches = _paths(value) + if matches: + raise ValueError("Direct PII is not allowed in the asynchronous ride-event pipeline") diff --git a/tests/test_aws_security_boundaries.py b/tests/test_aws_security_boundaries.py new file mode 100644 index 0000000..4c26fb3 --- /dev/null +++ b/tests/test_aws_security_boundaries.py @@ -0,0 +1,63 @@ +import base64 +import json + +import pytest + +from serverless.notification_worker import handler as notification_handler +from serverless.ride_event_processor import handler as ride_handler + + +class RecordingSqs: + def __init__(self): + self.messages = [] + + def send_message(self, **kwargs): + self.messages.append(kwargs) + return {"MessageId": "m-1"} + + +class RecordingSns: + def __init__(self): + self.messages = [] + + def publish(self, **kwargs): + self.messages.append(kwargs) + return {"MessageId": "n-1"} + + +def msk_event(payload): + value = base64.b64encode(json.dumps(payload).encode()).decode() + return { + "records": { + "ride.events-0": [ + {"topic": "ride.events", "partition": 0, "offset": 1, "value": value} + ] + } + } + + +def test_msk_processor_rejects_direct_pii(monkeypatch): + monkeypatch.setenv("PROCESSED_EVENTS_QUEUE_URL", "https://sqs.example/queue") + sqs = RecordingSqs() + with pytest.raises(ValueError, match="Direct PII"): + ride_handler.lambda_handler( + msk_event({"event_type": "ride.requested", "email": "rider@example.com"}), + None, + sqs_client=sqs, + ) + assert sqs.messages == [] + + +def test_notification_worker_returns_partial_failure_for_direct_pii(monkeypatch): + monkeypatch.setenv("NOTIFICATION_TOPIC_ARN", "arn:aws:sns:us-east-1:123:topic") + sns = RecordingSns() + body = json.dumps( + {"event": {"event_type": "trip.completed", "phone_number": "+15555550100"}} + ) + result = notification_handler.lambda_handler( + {"Records": [{"messageId": "bad-pii", "body": body}]}, + None, + sns_client=sns, + ) + assert result["batchItemFailures"] == [{"itemIdentifier": "bad-pii"}] + assert sns.messages == [] diff --git a/tests/test_broker_contracts.py b/tests/test_broker_contracts.py new file mode 100644 index 0000000..6ceeaa6 --- /dev/null +++ b/tests/test_broker_contracts.py @@ -0,0 +1,30 @@ +import pytest + +from kafka_bus import KafkaEventBus +from rabbitmq_bus import RabbitMQEventBus +from redis_bus import RedisEventBus + + +@pytest.mark.asyncio +async def test_kafka_publish_requires_connect(): + with pytest.raises(RuntimeError, match="Call connect"): + await KafkaEventBus().publish("rides", {"event_type": "ride.requested"}) + + +@pytest.mark.asyncio +async def test_redis_publish_requires_connect(): + with pytest.raises(RuntimeError, match="Call connect"): + await RedisEventBus().publish("rides", {"event_type": "ride.requested"}) + + +@pytest.mark.asyncio +async def test_rabbitmq_publish_requires_connect(): + with pytest.raises(RuntimeError, match="Call connect"): + await RabbitMQEventBus().publish("rides", {"event_type": "ride.requested"}) + + +@pytest.mark.asyncio +async def test_broker_close_is_safe_before_connect(): + await KafkaEventBus().close() + await RedisEventBus().close() + await RabbitMQEventBus().close() diff --git a/tests/test_security_boundaries.py b/tests/test_security_boundaries.py new file mode 100644 index 0000000..082960e --- /dev/null +++ b/tests/test_security_boundaries.py @@ -0,0 +1,68 @@ +import json + +import pytest +from fastapi import HTTPException +from fastapi.security import HTTPAuthorizationCredentials + +import auth +from pii_policy import assert_no_direct_pii, direct_pii_paths +from runtime_secrets import load_secret_json, resolve_driver_redis_url + + +class FakeSecretsClient: + def __init__(self, payload): + self.payload = payload + self.secret_ids = [] + + def get_secret_value(self, SecretId): + self.secret_ids.append(SecretId) + return {"SecretString": json.dumps(self.payload)} + + +@pytest.mark.asyncio +async def test_auth_is_credential_free_when_disabled(monkeypatch): + monkeypatch.setenv("AUTH_REQUIRED", "false") + assert await auth.require_authenticated_request(None) is None + + +@pytest.mark.asyncio +async def test_auth_rejects_missing_bearer_when_enabled(monkeypatch): + monkeypatch.setenv("AUTH_REQUIRED", "true") + with pytest.raises(HTTPException) as exc_info: + await auth.require_authenticated_request(None) + assert exc_info.value.status_code == 401 + + +@pytest.mark.asyncio +async def test_auth_delegates_token_verification_without_network(monkeypatch): + monkeypatch.setenv("AUTH_REQUIRED", "true") + monkeypatch.setattr(auth, "verify_bearer_token", lambda token: {"sub": token}) + credentials = HTTPAuthorizationCredentials(scheme="Bearer", credentials="test-token") + assert await auth.require_authenticated_request(credentials) == {"sub": "test-token"} + + +def test_secrets_manager_runtime_resolution_without_aws_credentials(monkeypatch): + monkeypatch.setenv("DRIVER_LOCATION_REDIS_SECRET_ID", "ride-sharing/dev/runtime") + client = FakeSecretsClient({"redis_url": "redis://private.example:6379/0"}) + assert resolve_driver_redis_url(client=client) == "redis://private.example:6379/0" + assert client.secret_ids == ["ride-sharing/dev/runtime"] + + +def test_secret_loader_rejects_non_json_secret(): + class BadClient: + def get_secret_value(self, SecretId): + return {"SecretString": "not-json"} + + with pytest.raises(RuntimeError, match="valid JSON"): + load_secret_json("secret-id", client=BadClient()) + + +def test_pii_policy_detects_nested_direct_identifiers(): + payload = {"ride_id": "ride-1", "contact": {"email": "rider@example.com"}} + assert direct_pii_paths(payload) == ["contact.email"] + with pytest.raises(ValueError, match="Direct PII fields"): + assert_no_direct_pii(payload) + + +def test_pii_policy_allows_pseudonymous_operational_identifiers(): + assert_no_direct_pii({"ride_id": "ride-1", "driver_id": "driver-7", "probe_id": "p-1"}) From e71d248a410849c2e00199316da1ff5b8c24ca6a Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Mon, 31 Aug 2026 22:21:27 -0400 Subject: [PATCH 03/21] feat: add optional real MSK dev environment --- infra/aws/dev_msk.tf | 88 ++++++++++++++++++++++++++++++++++++++++++ infra/aws/main.tf | 23 +++++------ infra/aws/outputs.tf | 22 ++++++++++- infra/aws/security.tf | 46 ++++++++++++++++++++++ infra/aws/variables.tf | 50 +++++++++++++++++++++++- 5 files changed, 216 insertions(+), 13 deletions(-) create mode 100644 infra/aws/dev_msk.tf create mode 100644 infra/aws/security.tf diff --git a/infra/aws/dev_msk.tf b/infra/aws/dev_msk.tf new file mode 100644 index 0000000..bcf68e4 --- /dev/null +++ b/infra/aws/dev_msk.tf @@ -0,0 +1,88 @@ +resource "aws_security_group" "dev_msk" { + count = var.create_dev_msk_cluster ? 1 : 0 + + name_prefix = "${var.name_prefix}-dev-msk-" + description = "Private security group for the optional development MSK Serverless cluster." + vpc_id = var.dev_msk_vpc_id + + ingress { + description = "IAM-authenticated Kafka traffic within the cluster security group" + from_port = 9098 + to_port = 9098 + protocol = "tcp" + self = true + } + + egress { + from_port = 0 + to_port = 0 + protocol = "-1" + cidr_blocks = ["0.0.0.0/0"] + } +} + +resource "aws_msk_serverless_cluster" "dev" { + count = var.create_dev_msk_cluster ? 1 : 0 + + cluster_name = "${var.name_prefix}-dev" + + vpc_config { + subnet_ids = var.dev_msk_subnet_ids + security_group_ids = [aws_security_group.dev_msk[0].id] + } + + client_authentication { + sasl { + iam { + enabled = true + } + } + } +} + +resource "aws_sqs_queue" "integration_probe" { + count = var.enable_integration_probe ? 1 : 0 + + name = "${var.name_prefix}-integration-probe" + message_retention_seconds = 3600 +} + +data "aws_iam_policy_document" "integration_probe" { + count = var.enable_integration_probe ? 1 : 0 + + statement { + sid = "AllowRideNotificationTopic" + actions = ["sqs:SendMessage"] + resources = [ + aws_sqs_queue.integration_probe[0].arn, + ] + + principals { + type = "Service" + identifiers = ["sns.amazonaws.com"] + } + + condition { + test = "ArnEquals" + variable = "aws:SourceArn" + values = [aws_sns_topic.ride_notifications.arn] + } + } +} + +resource "aws_sqs_queue_policy" "integration_probe" { + count = var.enable_integration_probe ? 1 : 0 + + queue_url = aws_sqs_queue.integration_probe[0].url + policy = data.aws_iam_policy_document.integration_probe[0].json +} + +resource "aws_sns_topic_subscription" "integration_probe" { + count = var.enable_integration_probe ? 1 : 0 + + topic_arn = aws_sns_topic.ride_notifications.arn + protocol = "sqs" + endpoint = aws_sqs_queue.integration_probe[0].arn + + depends_on = [aws_sqs_queue_policy.integration_probe] +} diff --git a/infra/aws/main.tf b/infra/aws/main.tf index a1dab9d..ae3c6e8 100644 --- a/infra/aws/main.tf +++ b/infra/aws/main.tf @@ -1,13 +1,14 @@ locals { - processor_function_name = "${var.name_prefix}-ride-event-processor" - notification_function_name = "${var.name_prefix}-notification-worker" - msk_topic_arn = var.msk_cluster_arn == null ? null : "${replace( - var.msk_cluster_arn, + processor_function_name = "${var.name_prefix}-ride-event-processor" + notification_function_name = "${var.name_prefix}-notification-worker" + effective_msk_cluster_arn = var.msk_cluster_arn != null ? var.msk_cluster_arn : try(aws_msk_serverless_cluster.dev[0].arn, null) + msk_topic_arn = local.effective_msk_cluster_arn == null ? null : "${replace( + local.effective_msk_cluster_arn, ":cluster/", ":topic/" )}/${var.ride_events_topic}" - msk_group_arn = var.msk_cluster_arn == null ? null : "${replace( - var.msk_cluster_arn, + msk_group_arn = local.effective_msk_cluster_arn == null ? null : "${replace( + local.effective_msk_cluster_arn, ":cluster/", ":group/" )}/${var.consumer_group_id}" @@ -84,7 +85,7 @@ data "aws_iam_policy_document" "processor_permissions" { } dynamic "statement" { - for_each = var.msk_cluster_arn == null ? [] : [var.msk_cluster_arn] + for_each = local.effective_msk_cluster_arn == null ? [] : [local.effective_msk_cluster_arn] content { sid = "DiscoverMskCluster" @@ -98,7 +99,7 @@ data "aws_iam_policy_document" "processor_permissions" { } dynamic "statement" { - for_each = var.msk_cluster_arn == null ? [] : [var.msk_cluster_arn] + for_each = local.effective_msk_cluster_arn == null ? [] : [local.effective_msk_cluster_arn] content { sid = "ConnectToMskCluster" @@ -134,7 +135,7 @@ data "aws_iam_policy_document" "processor_permissions" { } dynamic "statement" { - for_each = var.msk_cluster_arn == null ? [] : [1] + for_each = local.effective_msk_cluster_arn == null ? [] : [1] content { sid = "ManageMskNetworkInterfaces" @@ -250,9 +251,9 @@ resource "aws_lambda_event_source_mapping" "processed_events" { } resource "aws_lambda_event_source_mapping" "ride_events_msk" { - count = var.msk_cluster_arn == null ? 0 : 1 + count = local.effective_msk_cluster_arn == null ? 0 : 1 - event_source_arn = var.msk_cluster_arn + event_source_arn = local.effective_msk_cluster_arn function_name = aws_lambda_function.ride_event_processor.arn topics = [var.ride_events_topic] starting_position = "LATEST" diff --git a/infra/aws/outputs.tf b/infra/aws/outputs.tf index 3b6c9c1..9a7d561 100644 --- a/infra/aws/outputs.tf +++ b/infra/aws/outputs.tf @@ -33,7 +33,27 @@ output "cloudwatch_dashboard_name" { value = aws_cloudwatch_dashboard.serverless.dashboard_name } +output "effective_msk_cluster_arn" { + description = "ARN of the existing or optionally created development MSK cluster used by the Lambda event-source mapping." + value = local.effective_msk_cluster_arn +} + output "msk_event_source_enabled" { description = "Whether Terraform creates the Amazon MSK to Lambda event-source mapping." - value = var.msk_cluster_arn != null + value = local.effective_msk_cluster_arn != null +} + +output "integration_probe_queue_url" { + description = "SQS probe queue used to measure the real MSK-to-SNS event path when enabled." + value = try(aws_sqs_queue.integration_probe[0].url, null) +} + +output "runtime_secret_arn" { + description = "ARN of the optional KMS-protected runtime secret metadata resource." + value = try(aws_secretsmanager_secret.runtime[0].arn, null) +} + +output "runtime_secret_reader_policy_arn" { + description = "ARN of the optional least-privilege policy for reading the runtime secret." + value = try(aws_iam_policy.runtime_secret_reader[0].arn, null) } diff --git a/infra/aws/security.tf b/infra/aws/security.tf new file mode 100644 index 0000000..3d03c1e --- /dev/null +++ b/infra/aws/security.tf @@ -0,0 +1,46 @@ +resource "aws_kms_key" "runtime_secrets" { + count = var.create_runtime_secret ? 1 : 0 + + description = "KMS key for ride-sharing runtime secrets." + enable_key_rotation = true + deletion_window_in_days = 7 +} + +resource "aws_kms_alias" "runtime_secrets" { + count = var.create_runtime_secret ? 1 : 0 + + name = "alias/${var.name_prefix}-runtime-secrets" + target_key_id = aws_kms_key.runtime_secrets[0].key_id +} + +resource "aws_secretsmanager_secret" "runtime" { + count = var.create_runtime_secret ? 1 : 0 + + name = var.runtime_secret_name != null ? var.runtime_secret_name : "${var.name_prefix}/dev/runtime" + description = "Runtime configuration secret metadata. Populate values out-of-band; Terraform does not store secret material." + kms_key_id = aws_kms_key.runtime_secrets[0].arn + recovery_window_in_days = 7 +} + +data "aws_iam_policy_document" "runtime_secret_reader" { + count = var.create_runtime_secret ? 1 : 0 + + statement { + sid = "ReadRuntimeSecret" + actions = ["secretsmanager:GetSecretValue"] + resources = [aws_secretsmanager_secret.runtime[0].arn] + } + + statement { + sid = "DecryptRuntimeSecret" + actions = ["kms:Decrypt"] + resources = [aws_kms_key.runtime_secrets[0].arn] + } +} + +resource "aws_iam_policy" "runtime_secret_reader" { + count = var.create_runtime_secret ? 1 : 0 + + name = "${var.name_prefix}-runtime-secret-reader" + policy = data.aws_iam_policy_document.runtime_secret_reader[0].json +} diff --git a/infra/aws/variables.tf b/infra/aws/variables.tf index 166bf22..3d39762 100644 --- a/infra/aws/variables.tf +++ b/infra/aws/variables.tf @@ -11,12 +11,47 @@ variable "name_prefix" { } variable "msk_cluster_arn" { - description = "Optional Amazon MSK cluster ARN. When null, the MSK event source mapping is not created." + description = "Optional existing Amazon MSK cluster ARN. When null, Terraform can optionally create a development MSK Serverless cluster." type = string default = null nullable = true } +variable "create_dev_msk_cluster" { + description = "Create a private IAM-authenticated MSK Serverless cluster for development integration measurements. Disabled by default because it incurs AWS cost." + type = bool + default = false +} + +variable "dev_msk_vpc_id" { + description = "VPC ID used by the optional development MSK Serverless cluster." + type = string + default = null + nullable = true + + validation { + condition = !var.create_dev_msk_cluster || var.dev_msk_vpc_id != null + error_message = "dev_msk_vpc_id is required when create_dev_msk_cluster=true." + } +} + +variable "dev_msk_subnet_ids" { + description = "Private subnet IDs used by the optional development MSK Serverless cluster." + type = list(string) + default = [] + + validation { + condition = !var.create_dev_msk_cluster || length(var.dev_msk_subnet_ids) >= 2 + error_message = "At least two private subnet IDs are required when create_dev_msk_cluster=true." + } +} + +variable "enable_integration_probe" { + description = "Create an SNS-subscribed SQS probe queue for real end-to-end MSK integration measurements." + type = bool + default = false +} + variable "ride_events_topic" { description = "Kafka topic consumed by the ride-event processor Lambda." type = string @@ -85,3 +120,16 @@ variable "sqs_oldest_message_age_seconds" { error_message = "sqs_oldest_message_age_seconds must be at least 60 seconds." } } + +variable "create_runtime_secret" { + description = "Create KMS-protected Secrets Manager secret metadata and a least-privilege reader policy. Secret values are never stored in Terraform." + type = bool + default = false +} + +variable "runtime_secret_name" { + description = "Optional Secrets Manager name for application runtime configuration." + type = string + default = null + nullable = true +} From 7a6b33a72e5763e677b07338fd629220c8eb43ee Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Mon, 31 Aug 2026 22:22:51 -0400 Subject: [PATCH 04/21] feat: add real MSK evidence harness --- .github/workflows/aws-msk-integration.yml | 109 ++++++++++++ .github/workflows/aws-serverless.yml | 20 ++- evidence/aws-msk-integration-results.json | 7 + scripts/aws_msk_e2e.py | 201 ++++++++++++++++++++++ scripts/k8s_release_evidence.py | 76 ++++++++ tests/test_release_evidence_tools.py | 59 +++++++ 6 files changed, 468 insertions(+), 4 deletions(-) create mode 100644 .github/workflows/aws-msk-integration.yml create mode 100644 evidence/aws-msk-integration-results.json create mode 100644 scripts/aws_msk_e2e.py create mode 100644 scripts/k8s_release_evidence.py create mode 100644 tests/test_release_evidence_tools.py diff --git a/.github/workflows/aws-msk-integration.yml b/.github/workflows/aws-msk-integration.yml new file mode 100644 index 0000000..1cb791e --- /dev/null +++ b/.github/workflows/aws-msk-integration.yml @@ -0,0 +1,109 @@ +name: Real AWS MSK Integration Evidence + +on: + workflow_dispatch: + inputs: + aws_region: + description: AWS region containing the development VPC + required: true + default: us-east-1 + type: string + vpc_id: + description: Development VPC ID + required: true + type: string + subnet_ids_json: + description: JSON array of at least two private subnet IDs + required: true + type: string + samples: + description: Number of sequential end-to-end probes + required: true + default: "20" + type: string + destroy_after: + description: Destroy the temporary MSK integration environment after evidence collection + required: true + default: true + type: boolean + +permissions: + contents: read + id-token: write + +jobs: + real-msk-evidence: + name: Real MSK end-to-end evidence + runs-on: [self-hosted, linux, aws-msk-dev] + environment: development + timeout-minutes: 90 + env: + AWS_REGION: ${{ inputs.aws_region }} + TF_VAR_aws_region: ${{ inputs.aws_region }} + TF_VAR_create_dev_msk_cluster: "true" + TF_VAR_enable_integration_probe: "true" + TF_VAR_dev_msk_vpc_id: ${{ inputs.vpc_id }} + TF_VAR_dev_msk_subnet_ids: ${{ inputs.subnet_ids_json }} + LAMBDA_REQUESTS_PER_MILLION_USD: ${{ vars.LAMBDA_REQUESTS_PER_MILLION_USD }} + SQS_REQUESTS_PER_MILLION_USD: ${{ vars.SQS_REQUESTS_PER_MILLION_USD }} + SNS_PUBLISHES_PER_MILLION_USD: ${{ vars.SNS_PUBLISHES_PER_MILLION_USD }} + MSK_CLUSTER_HOURLY_USD: ${{ vars.MSK_CLUSTER_HOURLY_USD }} + + steps: + - name: Checkout repository + uses: actions/checkout@v4 + + - name: Configure AWS credentials with OIDC + uses: aws-actions/configure-aws-credentials@v4 + with: + role-to-assume: ${{ secrets.AWS_MSK_INTEGRATION_ROLE_ARN }} + aws-region: ${{ inputs.aws_region }} + + - name: Set up Terraform + uses: hashicorp/setup-terraform@v3 + with: + terraform_version: "1.9.8" + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.11" + + - name: Install integration dependencies + run: python -m pip install -r requirements.txt -r requirements-dev.txt + + - name: Apply development integration environment + working-directory: infra/aws + run: | + terraform init + terraform apply -auto-approve + + - name: Allow event source mapping to become active + run: sleep 60 + + - name: Capture Terraform outputs + id: terraform + working-directory: infra/aws + run: | + echo "cluster_arn=$(terraform output -raw effective_msk_cluster_arn)" >> "$GITHUB_OUTPUT" + echo "probe_queue_url=$(terraform output -raw integration_probe_queue_url)" >> "$GITHUB_OUTPUT" + + - name: Measure real end-to-end path + run: | + python scripts/aws_msk_e2e.py \ + --cluster-arn '${{ steps.terraform.outputs.cluster_arn }}' \ + --probe-queue-url '${{ steps.terraform.outputs.probe_queue_url }}' \ + --samples '${{ inputs.samples }}' \ + --output evidence/aws-msk-integration-results.json + + - name: Upload measured evidence + uses: actions/upload-artifact@v4 + with: + name: aws-msk-integration-evidence-${{ github.sha }} + path: evidence/aws-msk-integration-results.json + if-no-files-found: error + + - name: Destroy temporary integration environment + if: ${{ always() && inputs.destroy_after }} + working-directory: infra/aws + run: terraform destroy -auto-approve diff --git a/.github/workflows/aws-serverless.yml b/.github/workflows/aws-serverless.yml index 6e2fc85..dd9a73f 100644 --- a/.github/workflows/aws-serverless.yml +++ b/.github/workflows/aws-serverless.yml @@ -5,22 +5,30 @@ on: paths: - "serverless/**" - "infra/aws/**" + - "scripts/aws_msk_e2e.py" - "tests/test_aws_lambda_handlers.py" + - "tests/test_aws_security_boundaries.py" + - "tests/test_release_evidence_tools.py" - ".github/workflows/aws-serverless.yml" + - ".github/workflows/aws-msk-integration.yml" push: branches: ["main"] paths: - "serverless/**" - "infra/aws/**" + - "scripts/aws_msk_e2e.py" - "tests/test_aws_lambda_handlers.py" + - "tests/test_aws_security_boundaries.py" + - "tests/test_release_evidence_tools.py" - ".github/workflows/aws-serverless.yml" + - ".github/workflows/aws-msk-integration.yml" permissions: contents: read jobs: - lambda-unit-tests: - name: Lambda unit tests + aws-unit-tests: + name: AWS credential-free unit tests runs-on: ubuntu-latest steps: @@ -41,8 +49,12 @@ jobs: python -m pip install --upgrade pip python -m pip install -r requirements.txt -r requirements-dev.txt - - name: Run Lambda handler tests - run: pytest -q tests/test_aws_lambda_handlers.py + - name: Run AWS and evidence unit tests + run: | + pytest -q \ + tests/test_aws_lambda_handlers.py \ + tests/test_aws_security_boundaries.py \ + tests/test_release_evidence_tools.py terraform-validate: name: Terraform format and validate diff --git a/evidence/aws-msk-integration-results.json b/evidence/aws-msk-integration-results.json new file mode 100644 index 0000000..f3664d4 --- /dev/null +++ b/evidence/aws-msk-integration-results.json @@ -0,0 +1,7 @@ +{ + "schema_version": 1, + "status": "not_run", + "reason": "A real AWS development MSK run requires an authorized AWS account and a runner with private VPC network access. No latency or cost result is claimed until that workflow executes.", + "latency_ms": null, + "cost": null +} diff --git a/scripts/aws_msk_e2e.py b/scripts/aws_msk_e2e.py new file mode 100644 index 0000000..716c284 --- /dev/null +++ b/scripts/aws_msk_e2e.py @@ -0,0 +1,201 @@ +"""Measure a real MSK -> Lambda -> SQS -> Lambda -> SNS path from inside the MSK VPC.""" + +import argparse +import json +import math +import os +import statistics +import time +import uuid +from pathlib import Path +from typing import Any + +import boto3 +from aws_msk_iam_sasl_signer import MSKAuthTokenProvider +from kafka import KafkaProducer + + +class MskTokenProvider: + def __init__(self, region: str): + self.region = region + + def token(self) -> str: + token, _ = MSKAuthTokenProvider.generate_auth_token(self.region) + return token + + +def percentile(values: list[float], percentile_value: float) -> float: + if not values: + raise ValueError("values must not be empty") + ordered = sorted(values) + rank = (len(ordered) - 1) * percentile_value + lower = math.floor(rank) + upper = math.ceil(rank) + if lower == upper: + return ordered[lower] + weight = rank - lower + return ordered[lower] * (1 - weight) + ordered[upper] * weight + + +def unwrap_sns_sqs_body(body: str) -> dict[str, Any]: + outer = json.loads(body) + message = outer.get("Message") + if not isinstance(message, str): + raise ValueError("SQS probe message does not contain an SNS Message string") + payload = json.loads(message) + if not isinstance(payload, dict): + raise TypeError("SNS Message must decode to a JSON object") + return payload + + +def estimate_variable_cost_usd( + *, + samples: int, + elapsed_seconds: float, + lambda_requests_per_million_usd: float, + sqs_requests_per_million_usd: float, + sns_publishes_per_million_usd: float, + msk_cluster_hourly_usd: float, +) -> float: + lambda_requests = samples * 2 + sqs_requests = samples * 4 + sns_publishes = samples + return ( + lambda_requests / 1_000_000 * lambda_requests_per_million_usd + + sqs_requests / 1_000_000 * sqs_requests_per_million_usd + + sns_publishes / 1_000_000 * sns_publishes_per_million_usd + + elapsed_seconds / 3600 * msk_cluster_hourly_usd + ) + + +def env_float(name: str) -> float: + value = os.getenv(name) + return float(value) if value else 0.0 + + +def build_producer(bootstrap_servers: list[str], region: str) -> KafkaProducer: + return KafkaProducer( + bootstrap_servers=bootstrap_servers, + security_protocol="SASL_SSL", + sasl_mechanism="OAUTHBEARER", + sasl_oauth_token_provider=MskTokenProvider(region), + value_serializer=lambda value: json.dumps(value).encode("utf-8"), + request_timeout_ms=30000, + api_version_auto_timeout_ms=30000, + ) + + +def receive_probe(sqs_client: Any, queue_url: str, probe_id: str, timeout_seconds: int) -> int: + deadline = time.time() + timeout_seconds + while time.time() < deadline: + response = sqs_client.receive_message( + QueueUrl=queue_url, + MaxNumberOfMessages=10, + WaitTimeSeconds=10, + ) + for message in response.get("Messages", []): + receipt_handle = message.get("ReceiptHandle") + payload = unwrap_sns_sqs_body(message["Body"]) + if receipt_handle: + sqs_client.delete_message(QueueUrl=queue_url, ReceiptHandle=receipt_handle) + if payload.get("probe_id") == probe_id: + return time.time_ns() + raise TimeoutError(f"Probe {probe_id} was not observed before timeout") + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--cluster-arn", required=True) + parser.add_argument("--topic", default="ride.events") + parser.add_argument("--probe-queue-url", required=True) + parser.add_argument("--samples", type=int, default=20) + parser.add_argument("--timeout-seconds", type=int, default=120) + parser.add_argument( + "--output", + type=Path, + default=Path("evidence/aws-msk-integration-results.json"), + ) + args = parser.parse_args() + + if args.samples < 1: + raise ValueError("samples must be at least 1") + + region = os.getenv("AWS_REGION", "us-east-1") + kafka_client = boto3.client("kafka", region_name=region) + sqs_client = boto3.client("sqs", region_name=region) + brokers = kafka_client.get_bootstrap_brokers(ClusterArn=args.cluster_arn) + bootstrap = brokers.get("BootstrapBrokerStringSaslIam") + if not bootstrap: + raise RuntimeError("MSK cluster did not return IAM bootstrap brokers") + + producer = build_producer(bootstrap.split(","), region) + latencies_ms: list[float] = [] + run_started = time.time() + + try: + for index in range(args.samples): + probe_id = f"probe-{uuid.uuid4()}" + sent_ns = time.time_ns() + payload = { + "event_type": "trip.completed", + "ride_id": f"synthetic-{index}", + "probe_id": probe_id, + "sent_at_ns": sent_ns, + } + producer.send(args.topic, payload).get(timeout=30) + received_ns = receive_probe( + sqs_client, + args.probe_queue_url, + probe_id, + args.timeout_seconds, + ) + latencies_ms.append((received_ns - sent_ns) / 1_000_000) + finally: + producer.close(timeout=10) + + elapsed_seconds = time.time() - run_started + pricing_inputs = { + "lambda_requests_per_million_usd": env_float("LAMBDA_REQUESTS_PER_MILLION_USD"), + "sqs_requests_per_million_usd": env_float("SQS_REQUESTS_PER_MILLION_USD"), + "sns_publishes_per_million_usd": env_float("SNS_PUBLISHES_PER_MILLION_USD"), + "msk_cluster_hourly_usd": env_float("MSK_CLUSTER_HOURLY_USD"), + } + cost = estimate_variable_cost_usd( + samples=args.samples, + elapsed_seconds=elapsed_seconds, + **pricing_inputs, + ) + + result = { + "schema_version": 1, + "status": "measured", + "measurement_scope": "amazon-msk-to-lambda-to-sqs-to-lambda-to-sns-probe-sqs", + "sample_count": len(latencies_ms), + "latency_ms": { + "min": min(latencies_ms), + "mean": statistics.mean(latencies_ms), + "p50": percentile(latencies_ms, 0.50), + "p95": percentile(latencies_ms, 0.95), + "p99": percentile(latencies_ms, 0.99), + "max": max(latencies_ms), + }, + "cost": { + "measurement_type": "request-count-and-runtime estimate", + "estimated_variable_cost_usd": cost, + "pricing_inputs": pricing_inputs, + "pricing_inputs_complete": all(value > 0 for value in pricing_inputs.values()), + "includes_data_transfer_or_cloudwatch": False, + "note": "This is not AWS billing data. Use Cost Explorer after the run for billed-cost evidence.", + }, + "cluster_arn": args.cluster_arn, + "topic": args.topic, + "aws_region": region, + "elapsed_seconds": elapsed_seconds, + } + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n", encoding="utf-8") + print(json.dumps(result, indent=2, sort_keys=True)) + + +if __name__ == "__main__": + main() diff --git a/scripts/k8s_release_evidence.py b/scripts/k8s_release_evidence.py new file mode 100644 index 0000000..b66a8de --- /dev/null +++ b/scripts/k8s_release_evidence.py @@ -0,0 +1,76 @@ +"""Create deterministic Kubernetes promotion or rollback evidence JSON.""" + +import argparse +import hashlib +import json +from datetime import datetime, timezone +from pathlib import Path + + +def manifest_sha256(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def build_evidence( + *, + action: str, + environment: str, + image: str, + git_sha: str, + manifest: Path, + deployment_status: str, + from_revision: str | None = None, + to_revision: str | None = None, +) -> dict[str, object]: + if action not in {"promote", "rollback"}: + raise ValueError("action must be promote or rollback") + if environment not in {"dev", "staging", "prod"}: + raise ValueError("environment must be dev, staging, or prod") + if deployment_status not in {"rendered", "applied"}: + raise ValueError("deployment_status must be rendered or applied") + if environment == "prod" and "@sha256:" not in image: + raise ValueError("production promotion requires an immutable image digest") + + return { + "schema_version": 1, + "recorded_at": datetime.now(timezone.utc).isoformat(), + "action": action, + "environment": environment, + "deployment_status": deployment_status, + "image": image, + "git_sha": git_sha, + "manifest_sha256": manifest_sha256(manifest), + "from_revision": from_revision, + "to_revision": to_revision, + } + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--action", choices=["promote", "rollback"], required=True) + parser.add_argument("--environment", choices=["dev", "staging", "prod"], required=True) + parser.add_argument("--image", required=True) + parser.add_argument("--git-sha", required=True) + parser.add_argument("--manifest", type=Path, required=True) + parser.add_argument("--deployment-status", choices=["rendered", "applied"], required=True) + parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--from-revision") + parser.add_argument("--to-revision") + args = parser.parse_args() + + evidence = build_evidence( + action=args.action, + environment=args.environment, + image=args.image, + git_sha=args.git_sha, + manifest=args.manifest, + deployment_status=args.deployment_status, + from_revision=args.from_revision, + to_revision=args.to_revision, + ) + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(json.dumps(evidence, indent=2, sort_keys=True) + "\n", encoding="utf-8") + + +if __name__ == "__main__": + main() diff --git a/tests/test_release_evidence_tools.py b/tests/test_release_evidence_tools.py new file mode 100644 index 0000000..3cf50f2 --- /dev/null +++ b/tests/test_release_evidence_tools.py @@ -0,0 +1,59 @@ +import json +from pathlib import Path + +import pytest + +from scripts.aws_msk_e2e import estimate_variable_cost_usd, percentile, unwrap_sns_sqs_body +from scripts.k8s_release_evidence import build_evidence, manifest_sha256 + + +def test_percentile_interpolates_deterministically(): + values = [10.0, 20.0, 30.0, 40.0] + assert percentile(values, 0.50) == 25.0 + assert percentile(values, 0.95) == pytest.approx(38.5) + + +def test_cost_estimate_uses_explicit_pricing_inputs(): + cost = estimate_variable_cost_usd( + samples=100, + elapsed_seconds=3600, + lambda_requests_per_million_usd=1.0, + sqs_requests_per_million_usd=1.0, + sns_publishes_per_million_usd=1.0, + msk_cluster_hourly_usd=2.0, + ) + assert cost == pytest.approx(2.0007) + + +def test_unwrap_sns_sqs_body_returns_original_notification_payload(): + body = json.dumps({"Message": json.dumps({"event_type": "trip.completed", "probe_id": "p1"})}) + assert unwrap_sns_sqs_body(body)["probe_id"] == "p1" + + +def test_release_evidence_records_manifest_hash(tmp_path: Path): + manifest = tmp_path / "rendered.yaml" + manifest.write_text("kind: Deployment\n", encoding="utf-8") + evidence = build_evidence( + action="promote", + environment="staging", + image="ghcr.io/example/app:v1.2.3", + git_sha="abc123", + manifest=manifest, + deployment_status="rendered", + ) + assert evidence["manifest_sha256"] == manifest_sha256(manifest) + assert evidence["deployment_status"] == "rendered" + + +def test_production_evidence_requires_immutable_digest(tmp_path: Path): + manifest = tmp_path / "rendered.yaml" + manifest.write_text("kind: Deployment\n", encoding="utf-8") + with pytest.raises(ValueError, match="immutable image digest"): + build_evidence( + action="promote", + environment="prod", + image="ghcr.io/example/app:latest", + git_sha="abc123", + manifest=manifest, + deployment_status="rendered", + ) From 34ac1bb18ed5dcc9c719a7b1a0de5ea3627d41fa Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Mon, 31 Aug 2026 22:23:53 -0400 Subject: [PATCH 05/21] feat: add Kubernetes promotion and rollback evidence --- .github/workflows/kubernetes-promotion.yml | 113 ++++++++++++++++++ .github/workflows/kubernetes-rollback.yml | 75 ++++++++++++ .github/workflows/kubernetes-validation.yml | 50 ++++++++ evidence/kubernetes-deployment-results.json | 5 + infra/kubernetes/api-gateway-deployment.yaml | 44 ++++++- infra/kubernetes/kustomization.yaml | 19 +++ .../overlays/dev/kustomization.yaml | 22 ++++ .../overlays/prod/kustomization.yaml | 24 ++++ .../overlays/staging/kustomization.yaml | 24 ++++ tests/test_kubernetes_manifests.py | 32 +++++ 10 files changed, 404 insertions(+), 4 deletions(-) create mode 100644 .github/workflows/kubernetes-promotion.yml create mode 100644 .github/workflows/kubernetes-rollback.yml create mode 100644 .github/workflows/kubernetes-validation.yml create mode 100644 evidence/kubernetes-deployment-results.json create mode 100644 infra/kubernetes/kustomization.yaml create mode 100644 infra/kubernetes/overlays/dev/kustomization.yaml create mode 100644 infra/kubernetes/overlays/prod/kustomization.yaml create mode 100644 infra/kubernetes/overlays/staging/kustomization.yaml create mode 100644 tests/test_kubernetes_manifests.py diff --git a/.github/workflows/kubernetes-promotion.yml b/.github/workflows/kubernetes-promotion.yml new file mode 100644 index 0000000..8f5ea94 --- /dev/null +++ b/.github/workflows/kubernetes-promotion.yml @@ -0,0 +1,113 @@ +name: Kubernetes Promotion Evidence + +on: + workflow_dispatch: + inputs: + environment: + description: Target environment + required: true + type: choice + options: [dev, staging, prod] + image: + description: Full image reference; production requires @sha256 digest + required: true + type: string + deploy: + description: Apply to the configured cluster instead of render-only evidence + required: true + default: false + type: boolean + +permissions: + contents: read + +jobs: + promote: + runs-on: ubuntu-latest + environment: ${{ inputs.environment }} + steps: + - uses: actions/checkout@v4 + + - name: Set up Kustomize + uses: imranismail/setup-kustomize@v2 + + - name: Set up kubectl + uses: azure/setup-kubectl@v4 + + - name: Render immutable candidate manifest + env: + TARGET_ENV: ${{ inputs.environment }} + IMAGE: ${{ inputs.image }} + run: | + set -euo pipefail + cd "infra/kubernetes/overlays/${TARGET_ENV}" + kustomize edit set image "api-gateway=${IMAGE}" + kustomize build . > "$RUNNER_TEMP/rendered.yaml" + grep -n "image:" "$RUNNER_TEMP/rendered.yaml" + + - name: Client-side manifest validation + run: kubectl apply --dry-run=client --validate=false -f "$RUNNER_TEMP/rendered.yaml" + + - name: Configure cluster credentials + if: ${{ inputs.deploy }} + env: + KUBE_CONFIG_DATA: ${{ secrets.KUBE_CONFIG_DATA }} + run: | + test -n "$KUBE_CONFIG_DATA" + printf '%s' "$KUBE_CONFIG_DATA" | base64 --decode > "$RUNNER_TEMP/kubeconfig" + echo "KUBECONFIG=$RUNNER_TEMP/kubeconfig" >> "$GITHUB_ENV" + + - name: Capture current revision + if: ${{ inputs.deploy }} + id: before + run: | + revision=$(kubectl rollout history deployment/api-gateway | awk 'NR>2 && $1 ~ /^[0-9]+$/ {print $1}' | tail -n 1) + echo "revision=${revision:-unknown}" >> "$GITHUB_OUTPUT" + + - name: Apply and verify rollout + if: ${{ inputs.deploy }} + run: | + kubectl apply -f "$RUNNER_TEMP/rendered.yaml" + kubectl rollout status deployment/api-gateway --timeout=180s + + - name: Capture promoted revision + if: ${{ inputs.deploy }} + id: after + run: | + revision=$(kubectl rollout history deployment/api-gateway | awk 'NR>2 && $1 ~ /^[0-9]+$/ {print $1}' | tail -n 1) + echo "revision=${revision:-unknown}" >> "$GITHUB_OUTPUT" + + - name: Record render-only evidence + if: ${{ !inputs.deploy }} + run: | + python scripts/k8s_release_evidence.py \ + --action promote \ + --environment '${{ inputs.environment }}' \ + --image '${{ inputs.image }}' \ + --git-sha '${{ github.sha }}' \ + --manifest "$RUNNER_TEMP/rendered.yaml" \ + --deployment-status rendered \ + --output evidence/kubernetes-deployment-results.json + + - name: Record applied promotion evidence + if: ${{ inputs.deploy }} + run: | + python scripts/k8s_release_evidence.py \ + --action promote \ + --environment '${{ inputs.environment }}' \ + --image '${{ inputs.image }}' \ + --git-sha '${{ github.sha }}' \ + --manifest "$RUNNER_TEMP/rendered.yaml" \ + --deployment-status applied \ + --from-revision '${{ steps.before.outputs.revision }}' \ + --to-revision '${{ steps.after.outputs.revision }}' \ + --output evidence/kubernetes-deployment-results.json + + - name: Upload promotion evidence + uses: actions/upload-artifact@v4 + with: + name: kubernetes-promotion-${{ inputs.environment }}-${{ github.sha }} + path: | + ${{ runner.temp }}/rendered.yaml + evidence/kubernetes-deployment-results.json + if-no-files-found: error diff --git a/.github/workflows/kubernetes-rollback.yml b/.github/workflows/kubernetes-rollback.yml new file mode 100644 index 0000000..2665cdb --- /dev/null +++ b/.github/workflows/kubernetes-rollback.yml @@ -0,0 +1,75 @@ +name: Kubernetes Rollback Evidence + +on: + workflow_dispatch: + inputs: + environment: + description: Environment to roll back + required: true + type: choice + options: [dev, staging, prod] + revision: + description: Deployment revision to restore + required: true + type: string + +permissions: + contents: read + +jobs: + rollback: + runs-on: ubuntu-latest + environment: ${{ inputs.environment }} + steps: + - uses: actions/checkout@v4 + + - name: Set up kubectl + uses: azure/setup-kubectl@v4 + + - name: Configure cluster credentials + env: + KUBE_CONFIG_DATA: ${{ secrets.KUBE_CONFIG_DATA }} + run: | + test -n "$KUBE_CONFIG_DATA" + printf '%s' "$KUBE_CONFIG_DATA" | base64 --decode > "$RUNNER_TEMP/kubeconfig" + echo "KUBECONFIG=$RUNNER_TEMP/kubeconfig" >> "$GITHUB_ENV" + + - name: Capture current revision + id: before + run: | + revision=$(kubectl rollout history deployment/api-gateway | awk 'NR>2 && $1 ~ /^[0-9]+$/ {print $1}' | tail -n 1) + echo "revision=${revision:-unknown}" >> "$GITHUB_OUTPUT" + + - name: Roll back and verify + run: | + kubectl rollout undo deployment/api-gateway --to-revision='${{ inputs.revision }}' + kubectl rollout status deployment/api-gateway --timeout=180s + kubectl get deployment api-gateway -o yaml > "$RUNNER_TEMP/rollback-deployment.yaml" + + - name: Capture rollback image + id: image + run: | + image=$(kubectl get deployment api-gateway -o jsonpath='{.spec.template.spec.containers[?(@.name=="api-gateway")].image}') + echo "image=$image" >> "$GITHUB_OUTPUT" + + - name: Record rollback evidence + run: | + python scripts/k8s_release_evidence.py \ + --action rollback \ + --environment '${{ inputs.environment }}' \ + --image '${{ steps.image.outputs.image }}' \ + --git-sha '${{ github.sha }}' \ + --manifest "$RUNNER_TEMP/rollback-deployment.yaml" \ + --deployment-status applied \ + --from-revision '${{ steps.before.outputs.revision }}' \ + --to-revision '${{ inputs.revision }}' \ + --output evidence/kubernetes-deployment-results.json + + - name: Upload rollback evidence + uses: actions/upload-artifact@v4 + with: + name: kubernetes-rollback-${{ inputs.environment }}-${{ github.sha }} + path: | + ${{ runner.temp }}/rollback-deployment.yaml + evidence/kubernetes-deployment-results.json + if-no-files-found: error diff --git a/.github/workflows/kubernetes-validation.yml b/.github/workflows/kubernetes-validation.yml new file mode 100644 index 0000000..573a1e8 --- /dev/null +++ b/.github/workflows/kubernetes-validation.yml @@ -0,0 +1,50 @@ +name: Kubernetes Manifest Validation + +on: + pull_request: + paths: + - "infra/kubernetes/**" + - "scripts/k8s_release_evidence.py" + - "tests/test_kubernetes_manifests.py" + - "tests/test_release_evidence_tools.py" + - ".github/workflows/kubernetes-*.yml" + push: + branches: ["main"] + paths: + - "infra/kubernetes/**" + - "scripts/k8s_release_evidence.py" + - "tests/test_kubernetes_manifests.py" + - "tests/test_release_evidence_tools.py" + - ".github/workflows/kubernetes-*.yml" + +permissions: + contents: read + +jobs: + validate: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - name: Set up Kustomize + uses: imranismail/setup-kustomize@v2 + + - name: Render all environment overlays + run: | + set -euo pipefail + for environment in dev staging prod; do + kustomize build "infra/kubernetes/overlays/${environment}" > "/tmp/${environment}.yaml" + test -s "/tmp/${environment}.yaml" + done + + - name: Reject mutable latest images and legacy placeholders + run: | + ! grep -R --line-number -E 'yourdockerhub|image: .*:latest' infra/kubernetes + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.11" + + - name: Run manifest/evidence tests + run: pytest -q tests/test_kubernetes_manifests.py tests/test_release_evidence_tools.py diff --git a/evidence/kubernetes-deployment-results.json b/evidence/kubernetes-deployment-results.json new file mode 100644 index 0000000..bf8012c --- /dev/null +++ b/evidence/kubernetes-deployment-results.json @@ -0,0 +1,5 @@ +{ + "schema_version": 1, + "status": "not_run", + "reason": "Promotion and rollback workflows generate evidence artifacts when explicitly run. This placeholder does not claim a Kubernetes deployment occurred." +} diff --git a/infra/kubernetes/api-gateway-deployment.yaml b/infra/kubernetes/api-gateway-deployment.yaml index c0d0a5a..12c1f67 100644 --- a/infra/kubernetes/api-gateway-deployment.yaml +++ b/infra/kubernetes/api-gateway-deployment.yaml @@ -2,18 +2,54 @@ apiVersion: apps/v1 kind: Deployment metadata: name: api-gateway + labels: + app.kubernetes.io/name: api-gateway + app.kubernetes.io/part-of: ride-sharing-platform spec: replicas: 2 + revisionHistoryLimit: 10 + strategy: + type: RollingUpdate + rollingUpdate: + maxUnavailable: 0 + maxSurge: 1 selector: matchLabels: - app: api-gateway + app.kubernetes.io/name: api-gateway template: metadata: labels: - app: api-gateway + app.kubernetes.io/name: api-gateway + app.kubernetes.io/part-of: ride-sharing-platform spec: containers: - name: api-gateway - image: yourdockerhub/api-gateway:latest + image: api-gateway:local + imagePullPolicy: IfNotPresent + envFrom: + - configMapRef: + name: ride-sharing-runtime ports: - - containerPort: 8000 \ No newline at end of file + - name: http + containerPort: 8000 + readinessProbe: + httpGet: + path: /driver-location/ready + port: http + initialDelaySeconds: 5 + periodSeconds: 10 + timeoutSeconds: 3 + livenessProbe: + httpGet: + path: /driver-location/health + port: http + initialDelaySeconds: 10 + periodSeconds: 20 + timeoutSeconds: 3 + resources: + requests: + cpu: 100m + memory: 128Mi + limits: + cpu: 500m + memory: 512Mi diff --git a/infra/kubernetes/kustomization.yaml b/infra/kubernetes/kustomization.yaml new file mode 100644 index 0000000..eb6297e --- /dev/null +++ b/infra/kubernetes/kustomization.yaml @@ -0,0 +1,19 @@ +apiVersion: kustomize.config.k8s.io/v1beta1 +kind: Kustomization + +resources: + - api-gateway-deployment.yaml + - api-gateway-hpa.yaml + +configMapGenerator: + - name: ride-sharing-runtime + literals: + - APP_ENV=local + - AUTH_REQUIRED=false + - AUTH_ISSUER= + - AUTH_AUDIENCE= + - AUTH_TOKEN_USE=access + - DRIVER_LOCATION_REDIS_SECRET_ID= + +generatorOptions: + disableNameSuffixHash: true diff --git a/infra/kubernetes/overlays/dev/kustomization.yaml b/infra/kubernetes/overlays/dev/kustomization.yaml new file mode 100644 index 0000000..d372942 --- /dev/null +++ b/infra/kubernetes/overlays/dev/kustomization.yaml @@ -0,0 +1,22 @@ +apiVersion: kustomize.config.k8s.io/v1beta1 +kind: Kustomization + +resources: + - ../.. + +labels: + - pairs: + app.kubernetes.io/environment: dev + +images: + - name: api-gateway + newName: ghcr.io/coreyleath-code/scalable-event-driven-ride-sharing-platform + newTag: dev-unset + +configMapGenerator: + - name: ride-sharing-runtime + behavior: merge + literals: + - APP_ENV=dev + - AUTH_REQUIRED=false + - DRIVER_LOCATION_REDIS_SECRET_ID=ride-sharing/dev/runtime diff --git a/infra/kubernetes/overlays/prod/kustomization.yaml b/infra/kubernetes/overlays/prod/kustomization.yaml new file mode 100644 index 0000000..ec43754 --- /dev/null +++ b/infra/kubernetes/overlays/prod/kustomization.yaml @@ -0,0 +1,24 @@ +apiVersion: kustomize.config.k8s.io/v1beta1 +kind: Kustomization + +resources: + - ../.. + +labels: + - pairs: + app.kubernetes.io/environment: prod + +images: + - name: api-gateway + newName: ghcr.io/coreyleath-code/scalable-event-driven-ride-sharing-platform + newTag: prod-unset + +configMapGenerator: + - name: ride-sharing-runtime + behavior: merge + literals: + - APP_ENV=prod + - AUTH_REQUIRED=true + - AUTH_ISSUER=https://replace-me.example + - AUTH_AUDIENCE=replace-me + - DRIVER_LOCATION_REDIS_SECRET_ID=ride-sharing/prod/runtime diff --git a/infra/kubernetes/overlays/staging/kustomization.yaml b/infra/kubernetes/overlays/staging/kustomization.yaml new file mode 100644 index 0000000..7d0c9c7 --- /dev/null +++ b/infra/kubernetes/overlays/staging/kustomization.yaml @@ -0,0 +1,24 @@ +apiVersion: kustomize.config.k8s.io/v1beta1 +kind: Kustomization + +resources: + - ../.. + +labels: + - pairs: + app.kubernetes.io/environment: staging + +images: + - name: api-gateway + newName: ghcr.io/coreyleath-code/scalable-event-driven-ride-sharing-platform + newTag: staging-unset + +configMapGenerator: + - name: ride-sharing-runtime + behavior: merge + literals: + - APP_ENV=staging + - AUTH_REQUIRED=true + - AUTH_ISSUER=https://replace-me.example + - AUTH_AUDIENCE=replace-me + - DRIVER_LOCATION_REDIS_SECRET_ID=ride-sharing/staging/runtime diff --git a/tests/test_kubernetes_manifests.py b/tests/test_kubernetes_manifests.py new file mode 100644 index 0000000..38fb10a --- /dev/null +++ b/tests/test_kubernetes_manifests.py @@ -0,0 +1,32 @@ +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +K8S = ROOT / "infra" / "kubernetes" + + +def test_base_deployment_has_parameterized_image_and_rollout_safety(): + deployment = (K8S / "api-gateway-deployment.yaml").read_text(encoding="utf-8") + assert "image: api-gateway:local" in deployment + assert "revisionHistoryLimit: 10" in deployment + assert "maxUnavailable: 0" in deployment + assert "yourdockerhub" not in deployment + assert ":latest" not in deployment + + +def test_environment_overlays_are_explicit_and_do_not_use_latest(): + for environment in ("dev", "staging", "prod"): + overlay = (K8S / "overlays" / environment / "kustomization.yaml").read_text( + encoding="utf-8" + ) + assert f"APP_ENV={environment}" in overlay + assert "newName: ghcr.io/coreyleath-code/scalable-event-driven-ride-sharing-platform" in overlay + assert ":latest" not in overlay + + +def test_staging_and_prod_require_authentication(): + for environment in ("staging", "prod"): + overlay = (K8S / "overlays" / environment / "kustomization.yaml").read_text( + encoding="utf-8" + ) + assert "AUTH_REQUIRED=true" in overlay From c6567b61a8e850b682a82e884c7da3b0df9591ac Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Mon, 31 Aug 2026 22:26:13 -0400 Subject: [PATCH 06/21] docs: document production hardening and evidence boundaries --- .env.example | 9 +- CHANGELOG.md | 28 +- README.md | 384 ++++++++++++---------------- docs/aws-msk-integration.md | 68 +++++ docs/aws-serverless-architecture.md | 130 +++------- docs/kubernetes-promotion.md | 54 ++++ docs/security-and-pii.md | 81 ++++++ 7 files changed, 417 insertions(+), 337 deletions(-) create mode 100644 docs/aws-msk-integration.md create mode 100644 docs/kubernetes-promotion.md create mode 100644 docs/security-and-pii.md diff --git a/.env.example b/.env.example index 4a9182e..6bce32a 100644 --- a/.env.example +++ b/.env.example @@ -1,4 +1,9 @@ API_PORT=8000 KAFKA_BROKER=localhost:9092 -DB_URL=postgres://user:password@localhost:5432/rides -REDIS_URL=redis://localhost:6379 +DRIVER_LOCATION_REDIS_URL=redis://localhost:6379/0 +DRIVER_LOCATION_REDIS_SECRET_ID= +AUTH_REQUIRED=false +AUTH_ISSUER= +AUTH_AUDIENCE= +AUTH_JWKS_URL= +AUTH_TOKEN_USE=access diff --git a/CHANGELOG.md b/CHANGELOG.md index 9fd45b8..de231cc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,19 +4,25 @@ ### Added - AWS Lambda ride-event processor for Amazon MSK batches with Kafka-derived idempotency keys. -- SQS buffering and dead-letter queue handling between asynchronous AWS workers. -- AWS Lambda notification worker with SQS partial-batch failure responses and SNS publication. -- Terraform reference infrastructure for Lambda, SQS, SNS, CloudWatch Logs, IAM, and optional Amazon MSK event-source mapping. -- CloudWatch serverless dashboard for Lambda health, p95 duration, throttles, concurrency, SQS backlog, oldest-message age, DLQ depth, and recent failure logs. -- CloudWatch operational alarms for Lambda errors/throttles/duration, queue lag, DLQ activity, and record-level notification failures. -- Dedicated operational-alerts SNS topic with optional email subscription. -- CloudWatch Logs metric filter for record-level notification processing failures. -- Credential-free Lambda unit tests and Terraform validation workflow. -- AWS serverless architecture, CloudWatch operations, and deployment documentation. +- SQS buffering/DLQ handling and an SNS notification worker with partial-batch failure responses. +- Terraform reference infrastructure for Lambda, SQS, SNS, IAM, optional Amazon MSK, and CloudWatch observability. +- CloudWatch dashboard, alarms, operational SNS notifications, and a log-derived record-failure metric. +- Optional private IAM-authenticated MSK Serverless development environment and SNS-to-SQS integration probe. +- Real-MSK end-to-end latency evidence harness and manual VPC-runner workflow with explicit cost-estimate metadata. +- OIDC/Cognito-style JWT authentication boundary for sensitive driver-location endpoints. +- AWS Secrets Manager runtime resolver plus optional KMS/secret metadata/least-privilege reader-policy Terraform resources. +- Direct-PII guards in both Lambda stages and PII-safe application/event logging behavior. +- Kustomize dev/staging/prod overlays with parameterized images and environment configuration. +- Kubernetes promotion and rollback workflows with rendered/applied evidence artifacts and revision tracking. +- Credential-free unit tests for authentication, secrets, PII boundaries, broker connection contracts, AWS evidence helpers, and Kubernetes manifest policy. +- AWS, security/PII, real-MSK evidence, and Kubernetes promotion documentation. ### Changed -- README now documents the hybrid container/service + serverless architecture, CloudWatch observability path, and separates measured evidence from architecture targets. -- Python packaging now includes the `serverless` package tree. +- README now distinguishes implemented controls, real measurements, unexecuted integration evidence, architecture targets, and remaining production gaps. +- Root Kafka/Redis/RabbitMQ adapters are directly importable and expose explicit pre-connect failure behavior. +- Event bus and driver telemetry consumer no longer log raw event bodies, coordinates, or validation payload details. +- Kubernetes deployment no longer uses `yourdockerhub/api-gateway:latest`; it includes rolling-update safety, probes, resource requests/limits, and revision history. +- Python packaging includes authentication, secret-resolution, and PII-policy modules. ## [1.1.0] - 2026-08-21 ### Added diff --git a/README.md b/README.md index 3f797ee..00a8c52 100644 --- a/README.md +++ b/README.md @@ -2,157 +2,120 @@ [![CI](https://github.com/CoreyLeath-code/Scalable-Event-Driven-Ride-Sharing-Platform/actions/workflows/ci.yml/badge.svg)](https://github.com/CoreyLeath-code/Scalable-Event-Driven-Ride-Sharing-Platform/actions/workflows/ci.yml) [![AWS Serverless Validation](https://github.com/CoreyLeath-code/Scalable-Event-Driven-Ride-Sharing-Platform/actions/workflows/aws-serverless.yml/badge.svg)](https://github.com/CoreyLeath-code/Scalable-Event-Driven-Ride-Sharing-Platform/actions/workflows/aws-serverless.yml) +[![Kubernetes Validation](https://github.com/CoreyLeath-code/Scalable-Event-Driven-Ride-Sharing-Platform/actions/workflows/kubernetes-validation.yml/badge.svg)](https://github.com/CoreyLeath-code/Scalable-Event-Driven-Ride-Sharing-Platform/actions/workflows/kubernetes-validation.yml) [![System Hygiene Matrix](https://github.com/CoreyLeath-code/Scalable-Event-Driven-Ride-Sharing-Platform/actions/workflows/hygiene-matrix.yml/badge.svg)](https://github.com/CoreyLeath-code/Scalable-Event-Driven-Ride-Sharing-Platform/actions/workflows/hygiene-matrix.yml) ![Python](https://img.shields.io/badge/Python-3.11-3776AB?logo=python&logoColor=white) ![FastAPI](https://img.shields.io/badge/FastAPI-Framework-009688?logo=fastapi&logoColor=white) ![Kafka](https://img.shields.io/badge/Kafka-Event%20Streaming-231F20?logo=apachekafka&logoColor=white) ![AWS Lambda](https://img.shields.io/badge/AWS-Lambda-FF9900?logo=awslambda&logoColor=white) +![Amazon MSK](https://img.shields.io/badge/AWS-Amazon%20MSK-FF9900?logo=amazonaws&logoColor=white) ![CloudWatch](https://img.shields.io/badge/AWS-CloudWatch-FF4F8B?logo=amazoncloudwatch&logoColor=white) ![Terraform](https://img.shields.io/badge/Terraform-IaC-844FBA?logo=terraform&logoColor=white) -![Redis](https://img.shields.io/badge/Redis-State-DC382D?logo=redis&logoColor=white) -![Docker](https://img.shields.io/badge/Docker-Containerized-2496ED?logo=docker&logoColor=white) -![Kubernetes](https://img.shields.io/badge/Kubernetes-Orchestrated-326CE5?logo=kubernetes&logoColor=white) +![Kubernetes](https://img.shields.io/badge/Kubernetes-Kustomize-326CE5?logo=kubernetes&logoColor=white) -A production-style reference implementation for event-driven ride-sharing backend concepts: asynchronous ride events, driver-location state, matching and pricing primitives, broker adapters, containerized services, load-balancing, reproducible benchmarks, an optional AWS serverless event-processing path, and Terraform-managed CloudWatch observability. +A production-shaped reference implementation for event-driven ride-sharing backend concepts: asynchronous ride events, driver-location state, matching/pricing primitives, Kafka/Redis/RabbitMQ broker adapters, load-balanced APIs, an optional AWS Lambda/MSK path, CloudWatch operations, authentication/secrets/PII boundaries, and evidence-oriented Kubernetes promotion workflows. -The repository intentionally separates **implemented and reproducible behavior** from **production architecture targets**. It does not claim to be a complete live ride-sharing product and does not process real payments, identity data, or production rider/driver PII. +**Evidence rule:** implemented behavior, measured evidence, architecture targets, and unexecuted deployment references are kept separate. This repository does not claim to be a live production ride-sharing system and does not claim cloud latency, cost, SLO, security, or deployment results that have not been reproduced. ## Architecture -### Hybrid service + serverless design - ```mermaid flowchart LR - Client[Rider / Driver Client] --> Gateway[API Gateway / FastAPI] - Gateway --> Services[API + Long-running Services] - Services --> Kafka[(Kafka / Amazon MSK)] - Services --> Redis[(Redis Driver State)] + Client[Rider / Driver Client] --> Auth[OIDC / Cognito-style JWT Boundary] + Auth --> API[FastAPI Driver/API Services] + Secrets[AWS Secrets Manager] --> API + API --> Redis[(Redis Driver State)] + API --> Kafka[(Kafka / Amazon MSK)] - Kafka --> Matching[Matching / Trip / Pricing Workers] + Kafka --> Matching[Matching / Trip / Pricing] Kafka --> Lambda1[AWS Lambda\nRide Event Processor] Lambda1 --> SQS[(Amazon SQS)] SQS --> Lambda2[AWS Lambda\nNotification Worker] - SQS -. repeated failures .-> DLQ[(SQS DLQ)] + SQS -. retry exhaustion .-> DLQ[(SQS DLQ)] Lambda2 --> SNS[(Amazon SNS)] Lambda1 --> CW[CloudWatch Logs + Metrics] Lambda2 --> CW SQS --> CW DLQ --> CW - CW --> Dashboard[CloudWatch Dashboard] - CW --> Alarms[CloudWatch Alarms] - Alarms --> OpsSNS[(SNS Operational Alerts)] -``` - -The design deliberately keeps latency-sensitive driver-location and matching responsibilities in long-running services while using Lambda for independently scalable asynchronous work. This avoids treating serverless as a universal replacement for services that may require stable low-latency execution. + CW --> Dash[Dashboard + Alarms] + Dash --> Ops[(SNS Operational Alerts)] -### Core event flow - -```text -ride.requested -> matching-service -driver.matched -> trip-service -trip.started -> pricing-service -trip.completed -> payment-service -payment.processed -> notification-service + K8s[Kustomize dev / staging / prod] --> API ``` -The local event-bus abstraction supports asynchronous publish/subscribe behavior and can be backed by Kafka, Redis Streams, or RabbitMQ. The AWS extension adds an Amazon MSK -> Lambda -> SQS -> Lambda -> SNS reference path without removing the local development workflow. +Latency-sensitive driver-location and matching responsibilities remain long-running services. Lambda is used for independently scalable asynchronous work rather than treated as a universal replacement for services that may need stable low-latency execution. -## AWS Lambda serverless extension +## Implemented engineering boundaries -The AWS implementation lives under `serverless/` and `infra/aws/`. +### Event processing and reliability -### Ride-event processor Lambda +- in-memory async event bus plus Kafka, Redis Streams, and RabbitMQ adapters; +- Docker-backed broker round-trip integration tests; +- Amazon MSK -> Lambda -> SQS -> Lambda -> SNS reference path; +- Kafka `topic:partition:offset` idempotency key propagation; +- SQS buffering, bounded retries, DLQ, and partial-batch failure handling; +- NGINX/load-balanced driver-location API with Redis shared state; +- liveness/readiness separation and loss-of-one-replica integration coverage. -`serverless/ride_event_processor/handler.py` consumes the Amazon MSK event shape, base64-decodes JSON event values, validates that each event contains `event_type` or `type`, and forwards a normalized envelope to SQS. +### CloudWatch observability -Each forwarded message includes: +Terraform defines: -```json -{ - "idempotency_key": "ride.events:0:42", - "source": "amazon-msk", - "topic": "ride.events", - "partition": 0, - "offset": 42, - "event": { - "event_type": "ride.requested", - "ride_id": "ride-123" - } -} -``` +- Lambda log groups with retention; +- Lambda invocations, errors, p95 duration, throttles, and concurrency dashboard panels; +- SQS backlog, oldest-message age, and DLQ depth panels; +- log-derived record-level notification failure metric; +- alarms for errors, throttles, high p95 duration, queue lag, DLQ activity, and record failures; +- a dedicated operational SNS topic separate from product notifications. -The `topic:partition:offset` key gives downstream consumers a stable identifier for idempotency in an at-least-once delivery model. The reference handler fails on malformed records or failed SQS sends so the source batch can be retried rather than silently dropped. +See [`docs/cloudwatch-observability.md`](docs/cloudwatch-observability.md). -### Notification worker Lambda +### Authentication, secrets, and PII -`serverless/notification_worker/handler.py` consumes the processed-events SQS queue and publishes selected lifecycle events to SNS: +Sensitive driver-location collection endpoints support enforced RS256 OIDC JWT validation through `AUTH_REQUIRED=true`. Health/readiness endpoints remain public for infrastructure probes. -- `driver.matched` -- `trip.started` -- `trip.completed` -- `payment.processed` +Runtime Redis configuration can be resolved from AWS Secrets Manager via `DRIVER_LOCATION_REDIS_SECRET_ID`; local development can use `DRIVER_LOCATION_REDIS_URL`. Terraform can create KMS-protected secret metadata and a least-privilege reader policy, but deliberately does not place secret values in Terraform state. -It returns SQS partial-batch failure responses so one bad record does not force successful records in the same batch to be retried. Repeatedly failing messages are routed to the configured dead-letter queue. +The asynchronous Lambda path rejects direct identifier fields such as email, phone, names, home addresses, SSNs, and payment/bank identifiers before SQS/SNS fan-out. Event-bus and consumer logs avoid raw event bodies and coordinates. -### Terraform-provisioned AWS resources +See [`docs/security-and-pii.md`](docs/security-and-pii.md). -`infra/aws/` defines a deployable reference stack for: +### Kubernetes environments and release evidence -- two Python 3.11 Lambda functions; -- processed-events SQS queue; -- SQS dead-letter queue; -- ride-notifications SNS topic; -- CloudWatch log groups with configurable retention; -- CloudWatch dashboard, alarms, and log-derived failure metric; -- dedicated operational-alerts SNS topic with optional email subscription; -- separate Lambda execution roles and scoped runtime policies; -- optional Amazon MSK event-source mapping; -- deterministic Lambda ZIP packaging through the Terraform archive provider. +Kubernetes uses one Kustomize base plus explicit `dev`, `staging`, and `prod` overlays. The old hard-coded `yourdockerhub/api-gateway:latest` image is removed. -The MSK event-source mapping is disabled when `msk_cluster_arn` is `null`, allowing the Terraform configuration to be statically validated without requiring an existing Kafka cluster. +The deployment now includes: -See [`docs/aws-serverless-architecture.md`](docs/aws-serverless-architecture.md) for event contracts and deployment notes, and [`docs/cloudwatch-observability.md`](docs/cloudwatch-observability.md) for dashboard coverage, alarms, thresholds, triage flow, and the production observability boundary. +- parameterized image replacement; +- rolling-update policy with `maxUnavailable: 0`; +- revision history for rollback; +- liveness/readiness probes; +- CPU/memory requests and limits required for meaningful HPA behavior; +- environment-specific auth and Secrets Manager identifiers. -## CloudWatch observability +Manual promotion and rollback workflows upload JSON evidence containing environment, source commit, image, rendered-manifest SHA-256, applied/rendered status, and deployment revisions when a real cluster change occurs. Production evidence requires an immutable image digest. -The AWS reference stack now includes an operator-focused observability layer instead of stopping at raw Lambda logs. +See [`docs/kubernetes-promotion.md`](docs/kubernetes-promotion.md). -### Dashboard coverage +## Real Amazon MSK development integration -The `${name_prefix}-serverless-observability` dashboard tracks: +Terraform can optionally create a **private IAM-authenticated MSK Serverless development cluster** plus an SNS-subscribed SQS probe queue. Both are disabled by default because they create billable resources. -- Lambda invocations and invocation-level errors; -- p95 Lambda duration for the ride-event processor and notification worker; -- Lambda throttles and concurrent executions; -- processed-events SQS visible backlog; -- age of the oldest SQS message; -- dead-letter queue depth; -- record-level notification processing failures; -- a CloudWatch Logs Insights table for recent notification worker failures. +The manual `Real AWS MSK Integration Evidence` workflow is designed for a VPC-connected self-hosted runner. It sends synthetic `trip.completed` events through: -Native AWS service metrics are used where possible. A custom `RideSharing/Serverless::NotificationProcessingFailures` metric is derived from the notification worker log stream because SQS partial-batch failures can occur even when the Lambda invocation itself is not counted as a native Lambda `Errors` datapoint. - -### Alarm coverage - -Terraform creates operational alarms for: +```text +MSK -> Lambda -> SQS -> Lambda -> SNS -> probe SQS +``` -- ride-event processor errors; -- notification worker errors; -- throttling on either Lambda; -- p95 duration approaching the configured Lambda timeout; -- SQS oldest-message age indicating consumer lag; -- any visible message in the DLQ; -- record-level notification processing failures. +and writes min/mean/p50/p95/p99/max wall-clock latency plus a transparent request/runtime cost estimate. -All alarms publish to a dedicated operational SNS topic, separate from application ride notifications. Set `alarm_email` to create an optional email subscription; AWS requires confirmation before that subscription begins receiving notifications. +**Current cloud evidence status:** `evidence/aws-msk-integration-results.json` is `not_run`. No real MSK latency or cost result is claimed yet because this chat/repository session does not have an authorized AWS development account and VPC-connected runner. The workflow/harness is implemented so a real run can generate reviewable evidence later. -Default thresholds are development/reference values, not claimed production SLOs. They are configurable through Terraform variables and should be replaced with thresholds derived from measured traffic and an explicit SLO/error-budget policy in a real deployment. +See [`docs/aws-msk-integration.md`](docs/aws-msk-integration.md). ## Quick start -### Python validation - ```bash git clone https://github.com/CoreyLeath-code/Scalable-Event-Driven-Ride-Sharing-Platform.git cd Scalable-Event-Driven-Ride-Sharing-Platform @@ -163,17 +126,13 @@ python -m pip install -r requirements.txt -r requirements-dev.txt pytest ``` -On Windows PowerShell, activate the environment with `.venv\Scripts\Activate.ps1`. +Windows PowerShell activation: -### Reproduce quality checks and benchmarks - -```bash -make reproduce +```powershell +.venv\Scripts\Activate.ps1 ``` -The reproducibility path generates or refreshes evidence artifacts such as `benchmark-results.json`, `coverage.xml`, and `reproducibility-results.json`. Performance or quality numbers should be cited from those generated artifacts together with the commit, runtime, workload, and hardware context that produced them. - -### Containerized driver-location demo +### Run the local Compose demo ```bash docker compose up --build @@ -181,9 +140,13 @@ curl http://localhost:8000/driver-location/health curl http://localhost:8000/driver-location/ready ``` -The public Compose endpoint uses NGINX in front of driver-location API replicas. The readiness endpoint validates the configured Redis-backed driver store. +### Reproduce quality/benchmark artifacts + +```bash +make reproduce +``` -### Validate the AWS Terraform configuration +### Validate AWS IaC without deploying ```bash cd infra/aws @@ -192,118 +155,98 @@ terraform init -backend=false terraform validate ``` -The repository's AWS validation workflow also checks Terraform formatting and validation on relevant pull requests. - -### Plan CloudWatch alerts and optional email delivery - -With AWS credentials configured for a development account: +### Render Kubernetes environments ```bash -cd infra/aws -terraform init -terraform plan \ - -var='aws_region=us-east-1' \ - -var='alarm_email=operator@example.com' +kustomize build infra/kubernetes/overlays/dev +kustomize build infra/kubernetes/overlays/staging +kustomize build infra/kubernetes/overlays/prod ``` -If `alarm_email` is supplied, the email endpoint must confirm the SNS subscription before alarms can be delivered there. +## Authentication configuration -### Plan the optional Amazon MSK integration +Local development is intentionally credential-free by default: -```bash -cd infra/aws -terraform init -terraform plan \ - -var='aws_region=us-east-1' \ - -var='msk_cluster_arn=arn:aws:kafka:us-east-1:123456789012:cluster/example/uuid' +```text +AUTH_REQUIRED=false +``` + +For an authenticated environment configure: + +```text +AUTH_REQUIRED=true +AUTH_ISSUER=https://your-oidc-issuer +AUTH_AUDIENCE=your-client-or-resource-audience +AUTH_TOKEN_USE=access ``` -Before applying in a real environment, verify MSK authentication mode, network reachability, topic names, encryption requirements, quotas, alarm ownership, and organization-specific IAM controls. +Staging/prod Kustomize overlays contain explicit issuer/audience placeholders so a real deployment cannot be mistaken for a fully configured identity environment. ## Project structure ```text . -|-- .github/workflows/ # CI, AWS validation, hygiene, release, CD -|-- benchmarks/ # Reproducible benchmark harness -|-- docs/ # Architecture, AWS, observability, failure handling +|-- .github/workflows/ +| |-- ci.yml +| |-- aws-serverless.yml +| |-- aws-msk-integration.yml +| |-- kubernetes-validation.yml +| |-- kubernetes-promotion.yml +| `-- kubernetes-rollback.yml +|-- docs/ +|-- evidence/ |-- infra/ -| |-- aws/ # Lambda/SQS/SNS/MSK/CloudWatch Terraform reference -| `-- kubernetes/ # Kubernetes deployment/HPA manifests -|-- load-tests/ # Locust scenario +| |-- aws/ +| `-- kubernetes/ +| `-- overlays/{dev,staging,prod}/ +|-- scripts/ +| |-- aws_msk_e2e.py +| `-- k8s_release_evidence.py |-- serverless/ -| |-- ride_event_processor/ # Amazon MSK -> SQS Lambda -| `-- notification_worker/ # SQS -> SNS Lambda -|-- services/ # Service entrypoint examples -|-- shared/ # Shared config, logging, event schemas/adapters -|-- tests/ # Unit and integration behavior tests +| |-- ride_event_processor/ +| `-- notification_worker/ +|-- tests/ +|-- auth.py +|-- runtime_secrets.py +|-- pii_policy.py |-- Dockerfile |-- docker-compose.yml -|-- Makefile -|-- pyproject.toml -|-- requirements.txt `-- README.md ``` -## Reliability and failure handling - -The project models several failure boundaries instead of assuming a happy path: - -- broker adapters are exercised separately from Docker-free unit tests; -- SQS provides buffering between the AWS event processor and notification worker; -- the SQS queue has a dead-letter queue with a bounded receive count; -- the notification worker uses partial-batch responses for record-level retries; -- Kafka metadata is preserved to support downstream idempotency; -- CloudWatch alarms surface invocation errors, throttles, high p95 duration, queue lag, DLQ activity, and record-level notification failures; -- a dedicated operational SNS topic separates operator alerts from product-facing ride notifications; -- driver-location liveness and readiness are separate endpoints; -- Compose integration validates that the public gateway remains ready after loss of one API replica. +## Validation and test coverage -A real production system would still require persistent idempotency storage, distributed tracing, secrets management, authentication/authorization, PII controls, measured SLOs/error budgets, multi-AZ capacity testing, disaster recovery, payment-provider integration, and documented operational ownership. - -## Validation and CI - -The main CI workflow validates: - -- Docker Compose configuration; -- Black formatting; -- Ruff linting; -- mypy checks for core Python modules; -- pytest behavior and coverage; -- benchmark JSON generation and validation; -- broker integration tests against Docker-hosted Kafka, Redis, and RabbitMQ; -- NGINX/load-balanced driver-location behavior. - -The AWS serverless workflow additionally validates: +| Boundary | Credential-free validation | Real integration path | +| --- | --- | --- | +| API | endpoint/store tests + auth dependency tests | authenticated deployment smoke test is environment-specific | +| Consumer | valid/invalid event behavior + PII-safe logging boundary | broker-fed consumer path | +| Kafka/Redis/RabbitMQ | connection-state contract tests | Docker broker round trips | +| Lambda | injected fake SQS/SNS clients + malformed/PII tests | real MSK workflow | +| Terraform | `fmt`, `init -backend=false`, `validate` | manual development apply | +| Kubernetes | render all Kustomize overlays + manifest policy tests | promotion/rollback workflows | -- Lambda handler unit tests without AWS credentials; -- recursive Terraform formatting; -- Terraform initialization without a backend; -- Terraform static validation, including the CloudWatch dashboard, alarms, SNS alerting, and metric filter resources. +The default pytest marker expression excludes Docker broker integration and real AWS integration markers so unit tests remain credential-free. -## Research-style benchmarks and evidence +## Research-style evidence -Measured evidence belongs in generated, versioned artifacts rather than hand-maintained marketing claims. +Measured evidence belongs in generated/versioned artifacts rather than manually typed marketing numbers. | Area | Reproduction path | Evidence | | --- | --- | --- | -| Event-bus publish/delivery | benchmark harness | `benchmark.event_bus` | -| Matching engine | deterministic synthetic candidates | `benchmark.matching` | -| Driver location store | telemetry upserts | `benchmark.location_store` | -| Pricing engine | synthetic demand/supply inputs | `benchmark.pricing` | -| Test coverage | `pytest --cov=.` | `coverage.xml` | -| Quality checks | `make reproduce` / CI | reproducibility command results | -| AWS Lambda behavior | pytest | `tests/test_aws_lambda_handlers.py` | -| AWS IaC validity | Terraform CI | `terraform validate` | -| CloudWatch resource validity | Terraform CI | `terraform validate` | - -For performance claims, record warm-up behavior, sample count, concurrency, median, p95/p99, throughput, memory, runtime version, commit SHA, and hardware or cloud configuration. +| Local event bus / matching / location / pricing | benchmark harness | `benchmark-results.json` | +| Test coverage | pytest-cov | `coverage.xml` | +| Quality checks | `make reproduce` / CI | `reproducibility-results.json` | +| AWS Lambda behavior | pytest | AWS handler/security tests | +| AWS IaC | Terraform CI | `terraform validate` result | +| CloudWatch resources | Terraform CI | `terraform validate` result | +| Real MSK e2e | manual VPC workflow | `aws-msk-integration-results.json` artifact | +| Kubernetes promotion/rollback | manual workflows | rendered manifest + deployment evidence JSON | -## Architecture targets — not measured production claims +For any performance claim, record the source commit, workload, sample count, concurrency, warm-up behavior, median/p95/p99, runtime configuration, and environment. Cost estimates must state their pricing inputs and excluded charges. -The following remain design targets for a future production deployment. They are **not** asserted as results of the local benchmark harness or the AWS reference path. +## Architecture targets — not measured production claims -| Capability | Target | +| Capability | Design target | | --- | ---: | | Ride request throughput | 10,000+ requests/sec | | Driver telemetry ingestion | 5,000+ events/sec | @@ -313,55 +256,48 @@ The following remain design targets for a future production deployment. They are | Autoscaling response | under 8 seconds | | CI/CD pipeline time | under 90 seconds | +These remain architecture targets until a deployment-specific benchmark produces evidence. + ## Production-readiness boundary -This repository demonstrates engineering patterns and reproducible reference components. A real ride-sharing deployment would still need, at minimum: +The repository now implements authentication hooks, secret retrieval, PII event guards, observability, environment overlays, and deployment evidence workflows. It still does **not** justify a full production claim. Remaining work includes: -- authenticated rider and driver identities; -- authorization and tenant/security boundaries; -- encrypted secrets and PII lifecycle controls; -- production broker/datastore environments; -- payment-provider integration and financial controls; -- durable trip state and transactional consistency rules; -- geospatial indexing appropriate to the selected datastore; -- measured SLOs, paging policy, escalation paths, tracing, and incident-response ownership; -- capacity, chaos, failover, and disaster-recovery testing; -- environment-specific Kubernetes or cloud deployment configuration; -- security review and supply-chain policy enforcement. +- durable idempotency storage and controlled DLQ replay; +- fine-grained authorization/scopes and object-level/tenant access controls; +- OpenTelemetry traces correlating API requests, Kafka offsets, Lambda requests, and SQS messages; +- measured SLIs/SLOs/error budgets with paging and escalation ownership; +- secret rotation procedures and production workload-identity verification; +- capacity, chaos, multi-AZ failover, and disaster-recovery evidence; +- payment/financial controls and durable trip-state consistency; +- production privacy retention/deletion/audit policy; +- security review and supply-chain enforcement tied to the actual deployment. ## Engineering roadmap -1. Add persistent idempotency storage for Lambda side effects and controlled DLQ replay tooling. -2. Add OpenTelemetry traces that correlate API requests, Kafka offsets, Lambda requests, and SQS message IDs. -3. Define measured service-level indicators, SLOs, error budgets, alarm severity, and paging/escalation policy. -4. Add an integration environment using a real development MSK cluster and record measured end-to-end latency/cost results. -5. Add CloudWatch Synthetics or equivalent canary coverage for externally observable service paths where it adds value. -6. Add authentication, secrets management, and explicit PII boundaries before any production claim. -7. Parameterize Kubernetes image names/environments and add deployment promotion/rollback evidence. -8. Expand API, consumer, broker, and AWS integration coverage while keeping unit tests credential-free. +1. Persist Lambda idempotency keys and add controlled DLQ replay tooling. +2. Add OpenTelemetry distributed tracing across API, Kafka/MSK, Lambda, and SQS boundaries. +3. Define measured SLIs, SLOs, error budgets, alarm severity, paging, and escalation policy from real traffic. +4. Add fine-grained scopes/roles and object-level authorization beyond authentication. +5. Add CloudWatch Synthetics or equivalent canaries for externally observable paths. +6. Run the real MSK and Kubernetes workflows in an authorized development environment and commit reviewed evidence artifacts. +7. Add capacity, failover, chaos, and disaster-recovery experiments before any production claim. ## Q&A -**Why use both Kubernetes/services and Lambda?** -They solve different workload shapes. Long-running services are retained for latency-sensitive request and state paths, while Lambda handles asynchronous work that benefits from independent event-driven scaling. - -**Why use native CloudWatch metrics instead of emitting a custom metric for everything?** -Lambda and SQS already publish useful operational metrics. The reference stack reuses those metrics and adds a custom log-derived metric only for record-level SQS notification failures that can be invisible to the native Lambda `Errors` metric. - -**Does this repository deploy Amazon MSK automatically?** -No. Terraform accepts an existing `msk_cluster_arn`. When the value is omitted, the MSK event-source mapping is not created. +**Does the repository now automatically create MSK?** +No. `create_dev_msk_cluster` defaults to `false`. An existing `msk_cluster_arn` can still be supplied, or the optional development MSK Serverless cluster can be explicitly enabled. -**Does the Lambda code require AWS during tests?** -No. Unit tests inject fake SQS/SNS clients. `boto3` is imported lazily only when the handlers need a real AWS SDK client at runtime. +**Are real MSK latency numbers in the README?** +No. The measurement harness exists, but the checked-in evidence says `not_run` until an authorized VPC-connected workflow actually executes. -**Will setting `alarm_email` immediately send alerts?** -No. Terraform can create the SNS email subscription, but AWS requires the recipient to confirm the subscription before delivery begins. +**Are secrets stored in Terraform?** +No secret values are created by Terraform. The optional stack creates secret metadata/KMS/policy resources; the secret value is populated out-of-band and fetched at runtime. -**Is the AWS path production-ready?** -No production claim is made. It is a deployable reference implementation with retries, a DLQ, scoped IAM roles, CloudWatch monitoring/alerting, tests, and Terraform validation. A real deployment still needs environment-specific networking, authentication, persistent idempotency, tracing, measured SLOs, security, capacity, and operational evidence. +**Is authentication enough for production authorization?** +No. JWT verification establishes an authenticated identity boundary. Fine-grained roles/scopes, tenant boundaries, and object-level policy are still required. -**Where should measured results live?** -In versioned generated artifacts tied to a commit and reproduction command, not as uncited numbers in the README. +**Why keep containers/Kubernetes if Lambda exists?** +The workloads have different shapes. Long-running services remain appropriate for latency-sensitive/stateful request paths, while Lambda handles independently scalable asynchronous work. ## License diff --git a/docs/aws-msk-integration.md b/docs/aws-msk-integration.md new file mode 100644 index 0000000..b16a208 --- /dev/null +++ b/docs/aws-msk-integration.md @@ -0,0 +1,68 @@ +# Real Amazon MSK Development Integration Evidence + +This repository includes a deliberately opt-in path for measuring the real AWS event chain: + +```text +MSK Serverless -> Lambda ride-event processor -> SQS -> Lambda notification worker -> SNS -> probe SQS +``` + +The local/unit test suite does **not** require AWS credentials. Real-cloud evidence is produced only by the manual `Real AWS MSK Integration Evidence` workflow. + +## Why the workflow requires a VPC-connected runner + +The development MSK Serverless cluster is private and uses IAM authentication. The integration producer must therefore run from a network path that can reach the MSK VPC. The workflow targets a self-hosted runner labeled: + +```text +self-hosted, linux, aws-msk-dev +``` + +This avoids making the Kafka brokers public merely so a GitHub-hosted runner can reach them. + +## AWS resources + +When `create_dev_msk_cluster=true`, Terraform can create: + +- a private MSK Serverless cluster; +- an MSK security group; +- the existing Lambda event-source mapping using the created cluster ARN; +- an optional SNS-subscribed SQS probe queue when `enable_integration_probe=true`. + +Both features are disabled by default because they create billable AWS resources. + +## Authentication + +The development cluster enables MSK IAM authentication. The Lambda execution role receives the scoped `kafka` and `kafka-cluster` permissions needed by the existing event-source mapping. The measurement runner uses the AWS MSK IAM SASL signer and short-lived AWS workload credentials. + +AWS reference: https://docs.aws.amazon.com/lambda/latest/dg/msk-cluster-auth.html + +## Manual workflow prerequisites + +Create a GitHub `development` environment with: + +- secret `AWS_MSK_INTEGRATION_ROLE_ARN`: IAM role trusted by GitHub OIDC and allowed to manage the development integration resources; +- a VPC-connected self-hosted runner carrying the `aws-msk-dev` label. + +When dispatching the workflow, provide: + +- AWS region; +- development VPC ID; +- a JSON array containing at least two private subnet IDs; +- probe sample count; +- whether the temporary stack should be destroyed after the run. + +## Evidence artifact + +`scripts/aws_msk_e2e.py` publishes synthetic `trip.completed` events containing only pseudonymous operational IDs. For each event it waits until the matching `probe_id` reaches the SNS-subscribed probe queue. + +The generated JSON records: + +- sample count; +- min/mean/p50/p95/p99/max end-to-end latency; +- cluster ARN, region, topic, and elapsed test time; +- an explicit request-count/runtime cost estimate based on pricing inputs supplied to the workflow. + +The cost value is labeled an **estimate**, not AWS billing data. It intentionally does not claim to include every charge such as data transfer or CloudWatch ingestion. For billed-cost evidence, correlate the run with AWS Cost Explorer after billing data is available. + +## Current evidence status + +`evidence/aws-msk-integration-results.json` is checked in with `status: not_run`. That is intentional. No cloud latency or cost number should be copied into the README until an authorized real-AWS workflow run produces the evidence artifact and that artifact is reviewed/committed with its source commit and environment details. diff --git a/docs/aws-serverless-architecture.md b/docs/aws-serverless-architecture.md index fb907d3..4d57d27 100644 --- a/docs/aws-serverless-architecture.md +++ b/docs/aws-serverless-architecture.md @@ -1,19 +1,20 @@ # AWS Serverless Extension -This repository keeps the latency-sensitive driver-location and matching paths as long-running services while adding AWS Lambda for asynchronous event processing. The AWS path is an optional deployment reference; it does not replace the local Kafka/Redis/Docker workflow. +This repository keeps latency-sensitive driver-location and matching paths as long-running services while using AWS Lambda for asynchronous ride-event processing. ## Event flow ```mermaid flowchart LR Client[Rider / Driver Client] --> API[FastAPI / API Gateway] - API --> Kafka[(Kafka / Amazon MSK)] - Kafka --> Matching[Matching + Trip Services] - Kafka --> Processor[AWS Lambda\nride-event-processor] + API --> MSK[(Kafka / Amazon MSK)] + MSK --> Matching[Matching + Trip Services] + MSK --> Processor[AWS Lambda\nride-event-processor] Processor --> Queue[(Amazon SQS\nprocessed-events)] Queue --> Worker[AWS Lambda\nnotification-worker] Queue -. repeated failures .-> DLQ[(SQS DLQ)] Worker --> SNS[(Amazon SNS\nride-notifications)] + SNS -. integration evidence .-> Probe[(SQS probe queue)] Processor --> CW[CloudWatch Logs + Metrics] Worker --> CW @@ -24,55 +25,40 @@ flowchart LR Alarms --> OpsSNS[(SNS operational-alerts)] ``` -## Why Lambda is used here - -Lambda is reserved for asynchronous work that can scale independently from the request path: - -- decoding and validating Amazon MSK event batches; -- preserving Kafka topic, partition, and offset as an idempotency key; -- handing events to SQS for retry isolation and buffering; -- publishing selected lifecycle events to SNS; -- reporting per-message SQS failures so successful records are not retried unnecessarily. - -Real-time matching and driver-location serving remain container/service concerns in this reference architecture. That avoids claiming that serverless is automatically the best choice for every latency-sensitive workload. - ## Delivery semantics -The MSK processor intentionally fails its invocation when a record cannot be decoded or SQS cannot accept the message. Kafka/Lambda delivery is therefore treated as **at least once**. Every forwarded envelope contains an `idempotency_key` derived from `topic:partition:offset`; downstream production consumers should persist and enforce that key before applying non-idempotent side effects. +The MSK processor fails an invocation when a record cannot be decoded, contains prohibited direct PII, or cannot be handed to SQS. The path is therefore treated as **at least once**. Every forwarded envelope carries a Kafka-derived `topic:partition:offset` idempotency key. A production implementation still needs persistent idempotency enforcement before non-idempotent side effects. -The notification worker uses SQS partial-batch failure responses. Invalid or failed messages are returned in `batchItemFailures`, allowing successful messages in the same batch to remain complete. After the configured receive limit, repeatedly failing messages move to the dead-letter queue. +The notification worker uses SQS partial-batch failure responses. Repeatedly failing records eventually move to the DLQ. ## Terraform layout ```text infra/aws/ -|-- versions.tf -|-- variables.tf |-- main.tf +|-- dev_msk.tf |-- observability.tf +|-- security.tf +|-- variables.tf |-- outputs.tf +|-- versions.tf `-- build/ # generated Lambda ZIPs are ignored ``` -The Terraform configuration creates: - -- `ride-event-processor` Lambda; -- `notification-worker` Lambda; -- processed-events SQS queue; -- processed-events dead-letter queue; -- ride-notifications SNS topic; -- CloudWatch log groups with configurable retention; -- CloudWatch dashboard for Lambda/SQS operational health; -- CloudWatch alarms for errors, throttles, high p95 duration, SQS lag, DLQ activity, and record-level notification failures; -- dedicated operational-alerts SNS topic with optional email subscription; -- separate Lambda execution roles with scoped SQS/SNS permissions; -- optional Amazon MSK event-source mapping. +The reference stack can define: -See [`cloudwatch-observability.md`](cloudwatch-observability.md) for dashboard panels, alarm thresholds, triage guidance, and the monitoring production boundary. +- two Python Lambda functions; +- processed-events SQS queue + DLQ; +- rider/driver notification SNS topic; +- CloudWatch dashboard, alarms, log groups, and operator SNS alerts; +- optional existing MSK event-source mapping; +- optional private IAM-authenticated MSK Serverless development cluster; +- optional SNS->SQS integration probe sink; +- optional KMS-protected Secrets Manager metadata and least-privilege reader policy. -## Validate without deploying +All billable development integration and secret resources are disabled by default. -No AWS credentials are required for formatting and static validation: +## Validate without AWS credentials ```bash cd infra/aws @@ -81,74 +67,18 @@ terraform init -backend=false terraform validate ``` -The Lambda handlers are covered by normal repository tests and use injected fake AWS clients in tests, so unit validation does not require `boto3`, AWS credentials, or a live AWS account. - -## Plan with CloudWatch alert delivery - -```bash -cd infra/aws -terraform init -terraform plan \ - -var='aws_region=us-east-1' \ - -var='alarm_email=operator@example.com' -``` +The regular AWS unit tests inject fake clients and do not contact AWS. -When `alarm_email` is supplied, AWS requires the email recipient to confirm the SNS subscription before alarm messages are delivered. +## Real MSK evidence -## Plan with an Amazon MSK cluster +See [`aws-msk-integration.md`](aws-msk-integration.md). A VPC-connected manual workflow can create a temporary MSK Serverless cluster, run synthetic probes through the complete asynchronous path, generate latency/cost-estimate JSON, and optionally destroy the temporary stack. -```bash -cd infra/aws -terraform init -terraform plan \ - -var='aws_region=us-east-1' \ - -var='msk_cluster_arn=arn:aws:kafka:us-east-1:123456789012:cluster/example/uuid' -``` - -`msk_cluster_arn` defaults to `null`. When it is omitted, Terraform still creates the Lambda/SQS/SNS/CloudWatch reference stack but does not create the MSK event-source mapping. - -Before applying this against a real environment, verify the cluster's authentication mode, network reachability, Kafka topic name, encryption settings, service quotas, organization-specific IAM requirements, alarm ownership, and notification routing. Production workloads should also add tracing, secret management, idempotency persistence, load testing, measured SLOs/error budgets, and cost controls appropriate to the environment. - -## Lambda event contracts - -### Amazon MSK input - -The processor expects the standard Lambda MSK record shape with base64-encoded JSON in each record's `value` field. The decoded object must contain `event_type` or `type`. - -Example decoded event: - -```json -{ - "event_type": "ride.requested", - "ride_id": "ride-123", - "rider_id": "rider-7" -} -``` - -Forwarded SQS envelope: - -```json -{ - "idempotency_key": "ride.events:0:42", - "source": "amazon-msk", - "topic": "ride.events", - "partition": 0, - "offset": 42, - "event": { - "event_type": "ride.requested", - "ride_id": "ride-123", - "rider_id": "rider-7" - } -} -``` +No real-cloud number is claimed until that workflow has actually run. -### SQS notification input +## Security boundaries -The notification worker publishes these event types to SNS: +See [`security-and-pii.md`](security-and-pii.md). The Lambda ZIPs contain the direct-PII policy so the event boundary is enforced in the deployed function artifact itself rather than only in local application code. -- `driver.matched` -- `trip.started` -- `trip.completed` -- `payment.processed` +## Observability -Other valid events are acknowledged and skipped rather than treated as failures. +See [`cloudwatch-observability.md`](cloudwatch-observability.md) for dashboards, alarm thresholds, Logs Insights coverage, and the monitoring production boundary. diff --git a/docs/kubernetes-promotion.md b/docs/kubernetes-promotion.md new file mode 100644 index 0000000..f51d10c --- /dev/null +++ b/docs/kubernetes-promotion.md @@ -0,0 +1,54 @@ +# Kubernetes Environment Promotion and Rollback Evidence + +The Kubernetes manifests use a Kustomize base with explicit `dev`, `staging`, and `prod` overlays. + +## Layout + +```text +infra/kubernetes/ +|-- api-gateway-deployment.yaml +|-- api-gateway-hpa.yaml +|-- kustomization.yaml +`-- overlays/ + |-- dev/kustomization.yaml + |-- staging/kustomization.yaml + `-- prod/kustomization.yaml +``` + +The base deployment uses the logical image name `api-gateway:local`. Each overlay supplies the registry/repository and an environment placeholder tag. Promotion workflows replace that image in the ephemeral workflow workspace; repository manifests do not use `latest`. + +## Environment boundaries + +- `dev`: authentication disabled by default for controlled development. +- `staging`: authentication required; issuer/audience placeholders must be replaced before a real apply. +- `prod`: authentication required and deployment evidence requires an immutable `@sha256:` image digest. + +Each overlay also sets a non-secret Secrets Manager identifier such as `ride-sharing/prod/runtime`. Workload credentials—not static AWS access keys—should authorize the pod to read that secret. + +## Promotion workflow + +`Kubernetes Promotion Evidence` supports two modes: + +1. render-only (default): build the chosen overlay, validate it client-side, and upload the rendered manifest plus evidence JSON; +2. apply: load environment-scoped `KUBE_CONFIG_DATA`, capture the current deployment revision, apply the rendered manifest, wait for rollout success, then record before/after revisions. + +The evidence JSON records: + +- source Git commit; +- environment; +- image reference; +- whether the manifest was only rendered or actually applied; +- SHA-256 of the rendered manifest; +- before/after deployment revisions when available. + +## Rollback workflow + +`Kubernetes Rollback Evidence` requires an explicit revision. It captures the current revision, executes `kubectl rollout undo --to-revision`, waits for rollout success, snapshots the resulting Deployment YAML, and uploads a rollback evidence artifact. + +## GitHub environment secret + +For actual cluster changes, configure `KUBE_CONFIG_DATA` as a base64-encoded kubeconfig in each GitHub Environment (`dev`, `staging`, `prod`). In a stronger production setup, replace static kubeconfig material with short-lived cloud workload identity and environment protection rules. + +## Current evidence status + +`evidence/kubernetes-deployment-results.json` says `not_run` until a promotion or rollback workflow is explicitly executed. Rendered manifests and workflow success are reproducible evidence; a real deployment/rollback claim requires the corresponding uploaded workflow artifact. diff --git a/docs/security-and-pii.md b/docs/security-and-pii.md new file mode 100644 index 0000000..e54992a --- /dev/null +++ b/docs/security-and-pii.md @@ -0,0 +1,81 @@ +# Authentication, Secrets, and PII Boundaries + +This repository implements explicit security boundaries while keeping local development and unit tests credential-free. + +## API authentication boundary + +Sensitive driver-location collection endpoints use the FastAPI `require_authenticated_request` dependency: + +- `/driver-location/drivers` +- `/driver-location/drivers/{driver_id}` +- `/driver-location/count` + +Liveness and readiness remain unauthenticated so Kubernetes and load balancers can probe the service: + +- `/driver-location/health` +- `/driver-location/ready` + +Authentication is disabled by default for local development. Set `AUTH_REQUIRED=true` to enforce RS256 OIDC JWT verification. + +Required production/staging configuration: + +```text +AUTH_REQUIRED=true +AUTH_ISSUER=https:// +AUTH_AUDIENCE= +AUTH_TOKEN_USE=access +``` + +`AUTH_JWKS_URL` may override the default `/.well-known/jwks.json` location. The verifier accepts an OIDC `aud` claim or Cognito-style `client_id` audience and verifies the configured issuer. + +Authentication is not the same as a complete authorization model. A production system still needs scopes/roles, object-level access checks, tenant boundaries, and policy ownership. + +## Runtime secrets + +`runtime_secrets.py` resolves the driver Redis connection in this order: + +1. `DRIVER_LOCATION_REDIS_SECRET_ID` -> AWS Secrets Manager JSON field `redis_url`; +2. local-development fallback `DRIVER_LOCATION_REDIS_URL`. + +The application does not log the secret value or raw Secrets Manager response. + +Terraform can optionally create: + +- a customer-managed KMS key with rotation enabled; +- Secrets Manager secret **metadata**; +- a least-privilege IAM policy containing `secretsmanager:GetSecretValue` and `kms:Decrypt` for only that secret/key. + +Terraform intentionally does **not** create a `secret_version`; secret material is populated out-of-band so it is not written into Terraform configuration or state by this reference implementation. + +Example secret JSON: + +```json +{ + "redis_url": "redis://private-host:6379/0" +} +``` + +AWS recommends Secrets Manager instead of Lambda/application environment variables for credentials and sensitive authorization material. + +## Direct PII policy + +The asynchronous event path rejects fields whose names represent direct identifiers, including: + +- email addresses; +- phone numbers; +- first/full/last names; +- home/street addresses; +- SSNs; +- payment card, bank account, routing, or CVV fields. + +The Lambda MSK processor rejects such payloads before forwarding them to SQS. The notification worker applies the same check before SNS publication, so direct injection into the queue cannot bypass the boundary. + +Pseudonymous operational identifiers such as `ride_id`, `driver_id`, and `probe_id` are permitted by this narrow guard. They can still be personal data in a real system and require retention/access policies appropriate to the deployment. + +## Logging boundary + +The in-memory event bus no longer logs message bodies. The driver telemetry consumer no longer logs coordinates or validation exception details that may echo input values. Operational logs contain topic names, record counts, message IDs, and pseudonymous IDs rather than raw rider/driver payloads. + +## Production-claim rule + +These controls demonstrate implemented boundaries; they are not a certification or proof of regulatory compliance. A production claim still requires threat modeling, authorization design, key/secret rotation procedures, retention/deletion policy, audit logging, access reviews, incident response, and verification against the actual deployment environment. From 584962572fa0e4978e51f7fb9f5f9adc58d7339c Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Mon, 31 Aug 2026 22:26:55 -0400 Subject: [PATCH 07/21] test: expand API and consumer hardening coverage --- .github/workflows/kubernetes-promotion.yml | 17 ++++++ tests/test_api_security_integration.py | 25 +++++++++ tests/test_consumer_privacy.py | 64 ++++++++++++++++++++++ 3 files changed, 106 insertions(+) create mode 100644 tests/test_api_security_integration.py create mode 100644 tests/test_consumer_privacy.py diff --git a/.github/workflows/kubernetes-promotion.yml b/.github/workflows/kubernetes-promotion.yml index 8f5ea94..545c3c6 100644 --- a/.github/workflows/kubernetes-promotion.yml +++ b/.github/workflows/kubernetes-promotion.yml @@ -45,6 +45,23 @@ jobs: kustomize build . > "$RUNNER_TEMP/rendered.yaml" grep -n "image:" "$RUNNER_TEMP/rendered.yaml" + - name: Enforce promotion policy before apply + env: + TARGET_ENV: ${{ inputs.environment }} + IMAGE: ${{ inputs.image }} + DEPLOY: ${{ inputs.deploy }} + run: | + set -euo pipefail + if [[ "$TARGET_ENV" == "prod" && "$IMAGE" != *@sha256:* ]]; then + echo "Production promotion requires an immutable @sha256 image digest." >&2 + exit 1 + fi + if [[ "$DEPLOY" == "true" && "$TARGET_ENV" != "dev" ]] && \ + grep -q "replace-me" "$RUNNER_TEMP/rendered.yaml"; then + echo "Staging/prod OIDC placeholders must be replaced before an actual deployment." >&2 + exit 1 + fi + - name: Client-side manifest validation run: kubectl apply --dry-run=client --validate=false -f "$RUNNER_TEMP/rendered.yaml" diff --git a/tests/test_api_security_integration.py b/tests/test_api_security_integration.py new file mode 100644 index 0000000..501f8ee --- /dev/null +++ b/tests/test_api_security_integration.py @@ -0,0 +1,25 @@ +from fastapi.routing import APIRoute + +import api_router +from auth import require_authenticated_request + + +def route_by_path(path: str) -> APIRoute: + for route in api_router.router.routes: + if isinstance(route, APIRoute) and route.path == path: + return route + raise AssertionError(f"route not found: {path}") + + +def dependency_calls(route: APIRoute): + return {dependency.call for dependency in route.dependant.dependencies} + + +def test_sensitive_driver_routes_wire_auth_dependency(): + for path in ("/drivers", "/drivers/{driver_id}", "/count"): + assert require_authenticated_request in dependency_calls(route_by_path(path)) + + +def test_health_and_readiness_remain_probeable_without_auth_dependency(): + for path in ("/health", "/ready"): + assert require_authenticated_request not in dependency_calls(route_by_path(path)) diff --git a/tests/test_consumer_privacy.py b/tests/test_consumer_privacy.py new file mode 100644 index 0000000..37d67e9 --- /dev/null +++ b/tests/test_consumer_privacy.py @@ -0,0 +1,64 @@ +import pytest + +from consumer import DriverLocationConsumer + + +class RecordingStore: + def __init__(self): + self.events = [] + + def upsert_driver(self, event): + self.events.append(event) + + +class RecordingLogger: + def __init__(self): + self.entries = [] + + def info(self, message, *args): + self.entries.append(("info", message, args)) + + def error(self, message, *args): + self.entries.append(("error", message, args)) + + +@pytest.mark.asyncio +async def test_consumer_logs_do_not_include_coordinates(): + logger = RecordingLogger() + consumer = DriverLocationConsumer(event_bus=None, store=RecordingStore()) + consumer.logger = logger + + await consumer.handle_driver_location( + { + "driver_id": "driver-1", + "lat": 40.7128, + "lon": -74.006, + "timestamp": "2026-08-05T12:00:00Z", + "status": "available", + } + ) + + rendered = repr(logger.entries) + assert "40.7128" not in rendered + assert "-74.006" not in rendered + assert "driver-1" in rendered + + +@pytest.mark.asyncio +async def test_invalid_consumer_log_does_not_echo_input_values(): + logger = RecordingLogger() + consumer = DriverLocationConsumer(event_bus=None, store=RecordingStore()) + consumer.logger = logger + + await consumer.handle_driver_location( + { + "driver_id": "sensitive@example.com", + "lat": "invalid-coordinate", + "lon": -74.006, + "status": "available", + } + ) + + rendered = repr(logger.entries) + assert "sensitive@example.com" not in rendered + assert "invalid-coordinate" not in rendered From e2daf67eae3611d9530ac0157f4536cc6caccfc9 Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Mon, 31 Aug 2026 22:31:37 -0400 Subject: [PATCH 08/21] fix: normalize Terraform and Kustomize validation --- infra/aws/main.tf | 9 ++- .../base/api-gateway-deployment.yaml | 55 +++++++++++++++++++ infra/kubernetes/base/api-gateway-hpa.yaml | 18 ++++++ infra/kubernetes/base/kustomization.yaml | 19 +++++++ infra/kubernetes/kustomization.yaml | 16 +----- .../overlays/dev/kustomization.yaml | 2 +- .../overlays/prod/kustomization.yaml | 2 +- .../overlays/staging/kustomization.yaml | 2 +- tests/test_aws_security_boundaries.py | 15 +++-- tests/test_kubernetes_manifests.py | 5 +- 10 files changed, 114 insertions(+), 29 deletions(-) create mode 100644 infra/kubernetes/base/api-gateway-deployment.yaml create mode 100644 infra/kubernetes/base/api-gateway-hpa.yaml create mode 100644 infra/kubernetes/base/kustomization.yaml diff --git a/infra/aws/main.tf b/infra/aws/main.tf index ae3c6e8..f9557b9 100644 --- a/infra/aws/main.tf +++ b/infra/aws/main.tf @@ -1,7 +1,10 @@ locals { - processor_function_name = "${var.name_prefix}-ride-event-processor" - notification_function_name = "${var.name_prefix}-notification-worker" - effective_msk_cluster_arn = var.msk_cluster_arn != null ? var.msk_cluster_arn : try(aws_msk_serverless_cluster.dev[0].arn, null) + effective_msk_cluster_arn = var.msk_cluster_arn != null ? var.msk_cluster_arn : try(aws_msk_serverless_cluster.dev[0].arn, null) +} + +locals { + processor_function_name = "${var.name_prefix}-ride-event-processor" + notification_function_name = "${var.name_prefix}-notification-worker" msk_topic_arn = local.effective_msk_cluster_arn == null ? null : "${replace( local.effective_msk_cluster_arn, ":cluster/", diff --git a/infra/kubernetes/base/api-gateway-deployment.yaml b/infra/kubernetes/base/api-gateway-deployment.yaml new file mode 100644 index 0000000..12c1f67 --- /dev/null +++ b/infra/kubernetes/base/api-gateway-deployment.yaml @@ -0,0 +1,55 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + name: api-gateway + labels: + app.kubernetes.io/name: api-gateway + app.kubernetes.io/part-of: ride-sharing-platform +spec: + replicas: 2 + revisionHistoryLimit: 10 + strategy: + type: RollingUpdate + rollingUpdate: + maxUnavailable: 0 + maxSurge: 1 + selector: + matchLabels: + app.kubernetes.io/name: api-gateway + template: + metadata: + labels: + app.kubernetes.io/name: api-gateway + app.kubernetes.io/part-of: ride-sharing-platform + spec: + containers: + - name: api-gateway + image: api-gateway:local + imagePullPolicy: IfNotPresent + envFrom: + - configMapRef: + name: ride-sharing-runtime + ports: + - name: http + containerPort: 8000 + readinessProbe: + httpGet: + path: /driver-location/ready + port: http + initialDelaySeconds: 5 + periodSeconds: 10 + timeoutSeconds: 3 + livenessProbe: + httpGet: + path: /driver-location/health + port: http + initialDelaySeconds: 10 + periodSeconds: 20 + timeoutSeconds: 3 + resources: + requests: + cpu: 100m + memory: 128Mi + limits: + cpu: 500m + memory: 512Mi diff --git a/infra/kubernetes/base/api-gateway-hpa.yaml b/infra/kubernetes/base/api-gateway-hpa.yaml new file mode 100644 index 0000000..f8af884 --- /dev/null +++ b/infra/kubernetes/base/api-gateway-hpa.yaml @@ -0,0 +1,18 @@ +apiVersion: autoscaling/v2 +kind: HorizontalPodAutoscaler +metadata: + name: api-gateway-hpa +spec: + scaleTargetRef: + apiVersion: apps/v1 + kind: Deployment + name: api-gateway + minReplicas: 2 + maxReplicas: 10 + metrics: + - type: Resource + resource: + name: cpu + target: + type: Utilization + averageUtilization: 70 diff --git a/infra/kubernetes/base/kustomization.yaml b/infra/kubernetes/base/kustomization.yaml new file mode 100644 index 0000000..eb6297e --- /dev/null +++ b/infra/kubernetes/base/kustomization.yaml @@ -0,0 +1,19 @@ +apiVersion: kustomize.config.k8s.io/v1beta1 +kind: Kustomization + +resources: + - api-gateway-deployment.yaml + - api-gateway-hpa.yaml + +configMapGenerator: + - name: ride-sharing-runtime + literals: + - APP_ENV=local + - AUTH_REQUIRED=false + - AUTH_ISSUER= + - AUTH_AUDIENCE= + - AUTH_TOKEN_USE=access + - DRIVER_LOCATION_REDIS_SECRET_ID= + +generatorOptions: + disableNameSuffixHash: true diff --git a/infra/kubernetes/kustomization.yaml b/infra/kubernetes/kustomization.yaml index eb6297e..271aa1a 100644 --- a/infra/kubernetes/kustomization.yaml +++ b/infra/kubernetes/kustomization.yaml @@ -2,18 +2,4 @@ apiVersion: kustomize.config.k8s.io/v1beta1 kind: Kustomization resources: - - api-gateway-deployment.yaml - - api-gateway-hpa.yaml - -configMapGenerator: - - name: ride-sharing-runtime - literals: - - APP_ENV=local - - AUTH_REQUIRED=false - - AUTH_ISSUER= - - AUTH_AUDIENCE= - - AUTH_TOKEN_USE=access - - DRIVER_LOCATION_REDIS_SECRET_ID= - -generatorOptions: - disableNameSuffixHash: true + - base diff --git a/infra/kubernetes/overlays/dev/kustomization.yaml b/infra/kubernetes/overlays/dev/kustomization.yaml index d372942..02a3231 100644 --- a/infra/kubernetes/overlays/dev/kustomization.yaml +++ b/infra/kubernetes/overlays/dev/kustomization.yaml @@ -2,7 +2,7 @@ apiVersion: kustomize.config.k8s.io/v1beta1 kind: Kustomization resources: - - ../.. + - ../../base labels: - pairs: diff --git a/infra/kubernetes/overlays/prod/kustomization.yaml b/infra/kubernetes/overlays/prod/kustomization.yaml index ec43754..574a94a 100644 --- a/infra/kubernetes/overlays/prod/kustomization.yaml +++ b/infra/kubernetes/overlays/prod/kustomization.yaml @@ -2,7 +2,7 @@ apiVersion: kustomize.config.k8s.io/v1beta1 kind: Kustomization resources: - - ../.. + - ../../base labels: - pairs: diff --git a/infra/kubernetes/overlays/staging/kustomization.yaml b/infra/kubernetes/overlays/staging/kustomization.yaml index 7d0c9c7..cf5d680 100644 --- a/infra/kubernetes/overlays/staging/kustomization.yaml +++ b/infra/kubernetes/overlays/staging/kustomization.yaml @@ -2,7 +2,7 @@ apiVersion: kustomize.config.k8s.io/v1beta1 kind: Kustomization resources: - - ../.. + - ../../base labels: - pairs: diff --git a/tests/test_aws_security_boundaries.py b/tests/test_aws_security_boundaries.py index 4c26fb3..f0454ff 100644 --- a/tests/test_aws_security_boundaries.py +++ b/tests/test_aws_security_boundaries.py @@ -30,9 +30,14 @@ def msk_event(payload): return { "records": { "ride.events-0": [ - {"topic": "ride.events", "partition": 0, "offset": 1, "value": value} - ] - } + { + "topic": "ride.events", + "partition": 0, + "offset": 1, + "value": value, + }, + ], + }, } @@ -51,9 +56,7 @@ def test_msk_processor_rejects_direct_pii(monkeypatch): def test_notification_worker_returns_partial_failure_for_direct_pii(monkeypatch): monkeypatch.setenv("NOTIFICATION_TOPIC_ARN", "arn:aws:sns:us-east-1:123:topic") sns = RecordingSns() - body = json.dumps( - {"event": {"event_type": "trip.completed", "phone_number": "+15555550100"}} - ) + body = json.dumps({"event": {"event_type": "trip.completed", "phone_number": "+15555550100"}}) result = notification_handler.lambda_handler( {"Records": [{"messageId": "bad-pii", "body": body}]}, None, diff --git a/tests/test_kubernetes_manifests.py b/tests/test_kubernetes_manifests.py index 38fb10a..0fec83e 100644 --- a/tests/test_kubernetes_manifests.py +++ b/tests/test_kubernetes_manifests.py @@ -6,7 +6,7 @@ def test_base_deployment_has_parameterized_image_and_rollout_safety(): - deployment = (K8S / "api-gateway-deployment.yaml").read_text(encoding="utf-8") + deployment = (K8S / "base" / "api-gateway-deployment.yaml").read_text(encoding="utf-8") assert "image: api-gateway:local" in deployment assert "revisionHistoryLimit: 10" in deployment assert "maxUnavailable: 0" in deployment @@ -15,12 +15,13 @@ def test_base_deployment_has_parameterized_image_and_rollout_safety(): def test_environment_overlays_are_explicit_and_do_not_use_latest(): + expected_image = "newName: ghcr.io/coreyleath-code/scalable-event-driven-ride-sharing-platform" for environment in ("dev", "staging", "prod"): overlay = (K8S / "overlays" / environment / "kustomization.yaml").read_text( encoding="utf-8" ) assert f"APP_ENV={environment}" in overlay - assert "newName: ghcr.io/coreyleath-code/scalable-event-driven-ride-sharing-platform" in overlay + assert expected_image in overlay assert ":latest" not in overlay From 310b313c87a5a27c9a585f8965bd7046ea5ec72b Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Mon, 31 Aug 2026 22:33:56 -0400 Subject: [PATCH 09/21] fix: install pytest in Kubernetes validation --- .github/workflows/kubernetes-validation.yml | 3 +++ 1 file changed, 3 insertions(+) diff --git a/.github/workflows/kubernetes-validation.yml b/.github/workflows/kubernetes-validation.yml index 573a1e8..ba12cb5 100644 --- a/.github/workflows/kubernetes-validation.yml +++ b/.github/workflows/kubernetes-validation.yml @@ -46,5 +46,8 @@ jobs: with: python-version: "3.11" + - name: Install test dependency + run: python -m pip install pytest + - name: Run manifest/evidence tests run: pytest -q tests/test_kubernetes_manifests.py tests/test_release_evidence_tools.py From 77688ee6e2d7a06995a23363a4deb0dc641a2edd Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Mon, 31 Aug 2026 22:34:03 -0400 Subject: [PATCH 10/21] fix: normalize Kubernetes manifest tests --- tests/test_kubernetes_manifests.py | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/tests/test_kubernetes_manifests.py b/tests/test_kubernetes_manifests.py index 0fec83e..c4f8047 100644 --- a/tests/test_kubernetes_manifests.py +++ b/tests/test_kubernetes_manifests.py @@ -15,11 +15,13 @@ def test_base_deployment_has_parameterized_image_and_rollout_safety(): def test_environment_overlays_are_explicit_and_do_not_use_latest(): - expected_image = "newName: ghcr.io/coreyleath-code/scalable-event-driven-ride-sharing-platform" + expected_image = ( + "newName: ghcr.io/coreyleath-code/" + "scalable-event-driven-ride-sharing-platform" + ) for environment in ("dev", "staging", "prod"): - overlay = (K8S / "overlays" / environment / "kustomization.yaml").read_text( - encoding="utf-8" - ) + overlay_path = K8S / "overlays" / environment / "kustomization.yaml" + overlay = overlay_path.read_text(encoding="utf-8") assert f"APP_ENV={environment}" in overlay assert expected_image in overlay assert ":latest" not in overlay @@ -27,7 +29,6 @@ def test_environment_overlays_are_explicit_and_do_not_use_latest(): def test_staging_and_prod_require_authentication(): for environment in ("staging", "prod"): - overlay = (K8S / "overlays" / environment / "kustomization.yaml").read_text( - encoding="utf-8" - ) + overlay_path = K8S / "overlays" / environment / "kustomization.yaml" + overlay = overlay_path.read_text(encoding="utf-8") assert "AUTH_REQUIRED=true" in overlay From 21e4cb949c0270878adfd8b4168c691dcde2e1e1 Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Mon, 31 Aug 2026 22:36:39 -0400 Subject: [PATCH 11/21] fix: keep MSK evidence helpers dependency-light --- scripts/aws_msk_e2e.py | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/scripts/aws_msk_e2e.py b/scripts/aws_msk_e2e.py index 716c284..42d2b5f 100644 --- a/scripts/aws_msk_e2e.py +++ b/scripts/aws_msk_e2e.py @@ -10,16 +10,14 @@ from pathlib import Path from typing import Any -import boto3 -from aws_msk_iam_sasl_signer import MSKAuthTokenProvider -from kafka import KafkaProducer - class MskTokenProvider: def __init__(self, region: str): self.region = region def token(self) -> str: + from aws_msk_iam_sasl_signer import MSKAuthTokenProvider + token, _ = MSKAuthTokenProvider.generate_auth_token(self.region) return token @@ -73,7 +71,9 @@ def env_float(name: str) -> float: return float(value) if value else 0.0 -def build_producer(bootstrap_servers: list[str], region: str) -> KafkaProducer: +def build_producer(bootstrap_servers: list[str], region: str) -> Any: + from kafka import KafkaProducer + return KafkaProducer( bootstrap_servers=bootstrap_servers, security_protocol="SASL_SSL", @@ -104,6 +104,8 @@ def receive_probe(sqs_client: Any, queue_url: str, probe_id: str, timeout_second def main() -> None: + import boto3 + parser = argparse.ArgumentParser() parser.add_argument("--cluster-arn", required=True) parser.add_argument("--topic", default="ride.events") From ba1d776c3d1fdd71e020ccc87459a16c61f2bf08 Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Mon, 31 Aug 2026 22:36:48 -0400 Subject: [PATCH 12/21] fix: match Black formatting for manifest tests --- tests/test_kubernetes_manifests.py | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/tests/test_kubernetes_manifests.py b/tests/test_kubernetes_manifests.py index c4f8047..9f949bc 100644 --- a/tests/test_kubernetes_manifests.py +++ b/tests/test_kubernetes_manifests.py @@ -15,10 +15,7 @@ def test_base_deployment_has_parameterized_image_and_rollout_safety(): def test_environment_overlays_are_explicit_and_do_not_use_latest(): - expected_image = ( - "newName: ghcr.io/coreyleath-code/" - "scalable-event-driven-ride-sharing-platform" - ) + expected_image = "newName: ghcr.io/coreyleath-code/scalable-event-driven-ride-sharing-platform" for environment in ("dev", "staging", "prod"): overlay_path = K8S / "overlays" / environment / "kustomization.yaml" overlay = overlay_path.read_text(encoding="utf-8") From 004f05c1b10c1948f57230c41c0be588bae36e68 Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Mon, 31 Aug 2026 22:39:03 -0400 Subject: [PATCH 13/21] ci: show Black diff in release gate --- .github/workflows/release.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 9cc3b77..861b5d8 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -65,7 +65,7 @@ jobs: - name: Run quality gates run: | - black --check . --line-length=100 + black --diff --check . --line-length=100 ruff check . mypy models.py utils.py event_bus.py location_store.py matching_engine.py pricing_engine.py consumer.py --ignore-missing-imports pytest From 3407c49aa8059410698ed1b084308b250dda0283 Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Tue, 1 Sep 2026 11:18:00 -0400 Subject: [PATCH 14/21] fix: format release evidence tests --- tests/test_release_evidence_tools.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/tests/test_release_evidence_tools.py b/tests/test_release_evidence_tools.py index 3cf50f2..d1c1f2c 100644 --- a/tests/test_release_evidence_tools.py +++ b/tests/test_release_evidence_tools.py @@ -3,7 +3,11 @@ import pytest -from scripts.aws_msk_e2e import estimate_variable_cost_usd, percentile, unwrap_sns_sqs_body +from scripts.aws_msk_e2e import ( + estimate_variable_cost_usd, + percentile, + unwrap_sns_sqs_body, +) from scripts.k8s_release_evidence import build_evidence, manifest_sha256 @@ -26,7 +30,9 @@ def test_cost_estimate_uses_explicit_pricing_inputs(): def test_unwrap_sns_sqs_body_returns_original_notification_payload(): - body = json.dumps({"Message": json.dumps({"event_type": "trip.completed", "probe_id": "p1"})}) + body = json.dumps( + {"Message": json.dumps({"event_type": "trip.completed", "probe_id": "p1"})} + ) assert unwrap_sns_sqs_body(body)["probe_id"] == "p1" From d7a12f4e24ad16c5a9d764bfa2dbfa779cd2d958 Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Tue, 1 Sep 2026 11:20:53 -0400 Subject: [PATCH 15/21] fix: format Kubernetes manifest tests --- tests/test_kubernetes_manifests.py | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/test_kubernetes_manifests.py b/tests/test_kubernetes_manifests.py index 9f949bc..5e12cad 100644 --- a/tests/test_kubernetes_manifests.py +++ b/tests/test_kubernetes_manifests.py @@ -1,6 +1,5 @@ from pathlib import Path - ROOT = Path(__file__).resolve().parents[1] K8S = ROOT / "infra" / "kubernetes" From b3d94cacee09b561f0c19b94faf754879aa5a7aa Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Tue, 1 Sep 2026 11:21:34 -0400 Subject: [PATCH 16/21] fix: match Black formatting in release tests --- tests/test_release_evidence_tools.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/tests/test_release_evidence_tools.py b/tests/test_release_evidence_tools.py index d1c1f2c..6f96453 100644 --- a/tests/test_release_evidence_tools.py +++ b/tests/test_release_evidence_tools.py @@ -30,9 +30,7 @@ def test_cost_estimate_uses_explicit_pricing_inputs(): def test_unwrap_sns_sqs_body_returns_original_notification_payload(): - body = json.dumps( - {"Message": json.dumps({"event_type": "trip.completed", "probe_id": "p1"})} - ) + body = json.dumps({"Message": json.dumps({"event_type": "trip.completed", "probe_id": "p1"})}) assert unwrap_sns_sqs_body(body)["probe_id"] == "p1" From a533f6a0caea37b293c296813a247f4db696cfde Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Tue, 1 Sep 2026 11:23:12 -0400 Subject: [PATCH 17/21] ci: pin Black target to Python 3.11 --- pyproject.toml | 1 + 1 file changed, 1 insertion(+) diff --git a/pyproject.toml b/pyproject.toml index 02b5f0f..3d9ff4f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -54,6 +54,7 @@ namespaces = true [tool.black] line-length = 100 +target-version = ["py311"] [tool.pytest.ini_options] pythonpath = ["."] From 70356f46250cb2ac829295bede70a41af226e68d Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Tue, 1 Sep 2026 11:23:18 -0400 Subject: [PATCH 18/21] ci: pin Black formatter version --- requirements-dev.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/requirements-dev.txt b/requirements-dev.txt index 40accf8..b2908ad 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -1,6 +1,6 @@ # Linters and formatters flake8 -black +black==26.5.1 mypy ruff From b39950f306215113696436f60c41acf9cf998f25 Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Tue, 1 Sep 2026 11:24:26 -0400 Subject: [PATCH 19/21] fix: satisfy Ruff for FastAPI dependency injection --- auth.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/auth.py b/auth.py index 3e9fe5d..ef22b14 100644 --- a/auth.py +++ b/auth.py @@ -68,7 +68,7 @@ def verify_bearer_token(token: str) -> dict[str, Any]: async def require_authenticated_request( - credentials: HTTPAuthorizationCredentials | None = Depends(_bearer), + credentials: HTTPAuthorizationCredentials | None = Depends(_bearer), # noqa: B008 ) -> dict[str, Any] | None: """Require a valid bearer token only when AUTH_REQUIRED is enabled.""" if not auth_required(): From 1fc8d1980a14b4c07dfa4118b09b1f6fa1ea51f5 Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Tue, 1 Sep 2026 11:24:37 -0400 Subject: [PATCH 20/21] fix: use TypeError for invalid secret payload type --- runtime_secrets.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/runtime_secrets.py b/runtime_secrets.py index bee383e..3fbff99 100644 --- a/runtime_secrets.py +++ b/runtime_secrets.py @@ -25,7 +25,7 @@ def load_secret_json(secret_id: str, *, client: Any = None) -> dict[str, Any]: raise RuntimeError("Secrets Manager secret must contain valid JSON") from exc if not isinstance(payload, dict): - raise RuntimeError("Secrets Manager secret JSON must be an object") + raise TypeError("Secrets Manager secret JSON must be an object") return payload From c8f0229c4ea36047c088bd6029469993000a8d69 Mon Sep 17 00:00:00 2001 From: Corey Leath Date: Tue, 1 Sep 2026 11:25:03 -0400 Subject: [PATCH 21/21] fix: use TypeError for malformed SNS message type --- scripts/aws_msk_e2e.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scripts/aws_msk_e2e.py b/scripts/aws_msk_e2e.py index 42d2b5f..1623915 100644 --- a/scripts/aws_msk_e2e.py +++ b/scripts/aws_msk_e2e.py @@ -39,7 +39,7 @@ def unwrap_sns_sqs_body(body: str) -> dict[str, Any]: outer = json.loads(body) message = outer.get("Message") if not isinstance(message, str): - raise ValueError("SQS probe message does not contain an SNS Message string") + raise TypeError("SQS probe message does not contain an SNS Message string") payload = json.loads(message) if not isinstance(payload, dict): raise TypeError("SNS Message must decode to a JSON object")