experimental windows method

docs/requirements.rst

@@ -65,8 +65,7 @@ Requires:
 Windows
 ~~~~~~~
 
-Not officially supported, however can be made to work with Vagrant. Requires
-cmd.exe with Administrator access. See :doc:`windows` for more information.
+Experimental built-in support available. See :doc:`windows` for more information.
 
 
 Server side Requirements

docs/windows.rst

@@ -1,7 +1,13 @@
 Microsoft Windows
 =================
-Currently there is no built in support for running sshuttle directly on
-Microsoft Windows.
+
+Experimental support::
+
+Experimental built-in support for Windows is availble through `windivert` method.
+You have to install https://pypi.org/project/pydivert pacakge.  You need Administrator privileges to use windivert method
+
+
+Use Linux VM on Windows::
 
 What we can really do is to create a Linux VM with Vagrant (or simply
 Virtualbox if you like). In the Vagrant settings, remember to turn on bridged

sshuttle/__main__.py

@@ -1,4 +1,8 @@
 """Coverage.py's main entry point."""
 import sys
+import os
 from sshuttle.cmdline import main
-sys.exit(main())
+from sshuttle.helpers import debug3
+exit_code=main()
+debug3("Exiting process %r (pid:%s) with code %s" % (sys.argv, os.getpid(), exit_code,))
+sys.exit(exit_code)

sshuttle/client.py

@@ -14,7 +14,7 @@ import sshuttle.ssyslog as ssyslog
 import sshuttle.sdnotify as sdnotify
 from sshuttle.ssnet import SockWrapper, Handler, Proxy, Mux, MuxWrapper
 from sshuttle.helpers import log, debug1, debug2, debug3, Fatal, islocal, \
-    resolvconf_nameservers, which
+    resolvconf_nameservers, which, is_admin_user
 from sshuttle.methods import get_method, Features
 from sshuttle import __version__
 try:
@@ -219,67 +219,97 @@ class FirewallClient:
         # A list of commands that we can try to run to start the firewall.
         argv_tries = []
 
-        if os.getuid() == 0:  # No need to elevate privileges
+        if is_admin_user():  # No need to elevate privileges
             argv_tries.append(argvbase)
         else:
-            # Linux typically uses sudo; OpenBSD uses doas. However, some
-            # Linux distributions are starting to use doas.
-            sudo_cmd = ['sudo', '-p', '[local sudo] Password: ']
-            doas_cmd = ['doas']
-
-            # For clarity, try to replace executable name with the
-            # full path.
-            doas_path = which("doas")
-            if doas_path:
-                doas_cmd[0] = doas_path
-            sudo_path = which("sudo")
-            if sudo_path:
-                sudo_cmd[0] = sudo_path
-
-            # sudo_pythonpath indicates if we should set the
-            # PYTHONPATH environment variable when elevating
-            # privileges. This can be adjusted with the
-            # --no-sudo-pythonpath option.
-            if sudo_pythonpath:
-                pp_prefix = ['/usr/bin/env',
-                             'PYTHONPATH=%s' %
-                             os.path.dirname(os.path.dirname(__file__))]
-                sudo_cmd = sudo_cmd + pp_prefix
-                doas_cmd = doas_cmd + pp_prefix
-
-            # Final order should be: sudo/doas command, env
-            # pythonpath, and then argvbase (sshuttle command).
-            sudo_cmd = sudo_cmd + argvbase
-            doas_cmd = doas_cmd + argvbase
-
-            # If we can find doas and not sudo or if we are on
-            # OpenBSD, try using doas first.
-            if (doas_path and not sudo_path) or \
-               platform.platform().startswith('OpenBSD'):
-                argv_tries = [doas_cmd, sudo_cmd, argvbase]
+            if sys.platform == 'win32':
+                argv_tries.append(argvbase)
+                # runas_path = which("runas")
+                # if runas_path:
+                    # argv_tries.append(['runas' , '/noprofile', '/user:Administrator',  'python'])
             else:
-                argv_tries = [sudo_cmd, doas_cmd, argvbase]
+                # Linux typically uses sudo; OpenBSD uses doas. However, some
+                # Linux distributions are starting to use doas.
+                sudo_cmd = ['sudo', '-p', '[local sudo] Password: ']
+                doas_cmd = ['doas']
+
+                # For clarity, try to replace executable name with the
+                # full path.
+                doas_path = which("doas")
+                if doas_path:
+                    doas_cmd[0] = doas_path
+                sudo_path = which("sudo")
+                if sudo_path:
+                    sudo_cmd[0] = sudo_path
+
+                # sudo_pythonpath indicates if we should set the
+                # PYTHONPATH environment variable when elevating
+                # privileges. This can be adjusted with the
+                # --no-sudo-pythonpath option.
+                if sudo_pythonpath:
+                    pp_prefix = ['/usr/bin/env',
+                                'PYTHONPATH=%s' %
+                                os.path.dirname(os.path.dirname(__file__))]
+                    sudo_cmd = sudo_cmd + pp_prefix
+                    doas_cmd = doas_cmd + pp_prefix
+
+                # Final order should be: sudo/doas command, env
+                # pythonpath, and then argvbase (sshuttle command).
+                sudo_cmd = sudo_cmd + argvbase
+                doas_cmd = doas_cmd + argvbase
+
+                # If we can find doas and not sudo or if we are on
+                # OpenBSD, try using doas first.
+                if (doas_path and not sudo_path) or \
+                platform.platform().startswith('OpenBSD'):
+                    argv_tries = [doas_cmd, sudo_cmd, argvbase]
+                else:
+                    argv_tries = [sudo_cmd, doas_cmd, argvbase]
 
         # Try all commands in argv_tries in order. If a command
         # produces an error, try the next one. If command is
         # successful, set 'success' variable and break.
         success = False
         for argv in argv_tries:
-            # we can't use stdin/stdout=subprocess.PIPE here, as we
-            # normally would, because stupid Linux 'su' requires that
-            # stdin be attached to a tty. Instead, attach a
-            # *bidirectional* socket to its stdout, and use that for
-            # talking in both directions.
-            (s1, s2) = socket.socketpair()
 
-            def setup():
-                # run in the child process
-                s2.close()
+            if sys.platform != 'win32':
+                # we can't use stdin/stdout=subprocess.PIPE here, as we
+                # normally would, because stupid Linux 'su' requires that
+                # stdin be attached to a tty. Instead, attach a
+                # *bidirectional* socket to its stdout, and use that for
+                # talking in both directions.
+                (s1, s2) = socket.socketpair()
+                pstdout = s1
+                pstdin = s1
+                penv = None
+                def preexec_fn():
+                    # run in the child process
+                    s2.close()
+                def get_pfile():
+                    s1.close()
+                    return s2.makefile('rwb')
+
+            else:
+                (s1, s2) = socket.socketpair()
+                pstdout = None
+                pstdin = ssubprocess.PIPE
+                preexec_fn = None
+                penv = os.environ.copy()
+                penv['PYTHONPATH'] =  os.path.dirname(os.path.dirname(__file__))
+                def get_pfile():
+                    import base64
+                    socket_share_data = s1.share(self.p.pid)
+                    s1.close()
+                    socket_share_data_b64 = base64.b64encode(socket_share_data)
+                    # debug3(f"{socket_share_data_b64=}")
+                    self.p.stdin.write(socket_share_data_b64 + b'\n')
+                    self.p.stdin.flush()
+                    return s2.makefile('rwb')
 
             try:
                 debug1("Starting firewall manager with command: %r" % argv)
-                self.p = ssubprocess.Popen(argv, stdout=s1, stdin=s1,
-                                           preexec_fn=setup)
+                self.p = ssubprocess.Popen(argv, stdout=pstdout, stdin=pstdin, env=penv,
+                                           preexec_fn=preexec_fn)
                 # No env: Talking to `FirewallClient.start`, which has no i18n.
             except OSError as e:
                 # This exception will occur if the program isn't
@@ -287,11 +317,15 @@ class FirewallClient:
                 debug1('Unable to start firewall manager. Popen failed. '
                        'Command=%r Exception=%s' % (argv, e))
                 continue
-
             self.argv = argv
-            s1.close()
-            self.pfile = s2.makefile('rwb')
-            line = self.pfile.readline()
+            
+            self.pfile = get_pfile()
+
+            try:
+                line = self.pfile.readline()
+            except ConnectionResetError:
+                # happens in Windows, when subprocess exists
+                line=''
 
             rv = self.p.poll()   # Check if process is still running
             if rv:
@@ -327,14 +361,14 @@ class FirewallClient:
                        'Command=%r' % (skipped_text, self.argv))
                 continue
 
-            method_name = line[6:-1]
+            method_name = line.strip()[6:]
             self.method = get_method(method_name.decode("ASCII"))
             self.method.set_firewall(self)
             success = True
             break
 
         if not success:
-            raise Fatal("All attempts to elevate privileges failed.")
+            raise Fatal("All attempts to run firewall client with elevated privileges were failed.")
 
     def setup(self, subnets_include, subnets_exclude, nslist,
               redirectport_v6, redirectport_v4, dnsport_v6, dnsport_v4, udp,
@@ -397,9 +431,9 @@ class FirewallClient:
                          (udp, user, group, bytes(self.tmark, 'ascii'), os.getpid()))
         self.pfile.flush()
 
-        line = self.pfile.readline()
+        line = self.pfile.readline().strip()
         self.check()
-        if line != b'STARTED\n':
+        if line != b'STARTED':
             raise Fatal('%r expected STARTED, got %r' % (self.argv, line))
 
     def sethostip(self, hostname, ip):
@@ -562,24 +596,26 @@ def _main(tcp_listener, udp_listener, fw, ssh_cmd, remotename,
                          auto_nets=auto_nets))
     except socket.error as e:
         if e.args[0] == errno.EPIPE:
+            debug3('Error: EPIPE: ' + repr(e))
             raise Fatal("failed to establish ssh session (1)")
         else:
             raise
-    mux = Mux(serversock.makefile("rb"), serversock.makefile("wb"))
+    rfile, wfile = serversock.makefile("rb"), serversock.makefile("wb")
+    mux = Mux(rfile, wfile)
     handlers.append(mux)
 
     expected = b'SSHUTTLE0001'
-
     try:
         v = 'x'
         while v and v != b'\0':
-            v = serversock.recv(1)
+            v = rfile.read(1)
         v = 'x'
         while v and v != b'\0':
-            v = serversock.recv(1)
-        initstring = serversock.recv(len(expected))
+            v = rfile.read(1)
+        initstring = rfile.read(len(expected))
     except socket.error as e:
         if e.args[0] == errno.ECONNRESET:
+            debug3('Error: ECONNRESET ' + repr(e))
             raise Fatal("failed to establish ssh session (2)")
         else:
             raise

sshuttle/firewall.py

@@ -7,10 +7,12 @@ import os
 import platform
 import traceback
 import subprocess as ssubprocess
+import base64
+import io
 
 import sshuttle.ssyslog as ssyslog
 import sshuttle.helpers as helpers
-from sshuttle.helpers import log, debug1, debug2, Fatal
+from sshuttle.helpers import is_admin_user, log, debug1, debug2, Fatal
 from sshuttle.methods import get_auto_method, get_method
 
 HOSTSFILE = '/etc/hosts'
@@ -87,8 +89,8 @@ def firewall_exit(signum, frame):
 
 
 # Isolate function that needs to be replaced for tests
-def setup_daemon():
-    if os.getuid() != 0:
+def _setup_daemon_unix():
+    if not is_admin_user():
         raise Fatal('You must be root (or enable su/sudo) to set the firewall')
 
     # don't disappear if our controlling terminal or stdout/stderr
@@ -113,6 +115,25 @@ def setup_daemon():
     return sys.stdin, sys.stdout
 
 
+def _setup_daemon_windows():
+    if not is_admin_user():
+        raise Fatal('You must be administrator to set the firewall')
+
+    signal.signal(signal.SIGTERM, firewall_exit)
+    signal.signal(signal.SIGINT, firewall_exit)
+    socket_share_data_b64 = sys.stdin.readline()
+    # debug3(f'FROM_SHARE ${socket_share_data_b64=}')
+    socket_share_data = base64.b64decode(socket_share_data_b64)
+    sock = socket.fromshare(socket_share_data)
+    sys.stdin = io.TextIOWrapper(sock.makefile('rb'))
+    sys.stdout = io.TextIOWrapper(sock.makefile('wb'))
+    return sys.stdin, sys.stdout
+
+if sys.platform == 'win32':
+    setup_daemon = _setup_daemon_windows
+else:
+    setup_daemon = _setup_daemon_unix
+
 # Note that we're sorting in a very particular order:
 # we need to go from smaller, more specific, port ranges, to larger,
 # less-specific, port ranges. At each level, we order by subnet
@@ -190,9 +211,13 @@ def main(method_name, syslog):
     # we wait until we get some input before creating the rules.  That way,
     # sshuttle can launch us as early as possible (and get sudo password
     # authentication as early in the startup process as possible).
-    line = stdin.readline(128)
-    if not line:
-        return  # parent died; nothing to do
+    try:
+        line = stdin.readline(128)
+        if not line:
+            return  # parent died; nothing to do
+    except ConnectionResetError:   
+        # On windows, this is thrown when parent process closes it's socket pair end
+        return
 
     subnets = []
     if line != 'ROUTES\n':

sshuttle/helpers.py

@@ -220,3 +220,14 @@ def which(file, mode=os.F_OK | os.X_OK):
     else:
         debug2("which() could not find '%s' in %s" % (file, path))
     return rv
+
+def is_admin_user():
+    if sys.platform == 'win32':
+        import ctypes
+        # https://stackoverflow.com/questions/130763/request-uac-elevation-from-within-a-python-script/41930586#41930586
+        try:
+            return ctypes.windll.shell32.IsUserAnAdmin()
+        except:
+            return False
+
+    return os.getuid() != 0

sshuttle/methods/__init__.py

@@ -1,6 +1,7 @@
 import importlib
 import socket
 import struct
+import sys
 import errno
 import ipaddress
 from sshuttle.helpers import Fatal, debug3
@@ -109,7 +110,7 @@ def get_method(method_name):
 def get_auto_method():
     debug3("Selecting a method automatically...")
     # Try these methods, in order:
-    methods_to_try = ["nat", "nft", "pf", "ipfw"]
+    methods_to_try = ["nat", "nft", "pf", "ipfw"] if sys.platform != "win32" else ["windivert"]
     for m in methods_to_try:
         method = get_method(m)
         if method.is_supported():

sshuttle/methods/windivert.py

@@ -0,0 +1,126 @@
+import sys
+import ipaddress
+import threading
+from collections import namedtuple
+
+
+try:
+    import pydivert
+except ImportError:
+    raise Fatal('Could not import pydivert module. windivert requires https://pypi.org/project/pydivert')
+
+from sshuttle.methods import BaseMethod
+from sshuttle.helpers import log, debug1, debug2, Fatal
+
+# https://reqrypt.org/windivert-doc.html#divert_iphdr
+
+
+ConnectionTuple = namedtuple(
+    "ConnectionTuple", ["protocol", "src_addr", "src_port", "dst_addr", "dst_port"]
+)
+
+class ConnectionTracker:
+    def __init__(self) -> None:
+        self.d = {}
+
+    def add_tcp(self, src_addr, src_port, dst_addr, dst_port):
+        k = ("TCP", src_addr, src_port)
+        v = (dst_addr, dst_port)
+        if self.d.get(k) != v:
+            debug1("Adding tcp connection to tracker:" + repr((src_addr, src_port, dst_addr, dst_port)))
+            self.d[k] = v
+
+    def get_tcp(self, src_addr, src_port):
+        try:
+            return ConnectionTuple(
+                "TCP", src_addr, src_port, *self.d[("TCP", src_addr, src_port)]
+            )
+        except KeyError:
+            return None
+
+
+class Method(BaseMethod):
+
+    def setup_firewall(self, port, dnsport, nslist, family, subnets, udp,
+                       user, tmark):
+        log( f"{port=}, {dnsport=}, {nslist=}, {family=}, {subnets=}, {udp=}, {user=}, {tmark=}")
+        #  port=12300, dnsport=0, nslist=[], family=<AddressFamily.AF_INET: 2>, 
+        # subnets=[(2, 24, False, '10.111.10.0', 0, 0), (2, 16, False, '169.254.0.0', 0, 0), (2, 24, False, '172.31.0.0', 0, 0), (2, 16, False, '192.168.0.0', 0, 0), (2, 32, True, '0.0.0.0', 0, 0)],
+        #  udp=False, user=None, tmark='0x01' 
+        self.conntrack = ConnectionTracker()
+        proxy_addr = "10.0.2.15"
+
+        subnet_addreses = []
+        for (_, mask, exclude, network_addr, fport, lport) in subnets:
+            if exclude:
+                continue
+            assert fport == 0, 'custom port range not supported'
+            assert lport == 0,  'custom port range not supported'
+            subnet_addreses.append("%s/%s" % (network_addr, mask))
+
+        debug2("subnet_addreses=%s proxy_addr=%s:%s" % (subnet_addreses,proxy_addr,port))
+
+        # check permission
+        with pydivert.WinDivert('false'):
+            pass
+
+        threading.Thread(name='outbound_divert', target=self._outbound_divert, args=(subnet_addreses, proxy_addr, port),  daemon=True).start()
+        threading.Thread(name='inbound_divert', target=self._inbound_divert, args=(proxy_addr, port),  daemon=True).start()
+
+    def restore_firewall(self, port, family, udp, user):
+        pass
+
+    def get_supported_features(self):
+        result = super(Method, self).get_supported_features()
+        result.user = False
+        result.dns = False
+        result.ipv6 = False
+        return result
+
+    def get_tcp_dstip(self, sock):
+        return ('172.31.0.141', 80)
+
+    def is_supported(self):
+        if sys.platform == 'win32':
+            return True
+        return False
+
+
+
+    def _outbound_divert(self, subnets, proxy_addr, proxy_port):
+        # with pydivert.WinDivert(f"outbound and tcp and ip.DstAddr == {subnet}") as w:
+        filter = "outbound and ip and tcp"
+        subnet_selectors = []
+        for cidr in subnets:
+            ip_network = ipaddress.ip_network(cidr)
+            first_ip = ip_network.network_address
+            last_ip = ip_network.broadcast_address
+            subnet_selectors.append(f"(ip.DstAddr >= {first_ip} and ip.DstAddr <= {last_ip})")
+        filter = f"{filter} and ({'or'.join(subnet_selectors)}) "
+
+        debug1(f"[OUTBOUND] {filter=}")
+        with pydivert.WinDivert(filter) as w:
+            for pkt in w:
+                # debug3(repr(pkt))
+                self.conntrack.add_tcp(pkt.src_addr, pkt.src_port, pkt.dst_addr, pkt.dst_port)
+                pkt.ipv4.dst_addr = proxy_addr
+                pkt.tcp.dst_port = proxy_port
+                w.send(pkt, recalculate_checksum=True)
+
+
+    def _inbound_divert(self, proxy_addr, proxy_port):
+        filter = f"inbound and ip and tcp and ip.SrcAddr == {proxy_addr} and tcp.SrcPort == {proxy_port}"
+        debug2(f"[INBOUND] {filter=}")
+        with pydivert.WinDivert(filter) as w:
+            for pkt in w:
+                # debug2(repr(conntrack.d))
+                # debug2(repr((pkt.src_addr, pkt.src_port, pkt.dst_addr, pkt.dst_port)))
+                conn = self.conntrack.get_tcp(pkt.dst_addr, pkt.dst_port)
+                if not conn:
+                    debug2("Unexpcted packet:" + repr((pkt.protocol,pkt.src_addr,pkt.src_port,pkt.dst_addr,pkt.dst_port)))
+                    continue
+                pkt.ipv4.src_addr = conn.dst_addr
+                pkt.tcp.src_port = conn.dst_port
+                w.send(pkt, recalculate_checksum=True)
+
+

sshuttle/options.py

@@ -1,5 +1,6 @@
 import re
 import socket
+import sys
 from argparse import ArgumentParser, Action, ArgumentTypeError as Fatal
 
 from sshuttle import __version__
@@ -236,7 +237,7 @@ parser.add_argument(
 
 parser.add_argument(
     "--method",
-    choices=["auto", "nat", "nft", "tproxy", "pf", "ipfw"],
+    choices=["auto", "nat", "nft", "tproxy", "pf", "ipfw"] if sys.platform != 'win32' else ["auto", "windivert"],
     metavar="TYPE",
     default="auto",
     help="""

sshuttle/ssh.py

@@ -175,9 +175,10 @@ def connect(ssh_cmd, rhostport, python, stderr, add_cmd_delimiter, options):
             # case, sshuttle might not work at all since it is not
             # possible to run python on the remote machine---even if
             # it is present.
+            devnull='/dev/null'
             pycmd = ("P=python3; $P -V 2>%s || P=python; "
                      "exec \"$P\" -c %s; exit 97") % \
-                (os.devnull, quote(pyscript))
+                (devnull, quote(pyscript))
             pycmd = ("/bin/sh -c {}".format(quote(pycmd)))
 
         if password is not None:
@@ -203,19 +204,56 @@ def connect(ssh_cmd, rhostport, python, stderr, add_cmd_delimiter, options):
         raise Fatal("Failed to find '%s' in path %s" % (argv[0], get_path()))
     argv[0] = abs_path
 
-    (s1, s2) = socket.socketpair()
 
-    def setup():
-        # runs in the child process
-        s2.close()
-    s1a, s1b = os.dup(s1.fileno()), os.dup(s1.fileno())
-    s1.close()
+    if sys.platform != 'win32':
+        (s1, s2) = socket.socketpair()
+        def preexec_fn():
+            # runs in the child process
+            s2.close()
+        pstdin, pstdout = os.dup(s1.fileno()), os.dup(s1.fileno())
+        s1.close()
 
-    debug2('executing: %r' % argv)
-    p = ssubprocess.Popen(argv, stdin=s1a, stdout=s1b, preexec_fn=setup,
-                          close_fds=True, stderr=stderr)
-    os.close(s1a)
-    os.close(s1b)
-    s2.sendall(content)
-    s2.sendall(content2)
-    return p, s2
+        def get_serversock():
+            os.close(pstdin)
+            os.close(pstdout)
+            return s2
+    else:
+        (s1, s2) = socket.socketpair()
+        preexec_fn = None
+        pstdin = ssubprocess.PIPE
+        pstdout = ssubprocess.PIPE
+        def get_serversock():
+            import threading
+            def steam_stdout_to_sock():
+                while True:
+                    data = p.stdout.read(1)
+                    if not data:
+                        debug2("EOF on ssh process stdout. Process probably exited")
+                        break
+                    n = s1.sendall(data)
+                    print("<<<<< p.stdout.read()", len(data), '->', n, data[:min(32,len(data))])
+            def stream_sock_to_stdin():
+                while True:
+                    data = s1.recv(16384)
+                    if not data:
+                        print(">>>>>> EOF stream_sock_to_stdin")
+                        break
+                    n = p.stdin.write(data)
+                    print(">>>>>> s1.recv()", len(data) , "->" , n , data[:min(32,len(data))])
+                    p.communicate
+            threading.Thread(target=steam_stdout_to_sock,  name='steam_stdout_to_sock', daemon=True).start()
+            threading.Thread(target=stream_sock_to_stdin, name='stream_sock_to_stdin',  daemon=True).start()
+            # s2.setblocking(False)
+            return s2
+
+    # https://stackoverflow.com/questions/48671215/howto-workaround-of-close-fds-true-and-redirect-stdout-stderr-on-windows
+    close_fds = False if sys.platform == 'win32' else True
+
+    debug2("executing: %r" % argv)
+    p = ssubprocess.Popen(argv, stdin=pstdin, stdout=pstdout, preexec_fn=preexec_fn,
+                        close_fds=close_fds, stderr=stderr, bufsize=0)
+
+    serversock = get_serversock()
+    serversock.sendall(content)
+    serversock.sendall(content2)
+    return p, serversock

sshuttle/ssnet.py

@@ -4,7 +4,9 @@ import socket
 import errno
 import select
 import os
-import fcntl
+
+if sys.platform != "win32":
+    import fcntl
 
 from sshuttle.helpers import b, log, debug1, debug2, debug3, Fatal
 
@@ -213,7 +215,10 @@ class SockWrapper:
             return 0  # still connecting
         self.wsock.setblocking(False)
         try:
-            return _nb_clean(os.write, self.wsock.fileno(), buf)
+            if sys.platform == 'win32':
+                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:
@@ -236,7 +241,10 @@ class SockWrapper:
             return
         self.rsock.setblocking(False)
         try:
-            return _nb_clean(os.read, self.rsock.fileno(), 65536)
+            if sys.platform == 'win32':
+                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)
@@ -431,15 +439,22 @@ class Mux(Handler):
                 callback(cmd, data)
 
     def flush(self):
-        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)
+        if sys.platform != "win32":
+            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]:
-            wrote = _nb_clean(os.write, self.wfile.fileno(), self.outbuf[0])
+            if sys.platform == 'win32':
+                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:]
@@ -447,18 +462,26 @@ class Mux(Handler):
             self.outbuf[0:1] = []
 
     def fill(self):
-        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)
+        if sys.platform != "win32":
+            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.
-            read = _nb_clean(os.read, self.rfile.fileno(),
-                             min(1048576, LATENCY_BUFFER_SIZE))
+            if sys.platform == 'win32':
+                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)