Authenticate flex-auth calls with a rotating caller token
Closes the caller side of FLEX-WP-0015. FlexAuthHTTPAdapter reads the audience-scoped projected ServiceAccount token from a file on every decision, so hourly rotation needs no restart, and runtime configuration now requires USER_ENGINE_FLEX_AUTH_TOKEN_FILE. A missing, empty, or unreadable token file fails closed as a denial without reaching flex-auth: OSError joins the caught set and an empty read raises. Coverage proves all three unusable-token cases deny before any request is made, and that neither the deny reason nor the decision repr carries the token value. Tenant-authority reads now identify user-engine as actor `user-engine` under the protected tenant.read action, keeping tenant ids opaque and URL-encoded. Contract: docs/flex-auth-caller-identity.md. Full suite: 148 tests, 3 provider-gated skips. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
parent
6f6bbf5e4a
commit
4622b64061
7 changed files with 174 additions and 8 deletions
|
|
@ -1,5 +1,6 @@
|
|||
import io
|
||||
import json
|
||||
import tempfile
|
||||
import unittest
|
||||
from datetime import UTC, datetime
|
||||
from unittest.mock import patch
|
||||
|
|
@ -43,6 +44,72 @@ class PlatformAdapterTests(unittest.TestCase):
|
|||
self.assertEqual(decision.effect, AuthorizationEffect.DENY)
|
||||
self.assertEqual(decision.reason, "authorization service unavailable")
|
||||
|
||||
def test_flex_auth_reads_rotating_caller_token_for_each_decision(self):
|
||||
with tempfile.NamedTemporaryFile(mode="w+", encoding="utf-8") as token_file:
|
||||
token_file.write("projected-token-1\n")
|
||||
token_file.flush()
|
||||
body = _Response(json.dumps({"id": "decision:1", "effect": "allow"}).encode())
|
||||
adapter = FlexAuthHTTPAdapter(
|
||||
base_url="http://flex-auth", bearer_token_file=token_file.name
|
||||
)
|
||||
with patch("user_engine.adapters.flex_auth.urlopen", return_value=body) as call:
|
||||
adapter.check(_request())
|
||||
self.assertEqual(
|
||||
call.call_args.args[0].get_header("Authorization"),
|
||||
"Bearer projected-token-1",
|
||||
)
|
||||
|
||||
token_file.seek(0)
|
||||
token_file.truncate()
|
||||
token_file.write("projected-token-2\n")
|
||||
token_file.flush()
|
||||
body = _Response(json.dumps({"id": "decision:2", "effect": "allow"}).encode())
|
||||
with patch("user_engine.adapters.flex_auth.urlopen", return_value=body) as call:
|
||||
adapter.check(_request())
|
||||
self.assertEqual(
|
||||
call.call_args.args[0].get_header("Authorization"),
|
||||
"Bearer projected-token-2",
|
||||
)
|
||||
|
||||
def test_flex_auth_fails_closed_on_unusable_caller_token(self):
|
||||
"""A caller that cannot prove its identity must never reach the service."""
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
empty = f"{directory}/empty-token"
|
||||
with open(empty, "w", encoding="utf-8") as handle:
|
||||
handle.write(" \n")
|
||||
unusable = {
|
||||
"missing": f"{directory}/absent-token",
|
||||
"empty": empty,
|
||||
"unreadable": directory,
|
||||
}
|
||||
for label, path in unusable.items():
|
||||
with self.subTest(token=label):
|
||||
adapter = FlexAuthHTTPAdapter(
|
||||
base_url="http://flex-auth", bearer_token_file=path
|
||||
)
|
||||
with patch("user_engine.adapters.flex_auth.urlopen") as call:
|
||||
decision = adapter.check(_request())
|
||||
call.assert_not_called()
|
||||
self.assertEqual(decision.effect, AuthorizationEffect.DENY)
|
||||
self.assertEqual(
|
||||
decision.reason, "authorization service unavailable"
|
||||
)
|
||||
|
||||
def test_flex_auth_deny_reason_never_carries_the_caller_token(self):
|
||||
with tempfile.NamedTemporaryFile(mode="w+", encoding="utf-8") as token_file:
|
||||
token_file.write("super-secret-projected-token\n")
|
||||
token_file.flush()
|
||||
adapter = FlexAuthHTTPAdapter(
|
||||
base_url="http://flex-auth", bearer_token_file=token_file.name
|
||||
)
|
||||
with patch(
|
||||
"user_engine.adapters.flex_auth.urlopen", side_effect=URLError("down")
|
||||
):
|
||||
decision = adapter.check(_request())
|
||||
self.assertEqual(decision.effect, AuthorizationEffect.DENY)
|
||||
self.assertNotIn("super-secret-projected-token", str(decision.reason))
|
||||
self.assertNotIn("super-secret-projected-token", repr(decision))
|
||||
|
||||
def test_invitation_delivery_calls_mail_and_event_with_idempotency(self):
|
||||
adapter = HTTPOutboxDeliveryAdapter(
|
||||
event_url="http://events", mail_url="http://mail",
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue