Pull request 650: Fix crashes

Merge in ADGUARD-CORE-LIBS/dns-libs from fix/crashes to master Squashed commit of the following: commit ff481860b94b46cd0db42c29150a59052fe0556d Author: Sergey Fionov <[email protected]> Date: Thu Apr 25 12:22:34 2024 +0300 skipci: fix commit 0812e63c90a8f1e1ab2e4638c0c38a47a436dfaf Author: Sergey Fionov <[email protected]> Date: Thu Apr 25 12:14:52 2024 +0300 Add test for dnsfilter commit acef6405b1ce529b9f4d7b4ab24d9596f7e25679 Author: Sergey Fionov <[email protected]> Date: Thu Apr 25 12:10:20 2024 +0300 Add EventLoop test, fix code commit b200ffc5a6c2fc85bd44a335f6743d027fc510cc Author: Sergey Fionov <[email protected]> Date: Thu Apr 25 10:02:38 2024 +0300 Fix crash in event loop commit 24d42013b30b0f2fed11eda8099f4146283c84ac Author: Sergey Fionov <[email protected]> Date: Thu Apr 25 09:05:24 2024 +0300 Make code more readable in handlePacket commit bff503fa26e70e073bfd1a704bd36f3d70e4ce9a Author: Sergey Fionov <[email protected]> Date: Thu Apr 25 08:51:35 2024 +0300 Deadlock avoidment in AGDnsProxy commit c631925da78b6cf41bfabff32d7852940575e575 Author: Sergey Fionov <[email protected]> Date: Wed Apr 24 22:42:52 2024 +0300 Synchronize AGDnsProxy class commit ab9849e8c5b2d1582a942d953da68fb797d8ac6f Author: Sergey Fionov <[email protected]> Date: Wed Apr 24 22:27:12 2024 +0300 Fix system resolver live test commit 18b2e9af54e612ecc8f3518370614e6d8c6857d1 Author: Sergey Fionov <[email protected]> Date: Wed Apr 24 22:26:57 2024 +0300 Remove unneeded use of any_of commit 13aea9f0861f154d530de47fd2c4538a10f3f290 Author: Sergey Fionov <[email protected]> Date: Wed Apr 24 22:26:22 2024 +0300 Properly handle socket error commit 663f516a6baded6119e039523ec692cd9b77e6c6 Author: Sergey Fionov <[email protected]> Date: Wed Apr 24 22:25:53 2024 +0300 Fix crash in uv_count_bufs commit cd827d9f79cb62cd2aea70c7420f9766d5f66f08 Author: Sergey Fionov <[email protected]> Date: Wed Apr 24 22:23:07 2024 +0300 Fix crash in search_in_cidrs
AdguardTeam · Apr 25, 2024 · e9dbaf6 · e9dbaf6
1 parent daa68c4
commit e9dbaf6
Show file tree

Hide file tree

Showing 9 changed files with 95 additions and 60 deletions.
diff --git a/common/event_loop.cpp b/common/event_loop.cpp
@@ -100,8 +100,8 @@ void EventLoop::execute_stopper_iteration() noexcept {
 }
 
 void EventLoop::stop() {
+    m_stopping = true;
     submit([this]{
-        m_stopping = true;
         m_stopper = Uv<uv_idle_t>::create_with_parent(this);
         uv_idle_init(m_handle->raw(), m_stopper->raw());
         uv_idle_start(m_stopper->raw(), [](uv_idle_t *idle) {
@@ -141,7 +141,7 @@ void EventLoop::start() {
 
 EventLoop::~EventLoop() {
     dbglog(m_log, "Destroying");
-    if (m_running) {
+    if (m_running && !m_stopping) {
         errlog(m_log, "Event loop was not stopped before destruction");
         abort();
     }

diff --git a/common/test/test_event_loop.cpp b/common/test/test_event_loop.cpp
@@ -19,6 +19,12 @@ TEST(EventLoop, Submit) {
     loop->join();
 }
 
+TEST(EventLoop, InstantDestructionAfterStop) {
+    auto loop = EventLoop::create();
+    loop->start();
+    loop->stop();
+}
+
 TEST(EventLoop, Coro) {
     static Logger log("Coro");
     auto loop = EventLoop::create();

diff --git a/dnsfilter/filter.cpp b/dnsfilter/filter.cpp
@@ -736,7 +736,11 @@ void Filter::Impl::search_in_cidrs(MatchArg &match) const {
         return;
     }
 
-    CidrRange seek(match.ctx.host);
+    if (!match.ctx.ip_as_cidr.has_value()) {
+        return;
+    }
+
+    CidrRange &seek = *match.ctx.ip_as_cidr;
     auto last_includes = this->cidrs_table.upper_bound(seek);
     for (auto iter = this->cidrs_table.begin(); iter != last_includes; ++iter) {
         match_by_file_position(match, iter->second);

diff --git a/dnsfilter/test/dnsfilter_test.cpp b/dnsfilter/test/dnsfilter_test.cpp
@@ -1481,6 +1481,11 @@ static const CidrTestSample CIDR_TEST_SAMPLES[] = {
                 .ip = "feed::beef",
                 .rules = {"feed::be00/128"},
         },
+        {
+                .ip = "fe80::1%1",
+                .rules = {"fe80::/10"},
+                .expected_match = {"fe80::/10"},
+        },
 };
 
 INSTANTIATE_TEST_SUITE_P(DnsFilter, Cidr, testing::ValuesIn(CIDR_TEST_SAMPLES));

diff --git a/net/tcp_stream.cpp b/net/tcp_stream.cpp
@@ -94,23 +94,20 @@ Error<SocketError> TcpStream::connect(ConnectParameters params) {
     return {};
 }
 
+struct UvWrite : public uv_write_t {
+    Uint8Vector buf;
+};
+
 void TcpStream::on_write(uv_write_t *req, int status) {
-    auto *weak_data = Uv<uv_tcp_t>::weak_from_data(req->data);
-    if (auto tcp_handle = weak_data->lock()) {
-        auto *self = (TcpStream *) tcp_handle->parent();
-        self->m_writes.erase(req);
-    }
-    delete weak_data;
-    delete req;
+    delete static_cast<UvWrite *>(req);
 }
 
 Error<SocketError> TcpStream::send(Uint8View data) {
     log_stream(this, trace, "{}", data.size());
 
-    auto req = new uv_write_t;
-    req->data = new UvWeak<uv_tcp_t>(m_tcp);
-    m_writes[req].assign(data.begin(), data.begin() + data.size());
-    uv_buf_t buf = uv_buf_init((char *) m_writes[req].data(), m_writes[req].size());
+    auto req = new UvWrite{};
+    req->buf.assign(data.begin(), data.begin() + data.size());
+    uv_buf_t buf = uv_buf_init((char *) req->buf.data(), req->buf.size());
     if (int err = uv_write(req, (uv_stream_t *) m_tcp->raw(), &buf, 1, &on_write)) {
         log_stream(this, dbg, "Failed to write data");
         return make_error(SocketError::AE_SOCK_ERROR, make_error(uv_errno_t(err)));
@@ -220,7 +217,12 @@ void TcpStream::on_read(uv_stream_t *stream, ssize_t nread, const uv_buf_t *buf)
             }
             return;
         }
-        dbglog(self->m_log, "Read error: {}", nread);
+        dbglog(self->m_log, "Read error: {}", uv_strerror(nread));
+        Error<SocketError> err = make_error(SocketError::AE_SOCK_ERROR,
+                make_error(uv_errno_t(nread)));
+        if (Callbacks cbx = self->get_callbacks(); cbx.on_close != nullptr) {
+            cbx.on_close(cbx.arg, err);
+        }
         return;
     }
 

diff --git a/net/tcp_stream.h b/net/tcp_stream.h
@@ -28,7 +28,6 @@ class TcpStream : public Socket {
     Callbacks m_callbacks = {};
     bool m_connected = false;
     std::optional<std::chrono::microseconds> m_current_timeout;
-    HashMap<uv_write_t *, Uint8Vector> m_writes;
     HashMap<char *, std::unique_ptr<char[]>> m_reads;
 
     [[nodiscard]] std::optional<evutil_socket_t> get_fd() const override;

diff --git a/platform/mac/framework/AGDnsProxy.mm b/platform/mac/framework/AGDnsProxy.mm
@@ -34,6 +34,8 @@
 static constexpr int BINDFD_WAIT_MS = 10000;
 static constexpr int ERR_BIND_IN_USE = 14;
 
+static const char *IS_DNS_QUEUE_KEY = "isAGDnsProxyQueue";
+
 /**
  * @param str an STL string
  * @return an NSString converted from the C++ string
@@ -181,7 +183,7 @@ static uint16_t udp_checksum_v6(const struct iphdr6 *ip6_header,
     return htons(sum);
 }
 
-static void *create_response_packet(const struct iphdr *ip_header, const struct udphdr *udp_header,
+static CFDataRef create_response_packet(const struct iphdr *ip_header, const struct udphdr *udp_header,
         const std::vector<uint8_t> &payload) {
     struct udphdr reverse_udp_header = {};
     reverse_udp_header.uh_sport = udp_header->uh_dport;
@@ -208,12 +210,11 @@ static uint16_t udp_checksum_v6(const struct iphdr6 *ip6_header,
     [reverse_packet appendBytes: &reverse_udp_header length: sizeof(reverse_udp_header)];
     [reverse_packet appendBytes: payload.data() length: payload.size()];
 
-    return (__bridge_retained void *) reverse_packet;
+    return (__bridge_retained CFDataRef) reverse_packet;
 }
 
-static void *create_response_packet_v6(const struct iphdr6 *ip6_header,
-                                         const struct udphdr *udp_header,
-                                         const std::vector<uint8_t> &payload) {
+static CFDataRef create_response_packet_v6(const struct iphdr6 *ip6_header, const struct udphdr *udp_header,
+        const std::vector<uint8_t> &payload) {
     struct udphdr resp_udp_header = {};
     resp_udp_header.uh_sport = udp_header->uh_dport;
     resp_udp_header.uh_dport = udp_header->uh_sport;
@@ -233,7 +234,7 @@ static uint16_t udp_checksum_v6(const struct iphdr6 *ip6_header,
     [response_packet appendBytes: &resp_udp_header length: sizeof(resp_udp_header)];
     [response_packet appendBytes: payload.data() length: payload.size()];
 
-    return (__bridge_retained void *) response_packet;
+    return (__bridge_retained CFDataRef) response_packet;
 }
 
 static ServerStamp convert_stamp(AGDnsStamp *stamp) {
@@ -907,8 +908,15 @@ - (void)dealloc
 
 - (void)stop {
     if (initialized) {
-        self->proxy.deinit();
-        initialized = NO;
+        auto block = ^{
+            self->proxy.deinit();
+            initialized = NO;
+        };
+        if (dispatch_get_specific(IS_DNS_QUEUE_KEY) == (void *) 0x1) {
+            block();
+        } else {
+            dispatch_sync(queue, block);
+        }
     }
 }
 
@@ -1288,6 +1296,7 @@ - (instancetype) initWithConfig: (AGDnsProxyConfig *) config
 
     void *obj = (__bridge void *)self;
     self->queue = dispatch_queue_create("com.adguard.dnslibs.AGDnsProxy.queue", nil);
+    dispatch_queue_set_specific(self->queue, IS_DNS_QUEUE_KEY, (void *) 0x1, nullptr);
     self->events = handler;
     DnsProxyEvents native_events = {};
     if (handler != nil && handler.onRequestProcessed != nil) {
@@ -1419,7 +1428,7 @@ - (instancetype) initWithConfig: (AGDnsProxyConfig *) config
     return self;
 }
 
-static coro::Task<void *> handleIPv4Packet(AGDnsProxy *self, NSData *packet)
+static coro::Task<CFDataRef> handleIPv4Packet(AGDnsProxy *self, NSData *packet)
 {
     auto *ip_header = (struct iphdr *) packet.bytes;
     // @todo: handle tcp packets also
@@ -1453,7 +1462,7 @@ - (instancetype) initWithConfig: (AGDnsProxyConfig *) config
     co_return create_response_packet(ip_header, udp_header, response);
 }
 
-static coro::Task<void *> handleIPv6Packet(AGDnsProxy *self, NSData *packet)
+static coro::Task<CFDataRef> handleIPv6Packet(AGDnsProxy *self, NSData *packet)
 {
     auto *ip_header = (struct iphdr6 *) packet.bytes;
     // @todo: handle tcp packets also
@@ -1489,39 +1498,51 @@ - (instancetype) initWithConfig: (AGDnsProxyConfig *) config
 
 - (void)handlePacket:(NSData *)packet completionHandler:(void(^)(NSData *)) completionHandler
 {
-    coro::run_detached([](AGDnsProxy *self, NSData *packet, void (^completionHandler)(NSData *)) -> coro::Task<void> {
-        auto *ip_header = (const struct iphdr *)packet.bytes;
-        void *reply = nullptr;
-        if (ip_header->ip_v == 4) {
-            reply = co_await handleIPv4Packet(self, packet);
-        } else if (ip_header->ip_v == 6) {
-            reply = co_await handleIPv6Packet(self, packet);
-        } else {
-            dbglog(*self->log, "Wrong IP version: %u", ip_header->ip_v);
+    dispatch_async(queue, ^{
+        if (!initialized) {
+            completionHandler([NSData new]);
+            return;
         }
+        coro::run_detached([](AGDnsProxy *self, NSData *packet, void (^completionHandler)(NSData *)) -> coro::Task<void> {
+            auto *ip_header = (const struct iphdr *)packet.bytes;
+            CFDataRef reply = nullptr;
+            if (ip_header->ip_v == 4) {
+                reply = co_await handleIPv4Packet(self, packet);
+            } else if (ip_header->ip_v == 6) {
+                reply = co_await handleIPv6Packet(self, packet);
+            } else {
+                dbglog(*self->log, "Wrong IP version: %u", ip_header->ip_v);
+            }
 
-        @autoreleasepool {
-            completionHandler((__bridge_transfer NSData *) reply);
-        }
-    }(self, packet, completionHandler));
+            @autoreleasepool {
+                completionHandler((__bridge_transfer NSData *) reply);
+            }
+        }(self, packet, completionHandler));
+    });
 }
 
 - (void)handleMessage:(NSData *)message
              withInfo:(AGDnsMessageInfo *)info
 withCompletionHandler:(void (^)(NSData *))handler {
-    coro::run_detached([](AGDnsProxy *self, NSData *message, AGDnsMessageInfo *info,
-            void (^handler)(NSData *)) -> coro::Task<void> {
-        std::optional<DnsMessageInfo> cpp_info;
-        if (info) {
-            cpp_info.emplace();
-            cpp_info->transparent = info.transparent;
-        }
-        auto result = co_await self->proxy.handle_message({(uint8_t *) message.bytes, (size_t) message.length},
-                                                          opt_as_ptr(cpp_info));
-        @autoreleasepool {
-            handler([NSData dataWithBytes:result.data() length:result.size()]);
+    dispatch_async(queue, ^{
+        if (!initialized) {
+            handler([NSData new]);
+            return;
         }
-    }(self, message, info, handler));
+        coro::run_detached([](AGDnsProxy *self, NSData *message, AGDnsMessageInfo *info,
+                                   void (^handler)(NSData *)) -> coro::Task<void> {
+            std::optional<DnsMessageInfo> cpp_info;
+            if (info) {
+                cpp_info.emplace();
+                cpp_info->transparent = info.transparent;
+            }
+            auto result = co_await self->proxy.handle_message({(uint8_t *) message.bytes, (size_t) message.length},
+                    opt_as_ptr(cpp_info));
+            @autoreleasepool {
+                handler([NSData dataWithBytes:result.data() length:result.size()]);
+            }
+        }(self, message, info, handler));
+    });
 }
 
 + (BOOL) isValidRule: (NSString *) str

diff --git a/upstream/dns_framed.h b/upstream/dns_framed.h
@@ -99,10 +99,9 @@ class DnsFramedConnection : public Connection, public std::enable_shared_from_th
                 self->finish_request(request_id, Reply{make_error(DnsError::AE_TIMED_OUT)});
             }
         };
-        return parallel::any_of<void>(
-                Awaitable{.self = this, .req = request},
-                wait_timeout(m_loop, weak_from_this(), request->timeout, request->request_id)
-        );
+        coro::run_detached(
+                wait_timeout(m_loop, weak_from_this(), request->timeout, request->request_id));
+        return Awaitable{.self = this, .req = request};
     }
 
     auto wait_response(Request *request) {
@@ -125,10 +124,9 @@ class DnsFramedConnection : public Connection, public std::enable_shared_from_th
                 self->finish_request(request_id, Reply{make_error(DnsError::AE_TIMED_OUT)});
             }
         };
-        return parallel::any_of<void>(
-                Awaitable{.self = this, .req = request},
-                wait_timeout(m_loop, weak_from_this(), request->timeout, request->request_id)
-        );
+        coro::run_detached(
+                wait_timeout(m_loop, weak_from_this(), request->timeout, request->request_id));
+        return Awaitable{.self = this, .req = request};
     }
 };
 

diff --git a/upstream/test/test_system_resolver.cpp b/upstream/test/test_system_resolver.cpp
@@ -22,7 +22,7 @@ class SystemResolverTest : public ::testing::Test {
         Logger::set_log_level(LOG_LEVEL_TRACE);
         m_loop = EventLoop::create();
         m_loop->start();
-        m_resolver = std::move(SystemResolver::create(m_loop.get(), 0).value());
+        m_resolver = std::move(SystemResolver::create(m_loop.get(), Secs{5}, 0).value());
     }
     void TearDown() override {
         m_resolver.reset();