diff --git a/GitMirror/mirror.py b/GitMirror/mirror.py index df2b889b..4738b518 100644 --- a/GitMirror/mirror.py +++ b/GitMirror/mirror.py @@ -24,6 +24,7 @@ import secrets as py_secrets import sqlite3 import subprocess import threading +import urllib.parse import requests @@ -81,7 +82,6 @@ class MirrorConfig: os.makedirs(self.WorkDir, exist_ok=True) self.ReposDir = os.path.join(self.WorkDir, 'repos') os.makedirs(self.ReposDir, exist_ok=True) - self.CredFile = os.path.join(self.WorkDir, 'git-credentials') self.DbFile = os.path.join(self.WorkDir, 'mirror.db') # Gitea 令牌缺省从 secrets.ini 的 [gitea] API_TOKEN 读取（与主站共用） @@ -301,7 +301,12 @@ class GithubClient: # --------------------------------------------------------------------------- class MirrorManager: - """单个仓库的镜像同步（bare clone + fetch + push）""" + """单个仓库的镜像同步（bare 仓库 + fetch + push） + + 凭据处理：把带凭据的地址写入本地 bare 仓库的 remote 配置（origin=Gitea、 + github=GitHub），命令只引用 remote 名。这样令牌既不出现在 argv，也不依赖 + git credential helper（避免被全局 GCM/OAuth 凭据抢先命中）。 + """ def __init__(self, Cfg: MirrorConfig, Gitea: GiteaClient, Github: GithubClient, Store: MirrorStore): self.Cfg = Cfg @@ -311,7 +316,6 @@ class MirrorManager: self._Locks: dict[str, threading.Lock] = {} self._LocksGuard = threading.Lock() self._LfsAvailable = self._DetectLfs() - self._WriteCredentials() # -- 基础设施 -- @@ -326,39 +330,47 @@ class MirrorManager: logger.warning('未检测到 git lfs，将跳过 LFS 对象同步') return False - def _WriteCredentials(self) -> None: - """写入 git credential store 文件（600 权限）""" - Lines = [ - f'https://oauth2:{self.Cfg.GiteaToken}@{self._GiteaHost()}', - f'https://{GITHUB_AUTH_USER}:{self.Cfg.GithubToken}@github.com', - ] - with open(self.Cfg.CredFile, 'w', encoding='utf-8') as Fh: - Fh.write('\n'.join(Lines) + '\n') - try: - os.chmod(self.Cfg.CredFile, 0o600) - except OSError: - pass + @staticmethod + def _AuthUrl(Url: str, User: str, Token: str) -> str: + """把令牌嵌入 URL（只写入本地 remote 配置，不进 argv）""" + SafeUser = urllib.parse.quote(User, safe='') + SafeToken = urllib.parse.quote(Token, safe='') + return Url.replace('https://', f'https://{SafeUser}:{SafeToken}@', 1) - def _GiteaHost(self) -> str: - return self.Cfg.GiteaBaseUrl.split('://', 1)[-1].rstrip('/') + @staticmethod + def _EnsureRemote(Local: str, RunGit, Name: str, Url: str) -> None: + Remotes = RunGit(['-C', Local, 'remote']) + if Name in Remotes.split(): + RunGit(['-C', Local, 'remote', 'set-url', Name, Url]) + else: + RunGit(['-C', Local, 'remote', 'add', Name, Url]) def LockFor(self, RepoName: str) -> threading.Lock: with self._LocksGuard: return self._Locks.setdefault(RepoName, threading.Lock()) + def _DropRefs(self, Local: str, Prefixes: list[str]) -> None: + """删除指定命名空间的本地 ref（避免把 Gitea 隐藏 ref 推到 GitHub）""" + for Prefix in Prefixes: + Output = self._Git(['-C', Local, 'for-each-ref', '--format=%(refname)', Prefix]) + for RefName in Output.split(): + self._Git(['-C', Local, 'update-ref', '-d', RefName]) + def _Sanitize(self, Text: str) -> str: if not Text: return '' for Secret in (self.Cfg.GithubToken, self.Cfg.GiteaToken): if Secret: Text = Text.replace(Secret, '***') + Text = Text.replace(urllib.parse.quote(Secret, safe=''), '***') return Text def _Git(self, Args: list[str], Cwd: str | None = None) -> str: Cmd = [ 'git', '-c', 'core.longpaths=true', - '-c', f'credential.helper=store --file={self.Cfg.CredFile}', + # 清空 credential helper，杜绝全局 GCM/OAuth 凭据抢先命中 + '-c', 'credential.helper=', ] + Args Env = os.environ.copy() Env['GIT_TERMINAL_PROMPT'] = '0' @@ -386,8 +398,10 @@ class MirrorManager: GithubName = self._GithubName(RepoName) Private = bool(Repo.get('private', True)) Local = os.path.join(self.Cfg.ReposDir, f'{RepoName}.git') - SourceUrl = f'{self.Cfg.GiteaBaseUrl}/{self.Cfg.GiteaOrg}/{RepoName}.git' - TargetUrl = f'https://github.com/{self.Cfg.GithubOrg}/{GithubName}.git' + SourceUrl = self._AuthUrl( + f'{self.Cfg.GiteaBaseUrl}/{self.Cfg.GiteaOrg}/{RepoName}.git', 'oauth2', self.Cfg.GiteaToken) + TargetUrl = self._AuthUrl( + f'https://github.com/{self.Cfg.GithubOrg}/{GithubName}.git', GITHUB_AUTH_USER, self.Cfg.GithubToken) with self.LockFor(RepoName): logger.info('开始镜像 %s -> %s/%s', RepoName, self.Cfg.GithubOrg, GithubName) @@ -396,15 +410,24 @@ class MirrorManager: try: self.Github.EnsureRepo(GithubName, Private) if not os.path.isdir(Local): - self._Git(['clone', '--mirror', SourceUrl, Local]) - else: - self._Git(['-C', Local, 'remote', 'set-url', 'origin', SourceUrl]) + self._Git(['init', '--bare', Local]) + self._EnsureRemote(Local, self._Git, 'origin', SourceUrl) + self._EnsureRemote(Local, self._Git, 'github', TargetUrl) + # 只镜像 heads/tags/notes：`refs/pull/*` 是 Gitea 的隐藏 ref， + # 推到 GitHub 会被拒（deny updating a hidden ref）。 + self._Git(['-C', Local, 'config', '--replace-all', 'remote.origin.fetch', + '+refs/heads/*:refs/heads/*']) + self._Git(['-C', Local, 'config', '--add', 'remote.origin.fetch', + '+refs/tags/*:refs/tags/*']) + self._Git(['-C', Local, 'config', '--add', 'remote.origin.fetch', + '+refs/notes/*:refs/notes/*']) + self._DropRefs(Local, ['refs/remotes', 'refs/pull']) self._Git(['-C', Local, 'fetch', '--prune', '--prune-tags', '--force', 'origin']) if self._LfsAvailable: - self._Git(['-C', Local, 'lfs', 'fetch', '--all']) - self._Git(['-C', Local, 'push', '--mirror', TargetUrl]) + self._Git(['-C', Local, 'lfs', 'fetch', '--all', 'origin']) + self._Git(['-C', Local, 'push', '--mirror', 'github']) if self._LfsAvailable: - self._Git(['-C', Local, 'lfs', 'push', '--all', TargetUrl]) + self._Git(['-C', Local, 'lfs', 'push', '--all', 'github']) except MirrorError as Error: self.Store.SetStatus(RepoName, Status='failed', LastError=str(Error)[:1000]) raise