"""Controlled snapshot mirrors. Preparation is offline; delivery is explicit. One source checkout, source main branch and Git destination pair. Only a newly created, dedicated bare Git target or this engine's own target can be used. Git plumbing avoids checkout filters, index state, source hooks and native export side effects. A single correspondence record plus commit trailers supports retry. """ from __future__ import annotations import base64 from contextlib import contextmanager import fcntl import json import os from pathlib import Path import re import shlex import stat import subprocess import sys import tempfile import urllib.request from cli.v1_io import Refusal, atomic, digest, read from cli.v1_revision import OID, bounded_run, revision, unique_json RECORD = '.overseer/local/git-bridge.json' MARKER = 'overseer-mirror.json' HEX = re.compile(r'[0-9a-f]{40}\Z') CONFIG = b'[core]\n\trepositoryformatversion = 0\n\tfilemode = true\n\tbare = true\n\thooksPath = /dev/null\n\tlogAllRefUpdates = false\n' def packed(value): return (json.dumps(value, sort_keys=True, indent=2) + '\n').encode() def env(): result = {k: v for k, v in os.environ.items() if not k.startswith(('GIT_', 'MUSE_', 'PYTHON')) and k not in {'VIRTUAL_ENV', 'SSH_ASKPASS', 'SSH_ASKPASS_REQUIRE', 'GH_DEBUG'}} result.update(PATH='/usr/bin:/bin', GIT_CONFIG_NOSYSTEM='1', GIT_CONFIG_GLOBAL='/dev/null', GIT_TERMINAL_PROMPT='0', GIT_ASKPASS='/usr/bin/false', GIT_OPTIONAL_LOCKS='0', LC_ALL='C', GH_HOST='github.com', GH_PROMPT_DISABLED='1', GIT_AUTHOR_NAME='Overseer mirror', GIT_AUTHOR_EMAIL='mirror@example.invalid', GIT_COMMITTER_NAME='Overseer mirror', GIT_COMMITTER_EMAIL='mirror@example.invalid') return result def git(target, *args, data=None, optional=False, options=()): proc = subprocess.run(['/usr/bin/git', '--no-replace-objects', '-c', 'core.hooksPath=/dev/null', '-c', 'core.fsmonitor=false', '-c', 'gc.auto=0', *options, '-C', str(target), *args], input=data, env=env(), capture_output=True, timeout=30) if proc.returncode and not optional: raise Refusal('mirror_git_failed: ' + args[0]) return proc.stdout if not proc.returncode else b'' def head(target): return git(target, 'rev-parse', '--verify', 'HEAD', optional=True).decode().strip() or 'absent' def physical(value): if not isinstance(value, str): raise Refusal('mirror_path_requires_physical_absolute_root') path = Path(value) if not path.is_absolute() or path.resolve() != path or any(p.is_symlink() for p in (path, *path.parents)): raise Refusal('mirror_path_requires_physical_absolute_root') return path def validate_plan(plan, root, config): from cli.v1 import keys, KIT fields = ('schema', 'source', 'destination', 'target', 'expected_target', 'executables') keys(plan, (*fields, 'reconciliation') if 'reconciliation' in plan else fields) keys(plan['source'], ('repo_id', 'branch', 'revision', 'hub_url')) dest_fields = ('url', 'repository', 'branch', 'base', 'expected_head') keys(plan['destination'], (*dest_fields, 'expected_base') if 'expected_base' in plan['destination'] else dest_fields) source, dest = plan['source'], plan['destination'] if plan['schema'] != 1 or config['vcs'] != 'muse': raise Refusal('mirror_requires_explicit_muse_authority') if source['branch'] != 'main' or not OID.fullmatch(source['revision']): raise Refusal('mirror_requires_approved_main_revision') if source['repo_id'] != config['muse']['repo_id']: raise Refusal('mirror_source_identity_mismatch') # Public staging is the tested authority. Do not bypass unresolved production trust. if not re.fullmatch(r'https://staging\.musehub\.ai/[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+', source['hub_url']): raise Refusal('mirror_authority_url_unsupported') if not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9_.-]*/[A-Za-z0-9][A-Za-z0-9_.-]*', dest['repository']): raise Refusal('mirror_github_repository_invalid') if dest['url'] != 'https://github.com/' + dest['repository'] + '.git': raise Refusal('mirror_destination_repository_mismatch') for branch in (dest['branch'], dest['base']): if not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9_/-]*', branch) or branch.endswith('/') or '//' in branch: raise Refusal('mirror_git_branch_invalid') if dest['branch'] in {'main', 'master', dest['base']}: raise Refusal('mirror_requires_distribution_branch') for expected in (plan['expected_target'], dest['expected_head']): if expected != 'absent' and not HEX.fullmatch(expected): raise Refusal('mirror_expected_head_invalid') if 'expected_base' in dest and not HEX.fullmatch(dest['expected_base']): raise Refusal('mirror_expected_base_invalid') if 'reconciliation' in plan: from cli.v1_mirror_history import validate validate(plan) elif plan['expected_target'] == 'absent' and dest['expected_head'] != 'absent': raise Refusal('mirror_legacy_destination_requires_reconciliation') ex = plan['executables'] if not isinstance(ex, list) or any(not isinstance(p, str) for p in ex) or ex != sorted(set(ex)): raise Refusal('mirror_executable_policy_invalid') target = physical(plan['target']) for protected in (root, Path(config['muse']['store']), KIT): if target.is_relative_to(protected) or protected.is_relative_to(target): raise Refusal('mirror_target_overlap') if not target.parent.is_dir(): raise Refusal('mirror_target_parent_missing') return target def binding(plan, config): pair = {'source_root': config['repo']['root'], 'checkout_id': config['repo']['id'], 'source_id': plan['source']['repo_id'], 'source_branch': plan['source']['branch'], 'hub_url': plan['source']['hub_url'], 'target': plan['target'], 'destination': {k: v for k, v in plan['destination'].items() if k not in {'expected_head', 'expected_base'}}} if 'reconciliation' in plan: pair['reconciliation'] = plan['reconciliation'] return pair @contextmanager def locked(root, target): handles = [] try: # Lock before target creation. Physical aliases share the same directory inode. for path in (root, target.parent): fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) handles.append(fd) try: fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: raise Refusal('mirror_operator_busy') from None yield finally: for fd in handles: os.close(fd) def check_source(root, config, plan): state = revision(root, config) if (state.branch, state.head, state.source_id) != ( plan['source']['branch'], plan['source']['revision'], plan['source']['repo_id']): raise Refusal('mirror_source_drift') python = config['muse']['runtime']['python'] helper = Path(__file__).with_name('v1_mirror_reader.py') proc = bounded_run([python, '-I', '-B', str(helper), str(root), json.dumps(plan['executables'])], cwd=root, env=env(), limit=48 * 1024 * 1024, timeout=30) if proc.returncode: try: reason = json.loads(proc.stdout, object_pairs_hook=unique_json)['error'] if not isinstance(reason, str) or len(reason) > 300: raise ValueError() except (ValueError, KeyError, TypeError): reason = 'invalid reader response' raise Refusal('mirror_projection_refused: ' + reason) value = json.loads(proc.stdout, object_pairs_hook=unique_json) if (value['state']['runtime'] != config['muse']['runtime'] or value['state']['head'] != state.head or value['state']['repo_id'] != state.source_id or value['state']['branch'] != state.branch or value['state']['root'] != str(root) or value['state']['store'] != config['muse']['store']): raise Refusal('mirror_source_changed') return value def inspect_target(target, pair): physical(str(target)) if not target.exists(): return 'absent' if not target.is_dir() or not list(target.iterdir()): raise Refusal('mirror_target_not_owned') if read(target, MARKER, missing=True) != packed(pair): raise Refusal('mirror_foreign_target') if read(target, 'config') != CONFIG: raise Refusal('mirror_target_config_or_hooks_changed') allowed = {'HEAD', 'config', MARKER, 'objects', 'refs'} if set(p.name for p in target.iterdir()) - allowed: raise Refusal('mirror_dirty_or_hooked_target') for path in target.rglob('*'): mode = path.lstat() if stat.S_ISLNK(mode.st_mode) or (not stat.S_ISDIR(mode.st_mode) and (not stat.S_ISREG(mode.st_mode) or mode.st_nlink != 1 or mode.st_mode & 0o7111)): raise Refusal('mirror_target_unsafe_path') rel = path.relative_to(target).as_posix() if path.is_dir() and rel not in {'objects', 'refs', 'refs/heads'}: ref_dir = 'refs/heads/' + pair['destination']['branch'] if not re.fullmatch(r'objects/[0-9a-f]{2}', rel) and not ref_dir.startswith(rel + '/'): raise Refusal('mirror_foreign_directory') if path.is_file() and rel.startswith('refs/') and rel != 'refs/heads/' + pair['destination']['branch']: raise Refusal('mirror_foreign_ref_file') if path.is_file() and rel.startswith('objects/') and not re.fullmatch(r'objects/[0-9a-f]{2}/[0-9a-f]{38}', rel): raise Refusal('mirror_foreign_object_or_alternate') branch = pair['destination']['branch'] if read(target, 'HEAD') != ('ref: refs/heads/' + branch + '\n').encode(): raise Refusal('mirror_target_branch_changed') refs = git(target, 'for-each-ref', '--format=%(refname)').decode().splitlines() if set(refs) - {'refs/heads/' + branch}: raise Refusal('mirror_foreign_refs') return head(target) def create_target(target, pair, plan=None, files=None): # Initialize a sibling first: a killed init never leaves an ambiguous target. with tempfile.TemporaryDirectory(prefix='.overseer-mirror-', dir=target.parent) as name: stage = Path(name) git(stage, 'init', '--bare', '--template=', '--initial-branch=' + pair['destination']['branch']) (stage / 'config').write_bytes(CONFIG) # Git init's empty metadata directories are unnecessary for loose objects. (stage / 'objects/info').rmdir() (stage / 'objects/pack').rmdir() (stage / 'refs/tags').rmdir() (stage / MARKER).write_bytes(packed(pair)) if plan and 'reconciliation' in plan: from cli.v1_mirror_history import import_history import_history(stage, plan, files) inspect_target(stage, pair) if target.exists(): raise Refusal('mirror_target_appeared') os.rename(stage, target) def tree(target, files): entries = {} for path, item in files.items(): node = entries parts = path.split('/') for part in parts[:-1]: node = node.setdefault(part, {}) data = base64.b64decode(item['data'], validate=True) oid = git(target, 'hash-object', '-w', '--stdin', data=data).decode().strip() node[parts[-1]] = (item['mode'], oid) def write(node): rows = [] for name, item in sorted(node.items()): mode, oid, kind = ('040000', write(item), 'tree') if isinstance(item, dict) else (*item, 'blob') rows.append(f'{mode} {kind} {oid}\t{name}'.encode() + b'\0') return git(target, 'mktree', '-z', data=b''.join(rows)).decode().strip() return write(entries) def verify_tree(target, commit, files): found = {} for row in git(target, 'ls-tree', '-r', '-z', commit).split(b'\0'): if not row: continue meta, path = row.split(b'\t', 1) mode, kind, oid = meta.decode().split() path = path.decode() if kind != 'blob' or path not in files or mode != files[path]['mode']: raise Refusal('mirror_tree_paths_or_modes_mismatch') content = git(target, 'cat-file', 'blob', oid) if content != base64.b64decode(files[path]['data'], validate=True): raise Refusal('mirror_tree_bytes_mismatch') found[path] = True if set(found) != set(files): raise Refusal('mirror_tree_paths_mismatch') def expected_parents(plan): if plan['expected_target'] != 'absent': return [plan['expected_target']] if 'reconciliation' in plan: r = plan['reconciliation'] return [r['mirror_head'], r['base_head']] return [] def correspondence(target, commit, plan, plan_hash, files): if commit == 'absent': return False message = git(target, 'show', '-s', '--format=%B', commit).decode() trailers = f'Muse-Source: {plan["source"]["revision"]}\nOverseer-Mirror-Plan: {plan_hash}\n' if not message.endswith(trailers + '\n'): return False parents = git(target, 'show', '-s', '--format=%P', commit).decode().strip() if parents != ' '.join(expected_parents(plan)): raise Refusal('mirror_commit_parent_mismatch') verify_tree(target, commit, files) return True def save_record(root, old, pair, plan, plan_hash, commit): record = {'binding': pair, 'last_export': {'muse_branch': plan['source']['branch'], 'muse_commit_id': plan['source']['revision'], 'git_remote': plan['destination']['url'], 'git_ref': plan['destination']['branch'], 'git_sha': commit, 'plan_sha256': plan_hash, 'parents': expected_parents(plan), 'tree': git(Path(plan['target']), 'rev-parse', commit + '^{tree}').decode().strip()}} data = packed(record) if data != old: atomic(root, RECORD, data, expected='absent' if old is None else digest(old)) def execute(root, config, original, plan, plan_hash, operation, transport=None): target = validate_plan(plan, root, config) initial_reconciliation = 'reconciliation' in plan and plan['expected_target'] == 'absent' if operation == 'reconcile' and not initial_reconciliation: raise Refusal('mirror_reconcile_requires_initial_history_plan') result = {'ok': False, 'repository': config['repo'], 'source': plan['source'], 'target': str(target), 'destination': plan['destination'], 'plan_sha256': plan_hash, 'export': 'not_prepared', 'push': 'not_attempted', 'pr': 'not_attempted', 'authority': 'approved_input_not_checked_offline'} if 'reconciliation' in plan: result['reconciliation'] = plan['reconciliation'] with locked(root, target): try: pair = binding(plan, config) old = read(root, RECORD, missing=True) if old and json.loads(old)['binding'] != pair: raise Refusal('mirror_pair_changed_requires_reconciliation') observed = inspect_target(target, pair) projection = check_source(root, config, plan) files = projection['files'] result.update(excluded=projection['excluded'], files=len(files), directories_omitted=projection['directories_omitted']) already = observed != 'absent' and correspondence(target, observed, plan, plan_hash, files) if not already and observed != plan['expected_target']: raise Refusal('mirror_target_head_drift') if not already and operation != ('reconcile' if initial_reconciliation else 'prepare'): raise Refusal('mirror_preparation_required') if not already: if observed == 'absent' and target.exists(): # Only an owned, unborn target left by an interrupted prepare. inspect_target(target, pair) elif observed == 'absent': create_target(target, pair, plan, files) else: if not old or json.loads(old)['last_export']['git_sha'] != observed: raise Refusal('mirror_target_mapping_mismatch') # Refuse legacy/foreign content instead of sweeping it away. previous_plan = json.loads(old)['last_export']['plan_sha256'] message = git(target, 'show', '-s', '--format=%B', observed).decode() if f'Overseer-Mirror-Plan: {previous_plan}\n' not in message: raise Refusal('mirror_target_correspondence_missing') previous = json.loads(old)['last_export'] actual = git(target, 'show', '-s', '--format=%T%n%P', observed).decode().splitlines() trailers = f'Muse-Source: {previous["muse_commit_id"]}\nOverseer-Mirror-Plan: {previous_plan}\n\n' if actual != [previous['tree'], ' '.join(previous['parents'])] or not message.endswith(trailers): raise Refusal('mirror_target_correspondence_mismatch') if initial_reconciliation: from cli.v1_mirror_history import verify_history verify_history(target, plan, files) new_tree = tree(target, files) same_tree = observed != 'absent' and git(target, 'rev-parse', observed + '^{tree}').decode().strip() == new_tree message = f'Mirror approved Muse snapshot\n\nMuse-Source: {plan["source"]["revision"]}\nOverseer-Mirror-Plan: {plan_hash}\n' parents = [arg for parent in expected_parents(plan) for arg in ('-p', parent)] commit = git(target, 'commit-tree', new_tree, *parents, data=message.encode()).decode().strip() verify_tree(target, commit, files) # Recheck immutable source, working projection, config and target before the CAS. if check_source(root, config, plan) != projection or read(root, '.overseer/config.yaml') != original: raise Refusal('mirror_source_or_config_changed') if inspect_target(target, pair) != observed: raise Refusal('mirror_target_changed_before_commit') if initial_reconciliation: verify_history(target, plan, files) git(target, 'update-ref', 'refs/heads/' + plan['destination']['branch'], commit, '0' * 40 if observed == 'absent' else observed) result['export'] = 'mapped_no_content_change' if same_tree else 'prepared' observed = commit else: result['export'] = 'no_change_verified' if initial_reconciliation: from cli.v1_mirror_history import verify_history verify_history(target, plan, files) result['git_head'] = observed # Recover an interrupted record write using exact tree, parent and trailers. if not correspondence(target, observed, plan, plan_hash, files): raise Refusal('mirror_correspondence_failed') if operation != 'verify': save_record(root, old, pair, plan, plan_hash, observed) if operation == 'deliver': if transport is None: raise Refusal('mirror_delivery_transport_missing') deliver(result, transport, lambda: recheck(root, config, original, plan, plan_hash, pair, projection, observed)) else: result['ok'] = True except (Refusal, OSError, ValueError, KeyError, TypeError, subprocess.SubprocessError) as exc: result['error'] = str(exc) if isinstance(exc, Refusal) else type(exc).__name__ return result def recheck(root, config, original, plan, plan_hash, pair, projection, commit): if read(root, '.overseer/config.yaml') != original or check_source(root, config, plan) != projection: raise Refusal('mirror_source_or_config_changed') if inspect_target(Path(plan['target']), pair) != commit: raise Refusal('mirror_target_drift_before_delivery') if not correspondence(Path(plan['target']), commit, plan, plan_hash, projection['files']): raise Refusal('mirror_correspondence_failed') if 'reconciliation' in plan and plan['expected_target'] == 'absent': from cli.v1_mirror_history import verify_history verify_history(Path(plan['target']), plan, projection['files']) def deliver(result, transport, revalidate): """Delivery state is observed every time; an earlier success flag is never proof.""" source, dest, commit = result['source'], result['destination'], result['git_head'] def observe_authority(): result['authority'] = 'outcome_unknown' value = transport.authority(source) if value != (source['repo_id'], source['revision']): result['authority'] = 'unknown_or_drifted' raise Refusal('mirror_authority_unknown_or_drifted') result['authority'] = 'observed_matching' return value def observe_remote(): expected_base = dest.get('expected_base', result.get('reconciliation', {}).get('base_head')) if expected_base is not None: base = transport.remote_head(dict(dest, branch=dest['base'])) result['remote_base'] = base if base != expected_base: raise Refusal('mirror_remote_base_drift') value = transport.remote_head(dest) result['remote_head'] = value return value observation = observe_authority() revalidate() remote = observe_remote() if remote not in {dest['expected_head'], commit}: raise Refusal('mirror_remote_head_drift') if remote != commit: # Repeat immediately before the write, including the authoritative source. revalidate() if observe_authority() != observation or observe_remote() != remote: raise Refusal('mirror_remote_changed_before_push') result['push'] = 'outcome_unknown' try: transport.push(Path(result['target']), dest, commit) except (Refusal, OSError, subprocess.SubprocessError): # A lost response may follow a successful push. Readback resolves it. result['push'] = 'interrupted' if observe_remote() != commit: result['push'] = 'pending_retry' raise Refusal('mirror_push_incomplete') result['push'] = 'verified' else: result['push'] = 'already_delivered' revalidate() if observe_authority() != observation or observe_remote() != commit: result['push'] = 'remote_drift_after_push' raise Refusal('mirror_remote_changed_before_pr') result['pr'] = 'pending_retry' result['pr_url'] = transport.ensure_pr(Path(result['target']), dest, source, commit) result['pr'] = 'verified' result['ok'] = True class NetworkDelivery: """Explicit delivery adapter; tests substitute transport, never real remotes. Only public staging JSON refs are supported. Private/authenticated MuseHub and production trust are outside R2. No automatic login, trust reset or URL fallback. """ def __init__(self, gh): self.gh = physical(gh) if not self.gh.is_file() or not os.access(self.gh, os.X_OK): raise Refusal('mirror_gh_executable_required') def authority(self, source): class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, *args, **kwargs): raise Refusal('mirror_authority_redirect_refused') request = urllib.request.Request(source['hub_url'] + '/refs', headers={'Accept': 'application/json'}) with urllib.request.build_opener(NoRedirect).open(request, timeout=15) as response: raw = response.read(2 * 1024 * 1024 + 1) if len(raw) > 2 * 1024 * 1024: raise Refusal('mirror_authority_response_too_large') value = json.loads(raw, object_pairs_hook=unique_json) if not isinstance(value, dict) or value.get('domain') != 'code': raise Refusal('mirror_authority_domain_mismatch') heads = value.get('branch_heads') if not isinstance(heads, dict): raise Refusal('mirror_authority_refs_invalid') return value.get('repo_id'), heads.get(source['branch']) def remote_head(self, dest): raw = git(Path('/'), 'ls-remote', '--refs', dest['url'], 'refs/heads/' + dest['branch'], options=self.git_options(dest)) rows = raw.decode().splitlines() if not rows: return 'absent' fields = rows[0].split('\t') if len(rows) != 1 or len(fields) != 2 or fields[1] != 'refs/heads/' + dest['branch'] or not HEX.fullmatch(fields[0]): raise Refusal('mirror_remote_response_invalid') return rows[0].split('\t')[0] def push(self, target, dest, commit): git(target, 'push', '--porcelain', dest['url'], commit + ':refs/heads/' + dest['branch'], options=self.git_options(dest)) def git_options(self, dest): repository = dest['url'].removeprefix('https://github.com/').removesuffix('.git') if (not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9_.-]*/[A-Za-z0-9][A-Za-z0-9_.-]*', repository) or dest['url'] != 'https://github.com/' + repository + '.git' or dest.get('repository', repository) != repository): raise Refusal('mirror_credential_destination_invalid') helper = Path(__file__).with_name('v1_git_credential.py') command = '!' + shlex.join([sys.executable, '-I', '-B', str(helper), str(self.gh), repository]) # Reset helpers, constrain protocol/path, and disable redirects. Tokens never # enter argv, URLs, persistent config, logs or the correspondence record. values = ('credential.helper=', 'credential.useHttpPath=true', 'credential.' + dest['url'] + '.helper=' + command, 'http.followRedirects=false', 'protocol.allow=never', 'protocol.https.allow=always') return tuple(arg for value in values for arg in ('-c', value)) def gh_run(self, target, *args): proc = subprocess.run([str(self.gh), *args], cwd=target, env=env(), capture_output=True, timeout=30) if proc.returncode: raise Refusal('mirror_pr_command_failed') return proc.stdout def ensure_pr(self, target, dest, source, commit): args = ('pr', 'list', '--repo', dest['repository'], '--head', dest['branch'], '--base', dest['base'], '--state', 'open', '--json', 'url,headRefOid,isCrossRepository') def observe(): rows = json.loads(self.gh_run(target, *args), object_pairs_hook=unique_json) if not isinstance(rows, list): raise Refusal('mirror_pr_response_invalid') if len(rows) > 1: raise Refusal('mirror_pr_ambiguous') if not rows: return None row = rows[0] if (not isinstance(row, dict) or set(row) != {'url', 'headRefOid', 'isCrossRepository'} or not isinstance(row['headRefOid'], str) or not HEX.fullmatch(row['headRefOid']) or type(row['isCrossRepository']) is not bool): raise Refusal('mirror_pr_response_invalid') if row['headRefOid'] != commit or row['isCrossRepository']: raise Refusal('mirror_pr_head_mismatch') url = row['url'] if not isinstance(url, str) or not re.fullmatch( re.escape('https://github.com/' + dest['repository'] + '/pull/') + r'[1-9][0-9]*', url): raise Refusal('mirror_pr_not_verified') return url url = observe() if url is None: # A file preserves literal newlines and avoids shell interpretation. with tempfile.NamedTemporaryFile(mode='w', encoding='utf-8') as body: body.write(f'Approved Muse snapshot: {source["revision"]}\nSource: {source["hub_url"]}\nGit mirror: {commit}\n') body.flush() self.gh_run(target, 'pr', 'create', '--repo', dest['repository'], '--head', dest['branch'], '--base', dest['base'], '--title', 'Mirror approved Muse snapshot', '--body-file', body.name) url = observe() if not isinstance(url, str) or not re.fullmatch( re.escape('https://github.com/' + dest['repository'] + '/pull/') + r'[1-9][0-9]*', url): raise Refusal('mirror_pr_not_verified') return url def command(root, config, original, args): raw = read(root, args.plan_file) if digest(raw) != args.expect_plan: raise Refusal('mirror_plan_digest_mismatch') plan = json.loads(raw, object_pairs_hook=unique_json) try: validate_plan(plan, root, config) except (TypeError, KeyError, AttributeError): raise Refusal('mirror_plan_invalid') from None transport = NetworkDelivery(args.gh) if args.operation == 'deliver' else None return execute(root, config, original, plan, digest(raw), args.operation, transport)