"""The settings page's server and probes, offline: routes against the fake Companion, the LAN binding, the editing-code gate, the probe parsers on captured output, and the teaching pages.""" import json import os import socket import sys import tempfile import threading import time import unittest import urllib.error import urllib.request from http.server import ThreadingHTTPServer ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.insert(0, ROOT) sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import status_panel as sp # noqa: E402 from settings import keyguide, probes, server # noqa: E402 from test_status_panel import CONFIG, FakeCompanion, healthy_snapshot # noqa: E402 def get(port, path, json_body=None, origin=None): data = json.dumps(json_body).encode() if json_body is not None else None req = urllib.request.Request("http://127.0.0.1:%d%s" % (port, path), data=data, headers={"Content-Type": "application/json", **({"Origin": origin} if origin else {})}) try: with urllib.request.urlopen(req, timeout=10) as r: body = r.read() return r.status, body except urllib.error.HTTPError as e: return e.code, e.read() class TestServer(unittest.TestCase): @classmethod def setUpClass(cls): snap = healthy_snapshot(time.time()) FakeCompanion.store = {"conns": snap.conns, "vars": dict(snap.vars), "custom": {"sp_state": json.dumps({"all": {"x": "ALL OK", "b": 1, "f": 2, "b2": 1, "f2": 2, "s": 20}}), "sp_hb": str(int(time.time())), "sp_cmd": "", "flash_tick": "0"}, "restarts": []} cls.fake = ThreadingHTTPServer(("127.0.0.1", 0), FakeCompanion) threading.Thread(target=cls.fake.serve_forever, daemon=True).start() cls.data = tempfile.mkdtemp(prefix="sp-settings-") cfg = json.loads(json.dumps(CONFIG)) cfg["companion"] = {"host": "127.0.0.1", "port": cls.fake.server_address[1]} cfg["site"]["editing_code"] = "apple river stone" json.dump(cfg, open(os.path.join(cls.data, "config.json"), "w")) now = time.time() json.dump({"pid": 1, "started_at": now - 100, "config_sha": "x", "version": "test"}, open(os.path.join(cls.data, "started.json"), "w")) json.dump({"live": True, "reason": "a HyperDeck is recording", "time": now}, open(os.path.join(cls.data, "live-now.json"), "w")) json.dump({"time": now, "conns": {}, "faults": {"atem": [{"check": "ATEM", "type": "conn", "sub": "conn", "level": "err", "msg": "NO ANSWER", "since": now - 60, "acked": False}], "midi": [{"check": "tracks/NA Tracks/Lighting PC", "type": "midi_peer", "sub": "peer", "level": "err", "msg": "FX LIGHTS MIDI LOST", "since": now - 60, "acked": False}]}, "midi": {"NA Ingest": {"peers": [{"name": "Lighting PC", "state": "running"}]}}, "remote_midi": {}, "http": {"http://10.6.7.13:8080/": True}, "ping": {"10.1.3.227": 0}, "service_on": True, "dante_map": {}}, open(os.path.join(cls.data, "snapshot.json"), "w")) sock = socket.socket(); sock.bind(("127.0.0.1", 0)); cls.port = sock.getsockname()[1]; sock.close() cls.site, cls.servers = server.serve(cls.data, cls.port, ["127.0.0.1"]) @classmethod def tearDownClass(cls): for s in cls.servers: s.shutdown() cls.fake.shutdown() def test_reads_are_open_and_secrets_are_masked(self): status, body = get(self.port, "/api/config") self.assertEqual(status, 200) cfg = json.loads(body)["config"] self.assertEqual(cfg["site"]["editing_code"], "set (17 characters)") self.assertNotIn("apple river stone", body.decode()) def test_overview_shows_the_deck_the_problems_and_the_live_state(self): status, body = get(self.port, "/api/overview") o = json.loads(body) self.assertEqual(o["deck"]["all"]["x"], "ALL OK") self.assertTrue(o["companion_ok"]) self.assertEqual([p["text"] for p in o["problems"]], ["ATEM: NO ANSWER", "MIDI: FX LIGHTS MIDI LOST"]) words = {c["alias"]: c["word"] for c in o["computers"]} self.assertEqual((words["tracks"], words["lighting"], words["prop"], words["ingest"]), ("Problem", "Problem", "Not looked yet", "OK")) self.assertIn("hold the ATEM key", o["problems"][0]["action"]) self.assertTrue(o["live"]["live"]) self.assertEqual(len(o["computers"]), len(CONFIG["hosts"])) self.assertEqual(o["watchdog"]["version"], "test") def test_companion_view_lists_connections_with_their_watchers(self): o = json.loads(get(self.port, "/api/companion")[1]) by = {c["label"]: c for c in o["connections"]} self.assertIn("hd2", by["HyperDeck_2"]["watched_by"]) self.assertTrue(o["lights"]["variable sp_state"]) self.assertEqual(o["missing"], []) def test_machines_use_what_the_watchdog_sees(self): o = json.loads(get(self.port, "/api/machines")[1]) ingest = next(c for c in o["computers"] if c["self"]) self.assertEqual(ingest["word"], "OK") bible = next(c for c in o["computers"] if c["alias"] == "bibletv") web = next(i for i in bible["items"] if i["kind"] == "web") self.assertEqual(web["light"], "ok") # from snapshot.json, not a probe self.assertIn("mac", o["setup_steps"]) def test_gear_and_keys_and_pages(self): gear = json.loads(get(self.port, "/api/gear")[1]) self.assertTrue(any(r["name"] == "HyperDeck_2" and "hd2" in r["watched_by"] for r in gear)) keys = json.loads(get(self.port, "/api/keys")[1]) hd2 = next(t for t in keys["tiles"] if t["id"] == "hd2") self.assertIn("NO DISK", hd2["says"]) self.assertIn("off and on", hd2["holding"]) self.assertEqual(keys["hold"], "HOLD 2s") for page in ("/key-guide.html", "/cheat.html", "/where.html", "/", "/app.js", "/style.css"): status, body = get(self.port, page) self.assertEqual(status, 200, page) guide = get(self.port, "/key-guide.html")[1].decode() for t in CONFIG["tiles"]: self.assertIn(t["label"], guide) self.assertNotIn("?", keyguide.cheat_sheet(CONFIG, "http://x/").split("

Where to look")[0].split("

")[1][:200]) def test_logs_are_whitelisted(self): self.assertEqual(get(self.port, "/api/logs/nope")[0], 404) self.assertEqual(get(self.port, "/api/logs/watchdog")[0], 200) self.assertEqual(json.loads(get(self.port, "/api/logs/autofix")[1])["file"], None) def test_editing_code_gate_and_lockout(self): self.site.attempts.clear() self.assertFalse(json.loads(get(self.port, "/api/editing/check", {"code": "nope"})[1])["ok"]) self.assertTrue(json.loads(get(self.port, "/api/editing/check", {"code": "Apple River Stone"})[1])["ok"]) for _ in range(5): get(self.port, "/api/editing/check", {"code": "wrong"}) r = json.loads(get(self.port, "/api/editing/check", {"code": "apple river stone"})[1]) self.assertFalse(r["ok"]) self.assertIn("locked_for", r) self.site.attempts.clear() self.assertEqual(get(self.port, "/api/editing/check", {"code": "x"}, origin="http://evil.example")[0], 403) def test_probe_is_refused_while_a_fix_runs(self): json.dump({"fix-x": {"key": "hd2"}}, open(os.path.join(self.data, "fixes-running.json"), "w")) try: self.assertEqual(get(self.port, "/api/machines/probe", {})[0], 409) finally: os.remove(os.path.join(self.data, "fixes-running.json")) class TestBindingAndProbes(unittest.TestCase): def test_bind_addresses(self): ifaces = [("en0", "10.1.1.1", True), ("en1", "10.6.3.222", True), ("en17", "169.254.1.2", True), ("en20", "10.66.1.2", True), ("en22", "10.104.40.6", True)] self.assertEqual(server.bind_addresses(CONFIG, ifaces), ["127.0.0.1", "10.1.1.1", "10.6.3.222"]) # the LAN ports only cfg = json.loads(json.dumps(CONFIG)) cfg["hosts"]["ingest"]["networks"] = [] self.assertEqual(server.bind_addresses(cfg, ifaces), ["127.0.0.1", "10.1.1.1", "10.6.3.222", "169.254.1.2", "10.66.1.2", "10.104.40.6"]) self.assertTrue(server.allowed_client("192.168.4.7")) self.assertFalse(server.allowed_client("8.8.8.8")) def test_mac_probe_parsing(self): out = ("SP_NAME=NA-Tracks.localdomain\nSP_OSV=26.5.1\nSP_PY=Python 3.9.6\nSP_KEY=1\nSP_AGENTS\nPID\tStatus\tLabel\n" "780\t0\tapplication.com.audinate.DanteVirtualSoundcard.1.2\n-\t0\tcom.google.keystone.agent\nSP_PROCS\n" "/Applications/Ableton Live 11 Suite.app/Contents/MacOS/Live\n/usr/sbin/sshd\n/Library/Audio/dvsd\nSP_APPS\nAbleton Live 11 Suite.app\n" "SP_HELPER\n-rwxr-xr-x 1 root wheel 375 Oct 8 12:05 /usr/local/sbin/status-panel-dvs-start\n/usr/local/sbin/status-panel-dvs-start\n" "SP_SOLO\n\nSP_NOW\n1791900000\nSP_END\n") parsed = probes.sections(out, ("SP_AGENTS", "SP_PROCS", "SP_APPS", "SP_HELPER", "SP_SOLO", "SP_NOW", "SP_END")) items, found = probes.mac_items(CONFIG["hosts"]["tracks"], parsed, CONFIG) by = {i["name"]: i for i in items} self.assertEqual(by["Ableton Live 11 Suite"]["light"], "ok") self.assertEqual(by["the DVS helper"]["light"], "ok") self.assertEqual(by["Dante Virtual Soundcard"]["light"], "ok") self.assertEqual(probes.values(out)["KEY"], "1") stopped = out.replace("/Contents/MacOS/Live\n", "") items, _ = probes.mac_items(CONFIG["hosts"]["tracks"], probes.sections(stopped, ("SP_AGENTS", "SP_PROCS", "SP_APPS", "SP_HELPER", "SP_SOLO", "SP_NOW", "SP_END")), CONFIG) self.assertEqual({i["name"]: i for i in items}["Ableton Live 11 Suite"]["reason"], "not running") def test_windows_probe_parsing(self): out = ('SP_NAME=LIGHTING-PC\r\nSP_TASKS\r\n"\\StatusPanel\\Start Arena","N/A","Running"\r\n"\\StatusPanel\\Start rtpMIDI","N/A","Ready"\r\n' '"\\StatusPanel\\Start Vista","N/A","Running"\r\nSP_PROCS\r\n"Vista.exe","1234","Console","1","100 K"\r\n"Arena.exe","1","Console","1","1 K"\r\n' 'SP_SVC\r\nSERVICE_NAME: sshd\r\n STATE : 4 RUNNING\r\nSERVICE_NAME: rtpMIDIService\r\n STATE : 4 RUNNING\r\n' 'SP_PS\r\nPublic\r\nEnabled\r\n10.1.1.1,10.6.3.222\r\nSP_END\r\n') parsed = probes.sections(out, ("SP_TASKS", "SP_PROCS", "SP_SVC", "SP_PS", "SP_END")) items, found, setup = probes.windows_items(CONFIG["hosts"]["lighting"], parsed) by = {i["name"]: i for i in items} self.assertEqual(by["Start Vista"]["light"], "ok") self.assertEqual(by["Start rtpMIDI"]["reason"], "rtpMIDI.exe is not running") self.assertEqual(by["rtpMIDIService"]["light"], "ok") self.assertEqual(by["sshd"]["light"], "ok") self.assertEqual((setup["profile"], setup["wake_on_lan"], setup["firewall_allow"]), ("Public", "Enabled", "10.1.1.1,10.6.3.222")) self.assertIn("\\StatusPanel\\Start Arena", found["tasks"]) def test_probe_commands_never_carry_names_from_the_settings(self): cfg = json.loads(json.dumps(CONFIG)) cfg["hosts"]["tracks"]["runs"][1]["process"] = "x; rm -rf ~" # a hostile name in the settings cfg["site"]["dvs_helper"] = "/usr/local/sbin/x; rm -rf ~" # and a hostile path line = probes.MAC_PROBE.format(**probes.quoted_paths(cfg)) self.assertNotIn("rm -rf ~", line.replace("'/usr/local/sbin/x; rm -rf ~'", "")) # only inside quotes self.assertNotIn("x; rm -rf ~\n", line) self.assertNotIn("Live", line) # names are matched here, never sent self.assertIn('"$HOME"\'/Library/Application Support/status-panel\'', probes.quoted_paths(CONFIG)["solo"]) if __name__ == "__main__": unittest.main()