"""Auth provider scaffold regression tests.""" from __future__ import annotations import unittest import base64 from contextlib import contextmanager from datetime import datetime, timezone from typing import Any from urllib.parse import parse_qs, urlencode, urlsplit from fastapi import HTTPException, Response from starlette.requests import Request from unittest.mock import AsyncMock, patch from . import auth_sessions, deps from .config import Settings, settings from .deps import Principal, Role from .routes import admin as admin_routes from .routes import auth as auth_routes from .routes import sessions as session_routes from .saml import inflate_redirect_request @contextmanager def patched_settings(**values: Any): previous = {key: getattr(settings, key) for key in values} for key, value in values.items(): setattr(settings, key, value) try: yield finally: for key, value in previous.items(): setattr(settings, key, value) def _request( headers: list[tuple[bytes, bytes]] | None = None, path: str = "/auth/login", ) -> Request: return Request( { "type": "http", "method": "GET", "path": path, "headers": headers or [(b"host", b"localhost:8000")], } ) def _form_request(path: str, data: dict[str, str]) -> Request: body = urlencode(data).encode("utf-8") sent = False async def receive() -> dict[str, Any]: nonlocal sent if sent: return {"type": "http.request", "body": b"", "more_body": False} sent = True return {"type": "http.request", "body": body, "more_body": False} return Request( { "type": "http", "method": "POST", "path": path, "headers": [ (b"host", b"localhost:8000"), (b"content-type", b"application/x-www-form-urlencoded"), ], }, receive, ) def _fixture_saml_response( *, email: str = "learner@hs.ac.kr", display_name: str = "SAML Learner", role: str = "learner", cohort: str | None = None, ) -> str: cohort_attr = ( f'\n {cohort}' if cohort else "" ) xml = f""" {email} {email} {display_name} {role}{cohort_attr} """ return base64.b64encode(xml.encode("utf-8")).decode("ascii") def _managed_user( *, email: str, role: str = "learner", admin_access: bool = False, ) -> auth_sessions.ManagedUser: return auth_sessions.ManagedUser( user_id=auth_sessions.user_id_from_email(email), email=email, display_name=email, role=role, admin_access=admin_access, account_status="approved", cohort_ids=[], affiliation="", legal_name="", department="", grade_level="", phone="", contact_address="", nickname="", self_introduction="", avatar_url="", consent_at=None, profile_completed_at=1.0, terms_agreed_at=1.0, privacy_agreed_at=1.0, terms_version="", privacy_version="", created_at=1.0, last_seen_at=1.0, ) class AuthProviderScaffoldTest(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self) -> None: auth_routes._oauth_states.clear() auth_routes._saml_states.clear() auth_sessions._sessions.clear() auth_sessions._users.clear() auth_sessions._email_index.clear() auth_sessions._inactive_emails.clear() async def asyncTearDown(self) -> None: auth_routes._oauth_states.clear() auth_routes._saml_states.clear() auth_sessions._sessions.clear() auth_sessions._users.clear() auth_sessions._email_index.clear() auth_sessions._inactive_emails.clear() async def test_auth_config_reports_google_and_saml_provider_status(self) -> None: with patched_settings( oauth_google_client_id="google-client", oauth_google_client_secret="google-secret", auth_saml_enabled=True, saml_sp_entity_id="https://api-vignette.chanpaca.net/auth/saml/metadata", saml_sso_url="https://sso.hs.ac.kr/idp/profile/SAML2/Redirect/SSO", ): config = await auth_routes.auth_config(_request()) self.assertTrue(config.google_oauth_configured) self.assertTrue(config.saml_configured) providers = {item.provider: item for item in config.providers} self.assertTrue(providers["google"].enabled) self.assertTrue(providers["saml"].configured) self.assertTrue(providers["saml"].enabled) self.assertEqual(providers["saml"].login_path, "/auth/login?provider=saml") async def test_auth_config_allows_dev_login_from_configured_tailnet_origin(self) -> None: request = _request( [ (b"host", b"127.0.0.1:8000"), (b"origin", b"https://alpaca-home.taile93291.ts.net"), ] ) with patched_settings( environment="dev", auth_dev_login_enabled=True, auth_dev_login_extra_origins=["https://alpaca-home.taile93291.ts.net"], ): config = await auth_routes.auth_config(request) self.assertTrue(config.dev_login_enabled) self.assertEqual(config.allowed_email_domains, []) async def test_learner_can_accept_and_withdraw_practice_consent(self) -> None: with patch.object(auth_sessions, "get_pool", side_effect=RuntimeError("no db")): _, user = await auth_sessions.create_session( email="learner@hs.ac.kr", display_name="Learner", role="learner", external_id="dev:learner@hs.ac.kr", ) principal = Principal( user_id=user.user_id, role=Role.LEARNER, cohort_ids=user.cohort_ids, email=user.email, display_name=user.display_name, consent_at=user.consent_at, ) accepted = await auth_routes.accept_consent( auth_routes.ConsentRequest(accepted=True), principal, ) self.assertIsNotNone(accepted.consent_at) self.assertEqual(principal.consent_at, accepted.consent_at) self.assertTrue(await auth_sessions.user_has_consent(user.user_id)) withdrawn = await auth_routes.withdraw_consent(principal) self.assertIsNone(withdrawn.consent_at) self.assertIsNone(principal.consent_at) self.assertFalse(await auth_sessions.user_has_consent(user.user_id)) async def test_consent_rejects_non_learner_and_unaccepted_body(self) -> None: teacher = Principal( user_id="00000000-0000-0000-0000-000000000501", role=Role.TEACHER, email="teacher@hs.ac.kr", display_name="Teacher", ) learner = Principal( user_id="00000000-0000-0000-0000-000000000502", role=Role.LEARNER, email="learner@hs.ac.kr", display_name="Learner", ) with self.assertRaises(auth_routes.HTTPException) as teacher_error: await auth_routes.accept_consent(auth_routes.ConsentRequest(), teacher) with self.assertRaises(auth_routes.HTTPException) as learner_error: await auth_routes.accept_consent(auth_routes.ConsentRequest(accepted=False), learner) self.assertEqual(teacher_error.exception.status_code, 403) self.assertEqual(learner_error.exception.status_code, 400) async def test_new_provider_account_waits_for_admin_approval(self) -> None: with ( patched_settings( environment="dev", auth_new_user_default_status="pending", auth_super_admin_emails=["yunchan@twentyoz.kr"], auth_admin_emails=[], auth_teacher_emails=[], auth_approved_emails=[], ), patch.object(auth_sessions, "get_pool", side_effect=RuntimeError("no db")), ): sid, user = await auth_sessions.create_session( email="pending@hs.ac.kr", display_name="Pending Learner", role="learner", external_id="google:pending-provider-sub", ) self.assertEqual(user.account_status, "pending") me = await auth_routes._me_response(user) self.assertEqual(me.account_status, "pending") self.assertTrue(me.approval_required) principal = await deps.get_current_principal( _request(path="/auth/me"), dev_session_cookie=sid, ) self.assertEqual(principal.account_status, "pending") with self.assertRaises(auth_routes.HTTPException) as blocked: await deps.get_current_principal( _request(path="/personas"), dev_session_cookie=sid, ) self.assertEqual(blocked.exception.status_code, 403) self.assertEqual(blocked.exception.detail, "account_pending") async def test_dev_login_account_is_auto_approved_for_local_e2e(self) -> None: with ( patched_settings( environment="dev", auth_new_user_default_status="pending", auth_super_admin_emails=["yunchan@twentyoz.kr"], auth_admin_emails=[], auth_teacher_emails=[], auth_approved_emails=[], ), patch.object(auth_sessions, "get_pool", side_effect=RuntimeError("no db")), ): sid, user = await auth_sessions.create_session( email="dev-learner@hs.ac.kr", display_name="Dev Learner", role="learner", external_id="dev:dev-learner@hs.ac.kr", ) self.assertEqual(user.account_status, "approved") principal = await deps.get_current_principal( _request(path="/personas"), dev_session_cookie=sid, ) self.assertEqual(principal.account_status, "approved") async def test_dev_login_can_pin_shared_e2e_cohort_without_production_mapping( self, ) -> None: with ( patched_settings( environment="dev", auth_dev_login_enabled=True, auth_allowed_email_domains=["hs.ac.kr"], auth_domain_cohort_map={}, auth_email_cohort_map={}, ), patch.object(auth_sessions, "get_pool", side_effect=RuntimeError("no db")), ): me = await auth_routes.dev_login( _request(path="/auth/dev-login"), auth_routes.DevLoginRequest( email="cohort-e2e@hs.ac.kr", role="teacher", display_name="Cohort E2E Teacher", cohort_ids=[" e2e-hanshin ", "e2e-hanshin"], ), Response(), ) self.assertEqual(me.cohort_ids, ["e2e-hanshin"]) async def test_google_identity_accepts_any_verified_domain_and_auto_approves( self, ) -> None: with ( patched_settings( environment="dev", auth_dev_login_enabled=True, auth_allowed_email_domains=["hs.ac.kr"], auth_new_user_default_status="pending", auth_super_admin_emails=["yunchan@twentyoz.kr"], auth_admin_emails=[], auth_teacher_emails=[], auth_approved_emails=[], ), patch.object(auth_sessions, "get_pool", side_effect=RuntimeError("no db")), ): super_admin = Principal( user_id="00000000-0000-0000-0000-000000000604", role=Role.ADMIN, email="yunchan@twentyoz.kr", display_name="Yun Chan", admin_access=True, super_admin=True, ) created = await admin_routes.create_user( admin_routes.AdminUserCreate( email="forced.gmail@gmail.com", display_name="Forced Gmail", role="learner", cohort_ids=["manual-cohort"], ), super_admin, ) self.assertEqual(created.account_status, "pending") self.assertEqual( auth_routes.validate_google_identity( email="unmanaged@gmail.com", email_verified=True, ), "unmanaged@gmail.com", ) google_email = auth_routes.validate_google_identity( email="forced.gmail@gmail.com", email_verified=True, ) google_managed = await auth_sessions.get_managed_user_by_email(google_email) self.assertEqual(google_email, created.email) self.assertIsNotNone(google_managed) self.assertEqual(google_managed.user_id, created.user_id) _, google_user = await auth_sessions.create_session( email="forced.gmail@gmail.com", display_name="Provider Name", role="learner", external_id="google:external-participant", account_status="approved", ) self.assertEqual(google_user.user_id, created.user_id) self.assertEqual(google_user.account_status, "approved") self.assertEqual(google_user.cohort_ids, ["manual-cohort"]) suspended = await admin_routes.patch_user( created.user_id, admin_routes.AdminUserPatch(account_status="suspended"), super_admin, ) self.assertEqual(suspended.account_status, "suspended") _, suspended_login = await auth_sessions.create_session( email="forced.gmail@gmail.com", display_name="Provider Name", role="learner", external_id="google:external-participant", account_status="approved", ) self.assertEqual(suspended_login.account_status, "suspended") # Google is open to every verified account, while the local dev-login # fixture remains constrained to its explicit test domains. with self.assertRaises(auth_routes.HTTPException) as denied: await auth_routes.dev_login( _request(path="/auth/dev-login"), auth_routes.DevLoginRequest( email="unmanaged@gmail.com", role="learner", display_name="Unmanaged Gmail", ), Response(), ) self.assertEqual(denied.exception.status_code, 403) self.assertEqual(denied.exception.detail, "email domain is not allowed") with self.assertRaises(auth_routes.HTTPException) as unverified: auth_routes.validate_google_identity( email="unmanaged@gmail.com", email_verified=False, ) self.assertEqual(unverified.exception.status_code, 403) async def test_super_admin_email_is_admin_and_auto_approved(self) -> None: with ( patched_settings( environment="dev", auth_new_user_default_status="pending", auth_super_admin_emails=["yunchan@twentyoz.kr", "hoonjungkoo@hs.ac.kr"], auth_admin_emails=[], auth_teacher_emails=[], auth_approved_emails=[], ), patch.object(auth_sessions, "get_pool", side_effect=RuntimeError("no db")), ): role = auth_routes._role_for_email("yunchan@twentyoz.kr") _, user = await auth_sessions.create_session( email="yunchan@twentyoz.kr", display_name="Yun Chan", role=role.value, external_id="google:yunchan", ) self.assertEqual(role, Role.ADMIN) self.assertEqual(user.role, "admin") self.assertTrue(user.admin_access) self.assertEqual(user.account_status, "approved") hoonjung_role = auth_routes._role_for_email("hoonjungkoo@hs.ac.kr") self.assertEqual(hoonjung_role, Role.ADMIN) async def test_existing_primary_role_super_admin_persists_admin_access_on_login( self, ) -> None: email = "persistent-root@twentyoz.kr" existing = _managed_user(email=email, role="learner", admin_access=False) auth_sessions._users[existing.user_id] = existing auth_sessions._email_index[email] = existing.user_id with ( patched_settings( environment="dev", auth_super_admin_emails=[email], auth_admin_emails=[], auth_teacher_emails=[], auth_approved_emails=[], ), patch.object(auth_sessions, "get_pool", side_effect=RuntimeError("no db")), ): _, user = await auth_sessions.create_session( email=email, display_name="Persistent Root", role="learner", external_id="google:persistent-root", ) self.assertEqual(user.role, "learner") self.assertTrue(user.super_admin) self.assertTrue(user.admin_access) self.assertTrue(auth_sessions._users[existing.user_id].admin_access) async def test_pending_provider_user_enqueues_account_approval_notification(self) -> None: with ( patched_settings( environment="dev", auth_new_user_default_status="pending", auth_super_admin_emails=[], auth_admin_emails=[], auth_teacher_emails=[], auth_approved_emails=[], ), patch.object(auth_sessions, "get_pool", side_effect=RuntimeError("no db")), patch.object( auth_sessions.notifications, "enqueue_account_pending_approval", AsyncMock(), ) as enqueue_approval, ): _, user = await auth_sessions.create_session( email="new-approval@hs.ac.kr", display_name="New Approval", role="learner", external_id="google:new-approval", ) self.assertEqual(user.account_status, "pending") enqueue_approval.assert_awaited_once_with( user_id=user.user_id, email="new-approval@hs.ac.kr", display_name="New Approval", role="learner", ) async def test_super_admin_can_enter_teacher_and_learner_role_guards(self) -> None: principal = Principal( user_id="00000000-0000-0000-0000-000000000601", role=Role.LEARNER, email="hoonjungkoo@hs.ac.kr", display_name="Super Admin", admin_access=True, super_admin=True, ) teacher_checker = deps.require_role(Role.TEACHER, Role.ADMIN) teacher_view = await teacher_checker(principal) self.assertEqual(teacher_view.role, Role.ADMIN) self.assertTrue(principal.can_access_role(Role.LEARNER)) self.assertTrue(principal.can_access_role(Role.TEACHER)) self.assertTrue(principal.can_access_role(Role.ADMIN)) async def test_admin_role_can_enter_all_role_spaces_without_super_admin(self) -> None: principal = Principal( user_id="00000000-0000-0000-0000-000000000602", role=Role.ADMIN, email="operator@twentyoz.kr", display_name="Operator Admin", admin_access=True, super_admin=False, ) teacher_checker = deps.require_role(Role.TEACHER) teacher_view = await teacher_checker(principal) learner_view = session_routes._ensure_learner(principal) self.assertEqual(teacher_view.role, Role.TEACHER) self.assertEqual(learner_view.role, Role.LEARNER) self.assertTrue(principal.can_access_role(Role.LEARNER)) self.assertTrue(principal.can_access_role(Role.TEACHER)) self.assertTrue(principal.can_access_role(Role.ADMIN)) async def test_admin_role_can_enter_admin_console_even_with_stale_access_flag(self) -> None: principal = Principal( user_id="00000000-0000-0000-0000-000000000604", role=Role.ADMIN, email="stale-admin@twentyoz.kr", display_name="Stale Admin", admin_access=False, super_admin=False, ) admin_checker = deps.require_admin_access() admin_view = await admin_checker(principal) self.assertEqual(admin_view.role, Role.ADMIN) self.assertTrue(principal.can_access_role(Role.ADMIN)) async def test_admin_access_flag_without_admin_role_does_not_enter_learner_space(self) -> None: principal = Principal( user_id="00000000-0000-0000-0000-000000000603", role=Role.TEACHER, email="teacher-admin@hs.ac.kr", display_name="Teacher Admin", admin_access=True, super_admin=False, ) self.assertTrue(principal.can_access_role(Role.ADMIN)) self.assertFalse(principal.can_access_role(Role.LEARNER)) with self.assertRaises(HTTPException): session_routes._ensure_learner(principal) async def test_only_super_admin_can_grant_admin_access(self) -> None: with ( patched_settings( environment="dev", auth_super_admin_emails=["yunchan@twentyoz.kr"], auth_admin_emails=[], auth_teacher_emails=[], auth_approved_emails=[], ), patch.object(auth_sessions, "get_pool", side_effect=RuntimeError("no db")), ): target = await auth_sessions.upsert_managed_user( auth_sessions.ManagedUserUpsertInput( email="admin-grant-target@hs.ac.kr", display_name="Grant Target", role="learner", external_id="dev:admin-grant-target@hs.ac.kr", ) ) operator = Principal( user_id="00000000-0000-0000-0000-000000000602", role=Role.LEARNER, email="operator@hs.ac.kr", display_name="Operator", admin_access=True, super_admin=False, ) with self.assertRaises(auth_routes.HTTPException) as denied: await admin_routes.patch_user( target.user_id, admin_routes.AdminUserPatch(admin_access=True), operator, ) self.assertEqual(denied.exception.status_code, 403) super_admin = Principal( user_id="00000000-0000-0000-0000-000000000603", role=Role.ADMIN, email="yunchan@twentyoz.kr", display_name="Yun Chan", admin_access=True, super_admin=True, ) updated = await admin_routes.patch_user( target.user_id, admin_routes.AdminUserPatch(admin_access=True), super_admin, ) self.assertTrue(updated.admin_access) self.assertEqual(updated.role, "learner") async def test_super_admin_oauth_upsert_casts_nullable_admin_access_as_boolean( self, ) -> None: class RecordingConn: def __init__(self) -> None: self.queries: list[str] = [] async def fetchrow(self, query: str, *args: Any) -> dict[str, Any] | None: self.queries.append(query) if len(self.queries) == 1: return None now = datetime.now(timezone.utc) return { "user_id": "00000000-0000-0000-0000-000000000605", "email": "yunchan@twentyoz.kr", "display_name": "Yun Chan", "role": "learner", "admin_access": True, "account_status": "approved", "cohort": "", "affiliation": "", "created_at": now, "last_seen_at": now, } class RecordingAcquire: def __init__(self, conn: RecordingConn) -> None: self.conn = conn async def __aenter__(self) -> RecordingConn: return self.conn async def __aexit__(self, exc_type, exc, tb) -> None: return None class RecordingPool: def __init__(self, conn: RecordingConn) -> None: self.conn = conn def acquire(self) -> RecordingAcquire: return RecordingAcquire(self.conn) conn = RecordingConn() with ( patched_settings( auth_super_admin_emails=["yunchan@twentyoz.kr"], auth_admin_emails=[], ), patch.object(auth_sessions, "get_pool", return_value=RecordingPool(conn)), ): user = await auth_sessions.upsert_managed_user( auth_sessions.ManagedUserUpsertInput( email="yunchan@twentyoz.kr", display_name="Yun Chan", role="learner", external_id="google:111856072590637505974", ) ) self.assertTrue(user.admin_access) self.assertIn("$7::boolean", conn.queries[0]) self.assertIn("$9::boolean", conn.queries[1]) async def test_google_oauth_promotes_preregistered_pending_user_in_db(self) -> None: class RecordingConn: def __init__(self) -> None: self.query = "" self.args: tuple[Any, ...] = () async def fetchrow(self, query: str, *args: Any) -> dict[str, Any]: self.query = query self.args = args now = datetime.now(timezone.utc) return { "user_id": "00000000-0000-0000-0000-000000000606", "email": "preregistered@gmail.com", "display_name": "Preregistered", "role": "learner", "admin_access": False, "account_status": "approved", "cohort": "pilot", "affiliation": "", "created_at": now, "last_seen_at": now, } class RecordingAcquire: def __init__(self, conn: RecordingConn) -> None: self.conn = conn async def __aenter__(self) -> RecordingConn: return self.conn async def __aexit__(self, exc_type, exc, tb) -> None: return None class RecordingPool: def __init__(self, conn: RecordingConn) -> None: self.conn = conn def acquire(self) -> RecordingAcquire: return RecordingAcquire(self.conn) conn = RecordingConn() with patch.object(auth_sessions, "get_pool", return_value=RecordingPool(conn)): user = await auth_sessions.upsert_managed_user( auth_sessions.ManagedUserUpsertInput( email="preregistered@gmail.com", display_name="Preregistered", role="learner", external_id="google:preregistered-sub", account_status="approved", ) ) self.assertEqual(user.account_status, "approved") self.assertIn("WHEN $8::text = 'approved' THEN 'approved'", conn.query) self.assertEqual(conn.args[7], "approved") async def test_auth_config_allows_dev_login_from_configured_tailnet_forwarded_host(self) -> None: request = _request( [ (b"host", b"127.0.0.1:8000"), (b"x-forwarded-host", b"alpaca-home.taile93291.ts.net"), ] ) with patched_settings( environment="dev", auth_dev_login_enabled=True, auth_dev_login_extra_origins=["https://alpaca-home.taile93291.ts.net"], ): config = await auth_routes.auth_config(request) self.assertTrue(config.dev_login_enabled) async def test_auth_config_allows_dev_login_config_probe_without_origin_when_extra_origin_is_set(self) -> None: request = _request([(b"host", b"127.0.0.1:8000")]) with patched_settings( environment="dev", auth_dev_login_enabled=True, auth_dev_login_extra_origins=["https://alpaca-home.taile93291.ts.net"], ): config = await auth_routes.auth_config(request) self.assertTrue(config.dev_login_enabled) async def test_auth_config_keeps_dev_login_closed_for_public_origin(self) -> None: request = _request( [ (b"host", b"127.0.0.1:8000"), (b"origin", b"https://vignette.chanpaca.net"), (b"x-forwarded-host", b"api-vignette.chanpaca.net"), (b"x-forwarded-proto", b"https"), ] ) with patched_settings( environment="dev", auth_dev_login_enabled=True, auth_dev_login_extra_origins=["https://alpaca-home.taile93291.ts.net"], ): config = await auth_routes.auth_config(request) self.assertFalse(config.dev_login_enabled) async def test_frontend_origin_map_routes_vnet_api_callbacks_to_vnet_frontend(self) -> None: request = _request([(b"host", b"api-vnet.18ka.net")]) with patched_settings( frontend_base_url="https://vignette.chanpaca.net", frontend_origin_map={"api-vnet.18ka.net": "https://vnet.18ka.net"}, cors_origins=["https://vignette.chanpaca.net", "https://vnet.18ka.net"], ): origin = auth_routes._frontend_origin_for_request(request) self.assertEqual(origin, "https://vnet.18ka.net") async def test_saml_login_builds_redirect_authn_request_and_relay_state(self) -> None: with patched_settings( auth_saml_enabled=True, saml_sp_entity_id="https://api-vignette.chanpaca.net/auth/saml/metadata", saml_sso_url="https://sso.hs.ac.kr/idp/profile/SAML2/Redirect/SSO", ): response = await auth_routes.login(_request(), provider="saml", next="/learn") self.assertEqual(response.status_code, 302) location = response.headers["location"] self.assertTrue(location.startswith("https://sso.hs.ac.kr/idp/profile/SAML2/Redirect/SSO")) query = parse_qs(urlsplit(location).query) relay_state = query["RelayState"][0] self.assertIn(relay_state, auth_routes._saml_states) self.assertEqual(auth_routes._saml_states[relay_state].next_path, "/learn") xml = inflate_redirect_request(query["SAMLRequest"][0]) self.assertIn('Destination="https://sso.hs.ac.kr/idp/profile/SAML2/Redirect/SSO"', xml) self.assertIn( 'AssertionConsumerServiceURL="https://api-vignette.chanpaca.net/auth/saml/acs"', xml, ) self.assertIn( "https://api-vignette.chanpaca.net/auth/saml/metadata", xml, ) self.assertIn(auth_routes._saml_states[relay_state].request_id, xml) async def test_saml_acs_fixture_sets_opaque_cookie_without_browser_tokens(self) -> None: relay_state = "relay-state" auth_routes._saml_states[relay_state] = auth_routes.SamlState( request_id="_request", next_path="/learn", created_at=1_800_000_000.0, ) request = _form_request( "/auth/saml/acs", { "RelayState": relay_state, "SAMLResponse": _fixture_saml_response(role="teacher"), }, ) create_session_mock = AsyncMock(return_value=("opaque-session", object())) with ( patched_settings( auth_saml_enabled=True, saml_sp_entity_id="https://api-vignette.chanpaca.net/auth/saml/metadata", saml_sso_url="https://sso.hs.ac.kr/idp/profile/SAML2/Redirect/SSO", saml_x509_cert_fingerprint="", frontend_base_url="https://vignette.test", environment="dev", ), patch.object(auth_routes, "create_session", create_session_mock), ): response = await auth_routes.saml_acs(request) self.assertEqual(response.status_code, 302) self.assertEqual(response.headers["location"], "https://vignette.test/learn") create_session_mock.assert_awaited_once_with( email="learner@hs.ac.kr", display_name="SAML Learner", role="teacher", cohort_ids=[], external_id="saml:learner@hs.ac.kr", ) cookie_blob = "\n".join( value.decode("latin1") for name, value in response.raw_headers if name.lower() == b"set-cookie" ) self.assertIn("__Host-vignette_sid=opaque-session", cookie_blob) async def test_saml_acs_maps_cohort_claim_into_session(self) -> None: relay_state = "relay-state" auth_routes._saml_states[relay_state] = auth_routes.SamlState( request_id="_request", next_path="/teach", created_at=1_800_000_000.0, ) request = _form_request( "/auth/saml/acs", { "RelayState": relay_state, "SAMLResponse": _fixture_saml_response( email="teacher@hs.ac.kr", display_name="Teacher", role="teacher", cohort="counseling-2026-a, lab-b", ), }, ) create_session_mock = AsyncMock(return_value=("opaque-session", object())) with ( patched_settings( auth_saml_enabled=True, saml_sp_entity_id="https://api-vignette.chanpaca.net/auth/saml/metadata", saml_sso_url="https://sso.hs.ac.kr/idp/profile/SAML2/Redirect/SSO", saml_x509_cert_fingerprint="", frontend_base_url="https://vignette.test", environment="dev", ), patch.object(auth_routes, "create_session", create_session_mock), ): response = await auth_routes.saml_acs(request) self.assertEqual(response.status_code, 302) create_session_mock.assert_awaited_once_with( email="teacher@hs.ac.kr", display_name="Teacher", role="teacher", cohort_ids=["counseling-2026-a", "lab-b"], external_id="saml:teacher@hs.ac.kr", ) cookie_blob = "\n".join( value.decode("latin1") for name, value in response.raw_headers if name.lower() == b"set-cookie" ) self.assertIn("HttpOnly", cookie_blob) self.assertIn("Secure", cookie_blob) self.assertNotIn("SAMLResponse", cookie_blob) self.assertNotIn(relay_state, auth_routes._saml_states) async def test_saml_acs_sends_admin_entitled_generic_next_to_admin(self) -> None: relay_state = "relay-state" auth_routes._saml_states[relay_state] = auth_routes.SamlState( request_id="_request", next_path="/teach", created_at=1_800_000_000.0, ) request = _form_request( "/auth/saml/acs", { "RelayState": relay_state, "SAMLResponse": _fixture_saml_response( email="operator@hs.ac.kr", display_name="Operator", role="teacher", ), }, ) managed = _managed_user(email="operator@hs.ac.kr", role="learner", admin_access=True) create_session_mock = AsyncMock(return_value=("opaque-session", object())) with ( patched_settings( auth_saml_enabled=True, saml_sp_entity_id="https://api-vignette.chanpaca.net/auth/saml/metadata", saml_sso_url="https://sso.hs.ac.kr/idp/profile/SAML2/Redirect/SSO", saml_x509_cert_fingerprint="", frontend_base_url="https://vignette.test", environment="dev", ), patch.object(auth_routes, "get_managed_user_by_email", AsyncMock(return_value=managed)), patch.object(auth_routes, "create_session", create_session_mock), ): response = await auth_routes.saml_acs(request) self.assertEqual(response.status_code, 302) self.assertEqual(response.headers["location"], "https://vignette.test/admin") create_session_mock.assert_awaited_once_with( email="operator@hs.ac.kr", display_name="Operator", role="learner", cohort_ids=[], external_id="saml:operator@hs.ac.kr", ) self.assertNotIn(relay_state, auth_routes._saml_states) async def test_saml_acs_rejects_bad_relay_state(self) -> None: request = _form_request( "/auth/saml/acs", { "RelayState": "bad-relay", "SAMLResponse": _fixture_saml_response(), }, ) with patched_settings( auth_saml_enabled=True, saml_sp_entity_id="https://api-vignette.chanpaca.net/auth/saml/metadata", saml_sso_url="https://sso.hs.ac.kr/idp/profile/SAML2/Redirect/SSO", frontend_base_url="https://vignette.test", ): response = await auth_routes.saml_acs(request) self.assertEqual(response.status_code, 302) self.assertIn("oauth=saml_invalid_state", response.headers["location"]) async def test_saml_acs_rejects_when_signature_fingerprint_is_configured(self) -> None: relay_state = "relay-state" auth_routes._saml_states[relay_state] = auth_routes.SamlState( request_id="_request", next_path="/learn", created_at=1_800_000_000.0, ) request = _form_request( "/auth/saml/acs", { "RelayState": relay_state, "SAMLResponse": _fixture_saml_response(), }, ) with patched_settings( auth_saml_enabled=True, saml_sp_entity_id="https://api-vignette.chanpaca.net/auth/saml/metadata", saml_sso_url="https://sso.hs.ac.kr/idp/profile/SAML2/Redirect/SSO", saml_x509_cert_fingerprint="AA:BB:CC", frontend_base_url="https://vignette.test", ): response = await auth_routes.saml_acs(request) self.assertEqual(response.status_code, 302) self.assertIn("oauth=saml_signature_verification_required", response.headers["location"]) self.assertIn(relay_state, auth_routes._saml_states) async def test_saml_acs_unsigned_fixture_is_dev_only(self) -> None: relay_state = "relay-state" auth_routes._saml_states[relay_state] = auth_routes.SamlState( request_id="_request", next_path="/learn", created_at=1_800_000_000.0, ) request = _form_request( "/auth/saml/acs", { "RelayState": relay_state, "SAMLResponse": _fixture_saml_response(), }, ) with patched_settings( auth_saml_enabled=True, saml_sp_entity_id="https://api-vignette.chanpaca.net/auth/saml/metadata", saml_sso_url="https://sso.hs.ac.kr/idp/profile/SAML2/Redirect/SSO", saml_x509_cert_fingerprint="", frontend_base_url="https://vignette.test", environment="prod", ): response = await auth_routes.saml_acs(request) self.assertEqual(response.status_code, 302) self.assertIn("oauth=saml_fixture_acs_dev_only", response.headers["location"]) self.assertIn(relay_state, auth_routes._saml_states) async def test_unknown_provider_still_fails_as_unsupported(self) -> None: response = await auth_routes.login(_request(), provider="github") self.assertEqual(response.status_code, 302) self.assertIn("oauth=unsupported_provider", response.headers["location"]) async def test_google_login_uses_pkce_state_without_exposing_secret(self) -> None: with patched_settings( environment="prod", oauth_google_client_id="google-client", oauth_google_client_secret="google-secret", oauth_redirect_uri="https://api-vignette.test/auth/callback", ): response = await auth_routes.login(_request(), provider="google", next="//evil.test") self.assertEqual(response.status_code, 302) location = response.headers["location"] self.assertTrue(location.startswith(auth_routes.GOOGLE_AUTHORIZE_URL)) self.assertNotIn("google-secret", location) query = parse_qs(urlsplit(location).query) state = query["state"][0] self.assertIn(state, auth_routes._oauth_states) stored = auth_routes._oauth_states[state] self.assertEqual(stored.next_path, "/") self.assertEqual(query["client_id"], ["google-client"]) self.assertEqual(query["redirect_uri"], ["https://api-vignette.test/auth/callback"]) self.assertEqual(query["response_type"], ["code"]) self.assertEqual(query["code_challenge_method"], ["S256"]) self.assertEqual( query["code_challenge"], [auth_routes._pkce_challenge(stored.code_verifier)], ) cookie_blob = "\n".join( value.decode("latin1") for name, value in response.raw_headers if name.lower() == b"set-cookie" ) self.assertIn("__Host-vignette_oauth_state=", cookie_blob) self.assertIn("HttpOnly", cookie_blob) self.assertIn("Secure", cookie_blob) async def test_dev_google_login_rejects_public_callback_redirect(self) -> None: request = _request( [ (b"host", b"127.0.0.1:8010"), (b"x-forwarded-host", b"alpaca-home.taile93291.ts.net"), (b"x-forwarded-proto", b"https"), ] ) with patched_settings( environment="dev", auth_dev_login_enabled=True, auth_dev_login_extra_origins=["https://alpaca-home.taile93291.ts.net"], oauth_google_client_id="google-client", oauth_google_client_secret="google-secret", oauth_redirect_uri="https://api-vignette.chanpaca.net/auth/callback", frontend_base_url="https://vignette.chanpaca.net", cors_origins=["https://vignette.chanpaca.net"], ): response = await auth_routes.login(request, provider="google", next="/learn") self.assertEqual(response.status_code, 302) location = response.headers["location"] self.assertIn("https://alpaca-home.taile93291.ts.net/login", location) self.assertIn("oauth=local_oauth_unavailable", location) self.assertFalse(auth_routes._oauth_states) async def test_google_callback_accepts_unregistered_external_email(self) -> None: state = "external-email-state" auth_routes._oauth_states[state] = auth_routes.OAuthState( code_verifier="verifier", next_path="/learn", created_at=1_800_000_000.0, ) class FakeResponse: def __init__(self, payload: dict[str, Any]) -> None: self.status_code = 200 self._payload = payload def json(self) -> dict[str, Any]: return self._payload class FakeAsyncClient: def __init__(self, *args: Any, **kwargs: Any) -> None: pass async def __aenter__(self) -> "FakeAsyncClient": return self async def __aexit__(self, exc_type: object, exc: object, tb: object) -> None: return None async def post(self, url: str, **kwargs: Any) -> FakeResponse: return FakeResponse({"id_token": "id-token"}) async def get(self, url: str, **kwargs: Any) -> FakeResponse: return FakeResponse( { "aud": "google-client", "iss": "https://accounts.google.com", "email": "unregistered@gmail.com", "email_verified": "true", "name": "Unregistered External", "sub": "google-external-sub", } ) create_session_mock = AsyncMock(return_value=("opaque-session", object())) with ( patched_settings( oauth_google_client_id="google-client", oauth_google_client_secret="google-secret", frontend_base_url="https://vignette.test", environment="prod", auth_allowed_email_domains=["hs.ac.kr"], ), patch.object(auth_routes.httpx, "AsyncClient", FakeAsyncClient), patch.object( auth_routes, "get_managed_user_by_email", AsyncMock(return_value=None), ), patch.object(auth_routes, "create_session", create_session_mock), ): response = await auth_routes.callback( _request(), code="auth-code", state=state, oauth_state_cookie=state, ) self.assertEqual(response.status_code, 302) self.assertEqual(response.headers["location"], "https://vignette.test/learn") create_session_mock.assert_awaited_once_with( email="unregistered@gmail.com", display_name="Unregistered External", role="learner", cohort_ids=[], external_id="google:google-external-sub", account_status="approved", ) async def test_google_callback_sets_opaque_cookie_without_browser_tokens(self) -> None: state = "state-token" auth_routes._oauth_states[state] = auth_routes.OAuthState( code_verifier="verifier", next_path="/learn", created_at=1_800_000_000.0, ) class FakeResponse: def __init__(self, status_code: int, payload: dict[str, Any]) -> None: self.status_code = status_code self._payload = payload def json(self) -> dict[str, Any]: return self._payload class FakeAsyncClient: def __init__(self, *args: Any, **kwargs: Any) -> None: self.calls: list[tuple[str, str, dict[str, Any]]] = [] async def __aenter__(self) -> "FakeAsyncClient": return self async def __aexit__(self, exc_type: object, exc: object, tb: object) -> None: return None async def post(self, url: str, **kwargs: Any) -> FakeResponse: self.calls.append(("POST", url, kwargs)) return FakeResponse( 200, {"id_token": "id-token", "access_token": "browser-must-not-see-this"}, ) async def get(self, url: str, **kwargs: Any) -> FakeResponse: self.calls.append(("GET", url, kwargs)) return FakeResponse( 200, { "aud": "google-client", "iss": "https://accounts.google.com", "email": "learner@hs.ac.kr", "email_verified": "true", "name": "Learner", "hd": "hs.ac.kr", }, ) create_session_mock = AsyncMock(return_value=("opaque-session", object())) with ( patched_settings( oauth_google_client_id="google-client", oauth_google_client_secret="google-secret", oauth_redirect_uri="https://api-vignette.test/auth/callback", frontend_base_url="https://vignette.test", environment="prod", auth_domain_cohort_map={"hs.ac.kr": "hanshin-2026"}, auth_email_cohort_map={"learner@hs.ac.kr": "pilot-a"}, ), patch.object(auth_routes.httpx, "AsyncClient", FakeAsyncClient), patch.object( auth_routes, "get_managed_user_by_email", AsyncMock(return_value=None), ), patch.object(auth_routes, "create_session", create_session_mock), ): response = await auth_routes.callback( _request(), code="auth-code", state=state, oauth_state_cookie=state, ) self.assertEqual(response.status_code, 302) self.assertEqual(response.headers["location"], "https://vignette.test/learn") create_session_mock.assert_awaited_once_with( email="learner@hs.ac.kr", display_name="Learner", role="learner", cohort_ids=["pilot-a", "hanshin-2026"], external_id="google:learner@hs.ac.kr", account_status="approved", ) cookie_blob = "\n".join( value.decode("latin1") for name, value in response.raw_headers if name.lower() == b"set-cookie" ) self.assertIn("__Host-vignette_sid=opaque-session", cookie_blob) self.assertIn("HttpOnly", cookie_blob) self.assertIn("Secure", cookie_blob) self.assertNotIn("id-token", cookie_blob) self.assertNotIn("browser-must-not-see-this", cookie_blob) self.assertNotIn(state, auth_routes._oauth_states) async def test_google_callback_sends_admin_entitled_generic_next_to_admin(self) -> None: state = "state-token" auth_routes._oauth_states[state] = auth_routes.OAuthState( code_verifier="verifier", next_path="/learn", created_at=1_800_000_000.0, ) managed = _managed_user(email="operator@hs.ac.kr", role="learner", admin_access=True) class FakeResponse: def __init__(self, status_code: int, payload: dict[str, Any]) -> None: self.status_code = status_code self._payload = payload def json(self) -> dict[str, Any]: return self._payload class FakeAsyncClient: def __init__(self, *args: Any, **kwargs: Any) -> None: pass async def __aenter__(self) -> "FakeAsyncClient": return self async def __aexit__(self, exc_type: object, exc: object, tb: object) -> None: return None async def post(self, url: str, **kwargs: Any) -> FakeResponse: return FakeResponse(200, {"id_token": "id-token"}) async def get(self, url: str, **kwargs: Any) -> FakeResponse: return FakeResponse( 200, { "aud": "google-client", "iss": "https://accounts.google.com", "email": "operator@hs.ac.kr", "email_verified": "true", "name": "Operator", "hd": "hs.ac.kr", }, ) create_session_mock = AsyncMock(return_value=("opaque-session", object())) with ( patched_settings( oauth_google_client_id="google-client", oauth_google_client_secret="google-secret", oauth_redirect_uri="https://api-vignette.test/auth/callback", frontend_base_url="https://vignette.test", environment="prod", auth_admin_emails=[], auth_super_admin_emails=[], ), patch.object(auth_routes.httpx, "AsyncClient", FakeAsyncClient), patch.object(auth_routes, "get_managed_user_by_email", AsyncMock(return_value=managed)), patch.object(auth_routes, "create_session", create_session_mock), ): response = await auth_routes.callback( _request(), code="auth-code", state=state, oauth_state_cookie=state, ) self.assertEqual(response.status_code, 302) self.assertEqual(response.headers["location"], "https://vignette.test/admin") create_session_mock.assert_awaited_once_with( email="operator@hs.ac.kr", display_name="Operator", role="learner", cohort_ids=[], external_id="google:operator@hs.ac.kr", account_status="approved", ) self.assertEqual( auth_routes._post_login_next_path( "/learn/session/session-id", email="operator@hs.ac.kr", role=Role.LEARNER, managed_user=managed, ), "/learn/session/session-id", ) async def test_google_callback_accepts_signed_state_after_process_restart(self) -> None: with patched_settings( environment="prod", oauth_google_client_id="google-client", oauth_google_client_secret="google-secret", oauth_redirect_uri="https://api-vignette.test/auth/callback", frontend_base_url="https://vignette.test", session_secret="signed-oauth-state-secret", ): login_response = await auth_routes.login(_request(), provider="google", next="/learn") query = parse_qs(urlsplit(login_response.headers["location"]).query) state = query["state"][0] expected_verifier = auth_routes._oauth_states[state].code_verifier auth_routes._oauth_states.clear() calls: list[tuple[str, str, dict[str, Any]]] = [] class FakeResponse: def __init__(self, status_code: int, payload: dict[str, Any]) -> None: self.status_code = status_code self._payload = payload def json(self) -> dict[str, Any]: return self._payload class FakeAsyncClient: def __init__(self, *args: Any, **kwargs: Any) -> None: pass async def __aenter__(self) -> "FakeAsyncClient": return self async def __aexit__(self, exc_type: object, exc: object, tb: object) -> None: return None async def post(self, url: str, **kwargs: Any) -> FakeResponse: calls.append(("POST", url, kwargs)) return FakeResponse(200, {"id_token": "id-token"}) async def get(self, url: str, **kwargs: Any) -> FakeResponse: calls.append(("GET", url, kwargs)) return FakeResponse( 200, { "aud": "google-client", "iss": "https://accounts.google.com", "email": "learner@hs.ac.kr", "email_verified": "true", "name": "Learner", "hd": "hs.ac.kr", }, ) with ( patched_settings( environment="prod", oauth_google_client_id="google-client", oauth_google_client_secret="google-secret", oauth_redirect_uri="https://api-vignette.test/auth/callback", frontend_base_url="https://vignette.test", session_secret="signed-oauth-state-secret", ), patch.object(auth_routes.httpx, "AsyncClient", FakeAsyncClient), patch.object( auth_routes, "get_managed_user_by_email", AsyncMock(return_value=None), ), patch.object(auth_routes, "create_session", AsyncMock(return_value=("opaque-session", object()))), ): response = await auth_routes.callback( _request(), code="auth-code", state=state, oauth_state_cookie=state, ) self.assertEqual(response.status_code, 302) self.assertEqual(response.headers["location"], "https://vignette.test/learn") self.assertEqual(calls[0][2]["data"]["code_verifier"], expected_verifier) async def test_google_callback_requires_state_cookie_match(self) -> None: with patched_settings( environment="prod", oauth_google_client_id="google-client", oauth_google_client_secret="google-secret", oauth_redirect_uri="https://api-vignette.test/auth/callback", frontend_base_url="https://vignette.test", session_secret="signed-oauth-state-secret", ): login_response = await auth_routes.login(_request(), provider="google", next="/learn") state = parse_qs(urlsplit(login_response.headers["location"]).query)["state"][0] auth_routes._oauth_states.clear() with patched_settings( environment="prod", oauth_google_client_id="google-client", oauth_google_client_secret="google-secret", oauth_redirect_uri="https://api-vignette.test/auth/callback", frontend_base_url="https://vignette.test", session_secret="signed-oauth-state-secret", ): response = await auth_routes.callback(_request(), code="auth-code", state=state) self.assertEqual(response.status_code, 302) self.assertIn("oauth=invalid_state", response.headers["location"]) async def test_google_callback_maps_provider_error_reason(self) -> None: with patched_settings( environment="prod", oauth_google_client_id="google-client", oauth_google_client_secret="google-secret", oauth_redirect_uri="https://api-vignette.test/auth/callback", frontend_base_url="https://vignette.test", session_secret="signed-oauth-state-secret", ): response = await auth_routes.callback( _request(), state="provider-state", error="access_denied", error_description="The user denied access.", ) self.assertEqual(response.status_code, 302) self.assertEqual(response.headers["location"], "https://vignette.test/login?oauth=access_denied") def test_session_cookie_is_host_prefixed_httponly_secure_lax_without_domain(self) -> None: response = Response() with patched_settings(environment="prod", cookie_name="__Host-vignette_sid"): auth_routes._set_session_cookie(response, "opaque-session") cookie_blob = "\n".join( value.decode("latin1") for name, value in response.raw_headers if name.lower() == b"set-cookie" ) self.assertIn("__Host-vignette_sid=opaque-session", cookie_blob) self.assertIn("HttpOnly", cookie_blob) self.assertIn("Secure", cookie_blob) self.assertIn("SameSite=lax", cookie_blob) self.assertIn("Path=/", cookie_blob) self.assertNotIn("Domain=", cookie_blob) self.assertNotIn("vignette_sid=opaque-session", cookie_blob.replace("__Host-vignette_sid", "")) def test_dev_login_sets_secondary_local_cookie_only_in_dev(self) -> None: response = Response() with patched_settings(environment="dev", cookie_name="__Host-vignette_sid"): auth_routes._set_session_cookie(response, "dev-session") cookie_blob = "\n".join( value.decode("latin1") for name, value in response.raw_headers if name.lower() == b"set-cookie" ) self.assertIn("__Host-vignette_sid=dev-session", cookie_blob) self.assertIn("vignette_sid=dev-session", cookie_blob) def test_saml_enabled_requires_placeholder_config(self) -> None: with self.assertRaises(ValueError) as caught: Settings(auth_saml_enabled=True) error = str(caught.exception) self.assertIn("SAML_SP_ENTITY_ID", error) self.assertIn("SAML_SSO_URL", error) cfg = Settings( auth_saml_enabled=True, saml_sp_entity_id="https://api-vignette.chanpaca.net/auth/saml/metadata", saml_sso_url="https://sso.hs.ac.kr/idp/profile/SAML2/Redirect/SSO", ) self.assertTrue(cfg.auth_saml_enabled) if __name__ == "__main__": unittest.main()