diff --git a/muxplex/auth.py b/muxplex/auth.py index 4d5a7ba..a7535e5 100644 --- a/muxplex/auth.py +++ b/muxplex/auth.py @@ -136,6 +136,9 @@ def authenticate_pam(username: str, password: str) -> bool: # Paths that bypass auth (login page itself, static assets it needs) _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): """FastAPI middleware that enforces authentication on non-localhost requests.""" @@ -155,13 +158,10 @@ class AuthMiddleware(BaseHTTPMiddleware): self.password = password async def dispatch(self, request: Request, call_next) -> Response: - # 1. Localhost bypass — check both client IP and server hostname. - # In production, client.host is the connecting IP (127.0.0.1 for local). - # In tests (TestClient), base_url affects request.url.hostname instead. + # 1. Localhost bypass — client.host is the socket-level IP and cannot + # be forged by the client (unlike the HTTP Host header). client_host = request.client.host if request.client else "" - server_host = request.url.hostname or "" - _localhost_addrs = ("127.0.0.1", "::1", "localhost") - if client_host in _localhost_addrs or server_host in _localhost_addrs: + if client_host in _LOCALHOST_ADDRS: return await call_next(request) # 2. Exempt paths (login page, auth endpoints) @@ -177,6 +177,7 @@ class AuthMiddleware(BaseHTTPMiddleware): auth_header = request.headers.get("authorization", "") if auth_header.lower().startswith("basic "): try: + # Strip "Basic " prefix (6 chars) before base64-decoding decoded = base64.b64decode(auth_header[6:]).decode() username, _, pw = decoded.partition(":") if self._check_credentials(username, pw): diff --git a/muxplex/tests/test_auth.py b/muxplex/tests/test_auth.py index 895d983..8c6ceec 100644 --- a/muxplex/tests/test_auth.py +++ b/muxplex/tests/test_auth.py @@ -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: """Create a minimal FastAPI app with AuthMiddleware for testing.""" 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(): """Requests from 127.0.0.1 pass through without auth.""" 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") assert response.status_code == 200 assert response.text == "OK" @@ -280,7 +303,8 @@ def test_middleware_valid_session_cookie_passes(): app = _make_test_app() cookie = create_session_cookie("test-secret", ttl_seconds=3600) 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.text == "OK" @@ -289,7 +313,8 @@ def test_middleware_tampered_cookie_redirects(): """Non-localhost request with a tampered cookie redirects to /login.""" app = _make_test_app() 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 "/login" in response.headers["location"]