diff options
Diffstat (limited to 'packages/meshbay-cli/tests/fake_node.py')
| -rw-r--r-- | packages/meshbay-cli/tests/fake_node.py | 86 |
1 files changed, 85 insertions, 1 deletions
diff --git a/packages/meshbay-cli/tests/fake_node.py b/packages/meshbay-cli/tests/fake_node.py index e733c56..f53eb01 100644 --- a/packages/meshbay-cli/tests/fake_node.py +++ b/packages/meshbay-cli/tests/fake_node.py @@ -9,16 +9,30 @@ the node's code: a node the CLI must work against. """ import asyncio +import base64 import hashlib import math import secrets +from cryptography.hazmat.primitives.asymmetric.ed25519 import ( + Ed25519PrivateKey, + Ed25519PublicKey, +) +from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey +from fake_hub import USER_ID +from meshbay_cli.client.keys import Identity from meshbay_cli.client.session import GroupSession, NodeRefusal +from meshbay_common.device import ( + device_add_transcript, + device_code_hash, + device_revoke_transcript, +) from meshbay_common.groupbox import PURPOSE_INDEX, PURPOSE_UPLOAD, seal, unseal from meshbay_common.protocol import UPLOAD_PROBE_INDEX, file_upload_ack_wire from meshbay_common.webcrypto import chunk_key_aes, encrypt_chunk_aes GROUP_ID = "g1" +NODE_PK = base64.b64encode(b"\x05" * 32).decode() CHUNK = 1024 * 1024 @@ -35,6 +49,14 @@ class FakeNode: self.refuse_upload_code = "" self.chunk_requests = 0 self.chunk_delay = 0.0 + # This machine's key here, the account's other devices, and the requests + # waiting for an approval — each `{pk_ed25519, pk_x25519, code_hash}`. + self.identity = Identity(Ed25519PrivateKey.generate(), X25519PrivateKey.generate()) + self.nonce = secrets.token_bytes(32) + self.devices = [{"pk_ed25519": self.identity.pk_ed, "pk_x25519": self.identity.pk_x, + "label": "", "pinned_at": "2026-10-01T10:00:00+00:00", + "pinned_via": "code"}] + self.requests: list[dict] = [] def add(self, path: str, name: str, data: bytes) -> dict: fid = hashlib.blake2b(data + path.encode() + name.encode(), digest_size=32).hexdigest() @@ -47,7 +69,17 @@ class FakeNode: return entry def session(self) -> GroupSession: - return GroupSession(None, FakeChannel(self), GROUP_ID, "node-1", "pk", self.gek, {}) + return GroupSession(None, FakeChannel(self), GROUP_ID, "node-1", NODE_PK, self.gek, + {}, self.identity, self.nonce) + + def request_device(self, code: str) -> Identity: + """Another machine files a request with `code`, as it would before its proof.""" + other = Identity(Ed25519PrivateKey.generate(), X25519PrivateKey.generate()) + self.requests.append({"pk_ed25519": other.pk_ed, "pk_x25519": other.pk_x, + "created_at": "2026-10-14T10:00:00+00:00", + "code_hash": device_code_hash(code.replace("-", ""), + other.pk_ed, other.pk_x)}) + return other def stored(self, path: str) -> bytes | None: parent, _, name = path.rpartition("/") @@ -162,5 +194,57 @@ class FakeChannel: dir=folder) + # ── Devices ────────────────────────────────────────────────────────────── + + def _device_list(self, msg): + n = self.node + return {"type": "device_list_result", "pending": len(n.requests), + "devices": [{**d, "is_this_one": d["pk_ed25519"] == n.identity.pk_ed} + for d in n.devices]} + + def _device_lookup(self, msg): + return {"type": "device_lookup_result", "requests": list(self.node.requests)} + + def _device_add(self, msg): + n = self.node + transcript = device_add_transcript( + node_pk_b64=NODE_PK, user_id=USER_ID, pk_ed25519_b64=msg["pk_ed25519"], + pk_x25519_b64=msg["pk_x25519"], nonce_node=n.nonce, ts=msg["ts"]) + if not _signed_by_a_device(n, transcript, msg["sig"]): + return {"type": "error", "detail": "Not signed by a device already paired here"} + req = next((r for r in n.requests if r["code_hash"] == msg["code_hash"] + and r["pk_ed25519"] == msg["pk_ed25519"]), None) + if req is None: + return {"type": "error", "detail": "No pending request"} + n.requests.remove(req) + n.devices.append({"pk_ed25519": req["pk_ed25519"], "pk_x25519": req["pk_x25519"], + "label": "", "pinned_at": "2026-10-14T10:01:00+00:00", + "pinned_via": "device"}) + return {"type": "device_add_ack", "pk_ed25519": req["pk_ed25519"]} + + def _device_revoke(self, msg): + n = self.node + transcript = device_revoke_transcript( + node_pk_b64=NODE_PK, user_id=USER_ID, pk_ed25519_b64=msg["pk_ed25519"], + nonce_node=n.nonce, ts=msg["ts"]) + if not _signed_by_a_device(n, transcript, msg["sig"]): + return {"type": "error", "detail": "Not signed by a device already paired here"} + if len(n.devices) <= 1: + return {"type": "error", "detail": "This is your only device here"} + n.devices = [d for d in n.devices if d["pk_ed25519"] != msg["pk_ed25519"]] + return {"type": "device_add_ack", "revoked": msg["pk_ed25519"]} + + +def _signed_by_a_device(node: FakeNode, transcript: bytes, sig: str) -> bool: + for d in node.devices: + try: + Ed25519PublicKey.from_public_bytes(base64.b64decode(d["pk_ed25519"])).verify( + base64.b64decode(sig), transcript) + return True + except Exception: + continue + return False + + def chunks_of(size: int, chunk: int) -> int: return max(1, math.ceil(size / chunk)) |