christianandClaude Fable 5.1
01c18ba5ef
Overview (problems in the room with the hold action, the live deck, one line per computer), Keys (both pages, what each key will say and what holding does), Computers (map with the keys that turn red, cards with a light per thing that runs there from the watchdog's snapshot and a read-only probe, Gear, Setup), Logs, History (the installer's copies for now), Help (four cards, the generated key guide, cheat sheet and Where things are, glossary, Companion facts), Site setup and Fixes read-only. The editing code is checked (with a lockout) but nothing can change yet. Also: faults are attributed to the computers their check is about; the installer refuses while a HyperDeck is recording (--now overrides); the Lighting PC's MIDI session is listed; the probes read sudo's answer correctly and expand $HOME outside the quotes. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
377 lines
18 KiB
Python
377 lines
18 KiB
Python
"""Read-only looks at the computers, for the Computers screen.
|
|
|
|
Every probe is a fixed command. Whole lists come back (every agent, every process, every task,
|
|
every service) and are matched here, so no name from the settings is ever put into a command
|
|
line. The only values that reach a command are two folder paths from the site block, shell-quoted.
|
|
Nothing is ever changed, nothing is woken, and a probe never runs while the watchdog runs a fix.
|
|
"""
|
|
import glob
|
|
import json
|
|
import os
|
|
import re
|
|
import shlex
|
|
import socket
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
|
|
from settings import PARENT # noqa: F401 (puts the project folder on sys.path)
|
|
import status_panel as sp
|
|
|
|
TIMEOUT = 8
|
|
|
|
MAC_PROBE = (
|
|
'echo "SP_NAME=$(uname -n)"; echo "SP_OSV=$(sw_vers -productVersion 2>/dev/null)"; '
|
|
'echo "SP_PY=$(/usr/bin/python3 --version 2>&1)"; '
|
|
'echo "SP_KEY=$(grep -c status-panel ~/.ssh/authorized_keys 2>/dev/null)"; '
|
|
'echo SP_AGENTS; launchctl list 2>/dev/null; echo SP_PROCS; ps -axo comm= 2>/dev/null; '
|
|
'echo SP_APPS; ls /Applications 2>/dev/null; echo SP_HELPER; ls -l {helper} 2>&1; sudo -n -l {helper} 2>&1 | tail -1; '
|
|
'echo SP_SOLO; cat {solo}/solo-mode.json 2>/dev/null; echo; stat -f %m {solo}/solo-mode.json 2>/dev/null; '
|
|
'echo SP_NOW; date +%s; echo SP_END'
|
|
)
|
|
WIN_PROBE = (
|
|
'echo SP_NAME=%COMPUTERNAME% & echo SP_TASKS & schtasks /Query /FO CSV /NH 2>nul & echo SP_PROCS & tasklist /FO CSV /NH 2>nul & '
|
|
'echo SP_SVC & sc query type= service state= all 2>nul & echo SP_PS & powershell -NoProfile -Command '
|
|
'"(Get-NetConnectionProfile | Select -First 1 -ExpandProperty NetworkCategory); '
|
|
'(Get-NetAdapterPowerManagement | Select -First 1 -ExpandProperty WakeOnMagicPacket); '
|
|
'((Get-NetFirewallRule -Name OpenSSH-Server-In-TCP -ErrorAction SilentlyContinue | Get-NetFirewallAddressFilter | '
|
|
'Select -ExpandProperty RemoteAddress) -join \',\')" 2>nul & echo SP_END'
|
|
)
|
|
|
|
|
|
def shell_path(path):
|
|
"""A path for a shell line: quoted, except that a leading ~ becomes $HOME outside the quotes."""
|
|
if path.startswith("~/"):
|
|
return '"$HOME"' + shlex.quote(path[1:])
|
|
return shlex.quote(path)
|
|
|
|
|
|
def quoted_paths(cfg):
|
|
site = cfg.get("site", {})
|
|
return {"helper": shell_path(site.get("dvs_helper", "/usr/local/sbin/status-panel-dvs-start")),
|
|
"solo": shell_path(site.get("solo_dir", "~/Library/Application Support/status-panel"))}
|
|
|
|
|
|
def run_ssh(cfg, alias, command, timeout=TIMEOUT):
|
|
try:
|
|
r = subprocess.run(sp.ssh_cmd(cfg, alias) + [command], capture_output=True, text=True, timeout=timeout)
|
|
return r.returncode, r.stdout, r.stderr
|
|
except subprocess.TimeoutExpired:
|
|
return None, "", "no answer within %d s" % timeout
|
|
except OSError as e:
|
|
return None, "", str(e)
|
|
|
|
|
|
def sections(out, markers):
|
|
"""Split probe output at the SP_... marker lines: {marker: [lines]}."""
|
|
parts, current = {}, None
|
|
for line in out.replace("\r", "").split("\n"):
|
|
key = line.strip()
|
|
if key in markers:
|
|
current = key
|
|
parts[current] = []
|
|
elif current:
|
|
parts[current].append(line)
|
|
return parts
|
|
|
|
|
|
def values(lines_or_out):
|
|
"""SP_KEY=value lines -> {"KEY": "value"}."""
|
|
text = "\n".join(lines_or_out) if isinstance(lines_or_out, list) else lines_or_out
|
|
return dict(re.findall(r"^SP_([A-Z]+)=(.*)$", text.replace("\r", ""), re.M))
|
|
|
|
|
|
def parse_launchctl(lines):
|
|
"""{label: pid or None} from `launchctl list`."""
|
|
agents = {}
|
|
for line in lines:
|
|
parts = line.split("\t")
|
|
if len(parts) == 3 and parts[2] and parts[2] != "Label":
|
|
agents[parts[2]] = int(parts[0]) if parts[0].isdigit() else None
|
|
return agents
|
|
|
|
|
|
def parse_csv_rows(lines):
|
|
out = []
|
|
for line in lines:
|
|
if not line.startswith('"'):
|
|
continue
|
|
out.append([c.strip('"') for c in line.strip().split('","')])
|
|
return out
|
|
|
|
|
|
def parse_sc(lines):
|
|
"""{service name: state word} from `sc query`."""
|
|
services, name = {}, None
|
|
for line in lines:
|
|
m = re.match(r"^\s*SERVICE_NAME:\s*(.+?)\s*$", line)
|
|
if m:
|
|
name = m.group(1)
|
|
m = re.match(r"^\s*STATE\s*:\s*\d+\s+(\w+)", line)
|
|
if m and name:
|
|
services[name] = m.group(1)
|
|
return services
|
|
|
|
|
|
def light(ok, reason=""):
|
|
return {"light": "ok" if ok else "problem", "reason": reason if not ok else ""}
|
|
|
|
|
|
def mac_items(host, parsed, cfg):
|
|
"""One line per thing that runs on a Mac, judged from the probe's lists."""
|
|
agents = parse_launchctl(parsed.get("SP_AGENTS", []))
|
|
procs = set(l.strip() for l in parsed.get("SP_PROCS", []))
|
|
proc_names = {os.path.basename(p) for p in procs} | {p.split("/")[-1] for p in procs}
|
|
apps = set(l.strip() for l in parsed.get("SP_APPS", []))
|
|
helper = "\n".join(parsed.get("SP_HELPER", []))
|
|
solo_lines = [l for l in parsed.get("SP_SOLO", []) if l.strip()]
|
|
now = int(("\n".join(parsed.get("SP_NOW", [])).strip() or "0") or 0)
|
|
items = []
|
|
for run in host.get("runs", []):
|
|
kind, name = run.get("kind"), run.get("name") or run.get("label") or run.get("process") or run.get("kind")
|
|
item = {"name": name, "kind": kind, "watched": run.get("watched", True)}
|
|
if kind == "sshd":
|
|
item.update(light(True))
|
|
elif kind == "app":
|
|
proc = run.get("process", "")
|
|
running = any(p == proc or p.endswith("/" + proc) for p in procs) or proc in proc_names
|
|
installed = (not run.get("path")) or True
|
|
item.update(light(running, "not running" if installed else "not installed"))
|
|
if not running and run.get("watched", True) is False:
|
|
item["light"] = "not_needed"
|
|
elif kind == "launchagent":
|
|
label = run.get("label", "")
|
|
if label not in agents:
|
|
item.update(light(False, "not loaded"))
|
|
elif agents[label] is None:
|
|
item.update(light(False, "loaded but not running"))
|
|
else:
|
|
item.update(light(True))
|
|
if run.get("watched", True) is False and item["light"] == "problem":
|
|
item["light"] = "not_needed"
|
|
elif kind == "disabled":
|
|
label = run.get("label", "")
|
|
item.update(light(label not in agents, "loaded - it must not run here"))
|
|
elif kind == "daemon":
|
|
want = run.get("process", "").lower().replace(" ", "")
|
|
running = any(want in p.lower().replace(" ", "") for p in procs) or any(want in a.lower() for a in agents) \
|
|
or (want.startswith("dantevirtual") and any("dvsd" in p for p in procs))
|
|
item.update(light(running, "not running"))
|
|
elif kind == "helper_dvs":
|
|
# `sudo -n -l <helper>` answers with the command itself when it is allowed without a password.
|
|
lines = [l for l in helper.splitlines() if l.strip()]
|
|
installed = "No such file" not in helper and "-rwx" in helper
|
|
last = lines[-1] if lines else ""
|
|
allowed = installed and "status-panel-dvs-start" in last and "not allowed" not in last.lower() and "may not" not in last.lower() \
|
|
and "password" not in last.lower()
|
|
item.update(light(installed and allowed, "not installed" if not installed else "sudo rule missing"))
|
|
elif kind == "helper_solo":
|
|
label = run.get("label", "org.waterschurch.status-panel.solo-mode")
|
|
try:
|
|
data = json.loads(solo_lines[0]) if solo_lines else {}
|
|
age = now - int(solo_lines[1]) if len(solo_lines) > 1 and now else None
|
|
except (ValueError, IndexError):
|
|
data, age = {}, None
|
|
running = agents.get(label) is not None
|
|
fresh = age is not None and age < 60
|
|
trusted = bool(data.get("trusted"))
|
|
if not running:
|
|
item.update(light(False, "helper not running"))
|
|
elif not trusted:
|
|
item.update(light(False, "Accessibility not allowed on that Mac"))
|
|
elif not fresh:
|
|
item.update(light(False, "helper not reporting"))
|
|
else:
|
|
item.update(light(True))
|
|
item["detail"] = {"afl": data.get("afl"), "latch": data.get("latch"), "age": age}
|
|
elif kind == "midi_session":
|
|
item.update({"light": "unknown", "reason": "see the MIDI key"})
|
|
elif kind in ("folder",):
|
|
item.update({"light": "unknown", "reason": "not checked here"})
|
|
else:
|
|
item.update({"light": "unknown", "reason": ""})
|
|
items.append(item)
|
|
return items, {"agents": sorted(a for a in agents if not a.startswith(("com.apple.", "application."))),
|
|
"processes": sorted(proc_names)[:200]}
|
|
|
|
|
|
def windows_items(host, parsed):
|
|
tasks = {row[0]: row for row in parse_csv_rows(parsed.get("SP_TASKS", [])) if row}
|
|
procs = {row[0].lower() for row in parse_csv_rows(parsed.get("SP_PROCS", [])) if row}
|
|
services = parse_sc(parsed.get("SP_SVC", []))
|
|
ps = [l.strip() for l in parsed.get("SP_PS", []) if l.strip()]
|
|
profile, wol, allow = (ps + ["", "", ""])[:3]
|
|
items = []
|
|
for run in host.get("runs", []):
|
|
kind, name = run.get("kind"), run.get("name") or run.get("kind")
|
|
item = {"name": name, "kind": kind, "watched": run.get("watched", True)}
|
|
if kind == "sshd":
|
|
item.update(light(services.get("sshd") == "RUNNING", "the SSH service is %s" % services.get("sshd", "missing")))
|
|
elif kind == "task":
|
|
full = "\\StatusPanel\\" + run["name"]
|
|
if full not in tasks:
|
|
item.update(light(False, "no scheduled task %s" % full))
|
|
else:
|
|
exe = (run.get("exe") or "").lower()
|
|
running = exe in procs if exe else None
|
|
if running is False:
|
|
item.update(light(False, "%s is not running" % run.get("exe")))
|
|
else:
|
|
item.update(light(True))
|
|
item["task_state"] = tasks.get(full, ["", "", ""])[2] if full in tasks else None
|
|
elif kind == "service":
|
|
state = services.get(run["name"])
|
|
item.update(light(state == "RUNNING", "service %s is %s" % (run["name"], state or "missing")))
|
|
elif kind in ("web", "artnet"):
|
|
item.update({"light": "unknown", "reason": "see the key that watches it"})
|
|
else:
|
|
item.update({"light": "unknown", "reason": ""})
|
|
items.append(item)
|
|
setup = {"profile": profile, "wake_on_lan": wol, "firewall_allow": allow}
|
|
return items, {"tasks": sorted(t for t in tasks if "StatusPanel" in t), "services": sorted(k for k, v in services.items()
|
|
if v == "RUNNING" and not k.lower().startswith(("w", "s", "d", "e", "n", "p", "t", "u", "c")))[:60]}, setup
|
|
|
|
|
|
def probe_host(cfg, alias, host):
|
|
"""One computer, over ssh. Returns the record the Computers screen shows."""
|
|
started = time.time()
|
|
rec = {"alias": alias, "name": host.get("name", alias), "os": host.get("os", "mac"), "time": started,
|
|
"reachable": False, "items": [], "found": {}, "setup": {}}
|
|
if host.get("self"):
|
|
return probe_self(cfg, alias, host)
|
|
if host.get("os") == "windows":
|
|
rc, out, err = run_ssh(cfg, alias, WIN_PROBE, TIMEOUT + 6)
|
|
if rc is None or "SP_END" not in out:
|
|
rec["error"] = err.strip() or "no answer"
|
|
return rec
|
|
rec["reachable"] = True
|
|
parsed = sections(out, ("SP_TASKS", "SP_PROCS", "SP_SVC", "SP_PS", "SP_END"))
|
|
rec["computer_name"] = values(out).get("NAME", "").strip()
|
|
rec["items"], rec["found"], rec["setup"] = windows_items(host, parsed)
|
|
rec["setup"]["key_ok"] = True
|
|
rec["setup"]["shell"] = "cmd" if rec["computer_name"] and "%" not in rec["computer_name"] else "not cmd"
|
|
else:
|
|
rc, out, err = run_ssh(cfg, alias, MAC_PROBE.format(**quoted_paths(cfg)))
|
|
if rc is None or "SP_END" not in out:
|
|
rec["error"] = err.strip() or "no answer"
|
|
return rec
|
|
rec["reachable"] = True
|
|
v = values(out)
|
|
parsed = sections(out, ("SP_AGENTS", "SP_PROCS", "SP_APPS", "SP_HELPER", "SP_SOLO", "SP_NOW", "SP_END"))
|
|
rec["computer_name"], rec["os_version"], rec["python"] = v.get("NAME", ""), v.get("OSV", ""), v.get("PY", "")
|
|
rec["items"], rec["found"] = mac_items(host, parsed, cfg)
|
|
rec["setup"] = {"key_ok": (v.get("KEY") or "0").strip() not in ("", "0"), "python_ok": "3." in v.get("PY", "")}
|
|
rec["took"] = round(time.time() - started, 1)
|
|
return rec
|
|
|
|
|
|
def listening_ports():
|
|
try:
|
|
out = subprocess.run(["/usr/sbin/lsof", "-nP", "-iTCP", "-sTCP:LISTEN"], capture_output=True, text=True, timeout=6).stdout
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
return []
|
|
ports = set()
|
|
for line in out.splitlines()[1:]:
|
|
parts = line.split()
|
|
if len(parts) > 8:
|
|
ports.add("%s %s" % (parts[0][:12], parts[8]))
|
|
return sorted(ports)
|
|
|
|
|
|
def interfaces():
|
|
"""[(name, ipv4, active?)] of this Mac's network ports."""
|
|
try:
|
|
out = subprocess.run(["/sbin/ifconfig"], capture_output=True, text=True, timeout=5).stdout
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
return []
|
|
found = []
|
|
for block in re.split(r"\n(?=\w)", out):
|
|
name = block.split(":")[0]
|
|
ip = re.search(r"inet (\d+\.\d+\.\d+\.\d+)", block)
|
|
if ip and name != "lo0":
|
|
found.append((name, ip.group(1), "status: active" in block))
|
|
return found
|
|
|
|
|
|
def probe_self(cfg, alias, host):
|
|
"""This Mac: no ssh; the same questions asked locally, plus the things only it has."""
|
|
started = time.time()
|
|
rec = {"alias": alias, "name": host.get("name", "this Mac"), "os": "mac", "time": started, "reachable": True,
|
|
"items": [], "found": {}, "setup": {}}
|
|
try:
|
|
agents_out = subprocess.run(["/bin/launchctl", "list"], capture_output=True, text=True, timeout=5).stdout
|
|
procs_out = subprocess.run(["/bin/ps", "-axo", "comm="], capture_output=True, text=True, timeout=5).stdout
|
|
except (OSError, subprocess.TimeoutExpired) as e:
|
|
rec["error"] = str(e)
|
|
return rec
|
|
paths = quoted_paths(cfg)
|
|
helper = subprocess.run(["/bin/zsh", "-c", "ls -l %s 2>&1; sudo -n -l %s 2>&1 | tail -1" % (paths["helper"], paths["helper"])],
|
|
capture_output=True, text=True, timeout=6).stdout
|
|
parsed = {"SP_AGENTS": agents_out.splitlines(), "SP_PROCS": procs_out.splitlines(), "SP_APPS": [],
|
|
"SP_HELPER": helper.splitlines(), "SP_SOLO": [], "SP_NOW": [str(int(time.time()))]}
|
|
rec["items"], rec["found"] = mac_items(host, parsed, cfg)
|
|
for item, run in zip(rec["items"], host.get("runs", [])):
|
|
if run.get("kind") == "folder":
|
|
path = os.path.expanduser(run["path"])
|
|
item.update(light(os.path.isdir(path), "folder not found (is Google Drive signed in?)"))
|
|
rec["found"]["ports"] = listening_ports()
|
|
ifaces = interfaces()
|
|
rec["found"]["interfaces"] = [{"iface": n, "ip": ip, "active": a} for n, ip, a in ifaces]
|
|
key = os.path.expanduser(cfg.get("ssh_key", "~/.ssh/id_status_panel"))
|
|
atem = cfg.get("atem") or {}
|
|
rec["setup"] = {
|
|
"login_key": os.path.exists(key) and os.path.exists(key + ".pub"),
|
|
"python_ok": (3, 9) <= tuple(int(x) for x in subprocess.run(["/usr/bin/python3", "-c", "import sys; print('%d.%d' % sys.version_info[:2])"],
|
|
capture_output=True, text=True, timeout=5).stdout.strip().split(".")) <= (3, 12),
|
|
"companion_db": bool(sp.companion_db_path(cfg)),
|
|
"netgear_keychain": "acct" in subprocess.run(["/usr/bin/security", "find-generic-password", "-s", "netgear-status-panel"],
|
|
capture_output=True, text=True, timeout=5).stdout if cfg.get("dante_switches") else None,
|
|
"atem_node": bool(sp.newest(atem["node"])) if atem.get("node") else None,
|
|
"atem_library": bool(sp.newest(atem["library"])) if atem.get("library") else None,
|
|
"switches": {s: port_open(s, 23) for s in cfg.get("dante_switches", [])},
|
|
"lan_ports": [n for n in host.get("networks", []) if n.get("use") == "LAN"],
|
|
}
|
|
rec["took"] = round(time.time() - started, 1)
|
|
return rec
|
|
|
|
|
|
def port_open(host, port, timeout=2):
|
|
try:
|
|
with socket.create_connection((host, port), timeout=timeout):
|
|
return True
|
|
except OSError:
|
|
return False
|
|
|
|
|
|
def probe_all(cfg, data_dir=None, only=None):
|
|
"""Every computer at once (threads), results cached in machines-live.json."""
|
|
results, threads = {}, []
|
|
for alias, host in cfg.get("hosts", {}).items():
|
|
if only and alias not in only:
|
|
continue
|
|
|
|
def work(alias=alias, host=host):
|
|
try:
|
|
results[alias] = probe_host(cfg, alias, host)
|
|
except Exception as e: # one broken probe must not hide the others
|
|
results[alias] = {"alias": alias, "name": host.get("name", alias), "reachable": False, "error": str(e),
|
|
"items": [], "found": {}, "setup": {}, "time": time.time()}
|
|
t = threading.Thread(target=work, daemon=True)
|
|
t.start()
|
|
threads.append(t)
|
|
for t in threads:
|
|
t.join(TIMEOUT + 10)
|
|
if data_dir:
|
|
path = os.path.join(data_dir, "machines-live.json")
|
|
try:
|
|
old = json.load(open(path)) if os.path.exists(path) else {}
|
|
except (OSError, ValueError):
|
|
old = {}
|
|
old.update(results)
|
|
sp.write_json(path, old)
|
|
try:
|
|
os.chmod(path, 0o600)
|
|
except OSError:
|
|
pass
|
|
return results
|