feat(netprobe): HTTP(S) + DoH endpoint probes via libcurl (Refs #697)

conanfile.py

@@ -28,12 +28,16 @@ def requirements(self):
         self.requires("yaml-cpp/0.9.0")
         self.requires("robin-hood-hashing/3.11.5")
         self.requires("libcurl/8.20.0")
+        self.requires("libnghttp2/1.61.0")
         if (
             "libc" not in self.settings.compiler.fields
             or self.settings.compiler.libc != "musl"
         ):
             self.requires("sentry-crashpad/0.6.5")
 
+    def configure(self):
+        self.options["libcurl"].with_nghttp2 = True
+
     def build_requirements(self):
         self.tool_requires("protobuf/6.33.5")
 

libs/CMakeLists.txt

@@ -4,4 +4,5 @@ add_subdirectory(visor_test)
 add_subdirectory(visor_transaction)
 add_subdirectory(visor_tcp)
 add_subdirectory(visor_dns)
-add_subdirectory(visor_utils)
+add_subdirectory(visor_utils)
+add_subdirectory(visor_http_client)

libs/visor_http_client/CMakeLists.txt

@@ -0,0 +1,17 @@
+message(STATUS "Visor HTTP Client Library")
+
+find_package(CURL REQUIRED)
+find_package(uvw REQUIRED)
+find_package(httplib REQUIRED)
+find_package(Catch2 REQUIRED)
+
+add_library(VisorHttpClient STATIC HttpClient.cpp)
+# Namespaced alias for consistency with the other libs (Visor::Lib::Dns, Visor::Lib::Tcp, ...)
+# and cleaner downstream consumption.
+add_library(Visor::Lib::Http ALIAS VisorHttpClient)
+target_include_directories(VisorHttpClient PUBLIC ${CMAKE_CURRENT_SOURCE_DIR})
+target_link_libraries(VisorHttpClient PUBLIC CURL::libcurl uvw::uvw)
+
+add_executable(unit-tests-visor-http-client test_http_client.cpp)
+target_link_libraries(unit-tests-visor-http-client PRIVATE VisorHttpClient Catch2::Catch2WithMain httplib::httplib)
+add_test(NAME unit-tests-visor-http-client COMMAND unit-tests-visor-http-client)

libs/visor_http_client/HttpClient.h

@@ -0,0 +1,80 @@
+#pragma once
+#include "HttpTypes.h"
+#include <curl/curl.h>
+#include <functional>
+#include <memory>
+#include <optional>
+#include <unordered_map>
+#include <uvw/loop.h>
+#include <uvw/poll.h>
+#include <uvw/timer.h>
+
+namespace visor::http {
+
+// Threading contract: HttpClient is CONSTRUCTED on the control thread, before the
+// netprobe io thread is spawned (no concurrency at that point — same as _async_h/_timer
+// in NetProbeInputStream). Thereafter request(), the curl/uvw callbacks, and close()
+// run ONLY on the netprobe io (loop) thread. curl_multi is not thread-safe, so one
+// HttpClient is touched by exactly one thread for its whole active life.
+//
+// Reentrancy: the ResultCallback (on_done) MAY call request() — a follow-up request issued
+// from within a completion callback is safe. The `_processing` guard makes this work:
+//   - check_multi_info() collects (callback, result) pairs FIRST, then fires callbacks; and
+//     it sets _processing=true for the whole drain, so a re-entrant check_multi_info() (e.g.
+//     triggered by a request() from inside a callback) returns immediately instead of
+//     recursively mutating _easy/_sockets under the outer drain.
+//   - request(), when called while _processing, does NOT drive curl_multi_socket_action /
+//     check_multi_info() inline; it adds the handle and arms the timer for 0 ms so the kick
+//     happens on the next loop iteration, after the outer drain unwinds.
+// close() from inside on_done is still discouraged: the drain loop uses local copies so it
+// won't crash, but tearing the client down mid-callback is confusing — prefer deferring it.
+class HttpClient
+{
+public:
+    using ResultCallback = std::function<void(const HttpResult &)>;
+    explicit HttpClient(std::shared_ptr<uvw::loop> loop);
+    ~HttpClient();
+    void request(const HttpRequest &req, ResultCallback on_done);
+    void close();
+
+private:
+    // per-easy-handle context (owns the result callback + a write sink)
+    struct EasyContext {
+        CURL *easy{nullptr};
+        ResultCallback on_done;
+        char errbuf[CURL_ERROR_SIZE]{};
+        curl_slist *headers{nullptr};   // owned; freed in dtor (after curl_easy_cleanup)
+        bool capture{false};
+        std::string response;           // captured body (bounded to 64 KB)
+        ~EasyContext() { if (headers) curl_slist_free_all(headers); }
+    };
+    // per-socket context: a uvw poll handle curl watches (owned in _sockets below)
+    struct SocketContext {
+        std::shared_ptr<uvw::poll_handle> poll;
+        curl_socket_t sockfd{};
+    };
+
+    static int socket_cb(CURL *easy, curl_socket_t s, int what, void *userp, void *socketp);
+    static int timer_cb(CURLM *multi, long timeout_ms, void *userp);
+    static size_t write_discard(char *ptr, size_t size, size_t nmemb, void *userdata);
+    static size_t write_capture(char *ptr, size_t size, size_t nmemb, void *userdata);
+    void on_socket_event(curl_socket_t sockfd, int events);
+    void on_timeout();
+    void check_multi_info();
+
+    std::shared_ptr<uvw::loop> _loop;
+    CURLM *_multi{nullptr};
+    bool _closed{false};
+    bool _processing{false}; // true while check_multi_info() drains; guards re-entrant request()/check_multi_info()
+    std::shared_ptr<uvw::timer_handle> _timer;
+    std::unordered_map<CURL *, std::unique_ptr<EasyContext>> _easy;
+    // we OWN the socket contexts here (curl_multi_assign stores the bare pointer);
+    // close() sweeps these directly rather than relying on CURL_POLL_REMOVE firing for all.
+    std::unordered_map<curl_socket_t, std::unique_ptr<SocketContext>> _sockets;
+};
+
+// Validate that `url` is a well-formed http/https URL (uses libcurl's URL parser internally, so
+// callers don't reach into the curl API). Returns std::nullopt when valid; otherwise a short,
+// human-readable reason (e.g. "is not a valid http(s) URL: '<url>'") the caller can surface.
+std::optional<std::string> validate_http_url(const std::string &url);
+}

libs/visor_http_client/HttpTypes.h

@@ -0,0 +1,34 @@
+#pragma once
+#include <cstdint>
+#include <string>
+#include <vector>
+
+namespace visor::http {
+
+struct HttpTimings {
+    uint64_t total_us{0};
+    uint64_t dns_us{0};
+    uint64_t connect_us{0};
+    uint64_t tls_us{0};
+    uint64_t ttfb_us{0};
+};
+struct HttpRequest {
+    std::string url;
+    std::string method{"GET"};
+    uint64_t timeout_ms{0};
+    bool follow_redirects{true};
+    bool verify_tls{true};
+    std::string body;                   // request body bytes (empty => no body)
+    std::vector<std::string> headers;   // extra request headers, each "Key: Value"
+    bool capture_response{false};       // when true, capture the response body
+};
+struct HttpResult {
+    bool transport_ok{false};
+    long curl_code{0};
+    long status_code{0};
+    HttpTimings timings;
+    std::string response_body;          // populated only when HttpRequest.capture_response
+    std::string content_type;           // raw response Content-Type header when transport_ok (compare case-insensitively)
+    std::string error_msg;              // human-readable curl error detail when !transport_ok
+};
+}

libs/visor_http_client/test_http_client.cpp

@@ -0,0 +1,234 @@
+#include "HttpClient.h"
+#include <catch2/catch_test_macros.hpp>
+#include <httplib.h>
+#include <uvw/loop.h>
+#include <thread>
+
+using namespace visor::http;
+
+// Spin a throwaway httplib server on an ephemeral port; returns port.
+static int start_test_server(httplib::Server &svr, std::thread &t)
+{
+    svr.Get("/ok", [](const httplib::Request &, httplib::Response &res) { res.set_content("hi", "text/plain"); });
+    svr.Get("/notfound", [](const httplib::Request &, httplib::Response &res) { res.status = 404; });
+    svr.Get("/error", [](const httplib::Request &, httplib::Response &res) { res.status = 500; });
+    svr.Get("/slow", [](const httplib::Request &, httplib::Response &res) {
+        std::this_thread::sleep_for(std::chrono::milliseconds(500));
+        res.set_content("late", "text/plain");
+    });
+    svr.Post("/echo", [](const httplib::Request &req, httplib::Response &res) {
+        res.set_content(req.body, "application/octet-stream");
+    });
+    int port = svr.bind_to_any_port("127.0.0.1");
+    REQUIRE(port > 0);
+    t = std::thread([&svr] { svr.listen_after_bind(); });
+    // Wait until the server is actually accepting connections before the test
+    // connects — avoids a connect-before-listen race on slow/loaded CI machines.
+    svr.wait_until_ready();
+    return port;
+}
+
+// Arm a watchdog timer that calls loop->stop() after timeout_ms.
+// The handle is unreferenced so loop->run() returns as soon as the real work
+// handles close — the watchdog fires only if the loop would otherwise stall
+// forever (hang-guard). The caller must close the returned handle after
+// loop->run() completes.
+static std::shared_ptr<uvw::timer_handle> arm_watchdog(std::shared_ptr<uvw::loop> loop, uint64_t timeout_ms)
+{
+    auto wd = loop->resource<uvw::timer_handle>();
+    wd->on<uvw::timer_event>([loop](const auto &, auto &) { loop->stop(); });
+    wd->start(uvw::timer_handle::time{timeout_ms}, uvw::timer_handle::time{0});
+    // Unreference: the watchdog must not prevent the loop from exiting when all
+    // real work handles (curl poll + timer) have closed. uv_unref makes the
+    // handle "idle" w.r.t. loop liveness — the loop exits when only unreferenced
+    // handles remain active, then we close the watchdog in disarm_watchdog.
+    wd->unreference();
+    return wd;
+}
+
+// Disarm the watchdog and drain its close on the loop.
+static void disarm_watchdog(std::shared_ptr<uvw::loop> loop, std::shared_ptr<uvw::timer_handle> wd)
+{
+    if (wd && !wd->closing()) {
+        wd->stop();
+        wd->close();
+        loop->run();
+    }
+}
+
+TEST_CASE("HttpClient basic results", "[http][client]")
+{
+    httplib::Server svr;
+    std::thread server_thread;
+    int port = start_test_server(svr, server_thread);
+
+    auto loop = uvw::loop::create();
+    HttpClient client(loop);
+    std::string base = "http://127.0.0.1:" + std::to_string(port);
+
+    std::vector<HttpResult> results;
+    auto on_done = [&](const HttpResult &r) { results.push_back(r); };
+    // Named init (not positional aggregate) so adding HttpRequest fields doesn't warn.
+    auto get_req = [](std::string url, uint64_t timeout_ms) {
+        HttpRequest r;
+        r.url = std::move(url);
+        r.timeout_ms = timeout_ms;
+        return r;
+    };
+
+    SECTION("200 OK")
+    {
+        client.request(get_req(base + "/ok", 2000), on_done);
+        auto wd = arm_watchdog(loop, 5000);
+        loop->run();
+        disarm_watchdog(loop, wd);
+        REQUIRE(results.size() == 1);
+        CHECK(results[0].transport_ok);
+        CHECK(results[0].status_code == 200);
+        const auto &t = results[0].timings;
+        CHECK(t.total_us > 0);
+        // Per-phase delta sanity (plain HTTP, no TLS): the TLS phase must be 0, and every
+        // phase delta must stay within the total — guards the delta arithmetic against a
+        // wrong base or an unsigned underflow producing a huge bogus value.
+        CHECK(t.tls_us == 0);
+        CHECK(t.dns_us <= t.total_us);
+        CHECK(t.connect_us <= t.total_us);
+        CHECK(t.ttfb_us <= t.total_us);
+    }
+    SECTION("404")
+    {
+        client.request(get_req(base + "/notfound", 2000), on_done);
+        auto wd = arm_watchdog(loop, 5000);
+        loop->run();
+        disarm_watchdog(loop, wd);
+        REQUIRE(results.size() == 1);
+        CHECK(results[0].transport_ok);
+        CHECK(results[0].status_code == 404);
+    }
+    SECTION("connection refused")
+    {
+        client.request(get_req("http://127.0.0.1:1/x", 2000), on_done);
+        auto wd = arm_watchdog(loop, 5000);
+        loop->run();
+        disarm_watchdog(loop, wd);
+        REQUIRE(results.size() == 1);
+        CHECK_FALSE(results[0].transport_ok);
+    }
+    SECTION("timeout")
+    {
+        client.request(get_req(base + "/slow", 100), on_done);
+        auto wd = arm_watchdog(loop, 5000);
+        loop->run();
+        disarm_watchdog(loop, wd);
+        REQUIRE(results.size() == 1);
+        CHECK_FALSE(results[0].transport_ok);
+        CHECK(results[0].curl_code == CURLE_OPERATION_TIMEDOUT);
+    }
+    SECTION("close while in flight")
+    {
+        // Issue the slow request but then immediately close() before loop->run() completes it.
+        // The process must not crash and results must stay empty (interrupted = no metric recorded).
+        client.request(get_req(base + "/slow", 5000), on_done);
+        client.close();
+        auto wd = arm_watchdog(loop, 3000);
+        loop->run();
+        disarm_watchdog(loop, wd);
+        CHECK(results.empty());
+    }
+
+    // close() is idempotent — this is a no-op for the "close while in flight" section.
+    client.close();
+    loop->run();          // drain handle closes
+    svr.stop();
+    if (server_thread.joinable()) server_thread.join();
+}
+
+TEST_CASE("HttpClient POST body + headers + response capture", "[http][client]")
+{
+    httplib::Server svr;
+    std::thread server_thread;
+    int port = start_test_server(svr, server_thread);
+    auto loop = uvw::loop::create();
+    HttpClient client(loop);
+    std::string base = "http://127.0.0.1:" + std::to_string(port);
+
+    std::vector<HttpResult> results;
+    auto on_done = [&](const HttpResult &r) { results.push_back(r); };
+
+    SECTION("POST echoes body when capture_response")
+    {
+        HttpRequest req;
+        req.url = base + "/echo";
+        req.method = "POST";
+        req.body = std::string("\x00\x01hello", 7); // binary-safe
+        req.headers = {"Content-Type: application/octet-stream"};
+        req.capture_response = true;
+        req.timeout_ms = 2000;
+        client.request(req, on_done);
+        auto wd = arm_watchdog(loop, 5000);
+        loop->run();
+        disarm_watchdog(loop, wd);
+        REQUIRE(results.size() == 1);
+        CHECK(results[0].transport_ok);
+        CHECK(results[0].status_code == 200);
+        CHECK(results[0].response_body == std::string("\x00\x01hello", 7));
+    }
+    SECTION("body NOT captured by default")
+    {
+        HttpRequest req;
+        req.url = base + "/ok";
+        req.capture_response = false;
+        req.timeout_ms = 2000;
+        client.request(req, on_done);
+        auto wd = arm_watchdog(loop, 5000);
+        loop->run();
+        disarm_watchdog(loop, wd);
+        REQUIRE(results.size() == 1);
+        CHECK(results[0].response_body.empty());
+    }
+
+    client.close();
+    loop->run();
+    svr.stop();
+    if (server_thread.joinable()) server_thread.join();
+}
+
+TEST_CASE("HttpClient request() issued from within a completion callback is safe", "[http][client]")
+{
+    httplib::Server svr;
+    std::thread server_thread;
+    int port = start_test_server(svr, server_thread);
+
+    auto loop = uvw::loop::create();
+    HttpClient client(loop);
+    std::string base = "http://127.0.0.1:" + std::to_string(port);
+
+    std::vector<long> statuses;
+    // In the first request's completion callback, issue a SECOND request (reentrant). The
+    // _processing guard must keep this safe and let both complete.
+    auto second = [&](const HttpResult &r) { statuses.push_back(r.status_code); };
+    auto first = [&](const HttpResult &r) {
+        statuses.push_back(r.status_code);
+        HttpRequest r2;
+        r2.url = base + "/notfound";
+        r2.timeout_ms = 2000;
+        client.request(r2, second); // reentrant: called from inside check_multi_info()'s callback fire
+    };
+    HttpRequest r1;
+    r1.url = base + "/ok";
+    r1.timeout_ms = 2000;
+    client.request(r1, first);
+
+    auto wd = arm_watchdog(loop, 5000);
+    loop->run();
+    disarm_watchdog(loop, wd);
+
+    REQUIRE(statuses.size() == 2);
+    CHECK(statuses[0] == 200);
+    CHECK(statuses[1] == 404);
+
+    client.close();
+    loop->run();
+    svr.stop();
+    if (server_thread.joinable()) server_thread.join();
+}