tegwick
919ce25a37
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a140c4-354a-7823-8f0d-8f717397f786
100 lines
6.6 KiB
Python
100 lines
6.6 KiB
Python
"""Offline tool regression using the producer's existing synthetic signer only."""
|
|
import copy
|
|
from datetime import datetime,timedelta,timezone
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
ROOT=Path(__file__).resolve().parents[1]
|
|
@unittest.skipUnless(os.environ.get('SECRETS_ENGINE_SOURCE_ROOT'), 'requires reviewed Secrets consumer source/bundle for integration checks')
|
|
class NativeSignerReceiptTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
global MODULE
|
|
spec=importlib.util.spec_from_file_location('signer_receipt',ROOT/'tools/verify_native_signer_receipt.py')
|
|
MODULE=importlib.util.module_from_spec(spec);spec.loader.exec_module(MODULE)
|
|
consumer=Path(os.environ['SECRETS_ENGINE_SOURCE_ROOT'])
|
|
cls.scratch=tempfile.TemporaryDirectory();cls.base=Path(cls.scratch.name)
|
|
env={**os.environ,'GOFLAGS':'-buildvcs=false'}
|
|
cls.verifier=cls.base/'verify';cls.signer=cls.base/'synthetic-sign'
|
|
for binary,package in [(cls.verifier,'.'),(cls.signer,'./testsupport/signer')]:
|
|
subprocess.run(['go','build','-p','2','-o',str(binary),package],cwd=consumer/'verifier',env=env,check=True)
|
|
cls.now=datetime.now(timezone.utc)
|
|
env=json.loads((consumer/'tests/fixtures/pdp-signature/decision_rotate_signed.json').read_text())
|
|
env['provenance']['caller']={'mode':'enforce','principal':'system:serviceaccount:secrets-engine:secrets-engine','audience':'flex-auth','not_after':(cls.now+timedelta(minutes=20)).isoformat()}
|
|
env['lifetime']['not_before']=(cls.now-timedelta(seconds=1)).isoformat();env['lifetime']['expires_at']=(cls.now+timedelta(minutes=15)).isoformat()
|
|
def sign(envelope):
|
|
return json.loads(subprocess.check_output([str(cls.signer)],input=json.dumps(envelope).encode()))
|
|
env['diagnostics']['target']={'phase':{'uses':1}}
|
|
cls.allow=sign(env)
|
|
req=json.loads((ROOT/'examples/secrets-engine/check_request_allow_rotate.json').read_text())
|
|
(cls.base/'allow-request.json').write_text(json.dumps(req))
|
|
denied=copy.deepcopy(req);denied['action']='execute-ccr'
|
|
cls.deny=copy.deepcopy(env);cls.deny['effect']='deny';cls.deny['binding']['action']='execute-ccr';cls.deny['binding']['submitted_request_digest']=MODULE.request_digest(denied);cls.deny=sign(cls.deny)
|
|
(cls.base/'deny-request.json').write_text(json.dumps(denied))
|
|
provenance=cls.allow['provenance'];cls.trust={key:provenance[key] for key in ('evaluator','mode','policy_package','policy_version','policy_package_digest')}
|
|
cls.trust.update(schema='secrets-engine.pdp-trust.v1',keys={'testdata-ed25519':'IVL40Zt5HSRFMkLhXy6rbLfP-ntqXtMAl5YOBpiB2xI'})
|
|
@classmethod
|
|
def tearDownClass(cls):cls.scratch.cleanup()
|
|
def setUp(self):
|
|
for label,envelope in [('allow',self.allow),('deny',self.deny)]:
|
|
for kind in ('envelope','record'):(self.base/f'{label}-{kind}.json').write_text(json.dumps(envelope))
|
|
(self.base/'trust.json').write_text(json.dumps(self.trust))
|
|
self.config={'schema':'access-engine.signer-receipt-check.v1','trust_file':str(self.base/'trust.json'),'verifier':str(self.verifier),'verifier_sha256':hashlib.sha256(self.verifier.read_bytes()).hexdigest(),'expected_caller':{'mode':'enforce','principal':'system:serviceaccount:secrets-engine:secrets-engine','audience':'flex-auth'}}
|
|
for label in ('allow','deny'):
|
|
for kind in ('request','envelope','record'):self.config[f'{label}_{kind}']=str(self.base/f'{label}-{kind}.json')
|
|
def test_genuine_signed_receipt_and_refusal_matrix(self):
|
|
result=MODULE.verify(self.config,now=self.now)
|
|
self.assertTrue(all(result['checks'].values()));self.assertEqual(len(result['checks']),7)
|
|
def test_changed_source_record_refuses(self):
|
|
record=copy.deepcopy(self.allow);record['effect']='deny';(self.base/'allow-record.json').write_text(json.dumps(record))
|
|
with self.assertRaises(ValueError):MODULE.verify(self.config,now=self.now)
|
|
def test_nested_record_number_type_change_refuses(self):
|
|
for replacement in (True,1.0):
|
|
with self.subTest(replacement_type=type(replacement).__name__):
|
|
record=copy.deepcopy(self.allow)
|
|
record['diagnostics']['target']['phase']['uses']=replacement
|
|
(self.base/'allow-record.json').write_text(json.dumps(record))
|
|
with self.assertRaises(ValueError):MODULE.verify(self.config,now=self.now)
|
|
|
|
def test_receipt_hashes_use_the_once_read_verified_bytes(self):
|
|
actual_read=MODULE.read
|
|
seen={}
|
|
original={str(self.base/f'{label}-{suffix}.json'):actual_read(str(self.base/f'{label}-{suffix}.json')) for label in ('allow','deny') for suffix in ('request','envelope','record')}
|
|
def once(path):
|
|
if path in original:
|
|
seen[path]=seen.get(path,0)+1
|
|
self.assertEqual(seen[path],1,'receipt was reopened after verification')
|
|
return actual_read(path)
|
|
with mock.patch.object(MODULE,'read',side_effect=once):
|
|
result=MODULE.verify(self.config,now=self.now)
|
|
self.assertEqual(len(seen),6)
|
|
for label in ('allow','deny'):
|
|
for suffix in ('request','envelope','record'):
|
|
self.assertEqual(result['input_sha256'][label+'_'+suffix],hashlib.sha256(original[str(self.base/f'{label}-{suffix}.json')]).hexdigest())
|
|
|
|
def test_oversized_file_read_is_bounded(self):
|
|
stream=mock.MagicMock();stream.__enter__.return_value=stream;stream.read.return_value=b'x'*524289
|
|
with mock.patch.object(Path,'open',return_value=stream):
|
|
with self.assertRaises(ValueError):MODULE.read('/absolute/oversized.json')
|
|
stream.read.assert_called_once_with(524289)
|
|
|
|
def test_unadmitted_helper_digest_refuses(self):
|
|
self.config['verifier_sha256']='0'*64
|
|
with self.assertRaises(ValueError):MODULE.verify(self.config,now=self.now)
|
|
def test_disabled_caller_admission_refuses(self):
|
|
self.config['expected_caller']['mode']='disabled'
|
|
with self.assertRaises(ValueError):MODULE.verify(self.config,now=self.now)
|
|
def test_expired_captured_allow_refuses(self):
|
|
with self.assertRaises(MODULE.DecisionError):MODULE.verify(self.config,now=self.now+timedelta(minutes=16))
|
|
def test_wrong_independent_policy_pin_refuses(self):
|
|
wrong=copy.deepcopy(self.trust);wrong['policy_package_digest']='sha256:'+'0'*64;(self.base/'trust.json').write_text(json.dumps(wrong))
|
|
with self.assertRaises(MODULE.DecisionError):MODULE.verify(self.config,now=self.now)
|
|
|
|
if __name__=='__main__':unittest.main()
|