"""Auth provider scaffold regression tests."""
from __future__ import annotations
import unittest
import base64
from contextlib import contextmanager
from typing import Any
from urllib.parse import parse_qs, urlencode, urlsplit
from fastapi import 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 .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")
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)
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_admin_created_external_domain_user_can_login(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="teacher",
account_status="approved",
cohort_ids=["manual-cohort"],
),
super_admin,
)
me = await auth_routes.dev_login(
_request(path="/auth/dev-login"),
auth_routes.DevLoginRequest(
email="forced.gmail@gmail.com",
role="learner",
display_name="Provider Name",
),
Response(),
)
self.assertEqual(me.user_id, created.user_id)
self.assertEqual(me.email, "forced.gmail@gmail.com")
self.assertEqual(me.role, "teacher")
self.assertEqual(me.account_status, "approved")
self.assertEqual(me.cohort_ids, ["manual-cohort"])
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)
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_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_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(
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_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_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_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, "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",
)
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_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, "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()