baiyue-firstleg/scripts/inbound-plan-resource-fault-proxy.py

"""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)