status-indicators/settings/probes.py

"""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