"""TEST-only one-shot resource-confirm fault for one frozen container batch. No Docker, SQL or business mutation is performed by this module. A separately reviewed runtime must install/restore routing and own the private configuration. The shared HTTP server/transport provides bounded bodies and a loopback control listener; only this Router decides which request may receive the synthetic 503. """ import argparse import hashlib import hmac import importlib.util import json import re import threading import time from pathlib import Path _spec = importlib.util.spec_from_file_location( 'dependency_proxy_http', Path(__file__).with_name('order-submit-dependency-fault-proxy.py')) _http = importlib.util.module_from_spec(_spec) _spec.loader.exec_module(_http) CONFIRM_PATH = '/internal/v1/warehouse-pallet-batches/confirm' class Router: def __init__(self, config, forward=_http.transport, clock=time.monotonic, record=None): self.config, self.forward, self.clock = config, forward, clock self.record = record or (lambda row: None) self.lock = threading.Lock() self.mode, self.deadline, self.generation = 'PASS', 0, 0 self.counts = {'forwarded': 0, 'faulted': 0} if config.get('environment') != 'TEST': raise ValueError('EXPLICIT_TEST_REQUIRED') for key in ('containerBookingId', 'warehouseId'): if not re.fullmatch(r'[1-9][0-9]{0,19}', str(config.get(key, ''))): raise ValueError('EXACT_CONTAINER_AND_WAREHOUSE_REQUIRED') if not config.get('lastlegIps') or not config.get('internalToken') or not config.get('controlToken'): raise ValueError('AUTHENTICATED_CALLER_SCOPE_REQUIRED') if not re.fullmatch(r'[a-f0-9]{64}', config.get('batchPrefixHash', '')): raise ValueError('REVIEWED_BATCH_PREFIX_REQUIRED') self.prefix = 'CIP:' + config['batchPrefixHash'] + ':' # Coordinator hashes ObjectMapper.writeValueAsString(String bookingId), including JSON quotes. expected = hashlib.sha256(json.dumps(str(config['containerBookingId']), separators=(',', ':')).encode()).hexdigest() if expected != config['batchPrefixHash']: raise ValueError('CONTAINER_BATCH_PREFIX_MISMATCH') def status(self): with self.lock: self._expire() return {'mode': self.mode, 'generation': self.generation, 'remainingSeconds': max(0, self.deadline - self.clock()), **self.counts} def _expire(self): if self.mode != 'PASS' and self.clock() >= self.deadline: self.mode, self.deadline = 'PASS', 0 def set_mode(self, token, mode, ttl=0): if not hmac.compare_digest(str(token), self.config['controlToken']): raise PermissionError('CONTROL_AUTH_REQUIRED') with self.lock: self.mode, self.deadline = 'PASS', 0 if mode not in ('PASS', 'CONFIRM_ONCE') or (mode != 'PASS' and (type(ttl) is not int or not 1 <= ttl <= 60)): raise ValueError('INVALID_MODE_OR_TTL') self.mode = mode self.deadline = self.clock() + ttl if mode != 'PASS' else 0 self.generation += 1 return self.status() def dispatch(self, method, path, headers, body, peer): normalized = {k.lower(): v for k, v in headers.items()} token = normalized.get('x-internal-service-token', '') authenticated = bool(token) and hmac.compare_digest(token, self.config['internalToken']) payload = None if method == 'POST' and path == CONFIRM_PATH: try: payload = json.loads(body) except (ValueError, UnicodeDecodeError): pass selected = (isinstance(payload, dict) and authenticated and peer in self.config['lastlegIps'] and str(payload.get('warehouseId')) == str(self.config['warehouseId']) and isinstance(payload.get('batchKey'), str) and re.fullmatch(re.escape(self.prefix) + r'[A-Za-z0-9_-]{1,80}', payload['batchKey']) is not None and isinstance(payload.get('commandKey'), str) and bool(payload['commandKey'])) with self.lock: self._expire() fault = selected and self.mode == 'CONFIRM_ONCE' generation = self.generation if fault: self.mode, self.deadline = 'PASS', 0 self.counts['faulted'] += 1 if fault: result = (503, {'Content-Type': 'text/plain'}, b'TEST selected resource confirmation unavailable') else: result = self.forward(self.config['upstream'], method, path, headers, body) with self.lock: self.counts['forwarded'] += 1 if selected: self.record({'method': method, 'path': path, 'matched': fault, 'forwarded': not fault, 'responseStatus': result[0], 'generation': generation, 'bodySha256': hashlib.sha256(body).hexdigest(), 'containerBookingId': self.config['containerBookingId'], 'warehouseId': self.config['warehouseId'], 'internalTokenValidated': True}) return result if __name__ == '__main__': parser = argparse.ArgumentParser() parser.add_argument('--config', required=True) # Reuse only the existing transport/server, not its Order routing policy. _http.Router = Router _http.main(parser.parse_args().config)