dashboard/app/plugins/sessions.py

import asyncio
import re
from typing import Annotated, Any, cast

from fastapi import APIRouter, Form
from fastapi.responses import HTMLResponse

from app.plugins.base import Plugin
from app.render import render
from app.utils.subprocess import run_async, run_json_async

router = APIRouter(prefix="/api/plugins/sessions", tags=["plugins"])

LIST_TIMEOUT = 5
TERMINATE_TIMEOUT = 5
SESSION_ID_RE = re.compile(r"^[A-Za-z0-9]+$")

_terminate_lock = asyncio.Lock()


async def _list() -> tuple[list[dict[str, Any]], str]:
    """List the loginctl sessions of the user class.

    Runs `loginctl --json=short list-sessions` with a 5 s timeout (the
    child is killed on timeout) and keeps only entries whose class is
    "user" — greeter, lock-screen, overlay, and manager sessions are
    not shown. Every failure mode — missing binary, timeout, non-zero
    exit, bad JSON — is returned as a short error string rather than
    raised, so the UI can show a degraded state.

    Returns:
        (user-class session entries, "") on success, else ([], error
        description).
    """
    data, err = await run_json_async(
        ["loginctl", "--json=short", "list-sessions"], timeout=LIST_TIMEOUT
    )
    if err:
        return [], err[:200]
    if not isinstance(data, list):
        return [], "unexpected loginctl output"
    dicts: list[dict[str, Any]] = [e for e in cast("list[Any]", data) if isinstance(e, dict)]
    items = [e for e in dicts if e.get("class") == "user"]
    return items, ""


def _rows(entries: list[dict[str, Any]]) -> list[dict[str, str | bool]]:
    """Shape loginctl session entries into table rows for the UI.

    seat and tty are null for seatless sessions (e.g. SSH) and render
    as "—". idle defaults to False when the field is missing or not a
    boolean (newer systemd versions emit a boolean, older ones a
    "yes"/"no" string that degrades to "active" here).

    Args:
        entries: user-class entries from _list().

    Returns:
        One row per session: id, user, seat, tty, idle.
    """
    rows: list[dict[str, str | bool]] = []
    for e in entries:
        seat = e.get("seat")
        tty = e.get("tty")
        rows.append({
            "id": str(e.get("session", "?")),
            "user": str(e.get("user", "?")),
            "seat": seat if isinstance(seat, str) and seat else "—",
            "tty": tty if isinstance(tty, str) and tty else "—",
            "idle": e.get("idle") is True,
        })
    return rows


def _context(entries: list[dict[str, Any]], error: str, message: str = "") -> dict[str, Any]:
    """Build the template context shared by the state and skeleton fragments.

    Args:
        entries: user-class entries from _list().
        error: error string to display (from _list or a caller), "".
        message: transient success message to display, "".

    Returns:
        Context with sessions rows, message, and error.
    """
    return {
        "sessions": _rows(entries),
        "message": message,
        "error": error,
    }


async def _state(message: str = "", error: str = "") -> str:
    """Render the state fragment (polling view).

    Args:
        message: transient success message, or "".
        error: error to display (overrides the _list error), or "".

    Returns:
        The rendered sessions_state.html fragment.
    """
    entries, err = await _list()
    if error:
        err = error
    return render("plugins/sessions_state.html", **_context(entries, err, message))


async def _skeleton(message: str = "", error: str = "") -> str:
    """Render the full skeleton fragment (initial + post-action view).

    Args:
        message: transient success message, or "".
        error: error to display (overrides the _list error), or "".

    Returns:
        The rendered sessions_skeleton.html fragment.
    """
    entries, err = await _list()
    if error:
        err = error
    return render("plugins/sessions_skeleton.html", **_context(entries, err, message))


@router.get("/state")
async def state():
    """Poll endpoint: return the state fragment."""
    return HTMLResponse(await _state())


@router.post("/terminate")
async def terminate(session: Annotated[str, Form()]):
    """Terminate one session and re-render the skeleton.

    The requested session id is validated against a character whitelist
    and against a fresh _list() (which also supplies the user name for
    the result message) before `sudo loginctl terminate-session <id>` is
    run; unknown values are reported in the fragment, not raised.
    Serialized by a module-level lock so rapid double-clicks cannot
    issue the same terminate twice.

    Args:
        session: session id from the form.

    Returns:
        The skeleton fragment with a success message or error.
    """
    async with _terminate_lock:
        if not SESSION_ID_RE.match(session):
            return HTMLResponse(await _skeleton(error=f"invalid session id: {session}"))
        entries, err = await _list()
        if err:
            return HTMLResponse(await _skeleton(error=err))
        match = next((e for e in entries if str(e.get("session")) == session), None)
        if match is None:
            return HTMLResponse(await _skeleton(error=f"unknown session id: {session}"))
        rc, _out, err_out = await run_async(
            ["sudo", "loginctl", "--json=short", "terminate-session", session],
            timeout=TERMINATE_TIMEOUT,
        )
        if rc != 0:
            return HTMLResponse(
                await _skeleton(error=(err_out.strip() or f"terminate-session failed (rc={rc})")[:200])
            )
        user = str(match.get("user", "?"))
        return HTMLResponse(await _skeleton(message=f"terminated session {session} ({user})"))


plugin = Plugin(
    id="sessions",
    title="Active Sessions",
    description="User-class loginctl sessions and their idle state.",
    skeleton_fn=_skeleton,
)