225 lines
7.2 KiB
Python
225 lines
7.2 KiB
Python
import logging
|
|
from typing import List
|
|
|
|
from fastapi import Body, Depends, FastAPI, HTTPException, Request, Response, status
|
|
from fastapi.responses import HTMLResponse, RedirectResponse
|
|
from fastapi.templating import Jinja2Templates
|
|
|
|
from backend import actions
|
|
from backend.actions import ActionError
|
|
from backend.auth import (
|
|
authenticate_admin_user,
|
|
get_current_admin,
|
|
is_authorized_admin,
|
|
is_account_locked,
|
|
issue_token,
|
|
list_manageable_users,
|
|
)
|
|
from backend.models import ActionRequest, ActionResponse, LoginRequest, LoginResponse, UserStatus
|
|
from backend.oidc import OIDCClient, OIDCError
|
|
from backend.settings import Settings, get_settings
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s %(levelname)s %(name)s - %(message)s",
|
|
)
|
|
logger = logging.getLogger("skd")
|
|
|
|
app = FastAPI(title="Safe Kiddo Daemon", version="1.0.0")
|
|
templates = Jinja2Templates(directory="backend/templates")
|
|
|
|
|
|
def get_oidc_client(settings: Settings = Depends(get_settings)) -> OIDCClient:
|
|
if not settings.oidc_enabled:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail="OIDC not configured",
|
|
)
|
|
try:
|
|
return OIDCClient(settings)
|
|
except OIDCError as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=str(exc)
|
|
) from exc
|
|
|
|
|
|
def validate_user(username: str, settings: Settings = Depends(get_settings)) -> str:
|
|
allowed = set(list_manageable_users(settings))
|
|
if username not in allowed:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="User not allowed")
|
|
return username
|
|
|
|
|
|
@app.get("/health")
|
|
def health(settings: Settings = Depends(get_settings)) -> dict:
|
|
return {"status": "ok", "dry_run": settings.dry_run}
|
|
|
|
|
|
@app.get("/me")
|
|
def whoami(
|
|
current_user: str = Depends(get_current_admin),
|
|
settings: Settings = Depends(get_settings),
|
|
) -> dict:
|
|
auth_mode = "pam+oidc" if settings.oidc_enabled else "pam"
|
|
return {"user": current_user, "auth_mode": auth_mode}
|
|
|
|
|
|
@app.post("/login", response_model=LoginResponse)
|
|
def login(
|
|
payload: LoginRequest,
|
|
response: Response,
|
|
settings: Settings = Depends(get_settings),
|
|
) -> LoginResponse:
|
|
authenticate_admin_user(payload.username, payload.password, settings)
|
|
token = issue_token(payload.username, settings)
|
|
response.set_cookie(
|
|
settings.session_cookie_name,
|
|
token,
|
|
max_age=settings.token_ttl_seconds,
|
|
httponly=True,
|
|
secure=settings.session_cookie_secure,
|
|
samesite="lax",
|
|
)
|
|
return LoginResponse(token=token, expires_in=settings.token_ttl_seconds)
|
|
|
|
|
|
@app.get("/login/oidc/start")
|
|
def oidc_start(
|
|
settings: Settings = Depends(get_settings),
|
|
oidc: OIDCClient = Depends(get_oidc_client),
|
|
):
|
|
state = oidc.build_state_token()
|
|
redirect = RedirectResponse(url=oidc.authorization_url(state))
|
|
redirect.set_cookie(
|
|
settings.oidc_state_cookie_name,
|
|
state,
|
|
max_age=300,
|
|
httponly=True,
|
|
secure=settings.session_cookie_secure,
|
|
samesite="lax",
|
|
)
|
|
return redirect
|
|
|
|
|
|
@app.get("/login/oidc/callback")
|
|
def oidc_callback(
|
|
request: Request,
|
|
code: str,
|
|
state: str,
|
|
settings: Settings = Depends(get_settings),
|
|
oidc: OIDCClient = Depends(get_oidc_client),
|
|
):
|
|
stored_state = request.cookies.get(settings.oidc_state_cookie_name, "")
|
|
if not oidc.is_state_valid(state, stored_state):
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid OIDC state")
|
|
|
|
claims = oidc.exchange_code_for_claims(code)
|
|
username = oidc.extract_username(claims)
|
|
if not username:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Missing username claim")
|
|
if not is_authorized_admin(username, settings, mode="oidc"):
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="User not authorized to log in")
|
|
|
|
token = issue_token(username, settings)
|
|
redirect = RedirectResponse(url="/")
|
|
redirect.set_cookie(
|
|
settings.session_cookie_name,
|
|
token,
|
|
max_age=settings.token_ttl_seconds,
|
|
httponly=True,
|
|
secure=settings.session_cookie_secure,
|
|
samesite="lax",
|
|
)
|
|
redirect.delete_cookie(settings.oidc_state_cookie_name)
|
|
return redirect
|
|
|
|
|
|
@app.get("/login/oidc/status")
|
|
def oidc_status(settings: Settings = Depends(get_settings)) -> dict:
|
|
return {"enabled": settings.oidc_enabled}
|
|
|
|
|
|
@app.get("/users", response_model=List[UserStatus], dependencies=[Depends(get_current_admin)])
|
|
def users(settings: Settings = Depends(get_settings)) -> List[UserStatus]:
|
|
logged_in = set(actions.list_logged_in_users())
|
|
targets = list_manageable_users(settings)
|
|
return [
|
|
UserStatus(
|
|
user=user,
|
|
logged_in=user in logged_in,
|
|
account_locked=is_account_locked(user),
|
|
)
|
|
for user in targets
|
|
]
|
|
|
|
|
|
@app.post(
|
|
"/users/{username}/disable",
|
|
response_model=ActionResponse,
|
|
dependencies=[Depends(get_current_admin)],
|
|
)
|
|
def disable_user(
|
|
username: str = Depends(validate_user),
|
|
payload: ActionRequest | None = Body(default=None),
|
|
settings: Settings = Depends(get_settings),
|
|
current_user: str = Depends(get_current_admin),
|
|
) -> ActionResponse:
|
|
if username == current_user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Cannot disable current user",
|
|
)
|
|
try:
|
|
steps = actions.disable_user(
|
|
username,
|
|
countdown=payload.countdown if payload else None,
|
|
sound=payload.sound if payload else None,
|
|
message=payload.message if payload else None,
|
|
)
|
|
except ActionError as exc:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
|
|
except Exception as exc: # pragma: no cover - safeguard
|
|
logger.exception("Failed to disable %s", username)
|
|
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="failed") from exc
|
|
|
|
logged_in = username in actions.list_logged_in_users()
|
|
return ActionResponse(
|
|
user=username,
|
|
action="disable",
|
|
dry_run=settings.dry_run,
|
|
steps=steps,
|
|
logged_in=logged_in,
|
|
)
|
|
|
|
|
|
@app.post(
|
|
"/users/{username}/enable",
|
|
response_model=ActionResponse,
|
|
dependencies=[Depends(get_current_admin)],
|
|
)
|
|
def enable_user(
|
|
username: str = Depends(validate_user),
|
|
settings: Settings = Depends(get_settings),
|
|
) -> ActionResponse:
|
|
try:
|
|
steps = actions.enable_user(username)
|
|
except ActionError as exc:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
|
|
except Exception as exc: # pragma: no cover - safeguard
|
|
logger.exception("Failed to enable %s", username)
|
|
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="failed") from exc
|
|
|
|
logged_in = username in actions.list_logged_in_users()
|
|
return ActionResponse(
|
|
user=username,
|
|
action="enable",
|
|
dry_run=settings.dry_run,
|
|
steps=steps,
|
|
logged_in=logged_in,
|
|
)
|
|
|
|
|
|
@app.get("/", response_class=HTMLResponse)
|
|
def index(request: Request) -> HTMLResponse:
|
|
return templates.TemplateResponse("index.html", {"request": request})
|