100 lines
5.6 KiB
Python
100 lines
5.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Offline exact-source owner review verification. No cluster/credential access."""
|
|
import copy
|
|
import hashlib
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tarfile
|
|
import tempfile
|
|
import yaml
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
WAVE = HERE.parent
|
|
ACCESS = Path('/home/worsch/access-engine')
|
|
CHART = 'dc042022f7c30671cf0a7605cf43ef6d96d04242'
|
|
IMAGE = '84c78e072c8f3daf2bc3b90f8efc8fbeb16c3cf0'
|
|
def digest(data):
|
|
return hashlib.sha256(data).hexdigest()
|
|
def gitbytes(revision, path):
|
|
return subprocess.check_output(['git', '-C', str(ACCESS), 'show', f'{revision}:{path}'])
|
|
def write(name, data):
|
|
(HERE / name).write_text(json.dumps(data, indent=2) + '\n')
|
|
|
|
pins = {}
|
|
for rel in ['access-native-admission/rollout-r2/manifest.json',
|
|
'platform-signer-custody/native-adapter-r2/native-packet-p4/manifest.json',
|
|
'native-preflight-r1/live-preflight.json',
|
|
'native-preflight-r1/cluster-native-plan-review.json']:
|
|
pins[rel] = digest((WAVE / rel).read_bytes())
|
|
assert pins['access-native-admission/rollout-r2/manifest.json'] == '5be7a7f438624dcb35b7dbc96e7c9e5a8e2d5d1b1966cb4338aa5aecbe3a8b59'
|
|
assert pins['platform-signer-custody/native-adapter-r2/native-packet-p4/manifest.json'] == '9eec861838a8e9138f2c821cff3014998fbe74bb01447642bf50d44e8b3b7cb3'
|
|
rollout = WAVE / 'access-native-admission/rollout-r2'
|
|
manifest = json.loads((rollout / 'manifest.json').read_text())
|
|
for path, expected in manifest['files'].items():
|
|
assert 'sha256:' + digest((rollout / path).read_bytes()) == expected, path
|
|
evidence = json.loads((rollout / 'render-evidence.json').read_text())
|
|
for path, expected in evidence['source_files'].items():
|
|
assert 'sha256:' + digest(gitbytes(CHART, path)) == expected, path
|
|
if path.startswith('examples/'):
|
|
assert gitbytes(CHART, path) == gitbytes(IMAGE, path)
|
|
|
|
with tempfile.TemporaryDirectory(prefix='access-plan-review-') as temporary:
|
|
root = Path(temporary)
|
|
archive = subprocess.check_output(['git', '-C', str(ACCESS), 'archive', CHART])
|
|
with tarfile.open(fileobj=io.BytesIO(archive)) as tar:
|
|
tar.extractall(root, filter='data')
|
|
states = []
|
|
for phase in ['stopped', 'activation']:
|
|
command = ['helm', 'template', 'flex-auth-secrets-engine', str(root / 'charts/flex-auth'),
|
|
'--namespace', 'flex-auth', '-f', str(root / 'values/secrets-engine.yaml'),
|
|
'-f', str(rollout / f'{phase}-values.proposed.yaml')]
|
|
rendered = subprocess.check_output(command)
|
|
assert rendered == (rollout / f'{phase}-source-render.yaml').read_bytes()
|
|
states.append([x for x in yaml.safe_load_all(rendered) if x])
|
|
stopped, ready = states
|
|
normalized = copy.deepcopy(ready)
|
|
next(x for x in normalized if x['kind'] == 'Deployment')['spec']['replicas'] = 0
|
|
assert normalized == stopped
|
|
deployment = next(x for x in stopped if x['kind'] == 'Deployment')
|
|
assert deployment['spec']['strategy'] == {'type': 'Recreate'}
|
|
assert deployment['spec']['replicas'] == 0
|
|
assert not any(x['kind'] == 'Secret' for x in stopped)
|
|
pvc = next(x for x in stopped if x['kind'] == 'PersistentVolumeClaim')
|
|
assert pvc['metadata']['annotations']['helm.sh/resource-policy'] == 'keep'
|
|
assert pvc['spec']['resources']['requests']['storage'] == '1Gi'
|
|
|
|
paths = ['AGENTS.md', 'INTENT.md', 'SCOPE.md', '.forgejo/workflows/image.yaml',
|
|
'.forgejo/workflows/ci-smoke.yaml', '.github/workflows/ci.yml', 'Makefile',
|
|
'deploy/README.md', 'railiance/app.toml', 'tools/publish-runtime-images.sh',
|
|
'internal/emission/outbox.go', 'internal/emission/lock_linux.go',
|
|
'internal/emission/lock_other.go', 'internal/emission/delivery.go',
|
|
'internal/decision/engine.go', 'cmd/access-engine/main.go']
|
|
source_pins = {p: digest(gitbytes(CHART, p)) for p in paths}
|
|
source_pins['workplans/FLEX-WP-0024-decision-envelope-authenticity.md'] = digest(
|
|
(ACCESS / 'workplans/FLEX-WP-0024-decision-envelope-authenticity.md').read_bytes())
|
|
image_source = {p: digest(gitbytes(IMAGE, p)) for p in paths if p.startswith(('internal/', 'cmd/'))}
|
|
for p, sha in image_source.items():
|
|
assert source_pins[p] == sha, p
|
|
for p in ['.forgejo/workflows/image.yaml', '.forgejo/workflows/ci-smoke.yaml', '.github/workflows/ci.yml']:
|
|
data = gitbytes(CHART, p).decode()
|
|
assert 'helm upgrade' not in data and 'kubectl apply' not in data
|
|
|
|
preflight = json.loads((WAVE / 'native-preflight-r1/live-preflight.json').read_text())
|
|
observations = preflight['observations']
|
|
writer_observations = {k: observations[k] for k in ['hpa', 'jobs', 'cronjobs', 'daemonsets', 'statefulsets']}
|
|
assert all(x['ok'] and not x['rows'] for x in writer_observations.values())
|
|
write('verification.json', {
|
|
'schema': 'access-engine.native-plan-offline-verification.v1',
|
|
'frozen_input_sha256': pins, 'chart_revision': CHART, 'image_revision': IMAGE,
|
|
'reviewed_source_sha256': {p: 'sha256:' + value for p, value in source_pins.items()},
|
|
'image_runtime_source_sha256': {p: 'sha256:' + value for p, value in image_source.items()},
|
|
'frozen_manifest_file_hashes_verified': True, 'chart_source_hashes_verified': True,
|
|
'actual_exact_helm_renders_equal_frozen_bytes': True,
|
|
'stopped_ready_only_replicas_differ': True, 'runtime_sources_identical_chart_and_image': True,
|
|
'reviewed_repo_ci_has_no_helm_upgrade_or_kubectl_apply': True,
|
|
'dated_namespace_competing_controller_observations': writer_observations,
|
|
'observation_time': preflight['observed_at'], 'live_execution': False,
|
|
'no_credential_or_kubeconfig_access': True})
|
|
print('Verified frozen R2/P4 hashes, exact source files and two byte-identical Helm renders; no native access.')
|