Fix MITM using unique source ports for each server-side connection (finos#577)

* Fix MITM using unique source ports for each server-side connection

Each new client SYN now increments the server-side source port so the
server sees a distinct 4-tuple per client/reconnection, avoiding TCP
state machine corruption when clients reconnect.

The packet filter is updated to accept packets on the current
server_conn.local_port in addition to mitm_port, so server responses
to the unique port are not dropped.

Closes finos#576

* Disable Python stdout buffering in embedded interpreter

Python's stdout is block-buffered when connected to a pipe (e.g. ctest
in CI). Forked child processes exit via std::exit() without calling
Py_Finalize(), so Python's internal BufferedWriter is never flushed and
all print() output is silently dropped.

Use Py_UnbufferedStdioFlag on Python < 3.12 and PyConfig.buffered_stdio
on Python >= 3.12 (where the flag is deprecated) to make stdout
unbuffered before Py_Initialize().

* Re-enable UVYMQMitmTest/Reconnect on Windows

The test was disabled in finos#576 due to TCP state machine corruption caused
by the MITM always reusing the same source port for server-side
connections. This is fixed by the unique-port counter introduced in this
branch.

* Consolidate MITM platform IP selection into getMitmIPs() helper

Removes repeated #ifdef __linux__ / _WIN32 blocks from each MITM test
by introducing a MitmIPs struct and getMitmIPs() in testing.h/cpp.

* Assign static unique ports to MITM tests, remove randomPort()

Replaces randomPort() calls with fixed ports 23575-23578 for the four
MITM tests, matching the pattern of the existing static remote_ports
(23571-23574). Removes the now-unused randomPort() utility.

* Apply getMitmIPs() and static ports to uv_ymq MITM tests

Same cleanup as done for test_ymq.cpp: replace repeated #ifdef blocks
with getMitmIPs() and assign static unique mitm ports (23579-23582).

Author: magniloquency

tests/cpp/uv_ymq/test_mitm.cpp

@@ -136,15 +136,9 @@ TestResult reconnectClientMain(std::string address)
 // for more, see the python mitm files
 TEST_F(UVYMQMitmTest, Passthrough)
 {
-#ifdef __linux__
-    auto mitm_ip   = "192.0.2.4";
-    auto remote_ip = "192.0.2.3";
-#else
-    auto mitm_ip   = "127.0.0.1";
-    auto remote_ip = "127.0.0.1";
-#endif
-    auto mitm_port   = randomPort();
-    auto remote_port = 23571;
+    auto [mitm_ip, remote_ip] = getMitmIPs();
+    auto mitm_port            = 23579;
+    auto remote_port          = 23571;
 
     // the Python program must be the first and only the first function passed to test()
     // we must also pass `true` as the third argument to ensure that Python is fully started
@@ -162,15 +156,9 @@ TEST_F(UVYMQMitmTest, Passthrough)
 // this is the same as the above, but both the client and server use raw sockets
 TEST_F(UVYMQMitmTest, PassthroughRaw)
 {
-#ifdef __linux__
-    auto mitm_ip   = "192.0.2.4";
-    auto remote_ip = "192.0.2.3";
-#else
-    auto mitm_ip   = "127.0.0.1";
-    auto remote_ip = "127.0.0.1";
-#endif
-    auto mitm_port   = randomPort();
-    auto remote_port = 23574;
+    auto [mitm_ip, remote_ip] = getMitmIPs();
+    auto mitm_port            = 23580;
+    auto remote_port          = 23574;
 
     // the Python program must be the first and only the first function passed to test()
     // we must also pass `true` as the third argument to ensure that Python is fully started
@@ -185,22 +173,11 @@ TEST_F(UVYMQMitmTest, PassthroughRaw)
 }
 
 // this test uses the mitm to test the reconnect logic of uv_ymq by sending RST packets
-#ifdef _WIN32
-// See https://github.com/finos/opengris-scaler/issues/576
-TEST_F(UVYMQMitmTest, DISABLED_Reconnect)
-#else
 TEST_F(UVYMQMitmTest, Reconnect)
-#endif
 {
-#ifdef __linux__
-    auto mitm_ip   = "192.0.2.4";
-    auto remote_ip = "192.0.2.3";
-#else
-    auto mitm_ip   = "127.0.0.1";
-    auto remote_ip = "127.0.0.1";
-#endif
-    auto mitm_port   = randomPort();
-    auto remote_port = 23572;
+    auto [mitm_ip, remote_ip] = getMitmIPs();
+    auto mitm_port            = 23581;
+    auto remote_port          = 23572;
 
     auto result = test(
         30,
@@ -215,15 +192,9 @@ TEST_F(UVYMQMitmTest, Reconnect)
 // in this test, the mitm drops a random % of packets arriving from the client and server
 TEST_F(UVYMQMitmTest, RandomlyDropPackets)
 {
-#ifdef __linux__
-    auto mitm_ip   = "192.0.2.4";
-    auto remote_ip = "192.0.2.3";
-#else
-    auto mitm_ip   = "127.0.0.1";
-    auto remote_ip = "127.0.0.1";
-#endif
-    auto mitm_port   = randomPort();
-    auto remote_port = 23573;
+    auto [mitm_ip, remote_ip] = getMitmIPs();
+    auto mitm_port            = 23582;
+    auto remote_port          = 23573;
 
     auto result = test(
         180,

tests/cpp/ymq/common/testing.cpp

@@ -79,7 +79,25 @@ void ensurePythonInitialized()
         return;
 
     ensurePythonHome();
+
+    // Disable Python's stdout buffering so that print() output is flushed
+    // immediately when stdout is a pipe (e.g. under ctest in CI), rather than
+    // being lost when the forked child exits without calling Py_Finalize().
+#if PY_MAJOR_VERSION == 3 && PY_MINOR_VERSION >= 12
+    // Py_UnbufferedStdioFlag is deprecated in Python 3.12+; use PyConfig instead.
+    {
+        PyConfig config;
+        PyConfig_InitPythonConfig(&config);
+        config.buffered_stdio = 0;
+        PyStatus status       = Py_InitializeFromConfig(&config);
+        PyConfig_Clear(&config);
+        if (PyStatus_Exception(status))
+            throw std::runtime_error("failed to initialize Python interpreter");
+    }
+#else
+    Py_UnbufferedStdioFlag = 1;
     Py_Initialize();
+#endif
 
     // add the cwd to the path
     {
@@ -161,6 +179,15 @@ TestResult runPython(const char* path, std::vector<std::optional<std::string>> a
     return TestResult::Failure;
 }
 
+MitmIPs getMitmIPs()
+{
+#ifdef __linux__
+    return {"192.0.2.4", "192.0.2.3"};
+#else
+    return {"127.0.0.1", "127.0.0.1"};
+#endif
+}
+
 TestResult runMitm(
     std::string testCase,
     std::string mitmIp,

tests/cpp/ymq/common/testing.h

@@ -46,6 +46,14 @@ int getListenerPid();
 
 TestResult runPython(const char* path, std::vector<std::optional<std::string>> argv = {});
 
+struct MitmIPs {
+    const char* mitmIp;
+    const char* remoteIp;
+};
+
+// returns the platform-appropriate IP addresses for MITM tests
+MitmIPs getMitmIPs();
+
 TestResult runMitm(
     std::string testCase,
     std::string mitmIp,

tests/cpp/ymq/common/utils.cpp

@@ -1,7 +1,6 @@
 #include "tests/cpp/ymq/common/utils.h"
 
 #include <filesystem>
-#include <random>
 
 // change the current working directory to the project root
 // this is important for finding the python mitm script
@@ -18,10 +17,3 @@ void chdirToProjectRoot()
         }
     }
 }
-
-unsigned short randomPort(unsigned short minPort, unsigned short maxPort)
-{
-    static thread_local std::mt19937_64 rng(std::random_device {}());
-    std::uniform_int_distribution<unsigned int> dist(minPort, maxPort);
-    return static_cast<unsigned short>(dist(rng));
-}

tests/cpp/ymq/common/utils.h

@@ -12,5 +12,3 @@ void raiseSocketError(const char* msg);
 // change the current working directory to the project root
 // this is important for finding the python mitm script
 void chdirToProjectRoot();
-
-unsigned short randomPort(unsigned short minPort = 1024, unsigned short maxPort = 65535);

tests/cpp/ymq/py_mitm/main.py

@@ -54,7 +54,8 @@ def main(pid: int, mitm_ip: str, mitm_port: int, remote_ip: str, server_port: in
 
     # the port that the mitm uses to connect to the server
     # we increment the port for each new connection to avoid collisions
-    server_conn = TCPConnection(mitm_ip, mitm_port, remote_ip, server_port)
+    next_server_port = mitm_port
+    server_conn = TCPConnection(mitm_ip, next_server_port, remote_ip, server_port)
 
     # tracks the state of each connection
     client_closed = False
@@ -66,7 +67,8 @@ def main(pid: int, mitm_ip: str, mitm_port: int, remote_ip: str, server_port: in
         if not pkt.haslayer(IP) or not pkt.haslayer(TCP):
             continue
 
-        if pkt.sport != mitm_port and pkt.dport != mitm_port:
+        active_ports = {mitm_port, server_conn.local_port}
+        if pkt.sport not in active_ports and pkt.dport not in active_ports:
             continue
 
         ip = pkt[IP]
@@ -81,7 +83,8 @@ def main(pid: int, mitm_ip: str, mitm_port: int, remote_ip: str, server_port: in
                 print(f"[*] Client {ip.src}:{tcp.sport} connecting to {ip.dst}:{tcp.dport}")
                 client_conn = sender
 
-                server_conn = TCPConnection(mitm_ip, mitm_port, remote_ip, server_port)
+                next_server_port += 1
+                server_conn = TCPConnection(mitm_ip, next_server_port, remote_ip, server_port)
 
         if tcp.flags.S and tcp.flags.A:  # SYN-ACK from server
             if sender == server_conn:

tests/cpp/ymq/test_ymq.cpp

@@ -669,16 +669,9 @@ TEST_P(CcYmqTestSuiteParametrized, TestClientSendBigMessageToServer)
 // for more, see the python mitm files
 TEST(CcYmqTestSuite, TestMitmPassthrough)
 {
-#ifdef __linux__
-    auto mitm_ip   = "192.0.2.4";
-    auto remote_ip = "192.0.2.3";
-#endif  // __linux__
-#ifdef _WIN32
-    auto mitm_ip   = "127.0.0.1";
-    auto remote_ip = "127.0.0.1";
-#endif  // _WIN32
-    auto mitm_port   = randomPort();
-    auto remote_port = 23571;
+    auto [mitm_ip, remote_ip] = getMitmIPs();
+    auto mitm_port            = 23575;
+    auto remote_port          = 23571;
 
     // the Python program must be the first and only the first function passed to test()
     // we must also pass `true` as the third argument to ensure that Python is fully started
@@ -696,16 +689,9 @@ TEST(CcYmqTestSuite, TestMitmPassthrough)
 // this is the same as the above, but both the client and server use raw sockets
 TEST(CcYmqTestSuite, TestMitmPassthroughRaw)
 {
-#ifdef __linux__
-    auto mitm_ip   = "192.0.2.4";
-    auto remote_ip = "192.0.2.3";
-#endif  // __linux__
-#ifdef _WIN32
-    auto mitm_ip   = "127.0.0.1";
-    auto remote_ip = "127.0.0.1";
-#endif  // _WIN32
-    auto mitm_port   = randomPort();
-    auto remote_port = 23574;
+    auto [mitm_ip, remote_ip] = getMitmIPs();
+    auto mitm_port            = 23576;
+    auto remote_port          = 23574;
 
     // the Python program must be the first and only the first function passed to test()
     // we must also pass `true` as the third argument to ensure that Python is fully started
@@ -722,16 +708,9 @@ TEST(CcYmqTestSuite, TestMitmPassthroughRaw)
 // this test uses the mitm to test the reconnect logic of YMQ by sending RST packets
 TEST(CcYmqTestSuite, TestMitmReconnect)
 {
-#ifdef __linux__
-    auto mitm_ip   = "192.0.2.4";
-    auto remote_ip = "192.0.2.3";
-#endif  // __linux__
-#ifdef _WIN32
-    auto mitm_ip   = "127.0.0.1";
-    auto remote_ip = "127.0.0.1";
-#endif  // _WIN32
-    auto mitm_port   = randomPort();
-    auto remote_port = 23572;
+    auto [mitm_ip, remote_ip] = getMitmIPs();
+    auto mitm_port            = 23577;
+    auto remote_port          = 23572;
 
     auto result = test(
         30,
@@ -746,16 +725,9 @@ TEST(CcYmqTestSuite, TestMitmReconnect)
 // in this test, the mitm drops a random % of packets arriving from the client and server
 TEST(CcYmqTestSuite, TestMitmRandomlyDropPackets)
 {
-#ifdef __linux__
-    auto mitm_ip   = "192.0.2.4";
-    auto remote_ip = "192.0.2.3";
-#endif  // __linux__
-#ifdef _WIN32
-    auto mitm_ip   = "127.0.0.1";
-    auto remote_ip = "127.0.0.1";
-#endif  // _WIN32
-    auto mitm_port   = randomPort();
-    auto remote_port = 23573;
+    auto [mitm_ip, remote_ip] = getMitmIPs();
+    auto mitm_port            = 23578;
+    auto remote_port          = 23573;
 
     auto result = test(
         180,