1228cd8407
KUBECTL_ROUTE (agent default, founder-sudo = ssh railiance01 sudo -n) for the bounded proof scripts; one attended script with go/no-go gates, typed ADOPT confirmation, prune/force-off SSA sync and post-checks; evaluator and tests. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 527080@bnt-lap001 Assistant-Session: 5a0e864a-67be-4605-a77a-16635f7dd292
365 lines
21 KiB
Python
365 lines
21 KiB
Python
#!/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())
|