Commitc56e86ddRecorded5 Aug 2026Repositoryapiary
Port the transport failure harness into test/harness
Message
The freeze-proxy and MCP-stdio driver that measured the two worker-transport blockers lived outside any repo and would have died with the session that built them. They are the only instrument in reach that can construct a blackholed socket -- a route that stops carrying bytes without ever sending a FIN or an RST -- which is the condition both blockers are about.
The SYN-drop mode is the subtle part. It stops new connections by filling the accept queue and never accepting, so the kernel drops further SYNs and a connect hangs for its full retransmit budget (~135s). If the queue does not actually fill, the listening socket REFUSES instead and the connect fails in ~5s. Those are different failures, not a strong and a weak version of one, and an experiment that believes it is measuring the first while measuring the second still produces numbers that still look like results.
So the control that was previously run by hand before first use is now built in and runs every time:
assert_accepting() negative control -- the probe can see a HEALTHY
port, so its verdict on an unhealthy one means
something
assert_blackholing() positive control -- new connections HANG; raises
loudly and specifically if they are refusedproxy_control_test.py breaks the mode three ways (listener closed -> refused, blackhole_new() a no-op, backlog too large so the kernel completes the handshake itself) and asserts the control goes red for each. A control only ever seen passing is indistinguishable from one that always passes.
Two changes to the original while porting:
- Frozen connections are keyed by a monotonic counter rather than by Python id(). id()s are reused after GC, which was raised as a candidate explanation for a real measurement and had to be ruled out by hand; a counter cannot collide, so that hypothesis cannot arise again.
- Both controls score their OWN probe rather than probing separately and comparing. Two connects can land on different port states -- that is a race, not a check, and it produced one flaky result before it was fixed.
README records the teardown hazard: stop these servers by the exact PID in state.env, never by pattern. Concurrent harnesses on one host are a normal condition, and a pattern kill takes somebody else's server along with the killing session itself.
Changed
.gitignore | 1 +
test/harness/env-down.sh | 17 +++++++++
test/harness/env-up.sh | 78 +++++++++++++++++++++++++++++++++++++++++
test/harness/mcpdrv.py | 268 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
test/harness/proxy.py | 299 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
test/harness/proxy_control_test.py | 88 ++++++++++++++++++++++++++++++++++++++++++++++
6 files changed, 751 insertions(+)Diff
.gitignoremodified
@@ -2,3 +2,4 @@ build/
2
.sigil/deps/ 3
*.sgb 4
*.core+5
__pycache__/test/harness/env-down.shadded
@@ -0,0 +1,17 @@
+1
#!/usr/bin/env bash+2
#+3
# Tear down the throwaway enclave-server started by env-up.sh.+4
#+5
# Kills ONLY the PID recorded in state.env. Never a pattern match:+6
# another harness may be running the same binary on this host, and an+7
# agent session's own command line typically contains the task name it+8
# would match on. See README.md.+9
set -u+10
RUN=${HARNESS_RUN:-/tmp/apiary-harness}+11
[ -f "$RUN/state.env" ] || { echo "no $RUN/state.env -- nothing to stop"; exit 0; }+12
. "$RUN/state.env"+13
if kill -0 "$SERVER_PID" 2>/dev/null; then+14
kill "$SERVER_PID" && echo "killed enclave server pid=$SERVER_PID"+15
else+16
echo "server pid=$SERVER_PID already gone"+17
fitest/harness/env-up.shadded
@@ -0,0 +1,78 @@
+1
#!/usr/bin/env bash+2
#+3
# Stand up a throwaway enclave-server + seeded DB for the apiary+4
# transport harness. Writes state to $HARNESS_RUN/state.env so the+5
# python drivers can read it.+6
#+7
# ./env-up.sh -> seeds, starts server, prints the run dir+8
# ./env-down.sh -> kills the server by RECORDED PID, never pkill+9
#+10
# NEVER stop this server with a pattern match. See README.md --+11
# concurrent enclave harnesses on one host are normal, and a pattern+12
# kill takes somebody else's server AND the killing session with it.+13
set -u+14
+15
RUN=${HARNESS_RUN:-/tmp/apiary-harness}+16
PORT=${HARNESS_PORT:-49771}+17
ENCLAVE_BIN=${ENCLAVE_BIN:-}+18
DB="$RUN/harness.db"+19
+20
OWNER=${HARNESS_OWNER:-owner-1}+21
OWNER_PASS=${HARNESS_OWNER_PASS:-harness-admin}+22
LEADER=${HARNESS_LEADER:-leader-1}+23
CHANNEL=${HARNESS_CHANNEL:-#coord}+24
GROUP=${HARNESS_GROUP:-test-workers}+25
+26
if [ -z "$ENCLAVE_BIN" ]; then+27
echo "FATAL: set ENCLAVE_BIN to an enclave-server binary." >&2+28
echo " e.g. ENCLAVE_BIN=/path/to/enclave ./env-up.sh" >&2+29
exit 1+30
fi+31
if [ ! -x "$ENCLAVE_BIN" ]; then+32
echo "FATAL: enclave binary not found/executable at $ENCLAVE_BIN" >&2+33
exit 1+34
fi+35
+36
mkdir -p "$RUN"+37
rm -f "$DB" "$DB-shm" "$DB-wal"+38
+39
ENCLAVE_DB_PATH="$DB" "$ENCLAVE_BIN" user add "$OWNER" \+40
--password "$OWNER_PASS" --role admin >/dev/null \+41
|| { echo "FATAL: user add failed" >&2; exit 1; }+42
ENCLAVE_DB_PATH="$DB" "$ENCLAVE_BIN" group create "$GROUP" >/dev/null 2>&1 || true+43
+44
LEADER_OUT=$(ENCLAVE_DB_PATH="$DB" "$ENCLAVE_BIN" \+45
bot register "$LEADER" --owner "$OWNER" --allow-services yes \+46
--auto-join "$CHANNEL")+47
LEADER_TOKEN=$(echo "$LEADER_OUT" | grep -oE '[a-f0-9]{64}' | head -1)+48
if [ -z "$LEADER_TOKEN" ]; then+49
echo "FATAL: no leader token in bot-register output" >&2+50
echo "$LEADER_OUT" >&2+51
exit 1+52
fi+53
+54
ENCLAVE_LOG_LEVEL=${ENCLAVE_LOG_LEVEL:-info} ENCLAVE_PORT="$PORT" \+55
ENCLAVE_DB_PATH="$DB" "$ENCLAVE_BIN" serve > "$RUN/server.log" 2>&1 &+56
SERVER=$!+57
sleep 2+58
if ! kill -0 "$SERVER" 2>/dev/null; then+59
echo "FATAL: server died on startup" >&2+60
cat "$RUN/server.log" >&2+61
exit 1+62
fi+63
+64
cat > "$RUN/state.env" <<EOF+65
RUN=$RUN+66
PORT=$PORT+67
DB=$DB+68
SERVER_PID=$SERVER+69
ENCLAVE_BIN=$ENCLAVE_BIN+70
LEADER_TOKEN=$LEADER_TOKEN+71
LEADER_NICK=$LEADER+72
OWNER=$OWNER+73
CHANNEL=$CHANNEL+74
EOF+75
+76
echo "RUN=$RUN PORT=$PORT SERVER_PID=$SERVER"+77
echo "leader token: ${LEADER_TOKEN:0:8}...(${#LEADER_TOKEN} chars)"+78
echo "stop with: HARNESS_RUN=$RUN $(dirname "$0")/env-down.sh"test/harness/mcpdrv.pyadded
@@ -0,0 +1,268 @@
+1
#!/usr/bin/env python3+2
"""Drive apiary (an MCP-stdio server) as a scriptable client.+3
+4
Apiary has no CLI; running the binary speaks MCP JSON-RPC on+5
stdin/stdout. This module spawns one apiary process per bot, performs+6
the MCP handshake, and lets a test script call tools and observe+7
inbound `notifications/claude/channel` events -- which is exactly the+8
signal that says "this bot's trusted-set filter delivered that+9
message".+10
+11
Everything is line-delimited JSON on stdout; stderr carries apiary's+12
own structured log.+13
+14
Configuration comes from the environment so a single experiment script+15
can be pointed at two different binaries for a matched pair:+16
+17
APIARY_BIN path to the apiary release bundle to drive (required)+18
HARNESS_RUN state/log directory (default /tmp/apiary-harness)+19
"""+20
+21
import json+22
import os+23
import queue+24
import subprocess+25
import sys+26
import threading+27
import time+28
+29
RUN_DIR = os.environ.get("HARNESS_RUN", "/tmp/apiary-harness")+30
+31
# Identifiers used by env-up.sh's seeded database. Generic on purpose:+32
# this is a public repo and these must not encode any particular+33
# deployment's nicks, hosts or channels.+34
OWNER = os.environ.get("HARNESS_OWNER", "owner-1")+35
LEADER_NICK = os.environ.get("HARNESS_LEADER", "leader-1")+36
CHANNEL = os.environ.get("HARNESS_CHANNEL", "#coord")+37
WORKER_GROUP = os.environ.get("HARNESS_GROUP", "test-workers")+38
+39
+40
def apiary_bin():+41
b = os.environ.get("APIARY_BIN")+42
if not b:+43
raise SystemExit(+44
"APIARY_BIN is not set. Point it at the apiary release "+45
"bundle to drive, e.g.\n"+46
" APIARY_BIN=$PWD/build/release/bin/apiary python3 "+47
"test/harness/exp_silent_loss.py\n"+48
"Refusing to guess: a matched pair is worthless if both "+49
"arms silently run the same binary.")+50
if not os.access(b, os.X_OK):+51
raise SystemExit(f"APIARY_BIN is not executable: {b}")+52
return b+53
+54
+55
def load_state(run_dir=None):+56
"""Read env-up.sh's state.env."""+57
run_dir = run_dir or RUN_DIR+58
path = os.path.join(run_dir, "state.env")+59
if not os.path.exists(path):+60
raise SystemExit(+61
f"no {path} -- run test/harness/env-up.sh first")+62
state = {}+63
with open(path) as f:+64
for line in f:+65
k, _, v = line.strip().partition("=")+66
if k:+67
state[k] = v+68
return state+69
+70
+71
class Bot:+72
def __init__(self, label, env, bin=None, log_dir=None):+73
self.label = label+74
self.env = dict(os.environ)+75
# Scrub any inherited APIARY_* so a stray value from the shell+76
# cannot silently change what we think we are testing.+77
for k in list(self.env):+78
if k.startswith("APIARY_"):+79
del self.env[k]+80
self.env.update({k: str(v) for k, v in env.items()})+81
log_dir = log_dir or RUN_DIR+82
os.makedirs(log_dir, exist_ok=True)+83
self.stderr_path = os.path.join(log_dir, f"{label}.stderr")+84
self._stderr_f = open(self.stderr_path, "w")+85
self.bin = bin or apiary_bin()+86
self.proc = subprocess.Popen(+87
[self.bin],+88
stdin=subprocess.PIPE,+89
stdout=subprocess.PIPE,+90
stderr=self._stderr_f,+91
env=self.env,+92
text=True,+93
bufsize=1,+94
)+95
self._id = 0+96
self._responses = {}+97
self._resp_lock = threading.Condition()+98
self.notifications = []+99
self._notif_lock = threading.Condition()+100
self.raw = []+101
self._reader = threading.Thread(target=self._read_loop, daemon=True)+102
self._reader.start()+103
+104
# ---- plumbing -------------------------------------------------+105
+106
def _read_loop(self):+107
for line in self.proc.stdout:+108
line = line.strip()+109
if not line:+110
continue+111
self.raw.append(line)+112
try:+113
msg = json.loads(line)+114
except json.JSONDecodeError:+115
continue+116
if "id" in msg and ("result" in msg or "error" in msg):+117
with self._resp_lock:+118
self._responses[msg["id"]] = msg+119
self._resp_lock.notify_all()+120
elif msg.get("method") == "notifications/claude/channel":+121
with self._notif_lock:+122
self.notifications.append(msg["params"])+123
self._notif_lock.notify_all()+124
+125
def _send(self, obj):+126
self.proc.stdin.write(json.dumps(obj) + "\n")+127
self.proc.stdin.flush()+128
+129
def _request(self, method, params=None, timeout=60):+130
self._id += 1+131
rid = self._id+132
self._send({"jsonrpc": "2.0", "id": rid, "method": method,+133
"params": params or {}})+134
deadline = time.time() + timeout+135
with self._resp_lock:+136
while rid not in self._responses:+137
remaining = deadline - time.time()+138
if remaining <= 0:+139
raise TimeoutError(f"{self.label}: {method} timed out")+140
self._resp_lock.wait(remaining)+141
return self._responses.pop(rid)+142
+143
# ---- MCP ------------------------------------------------------+144
+145
def initialize(self, timeout=60):+146
r = self._request("initialize", {+147
"protocolVersion": "2024-11-05",+148
"capabilities": {},+149
"clientInfo": {"name": "apiary-harness", "version": "0.0"},+150
}, timeout=timeout)+151
self._send({"jsonrpc": "2.0", "method": "notifications/initialized",+152
"params": {}})+153
return r+154
+155
def tools(self):+156
r = self._request("tools/list")+157
return sorted(t["name"] for t in r["result"]["tools"])+158
+159
def call(self, tool, timeout=60, **args):+160
r = self._request("tools/call", {"name": tool, "arguments": args},+161
timeout=timeout)+162
if "error" in r:+163
return f"<JSONRPC-ERROR {json.dumps(r['error'])}>"+164
parts = [c.get("text", "") for c in r["result"].get("content", [])]+165
return "\n".join(parts)+166
+167
def wait_connected(self, timeout=25.0):+168
"""Block until the bridge logs a successful connect."""+169
deadline = time.time() + timeout+170
while time.time() < deadline:+171
if self.grep_stderr("Apiary bridge connected"):+172
return True+173
time.sleep(0.1)+174
return False+175
+176
# ---- observation ---------------------------------------------+177
+178
def wait_notification(self, match, timeout=20):+179
"""Wait for an inbound channel notification whose content+180
contains `match`. Returns the params dict, or None on timeout."""+181
deadline = time.time() + timeout+182
seen = 0+183
with self._notif_lock:+184
while True:+185
while seen < len(self.notifications):+186
n = self.notifications[seen]+187
seen += 1+188
if match in n.get("content", ""):+189
return n+190
remaining = deadline - time.time()+191
if remaining <= 0:+192
return None+193
self._notif_lock.wait(remaining)+194
+195
def contents(self):+196
return [n.get("content", "") for n in self.notifications]+197
+198
def alive(self):+199
return self.proc.poll() is None+200
+201
def stderr_text(self):+202
self._stderr_f.flush()+203
with open(self.stderr_path) as f:+204
return f.read()+205
+206
def grep_stderr(self, pat):+207
return [l for l in self.stderr_text().splitlines() if pat in l]+208
+209
def close(self):+210
try:+211
self.proc.terminate()+212
self.proc.wait(timeout=10)+213
except Exception:+214
try:+215
self.proc.kill()+216
except Exception:+217
pass+218
try:+219
self._stderr_f.close()+220
except Exception:+221
pass+222
+223
+224
def leader_env(port, token, nick=None, host="127.0.0.1",+225
channel=None, owner=None, **extra):+226
e = {+227
"APIARY_ENCLAVE_HOST": host,+228
"APIARY_ENCLAVE_PORT": port,+229
"APIARY_ENCLAVE_TLS": "no",+230
"APIARY_USER": nick or LEADER_NICK,+231
"APIARY_TOKEN": token,+232
"APIARY_CHANNEL": channel or CHANNEL,+233
"APIARY_MODE": "leader",+234
"APIARY_OWNER_NICK": owner or OWNER,+235
"APIARY_REPORTS_TO": owner or OWNER,+236
"APIARY_SERVICES_TIMEOUT": "8.0",+237
"APIARY_WORKER_GROUP": WORKER_GROUP,+238
}+239
e.update(extra)+240
return e+241
+242
+243
def worker_env(port, token, nick, reports_to=None,+244
host="127.0.0.1", channel=None, owner=None, **extra):+245
e = {+246
"APIARY_ENCLAVE_HOST": host,+247
"APIARY_ENCLAVE_PORT": port,+248
"APIARY_ENCLAVE_TLS": "no",+249
"APIARY_USER": nick,+250
"APIARY_TOKEN": token,+251
"APIARY_CHANNEL": channel or CHANNEL,+252
"APIARY_MODE": "worker",+253
"APIARY_OWNER_NICK": owner or OWNER,+254
"APIARY_REPORTS_TO": reports_to or LEADER_NICK,+255
"APIARY_SERVICES_TIMEOUT": "8.0",+256
}+257
e.update(extra)+258
return e+259
+260
+261
def say(*a):+262
print(*a, flush=True)+263
+264
+265
def check(label, ok, detail=""):+266
say(f" [{'PASS' if ok else 'FAIL'}] {label}"+267
+ (f" -- {detail}" if detail else ""))+268
return bool(ok)test/harness/proxy.pyadded
@@ -0,0 +1,299 @@
+1
"""Freeze-proxy: construct network failures that a plain socket close+2
cannot imitate.+3
+4
Two modes, and they are DIFFERENT failures:+5
+6
freeze_existing() -- hold established connections open and silently+7
swallow bytes in both directions. No FIN, no+8
RST. This is what a route change looks like to+9
the far end: the peer's kernel keeps the socket+10
ESTABLISHED forever and writes succeed into the+11
local send buffer.+12
+13
blackhole_new() -- make NEW connections HANG instead of being+14
refused, by filling the accept queue and never+15
accepting again so the kernel drops further+16
SYNs.+17
+18
`blackhole_new()` is the subtle one and it carries a POSITIVE CONTROL+19
you must not remove -- see `assert_blackholing()` below.+20
"""+21
import select+22
import socket+23
import threading+24
import time+25
+26
+27
class ProxyControlFailed(AssertionError):+28
"""The proxy is not producing the failure it claims to produce.+29
+30
`observed` carries the probe classification that failed the+31
control, so a caller scores the SAME probe the control scored.+32
Probing twice and comparing the two answers is a race, not a+33
check -- the port's state can differ between them.+34
"""+35
+36
def __init__(self, message, observed=None):+37
super().__init__(message)+38
self.observed = observed+39
+40
+41
class Proxy:+42
"""TCP relay with a switchable mode."""+43
+44
def __init__(self, listen_port, target_port, host="127.0.0.1"):+45
self.mode = "pass"+46
self.host = host+47
self.listen_port = listen_port+48
self.target_port = target_port+49
self.conns = [] # list of [key, client_sock, upstream_sock]+50
self.fillers = []+51
self._next_key = 0+52
self.srv = socket.socket()+53
self.srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)+54
self.srv.bind((host, listen_port))+55
# Tiny backlog on purpose: `blackhole_new()` fills the accept+56
# queue and stops accepting, so further SYNs are DROPPED by the+57
# kernel rather than refused. Refused fails fast (~5s) and would+58
# test the wrong failure entirely; a route that has gone away+59
# drops, and the caller waits out the kernel's SYN-retransmit+60
# budget (~135s).+61
self.srv.listen(1)+62
# Frozen connections, keyed by our own monotonic counter.+63
#+64
# The spike version keyed this on `id(pair)`. CPython reuses+65
# id()s after GC, so a connection created after a reconnect+66
# could in principle inherit a stale frozen mark -- which was+67
# raised as a candidate explanation (H1) for a real measurement+68
# and had to be ruled out by hand. A counter cannot collide, so+69
# the hypothesis cannot arise again.+70
self.frozen_at = {}+71
threading.Thread(target=self._accept_loop, daemon=True).start()+72
+73
# ---- failure modes -------------------------------------------+74
+75
def freeze_existing(self):+76
"""Blackhole every currently-established connection."""+77
now = time.time()+78
for entry in list(self.conns):+79
self.frozen_at[entry[0]] = now+80
return now+81
+82
def sever_existing(self):+83
"""Close every established connection cleanly (FIN both ways).+84
+85
This is the SERVER-RESTART shape, not the route-change shape:+86
the peer sees EOF immediately and reconnects in seconds. Use it+87
when you want to exercise the reconnect path itself rather than+88
the detection that precedes it -- e.g. to inspect what state a+89
reconnected connection comes back with.+90
"""+91
n = 0+92
for entry in list(self.conns):+93
for s in entry[1:]:+94
try:+95
s.close()+96
except OSError:+97
pass+98
n += 1+99
self.conns = []+100
return n+101
+102
def blackhole_new(self):+103
"""Make NEW connections hang instead of being refused.+104
+105
Call `assert_blackholing()` afterwards. Do not assume this+106
worked.+107
"""+108
self.mode = "blackhole"+109
time.sleep(0.3) # let the accept loop park+110
for _ in range(4):+111
c = socket.socket()+112
c.settimeout(1.5)+113
try:+114
c.connect((self.host, self.listen_port))+115
self.fillers.append(c)+116
except OSError:+117
pass # queue already full: good+118
+119
# ---- controls ------------------------------------------------+120
+121
def probe_connect(self, timeout=2.0):+122
"""Classify what a fresh connection to the proxy port does.+123
+124
Returns one of "connected" / "hung" / "refused" / "error:...".+125
This is the instrument the two controls below score.+126
"""+127
s = socket.socket()+128
s.settimeout(timeout)+129
try:+130
s.connect((self.host, self.listen_port))+131
return "connected"+132
except socket.timeout:+133
return "hung"+134
except ConnectionRefusedError:+135
return "refused"+136
except OSError as e:+137
return f"error:{e.__class__.__name__}"+138
finally:+139
try:+140
s.close()+141
except OSError:+142
pass+143
+144
def assert_accepting(self, timeout=2.0):+145
"""NEGATIVE CONTROL -- prove the probe can see a HEALTHY port.+146
+147
Without this, `assert_blackholing()` is unfalsifiable: a probe+148
that reports "hung" unconditionally (wrong host, wrong port,+149
a bug in the timeout handling) would pass it for free. Run this+150
BEFORE blackholing, against the same port, with the same probe.+151
"""+152
got = self.probe_connect(timeout=timeout)+153
if got != "connected":+154
raise ProxyControlFailed(+155
"negative control FAILED: a fresh connection to the "+156
f"proxy port before blackholing reported {got!r}, not "+157
"'connected'. The probe cannot see a healthy port, so "+158
"its verdict on an unhealthy one means nothing.", got)+159
return got+160
+161
def assert_blackholing(self, timeout=2.0):+162
"""POSITIVE CONTROL for `blackhole_new()` -- DO NOT REMOVE.+163
+164
The mode claims new connections are DROPPED (SYNs go+165
unanswered, the caller hangs for the kernel's full retransmit+166
budget, ~135s). If the accept queue did not actually fill --+167
a larger backlog, a kernel that answers anyway, a filler+168
connect that raced -- the listening socket instead REFUSES,+169
which fails in ~5s.+170
+171
Refused and dropped are DIFFERENT FAILURES, not a strong and a+172
weak version of one. An experiment that believes it is+173
measuring a blackholed route while measuring a refused one is+174
wrong in a way nothing downstream can detect: it still produces+175
numbers, and the numbers still look like results.+176
+177
This was verified by hand before the mode was first used. That+178
verification is now here so it runs every time.+179
"""+180
got = self.probe_connect(timeout=timeout)+181
if got == "hung":+182
return got+183
if got == "refused":+184
raise ProxyControlFailed(+185
"SYN-drop positive control FAILED: a fresh connection "+186
"was REFUSED, not dropped. The accept queue did not "+187
"fill, so this proxy is producing a ~5s "+188
"connection-refused, NOT a ~135s blackhole. Any "+189
"measurement taken now is of a different failure.", got)+190
raise ProxyControlFailed(+191
"SYN-drop positive control FAILED: a fresh connection "+192
f"reported {got!r}; expected 'hung'. New connections are "+193
"not being dropped.", got)+194
+195
# ---- plumbing ------------------------------------------------+196
+197
def _accept_loop(self):+198
while True:+199
if self.mode == "blackhole":+200
time.sleep(0.2) # never accept: SYNs get dropped+201
continue+202
if self.mode == "closed":+203
return+204
r, _, _ = select.select([self.srv], [], [], 0.2)+205
if not r:+206
continue+207
try:+208
c, _ = self.srv.accept()+209
except OSError:+210
return+211
try:+212
u = socket.create_connection((self.host, self.target_port))+213
except OSError:+214
c.close()+215
continue+216
self._next_key += 1+217
key = self._next_key+218
self.conns.append([key, c, u])+219
threading.Thread(target=self._pump, args=(c, u, key),+220
daemon=True).start()+221
threading.Thread(target=self._pump, args=(u, c, key),+222
daemon=True).start()+223
+224
def _pump(self, a, b, key):+225
a.settimeout(0.5)+226
while True:+227
try:+228
data = a.recv(65536)+229
except socket.timeout:+230
continue+231
except OSError:+232
return+233
if not data:+234
try:+235
b.shutdown(socket.SHUT_WR)+236
except OSError:+237
pass+238
return+239
if self.frozen_at.get(key):+240
continue # swallow: no FIN, no RST, bytes vanish+241
try:+242
b.sendall(data)+243
except OSError:+244
return+245
+246
def close(self):+247
self.mode = "closed"+248
for c in self.fillers:+249
try:+250
c.close()+251
except OSError:+252
pass+253
try:+254
self.srv.close()+255
except OSError:+256
pass+257
+258
+259
def _self_test():+260
"""Demonstrate BOTH controls firing, in both directions.+261
+262
Run: python3 proxy.py+263
+264
Prints the classification of a fresh connect before and after+265
`blackhole_new()`. A run in which both lines say the same thing is+266
a broken harness whatever the words are.+267
"""+268
# Target port is deliberately something nothing listens on: the+269
# controls score the LISTENING side, and never need an upstream.+270
p = Proxy(0, 9, host="127.0.0.1")+271
# Port 0 asked the kernel for an ephemeral port; read back what it+272
# actually bound so the probe aims at the right place.+273
p.listen_port = p.srv.getsockname()[1]+274
try:+275
# Each control probes ONCE and we score that same probe. An+276
# earlier version probed separately and compared, which is a+277
# race rather than a check: the port's state can change between+278
# two connects, and it did.+279
before = p.assert_accepting()+280
print(f" before blackhole_new(): fresh connect -> {before!r}")+281
print(" [PASS] negative control: probe sees a HEALTHY port")+282
+283
p.blackhole_new()+284
after = p.assert_blackholing()+285
print(f" after blackhole_new(): fresh connect -> {after!r}")+286
print(" [PASS] positive control: new connections are DROPPED, "+287
"not refused")+288
+289
if before == after:+290
raise ProxyControlFailed(+291
f"both probes reported {before!r} -- the probe is not "+292
"discriminating, it is just returning a constant", before)+293
print(f"\n controls discriminate: {before!r} -> {after!r}")+294
finally:+295
p.close()+296
+297
+298
if __name__ == "__main__":+299
_self_test()test/harness/proxy_control_test.pyadded
@@ -0,0 +1,88 @@
+1
#!/usr/bin/env python3+2
"""Prove the SYN-drop positive control can go RED.+3
+4
`proxy.py --self-test` shows the control passing. A control that has+5
only ever been seen passing is indistinguishable from a control that+6
always passes, so this script breaks the SYN-drop mode three different+7
ways and asserts the control catches each one.+8
+9
Run: python3 test/harness/proxy_control_test.py+10
+11
The sabotages are applied by monkeypatching a live Proxy instance --+12
`proxy.py` itself is never edited, so there is no restore step that+13
could silently discard work.+14
+15
Needs no enclave-server and no apiary binary: it scores the proxy's+16
listening socket only.+17
"""+18
import os+19
import time+20
import sys+21
+22
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))+23
from proxy import Proxy, ProxyControlFailed # noqa: E402+24
+25
+26
def run(name, sabotage, expect_probe):+27
p = Proxy(0, 9)+28
# Port 0 asked the kernel for an ephemeral port; read back what it+29
# actually bound so the probe aims at the right place.+30
p.listen_port = p.srv.getsockname()[1]+31
try:+32
# Fixture validity: the instrument still sees a healthy port+33
# BEFORE the sabotage. Without this a RED below could just mean+34
# the probe was broken from the start.+35
p.assert_accepting()+36
sabotage(p)+37
time.sleep(0.3) # let the accept loop park/exit before probing+38
try:+39
# Score the control's OWN probe (carried on the exception).+40
# Probing separately and comparing would be a race: the two+41
# connects can land on different port states.+42
got = p.assert_blackholing()+43
except ProxyControlFailed as e:+44
ok = (e.observed == expect_probe)+45
print(f" [{'RED' if ok else 'RED (unexpected shape)'}] {name}: "+46
f"probe={e.observed!r} (expected {expect_probe!r})")+47
print(f" -> {str(e).splitlines()[0][:96]}...")+48
return ok+49
print(f" [FALSE GREEN] {name}: probe={got!r} but the control PASSED")+50
return False+51
finally:+52
p.close()+53
+54
+55
def sab_close_listener(p):+56
"""Degradation: the listener is CLOSED rather than parked, so the+57
kernel REFUSES. Refused fails in ~5s where a blackhole takes ~135s+58
-- a different failure, and the one this control exists to catch."""+59
p.mode = "closed"+60
p.srv.close()+61
+62
+63
def sab_noop(p):+64
"""Degradation: blackhole_new() silently did nothing at all."""+65
+66
+67
def sab_big_backlog(p):+68
"""Degradation: the backlog is large enough that the filler connects+69
do not fill it, so the kernel completes the handshake by itself and+70
connect() SUCCEEDS even though userspace never accepts."""+71
p.srv.listen(128)+72
p.blackhole_new()+73
+74
+75
def main():+76
print("== sabotaging the SYN-drop positive control ==")+77
results = [+78
run("listener closed -> refused", sab_close_listener, "refused"),+79
run("blackhole_new() no-op", sab_noop, "connected"),+80
run("backlog too large", sab_big_backlog, "connected"),+81
]+82
print(f"\n {sum(results)}/{len(results)} sabotages produced the "+83
"expected RED")+84
return 0 if all(results) else 1+85
+86
+87
if __name__ == "__main__":+88
sys.exit(main())