diff --git a/CHANGELOG.md b/CHANGELOG.md index b9c974e..6175b0d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,13 @@ named `git` was skipped, so that renaming or hard-linking the gitcache executable to `git` resulted in an infinite recursion. As the check now compares the candidates with the gitcache executable, a symlink pointing to the real git command is no longer skipped either. +- Bugfix: Keep the user of `ssh://` and scp-style URLs when normalizing URLs and stripping credentials. The + user of an SSH URL is part of the address, not a secret. Dropping it left mirrors created from SSH URLs + unable to authenticate on `gitcache -u` and whenever the same repository was reached through an HTTPS URL + (#159, thanks to waipeng). +- Bugfix: Set the ssh user of the mirror remote to the user of the command URL. This restores the user on + mirrors created by v1.0.31 to v1.0.34 on their next use with an SSH URL, and follows the user when the same + repository is reached with different ssh users (thanks to Bilb). ## v1.0.34 diff --git a/src/git_cache/database.py b/src/git_cache/database.py index 9cd7c95..f321d25 100644 --- a/src/git_cache/database.py +++ b/src/git_cache/database.py @@ -72,6 +72,18 @@ def add(self, url: str, path: str) -> None: } self._save() + def set_url(self, path: str, url: str) -> None: + """Set the upstream URL of an existing entry, keeping its counters. + + Args: + path (str): The path of the repository mirror. + url (str): The upstream repository URL. + """ + with portalocker.Lock(GITCACHE_DB_LOCK): + self._load() + self.database[path]["url"] = url + self._save() + def remove(self, path: str) -> None: """Remove an entry from the database. diff --git a/src/git_cache/git_mirror.py b/src/git_cache/git_mirror.py index 1eed233..e1a6df7 100644 --- a/src/git_cache/git_mirror.py +++ b/src/git_cache/git_mirror.py @@ -27,7 +27,7 @@ from .database import Database from .git_options import GitOptions from .global_settings import GITCACHE_DIR -from .helpers import rmtree, strip_credentials +from .helpers import get_user, is_ssh_url, keep_username, rmtree, strip_credentials from .invocation_log import record_cache # ----------------------------------------------------------------------------- @@ -525,24 +525,21 @@ def _clone(self, ref=None): return self._remove_credentials_from_remote() + def _set_remote_url(self, url: str) -> bool: + """Set the URL of the origin remote of the mirror.""" + command = [self.config.get("System", "RealGit"), "-C", self.git_dir, "remote", "set-url", "origin", url] + cmd_retval = simple_call_command(command) + if cmd_retval != 0: + LOG.error("Command '%s' gave return code of %d!", command, cmd_retval) + return False + return True + def _remove_credentials_from_remote(self): """Remove any credentials from the mirror remote URLs.""" safe_url = self.strip_credentials(self.url) if self.url != safe_url: LOG.info("Removing credentials from the mirror remote URLs.") - command = [ - self.config.get("System", "RealGit"), - "-C", - self.git_dir, - "remote", - "set-url", - "origin", - safe_url, - ] - cmd_retval = simple_call_command(command) - if cmd_retval != 0: - LOG.error("Command '%s' gave return code of %d!", command, cmd_retval) - return False + return self._set_remote_url(safe_url) return True @@ -551,22 +548,41 @@ def _add_credentials_to_remote(self): safe_url = self.strip_credentials(self.url) if self.url != safe_url: LOG.info("Temporarily restoring credentials on the mirror remote URLs.") - command = [ - self.config.get("System", "RealGit"), - "-C", - self.git_dir, - "remote", - "set-url", - "origin", - self.url, - ] - cmd_retval = simple_call_command(command) - if cmd_retval != 0: - LOG.error("Command '%s' gave return code of %d!", command, cmd_retval) + return self._set_remote_url(self.url) + + if self._remote_user_differs(): + LOG.info("Setting the user of the mirror remote URL to the one of the mirror URL.") + if not self._set_remote_url(self.url): return False + self.database.set_url(self.path, self.normalize_url(self.url)) return True + def _remote_user_differs(self) -> bool: + """Check whether the mirror remote is an ssh URL of the same repository with another user. + + The ssh user selects the account the mirror authenticates with, so the + remote follows the user of the mirror URL. This also repairs mirrors of + gitcache v1.0.31 to v1.0.34, which stripped the user from the remote. + + Return: + Returns True if the mirror URL is an ssh URL with a user and the + remote is an ssh URL of the same repository with a different user + or without one. + """ + user = get_user(self.url) + if not user or not is_ssh_url(self.url): + return False + + command = [self.config.get("System", "RealGit"), "-C", self.git_dir, "remote", "get-url", "origin"] + retval, remote_url = getstatusoutput(command) + if retval != 0: + return False + remote_url = remote_url.strip() + if not is_ssh_url(remote_url) or get_user(remote_url) == user: + return False + return self.get_mirror_path(remote_url) == self.path + def _update(self, ref=None, handle_gc_error=True): """Update the mirror. @@ -756,7 +772,8 @@ def normalize_url(url: str) -> str: path = path[:-1] if path.endswith(".git"): path = path[:-4] - return f"{match.group(1)}://{match.group(3)}{match.group(4) or ''}/{path}" + creds = keep_username(match.group(2)) if match.group(1).lower() == "ssh" else "" + return f"{match.group(1)}://{creds}{match.group(3)}{match.group(4) or ''}/{path}" if match := RE_URL_WITHOUT_PROTO.match(url): path = posixpath.normpath(match.group(3)) @@ -766,7 +783,9 @@ def normalize_url(url: str) -> str: path = path[:-1] if path.endswith(".git"): path = path[:-4] - return f"{match.group(2)}:{path}" + # SCP-style URLs (user@host:path) have no syntax for a password, so the + # username portion can be kept as-is without needing to strip anything. + return f"{match.group(1) or ''}{match.group(2)}:{path}" return url diff --git a/src/git_cache/helpers.py b/src/git_cache/helpers.py index 3a8a9d4..f0ae873 100644 --- a/src/git_cache/helpers.py +++ b/src/git_cache/helpers.py @@ -18,6 +18,7 @@ import os import re import shutil +from typing import Optional # Pattern to match ssh, git, http[s] and ftp[s]: # [user@] [:port] @@ -120,12 +121,87 @@ def resetperms(path): shutil.rmtree(name, onerror=onerror) +def keep_username(creds: Optional[str]) -> str: + """Reduce a credentials part of an URL to the plain 'user@' prefix. + + Args: + creds (str): The credentials part of an URL including the trailing '@', + e.g. 'user@' or 'user:password@'. May be None or empty. + + Return: + Returns the 'user@' prefix without any password, or an empty string + if no credentials were given. + """ + if not creds: + return "" + return creds[:-1].split(":", 1)[0] + "@" + + +def get_user(url: str) -> Optional[str]: + """Get the user of the specified url. + + Args: + url (str): The URL of the repository. + + Return: + Returns the user without any password or None if the URL has no user. + """ + if _RE_URL_WITH_FILE.match(url): + return None + + if match := _RE_URL_WITH_PROTO.match(url): + creds = match.group(2) + elif match := _RE_URL_WITHOUT_PROTO.match(url): + creds = match.group(1) + else: + return None + + return keep_username(creds)[:-1] or None + + +def is_ssh_url(url: str) -> bool: + """Check whether the specified url is an ssh:// or scp-style URL. + + Args: + url (str): The URL of the repository. + + Return: + Returns True if the URL uses ssh. + """ + if _RE_URL_WITH_FILE.match(url): + return False + + if match := _RE_URL_WITH_PROTO.match(url): + return match.group(1).lower() == "ssh" + + return _RE_URL_WITHOUT_PROTO.match(url) is not None + + +def _strip_password(creds: str, mask: bool = False) -> str: + """Remove the password from a credentials part of an SSH URL. + + The username is required for SSH authentication and therefore kept. + + Args: + creds (str): The credentials part of an URL including the trailing '@', + e.g. 'user@' or 'user:password@'. + mask (bool): If set to True the password is replaced with [MASKED]. + + Return: + Returns 'user@' resp. 'user:[MASKED]@' if masking a password. + """ + has_password = ":" in creds[:-1] + if has_password and mask: + return keep_username(creds)[:-1] + ":[MASKED]@" + return keep_username(creds) + + def strip_credentials(url: str, mask: bool = False) -> str: """Remove any credentials from the specified url. Args: url (str): The URL of the repository. - mask (bool): If set to True the credentials are replaced with [MASKED]. + mask (bool): If set to True the removed parts are replaced with [MASKED]. Return: Returns the URL without credentials. @@ -134,12 +210,17 @@ def strip_credentials(url: str, mask: bool = False) -> str: return url if match := _RE_URL_WITH_PROTO.match(url): - masked_creds = "[MASKED]@" if match.group(2) and mask else "" - return f"{match.group(1)}://{masked_creds}{match.group(3)}{match.group(4) or ''}/{match.group(5)}" + proto, creds, host, port, path = match.groups() + if not creds: + return url + if proto.lower() == "ssh": + return f"{proto}://{_strip_password(creds, mask)}{host}{port or ''}/{path}" + return f"{proto}://{'[MASKED]@' if mask else ''}{host}{port or ''}/{path}" if match := _RE_URL_WITHOUT_PROTO.match(url): - masked_creds = "[MASKED]@" if match.group(1) and mask else "" - return f"{masked_creds}{match.group(2)}:{match.group(3)}" + # For SCP-style URLs (user@host:path), keep the username as-is. SCP syntax + # has no place for a password, so there is nothing to strip or mask here. + return url return url diff --git a/test/functional_tests/test_10_clone.py b/test/functional_tests/test_10_clone.py index b74e006..0d89ff3 100644 --- a/test/functional_tests/test_10_clone.py +++ b/test/functional_tests/test_10_clone.py @@ -247,7 +247,7 @@ def test_clone_recurse_submodules_remote_submodules(gitcache_ifc: GitcacheIfc): def test_clone_via_ssh(gitcache_ifc: GitcacheIfc, remote_url: str): """Test clone via ssh URLs.""" gitcache_ifc.run_ok(["git", "-C", gitcache_ifc.workspace.workspace_path, "clone", remote_url]) - db_url = remote_url[:-4].replace("git@", "") + db_url = remote_url[:-4] assert "github.com/seeraven/gitcache" in gitcache_ifc.db_field("mirror-dir", db_url) @@ -262,7 +262,7 @@ def test_clone_via_ssh(gitcache_ifc: GitcacheIfc, remote_url: str): def test_clone_via_ssh_and_port(gitcache_ifc: GitcacheIfc, remote_url: str): """Test clone via ssh URLs.""" gitcache_ifc.run_ok(["git", "-C", gitcache_ifc.workspace.workspace_path, "clone", remote_url]) - db_url = remote_url[:-4].replace("git@", "") + db_url = remote_url[:-4] assert "github.com_22/seeraven/gitcache" in gitcache_ifc.db_field("mirror-dir", db_url) diff --git a/test/functional_tests/test_12_submodule_update.py b/test/functional_tests/test_12_submodule_update.py index e36e988..04d5835 100644 --- a/test/functional_tests/test_12_submodule_update.py +++ b/test/functional_tests/test_12_submodule_update.py @@ -397,7 +397,7 @@ def test_relative_submodule_ssh_specs( gitmodules.write("\tpath = gitcache\n") gitmodules.write(f"\turl = {submodule_url}\n") gitcache_ifc.run_ok(["git", "-C", checkout, "submodule", "update", "--init"]) - assert 1 == gitcache_ifc.db_field("clones", final_submodule_url.replace("git@", "")) + assert 1 == gitcache_ifc.db_field("clones", final_submodule_url) def test_submodule_on_orphaned_commit(gitcache_ifc: GitcacheIfc): diff --git a/test/functional_tests/test_13_fetch.py b/test/functional_tests/test_13_fetch.py index 40c9dad..1617894 100644 --- a/test/functional_tests/test_13_fetch.py +++ b/test/functional_tests/test_13_fetch.py @@ -113,7 +113,7 @@ def test_fetch_in_empty_repo_on_workhorse(gitcache_ifc: GitcacheIfc, remote_url: checkout = os.path.join(gitcache_ifc.workspace.workspace_path, "gitcache") gitcache_ifc.run_ok(["git", "init", checkout]) gitcache_ifc.run_ok(["git", "-C", checkout, "fetch", remote_url]) - db_url = remote_url[:-4].replace("git@", "") + db_url = remote_url[:-4] assert 0 == gitcache_ifc.db_field("mirror-updates", db_url) assert 0 == gitcache_ifc.db_field("clones", db_url) assert 1 == gitcache_ifc.db_field("updates", db_url) diff --git a/test/functional_tests/test_14_ls_remote.py b/test/functional_tests/test_14_ls_remote.py index 7fa6990..a24ab87 100644 --- a/test/functional_tests/test_14_ls_remote.py +++ b/test/functional_tests/test_14_ls_remote.py @@ -84,7 +84,7 @@ def test_ls_remote_without_initial_clone(gitcache_ifc: GitcacheIfc, remote_url: def test_ls_remote_without_initial_clone_on_workhorse(gitcache_ifc: GitcacheIfc, remote_url: str, mirror_dir: str): """Test the 'git ls-remote' command without an initial checkout.""" gitcache_ifc.run_ok(["git", "ls-remote", remote_url]) - db_url = remote_url[:-4].replace("git@", "") + db_url = remote_url[:-4] assert 0 == gitcache_ifc.db_field("mirror-updates", db_url) assert 0 == gitcache_ifc.db_field("clones", db_url) assert 0 == gitcache_ifc.db_field("updates", db_url) diff --git a/test/functional_tests/test_15_lfs.py b/test/functional_tests/test_15_lfs.py index be99989..347d540 100644 --- a/test/functional_tests/test_15_lfs.py +++ b/test/functional_tests/test_15_lfs.py @@ -38,7 +38,7 @@ def test_lfs_fetch(gitcache_ifc: GitcacheIfc, remote_url: str): # Initial clone checkout = os.path.join(gitcache_ifc.workspace.workspace_path, "lfs-example") gitcache_ifc.run_ok(["git", "clone", remote_url, checkout]) - db_url = remote_url.replace("git@", "") + db_url = remote_url assert 0 == gitcache_ifc.db_field("mirror-updates", db_url) assert 1 == gitcache_ifc.db_field("lfs-updates", db_url) @@ -81,7 +81,7 @@ def test_lfs_pull(gitcache_ifc: GitcacheIfc, remote_url: str): """Test not caching ls-remotes from local filesystem.""" checkout = os.path.join(gitcache_ifc.workspace.workspace_path, "lfs-example") gitcache_ifc.run_ok(["git", "clone", remote_url, checkout]) - db_url = remote_url.replace("git@", "") + db_url = remote_url assert 0 == gitcache_ifc.db_field("mirror-updates", db_url) assert 1 == gitcache_ifc.db_field("lfs-updates", db_url) diff --git a/test/unittests/test_git_cache_database.py b/test/unittests/test_git_cache_database.py index 228061c..a16c1d8 100644 --- a/test/unittests/test_git_cache_database.py +++ b/test/unittests/test_git_cache_database.py @@ -91,6 +91,12 @@ def test_database(self): self.assertEqual(1, database.get(repo_abs_path)["clones"]) self.assertEqual(1, database.get(repo_abs_path)["updates"]) + database.set_url(repo_abs_path, "http://dummy/other") + self.assertEqual("http://dummy/other", database.get(repo_abs_path)["url"]) + self.assertEqual(1, database.get(repo_abs_path)["mirror-updates"]) + self.assertEqual(1, database.get(repo_abs_path)["clones"]) + self.assertEqual(1, database.get(repo_abs_path)["updates"]) + database.clear_counters(repo_abs_path) self.assertEqual(0, database.get(repo_abs_path)["mirror-updates"]) self.assertEqual(0, database.get(repo_abs_path)["clones"]) diff --git a/test/unittests/test_git_cache_git_mirror_add_credentials_to_remote.py b/test/unittests/test_git_cache_git_mirror_add_credentials_to_remote.py new file mode 100644 index 0000000..de7e3d0 --- /dev/null +++ b/test/unittests/test_git_cache_git_mirror_add_credentials_to_remote.py @@ -0,0 +1,130 @@ +# +# Copyright (c) 2026 by Audric Ackermann +# All rights reserved. +# This file is part of gitcache (https://github.com/seeraven/gitcache) +# and is released under the "BSD 3-Clause License". Please see the LICENSE file +# that is included as part of this package. +# +"""Unit tests of the git_cache.git_mirror module testing _add_credentials_to_remote().""" + +# ----------------------------------------------------------------------------- +# Module Import +# ----------------------------------------------------------------------------- +from unittest import TestCase, mock + +from git_cache.git_mirror import GitMirror + + +# ----------------------------------------------------------------------------- +# Test Class +# ----------------------------------------------------------------------------- +class GitCacheAddCredentialsToRemoteTest(TestCase): + """Test the :func:`git_cache.git_mirror.GitMirror._add_credentials_to_remote` function.""" + + def _mirror(self, url: str) -> GitMirror: + mirror = GitMirror.__new__(GitMirror) + mirror.url = url + mirror.path = GitMirror.get_mirror_path(url) + mirror.git_dir = f"{mirror.path}/git" + mirror.config = mock.MagicMock() + mirror.config.get.return_value = "git" + mirror.database = mock.MagicMock() + return mirror + + def _run(self, url: str, remote_url: str): + """Run _add_credentials_to_remote() with the given remote. + + Return: + Returns the URLs passed to 'git remote set-url' and the URLs written to the database. + """ + mirror = self._mirror(url) + with ( + mock.patch("git_cache.git_mirror.getstatusoutput", return_value=(0, remote_url + "\n")), + mock.patch("git_cache.git_mirror.simple_call_command", return_value=0) as set_url, + ): + self.assertTrue(mirror._add_credentials_to_remote()) # pylint: disable=protected-access + remote_urls = [call.args[0][-1] for call in set_url.call_args_list] + database_urls = [call.args[1] for call in mirror.database.set_url.call_args_list] + return remote_urls, database_urls + + def test_remote_lost_user_is_restored(self): + """_add_credentials_to_remote(): Restore the user on a remote stripped by v1.0.31 to v1.0.34.""" + self.assertEqual( + self._run("git@github.com:org/repo.git", "github.com:org/repo.git"), + (["git@github.com:org/repo.git"], ["git@github.com:org/repo"]), + ) + self.assertEqual( + self._run("git@github.com:org/repo", "github.com:org/repo.git"), + (["git@github.com:org/repo"], ["git@github.com:org/repo"]), + ) + self.assertEqual( + self._run("ssh://git@github.com/org/repo.git", "ssh://github.com/org/repo.git"), + (["ssh://git@github.com/org/repo.git"], ["ssh://git@github.com/org/repo"]), + ) + self.assertEqual( + self._run("git@github.com:org/repo.git", "ssh://github.com/org/repo.git"), + (["git@github.com:org/repo.git"], ["git@github.com:org/repo"]), + ) + + def test_remote_with_other_user_follows_mirror_url(self): + """_add_credentials_to_remote(): Set the remote to the mirror URL when the ssh user differs.""" + self.assertEqual( + self._run("bob@host:org/repo.git", "alice@host:org/repo.git"), + (["bob@host:org/repo.git"], ["bob@host:org/repo"]), + ) + self.assertEqual( + self._run("ssh://bob@host/org/repo.git", "ssh://alice@host/org/repo.git"), + (["ssh://bob@host/org/repo.git"], ["ssh://bob@host/org/repo"]), + ) + + def test_remote_with_same_user_is_left_alone(self): + """_add_credentials_to_remote(): Keep a remote that already has the user.""" + self.assertEqual(self._run("git@github.com:org/repo.git", "git@github.com:org/repo.git"), ([], [])) + self.assertEqual(self._run("git@github.com:org/repo.git", "https://github.com/org/repo.git"), ([], [])) + + def test_remote_of_other_repository_is_left_alone(self): + """_add_credentials_to_remote(): Never point the mirror at another repository.""" + self.assertEqual(self._run("git@github.com:org/repo.git", "github.com:org/other.git"), ([], [])) + self.assertEqual(self._run("git@github.com:org/repo.git", "alice@github.com:org/other.git"), ([], [])) + + def test_failed_set_url_keeps_database_url(self): + """_add_credentials_to_remote(): Leave the database alone when 'git remote set-url' fails.""" + mirror = self._mirror("git@github.com:org/repo.git") + with ( + mock.patch("git_cache.git_mirror.getstatusoutput", return_value=(0, "github.com:org/repo.git\n")), + mock.patch("git_cache.git_mirror.simple_call_command", return_value=1), + ): + self.assertFalse(mirror._add_credentials_to_remote()) # pylint: disable=protected-access + mirror.database.set_url.assert_not_called() + + def test_url_without_user_does_not_touch_remote(self): + """_add_credentials_to_remote(): A URL without user has nothing to restore.""" + mirror = self._mirror("github.com:org/repo.git") + with ( + mock.patch("git_cache.git_mirror.getstatusoutput") as get_url, + mock.patch("git_cache.git_mirror.simple_call_command") as set_url, + ): + self.assertTrue(mirror._add_credentials_to_remote()) # pylint: disable=protected-access + get_url.assert_not_called() + set_url.assert_not_called() + mirror.database.set_url.assert_not_called() + + def test_non_ssh_url_never_rewrites_remote(self): + """_remote_user_differs(): Only an ssh mirror URL may set the user of the remote.""" + for url in ("https://token@github.com/org/repo.git", "git://user@github.com/org/repo.git"): + mirror = self._mirror(url) + with mock.patch("git_cache.git_mirror.getstatusoutput") as get_url: + self.assertFalse(mirror._remote_user_differs(), url) # pylint: disable=protected-access + get_url.assert_not_called() + + def test_http_credentials_are_restored(self): + """_add_credentials_to_remote(): Restore http credentials without reading the remote.""" + mirror = self._mirror("https://user:secret@github.com/org/repo.git") + with ( + mock.patch("git_cache.git_mirror.getstatusoutput") as get_url, + mock.patch("git_cache.git_mirror.simple_call_command", return_value=0) as set_url, + ): + self.assertTrue(mirror._add_credentials_to_remote()) # pylint: disable=protected-access + get_url.assert_not_called() + self.assertEqual(set_url.call_args.args[0][-1], "https://user:secret@github.com/org/repo.git") + mirror.database.set_url.assert_not_called() diff --git a/test/unittests/test_git_cache_git_mirror_normalize_url.py b/test/unittests/test_git_cache_git_mirror_normalize_url.py new file mode 100644 index 0000000..24d34df --- /dev/null +++ b/test/unittests/test_git_cache_git_mirror_normalize_url.py @@ -0,0 +1,68 @@ +# +# Copyright (c) 2026 by Audric Ackermann +# All rights reserved. +# This file is part of gitcache (https://github.com/seeraven/gitcache) +# and is released under the "BSD 3-Clause License". Please see the LICENSE file +# that is included as part of this package. +# +"""Unit tests of the git_cache.git_mirror module testing normalize_url().""" + +# ----------------------------------------------------------------------------- +# Module Import +# ----------------------------------------------------------------------------- +from unittest import TestCase + +from git_cache.git_mirror import GitMirror + + +# ----------------------------------------------------------------------------- +# Test Class +# ----------------------------------------------------------------------------- +class GitCacheNormalizeUrlTest(TestCase): + """Test the :func:`git_cache.git_mirror.GitMirror.normalize_url` function.""" + + def _assert_normalized(self, url: str, expected: str): + self.assertEqual(GitMirror.normalize_url(url), expected, url) + self.assertEqual(GitMirror.normalize_url(expected), expected, expected) + + def test_ssh_user_is_kept(self): + """git_cache.git_mirror.GitMirror.normalize_url(): Keep the user of ssh:// and scp-style URLs.""" + self._assert_normalized("ssh://git@github.com/org/repo.git", "ssh://git@github.com/org/repo") + self._assert_normalized("ssh://git@github.com:2222/org/repo.git", "ssh://git@github.com:2222/org/repo") + self._assert_normalized("ssh://github.com/org/repo.git", "ssh://github.com/org/repo") + self._assert_normalized("git@github.com:org/repo.git", "git@github.com:org/repo") + self._assert_normalized("github.com:org/repo.git", "github.com:org/repo") + + def test_ssh_password_is_dropped(self): + """git_cache.git_mirror.GitMirror.normalize_url(): Drop the password of ssh:// URLs but keep the user.""" + self._assert_normalized("ssh://user:secret@github.com/org/repo.git", "ssh://user@github.com/org/repo") + + def test_other_credentials_are_dropped(self): + """git_cache.git_mirror.GitMirror.normalize_url(): Drop user and password of http(s), ftp(s) and git URLs.""" + self._assert_normalized("https://user:secret@github.com/org/repo.git", "https://github.com/org/repo") + self._assert_normalized("https://token@github.com/org/repo", "https://github.com/org/repo") + self._assert_normalized("http://user:secret@localhost:7698/org/repo.git", "http://localhost:7698/org/repo") + self._assert_normalized("ftps://user:secret@example.com/repo", "ftps://example.com/repo") + self._assert_normalized("git://user@github.com/org/repo.git", "git://github.com/org/repo") + + def test_path_is_normalized(self): + """git_cache.git_mirror.GitMirror.normalize_url(): Strip trailing slashes, '.git' and relative parts.""" + for user in ["", "git@"]: + self._assert_normalized(f"ssh://{user}github.com/org/repo/", f"ssh://{user}github.com/org/repo") + self._assert_normalized(f"ssh://{user}github.com/org/repo.git/", f"ssh://{user}github.com/org/repo") + self._assert_normalized(f"ssh://{user}github.com/org/../org/repo.git", f"ssh://{user}github.com/org/repo") + self._assert_normalized(f"ssh://{user}github.com/../org/repo.git", f"ssh://{user}github.com/org/repo") + self._assert_normalized(f"{user}github.com:org/repo/", f"{user}github.com:org/repo") + self._assert_normalized(f"{user}github.com:org/repo.git/", f"{user}github.com:org/repo") + self._assert_normalized(f"{user}github.com:org/../org/repo.git", f"{user}github.com:org/repo") + self._assert_normalized(f"{user}github.com:../org/repo.git", f"{user}github.com:org/repo") + + def test_local_urls_are_untouched(self): + """git_cache.git_mirror.GitMirror.normalize_url(): Keep file:// URLs and local paths.""" + for url in ("file:///tmp/repo.git", "file:///tmp/a/../repo.git/", "/tmp/repo.git", "../repo"): + self._assert_normalized(url, url) + + +# ----------------------------------------------------------------------------- +# EOF +# ----------------------------------------------------------------------------- diff --git a/test/unittests/test_git_cache_helpers_get_user.py b/test/unittests/test_git_cache_helpers_get_user.py new file mode 100644 index 0000000..ad7f50e --- /dev/null +++ b/test/unittests/test_git_cache_helpers_get_user.py @@ -0,0 +1,42 @@ +# +# Copyright (c) 2026 by Audric Ackermann +# All rights reserved. +# This file is part of gitcache (https://github.com/seeraven/gitcache) +# and is released under the "BSD 3-Clause License". Please see the LICENSE file +# that is included as part of this package. +# +"""Unit tests of the git_cache.helpers module testing get_user().""" + +# ----------------------------------------------------------------------------- +# Module Import +# ----------------------------------------------------------------------------- +from unittest import TestCase + +from git_cache.helpers import get_user + + +# ----------------------------------------------------------------------------- +# Test Class +# ----------------------------------------------------------------------------- +class GitCacheGetUserTest(TestCase): + """Test the :func:`git_cache.helpers.get_user` function.""" + + def test_urls_with_user(self): + """git_cache.helpers.get_user(): Return the user of URLs carrying one.""" + self.assertEqual(get_user("git@github.com:org/repo.git"), "git") + self.assertEqual(get_user("ssh://git@github.com/org/repo.git"), "git") + self.assertEqual(get_user("ssh://git@github.com:2222/org/repo.git"), "git") + self.assertEqual(get_user("ssh://user:secret@github.com/org/repo.git"), "user") + self.assertEqual(get_user("https://user:secret@github.com/org/repo.git"), "user") + self.assertEqual(get_user("https://token@github.com/org/repo"), "token") + + def test_urls_without_user(self): + """git_cache.helpers.get_user(): Return None for URLs without a user.""" + for url in ( + "github.com:org/repo.git", + "ssh://github.com/org/repo.git", + "https://github.com/org/repo.git", + "file:///tmp/repo.git", + "/tmp/repo.git", + ): + self.assertIsNone(get_user(url), url) diff --git a/test/unittests/test_git_cache_helpers_is_ssh_url.py b/test/unittests/test_git_cache_helpers_is_ssh_url.py new file mode 100644 index 0000000..b2840c7 --- /dev/null +++ b/test/unittests/test_git_cache_helpers_is_ssh_url.py @@ -0,0 +1,45 @@ +# +# Copyright (c) 2026 by Audric Ackermann +# All rights reserved. +# This file is part of gitcache (https://github.com/seeraven/gitcache) +# and is released under the "BSD 3-Clause License". Please see the LICENSE file +# that is included as part of this package. +# +"""Unit tests of the git_cache.helpers module testing is_ssh_url().""" + +# ----------------------------------------------------------------------------- +# Module Import +# ----------------------------------------------------------------------------- +from unittest import TestCase + +from git_cache.helpers import is_ssh_url + + +# ----------------------------------------------------------------------------- +# Test Class +# ----------------------------------------------------------------------------- +class GitCacheIsSshUrlTest(TestCase): + """Test the :func:`git_cache.helpers.is_ssh_url` function.""" + + def test_ssh_urls(self): + """git_cache.helpers.is_ssh_url(): Detect ssh:// and scp-style URLs.""" + for url in ( + "git@github.com:org/repo.git", + "github.com:org/repo.git", + "ssh://git@github.com/org/repo.git", + "ssh://github.com:2222/org/repo.git", + ): + self.assertTrue(is_ssh_url(url), url) + + def test_other_urls(self): + """git_cache.helpers.is_ssh_url(): Reject http(s), ftp(s), git://, file:// and local paths.""" + for url in ( + "https://github.com/org/repo.git", + "https://user:secret@github.com/org/repo.git", + "ftp://example.com/repo", + "git://github.com/org/repo.git", + "file:///tmp/repo.git", + "/tmp/repo.git", + "../repo", + ): + self.assertFalse(is_ssh_url(url), url) diff --git a/test/unittests/test_git_cache_helpers_strip_credentials.py b/test/unittests/test_git_cache_helpers_strip_credentials.py new file mode 100644 index 0000000..1328566 --- /dev/null +++ b/test/unittests/test_git_cache_helpers_strip_credentials.py @@ -0,0 +1,62 @@ +# +# Copyright (c) 2026 by Audric Ackermann +# All rights reserved. +# This file is part of gitcache (https://github.com/seeraven/gitcache) +# and is released under the "BSD 3-Clause License". Please see the LICENSE file +# that is included as part of this package. +# +"""Unit tests of the git_cache.helpers module testing strip_credentials().""" + +# ----------------------------------------------------------------------------- +# Module Import +# ----------------------------------------------------------------------------- +from unittest import TestCase + +from git_cache.helpers import strip_credentials + + +# ----------------------------------------------------------------------------- +# Test Class +# ----------------------------------------------------------------------------- +class GitCacheStripCredentialsTest(TestCase): + """Test the :func:`git_cache.helpers.strip_credentials` function.""" + + def test_http_credentials_are_stripped(self): + """git_cache.helpers.strip_credentials(): Strip user and password of http(s)/ftp(s) URLs.""" + self.assertEqual( + strip_credentials("https://user:secret@github.com/org/repo.git"), "https://github.com/org/repo.git" + ) + self.assertEqual(strip_credentials("https://token@github.com/org/repo"), "https://github.com/org/repo") + self.assertEqual( + strip_credentials("http://user:secret@localhost:7698/org/repo"), "http://localhost:7698/org/repo" + ) + self.assertEqual(strip_credentials("ftps://user:secret@example.com/repo"), "ftps://example.com/repo") + self.assertEqual(strip_credentials("https://github.com/org/repo.git"), "https://github.com/org/repo.git") + + def test_http_credentials_are_masked(self): + """git_cache.helpers.strip_credentials(): Mask user and password of http(s) URLs.""" + self.assertEqual( + strip_credentials("https://user:secret@github.com/org/repo.git", mask=True), + "https://[MASKED]@github.com/org/repo.git", + ) + self.assertEqual( + strip_credentials("https://github.com/org/repo.git", mask=True), "https://github.com/org/repo.git" + ) + + def test_ssh_user_is_kept(self): + """git_cache.helpers.strip_credentials(): Keep the user of ssh:// and scp-style URLs.""" + for url in ( + "ssh://git@github.com/org/repo.git", + "ssh://git@github.com:2222/org/repo.git", + "git@github.com:org/repo.git", + "git@github.com:org/repo", + "github.com:org/repo.git", + ): + self.assertEqual(strip_credentials(url), url) + self.assertEqual(strip_credentials(url, mask=True), url) + + def test_local_urls_are_untouched(self): + """git_cache.helpers.strip_credentials(): Keep file:// URLs and local paths.""" + for url in ("file:///tmp/repo.git", "/tmp/repo.git", "../repo"): + self.assertEqual(strip_credentials(url), url) + self.assertEqual(strip_credentials(url, mask=True), url)