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..2512bd0 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 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 - Supports WebSocket and WebSocket+SSL TCP connections for both user @@ -18,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) @@ -157,8 +165,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 +315,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 +400,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..a23652e 160000 --- a/modules/mavlink +++ b/modules/mavlink @@ -1 +1 @@ -Subproject commit 7bdb2dc7d679629eb80e7f28973334ca860f11c9 +Subproject commit a23652e3c7d1d7a94b1c2d354a13ce4858cd0996 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..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 @@ -31,7 +31,17 @@ 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_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', 'tests/test_drop_lost_request.py', 'tests/test_websocket_decode.py', @@ -157,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', @@ -175,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)') @@ -186,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) @@ -206,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) @@ -230,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/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_binlog_capture.py b/tests/test_binlog_capture.py index a5ebede..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) @@ -87,7 +89,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,20 +105,37 @@ 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: ' + ''.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 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_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_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_sysid32.py b/tests/test_sysid32.py new file mode 100644 index 0000000..80ff822 --- /dev/null +++ b/tests/test_sysid32.py @@ -0,0 +1,201 @@ +"""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: + 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) + 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/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) 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: 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(