1268 lines
50 KiB
Python
1268 lines
50 KiB
Python
"""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 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 <saml:Attribute Name="cohort"><saml:AttributeValue>{cohort}</saml:AttributeValue></saml:Attribute>'
|
|
if cohort
|
|
else ""
|
|
)
|
|
xml = f"""<samlp:Response xmlns:samlp="urn:oasis:names:tc:SAML:2.0:protocol" xmlns:saml="urn:oasis:names:tc:SAML:2.0:assertion">
|
|
<saml:Assertion>
|
|
<saml:Subject><saml:NameID>{email}</saml:NameID></saml:Subject>
|
|
<saml:AttributeStatement>
|
|
<saml:Attribute Name="email"><saml:AttributeValue>{email}</saml:AttributeValue></saml:Attribute>
|
|
<saml:Attribute Name="displayName"><saml:AttributeValue>{display_name}</saml:AttributeValue></saml:Attribute>
|
|
<saml:Attribute Name="role"><saml:AttributeValue>{role}</saml:AttributeValue></saml:Attribute>{cohort_attr}
|
|
</saml:AttributeStatement>
|
|
</saml:Assertion>
|
|
</samlp:Response>"""
|
|
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)
|
|
|
|
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_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_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(
|
|
"<saml:Issuer>https://api-vignette.chanpaca.net/auth/saml/metadata</saml:Issuer>",
|
|
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_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_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",
|
|
)
|
|
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, "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()
|