vignette/apps/api/app/test_auth_providers.py

"""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",
) -> str:
    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>
    </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_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=[],
        )
        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("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(
            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)],
        )

    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",
                    },
                )

        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",
            ),
            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)

        self.assertEqual(response.status_code, 302)
        self.assertEqual(response.headers["location"], "https://vignette.test/learn")
        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)

    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()