refactoring to make it better structured

hack/compose.yml

@@ -25,5 +25,4 @@ services:
 networks:
   default:
     driver: bridge
-    enable_ipv6: true
+    # internal: true
-    internal: true

hack/exec-iperf

@@ -1,15 +0,0 @@
-#!/usr/bin/env bash
-set -e
-
-
-node=$1
-
-if [[ ! $node =~ [1-9]+ ]]; then
-    echo "node argument missing. should be '1' , '2' etc"
-    exit 2
-fi
-shift
-
-ip="10.55.$node.77"
-
-exec iperf3 --client "$ip"  --port 5001

hack/exec-tool

@@ -0,0 +1,37 @@
+#!/usr/bin/env bash
+set -e
+
+tool=${1?:"tool argument missing. should be one of iperf3,ping,curl,ab"}
+node=${2?:"node argument missing. should be '1' , '2' etc"}
+shift 2
+
+ip="10.55.$node.77"
+connect_timeout_sec=3
+
+function with_set_x() {
+    set -x
+    "$@"
+    {
+        ec=$?
+        set +x
+        return $ec
+    } 2>/dev/null
+}
+
+case "$tool" in
+ping)
+    with_set_x exec ping -W $connect_timeout_sec "$@" "$ip"
+    ;;
+iperf3)
+    port=5001
+    with_set_x exec iperf3 --client "$ip" --port=$port --connect-timeout=$connect_timeout_sec "$@"
+    ;;
+curl)
+    port=8080
+    with_set_x exec curl "http://$ip:$port/" -v --connect-timeout $connect_timeout_sec "$@"
+    ;;
+ab)
+    port=8080
+    with_set_x exec ab -n 100 -c 20 -s $connect_timeout_sec "$@" "http://$ip:$port/" 
+    ;;
+esac

hack/run-benchmark

@@ -0,0 +1,40 @@
+#!/usr/bin/env bash
+set -e
+cd "$(dirname "$0")"
+
+function with_set_x() {
+    set -x
+    "$@"
+    { ec=$?; set +x;return $ec; } 2>/dev/null
+}
+
+
+./test-bed up -d
+
+
+
+benchmark() {
+    local sshuttle_bin="${1?:}"
+    echo -e "\n======== Benchmarking sshuttle: $sshuttle_bin  ========"
+    if [[ "$sshuttle_bin" == dev ]]; then
+        sshuttle_bin="../run"
+    fi
+    SSHUTTLE_BIN=$sshuttle_bin ./exec-sshuttle 1 --listen 55771 &
+    sshuttle_pid=$!
+    trap 'kill -0 $sshuttle_pid &>/dev/null && kill -15 $sshuttle_pid' EXIT
+    while ! nc -z localhost 55771; do sleep 0.1; done
+    sleep 1
+    ./exec-tool iperf3 1 --time=4
+    with_set_x kill -15 $sshuttle_pid
+    wait $sshuttle_pid || true
+}
+
+
+if [[ "$1" ]]; then
+    benchmark "$1"
+else
+    benchmark "${SSHUTTLE_BIN:-/bin/sshuttle}"
+    benchmark dev
+fi
+
+

hack/setup.service

@@ -5,16 +5,20 @@ set -e
 
 echo -e ">>> Setting up $(hostname) | id: $(id) | $(python --version) \nip: $(ip a)\n route: $(ip r)"
 
+function with_set_x() {
+    set -x
+    "$@"
+    { ec=$?; set +x;return $ec; } 2>/dev/null
+}
+
+
 iface="$(ip route | awk '/default/ { print $5 }')"
 default_gw="$(ip route | awk '/default/ { print $3 }')"
 for addr in ${IP_ADDRESSES//,/ }; do
     echo ">>> Adding $addr to interface $iface"
     net_addr=$(ipcalc -n "$addr" | awk -F= '{print $2}')
-    (
+    with_set_x ip addr add "$addr" dev "$iface"
-        set -ex
+    with_set_x ip route add "$net_addr" via "$default_gw" dev "$iface" # so that sshuttle -N can discover routes
-        ip addr add "$addr" dev "$iface"
-        ip route add "$net_addr" via "$default_gw" dev "$iface" # so that sshuttle -N can discover routes
-    )
 done
 
 echo ">>> Starting iperf3 server"

hack/test-bed

@@ -0,0 +1,30 @@
+#!/usr/bin/env bash
+set -e
+cd "$(dirname "$0")"
+
+
+if [[ -z $1 || $1 = -* ]]; then
+    set -- up "$@"
+fi
+
+function with_set_x() {
+    set -x
+    "$@"
+    { ec=$?; set +x;return $ec; } 2>/dev/null
+}
+
+function build() {
+    # podman build -t ghcr.io/sshuttle/sshuttle-testbed .
+    with_set_x docker build -t ghcr.io/sshuttle/sshuttle-testbed -f Containerfile .
+}
+
+function compose() {
+    # podman-compose "$@"
+    with_set_x docker compose "$@"
+}
+
+
+if [[ $* = *--build* ]]; then
+    build
+fi
+compose "$@"

hack/test-bed-up

@@ -1,9 +0,0 @@
-#!/usr/bin/env bash
-set -e
-cd "$(dirname "$0")"
-
-# podman build -t ghcr.io/sshuttle/sshuttle-testbed .
-# podman-compose up
-
-docker build -t ghcr.io/sshuttle/sshuttle-testbed -f Containerfile .
-docker compose up

sshuttle/__main__.py

@@ -3,6 +3,9 @@ import sys
 import os
 from sshuttle.cmdline import main
 from sshuttle.helpers import debug3
+from sshuttle import __version__
+
+debug3("Starting cmd %r (pid:%s) | sshuttle: %s | Python: %s" % (sys.argv, os.getpid(), __version__, sys.version))
 exit_code = main()
-debug3("Exiting process %r (pid:%s) with code %s" % (sys.argv, os.getpid(), exit_code,))
+debug3("Exiting cmd %r (pid:%s) with code %s" % (sys.argv, os.getpid(), exit_code,))
 sys.exit(exit_code)

sshuttle/assembler.py

@@ -3,24 +3,27 @@ import zlib
 import types
 import platform
 
-verbosity = verbosity  # noqa: F821 must be a previously defined global
+stdin = stdin  # type: typing.BinaryIO # noqa: F821 must be a previously defined global
+verbosity = verbosity  # type: int # noqa: F821 must be a previously defined global
 if verbosity > 0:
     sys.stderr.write(' s: Running server on remote host with %s (version %s)\n'
                      % (sys.executable, platform.python_version()))
+
 z = zlib.decompressobj()
+
 while 1:
-    name = sys.stdin.readline().strip()
+    name = stdin.readline().strip()
     if name:
-        # python2 compat: in python2 sys.stdin.readline().strip() -> str
+        # python2 compat: in python2 stdin.readline().strip() -> str
-        #                 in python3 sys.stdin.readline().strip() -> bytes
+        #                 in python3 stdin.readline().strip() -> bytes
         # (see #481)
         if sys.version_info >= (3, 0):
             name = name.decode("ASCII")
-        nbytes = int(sys.stdin.readline())
+        nbytes = int(stdin.readline())
         if verbosity >= 2:
             sys.stderr.write(' s: assembling %r (%d bytes)\n'
                              % (name, nbytes))
-        content = z.decompress(sys.stdin.read(nbytes))
+        content = z.decompress(stdin.read(nbytes))
 
         module = types.ModuleType(name)
         parents = name.rsplit(".", 1)
@@ -44,6 +47,7 @@ sshuttle.helpers.verbose = verbosity
 
 import sshuttle.cmdline_options as options  # noqa: E402
 from sshuttle.server import main  # noqa: E402
+
 main(options.latency_control, options.latency_buffer_size,
      options.auto_hosts, options.to_nameserver,
      options.auto_nets)

sshuttle/client.py

@@ -391,14 +391,14 @@ class FirewallClient:
                        'Command=%r' % (skipped_text, self.argv))
                 continue
 
-            method_name = line.strip()[6:]
+            method_name = line[6:-1]
             self.method = get_method(method_name.decode("ASCII"))
             self.method.set_firewall(self)
             success = True
             break
 
         if not success:
-            raise Fatal("All attempts to run firewall client with elevated privileges were failed.")
+            raise Fatal("All attempts to run firewall client process with elevated privileges were failed.")
 
     def setup(self, subnets_include, subnets_exclude, nslist,
               redirectport_v6, redirectport_v4, dnsport_v6, dnsport_v4, udp,
@@ -461,9 +461,9 @@ class FirewallClient:
                          (udp, user, group, bytes(self.tmark, 'ascii'), os.getpid()))
         self.pfile.flush()
 
-        line = self.pfile.readline().strip()
+        line = self.pfile.readline()
         self.check()
-        if line != b'STARTED':
+        if line != b'STARTED\n':
             raise Fatal('%r expected STARTED, got %r' % (self.argv, line))
 
     def sethostip(self, hostname, ip):
@@ -615,7 +615,7 @@ def _main(tcp_listener, udp_listener, fw, ssh_cmd, remotename,
     debug1('Connecting to server...')
 
     try:
-        (serverproc, serversock) = ssh.connect(
+        (serverproc, rfile, wfile) = ssh.connect(
             ssh_cmd, remotename, python,
             stderr=ssyslog._p and ssyslog._p.stdin,
             add_cmd_delimiter=add_cmd_delimiter,
@@ -630,7 +630,6 @@ def _main(tcp_listener, udp_listener, fw, ssh_cmd, remotename,
             raise Fatal("failed to establish ssh session (1)")
         else:
             raise
-    rfile, wfile = serversock.makefile("rb"), serversock.makefile("wb")
     mux = Mux(rfile, wfile)
     handlers.append(mux)
 
@@ -887,10 +886,7 @@ def main(listenip_v6, listenip_v4,
 
     # listenip_v4 contains user specified value or it is set to "auto".
     if listenip_v4 == "auto":
-        if sys.platform == 'win32':
+        listenip_v4 = ('127.0.0.1' if avail.loopback_port else '0.0.0.0', 0)
-            listenip_v4 = ('0.0.0.0', 0)  # windivert method won't work with loopback interface
-        else:
-            listenip_v4 = ('127.0.0.1', 0)
         debug1("Using default IPv4 listen address " + listenip_v4[0])
 
     # listenip_v6 is...
@@ -901,10 +897,7 @@ def main(listenip_v6, listenip_v4,
         debug1("IPv6 disabled by --disable-ipv6")
     if listenip_v6 == "auto":
         if avail.ipv6:
-            if sys.platform == 'win32':
+            listenip_v6 = ('::1' if avail.loopback_port else '::', 0)
-                listenip_v6 = ('::', 0)  # windivert method won't work with loopback interface
-            else:
-                listenip_v6 = ('::1', 0)
             debug1("IPv6 enabled: Using default IPv6 listen address " + listenip_v6[0])
         else:
             debug1("IPv6 disabled since it isn't supported by method "

sshuttle/firewall.py

@@ -84,14 +84,17 @@ def firewall_exit(signum, frame):
     # the typical exit process as described above.
     global sshuttle_pid
     if sshuttle_pid:
-        debug1("Relaying SIGINT to sshuttle process %d\n" % sshuttle_pid)
+        debug1("Relaying interupt signal to sshuttle process %d\n" % sshuttle_pid)
-        os.kill(sshuttle_pid, signal.SIGINT)
+        if sys.platform == 'win32':
+            sig = signal.CTRL_C_EVENT
+        else:
+            sig = signal.SIGINT
+        os.kill(sshuttle_pid, sig)
 
 
-# Isolate function that needs to be replaced for tests
+def _setup_daemon_for_unix_like():
-def _setup_daemon_unix():
     if not is_admin_user():
-        raise Fatal('You must be root (or enable su/sudo) to set the firewall')
+        raise Fatal('You must have root privileges (or enable su/sudo) to set the firewall')
 
     # don't disappear if our controlling terminal or stdout/stderr
     # disappears; we still have to clean up.
@@ -115,7 +118,7 @@ def _setup_daemon_unix():
     return sys.stdin, sys.stdout
 
 
-def _setup_daemon_windows():
+def _setup_daemon_for_windows():
     if not is_admin_user():
         raise Fatal('You must be administrator to set the firewall')
 
@@ -128,7 +131,7 @@ def _setup_daemon_windows():
         debug3('Using shared socket for communicating with sshuttle client process')
         socket_share_data_b64 = line[len(socket_share_data_prefix):]
         socket_share_data = base64.b64decode(socket_share_data_b64)
-        sock = socket.fromshare(socket_share_data)
+        sock = socket.fromshare(socket_share_data)  # type: socket.socket
         sys.stdin = io.TextIOWrapper(sock.makefile('rb', buffering=0))
         sys.stdout = io.TextIOWrapper(sock.makefile('wb', buffering=0), write_through=True)
         sock.close()
@@ -140,10 +143,11 @@ def _setup_daemon_windows():
     return sys.stdin, sys.stdout
 
 
+# Isolate function that needs to be replaced for tests
 if sys.platform == 'win32':
-    setup_daemon = _setup_daemon_windows
+    setup_daemon = _setup_daemon_for_windows
 else:
-    setup_daemon = _setup_daemon_unix
+    setup_daemon = _setup_daemon_for_unix_like
 
 
 # Note that we're sorting in a very particular order:
@@ -226,10 +230,9 @@ def main(method_name, syslog):
     try:
         line = stdin.readline(128)
         if not line:
-            # parent probably exited
+            return  # parent probably exited
-            return
+    except ConnectionResetError as e:
-    except IOError as e:
+        # On windows, ConnectionResetError is thrown when parent process closes it's socket pair end
-        # On windows, this ConnectionResetError is thrown when parent process closes it's socket pair end
         debug3('read from stdin failed: %s' % (e,))
         return
 
@@ -343,13 +346,13 @@ def main(method_name, syslog):
         except NotImplementedError:
             pass
 
-        if sys.platform != 'win32':
+        if sys.platform == 'linux':
             flush_systemd_dns_cache()
 
         try:
             stdout.write('STARTED\n')
             stdout.flush()
-        except IOError as e:
+        except IOError as e:  # the parent process probably died
             debug3('write to stdout failed: %s' % (e,))
             return
 
@@ -410,7 +413,7 @@ def main(method_name, syslog):
             except Exception:
                 debug2('An error occurred, ignoring it.')
 
-        if sys.platform != 'win32':
+        if sys.platform == 'linux':
             try:
                 flush_systemd_dns_cache()
             except Exception:

sshuttle/helpers.py

@@ -3,6 +3,10 @@ import socket
 import errno
 import os
 
+
+if sys.platform != "win32":
+    import fcntl
+
 logprefix = ''
 verbose = 0
 
@@ -14,10 +18,10 @@ def b(s):
 def log(s):
     global logprefix
     try:
-        try:
+        sys.stdout.flush()
-            sys.stdout.flush()
+    except IOError:
-        except (IOError, ValueError):
+        pass
-            pass
+    try:
         # Put newline at end of string if line doesn't have one.
         if not s.endswith("\n"):
             s = s+"\n"
@@ -234,4 +238,19 @@ def is_admin_user():
         except Exception:
             return False
 
+    # TODO(nom3ad): for sys.platform == 'linux', support capabilities check for non-root users. (CAP_NET_ADMIN might be enough?)
     return os.getuid() == 0
+
+
+def set_non_blocking_io(fd):
+    if sys.platform != "win32":
+        try:
+            os.set_blocking(fd, False)
+        except AttributeError:
+            # python < 3.5
+            flags = fcntl.fcntl(fd, fcntl.F_GETFL)
+            flags |= os.O_NONBLOCK
+            fcntl.fcntl(fd, fcntl.F_SETFL, flags)
+    else:
+        _sock = socket.fromfd(fd, socket.AF_INET, socket.SOCK_STREAM)
+        _sock.setblocking(False)

sshuttle/methods/__init__.py

@@ -46,6 +46,7 @@ class BaseMethod(object):
     @staticmethod
     def get_supported_features():
         result = Features()
+        result.loopback_port = True
         result.ipv4 = True
         result.ipv6 = False
         result.udp = False

sshuttle/methods/windivert.py

@@ -350,6 +350,7 @@ class Method(BaseMethod):
 
     def get_supported_features(self):
         result = super(Method, self).get_supported_features()
+        result.loopback_port = False
         result.user = False
         result.dns = False
         result.ipv6 = False
@@ -444,7 +445,7 @@ class Method(BaseMethod):
         if not ip_filters:
             raise Fatal("At least ipv4 or ipv6 address is expected")
         filter = f"{direction} and {proto.filter} and ({' or '.join(ip_filters)}) and tcp.SrcPort=={self.proxy_port}"
-        debug2(f"[INGRESS] {filter=}")
+        debug1(f"[INGRESS] {filter=}")
         with pydivert.WinDivert(filter) as w:
             ready_cb()
             for pkt in w:

sshuttle/options.py

@@ -235,9 +235,14 @@ parser.add_argument(
     """
 )
 
+if sys.platform == 'win32':
+    method_choices = ["auto", "windivert"]
+else:
+    method_choices = ["auto", "nat", "tproxy", "pf", "ipfw"]
+
 parser.add_argument(
     "--method",
-    choices=["auto", "nat", "nft", "tproxy", "pf", "ipfw"] if sys.platform != 'win32' else ["auto", "windivert"],
+    choices=method_choices,
     metavar="TYPE",
     default="auto",
     help="""

sshuttle/server.py

@@ -281,7 +281,7 @@ def main(latency_control, latency_buffer_size, auto_hosts, to_nameserver,
         sys.stdout.flush()
 
         handlers = []
-        mux = Mux(sys.stdin, sys.stdout)
+        mux = Mux(sys.stdin.buffer, sys.stdout.buffer)
         handlers.append(mux)
 
         debug1('auto-nets:' + str(auto_nets))

sshuttle/ssh.py

@@ -115,8 +115,8 @@ def connect(ssh_cmd, rhostport, python, stderr, add_cmd_delimiter, options):
     pyscript = r"""
                 import sys, os;
                 verbosity=%d;
-                sys.stdin = os.fdopen(0, "rb");
+                stdin = os.fdopen(0, "rb");
-                exec(compile(sys.stdin.read(%d), "assembler.py", "exec"));
+                exec(compile(stdin.read(%d), "assembler.py", "exec"));
                 sys.exit(98);
                 """ % (helpers.verbose or 0, len(content))
     pyscript = re.sub(r'\s+', ' ', pyscript.strip())
@@ -213,24 +213,26 @@ def connect(ssh_cmd, rhostport, python, stderr, add_cmd_delimiter, options):
             s2.close()
         s1.close()
 
-        def get_serversock():
+        def get_server_io():
             os.close(pstdin)
             os.close(pstdout)
-            return s2
+            return s2.makefile("rb", buffering=0), s2.makefile("wb", buffering=0)
     else:
-        # In Windows python implementation it seems not possible to use sockets as subprocess stdio
+        # In Windows CPython, we can't use BSD sockets as subprocess stdio
-        # Also select.select() won't work on pipes.
+        # and select.select() used in ssnet.py won't work on Windows pipes.
-        # So we have to use both socketpair and pipes together along with reader/writer threads to
+        # So we have to use both socketpair (for select.select) and pipes (for subprocess.Popen) together
-        # stream data between them
+        # along with reader/writer threads to stream data between them
-        # NOTE: Their can be a way to use sockets as stdio with some hacks.
+        # NOTE: Their could be a better way. Need to investigate further on this.
+        #   Either to use sockets as stdio for subprocess. Or to use pipes but with a select() alternative
         #   https://stackoverflow.com/questions/4993119/redirect-io-of-process-to-windows-socket
+
         (s1, s2) = socket.socketpair()
         pstdin = ssubprocess.PIPE
         pstdout = ssubprocess.PIPE
 
         preexec_fn = None
 
-        def get_serversock():
+        def get_server_io():
             import threading
 
             def stream_stdout_to_sock():
@@ -267,7 +269,7 @@ def connect(ssh_cmd, rhostport, python, stderr, add_cmd_delimiter, options):
     p = ssubprocess.Popen(argv, stdin=pstdin, stdout=pstdout, preexec_fn=preexec_fn,
                           close_fds=close_fds, stderr=stderr, bufsize=0)
 
-    serversock = get_serversock()
+    rfile, wfile = get_server_io()
-    serversock.sendall(content)
+    wfile.write(content)
-    serversock.sendall(content2)
+    wfile.write(content2)
-    return p, serversock
+    return p, rfile, wfile

sshuttle/ssnet.py

@@ -5,10 +5,7 @@ import errno
 import select
 import os
 
-if sys.platform != "win32":
+from sshuttle.helpers import b, log, debug1, debug2, debug3, Fatal, set_non_blocking_io
-    import fcntl
-
-from sshuttle.helpers import b, log, debug1, debug2, debug3, Fatal
 
 MAX_CHANNEL = 65535
 LATENCY_BUFFER_SIZE = 32768
@@ -215,10 +212,7 @@ class SockWrapper:
             return 0  # still connecting
         self.wsock.setblocking(False)
         try:
-            if sys.platform == 'win32':
+            return _nb_clean(self.wsock.send, buf)
-                return _nb_clean(self.wsock.send, buf)
-            else:
-                return _nb_clean(os.write, self.wsock.fileno(), buf)
         except OSError:
             _, e = sys.exc_info()[:2]
             if e.errno == errno.EPIPE:
@@ -241,10 +235,7 @@ class SockWrapper:
             return
         self.rsock.setblocking(False)
         try:
-            if sys.platform == 'win32':
+            return _nb_clean(self.rsock.recv, 65536)
-                return _nb_clean(self.rsock.recv, 65536)
-            else:
-                return _nb_clean(os.read, self.rsock.fileno(), 65536)
         except OSError:
             _, e = sys.exc_info()[:2]
             self.seterr('uread: %s' % e)
@@ -439,22 +430,9 @@ class Mux(Handler):
                 callback(cmd, data)
 
     def flush(self):
-        if sys.platform != "win32":
+        set_non_blocking_io(self.wfile.fileno())
-            try:
-                os.set_blocking(self.wfile.fileno(), False)
-            except AttributeError:
-                # python < 3.5
-                flags = fcntl.fcntl(self.wfile.fileno(), fcntl.F_GETFL)
-                flags |= os.O_NONBLOCK
-                fcntl.fcntl(self.wfile.fileno(), fcntl.F_SETFL, flags)
-        else:
-            self.wfile.raw._sock.setblocking(False)
-
         if self.outbuf and self.outbuf[0]:
-            if sys.platform == 'win32':
+            wrote = _nb_clean(os.write, self.wfile.fileno(), self.outbuf[0])
-                wrote = _nb_clean(self.wfile.raw._sock.send, self.outbuf[0])
-            else:
-                wrote = _nb_clean(os.write, self.wfile.fileno(), self.outbuf[0])
             debug2('mux wrote: %r/%d' % (wrote, len(self.outbuf[0])))
             if wrote:
                 self.outbuf[0] = self.outbuf[0][wrote:]
@@ -462,24 +440,11 @@ class Mux(Handler):
             self.outbuf[0:1] = []
 
     def fill(self):
-        if sys.platform != "win32":
+        set_non_blocking_io(self.rfile.fileno())
-            try:
-                os.set_blocking(self.rfile.fileno(), False)
-            except AttributeError:
-                # python < 3.5
-                flags = fcntl.fcntl(self.rfile.fileno(), fcntl.F_GETFL)
-                flags |= os.O_NONBLOCK
-                fcntl.fcntl(self.rfile.fileno(), fcntl.F_SETFL, flags)
-        else:
-            self.rfile.raw._sock.setblocking(False)
-
         try:
             # If LATENCY_BUFFER_SIZE is inappropriately large, we will
             # get a MemoryError here. Read no more than 1MiB.
-            if sys.platform == 'win32':
+            read = _nb_clean(self.rfile.read, min(1048576, LATENCY_BUFFER_SIZE))
-                read = _nb_clean(self.rfile.raw._sock.recv, min(1048576, LATENCY_BUFFER_SIZE))
-            else:
-                read = _nb_clean(os.read, self.rfile.fileno(), min(1048576, LATENCY_BUFFER_SIZE))
         except OSError:
             _, e = sys.exc_info()[:2]
             raise Fatal('other end: %r' % e)