Use config.json for detachKeys

Signed-off-by: Fumiaki Matsushima <[email protected]>

Author: mtsmfm

docker/api/client.py

@@ -32,7 +32,7 @@
 )
 from ..tls import TLSConfig
 from ..transport import SSLAdapter, UnixAdapter
-from ..utils import utils, check_resource, update_headers
+from ..utils import utils, check_resource, update_headers, config
 from ..utils.socket import frames_iter, socket_raw_iter
 from ..utils.json_stream import json_stream
 try:
@@ -106,6 +106,7 @@ def __init__(self, base_url=None, version=None,
         self.headers['User-Agent'] = user_agent
 
         self._auth_configs = auth.load_config()
+        self._general_configs = config.load_general_config()
 
         base_url = utils.parse_host(
             base_url, IS_WINDOWS_PLATFORM, tls=bool(tls)

docker/api/container.py

@@ -66,6 +66,7 @@ def attach_socket(self, container, params=None, ws=False):
             container (str): The container to attach to.
             params (dict): Dictionary of request parameters (e.g. ``stdout``,
                 ``stderr``, ``stream``).
+                For ``detachKeys``, ~/.docker/config.json is used by default.
             ws (bool): Use websockets instead of raw HTTP.
 
         Raises:
@@ -79,6 +80,11 @@ def attach_socket(self, container, params=None, ws=False):
                 'stream': 1
             }
 
+        if 'detachKeys' not in params \
+                and 'detachKeys' in self._general_configs:
+
+            params['detachKeys'] = self._general_configs['detachKeys']
+
         if ws:
             return self._attach_websocket(container, params)
 

docker/api/exec_api.py

@@ -9,7 +9,7 @@ class ExecApiMixin(object):
     @utils.check_resource('container')
     def exec_create(self, container, cmd, stdout=True, stderr=True,
                     stdin=False, tty=False, privileged=False, user='',
-                    environment=None, workdir=None):
+                    environment=None, workdir=None, detach_keys=None):
         """
         Sets up an exec instance in a running container.
 
@@ -27,6 +27,11 @@ def exec_create(self, container, cmd, stdout=True, stderr=True,
                 the following format ``["PASSWORD=xxx"]`` or
                 ``{"PASSWORD": "xxx"}``.
             workdir (str): Path to working directory for this exec session
+            detach_keys (str): Override the key sequence for detaching
+                a container. Format is a single character `[a-Z]`
+                or `ctrl-<value>` where `<value>` is one of:
+                `a-z`, `@`, `^`, `[`, `,` or `_`.
+                ~/.docker/config.json is used by default.
 
         Returns:
             (dict): A dictionary with an exec ``Id`` key.
@@ -74,6 +79,11 @@ def exec_create(self, container, cmd, stdout=True, stderr=True,
                 )
             data['WorkingDir'] = workdir
 
+        if detach_keys:
+            data['detachKeys'] = detach_keys
+        elif 'detachKeys' in self._general_configs:
+            data['detachKeys'] = self._general_configs['detachKeys']
+
         url = self._url('/containers/{0}/exec', container)
         res = self._post_json(url, data=data)
         return self._result(res, True)

docker/auth.py

@@ -1,18 +1,15 @@
 import base64
 import json
 import logging
-import os
 
 import dockerpycreds
 import six
 
 from . import errors
-from .constants import IS_WINDOWS_PLATFORM
+from .utils import config
 
 INDEX_NAME = 'docker.io'
 INDEX_URL = 'https://index.{0}/v1/'.format(INDEX_NAME)
-DOCKER_CONFIG_FILENAME = os.path.join('.docker', 'config.json')
-LEGACY_DOCKER_CONFIG_FILENAME = '.dockercfg'
 TOKEN_USERNAME = '<token>'
 
 log = logging.getLogger(__name__)
@@ -105,10 +102,10 @@ def resolve_authconfig(authconfig, registry=None):
         log.debug("Found {0}".format(repr(registry)))
         return authconfig[registry]
 
-    for key, config in six.iteritems(authconfig):
+    for key, conf in six.iteritems(authconfig):
         if resolve_index_name(key) == registry:
             log.debug("Found {0}".format(repr(key)))
-            return config
+            return conf
 
     log.debug("No entry found")
     return None
@@ -223,44 +220,6 @@ def parse_auth(entries, raise_on_error=False):
     return conf
 
 
-def find_config_file(config_path=None):
-    paths = list(filter(None, [
-        config_path,  # 1
-        config_path_from_environment(),  # 2
-        os.path.join(home_dir(), DOCKER_CONFIG_FILENAME),  # 3
-        os.path.join(home_dir(), LEGACY_DOCKER_CONFIG_FILENAME),  # 4
-    ]))
-
-    log.debug("Trying paths: {0}".format(repr(paths)))
-
-    for path in paths:
-        if os.path.exists(path):
-            log.debug("Found file at path: {0}".format(path))
-            return path
-
-    log.debug("No config file found")
-
-    return None
-
-
-def config_path_from_environment():
-    config_dir = os.environ.get('DOCKER_CONFIG')
-    if not config_dir:
-        return None
-    return os.path.join(config_dir, os.path.basename(DOCKER_CONFIG_FILENAME))
-
-
-def home_dir():
-    """
-    Get the user's home directory, using the same logic as the Docker Engine
-    client - use %USERPROFILE% on Windows, $HOME/getuid on POSIX.
-    """
-    if IS_WINDOWS_PLATFORM:
-        return os.environ.get('USERPROFILE', '')
-    else:
-        return os.path.expanduser('~')
-
-
 def load_config(config_path=None):
     """
     Loads authentication data from a Docker configuration file in the given
@@ -269,7 +228,7 @@ def load_config(config_path=None):
         explicit config_path parameter > DOCKER_CONFIG environment variable >
         ~/.docker/config.json > ~/.dockercfg
     """
-    config_file = find_config_file(config_path)
+    config_file = config.find_config_file(config_path)
 
     if not config_file:
         return {}

docker/utils/config.py

@@ -0,0 +1,65 @@
+import json
+import logging
+import os
+
+from ..constants import IS_WINDOWS_PLATFORM
+
+DOCKER_CONFIG_FILENAME = os.path.join('.docker', 'config.json')
+LEGACY_DOCKER_CONFIG_FILENAME = '.dockercfg'
+
+log = logging.getLogger(__name__)
+
+
+def find_config_file(config_path=None):
+    paths = list(filter(None, [
+        config_path,  # 1
+        config_path_from_environment(),  # 2
+        os.path.join(home_dir(), DOCKER_CONFIG_FILENAME),  # 3
+        os.path.join(home_dir(), LEGACY_DOCKER_CONFIG_FILENAME),  # 4
+    ]))
+
+    log.debug("Trying paths: {0}".format(repr(paths)))
+
+    for path in paths:
+        if os.path.exists(path):
+            log.debug("Found file at path: {0}".format(path))
+            return path
+
+    log.debug("No config file found")
+
+    return None
+
+
+def config_path_from_environment():
+    config_dir = os.environ.get('DOCKER_CONFIG')
+    if not config_dir:
+        return None
+    return os.path.join(config_dir, os.path.basename(DOCKER_CONFIG_FILENAME))
+
+
+def home_dir():
+    """
+    Get the user's home directory, using the same logic as the Docker Engine
+    client - use %USERPROFILE% on Windows, $HOME/getuid on POSIX.
+    """
+    if IS_WINDOWS_PLATFORM:
+        return os.environ.get('USERPROFILE', '')
+    else:
+        return os.path.expanduser('~')
+
+
+def load_general_config(config_path=None):
+    config_file = find_config_file(config_path)
+
+    if not config_file:
+        return {}
+
+    try:
+        with open(config_file) as f:
+            return json.load(f)
+    except Exception as e:
+        log.debug(e)
+        pass
+
+    log.debug("All parsing attempts failed - returning empty config")
+    return {}

tests/helpers.py

@@ -5,6 +5,9 @@
 import tarfile
 import tempfile
 import time
+import re
+import socket
+import six
 
 import docker
 import pytest
@@ -102,3 +105,22 @@ def force_leave_swarm(client):
 
 def swarm_listen_addr():
     return '0.0.0.0:{0}'.format(random.randrange(10000, 25000))
+
+
+def assert_socket_closed_with_keys(sock, inputs):
+    if six.PY3:
+        sock = sock._sock
+
+    for i in inputs:
+        sock.send(i)
+        time.sleep(1)
+
+    with pytest.raises(socket.error):
+        sock.send(b"make sure the socket is closed\n")
+
+
+def ctrl_with(char):
+    if re.match('[a-z]', char):
+        return chr(ord(char) - ord('a') + 1).encode('ascii')
+    else:
+        raise(Exception('char must be [a-z]'))

tests/integration/api_container_test.py

@@ -1,4 +1,5 @@
 import os
+import re
 import signal
 import tempfile
 from datetime import datetime
@@ -15,8 +16,9 @@
 
 from .base import BUSYBOX, BaseAPIIntegrationTest
 from .. import helpers
-from ..helpers import requires_api_version
-import re
+from ..helpers import (
+    requires_api_version, ctrl_with, assert_socket_closed_with_keys
+)
 
 
 class ListContainersTest(BaseAPIIntegrationTest):
@@ -1223,6 +1225,58 @@ def test_attach_no_stream(self):
         output = self.client.attach(container, stream=False, logs=True)
         assert output == 'hello\n'.encode(encoding='ascii')
 
+    def test_detach_with_default(self):
+        container = self.client.create_container(
+            BUSYBOX, '/bin/sh',
+            detach=True, stdin_open=True, tty=True
+        )
+        id = container['Id']
+        self.tmp_containers.append(id)
+        self.client.start(id)
+
+        sock = self.client.attach_socket(
+            container,
+            {'stdin': True, 'stream': True}
+        )
+
+        assert_socket_closed_with_keys(sock, [ctrl_with('p'), ctrl_with('q')])
+
+    def test_detach_with_config_file(self):
+        self.client._general_configs['detachKeys'] = 'ctrl-p'
+
+        container = self.client.create_container(
+            BUSYBOX, '/bin/sh',
+            detach=True, stdin_open=True, tty=True
+        )
+        id = container['Id']
+        self.tmp_containers.append(id)
+        self.client.start(id)
+
+        sock = self.client.attach_socket(
+            container,
+            {'stdin': True, 'stream': True}
+        )
+
+        assert_socket_closed_with_keys(sock, [ctrl_with('p')])
+
+    def test_detach_with_arg(self):
+        self.client._general_configs['detachKeys'] = 'ctrl-p'
+
+        container = self.client.create_container(
+            BUSYBOX, '/bin/sh',
+            detach=True, stdin_open=True, tty=True
+        )
+        id = container['Id']
+        self.tmp_containers.append(id)
+        self.client.start(id)
+
+        sock = self.client.attach_socket(
+            container,
+            {'stdin': True, 'stream': True, 'detachKeys': 'ctrl-x'}
+        )
+
+        assert_socket_closed_with_keys(sock, [ctrl_with('x')])
+
 
 class PauseTest(BaseAPIIntegrationTest):
     def test_pause_unpause(self):

tests/integration/api_exec_test.py

@@ -2,7 +2,9 @@
 from docker.utils.socket import read_exactly
 
 from .base import BaseAPIIntegrationTest, BUSYBOX
-from ..helpers import requires_api_version
+from ..helpers import (
+    requires_api_version, ctrl_with, assert_socket_closed_with_keys
+)
 
 
 class ExecTest(BaseAPIIntegrationTest):
@@ -148,3 +150,44 @@ def test_exec_command_with_workdir(self):
         res = self.client.exec_create(container, 'pwd', workdir='/var/www')
         exec_log = self.client.exec_start(res)
         assert exec_log == b'/var/www\n'
+
+    def test_detach_with_default(self):
+        container = self.client.create_container(BUSYBOX, 'cat',
+                                                 detach=True, stdin_open=True)
+        id = container['Id']
+        self.client.start(id)
+        self.tmp_containers.append(id)
+
+        exec_id = self.client.exec_create(id, '/bin/sh', stdin=True, tty=True)
+        sock = self.client.exec_start(exec_id, tty=True, socket=True)
+
+        assert_socket_closed_with_keys(sock, [ctrl_with('p'), ctrl_with('q')])
+
+    def test_detach_with_config_file(self):
+        self.client._general_configs['detachKeys'] = 'ctrl-p'
+        container = self.client.create_container(BUSYBOX, 'cat',
+                                                 detach=True, stdin_open=True)
+        id = container['Id']
+        self.client.start(id)
+        self.tmp_containers.append(id)
+
+        exec_id = self.client.exec_create(id, '/bin/sh', stdin=True, tty=True)
+        sock = self.client.exec_start(exec_id, tty=True, socket=True)
+
+        assert_socket_closed_with_keys(sock, [ctrl_with('p')])
+
+    def test_detach_with_arg(self):
+        self.client._general_configs['detachKeys'] = 'ctrl-p'
+        container = self.client.create_container(BUSYBOX, 'cat',
+                                                 detach=True, stdin_open=True)
+        id = container['Id']
+        self.client.start(id)
+        self.tmp_containers.append(id)
+
+        exec_id = self.client.exec_create(
+            id, '/bin/sh',
+            stdin=True, tty=True, detach_keys='ctrl-x'
+        )
+        sock = self.client.exec_start(exec_id, tty=True, socket=True)
+
+        assert_socket_closed_with_keys(sock, [ctrl_with('x')])