diff --git a/muxplex/main.py b/muxplex/main.py index 0d188e3..953cfc6 100644 --- a/muxplex/main.py +++ b/muxplex/main.py @@ -21,13 +21,16 @@ import websockets import websockets.exceptions from websockets.typing import Subprotocol -from fastapi import FastAPI, HTTPException, WebSocket +from fastapi import FastAPI, Form, HTTPException, Request, WebSocket from fastapi.responses import HTMLResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel +from starlette.responses import RedirectResponse from muxplex.auth import ( AuthMiddleware, + authenticate_pam, + create_session_cookie, generate_and_save_password, load_or_create_secret, load_password, @@ -485,6 +488,42 @@ async def login_page(): ) +@app.post("/login") +async def post_login( + request: Request, + username: str = Form(default=""), + password: str = Form(default=""), +) -> RedirectResponse: + """Validate credentials and issue a session cookie on success. + + In PAM mode, delegates to authenticate_pam(username, password). + In password mode, compares the submitted password to _auth_password. + + On success: redirect to / with a signed muxplex_session cookie. + On failure: redirect to /login?error=1. + """ + # Validate credentials + if _auth_mode == "pam": + valid = authenticate_pam(username, password) + else: + valid = password == _auth_password + + if not valid: + return RedirectResponse("/login?error=1", status_code=303) + + # Issue session cookie + cookie_value = create_session_cookie(_auth_secret, _auth_ttl) + response = RedirectResponse("/", status_code=303) + response.set_cookie( + "muxplex_session", + cookie_value, + httponly=True, + samesite="strict", + max_age=_auth_ttl if _auth_ttl > 0 else None, + ) + return response + + @app.get("/auth/mode") async def auth_mode_endpoint(): """Return the current auth mode and running username.""" diff --git a/muxplex/tests/test_api.py b/muxplex/tests/test_api.py index 45cafc0..4bad798 100644 --- a/muxplex/tests/test_api.py +++ b/muxplex/tests/test_api.py @@ -705,3 +705,72 @@ def test_get_auth_mode_returns_json(client): data = response.json() assert "mode" in data assert data["mode"] in ("pam", "password") + + +# --------------------------------------------------------------------------- +# POST /login +# --------------------------------------------------------------------------- + + +def test_post_login_correct_password_redirects_to_root(monkeypatch): + """POST /login with correct password: 303 redirect to / with muxplex_session cookie.""" + import muxplex.main as main_module + + monkeypatch.setenv("MUXPLEX_PASSWORD", "test-password") + monkeypatch.setattr(main_module, "_auth_mode", "password") + monkeypatch.setattr(main_module, "_auth_password", "test-password") + + with TestClient(app, follow_redirects=False) as c: + response = c.post( + "/login", data={"username": "user", "password": "test-password"} + ) + + assert response.status_code == 303 + assert response.headers["location"] == "/" + assert "muxplex_session" in response.cookies + + +def test_post_login_wrong_password_redirects_to_login_error(monkeypatch): + """POST /login with wrong password: 303 redirect to /login?error=1.""" + import muxplex.main as main_module + + monkeypatch.setenv("MUXPLEX_PASSWORD", "test-password") + monkeypatch.setattr(main_module, "_auth_mode", "password") + monkeypatch.setattr(main_module, "_auth_password", "test-password") + + with TestClient(app, follow_redirects=False) as c: + response = c.post( + "/login", data={"username": "user", "password": "wrong-password"} + ) + + assert response.status_code == 303 + assert "error=1" in response.headers["location"] + + +def test_post_login_pam_mode_correct_creds(monkeypatch): + """POST /login in PAM mode with correct creds: 303 to / with muxplex_session cookie.""" + import muxplex.main as main_module + + monkeypatch.setattr(main_module, "_auth_mode", "pam") + monkeypatch.setattr("muxplex.main.authenticate_pam", lambda u, p: True) + + with TestClient(app, follow_redirects=False) as c: + response = c.post("/login", data={"username": "user", "password": "correct"}) + + assert response.status_code == 303 + assert response.headers["location"] == "/" + assert "muxplex_session" in response.cookies + + +def test_post_login_pam_mode_wrong_creds(monkeypatch): + """POST /login in PAM mode with wrong creds: 303 redirect to /login?error=1.""" + import muxplex.main as main_module + + monkeypatch.setattr(main_module, "_auth_mode", "pam") + monkeypatch.setattr("muxplex.main.authenticate_pam", lambda u, p: False) + + with TestClient(app, follow_redirects=False) as c: + response = c.post("/login", data={"username": "user", "password": "wrong"}) + + assert response.status_code == 303 + assert "error=1" in response.headers["location"]