Merge pull request #2299 from Keerthana-64/redact-config-url-credentials

fix(util): redact URL credentials in config values on the command line

Author: Byron

git/util.py

@@ -652,42 +652,60 @@ def expand_path(p: Union[None, PathLike], expand_vars: bool = True) -> Optional[
         return None
 
 
+def _redact_url_credentials(text: str) -> Optional[str]:
+    """Return `text` with the userinfo of a URL redacted, or ``None`` if it has none."""
+    try:
+        url = urlsplit(text)
+    except ValueError:
+        # This is not a valid URL.
+        return None
+    if url.password is None and url.username is None:
+        return None
+
+    # Match urllib.parse's userinfo boundary. Keeping the raw hostinfo
+    # preserves hostname case, IPv6 brackets, and port formatting.
+    _, _, hostinfo = url.netloc.rpartition("@")
+    redacted = "*****:*****" if url.password is not None else "*****"
+    return urlunsplit(url._replace(netloc=f"{redacted}@{hostinfo}"))
+
+
 def remove_password_if_present(cmdline: Sequence[str]) -> List[str]:
     """Redact credentials in URLs and HTTP Authorization extra headers in a command line.
 
+    URLs are also redacted in the key and value of a configuration assignment, such
+    as ``http.proxy=<url>``. Assignments are recognized in the ``-c <key>=<value>``,
+    ``-c<key>=<value>``, and ``--config=<key>=<value>`` forms.
+
     If nothing is found, this just returns the command line as-is.
 
     This should be used for every log line that print a command line, as well as
     exception messages.
     """
     new_cmdline = []
-    for index, to_parse in enumerate(cmdline):
-        new_cmdline.append(to_parse)
-        config_key, separator, header = to_parse.partition("=")
-        header_name, colon, _ = header.partition(":")
+    for to_parse in cmdline:
+        if to_parse.startswith("--config="):
+            option = "--config="
+        elif to_parse.startswith("-c"):
+            option = "-c"
+        else:
+            option = ""
+        assignment = to_parse[len(option) :]
+        # Redact the whole URL first, before splitting on a possible '=' in userinfo.
+        assignment = _redact_url_credentials(assignment) or assignment
+        config_key, separator, value = assignment.partition("=")
+        header_name, colon, _ = value.partition(":")
         if (
             separator
             and colon
             and config_key.lower().endswith(".extraheader")
             and header_name.strip().lower() == "authorization"
         ):
-            new_cmdline[index] = "%s%s%s%s *****" % (config_key, separator, header_name, colon)
-            continue
-        try:
-            url = urlsplit(to_parse)
-            # Remove password from the URL if present.
-            if url.password is None and url.username is None:
-                continue
-
-            # Match urllib.parse's userinfo boundary. Keeping the raw hostinfo
-            # preserves hostname case, IPv6 brackets, and port formatting.
-            _, _, hostinfo = url.netloc.rpartition("@")
-            redacted = "*****:*****" if url.password is not None else "*****"
-            url = url._replace(netloc=f"{redacted}@{hostinfo}")
-            new_cmdline[index] = urlunsplit(url)
-        except ValueError:
-            # This is not a valid URL.
+            new_cmdline.append("%s%s%s%s%s *****" % (option, config_key, separator, header_name, colon))
             continue
+
+        if separator:
+            value = _redact_url_credentials(value) or value
+        new_cmdline.append(option + config_key + separator + value)
     return new_cmdline
 
 

test/test_git.py

@@ -10,7 +10,6 @@
 import logging
 import os
 import os.path as osp
-from pathlib import Path
 import pickle
 import re
 import shutil
@@ -20,14 +19,14 @@
 import tempfile
 import threading
 import time
+from pathlib import Path
 from unittest import mock, skipUnless
 
 import ddt
 
 from git import Git, GitCommandError, GitCommandNotFound, Repo, cmd, refresh
 from git.exc import UnsafeOptionError
 from git.util import cwd, finalize_process
-
 from test.lib import TestBase, fixture_path, with_rw_directory
 
 
@@ -547,6 +546,38 @@ def test_it_redacts_authorization_extra_header_from_error(self):
         self.assertNotIn(token, message)
         self.assertIn("http.extraHeader=Authorization: *****", message)
 
+    @ddt.data(
+        (
+            "--config=http.https://[email protected]=http://bob:token@proxy",
+            "--config=http.https://*****@host.proxy=http://*****:*****@proxy",
+        ),
+        (
+            "-curl.https://user:token@host/.insteadOf=https://old/",
+            "-curl.https://*****:*****@host/.insteadOf=https://old/",
+        ),
+    )
+    @ddt.unpack
+    def test_clone_redacts_config_urls_from_trace_and_error(self, option, expected):
+        for trace in (True, "full"):
+            with self.subTest(trace=trace), tempfile.TemporaryDirectory() as directory:
+                with mock.patch.object(Git, "GIT_PYTHON_TRACE", trace):
+                    with self.assertLogs("git.cmd", level="INFO") as logs:
+                        with self.assertRaises(GitCommandError) as context:
+                            Repo.clone_from(
+                                osp.join(directory, "missing-source"),
+                                osp.join(directory, "destination"),
+                                multi_options=[option],
+                                allow_unsafe_options=True,
+                            )
+                self.assertIn(expected, context.exception.command)
+                message = str(context.exception)
+                assert logs is not None
+                log_output = " ".join(logs.output)
+                self.assertIn(expected, log_output)
+                for credential in ("alice", "bob", "user", "token"):
+                    self.assertNotIn(credential, message)
+                    self.assertNotIn(credential, log_output)
+
     def test_it_accepts_environment_variables(self):
         filename = fixture_path("ls_tree_empty")
         with open(filename, "r") as fh:

test/test_util.py

@@ -849,13 +849,70 @@ def test_remove_empty_password_keeps_host_intact(self):
         ("//user:pass@[2001:db8::1]:0080/repo", "//*****:*****@[2001:db8::1]:0080/repo"),
         ("https://user:p@[email protected]/repo", "https://*****:*****@example.com/repo"),
         ("https://user:@example.com/repo", "https://*****:*****@example.com/repo"),
+        ("https://user:[email protected]/repo", "https://*****:*****@example.com/repo"),
         ("https://@example.com/repo", "https://*****@example.com/repo"),
         ("https://example.com/repo@name?q=a@b#c@d", "https://example.com/repo@name?q=a@b#c@d"),
     )
     @ddt.unpack
     def test_remove_password_preserves_url_components(self, url, expected):
         assert remove_password_if_present([url]) == [expected]
 
+    @ddt.data(
+        (
+            ["-c", "http.proxy=http://user:[email protected]:3128"],
+            ["-c", "http.proxy=http://*****:*****@proxy.example.com:3128"],
+        ),
+        (
+            ["-c", "remote.origin.url=https://user:[email protected]/repo"],
+            ["-c", "remote.origin.url=https://*****:*****@example.com/repo"],
+        ),
+        (
+            ["--config=http.proxy=http://user:[email protected]:3128"],
+            ["--config=http.proxy=http://*****:*****@proxy.example.com:3128"],
+        ),
+        (
+            ["--config=url.https://user:[email protected]/.insteadOf=https://example.com/"],
+            ["--config=url.https://*****:*****@example.com/.insteadOf=https://example.com/"],
+        ),
+        (
+            ["--config=http.extraHeader=Authorization: Bearer pass"],
+            ["--config=http.extraHeader=Authorization: *****"],
+        ),
+        (
+            ["--config=http.https://[email protected]=http://bob:token@proxy"],
+            ["--config=http.https://*****@host.proxy=http://*****:*****@proxy"],
+        ),
+        (
+            ["-c", "http.https://[email protected]=http://bob:token@proxy"],
+            ["-c", "http.https://*****@host.proxy=http://*****:*****@proxy"],
+        ),
+        (
+            ["-chttp.https://[email protected]=http://bob:token@proxy"],
+            ["-chttp.https://*****@host.proxy=http://*****:*****@proxy"],
+        ),
+        (
+            ["-curl.https://user:token@host/.insteadOf=https://old/"],
+            ["-curl.https://*****:*****@host/.insteadOf=https://old/"],
+        ),
+        (
+            ["-chttp.proxy=http://bob:token@proxy"],
+            ["-chttp.proxy=http://*****:*****@proxy"],
+        ),
+        (
+            ["--config=http.https://alice:token@host/.extraHeader=Authorization: Bearer pass"],
+            ["--config=http.https://*****:*****@host/.extraHeader=Authorization: *****"],
+        ),
+        (
+            ["-chttp.extraHeader=Authorization: Bearer pass"],
+            ["-chttp.extraHeader=Authorization: *****"],
+        ),
+        (["-c", "user.name=pass"], ["-c", "user.name=pass"]),
+        (["-cuser.name=pass"], ["-cuser.name=pass"]),
+    )
+    @ddt.unpack
+    def test_remove_password_from_config_values(self, cmdline, expected):
+        assert remove_password_if_present(cmdline) == expected
+
 
 def test_mode_str_to_int_accepts_bytes():
     assert mode_str_to_int("100644") == 0o100644