mirror of
https://github.com/openglow-org/forgefirm.git
synced 2026-10-15 17:19:55 -07:00
Catalog: exthost.panel-install leaves
exthost.panel-install leaves for the host proofs that carry it; the catalog goes from 88 to 87. - The upload sink is forgectrl's tests/host_ext.sh (005b159): 403 without the token, 409 and not idle while cnc/state reads running with the host never run, the host's refusal at 400 in its words with the staging file gone, an archive the host takes staged byte for byte as tier unverified with consent button, the discard, an install with nothing staged at 409, and a key without the button at 409 with no key written. The consent each tier takes, the phrase, the grants' form, the button the install and a key are passed, and the key's name and the host's refusal of it are forgectrl's extpkg_test; the tiers themselves are forgeext's install_test.py. - Its section header, "installing through the panel", held only it, and EXT_STAGE and the contextlib import, which only it read, go with it. Its requires line goes with the function. Three stay. exthost.package-routes: extpkg_test has the status document, the holds, disable and enable, remove, a bad action and a bad id at 400 with nothing run, and the host's refusal at 409, but not the 403 without the login on GET /ext/status and POST /ext/package, which is main.c's guard and no case of extpkg_test's. exthost.wizard: a package's check takes no lease (wizdark.c), and the lease is held in memory, so there is no lease file to assert absent at the prompt. exthost.catalog: install_test.py and extpkg_test assert every listed claim, and what is left is the image's own key. The coverage lint names no uncovered path, so no HOST_TESTED row is added: forgectrl's src/auth.* stays covered by other catalog tests, and src/extpkg.* by the three that stay. test_setup_suite.py (31) and test_manifest.py (34) pass. The coverage lint on the tree manifest finds no uncovered path and no host-tested row problem across the 87 tests; its one finding is setup.check-lens's empty src/lensfind.* entry. pyflakes reports nothing on exthost.py. The full forgetest suite was not run. Fingerprints: exthost.py's shared text moves, so 8 tests move: the 6 left in exthost.py, and exthost.catalog and exthost.wizard in the modules that read it. The pending pin bumps from forgectrl c274f19 and forgeext 2d0391e already owe 7 of them; exthost.platform is the one this adds. releases/ and TEST-AUDIT.md are unchanged.
This commit is contained in:
ScottW514
@@ -4,7 +4,6 @@
|
||||
# SPDX-License-Identifier: MIT
|
||||
|
||||
"""exthost.* - what holds an extension package: the image's sandbox platform."""
|
||||
import contextlib
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
@@ -1706,187 +1705,3 @@ def package_routes(ctx):
|
||||
ctx.check(left == found_tree, "the extension root is not as found: %s", sorted(set(left) ^ set(found_tree)))
|
||||
ctx.check(not os.listdir(REQUIRED_HOLDS), "a required hold is still named at the end: %s", os.listdir(REQUIRED_HOLDS))
|
||||
|
||||
|
||||
# ------------------------------------------- installing through the panel
|
||||
|
||||
EXT_STAGE = "/data/forgefirm/tmp/ext-upload.ffx"
|
||||
|
||||
|
||||
@test("exthost.panel-install", title="A package is installed through the panel with the consent its tier takes",
|
||||
subsystem="exthost", kind="operator", mode="grbl", est_min=4,
|
||||
covers=[("forgectrl", "src/extpkg.*"), ("forgectrl", "src/main.c"), ("forgectrl", "src/auth.*"),
|
||||
("forgeext", "src/main.c"), ("forgeext", "src/install.*"), ("forgeext", "src/pkg.*")],
|
||||
requires=["exthost.package-routes"], actions=["button"],
|
||||
steps=["The machine idle in GRBL mode; nothing moves and nothing fires. The test asks for the button to be "
|
||||
"HELD for a few seconds while it installs an unsigned package: hold it when the notice says so."],
|
||||
description="Extensions stay as found; the packages are installed and never run. POST /ext/upload without "
|
||||
"the login is 403; an upload the host will not take (bytes that are no archive) is 400 in the "
|
||||
"host's words and leaves no staged file. The reference package signed with a key nobody "
|
||||
"trusts uploads as tier unverified with consent button: POST /ext/install without the button "
|
||||
"held is 409 and the staged file stays, the typed phrase is no substitute, and with the "
|
||||
"button held it installs with the hold the request granted and is listed unverified. With "
|
||||
"the same key added as the owner's, the upload reads community with consent typed: the "
|
||||
"install without the phrase is 400, with the phrase and without the grant 409 in the "
|
||||
"host's words, with a grant that has not the form of a capability 400, and with the phrase "
|
||||
"and the grant it installs. The key itself goes in through the panel: without the button "
|
||||
"held it is 409 and no file lands, a name with a space or a path is 400, a key that is no "
|
||||
"key is 409 from the host, and with the button held it is added and listed with its id; "
|
||||
"removed again, the same archive reads unverified. An install with nothing staged is 409, and a discarded upload "
|
||||
"is gone. Both packages are removed through the route; the key, the work directory, and "
|
||||
"the staged file are removed and the extension root is as found.")
|
||||
def panel_install(ctx):
|
||||
import shutil
|
||||
import tempfile
|
||||
fc = ctx.forgectrl
|
||||
ev = ctx.evidence
|
||||
ctx.check(len(_host_pids()) == 1, "the extension host is not one running process: %s", _host_pids())
|
||||
ctx.check(fc.wait_idle(timeout=30, abort=ctx.aborted), "machine not idle: an upload is refused")
|
||||
ctx.check(REF_ID not in [x.get("id") for x in _forgeext("list").get("packages", [])], "%s is already installed", REF_ID)
|
||||
found_tree = _tree(EXT_ROOT)
|
||||
work = tempfile.mkdtemp(prefix="forgetest-ffx.")
|
||||
owner_key = os.path.join(EXT_ROOT, "keys", REF_KEY + ".pub")
|
||||
|
||||
def upload(path=None, raw=None, login=True):
|
||||
mark = "forgetestExtBoundary7d1"
|
||||
with open(path, "rb") if path else contextlib.nullcontext() as f:
|
||||
data = f.read() if path else raw
|
||||
body = ('--%s\r\nContent-Disposition: form-data; name="file"; filename="package.ffx"\r\n'
|
||||
'Content-Type: application/octet-stream\r\n\r\n' % mark).encode() + data + ("\r\n--%s--\r\n" % mark).encode()
|
||||
hdrs = {"Content-Type": "multipart/form-data; boundary=%s" % mark}
|
||||
if login:
|
||||
return fc.post("/ext/upload", data=body, headers=hdrs)
|
||||
st, reply, _ = request(fc.base, "POST", "/ext/upload", data=body, headers=dict(hdrs, Host=fc.host_header()))
|
||||
return st, reply
|
||||
|
||||
def install(**form):
|
||||
return fc.post("/ext/install", data=form)
|
||||
|
||||
def installed():
|
||||
return next((x for x in _forgeext("list").get("packages", []) if x.get("id") == REF_ID), None)
|
||||
|
||||
try:
|
||||
archive, pub = _pack_reference(work, lan_ip(), more_caps=["hold"])
|
||||
|
||||
st, reply = upload(archive, login=False)
|
||||
ctx.check(st == 403 and not os.path.exists(EXT_STAGE), "an upload without the login -> %s, staged %s", st, os.path.exists(EXT_STAGE))
|
||||
st, reply = upload(raw=b"these bytes are no archive\n" * 40)
|
||||
ev["not_an_archive"] = [st, reply if isinstance(reply, str) else ""]
|
||||
ctx.log("an upload that is no archive -> %s %s", st, reply if isinstance(reply, str) else "")
|
||||
ctx.check(st == 400 and not os.path.exists(EXT_STAGE), "an upload the host will not take -> %s, staged file %s", st,
|
||||
"kept" if os.path.exists(EXT_STAGE) else "gone")
|
||||
st, reply = install(grants="hold", phrase=SAFETY_PHRASE)
|
||||
ctx.check(st == 409 and isinstance(reply, str) and "staged" in reply, "an install with nothing staged -> %s %r", st, reply)
|
||||
|
||||
# nobody the machine trusts signed it: the button, held
|
||||
st, doc = upload(archive)
|
||||
ev["unverified_upload"] = {k: (doc or {}).get(k) for k in ("tier", "consent", "needs_grant")} if isinstance(doc, dict) else doc
|
||||
ctx.log("the upload, its key unknown to the machine -> %s %s", st, ev["unverified_upload"])
|
||||
ctx.check(st == 200 and isinstance(doc, dict) and doc.get("tier") == "unverified" and doc.get("consent") == "button"
|
||||
and doc.get("needs_grant") == ["hold"] and (doc.get("package") or {}).get("id") == REF_ID,
|
||||
"the unverified upload -> %s %s", st, doc)
|
||||
st, reply = install(grants="hold", phrase=SAFETY_PHRASE)
|
||||
ev["unverified_without_the_button"] = [st, reply if isinstance(reply, str) else ""]
|
||||
ctx.check(st == 409 and isinstance(reply, str) and "button" in reply and os.path.exists(EXT_STAGE) and not installed(),
|
||||
"an unverified install without the button held -> %s %r", st, reply)
|
||||
done = {}
|
||||
|
||||
def held_install():
|
||||
st_, reply_ = install(grants="hold")
|
||||
done["last"] = [st_, reply_ if isinstance(reply_, str) else "ok"]
|
||||
return st_ == 200
|
||||
|
||||
# A person holds the button; the bench actuator's press is a half-second pulse, so it is pressed again
|
||||
# until a request has landed inside one.
|
||||
for _ in range(10):
|
||||
if ctx.act("button", "press", text="HOLD the button now, for a few seconds: an unsigned package is being installed.",
|
||||
until=held_install, timeout=4, fail=False, ms=500) is not None:
|
||||
break
|
||||
pkg = installed()
|
||||
ev["unverified_installed"] = {k: (pkg or {}).get(k) for k in ("tier", "grants", "hold", "enabled")}
|
||||
ctx.log("with the button held: %s, listed %s", done.get("last"), ev["unverified_installed"])
|
||||
ctx.check(pkg and pkg.get("tier") == "unverified" and pkg.get("grants") == ["hold"] and not os.path.exists(EXT_STAGE),
|
||||
"with the button held the package is not installed as unverified with its grant: %s (%s)", pkg, done.get("last"))
|
||||
st, reply = fc.post("/ext/package", data={"id": REF_ID, "action": "remove"})
|
||||
ctx.check(st == 200 and not installed(), "remove through the route -> %s", st)
|
||||
|
||||
# the same key, now the owner's, added through the panel with the machine's button held
|
||||
with open(pub) as f:
|
||||
keytext = f.read().strip()
|
||||
st, reply = fc.post("/ext/key", data={"name": REF_KEY, "key": keytext})
|
||||
ev["key_without_the_button"] = [st, reply if isinstance(reply, str) else ""]
|
||||
ctx.check(st == 409 and isinstance(reply, str) and "button" in reply and not os.path.exists(owner_key),
|
||||
"a key added without the button held -> %s %r", st, reply)
|
||||
for name, form, want in (("a name with a space", {"name": "a maker", "key": keytext}, 400),
|
||||
("a name that is a path", {"name": "../../etc/passwd", "key": keytext}, 400),
|
||||
("no key at all", {"name": REF_KEY}, 400),
|
||||
("a key that is no key", {"name": REF_KEY, "key": "this is no key"}, 409)):
|
||||
st, reply = fc.post("/ext/key", data=form)
|
||||
ev[name] = st
|
||||
ctx.check(st == want and not os.path.exists(owner_key), "%s -> %s, expected %s", name, st, want)
|
||||
keyed = {}
|
||||
|
||||
def held_key():
|
||||
st_, reply_ = fc.post("/ext/key", data={"name": REF_KEY, "key": keytext})
|
||||
keyed["last"] = [st_, reply_ if isinstance(reply_, str) else "ok"]
|
||||
return st_ == 200
|
||||
|
||||
for _ in range(10):
|
||||
if ctx.act("button", "press", text="HOLD the button now, for a few seconds: a key is being added.",
|
||||
until=held_key, timeout=4, fail=False, ms=500) is not None:
|
||||
break
|
||||
st, doc = fc.get("/ext/status")
|
||||
keys = {k.get("name"): k.get("key") for k in ((doc or {}).get("keys") or [])}
|
||||
ev["keys"] = keys
|
||||
ctx.log("with the button held: %s, the keys now %s", keyed.get("last"), sorted(keys))
|
||||
ctx.check(REF_KEY in keys and os.path.exists(owner_key), "the key was not added with the button held: %s (%s)",
|
||||
sorted(keys), keyed.get("last"))
|
||||
ctx.check(len(keys.get(REF_KEY) or "") == 64, "the key is listed without its id: %s", keys)
|
||||
|
||||
st, doc = upload(archive)
|
||||
ev["community_upload"] = {k: (doc or {}).get(k) for k in ("tier", "consent", "needs_grant")} if isinstance(doc, dict) else doc
|
||||
ctx.check(st == 200 and isinstance(doc, dict) and doc.get("tier") == "community" and doc.get("consent") == "typed",
|
||||
"the community upload -> %s %s", st, doc)
|
||||
for name, form, want, words in (
|
||||
("no phrase", {"grants": "hold"}, 400, "type I UNDERSTAND"),
|
||||
("the phrase in another case", {"grants": "hold", "phrase": SAFETY_PHRASE.lower()}, 400, "type I UNDERSTAND"),
|
||||
("the phrase and no grant", {"phrase": SAFETY_PHRASE}, 409, "hold"),
|
||||
("a grant with a shell's words", {"phrase": SAFETY_PHRASE, "grants": "hold;reboot"}, 400, "capability names"),
|
||||
("a grant that is an option", {"phrase": SAFETY_PHRASE, "grants": "--consent-unverified"}, 400, "capability names")):
|
||||
st, reply = install(**form)
|
||||
ev[name] = st
|
||||
ctx.log("POST /ext/install, %s -> %s %s", name, st, reply if isinstance(reply, str) else "")
|
||||
ctx.check(st == want and isinstance(reply, str) and words in reply and not installed() and os.path.exists(EXT_STAGE),
|
||||
"%s -> %s %r, expected %s with the package not installed and the staged file kept", name, st, reply, want)
|
||||
st, reply = install(grants="hold", phrase=SAFETY_PHRASE)
|
||||
pkg = installed()
|
||||
ctx.check(st == 200 and pkg and pkg.get("tier") == "community" and pkg.get("grants") == ["hold"]
|
||||
and not os.path.exists(EXT_STAGE), "the community install -> %s, listed %s", st, pkg)
|
||||
st, reply = fc.post("/ext/package", data={"id": REF_ID, "action": "remove"})
|
||||
ctx.check(st == 200 and not installed(), "remove through the route -> %s", st)
|
||||
|
||||
# the key removed: the same archive reads unverified again
|
||||
st, reply = fc.post("/ext/key/remove", data={"name": REF_KEY})
|
||||
ctx.check(st == 200 and not os.path.exists(owner_key), "the key was not removed -> %s %r", st, reply)
|
||||
st, reply = fc.post("/ext/key/remove", data={"name": REF_KEY})
|
||||
ctx.check(st == 409, "removing a key that is not there -> %s", st)
|
||||
|
||||
st, doc = upload(archive)
|
||||
ev["after_the_key_went"] = (doc or {}).get("tier") if isinstance(doc, dict) else doc
|
||||
ctx.check(st == 200 and isinstance(doc, dict) and doc.get("tier") == "unverified" and os.path.exists(EXT_STAGE),
|
||||
"without the key the same archive reads %s", ev["after_the_key_went"])
|
||||
st, reply = fc.post("/ext/upload/discard")
|
||||
ctx.check(st == 200 and not os.path.exists(EXT_STAGE), "discard -> %s, staged file %s", st,
|
||||
"kept" if os.path.exists(EXT_STAGE) else "gone")
|
||||
st, reply = install(grants="hold", phrase=SAFETY_PHRASE)
|
||||
ctx.check(st == 409 and not installed(), "an install after the discard -> %s", st)
|
||||
finally:
|
||||
ctx.clear_notice()
|
||||
if installed():
|
||||
_forgeext("remove", REF_ID)
|
||||
for path in (owner_key, EXT_STAGE):
|
||||
if os.path.exists(path):
|
||||
os.remove(path)
|
||||
shutil.rmtree(work, ignore_errors=True)
|
||||
left = _tree(EXT_ROOT)
|
||||
ctx.check(left == found_tree, "the extension root is not as found: %s", sorted(set(left) ^ set(found_tree)))
|
||||
|
||||
|
||||
Reference in New Issue
Block a user