From 2a09eac23be1025e49bdbd65249b115ee3364f7d Mon Sep 17 00:00:00 2001 From: Vladimir Eremin Date: Sun, 11 Oct 2026 16:26:20 +0300 Subject: [PATCH] feat(replication): add authenticated readiness adapter Run a fixed application probe behind an authenticated local endpoint with bounded concurrency, deadlines, and no-store responses. Keep application safety checks and production enrollment explicit integration gates. Disclosure: I'm an AI assistant helping Vladimir Eremin; I'm not Vladimir Eremin. --- assets/composition-readiness/README.md | 27 +++ assets/composition-readiness/adapter.py | 167 +++++++++++++++++++ assets/composition-readiness/test_adapter.py | 53 ++++++ 3 files changed, 247 insertions(+) create mode 100644 assets/composition-readiness/README.md create mode 100644 assets/composition-readiness/adapter.py create mode 100644 assets/composition-readiness/test_adapter.py diff --git a/assets/composition-readiness/README.md b/assets/composition-readiness/README.md new file mode 100644 index 0000000..ddbd3f8 --- /dev/null +++ b/assets/composition-readiness/README.md @@ -0,0 +1,27 @@ +# Authenticated composition readiness adapter + +This dependency-free Python adapter exposes one local HTTP endpoint for the readiness-gated Worker and origin proxy. It runs an operator-supplied executable for every authenticated probe; exit0 means ready, every other outcome means unavailable. The executable must establish the application's actual readiness contract. A successful TCP connection or container health flag is not enough for replicated SSO. + +Example private configuration: + +```json +{ + "host": "127.0.0.1", + "port": 9092, + "path": "/.well-known/apex-ready", + "token_file": "/private/readiness-token", + "command": ["/usr/local/bin/python3", "/assets/probe.py", "--config", "/private/probe.json"], + "timeout": 1.5, + "slots": 4 +} +``` + +Run `python3 adapter.py --config /private/adapter.json` inside the application frontend namespace. Keep the token and probe credentials private. A64-character lowercase hexadecimal random token is suitable. This adapter does not need issuer private keys, a Docker socket, root, writable application data or network-administration capabilities. Bind only to loopback; route authorized origin readiness through a separately validated local proxy. + +Only GET at the exact canonical path, with one matching `X-APEX-Ready-Token` header, can execute the fixed command. Missing/wrong/duplicate credentials return403; other paths404; supported non-GET methods405. Ready returns204; probe failure, timeout or concurrency exhaustion returns503. These responses have empty bodies and `Cache-Control: no-store`. The adapter does not log requests, tokens or command output, cache success, or accept a client-supplied command. Each connection closes after its response. + +Probe timeout is bounded to1.5seconds so an origin can answer within the Worker's two-second deadline. A timeout kills the probe's process group. Probe slots and request threads are bounded; slow socket operations time out after two seconds. Operators must account for proxy overhead and actual probe execution time. Probe programs must not detach into a new session or launch background jobs. + +The SSO probe must verify the local Authelia instance/config generation, local authenticated SQL and directory access, completed migration, SQL Primary/Synced/write authority, and fresh session-controller authority plus actual verified TLS application access to its selected primary. It must reject uncertain/stale state and avoid inventing a competing election. A short-lived controller status file alone is not a consensus lease. The generic adapter supplies none of those application checks; do not deploy a placeholder-success probe or claim these tests establish HA. + +Run `python3 test_adapter.py` for loopback HTTP and subprocess behavior checks. No third-party packages are required. Production native image/probe/proxy integration and location-loss tests remain separate acceptance gates. diff --git a/assets/composition-readiness/adapter.py b/assets/composition-readiness/adapter.py new file mode 100644 index 0000000..830b7c0 --- /dev/null +++ b/assets/composition-readiness/adapter.py @@ -0,0 +1,167 @@ +"""Authenticated readiness endpoint for a fixed, bounded application probe.""" +import argparse +import hmac +import http.server +import json +import os +import pathlib +import re +import signal +import subprocess +import threading + + +class ReadinessServer(http.server.ThreadingHTTPServer): + daemon_threads = True + block_on_close = False + allow_reuse_address = True + + def __init__(self, address, *, token, path, command, timeout=1.5, slots=4): + if not isinstance(token, str) or not re.fullmatch(r'[!-~]{16,512}', token): + raise ValueError('A private visible-ASCII token is required') + if not isinstance(path, str) or not re.fullmatch(r'/[A-Za-z0-9._~/-]+', path) or any(part in ('', '.', '..') for part in path[1:].split('/')): + raise ValueError('A canonical absolute readiness path is required') + if not isinstance(command, list) or not command or not all(isinstance(arg, str) and '\0' not in arg for arg in command) or not os.path.isabs(command[0]): + raise ValueError('A fixed absolute executable and argv are required') + if isinstance(timeout, bool) or not 0 < timeout <= 1.5 or isinstance(slots, bool) or not isinstance(slots, int) or not 1 <= slots <= 16: + raise ValueError('Invalid probe bounds') + self.token = token.encode('ascii') + self.ready_path = path + self.command = command + self.timeout = timeout + self.probe_slots = threading.BoundedSemaphore(slots) + self.connection_slots = threading.BoundedSemaphore(slots * 2 + 4) + self.active = set() + self.active_lock = threading.Lock() + self.closing = False + super().__init__(address, Handler) + + def get_request(self): + connection, address = super().get_request() + connection.settimeout(2) + return connection, address + + def process_request(self, request, address): + if not self.connection_slots.acquire(blocking=False): + try: + request.sendall(b'HTTP/1.1 503 Service Unavailable\r\nCache-Control: no-store\r\nContent-Length: 0\r\nConnection: close\r\n\r\n') + except OSError: + pass + finally: + self.shutdown_request(request) + return + try: + super().process_request(request, address) + except BaseException: + self.connection_slots.release() + raise + + def process_request_thread(self, request, address): + try: + super().process_request_thread(request, address) + finally: + self.connection_slots.release() + + def handle_error(self, request, address): + pass + + @staticmethod + def stop_probe(process): + try: + os.killpg(process.pid, signal.SIGKILL) + except ProcessLookupError: + pass + process.wait() + + def probe(self): + if not self.probe_slots.acquire(blocking=False): + return False + process = None + try: + with self.active_lock: + if self.closing: + return False + process = subprocess.Popen(self.command, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True, close_fds=True) + self.active.add(process) + try: + return process.wait(timeout=self.timeout) == 0 + except subprocess.TimeoutExpired: + self.stop_probe(process) + return False + except (OSError, ValueError): + return False + finally: + if process is not None: + with self.active_lock: + self.active.discard(process) + self.probe_slots.release() + + def server_close(self): + with self.active_lock: + self.closing = True + active = list(self.active) + for process in active: + self.stop_probe(process) + super().server_close() + + +class Handler(http.server.BaseHTTPRequestHandler): + server_version = 'APEX' + sys_version = '' + + def log_message(self, *args): + pass + + def send_error(self, code, message=None, explain=None): + self.respond(code) + + def respond(self, status): + if self.request_version == 'HTTP/0.9': + self.request_version = 'HTTP/1.0' + self.send_response(status) + self.send_header('Cache-Control', 'no-store') + self.send_header('Content-Length', '0') + self.send_header('Connection', 'close') + self.end_headers() + self.close_connection = True + + def dispatch(self): + target = self.requestline.split()[1] + if target != self.server.ready_path or self.path != self.server.ready_path: + return self.respond(404) + if self.command != 'GET': + return self.respond(405) + headers = self.headers.get_all('X-APEX-Ready-Token', []) + supplied = headers[0].encode('latin-1') if len(headers) == 1 else b'' + if len(headers) != 1 or not hmac.compare_digest(supplied, self.server.token): + return self.respond(403) + return self.respond(204 if self.server.probe() else 503) + + do_GET = dispatch + do_HEAD = dispatch + do_POST = dispatch + do_PUT = dispatch + do_DELETE = dispatch + do_OPTIONS = dispatch + do_PATCH = dispatch + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument('--config', required=True) + args = parser.parse_args() + config = json.loads(pathlib.Path(args.config).read_text()) + token = pathlib.Path(config.pop('token_file')).read_text().strip() + address = (config.pop('host', '127.0.0.1'), config.pop('port', 9092)) + server = ReadinessServer(address, token=token, **config) + def terminate(signum, frame): + raise SystemExit(0) + signal.signal(signal.SIGTERM, terminate) + try: + server.serve_forever() + finally: + server.server_close() + + +if __name__ == '__main__': + main() diff --git a/assets/composition-readiness/test_adapter.py b/assets/composition-readiness/test_adapter.py new file mode 100644 index 0000000..adb86a5 --- /dev/null +++ b/assets/composition-readiness/test_adapter.py @@ -0,0 +1,53 @@ +import concurrent.futures,http.client,pathlib,socket,sys,tempfile,threading,time,unittest +from adapter import ReadinessServer +class AdapterTests(unittest.TestCase): + def setUp(self): + self.tmp=tempfile.TemporaryDirectory();self.root=pathlib.Path(self.tmp.name);self.marker=self.root/'ran';self.servers=[] + def tearDown(self): + for s,t in self.servers:s.shutdown();s.server_close();t.join() + self.tmp.cleanup() + def server(self,code=None,timeout=.3,slots=2): + code=code or f"from pathlib import Path;Path({str(self.marker)!r}).write_text('yes')" + server=ReadinessServer(('127.0.0.1',0),token='private-test-token',path='/.well-known/apex-ready',command=[sys.executable,'-c',code],timeout=timeout,slots=slots) + t=threading.Thread(target=server.serve_forever,kwargs={'poll_interval':.01},daemon=True);t.start();self.servers.append((server,t));return server + def request(self,s,path='/.well-known/apex-ready',token='private-test-token',method='GET',extra_token=False): + c=http.client.HTTPConnection(*s.server_address,timeout=3);c.putrequest(method,path) + if token is not None:c.putheader('X-APEX-Ready-Token',token) + if extra_token:c.putheader('X-APEX-Ready-Token',token) + c.endheaders();r=c.getresponse();out=(r.status,r.getheader('Cache-Control'),r.read());c.close();return out + def test_authenticated_success_runs_fixed_probe(self): + self.assertEqual(self.request(self.server()),(204,'no-store',b''));self.assertTrue(self.marker.exists()) + def test_absent_wrong_and_duplicate_tokens_never_probe(self): + s=self.server() + for token,duplicate in [(None,False),('bad',False),('private-test-token',True)]:self.assertEqual(self.request(s,token=token,extra_token=duplicate)[0],403) + self.assertFalse(self.marker.exists()) + def test_noncanonical_paths_and_methods_never_probe(self): + s=self.server() + for path in ['/other','/.well-known/apex-ready?x=1','/.well-known/%61pex-ready','//.well-known/apex-ready']: + self.assertEqual(self.request(s,path=path)[0],404) + self.assertEqual(self.request(s,method='POST')[0],405);self.assertFalse(self.marker.exists()) + def test_nonzero_exit_and_missing_executable_fail_closed(self): + s=self.server('raise SystemExit(1)');self.assertEqual(self.request(s),(503,'no-store',b''));s.command=['/nonexistent/apex-probe'];self.assertEqual(self.request(s)[0],503) + def test_timeout_terminates_probe_and_descendant(self): + code=f"import subprocess,time,sys;subprocess.Popen([sys.executable,'-c',\"import time,pathlib;time.sleep(.8);pathlib.Path({str(self.marker)!r}).write_text('leaked')\"]);time.sleep(3)" + s=self.server(code,timeout=.1);start=time.monotonic();self.assertEqual(self.request(s)[0],503);self.assertLess(time.monotonic()-start,.7);time.sleep(.9);self.assertFalse(self.marker.exists()) + def test_concurrency_limit_rejects_before_spawn(self): + code=f"from pathlib import Path;import time;Path({str(self.marker)!r}).write_text('yes');time.sleep(.4)" + s=self.server(code,timeout=1,slots=1) + with concurrent.futures.ThreadPoolExecutor() as pool: + first=pool.submit(self.request,s) + for _ in range(100): + if self.marker.exists():break + time.sleep(.005) + self.assertTrue(self.marker.exists());self.assertEqual(self.request(s)[0],503);self.assertEqual(first.result()[0],204) + def test_unsupported_method_is_empty_and_not_cached(self): + self.assertEqual(self.request(self.server(),method='TRACE'),(501,'no-store',b''));self.assertFalse(self.marker.exists()) + def test_parser_error_is_empty_and_not_cached(self): + s=self.server() + with socket.create_connection(s.server_address,timeout=2) as sock: + sock.sendall(b'GET /.well-known/apex-ready HTTP/banana\r\n\r\n') + r=http.client.HTTPResponse(sock);r.begin();self.assertEqual((r.status,r.getheader('Cache-Control'),r.read()),(400,'no-store',b'')) + self.assertFalse(self.marker.exists()) + def test_no_cached_readiness(self): + s=self.server();self.assertEqual(self.request(s)[0],204);s.command=[sys.executable,'-c','raise SystemExit(1)'];self.assertEqual(self.request(s)[0],503) +if __name__=='__main__':unittest.main()