tegwick
eba6e1fdbb
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a1346b-72a4-7d40-b8bd-8b8ca346cfa2
221 lines
13 KiB
Python
221 lines
13 KiB
Python
import base64
|
|
import copy
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import jsonpatch
|
|
|
|
ROOT = Path(__file__).parent
|
|
def load(name, filename):
|
|
spec = importlib.util.spec_from_file_location(name, ROOT / filename)
|
|
module = importlib.util.module_from_spec(spec); spec.loader.exec_module(module)
|
|
return module
|
|
a = load('activation', 'playwright-activation.py')
|
|
f = load('registration_fixture', 'test_playwright_registration.py')
|
|
REVISION = '5dd74dfbdf3fd38adb1bfc1bdc1326181435f0e9'
|
|
IDENTITIES = [f.IDENTITY, dict(f.IDENTITY, agent_id='second', service_subject='service:second')]
|
|
|
|
|
|
def fixtures():
|
|
secret, raw = f.fixture()
|
|
secret.update(kind='Secret', apiVersion='v1')
|
|
secret['metadata'].update(name='keycape-config', namespace='sso', uid='00000000-0000-4000-8000-000000000001')
|
|
deployment = {'apiVersion': 'apps/v1', 'kind': 'Deployment', 'metadata': {
|
|
'name': 'keycape', 'namespace': 'sso', 'uid': '00000000-0000-4000-8000-000000000002',
|
|
'resourceVersion': '12', 'generation': 5}, 'spec': {'replicas': 1, 'strategy': {'type': 'Recreate'},
|
|
'template': {'metadata': {'labels': {'app': 'keycape'}}, 'spec': {
|
|
'containers': [{'name': 'keycape', 'image': 'forgejo.coulomb.social/coulomb/key-cape@sha256:' + 'a' * 64,
|
|
'env': [{'name': 'KEYCAPE_CONFIG', 'value': '/etc/keycape/config.yaml'}],
|
|
'volumeMounts': [{'name': 'registry', 'mountPath': '/etc/keycape', 'readOnly': True}]}],
|
|
'volumes': [{'name': 'registry', 'secret': {'secretName': 'keycape-config'}}]}}}}
|
|
controller(deployment)
|
|
return secret, deployment
|
|
|
|
|
|
def controller(deployment):
|
|
# Simulates observed controller completion, separate from patch admission.
|
|
replicas = deployment['spec']['replicas']
|
|
deployment['status'] = {'observedGeneration': deployment['metadata']['generation'],
|
|
**{k: replicas for k in ['replicas', 'readyReplicas', 'availableReplicas', 'updatedReplicas']}}
|
|
|
|
|
|
class World:
|
|
def __init__(self, secret, deployment):
|
|
self.secret, self.deployment = copy.deepcopy(secret), copy.deepcopy(deployment)
|
|
self.pods = {'kind': 'PodList', 'metadata': {}, 'items': []}
|
|
self.history = []
|
|
|
|
def step(self, proposal, phase, observe=True):
|
|
operation = a.propose_step(proposal, a.digest(proposal), phase, self.secret, self.deployment, self.pods)
|
|
if operation is None: return None
|
|
attr = 'secret' if operation['kind'] == 'Secret' else 'deployment'
|
|
# Independent RFC 6902 implementation verifies actual generated CAS tests.
|
|
updated = jsonpatch.apply_patch(getattr(self, attr), operation['patch'], in_place=False)
|
|
updated['metadata']['resourceVersion'] = str(int(updated['metadata']['resourceVersion']) + 1)
|
|
if attr == 'deployment': updated['metadata']['generation'] += 1
|
|
self.history.append(phase)
|
|
setattr(self, attr, updated)
|
|
receipt = a.verify_response(operation, updated)
|
|
if attr == 'deployment' and observe: controller(updated)
|
|
return receipt
|
|
|
|
|
|
class ActivationTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.secret, self.deployment = fixtures()
|
|
self.proposal = a.build(self.secret, self.deployment, IDENTITIES, REVISION)
|
|
|
|
def test_exact_sequence_keeps_original_source_and_returns_safe_receipts(self):
|
|
world = World(self.secret, self.deployment)
|
|
original = copy.deepcopy((self.secret, self.deployment))
|
|
for phase in ['stop', 'registry', 'bindings', 'resume']:
|
|
receipt = world.step(self.proposal, phase)
|
|
self.assertTrue(receipt['readback_matches'])
|
|
self.assertNotIn('synthetic-private-marker', json.dumps(receipt))
|
|
self.assertEqual(world.deployment['spec']['replicas'], 1)
|
|
self.assertEqual(world.secret['data'], self.proposal['after_data'])
|
|
self.assertEqual((self.secret, self.deployment), original)
|
|
self.assertNotIn('synthetic-private-marker', json.dumps(a.summary(self.proposal)))
|
|
|
|
def test_rollback_preserves_preexisting_public_and_agent_registrations(self):
|
|
encoded, _ = a.registration.prepare_registry(self.secret, [IDENTITIES[0]])
|
|
self.secret['data']['config.yaml'] = encoded
|
|
binding = a.registration.registry.env_binding(a.registration.agent_client(IDENTITIES[0]))
|
|
a.container(self.deployment['spec'])['env'].append(binding)
|
|
proposal = a.build(self.secret, self.deployment, IDENTITIES, REVISION)
|
|
self.assertEqual(proposal['created_client_ids'], ['playwright-agent-second'])
|
|
world = World(self.secret, self.deployment)
|
|
for phase in ['stop', 'registry', 'bindings', 'resume', 'rollback-stop',
|
|
'rollback-registry', 'rollback-bindings', 'rollback-resume']:
|
|
world.step(proposal, phase)
|
|
self.assertEqual(world.secret['data'], self.secret['data'])
|
|
self.assertEqual(world.deployment['spec'], self.deployment['spec'])
|
|
|
|
def test_each_partial_state_recovers_without_removing_original_entries(self):
|
|
phases = ['stop', 'registry', 'bindings', 'resume']
|
|
for count in range(5):
|
|
world = World(self.secret, self.deployment)
|
|
for phase in phases[:count]: world.step(self.proposal, phase)
|
|
for phase in ['rollback-stop', 'rollback-registry', 'rollback-bindings', 'rollback-resume']:
|
|
world.step(self.proposal, phase)
|
|
self.assertEqual(world.secret['data'], self.secret['data'])
|
|
self.assertEqual(world.deployment['spec'], self.deployment['spec'])
|
|
|
|
def test_identical_complete_registration_is_noop_without_outage(self):
|
|
world = World(self.secret, self.deployment)
|
|
for phase in ['stop', 'registry', 'bindings', 'resume']: world.step(self.proposal, phase)
|
|
proposal = a.build(world.secret, world.deployment, IDENTITIES, REVISION)
|
|
self.assertFalse(a.summary(proposal)['change_needed'])
|
|
before = len(world.history)
|
|
for phase in ['stop', 'registry', 'bindings', 'resume']: self.assertIsNone(world.step(proposal, phase))
|
|
self.assertEqual(len(world.history), before)
|
|
|
|
def test_rollback_before_any_execution_never_stops_the_issuer(self):
|
|
world = World(self.secret, self.deployment)
|
|
for phase in ['rollback-stop', 'rollback-registry', 'rollback-bindings', 'rollback-resume']:
|
|
self.assertIsNone(world.step(self.proposal, phase))
|
|
self.assertEqual(world.history, [])
|
|
|
|
def test_reordered_phases_and_unobserved_controller_are_refused(self):
|
|
world = World(self.secret, self.deployment)
|
|
for phase in ['registry', 'bindings', 'resume']:
|
|
with self.subTest(phase=phase), self.assertRaises(a.Refused): world.step(self.proposal, phase)
|
|
world.step(self.proposal, 'stop', observe=False)
|
|
with self.assertRaises(a.Refused): world.step(self.proposal, 'registry')
|
|
controller(world.deployment)
|
|
with self.assertRaises(a.Refused): world.step(self.proposal, 'resume')
|
|
|
|
def test_terminating_direct_projected_and_env_consumers_prevent_registry_write(self):
|
|
specs = [{'volumes': [{'secret': {'secretName': 'keycape-config'}}]},
|
|
{'volumes': [{'projected': {'sources': [{'secret': {'name': 'keycape-config'}}]}}]},
|
|
{'containers': [{'envFrom': [{'secretRef': {'name': 'keycape-config'}}]}]},
|
|
{'initContainers': [{'env': [{'valueFrom': {'secretKeyRef': {'name': 'keycape-config'}}}]}]}]
|
|
for spec in specs:
|
|
world = World(self.secret, self.deployment); world.step(self.proposal, 'stop')
|
|
world.pods['items'] = [{'metadata': {'namespace': 'sso', 'deletionTimestamp': 'synthetic'}, 'spec': spec}]
|
|
with self.subTest(spec=spec), self.assertRaises(a.Refused): world.step(self.proposal, 'registry')
|
|
|
|
def test_partial_pod_inventory_refused(self):
|
|
world = World(self.secret, self.deployment); world.step(self.proposal, 'stop')
|
|
world.pods['metadata']['continue'] = 'another-page'
|
|
with self.assertRaises(a.Refused): world.step(self.proposal, 'registry')
|
|
|
|
def test_original_revision_and_proposal_digest_are_bound(self):
|
|
for attr in ['secret', 'deployment']:
|
|
world = World(self.secret, self.deployment)
|
|
getattr(world, attr)['metadata']['resourceVersion'] = '999'
|
|
with self.subTest(attr=attr), self.assertRaises(a.Refused): world.step(self.proposal, 'stop')
|
|
changed = copy.deepcopy(self.proposal); changed['after_data']['other'] = 'changed'
|
|
with self.assertRaises(a.Refused):
|
|
a.propose_step(changed, a.digest(self.proposal), 'stop', self.secret, self.deployment, {'kind': 'PodList', 'items': []})
|
|
|
|
def test_independent_jsonpatch_rejects_racing_version_or_spec(self):
|
|
operation = a.propose_step(self.proposal, a.digest(self.proposal), 'stop',
|
|
self.secret, self.deployment, {'kind': 'PodList', 'items': []})
|
|
for field in ['version', 'spec']:
|
|
raced = copy.deepcopy(self.deployment)
|
|
if field == 'version': raced['metadata']['resourceVersion'] = '999'
|
|
else: raced['spec']['replicas'] = 2
|
|
with self.subTest(field=field), self.assertRaises(jsonpatch.JsonPatchTestFailed):
|
|
jsonpatch.apply_patch(raced, operation['patch'])
|
|
|
|
def test_private_registry_patch_checks_complete_data_and_uid(self):
|
|
world = World(self.secret, self.deployment); world.step(self.proposal, 'stop')
|
|
operation = a.propose_step(self.proposal, a.digest(self.proposal), 'registry', world.secret, world.deployment, world.pods)
|
|
for field in ['data', 'uid']:
|
|
raced = copy.deepcopy(world.secret)
|
|
if field == 'data': raced['data']['other'] = 'racing-write'
|
|
else: raced['metadata']['uid'] = 'recreated'
|
|
with self.subTest(field=field), self.assertRaises(jsonpatch.JsonPatchTestFailed):
|
|
jsonpatch.apply_patch(raced, operation['patch'])
|
|
|
|
def test_unrelated_registry_or_deployment_drift_refuses_rollback(self):
|
|
for field in ['data', 'spec']:
|
|
world = World(self.secret, self.deployment)
|
|
for phase in ['stop', 'registry', 'bindings']: world.step(self.proposal, phase)
|
|
if field == 'data': world.secret['data']['unrelated-new-field'] = 'keep'
|
|
else: world.deployment['spec']['template']['metadata']['labels']['new'] = 'keep'
|
|
with self.subTest(field=field), self.assertRaises(a.Refused): world.step(self.proposal, 'rollback-registry')
|
|
|
|
def test_readback_mismatch_refused_without_raw_output(self):
|
|
operation = a.propose_step(self.proposal, a.digest(self.proposal), 'stop', self.secret, self.deployment, {})
|
|
with self.assertRaisesRegex(a.Refused, '^mutation_readback_mismatch$'):
|
|
a.verify_response(operation, self.deployment)
|
|
|
|
def test_unsupported_deployment_and_binding_drift_refused(self):
|
|
changes = [lambda d: d['spec'].update(replicas=2),
|
|
lambda d: d['spec']['strategy'].update(type='RollingUpdate'),
|
|
lambda d: a.container(d['spec']).update(image='unreviewed:latest'),
|
|
lambda d: a.container(d['spec']).update(envFrom=[{'secretRef': {'name': 'other'}}]),
|
|
lambda d: a.container(d['spec'])['env'].append({'name': 'KEYCAPE_CONFIG', 'value': 'other'}),
|
|
lambda d: a.container(d['spec'])['env'].append({'name': 'KEYCAPE_PLAYWRIGHT_AGENT_TEST_AGENT_CLIENT_SECRET', 'value': 'synthetic-private-marker'})]
|
|
for change in changes:
|
|
deployment = copy.deepcopy(self.deployment); change(deployment)
|
|
with self.subTest(change=change), self.assertRaises(a.Refused):
|
|
a.build(self.secret, deployment, IDENTITIES, REVISION)
|
|
|
|
def test_native_verifier_is_pinned_contained_and_ignores_ambient_environment(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
binary = Path(folder) / 'verifier'; binary.write_bytes(b'synthetic-binary')
|
|
sha = hashlib.sha256(binary.read_bytes()).hexdigest()
|
|
with patch.object(a.subprocess, 'run') as run:
|
|
run.return_value = subprocess.CompletedProcess([], 0, b'{"valid": true, "errors": []}', b'')
|
|
self.assertTrue(a.check_registry(self.proposal, binary, sha)['proposed_registry_valid'])
|
|
self.assertEqual(run.call_count, 2)
|
|
self.assertEqual(run.call_args.kwargs['env'], {'PATH': '/usr/bin:/bin'})
|
|
self.assertIn('synthetic-private-marker', run.call_args.kwargs['input'].decode())
|
|
self.assertNotIn('synthetic-private-marker', str(run.call_args.args))
|
|
run.return_value = subprocess.CompletedProcess([], 1, b'{"valid": false, "errors": ["synthetic-private-marker"]}', b'synthetic-private-marker')
|
|
with self.assertRaisesRegex(a.Refused, '^native_registry_check_failed$'):
|
|
a.check_registry(self.proposal, binary, sha)
|
|
with self.assertRaisesRegex(a.Refused, '^native_verifier_digest_mismatch$'):
|
|
a.check_registry(self.proposal, binary, 'wrong')
|
|
|
|
|
|
if __name__ == '__main__': unittest.main()
|