github_auth.py

"""One shared, bounded GitHub browser login for background Git operations."""
import os
from pathlib import Path
import re
import shutil
import signal
import subprocess
import threading
import time
from urllib.parse import urlsplit

from network_settings import git_environment

LOGIN_TIMEOUT = 180


def needs_github_login(url, error):
    parsed = urlsplit(url)
    return (parsed.scheme == "https" and parsed.hostname == "github.com"
            and parsed.port in {None, 443}
            and bool(re.search(r"terminal prompts disabled|interactivity has been disabled|"
                               r"could not read (?:Username|Password)|unable to get password|"
                               r"authentication failed|invalid username or (?:password|token)", error, re.I)))


def browser_login(settings, timeout=LOGIN_TIMEOUT):
    env = git_environment(settings, "https://github.com")
    # Keep ordinary Git commands noninteractive; explicitly allow this browser flow.
    env.pop("GIT_TERMINAL_PROMPT", None)
    env.update(GCM_INTERACTIVE="always", GCM_GUI_PROMPT="true", LC_ALL="C")
    command = [shutil.which("git") or "git", "-c", "credential.interactive=always",
               "credential-manager", "github", "login", "--url", "https://github.com", "--browser", "--force"]
    proc = subprocess.Popen(command, cwd=Path(__file__).resolve().parent, env=env,
                            stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                            creationflags=subprocess.CREATE_NO_WINDOW if os.name == "nt" else 0,
                            start_new_session=os.name != "nt")
    try:
        _, error = proc.communicate(timeout=timeout)
    except subprocess.TimeoutExpired:
        if os.name == "nt":
            subprocess.run(["taskkill", "/PID", str(proc.pid), "/T", "/F"],
                           capture_output=True, creationflags=subprocess.CREATE_NO_WINDOW)
        else:
            try:
                os.killpg(proc.pid, signal.SIGKILL)
            except ProcessLookupError:
                pass
        if proc.poll() is None:
            proc.kill()
        proc.communicate()
        raise ValueError("GitHub 登录等待超时（3 分钟）。请重新获取分支或继续任务，并在浏览器中完成登录。") from None
    if proc.returncode:
        # Never forward OAuth output, callback addresses, or codes into task logs.
        if b"not a git command" in error.lower():
            raise ValueError("未安装 Git Credential Manager，请安装包含该组件的 Git for Windows 后重试。")
        raise ValueError("GitHub 浏览器登录未完成，请检查网络代理并重试；也可在终端运行 git credential-manager github login --browser。")


class GitHubLogin:
    def __init__(self):
        self.condition = threading.Condition()
        self.running = False
        self.generation = 0
        self.finished_at = float("-inf")
        self.error = None

    def version(self):
        with self.condition:
            return self.generation

    def snapshot(self):
        with self.condition:
            return {"running": self.running}

    def ensure_authenticated(self, settings, observed_version):
        with self.condition:
            while self.running:
                self.condition.wait()
            # Share the outcome with concurrent requests and avoid repeated popups.
            if observed_version != self.generation or time.monotonic() - self.finished_at < 30:
                if self.error:
                    raise ValueError(self.error)
                return
            self.running = True
        error = None
        try:
            browser_login(settings)
        except (ValueError, OSError, subprocess.SubprocessError) as exc:
            error = str(exc)
        finally:
            with self.condition:
                self.running = False
                self.generation += 1
                self.finished_at = time.monotonic()
                self.error = error
                self.condition.notify_all()
        if error:
            raise ValueError(error)


github_login = GitHubLogin()