context: load TLS for contexts without a client cert

docker/context/context.py

@@ -113,33 +113,43 @@ def _load_meta(cls, name):
             metadata["Endpoints"][k]["Host"] = v.get(
                 "Host", get_context_host(None, False))
             metadata["Endpoints"][k]["SkipTLSVerify"] = bool(
-                v.get("SkipTLSVerify", True))
+                v.get("SkipTLSVerify", False))
 
         return metadata
 
     def _load_certs(self):
         certs = {}
         tls_dir = get_tls_dir(self.name)
         for endpoint in self.endpoints.keys():
-            if not os.path.isdir(os.path.join(tls_dir, endpoint)):
-                continue
             ca_cert = None
             cert = None
             key = None
-            for filename in os.listdir(os.path.join(tls_dir, endpoint)):
-                if filename.startswith("ca"):
-                    ca_cert = os.path.join(tls_dir, endpoint, filename)
-                elif filename.startswith("cert"):
-                    cert = os.path.join(tls_dir, endpoint, filename)
-                elif filename.startswith("key"):
-                    key = os.path.join(tls_dir, endpoint, filename)
-            if all([ca_cert, cert, key]):
-                verify = None
-                if endpoint == "docker" and not self.endpoints["docker"].get(
-                        "SkipTLSVerify", False):
-                    verify = True
-                certs[endpoint] = TLSConfig(
-                    client_cert=(cert, key), ca_cert=ca_cert, verify=verify)
+            endpoint_dir = os.path.join(tls_dir, endpoint)
+            if os.path.isdir(endpoint_dir):
+                for filename in os.listdir(endpoint_dir):
+                    if filename.startswith("ca"):
+                        ca_cert = os.path.join(endpoint_dir, filename)
+                    elif filename.startswith("cert"):
+                        cert = os.path.join(endpoint_dir, filename)
+                    elif filename.startswith("key"):
+                        key = os.path.join(endpoint_dir, filename)
+            skip_verify = False
+            if endpoint == "docker":
+                # Like the Docker CLI, the docker endpoint uses TLS when it
+                # has any TLS material or skip-tls-verify is set; a client
+                # certificate is optional.
+                skip_verify = self.endpoints["docker"].get(
+                    "SkipTLSVerify", False)
+                if not (ca_cert or cert or key or skip_verify):
+                    continue
+            elif not all([ca_cert, cert, key]):
+                continue
+            verify = None
+            if endpoint == "docker" and not skip_verify:
+                verify = True
+            client_cert = (cert, key) if cert and key else None
+            certs[endpoint] = TLSConfig(
+                client_cert=client_cert, ca_cert=ca_cert, verify=verify)
         self.tls_cfg = certs
         self.tls_path = tls_dir
 
@@ -233,9 +243,8 @@ def TLSConfig(self):
     def TLSMaterial(self):
         certs = {}
         for endpoint, tls in self.tls_cfg.items():
-            cert, key = tls.cert
-            certs[endpoint] = list(
-                map(os.path.basename, [tls.ca_cert, cert, key]))
+            files = [tls.ca_cert, *(tls.cert or ())]
+            certs[endpoint] = [os.path.basename(f) for f in files if f]
         return {
             "TLSMaterial": certs
         }

tests/unit/context_test.py

@@ -1,4 +1,10 @@
+import hashlib
+import json
+import os
+import shutil
+import tempfile
 import unittest
+from unittest import mock
 
 import pytest
 
@@ -49,3 +55,91 @@ def test_context_inspect_without_params(self):
             DEFAULT_NPIPE,
             DEFAULT_UNIX_SOCKET[5:],
         )
+
+
+class LoadContextTLSTest(unittest.TestCase):
+    """Contexts written by ``docker context create`` use TLS whenever the
+    endpoint has TLS material or ``skip-tls-verify=true``, even without a
+    client certificate."""
+
+    def setUp(self):
+        self.config_dir = tempfile.mkdtemp()
+        self.addCleanup(shutil.rmtree, self.config_dir)
+        patcher = mock.patch.dict(
+            os.environ, {'DOCKER_CONFIG': self.config_dir}
+        )
+        patcher.start()
+        self.addCleanup(patcher.stop)
+        with open(os.path.join(self.config_dir, 'config.json'), 'w') as f:
+            json.dump({}, f)
+
+    def write_context(self, name, skip_tls_verify=False, tls_files=()):
+        context_id = hashlib.sha256(name.encode('utf-8')).hexdigest()
+        meta_dir = os.path.join(
+            self.config_dir, 'contexts', 'meta', context_id
+        )
+        os.makedirs(meta_dir)
+        with open(os.path.join(meta_dir, 'meta.json'), 'w') as f:
+            json.dump({
+                'Name': name,
+                'Metadata': {},
+                'Endpoints': {'docker': {
+                    'Host': 'tcp://testdomain:2376',
+                    'SkipTLSVerify': skip_tls_verify,
+                }},
+            }, f)
+        if tls_files:
+            tls_dir = os.path.join(
+                self.config_dir, 'contexts', 'tls', context_id, 'docker'
+            )
+            os.makedirs(tls_dir)
+            for filename in tls_files:
+                with open(os.path.join(tls_dir, filename), 'w') as f:
+                    f.write('pem')
+            return tls_dir
+
+    def test_ca_only_context_uses_tls(self):
+        tls_dir = self.write_context('ca-only', tls_files=['ca.pem'])
+        ctx = ContextAPI.get_context('ca-only')
+        tls = ctx.TLSConfig
+        assert tls is not None
+        assert tls.ca_cert == os.path.join(tls_dir, 'ca.pem')
+        assert tls.cert is None
+        assert tls.verify is True
+        assert ctx.TLSMaterial == {'TLSMaterial': {'docker': ['ca.pem']}}
+
+        params = ContextAPI.kwargs_from_context(name='ca-only')
+        client = docker.APIClient(version='1.45', **params)
+        assert client.base_url == 'https://testdomain:2376'
+        assert client.verify == os.path.join(tls_dir, 'ca.pem')
+        assert client.cert is None
+
+    def test_skip_tls_verify_context_uses_tls(self):
+        self.write_context('insecure', skip_tls_verify=True)
+        ctx = ContextAPI.get_context('insecure')
+        tls = ctx.TLSConfig
+        assert tls is not None
+        assert not tls.verify
+
+        params = ContextAPI.kwargs_from_context(name='insecure')
+        client = docker.APIClient(version='1.45', **params)
+        assert client.base_url == 'https://testdomain:2376'
+        assert not client.verify
+
+    def test_client_cert_context_uses_tls(self):
+        tls_dir = self.write_context(
+            'mtls', tls_files=['ca.pem', 'cert.pem', 'key.pem']
+        )
+        tls = ContextAPI.get_context('mtls').TLSConfig
+        assert tls.ca_cert == os.path.join(tls_dir, 'ca.pem')
+        assert tls.cert == (
+            os.path.join(tls_dir, 'cert.pem'),
+            os.path.join(tls_dir, 'key.pem'),
+        )
+        assert tls.verify is True
+
+    def test_plain_tcp_context_has_no_tls(self):
+        self.write_context('plain')
+        assert ContextAPI.get_context('plain').TLSConfig is None
+        params = ContextAPI.kwargs_from_context(name='plain')
+        assert params == {'base_url': 'tcp://testdomain:2376'}