Gate web UIs behind auth and escape currency XSS

The po-web-ui and workorder-web-ui handlers had no auth: any
invocation path returned the full PO/WO DB. Add a fail-closed
shared-secret gate (X-Auth-Token / Bearer, constant-time compared to
WEB_UI_AUTH_TOKEN) so a future re-attached Function URL cannot
re-expose the data (URLs removed under INFRA-74). Wire the token from
the SSM String param /procurement-ingest/web-ui-auth-token.

Also fix stored XSS in po-web-ui fmt_currency: the non-numeric
fallback returned str(val) unescaped, so a prompt-injected email
could make Claude emit total_amount as <script>. Escape it.

Refs: INFRA-74
This commit is contained in:
Adam Moussa 2026-06-17 11:37:49 -04:00
parent 5785f39590
commit e97e740c5a
4 changed files with 113 additions and 1 deletions

View file

@ -191,6 +191,19 @@ class PoIngestStack(Stack):
log_retention=logs.RetentionDays.TWO_MONTHS,
environment={
"PO_TABLE": "purchase-orders",
# Defense-in-depth shared secret for the web UI handler. The
# handler fails closed if this is unset, so any future invocation
# path cannot re-expose the PO DB unauthenticated. Sourced from the
# SSM String parameter /procurement-ingest/web-ui-auth-token; create
# it before deploy. NOTE: a plaintext String (not SecureString) is
# used deliberately — value_for_string_parameter resolves String
# params into Lambda env vars at deploy via {{resolve:ssm:...}};
# SecureString dynamic references are NOT resolved in Lambda env
# vars and would pass the token literally. This is a defense-in-depth
# floor for a currently-detached URL (INFRA-74), not a primary auth.
"WEB_UI_AUTH_TOKEN": ssm.StringParameter.value_for_string_parameter(
self, "/procurement-ingest/web-ui-auth-token"
),
},
)

View file

@ -201,6 +201,19 @@ class WorkorderIngestStack(Stack):
environment={
"WORK_ORDERS_TABLE": work_orders_table.table_name,
"COMMENTS_TABLE": comments_table.table_name,
# Defense-in-depth shared secret for the web UI handler. The
# handler fails closed if this is unset, so any future invocation
# path cannot re-expose the WO DB unauthenticated. Sourced from the
# SSM String parameter /procurement-ingest/web-ui-auth-token; create
# it before deploy. NOTE: a plaintext String (not SecureString) is
# used deliberately — value_for_string_parameter resolves String
# params into Lambda env vars at deploy via {{resolve:ssm:...}};
# SecureString dynamic references are NOT resolved in Lambda env
# vars and would pass the token literally. This is a defense-in-depth
# floor for a currently-detached URL (INFRA-74), not a primary auth.
"WEB_UI_AUTH_TOKEN": ssm.StringParameter.value_for_string_parameter(
self, "/procurement-ingest/web-ui-auth-token"
),
},
)

View file

@ -5,6 +5,7 @@ Serves a simple HTML dashboard for viewing purchase orders.
Accessed via Lambda Function URL.
"""
import hmac
import json
import os
from decimal import Decimal
@ -16,6 +17,40 @@ dynamodb = boto3.resource("dynamodb")
PO_TABLE = os.environ.get("PO_TABLE", "purchase-orders")
# Defense-in-depth auth gate. The public Function URL was removed (INFRA-74), but
# the handler must still refuse unauthenticated requests so any future invocation
# path (re-attached Function URL, API Gateway, etc.) does not re-expose the whole
# PO DB. Callers must present the shared secret in the X-Auth-Token header (or
# Authorization: Bearer <token>). The secret is supplied via the WEB_UI_AUTH_TOKEN
# env var (wire to Secrets Manager / SSM at deploy time). If it is unset, the
# handler fails closed and denies every request.
WEB_UI_AUTH_TOKEN = os.environ.get("WEB_UI_AUTH_TOKEN")
def _header(event: dict, name: str) -> str:
"""Case-insensitive header lookup from a Lambda Function URL / APIGW event."""
headers = event.get("headers") or {}
name_lower = name.lower()
for key, value in headers.items():
if key.lower() == name_lower:
return value or ""
return ""
def is_authenticated(event: dict) -> bool:
"""Constant-time check of the request's shared secret against the configured
token. Fails closed when no token is configured."""
if not WEB_UI_AUTH_TOKEN:
return False
presented = _header(event, "x-auth-token")
if not presented:
auth = _header(event, "authorization")
if auth.lower().startswith("bearer "):
presented = auth[7:].strip()
if not presented:
return False
return hmac.compare_digest(presented, WEB_UI_AUTH_TOKEN)
def get_purchase_orders(limit=500):
table = dynamodb.Table(PO_TABLE)
@ -60,7 +95,10 @@ def fmt_currency(val):
val = float(val)
if isinstance(val, (int, float)):
return f"${val:,.2f}"
return str(val)
# Non-numeric fallback: a prompt-injected email can make Claude return
# total_amount/amount as an arbitrary string. Escape it before it is
# interpolated raw into the HTML to prevent stored XSS.
return esc(str(val))
def render_po_detail(po):
@ -250,6 +288,13 @@ def render_po_list(purchase_orders):
def handler(event, context):
if not is_authenticated(event):
return {
"statusCode": 401,
"headers": {"Content-Type": "text/html"},
"body": "<h1>401 Unauthorized</h1>",
}
path = event.get("rawPath", "/")
qs = event.get("queryStringParameters") or {}

View file

@ -5,6 +5,7 @@ Serves a simple HTML dashboard for viewing work orders and comments.
Accessed via Lambda Function URL.
"""
import hmac
import json
import os
from html import escape as esc
@ -16,6 +17,39 @@ dynamodb = boto3.resource("dynamodb")
WORK_ORDERS_TABLE = os.environ.get("WORK_ORDERS_TABLE", "WorkOrders")
COMMENTS_TABLE = os.environ.get("COMMENTS_TABLE", "WorkOrderComments")
# Defense-in-depth auth gate. The public Function URL was removed (INFRA-74), but
# the handler must still refuse unauthenticated requests so any future invocation
# path does not re-expose the whole WO DB. Callers must present the shared secret
# in the X-Auth-Token header (or Authorization: Bearer <token>). The secret is
# supplied via the WEB_UI_AUTH_TOKEN env var (wire to Secrets Manager / SSM at
# deploy time). If it is unset, the handler fails closed and denies every request.
WEB_UI_AUTH_TOKEN = os.environ.get("WEB_UI_AUTH_TOKEN")
def _header(event: dict, name: str) -> str:
"""Case-insensitive header lookup from a Lambda Function URL / APIGW event."""
headers = event.get("headers") or {}
name_lower = name.lower()
for key, value in headers.items():
if key.lower() == name_lower:
return value or ""
return ""
def is_authenticated(event: dict) -> bool:
"""Constant-time check of the request's shared secret against the configured
token. Fails closed when no token is configured."""
if not WEB_UI_AUTH_TOKEN:
return False
presented = _header(event, "x-auth-token")
if not presented:
auth = _header(event, "authorization")
if auth.lower().startswith("bearer "):
presented = auth[7:].strip()
if not presented:
return False
return hmac.compare_digest(presented, WEB_UI_AUTH_TOKEN)
def get_work_orders(limit=500):
table = dynamodb.Table(WORK_ORDERS_TABLE)
@ -228,6 +262,13 @@ def render_work_orders_list(work_orders):
def handler(event, context):
if not is_authenticated(event):
return {
"statusCode": 401,
"headers": {"Content-Type": "text/html"},
"body": "<h1>401 Unauthorized</h1>",
}
path = event.get("rawPath", "/")
qs = event.get("queryStringParameters") or {}