test/harness/proxy_control_test.py

1#!/usr/bin/env python3

2"""Prove the SYN-drop positive control can go RED.

4`proxy.py --self-test` shows the control passing. A control that has

5only ever been seen passing is indistinguishable from a control that

6always passes, so this script breaks the SYN-drop mode three different

7ways and asserts the control catches each one.

9Run: python3 test/harness/proxy_control_test.py

11The sabotages are applied by monkeypatching a live Proxy instance --

12`proxy.py` itself is never edited, so there is no restore step that

13could silently discard work.

15Needs no enclave-server and no apiary binary: it scores the proxy's

16listening socket only.

17"""

18import os

19import time

20import sys

22sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

23from proxy import Proxy, ProxyControlFailed # noqa: E402

26def 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()

55def 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()

63def sab_noop(p):

64 """Degradation: blackhole_new() silently did nothing at all."""

67def 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()

75def 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

87if __name__ == "__main__":

88 sys.exit(main())