1dc1d2f3a1
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade
127 lines
8.2 KiB
Python
127 lines
8.2 KiB
Python
import base64
|
|
from datetime import datetime, timezone, timedelta
|
|
import importlib.util
|
|
from pathlib import Path
|
|
import unittest
|
|
from unittest import mock
|
|
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
|
|
from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
|
|
|
|
SPEC=importlib.util.spec_from_file_location('readiness',Path(__file__).parents[1]/'scripts/activity_promotion_readiness.py')
|
|
p=importlib.util.module_from_spec(SPEC);SPEC.loader.exec_module(p)
|
|
NOW=datetime(2026,10,11,20,tzinfo=timezone.utc)
|
|
|
|
|
|
def observation():
|
|
return {'captured_at':NOW.isoformat(),'estate_ip':'92.205.62.239','cluster_uid':'cluster-uid','application_uid':'app-uid',
|
|
'source':{'repoURL':'https://forgejo.coulomb.social/coulomb/activity-core.git','path':'k8s/gitops','targetRevision':p.PIN},
|
|
'project':'activity-core','revision':p.PIN,'destination':{'server':'https://kubernetes.default.svc','namespace':'activity-core'},
|
|
'automated_present':False,'pending_operation':False,
|
|
'sync_options':['CreateNamespace=false','ApplyOutOfSyncOnly=true','PruneLast=true','FailOnSharedResource=true'],
|
|
'sync':'Synced','health':'Healthy','condition_types':[],'project_scope_matches':True,
|
|
'operation':{'phase':'Succeeded','finishedAt':'2026-10-04T14:13:25Z'},
|
|
'resources':[dict(zip(('group','kind','namespace','name'),r),status='Synced') for r in sorted(p.RESOURCES)],
|
|
'deployments':[{'name':name,'generation_observed':True,'replicas':1,'ready':1,'updated':1,'available':1}
|
|
for name in ('actcore-api','actcore-worker','actcore-event-router')]}
|
|
|
|
|
|
def payload():
|
|
return {'schema_version':'activity-core-release-acceptance/v1','owner_task':p.OWNER_TASK,'issuer':'owner-acceptance-principal',
|
|
'authority':p.AUTHORITY,'application':'activity-core','cluster_uid':'cluster-uid','adoption_revision':p.PIN,
|
|
'targets':dict(p.TARGETS),'release_principal':'activity-core-release-broker','broker_revision':'a'*40,
|
|
'checks':{c:True for c in p.CHECKS},'issued_at':NOW.isoformat(),'expires_at':(NOW+timedelta(minutes=4)).isoformat(),
|
|
'continuous_health':{'revision':p.PIN,'continuous':True,'max_gap_seconds':90,'observed_at':NOW.isoformat(),
|
|
'healthy_since':(NOW-timedelta(days=1)).isoformat()}}
|
|
|
|
|
|
class GateTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.private=Ed25519PrivateKey.generate()
|
|
self.public=self.private.public_key().public_bytes(Encoding.Raw,PublicFormat.Raw)
|
|
|
|
def envelope(self,data=None):
|
|
data=data or payload()
|
|
return {'payload':data,'signature':base64.b64encode(self.private.sign(p.canonical(data))).decode()}
|
|
|
|
def evaluate(self,envelope=None,obs=None,key=None):
|
|
return p.readiness(obs or observation(),envelope=envelope or self.envelope(),trusted_key=key or self.public,
|
|
trusted_principal='owner-acceptance-principal',now=NOW)
|
|
|
|
def test_complete_signed_fixture_passes_without_activation(self):
|
|
result=self.evaluate()
|
|
self.assertEqual(result['status'],'ready')
|
|
self.assertFalse(result['native_mutation'])
|
|
self.assertFalse(result['automatic_sync_candidate'])
|
|
|
|
def test_snapshot_and_elapsed_time_cannot_substitute_for_owner_proof(self):
|
|
result=p.readiness(observation(),now=NOW)
|
|
self.assertEqual(result['status'],'waiting')
|
|
self.assertEqual(result['depends_on'],['ACTIVITY-WP-0041-T03'])
|
|
self.assertFalse(result['needs_human'])
|
|
self.assertTrue(result['adoption_snapshot_is_not_continuous_observation'])
|
|
|
|
def test_every_acceptance_check_required(self):
|
|
for check in p.CHECKS:
|
|
for missing in (False,None,'true'):
|
|
data=payload();data['checks'][check]=missing
|
|
with self.subTest(check=check,value=missing):self.assertEqual(self.evaluate(self.envelope(data))['status'],'unverified')
|
|
|
|
def test_unknown_key_forged_tampered_or_self_introduced_trust_refused(self):
|
|
envelope=self.envelope();envelope['payload']['release_principal']='attacker'
|
|
self.assertEqual(self.evaluate(envelope)['status'],'unverified')
|
|
unknown=Ed25519PrivateKey.generate().public_key().public_bytes(Encoding.Raw,PublicFormat.Raw)
|
|
self.assertEqual(self.evaluate(key=unknown)['status'],'unverified')
|
|
envelope=self.envelope();envelope['public_key']=self.public.hex()
|
|
self.assertEqual(self.evaluate(envelope)['status'],'unverified')
|
|
envelope=self.envelope();envelope['signature']='invalid-base64!'
|
|
self.assertEqual(self.evaluate(envelope)['status'],'unverified')
|
|
|
|
def test_wrong_scope_identity_cluster_commit_or_expired_owner_proof_refused(self):
|
|
for key,value in [('authority','admin'),('application','other'),('cluster_uid','other'),
|
|
('adoption_revision','b'*40),('owner_task','other'),('issuer','producer'),
|
|
('broker_revision','main'),('release_principal',''),
|
|
('issued_at',(NOW-timedelta(minutes=6)).isoformat()),
|
|
('issued_at',(NOW+timedelta(seconds=1)).isoformat()),
|
|
('expires_at',NOW.isoformat())]:
|
|
data=payload();data[key]=value
|
|
with self.subTest(key=key):self.assertEqual(self.evaluate(self.envelope(data))['status'],'unverified')
|
|
for key,value in [('prune',True),('root_sync_selects_only','*'),('child_path','other')]:
|
|
data=payload();data['targets'][key]=value
|
|
with self.subTest(target=key):self.assertEqual(self.evaluate(self.envelope(data))['status'],'unverified')
|
|
|
|
def test_continuous_sampling_requires_owner_signed_fresh_bounded_interval(self):
|
|
for key,value in [('continuous',False),('max_gap_seconds',91),('max_gap_seconds',True),
|
|
('revision','b'*40),('healthy_since',(NOW-timedelta(hours=23)).isoformat()),
|
|
('observed_at',(NOW-timedelta(seconds=91)).isoformat()),
|
|
('observed_at',(NOW+timedelta(seconds=1)).isoformat())]:
|
|
data=payload();data['continuous_health'][key]=value
|
|
with self.subTest(key=key):self.assertEqual(self.evaluate(self.envelope(data))['status'],'unverified')
|
|
|
|
def test_exact_live_adoption_scope_and_freshness_refusals(self):
|
|
for key,value in [('estate_ip','other'),('project_scope_matches',False),('project','other'),
|
|
('automated_present',True),('pending_operation',True),('revision','b'*40),
|
|
('health','Degraded'),('sync','OutOfSync'),('condition_types',['ComparisonError']),
|
|
('captured_at',(NOW-timedelta(minutes=6)).isoformat()),
|
|
('captured_at',(NOW+timedelta(seconds=1)).isoformat()),
|
|
('resources',[]),('deployments',[])]:
|
|
data=observation();data[key]=value
|
|
with self.subTest(key=key):self.assertEqual(self.evaluate(obs=data)['status'],'unverified')
|
|
data=observation();data['resources'].append(data['resources'][0])
|
|
self.assertEqual(self.evaluate(obs=data)['status'],'unverified')
|
|
for key,value in [('ready',0),('generation_observed',False),('updated',0),('replicas',0)]:
|
|
data=observation();data['deployments'][0][key]=value
|
|
with self.subTest(deployment=key):self.assertEqual(self.evaluate(obs=data)['status'],'unverified')
|
|
|
|
def test_no_signature_verification_attempt_without_trusted_operator_config(self):
|
|
self.assertEqual(p.readiness(observation(),envelope=self.envelope(),now=NOW)['status'],'waiting')
|
|
|
|
def test_protected_subprocess_errors_and_timeouts_never_leak_canary(self):
|
|
import subprocess
|
|
from types import SimpleNamespace
|
|
canary='SECRET-CANARY-NOT-OUTPUT'
|
|
with mock.patch.object(p.subprocess,'run',return_value=SimpleNamespace(returncode=1,stdout=canary,stderr=canary)):
|
|
with self.assertRaises(ValueError) as exc:p.run(['kubectl','get','nodes'])
|
|
self.assertNotIn(canary,str(exc.exception))
|
|
with mock.patch.object(p.subprocess,'run',side_effect=subprocess.TimeoutExpired('cmd',1,output=canary,stderr=canary)):
|
|
with self.assertRaises(ValueError) as exc:p.run(['kubectl','get','nodes'])
|
|
self.assertNotIn(canary,str(exc.exception))
|