Merge pull request #2165 from docker/ssh_protocol_support

SSH protocol support

16 files changed, 290 insertions, 108 deletions

diff --git a/docker/api/client.py b/docker/api/client.py
index 91da1c8..197846d 100644
--- a/docker/api/client.py
+++ b/docker/api/client.py
@@ -39,6 +39,11 @@ try:
 except ImportError:
     pass
 
+try:
+    from ..transport import SSHAdapter
+except ImportError:
+    pass
+
 
 class APIClient(
         requests.Session,
@@ -141,6 +146,18 @@ class APIClient(
                 )
             self.mount('http+docker://', self._custom_adapter)
             self.base_url = 'http+docker://localnpipe'
+        elif base_url.startswith('ssh://'):
+            try:
+                self._custom_adapter = SSHAdapter(
+                    base_url, timeout, pool_connections=num_pools
+                )
+            except NameError:
+                raise DockerException(
+                    'Install paramiko package to enable ssh:// support'
+                )
+            self.mount('http+docker://ssh', self._custom_adapter)
+            self._unmount('http://', 'https://')
+            self.base_url = 'http+docker://ssh'
         else:
             # Use SSLAdapter for the ability to specify SSL version
             if isinstance(tls, TLSConfig):
@@ -279,6 +296,8 @@ class APIClient(
         self._raise_for_status(response)
         if self.base_url == "http+docker://localnpipe":
             sock = response.raw._fp.fp.raw.sock
+        elif self.base_url.startswith('http+docker://ssh'):
+            sock = response.raw._fp.fp.channel
         elif six.PY3:
             sock = response.raw._fp.fp.raw
             if self.base_url.startswith("https://"):
diff --git a/docker/transport/__init__.py b/docker/transport/__init__.py
index abbee18..d2cf2a7 100644
--- a/docker/transport/__init__.py
+++ b/docker/transport/__init__.py
@@ -6,3 +6,8 @@ try:
     from .npipesocket import NpipeSocket
 except ImportError:
     pass
+
+try:
+    from .sshconn import SSHAdapter
+except ImportError:
+    pass
diff --git a/docker/transport/npipesocket.py b/docker/transport/npipesocket.py
index c04b39d..ef02031 100644
--- a/docker/transport/npipesocket.py
+++ b/docker/transport/npipesocket.py
@@ -87,10 +87,6 @@ class NpipeSocket(object):
     def dup(self):
         return NpipeSocket(self._handle)
 
-    @check_closed
-    def fileno(self):
-        return int(self._handle)
-
     def getpeername(self):
         return self._address
 
diff --git a/docker/transport/sshconn.py b/docker/transport/sshconn.py
new file mode 100644
index 0000000..0f6bb51
--- /dev/null
+++ b/docker/transport/sshconn.py
@@ -0,0 +1,107 @@
+import paramiko
+import requests.adapters
+import six
+
+from .. import constants
+
+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 SSHConnection(httplib.HTTPConnection, object):
+    def __init__(self, ssh_transport, timeout=60):
+        super(SSHConnection, self).__init__(
+            'localhost', timeout=timeout
+        )
+        self.ssh_transport = ssh_transport
+        self.timeout = timeout
+
+    def connect(self):
+        sock = self.ssh_transport.open_session()
+        sock.settimeout(self.timeout)
+        sock.exec_command('docker system dial-stdio')
+        self.sock = sock
+
+
+class SSHConnectionPool(urllib3.connectionpool.HTTPConnectionPool):
+    scheme = 'ssh'
+
+    def __init__(self, ssh_client, timeout=60, maxsize=10):
+        super(SSHConnectionPool, self).__init__(
+            'localhost', timeout=timeout, maxsize=maxsize
+        )
+        self.ssh_transport = ssh_client.get_transport()
+        self.timeout = timeout
+
+    def _new_conn(self):
+        return SSHConnection(self.ssh_transport, self.timeout)
+
+    # When re-using connections, urllib3 calls fileno() on our
+    # SSH channel instance, quickly overloading our fd limit. To avoid this,
+    # we override _get_conn
+    def _get_conn(self, timeout):
+        conn = None
+        try:
+            conn = self.pool.get(block=self.block, timeout=timeout)
+
+        except AttributeError:  # self.pool is None
+            raise urllib3.exceptions.ClosedPoolError(self, "Pool is closed.")
+
+        except six.moves.queue.Empty:
+            if self.block:
+                raise urllib3.exceptions.EmptyPoolError(
+                    self,
+                    "Pool reached maximum size and no more "
+                    "connections are allowed."
+                )
+            pass  # Oh well, we'll create a new connection then
+
+        return conn or self._new_conn()
+
+
+class SSHAdapter(requests.adapters.HTTPAdapter):
+
+    __attrs__ = requests.adapters.HTTPAdapter.__attrs__ + [
+        'pools', 'timeout', 'ssh_client',
+    ]
+
+    def __init__(self, base_url, timeout=60,
+                 pool_connections=constants.DEFAULT_NUM_POOLS):
+        self.ssh_client = paramiko.SSHClient()
+        self.ssh_client.load_system_host_keys()
+
+        parsed = six.moves.urllib_parse.urlparse(base_url)
+        self.ssh_client.connect(
+            parsed.hostname, parsed.port, parsed.username,
+        )
+        self.timeout = timeout
+        self.pools = RecentlyUsedContainer(
+            pool_connections, dispose_func=lambda p: p.close()
+        )
+        super(SSHAdapter, self).__init__()
+
+    def get_connection(self, url, proxies=None):
+        with self.pools.lock:
+            pool = self.pools.get(url)
+            if pool:
+                return pool
+
+            pool = SSHConnectionPool(
+                self.ssh_client, self.timeout
+            )
+            self.pools[url] = pool
+
+        return pool
+
+    def close(self):
+        self.pools.clear()
+        self.ssh_client.close()
diff --git a/docker/types/daemon.py b/docker/types/daemon.py
index ee8624e..700f9a9 100644
--- a/docker/types/daemon.py
+++ b/docker/types/daemon.py
@@ -5,6 +5,8 @@ try:
 except ImportError:
     import urllib3
 
+from ..errors import DockerException
+
 
 class CancellableStream(object):
     """
@@ -55,9 +57,17 @@ class CancellableStream(object):
                 elif hasattr(sock_raw, '_sock'):
                     sock = sock_raw._sock
 
+            elif hasattr(sock_fp, 'channel'):
+                # We're working with a paramiko (SSH) channel, which doesn't
+                # support cancelable streams with the current implementation
+                raise DockerException(
+                    'Cancellable streams not supported for the SSH protocol'
+                )
             else:
                 sock = sock_fp._sock
-            if isinstance(sock, urllib3.contrib.pyopenssl.WrappedSocket):
+
+            if hasattr(urllib3.contrib, 'pyopenssl') and isinstance(
+                    sock, urllib3.contrib.pyopenssl.WrappedSocket):
                 sock = sock.socket
 
             sock.shutdown(socket.SHUT_RDWR)
diff --git a/docker/utils/utils.py b/docker/utils/utils.py
index fe3b9a5..4e04caf 100644
--- a/docker/utils/utils.py
+++ b/docker/utils/utils.py
@@ -1,10 +1,11 @@
 import base64
+import json
 import os
 import os.path
-import json
 import shlex
-from distutils.version import StrictVersion
+import string
 from datetime import datetime
+from distutils.version import StrictVersion
 
 import six
 
@@ -13,11 +14,12 @@ from .. import tls
 
 if six.PY2:
     from urllib import splitnport
+    from urlparse import urlparse
 else:
-    from urllib.parse import splitnport
+    from urllib.parse import splitnport, urlparse
 
 DEFAULT_HTTP_HOST = "127.0.0.1"
-DEFAULT_UNIX_SOCKET = "http+unix://var/run/docker.sock"
+DEFAULT_UNIX_SOCKET = "http+unix:///var/run/docker.sock"
 DEFAULT_NPIPE = 'npipe:////./pipe/docker_engine'
 
 BYTE_UNITS = {
@@ -212,75 +214,93 @@ def parse_repository_tag(repo_name):
     return repo_name, None
 
 
-# Based on utils.go:ParseHost http://tinyurl.com/nkahcfh
-# fd:// protocol unsupported (for obvious reasons)
-# Added support for http and https
-# Protocol translation: tcp -> http, unix -> http+unix
 def parse_host(addr, is_win32=False, tls=False):
-    proto = "http+unix"
-    port = None
     path = ''
+    port = None
+    host = None
 
+    # Sensible defaults
     if not addr and is_win32:
-        addr = DEFAULT_NPIPE
-
+        return DEFAULT_NPIPE
     if not addr or addr.strip() == 'unix://':
         return DEFAULT_UNIX_SOCKET
 
     addr = addr.strip()
-    if addr.startswith('http://'):
-        addr = addr.replace('http://', 'tcp://')
-    if addr.startswith('http+unix://'):
-        addr = addr.replace('http+unix://', 'unix://')
 
-    if addr == 'tcp://':
+    parsed_url = urlparse(addr)
+    proto = parsed_url.scheme
+    if not proto or any([x not in string.ascii_letters + '+' for x in proto]):
+        # https://bugs.python.org/issue754016
+        parsed_url = urlparse('//' + addr, 'tcp')
+        proto = 'tcp'
+
+    if proto == 'fd':
+        raise errors.DockerException('fd protocol is not implemented')
+
+    # These protos are valid aliases for our library but not for the
+    # official spec
+    if proto == 'http' or proto == 'https':
+        tls = proto == 'https'
+        proto = 'tcp'
+    elif proto == 'http+unix':
+        proto = 'unix'
+
+    if proto not in ('tcp', 'unix', 'npipe', 'ssh'):
         raise errors.DockerException(
-            "Invalid bind address format: {0}".format(addr)
+            "Invalid bind address protocol: {}".format(addr)
+        )
+
+    if proto == 'tcp' and not parsed_url.netloc:
+        # "tcp://" is exceptionally disallowed by convention;
+        # omitting a hostname for other protocols is fine
+        raise errors.DockerException(
+            'Invalid bind address format: {}'.format(addr)
         )
-    elif addr.startswith('unix://'):
-        addr = addr[7:]
-    elif addr.startswith('tcp://'):
-        proto = 'http{0}'.format('s' if tls else '')
-        addr = addr[6:]
-    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:
-        if "://" in addr:
-            raise errors.DockerException(
-                "Invalid bind address protocol: {0}".format(addr)
-            )
-        proto = "https" if tls else "http"
 
-    if proto in ("http", "https"):
-        address_parts = addr.split('/', 1)
-        host = address_parts[0]
-        if len(address_parts) == 2:
-            path = '/' + address_parts[1]
-        host, port = splitnport(host)
+    if any([
+        parsed_url.params, parsed_url.query, parsed_url.fragment,
+        parsed_url.password
+    ]):
+        raise errors.DockerException(
+            'Invalid bind address format: {}'.format(addr)
+        )
 
-        if port is None:
-            raise errors.DockerException(
-                "Invalid port: {0}".format(addr)
-            )
+    if parsed_url.path and proto == 'ssh':
+        raise errors.DockerException(
+            'Invalid bind address format: no path allowed for this protocol:'
+            ' {}'.format(addr)
+        )
+    else:
+        path = parsed_url.path
+        if proto == 'unix' and parsed_url.hostname is not None:
+            # For legacy reasons, we consider unix://path
+            # to be valid and equivalent to unix:///path
+            path = '/'.join((parsed_url.hostname, path))
+
+    if proto in ('tcp', 'ssh'):
+        # parsed_url.hostname strips brackets from IPv6 addresses,
+        # which can be problematic hence our use of splitnport() instead.
+        host, port = splitnport(parsed_url.netloc)
+        if port is None or port < 0:
+            if proto != 'ssh':
+                raise errors.DockerException(
+                    'Invalid bind address format: port is required:'
+                    ' {}'.format(addr)
+                )
+            port = 22
 
         if not host:
             host = DEFAULT_HTTP_HOST
-    else:
-        host = addr
 
-    if proto in ("http", "https") and port == -1:
-        raise errors.DockerException(
-            "Bind address needs a port: {0}".format(addr))
+    # Rewrite schemes to fit library internals (requests adapters)
+    if proto == 'tcp':
+        proto = 'http{}'.format('s' if tls else '')
+    elif proto == 'unix':
+        proto = 'http+unix'
 
-    if proto == "http+unix" or proto == 'npipe':
-        return "{0}://{1}".format(proto, host).rstrip('/')
-    return "{0}://{1}:{2}{3}".format(proto, host, port, path).rstrip('/')
+    if proto in ('http+unix', 'npipe'):
+        return "{}://{}".format(proto, path).rstrip('/')
+    return '{0}://{1}:{2}{3}'.format(proto, host, port, path).rstrip('/')
 
 
 def parse_devices(devices):
diff --git a/requirements.txt b/requirements.txt
index c0ce59a..d13e9d6 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -9,6 +9,7 @@ enum34==1.1.6
 idna==2.5
 ipaddress==1.0.18
 packaging==16.8
+paramiko==2.4.2
 pycparser==2.17
 pyOpenSSL==18.0.0
 pyparsing==2.2.0
diff --git a/setup.py b/setup.py
index 390783d..3ad572b 100644
--- a/setup.py
+++ b/setup.py
@@ -42,6 +42,9 @@ extras_require = {
     # 'requests[security] >= 2.5.2, != 2.11.0, != 2.12.2'
     'tls': ['pyOpenSSL>=17.5.0', 'cryptography>=1.3.4', 'idna>=2.0.0'],
 
+    # Only required when connecting using the ssh:// protocol
+    'ssh': ['paramiko>=2.4.2'],
+
 }
 
 version = None
diff --git a/tests/helpers.py b/tests/helpers.py
index b36d6d7..f912bd8 100644
--- a/tests/helpers.py
+++ b/tests/helpers.py
@@ -10,6 +10,7 @@ import six
 import socket
 
 import docker
+import paramiko
 import pytest
 
 
@@ -121,6 +122,9 @@ def assert_cat_socket_detached_with_keys(sock, inputs):
     if getattr(sock, 'family', -9) == getattr(socket, 'AF_UNIX', -1):
         with pytest.raises(socket.error):
             sock.sendall(b'make sure the socket is closed\n')
+    elif isinstance(sock, paramiko.Channel):
+        with pytest.raises(OSError):
+            sock.sendall(b'make sure the socket is closed\n')
     else:
         sock.sendall(b"make sure the socket is closed\n")
         data = sock.recv(128)
diff --git a/tests/integration/api_container_test.py b/tests/integration/api_container_test.py
index 6ce846b..02f3603 100644
--- a/tests/integration/api_container_test.py
+++ b/tests/integration/api_container_test.py
@@ -883,6 +883,8 @@ Line2'''
         assert logs == (snippet + '\n').encode(encoding='ascii')
 
     @pytest.mark.timeout(5)
+    @pytest.mark.skipif(os.environ.get('DOCKER_HOST', '').startswith('ssh://'),
+                        reason='No cancellable streams over SSH')
     def test_logs_streaming_and_follow_and_cancel(self):
         snippet = 'Flowering Nights (Sakuya Iyazoi)'
         container = self.client.create_container(
@@ -1255,6 +1257,8 @@ class AttachContainerTest(BaseAPIIntegrationTest):
         assert output == 'hello\n'.encode(encoding='ascii')
 
     @pytest.mark.timeout(5)
+    @pytest.mark.skipif(os.environ.get('DOCKER_HOST', '').startswith('ssh://'),
+                        reason='No cancellable streams over SSH')
     def test_attach_stream_and_cancel(self):
         container = self.client.create_container(
             BUSYBOX, 'sh -c "echo hello && sleep 60"',
diff --git a/tests/integration/api_network_test.py b/tests/integration/api_network_test.py
index b6726d0..db37cbd 100644
--- a/tests/integration/api_network_test.py
+++ b/tests/integration/api_network_test.py
@@ -8,8 +8,8 @@ from .base import BaseAPIIntegrationTest, BUSYBOX
 
 class TestNetworks(BaseAPIIntegrationTest):
     def tearDown(self):
-        super(TestNetworks, self).tearDown()
         self.client.leave_swarm(force=True)
+        super(TestNetworks, self).tearDown()
 
     def create_network(self, *args, **kwargs):
         net_name = random_name()
diff --git a/tests/integration/api_plugin_test.py b/tests/integration/api_plugin_test.py
index 1150b09..38f9d12 100644
--- a/tests/integration/api_plugin_test.py
+++ b/tests/integration/api_plugin_test.py
@@ -3,7 +3,7 @@ import os
 import docker
 import pytest
 
-from .base import BaseAPIIntegrationTest, TEST_API_VERSION
+from .base import BaseAPIIntegrationTest
 from ..helpers import requires_api_version
 
 SSHFS = 'vieux/sshfs:latest'
@@ -13,27 +13,27 @@ SSHFS = 'vieux/sshfs:latest'
 class PluginTest(BaseAPIIntegrationTest):
     @classmethod
     def teardown_class(cls):
-        c = docker.APIClient(
-            version=TEST_API_VERSION, timeout=60,
-            **docker.utils.kwargs_from_env()
-        )
+        client = cls.get_client_instance()
         try:
-            c.remove_plugin(SSHFS, force=True)
+            client.remove_plugin(SSHFS, force=True)
         except docker.errors.APIError:
             pass
 
     def teardown_method(self, method):
+        client = self.get_client_instance()
         try:
-            self.client.disable_plugin(SSHFS)
+            client.disable_plugin(SSHFS)
         except docker.errors.APIError:
             pass
 
         for p in self.tmp_plugins:
             try:
-                self.client.remove_plugin(p, force=True)
+                client.remove_plugin(p, force=True)
             except docker.errors.APIError:
                 pass
 
+        client.close()
+
     def ensure_plugin_installed(self, plugin_name):
         try:
             return self.client.inspect_plugin(plugin_name)
diff --git a/tests/integration/api_swarm_test.py b/tests/integration/api_swarm_test.py
index dbf3786..b58dabc 100644
--- a/tests/integration/api_swarm_test.py
+++ b/tests/integration/api_swarm_test.py
@@ -13,14 +13,13 @@ class SwarmTest(BaseAPIIntegrationTest):
         self._unlock_key = None
 
     def tearDown(self):
-        super(SwarmTest, self).tearDown()
         try:
             if self._unlock_key:
                 self.client.unlock_swarm(self._unlock_key)
         except docker.errors.APIError:
             pass
-
         force_leave_swarm(self.client)
+        super(SwarmTest, self).tearDown()
 
     @requires_api_version('1.24')
     def test_init_swarm_simple(self):
diff --git a/tests/integration/base.py b/tests/integration/base.py
index 56c23ed..262769d 100644
--- a/tests/integration/base.py
+++ b/tests/integration/base.py
@@ -29,41 +29,44 @@ class BaseIntegrationTest(unittest.TestCase):
 
     def tearDown(self):
         client = docker.from_env(version=TEST_API_VERSION)
-        for img in self.tmp_imgs:
-            try:
-                client.api.remove_image(img)
-            except docker.errors.APIError:
-                pass
-        for container in self.tmp_containers:
-            try:
-                client.api.remove_container(container, force=True, v=True)
-            except docker.errors.APIError:
-                pass
-        for network in self.tmp_networks:
-            try:
-                client.api.remove_network(network)
-            except docker.errors.APIError:
-                pass
-        for volume in self.tmp_volumes:
-            try:
-                client.api.remove_volume(volume)
-            except docker.errors.APIError:
-                pass
-
-        for secret in self.tmp_secrets:
-            try:
-                client.api.remove_secret(secret)
-            except docker.errors.APIError:
-                pass
-
-        for config in self.tmp_configs:
-            try:
-                client.api.remove_config(config)
-            except docker.errors.APIError:
-                pass
-
-        for folder in self.tmp_folders:
-            shutil.rmtree(folder)
+        try:
+            for img in self.tmp_imgs:
+                try:
+                    client.api.remove_image(img)
+                except docker.errors.APIError:
+                    pass
+            for container in self.tmp_containers:
+                try:
+                    client.api.remove_container(container, force=True, v=True)
+                except docker.errors.APIError:
+                    pass
+            for network in self.tmp_networks:
+                try:
+                    client.api.remove_network(network)
+                except docker.errors.APIError:
+                    pass
+            for volume in self.tmp_volumes:
+                try:
+                    client.api.remove_volume(volume)
+                except docker.errors.APIError:
+                    pass
+
+            for secret in self.tmp_secrets:
+                try:
+                    client.api.remove_secret(secret)
+                except docker.errors.APIError:
+                    pass
+
+            for config in self.tmp_configs:
+                try:
+                    client.api.remove_config(config)
+                except docker.errors.APIError:
+                    pass
+
+            for folder in self.tmp_folders:
+                shutil.rmtree(folder)
+        finally:
+            client.close()
 
 
 class BaseAPIIntegrationTest(BaseIntegrationTest):
diff --git a/tests/integration/models_containers_test.py b/tests/integration/models_containers_test.py
index ab41ea5..b48f6fb 100644
--- a/tests/integration/models_containers_test.py
+++ b/tests/integration/models_containers_test.py
@@ -1,3 +1,4 @@
+import os
 import tempfile
 import threading
 
@@ -146,6 +147,8 @@ class ContainerCollectionTest(BaseIntegrationTest):
         assert logs[1] == b'world\n'
 
     @pytest.mark.timeout(5)
+    @pytest.mark.skipif(os.environ.get('DOCKER_HOST', '').startswith('ssh://'),
+                        reason='No cancellable streams over SSH')
     def test_run_with_streamed_logs_and_cancel(self):
         client = docker.from_env(version=TEST_API_VERSION)
         out = client.containers.run(
diff --git a/tests/unit/utils_test.py b/tests/unit/utils_test.py
index 8880cfe..c862a1c 100644
--- a/tests/unit/utils_test.py
+++ b/tests/unit/utils_test.py
@@ -272,6 +272,11 @@ class ParseHostTest(unittest.TestCase):
             'tcp://',
             'udp://127.0.0.1',
             'udp://127.0.0.1:2375',
+            'ssh://:22/path',
+            'tcp://netloc:3333/path?q=1',
+            'unix:///sock/path#fragment',
+            'https://netloc:3333/path;params',
+            'ssh://:clearpassword@host:22',
         ]
 
         valid_hosts = {
@@ -281,7 +286,7 @@ class ParseHostTest(unittest.TestCase):
             'http://:7777': 'http://127.0.0.1:7777',
             'https://kokia.jp:2375': 'https://kokia.jp:2375',
             'unix:///var/run/docker.sock': 'http+unix:///var/run/docker.sock',
-            'unix://': 'http+unix://var/run/docker.sock',
+            'unix://': 'http+unix:///var/run/docker.sock',
             '12.234.45.127:2375/docker/engine': (
                 'http://12.234.45.127:2375/docker/engine'
             ),
@@ -294,6 +299,9 @@ class ParseHostTest(unittest.TestCase):
             '[fd12::82d1]:2375/docker/engine': (
                 'http://[fd12::82d1]:2375/docker/engine'
             ),
+            'ssh://': 'ssh://127.0.0.1:22',
+            'ssh://user@localhost:22': 'ssh://user@localhost:22',
+            'ssh://user@remote': 'ssh://user@remote:22',
         }
 
         for host in invalid_hosts:
@@ -304,7 +312,7 @@ class ParseHostTest(unittest.TestCase):
             assert parse_host(host, None) == expected
 
     def test_parse_host_empty_value(self):
-        unix_socket = 'http+unix://var/run/docker.sock'
+        unix_socket = 'http+unix:///var/run/docker.sock'
         npipe = 'npipe:////./pipe/docker_engine'
 
         for val in [None, '']: