feat(security): validate nation_slug + idempotent Stripe webhook (fixes #13)

infrastructure/template.yaml

@@ -126,6 +126,32 @@ Resources:
         - Key: Project
           Value: nat
 
+  # StripeEventsTable - records processed Stripe webhook event IDs so duplicate
+  # deliveries (Stripe retries) are no-ops. TTL purges old markers automatically.
+  StripeEventsTable:
+    Type: AWS::DynamoDB::Table
+    DeletionPolicy: Retain
+    UpdateReplacePolicy: Retain
+    Properties:
+      TableName: !Sub 'nat-stripe-events-${Environment}'
+      BillingMode: PAY_PER_REQUEST
+      AttributeDefinitions:
+        - AttributeName: event_id
+          AttributeType: S
+      KeySchema:
+        - AttributeName: event_id
+          KeyType: HASH
+      TimeToLiveSpecification:
+        AttributeName: expires_at
+        Enabled: true
+      PointInTimeRecoverySpecification:
+        PointInTimeRecoveryEnabled: true
+      Tags:
+        - Key: Environment
+          Value: !Ref Environment
+        - Key: Project
+          Value: nat
+
   # SessionStateTable - server-authoritative confirmation + undo state
   # Holds per-session destructive-action confirmations and undo history so that
   # neither can be forged via the request body. Rows expire via the TTL.
@@ -213,6 +239,8 @@ Resources:
                   - !Sub '${TenantsTable.Arn}/index/*'
                   - !GetAtt UsersTable.Arn
                   - !Sub '${UsersTable.Arn}/index/*'
+                  - !GetAtt StripeEventsTable.Arn
+                  - !Sub '${StripeEventsTable.Arn}/index/*'
                   - !GetAtt SessionStateTable.Arn
                   - !Sub '${SessionStateTable.Arn}/index/*'
                   - !GetAtt OAuthStateTable.Arn
@@ -335,6 +363,7 @@ Resources:
           NATIONS_TABLE: !Ref NationsTable
           TENANTS_TABLE: !Ref TenantsTable
           USERS_TABLE: !Ref UsersTable
+          STRIPE_EVENTS_TABLE: !Ref StripeEventsTable
           STRIPE_WEBHOOK_SECRET_NAME: !Sub 'nat/stripe-webhook-secret-${Environment}'
       Tags:
         - Key: Environment
@@ -1015,6 +1044,18 @@ Outputs:
     Export:
       Name: !Sub '${AWS::StackName}-UsersTableArn'
 
+  StripeEventsTableName:
+    Description: Name of the Stripe events idempotency DynamoDB table
+    Value: !Ref StripeEventsTable
+    Export:
+      Name: !Sub '${AWS::StackName}-StripeEventsTableName'
+
+  StripeEventsTableArn:
+    Description: ARN of the Stripe events idempotency DynamoDB table
+    Value: !GetAtt StripeEventsTable.Arn
+    Export:
+      Name: !Sub '${AWS::StackName}-StripeEventsTableArn'
+
   SessionStateTableName:
     Description: Name of the Session State DynamoDB table
     Value: !Ref SessionStateTable

src/lambdas/nat_agent/handler.py

@@ -38,6 +38,11 @@
         authenticate_request,
     )
 
+try:  # Resolve in both pytest (repo root) and flattened Lambda packages.
+    from src.lambdas.shared.validation import is_valid_nation_slug
+except ModuleNotFoundError:  # pragma: no cover - exercised only in Lambda
+    from shared.validation import is_valid_nation_slug  # type: ignore[no-redef]
+
 logger = logging.getLogger()
 logger.setLevel(logging.INFO)
 
@@ -373,6 +378,17 @@ def handler(event: dict[str, Any], context: Any) -> LambdaResponse:
         user_id = session.user_id
         nation_slug = session.nation_slug
 
+        # Defense in depth: the slug is token-attested (minted post-validation at
+        # OAuth connect), but it is interpolated into Secrets Manager / DynamoDB
+        # lookups, so reject a malformed value rather than build a bad key.
+        if not is_valid_nation_slug(nation_slug):
+            logger.error(f"Invalid nation_slug in session token: {nation_slug!r}")
+            return {
+                "statusCode": 400,
+                "body": json.dumps({"error": "Invalid nation_slug format"}),
+                "headers": headers,
+            }
+
         # Extract request payload (identity fields are ignored if present).
         query = body.get("query")
         page_context = body.get("context", {})

src/lambdas/nat_agent_streaming/handler.py

@@ -58,6 +58,11 @@
         record_pending_confirmation,
     )
 
+try:  # Resolve in both pytest (repo root) and flattened Lambda packages.
+    from src.lambdas.shared.validation import is_valid_nation_slug
+except ModuleNotFoundError:  # pragma: no cover - exercised only in Lambda
+    from shared.validation import is_valid_nation_slug  # type: ignore[no-redef]
+
 logger = logging.getLogger()
 logger.setLevel(logging.INFO)
 
@@ -604,6 +609,17 @@ async def process_streaming_request(body: dict[str, Any]) -> AsyncGenerator[str,
         })
         return
 
+    # Defense in depth: the slug is token-attested (minted post-validation at
+    # OAuth connect), but it is interpolated into Secrets Manager / DynamoDB
+    # lookups, so reject a malformed value rather than build a bad key.
+    if not is_valid_nation_slug(nation_slug):
+        logger.error(f"Invalid nation_slug in session token: {nation_slug!r}")
+        yield format_sse_event(SSE_EVENT_ERROR, {
+            "error": "Invalid nation_slug format",
+            "error_code": "BAD_REQUEST",
+        })
+        return
+
     logger.info(f"Processing streaming query for nation {nation_slug}, user {user_id}: {query[:100]}...")
 
     # Check if billing cycle has reset for this nation (must run before the

src/lambdas/nb_oauth_callback/handler.py

@@ -13,6 +13,7 @@
 import json
 import logging
 import os
+import re
 from datetime import datetime, timezone
 from typing import Any, TypedDict
 from urllib.parse import parse_qs, urlencode
@@ -59,6 +60,19 @@
     "ERROR_REDIRECT_URL", "https://natassistant.com/connection-error"
 )
 
+# The nation slug arrives in the (client-supplied) OAuth ``state`` and is the
+# most security-sensitive entry point: it is interpolated into the Secrets
+# Manager path ``nat/nation/{slug}/nb-tokens`` and the NationBuilder token URL
+# host ``https://{slug}.nationbuilder.com``. Reject malformed values before
+# either is constructed. Mirrors shared.validation.NATION_SLUG_PATTERN (this
+# Lambda is packaged without shared/, so the pattern is inlined).
+NATION_SLUG_PATTERN = re.compile(r"^[a-z0-9-]{1,63}\Z")
+
+
+def is_valid_nation_slug(slug: Any) -> bool:
+    """Return True if *slug* is a non-empty, well-formed nation slug."""
+    return isinstance(slug, str) and NATION_SLUG_PATTERN.match(slug) is not None
+
 
 class LambdaResponse(TypedDict):
     """Lambda response type."""
@@ -439,6 +453,19 @@ def handler(event: dict[str, Any], context: Any) -> LambdaResponse:
         nb_slug = state_data["nb_slug"]
         redirect_uri = state_data["redirect_uri"]
 
+        if not user_id or not nb_slug or not redirect_uri:
+            logger.error("State missing required fields")
+            error_url = f"{ERROR_REDIRECT_URL}?error=invalid_state"
+            return create_redirect_response(error_url)
+
+        # Reject a malformed slug before it reaches Secrets Manager / the NB token
+        # URL. This endpoint is a browser OAuth redirect, so a rejected slug is
+        # surfaced as an error redirect rather than a 400 JSON body.
+        if not is_valid_nation_slug(nb_slug):
+            logger.error(f"Invalid nation_slug in OAuth state: {nb_slug!r}")
+            error_url = f"{ERROR_REDIRECT_URL}?error=invalid_nation"
+            return create_redirect_response(error_url)
+
         # admin_email is no longer carried in the (now trusted, single-use)
         # state. Populating it from NationBuilder userinfo is a follow-up.
         user_email = None

src/lambdas/shared/validation.py

@@ -0,0 +1,43 @@
+"""
+Input validation helpers shared across Lambda handlers.
+
+The canonical home for ``nation_slug`` format validation. A nation slug flows
+directly into DynamoDB keys, Secrets Manager secret names
+(``nat/nation/{slug}/nb-tokens``) and NationBuilder API URLs
+(``https://{slug}.nationbuilder.com``), so every entry point that accepts one
+must reject malformed values before they are used.
+
+Only the Lambdas that bundle ``shared/`` at deploy time (the agent handlers)
+import this module. The standalone handlers (stripe_checkout, stripe_webhook,
+nb_oauth_callback) are packaged without ``shared/`` and inline an identical
+``NATION_SLUG_PATTERN`` instead.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+# NationBuilder slugs are lowercase alphanumeric plus hyphen. The 63-char ceiling
+# matches a DNS label (slugs appear as the subdomain of *.nationbuilder.com).
+NATION_SLUG_PATTERN = re.compile(r"^[a-z0-9-]{1,63}\Z")
+
+
+class InvalidNationSlugError(ValueError):
+    """Raised when a nation_slug does not match the required format."""
+
+
+def is_valid_nation_slug(slug: Any) -> bool:
+    """Return True if *slug* is a non-empty, well-formed nation slug."""
+    return isinstance(slug, str) and NATION_SLUG_PATTERN.match(slug) is not None
+
+
+def validate_nation_slug(slug: Any) -> str:
+    """Return *slug* unchanged if valid, else raise :class:`InvalidNationSlugError`."""
+    if not is_valid_nation_slug(slug):
+        raise InvalidNationSlugError(
+            f"Invalid nation_slug format: {slug!r} (must match ^[a-z0-9-]{{1,63}}$)"
+        )
+    # is_valid_nation_slug guarantees a str; assert narrows it for the type checker.
+    assert isinstance(slug, str)
+    return slug

src/lambdas/stripe_checkout/handler.py

@@ -10,6 +10,7 @@
 import json
 import logging
 import os
+import re
 from typing import Any, TypedDict
 
 import boto3
@@ -18,6 +19,17 @@
 logger = logging.getLogger()
 logger.setLevel(logging.INFO)
 
+# nation_slug is validated here because it is forwarded to Stripe as checkout
+# metadata and later used by the webhook as a DynamoDB key. Mirrors
+# shared.validation.NATION_SLUG_PATTERN (this Lambda is packaged without
+# shared/, so the pattern is inlined).
+NATION_SLUG_PATTERN = re.compile(r"^[a-z0-9-]{1,63}\Z")
+
+
+def is_valid_nation_slug(slug: Any) -> bool:
+    """Return True if *slug* is a non-empty, well-formed nation slug."""
+    return isinstance(slug, str) and NATION_SLUG_PATTERN.match(slug) is not None
+
 # Environment variables
 STRIPE_SECRET_KEY_NAME = os.environ.get(
     "STRIPE_SECRET_KEY_NAME", "nat/stripe-secret-key"
@@ -174,6 +186,15 @@ def handler(event: dict[str, Any], context: Any) -> LambdaResponse:
                 "headers": headers,
             }
 
+        if not is_valid_nation_slug(nation_slug):
+            return {
+                "statusCode": 400,
+                "body": json.dumps({
+                    "error": "Invalid nation_slug format (must match ^[a-z0-9-]{1,63}$)"
+                }),
+                "headers": headers,
+            }
+
         if plan not in STRIPE_PRICE_IDS:
             return {
                 "statusCode": 400,