From 22be933ebf32fd2f3f2bd72e698c6a489a5a2cb3 Mon Sep 17 00:00:00 2001 From: Brian Krabach Date: Sat, 28 Mar 2026 21:31:48 -0700 Subject: [PATCH] feat(auth): request middleware with localhost bypass and session cookie check - Add AuthMiddleware class extending BaseHTTPMiddleware - Localhost bypass: checks both client.host and url.hostname for 127.0.0.1/::1/localhost - Exempt paths: /login, /auth/mode, /auth/logout bypass auth - Valid session cookie (muxplex_session) passes through - Basic auth header: valid password passes, invalid returns 401 - API clients (Accept: application/json) get 401 on auth failure - Browser clients get 307 redirect to /login on auth failure - _check_credentials dispatches to PAM or password comparison - Add 8 middleware tests to test_auth.py Co-authored-by: Amplifier --- muxplex/auth.py | 73 +++++++++++++++++++++++++ muxplex/tests/test_auth.py | 108 +++++++++++++++++++++++++++++++++++++ 2 files changed, 181 insertions(+) diff --git a/muxplex/auth.py b/muxplex/auth.py index 13246f1..4d5a7ba 100644 --- a/muxplex/auth.py +++ b/muxplex/auth.py @@ -2,10 +2,14 @@ muxplex authentication — password and signing secret file management. """ +import base64 import secrets from pathlib import Path from itsdangerous import BadSignature, SignatureExpired, TimestampSigner +from starlette.middleware.base import BaseHTTPMiddleware +from starlette.requests import Request +from starlette.responses import JSONResponse, RedirectResponse, Response # --------------------------------------------------------------------------- @@ -123,3 +127,72 @@ def authenticate_pam(username: str, password: str) -> bool: if username != running_user: return False return pam.authenticate(username, password, service="login") + + +# --------------------------------------------------------------------------- +# Auth middleware +# --------------------------------------------------------------------------- + +# Paths that bypass auth (login page itself, static assets it needs) +_AUTH_EXEMPT_PATHS = {"/login", "/auth/mode", "/auth/logout"} + + +class AuthMiddleware(BaseHTTPMiddleware): + """FastAPI middleware that enforces authentication on non-localhost requests.""" + + def __init__( + self, + app, + auth_mode: str, + secret: str, + ttl_seconds: int, + password: str = "", + ): + super().__init__(app) + self.auth_mode = auth_mode + self.secret = secret + self.ttl_seconds = ttl_seconds + 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. + 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: + return await call_next(request) + + # 2. Exempt paths (login page, auth endpoints) + if request.url.path in _AUTH_EXEMPT_PATHS: + return await call_next(request) + + # 3. Valid session cookie + cookie = request.cookies.get("muxplex_session") + if cookie and verify_session_cookie(self.secret, cookie, self.ttl_seconds): + return await call_next(request) + + # 4. Authorization: Basic header + auth_header = request.headers.get("authorization", "") + if auth_header.lower().startswith("basic "): + try: + decoded = base64.b64decode(auth_header[6:]).decode() + username, _, pw = decoded.partition(":") + if self._check_credentials(username, pw): + return await call_next(request) + except Exception: + pass + return JSONResponse({"detail": "Invalid credentials"}, status_code=401) + + # 5. No auth — redirect browsers, 401 for API clients + accept = request.headers.get("accept", "") + if "application/json" in accept: + return JSONResponse({"detail": "Authentication required"}, status_code=401) + return RedirectResponse(url="/login", status_code=307) + + def _check_credentials(self, username: str, password: str) -> bool: + """Validate credentials against the configured auth mode.""" + if self.auth_mode == "pam": + return authenticate_pam(username, password) + return password == self.password diff --git a/muxplex/tests/test_auth.py b/muxplex/tests/test_auth.py index 31da6f3..895d983 100644 --- a/muxplex/tests/test_auth.py +++ b/muxplex/tests/test_auth.py @@ -1,10 +1,17 @@ """Tests for muxplex/auth.py — authentication module.""" +import base64 import os import pwd import stat from pathlib import Path +from fastapi import FastAPI +from fastapi.testclient import TestClient +from starlette.responses import PlainTextResponse + +from muxplex.auth import AuthMiddleware, create_session_cookie + # --------------------------------------------------------------------------- # Password file management @@ -233,3 +240,104 @@ def test_authenticate_pam_wrong_user_rejected(monkeypatch): # Mock PAM to always return True — but wrong username should still fail monkeypatch.setattr("pam.authenticate", lambda u, p, service="login": True) assert authenticate_pam(wrong_user, "any-password") is False + + +# --------------------------------------------------------------------------- +# Auth middleware +# --------------------------------------------------------------------------- + + +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() + + test_app.add_middleware( + AuthMiddleware, + auth_mode=auth_mode, + secret="test-secret", + ttl_seconds=3600, + password=password, + ) + + @test_app.get("/protected") + async def protected(): + return PlainTextResponse("OK") + + return test_app + + +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") + response = client.get("/protected") + assert response.status_code == 200 + assert response.text == "OK" + + +def test_middleware_valid_session_cookie_passes(): + """Non-localhost request with a valid session cookie passes through.""" + 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}) + assert response.status_code == 200 + assert response.text == "OK" + + +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"}) + assert response.status_code == 307 + assert "/login" in response.headers["location"] + + +def test_middleware_no_cookie_non_localhost_redirects(): + """Non-localhost request with no 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") + assert response.status_code == 307 + assert "/login" in response.headers["location"] + + +def test_middleware_basic_auth_valid_password(): + """Non-localhost request with valid Basic auth header passes through.""" + app = _make_test_app(auth_mode="password", password="test-pw") + client = TestClient(app, base_url="http://192.168.1.1") + creds = base64.b64encode(b":test-pw").decode() + response = client.get("/protected", headers={"Authorization": f"Basic {creds}"}) + assert response.status_code == 200 + assert response.text == "OK" + + +def test_middleware_basic_auth_invalid_password(): + """Non-localhost request with wrong Basic auth header returns 401.""" + app = _make_test_app(auth_mode="password", password="test-pw") + client = TestClient(app, base_url="http://192.168.1.1") + creds = base64.b64encode(b":wrong-pw").decode() + response = client.get("/protected", headers={"Authorization": f"Basic {creds}"}) + assert response.status_code == 401 + + +def test_middleware_json_request_gets_401_not_redirect(): + """Non-localhost API request (Accept: application/json) gets 401, not redirect.""" + app = _make_test_app() + client = TestClient(app, base_url="http://192.168.1.1", follow_redirects=False) + response = client.get("/protected", headers={"Accept": "application/json"}) + assert response.status_code == 401 + + +def test_middleware_login_path_excluded(): + """/login path is excluded from auth to avoid redirect loops.""" + app = _make_test_app() + + @app.get("/login") + async def login(): + return PlainTextResponse("login page") + + client = TestClient(app, base_url="http://192.168.1.1") + response = client.get("/login") + assert response.status_code == 200