bugfix: Keep the user of ssh URLs and restore it on mirrors stripped by v1.0.31 to v1.0.34

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
 

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.
 

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
 

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]:
 #                                  <proto>      [user@]  <host>  [:port]   <path>
@@ -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
 

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)
 
 

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):

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)

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)

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)
 

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"])