porchlight/tests/test_forward_auth.py

import re
from datetime import UTC, datetime
from urllib.parse import parse_qs, urlsplit

from argon2 import PasswordHasher
from httpx import AsyncClient

from porchlight.authn.password import PasswordService
from porchlight.models import PasswordCredential, User


async def _create_user(
    client: AsyncClient,
    *,
    username: str = "alice",
    password: str = "password123!Secure",
    groups: list[str] | None = None,
    email: str | None = None,
) -> User:
    app = client._transport.app  # type: ignore[union-attr]
    user = User(
        userid="forward-user-01",
        username=username,
        preferred_username="Alice",
        email=email,
        groups=groups or ["users"],
        created_at=datetime.now(UTC),
        updated_at=datetime.now(UTC),
    )
    await app.state.user_repo.create(user)

    svc = PasswordService(hasher=PasswordHasher(time_cost=1, memory_cost=8192))
    await app.state.credential_repo.create_password(
        PasswordCredential(user_id=user.userid, password_hash=svc.hash(password))
    )
    return user


async def _login(client: AsyncClient, *, username: str = "alice", password: str = "password123!Secure") -> None:
    token = await _login_page_csrf(client)
    res = await client.post(
        "/login/password",
        data={"username": username, "password": password},
        headers={"HX-Request": "true", "X-CSRF-Token": token},
    )
    assert res.status_code == 200
    assert res.headers.get("HX-Redirect")


async def _login_page_csrf(client: AsyncClient, return_to: str | None = None) -> str:
    url = "/login"
    if return_to is not None:
        url = f"/login?return_to={return_to}"
    resp = await client.get(url)
    match = re.search(r'name="csrf-token" content="([^"]+)"', resp.text)
    assert match, "CSRF meta tag not found in page"
    return match.group(1)


def _allow_forward_auth_hosts(client: AsyncClient, hosts: list[str]) -> None:
    app = client._transport.app  # type: ignore[union-attr]
    app.state.settings.forward_auth_allowed_redirect_hosts = hosts


async def test_forward_auth_redirects_unauthenticated_to_login(client: AsyncClient) -> None:
    res = await client.get("/forward-auth", follow_redirects=False)

    assert res.status_code == 303
    assert res.headers["location"] == "http://localhost:8000/login"
    assert res.headers["cache-control"] == "no-store"


async def test_forward_auth_redirect_includes_allowed_return_to(client: AsyncClient) -> None:
    _allow_forward_auth_hosts(client, ["app.example.com"])

    res = await client.get(
        "/forward-auth",
        headers={
            "X-Forwarded-Proto": "https",
            "X-Forwarded-Host": "app.example.com",
            "X-Forwarded-Uri": "/private?tab=1",
        },
        follow_redirects=False,
    )

    assert res.status_code == 303
    location = urlsplit(res.headers["location"])
    assert location.scheme == "http"
    assert location.netloc == "localhost:8000"
    assert location.path == "/login"
    assert parse_qs(location.query) == {"return_to": ["https://app.example.com/private?tab=1"]}


async def test_forward_auth_ignores_disallowed_return_host(client: AsyncClient) -> None:
    _allow_forward_auth_hosts(client, ["app.example.com"])

    res = await client.get(
        "/forward-auth",
        headers={
            "X-Forwarded-Proto": "https",
            "X-Forwarded-Host": "evil.example.net",
            "X-Forwarded-Uri": "/private",
        },
        follow_redirects=False,
    )

    assert res.status_code == 303
    assert res.headers["location"] == "http://localhost:8000/login"


async def test_forward_auth_allows_non_safe_proxy_subrequests_without_csrf(client: AsyncClient) -> None:
    res = await client.post("/forward-auth", follow_redirects=False)

    assert res.status_code == 303
    assert res.headers["location"] == "http://localhost:8000/login"


async def test_login_consumes_allowed_forward_auth_return_to(client: AsyncClient) -> None:
    _allow_forward_auth_hosts(client, ["app.example.com"])
    await _create_user(client)

    token = await _login_page_csrf(client, "https://app.example.com/private")
    res = await client.post(
        "/login/password",
        data={"username": "alice", "password": "password123!Secure"},
        headers={"HX-Request": "true", "X-CSRF-Token": token},
    )

    assert res.status_code == 200
    assert res.headers["HX-Redirect"] == "https://app.example.com/private"

    res2 = await client.post(
        "/login/password",
        data={"username": "alice", "password": "password123!Secure"},
        headers={"HX-Request": "true", "X-CSRF-Token": await _login_page_csrf(client)},
    )
    assert res2.headers["HX-Redirect"] == "/manage/credentials"


async def test_login_ignores_disallowed_forward_auth_return_to(client: AsyncClient) -> None:
    _allow_forward_auth_hosts(client, ["app.example.com"])
    await _create_user(client)

    await _login_page_csrf(client, "https://app.example.com/private")
    token = await _login_page_csrf(client, "https://evil.example.net/private")
    res = await client.post(
        "/login/password",
        data={"username": "alice", "password": "password123!Secure"},
        headers={"HX-Request": "true", "X-CSRF-Token": token},
    )

    assert res.status_code == 200
    assert res.headers["HX-Redirect"] == "/manage/credentials"


async def test_forward_auth_authenticated_user_returns_identity_headers(client: AsyncClient) -> None:
    await _create_user(client, groups=["users", "admin"], email="alice@example.com")
    await _login(client)

    res = await client.get("/forward-auth", follow_redirects=False)

    assert res.status_code == 204
    assert res.headers["remote-user"] == "alice"
    assert res.headers["remote-email"] == "alice@example.com"
    assert res.headers["remote-groups"] == "admin,users"
    assert res.headers["x-forwarded-user"] == "alice"
    assert res.headers["x-forwarded-userid"] == "forward-user-01"
    assert res.headers["x-forwarded-preferred-username"] == "Alice"
    assert res.headers["x-forwarded-email"] == "alice@example.com"
    assert res.headers["x-forwarded-groups"] == "admin,users"
    assert res.headers["cache-control"] == "no-store"


async def test_forward_auth_rejects_inactive_session_user(client: AsyncClient) -> None:
    user = await _create_user(client)
    await _login(client)

    app = client._transport.app  # type: ignore[union-attr]
    await app.state.user_repo.update(user.model_copy(update={"active": False}))

    res = await client.get("/forward-auth", follow_redirects=False)

    assert res.status_code == 303
    assert "remote-user" not in res.headers