diff --git a/Makefile b/Makefile index e69dfa7..36e1624 100644 --- a/Makefile +++ b/Makefile @@ -73,14 +73,14 @@ mavlink.o: mavlink.cpp mavlink.h $(MAVLINK_DIR)/protocol.h # Dependencies. mavlink.h includes keydb.h, so any object that pulls in # mavlink.h transitively depends on keydb.h too. -supportproxy.o: supportproxy.cpp mavlink.h util.h keydb.h conntdb.h tlog.h session.h cleanup.h websocket.h +supportproxy.o: supportproxy.cpp mavlink.h util.h keydb.h conntdb.h tlog.h binlog.h session.h cleanup.h websocket.h mavlink.o: mavlink.cpp mavlink.h keydb.h $(MAVLINK_DIR)/protocol.h util.o: util.cpp util.h keydb.o: keydb.cpp keydb.h conntdb.o: conntdb.cpp conntdb.h tlog.o: tlog.cpp tlog.h session.h session.o: session.cpp session.h -binlog.o: binlog.cpp binlog.h session.h mavlink.h util.h $(MAVLINK_DIR)/protocol.h +binlog.o: binlog.cpp binlog.h session.h mavlink.h util.h cleanup.h $(MAVLINK_DIR)/protocol.h cleanup.o: cleanup.cpp cleanup.h keydb.h websocket.o: websocket.cpp websocket.h util.h diff --git a/binlog.cpp b/binlog.cpp index 4e41bc2..59dd255 100644 --- a/binlog.cpp +++ b/binlog.cpp @@ -5,6 +5,7 @@ #include "session.h" #include "mavlink.h" #include "util.h" +#include "cleanup.h" #include "libraries/mavlink2/generated/all/mavlink.h" @@ -139,7 +140,9 @@ void BinlogWriter::refresh_other_sessions_bytes() continue; } } - other_sessions_bytes_ += st.st_size; + // allocated size, not apparent: sparse .bin files would + // otherwise overstate their disk cost and starve the quota + other_sessions_bytes_ += off_t(st.st_blocks) * 512; } closedir(dd); } @@ -184,6 +187,19 @@ void BinlogWriter::handle_block(uint32_t port2, unsigned session_n, mavlink_remote_log_data_block_t blk {}; mavlink_msg_remote_log_data_block_decode(&msg, &blk); + // Vehicle-side log restart detected from the data itself: the + // vehicle rebooted (or its client-timeout fired) and began a new + // log from block 0 before we saw the SYSTEM_TIME backward jump. + // Without this, block 0 overwrites the head of the old session + // file, and when SYSTEM_TIME finally shows the jump the rotation + // waits for a fresh seqno=0 that never comes. + if (fp != nullptr && blk.seqno == 0 + && highest_seen >= SEQNO0_RESTART_MIN_HIGHEST) { + ::printf("binlog: seqno=0 with highest_seen=%u — vehicle log " + "restart, rotating\n", unsigned(highest_seen)); + rotate_for_reboot(); + } + // Strict-start gate. Without this we sparse-extend the file out // to whatever the vehicle's current seqno is — a vehicle that // was already streaming when SupportProxy activated will have @@ -196,6 +212,17 @@ void BinlogWriter::handle_block(uint32_t port2, unsigned session_n, // vehicle's new boot sends seqno=0 to start the new file. if (fp == nullptr) { if (blk.seqno != 0) { + // vehicle is streaming mid-log at a closed file: only a + // restart from seqno 0 can unblock it. tick() sends STOP + // to trigger that promptly. Latch the sender's identity + // so the STOP is targeted, not broadcast. + if (target_system == 0) { + target_system = msg.sysid; + target_component = msg.compid; + } + if (gated_block_count_ < STOP_MIN_GATED_BLOCKS) { + gated_block_count_++; + } return; } unsigned n = (pending_session_n_ != 0) ? pending_session_n_ : session_n; @@ -203,6 +230,7 @@ void BinlogWriter::handle_block(uint32_t port2, unsigned session_n, return; } pending_session_n_ = 0; + gated_block_count_ = 0; } // Caps to limit damage from a malicious or buggy peer sending a @@ -223,13 +251,52 @@ void BinlogWriter::handle_block(uint32_t port2, unsigned session_n, (long long)MAX_FORWARD_JUMP_BYTES); return; } - if (other_sessions_bytes_ + prospective_size > MAX_PER_PORT2_BYTES) { + const off_t quota = port2_quota_bytes(); + // Charge our own file by its allocated size, matching how every + // other file is counted: a legitimate forward jump can grow the + // sparse logical extent by up to MAX_FORWARD_JUMP_BYTES while + // allocating almost nothing, and charging the logical size would + // falsely trip the quota (and the cleanup pass, which sees only + // allocated bytes, would rightly refuse to free anything). + // Project growth as one filesystem block, not BLOCK_BYTES: a + // 200-byte write into an unallocated region allocates a whole + // block, and projecting less would let the total overshoot the + // quota by the difference. + off_t own_alloc = 0; + off_t write_growth = off_t(BLOCK_BYTES); + { + struct stat fst; + if (fstat(fileno(fp), &fst) == 0) { + own_alloc = off_t(fst.st_blocks) * 512; + if (off_t(fst.st_blksize) > write_growth) { + write_growth = off_t(fst.st_blksize); + } + } + } + const off_t projected = other_sessions_bytes_ + own_alloc + write_growth; + if (projected > quota) { + // Try to free space now rather than dropping every block until + // the hourly cleanup pass: age out the oldest sessions for this + // port2 and re-baseline. Rate-limited so a dir that genuinely + // can't get under quota isn't rescanned per block. + double cnow = time_seconds(); + if (cnow - last_quota_cleanup_s_ >= QUOTA_CLEANUP_MIN_INTERVAL_S) { + last_quota_cleanup_s_ = cnow; + // the pass counts our own on-disk allocation itself; the + // extra headroom we need beyond that is this write + log_cleanup_port2_quota(port2_, base_dir_.c_str(), + write_growth); + refresh_other_sessions_bytes(); + } + } + if (other_sessions_bytes_ + own_alloc + write_growth > quota) { ::printf("binlog: dropping seqno=%u (port2=%u total would be " "%lld > %lld byte quota; cleanup pass will age out " "old sessions)\n", unsigned(blk.seqno), unsigned(port2_), - (long long)(other_sessions_bytes_ + prospective_size), - (long long)MAX_PER_PORT2_BYTES); + (long long)(other_sessions_bytes_ + own_alloc + + write_growth), + (long long)quota); return; } @@ -363,6 +430,23 @@ void BinlogWriter::tick(MAVLink &user_link) { double now_s = time_seconds(); + // A vehicle streaming mid-log with no file open (fresh child + // attached mid-flight, or post-rotation) can't make progress until + // it restarts from seqno 0; left alone, only its 10 s no-ACK + // client timeout gets it there. Send STOP so it stops now — the + // START logic below then restarts it from 0 within a second or so + // (ArduPilot ignores STARTs while streaming, but honours STOP). + // The threshold keeps a lone stale block from stopping a healthy + // stream that is about to deliver its seqno 0. + if (fp == nullptr && gated_block_count_ >= STOP_MIN_GATED_BLOCKS + && now_s - last_stop_sent_s >= STOP_REPEAT_S) { + if (send_stop_packet(user_link)) { + last_stop_sent_s = now_s; + // make the follow-up START prompt in both START loops + last_start_sent_s = 0.0; + } + } + if (!any_block_seen) { // Vehicle hasn't begun streaming yet. Send the magic // REMOTE_LOG_BLOCK_STATUS(status=ACK, @@ -436,7 +520,7 @@ void BinlogWriter::tick(MAVLink &user_link) } } -bool BinlogWriter::send_start_packet(MAVLink &user_link) +bool BinlogWriter::send_magic_packet(MAVLink &user_link, uint32_t magic_seqno) { // See the long comment in send_status() about why we can't use // user_link.send_message() — the pack_chan call finalises the @@ -448,7 +532,7 @@ bool BinlogWriter::send_start_packet(MAVLink &user_link) PROXY_SYSID, PROXY_COMPID, CHAN_COMM1, &msg, /*target_system*/ target_system, /*target_component*/ target_component, - /*seqno*/ MAV_REMOTE_LOG_DATA_BLOCK_START, + /*seqno*/ magic_seqno, /*status*/ MAV_REMOTE_LOG_DATA_BLOCK_ACK); uint8_t buf[MAVLINK_MAX_PACKET_LEN]; uint16_t len = mavlink_msg_to_send_buffer(buf, &msg); @@ -458,6 +542,16 @@ bool BinlogWriter::send_start_packet(MAVLink &user_link) return user_link.send_buf(buf, len) == ssize_t(len); } +bool BinlogWriter::send_start_packet(MAVLink &user_link) +{ + return send_magic_packet(user_link, MAV_REMOTE_LOG_DATA_BLOCK_START); +} + +bool BinlogWriter::send_stop_packet(MAVLink &user_link) +{ + return send_magic_packet(user_link, MAV_REMOTE_LOG_DATA_BLOCK_STOP); +} + void BinlogWriter::observe(const mavlink_message_t &msg) { if (msg.msgid != MAVLINK_MSG_ID_SYSTEM_TIME) { @@ -498,6 +592,10 @@ bool BinlogWriter::rotate_for_reboot() // (whose _sending_to_client is now false post-reboot) resumes // streaming without waiting out the keep-alive interval. last_start_sent_s = 0.0; + // Stale pre-rotation blocks must re-accumulate before a STOP nudge + // fires: the restarted stream's seqno 0 is usually already on the + // way and must not be interrupted by a lone straggler. + gated_block_count_ = 0; // Re-arm the first-message-seen guard so the new boot's first // SYSTEM_TIME becomes the new watermark, not a spurious second // trigger of the backward-jump check. diff --git a/binlog.h b/binlog.h index a5543f8..981b873 100644 --- a/binlog.h +++ b/binlog.h @@ -147,6 +147,19 @@ class BinlogWriter { // How often to re-send START until the vehicle starts streaming. // ArduPilot's client-timeout is 10 s so 1 Hz is comfortable. static constexpr double START_REPEAT_S = 1.0; + // Throttle for the STOP nudge sent while a vehicle streams mid-log + // blocks at us with no file open (see gated_block_count_). + static constexpr double STOP_REPEAT_S = 2.0; + // Gated blocks required before the first STOP: one or two stale + // packets (e.g. delayed pre-reboot blocks right after a rotation) + // must not stop a healthy stream that is about to deliver its + // seqno 0; a genuinely mid-log stream reaches this in well under + // a second. Residual: ArduPilot has a single remote-log stream + // with no ownership check, so if another collector (e.g. a local + // MAVProxy dataflash_logger) is also attached to the vehicle, our + // STOP restarts its stream too — inherent to the protocol, and + // two collectors on one vehicle fight over ACKs regardless. + static constexpr unsigned STOP_MIN_GATED_BLOCKS = 3; // Keep-alive START cadence after streaming has begun: defence in // depth so a post-reboot vehicle (whose _sending_to_client got // cleared) resumes streaming within ~5 s of the next keepalive. @@ -158,6 +171,11 @@ class BinlogWriter { // wobble and well below the multi-second pause a real reboot // creates. static constexpr uint32_t REBOOT_TIME_BACKWARD_MS = 10000; + // A block-0 arriving while the file is open with highest_seen at + // least this far along is a vehicle-side log restart, not a + // retransmission: AP_Logger_MAVLink's resend window is ~32 blocks, + // so a legitimate block-0 retry can never arrive this late. + static constexpr uint32_t SEQNO0_RESTART_MIN_HIGHEST = 200; // Caps to limit the damage from an attacker (or a buggy vehicle) // sending a giant seqno on the unsigned-by-default user-side port. // A bare seqno=0 followed by seqno=2^32-1 would otherwise sparse- @@ -170,10 +188,11 @@ class BinlogWriter { // in a single seqno step. Covers ~30 minutes of streaming at // 400 blocks/s; anything bigger is unambiguously bogus. static constexpr off_t MAX_FORWARD_JUMP_BYTES = off_t(100) * 1024 * 1024; - // 2. Per-port-pair on-disk quota: total size of all .tlog + .bin - // files under logs// may not exceed 1 GiB. The hourly - // cleanup loop also enforces this by deleting oldest files. - static constexpr off_t MAX_PER_PORT2_BYTES = off_t(1024) * 1024 * 1024; + // 2. Per-port-pair on-disk quota: total allocated size of all + // .tlog + .bin files under logs// may not exceed + // port2_quota_bytes() (cleanup.h; default 1 GiB, env- + // overridable). The hourly cleanup loop enforces the same + // quota by deleting oldest files. FILE *fp = nullptr; @@ -212,9 +231,20 @@ class BinlogWriter { double now_s); void drop_stale_nack_state(double now_s); bool send_status(MAVLink &user_link, uint32_t seqno, uint8_t status); - // The magic START packet emit, factored out so both the pre-stream - // 1 Hz loop and the streaming-mode keepalive call it. + // The magic START/STOP packet emit, shared by the pre-stream 1 Hz + // loop, the streaming-mode keepalive and the mid-log STOP nudge. + bool send_magic_packet(MAVLink &user_link, uint32_t magic_seqno); bool send_start_packet(MAVLink &user_link); + bool send_stop_packet(MAVLink &user_link); + + // Count of DATA_BLOCKs rejected by the strict-start gate (no file + // open, seqno != 0) since the last open/rotation: the vehicle is + // streaming mid-log and can only recover by restarting from + // seqno 0. Once it passes STOP_MIN_GATED_BLOCKS, tick() sends + // STOP so that happens now instead of after the vehicle's 10 s + // no-ACK client timeout. Saturating. + unsigned gated_block_count_ = 0; + double last_stop_sent_s = 0.0; // Captured on the first successful open() so rotate_for_reboot() // can re-scan the per-day dir for a fresh session N without @@ -233,6 +263,12 @@ class BinlogWriter { off_t other_sessions_bytes_ = 0; unsigned writes_since_quota_refresh_ = 0; void refresh_other_sessions_bytes(); + // Rate limit for the write-time quota-breach cleanup: when the + // gate trips we run the per-port2 quota pass immediately (instead + // of dropping blocks until the hourly pass), but at most once per + // this interval so a genuinely full dir isn't rescanned per block. + static constexpr double QUOTA_CLEANUP_MIN_INTERVAL_S = 30.0; + double last_quota_cleanup_s_ = 0.0; // Per-entry MAVLink sysid filter for SYSTEM_TIME-based reboot // detection. 0 = match any (default). Set from KeyEntry.fc_sysid diff --git a/cleanup.cpp b/cleanup.cpp index 0e7408a..9ad73a5 100644 --- a/cleanup.cpp +++ b/cleanup.cpp @@ -5,6 +5,7 @@ #include "keydb.h" #include +#include #include #include #include @@ -19,13 +20,33 @@ #include #include -namespace { +off_t port2_quota_bytes(void) +{ + static off_t cached = -1; + if (cached >= 0) { + return cached; + } + cached = off_t(1024) * 1024 * 1024; // 1 GiB default + const char *env = getenv("SUPPORTPROXY_PORT2_QUOTA_BYTES"); + if (env != nullptr && *env != '\0') { + // strict: plain positive bytes only. A prefix parse would turn + // a well-meant "1GB" into a 1-byte quota and let the cleanup + // pass delete nearly the whole log tree. + char *endp = nullptr; + errno = 0; + long long v = strtoll(env, &endp, 10); + if (errno == 0 && endp != env && *endp == '\0' && v > 0) { + cached = off_t(v); + } else { + ::printf("ignoring invalid SUPPORTPROXY_PORT2_QUOTA_BYTES " + "'%s' (want plain bytes); using %lld\n", + env, (long long)cached); + } + } + return cached; +} -// Per-port-pair on-disk quota: matches BinlogWriter::MAX_PER_PORT2_BYTES -// in binlog.h. Kept in sync there. Total .tlog + .bin under logs// -// across all date dirs may not exceed this; oldest files are deleted -// first. -constexpr off_t MAX_PER_PORT2_BYTES = off_t(1024) * 1024 * 1024; // 1 GiB +namespace { struct PassCtx { const char *base_dir; @@ -53,7 +74,22 @@ static bool is_session_file(const char *name) about to overflow. Walks every date dir under logs//, sorts files by mtime ascending, deletes from the head. */ -static void enforce_port2_quota(uint32_t port2, const char *base_dir) +// Files whose mtime is within this window are treated as belonging to +// a live session and are never deleted by the quota pass: on Linux the +// unlink would succeed while the writer keeps appending to an +// invisible unlinked inode — the log is lost on close and the disk +// usage stops being counted. +// +// mtime is a heuristic, not ownership: the hourly pass runs in the +// cleanup child and cannot know which files other children hold open. +// An open file idle for longer than the grace (a stalled stream) can +// still be unlinked, and a just-closed session is protected slightly +// longer than needed. Both are acceptable: a healthy binlog/tlog +// writes many times per second. +static constexpr time_t ACTIVE_FILE_GRACE_S = 60; + +static void enforce_port2_quota(uint32_t port2, const char *base_dir, + off_t needed = 0) { char port_dir[768]; snprintf(port_dir, sizeof(port_dir), "%s/%u", base_dir, port2); @@ -97,29 +133,45 @@ static void enforce_port2_quota(uint32_t port2, const char *base_dir) if (stat(fpath, &fst) != 0) { continue; } - items.push_back({fpath, fst.st_size, fst.st_mtime, date_dir}); - total += fst.st_size; + // allocated size, not apparent: .bin files are sparse and + // st_size wildly overstates what they cost on disk + const off_t alloc = off_t(fst.st_blocks) * 512; + total += alloc; + if (time(nullptr) - fst.st_mtime < ACTIVE_FILE_GRACE_S) { + // live session file: count it, never delete it + continue; + } + items.push_back({fpath, alloc, fst.st_mtime, date_dir}); } closedir(dd); } closedir(d); - if (total <= MAX_PER_PORT2_BYTES) { + // `needed` is the caller's prospective growth: a write-time breach + // can happen with total still at or just under the quota, and + // without accounting for it here the pass would free nothing and + // the caller's write would be dropped forever. + const off_t quota = port2_quota_bytes(); + if (total + needed <= quota) { return; } - // Sort oldest-first and delete until under quota. + // Sort oldest-first and delete down to 80% of quota, not just + // under it: freeing to the brim meant an active session's growth + // re-breached the cap within minutes and blocked binlog writes + // until the next hourly pass. + const off_t target = quota - quota / 5; std::sort(items.begin(), items.end(), [](const Item &a, const Item &b) { return a.mtime < b.mtime; }); for (const auto &it : items) { - if (total <= MAX_PER_PORT2_BYTES) { + if (total + needed <= target) { break; } if (unlink(it.path.c_str()) == 0) { ::printf("log cleanup: removed %s for quota " "(port2=%u total %lld > %lld)\n", it.path.c_str(), unsigned(port2), - (long long)total, (long long)MAX_PER_PORT2_BYTES); + (long long)total, (long long)quota); total -= it.size; // Try rmdir on the date dir in case this was its last file; // harmless if it isn't. @@ -257,6 +309,12 @@ static void sleep_seconds(double s) } // namespace +void log_cleanup_port2_quota(unsigned port2, const char *base_dir, + off_t needed) +{ + enforce_port2_quota(port2, base_dir, needed); +} + void log_cleanup_once(const char *base_dir) { auto *db = db_open(); diff --git a/cleanup.h b/cleanup.h index 6eef4ef..f37b662 100644 --- a/cleanup.h +++ b/cleanup.h @@ -3,6 +3,16 @@ */ #pragma once +#include + +/* + Per-port-pair on-disk quota (bytes) for logs//. Default 1 GiB; + override with SUPPORTPROXY_PORT2_QUOTA_BYTES (integer bytes). Shared + by the cleanup quota pass and binlog's write-time gate so the two + always agree. + */ +off_t port2_quota_bytes(void); + /* Run forever: every SUPPORTPROXY_CLEANUP_INTERVAL seconds (default 3600, env var override accepts a float for tests), traverse keys.tdb and @@ -19,3 +29,14 @@ void log_cleanup_loop(const char *base_dir = "logs"); test suite so it can drive cleanup without the sleep loop. */ void log_cleanup_once(const char *base_dir = "logs"); + +/* + Run just the quota pass for a single port pair, synchronously. + Used by the binlog writer when a write-time quota breach needs + relief now rather than at the next hourly pass. `needed` is extra + headroom (bytes) the caller wants on top of what's on disk: the + pass frees when total+needed exceeds the quota, so a prospective + breach at total == quota still gets relief. + */ +void log_cleanup_port2_quota(unsigned port2, const char *base_dir = "logs", + off_t needed = 0); diff --git a/conntdb.cpp b/conntdb.cpp index 8964e65..0d3a39f 100644 --- a/conntdb.cpp +++ b/conntdb.cpp @@ -145,6 +145,41 @@ static int collect_port2(struct tdb_context *db, TDB_DATA key, return 0; } +struct drop_mask_ctx { + int port2; + uint32_t mask; +}; + +static int collect_drop_mask(struct tdb_context *db, TDB_DATA key, + TDB_DATA data, void *ptr) +{ + (void)db; + auto *c = (struct drop_mask_ctx *)ptr; + if (key.dsize != sizeof(struct ConnKey) || + data.dsize < CONNENTRY_MIN_SIZE) { + return 0; + } + struct ConnKey k {}; + memcpy(&k, key.dptr, sizeof(k)); + if (k.port2 != c->port2 || k.conn_index < 0 || k.conn_index >= 32) { + return 0; + } + struct ConnEntry e {}; + size_t copy = data.dsize < sizeof(e) ? data.dsize : sizeof(e); + memcpy(&e, data.dptr, copy); + if (e.magic == CONN_MAGIC && (e.flags & CONN_FLAG_DROP_REQUESTED) != 0) { + c->mask |= 1u << k.conn_index; + } + return 0; +} + +uint32_t conn_drop_mask(TDB_CONTEXT *db, int port2) +{ + struct drop_mask_ctx c { port2, 0 }; + tdb_traverse(db, collect_drop_mask, &c); + return c.mask; +} + int conn_delete_for_port2(TDB_CONTEXT *db, int port2) { struct port2_filter f { port2, {} }; diff --git a/conntdb.h b/conntdb.h index 710851b..b973121 100644 --- a/conntdb.h +++ b/conntdb.h @@ -82,6 +82,11 @@ bool conn_delete(TDB_CONTEXT *db, int port2, int conn_index); // Returns the number of records removed. int conn_delete_for_port2(TDB_CONTEXT *db, int port2); +// Bitmask of conn_index values (0..31) whose record for this port2 has +// CONN_FLAG_DROP_REQUESTED set. Used by the heartbeat snapshot so its +// delete+rewrite can't wipe a drop request that raced it. +uint32_t conn_drop_mask(TDB_CONTEXT *db, int port2); + // One-shot helpers used by the parent (open + transaction internally). void conn_recreate_empty(void); void conn_remove_port2(int port2); diff --git a/keydb.cpp b/keydb.cpp index 6378124..140876f 100644 --- a/keydb.cpp +++ b/keydb.cpp @@ -1,10 +1,37 @@ #include "keydb.h" #include #include +#include +#include +/* + Open keys.tdb. EBUSY can happen briefly when a concurrent transaction + (keydb.py, the web admin) holds the open lock; retry with a short + backoff like conn_db_open() does so a transient collision doesn't + fail a reload or a signing-key load. + */ TDB_CONTEXT *db_open(void) { - return tdb_open(KEY_FILE, 1000, 0, O_RDWR | O_CREAT, 0600); + static const struct timespec backoffs[] = { + {0, 0}, + {0, 10 * 1000 * 1000}, // 10ms + {0, 50 * 1000 * 1000}, + {0, 100 * 1000 * 1000}, + {0, 250 * 1000 * 1000}, + }; + for (size_t i = 0; i < sizeof(backoffs) / sizeof(backoffs[0]); i++) { + if (i > 0) { + nanosleep(&backoffs[i], nullptr); + } + auto *db = tdb_open(KEY_FILE, 1000, 0, O_RDWR | O_CREAT, 0600); + if (db != nullptr) { + return db; + } + if (errno != EBUSY) { + return nullptr; + } + } + return nullptr; } TDB_CONTEXT *db_open_transaction(void) @@ -13,7 +40,10 @@ TDB_CONTEXT *db_open_transaction(void) if (db == nullptr) { return db; } - tdb_transaction_start(db); + if (tdb_transaction_start(db) != 0) { + tdb_close(db); + return nullptr; + } return db; } diff --git a/mavlink.cpp b/mavlink.cpp index 4c9c033..6f2af8b 100644 --- a/mavlink.cpp +++ b/mavlink.cpp @@ -70,6 +70,10 @@ void MAVLink::init(int _fd, mavlink_channel_t _chan, bool signing_required, bool allow_websocket = _allow_websocket; got_bad_signature[chan] = false; use_sendto = false; + // slot reuse: a previous WebSocket peer's wrapper is deleted by the + // owner on close; without this reset the next (plain) peer on the + // same slot would send through the dangling pointer + ws = nullptr; ZERO_STRUCT(signing_streams); ZERO_STRUCT(signing); @@ -289,6 +293,27 @@ void MAVLink::load_signing_key(void) return; } + // safe-fail on all-zero stored key: leave key_loaded=false so + // receive_message() rejects every frame on this channel rather than + // wiring NULL signing (which the generated parser treats as + // "signature OK") into status->signing. + bool stored_all_zero = (key.timestamp == 0); + for (uint8_t i=0; stored_all_zero && isigning = nullptr; + status_z->signing_streams = nullptr; + } + db_close(db); + return; + } + key_loaded = true; memcpy(signing.secret_key, key.secret_key, sizeof(key.secret_key)); @@ -324,22 +349,10 @@ void MAVLink::load_signing_key(void) signing.flags = MAVLINK_SIGNING_FLAG_SIGN_OUTGOING; signing.accept_unsigned_callback = accept_unsigned_callback; - // if timestamp and key are all zero then we disable signing - bool all_zero = (key.timestamp == 0); - for (uint8_t i=0; isigning = nullptr; - status->signing_streams = nullptr; - } else { - status->signing = &signing; - status->signing_streams = &signing_streams; - } + // the all-zero "disable signing" case was already handled with an + // early return above, so this branch always wires real signing in. + status->signing = &signing; + status->signing_streams = &signing_streams; db_close(db); } @@ -501,6 +514,22 @@ void MAVLink::handle_setup_signing(const mavlink_message_t &msg) mavlink_setup_signing_t packet; mavlink_msg_setup_signing_decode(&msg, &packet); + // refuse rotation to an all-zero key + zero timestamp: that combo is + // exactly what load_signing_key() treats as "signing disabled", so + // accepting it would turn a key-rotation request into a persistent + // auth downgrade. Defence in depth — load_signing_key() also + // safe-fails on this combination if it lands in keys.tdb some other way. + bool all_zero = (packet.initial_timestamp == 0); + for (uint8_t i=0; all_zero && i #include #include +#include #include #include "mavlink.h" @@ -51,6 +52,18 @@ the matching ConnEntry first; the handler just sets a flag and main_loop scans connections.tdb to find the target slot(s). */ +// Engineer-side conn2 slots that haven't validated a signed packet +// within CONN2_PREAUTH_SECONDS of accept/first-tuple are closed. Cuts +// off DoS where unauthenticated clients camp on conn2 slots and block +// legitimate signed engineers. +static constexpr time_t CONN2_PREAUTH_SECONDS = 5; + +// User-side conn1 in bidi-sign mode: TCP/WS latch immediately but if +// no signed user packet validates within this window we break the +// per-pair child so the parent reopens the listener. UDP defers the +// latch until validation, so this guard only fires for TCP/WS. +static constexpr time_t CONN1_BIDI_PREAUTH_SECONDS = 5; + static volatile sig_atomic_t g_drops_pending = 0; static void sigusr1_handler(int) @@ -80,6 +93,13 @@ struct listen_port { static struct listen_port *ports; +// epoll instance used by wait_connection(). handle_connection() must +// deregister a pair's sockets from it before closing them at fork +// time: the child keeps the open file descriptions alive, so close() +// alone leaves live registrations that wake the parent for every +// packet the child handles. +static int g_epfd = -1; + // PID of the long-lived log-cleanup child forked from main() that // ages out old .tlog / .bin files. Tracked separately from // per-port-pair children so check_children() can respawn it if it @@ -216,6 +236,7 @@ class Connection2 { close_fd(sock); tcp_active = false; used = false; + mav.set_ws(nullptr); delete ws; ws = nullptr; connected_at = 0; @@ -261,6 +282,17 @@ static void main_loop(struct listen_port *p) // which writer activates first or whether one of them never does. const unsigned session_n = next_session_n(uint32_t(p->port2), "logs"); + // bidi: initialise the user-side validator once. Re-initialising + // per unsigned datagram (as the pre-latch path originally did) + // opens keys.tdb and forks a timestamp-save child every packet; + // under a stream of unsigned traffic the tdb lock collisions stall + // the main loop long enough to blow the engineer pre-auth window + // and WebSocket handshake timeouts. Parser and signing state carry + // across datagrams safely — validation still gates the latch. + if (bidi && p->sock1_udp != -1) { + mav1.init(p->sock1_udp, CHAN_COMM1, true, false, false, conn1_key_id); + } + // tlog: opened lazily on first received frame so an idle child that // never sees traffic doesn't leave behind an empty session file. TlogWriter tlog; @@ -324,6 +356,21 @@ static void main_loop(struct listen_port *p) fdmax = MAX(fdmax, p->sock1_tcp); fdmax = MAX(fdmax, p->sock2_listen); + // Close an engineer slot and keep the counters consistent. + // max_conn2_count is a scan watermark (highest used slot + 1): it + // may only shrink when the top slot(s) become free. Shrinking it + // because the counts happened to match dropped still-used higher + // slots out of every scan loop, silently freezing those engineers. + auto close_conn2 = [&](Connection2 &c2) { + c2.close(); + if (conn2_count > 0) { + conn2_count--; + } + while (max_conn2_count > 0 && !conn2[max_conn2_count-1].used) { + max_conn2_count--; + } + }; + // Pull DROP_REQUESTED entries for our port2 out of connections.tdb, // close the matching slots, and delete the records. Returns true if // the user side was dropped (caller should exit main_loop). @@ -377,13 +424,7 @@ static void main_loop(struct listen_port *p) if (c2.used) { printf("[%d] %s drop conn2[%d] requested\n", p->port2, time_string(), idx - 1); - c2.close(); - if (conn2_count > 0) { - conn2_count--; - } - if (max_conn2_count > 0 && idx == max_conn2_count) { - max_conn2_count--; - } + close_conn2(c2); } } } @@ -436,23 +477,60 @@ static void main_loop(struct listen_port *p) now = time_seconds(); if (max_conn2_count > MAX_COMM2_LINKS) { - printf("BUG: max_conn2_count=%d\n", int(max_conn2_count)); - exit(1); + // formerly exit(1). With the UDP idle-close path now decrementing + // conn2_count properly this should not be reachable, but if some + // other path leaks it we'd rather log loudly and clamp than kill + // the child for the whole port pair. + printf("BUG: max_conn2_count=%d, clamping to %d\n", + int(max_conn2_count), int(MAX_COMM2_LINKS)); + max_conn2_count = MAX_COMM2_LINKS; } /* - check for dead UDP conn2 + check for dead UDP conn2 + pre-auth deadline on any conn2 slot */ + const time_t wall_now = time(nullptr); for (uint8_t i=0; i 10) { - printf("[%d] %s dead UDP conn2[%u]\n", + if (!c2.used) { + continue; + } + bool close_this = false; + const char *why = ""; + if (!c2.mav.is_authenticated() && + wall_now - c2.connected_at > CONN2_PREAUTH_SECONDS) { + // pre-auth deadline: applies to TCP, WS and UDP. Compare + // against connected_at (not last_pkt) so an attacker can't + // keep the slot alive by spamming unsigned/wrong-key + // traffic that refreshes last_pkt. + close_this = true; + why = "pre-auth deadline"; + } else if (c2.is_udp && now - c2.last_pkt > 10) { + // authenticated UDP: existing idle close + close_this = true; + why = "idle"; + } + if (close_this) { + printf("[%d] %s closing %s conn2[%u] (%s)\n", unsigned(p->port2), time_string(), - unsigned(i)); - c2.close(); + c2.is_udp ? "UDP" : "TCP", + unsigned(i), why); + close_conn2(c2); } } + // bidi user-side pre-auth deadline: TCP/WS latches before the + // first signed packet validates. If we don't see a valid signed + // packet in time, exit the child so the parent reopens the + // listener for the legitimate signed user. (UDP defers the latch + // until validation, so it never enters this state.) + if (bidi && have_conn1 && !mav1.is_authenticated() && mav1_is_tcp && + wall_now - mav1_connected_at > CONN1_BIDI_PREAUTH_SECONDS) { + printf("[%d] %s bidi conn1 pre-auth timeout — restarting child\n", + unsigned(p->port2), time_string()); + break; + } + /* check for UDP user data */ @@ -467,18 +545,50 @@ static void main_loop(struct listen_port *p) last_pkt1 = now; count1++; if (!have_conn1) { - if (connect(p->sock1_udp, (struct sockaddr *)&from, fromlen) != 0) { - break; + if (bidi) { + // bidi pre-auth: validate the signature *before* + // committing the listener to this tuple. Unsigned + // or wrong-key senders cannot latch conn1 and + // deny the legitimate signed user. mav1 was + // initialised once at child start. + uint8_t *vbuf = buf; + ssize_t vn = n; + mavlink_message_t vmsg{}; + bool validated = false; + while (vn > 0 && mav1.receive_message(vbuf, vn, vmsg)) { + validated = true; + } + if (!validated) { + continue; // drop, leave listener open + } + if (connect(p->sock1_udp, (struct sockaddr *)&from, fromlen) != 0) { + break; + } + have_conn1 = true; + mav1_peer = from; + mav1_connected_at = time(nullptr); + mav1_is_tcp = false; + last_conn_save_s = 0; + printf("[%d] %s have UDP conn1 (bidi-validated) from %s\n", + unsigned(p->port2), time_string(), addr_to_str(from)); + // bytes were already consumed during validation; + // skip the main parse path below for this datagram + n = 0; + } else { + if (connect(p->sock1_udp, (struct sockaddr *)&from, fromlen) != 0) { + break; + } + mav1.init(p->sock1_udp, CHAN_COMM1, bidi, false, false, conn1_key_id); + have_conn1 = true; + mav1_peer = from; + mav1_connected_at = time(nullptr); + mav1_is_tcp = false; + // trigger an immediate connections.tdb snapshot on + // the next loop iteration so the web UI sees the + // new conn quickly + last_conn_save_s = 0; + printf("[%d] %s have UDP conn1 for from %s\n", unsigned(p->port2), time_string(), addr_to_str(from)); } - mav1.init(p->sock1_udp, CHAN_COMM1, bidi, false, false, conn1_key_id); - have_conn1 = true; - mav1_peer = from; - mav1_connected_at = time(nullptr); - mav1_is_tcp = false; - // trigger an immediate connections.tdb snapshot on the next - // loop iteration so the web UI sees the new conn quickly - last_conn_save_s = 0; - printf("[%d] %s have UDP conn1 for from %s\n", unsigned(p->port2), time_string(), addr_to_str(from)); } mavlink_message_t msg {}; // Parse user-side bytes whenever there's anywhere for them to @@ -501,11 +611,7 @@ static void main_loop(struct listen_port *p) } if (!c2.is_udp && c2.sock != -1) { if (!c2.mav.send_message(msg)) { - c2.close(); - if (conn2_count == max_conn2_count) { - max_conn2_count--; - } - conn2_count--; + close_conn2(c2); } else { c2.tx_msgs++; } @@ -577,7 +683,13 @@ static void main_loop(struct listen_port *p) if (idx != -1) { mavlink_message_t msg {}; - if (have_conn1) { + // count1>0 means we've processed at least one user-side + // event, so conn1's transport is decided (raw vs + // WebSocket). Forwarding before that can write plain + // MAVLink onto a freshly-accepted TCP socket that then + // turns out to be WebSocket — landing ahead of the HTTP + // 101 and corrupting the handshake. + if (have_conn1 && count1 > 0) { uint8_t *buf0 = buf; bool failed = false; auto &c2 = conn2[idx]; @@ -634,14 +746,29 @@ static void main_loop(struct listen_port *p) FD_ISSET(p->sock1_tcp, &fds)) { close_fd(p->sock1_udp); - if (count1 == 0 && WebSocket::detect(p->sock1_tcp)) { - p->ws = new WebSocket(p->sock1_tcp); - if (p->ws == nullptr) { - break; + if (count1 == 0 && !p->ws) { + ws_detect_t d = WebSocket::detect(p->sock1_tcp); + if (d == WS_MORE) { + // fragmented handshake prefix: wait for more bytes + // rather than committing to raw. Committing early + // would leave mav1 in raw mode and forward raw + // MAVLink onto a socket that is actually WebSocket, + // corrupting the handshake. Brief sleep bounds CPU + // while the rest of the request arrives; the conn1 + // idle timeout still applies. + struct timespec ts { 0, 2 * 1000 * 1000 }; + nanosleep(&ts, nullptr); + continue; + } + if (d == WS_YES) { + p->ws = new WebSocket(p->sock1_tcp); + if (p->ws == nullptr) { + break; + } + mav1.set_ws(p->ws); + printf("[%d] %s WebSocket%s conn1\n", unsigned(p->port2), time_string(), + p->ws->is_SSL()?" SSL":""); } - mav1.set_ws(p->ws); - printf("[%d] %s WebSocket%s conn1\n", unsigned(p->port2), time_string(), - p->ws->is_SSL()?" SSL":""); } ssize_t n; if (p->ws) { @@ -675,11 +802,7 @@ static void main_loop(struct listen_port *p) continue; } if (!c2.mav.send_message(msg)) { - c2.close(); - if (conn2_count == max_conn2_count) { - max_conn2_count--; - } - conn2_count--; + close_conn2(c2); } else { c2.tx_msgs++; } @@ -746,13 +869,23 @@ static void main_loop(struct listen_port *p) continue; } if (FD_ISSET(c2.sock, &fds)) { - if (!c2.tcp_active && WebSocket::detect(c2.sock)) { - c2.ws = new WebSocket(c2.sock); - if (c2.ws == nullptr) { - break; + if (!c2.tcp_active && !c2.ws) { + ws_detect_t d = WebSocket::detect(c2.sock); + if (d == WS_MORE) { + // fragmented handshake prefix; wait for the rest + // rather than misclassifying it as raw MAVLink + struct timespec ts { 0, 2 * 1000 * 1000 }; + nanosleep(&ts, nullptr); + continue; + } + if (d == WS_YES) { + c2.ws = new WebSocket(c2.sock); + if (c2.ws == nullptr) { + break; + } + c2.mav.set_ws(c2.ws); + printf("[%d] %s WebSocket%s conn2\n", unsigned(p->port2), time_string(), c2.ws->is_SSL()?" SSL":""); } - c2.mav.set_ws(c2.ws); - printf("[%d] %s WebSocket%s conn2\n", unsigned(p->port2), time_string(), c2.ws->is_SSL()?" SSL":""); } ssize_t n; if (c2.ws) { @@ -763,9 +896,7 @@ static void main_loop(struct listen_port *p) if (c2.ws) { if (n < 0) { printf("[%d] %s EOF TCP conn2[%u]\n", unsigned(p->port2), time_string(), unsigned(i+1)); - c2.close(); - if (conn2_count == max_conn2_count) { max_conn2_count--; } - conn2_count--; + close_conn2(c2); continue; } if (n == 0) { @@ -775,9 +906,7 @@ static void main_loop(struct listen_port *p) } else { if (n <= 0) { printf("[%d] %s EOF TCP conn2[%u]\n", unsigned(p->port2), time_string(), unsigned(i+1)); - c2.close(); - if (conn2_count == max_conn2_count) { max_conn2_count--; } - conn2_count--; + close_conn2(c2); continue; } } @@ -785,7 +914,9 @@ static void main_loop(struct listen_port *p) count2++; c2.tcp_active = true; mavlink_message_t msg {}; - if (have_conn1) { + // see the note at the UDP-engineer forward: don't forward + // to conn1 until its transport is decided (count1>0) + if (have_conn1 && count1 > 0) { uint8_t *buf0 = buf; bool failed = false; while (n > 0 && c2.mav.receive_message(buf0, n, msg)) { @@ -826,10 +957,19 @@ static void main_loop(struct listen_port *p) double snap_now = time_seconds(); if (snap_now - last_conn_save_s > 5) { last_conn_save_s = snap_now; + // Safety net: rescan for webadmin drop requests on the + // snapshot cadence. A request whose SIGUSR1 raced a + // snapshot rewrite would otherwise be lost for good. + if (process_drops()) { + break; + } signal(SIGCHLD, SIG_IGN); if (fork() == 0) { auto *db = conn_db_open_transaction(); if (db != nullptr) { + // drop requests that landed since we forked must + // survive the delete+rewrite below + const uint32_t drop_mask = conn_drop_mask(db, p->port2); conn_delete_for_port2(db, p->port2); time_t now_t = time(nullptr); if (have_conn1) { @@ -850,6 +990,9 @@ static void main_loop(struct listen_port *p) e.transport = mav1_is_tcp ? CONN_TRANSPORT_TCP : CONN_TRANSPORT_UDP; } e.is_user = 1; + if (drop_mask & 1u) { + e.flags |= CONN_FLAG_DROP_REQUESTED; + } conn_write(db, e); } for (uint8_t i = 0; i < max_conn2_count; i++) { @@ -876,6 +1019,9 @@ static void main_loop(struct listen_port *p) e.transport = CONN_TRANSPORT_TCP; } e.is_user = 0; + if (drop_mask & (1u << (i + 1))) { + e.flags |= CONN_FLAG_DROP_REQUESTED; + } conn_write(db, e); } conn_db_close_commit(db); @@ -960,11 +1106,14 @@ static void open_sockets(struct listen_port *p) } /* - check for child exit + check for child exit. Returns true if a per-port-pair child was + reaped (the caller should refresh the epoll set so the reopened + listeners are watched again). */ -static void check_children(void) +static bool check_children(void) { int wstatus = 0; + bool reaped = false; while (true) { pid_t pid = waitpid(-1, &wstatus, WNOHANG); if (pid <= 0) { @@ -984,6 +1133,7 @@ static void check_children(void) // drop any live-connection records the child wrote conn_remove_port2(p->port2); found_child = true; + reaped = true; // Don't reopen listening sockets for an entry that was // removed from keys.tdb between fork and exit; that would // rebind the port for a record that no longer exists. @@ -997,6 +1147,7 @@ static void check_children(void) printf("No child for %d found\n", int(pid)); } } + return reaped; } /* @@ -1008,6 +1159,16 @@ static void fork_cleanup_child(void) { pid_t pid = fork(); if (pid == 0) { + // die with the parent: this loop never exits on its own, so + // without this every proxy shutdown leaked an orphan process + prctl(PR_SET_PDEATHSIG, SIGTERM); + if (getppid() == 1) { + _exit(0); // parent already gone before prctl took effect + } + if (g_epfd != -1) { + close(g_epfd); + g_epfd = -1; + } for (auto *p = ports; p; p = p->next) { close_sockets(p); } @@ -1022,13 +1183,44 @@ static void fork_cleanup_child(void) printf("log cleanup child %d started\n", int(pid)); } +/* + deregister a pair's listening sockets from the parent's epoll set + */ +static void epoll_del_sockets(struct listen_port *p) +{ + if (g_epfd == -1) { + return; + } + const int fds[4] = { p->sock1_udp, p->sock2_udp, + p->sock1_tcp, p->sock2_listen }; + for (int fd : fds) { + if (fd != -1) { + epoll_ctl(g_epfd, EPOLL_CTL_DEL, fd, nullptr); + } + } +} + /* handle a new connection */ static void handle_connection(struct listen_port *p) { pid_t pid = fork(); + if (pid < 0) { + // keep the sockets open and registered; the pending event fires + // again and we retry. Storing -1 in p->pid would kill the pair + // for good (nothing ever reaps pid -1) and a later + // kill(p->pid, ...) would signal every process on the system. + printf("[%d] fork failed - %s\n", p->port2, strerror(errno)); + return; + } if (pid == 0) { + // the epoll instance is the parent's; drop our copy so a + // parent-side close/recreate doesn't leave it pinned here + if (g_epfd != -1) { + close(g_epfd); + g_epfd = -1; + } for (auto *p2 = ports; p2; p2=p2->next) { if (p2 != p) { close_sockets(p2); @@ -1040,6 +1232,9 @@ static void handle_connection(struct listen_port *p) p->pid = pid; printf("[%d] New child %d\n", p->port2, int(p->pid)); + // deregister before close: fork gave the child references to these + // descriptions, so close() alone leaves the registrations live + epoll_del_sockets(p); close_sockets(p); } @@ -1057,8 +1252,10 @@ static void reload_ports(void) // even if keydb.py / the web admin UI is mutating in parallel auto *db = db_open_transaction(); if (db == nullptr) { - printf("Database not found\n"); - exit(1); + // transient open/lock failure: skip this cycle rather than + // killing the proxy (and every active session) at runtime + printf("reload: failed to open %s - %s\n", KEY_FILE, strerror(errno)); + return; } tdb_traverse(db, handle_record, nullptr); db_close_cancel(db); @@ -1097,6 +1294,7 @@ static void wait_connection(void) perror("epoll_create1"); exit(1); } + g_epfd = epfd; /* rebuild epoll structure for current list of connections @@ -1143,19 +1341,6 @@ static void wait_connection(void) break; } - if (ret == 0) { - check_children(); - double now = time_seconds(); - if (now - last_reload > 5) { - last_reload = now; - reload_ports(); - close(epfd); - epfd = epoll_create1(0); - rebuild_epoll_set(); - } - continue; - } - for (int i = 0; i < ret; i++) { int fd = events[i].data.fd; @@ -1168,6 +1353,27 @@ static void wait_connection(void) } } } + + /* + Housekeeping must run on every iteration, not only when + epoll_wait times out. Children inherit the listening sockets, + so their traffic can keep waking us via registrations that + survived our close() after fork; gating on ret==0 starved + child reaping and DB reloads whenever any session was busy, + leaving killed connections dead until all sessions went idle. + Reaping a child forces an immediate reload so its reopened + listeners get back into the epoll set without the 5s wait. + */ + const bool reaped = check_children(); + double now = time_seconds(); + if (reaped || now - last_reload > 5) { + last_reload = now; + reload_ports(); + close(epfd); + epfd = epoll_create1(0); + g_epfd = epfd; + rebuild_epoll_set(); + } } close(epfd); } @@ -1175,6 +1381,9 @@ static void wait_connection(void) int main(int argc, char *argv[]) { setvbuf(stdout, nullptr, _IOLBF, 4096); + // a peer-closed TCP/WS/SSL connection must fail the write with + // EPIPE, not kill the child (and its whole session) with SIGPIPE + signal(SIGPIPE, SIG_IGN); printf("Opening sockets\n"); // Wipe any connections.tdb records left behind by a previous run. // Per-port-pair children write into this file; on a fresh start no diff --git a/tests/test_binlog_capture.py b/tests/test_binlog_capture.py index 3c38c3b..2713c61 100644 --- a/tests/test_binlog_capture.py +++ b/tests/test_binlog_capture.py @@ -74,9 +74,14 @@ def _setup_db(workdir, port_user, port_eng, name, passphrase, *flags): db.close() -def _start_proxy(workdir, port_eng): +def _start_proxy(workdir, port_eng, quota_bytes=None, cleanup_interval=None): + env = os.environ.copy() + if quota_bytes is not None: + env['SUPPORTPROXY_PORT2_QUOTA_BYTES'] = str(quota_bytes) + if cleanup_interval is not None: + env['SUPPORTPROXY_CLEANUP_INTERVAL'] = str(cleanup_interval) proc = subprocess.Popen( - [SUPPORTPROXY_BIN], cwd=str(workdir), + [SUPPORTPROXY_BIN], cwd=str(workdir), env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, bufsize=1, text=True, ) @@ -111,6 +116,22 @@ def _terminate(proc): proc.wait(timeout=2) if hasattr(proc, '_thread'): proc._thread.join(timeout=2) + # A per-pair child outlives the parent by up to its 10s conn1 idle + # timeout and keeps the fixed test ports bound; if the next test + # starts inside that window its proxy can't bind and every packet + # goes to the stale child. Probe the TCP listen port until it's + # free. + deadline = time.time() + 15 + while time.time() < deadline: + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + try: + s.bind(('127.0.0.1', PORT_ENG)) + s.close() + return + except OSError: + s.close() + time.sleep(0.3) def _bin_path(workdir, port_eng, n=1): @@ -155,6 +176,29 @@ def _send_system_time(sock, dest, time_boot_ms, sysid=1, sock.sendto(buf, dest) +def _recv_block_status_msgs(sock, timeout=2.0): + """Drain incoming UDP packets, return the decoded + REMOTE_LOG_BLOCK_STATUS message objects seen within `timeout`.""" + from pymavlink.dialects.v20 import all as mav + mav_obj = mav.MAVLink(file=None) + sock.settimeout(0.1) + out = [] + deadline = time.time() + timeout + while time.time() < deadline: + try: + data, _ = sock.recvfrom(2048) + except socket.timeout: + continue + try: + msgs = mav_obj.parse_buffer(data) or [] + except mav.MAVError: + continue + for m in msgs: + if m.get_type() == 'REMOTE_LOG_BLOCK_STATUS': + out.append(m) + return out + + def _recv_block_statuses(sock, timeout=2.0): """Drain incoming UDP packets, decode REMOTE_LOG_BLOCK_STATUS, return list of (seqno, status) seen within `timeout`.""" @@ -809,39 +853,180 @@ def test_seqno_jump_above_100mb_rejected(self, proxy_workdir): finally: _terminate(proc) - def test_disk_quota_blocks_writes_when_over_cap(self, proxy_workdir): - """Pre-seed the per-port2 logs// tree with sparse files - totalling > 1 GiB. Subsequent legitimate blocks must be - dropped at write time because the per-port-pair quota - (MAX_PER_PORT2_BYTES = 1 GiB in binlog.h) is already - breached.""" - _setup_db(proxy_workdir, PORT_USER, PORT_ENG, 'bintest', 'bp', - 'binlog') - # Pre-seed BEFORE starting the proxy so the cleanup pass - # doesn't get a chance to age them out. - date_dir = (proxy_workdir / 'logs' / str(PORT_ENG) / _today_str()) + def _seed_prefill_after_startup(self, proxy_workdir, proc, nbytes, + sparse=False, age_seconds=7200): + """Create logs///prefill.bin AFTER the proxy's + startup cleanup pass (which would delete an over-quota file + seeded before it). Waits for the cleanup child's start line; + its startup pass over the still-empty logs tree is + instantaneous. mtime is backdated so quota passes treat it as + the oldest file.""" + deadline = time.time() + 5 + while time.time() < deadline and not any( + 'log cleanup child' in l for l in proc._lines): + time.sleep(0.05) + time.sleep(0.3) + date_dir = (proxy_workdir / 'logs' / str(PORT_ENG) + / _today_str()) date_dir.mkdir(parents=True, exist_ok=True) prefilled = date_dir / 'prefill.bin' - # Sparse: 1.2 GiB apparent size, ~0 bytes actually allocated. with open(prefilled, 'wb') as f: - f.truncate(int(1.2 * 1024 * 1024 * 1024)) + if sparse: + f.truncate(nbytes) + else: + f.write(b'\xee' * nbytes) + now = time.time() + os.utime(prefilled, (now - age_seconds, now - age_seconds)) + return prefilled + + def test_quota_breach_triggers_immediate_cleanup(self, proxy_workdir): + """A write-time quota breach must age out old sessions right + away and let the write proceed — not silently drop blocks for + up to an hour until the next cleanup pass (which is how a + preflight-rebooted aircraft ended up with an empty .bin in the + field).""" + _setup_db(proxy_workdir, PORT_USER, PORT_ENG, 'bintest', 'bp', + 'binlog') - proc = _start_proxy(proxy_workdir, PORT_ENG) + proc = _start_proxy(proxy_workdir, PORT_ENG, quota_bytes=300000) + try: + prefilled = self._seed_prefill_after_startup( + proxy_workdir, proc, 400 * 1024) + + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.bind(('127.0.0.1', 0)) + dest = ('127.0.0.1', PORT_USER) + + # seqno=0 opens session1.bin; the quota is breached by the + # (old) prefill, which the immediate cleanup must delete so + # the write lands. + _send_data_block(sock, dest, 0, b'\xaa' * 50) + + bin_path = _bin_path(proxy_workdir, PORT_ENG) + assert _wait_for( + lambda: bin_path.exists() and bin_path.stat().st_size >= 200, + timeout=5.0), \ + 'write did not proceed after quota breach; proxy log:\n%s' \ + % ''.join(proc._lines[-10:]) + assert not prefilled.exists(), \ + 'old session should have been aged out by immediate cleanup' + + sock.close() + finally: + _terminate(proc) + + def test_quota_boundary_prospective_breach_frees(self, proxy_workdir): + """Old sessions sitting exactly at (or just under) the quota: + current usage doesn't exceed the cap, but the first block's + prospective growth does. The cleanup must account for the + caller's needed headroom and free anyway — otherwise no write + ever succeeds and the stall is permanent.""" + _setup_db(proxy_workdir, PORT_USER, PORT_ENG, 'bintest', 'bp', + 'binlog') + + # prefill = 73 pages = 299008 allocated bytes; quota 299100. + # total (299008) <= quota, but total + 200 > quota. + proc = _start_proxy(proxy_workdir, PORT_ENG, quota_bytes=299100) + try: + prefilled = self._seed_prefill_after_startup( + proxy_workdir, proc, 299008) + + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.bind(('127.0.0.1', 0)) + dest = ('127.0.0.1', PORT_USER) + _send_data_block(sock, dest, 0, b'\xaa' * 50) + + bin_path = _bin_path(proxy_workdir, PORT_ENG) + assert _wait_for( + lambda: bin_path.exists() and bin_path.stat().st_size >= 200, + timeout=5.0), \ + 'boundary breach never freed; proxy log:\n%s' \ + % ''.join(proc._lines[-10:]) + assert not prefilled.exists(), \ + 'old session should have been aged out' + + sock.close() + finally: + _terminate(proc) + + def test_sparse_forward_jump_not_charged_as_logical_size( + self, proxy_workdir): + """A legitimate forward jump makes the active .bin sparse: its + logical size can far exceed its allocated size. The quota gate + must charge the allocated size (like everything else), or the + jump falsely trips the quota — and the cleanup pass, seeing + only allocated bytes under the cap, rightly frees nothing, so + the stall would be permanent.""" + _setup_db(proxy_workdir, PORT_USER, PORT_ENG, 'bintest', 'bp', + 'binlog') + + proc = _start_proxy(proxy_workdir, PORT_ENG, quota_bytes=300000) + try: + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.bind(('127.0.0.1', 0)) + dest = ('127.0.0.1', PORT_USER) + + # seqno 0 opens the file; seqno 1600 makes the logical + # size 320,200 bytes (> 300 KB quota) while allocating + # only ~2 pages. + _send_data_block(sock, dest, 0, b'\xaa' * 50) + _send_data_block(sock, dest, 1600, b'\xbb' * 50) + + bin_path = _bin_path(proxy_workdir, PORT_ENG) + assert _wait_for( + lambda: bin_path.exists() + and bin_path.stat().st_size == 1600 * 200 + 200, + timeout=5.0), \ + 'sparse jump falsely tripped the quota; proxy log:\n%s' \ + % ''.join(proc._lines[-10:]) + + sock.close() + finally: + _terminate(proc) + + def test_quota_pass_never_deletes_active_files(self, proxy_workdir): + """A live session's files (fresh mtime) must never be unlinked + by the quota pass, even when they alone exceed the quota: the + writer would keep appending to an invisible unlinked inode and + the whole log would be lost on close.""" + _setup_db(proxy_workdir, PORT_USER, PORT_ENG, 'bintest', 'bp', + 'binlog') + + # Fast cleanup ticks so the quota pass runs several times + # inside the test window. + proc = _start_proxy(proxy_workdir, PORT_ENG, quota_bytes=100000, + cleanup_interval='0.3') + try: + # 150 KB with a *fresh* mtime — over quota, but "active". + prefilled = self._seed_prefill_after_startup( + proxy_workdir, proc, 150 * 1024, age_seconds=0) + + # Several 0.3s cleanup ticks pass; the file must survive. + time.sleep(2.0) + assert prefilled.exists(), \ + 'quota pass deleted an active session file; log:\n%s' \ + % ''.join(proc._lines[-10:]) + finally: + _terminate(proc) + + def test_quota_still_blocks_when_nothing_to_free(self, proxy_workdir): + """When the quota genuinely can't be met (nothing deletable), + blocks must still be dropped rather than written over quota.""" + _setup_db(proxy_workdir, PORT_USER, PORT_ENG, 'bintest', 'bp', + 'binlog') + + # quota smaller than a single 200-byte block: no amount of + # cleanup can make the write fit. + proc = _start_proxy(proxy_workdir, PORT_ENG, quota_bytes=100) try: sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.bind(('127.0.0.1', 0)) dest = ('127.0.0.1', PORT_USER) - # seqno=0 will pass the strict-start gate and open - # session1.bin, but the quota check should reject the - # actual write — session1.bin should stay empty. _send_data_block(sock, dest, 0, b'\xaa' * 50) time.sleep(1.0) bin_path = _bin_path(proxy_workdir, PORT_ENG) - # session1.bin may or may not exist (open() succeeds; write - # is rejected). Either is acceptable. If it exists, it must - # be 0 bytes. if bin_path.exists(): sz = bin_path.stat().st_size assert sz == 0, \ @@ -852,6 +1037,200 @@ def test_disk_quota_blocks_writes_when_over_cap(self, proxy_workdir): finally: _terminate(proc) + def test_sparse_apparent_size_does_not_consume_quota(self, proxy_workdir): + """The quota counts allocated bytes (st_blocks), not apparent + size: a sparse file with a huge apparent size but no allocated + blocks must not starve the binlog of its quota.""" + _setup_db(proxy_workdir, PORT_USER, PORT_ENG, 'bintest', 'bp', + 'binlog') + + proc = _start_proxy(proxy_workdir, PORT_ENG, quota_bytes=300000) + try: + # 10 MB apparent, ~0 allocated — far over the 300 KB quota + # by apparent size, well under it by allocated size. + self._seed_prefill_after_startup(proxy_workdir, proc, + 10 * 1024 * 1024, + sparse=True) + + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.bind(('127.0.0.1', 0)) + dest = ('127.0.0.1', PORT_USER) + + for seq in range(3): + _send_data_block(sock, dest, seq, b'\xaa' * 50) + + assert _wait_for( + lambda: _bin_path(proxy_workdir, PORT_ENG).exists() + and _bin_path(proxy_workdir, PORT_ENG).stat() + .st_size >= 600, + timeout=5.0), \ + 'writes blocked by sparse apparent size; proxy log:\n%s' \ + % ''.join(proc._lines[-10:]) + + sock.close() + finally: + _terminate(proc) + + def test_instream_seqno0_restart_rotates(self, proxy_workdir): + """A block-0 arriving mid-stream (highest_seen far along) is a + vehicle log restart — the vehicle rebooted and restarted from + seqno 0 before the proxy saw a SYSTEM_TIME backward jump. The + proxy must rotate to a new session file instead of overwriting + the old log's head and then stalling on a seqno=0 that never + comes again.""" + _setup_db(proxy_workdir, PORT_USER, PORT_ENG, 'bintest', 'bp', + 'binlog') + proc = _start_proxy(proxy_workdir, PORT_ENG) + try: + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.bind(('127.0.0.1', 0)) + dest = ('127.0.0.1', PORT_USER) + + # Establish session1: blocks 0..2, then a jump to 250 so + # highest_seen clears the restart threshold (200). + for seq in range(3): + _send_data_block(sock, dest, seq, b'\x11' * 50) + _send_data_block(sock, dest, 250, b'\x22' * 50) + assert _wait_for( + lambda: _bin_path(proxy_workdir, PORT_ENG).exists() + and _bin_path(proxy_workdir, PORT_ENG).stat() + .st_size >= 250 * 200, + timeout=5.0), 'session1 never established' + + # "Reboot": the vehicle restarts its log from seqno 0. + _send_data_block(sock, dest, 0, b'\x33' * 50) + _send_data_block(sock, dest, 1, b'\x33' * 50) + + session2 = _bin_path(proxy_workdir, PORT_ENG, n=2) + assert _wait_for( + lambda: session2.exists() and session2.stat().st_size >= 400, + timeout=5.0), \ + 'no rotation on in-stream seqno=0; proxy log:\n%s' \ + % ''.join(proc._lines[-10:]) + + # session1's head must be intact (not overwritten by the + # new boot's block 0). + with open(_bin_path(proxy_workdir, PORT_ENG), 'rb') as f: + head = f.read(200) + assert head[:50] == b'\x11' * 50, \ + 'old session head overwritten by post-restart block' + + sock.close() + finally: + _terminate(proc) + + def test_midstream_vehicle_gets_stop_nudge(self, proxy_workdir): + """A vehicle streaming mid-log at a proxy with no file open + (proxy restarted mid-flight, or post-rotation) can only recover + by restarting from seqno 0. ArduPilot ignores redundant STARTs + while streaming but honours STOP, so the proxy must send the + STOP magic instead of waiting ~10 s for the vehicle's no-ACK + client timeout.""" + from pymavlink.dialects.v20 import all as mav + _setup_db(proxy_workdir, PORT_USER, PORT_ENG, 'bintest', 'bp', + 'binlog') + proc = _start_proxy(proxy_workdir, PORT_ENG) + try: + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.bind(('127.0.0.1', 0)) + dest = ('127.0.0.1', PORT_USER) + + # Mid-log blocks (from sysid 1): rejected by the + # strict-start gate. Three of them, to clear the lone- + # straggler threshold. + for seq in (300, 301, 302): + _send_data_block(sock, dest, seq, b'\x44' * 50) + + # The proxy must nudge us with the STOP magic — addressed + # to the vehicle it heard, not broadcast. + stop_msg = None + deadline = time.time() + 6 + while time.time() < deadline and stop_msg is None: + for m in _recv_block_status_msgs(sock, timeout=1.0): + if m.seqno == mav.MAV_REMOTE_LOG_DATA_BLOCK_STOP: + stop_msg = m + break + assert stop_msg is not None, \ + 'no STOP nudge for mid-log stream; proxy log:\n%s' \ + % ''.join(proc._lines[-10:]) + assert stop_msg.target_system == 1, \ + 'STOP not targeted at the sending vehicle: target=%d' \ + % stop_msg.target_system + + # "Vehicle" reacts like AP_Logger_MAVLink: stops, then the + # START restarts it from seqno 0 — data must now land. + _send_data_block(sock, dest, 0, b'\x55' * 50) + _send_data_block(sock, dest, 1, b'\x55' * 50) + bin_path = _bin_path(proxy_workdir, PORT_ENG) + assert _wait_for( + lambda: bin_path.exists() and bin_path.stat().st_size >= 400, + timeout=5.0), 'restart from seqno 0 did not open the file' + + sock.close() + finally: + _terminate(proc) + + def test_malformed_quota_env_ignored(self, proxy_workdir): + """SUPPORTPROXY_PORT2_QUOTA_BYTES='1GB' must be rejected, not + prefix-parsed to a 1-byte quota (which would block all writes + and let the cleanup delete nearly the whole log tree).""" + _setup_db(proxy_workdir, PORT_USER, PORT_ENG, 'bintest', 'bp', + 'binlog') + proc = _start_proxy(proxy_workdir, PORT_ENG, quota_bytes='1GB') + try: + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.bind(('127.0.0.1', 0)) + dest = ('127.0.0.1', PORT_USER) + _send_data_block(sock, dest, 0, b'\xaa' * 50) + + bin_path = _bin_path(proxy_workdir, PORT_ENG) + assert _wait_for( + lambda: bin_path.exists() and bin_path.stat().st_size >= 200, + timeout=5.0), \ + "writes blocked - '1GB' was prefix-parsed; proxy log:\n%s" \ + % ''.join(proc._lines[-10:]) + sock.close() + finally: + _terminate(proc) + + def test_single_stale_block_no_stop_nudge(self, proxy_workdir): + """One stale mid-log block (e.g. a delayed pre-reboot packet + right after rotation) must NOT trigger the STOP nudge — that + would stop a healthy stream whose seqno 0 is already on the + way. Three or more gated blocks must.""" + from pymavlink.dialects.v20 import all as mav + _setup_db(proxy_workdir, PORT_USER, PORT_ENG, 'bintest', 'bp', + 'binlog') + proc = _start_proxy(proxy_workdir, PORT_ENG) + try: + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.bind(('127.0.0.1', 0)) + dest = ('127.0.0.1', PORT_USER) + + def _saw_stop(timeout): + deadline = time.time() + timeout + while time.time() < deadline: + for m in _recv_block_status_msgs(sock, timeout=0.5): + if m.seqno == mav.MAV_REMOTE_LOG_DATA_BLOCK_STOP: + return True + return False + + # A lone straggler: no STOP. + _send_data_block(sock, dest, 500, b'\x66' * 50) + assert not _saw_stop(3.0), \ + 'STOP fired on a single stale block' + + # Two more gated blocks cross the threshold: STOP fires. + _send_data_block(sock, dest, 501, b'\x66' * 50) + _send_data_block(sock, dest, 502, b'\x66' * 50) + assert _saw_stop(4.0), \ + 'no STOP after threshold; proxy log:\n%s' \ + % ''.join(proc._lines[-10:]) + + sock.close() + finally: + _terminate(proc) + def test_late_old_block_after_rotation_dropped(self, proxy_workdir): """After SYSTEM_TIME-detected reboot, rotate_for_reboot() closes the file and arms pending_session_n_ but does NOT diff --git a/tests/test_conn2_slot_orphan.py b/tests/test_conn2_slot_orphan.py new file mode 100644 index 0000000..04f89e0 --- /dev/null +++ b/tests/test_conn2_slot_orphan.py @@ -0,0 +1,191 @@ +"""Regression test: closing one engineer slot must not orphan the others. + +The conn2 close paths used to shrink the scan watermark with + if (conn2_count == max_conn2_count) max_conn2_count--; +which fires whenever the slot table has no holes, regardless of which +slot closed. With engineers in slots 0 and 1, an EOF on slot 0 dropped +the watermark to 1 and slot 1 fell out of every scan loop (select set, +reads, forwarding, idle close, connections.tdb snapshot): that engineer +silently stopped receiving anything until a third connection happened +to raise the watermark again. + +This test connects a user plus two signed TCP engineers, disconnects +engineer #1, and asserts engineer #2 still receives the user's traffic. +""" +import hashlib +import os +import signal +import subprocess +import sys +import threading +import time + +import pytest + +_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) + +import conntdb_lib # noqa: E402 +import keydb_lib # noqa: E402 + +SUPPORTPROXY_BIN = os.path.join(_REPO_ROOT, 'supportproxy') + +_W = int(os.environ.get('PYTEST_XDIST_WORKER', 'gw0')[2:] + if os.environ.get('PYTEST_XDIST_WORKER', 'gw0').startswith('gw') else 0) +PORT_USER = 17800 + _W * 4 +PORT_ENG = 17801 + _W * 4 + +os.environ.setdefault('MAVLINK_DIALECT', 'all') +os.environ.setdefault('MAVLINK20', '1') + + +@pytest.fixture +def proxy_workdir(tmp_path): + p = tmp_path / 'work' + p.mkdir() + db_path = str(p / 'keys.tdb') + db = keydb_lib.init_db(db_path) + db.transaction_start() + keydb_lib.add_entry(db, PORT_USER, PORT_ENG, 'orphan_test', 'orphanpw') + db.transaction_prepare_commit() + db.transaction_commit() + db.close() + return p + + +def _start_proxy(workdir): + proc = subprocess.Popen( + [SUPPORTPROXY_BIN], cwd=str(workdir), + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, + bufsize=1, text=True, + ) + proc._lines = [] + proc._ready = threading.Event() + + def _drain(): + for line in iter(proc.stdout.readline, ''): + proc._lines.append(line) + if 'Added port %d/%d' % (PORT_USER, PORT_ENG) in line: + proc._ready.set() + proc.stdout.close() + + proc._thread = threading.Thread(target=_drain, daemon=True) + proc._thread.start() + if not proc._ready.wait(timeout=10): + proc.kill() + proc.wait(timeout=2) + raise RuntimeError('proxy did not load test port pair') + return proc + + +def _terminate(proc): + proc.send_signal(signal.SIGTERM) + try: + proc.wait(timeout=5) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait(timeout=2) + + +def _wait(predicate, timeout=5.0): + deadline = time.time() + timeout + while time.time() < deadline: + if predicate(): + return True + time.sleep(0.1) + return False + + +def _list_conn_indices(workdir, port2): + path = str(workdir / 'connections.tdb') + return sorted(c.conn_index for c in conntdb_lib.list_active(path) + if c.port2 == port2) + + +def _recv_user_heartbeat(conn, user_sysid, timeout): + """True if ``conn`` receives a HEARTBEAT originating from the user + (srcSystem == user_sysid) within ``timeout`` seconds.""" + deadline = time.time() + timeout + while time.time() < deadline: + m = conn.recv_match(type='HEARTBEAT', blocking=True, timeout=0.5) + if m is not None and m.get_srcSystem() == user_sysid: + return True + return False + + +@pytest.mark.skipif(not os.path.exists(SUPPORTPROXY_BIN), + reason='supportproxy binary not built') +class TestConn2SlotOrphan: + def test_engineer2_survives_engineer1_disconnect(self, proxy_workdir): + from pymavlink import mavutil + secret = hashlib.sha256(b'orphanpw').digest() + + proc = _start_proxy(proxy_workdir) + stop_ev = threading.Event() + user = eng1 = eng2 = None + try: + # User side streams continuously in the background so the + # child keeps iterating and forwarding throughout the test. + user = mavutil.mavlink_connection( + 'udpout:127.0.0.1:%d' % PORT_USER, + source_system=10, source_component=1) + + def _user_stream(): + while not stop_ev.is_set(): + user.mav.heartbeat_send(0, 0, 0, 0, 0) + time.sleep(0.1) + + t = threading.Thread(target=_user_stream, daemon=True) + t.start() + + assert _wait(lambda: 0 in _list_conn_indices( + proxy_workdir, PORT_ENG), timeout=10), 'user never latched' + + # Engineer #1 into slot 0 (conn_index 1). + eng1 = mavutil.mavlink_connection( + 'tcp:127.0.0.1:%d' % PORT_ENG, + source_system=11, source_component=1) + eng1.setup_signing(secret, sign_outgoing=True) + for _ in range(10): + eng1.mav.heartbeat_send(0, 0, 0, 0, 0) + time.sleep(0.1) + assert _wait(lambda: 1 in _list_conn_indices( + proxy_workdir, PORT_ENG), timeout=5), 'engineer #1 missing' + + # Engineer #2 into slot 1 (conn_index 2). + eng2 = mavutil.mavlink_connection( + 'tcp:127.0.0.1:%d' % PORT_ENG, + source_system=12, source_component=1) + eng2.setup_signing(secret, sign_outgoing=True) + for _ in range(10): + eng2.mav.heartbeat_send(0, 0, 0, 0, 0) + time.sleep(0.1) + assert _wait(lambda: _list_conn_indices( + proxy_workdir, PORT_ENG) == [0, 1, 2], timeout=5), \ + 'expected slots 0,1,2; got %r' % ( + _list_conn_indices(proxy_workdir, PORT_ENG),) + + # Sanity: engineer #2 sees the user's heartbeats. + assert _recv_user_heartbeat(eng2, 10, timeout=5), \ + 'engineer #2 not receiving user traffic before disconnect' + + # Engineer #1 disconnects abruptly. + eng1.close() + eng1 = None + time.sleep(1.0) # let the proxy process the EOF + + # Drain anything forwarded before the EOF was processed. + while eng2.recv_match(blocking=False) is not None: + pass + + # Engineer #2 must keep receiving the user's traffic. + assert _recv_user_heartbeat(eng2, 10, timeout=5), \ + 'engineer #2 orphaned after engineer #1 disconnect; ' \ + 'recent proxy output:\n%s' % ''.join(proc._lines[-15:]) + finally: + stop_ev.set() + for c in (eng1, eng2, user): + if c is not None: + c.close() + _terminate(proc) diff --git a/tests/test_connections.py b/tests/test_connections.py index 14e85bb..a5c0976 100644 --- a/tests/test_connections.py +++ b/tests/test_connections.py @@ -149,38 +149,72 @@ def check_supportproxy_output(self, test_server, expected_messages, num_lines=5) def create_connection(self, connection_type, port, source_system=1, source_component=1): - """Create a connection of the given type (udp / tcp / ws / wss).""" + """Create a connection of the given type (udp / tcp / ws / wss). + + TCP/WS/WSS construction connects (and, for WS, handshakes) + synchronously. Under the CI runner's heavy -j oversubscription + the proxy child can be slow enough to answer that pymavlink's + own connect retries are exhausted — leaving self.sock == None, + which older pymavlink then dereferences (self.sock.fileno()) + and crashes. Retry construction a few times so a transient + connect starvation doesn't fail the test.""" + def _connect(): + if connection_type == 'udp': + return mavutil.mavlink_connection( + f'udpout:localhost:{port}', + source_system=source_system, + source_component=source_component, + use_native=False + ) + elif connection_type == 'tcp': + return mavutil.mavlink_connection( + f'tcp:localhost:{port}', + source_system=source_system, + source_component=source_component, + autoreconnect=True, + use_native=False + ) + elif connection_type == 'ws': + return mavutil.mavlink_connection( + f'ws:localhost:{port}', + source_system=source_system, + source_component=source_component, + use_native=False, + ) + elif connection_type == 'wss': + return mavutil.mavlink_connection( + f'wss:localhost:{port}', + source_system=source_system, + source_component=source_component, + use_native=False, + ) + else: + raise ValueError(f"Unknown connection type: {connection_type}") + + # UDP is connectionless and never fails to "connect"; only the + # stream transports need the retry. if connection_type == 'udp': - return mavutil.mavlink_connection( - f'udpout:localhost:{port}', - source_system=source_system, - source_component=source_component, - use_native=False - ) - elif connection_type == 'tcp': - return mavutil.mavlink_connection( - f'tcp:localhost:{port}', - source_system=source_system, - source_component=source_component, - autoreconnect=True, - use_native=False - ) - elif connection_type == 'ws': - return mavutil.mavlink_connection( - f'ws:localhost:{port}', - source_system=source_system, - source_component=source_component, - use_native=False, - ) - elif connection_type == 'wss': - return mavutil.mavlink_connection( - f'wss:localhost:{port}', - source_system=source_system, - source_component=source_component, - use_native=False, - ) - else: - raise ValueError(f"Unknown connection type: {connection_type}") + return _connect() + last_err = None + for attempt in range(8): + try: + conn = _connect() + # older pymavlink returns an object with fd=None on a + # failed WS connect instead of raising; treat that as a + # retryable failure too. + if getattr(conn, 'fd', 0) is None: + try: + conn.close() + except Exception: + pass + raise ConnectionError('connect left socket unset') + return conn + except (AttributeError, ConnectionError, OSError) as e: + last_err = e + time.sleep(0.5) + raise RuntimeError( + "could not establish %s connection to port %d after retries: %r" + % (connection_type, port, last_err)) def setup_signing(self, connection, signing_key=None, enable_signing=True): """Setup signing for a connection.""" @@ -235,37 +269,48 @@ def run_test_scenario(self, test_server, user_conn_type, engineer_conn_type, """ user_conn = None engineer_conn = None + stop_sending = threading.Event() + sender_threads = [] try: port1, port2 = ports if ports is not None else TEST_PORTS - # Create connections + # User connection first; the engineer is connected only + # after the user-side wait below. Connecting the engineer + # up front left its socket idle while + # wait_for_connection_user burned its full 10s in the + # bidi negative tests (no conn1 latch by design), and the + # proxy's 5s conn2 pre-auth deadline then closed it before + # any signed traffic flowed — a flaky handshake/reconnect + # race, not proxy misbehaviour. An engineer must start + # signing within 5s of connecting. user_conn = self.create_connection(user_conn_type, port1, source_system=1) - engineer_conn = self.create_connection(engineer_conn_type, port2, - source_system=2) - - # Setup signing for engineer if provided - self.setup_signing(engineer_conn, engineer_signing_key, - enable_signing=(engineer_signing_key is not None)) # Setup signing for user (bidi-sign tests) self.setup_signing(user_conn, user_signing_key, enable_signing=(user_signing_key is not None)) # Start continuous message sending - stop_sending = threading.Event() - user_sender = self.create_message_sender( user_conn, ['heartbeat_user', 'system_time'], stop_sending) - engineer_sender = self.create_message_sender( - engineer_conn, ['heartbeat_engineer'], stop_sending) - user_thread = threading.Thread(target=user_sender) - engineer_thread = threading.Thread(target=engineer_sender) user_thread.start() + sender_threads.append(user_thread) self.wait_for_connection_user(test_server) + + # Now the engineer: connect, sign, and start sending + # immediately so it authenticates well inside the proxy's + # pre-auth window. + engineer_conn = self.create_connection(engineer_conn_type, port2, + source_system=2) + self.setup_signing(engineer_conn, engineer_signing_key, + enable_signing=(engineer_signing_key is not None)) + engineer_sender = self.create_message_sender( + engineer_conn, ['heartbeat_engineer'], stop_sending) + engineer_thread = threading.Thread(target=engineer_sender) engineer_thread.start() + sender_threads.append(engineer_thread) # Test for specified duration. Drain *all* available messages # per iteration (not just one per type) so a burst of buffered @@ -278,7 +323,18 @@ def run_test_scenario(self, test_server, user_conn_type, engineer_conn_type, for i in range(test_duration): time.sleep(1.0) while True: - m = engineer_conn.recv_match(blocking=False) + # A bidi negative test's unsigned/wrong-key user is + # dropped by the proxy's pre-auth deadline, which + # tears the child down and resets the engineer + # socket. That's expected: nothing was forwarded, so + # treat a transport error as end-of-stream rather + # than an error. A positive test never trips the + # deadline, so a reset there still surfaces as an + # under-count assertion failure. + try: + m = engineer_conn.recv_match(blocking=False) + except (ConnectionError, OSError): + m = None if m is None: break t = m.get_type() @@ -287,18 +343,20 @@ def run_test_scenario(self, test_server, user_conn_type, engineer_conn_type, elif t == 'SYSTEM_TIME': system_time_count += 1 - # Stop sending - stop_sending.set() - user_thread.join() - engineer_thread.join() - return heartbeat_count, system_time_count finally: - if user_conn: - user_conn.close() - if engineer_conn: - engineer_conn.close() + # Stop senders before closing their connections, even when + # we get here via an exception mid-setup. + stop_sending.set() + for t in sender_threads: + t.join() + for conn in (user_conn, engineer_conn): + if conn: + try: + conn.close() + except (ConnectionError, OSError): + pass # Wait for SupportProxy to close connections before next test self.wait_for_connection_close(test_server) diff --git a/tests/test_drop_lost_request.py b/tests/test_drop_lost_request.py new file mode 100644 index 0000000..39bcbec --- /dev/null +++ b/tests/test_drop_lost_request.py @@ -0,0 +1,169 @@ +"""Regression test: a drop request must not be silently lost. + +The webadmin kill path sets CONN_FLAG_DROP_REQUESTED on the record and +sends SIGUSR1. The child's 5s connections.tdb snapshot used to +delete-and-rewrite every record for its port2 with flags=0, so a flag +that landed while a snapshot was in flight was wiped before the child's +scan ran — the kill silently did nothing, and nothing ever retried. + +The fix is twofold: the snapshot rewrite preserves DROP_REQUESTED, and +the child rescans for drop requests on the snapshot cadence so a +request whose SIGUSR1 raced the rewrite is still honoured. + +This test simulates the lost-signal case directly: it sets the flag on +an engineer record *without* sending SIGUSR1 and asserts the connection +is dropped anyway within a couple of snapshot cycles. +""" +import hashlib +import os +import signal +import subprocess +import sys +import threading +import time + +import pytest + +_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) + +import conntdb_lib # noqa: E402 +import keydb_lib # noqa: E402 + +SUPPORTPROXY_BIN = os.path.join(_REPO_ROOT, 'supportproxy') + +_W = int(os.environ.get('PYTEST_XDIST_WORKER', 'gw0')[2:] + if os.environ.get('PYTEST_XDIST_WORKER', 'gw0').startswith('gw') else 0) +PORT_USER = 17900 + _W * 4 +PORT_ENG = 17901 + _W * 4 + +os.environ.setdefault('MAVLINK_DIALECT', 'all') +os.environ.setdefault('MAVLINK20', '1') + + +@pytest.fixture +def proxy_workdir(tmp_path): + p = tmp_path / 'work' + p.mkdir() + db_path = str(p / 'keys.tdb') + db = keydb_lib.init_db(db_path) + db.transaction_start() + keydb_lib.add_entry(db, PORT_USER, PORT_ENG, 'lostdrop', 'lostpw') + db.transaction_prepare_commit() + db.transaction_commit() + db.close() + return p + + +def _start_proxy(workdir): + proc = subprocess.Popen( + [SUPPORTPROXY_BIN], cwd=str(workdir), + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, + bufsize=1, text=True, + ) + proc._lines = [] + proc._ready = threading.Event() + + def _drain(): + for line in iter(proc.stdout.readline, ''): + proc._lines.append(line) + if 'Added port %d/%d' % (PORT_USER, PORT_ENG) in line: + proc._ready.set() + proc.stdout.close() + + proc._thread = threading.Thread(target=_drain, daemon=True) + proc._thread.start() + if not proc._ready.wait(timeout=10): + proc.kill() + proc.wait(timeout=2) + raise RuntimeError('proxy did not load test port pair') + return proc + + +def _terminate(proc): + proc.send_signal(signal.SIGTERM) + try: + proc.wait(timeout=5) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait(timeout=2) + + +def _wait(predicate, timeout=5.0): + deadline = time.time() + timeout + while time.time() < deadline: + if predicate(): + return True + time.sleep(0.1) + return False + + +def _list_conn_indices(workdir, port2): + path = str(workdir / 'connections.tdb') + return sorted(c.conn_index for c in conntdb_lib.list_active(path) + if c.port2 == port2) + + +@pytest.mark.skipif(not os.path.exists(SUPPORTPROXY_BIN), + reason='supportproxy binary not built') +class TestDropRequestNotLost: + def test_flag_without_signal_still_drops(self, proxy_workdir): + from pymavlink import mavutil + secret = hashlib.sha256(b'lostpw').digest() + + proc = _start_proxy(proxy_workdir) + stop_ev = threading.Event() + user = eng = None + try: + # Continuous user traffic keeps the child's loop (and its + # 5s snapshot/scan cadence) running throughout. + user = mavutil.mavlink_connection( + 'udpout:127.0.0.1:%d' % PORT_USER, + source_system=10, source_component=1) + + def _user_stream(): + while not stop_ev.is_set(): + user.mav.heartbeat_send(0, 0, 0, 0, 0) + time.sleep(0.1) + + t = threading.Thread(target=_user_stream, daemon=True) + t.start() + + # Signed TCP engineer: no idle close, no auto-reconnect, so + # the slot only goes away if the drop request is honoured. + eng = mavutil.mavlink_connection( + 'tcp:127.0.0.1:%d' % PORT_ENG, + source_system=11, source_component=1) + eng.setup_signing(secret, sign_outgoing=True) + for _ in range(10): + eng.mav.heartbeat_send(0, 0, 0, 0, 0) + time.sleep(0.1) + + assert _wait(lambda: _list_conn_indices( + proxy_workdir, PORT_ENG) == [0, 1], timeout=10), \ + 'expected slots 0,1; got %r' % ( + _list_conn_indices(proxy_workdir, PORT_ENG),) + + # Set the drop flag but do NOT send SIGUSR1 — this is the + # state after the signal raced a snapshot rewrite and the + # child's scan found nothing. + pid = conntdb_lib._flip_drop_flag( + str(proxy_workdir / 'connections.tdb'), PORT_ENG, 1) + assert pid, 'engineer record missing when setting drop flag' + + # The request must still be honoured within a couple of + # snapshot cycles (5s cadence); the user connection stays. + assert _wait(lambda: 1 not in _list_conn_indices( + proxy_workdir, PORT_ENG), timeout=15), \ + 'drop request was silently lost; slots: %r\n%s' % ( + _list_conn_indices(proxy_workdir, PORT_ENG), + ''.join(proc._lines[-15:])) + assert 0 in _list_conn_indices(proxy_workdir, PORT_ENG), \ + 'user connection should survive an engineer drop' + finally: + stop_ev.set() + for c in (eng, user): + if c is not None: + c.close() + _terminate(proc) diff --git a/tests/test_engineer_preauth_pool.py b/tests/test_engineer_preauth_pool.py new file mode 100644 index 0000000..fca77ff --- /dev/null +++ b/tests/test_engineer_preauth_pool.py @@ -0,0 +1,87 @@ +"""Regression test for Codex finding #2 (engineer-side pre-auth slot +DoS) and the structural half of #3. + +Pre-fix: an unauthenticated client could open a TCP connection or send +a single UDP datagram and consume a conn2 slot indefinitely. With +MAX_COMM2_LINKS = 100 slots per port pair, ~100 idle attacker sockets +DoS the legitimate support engineer. + +Fix: the per-pair child closes any conn2 slot that has not validated +a signed packet within CONN2_PREAUTH_SECONDS (=5s), measured from +slot creation. Slot counters are decremented properly on close, so +unauthenticated churn no longer monotonically inflates max_conn2_count. +""" +import os +import socket +import time + +import pytest +from pymavlink import mavutil + +os.environ.setdefault('MAVLINK_DIALECT', 'all') +os.environ.setdefault('MAVLINK20', '1') + +from test_config import TEST_PORT_USER, TEST_PORT_ENGINEER, TEST_PASSPHRASE +from test_connections import passphrase_to_key, BaseConnectionTest + + +class TestEngineerPreauthPool(BaseConnectionTest): + + def test_unsigned_tcp_hog_doesnt_block_signed_engineer(self, test_server): + # Fill most engineer slots with unauthenticated TCP connections. + # Pre-fix these would camp until the proxy died / the user gave + # up; post-fix the pre-auth deadline reaps them after ~5 s. + hogs = [] + try: + for _ in range(60): + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.settimeout(2.0) + s.connect(("127.0.0.1", TEST_PORT_ENGINEER)) + hogs.append(s) + + # Wait past the pre-auth deadline so the proxy reaps them. + time.sleep(7) + + # consume stale output, then bring up a legitimate signed + # engineer + a user side so the forwarding path is alive + test_server.get_new_output_since_last_check() + + user = mavutil.mavlink_connection( + f'udpout:127.0.0.1:{TEST_PORT_USER}', source_system=1) + user.mav.heartbeat_send( + mavutil.mavlink.MAV_TYPE_QUADROTOR, + mavutil.mavlink.MAV_AUTOPILOT_ARDUPILOTMEGA, 0, 0, 0) + time.sleep(0.3) + + key = passphrase_to_key(TEST_PASSPHRASE) + eng = mavutil.mavlink_connection( + f'udpout:127.0.0.1:{TEST_PORT_ENGINEER}', + source_system=11, source_component=21) + eng.setup_signing(key, sign_outgoing=True) + for _ in range(6): + eng.mav.heartbeat_send( + mavutil.mavlink.MAV_TYPE_GCS, + mavutil.mavlink.MAV_AUTOPILOT_INVALID, 0, 0, 0) + time.sleep(0.3) + eng.close() + user.close() + finally: + for s in hogs: + try: + s.close() + except OSError: + pass + + self.assert_with_proxy_log( + test_server, + test_server.proc.poll() is None, + "supportproxy died under TCP hog + signed engineer connect") + + stdout, stderr = test_server.get_new_output_since_last_check() + combined = stdout + stderr + # the legitimate engineer must have validated at least one signed + # packet (proves the slot was available) + self.assert_with_proxy_log( + test_server, + "Got good signature" in combined, + "legitimate signed engineer never validated — slots not freed") diff --git a/tests/test_engineer_udp_churn.py b/tests/test_engineer_udp_churn.py new file mode 100644 index 0000000..bfea87e --- /dev/null +++ b/tests/test_engineer_udp_churn.py @@ -0,0 +1,45 @@ +"""Regression test for Codex finding #3 (counter-leak half): the +engineer UDP idle-timeout close path did not decrement +conn2_count/max_conn2_count, so repeated unauthenticated UDP churn +across MAX_COMM2_LINKS+ tuples drove max_conn2_count past the cap and +triggered the proxy's BUG exit(1) at supportproxy.cpp:438. + +The fix decrements both counters on idle close and (defence in depth) +clamps + warns instead of exiting if the BUG guard ever fires. +""" +import socket +import time + +import pytest + +from test_config import TEST_PORT_ENGINEER +from test_connections import BaseConnectionTest + + +def _churn(port, count): + """Send one short UDP datagram from `count` distinct ephemeral + source ports. Each new tuple consumes a fresh conn2 slot at the + proxy.""" + for _ in range(count): + s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + s.sendto(b"\xff" * 8, ("127.0.0.1", port)) + finally: + s.close() + + +class TestEngineerUdpChurn(BaseConnectionTest): + + def test_udp_churn_does_not_BUG_exit(self, test_server): + # MAX_COMM2_LINKS is 100. Burst 50 unsigned tuples; wait past the + # 10s idle close threshold; then burst 60 more. Pre-fix this drove + # max_conn2_count to 110 and tripped the BUG exit. + _churn(TEST_PORT_ENGINEER, 50) + time.sleep(11) + _churn(TEST_PORT_ENGINEER, 60) + time.sleep(1) + + self.assert_with_proxy_log( + test_server, + test_server.proc.poll() is None, + "supportproxy died after engineer UDP-tuple churn") diff --git a/tests/test_log_cleanup.py b/tests/test_log_cleanup.py index 3aa0a53..02f2c60 100644 --- a/tests/test_log_cleanup.py +++ b/tests/test_log_cleanup.py @@ -62,9 +62,11 @@ def _seed(workdir, port2, date, session_name, age_seconds): return f -def _start_proxy(workdir, interval='0.3'): +def _start_proxy(workdir, interval='0.3', quota_bytes=None): env = os.environ.copy() env['SUPPORTPROXY_CLEANUP_INTERVAL'] = interval + if quota_bytes is not None: + env['SUPPORTPROXY_PORT2_QUOTA_BYTES'] = str(quota_bytes) proc = subprocess.Popen([SUPPORTPROXY_BIN], cwd=str(workdir), env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) return proc @@ -149,37 +151,36 @@ def test_missing_logs_dir_is_ok(self, proxy_workdir): _terminate(proc) def test_quota_deletes_oldest_even_with_retention_zero(self, proxy_workdir): - """The per-port-pair 1 GiB quota is enforced INDEPENDENTLY of - the per-entry retention. With retention=0 (keep forever) the + """The per-port-pair quota is enforced INDEPENDENTLY of the + per-entry retention. With retention=0 (keep forever) the retention pass is a no-op, but the quota pass must still - delete oldest files until total <= 1 GiB. + delete oldest files until under quota. - Uses sparse files (truncate) to fake large apparent sizes - without consuming real disk.""" + Uses SUPPORTPROXY_PORT2_QUOTA_BYTES to shrink the quota so the + test can use small files with real allocated blocks (the quota + counts allocated size, so sparse files don't work here).""" workdir, _, db = proxy_workdir _add_entry(db, 26401, 26402, retention=0.0) # forever - # Three sparse session files, totalling 1.2 GiB. Sort order - # oldest -> newest: a (600 MB), b (400 MB), c (200 MB). - # Quota cap = 1 GiB → 1.2 GiB > cap → drop oldest until under. - # After deleting a (600 MB), total = 600 MB <= 1 GiB. Stops. + # Three real session files, totalling ~1.2 MB against a 1 MB + # quota. Sort order oldest -> newest: a (600 KB), b (400 KB), + # c (200 KB). Deleting a gets under quota; b and c survive. a = _seed(workdir, 26402, '2026-05-08', 'session1.bin', age_seconds=300) b = _seed(workdir, 26402, '2026-05-09', 'session1.bin', age_seconds=200) c = _seed(workdir, 26402, '2026-05-10', 'session1.bin', age_seconds=100) - # Resize each to a large apparent size (sparse). - os.truncate(a, 600 * 1024 * 1024) - os.truncate(b, 400 * 1024 * 1024) - os.truncate(c, 200 * 1024 * 1024) - # Re-set mtimes after the truncate (which updates them). + a.write_bytes(b'\xaa' * (600 * 1024)) + b.write_bytes(b'\xbb' * (400 * 1024)) + c.write_bytes(b'\xcc' * (200 * 1024)) + # Re-set mtimes after the writes (which update them). now = time.time() os.utime(a, (now - 300, now - 300)) os.utime(b, (now - 200, now - 200)) os.utime(c, (now - 100, now - 100)) - proc = _start_proxy(workdir) + proc = _start_proxy(workdir, quota_bytes=1024 * 1024) try: # Quota pass runs once at startup + on each cleanup tick. assert _wait_for_state(lambda: not a.exists(), timeout=5.0), \ @@ -189,3 +190,39 @@ def test_quota_deletes_oldest_even_with_retention_zero(self, proxy_workdir): assert c.exists(), 'newest file deleted unexpectedly' finally: _terminate(proc) + + def test_quota_frees_headroom_below_cap(self, proxy_workdir): + """The quota pass must free down to ~80% of the cap, not stop + just under it — otherwise an active session's growth re-breaches + the cap within minutes and binlog writes stall until the next + hourly pass.""" + workdir, _, db = proxy_workdir + _add_entry(db, 26501, 26502, retention=0.0) # forever + + # a (150 KB, oldest), b (150 KB), c (800 KB, newest): total + # ~1.1 MB against a 1 MB quota. Deleting a alone gets under the + # cap (~950 KB) but not under the 80% target (~819 KB); b must + # go too. c (newest) survives. + a = _seed(workdir, 26502, '2026-05-08', 'session1.bin', + age_seconds=300) + b = _seed(workdir, 26502, '2026-05-09', 'session1.bin', + age_seconds=200) + c = _seed(workdir, 26502, '2026-05-10', 'session1.bin', + age_seconds=100) + a.write_bytes(b'\xaa' * (150 * 1024)) + b.write_bytes(b'\xbb' * (150 * 1024)) + c.write_bytes(b'\xcc' * (800 * 1024)) + now = time.time() + os.utime(a, (now - 300, now - 300)) + os.utime(b, (now - 200, now - 200)) + os.utime(c, (now - 100, now - 100)) + + proc = _start_proxy(workdir, quota_bytes=1024 * 1024) + try: + assert _wait_for_state(lambda: not a.exists(), timeout=5.0), \ + 'oldest file should have been removed by quota pass' + assert _wait_for_state(lambda: not b.exists(), timeout=5.0), \ + 'quota pass stopped at the cap instead of freeing headroom' + assert c.exists(), 'newest file deleted unexpectedly' + finally: + _terminate(proc) diff --git a/tests/test_parent_housekeeping.py b/tests/test_parent_housekeeping.py new file mode 100644 index 0000000..c911bc5 --- /dev/null +++ b/tests/test_parent_housekeeping.py @@ -0,0 +1,184 @@ +"""Regression test: parent housekeeping must not be starved by traffic. + +The parent used to run check_children()/reload_ports() only when +epoll_wait() returned 0 (a full second with no events). Because the +per-pair children inherit the listening sockets, the parent's epoll +registrations survive its close() after fork and fire for every UDP +packet on any active session. With one busy session anywhere, the +parent never reaped exited children and never reopened their +listeners: a connection killed via the web UI never came back even +though its user kept sending UDP packets. + +This test runs two port pairs: pair A streams continuously, pair B's +user connection is killed via the CONN_FLAG_DROP_REQUESTED + SIGUSR1 +path. Pair B's user keeps transmitting; a fresh child must pick the +session up again while pair A's stream never pauses. +""" +import hashlib +import os +import signal +import subprocess +import sys +import threading +import time + +import pytest + +_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) + +import conntdb_lib # noqa: E402 +import keydb_lib # noqa: E402 + +SUPPORTPROXY_BIN = os.path.join(_REPO_ROOT, 'supportproxy') + +_W = int(os.environ.get('PYTEST_XDIST_WORKER', 'gw0')[2:] + if os.environ.get('PYTEST_XDIST_WORKER', 'gw0').startswith('gw') else 0) +PORT_A_USER = 17700 + _W * 8 +PORT_A_ENG = 17701 + _W * 8 +PORT_B_USER = 17702 + _W * 8 +PORT_B_ENG = 17703 + _W * 8 + +os.environ.setdefault('MAVLINK_DIALECT', 'all') +os.environ.setdefault('MAVLINK20', '1') + + +@pytest.fixture +def proxy_workdir(tmp_path): + p = tmp_path / 'work' + p.mkdir() + db_path = str(p / 'keys.tdb') + db = keydb_lib.init_db(db_path) + db.transaction_start() + keydb_lib.add_entry(db, PORT_A_USER, PORT_A_ENG, 'hk_busy', 'busypw') + keydb_lib.add_entry(db, PORT_B_USER, PORT_B_ENG, 'hk_victim', 'victimpw') + db.transaction_prepare_commit() + db.transaction_commit() + db.close() + return p + + +def _start_proxy(workdir): + proc = subprocess.Popen( + [SUPPORTPROXY_BIN], cwd=str(workdir), + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, + bufsize=1, text=True, + ) + proc._lines = [] + proc._ready = threading.Event() + markers = { + 'Added port %d/%d' % (PORT_A_USER, PORT_A_ENG): False, + 'Added port %d/%d' % (PORT_B_USER, PORT_B_ENG): False, + } + + def _drain(): + for line in iter(proc.stdout.readline, ''): + proc._lines.append(line) + for m in markers: + if m in line: + markers[m] = True + if all(markers.values()): + proc._ready.set() + proc.stdout.close() + + proc._thread = threading.Thread(target=_drain, daemon=True) + proc._thread.start() + if not proc._ready.wait(timeout=10): + proc.kill() + proc.wait(timeout=2) + raise RuntimeError('proxy did not load both test port pairs') + return proc + + +def _terminate(proc): + proc.send_signal(signal.SIGTERM) + try: + proc.wait(timeout=5) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait(timeout=2) + + +def _wait(predicate, timeout=5.0, interval=0.1): + deadline = time.time() + timeout + while time.time() < deadline: + if predicate(): + return True + time.sleep(interval) + return False + + +def _pid_for(workdir, port2, conn_index): + path = str(workdir / 'connections.tdb') + for c in conntdb_lib.list_active(path): + if c.port2 == port2 and c.conn_index == conn_index: + return c.pid + return None + + +class _Blaster(threading.Thread): + """Send heartbeats to a UDP port at a fixed interval until stopped.""" + + def __init__(self, port, interval, source_system): + super().__init__(daemon=True) + from pymavlink import mavutil + self.conn = mavutil.mavlink_connection( + 'udpout:127.0.0.1:%d' % port, + source_system=source_system, source_component=1) + self.interval = interval + self.stop_ev = threading.Event() + + def run(self): + while not self.stop_ev.is_set(): + self.conn.mav.heartbeat_send(0, 0, 0, 0, 0) + time.sleep(self.interval) + + def stop(self): + self.stop_ev.set() + self.join(timeout=2) + self.conn.close() + + +@pytest.mark.skipif(not os.path.exists(SUPPORTPROXY_BIN), + reason='supportproxy binary not built') +class TestParentHousekeeping: + def test_killed_pair_recovers_while_other_pair_streams(self, proxy_workdir): + proc = _start_proxy(proxy_workdir) + busy = victim = None + try: + # Pair A: continuous fast stream, keeps the parent's epoll busy + # via the stale registrations of the forked child's sockets. + busy = _Blaster(PORT_A_USER, interval=0.005, source_system=10) + busy.start() + # Pair B: normal-rate user, keeps transmitting through the kill. + victim = _Blaster(PORT_B_USER, interval=0.1, source_system=20) + victim.start() + + # Both children latch their user connection. + assert _wait(lambda: _pid_for(proxy_workdir, PORT_A_ENG, 0), + timeout=10), 'pair A user never registered' + assert _wait(lambda: _pid_for(proxy_workdir, PORT_B_ENG, 0), + timeout=10), 'pair B user never registered' + old_pid = _pid_for(proxy_workdir, PORT_B_ENG, 0) + + # Kill pair B's user connection via the web UI mechanism. + ok = conntdb_lib.request_drop( + str(proxy_workdir / 'connections.tdb'), PORT_B_ENG, 0) + assert ok + + # Pair B's user is still sending; the parent must reap the + # exited child, reopen the listeners and fork a fresh child + # even though pair A's stream never pauses. + assert _wait( + lambda: _pid_for(proxy_workdir, PORT_B_ENG, 0) + not in (None, old_pid), + timeout=20), \ + 'killed pair never came back while other pair streamed; ' \ + 'recent proxy output:\n%s' % ''.join(proc._lines[-15:]) + finally: + if busy: + busy.stop() + if victim: + victim.stop() + _terminate(proc) diff --git a/tests/test_setup_signing_guard.py b/tests/test_setup_signing_guard.py new file mode 100644 index 0000000..3f9623a --- /dev/null +++ b/tests/test_setup_signing_guard.py @@ -0,0 +1,88 @@ +"""Regression test for Codex finding #6: a signed SETUP_SIGNING with +all-zero secret_key and zero initial_timestamp would persist into +keys.tdb. The next load_signing_key() would then unwire status->signing +and the generated parser would accept signed-flag frames without +checking a secret. The fix rejects that exact combination at +handle_setup_signing(), and load_signing_key() safe-fails if zero ever +lands in keys.tdb through another path. +""" +import os +import time + +import pytest +from pymavlink import mavutil + +os.environ.setdefault('MAVLINK_DIALECT', 'all') +os.environ.setdefault('MAVLINK20', '1') + +from test_config import TEST_PORT_USER, TEST_PORT_ENGINEER, TEST_PASSPHRASE +from test_connections import passphrase_to_key, BaseConnectionTest + + +class TestSetupSigningGuard(BaseConnectionTest): + + def test_all_zero_setup_signing_rejected(self, test_server): + self.wait_for_connection_close(test_server) + # consume any stale output so the log assertion below only sees + # output produced by this test + test_server.get_new_output_since_last_check() + + key = passphrase_to_key(TEST_PASSPHRASE) + + # establish a user side so the engineer→user forward path exists + user = mavutil.mavlink_connection( + f'udpout:127.0.0.1:{TEST_PORT_USER}', source_system=1) + user.mav.heartbeat_send( + mavutil.mavlink.MAV_TYPE_QUADROTOR, + mavutil.mavlink.MAV_AUTOPILOT_ARDUPILOTMEGA, 0, 0, 0) + time.sleep(0.3) + + eng = mavutil.mavlink_connection( + f'udpout:127.0.0.1:{TEST_PORT_ENGINEER}', + source_system=11, source_component=21) + eng.setup_signing(key, sign_outgoing=True) + + try: + # drive enough signed heartbeats for the proxy to log + # 'Got good signature' (and to defeat any pymavlink first- + # packet edge case) + for _ in range(6): + eng.mav.heartbeat_send( + mavutil.mavlink.MAV_TYPE_GCS, + mavutil.mavlink.MAV_AUTOPILOT_INVALID, + 0, 0, 0) + time.sleep(0.3) + + # malicious SETUP_SIGNING: zero key + zero timestamp + eng.mav.setup_signing_send(0, 0, b'\x00' * 32, 0) + time.sleep(1.0) + + # post-rejection, signed heartbeats with the original key + # must still validate (the proxy did not rotate its key) + for _ in range(3): + eng.mav.heartbeat_send( + mavutil.mavlink.MAV_TYPE_GCS, + mavutil.mavlink.MAV_AUTOPILOT_INVALID, + 0, 0, 0) + time.sleep(0.3) + finally: + eng.close() + user.close() + + self.assert_with_proxy_log( + test_server, + test_server.proc.poll() is None, + "supportproxy died after a malicious SETUP_SIGNING") + + # the rejection log line must appear, and the post-rejection + # signed traffic must not have triggered a 'Set new signing key' + stdout, stderr = test_server.get_new_output_since_last_check() + combined = stdout + stderr + self.assert_with_proxy_log( + test_server, + "Rejecting SETUP_SIGNING" in combined, + "expected 'Rejecting SETUP_SIGNING' log line not seen") + self.assert_with_proxy_log( + test_server, + "Set new signing key" not in combined, + "proxy persisted the all-zero key (saw 'Set new signing key')") diff --git a/tests/test_websocket_decode.py b/tests/test_websocket_decode.py new file mode 100644 index 0000000..740a828 --- /dev/null +++ b/tests/test_websocket_decode.py @@ -0,0 +1,79 @@ +"""Regression test for the WebSocket extended-payload-length overflow +(Codex security scan finding #1). Pre-fix, a crafted 0x7f-length frame +with payload_len near UINT64_MAX wrapped the completeness check in +WebSocket::decode() and let the unmask loop / memmove walk over the +fixed 1024-byte pending[] buffer. The fix bounds payload_len before +any addition; the proxy now drops the frame and stays up. +""" +import base64 +import socket +import struct +import time + +import pytest + +from test_config import TEST_PORT_ENGINEER +from test_connections import BaseConnectionTest + + +def _ws_handshake(s): + """Minimal Sec-WebSocket-Key handshake. We only need the proxy to + switch into WebSocket framing mode; we don't validate the Accept + value on our end.""" + key = base64.b64encode(b"x" * 16).decode() + req = ( + "GET / HTTP/1.1\r\n" + "Host: localhost\r\n" + "Upgrade: websocket\r\n" + "Connection: Upgrade\r\n" + f"Sec-WebSocket-Key: {key}\r\n" + "Sec-WebSocket-Version: 13\r\n" + "\r\n" + ).encode() + s.sendall(req) + s.settimeout(3.0) + buf = b"" + while b"\r\n\r\n" not in buf: + chunk = s.recv(4096) + if not chunk: + break + buf += chunk + assert b"101" in buf, f"no 101 Switching Protocols: {buf!r}" + + +def _craft_127_frame(payload_len, payload=b""): + """Build a binary WS frame using the 64-bit extended length encoding, + masked (clients must mask). Zero mask leaves the payload unchanged.""" + hdr = bytes([0x82, 0x80 | 127]) + struct.pack(">Q", payload_len) + hdr += b"\x00\x00\x00\x00" + return hdr + payload + + +class TestWebSocketDecodeOverflow(BaseConnectionTest): + + @pytest.mark.parametrize("payload_len", [ + 0xFFFFFFFFFFFFFFFF, # UINT64_MAX: wraps the pre-fix add to ~14 + 1 << 63, # high bit set + 2048, # just over pending[1024] + ], ids=["uint64_max", "msb_set", "just_over_buffer"]) + def test_oversized_127_frame_does_not_crash(self, test_server, payload_len): + # use the engineer port: it supports multiple parallel TCP + # connections, so parametrized cases don't fight over conn1. + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.settimeout(5.0) + s.connect(("127.0.0.1", TEST_PORT_ENGINEER)) + try: + _ws_handshake(s) + s.sendall(_craft_127_frame(payload_len)) + # let the proxy reach decode() and reject + time.sleep(0.3) + finally: + s.close() + + self.assert_with_proxy_log( + test_server, + test_server.proc.poll() is None, + "supportproxy died after crafted 0x7f frame with " + f"payload_len=0x{payload_len:x}", + ) + diff --git a/tests/test_ws_handshake_ordering.py b/tests/test_ws_handshake_ordering.py new file mode 100644 index 0000000..b5c4bc1 --- /dev/null +++ b/tests/test_ws_handshake_ordering.py @@ -0,0 +1,195 @@ +"""Regression test: the proxy must not write MAVLink frames to a +would-be WebSocket user before it has sent the HTTP upgrade (101) +response. If it does, the frame lands ahead of the "HTTP/1.1 101" +status line and corrupts the handshake — the peer's WS parser chokes +on binary garbage where the status line should be (observed in the +field as `illegal status line: bytearray(b'\\xfd\\t...HTTP/1.1 101')`, +0xfd being the MAVLink2 magic). + +Two windows are covered, each with its own fresh proxy so a sticky +conn1 can't couple the cases: + + 1. TCP accepted, transport not yet known: conn1 is latched but no + user data has been seen, so WebSocket hasn't been detected and + mav1 is still raw-TCP. An engineer frame forwarded now goes out + as plain MAVLink on the raw socket. (Fixed by gating the + engineer->user forward on count1>0.) + 2. WebSocket detected but handshake unfinished: mav1 wraps the + WebSocket but done_headers is false, so a frame would be WS-framed + and written before the 101. (Fixed by WebSocket::send dropping + until done_headers.) +""" +import hashlib +import os +import signal +import socket +import subprocess +import sys +import threading +import time + +import pytest + +_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) + +import keydb_lib # noqa: E402 + +SUPPORTPROXY_BIN = os.path.join(_REPO_ROOT, 'supportproxy') + +_W = int(os.environ.get('PYTEST_XDIST_WORKER', 'gw0')[2:] + if os.environ.get('PYTEST_XDIST_WORKER', 'gw0').startswith('gw') else 0) +PORT_USER = 18000 + _W * 4 +PORT_ENG = 18001 + _W * 4 + +os.environ.setdefault('MAVLINK_DIALECT', 'all') +os.environ.setdefault('MAVLINK20', '1') + + +@pytest.fixture +def proxy_workdir(tmp_path): + p = tmp_path / 'work' + p.mkdir() + db = keydb_lib.init_db(str(p / 'keys.tdb')) + db.transaction_start() + keydb_lib.add_entry(db, PORT_USER, PORT_ENG, 'wshs', 'wshspw') + db.transaction_prepare_commit() + db.transaction_commit() + db.close() + return p + + +def _start_proxy(workdir): + proc = subprocess.Popen( + [SUPPORTPROXY_BIN], cwd=str(workdir), + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, + bufsize=1, text=True, + ) + proc._lines = [] + proc._ready = threading.Event() + + def _drain(): + for line in iter(proc.stdout.readline, ''): + proc._lines.append(line) + if 'Added port %d/%d' % (PORT_USER, PORT_ENG) in line: + proc._ready.set() + proc.stdout.close() + + proc._thread = threading.Thread(target=_drain, daemon=True) + proc._thread.start() + if not proc._ready.wait(timeout=10): + proc.kill() + proc.wait(timeout=2) + raise RuntimeError('proxy did not load test port pair') + return proc + + +def _terminate(proc): + proc.send_signal(signal.SIGTERM) + try: + proc.wait(timeout=5) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait(timeout=2) + + +def _drive(split_at, hold_s=1.0): + """Connect a raw user and send the WebSocket upgrade request in two + parts split at byte ``split_at``: the first part, then (while a + signed engineer streams forwardable data) the rest. Returns the + server's first response bytes. + + split_at == 0 -> send nothing first: proxy still in raw-TCP mode + when it tries to forward (transport undecided). + 0 < split_at < 14 -> first read is a partial handshake prefix + (< the 14-byte "GET / HTTP/1.1" needed to + classify): must not be mistaken for raw MAVLink. + split_at past the request line, before the key -> WebSocket + detected but handshake not yet complete. + """ + from pymavlink import mavutil + import base64 + secret = hashlib.sha256(b'wshspw').digest() + + key = base64.b64encode(b"z" * 16).decode() + req = ("GET / HTTP/1.1\r\nHost: localhost\r\n" + "Upgrade: websocket\r\nConnection: Upgrade\r\n" + "Sec-WebSocket-Key: %s\r\nSec-WebSocket-Version: 13\r\n\r\n" + % key).encode() + + u = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + u.settimeout(5.0) + u.connect(("127.0.0.1", PORT_USER)) + if split_at: + u.sendall(req[:split_at]) + time.sleep(0.3) + + eng = mavutil.mavlink_connection( + 'udpout:127.0.0.1:%d' % PORT_ENG, + source_system=11, source_component=21) + eng.setup_signing(secret, sign_outgoing=True) + deadline = time.time() + hold_s + while time.time() < deadline: + eng.mav.heartbeat_send(0, 0, 0, 0, 0) + time.sleep(0.1) + + u.sendall(req[split_at:]) + + buf = b"" + rd = time.time() + 5 + while b"\r\n\r\n" not in buf and time.time() < rd: + try: + chunk = u.recv(4096) + except socket.timeout: + break + if not chunk: + break + buf += chunk + + eng.close() + u.close() + return buf + + +@pytest.mark.skipif(not os.path.exists(SUPPORTPROXY_BIN), + reason='supportproxy binary not built') +class TestWSHandshakeOrdering: + def _check(self, proc, split_at, why): + buf = _drive(split_at=split_at) + assert buf.startswith(b"HTTP/1.1 101"), \ + ("%s; first bytes: %r\nproxy log:\n%s" + % (why, buf[:64], ''.join(proc._lines[-12:]))) + + def test_no_raw_forward_before_ws_detect(self, proxy_workdir): + # Transport undecided: send nothing before the engineer streams, + # so the proxy is still in raw-TCP mode when it tries to forward. + proc = _start_proxy(proxy_workdir) + try: + self._check(proc, 0, + "handshake corrupted by a raw pre-detect forward") + finally: + _terminate(proc) + + def test_fragmented_prefix_not_misclassified(self, proxy_workdir): + # First read is a 9-byte prefix of "GET / HTTP/1.1" (< the 14 + # needed to classify). It must be held as "maybe WebSocket", not + # committed to raw — otherwise the engineer's forwarded frame + # goes out as raw MAVLink ahead of the eventual 101. + proc = _start_proxy(proxy_workdir) + try: + self._check(proc, 9, + "fragmented GET prefix misclassified as raw") + finally: + _terminate(proc) + + def test_no_ws_forward_before_handshake(self, proxy_workdir): + # WebSocket detected (full request line sent) but Sec-WebSocket- + # Key withheld, so done_headers stays false while the engineer + # streams. + proc = _start_proxy(proxy_workdir) + try: + self._check(proc, len(b"GET / HTTP/1.1\r\nHost: localhost\r\n"), + "handshake corrupted by a pre-handshake WS forward") + finally: + _terminate(proc) diff --git a/util.cpp b/util.cpp index 7e31fd8..aa46187 100644 --- a/util.cpp +++ b/util.cpp @@ -52,8 +52,9 @@ int open_socket_in_udp(int port) setsockopt(res,SOL_SOCKET,SO_REUSEADDR,(char *)&one,sizeof(one)); - if (bind(res, (struct sockaddr *)&sock, sizeof(sock)) < 0) { - return(-1); + if (bind(res, (struct sockaddr *)&sock, sizeof(sock)) < 0) { + close(res); + return(-1); } return res; @@ -94,10 +95,12 @@ int open_socket_in_tcp(int port) set_tcp_options(res); if (bind(res, (struct sockaddr *)&sock, sizeof(sock)) < 0) { - return(-1); + close(res); + return(-1); } if (listen(res, 100) != 0) { + close(res); return(-1); } diff --git a/websocket.cpp b/websocket.cpp index 49ae697..3fc7c94 100644 --- a/websocket.cpp +++ b/websocket.cpp @@ -26,20 +26,43 @@ static uint8_t wss_prefix[] { 0x16, 0x03, 0x01 }; see if this could be a WebSocket connection by looking at the first packet */ -bool WebSocket::detect(int fd) +ws_detect_t WebSocket::detect(int fd) { + const size_t ws_len = strlen(ws_prefix); // 14 ("GET / HTTP/1.1") + const size_t wss_len = sizeof(wss_prefix); // 3 (TLS ClientHello) uint8_t peekbuf[14] {}; - const ssize_t peekn = ::recv(fd, peekbuf, sizeof(peekbuf), MSG_PEEK); - if (peekn >= ssize_t(sizeof(wss_prefix)) && memcmp(wss_prefix, peekbuf, sizeof(wss_prefix)) == 0) { - // SSL connection - return true; + ssize_t peekn = ::recv(fd, peekbuf, sizeof(peekbuf), MSG_PEEK); + if (peekn <= 0) { + return WS_MORE; // nothing readable yet; try again + } + const size_t n = size_t(peekn); + + // TLS ClientHello (wss). Compare only the bytes we have. + const size_t wss_cmp = n < wss_len ? n : wss_len; + if (memcmp(wss_prefix, peekbuf, wss_cmp) == 0) { + if (n >= wss_len) { + return WS_YES; + } + return WS_MORE; // matches so far, need more to be sure } - if (peekn >= ssize_t(sizeof(ws_prefix)) && - strncmp(ws_prefix, (const char *)peekbuf, strlen(ws_prefix)) == 0) { - return true; + + // HTTP upgrade (ws). NOTE: ws_prefix is a char*, so its length is + // strlen(), not sizeof() — the old sizeof() admitted an 8-byte + // prefix and then strncmp'd 14 bytes against a half-filled buffer, + // misclassifying a fragmented "GET / HTTP/1.1" as raw MAVLink. + const size_t ws_cmp = n < ws_len ? n : ws_len; + if (strncmp(ws_prefix, (const char *)peekbuf, ws_cmp) == 0) { + if (n >= ws_len) { + return WS_YES; + } + return WS_MORE; // matches so far, need more to be sure } - return false; + + // Doesn't match either handshake prefix. A raw MAVLink2 frame + // starts with 0xFD (v1: 0xFE), never 'G' or 0x16, so this is a + // definite raw connection even from a single byte. + return WS_NO; } /* @@ -86,6 +109,22 @@ WebSocket::WebSocket(int _fd) check_headers(); } +/* + destructor: release the SSL objects. The socket fd is owned by the + caller (Connection2 / listen_port) and is closed there, never here. + */ +WebSocket::~WebSocket() +{ + if (ssl) { + SSL_free(ssl); + ssl = nullptr; + } + if (ctx) { + SSL_CTX_free(ctx); + ctx = nullptr; + } +} + void WebSocket::check_headers(void) { auto len = strnlen((const char *)pending, npending); @@ -127,8 +166,7 @@ void WebSocket::fill_pending(void) return; } ERR_print_errors_fp(stdout); - close(fd); - fd = -1; + fd = -1; // owner closes the socket return; } printf("SSL handshake completed\n"); @@ -142,13 +180,11 @@ void WebSocket::fill_pending(void) } if (err == SSL_ERROR_ZERO_RETURN) { // orderly shutdown - close(fd); - fd = -1; + fd = -1; // owner closes the socket return; } ERR_print_errors_fp(stdout); - close(fd); - fd = -1; + fd = -1; // owner closes the socket return; } } else { @@ -157,14 +193,12 @@ void WebSocket::fill_pending(void) if (errno == EAGAIN || errno == EWOULDBLOCK) { return; } - close(fd); - fd = -1; + fd = -1; // owner closes the socket return; } if (n == 0) { // EOF - close(fd); - fd = -1; + fd = -1; // owner closes the socket return; } } @@ -174,7 +208,9 @@ void WebSocket::fill_pending(void) /* decode an incoming WebSocket packet and overwrite buf with the decoded data - return the number of decoded payload bytes, or -1 on error + return the number of decoded payload bytes, -1 if the frame is + incomplete (wait for more data), or -2 if the frame can never be + decoded (it doesn't fit in pending[]; the stream is unrecoverable) */ ssize_t WebSocket::decode(uint8_t *buf, size_t n, size_t &used) { @@ -196,6 +232,17 @@ ssize_t WebSocket::decode(uint8_t *buf, size_t n, size_t &used) pos += 8; } + // bound payload_len before the completeness checks below: pending[] is + // fixed-size, and an attacker-supplied payload_len near UINT64_MAX would + // wrap "pos + 4 + payload_len" to a small number, letting the check pass. + // fill_pending() keeps one byte for a NUL, so a frame needing more than + // sizeof(pending)-1 bytes can never complete: waiting for more data + // would wedge the connection forever, so fail it instead. + const size_t mask_bytes = masked ? 4 : 0; + if (payload_len > (sizeof(pending)-1) - pos - mask_bytes) { + return -2; + } + if (masked) { if (n < pos + 4 + payload_len) { return -1; @@ -278,7 +325,7 @@ bool WebSocket::send_handshake(const std::string &key) return false; // try again later } ERR_print_errors_fp(stdout); - close(fd); fd = -1; + fd = -1; // owner closes the socket return false; } } else { @@ -287,7 +334,7 @@ bool WebSocket::send_handshake(const std::string &key) if (errno == EAGAIN || errno == EWOULDBLOCK) { return false; } - close(fd); fd = -1; + fd = -1; // owner closes the socket return false; } } @@ -301,6 +348,16 @@ bool WebSocket::send_handshake(const std::string &key) */ ssize_t WebSocket::send(const void *buf, size_t n) { + if (!done_headers) { + // The HTTP upgrade response hasn't been sent yet. Writing a + // MAVLink frame onto the socket now would land *before* the + // "HTTP/1.1 101" line and corrupt the handshake (the peer's WS + // parser sees binary garbage as the status line). Drop the + // frame but report it as sent so the caller doesn't treat it as + // a dead link and tear the session down; the handshake + // completes on the next read and forwarding resumes. + return n; + } uint8_t header[10]; size_t header_len = 0; header[0] = 0x82; // FIN + binary opcode @@ -331,7 +388,7 @@ ssize_t WebSocket::send(const void *buf, size_t n) return 0; // try again later } ERR_print_errors_fp(stdout); - close(fd); fd = -1; + fd = -1; // owner closes the socket return -1; } } else { @@ -340,7 +397,7 @@ ssize_t WebSocket::send(const void *buf, size_t n) if (errno == EAGAIN || errno == EWOULDBLOCK) { return 0; } - close(fd); fd = -1; + fd = -1; // owner closes the socket return -1; } } @@ -361,9 +418,20 @@ ssize_t WebSocket::recv(void *buf, size_t n) } if (!done_headers) { check_headers(); + if (!done_headers) { + // don't decode partial HTTP upgrade headers as a frame: + // that consumed header bytes and broke the handshake when + // the request arrived fragmented + return 0; + } } size_t used; auto decode_len = decode(pending, npending, used); + if (decode_len == -2) { + // unrecoverable frame; fail the connection so the owner closes it + fd = -1; // owner closes the socket + return -1; + } if (decode_len == -1) { return 0; } diff --git a/websocket.h b/websocket.h index 334aded..f4b7e93 100644 --- a/websocket.h +++ b/websocket.h @@ -9,11 +9,21 @@ #include #include +// Result of peeking at a new TCP stream's first bytes. +// WS_NO - definitely not a WebSocket/TLS handshake (raw MAVLink) +// WS_YES - a WebSocket (HTTP upgrade) or TLS ClientHello +// WS_MORE - the bytes so far are a prefix of a handshake but there +// aren't enough yet to decide; the caller must wait for +// more rather than committing to raw (committing early +// misclassifies a fragmented handshake as raw MAVLink) +enum ws_detect_t { WS_NO, WS_YES, WS_MORE }; + class WebSocket { public: WebSocket(int fd); + ~WebSocket(); - static bool detect(int fd); + static ws_detect_t detect(int fd); ssize_t send(const void *buf, size_t n); ssize_t recv(void *buf, size_t n); bool is_SSL(void) const {