Merge pull request #6603 from cjerdonek/issue-3799-trusted-host-requi…

…rements-txt

Fully support --trusted-host inside requirements files

docs/html/reference/pip_install.rst

@@ -153,6 +153,7 @@ The following options are supported:
   *  :ref:`--no-binary <install_--no-binary>`
   *  :ref:`--only-binary <install_--only-binary>`
   *  :ref:`--require-hashes <--require-hashes>`
+  *  :ref:`--trusted-host <--trusted-host>`
 
 For example, to specify :ref:`--no-index <--no-index>` and two
 :ref:`--find-links <--find-links>` locations:

news/3799.feature

@@ -0,0 +1 @@
+Fully support using ``--trusted-host`` inside requirements files.

src/pip/_internal/download.py

@@ -588,6 +588,8 @@ def __init__(self, *args, **kwargs):
         # well as any https:// host that we've marked as ignoring TLS errors
         # for.
         insecure_adapter = InsecureHTTPAdapter(max_retries=retries)
+        # Save this for later use in add_insecure_host().
+        self._insecure_adapter = insecure_adapter
 
         self.mount("https://", secure_adapter)
         self.mount("http://", insecure_adapter)
@@ -598,7 +600,11 @@ def __init__(self, *args, **kwargs):
         # We want to use a non-validating adapter for any requests which are
         # deemed insecure.
         for host in insecure_hosts:
-            self.mount("https://{}/".format(host), insecure_adapter)
+            self.add_insecure_host(host)
+
+    def add_insecure_host(self, host):
+        # type: (str) -> None
+        self.mount('https://{}/'.format(host), self._insecure_adapter)
 
     def request(self, method, url, *args, **kwargs):
         # Allow setting a default timeout on a session

src/pip/_internal/index.py

@@ -608,8 +608,8 @@ def create(
         # type: (...) -> PackageFinder
         """Create a PackageFinder.
 
-        :param trusted_hosts: Domains that we won't emit warnings for when
-            not using HTTPS.
+        :param trusted_hosts: Domains not to emit warnings for when not using
+            HTTPS.
         :param session: The Session to use to make requests.
         :param format_control: A FormatControl object or None. Used to control
             the selection of source packages / binary packages when consulting
@@ -676,12 +676,23 @@ def set_allow_all_prereleases(self):
         # type: () -> None
         self.candidate_evaluator.allow_all_prereleases = True
 
-    def extend_trusted_hosts(self, hosts):
-        # type: (List[str]) -> None
-        for host in hosts:
-            if host in self.trusted_hosts:
-                continue
-            self.trusted_hosts.append(host)
+    def add_trusted_host(self, host, source=None):
+        # type: (str, Optional[str]) -> None
+        """
+        :param source: An optional source string, for logging where the host
+            string came from.
+        """
+        # It is okay to add a previously added host because PipSession stores
+        # the resulting prefixes in a dict.
+        msg = 'adding trusted host: {!r}'.format(host)
+        if source is not None:
+            msg += ' (from {})'.format(source)
+        logger.info(msg)
+        self.session.add_insecure_host(host)
+        if host in self.trusted_hosts:
+            return
+
+        self.trusted_hosts.append(host)
 
     def iter_secure_origins(self):
         # type: () -> Iterator[SecureOrigin]

src/pip/_internal/req/req_file.py

@@ -250,8 +250,9 @@ def process_line(
             finder.find_links.append(value)
         if opts.pre:
             finder.set_allow_all_prereleases()
-        if opts.trusted_hosts:
-            finder.extend_trusted_hosts(opts.trusted_hosts)
+        for host in opts.trusted_hosts or []:
+            source = 'line {} of {}'.format(line_number, filename)
+            finder.add_trusted_host(host, source=source)
 
 
 def break_args_options(line):

tests/unit/test_index.py

@@ -163,16 +163,61 @@ def test_create__target_python(self):
         assert actual_target_python is target_python
         assert actual_target_python.py_version_info == (3, 7, 3)
 
-    def test_extend_trusted_hosts(self):
-        trusted_hosts = ['host1', 'host2']
-        finder = make_test_finder(trusted_hosts=trusted_hosts)
-
-        # Check that extend_trusted_hosts() prevents duplicates.
-        finder.extend_trusted_hosts(['host2', 'host3', 'host2'])
-        assert finder.trusted_hosts == ['host1', 'host2', 'host3'], (
+    def test_add_trusted_host(self):
+        # Leave a gap to test how the ordering is affected.
+        trusted_hosts = ['host1', 'host3']
+        session = PipSession(insecure_hosts=trusted_hosts)
+        finder = make_test_finder(
+            session=session,
+            trusted_hosts=trusted_hosts,
+        )
+        insecure_adapter = session._insecure_adapter
+        prefix2 = 'https://host2/'
+        prefix3 = 'https://host3/'
+
+        # Confirm some initial conditions as a baseline.
+        assert finder.trusted_hosts == ['host1', 'host3']
+        assert session.adapters[prefix3] is insecure_adapter
+        assert prefix2 not in session.adapters
+
+        # Test adding a new host.
+        finder.add_trusted_host('host2')
+        assert finder.trusted_hosts == ['host1', 'host3', 'host2']
+        # Check that prefix3 is still present.
+        assert session.adapters[prefix3] is insecure_adapter
+        assert session.adapters[prefix2] is insecure_adapter
+
+        # Test that adding the same host doesn't create a duplicate.
+        finder.add_trusted_host('host3')
+        assert finder.trusted_hosts == ['host1', 'host3', 'host2'], (
             'actual: {}'.format(finder.trusted_hosts)
         )
 
+    def test_add_trusted_host__logging(self, caplog):
+        """
+        Test logging when add_trusted_host() is called.
+        """
+        trusted_hosts = ['host1']
+        session = PipSession(insecure_hosts=trusted_hosts)
+        finder = make_test_finder(
+            session=session,
+            trusted_hosts=trusted_hosts,
+        )
+        with caplog.at_level(logging.INFO):
+            # Test adding an existing host.
+            finder.add_trusted_host('host1', source='somewhere')
+            finder.add_trusted_host('host2')
+            # Test calling add_trusted_host() on the same host twice.
+            finder.add_trusted_host('host2')
+
+        actual = [(r.levelname, r.message) for r in caplog.records]
+        expected = [
+            ('INFO', "adding trusted host: 'host1' (from somewhere)"),
+            ('INFO', "adding trusted host: 'host2'"),
+            ('INFO', "adding trusted host: 'host2'"),
+        ]
+        assert actual == expected
+
     def test_iter_secure_origins(self):
         trusted_hosts = ['host1', 'host2']
         finder = make_test_finder(trusted_hosts=trusted_hosts)

tests/unit/test_req_file.py

@@ -1,3 +1,4 @@
+import logging
 import os
 import subprocess
 import textwrap
@@ -314,9 +315,21 @@ def test_set_finder_extra_index_urls(self, finder):
         list(process_line("--extra-index-url=url", "file", 1, finder=finder))
         assert finder.index_urls == ['url']
 
-    def test_set_finder_trusted_host(self, finder):
-        list(process_line("--trusted-host=url", "file", 1, finder=finder))
-        assert finder.trusted_hosts == ['url']
+    def test_set_finder_trusted_host(self, caplog, finder):
+        with caplog.at_level(logging.INFO):
+            list(process_line(
+                "--trusted-host=host", "file.txt", 1, finder=finder,
+            ))
+        assert finder.trusted_hosts == ['host']
+        session = finder.session
+        assert session.adapters['https://host/'] is session._insecure_adapter
+
+        # Test the log message.
+        actual = [(r.levelname, r.message) for r in caplog.records]
+        expected = [
+            ('INFO', "adding trusted host: 'host' (from line 1 of file.txt)"),
+        ]
+        assert actual == expected
 
     def test_noop_always_unzip(self, finder):
         # noop, but confirm it can be set