From 2802d043687767fdc95d09fc284548427b536bf4 Mon Sep 17 00:00:00 2001 From: Adam Moussa Date: Wed, 15 Jul 2026 19:33:16 -0400 Subject: [PATCH] Add TTL to web UI auth token cache for rotation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The web-ui handlers cached the Secrets Manager auth token at module level with no expiry, so a rotated secret was only picked up when the warm container recycled — an emergency rotation could take hours to take effect. Cache the fetched value for a 5-minute TTL instead, so a rotated token propagates within the TTL while still avoiding a Secrets Manager call on every request. Still fails closed when the secret is unset or unreadable. Refs: INFRA-74 --- lambdas/po/web_ui/handler.py | 20 ++++++++++++++++---- lambdas/wo/web_ui/handler.py | 20 ++++++++++++++++---- 2 files changed, 32 insertions(+), 8 deletions(-) diff --git a/lambdas/po/web_ui/handler.py b/lambdas/po/web_ui/handler.py index 714c79a..9e8d48f 100644 --- a/lambdas/po/web_ui/handler.py +++ b/lambdas/po/web_ui/handler.py @@ -8,6 +8,7 @@ Accessed via Lambda Function URL. import hmac import json import os +import time from decimal import Decimal from html import escape as esc @@ -26,17 +27,27 @@ PO_TABLE = os.environ.get("PO_TABLE", "purchase-orders") # 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") +# Refresh the cached token this often so a rotated secret propagates without +# waiting for the execution environment to recycle (emergency-rotation path). +_AUTH_TOKEN_CACHE_TTL_SECONDS = 300 _auth_token_cache = None +_auth_token_cached_at = 0.0 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. + Cached in the warm container for a short TTL so we don't hit Secrets Manager + on every request, while still picking up a rotated secret within the TTL + rather than only when the execution environment recycles. Returns None when + not configured or unreadable (the caller then fails closed). """ - global _auth_token_cache - if _auth_token_cache is not None: + global _auth_token_cache, _auth_token_cached_at + now = time.monotonic() + if ( + _auth_token_cache is not None + and now - _auth_token_cached_at < _AUTH_TOKEN_CACHE_TTL_SECONDS + ): return _auth_token_cache if not _WEB_UI_AUTH_TOKEN_SECRET_ARN: return None @@ -44,6 +55,7 @@ def _get_auth_token() -> str | None: try: secret = secrets.get_secret_value(SecretId=_WEB_UI_AUTH_TOKEN_SECRET_ARN) _auth_token_cache = secret["SecretString"] + _auth_token_cached_at = now return _auth_token_cache except Exception: return None diff --git a/lambdas/wo/web_ui/handler.py b/lambdas/wo/web_ui/handler.py index 8bc603a..21f64aa 100644 --- a/lambdas/wo/web_ui/handler.py +++ b/lambdas/wo/web_ui/handler.py @@ -8,6 +8,7 @@ Accessed via Lambda Function URL. import hmac import json import os +import time from html import escape as esc import boto3 @@ -25,17 +26,27 @@ COMMENTS_TABLE = os.environ.get("COMMENTS_TABLE", "WorkOrderComments") # 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") +# Refresh the cached token this often so a rotated secret propagates without +# waiting for the execution environment to recycle (emergency-rotation path). +_AUTH_TOKEN_CACHE_TTL_SECONDS = 300 _auth_token_cache = None +_auth_token_cached_at = 0.0 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. + Cached in the warm container for a short TTL so we don't hit Secrets Manager + on every request, while still picking up a rotated secret within the TTL + rather than only when the execution environment recycles. Returns None when + not configured or unreadable (the caller then fails closed). """ - global _auth_token_cache - if _auth_token_cache is not None: + global _auth_token_cache, _auth_token_cached_at + now = time.monotonic() + if ( + _auth_token_cache is not None + and now - _auth_token_cached_at < _AUTH_TOKEN_CACHE_TTL_SECONDS + ): return _auth_token_cache if not _WEB_UI_AUTH_TOKEN_SECRET_ARN: return None @@ -43,6 +54,7 @@ def _get_auth_token() -> str | None: try: secret = secrets.get_secret_value(SecretId=_WEB_UI_AUTH_TOKEN_SECRET_ARN) _auth_token_cache = secret["SecretString"] + _auth_token_cached_at = now return _auth_token_cache except Exception: return None