coulomb/railiance-platform
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Execute task-plan catalogue validation from verified source bytes

Browse source 
Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a13690-09e0-7cd1-964b-ec83cdbcd348
This commit is contained in:
codex 2026-10-13 20:55:05 +02:00
parent 7c79293831
commit b98d17c09f
5 changed files with 139 additions and 11 deletions
Download patch file Download diff file Expand all files Collapse all files

55
scripts/task_execution_plan.py
View file

@ -5,15 +5,17 @@ Six fixed calls on two structure objects. Recompile immediately before each call
Canonical PDP request digest is opaque: the authenticated mediator verifies it.
"""
from __future__ import annotations
import builtins
import hashlib
import json
import os
from pathlib import Path
import re
import stat
import sys
import types
from dataclasses import dataclass
from uuid import UUID
from uuid import UUID, uuid4
import yaml
class Refused(Exception): pass
@ -123,6 +125,49 @@ class Compiler:
finally:
if fd is not None: os.close(fd)
def _catalogue_entry(self, snapshot):
# Execute the verified bytes, not cached imports or timestamp-valid .pyc
# files. Private names support dataclass annotation lookup without replacing
# the application's secrets_engine modules, even during concurrent compiles.
prefix = '_platform_catalogue_' + uuid4().hex
modules = {}
registered = []
def snapshot_import(name, globals=None, locals=None, fromlist=(), level=0):
if name == 'secrets_engine' or name.startswith('secrets_engine.'):
need(level == 0 and name in modules, 'unpinned_catalogue_import')
module = modules[name]
if fromlist:
need(all(hasattr(module, item) for item in fromlist), 'unpinned_catalogue_import')
return module
return modules['secrets_engine']
need(level == 0, 'unpinned_catalogue_import')
return builtins.__import__(name, globals, locals, fromlist, level)
try:
sources = [('secrets_engine', self.dependencies[0]),
('secrets_engine.errors', self.dependencies[1]),
('secrets_engine.redact', self.dependencies[2]),
('secrets_engine.catalog', self.catalog_code)]
for name, path in sources:
private_name = prefix + name[len('secrets_engine'):]
module = types.ModuleType(private_name)
module.__file__ = str(path)
module.__package__ = prefix
module.__dict__['__builtins__'] = dict(vars(builtins), __import__=snapshot_import)
modules[name] = module
sys.modules[private_name] = module
registered.append(private_name)
if name != 'secrets_engine':
setattr(modules['secrets_engine'], name.rsplit('.', 1)[1], module)
exec(compile(snapshot[str(path)], str(path), 'exec'), module.__dict__)
return modules['secrets_engine.catalog'].validate_entry(
yaml.load(snapshot[str(self.catalog)], Loader=UniqueLoader),
source='<owner-pinned-catalogue>')
finally:
for name in registered:
sys.modules.pop(name, None)
def _derive(self):
# No native operations, CLI applier, subprocess, network or ambient credentials.
snapshot={p:self.read(Path(p)) for p in self.pins}
@ -147,13 +192,7 @@ class Compiler:
generated=owner.generated_policy_hcl(ccr)
need(owner.normalized_policy_body(snapshot[str(policy)].decode())==owner.normalized_policy_body(generated),'owner_policy_not_generated')
# Use the actual consumer catalog validator/defaults, not an alternative broker catalog.
import secrets_engine.catalog as catalogue
import secrets_engine.errors as errors_module
import secrets_engine.redact as redact_module
import secrets_engine
for module,p in [(catalogue,self.catalog_code),(secrets_engine,self.dependencies[0]),(errors_module,self.dependencies[1]),(redact_module,self.dependencies[2])]:
need(Path(module.__file__).resolve()==p,'catalogue_runtime_source_mismatch')
entry=catalogue.validate_entry(yaml.load(snapshot[str(self.catalog)],Loader=UniqueLoader),source='<owner-pinned-catalogue>')
entry=self._catalogue_entry(snapshot)
need(entry.id==ccr['access_frontdoor']['catalog_id'] and entry.stage=='prod'
and ccr['target']['environment']=='production' and entry.kind=='kv'
and entry.mount_management=='existing'