216 lines
9.2 KiB
Python
216 lines
9.2 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from datetime import UTC, datetime, timedelta
|
|
from urllib.parse import parse_qs, urlparse
|
|
from uuid import uuid4
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from services.control_api.auth.bff import BFFService
|
|
from services.control_api.auth.config import OIDCSettings
|
|
from services.control_api.auth.errors import (
|
|
CSRFError,
|
|
InvalidState,
|
|
InvalidToken,
|
|
ProviderUnavailable,
|
|
RevokedMachinePrincipal,
|
|
SessionExpired,
|
|
)
|
|
from services.control_api.auth.fake_provider import FakeOIDCProvider
|
|
from services.control_api.auth.machine import (
|
|
MachineClientBinding,
|
|
MachinePrincipalAuthenticator,
|
|
PrincipalMapper,
|
|
RevocationCache,
|
|
)
|
|
from services.control_api.auth.models import ManagementRole, PrincipalKind
|
|
from services.control_api.auth.provider import OIDCProviderClient
|
|
from services.control_api.auth.router import build_auth_router
|
|
|
|
|
|
class AuthFixture(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.now = datetime.now(UTC)
|
|
self.fake = FakeOIDCProvider()
|
|
self.settings = OIDCSettings(
|
|
issuer=self.fake.issuer,
|
|
audience=self.fake.audience,
|
|
client_id="hub-bff",
|
|
redirect_uri="https://hub.test/auth/callback",
|
|
secure_cookie=True,
|
|
role_group_map={"hub-viewer": ManagementRole.VIEWER},
|
|
)
|
|
self.provider = OIDCProviderClient(self.settings, transport=self.fake.transport())
|
|
|
|
def tearDown(self) -> None:
|
|
self.provider.close()
|
|
|
|
def test_pkce_bff_happy_path_logout_csrf_and_no_token_projection(self) -> None:
|
|
mapper = PrincipalMapper(role_group_map=self.settings.role_group_map)
|
|
bff = BFFService(self.settings, self.provider, principal_mapper=mapper.map_claims)
|
|
first_url = bff.begin_login(now=self.now)
|
|
second_url = bff.begin_login(now=self.now)
|
|
first_query = parse_qs(urlparse(first_url).query)
|
|
self.assertNotEqual(first_query["state"], parse_qs(urlparse(second_url).query)["state"])
|
|
self.assertEqual(first_query["code_challenge_method"], ["S256"])
|
|
self.assertNotIn("access_token", first_url)
|
|
code = self.fake.authorize(first_url)
|
|
session = bff.complete_login(code=code, state=first_query["state"][0], now=self.now)
|
|
self.assertTrue(session.set_cookie.lower().find("httponly") >= 0)
|
|
self.assertIn("secure", session.set_cookie.lower())
|
|
self.assertIn("samesite=strict", session.set_cookie.lower())
|
|
metadata = bff.session_metadata(session.value, now=self.now)
|
|
self.assertEqual(metadata.kind, PrincipalKind.MANAGEMENT_USER)
|
|
self.assertEqual(metadata.roles, frozenset({ManagementRole.VIEWER}))
|
|
self.assertNotIn("access_token", metadata.model_dump_json())
|
|
with self.assertRaises(CSRFError):
|
|
bff.logout(session.value, "wrong-csrf")
|
|
bff.logout(session.value, session.csrf_token)
|
|
with self.assertRaises(SessionExpired):
|
|
bff.require_session(session.value, now=self.now)
|
|
|
|
def test_state_replay_and_expiry_are_rejected(self) -> None:
|
|
bff = BFFService(
|
|
self.settings, self.provider, principal_mapper=PrincipalMapper().map_claims
|
|
)
|
|
url = bff.begin_login(now=self.now)
|
|
query = parse_qs(urlparse(url).query)
|
|
code = self.fake.authorize(url)
|
|
bff.complete_login(code=code, state=query["state"][0], now=self.now)
|
|
with self.assertRaises(InvalidState):
|
|
bff.complete_login(code=code, state=query["state"][0], now=self.now)
|
|
expired_url = bff.begin_login(now=self.now - timedelta(minutes=11))
|
|
expired_query = parse_qs(urlparse(expired_url).query)
|
|
with self.assertRaises(InvalidState):
|
|
bff.complete_login(
|
|
code=self.fake.authorize(expired_url),
|
|
state=expired_query["state"][0],
|
|
now=self.now,
|
|
)
|
|
|
|
def test_bad_signature_issuer_audience_expiry_and_key_rotation(self) -> None:
|
|
valid = self.fake.issue_machine_token("machine-one", now=self.now)
|
|
with self.assertRaises(InvalidToken):
|
|
self.provider.validate_token(valid[:-2], now=self.now)
|
|
self.fake.rotate_keys()
|
|
rotated = self.fake.issue_machine_token("machine-one", now=self.now)
|
|
self.provider.validate_token(rotated, now=self.now)
|
|
with self.assertRaises(InvalidToken):
|
|
self.provider.validate_token(rotated, now=self.now + timedelta(minutes=6))
|
|
with self.assertRaises(InvalidToken):
|
|
self.provider.validate_token(
|
|
self.fake.issue_machine_token(
|
|
"machine-one", now=self.now, issuer="https://foreign.oidc.test"
|
|
),
|
|
now=self.now,
|
|
)
|
|
with self.assertRaises(InvalidToken):
|
|
self.provider.validate_token(
|
|
self.fake.issue_machine_token(
|
|
"machine-one", now=self.now, audience="wrong-audience"
|
|
),
|
|
now=self.now,
|
|
)
|
|
foreign = FakeOIDCProvider(issuer="https://foreign.oidc.test", audience=self.fake.audience)
|
|
try:
|
|
with self.assertRaises(InvalidToken):
|
|
self.provider.validate_token(
|
|
foreign.issue_machine_token("machine-one", now=self.now), now=self.now
|
|
)
|
|
finally:
|
|
foreign_token = None
|
|
self.assertIsNone(foreign_token)
|
|
|
|
def test_trailing_slash_issuer_is_validated_exactly(self) -> None:
|
|
provider_fixture = FakeOIDCProvider(issuer="https://auth.oidc.test/application/o/hub/")
|
|
settings = OIDCSettings(
|
|
issuer=provider_fixture.issuer,
|
|
audience=provider_fixture.audience,
|
|
client_id="hub-bff",
|
|
redirect_uri="https://hub.test/auth/callback",
|
|
)
|
|
provider = OIDCProviderClient(settings, transport=provider_fixture.transport())
|
|
try:
|
|
provider.validate_token(
|
|
provider_fixture.issue_machine_token("machine-one", now=self.now), now=self.now
|
|
)
|
|
finally:
|
|
provider.close()
|
|
mismatched = OIDCProviderClient(
|
|
settings.model_copy(update={"issuer": str(provider_fixture.issuer).rstrip("/")}),
|
|
transport=provider_fixture.transport(),
|
|
)
|
|
try:
|
|
with self.assertRaises(ProviderUnavailable):
|
|
mismatched.discover()
|
|
finally:
|
|
mismatched.close()
|
|
|
|
def test_machine_subjects_are_distinct_and_revocation_is_bounded(self) -> None:
|
|
first_id = uuid4()
|
|
second_id = uuid4()
|
|
revoked: set = set()
|
|
cache = RevocationCache(lambda client_id: client_id in revoked, ttl_seconds=60)
|
|
mapper = PrincipalMapper(
|
|
machine_subjects={
|
|
"machine-one": MachineClientBinding("machine-one", first_id),
|
|
"machine-two": MachineClientBinding("machine-two", second_id),
|
|
}
|
|
)
|
|
auth = MachinePrincipalAuthenticator(self.provider, mapper, cache)
|
|
first = auth.authenticate(
|
|
"Bearer " + self.fake.issue_machine_token("machine-one", now=self.now)
|
|
)
|
|
second = auth.authenticate(
|
|
"Bearer " + self.fake.issue_machine_token("machine-two", now=self.now)
|
|
)
|
|
self.assertEqual(first.client_id, first_id)
|
|
self.assertEqual(second.client_id, second_id)
|
|
self.assertNotEqual(first.subject, second.subject)
|
|
revoked.add(first_id)
|
|
self.assertFalse(cache.is_revoked(first_id))
|
|
cache.invalidate(first_id)
|
|
with self.assertRaises(RevokedMachinePrincipal):
|
|
auth.authenticate(
|
|
"Bearer " + self.fake.issue_machine_token("machine-one", now=self.now)
|
|
)
|
|
|
|
def test_fastapi_router_never_returns_bearer_tokens(self) -> None:
|
|
bff = BFFService(
|
|
self.settings, self.provider, principal_mapper=PrincipalMapper().map_claims
|
|
)
|
|
login_url = bff.begin_login(now=self.now)
|
|
query = parse_qs(urlparse(login_url).query)
|
|
session = bff.complete_login(
|
|
code=self.fake.authorize(login_url), state=query["state"][0], now=self.now
|
|
)
|
|
app = FastAPI()
|
|
app.include_router(build_auth_router(bff))
|
|
client = TestClient(app)
|
|
|
|
anonymous = client.get("/auth/session")
|
|
self.assertEqual(anonymous.status_code, 200)
|
|
self.assertEqual(anonymous.json(), {"authenticated": False})
|
|
self.assertEqual(anonymous.headers["cache-control"], "no-store")
|
|
|
|
client.cookies.set(self.settings.cookie_name, "invalid-session")
|
|
invalid = client.get("/auth/session")
|
|
self.assertEqual(invalid.status_code, 401)
|
|
client.cookies.clear()
|
|
|
|
client.cookies.set(self.settings.cookie_name, session.value)
|
|
response = client.get("/auth/session")
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertNotIn("access_token", response.text)
|
|
self.assertNotIn("refresh_token", response.text)
|
|
csrf_token = response.json()["csrf_token"]
|
|
logout = client.post("/auth/logout", headers={"X-CSRF-Token": csrf_token})
|
|
self.assertEqual(logout.status_code, 204)
|
|
self.assertIn("Max-Age=0", logout.headers["set-cookie"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|