"""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()