"""Tests for phonebox-sync: record verification and directory output. Run via the `phonebox-global-sync` flake check; needs the `cryptography` python package and the `yggdrasil` binary on PATH. """ import base64 import importlib.util import ipaddress import json import os import subprocess import sys import tempfile import unittest from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey spec = importlib.util.spec_from_file_location("phonebox_sync", os.environ["PHONEBOX_SYNC"]) sync = importlib.util.module_from_spec(spec) spec.loader.exec_module(sync) PORT = 7946 class Node: """A phonebox node identity: yggdrasil key, address and signed number.""" def __init__(self, number: str, owner: str = ""): self.key = Ed25519PrivateKey.generate() self.public = self.key.public_key().public_bytes_raw() self.address = str(sync.yggdrasil_address(self.public)) self.number = number self.owner = owner def member(self, name="node", **overrides) -> dict: tags = { "number": self.number, "key": self.public.hex(), "sig": base64.b64encode(self.key.sign(b"phonebox:" + self.number.encode())).decode(), "owner": self.owner, } member = {"name": name, "addr": f"[{self.address}]:{PORT}", "port": PORT, "status": "alive", "tags": tags} for k, v in overrides.items(): (tags if k in tags else member)[k] = v return member def yggdrasil_binary_address(key: Ed25519PrivateKey) -> str: private = key.private_bytes_raw() + key.public_key().public_bytes_raw() out = subprocess.run( ["yggdrasil", "-useconf", "-address"], input=json.dumps({"PrivateKey": private.hex()}), check=True, capture_output=True, text=True, ) return out.stdout.strip() class AddressDerivation(unittest.TestCase): def test_known_vector(self): # A real clan machine key and the address yggdrasil assigned to it. key = bytes.fromhex("0ec53986a2bdca8a43f74063aeab6a958fc697c528c83e7281365072926886f0") self.assertEqual(str(sync.yggdrasil_address(key)), "204:2758:cf2b:a846:aeb7:8117:f38a:2a92") def test_matches_yggdrasil_binary(self): for _ in range(16): key = Ed25519PrivateKey.generate() expected = ipaddress.IPv6Address(yggdrasil_binary_address(key)) self.assertEqual(sync.yggdrasil_address(key.public_key().public_bytes_raw()), expected) class RecordVerification(unittest.TestCase): def setUp(self): self.node = Node("482913", owner="alice") def test_valid_record(self): number, key, address, owner = sync.verified_record(self.node.member()) self.assertEqual((number, key, address, owner), ("482913", self.node.public.hex(), self.node.address, "alice")) def test_record_from_foreign_address_is_rejected(self): other = Node("111111") with self.assertRaisesRegex(ValueError, "does not derive"): sync.verified_record(self.node.member(addr=f"[{other.address}]:{PORT}")) def test_signature_over_other_number_is_rejected(self): with self.assertRaisesRegex(ValueError, "bad signature"): sync.verified_record(self.node.member(number="999999")) def test_garbage_signature_is_rejected(self): with self.assertRaisesRegex(ValueError, "malformed"): sync.verified_record(self.node.member(sig="not base64!")) def test_missing_tags_are_rejected(self): with self.assertRaisesRegex(ValueError, "malformed"): sync.verified_record({"name": "foreign", "addr": f"[{self.node.address}]:{PORT}", "tags": {}}) def test_number_must_be_six_digits_without_leading_zero(self): for bad in ["012345", "12345", "1234567", "12a456", ""]: node = Node(bad) with self.assertRaisesRegex(ValueError, "invalid number", msg=bad): sync.verified_record(node.member()) def test_owner_defaults_to_empty(self): member = self.node.member() del member["tags"]["owner"] self.assertEqual(sync.verified_record(member)[3], "") class Directory(unittest.TestCase): def test_invalid_records_are_skipped_not_fatal(self): good = Node("482913", owner="alice") bad = Node("555555") members = [bad.member(sig="AAAA"), good.member(), {"name": "x", "addr": "[200::1]:7946", "tags": {}}] self.assertEqual(sync.directory(members), {"482913": (good.public.hex(), good.address, "alice")}) def test_collision_lowest_key_wins_regardless_of_order(self): a, b = Node("482913"), Node("482913") winner = min((a, b), key=lambda n: n.public.hex()) for members in ([a.member("a"), b.member("b")], [b.member("b"), a.member("a")]): self.assertEqual(sync.directory(members)["482913"][1], winner.address) class Main(unittest.TestCase): def run_sync(self, members: list[dict], state: str) -> None: bindir = tempfile.mkdtemp() with open(os.path.join(bindir, "serf"), "w") as f: f.write("#!/bin/sh\ncat <<'EOF'\n" + json.dumps({"members": members}) + "\nEOF\n") os.chmod(os.path.join(bindir, "serf"), 0o755) env_path = os.environ["PATH"] os.environ["PATH"] = bindir + os.pathsep + env_path argv = sys.argv sys.argv = ["phonebox-sync", "127.0.0.1:7373", state] try: sync.main() finally: sys.argv = argv os.environ["PATH"] = env_path def test_writes_numbers_and_contacts_and_removes_stale(self): state = tempfile.mkdtemp() os.makedirs(os.path.join(state, "numbers")) with open(os.path.join(state, "numbers", "111111"), "w") as f: f.write("200::dead") a, b = Node("482913", owner="alice"), Node("555555", owner="bob") self.run_sync([b.member("b"), a.member("a")], state) numbers = os.path.join(state, "numbers") self.assertEqual(sorted(os.listdir(numbers)), ["482913", "555555"]) with open(os.path.join(numbers, "482913")) as f: self.assertEqual(f.read(), a.address) # no trailing newline: read by FILE() in the dialplan with open(os.path.join(state, "contacts.txt")) as f: self.assertEqual(f.read(), "482913\t\t: \t\talice\n555555\t\t: \t\tbob\n") def test_empty_membership_clears_directory(self): state = tempfile.mkdtemp() self.run_sync([Node("482913").member()], state) self.run_sync([], state) self.assertEqual(os.listdir(os.path.join(state, "numbers")), []) with open(os.path.join(state, "contacts.txt")) as f: self.assertEqual(f.read(), "") if __name__ == "__main__": unittest.main()