#!/usr/bin/env python3 # SPDX-License-Identifier: Apache-2.0 """End to end: the birdhouse camera simulator reports to a HEART Monitor site running on this machine. Starts the monitor site (heart-monitor/test/serve.mjs, with an in-memory database and an email outbox), runs the simulator with every staged fault at 5x speed (simulated clock from 07:00, so the birds are about), signs in with an emailed code, registers the device through the site's own form, and checks for about four minutes that: * every message the simulator serves is valid against the official schemas (spec/schemas/1.0), Playbooks (heart.rule), the control policy with its allowlist (heart.control) and Actions (heart.action) included; * the public views hold nothing but public data, and private views need the owner key; * the owner sets the zoom at once, inside its range, and an out-of-range zoom is refused; * an IR light change (physical) waits: an approval over the network with only the key is refused with 403 presence-required, and the change runs only after the button file is touched; * both zoom Playbooks fired, every automatic change stayed inside the allowlist envelope, and refused ones say why; * every staged fault shows on the owner's device page with its standard fault code; * no reading, bird count, egg count or zoom ever reaches the site. Needs: Node 20+, Python 3 with jsonschema and referencing (pip install "jsonschema[format]>=4.18" referencing). Usage, from the project's top folder: python3 birdhouseSimulator/test_e2e.py """ import json import pathlib import re import subprocess import sys import tempfile import time import urllib.error import urllib.request HERE = pathlib.Path(__file__).resolve().parent ROOT = HERE.parent sys.path.insert(0, str(ROOT / "heart-monitor" / "test")) import e2e # noqa: E402 the monitor site's own end-to-end helpers (sign in, forms, outbox) import jsonschema # noqa: E402 from referencing import Registry, Resource # noqa: E402 SITE_PORT, API_PORT, RUN_SEC = 8801, 43301, 260 e2e.SITE_PORT, e2e.SITE = SITE_PORT, f"http://127.0.0.1:{SITE_PORT}" STAGED = ["storage.nearly-full", "storage.full", "power.battery-low", "performance.too-hot", "connectivity.link-lost", "system.main-function-lost"] RULES = ["zoom-out-for-a-crowd", "zoom-in-on-the-eggs"] ENV = {"min": 1.0, "max": 4.0, "maxStep": 1.0, "maxPerDay": 12, "cooldownSec": 300} NEVER_ON_SITE = ["Cel", "dBm", "zoom", "eggs", "egg_", "birds_in_view", "rssi", "pictures_today", "sd_free", "ir_level"] registry = Registry() for p in (ROOT / "spec" / "schemas" / "1.0").glob("*.json"): d = json.loads(p.read_text()) registry = registry.with_resource(d["$id"], Resource.from_contents(d)) validator = jsonschema.Draft202012Validator(json.loads((ROOT / "spec/schemas/1.0/envelope.schema.json").read_text()), registry=registry) results = [] def check(name, ok, detail=""): results.append(ok) print(f"{'PASS' if ok else 'FAIL'} {name}{'' if ok else ': ' + detail}", flush=True) def call(method, path, key=None, body=None): data = json.dumps(body).encode() if body is not None else (b"{}" if method != "GET" else None) req = urllib.request.Request(f"http://127.0.0.1:{API_PORT}{path}", data=data, method=method, headers={"Content-Type": "application/json", **({"Authorization": f"Bearer {key}"} if key else {})}) try: with urllib.request.urlopen(req, timeout=10) as r: return r.status, json.loads(r.read() or b"{}") except urllib.error.HTTPError as e: try: return e.code, json.loads(e.read() or b"{}") except ValueError: return e.code, {} def get(path, key=None): s, b = call("GET", path, key) if s != 200: raise OSError(f"GET {path} answered {s}") return b def prop(name, key): return next(p["data"]["value"] for p in get("/heart/v1/properties", key)["items"] if p["data"]["name"].endswith("." + name)) def main(): work = pathlib.Path(tempfile.mkdtemp(prefix="heart-birdhouse-")) site = subprocess.Popen(["node", str(ROOT / "heart-monitor/test/serve.mjs"), str(SITE_PORT)], stderr=subprocess.DEVNULL) log_path = work / "simulator.log" sim = None try: def site_up(): try: return e2e.http("GET", "/monitor")[0] == 200 except OSError: return False e2e.until(site_up, 20, "the monitor site") with open(log_path, "w") as log: sim = subprocess.Popen([sys.executable, str(HERE / "birdhouseSimulator.py"), "--monitor-url", e2e.SITE, "--scenario", "all", "--speed", "5", "--seed", "7", "--clock", "07:00", "--duration", str(RUN_SEC), "--port", str(API_PORT), "--state-file", str(work / "birdhouse-monitor.json")], stdout=log, stderr=log, cwd=work) text = log_path.read_text m = e2e.until(lambda: re.search(r"register\?id=([0-9a-f]{16})\n\s+3\. Enter the claim code\s+([0-9A-Z-]{23})", text()), 20, "the claim code on the console") dev, code = m.group(1), m.group(2) owner = e2e.login("owner@example.com") s = e2e.http("POST", "/register", {"id": dev, "code": code.lower(), "name": "Birdhouse demo"}, cookie=owner)[0] check("the owner registers the device with its claim code", s in (200, 303), f"status {s}") key = e2e.until(lambda: re.search(r"owner key for private views: (\S+)\)", text()), 10, "owner key").group(1) # The owner's own changes: the zoom (not physical) runs at once; the IR light (physical) waits for the button. s, a = call("POST", "/heart/v1/capabilities/camera.zoom/invoke", key, {"arguments": {"zoom": 1.5}}) check("the owner sets the zoom at once with the owner key", s == 200 and a["data"]["status"] == "succeeded" and a["data"]["initiator"]["kind"] == "human" and prop("zoom", key) == 1.5, f"{s} {a}") s1 = call("POST", "/heart/v1/capabilities/camera.zoom/invoke", key, {"arguments": {"zoom": 5.0}})[0] s2 = call("POST", "/heart/v1/capabilities/camera.zoom/invoke", None, {"arguments": {"zoom": 2.0}})[0] check("a zoom outside 1.0 to 4.0 is refused (422), and no change runs without the key (401)", s1 == 422 and s2 == 401, f"{s1} {s2}") ir_before = prop("ir_level", key) s, a = call("POST", "/heart/v1/capabilities/camera.ir_level/invoke", key, {"arguments": {"level": 40}}) aid = a.get("data", {}).get("id", "") check("an IR light change is physical and waits for a person (202 awaiting-approval)", s == 202 and a["data"]["status"] == "awaiting-approval" and a["data"]["physical"] is True, f"{s} {a}") s1, p1 = call("POST", f"/heart/v1/actions/{aid}/decision", key, {"decision": "approve"}) s2, p2 = call("POST", f"/heart/v1/actions/{aid}/decision", key, {"decision": "approve", "presence": {"method": "device-button", "digest": "0" * 64, "proof": "x"}}) still = get(f"/heart/v1/actions/{aid}", key)["data"]["status"] check("approving the IR change over the network with only the key is refused (403 presence-required)", s1 == 403 and p1.get("code") == "presence-required" and s2 == 403 and still == "awaiting-approval" and prop("ir_level", key) == ir_before, f"{s1} {p1.get('code')} {s2} {still}") (work / "button.press").touch() done = e2e.until(lambda: get(f"/heart/v1/actions/{aid}", key)["data"]["status"] != "awaiting-approval", 10, "the button") act = get(f"/heart/v1/actions/{aid}", key)["data"] check("the IR change runs once the button at the box is pressed (touch button.press)", done and act["status"] == "succeeded" and act["decision"]["presenceMethod"] == "device-button" and prop("ir_level", key) == 40, str(act)) s = call("POST", "/heart/v1/capabilities/camera.ir_level/invoke", key, {"arguments": {"level": 80}})[0] check("an IR level above the firmware's 60 % is refused (422)", s == 422, f"status {s}") invalid, leaks, pages = [], [], [] end = time.time() + RUN_SEC - 15 while time.time() < end and sim.poll() is None: try: msgs = ([get("/.well-known/heart-card.json"), get("/.well-known/heart-card.json", key), get("/heart/v1/pulse"), get("/heart/v1/pulse", key), get("/heart/v1/tending/autonomy", key)] + get("/heart/v1/arrhythmias")["items"] + get("/heart/v1/arrhythmias", key)["items"] + get("/heart/v1/properties")["items"] + get("/heart/v1/properties", key)["items"] + get("/heart/v1/tending/rules", key)["items"] + get("/heart/v1/actions", key)["items"]) except OSError: break for e in msgs: errs = [x.message[:160] for x in validator.iter_errors(e)] if errs: invalid.append((e.get("type"), errs[0])) public = get("/heart/v1/properties")["items"] + get("/heart/v1/arrhythmias")["items"] + [get("/heart/v1/pulse")] leaks += [e["type"] for e in public if e["heartprivacy"] != "public"] leaks += [k for e in get("/heart/v1/arrhythmias")["items"] for k in e["data"] if k not in ("id", "code", "severity", "status", "detectedAt", "resolvedAt")] if "capabilities" in get("/.well-known/heart-card.json")["data"]: leaks.append("card capabilities") pages.append(e2e.http("GET", f"/device/{dev}", cookie=owner)[2]) time.sleep(10) check("every served message is valid against the official schemas (card, pulse, alerts, properties, rules, control, actions)", not invalid, str(invalid[:3])) check("public views carry only public data", not leaks, str(sorted(set(leaks)))) check("Playbooks, Actions and the visit timeline need the owner key", all(call("GET", p)[0] == 403 for p in ("/heart/v1/tending/rules", "/heart/v1/actions", "/heart/v1/visits")) and call("GET", "/heart/v1/visits", key)[0] == 200) rules = [r["data"] for r in get("/heart/v1/tending/rules", key)["items"]] control = get("/heart/v1/tending/autonomy", key)["data"] entry = next((e for e in control.get("allowlist", []) if e["target"] == "camera.zoom"), {}) envz = entry.get("envelope", {}) check("both Playbooks are owner-adopted and enabled, and the zoom is allowlisted for rule only, with its envelope", sorted(r["id"] for r in rules) == sorted(RULES) and all(r["origin"] == "owner" and r["status"] == "adopted" and r["enabled"] for r in rules) and entry.get("initiators") == ["rule"] and envz.get("maxPerDay") == ENV["maxPerDay"] and envz.get("cooldownSec") == ENV["cooldownSec"] and envz.get("arguments", {}).get("zoom") == {k: ENV[k] for k in ("min", "max", "maxStep")}, str(control)) actions = [a["data"] for a in get("/heart/v1/actions", key)["items"]] auto = [a for a in actions if a["initiator"]["kind"] == "rule"] ran = sorted((a for a in auto if a["status"] == "succeeded"), key=lambda a: a["result"]["simSec"]) refused = [a for a in auto if a["status"] != "succeeded"] fired = {a["initiator"]["id"] for a in ran} check("both zoom Playbooks fired and changed the zoom", fired == set(RULES), f"ran {sorted(fired)}") bad = [a for a in ran if not (ENV["min"] <= a["result"]["zoomAfter"] <= ENV["max"]) or abs(a["result"]["zoomAfter"] - a["result"]["zoomBefore"]) > ENV["maxStep"] + 1e-9 or a["arguments"]["zoom"] != a["result"]["zoomAfter"] or a["risk"] != "write" or a["physical"] is not False] gaps = [b["result"]["simSec"] - a["result"]["simSec"] for a, b in zip(ran, ran[1:])] check("every automatic zoom change stayed inside the envelope (1.0 to 4.0, step at most 1.0, 300 s apart, 12 a day)", not bad and all(g >= ENV["cooldownSec"] for g in gaps) and len(ran) <= ENV["maxPerDay"], f"bad {bad[:2]} gaps {gaps}") check("refused automatic changes are recorded with the reason, and nothing was clipped", all(a["status"] == "denied" and a.get("error", "").startswith("Refused") and "proposed" in a["result"] and a["arguments"]["zoom"] == a["result"]["proposed"] for a in refused), str([a for a in refused if a["status"] != "denied"][:2])) print(f" ({len(ran)} automatic zoom changes ran, {len(refused)} refused; " f"{sum(1 for a in actions if a['initiator']['kind'] == 'human')} owner Actions)", flush=True) check("other changes are refused (403 capability-not-allowed)", all(call(m_, p_, key)[0] == 403 for m_, p_ in (("PUT", "/heart/v1/properties"), ("PUT", "/heart/v1/tending/rules/x"), ("PUT", "/heart/v1/tending/allowlist"), ("DELETE", "/heart/v1/tending/rules/x")))) shown = set(re.findall(r"(?:system|storage|power|performance|connectivity)\.[a-z-]+", " ".join(pages))) missing = [c for c in STAGED if c not in shown] check("every staged fault shows on the owner's page with its fault code", not missing, f"missing {missing}") found = [w for w in NEVER_ON_SITE if any(w in p for p in pages)] check("no reading, bird count, egg count or zoom reaches the site", not found, f"found {found}") public_page = e2e.http("GET", f"/device/{dev}")[0] check("the device page is private until shared", public_page in (403, 404), f"status {public_page}") finally: if sim: try: sim.wait(timeout=RUN_SEC + 60) except subprocess.TimeoutExpired: sim.kill() site.terminate() site.wait(timeout=10) print(f"\n{sum(results)}/{len(results)} passed (simulator log: {log_path})") return 0 if results and all(results) else 1 if __name__ == "__main__": sys.exit(main())