Block unsafe options and protocols by default

Author: stsewd

git/cmd.py

@@ -4,6 +4,7 @@
 # This module is part of GitPython and is released under
 # the BSD License: http://www.opensource.org/licenses/bsd-license.php
 from __future__ import annotations
+import re
 from contextlib import contextmanager
 import io
 import logging
@@ -24,7 +25,7 @@
 from git.exc import CommandError
 from git.util import is_cygwin_git, cygpath, expand_path, remove_password_if_present
 
-from .exc import GitCommandError, GitCommandNotFound
+from .exc import GitCommandError, GitCommandNotFound, UnsafeOptionError, UnsafeProtocolError
 from .util import (
     LazyMixin,
     stream_copy,
@@ -262,6 +263,8 @@ class Git(LazyMixin):
 
     _excluded_ = ("cat_file_all", "cat_file_header", "_version_info")
 
+    re_unsafe_protocol = re.compile("(.+)::.+")
+
     def __getstate__(self) -> Dict[str, Any]:
         return slots_to_dict(self, exclude=self._excluded_)
 
@@ -454,6 +457,48 @@ def polish_url(cls, url: str, is_cygwin: Union[None, bool] = None) -> PathLike:
             url = url.replace("\\\\", "\\").replace("\\", "/")
         return url
 
+    @classmethod
+    def check_unsafe_protocols(cls, url: str) -> None:
+        """
+        Check for unsafe protocols.
+
+        Apart from the usual protocols (http, git, ssh),
+        Git allows "remote helpers" that have the form `<transport>::<address>`,
+        one of these helpers (`ext::`) can be used to invoke any arbitrary command.
+
+        See:
+
+        - https://git-scm.com/docs/gitremote-helpers
+        - https://git-scm.com/docs/git-remote-ext
+        """
+        match = cls.re_unsafe_protocol.match(url)
+        if match:
+            protocol = match.group(1)
+            raise UnsafeProtocolError(
+                f"The `{protocol}::` protocol looks suspicious, use `allow_unsafe_protocols=True` to allow it."
+            )
+
+    @classmethod
+    def check_unsafe_options(cls, options: List[str], unsafe_options: List[str]) -> None:
+        """
+        Check for unsafe options.
+
+        Some options that are passed to `git <command>` can be used to execute
+        arbitrary commands, this are blocked by default.
+        """
+        # Options can be of the form `foo` or `--foo bar` `--foo=bar`,
+        # so we need to check if they start with "--foo" or if they are equal to "foo".
+        bare_options = [
+            option.lstrip("-")
+            for option in unsafe_options
+        ]
+        for option in options:
+            for unsafe_option, bare_option in zip(unsafe_options, bare_options):
+                if option.startswith(unsafe_option) or option == bare_option:
+                    raise UnsafeOptionError(
+                        f"{unsafe_option} is not allowed, use `allow_unsafe_options=True` to allow it."
+                    )
+
     class AutoInterrupt(object):
         """Kill/Interrupt the stored process instance once this instance goes out of scope. It is
         used to prevent processes piling up in case iterators stop reading.

git/exc.py

@@ -37,8 +37,12 @@ class NoSuchPathError(GitError, OSError):
     """Thrown if a path could not be access by the system."""
 
 
-class UnsafeOptionsUsedError(GitError):
-    """Thrown if unsafe protocols or options are passed without overridding."""
+class UnsafeProtocolError(GitError):
+    """Thrown if unsafe protocols are passed without being explicitly allowed."""
+
+
+class UnsafeOptionError(GitError):
+    """Thrown if unsafe options are passed without being explicitly allowed."""
 
 
 class CommandError(GitError):

git/remote.py

@@ -535,6 +535,23 @@ class Remote(LazyMixin, IterableObj):
     __slots__ = ("repo", "name", "_config_reader")
     _id_attribute_ = "name"
 
+    unsafe_git_fetch_options = [
+        # This option allows users to execute arbitrary commands.
+        # https://git-scm.com/docs/git-fetch#Documentation/git-fetch.txt---upload-packltupload-packgt
+        "--upload-pack",
+    ]
+    unsafe_git_pull_options = [
+        # This option allows users to execute arbitrary commands.
+        # https://git-scm.com/docs/git-pull#Documentation/git-pull.txt---upload-packltupload-packgt
+        "--upload-pack"
+    ]
+    unsafe_git_push_options = [
+        # This option allows users to execute arbitrary commands.
+        # https://git-scm.com/docs/git-push#Documentation/git-push.txt---execltgit-receive-packgt
+        "--receive-pack",
+        "--exec",
+    ]
+
     def __init__(self, repo: "Repo", name: str) -> None:
         """Initialize a remote instance
 
@@ -611,7 +628,9 @@ def iter_items(cls, repo: "Repo", *args: Any, **kwargs: Any) -> Iterator["Remote
             yield Remote(repo, section[lbound + 1 : rbound])
         # END for each configuration section
 
-    def set_url(self, new_url: str, old_url: Optional[str] = None, **kwargs: Any) -> "Remote":
+    def set_url(
+        self, new_url: str, old_url: Optional[str] = None, allow_unsafe_protocols: bool = False, **kwargs: Any
+    ) -> "Remote":
         """Configure URLs on current remote (cf command git remote set_url)
 
         This command manages URLs on the remote.
@@ -620,15 +639,17 @@ def set_url(self, new_url: str, old_url: Optional[str] = None, **kwargs: Any) ->
         :param old_url: when set, replaces this URL with new_url for the remote
         :return: self
         """
+        if not allow_unsafe_protocols:
+            Git.check_unsafe_protocols(new_url)
         scmd = "set-url"
         kwargs["insert_kwargs_after"] = scmd
         if old_url:
-            self.repo.git.remote(scmd, self.name, new_url, old_url, **kwargs)
+            self.repo.git.remote(scmd, "--", self.name, new_url, old_url, **kwargs)
         else:
-            self.repo.git.remote(scmd, self.name, new_url, **kwargs)
+            self.repo.git.remote(scmd, "--", self.name, new_url, **kwargs)
         return self
 
-    def add_url(self, url: str, **kwargs: Any) -> "Remote":
+    def add_url(self, url: str, allow_unsafe_protocols: bool = False, **kwargs: Any) -> "Remote":
         """Adds a new url on current remote (special case of git remote set_url)
 
         This command adds new URLs to a given remote, making it possible to have
@@ -637,6 +658,8 @@ def add_url(self, url: str, **kwargs: Any) -> "Remote":
         :param url: string being the URL to add as an extra remote URL
         :return: self
         """
+        if not allow_unsafe_protocols:
+            Git.check_unsafe_protocols(url)
         return self.set_url(url, add=True)
 
     def delete_url(self, url: str, **kwargs: Any) -> "Remote":
@@ -729,7 +752,7 @@ def stale_refs(self) -> IterableList[Reference]:
         return out_refs
 
     @classmethod
-    def create(cls, repo: "Repo", name: str, url: str, **kwargs: Any) -> "Remote":
+    def create(cls, repo: "Repo", name: str, url: str, allow_unsafe_protocols: bool = False, **kwargs: Any) -> "Remote":
         """Create a new remote to the given repository
         :param repo: Repository instance that is to receive the new remote
         :param name: Desired name of the remote
@@ -739,7 +762,10 @@ def create(cls, repo: "Repo", name: str, url: str, **kwargs: Any) -> "Remote":
         :raise GitCommandError: in case an origin with that name already exists"""
         scmd = "add"
         kwargs["insert_kwargs_after"] = scmd
-        repo.git.remote(scmd, name, Git.polish_url(url), **kwargs)
+        url = Git.polish_url(url)
+        if not allow_unsafe_protocols:
+            Git.check_unsafe_protocols(url)
+        repo.git.remote(scmd, "--", name, url, **kwargs)
         return cls(repo, name)
 
     # add is an alias
@@ -921,6 +947,8 @@ def fetch(
         progress: Union[RemoteProgress, None, "UpdateProgress"] = None,
         verbose: bool = True,
         kill_after_timeout: Union[None, float] = None,
+        allow_unsafe_protocols: bool = False,
+        allow_unsafe_options: bool = False,
         **kwargs: Any,
     ) -> IterableList[FetchInfo]:
         """Fetch the latest changes for this remote
@@ -963,6 +991,14 @@ def fetch(
         else:
             args = [refspec]
 
+        if not allow_unsafe_protocols:
+            for ref in args:
+                if ref:
+                    Git.check_unsafe_protocols(ref)
+
+        if not allow_unsafe_options:
+            Git.check_unsafe_options(options=list(kwargs.keys()), unsafe_options=self.unsafe_git_fetch_options)
+
         proc = self.repo.git.fetch(
             "--", self, *args, as_process=True, with_stdout=False, universal_newlines=True, v=verbose, **kwargs
         )
@@ -976,6 +1012,8 @@ def pull(
         refspec: Union[str, List[str], None] = None,
         progress: Union[RemoteProgress, "UpdateProgress", None] = None,
         kill_after_timeout: Union[None, float] = None,
+        allow_unsafe_protocols: bool = False,
+        allow_unsafe_options: bool = False,
         **kwargs: Any,
     ) -> IterableList[FetchInfo]:
         """Pull changes from the given branch, being the same as a fetch followed
@@ -990,6 +1028,16 @@ def pull(
             # No argument refspec, then ensure the repo's config has a fetch refspec.
             self._assert_refspec()
         kwargs = add_progress(kwargs, self.repo.git, progress)
+
+        if not allow_unsafe_protocols and refspec:
+            if isinstance(refspec, str):
+                Git.check_unsafe_protocols(refspec)
+            else:
+                for ref in refspec:
+                    Git.check_unsafe_protocols(ref)
+        if not allow_unsafe_options:
+            Git.check_unsafe_options(options=list(kwargs.keys()), unsafe_options=self.unsafe_git_pull_options)
+
         proc = self.repo.git.pull(
             "--", self, refspec, with_stdout=False, as_process=True, universal_newlines=True, v=True, **kwargs
         )
@@ -1003,6 +1051,8 @@ def push(
         refspec: Union[str, List[str], None] = None,
         progress: Union[RemoteProgress, "UpdateProgress", Callable[..., RemoteProgress], None] = None,
         kill_after_timeout: Union[None, float] = None,
+        allow_unsafe_protocols: bool = False,
+        allow_unsafe_options: bool = False,
         **kwargs: Any,
     ) -> IterableList[PushInfo]:
         """Push changes from source branch in refspec to target branch in refspec.
@@ -1033,6 +1083,17 @@ def push(
             If the operation fails completely, the length of the returned IterableList will
             be 0."""
         kwargs = add_progress(kwargs, self.repo.git, progress)
+
+        if not allow_unsafe_protocols and refspec:
+            if isinstance(refspec, str):
+                Git.check_unsafe_protocols(refspec)
+            else:
+                for ref in refspec:
+                    Git.check_unsafe_protocols(ref)
+
+        if not allow_unsafe_options:
+            Git.check_unsafe_options(options=list(kwargs.keys()), unsafe_options=self.unsafe_git_push_options)
+
         proc = self.repo.git.push(
             "--",
             self,

git/repo/base.py

@@ -25,7 +25,6 @@
     GitCommandError,
     InvalidGitRepositoryError,
     NoSuchPathError,
-    UnsafeOptionsUsedError,
 )
 from git.index import IndexFile
 from git.objects import Submodule, RootModule, Commit
@@ -133,7 +132,18 @@ class Repo(object):
     re_envvars = re.compile(r"(\$(\{\s?)?[a-zA-Z_]\w*(\}\s?)?|%\s?[a-zA-Z_]\w*\s?%)")
     re_author_committer_start = re.compile(r"^(author|committer)")
     re_tab_full_line = re.compile(r"^\t(.*)$")
-    re_config_protocol_option = re.compile(r"-[-]?c(|onfig)\s+protocol\.", re.I)
+
+    unsafe_git_clone_options = [
+        # This option allows users to execute arbitrary commands.
+        # https://git-scm.com/docs/git-clone#Documentation/git-clone.txt---upload-packltupload-packgt
+        "--upload-pack",
+        "-u",
+        # Users can override configuration variables
+        # like `protocol.allow` or `core.gitProxy` to execute arbitrary commands.
+        # https://git-scm.com/docs/git-clone#Documentation/git-clone.txt---configltkeygtltvaluegt
+        "--config",
+        "-c",
+    ]
 
     # invariants
     # represents the configuration level of a configuration file
@@ -961,7 +971,7 @@ def blame(
         file: str,
         incremental: bool = False,
         rev_opts: Optional[List[str]] = None,
-        **kwargs: Any
+        **kwargs: Any,
     ) -> List[List[Commit | List[str | bytes] | None]] | Iterator[BlameEntry] | None:
         """The blame information for the given file at the given revision.
 
@@ -1152,6 +1162,8 @@ def _clone(
         odb_default_type: Type[GitCmdObjectDB],
         progress: Union["RemoteProgress", "UpdateProgress", Callable[..., "RemoteProgress"], None] = None,
         multi_options: Optional[List[str]] = None,
+        allow_unsafe_protocols: bool = False,
+        allow_unsafe_options: bool = False,
         **kwargs: Any,
     ) -> "Repo":
         odbt = kwargs.pop("odbt", odb_default_type)
@@ -1173,6 +1185,12 @@ def _clone(
         multi = None
         if multi_options:
             multi = shlex.split(" ".join(multi_options))
+
+        if not allow_unsafe_protocols:
+            Git.check_unsafe_protocols(str(url))
+        if not allow_unsafe_options and multi_options:
+            Git.check_unsafe_options(options=multi_options, unsafe_options=cls.unsafe_git_clone_options)
+
         proc = git.clone(
             multi,
             "--",
@@ -1221,27 +1239,13 @@ def _clone(
         # END handle remote repo
         return repo
 
-    @classmethod
-    def unsafe_options(
-        cls,
-        url: str,
-        multi_options: Optional[List[str]] = None,
-    ) -> bool:
-        if "ext::" in url:
-            return True
-        if multi_options is not None:
-            if any(["--upload-pack" in m for m in multi_options]):
-                return True
-            if any([re.match(cls.re_config_protocol_option, m) for m in multi_options]):
-                return True
-        return False
-
     def clone(
         self,
         path: PathLike,
         progress: Optional[Callable] = None,
         multi_options: Optional[List[str]] = None,
-        unsafe_protocols: bool = False,
+        allow_unsafe_protocols: bool = False,
+        allow_unsafe_options: bool = False,
         **kwargs: Any,
     ) -> "Repo":
         """Create a clone from this repository.
@@ -1259,15 +1263,15 @@ def clone(
             * All remaining keyword arguments are given to the git-clone command
 
         :return: ``git.Repo`` (the newly cloned repo)"""
-        if not unsafe_protocols and self.unsafe_options(path, multi_options):
-            raise UnsafeOptionsUsedError(f"{path} requires unsafe_protocols flag")
         return self._clone(
             self.git,
             self.common_dir,
             path,
             type(self.odb),
             progress,
             multi_options,
+            allow_unsafe_protocols=allow_unsafe_protocols,
+            allow_unsafe_options=allow_unsafe_options,
             **kwargs,
         )
 
@@ -1279,7 +1283,8 @@ def clone_from(
         progress: Optional[Callable] = None,
         env: Optional[Mapping[str, str]] = None,
         multi_options: Optional[List[str]] = None,
-        unsafe_protocols: bool = False,
+        allow_unsafe_protocols: bool = False,
+        allow_unsafe_options: bool = False,
         **kwargs: Any,
     ) -> "Repo":
         """Create a clone from the given URL
@@ -1300,9 +1305,17 @@ def clone_from(
         git = cls.GitCommandWrapperType(os.getcwd())
         if env is not None:
             git.update_environment(**env)
-        if not unsafe_protocols and cls.unsafe_options(url, multi_options):
-            raise UnsafeOptionsUsedError(f"{url} requires unsafe_protocols flag")
-        return cls._clone(git, url, to_path, GitCmdObjectDB, progress, multi_options, **kwargs)
+        return cls._clone(
+            git,
+            url,
+            to_path,
+            GitCmdObjectDB,
+            progress,
+            multi_options,
+            allow_unsafe_protocols=allow_unsafe_protocols,
+            allow_unsafe_options=allow_unsafe_options,
+            **kwargs,
+        )
 
     def archive(
         self,