| 1 |
#!/usr/bin/env python3 |
| 2 |
"""Real HTTP workers on disposable Apache, Nginx/FPM, Caddy/FPM and OLS.""" |
| 3 |
import argparse |
| 4 |
import concurrent.futures |
| 5 |
import json |
| 6 |
import os |
| 7 |
from pathlib import Path |
| 8 |
import secrets |
| 9 |
import subprocess |
| 10 |
import time |
| 11 |
import urllib.error |
| 12 |
import urllib.parse |
| 13 |
import urllib.request |
| 14 |
|
| 15 |
HERE = Path(__file__).resolve().parent |
| 16 |
PROTECTED = """# BEGIN SyncBasalam Async |
| 17 |
<IfModule LiteSpeed> |
| 18 |
<IfModule mod_rewrite.c> |
| 19 |
RewriteEngine On |
| 20 |
RewriteCond %{REQUEST_URI} /wp-admin/admin-ajax\\.php$ |
| 21 |
RewriteCond %{QUERY_STRING} (^|&)action=sync_basalam_run_jobs_async(&|$) |
| 22 |
RewriteRule .* - [E=noabort:1] |
| 23 |
</IfModule> |
| 24 |
</IfModule> |
| 25 |
# END SyncBasalam Async |
| 26 |
""" |
| 27 |
SERVERS = { |
| 28 |
'apache': (8141, 'apache', ['apache'], 'apache2handler'), |
| 29 |
'nginx': (8142, 'fpm', ['fpm', 'nginx'], 'fpm-fcgi'), |
| 30 |
'caddy': (8143, 'caddy-fpm', ['caddy-fpm', 'caddy'], 'fpm-fcgi'), |
| 31 |
'ols': (8144, 'ols', ['ols'], 'litespeed'), |
| 32 |
} |
| 33 |
|
| 34 |
|
| 35 |
class Matrix: |
| 36 |
def __init__(self, args): |
| 37 |
self.args = args |
| 38 |
self.site = args.site_dir.resolve() |
| 39 |
self.output = args.results.resolve() |
| 40 |
self.output.mkdir(parents=True, exist_ok=True) |
| 41 |
self.token = secrets.token_hex(32) |
| 42 |
self.env = dict(os.environ, MATRIX_TOKEN=self.token, MATRIX_SITE_DIR=str(self.site)) |
| 43 |
self.command = ['docker', 'compose', '-f', str(HERE / 'compose.yml')] |
| 44 |
self.report = {'success': False, 'servers': [], 'source_commit': self.git_head()} |
| 45 |
self.assertions = 0 |
| 46 |
|
| 47 |
def git_head(self): |
| 48 |
return subprocess.check_output(['git', 'rev-parse', 'HEAD'], cwd=HERE, text=True).strip() |
| 49 |
|
| 50 |
def compose(self, *args, check=True): |
| 51 |
result = subprocess.run(self.command + list(args), env=self.env, text=True, capture_output=True) |
| 52 |
if check and result.returncode: |
| 53 |
raise RuntimeError(result.stdout + result.stderr) |
| 54 |
return result |
| 55 |
|
| 56 |
def http(self, fields, control=True, timeout=15): |
| 57 |
body = urllib.parse.urlencode(fields).encode() |
| 58 |
headers = {'X-Matrix-Token': self.token} if control else {} |
| 59 |
url = self.url if control else self.url + '?action=sync_basalam_run_jobs_async' |
| 60 |
request = urllib.request.Request(url, data=body, headers=headers) |
| 61 |
try: |
| 62 |
with urllib.request.urlopen(request, timeout=timeout) as response: |
| 63 |
return response.status, json.loads(response.read()) |
| 64 |
except urllib.error.HTTPError as error: |
| 65 |
raw = error.read() |
| 66 |
try: |
| 67 |
data = json.loads(raw) |
| 68 |
except json.JSONDecodeError: |
| 69 |
data = {'body': raw.decode(errors='replace')[:1000]} |
| 70 |
return error.code, data |
| 71 |
|
| 72 |
def control(self, op='state', **fields): |
| 73 |
status, data = self.http(dict(action='sync_basalam_matrix_control', op=op, **fields)) |
| 74 |
if status != 200 or not data.get('success'): |
| 75 |
raise RuntimeError(f'Control {op}: {status} {data}') |
| 76 |
self.last_state = data['data'] |
| 77 |
return data['data'] |
| 78 |
|
| 79 |
def worker(self, nonce=None): |
| 80 |
if nonce is None: |
| 81 |
nonce = self.control()['nonce'] |
| 82 |
try: |
| 83 |
return self.http(dict(action='sync_basalam_run_jobs_async', nonce=nonce), False, 90) |
| 84 |
except (OSError, ValueError) as error: |
| 85 |
return 0, {'error': str(error)} |
| 86 |
|
| 87 |
def wait(self, predicate, seconds=20): |
| 88 |
deadline = time.monotonic() + seconds |
| 89 |
last = None |
| 90 |
while time.monotonic() < deadline: |
| 91 |
last = self.control() |
| 92 |
if predicate(last): |
| 93 |
return last |
| 94 |
time.sleep(.15) |
| 95 |
raise RuntimeError(f'Timed out waiting for worker: {last}') |
| 96 |
|
| 97 |
def check(self, condition, description): |
| 98 |
if not condition: |
| 99 |
raise AssertionError(description) |
| 100 |
self.assertions += 1 |
| 101 |
|
| 102 |
def completed(self, key): |
| 103 |
state = self.wait(lambda s: s['events'].get(key, {}).get('returns') == 1 and s['lock_free']) |
| 104 |
self.check(not state['jobs'], f'{key}: success must delete the queue record') |
| 105 |
self.check(state['events'][key]['starts'] == 1, f'{key}: job executed exactly once') |
| 106 |
return state |
| 107 |
|
| 108 |
def ready(self): |
| 109 |
deadline = time.monotonic() + 45 |
| 110 |
while time.monotonic() < deadline: |
| 111 |
try: |
| 112 |
return self.control() |
| 113 |
except (OSError, ValueError, RuntimeError): |
| 114 |
time.sleep(.3) |
| 115 |
raise RuntimeError(f'HTTP server did not become ready: {self.url}') |
| 116 |
|
| 117 |
def run_server(self, name): |
| 118 |
port, backend, services, expected_sapi = SERVERS[name] |
| 119 |
self.url = f'http://127.0.0.1:{port}/wp-admin/admin-ajax.php' |
| 120 |
self.compose('up', '-d', *services) |
| 121 |
state = self.ready() |
| 122 |
self.check(state['sapi'] == expected_sapi, f'{name}: actual web SAPI must be {expected_sapi}') |
| 123 |
result = {'server': name, 'runtime': {k: state[k] for k in ['sapi', 'php', 'server']}, 'scenarios': []} |
| 124 |
start_assertions = self.assertions |
| 125 |
self.report['servers'].append(result) |
| 126 |
scenario = lambda text: (result['scenarios'].append(text), print(f'{name}: PASS {text}', flush=True)) |
| 127 |
|
| 128 |
self.control('reset') |
| 129 |
status, _ = self.worker('invalid-matrix-nonce') |
| 130 |
self.check(status == 403, f'{name}: invalid nonce must be rejected') |
| 131 |
status, body = self.worker() |
| 132 |
self.check(status == 200 and body['data']['processed'] == 0, 'Valid empty worker must return zero') |
| 133 |
scenario('real HTTP nonce rejection, acceptance and empty batch') |
| 134 |
|
| 135 |
self.control('reset') |
| 136 |
self.control('enqueue', key='success') |
| 137 |
status, body = self.worker() |
| 138 |
self.check(status == 200 and body['data']['processed'] == 1, 'Worker must execute one job') |
| 139 |
state = self.completed('success') |
| 140 |
self.check(state['notice'] is None, 'Healthy runtime must not alert') |
| 141 |
self.check(state['health']['last_returned_protected'] is (True if name == 'ols' else None), 'Protection must match the real runtime') |
| 142 |
self.check(state['autoload'] in ['off', 'no'], 'Health metadata must not autoload') |
| 143 |
result['healthy_evidence'] = state |
| 144 |
scenario('actual queue success, runtime protection, cleanup and non-autoloaded health') |
| 145 |
|
| 146 |
self.control('reset') |
| 147 |
self.control('enqueue', key='rejected', mode='reject') |
| 148 |
self.worker() |
| 149 |
state = self.control() |
| 150 |
self.check(state['jobs'][0]['status'] == 'failed' and '422' in state['jobs'][0]['error_message'], 'API rejection must be preserved') |
| 151 |
self.check(state['notice'] is None, 'API rejection must not produce a hosting alert') |
| 152 |
scenario('explicit API 422 remains separate from host interruption') |
| 153 |
|
| 154 |
self.control('reset') |
| 155 |
began = time.monotonic() |
| 156 |
self.control('enqueue', key='disconnected', seconds=3, dispatch=1) |
| 157 |
dispatch_seconds = time.monotonic() - began |
| 158 |
state = self.completed('disconnected') |
| 159 |
self.check(dispatch_seconds < 2.5, 'Nonblocking caller must return before the three-second job completes') |
| 160 |
self.check(state['events']['disconnected']['returned_at'] - state['events']['disconnected']['started_at'] >= 2.9, 'Background job must finish after client disconnection') |
| 161 |
self.check(state['notice'] is None, 'Completed background request must not alert') |
| 162 |
result['disconnect_evidence'] = {'caller_seconds': round(dispatch_seconds, 3), 'worker': state['events']['disconnected']} |
| 163 |
scenario('normal nonblocking WordPress loopback survives client disconnection') |
| 164 |
|
| 165 |
self.control('reset') |
| 166 |
job = self.control('enqueue', key='concurrent', seconds=3)['id'] |
| 167 |
with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool: |
| 168 |
futures = [pool.submit(self.worker) for _ in range(4)] |
| 169 |
self.wait(lambda s: not s['lock_free'] and s['events'].get('concurrent', {}).get('starts') == 1) |
| 170 |
self.control('age', id=job) |
| 171 |
html = self.control('notice')['html'] |
| 172 |
self.check('اجرای ه� |
| 173 |
گا� |
| 174 |
سازی نی� |
| 175 |
هکاره' not in html, 'Slow live worker must not be called interrupted') |
| 176 |
responses = [f.result() for f in futures] |
| 177 |
self.check(all(status == 200 for status, _ in responses), 'All concurrent requests must return HTTP 200') |
| 178 |
self.check(sorted(body['data']['processed'] for _, body in responses) == [0, 0, 0, 1], 'Only one runner may execute the job') |
| 179 |
self.completed('concurrent') |
| 180 |
scenario('four real HTTP workers, real advisory lock, no slow-worker false alarm') |
| 181 |
|
| 182 |
self.control('reset') |
| 183 |
job = self.control('enqueue', key='killed', seconds=60)['id'] |
| 184 |
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool: |
| 185 |
future = pool.submit(self.worker) |
| 186 |
state = self.wait(lambda s: not s['lock_free'] and 'killed' in s['events']) |
| 187 |
pid = int(state['events']['killed']['pid']) |
| 188 |
self.check(pid > 1, 'Kill must target a PHP worker, never container init') |
| 189 |
self.compose('exec', '-T', backend, 'sh', '-c', 'kill -9 "$1"', 'matrix-kill', str(pid)) |
| 190 |
future.result() |
| 191 |
self.wait(lambda s: s['lock_free']) |
| 192 |
self.control('age', id=job) |
| 193 |
html = self.control('notice')['html'] |
| 194 |
state = self.control() |
| 195 |
self.check(state['notice']['class'] == 'notice-error' and state['notice']['job_id'] == job, 'A later HTTP admin request must detect the dead worker') |
| 196 |
self.check(str(job) in html and 'role="alert"' in html, 'Admin notice must identify the job') |
| 197 |
self.check(('noabort' in html) == (name == 'ols'), 'Hosting advice must match actual runtime') |
| 198 |
self.check(self.control('notice', role='anonymous')['html'] == '', 'Anonymous callers cannot see the admin notice') |
| 199 |
result['kill_evidence'] = {'pid': pid, 'health': state['health'], 'notice': state['notice']} |
| 200 |
scenario('SIGKILL of actual HTTP PHP worker, independent detection and admin permissions') |
| 201 |
|
| 202 |
self.compose('restart', backend) |
| 203 |
self.ready() |
| 204 |
self.check(self.control()['notice']['job_id'] == job, 'Interruption evidence must persist across server restart') |
| 205 |
self.control('retry', id=job) |
| 206 |
self.worker() |
| 207 |
state = self.control() |
| 208 |
self.check(not state['jobs'] and state['notice'] is None, 'Normal stale recovery must complete the job and clear warning') |
| 209 |
recovered = state['events']['killed'] |
| 210 |
attempts = [entry['attempts'] for entry in recovered['history']] |
| 211 |
# A webserver may replay a failed upstream request. Each real execution |
| 212 |
# must be a distinct queue attempt; only the recovered attempt returns. |
| 213 |
self.check(recovered['starts'] >= 2 and recovered['returns'] == 1 and attempts == list(range(recovered['starts'])), 'Every recovery start must be a distinct queue attempt, with exactly one return') |
| 214 |
result['recovery_evidence'] = recovered |
| 215 |
scenario('restart persistence, real stale recovery and warning clearance') |
| 216 |
|
| 217 |
self.control('reset') |
| 218 |
self.control('lease') |
| 219 |
self.control('enqueue', key='orphaned-lease', dispatch=1) |
| 220 |
self.completed('orphaned-lease') |
| 221 |
scenario('legacy dispatch lock without expiration cannot block automatic execution') |
| 222 |
|
| 223 |
self.control('reset') |
| 224 |
job = self.control('enqueue', key='fatal', mode='fatal')['id'] |
| 225 |
self.worker() |
| 226 |
state = self.wait(lambda s: s['lock_free']) |
| 227 |
self.check(state['jobs'][0]['status'] == 'processing', 'Fatal PHP error must leave observable interrupted job') |
| 228 |
self.control('age', id=job) |
| 229 |
self.control('notice') |
| 230 |
self.check(self.control()['notice']['job_id'] == job, 'Fatal PHP error must be detected on a later admin request') |
| 231 |
self.control('retry', id=job) |
| 232 |
self.worker() |
| 233 |
self.check(self.control()['notice'] is None, 'Successful retry after fatal must clear interruption') |
| 234 |
scenario('actual PHP fatal, later detection and successful recovery') |
| 235 |
|
| 236 |
if name in ['nginx', 'caddy']: |
| 237 |
self.control('reset') |
| 238 |
self.compose('exec', '-T', backend, 'sh', '-c', "printf '%s\\n' '[www]' 'request_terminate_timeout = 4s' > /usr/local/etc/php-fpm.d/zz-matrix-timeout.conf") |
| 239 |
self.compose('restart', backend) |
| 240 |
self.ready() |
| 241 |
job = self.control('enqueue', key='fpm-timeout', seconds=10)['id'] |
| 242 |
status, _ = self.worker() |
| 243 |
self.check(status in [0, 502, 503, 504], 'FPM timeout must interrupt HTTP execution') |
| 244 |
state = self.wait(lambda s: s['lock_free']) |
| 245 |
self.check(state['events']['fpm-timeout']['returns'] == 0, 'FPM must actually terminate the worker before return') |
| 246 |
self.control('age', id=job) |
| 247 |
self.control('notice') |
| 248 |
self.check(self.control()['notice']['job_id'] == job, 'Resource timeout must produce an interruption alert') |
| 249 |
self.compose('exec', '-T', backend, 'rm', '/usr/local/etc/php-fpm.d/zz-matrix-timeout.conf') |
| 250 |
self.compose('restart', backend) |
| 251 |
self.ready() |
| 252 |
self.control('retry', id=job) |
| 253 |
self.worker() |
| 254 |
self.check(self.control()['notice'] is None, 'Recovery after FPM timeout must clear alert') |
| 255 |
scenario('real PHP-FPM request timeout, interruption alert and recovery') |
| 256 |
|
| 257 |
if name == 'ols': |
| 258 |
self.control('reset') |
| 259 |
(self.site / '.htaccess').write_text('') |
| 260 |
self.compose('restart', 'ols') |
| 261 |
self.ready() |
| 262 |
self.control('enqueue', key='unprotected') |
| 263 |
self.worker() |
| 264 |
state = self.completed('unprotected') |
| 265 |
self.check(state['notice']['class'] == 'notice-warning', 'Actual LSAPI without rule must report unverified protection') |
| 266 |
self.check(state['health']['last_returned_protected'] is False, 'Missing actual noabort flag must be recorded') |
| 267 |
result['unprotected_evidence'] = state |
| 268 |
scenario('actual LSAPI without htaccess protection warns even after fast completion') |
| 269 |
(self.site / '.htaccess').write_text(PROTECTED) |
| 270 |
self.compose('restart', 'ols') |
| 271 |
self.ready() |
| 272 |
self.control('enqueue', key='protection-restored', seconds=3, dispatch=1) |
| 273 |
state = self.completed('protection-restored') |
| 274 |
self.check(state['notice'] is None and state['health']['last_returned_protected'] is True, 'Restoring actual scoped rule must clear warning after real execution') |
| 275 |
scenario('restored scoped htaccess rule is observed in LSAPI and clears warning') |
| 276 |
|
| 277 |
self.control('reset') |
| 278 |
result['assertions'] = self.assertions - start_assertions |
| 279 |
result['success'] = True |
| 280 |
self.compose('stop', *services) |
| 281 |
|
| 282 |
def run(self): |
| 283 |
self.check((self.site / 'wp-load.php').is_file(), 'Supply a disposable WordPress core') |
| 284 |
self.check((self.site / 'wp-content/plugins/woocommerce/woocommerce.php').is_file(), 'Supply WooCommerce in the disposable site') |
| 285 |
if (self.site / 'wp-config.php').exists() and not ( |
| 286 |
(self.site / '.sync-basalam-matrix').is_file() and (self.site / 'wp-config.php').stat().st_size == 0 |
| 287 |
): |
| 288 |
raise RuntimeError('Refusing a directory containing wp-config.php; use an unconfigured disposable copy.') |
| 289 |
(self.site / '.sync-basalam-matrix').write_text('Disposable sync-basalam HTTP worker test site\n') |
| 290 |
(self.site / 'wp-content/mu-plugins').mkdir(parents=True, exist_ok=True) |
| 291 |
(self.site / 'wp-config.php').touch() |
| 292 |
(self.site / 'wp-content/mu-plugins/matrix.php').touch() |
| 293 |
(self.site / 'wp-content/plugins/sync-basalam').mkdir(exist_ok=True) |
| 294 |
(self.site / '.htaccess').write_text(PROTECTED) |
| 295 |
try: |
| 296 |
self.compose('up', '-d', '--wait', 'db', 'apache') |
| 297 |
self.compose('exec', '-T', 'apache', 'php', '/matrix/install.php') |
| 298 |
self.compose('stop', 'apache') |
| 299 |
for name in self.args.servers: |
| 300 |
self.run_server(name) |
| 301 |
self.report['success'] = True |
| 302 |
except BaseException as error: |
| 303 |
self.report['error'] = str(error) |
| 304 |
self.report['last_state'] = getattr(self, 'last_state', None) |
| 305 |
raise |
| 306 |
finally: |
| 307 |
self.report['assertions'] = self.assertions |
| 308 |
self.output.joinpath('results.json').write_text(json.dumps(self.report, indent=2, ensure_ascii=False)) |
| 309 |
self.output.joinpath('containers.log').write_text(self.compose('logs', '--no-color', check=False).stdout) |
| 310 |
images = subprocess.run(['docker', 'image', 'inspect', 'wordpress:php8.2-apache', 'wordpress:php8.2-fpm', 'nginx:1.29-alpine', 'caddy:2-alpine', 'litespeedtech/openlitespeed:1.9.3-lsphp83'], text=True, capture_output=True) |
| 311 |
if images.returncode == 0: |
| 312 |
self.output.joinpath('images.json').write_text(json.dumps([{k: image[k] for k in ['Id', 'RepoDigests', 'Architecture']} for image in json.loads(images.stdout)], indent=2)) |
| 313 |
if not self.args.keep: |
| 314 |
self.compose('down', '-v', check=False) |
| 315 |
print(f'PASS {len(self.report["servers"])} webservers, {self.assertions} assertions', flush=True) |
| 316 |
|
| 317 |
|
| 318 |
if __name__ == '__main__': |
| 319 |
parser = argparse.ArgumentParser(description=__doc__) |
| 320 |
parser.add_argument('--site-dir', type=Path, required=True) |
| 321 |
parser.add_argument('--results', type=Path, required=True) |
| 322 |
parser.add_argument('--servers', nargs='+', choices=SERVERS, default=list(SERVERS)) |
| 323 |
parser.add_argument('--keep', action='store_true', help='Keep disposable containers after the run for debugging') |
| 324 |
Matrix(parser.parse_args()).run() |
| 325 |
|
