229 lines
9.7 KiB
Python
229 lines
9.7 KiB
Python
# tests/test_demo_token.py
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
from services.control_api.auth.errors import InvalidToken
|
|
from services.control_api.auth.models import PrincipalKind
|
|
from services.control_api.persistence.demo_client import (
|
|
DEMO_CLIENT_ID,
|
|
DEMO_CLIENT_ISSUER,
|
|
DEMO_CLIENT_SUBJECT,
|
|
)
|
|
from services.runtime.demo_token import (
|
|
MAX_TTL_HOURS,
|
|
MIN_TTL_HOURS,
|
|
DemoTokenAuthenticator,
|
|
load_demo_revocations,
|
|
mint_demo_token,
|
|
verify_demo_token,
|
|
)
|
|
|
|
KEY = b"x" * 32
|
|
OTHER_KEY = b"y" * 32
|
|
|
|
|
|
class DemoTokenCoreTests(unittest.TestCase):
|
|
def test_mint_and_verify_round_trip(self) -> None:
|
|
token, jti = mint_demo_token(KEY, ttl_hours=48, now=1_000_000)
|
|
claims = verify_demo_token(token, KEY, now=1_000_100)
|
|
self.assertEqual(claims.jti, jti)
|
|
self.assertEqual(claims.issued_at, 1_000_000)
|
|
self.assertEqual(claims.expires_at, 1_000_000 + 48 * 3600)
|
|
|
|
def test_ttl_bounds_enforced(self) -> None:
|
|
with self.assertRaises(ValueError):
|
|
mint_demo_token(KEY, ttl_hours=MIN_TTL_HOURS - 1)
|
|
with self.assertRaises(ValueError):
|
|
mint_demo_token(KEY, ttl_hours=MAX_TTL_HOURS + 1)
|
|
mint_demo_token(KEY, ttl_hours=MIN_TTL_HOURS)
|
|
mint_demo_token(KEY, ttl_hours=MAX_TTL_HOURS)
|
|
|
|
def test_expired_token_rejected(self) -> None:
|
|
token, _ = mint_demo_token(KEY, ttl_hours=1, now=1_000_000)
|
|
with self.assertRaises(InvalidToken):
|
|
verify_demo_token(token, KEY, now=1_000_000 + 3601)
|
|
|
|
def test_wrong_key_rejected(self) -> None:
|
|
token, _ = mint_demo_token(KEY, ttl_hours=1, now=1_000_000)
|
|
with self.assertRaises(InvalidToken):
|
|
verify_demo_token(token, OTHER_KEY, now=1_000_100)
|
|
|
|
def test_tampered_payload_rejected(self) -> None:
|
|
token, _ = mint_demo_token(KEY, ttl_hours=1, now=1_000_000)
|
|
prefix, rest = token.split("demo_", 1)
|
|
payload_b64, signature_b64 = rest.split(".")
|
|
tampered = f"{prefix}demo_{payload_b64}A.{signature_b64}"
|
|
with self.assertRaises(InvalidToken):
|
|
verify_demo_token(tampered, KEY, now=1_000_100)
|
|
|
|
def test_non_demo_token_rejected(self) -> None:
|
|
with self.assertRaises(InvalidToken):
|
|
verify_demo_token("not-a-demo-token", KEY)
|
|
|
|
def test_malformed_shape_rejected(self) -> None:
|
|
with self.assertRaises(InvalidToken):
|
|
verify_demo_token("demo_onlyonepart", KEY)
|
|
with self.assertRaises(InvalidToken):
|
|
verify_demo_token("demo_a.b.c", KEY)
|
|
|
|
|
|
class DemoTokenAuthenticatorTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.temp_dir = tempfile.TemporaryDirectory()
|
|
self.revocations_path = Path(self.temp_dir.name) / "revocations.json"
|
|
|
|
def tearDown(self) -> None:
|
|
self.temp_dir.cleanup()
|
|
|
|
def test_missing_revocations_file_means_nothing_revoked(self) -> None:
|
|
self.assertEqual(load_demo_revocations(self.revocations_path), frozenset())
|
|
|
|
def test_revocations_file_is_read(self) -> None:
|
|
self.revocations_path.write_text(
|
|
json.dumps({"schemaVersion": "1.0", "revokedJtis": ["abc", "def"]}),
|
|
encoding="utf-8",
|
|
)
|
|
self.assertEqual(load_demo_revocations(self.revocations_path), frozenset({"abc", "def"}))
|
|
|
|
def test_revocations_path_that_is_a_directory_fails_closed(self) -> None:
|
|
"""Docker creates a directory at a bind-mount target whose host source
|
|
is missing. That must raise, not read as 'nothing revoked'."""
|
|
self.revocations_path.mkdir()
|
|
with self.assertRaises(ValueError):
|
|
load_demo_revocations(self.revocations_path)
|
|
|
|
def test_invalid_revocations_file_raises(self) -> None:
|
|
self.revocations_path.write_text("not json", encoding="utf-8")
|
|
with self.assertRaises(ValueError):
|
|
load_demo_revocations(self.revocations_path)
|
|
|
|
def test_valid_token_authenticates_as_fixed_demo_principal(self) -> None:
|
|
# DemoTokenAuthenticator has no test-clock hook by design (its
|
|
# __call__ signature is fixed to authorization_header only), so it
|
|
# always verifies against the real wall clock -- mint against real
|
|
# "now" here rather than a fixed past epoch that would always read
|
|
# as already-expired.
|
|
token, _ = mint_demo_token(KEY, ttl_hours=1)
|
|
authenticator = DemoTokenAuthenticator(KEY, self.revocations_path)
|
|
principal = authenticator(f"Bearer {token}")
|
|
self.assertEqual(principal.client_id, DEMO_CLIENT_ID)
|
|
self.assertEqual(principal.issuer, DEMO_CLIENT_ISSUER)
|
|
self.assertEqual(principal.subject, DEMO_CLIENT_SUBJECT)
|
|
self.assertEqual(principal.kind, PrincipalKind.MACHINE_CLIENT)
|
|
|
|
def test_key_loader_supports_secret_rotation_without_router_restart(self) -> None:
|
|
first = b"a" * 32
|
|
second = b"b" * 32
|
|
current = [first]
|
|
authenticator = DemoTokenAuthenticator(
|
|
None, self.revocations_path, key_loader=lambda: current[0]
|
|
)
|
|
first_token, _ = mint_demo_token(first, ttl_hours=1)
|
|
authenticator(f"Bearer {first_token}")
|
|
current[0] = second
|
|
second_token, _ = mint_demo_token(second, ttl_hours=1)
|
|
authenticator(f"Bearer {second_token}")
|
|
with self.assertRaises(InvalidToken):
|
|
authenticator(f"Bearer {first_token}")
|
|
|
|
def test_authenticator_requires_exactly_one_key_source(self) -> None:
|
|
with self.assertRaises(ValueError):
|
|
DemoTokenAuthenticator(None, self.revocations_path)
|
|
with self.assertRaises(ValueError):
|
|
DemoTokenAuthenticator(KEY, self.revocations_path, key_loader=lambda: KEY)
|
|
|
|
def test_revoked_jti_rejected(self) -> None:
|
|
# Real "now" so the InvalidToken raised below is actually caused by
|
|
# the revocation check, not masked by an unrelated expiry failure.
|
|
token, jti = mint_demo_token(KEY, ttl_hours=1)
|
|
self.revocations_path.write_text(
|
|
json.dumps({"schemaVersion": "1.0", "revokedJtis": [jti]}), encoding="utf-8"
|
|
)
|
|
authenticator = DemoTokenAuthenticator(KEY, self.revocations_path)
|
|
with self.assertRaises(InvalidToken):
|
|
authenticator(f"Bearer {token}")
|
|
|
|
def test_missing_or_malformed_header_rejected(self) -> None:
|
|
authenticator = DemoTokenAuthenticator(KEY, self.revocations_path)
|
|
with self.assertRaises(InvalidToken):
|
|
authenticator(None)
|
|
with self.assertRaises(InvalidToken):
|
|
authenticator("not-bearer-shaped")
|
|
|
|
def test_tampered_payload_can_never_resolve_to_a_different_client(self) -> None:
|
|
"""Defense in depth: a hand-crafted payload that injects an extra
|
|
"project" field still resolves to the one fixed demo client, because
|
|
the authenticator never reads project/client identity from the
|
|
token payload at all -- only jti/iat/exp are ever consulted."""
|
|
import hashlib
|
|
import hmac as hmac_module
|
|
import time
|
|
|
|
from services.runtime.demo_token import _b64url_encode
|
|
|
|
now = int(time.time())
|
|
forged_payload = json.dumps(
|
|
{
|
|
"jti": "forged-jti",
|
|
"iat": now,
|
|
"exp": now + 3600,
|
|
"project": "some-other-project",
|
|
"client_id": str(uuid4()),
|
|
},
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
).encode("utf-8")
|
|
payload_b64 = _b64url_encode(forged_payload)
|
|
signature = hmac_module.new(KEY, payload_b64.encode("ascii"), hashlib.sha256).digest()
|
|
forged_token = f"demo_{payload_b64}.{_b64url_encode(signature)}"
|
|
|
|
authenticator = DemoTokenAuthenticator(KEY, self.revocations_path)
|
|
principal = authenticator(f"Bearer {forged_token}")
|
|
self.assertEqual(principal.client_id, DEMO_CLIENT_ID)
|
|
self.assertEqual(principal.issuer, DEMO_CLIENT_ISSUER)
|
|
self.assertEqual(principal.subject, DEMO_CLIENT_SUBJECT)
|
|
|
|
def test_expired_demo_principal_is_rejected_by_the_existing_authorizer(self) -> None:
|
|
"""Defense in depth at the OTHER layer: even if a demo principal's
|
|
own expiry check were ever bypassed, the existing, unmodified
|
|
AliasAuthorizer.authorize() still independently rejects it, because
|
|
DemoTokenAuthenticator always sets expires_at from the token's real
|
|
exp claim, never a far-future placeholder."""
|
|
from datetime import timedelta
|
|
|
|
from pydantic import AnyHttpUrl, TypeAdapter
|
|
|
|
from services.dataplane_router.authz import AliasAuthorizer
|
|
from services.dataplane_router.errors import RouterError
|
|
from services.dataplane_router.models import AliasRecord, ClientBinding
|
|
|
|
token, _ = mint_demo_token(KEY, ttl_hours=1)
|
|
authenticator = DemoTokenAuthenticator(KEY, self.revocations_path)
|
|
principal = authenticator(f"Bearer {token}")
|
|
record = AliasRecord(
|
|
project_slug="mobilityops",
|
|
environment="prod",
|
|
alias="/mcp/mobilityops/prod",
|
|
upstream_url=TypeAdapter(AnyHttpUrl).validate_python("https://gateway.invalid/x"),
|
|
enabled=True,
|
|
bindings=(
|
|
ClientBinding(
|
|
client_id=DEMO_CLIENT_ID,
|
|
issuer=DEMO_CLIENT_ISSUER,
|
|
subject=DEMO_CLIENT_SUBJECT,
|
|
enabled=True,
|
|
),
|
|
),
|
|
)
|
|
# principal.expires_at reflects the token's real exp claim; evaluate
|
|
# "now" as later than that to prove the authorizer's own check fires.
|
|
with self.assertRaises(RouterError):
|
|
AliasAuthorizer().authorize(
|
|
principal, record, now=principal.expires_at + timedelta(seconds=100)
|
|
)
|