WIP: Ping interop in py-libp2p and rust-libp2p

Makefile

@@ -46,7 +46,7 @@ typecheck:
 	pre-commit run mypy-local --all-files && pre-commit run pyrefly-local --all-files
 
 test:
-	python -m pytest tests -n auto
+	python -m pytest tests -n auto --timeout=300
 
 pr: clean fix lint typecheck test
 

docs/conf.py

@@ -290,7 +290,7 @@
 ]
 
 # Prevent autodoc from trying to import module from tests.factories
-autodoc_mock_imports = ["tests.factories"]
+autodoc_mock_imports = ["tests.factories", "redis"]
 
 # Documents to append as an appendix to all manuals.
 # texinfo_appendices = []

docs/index.rst

@@ -18,6 +18,7 @@ The Python implementation of the libp2p networking stack
 
     Examples <examples>
     API <libp2p>
+    Interop <interop>
 
 .. toctree::
     :maxdepth: 1

docs/interop.rst

@@ -0,0 +1,29 @@
+interop package
+===============
+
+Submodules
+----------
+
+interop.arch module
+-------------------
+
+.. automodule:: interop.arch
+   :members:
+   :show-inheritance:
+   :undoc-members:
+
+interop.lib module
+------------------
+
+.. automodule:: interop.lib
+   :members:
+   :show-inheritance:
+   :undoc-members:
+
+Module contents
+---------------
+
+.. automodule:: interop
+   :members:
+   :show-inheritance:
+   :undoc-members:

examples/ping/ping.py

@@ -1,4 +1,5 @@
 import argparse
+import logging
 
 import multiaddr
 import trio
@@ -55,8 +56,8 @@ async def send_ping(stream: INetStream) -> None:
 
 
 async def run(port: int, destination: str) -> None:
-    listen_addr = multiaddr.Multiaddr(f"/ip4/0.0.0.0/tcp/{port}")
-    host = new_host(listen_addrs=[listen_addr])
+    listen_addr = multiaddr.Multiaddr(f"/ip4/127.0.0.1/tcp/{port}")
+    host = new_host()
 
     async with host.run(listen_addrs=[listen_addr]), trio.open_nursery() as nursery:
         # Start the peer-store cleanup task
@@ -106,8 +107,20 @@ def main() -> None:
         type=str,
         help=f"destination multiaddr string, e.g. {example_maddr}",
     )
+    parser.add_argument(
+        "-v", "--verbose", action="store_true", help="enable verbose logging"
+    )
+
     args = parser.parse_args()
 
+    if args.verbose:
+        # Enable even more detailed logging
+        logging.getLogger("libp2p").setLevel(logging.DEBUG)
+        logging.getLogger("libp2p.network").setLevel(logging.DEBUG)
+        logging.getLogger("libp2p.transport").setLevel(logging.DEBUG)
+        logging.getLogger("libp2p.security").setLevel(logging.DEBUG)
+        logging.getLogger("libp2p.stream_muxer").setLevel(logging.DEBUG)
+
     try:
         trio.run(run, *(args.port, args.destination))
     except KeyboardInterrupt:

interop/README.md

@@ -0,0 +1,25 @@
+These commands are to be run in `./interop/exec`
+
+## Redis
+
+```bash
+docker run -p 6379:6379 -it redis:latest
+```
+
+## Listener
+
+```bash
+transport=tcp ip=127.0.0.1  redis_addr=6379 port=8001 test_timeout_seconds=180 security=noise muxer=yamux is_dialer=false python3 interop/exec/native_ping.py
+```
+
+## Dialer
+
+```bash
+transport=tcp ip=127.0.0.1  redis_addr=6379 port=8001 test_timeout_seconds=180 security=noise  muxer=yamux is_dialer=true python3 interop/exec/native_ping.py
+```
+
+## From the Rust-side (Listener)
+
+```bash
+RUST_LOG=debug redis_addr=localhost:6379 ip="0.0.0.0" transport=tcp security=noise muxer=yamux is_dialer="false" cargo run --bin native_ping
+```

interop/arch.py

@@ -0,0 +1,157 @@
+from dataclasses import (
+    dataclass,
+)
+import logging
+
+from cryptography.hazmat.primitives.asymmetric import (
+    x25519,
+)
+import multiaddr
+import redis
+import trio
+
+from libp2p import (
+    new_host,
+)
+from libp2p.crypto.keys import (
+    KeyPair,
+)
+from libp2p.crypto.rsa import (
+    create_new_key_pair,
+)
+from libp2p.custom_types import (
+    TProtocol,
+)
+from libp2p.security.insecure.transport import (
+    PLAINTEXT_PROTOCOL_ID,
+    InsecureTransport,
+)
+from libp2p.security.noise.transport import (
+    Transport as NoiseTransport,
+)
+import libp2p.security.secio.transport as secio
+from libp2p.stream_muxer.mplex.mplex import (
+    MPLEX_PROTOCOL_ID,
+    Mplex,
+)
+from libp2p.stream_muxer.yamux.yamux import (
+    PROTOCOL_ID as YAMUX_PROTOCOL_ID,
+    Yamux,
+)
+
+# Configure detailed logging
+logging.basicConfig(
+    level=logging.DEBUG,
+    format="%(asctime)s - %(levelname)s - %(message)s",
+    handlers=[
+        logging.StreamHandler(),
+        logging.FileHandler("ping_debug.log", mode="w", encoding="utf-8"),
+    ],
+)
+
+
+def generate_new_rsa_identity() -> KeyPair:
+    return create_new_key_pair()
+
+
+def create_noise_keypair():
+    """Create a Noise protocol keypair for secure communication"""
+    try:
+        x25519_private_key = x25519.X25519PrivateKey.generate()
+
+        class NoisePrivateKey:
+            def __init__(self, key):
+                self._key = key
+
+            def to_bytes(self):
+                return self._key.private_bytes_raw()
+
+            def public_key(self):
+                return NoisePublicKey(self._key.public_key())
+
+            def get_public_key(self):
+                return NoisePublicKey(self._key.public_key())
+
+        class NoisePublicKey:
+            def __init__(self, key):
+                self._key = key
+
+            def to_bytes(self):
+                return self._key.public_bytes_raw()
+
+        return NoisePrivateKey(x25519_private_key)
+    except Exception as e:
+        logging.error(f"Failed to create Noise keypair: {e}")
+        return None
+
+
+async def build_host(transport: str, ip: str, port: str, sec_protocol: str, muxer: str):
+    match (sec_protocol, muxer):
+        case ("insecure", "mplex"):
+            key_pair = create_new_key_pair()
+            host = new_host(
+                key_pair,
+                {TProtocol(MPLEX_PROTOCOL_ID): Mplex},
+                {
+                    TProtocol(PLAINTEXT_PROTOCOL_ID): InsecureTransport(key_pair),
+                    TProtocol(secio.ID): secio.Transport(key_pair),
+                },
+            )
+            muladdr = multiaddr.Multiaddr(f"/ip4/{ip}/tcp/{port}")
+            return (host, muladdr)
+        case ("insecure", "yamux"):
+            key_pair = create_new_key_pair()
+            host = new_host(
+                key_pair,
+                {TProtocol(YAMUX_PROTOCOL_ID): Yamux},
+                {
+                    TProtocol(PLAINTEXT_PROTOCOL_ID): InsecureTransport(key_pair),
+                    TProtocol(secio.ID): secio.Transport(key_pair),
+                },
+            )
+            muladdr = multiaddr.Multiaddr(f"/ip4/{ip}/tcp/{port}")
+            return (host, muladdr)
+        case ("noise", "yamux"):
+            key_pair = generate_new_rsa_identity()
+            logging.debug("Generated RSA keypair")
+
+            noise_privkey = create_noise_keypair()
+            if not noise_privkey:
+                print("[ERROR] Failed to create Noise keypair")
+                return 1
+            logging.debug("Generated Noise keypair")
+
+            noise_transport = NoiseTransport(key_pair, noise_privkey)
+            logging.debug(f"Noise transport initialized: {noise_transport}")
+            sec_opt = {TProtocol("/noise"): noise_transport}
+            muxer_opt = {TProtocol(YAMUX_PROTOCOL_ID): Yamux}
+
+            logging.info(f"Using muxer: {muxer_opt}")
+
+            host = new_host(key_pair=key_pair, sec_opt=sec_opt, muxer_opt=muxer_opt)
+            muladdr = multiaddr.Multiaddr(f"/ip4/{ip}/tcp/{port}")
+            return (host, muladdr)
+        case _:
+            raise ValueError("Protocols not supported")
+
+
+@dataclass
+class RedisClient:
+    client: redis.Redis
+
+    def brpop(self, key: str, timeout: float) -> list[str]:
+        result = self.client.brpop([key], timeout)
+        return [result[1]] if result else []
+
+    def rpush(self, key: str, value: str) -> None:
+        self.client.rpush(key, value)
+
+
+async def main():
+    client = RedisClient(redis.Redis(host="localhost", port=6379, db=0))
+    client.rpush("test", "hello")
+    print(client.blpop("test", timeout=5))
+
+
+if __name__ == "__main__":
+    trio.run(main)

interop/exec/config/mod.py

@@ -0,0 +1,54 @@
+from dataclasses import (
+    dataclass,
+)
+import os
+
+
+def str_to_bool(val: str) -> bool:
+    return val.lower() in ("true", "1")
+
+
+class ConfigError(Exception):
+    """Raised when the required environment variables are missing or invalid"""
+
+
+@dataclass
+class Config:
+    transport: str
+    sec_protocol: str | None
+    muxer: str | None
+    ip: str
+    is_dialer: bool
+    test_timeout: int
+    redis_addr: str
+    port: str
+
+    @classmethod
+    def from_env(cls) -> "Config":
+        try:
+            transport = os.environ["transport"]
+            ip = os.environ["ip"]
+        except KeyError as e:
+            raise ConfigError(f"{e.args[0]} env variable not set") from None
+
+        try:
+            is_dialer = str_to_bool(os.environ.get("is_dialer", "true"))
+            test_timeout = int(os.environ.get("test_timeout", "180"))
+        except ValueError as e:
+            raise ConfigError(f"Invalid value in env: {e}") from None
+
+        redis_addr = os.environ.get("redis_addr", 6379)
+        sec_protocol = os.environ.get("security")
+        muxer = os.environ.get("muxer")
+        port = os.environ.get("port", "8000")
+
+        return cls(
+            transport=transport,
+            sec_protocol=sec_protocol,
+            muxer=muxer,
+            ip=ip,
+            is_dialer=is_dialer,
+            test_timeout=test_timeout,
+            redis_addr=redis_addr,
+            port=port,
+        )

interop/exec/native_ping.py

@@ -0,0 +1,33 @@
+import trio
+
+from interop.exec.config.mod import (
+    Config,
+    ConfigError,
+)
+from interop.lib import (
+    run_test,
+)
+
+
+async def main() -> None:
+    try:
+        config = Config.from_env()
+    except ConfigError as e:
+        print(f"Config error: {e}")
+        return
+
+    # Uncomment and implement when ready
+    _ = await run_test(
+        config.transport,
+        config.ip,
+        config.port,
+        config.is_dialer,
+        config.test_timeout,
+        config.redis_addr,
+        config.sec_protocol,
+        config.muxer,
+    )
+
+
+if __name__ == "__main__":
+    trio.run(main)