diff --git a/CLAUDE.md b/CLAUDE.md index b1e28b6..06fb8a2 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -10,7 +10,7 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co - **Stick-table contract**: `python3 scripts/test-stick-table-contract.py` - offline; holds the templates' `store` clauses, `STICK_TABLE_FIELD_CONTRACT`, and every consumer to each other. Run it after touching any `stick-table` line. - **Runtime-map contract**: `python3 scripts/test-runtime-map-contract.py` - offline; asserts the runtime map commands are `@1`-prefixed, reference the map by FILE PATH (never `#`), carry the value `1`, and that every captured rejection is classified as a failure. Run it after touching any `add map`/`del map`/`clear map` path. - **Certificate destruction safety**: `python3 scripts/test-cert-write-safety.py` - offline; asserts a live `.pem` is never truncated, removed, or its certbot lineage deleted while any configured domain still references it (one bundle serves many names, so `ssl_cert_path` is routinely shared). Run it after touching any `os.remove`/`certbot delete`/PEM-write path. -- **API thread-pool safety**: `python3 scripts/test-gunicorn-socket-timeouts.py` - runs REAL gunicorn (~30s); asserts a silent/half-written client cannot pin a gthread pool thread (per-connection socket timeout, `HAPROXY_MGR_CLIENT_TIMEOUT`), slow legit requests are not cut off, the wedge watchdog (`HAPROXY_MGR_WEDGE_SECONDS`) replaces a fully stuck worker, and an in-process call to a stalled upstream times out. Run it after touching `haproxy_gunicorn_worker.py`, start-up.sh's gunicorn lines, or upgrading gunicorn (the worker overrides `ThreadWorker.enqueue_req`). +- **API thread-pool safety**: `python3 scripts/test-gunicorn-socket-timeouts.py` - runs REAL gunicorn (~60s); asserts a silent/half-written client cannot pin a gthread pool thread (per-connection socket timeout, `HAPROXY_MGR_CLIENT_TIMEOUT`), 2050 never-sending connections are reaped (`HAPROXY_MGR_FIRST_BYTE_TIMEOUT`), slowloris header drips are cut (`HAPROXY_MGR_HEADER_DEADLINE`), slow legit requests are not cut off, the wedge watchdog (`HAPROXY_MGR_WEDGE_SECONDS`) replaces a fully stuck worker, and an in-process call to a stalled upstream times out. Run it after touching `haproxy_gunicorn_worker.py`, start-up.sh's gunicorn lines, or upgrading gunicorn (the worker overrides gthread internals: `enqueue_req`, `accept`, `on_client_socket_readable`, `murder_keepalived`, `handle`, `handle_request`). - **Manual Testing**: Run `curl` commands against `http://localhost:8000` endpoints as shown in README.md ### Reading stick tables (and why it is easy to get silently wrong) @@ -168,6 +168,6 @@ re-reads it on reload, so a failed runtime command must degrade to - Stats page on port 8404 (localhost only inside container) - Management interface on port 8000 should be firewall-protected in production - Dockerfile HEALTHCHECK runs `scripts/healthcheck.sh`: port 8000 `/health` (4s max) AND port 80 (HAProxy). WHP's `lib-edge-container.sh` passes its own `--health-cmd`, which REPLACES the image's; it must call `/haproxy/scripts/healthcheck.sh` or the fleet never probes :8000 -- gunicorn uses `haproxy_gunicorn_worker.TimeoutThreadWorker` (gthread + client socket timeout + wedge watchdog). Stock gthread leaves client sockets blocking with NO timeout; `--timeout` only watches the main thread's heartbeat +- gunicorn uses `haproxy_gunicorn_worker.TimeoutThreadWorker` (gthread + client socket timeout + first-byte deadline + total header deadline + wedge watchdog). Stock gthread leaves client sockets blocking with NO timeout; `--timeout` only watches the main thread's heartbeat - Supports deployment on servers with git directory at `/root/whp` and web file sync via rsync to `/docker/whp/web/` - HAProxy is version 3.0.11 \ No newline at end of file diff --git a/haproxy_gunicorn_worker.py b/haproxy_gunicorn_worker.py index e497ad1..9b5d1ec 100644 --- a/haproxy_gunicorn_worker.py +++ b/haproxy_gunicorn_worker.py @@ -1,4 +1,4 @@ -"""gunicorn gthread worker that cannot be wedged by a silent client. +"""gunicorn gthread worker that cannot be wedged by silent or slow clients. Incident 2026-10-09 (sdbees), and very likely the 2026-07-07 whp01 one too ------------------------------------------------------------------------- @@ -16,11 +16,10 @@ blocked were gunicorn's own reads of the CLIENT socket: Once the main loop sees a connection become readable it hands it to a pool thread and turns the socket fully blocking -- timeout None -- before parsing the request. Any peer that sends a few bytes and then waits (a TLS ClientHello -sent to this plain-HTTP port, a port scanner, a half-written request, a body -shorter than its Content-Length) pins that pool thread forever. :8000 is -published on 0.0.0.0 by the deploy (`-p 8000:8000`, which bypasses firewalld), -and on sdbees internet scanners were measured connecting to it. Each such -connection leaks one thread; 4 per worker exhausts the pool. +sent to this plain-HTTP port, a port scanner, a half-written request) pins +that pool thread forever. :8000 was published on 0.0.0.0 by the deploy +(`-p 8000:8000`, which bypasses firewalld) and internet scanners were measured +connecting to it. 4 such connections exhaust a worker's pool. gunicorn's --timeout does not catch this: for gthread it only checks that the worker's MAIN thread keeps heart-beating, and the main thread keeps polling @@ -29,39 +28,78 @@ a worker whose threads never finish a request never reaches the limit. socket.setdefaulttimeout() does not help: setblocking(True) explicitly resets the timeout to None, overriding any default. -What this worker changes (and nothing else) -------------------------------------------- -1. Every client connection gets an explicit socket timeout - (HAPROXY_MGR_CLIENT_TIMEOUT, default 20s) at the exact point gthread makes it - blocking. It bounds each recv/send on the client socket -- request line, - headers, the body the app reads, and response writes. It does NOT bound how - long the app takes to handle a request (a long certbot run is a subprocess - with its own timeout and touches the client socket only once it is done). - A stalled peer now costs one thread for at most that long, then the - connection is closed and the thread is returned to the pool. +Stock gthread has two more exhaustion paths on the same port (review F1/F2): -2. A per-worker watchdog thread (HAPROXY_MGR_WEDGE_SECONDS, default 1200s; 0 - disables). If EVERY pool thread has been inside the same request for longer - than that -- i.e. the pool has not turned over at all for 20 minutes, longer - than the slowest legitimate request (certbot renew, 900s) -- the worker logs - every thread's Python stack to the error log and exits. The gunicorn master - respawns it in about a second, without touching HAProxy or the edge. - This is a backstop for blocking sources we have not found yet; the fix for - the known one is (1). +* A connection that NEVER sends a byte never becomes readable, so it is never + handed to a thread and never gets any deadline -- but it holds one of the + worker's ``worker_connections`` (1000) slots. At 1000 the run loop stops + polling altogether and only waits on futures; none exist, so even the + attacker's FINs are never seen and the worker is wedged PERMANENTLY, still + heart-beating, with no request in flight for a watchdog to notice. +* Slowloris: one header byte every monotonic start time + self._inflight = {} # thread ident -> _InFlight self._inflight_lock = threading.Lock() + self._pending = deque() # accepted, not yet readable (FIFO by deadline) # -- (1) client socket timeout ------------------------------------------ def enqueue_req(self, conn): @@ -132,21 +186,99 @@ class TimeoutThreadWorker(ThreadWorker): fs = self.tpool.submit(self.handle, conn) self._wrap_future(fs, conn) - # -- busy tracking -------------------------------------------------------- + # -- (2) first-byte deadline for never-readable connections --------------- + def accept(self, server, listener): + # ThreadWorker.accept (23.0.0) plus: a deadline, and the conn is + # tracked in self._pending so murder_keepalived() can reap it. + try: + sock, client = listener.accept() + conn = TConn(self.cfg, sock, client, server) + conn.timeout = time.time() + FIRST_BYTE_TIMEOUT + + self.nr_conns += 1 + with self._lock: + self._pending.append(conn) + self.poller.register(conn.sock, selectors.EVENT_READ, + partial(self.on_client_socket_readable, conn)) + except OSError as e: + if e.errno not in (errno.EAGAIN, errno.ECONNABORTED, + errno.EWOULDBLOCK): + raise + + def on_client_socket_readable(self, conn, client): + # ThreadWorker version only knows about self._keep; a first request + # comes off self._pending instead. + with self._lock: + self.poller.unregister(client) + queue = self._keep if conn.initialized else self._pending + try: + queue.remove(conn) + except ValueError: + # already reaped (race with murder_keepalived) + return + self.enqueue_req(conn) + + def murder_keepalived(self): + # Runs on every pass of ThreadWorker.run() -- including the branch + # taken when nr_conns >= worker_connections, where the poller is not + # serviced. Reaping here is what breaks that permanent deadlock. + super().murder_keepalived() + if FIRST_BYTE_TIMEOUT <= 0: + return + now = time.time() + while True: + with self._lock: + if not self._pending or self._pending[0].timeout > now: + break + conn = self._pending.popleft() + try: + self.poller.unregister(conn.sock) + except (OSError, KeyError, ValueError): + pass + self.nr_conns -= 1 + conn.close() + + # -- (3) header deadline + busy tracking ---------------------------------- def handle(self, conn): ident = threading.get_ident() + now = time.monotonic() + entry = _InFlight(conn, now, now + HEADER_DEADLINE if HEADER_DEADLINE > 0 else None) with self._inflight_lock: - self._inflight[ident] = time.monotonic() + self._inflight[ident] = entry try: return super().handle(conn) finally: with self._inflight_lock: self._inflight.pop(ident, None) + def handle_request(self, req, conn): + # The parser has produced a complete request line + headers. + entry = self._inflight.get(threading.get_ident()) + if entry is not None: + entry.header_deadline = None + return super().handle_request(req, conn) + + def _sweep_header_deadlines(self): + now = time.monotonic() + with self._inflight_lock: + late = [e for e in self._inflight.values() + if e.header_deadline is not None and e.header_deadline < now] + for e in late: + e.header_deadline = None # one shutdown per connection + try: + # shutdown, not close: wakes the blocked recv() with EOF + # without releasing the fd while another thread is in it. + e.conn.sock.shutdown(socket.SHUT_RDWR) + except OSError: + pass + if late: + self.log.info('Cut %d client connection(s) still sending request headers ' + 'after %ss (HAPROXY_MGR_HEADER_DEADLINE)', len(late), HEADER_DEADLINE) + def pool_status(self): now = time.monotonic() with self._inflight_lock: - starts = list(self._inflight.values()) + starts = [e.start for e in self._inflight.values()] return { 'pid': os.getpid(), 'threads': self.cfg.threads, @@ -155,13 +287,13 @@ class TimeoutThreadWorker(ThreadWorker): 'client_timeout_seconds': CLIENT_TIMEOUT, } - # -- (2) watchdog --------------------------------------------------------- + # -- (4) watchdog --------------------------------------------------------- def _wedged_for(self): """Seconds the pool has been completely wedged (every thread busy in its current request), measured from the YOUNGEST in-flight request. 0 when at least one thread is free.""" with self._inflight_lock: - starts = list(self._inflight.values()) + starts = [e.start for e in self._inflight.values()] if len(starts) < self.cfg.threads: return 0.0 return time.monotonic() - max(starts) @@ -175,11 +307,10 @@ class TimeoutThreadWorker(ThreadWorker): ident, names.get(ident, '?'), ''.join(traceback.format_stack(frame)))) return '\n'.join(out) - def _watchdog(self): - while self.alive: - time.sleep(WATCHDOG_INTERVAL) - wedged = self._wedged_for() - if wedged > WEDGE_SECONDS: + def _check_wedged(self): + wedged = self._wedged_for() + if WEDGE_SECONDS > 0 and wedged > WEDGE_SECONDS: + try: self.log.critical( 'WEDGED WORKER (pid %s): all %s threads busy in the same request ' 'for %.0fs > HAPROXY_MGR_WEDGE_SECONDS=%s. Exiting so the master ' @@ -191,11 +322,30 @@ class TimeoutThreadWorker(ThreadWorker): h.flush() except Exception: pass + finally: # os._exit, not SIGTERM: a graceful exit joins the (stuck, # non-daemon) pool threads at interpreter shutdown and would # hang until the master's --timeout escalates to SIGKILL. os._exit(WEDGED_EXIT_CODE) + def _monitor(self): + """One daemon thread: header-deadline sweep every second, wedge check + every WATCHDOG_INTERVAL. Never allowed to die silently.""" + last_wedge_check = time.monotonic() + while self.alive: + time.sleep(MONITOR_TICK) + try: + if HEADER_DEADLINE > 0: + self._sweep_header_deadlines() + if time.monotonic() - last_wedge_check >= WATCHDOG_INTERVAL: + last_wedge_check = time.monotonic() + self._check_wedged() + except Exception: + try: + self.log.exception('haproxy_gunicorn_worker monitor error (continuing)') + except Exception: + pass + def init_process(self): global _current_worker _current_worker = self @@ -203,7 +353,11 @@ class TimeoutThreadWorker(ThreadWorker): self.log.error_log.addFilter(_QuietClientTimeout()) except AttributeError: pass - if WEDGE_SECONDS > 0: - threading.Thread(target=self._watchdog, name='wedge-watchdog', - daemon=True).start() + if gunicorn.__version__ != TESTED_GUNICORN_VERSION: + self.log.warning( + 'haproxy_gunicorn_worker overrides gthread internals as of gunicorn %s; ' + 'running %s -- re-run scripts/test-gunicorn-socket-timeouts.py', + TESTED_GUNICORN_VERSION, gunicorn.__version__) + threading.Thread(target=self._monitor, name='worker-monitor', + daemon=True).start() super().init_process() # runs the event loop; does not return diff --git a/haproxy_manager.py b/haproxy_manager.py index 8d0bb00..91a9c15 100644 --- a/haproxy_manager.py +++ b/haproxy_manager.py @@ -7,6 +7,7 @@ import jinja2 import socket import psutil import functools +import hmac import logging from datetime import datetime, timedelta import json @@ -165,7 +166,12 @@ def require_api_key(f): def decorated_function(*args, **kwargs): if API_KEY: auth_header = request.headers.get('Authorization') - if not auth_header or auth_header != f'Bearer {API_KEY}': + # Constant-time comparison so response timing doesn't leak how + # much of a guessed key matched. Compared as bytes because + # compare_digest rejects non-ASCII str. + if not auth_header or not hmac.compare_digest( + auth_header.encode('utf-8', 'surrogateescape'), + f'Bearer {API_KEY}'.encode('utf-8', 'surrogateescape')): return jsonify({'error': 'Unauthorized - Invalid or missing API key'}), 401 return f(*args, **kwargs) return decorated_function diff --git a/scripts/start-up.sh b/scripts/start-up.sh index a700b3b..170dd26 100755 --- a/scripts/start-up.sh +++ b/scripts/start-up.sh @@ -94,11 +94,15 @@ MAX_REQ_JITTER="${HAPROXY_MGR_MAX_REQUESTS_JITTER:-100}" # Worker class (incident 2026-10-09): stock gthread makes every client socket # fully blocking with NO timeout, so any peer that sends a few bytes and goes # quiet (TLS hello on this plain-HTTP port, scanners -- :8000 is published on -# 0.0.0.0) pins a pool thread forever; 4 of them wedge a worker. This subclass -# sets a per-connection socket timeout (HAPROXY_MGR_CLIENT_TIMEOUT, default -# 20s) and runs a watchdog that exits a worker whose threads have ALL been -# stuck for HAPROXY_MGR_WEDGE_SECONDS (default 1200) so the master respawns it. -# See haproxy_gunicorn_worker.py. --timeout below does NOT cover either case: +# 0.0.0.0) pins a pool thread forever; 4 of them wedge a worker. Stock gthread +# also never reaps connections that send nothing (2x1000 of them wedge it +# permanently) and has no total header-read bound (slowloris). This subclass +# adds: a per-connection socket timeout (HAPROXY_MGR_CLIENT_TIMEOUT, 20s), a +# first-byte deadline (HAPROXY_MGR_FIRST_BYTE_TIMEOUT, 10s), a total +# request-line+headers deadline (HAPROXY_MGR_HEADER_DEADLINE, 10s), and a +# watchdog that exits a worker whose threads have ALL been stuck for +# HAPROXY_MGR_WEDGE_SECONDS (3600) so the master respawns it. +# See haproxy_gunicorn_worker.py. --timeout below does NOT cover any of these: # for gthread it only watches the main thread's heartbeat. WORKER_CLASS="haproxy_gunicorn_worker.TimeoutThreadWorker" diff --git a/scripts/test-gunicorn-socket-timeouts.py b/scripts/test-gunicorn-socket-timeouts.py index 438b6c4..243eebf 100644 --- a/scripts/test-gunicorn-socket-timeouts.py +++ b/scripts/test-gunicorn-socket-timeouts.py @@ -19,6 +19,10 @@ pin: * that a legitimately SLOW request (handling time > client timeout) is not cut off -- the timeout bounds socket I/O, not request duration; * the watchdog: a worker whose every thread is stuck is replaced; + * review round 1, at production shape (2 workers x 4 threads): 2050 + connections that never send a byte are reaped (F1, stock gthread wedges + permanently), and slowloris drip connections are cut by the total header + deadline while /health stays responsive (F2); * the timeout is logged as one line, not a traceback per scanner; * haproxy_manager's socket.setdefaulttimeout: an in-process client call to a stalled upstream (accepts, never replies) times out instead of hanging; @@ -29,7 +33,7 @@ Running python3 scripts/test-gunicorn-socket-timeouts.py stdlib unittest, like the other scripts/test-*.py; needs only the app's own -requirements (gunicorn, Flask...). Runs in the Docker build gate. ~25s. +requirements (gunicorn, Flask...). Runs in the Docker build gate. ~70s. """ import os @@ -89,7 +93,7 @@ def http_get(port, path='/', timeout=5): class Gunicorn: """A real gunicorn master on a free port, with its output captured.""" - def __init__(self, worker_class, threads=1, env=None): + def __init__(self, worker_class, threads=1, env=None, workers=1): self.tmp = tempfile.mkdtemp(prefix='gunicorn-timeout-test-') with open(os.path.join(self.tmp, 'tapp.py'), 'w') as fh: fh.write(WSGI_APP) @@ -101,7 +105,7 @@ class Gunicorn: self.proc = subprocess.Popen( [sys.executable, '-m', 'gunicorn', '--bind', '127.0.0.1:%d' % self.port, - '--workers', '1', '--threads', str(threads), + '--workers', str(workers), '--threads', str(threads), '--worker-class', worker_class, '--timeout', '120', '--graceful-timeout', '2', '--log-level', 'info', '--error-logfile', '-', @@ -290,6 +294,128 @@ class TimeoutWorkerTests(unittest.TestCase): pass +def raise_nofile(n): + try: + import resource + soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE) + if soft < n: + resource.setrlimit(resource.RLIMIT_NOFILE, (min(n, hard), hard)) + return resource.getrlimit(resource.RLIMIT_NOFILE)[0] + except (ImportError, ValueError, OSError): + return None + + +class ProductionShapeExhaustion(unittest.TestCase): + """Review round 1 (F1/F2), at the production shape: 2 workers x 4 threads, + default worker_connections=1000.""" + + def start(self, env): + g = Gunicorn(FIXED, threads=4, workers=2, env=env) + self.addCleanup(g.stop) + return g + + def test_f1_2050_idle_conns_reaped_and_health_recovers(self): + # Stock gthread: 2x1000 never-readable conns fill every slot, the run + # loop stops polling, and the worker stays dead even after the + # attacker closes everything. Here they must be reaped after + # FIRST_BYTE_TIMEOUT -- even while the attacker still holds them open. + if (raise_nofile(4500) or 0) < 4300: + self.skipTest('RLIMIT_NOFILE too low for 2050 client sockets') + g = self.start({'HAPROXY_MGR_FIRST_BYTE_TIMEOUT': '3'}) + socks = [] + self.addCleanup(lambda: [x.close() for x in socks]) + for _ in range(2050): + s = socket.socket() + s.settimeout(2) + s.connect(('127.0.0.1', g.port)) + socks.append(s) + # Attacker keeps every socket open. Within FIRST_BYTE_TIMEOUT plus the + # time to drain the backlog, the API must answer again. + deadline = time.time() + 20 + ok = None + while time.time() < deadline: + try: + ok = http_get(g.port, '/', timeout=3) + break + except Exception: + time.sleep(0.5) + self.assertIsNotNone(ok, '/health never recovered while idle conns were held') + held = time.time() + # The reaped sockets were closed by the server. + closed = sum(1 for x in socks[:1500] if wait_closed_by_server(x, 0.01)) + self.assertGreater(closed, 1000, 'server did not close idle conns (%d)' % closed) + for x in socks: + x.close() + socks.clear() + time.sleep(1) + for _ in range(5): + self.assertTrue(http_get(g.port, '/', timeout=3).startswith('pid=')) + self.assertLess(held, time.time()) + + def test_f2_drip_connections_cut_and_health_responsive_throughout(self): + # 8 connections (= every thread at production shape) each drip one + # header byte per second -- far under the 20s per-recv timeout, so only + # the total HEADER_DEADLINE can cut them. + g = self.start({'HAPROXY_MGR_HEADER_DEADLINE': '3'}) # CLIENT_TIMEOUT stays 20 + stop = threading.Event() + cut_after = [] + + def drip(): + s = socket.create_connection(('127.0.0.1', g.port)) + t0 = time.time() + req = b'GET / HTTP/1.1\r\nX-a: ' + b'a' * 1000 + try: + for i in range(len(req)): + if stop.is_set(): + return + s.send(req[i:i + 1]) + time.sleep(1.0) + s.settimeout(0.01) + try: + if s.recv(1) == b'': + raise ConnectionResetError + except socket.timeout: + pass + except OSError: + cut_after.append(time.time() - t0) + finally: + s.close() + + drippers = [threading.Thread(target=drip, daemon=True) for _ in range(8)] + for t in drippers: + t.start() + self.addCleanup(stop.set) + time.sleep(0.5) + worst = 0.0 + t_end = time.time() + 12 + while time.time() < t_end: + t0 = time.time() + self.assertTrue(http_get(g.port, '/', timeout=6).startswith('pid=')) + worst = max(worst, time.time() - t0) + time.sleep(0.5) + stop.set() + self.assertEqual(len(cut_after), 8, 'not every drip connection was cut: %r' % cut_after) + # The deadline is armed when a THREAD picks the connection up, not at + # accept: a request queued behind busy threads (e.g. legit certbot + # runs) must not be cut for waiting. So a drip conn that was queued + # (the 2 workers rarely split 8 conns 4/4) is cut at up to about + # 2 x deadline; one that got a thread at once within ~deadline + 1s. + self.assertLessEqual(min(cut_after), 3 + 2, cut_after) + self.assertTrue(all(c < 2 * 3 + 2 for c in cut_after), cut_after) + self.assertLess(worst, 5.0) + self.assertIn('HAPROXY_MGR_HEADER_DEADLINE', g.output()) + + def test_f2_slow_but_complete_headers_within_deadline_are_served(self): + g = self.start({'HAPROXY_MGR_HEADER_DEADLINE': '4'}) + s = socket.create_connection(('127.0.0.1', g.port)) + self.addCleanup(s.close) + for part in (b'GET / HTTP/1.1\r\n', b'Host: x\r\n', b'\r\n'): + s.sendall(part) + time.sleep(1) + s.settimeout(5) + self.assertIn(b'200 OK', s.recv(4096)) + + class StalledUpstreamDefaultTimeout(unittest.TestCase): """haproxy_manager sets socket.setdefaulttimeout() at import. An in-process client call to an upstream that accepts and never replies must time out."""