porchlight/tests/test_forward_auth.py

import re
from datetime import UTC, datetime
from unittest.mock import MagicMock
from urllib.parse import parse_qs, urlsplit

from argon2 import PasswordHasher
from httpx import AsyncClient

from porchlight.authn.password import PasswordService
from porchlight.authn.routes import _login_redirect_target
from porchlight.forward_auth import FORWARD_AUTH_RETURN_TO_SESSION_KEY
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 if groups is not None else ["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_rejects_backslash_wildcard_return_host(client: AsyncClient) -> None:
    _allow_forward_auth_hosts(client, ["*.apps.example.com"])

    res = await client.get(
        "/forward-auth",
        headers={
            "X-Forwarded-Proto": "https",
            "X-Forwarded-Host": r"evil.com\x.apps.example.com",
            "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_bare_hosts_require_https_default_port(client: AsyncClient) -> None:
    _allow_forward_auth_hosts(client, ["app.example.com"])

    http_res = await client.get(
        "/forward-auth",
        headers={
            "X-Forwarded-Proto": "http",
            "X-Forwarded-Host": "app.example.com",
            "X-Forwarded-Uri": "/private",
        },
        follow_redirects=False,
    )
    custom_port_res = await client.get(
        "/forward-auth",
        headers={
            "X-Forwarded-Proto": "https",
            "X-Forwarded-Host": "app.example.com:1337",
            "X-Forwarded-Uri": "/private",
        },
        follow_redirects=False,
    )

    assert http_res.headers["location"] == "http://localhost:8000/login"
    assert custom_port_res.headers["location"] == "http://localhost:8000/login"


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

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

    assert parse_qs(urlsplit(res.headers["location"]).query) == {"return_to": ["https://app.example.com/private"]}


async def test_forward_auth_supports_scheme_qualified_ipv6_patterns(client: AsyncClient) -> None:
    _allow_forward_auth_hosts(client, ["http://[::1]:8080"])

    res = await client.get(
        "/forward-auth",
        headers={
            "X-Forwarded-Proto": "http",
            "X-Forwarded-Host": "[::1]:8080",
            "X-Forwarded-Uri": "/private",
        },
        follow_redirects=False,
    )

    assert parse_qs(urlsplit(res.headers["location"]).query) == {"return_to": ["http://[::1]:8080/private"]}


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_login_rejects_backslash_wildcard_return_to(client: AsyncClient) -> None:
    _allow_forward_auth_hosts(client, ["*.apps.example.com"])
    await _create_user(client)

    token = await _login_page_csrf(client, "https://evil.com%5Cx.apps.example.com/")
    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_authenticated_user_returns_empty_identity_headers(client: AsyncClient) -> None:
    await _create_user(client, groups=[], email=None)
    await _login(client)

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

    assert res.status_code == 204
    assert res.headers["remote-email"] == ""
    assert res.headers["remote-groups"] == ""
    assert res.headers["x-forwarded-email"] == ""
    assert res.headers["x-forwarded-groups"] == ""


async def test_forward_auth_unsafe_header_values_are_zeroed(client: AsyncClient) -> None:
    user = await _create_user(client)
    app = client._transport.app  # type: ignore[union-attr]
    await app.state.user_repo.update(user.model_copy(update={"preferred_username": "bad\x0bname"}))
    await _login(client)

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

    assert res.status_code == 204
    assert res.headers["x-forwarded-preferred-username"] == ""


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


def test_login_redirect_target_clears_forward_auth_when_oidc_wins() -> None:
    request = MagicMock()
    request.session = {
        "oidc_auth_request": {"client_id": "test-rp"},
        FORWARD_AUTH_RETURN_TO_SESSION_KEY: "https://app.example.com/private",
    }

    assert _login_redirect_target(request) == "/authorization/complete"
    assert FORWARD_AUTH_RETURN_TO_SESSION_KEY not in request.session