"""
Web UI Lambda.
Serves a simple HTML dashboard for viewing purchase orders.
Accessed via Lambda Function URL.
"""
import hmac
import json
import os
from decimal import Decimal
from html import escape as esc
import boto3
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 ). 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)
items = []
response = table.scan()
items.extend(response.get("Items", []))
while "LastEvaluatedKey" in response:
response = table.scan(ExclusiveStartKey=response["LastEvaluatedKey"])
items.extend(response.get("Items", []))
items.sort(key=lambda x: x.get("processed_at", ""), reverse=True)
return items[:limit]
def render_badge(value, color_map):
if not value:
value = "unknown"
color = color_map.get(value.lower(), "#9ca3af")
label = esc(value.replace("_", " ").title())
return f'{label}'
STATUS_COLORS = {
"issued": "#3b82f6",
"pending buyer action": "#f59e0b",
"open": "#3b82f6",
"closed": "#10b981",
"cancelled": "#ef4444",
"soft closed": "#6b7280",
}
EMAIL_TYPE_COLORS = {
"new_po": "#3b82f6",
"revision": "#f59e0b",
"cancellation": "#ef4444",
}
def fmt_currency(val):
if val is None:
return ""
if isinstance(val, Decimal):
val = float(val)
if isinstance(val, (int, float)):
return f"${val:,.2f}"
# 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):
po_number = esc(po.get("po_number", ""))
fields = [
("PO Number", po_number),
("Status", render_badge(po.get("po_status", ""), STATUS_COLORS)),
("Email Type", render_badge(po.get("email_type", ""), EMAIL_TYPE_COLORS)),
("Total Amount", fmt_currency(po.get("total_amount"))),
("Currency", esc(po.get("currency", "")) or None),
("Supplier", esc((po.get("supplier") or {}).get("name") or "") or None),
("Site Code", esc(po.get("site_code", "")) or None),
("State", esc(po.get("state", "")) or None),
("Trade", esc(po.get("trade", "")) or None),
("Fiscal Year", esc(po.get("fiscal_year", "")) or None),
("Coupa Category", esc(po.get("coupa_category", "")) or None),
("Submitted By", esc(po.get("submitted_by", "")) or None),
("On Behalf Of", esc(po.get("on_behalf_of", "")) or None),
("Order Date", esc(po.get("order_date", "")) or None),
("Revision Date", esc(po.get("revision_date", "")) or None),
("Payment Terms", esc(po.get("payment_terms", "")) or None),
("Requisition #", esc(po.get("requisition_number", "")) or None),
("Department", esc(po.get("department", "")) or None),
("Data Source", esc(po.get("data_source", "")) or None),
("Processed At", esc(po.get("processed_at", "")) or None),
]
ship_to = po.get("ship_to") or {}
if any(ship_to.values()):
ship_parts = []
if ship_to.get("name"):
ship_parts.append(esc(ship_to["name"]))
if ship_to.get("address"):
ship_parts.append(esc(ship_to["address"]))
if ship_to.get("location_code"):
ship_parts.append(f"Location: {esc(ship_to['location_code'])}")
if ship_to.get("attn"):
ship_parts.append(f"Attn: {esc(ship_to['attn'])}")
fields.append(("Ship To", " ".join(ship_parts)))
view_url = po.get("view_order_url")
if view_url and view_url.startswith(("https://", "http://")):
escaped_url = esc(view_url, quote=True)
fields.append(
(
"Coupa Link",
f'View in Coupa',
)
)
details_html = ""
for label, value in fields:
if value:
details_html += f"""
{label}
{value}
"""
# Line items
line_items = po.get("line_items") or []
items_html = ""
if line_items:
rows = ""
for item in line_items:
qty = esc(str(item.get("quantity", "") or ""))
unit = esc(str(item.get("unit", "") or ""))
price = esc(str(item.get("price", "") or ""))
rows += f"""