65 lines
2.0 KiB
Python
65 lines
2.0 KiB
Python
"""Dashboard login — minimal cookie-session auth.
|
|
|
|
Credentials come from `core_config.json` → `auth.{username,password}`.
|
|
The session is signed by Starlette's SessionMiddleware (stateless cookie).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, HTTPException, Request
|
|
from fastapi.responses import HTMLResponse, RedirectResponse
|
|
from pydantic import BaseModel
|
|
|
|
from Project.Sanad.config import BASE_DIR
|
|
from Project.Sanad.core.config_loader import section as _cfg_section
|
|
from Project.Sanad.core.logger import get_logger
|
|
|
|
router = APIRouter()
|
|
log = get_logger("dashboard.auth")
|
|
|
|
_AUTH_CFG = _cfg_section("core", "auth") or {}
|
|
USERNAME = _AUTH_CFG.get("username", "admin")
|
|
PASSWORD = _AUTH_CFG.get("password", "admin")
|
|
|
|
LOGIN_PAGE = BASE_DIR / "dashboard" / "static" / "login.html"
|
|
|
|
|
|
def is_authed(request: Request) -> bool:
|
|
return bool(request.session.get("user"))
|
|
|
|
|
|
class LoginPayload(BaseModel):
|
|
username: str
|
|
password: str
|
|
|
|
|
|
@router.get("/login", include_in_schema=False)
|
|
async def login_page(request: Request):
|
|
if is_authed(request):
|
|
return RedirectResponse("/", status_code=303)
|
|
if LOGIN_PAGE.exists():
|
|
return HTMLResponse(LOGIN_PAGE.read_text(encoding="utf-8"))
|
|
return HTMLResponse("<h1>Login page missing</h1>", status_code=500)
|
|
|
|
|
|
@router.post("/api/auth/login")
|
|
async def login(request: Request, payload: LoginPayload):
|
|
if payload.username == USERNAME and payload.password == PASSWORD:
|
|
request.session["user"] = payload.username
|
|
log.info("login OK: %s", payload.username)
|
|
return {"ok": True, "user": payload.username}
|
|
log.warning("login FAILED: %s", payload.username)
|
|
raise HTTPException(401, "Invalid username or password")
|
|
|
|
|
|
@router.post("/api/auth/logout")
|
|
async def logout(request: Request):
|
|
user = request.session.pop("user", None)
|
|
log.info("logout: %s", user)
|
|
return {"ok": True}
|
|
|
|
|
|
@router.get("/api/auth/me")
|
|
async def whoami(request: Request):
|
|
return {"authenticated": is_authed(request), "user": request.session.get("user")}
|