#!/usr/bin/env python3
"""Owned native server and fault proxies, controlled over private stdin/stdout RPC."""
import argparse
import base64
import hashlib
import http.client
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import json
import os
from pathlib import Path
import shutil
import signal
import socket
import ssl
import subprocess
import sys
import tempfile
import threading
import time
import urllib.request
import urllib.parse
import uuid
from xml.sax.saxutils import escape
PASSWORD = 'test-password'
AUTH = 'Basic ' + base64.b64encode(('default:' + PASSWORD).encode()).decode()
def port():
with socket.socket() as listener:
listener.bind(('127.0.0.1', 0))
return listener.getsockname()[1]
def binary_for(entry):
if sys.platform != 'darwin' or os.uname().machine != 'arm64':
raise RuntimeError('Candidate matrix requires native Apple Silicon; Linux execution is deferred')
cache = Path.home() / 'Library/Caches/clickhouse-migrate' / entry['version']
cache.mkdir(parents=True, exist_ok=True)
binary = cache / 'clickhouse'
def digest(path):
with path.open('rb') as source:
return hashlib.file_digest(source, 'sha256').hexdigest()
if binary.exists() and digest(binary) == entry['sha256']:
return binary
with tempfile.NamedTemporaryFile(prefix='download-', dir=cache, delete=False) as destination:
candidate = Path(destination.name)
try:
with urllib.request.urlopen(entry['url'], timeout=60) as response:
shutil.copyfileobj(response, destination)
destination.flush()
if digest(candidate) != entry['sha256']:
raise RuntimeError('Server binary digest mismatch')
candidate.chmod(0o755)
candidate.replace(binary)
finally:
candidate.unlink(missing_ok=True)
return binary
class NativeServer:
def __init__(self, entry, deployment=None):
self.entry = entry
self.root = Path(tempfile.mkdtemp(prefix='clickhouse-migrate-server-', dir='/tmp'))
self.port = port()
self.tcp_port = port()
self.container = None
self.docker = None
if 'image' in entry:
context = os.environ.get('CLICKHOUSE_MIGRATE_DOCKER_CONTEXT')
if not context:
raise RuntimeError('OCI candidate requires an explicit isolated Docker context')
self.docker = ['docker', '--context', context]
endpoint = json.loads(subprocess.check_output(self.docker + ['context', 'inspect', context], text=True))[0]['Endpoints']['docker']['Host']
if not endpoint.startswith('unix://'):
raise RuntimeError('Owned OCI fixture requires a local Unix Docker socket')
image = json.loads(subprocess.check_output(self.docker + ['image', 'inspect', entry['image']], text=True))[0]
if image['Architecture'] != entry['architecture'] or image['Os'] != 'linux':
raise RuntimeError('OCI candidate platform mismatch; pre-pull the exact matrix digest')
self.binary = None
else:
self.binary = binary_for(entry)
self.process = None
self.log = (self.root / 'console.log').open('w')
self.proxies = []
self.keep = True
for directory in ['data', 'tmp', 'user_files', 'format_schemas']:
(self.root / directory).mkdir()
users = self.root / 'users.xml'
users.write_text('''<clickhouse><profiles><default><max_threads>2</max_threads><async_insert>1</async_insert><wait_for_async_insert>0</wait_for_async_insert><async_insert_busy_timeout_ms>10000</async_insert_busy_timeout_ms><mutations_sync>0</mutations_sync><alter_sync>0</alter_sync><fsync_metadata>0</fsync_metadata><max_result_rows>1</max_result_rows><result_overflow_mode>break</result_overflow_mode><max_block_size>1</max_block_size></default></profiles>
<users><default><password>test-password</password><networks><ip>127.0.0.1</ip></networks>
<profile>default</profile><quota>default</quota><access_management>1</access_management></default></users>
<quotas><default><interval><duration>3600</duration><queries>0</queries><errors>0</errors>
<result_rows>0</result_rows><read_rows>0</read_rows><execution_time>0</execution_time></interval></default></quotas></clickhouse>''')
root = '/fixture' if self.docker else escape(str(self.root))
if self.docker:
users.write_text(users.read_text().replace('<ip>127.0.0.1</ip>', '<ip>::/0</ip>'))
users_path = '/fixture/users.xml' if self.docker else str(users)
listener = '0.0.0.0' if self.docker else '127.0.0.1'
http_port = 8123 if self.docker else self.port
tcp_port = 9000 if self.docker else self.tcp_port
(self.root / 'config.xml').write_text(f'''<clickhouse>
<logger><level>warning</level><log>{root}/server.log</log><errorlog>{root}/error.log</errorlog></logger>
<path>{root}/data/</path><tmp_path>{root}/tmp/</tmp_path><user_files_path>{root}/user_files/</user_files_path>
<access_control_path>{root}/access/</access_control_path>
<format_schema_path>{root}/format_schemas/</format_schema_path><users_config>{users_path}</users_config>
<listen_host>{listener}</listen_host><http_port>{http_port}</http_port><tcp_port>{tcp_port}</tcp_port>
<max_connections>64</max_connections><max_thread_pool_size>128</max_thread_pool_size>
<background_schedule_pool_size>8</background_schedule_pool_size><background_buffer_flush_schedule_pool_size>2</background_buffer_flush_schedule_pool_size>
<background_message_broker_schedule_pool_size>2</background_message_broker_schedule_pool_size>
<background_distributed_schedule_pool_size>2</background_distributed_schedule_pool_size>
<background_pool_size>16</background_pool_size><background_move_pool_size>2</background_move_pool_size>
<background_fetches_pool_size>2</background_fetches_pool_size><background_common_pool_size>2</background_common_pool_size>
<mark_cache_size>0</mark_cache_size><uncompressed_cache_size>0</uncompressed_cache_size>
</clickhouse>''')
if deployment:
extra = deployment['config']
config = self.root / 'config.xml'
config.write_text(config.read_text().replace('</clickhouse>', extra + '</clickhouse>'))
try:
if self.docker:
name = deployment['name'] if deployment else 'clickhouse-migrate-' + uuid.uuid4().hex
subprocess.run(self.docker + ['create', '--name', name, '--label', 'clickhouse-migrate.fixture=owned',
*(['--network', deployment['network']] if deployment else []),
'--user', str(os.getuid()), '--memory', '2g', '--cpus', '2',
'-p', '127.0.0.1::8123', '-p', '127.0.0.1::9000',
'--entrypoint', '/bin/sh', entry['image'], '-c', 'exec sleep 2147483647'],
check=True, capture_output=True, timeout=20)
self.container = name
subprocess.run(self.docker + ['cp', str(self.root) + '/.', name + ':/fixture'], check=True, capture_output=True, timeout=20)
# Docker allocates ports on its own host, not the client OS.
# Keep the namespace alive while restarting only ClickHouse so
# these exact bindings remain reserved for the fixture lifetime.
subprocess.run(self.docker + ['start', name], check=True, capture_output=True, timeout=20)
self.port = int(subprocess.check_output(self.docker + ['port', name, '8123/tcp'], text=True, timeout=5).strip().rsplit(':', 1)[1])
self.tcp_port = int(subprocess.check_output(self.docker + ['port', name, '9000/tcp'], text=True, timeout=5).strip().rsplit(':', 1)[1])
self.start()
deadline = time.monotonic() + 30
while time.monotonic() < deadline:
if self.process.poll() is not None:
raise RuntimeError(f'Owned server exited before readiness: {self.process.returncode}')
try:
with urllib.request.urlopen(self.endpoint + '/ping', timeout=.2) as response:
if response.read() == b'Ok.\n':
break
except OSError:
pass
time.sleep(.05)
else:
raise RuntimeError('Owned server readiness deadline expired')
identity = self.query('SELECT version() AS version, toString(serverUUID()) AS server_id')[0]
version = identity['version']
if identity['server_id'] != self.retained_uuid():
raise RuntimeError('Endpoint is not the owned server data directory')
if version != entry['version']:
raise RuntimeError('Candidate server version mismatch')
except BaseException:
print(f"Owned server failure evidence: {self.root}", file=sys.stderr, flush=True)
self.close(False)
raise
@property
def endpoint(self):
return f'http://127.0.0.1:{self.port}'
def query(self, sql):
request = urllib.request.Request(self.endpoint + '/?default_format=JSONEachRow&wait_end_of_query=1&async_insert=0&mutations_sync=1&alter_sync=1&fsync_metadata=1&max_result_rows=0&max_result_bytes=0&result_overflow_mode=throw&read_overflow_mode=throw', data=sql.encode(), headers={'Authorization': AUTH}, method='POST')
with urllib.request.urlopen(request, timeout=10) as response:
return [json.loads(line) for line in response.read().splitlines() if line.strip()]
def retained_uuid(self):
if self.docker:
return subprocess.check_output(self.docker + ['exec', self.container, 'cat', '/fixture/data/uuid'], text=True, timeout=5).strip()
return (self.root / 'data/uuid').read_text().strip()
def start(self):
command = self.docker + ['exec', self.container, '/bin/sh', '-c', 'echo $$ > /fixture/owned-server.pid; exec clickhouse server --config-file=/fixture/config.xml'] if self.docker else [str(self.binary), 'server', '--config-file', str(self.root / 'config.xml')]
self.process = subprocess.Popen(command, stdout=self.log, stderr=subprocess.STDOUT, start_new_session=True)
def stop(self, sig):
if self.process is not None and self.process.poll() is None:
if self.docker:
subprocess.run(self.docker + ['exec', self.container, '/bin/sh', '-c', 'kill -s "$1" "$(cat /fixture/owned-server.pid)"', 'sh', signal.Signals(sig).name.removeprefix('SIG')], check=True, capture_output=True, timeout=10)
else:
os.killpg(self.process.pid, sig)
try:
self.process.wait(timeout=10)
except subprocess.TimeoutExpired:
if self.docker:
subprocess.run(self.docker + ['exec', self.container, '/bin/sh', '-c', 'kill -KILL "$(cat /fixture/owned-server.pid)"'], check=True, capture_output=True, timeout=10)
else:
os.killpg(self.process.pid, signal.SIGKILL)
self.process.wait(timeout=5)
def restart(self):
"""Kill/restart only this owned process, preserving its data/config/UUID."""
self.stop(signal.SIGKILL)
self.start()
deadline = time.monotonic() + 30
while time.monotonic() < deadline:
if self.process.poll() is not None:
raise RuntimeError('Owned restarted server exited')
try:
identity = self.query('SELECT version() AS version, toString(serverUUID()) AS server_id')[0]
if identity['version'] == self.entry['version'] and identity['server_id'] == self.retained_uuid():
return
except OSError:
pass
time.sleep(.05)
raise RuntimeError('Owned server restart deadline expired')
def close(self, success):
for proxy in self.proxies:
proxy.close()
try:
self.stop(signal.SIGTERM)
finally:
if self.container:
# Copy only this fixture's private data/logs before removing its
# recorded container. No ambient container or host volume is used.
if not success:
evidence = self.root / 'container-evidence'
evidence.mkdir()
subprocess.run(self.docker + ['cp', self.container + ':/fixture/.', str(evidence)], capture_output=True, timeout=30, check=True)
subprocess.run(self.docker + ['rm', '--force', self.container], capture_output=True, timeout=15, check=True)
self.log.close()
if success:
shutil.rmtree(self.root)
def leading_sql(body):
"""Classify without recording bodies; account for legal leading comments."""
while True:
body = body.lstrip()
if body.startswith((b'--', b'//', b'#!', b'# ')):
ends = [index for mark in (b'\r', b'\n') if (index := body.find(mark)) >= 0]
if not ends:
return b''
body = body[min(ends) + 1:]
elif body.startswith(b'/*'):
depth, cursor = 1, 2
while depth and cursor < len(body):
if body[cursor:cursor + 2] == b'/*':
depth += 1
cursor += 2
elif body[cursor:cursor + 2] == b'*/':
depth -= 1
cursor += 2
else:
cursor += 1
if depth:
return b''
body = body[cursor:]
else:
return body.upper()
class FaultProxy:
def __init__(self, target, tls=False):
self.target = target
self.tls = tls
self.mode = 'pass'
self.query_id = None
self.stage = None
self.step = None
self.rules = []
self.release = threading.Event()
self.identity_spoof = False
self.closed = False
self.events = []
self.open_connections = 0
self.lock = threading.Lock()
proxy = self
class Handler(BaseHTTPRequestHandler):
protocol_version = 'HTTP/1.1'
def setup(self):
super().setup()
with proxy.lock:
proxy.open_connections += 1
def finish(self):
try:
super().finish()
finally:
with proxy.lock:
proxy.open_connections -= 1
def log_message(self, *args):
pass
def do_POST(self):
body = self.rfile.read(int(self.headers.get('Content-Length', '0')))
with proxy.lock:
stage = None
step = None
header = leading_sql(body)
if header.startswith(b'CREATE DATABASE IF NOT EXISTS ') and header.endswith(b' ENGINE=ATOMIC'):
stage = 'ddl-database'
elif header.startswith(b'CREATE TABLE IF NOT EXISTS ') and b'(SEQUENCE UINT64, EVENT_ID UUID, LEDGER_ID UUID, EVENT STRING)' in header:
stage = 'ddl-table'
elif b'FORMAT JSONEachRow\n' in body:
try:
payload = json.loads(body.split(b'FORMAT JSONEachRow\n', 1)[1])
event = json.loads(payload['event'])
stage = event['kind']
context = event.get('context') or {}
step = (context.get('step') or {}).get('name')
except (ValueError, KeyError, TypeError):
pass
if header.startswith(b'SELECT TOSTRING(SEQUENCE) AS SEQUENCE, TOSTRING(EVENT_ID)'):
stage = 'history-read'
if stage is None and header.startswith((b'CREATE ', b'ALTER ', b'INSERT ', b'DROP ', b'TRUNCATE ', b'RENAME ', b'EXCHANGE ')):
stage = 'application-sql'
mode = proxy.mode
if proxy.stage is not None and stage != proxy.stage:
mode = 'pass'
query_id = urllib.parse.parse_qs(urllib.parse.urlsplit(self.path).query).get('query_id', [''])[0]
if proxy.query_id is not None and query_id != proxy.query_id:
mode = 'pass'
if proxy.step is not None and step != proxy.step:
mode = 'pass'
for rule in proxy.rules:
if rule['stage'] == stage and (rule.get('step') is None or rule['step'] == step):
mode = rule['mode']
if mode == 'delay-first':
mode = 'delay' if not proxy.events else 'pass'
if mode == 'close-second':
mode = 'pass' if not proxy.events else 'close'
event_index = len(proxy.events)
proxy.events.append({'path': self.path, 'mode': mode, 'stage': stage, 'connection': self.client_address[1], 'forwarded': mode not in ['redirect', 'close', 'hold-before'], 'bodyPrefixSent': False})
if mode == 'close':
self.close_connection = True
return
if mode == 'redirect':
self.send_response(307)
self.send_header('Location', proxy.redirect_url)
self.send_header('Content-Length', '0')
self.end_headers()
return
if mode == 'hold-before':
if not proxy.release.wait(timeout=15):
self.close_connection = True
return
with proxy.lock:
proxy.events[event_index]['forwarded'] = True
upstream = http.client.HTTPConnection('127.0.0.1', target.port, timeout=20)
try:
upstream.request('POST', self.path, body=body, headers={'Authorization': self.headers.get('Authorization', ''), 'Content-Type': 'text/plain'})
response = upstream.getresponse()
data = response.read()
headers = dict(response.getheaders())
status = response.status
finally:
upstream.close()
with proxy.lock:
proxy.events[event_index]['responseBytes'] = len(data)
proxy.events[event_index]['maxRowBytes'] = max(map(len, data.splitlines()), default=0)
if mode == 'drop':
self.close_connection = True
return
if mode == 'delay':
time.sleep(2)
if proxy.identity_spoof and data.strip():
rows = [json.loads(line) for line in data.splitlines() if line.strip()]
if rows and 'server_id' in rows[0]:
rows[0]['server_id'] = '00000000-0000-0000-0000-000000000000'
data = b''.join(json.dumps(row).encode() + b'\n' for row in rows)
if mode in ['unsupported-version', 'prerelease-version', 'weakened-policy']:
rows = [json.loads(line) for line in data.splitlines() if line.strip()]
if mode == 'unsupported-version':
rows[0]['version'] = '26.8.28.15'
elif mode == 'prerelease-version':
rows[0]['version'] = '26.8.28.14-rc1'
else:
rows[0]['mutations_sync'] = 0
data = b''.join(json.dumps(row).encode() + b'\n' for row in rows)
if mode == 'weakened-storage' and data.strip():
rows = [json.loads(line) for line in data.splitlines() if line.strip()]
for row in rows:
if 'is_remote' in row:
row['is_remote'], row['type'] = 1, 'ObjectStorage'
data = b''.join(json.dumps(row).encode() + b'\n' for row in rows)
if mode == 'append-late-error':
data += b'Code: 999. DB::Exception: SQL_SECRET_MARKER (SYNTHETIC_ERROR)\n'
if mode == 'oversized-history-row':
data += b'{"oversized":"' + b'x' * 1048577 + b'"}\n'
if mode in ['history-gap', 'history-identity-conflict', 'history-duplicate-conflict']:
rows = [json.loads(line) for line in data.splitlines() if line.strip()]
extra = dict(max(rows, key=lambda row: int(row['sequence'])))
event = json.loads(extra['event'])
if mode == 'history-gap':
event['sequence'] = int(event['sequence']) + 2
event['eventId'] = 'ffffffff-ffff-4fff-8fff-ffffffffffff'
elif mode == 'history-identity-conflict':
event['sequence'] = int(event['sequence']) + 1
else:
event['audit']['actor'] = 'conflicting-operator'
extra['sequence'], extra['event_id'] = str(event['sequence']), event['eventId']
extra['event'] = json.dumps(event, sort_keys=True, separators=(',', ':'), ensure_ascii=False)
rows.append(extra)
data = b''.join(json.dumps(row, ensure_ascii=False).encode() + b'\n' for row in rows)
if mode == 'late-error':
status, data = 200, b'Code: 999. DB::Exception: SQL_SECRET_MARKER (SYNTHETIC_ERROR)\n'
self.send_response(status)
for name in ['X-ClickHouse-Exception-Code', 'Content-Type']:
if name in headers:
self.send_header(name, headers[name])
if mode == 'switch-identity':
proxy.identity_spoof = True
self.send_header('Connection', 'close')
self.close_connection = True
self.send_header('Content-Length', str(len(data) + (20 if mode == 'truncate' else 0)))
self.end_headers()
try:
if mode == 'delay-history-body':
pivot = len(data) // 2
self.wfile.write(data[:pivot])
self.wfile.flush()
with proxy.lock:
proxy.events[event_index]['bodyPrefixSent'] = True
time.sleep(2)
self.wfile.write(data[pivot:])
else:
self.wfile.write(data)
self.wfile.flush()
except OSError:
pass
if mode == 'truncate':
self.close_connection = True
self.http = ThreadingHTTPServer(('127.0.0.1', 0), Handler)
if tls:
cert = target.root / ('proxy-cert-' + str(self.http.server_port) + '.pem')
key = target.root / ('proxy-key-' + str(self.http.server_port) + '.pem')
subprocess.run(['openssl', 'req', '-x509', '-newkey', 'rsa:2048', '-nodes', '-days', '1',
'-keyout', str(key), '-out', str(cert), '-subj', '/CN=localhost',
'-addext', 'subjectAltName=DNS:localhost,IP:127.0.0.1'],
check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
context.load_cert_chain(cert, key)
self.http.socket = context.wrap_socket(self.http.socket, server_side=True)
self.http.daemon_threads = True
self.http.block_on_close = False
self.redirect_url = self.endpoint
self.thread = threading.Thread(target=self.http.serve_forever, daemon=True)
self.thread.start()
@property
def endpoint(self):
return f'{"https" if self.tls else "http"}://127.0.0.1:{self.http.server_port}'
def close(self):
if self.closed:
return
self.closed = True
self.release.set()
self.http.shutdown()
self.http.server_close()
self.thread.join(timeout=2)
with self.lock:
evidence = {'receivedCount': len(self.events), 'events': list(self.events)}
(self.target.root / f'proxy-{self.http.server_port}-events.json').write_text(json.dumps(evidence, indent=2) + '\n')
def serve(matrix, entry_id, replicated=False):
def stop_signal(*_):
raise SystemExit(1)
signal.signal(signal.SIGTERM, stop_signal)
entries = json.loads(Path(matrix).read_text())['entries']
entry = next(value for value in entries if value['id'] == entry_id)
if replicated:
from replicated_fixture import ReplicatedServer
server = ReplicatedServer(entry)
else:
server = NativeServer(entry)
print(json.dumps({'endpoint': server.endpoint, 'root': str(server.root), 'version': entry['version']}), flush=True)
success = False
try:
for line in sys.stdin:
request = json.loads(line)
operation = request['operation']
try:
if operation == 'query-replica':
result = server.request(request['replica'], request['sql'], request.get('settings'), request.get('user', 'default'))
elif operation == 'stop-replica':
server.stop_replica(request['replica'])
result = None
elif operation == 'start-replica':
server.start_replica(request['replica'])
result = None
elif operation == 'restart':
server.restart()
result = None
elif operation == 'query':
result = server.query(request['sql'])
elif operation == 'proxy-start':
proxy = FaultProxy(server, bool(request.get("tls", False)))
server.proxies.append(proxy)
result = {'id': len(server.proxies)-1, 'endpoint': proxy.endpoint}
elif operation == 'proxy-stop':
server.proxies[request['id']].close()
result = None
elif operation == 'proxy-mode':
proxy = server.proxies[request['id']]
with proxy.lock:
proxy.mode = request['mode']
proxy.query_id = request.get('queryId')
proxy.stage = request.get('stage')
proxy.step = request.get('step')
proxy.rules.clear()
proxy.release.clear()
result = None
elif operation == 'proxy-rule':
proxy = server.proxies[request['id']]
with proxy.lock:
proxy.rules.append(request)
result = None
elif operation == 'proxy-release':
server.proxies[request['id']].release.set()
result = None
elif operation == 'proxy-open-connections':
proxy = server.proxies[request['id']]
with proxy.lock:
result = proxy.open_connections
elif operation == 'proxy-events':
proxy = server.proxies[request['id']]
with proxy.lock:
result = list(proxy.events)
elif operation == 'proxy-redirect':
server.proxies[request['id']].redirect_url = request['endpoint']
result = None
elif operation == 'stop':
success = bool(request['success'])
print(json.dumps({'result': None}), flush=True)
break
else:
raise RuntimeError('Unknown fixture operation')
print(json.dumps({'result': result}), flush=True)
except Exception:
print(json.dumps({'fixtureError': 'owned-fixture-operation-failed'}), flush=True)
finally:
server.close(success)
if __name__ == '__main__':
parser = argparse.ArgumentParser()
parser.add_argument('--matrix', required=True)
parser.add_argument('--entry', default='lts-native')
parser.add_argument('--replicated', action='store_true')
options = parser.parse_args()
serve(options.matrix, options.entry, options.replicated)