). The secret is
# fetched at runtime from Secrets Manager to keep it out of CloudFormation
# templates and Lambda env vars. If the ARN is unset or the secret is missing, the
# handler fails closed and denies every request.
_WEB_UI_AUTH_TOKEN_SECRET_ARN = os.environ.get("WEB_UI_AUTH_TOKEN_SECRET_ARN")
_auth_token_cache = None
def _get_auth_token() -> str | None:
"""Fetch the shared web UI auth token from Secrets Manager.
Cached at module level across warm invocations so we don't hit Secrets
Manager on every request. Returns None when not configured.
"""
global _auth_token_cache
if _auth_token_cache is not None:
return _auth_token_cache
if not _WEB_UI_AUTH_TOKEN_SECRET_ARN:
return None
secrets = boto3.client("secretsmanager")
try:
secret = secrets.get_secret_value(SecretId=_WEB_UI_AUTH_TOKEN_SECRET_ARN)
_auth_token_cache = secret["SecretString"]
return _auth_token_cache
except Exception:
return None
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."""
token = _get_auth_token()
if not 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, token)
def get_work_orders(limit=500):
table = dynamodb.Table(WORK_ORDERS_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("updated_at", ""), reverse=True)
return items[:limit]
def get_comments(work_order_id):
table = dynamodb.Table(COMMENTS_TABLE)
items = []
kwargs = {
"KeyConditionExpression": "work_order_id = :woid",
"ExpressionAttributeValues": {":woid": work_order_id},
}
response = table.query(**kwargs)
items.extend(response.get("Items", []))
while "LastEvaluatedKey" in response:
response = table.query(**kwargs, ExclusiveStartKey=response["LastEvaluatedKey"])
items.extend(response.get("Items", []))
items.sort(key=lambda x: x.get("created_at", ""), reverse=True)
return items
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 = {
"new": "#3b82f6",
"assigned": "#8b5cf6",
"in_progress": "#f59e0b",
"on_hold": "#6b7280",
"completed": "#10b981",
"cancelled": "#ef4444",
"unknown": "#9ca3af",
}
RECORD_TYPE_COLORS = {
"new_work_order": "#3b82f6",
"comment": "#8b5cf6",
"update": "#f59e0b",
"cancellation": "#ef4444",
"unknown": "#9ca3af",
}
def render_work_order_detail(wo, events):
events_html = ""
if events:
for e in events:
record_type = e.get("record_type", "unknown")
commenter = esc(e.get("commenter", ""))
created = esc(e.get("created_at", ""))
text = esc(e.get("text", ""))
badge = render_badge(record_type, RECORD_TYPE_COLORS)
commenter_str = (
f"{commenter} — " if commenter else ""
)
border_colors = {
"new_work_order": "#3b82f6",
"comment": "#8b5cf6",
"update": "#f59e0b",
"cancellation": "#ef4444",
}
border = border_colors.get(record_type, "#94a3b8")
events_html += f"""
{badge} {commenter_str}{created}
{"
" + text + "
" if text else ""}
"""
else:
events_html = 'No events yet.
'
wo_id = esc(wo.get("work_order_id", ""))
fields = [
("Description", esc(wo.get("description", "")) or None),
("Status", render_badge(wo.get("wo_status", "unknown"), STATUS_COLORS)),
(
"Record Type",
render_badge(wo.get("record_type", "unknown"), RECORD_TYPE_COLORS),
),
("Site Code", esc(wo.get("site_code", "")) or None),
("Building", esc(wo.get("building", "")) or None),
("Address", esc(wo.get("address", "")) or None),
("Severity", esc(wo.get("severity", "")) or None),
("Priority", esc(wo.get("priority", "")) or None),
("Due Date", esc(wo.get("due_date", "")) or None),
("Date Reported", esc(wo.get("date_reported", "")) or None),
("Scheduled Start", esc(wo.get("scheduled_start", "")) or None),
("Assigned To", esc(wo.get("assigned_to", "")) or None),
("Created", esc(wo.get("created_at", "")) or None),
("Last Updated", esc(wo.get("updated_at", "")) or None),
]
details_html = ""
for label, value in fields:
if value:
details_html += f"""
"""
return f"""
WO {wo_id} - Sea Haven
Work Order {wo_id}
{details_html}
Events ({len(events)})
{events_html}
"""
def render_work_orders_list(work_orders):
rows = ""
for wo in work_orders:
wo_id = esc(wo.get("work_order_id", ""))
desc = esc(wo.get("description", ""))
site = esc(wo.get("site_code", ""))
status = wo.get("wo_status", "unknown")
record_type = wo.get("record_type", "unknown")
due = esc(wo.get("due_date", ""))
updated = esc((wo.get("updated_at") or "")[:16])
rows += f"""
| {wo_id} |
{desc} |
{site} |
{render_badge(record_type, RECORD_TYPE_COLORS)} |
{render_badge(status, STATUS_COLORS)} |
{due} |
{updated} |
"""
return f"""
Work Orders - Sea Haven
Work Orders
{len(work_orders)} total
| WO # |
Description |
Site |
Last Action |
Status |
Due Date |
Updated |
{rows if rows else '| No work orders yet. |
'}
"""
def handler(event, context):
if not is_authenticated(event):
return {
"statusCode": 401,
"headers": {"Content-Type": "text/html"},
"body": "401 Unauthorized
",
}
path = event.get("rawPath", "/")
qs = event.get("queryStringParameters") or {}
if path == "/wo" and "id" in qs:
wo_id = qs["id"]
# Get work order
table = dynamodb.Table(WORK_ORDERS_TABLE)
result = table.get_item(Key={"work_order_id": wo_id})
wo = result.get("Item")
if not wo:
return {
"statusCode": 404,
"headers": {"Content-Type": "text/html"},
"body": "Work order not found
",
}
events = get_comments(wo_id)
html = render_work_order_detail(wo, events)
else:
work_orders = get_work_orders()
html = render_work_orders_list(work_orders)
return {
"statusCode": 200,
"headers": {"Content-Type": "text/html"},
"body": html,
}