__init__.py (10095B)
1 """A static pipeline laboratory, not a CI executor or hard resource sandbox.""" 2 from dataclasses import asdict, dataclass 3 import hashlib 4 from importlib.resources import files 5 import json 6 import os 7 from pathlib import Path 8 import re 9 import stat 10 11 from tricu import Library 12 from ._bundle_pin import BUNDLE_SHA256 13 14 PROTOCOL = 'tricu.workflow-demo.plan.v1' 15 MAX_TEXT = 65536 16 17 18 class ProtocolError(RuntimeError): 19 pass 20 21 22 class Denied(RuntimeError): 23 pass 24 25 26 @dataclass(frozen=True) 27 class Stage: 28 name: str 29 action: tuple 30 approval: bool = False 31 32 33 def identifier(value): 34 if not isinstance(value, str) or not re.fullmatch('[a-z][a-z0-9-]{0,31}', value): 35 raise ProtocolError('invalid stage name') 36 return value 37 38 39 def basename(value): 40 if not re.fullmatch('[a-z][a-z0-9._-]{0,63}', value): 41 raise ProtocolError('artifact names must be simple basenames') 42 return value 43 44 45 def validate_action(action): 46 if len(action) != 3 or action[0] not in ('write', 'check', 'publish'): 47 raise ProtocolError('unknown or malformed action') 48 basename(action[1]) 49 if action[0] == 'publish': 50 basename(action[2]) 51 elif len(action[2].encode('utf8')) > MAX_TEXT: 52 raise ProtocolError('artifact text exceeds 64 KiB') 53 return tuple(action) 54 55 56 def definition_id(stages): 57 payload = json.dumps([asdict(stage) for stage in stages], ensure_ascii=False, 58 sort_keys=True, separators=(',', ':')).encode('utf8') 59 return hashlib.sha256(PROTOCOL.encode()+b'\0'+payload).hexdigest() 60 61 62 class Session: 63 def __init__(self): 64 payload = files(__package__).joinpath('pipeline.arboricx').read_bytes() 65 if hashlib.sha256(payload).hexdigest() != BUNDLE_SHA256: 66 raise ProtocolError('packaged pipeline checksum mismatch') 67 self.context = Library().context() 68 try: 69 self.exports = {name: self.context.load_bundle(payload, name) 70 for name in ('pipeline', 'requireApproval', 'explain', 'execute')} 71 except BaseException: 72 self.context.close() 73 raise 74 75 def __enter__(self): return self 76 def __exit__(self, *_): self.close() 77 def close(self): self.context.close() 78 79 def _list(self, roots, value, limit): 80 result = [] 81 while self.context.kind(value) == 2: 82 if len(result) >= limit: 83 raise ProtocolError('list exceeds workflow protocol bound') 84 result.append(roots.child(value, 0)) 85 value = roots.child(value, 1) 86 if self.context.kind(value) != 0: 87 raise ProtocolError('improper workflow list') 88 return result 89 90 def _strings(self, roots, value, limit): 91 return [self.context.to_string(item) for item in self._list(roots, value, limit)] 92 93 def stages(self, roots, plan): 94 stages = [] 95 for value in self._list(roots, plan, 16): 96 fields = self._list(roots, value, 3) 97 if len(fields) != 3: 98 raise ProtocolError('invalid stage shape') 99 tag, name = (self.context.to_string(v) for v in fields[:2]) 100 guarded = tag == 'approval' 101 if guarded: 102 inner = self._list(roots, fields[2], 3) 103 if len(inner) != 3 or self.context.to_string(inner[0]) != 'stage' or self.context.to_string(inner[1]) != name: 104 raise ProtocolError('approval must wrap its named stage once') 105 fields = inner 106 elif tag != 'stage': 107 raise ProtocolError('unknown stage tag') 108 stages.append(Stage(identifier(name), validate_action(self._strings(roots, fields[2], 3)), guarded)) 109 if not stages or len({stage.name for stage in stages}) != len(stages): 110 raise ProtocolError('pipeline requires distinct stage names') 111 return stages 112 113 def plan(self, roots, text, approval=False): 114 if len(text.encode('utf8')) > MAX_TEXT: 115 raise ValueError('payload exceeds 64 KiB') 116 plan = roots.apply(self.exports['pipeline'], roots.string(text)) 117 if approval: 118 plan = self.guard(roots, plan, 'deploy') 119 return plan 120 121 def guard(self, roots, plan, target): 122 if target not in {stage.name for stage in self.stages(roots, plan)}: 123 raise ProtocolError('approval target not found') 124 return roots.apply_many(self.exports['requireApproval'], roots.string(target), plan) 125 126 def describe(self, roots, plan): 127 stages = self.stages(roots, plan) 128 explanation = self.context.to_string(roots.apply(self.exports['explain'], plan)) 129 return dict(protocol=PROTOCOL, definition_id=definition_id(stages), 130 executable_bundle_sha256=BUNDLE_SHA256, 131 stages=[asdict(stage) for stage in stages], explanation=explanation) 132 133 def _start(self, roots, plan): 134 return roots.apply(self.exports['execute'], plan) 135 136 def execute(self, roots, plan, executor): 137 stages = self.stages(roots, plan) 138 identity = definition_id(stages) 139 expected = [] 140 for stage in stages: 141 if stage.approval: 142 expected.append((stage.name, ('approve',))) 143 expected.append((stage.name, stage.action)) 144 cursor, events, failure = 0, [], None 145 approved = set() # A grant alone cannot manufacture a completed gate. 146 message = self._start(roots, plan) 147 while True: 148 parts = self._list(roots, message, 4) 149 if not parts: 150 raise ProtocolError('empty execution message') 151 tag = self.context.to_string(parts[0]) 152 if failure is not None: 153 if len(parts) != 3 or tag != 'failed' or self._strings(roots, message, 3)[1:] != list(failure): 154 raise ProtocolError('interpreter continued after a rejected action') 155 return dict(outcome='failed', definition_id=identity, events=events, 156 failed_stage=failure[0], reason=failure[1]) 157 if tag == 'done' and len(parts) == 1 and cursor == len(expected): 158 return dict(outcome='completed', definition_id=identity, events=events) 159 if tag != 'request' or len(parts) != 4 or cursor >= len(expected): 160 raise ProtocolError('unexpected execution message') 161 name = self.context.to_string(parts[1]) 162 action = tuple(self._strings(roots, parts[2], 3)) 163 if (name, action) != expected[cursor]: 164 raise ProtocolError('execution request differs from inspected declaration') 165 cursor += 1 166 try: 167 if action == ('approve',): 168 executor.approve(identity, name) 169 approved.add(name) 170 else: 171 executor.perform(name, action, approved) 172 response = ['ok'] 173 status = 'ok' 174 except (Denied, OSError, ValueError) as error: 175 status = 'denied' if isinstance(error, Denied) else 'error' 176 failure = (name, str(error)) 177 response = ['error', str(error)] 178 events.append(dict(stage=name, action=list(action), status=status)) 179 reply = roots.leaf() 180 for text in reversed(response): 181 reply = roots.fork(roots.string(text), reply) 182 message = roots.apply(parts[3], reply) 183 184 185 class WorkspaceExecutor: 186 """Explicit authority over bounded, create-only local artifact operations.""" 187 def __init__(self, workspace, capabilities=(), grants=()): 188 self.capabilities = frozenset(capabilities) 189 if not self.capabilities <= {'write', 'check', 'publish'}: 190 raise ValueError('unknown capability') 191 self.grants = frozenset(grants) # (definition ID, stage name) 192 self.fd = os.open(Path(workspace), os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) 193 194 def __enter__(self): return self 195 def __exit__(self, *_): self.close() 196 def close(self): 197 if self.fd is not None: 198 os.close(self.fd) 199 self.fd = None 200 201 def _active(self): 202 if self.fd is None: 203 raise Denied('executor is closed') 204 205 def approve(self, identity, stage): 206 self._active() 207 if (identity, stage) not in self.grants: 208 raise Denied('no approval grant for this definition and stage') 209 210 def _read(self, name): 211 fd = os.open(basename(name), os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=self.fd) 212 try: 213 if not stat.S_ISREG(os.fstat(fd).st_mode): 214 raise ValueError('artifact is not a regular file') 215 payload = bytearray() 216 while len(payload) <= MAX_TEXT: 217 part = os.read(fd, min(4096, MAX_TEXT+1-len(payload))) 218 if not part: 219 break 220 payload.extend(part) 221 if len(payload) > MAX_TEXT: 222 raise ValueError('artifact exceeds 64 KiB') 223 return bytes(payload) 224 finally: 225 os.close(fd) 226 227 def _write(self, name, payload): 228 if len(payload) > MAX_TEXT: 229 raise ValueError('artifact exceeds 64 KiB') 230 fd = os.open(basename(name), os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 231 0o600, dir_fd=self.fd) 232 with os.fdopen(fd, 'wb') as stream: 233 stream.write(payload) 234 235 def perform(self, name, action, approved): 236 self._active() 237 kind, source, argument = validate_action(action) 238 if kind not in self.capabilities: 239 raise Denied('capability not granted: '+kind) 240 if kind == 'publish' and name not in approved: 241 raise Denied('publication requires a completed approval boundary') 242 if kind == 'write': 243 self._write(source, argument.encode('utf8')) 244 elif kind == 'check': 245 if self._read(source) != argument.encode('utf8'): 246 raise ValueError('artifact check failed') 247 else: 248 self._write(argument, self._read(source))