148 lines
6.2 KiB
Python
148 lines
6.2 KiB
Python
"""Rebuild commits the git transport cannot fetch, using the Gitea HTTP API.
|
|
|
|
Why this exists: git.rpchannel.com has repeatedly served `info/refs` and the
|
|
whole HTTP API fine while `git-upload-pack` / `git-receive-pack` 504 (the
|
|
server-side git process hangs -- a 32KB push body 504s just as a full fetch
|
|
does). That leaves the local checkout unable to rebase onto a remote head it
|
|
can see but cannot download.
|
|
|
|
The API exposes the same objects the pack protocol would have shipped, so the
|
|
missing commits can be rebuilt locally:
|
|
|
|
/api/v1/repos/<o>/<r>/git/commits/<sha> author, committer, message, parents
|
|
/api/v1/repos/<o>/<r>/git/trees/<sha>?recursive=1 every path, mode and blob sha
|
|
/api/v1/repos/<o>/<r>/git/blobs/<sha> base64 content
|
|
|
|
This is a reconstruction, not a copy, so it is only trustworthy because it is
|
|
self-verifying: every blob is accepted only if its SHA-1 matches what the
|
|
server reported, and a commit is written only if the assembled object hashes to
|
|
the exact sha being recovered. A mismatch aborts instead of writing something
|
|
plausible.
|
|
|
|
Two traps, both of which produce a silently wrong result rather than an error:
|
|
* `git update-index --index-info` input must be bytes. With text=True on
|
|
Windows the newlines become CRLF, every line is rejected, and write-tree
|
|
happily returns the EMPTY tree (4b825dc6...).
|
|
* Gitea fills the commit's `tree.sha` field with the COMMIT sha. Do not use it
|
|
as a check -- verify the final commit sha instead.
|
|
|
|
Usage:
|
|
python recover_commits_via_gitea_api.py <repo_dir> <sha> [<sha> ...]
|
|
|
|
Recover in parent-first order. Afterwards the shas are real local objects:
|
|
git update-ref refs/remotes/origin/<branch> <newest_sha>
|
|
git rebase <newest_sha>
|
|
"""
|
|
|
|
import base64
|
|
import datetime
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import urllib.request
|
|
|
|
|
|
def api_base(repo_dir):
|
|
"""Derive the Gitea API root from origin's URL."""
|
|
url = subprocess.run(("git", "-C", repo_dir, "remote", "get-url", "origin"),
|
|
capture_output=True, text=True, check=True).stdout.strip()
|
|
m = re.match(r"https?://([^/]+)/([^/]+)/([^/]+?)(?:\.git)?/?$", url)
|
|
if not m:
|
|
sys.exit(f"cannot parse a Gitea owner/repo out of origin url: {url}")
|
|
host, owner, repo = m.groups()
|
|
return f"http://{host}/api/v1/repos/{owner}/{repo}"
|
|
|
|
|
|
def get(url):
|
|
with urllib.request.urlopen(url, timeout=90) as f:
|
|
return json.load(f)
|
|
|
|
|
|
def git_date(iso):
|
|
"""'2026-07-30T22:19:56+08:00' -> ('1785...', '+0800')"""
|
|
dt = datetime.datetime.fromisoformat(iso)
|
|
total = int(dt.utcoffset().total_seconds())
|
|
sign = "+" if total >= 0 else "-"
|
|
total = abs(total)
|
|
return str(int(dt.timestamp())), f"{sign}{total // 3600:02d}{(total % 3600) // 60:02d}"
|
|
|
|
|
|
def recover(sha, base, env, index_file):
|
|
meta = get(f"{base}/git/commits/{sha}")
|
|
tree = get(f"{base}/git/trees/{sha}?recursive=true&per_page=1000")
|
|
if tree.get("truncated"):
|
|
sys.exit(f"{sha[:8]}: tree listing truncated -- raise per_page")
|
|
|
|
blobs = [e for e in tree["tree"] if e["type"] == "blob"]
|
|
fetched = 0
|
|
for e in blobs:
|
|
present = subprocess.run(("git", "cat-file", "-e", e["sha"] + "^{object}"),
|
|
env=env, capture_output=True).returncode == 0
|
|
if present:
|
|
continue
|
|
raw = base64.b64decode(get(f"{base}/git/blobs/{e['sha']}")["content"])
|
|
got = hashlib.sha1(b"blob %d\0" % len(raw) + raw).hexdigest()
|
|
if got != e["sha"]:
|
|
sys.exit(f"blob mismatch for {e['path']}: {got} != {e['sha']}")
|
|
w = subprocess.run(("git", "hash-object", "-t", "blob", "-w", "--stdin"),
|
|
env=env, input=raw, capture_output=True)
|
|
if w.stdout.decode().strip() != e["sha"]:
|
|
sys.exit(f"write-back drift for {e['path']}")
|
|
fetched += 1
|
|
|
|
if os.path.exists(index_file):
|
|
os.remove(index_file)
|
|
payload = "".join(f"{e['mode']} {e['sha']}\t{e['path']}\n" for e in blobs)
|
|
r = subprocess.run(("git", "update-index", "--index-info"), env=env,
|
|
input=payload.encode("utf-8"), capture_output=True)
|
|
if r.returncode:
|
|
sys.exit(f"update-index failed: {r.stderr.decode()}")
|
|
tree_sha = subprocess.run(("git", "write-tree"), env=env,
|
|
capture_output=True, text=True, check=True).stdout.strip()
|
|
|
|
c = meta["commit"]
|
|
ats, atz = git_date(c["author"]["date"])
|
|
cts, ctz = git_date(c["committer"]["date"])
|
|
head = f"tree {tree_sha}\n"
|
|
for p in meta.get("parents", []):
|
|
head += f"parent {p['sha']}\n"
|
|
head += (f"author {c['author']['name']} <{c['author']['email']}> {ats} {atz}\n"
|
|
f"committer {c['committer']['name']} <{c['committer']['email']}> {cts} {ctz}\n\n")
|
|
|
|
msg = c["message"]
|
|
for variant in (msg, msg + "\n", msg.rstrip("\n") + "\n", msg.rstrip("\n")):
|
|
body = (head + variant).encode("utf-8")
|
|
if hashlib.sha1(b"commit %d\0" % len(body) + body).hexdigest() == sha:
|
|
w = subprocess.run(("git", "hash-object", "-t", "commit", "-w", "--stdin"),
|
|
env=env, input=body, capture_output=True)
|
|
if w.stdout.decode().strip() != sha:
|
|
sys.exit(f"{sha[:8]}: commit write drift")
|
|
print(f"OK {sha[:8]} tree={tree_sha[:8]} blobs_fetched={fetched}")
|
|
return True
|
|
print(f"FAIL {sha[:8]} no reconstruction hashes to this sha "
|
|
f"(signed commit, or the API is reporting different metadata)")
|
|
return False
|
|
|
|
|
|
def main():
|
|
if len(sys.argv) < 3:
|
|
sys.exit(__doc__)
|
|
repo_dir, shas = sys.argv[1], sys.argv[2:]
|
|
git_dir = subprocess.run(("git", "-C", repo_dir, "rev-parse", "--absolute-git-dir"),
|
|
capture_output=True, text=True, check=True).stdout.strip()
|
|
index_file = os.path.join(os.environ.get("TEMP", "."), "recover_gitea_index")
|
|
# A private index keeps the caller's working tree and real index untouched.
|
|
env = dict(os.environ, GIT_DIR=git_dir, GIT_INDEX_FILE=index_file)
|
|
base = api_base(repo_dir)
|
|
ok = all(recover(s, base, env, index_file) for s in shas)
|
|
if os.path.exists(index_file):
|
|
os.remove(index_file)
|
|
return 0 if ok else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|