feat(auth): accept portal Cognito ID tokens (DEV-238) (#192)

* feat(auth): accept portal Cognito ID tokens

* fix(auth): distinguish portal verification outages

* docs(auth): document Cognito workspace variables

README.md

@@ -111,7 +111,9 @@ sit in ALARM between runs). Alarm names follow `meal-order-manager-<fn>-<signal>
 
 ## Authentication
 
-Google Identity Services (OAuth) with tokeninfo endpoint verification. Accepts both `seahavenind.com` and `seahaven.com` Google Workspace domains. Cloud form generation and Lambda order submission fail closed unless `/meal-order-manager/google-client-id` is configured. The local Flask workflow can still use manual name and email entry when Google auth is not configured.
+Google Identity Services and portal Cognito ID tokens coexist until portal cutover. Google tokens use the tokeninfo endpoint. Portal tokens are verified locally against the configured Cognito issuer, audience, signature, expiry, token use, and email domain. Both paths accept only `seahavenind.com` and `seahaven.com` identities and fail closed when their SSM configuration is unavailable. The local Flask workflow can still use manual name and email entry when Google auth is not configured.
+
+Set the `portal_cognito_issuer` and `portal_cognito_audience` HCP Terraform workspace variables from the matching internal-portal stage outputs. Switch both values together when moving from dev validation to the production portal pool.
 
 ## Admin Panel
 

functions/admin_authorizer/handler.py

@@ -1,12 +1,9 @@
 """API Gateway (HTTP API) Lambda authorizer for the meal-order-manager admin API.
 
-Gates every ``/api/admin/*`` route at the gateway so they are no longer
-AuthorizationType NONE. The authorizer validates the same Google ID token the
-admin panel already sends in the ``Authorization: Bearer <token>`` header and
-confirms the caller is in the ``admin_emails`` allow-list (DynamoDB
-CONFIG/SETTINGS) — exactly the checks ``submit_order``'s ``_verify_admin`` does
-in-handler today. No client change is required: the existing admin UI already
-sends this header.
+Gates every ``/api/admin/*`` route at the gateway. The authorizer accepts the
+existing Google ID token or a portal Cognito ID token in the Authorization
+header, then confirms the caller is in the ``admin_emails`` allow-list
+(DynamoDB CONFIG/SETTINGS).
 
 Returns the HTTP API v2 *simple response* shape (``{"isAuthorized": bool}``);
 a False result causes API Gateway to return 403 before the integration runs.
@@ -21,6 +18,11 @@ import urllib.error
 import urllib.parse
 import urllib.request
 
+from shared.cognito import (
+    CognitoVerificationUnavailable,
+    looks_like_cognito_token,
+    verify_cognito_id_token,
+)
 from shared.db import get_settings
 from shared.secrets import get_parameter
 
@@ -89,20 +91,33 @@ def _extract_token(event) -> str:
     return ""
 
 
-def lambda_handler(event, context):
-    if not os.environ.get("GOOGLE_CLIENT_ID_PARAM", ""):
-        logger.error("Authorizer misconfigured: GOOGLE_CLIENT_ID_PARAM unset")
-        return _DENY
+def _verify_portal_token(token: str) -> dict | None:
+    return verify_cognito_id_token(
+        token,
+        os.environ.get("PORTAL_COGNITO_ISSUER_PARAM", ""),
+        os.environ.get("PORTAL_COGNITO_AUDIENCE_PARAM", ""),
+    )
 
+
+def lambda_handler(event, context):
     token = _extract_token(event)
     if not token:
         return _DENY
 
+    if looks_like_cognito_token(token):
+        try:
+            user_info = _verify_portal_token(token)
+        except CognitoVerificationUnavailable:
+            logger.error("Cognito verification service unavailable; denying")
+            return _DENY
+    else:
+        if not os.environ.get("GOOGLE_CLIENT_ID_PARAM", ""):
+            logger.error("Authorizer misconfigured: GOOGLE_CLIENT_ID_PARAM unset")
+            return _DENY
         client_id = _get_google_client_id()
         if not client_id:
             logger.error("Google client ID unavailable; denying")
             return _DENY
-
         user_info = _verify_google_token(token, client_id)
     if user_info is None:
         return _DENY

functions/submit_order/handler.py

@@ -13,6 +13,11 @@ from zoneinfo import ZoneInfo
 
 import boto3
 
+from shared.cognito import (
+    CognitoVerificationUnavailable,
+    looks_like_cognito_token,
+    verify_cognito_id_token,
+)
 from shared.db import (
     current_week,
     delete_order,
@@ -156,25 +161,47 @@ def _authentication_service_unavailable() -> dict:
     return response(503, {"error": "Authentication service temporarily unavailable"})
 
 
-def _verify_admin(event) -> tuple[dict | None, dict | None]:
-    """Verify Google auth and admin access. Returns (user_info, error_response)."""
-    if not _google_auth_configured():
-        return None, response(403, {"error": "Authentication not configured"})
+def _extract_bearer_token(event) -> str:
+    headers = event.get("headers") or {}
+    raw = headers.get("authorization") or headers.get("Authorization") or ""
+    if raw.lower().startswith("bearer "):
+        return raw[7:].strip()
+    return ""
 
-    token = (
-        event.get("headers", {})
-        .get("authorization", "")
-        .removeprefix("Bearer ")
-        .strip()
+
+def _verify_portal_token(token: str) -> dict | None:
+    return verify_cognito_id_token(
+        token,
+        os.environ.get("PORTAL_COGNITO_ISSUER_PARAM", ""),
+        os.environ.get("PORTAL_COGNITO_AUDIENCE_PARAM", ""),
     )
+
+
+def _verify_admin(event) -> tuple[dict | None, dict | None]:
+    """Verify portal or Google auth and admin access."""
+    token = _extract_bearer_token(event)
     if not token:
         return None, response(403, {"error": "Authentication required"})
 
+    if looks_like_cognito_token(token):
+        try:
+            user_info = _verify_portal_token(token)
+        except CognitoVerificationUnavailable:
+            return None, _authentication_service_unavailable()
+        if user_info is None:
+            return None, response(
+                403, {"error": "Invalid or unauthorized portal account"}
+            )
+    else:
+        if not _google_auth_configured():
+            return None, response(403, {"error": "Authentication not configured"})
         user_info, status = _verify_google_token(token)
         if status == "unavailable":
             return None, _authentication_service_unavailable()
         if user_info is None:
-        return None, response(403, {"error": "Invalid or unauthorized Google account"})
+            return None, response(
+                403, {"error": "Invalid or unauthorized Google account"}
+            )
 
     if user_info["email"].lower() not in _get_admin_emails():
         return None, response(403, {"error": "Admin access required"})
@@ -519,10 +546,18 @@ def handle_submit(event):
     except json.JSONDecodeError:
         return response(400, {"error": "Invalid JSON"})
 
-    # The Lambda is the cloud submission path, so Google authentication is always
-    # required. Local/manual submissions are handled only by src/server/app.py.
+    portal_token = _extract_bearer_token(event)
+    if portal_token:
+        try:
+            user_info = _verify_portal_token(portal_token)
+        except CognitoVerificationUnavailable:
+            return _authentication_service_unavailable()
+        if user_info is None:
+            return response(403, {"error": "Invalid or unauthorized portal account"})
+    else:
+        # The Lambda is the cloud submission path, so one of the two verified
+        # identity providers is required. Local/manual submissions remain in Flask.
         google_token = body.get("google_id_token")
-
         try:
             client_id = _get_google_client_id()
         except Exception as exc:

src/shared/requirements.txt

@@ -1,2 +1,3 @@
 boto3==1.43.78
 fpdf2==2.8.8
+PyJWT[crypto]==2.14.0

src/shared/shared/cognito.py

@@ -0,0 +1,119 @@
+"""Verification for portal Cognito ID tokens."""
+
+import logging
+import re
+from functools import lru_cache
+
+import jwt
+from jwt import PyJWKClient
+from jwt.exceptions import PyJWKClientConnectionError, PyJWKClientError, PyJWTError
+
+from shared.secrets import get_parameter
+
+logger = logging.getLogger(__name__)
+
+ALLOWED_EMAIL_DOMAINS = {"seahavenind.com", "seahaven.com"}
+_COGNITO_ISSUER_RE = re.compile(
+    r"^https://cognito-idp\.[a-z0-9-]+\.amazonaws\.com/[A-Za-z0-9_-]+$"
+)
+
+
+class CognitoVerificationUnavailable(RuntimeError):
+    """Trusted Cognito configuration or JWKS could not be loaded."""
+
+
+def looks_like_cognito_token(token: str) -> bool:
+    """Return whether untrusted claims identify a Cognito ID token."""
+    try:
+        claims = jwt.decode(
+            token,
+            options={
+                "verify_signature": False,
+                "verify_exp": False,
+                "verify_aud": False,
+            },
+        )
+    except PyJWTError:
+        return False
+    issuer = claims.get("iss")
+    return (
+        isinstance(issuer, str)
+        and bool(_COGNITO_ISSUER_RE.fullmatch(issuer))
+        and claims.get("token_use") == "id"
+    )
+
+
+@lru_cache(maxsize=4)
+def _jwk_client(issuer: str) -> PyJWKClient:
+    return PyJWKClient(
+        f"{issuer}/.well-known/jwks.json",
+        cache_keys=True,
+        lifespan=300,
+        timeout=5,
+    )
+
+
+def verify_cognito_id_token(
+    token: str, issuer_param: str, audience_param: str
+) -> dict | None:
+    """Verify a portal token and return its trusted identity claims."""
+    if not token or not issuer_param or not audience_param:
+        logger.error("Cognito verification is not configured")
+        return None
+
+    try:
+        issuer = get_parameter(issuer_param, decrypt=False).rstrip("/")
+        audience = get_parameter(audience_param, decrypt=False)
+    except Exception as exc:
+        logger.error("Failed to load Cognito verification parameters: %s", exc)
+        raise CognitoVerificationUnavailable from exc
+
+    if not _COGNITO_ISSUER_RE.fullmatch(issuer) or not audience:
+        logger.error("Cognito verification parameters are invalid")
+        raise CognitoVerificationUnavailable
+
+    try:
+        signing_key = _jwk_client(issuer).get_signing_key_from_jwt(token)
+        claims = jwt.decode(
+            token,
+            signing_key.key,
+            algorithms=["RS256"],
+            audience=audience,
+            issuer=issuer,
+            options={
+                "require": [
+                    "aud",
+                    "email",
+                    "email_verified",
+                    "exp",
+                    "iat",
+                    "iss",
+                    "name",
+                    "token_use",
+                ]
+            },
+        )
+    except PyJWKClientConnectionError as exc:
+        logger.error("Cognito JWKS is unavailable: %s", type(exc).__name__)
+        raise CognitoVerificationUnavailable from exc
+    except OSError as exc:
+        logger.error("Cognito JWKS is unavailable: %s", type(exc).__name__)
+        raise CognitoVerificationUnavailable from exc
+    except (PyJWKClientError, PyJWTError, TypeError, ValueError) as exc:
+        logger.warning("Cognito token rejected: %s", type(exc).__name__)
+        return None
+
+    email = claims.get("email")
+    name = claims.get("name")
+    if claims.get("token_use") != "id":
+        return None
+    if claims.get("email_verified") is not True:
+        return None
+    if not isinstance(email, str) or not isinstance(name, str):
+        return None
+    if not email.strip() or not name.strip() or "@" not in email:
+        return None
+    if email.rsplit("@", 1)[1].lower() not in ALLOWED_EMAIL_DOMAINS:
+        return None
+
+    return {"name": name.strip(), "email": email.strip()}

terraform/lambda.tf

@@ -39,6 +39,8 @@ resource "aws_lambda_function" "submit_order" {
     variables = merge(local.common_env, {
       SLACK_NOTIFIER_ARN            = aws_lambda_function.slack_notifier.arn
       GOOGLE_CLIENT_ID_PARAM        = local.google_client_id_param
+      PORTAL_COGNITO_ISSUER_PARAM   = aws_ssm_parameter.portal_cognito_issuer.name
+      PORTAL_COGNITO_AUDIENCE_PARAM = aws_ssm_parameter.portal_cognito_audience.name
     })
   }
 
@@ -71,6 +73,8 @@ resource "aws_lambda_function" "admin_authorizer" {
   environment {
     variables = merge(local.common_env, {
       GOOGLE_CLIENT_ID_PARAM        = local.google_client_id_param
+      PORTAL_COGNITO_ISSUER_PARAM   = aws_ssm_parameter.portal_cognito_issuer.name
+      PORTAL_COGNITO_AUDIENCE_PARAM = aws_ssm_parameter.portal_cognito_audience.name
     })
   }
 

terraform/locals.tf

@@ -22,6 +22,9 @@ locals {
   # (data.tf) but never owns it.
   google_client_id_param = "${local.ssm_prefix}/google-client-id"
 
+  portal_cognito_issuer_param   = "${local.ssm_prefix}/portal-cognito-issuer"
+  portal_cognito_audience_param = "${local.ssm_prefix}/portal-cognito-audience"
+
   # shared/slack.py resolves the token by NAME, while the IAM grant is scoped to
   # the ARN in var.slack_bot_secret_arn. Both must refer to the same secret.
   slack_bot_secret_name = "${local.project}/slack-bot-token"

terraform/ssm.tf

@@ -11,6 +11,20 @@ resource "aws_ssm_parameter" "slack_channel_id" {
   description = "Slack channel ID for meal order notifications"
 }
 
+resource "aws_ssm_parameter" "portal_cognito_issuer" {
+  name        = local.portal_cognito_issuer_param
+  type        = "String"
+  value       = var.portal_cognito_issuer
+  description = "Trusted portal Cognito user-pool issuer for ID-token verification"
+}
+
+resource "aws_ssm_parameter" "portal_cognito_audience" {
+  name        = local.portal_cognito_audience_param
+  type        = "String"
+  value       = var.portal_cognito_audience
+  description = "Trusted portal Cognito app client ID for ID-token verification"
+}
+
 # ---------------------------------------------------------------------------
 # Deploy-time lookups
 # ---------------------------------------------------------------------------

terraform/terraform.tfvars.example

@@ -16,3 +16,8 @@ slack_bot_secret_arn = "arn:aws:secretsmanager:us-east-1:011934824531:secret:mea
 # Slack channel that receives meal order notifications. Written to
 # /meal-order-manager/slack-channel-id.
 slack_channel_id = "C00000000000"
+
+# Portal Cognito pool and public app client accepted by the meals API. Use the
+# dev pool during portal validation, then switch both values together at cutover.
+portal_cognito_issuer   = "https://cognito-idp.us-east-1.amazonaws.com/us-east-1_EXAMPLE"
+portal_cognito_audience = "examplepublicappclientid"

terraform/variables.tf

@@ -31,6 +31,26 @@ variable "slack_channel_id" {
   type        = string
 }
 
+variable "portal_cognito_issuer" {
+  description = "Exact issuer URL for the portal Cognito user pool whose ID tokens meal-order-manager accepts."
+  type        = string
+
+  validation {
+    condition     = can(regex("^https://cognito-idp\\.[a-z0-9-]+\\.amazonaws\\.com/[A-Za-z0-9_-]+$", var.portal_cognito_issuer))
+    error_message = "portal_cognito_issuer must be an exact Cognito user-pool issuer URL without a trailing slash."
+  }
+}
+
+variable "portal_cognito_audience" {
+  description = "Portal Cognito app client ID required in accepted ID-token aud claims."
+  type        = string
+
+  validation {
+    condition     = length(trimspace(var.portal_cognito_audience)) > 0
+    error_message = "portal_cognito_audience must not be empty."
+  }
+}
+
 variable "checkcomponents_queue_url" {
   description = "paychex-checkcomponents SQS URL. Empty skips the weekly SendMessage."
   type        = string

tests/test_admin_authorizer.py

@@ -62,6 +62,57 @@ def test_valid_user_but_not_admin_denied(mock_cid, mock_verify, mock_settings):
     assert authorizer.lambda_handler(_event(), None) == {"isAuthorized": False}
 
 
+@patch("admin_authorizer_handler.get_settings")
+@patch(
+    "admin_authorizer_handler._verify_portal_token",
+    return_value={"name": "Portal Admin", "email": ADMIN_EMAIL},
+)
+@patch("admin_authorizer_handler.looks_like_cognito_token", return_value=True)
+def test_valid_portal_admin_allowed(mock_looks_like, mock_verify, mock_settings):
+    mock_settings.return_value = {"admin_emails": [ADMIN_EMAIL]}
+
+    assert authorizer.lambda_handler(_event("portal-id-token"), None) == {
+        "isAuthorized": True
+    }
+
+
+@patch("admin_authorizer_handler.get_settings")
+@patch(
+    "admin_authorizer_handler._verify_portal_token",
+    return_value={"name": "Portal User", "email": "user@seahavenind.com"},
+)
+@patch("admin_authorizer_handler.looks_like_cognito_token", return_value=True)
+def test_portal_non_admin_denied(mock_looks_like, mock_verify, mock_settings):
+    mock_settings.return_value = {"admin_emails": [ADMIN_EMAIL]}
+
+    assert authorizer.lambda_handler(_event("portal-id-token"), None) == {
+        "isAuthorized": False
+    }
+
+
+@patch("admin_authorizer_handler._verify_portal_token", return_value=None)
+@patch("admin_authorizer_handler.looks_like_cognito_token", return_value=True)
+def test_invalid_portal_token_denied_without_google_fallback(
+    mock_looks_like, mock_verify
+):
+    with patch("admin_authorizer_handler._get_google_client_id") as mock_google:
+        assert authorizer.lambda_handler(_event("bad-portal-token"), None) == {
+            "isAuthorized": False
+        }
+    mock_google.assert_not_called()
+
+
+@patch(
+    "admin_authorizer_handler._verify_portal_token",
+    side_effect=authorizer.CognitoVerificationUnavailable,
+)
+@patch("admin_authorizer_handler.looks_like_cognito_token", return_value=True)
+def test_portal_verification_outage_denied(mock_looks_like, mock_verify):
+    assert authorizer.lambda_handler(_event("portal-id-token"), None) == {
+        "isAuthorized": False
+    }
+
+
 @patch.dict(
     os.environ, {"GOOGLE_CLIENT_ID_PARAM": "/meal-order-manager/google-client-id"}
 )

tests/test_cognito.py

@@ -0,0 +1,140 @@
+"""Unit tests for portal Cognito ID-token verification."""
+
+import time
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import jwt
+import pytest
+from cryptography.hazmat.primitives.asymmetric import rsa
+
+from shared import cognito
+
+ISSUER = "https://cognito-idp.us-east-1.amazonaws.com/us-east-1_TEST"
+AUDIENCE = "portal-client-id"
+ISSUER_PARAM = "/meal-order-manager/portal-cognito-issuer"
+AUDIENCE_PARAM = "/meal-order-manager/portal-cognito-audience"
+
+
+@pytest.fixture(scope="module")
+def signing_key():
+    private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+    return private_key, private_key.public_key()
+
+
+def _claims(**overrides):
+    now = int(time.time())
+    claims = {
+        "iss": ISSUER,
+        "aud": AUDIENCE,
+        "token_use": "id",
+        "exp": now + 300,
+        "iat": now,
+        "email": "employee@seahavenind.com",
+        "email_verified": True,
+        "name": "Test Employee",
+    }
+    claims.update(overrides)
+    return claims
+
+
+def _token(private_key, **overrides):
+    return jwt.encode(
+        _claims(**overrides),
+        private_key,
+        algorithm="RS256",
+        headers={"kid": "test-key"},
+    )
+
+
+def _parameter(name, decrypt=False):
+    assert decrypt is False
+    return {ISSUER_PARAM: ISSUER, AUDIENCE_PARAM: AUDIENCE}[name]
+
+
+def _verify(token, public_key):
+    jwks = MagicMock()
+    jwks.get_signing_key_from_jwt.return_value = SimpleNamespace(key=public_key)
+    with (
+        patch("shared.cognito.get_parameter", side_effect=_parameter),
+        patch("shared.cognito._jwk_client", return_value=jwks),
+    ):
+        return cognito.verify_cognito_id_token(token, ISSUER_PARAM, AUDIENCE_PARAM)
+
+
+def test_valid_id_token_returns_trusted_identity(signing_key):
+    private_key, public_key = signing_key
+
+    result = _verify(_token(private_key), public_key)
+
+    assert result == {
+        "name": "Test Employee",
+        "email": "employee@seahavenind.com",
+    }
+
+
+@pytest.mark.parametrize(
+    "overrides",
+    [
+        {"iss": "https://cognito-idp.us-east-1.amazonaws.com/us-east-1_OTHER"},
+        {"aud": "other-client"},
+        {"token_use": "access"},
+        {"exp": int(time.time()) - 60},
+        {"email": "employee@example.com"},
+        {"email_verified": False},
+        {"name": ""},
+    ],
+)
+def test_invalid_claims_are_rejected(signing_key, overrides):
+    private_key, public_key = signing_key
+
+    assert _verify(_token(private_key, **overrides), public_key) is None
+
+
+def test_missing_required_claim_is_rejected(signing_key):
+    private_key, public_key = signing_key
+    claims = _claims()
+    del claims["email"]
+    token = jwt.encode(
+        claims, private_key, algorithm="RS256", headers={"kid": "test-key"}
+    )
+
+    assert _verify(token, public_key) is None
+
+
+def test_ssm_failure_is_rejected(signing_key):
+    private_key, _ = signing_key
+    with patch(
+        "shared.cognito.get_parameter", side_effect=RuntimeError("SSM unavailable")
+    ):
+        with pytest.raises(cognito.CognitoVerificationUnavailable):
+            cognito.verify_cognito_id_token(
+                _token(private_key), ISSUER_PARAM, AUDIENCE_PARAM
+            )
+
+
+def test_jwks_failure_is_rejected(signing_key):
+    private_key, _ = signing_key
+    jwks = MagicMock()
+    jwks.get_signing_key_from_jwt.side_effect = OSError("JWKS unavailable")
+    with (
+        patch("shared.cognito.get_parameter", side_effect=_parameter),
+        patch("shared.cognito._jwk_client", return_value=jwks),
+    ):
+        with pytest.raises(cognito.CognitoVerificationUnavailable):
+            cognito.verify_cognito_id_token(
+                _token(private_key), ISSUER_PARAM, AUDIENCE_PARAM
+            )
+
+
+def test_cognito_token_detection_is_untrusted_routing_hint(signing_key):
+    private_key, _ = signing_key
+
+    assert cognito.looks_like_cognito_token(_token(private_key)) is True
+    assert cognito.looks_like_cognito_token("not-a-jwt") is False
+    assert (
+        cognito.looks_like_cognito_token(
+            _token(private_key, iss="https://accounts.google.com")
+        )
+        is False
+    )

tests/test_submit_order.py

@@ -610,6 +610,116 @@ def test_total_summation_multiple_items(
 # ===========================================================================
 
 
+@patch("submit_order_handler._lambda")
+@patch("submit_order_handler.put_order")
+@patch("submit_order_handler.get_form_status", return_value="open")
+@patch("submit_order_handler.current_week", return_value="2026-W20")
+@patch(
+    "submit_order_handler.get_settings",
+    return_value={"bulk_discount_percent": 0, "company_subsidy_percent": 0},
+)
+@patch("submit_order_handler._get_google_client_id")
+@patch(
+    "submit_order_handler._verify_portal_token",
+    return_value={
+        "name": "Portal Employee",
+        "email": "portal.employee@seahavenind.com",
+    },
+)
+@patch("submit_order_handler.get_menu")
+def test_portal_token_identity_takes_precedence(
+    mock_get_menu,
+    mock_portal,
+    mock_google_client_id,
+    mock_settings,
+    mock_week,
+    mock_status,
+    mock_put,
+    mock_lambda,
+):
+    items = _make_items([(10.00, 1)])
+    mock_get_menu.return_value = _menu_doc_from_retail_pairs([(10.00, 1)])
+    event = _submit_event(
+        items,
+        employee_name="Body Name",
+        employee_email="body@example.com",
+        extra_body={"google_id_token": "body-google-token"},
+    )
+    event["headers"] = {"authorization": "Bearer portal-id-token"}
+
+    status, body = _parse_response(submit_order_handler.lambda_handler(event, None))
+
+    assert status == 200, body
+    saved_order = mock_put.call_args.args[2]
+    assert saved_order["employee_name"] == "Portal Employee"
+    assert saved_order["employee_email"] == "portal.employee@seahavenind.com"
+    mock_portal.assert_called_once_with("portal-id-token")
+    mock_google_client_id.assert_not_called()
+
+
+@patch("submit_order_handler.put_order")
+@patch("submit_order_handler._get_google_client_id")
+@patch("submit_order_handler._verify_portal_token", return_value=None)
+def test_invalid_portal_token_never_falls_back_to_google_or_body(
+    mock_portal, mock_google_client_id, mock_put
+):
+    event = _submit_event(
+        _make_items([(10.00, 1)]),
+        employee_name="Body Name",
+        employee_email="body@seahavenind.com",
+        extra_body={"google_id_token": "valid-google-token"},
+    )
+    event["headers"] = {"Authorization": "Bearer invalid-portal-token"}
+
+    status, body = _parse_response(submit_order_handler.lambda_handler(event, None))
+
+    assert status == 403
+    assert body == {"error": "Invalid or unauthorized portal account"}
+    mock_portal.assert_called_once_with("invalid-portal-token")
+    mock_google_client_id.assert_not_called()
+    mock_put.assert_not_called()
+
+
+@patch("submit_order_handler.put_order")
+@patch(
+    "submit_order_handler._verify_portal_token",
+    side_effect=submit_order_handler.CognitoVerificationUnavailable,
+)
+def test_portal_verification_outage_returns_503(mock_portal, mock_put):
+    event = _submit_event(_make_items([(10.00, 1)]))
+    event["headers"] = {"authorization": "Bearer portal-id-token"}
+
+    status, body = _parse_response(submit_order_handler.lambda_handler(event, None))
+
+    assert status == 503
+    assert body == {"error": "Authentication service temporarily unavailable"}
+    mock_put.assert_not_called()
+
+
+@patch(
+    "submit_order_handler.get_settings",
+    return_value={"admin_emails": ["portal.admin@seahaven.com"]},
+)
+@patch(
+    "submit_order_handler._verify_portal_token",
+    return_value={"name": "Portal Admin", "email": "portal.admin@seahaven.com"},
+)
+@patch("submit_order_handler.looks_like_cognito_token", return_value=True)
+def test_in_handler_admin_check_accepts_portal_token(
+    mock_looks_like, mock_portal, mock_settings
+):
+    event = _make_event(
+        method="GET",
+        path="/api/admin/orders",
+        headers={"authorization": "Bearer portal-id-token"},
+    )
+
+    user, error = submit_order_handler._verify_admin(event)
+
+    assert error is None
+    assert user == {"name": "Portal Admin", "email": "portal.admin@seahaven.com"}
+
+
 @patch("submit_order_handler._lambda")
 @patch("submit_order_handler.put_order")
 @patch("submit_order_handler.get_form_status", return_value="open")

tests/test_terraform_cognito_auth.py

@@ -0,0 +1,47 @@
+"""Structural checks for portal Cognito ID-token configuration."""
+
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+TERRAFORM = ROOT / "terraform"
+
+
+def _read(name: str) -> str:
+    return (TERRAFORM / name).read_text()
+
+
+def test_portal_cognito_parameters_are_managed():
+    locals_tf = _read("locals.tf")
+    ssm_tf = _read("ssm.tf")
+    variables_tf = _read("variables.tf")
+
+    assert (
+        'portal_cognito_issuer_param   = "${local.ssm_prefix}/portal-cognito-issuer"'
+        in locals_tf
+    )
+    assert (
+        'portal_cognito_audience_param = "${local.ssm_prefix}/portal-cognito-audience"'
+        in locals_tf
+    )
+    assert 'resource "aws_ssm_parameter" "portal_cognito_issuer"' in ssm_tf
+    assert 'resource "aws_ssm_parameter" "portal_cognito_audience"' in ssm_tf
+    assert 'variable "portal_cognito_issuer"' in variables_tf
+    assert 'variable "portal_cognito_audience"' in variables_tf
+
+
+def test_both_api_lambdas_receive_parameter_names():
+    lambda_tf = _read("lambda.tf")
+
+    assert lambda_tf.count("PORTAL_COGNITO_ISSUER_PARAM") == 2
+    assert lambda_tf.count("PORTAL_COGNITO_AUDIENCE_PARAM") == 2
+    assert lambda_tf.count("aws_ssm_parameter.portal_cognito_issuer.name") == 2
+    assert lambda_tf.count("aws_ssm_parameter.portal_cognito_audience.name") == 2
+
+
+def test_submit_route_remains_public_for_google_compatibility():
+    locals_tf = _read("locals.tf")
+    submit_route = locals_tf.split("submit_order = {", 1)[1].split("}", 1)[0]
+
+    assert 'route_key         = "POST /api/submit-order"' in submit_route
+    assert 'authorizer        = "NONE"' in submit_route
+    assert 'authorizer_type                   = "JWT"' not in _read("apigateway.tf")