Merge pull request #3156 from tk0miya/3095_tls_cacerts

Fix #3095: Add tls_verify and tls_cacerts to support self-signed servers

CHANGES

@@ -9,6 +9,9 @@ Incompatible changes
 Features added
 --------------
 
+* #3095: Add :confval:`tls_verify` and :confval:`tls_cacerts` to support
+  self-signed HTTPS servers in linkcheck and intersphinx
+
 Bugs fixed
 ----------
 

doc/config.rst

@@ -318,6 +318,21 @@ General configuration
 
    .. versionadded:: 1.3
 
+.. confval:: tls_verify
+
+   If true, Sphinx verifies server certifications.  Default is ``True``.
+
+   .. versionadded:: 1.5
+
+.. confval:: tls_cacerts
+
+   A path to a certification file of CA or a path to directory which
+   contains the certificates.  This also allows a dictionary mapping
+   hostname to the path to certificate file.
+   The certificates are used to verify server certifications.
+
+   .. versionadded:: 1.5
+
 Project information
 -------------------
 

sphinx/builders/linkcheck.py

@@ -32,10 +32,10 @@ except ImportError:
         pass
 
 from sphinx.builders import Builder
-from sphinx.util import encode_uri
+from sphinx.util import encode_uri, requests
 from sphinx.util.console import purple, red, darkgreen, darkgray, \
     darkred, turquoise
-from sphinx.util.requests import requests, useragent_header, is_ssl_error
+from sphinx.util.requests import is_ssl_error
 
 
 class AnchorCheckParser(HTMLParser):
@@ -87,7 +87,6 @@ class CheckExternalLinksBuilder(Builder):
         self.good = set()
         self.broken = {}
         self.redirected = {}
-        self.headers = dict(useragent_header)
         # set a timeout for non-responding servers
         socket.setdefaulttimeout(5.0)
         # create output file
@@ -131,7 +130,7 @@ class CheckExternalLinksBuilder(Builder):
             try:
                 if anchor and self.app.config.linkcheck_anchors:
                     # Read the whole document and see if #anchor exists
-                    response = requests.get(req_url, stream=True, headers=self.headers,
+                    response = requests.get(req_url, stream=True, config=self.app.config,
                                             **kwargs)
                     found = check_anchor(response, unquote(anchor))
 
@@ -141,12 +140,12 @@ class CheckExternalLinksBuilder(Builder):
                     try:
                         # try a HEAD request first, which should be easier on
                         # the server and the network
-                        response = requests.head(req_url, headers=self.headers, **kwargs)
+                        response = requests.head(req_url, config=self.app.config, **kwargs)
                         response.raise_for_status()
                     except HTTPError as err:
                         # retry with GET request if that fails, some servers
                         # don't like HEAD requests.
-                        response = requests.get(req_url, stream=True, headers=self.headers,
+                        response = requests.get(req_url, stream=True, config=self.app.config,
                                                 **kwargs)
                         response.raise_for_status()
             except HTTPError as err:

sphinx/config.py

@@ -112,6 +112,9 @@ class Config(object):
                           'code-block': l_('Listing %s')},
                          'env'),
 
+        tls_verify = (True, 'env'),
+        tls_cacerts = (None, 'env'),
+
         # pre-initialized confval for HTML builder
         html_translator_class = (None, 'html', string_classes),
     )

sphinx/ext/intersphinx.py

@@ -41,7 +41,7 @@ from docutils.utils import relative_path
 import sphinx
 from sphinx.locale import _
 from sphinx.builders.html import INVENTORY_FILENAME
-from sphinx.util.requests import requests, useragent_header
+from sphinx.util import requests
 
 
 UTF8StreamReader = codecs.lookup('utf-8')[2]
@@ -145,7 +145,7 @@ def _strip_basic_auth(url):
     return urlunsplit(frags)
 
 
-def _read_from_url(url, timeout=None):
+def _read_from_url(url, config=None):
     """Reads data from *url* with an HTTP *GET*.
 
     This function supports fetching from resources which use basic HTTP auth as
@@ -161,7 +161,7 @@ def _read_from_url(url, timeout=None):
     :return: data read from resource described by *url*
     :rtype: ``file``-like object
     """
-    r = requests.get(url, stream=True, timeout=timeout, headers=dict(useragent_header))
+    r = requests.get(url, stream=True, config=config, timeout=config.intersphinx_timeout)
     r.raise_for_status()
     r.raw.url = r.url
     return r.raw
@@ -202,7 +202,7 @@ def fetch_inventory(app, uri, inv):
         uri = _strip_basic_auth(uri)
     try:
         if '://' in inv:
-            f = _read_from_url(inv, timeout=app.config.intersphinx_timeout)
+            f = _read_from_url(inv, config=app.config)
         else:
             f = open(path.join(app.srcdir, inv), 'rb')
     except Exception as err:

sphinx/util/requests.py

@@ -14,7 +14,10 @@ from __future__ import absolute_import
 import requests
 import warnings
 import pkg_resources
-from requests.packages.urllib3.exceptions import SSLError
+
+from six import string_types
+from six.moves.urllib.parse import urlsplit
+from requests.packages.urllib3.exceptions import SSLError, InsecureRequestWarning
 
 # try to load requests[security]
 try:
@@ -45,6 +48,7 @@ useragent_header = [('User-Agent',
 
 
 def is_ssl_error(exc):
+    """Check an exception is SSLError."""
     if isinstance(exc, SSLError):
         return True
     else:
@@ -53,3 +57,57 @@ def is_ssl_error(exc):
             return True
         else:
             return False
+
+
+def _get_tls_cacert(url, config):
+    """Get addiotinal CA cert for a specific URL.
+
+    This also returns ``False`` if verification is disabled.
+    And returns ``True`` if additional CA cert not found.
+    """
+    if not config.tls_verify:
+        return False
+
+    certs = getattr(config, 'tls_cacerts', None)
+    if not certs:
+        return True
+    elif isinstance(certs, (string_types, tuple)):
+        return certs
+    else:
+        hostname = urlsplit(url)[1]
+        if '@' in hostname:
+            hostname = hostname.split('@')[1]
+
+        return certs.get(hostname, True)
+
+
+def get(url, **kwargs):
+    """Sends a GET request like requests.get().
+
+    This sets up User-Agent header and TLS verification automatically."""
+    kwargs.setdefault('headers', dict(useragent_header))
+    config = kwargs.pop('config', None)
+    if config:
+        kwargs.setdefault('verify', _get_tls_cacert(url, config))
+
+    with warnings.catch_warnings():
+        if not kwargs.get('verify'):
+            # ignore InsecureRequestWarning if verify=False
+            warnings.filterwarnings("ignore", category=InsecureRequestWarning)
+        return requests.get(url, **kwargs)
+
+
+def head(url, **kwargs):
+    """Sends a HEAD request like requests.head().
+
+    This sets up User-Agent header and TLS verification automatically."""
+    kwargs.setdefault('headers', dict(useragent_header))
+    config = kwargs.pop('config', None)
+    if config:
+        kwargs.setdefault('verify', _get_tls_cacert(url, config))
+
+    with warnings.catch_warnings():
+        if not kwargs.get('verify'):
+            # ignore InsecureRequestWarning if verify=False
+            warnings.filterwarnings("ignore", category=InsecureRequestWarning)
+        return requests.get(url, **kwargs)