fix(auth): remove Host-header auth bypass; inject client IP in tests via ASGI middleware

The server_host check (request.url.hostname) read from the user-controlled
HTTP Host header, allowing any attacker to bypass authentication by sending
'Host: 127.0.0.1'. Remove it entirely — request.client.host is the
socket-level IP and cannot be forged.

Fix the localhost test by introducing _InjectClientMiddleware, a thin ASGI
wrapper that writes a real IP into the scope's 'client' field. This lets the
test exercise the actual localhost check with 127.0.0.1 rather than relying
on the URL hostname.

Also:
- Promote _LOCALHOST_ADDRS to module-level constant (no-op per request)
- Add clarifying comment on auth_header[6:] magic slice
- Replace per-request cookies= kwarg with client.cookies.set() (Starlette deprecation)
This commit is contained in:
Brian Krabach
2026-03-28 21:40:22 -07:00
parent 22be933ebf
commit 4e590345c3
2 changed files with 35 additions and 9 deletions
+7 -6
View File
@@ -136,6 +136,9 @@ def authenticate_pam(username: str, password: str) -> bool:
# Paths that bypass auth (login page itself, static assets it needs) # Paths that bypass auth (login page itself, static assets it needs)
_AUTH_EXEMPT_PATHS = {"/login", "/auth/mode", "/auth/logout"} _AUTH_EXEMPT_PATHS = {"/login", "/auth/mode", "/auth/logout"}
# Socket-level localhost addresses — cannot be forged via HTTP headers
_LOCALHOST_ADDRS = {"127.0.0.1", "::1"}
class AuthMiddleware(BaseHTTPMiddleware): class AuthMiddleware(BaseHTTPMiddleware):
"""FastAPI middleware that enforces authentication on non-localhost requests.""" """FastAPI middleware that enforces authentication on non-localhost requests."""
@@ -155,13 +158,10 @@ class AuthMiddleware(BaseHTTPMiddleware):
self.password = password self.password = password
async def dispatch(self, request: Request, call_next) -> Response: async def dispatch(self, request: Request, call_next) -> Response:
# 1. Localhost bypass — check both client IP and server hostname. # 1. Localhost bypass — client.host is the socket-level IP and cannot
# In production, client.host is the connecting IP (127.0.0.1 for local). # be forged by the client (unlike the HTTP Host header).
# In tests (TestClient), base_url affects request.url.hostname instead.
client_host = request.client.host if request.client else "" client_host = request.client.host if request.client else ""
server_host = request.url.hostname or "" if client_host in _LOCALHOST_ADDRS:
_localhost_addrs = ("127.0.0.1", "::1", "localhost")
if client_host in _localhost_addrs or server_host in _localhost_addrs:
return await call_next(request) return await call_next(request)
# 2. Exempt paths (login page, auth endpoints) # 2. Exempt paths (login page, auth endpoints)
@@ -177,6 +177,7 @@ class AuthMiddleware(BaseHTTPMiddleware):
auth_header = request.headers.get("authorization", "") auth_header = request.headers.get("authorization", "")
if auth_header.lower().startswith("basic "): if auth_header.lower().startswith("basic "):
try: try:
# Strip "Basic " prefix (6 chars) before base64-decoding
decoded = base64.b64decode(auth_header[6:]).decode() decoded = base64.b64decode(auth_header[6:]).decode()
username, _, pw = decoded.partition(":") username, _, pw = decoded.partition(":")
if self._check_credentials(username, pw): if self._check_credentials(username, pw):
+28 -3
View File
@@ -247,6 +247,25 @@ def test_authenticate_pam_wrong_user_rejected(monkeypatch):
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
class _InjectClientMiddleware:
"""Thin ASGI wrapper that injects a fake client address into the scope.
Starlette's TestClient sets request.client.host to "testclient" regardless
of base_url. Wrapping the app with this middleware lets tests supply a
specific socket-level IP so the AuthMiddleware localhost check is exercised
with real values rather than relying on the (user-controlled) Host header.
"""
def __init__(self, app, client_host: str, client_port: int = 50000):
self.app = app
self._client = (client_host, client_port)
async def __call__(self, scope, receive, send):
if scope.get("type") == "http":
scope = {**scope, "client": self._client}
await self.app(scope, receive, send)
def _make_test_app(auth_mode: str = "password", password: str = "test-pw") -> FastAPI: def _make_test_app(auth_mode: str = "password", password: str = "test-pw") -> FastAPI:
"""Create a minimal FastAPI app with AuthMiddleware for testing.""" """Create a minimal FastAPI app with AuthMiddleware for testing."""
test_app = FastAPI() test_app = FastAPI()
@@ -269,7 +288,11 @@ def _make_test_app(auth_mode: str = "password", password: str = "test-pw") -> Fa
def test_middleware_localhost_bypasses_auth(): def test_middleware_localhost_bypasses_auth():
"""Requests from 127.0.0.1 pass through without auth.""" """Requests from 127.0.0.1 pass through without auth."""
app = _make_test_app() app = _make_test_app()
client = TestClient(app, base_url="http://127.0.0.1") # TestClient always sets request.client.host to "testclient". Wrap the app
# with _InjectClientMiddleware to set the socket-level client IP to 127.0.0.1
# so the middleware's localhost check is exercised with a real address.
app_with_client = _InjectClientMiddleware(app, "127.0.0.1")
client = TestClient(app_with_client)
response = client.get("/protected") response = client.get("/protected")
assert response.status_code == 200 assert response.status_code == 200
assert response.text == "OK" assert response.text == "OK"
@@ -280,7 +303,8 @@ def test_middleware_valid_session_cookie_passes():
app = _make_test_app() app = _make_test_app()
cookie = create_session_cookie("test-secret", ttl_seconds=3600) cookie = create_session_cookie("test-secret", ttl_seconds=3600)
client = TestClient(app, base_url="http://192.168.1.1") client = TestClient(app, base_url="http://192.168.1.1")
response = client.get("/protected", cookies={"muxplex_session": cookie}) client.cookies.set("muxplex_session", cookie)
response = client.get("/protected")
assert response.status_code == 200 assert response.status_code == 200
assert response.text == "OK" assert response.text == "OK"
@@ -289,7 +313,8 @@ def test_middleware_tampered_cookie_redirects():
"""Non-localhost request with a tampered cookie redirects to /login.""" """Non-localhost request with a tampered cookie redirects to /login."""
app = _make_test_app() app = _make_test_app()
client = TestClient(app, base_url="http://192.168.1.1", follow_redirects=False) client = TestClient(app, base_url="http://192.168.1.1", follow_redirects=False)
response = client.get("/protected", cookies={"muxplex_session": "bad.cookie.value"}) client.cookies.set("muxplex_session", "bad.cookie.value")
response = client.get("/protected")
assert response.status_code == 307 assert response.status_code == 307
assert "/login" in response.headers["location"] assert "/login" in response.headers["location"]