add support for git credential helpers

Introduces `GitCredentialsHttpClient` that will retry http requests
failing with 401/404 with authorization headers generated from
credentials returned by git credentials helpers (if any).

https://www.git-scm.com/docs/gitcredentials

fixes jelmer#873

dulwich/client.py

@@ -46,7 +46,8 @@
 import socket
 import subprocess
 import sys
-from typing import Any, Callable, Dict, List, Optional, Set, Tuple, IO
+from typing import (IO, Any, Callable, Dict, Generator, List, Optional, Set,
+                    Tuple)
 
 from urllib.parse import (
     quote as urlquote,
@@ -59,7 +60,8 @@
 
 
 import dulwich
-from dulwich.config import get_xdg_config_home_path
+from dulwich.config import Config, StackedConfig, get_xdg_config_home_path
+from dulwich.credentials import CredentialHelper, CredentialNotFoundError
 from dulwich.errors import (
     GitProtocolError,
     NotGitRepository,
@@ -2165,7 +2167,7 @@ def __init__(
         base_url,
         dumb=None,
         pool_manager=None,
-        config=None,
+        config: Optional[Config] = None,
         username=None,
         password=None,
         **kwargs
@@ -2236,7 +2238,63 @@ def _http_request(self, url, headers=None, data=None):
         return resp, resp.read
 
 
-HttpGitClient = Urllib3HttpGitClient
+class GitCredentialsHttpClient(Urllib3HttpGitClient):
+    """HTTP git client which uses credentials from git credential helpers"""
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        if not self.config:
+            # Note that using the StackedConfig default backends does not include
+            # the repo's local git config, but this should only happen for
+            # clones (no local config yet) and for dulwich CLI commands
+            self.config = StackedConfig.default()
+        self.credentials = None
+
+    def _http_request(self, url, *args, **kwargs):
+        while True:
+            try:
+                return super()._http_request(url, *args, **kwargs)
+            except (
+                HTTPUnauthorized,
+                NotGitRepository,  # some git servers returns 404 instead of 401 (github)
+            ) as exc:
+                if self.credentials is None:
+                    self.credentials = self.get_credentials(url)
+                try:
+                    credentials = next(self.credentials)
+                except StopIteration:
+                    raise exc
+
+                import base64
+
+                basic_auth = credentials[b"username"] + b":" + credentials[b"password"]
+                self.pool_manager.headers.update(
+                    {
+                        "authorization": f"Basic {base64.b64encode(basic_auth).decode('ascii')}"
+                    }
+                )
+
+    def get_credentials(self, url: str) -> Generator[Dict[bytes, bytes], None, None]:
+        assert self.config
+        if isinstance(self.config, StackedConfig):
+            backends = self.config.backends
+        else:
+            backends = [self.config]
+
+        for config in backends:
+            try:
+                helper = CredentialHelper.from_config(config, url)
+            except KeyError:
+                # no credential helpers in the given config
+                continue
+
+            try:
+                yield helper.get(url)
+            except CredentialNotFoundError:
+                continue
+
+
+HttpGitClient = GitCredentialsHttpClient
 
 
 def _win32_url_to_path(parsed) -> str:

dulwich/credentials.py

@@ -0,0 +1,141 @@
+# credentials.py -- support for git credential helpers
+
+# Copyright (C) 2022 Daniele Trifirò <[email protected]>
+#
+# Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU
+# General Public License as public by the Free Software Foundation; version 2.0
+# or (at your option) any later version. You can redistribute it and/or
+# modify it under the terms of either of these two licenses.
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# You should have received a copy of the licenses; if not, see
+# <http://www.gnu.org/licenses/> for a copy of the GNU General Public License
+# and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache
+# License, Version 2.0.
+#
+
+"""Support for git credential helpers
+
+https://git-scm.com/book/en/v2/Git-Tools-Credential-Storage
+
+Currently Dulwich supports only the `get` operation
+
+"""
+import os
+import shlex
+import shutil
+import subprocess
+import sys
+from typing import Any, Dict, List, Optional, Union
+
+from dulwich.config import ConfigDict
+
+
+class CredentialNotFoundError(Exception):
+    pass
+
+
+class CredentialHelper:
+    """Helper for retrieving credentials for http/https git remotes
+
+    Usage:
+    >>> helper = CredentialHelper("store") # Use `git credential-store`
+    >>> credentials = helper.get("https://github.com/dtrifiro/aprivaterepo")
+    >>> username = credentials["username"]
+    >>> password = credentials["password"]
+    """
+
+    def __init__(self, command: str):
+        self._command = command
+        self._run_kwargs: Dict[str, Any] = {}
+        if self._command[0] == "!":
+            # On Windows this will only work in git-bash and/or WSL2
+            self._run_kwargs["shell"] = True
+
+    def _prepare_command(self) -> Union[str, List[str]]:
+        if self._command[0] == "!":
+            return self._command[1:]
+
+        argv = shlex.split(self._command)
+        if sys.platform == "win32":
+            # Windows paths are mangled by shlex
+            argv[0] = self._command.split(maxsplit=1)[0]
+
+        if os.path.isabs(argv[0]):
+            return argv
+
+        executable = f"git-credential-{argv[0]}"
+        if not shutil.which(executable) and shutil.which("git"):
+            # If the helper cannot be found in PATH, it might be
+            # a C git helper in GIT_EXEC_PATH
+            git_exec_path = subprocess.check_output(
+                ("git", "--exec-path"),
+                universal_newlines=True,  # TODO: replace universal_newlines with `text` when dropping 3.6
+            ).strip()
+            if shutil.which(executable, path=git_exec_path):
+                executable = os.path.join(git_exec_path, executable)
+
+        return [executable, *argv[1:]]
+
+    def get(self, url: str) -> Dict[bytes, bytes]:
+        cmd = self._prepare_command()
+        if isinstance(cmd, str):
+            cmd += " get"
+        else:
+            cmd.append("get")
+
+        helper_input = f"url={url}{os.linesep}".encode("ascii")
+
+        try:
+            res = subprocess.run(  # type: ignore # breaks on 3.6
+                cmd,
+                check=True,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                input=helper_input,
+                **self._run_kwargs,
+            )
+        except subprocess.CalledProcessError as exc:
+            raise CredentialNotFoundError(exc.stderr) from exc
+        except FileNotFoundError as exc:
+            raise CredentialNotFoundError("Helper not found") from exc
+
+        credentials = {}
+        for line in res.stdout.strip().splitlines():
+            try:
+                key, value = line.split(b"=")
+                credentials[key] = value
+            except ValueError:
+                continue
+
+        if not all(
+            (credentials, b"username" in credentials, b"password" in credentials)
+        ):
+            raise CredentialNotFoundError("Could not get credentials from helper")
+
+        return credentials
+
+    def store(self, *args, **kwargs):
+        """Store the credential, if applicable to the helper"""
+        raise NotImplementedError
+
+    def erase(self, *args, **kwargs):
+        """Remove a matching credential, if any, from the helper’s storage"""
+        raise NotImplementedError
+
+    @classmethod
+    def from_config(
+        cls, config: ConfigDict, url: Optional[str] = None
+    ) -> "CredentialHelper":
+        # We will try to get the url-specific credential section, in case that
+        # is not defined, config.get() will fallback to the generic section.
+        encoding = config.encoding or sys.getdefaultencoding()
+        section = (b"credential", url.encode(encoding)) if url else (b"credential",)
+        command = config.get(section, b"helper")
+        assert command and isinstance(command, bytes)
+        return cls(command.decode(encoding))

dulwich/tests/__init__.py

@@ -109,6 +109,7 @@ def self_test_suite():
         "bundle",
         "client",
         "config",
+        "credentials",
         "diff_tree",
         "fastexport",
         "file",