Send a projected flex-auth caller token on every check
All checks were successful
Build and Publish Container Image / build-and-push (push) Successful in 1m7s

TENANT_ENGINE_FLEX_AUTH_TOKEN_FILE is read per request so hourly
projection rotation needs no restart. Missing or unreadable file fails
closed as a local deny and never calls flex-auth. Needed before
flex-auth-tenant-engine can enforce (FLEX-WP-0015-T02).
This commit is contained in:
tegwick 2026-08-19 14:31:39 +02:00
parent 0809af063c
commit 2063470ac8
4 changed files with 38 additions and 2 deletions

View file

@ -62,9 +62,11 @@ class FlexAuthCheckClient:
base_url: str,
timeout_seconds: float = 3.0,
transport: httpx.BaseTransport | None = None,
bearer_token_file: str | None = None,
) -> None:
self.base_url = base_url.rstrip("/")
self.timeout_seconds = timeout_seconds
self.bearer_token_file = bearer_token_file
self._client = httpx.Client(
base_url=self.base_url,
timeout=httpx.Timeout(timeout_seconds),
@ -73,8 +75,17 @@ class FlexAuthCheckClient:
def is_allowed(self, request: CheckRequest) -> bool:
try:
response = self._client.post("/v1/check", json=request.to_json())
except httpx.HTTPError:
headers: dict[str, str] = {}
if self.bearer_token_file:
# Projected ServiceAccount tokens rotate. Read on each check
# instead of pinning the token for the lifetime of the process.
with open(self.bearer_token_file, encoding="utf-8") as token_file:
token = token_file.read().strip()
if not token:
return False
headers["Authorization"] = f"Bearer {token}"
response = self._client.post("/v1/check", json=request.to_json(), headers=headers)
except (httpx.HTTPError, OSError):
return False
if response.status_code != 200: