Add support for SSL_CERT_FILE and SSL_CERT_DIR (#307)

docs/environment_variables.md

@@ -2,6 +2,12 @@ Environment Variables
 =====================
 
 The HTTPX library can be configured via environment variables.
+Environment variables are used by default. To ignore environment variables, `trust_env` has to be set `False`.
+There are two ways to set `trust_env` to disable environment variables:
+
+* On the client via `httpx.Client(trust_env=False)`
+* Per request via `client.get("<url>", trust_env=False)`
+
 Here is a list of environment variables that HTTPX recognizes
 and what function they serve:
 
@@ -80,6 +86,36 @@ CLIENT_HANDSHAKE_TRAFFIC_SECRET XXXX
 CLIENT_TRAFFIC_SECRET_0 XXXX
 ```
 
+`SSL_CERT_FILE`
+-----------
+
+Valid values: a filename
+
+if this environment variable is set then HTTPX will load
+CA certificate from the specified file instead of the default
+location.
+
+Example:
+
+```console
+SSL_CERT_FILE=/path/to/ca-certs/ca-bundle.crt python -c "import httpx; httpx.get('https://example.com')"
+```
+
+`SSL_CERT_DIR`
+-----------
+
+Valid values: a directory
+
+if this environment variable is set then HTTPX will load
+CA certificates from the specified location instead of the default
+location.
+
+Example:
+
+```console
+SSL_CERT_DIR=/path/to/ca-certs/ python -c "import httpx; httpx.get('https://example.com')"
+```
+
 `HTTP_PROXY`, `HTTPS_PROXY`, `ALL_PROXY`
 ----------------------------------------
 

httpx/config.py

@@ -6,6 +6,7 @@ from pathlib import Path
 import certifi
 
 from .__version__ import __version__
+from .utils import get_ca_bundle_from_env
 
 CertTypes = typing.Union[str, typing.Tuple[str, str], typing.Tuple[str, str, str]]
 VerifyTypes = typing.Union[str, bool, ssl.SSLContext]
@@ -117,6 +118,11 @@ class SSLConfig:
         """
         Return an SSL context for verified connections.
         """
+        if self.trust_env and self.verify is True:
+            ca_bundle = get_ca_bundle_from_env()
+            if ca_bundle is not None:
+                self.verify = ca_bundle  # type: ignore
+
         if isinstance(self.verify, bool):
             ca_bundle_path = DEFAULT_CA_BUNDLE_PATH
         elif Path(self.verify).exists():

httpx/utils.py

@@ -111,6 +111,18 @@ def get_netrc_login(host: str) -> typing.Optional[typing.Tuple[str, str, str]]:
     return netrc_info.authenticators(host)  # type: ignore
 
 
+def get_ca_bundle_from_env() -> typing.Optional[str]:
+    if "SSL_CERT_FILE" in os.environ:
+        ssl_file = Path(os.environ["SSL_CERT_FILE"])
+        if ssl_file.is_file():
+            return str(ssl_file)
+    if "SSL_CERT_DIR" in os.environ:
+        ssl_path = Path(os.environ["SSL_CERT_DIR"])
+        if ssl_path.is_dir():
+            return str(ssl_path)
+    return None
+
+
 def parse_header_links(value: str) -> typing.List[typing.Dict[str, str]]:
     """
     Returns a list of parsed link headers, for more info see:

tests/test_config.py

@@ -1,5 +1,8 @@
+import os
+import socket
 import ssl
 import sys
+from pathlib import Path
 
 import pytest
 
@@ -26,6 +29,30 @@ def test_load_ssl_config_verify_existing_file():
     assert context.check_hostname is True
 
 
+@pytest.mark.parametrize("config", ("SSL_CERT_FILE", "SSL_CERT_DIR"))
+def test_load_ssl_config_verify_env_file(https_server, ca_cert_pem_file, config):
+    os.environ[config] = (
+        ca_cert_pem_file
+        if config.endswith("_FILE")
+        else str(Path(ca_cert_pem_file).parent)
+    )
+    ssl_config = httpx.SSLConfig(trust_env=True)
+    context = ssl_config.load_ssl_context()
+    assert context.verify_mode == ssl.VerifyMode.CERT_REQUIRED
+    assert context.check_hostname is True
+    assert ssl_config.verify == os.environ[config]
+
+    # Skipping 'SSL_CERT_DIR' functional test for now because
+    # we're unable to get the certificate within the directory to
+    # load into the SSLContext. :(
+    if config == "SSL_CERT_FILE":
+        host = https_server.url.host
+        port = https_server.url.port
+        conn = socket.create_connection((host, port))
+        context.wrap_socket(conn, server_hostname=host)
+        assert len(context.get_ca_certs()) == 1
+
+
 def test_load_ssl_config_verify_directory():
     path = httpx.config.DEFAULT_CA_BUNDLE_PATH.parent
     ssl_config = httpx.SSLConfig(verify=path)

tests/test_utils.py

@@ -8,6 +8,7 @@ import httpx
 from httpx import utils
 from httpx.utils import (
     ElapsedTimer,
+    get_ca_bundle_from_env,
     get_environment_proxies,
     get_netrc_login,
     guess_json_utf,
@@ -120,6 +121,42 @@ async def test_httpx_debug_enabled_stderr_logging(server, capsys, httpx_debug):
     logging.getLogger("httpx").handlers = []
 
 
+def test_get_ssl_cert_file():
+    # Two environments is not set.
+    assert get_ca_bundle_from_env() is None
+
+    os.environ["SSL_CERT_DIR"] = "tests/"
+    # SSL_CERT_DIR is correctly set, SSL_CERT_FILE is not set.
+    assert get_ca_bundle_from_env() == "tests"
+
+    del os.environ["SSL_CERT_DIR"]
+    os.environ["SSL_CERT_FILE"] = "tests/test_utils.py"
+    # SSL_CERT_FILE is correctly set, SSL_CERT_DIR is not set.
+    assert get_ca_bundle_from_env() == "tests/test_utils.py"
+
+    os.environ["SSL_CERT_FILE"] = "wrongfile"
+    # SSL_CERT_FILE is set with wrong file,  SSL_CERT_DIR is not set.
+    assert get_ca_bundle_from_env() is None
+
+    del os.environ["SSL_CERT_FILE"]
+    os.environ["SSL_CERT_DIR"] = "wrongpath"
+    # SSL_CERT_DIR is set with wrong path,  SSL_CERT_FILE is not set.
+    assert get_ca_bundle_from_env() is None
+
+    os.environ["SSL_CERT_DIR"] = "tests/"
+    os.environ["SSL_CERT_FILE"] = "tests/test_utils.py"
+    # Two environments is correctly set.
+    assert get_ca_bundle_from_env() == "tests/test_utils.py"
+
+    os.environ["SSL_CERT_FILE"] = "wrongfile"
+    # Two environments is set but SSL_CERT_FILE is not a file.
+    assert get_ca_bundle_from_env() == "tests"
+
+    os.environ["SSL_CERT_DIR"] = "wrongpath"
+    # Two environments is set but both are not correct.
+    assert get_ca_bundle_from_env() is None
+
+
 @pytest.mark.asyncio
 async def test_elapsed_timer():
     with ElapsedTimer() as timer: