AtlatestRepositoryapiary

apiary / tree / test / harnessproxy.py

1"""Freeze-proxy: construct network failures that a plain socket close
2cannot imitate.
3
4Two 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.
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.
18`blackhole_new()` is the subtle one and it carries a POSITIVE CONTROL
19you must not remove -- see `assert_blackholing()` below.
20"""
21import select
22import socket
23import threading
24import time
27class ProxyControlFailed(AssertionError):
28 """The proxy is not producing the failure it claims to produce.
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 """
36 def __init__(self, message, observed=None):
37 super().__init__(message)
38 self.observed = observed
41class Proxy:
42 """TCP relay with a switchable mode."""
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()
73 # ---- failure modes -------------------------------------------
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
82 def sever_existing(self):
83 """Close every established connection cleanly (FIN both ways).
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
102 def blackhole_new(self):
103 """Make NEW connections hang instead of being refused.
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
119 # ---- controls ------------------------------------------------
121 def probe_connect(self, timeout=2.0):
122 """Classify what a fresh connection to the proxy port does.
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
144 def assert_accepting(self, timeout=2.0):
145 """NEGATIVE CONTROL -- prove the probe can see a HEALTHY port.
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
161 def assert_blackholing(self, timeout=2.0):
162 """POSITIVE CONTROL for `blackhole_new()` -- DO NOT REMOVE.
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.
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.
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)
195 # ---- plumbing ------------------------------------------------
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()
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
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
259def _self_test():
260 """Demonstrate BOTH controls firing, in both directions.
262 Run: python3 proxy.py
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")
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")
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()
298if __name__ == "__main__":
299 _self_test()