diff --git a/.github/INTEROPERABILITY.md b/.github/INTEROPERABILITY.md new file mode 100644 index 000000000..36a44729b --- /dev/null +++ b/.github/INTEROPERABILITY.md @@ -0,0 +1,47 @@ +# Hosted network checks + +The Interoperability workflow follows normal CI runs for pull requests, pushes, schedules and manual runs. Fork pull requests are included. The controller and worker orchestration always come from the protected default branch; the candidate revision comes from the upstream CI run. + +The current network check runs four Linux canaries as two isolated pairs. It checks quarantine, permitted connections, cross-cohort denial, a forbidden port, revocation and device removal. The connected transfer stage needs endpoint services and transport orchestration for Linux, FreeBSD, NetBSD, OpenBSD, macOS, Solaris and Cygwin, covering 42 directed pairs, both client roles and SSH and daemon transports. + +## Trust boundaries + +Only the isolation controller and recovery job have OIDC permission. Workers generate temporary RSA recipients and receive encrypted, single-use quarantine keys. RSA-OAEP uses 3072-bit keys and SHA-256. Private keys and enrolment keys are removed after connection; plaintext keys never enter artifacts or job outputs. Enrolment keys expire after 900 seconds and create ephemeral nodes. + +The controller validates each node against the provider before admission. Address-specific grants use conditional policy updates. One workflow holds admission at a time; other cohorts wait in quarantine within a fixed deadline. Unexpected provider errors fail immediately. Separate GitHub-run isolation remains an unqualified case. + +Admission leases carry GitHub build-provenance attestations. Recovery verifies the repository, workflow, source ref and signing revision before using a downloaded lease. Reassigned addresses require review rather than deleting another cohort's access. Failed or cancelled runs trigger a separate recovery workflow. Missing artifacts or failed verification stop recovery and need maintainer attention. + +Candidate code is never checked out or executed by these credential-bearing jobs. The current workflow runs only trusted canaries. Native candidate tests run separately through CI. Before connected transfers execute candidate binaries, their fixtures must retain this separation and bind all endpoint identities and results to the candidate revision. + +## Deployment + +Install the trusted workflows and helpers on master in a small bootstrap PR before relying on them from the rework branch. GitHub only activates [workflow_run listeners](https://docs.github.com/en/actions/reference/workflows-and-actions/events-that-trigger-workflows#workflow_run) from the default branch. The bootstrap consists of interoperability.yml, interoperability-cleanup.yml, the interop helpers and their two test scripts. + +Write access to these files on master carries the controller's test-tailnet authority. Require maintainer review of changes to them before enabling the broader federation rules. + +The controller federated identity keeps its existing client ID and audience. Configure these fields: + +Field | Value +--- | --- +Issuer | https://token.actions.githubusercontent.com +Subject | repo:RsyncProject/rsync:ref:refs/heads/master +repository_id | 266877886 +repository_owner_id | 166180282 +event_name | workflow_run +workflow_ref | RsyncProject/rsync/.github/workflows/interoperability*.yml@refs/heads/master +runner_environment | github-hosted + +Remove the old actor_id, head_ref and job_workflow_ref restrictions. Keep write access to auth_keys, policy_file, devices:core and devices:posture_attributes. The posture scope is required by policy-file write access. Restrict tagged enrolment to tag:quarantine. No user, DNS, route or credential-administration scopes are needed. + +With no active runs, rename tag:rsync-quarantine to tag:quarantine in tagOwners and the controller credential's tag restrictions. Preserve the explicit empty acls baseline and any unrelated policy content. Revoke the obsolete worker federated identity after the brokered workflow is qualified; worker jobs have no OIDC permission. + +The workflow needs contents:read and actions:read. The source and result jobs also use statuses:write. The controller needs id-token:write and attestations:write; recovery uses id-token:write and attestations:read. All actions are pinned. The new dependency is GitHub's actions/attest v4.2.2. Artifact storage records are disabled, so artifact-metadata write access is unnecessary. + +Publish the full CI rework first so its unprivileged diagnostics exercise the new helpers before the bootstrap is reviewed. After the bootstrap merges, rerun CI on the rework branch to activate the listener. Qualify Interoperability through revocation and cleanup before merging the CI router. Once ci.yml is on the default branch and existing PRs have fresh runs, require the Interoperability commit status and the Required check from GitHub Actions in branch protection. The latter appears as CI / Required in the Actions interface. Enabling those requirements before the router is deployed would block PRs using the legacy workflows. The commit status is posted to the upstream candidate SHA. Before posting, the publisher checks for a newer CI run or attempt and rejects obsolete results. + +## Recovery + +The controller performs cleanup on ordinary failure. The separate recovery workflow handles failed and cancelled Interoperability runs using the attested lease. An unavailable or invalid lease, reassigned addresses or provider errors leave a failed recovery job. Review the exact recorded node IDs, key IDs and grants before changing the test tailnet. Never replace the entire policy to clear one run. + +For rollback, disable new Interoperability runs first. Keep the controller identity available while active runs finish or are reconciled from their leases. Remove its authority after the recorded grants, nodes and keys have been cleared, then adjust the required status setting before removing the workflows. diff --git a/.github/scripts/interop-control.py b/.github/scripts/interop-control.py new file mode 100644 index 000000000..8872e8ccf --- /dev/null +++ b/.github/scripts/interop-control.py @@ -0,0 +1,312 @@ +import argparse +import json +import os +from pathlib import Path +import re +import subprocess +import sys +import time + +from interop_enrol import seal +from interop_network import NetworkError, SLOTS, TAILNET_API, Tailnet, artifact_name, artifacts, context, download_records, fetch, decode, validate_preflight, workflow_event +from interop_policy import TAILNET, admission, check_policy, update_policy + + +class AdmissionBusy(Exception): + pass + + +def write(root, name, **data): + (root / (name + '.json')).write_text(json.dumps(dict(context(), **data)) + '\n') + + +def receipts(phase, nodes): + records = artifacts(phase, SLOTS, timeout=300) + if set(records) != set(SLOTS): + raise NetworkError('Missing worker receipt') + for slot, record in records.items(): + if record.get('slot') != slot or record.get('node_id') != nodes[slot]['node_id'] or record.get('passed') is not True: + raise NetworkError('Invalid worker receipt') + return records + + +def prepare(root, api): + validate_preflight(json.loads((root / 'preflight.json').read_text())) + nodes = artifacts('ready', SLOTS, timeout=300) + devices = api.devices() + keys = json.loads((root / 'keys.json').read_text())['keys'] + ids, ips, grants, owned = set(), set(), [], [] + for slot, node in nodes.items(): + if (node.get('slot') != slot or node.get('oidc_available') is not False + or node.get('key_id') != keys.get(slot) + or node.get('preferences') != {'RouteAll': False, 'CorpDNS': False} + or not isinstance(node.get('nonce'), str) or not re.fullmatch(r'[0-9a-f]{48}', node['nonce'])): + raise NetworkError('Invalid worker readiness') + node_id, ip = node['node_id'], node['address'] + if node_id in ids or ip in ips: + raise NetworkError('Duplicate worker identity') + ids.add(node_id) + ips.add(ip) + matches = [device for device in devices if device.get('nodeId') == node_id] + hostname = 'ci-' + os.environ['GITHUB_RUN_ID'] + '-' + os.environ['GITHUB_RUN_ATTEMPT'] + '-' + slot + if len(matches) != 1 or matches[0].get('hostname') != hostname: + raise NetworkError('Worker device does not belong to this attempt') + owned.append(dict(node_id=node_id, device_id=matches[0]['id'])) + for cohort in ('a', 'b'): + roles = (cohort + '1', cohort + '2') + manifest = dict(context(), tailnet=TAILNET, repository_id='266877886', + nodes={slot: {key: nodes[slot][key] for key in ('node_id', 'address')} for slot in roles}) + grants.append(admission(manifest, context(), devices, roles=roles)) + write(root, 'lease', nodes=nodes, grants=grants, devices=owned, keys=keys, + control_sha=os.environ['GITHUB_WORKFLOW_SHA']) + # Validate the existing policy without granting access before quarantine probes. + def unchanged(policy): + for grant in grants: + update_policy(policy, grant) + return policy + api.change(unchanged) + + +def admit(root, api, lease): + receipts('quarantine', lease['nodes']) + def add(policy): + original = policy + for grant in lease['grants']: + policy = update_policy(policy, grant) + if original.get('grants'): + raise AdmissionBusy() + return policy + # Leave time to publish admission within the workers' 600-second wait. + deadline = time.monotonic() + 480 + while True: + try: + version = api.change(add) + break + except AdmissionBusy: + if time.monotonic() >= deadline: + raise NetworkError('Another cohort did not release admission') from None + time.sleep(5) + now = time.time() + write(root, 'admission', nodes=lease['nodes'], policy=version, admitted_at=now, start_at=now + 60) + + +def revoke(api, lease): + def remove(policy): + for grant in lease['grants']: + if grant in policy.get('grants', []): + policy = update_policy(policy, grant, previous=grant, remove=True) + else: + update_policy(policy, grant) + return policy + return api.change(remove) + + +def finish(root, api, lease): + error = None + try: + if all((root / (name + '.json')).exists() for name in ('admission', 'verified', 'revocation')): + records = receipts('revoked', lease['nodes']) + window = json.loads((root / 'revocation.json').read_text()) + if any(not window['end_at'] <= record.get('finished_at', 0) <= window['end_at'] + 30 + for record in records.values()): + raise NetworkError('Revocation receipt is outside the live-listener window') + else: + raise NetworkError('Admission or revocation did not complete') + except ValueError: + error = True + finally: + try: + revoke(api, lease) + except ValueError: + error = True + for device in lease['devices']: + try: + api.delete(device['device_id']) + except ValueError: + error = True + for key_id in lease.get('keys', {}).values(): + try: + api.delete_key(key_id) + except NetworkError: + error = True + remaining = {device.get('nodeId') for device in api.devices()} + if any(device['node_id'] in remaining for device in lease['devices']): + raise NetworkError('Worker devices remain after cleanup') + if error: + raise NetworkError('Live isolation proof or cleanup did not complete') + write(root, 'result', passed=True, scope='two live cohorts in one GitHub run', + concurrent_github_runs_exercised=False, slots=list(SLOTS), finished_at=time.time(), + admission_policy=json.loads((root / 'admission.json').read_text()).get('policy'), + revocation_policy=json.loads((root / 'revocation.json').read_text()).get('policy')) + + +def broker(root, api): + public = artifacts('public', SLOTS, timeout=300) + keys, recipients = {}, {} + if len({record.get('public_key') for record in public.values()}) != len(SLOTS): + raise NetworkError('Enrolment recipients share a public key') + for slot in SLOTS: + key_id, key = api.create_key(slot) + keys[slot] = key_id + write(root, 'keys', keys=keys) + recipients[slot] = seal(root, slot, public[slot], key_id, key) + write(root, 'enrolments', recipients=recipients) + + +def publish_status(state): + event = workflow_event() + source = event['workflow_run'] + body, _ = fetch('https://api.github.com/repos/RsyncProject/rsync/actions/runs/' + str(source['id']), + os.environ['GH_TOKEN'], operation='Check upstream CI attempt') + if decode(body).get('run_attempt') != source['run_attempt']: + raise NetworkError('A newer CI attempt superseded this result') + body, _ = fetch('https://api.github.com/repos/RsyncProject/rsync/actions/workflows/ci.yml/runs?head_sha=' + + context()['source_sha'] + '&per_page=1', os.environ['GH_TOKEN'], operation='Check latest CI run') + runs = decode(body).get('workflow_runs', []) + if not runs or runs[0].get('id') != source['id']: + raise NetworkError('A newer CI run superseded this result') + description = 'CI ' + str(source['id']) + '/' + str(source['run_attempt']) + ': network isolation' + payload = dict(state=state, context='Interoperability', description=description, + target_url='https://github.com/RsyncProject/rsync/actions/runs/' + os.environ['GITHUB_RUN_ID']) + fetch('https://api.github.com/repos/RsyncProject/rsync/statuses/' + context()['source_sha'], os.environ['GH_TOKEN'], + json.dumps(payload).encode(), {'Content-Type': 'application/json'}, operation='Publish interoperability status') + + +def metadata(root): + state = context() + source = workflow_event()['workflow_run'] + body, _ = fetch('https://api.github.com/repos/RsyncProject/rsync/actions/runs/' + str(source['id']), + os.environ['GH_TOKEN'], operation='Validate upstream CI run') + run = decode(body) + if (run.get('path') != '.github/workflows/ci.yml' or run.get('name') != 'CI' + or run.get('head_sha') != state['source_sha'] or run.get('run_attempt') != source['run_attempt'] + or run.get('repository', {}).get('id') != 266877886): + raise NetworkError('Upstream CI run does not match its event') + write(root, 'context', upstream_run_id=str(source['id']), upstream_attempt=str(source['run_attempt']), + control_sha=os.environ['GITHUB_WORKFLOW_SHA']) + publish_status('pending') + + +def cleanup(root, api): + state = context() + path = root / 'lease.json' + if path.is_symlink() or path.stat().st_size > 65536: + raise NetworkError('Invalid admission lease') + lease = json.loads(path.read_text()) + control_sha = lease.get('control_sha', '') + if not isinstance(control_sha, str) or not re.fullmatch(r'[0-9a-f]{40}', control_sha): + raise NetworkError('Admission lease has an invalid control revision') + env = {key: os.environ[key] for key in ('PATH', 'HOME', 'GH_TOKEN') if key in os.environ} + env['GH_PROMPT_DISABLED'] = '1' + result = subprocess.run(['gh', 'attestation', 'verify', str(root / 'lease.json'), '--repo', 'RsyncProject/rsync', + '--signer-workflow', 'RsyncProject/rsync/.github/workflows/interoperability.yml', + '--signer-digest', control_sha, + '--source-ref', 'refs/heads/master', '--deny-self-hosted-runners'], + capture_output=True, timeout=60, env=env) + if result.returncode: + raise NetworkError('Admission lease attestation failed') + if any(lease.get(key) != value for key, value in state.items()): + raise NetworkError('Cleanup lease belongs to a different run') + members = {node['address'] for node in lease['nodes'].values()} + owners = {node['node_id'] for node in lease['nodes'].values()} + for device in api.devices(): + if members.intersection(device.get('addresses', [])) and device.get('nodeId') not in owners: + raise NetworkError('Admission addresses have been reassigned; cleanup requires review') + revoke(api, lease) + for device in lease['devices']: + api.delete(device['device_id']) + for key_id in lease['keys'].values(): + api.delete_key(key_id) + remaining = {device.get('nodeId') for device in api.devices()} + if owners.intersection(remaining): + raise NetworkError('Owned devices remain after recovery') + + +def recovery_records(root): + source = workflow_event()['workflow_run'] + run_id, attempt = str(source['id']), source['run_attempt'] + for kind, filename in (('context', 'context.json'), ('lease', 'lease.json')): + name = artifact_name(kind, attempt=attempt) + record, raw = download_records(run_id, {name: None}, timeout=60)[None] + if record.get('run_id') != run_id or record.get('run_attempt') != str(attempt): + raise NetworkError('Recovery artifact belongs to a different run') + (root / filename).write_bytes(raw) + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument('phase', choices=('metadata', 'status', 'preflight', 'broker', 'prepare', 'topology', 'admit', 'verify', 'revoke', 'finish', 'recovery-records', 'cleanup')) + parser.add_argument('--root', type=Path, required=True) + args = parser.parse_args() + root = args.root.resolve() + root.mkdir(parents=True, exist_ok=True) + try: + recovery = workflow_event()['workflow']['path'] == '.github/workflows/interoperability.yml' + if recovery != (args.phase in ('recovery-records', 'cleanup')): + raise NetworkError('Operation does not match the triggering workflow') + if args.phase == 'recovery-records': + recovery_records(root) + return 0 + context() + if args.phase == 'metadata': + metadata(root) + return 0 + if args.phase == 'status': + jobs = decode(os.environ['INTEROPERABILITY_JOBS']) + success = set(jobs) == {'checks', 'metadata', 'workers', 'controller'} and all(job.get('result') == 'success' for job in jobs.values()) + if success: + proof = artifacts('result')[None] + success = proof.get('passed') is True and proof.get('slots') == list(SLOTS) + publish_status('success' if success else 'failure') + return 0 if success else 1 + path = root / 'lease.json' + if not path.exists() and args.phase in ('revoke', 'finish'): + keys_path = root / 'keys.json' + if keys_path.exists(): + api = Tailnet() + keys = json.loads(keys_path.read_text()) + if any(keys.get(key) != value for key, value in context().items()): + raise NetworkError('Key journal belongs to a different run') + for key_id in keys['keys'].values(): + api.delete_key(key_id) + return 0 + api = Tailnet() + if args.phase == 'preflight': + api.devices() + policy, _ = api.policy() + try: + check_policy(policy) + except ValueError: + raise NetworkError('Test tailnet policy does not have the required isolation baseline') from None + write(root, 'preflight', api=TAILNET_API, readable=['acl', 'devices']) + elif args.phase == 'broker': + broker(root, api) + elif args.phase == 'cleanup': + cleanup(root, api) + elif args.phase == 'prepare': + prepare(root, api) + else: + lease = json.loads(path.read_text()) + if args.phase == 'topology': + write(root, 'topology', nodes=lease['nodes'], start_at=time.time() + 60) + elif args.phase == 'admit': + admit(root, api, lease) + elif args.phase == 'verify': + receipts('exercise', lease['nodes']) + write(root, 'verified', passed=True) + elif args.phase == 'revoke': + version = revoke(api, lease) + now = time.time() + start = now + 60 + write(root, 'revocation', policy=version, revoked_at=now, start_at=start, end_at=start + 90) + else: + finish(root, api, lease) + except (ValueError, KeyError, OSError, subprocess.SubprocessError) as error: + detail = str(error) if isinstance(error, NetworkError) else type(error).__name__ + print('Interoperability controller failed during ' + args.phase + ': ' + detail, file=sys.stderr) + return 1 + return 0 + + +if __name__ == '__main__': + raise SystemExit(main()) diff --git a/.github/scripts/interop-worker.py b/.github/scripts/interop-worker.py new file mode 100644 index 000000000..1e865e2f6 --- /dev/null +++ b/.github/scripts/interop-worker.py @@ -0,0 +1,257 @@ +import argparse +import errno +import json +import os +from pathlib import Path +import secrets +import selectors +import signal +import socket +import subprocess +import sys +import time + +from interop_network import NetworkError, SLOTS, artifacts, context, worker_authority +from interop_policy import address + + +def write(root, name, data): + (root / (name + '.json')).write_text(json.dumps(dict(context(), **data)) + '\n') + + +def connect(ip, port, nonce=None): + try: + connection = socket.create_connection((ip, port), timeout=2) + except OSError as error: + if error.errno not in (errno.ETIMEDOUT, errno.ECONNREFUSED, errno.EHOSTUNREACH, errno.ENETUNREACH, + errno.EACCES, errno.EPERM) and not isinstance(error, socket.timeout): + raise + return False + with connection: + if nonce is not None: + reply = b'' + while len(reply) < len(nonce): + part = connection.recv(len(nonce) - len(reply)) + if not part: + break + reply += part + if reply != nonce.encode(): + raise NetworkError('Canary identity mismatch') + return True + + +def health(ready): + for port in (18731, 18732): + if not connect(ready['address'], port, ready['nonce']): + raise NetworkError('Local canary is not listening') + + +def denied(ip, port): + if connect(ip, port): + raise NetworkError('Forbidden TCP connection succeeded') + + +def wait_until(value): + if type(value) not in (int, float) or not time.time() - 5 <= value <= time.time() + 180: + raise NetworkError('Invalid probe barrier') + time.sleep(max(0, value - time.time())) + + +def serve(root): + ready = json.loads((root / 'ready.json').read_text()) + with selectors.DefaultSelector() as selector: + listeners = [] + try: + for port in (18731, 18732): + listener = socket.socket() + listeners.append(listener) + listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + listener.bind((ready['address'], port)) + listener.listen(32) + listener.setblocking(False) + selector.register(listener, selectors.EVENT_READ) + (root / 'listening').touch() + deadline = time.monotonic() + 1500 + while time.monotonic() < deadline: + for key, _ in selector.select(min(2, max(0, deadline - time.monotonic()))): + with key.fileobj.accept()[0] as connection: + connection.settimeout(2) + try: + connection.sendall(ready['nonce'].encode()) + except (BrokenPipeError, ConnectionResetError, socket.timeout): + pass + finally: + for listener in listeners: + listener.close() + + +def configure_network(): + subprocess.run(['sudo', 'tailscale', 'set', '--accept-routes=false', '--accept-dns=false'], + capture_output=True, timeout=15, check=True) + result = subprocess.run(['sudo', 'tailscale', 'debug', 'prefs'], capture_output=True, timeout=15, check=True) + prefs = json.loads(result.stdout) + if any(prefs.get(key) is not False for key in ('RouteAll', 'CorpDNS')): + raise NetworkError('Route or DNS acceptance is still enabled') + return {key: prefs[key] for key in ('RouteAll', 'CorpDNS')} + + +def prepare(root, slot): + prefs = configure_network() + oidc = worker_authority() + enrolment = json.loads((root / 'enrolment.json').read_text()) + if any(enrolment.get(key) != value for key, value in dict(context(), slot=slot).items()): + raise NetworkError('Worker enrolment record is stale') + result = subprocess.run(['tailscale', 'status', '--json'], capture_output=True, timeout=15, check=True) + node = json.loads(result.stdout)['Self'] + ips = [ip for ip in node['TailscaleIPs'] if ':' not in ip] + if len(ips) != 1 or not isinstance(node['ID'], str) or not node['ID']: + raise NetworkError('Missing local Tailscale identity') + ready = dict(slot=slot, node_id=node['ID'], address=address(ips[0]), nonce=secrets.token_hex(24), + oidc_available=oidc, key_id=enrolment['key_id']) + ready['preferences'] = prefs + write(root, 'ready', ready) + env = {key: os.environ[key] for key in ('PATH', 'LANG', 'RUNNER_TRACKING_ID') if key in os.environ} + with (root / 'server.log').open('wb') as log: + child = subprocess.Popen([sys.executable, '-B', '-S', str(Path(__file__).resolve()), 'serve', slot, + '--root', str(root)], env=env, stdin=subprocess.DEVNULL, + stdout=log, stderr=subprocess.STDOUT, start_new_session=True) + start = Path('/proc') / str(child.pid) / 'stat' + (root / 'process.json').write_text(json.dumps(dict(pid=child.pid, start=start.read_text().split()[21]))) + deadline = time.monotonic() + 10 + while not (root / 'listening').exists(): + if child.poll() is not None or time.monotonic() >= deadline: + raise NetworkError('Canary did not start') + time.sleep(0.1) + health(ready) + + +def probe(root, slot, phase): + ready = json.loads((root / 'ready.json').read_text()) + topology = artifacts('topology')[None] + nodes = topology['nodes'] + if set(nodes) != set(SLOTS) or nodes[slot] != ready: + raise NetworkError('Topology does not match this worker') + for name, node in nodes.items(): + if node.get('slot') != name or address(node['address']) != node['address']: + raise NetworkError('Invalid peer identity') + peer = nodes[slot[0] + ('2' if slot[1] == '1' else '1')] + health(ready) + if phase == 'quarantine': + wait_until(topology['start_at']) + for name, node in nodes.items(): + if name != slot: + denied(node['address'], 18731) + elif phase == 'exercise': + release = artifacts('admission')[None] + if release['nodes'] != nodes: + raise NetworkError('Admission changed endpoint identities') + wait_until(release['start_at']) + deadline = time.monotonic() + 45 + while not connect(peer['address'], 18731, peer['nonce']): + if time.monotonic() >= deadline: + raise NetworkError('Permitted peer connection did not succeed') + time.sleep(1) + for name, node in nodes.items(): + if name[0] != slot[0]: + denied(node['address'], 18731) + denied(peer['address'], 18732) + if not connect(peer['address'], 18731, peer['nonce']): + raise NetworkError('Permitted connection failed after denial probes') + else: + release = artifacts('revocation')[None] + wait_until(release['start_at']) + if release['end_at'] != release['start_at'] + 90: + raise NetworkError('Invalid revocation window') + deadline = time.monotonic() + 45 + while connect(peer['address'], 18731): + if time.monotonic() >= deadline: + raise NetworkError('Revoked connection remained permitted') + time.sleep(1) + for _ in range(3): + denied(peer['address'], 18731) + remaining = release['end_at'] - time.time() + if remaining < 0: + raise NetworkError('Revocation probe exceeded the live-listener window') + time.sleep(remaining) + health(ready) + write(root, phase, dict(slot=slot, node_id=ready['node_id'], passed=True, finished_at=time.time())) + + +def process_state(pid): + try: + fields = (Path('/proc') / str(pid) / 'stat').read_text().split() + except FileNotFoundError: + return None + return fields[2], fields[21] + + +def stop(root): + path = root / 'process.json' + if not path.exists(): + return + state = json.loads(path.read_text()) + current = process_state(state['pid']) + if current is None: + return + if current[1] != state['start']: + raise NetworkError('Canary process identity changed') + try: + if os.getpgid(state['pid']) != state['pid']: + raise NetworkError('Canary process group changed') + os.killpg(state['pid'], signal.SIGTERM) + except ProcessLookupError: + return + deadline = time.monotonic() + 5 + while time.monotonic() < deadline: + current = process_state(state['pid']) + if current is None or current[0] == 'Z': + return + if current[1] != state['start']: + raise NetworkError('Canary process identity changed during shutdown') + time.sleep(0.1) + current = process_state(state['pid']) + if current is None or current[0] == 'Z': + return + if current[1] != state['start']: + raise NetworkError('Canary process identity changed before forced shutdown') + try: + os.killpg(state['pid'], signal.SIGKILL) + except ProcessLookupError: + return + deadline = time.monotonic() + 5 + while time.monotonic() < deadline: + current = process_state(state['pid']) + if current is None or current[0] == 'Z': + return + time.sleep(0.1) + raise NetworkError('Canary process did not stop') + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument('phase', choices=('prepare', 'serve', 'quarantine', 'exercise', 'revoked', 'stop')) + parser.add_argument('slot', choices=SLOTS) + parser.add_argument('--root', type=Path, required=True) + args = parser.parse_args() + root = args.root.resolve() + root.mkdir(parents=True, exist_ok=True) + try: + if args.phase == 'serve': + serve(root) + elif args.phase == 'stop': + stop(root) + else: + context() + if args.phase == 'prepare': + prepare(root, args.slot) + else: + probe(root, args.slot, args.phase) + except (ValueError, KeyError, OSError, subprocess.SubprocessError) as error: + detail = str(error) if isinstance(error, NetworkError) else type(error).__name__ + print('Interoperability worker failed during ' + args.phase + ': ' + detail, file=sys.stderr) + return 1 + return 0 + + +if __name__ == '__main__': + raise SystemExit(main()) diff --git a/.github/scripts/interop_enrol.py b/.github/scripts/interop_enrol.py new file mode 100644 index 000000000..971c84022 --- /dev/null +++ b/.github/scripts/interop_enrol.py @@ -0,0 +1,118 @@ +import argparse +import base64 +import hashlib +import json +import os +from pathlib import Path +import re +import subprocess +import sys + +from interop_network import NetworkError, SLOTS, artifacts, context, require_no_oidc + + +def openssl(args, data=None): + env = {key: os.environ[key] for key in ('PATH', 'SYSTEMROOT') if key in os.environ} + try: + result = subprocess.run(['openssl', *args], input=data, capture_output=True, env=env, timeout=30) + except (OSError, subprocess.SubprocessError): + raise NetworkError('Enrolment cryptography could not complete') from None + if result.returncode: + raise NetworkError('Enrolment cryptography failed') + return result.stdout + + +def label(slot): + value = json.dumps(dict(context(), slot=slot), sort_keys=True, separators=(',', ':')).encode() + return hashlib.sha256(value).hexdigest() + + +def protect(root): + root.mkdir(mode=0o700, parents=True, exist_ok=True) + os.chmod(root, 0o700) + + +def generate(root, slot): + require_no_oidc() + protect(root) + private = root / 'enrol-private.pem' + with private.open('xb') as stream: + os.chmod(private, 0o600) + stream.write(openssl(['genpkey', '-algorithm', 'RSA', '-pkeyopt', 'rsa_keygen_bits:3072'])) + public = openssl(['pkey', '-in', str(private), '-pubout']).decode('ascii') + record = dict(context(), slot=slot, public_key=public) + (root / 'public-key.json').write_text(json.dumps(record) + '\n') + + +def seal(root, slot, record, key_id, key): + if any(record.get(k) != v for k, v in dict(context(), slot=slot).items()): + raise NetworkError('Enrolment public key has stale context') + public = record.get('public_key') + if not isinstance(public, str) or not 100 < len(public) < 2048: + raise NetworkError('Invalid enrolment public key') + path = root / ('public-' + slot + '.pem') + path.write_text(public) + description = openssl(['pkey', '-pubin', '-in', str(path), '-text', '-noout']).decode('ascii') + if not re.search(r'^(?:RSA )?Public-Key: \(3072 bit\)$', description, re.M): + raise NetworkError('Unexpected enrolment public key size') + encrypted = openssl(['pkeyutl', '-encrypt', '-pubin', '-inkey', str(path), + '-pkeyopt', 'rsa_padding_mode:oaep', '-pkeyopt', 'rsa_oaep_md:sha256', + '-pkeyopt', 'rsa_mgf1_md:sha256', '-pkeyopt', 'rsa_oaep_label:' + label(slot)], key.encode()) + return dict(context(), slot=slot, key_id=key_id, public_key_sha256=hashlib.sha256(public.encode()).hexdigest(), + encrypted_key=base64.b64encode(encrypted).decode()) + + +def decrypt(root, slot, record): + if any(record.get(k) != v for k, v in dict(context(), slot=slot).items()): + raise NetworkError('Enrolment key has stale context') + public = json.loads((root / 'public-key.json').read_text())['public_key'] + if record.get('public_key_sha256') != hashlib.sha256(public.encode()).hexdigest(): + raise NetworkError('Enrolment key has the wrong recipient') + encoded = record.get('encrypted_key') + if not isinstance(encoded, str) or len(encoded) != 512: + raise NetworkError('Invalid encrypted enrolment key') + cipher = base64.b64decode(encoded, validate=True) + key = openssl(['pkeyutl', '-decrypt', '-inkey', str(root / 'enrol-private.pem'), + '-pkeyopt', 'rsa_padding_mode:oaep', '-pkeyopt', 'rsa_oaep_md:sha256', + '-pkeyopt', 'rsa_mgf1_md:sha256', '-pkeyopt', 'rsa_oaep_label:' + label(slot)], cipher) + if not key.startswith(b'tskey-auth-') or len(key) > 256 or any(char < 33 or char > 126 for char in key): + raise NetworkError('Invalid decrypted enrolment key') + with (root / 'enrol-key').open('xb') as stream: + os.chmod(root / 'enrol-key', 0o600) + stream.write(key) + (root / 'enrolment.json').write_text(json.dumps({k: v for k, v in record.items() if k != 'encrypted_key'}) + '\n') + + +def forget(root): + for name in ('enrol-private.pem', 'enrol-key'): + try: + (root / name).unlink() + except FileNotFoundError: + pass + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument('phase', choices=('generate', 'receive', 'forget')) + parser.add_argument('slot', choices=SLOTS) + parser.add_argument('--root', type=Path, required=True) + args = parser.parse_args() + try: + if args.phase == 'generate': + generate(args.root, args.slot) + elif args.phase == 'receive': + records = artifacts('enrol', timeout=300)[None]['recipients'] + if set(records) != set(SLOTS): + raise NetworkError('Incomplete enrolment recipients') + decrypt(args.root, args.slot, records[args.slot]) + else: + forget(args.root) + except (ValueError, KeyError, OSError) as error: + detail = str(error) if isinstance(error, NetworkError) else type(error).__name__ + print('Worker enrolment failed: ' + detail, file=sys.stderr) + return 1 + return 0 + + +if __name__ == '__main__': + raise SystemExit(main()) diff --git a/.github/scripts/interop_network.py b/.github/scripts/interop_network.py new file mode 100644 index 000000000..b5856f217 --- /dev/null +++ b/.github/scripts/interop_network.py @@ -0,0 +1,302 @@ +import io +import json +import os +import re +import time +import urllib.error +import urllib.parse +import urllib.request +import zipfile + +from interop_policy import QUARANTINE, TAILNET + +CLIENT_ID = 'TG2FEugs2S11CNTRL-kx3fW8WXMt11CNTRL' +SLOTS = ('a1', 'a2', 'b1', 'b2') +TAILNET_API = 'https://api.tailscale.com/api/v2/tailnet/-' + + +class NetworkError(ValueError): + def __init__(self, message, code=None): + super().__init__(message) + self.code = code + + +class NoRedirect(urllib.request.HTTPRedirectHandler): + def redirect_request(self, *args): + return None + + +def decode(body): + try: + result = json.loads(body) + except ValueError: + raise NetworkError('Invalid JSON response') from None + if not isinstance(result, dict): + raise NetworkError('Expected a JSON object') + return result + + +def fetch(url, token=None, data=None, headers=None, allowed_redirect=False, method=None, operation='Network request'): + options = {'Accept': 'application/json', **(headers or {})} + if token: + options['Authorization'] = 'Bearer ' + token + try: + with urllib.request.build_opener(NoRedirect()).open( + urllib.request.Request(url, data=data, headers=options, method=method), timeout=30) as response: + body = response.read(1048577) + if len(body) > 1048576: + raise NetworkError(operation + ': response exceeds limit') + return body, response.headers + except urllib.error.HTTPError as error: + error.close() + if error.code == 302 and allowed_redirect: + target = error.headers.get('Location', '') + parts = urllib.parse.urlsplit(target) + if (parts.scheme != 'https' or not (parts.hostname or '').endswith('.blob.core.windows.net') + or parts.username or parts.password or parts.port not in (None, 443)): + raise NetworkError('Unexpected artifact download host') from None + return fetch(target, operation=operation) + raise NetworkError(operation + ': HTTP ' + str(error.code), error.code) from None + except (OSError, urllib.error.URLError): + raise NetworkError(operation + ': connection failed') from None + + +def context(): + env = os.environ + event = workflow_event() + source = event['workflow_run'] + if event['workflow']['path'] == '.github/workflows/interoperability.yml': + path = os.path.join(env['RUNNER_TEMP'], 'interoperability-cleanup', 'context.json') + with open(path, 'rb') as stream: + data = stream.read(65537) + if len(data) > 65536: + raise NetworkError('Cleanup context exceeds limit') + metadata = decode(data) + if (metadata.get('run_id') != str(source['id']) or metadata.get('run_attempt') != str(source['run_attempt']) + or not re.fullmatch(r'[0-9a-f]{40}', metadata.get('control_sha', ''))): + raise NetworkError('Cleanup context does not match its trusted workflow') + result = {key: metadata[key] for key in ('run_id', 'run_attempt', 'source_sha')} + else: + result = dict(run_id=env['GITHUB_RUN_ID'], run_attempt=env['GITHUB_RUN_ATTEMPT'], source_sha=source['head_sha']) + if (any(not re.fullmatch(r'[1-9][0-9]*', result[key]) for key in ('run_id', 'run_attempt')) + or not re.fullmatch(r'[0-9a-f]{40}', result['source_sha'])): + raise NetworkError('Invalid workflow identity') + return result + + +def workflow_event(): + env = os.environ + if (env.get('GITHUB_REPOSITORY_ID') != '266877886' or env.get('GITHUB_EVENT_NAME') != 'workflow_run' + or env.get('GITHUB_REF') != 'refs/heads/master'): + raise NetworkError('Unexpected CI context') + with open(env['GITHUB_EVENT_PATH'], 'rb') as stream: + data = stream.read(2097153) + if len(data) > 2097152: + raise NetworkError('Workflow event exceeds limit') + event = decode(data) + repository = event.get('repository', {}) + source = event.get('workflow_run', {}) + path = event.get('workflow', {}).get('path') + if (repository.get('id') != 266877886 or repository.get('owner', {}).get('id') != 166180282 + or path not in ('.github/workflows/ci.yml', '.github/workflows/interoperability.yml') + or type(source.get('id')) is not int or source['id'] <= 0 + or type(source.get('run_attempt')) is not int or source['run_attempt'] <= 0 + or not re.fullmatch(r'[0-9a-f]{40}', source.get('head_sha', ''))): + raise NetworkError('Unexpected upstream workflow') + if path == '.github/workflows/ci.yml': + if (event.get('action') != 'in_progress' or source.get('name') != 'CI' + or source.get('event') not in ('pull_request', 'push', 'schedule', 'workflow_dispatch')): + raise NetworkError('Unsupported upstream CI event') + elif (event.get('action') != 'completed' or source.get('name') != 'Interoperability' + or source.get('event') != 'workflow_run'): + raise NetworkError('Unsupported cleanup event') + return event + + +def identity_token(): + context() + parts = urllib.parse.urlsplit(os.environ['ACTIONS_ID_TOKEN_REQUEST_URL']) + if (parts.scheme != 'https' or not (parts.hostname or '').endswith('.actions.githubusercontent.com') + or parts.username or parts.password or parts.port not in (None, 443)): + raise NetworkError('Unexpected OIDC endpoint') + query = [(k, v) for k, v in urllib.parse.parse_qsl(parts.query) if k != 'audience'] + query.append(('audience', 'api.tailscale.com/' + CLIENT_ID)) + url = urllib.parse.urlunsplit(parts._replace(query=urllib.parse.urlencode(query))) + data, _ = fetch(url, os.environ['ACTIONS_ID_TOKEN_REQUEST_TOKEN'], operation='GitHub OIDC request') + token = decode(data).get('value') + if not isinstance(token, str) or not token or len(token) > 32768: + raise NetworkError('Invalid OIDC response') + return token + + +def exchange(): + data = urllib.parse.urlencode({'client_id': CLIENT_ID, 'jwt': identity_token()}).encode() + body, _ = fetch('https://api.tailscale.com/api/v2/oauth/token-exchange', data=data, + headers={'Content-Type': 'application/x-www-form-urlencoded'}, + operation='Tailscale controller token exchange') + token = decode(body).get('access_token') + if not isinstance(token, str) or not token or len(token) > 32768: + raise NetworkError('Invalid federation response') + return token + + +def require_no_oidc(): + if any(os.environ.get(key) for key in ('ACTIONS_ID_TOKEN_REQUEST_URL', 'ACTIONS_ID_TOKEN_REQUEST_TOKEN')): + raise NetworkError('Worker has OIDC request variables') + + +def worker_authority(): + require_no_oidc() + validate_preflight(artifacts('preflight', timeout=300)[None]) + return False + + +def validate_preflight(record): + if (any(record.get(key) != value for key, value in context().items()) + or record.get('api') != TAILNET_API or record.get('readable') != ['acl', 'devices']): + raise NetworkError('Missing controller API positive control') + + +def artifact_name(kind, slot=None, attempt=None): + suffix = '-' + slot if slot else '' + return 'interoperability-' + (context()['run_attempt'] if attempt is None else str(attempt)) + '-' + kind + suffix + + +def artifacts(kind, slots=(None,), timeout=600): + state = context() + expected = {artifact_name(kind, slot): slot for slot in slots} + records = download_records(state['run_id'], expected, timeout) + result = {} + for slot, (record, _) in records.items(): + if any(record.get(key) != value for key, value in state.items()): + raise NetworkError('Stale rendezvous artifact') + result[slot] = record + return result + + +def download_records(run_id, expected, timeout=600): + if not isinstance(run_id, str) or not re.fullmatch(r'[1-9][0-9]*', run_id): + raise NetworkError('Invalid artifact source run') + url = ('https://api.github.com/repos/RsyncProject/rsync/actions/runs/' + + run_id + '/artifacts?per_page=100') + deadline = time.monotonic() + timeout + cached, etag = None, None + while time.monotonic() < deadline: + try: + body, headers = fetch(url, os.environ['GH_TOKEN'], headers={'If-None-Match': etag} if etag else {}, + operation='List GitHub artifacts') + except NetworkError as error: + if error.code != 304 or cached is None: + raise + records = cached + else: + listing = decode(body) + records = listing['artifacts'] + count = listing['total_count'] + if type(count) is not int or count < 0 or count > 1000: + raise NetworkError('Artifact listing exceeds rendezvous limit') + for page in range(2, (count + 99) // 100 + 1): + body, _ = fetch(url + '&page=' + str(page), os.environ['GH_TOKEN'], operation='List GitHub artifacts') + records += decode(body)['artifacts'] + records = list({item['id']: item for item in records}.values()) + cached, etag = records, headers.get('ETag') + found = {name: [item for item in records if item['name'] == name and not item['expired']] + for name in expected} + if any(len(items) > 1 for items in found.values()): + raise NetworkError('Duplicate rendezvous artifact') + if all(len(items) == 1 for items in found.values()): + result = {} + for name, items in found.items(): + artifact_id = items[0]['id'] + if type(artifact_id) is not int or artifact_id <= 0: + raise NetworkError('Invalid artifact identity') + download = 'https://api.github.com/repos/RsyncProject/rsync/actions/artifacts/' + str(artifact_id) + '/zip' + try: + data, _ = fetch(download, os.environ['GH_TOKEN'], allowed_redirect=True, operation='Download GitHub artifact') + except NetworkError as error: + if error.code != 404: + raise + # The listing can precede archive availability. + result = None + break + with zipfile.ZipFile(io.BytesIO(data)) as archive: + files = [item for item in archive.infolist() if not item.is_dir()] + if len(files) != 1 or files[0].file_size > 65536: + raise NetworkError('Invalid rendezvous artifact') + raw = archive.read(files[0]) + result[expected[name]] = decode(raw), raw + if result is not None: + return result + time.sleep(5) + raise NetworkError('Timed out waiting for rendezvous artifacts') + + +class Tailnet: + def __init__(self): + if not CLIENT_ID.startswith(TAILNET + '-'): + raise NetworkError('Controller credential does not match the configured test namespace') + self.token = exchange() + self.base = TAILNET_API + + def devices(self): + body, _ = fetch(self.base + '/devices', self.token, operation='List Tailscale devices') + devices = decode(body).get('devices') + if not isinstance(devices, list) or any(not isinstance(device, dict) for device in devices): + raise NetworkError('Invalid device list') + return devices + + def policy(self): + body, headers = fetch(self.base + '/acl', self.token, operation='Read Tailscale policy') + return decode(body), headers + + def change(self, transform): + for _ in range(5): + policy, headers = self.policy() + etag = headers.get('ETag') + if not etag: + raise NetworkError('Policy response has no concurrency token') + changed = transform(policy) + try: + _, applied = fetch(self.base + '/acl', self.token, json.dumps(changed).encode(), + {'Content-Type': 'application/hujson', 'If-Match': etag}, operation='Update Tailscale policy') + return dict(before=etag, after=applied.get('ETag')) + except NetworkError as error: + if error.code != 412: + raise + raise NetworkError('Policy changed during every admission attempt') + + def delete(self, device_id): + if not isinstance(device_id, str) or not re.fullmatch(r'[A-Za-z0-9]+', device_id): + raise NetworkError('Invalid device API identity') + try: + fetch('https://api.tailscale.com/api/v2/device/' + device_id, self.token, method='DELETE', + operation='Delete Tailscale device') + except NetworkError as error: + if error.code != 404: + raise + + def create_key(self, slot): + if slot not in SLOTS: + raise NetworkError('Unexpected enrolment role') + state = context() + description = 'ci-' + state['run_id'] + '-' + state['run_attempt'] + '-' + slot + payload = dict(expirySeconds=900, description=description, + capabilities=dict(devices=dict(create=dict(reusable=False, ephemeral=True, preauthorized=True, + tags=[QUARANTINE])))) + body, _ = fetch(self.base + '/keys', self.token, json.dumps(payload).encode(), + {'Content-Type': 'application/json'}, operation='Create quarantine key') + result = decode(body) + if (not isinstance(result.get('id'), str) or not re.fullmatch(r'[A-Za-z0-9]+', result['id']) + or not isinstance(result.get('key'), str) or not result['key'].startswith('tskey-auth-') + or len(result['key']) > 256): + raise NetworkError('Invalid quarantine key response') + return result['id'], result['key'] + + def delete_key(self, key_id): + if not isinstance(key_id, str) or not re.fullmatch(r'[A-Za-z0-9]+', key_id): + raise NetworkError('Invalid enrolment key identity') + try: + fetch(self.base + '/keys/' + key_id, self.token, method='DELETE', operation='Revoke quarantine key') + except NetworkError as error: + if error.code != 404: + raise diff --git a/.github/scripts/interop_policy.py b/.github/scripts/interop_policy.py new file mode 100644 index 000000000..ef6e949b0 --- /dev/null +++ b/.github/scripts/interop_policy.py @@ -0,0 +1,101 @@ +import copy +import ipaddress +import re + +TAILNET = 'TG2FEugs2S11CNTRL' +QUARANTINE = 'tag:quarantine' +PORTS = ['tcp:18731'] + + +def address(value): + parsed = ipaddress.ip_address(value) + if not isinstance(value, str) or not isinstance(parsed, ipaddress.IPv4Address) or parsed not in ipaddress.ip_network('100.64.0.0/10'): + raise ValueError('Expected an allocated Tailscale IPv4 address') + return str(parsed) + + +def admission(manifest, context, devices, roles): + if set(context) != {'run_id', 'run_attempt', 'source_sha'}: + raise ValueError('Unexpected workflow context') + expected = dict(context, repository_id='266877886', tailnet=TAILNET) + for key, value in expected.items(): + if manifest.get(key) != value: + raise ValueError('Readiness manifest does not match the workflow context') + for key in ('run_id', 'run_attempt'): + if not isinstance(context.get(key), str) or not re.fullmatch(r'[1-9][0-9]*', context[key]): + raise ValueError('Invalid workflow run identity') + if not isinstance(context.get('source_sha'), str) or not re.fullmatch(r'[0-9a-f]{40}', context['source_sha']): + raise ValueError('Invalid source revision') + nodes = manifest.get('nodes') + if not isinstance(nodes, dict) or set(nodes) != set(roles): + raise ValueError('Missing or unexpected endpoint role') + ids, addresses = set(), set() + for role in roles: + node = nodes[role] + if not isinstance(node, dict) or set(node) != {'node_id', 'address'}: + raise ValueError('Invalid endpoint identity') + node_id, ip = node['node_id'], address(node['address']) + if not isinstance(node_id, str) or not node_id or node_id in ids or ip in addresses: + raise ValueError('Endpoint identities are not distinct') + matches = [device for device in devices if device.get('nodeId') == node_id] + if len(matches) != 1: + raise ValueError('Endpoint is not uniquely registered in the tailnet') + device = matches[0] + if device.get('authorized') is not True or device.get('tags') != [QUARANTINE] or ip not in device.get('addresses', []): + raise ValueError('Endpoint does not match its registered quarantine identity') + ids.add(node_id) + addresses.add(ip) + ips = sorted(addresses) + return {'src': ips, 'dst': ips, 'ip': list(PORTS)} + + +def validate_grant(grant): + if not isinstance(grant, dict) or set(grant) != {'src', 'dst', 'ip'}: + raise ValueError('Unexpected network grant') + src, dst = grant['src'], grant['dst'] + if not isinstance(src, list) or not src or src != dst or grant['ip'] != PORTS: + raise ValueError('Only cohort-local fixture ports are permitted') + if len(set(src)) != len(src) or any(address(ip) != ip for ip in src): + raise ValueError('Invalid cohort addresses') + return set(src) + + +def check_policy(policy): + if policy.get('acls') != [] or policy.get('ssh', []) != []: + raise ValueError('An explicit deny-all baseline is required') + for key in ('autoApprovers', 'nodeAttrs', 'ipsets'): + if policy.get(key): + raise ValueError('Unexpected network policy capabilities') + grants = policy.get('grants', []) + if not isinstance(grants, list): + raise ValueError('Invalid grants') + allocated = set() + for existing in grants: + ips = validate_grant(existing) + if allocated & ips: + raise ValueError('Existing cohorts overlap') + allocated.update(ips) + return grants + + +def update_policy(policy, grant, previous=None, remove=False): + members = validate_grant(grant) + grants = check_policy(policy) + retained = copy.deepcopy(grants) + if previous is not None: + validate_grant(previous) + if previous != grant: + raise ValueError('The admission lease does not match this cohort') + if retained.count(previous) != 1: + raise ValueError('The recorded admission lease is missing or duplicated') + retained.remove(previous) + elif remove: + raise ValueError('Cleanup requires the recorded admission lease') + for existing in retained: + if members & set(existing['src']): + raise ValueError('Admission overlaps another cohort') + if not remove: + retained.append(copy.deepcopy(grant)) + result = copy.deepcopy(policy) + result['grants'] = retained + return result diff --git a/.github/scripts/test-interop-policy.py b/.github/scripts/test-interop-policy.py new file mode 100644 index 000000000..6990a5fc9 --- /dev/null +++ b/.github/scripts/test-interop-policy.py @@ -0,0 +1,76 @@ +import copy +import unittest + +from interop_policy import PORTS, QUARANTINE, TAILNET, admission, update_policy + +ROLES = ('a', 'b') + + +class Policy(unittest.TestCase): + def setUp(self): + self.context = dict(run_id='123', run_attempt='1', source_sha='a' * 40) + self.manifest = dict(self.context, tailnet=TAILNET, repository_id='266877886', nodes={}) + self.devices = [] + for i, role in enumerate(ROLES, 1): + ip, node_id = f'100.64.0.{i}', f'node-{i}' + self.manifest['nodes'][role] = dict(address=ip, node_id=node_id) + self.devices.append(dict(nodeId=node_id, addresses=[ip], tags=[QUARANTINE], authorized=True)) + + def test_admission(self): + grant = admission(self.manifest, self.context, self.devices, ROLES) + self.assertEqual(len(grant['src']), 2) + self.assertEqual(grant['src'], grant['dst']) + self.assertEqual(grant['ip'], PORTS) + + def test_wrong_context(self): + for field, value in dict(run_id='124', run_attempt='2', source_sha='b' * 40, + tailnet='other', repository_id='1').items(): + with self.subTest(field=field), self.assertRaises(ValueError): + admission(dict(self.manifest, **{field: value}), self.context, self.devices, ROLES) + + def test_endpoint_identity(self): + for role in ROLES: + manifest = copy.deepcopy(self.manifest) + manifest['nodes'].pop(role) + with self.subTest(role=role), self.assertRaises(ValueError): + admission(manifest, self.context, self.devices, ROLES) + for field, value in dict(address='127.0.0.1', node_id='node-2').items(): + manifest = copy.deepcopy(self.manifest) + manifest['nodes']['a'][field] = value + with self.subTest(field=field), self.assertRaises(ValueError): + admission(manifest, self.context, self.devices, ROLES) + + def test_provider_state(self): + for field, value in dict(authorized=False, tags=['tag:other'], addresses=['100.64.1.1']).items(): + devices = copy.deepcopy(self.devices) + devices[0][field] = value + with self.subTest(field=field), self.assertRaises(ValueError): + admission(self.manifest, self.context, devices, ROLES) + + def test_other_run_is_preserved(self): + grant = admission(self.manifest, self.context, self.devices, ROLES) + other = dict(src=['100.64.1.1'], dst=['100.64.1.1'], ip=list(PORTS)) + policy = dict(acls=[], grants=[other]) + updated = update_policy(policy, grant) + self.assertEqual(updated['grants'], [other, grant]) + self.assertEqual(policy['grants'], [other]) + self.assertEqual(update_policy(updated, grant, previous=grant, remove=True)['grants'], [other]) + with self.assertRaises(ValueError): + update_policy(updated, grant) + with self.assertRaises(ValueError): + update_policy(policy, grant, previous=grant, remove=True) + with self.assertRaises(ValueError): + update_policy(updated, grant, previous=other, remove=True) + with self.assertRaises(ValueError): + update_policy(dict(acls=[], grants=[other, other]), grant) + + def test_broad_access_is_rejected(self): + grant = admission(self.manifest, self.context, self.devices, ROLES) + for policy in ({}, dict(acls=[{}]), dict(acls=[], ssh=[{}]), + dict(acls=[], grants=[dict(src=['*'], dst=['*'], ip=['*'])])): + with self.subTest(policy=policy), self.assertRaises(ValueError): + update_policy(policy, grant) + + +if __name__ == '__main__': + unittest.main() diff --git a/.github/scripts/test-interop.py b/.github/scripts/test-interop.py new file mode 100644 index 000000000..e98aff063 --- /dev/null +++ b/.github/scripts/test-interop.py @@ -0,0 +1,560 @@ +import copy +import io +import json +import os +from pathlib import Path +import runpy +import socket +import subprocess +import sys +import tempfile +import threading +import time +import unittest +from unittest.mock import Mock, patch +import urllib.error +import zipfile + +import interop_network as network +import interop_enrol as enrol +from interop_policy import PORTS, QUARANTINE + +worker = runpy.run_path(str(Path(__file__).with_name('interop-worker.py'))) +control = runpy.run_path(str(Path(__file__).with_name('interop-control.py'))) + + +class Isolation(unittest.TestCase): + def setUp(self): + self.directory = tempfile.TemporaryDirectory() + self.addCleanup(self.directory.cleanup) + self.root = Path(self.directory.name) + self.event = dict(action='in_progress', repository=dict(id=266877886, owner=dict(id=166180282)), + workflow=dict(path='.github/workflows/ci.yml'), + workflow_run=dict(id=456, run_attempt=3, head_sha='a' * 40, name='CI', event='pull_request', + head_repository=dict(id=999, full_name='contributor/rsync'))) + (self.root / 'event.json').write_text(json.dumps(self.event)) + self.env = dict(GITHUB_REPOSITORY_ID='266877886', GITHUB_EVENT_NAME='workflow_run', + GITHUB_REF='refs/heads/master', GITHUB_RUN_ID='123', GITHUB_WORKFLOW_SHA='c' * 40, + GITHUB_EVENT_PATH=str(self.root / 'event.json'), RUNNER_TEMP=str(self.root), + GITHUB_RUN_ATTEMPT='2', GITHUB_SHA='c' * 40, GH_TOKEN='fixture-not-a-credential') + self.environment = patch.dict(os.environ, self.env) + self.environment.start() + self.addCleanup(self.environment.stop) + self.preflight = dict(network.context(), api=network.TAILNET_API, readable=['acl', 'devices']) + (self.root / 'preflight.json').write_text(json.dumps(self.preflight)) + (self.root / 'keys.json').write_text(json.dumps(dict(network.context(), keys={slot: slot for slot in network.SLOTS}))) + self.nodes, self.devices = {}, [] + for i, slot in enumerate(network.SLOTS, 1): + self.nodes[slot] = dict(network.context(), slot=slot, node_id='n' + str(i), address='100.64.0.' + str(i), + nonce=str(i) * 48, oidc_available=False, key_id=slot) + self.nodes[slot]['preferences'] = dict(RouteAll=False, CorpDNS=False) + self.devices.append(dict(nodeId='n' + str(i), id=str(i), hostname='ci-123-2-' + slot, + addresses=['100.64.0.' + str(i)], authorized=True, tags=[QUARANTINE])) + + def test_context(self): + self.assertEqual(network.context()['run_id'], '123') + self.assertEqual(network.context()['source_sha'], 'a' * 40) + for key, value in dict(GITHUB_EVENT_NAME='pull_request_target', GITHUB_REF='refs/pull/7/merge', + GITHUB_RUN_ID='../123', GITHUB_RUN_ATTEMPT='0').items(): + with self.subTest(key=key), patch.dict(os.environ, {key: value}), self.assertRaises(network.NetworkError): + network.context() + for event in ('pull_request', 'push', 'schedule', 'workflow_dispatch'): + self.event['workflow_run']['event'] = event + (self.root / 'event.json').write_text(json.dumps(self.event)) + self.assertEqual(network.context()['source_sha'], 'a' * 40) + + def test_untrusted_workflow_context(self): + for field, value in (('path', '.github/workflows/untrusted.yml'), ('owner', 1), ('repository', 1), ('event', 'pull_request_target')): + event = copy.deepcopy(self.event) + if field == 'path': + event['workflow']['path'] = value + elif field == 'owner': + event['repository']['owner']['id'] = value + elif field == 'repository': + event['repository']['id'] = value + else: + event['workflow_run']['event'] = value + (self.root / 'event.json').write_text(json.dumps(event)) + with self.subTest(field=field), self.assertRaises(network.NetworkError): + network.context() + + def test_encrypted_single_use_enrolment(self): + enrol.generate(self.root, 'a1') + public = json.loads((self.root / 'public-key.json').read_text()) + key = 'tskey-auth-test-fixture' + record = enrol.seal(self.root, 'a1', public, 'fixtureKey', key) + self.assertNotIn(key, json.dumps(record)) + enrol.decrypt(self.root, 'a1', record) + self.assertEqual((self.root / 'enrol-key').read_text(), key) + self.assertEqual((self.root / 'enrol-key').stat().st_mode & 0o777, 0o600) + enrol.forget(self.root) + self.assertFalse((self.root / 'enrol-private.pem').exists()) + self.assertFalse((self.root / 'enrol-key').exists()) + + def test_enrolment_rejects_replay_and_tampering(self): + enrol.generate(self.root, 'a1') + public = json.loads((self.root / 'public-key.json').read_text()) + record = enrol.seal(self.root, 'a1', public, 'fixtureKey', 'tskey-auth-test-fixture') + for field, value in dict(slot='b1', run_attempt='1', source_sha='b' * 40, public_key_sha256='0' * 64, + encrypted_key='A' * 512).items(): + with self.subTest(field=field), self.assertRaises(network.NetworkError): + enrol.decrypt(self.root, 'a1', dict(record, **{field: value})) + with patch.dict(os.environ, GITHUB_RUN_ATTEMPT='3'), self.assertRaises(network.NetworkError): + enrol.decrypt(self.root, 'a1', dict(record, run_attempt='3')) + self.assertFalse((self.root / 'enrol-key').exists()) + + def test_auth_key_scope(self): + with patch.object(network, 'exchange', return_value='fixture'), patch.object(network, 'fetch', + return_value=(b'{"id":"fixtureKey","key":"tskey-auth-test-fixture"}', {})) as fetch: + self.assertEqual(network.Tailnet().create_key('a1'), ('fixtureKey', 'tskey-auth-test-fixture')) + payload = json.loads(fetch.call_args[0][2]) + self.assertEqual(payload['expirySeconds'], 900) + self.assertEqual(payload['capabilities']['devices']['create'], + dict(reusable=False, ephemeral=True, preauthorized=True, tags=[QUARANTINE])) + + def test_status_cannot_overwrite_a_newer_run(self): + responses = [(b'{"run_attempt":3}', {}), (b'{"workflow_runs":[{"id":456}]}', {}), (b'{}', {})] + fetch = Mock(side_effect=responses) + with patch.dict(control['publish_status'].__globals__, fetch=fetch): + control['publish_status']('success') + self.assertTrue(fetch.call_args[0][0].endswith('/statuses/' + 'a' * 40)) + for responses in ([(b'{"run_attempt":4}', {})], + [(b'{"run_attempt":3}', {}), (b'{"workflow_runs":[{"id":457}]}', {})]): + fetch = Mock(side_effect=responses) + with patch.dict(control['publish_status'].__globals__, fetch=fetch), self.assertRaisesRegex(network.NetworkError, 'superseded'): + control['publish_status']('success') + + def test_cleanup_requires_attested_lease(self): + lease = dict(network.context(), nodes=self.nodes, keys={}, devices=[], grants=[], control_sha='c' * 40) + (self.root / 'lease.json').write_text(json.dumps(lease)) + class API: + def devices(api): + self.fail('API accessed before attestation verification') + with patch.object(subprocess, 'run', return_value=subprocess.CompletedProcess([], 1)): + with self.assertRaisesRegex(network.NetworkError, 'attestation'): + control['cleanup'](self.root, API()) + + def test_recovery_records_keep_the_attested_bytes(self): + state = network.context() + event = copy.deepcopy(self.event) + event.update(action='completed', workflow=dict(path='.github/workflows/interoperability.yml')) + event['workflow_run'].update(id=123, run_attempt=2, head_sha='c' * 40, name='Interoperability', event='workflow_run') + (self.root / 'event.json').write_text(json.dumps(event)) + root = self.root / 'interoperability-cleanup' + root.mkdir() + metadata = dict(state, control_sha='c' * 40) + lease = dict(metadata, nodes=self.nodes, grants=[], devices=[], keys={}) + meta_raw, lease_raw = json.dumps(metadata, indent=2).encode(), json.dumps(lease, indent=4).encode() + records = [{None: (metadata, meta_raw)}, {None: (lease, lease_raw)}] + with patch.dict(control['recovery_records'].__globals__, download_records=Mock(side_effect=records)): + control['recovery_records'](root) + self.assertEqual((root / 'lease.json').read_bytes(), lease_raw) + self.assertEqual(network.context(), state) + event['workflow_run']['event'] = 'pull_request' + (self.root / 'event.json').write_text(json.dumps(event)) + with self.assertRaises(network.NetworkError): + network.context() + + def test_cleanup_does_not_touch_reassigned_addresses(self): + (self.root / 'lease.json').write_text(json.dumps(dict(network.context(), nodes=self.nodes, keys={}, devices=[], grants=[], control_sha='c' * 40))) + class API: + def devices(api): + return [dict(nodeId='foreign', addresses=['100.64.0.1'])] + def change(api, transform): + self.fail('Foreign policy was changed') + with patch.object(subprocess, 'run', return_value=subprocess.CompletedProcess([], 0)) as verify: + with self.assertRaisesRegex(network.NetworkError, 'reassigned'): + control['cleanup'](self.root, API()) + self.assertIn('--signer-digest', verify.call_args[0][0]) + self.assertIn('--deny-self-hosted-runners', verify.call_args[0][0]) + + def test_success_requires_revocation_receipts_and_resource_removal(self): + policies = dict(before='v1', after='v2') + for name, data in dict(admission=dict(policy=policies), verified=dict(passed=True), + revocation=dict(end_at=100, policy=dict(before='v2', after='v3'))).items(): + (self.root / (name + '.json')).write_text(json.dumps(data)) + removed = [] + class API: + def change(api, transform): + transform(dict(acls=[], grants=[])) + def delete(api, node_id): + removed.append(node_id) + def delete_key(api, key_id): + removed.append(key_id) + def devices(api): + return [] + lease = dict(nodes=self.nodes, grants=[], keys=dict(a1='key1'), devices=[dict(node_id='n1', device_id='1')]) + records = {slot: dict(finished_at=100) for slot in network.SLOTS} + with patch.dict(control['finish'].__globals__, receipts=lambda *args: records): + control['finish'](self.root, API(), lease) + result = json.loads((self.root / 'result.json').read_text()) + self.assertTrue(result['passed']) + self.assertEqual(result['admission_policy'], policies) + self.assertEqual(removed, ['1', 'key1']) + + def test_network_preferences(self): + prefs = dict(RouteAll=False, CorpDNS=False, Config=dict(PrivateNodeKey='fixture-not-a-key')) + with patch.object(subprocess, 'run', return_value=subprocess.CompletedProcess([], 0, json.dumps(prefs))) as run: + self.assertEqual(worker['configure_network'](), dict(RouteAll=False, CorpDNS=False)) + self.assertEqual(run.call_args_list[0][0][0], ['sudo', 'tailscale', 'set', '--accept-routes=false', '--accept-dns=false']) + self.assertEqual(run.call_args_list[1][0][0], ['sudo', 'tailscale', 'debug', 'prefs']) + for key in ('RouteAll', 'CorpDNS'): + for value in (True, None, 0, 'false'): + with self.subTest(key=key, value=value), patch.object(subprocess, 'run', + return_value=subprocess.CompletedProcess([], 0, json.dumps(dict(prefs, **{key: value})))): + with self.assertRaises(network.NetworkError): + worker['configure_network']() + with patch.object(subprocess, 'run', side_effect=subprocess.CalledProcessError(1, ['tailscale'])): + with self.assertRaises(subprocess.CalledProcessError): + worker['configure_network']() + def fail_setup(): + raise ValueError('setup failed') + with patch.dict(worker['prepare'].__globals__, configure_network=fail_setup, + worker_authority=lambda: self.fail('Authority checked before network setup')): + with self.assertRaises(ValueError): + worker['prepare'](self.root, 'a1') + self.assertFalse((self.root / 'ready.json').exists()) + + def test_redacted_http_errors(self): + class Opener: + def open(self, *args, **kwargs): + raise urllib.error.HTTPError('https://example.test/private-token', 403, 'private body', {}, io.BytesIO(b'private body')) + with patch.object(network.urllib.request, 'build_opener', return_value=Opener()): + with self.assertRaises(network.NetworkError) as caught: + network.fetch('https://example.test', 'fixture-bearer') + self.assertEqual(str(caught.exception), 'Network request: HTTP 403') + self.assertEqual(caught.exception.code, 403) + with patch.object(network.urllib.request, 'build_opener', return_value=Opener()): + with self.assertRaisesRegex(network.NetworkError, 'Worker acl access probe: HTTP 403'): + network.fetch('https://example.test', 'fixture-bearer', operation='Worker acl access probe') + + def test_artifact_redirect_drops_authorisation(self): + requests = [] + class Response: + headers = {} + def __enter__(self): + return self + def __exit__(self, *args): + pass + def read(self, size): + return b'{}' + class Opener: + def open(self, request, **kwargs): + requests.append(request) + if len(requests) == 1: + raise urllib.error.HTTPError(request.full_url, 302, 'redirect', + {'Location': 'https://fixture.blob.core.windows.net/result?fixture=signed'}, io.BytesIO()) + return Response() + with patch.object(network.urllib.request, 'build_opener', return_value=Opener()): + network.fetch('https://api.github.com/example', 'fixture-token', allowed_redirect=True) + self.assertEqual(requests[0].get_header('Authorization'), 'Bearer fixture-token') + self.assertIsNone(requests[1].get_header('Authorization')) + + def test_cas_preserves_other_cohort(self): + own = dict(src=['100.64.0.1'], dst=['100.64.0.1'], ip=list(PORTS)) + foreign = dict(src=['100.64.1.1'], dst=['100.64.1.1'], ip=list(PORTS)) + initial, concurrent = dict(acls=[], grants=[]), dict(acls=[], grants=[foreign]) + responses = [(json.dumps(initial).encode(), {'ETag': 'v1'}), network.NetworkError('HTTP 412', 412), + (json.dumps(concurrent).encode(), {'ETag': 'v2'}), (b'{}', {'ETag': 'v3'})] + with patch.object(network, 'exchange', return_value='fixture'), patch.object(network, 'fetch', side_effect=responses) as fetch: + from interop_policy import update_policy + version = network.Tailnet().change(lambda policy: update_policy(policy, own)) + self.assertEqual(version, dict(before='v2', after='v3')) + final = fetch.call_args_list[-1] + self.assertEqual(json.loads(final[0][2])['grants'], [foreign, own]) + self.assertEqual(final[0][3]['If-Match'], 'v2') + self.assertTrue(all(call[0][0] == 'https://api.tailscale.com/api/v2/tailnet/-/acl' for call in fetch.call_args_list)) + with patch.object(network, 'exchange', return_value='fixture'), patch.object(network, 'fetch', return_value=(b'{}', {})): + with self.assertRaises(network.NetworkError): + network.Tailnet().change(lambda value: value) + + def test_controller_preflight_precedes_worker_wait(self): + calls = [] + class API: + def devices(api): + calls.append('devices') + def policy(api): + calls.append('policy') + return dict(acls=[], grants=[]), {} + args = ['interop-control.py', 'preflight', '--root', str(self.root)] + with patch.object(sys, 'argv', args), patch.dict(control['main'].__globals__, Tailnet=API, + prepare=lambda root, api: calls.append('prepare')): + self.assertEqual(control['main'](), 0) + self.assertEqual(calls, ['devices', 'policy']) + self.assertEqual(json.loads((self.root / 'preflight.json').read_text()), self.preflight) + calls.clear() + with patch.object(sys, 'argv', args), patch.dict(control['main'].__globals__, Tailnet=API, + prepare=lambda root, api: self.fail('Worker wait started after a failed preflight')), \ + patch.object(API, 'policy', side_effect=network.NetworkError('Read Tailscale policy: HTTP 404', 404)), \ + patch.object(sys, 'stderr', io.StringIO()) as stderr: + self.assertEqual(control['main'](), 1) + self.assertIn('Read Tailscale policy: HTTP 404', stderr.getvalue()) + with patch.object(sys, 'argv', args), patch.dict(control['main'].__globals__, Tailnet=API), \ + patch.object(API, 'policy', return_value=({}, {})), patch.object(sys, 'stderr', io.StringIO()): + self.assertEqual(control['main'](), 1) + with patch.object(sys, 'argv', ['interop-control.py', 'revoke', '--root', str(self.root)]), \ + patch.dict(control['main'].__globals__, Tailnet=lambda: self.fail('No lease requires no API access')): + (self.root / 'keys.json').unlink() + self.assertEqual(control['main'](), 0) + + def test_worker_has_no_oidc_authority(self): + with patch.object(network, 'artifacts', return_value={None: self.preflight}): + self.assertIs(network.worker_authority(), False) + for key in ('ACTIONS_ID_TOKEN_REQUEST_URL', 'ACTIONS_ID_TOKEN_REQUEST_TOKEN'): + with patch.dict(os.environ, {key: 'fixture'}), self.assertRaisesRegex(network.NetworkError, 'OIDC'): + network.worker_authority() + with patch.dict(os.environ, {key: 'fixture'}), self.assertRaisesRegex(network.NetworkError, 'OIDC'): + enrol.generate(self.root, 'a1') + self.assertFalse((self.root / 'enrol-private.pem').exists()) + + def test_denial_requires_current_positive_control(self): + for field, value in dict(run_attempt='1', source_sha='b' * 40, api='https://example.test', readable=['devices']).items(): + with self.subTest(field=field): + with patch.object(network, 'artifacts', return_value={None: dict(self.preflight, **{field: value})}), \ + patch.object(network, 'fetch') as fetch: + with self.assertRaisesRegex(network.NetworkError, 'positive control'): + network.worker_authority() + fetch.assert_not_called() + (self.root / 'preflight.json').write_text(json.dumps(dict(self.preflight, run_attempt='1'))) + with patch.dict(control['prepare'].__globals__, artifacts=lambda *args, **kwargs: self.fail('Stale preflight accepted')): + with self.assertRaisesRegex(network.NetworkError, 'positive control'): + control['prepare'](self.root, None) + + def test_setup_guard_messages_are_reported(self): + args = ['interop-worker.py', 'prepare', 'a1', '--root', str(self.root)] + for message in ('Worker obtained a controller credential', 'Route or DNS acceptance is still enabled'): + def fail(root, slot): + raise network.NetworkError(message) + with patch.object(sys, 'argv', args), patch.dict(worker['main'].__globals__, prepare=fail), \ + patch.object(sys, 'stderr', io.StringIO()) as stderr: + self.assertEqual(worker['main'](), 1) + self.assertIn(message, stderr.getvalue()) + + def test_policy_lifecycle(self): + foreign = dict(src=['100.64.1.1'], dst=['100.64.1.1'], ip=list(PORTS)) + class API: + policy = dict(acls=[], grants=[]) + def devices(api): + return self.devices + def change(api, transform): + api.policy = transform(copy.deepcopy(api.policy)) + api = API() + with patch.dict(control['prepare'].__globals__, artifacts=lambda *args, **kwargs: self.nodes): + control['prepare'](self.root, api) + self.assertEqual(api.policy['grants'], []) + lease = json.loads((self.root / 'lease.json').read_text()) + with patch.dict(control['admit'].__globals__, receipts=lambda *args: None): + control['admit'](self.root, api, lease) + self.assertEqual(len(api.policy['grants']), 2) + api.policy['grants'].append(foreign) + control['revoke'](api, lease) + self.assertEqual(api.policy['grants'], [foreign]) + control['revoke'](api, lease) + self.assertEqual(api.policy['grants'], [foreign]) + api.policy['acls'] = [{}] + with self.assertRaises(ValueError): + control['revoke'](api, lease) + + def test_admission_waits_without_overwriting_another_cohort(self): + own = dict(src=['100.64.0.1', '100.64.0.2'], dst=['100.64.0.1', '100.64.0.2'], ip=list(PORTS)) + foreign = dict(src=['100.64.1.1'], dst=['100.64.1.1'], ip=list(PORTS)) + class API: + calls = 0 + def change(api, transform): + api.calls += 1 + result = transform(dict(acls=[], grants=[foreign] if api.calls == 1 else [])) + self.assertEqual(result['grants'], [own]) + api = API() + with patch.dict(control['admit'].__globals__, receipts=lambda *args: None), patch.object(time, 'sleep') as sleep: + control['admit'](self.root, api, dict(nodes=self.nodes, grants=[own])) + self.assertEqual(api.calls, 2) + sleep.assert_called_once_with(5) + + def test_admission_wait_is_bounded(self): + own = dict(src=['100.64.0.1'], dst=['100.64.0.1'], ip=list(PORTS)) + foreign = dict(src=['100.64.1.1'], dst=['100.64.1.1'], ip=list(PORTS)) + class API: + def change(api, transform): + transform(dict(acls=[], grants=[foreign])) + with patch.dict(control['admit'].__globals__, receipts=lambda *args: None), \ + patch.object(time, 'monotonic', side_effect=[0, 1, 481]), patch.object(time, 'sleep'): + with self.assertRaisesRegex(network.NetworkError, 'release admission'): + control['admit'](self.root, API(), dict(nodes=self.nodes, grants=[own])) + self.assertFalse((self.root / 'admission.json').exists()) + + def test_cleanup_attempts_every_device_after_policy_failure(self): + deleted = [] + class API: + def change(api, transform): + raise network.NetworkError('HTTP 500', 500) + def delete(api, device_id): + deleted.append(device_id) + if device_id == '1': + raise network.NetworkError('HTTP 500', 500) + def devices(api): + return [] + lease = dict(nodes=self.nodes, grants=[], devices=[dict(node_id='n1', device_id='1'), dict(node_id='n2', device_id='2')]) + with self.assertRaises(ValueError): + control['finish'](self.root, API(), lease) + self.assertEqual(deleted, ['1', '2']) + self.assertFalse((self.root / 'result.json').exists()) + + def test_readiness_identity(self): + class API: + def devices(api): + return self.devices + def change(api, transform): + transform(dict(acls=[])) + for field, value in dict(nonce='', oidc_available=True, key_id='foreign', + node_id='n2', address='100.64.0.2', preferences={}).items(): + nodes = copy.deepcopy(self.nodes) + nodes['a1'][field] = value + with self.subTest(field=field), patch.dict(control['prepare'].__globals__, artifacts=lambda *args, **kwargs: nodes): + with self.assertRaises(ValueError): + control['prepare'](self.root, API()) + + def test_live_tcp(self): + with socket.socket() as listener: + listener.bind(('127.0.0.1', 0)) + listener.listen(2) + listener.settimeout(5) + port = listener.getsockname()[1] + def reply(): + for _ in range(3): + with listener.accept()[0] as peer: + peer.sendall(b'nonce') + thread = threading.Thread(target=reply) + thread.start() + try: + self.assertTrue(worker['connect']('127.0.0.1', port, 'nonce')) + with self.assertRaisesRegex(ValueError, 'Forbidden'): + worker['denied']('127.0.0.1', port) + with self.assertRaisesRegex(ValueError, 'identity mismatch'): + worker['connect']('127.0.0.1', port, 'wrong') + finally: + thread.join(5) + self.assertFalse(thread.is_alive()) + self.assertFalse(worker['connect']('127.0.0.1', port)) + with patch.dict(worker['health'].__globals__, connect=lambda *args: False): + with self.assertRaisesRegex(ValueError, 'not listening'): + worker['health'](dict(address='127.0.0.1', nonce='nonce')) + + def test_exercise_probes_both_foreign_nodes_and_wrong_port(self): + (self.root / 'ready.json').write_text(json.dumps(self.nodes['a1'])) + connections, denials = [], [] + def connected(ip, port, nonce=None): + connections.append((ip, port, nonce)) + return True + def published(kind): + return {None: dict(nodes=self.nodes, start_at=time.time())} + globals_ = worker['probe'].__globals__ + with patch.dict(globals_, artifacts=published, connect=connected, denied=lambda *args: denials.append(args), + wait_until=lambda value: None): + worker['probe'](self.root, 'a1', 'exercise') + self.assertEqual(denials, [('100.64.0.3', 18731), ('100.64.0.4', 18731), ('100.64.0.2', 18732)]) + self.assertEqual(connections.count(('100.64.0.2', 18731, '2' * 48)), 2) + self.assertEqual(connections.count(('100.64.0.1', 18732, '1' * 48)), 2) + self.assertTrue(json.loads((self.root / 'exercise.json').read_text())['passed']) + + def test_revocation_keeps_listeners_alive_through_the_barrier(self): + (self.root / 'ready.json').write_text(json.dumps(self.nodes['a1'])) + start = time.time() + connections, denials = [], [] + def connected(ip, port, nonce=None): + connections.append((ip, port)) + return ip == '100.64.0.1' + def published(kind): + return {None: dict(nodes=self.nodes, start_at=start, end_at=start + 90)} + with patch.dict(worker['probe'].__globals__, artifacts=published, connect=connected, + denied=lambda *args: denials.append(args), wait_until=lambda value: None), patch.object(time, 'sleep') as sleep: + worker['probe'](self.root, 'a1', 'revoked') + self.assertEqual(denials, [('100.64.0.2', 18731)] * 3) + self.assertTrue(89 < sleep.call_args[0][0] <= 90) + self.assertEqual(connections.count(('100.64.0.1', 18732)), 2) + + def test_late_barriers_are_rejected(self): + for value in (time.time() - 10, time.time() + 300, True, 'now', float('nan')): + with self.subTest(value=value), self.assertRaises(ValueError): + worker['wait_until'](value) + + def test_artifacts_reject_stale_duplicate_and_redirected_identity(self): + def zipped(record): + output = io.BytesIO() + with zipfile.ZipFile(output, 'w') as archive: + archive.writestr('ready.json', json.dumps(record)) + return output.getvalue() + item = dict(name=network.artifact_name('ready', 'a1'), id=1, expired=False, + archive_download_url='https://attacker.invalid') + listing = json.dumps(dict(artifacts=[item], total_count=1)).encode() + with patch.object(network, 'fetch', side_effect=[(listing, {}), (zipped(self.nodes['a1']), {})]) as fetch: + self.assertEqual(network.artifacts('ready', ('a1',))['a1'], self.nodes['a1']) + self.assertEqual(fetch.call_args[0][0], 'https://api.github.com/repos/RsyncProject/rsync/actions/artifacts/1/zip') + responses = [(listing, {}), network.NetworkError('Download GitHub artifact: HTTP 404', 404), + (listing, {}), (zipped(self.nodes['a1']), {})] + with patch.object(network, 'fetch', side_effect=responses) as fetch, patch.object(time, 'sleep') as sleep: + self.assertEqual(network.artifacts('ready', ('a1',))['a1'], self.nodes['a1']) + self.assertEqual(fetch.call_args_list[0][0][0], fetch.call_args_list[2][0][0]) + sleep.assert_called_once_with(5) + responses = [(listing, {'ETag': 'list-v1'}), network.NetworkError('Not ready', 404), + network.NetworkError('Not modified', 304), (zipped(self.nodes['a1']), {})] + with patch.object(network, 'fetch', side_effect=responses) as fetch, patch.object(time, 'sleep'): + self.assertEqual(network.artifacts('ready', ('a1',))['a1'], self.nodes['a1']) + self.assertEqual(fetch.call_args_list[2][1]['headers'], {'If-None-Match': 'list-v1'}) + for code in (401, 403, 500): + with self.subTest(code=code), patch.object(network, 'fetch', + side_effect=[(listing, {}), network.NetworkError('Download failed', code)]), patch.object(time, 'sleep') as sleep: + with self.assertRaises(network.NetworkError) as caught: + network.artifacts('ready', ('a1',)) + self.assertEqual(caught.exception.code, code) + sleep.assert_not_called() + with patch.object(network, 'fetch', side_effect=[(listing, {}), network.NetworkError('Not ready', 404)]), \ + patch.object(time, 'monotonic', side_effect=[0, 0, 2]), patch.object(time, 'sleep'): + with self.assertRaisesRegex(network.NetworkError, 'Timed out'): + network.artifacts('ready', ('a1',), timeout=1) + stale = dict(self.nodes['a1'], run_attempt='1') + with patch.object(network, 'fetch', side_effect=[(listing, {}), (zipped(stale), {})]): + with self.assertRaisesRegex(network.NetworkError, 'Stale'): + network.artifacts('ready', ('a1',)) + listing = json.dumps(dict(artifacts=[item, dict(item, id=2)], total_count=2)).encode() + with patch.object(network, 'fetch', return_value=(listing, {})): + with self.assertRaisesRegex(network.NetworkError, 'Duplicate'): + network.artifacts('ready', ('a1',)) + + def test_receipts_reject_missing_or_failed_worker(self): + records = {slot: dict(node, passed=True) for slot, node in self.nodes.items()} + with patch.dict(control['receipts'].__globals__, artifacts=lambda *args, **kwargs: {}): + with self.assertRaises(ValueError): + control['receipts']('exercise', self.nodes) + for field, value in dict(slot='b1', node_id='other', passed=False).items(): + modified = copy.deepcopy(records) + modified['a1'][field] = value + with patch.dict(control['receipts'].__globals__, artifacts=lambda *args, **kwargs: modified): + with self.assertRaises(ValueError): + control['receipts']('exercise', self.nodes) + + @unittest.skipUnless(sys.platform == 'linux', 'Linux canary process ownership') + def test_canary_lifecycle(self): + ready = dict(address='127.0.0.1', nonce='test-canary') + (self.root / 'ready.json').write_text(json.dumps(ready)) + command = [sys.executable, '-B', '-S', str(Path(__file__).with_name('interop-worker.py')), + 'serve', 'a1', '--root', str(self.root)] + with subprocess.Popen(command, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, start_new_session=True) as child: + try: + deadline = time.monotonic() + 5 + while not (self.root / 'listening').exists(): + if child.poll() is not None or time.monotonic() > deadline: + self.fail('Canary startup failed') + time.sleep(0.05) + worker['health'](ready) + stat = Path('/proc') / str(child.pid) / 'stat' + (self.root / 'process.json').write_text(json.dumps(dict(pid=child.pid, start=stat.read_text().split()[21]))) + worker['stop'](self.root) + self.assertEqual(child.wait(timeout=5), -15) + finally: + if child.poll() is None: + child.kill() + child.wait(timeout=5) + + +if __name__ == '__main__': + unittest.main() diff --git a/.github/workflows/interoperability-cleanup.yml b/.github/workflows/interoperability-cleanup.yml new file mode 100644 index 000000000..b3174f08b --- /dev/null +++ b/.github/workflows/interoperability-cleanup.yml @@ -0,0 +1,35 @@ +name: Interoperability cleanup + +on: + workflow_run: + workflows: [Interoperability] + types: [completed] + +permissions: + contents: read + actions: read + +jobs: + cleanup: + if: ${{ github.event.workflow_run.conclusion != 'success' }} + name: Reconcile admission + runs-on: ubuntu-24.04 + timeout-minutes: 10 + permissions: + contents: read + actions: read + id-token: write + attestations: read + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + ref: ${{ github.workflow_sha }} + persist-credentials: false + - name: Validate source and retrieve bounded recovery records + env: + GH_TOKEN: ${{ github.token }} + run: python3 -B -S .github/scripts/interop-control.py recovery-records --root "$RUNNER_TEMP/interoperability-cleanup" + - name: Verify lease and reconcile resources + env: + GH_TOKEN: ${{ github.token }} + run: python3 -B -S .github/scripts/interop-control.py cleanup --root "$RUNNER_TEMP/interoperability-cleanup" diff --git a/.github/workflows/interoperability.yml b/.github/workflows/interoperability.yml new file mode 100644 index 000000000..7950631d3 --- /dev/null +++ b/.github/workflows/interoperability.yml @@ -0,0 +1,253 @@ +name: Interoperability +run-name: CI ${{ github.event.workflow_run.id }} attempt ${{ github.event.workflow_run.run_attempt }} + +on: + workflow_run: + workflows: [CI] + types: [in_progress] + +permissions: + contents: read + actions: read + +concurrency: + group: interoperability-${{ github.event.workflow_run.id }}-${{ github.event.workflow_run.run_attempt }} + cancel-in-progress: false + +jobs: + checks: + name: Control-plane checks + runs-on: ubuntu-24.04 + timeout-minutes: 10 + container: python:3.7.17-bullseye@sha256:f36f6fe9ddb2622aab9803944cbe7e2bc50be75d624a081f7895e60e7c6b7f0a + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + ref: ${{ github.workflow_sha }} + persist-credentials: false + - name: Check admission, enrolment and recovery + run: | + python3 -B -S .github/scripts/test-interop-policy.py + python3 -B -S .github/scripts/test-interop.py + + metadata: + name: Source + needs: checks + runs-on: ubuntu-24.04 + timeout-minutes: 5 + permissions: + contents: read + actions: read + statuses: write + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + ref: ${{ github.workflow_sha }} + persist-credentials: false + - name: Validate source and publish pending status + env: + GH_TOKEN: ${{ github.token }} + run: python3 -B -S .github/scripts/interop-control.py metadata --root "$RUNNER_TEMP/interoperability" + - name: Upload source context + uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2 + with: + name: interoperability-${{ github.run_attempt }}-context + path: ${{ runner.temp }}/interoperability/context.json + if-no-files-found: error + retention-days: 7 + + workers: + name: Isolation (${{ matrix.slot }}) + needs: metadata + runs-on: ubuntu-24.04 + timeout-minutes: 25 + permissions: + contents: read + actions: read + strategy: + fail-fast: false + matrix: + slot: [a1, a2, b1, b2] + env: + GH_TOKEN: ${{ github.token }} + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + ref: ${{ github.workflow_sha }} + persist-credentials: false + - name: Generate enrolment recipient + run: python3 -B -S .github/scripts/interop_enrol.py generate ${{ matrix.slot }} --root "$RUNNER_TEMP/interoperability" + - name: Upload recipient key + uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2 + with: + name: interoperability-${{ github.run_attempt }}-public-${{ matrix.slot }} + path: ${{ runner.temp }}/interoperability/public-key.json + if-no-files-found: error + retention-days: 7 + - name: Receive single-use key + run: python3 -B -S .github/scripts/interop_enrol.py receive ${{ matrix.slot }} --root "$RUNNER_TEMP/interoperability" + - name: Connect quarantine node + uses: tailscale/github-action@064b10558093f38325e838ddea35486bdc6b1d5c # v4 + with: + authkey: file:${{ runner.temp }}/interoperability/enrol-key + hostname: ci-${{ github.run_id }}-${{ github.run_attempt }}-${{ matrix.slot }} + version: 1.104.4 + args: --accept-dns=false + retry: '1' + log-mode: quiet + - name: Remove enrolment secrets + if: always() + run: python3 -B -S .github/scripts/interop_enrol.py forget ${{ matrix.slot }} --root "$RUNNER_TEMP/interoperability" + - name: Start canary and check credential separation + run: python3 -B -S .github/scripts/interop-worker.py prepare ${{ matrix.slot }} --root "$RUNNER_TEMP/interoperability" + - name: Upload readiness + uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2 + with: + name: interoperability-${{ github.run_attempt }}-ready-${{ matrix.slot }} + path: ${{ runner.temp }}/interoperability/ready.json + if-no-files-found: error + retention-days: 7 + - name: Check quarantine + run: python3 -B -S .github/scripts/interop-worker.py quarantine ${{ matrix.slot }} --root "$RUNNER_TEMP/interoperability" + - name: Upload quarantine result + uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2 + with: + name: interoperability-${{ github.run_attempt }}-quarantine-${{ matrix.slot }} + path: ${{ runner.temp }}/interoperability/quarantine.json + if-no-files-found: error + retention-days: 7 + - name: Check cohort isolation + run: python3 -B -S .github/scripts/interop-worker.py exercise ${{ matrix.slot }} --root "$RUNNER_TEMP/interoperability" + - name: Upload isolation result + uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2 + with: + name: interoperability-${{ github.run_attempt }}-exercise-${{ matrix.slot }} + path: ${{ runner.temp }}/interoperability/exercise.json + if-no-files-found: error + retention-days: 7 + - name: Check revocation + run: python3 -B -S .github/scripts/interop-worker.py revoked ${{ matrix.slot }} --root "$RUNNER_TEMP/interoperability" + - name: Upload revocation result + uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2 + with: + name: interoperability-${{ github.run_attempt }}-revoked-${{ matrix.slot }} + path: ${{ runner.temp }}/interoperability/revoked.json + if-no-files-found: error + retention-days: 7 + - name: Stop canary + if: always() + run: python3 -B -S .github/scripts/interop-worker.py stop ${{ matrix.slot }} --root "$RUNNER_TEMP/interoperability" + + controller: + name: Isolation controller + needs: metadata + runs-on: ubuntu-24.04 + timeout-minutes: 40 + permissions: + contents: read + actions: read + id-token: write + attestations: write + env: + GH_TOKEN: ${{ github.token }} + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + ref: ${{ github.workflow_sha }} + persist-credentials: false + - name: Check controller API access + run: python3 -B -S .github/scripts/interop-control.py preflight --root "$RUNNER_TEMP/interoperability" + - name: Upload API positive control + uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2 + with: + name: interoperability-${{ github.run_attempt }}-preflight + path: ${{ runner.temp }}/interoperability/preflight.json + if-no-files-found: error + retention-days: 7 + - name: Issue single-use quarantine keys + run: python3 -B -S .github/scripts/interop-control.py broker --root "$RUNNER_TEMP/interoperability" + - name: Upload encrypted enrolment keys + uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2 + with: + name: interoperability-${{ github.run_attempt }}-enrol + path: ${{ runner.temp }}/interoperability/enrolments.json + if-no-files-found: error + retention-days: 7 + - name: Validate quarantine membership + run: python3 -B -S .github/scripts/interop-control.py prepare --root "$RUNNER_TEMP/interoperability" + - name: Attest admission lease + uses: actions/attest@1e69f48acb82d1966a394da916b4c1698aa569d6 # v4.2.2 + with: + subject-path: ${{ runner.temp }}/interoperability/lease.json + create-storage-record: false + - name: Upload admission lease + uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2 + if: always() + with: + name: interoperability-${{ github.run_attempt }}-lease + path: ${{ runner.temp }}/interoperability/lease.json + if-no-files-found: error + retention-days: 7 + - name: Schedule quarantine probes + run: python3 -B -S .github/scripts/interop-control.py topology --root "$RUNNER_TEMP/interoperability" + - name: Upload quarantine topology + uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2 + with: + name: interoperability-${{ github.run_attempt }}-topology + path: ${{ runner.temp }}/interoperability/topology.json + if-no-files-found: error + retention-days: 7 + - name: Admit isolated cohorts + run: python3 -B -S .github/scripts/interop-control.py admit --root "$RUNNER_TEMP/interoperability" + - name: Upload admission + uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2 + with: + name: interoperability-${{ github.run_attempt }}-admission + path: ${{ runner.temp }}/interoperability/admission.json + if-no-files-found: error + retention-days: 7 + - name: Collect isolation results + run: python3 -B -S .github/scripts/interop-control.py verify --root "$RUNNER_TEMP/interoperability" + - name: Revoke cohort grants + if: always() + run: python3 -B -S .github/scripts/interop-control.py revoke --root "$RUNNER_TEMP/interoperability" + - name: Upload revocation + uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2 + if: always() + with: + name: interoperability-${{ github.run_attempt }}-revocation + path: ${{ runner.temp }}/interoperability/revocation.json + if-no-files-found: error + retention-days: 7 + - name: Verify revocation and remove devices + if: always() + run: python3 -B -S .github/scripts/interop-control.py finish --root "$RUNNER_TEMP/interoperability" + - name: Upload isolation proof + uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2 + if: always() + with: + name: interoperability-${{ github.run_attempt }}-result + path: ${{ runner.temp }}/interoperability/result.json + if-no-files-found: error + retention-days: 7 + + result: + name: Result + if: always() + needs: [checks, metadata, workers, controller] + runs-on: ubuntu-24.04 + timeout-minutes: 5 + permissions: + contents: read + actions: read + statuses: write + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + ref: ${{ github.workflow_sha }} + persist-credentials: false + - name: Publish result for candidate revision + env: + GH_TOKEN: ${{ github.token }} + INTEROPERABILITY_JOBS: ${{ toJSON(needs) }} + run: python3 -B -S .github/scripts/interop-control.py status --root "$RUNNER_TEMP/interoperability"