#!/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
def verify_apple_image(entry):
"""Require the exact local matrix digest and Linux architecture; never a tag."""
image = json.loads(subprocess.check_output(
['container', 'image', 'inspect', entry['image']], text=True))[0]
digest = entry['image'].split('@', 1)[1]
if not any(v.get('digest') == digest and
v.get('platform', {}).get('architecture') == entry['architecture'] and
v.get('platform', {}).get('os') == 'linux'
for v in image.get('variants', [])):
raise RuntimeError('Apple Container candidate digest/platform mismatch; pre-pull the exact matrix digest')
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.runtime = None
if 'image' in entry:
self.runtime = ['container']
verify_apple_image(entry)
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.runtime else escape(str(self.root))
if self.runtime:
users.write_text(users.read_text().replace('<ip>127.0.0.1</ip>', '<ip>::/0</ip>'))
users_path = '/fixture/users.xml' if self.runtime else str(users)
listener = '0.0.0.0' if self.runtime else '127.0.0.1'
http_port = 8123 if self.runtime else self.port
tcp_port = 9000 if self.runtime 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.runtime:
name = deployment['name'] if deployment else 'clickhouse-migrate-' + uuid.uuid4().hex
self.container = name
subprocess.run(self.runtime + ['run', '--detach', '--name', name,
'--label', 'clickhouse-migrate.fixture=owned',
*(['--network', deployment['network']] if deployment else []),
'--user', str(os.getuid()), '--memory', '2g', '--cpus', '2',
'--platform', 'linux/' + entry['architecture'],
'--volume', str(self.root) + ':/fixture',
'-p', f'127.0.0.1:{self.port}:8123',
'-p', f'127.0.0.1:{self.tcp_port}:9000',
'--entrypoint', '/bin/sh', entry['image'], '-c', 'exec sleep 2147483647'],
check=True, capture_output=True, timeout=60)
if not (deployment and deployment.get('defer_start')):
self.ready()
except BaseException:
print(f"Owned server failure evidence: {self.root}", file=sys.stderr, flush=True)
self.close(False)
raise
def ready(self):
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 != self.entry['version']:
raise RuntimeError('Candidate server version mismatch')
@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.runtime:
return subprocess.check_output(self.runtime + ['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.runtime + ['exec', self.container, '/bin/sh', '-c', 'echo $$ > /fixture/owned-server.pid; exec clickhouse server --config-file=/fixture/config.xml'] if self.runtime 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.runtime:
subprocess.run(self.runtime + ['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.runtime:
subprocess.run(self.runtime + ['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:
# The private bind mount retains failure evidence on the host.
subprocess.run(self.runtime + ['delete', '--force', self.container], capture_output=True, timeout=30, 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)