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