284 lines
17 KiB
Python
284 lines
17 KiB
Python
"""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)))
|