Mixnet: Add basic structure and topology construction (#44)

mixnet/bls.py

@@ -0,0 +1,13 @@
+from typing import TypeAlias
+
+import blspy
+
+from mixnet.utils import random_bytes
+
+BlsPrivateKey: TypeAlias = blspy.PrivateKey
+BlsPublicKey: TypeAlias = blspy.G1Element
+
+
+def generate_bls() -> BlsPrivateKey:
+    seed = random_bytes(32)
+    return blspy.BasicSchemeMPL.key_gen(seed)

mixnet/fisheryates.py

@@ -0,0 +1,21 @@
+import random
+from typing import List
+
+
+class FisherYates:
+    @staticmethod
+    def shuffle(elements: List, entropy: bytes) -> List:
+        """
+        Fisher-Yates shuffling algorithm.
+        In Python, random.shuffle implements the Fisher-Yates shuffling.
+        https://en.wikipedia.org/wiki/Fisher%E2%80%93Yates_shuffle
+        https://softwareengineering.stackexchange.com/a/215780
+        :param elements: elements to be shuffled
+        :param entropy: a seed for deterministic sampling
+        """
+        out = elements.copy()
+        random.seed(a=entropy, version=2)
+        random.shuffle(out)
+        # reset seed
+        random.seed()
+        return out

mixnet/mixnet.py

@@ -0,0 +1,66 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import List, TypeAlias
+
+from cryptography.hazmat.primitives.asymmetric.x25519 import (
+    X25519PrivateKey,
+    X25519PublicKey,
+)
+
+from mixnet.bls import BlsPrivateKey, BlsPublicKey
+from mixnet.fisheryates import FisherYates
+
+NodeId: TypeAlias = BlsPublicKey
+# 32-byte that represents an IP address and a port of a mix node.
+NodeAddress: TypeAlias = bytes
+
+
+@dataclass
+class Mixnet:
+    mix_nodes: List[MixNode]
+
+    # Build a new topology deterministically using an entropy.
+    # The entropy is expected to be injected from outside.
+    #
+    # TODO: Implement constructing a new topology in advance to minimize the topology transition time.
+    #       https://www.notion.so/Mixnet-Specification-807b624444a54a4b88afa1cc80e100c2?pvs=4#9a7f6089e210454bb11fe1c10fceff68
+    def build_topology(
+        self,
+        entropy: bytes,
+        n_layers: int,
+        n_nodes_per_layer: int,
+    ) -> MixnetTopology:
+        num_nodes = n_nodes_per_layer * n_layers
+        assert num_nodes < len(self.mix_nodes)
+
+        shuffled = FisherYates.shuffle(self.mix_nodes, entropy)
+        sampled = shuffled[:num_nodes]
+        layers = []
+        for l in range(n_layers):
+            start = l * n_nodes_per_layer
+            layer = sampled[start : start + n_nodes_per_layer]
+            layers.append(layer)
+        return MixnetTopology(layers)
+
+
+@dataclass
+class MixNode:
+    identity_public_key: BlsPublicKey
+    encryption_public_key: X25519PublicKey
+    addr: NodeAddress
+
+    def __init__(
+        self,
+        identity_private_key: BlsPrivateKey,
+        encryption_private_key: X25519PrivateKey,
+        addr: NodeAddress,
+    ):
+        self.identity_public_key = identity_private_key.get_g1()
+        self.encryption_public_key = encryption_private_key.public_key()
+        self.addr = addr
+
+
+@dataclass
+class MixnetTopology:
+    layers: List[List[MixNode]]

mixnet/test_fisheryates.py

@@ -0,0 +1,21 @@
+from unittest import TestCase
+
+from mixnet.fisheryates import FisherYates
+
+
+class TestFisherYates(TestCase):
+    def test_shuffle(self):
+        entropy = b"hello"
+        elems = [1, 2, 3, 4, 5]
+
+        shuffled1 = FisherYates.shuffle(elems, entropy)
+        self.assertEqual(sorted(elems), sorted(shuffled1))
+
+        # shuffle again with the same entropy
+        shuffled2 = FisherYates.shuffle(elems, entropy)
+        self.assertEqual(shuffled1, shuffled2)
+
+        # shuffle with a different entropy
+        shuffled3 = FisherYates.shuffle(elems, b"world")
+        self.assertNotEqual(shuffled1, shuffled3)
+        self.assertEqual(sorted(elems), sorted(shuffled3))

mixnet/test_mixnet.py

@@ -0,0 +1,21 @@
+from unittest import TestCase
+
+from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
+
+from mixnet.bls import generate_bls
+from mixnet.mixnet import Mixnet, MixNode
+from mixnet.utils import random_bytes
+
+
+class TestMixnet(TestCase):
+    def test_build_topology(self):
+        nodes = [
+            MixNode(generate_bls(), X25519PrivateKey.generate(), random_bytes(32))
+            for _ in range(12)
+        ]
+        mixnet = Mixnet(nodes)
+
+        topology = mixnet.build_topology(b"entropy", 3, 3)
+        self.assertEqual(len(topology.layers), 3)
+        for layer in topology.layers:
+            self.assertEqual(len(layer), 3)

mixnet/utils.py

@@ -0,0 +1,6 @@
+from random import randint
+
+
+def random_bytes(size: int) -> bytes:
+    assert size >= 0
+    return bytes([randint(0, 255) for _ in range(size)])

requirements.txt

@@ -1,2 +1,6 @@
-blspy~=1.0.16
-scipy~=1.10.1
+blspy==1.0.16
+cffi==1.16.0
+cryptography==41.0.7
+numpy==1.26.2
+pycparser==2.21
+scipy==1.10.1