mirror of
https://github.com/openglow-org/forgefirm.git
synced 2026-09-27 16:51:12 -07:00
ScottW514
b90c53c739
The panel moved out of ui.c into src/ui/{index.html,panel.css,panel.js},
bundled into index_html[] by CMake. forgectrl.panel-serves now covers
src/ui/** and checks that the served page carries no external asset
reference - the bundle is what keeps the panel a single self-contained
response, and this is where that would break. Fetch-verified and
cross-built in the build tree (bitbake forgectrl).
235 lines
12 KiB
Python
235 lines
12 KiB
Python
"""forgectrl.* - the machine-services daemon's API, access control, and panel."""
|
|
import json
|
|
import socket
|
|
import time
|
|
|
|
from ..catalog import test
|
|
from .. import hw
|
|
|
|
_COVERS_AUTH = [("forgectrl", "src/auth.*"), ("forgectrl", "src/main.c")]
|
|
|
|
|
|
def lan_ip():
|
|
"""The board's own non-loopback IPv4 (the address a LAN client would
|
|
use), or None."""
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
try:
|
|
s.connect(("10.255.255.255", 9))
|
|
ip = s.getsockname()[0]
|
|
except OSError:
|
|
ip = None
|
|
finally:
|
|
s.close()
|
|
if ip and not ip.startswith("127."):
|
|
return ip
|
|
return None
|
|
|
|
|
|
@test("forgectrl.auth", title="API access control", subsystem="forgectrl", kind="auto", est_min=1,
|
|
covers=_COVERS_AUTH,
|
|
description="Every state-changing endpoint refuses an unauthenticated write; a non-literal "
|
|
"Host, a non-literal Origin and a cross-site Sec-Fetch-Site are refused; the "
|
|
"cooling report channel refuses a non-loopback peer; the fuse view is two-factor "
|
|
"(token and the physical button) and refused without either; "
|
|
"the flash and factory-restore chain is refused unauthenticated.")
|
|
def auth(ctx):
|
|
fc = ctx.forgectrl
|
|
ev = ctx.evidence
|
|
|
|
st, body = fc.get("/status")
|
|
ctx.log("GET /status -> %s", st)
|
|
ctx.check(st == 200, "GET /status -> %s", st)
|
|
|
|
# unauthenticated writes: every one must be refused before it acts
|
|
for path, params in (("/controller/stop", None), ("/controller/start", None),
|
|
("/mode", {"controller": "grbl"}), ("/settings", {"ui_units": "mm"}),
|
|
("/diag/flow-verify", None), ("/diag/abort", None),
|
|
("/update/apply", None), ("/boot", {"slot": "a"}),
|
|
("/system/reboot", None), ("/restore/factory", None)):
|
|
st, body = fc.post(path, params=params, auth=False)
|
|
ctx.log("POST %s (no token) -> %s %s", path, st, body if isinstance(body, dict) else "")
|
|
ev["noauth " + path] = st
|
|
ctx.check(st == 403, "POST %s without a token -> %s, expected 403", path, st)
|
|
ctx.check(isinstance(body, dict) and body.get("error") == "authentication required",
|
|
"POST %s without a token: unexpected body %r", path, body)
|
|
|
|
# the upload sink refuses during body parse; only the status is asserted
|
|
st, body = fc.post("/update/upload", data=b"not a firmware archive", auth=False,
|
|
headers={"Content-Type": "application/octet-stream"})
|
|
ev["noauth /update/upload"] = st
|
|
ctx.log("POST /update/upload (no token) -> %s", st)
|
|
ctx.check(st in (400, 403), "POST /update/upload without a token -> %s", st)
|
|
|
|
# origin checks (read endpoint, so only the origin layer decides)
|
|
st, body = fc.get("/status", headers={"Host": "evil.example.net"})
|
|
ev["host_name"] = st
|
|
ctx.log("GET /status Host=evil.example.net -> %s", st)
|
|
ctx.check(st == 403, "a DNS-name Host was accepted (%s)", st)
|
|
st, body = fc.get("/status", headers={"Origin": "http://evil.example.net"})
|
|
ev["origin_name"] = st
|
|
ctx.log("GET /status Origin=http://evil.example.net -> %s", st)
|
|
ctx.check(st == 403, "a DNS-name Origin was accepted (%s)", st)
|
|
st, body = fc.get("/status", headers={"Sec-Fetch-Site": "cross-site"})
|
|
ev["sfs_cross"] = st
|
|
ctx.log("GET /status Sec-Fetch-Site=cross-site -> %s", st)
|
|
ctx.check(st == 403, "a cross-site fetch was accepted (%s)", st)
|
|
st, body = fc.get("/status", headers={"Sec-Fetch-Site": "same-origin", "Origin": "http://127.0.0.1:8080"})
|
|
ctx.check(st == 200, "same-origin literal Origin refused (%s)", st)
|
|
|
|
# the fuse view is two-factor: the token AND the physical button held.
|
|
# Without the token: authentication refused; with the token and nobody
|
|
# at the button: refused with the button message. The identity itself is
|
|
# never fetched (it would land in this log).
|
|
st, body = fc.get("/fuse-identity", auth=False)
|
|
ev["fuse_noauth"] = st
|
|
ctx.log("GET /fuse-identity (no token) -> %s %s", st, body if isinstance(body, dict) else "")
|
|
ctx.check(st == 403 and isinstance(body, dict) and body.get("error") == "authentication required",
|
|
"GET /fuse-identity without token -> %s %r", st, body)
|
|
st, body = fc.get("/fuse-identity")
|
|
ev["fuse_token_no_button"] = st
|
|
ctx.log("GET /fuse-identity (token, button not held) -> %s %s", st, body if isinstance(body, dict) else "")
|
|
msg = body.get("error", "") if isinstance(body, dict) else str(body)
|
|
ctx.check(st == 403 and "button" in msg,
|
|
"GET /fuse-identity with the token but no button -> %s %r (expected the two-factor refusal)",
|
|
st, body)
|
|
|
|
# the cooling report channel: loopback only, even with a token
|
|
ip = lan_ip()
|
|
ev["lan_ip"] = ip
|
|
ctx.check(ip, "cannot determine the board's LAN address")
|
|
port = fc.base.rsplit(":", 1)[-1]
|
|
lan = hw.Forgectrl("http://%s:%s" % (ip, port), token=fc.token)
|
|
st, body = lan.post("/cool/state", params={"mode": "idle", "armed": "0"})
|
|
ev["cool_state_from_lan"] = st
|
|
ctx.log("POST /cool/state from %s -> %s %s", ip, st, body if isinstance(body, dict) else "")
|
|
ctx.check(st == 403 and isinstance(body, dict) and body.get("error") == "loopback only",
|
|
"/cool/state accepted a non-loopback peer (%s %r)", st, body)
|
|
|
|
|
|
@test("forgectrl.settings-bounds", title="Settings validation and restore", subsystem="forgectrl",
|
|
kind="auto", est_min=1,
|
|
covers=[("forgectrl", "src/settings.*"), ("forgectrl", "src/main.c"), ("forgectrl", "src/cam.c")],
|
|
description="An over-length value and an out-of-range value are refused (400) and leave the "
|
|
"settings byte-identical; an in-range value is accepted (200). The lid lamp "
|
|
"idles at lid_lamp_idle (unset = 236), an out-of-range level is refused, a new "
|
|
"level applies to the lamp at once, and clearing it returns the default.")
|
|
def settings_bounds(ctx):
|
|
fc = ctx.forgectrl
|
|
ev = ctx.evidence
|
|
before = fc.settings()
|
|
ev["keys"] = len(before)
|
|
|
|
st, body = fc.post("/settings", data={"gf_serial": "X" * 300})
|
|
ev["overlong"] = st
|
|
ctx.log("POST /settings gf_serial=<300 chars> -> %s %s", st, body if isinstance(body, dict) else "")
|
|
ctx.check(st == 400, "over-length value -> %s, expected 400", st)
|
|
|
|
st, body = fc.post("/settings", data={"laser_disarm_s": "99999"})
|
|
ev["out_of_range"] = st
|
|
ctx.log("POST /settings laser_disarm_s=99999 -> %s", st)
|
|
ctx.check(st == 400, "out-of-range value -> %s, expected 400", st)
|
|
|
|
st, body = fc.post("/settings", data={"no_such_key_forgetest": "1"})
|
|
ev["unknown_key"] = st
|
|
ctx.log("POST /settings no_such_key_forgetest=1 -> %s", st)
|
|
ctx.check(st in (400, 404), "unknown key -> %s, expected 400", st)
|
|
|
|
after = fc.settings()
|
|
ctx.check(json.dumps(after, sort_keys=True) == json.dumps(before, sort_keys=True),
|
|
"settings changed after refused writes")
|
|
ctx.log("settings unchanged after the refused writes")
|
|
|
|
# an accepted in-range write: rewrite a present key with its own value
|
|
key = None
|
|
for k in ("ui_units", "laser_disarm_s", "cool_flow_rise", "rail_settle_s"):
|
|
v = before.get(k)
|
|
if isinstance(v, str) and v != "":
|
|
key = k
|
|
break
|
|
if key is None:
|
|
key, val = "ui_units", "mm"
|
|
ctx.log("no settable key is present; writing %s=%s (recorded in evidence)", key, val)
|
|
else:
|
|
val = before[key]
|
|
st, body = fc.post("/settings", data={key: val})
|
|
ev["accepted"] = {"key": key, "value": val, "status": st}
|
|
ctx.log("POST /settings %s=%s -> %s", key, val, st)
|
|
ctx.check(st == 200, "in-range write -> %s, expected 200", st)
|
|
final = fc.settings()
|
|
others_before = {k: v for k, v in before.items() if k != key}
|
|
others_after = {k: v for k, v in final.items() if k != key}
|
|
ctx.check(others_before == others_after, "other settings changed by the write")
|
|
ctx.check(final.get(key) == val, "%s reads back %r, wrote %r", key, final.get(key), val)
|
|
|
|
# the lid lamp's idle level: resting at the setting, bounded, applied live
|
|
lamp_was = (before.get("lid_lamp_idle") or "").strip()
|
|
want = lamp_was or "236"
|
|
got = ctx.sysfs("pic/lid_led")
|
|
ev["lid_lamp"] = {"setting": lamp_was, "resting": got}
|
|
ctx.log("lid lamp: setting %r, pic/lid_led=%s (expected %s)", lamp_was, got, want)
|
|
ctx.check(got == want, "lid lamp rests at %s, lid_lamp_idle is %s", got, want)
|
|
for bad in ("256", "-1", "bright"):
|
|
st, body = fc.post("/settings", data={"lid_lamp_idle": bad})
|
|
ctx.check(st == 400, "lid_lamp_idle=%s -> %s, expected 400", bad, st)
|
|
ctx.log("lid_lamp_idle 256 / -1 / bright refused")
|
|
try_level = "100" if want != "100" else "120"
|
|
st, body = fc.post("/settings", data={"lid_lamp_idle": try_level})
|
|
ctx.check(st == 200, "lid_lamp_idle=%s -> %s, expected 200", try_level, st)
|
|
applied = None
|
|
t0 = time.time()
|
|
while time.time() - t0 < 5:
|
|
applied = ctx.sysfs("pic/lid_led")
|
|
if applied == try_level:
|
|
break
|
|
ctx.sleep(0.2)
|
|
ctx.log("lid_lamp_idle=%s -> pic/lid_led=%s after %.1f s", try_level, applied, time.time() - t0)
|
|
# an empty value clears the key: the query-string form carries it
|
|
st, body = (fc.post("/settings", params={"lid_lamp_idle": ""}) if not lamp_was
|
|
else fc.post("/settings", data={"lid_lamp_idle": lamp_was}))
|
|
ctx.check(st == 200, "restoring lid_lamp_idle=%r -> %s", lamp_was, st)
|
|
t0 = time.time()
|
|
back = None
|
|
while time.time() - t0 < 5:
|
|
back = ctx.sysfs("pic/lid_led")
|
|
if back == want:
|
|
break
|
|
ctx.sleep(0.2)
|
|
ev["lid_lamp"].update({"applied": applied, "restored": back})
|
|
ctx.check(applied == try_level, "lamp did not follow lid_lamp_idle=%s (reads %s)", try_level, applied)
|
|
ctx.check(back == want, "lamp did not return to %s after the restore (reads %s)", want, back)
|
|
ctx.log("lid lamp follows the setting live and returns to %s", want)
|
|
|
|
|
|
@test("forgectrl.panel-serves", title="Control panel and status endpoints", subsystem="forgectrl",
|
|
kind="auto", est_min=1,
|
|
covers=[("forgectrl", "src/ui.*"), ("forgectrl", "src/ui/**"), ("forgectrl", "src/status.*"),
|
|
("forgectrl", "src/cam.c"), ("forgectrl", "src/main.c")],
|
|
description="The panel page is served, /status carries the machine telemetry the panel and "
|
|
"the acceptance tool read, and /cam/status answers.")
|
|
def panel_serves(ctx):
|
|
fc = ctx.forgectrl
|
|
ev = ctx.evidence
|
|
st, body = fc.get("/", raw=True)
|
|
ev["panel_status"] = st
|
|
ctx.log("GET / -> %s (%d bytes)", st, len(body) if body else 0)
|
|
ctx.check(st == 200, "GET / -> %s", st)
|
|
text = body.decode("utf-8", "replace")
|
|
ctx.check("<html" in text.lower() and "ForgeFIRM" in text, "the panel does not look like the panel")
|
|
ctx.check(fc.token and fc.token in text, "the panel does not embed the bearer token")
|
|
ctx.check("<link " not in text and "<script src=" not in text,
|
|
"the panel references an external asset (the build did not bundle src/ui/)")
|
|
|
|
s = fc.status()
|
|
for key in ("state", "switches", "coolant", "fans"):
|
|
ctx.check(key in s, "/status lacks %r", key)
|
|
ev["state"] = s.get("state")
|
|
ev["switches"] = s.get("switches")
|
|
ctx.log("/status state=%s switches=%s", s.get("state"), s.get("switches"))
|
|
for key in ("lid", "button", "interlock_ok", "head", "hv_enable"):
|
|
ctx.check(key in (s.get("switches") or {}), "/status switches lacks %r", key)
|
|
|
|
st, cam = fc.get("/cam/status")
|
|
ev["cam_status"] = st
|
|
ctx.log("GET /cam/status -> %s %s", st, cam)
|
|
ctx.check(st == 200 and isinstance(cam, dict) and "running" in cam, "GET /cam/status -> %s", st)
|