#!/usr/bin/env python3 # SPDX-License-Identifier: Apache-2.0 """End to end: the greenhouse simulator reports to a HEART Monitor site running on this machine, and runs the brain-proposes, HEART-decides loop of Core Standard Section 7. 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, signs in with an emailed code, registers the device through the site's own form, and checks for about four and a half minutes that: * every message the simulator serves is valid against the official schemas (spec/schemas/1.0), and so is every Consultation (consultation.schema.json), Proposal and Trial; * the public views hold nothing but public data, and Tending, Trials and the Rhythm need the owner key; * the mock brain's Proposal waits for a person: nothing runs before the button is pressed, an approval over the network with the key alone is refused, and the approval digest covers every step and the undo; * Proposals that break the rules are refused with the right problem (400, 401, 403, 404, 409, 422); * one Trial is kept and becomes a proposed learned Playbook that never runs; one is undone by the hot spell, restoring the stored value; Stop undoes a running Trial and then refuses new Proposals; * every Trial moved only along the allowed transitions, and every step stayed inside the envelope and its waitSec; * every staged fault shows on the owner's device page with its standard fault code; * no reading, schedule, Proposal or door visit ever reaches the site, and the page stays private. Beside it, on its own ports, runs the brain switch scenario (D36, API Standard 7.4): the greenhouse gets its brain only through a Command Center (the same site with HEART_TEST_CC=1) on a 30-second lease, with the MOCK brain behind the simulator's upstream gateway (--mock-gateway) and a proxy of this test as the network between them. It checks a forged answer is rejected and counted, the owner's cut undoes the brain's running Trial within one lease while watering goes on, a restore on the Command Center still waits for the owner's button, and a lost link makes the greenhouse brainless by itself when the lease runs out (ai.lease-expired). `--brain-only` runs it alone. Needs: Node 20+, Python 3 with jsonschema and referencing (pip install "jsonschema[format]>=4.18" referencing). Usage, from the project's top folder: python3 greenhouseSimulator/test_e2e.py """ import base64 import hashlib import json import os import pathlib import re import secrets import subprocess import sys import tempfile import threading import time import urllib.error import urllib.request import uuid from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer HERE = pathlib.Path(__file__).resolve().parent ROOT = HERE.parent sys.path.insert(0, str(ROOT / "heart-monitor" / "test")) sys.path.insert(0, str(HERE)) import e2e # noqa: E402 the monitor site's own end-to-end helpers (sign in, forms, outbox) import greenhouseSimulator as gs # noqa: E402 its canonical JSON, Ed25519 and signature check, for the brain scenario import jsonschema # noqa: E402 from referencing import Registry, Resource # noqa: E402 SITE_PORT, API_PORT, RUN_SEC, SPEED = 8803, 43303, 280, 5 e2e.SITE_PORT, e2e.SITE = SITE_PORT, f"http://127.0.0.1:{SITE_PORT}" SYSTEM_ID = "org.heartaispec.demo.greenhouse" EXT = SYSTEM_ID STAGED = ["connectivity.link-lost", "sensors.stuck", "actuators.not-responding", "performance.overshoot", "performance.too-hot", "system.main-function-degraded", "system.main-function-lost"] NEVER_ON_SITE = ["soil_moisture", "minutes_per_day", "watering_minutes", "tank_level", "mock-brain", "greenhouse.door", "set_watering", "Cel"] ENVELOPE = {"min": 15, "max": 45, "maxStep": 3} # the owner's envelope the simulator declares ALLOWED = {"awaiting-approval": {"running", "refused", "stopped"}, "running": {"kept", "undoing", "refused"}, "undoing": {"undone", "stopped", "undo-failed"}} # Core Standard 7.5 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)) def validator(name): return jsonschema.Draft202012Validator(json.loads((ROOT / "spec/schemas/1.0" / name).read_text()), registry=registry) envelope_v, consultation_v = validator("envelope.schema.json"), validator("consultation.schema.json") 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, body=None, key=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): return call("GET", path, key=key)[1] def canonical(obj): return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode() def proposal(target, plan, **change): """A Proposal as a remote brain would send it: no current values, only the change and how to check it.""" p = {"id": str(uuid.uuid4()), "target": target, "arguments": {"minutes_per_day": plan[-1]}, "steps": [{"arguments": {"minutes_per_day": m}, **({"waitSec": 120} if i < len(plan) - 1 else {})} for i, m in enumerate(plan)], "expect": [{"signal": "soil_moisture_zone3", "feature": "mean", "op": ">=", "value": 30, "windowSec": 240, "withinSec": 360}], "undoIf": {"any": [{"signal": "soil_moisture_zone3", "feature": "value", "op": "<", "value": 15, "withinSec": 360}]}, "basis": "Test brain: water zone 3 a little less.", "brain": {"tier": "cloud", "model": "test"}, "createdAt": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())} p.update(change) return {k: v for k, v in p.items() if v is not None} def prop_value(name, key): items = get("/heart/v1/properties", key)["items"] return next(e["data"]["value"] for e in items if e["data"]["name"] == f"{SYSTEM_ID}.{name}") def trials(key): return [e["data"] for e in get("/heart/v1/tending/trials", key)["items"]] def trial_on(target, key, states=None): found = [t for t in trials(key) if t["target"] == target and (states is None or t["state"] in states)] return found[0] if found else None def monitor_scenario(): work = pathlib.Path(tempfile.mkdtemp(prefix="heart-greenhouse-")) 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 / "greenhouseSimulator.py"), "--monitor-url", e2e.SITE, "--scenario", "all", "--speed", str(SPEED), "--seed", "7", "--duration", str(RUN_SEC), "--port", str(API_PORT), "--state-file", str(work / "greenhouse-monitor.json")], stdout=log, stderr=log, cwd=work) start = time.time() 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": "Greenhouse 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) brain = e2e.until(lambda: re.search(r"scope heart\.propose only\): (\S+)", text()), 10, "brain key").group(1) press = lambda: (work / "button.press").touch() # noqa: E731 a person at the greenhouse presses the button # --- Trial A: the mock brain's Proposal for zone 1, approved at the greenhouse, kept. a = e2e.until(lambda: trial_on("set_watering_zone1", key, ["awaiting-approval"]), 30, "the brain's Proposal for zone 1") acts = [e["data"] for e in get("/heart/v1/actions", key)["items"]] first = next(x for x in acts if x.get("trialId") == a["id"]) time.sleep(3) # some 15 simulated seconds: nothing may happen while it waits a = trial_on("set_watering_zone1", key) first = get(f"/heart/v1/actions/{first['id']}", key)["data"] check("the brain's Proposal waits for a person: nothing ran before approval", a["state"] == "awaiting-approval" and not a["steps"] and "startedAt" not in a and prop_value("watering_minutes_zone1", key) == 30 and first["status"] == "awaiting-approval" and first["initiator"]["kind"] == "ai" and first["physical"] is True and first["risk"] == "destructive", json.dumps({"trial": a.get("state"), "steps": a.get("steps"), "action": first.get("status")})) props = [e["data"] for e in get("/heart/v1/tending/proposals", key)["items"]] pa = next(p for p in props if p["id"] == a["proposalId"]) digest = hashlib.sha256(canonical({"system": SYSTEM_ID, "id": first["id"], "operation": a["target"], "arguments": a["plan"][0], "physical": True, "plan": a["plan"], "restore": a["previous"]})).hexdigest() check("the HEART read the value to restore itself, and one digest covers every step and the undo", a["previous"] == {"minutes_per_day": 30} and first["digest"] == digest and len(a["plan"]) == 2 and a["plan"][-1] == pa["arguments"] and "previous" not in pa and "current" not in json.dumps(pa), f"{a['previous']} {first['digest']} {digest}") s1 = call("POST", f"/heart/v1/actions/{first['id']}/decision", {"decision": "approve"}, key) s2 = call("POST", f"/heart/v1/actions/{first['id']}/decision", {"decision": "approve", "presence": {"method": "device-button", "digest": digest, "proof": "pressed"}}, key) check("approving over the network with the key alone is refused (403 presence-required)", s1[0] == 403 and s1[1].get("code") == "presence-required" and s2[0] == 403 and s2[1].get("code") == "presence-invalid", f"{s1} {s2}") refusals = { "no key (401)": (call("POST", "/heart/v1/tending/proposals", {"proposal": proposal("set_watering_zone3", [27])}), 401, "unauthenticated"), "owner key, not a brain token (403)": (call("POST", "/heart/v1/tending/proposals", {"proposal": proposal("set_watering_zone3", [27])}, key), 403, "insufficient-clearance"), "no undoIf (400)": (call("POST", "/heart/v1/tending/proposals", {"proposal": proposal("set_watering_zone3", [27], undoIf=None)}, brain), 400, "invalid-request"), "unknown target (404)": (call("POST", "/heart/v1/tending/proposals", {"proposal": proposal("open_roof_vents", [27])}, brain), 404, "not-found"), "target closed to ai (403)": (call("POST", "/heart/v1/tending/proposals", {"proposal": proposal("flush_lines", [27])}, brain), 403, "caller-not-allowed"), "outside the declared range (422)": (call("POST", "/heart/v1/tending/proposals", {"proposal": proposal("set_watering_zone3", [80])}, brain), 422, "out-of-range"), "a step bigger than maxStep (422)": (call("POST", "/heart/v1/tending/proposals", {"proposal": proposal("set_watering_zone3", [20])}, brain), 422, "out-of-range"), "a second Trial on a live target (409)": (call("POST", "/heart/v1/tending/proposals", {"proposal": proposal("set_watering_zone1", [27])}, brain), 409, "conflict"), } bad = {k: (r[0], r[1].get("code")) for k, (r, st, cd) in refusals.items() if (r[0], r[1].get("code")) != (st, cd)} check("Proposals that break the rules are refused with the right problem", not bad, str(bad)) check("nothing ran after the refused approvals either", trial_on("set_watering_zone1", key)["state"] == "awaiting-approval" and prop_value("watering_minutes_zone1", key) == 30) press() invalid, leaks, pages, bad_consult, seen = [], [], [], [], {} stage, b_id, c_id, notes, next_poll = "A", None, None, {}, 0.0 end = start + RUN_SEC - 15 while time.time() < end and sim.poll() is None: try: for t in trials(key): seen[t["id"]] = t if stage == "A": t = trial_on("set_watering_zone1", key) if t["state"] in ("kept", "undone", "refused", "stopped", "undo-failed"): notes["A"], stage = t, "B" elif stage == "B": t = trial_on("set_watering_zone2", key) if t and t["state"] == "awaiting-approval" and b_id is None: b_id = t["id"] press() elif t and t["state"] in ("kept", "undone", "refused", "stopped", "undo-failed"): notes["B"], notes["B-minutes"], stage = t, prop_value("watering_minutes_zone2", key), "C" elif stage == "C": r = call("POST", "/heart/v1/tending/proposals", {"proposal": proposal("set_watering_zone3", [27, 24])}, brain) notes["C-202"] = r[0] == 202 and r[1]["data"]["status"] == "awaiting-approval" c_id = r[1].get("data", {}).get("trialId") press() e2e.until(lambda: get(f"/heart/v1/tending/trials/{c_id}", key)["data"]["steps"], 15, "the first step of Trial C") notes["C-27"] = prop_value("watering_minutes_zone3", key) notes["stop"] = call("POST", "/heart/v1/tending/stop", None, key) notes["C"] = get(f"/heart/v1/tending/trials/{c_id}", key)["data"] notes["C-30"] = prop_value("watering_minutes_zone3", key) notes["after-stop"] = call("POST", "/heart/v1/tending/proposals", {"proposal": proposal("set_watering_zone3", [27])}, brain) notes["by-brain"] = call("PUT", "/heart/v1/tending/autonomy", {"max": 3}, brain) notes["resume"] = call("PUT", "/heart/v1/tending/autonomy", {"max": 3}, key) notes["control"] = get("/heart/v1/tending/autonomy", key)["data"] notes["mode"] = notes["control"]["mode"] stage = "faults" if time.time() >= next_poll: next_poll = time.time() + 10 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/rhythm", key)["items"] + get("/heart/v1/actions", key)["items"] + get("/heart/v1/tending/proposals", key)["items"] + get("/heart/v1/tending/trials", key)["items"] + get("/heart/v1/tending/rules", key)["items"]) for e in msgs: errs = [x.message[:160] for x in envelope_v.iter_errors(e)] if errs: invalid.append((e.get("type"), errs[0])) for c in get("/heart/v1/tending/consultations", key)["items"]: bad_consult += [x.message[:160] for x in consultation_v.iter_errors(c)] 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")] leaks += ["capabilities"] if "capabilities" in get("/.well-known/heart-card.json")["data"] else [] pages.append(e2e.http("GET", f"/device/{dev}", cookie=owner)[2]) except OSError: break time.sleep(1) A, B, C = notes.get("A") or {}, notes.get("B") or {}, notes.get("C") or {} rules = [e["data"] for e in get("/heart/v1/tending/rules", key)["items"]] if sim.poll() is None else [] all_trials = list(seen.values()) check("every served message is valid against the official schemas", not invalid, str(invalid[:3])) check("every Consultation is valid against consultation.schema.json, and there was one", not bad_consult and bool(get("/heart/v1/tending/consultations", key)["items"]), str(bad_consult[:3])) check("public views carry only public data", not leaks, str(sorted(set(leaks)))) check("Tending, Trials, Proposals and the Rhythm need the owner key", all(call("GET", p)[0] == 403 for p in ("/heart/v1/tending/trials", "/heart/v1/tending/proposals", "/heart/v1/tending/consultations", "/heart/v1/rhythm", "/heart/v1/actions")) and call("POST", "/heart/v1/tending/stop")[0] == 401) check("direct changes are refused: the schedule changes only in an approved Trial", all(call(m_, "/heart/v1/properties", {"value": 60}, key)[0] == 403 for m_ in ("PUT", "POST", "PATCH", "DELETE"))) check("Trial A (zone 1, mock brain) ran only after the button, and was kept", A.get("state") == "kept" and A["ext"][EXT]["history"] == ["awaiting-approval", "running", "kept"] and A.get("approvedBy", {}).get("kind") == "human" and len(A["steps"]) == 2 and all(s["guard"] == "passed" for s in A["steps"]) and all(c["held"] for c in A["steps"][-1].get("checks", [{}])), json.dumps({k: A.get(k) for k in ("state", "reason")})) first = get(f"/heart/v1/actions/{A.get('approvedBy', {}).get('actionId')}", key).get("data", {}) if A else {} check("the approval was a press at the greenhouse, before any step", first.get("decision", {}).get("presenceMethod") == "device-button" and all(s["at"] >= first["decision"]["at"] for s in A.get("steps", [])), json.dumps(first.get("decision"))) lr = [r for r in rules if r.get("id") == A.get("ruleProposed")] check("a kept Trial becomes a PROPOSED learned Playbook that never runs", len(lr) == 1 and lr[0]["origin"] == "learned" and lr[0]["status"] == "proposed" and lr[0]["enabled"] is False and lr[0]["evidence"]["trials"] == [A["id"]] and lr[0]["when"]["op"] == ">" and lr[0]["when"]["value"] == 45 and lr[0]["envelope"]["arguments"]["minutes_per_day"] == {"min": 24, "max": 30, "maxStep": 3} and not any("ruleId" in t for t in all_trials) and call("PUT", f"/heart/v1/tending/rules/{lr[0]['id']}", {"status": "adopted"}, key)[0] == 501, json.dumps(lr)) check("the hot spell undid Trial B (zone 2) and restored the stored value", B.get("state") == "undone" and "undoIf held" in B.get("reason", "") and notes.get("B-minutes") == 30 and B["ext"][EXT]["history"] == ["awaiting-approval", "running", "undoing", "undone"] and len(B["steps"]) >= 1, json.dumps({k: B.get(k) for k in ("state", "reason")}) + f" minutes {notes.get('B-minutes')}") check("Stop undid a running Trial (zone 3), then refused new Proposals until a person resumes", notes.get("C-202") and notes.get("C-27") == 27 and notes.get("stop", (0,))[0] == 200 and C.get("state") == "stopped" and notes.get("C-30") == 30 and C["ext"][EXT]["history"] == ["awaiting-approval", "running", "undoing", "stopped"] and notes["after-stop"][0] == 409 and notes["after-stop"][1].get("code") == "autonomy-stopped" and notes["resume"][0] == 202 and notes["resume"][1]["data"]["status"] == "awaiting-approval" and notes["resume"][1]["data"]["operation"] == "PUT /heart/v1/tending/autonomy" and notes["control"]["max"] == 2 and notes["mode"] == "observe" and notes["by-brain"][0] == 403 and notes["by-brain"][1].get("code") == "capability-not-allowed", json.dumps({k: notes.get(k) for k in ("C-202", "C-27", "C-30", "control", "by-brain")}) + f" {C.get('state')}") moves = [(t["id"][:8], a_, b_) for t in all_trials for a_, b_ in zip(t["ext"][EXT]["history"], t["ext"][EXT]["history"][1:]) if b_ not in ALLOWED.get(a_, set())] check("every Trial moved only along the allowed transitions", not moves and len(all_trials) >= 3, str(moves)) out = [] for t in all_trials: last, sec, waits = t["previous"]["minutes_per_day"], t["ext"][EXT]["stepSimSec"], t["ext"][EXT]["waitSec"] for i, s in enumerate(t["steps"]): v = s["arguments"]["minutes_per_day"] if not ENVELOPE["min"] <= v <= ENVELOPE["max"] or abs(v - last) > ENVELOPE["maxStep"]: out.append((t["target"], last, v)) if i and sec[i] - sec[i - 1] < waits[i - 1]: out.append((t["target"], "waitSec", sec[i] - sec[i - 1])) last = v check("every step stayed inside the envelope (min, max, maxStep) and waited its waitSec", not out, str(out)) shown = set(re.findall(r"(?:system|sensors|power|connectivity|performance|actuators)\.[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, schedule, Proposal or door visit 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"\nthe monitor scenario: {sum(results)}/{len(results)} passed (simulator log: {log_path})", flush=True) # ------------------------------------------------------------------------------------------------ the brain switch (D36) # The greenhouse gets its brain only through a Command Center (heart-monitor, HEART_TEST_CC=1), on a 30 s lease. The # MOCK brain sits behind the simulator's small upstream Reasoning Gateway (--mock-gateway), so every Proposal reaches # the greenhouse signed by the Command Center. Between the greenhouse and the Command Center sits a PROXY run by this # test: it is "the network", which can forge one answer (an attacker on the path) or go silent (the link is lost). CC_PORT, GW_PORT, PX_PORT, BRAIN_API_PORT, BRAIN_SPEED, LEASE_SEC = 8804, 8805, 8806, 43304, 6, 30 TEST_SITE_SEED = "dGVzdC1vbmx5LXNpdGUta2V5LW5ldmVyLWluLXByb2Q" # serve.mjs's test-only SITE_SIGNING_KEY MODEL = {"id": "mock/greenhouse", "version": "1"} def _b64u(b): return base64.urlsafe_b64encode(b).rstrip(b"=").decode() def _unb64u(s): return base64.urlsafe_b64decode(s + "=" * (-len(s) % 4)) def _ed_secret(seed): h = hashlib.sha512(seed).digest() a = int.from_bytes(h[:32], "little") & ((1 << 254) - 8) | (1 << 254) return a, h[32:] def _ed_point(p): x, y, z, _ = p zi = pow(z, gs._P - 2, gs._P) x, y = x * zi % gs._P, y * zi % gs._P return (y | (x & 1) << 255).to_bytes(32, "little") def ed_public(seed): return _ed_point(gs._times(_ed_secret(seed)[0], gs._BASE)) def ed_sign(seed, msg): """Ed25519 (RFC 8032) signing, standard library only: for the test's attacker key and its test signals.""" a, prefix = _ed_secret(seed) pub = _ed_point(gs._times(a, gs._BASE)) r = int.from_bytes(hashlib.sha512(prefix + msg).digest(), "little") % gs._L big_r = _ed_point(gs._times(r, gs._BASE)) k = int.from_bytes(hashlib.sha512(big_r + pub + msg).digest(), "little") % gs._L return big_r + ((r + k * a) % gs._L).to_bytes(32, "little") class Network: """The path between the greenhouse and the Command Center. pass: forwards. forge: replaces the next brain answer with another Proposal, re-digested and signed with an attacker's key. blocked: answers nothing at all.""" def __init__(self, port, upstream): self.upstream, self.blocked, self.forge, self.log, self.forged = upstream, False, False, [], [] self.attacker = secrets.token_bytes(32) net = self class H(BaseHTTPRequestHandler): def do_POST(self): body = self.rfile.read(int(self.headers.get("Content-Length") or 0)) if net.blocked: # silence: the device's own 10 s timeout ends the wait end = time.time() + 15 while net.blocked and time.time() < end: time.sleep(0.2) self.close_connection = True return req = urllib.request.Request(net.upstream + self.path, data=body, method="POST", headers={k: v for k, v in self.headers.items() if k.lower() in ("content-type", "authorization")}) try: with urllib.request.urlopen(req, timeout=20) as r: status, raw = r.status, r.read() except urllib.error.HTTPError as e: status, raw = e.code, e.read() if self.path.endswith("/reason") and status == 200 and net.forge: net.forge, raw = False, net.forged_answer(raw) try: reply = json.loads(raw or b"{}") except ValueError: reply = {} net.log.append({"at": time.time(), "path": self.path, "request": json.loads(body or b"{}"), "status": status, "reply": reply}) self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(raw))) self.end_headers() self.wfile.write(raw) def log_message(self, *a): pass self.srv = ThreadingHTTPServer(("127.0.0.1", port), H) threading.Thread(target=self.srv.serve_forever, daemon=True).start() def forged_answer(self, raw): genuine = json.loads(raw) signed = genuine.pop("signed") bad = dict(genuine, id=str(uuid.uuid4()), arguments={"minutes_per_day": 45}, steps=[{"arguments": {"minutes_per_day": 45}}], basis="Water much more (FORGED on the path).") digest = hashlib.sha256(gs.jcs(bad).encode()).hexdigest() body = {"system": SYSTEM_ID, "lease": signed["lease"], "requestDigest": signed["requestDigest"], "answerDigest": digest, "model": signed["model"], "at": signed["at"]} bad["signed"] = dict(signed, answerDigest=digest, signature=_b64u(ed_sign(self.attacker, gs.jcs(body).encode()))) self.forged.append(bad["id"]) return json.dumps(bad).encode() def seen(self, what, since=0.0): return [x for x in self.log if x["path"].endswith("/" + what) and x["at"] >= since] def brain_main(): global API_PORT API_PORT = BRAIN_API_PORT e2e.SITE_PORT, e2e.SITE = CC_PORT, f"http://127.0.0.1:{CC_PORT}" t0 = time.time() work = pathlib.Path(tempfile.mkdtemp(prefix="heart-greenhouse-brain-")) mon = ROOT / "heart-monitor" if (mon / "worker/worker.js").stat().st_mtime < (mon / "src/worker.src.js").stat().st_mtime: subprocess.run(["node", str(mon / "tools/build-worker.mjs")], check=True, stdout=subprocess.DEVNULL) env = dict(os.environ, HEART_TEST_CC="1", HEART_TEST_CC_PROVIDER=f"http://127.0.0.1:{GW_PORT}", HEART_TEST_CC_MODELS=json.dumps([dict(MODEL, uses=["explain", "propose"])])) site = subprocess.Popen(["node", str(mon / "test/serve.mjs"), str(CC_PORT)], env=env, stderr=subprocess.DEVNULL) net = Network(PX_PORT, e2e.SITE) via = f"http://127.0.0.1:{PX_PORT}" log_path, sim, timings = work / "simulator.log", None, {} site_pub = _b64u(ed_public(_unb64u(TEST_SITE_SEED))) try: e2e.until(lambda: _site_up(), 20, "the Command Center site") with open(log_path, "w") as f: sim = subprocess.Popen([sys.executable, str(HERE / "greenhouseSimulator.py"), "--monitor-url", via, "--command-center", via, "--mock-gateway", str(GW_PORT), "--speed", str(BRAIN_SPEED), "--seed", "7", "--duration", "300", "--port", str(API_PORT), "--state-file", str(work / "greenhouse-monitor.json")], stdout=f, stderr=f, 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") dev, code = m.group(1), m.group(2) key = e2e.until(lambda: re.search(r"owner key for private views: (\S+)\)", text()), 10, "owner key").group(1) brain_key = e2e.until(lambda: re.search(r"scope heart\.propose only\): (\S+)", text()), 10, "brain key").group(1) e2e.until(lambda: "knows this device" in text(), 20, "the hello") owner = e2e.login("owner@example.com") s = e2e.http("POST", "/register", {"id": dev, "code": code.lower(), "name": "Greenhouse brain demo"}, cookie=owner)[0] check("brain: the owner registers the greenhouse at the Command Center (its monitor site)", s in (200, 303), f"status {s}") press = lambda: (work / "button.press").touch() # noqa: E731 bview = lambda k=None: get("/heart/v1/brain", k) # noqa: E731 state = lambda: bview(key)["data"] # noqa: E731 alerts = lambda: [e["data"] for e in get("/heart/v1/arrhythmias", key)["items"]] # noqa: E731 open_codes = lambda: {a["code"] for a in alerts() if a["status"] != "resolved"} # noqa: E731 pulse_limit = lambda: get("/heart/v1/pulse", key)["data"]["autonomy"].get("limitedBy") # noqa: E731 msgs = [] # --- No brain yet: brainless, so the open problem is handled by rules alone, and a Proposal is refused. e2e.until(lambda: "performance.overshoot" in open_codes(), 40, "a zone wetter than its band") time.sleep(3) r409 = call("POST", "/heart/v1/tending/proposals", {"proposal": proposal("set_watering_zone3", [27])}, brain_key) card = get("/.well-known/heart-card.json")["data"] st0, pub0 = state(), bview() msgs += [bview(), bview(key), get("/.well-known/heart-card.json", key)] check("brain: with no lease it is brainless and consults nobody, though a zone is too wet", st0["state"] == "brainless" and st0["why"] == "no-lease" and pub0["data"] == {"state": "brainless", "why": "no-lease"} and pub0["heartprivacy"] == "public" and "brain" in card["modules"] and pulse_limit() == "brain-cut" and not get("/heart/v1/tending/consultations", key)["items"] and not net.seen("reason") and r409[0] == 409 and r409[1].get("code") == "conflict" and st0["rejectedAnswers"] == 0 and call("POST", "/heart/v1/brain/restore", None, key)[0] == 409, json.dumps({"state": st0, "public": pub0["data"], "post": r409[0], "limitedBy": pulse_limit()})) # --- The owner gives it a brain on the Command Center. The network forges the first answer. net.forge = True form = [("on", "1"), ("model", f"{MODEL['id']}@{MODEL['version']}"), ("use", "explain"), ("use", "propose"), ("use", "plan"), ("lease_sec", str(LEASE_SEC)), ("offline_sec", "0")] t_on = time.time() s = e2e.http("POST", f"/commandcenter/device/{dev}/brain", form, cookie=owner)[0] e2e.until(lambda: state()["state"] == "leased", 20, "the first lease") timings["first lease after the owner turned the brain on"] = time.time() - t_on lease_reply = [x for x in net.seen("lease") if "lease" in x["reply"]][-1]["reply"] st = state() check("brain: the owner turns the brain on; the greenhouse takes a lease signed with the key it pinned at hello", s == 200 and st["lease"]["id"] == lease_reply["lease"]["lease"] and st["lease"]["models"] == [MODEL] and set(st["lease"]["uses"]) == {"explain", "propose", "plan"} and pulse_limit() is None and gs.signed_by(site_pub, lease_reply["lease"], lease_reply["signature"]) and "why" not in st, json.dumps(st)) # (d) The forged answer: signed, but with another key. Rejected, counted, never used. e2e.until(lambda: state()["rejectedAnswers"] >= 1, 30, "the forged answer to be rejected") forged = net.forged[0] if net.forged else None r403 = call("POST", "/heart/v1/tending/proposals", {"proposal": proposal("set_watering_zone3", [27])}, brain_key) names = [e["data"]["name"] for e in get("/heart/v1/rhythm", key)["items"]] props = [e["data"] for e in get("/heart/v1/tending/proposals", key)["items"]] check("brain (d): an answer signed with another key is rejected, counted (ai.answer-rejected) and not used; " "so is an unsigned Proposal", forged is not None and forged not in [p["id"] for p in props] and not any(t["plan"][-1] == {"minutes_per_day": 45} for t in trials(key)) and "ai.answer-rejected" in open_codes() and "ai.answer-rejected" in names and r403[0] == 403 and r403[1].get("code") == "signature-invalid" and state()["rejectedAnswers"] == 2 and "the signature is not the Command Center's" in text() and state()["state"] == "leased", json.dumps({"forged": forged, "rejected": state()["rejectedAnswers"], "post": r403})) # The next answer is genuine: the Command Center signed it for exactly the Consultation that was sent. a = e2e.until(lambda: trial_on("set_watering_zone1", key, ["awaiting-approval"]), 60, "a signed Proposal for zone 1") p = next(x for x in (e["data"] for e in get("/heart/v1/tending/proposals", key)["items"]) if x["id"] == a["proposalId"]) sent = [x for x in net.seen("reason") if x["status"] == 200 and x["reply"].get("id") == p["id"]] cons = get("/heart/v1/tending/consultations", key)["items"] digest = lambda o: hashlib.sha256(gs.jcs(o).encode()).hexdigest() # noqa: E731 bare = {k: v for k, v in p.items() if k != "signed"} sig_body = {"system": SYSTEM_ID, "lease": p["signed"]["lease"], "requestDigest": p["signed"]["requestDigest"], "answerDigest": p["signed"]["answerDigest"], "model": p["signed"]["model"], "at": p["signed"]["at"]} check("brain: the MOCK brain's Proposal came through the Command Center, signed for exactly this Consultation and lease", len(sent) == 1 and sent[0]["request"]["request"] in cons and p["signed"]["requestDigest"] == digest(sent[0]["request"]["request"]) and p["signed"]["answerDigest"] == digest(bare) and p["signed"]["lease"] == state()["lease"]["id"] and p["signed"]["model"] == MODEL and gs.signed_by(site_pub, sig_body, p["signed"]["signature"]) and p["brain"]["tier"] == "cloud", json.dumps(p.get("signed"))) press() e2e.until(lambda: (lambda t: t["state"] == "running" and t["steps"])(trial_on("set_watering_zone1", key)), 15, "the Trial to start") check("brain: a person approved it at the greenhouse and the Trial runs (30 to 27 min/day)", prop_value("watering_minutes_zone1", key) == 27) # (a) The owner cuts the brain on the Command Center. n_props = len(get("/heart/v1/tending/proposals", key)["items"]) n_cons = len(get("/heart/v1/tending/consultations", key)["items"]) t_cut = time.time() s = e2e.http("POST", f"/commandcenter/device/{dev}/cut", {}, cookie=owner)[0] e2e.until(lambda: state()["state"] == "brainless", LEASE_SEC + 5, "brainless after the cut") t_off = time.time() timings["owner's cut until brainless"] = t_off - t_cut tr = get(f"/heart/v1/tending/trials/{a['id']}", key)["data"] st, water0 = state(), prop_value("water_used_today", key) msgs += [bview(), bview(key), get("/heart/v1/pulse", key)] check(f"brain (a): the owner's cut reaches the greenhouse within one lease ({t_off - t_cut:.1f} s <= {LEASE_SEC} s): " "brain-off, checked, and the running brain Trial is undone", s == 200 and t_off - t_cut <= LEASE_SEC and st["why"] == "brain-off" and "lease" not in st and tr["state"] == "stopped" and tr["ext"][EXT]["history"][-3:] == ["running", "undoing", "stopped"] and "brain was cut" in tr.get("reason", "") and prop_value("watering_minutes_zone1", key) == 30 and "ai.brain-revoked" in open_codes() and pulse_limit() == "brain-cut" and bview()["data"] == {"state": "brainless", "why": "brain-off"}, json.dumps({"state": st, "trial": tr["state"], "history": tr["ext"][EXT]["history"]})) time.sleep(15) # 90 greenhouse seconds without a brain off_net = [x for x in net.seen("reason", t_off)] check("brain (a): while brainless no brain is asked and no Proposal is used, though a zone is still too wet", not off_net and len(get("/heart/v1/tending/proposals", key)["items"]) == n_props and len(get("/heart/v1/tending/consultations", key)["items"]) == n_cons and "performance.overshoot" in open_codes() and state()["state"] == "brainless", f"reason calls {len(off_net)}, proposals {n_props}, open {sorted(open_codes())}") water1 = prop_value("water_used_today", key) check("brain (a): watering goes on, on the controller's schedule, and Minimum Tending still notices", water1 > water0 and get("/heart/v1/pulse", key)["data"]["status"] in ("degraded", "failing"), f"water {water0} -> {water1}") # (b) Restore on the Command Center (an emailed code), then the owner at the greenhouse. t_restore = time.time() e2e.http("POST", f"/commandcenter/device/{dev}/restore", {}, cookie=owner) mail = [m_ for m_ in json.loads(e2e.http("GET", "/__test/outbox")[2]) if m_["to"] == "owner@example.com"][-1] rcode = re.search(r"^ {4}(\d{6})$", mail["body"], re.M).group(1) s = e2e.http("POST", f"/commandcenter/device/{dev}/restore/verify", {"code": rcode}, cookie=owner)[0] e2e.until(lambda: state().get("why") == "restore-pending", 20, "restore-pending") e2e.until(lambda: len([x for x in net.seen("lease", t_restore) if "lease" in x["reply"]]) >= 3, 30, "three new leases offered") st = state() offered = [x for x in net.seen("lease", t_restore) if "lease" in x["reply"]] r401 = call("POST", "/heart/v1/brain/restore", None) r202 = call("POST", "/heart/v1/brain/restore", None, key) act = r202[1].get("data", {}) r403 = call("POST", f"/heart/v1/actions/{act.get('id')}/decision", {"decision": "approve"}, key) time.sleep(6) still = state() msgs += [bview(), bview(key), r202[1]] check("brain (b): after the Command Center restores, the greenhouse still refuses every new lease (restore-pending) " "until its owner approves at the greenhouse", s == 200 and st["state"] == "brainless" and len(offered) >= 3 and all(not x["request"]["restore"] for x in offered) and bview()["data"] == {"state": "brainless", "why": "restore-pending"} and r401[0] == 401 and r202[0] == 202 and act.get("status") == "awaiting-approval" and act.get("risk") == "destructive" and act.get("operation") == "POST /heart/v1/brain/restore" and r403[0] == 403 and still["state"] == "brainless" and still["why"] == "restore-pending", json.dumps({"state": st, "offered": len(offered), "restore": r202[0], "after": still})) t_press = time.time() press() e2e.until(lambda: state()["state"] == "leased", 20, "the lease after the owner's approval") timings["owner's press until leased"] = time.time() - t_press took = [x for x in net.seen("lease", t_press) if "lease" in x["reply"]] check("brain (b): the owner pressed the button; the next lease asks restore and is accepted, once", bool(took) and took[0]["request"]["restore"] is True and took[0]["reply"]["lease"].get("restore") is True and state()["lease"]["id"] == took[-1]["reply"]["lease"]["lease"] and "ai.brain-revoked" not in open_codes() and get(f"/heart/v1/actions/{act.get('id')}", key)["data"]["status"] == "succeeded" and pulse_limit() is None, json.dumps([x["request"] for x in took])) # Signals sent straight to the greenhouse: only a signed, fresh, new one is acted on, and it only lowers risk. def signal(kind, seed=_unb64u(TEST_SITE_SEED), age=0, sid=None, **extra): sg = {"system": SYSTEM_ID, "id": sid or str(uuid.uuid4()), "kind": kind, "at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(time.time() - age)), **extra} return {"signal": sg, "signature": _b64u(ed_sign(seed, gs.jcs(sg).encode()))} forged_sig = call("POST", "/heart/v1/brain/signal", signal("brain-off", seed=net.attacker)) other = signal("brain-off") other["signal"]["system"] = "org.example.other" other["signature"] = _b64u(ed_sign(_unb64u(TEST_SITE_SEED), gs.jcs(other["signal"]).encode())) other_sys = call("POST", "/heart/v1/brain/signal", other) stale = call("POST", "/heart/v1/brain/signal", signal("brain-off", age=400)) good = signal("revoke-use", use="explain") ok = call("POST", "/heart/v1/brain/signal", good) again = call("POST", "/heart/v1/brain/signal", good) st = state() check("brain: a signal is acted on only if signed with the pinned key, for this system, fresh and new", forged_sig[0] == 403 and forged_sig[1].get("code") == "signature-invalid" and other_sys[0] == 403 and stale[0] == 409 and ok[0] == 200 and again[0] == 409 and st["state"] == "leased" and "explain" not in st["lease"]["uses"], json.dumps([forged_sig[0], other_sys[0], stale[0], ok[0], again[0], st.get("lease", {}).get("uses")])) # (c) The network to the Command Center goes silent. No message is needed: the lease runs out by itself. e2e.until(lambda: net.seen("lease", time.time() - 1.5), LEASE_SEC, "a fresh renewal") # start just after one net.blocked = True t_block = time.time() last_ok = max(x["at"] for x in net.log if x["path"].endswith("/lease") and "lease" in x["reply"]) e2e.until(lambda: state()["state"] == "brainless", LEASE_SEC + 10, "brainless with the link lost") t_off = time.time() timings["link lost until brainless"] = t_off - t_block timings["last renewal until brainless (one lease)"] = t_off - last_ok st, codes_c, limit_c, r409_c = state(), open_codes(), pulse_limit(), call("POST", "/heart/v1/brain/restore", None, key)[0] unreachable = bool(e2e.until(lambda: "cannot be reached" in text(), 15, "the timed-out renewal in the log")) check(f"brain (c): with the Command Center unreachable the greenhouse goes brainless by itself when the lease ends " f"({t_off - last_ok:.1f} s after the last renewal), with ai.lease-expired", st["why"] == "lease-expired" and "ai.lease-expired" in codes_c and LEASE_SEC - 3 <= t_off - last_ok <= LEASE_SEC + 2.5 and limit_c == "brain-cut" and r409_c == 409 and unreachable, json.dumps({"state": st, "open": sorted(codes_c), "limitedBy": limit_c, "restore": r409_c, "sinceBlock": round(t_off - t_block, 1), "sinceRenewal": round(t_off - last_ok, 1)})) net.blocked = False t_back = time.time() e2e.until(lambda: state()["state"] == "leased", 30, "a lease once the link is back") timings["link back until leased"] = time.time() - t_back took = [x for x in net.seen("lease", t_back) if "lease" in x["reply"]] check("brain (c): a lease that merely ran out is not a cut: the next lease is taken without the owner", bool(took) and not took[-1]["reply"]["lease"].get("restore") and "ai.lease-expired" not in open_codes() and state()["rejectedAnswers"] == 2, json.dumps(state())) msgs += [bview(), bview(key), get("/heart/v1/pulse", key), get("/heart/v1/pulse"), get("/.well-known/heart-card.json", key)] msgs += (get("/heart/v1/arrhythmias", key)["items"] + get("/heart/v1/rhythm", key)["items"] + get("/heart/v1/actions", key)["items"] + get("/heart/v1/tending/proposals", key)["items"] + get("/heart/v1/tending/trials", key)["items"]) invalid = [(e.get("type"), x.message[:160]) for e in msgs for x in envelope_v.iter_errors(e)] bad_c = [x.message[:160] for c in get("/heart/v1/tending/consultations", key)["items"] for x in consultation_v.iter_errors(c)] moves = [(t["id"][:8], a_, b_) for t in trials(key) for a_, b_ in zip(t["ext"][EXT]["history"], t["ext"][EXT]["history"][1:]) if b_ not in ALLOWED.get(a_, set())] check("brain: every message served (brain views, Pulse, Card, Proposals, Trials, Actions, Rhythm) is valid against " "the official schemas, and every Trial moved only along allowed transitions", not invalid and not bad_c and not moves, str(invalid[:3] + bad_c[:3] + moves[:3])) check("brain: the simulator logged no error", "Traceback" not in text(), text()[-800:]) except AssertionError as e: check("brain: the scenario ran to the end", False, str(e)) finally: if sim: sim.terminate() try: sim.wait(timeout=10) except subprocess.TimeoutExpired: sim.kill() net.srv.shutdown() site.terminate() site.wait(timeout=10) for k, v in timings.items(): print(f" timing: {k}: {v:.1f} s") print(f"\nthe brain switch scenario: {sum(results)}/{len(results)} passed in {time.time() - t0:.0f} s (simulator log: {log_path})", flush=True) return 0 if results and all(results) else 1 def _site_up(): try: return e2e.http("GET", "/monitor")[0] == 200 except OSError: return False def main(): if "--brain-only" in sys.argv: return brain_main() # The brain switch scenario runs beside the monitor scenario, on its own ports, so the whole test takes no longer. out = tempfile.NamedTemporaryFile("w+", prefix="heart-greenhouse-brain-", suffix=".log", delete=False) brain = subprocess.Popen([sys.executable, str(HERE / "test_e2e.py"), "--brain-only"], stdout=out, stderr=subprocess.STDOUT, cwd=ROOT) print("== the monitor scenario (and, beside it, the brain switch scenario)", flush=True) try: monitor_scenario() finally: try: brain_rc = brain.wait(timeout=600) except subprocess.TimeoutExpired: brain.kill() brain_rc = 1 text = pathlib.Path(out.name).read_text() print("\n== the brain switch scenario (D36)\n" + text, flush=True) brain_results = [line.startswith("PASS") for line in text.splitlines() if line[:4] in ("PASS", "FAIL")] total = results + brain_results ok = bool(results) and bool(brain_results) and all(total) and brain_rc == 0 print(f"{sum(total)}/{len(total)} passed{'' if ok else ' (FAILED)'}", flush=True) return 0 if ok else 1 if __name__ == "__main__": sys.exit(main())