114 lines
5.4 KiB
Python
114 lines
5.4 KiB
Python
"""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)
|