pyr0ball
251109ae96
- wizard.py: wrap syslog_port int() in try/except to default 514 on non-numeric input - ContextView: add try/catch to doDelete, doDeleteFact, addFact for network errors - ContextView: arrow-key navigation for tablist (ArrowLeft/ArrowRight) - DiagnoseView: arrow-key navigation for tablist (ArrowLeft/ArrowRight) - WizardOverlay: reset current_step to last schema step when clicking 'Go back and edit' - WizardOverlay: focus trap on Tab/Shift+Tab within dialog element
128 lines
4.4 KiB
Python
128 lines
4.4 KiB
Python
"""Wizard state machine — MIT (structured Q&A); BSL gate reserved for LLM path."""
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from app.context.store import add_fact
|
|
|
|
WIZARD_STEPS: list[dict[str, Any]] = [
|
|
{
|
|
"step": 1,
|
|
"id": "os",
|
|
"title": "What operating system is this server running?",
|
|
"type": "select",
|
|
"options": ["Linux (systemd/journald)", "Linux (other init)", "Other"],
|
|
"optional": False,
|
|
"help": "This determines which log sources Turnstone can watch.",
|
|
},
|
|
{
|
|
"step": 2,
|
|
"id": "hostname",
|
|
"title": "What is this server's hostname?",
|
|
"type": "text",
|
|
"placeholder": "e.g. heimdall.local",
|
|
"optional": False,
|
|
"help": "Used to label log sources in diagnosis results.",
|
|
},
|
|
{
|
|
"step": 3,
|
|
"id": "services",
|
|
"title": "Which named systemd services do you want to monitor?",
|
|
"type": "text",
|
|
"placeholder": "e.g. plex.service, sonarr.service",
|
|
"optional": True,
|
|
"help": "Comma-separated. Leave blank to collect all journal output.",
|
|
},
|
|
{
|
|
"step": 4,
|
|
"id": "docker",
|
|
"title": "Are you running Docker or Podman containers?",
|
|
"type": "select",
|
|
"options": ["Yes — Docker", "Yes — Podman", "No"],
|
|
"optional": False,
|
|
"help": "Turnstone can tail container log streams directly.",
|
|
},
|
|
{
|
|
"step": 5,
|
|
"id": "syslog",
|
|
"title": "Do any network devices send syslog UDP to this server?",
|
|
"type": "select",
|
|
"options": ["Yes — UDP 514", "Yes — custom port", "No"],
|
|
"optional": False,
|
|
"help": "Routers, switches, and APs can forward logs via UDP syslog.",
|
|
},
|
|
{
|
|
"step": 6,
|
|
"id": "syslog_port",
|
|
"title": "What UDP port does your syslog device send to?",
|
|
"type": "text",
|
|
"placeholder": "e.g. 514",
|
|
"optional": True,
|
|
"condition": {"step_id": "syslog", "value": "Yes — custom port"},
|
|
"help": "Only needed if you chose 'custom port' above.",
|
|
},
|
|
]
|
|
|
|
TOTAL_STEPS: int = len(WIZARD_STEPS)
|
|
|
|
|
|
def get_schema() -> list[dict[str, Any]]:
|
|
"""Return the wizard schema (all steps)."""
|
|
return WIZARD_STEPS
|
|
|
|
|
|
def advance_step(session: dict[str, Any], step_id: str, answer: Any) -> dict[str, Any]:
|
|
"""Return a new session dict with the answer recorded and current_step incremented."""
|
|
answers = {**session.get("answers", {}), step_id: answer}
|
|
return {**session, "answers": answers, "current_step": session.get("current_step", 1) + 1}
|
|
|
|
|
|
def is_complete(session: dict[str, Any]) -> bool:
|
|
"""Check if wizard session has progressed past all steps."""
|
|
return session.get("current_step", 1) > TOTAL_STEPS
|
|
|
|
|
|
def apply_session(db_path: Path, session: dict[str, Any]) -> dict[str, Any]:
|
|
"""Write context facts and return source config from a completed wizard session."""
|
|
answers: dict[str, Any] = session.get("answers", {})
|
|
|
|
facts_written = 0
|
|
hostname = str(answers.get("hostname") or "this-host")
|
|
|
|
if answers.get("hostname"):
|
|
add_fact(db_path, "host", "hostname", hostname, source="wizard")
|
|
facts_written += 1
|
|
if answers.get("os"):
|
|
add_fact(db_path, "host", "os", str(answers["os"]), source="wizard")
|
|
facts_written += 1
|
|
|
|
sources: list[dict[str, Any]] = []
|
|
|
|
services_raw = str(answers.get("services") or "")
|
|
services = [s.strip() for s in services_raw.split(",") if s.strip()]
|
|
if services:
|
|
for svc in services:
|
|
sources.append({"type": "journald", "id": f"journal:{hostname}:{svc}", "unit": svc})
|
|
else:
|
|
sources.append({"type": "journald", "id": f"journal:{hostname}"})
|
|
|
|
docker_answer = str(answers.get("docker") or "No")
|
|
if "Docker" in docker_answer:
|
|
sources.append({"type": "docker", "id": f"docker:{hostname}"})
|
|
elif "Podman" in docker_answer:
|
|
sources.append({"type": "docker", "id": f"podman:{hostname}", "runtime": "podman"})
|
|
|
|
syslog_answer = str(answers.get("syslog") or "No")
|
|
if syslog_answer.startswith("Yes"):
|
|
try:
|
|
port = int(answers.get("syslog_port") or 514)
|
|
except (ValueError, TypeError):
|
|
port = 514
|
|
sources.append({"type": "syslog", "id": f"syslog:{hostname}", "port": port})
|
|
|
|
return {
|
|
"facts_written": facts_written,
|
|
"sources": sources,
|
|
"source_count": len(sources),
|
|
}
|