Merge pull request #1079 from docker/1024-npipe-support

npipe support

Author: shin-

docker/client.py

@@ -14,7 +14,6 @@
 
 import json
 import struct
-import sys
 
 import requests
 import requests.exceptions
@@ -26,10 +25,14 @@
 from . import constants
 from . import errors
 from .auth import auth
-from .unixconn import unixconn
 from .ssladapter import ssladapter
-from .utils import utils, check_resource, update_headers, kwargs_from_env
 from .tls import TLSConfig
+from .transport import UnixAdapter
+from .utils import utils, check_resource, update_headers, kwargs_from_env
+try:
+    from .transport import NpipeAdapter
+except ImportError:
+    pass
 
 
 def from_env(**kwargs):
@@ -59,11 +62,26 @@ def __init__(self, base_url=None, version=None,
 
         self._auth_configs = auth.load_config()
 
-        base_url = utils.parse_host(base_url, sys.platform, tls=bool(tls))
+        base_url = utils.parse_host(
+            base_url, constants.IS_WINDOWS_PLATFORM, tls=bool(tls)
+        )
         if base_url.startswith('http+unix://'):
-            self._custom_adapter = unixconn.UnixAdapter(base_url, timeout)
+            self._custom_adapter = UnixAdapter(base_url, timeout)
             self.mount('http+docker://', self._custom_adapter)
             self.base_url = 'http+docker://localunixsocket'
+        elif base_url.startswith('npipe://'):
+            if not constants.IS_WINDOWS_PLATFORM:
+                raise errors.DockerException(
+                    'The npipe:// protocol is only supported on Windows'
+                )
+            try:
+                self._custom_adapter = NpipeAdapter(base_url, timeout)
+            except NameError:
+                raise errors.DockerException(
+                    'Install pypiwin32 package to enable npipe:// support'
+                )
+            self.mount('http+docker://', self._custom_adapter)
+            self.base_url = 'http+docker://localnpipe'
         else:
             # Use SSLAdapter for the ability to specify SSL version
             if isinstance(tls, TLSConfig):

docker/constants.py

@@ -1,3 +1,5 @@
+import sys
+
 DEFAULT_DOCKER_API_VERSION = '1.22'
 DEFAULT_TIMEOUT_SECONDS = 60
 STREAM_HEADER_SIZE_BYTES = 8
@@ -8,3 +10,5 @@
 INSECURE_REGISTRY_DEPRECATION_WARNING = \
     'The `insecure_registry` argument to {} ' \
     'is deprecated and non-functional. Please remove it.'
+
+IS_WINDOWS_PLATFORM = (sys.platform == 'win32')

docker/transport/__init__.py

@@ -0,0 +1,6 @@
+# flake8: noqa
+from .unixconn import UnixAdapter
+try:
+    from .npipeconn import NpipeAdapter
+except ImportError:
+    pass

docker/transport/npipeconn.py

@@ -0,0 +1,80 @@
+import six
+import requests.adapters
+
+from .npipesocket import NpipeSocket
+
+if six.PY3:
+    import http.client as httplib
+else:
+    import httplib
+
+try:
+    import requests.packages.urllib3 as urllib3
+except ImportError:
+    import urllib3
+
+
+RecentlyUsedContainer = urllib3._collections.RecentlyUsedContainer
+
+
+class NpipeHTTPConnection(httplib.HTTPConnection, object):
+    def __init__(self, npipe_path, timeout=60):
+        super(NpipeHTTPConnection, self).__init__(
+            'localhost', timeout=timeout
+        )
+        self.npipe_path = npipe_path
+        self.timeout = timeout
+
+    def connect(self):
+        sock = NpipeSocket()
+        sock.settimeout(self.timeout)
+        sock.connect(self.npipe_path)
+        self.sock = sock
+
+
+class NpipeHTTPConnectionPool(urllib3.connectionpool.HTTPConnectionPool):
+    def __init__(self, npipe_path, timeout=60):
+        super(NpipeHTTPConnectionPool, self).__init__(
+            'localhost', timeout=timeout
+        )
+        self.npipe_path = npipe_path
+        self.timeout = timeout
+
+    def _new_conn(self):
+        return NpipeHTTPConnection(
+            self.npipe_path, self.timeout
+        )
+
+
+class NpipeAdapter(requests.adapters.HTTPAdapter):
+    def __init__(self, base_url, timeout=60):
+        self.npipe_path = base_url.replace('npipe://', '')
+        self.timeout = timeout
+        self.pools = RecentlyUsedContainer(
+            10, dispose_func=lambda p: p.close()
+        )
+        super(NpipeAdapter, self).__init__()
+
+    def get_connection(self, url, proxies=None):
+        with self.pools.lock:
+            pool = self.pools.get(url)
+            if pool:
+                return pool
+
+            pool = NpipeHTTPConnectionPool(
+                self.npipe_path, self.timeout
+            )
+            self.pools[url] = pool
+
+        return pool
+
+    def request_url(self, request, proxies):
+        # The select_proxy utility in requests errors out when the provided URL
+        # doesn't have a hostname, like is the case when using a UNIX socket.
+        # Since proxies are an irrelevant notion in the case of UNIX sockets
+        # anyway, we simply return the path URL directly.
+        # See also: https://github.com/docker/docker-py/issues/811
+        return request.path_url
+
+    def close(self):
+        self.pools.clear()

docker/transport/npipesocket.py

@@ -0,0 +1,191 @@
+import functools
+import io
+
+import win32file
+import win32pipe
+
+cSECURITY_SQOS_PRESENT = 0x100000
+cSECURITY_ANONYMOUS = 0
+cPIPE_READMODE_MESSAGE = 2
+
+
+def check_closed(f):
+    @functools.wraps(f)
+    def wrapped(self, *args, **kwargs):
+        if self._closed:
+            raise RuntimeError(
+                'Can not reuse socket after connection was closed.'
+            )
+        return f(self, *args, **kwargs)
+    return wrapped
+
+
+class NpipeSocket(object):
+    """ Partial implementation of the socket API over windows named pipes.
+        This implementation is only designed to be used as a client socket,
+        and server-specific methods (bind, listen, accept...) are not
+        implemented.
+    """
+    def __init__(self, handle=None):
+        self._timeout = win32pipe.NMPWAIT_USE_DEFAULT_WAIT
+        self._handle = handle
+        self._closed = False
+
+    def accept(self):
+        raise NotImplementedError()
+
+    def bind(self, address):
+        raise NotImplementedError()
+
+    def close(self):
+        self._handle.Close()
+        self._closed = True
+
+    @check_closed
+    def connect(self, address):
+        win32pipe.WaitNamedPipe(address, self._timeout)
+        handle = win32file.CreateFile(
+            address,
+            win32file.GENERIC_READ | win32file.GENERIC_WRITE,
+            0,
+            None,
+            win32file.OPEN_EXISTING,
+            cSECURITY_ANONYMOUS | cSECURITY_SQOS_PRESENT,
+            0
+        )
+        self.flags = win32pipe.GetNamedPipeInfo(handle)[0]
+
+        self._handle = handle
+        self._address = address
+
+    @check_closed
+    def connect_ex(self, address):
+        return self.connect(address)
+
+    @check_closed
+    def detach(self):
+        self._closed = True
+        return self._handle
+
+    @check_closed
+    def dup(self):
+        return NpipeSocket(self._handle)
+
+    @check_closed
+    def fileno(self):
+        return int(self._handle)
+
+    def getpeername(self):
+        return self._address
+
+    def getsockname(self):
+        return self._address
+
+    def getsockopt(self, level, optname, buflen=None):
+        raise NotImplementedError()
+
+    def ioctl(self, control, option):
+        raise NotImplementedError()
+
+    def listen(self, backlog):
+        raise NotImplementedError()
+
+    def makefile(self, mode=None, bufsize=None):
+        if mode.strip('b') != 'r':
+            raise NotImplementedError()
+        rawio = NpipeFileIOBase(self)
+        if bufsize is None:
+            bufsize = io.DEFAULT_BUFFER_SIZE
+        return io.BufferedReader(rawio, buffer_size=bufsize)
+
+    @check_closed
+    def recv(self, bufsize, flags=0):
+        err, data = win32file.ReadFile(self._handle, bufsize)
+        return data
+
+    @check_closed
+    def recvfrom(self, bufsize, flags=0):
+        data = self.recv(bufsize, flags)
+        return (data, self._address)
+
+    @check_closed
+    def recvfrom_into(self, buf, nbytes=0, flags=0):
+        return self.recv_into(buf, nbytes, flags), self._address
+
+    @check_closed
+    def recv_into(self, buf, nbytes=0):
+        readbuf = buf
+        if not isinstance(buf, memoryview):
+            readbuf = memoryview(buf)
+
+        err, data = win32file.ReadFile(
+            self._handle,
+            readbuf[:nbytes] if nbytes else readbuf
+        )
+        return len(data)
+
+    @check_closed
+    def send(self, string, flags=0):
+        err, nbytes = win32file.WriteFile(self._handle, string)
+        return nbytes
+
+    @check_closed
+    def sendall(self, string, flags=0):
+        return self.send(string, flags)
+
+    @check_closed
+    def sendto(self, string, address):
+        self.connect(address)
+        return self.send(string)
+
+    def setblocking(self, flag):
+        if flag:
+            return self.settimeout(None)
+        return self.settimeout(0)
+
+    def settimeout(self, value):
+        if value is None:
+            self._timeout = win32pipe.NMPWAIT_NOWAIT
+        elif not isinstance(value, (float, int)) or value < 0:
+            raise ValueError('Timeout value out of range')
+        elif value == 0:
+            self._timeout = win32pipe.NMPWAIT_USE_DEFAULT_WAIT
+        else:
+            self._timeout = value
+
+    def gettimeout(self):
+        return self._timeout
+
+    def setsockopt(self, level, optname, value):
+        raise NotImplementedError()
+
+    @check_closed
+    def shutdown(self, how):
+        return self.close()
+
+
+class NpipeFileIOBase(io.RawIOBase):
+    def __init__(self, npipe_socket):
+        self.sock = npipe_socket
+
+    def close(self):
+        super(NpipeFileIOBase, self).close()
+        self.sock = None
+
+    def fileno(self):
+        return self.sock.fileno()
+
+    def isatty(self):
+        return False
+
+    def readable(self):
+        return True
+
+    def readinto(self, buf):
+        return self.sock.recv_into(buf)
+
+    def seekable(self):
+        return False
+
+    def writable(self):
+        return False

docker/unixconn/unixconn.py renamed to docker/transport/unixconn.py

@@ -383,13 +383,13 @@ def parse_repository_tag(repo_name):
 # fd:// protocol unsupported (for obvious reasons)
 # Added support for http and https
 # Protocol translation: tcp -> http, unix -> http+unix
-def parse_host(addr, platform=None, tls=False):
+def parse_host(addr, is_win32=False, tls=False):
     proto = "http+unix"
     host = DEFAULT_HTTP_HOST
     port = None
     path = ''
 
-    if not addr and platform == 'win32':
+    if not addr and is_win32:
         addr = '{0}:{1}'.format(DEFAULT_HTTP_HOST, 2375)
 
     if not addr or addr.strip() == 'unix://':
@@ -413,6 +413,9 @@ def parse_host(addr, platform=None, tls=False):
     elif addr.startswith('https://'):
         proto = "https"
         addr = addr[8:]
+    elif addr.startswith('npipe://'):
+        proto = 'npipe'
+        addr = addr[8:]
     elif addr.startswith('fd://'):
         raise errors.DockerException("fd protocol is not implemented")
     else:
@@ -448,7 +451,7 @@ def parse_host(addr, platform=None, tls=False):
     else:
         host = addr
 
-    if proto == "http+unix":
+    if proto == "http+unix" or proto == 'npipe':
         return "{0}://{1}".format(proto, host)
     return "{0}://{1}:{2}{3}".format(proto, host, port, path)