diff --git a/README.md b/README.md index d5cbe44..980e1b7 100644 --- a/README.md +++ b/README.md @@ -96,7 +96,6 @@ 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. @@ -124,33 +123,6 @@ 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 diff --git a/porchlight.dev.toml b/porchlight.dev.toml index 65e9880..f5dd782 100644 --- a/porchlight.dev.toml +++ b/porchlight.dev.toml @@ -11,7 +11,6 @@ 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.] section defines one client. diff --git a/porchlight.example.toml b/porchlight.example.toml index a42d66e..801155b 100644 --- a/porchlight.example.toml +++ b/porchlight.example.toml @@ -11,7 +11,6 @@ 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.] section defines one client. diff --git a/src/porchlight/app.py b/src/porchlight/app.py index ef8d21e..272f9ce 100644 --- a/src/porchlight/app.py +++ b/src/porchlight/app.py @@ -21,7 +21,6 @@ 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 @@ -144,7 +143,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", "/forward-auth"}, + exempt_paths={"/token", "/userinfo"}, check_origin=settings.issuer, ) app.add_middleware( @@ -182,7 +181,6 @@ 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) diff --git a/src/porchlight/authn/routes.py b/src/porchlight/authn/routes.py index c01fb13..979cbc9 100644 --- a/src/porchlight/authn/routes.py +++ b/src/porchlight/authn/routes.py @@ -6,12 +6,6 @@ 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 @@ -26,11 +20,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: - return forward_auth_return_to return "/manage/credentials" @@ -58,12 +48,9 @@ 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 @@ -71,7 +58,6 @@ 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") @@ -174,7 +160,6 @@ 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) diff --git a/src/porchlight/config.py b/src/porchlight/config.py index 2ac30ac..20ff9ea 100644 --- a/src/porchlight/config.py +++ b/src/porchlight/config.py @@ -69,13 +69,6 @@ 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" diff --git a/src/porchlight/forward_auth.py b/src/porchlight/forward_auth.py deleted file mode 100644 index 07a9574..0000000 --- a/src/porchlight/forward_auth.py +++ /dev/null @@ -1,340 +0,0 @@ -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) diff --git a/tests/test_config.py b/tests/test_config.py index 9067af4..8a91127 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -15,7 +15,6 @@ 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" @@ -50,7 +49,6 @@ 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" @@ -64,7 +62,6 @@ 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"] diff --git a/tests/test_forward_auth.py b/tests/test_forward_auth.py deleted file mode 100644 index f085e24..0000000 --- a/tests/test_forward_auth.py +++ /dev/null @@ -1,313 +0,0 @@ -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