diff --git a/AGENTS.md b/AGENTS.md index 544b6d6..dbb4d28 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -114,7 +114,8 @@ GTest suites under `tests/` (all must stay green; run via `ctest` after a normal | `test_registration_handshake.cpp` | REG1/REG2/REG3 state machine, malformed-frame contracts | | `test_extended_keepalive.cpp` | Extended-KA activation, fallback, and edge semantics | | `test_reg_race.cpp` | REG3/NGP race, concurrent multi-interface registration | -| `test_group_limits.cpp` | MAX_GROUPS exhaustion, REG_ERR at cap | +| `test_group_limits.cpp` | MAX_GROUPS exhaustion, REG_ERR at cap (fillers are data-seen — see RECEIVER HARDENING) | +| `test_ghost_group_eviction.cpp` | Ghost-group reaping/eviction at PENDING_GROUP_TIMEOUT; data-seen groups protected | | `test_timeout_cleanup.cpp` | Per-connection and group timeout/cleanup paths | | `test_identity_hooks.cpp` | GroupIdentity extension hooks (see `docs/EXTENSION_POINTS.md`) | | `test_telemetry_emit.cpp` | ADR-001 stats-file serialization, atomic publish, staleness | @@ -145,6 +146,40 @@ The TS binding reader lives in `bindings/typescript/src/sender/` alongside the e spawn/args helpers. It is an **additive** export — existing exports (`srtlaSendOptionsSchema`, `buildSrtlaSendArgs`, spawn helpers) are frozen and unchanged. +## RECEIVER HARDENING + +`srtla_rec` is a pre-auth UDP relay: a REG1 creates a connection group before any +SRT handshake, and the actual stream auth happens downstream at the SRT server. +Two upstream commits (`irlserver/main` `7855012`, `39e324a`) close the resulting +pre-auth abuse surfaces. All knobs live in `src/receiver_config.h`. + +**1. Ghost-group eviction (anti table-exhaustion DoS).** A group that registered +but never forwarded real SRT data is a "ghost". `ConnectionGroup::mark_data_seen()` +is set on the first forwarded SRT packet (`SRTLAHandler::process_single_packet`), +promoting the group to non-evictable. + +- Empty groups are reaped at `PENDING_GROUP_TIMEOUT` (5 s) if they never saw data, + vs `GROUP_TIMEOUT` (30 s) once promoted (`ConnectionRegistry::cleanup_inactive`). +- At `MAX_GROUPS` (200) a new REG1 evicts the oldest ghost before returning + `REG_ERR` (`evict_oldest_pending_group()`), so a REG1 flood cannot lock out the + real broadcaster. Eviction skips any group with live connections or `data_seen`. + +**2. Per-IP auth-fail rate limiter.** `src/utils/auth_rate_limiter.{cpp,h}` +(linked into `receiver_core_obj`). A failed SRT auth — a libsrt handshake reject, +or (as srt-live-server does) an SRT `SHUTDOWN` before the group is `established` +(server ACK seen) — is counted per source IP; a failed-auth group is torn down +immediately to reclaim its slot. Keys are IP-only so port rotation does not evade. + +- `AUTH_FAIL_THRESHOLD` = 5 failures within +- `AUTH_FAIL_WINDOW` = 60 s trips a block; new REG1s are refused for +- `AUTH_FAIL_COOLDOWN` = 60 s. Lenient by design so a mistyped passphrase or + several broadcasters behind one NAT are not locked out. + +Test-infra note: pre-existing receiver tests model **real** streaming groups, so +their group factories call `mark_data_seen()` (`test_group_limits`, +`test_timeout_cleanup`); the ghost/eviction behavior itself is pinned by +`test_ghost_group_eviction.cpp`. + ## ANTI-PATTERNS - Don't modify the TS bindings API without checking `UPSTREAM MERGE STATUS` above — existing exports are frozen; new functionality must be additive diff --git a/CMakeLists.txt b/CMakeLists.txt index 9c8022c..2ccac40 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -49,7 +49,8 @@ add_library(receiver_core_obj OBJECT src/protocol/srtla_handler.cpp src/protocol/srt_handler.cpp src/utils/network_utils.cpp - src/utils/nak_dedup.cpp) + src/utils/nak_dedup.cpp + src/utils/auth_rate_limiter.cpp) target_include_directories(receiver_core_obj PUBLIC "${CMAKE_CURRENT_SOURCE_DIR}/src") diff --git a/src/common.c b/src/common.c index 06da0a0..35172ee 100644 --- a/src/common.c +++ b/src/common.c @@ -110,6 +110,19 @@ int is_srtla_keepalive(void *pkt, int n) { return get_srt_type(pkt, n) == SRTLA_TYPE_KEEPALIVE; } +int is_srt_shutdown(void *pkt, int n) { + return get_srt_type(pkt, n) == SRT_TYPE_SHUTDOWN; +} + +int is_srt_handshake_reject(void *pkt, int n) { + if (n < (int)sizeof(srt_handshake_t)) + return 0; + if (get_srt_type(pkt, n) != SRT_TYPE_HANDSHAKE) + return 0; + uint32_t hs_type = be32toh(((srt_handshake_t *)pkt)->handshake_type); + return hs_type >= SRT_REJECTION_CODE_BASE && hs_type < 0x80000000u; +} + int is_srtla_reg1(void *pkt, int len) { if (len != SRTLA_TYPE_REG1_LEN) return 0; diff --git a/src/common.h b/src/common.h index 95674da..80cce59 100644 --- a/src/common.h +++ b/src/common.h @@ -103,6 +103,13 @@ int is_srt_ack(void *pkt, int n); int is_srt_nak(void *pkt, int n); int is_srt_shutdown(void *pkt, int n); +// SRT signals a rejected handshake by setting the handshake type field to a +// failure code (URQ_FAILURE_TYPES base). Accepted handshakes use small +// (induction=1) or negative-as-unsigned (conclusion/agreement) values, so a +// value in [1000, 0x80000000) reliably identifies an auth/connection refusal. +#define SRT_REJECTION_CODE_BASE 1000 +int is_srt_handshake_reject(void *pkt, int n); + int is_srtla_keepalive(void *pkt, int len); int is_srtla_reg1(void *pkt, int len); int is_srtla_reg2(void *pkt, int len); diff --git a/src/connection/connection_group.h b/src/connection/connection_group.h index 6eac17a..dc07f51 100644 --- a/src/connection/connection_group.h +++ b/src/connection/connection_group.h @@ -56,6 +56,21 @@ class ConnectionGroup { time_t created_at() const { return created_at_; } + // True once the group has forwarded at least one real SRT packet. Used to + // distinguish authenticated, streaming groups from unauthenticated "ghost" + // groups created by a REG1 flood, which are reaped/evicted aggressively. + bool has_seen_data() const { return data_seen_; } + void mark_data_seen() { data_seen_ = true; } + + // True once the SRT server has ACKed media for this group, i.e. the session + // was accepted and is running. The SRT server (srt-live-server) accepts the + // handshake before checking stream auth and, on failure, simply closes the + // socket (SHUTDOWN) instead of sending a handshake rejection. A SHUTDOWN + // before the group is established therefore indicates a rejected/failed + // connection rather than a legitimate end-of-stream. + bool is_established() const { return established_; } + void mark_established() { established_ = true; } + int srt_socket() const { return srt_sock_; } void set_srt_socket(int sock); @@ -90,6 +105,8 @@ class ConnectionGroup { std::vector conns_; GroupIdentity identity_; time_t created_at_ = 0; + bool data_seen_ = false; + bool established_ = false; int srt_sock_ = -1; struct sockaddr_storage last_addr_ {}; diff --git a/src/connection/connection_registry.cpp b/src/connection/connection_registry.cpp index f9aa2c5..2bbdf41 100644 --- a/src/connection/connection_registry.cpp +++ b/src/connection/connection_registry.cpp @@ -66,6 +66,27 @@ void ConnectionRegistry::remove_group(const ConnectionGroupPtr &group) { groups_.erase(it); } +bool ConnectionRegistry::evict_oldest_pending_group() { + ConnectionGroupPtr oldest; + for (auto &group : groups_) { + if (!group->connections().empty() || group->has_seen_data()) { + continue; + } + if (!oldest || group->created_at() < oldest->created_at()) { + oldest = group; + } + } + + if (!oldest) { + return false; + } + + spdlog::warn("[Group: {}] Evicting pending group to admit new registration (group table full)", + static_cast(oldest.get())); + remove_group(oldest); + return true; +} + ConnectionGroupPtr ConnectionRegistry::find_group_by_id(const char *id) { for (auto &group : groups_) { if (NetworkUtils::constant_time_compare(group->id().data(), id, SRTLA_ID_LEN) == 0) { @@ -167,7 +188,8 @@ void ConnectionRegistry::cleanup_inactive(time_t current_time, } } - if (connections.empty() && (group->created_at() + GROUP_TIMEOUT) < current_time) { + time_t empty_timeout = group->has_seen_data() ? GROUP_TIMEOUT : PENDING_GROUP_TIMEOUT; + if (connections.empty() && (group->created_at() + empty_timeout) < current_time) { if (on_group_reaped) { on_group_reaped(group->identity()); } diff --git a/src/connection/connection_registry.h b/src/connection/connection_registry.h index bbeb858..e6fbd9d 100644 --- a/src/connection/connection_registry.h +++ b/src/connection/connection_registry.h @@ -19,6 +19,11 @@ class ConnectionRegistry { void add_group(const ConnectionGroupPtr &group); void remove_group(const ConnectionGroupPtr &group); + // Evicts the oldest group that registered but never forwarded real SRT data + // (no connections, no traffic). Returns true if one was evicted. Used to + // admit a legitimate registration when the table is full of ghost groups. + bool evict_oldest_pending_group(); + ConnectionGroupPtr find_group_by_id(const char *id); void find_by_address(const struct sockaddr_storage *addr, ConnectionGroupPtr &out_group, diff --git a/src/protocol/srt_handler.cpp b/src/protocol/srt_handler.cpp index eb81572..ec5a171 100644 --- a/src/protocol/srt_handler.cpp +++ b/src/protocol/srt_handler.cpp @@ -6,6 +6,13 @@ #include "pad_sendto.h" #include + +static inline int is_srt_handshake(const void *pkt, int n) { + if (n < 16) return 0; + const unsigned char *p = (const unsigned char *)pkt; + return (p[0] == 0x80) && (p[1] == 0x00); +} +#include #include #include #include @@ -23,8 +30,10 @@ namespace srtla::protocol { SRTHandler::SRTHandler(int srtla_socket, const struct sockaddr_storage &srt_addr, int epoll_fd, - connection::ConnectionRegistry ®istry) - : srtla_socket_(srtla_socket), srt_addr_(srt_addr), epoll_fd_(epoll_fd), registry_(registry) {} + connection::ConnectionRegistry ®istry, + utils::AuthRateLimiter &rate_limiter) + : srtla_socket_(srtla_socket), srt_addr_(srt_addr), epoll_fd_(epoll_fd), + registry_(registry), rate_limiter_(rate_limiter) {} void SRTHandler::handle_srt_data(connection::ConnectionGroupPtr group) { if (!group) { @@ -40,6 +49,27 @@ void SRTHandler::handle_srt_data(connection::ConnectionGroupPtr group) { return; } + // An SRT ACK from the server means media is flowing: the connection was + // accepted and stream auth passed. Mark the group established so a later + // SHUTDOWN is treated as a normal end-of-stream rather than a rejection. + if (is_srt_ack(buf, n)) { + group->mark_established(); + } + + // Detect a failed/rejected connection and throttle the source IP. Two + // shapes: a libsrt-native handshake rejection (type >= failure base), or + // — as srt-live-server does — the server accepts the handshake then closes + // the socket (SHUTDOWN) before the session is established because stream + // auth failed. We still relay the packet below so the client sees it, then + // tear the group down (see end of function). + bool failed_auth = is_srt_handshake_reject(buf, n) || + (is_srt_shutdown(buf, n) && !group->is_established()); + if (failed_auth) { + rate_limiter_.record_failure(group->last_address(), ::time(nullptr)); + spdlog::warn("[Group: {}] SRT connection rejected before established; recorded auth failure", + static_cast(group.get())); + } + // Broadcast ACKs and NAKs to all connections to ensure they reach the // sender even if some connections are dead. Other packets go to last_address. if (is_srt_ack(buf, n) || is_srt_nak(buf, n)) { @@ -91,6 +121,18 @@ void SRTHandler::handle_srt_data(connection::ConnectionGroupPtr group) { static_cast(group.get())); } } + + if (failed_auth) { + // The client has now been sent the rejection; reclaim the group's slot + // immediately rather than waiting for it to time out, so repeated + // failed-auth attempts cannot tie up the group table. Safe here: we + // hold a strong ref via the by-value `group` param (the object outlives + // this call), and the main loop stops using stale epoll pointers once + // the group count shrinks. + spdlog::info("[Group: {}] Tearing down failed-auth group", + static_cast(group.get())); + remove_group(group); + } } bool SRTHandler::forward_to_srt_server(connection::ConnectionGroupPtr group, const char *buffer, int length) { diff --git a/src/protocol/srt_handler.h b/src/protocol/srt_handler.h index 7b609b7..f0ae92f 100644 --- a/src/protocol/srt_handler.h +++ b/src/protocol/srt_handler.h @@ -3,6 +3,7 @@ #include #include "../connection/connection_registry.h" +#include "../utils/auth_rate_limiter.h" #include "../utils/network_utils.h" namespace srtla::protocol { @@ -12,7 +13,8 @@ class SRTHandler { SRTHandler(int srtla_socket, const struct sockaddr_storage &srt_addr, int epoll_fd, - connection::ConnectionRegistry ®istry); + connection::ConnectionRegistry ®istry, + utils::AuthRateLimiter &rate_limiter); void handle_srt_data(connection::ConnectionGroupPtr group); bool forward_to_srt_server(connection::ConnectionGroupPtr group, const char *buffer, int length); @@ -25,6 +27,7 @@ class SRTHandler { struct sockaddr_storage srt_addr_ {}; int epoll_fd_; connection::ConnectionRegistry ®istry_; + utils::AuthRateLimiter &rate_limiter_; }; } // namespace srtla::protocol diff --git a/src/protocol/srtla_handler.cpp b/src/protocol/srtla_handler.cpp index 901f03f..90a61e5 100644 --- a/src/protocol/srtla_handler.cpp +++ b/src/protocol/srtla_handler.cpp @@ -51,11 +51,13 @@ inline bool is_duplicate_nak(ConnectionGroupPtr group, const char *buffer, int l SRTLAHandler::SRTLAHandler(int srtla_socket, connection::ConnectionRegistry ®istry, SRTHandler &srt_handler, - quality::MetricsCollector &metrics_collector) + quality::MetricsCollector &metrics_collector, + utils::AuthRateLimiter &rate_limiter) : srtla_socket_(srtla_socket), registry_(registry), srt_handler_(srt_handler), - metrics_(metrics_collector) {} + metrics_(metrics_collector), + rate_limiter_(rate_limiter) {} int SRTLAHandler::process_packets(time_t ts) { // Pre-allocate buffers for batch receive @@ -137,6 +139,9 @@ void SRTLAHandler::process_single_packet(const char *buf, int n, return; } + // Real SRT traffic: promote the group out of "pending" so it is no longer + // subject to aggressive ghost-group reaping or eviction under a REG1 flood. + group->mark_data_seen(); group->set_last_address(*srtla_addr); metrics_.on_packet_received(conn, static_cast(n)); @@ -190,7 +195,25 @@ void SRTLAHandler::send_keepalive(const ConnectionPtr &conn, time_t ts) { } int SRTLAHandler::register_group(const struct sockaddr_storage *addr, const char *buffer, time_t ts) { - if (registry_.groups().size() >= MAX_GROUPS) { + // Refuse registrations from a source IP that recently failed SRT auth + // repeatedly. Blunts a client brute forcing streamid/passphrase and stops + // it from churning ghost groups, without affecting honest broadcasters. + if (rate_limiter_.is_blocked(*addr, ts)) { + uint16_t header = htobe16(SRTLA_TYPE_REG_ERR); + pad_sendto(srtla_socket_, &header, sizeof(header), 0, + reinterpret_cast(addr), kAddrLen); + spdlog::warn("[{}:{}] Group registration refused: source throttled for repeated auth failures", + print_addr(const_cast(reinterpret_cast(addr))), + port_no(const_cast(reinterpret_cast(addr)))); + return -1; + } + + // When the group table is full, try to reclaim a slot from a ghost group + // (registered but never streamed) before rejecting. This keeps an + // unauthenticated REG1 flood from locking out the real broadcaster: its + // REG1 evicts the oldest ghost, and once it completes REG2 and starts + // streaming the group is marked as having data and is no longer evictable. + if (registry_.groups().size() >= MAX_GROUPS && !registry_.evict_oldest_pending_group()) { uint16_t header = htobe16(SRTLA_TYPE_REG_ERR); pad_sendto(srtla_socket_, &header, sizeof(header), 0, reinterpret_cast(addr), kAddrLen); diff --git a/src/protocol/srtla_handler.h b/src/protocol/srtla_handler.h index 7e30b4a..0365f93 100644 --- a/src/protocol/srtla_handler.h +++ b/src/protocol/srtla_handler.h @@ -3,6 +3,7 @@ #include "srt_handler.h" #include "../connection/connection_registry.h" #include "../quality/metrics_collector.h" +#include "../utils/auth_rate_limiter.h" #include "../utils/nak_dedup.h" namespace srtla::protocol { @@ -15,7 +16,8 @@ class SRTLAHandler { SRTLAHandler(int srtla_socket, connection::ConnectionRegistry ®istry, SRTHandler &srt_handler, - quality::MetricsCollector &metrics_collector); + quality::MetricsCollector &metrics_collector, + utils::AuthRateLimiter &rate_limiter); // Process multiple packets in a batch using recvmmsg int process_packets(time_t ts); @@ -48,6 +50,7 @@ class SRTLAHandler { connection::ConnectionRegistry ®istry_; SRTHandler &srt_handler_; quality::MetricsCollector &metrics_; + utils::AuthRateLimiter &rate_limiter_; }; } // namespace srtla::protocol diff --git a/src/receiver_config.h b/src/receiver_config.h index bfaaa4a..c31865a 100644 --- a/src/receiver_config.h +++ b/src/receiver_config.h @@ -20,11 +20,25 @@ inline constexpr int MAX_GROUPS = 200; inline constexpr int CLEANUP_PERIOD = 3; inline constexpr int GROUP_TIMEOUT = 30; +// Groups that registered (REG1) but never forwarded real SRT data are reaped +// aggressively. A legitimate broadcaster completes REG2 and starts the SRT +// handshake within a fraction of a second, so this only targets the "ghost" +// groups left behind by an unauthenticated REG1 flood (resource-exhaustion DoS). +inline constexpr int PENDING_GROUP_TIMEOUT = 5; inline constexpr int CONN_TIMEOUT = 15; inline constexpr int KEEPALIVE_PERIOD = 1; inline constexpr int RECOVERY_CHANCE_PERIOD = 5; +// Per-source-IP throttle for SRT authentication failures. srtla_rec relays the +// SRT handshake but never authenticates itself; when the SRT server rejects a +// handshake we count it against the source IP and refuse new registrations once +// it crosses the threshold within the window. Tuned leniently so a mistyped +// passphrase or several broadcasters behind one NAT are not locked out. +inline constexpr int AUTH_FAIL_THRESHOLD = 5; // failures within window to trip +inline constexpr int AUTH_FAIL_WINDOW = 60; // seconds +inline constexpr int AUTH_FAIL_COOLDOWN = 60; // seconds blocked once tripped + inline constexpr int CONN_QUALITY_EVAL_PERIOD = 5; inline constexpr double MIN_ACCEPTABLE_TOTAL_BANDWIDTH_KBPS = 1000.0; inline constexpr int MAX_ERROR_POINTS = 40; diff --git a/src/receiver_main.cpp b/src/receiver_main.cpp index d38f666..d5a6843 100644 --- a/src/receiver_main.cpp +++ b/src/receiver_main.cpp @@ -18,6 +18,7 @@ #include "quality/metrics_collector.h" #include "quality/quality_evaluator.h" #include "receiver_config.h" +#include "utils/auth_rate_limiter.h" #include "utils/network_utils.h" extern "C" { @@ -151,10 +152,11 @@ int main(int argc, char **argv) { srtla::connection::ConnectionRegistry registry; srtla::quality::MetricsCollector metrics_collector; + srtla::utils::AuthRateLimiter rate_limiter; srtla::protocol::SRTHandler srt_handler(srtla_sock, srt_addr, epoll_fd, - registry); + registry, rate_limiter); srtla::protocol::SRTLAHandler srtla_handler(srtla_sock, registry, srt_handler, - metrics_collector); + metrics_collector, rate_limiter); srtla::quality::QualityEvaluator quality_evaluator; srtla::quality::LoadBalancer load_balancer; @@ -202,6 +204,7 @@ int main(int argc, char **argv) { } registry.cleanup_inactive(ts, keepalive_callback); + rate_limiter.cleanup(ts); for (auto &group : registry.groups()) { quality_evaluator.evaluate_group(group, ts); load_balancer.adjust_weights(group, ts); diff --git a/src/utils/auth_rate_limiter.cpp b/src/utils/auth_rate_limiter.cpp new file mode 100644 index 0000000..9cd6798 --- /dev/null +++ b/src/utils/auth_rate_limiter.cpp @@ -0,0 +1,62 @@ +#include "auth_rate_limiter.h" + +#include + +#include + +#include "../receiver_config.h" + +namespace srtla::utils { + +std::string AuthRateLimiter::key_for(const struct sockaddr_storage &addr) { + if (addr.ss_family == AF_INET6) { + auto *a = reinterpret_cast(&addr); + return std::string("6:") + + std::string(reinterpret_cast(&a->sin6_addr), sizeof(a->sin6_addr)); + } + + auto *a = reinterpret_cast(&addr); + return std::string("4:") + + std::string(reinterpret_cast(&a->sin_addr), sizeof(a->sin_addr)); +} + +void AuthRateLimiter::record_failure(const struct sockaddr_storage &addr, time_t now) { + auto &entry = entries_[key_for(addr)]; + + if (entry.window_start == 0 || (now - entry.window_start) > AUTH_FAIL_WINDOW) { + entry.window_start = now; + entry.failures = 0; + } + + entry.failures++; + if (entry.failures >= AUTH_FAIL_THRESHOLD) { + entry.blocked_until = now + AUTH_FAIL_COOLDOWN; + entry.failures = 0; + entry.window_start = 0; + spdlog::warn("Source IP throttled after {} SRT auth failures (cooldown {}s)", + AUTH_FAIL_THRESHOLD, AUTH_FAIL_COOLDOWN); + } +} + +bool AuthRateLimiter::is_blocked(const struct sockaddr_storage &addr, time_t now) const { + auto it = entries_.find(key_for(addr)); + if (it == entries_.end()) { + return false; + } + return it->second.blocked_until > now; +} + +void AuthRateLimiter::cleanup(time_t now) { + for (auto it = entries_.begin(); it != entries_.end();) { + const Entry &e = it->second; + bool blocked = e.blocked_until > now; + bool window_active = e.window_start != 0 && (now - e.window_start) <= AUTH_FAIL_WINDOW; + if (!blocked && !window_active) { + it = entries_.erase(it); + } else { + ++it; + } + } +} + +} // namespace srtla::utils diff --git a/src/utils/auth_rate_limiter.h b/src/utils/auth_rate_limiter.h new file mode 100644 index 0000000..d49eee5 --- /dev/null +++ b/src/utils/auth_rate_limiter.h @@ -0,0 +1,43 @@ +#pragma once + +#include +#include +#include + +#include + +namespace srtla::utils { + +// Tracks SRT authentication failures per source IP and refuses new group +// registrations from an address that keeps failing within a window. +// +// srtla_rec is a dumb SRT relay: the actual auth (streamid/passphrase) happens +// at the SRT server, whose rejection flows back through the relay. Counting +// those rejections lets us throttle a source that is brute forcing or otherwise +// repeatedly failing auth, without affecting legitimate broadcasters. Keys are +// IP-only (port stripped) so an attacker cannot evade by rotating source ports. +class AuthRateLimiter { +public: + void record_failure(const struct sockaddr_storage &addr, time_t now); + bool is_blocked(const struct sockaddr_storage &addr, time_t now) const; + + // Drops stale entries; call periodically from the cleanup loop. + void cleanup(time_t now); + + // Tracked-IP count: read-only observability for the cleanup reclamation + // path (no other live counter is exposed); does not affect throttling. + std::size_t tracked_entry_count() const { return entries_.size(); } + +private: + struct Entry { + int failures = 0; + time_t window_start = 0; + time_t blocked_until = 0; + }; + + static std::string key_for(const struct sockaddr_storage &addr); + + std::unordered_map entries_; +}; + +} // namespace srtla::utils diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index b11c125..f6d2b78 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -66,6 +66,22 @@ target_link_libraries(test_reg_race PRIVATE receiver_core_obj common_obj) srtla_add_test(test_group_limits test_group_limits.cpp) target_link_libraries(test_group_limits PRIVATE receiver_core_obj common_obj) +# Family 6 (Task 1, RED): ghost-group eviction DoS hardening (upstream 7855012, +# pre-cherry-pick). Drives the real registration path (handler_harness.h) and the +# cleanup reaper (injected clock) to assert behavior absent on current main: +# never-streamed ghosts reaped at PENDING_GROUP_TIMEOUT, oldest-ghost eviction +# admits a registration under table pressure, and groups that forwarded real SRT +# data are promoted (neither reaped early nor evicted). RED until the fix lands. +# Defined inline (like test_registration_handshake) so a "ghost_group." TEST_PREFIX +# makes `ctest -R ghost_group` select exactly this suite. Links like test_group_limits. +add_executable(test_ghost_group_eviction test_ghost_group_eviction.cpp) +target_link_libraries(test_ghost_group_eviction + PRIVATE GTest::gtest_main receiver_core_obj common_obj) +target_include_directories(test_ghost_group_eviction PRIVATE ${CMAKE_SOURCE_DIR}/src) +target_compile_features(test_ghost_group_eviction PRIVATE cxx_std_17) +target_compile_options(test_ghost_group_eviction PRIVATE -Wall -Wextra) +gtest_discover_tests(test_ghost_group_eviction TEST_PREFIX "ghost_group.") + # Family 3: timeout / cleanup via ConnectionRegistry::cleanup_inactive(ts, cb). srtla_add_test(test_timeout_cleanup test_timeout_cleanup.cpp) target_link_libraries(test_timeout_cleanup PRIVATE receiver_core_obj common_obj) @@ -107,3 +123,16 @@ find_package(Threads REQUIRED) target_link_libraries(test_telemetry_emit PRIVATE Threads::Threads) target_compile_definitions(test_telemetry_emit PRIVATE TELEMETRY_GOLDEN_DIR="${CMAKE_CURRENT_SOURCE_DIR}/golden") + +# Per-IP SRT auth-failure throttle (upstream 39e324a) + is_srt_shutdown classifier. +# Pure time-as-parameter logic: no clock injection, no real waits. Links +# receiver_core_obj (AuthRateLimiter) and common_obj (is_srt_shutdown). +# Inline registration with a "auth_rate_limiter." TEST_PREFIX so +# `ctest -R auth_rate_limiter` selects exactly this suite (ghost_group idiom). +add_executable(test_auth_rate_limiter test_auth_rate_limiter.cpp) +target_link_libraries(test_auth_rate_limiter + PRIVATE GTest::gtest_main receiver_core_obj common_obj) +target_include_directories(test_auth_rate_limiter PRIVATE ${CMAKE_SOURCE_DIR}/src) +target_compile_features(test_auth_rate_limiter PRIVATE cxx_std_17) +target_compile_options(test_auth_rate_limiter PRIVATE -Wall -Wextra) +gtest_discover_tests(test_auth_rate_limiter TEST_PREFIX "auth_rate_limiter.") diff --git a/tests/handler_harness.h b/tests/handler_harness.h index b5ae755..c0d292f 100644 --- a/tests/handler_harness.h +++ b/tests/handler_harness.h @@ -207,8 +207,8 @@ class HandlerHarness { } epoll_fd_ = ::epoll_create1(0); srt_addr_ = loopback_addr(9); // discard port; unused on the reg/keepalive paths - srt_handler_ = std::make_unique(recv_sock_, srt_addr_, epoll_fd_, registry_); - handler_ = std::make_unique(recv_sock_, registry_, *srt_handler_, metrics_); + srt_handler_ = std::make_unique(recv_sock_, srt_addr_, epoll_fd_, registry_, rate_limiter_); + handler_ = std::make_unique(recv_sock_, registry_, *srt_handler_, metrics_, rate_limiter_); } ~HandlerHarness() { @@ -250,6 +250,7 @@ class HandlerHarness { struct sockaddr_storage srt_addr_ {}; connection::ConnectionRegistry registry_; quality::MetricsCollector metrics_; + utils::AuthRateLimiter rate_limiter_; std::unique_ptr srt_handler_; std::unique_ptr handler_; }; diff --git a/tests/test_auth_rate_limiter.cpp b/tests/test_auth_rate_limiter.cpp new file mode 100644 index 0000000..76d54f1 --- /dev/null +++ b/tests/test_auth_rate_limiter.cpp @@ -0,0 +1,186 @@ +/* + srtla - SRT transport proxy with link aggregation + Copyright (C) 2026 CeraLive + + Locks the per-source-IP SRT auth-failure throttle introduced by upstream + cherry-pick 39e324a (src/utils/auth_rate_limiter.{cpp,h}) plus the + is_srt_shutdown classifier that drives the receiver's failed-auth signal. + + The limiter takes wall-clock time as an explicit `time_t now` parameter on + every method, so these tests advance time by passing values -- no real + waits, no clock injection needed. Thresholds (5 failures / 60s window / 60s + cooldown) are the picked values, pinned by the static_asserts below; the + behavioral cases assert the algorithm relative to those named constants. +*/ + +#include + +#include +#include +#include + +#include +#include + +#include "receiver_config.h" +#include "utils/auth_rate_limiter.h" + +extern "C" { +#include "common.h" +} + +using srtla::utils::AuthRateLimiter; + +static_assert(srtla::AUTH_FAIL_THRESHOLD == 5, "picked threshold (39e324a)"); +static_assert(srtla::AUTH_FAIL_WINDOW == 60, "picked window seconds (39e324a)"); +static_assert(srtla::AUTH_FAIL_COOLDOWN == 60, "picked cooldown seconds (39e324a)"); + +namespace { + +constexpr int kThreshold = srtla::AUTH_FAIL_THRESHOLD; +constexpr int kWindow = srtla::AUTH_FAIL_WINDOW; +constexpr int kCooldown = srtla::AUTH_FAIL_COOLDOWN; + +// Non-zero base so test timestamps never collide with the window_start==0 +// "uninitialized" sentinel inside the limiter. +constexpr time_t kBase = 100000; + +struct sockaddr_storage make_addr_v4(const char *ip, uint16_t port) { + struct sockaddr_storage ss {}; + auto *a = reinterpret_cast(&ss); + a->sin_family = AF_INET; + a->sin_port = htons(port); + inet_pton(AF_INET, ip, &a->sin_addr); + return ss; +} + +void make_control_packet(uint8_t *out, uint16_t srt_type) { + std::memset(out, 0, 16); + uint16_t type_be = htons(srt_type); + std::memcpy(out, &type_be, sizeof(type_be)); +} + +} // namespace + +// (a) record_failure accrues per source IP; below threshold stays allowed. +TEST(AuthRateLimiter, RecordFailureCountsPerSourceIpBelowThreshold) { + AuthRateLimiter rl; + auto attacker = make_addr_v4("203.0.113.10", 40000); + for (int i = 0; i < kThreshold - 1; ++i) { + rl.record_failure(attacker, kBase + i); + } + EXPECT_FALSE(rl.is_blocked(attacker, kBase + kThreshold)); + EXPECT_EQ(rl.tracked_entry_count(), 1u); + EXPECT_FALSE(rl.is_blocked(make_addr_v4("203.0.113.99", 40000), kBase)); +} + +// (b) 5 failures inside the 60s window -> the 6th registration is BLOCKED. +TEST(AuthRateLimiter, FifthFailureInWindowBlocksSixthRegistration) { + AuthRateLimiter rl; + auto attacker = make_addr_v4("203.0.113.10", 40000); + for (int i = 0; i < kThreshold; ++i) { + rl.record_failure(attacker, kBase + i); + } + EXPECT_TRUE(rl.is_blocked(attacker, kBase + kThreshold)); +} + +// (c) 5 failures spread so no 5 land in any 60s window -> never blocked. +TEST(AuthRateLimiter, FailuresSpreadAcrossSlidingWindowNeverBlock) { + AuthRateLimiter rl; + auto attacker = make_addr_v4("203.0.113.10", 40000); + const time_t offsets[] = {0, 30, 65, 95, 130}; + for (time_t off : offsets) { + rl.record_failure(attacker, kBase + off); + } + EXPECT_FALSE(rl.is_blocked(attacker, kBase + 130)); + EXPECT_FALSE(rl.is_blocked(attacker, kBase + 200)); +} + +// (d) A blocked IP unblocks once the 60s cooldown elapses (strict expiry). +TEST(AuthRateLimiter, BlockedIpUnblocksAfterCooldownElapses) { + AuthRateLimiter rl; + auto attacker = make_addr_v4("203.0.113.10", 40000); + for (int i = 0; i < kThreshold; ++i) { + rl.record_failure(attacker, kBase + i); + } + const time_t tripped = kBase + (kThreshold - 1); + EXPECT_TRUE(rl.is_blocked(attacker, tripped + 1)); + EXPECT_TRUE(rl.is_blocked(attacker, tripped + kCooldown - 1)); + EXPECT_FALSE(rl.is_blocked(attacker, tripped + kCooldown)); + EXPECT_FALSE(rl.is_blocked(attacker, tripped + kCooldown + 1)); +} + +// (e) Keys are IP-only: rotating the source port does not evade the block. +TEST(AuthRateLimiter, IpKeyingPortRotationDoesNotEvadeBlock) { + AuthRateLimiter rl; + for (int i = 0; i < kThreshold; ++i) { + rl.record_failure(make_addr_v4("203.0.113.10", 40000 + i), kBase + i); + } + EXPECT_TRUE(rl.is_blocked(make_addr_v4("203.0.113.10", 55555), kBase + kThreshold)); + EXPECT_EQ(rl.tracked_entry_count(), 1u); +} + +// (f) Distinct IPs are independent: a tripped attacker does not lock out a +// separate neighbor sitting at 4 (below-threshold) failures. +TEST(AuthRateLimiter, DistinctIpsAreIndependentNeighborNotLockedOut) { + AuthRateLimiter rl; + auto attacker = make_addr_v4("203.0.113.10", 40000); + auto neighbor = make_addr_v4("203.0.113.11", 40000); + for (int i = 0; i < kThreshold; ++i) { + rl.record_failure(attacker, kBase + i); + } + for (int i = 0; i < kThreshold - 1; ++i) { + rl.record_failure(neighbor, kBase + i); + } + EXPECT_TRUE(rl.is_blocked(attacker, kBase + kThreshold)); + EXPECT_FALSE(rl.is_blocked(neighbor, kBase + kThreshold)); +} + +// (g) cleanup reclaims stale entries, but retains active windows / live blocks. +TEST(AuthRateLimiter, StaleEntryCleanupReclaimsExpiredEntries) { + AuthRateLimiter rl; + + auto idle = make_addr_v4("203.0.113.10", 40000); + rl.record_failure(idle, kBase); + EXPECT_EQ(rl.tracked_entry_count(), 1u); + rl.cleanup(kBase + 1); + EXPECT_EQ(rl.tracked_entry_count(), 1u); + rl.cleanup(kBase + kWindow + 1); + EXPECT_EQ(rl.tracked_entry_count(), 0u); + + auto blocked = make_addr_v4("203.0.113.20", 40000); + for (int i = 0; i < kThreshold; ++i) { + rl.record_failure(blocked, kBase + i); + } + const time_t tripped = kBase + (kThreshold - 1); + rl.cleanup(tripped + 1); + EXPECT_EQ(rl.tracked_entry_count(), 1u); + rl.cleanup(tripped + kCooldown); + EXPECT_EQ(rl.tracked_entry_count(), 0u); +} + +// (h) is_srt_shutdown recognizes a SHUTDOWN and rejects other control packets. +TEST(IsSrtShutdown, ClassifiesShutdownAndRejectsOtherControlPackets) { + uint8_t pkt[16]; + + make_control_packet(pkt, SRT_TYPE_SHUTDOWN); + EXPECT_TRUE(is_srt_shutdown(pkt, sizeof(pkt))); + + make_control_packet(pkt, SRT_TYPE_ACK); + EXPECT_FALSE(is_srt_shutdown(pkt, sizeof(pkt))); + make_control_packet(pkt, SRT_TYPE_NAK); + EXPECT_FALSE(is_srt_shutdown(pkt, sizeof(pkt))); + make_control_packet(pkt, SRT_TYPE_HANDSHAKE); + EXPECT_FALSE(is_srt_shutdown(pkt, sizeof(pkt))); +} + +TEST(IsSrtShutdown, ShortPacketIsNotShutdown) { + uint8_t one_byte[1] = {0x80}; + EXPECT_FALSE(is_srt_shutdown(one_byte, 1)); +} + +// SRT data packets clear the high bit of byte 0 -> never a control SHUTDOWN. +TEST(IsSrtShutdown, DataPacketIsNotShutdown) { + uint8_t data[16] = {0x00, 0x00, 0x00, 0x05}; + EXPECT_FALSE(is_srt_shutdown(data, sizeof(data))); +} diff --git a/tests/test_ghost_group_eviction.cpp b/tests/test_ghost_group_eviction.cpp new file mode 100644 index 0000000..ad7e63c --- /dev/null +++ b/tests/test_ghost_group_eviction.cpp @@ -0,0 +1,340 @@ +/* + srtla - SRT transport proxy with link aggregation + Copyright (C) 2026 CeraLive + + RED tests — ghost-group eviction (pre-cherry-pick of upstream 7855012). + + Unauthenticated REG1 packets create connection groups before any SRT + handshake. Today an empty group is held for the full GROUP_TIMEOUT (30s) and + the table is hard-capped at MAX_GROUPS with REG_ERR at the cap, so an + attacker can flood "ghost" groups (registered, never streamed) and lock out + the real broadcaster. Commit 7855012 hardens this: + + * a group that never forwarded real SRT data is reaped at the shorter + PENDING_GROUP_TIMEOUT (5s) instead of GROUP_TIMEOUT (30s); + * when the table is full, a new REG1 evicts the OLDEST ghost group before + rejecting, instead of returning REG_ERR outright; + * once a group forwards real SRT traffic it is promoted (mark_data_seen) + and is no longer reaped early or evictable, so active streams and + cellular reconnects survive a flood. + + This suite asserts that hardened behavior. It is RED on current main: the + behavior is absent, so the discriminating cases below fail today and go + green after the cherry-pick — with no edits to this file. + + Compile-on-main constraint. The cherry-pick adds new public symbols + (ConnectionGroup::has_seen_data/mark_data_seen, + ConnectionRegistry::evict_oldest_pending_group, and the + PENDING_GROUP_TIMEOUT constant). Referencing any of them by name would break + compilation on current main and turn "RED test" into "build failure". So the + suite is written strictly against the public API that exists BOTH before and + after the cherry-pick: + + * promotion (data_seen) is induced only through the production data path — + a real SRT data packet pushed through SRTLAHandler, the sole caller of + mark_data_seen() — never by touching the flag directly; + * PENDING_GROUP_TIMEOUT (5s) is mirrored locally as kPendingGroupTimeout; + * the reaper is driven by ConnectionRegistry::cleanup_inactive(ts, cb) + through its injected logical clock — no wall-clock waits anywhere. + + Tests that need to distinguish before/after behavior assert on observable + public effects (group survival, REG2 vs REG_ERR replies), so they read as a + behavioral spec and pass unmodified once the fix lands. +*/ + +#include + +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +#include "connection/connection.h" +#include "connection/connection_group.h" +#include "connection/connection_registry.h" +#include "handler_harness.h" +#include "receiver_config.h" + +extern "C" { +#include "common.h" +} + +using srtla::MAX_GROUPS; +using srtla::connection::Connection; +using srtla::connection::ConnectionGroup; +using srtla::connection::ConnectionGroupPtr; +using srtla::connection::ConnectionRegistry; +using srtla::test::Client; +using srtla::test::extract_full_id; +using srtla::test::HandlerHarness; +using srtla::test::make_client_id; +using srtla::test::pkt_type; + +namespace { + +// Local mirror of the PENDING_GROUP_TIMEOUT the cherry-pick adds to +// receiver_config.h (5s). Declared here so the suite compiles against current +// main, where the constant does not yet exist. Kept well below GROUP_TIMEOUT +// (30s); every reaper advance in this file stays inside that window so a green +// result can only come from the pending-timeout path, never the 30s fallback. +constexpr time_t kPendingGroupTimeout = 5; + +// Strictly-increasing, far-apart logical time base per test. cleanup_inactive +// keeps a process-static `last_run` (shared across every ConnectionRegistry +// instance), so each test must start above the previous test's timestamps for +// its first cleanup to clear the once-per-CLEANUP_PERIOD throttle regardless of +// execution order (including --gtest_shuffle). Mirrors test_timeout_cleanup.cpp. +time_t fresh_base() { + static std::atomic base{1'000'000}; + return base.fetch_add(1'000'000); +} + +// A bare group with a unique id, no connections, no SRT socket and never having +// forwarded data: a "ghost" left behind by a REG1 flood. created_at is explicit +// so eviction's "oldest" choice is deterministic. +ConnectionGroupPtr make_ghost_group(uint32_t seed, time_t created_at) { + auto id = make_client_id(seed); + return std::make_shared(reinterpret_cast(id.data()), created_at); +} + +bool group_present(const ConnectionRegistry ®, const ConnectionGroupPtr &g) { + const auto &groups = reg.groups(); + return std::find(groups.begin(), groups.end(), g) != groups.end(); +} + +// Run a full REG1/REG2 handshake plus one real SRT data packet through the live +// handler so the resulting group is promoted via the production mark_data_seen +// path (post-cherry-pick) — without this file naming that method. The data +// packet is a minimal SRT data packet (high bit clear, >= SRT_MIN_LEN, not a +// REG/keepalive frame), so process_single_packet treats it as forwardable +// traffic. Returns the registered group with its connection still attached. +ConnectionGroupPtr register_streaming_group(HandlerHarness &h, Client &link, + uint32_t seed, time_t ts) { + link.send_reg1(make_client_id(seed)); + h.pump(ts); + std::vector reg2; + EXPECT_TRUE(link.recv_one(reg2)) << "REG1 should be answered with REG2"; + EXPECT_EQ(pkt_type(reg2), SRTLA_TYPE_REG2); + auto full_id = extract_full_id(reg2); + + link.send_reg2(full_id); + h.pump(ts); + std::vector reg3; + EXPECT_TRUE(link.recv_one(reg3)) << "REG2 should be answered with REG3"; + EXPECT_EQ(pkt_type(reg3), SRTLA_TYPE_REG3); + + ConnectionGroupPtr group = + h.registry().find_group_by_id(reinterpret_cast(full_id.data())); + EXPECT_NE(group, nullptr) << "handshake must leave a registered group"; + + // One real SRT data packet from the registered link. On the post-cherry-pick + // handler this calls group->mark_data_seen(); on current main it only runs + // metrics + forward. Either way it exercises the same public data path. + std::array data{}; + uint32_t sn = htobe32(1u); // high bit clear => SRT data packet, sn = 1 + std::memcpy(data.data(), &sn, sizeof(sn)); + (void)::send(link.fd(), data.data(), data.size(), 0); + h.pump(ts); + + // Confirm the packet actually reached the forward path on both versions, so + // a later "data-seen group survives" assertion can only be explained by the + // promotion, not by the setup silently dropping the packet. + if (group && !group->connections().empty()) { + EXPECT_GE(group->connections().front()->stats().packets_received, 1u) + << "the SRT data packet must reach SRTLAHandler's forward path"; + } + return group; +} + +// A loopback connection with a distinct source port, used to make a filler +// group non-evictable (connections() non-empty) without standing up a real +// handshake. Stands in for an active stream in the "nothing safe to evict" case. +std::shared_ptr make_loopback_conn(uint16_t port, time_t ts) { + struct sockaddr_storage ss; + std::memset(&ss, 0, sizeof(ss)); + auto *in = reinterpret_cast(&ss); + in->sin_family = AF_INET; + in->sin_addr.s_addr = htonl(INADDR_LOOPBACK); + in->sin_port = htons(port); + return std::make_shared(ss, ts); +} + +} // namespace + +// (a) A ghost group (registered, never forwarded SRT data) is reaped at +// PENDING_GROUP_TIMEOUT, not held for the full GROUP_TIMEOUT. +// RED on main: today an empty group survives until 30s, so it is still present +// at t0+6. +TEST(GhostGroupEviction, GhostGroupReapedAtPendingTimeout) { + time_t t0 = fresh_base(); + ConnectionRegistry reg; + reg.add_group(make_ghost_group(0x6057u, t0)); // no connections, never streamed + + // 6s of life: past PENDING_GROUP_TIMEOUT(5), far short of GROUP_TIMEOUT(30). + reg.cleanup_inactive(t0 + kPendingGroupTimeout + 1, nullptr); + + EXPECT_EQ(reg.groups().size(), 0u) + << "a never-streamed ghost group must be reaped at PENDING_GROUP_TIMEOUT, " + "not held for the full GROUP_TIMEOUT"; +} + +// (a, lower boundary) A ghost younger than PENDING_GROUP_TIMEOUT is NOT reaped +// yet — the aggressive reap must not fire immediately. Passes before and after +// the fix; pins that the pending window is a window, not a zero-grace drop. +TEST(GhostGroupEviction, GhostGroupRetainedBeforePendingTimeout) { + time_t t0 = fresh_base(); + ConnectionRegistry reg; + reg.add_group(make_ghost_group(0x6057u, t0)); + + // 4s: still inside PENDING_GROUP_TIMEOUT(5). + reg.cleanup_inactive(t0 + kPendingGroupTimeout - 1, nullptr); + + EXPECT_EQ(reg.groups().size(), 1u) + << "a ghost younger than PENDING_GROUP_TIMEOUT must not be reaped yet"; +} + +// (b) A flood that fills the table with ghost groups must not lock out a new +// registration: the oldest ghost is evicted and the REG1 is admitted (REG2). +// RED on main: at the cap a new REG1 is rejected with REG_ERR and no ghost is +// evicted. +TEST(GhostGroupEviction, FloodEvictsOldestGhostToAdmitRegistration) { + time_t t0 = fresh_base(); + HandlerHarness h; + + for (int i = 0; i < MAX_GROUPS; ++i) { + // created_at increases with i, so groups().front() is the oldest ghost. + h.registry().add_group(make_ghost_group(static_cast(i + 1), t0 + i)); + } + ASSERT_EQ(h.registry().groups().size(), static_cast(MAX_GROUPS)); + ConnectionGroupPtr oldest_ghost = h.registry().groups().front(); + + Client newcomer = h.make_client(); + newcomer.send_reg1(make_client_id(0xFEEDu)); + h.pump(t0 + 1); // ts is just the new group's created_at; no reaper involved + + std::vector reply; + ASSERT_TRUE(newcomer.recv_one(reply)) << "registration at cap must not be silently dropped"; + EXPECT_EQ(pkt_type(reply), SRTLA_TYPE_REG2) + << "a ghost flood must not lock out a new registration: the oldest ghost is " + "evicted and the REG1 admitted (was REG_ERR before the fix)"; + EXPECT_FALSE(group_present(h.registry(), oldest_ghost)) + << "the oldest ghost group is the eviction victim"; + EXPECT_EQ(h.registry().groups().size(), static_cast(MAX_GROUPS)) + << "table stays at the cap: one ghost out, one registration in"; +} + +// (c) A group that has forwarded real SRT data is promoted: it survives the +// pending-group reap while a never-streamed ghost beside it is reaped at the +// same 5s pass. RED on main: today the ghost is still alive at t0+6 (held to +// 30s), so the "ghost reaped" expectation fails. +TEST(GhostGroupEviction, DataSeenGroupSurvivesPendingReapWhileGhostReaped) { + time_t t0 = fresh_base(); + HandlerHarness h; + Client streamer = h.make_client(); + + // A promoted (data-seen) group and a never-streamed ghost, both born at t0. + ConnectionGroupPtr streamed = register_streaming_group(h, streamer, 0x57A1u, t0); + ASSERT_NE(streamed, nullptr); + ASSERT_FALSE(streamed->connections().empty()); + // Cellular reconnect: the uplink drops, the group goes empty but stays known. + streamed->connections().clear(); + + ConnectionGroupPtr ghost = make_ghost_group(0x6057u, t0); + h.registry().add_group(ghost); + + // One reap pass at t0+6: past PENDING_GROUP_TIMEOUT(5), short of GROUP_TIMEOUT(30). + h.registry().cleanup_inactive(t0 + kPendingGroupTimeout + 1, nullptr); + + EXPECT_TRUE(group_present(h.registry(), streamed)) + << "a group that forwarded real SRT data is promoted and must survive the " + "pending-group reap (kept until GROUP_TIMEOUT like any active stream)"; + EXPECT_FALSE(group_present(h.registry(), ghost)) + << "the never-streamed ghost beside it must be reaped at PENDING_GROUP_TIMEOUT"; +} + +// (d) Eviction under table pressure never touches a data-seen group — even when +// it is the OLDEST group of all (reconnect-safety). The real broadcaster +// streamed, lost its link, and must keep its slot while a ghost is evicted for +// the newcomer. RED on main: the newcomer is rejected with REG_ERR and no ghost +// is evicted. +TEST(GhostGroupEviction, EvictionSkipsDataSeenGroupUnderTablePressure) { + time_t t0 = fresh_base(); + HandlerHarness h; + Client streamer = h.make_client(); + + // Oldest group of all: registered and streamed at t0, then went empty. + ConnectionGroupPtr streamed = register_streaming_group(h, streamer, 0xB0A7u, t0); + ASSERT_NE(streamed, nullptr); + ASSERT_FALSE(streamed->connections().empty()); + streamed->connections().clear(); + + // Fill the rest of the table with never-streamed ghosts, all NEWER than the + // broadcaster (created_at t0+1 .. these are creation stamps, not a reaper + // clock advance — cleanup_inactive is never called in this test). + for (int i = 0; i < MAX_GROUPS - 1; ++i) { + h.registry().add_group(make_ghost_group(static_cast(i + 1), t0 + 1 + i)); + } + ASSERT_EQ(h.registry().groups().size(), static_cast(MAX_GROUPS)); + + // The eviction target is the oldest ghost (not the older, data-seen group). + ConnectionGroupPtr oldest_ghost; + for (const auto &g : h.registry().groups()) { + if (g == streamed) { + continue; + } + if (!oldest_ghost || g->created_at() < oldest_ghost->created_at()) { + oldest_ghost = g; + } + } + ASSERT_NE(oldest_ghost, nullptr); + + Client newcomer = h.make_client(); + newcomer.send_reg1(make_client_id(0xFEEDu)); + h.pump(t0 + 1); + + std::vector reply; + ASSERT_TRUE(newcomer.recv_one(reply)) << "registration at cap must get a reply"; + EXPECT_EQ(pkt_type(reply), SRTLA_TYPE_REG2) + << "table full of ghosts: the newcomer must evict a ghost and be admitted"; + EXPECT_TRUE(group_present(h.registry(), streamed)) + << "the data-seen broadcaster must never be evicted, even as the oldest group " + "(reconnect-safety)"; + EXPECT_FALSE(group_present(h.registry(), oldest_ghost)) + << "the oldest never-streamed ghost is the eviction victim"; +} + +// (d, contract guard) When the table is full but holds NO ghost — every group is +// actively connected or has streamed — there is nothing safe to evict, so a new +// registration is still refused with REG_ERR. Holds before AND after the fix: +// eviction reclaims only never-streamed ghosts and never steals a live slot. +TEST(GhostGroupEviction, AtMaxWithNoGhost_RegistrationStillRejected) { + time_t t0 = fresh_base(); + HandlerHarness h; + + for (int i = 0; i < MAX_GROUPS; ++i) { + ConnectionGroupPtr g = make_ghost_group(static_cast(i + 1), t0 + i); + // A live connection makes the group non-evictable (connections() non-empty), + // standing in for an active stream without 200 real handshakes. + g->add_connection(make_loopback_conn(static_cast(20000 + i), t0)); + h.registry().add_group(g); + } + ASSERT_EQ(h.registry().groups().size(), static_cast(MAX_GROUPS)); + + Client newcomer = h.make_client(); + newcomer.send_reg1(make_client_id(0xFEEDu)); + h.pump(t0 + 1); + + std::vector reply; + ASSERT_TRUE(newcomer.recv_one(reply)) << "registration at cap must get a reply"; + EXPECT_EQ(pkt_type(reply), SRTLA_TYPE_REG_ERR) + << "no ghost to reclaim => the REG_ERR cap contract is preserved"; + EXPECT_EQ(h.registry().groups().size(), static_cast(MAX_GROUPS)); +} diff --git a/tests/test_group_limits.cpp b/tests/test_group_limits.cpp index aa6f06d..e032d45 100644 --- a/tests/test_group_limits.cpp +++ b/tests/test_group_limits.cpp @@ -52,11 +52,16 @@ namespace { constexpr time_t kTs = 200000; -// A bare group with a unique id, no connections, no SRT socket — just enough to -// occupy a registry slot for the size guard. +// Filler for the MAX_GROUPS guard. mark_data_seen() is load-bearing: post-7855012 +// an empty group without data is an evictable ghost (reaped at PENDING_GROUP_TIMEOUT), +// so an unmarked filler would be evicted by the (MAX_GROUPS+1)-th REG1 instead of +// triggering REG_ERR. Marking it models the realistic "table full of real streams" +// state; the ghost-eviction path is covered by test_ghost_group_eviction.cpp. ConnectionGroupPtr make_mock_group(uint32_t seed) { auto id = make_client_id(seed); - return std::make_shared(reinterpret_cast(id.data()), kTs); + auto group = std::make_shared(reinterpret_cast(id.data()), kTs); + group->mark_data_seen(); + return group; } void fill_registry(HandlerHarness &h, int count) { diff --git a/tests/test_registration_handshake.cpp b/tests/test_registration_handshake.cpp index 949fe66..0c417c5 100644 --- a/tests/test_registration_handshake.cpp +++ b/tests/test_registration_handshake.cpp @@ -153,8 +153,8 @@ class RegHandshakeTest : public ::testing::Test { in->sin_addr.s_addr = htonl(INADDR_LOOPBACK); in->sin_port = htons(9999); - srt_handler_ = std::make_unique(srtla_fd_, srt_addr_, epoll_fd_, registry_); - handler_ = std::make_unique(srtla_fd_, registry_, *srt_handler_, metrics_); + srt_handler_ = std::make_unique(srtla_fd_, srt_addr_, epoll_fd_, registry_, rate_limiter_); + handler_ = std::make_unique(srtla_fd_, registry_, *srt_handler_, metrics_, rate_limiter_); } void TearDown() override { @@ -202,6 +202,7 @@ class RegHandshakeTest : public ::testing::Test { struct sockaddr_storage srt_addr_ {}; ConnectionRegistry registry_; MetricsCollector metrics_; + srtla::utils::AuthRateLimiter rate_limiter_; std::unique_ptr srt_handler_; std::unique_ptr handler_; std::vector client_fds_; diff --git a/tests/test_timeout_cleanup.cpp b/tests/test_timeout_cleanup.cpp index 92cbcd1..8f895f0 100644 --- a/tests/test_timeout_cleanup.cpp +++ b/tests/test_timeout_cleanup.cpp @@ -84,7 +84,9 @@ ConnectionPtr make_conn(uint16_t port, time_t last_received) { ConnectionGroupPtr make_group(time_t created_at) { std::array id{}; std::memcpy(id.data(), "timeout-cleanup-group", 21); - return std::make_shared(id.data(), created_at); + auto group = std::make_shared(id.data(), created_at); + group->mark_data_seen(); + return group; } } // namespace