feat(receiver): upstream hardening cherry-picks + test coverage (7855012 + 39e324a)

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

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")

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;

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);

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<ConnectionPtr> conns_;
     GroupIdentity identity_;
     time_t created_at_ = 0;
+    bool data_seen_ = false;
+    bool established_ = false;
     int srt_sock_ = -1;
     struct sockaddr_storage last_addr_ {};
 

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<void *>(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());
             }

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,

src/protocol/srt_handler.cpp

@@ -6,6 +6,13 @@
 #include "pad_sendto.h"
 
 #include <arpa/inet.h>
+
+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 <ctime>
 #include <fcntl.h>
 #include <sys/socket.h>
 #include <unistd.h>
@@ -23,8 +30,10 @@ namespace srtla::protocol {
 SRTHandler::SRTHandler(int srtla_socket,
                        const struct sockaddr_storage &srt_addr,
                        int epoll_fd,
-                       connection::ConnectionRegistry &registry)
-    : srtla_socket_(srtla_socket), srt_addr_(srt_addr), epoll_fd_(epoll_fd), registry_(registry) {}
+                       connection::ConnectionRegistry &registry,
+                       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<void *>(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<void *>(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<void *>(group.get()));
+        remove_group(group);
+    }
 }
 
 bool SRTHandler::forward_to_srt_server(connection::ConnectionGroupPtr group, const char *buffer, int length) {

src/protocol/srt_handler.h

@@ -3,6 +3,7 @@
 #include <sys/epoll.h>
 
 #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 &registry);
+               connection::ConnectionRegistry &registry,
+               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 &registry_;
+    utils::AuthRateLimiter &rate_limiter_;
 };
 
 } // namespace srtla::protocol

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 &registry,
                            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<size_t>(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<const struct sockaddr *>(addr), kAddrLen);
+        spdlog::warn("[{}:{}] Group registration refused: source throttled for repeated auth failures",
+                     print_addr(const_cast<struct sockaddr *>(reinterpret_cast<const struct sockaddr *>(addr))),
+                     port_no(const_cast<struct sockaddr *>(reinterpret_cast<const struct sockaddr *>(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<const struct sockaddr *>(addr), kAddrLen);

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 &registry,
                  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 &registry_;
     SRTHandler &srt_handler_;
     quality::MetricsCollector &metrics_;
+    utils::AuthRateLimiter &rate_limiter_;
 };
 
 } // namespace srtla::protocol

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;