More tests

git/objects/submodule/base.py

@@ -272,7 +272,16 @@ def _module_abspath(cls, parent_repo: "Repo", path: PathLike, name: str) -> Path
         # end
 
     @classmethod
-    def _clone_repo(cls, repo: "Repo", url: str, path: PathLike, name: str, **kwargs: Any) -> "Repo":
+    def _clone_repo(
+        cls,
+        repo: "Repo",
+        url: str,
+        path: PathLike,
+        name: str,
+        allow_unsafe_options: bool = False,
+        allow_unsafe_protocols: bool = False,
+        **kwargs: Any,
+    ) -> "Repo":
         """:return: Repo instance of newly cloned repository
         :param repo: our parent repository
         :param url: url to clone from
@@ -289,7 +298,13 @@ def _clone_repo(cls, repo: "Repo", url: str, path: PathLike, name: str, **kwargs
             module_checkout_path = osp.join(str(repo.working_tree_dir), path)
         # end
 
-        clone = git.Repo.clone_from(url, module_checkout_path, **kwargs)
+        clone = git.Repo.clone_from(
+            url,
+            module_checkout_path,
+            allow_unsafe_options=allow_unsafe_options,
+            allow_unsafe_protocols=allow_unsafe_protocols,
+            **kwargs,
+        )
         if cls._need_gitfile_submodules(repo.git):
             cls._write_git_file_and_module_config(module_checkout_path, module_abspath)
         # end
@@ -359,6 +374,8 @@ def add(
         depth: Union[int, None] = None,
         env: Union[Mapping[str, str], None] = None,
         clone_multi_options: Union[Sequence[TBD], None] = None,
+        allow_unsafe_options: bool = False,
+        allow_unsafe_protocols: bool = False,
     ) -> "Submodule":
         """Add a new submodule to the given repository. This will alter the index
         as well as the .gitmodules file, but will not create a new commit.
@@ -475,7 +492,16 @@ def add(
                 kwargs["multi_options"] = clone_multi_options
 
             # _clone_repo(cls, repo, url, path, name, **kwargs):
-            mrepo = cls._clone_repo(repo, url, path, name, env=env, **kwargs)
+            mrepo = cls._clone_repo(
+                repo,
+                url,
+                path,
+                name,
+                env=env,
+                allow_unsafe_options=allow_unsafe_options,
+                allow_unsafe_protocols=allow_unsafe_protocols,
+                **kwargs,
+            )
         # END verify url
 
         ## See #525 for ensuring git urls in config-files valid under Windows.
@@ -520,6 +546,8 @@ def update(
         keep_going: bool = False,
         env: Union[Mapping[str, str], None] = None,
         clone_multi_options: Union[Sequence[TBD], None] = None,
+        allow_unsafe_options: bool = False,
+        allow_unsafe_protocols: bool = False,
     ) -> "Submodule":
         """Update the repository of this submodule to point to the checkout
         we point at with the binsha of this instance.
@@ -643,6 +671,8 @@ def update(
                         n=True,
                         env=env,
                         multi_options=clone_multi_options,
+                        allow_unsafe_options=allow_unsafe_options,
+                        allow_unsafe_protocols=allow_unsafe_protocols,
                     )
                 # END handle dry-run
                 progress.update(

git/remote.py

@@ -658,9 +658,7 @@ def add_url(self, url: str, allow_unsafe_protocols: bool = False, **kwargs: Any)
         :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)
+        return self.set_url(url, add=True, allow_unsafe_protocols=allow_unsafe_protocols)
 
     def delete_url(self, url: str, **kwargs: Any) -> "Remote":
         """Deletes a new url on current remote (special case of git remote set_url)

test/test_remote.py

@@ -23,6 +23,8 @@
     GitCommandError,
 )
 from git.cmd import Git
+from pathlib import Path
+from git.exc import UnsafeOptionError, UnsafeProtocolError
 from test.lib import (
     TestBase,
     with_rw_repo,
@@ -690,6 +692,215 @@ def test_push_error(self, repo):
         with self.assertRaisesRegex(GitCommandError, "src refspec __BAD_REF__ does not match any"):
             rem.push("__BAD_REF__")
 
+    @with_rw_repo("HEAD")
+    def test_set_unsafe_url(self, rw_repo):
+        remote = rw_repo.remote("origin")
+        urls = [
+            "ext::sh -c touch% /tmp/pwn",
+            "fd::17/foo",
+        ]
+        for url in urls:
+            with self.assertRaises(UnsafeProtocolError):
+                remote.set_url(url)
+
+    @with_rw_repo("HEAD")
+    def test_set_unsafe_url_allowed(self, rw_repo):
+        remote = rw_repo.remote("origin")
+        urls = [
+            "ext::sh -c touch% /tmp/pwn",
+            "fd::17/foo",
+        ]
+        for url in urls:
+            remote.set_url(url, allow_unsafe_protocols=True)
+            assert list(remote.urls)[-1] == url
+
+    @with_rw_repo("HEAD")
+    def test_add_unsafe_url(self, rw_repo):
+        remote = rw_repo.remote("origin")
+        urls = [
+            "ext::sh -c touch% /tmp/pwn",
+            "fd::17/foo",
+        ]
+        for url in urls:
+            with self.assertRaises(UnsafeProtocolError):
+                remote.add_url(url)
+
+    @with_rw_repo("HEAD")
+    def test_add_unsafe_url_allowed(self, rw_repo):
+        remote = rw_repo.remote("origin")
+        urls = [
+            "ext::sh -c touch% /tmp/pwn",
+            "fd::17/foo",
+        ]
+        for url in urls:
+            remote.add_url(url, allow_unsafe_protocols=True)
+            assert list(remote.urls)[-1] == url
+
+    @with_rw_repo("HEAD")
+    def test_create_remote_unsafe_url(self, rw_repo):
+        urls = [
+            "ext::sh -c touch% /tmp/pwn",
+            "fd::17/foo",
+        ]
+        for url in urls:
+            with self.assertRaises(UnsafeProtocolError):
+                Remote.create(rw_repo, "origin", url)
+
+    @with_rw_repo("HEAD")
+    def test_create_remote_unsafe_url_allowed(self, rw_repo):
+        urls = [
+            "ext::sh -c touch% /tmp/pwn",
+            "fd::17/foo",
+        ]
+        for i, url in enumerate(urls):
+            remote = Remote.create(rw_repo, f"origin{i}", url, allow_unsafe_protocols=True)
+            assert remote.url == url
+
+    @with_rw_repo("HEAD")
+    def test_fetch_unsafe_url(self, rw_repo):
+        remote = rw_repo.remote("origin")
+        urls = [
+            "ext::sh -c touch% /tmp/pwn",
+            "fd::17/foo",
+        ]
+        for url in urls:
+            with self.assertRaises(UnsafeProtocolError):
+                remote.fetch(url)
+
+    @with_rw_repo("HEAD")
+    def test_fetch_unsafe_url_allowed(self, rw_repo):
+        remote = rw_repo.remote("origin")
+        urls = [
+            "ext::sh -c touch% /tmp/pwn",
+            "fd::17/foo",
+        ]
+        for url in urls:
+            # The URL will be allowed into the command, but the command will
+            # fail since we don't have that protocol enabled in the Git config file.
+            with self.assertRaises(GitCommandError):
+                remote.fetch(url, allow_unsafe_protocols=True)
+
+    @with_rw_repo("HEAD")
+    def test_fetch_unsafe_options(self, rw_repo):
+        remote = rw_repo.remote("origin")
+        tmp_dir = Path(tempfile.mkdtemp())
+        tmp_file = tmp_dir / "pwn"
+        unsafe_options = [{"upload-pack": f"touch {tmp_file}"}]
+        for unsafe_option in unsafe_options:
+            with self.assertRaises(UnsafeOptionError):
+                remote.fetch(**unsafe_option)
+
+    @with_rw_repo("HEAD")
+    def test_fetch_unsafe_options_allowed(self, rw_repo):
+        remote = rw_repo.remote("origin")
+        tmp_dir = Path(tempfile.mkdtemp())
+        tmp_file = tmp_dir / "pwn"
+        unsafe_options = [{"upload-pack": f"touch {tmp_file}"}]
+        for unsafe_option in unsafe_options:
+            # The options will be allowed, but the command will fail.
+            with self.assertRaises(GitCommandError):
+                remote.fetch(**unsafe_option, allow_unsafe_options=True)
+
+    @with_rw_repo("HEAD")
+    def test_pull_unsafe_url(self, rw_repo):
+        remote = rw_repo.remote("origin")
+        urls = [
+            "ext::sh -c touch% /tmp/pwn",
+            "fd::17/foo",
+        ]
+        for url in urls:
+            with self.assertRaises(UnsafeProtocolError):
+                remote.pull(url)
+
+    @with_rw_repo("HEAD")
+    def test_pull_unsafe_url_allowed(self, rw_repo):
+        remote = rw_repo.remote("origin")
+        urls = [
+            "ext::sh -c touch% /tmp/pwn",
+            "fd::17/foo",
+        ]
+        for url in urls:
+            # The URL will be allowed into the command, but the command will
+            # fail since we don't have that protocol enabled in the Git config file.
+            with self.assertRaises(GitCommandError):
+                remote.pull(url, allow_unsafe_protocols=True)
+
+    @with_rw_repo("HEAD")
+    def test_pull_unsafe_options(self, rw_repo):
+        remote = rw_repo.remote("origin")
+        tmp_dir = Path(tempfile.mkdtemp())
+        tmp_file = tmp_dir / "pwn"
+        unsafe_options = [{"upload-pack": f"touch {tmp_file}"}]
+        for unsafe_option in unsafe_options:
+            with self.assertRaises(UnsafeOptionError):
+                remote.pull(**unsafe_option)
+
+    @with_rw_repo("HEAD")
+    def test_pull_unsafe_options_allowed(self, rw_repo):
+        remote = rw_repo.remote("origin")
+        tmp_dir = Path(tempfile.mkdtemp())
+        tmp_file = tmp_dir / "pwn"
+        unsafe_options = [{"upload-pack": f"touch {tmp_file}"}]
+        for unsafe_option in unsafe_options:
+            # The options will be allowed, but the command will fail.
+            with self.assertRaises(GitCommandError):
+                remote.pull(**unsafe_option, allow_unsafe_options=True)
+
+    @with_rw_repo("HEAD")
+    def test_push_unsafe_url(self, rw_repo):
+        remote = rw_repo.remote("origin")
+        urls = [
+            "ext::sh -c touch% /tmp/pwn",
+            "fd::17/foo",
+        ]
+        for url in urls:
+            with self.assertRaises(UnsafeProtocolError):
+                remote.push(url)
+
+    @with_rw_repo("HEAD")
+    def test_push_unsafe_url_allowed(self, rw_repo):
+        remote = rw_repo.remote("origin")
+        urls = [
+            "ext::sh -c touch% /tmp/pwn",
+            "fd::17/foo",
+        ]
+        for url in urls:
+            # The URL will be allowed into the command, but the command will
+            # fail since we don't have that protocol enabled in the Git config file.
+            with self.assertRaises(GitCommandError):
+                remote.push(url, allow_unsafe_protocols=True)
+
+    @with_rw_repo("HEAD")
+    def test_push_unsafe_options(self, rw_repo):
+        remote = rw_repo.remote("origin")
+        tmp_dir = Path(tempfile.mkdtemp())
+        tmp_file = tmp_dir / "pwn"
+        unsafe_options = [
+            {
+                "receive-pack": f"touch {tmp_file}",
+                "exec": f"touch {tmp_file}",
+            }
+        ]
+        for unsafe_option in unsafe_options:
+            with self.assertRaises(UnsafeOptionError):
+                remote.push(**unsafe_option)
+
+    @with_rw_repo("HEAD")
+    def test_push_unsafe_options_allowed(self, rw_repo):
+        remote = rw_repo.remote("origin")
+        tmp_dir = Path(tempfile.mkdtemp())
+        tmp_file = tmp_dir / "pwn"
+        unsafe_options = [
+            {
+                "receive-pack": f"touch {tmp_file}",
+                "exec": f"touch {tmp_file}",
+            }
+        ]
+        for unsafe_option in unsafe_options:
+            # The options will be allowed, but the command will fail.
+            with self.assertRaises(GitCommandError):
+                remote.push(**unsafe_option, allow_unsafe_options=True)
+
 
 class TestTimeouts(TestBase):
     @with_rw_repo("HEAD", bare=False)

test/test_submodule.py

@@ -1026,7 +1026,7 @@ def test_update_clone_multi_options_argument(self, rwdir):
         )
 
         # Act
-        sm.update(init=True, clone_multi_options=["--config core.eol=true"])
+        sm.update(init=True, clone_multi_options=["--config core.eol=true"], allow_unsafe_options=True)
 
         # Assert
         sm_config = GitConfigParser(file_or_files=osp.join(parent.git_dir, "modules", sm_name, "config"))
@@ -1070,6 +1070,7 @@ def test_add_clone_multi_options_argument(self, rwdir):
             sm_name,
             url=self._small_repo_url(),
             clone_multi_options=["--config core.eol=true"],
+            allow_unsafe_options=True,
         )
 
         # Assert