Experimenting with Rudimentary SOCKS5 support

pyproject.toml

@@ -7,7 +7,7 @@ license = "MIT"
 readme = "README.md"
 
 [tool.poetry.dependencies]
-python = "^3.7"
+python = ">3.12"
 rns = ">=0.9.0"
 
 [tool.poetry.scripts]

rnsh/args.py

@@ -18,17 +18,23 @@ def _split_array_at(arr: [_T], at: _T) -> ([_T], [_T]):
 usage = \
 '''
 Usage:
-    rnsh -l [-c <configdir>] [-i <identityfile> | -s <service_name>] [-v... | -q...] -p
-    rnsh -l [-c <configdir>] [-i <identityfile> | -s <service_name>] [-v... | -q...] 
-            [-b <period>] [-n] [-a <identity_hash>] ([-a <identity_hash>] ...) [-A | -C]
-            [[--] <program> [<arg> ...]]
+    rnsh [--socks5] -l [-c <configdir>] [-i <identityfile> | -s <service_name>] [-v... | -q...] -p
+    rnsh [--socks5] -l [-c <configdir>] [-i <identityfile> | -s <service_name>] [-v... | -q...] 
+                    [-b <period>] [-n] [-a <identity_hash>] ([-a <identity_hash>] ...) [-A | -C]
+                    [[--] <program> [<arg> ...]]
     rnsh [-c <configdir>] [-i <identityfile>] [-v... | -q...] -p
+    rnsh [-c <configdir>] [-i <identityfile>] [-v... | -q...] [-N] [-m] [-w <timeout>]
+         [--socks5 [--socks5-host <host>] [--socks5-port <port>]]
+         <destination_hash>
     rnsh [-c <configdir>] [-i <identityfile>] [-v... | -q...] [-N] [-m] [-w <timeout>] 
          <destination_hash> [[--] <program> [<arg> ...]]
     rnsh -h
     rnsh --version
 
 Options:
+    --socks5                     Enable socks5 proxy mode.
+    --socks5-host HOST           SOCKS5 proxy host
+    --socks5-port PORT           SOCKS5 proxy port
     -c DIR --config DIR          Alternate Reticulum config directory to use
     -i FILE --identity FILE      Specific identity file to use
     -s NAME --service NAME       Service name for identity file if not default
@@ -79,6 +85,20 @@ def __init__(self, argv: [str]):
             args = docopt.docopt(usage, argv=self.docopts_argv[1:], version=f"rnsh {rnsh.__version__}")
             # json.dump(args, sys.stdout)
 
+            self.socks5 = "--socks5" in args
+            self.socks5_host = args.get("--socks5-host") or "127.0.0.1"
+            try:
+                if "--socks5-port" in args:
+                    port_string = args.get("--socks5-port")
+                    if port_string is None:
+                        self.socks5_port = 1080
+                    else:
+                        self.socks5_port = int(port_string)
+                else:
+                    self.socks5_port = 1080
+            except ValueError:
+                print("Invalid value for --socks5-port")
+                sys.exit(1)
             self.listen = args.get("--listen", None) or False
             self.service_name = args.get("--service", None)
             if self.listen and (self.service_name is None or len(self.service_name) > 0):

rnsh/rnsh.py

@@ -48,12 +48,11 @@
 import re
 import contextlib
 import rnsh.args
-import pwd
-import rnsh.protocol as protocol
-import rnsh.helpers as helpers
 import rnsh.loop
 import rnsh.listener as listener
 import rnsh.initiator as initiator
+from rnsh.socksext.socksproxy import SOCKS5Proxy
+from rnsh.socksext.counterpart import SOCKS5CounterPart
 
 module_logger = __logging.getLogger(__name__)
 
@@ -104,12 +103,11 @@ def print_identity(configdir, identitypath, service_name, include_destination: b
 verbose_set = False
 
 
-async def _rnsh_cli_main():
+async def _rnsh_cli_main(args):
     global verbose_set
     log = _get_logger("main")
     _loop = asyncio.get_running_loop()
     rnslogging.set_main_loop(_loop)
-    args = rnsh.args.Args(sys.argv)
     verbose_set = args.verbose > 0
 
     if args.print_identity:
@@ -156,12 +154,43 @@ async def _rnsh_cli_main():
         return 1
 
 
+
+async def _rnsocks_cli_main(args):
+    global verbose_set
+    log = _get_logger("main")
+    _loop = asyncio.get_running_loop()
+    rnslogging.set_main_loop(_loop)
+    args = rnsh.args.Args(sys.argv)
+    verbose_set = args.verbose > 0
+
+    if args.listen:
+        cpart = SOCKS5CounterPart()
+        cpart.run()
+    else:
+        if args.destination is None:
+            print("No destination specified for socks5 client mode, exiting")
+            return 1
+        host = args.socks5_host or "127.0.0.1"
+        port = 1080
+        if args.socks5_port is not None:
+            try:
+                port = int(args.socks5_port)
+            except Exception:
+                print("Invalid socks5 port specified, exiting")
+                return 1
+        proxy = SOCKS5Proxy(host=host, port=port, destination_hash=args.destination)
+        proxy.start()
+
 def rnsh_cli():
     global verbose_set
     return_code = 1
     exc = None
     try:
-        return_code = asyncio.run(_rnsh_cli_main())
+        args = rnsh.args.Args(sys.argv)
+        if args.socks5:
+            return_code = asyncio.run(_rnsocks_cli_main(args))
+        else:
+            return_code = asyncio.run(_rnsh_cli_main(args))
     except SystemExit:
         pass
     except KeyboardInterrupt:

rnsh/socksext/counterpart.py

@@ -0,0 +1,150 @@
+import os
+import sys
+import threading
+import socket
+import time
+import RNS
+
+from rnsh.socksext.socksproxy import SOCKS_APP_NAME
+from rnsh.socksext.protocol import RequestMessage
+
+COUNTERPART_IDENTITY_FILE = "socks5_identity"
+
+
+class SOCKS5CounterPart:
+    def __init__(self, announce_interval: int = 60):
+        self.reticulum = RNS.Reticulum(configdir=None, loglevel=RNS.LOG_INFO)
+        self.identity = self.load_or_create_identity()
+        self.destination = RNS.Destination(
+            self.identity, RNS.Destination.IN, RNS.Destination.SINGLE, SOCKS_APP_NAME
+        )
+        self.channels = {}
+        self.connections = {}
+        self.lock = threading.Lock()
+        self.next_link_id = 0
+        self.running = False
+        self.announce_interval = announce_interval
+
+    def load_or_create_identity(self):
+        if os.path.exists(COUNTERPART_IDENTITY_FILE):
+            identity = RNS.Identity.from_file(COUNTERPART_IDENTITY_FILE)
+            print("Loaded existing identity")
+        else:
+            identity = RNS.Identity()
+            identity.to_file(COUNTERPART_IDENTITY_FILE)
+            print("Created and saved new identity")
+        return identity
+
+    def handle_message(self, message: RequestMessage, link_id: int):
+        try:
+            data = message.data
+            parts = data.split(b":", 2)
+            if len(parts) < 2:
+                print(f"Invalid message format on link {link_id}")
+                return
+            command = parts[0].decode('utf-8')
+            handler_id = int(parts[1].decode('utf-8'))
+            payload = parts[2] if len(parts) > 2 else b""
+
+            if command == "CONNECT":
+                addr, port = payload.decode('utf-8').split(":", 1)
+                port = int(port)
+                print(f"Received CONNECT {handler_id} for {addr}:{port}")
+                sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+                sock.settimeout(10)
+                sock.connect((addr, port))
+                with self.lock:
+                    self.connections[handler_id] = sock
+                threading.Thread(target=self.relay_from_destination, args=(handler_id, sock, link_id), daemon=True).start()
+
+            elif command == "DATA":
+                print(f"Received {len(payload)} bytes for {handler_id}")
+                with self.lock:
+                    if handler_id in self.connections:
+                        sock = self.connections[handler_id]
+                        sock.sendall(payload)
+                    else:
+                        print(f"No connection for {handler_id}")
+        except Exception as e:
+            print(f"Error handling message: {e}")
+
+    def relay_from_destination(self, handler_id: int, sock: socket.socket, link_id: int):
+        max_retries = 5
+        retry_delay = 1  # seconds
+        try:
+            while True:
+                try:
+                    data = sock.recv(4096)
+                    if not data:
+                        print(f"Destination closed for handler {handler_id}")
+                        break
+                except socket.error as e:
+                    print(f"Socket recv error for handler {handler_id}: {e}")
+                    break  # Exit on recv error (destination likely closed)
+
+                with self.lock:
+                    if link_id not in self.channels:
+                        print(f"Channel for link {link_id} gone for handler {handler_id}")
+                        break
+                    channel = self.channels[link_id]
+                    response = RequestMessage()
+                    response.data = f"DATA:{handler_id}:".encode() + data
+
+                retries = 0
+                while retries < max_retries:
+                    try:
+                        channel.send(response)
+                        print(f"Sent {len(data)} bytes back for {handler_id} on link {link_id}")
+                        break
+                    except Exception as e:
+                        retries += 1
+                        print(f"Channel send failed for handler {handler_id} (retry {retries}/{max_retries}): {e}")
+                        if retries == max_retries:
+                            print(f"Max retries reached for handler {handler_id}, giving up")
+                            return  # Exit thread if retries exhausted
+                        time.sleep(retry_delay)
+
+        except Exception as e:
+            print(f"Unexpected error in relay for handler {handler_id}: {e}")
+        finally:
+            with self.lock:
+                if handler_id in self.connections:
+                    del self.connections[handler_id]
+            sock.close()
+
+    def link_established(self, link):
+        link_id = self.next_link_id
+        self.next_link_id += 1
+        print(f"Link {link_id} established from {link.get_remote_identity()}")
+        channel = link.get_channel()
+        channel.register_message_type(RequestMessage)
+        channel.add_message_handler(lambda msg: self.handle_message(msg, link_id))
+        with self.lock:
+            self.channels[link_id] = channel
+
+    def announce_loop(self):
+        while self.running:
+            self.destination.announce()
+            print(f"Announced destination {self.destination.hash.hex()}")
+            time.sleep(self.announce_interval)
+
+    def run(self):
+        print(f"Destination hash: {self.destination.hash.hex()}")
+        self.destination.set_link_established_callback(self.link_established)
+        self.destination.accepts_links(True)
+        self.running = True
+
+        threading.Thread(target=self.announce_loop, daemon=True).start()
+
+        self.destination.announce()
+        print("Counterpart running. Press Ctrl+C to exit.")
+        sys.stdout.flush()
+
+        try:
+            threading.Event().wait()
+        except KeyboardInterrupt:
+            print("Shutting down...")
+            self.running = False
+            self.reticulum.exit_handler()
+            sys.stdout.flush()
+            sys.exit(0)

rnsh/socksext/protocol.py

@@ -0,0 +1,23 @@
+import socket
+import threading
+from typing import Optional
+
+import RNS
+
+
+class SOCKS5Request:
+    def __init__(self, client_socket: socket.socket, addr: str, port: int, handler_id: int):
+        self.client_socket = client_socket
+        self.addr = addr
+        self.port = port
+        self.handler_id = handler_id
+        self.response: Optional[bytes] = None
+        self.event = threading.Event()
+
+
+class RequestMessage(RNS.MessageBase):
+    MSGTYPE = 0x0091
+    def pack(self):
+        return self.data
+    def unpack(self, raw):
+        self.data = raw