blog entry

scripts/git_repo.py

@@ -23,6 +23,10 @@ class GitRepositoryError(Exception):
     """An expected Git problem that can be displayed directly to a user."""
 
 
+class GitAuthenticationError(GitRepositoryError):
+    """Git could reach the remote but its HTTPS credential was rejected."""
+
+
 @dataclass(frozen=True)
 class RepositoryState:
     root: Path
@@ -100,15 +104,25 @@ class GitRepository:
             ) from exc
         return cls(root)
 
-    def refresh(self) -> RepositoryState:
+    def refresh(self, *, allow_prompt: bool = False) -> RepositoryState:
         """Read local state, safely fetch origin metadata, and compare commits."""
         before_fetch = self._read_state(compare_remote=False)
         if before_fetch.problem:
             return before_fetch
         result = self._git(
-            "fetch", "--no-tags", "--prune", "origin", check=False, timeout=60
+            "fetch",
+            "--no-tags",
+            "--prune",
+            "origin",
+            check=False,
+            timeout=60,
+            allow_prompt=allow_prompt,
         )
         if result.returncode != 0:
+            if _is_authentication_failure(result):
+                raise GitAuthenticationError(
+                    f"Authentication failed: {self._safe_detail(result)}"
+                )
             return replace(
                 before_fetch,
                 sync_state=UNAVAILABLE,
@@ -153,6 +167,7 @@ class GitRepository:
             f"HEAD:refs/heads/{current.upstream_branch}",
             check=False,
             timeout=60,
+            allow_prompt=self._origin_is_https(),
         )
         if result.returncode != 0:
             refreshed = self.refresh()
@@ -161,9 +176,45 @@ class GitRepository:
                     "Push failed because the remote changed and the histories now "
                     "diverge. Resolve this manually."
                 )
-            raise GitRepositoryError(f"Push failed: {self._safe_detail(result)}")
+            detail = self._safe_detail(result)
+            if _is_authentication_failure(result):
+                raise GitAuthenticationError(f"Push failed: {detail}")
+            raise GitRepositoryError(f"Push failed: {detail}")
         return self.refresh()
 
+    def clear_https_credential(self) -> str:
+        """Reject only the credential associated with this repository's HTTPS host."""
+        remote_result = self._git("remote", "get-url", "origin", check=False)
+        if remote_result.returncode != 0:
+            raise GitRepositoryError("Cannot determine the origin URL for sign-in.")
+        parsed = urlsplit(remote_result.stdout.strip())
+        if parsed.scheme != "https" or not parsed.hostname:
+            raise GitRepositoryError(
+                "Automatic sign-in recovery is available only for an HTTPS origin."
+            )
+        host = parsed.hostname
+        if parsed.port:
+            host = f"{host}:{parsed.port}"
+        credential_input = f"protocol=https\nhost={host}\n\n"
+        result = _run_git(
+            self.root,
+            "credential",
+            "reject",
+            check=False,
+            input_text=credential_input,
+        )
+        if result.returncode != 0:
+            raise GitRepositoryError(
+                f"Could not clear the rejected credential: {self._safe_detail(result)}"
+            )
+        return host
+
+    def _origin_is_https(self) -> bool:
+        result = self._git("remote", "get-url", "origin", check=False)
+        if result.returncode != 0:
+            return False
+        return urlsplit(result.stdout.strip()).scheme == "https"
+
     def publishing_changes(self) -> str:
         """Return every path that a publish operation would stage."""
         result = self._git(
@@ -433,8 +484,15 @@ class GitRepository:
         *args: str,
         check: bool,
         timeout: int = 15,
+        allow_prompt: bool = False,
     ) -> subprocess.CompletedProcess[str]:
-        result = _run_git(self.root, *args, check=False, timeout=timeout)
+        result = _run_git(
+            self.root,
+            *args,
+            check=False,
+            timeout=timeout,
+            allow_prompt=allow_prompt,
+        )
         if check and result.returncode != 0:
             raise GitRepositoryError(f"Git failed: {self._safe_detail(result)}")
         return result
@@ -448,6 +506,8 @@ def _run_git(
     *args: str,
     check: bool,
     timeout: int = 15,
+    input_text: str | None = None,
+    allow_prompt: bool = False,
 ) -> subprocess.CompletedProcess[str]:
     executable, bundled = _git_executable(cwd)
     command = [str(executable)]
@@ -462,8 +522,13 @@ def _run_git(
         "errors": "replace",
         "check": False,
         "timeout": timeout,
-        "env": {**os.environ, "GIT_TERMINAL_PROMPT": "0"},
+        "env": {
+            **os.environ,
+            "GIT_TERMINAL_PROMPT": "1" if allow_prompt else "0",
+        },
     }
+    if input_text is not None:
+        options["input"] = input_text
     if os.name == "nt":
         options["creationflags"] = subprocess.CREATE_NO_WINDOW
     try:
@@ -498,6 +563,20 @@ def _error_detail(result: subprocess.CompletedProcess[str]) -> str:
     )
 
 
+def _is_authentication_failure(result: subprocess.CompletedProcess[str]) -> bool:
+    detail = f"{result.stderr}\n{result.stdout}".casefold()
+    return any(
+        marker in detail
+        for marker in (
+            "authentication failed",
+            "failed to authenticate",
+            "invalid username or password",
+            "could not read username",
+            "terminal prompts disabled",
+        )
+    )
+
+
 def _display_url(value: str) -> str:
     """Remove HTTP credentials before a remote URL reaches the GUI."""
     parsed = urlsplit(value)