Native UDP: hardening fixes (slot leak, transient errors, rotation, visitor unregister)

src/net/quic/quic_proxy_datagram_client_socket.cc

@@ -401,7 +401,11 @@ void QuicProxyDatagramClientSocket::OnStreamClosed(int result) {
   // They may otherwise re-enter OnIOComplete after this close notification.
   weak_factory_.InvalidateWeakPtrs();
   next_state_ = STATE_DISCONNECTED;
-  datagram_visitor_registered_ = false;
+  // Match Close(): do not clear the flag while leaving the visitor registered.
+  if (datagram_visitor_registered_) {
+    stream_handle_->UnregisterHttp3DatagramVisitor();
+    datagram_visitor_registered_ = false;
+  }
   connect_request_sent_ = false;
   awaiting_connect_response_ = false;
   read_buf_ = nullptr;

src/net/tools/naive/naive_connect_udp_backend_test_bin.cc

@@ -852,6 +852,22 @@ void TestTargetAndQueueLimits() {
                  128 &&
              association_cap_backend.stats_for_testing().capacity_drops == 1,
          "association packet queue admits 128 and drops the 129th");
+  const size_t targets_before_new_drop =
+      association_cap_backend.target_count_for_testing();
+  const size_t factory_calls_before_new_drop = association_states.size();
+  association_cap_backend.Send(
+      DatagramFor(net::Socks5UdpEndpoint{
+                      .type = net::Socks5UdpAddressType::kDomain,
+                      .host = "new-target-at-capacity.test",
+                      .port = 443,
+                  },
+                  {0x03}),
+      net::CompletionOnceCallback());
+  Expect(association_cap_backend.target_count_for_testing() ==
+                 targets_before_new_drop &&
+             association_states.size() == factory_calls_before_new_drop &&
+             association_cap_backend.stats_for_testing().capacity_drops == 2,
+         "first datagram admission drop does not construct or leak a target");
 
   auto byte_state = std::make_shared<ScriptedTunnelState>();
   net::NaiveConnectUdpDatagramBackend byte_cap_backend(

src/net/tools/naive/naive_connect_udp_datagram_backend.cc

@@ -90,12 +90,25 @@ int NaiveConnectUdpDatagramBackend::Send(
   }
   const Socks5UdpTargetKey key(datagram.destination);
   auto it = targets_.find(key);
+  const bool association_queue_full =
+      queued_datagram_count_ >=
+          Socks5UdpBackendLimits::kMaxQueuedDatagramsPerAssociation ||
+      datagram.payload.size() >
+          Socks5UdpBackendLimits::kMaxQueuedPayloadBytesPerAssociation -
+              queued_payload_bytes_;
   if (it == targets_.end()) {
     if (targets_.size() >= Socks5UdpBackendLimits::kMaxTargets) {
       ++stats_.capacity_drops;
       RecordCounterEvent("target_capacity_drop", stats_.capacity_drops);
       return OK;
     }
+    // Reject before factory/emplace so a full association queue cannot churn
+    // unique destinations into throwaway tunnels.
+    if (association_queue_full) {
+      ++stats_.capacity_drops;
+      RecordCounterEvent("queue_capacity_drop", stats_.capacity_drops);
+      return OK;
+    }
     std::unique_ptr<NaiveConnectUdpTargetTunnel> tunnel =
         tunnel_factory_.Run(context_, datagram.destination);
     if (!tunnel) {
@@ -122,11 +135,7 @@ int NaiveConnectUdpDatagramBackend::Send(
   }
   if (entry->outbound_queue.size() >=
           Socks5UdpBackendLimits::kMaxQueuedDatagramsPerTarget ||
-      queued_datagram_count_ >=
-          Socks5UdpBackendLimits::kMaxQueuedDatagramsPerAssociation ||
-      datagram.payload.size() >
-          Socks5UdpBackendLimits::kMaxQueuedPayloadBytesPerAssociation -
-              queued_payload_bytes_) {
+      association_queue_full) {
     ++stats_.capacity_drops;
     RecordCounterEvent("queue_capacity_drop", stats_.capacity_drops);
     return OK;

src/net/tools/naive/naive_proxy.cc

@@ -605,7 +605,7 @@ void NaiveProxy::CleanUpIdleConnections() {
     ClosePendingSocks(id, ERR_TIMED_OUT);
   }
   for (const auto& [id, association] : udp_association_by_id_) {
-    if (now - association->last_activity() > idle_timeout_) {
+    if (association->ShouldExpire(now, idle_timeout_, tunnel_timeout_)) {
       idle_udp_associations.push_back(id);
     }
   }

src/net/tools/naive/naive_socks5_udp_association_test_bin.cc

@@ -20,6 +20,7 @@
 #include "base/run_loop.h"
 #include "base/task/single_thread_task_executor.h"
 #include "base/task/single_thread_task_runner.h"
+#include "base/time/time.h"
 #include "net/base/io_buffer.h"
 #include "net/base/ip_address.h"
 #include "net/base/ip_endpoint.h"
@@ -224,6 +225,9 @@ struct RelayState {
   bool recv_pending = false;
   bool send_pending = false;
   bool closed = false;
+  scoped_refptr<net::IOBuffer> pending_recv_buffer;
+  net::IPEndPoint* pending_recv_address = nullptr;
+  net::CompletionOnceCallback pending_recv_callback;
   scoped_refptr<net::IOBuffer> pending_send_buffer;
   net::CompletionOnceCallback pending_send_callback;
 };
@@ -236,6 +240,12 @@ void QueueRelayData(const std::shared_ptr<RelayState>& state,
       .data = std::move(data), .source = source, .result = net::OK});
 }
 
+void QueueRelayResult(const std::shared_ptr<RelayState>& state, int result) {
+  CHECK_LT(result, 0);
+  state->reads.push_back(
+      RelayReadEvent{.data = {}, .source = {}, .result = result});
+}
+
 class ScriptedDatagramSocket final : public net::DatagramServerSocket {
  public:
   ScriptedDatagramSocket(std::shared_ptr<RelayState> state,
@@ -258,9 +268,9 @@ class ScriptedDatagramSocket final : public net::DatagramServerSocket {
     CHECK(!state_->recv_pending);
     if (state_->reads.empty()) {
       state_->recv_pending = true;
-      pending_recv_buffer_ = base::WrapRefCounted(buffer);
-      pending_recv_address_ = address;
-      pending_recv_callback_ = std::move(callback);
+      state_->pending_recv_buffer = base::WrapRefCounted(buffer);
+      state_->pending_recv_address = address;
+      state_->pending_recv_callback = std::move(callback);
       return net::ERR_IO_PENDING;
     }
 
@@ -308,11 +318,11 @@ class ScriptedDatagramSocket final : public net::DatagramServerSocket {
     }
     state_->closed = true;
     ++state_->close_calls;
-    if (pending_recv_callback_) {
+    if (state_->pending_recv_callback) {
       ++state_->pending_recv_cancellations;
-      pending_recv_callback_.Reset();
-      pending_recv_buffer_.reset();
-      pending_recv_address_ = nullptr;
+      state_->pending_recv_callback.Reset();
+      state_->pending_recv_buffer.reset();
+      state_->pending_recv_address = nullptr;
       state_->recv_pending = false;
     }
     if (state_->pending_send_callback) {
@@ -367,12 +377,35 @@ class ScriptedDatagramSocket final : public net::DatagramServerSocket {
  private:
   const std::shared_ptr<RelayState> state_;
   const net::IPEndPoint local_endpoint_;
-  scoped_refptr<net::IOBuffer> pending_recv_buffer_;
-  net::IPEndPoint* pending_recv_address_ = nullptr;
-  net::CompletionOnceCallback pending_recv_callback_;
   net::NetLogWithSource net_log_;
 };
 
+void CompleteRelayRead(const std::shared_ptr<RelayState>& state,
+                       std::vector<uint8_t> data,
+                       const net::IPEndPoint& source) {
+  CHECK(state->recv_pending);
+  CHECK(state->pending_recv_callback);
+  CHECK_LE(data.size(),
+           static_cast<size_t>(state->pending_recv_buffer->size()));
+  state->pending_recv_buffer->first(data.size()).copy_from(data);
+  *state->pending_recv_address = source;
+  state->recv_pending = false;
+  state->pending_recv_buffer.reset();
+  state->pending_recv_address = nullptr;
+  std::move(state->pending_recv_callback).Run(static_cast<int>(data.size()));
+}
+
+void CompleteRelayReadError(const std::shared_ptr<RelayState>& state,
+                            int result) {
+  CHECK_LT(result, 0);
+  CHECK(state->recv_pending);
+  CHECK(state->pending_recv_callback);
+  state->recv_pending = false;
+  state->pending_recv_buffer.reset();
+  state->pending_recv_address = nullptr;
+  std::move(state->pending_recv_callback).Run(result);
+}
+
 void CompleteRelaySend(const std::shared_ptr<RelayState>& state, int result) {
   CHECK(state->send_pending);
   CHECK(state->pending_send_callback);
@@ -643,6 +676,93 @@ void TestSynchronousBackendEchoAndSendErrorStopsReads() {
          "synchronous SendTo error never reads closed relay");
 }
 
+void TestRecoverableRelayReadErrorsContinue() {
+  AssociationHarness harness;
+  BuildAssociation(harness);
+  for (int result : {net::ERR_CONNECTION_RESET, net::ERR_MSG_TOO_BIG,
+                     net::ERR_ADDRESS_UNREACHABLE}) {
+    QueueRelayResult(harness.relay, result);
+  }
+  QueueRelayData(harness.relay, ValidPacket(), harness.control_peer);
+
+  Expect(StartAssociation(harness) == net::ERR_IO_PENDING,
+         "recoverable synchronous read association starts");
+  RunUntilIdle();
+
+  Expect(harness.completion.calls == 0,
+         "recoverable synchronous read errors keep association active");
+  Expect(harness.backend->send_calls == 1,
+         "valid datagram after synchronous read errors reaches backend");
+  Expect(harness.relay->recv_pending,
+         "relay read rearms after synchronous recoverable errors");
+
+  if (harness.relay->recv_pending) {
+    CompleteRelayReadError(harness.relay, net::ERR_CONNECTION_RESET);
+    Expect(harness.completion.calls == 0 && harness.relay->recv_pending,
+           "asynchronous recoverable read error rearms receive");
+  }
+  if (harness.relay->recv_pending) {
+    CompleteRelayRead(harness.relay, ValidPacket(), harness.control_peer);
+    Expect(harness.backend->send_calls == 2 && harness.relay->recv_pending,
+           "valid datagram after asynchronous read error reaches backend");
+  }
+}
+
+void TestRecoverableRelayWriteErrorsContinue() {
+  AssociationHarness synchronous;
+  BuildAssociation(synchronous);
+  synchronous.backend->mode = BackendMode::kEchoSynchronously;
+  for (int result : {net::ERR_CONNECTION_RESET, net::ERR_MSG_TOO_BIG,
+                     net::ERR_ADDRESS_UNREACHABLE}) {
+    synchronous.relay->send_results.push_back(result);
+    QueueRelayData(synchronous.relay, ValidPacket(), synchronous.control_peer);
+  }
+  QueueRelayData(synchronous.relay, ValidPacket(), synchronous.control_peer);
+
+  Expect(StartAssociation(synchronous) == net::ERR_IO_PENDING,
+         "recoverable synchronous write association starts");
+  RunUntilIdle();
+  Expect(synchronous.completion.calls == 0,
+         "recoverable synchronous write errors keep association active");
+  Expect(synchronous.relay->send_calls == 4 &&
+             synchronous.relay->recv_pending,
+         "relay sends the datagram after synchronous write errors");
+
+  AssociationHarness asynchronous;
+  BuildAssociation(asynchronous);
+  asynchronous.backend->mode = BackendMode::kEchoSynchronously;
+  asynchronous.relay->send_results.push_back(net::ERR_IO_PENDING);
+  asynchronous.relay->send_results.push_back(net::ERR_IO_PENDING);
+  QueueRelayData(asynchronous.relay, ValidPacket(), asynchronous.control_peer);
+  QueueRelayData(asynchronous.relay, ValidPacket(), asynchronous.control_peer);
+  Expect(StartAssociation(asynchronous) == net::ERR_IO_PENDING,
+         "recoverable asynchronous write association starts");
+  RunUntilIdle();
+  Expect(asynchronous.relay->send_pending &&
+             asynchronous.relay->send_calls == 1,
+         "first asynchronous relay write is pending");
+  const base::TimeTicks activity_before =
+      asynchronous.association->last_activity();
+  CompleteRelaySend(asynchronous.relay, net::ERR_CONNECTION_RESET);
+  Expect(asynchronous.completion.calls == 0 &&
+             asynchronous.relay->send_calls == 2 &&
+             asynchronous.relay->send_pending &&
+             asynchronous.association->last_activity() == activity_before,
+         "asynchronous write error drops one response without idle refresh");
+}
+
+void TestFatalRelayReadErrorFinishesAssociation() {
+  AssociationHarness harness;
+  BuildAssociation(harness);
+  QueueRelayResult(harness.relay, net::ERR_FAILED);
+  Expect(StartAssociation(harness) == net::ERR_IO_PENDING,
+         "fatal read association starts");
+  RunUntilIdle();
+  Expect(harness.completion.calls == 1 &&
+             harness.completion.result == net::ERR_FAILED,
+         "non-recoverable relay read still finishes the association");
+}
+
 void TestSynchronousControlDataYieldsThenEof() {
   AssociationHarness harness;
   BuildAssociation(harness);
@@ -759,6 +879,33 @@ void TestResponseQueuePressure() {
          "response pressure does not terminate the SOCKS association");
 }
 
+void TestExpirationPolicy() {
+  AssociationHarness harness;
+  BuildAssociation(harness);
+  const base::TimeTicks created = harness.association->creation_time();
+  const base::TimeTicks last_activity = harness.association->last_activity();
+  const base::TimeDelta idle_timeout = base::Seconds(5);
+  const base::TimeDelta tunnel_timeout = base::Seconds(2);
+
+  Expect(!harness.association->ShouldExpire(
+             created + tunnel_timeout, base::Seconds(30), tunnel_timeout),
+         "association remains active at the exact maximum-age boundary");
+  Expect(harness.association->ShouldExpire(
+             created + tunnel_timeout + base::Milliseconds(1),
+             base::Seconds(30), tunnel_timeout),
+         "active association expires after its maximum lifetime");
+  Expect(!harness.association->ShouldExpire(
+             last_activity + idle_timeout, idle_timeout, base::Hours(1)),
+         "association remains active at the exact idle boundary");
+  Expect(harness.association->ShouldExpire(
+             last_activity + idle_timeout + base::Milliseconds(1), idle_timeout,
+             base::Hours(1)),
+         "idle association expires after the idle timeout");
+  Expect(!harness.association->ShouldExpire(
+             created + base::Seconds(1), idle_timeout, tunnel_timeout),
+         "fresh association does not expire");
+}
+
 }  // namespace
 
 int main() {
@@ -768,10 +915,14 @@ int main() {
   TestFirstRecvFromCompletesSynchronously();
   TestSynchronousInvalidBurstYieldsAndRecovers();
   TestSynchronousBackendEchoAndSendErrorStopsReads();
+  TestRecoverableRelayReadErrorsContinue();
+  TestRecoverableRelayWriteErrorsContinue();
+  TestFatalRelayReadErrorFinishesAssociation();
   TestSynchronousControlDataYieldsThenEof();
   TestPendingReadDestructionCancelsCallbacks();
   TestPendingBackendSendDestructionCancelsCallback();
   TestResponseQueuePressure();
+  TestExpirationPolicy();
 
   if (failures != 0) {
     std::cerr << "M2 G4/G5 deterministic association failures=" << failures

src/net/tools/naive/socks5_udp_association.cc

@@ -35,6 +35,19 @@ uint16_t RequestedClientPort(
   return control_socket->request_endpoint().port();
 }
 
+// Per-datagram relay failures (ICMP port-unreachable, oversize). Do not
+// Finish() the whole association.
+bool IsRecoverableRelayError(int result) {
+  switch (result) {
+    case ERR_CONNECTION_RESET:
+    case ERR_MSG_TOO_BIG:
+    case ERR_ADDRESS_UNREACHABLE:
+      return true;
+    default:
+      return false;
+  }
+}
+
 }  // namespace
 
 Socks5UdpAssociation::QueuedResponse::QueuedResponse(
@@ -66,6 +79,13 @@ Socks5UdpAssociation::Socks5UdpAssociation(
 
 Socks5UdpAssociation::~Socks5UdpAssociation() = default;
 
+bool Socks5UdpAssociation::ShouldExpire(base::TimeTicks now,
+                                        base::TimeDelta idle_timeout,
+                                        base::TimeDelta tunnel_timeout) const {
+  return now - last_activity_ > idle_timeout ||
+         now - created_at_ > tunnel_timeout;
+}
+
 int Socks5UdpAssociation::Start(CompletionOnceCallback callback) {
   CHECK(!completion_callback_);
   completion_callback_ = std::move(callback);
@@ -189,6 +209,9 @@ void Socks5UdpAssociation::OnRelayReadComplete(int result) {
 
 bool Socks5UdpAssociation::HandleRelayRead(int result) {
   if (result < 0) {
+    if (IsRecoverableRelayError(result)) {
+      return true;
+    }
     Finish(result);
     return false;
   }
@@ -314,6 +337,12 @@ void Socks5UdpAssociation::PumpRelayWrites() {
 void Socks5UdpAssociation::OnRelayWriteComplete(int result) {
   relay_write_pending_ = false;
   if (result < 0) {
+    if (IsRecoverableRelayError(result)) {
+      CHECK(!response_queue_.empty());
+      response_queue_.pop_front();
+      PumpRelayWrites();
+      return;
+    }
     Finish(result);
     return;
   }