Harden forward-auth redirect and header handling

Tighten forward-auth return URL validation to reject unsafe host syntax,
including backslash-based browser/parser mismatches, and normalize
scheme/port matching so bare allowlist entries are HTTPS default-port
only. Support explicit scheme-qualified targets, including IPv6
host:port entries.

Always emit deterministic identity headers with empty values when
attributes are absent, zero unsafe header values, and clear stale
forward-auth return targets when OIDC or registration flows take
precedence.

Add regression coverage for wildcard redirect bypasses, scheme/port
handling, IPv6 allowlist entries, empty identity headers, unsafe header
values, and stale return-target cleanup. Update docs with proxy
header-stripping guidance and the new allowlist semantics.

README.md

@@ -140,9 +140,16 @@ 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.
-Only exact hosts, `host:port` entries, or explicit `*.example.com` wildcard
+Bare host patterns are HTTPS-only and match the default HTTPS port; `host:port`
-suffixes listed in `forward_auth_allowed_redirect_hosts` are accepted as return
+patterns are also HTTPS-only unless you include an explicit scheme such as
-targets.
+`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:
 

src/porchlight/authn/routes.py

@@ -9,6 +9,7 @@ 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
@@ -25,6 +26,7 @@ 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:
@@ -60,7 +62,7 @@ def _establish_authenticated_session(request: Request, user: User, acr: str) ->
     request.session.clear()
     if pending_oidc is not None:
         request.session["oidc_auth_request"] = pending_oidc
-    if pending_forward_auth is not None:
+    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
@@ -172,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

@@ -71,8 +71,9 @@ class Settings(BaseSettings):
 
     # Reverse-proxy forward-auth. Hosts listed here are eligible post-login
     # redirect targets when /forward-auth receives X-Forwarded-* request
-    # metadata. Supports exact hosts, host:port, and explicit "*.example.com"
+    # metadata. Bare hosts and host:port entries are HTTPS-only; include an
-    # wildcard suffixes.
+    # 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

src/porchlight/forward_auth.py

@@ -1,5 +1,7 @@
+import ipaddress
+import re
 from dataclasses import dataclass
-from urllib.parse import quote, urlsplit, urlunsplit
+from urllib.parse import SplitResult, quote, urlsplit, urlunsplit
 
 from fastapi import APIRouter, Request
 from fastapi.responses import RedirectResponse
@@ -16,13 +18,28 @@ 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 | None = None
+    port: int
-    scheme: str | None = None
+    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))
@@ -61,6 +78,11 @@ def capture_forward_auth_return_to(request: Request) -> None:
         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)
@@ -75,25 +97,45 @@ def pop_forward_auth_return_to(request: Request) -> str | None:
 
 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 False
+        return None
 
-    if parsed.scheme not in ("http", "https") or parsed.hostname is None:
+    return_port = _return_target_port(parsed)
-        return False
+    if return_port is None or parsed.hostname is None:
-    if parsed.username is not None or parsed.password is not None:
+        return None
-        return False
 
+    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_port = parsed.port
+        return _effective_port(parsed.scheme, parsed.port)
     except ValueError:
-        return False
+        return None
-
-    for pattern in settings.forward_auth_allowed_redirect_hosts:
-        if _host_pattern_matches(pattern, parsed.hostname, return_port, parsed.scheme):
-            return True
-    return False
 
 
 def _login_redirect_response(request: Request) -> RedirectResponse:
@@ -146,47 +188,45 @@ def _first_header_value(request: Request, names: tuple[str, ...]) -> str | 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,
     }
 
-    if user.email:
+    return {name: _safe_header_value(value) for name, value in headers.items()}
-        headers["Remote-Email"] = user.email
-        headers["X-Forwarded-Email"] = user.email
-    if user.groups:
-        groups = ",".join(sorted(user.groups))
-        headers["Remote-Groups"] = groups
-        headers["X-Forwarded-Groups"] = groups
-
-    return {name: value for name, value in headers.items() if _is_safe_header_value(value)}
 
 
-def _is_safe_header_value(value: str) -> bool:
+def _safe_header_value(value: str) -> str:
-    if "\r" in value or "\n" in value:
-        return False
     try:
         value.encode("latin-1")
     except UnicodeEncodeError:
-        return False
+        return ""
-    return True
+    if any(_is_http_header_control(ch) for ch in value):
+        return ""
+    return value
 
 
-def _host_pattern_matches(pattern: str, hostname: str, port: int | None, scheme: str) -> bool:
+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 is not None and parsed.scheme != scheme:
+    if parsed.scheme != scheme or parsed.port != port:
-        return False
-    if parsed.port is not None and parsed.port != port:
         return False
 
-    hostname = hostname.lower().rstrip(".")
+    if parsed.wildcard:
-    if parsed.host.startswith("*."):
+        return hostname.endswith(f".{parsed.host}")
-        suffix = parsed.host[1:]
-        return hostname.endswith(suffix) and hostname != suffix.removeprefix(".")
     return hostname == parsed.host
 
 
@@ -196,20 +236,105 @@ def _parse_host_pattern(pattern: str) -> _HostPattern | None:
         return None
 
     if "://" in pattern:
-        try:
+        return _parse_scheme_qualified_host_pattern(pattern)
-            parsed_pattern = urlsplit(pattern)
+    return _parse_host_port_pattern(pattern, default_scheme="https")
-            pattern_host = parsed_pattern.hostname
-            pattern_port = parsed_pattern.port
-        except ValueError:
-            return None
-        if parsed_pattern.scheme not in ("http", "https") or pattern_host is None:
-            return None
-        return _HostPattern(host=pattern_host.rstrip("."), port=pattern_port, scheme=parsed_pattern.scheme)
 
-    pattern_port = None
+
-    if pattern.count(":") == 1:
+def _parse_scheme_qualified_host_pattern(pattern: str) -> _HostPattern | None:
-        possible_host, possible_port = pattern.rsplit(":", 1)
+    try:
-        if possible_port.isdigit():
+        parsed = urlsplit(pattern)
-            pattern = possible_host.rstrip(".")
+        if parsed.scheme not in ("http", "https"):
-            pattern_port = int(possible_port)
+            return None
-    return _HostPattern(host=pattern, port=pattern_port)
+        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_forward_auth.py

@@ -1,11 +1,14 @@
 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
 
 
@@ -23,7 +26,7 @@ async def _create_user(
         username=username,
         preferred_username="Alice",
         email=email,
-        groups=groups or ["users"],
+        groups=groups if groups is not None else ["users"],
         created_at=datetime.now(UTC),
         updated_at=datetime.now(UTC),
     )
@@ -108,6 +111,81 @@ async def test_forward_auth_ignores_disallowed_return_host(client: AsyncClient)
     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)
 
@@ -153,6 +231,21 @@ async def test_login_ignores_disallowed_forward_auth_return_to(client: AsyncClie
     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)
@@ -171,6 +264,31 @@ async def test_forward_auth_authenticated_user_returns_identity_headers(client:
     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)
@@ -182,3 +300,14 @@ async def test_forward_auth_rejects_inactive_session_user(client: AsyncClient) -
 
     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