bvandeusenandClaude Opus 5.5
329aa15e3d
A change Steward can't see happen (a config edited, a disk swapped) is recorded by whoever made it: what was done, its area, when (in the reader's own zone), the hosts, what it was done to and a detail. Its author or an admin edits or deletes it; recording, editing and deleting are audited and those audit rows stay off the timeline. log_change records one as "<user> via MCP", editable by the token's user. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
247 lines
11 KiB
Python
247 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from urllib.parse import urlencode
|
|
|
|
from quart import Blueprint, abort, current_app, redirect, render_template, request, session, url_for
|
|
from sqlalchemy import select
|
|
|
|
from steward.audit.service import mark_day_groups
|
|
from steward.auth.middleware import require_role, safe_next_url
|
|
from steward.core import formatting as fmt
|
|
from steward.core.audit import audit_request
|
|
from steward.core.paging import keyset, next_page
|
|
from steward.models.hosts import Host
|
|
from steward.models.timeline import TimelineEntry
|
|
from steward.models.users import UserRole
|
|
from steward.timeline import catalog, changes, config as tl_config, service as tl
|
|
|
|
timeline_bp = Blueprint("timeline", __name__)
|
|
|
|
|
|
def _filters() -> dict:
|
|
a = request.args
|
|
return {"host": a.get("host", "").strip(), "area": a.get("area", "").strip(),
|
|
"kind": a.get("kind", "").strip(), "hidden": a.get("hidden") == "1"}
|
|
|
|
|
|
@timeline_bp.get("/timeline/")
|
|
@require_role(UserRole.viewer)
|
|
async def timeline():
|
|
"""Everything that happened, newest first, a page at a time as the reader
|
|
scrolls; ?before=<cursor> is the next page's rows alone (core.paging)."""
|
|
f = _filters()
|
|
async with current_app.db_sessionmaker() as db:
|
|
cfg = await tl_config.load(db)
|
|
stmt = tl.query(cfg, host=f["host"], area=f["area"], kind=f["kind"],
|
|
include_hidden=f["hidden"])
|
|
stmt = keyset(stmt, TimelineEntry.at, TimelineEntry.id, request.args.get("before"))
|
|
rows, after = next_page((await db.execute(stmt)).scalars().all(), lambda e: (e.at, e.id))
|
|
items = _editable(await tl.views(db, cfg, rows))
|
|
hosts = [] if request.args.get("before") else \
|
|
list((await db.execute(select(Host.name).order_by(Host.name))).scalars())
|
|
last_group = mark_day_groups(items, fmt.zone(session.get("tz")), request.args.get("group", ""),
|
|
at=lambda item: item["entry"].at)
|
|
keep = {k: v for k, v in (("host", f["host"]), ("area", f["area"]), ("kind", f["kind"]),
|
|
("hidden", "1" if f["hidden"] else "")) if v}
|
|
back = "/timeline/" + ("?" + urlencode(keep) if keep else "")
|
|
ctx = {"items": items, "after": after, "last_group": last_group, "filters": f, "keep": keep, "back": back,
|
|
"areas": cfg.areas, "kinds": catalog.KINDS,
|
|
"hosts": hosts if not f["host"] or f["host"] in hosts else [f["host"], *hosts]}
|
|
if request.args.get("before"):
|
|
return await render_template("timeline/_rows.html", **ctx)
|
|
return await render_template("timeline/list.html", **ctx)
|
|
|
|
|
|
@timeline_bp.get("/timeline/host/<host_id>")
|
|
@require_role(UserRole.viewer)
|
|
async def host_card(host_id: str):
|
|
"""A host's newest entries, for its page (loaded into it)."""
|
|
async with current_app.db_sessionmaker() as db:
|
|
host = await db.get(Host, host_id)
|
|
if host is None:
|
|
return "", 404
|
|
cfg = await tl_config.load(db)
|
|
items = _editable(await tl.recent_for_host(db, cfg, host.name))
|
|
return await render_template("timeline/_host_card.html", host=host, items=items)
|
|
|
|
|
|
def _editable(items: list[dict]) -> list[dict]:
|
|
"""Mark the entries this reader may edit: changes they recorded, or any as an admin."""
|
|
for item in items:
|
|
item["editable"] = changes.can_edit(item["entry"], session.get("user_id"), session.get("user_role"))
|
|
return items
|
|
|
|
|
|
# ── Changes recorded by hand ────────────────────────────────────────────────
|
|
|
|
def _local(value: str | None) -> datetime | None:
|
|
"""A datetime-local field, in the reader's own zone."""
|
|
value = (value or "").strip()
|
|
if not value:
|
|
return None
|
|
try:
|
|
local = datetime.fromisoformat(value)
|
|
except ValueError:
|
|
raise ValueError(f"{value!r} is not a date and time.") from None
|
|
return local.replace(tzinfo=fmt.zone(session.get("tz"))).astimezone(timezone.utc)
|
|
|
|
|
|
def _as_local(when: datetime) -> str:
|
|
return when.astimezone(fmt.zone(session.get("tz"))).strftime("%Y-%m-%dT%H:%M")
|
|
|
|
|
|
async def _change_page(db, cfg, *, values: dict, entry=None, back: str, error: str | None = None,
|
|
status: int = 200):
|
|
from steward.models.hosts import Host
|
|
hosts = list((await db.execute(select(Host.name).order_by(Host.name))).scalars())
|
|
# A host named by an entry and since removed stays ticked, so a save keeps it.
|
|
hosts += [h for h in values["hosts"] if h not in hosts]
|
|
return await render_template("timeline/change_form.html", values=values, entry=entry, back=back,
|
|
error=error, hosts=hosts, areas=cfg.areas), status
|
|
|
|
|
|
def _back(raw: str | None) -> str:
|
|
"""Where saving returns: the page the form was opened from, else the timeline."""
|
|
return safe_next_url(raw) or "/timeline/"
|
|
|
|
|
|
async def _checked(db, cfg) -> tuple[dict | None, dict, str, str | None]:
|
|
"""(fields, values, back, error) from the posted form: the fields
|
|
changes.fields checked, or None and why not, with the form's own values
|
|
to show again."""
|
|
form = await request.form
|
|
values = {"summary": form.get("summary", ""), "area": form.get("area", ""),
|
|
"hosts": form.getlist("hosts"), "subject": form.get("subject", ""),
|
|
"note": form.get("note", ""), "at": form.get("at", "")}
|
|
back = _back(form.get("back"))
|
|
try:
|
|
f = await changes.fields(db, cfg, summary=values["summary"], area=values["area"],
|
|
hosts=values["hosts"], subject=values["subject"],
|
|
note=values["note"], at=_local(values["at"]))
|
|
except ValueError as exc:
|
|
return None, values, back, str(exc)
|
|
return f, values, back, None
|
|
|
|
|
|
@timeline_bp.get("/timeline/changes/new")
|
|
@require_role(UserRole.operator)
|
|
async def new_change():
|
|
"""The form for recording a change; ?host= ticks a host, ?back= is where saving returns."""
|
|
host = request.args.get("host", "").strip()
|
|
values = {"summary": "", "area": "", "hosts": [host] if host else [], "subject": "", "note": "",
|
|
"at": _as_local(datetime.now(timezone.utc))}
|
|
async with current_app.db_sessionmaker() as db:
|
|
cfg = await tl_config.load(db)
|
|
return await _change_page(db, cfg, values=values, back=_back(request.args.get("back")))
|
|
|
|
|
|
@timeline_bp.post("/timeline/changes")
|
|
@require_role(UserRole.operator)
|
|
async def record_change():
|
|
async with current_app.db_sessionmaker() as db:
|
|
async with db.begin():
|
|
cfg = await tl_config.load(db)
|
|
f, values, back, error = await _checked(db, cfg)
|
|
if error:
|
|
return await _change_page(db, cfg, values=values, back=back, error=error, status=400)
|
|
entry = changes.log(db, f, author_id=session.get("user_id"), actor=session.get("username", ""))
|
|
await audit_request("timeline.change_recorded", entity_type="timeline_entry", entity_id=entry.id,
|
|
detail={"name": entry.summary, "hosts": entry.hosts, "area": entry.area})
|
|
return redirect(back)
|
|
|
|
|
|
async def _own(db, entry_id: str):
|
|
"""The recorded change `entry_id`, when this reader may edit it; else 404 or 403."""
|
|
entry = await changes.get(db, entry_id)
|
|
if entry is None:
|
|
abort(404)
|
|
if not changes.can_edit(entry, session.get("user_id"), session.get("user_role")):
|
|
abort(403)
|
|
return entry
|
|
|
|
|
|
@timeline_bp.get("/timeline/changes/<entry_id>/edit")
|
|
@require_role(UserRole.operator)
|
|
async def edit_change(entry_id: str):
|
|
async with current_app.db_sessionmaker() as db:
|
|
entry = await _own(db, entry_id)
|
|
cfg = await tl_config.load(db)
|
|
values = {"summary": entry.summary, "area": entry.area, "hosts": list(entry.hosts),
|
|
"subject": entry.subject or "", "note": (entry.detail or {}).get("note") or "",
|
|
"at": _as_local(entry.at)}
|
|
return await _change_page(db, cfg, values=values, entry=entry,
|
|
back=_back(request.args.get("back")))
|
|
|
|
|
|
@timeline_bp.post("/timeline/changes/<entry_id>")
|
|
@require_role(UserRole.operator)
|
|
async def save_change(entry_id: str):
|
|
async with current_app.db_sessionmaker() as db:
|
|
async with db.begin():
|
|
entry = await _own(db, entry_id)
|
|
cfg = await tl_config.load(db)
|
|
f, values, back, error = await _checked(db, cfg)
|
|
if error:
|
|
return await _change_page(db, cfg, values=values, entry=entry, back=back,
|
|
error=error, status=400)
|
|
changes.edit(entry, f)
|
|
await audit_request("timeline.change_edited", entity_type="timeline_entry", entity_id=entry_id,
|
|
detail={"name": f["summary"]})
|
|
return redirect(back)
|
|
|
|
|
|
@timeline_bp.post("/timeline/changes/<entry_id>/delete")
|
|
@require_role(UserRole.operator)
|
|
async def delete_change(entry_id: str):
|
|
form = await request.form
|
|
async with current_app.db_sessionmaker() as db:
|
|
async with db.begin():
|
|
entry = await _own(db, entry_id)
|
|
summary = entry.summary
|
|
await db.delete(entry)
|
|
await audit_request("timeline.change_deleted", entity_type="timeline_entry", entity_id=entry_id,
|
|
detail={"name": summary})
|
|
return redirect(_back(form.get("back")))
|
|
|
|
|
|
@timeline_bp.get("/settings/timeline/")
|
|
@require_role(UserRole.admin)
|
|
async def settings_page():
|
|
async with current_app.db_sessionmaker() as db:
|
|
cfg = await tl_config.load(db)
|
|
from steward.core.settings import get_setting
|
|
retention = int(await get_setting(db, "timeline.retention_days") or 0)
|
|
return await render_template("settings/timeline.html", **_settings_ctx(cfg, retention))
|
|
|
|
|
|
def _settings_ctx(cfg, retention: int, error: str | None = None) -> dict:
|
|
by_kind = {k: [t for t in catalog.TYPES.values() if t.kind == k] for k in catalog.KINDS}
|
|
return {"cfg": cfg, "retention": retention, "error": error, "kinds": catalog.KINDS,
|
|
"by_kind": {k: v for k, v in by_kind.items() if v}}
|
|
|
|
|
|
@timeline_bp.post("/settings/timeline/")
|
|
@require_role(UserRole.admin)
|
|
async def save_settings():
|
|
from steward.core.settings import set_setting
|
|
form = await request.form
|
|
try:
|
|
retention = max(0, int(form.get("retention_days") or 0))
|
|
except ValueError:
|
|
retention = None
|
|
async with current_app.db_sessionmaker() as db:
|
|
async with db.begin():
|
|
cfg = await tl_config.load(db)
|
|
new, error = tl_config.edited(cfg, form)
|
|
if error is None and retention is None:
|
|
error = "Keep for: a whole number of days, or 0 to keep everything."
|
|
if error is not None:
|
|
kept = retention if retention is not None else 0
|
|
return await render_template("settings/timeline.html",
|
|
**_settings_ctx(cfg, kept, error)), 400
|
|
await tl_config.save(db, new)
|
|
await set_setting(db, "timeline.retention_days", retention)
|
|
await audit_request("settings.saved", detail={"section": "timeline"})
|
|
return redirect(url_for("timeline.settings_page"))
|