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

"""Explicit TEST/single-container runtime. Import performs no network or mutation.

Reuses the reviewed Order Docker journal/reconstruction/reconciliation methods;
only Lastleg's baiyue.warehouse.base-url is temporarily changed. No business/SQL.
"""
import ast
import base64
import copy
import hashlib
import json
import os
import re
import secrets
import signal
import socket
import subprocess
import sys
import time
import urllib.parse
import urllib.request
import urllib.error
from pathlib import Path

TARGET = 'firstleg-test-lastleg'
UPSTREAM = 'firstleg-test-warehouse-overseas'
PROPERTY = 'baiyue.warehouse.base-url'
STAGE = 'INITIALIZE'


def safe_diagnostic(error):
    """Never return exception messages, URLs, bodies, environment or identities."""
    stage, status = STAGE, None
    if isinstance(error, urllib.error.HTTPError):
        endpoint = urllib.parse.urlsplit(error.geturl()).path
        stage = {'/nacos/v3/auth/user/login': 'NACOS_LOGIN',
                 '/nacos/v3/admin/cs/config': 'NACOS_CONFIG_READ'}.get(endpoint, stage)
        status = error.code if type(error.code) is int and 100 <= error.code <= 599 else None
    kind = type(error).__name__
    diagnostic = {'stage': stage, 'errorType': kind if re.fullmatch(r'[A-Za-z_]{1,80}', kind) else 'Exception'}
    if isinstance(error, urllib.error.URLError) and not isinstance(error, urllib.error.HTTPError):
        reason = error.reason
        reason_type = type(reason).__name__
        if re.fullmatch(r'[A-Za-z_]{1,80}', reason_type): diagnostic['transportErrorType'] = reason_type
        number = getattr(reason, 'errno', None)
        if type(number) is int and -10000 <= number <= 10000: diagnostic['errno'] = number
    if type(status) is int and 100 <= status <= 599: diagnostic['httpStatus'] = status
    message = str(error)
    code = message if re.fullmatch(r'[A-Z_0-9]{1,90}', message) else 'INBOUND_RESOURCE_ACTION_FAILED'
    if status is not None: code = stage + '_HTTP_' + str(status)
    elif isinstance(error, urllib.error.URLError): code = stage + '_TRANSPORT_FAILED'
    return {'ok': False, 'error': code, 'diagnostic': diagnostic}


def validate_scope(scope):
    if set(scope) != {'environment', 'marker', 'containerBookingId', 'warehouseId', 'batchPrefixHash', 'planSha256'}:
        raise RuntimeError('INVALID_SCOPE')
    if scope['environment'] != 'TEST' or not re.fullmatch(r'[A-Za-z0-9_-]{1,80}', scope['marker']):
        raise RuntimeError('EXPLICIT_TEST_SCOPE_REQUIRED')
    if any(not re.fullmatch(r'[1-9][0-9]{0,19}', scope[key]) for key in ('containerBookingId', 'warehouseId')):
        raise RuntimeError('SINGLE_CONTAINER_WAREHOUSE_REQUIRED')
    expected = hashlib.sha256(json.dumps(scope['containerBookingId'], separators=(',', ':')).encode()).hexdigest()
    if scope['batchPrefixHash'] != expected or not re.fullmatch(r'[a-f0-9]{64}', scope['planSha256']):
        raise RuntimeError('BATCH_OR_PLAN_HASH_MISMATCH')


def runtime_class(base_source):
    # No __main__ executes. Only the known config import's dataId is adapted.
    tree = ast.parse(base_source)
    class Imports(ast.NodeTransformer):
        def visit_Constant(self, node):
            if node.value == 'order-service.yml': return ast.copy_location(ast.Constant('lastleg-service.yml'), node)
            return node
    tree = ast.fix_missing_locations(Imports().visit(tree))
    namespace = {'__name__': 'reviewed_resource_docker_runtime'}
    exec(compile(tree, '<reviewed-order-runtime>', 'exec'), namespace)
    namespace.update(ORDER=TARGET, PROPERTIES={'WAREHOUSE': PROPERTY}, validate_scope=validate_scope)
    Parent = namespace['Runtime']

    class Runtime(Parent):
        def __init__(self, scope, tools, base=Path('/tmp')):
            validate_scope(scope)
            self.scope, self.tools = scope, tools
            self.directory = base / ('chongchong-inbound-resource-' + scope['marker'])
            self.path = self.directory / 'state.json'
            self.lock = base / 'chongchong-inbound-resource.lock'

        def probes(self, state):
            target = self.verified_upstream(self.container(), state['routing'], state['internalToken'])
            if (target['Id'] != state['upstreamId'] or target['Image'] != state['upstreamImageId']
                    or self.tools['fingerprint'](target) != state['upstreamFingerprint']):
                raise RuntimeError('WAREHOUSE_UPSTREAM_CONFIGURATION_CHANGED')
            request = urllib.request.Request(state['upstreams']['WAREHOUSE'] + '/actuator/health',
                headers={'X-Internal-Service-Token': state['internalToken']})
            with namespace['http_open'](request, timeout=8) as response:
                value = json.load(response)
                if response.status != 200 or value.get('status') != 'UP':
                    raise RuntimeError('WAREHOUSE_UPSTREAM_HEALTH_FAILED')
            return {'WAREHOUSE': True}

        def verified_upstream(self, original, routing, token):
            target = self.tools['inspect'](UPSTREAM); self.tools['require_test'](target, UPSTREAM)
            target_env = self.tools['env_of'](target)
            address = urllib.parse.urlsplit(routing['upstreams']['WAREHOUSE'])
            ips = {n['IPAddress'] for name, n in target['NetworkSettings']['Networks'].items()
                   if name in original['NetworkSettings']['Networks']}
            if (address.scheme != 'http' or address.hostname not in ips or str(address.port) != target_env.get('SERVER_PORT')
                    or address.username or address.password or address.path not in ('', '/') or address.query or address.fragment
                    or target_env.get('DEPLOY_ENV') != 'test' or target['State'].get('Health', {}).get('Status') != 'healthy'
                    or len(token) < 16 or target_env.get('INTERNAL_SERVICE_TOKEN') != token):
                raise RuntimeError('UPSTREAM_NOT_VERIFIED_TEST_CONTAINER')
            return target

        def prepare(self, config):
            if self.directory.exists(): raise RuntimeError('STATE_EXISTS_USE_STATUS_OR_RESTORE')
            original = self.container()
            if original['Image'] != config.get('expectedImageId') or not re.fullmatch(r'sha256:[a-f0-9]{64}', original['Image']):
                raise RuntimeError('ORIGINAL_IMAGE_MISMATCH')
            if original['State'].get('Health', {}).get('Status') != 'healthy': raise RuntimeError('LASTLEG_NOT_HEALTHY')
            if original['Config']['Entrypoint'] != namespace['ENTRYPOINT'] or original['Config'].get('Cmd') not in (None, []):
                raise RuntimeError('UNREVIEWED_JAVA_ENTRYPOINT')
            if 'host.docker.internal:host-gateway' not in (original['HostConfig'].get('ExtraHosts') or []):
                raise RuntimeError('REVIEWED_PROXY_HOST_ROUTE_REQUIRED')
            env = self.tools['env_of'](original)
            token = env.get('INTERNAL_SERVICE_TOKEN', '')
            if len(token) < 16: raise RuntimeError('INTERNAL_TOKEN_UNAVAILABLE')
            routing = config['routing']
            if routing.get('configurationReviewed') is not True or set(routing['upstreams']) != {'WAREHOUSE'}:
                raise RuntimeError('REVIEWED_WAREHOUSE_ROUTING_REQUIRED')
            source = self.config_fingerprint(original, routing)
            if source != routing.get('sourceSha256'): raise RuntimeError('REVIEWED_NACOS_SOURCE_DIGEST_CHANGED')
            target = self.verified_upstream(original, routing, token)
            fd = os.open(self.lock, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
            with os.fdopen(fd, 'w') as stream: stream.write(self.scope['marker']); stream.flush(); os.fsync(stream.fileno())
            self.directory.mkdir(mode=0o700)
            sockets = [socket.socket(), socket.socket()]
            for sock in sockets: sock.bind(('127.0.0.1', 0))
            proxy_port, control_port = [sock.getsockname()[1] for sock in sockets]
            for sock in sockets: sock.close()
            state = {'scope': self.scope, 'phase': 'PREPARED', 'originalId': original['Id'], 'originalImageId': original['Image'],
                'originalFingerprint': self.tools['fingerprint'](original), 'originalBody': self.tools['recreation'](original, env, full=True),
                'internalToken': token, 'controlToken': secrets.token_urlsafe(32), 'proxyPort': proxy_port, 'controlPort': control_port,
                'upstreams': {'WAREHOUSE': routing['upstreams']['WAREHOUSE'].rstrip('/')}, 'routing': routing,
                'routingSourceSha256': source, 'lastlegIps': [n['IPAddress'] for n in original['NetworkSettings']['Networks'].values()],
                'upstreamId': target['Id'], 'upstreamImageId': target['Image'], 'upstreamFingerprint': self.tools['fingerprint'](target),
                'mutation': None, 'proxyLaunchIntent': False}
            self.save(state)
            self.probes(state)
            for name, field in [('proxy.py', 'proxySource'), ('order-submit-dependency-fault-proxy.py', 'httpProxySource')]:
                file = self.directory / name; file.write_bytes(base64.b64decode(config[field])); os.chmod(file, 0o600)
            namespace['private_write'](self.directory / 'proxy-config.json', {
                key: state[key] for key in ('internalToken', 'controlToken', 'proxyPort', 'controlPort', 'lastlegIps')} | {
                'environment': 'TEST', 'containerBookingId': self.scope['containerBookingId'], 'warehouseId': self.scope['warehouseId'],
                'batchPrefixHash': self.scope['batchPrefixHash'], 'upstream': state['upstreams']['WAREHOUSE']})
            state['phase'] = 'PROXY_START_INTENT'; state['proxyLaunchIntent'] = True; self.save(state)
            with open(self.directory / 'proxy-output.log', 'ab') as output:
                os.chmod(self.directory / 'proxy-output.log', 0o600)
                process = subprocess.Popen(['python3', str(self.directory / 'proxy.py'), '--config', str(self.directory / 'proxy-config.json')],
                    stdout=output, stderr=output, start_new_session=True)
            state['proxyPid'] = process.pid; state['phase'] = 'PROXY_STARTED'; self.save(state)
            return self.summary(state)

        def injected_entrypoint(self, state):
            value = copy.deepcopy(state['originalBody']['Entrypoint'])
            value[2] += ' --' + PROPERTY + '=http://host.docker.internal:' + str(state['proxyPort'])
            return value

        def matches_injected(self, current, state):
            view = copy.deepcopy(current); view['Config']['Entrypoint'] = state['originalBody']['Entrypoint']
            return (current['Image'] == state['originalImageId'] and current['Config']['Entrypoint'] == self.injected_entrypoint(state)
                    and self.tools['fingerprint'](view) == state['originalFingerprint'])

        def activate(self, state):
            if state['phase'] != 'PROXY_STARTED' or state.get('uncertain'): raise RuntimeError('ACTIVATE_NOT_RETRYABLE')
            if self.control(state)['mode'] != 'PASS': raise RuntimeError('PROXY_NOT_PASS')
            current = self.container()
            if current['Id'] != state['originalId'] or self.tools['fingerprint'](current) != state['originalFingerprint']:
                raise RuntimeError('ORIGINAL_CONFIGURATION_CHANGED')
            desired = copy.deepcopy(state['originalBody']); desired['Entrypoint'] = self.injected_entrypoint(state)
            self.replace(state, current, desired, 'INJECTED')
            state['injectedFingerprint'] = self.tools['fingerprint'](self.container()); self.save(state)
            return self.summary(state)

        def finish(self, state, _business=None):
            self.verify_restored(state)  # Includes healthy, full config/mount/network/image + Nacos + real upstream.
            if self.lock.is_symlink() or self.lock.read_text() != self.scope['marker']: raise RuntimeError('RUNTIME_LOCK_OWNER_CHANGED')
            pid = state.get('proxyPid')
            if not pid and state.get('proxyLaunchIntent'):
                file = self.directory / 'proxy.pid'
                if file.is_symlink() or not file.exists() or not file.read_text().isdigit(): raise RuntimeError('PROXY_PID_UNCERTAIN')
                pid = int(file.read_text())
            command = Path('/proc/' + str(pid) + '/cmdline') if pid else None
            if command and command.exists():
                if str(self.directory / 'proxy.py').encode() not in command.read_bytes().split(b'\0'):
                    raise RuntimeError('PROXY_PROCESS_OWNERSHIP_MISMATCH')
                os.kill(pid, signal.SIGTERM)
                for _ in range(40):
                    if not command.exists() or not command.read_bytes(): break
                    time.sleep(.1)
                else: raise RuntimeError('PROXY_NOT_STOPPED')
            state['restored'] = True; state['phase'] = 'FINISHED'
            for field in ('originalBody', 'desiredBody', 'internalToken', 'controlToken'): state.pop(field, None)
            self.save(state)
            file = self.directory / 'proxy-config.json'
            if file.exists(): file.unlink()
            self.lock.unlink()
            return {'restored': True, 'healthy': True, 'proxyStopped': True, 'lastlegRoutesVerified': True,
                    'originalImageId': state['originalImageId'], 'originalFingerprint': state['originalFingerprint']}
    return Runtime, namespace


def execute(config):
    global STAGE
    STAGE = 'LOAD_REVIEWED_PRIMITIVES'
    Runtime, namespace = runtime_class(base64.b64decode(config['baseSource']).decode())
    runtime = Runtime(config['scope'], namespace['common_tools'](base64.b64decode(config['commonSource']).decode()))
    action = config['action']
    if action == 'preflight':
        STAGE = 'LASTLEG_DOCKER_READ'
        current = runtime.container()
        STAGE = 'WAREHOUSE_UPSTREAM_IDENTITY'
        target = runtime.verified_upstream(current, config['routing'], runtime.tools['env_of'](current).get('INTERNAL_SERVICE_TOKEN', ''))
        STAGE = 'LASTLEG_FULL_FINGERPRINT'
        original_fingerprint = runtime.tools['fingerprint'](current)
        STAGE = 'NACOS_SOURCE_READ'
        source_hash = runtime.config_fingerprint(current, config['routing'])
        STAGE = 'WAREHOUSE_FULL_FINGERPRINT'
        upstream_fingerprint = runtime.tools['fingerprint'](target)
        return {'originalImageId': current['Image'], 'originalFingerprint': original_fingerprint,
                'sourceSha256': source_hash, 'runtimeChanged': False,
                'upstreamImageId': target['Image'], 'upstreamFingerprint': upstream_fingerprint}
    if action == 'prepare': return runtime.prepare(config)
    STAGE = 'RUNTIME_STATE_READ'
    state = runtime.load()
    if action == 'status':
        result = runtime.summary(state)
        if not state.get('restored'):
            try: result['proxy'] = runtime.control(state)
            except Exception: result['proxy'] = {'unavailable': True}
        return result
    if state.get('restored'): raise RuntimeError('ALREADY_FINISHED_USE_STATUS')
    if runtime.lock.is_symlink() or runtime.lock.read_text() != runtime.scope['marker']: raise RuntimeError('RUNTIME_LOCK_OWNER_CHANGED')
    if action == 'records':
        file = runtime.directory / 'proxy-evidence.jsonl'
        return {'records': [json.loads(line) for line in file.read_text().splitlines()] if file.exists() else []}
    if action == 'mode':
        if state['phase'] != 'INJECTED' or state.get('uncertain'): raise RuntimeError('FAULT_WINDOW_NOT_READY')
        mode, ttl = config.get('mode'), config.get('ttl', 0)
        if mode not in ('PASS', 'CONFIRM_ONCE') or (mode != 'PASS' and (type(ttl) is not int or not 1 <= ttl <= 60)):
            raise RuntimeError('INVALID_MODE_OR_TTL')
        return runtime.control(state, mode, ttl)
    if action == 'health': return {'healthy': runtime.container()['State'].get('Health', {}).get('Status') == 'healthy'}
    actions = {'activate': runtime.activate, 'restore': runtime.restore, 'reconcile': runtime.reconcile,
               'verify-restored': runtime.verify_restored, 'finish': runtime.finish}
    if action not in actions: raise RuntimeError('UNKNOWN_ACTION')
    return actions[action](state)


def main(config):
    validate_scope(config['scope'])
    import fcntl
    file = Path('/tmp') / ('chongchong-inbound-resource-' + config['scope']['marker'] + '.action.lock')
    if file.is_symlink(): raise RuntimeError('UNSAFE_ACTION_LOCK')
    fd = os.open(file, os.O_RDWR | os.O_CREAT, 0o600)
    try:
        try: fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError: raise RuntimeError('RUNTIME_ACTION_IN_PROGRESS_USE_STATUS_LATER')
        return execute(config)
    finally: os.close(fd)


def read_stdin_config(stream=None):
    """One bounded UTF-8 JSON stdin protocol; configuration is never argv."""
    global STAGE
    STAGE = 'READ_STDIN_CONFIG'
    raw = (sys.stdin.buffer if stream is None else stream).read(1024 * 1024 + 1)
    if len(raw) > 1024 * 1024: raise RuntimeError('RUNTIME_STDIN_CONFIG_TOO_LARGE')
    return json.loads(raw.decode('utf-8'))


if __name__ == '__main__':
    try: print('INBOUND_RESOURCE_JSON=' + json.dumps({'ok': True, 'data': main(read_stdin_config())}))
    except Exception as error:
        print('INBOUND_RESOURCE_JSON=' + json.dumps(safe_diagnostic(error)))