diff --git a/src/net/quic/quic_proxy_datagram_client_socket.cc b/src/net/quic/quic_proxy_datagram_client_socket.cc index e9fa07f684..da2f131344 100644 --- a/src/net/quic/quic_proxy_datagram_client_socket.cc +++ b/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; diff --git a/src/net/tools/naive/naive_connect_udp_backend_test_bin.cc b/src/net/tools/naive/naive_connect_udp_backend_test_bin.cc index a50283bdcd..c59288a96e 100644 --- a/src/net/tools/naive/naive_connect_udp_backend_test_bin.cc +++ b/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(); net::NaiveConnectUdpDatagramBackend byte_cap_backend( diff --git a/src/net/tools/naive/naive_connect_udp_datagram_backend.cc b/src/net/tools/naive/naive_connect_udp_datagram_backend.cc index 181ba9ec22..20210fce4a 100644 --- a/src/net/tools/naive/naive_connect_udp_datagram_backend.cc +++ b/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 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; diff --git a/src/net/tools/naive/naive_proxy.cc b/src/net/tools/naive/naive_proxy.cc index 4481bfa4b7..600adeaddd 100644 --- a/src/net/tools/naive/naive_proxy.cc +++ b/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); } } diff --git a/src/net/tools/naive/naive_socks5_udp_association_test_bin.cc b/src/net/tools/naive/naive_socks5_udp_association_test_bin.cc index ff36a3067e..c17c02d34e 100644 --- a/src/net/tools/naive/naive_socks5_udp_association_test_bin.cc +++ b/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 pending_recv_buffer; + net::IPEndPoint* pending_recv_address = nullptr; + net::CompletionOnceCallback pending_recv_callback; scoped_refptr pending_send_buffer; net::CompletionOnceCallback pending_send_callback; }; @@ -236,6 +240,12 @@ void QueueRelayData(const std::shared_ptr& state, .data = std::move(data), .source = source, .result = net::OK}); } +void QueueRelayResult(const std::shared_ptr& 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 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 state_; const net::IPEndPoint local_endpoint_; - scoped_refptr pending_recv_buffer_; - net::IPEndPoint* pending_recv_address_ = nullptr; - net::CompletionOnceCallback pending_recv_callback_; net::NetLogWithSource net_log_; }; +void CompleteRelayRead(const std::shared_ptr& state, + std::vector data, + const net::IPEndPoint& source) { + CHECK(state->recv_pending); + CHECK(state->pending_recv_callback); + CHECK_LE(data.size(), + static_cast(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(data.size())); +} + +void CompleteRelayReadError(const std::shared_ptr& 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& 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 diff --git a/src/net/tools/naive/socks5_udp_association.cc b/src/net/tools/naive/socks5_udp_association.cc index 552aeb7a10..33c7247491 100644 --- a/src/net/tools/naive/socks5_udp_association.cc +++ b/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; } diff --git a/src/net/tools/naive/socks5_udp_association.h b/src/net/tools/naive/socks5_udp_association.h index 7e9932c5d5..2eac0365e5 100644 --- a/src/net/tools/naive/socks5_udp_association.h +++ b/src/net/tools/naive/socks5_udp_association.h @@ -42,6 +42,10 @@ class Socks5UdpAssociation { unsigned int id() const { return id_; } base::TimeTicks last_activity() const { return last_activity_; } + base::TimeTicks creation_time() const { return created_at_; } + bool ShouldExpire(base::TimeTicks now, + base::TimeDelta idle_timeout, + base::TimeDelta tunnel_timeout) const; // Returns ERR_IO_PENDING while the association is active. The callback is // invoked once when the TCP control connection or UDP relay terminates. @@ -101,6 +105,7 @@ class Socks5UdpAssociation { CompletionOnceCallback completion_callback_; bool finished_ = false; + const base::TimeTicks created_at_ = base::TimeTicks::Now(); base::TimeTicks last_activity_ = base::TimeTicks::Now(); base::WeakPtrFactory weak_ptr_factory_{this};