From 929fc906f4fa9d8a0a5b13332a7cce65334dac46 Mon Sep 17 00:00:00 2001 From: Andrew Tridgell Date: Fri, 25 Sep 2026 08:26:40 +1000 Subject: [PATCH 1/4] Support full-width MAVLink system IDs and payload targets Preserve full source and target IDs while forwarding and re-signing received payloads, including unknown extension bytes. Widen log replies, reboot filters, status identities and configuration without changing the database layout. Use the pinned MAVLink generator for builds, CI and Docker, with end-to-end coverage of forwarding, signing, logging and uint32 configuration boundaries. --- .gitignore | 1 + Makefile | 12 +- README.md | 9 +- binlog.h | 6 +- docker/Dockerfile | 8 +- keydb.py | 2 +- keydb_lib.py | 4 +- mavlink.cpp | 18 ++- mavlink.h | 3 +- modules/mavlink | 2 +- regen_headers.sh | 9 +- scripts/run_tests.py | 4 +- scripts/setup_ci.sh | 4 +- supportproxy.cpp | 9 +- tests/test_keydb_log.py | 15 +- tests/test_sysid32.py | 202 ++++++++++++++++++++++++ tests/webadmin/test_wide_sysid_forms.py | 22 +++ webadmin/forms.py | 4 +- 18 files changed, 294 insertions(+), 40 deletions(-) create mode 100644 tests/test_sysid32.py create mode 100644 tests/webadmin/test_wide_sysid_forms.py diff --git a/.gitignore b/.gitignore index dad5b95..d1fd4d5 100644 --- a/.gitignore +++ b/.gitignore @@ -3,6 +3,7 @@ set_key supportproxy libraries/ +git-version.h venv/ __pycache__/ *.pyc diff --git a/Makefile b/Makefile index bb2579d..0576a47 100644 --- a/Makefile +++ b/Makefile @@ -42,16 +42,21 @@ help: @echo " CXX - C++ compiler (default: g++)" # Git submodules -modules: modules/mavlink/message_definitions/v1.0/all.xml +modules: modules/mavlink/message_definitions/v1.0/all.xml modules/mavlink/pymavlink/generator/mavgen.py modules/mavlink/message_definitions/v1.0/all.xml: @echo "Initializing git submodules..." @git submodule update --init --recursive +modules/mavlink/pymavlink/generator/mavgen.py: | modules/mavlink/message_definitions/v1.0/all.xml + @git submodule update --init --recursive + # MAVLink headers generation headers: $(MAVLINK_DIR)/protocol.h -$(MAVLINK_DIR)/protocol.h: modules/mavlink/message_definitions/v1.0/all.xml +MAVLINK_INPUTS := $(wildcard modules/mavlink/message_definitions/v1.0/*.xml modules/mavlink/pymavlink/generator/*.py modules/mavlink/pymavlink/generator/C/include_v2.0/*.h) + +$(MAVLINK_DIR)/protocol.h: modules/mavlink/message_definitions/v1.0/all.xml regen_headers.sh $(MAVLINK_INPUTS) | modules @echo "Generating MAVLink headers..." @./regen_headers.sh @@ -60,6 +65,9 @@ $(TARGET): $(OBJECTS) @echo "Linking $(TARGET)..." $(CXX) $(CXXFLAGS) -o $@ $^ $(LIBS) +# All users of generated types must rebuild together after a protocol update. +$(OBJECTS): $(MAVLINK_DIR)/protocol.h + # Object file compilation %.o: %.cpp @echo "Compiling $<..." diff --git a/README.md b/README.md index fbda681..bc92507 100644 --- a/README.md +++ b/README.md @@ -11,6 +11,8 @@ For more information on using the support proxy see https://support.ardupilot.or - Both support engineer and user can be on private networks - Supports many users running in parallel - Uses MAVLink2 signed connections from the support engineer +- Preserves 32-bit source system IDs and explicit header targets across + forwarding, signing and telemetry logs - Uses normal UDP/TCP forwarding in users GCS - Supports both TCP and UDP, including mixed connections - Supports WebSocket and WebSocket+SSL TCP connections for both user @@ -157,8 +159,9 @@ python3 -m venv --system-site-packages venv # Activate the virtual environment source venv/bin/activate -# Install pymavlink in the virtual environment -pip install pymavlink +# Install the pinned pymavlink with 32-bit system ID support +git submodule update --init --recursive +pip install ./modules/mavlink/pymavlink ``` ## Building SupportProxy @@ -306,6 +309,7 @@ SupportProxy can also be run using Docker for easier deployment and management. ## Building the Docker Image ```bash +git submodule update --init --recursive docker build -f docker/Dockerfile -t ap-supportproxy . ``` @@ -390,6 +394,7 @@ The `keydb.py` script provides comprehensive database management: ./keydb.py setname PORT2 NewName # Change name ./keydb.py setpass PORT2 NewPassPhrase # Change passphrase ./keydb.py setport1 PORT2 NewPORT1 # Change user port +./keydb.py setsysid PORT2 SYSID # FC reboot filter: 1..4294967295; 0 clears # Reset signing replay-protection timestamp (e.g. after clock skew) ./keydb.py resettimestamp PORT2 diff --git a/binlog.h b/binlog.h index 44b9404..27b9b90 100644 --- a/binlog.h +++ b/binlog.h @@ -112,7 +112,7 @@ class BinlogWriter { to packets with msg.sysid == sysid. Sourced from KeyEntry.fc_sysid at fork start. */ - void set_fc_sysid_filter(uint8_t sysid) { fc_sysid_filter_ = sysid; } + void set_fc_sysid_filter(uint32_t sysid) { fc_sysid_filter_ = sysid; } /* Periodic pump. Called once per main_loop iteration whenever @@ -230,7 +230,7 @@ class BinlogWriter { // First-seen sysid/compid of the vehicle on this log session, // used as target_{system,component} when we send ACK/NACK back. - uint8_t target_system = 0; + uint32_t target_system = 0; uint8_t target_component = 0; // Pending ACK queue (seqnos to ACK, FIFO). @@ -302,7 +302,7 @@ class BinlogWriter { // Per-entry MAVLink sysid filter for SYSTEM_TIME-based reboot // detection. 0 = match any (default). Set from KeyEntry.fc_sysid // by the per-port-pair child at fork. - uint8_t fc_sysid_filter_ = 0; + uint32_t fc_sysid_filter_ = 0; // Most-recently-seen SYSTEM_TIME.time_boot_ms from the autopilot. // 0 = nothing seen yet (used as a guard so the very first diff --git a/docker/Dockerfile b/docker/Dockerfile index 4b115d7..eb0e813 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -25,14 +25,14 @@ WORKDIR /app # Copy source code (from parent directory) COPY . . +# The build context excludes Git metadata; initialize submodules on the host. +RUN test -f modules/mavlink/pymavlink/tools/mavgen.py + # Create virtual environment with system site packages and install pymavlink # --system-site-packages allows access to system python3-tdb package RUN python3 -m venv --system-site-packages venv && \ . venv/bin/activate && \ - pip install --no-cache-dir pymavlink - -# Initialize git submodules and build -RUN git submodule update --init --recursive || true + pip install --no-cache-dir ./modules/mavlink/pymavlink # Build the supportproxy binary with venv activated RUN . venv/bin/activate && make clean && make diff --git a/keydb.py b/keydb.py index 11800f3..496433e 100755 --- a/keydb.py +++ b/keydb.py @@ -264,7 +264,7 @@ def main(): elif args.action == "setsysid": _expect(args.args, 2, "keydb.py setsysid PORT2 SYSID " - "(0 = match any, 1..255 = filter to that MAVLink sysid)") + "(0 = match any, 1..4294967295 = filter to that MAVLink sysid)") try: sysid = int(args.args[1]) except ValueError: diff --git a/keydb_lib.py b/keydb_lib.py index 8e99343..9dfd55e 100644 --- a/keydb_lib.py +++ b/keydb_lib.py @@ -796,8 +796,8 @@ def set_fc_sysid(db, port2, sysid): ke = KeyEntry(port2) if not ke.fetch(db): raise CLIError("No entry for port2 %d" % port2) - if sysid < 0 or sysid > 255: - raise CLIError("fc_sysid must be in 0..255 (got %r)" % sysid) + if sysid < 0 or sysid > 0xFFFFFFFF: + raise CLIError("fc_sysid must be in 0..4294967295 (got %r)" % sysid) ke.fc_sysid = int(sysid) ke.store(db) return ke diff --git a/mavlink.cpp b/mavlink.cpp index 6f2af8b..4c7c42a 100644 --- a/mavlink.cpp +++ b/mavlink.cpp @@ -188,7 +188,7 @@ bool MAVLink::receive_message(uint8_t *&buf, ssize_t &len, mavlink_message_t &ms bool MAVLink::send_message(const mavlink_message_t &msg) { mavlink_message_t msg2 = msg; - uint8_t buf[300]; + uint8_t buf[MAVLINK_MAX_PACKET_LEN]; if (is_tcp) { if (socket_is_dead(fd)) { return false; @@ -235,7 +235,19 @@ bool MAVLink::send_message(const mavlink_message_t &msg) // packet loss information status->current_tx_seq = msg.seq; - mavlink_finalize_message_buffer(&msg2, msg2.sysid, msg2.compid, status, min_len, max_len, crc_extra); + // Re-sign only the received payload bytes, retaining wide target IDs. + uint16_t finalized_len; + if (msg.incompat_flags & MAVLINK_IFLAG_TARGET32) { + finalized_len = mavlink_finalize_message_buffer_target( + &msg2, msg.sysid, msg.compid, status, min_len, msg.len, crc_extra, + msg.target_sysid); + } else { + finalized_len = mavlink_finalize_message_buffer( + &msg2, msg.sysid, msg.compid, status, min_len, msg.len, crc_extra); + } + if (finalized_len == 0) { + return false; + } uint16_t len = mavlink_msg_to_send_buffer(buf, &msg2); if (len > 0) { @@ -470,7 +482,7 @@ void MAVLink::mav_printf(uint8_t severity, const char *fmt, ...) severity, text, 0, 0); - uint8_t buf[300]; + uint8_t buf[MAVLINK_MAX_PACKET_LEN]; uint16_t len = mavlink_msg_to_send_buffer(buf, &msg); if (len > 0) { ::printf("[%d]: %s\n", key_id, text); diff --git a/mavlink.h b/mavlink.h index baa1c0d..92ef206 100644 --- a/mavlink.h +++ b/mavlink.h @@ -76,7 +76,8 @@ class MAVLink { double last_signing_warning_s = 0; // last source sysid and compid from a HEARTBEAT from user - uint8_t last_sysid, last_compid; + uint32_t last_sysid; + uint8_t last_compid; // count of signature errors for triggering message uint32_t bad_sig_count = 0; diff --git a/modules/mavlink b/modules/mavlink index 7bdb2dc..a755fae 160000 --- a/modules/mavlink +++ b/modules/mavlink @@ -1 +1 @@ -Subproject commit 7bdb2dc7d679629eb80e7f28973334ca860f11c9 +Subproject commit a755faeb4d1ae39b40b22550764417cdd28e19c4 diff --git a/regen_headers.sh b/regen_headers.sh index a8690be..34d8c5c 100755 --- a/regen_headers.sh +++ b/regen_headers.sh @@ -1,10 +1,11 @@ #!/bin/bash -# re-generate mavlink headers, assumes pymavlink is installed +# Generate headers with the pymavlink revision pinned by our MAVLink submodule. +set -euo pipefail +cd "$(dirname "$0")" +export PYTHONPATH="$PWD/modules/mavlink${PYTHONPATH:+:$PYTHONPATH}" echo "Generating mavlink2 headers" rm -rf libraries/mavlink2/generated -mavgen.py --no-validate --wire-protocol 2.0 --lang C modules/mavlink/message_definitions/v1.0/all.xml -o libraries/mavlink2/generated +python3 modules/mavlink/pymavlink/tools/mavgen.py --no-validate --wire-protocol 2.0 --lang C modules/mavlink/message_definitions/v1.0/all.xml -o libraries/mavlink2/generated ./git-version.sh - - diff --git a/scripts/run_tests.py b/scripts/run_tests.py index 90d96ec..148f1d6 100755 --- a/scripts/run_tests.py +++ b/scripts/run_tests.py @@ -31,7 +31,9 @@ PHASES = [ ('Connection Tests', ['tests/test_connections.py']), ('Authentication Tests', ['tests/test_authentication.py']), - ('Robustness Tests', ['tests/test_parent_housekeeping.py', + ('Robustness Tests', ['tests/test_sysid32.py', + 'tests/test_keydb_log.py', + 'tests/test_parent_housekeeping.py', 'tests/test_conn2_slot_orphan.py', 'tests/test_drop_lost_request.py', 'tests/test_websocket_decode.py', diff --git a/scripts/setup_ci.sh b/scripts/setup_ci.sh index cb66bd6..d202417 100755 --- a/scripts/setup_ci.sh +++ b/scripts/setup_ci.sh @@ -41,7 +41,9 @@ source venv/bin/activate echo "Installing Python dependencies..." pip install --upgrade pip -pip install pytest pytest-xdist pymavlink wsproto Flask Flask-WTF +git submodule update --init --recursive +pip install pytest pytest-xdist wsproto Flask Flask-WTF +pip install ./modules/mavlink/pymavlink echo "Verifying tdb module accessibility..." python3 -c "import tdb; print('tdb module is available')" diff --git a/supportproxy.cpp b/supportproxy.cpp index bc987f1..5b3b4f5 100644 --- a/supportproxy.cpp +++ b/supportproxy.cpp @@ -92,7 +92,7 @@ struct listen_port { uint32_t video_flags; uint32_t video_flags_hi; // slots past KEY_VIDEO_PORTS_INLINE uint32_t flags; - uint8_t fc_sysid; // 0 = match any; otherwise the FC's MAVLink + uint32_t fc_sysid; // 0 = match any; otherwise the FC's MAVLink // sysid for binlog reboot detection float tz_offset_hours; // log-naming timezone (GMT offset in hours) bool seen; // set true by handle_record() during reload_ports() @@ -182,7 +182,7 @@ static void video_stop_child(struct listen_port *p, const char *why) kill(p->video_pid, SIGTERM); } -static void upsert_port(int port1, int port2, uint32_t flags, uint8_t fc_sysid, +static void upsert_port(int port1, int port2, uint32_t flags, uint32_t fc_sysid, float tz_offset_hours, const uint32_t *video_ports, uint32_t video_flags, uint32_t video_flags_hi) { @@ -270,16 +270,13 @@ static int handle_record(struct tdb_context *db, TDB_DATA key, TDB_DATA data, vo memcpy(&port2, key.dptr, sizeof(int)); size_t copy = data.dsize < sizeof(KeyEntry) ? data.dsize : sizeof(KeyEntry); memcpy(&k, data.dptr, copy); - // KeyEntry.fc_sysid is uint32 for forward compat; the wire value is - // a MAVLink sysid (0..255), so truncate to uint8 once it crosses the - // C++/binlog boundary. The CLI / web UI already cap at 255. // The slots are split across two field groups on disk; hand // upsert_port one flat array so nothing downstream has to know. uint32_t vports[KEY_MAX_VIDEO_PORTS]; for (unsigned i = 0; i < KEY_MAX_VIDEO_PORTS; i++) { vports[i] = video_port_of(k, i); } - upsert_port(k.port1, port2, k.flags, uint8_t(k.fc_sysid), + upsert_port(k.port1, port2, k.flags, k.fc_sysid, k.tz_offset_hours, vports, k.video_flags, k.video_flags_hi); return 0; } diff --git a/tests/test_keydb_log.py b/tests/test_keydb_log.py index a54fbf1..c131244 100644 --- a/tests/test_keydb_log.py +++ b/tests/test_keydb_log.py @@ -260,16 +260,17 @@ def test_fc_sysid_default_is_zero(tmp_path): assert ke.fc_sysid == 0 -def test_set_fc_sysid_round_trip(tmp_path): +@pytest.mark.parametrize('sysid', [0, 42, 255, 256, 0x7FFFFFFF, 0x80000000, 0xFFFFFFFF]) +def test_set_fc_sysid_round_trip(tmp_path, sysid): p = str(tmp_path / 'keys.tdb') db = keydb_lib.init_db(p) db.transaction_start() keydb_lib.add_entry(db, 17101, 17102, 'sysid', 'pw') - keydb_lib.set_fc_sysid(db, 17102, 42) + keydb_lib.set_fc_sysid(db, 17102, sysid) ke = keydb_lib.KeyEntry(17102) ke.fetch(db) db.transaction_cancel() - assert ke.fc_sysid == 42 + assert ke.fc_sysid == sysid def test_set_fc_sysid_rejects_out_of_range(tmp_path): @@ -280,7 +281,7 @@ def test_set_fc_sysid_rejects_out_of_range(tmp_path): with pytest.raises(keydb_lib.CLIError): keydb_lib.set_fc_sysid(db, 17202, -1) with pytest.raises(keydb_lib.CLIError): - keydb_lib.set_fc_sysid(db, 17202, 256) + keydb_lib.set_fc_sysid(db, 17202, 0x100000000) db.transaction_cancel() @@ -288,11 +289,11 @@ def test_cli_setsysid_then_list_shows_sysid(tmp_path): p = str(tmp_path / 'keys.tdb') _run_cli(p, 'initialise') _run_cli(p, 'add', '17301', '17302', 'CliSysid', 'pw') - r = _run_cli(p, 'setsysid', '17302', '7') + r = _run_cli(p, 'setsysid', '17302', '4294967295') assert r.returncode == 0, r.stderr - assert 'fc_sysid=7' in r.stdout + assert 'fc_sysid=4294967295' in r.stdout r = _run_cli(p, 'list') - assert 'fc_sysid=7' in r.stdout + assert 'fc_sysid=4294967295' in r.stdout # Clearing back to 0 hides it from list output again. r = _run_cli(p, 'setsysid', '17302', '0') assert r.returncode == 0 diff --git a/tests/test_sysid32.py b/tests/test_sysid32.py new file mode 100644 index 0000000..3e82b66 --- /dev/null +++ b/tests/test_sysid32.py @@ -0,0 +1,202 @@ +"""Exercise full-width identities and explicit targets through the real proxy.""" +import hashlib +import os +from pathlib import Path +import signal +import socket +import subprocess +import time + +import pytest +from pymavlink import mavutil + +import keydb_lib +from test_config import SUPPORTPROXY_BIN +from test_binlog_capture import (_send_data_block, _send_system_time, + _recv_block_status_msgs, _wait_bin, _bin_path) + +KEY = hashlib.sha256(b'sysid32-test').digest() +WORKER = int(os.environ.get('PYTEST_XDIST_WORKER', 'gw0').removeprefix('gw')) +PORT_USER = 24000 + WORKER * 2 +PORT_ENG = PORT_USER + 1 + + +@pytest.fixture +def wide_proxy(tmp_path, request): + flags, sysid = getattr(request, 'param', (['tlog'], 0)) + db = keydb_lib.init_db(str(tmp_path / 'keys.tdb')) + db.transaction_start() + keydb_lib.add_entry(db, PORT_USER, PORT_ENG, 'wide', 'sysid32-test') + for flag in flags: + keydb_lib.set_flag(db, PORT_ENG, flag) + keydb_lib.set_fc_sysid(db, PORT_ENG, sysid) + db.transaction_prepare_commit() + db.transaction_commit() + db.close() + log = (tmp_path / 'proxy.log').open('w') + proc = subprocess.Popen([SUPPORTPROXY_BIN], cwd=tmp_path, stdout=log, + stderr=subprocess.STDOUT, start_new_session=True) + try: + want = {'%04X' % port for port in (PORT_USER, PORT_ENG)} + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + lines = Path('/proc/net/tcp').read_text().splitlines()[1:] + listening = {line.split()[1].split(':')[1] for line in lines + if line.split()[3] == '0A'} + if want <= listening: + break + assert proc.poll() is None, (tmp_path / 'proxy.log').read_text() + time.sleep(0.02) + else: + pytest.fail('Proxy did not start: ' + (tmp_path / 'proxy.log').read_text()) + yield tmp_path + finally: + if proc.poll() is None: + os.killpg(proc.pid, signal.SIGTERM) + proc.wait(timeout=10) + log.close() + + +def heartbeat(tag): + return mavutil.mavlink.MAVLink_heartbeat_message( + mavutil.mavlink.MAV_TYPE_QUADROTOR, + mavutil.mavlink.MAV_AUTOPILOT_ARDUPILOTMEGA, 0, tag, 0, 3) + + +def receive(link, kind, predicate, timeout=4): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + msg = link.recv_match(type=kind, blocking=True, timeout=0.1) + if msg is not None and predicate(msg): + return msg + pytest.fail('No matching %s received' % kind) + + +@pytest.mark.parametrize('transport', ['udpout', 'tcp', 'ws']) +@pytest.mark.parametrize('wide_proxy', [(['tlog'], 0), (['tlog', 'bidi_sign'], 0)], indirect=True) +def test_forward_wide_ids_and_targets(wide_proxy, transport, request): + bidi = 'bidi_sign' in request.node.callspec.params['wide_proxy'][0] + def connect(port, source): + endpoint = (transport + ':127.0.0.1:%d') % port + return mavutil.mavlink_connection(endpoint, source_system=source, source_component=11) + user = connect(PORT_USER, 0xFEDCBA98) + if bidi: + user.setup_signing(KEY, sign_outgoing=True) + user.mav.send(heartbeat(100)) + eng = connect(PORT_ENG, 0xFFFFFFFF) + eng.setup_signing(KEY, sign_outgoing=True) + try: + eng.mav.send(heartbeat(101)) + receive(user, 'HEARTBEAT', lambda m: m.custom_mode == 101) + cases = [(42, None), (0xFFFFFFFF, None), (42, 7), (42, 0), + (42, 0xFFFFFFFF), (0xFEDCBA98, 0x80000000)] + expected = set() + tag = 200 + for sender, receiver, signed in [(user, eng, True), (eng, user, bidi)]: + for source, target in cases: + sender.mav.srcSystem = source + if target is None: + msg = heartbeat(tag) + predicate = lambda m: m.custom_mode == tag + else: + msg = mavutil.mavlink.MAVLink_command_long_message(target, 250, 300, 1, + tag, 2, 3, 4, 5, 6, 7) + predicate = lambda m: m.param1 == tag + sender.mav.send(msg) + got = receive(receiver, msg.get_type(), predicate) + assert got.get_srcSystem() == source + assert got.get_srcComponent() == 11 + assert got.get_seq() == msg.get_seq() + assert got.get_target_system() == target + assert bool(got.get_header().incompat_flags & 1) == signed + if target is not None: + assert got.get_target_component() == 250 + expected.add((source, target, tag)) + tag += 1 + # Maximum-size targeted payloads must retain their contents. + msg = mavutil.mavlink.MAVLink_file_transfer_protocol_message(0, 0x87654321, 250, list(range(251))) + sender.mav.send(msg) + got = receive(receiver, 'FILE_TRANSFER_PROTOCOL', lambda m: m.get_target_system() == 0x87654321) + assert got.get_payload() == msg.get_payload() + assert got.get_payload()[1:3] == bytes([255, 250]) + assert list(got.payload) == list(range(251)) + # A future extension byte fills the 255-byte payload: the relay + # must retain it even though its dialect only knows 254 bytes. + payload = bytes(msg.get_payload()) + b'\xa5' + sender.write(msg._pack(sender.mav, msg.crc_extra, payload)) + got = receive(receiver, 'FILE_TRANSFER_PROTOCOL', lambda m: len(m.get_payload()) == 255) + assert got.get_payload() == payload + assert got.get_target_system() == 0x87654321 + assert len(got.get_msgbuf()) == (287 if signed else 274) + if not bidi: + # Legacy MAVLink1 user traffic still forwards with signing. + user.mav.srcSystem = 42 + user.mav.send(heartbeat(998), force_mavlink1=True) + got = receive(eng, 'HEARTBEAT', lambda m: m.custom_mode == 998) + assert got.get_srcSystem() == 42 + assert got.get_target_system() is None + # Incoming signatures must still be checked for full-width sources. + eng.mav.srcSystem = 0xFFFFFFFF + invalid = bytearray(heartbeat(999).pack(eng.mav)) + invalid[-1] ^= 1 + eng.write(invalid) + assert user.recv_match(type='HEARTBEAT', blocking=True, timeout=0.3) is None + eng.mav.send(heartbeat(1000)) + receive(user, 'HEARTBEAT', lambda m: m.custom_mode == 1000) + # Proxy-generated signing warnings use the full remembered vehicle ID. + user.mav.srcSystem = 0xFEDCBA98 + user.mav.send(heartbeat(1001)) + receive(eng, 'HEARTBEAT', lambda m: m.custom_mode == 1001) + eng.mav.signing.sign_outgoing = False + deadline = time.monotonic() + 3 + warning = None + while time.monotonic() < deadline: + eng.mav.send(heartbeat(1002)) + warning = eng.recv_match(type='STATUSTEXT', blocking=True, timeout=0.2) + if warning is not None: + break + assert warning is not None + assert warning.get_srcSystem() == 0xFEDCBA98 + assert warning.get_srcComponent() == 11 + assert warning.text == 'Need to use support signing key' + finally: + eng.close() + user.close() + logs = list(wide_proxy.rglob('*.tlog')) + assert logs + recorded = set() + for path in logs: + # The mmap indexer still assumes fixed-size MAVLink2 headers. + reader = mavutil.mavlogfile(str(path)) + try: + while (msg := reader.recv_match(type=['HEARTBEAT', 'COMMAND_LONG'])) is not None: + tag = msg.custom_mode if msg.get_type() == "HEARTBEAT" else int(msg.param1) + recorded.add((msg.get_srcSystem(), msg.get_target_system(), tag)) + finally: + reader.close() + assert expected <= recorded + + +@pytest.mark.parametrize('wide_proxy', [(['binlog'], 257), (['binlog'], 0x80000001), + (['binlog'], 0xFFFFFFFF)], indirect=True) +def test_wide_binlog_ack_and_reboot_filter(wide_proxy, request): + sysid = request.node.callspec.params['wide_proxy'][1] + dest = ('127.0.0.1', PORT_USER) + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock: + sock.bind(('127.0.0.1', 0)) + _send_system_time(sock, dest, 60000, sysid=sysid) + _send_data_block(sock, dest, 0, b'A' * 200, sysid=sysid) + assert _wait_bin(wide_proxy, PORT_ENG, min_size=200) + replies = _recv_block_status_msgs(sock, timeout=0.5) + assert any(m.seqno == 0 and m.get_target_system() == sysid for m in replies) + # Same low byte is a different vehicle and must not trigger rotation. + _send_system_time(sock, dest, 1000, sysid=sysid & 255) + _send_data_block(sock, dest, 1, b'B' * 200, sysid=sysid) + assert _wait_bin(wide_proxy, PORT_ENG, min_size=400) + assert not _bin_path(wide_proxy, PORT_ENG, 2).exists() + # The matching full-width source must trigger rotation. + _send_system_time(sock, dest, 1000, sysid=sysid) + _send_data_block(sock, dest, 0, b'C' * 200, sysid=sysid) + second = _wait_bin(wide_proxy, PORT_ENG, n=2, min_size=200) + assert second + assert second.read_bytes()[:200] == b'C' * 200 diff --git a/tests/webadmin/test_wide_sysid_forms.py b/tests/webadmin/test_wide_sysid_forms.py new file mode 100644 index 0000000..b40628a --- /dev/null +++ b/tests/webadmin/test_wide_sysid_forms.py @@ -0,0 +1,22 @@ +"""Full-width system ID configuration through both web forms.""" +import pytest +from _test_helpers import (ALICE_PASS, ALICE_PORT1, ALICE_PORT2, + BOB_PASS, BOB_PORT1, fetch_entry, login_as) + + +@pytest.mark.parametrize('admin', [False, True]) +@pytest.mark.parametrize('sysid', [0, 255, 256, 0x80000000, 0xFFFFFFFF, -1, 0x100000000]) +def test_fc_sysid_range(client, keydb_path, admin, sysid): + login_as(client, BOB_PORT1 if admin else ALICE_PORT1, + BOB_PASS if admin else ALICE_PASS) + url = '/admin/%d' % ALICE_PORT2 if admin else '/me/' + data = {'name': 'alice', 'fc_sysid': str(sysid), 'submit': 'Save'} + if admin: + data.update(port1=str(ALICE_PORT1), port2=str(ALICE_PORT2)) + result = client.post(url, data=data) + if 0 <= sysid <= 0xFFFFFFFF: + assert result.status_code == 302 + assert fetch_entry(keydb_path, ALICE_PORT2).fc_sysid == sysid + else: + assert result.status_code == 200 + assert fetch_entry(keydb_path, ALICE_PORT2).fc_sysid == 0 diff --git a/webadmin/forms.py b/webadmin/forms.py index 2102c68..c76f633 100644 --- a/webadmin/forms.py +++ b/webadmin/forms.py @@ -300,7 +300,7 @@ class OwnerEditForm(FlaskForm, _VideoOwnerFields): log_access = _log_access_field() fc_sysid = IntegerField( 'Flight-controller MAVLink sysid (0 = any)', description=_D_SYSID, - validators=[Optional(), NumberRange(min=0, max=255)]) + validators=[Optional(), NumberRange(min=0, max=0xFFFFFFFF)]) use_tz = BooleanField('Name logs in a fixed timezone', description=_D_USE_TZ) tz_offset_hours = FloatField( @@ -354,7 +354,7 @@ class AdminEditForm(FlaskForm, _VideoAdminFields): log_access = _log_access_field() fc_sysid = IntegerField( 'Flight-controller MAVLink sysid (0 = any)', description=_D_SYSID, - validators=[Optional(), NumberRange(min=0, max=255)]) + validators=[Optional(), NumberRange(min=0, max=0xFFFFFFFF)]) use_tz = BooleanField('Name logs in a fixed timezone', description=_D_USE_TZ) tz_offset_hours = FloatField( From 2e049e066c1e4da1ea1614ee833f757890aa7592 Mon Sep 17 00:00:00 2001 From: Andrew Tridgell Date: Mon, 28 Sep 2026 15:20:27 +1000 Subject: [PATCH 2/4] Use merged MAVLink system-ID support Pin the upstream merged generator and exercise its default mmap-backed tlog reader. Clarify payload targeting for 8-bit peers and the configuration limits when rolling back to an older proxy. --- README.md | 8 +++++++- modules/mavlink | 2 +- tests/test_sysid32.py | 3 +-- 3 files changed, 9 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index bc92507..2512bd0 100644 --- a/README.md +++ b/README.md @@ -11,7 +11,7 @@ For more information on using the support proxy see https://support.ardupilot.or - Both support engineer and user can be on private networks - Supports many users running in parallel - Uses MAVLink2 signed connections from the support engineer -- Preserves 32-bit source system IDs and explicit header targets across +- Preserves 32-bit source and destination system IDs across forwarding, signing and telemetry logs - Uses normal UDP/TCP forwarding in users GCS - Supports both TCP and UDP, including mixed connections @@ -20,6 +20,12 @@ For more information on using the support proxy see https://support.ardupilot.or - supports up to 8 simultaneous connections by support engineer - Optional video proxying alongside the MAVLink link, with recording +System IDs above 255 require peers that understand the MAVLink system-ID +extensions. Older peers cannot read these frames, including proxy diagnostics +sent with a wide vehicle ID. Before rolling back to an older proxy binary, +restore configured flight-controller system IDs to the 8-bit range; older +binaries truncate the binlog source filter, and 256 becomes the match-any value 0. + ## How It Works ![SupportProxy Architecture](supportproxy-diagram.svg) diff --git a/modules/mavlink b/modules/mavlink index a755fae..a23652e 160000 --- a/modules/mavlink +++ b/modules/mavlink @@ -1 +1 @@ -Subproject commit a755faeb4d1ae39b40b22550764417cdd28e19c4 +Subproject commit a23652e3c7d1d7a94b1c2d354a13ce4858cd0996 diff --git a/tests/test_sysid32.py b/tests/test_sysid32.py index 3e82b66..80ff822 100644 --- a/tests/test_sysid32.py +++ b/tests/test_sysid32.py @@ -166,8 +166,7 @@ def connect(port, source): assert logs recorded = set() for path in logs: - # The mmap indexer still assumes fixed-size MAVLink2 headers. - reader = mavutil.mavlogfile(str(path)) + reader = mavutil.mavlink_connection(str(path)) try: while (msg := reader.recv_match(type=['HEARTBEAT', 'COMMAND_LONG'])) is not None: tag = msg.custom_mode if msg.get_type() == "HEARTBEAT" else int(msg.param1) From d910fed93b6b769c11f0dc5c74769877338b7f9a Mon Sep 17 00:00:00 2001 From: Andrew Tridgell Date: Mon, 28 Sep 2026 15:20:27 +1000 Subject: [PATCH 3/4] Run all regression phases and isolate proxy test children An early pytest failure skipped later phases, and several regression files were omitted from the standard runner. Run every phase while preserving failure status, include the missing tests, and terminate each test proxy process group so orphaned children cannot retain ports used by subsequent tests. --- scripts/run_tests.py | 37 ++++++++++++++++++++++++--------- tests/test_binlog_capture.py | 18 ++++++++++++---- tests/test_conn2_slot_orphan.py | 18 ++++++++++++---- tests/test_drop_lost_request.py | 18 ++++++++++++---- tests/test_kill_drop.py | 18 ++++++++++++---- tests/test_run_tests.py | 35 +++++++++++++++++++++++++++++++ tests/test_tlog_capture.py | 18 ++++++++++++---- 7 files changed, 132 insertions(+), 30 deletions(-) create mode 100644 tests/test_run_tests.py diff --git a/scripts/run_tests.py b/scripts/run_tests.py index 148f1d6..0f705a3 100755 --- a/scripts/run_tests.py +++ b/scripts/run_tests.py @@ -2,8 +2,8 @@ """ Test runner for SupportProxy. -Default behaviour: build supportproxy, then run three pytest invocations -(connection, authentication, webadmin tests) — each phase has different +Default behaviour: build supportproxy, then run four pytest invocations +(connection, authentication, robustness, webadmin tests) — each phase has different cwd / keys.tdb / process expectations, so they stay isolated. Pass -j N for parallel test execution via pytest-xdist; each worker gets @@ -11,12 +11,12 @@ test gets its own worker), useful for the connection phase where the slowest worker pins wall-clock time. -Pass --list to enumerate tests across all three phases without running. +Pass --list to enumerate tests across all four phases without running. Pass test selectors as positional args (any pytest selector works: file path, dir, NodeID, -k expression). When selectors are present the runner does ONE pytest invocation against exactly what you asked for, -skipping the three-phase split. +skipping the four-phase split. """ import argparse import os @@ -32,6 +32,14 @@ ('Connection Tests', ['tests/test_connections.py']), ('Authentication Tests', ['tests/test_authentication.py']), ('Robustness Tests', ['tests/test_sysid32.py', + 'tests/test_binlog_capture.py', + 'tests/test_engineer_preauth_pool.py', + 'tests/test_engineer_udp_churn.py', + 'tests/test_kill_drop.py', + 'tests/test_log_cleanup.py', + 'tests/test_setup_signing_guard.py', + 'tests/test_tlog_capture.py', + 'tests/test_run_tests.py', 'tests/test_keydb_log.py', 'tests/test_parent_housekeeping.py', 'tests/test_conn2_slot_orphan.py', @@ -159,7 +167,7 @@ def build_pytest_cmd(j, extra_args, target_args, timing=False): def cmd_list(): - """Run pytest --collect-only -q across the three phases.""" + """Run pytest --collect-only -q across the four phases.""" for label, targets in PHASES: print('\n=== %s ===' % label, flush=True) subprocess.call([sys.executable, '-m', 'pytest', '--collect-only', @@ -177,7 +185,7 @@ def main(): ap.add_argument('--no-build', action='store_true', help='skip the make step (use existing supportproxy binary)') ap.add_argument('--list', action='store_true', - help='list all tests across the three phases and exit') + help='list all tests across the four phases and exit') ap.add_argument('--timing', action='store_true', help='print per-test timing at the end, sorted ascending ' '(slowest test last)') @@ -188,7 +196,7 @@ def main(): 'filter via pytest -k (multiple bare words OR\'d ' 'together). Mixing both is fine. With selectors ' 'the runner does one pytest invocation instead ' - 'of the three default phases.') + 'of the four default phases.') args = ap.parse_args() os.chdir(REPO_ROOT) @@ -208,6 +216,7 @@ def main(): sys.exit('ERROR: supportproxy binary not found') all_timings = [] + failed_phases = [] def run_one(extra_args, target_args): cmd = build_pytest_cmd(args.j, extra_args, target_args, args.timing) @@ -232,20 +241,28 @@ def run_one(extra_args, target_args): extra = ['-k', ' or '.join(keywords)] if keywords else [] if not paths: # No path given: search the whole tests/ tree so the keyword - # filter applies across all three phases. + # filter applies across all four phases. paths = ['tests/'] print('\n=== Running selected tests ===') run_one(extra, paths) else: - # Default: three separate phases (kept apart so phase 2 can wipe + # Default: four separate phases (kept apart so phase 2 can wipe # keys.tdb without disturbing phase 1's live supportproxy fixture). for label, targets in PHASES: print('\n=== Running %s ===' % label) - run_one([], targets) + try: + run_one([], targets) + except subprocess.CalledProcessError as exc: + failed_phases.append(label) + print("Phase %s failed with exit code %s" % (label, exc.returncode), flush=True) if args.timing: print_combined_timings(all_timings) + if failed_phases: + print('\nFailed phases: ' + ', '.join(failed_phases)) + return 1 + print('\nAll tests completed.') return 0 diff --git a/tests/test_binlog_capture.py b/tests/test_binlog_capture.py index a5ebede..975ce63 100644 --- a/tests/test_binlog_capture.py +++ b/tests/test_binlog_capture.py @@ -87,7 +87,7 @@ def _start_proxy(workdir, port_eng, quota_bytes=None, cleanup_interval=None): proc = subprocess.Popen( [SUPPORTPROXY_BIN], cwd=str(workdir), env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, - bufsize=1, text=True, + bufsize=1, text=True, start_new_session=True, ) proc._lines = [] proc._ready = threading.Event() @@ -103,7 +103,10 @@ def _drain(): proc._thread = threading.Thread(target=_drain, daemon=True) proc._thread.start() if not proc._ready.wait(timeout=10): - proc.kill() + try: + os.killpg(proc.pid, signal.SIGKILL) + except ProcessLookupError: + pass proc.wait(timeout=2) raise RuntimeError( 'proxy did not load test port pair; stdout: ' @@ -112,11 +115,18 @@ def _drain(): def _terminate(proc): - proc.send_signal(signal.SIGTERM) + # End the whole test session so children cannot keep the next test's ports. + try: + os.killpg(proc.pid, signal.SIGTERM) + except ProcessLookupError: + pass try: proc.wait(timeout=5) except subprocess.TimeoutExpired: - proc.kill() + try: + os.killpg(proc.pid, signal.SIGKILL) + except ProcessLookupError: + pass proc.wait(timeout=2) if hasattr(proc, '_thread'): proc._thread.join(timeout=2) diff --git a/tests/test_conn2_slot_orphan.py b/tests/test_conn2_slot_orphan.py index 04f89e0..5a411d5 100644 --- a/tests/test_conn2_slot_orphan.py +++ b/tests/test_conn2_slot_orphan.py @@ -58,7 +58,7 @@ def _start_proxy(workdir): proc = subprocess.Popen( [SUPPORTPROXY_BIN], cwd=str(workdir), stdout=subprocess.PIPE, stderr=subprocess.STDOUT, - bufsize=1, text=True, + bufsize=1, text=True, start_new_session=True, ) proc._lines = [] proc._ready = threading.Event() @@ -73,18 +73,28 @@ def _drain(): proc._thread = threading.Thread(target=_drain, daemon=True) proc._thread.start() if not proc._ready.wait(timeout=10): - proc.kill() + try: + os.killpg(proc.pid, signal.SIGKILL) + except ProcessLookupError: + pass proc.wait(timeout=2) raise RuntimeError('proxy did not load test port pair') return proc def _terminate(proc): - proc.send_signal(signal.SIGTERM) + # End the whole test session so children cannot keep the next test's ports. + try: + os.killpg(proc.pid, signal.SIGTERM) + except ProcessLookupError: + pass try: proc.wait(timeout=5) except subprocess.TimeoutExpired: - proc.kill() + try: + os.killpg(proc.pid, signal.SIGKILL) + except ProcessLookupError: + pass proc.wait(timeout=2) diff --git a/tests/test_drop_lost_request.py b/tests/test_drop_lost_request.py index 39bcbec..a1221f8 100644 --- a/tests/test_drop_lost_request.py +++ b/tests/test_drop_lost_request.py @@ -60,7 +60,7 @@ def _start_proxy(workdir): proc = subprocess.Popen( [SUPPORTPROXY_BIN], cwd=str(workdir), stdout=subprocess.PIPE, stderr=subprocess.STDOUT, - bufsize=1, text=True, + bufsize=1, text=True, start_new_session=True, ) proc._lines = [] proc._ready = threading.Event() @@ -75,18 +75,28 @@ def _drain(): proc._thread = threading.Thread(target=_drain, daemon=True) proc._thread.start() if not proc._ready.wait(timeout=10): - proc.kill() + try: + os.killpg(proc.pid, signal.SIGKILL) + except ProcessLookupError: + pass proc.wait(timeout=2) raise RuntimeError('proxy did not load test port pair') return proc def _terminate(proc): - proc.send_signal(signal.SIGTERM) + # End the whole test session so children cannot keep the next test's ports. + try: + os.killpg(proc.pid, signal.SIGTERM) + except ProcessLookupError: + pass try: proc.wait(timeout=5) except subprocess.TimeoutExpired: - proc.kill() + try: + os.killpg(proc.pid, signal.SIGKILL) + except ProcessLookupError: + pass proc.wait(timeout=2) diff --git a/tests/test_kill_drop.py b/tests/test_kill_drop.py index 6a4d795..45853b0 100644 --- a/tests/test_kill_drop.py +++ b/tests/test_kill_drop.py @@ -60,7 +60,7 @@ def _start_proxy(workdir): proc = subprocess.Popen( [SUPPORTPROXY_BIN], cwd=str(workdir), stdout=subprocess.PIPE, stderr=subprocess.STDOUT, - bufsize=1, text=True, + bufsize=1, text=True, start_new_session=True, ) proc._lines = [] proc._ready = threading.Event() @@ -75,18 +75,28 @@ def _drain(): proc._thread = threading.Thread(target=_drain, daemon=True) proc._thread.start() if not proc._ready.wait(timeout=10): - proc.kill() + try: + os.killpg(proc.pid, signal.SIGKILL) + except ProcessLookupError: + pass proc.wait(timeout=2) raise RuntimeError('proxy did not load test port pair') return proc def _terminate(proc): - proc.send_signal(signal.SIGTERM) + # End the whole test session so children cannot keep the next test's ports. + try: + os.killpg(proc.pid, signal.SIGTERM) + except ProcessLookupError: + pass try: proc.wait(timeout=5) except subprocess.TimeoutExpired: - proc.kill() + try: + os.killpg(proc.pid, signal.SIGKILL) + except ProcessLookupError: + pass proc.wait(timeout=2) diff --git a/tests/test_run_tests.py b/tests/test_run_tests.py new file mode 100644 index 0000000..3d8244b --- /dev/null +++ b/tests/test_run_tests.py @@ -0,0 +1,35 @@ +"""Keep a failing phase from hiding later CI coverage.""" +import importlib.util +from pathlib import Path +import subprocess +import sys + +import pytest + + +@pytest.mark.parametrize('timing', [False, True]) +@pytest.mark.parametrize('failure', [False, True]) +def test_all_phases_run_and_report_failures(monkeypatch, capsys, timing, failure): + path = Path(__file__).resolve().parents[1] / 'scripts' / 'run_tests.py' + spec = importlib.util.spec_from_file_location('supportproxy_test_runner', path) + runner = importlib.util.module_from_spec(spec) + spec.loader.exec_module(runner) + calls = [] + + def invoke(cmd, **kwargs): + calls.append(cmd) + if failure and 'tests/test_sysid32.py' in cmd: + raise subprocess.CalledProcessError(1, cmd) + return '' + + monkeypatch.setattr(runner, 'run', invoke) + monkeypatch.setattr(runner, 'run_capture', invoke) + monkeypatch.setattr(runner.os, 'chdir', lambda _: None) + monkeypatch.setattr(runner.os.path, 'isfile', lambda _: True) + monkeypatch.setattr(sys, 'argv', ['run_tests.py', '--no-build'] + (['--timing'] if timing else [])) + assert runner.main() == (1 if failure else 0) + assert any('tests/test_sysid32.py' in cmd for cmd in calls) + assert 'tests/webadmin/' in calls[-1] + output = capsys.readouterr().out + assert ('Failed phases: Robustness Tests' in output) == failure + assert ('All tests completed.' in output) != failure diff --git a/tests/test_tlog_capture.py b/tests/test_tlog_capture.py index f0c1880..cbbd71e 100644 --- a/tests/test_tlog_capture.py +++ b/tests/test_tlog_capture.py @@ -97,7 +97,7 @@ def _start_proxy(workdir, env_extra=None): proc = subprocess.Popen( [SUPPORTPROXY_BIN], cwd=str(workdir), env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, - bufsize=1, text=True, + bufsize=1, text=True, start_new_session=True, ) proc._lines = [] proc._ready = threading.Event() @@ -112,7 +112,10 @@ def _drain(): proc._thread = threading.Thread(target=_drain, daemon=True) proc._thread.start() if not proc._ready.wait(timeout=10): - proc.kill() + try: + os.killpg(proc.pid, signal.SIGKILL) + except ProcessLookupError: + pass proc.wait(timeout=2) raise RuntimeError( 'proxy did not load test port pair; stdout: ' @@ -131,11 +134,18 @@ def _wait_for_log(proc, needle, timeout=10.0): def _terminate(proc): - proc.send_signal(signal.SIGTERM) + # End the whole test session so children cannot keep the next test's ports. + try: + os.killpg(proc.pid, signal.SIGTERM) + except ProcessLookupError: + pass try: proc.wait(timeout=5) except subprocess.TimeoutExpired: - proc.kill() + try: + os.killpg(proc.pid, signal.SIGKILL) + except ProcessLookupError: + pass proc.wait(timeout=2) if hasattr(proc, '_thread'): proc._thread.join(timeout=2) From 8503e6addc042b23b375201369ca7995fd942736 Mon Sep 17 00:00:00 2001 From: Andrew Tridgell Date: Mon, 28 Sep 2026 15:37:37 +1000 Subject: [PATCH 4/4] tests: wait for proxy readiness and scope backend checks The binlog test could lose its first block before the sockets were bound, and the RTMP test stopped its publisher during the handshake. Wait for listeners and a joinable stream, settle shared ports for directly created sessions, and track only the tested proxy's backend PIDs so parallel workers cannot cause false orphan failures. --- tests/test_binlog_capture.py | 9 +++++++ tests/test_video_rtsp.py | 52 +++++++++++++++++++++++++++--------- 2 files changed, 48 insertions(+), 13 deletions(-) diff --git a/tests/test_binlog_capture.py b/tests/test_binlog_capture.py index 975ce63..0d9ea55 100644 --- a/tests/test_binlog_capture.py +++ b/tests/test_binlog_capture.py @@ -32,6 +32,8 @@ import pytest +from conftest import _wait_listening + _REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir)) if _REPO_ROOT not in sys.path: sys.path.insert(0, _REPO_ROOT) @@ -111,6 +113,13 @@ def _drain(): raise RuntimeError( 'proxy did not load test port pair; stdout: ' + ''.join(proc._lines)) + # The configuration log precedes binding the sockets. Losing block zero + # here leaves the strict-start gate closed for all later log blocks. + try: + _wait_listening([port_eng]) + except RuntimeError: + _terminate(proc) + raise return proc diff --git a/tests/test_video_rtsp.py b/tests/test_video_rtsp.py index 262669a..1c27388 100644 --- a/tests/test_video_rtsp.py +++ b/tests/test_video_rtsp.py @@ -84,6 +84,7 @@ def _workdir(tmp_path, record=True, publish_pass=None, class RtspSession: def __init__(self, workdir, with_mav=True): + assert _settle(), 'previous session still active: %s' % _settle_state() self.workdir = workdir self.proxy = Proxy(workdir) assert self.proxy.wait_for(r'video slot 0 listening'), self.proxy.log @@ -226,8 +227,6 @@ def session(tmp_path): to be released before yielding keeps a slow teardown from failing the next test rather than its own. """ - _settle() - made = {} def _start(**kw): @@ -246,9 +245,34 @@ def _start(**kw): def _ffmpeg_children(proxy): """ffmpeg processes descended from this proxy.""" - out = subprocess.run(['pgrep', '-a', '-x', 'ffmpeg'], + out = subprocess.run(['ps', '-eo', 'pid=,ppid=,comm='], capture_output=True, text=True).stdout - return [ln for ln in out.splitlines() if 'rtsp://127.0.0.1' in ln] + processes = {} + for line in out.splitlines(): + pid, parent, name = line.split(maxsplit=2) + processes[int(pid)] = (int(parent), name) + children = [] + for pid, (_, name) in processes.items(): + if name != 'ffmpeg': + continue + parent = processes[pid][0] + seen = set() + while parent in processes and parent not in seen: + if parent == proxy.proc.pid: + children.append(pid) + break + seen.add(parent) + parent = processes[parent][0] + return children + + +def test_ffmpeg_children_excludes_other_workers(monkeypatch): + from types import SimpleNamespace + monkeypatch.setattr(subprocess, 'run', lambda *a, **kw: SimpleNamespace(stdout=( + '100 1 supportproxy\n101 100 supportproxy\n102 101 ffmpeg\n' + '200 1 supportproxy\n201 200 ffmpeg\n202 1 ffmpeg\n'))) + proxy = SimpleNamespace(proc=SimpleNamespace(pid=100)) + assert _ffmpeg_children(proxy) == [102] @pytest.mark.integration @@ -344,28 +368,30 @@ def test_no_orphan_backend_after_the_publisher_leaves(self, session, clip): s = session() s.publish(clip) assert s.proxy.wait_for(r'RTSP backend pid \d+', timeout=20), s.proxy.log - assert _ffmpeg_children(s.proxy), 'no backend running while publishing' + backends = _ffmpeg_children(s.proxy) + assert backends, 'no backend running while publishing' s.stop_publisher() assert s.proxy.wait_for(r'RTSP publisher gone', timeout=25), s.proxy.log deadline = time.time() + 15 - while time.time() < deadline and _ffmpeg_children(s.proxy): + while time.time() < deadline and any(os.path.exists('/proc/%d' % pid) for pid in backends): time.sleep(0.5) - assert not _ffmpeg_children(s.proxy), \ - 'backend outlived the publisher: %r' % _ffmpeg_children(s.proxy) + assert not any(os.path.exists('/proc/%d' % pid) for pid in backends), \ + 'backend outlived the publisher: %r' % backends def test_backend_dies_with_the_proxy(self, session, clip): s = session() s.publish(clip) assert s.proxy.wait_for(r'RTSP backend pid \d+', timeout=20), s.proxy.log - assert _ffmpeg_children(s.proxy) + backends = _ffmpeg_children(s.proxy) + assert backends s.stop_publisher() s.proxy.stop() deadline = time.time() + 15 - while time.time() < deadline and _ffmpeg_children(s.proxy): + while time.time() < deadline and any(os.path.exists('/proc/%d' % pid) for pid in backends): time.sleep(0.5) - assert not _ffmpeg_children(s.proxy), \ - 'backend outlived the proxy: %r' % _ffmpeg_children(s.proxy) + assert not any(os.path.exists('/proc/%d' % pid) for pid in backends), \ + 'backend outlived the proxy: %r' % backends def _publish_with(url_suffix, clip, seconds=6): @@ -1448,7 +1474,7 @@ def test_no_orphan_backend_after_the_rtmp_publisher_leaves( s = RtspSession(_workdir(tmp_path, rtmp_path='PhoenixFPV/FPV')) try: s.publish_rtmp(clip) - assert s.proxy.wait_for(r'RTMP publisher', timeout=25), s.proxy.log + assert s.proxy.wait_for(r'join=ready', timeout=40), s.proxy.log s.stop_publisher() assert s.proxy.wait_for(r'publisher gone', timeout=30), s.proxy.log finally: