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
from pathlib import Path

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


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):
    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':
        current = runtime.container()
        target = runtime.verified_upstream(current, config['routing'], runtime.tools['env_of'](current).get('INTERNAL_SERVICE_TOKEN', ''))
        return {'originalImageId': current['Image'], 'originalFingerprint': runtime.tools['fingerprint'](current),
                'sourceSha256': runtime.config_fingerprint(current, config['routing']), 'runtimeChanged': False,
                'upstreamImageId': target['Image'], 'upstreamFingerprint': runtime.tools['fingerprint'](target)}
    if action == 'prepare': return runtime.prepare(config)
    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):
    import fcntl
    validate_scope(config['scope'])
    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)


if __name__ == '__main__':
    try: print('INBOUND_RESOURCE_JSON=' + json.dumps({'ok': True, 'data': main(json.loads(base64.b64decode(sys.argv[1])))}))
    except Exception as error:
        message = str(error)
        print('INBOUND_RESOURCE_JSON=' + json.dumps({'ok': False, 'error': message if re.fullmatch(r'[A-Z_0-9]{1,90}', message) else 'INBOUND_RESOURCE_ACTION_FAILED'}))