tavolo/server/packages/tavolo-platform/tests/test_auth.py

"""Tests for the auth helpers' OIDC user lookup and gating.

kaya-oidc (0.0.6) namespaces session state per provider as
``oidc:{provider}:user`` and tavolo wires an :class:`OIDCProviders`; these
tests pin both contracts so a kaya upgrade or a provider change cannot
silently break HTTP or websocket authentication.
"""
from __future__ import annotations

import unittest
from types import SimpleNamespace
from typing import Any

from kaya.oidc import OIDCConfig, OIDCProviders, OIDCUser
from kaya.session import InMemorySessionStore, Session, SessionMixin

from helpers import async_test
from tavolo.platform.auth import (
    current_user,
    display_name,
    get_ws_user,
    picture,
    require_auth,
    session_user,
)


def _providers(*names: str) -> OIDCProviders:
    session = SessionMixin(InMemorySessionStore())
    return OIDCProviders(
        [
            OIDCConfig(
                provider=name,
                issuer=f"http://localhost:8180/{name}",
                client_id="tavolo",
                redirect_uri=f"http://localhost:8080/auth/{name}/callback",
            )
            for name in names
        ],
        session=session,
    )


def _ctx(data: Any) -> Any:
    return SimpleNamespace(session=Session("sid", data))


def _ws(session: Any) -> Any:
    return SimpleNamespace(session=session)


class SessionUserTest(unittest.TestCase):
    def test_finds_default_provider_user(self) -> None:
        user = session_user(Session("sid", {"oidc:default:user": {"sub": "alice"}}))
        assert user is not None
        self.assertEqual("alice", user.sub)

    def test_finds_user_of_any_provider(self) -> None:
        user = session_user(Session("sid", {"oidc:google:user": {"sub": "alice"}}))
        assert user is not None
        self.assertEqual("alice", user.sub)

    def test_ignores_flow_state_and_tokens(self) -> None:
        session = Session(
            "sid",
            {
                "oidc:default:state": "state",
                "oidc:default:id_token": "token",
                "oidc:default:access_token": "token",
            },
        )
        self.assertIsNone(session_user(session))

    def test_ignores_legacy_unprefixed_user_key(self) -> None:
        session = Session("sid", {"oidc_user": {"sub": "alice"}})
        self.assertIsNone(session_user(session))

    def test_ignores_malformed_user_claims(self) -> None:
        session = Session("sid", {"oidc:default:user": "alice"})
        self.assertIsNone(session_user(session))


class GetWsUserTest(unittest.TestCase):
    def test_returns_user_from_provider_namespaced_session(self) -> None:
        session = Session(
            "sid",
            {"oidc:default:user": {"sub": "alice", "preferred_username": "Alice"}},
        )
        user = get_ws_user(_ws(session))
        assert user is not None
        self.assertEqual("alice", user.sub)

    def test_missing_session_returns_none(self) -> None:
        self.assertIsNone(get_ws_user(SimpleNamespace()))


class DisplayNameTest(unittest.TestCase):
    """The claim priority used for the name shown at the table."""

    def test_prefers_preferred_username_over_name(self) -> None:
        user = OIDCUser(
            {"sub": "alice", "preferred_username": "alice", "name": "Alice"}
        )
        self.assertEqual("alice", display_name(user))

    def test_falls_back_to_name(self) -> None:
        user = OIDCUser({"sub": "alice", "name": "Alice"})
        self.assertEqual("Alice", display_name(user))

    def test_ignores_email(self) -> None:
        # The app requests only the openid and profile scopes; an email
        # claim the provider still returns must not leak into the name.
        user = OIDCUser({"sub": "alice", "email": "alice@example.com"})
        self.assertEqual("alice", display_name(user))

    def test_falls_back_to_sub(self) -> None:
        user = OIDCUser({"sub": "alice"})
        self.assertEqual("alice", display_name(user))

    def test_ignores_empty_and_non_string_claims(self) -> None:
        user = OIDCUser({"sub": "alice", "preferred_username": "", "name": 42})
        self.assertEqual("alice", display_name(user))


class PictureTest(unittest.TestCase):
    """Only absolute http(s) picture claims are exported to the SPA."""

    def test_uses_an_absolute_http_url(self) -> None:
        user = OIDCUser({"sub": "alice", "picture": "https://cdn.example.com/a.png"})
        self.assertEqual("https://cdn.example.com/a.png", picture(user))

    def test_rejects_relative_and_script_schemes(self) -> None:
        for value in ("/relative.png", "javascript:alert(1)", "data:image/png;base64,x"):
            with self.subTest(picture=value):
                user = OIDCUser({"sub": "alice", "picture": value})
                self.assertIsNone(picture(user))

    def test_missing_claim_is_none(self) -> None:
        self.assertIsNone(picture(OIDCUser({"sub": "alice"})))


class CurrentUserTest(unittest.TestCase):
    def test_anonymous_session_has_no_user(self) -> None:
        self.assertIsNone(current_user(_providers("default"), _ctx({})))

    def test_returns_the_authenticated_provider_user(self) -> None:
        user = current_user(
            _providers("default", "google"),
            _ctx({"oidc:google:user": {"sub": "alice"}}),
        )
        assert user is not None
        self.assertEqual("alice", user.sub)

    def test_first_configured_provider_wins(self) -> None:
        # Single-identity sessions never hold two users; this only pins the
        # tie-break should they ever coexist.
        user = current_user(
            _providers("default", "google"),
            _ctx(
                {
                    "oidc:default:user": {"sub": "alice"},
                    "oidc:google:user": {"sub": "bob"},
                }
            ),
        )
        assert user is not None
        self.assertEqual("alice", user.sub)


class RequireAuthTest(unittest.TestCase):
    @async_test
    async def test_rejects_anonymous_requests(self) -> None:
        oidc = _providers("default")
        ctx = _ctx({})
        sent = {}

        async def send_bytes(*args: Any) -> None:
            sent["status"] = args[0]

        ctx.send_bytes = send_bytes
        handled = []

        @require_auth(oidc)
        async def handler(_ctx: Any) -> None:
            handled.append(True)

        await handler(ctx)
        self.assertEqual(401, sent["status"])
        self.assertEqual([], handled)

    @async_test
    async def test_allows_authenticated_requests(self) -> None:
        oidc = _providers("default")
        ctx = _ctx({"oidc:default:user": {"sub": "alice"}})
        handled = []

        @require_auth(oidc)
        async def handler(request_ctx: Any) -> None:
            user = current_user(oidc, request_ctx)
            assert user is not None
            handled.append(user.sub)

        await handler(ctx)
        self.assertEqual(["alice"], handled)