README.md

@@ -96,6 +96,7 @@ variables always take priority over file values.
 | `OIDC_OP_INVITE_TTL` | `86400` | Magic link expiry in seconds |
 | `OIDC_OP_MANAGE_CLIENT_ID` | `manage-app` | Client ID for the management UI |
 | `OIDC_OP_SESSION_HTTPS_ONLY` | `true` | Restrict session cookie to HTTPS (set `false` for local dev) |
+| `OIDC_OP_FORWARD_AUTH_ALLOWED_REDIRECT_HOSTS` | `[]` | JSON list of hosts allowed as post-login return targets for `/forward-auth` |
 | `OIDC_OP_CONFIG_FILE` | `porchlight.toml` | Path to TOML config file |
 
 Database migrations run automatically on startup.
@@ -123,6 +124,33 @@ startup. Only `client_secret` and `redirect_uris` are required; the other
 fields have sensible defaults (`response_types = ["code"]`,
 `scope = ["openid"]`, `token_endpoint_auth_method = "client_secret_basic"`).
 
+### Forward-auth for reverse proxies
+
+Porchlight exposes `/forward-auth` for proxies such as Traefik, Caddy, and
+nginx `auth_request`. Authenticated active users receive `204 No Content` with
+identity headers (`Remote-User`, `Remote-Email`, `Remote-Groups`, and
+`X-Forwarded-*` equivalents). Anonymous users receive a `303` redirect to the
+Porchlight login page.
+
+To redirect users back to a protected app after login, allow that app's host:
+
+```toml
+forward_auth_allowed_redirect_hosts = ["app.example.com", "*.apps.example.com"]
+```
+
+The endpoint uses `X-Forwarded-Proto`, `X-Forwarded-Host`, and
+`X-Forwarded-Uri` (or `X-Original-Url`) to reconstruct the original request URL.
+Bare host patterns are HTTPS-only and match the default HTTPS port; `host:port`
+patterns are also HTTPS-only unless you include an explicit scheme such as
+`http://localhost:9000` for local cleartext deployments. Exact hosts,
+`host:port` entries, scheme-qualified hosts, and explicit `*.example.com`
+wildcard suffixes listed in `forward_auth_allowed_redirect_hosts` are accepted
+as return targets.
+
+Configure the reverse proxy to strip any inbound `Remote-*` and `X-Forwarded-*`
+identity headers from client requests before applying the headers returned by
+`/forward-auth`.
+
 To use a config file at a different path:
 
 ```bash

porchlight.dev.toml

@@ -11,6 +11,7 @@ issuer = "https://auth.example.com"
 # sqlite_path = "data/oidc_op.db"
 # signing_key_path = "data/keys"
 # invite_ttl = 86400
+# forward_auth_allowed_redirect_hosts = ["app.example.com", "*.apps.example.com"]
 
 # Register OIDC Relying Party clients below.
 # Each [clients.<client-id>] section defines one client.

porchlight.example.toml

@@ -11,6 +11,7 @@ issuer = "https://auth.example.com"
 # sqlite_path = "data/oidc_op.db"
 # signing_key_path = "data/keys"
 # invite_ttl = 86400
+# forward_auth_allowed_redirect_hosts = ["app.example.com", "*.apps.example.com"]
 
 # Register OIDC Relying Party clients below.
 # Each [clients.<client-id>] section defines one client.

src/porchlight/app.py

@@ -21,6 +21,7 @@ from porchlight.authn.routes import router as authn_router
 from porchlight.authn.webauthn import WebAuthnService
 from porchlight.config import Settings, StorageBackend
 from porchlight.csrf import CSRFMiddleware, generate_csrf_token
+from porchlight.forward_auth import router as forward_auth_router
 from porchlight.invite.service import MagicLinkService
 from porchlight.manage.routes import router as manage_router
 from porchlight.oidc.endpoints import router as oidc_router
@@ -143,7 +144,7 @@ def create_app(settings: Settings | None = None) -> FastAPI:
     app.state.session_secret = session_secret
     app.add_middleware(
         CSRFMiddleware,  # ty: ignore[invalid-argument-type]
-        exempt_paths={"/token", "/userinfo"},
+        exempt_paths={"/token", "/userinfo", "/forward-auth"},
         check_origin=settings.issuer,
     )
     app.add_middleware(
@@ -181,6 +182,7 @@ def create_app(settings: Settings | None = None) -> FastAPI:
     # Routers
     app.include_router(admin_router)
     app.include_router(authn_router)
+    app.include_router(forward_auth_router)
     app.include_router(manage_router)
     app.include_router(oidc_router)
 

src/porchlight/authn/routes.py

@@ -6,6 +6,12 @@ from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
 from fido2.webauthn import AttestedCredentialData, AuthenticationResponse
 
 from porchlight.authn.acr import ACR_PASSWORD, ACR_WEBAUTHN, SESSION_ACR_KEY
+from porchlight.forward_auth import (
+    FORWARD_AUTH_RETURN_TO_SESSION_KEY,
+    capture_forward_auth_return_to,
+    clear_forward_auth_return_to,
+    pop_forward_auth_return_to,
+)
 from porchlight.models import User
 from porchlight.rate_limit import limiter
 from porchlight.userid import generate_unique_userid
@@ -20,7 +26,11 @@ def _login_redirect_target(request: Request) -> str:
     Otherwise, redirect to credential management.
     """
     if "oidc_auth_request" in request.session:
+        clear_forward_auth_return_to(request)
         return "/authorization/complete"
+    forward_auth_return_to = pop_forward_auth_return_to(request)
+    if forward_auth_return_to is not None:
+        return forward_auth_return_to
     return "/manage/credentials"
 
 
@@ -48,9 +58,12 @@ def _establish_authenticated_session(request: Request, user: User, acr: str) ->
     surface a truthful Authentication Context Class Reference (see acr.py).
     """
     pending_oidc = request.session.get("oidc_auth_request")
+    pending_forward_auth = request.session.get(FORWARD_AUTH_RETURN_TO_SESSION_KEY)
     request.session.clear()
     if pending_oidc is not None:
         request.session["oidc_auth_request"] = pending_oidc
+    elif pending_forward_auth is not None:
+        request.session[FORWARD_AUTH_RETURN_TO_SESSION_KEY] = pending_forward_auth
     request.session["userid"] = user.userid
     request.session["username"] = user.username
     request.session[SESSION_ACR_KEY] = acr
@@ -58,6 +71,7 @@ def _establish_authenticated_session(request: Request, user: User, acr: str) ->
 
 @router.get("/login", response_class=HTMLResponse)
 async def login_page(request: Request) -> HTMLResponse:
+    capture_forward_auth_return_to(request)
     templates = request.app.state.templates
     return templates.TemplateResponse(request, "login.html")
 
@@ -160,6 +174,7 @@ async def register_magic_link(request: Request, token: str) -> Response:
     # Magic-link registration is single-factor (email possession); mark it as
     # such. It normally redirects to credential setup rather than completing an
     # OIDC flow, but the session acr governs any later authorization too.
+    clear_forward_auth_return_to(request)
     _establish_authenticated_session(request, user, ACR_PASSWORD)
 
     return RedirectResponse("/manage/credentials?setup=1", status_code=303)

src/porchlight/config.py

@@ -69,6 +69,13 @@ class Settings(BaseSettings):
     # many hops from the right. Keep 0 unless deployed behind a known proxy.
     trusted_proxy_count: int = 0
 
+    # Reverse-proxy forward-auth. Hosts listed here are eligible post-login
+    # redirect targets when /forward-auth receives X-Forwarded-* request
+    # metadata. Bare hosts and host:port entries are HTTPS-only; include an
+    # explicit scheme such as "http://localhost:9000" for cleartext targets.
+    # Supports exact hosts and explicit "*.example.com" wildcard suffixes.
+    forward_auth_allowed_redirect_hosts: list[str] = []
+
     # Signing keys
     signing_key_path: str = "data/keys"
 

src/porchlight/forward_auth.py

@@ -0,0 +1,340 @@
+import ipaddress
+import re
+from dataclasses import dataclass
+from urllib.parse import SplitResult, quote, urlsplit, urlunsplit
+
+from fastapi import APIRouter, Request
+from fastapi.responses import RedirectResponse
+from starlette.responses import Response
+
+from porchlight.config import Settings
+from porchlight.dependencies import get_session_user
+from porchlight.models import User
+
+FORWARD_AUTH_RETURN_TO_SESSION_KEY = "forward_auth_return_to"
+
+router = APIRouter(tags=["forward-auth"])
+
+_FORWARD_AUTH_METHODS = ("GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS")
+_ORIGINAL_URL_HEADERS = ("x-original-url", "x-forwarded-url")
+_ORIGINAL_URI_HEADERS = ("x-forwarded-uri", "x-original-uri")
+_DNS_LABEL_RE = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$")
+_DEFAULT_PORTS = {"http": 80, "https": 443}
+_HEADER_MIN_VISIBLE = 0x20
+_HOST_MIN_VISIBLE = 0x21
+_ASCII_DELETE = 0x7F
+_C1_CONTROL_MAX = 0x9F
+_MAX_DNS_HOSTNAME_LENGTH = 253
+
+
+@dataclass(frozen=True)
+class _HostPattern:
+    host: str
+    port: int
+    scheme: str
+    wildcard: bool = False
+
+
+@dataclass(frozen=True)
+class _ReturnTarget:
+    host: str
+    port: int
+    scheme: str
+
+
+@router.api_route("/forward-auth", methods=list(_FORWARD_AUTH_METHODS))
+async def forward_auth(request: Request) -> Response:
+    """Reverse-proxy forward-auth endpoint.
+
+    Authenticated, active users receive a 204 with identity headers. Anonymous
+    users are redirected to the login page, optionally with a validated
+    post-login return URL reconstructed from proxy headers.
+    """
+    session_user = get_session_user(request)
+    if session_user is None:
+        return _login_redirect_response(request)
+
+    userid, _username = session_user
+    user = await request.app.state.user_repo.get_by_userid(userid)
+    if user is None or not user.active:
+        request.session.clear()
+        return _login_redirect_response(request)
+
+    response = Response(status_code=204, headers=_identity_headers(user))
+    response.headers["Cache-Control"] = "no-store"
+    return response
+
+
+def capture_forward_auth_return_to(request: Request) -> None:
+    """Store a validated forward-auth return URL from ``/login`` query params."""
+    return_to = request.query_params.get("return_to")
+    if return_to is None:
+        return
+
+    settings: Settings = request.app.state.settings
+    if is_forward_auth_return_to_allowed(settings, return_to):
+        request.session[FORWARD_AUTH_RETURN_TO_SESSION_KEY] = return_to
+    else:
+        request.session.pop(FORWARD_AUTH_RETURN_TO_SESSION_KEY, None)
+
+
+def clear_forward_auth_return_to(request: Request) -> None:
+    """Remove any stored forward-auth return URL from the current session."""
+    request.session.pop(FORWARD_AUTH_RETURN_TO_SESSION_KEY, None)
+
+
+def pop_forward_auth_return_to(request: Request) -> str | None:
+    """Consume the stored forward-auth return URL if it is still allowed."""
+    return_to = request.session.pop(FORWARD_AUTH_RETURN_TO_SESSION_KEY, None)
+    if not isinstance(return_to, str):
+        return None
+
+    settings: Settings = request.app.state.settings
+    if not is_forward_auth_return_to_allowed(settings, return_to):
+        return None
+    return return_to
+
+
+def is_forward_auth_return_to_allowed(settings: Settings, return_to: str) -> bool:
+    """Return true when ``return_to`` is an HTTP(S) URL on an allowed host."""
+    target = _parse_return_target(return_to)
+    if target is None:
+        return False
+
+    return any(
+        _host_pattern_matches(pattern, target.host, target.port, target.scheme)
+        for pattern in settings.forward_auth_allowed_redirect_hosts
+    )
+
+
+def _parse_return_target(return_to: str) -> _ReturnTarget | None:
+    try:
+        parsed = urlsplit(return_to)
+    except ValueError:
+        return None
+
+    return_port = _return_target_port(parsed)
+    if return_port is None or parsed.hostname is None:
+        return None
+
+    hostname = _normalize_url_hostname(parsed.hostname)
+    if hostname is None:
+        return None
+    return _ReturnTarget(host=hostname, port=return_port, scheme=parsed.scheme)
+
+
+def _return_target_port(parsed: SplitResult) -> int | None:
+    if (
+        parsed.scheme not in ("http", "https")
+        or parsed.hostname is None
+        or parsed.username is not None
+        or parsed.password is not None
+        or _has_unsafe_url_host_chars(parsed.netloc)
+    ):
+        return None
+    try:
+        return _effective_port(parsed.scheme, parsed.port)
+    except ValueError:
+        return None
+
+
+def _login_redirect_response(request: Request) -> RedirectResponse:
+    target = _forwarded_return_to(request)
+    response = RedirectResponse(_login_url(request, target), status_code=303)
+    response.headers["Cache-Control"] = "no-store"
+    return response
+
+
+def _login_url(request: Request, return_to: str | None) -> str:
+    settings: Settings = request.app.state.settings
+    url = f"{settings.issuer.rstrip('/')}/login"
+    if return_to is None:
+        return url
+    return f"{url}?return_to={quote(return_to, safe='')}"
+
+
+def _forwarded_return_to(request: Request) -> str | None:
+    settings: Settings = request.app.state.settings
+    absolute = _first_header_value(request, _ORIGINAL_URL_HEADERS)
+    if absolute is not None and is_forward_auth_return_to_allowed(settings, absolute):
+        return absolute
+
+    proto = _first_header_value(request, ("x-forwarded-proto",))
+    host = _first_header_value(request, ("x-forwarded-host",))
+    uri = _first_header_value(request, _ORIGINAL_URI_HEADERS)
+    if proto is None or host is None or uri is None:
+        return None
+
+    proto = proto.lower()
+    if proto not in ("http", "https") or not uri.startswith("/"):
+        return None
+
+    candidate = urlunsplit((proto, host, uri, "", ""))
+    if not is_forward_auth_return_to_allowed(settings, candidate):
+        return None
+    return candidate
+
+
+def _first_header_value(request: Request, names: tuple[str, ...]) -> str | None:
+    for name in names:
+        raw = request.headers.get(name)
+        if raw is None:
+            continue
+        first = raw.split(",", 1)[0].strip()
+        if first and "\r" not in first and "\n" not in first:
+            return first
+    return None
+
+
+def _identity_headers(user: User) -> dict[str, str]:
+    display_name = user.preferred_username or user.username
+    groups = ",".join(sorted(user.groups))
+    headers = {
+        "Remote-User": user.username,
+        "Remote-Email": user.email or "",
+        "Remote-Groups": groups,
+        "X-Forwarded-User": user.username,
+        "X-Forwarded-Userid": user.userid,
+        "X-Forwarded-Preferred-Username": display_name,
+        "X-Forwarded-Email": user.email or "",
+        "X-Forwarded-Groups": groups,
+    }
+
+    return {name: _safe_header_value(value) for name, value in headers.items()}
+
+
+def _safe_header_value(value: str) -> str:
+    try:
+        value.encode("latin-1")
+    except UnicodeEncodeError:
+        return ""
+    if any(_is_http_header_control(ch) for ch in value):
+        return ""
+    return value
+
+
+def _is_http_header_control(ch: str) -> bool:
+    codepoint = ord(ch)
+    return (codepoint < _HEADER_MIN_VISIBLE and ch != "\t") or _ASCII_DELETE <= codepoint <= _C1_CONTROL_MAX
+
+
+def _host_pattern_matches(pattern: str, hostname: str, port: int, scheme: str) -> bool:
+    parsed = _parse_host_pattern(pattern)
+    if parsed is None:
+        return False
+    if parsed.scheme != scheme or parsed.port != port:
+        return False
+
+    if parsed.wildcard:
+        return hostname.endswith(f".{parsed.host}")
+    return hostname == parsed.host
+
+
+def _parse_host_pattern(pattern: str) -> _HostPattern | None:
+    pattern = pattern.strip().lower().rstrip(".")
+    if not pattern:
+        return None
+
+    if "://" in pattern:
+        return _parse_scheme_qualified_host_pattern(pattern)
+    return _parse_host_port_pattern(pattern, default_scheme="https")
+
+
+def _parse_scheme_qualified_host_pattern(pattern: str) -> _HostPattern | None:
+    try:
+        parsed = urlsplit(pattern)
+        if parsed.scheme not in ("http", "https"):
+            return None
+        port = _effective_port(parsed.scheme, parsed.port)
+    except ValueError:
+        return None
+
+    if (
+        parsed.hostname is None
+        or parsed.username is not None
+        or parsed.password is not None
+        or parsed.path not in ("", "/")
+        or parsed.query
+        or parsed.fragment
+        or _has_unsafe_url_host_chars(parsed.netloc)
+    ):
+        return None
+    return _build_host_pattern(parsed.hostname, port=port, scheme=parsed.scheme)
+
+
+def _parse_host_port_pattern(pattern: str, default_scheme: str) -> _HostPattern | None:
+    try:
+        parsed = urlsplit(f"//{pattern}")
+        port = _effective_port(default_scheme, parsed.port)
+    except ValueError:
+        return None
+
+    if (
+        parsed.hostname is None
+        or parsed.username is not None
+        or parsed.password is not None
+        or parsed.path
+        or parsed.query
+        or parsed.fragment
+        or _has_unsafe_url_host_chars(parsed.netloc)
+    ):
+        return None
+    return _build_host_pattern(parsed.hostname, port=port, scheme=default_scheme)
+
+
+def _build_host_pattern(hostname: str, *, port: int, scheme: str) -> _HostPattern | None:
+    raw_hostname = hostname.lower().rstrip(".")
+    wildcard = raw_hostname.startswith("*.")
+    if wildcard:
+        normalized = _normalize_dns_hostname(raw_hostname.removeprefix("*."))
+        if normalized is None:
+            return None
+    else:
+        normalized = _normalize_url_hostname(raw_hostname)
+        if normalized is None:
+            return None
+    return _HostPattern(host=normalized, port=port, scheme=scheme, wildcard=wildcard)
+
+
+def _effective_port(scheme: str, port: int | None) -> int:
+    if port is not None:
+        return port
+    return _DEFAULT_PORTS[scheme]
+
+
+def _normalize_url_hostname(hostname: str) -> str | None:
+    hostname = hostname.lower().rstrip(".")
+    if _has_unsafe_hostname_chars(hostname):
+        return None
+
+    try:
+        ipaddress.ip_address(hostname)
+    except ValueError:
+        return _normalize_dns_hostname(hostname)
+    return hostname
+
+
+def _normalize_dns_hostname(hostname: str) -> str | None:
+    if _has_unsafe_hostname_chars(hostname):
+        return None
+    try:
+        hostname.encode("ascii")
+    except UnicodeEncodeError:
+        return None
+
+    labels = hostname.split(".")
+    if not labels or any(not _DNS_LABEL_RE.fullmatch(label) for label in labels):
+        return None
+    if len(hostname) > _MAX_DNS_HOSTNAME_LENGTH:
+        return None
+    return hostname
+
+
+def _has_unsafe_url_host_chars(value: str) -> bool:
+    return "\\" in value or any(ord(ch) < _HOST_MIN_VISIBLE or ord(ch) == _ASCII_DELETE for ch in value)
+
+
+def _has_unsafe_hostname_chars(value: str) -> bool:
+    if "\\" in value or "%" in value or "*" in value:
+        return True
+    return any(ord(ch) < _HOST_MIN_VISIBLE or ord(ch) == _ASCII_DELETE for ch in value)

tests/test_config.py

@@ -15,6 +15,7 @@ def test_default_settings() -> None:
     assert settings.sqlite_path == "data/oidc_op.db"
     assert settings.manage_client_id == "manage-app"
     assert settings.invite_ttl == 86400
+    assert settings.forward_auth_allowed_redirect_hosts == []
     assert settings.theme == "default"
 
 
@@ -49,6 +50,7 @@ def test_settings_from_toml_file(tmp_path: Path) -> None:
 issuer = "https://toml.example.com"
 debug = true
 sqlite_path = "custom/path.db"
+forward_auth_allowed_redirect_hosts = ["app.example.com", "*.apps.example.com"]
 
 [clients.my-app]
 client_secret = "secret123"
@@ -62,6 +64,7 @@ scope = ["openid", "profile"]
     assert settings.issuer == "https://toml.example.com"
     assert settings.debug is True
     assert settings.sqlite_path == "custom/path.db"
+    assert settings.forward_auth_allowed_redirect_hosts == ["app.example.com", "*.apps.example.com"]
     assert "my-app" in settings.clients
     assert settings.clients["my-app"].client_secret == "secret123"
     assert settings.clients["my-app"].redirect_uris == ["https://app.example.com/callback"]

tests/test_forward_auth.py

@@ -0,0 +1,313 @@
+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