clanService/phonebox-global: decentralized phone network over yggdrasil

New service alongside phonebox. Boxes get a random 6 digit number
(100000-999999) signed by the machine's yggdrasil key. A serf gossip
cluster on yggdrasil port 7946 exchanges the signed records;
phonebox-sync verifies key -> address derivation and the signature and
writes the number -> address map to /run/phonebox/numbers for the
asterisk dialplan. Any yggdrasil node running the service can reach any
other; clan-core's ygg-input filter is opened for SIP, RTP and gossip.

Includes a two-node VM test proving directory convergence and a call
over yggdrasil. The existing phonebox service is untouched.

modules/clan/phonebox-global/phonebox-sync.py

@@ -0,0 +1,111 @@
+"""Rebuild the phonebox number directory from serf membership.
+
+Usage: phonebox-sync <serf rpc addr> <state dir>
+
+Runs as a serf event handler. Every alive member carries its number record
+as tags (number, key, sig, owner). A record is accepted only when the
+yggdrasil address it is gossiped from derives from `key` and `sig` is a
+valid ed25519 signature of "phonebox:<number>" under that key, so a number
+can only ever be bound to the address of the node that signed it.
+
+Output, consumed by the asterisk dialplan:
+  <state dir>/numbers/<number>  the yggdrasil address, no trailing newline
+  <state dir>/contacts.txt      "<number> : <owner>" per line, for the fax
+"""
+
+import base64
+import ipaddress
+import json
+import os
+import subprocess
+import sys
+import syslog
+import tempfile
+
+from cryptography.exceptions import InvalidSignature
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
+
+
+def yggdrasil_address(public_key: bytes) -> ipaddress.IPv6Address:
+    """Port of yggdrasil-go address.AddrForKey.
+
+    Invert the key; the address is 0x02, the count of leading one bits,
+    then the bits following the first zero, truncated to 128 bits.
+    """
+    bits = "".join(f"{b ^ 0xFF:08b}" for b in public_key)
+    ones = len(bits) - len(bits.lstrip("1"))
+    body = int(bits[ones + 1:ones + 113], 2).to_bytes(14, "big")
+    return ipaddress.IPv6Address(bytes([0x02, ones]) + body)
+
+
+def verified_record(member: dict) -> tuple[str, str, str, str]:
+    """Return (number, key hex, address, owner) or raise ValueError."""
+    tags = member.get("tags", {})
+    try:
+        number, key_hex, sig = tags["number"], tags["key"], tags["sig"]
+        key = bytes.fromhex(key_hex)
+        host, _ = member["addr"].rsplit(":", 1)
+        address = ipaddress.IPv6Address(host.strip("[]"))
+        signature = base64.b64decode(sig, validate=True)
+    except (KeyError, ValueError) as e:
+        raise ValueError(f"malformed record: {e}")
+    if not (len(number) == 6 and number.isdigit() and number[0] != "0"):
+        raise ValueError(f"invalid number {number!r}")
+    if yggdrasil_address(key) != address:
+        raise ValueError(f"key does not derive to address {address}")
+    try:
+        Ed25519PublicKey.from_public_bytes(key).verify(signature, b"phonebox:" + number.encode())
+    except (ValueError, InvalidSignature):
+        raise ValueError(f"bad signature for number {number}")
+    return number, key_hex, str(address), tags.get("owner", "")
+
+
+def directory(members: list[dict]) -> dict[str, tuple[str, str, str]]:
+    """number -> (key hex, address, owner); on a collision the lowest key wins."""
+    book = {}
+    for member in members:
+        try:
+            number, key, address, owner = verified_record(member)
+        except ValueError as e:
+            syslog.syslog(syslog.LOG_WARNING, f"ignoring {member.get('name')}: {e}")
+            continue
+        if number in book:
+            syslog.syslog(syslog.LOG_WARNING, f"number {number} claimed by keys {book[number][0]} and {key}")
+            if book[number][0] <= key:
+                continue
+        book[number] = (key, address, owner)
+    return book
+
+
+def write_atomic(path: str, content: str) -> None:
+    fd, tmp = tempfile.mkstemp(dir=os.path.dirname(path))
+    with os.fdopen(fd, "w") as f:
+        f.write(content)
+    os.chmod(tmp, 0o644)
+    os.replace(tmp, path)
+
+
+def main() -> None:
+    rpc_addr, state_dir = sys.argv[1:]
+    syslog.openlog("phonebox-sync")
+    out = subprocess.run(
+        ["serf", "members", "-format=json", "-status=alive", f"-rpc-addr={rpc_addr}"],
+        check=True, capture_output=True, text=True,
+    ).stdout
+    book = directory(json.loads(out)["members"])
+
+    numbers_dir = os.path.join(state_dir, "numbers")
+    os.makedirs(numbers_dir, exist_ok=True)
+    for number, (_, address, _) in book.items():
+        write_atomic(os.path.join(numbers_dir, number), address)
+    for stale in set(os.listdir(numbers_dir)) - book.keys():
+        os.unlink(os.path.join(numbers_dir, stale))
+    write_atomic(
+        os.path.join(state_dir, "contacts.txt"),
+        "".join(f"{number}\t\t: \t\t{book[number][2]}\n" for number in sorted(book)),
+    )
+    syslog.syslog(syslog.LOG_INFO, f"directory has {len(book)} numbers")
+
+
+if __name__ == "__main__":
+    main()