728 lines
29 KiB
Python
728 lines
29 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 Response
|
|
from starlette.requests import Request
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from .config import Settings, settings
|
|
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) -> Request:
|
|
return Request(
|
|
{
|
|
"type": "http",
|
|
"method": "GET",
|
|
"path": "/auth/login",
|
|
"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")
|
|
|
|
|
|
class AuthProviderScaffoldTest(unittest.IsolatedAsyncioTestCase):
|
|
async def asyncSetUp(self) -> None:
|
|
auth_routes._oauth_states.clear()
|
|
auth_routes._saml_states.clear()
|
|
|
|
async def asyncTearDown(self) -> None:
|
|
auth_routes._oauth_states.clear()
|
|
auth_routes._saml_states.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_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_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()
|