Prepare signed lifecycle runtime admission and receipt verification

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a140c4-354a-7823-8f0d-8f717397f786

tests/test_native_signer_receipt.py

@@ -0,0 +1,100 @@
+"""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()