operator
41499e069d
Fingerprint map + pure resolver (account umbrella / key-level scope rule), live Warp provider on warp-* structures, broker lifecycle (up/down/cycle/exec/routes/status/bind), wireguard+socks boilerplate stubs, agent-manager bind integration. CLI carries emails and fingerprints only; key bytes never appear. 41 committed tests.
476 lines
18 KiB
Python
476 lines
18 KiB
Python
#!/usr/bin/env python3
|
|
"""test_identity_plane.py — Unit tests for the identity plane.
|
|
|
|
Slice 1 covers the pure resolver (map -> scope); provider command
|
|
shapes and broker state transitions use stubbed runners only.
|
|
No test touches the network, netns, or /etc/netvm.
|
|
"""
|
|
|
|
import importlib.util
|
|
import json
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
|
|
|
|
def load_bin(name, modname):
|
|
path = REPO_ROOT / "bin" / name
|
|
spec = importlib.util.spec_from_file_location(modname, path)
|
|
mod = importlib.util.module_from_spec(spec)
|
|
sys.modules[modname] = mod
|
|
spec.loader.exec_module(mod)
|
|
return mod
|
|
|
|
|
|
resolve = load_bin("identity-resolve.py", "identity_resolve")
|
|
provider = load_bin("identity-provider.py", "identity_provider")
|
|
broker = load_bin("identity-broker.py", "identity_broker")
|
|
|
|
|
|
class FakeProvider:
|
|
name = "fake"
|
|
ready = True
|
|
|
|
def __init__(self):
|
|
self.calls = []
|
|
|
|
def provision(self, label, run=None):
|
|
self.calls.append(("provision", label))
|
|
return {"ok": "true", "label": label,
|
|
"netns": "warp-" + label}
|
|
|
|
def teardown(self, label, run=None):
|
|
self.calls.append(("teardown", label))
|
|
return {"ok": "true", "label": label}
|
|
|
|
def cycle(self, label, run=None):
|
|
self.calls.append(("cycle", label))
|
|
return {"ok": "true", "label": label}
|
|
|
|
def exec(self, label, cmd, run=None):
|
|
self.calls.append(("exec", label, list(cmd)))
|
|
return 0, "fake-out"
|
|
|
|
def routes(self, label, run=None):
|
|
self.calls.append(("routes", label))
|
|
return {"label": label, "routes": "r", "wireguard": "w"}
|
|
|
|
def status(self, label, run=None):
|
|
self.calls.append(("status", label))
|
|
return {"label": label, "conf": "yes", "netns": "up",
|
|
"egress": "9.9.9.9"}
|
|
|
|
|
|
class BrokerCase(unittest.TestCase):
|
|
def setUp(self):
|
|
import tempfile
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.map_path = str(Path(self.tmp.name) / "map.json")
|
|
self.state_path = str(Path(self.tmp.name) / "state.json")
|
|
with open(self.map_path, "w") as f:
|
|
json.dump(MAP, f)
|
|
self.fake = FakeProvider()
|
|
self.orig = dict(broker.PROVIDERS)
|
|
broker.PROVIDERS["warp"] = self.fake
|
|
self.addCleanup(lambda: broker.PROVIDERS.update(self.orig))
|
|
|
|
def kw(self):
|
|
run = StubRun([])
|
|
return {"run": run, "map_path": self.map_path,
|
|
"state_path": self.state_path}
|
|
|
|
|
|
class TestBrokerOps(BrokerCase):
|
|
def test_up_unknown_fp_fails_closed(self):
|
|
with self.assertRaises(broker.BrokerError):
|
|
broker.op_up("sha256:nope", **self.kw())
|
|
self.assertEqual(self.fake.calls, [])
|
|
|
|
def test_up_provisions_and_records(self):
|
|
res = broker.op_up("sha256:aaa", **self.kw())
|
|
self.assertEqual(res["ok"], "true")
|
|
self.assertEqual(res["email"], "uma@example.com")
|
|
self.assertEqual(self.fake.calls,
|
|
[("provision", res["label"])])
|
|
state = broker.load_state(self.state_path)
|
|
self.assertIn("uma@example.com", state["scopes"])
|
|
|
|
def test_up_is_idempotent(self):
|
|
broker.op_up("sha256:aaa", **self.kw())
|
|
res = broker.op_up("sha256:aaa", **self.kw())
|
|
self.assertEqual(res["ok"], "exists")
|
|
self.assertEqual(len(self.fake.calls), 1)
|
|
|
|
def test_down_clears_scope(self):
|
|
up = broker.op_up("sha256:aaa", **self.kw())
|
|
res = broker.op_down("sha256:aaa", **self.kw())
|
|
self.assertEqual(res["unit"], "uma@example.com")
|
|
self.assertIn(("teardown", up["label"]), self.fake.calls)
|
|
state = broker.load_state(self.state_path)
|
|
self.assertEqual(state["scopes"], {})
|
|
|
|
def test_down_unknown_fails_closed(self):
|
|
with self.assertRaises(broker.BrokerError):
|
|
broker.op_down("sha256:nope", **self.kw())
|
|
self.assertEqual(self.fake.calls, [])
|
|
|
|
def test_cycle_bumps_count(self):
|
|
broker.op_up("sha256:aaa", **self.kw())
|
|
res = broker.op_cycle("sha256:aaa", **self.kw())
|
|
self.assertEqual(res["cycles"], 1)
|
|
state = broker.load_state(self.state_path)
|
|
self.assertEqual(state["scopes"]["uma@example.com"]["cycles"], 1)
|
|
|
|
def test_cycle_requires_up(self):
|
|
with self.assertRaises(broker.BrokerError):
|
|
broker.op_cycle("sha256:aaa", **self.kw())
|
|
|
|
def test_exec_passthrough(self):
|
|
broker.op_up("sha256:aaa", **self.kw())
|
|
rc, out = broker.op_exec("sha256:aaa", ["curl", "x"],
|
|
**self.kw())
|
|
self.assertEqual((rc, out), (0, "fake-out"))
|
|
|
|
def test_exec_requires_up(self):
|
|
with self.assertRaises(broker.BrokerError):
|
|
broker.op_exec("sha256:aaa", ["id"], **self.kw())
|
|
|
|
def test_status_emails_only(self):
|
|
broker.op_up("sha256:aaa", **self.kw())
|
|
st = broker.op_status(run=StubRun([]),
|
|
state_path=self.state_path)
|
|
blob = json.dumps(st).lower()
|
|
self.assertIn("uma@example.com", blob)
|
|
for banned in ("password", "secret", "bearer", "api_key",
|
|
"private"):
|
|
self.assertNotIn(banned, blob)
|
|
|
|
|
|
class TestBrokerBind(BrokerCase):
|
|
SCAN = {"runs": [
|
|
{"device": "bl", "type": "muse", "session": "muse,other",
|
|
"pane": "%1", "pid": 111},
|
|
{"device": "bl", "type": "muse", "session": "muse",
|
|
"pane": "%3", "pid": 333},
|
|
{"device": "tp", "type": "agy", "session": "x",
|
|
"pane": "%0", "pid": 999},
|
|
]}
|
|
|
|
def scan_path(self):
|
|
p = str(Path(self.tmp.name) / "runs.json")
|
|
with open(p, "w") as f:
|
|
json.dump(self.SCAN, f)
|
|
return p
|
|
|
|
def test_bind_attributes_group_members(self):
|
|
res = broker.op_bind(self.scan_path(), "muse", "sha256:aaa",
|
|
map_path=self.map_path,
|
|
state_path=self.state_path)
|
|
self.assertEqual(res["bound"], 2)
|
|
self.assertEqual(res["unit"], "uma@example.com")
|
|
state = broker.load_state(self.state_path)
|
|
self.assertEqual(len(state["bindings"]), 2)
|
|
pids = sorted(b["pid"] for b in state["bindings"])
|
|
self.assertEqual(pids, [111, 333])
|
|
|
|
def test_bind_rejects_unseen_session(self):
|
|
with self.assertRaises(broker.BrokerError) as ctx:
|
|
broker.op_bind(self.scan_path(), "ghost", "sha256:aaa",
|
|
map_path=self.map_path,
|
|
state_path=self.state_path)
|
|
self.assertIn("refusing to bind unseen runs",
|
|
str(ctx.exception))
|
|
|
|
def test_bind_rejects_unknown_fp(self):
|
|
with self.assertRaises(broker.BrokerError):
|
|
broker.op_bind(self.scan_path(), "muse", "sha256:nope",
|
|
map_path=self.map_path,
|
|
state_path=self.state_path)
|
|
|
|
def test_rebind_refreshes_without_dupes(self):
|
|
kw = {"map_path": self.map_path,
|
|
"state_path": self.state_path}
|
|
broker.op_bind(self.scan_path(), "muse", "sha256:aaa", **kw)
|
|
broker.op_bind(self.scan_path(), "muse", "sha256:aaa", **kw)
|
|
state = broker.load_state(self.state_path)
|
|
self.assertEqual(len(state["bindings"]), 2)
|
|
|
|
|
|
class TestBrokerCLI(BrokerCase):
|
|
def test_status_main_uses_state_flag(self):
|
|
import io
|
|
from contextlib import redirect_stdout
|
|
broker.op_up("sha256:aaa", **self.kw())
|
|
buf = io.StringIO()
|
|
with redirect_stdout(buf):
|
|
rc = broker.main(["--map", self.map_path,
|
|
"--state", self.state_path, "status"])
|
|
self.assertEqual(rc, 0)
|
|
out = json.loads(buf.getvalue())
|
|
self.assertEqual(len(out["scopes"]), 1)
|
|
self.assertEqual(out["scopes"][0]["email"], "uma@example.com")
|
|
|
|
def test_lookup_main_uses_map_flag(self):
|
|
import io
|
|
from contextlib import redirect_stdout
|
|
buf = io.StringIO()
|
|
with redirect_stdout(buf):
|
|
rc = resolve.main(["--map", self.map_path, "lookup",
|
|
"sha256:bbb"])
|
|
self.assertEqual(rc, 0)
|
|
out = json.loads(buf.getvalue())
|
|
self.assertEqual(out["scope"], "key")
|
|
|
|
def test_exec_main_dispatches_passthrough(self):
|
|
import io
|
|
from contextlib import redirect_stdout
|
|
broker.op_up("sha256:aaa", **self.kw())
|
|
buf = io.StringIO()
|
|
with redirect_stdout(buf):
|
|
rc = broker.main(["--map", self.map_path,
|
|
"--state", self.state_path,
|
|
"exec", "sha256:aaa", "--",
|
|
"curl", "x"])
|
|
self.assertEqual(rc, 0)
|
|
self.assertEqual(buf.getvalue(), "fake-out\n")
|
|
self.assertIn(("exec", broker.load_state(self.state_path)
|
|
["scopes"]["uma@example.com"]["label"],
|
|
["curl", "x"]),
|
|
[(c[0], c[1], c[2]) for c in self.fake.calls
|
|
if c[0] == "exec"])
|
|
|
|
|
|
class StubRun:
|
|
"""Scripted run function: match argv[0]/fragments -> (rc, out)."""
|
|
|
|
def __init__(self, script):
|
|
self.script = list(script)
|
|
self.calls = []
|
|
|
|
def __call__(self, cmd, timeout=120):
|
|
self.calls.append(list(cmd))
|
|
for i, (frag, rc, out) in enumerate(self.script):
|
|
if frag in " ".join(cmd):
|
|
del self.script[i]
|
|
return rc, out
|
|
return 0, ""
|
|
|
|
|
|
class TestWarpProviderShapes(unittest.TestCase):
|
|
def test_provision_new_identity(self):
|
|
run = StubRun([("test -f", 1, ""),
|
|
("netvm-new-identity.sh", 0, "installed"),
|
|
("netvm-node-up.sh", 0, "up")])
|
|
w = provider.WarpProvider()
|
|
res = w.provision("id-x", run=run)
|
|
self.assertEqual(res["ok"], "true")
|
|
self.assertEqual(res["netns"], "warp-id-x")
|
|
self.assertIn("identity=new", res["steps"])
|
|
self.assertEqual(len(run.calls), 3)
|
|
|
|
def test_provision_existing_identity_skips_generation(self):
|
|
run = StubRun([("test -f", 0, ""),
|
|
("netvm-node-up.sh", 0, "up")])
|
|
w = provider.WarpProvider()
|
|
res = w.provision("id-x", run=run)
|
|
self.assertIn("identity=exists", res["steps"])
|
|
joined = " ".join(" ".join(c) for c in run.calls)
|
|
self.assertNotIn("new-identity", joined)
|
|
|
|
def test_exec_shape(self):
|
|
run = StubRun([("netvm-exec.sh", 0, "1.2.3.4")])
|
|
w = provider.WarpProvider()
|
|
rc, out = w.exec("id-x", ["curl", "https://api.ipify.org"],
|
|
run=run)
|
|
self.assertEqual((rc, out), (0, "1.2.3.4"))
|
|
argv = run.calls[0]
|
|
self.assertIn("netvm-exec.sh", argv[0])
|
|
self.assertEqual(argv[1:3], ["id-x", "--"])
|
|
|
|
def test_teardown_shape(self):
|
|
run = StubRun([("netvm-node-down.sh", 0, "down")])
|
|
w = provider.WarpProvider()
|
|
res = w.teardown("id-x", run=run)
|
|
self.assertEqual(res["ok"], "true")
|
|
|
|
def test_cycle_shape(self):
|
|
run = StubRun([("netvm-node-down.sh", 0, "down"),
|
|
("rm -f", 0, ""),
|
|
("test -f", 1, ""),
|
|
("netvm-new-identity.sh", 0, "installed"),
|
|
("netvm-node-up.sh", 0, "up")])
|
|
w = provider.WarpProvider()
|
|
res = w.cycle("id-x", run=run)
|
|
self.assertEqual(res["ok"], "true")
|
|
self.assertEqual(len(run.calls), 5)
|
|
|
|
def test_cycle_refused_rm_fails_closed(self):
|
|
run = StubRun([("netvm-node-down.sh", 0, "down"),
|
|
("rm -f", 1, "denied")])
|
|
w = provider.WarpProvider()
|
|
with self.assertRaises(provider.ProviderError) as ctx:
|
|
w.cycle("id-x", run=run)
|
|
self.assertIn("Human: sudo rm", str(ctx.exception))
|
|
|
|
def test_bad_label_runs_nothing(self):
|
|
run = StubRun([])
|
|
w = provider.WarpProvider()
|
|
with self.assertRaises(provider.ProviderError):
|
|
w.provision("BAD LABEL!", run=run)
|
|
with self.assertRaises(provider.ProviderError):
|
|
w.exec("x" * 30, ["id"], run=run)
|
|
self.assertEqual(run.calls, [])
|
|
|
|
def test_status_shape(self):
|
|
run = StubRun([("test -f", 0, ""),
|
|
("netns list", 0, "warp-id-x (id: 3)"),
|
|
("netvm-exec.sh", 0, "9.9.9.9")])
|
|
w = provider.WarpProvider()
|
|
st = w.status("id-x", run=run)
|
|
self.assertEqual(st, {"label": "id-x", "conf": "yes",
|
|
"netns": "up", "egress": "9.9.9.9"})
|
|
|
|
def test_base_is_boilerplate(self):
|
|
p = provider.Provider()
|
|
with self.assertRaises(NotImplementedError):
|
|
p.provision("id-x", run=StubRun([]))
|
|
|
|
def test_ready_flags(self):
|
|
self.assertFalse(provider.Provider.ready)
|
|
self.assertTrue(provider.WarpProvider.ready)
|
|
self.assertFalse(provider.GenericWireGuardProvider.ready)
|
|
self.assertFalse(provider.SocksProxyProvider.ready)
|
|
|
|
def test_stubs_raise_not_implemented(self):
|
|
for cls in (provider.GenericWireGuardProvider,
|
|
provider.SocksProxyProvider):
|
|
with self.assertRaises(NotImplementedError):
|
|
cls().provision("id-x", run=StubRun([]))
|
|
|
|
def test_broker_refuses_boilerplate_provider(self):
|
|
import tempfile
|
|
with tempfile.TemporaryDirectory() as td:
|
|
mp = str(Path(td) / "map.json")
|
|
sp = str(Path(td) / "state.json")
|
|
with open(mp, "w") as f:
|
|
json.dump(MAP, f)
|
|
with self.assertRaises(broker.BrokerError) as ctx:
|
|
broker.op_up("sha256:aaa", run=StubRun([]),
|
|
map_path=mp, state_path=sp,
|
|
provider_name="socks")
|
|
self.assertIn("boilerplate", str(ctx.exception))
|
|
with self.assertRaises(broker.BrokerError) as ctx2:
|
|
broker.op_up("sha256:aaa", run=StubRun([]),
|
|
map_path=mp, state_path=sp,
|
|
provider_name="nope")
|
|
self.assertIn("unknown provider", str(ctx2.exception))
|
|
|
|
MAP = {
|
|
"accounts": {
|
|
"uma@example.com": {
|
|
"keys": [
|
|
{"fp": "sha256:aaa", "origins": ["uma@example.com"],
|
|
"label": "main"},
|
|
{"fp": "sha256:bbb",
|
|
"origins": ["uma@example.com", "provider-acct"],
|
|
"label": "openrouter"},
|
|
]
|
|
},
|
|
"nadia@example.com": {
|
|
"keys": [
|
|
{"fp": "sha256:ccc", "origins": ["nadia@example.com"]},
|
|
]
|
|
},
|
|
}
|
|
}
|
|
|
|
|
|
class TestFingerprint(unittest.TestCase):
|
|
def test_known_vector(self):
|
|
self.assertEqual(
|
|
resolve.fingerprint_hex(b"abc"),
|
|
"sha256:ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad")
|
|
|
|
def test_empty(self):
|
|
self.assertTrue(resolve.fingerprint_hex(b"").startswith("sha256:"))
|
|
self.assertEqual(len(resolve.fingerprint_hex(b"")), 7 + 64)
|
|
|
|
|
|
class TestResolveScope(unittest.TestCase):
|
|
def test_single_origin_rolls_up_to_account(self):
|
|
s = resolve.resolve_scope(MAP, "sha256:aaa")
|
|
self.assertEqual(s["scope"], "account")
|
|
self.assertEqual(s["unit"], "uma@example.com")
|
|
self.assertEqual(s["email"], "uma@example.com")
|
|
self.assertEqual(s["origins"], ["uma@example.com"])
|
|
|
|
def test_dual_origin_stays_at_key(self):
|
|
s = resolve.resolve_scope(MAP, "sha256:bbb")
|
|
self.assertEqual(s["scope"], "key")
|
|
self.assertEqual(s["unit"], "sha256:bbb")
|
|
self.assertEqual(s["email"], "uma@example.com")
|
|
self.assertEqual(s["origins"],
|
|
["uma@example.com", "provider-acct"])
|
|
|
|
def test_unknown_fingerprint(self):
|
|
self.assertIsNone(resolve.resolve_scope(MAP, "sha256:nope"))
|
|
self.assertIsNone(resolve.resolve_scope({}, "sha256:aaa"))
|
|
self.assertIsNone(resolve.resolve_scope(MAP, ""))
|
|
|
|
def test_ignores_underscore_keys(self):
|
|
m = {"_comment": "doc", "_example": {}, "accounts": {}}
|
|
self.assertIsNone(resolve.resolve_scope(m, "sha256:aaa"))
|
|
|
|
def test_no_key_material_in_result(self):
|
|
s = resolve.resolve_scope(MAP, "sha256:aaa")
|
|
blob = json.dumps(s).lower()
|
|
for banned in ("password", "secret", "bearer", "api_key",
|
|
"apikey", "private"):
|
|
self.assertNotIn(banned, blob)
|
|
|
|
|
|
class TestScopeSlug(unittest.TestCase):
|
|
def test_label_shape(self):
|
|
import re
|
|
s = resolve.resolve_scope(MAP, "sha256:aaa")
|
|
slug = resolve.scope_slug(s)
|
|
self.assertRegex(slug, r"^[a-z0-9][a-z0-9-]{0,22}$")
|
|
self.assertTrue(slug.startswith("id-"))
|
|
|
|
def test_deterministic_and_distinct(self):
|
|
a = resolve.resolve_scope(MAP, "sha256:aaa")
|
|
b = resolve.resolve_scope(MAP, "sha256:bbb")
|
|
c = resolve.resolve_scope(MAP, "sha256:ccc")
|
|
self.assertEqual(resolve.scope_slug(a), resolve.scope_slug(a))
|
|
self.assertEqual(len({resolve.scope_slug(a),
|
|
resolve.scope_slug(b),
|
|
resolve.scope_slug(c)}), 3)
|
|
|
|
def test_key_scope_uses_fp_prefix(self):
|
|
b = resolve.resolve_scope(MAP, "sha256:bbb")
|
|
self.assertIn("bbb", resolve.scope_slug(b))
|
|
|
|
|
|
class TestLoadMap(unittest.TestCase):
|
|
def test_missing_file_is_empty(self):
|
|
self.assertEqual(resolve.load_map("/nonexistent/x.json"),
|
|
{"accounts": {}})
|
|
|
|
def test_corrupt_file_is_empty(self):
|
|
import tempfile
|
|
with tempfile.NamedTemporaryFile("w", suffix=".json",
|
|
delete=False) as f:
|
|
f.write("{not json")
|
|
path = f.name
|
|
try:
|
|
self.assertEqual(resolve.load_map(path), {"accounts": {}})
|
|
finally:
|
|
Path(path).unlink()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|