lundberg/porchlight
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages 1 Wiki Activity Actions

Compare commits

base: lundberg:main
Branches Tags
lundberg:main
lundberg:v0.0.1
..
compare: lundberg:v0.0.1
Branches Tags
lundberg:main
lundberg:v0.0.1

No commits in common. "main" and "v0.0.1" have entirely different histories.
main ... v0.0.1

15 changed files with 29 additions and 767 deletions
Download patch file Download diff file Expand all files Collapse all files

25
Dockerfile
View file

@ -6,7 +6,7 @@ ENV PYTHONDONTWRITEBYTECODE=1 \
UV_COMPILE_BYTECODE=1 \
UV_LINK_MODE=copy
COPY --from=ghcr.io/astral-sh/uv:0.11.29 /uv /uvx /bin/
COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/
WORKDIR /app
@ -17,19 +17,17 @@ RUN uv sync --frozen --no-install-project --no-dev
# ---- Dev stage: hot-reload for local development ----
FROM base AS dev
# Also install dev dependencies. The project source is bind-mounted at runtime.
# Also install dev dependencies
RUN uv sync --frozen --no-install-project
# Source is bind-mounted at runtime via docker-compose
ENV OIDC_OP_ISSUER=http://localhost:8000 \
OIDC_OP_DEBUG=true \
OIDC_OP_SESSION_HTTPS_ONLY=false \
PATH="/app/.venv/bin:$PATH" \
PYTHONPATH=/app/src
OIDC_OP_SESSION_HTTPS_ONLY=false
EXPOSE 8000
CMD ["uvicorn", "porchlight.app:create_app", \
CMD ["uv", "run", "uvicorn", "porchlight.app:create_app", \
"--factory", "--host", "0.0.0.0", "--port", "8000", \
"--reload", "--reload-dir", "/app/src"]
@ -41,21 +39,10 @@ COPY README.md ./
COPY src/ src/
RUN uv sync --frozen --no-dev
# Use the venv built above directly — avoids `uv run` re-syncing (and
# needing network access to re-resolve/rebuild) at container startup.
ENV PATH="/app/.venv/bin:$PATH"
RUN groupadd --system porchlight \
&& useradd --system --gid porchlight --home-dir /app porchlight \
&& mkdir -p /app/data \
&& chown porchlight:porchlight /app/data \
&& mkdir -p /app/config \
&& chown porchlight:porchlight /app/config
USER porchlight
ENV OIDC_OP_ISSUER=http://localhost:8000
EXPOSE 8000
CMD ["uvicorn", "porchlight.app:create_app", \
CMD ["uv", "run", "uvicorn", "porchlight.app:create_app", \
"--factory", "--host", "0.0.0.0", "--port", "8000", \
"--workers", "4"]

28
README.md
View file

@ -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

4
docker-compose.yml
View file

@ -6,8 +6,8 @@ services:
ports:
- "8000:8000"
environment:
OIDC_OP_ISSUER: "${OIDC_OP_ISSUER:?set OIDC_OP_ISSUER to the public HTTPS URL}"
OIDC_OP_SESSION_SECRET: "${OIDC_OP_SESSION_SECRET:?set OIDC_OP_SESSION_SECRET to a random secret}"
OIDC_OP_ISSUER: "http://localhost:8000"
OIDC_OP_SESSION_SECRET: "change-me-in-production"
volumes:
- app-data:/app/data
profiles:

1
porchlight.dev.toml
View file

@ -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.<client-id>] section defines one client.

1
porchlight.example.toml
View file

@ -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.<client-id>] section defines one client.

4
src/porchlight/app.py
View file

@ -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)

15
src/porchlight/authn/routes.py
View file

@ -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)

7
src/porchlight/config.py
View file

@ -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"

340
src/porchlight/forward_auth.py
View file

@ -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)

BIN
src/porchlight/static/favicon.png
View file

Binary file not shown.
Side by side Swipe Overlay

Before

Width:  |  Height:  |  Size: 996 B

After

Width:  |  Height:  |  Size: 735 B

Before After
Before After

34
src/porchlight/static/logo.svg
View file

@ -1,27 +1,11 @@
<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 64 64" fill="none">
<title>Porchlight</title>
<style>
.structure { stroke: #292524; }
.door { fill: #e7e5e4; }
.glow { fill: #fef3c7; }
.lamp { fill: #d97706; }
@media (prefers-color-scheme: dark) {
.structure { stroke: #fafaf9; }
.door { fill: #44403c; }
.glow { fill: #78350f; }
.lamp { fill: #fbbf24; }
}
</style>
<!-- A simple porch: shelter, doorway, and a warm light left on. -->
<path class="structure" d="M8 25 32 7l24 18M13 25h38M16 25v28M48 25v28M10 53h44"
stroke-width="4" stroke-linecap="round" stroke-linejoin="round"/>
<path class="door" d="M22 52V38a10 10 0 0 1 20 0v14Z"/>
<path class="structure" d="M22 52V38a10 10 0 0 1 20 0v14"
stroke-width="3.5" stroke-linecap="round" stroke-linejoin="round"/>
<circle class="glow" cx="32" cy="22" r="8"/>
<path class="structure" d="M32 12v5" stroke-width="3" stroke-linecap="round"/>
<circle class="lamp" cx="32" cy="22" r="4"/>
<!-- Pentagon house shape with pointed roof -->
<path d="M32 6 L54 22 L50 46 H14 L10 22 Z" stroke="currentColor" stroke-width="3.5" stroke-linecap="round" stroke-linejoin="round" fill="none"/>
<!-- Light circle -->
<circle cx="32" cy="28" r="6.5" stroke="currentColor" stroke-width="3" fill="none"/>
<circle cx="32" cy="28" r="2" fill="#d97706"/>
<!-- Small dot accent (upper left) -->
<circle cx="19" cy="18" r="1.5" fill="currentColor"/>
<!-- Curved lower shield/visor -->
<path d="M16 44 Q24 58 32 58 Q40 58 48 44" stroke="currentColor" stroke-width="3.5" stroke-linecap="round" fill="none"/>
</svg>

Side by side Swipe Overlay

Before

Width:  |  Height:  |  Size: 1 KiB

After

Width:  |  Height:  |  Size: 694 B

Before After
Before After

3
tests/test_config.py
View file

@ -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"]

4
tests/test_csrf.py
View file

@ -91,7 +91,9 @@ class TestCSRFValidation:
app = _make_app()
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://testserver") as client:
token, _ = await _get_token_and_cookies(client)
response = await client.post("/echo", data={"csrf_token": token, "payload": "hello"})
response = await client.post(
"/echo", data={"csrf_token": token, "payload": "hello"}
)
assert response.status_code == 200
assert response.text == "hello"

313
tests/test_forward_auth.py
View file

@ -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

17
tests/test_oidc/test_e2e_flow.py
View file

@ -347,12 +347,16 @@ async def test_refresh_grant_does_not_mint_id_token(client: AsyncClient) -> None
assert "id_token" not in refreshed
async def _add_webauthn_credential(app: FastAPI, userid: str) -> tuple[ec.EllipticCurvePrivateKey, bytes]:
async def _add_webauthn_credential(
app: FastAPI, userid: str
) -> tuple[ec.EllipticCurvePrivateKey, bytes]:
"""Register a WebAuthn credential for an existing user; return (key, cred_id)."""
private_key = ec.generate_private_key(ec.SECP256R1())
cose_key = ES256.from_cryptography_key(private_key.public_key())
credential_id = secrets.token_bytes(32)
attested = AttestedCredentialData.create(aaguid=Aaguid.NONE, credential_id=credential_id, public_key=cose_key)
attested = AttestedCredentialData.create(
aaguid=Aaguid.NONE, credential_id=credential_id, public_key=cose_key
)
await app.state.credential_repo.create_webauthn(
WebAuthnCredential(user_id=userid, credential_id=credential_id, public_key=bytes(attested))
)
@ -360,14 +364,9 @@ async def _add_webauthn_credential(app: FastAPI, userid: str) -> tuple[ec.Ellipt
async def _login_webauthn_and_authorize(
client: AsyncClient,
userid: str,
credential: tuple[ec.EllipticCurvePrivateKey, bytes],
state: str,
nonce: str,
client: AsyncClient, userid: str, private_key: ec.EllipticCurvePrivateKey, credential_id: bytes, state: str, nonce: str
) -> str:
"""Authorize, log in via WebAuthn assertion, consent, return the auth code."""
private_key, credential_id = credential
auth_res = await client.get(
"/authorization",
params={
@ -433,6 +432,6 @@ async def test_webauthn_login_emits_mfa_acr(client: AsyncClient) -> None:
private_key, credential_id = await _add_webauthn_credential(app, "lusab-bansen")
state, nonce = secrets.token_urlsafe(16), secrets.token_urlsafe(16)
code = await _login_webauthn_and_authorize(client, "lusab-bansen", (private_key, credential_id), state, nonce)
code = await _login_webauthn_and_authorize(client, "lusab-bansen", private_key, credential_id, state, nonce)
token_data = await _exchange_token(client, code)
await _validate_id_token(client, token_data["id_token"], nonce, expected_acr=ACR_WEBAUTHN)