adam/procurement-ingest
1
0
Fork 0
mirror of https://github.com/Sea-Haven-Industries/procurement-ingest.git synced 2026-10-07 16:08:54 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Add TTL to web UI auth token cache for rotation

Browse source 
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
This commit is contained in:
Adam Moussa 2026-07-15 19:33:16 -04:00
parent 5b39f1dad1
commit 2802d04368
No known key found for this signature in database
2 changed files with 32 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
lambdas/po/web_ui/handler.py
View file

@ -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

20
lambdas/wo/web_ui/handler.py
View file

@ -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