CVE-2020-15720: Improper Certificate Validation (CWE-295)
value: In Dogtag PKI through 10.8.3, the pki.client.PKIConnection class did not enable python-requests certificate validation. Since the verify parameter was hard-coded in all request functions, it was not possible to override the setting. As a result, tools making use of this class, such as the pki-server command, may have been vulnerable to Person-in-the-Middle attacks in certain non-localhost use cases. This is fixed in 10.9.0-b1.
CWE description: The product does not validate, or incorrectly validates, a certificate.
CVSS v3 base score: 6.8
Published: 2020-07-14T14:15Z
Repository: https://github.com/dogtagpki/pki
Language: Python
Files changed: base/common/python/pki/client.py
Commit date: 2020-06-18 11:33:12 -0400
Commit message: Enable certificate verification in PKIConnection

To PKIConnection's initialization handler, we introduce a new argument,
cert_paths, which takes a string or iterable; each unit of which is
treated as a capath or cafile depending on whether or not it is a
directory. See ssl.SSLContext.load_verify_locations for more
information. This enables both PKI and IPA to specify independent CA
file locations at the same time and have fallback if this does not work.

Because some users might've already loaded the CA certificate into the
system-wide CA certificate store (if they're running Dogtag in
production), we also inclue the global trust store.

Resolves: rh-bz#1426572

Signed-off-by: Alexander Scheel <ascheel@redhat.com>

Diff: diff --git a/base/common/python/pki/client.py b/base/common/python/pki/client.py
index ebc67ced3..a0d2e0b74 100644
--- a/base/common/python/pki/client.py
+++ b/base/common/python/pki/client.py
@@ -23,15 +23,17 @@ from __future__ import absolute_import
 from __future__ import print_function
 
 import functools
 import inspect
 import logging
+import os
 import ssl
 import warnings
 
 import requests
 from requests import adapters
+from requests.adapters import DEFAULT_POOLBLOCK, DEFAULT_POOLSIZE, DEFAULT_RETRIES
 try:
     from requests.packages.urllib3.exceptions import InsecureRequestWarning
 except ImportError:
     from urllib3.exceptions import InsecureRequestWarning
 
@@ -53,22 +55,72 @@ def catch_insecure_warning(func):
             return func(self, *args, **kwargs)
     return wrapper
 
 
 class SSLContextAdapter(adapters.HTTPAdapter):
-    """Custom SSLContext Adapter for requests
     """
+    Custom SSLContext Adapter for requests
+    """
+
+    def __init__(self, pool_connections=DEFAULT_POOLSIZE,
+                 pool_maxsize=DEFAULT_POOLSIZE, max_retries=DEFAULT_RETRIES,
+                 pool_block=DEFAULT_POOLBLOCK, verify=True,
+                 cert_paths=None):
+        self.verify = verify
+        self.cafiles = []
+        self.capaths = []
+
+        cert_paths = cert_paths or []
+
+        if isinstance(cert_paths, str):
+            cert_paths = [cert_paths]
+
+        for path in cert_paths:
+            path = path and os.path.expanduser(path)
+
+            if os.path.isdir(path):
+                self.capaths.append(path)
+            elif os.path.exists(path):
+                self.cafiles.append(path)
+            else:
+                logger.warning("cert_path missing; not used for validation: %s",
+                               path)
+
+        # adapters.HTTPAdapter.__init__ calls our init_poolmanager, which needs
+        # our cafiles/capaths variables we set up above.
+        super(SSLContextAdapter, self).__init__(pool_connections=pool_connections,
+                                                pool_maxsize=pool_maxsize,
+                                                max_retries=max_retries,
+                                                pool_block=pool_block)
+
     def init_poolmanager(self, connections, maxsize,
                          block=adapters.DEFAULT_POOLBLOCK, **pool_kwargs):
         context = ssl.SSLContext(
             ssl.PROTOCOL_TLS  # pylint: disable=no-member
         )
 
         # Enable post handshake authentication for TLS 1.3
         if getattr(context, "post_handshake_auth", None) is not None:
             context.post_handshake_auth = True
 
+        # Load from the system trust store when possible; per documentation
+        # this call could silently fail and refuse to configure any
+        # certificates. In this instance, the user should provide a
+        # certificate manually.
+        context.set_default_verify_paths()
+
+        # Load any specific certificate paths that have been specified during
+        # adapter initialization.
+        for cafile in self.cafiles:
+            context.load_verify_locations(cafile=cafile)
+        for capath in self.capaths:
+            context.load_verify_locations(capath=capath)
+
+        if self.verify:
+            # Enable certificate verification
+            context.verify_mode = ssl.VerifyMode.CERT_REQUIRED  # pylint: disable=no-member
+
         pool_kwargs['ssl_context'] = context
         return super().init_poolmanager(
             connections, maxsize, block, **pool_kwargs
         )
 
@@ -79,11 +131,11 @@ class PKIConnection:
     subsystem.
     """
 
     def __init__(self, protocol='http', hostname='localhost', port='8080',
                  subsystem=None, accept='application/json',
-                 trust_env=None, verify=False):
+                 trust_env=None, verify=True, cert_paths=None):
         """
         Set the parameters for a python-requests based connection to a
         Dogtag subsystem.
         :param protocol: http or https
         :type protocol: str
@@ -101,10 +153,13 @@ class PKIConnection:
            requests settings (default: yes)
         :type trust_env: bool, None
         :param verify: verify TLS/SSL connections and configure CA certs
            (default: no)
         :type verify: None, bool, str
+        :param cert_paths: paths to CA certificates / directories in OpenSSL
+          format. (default: None)
+        :type cert_paths: None, str, list
         :return: PKIConnection object.
         """
 
         self.protocol = protocol
         self.hostname = hostname
@@ -121,11 +176,11 @@ class PKIConnection:
             self.serverURI = self.rootURI + '/' + subsystem
         else:
             self.serverURI = self.rootURI
 
         self.session = requests.Session()
-        self.session.mount("https://", SSLContextAdapter())
+        self.session.mount("https://", SSLContextAdapter(verify=verify, cert_paths=cert_paths))
         self.session.trust_env = trust_env
         self.session.verify = verify
 
         if accept:
             self.session.headers.update({'Accept': accept})

Vulnerable code:     """Custom SSLContext Adapter for requests
                 trust_env=None, verify=False):
        self.session.mount("https://", SSLContextAdapter())
Fixed code: import os
from requests.adapters import DEFAULT_POOLBLOCK, DEFAULT_POOLSIZE, DEFAULT_RETRIES
    Custom SSLContext Adapter for requests
    """

    def __init__(self, pool_connections=DEFAULT_POOLSIZE,
                 pool_maxsize=DEFAULT_POOLSIZE, max_retries=DEFAULT_RETRIES,
                 pool_block=DEFAULT_POOLBLOCK, verify=True,
                 cert_paths=None):
        self.verify = verify
        self.cafiles = []
        self.capaths = []

        cert_paths = cert_paths or []

        if isinstance(cert_paths, str):
            cert_paths = [cert_paths]

        for path in cert_paths:
            path = path and os.path.expanduser(path)

            if os.path.isdir(path):
                self.capaths.append(path)
            elif os.path.exists(path):
                self.cafiles.append(path)
            else:
                logger.warning("cert_path missing; not used for validation: %s",
                               path)

        # adapters.HTTPAdapter.__init__ calls our init_poolmanager, which needs
        # our cafiles/capaths variables we set up above.
        super(SSLContextAdapter, self).__init__(pool_connections=pool_connections,
                                                pool_maxsize=pool_maxsize,
                                                max_retries=max_retries,
                                                pool_block=pool_block)

        # Load from the system trust store when possible; per documentation
        # this call could silently fail and refuse to configure any
        # certificates. In this instance, the user should provide a
        # certificate manually.
        context.set_default_verify_paths()

        # Load any specific certificate paths that have been specified during
        # adapter initialization.
        for cafile in self.cafiles:
            context.load_verify_locations(cafile=cafile)
        for capath in self.capaths:
            context.load_verify_locations(capath=capath)

        if self.verify:
            # Enable certificate verification
            context.verify_mode = ssl.VerifyMode.CERT_REQUIRED  # pylint: disable=no-member

                 trust_env=None, verify=True, cert_paths=None):
        :param cert_paths: paths to CA certificates / directories in OpenSSL
          format. (default: None)
        :type cert_paths: None, str, list
        self.session.mount("https://", SSLContextAdapter(verify=verify, cert_paths=cert_paths))