#!/usr/bin/env python3 """Read-only Argo adoption proof. Capture/discard raw subprocess output on errors. Use --live on the workstation; SSH collects only allowlisted metadata on the verified estate. The explicit contract binds source, revision and resource set. A receipt is proof of observation, never authority for a subsequent mutation. """ from __future__ import annotations import argparse import base64 from datetime import datetime, timezone import hashlib import json from pathlib import Path import re import subprocess import tempfile PIN = 'd2dbc19c254247652c49fda8721c80d53bca206a' CONTRACT = { 'estate_ip': '92.205.62.239', 'application': 'openbao-secretstore', 'project': 'railiance-platform-addons', 'destination': {'server': 'https://kubernetes.default.svc', 'namespace': 'external-secrets'}, 'repo_url': 'https://forgejo.coulomb.social/coulomb/railiance-platform.git', 'source_path': 'argocd/platform-addons/openbao-secretstore', 'revision': PIN, 'resources': [{'group': 'external-secrets.io', 'kind': 'ClusterSecretStore', 'namespace': '', 'name': 'openbao'}], 'ready_checks': [{'kind': 'ClusterSecretStore', 'namespace': '', 'name': 'openbao', 'tracking_id': 'openbao-secretstore:external-secrets.io/ClusterSecretStore:external-secrets/openbao'}, {'kind': 'ExternalSecret', 'namespace': 'issue-core', 'name': 'issue-core-runtime'}], 'tracking_method': 'annotation', 'controller_version': 'v3.5.3', 'minimum_proving_seconds': 86400, 'maximum_observation_age_seconds': 300, } def _kube_route(): """Local-only import; this module's source may also run remotely without it.""" import sys here = str(Path(__file__).resolve().parent) if here not in sys.path: sys.path.insert(0, here) import kube_route return kube_route def checked_run(command, *, input=None, timeout=30): try: result = subprocess.run(command, input=input, capture_output=True, timeout=timeout) if result.returncode: raise RuntimeError('operation unverified; protected output suppressed') return result.stdout except (OSError, subprocess.SubprocessError): raise RuntimeError('operation unverified; protected output suppressed') from None def identity(resource): return tuple(resource.get(k) or '' for k in ('group', 'kind', 'namespace', 'name')) def timestamp(value): if not isinstance(value, str): raise ValueError('invalid proof timestamp') stamp = datetime.fromisoformat(value.replace('Z', '+00:00')) if stamp.tzinfo is None: raise ValueError('proof timestamp requires timezone') return stamp def validate_contract(contract): if (not re.fullmatch(r'[a-f0-9]{40}', contract['revision']) or not re.fullmatch(r'[a-z0-9][a-z0-9.-]*', contract['application']) or not re.fullmatch(r'[a-z0-9/-]+', contract['source_path']) or '..' in contract['source_path'] or contract['source_path'].startswith('/')): raise ValueError('invalid adoption contract') for check in contract['ready_checks']: if check['kind'] not in ('ClusterSecretStore', 'ExternalSecret'): raise ValueError('secret objects are excluded from the metadata contract') for key in ('namespace', 'name'): if not re.fullmatch(r'[a-z0-9.-]*', check[key]) or (key == 'name' and not check[key]): raise ValueError('invalid metadata identity') if (not contract['resources'] or len({identity(r) for r in contract['resources']}) != len(contract['resources']) or contract['minimum_proving_seconds'] < 86400 or not 0 < contract['maximum_observation_age_seconds'] <= 300): raise ValueError('invalid proof bounds') def tracking_settings(contract): cm = json.loads(checked_run(['kubectl', '-n', 'argocd', 'get', 'configmap', 'argocd-cm', '-o', 'json'])) controller = json.loads(checked_run(['kubectl', '-n', 'argocd', 'get', 'statefulset', 'argocd-application-controller', '-o', 'json'])) containers = controller['spec']['template']['spec']['containers'] images = [c['image'] for c in containers if c['name'] == 'argocd-application-controller'] if len(images) != 1 or not images[0].endswith(':' + contract['controller_version']): raise ValueError('expected Argo controller version unverified') data = cm.get('data', {}) # Argo 3.0 changed the default from label to annotation. Bind that default to # an observed pinned 3.x controller; never infer it for an unknown runtime. method = data.get('application.resourceTrackingMethod', 'annotation') label_key = data.get('application.instanceLabelKey', 'app.kubernetes.io/instance') if method != contract['tracking_method'] or data.get('installationID'): raise ValueError('tracking semantics differ from this adoption contract') return {'method': method, 'instance_label_key': label_key, 'controller_version': contract['controller_version'], 'installation_id_absent': True} def argo_desired(source, contract, settings): """Inject only Argo-owned tracking metadata, never copy live fields into source.""" import copy desired = copy.deepcopy(source) if (desired.get('apiVersion') != 'external-secrets.io/v1beta1' or desired.get('kind') != 'ClusterSecretStore' or desired.get('metadata', {}).get('name') != 'openbao' or desired['metadata'].get('namespace')): raise ValueError('pinned source identity differs from admitted store') if (settings['method'] != contract['tracking_method'] or settings['controller_version'] != contract['controller_version'] or settings['installation_id_absent'] is not True): raise ValueError('unsupported Argo render semantics') tracking = f"{contract['application']}:external-secrets.io/ClusterSecretStore:{contract['destination']['namespace']}/openbao" annotations = desired['metadata'].setdefault('annotations', {}) if 'argocd.argoproj.io/tracking-id' in annotations and annotations['argocd.argoproj.io/tracking-id'] != tracking: raise ValueError('source tracking annotation conflicts with Argo identity') annotations['argocd.argoproj.io/tracking-id'] = tracking return desired def changed_paths(left, right, path=''): """JSON pointers only: never return old/new values or annotation contents.""" if type(left) is not type(right): return [path] if isinstance(left, dict): result = [] for key in sorted(set(left) | set(right)): child = path + '/' + key.replace('~', '~0').replace('/', '~1') result.extend([child] if key not in left or key not in right else changed_paths(left[key], right[key], child)) return result return [] if left == right else [path] def stable_object(value): import copy value = copy.deepcopy(value) for field in ('creationTimestamp', 'managedFields', 'resourceVersion', 'uid', 'generation'): value['metadata'].pop(field, None) value.pop('status', None) return value def remote_diff(rendered, contract): def get(args, input=None): return json.loads(checked_run(['kubectl', *args], input=input)) nodes = get(['get', 'nodes', '-o', 'json']) if contract['estate_ip'] not in [a['address'] for n in nodes['items'] for a in n['status']['addresses']]: raise ValueError('estate identity unverified') settings = tracking_settings(contract) source = get(['create', '--dry-run=client', '--validate=false', '-f', '-', '-o', 'json'], rendered) normalized = json.dumps(argo_desired(source, contract, settings), sort_keys=True).encode() live = get(['get', '--raw', '/apis/external-secrets.io/v1beta1/clustersecretstores/openbao']) if live.get('apiVersion') != source['apiVersion']: raise ValueError('same source API-version read unverified') def diff(payload, server): command = ['kubectl', 'diff', *(['--server-side'] if server else []), '-f', '-'] result = subprocess.run(command, input=payload, capture_output=True, timeout=30) if result.returncode not in (0, 1): raise ValueError('native diff failed; output suppressed') return result.returncode raw_client = get(['apply', '--dry-run=server', '-f', '-', '-o', 'json'], rendered) raw_server = get(['apply', '--server-side', '--dry-run=server', '-f', '-', '-o', 'json'], rendered) normalized_client = get(['apply', '--dry-run=server', '-f', '-', '-o', 'json'], normalized) normalized_server = get(['apply', '--server-side', '--dry-run=server', '-f', '-', '-o', 'json'], normalized) return {'revision': contract['revision'], 'path': contract['source_path'], 'render_sha256': hashlib.sha256(rendered).hexdigest(), 'render_resource_count': 1, 'normalized_render_sha256': hashlib.sha256(normalized).hexdigest(), 'normalization': 'argocd-v3-annotation-tracking-destination-namespace', 'tracking_settings': settings, 'same_api_version': True, 'raw_server_exit_code': diff(rendered, True), 'raw_client_exit_code': diff(rendered, False), 'raw_client_changed_paths': changed_paths(stable_object(live), stable_object(raw_client)), 'raw_server_changed_paths': changed_paths(stable_object(live), stable_object(raw_server)), 'exit_code': diff(normalized, True), 'client_side_exit_code': diff(normalized, False), 'normalized_client_changed_paths': changed_paths(stable_object(live), stable_object(normalized_client)), 'normalized_server_changed_paths': changed_paths(stable_object(live), stable_object(normalized_server)), 'output_suppressed': True} def collect(contract): validate_contract(contract) def get(args): return json.loads(checked_run(['kubectl', *args, '-o', 'json'])) nodes = get(['get', 'nodes']) addresses = [a.get('address') for n in nodes['items'] for a in n.get('status', {}).get('addresses', [])] if contract['estate_ip'] not in addresses: raise ValueError('estate identity unverified') obj = get(['-n', 'argocd', 'get', 'application', contract['application']]) spec, status = obj['spec'], obj.get('status', {}) op = status.get('operationState', {}) sync_operation = op.get('operation', {}).get('sync', {}) app = {'name': obj['metadata']['name'], 'uid': obj['metadata']['uid'], 'project': spec.get('project'), 'destination': spec.get('destination'), 'source': {k: spec.get('source', {}).get(k) for k in ('repoURL', 'path', 'targetRevision')}, 'automated_present': 'automated' in spec.get('syncPolicy', {}), 'sync': status.get('sync', {}).get('status'), 'health': status.get('health', {}).get('status'), 'revision': status.get('sync', {}).get('revision'), 'pending_operation': bool(obj.get('operation')), 'condition_types': [c.get('type') for c in status.get('conditions', [])], 'operation': {k: op.get(k) for k in ('phase', 'startedAt', 'finishedAt')}, 'sync_prune': sync_operation.get('prune', False), 'sync_strategy_apply': 'apply' in sync_operation.get('syncStrategy', {}), 'operation_revision': sync_operation.get('revision'), 'resources': [{k: r.get(k) for k in ('group', 'kind', 'namespace', 'name', 'status')} for r in status.get('resources', [])]} checks = [] for check in contract['ready_checks']: args = ['-n', check['namespace']] if check['namespace'] else [] obj = get([*args, 'get', check['kind'], check['name']]) ready = [c for c in obj.get('status', {}).get('conditions', []) if c.get('type') == 'Ready'] checks.append({'kind': check['kind'], 'namespace': check['namespace'], 'name': check['name'], 'uid': obj['metadata']['uid'], 'ready': [{k: c.get(k) for k in ('status', 'lastTransitionTime')} for c in ready], 'tracking_matches': obj['metadata'].get('annotations', {}).get('argocd.argoproj.io/tracking-id') == check['tracking_id'] if 'tracking_id' in check else None}) return {'tracking_settings': tracking_settings(contract), 'captured_at': datetime.now(timezone.utc).isoformat(), 'estate_ip': contract['estate_ip'], 'application': app, 'ready_checks': checks} def verify(observation, contract, *, now=None): validate_contract(contract) now = now or datetime.now(timezone.utc) age = (now - timestamp(observation['captured_at'])).total_seconds() if not 0 <= age <= contract['maximum_observation_age_seconds']: raise ValueError('observation stale or from future') if observation['estate_ip'] != contract['estate_ip']: raise ValueError('estate identity mismatch') settings = observation.get('tracking_settings', {}) if (settings.get('method') != contract['tracking_method'] or settings.get('controller_version') != contract['controller_version'] or settings.get('installation_id_absent') is not True): raise ValueError('observed Argo tracking semantics unverified') app = observation['application'] if (app['name'] != contract['application'] or not app['uid'] or app['project'] != contract['project'] or app['source'] != {'repoURL': contract['repo_url'], 'path': contract['source_path'], 'targetRevision': contract['revision']} or app['revision'] != contract['revision'] or app.get('destination') != contract['destination']): raise ValueError('application source or identity mismatch') if app['automated_present'] is not False or app['pending_operation'] is not False: raise ValueError('manual idle policy unverified') if app['sync'] != 'Synced' or app['health'] != 'Healthy': raise ValueError('application sync or health unverified') if any(t != 'OrphanedResourceWarning' for t in app['condition_types']): raise ValueError('application condition prevents proof') op = app['operation'] if (op['phase'] != 'Succeeded' or app['sync_prune'] is not False or app['sync_strategy_apply'] is not True or app['operation_revision'] != contract['revision']): raise ValueError('successful pinned manual apply unverified') start, finish = timestamp(op['startedAt']), timestamp(op['finishedAt']) proof_age = (timestamp(observation['captured_at']) - finish).total_seconds() if finish < start or proof_age < contract['minimum_proving_seconds']: raise ValueError('completed adoption proving period unverified') resources = app['resources'] expected = {identity(r) for r in contract['resources']} if (len(resources) != len(expected) or {identity(r) for r in resources} != expected or any(r['status'] != 'Synced' for r in resources)): raise ValueError('exact tracked resource set unverified') checks = observation['ready_checks'] expected_checks = {(c['kind'], c['namespace'], c['name']): c for c in contract['ready_checks']} if len(checks) != len(expected_checks) or {(c['kind'], c['namespace'], c['name']) for c in checks} != set(expected_checks): raise ValueError('readiness checks incomplete') for check in checks: key = (check['kind'], check['namespace'], check['name']) ready = check['ready'] if not check['uid'] or len(ready) != 1 or ready[0]['status'] != 'True': raise ValueError('resource readiness unverified') if timestamp(ready[0]['lastTransitionTime']) > start: raise ValueError('resource readiness changed during adoption proving period') if 'tracking_id' in expected_checks[key] and check['tracking_matches'] is not True: raise ValueError('resource tracking identity unverified') diff = observation.get('pinned_render_diff', {}) allowed_raw_paths = { '/metadata/annotations/argocd.argoproj.io~1tracking-id', '/metadata/annotations/kubectl.kubernetes.io~1last-applied-configuration', } for mode in ('client', 'server'): paths = diff.get(f'raw_{mode}_changed_paths') code = diff.get(f'raw_{mode}_exit_code') if (not isinstance(paths, list) or any(not isinstance(path, str) for path in paths) or not set(paths) <= allowed_raw_paths or type(code) is not int or code not in (0, 1) or bool(paths) != bool(code)): raise ValueError('raw source drift contains unknown or substantive fields') if (diff.get('revision') != contract['revision'] or diff.get('path') != contract['source_path'] or type(diff.get('exit_code')) is not int or diff['exit_code'] != 0 or type(diff.get('client_side_exit_code')) is not int or diff['client_side_exit_code'] != 0 or diff.get('normalization') != 'argocd-v3-annotation-tracking-destination-namespace' or diff.get('tracking_settings') != settings or diff.get('same_api_version') is not True or diff.get('normalized_client_changed_paths') != [] or diff.get('normalized_server_changed_paths') != [] or not re.fullmatch(r'[a-f0-9]{64}', diff.get('normalized_render_sha256', '')) or diff.get('render_resource_count') != len(expected) or not re.fullmatch(r'[a-f0-9]{64}', diff.get('render_sha256', ''))): raise ValueError('exact pinned render zero-diff unverified') return {'schema_version': 'argocd-adoption-proof/v1', 'status': 'verified', 'application': contract['application'], 'revision': contract['revision'], 'captured_at': observation['captured_at'], 'verified_at': now.isoformat(), 'proving_seconds': proof_age, 'resource_count': len(expected), 'ready_count': len(checks), 'manual_apply_prune_off': True, 'warnings': sorted(set(app['condition_types'])), 'native_mutation': False, 'verifier_sha256': hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), 'contract_sha256': hashlib.sha256(json.dumps(contract,sort_keys=True).encode()).hexdigest(), 'observation_sha256': hashlib.sha256(json.dumps(observation,sort_keys=True).encode()).hexdigest()} def render_diff(source_root, contract): """Render a pinned tracked source into a disposable local directory; no checkout.""" with tempfile.TemporaryDirectory(prefix='adoption-pin-') as temporary: archive = checked_run(['git', '-C', str(source_root), 'archive', contract['revision'], contract['source_path']]) checked_run(['tar', '-x', '-C', temporary], input=archive) rendered = checked_run(['kubectl', 'kustomize', str(Path(temporary) / contract['source_path'])]) # Fixed admission for this phase: exactly one public ClusterSecretStore, # never forward a Secret or extra document to a diff operation. if re.findall(rb'^kind: (.+)$', rendered, re.M) != [b'ClusterSecretStore']: raise ValueError('pinned render resource kind unverified') encoded = base64.b64encode(rendered).decode() return json.loads(checked_run(_kube_route().ssh_argv(['python3', '-', '--remote-diff', '--render-base64', encoded]), input=Path(__file__).read_bytes(), timeout=150)) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--live', action='store_true') parser.add_argument('--remote-diff', action='store_true', help=argparse.SUPPRESS) parser.add_argument('--render-base64', help=argparse.SUPPRESS) parser.add_argument('--remote-collect', action='store_true', help=argparse.SUPPRESS) parser.add_argument('--observation', type=Path) parser.add_argument('--source-root', type=Path) parser.add_argument('--receipt', type=Path) args = parser.parse_args() try: if args.remote_diff: print(json.dumps(remote_diff(base64.b64decode(args.render_base64, validate=True), CONTRACT), sort_keys=True)) return 0 if args.remote_collect: print(json.dumps(collect(CONTRACT), sort_keys=True)) return 0 if args.live: if not args.source_root: raise ValueError('source root required for pinned rendering') raw = checked_run(_kube_route().ssh_argv(['python3', '-', '--remote-collect']), input=Path(__file__).read_bytes(), timeout=120) observation = json.loads(raw) observation['pinned_render_diff'] = render_diff(args.source_root, CONTRACT) if args.observation: args.observation.write_text(json.dumps(observation,indent=2)+'\n') elif args.observation: observation = json.loads(args.observation.read_text()) else: raise ValueError('live read or observation required') receipt = verify(observation, CONTRACT) except Exception: receipt = {'schema_version':'argocd-adoption-proof/v1', 'status':'unverified', 'error':'adoption proof incomplete; raw output suppressed', 'native_mutation':False} if args.receipt: args.receipt.write_text(json.dumps(receipt,indent=2)+'\n') print(json.dumps(receipt,sort_keys=True)) return 0 if receipt.get('status') == 'verified' else 1 if __name__ == '__main__': raise SystemExit(main())