Read immutable policy closure with its exact custody pin

tests/Broker.NativeProof/RuntimeEvidence/fixtures/seed-attribution-helper/helper/growing_log.py

@@ -1,7 +1,7 @@
 """Held-FD adapter for one source-bound, growing Recoil infolog."""
 import base64,hashlib,json,os,re,selectors,stat,subprocess,time
 from pathlib import Path
-from private_io import Refused,SHA,atomic_bytes_new,atomic_new,canonical,components,hash_artifact,need,read_bytes
+from private_io import Refused,SHA,atomic_bytes_new,atomic_new,canonical,components,hash_artifact,need,read_bytes,read_immutable_closure
 from runtime_identity import mark_failure,proc_bytes,start_ticks
 MAX_LOG=10*1024*1024
 MAX_POLICY_INPUT=16*1024*1024
@@ -248,9 +248,9 @@ def _settle(self,deadline,probes,raw_mode=False,pending_raw=None):
         self.counter_cause='settlement-probe-cap'
         raise mark_failure(Refused('growing log record settlement exhausted'),'infolog-record-settlement-exhausted','refused',None)
     @staticmethod
-    def _closure(path,digest,deadline):
+    def _closure(path,digest,deadline,physical_pin=None):
         need(time.monotonic()<deadline,'policy closure deadline')
-        raw=read_bytes(path,1024*1024);need(hashlib.sha256(raw).hexdigest()==digest,'policy closure manifest digest/bound')
+        raw=read_bytes(path,1024*1024)if physical_pin is None else read_immutable_closure(path,digest,physical_pin,1024*1024);need(hashlib.sha256(raw).hexdigest()==digest,'policy closure manifest digest/bound')
         value=json.loads(raw.decode('utf-8'),object_pairs_hook=_unique)
         need(isinstance(value,dict) and set(value)=={'schema','custodyRoot','managedRoot','provenanceRoot','runtimeRoots','searchLayout','managed','provenance','runtime','runtimeRoles','source','custody'} and value.get('schema')==CLOSURE_SCHEMA,'policy closure schema')
         pins=[]

tests/Broker.NativeProof/RuntimeEvidence/fixtures/seed-attribution-helper/helper/private_io.py

@@ -96,3 +96,32 @@ def atomic_bytes_new(path,encoded,maximum):
         directory=os.open(path.parent,os.O_DIRECTORY|os.O_NOFOLLOW);os.fsync(directory);os.close(directory)
     finally:temporary.unlink()
     return digest(encoded)
+
+def read_immutable_closure(path,expected_sha256,physical_pin,maximum=1048576):
+    """Read the prior-bound canonical0400 closure under its immutable0500 root.
+    This entry accepts only the full already-selected physical pin; no mode relaxation.
+    """
+    need(type(maximum)is int and 0<maximum<=1048576,'immutable closure read bound')
+    fields={'path','sha256','bytes','device','inode','uid','mode','nlink'}
+    need(type(physical_pin)is dict and set(physical_pin)==fields and physical_pin['path']==path and physical_pin['sha256']==expected_sha256 and SHA.fullmatch(expected_sha256 or ''),'full prior canonical closure pin')
+    path=components(path);anchor=path.parent.lstat()
+    need(stat.S_ISDIR(anchor.st_mode)and anchor.st_uid==os.geteuid()and stat.S_IMODE(anchor.st_mode)==0o500,'immutable closure anchor')
+    before=path.lstat()
+    need(stat.S_ISREG(before.st_mode)and before.st_uid==os.geteuid()and stat.S_IMODE(before.st_mode)==0o400 and before.st_nlink==1 and 0<before.st_size<=maximum,'immutable closure custody')
+    def matches(info):
+        return all(physical_pin[k]==v for k,v in {'bytes':info.st_size,'device':info.st_dev,'inode':info.st_ino,'uid':info.st_uid,'mode':stat.S_IMODE(info.st_mode),'nlink':info.st_nlink}.items())
+    need(matches(before),'prior immutable closure physical identity')
+    fd=os.open(path,os.O_RDONLY|os.O_NOFOLLOW|os.O_NONBLOCK|os.O_CLOEXEC)
+    try:
+        first=os.fstat(fd);need(matches(first),'opened immutable closure identity')
+        chunks=[];length=0
+        while length<=maximum:
+            block=os.read(fd,min(65536,maximum+1-length))
+            if not block:break
+            chunks.append(block);length+=len(block)
+        last=os.fstat(fd);after=path.lstat();parent=path.parent.lstat()
+        attrs=('st_dev','st_ino','st_uid','st_mode','st_nlink','st_size','st_mtime_ns','st_ctime_ns')
+        need(length==before.st_size and length<=maximum and all(getattr(before,k)==getattr(first,k)==getattr(last,k)==getattr(after,k)for k in attrs),'immutable closure changed during read')
+        need(all(getattr(anchor,k)==getattr(parent,k)for k in ('st_dev','st_ino','st_uid','st_mode')),'immutable closure parent generation drift')
+        raw=b''.join(chunks);need(digest(raw)==expected_sha256,'immutable closure byte pin');return raw
+    finally:os.close(fd)

tests/Broker.NativeProof/RuntimeEvidence/fixtures/seed-attribution-helper/helper/seed_policy.py

@@ -143,6 +143,14 @@ def _clr(observed):
 class SeedGrowingLog(GrowingLog):
  def bind(self,context):
   need(type(context)in [SeedPolicyContext,PolicyOnlyContext] and not hasattr(self,'context'),'closed concrete policy context');_validate(context);self.context=context;return self
+ def _closure(self,path,digest,deadline):
+  ctx=self.context;entry,a=_validate(ctx)
+  # Both closed production contexts obtain the original pin from their sealed profile.
+  roster=ctx.profile['physicalPins']if type(ctx)is PolicyOnlyContext else ctx.profile['policy']['physicalPins']
+  need(type(roster)is list,'selected canonical physical roster')
+  selected=[pin for pin in roster if pin['path']==path]
+  need(len(selected)==1 and selected[0]['sha256']==digest,'exact unique prior closure pin')
+  return GrowingLog._closure(path,digest,deadline,selected[0])
  def _run_policy(self,path,closure_path,closure_sha,encoded,deadline):
   ctx=self.context;p=ctx.profile['policy'];need((path,closure_path,closure_sha)==(p['apphost'],p['closurePath'],p['closureSha256']),'literal policy input mismatch');need(p['environment']=={'PATH':'/usr/bin:/bin'} and len(encoded)<=MAX_POLICY_INPUT,'literal policy environment/input cap')
   invocation=hashlib.sha256(os.urandom(32)).hexdigest();child=None;sel=None;diagnostic=bytearray();all_stdout=bytearray();output=bytearray();ready=None;offset=0;count=0;check='policy-child-start';stdin_open=True;input_registered=False

tests/Broker.NativeProof/RuntimeEvidence/fixtures/seed-attribution-helper/helper/test_immutable_closure.py

@@ -0,0 +1,23 @@
+"""Portable file-custody controls; no policy/compiler/process effects."""
+import unittest,tempfile,pathlib,os,hashlib,stat
+from unittest.mock import patch
+import private_io as p
+class ImmutableClosure(unittest.TestCase):
+ def setUp(self):
+  self.tmp=tempfile.TemporaryDirectory();self.root=pathlib.Path(self.tmp.name);self.file=self.root/'closure.json';self.file.write_bytes(b'{"canonical":"controlled"}\n');self.file.chmod(0o400);self.root.chmod(0o500)
+  s=self.file.stat();self.pin=dict(path=str(self.file),sha256=hashlib.sha256(self.file.read_bytes()).hexdigest(),bytes=s.st_size,device=s.st_dev,inode=s.st_ino,uid=s.st_uid,mode=stat.S_IMODE(s.st_mode),nlink=s.st_nlink)
+ def tearDown(self):self.root.chmod(0o700);self.tmp.cleanup()
+ def read(self,pin=None):return p.read_immutable_closure(str(self.file),self.pin['sha256'],self.pin if pin is None else pin)
+ def refused(self,fn):
+  with self.assertRaises(p.Refused):fn()
+ def test_exact_prior_pin_readonly(self):self.assertEqual(self.read(),b'{"canonical":"controlled"}\n')
+ def test_parent_must_be_immutable(self):self.root.chmod(0o700);self.refused(self.read)
+ def test_file_must_be_immutable(self):self.file.chmod(0o600);self.refused(self.read)
+ def test_full_pin_required(self):self.refused(lambda:self.read({'path':str(self.file)}))
+ def test_inode_drift(self):self.refused(lambda:self.read(dict(self.pin,inode=self.pin['inode']+1)))
+ def test_uid_drift(self):self.refused(lambda:self.read(dict(self.pin,uid=self.pin['uid']+1)))
+ def test_byte_drift(self):self.file.chmod(0o600);self.file.write_bytes(b'X'*self.pin['bytes']);self.file.chmod(0o400);self.refused(self.read)
+ def test_link_refuses(self):self.root.chmod(0o700);os.link(self.file,self.root/'alias');self.root.chmod(0o500);self.refused(self.read)
+ def test_symlink_refuses(self):self.root.chmod(0o700);self.file.rename(self.root/'real');self.file.symlink_to(self.root/'real');self.root.chmod(0o500);self.refused(self.read)
+ def test_generic_private_reader_unchanged(self):self.refused(lambda:p.read_bytes(str(self.file)))
+if __name__=='__main__':unittest.main()

tests/Broker.NativeProof/RuntimeEvidence/fixtures/seed-attribution-helper/policy_transport_runner.py

@@ -11,7 +11,7 @@
 from growing_log import _policy_observation
 from runtime_identity import start_ticks
 from snapshot_publication import bytes_new,json_new
-ROOT='/tmp/bar-policy-transport-qualification-20261004';PYTHON='/usr/bin/python3.14'
+ROOT='/tmp/bar-policy-transport-qualification-20261005-v2';PYTHON='/usr/bin/python3.14'
 CASES=['stable-consume','malformed-request','withheld-input']
 LIMITS=dict(wholeSeconds=180,cleanupReserve=8,termSeconds=3,killSeconds=3,attemptBytes=134217728,attemptFiles=1024,mapsBytes=4194304,ownedIdentities=128,maximumEvaluations=3,boundarySeconds=5,outputBytes=131072)
 need=m.need

tests/Broker.NativeProof/RuntimeEvidence/fixtures/seed-attribution-helper/run-source-controls.py

@@ -13,11 +13,11 @@ def forbidden(*args,**kwargs):raise RuntimeError('portable source controls prohi
 os.system=forbidden
 for name in ['fork','forkpty','posix_spawn','posix_spawnp','execv','execve','execvp','execvpe']:
  if hasattr(os,name):setattr(os,name,forbidden)
-FILES=('test_policy_transport_runner.py','test_qualify_prearm.py','test_native_evidence.py','helper/test_prearm_startup.py','helper/test_seed_policy.py','helper/test_capture_custody.py','helper/test_loaded_custody.py','helper/test_settings_custody.py','helper/test_proc_bytes.py','fixture/tests/test_prearm_observation.py')
+FILES=('helper/test_immutable_closure.py','test_policy_transport_runner.py','test_qualify_prearm.py','test_native_evidence.py','helper/test_prearm_startup.py','helper/test_seed_policy.py','helper/test_capture_custody.py','helper/test_loaded_custody.py','helper/test_settings_custody.py','helper/test_proc_bytes.py','fixture/tests/test_prearm_observation.py')
 suite=unittest.TestSuite()
 for index,name in enumerate(FILES):
  spec=importlib.util.spec_from_file_location('portable_seed_controls_'+str(index),ROOT/name);module=importlib.util.module_from_spec(spec);spec.loader.exec_module(module);suite.addTests(unittest.defaultTestLoader.loadTestsFromModule(module))
-expected=89
+expected=99
 if suite.countTestCases()!=expected:raise SystemExit('portable source control roster changed: '+str(suite.countTestCases()))
 result=unittest.TextTestRunner(verbosity=2).run(suite)
 if not result.wasSuccessful()or result.testsRun!=expected or result.skipped:raise SystemExit(1)