diff options
Diffstat (limited to 'packages/meshbay-cli/tests/fake_node.py')
| -rw-r--r-- | packages/meshbay-cli/tests/fake_node.py | 51 |
1 files changed, 47 insertions, 4 deletions
diff --git a/packages/meshbay-cli/tests/fake_node.py b/packages/meshbay-cli/tests/fake_node.py index f5fe631..8f4356b 100644 --- a/packages/meshbay-cli/tests/fake_node.py +++ b/packages/meshbay-cli/tests/fake_node.py @@ -32,6 +32,7 @@ from meshbay_common.device import ( from meshbay_common.groupbox import ( PURPOSE_CHAT_KEYS, PURPOSE_INDEX, + PURPOSE_ROSTER, PURPOSE_UPLOAD, seal, unseal, @@ -69,6 +70,12 @@ class FakeNode: self.ack: dict = {"chat_epoch": 1} # the sealed configuration, opened self.chat_epochs = {1: secrets.token_bytes(32)} self.chat: list[dict] = [] # stored rows, oldest first + # group_roster_resp's devices; this machine's own key is a root of its account. + self.roster: list[dict] = [{ + "user_id": USER_ID, "username": "alice-test", "pk_ed25519": self.identity.pk_ed, + "pk_x25519": self.identity.pk_x, "added_by_pk": "", "add_sig": "", + "add_nonce": "", "add_ts": 0}] + self.members: dict[str, Identity] = {} # sender → the device it signs with self.admin_ops: dict[str, dict] = {} def add(self, path: str, name: str, data: bytes, uploader_id: str = "") -> dict: @@ -86,17 +93,47 @@ class FakeNode: return GroupSession(None, FakeChannel(self), GROUP_ID, "node-1", NODE_PK, self.gek, dict(self.ack), self.identity, self.nonce, self.role) + def list_device(self, sender: str, device: Identity, *, + countersigned_by: Identity | None = None, proof: bool = True) -> None: + """A roster row for `sender`'s device: a root, or countersigned by another + of its devices — with the signature, or (`proof=False`) naming the + countersigner and carrying nothing, as a lying node would.""" + row = {"user_id": "u-" + sender, "username": sender, "pk_ed25519": device.pk_ed, + "pk_x25519": device.pk_x, "added_by_pk": "", "add_sig": "", + "add_nonce": "", "add_ts": 0} + if countersigned_by is not None: + row["added_by_pk"] = countersigned_by.pk_ed + if proof: + nonce, ts = secrets.token_bytes(32), 1_700_000_000 + row.update(add_nonce=base64.b64encode(nonce).decode(), add_ts=ts, + add_sig=base64.b64encode(countersigned_by.sk_ed.sign( + device_add_transcript( + node_pk_b64=NODE_PK, user_id="u-" + sender, + pk_ed25519_b64=device.pk_ed, + pk_x25519_b64=device.pk_x, + nonce_node=nonce, ts=ts))).decode()) + self.roster.append(row) + def say(self, text: str, *, sender: str = "bob-test", epoch: int = 1, - forge: bool = False) -> None: - """Another member's message, sealed and signed as a client does; `forge` - signs it with a key other than the device it names.""" - other = Identity(Ed25519PrivateKey.generate(), X25519PrivateKey.generate()) + forge: bool = False, device: Identity | None = None) -> Identity: + """Another member's message, sealed and signed as a client does, from + `device` (default: the sender's usual one, listed in the roster as a + root the first time); `forge` signs it with a key other than the device + it names.""" + if device is None: + if sender not in self.members: + self.members[sender] = Identity(Ed25519PrivateKey.generate(), + X25519PrivateKey.generate()) + self.list_device(sender, self.members[sender]) + device = self.members[sender] + other = device sealed = chatbox.seal(self.chat_epochs[epoch], GROUP_ID, epoch, other.pk_ed, base64.b64decode(other.pk_ed), other.sk_ed, {"text": text, "sender_name": sender}) if forge: sealed["sig"] = Ed25519PrivateKey.generate().sign(b"something else") self._store(base64.b64decode(other.pk_ed), epoch, sealed, "u-" + sender) + return other def _store(self, device: bytes, epoch: int, sealed: dict, sender_id: str) -> None: self.chat.append({"id": len(self.chat) + 1, "sender_id": sender_id, @@ -236,6 +273,12 @@ class FakeChannel: return {"type": "chat_keys_resp", "group_id": GROUP_ID, **seal(n.gek, PURPOSE_CHAT_KEYS, "chat_keys_resp", GROUP_ID, payload)} + def _group_roster_req(self, msg): + n = self.node + return {"type": "group_roster_resp", "group_id": GROUP_ID, + **seal(n.gek, PURPOSE_ROSTER, "group_roster_resp", GROUP_ID, + {"devices": n.roster, "node_pk": NODE_PK})} + def _chat_hist(self, msg): rows = self.node.chat if msg.get("before"): |