Bound task execution by absolute call and credential deadlines

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

scripts/task_execution_native_fixture.py

@@ -9,7 +9,12 @@ Python privacy is not credential isolation; production needs the owner runtime.
 """
 from __future__ import annotations
 from dataclasses import dataclass, field
-from datetime import datetime, timezone
+from contextlib import contextmanager
+from functools import wraps
+import signal
+import math
+import threading
+from datetime import datetime, timezone, timedelta
 import hashlib
 import http.client
 import json
@@ -19,10 +24,63 @@ from pathlib import Path
 from urllib.parse import urlsplit
 
 from task_execution_cleanup import CleanupResult
+from task_execution_recovery import _same_policy,_policy_absent,_json_unique,_invalid_constant
 from task_execution_journal import Binding, Cleanup, Journal, Refused
 from task_execution_plan import Compiler, Plan, cleanup_specs, encoded, content_hash
 
 
+def _budget_expired(_signum, _frame):
+    # Keep blocked teardown interruptible after the first execution timeout.
+    # This is unwind-only grace, never a refreshed native-effect budget.
+    signal.setitimer(signal.ITIMER_REAL,.05)
+    raise Refused('absolute native call budget exhausted')
+
+
+@contextmanager
+def _absolute_budget(seconds):
+    """Disposable owner-child bound; refuse unsupported/conflicting execution.
+
+    SIGALRM interrupts slow source reads/compilation and HTTP header/body drips.
+    Nested bounds preserve the shorter outer deadline; no timer refresh extends
+    authority. This fixture assumes its owner main-thread process, not a shared
+    arbitrary-thread production runtime.
+    """
+    if (threading.current_thread() is not threading.main_thread() or type(seconds) not in {int,float}
+            or not math.isfinite(seconds) or seconds<=0):
+        raise Refused('bounded owner main-thread execution required')
+    previous=signal.getsignal(signal.SIGALRM)
+    old,interval=signal.getitimer(signal.ITIMER_REAL)
+    if interval or (old>0 and previous is not _budget_expired):
+        raise Refused('conflicting owner execution timer')
+    started=time.monotonic()
+    try:
+        signal.signal(signal.SIGALRM,_budget_expired)
+        signal.setitimer(signal.ITIMER_REAL,min(seconds,old) if old>0 else seconds)
+        yield
+    finally:
+        signal.setitimer(signal.ITIMER_REAL,0)
+        elapsed=time.monotonic()-started
+        signal.signal(signal.SIGALRM,previous)
+        if old>0:
+            if old<=elapsed:raise Refused('outer native authority budget exhausted')
+            signal.setitimer(signal.ITIMER_REAL,old-elapsed)
+
+
+def _business_window(method):
+    @wraps(method)
+    def bounded(self,*args,**kwargs):
+        remaining=self._remaining_business()-5
+        with _absolute_budget(remaining):return method(self,*args,**kwargs)
+    return bounded
+
+
+def _call_window(method):
+    @wraps(method)
+    def bounded(self,*args,**kwargs):
+        with _absolute_budget(2):return method(self,*args,**kwargs)
+    return bounded
+
+
 @dataclass(frozen=True)
 class PrivateSession:
     token: str = field(repr=False)
@@ -101,6 +159,7 @@ class NativeFixtureAdapter:
         if self.meta['cleanup'] != cleanup_specs(binding.operation_id): raise Refused('cleanup plan mismatch')
         self.task=None; self._mint_attempted=False; self._policy_attempted=False; self._ordinal=0; self._next_metadata=0
         self._cleanup_steps={item.kind:0 for item in self.cleanup}
+        self._task_wall_deadline=None;self._task_monotonic_deadline=None
         self.recovery_custody=recovery_custody
         if recovery_custody is not None:
             from task_execution_recovery import FixtureRecoveryCustody
@@ -112,9 +171,17 @@ class NativeFixtureAdapter:
         if not self.journal._held or self.journal._admission!=self.binding or self.journal._consumed_admission!=self.binding:
             raise Refused('same live consumed owner admission required')
 
+    def _remaining_business(self):
+        bounds=[(self.deadline-datetime.now(timezone.utc)).total_seconds(),self.monotonic_deadline-time.monotonic()]
+        task_wall=getattr(self,'_task_wall_deadline',None)
+        task_mono=getattr(self,'_task_monotonic_deadline',None)
+        if task_wall is not None:bounds.append((task_wall-datetime.now(timezone.utc)).total_seconds())
+        if task_mono is not None:bounds.append(task_mono-time.monotonic())
+        return min(bounds)
+
     def _fresh(self):
         self._live()
-        if min((self.deadline-datetime.now(timezone.utc)).total_seconds(),self.monotonic_deadline-time.monotonic())<=5:
+        if self._remaining_business()<=5:
             raise Refused('business deadline exhausted')
         # The constructor independently rebuilt the complete immutable Plan and
         # grant. Recheck every pinned byte before each native effect. Metadata
@@ -122,6 +189,7 @@ class NativeFixtureAdapter:
         # creation reconstructs and compares its fixed body before issuance.
         for path in self.compiler.pins:self.compiler.read(Path(path))
 
+    @_call_window
     def _http(self, method, path, token, body=b''):
         # Fixed caller-generated paths only; http.client has no proxy or redirect
         # following. Never surface backend body/header or transport exception text.
@@ -133,13 +201,15 @@ class NativeFixtureAdapter:
                          headers={'X-Vault-Token':token,'Content-Type':'application/json'})
             response=conn.getresponse(); raw=response.read(1024*1024+1)
             if len(raw)>1024*1024: raise Refused('oversize native response')
-            data=json.loads(raw) if raw else {}
+            data=json.loads(raw,object_pairs_hook=_json_unique,parse_constant=_invalid_constant) if raw else {}
             if type(data) is not dict: raise Refused('invalid native response')
             return response.status,data
         except Exception:
             raise Refused('native transport or response unconfirmed') from None
         finally: conn.close()
 
+    @_business_window
+    @_call_window
     def _business(self, method, path, token, payload=b'', task=False):
         self._fresh()
         if task:
@@ -165,6 +235,7 @@ class NativeFixtureAdapter:
         if status not in allowed or data.get('errors'):
             raise Refused('native request refused or unverified')
 
+    @_business_window
     def materialize_grant(self):
         self._fresh()
         if self.journal.observe(self.binding)['state']!='grant-issuing' or self._policy_attempted:
@@ -173,12 +244,12 @@ class NativeFixtureAdapter:
             raise Refused('insufficient lifetime for exact fixed task TTL')
         root=self.admission.owner_token; path='sys/policies/acl/'+self.policy_name
         status,data=self._business('GET',path,root)
-        if status!=404 or data.get('data'):
+        if not _policy_absent(status,data):
             raise Refused('per-operation grant policy must be absent')
         self._policy_attempted=True
         status,data=self._business('PUT',path,root,encoded({'policy':self.hcl}));self._ok(status,data,{200,204})
         status,data=self._business('GET',path,root);self._ok(status,data,{200})
-        if data.get('data',{}).get('policy','').rstrip()!=self.hcl.rstrip(): raise Refused('grant policy readback mismatch')
+        if not _same_policy(data.get('data',{}).get('policy'),self.hcl): raise Refused('grant policy readback mismatch')
         body={'policies':[self.policy_name],'no_default_policy':True,'renewable':False,
               'ttl':str(self.ttl)+'s','explicit_max_ttl':str(self.ttl)+'s','num_uses':self.uses,
               'type':'service','display_name':'task-exec-'+self.binding.operation_id,'no_parent':False}
@@ -187,6 +258,7 @@ class NativeFixtureAdapter:
         if self.recovery_custody is not None:
             self.recovery_custody.checkpoint(self.binding,'mint-intent')
         self._mint_attempted=True
+        minted_wall=datetime.now(timezone.utc);minted_mono=time.monotonic()
         status,data=self._business('POST','auth/token/create',root,encoded(body));self._ok(status,data,{200})
         auth=data.get('auth',{})
         if type(auth.get('client_token')) is not str or not auth['client_token'] or type(auth.get('accessor')) is not str or not auth['accessor']:
@@ -202,9 +274,19 @@ class NativeFixtureAdapter:
                 or info.get('num_uses')!=self.uses or not 0<info.get('ttl',0)<=self.ttl
                 or info.get('explicit_max_ttl')!=self.ttl or info.get('orphan') is not False):
             raise Refused('native task token bounds mismatch')
+        try:
+            expiry=datetime.fromisoformat(info['expire_time'].replace('Z','+00:00'))
+            if expiry.tzinfo is None or expiry.utcoffset() is None:raise ValueError()
+            self._task_wall_deadline=min(expiry,minted_wall+timedelta(seconds=self.ttl))
+            self._task_monotonic_deadline=min(minted_mono+self.ttl,
+                time.monotonic()+(self._task_wall_deadline-datetime.now(timezone.utc)).total_seconds())
+            if self._remaining_business()<=5:raise ValueError()
+        except (KeyError,TypeError,ValueError,AttributeError):
+            raise Refused('native task expiry unconfirmed or exhausted') from None
         self.journal.transition(self.binding,'grant-issuing','grant-issued')
         return {'issued':True,'policy_name':self.policy_name,'ttl_ceiling_seconds':self.ttl,'use_ceiling':self.uses}
 
+    @_business_window
     def metadata_call(self, ordinal):
         self._fresh()
         if self.task is None or self.journal.observe(self.binding)['state']!='applying' or type(ordinal) is not int or not 0<=ordinal<6 or ordinal!=self._next_metadata:
@@ -245,6 +327,7 @@ class NativeFixtureAdapter:
         # exact canonical JSON comparison. True never equals integer 1 here.
         return value
 
+    @_business_window
     def verify_metadata(self):
         self._fresh()
         if self._next_metadata!=6 or self.journal.observe(self.binding)['state']!='applying':
@@ -267,6 +350,7 @@ class NativeFixtureAdapter:
         if digest!=expected.call_digest or actual['call_digest']!=digest or actual['outcome']!='pending':
             raise Refused('exact pending cleanup obligation required')
 
+    @_call_window
     def _cleanup_http(self,kind,method,path,token,body=b''):
         ordinal=next(i for i,item in enumerate(self.cleanup) if item.kind==kind)
         obligation=self.cleanup[ordinal];step=self._cleanup_steps[kind]
@@ -307,5 +391,5 @@ class NativeFixtureAdapter:
         if self._policy_attempted:
             status,data=self._cleanup_http('grant-policy-delete','DELETE',path,self.admission.owner_token);self._ok(status,data,{200,204})
         status,data=self._cleanup_http('grant-policy-delete','GET',path,self.admission.owner_token)
-        confirmed=status==404 and not data.get('data')
+        confirmed=_policy_absent(status,data)
         return CleanupResult(confirmed,content_hash(encoded({'kind':'grant-policy-delete','absence_confirmed':confirmed})))