v1_mirror.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """Controlled snapshot mirrors. Preparation is offline; delivery is explicit. |
| 2 | |
| 3 | One source checkout, source main branch and Git destination pair. Only a newly |
| 4 | created, dedicated bare Git target or this engine's own target can be used. |
| 5 | Git plumbing avoids checkout filters, index state, source hooks and native export |
| 6 | side effects. A single correspondence record plus commit trailers supports retry. |
| 7 | """ |
| 8 | from __future__ import annotations |
| 9 | |
| 10 | import base64 |
| 11 | from contextlib import contextmanager |
| 12 | import fcntl |
| 13 | import json |
| 14 | import os |
| 15 | from pathlib import Path |
| 16 | import re |
| 17 | import shlex |
| 18 | import stat |
| 19 | import subprocess |
| 20 | import sys |
| 21 | import tempfile |
| 22 | import urllib.request |
| 23 | |
| 24 | from cli.v1_io import Refusal, atomic, digest, read |
| 25 | from cli.v1_revision import OID, bounded_run, revision, unique_json |
| 26 | |
| 27 | RECORD = '.overseer/local/git-bridge.json' |
| 28 | MARKER = 'overseer-mirror.json' |
| 29 | HEX = re.compile(r'[0-9a-f]{40}\Z') |
| 30 | CONFIG = b'[core]\n\trepositoryformatversion = 0\n\tfilemode = true\n\tbare = true\n\thooksPath = /dev/null\n\tlogAllRefUpdates = false\n' |
| 31 | |
| 32 | |
| 33 | def packed(value): |
| 34 | return (json.dumps(value, sort_keys=True, indent=2) + '\n').encode() |
| 35 | |
| 36 | |
| 37 | def env(): |
| 38 | result = {k: v for k, v in os.environ.items() |
| 39 | if not k.startswith(('GIT_', 'MUSE_', 'PYTHON')) |
| 40 | and k not in {'VIRTUAL_ENV', 'SSH_ASKPASS', 'SSH_ASKPASS_REQUIRE', 'GH_DEBUG'}} |
| 41 | result.update(PATH='/usr/bin:/bin', GIT_CONFIG_NOSYSTEM='1', GIT_CONFIG_GLOBAL='/dev/null', |
| 42 | GIT_TERMINAL_PROMPT='0', GIT_ASKPASS='/usr/bin/false', GIT_OPTIONAL_LOCKS='0', LC_ALL='C', |
| 43 | GH_HOST='github.com', GH_PROMPT_DISABLED='1', |
| 44 | GIT_AUTHOR_NAME='Overseer mirror', GIT_AUTHOR_EMAIL='[email protected]', |
| 45 | GIT_COMMITTER_NAME='Overseer mirror', GIT_COMMITTER_EMAIL='[email protected]') |
| 46 | return result |
| 47 | |
| 48 | |
| 49 | def git(target, *args, data=None, optional=False, options=()): |
| 50 | proc = subprocess.run(['/usr/bin/git', '--no-replace-objects', '-c', 'core.hooksPath=/dev/null', |
| 51 | '-c', 'core.fsmonitor=false', '-c', 'gc.auto=0', *options, '-C', str(target), *args], |
| 52 | input=data, env=env(), capture_output=True, timeout=30) |
| 53 | if proc.returncode and not optional: |
| 54 | raise Refusal('mirror_git_failed: ' + args[0]) |
| 55 | return proc.stdout if not proc.returncode else b'' |
| 56 | |
| 57 | |
| 58 | def head(target): |
| 59 | return git(target, 'rev-parse', '--verify', 'HEAD', optional=True).decode().strip() or 'absent' |
| 60 | |
| 61 | |
| 62 | def physical(value): |
| 63 | if not isinstance(value, str): |
| 64 | raise Refusal('mirror_path_requires_physical_absolute_root') |
| 65 | path = Path(value) |
| 66 | if not path.is_absolute() or path.resolve() != path or any(p.is_symlink() for p in (path, *path.parents)): |
| 67 | raise Refusal('mirror_path_requires_physical_absolute_root') |
| 68 | return path |
| 69 | |
| 70 | |
| 71 | def validate_plan(plan, root, config): |
| 72 | from cli.v1 import keys, KIT |
| 73 | fields = ('schema', 'source', 'destination', 'target', 'expected_target', 'executables') |
| 74 | keys(plan, (*fields, 'reconciliation') if 'reconciliation' in plan else fields) |
| 75 | keys(plan['source'], ('repo_id', 'branch', 'revision', 'hub_url')) |
| 76 | dest_fields = ('url', 'repository', 'branch', 'base', 'expected_head') |
| 77 | keys(plan['destination'], (*dest_fields, 'expected_base') if 'expected_base' in plan['destination'] else dest_fields) |
| 78 | source, dest = plan['source'], plan['destination'] |
| 79 | if plan['schema'] != 1 or config['vcs'] != 'muse': |
| 80 | raise Refusal('mirror_requires_explicit_muse_authority') |
| 81 | if source['branch'] != 'main' or not OID.fullmatch(source['revision']): |
| 82 | raise Refusal('mirror_requires_approved_main_revision') |
| 83 | if source['repo_id'] != config['muse']['repo_id']: |
| 84 | raise Refusal('mirror_source_identity_mismatch') |
| 85 | # Public staging is the tested authority. Do not bypass unresolved production trust. |
| 86 | if not re.fullmatch(r'https://staging\.musehub\.ai/[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+', source['hub_url']): |
| 87 | raise Refusal('mirror_authority_url_unsupported') |
| 88 | if not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9_.-]*/[A-Za-z0-9][A-Za-z0-9_.-]*', dest['repository']): |
| 89 | raise Refusal('mirror_github_repository_invalid') |
| 90 | if dest['url'] != 'https://github.com/' + dest['repository'] + '.git': |
| 91 | raise Refusal('mirror_destination_repository_mismatch') |
| 92 | for branch in (dest['branch'], dest['base']): |
| 93 | if not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9_/-]*', branch) or branch.endswith('/') or '//' in branch: |
| 94 | raise Refusal('mirror_git_branch_invalid') |
| 95 | if dest['branch'] in {'main', 'master', dest['base']}: |
| 96 | raise Refusal('mirror_requires_distribution_branch') |
| 97 | for expected in (plan['expected_target'], dest['expected_head']): |
| 98 | if expected != 'absent' and not HEX.fullmatch(expected): |
| 99 | raise Refusal('mirror_expected_head_invalid') |
| 100 | if 'expected_base' in dest and not HEX.fullmatch(dest['expected_base']): |
| 101 | raise Refusal('mirror_expected_base_invalid') |
| 102 | if 'reconciliation' in plan: |
| 103 | from cli.v1_mirror_history import validate |
| 104 | validate(plan) |
| 105 | elif plan['expected_target'] == 'absent' and dest['expected_head'] != 'absent': |
| 106 | raise Refusal('mirror_legacy_destination_requires_reconciliation') |
| 107 | ex = plan['executables'] |
| 108 | if not isinstance(ex, list) or any(not isinstance(p, str) for p in ex) or ex != sorted(set(ex)): |
| 109 | raise Refusal('mirror_executable_policy_invalid') |
| 110 | target = physical(plan['target']) |
| 111 | for protected in (root, Path(config['muse']['store']), KIT): |
| 112 | if target.is_relative_to(protected) or protected.is_relative_to(target): |
| 113 | raise Refusal('mirror_target_overlap') |
| 114 | if not target.parent.is_dir(): |
| 115 | raise Refusal('mirror_target_parent_missing') |
| 116 | return target |
| 117 | |
| 118 | |
| 119 | def binding(plan, config): |
| 120 | pair = {'source_root': config['repo']['root'], 'checkout_id': config['repo']['id'], |
| 121 | 'source_id': plan['source']['repo_id'], 'source_branch': plan['source']['branch'], |
| 122 | 'hub_url': plan['source']['hub_url'], 'target': plan['target'], |
| 123 | 'destination': {k: v for k, v in plan['destination'].items() if k not in {'expected_head', 'expected_base'}}} |
| 124 | if 'reconciliation' in plan: |
| 125 | pair['reconciliation'] = plan['reconciliation'] |
| 126 | return pair |
| 127 | |
| 128 | |
| 129 | @contextmanager |
| 130 | def locked(root, target): |
| 131 | handles = [] |
| 132 | try: |
| 133 | # Lock before target creation. Physical aliases share the same directory inode. |
| 134 | for path in (root, target.parent): |
| 135 | fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) |
| 136 | handles.append(fd) |
| 137 | try: |
| 138 | fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) |
| 139 | except BlockingIOError: |
| 140 | raise Refusal('mirror_operator_busy') from None |
| 141 | yield |
| 142 | finally: |
| 143 | for fd in handles: |
| 144 | os.close(fd) |
| 145 | |
| 146 | |
| 147 | def check_source(root, config, plan): |
| 148 | state = revision(root, config) |
| 149 | if (state.branch, state.head, state.source_id) != ( |
| 150 | plan['source']['branch'], plan['source']['revision'], plan['source']['repo_id']): |
| 151 | raise Refusal('mirror_source_drift') |
| 152 | python = config['muse']['runtime']['python'] |
| 153 | helper = Path(__file__).with_name('v1_mirror_reader.py') |
| 154 | proc = bounded_run([python, '-I', '-B', str(helper), str(root), json.dumps(plan['executables'])], |
| 155 | cwd=root, env=env(), limit=48 * 1024 * 1024, timeout=30) |
| 156 | if proc.returncode: |
| 157 | try: |
| 158 | reason = json.loads(proc.stdout, object_pairs_hook=unique_json)['error'] |
| 159 | if not isinstance(reason, str) or len(reason) > 300: |
| 160 | raise ValueError() |
| 161 | except (ValueError, KeyError, TypeError): |
| 162 | reason = 'invalid reader response' |
| 163 | raise Refusal('mirror_projection_refused: ' + reason) |
| 164 | value = json.loads(proc.stdout, object_pairs_hook=unique_json) |
| 165 | if (value['state']['runtime'] != config['muse']['runtime'] |
| 166 | or value['state']['head'] != state.head or value['state']['repo_id'] != state.source_id |
| 167 | or value['state']['branch'] != state.branch or value['state']['root'] != str(root) |
| 168 | or value['state']['store'] != config['muse']['store']): |
| 169 | raise Refusal('mirror_source_changed') |
| 170 | return value |
| 171 | |
| 172 | |
| 173 | def inspect_target(target, pair): |
| 174 | physical(str(target)) |
| 175 | if not target.exists(): |
| 176 | return 'absent' |
| 177 | if not target.is_dir() or not list(target.iterdir()): |
| 178 | raise Refusal('mirror_target_not_owned') |
| 179 | if read(target, MARKER, missing=True) != packed(pair): |
| 180 | raise Refusal('mirror_foreign_target') |
| 181 | if read(target, 'config') != CONFIG: |
| 182 | raise Refusal('mirror_target_config_or_hooks_changed') |
| 183 | allowed = {'HEAD', 'config', MARKER, 'objects', 'refs'} |
| 184 | if set(p.name for p in target.iterdir()) - allowed: |
| 185 | raise Refusal('mirror_dirty_or_hooked_target') |
| 186 | for path in target.rglob('*'): |
| 187 | mode = path.lstat() |
| 188 | if stat.S_ISLNK(mode.st_mode) or (not stat.S_ISDIR(mode.st_mode) and |
| 189 | (not stat.S_ISREG(mode.st_mode) or mode.st_nlink != 1 or mode.st_mode & 0o7111)): |
| 190 | raise Refusal('mirror_target_unsafe_path') |
| 191 | rel = path.relative_to(target).as_posix() |
| 192 | if path.is_dir() and rel not in {'objects', 'refs', 'refs/heads'}: |
| 193 | ref_dir = 'refs/heads/' + pair['destination']['branch'] |
| 194 | if not re.fullmatch(r'objects/[0-9a-f]{2}', rel) and not ref_dir.startswith(rel + '/'): |
| 195 | raise Refusal('mirror_foreign_directory') |
| 196 | if path.is_file() and rel.startswith('refs/') and rel != 'refs/heads/' + pair['destination']['branch']: |
| 197 | raise Refusal('mirror_foreign_ref_file') |
| 198 | if path.is_file() and rel.startswith('objects/') and not re.fullmatch(r'objects/[0-9a-f]{2}/[0-9a-f]{38}', rel): |
| 199 | raise Refusal('mirror_foreign_object_or_alternate') |
| 200 | branch = pair['destination']['branch'] |
| 201 | if read(target, 'HEAD') != ('ref: refs/heads/' + branch + '\n').encode(): |
| 202 | raise Refusal('mirror_target_branch_changed') |
| 203 | refs = git(target, 'for-each-ref', '--format=%(refname)').decode().splitlines() |
| 204 | if set(refs) - {'refs/heads/' + branch}: |
| 205 | raise Refusal('mirror_foreign_refs') |
| 206 | return head(target) |
| 207 | |
| 208 | |
| 209 | def create_target(target, pair, plan=None, files=None): |
| 210 | # Initialize a sibling first: a killed init never leaves an ambiguous target. |
| 211 | with tempfile.TemporaryDirectory(prefix='.overseer-mirror-', dir=target.parent) as name: |
| 212 | stage = Path(name) |
| 213 | git(stage, 'init', '--bare', '--template=', '--initial-branch=' + pair['destination']['branch']) |
| 214 | (stage / 'config').write_bytes(CONFIG) |
| 215 | # Git init's empty metadata directories are unnecessary for loose objects. |
| 216 | (stage / 'objects/info').rmdir() |
| 217 | (stage / 'objects/pack').rmdir() |
| 218 | (stage / 'refs/tags').rmdir() |
| 219 | (stage / MARKER).write_bytes(packed(pair)) |
| 220 | if plan and 'reconciliation' in plan: |
| 221 | from cli.v1_mirror_history import import_history |
| 222 | import_history(stage, plan, files) |
| 223 | inspect_target(stage, pair) |
| 224 | if target.exists(): |
| 225 | raise Refusal('mirror_target_appeared') |
| 226 | os.rename(stage, target) |
| 227 | |
| 228 | |
| 229 | def tree(target, files): |
| 230 | entries = {} |
| 231 | for path, item in files.items(): |
| 232 | node = entries |
| 233 | parts = path.split('/') |
| 234 | for part in parts[:-1]: |
| 235 | node = node.setdefault(part, {}) |
| 236 | data = base64.b64decode(item['data'], validate=True) |
| 237 | oid = git(target, 'hash-object', '-w', '--stdin', data=data).decode().strip() |
| 238 | node[parts[-1]] = (item['mode'], oid) |
| 239 | def write(node): |
| 240 | rows = [] |
| 241 | for name, item in sorted(node.items()): |
| 242 | mode, oid, kind = ('040000', write(item), 'tree') if isinstance(item, dict) else (*item, 'blob') |
| 243 | rows.append(f'{mode} {kind} {oid}\t{name}'.encode() + b'\0') |
| 244 | return git(target, 'mktree', '-z', data=b''.join(rows)).decode().strip() |
| 245 | return write(entries) |
| 246 | |
| 247 | |
| 248 | def verify_tree(target, commit, files): |
| 249 | found = {} |
| 250 | for row in git(target, 'ls-tree', '-r', '-z', commit).split(b'\0'): |
| 251 | if not row: |
| 252 | continue |
| 253 | meta, path = row.split(b'\t', 1) |
| 254 | mode, kind, oid = meta.decode().split() |
| 255 | path = path.decode() |
| 256 | if kind != 'blob' or path not in files or mode != files[path]['mode']: |
| 257 | raise Refusal('mirror_tree_paths_or_modes_mismatch') |
| 258 | content = git(target, 'cat-file', 'blob', oid) |
| 259 | if content != base64.b64decode(files[path]['data'], validate=True): |
| 260 | raise Refusal('mirror_tree_bytes_mismatch') |
| 261 | found[path] = True |
| 262 | if set(found) != set(files): |
| 263 | raise Refusal('mirror_tree_paths_mismatch') |
| 264 | |
| 265 | |
| 266 | def expected_parents(plan): |
| 267 | if plan['expected_target'] != 'absent': |
| 268 | return [plan['expected_target']] |
| 269 | if 'reconciliation' in plan: |
| 270 | r = plan['reconciliation'] |
| 271 | return [r['mirror_head'], r['base_head']] |
| 272 | return [] |
| 273 | |
| 274 | |
| 275 | def correspondence(target, commit, plan, plan_hash, files): |
| 276 | if commit == 'absent': |
| 277 | return False |
| 278 | message = git(target, 'show', '-s', '--format=%B', commit).decode() |
| 279 | trailers = f'Muse-Source: {plan["source"]["revision"]}\nOverseer-Mirror-Plan: {plan_hash}\n' |
| 280 | if not message.endswith(trailers + '\n'): |
| 281 | return False |
| 282 | parents = git(target, 'show', '-s', '--format=%P', commit).decode().strip() |
| 283 | if parents != ' '.join(expected_parents(plan)): |
| 284 | raise Refusal('mirror_commit_parent_mismatch') |
| 285 | verify_tree(target, commit, files) |
| 286 | return True |
| 287 | |
| 288 | |
| 289 | def save_record(root, old, pair, plan, plan_hash, commit): |
| 290 | record = {'binding': pair, 'last_export': {'muse_branch': plan['source']['branch'], |
| 291 | 'muse_commit_id': plan['source']['revision'], 'git_remote': plan['destination']['url'], |
| 292 | 'git_ref': plan['destination']['branch'], 'git_sha': commit, 'plan_sha256': plan_hash, |
| 293 | 'parents': expected_parents(plan), |
| 294 | 'tree': git(Path(plan['target']), 'rev-parse', commit + '^{tree}').decode().strip()}} |
| 295 | data = packed(record) |
| 296 | if data != old: |
| 297 | atomic(root, RECORD, data, expected='absent' if old is None else digest(old)) |
| 298 | |
| 299 | |
| 300 | def execute(root, config, original, plan, plan_hash, operation, transport=None): |
| 301 | target = validate_plan(plan, root, config) |
| 302 | initial_reconciliation = 'reconciliation' in plan and plan['expected_target'] == 'absent' |
| 303 | if operation == 'reconcile' and not initial_reconciliation: |
| 304 | raise Refusal('mirror_reconcile_requires_initial_history_plan') |
| 305 | result = {'ok': False, 'repository': config['repo'], 'source': plan['source'], |
| 306 | 'target': str(target), 'destination': plan['destination'], 'plan_sha256': plan_hash, |
| 307 | 'export': 'not_prepared', 'push': 'not_attempted', 'pr': 'not_attempted', |
| 308 | 'authority': 'approved_input_not_checked_offline'} |
| 309 | if 'reconciliation' in plan: |
| 310 | result['reconciliation'] = plan['reconciliation'] |
| 311 | with locked(root, target): |
| 312 | try: |
| 313 | pair = binding(plan, config) |
| 314 | old = read(root, RECORD, missing=True) |
| 315 | if old and json.loads(old)['binding'] != pair: |
| 316 | raise Refusal('mirror_pair_changed_requires_reconciliation') |
| 317 | observed = inspect_target(target, pair) |
| 318 | projection = check_source(root, config, plan) |
| 319 | files = projection['files'] |
| 320 | result.update(excluded=projection['excluded'], files=len(files), |
| 321 | directories_omitted=projection['directories_omitted']) |
| 322 | already = observed != 'absent' and correspondence(target, observed, plan, plan_hash, files) |
| 323 | if not already and observed != plan['expected_target']: |
| 324 | raise Refusal('mirror_target_head_drift') |
| 325 | if not already and operation != ('reconcile' if initial_reconciliation else 'prepare'): |
| 326 | raise Refusal('mirror_preparation_required') |
| 327 | if not already: |
| 328 | if observed == 'absent' and target.exists(): |
| 329 | # Only an owned, unborn target left by an interrupted prepare. |
| 330 | inspect_target(target, pair) |
| 331 | elif observed == 'absent': |
| 332 | create_target(target, pair, plan, files) |
| 333 | else: |
| 334 | if not old or json.loads(old)['last_export']['git_sha'] != observed: |
| 335 | raise Refusal('mirror_target_mapping_mismatch') |
| 336 | # Refuse legacy/foreign content instead of sweeping it away. |
| 337 | previous_plan = json.loads(old)['last_export']['plan_sha256'] |
| 338 | message = git(target, 'show', '-s', '--format=%B', observed).decode() |
| 339 | if f'Overseer-Mirror-Plan: {previous_plan}\n' not in message: |
| 340 | raise Refusal('mirror_target_correspondence_missing') |
| 341 | previous = json.loads(old)['last_export'] |
| 342 | actual = git(target, 'show', '-s', '--format=%T%n%P', observed).decode().splitlines() |
| 343 | trailers = f'Muse-Source: {previous["muse_commit_id"]}\nOverseer-Mirror-Plan: {previous_plan}\n\n' |
| 344 | if actual != [previous['tree'], ' '.join(previous['parents'])] or not message.endswith(trailers): |
| 345 | raise Refusal('mirror_target_correspondence_mismatch') |
| 346 | if initial_reconciliation: |
| 347 | from cli.v1_mirror_history import verify_history |
| 348 | verify_history(target, plan, files) |
| 349 | new_tree = tree(target, files) |
| 350 | same_tree = observed != 'absent' and git(target, 'rev-parse', observed + '^{tree}').decode().strip() == new_tree |
| 351 | message = f'Mirror approved Muse snapshot\n\nMuse-Source: {plan["source"]["revision"]}\nOverseer-Mirror-Plan: {plan_hash}\n' |
| 352 | parents = [arg for parent in expected_parents(plan) for arg in ('-p', parent)] |
| 353 | commit = git(target, 'commit-tree', new_tree, *parents, data=message.encode()).decode().strip() |
| 354 | verify_tree(target, commit, files) |
| 355 | # Recheck immutable source, working projection, config and target before the CAS. |
| 356 | if check_source(root, config, plan) != projection or read(root, '.overseer/config.yaml') != original: |
| 357 | raise Refusal('mirror_source_or_config_changed') |
| 358 | if inspect_target(target, pair) != observed: |
| 359 | raise Refusal('mirror_target_changed_before_commit') |
| 360 | if initial_reconciliation: |
| 361 | verify_history(target, plan, files) |
| 362 | git(target, 'update-ref', 'refs/heads/' + plan['destination']['branch'], commit, |
| 363 | '0' * 40 if observed == 'absent' else observed) |
| 364 | result['export'] = 'mapped_no_content_change' if same_tree else 'prepared' |
| 365 | observed = commit |
| 366 | else: |
| 367 | result['export'] = 'no_change_verified' |
| 368 | if initial_reconciliation: |
| 369 | from cli.v1_mirror_history import verify_history |
| 370 | verify_history(target, plan, files) |
| 371 | result['git_head'] = observed |
| 372 | # Recover an interrupted record write using exact tree, parent and trailers. |
| 373 | if not correspondence(target, observed, plan, plan_hash, files): |
| 374 | raise Refusal('mirror_correspondence_failed') |
| 375 | if operation != 'verify': |
| 376 | save_record(root, old, pair, plan, plan_hash, observed) |
| 377 | if operation == 'deliver': |
| 378 | if transport is None: |
| 379 | raise Refusal('mirror_delivery_transport_missing') |
| 380 | deliver(result, transport, lambda: recheck(root, config, original, plan, plan_hash, pair, projection, observed)) |
| 381 | else: |
| 382 | result['ok'] = True |
| 383 | except (Refusal, OSError, ValueError, KeyError, TypeError, subprocess.SubprocessError) as exc: |
| 384 | result['error'] = str(exc) if isinstance(exc, Refusal) else type(exc).__name__ |
| 385 | return result |
| 386 | |
| 387 | |
| 388 | def recheck(root, config, original, plan, plan_hash, pair, projection, commit): |
| 389 | if read(root, '.overseer/config.yaml') != original or check_source(root, config, plan) != projection: |
| 390 | raise Refusal('mirror_source_or_config_changed') |
| 391 | if inspect_target(Path(plan['target']), pair) != commit: |
| 392 | raise Refusal('mirror_target_drift_before_delivery') |
| 393 | if not correspondence(Path(plan['target']), commit, plan, plan_hash, projection['files']): |
| 394 | raise Refusal('mirror_correspondence_failed') |
| 395 | if 'reconciliation' in plan and plan['expected_target'] == 'absent': |
| 396 | from cli.v1_mirror_history import verify_history |
| 397 | verify_history(Path(plan['target']), plan, projection['files']) |
| 398 | |
| 399 | |
| 400 | def deliver(result, transport, revalidate): |
| 401 | """Delivery state is observed every time; an earlier success flag is never proof.""" |
| 402 | source, dest, commit = result['source'], result['destination'], result['git_head'] |
| 403 | def observe_authority(): |
| 404 | result['authority'] = 'outcome_unknown' |
| 405 | value = transport.authority(source) |
| 406 | if value != (source['repo_id'], source['revision']): |
| 407 | result['authority'] = 'unknown_or_drifted' |
| 408 | raise Refusal('mirror_authority_unknown_or_drifted') |
| 409 | result['authority'] = 'observed_matching' |
| 410 | return value |
| 411 | def observe_remote(): |
| 412 | expected_base = dest.get('expected_base', result.get('reconciliation', {}).get('base_head')) |
| 413 | if expected_base is not None: |
| 414 | base = transport.remote_head(dict(dest, branch=dest['base'])) |
| 415 | result['remote_base'] = base |
| 416 | if base != expected_base: |
| 417 | raise Refusal('mirror_remote_base_drift') |
| 418 | value = transport.remote_head(dest) |
| 419 | result['remote_head'] = value |
| 420 | return value |
| 421 | observation = observe_authority() |
| 422 | revalidate() |
| 423 | remote = observe_remote() |
| 424 | if remote not in {dest['expected_head'], commit}: |
| 425 | raise Refusal('mirror_remote_head_drift') |
| 426 | if remote != commit: |
| 427 | # Repeat immediately before the write, including the authoritative source. |
| 428 | revalidate() |
| 429 | if observe_authority() != observation or observe_remote() != remote: |
| 430 | raise Refusal('mirror_remote_changed_before_push') |
| 431 | result['push'] = 'outcome_unknown' |
| 432 | try: |
| 433 | transport.push(Path(result['target']), dest, commit) |
| 434 | except (Refusal, OSError, subprocess.SubprocessError): |
| 435 | # A lost response may follow a successful push. Readback resolves it. |
| 436 | result['push'] = 'interrupted' |
| 437 | if observe_remote() != commit: |
| 438 | result['push'] = 'pending_retry' |
| 439 | raise Refusal('mirror_push_incomplete') |
| 440 | result['push'] = 'verified' |
| 441 | else: |
| 442 | result['push'] = 'already_delivered' |
| 443 | revalidate() |
| 444 | if observe_authority() != observation or observe_remote() != commit: |
| 445 | result['push'] = 'remote_drift_after_push' |
| 446 | raise Refusal('mirror_remote_changed_before_pr') |
| 447 | result['pr'] = 'pending_retry' |
| 448 | result['pr_url'] = transport.ensure_pr(Path(result['target']), dest, source, commit) |
| 449 | result['pr'] = 'verified' |
| 450 | result['ok'] = True |
| 451 | |
| 452 | |
| 453 | class NetworkDelivery: |
| 454 | """Explicit delivery adapter; tests substitute transport, never real remotes. |
| 455 | |
| 456 | Only public staging JSON refs are supported. Private/authenticated MuseHub and |
| 457 | production trust are outside R2. No automatic login, trust reset or URL fallback. |
| 458 | """ |
| 459 | def __init__(self, gh): |
| 460 | self.gh = physical(gh) |
| 461 | if not self.gh.is_file() or not os.access(self.gh, os.X_OK): |
| 462 | raise Refusal('mirror_gh_executable_required') |
| 463 | |
| 464 | def authority(self, source): |
| 465 | class NoRedirect(urllib.request.HTTPRedirectHandler): |
| 466 | def redirect_request(self, *args, **kwargs): |
| 467 | raise Refusal('mirror_authority_redirect_refused') |
| 468 | request = urllib.request.Request(source['hub_url'] + '/refs', headers={'Accept': 'application/json'}) |
| 469 | with urllib.request.build_opener(NoRedirect).open(request, timeout=15) as response: |
| 470 | raw = response.read(2 * 1024 * 1024 + 1) |
| 471 | if len(raw) > 2 * 1024 * 1024: |
| 472 | raise Refusal('mirror_authority_response_too_large') |
| 473 | value = json.loads(raw, object_pairs_hook=unique_json) |
| 474 | if not isinstance(value, dict) or value.get('domain') != 'code': |
| 475 | raise Refusal('mirror_authority_domain_mismatch') |
| 476 | heads = value.get('branch_heads') |
| 477 | if not isinstance(heads, dict): |
| 478 | raise Refusal('mirror_authority_refs_invalid') |
| 479 | return value.get('repo_id'), heads.get(source['branch']) |
| 480 | |
| 481 | def remote_head(self, dest): |
| 482 | raw = git(Path('/'), 'ls-remote', '--refs', dest['url'], 'refs/heads/' + dest['branch'], |
| 483 | options=self.git_options(dest)) |
| 484 | rows = raw.decode().splitlines() |
| 485 | if not rows: |
| 486 | return 'absent' |
| 487 | fields = rows[0].split('\t') |
| 488 | if len(rows) != 1 or len(fields) != 2 or fields[1] != 'refs/heads/' + dest['branch'] or not HEX.fullmatch(fields[0]): |
| 489 | raise Refusal('mirror_remote_response_invalid') |
| 490 | return rows[0].split('\t')[0] |
| 491 | |
| 492 | def push(self, target, dest, commit): |
| 493 | git(target, 'push', '--porcelain', dest['url'], commit + ':refs/heads/' + dest['branch'], |
| 494 | options=self.git_options(dest)) |
| 495 | |
| 496 | def git_options(self, dest): |
| 497 | repository = dest['url'].removeprefix('https://github.com/').removesuffix('.git') |
| 498 | if (not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9_.-]*/[A-Za-z0-9][A-Za-z0-9_.-]*', repository) |
| 499 | or dest['url'] != 'https://github.com/' + repository + '.git' |
| 500 | or dest.get('repository', repository) != repository): |
| 501 | raise Refusal('mirror_credential_destination_invalid') |
| 502 | helper = Path(__file__).with_name('v1_git_credential.py') |
| 503 | command = '!' + shlex.join([sys.executable, '-I', '-B', str(helper), str(self.gh), repository]) |
| 504 | # Reset helpers, constrain protocol/path, and disable redirects. Tokens never |
| 505 | # enter argv, URLs, persistent config, logs or the correspondence record. |
| 506 | values = ('credential.helper=', 'credential.useHttpPath=true', |
| 507 | 'credential.' + dest['url'] + '.helper=' + command, |
| 508 | 'http.followRedirects=false', 'protocol.allow=never', 'protocol.https.allow=always') |
| 509 | return tuple(arg for value in values for arg in ('-c', value)) |
| 510 | |
| 511 | def gh_run(self, target, *args): |
| 512 | proc = subprocess.run([str(self.gh), *args], cwd=target, env=env(), capture_output=True, timeout=30) |
| 513 | if proc.returncode: |
| 514 | raise Refusal('mirror_pr_command_failed') |
| 515 | return proc.stdout |
| 516 | |
| 517 | def ensure_pr(self, target, dest, source, commit): |
| 518 | args = ('pr', 'list', '--repo', dest['repository'], '--head', dest['branch'], '--base', dest['base'], |
| 519 | '--state', 'open', '--json', 'url,headRefOid,isCrossRepository') |
| 520 | def observe(): |
| 521 | rows = json.loads(self.gh_run(target, *args), object_pairs_hook=unique_json) |
| 522 | if not isinstance(rows, list): |
| 523 | raise Refusal('mirror_pr_response_invalid') |
| 524 | if len(rows) > 1: |
| 525 | raise Refusal('mirror_pr_ambiguous') |
| 526 | if not rows: |
| 527 | return None |
| 528 | row = rows[0] |
| 529 | if (not isinstance(row, dict) or set(row) != {'url', 'headRefOid', 'isCrossRepository'} |
| 530 | or not isinstance(row['headRefOid'], str) or not HEX.fullmatch(row['headRefOid']) |
| 531 | or type(row['isCrossRepository']) is not bool): |
| 532 | raise Refusal('mirror_pr_response_invalid') |
| 533 | if row['headRefOid'] != commit or row['isCrossRepository']: |
| 534 | raise Refusal('mirror_pr_head_mismatch') |
| 535 | url = row['url'] |
| 536 | if not isinstance(url, str) or not re.fullmatch( |
| 537 | re.escape('https://github.com/' + dest['repository'] + '/pull/') + r'[1-9][0-9]*', url): |
| 538 | raise Refusal('mirror_pr_not_verified') |
| 539 | return url |
| 540 | url = observe() |
| 541 | if url is None: |
| 542 | # A file preserves literal newlines and avoids shell interpretation. |
| 543 | with tempfile.NamedTemporaryFile(mode='w', encoding='utf-8') as body: |
| 544 | body.write(f'Approved Muse snapshot: {source["revision"]}\nSource: {source["hub_url"]}\nGit mirror: {commit}\n') |
| 545 | body.flush() |
| 546 | self.gh_run(target, 'pr', 'create', '--repo', dest['repository'], '--head', dest['branch'], |
| 547 | '--base', dest['base'], '--title', 'Mirror approved Muse snapshot', '--body-file', body.name) |
| 548 | url = observe() |
| 549 | if not isinstance(url, str) or not re.fullmatch( |
| 550 | re.escape('https://github.com/' + dest['repository'] + '/pull/') + r'[1-9][0-9]*', url): |
| 551 | raise Refusal('mirror_pr_not_verified') |
| 552 | return url |
| 553 | |
| 554 | |
| 555 | def command(root, config, original, args): |
| 556 | raw = read(root, args.plan_file) |
| 557 | if digest(raw) != args.expect_plan: |
| 558 | raise Refusal('mirror_plan_digest_mismatch') |
| 559 | plan = json.loads(raw, object_pairs_hook=unique_json) |
| 560 | try: |
| 561 | validate_plan(plan, root, config) |
| 562 | except (TypeError, KeyError, AttributeError): |
| 563 | raise Refusal('mirror_plan_invalid') from None |
| 564 | transport = NetworkDelivery(args.gh) if args.operation == 'deliver' else None |
| 565 | return execute(root, config, original, plan, digest(raw), args.operation, transport) |