feat: add POST /login handler with PAM and password mode auth
- Import authenticate_pam and create_session_cookie from muxplex.auth
- Import Form, Request from fastapi; RedirectResponse from starlette.responses
- Add @app.post('/login') handler that reads username and password form fields
- In PAM mode: delegates to authenticate_pam(username, password)
- In password mode: compares password to _auth_password
- On success: creates signed muxplex_session cookie (httponly, samesite=strict),
redirects to / with 303
- On failure: redirects to /login?error=1 with 303
- Add 4 tests covering all branches (password success/failure, PAM success/failure)
Closes task-2-post-login-handler
This commit is contained in:
+40
-1
@@ -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."""
|
||||
|
||||
@@ -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"]
|
||||
|
||||
Reference in New Issue
Block a user