tegwick
52dcc08f23
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade
84 lines
4.9 KiB
Python
84 lines
4.9 KiB
Python
import base64
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
|
|
from cryptography.hazmat.primitives.serialization import Encoding,PublicFormat
|
|
|
|
from activity_core.release_runtime import Admission, CLUSTER_UID, DurableAudit, verify_runtime_identity
|
|
|
|
|
|
def configuration(root):
|
|
kube=root/'kubeconfig';kube.write_text('fixture private kube configuration');kube.chmod(0o600)
|
|
git=root/'gitconfig';git.write_text('fixture private git configuration');git.chmod(0o600)
|
|
trust={}
|
|
for role in ('build','review','health','retention'):
|
|
public=Ed25519PrivateKey.generate().public_key().public_bytes(Encoding.Raw,PublicFormat.Raw)
|
|
trust[role]={'public_key':base64.b64encode(public).decode(),'principal':role,'roles':[role]}
|
|
value={'schema_version':'activity-core-release-runtime/v1','enabled':True,'principal':'activity-core-release-broker',
|
|
'cluster_uid':CLUSTER_UID,'state_directory':str(root/'state'),'kubeconfig':str(kube),'git_config':str(git),'receipt_trust':trust}
|
|
path=root/'runtime.json';path.write_text(json.dumps(value));path.chmod(0o600)
|
|
return path,value
|
|
|
|
|
|
def test_operator_private_config_drift_disable_and_custody_replacement_refused(tmp_path):
|
|
path,value=configuration(tmp_path);admission=Admission(path);admission.require()
|
|
assert admission.environment()['GIT_CONFIG_NOSYSTEM']=='1'
|
|
value['enabled']=False;path.write_text(json.dumps(value))
|
|
with pytest.raises(ValueError,match='disabled'):admission.require()
|
|
value['enabled']=True;value['principal']='borrowed-admin';path.write_text(json.dumps(value))
|
|
with pytest.raises(ValueError):admission.require()
|
|
value['principal']='activity-core-release-broker';path.write_text(json.dumps(value))
|
|
Path(value['kubeconfig']).write_text('replaced fixture custody')
|
|
with pytest.raises(ValueError,match='custody'):admission.require()
|
|
|
|
|
|
@pytest.mark.parametrize('delta',[{'enabled':False},{'principal':'operator'},{'cluster_uid':'other'},
|
|
{'schema_version':'other'},{'untrusted_transport_factory':'arbitrary-import'}])
|
|
def test_disabled_widened_configuration_cannot_construct_runtime(tmp_path,delta):
|
|
path,value=configuration(tmp_path);path.write_text(json.dumps(value|delta))
|
|
with pytest.raises(ValueError):Admission(path)
|
|
|
|
|
|
def test_public_permissions_symlink_or_incomplete_role_registry_refused(tmp_path):
|
|
path,value=configuration(tmp_path);path.chmod(0o644)
|
|
with pytest.raises(ValueError,match='private'):Admission(path)
|
|
path.chmod(0o600);alias=tmp_path/'symlink.json';alias.symlink_to(path)
|
|
with pytest.raises(ValueError,match='private'):Admission(alias)
|
|
del value['receipt_trust']['health'];path.write_text(json.dumps(value))
|
|
with pytest.raises(ValueError,match='independent'):Admission(path)
|
|
|
|
|
|
def identity_runner(*,admin=False,extra_access=False,wrong_cluster=False):
|
|
calls=[]
|
|
def run(argv,**kwargs):
|
|
calls.append(argv)
|
|
if 'namespace' in argv:return json.dumps({'metadata':{'uid':'wrong' if wrong_cluster else CLUSTER_UID}})
|
|
if 'whoami' in argv:return json.dumps({'status':{'userInfo':{'username':'root' if admin else 'system:serviceaccount:activity-core:activity-core-release-broker','groups':['system:masters'] if admin else ['system:serviceaccounts']}}})
|
|
assert '--raw=/apis/authorization.k8s.io/v1/selfsubjectaccessreviews' in argv
|
|
attrs=json.loads(kwargs['input'])['spec']['resourceAttributes']
|
|
allowed=attrs['resource']=='applications' and attrs.get('name') in {'activity-core','railiance-apps-root'}
|
|
if extra_access and attrs['resource']=='secrets':allowed=True
|
|
return json.dumps({'status':{'allowed':allowed}})
|
|
return run,calls
|
|
|
|
|
|
def test_actual_subject_scope_and_denial_checks_are_required_before_dispatch():
|
|
run,calls=identity_runner();result=verify_runtime_identity(run)
|
|
assert result=={'principal_verified':True,'allowed_checks':4,'denied_checks':7}
|
|
assert len(calls)==13
|
|
for kwargs in [{'admin':True},{'extra_access':True},{'wrong_cluster':True}]:
|
|
run,calls=identity_runner(**kwargs)
|
|
with pytest.raises(ValueError):verify_runtime_identity(run)
|
|
if kwargs.get('wrong_cluster'):assert len(calls)==1
|
|
|
|
|
|
def test_audit_sink_durably_deduplicates_and_refuses_collision_or_private_fields(tmp_path):
|
|
sink=DurableAudit(tmp_path/'audit.sqlite')
|
|
event={'event_id':'release:id:1','event_type':'release_transition','release_id':'id','application':'activity-core',
|
|
'phase':'complete','observed_at':'2026-10-11T20:00:00Z','candidate':'a'*40,'rollback':'b'*40}
|
|
sink(event);DurableAudit(tmp_path/'audit.sqlite')(event)
|
|
with sink.connect() as db:assert db.execute('SELECT count(*) FROM events').fetchone()[0]==1
|
|
with pytest.raises(ValueError,match='conflict'):sink(event|{'phase':'failed'})
|
|
with pytest.raises(ValueError):sink(event|{'credential':'secret fixture canary'})
|