v1_mirror_history.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """Bounded offline import of reviewed legacy history into a new owned mirror. |
| 2 | |
| 3 | Only complete SHA-1 v2 bundles with exactly the two destination refs are accepted. |
| 4 | No Git config, hooks, remote, worktree or refs are imported from another directory. |
| 5 | """ |
| 6 | import base64 |
| 7 | import os |
| 8 | from pathlib import Path |
| 9 | import re |
| 10 | import stat |
| 11 | |
| 12 | from cli.v1_io import Refusal, digest |
| 13 | |
| 14 | BUNDLE_LIMIT = 128 * 1024 * 1024 |
| 15 | SHA256 = re.compile(r'[0-9a-f]{64}\Z') |
| 16 | |
| 17 | |
| 18 | def validate(plan): |
| 19 | from cli.v1 import keys |
| 20 | from cli.v1_mirror import HEX, physical |
| 21 | r = plan['reconciliation'] |
| 22 | keys(r, ('bundle', 'bundle_sha256', 'mirror_head', 'base_head', 'delta')) |
| 23 | physical(r['bundle']) |
| 24 | if (not SHA256.fullmatch(r['bundle_sha256']) or not HEX.fullmatch(r['mirror_head']) |
| 25 | or not HEX.fullmatch(r['base_head']) or r['mirror_head'] == r['base_head']): |
| 26 | raise Refusal('mirror_reconciliation_identity_invalid') |
| 27 | if plan['expected_target'] == 'absent' and ( |
| 28 | plan['destination']['expected_head'] != r['mirror_head'] |
| 29 | or plan['destination'].get('expected_base', r['base_head']) != r['base_head']): |
| 30 | raise Refusal('mirror_reconciliation_head_mismatch') |
| 31 | if not isinstance(r['delta'], dict): |
| 32 | raise Refusal('mirror_reconciliation_delta_invalid') |
| 33 | for path, entry in r['delta'].items(): |
| 34 | if (not isinstance(path, str) or not path or path.startswith('/') |
| 35 | or any(p in ('', '.', '..') for p in path.split('/')) or '\0' in path): |
| 36 | raise Refusal('mirror_reconciliation_path_invalid') |
| 37 | keys(entry, ('before', 'after', 'mode_before', 'mode_after')) |
| 38 | for side in ('before', 'after'): |
| 39 | value, mode = entry[side], entry['mode_' + side] |
| 40 | if (value is None and mode is not None) or (value is not None and |
| 41 | (not isinstance(value, str) or not SHA256.fullmatch(value) |
| 42 | or mode not in ('100644', '100755'))): |
| 43 | raise Refusal('mirror_reconciliation_delta_invalid') |
| 44 | if entry['before'] is None and entry['after'] is None: |
| 45 | raise Refusal('mirror_reconciliation_delta_invalid') |
| 46 | |
| 47 | |
| 48 | def bundle_bytes(plan): |
| 49 | from cli.v1_mirror import physical |
| 50 | r = plan['reconciliation'] |
| 51 | path = physical(r['bundle']) |
| 52 | fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK) |
| 53 | with os.fdopen(fd, 'rb') as stream: |
| 54 | info = os.fstat(stream.fileno()) |
| 55 | if (not stat.S_ISREG(info.st_mode) or info.st_nlink != 1 |
| 56 | or info.st_mode & 0o7111 or info.st_size > BUNDLE_LIMIT): |
| 57 | raise Refusal('mirror_history_bundle_not_regular') |
| 58 | raw = stream.read(BUNDLE_LIMIT + 1) |
| 59 | if len(raw) > BUNDLE_LIMIT or digest(raw) != r['bundle_sha256']: |
| 60 | raise Refusal('mirror_history_bundle_digest_mismatch') |
| 61 | header, separator, pack = raw.partition(b'\n\n') |
| 62 | dest = plan['destination'] |
| 63 | expected = {f'{r["mirror_head"]} refs/heads/{dest["branch"]}'.encode(), |
| 64 | f'{r["base_head"]} refs/heads/{dest["base"]}'.encode()} |
| 65 | lines = header.split(b'\n') |
| 66 | if (not separator or lines[0] != b'# v2 git bundle' or len(lines) != 3 |
| 67 | or set(lines[1:]) != expected or not pack.startswith(b'PACK')): |
| 68 | raise Refusal('mirror_history_bundle_refs_or_prerequisites') |
| 69 | return pack |
| 70 | |
| 71 | |
| 72 | def import_history(target, plan, files): |
| 73 | from cli.v1_mirror import git |
| 74 | # Unpack directly into loose objects: no imported config, refs, packs or |
| 75 | # alternates. Git checks the pack and object structure before installation. |
| 76 | git(target, 'unpack-objects', '-r', '--strict', data=bundle_bytes(plan)) |
| 77 | verify_history(target, plan, files) |
| 78 | r = plan['reconciliation'] |
| 79 | reachable = {row.split(b' ', 1)[0] for row in git( |
| 80 | target, 'rev-list', '--objects', r['mirror_head'], r['base_head']).splitlines()} |
| 81 | actual = set(git(target, 'cat-file', '--batch-all-objects', '--batch-check=%(objectname)').splitlines()) |
| 82 | if reachable != actual: |
| 83 | raise Refusal('mirror_history_unrelated_objects') |
| 84 | |
| 85 | |
| 86 | def manifest(target, commit): |
| 87 | from cli.v1_mirror import git |
| 88 | result = {} |
| 89 | for row in git(target, 'ls-tree', '-r', '-z', commit).split(b'\0'): |
| 90 | if not row: |
| 91 | continue |
| 92 | meta, path = row.split(b'\t', 1) |
| 93 | mode, kind, oid = meta.decode().split() |
| 94 | if kind != 'blob' or mode not in ('100644', '100755'): |
| 95 | raise Refusal('mirror_history_tree_unsupported') |
| 96 | result[path.decode()] = {'sha256': digest(git(target, 'cat-file', 'blob', oid)), 'mode': mode} |
| 97 | return result |
| 98 | |
| 99 | |
| 100 | def delta(before, after): |
| 101 | return {p: {'before': before[p]['sha256'] if p in before else None, |
| 102 | 'after': after[p]['sha256'] if p in after else None, |
| 103 | 'mode_before': before[p]['mode'] if p in before else None, |
| 104 | 'mode_after': after[p]['mode'] if p in after else None} |
| 105 | for p in sorted(before.keys() | after.keys()) if before.get(p) != after.get(p)} |
| 106 | |
| 107 | |
| 108 | def verify_history(target, plan, files): |
| 109 | from cli.v1_mirror import git, HEX |
| 110 | r = plan['reconciliation'] |
| 111 | parents = [r['mirror_head'], r['base_head']] |
| 112 | for parent in parents: |
| 113 | if git(target, 'cat-file', '-t', parent) != b'commit\n': |
| 114 | raise Refusal('mirror_history_head_not_commit') |
| 115 | git(target, 'fsck', '--strict', '--no-reflogs', '--no-dangling', *parents) |
| 116 | common = git(target, 'merge-base', *parents, optional=True).decode().strip() |
| 117 | if not HEX.fullmatch(common): |
| 118 | raise Refusal('mirror_history_unrelated_heads') |
| 119 | trees = [git(target, 'rev-parse', p + '^{tree}') for p in parents] |
| 120 | if trees[0] != trees[1]: |
| 121 | raise Refusal('mirror_history_base_tree_differs') |
| 122 | after = {p: {'sha256': digest(base64.b64decode(item['data'], validate=True)), 'mode': item['mode']} |
| 123 | for p, item in files.items()} |
| 124 | if delta(manifest(target, parents[0]), after) != r['delta']: |
| 125 | raise Refusal('mirror_reconciliation_delta_mismatch') |