diff --git a/CLAUDE.md b/CLAUDE.md index 06fb8a2..ad30849 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 (~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`). +- **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`, counted from enqueue), 40 reconnecting drippers cannot starve legit requests, the :8000 peer filter (`HAPROXY_MGR_API_ALLOWED_PEERS`) refuses non-allowed peers, 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,7 @@ 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 +- **The :8000 API refuses peers other than 127.0.0.1/::1 and the container's default gateway** (`HAPROXY_MGR_API_ALLOWED_PEERS`, default `127.0.0.1,::1,gateway`; ports via `HAPROXY_MGR_PEER_FILTER_PORTS`, default `8000`). haproxy-manager shares `client-net` with every tenant container; host callers arrive from the gateway via docker-proxy. Only effective with the WHP `-p 127.0.0.1:8000:8000` publish (a 0.0.0.0 publish lets a tenant hairpin through the host and arrive from the gateway). A sidecar that must call :8000 directly needs its IP added - 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 9b5d1ec..7b56603 100644 --- a/haproxy_gunicorn_worker.py +++ b/haproxy_gunicorn_worker.py @@ -56,17 +56,20 @@ What this worker changes the main loop, freeing its slot (review F1). 3. HEADER_DEADLINE (HAPROXY_MGR_HEADER_DEADLINE, default 10s): total time a - pool thread may spend reading the request line + headers. Armed when the - thread picks the connection up, disarmed in handle_request() (which gunicorn - calls only once the parser has returned a complete request). A 1s monitor - loop shuts down (SHUT_RDWR, not close -- no fd-reuse race with the blocked - recv) the socket of any connection past it; the blocked recv returns EOF and - the thread is freed (review F2). Armed in handle(), NOT at accept/enqueue, - on purpose: a legit request queued behind busy threads (certbot runs) must - not be cut for waiting. A slow client queued behind other slow clients is - therefore bounded by (wait for a thread) + HEADER_DEADLINE. Bodies are - only read by routes behind require_api_key, so the unauthenticated surface - is exactly this phase. + client may take to deliver the request line + headers, counted from when + the connection was ENQUEUED for a thread, with a floor of + HAPROXY_MGR_HEADER_PICKUP_GRACE (1s) after a thread picks it up: + deadline = max(enqueue + 10s, pickup + 1s). Disarmed in handle_request() + (which gunicorn calls only once the parser has returned a complete + request). A 1s monitor loop shuts down (SHUT_RDWR, not close -- no fd-reuse + race with the blocked recv) the socket of any connection past it; the + blocked recv returns EOF and the thread is freed (review F2). A legit + request queued behind busy threads (certbot runs) already has its headers + in the kernel buffer and parses in milliseconds inside the 1s grace; a + slow client that sat in the queue gets only the grace, so reconnecting + drippers cannot starve the pool by queueing (re-review). Bodies are only + read by routes behind require_api_key, so the unauthenticated surface is + exactly this phase. 4. Wedge watchdog (HAPROXY_MGR_WEDGE_SECONDS, default 3600s; 0 disables). If EVERY pool thread has been inside the same request for longer than that the @@ -76,7 +79,20 @@ What this worker changes certbot (up to 180s) per domain; with (1)-(3) closing the known client-side sources, this is only a backstop for blocking sources not yet found. -5. pool_status() exposes busy/total threads and the oldest in-flight age so +5. Peer filter (re-review R1). haproxy-manager shares the client-net bridge + with every tenant container, so tenant PHP could reach haproxy-manager:8000 + directly; the 127.0.0.1 publish only removes the internet path. On the + ports in HAPROXY_MGR_PEER_FILTER_PORTS (default "8000" -- NOT the :8080 + default-page server, which HAProxy feeds over loopback) accept() closes + any connection whose peer is not in HAPROXY_MGR_API_ALLOWED_PEERS (default + "127.0.0.1,::1,gateway"; IPs/CIDRs; "gateway" = this container's default + gateway, which is where host callers arrive from through docker-proxy). + Refused before a thread or connection slot is used; logged at most once a + minute as a per-peer summary. Effective only once the edge publishes + 127.0.0.1:8000 -- with a 0.0.0.0 publish a tenant can hairpin through the + host and arrive from the gateway too. + +6. pool_status() exposes busy/total threads and the oldest in-flight age so /health can report saturation. This overrides gthread internals (enqueue_req, accept, @@ -87,6 +103,7 @@ real gunicorn in the Docker build gate and fails on behavioural drift. """ import errno +import ipaddress import logging import os import selectors @@ -115,6 +132,65 @@ HEADER_DEADLINE = _env_float('HAPROXY_MGR_HEADER_DEADLINE', 10) WEDGE_SECONDS = _env_float('HAPROXY_MGR_WEDGE_SECONDS', 3600) WATCHDOG_INTERVAL = _env_float('HAPROXY_MGR_WATCHDOG_INTERVAL', 15) MONITOR_TICK = 1.0 +# Once a thread picks a queued connection up it gets at least this long to +# finish its headers, even if its enqueue-based deadline has already passed. +HEADER_PICKUP_GRACE = _env_float('HAPROXY_MGR_HEADER_PICKUP_GRACE', 1) + +# Peer filter (review R1). Listener ports it applies to (comma-separated; +# empty disables) and the peers allowed through. "gateway" means this +# container's default gateway on its bridge network -- the address host-side +# callers arrive from via the 127.0.0.1:8000 publish (docker-proxy) and via a +# direct host->container-IP connect. Tenant containers on the same bridge +# (client-net) arrive from their OWN addresses and are refused. +PEER_FILTER_PORTS = os.environ.get('HAPROXY_MGR_PEER_FILTER_PORTS', '8000') +ALLOWED_PEERS_SPEC = os.environ.get('HAPROXY_MGR_API_ALLOWED_PEERS', '127.0.0.1,::1,gateway') +PEER_REJECT_LOG_INTERVAL = 60.0 + + +def _default_gateways(route_file='/proc/net/route'): + """IPv4 default gateway(s) from the kernel routing table.""" + gws = [] + try: + with open(route_file) as fh: + next(fh, None) + for line in fh: + f = line.split() + if len(f) >= 4 and f[1] == '00000000' and int(f[3], 16) & 0x2: + gws.append(str(ipaddress.IPv4Address(int(f[2], 16).to_bytes(4, 'little')))) + except (OSError, ValueError): + pass + return gws + + +def parse_allowed_peers(spec, gateways=None): + """Return (networks, problems). Tokens: IPs, CIDRs, or the word + 'gateway'. Unparseable tokens are reported, never silently widened.""" + nets, problems = [], [] + for tok in (t.strip() for t in (spec or '').split(',')): + if not tok: + continue + if tok.lower() == 'gateway': + gws = _default_gateways() if gateways is None else gateways + if not gws: + problems.append('gateway: no IPv4 default route found') + for g in gws: + nets.append(ipaddress.ip_network(g + '/32')) + continue + try: + nets.append(ipaddress.ip_network(tok, strict=False)) + except ValueError: + problems.append('unparseable entry %r' % tok) + return nets, problems + + +def peer_allowed(addr, nets): + try: + ip = ipaddress.ip_address(addr.split('%', 1)[0]) + except (ValueError, AttributeError): + return False + if ip.version == 6 and ip.ipv4_mapped is not None: + ip = ip.ipv4_mapped + return any(ip.version == n.version and ip in n for n in nets) TESTED_GUNICORN_VERSION = '23.0.0' @@ -173,6 +249,15 @@ class TimeoutThreadWorker(ThreadWorker): self._inflight = {} # thread ident -> _InFlight self._inflight_lock = threading.Lock() self._pending = deque() # accepted, not yet readable (FIFO by deadline) + self._filter_ports = set() + for tok in PEER_FILTER_PORTS.split(','): + tok = tok.strip() + if tok.isdigit(): + self._filter_ports.add(int(tok)) + self._allowed_nets = [] + self._rejected = {} # peer ip -> count since last log line + self._rejected_lock = threading.Lock() + self._last_reject_log = 0.0 # -- (1) client socket timeout ------------------------------------------ def enqueue_req(self, conn): @@ -183,6 +268,7 @@ class TimeoutThreadWorker(ThreadWorker): conn.init() if CLIENT_TIMEOUT > 0: conn.sock.settimeout(CLIENT_TIMEOUT) + conn.whp_enqueued = time.monotonic() # header deadline counts from here fs = self.tpool.submit(self.handle, conn) self._wrap_future(fs, conn) @@ -192,6 +278,14 @@ class TimeoutThreadWorker(ThreadWorker): # tracked in self._pending so murder_keepalived() can reap it. try: sock, client = listener.accept() + if self._filter_ports and self._server_port(server) in self._filter_ports \ + and not peer_allowed(client[0] if isinstance(client, tuple) else '', + self._allowed_nets): + # Refused before it costs a TConn, a connection slot, a poller + # registration or a thread. + sock.close() + self._note_rejected(client[0] if isinstance(client, tuple) else str(client)) + return conn = TConn(self.cfg, sock, client, server) conn.timeout = time.time() + FIRST_BYTE_TIMEOUT @@ -205,6 +299,32 @@ class TimeoutThreadWorker(ThreadWorker): errno.EWOULDBLOCK): raise + @staticmethod + def _server_port(server): + try: + return int(server[1]) + except (TypeError, IndexError, ValueError): + return None + + def _note_rejected(self, peer): + with self._rejected_lock: + self._rejected[peer] = self._rejected.get(peer, 0) + 1 + if time.monotonic() - self._last_reject_log >= PEER_REJECT_LOG_INTERVAL: + self._flush_rejected() + + def _flush_rejected(self): + with self._rejected_lock: + if not self._rejected: + return + items, self._rejected = self._rejected, {} + self._last_reject_log = time.monotonic() + total = sum(items.values()) + top = sorted(items.items(), key=lambda kv: -kv[1])[:5] + self.log.warning( + 'Refused %d connection(s) to the management API from %d non-allowed peer(s) ' + '(HAPROXY_MGR_API_ALLOWED_PEERS=%s): %s%s', total, len(items), ALLOWED_PEERS_SPEC, + ', '.join('%s x%d' % kv for kv in top), ' ...' if len(items) > 5 else '') + def on_client_socket_readable(self, conn, client): # ThreadWorker version only knows about self._keep; a first request # comes off self._pending instead. @@ -242,7 +362,16 @@ class TimeoutThreadWorker(ThreadWorker): def handle(self, conn): ident = threading.get_ident() now = time.monotonic() - entry = _InFlight(conn, now, now + HEADER_DEADLINE if HEADER_DEADLINE > 0 else None) + deadline = None + if HEADER_DEADLINE > 0: + # max(enqueue + D, pickup + grace): a legit request that waited in + # the queue behind busy threads already has its headers in the + # kernel buffer and parses in milliseconds; a slow client that + # waited gets only the grace, not a fresh full deadline (review: + # 40 reconnecting drippers otherwise starved legit requests). + enq = getattr(conn, 'whp_enqueued', now) + deadline = max(enq + HEADER_DEADLINE, now + HEADER_PICKUP_GRACE) + entry = _InFlight(conn, now, deadline) with self._inflight_lock: self._inflight[ident] = entry try: @@ -337,6 +466,8 @@ class TimeoutThreadWorker(ThreadWorker): try: if HEADER_DEADLINE > 0: self._sweep_header_deadlines() + if time.monotonic() - self._last_reject_log >= PEER_REJECT_LOG_INTERVAL: + self._flush_rejected() if time.monotonic() - last_wedge_check >= WATCHDOG_INTERVAL: last_wedge_check = time.monotonic() self._check_wedged() @@ -358,6 +489,14 @@ class TimeoutThreadWorker(ThreadWorker): '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__) + if self._filter_ports: + self._allowed_nets, problems = parse_allowed_peers(ALLOWED_PEERS_SPEC) + for prob in problems: + self.log.error('HAPROXY_MGR_API_ALLOWED_PEERS: %s -- that entry allows nothing ' + '(fail closed)', prob) + self.log.info('Management API peer filter on port(s) %s: allowing %s', + ','.join(str(p) for p in sorted(self._filter_ports)), + ', '.join(str(n) for n in self._allowed_nets) or 'NOTHING') threading.Thread(target=self._monitor, name='worker-monitor', daemon=True).start() super().init_process() # runs the event loop; does not return diff --git a/scripts/test-gunicorn-socket-timeouts.py b/scripts/test-gunicorn-socket-timeouts.py index 243eebf..1be1126 100644 --- a/scripts/test-gunicorn-socket-timeouts.py +++ b/scripts/test-gunicorn-socket-timeouts.py @@ -93,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, workers=1): + def __init__(self, worker_class, threads=1, env=None, workers=1, extra_ports=()): self.tmp = tempfile.mkdtemp(prefix='gunicorn-timeout-test-') with open(os.path.join(self.tmp, 'tapp.py'), 'w') as fh: fh.write(WSGI_APP) @@ -105,6 +105,7 @@ class Gunicorn: self.proc = subprocess.Popen( [sys.executable, '-m', 'gunicorn', '--bind', '127.0.0.1:%d' % self.port, + *sum((['--bind', '127.0.0.1:%d' % p] for p in extra_ports), []), '--workers', str(workers), '--threads', str(threads), '--worker-class', worker_class, '--timeout', '120', '--graceful-timeout', '2', @@ -416,6 +417,149 @@ class ProductionShapeExhaustion(unittest.TestCase): self.assertIn(b'200 OK', s.recv(4096)) +def refused(port, wait=2.0): + """True if the server closes a fresh connection without answering a + complete request (peer filter).""" + s = socket.create_connection(('127.0.0.1', port), timeout=wait) + try: + try: + s.sendall(b'GET / HTTP/1.1\r\nHost: x\r\n\r\n') + except OSError: + return True + s.settimeout(wait) + try: + return s.recv(4096) == b'' + except (ConnectionResetError, BrokenPipeError): + return True + except socket.timeout: + return False + finally: + s.close() + + +class PeerFilter(unittest.TestCase): + """Re-review R1: tenant containers share client-net with haproxy-manager, + so the API port must refuse peers that are not loopback / the host path.""" + + @classmethod + def setUpClass(cls): + sys.path.insert(0, REPO_DIR) + import haproxy_gunicorn_worker as w + cls.w = w + + def test_allowlist_parsing_and_matching(self): + nets, problems = self.w.parse_allowed_peers('127.0.0.1,::1,gateway', gateways=['172.18.0.1']) + self.assertEqual(problems, []) + ok = self.w.peer_allowed + self.assertTrue(ok('127.0.0.1', nets)) + self.assertTrue(ok('::1', nets)) + self.assertTrue(ok('::ffff:127.0.0.1', nets)) + self.assertTrue(ok('172.18.0.1', nets)) # docker-proxy / host + self.assertFalse(ok('172.18.0.7', nets)) # a tenant container + self.assertFalse(ok('127.0.0.2', nets)) + self.assertFalse(ok('', nets)) + self.assertFalse(ok('garbage', nets)) + + def test_bad_entries_fail_closed_and_are_reported(self): + nets, problems = self.w.parse_allowed_peers('nonsense,gateway', gateways=[]) + self.assertEqual(nets, []) + self.assertEqual(len(problems), 2, problems) + nets, _ = self.w.parse_allowed_peers('10.0.0.0/8') + self.assertTrue(self.w.peer_allowed('10.1.2.3', nets)) + + def test_default_gateway_parsed_from_route_table(self): + fn = os.path.join(tempfile.mkdtemp(), 'route') + with open(fn, 'w') as fh: + fh.write('Iface\tDestination\tGateway \tFlags\tRefCnt\tUse\tMetric\tMask\n' + 'eth0\t00000000\t010012AC\t0003\t0\t0\t0\t00000000\n' + 'eth0\t000012AC\t00000000\t0001\t0\t0\t0\t0000FFFF\n') + self.assertEqual(self.w._default_gateways(fn), ['172.18.0.1']) + + def test_filtered_port_refuses_unlisted_peer_other_port_unaffected(self): + other = free_port() + g = Gunicorn(FIXED, threads=1, extra_ports=(other,), env={ + 'HAPROXY_MGR_PEER_FILTER_PORTS': str(other), + 'HAPROXY_MGR_API_ALLOWED_PEERS': '10.255.255.1'}) + self.addCleanup(g.stop) + # g.port (unfiltered) works; `other` refuses 127.0.0.1 (not listed). + self.assertTrue(http_get(g.port, '/').startswith('pid=')) + for _ in range(5): + self.assertTrue(refused(other), 'filtered port answered an unlisted peer') + self.assertTrue(http_get(g.port, '/').startswith('pid=')) + out = g.output() + self.assertIn('Refused 1 connection(s) to the management API', out) # first one, immediately + self.assertEqual(out.count('Refused '), 1, 'reject logging is not rate-limited') + + def test_filtered_port_serves_listed_peer(self): + port = free_port() + g = Gunicorn(FIXED, threads=1, extra_ports=(port,), env={ + 'HAPROXY_MGR_PEER_FILTER_PORTS': str(port), + 'HAPROXY_MGR_API_ALLOWED_PEERS': '127.0.0.1,::1'}) + self.addCleanup(g.stop) + self.assertTrue(http_get(port, '/').startswith('pid=')) + self.assertNotIn('Refused', g.output()) + + +class QueuedDripStarvation(unittest.TestCase): + """Re-review: with the deadline armed at thread pickup, 40 reconnecting + drippers queued behind each other starved legit requests (every probe + >30s). Armed as max(enqueue + D, pickup + 1s) they must not.""" + + def test_40_reconnecting_drippers_do_not_starve_legit_requests(self): + g = Gunicorn(FIXED, threads=4, workers=2, env={}) # production defaults + self.addCleanup(g.stop) + stop = threading.Event() + self.addCleanup(stop.set) + + def drip(): + while not stop.is_set(): + try: + s = socket.create_connection(('127.0.0.1', g.port)) + except OSError: + time.sleep(0.2) + continue + try: + req = b'GET / HTTP/1.1\r\nX-a: ' + b'a' * 5000 + for i in range(len(req)): + if stop.is_set(): + break + s.send(req[i:i + 1]) + time.sleep(1.0) + except OSError: + pass + finally: + s.close() + + for _ in range(40): + threading.Thread(target=drip, daemon=True).start() + time.sleep(2) + lat = [] + t_end = time.time() + 25 + while time.time() < t_end: + t0 = time.time() + try: + http_get(g.port, '/', timeout=30) + lat.append(time.time() - t0) + except Exception: + lat.append(99.0) + time.sleep(1) + stop.set() + lat.sort() + median = lat[len(lat) // 2] + sys.stderr.write('\n 40 drippers: probes=%d median=%.2fs worst=%.2fs\n' + % (len(lat), median, lat[-1])) + # Armed at pickup, every probe here exceeded 30s (total starvation). + # With max(enqueue + D, pickup + 1s) queued drippers get ~1s of thread + # time, so legit requests keep getting through. They still queue: the + # equilibrium wait is ~(drippers/threads) x hold time, about 9s at 40 + # drippers on 8 threads -- measured, see the round-2 report. The + # remaining fix for that is not admitting partial requests to a thread + # at all; the peer filter makes it a local-only scenario. + self.assertNotIn(99.0, lat, 'a legit request was starved for 30s+') + self.assertLess(median, 12.0, lat) + self.assertLess(lat[-1], 20.0, lat) + + 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."""