feat: wire auth middleware into FastAPI app

- Add _resolve_auth() with PAM → MUXPLEX_PASSWORD env → password file → auto-generate fallback chain
- MUXPLEX_AUTH=password forces password mode regardless of PAM availability
- MUXPLEX_SESSION_TTL controls session cookie TTL (default 604800 = 7 days)
- app.add_middleware(AuthMiddleware, ...) called after app creation
- Update client fixture to set a valid session cookie for test compatibility
- Add test_non_localhost_without_auth_gets_redirected integration test

Co-authored-by: Amplifier <amplifier@amplified.dev>
This commit is contained in:
Brian Krabach
2026-03-28 21:47:01 -07:00
parent 4e590345c3
commit 88b21832f7
2 changed files with 104 additions and 2 deletions
+72
View File
@@ -12,6 +12,8 @@ import contextlib
import logging
import os
import pathlib
import pwd
import sys
import time
from typing import Literal
@@ -23,6 +25,13 @@ from fastapi import FastAPI, HTTPException, WebSocket
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel
from muxplex.auth import (
AuthMiddleware,
generate_and_save_password,
load_or_create_secret,
load_password,
pam_available,
)
from muxplex.bells import apply_bell_clear_rule, process_bell_flags
from muxplex.sessions import (
enumerate_sessions,
@@ -173,6 +182,69 @@ async def lifespan(app: FastAPI):
app = FastAPI(title="muxplex", version="0.1.0", lifespan=lifespan)
# ---------------------------------------------------------------------------
# Auth setup
# ---------------------------------------------------------------------------
def _resolve_auth() -> tuple[str, str]:
"""Determine auth mode and resolve password. Returns (auth_mode, password).
Fallback chain for non-localhost:
1. PAM available → ("pam", "")
2. MUXPLEX_PASSWORD env → ("password", <env value>)
3. ~/.config/muxplex/password file → ("password", <file value>)
4. Auto-generate → ("password", <generated>)
"""
# Explicit override: MUXPLEX_AUTH=password forces password mode
force_password = os.environ.get("MUXPLEX_AUTH", "").lower() == "password"
if not force_password and pam_available():
running_user = pwd.getpwuid(os.getuid()).pw_name
print(f" muxplex auth: PAM (user: {running_user})", file=sys.stderr)
return "pam", ""
if not force_password:
print(" muxplex auth: PAM unavailable, using password mode", file=sys.stderr)
# Password mode — resolve password
env_pw = os.environ.get("MUXPLEX_PASSWORD")
if env_pw:
print(" muxplex auth: password (env)", file=sys.stderr)
return "password", env_pw
file_pw = load_password()
if file_pw:
print(
f" muxplex auth: password (file: {load_password.__module__})",
file=sys.stderr,
)
return "password", file_pw
# Last resort: auto-generate
generated = generate_and_save_password()
from muxplex.auth import get_password_path
print(
f" muxplex auth: password generated — {generated} — saved to {get_password_path()}",
file=sys.stderr,
)
return "password", generated
_auth_mode, _auth_password = _resolve_auth()
_auth_secret = load_or_create_secret()
_auth_ttl = int(os.environ.get("MUXPLEX_SESSION_TTL", "604800"))
app.add_middleware(
AuthMiddleware,
auth_mode=_auth_mode,
secret=_auth_secret,
ttl_seconds=_auth_ttl,
password=_auth_password,
)
# ---------------------------------------------------------------------------
# Request / response models
# ---------------------------------------------------------------------------
+32 -2
View File
@@ -47,9 +47,19 @@ def patch_startup_and_state(tmp_path, monkeypatch):
@pytest.fixture
def client():
"""Return a TestClient that triggers the app lifespan on entry/exit."""
def client(monkeypatch):
"""Return a TestClient that triggers the app lifespan on entry/exit.
Sets a valid session cookie so existing tests bypass the AuthMiddleware
(TestClient uses host='testclient', which is not a localhost address).
"""
monkeypatch.setenv("MUXPLEX_PASSWORD", "test-password")
with TestClient(app) as c:
from muxplex.auth import create_session_cookie
from muxplex.main import _auth_secret, _auth_ttl
cookie = create_session_cookie(_auth_secret, _auth_ttl)
c.cookies.set("muxplex_session", cookie)
yield c
@@ -653,3 +663,23 @@ def test_terminal_ws_route_exists():
if isinstance(r, (APIRoute, APIWebSocketRoute)) and r.path == "/terminal/ws"
]
assert len(ws_routes) == 1, "Expected exactly one /terminal/ws route"
# ---------------------------------------------------------------------------
# Auth middleware integration
# ---------------------------------------------------------------------------
def test_non_localhost_without_auth_gets_redirected(monkeypatch):
"""A non-localhost request without credentials is redirected to /login."""
from fastapi.testclient import TestClient
from muxplex.main import app
# Ensure auth is active — set a known password via env
monkeypatch.setenv("MUXPLEX_PASSWORD", "test-pw-for-api")
with TestClient(app, base_url="http://192.168.1.1") as c:
response = c.get("/health", follow_redirects=False)
# Should be redirected to /login or get 307/401
assert response.status_code in (307, 401)