diff --git a/conanfile.py b/conanfile.py index 017a0f72a..048aebd3e 100644 --- a/conanfile.py +++ b/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") diff --git a/libs/CMakeLists.txt b/libs/CMakeLists.txt index e6480f56e..fadb19e11 100644 --- a/libs/CMakeLists.txt +++ b/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) \ No newline at end of file +add_subdirectory(visor_utils) +add_subdirectory(visor_http_client) diff --git a/libs/visor_http_client/CMakeLists.txt b/libs/visor_http_client/CMakeLists.txt new file mode 100644 index 000000000..11857bfc3 --- /dev/null +++ b/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) diff --git a/libs/visor_http_client/HttpClient.cpp b/libs/visor_http_client/HttpClient.cpp new file mode 100644 index 000000000..bb75528b6 --- /dev/null +++ b/libs/visor_http_client/HttpClient.cpp @@ -0,0 +1,384 @@ +#include "HttpClient.h" +#include +#include +#include +#include + +namespace visor::http { + +// libcurl requires a one-time, NOT-thread-safe global init before the first +// easy/multi handle. Multiple netprobe streams construct HttpClients on different +// control threads, so serialize it once (mirrors PingProbe's std::call_once). +// HttpClient is the ONLY conan-libcurl user in the tree (grep-confirmed; the vendored +// breakpad/sentry curl is a separate static copy with its own global state). The first +// HttpClient — hence this init — is constructed only when an HTTP netprobe stream starts, +// well after process/web-server startup. OpenSSL 3.x (the project's openssl/3.6.3) does +// thread-safe, idempotent auto-init, so curl_global_init(CURL_GLOBAL_DEFAULT) initializing +// OpenSSL does not destructively race other OpenSSL users. curl_global_cleanup() is +// intentionally omitted: it's a process-lifetime singleton; leaking it at exit is the +// conventional choice. +static void ensure_curl_global_init() +{ + static std::once_flag flag; + std::call_once(flag, [] { curl_global_init(CURL_GLOBAL_DEFAULT); }); +} + +std::optional validate_http_url(const std::string &url) +{ + // This may be the FIRST libcurl call (config validation runs before any HttpClient is + // constructed), and libcurl requires curl_global_init() before any other API. Ensure it + // (idempotent via std::call_once) so the URL API can't run pre-init on non-thread-safe builds. + ensure_curl_global_init(); + CURLU *h = curl_url(); + if (!h) { + return std::nullopt; // allocation failure — can't validate, so don't reject + } + auto rc = curl_url_set(h, CURLUPART_URL, url.c_str(), 0); + bool scheme_ok = false; + if (rc == CURLUE_OK) { + char *scheme = nullptr; + if (curl_url_get(h, CURLUPART_SCHEME, &scheme, 0) == CURLUE_OK && scheme) { + scheme_ok = (std::string(scheme) == "http" || std::string(scheme) == "https"); + curl_free(scheme); + } + } + curl_url_cleanup(h); + if (rc != CURLUE_OK || !scheme_ok) { + return std::string("is not a valid http(s) URL: '") + url + "'"; + } + return std::nullopt; +} + +HttpClient::HttpClient(std::shared_ptr loop) + : _loop(std::move(loop)) +{ + ensure_curl_global_init(); + _multi = curl_multi_init(); + if (!_multi) { + throw std::runtime_error("HttpClient: curl_multi_init() failed"); + } + curl_multi_setopt(_multi, CURLMOPT_SOCKETFUNCTION, &HttpClient::socket_cb); + curl_multi_setopt(_multi, CURLMOPT_SOCKETDATA, this); + curl_multi_setopt(_multi, CURLMOPT_TIMERFUNCTION, &HttpClient::timer_cb); + curl_multi_setopt(_multi, CURLMOPT_TIMERDATA, this); + _timer = _loop->resource(); + if (!_timer) { + curl_multi_cleanup(_multi); + _multi = nullptr; + throw std::runtime_error("HttpClient: unable to initialize timer handle"); + } + _timer->on([this](const auto &, auto &) { on_timeout(); }); +} + +HttpClient::~HttpClient() +{ + close(); +} + +// MUST be called on the loop thread (from NetProbeInputStream's async stop callback), +// BEFORE _io_loop->stop()/close(). uv_close is async, but it is requested here during +// the loop's poll phase, so the same iteration's closing-handles phase drains the poll +// handles/timer before run() returns — the loop is then quiescent for _io_loop->close() +// (the netprobe loop-quiescent teardown contract). In-flight transfers are dropped +// WITHOUT invoking their on_done (no metric recorded for an interrupted request). +void HttpClient::close() +{ + if (_closed) { + return; // idempotent (dtor may call after an explicit close) + } + _closed = true; + // 1) Tear curl down FIRST, while the SocketContexts are still alive. curl_multi_cleanup() + // and curl_multi_remove_handle() can synchronously invoke socket_cb(CURL_POLL_REMOVE), + // which dereferences the SocketContext stored via curl_multi_assign — so freeing the + // contexts before cleanup would be a use-after-free. Detaching the socket/timer + // callbacks first makes cleanup not call back at all. + if (_multi) { + curl_multi_setopt(_multi, CURLMOPT_SOCKETFUNCTION, nullptr); + curl_multi_setopt(_multi, CURLMOPT_TIMERFUNCTION, nullptr); + for (auto &kv : _easy) { + curl_multi_remove_handle(_multi, kv.first); + curl_easy_cleanup(kv.first); + } + _easy.clear(); + curl_multi_cleanup(_multi); + _multi = nullptr; + } + // 2) Now no curl callback can fire — close the uvw poll handles + timer on the loop + // thread. uv_close is async, but requested here in the loop's poll phase the close + // callbacks run in the SAME iteration's closing phase, so the loop is quiescent + // before _io_loop->close() (the netprobe loop-quiescent teardown contract). + for (auto &kv : _sockets) { + if (kv.second->poll && !kv.second->poll->closing()) { + kv.second->poll->close(); + } + } + _sockets.clear(); + if (_timer && !_timer->closing()) { + _timer->stop(); + _timer->close(); + } +} + +size_t HttpClient::write_discard(char *, size_t size, size_t nmemb, void *) +{ + return size * nmemb; // we don't keep the body +} + +size_t HttpClient::write_capture(char *ptr, size_t size, size_t nmemb, void *userdata) +{ + size_t n = size * nmemb; + auto *ctx = static_cast(userdata); + if (ctx) { + constexpr size_t kMaxBody = 64 * 1024; // a DNS-over-HTTPS message is well under 64 KB + if (ctx->response.size() < kMaxBody) { + ctx->response.append(ptr, (n < kMaxBody - ctx->response.size()) ? n : (kMaxBody - ctx->response.size())); + } + } + return n; // always consume so curl doesn't abort the transfer +} + +void HttpClient::request(const HttpRequest &req, ResultCallback on_done) +{ + if (_closed || !_multi) { + return; // client is closed; do not touch curl_multi + } + auto ctx = std::make_unique(); + ctx->on_done = std::move(on_done); + CURL *easy = curl_easy_init(); + if (!easy) { + HttpResult fail; + fail.transport_ok = false; + fail.curl_code = CURLE_FAILED_INIT; + if (ctx->on_done) ctx->on_done(fail); + return; + } + ctx->easy = easy; + curl_easy_setopt(easy, CURLOPT_URL, req.url.c_str()); + // Don't force CUSTOMREQUEST for GET (it can subtly change curl's behavior); use the + // native verb opts. HEAD => NOBODY; anything other than GET/HEAD => CUSTOMREQUEST. + if (req.method == "HEAD") { + curl_easy_setopt(easy, CURLOPT_NOBODY, 1L); + } else if (req.method != "GET") { + curl_easy_setopt(easy, CURLOPT_CUSTOMREQUEST, req.method.c_str()); + } + curl_easy_setopt(easy, CURLOPT_FOLLOWLOCATION, req.follow_redirects ? 1L : 0L); + // Bound the redirect chain (curl's default is unlimited) so a redirect loop can't burn the + // whole timeout budget, and restrict redirects to http/https. curl's DEFAULT redirect protocol + // set also permits ftp/ftps, so without this a `Location: ftp://...` would make an HTTP/DoH + // probe follow to a non-HTTP endpoint (unexpected outbound traffic + misleading results). + curl_easy_setopt(easy, CURLOPT_MAXREDIRS, 10L); + curl_easy_setopt(easy, CURLOPT_REDIR_PROTOCOLS_STR, "http,https"); + curl_easy_setopt(easy, CURLOPT_SSL_VERIFYPEER, req.verify_tls ? 1L : 0L); + curl_easy_setopt(easy, CURLOPT_SSL_VERIFYHOST, req.verify_tls ? 2L : 0L); + if (req.timeout_ms) curl_easy_setopt(easy, CURLOPT_TIMEOUT_MS, static_cast(req.timeout_ms)); + // We run on the netprobe io thread, not the main thread; CURLOPT_NOSIGNAL stops curl from + // using signals (e.g. SIGALRM with the standard name resolver), which is unsafe off-main-thread. + curl_easy_setopt(easy, CURLOPT_NOSIGNAL, 1L); + if (!req.body.empty()) { + // COPYPOSTFIELDS copies the bytes (curl owns them); size set first => binary-safe. + curl_easy_setopt(easy, CURLOPT_POSTFIELDSIZE, static_cast(req.body.size())); + curl_easy_setopt(easy, CURLOPT_COPYPOSTFIELDS, req.body.data()); + } + for (const auto &h : req.headers) { + struct curl_slist *appended = curl_slist_append(ctx->headers, h.c_str()); + if (!appended) { + // OOM: curl_slist_append returns null and leaves the existing list intact. Don't + // overwrite ctx->headers with null (that would leak the partial list and drop headers); + // abort the request cleanly. The partial list is freed by ~EasyContext at return. + curl_easy_cleanup(easy); + HttpResult fail; + fail.transport_ok = false; + fail.curl_code = CURLE_OUT_OF_MEMORY; + fail.error_msg = "curl_slist_append failed (out of memory)"; + if (ctx->on_done) ctx->on_done(fail); + return; + } + ctx->headers = appended; + } + if (ctx->headers) { + curl_easy_setopt(easy, CURLOPT_HTTPHEADER, ctx->headers); + } + ctx->capture = req.capture_response; + curl_easy_setopt(easy, CURLOPT_WRITEFUNCTION, ctx->capture ? &HttpClient::write_capture : &HttpClient::write_discard); + curl_easy_setopt(easy, CURLOPT_WRITEDATA, ctx.get()); + curl_easy_setopt(easy, CURLOPT_PRIVATE, ctx.get()); + curl_easy_setopt(easy, CURLOPT_ERRORBUFFER, ctx->errbuf); + _easy[easy] = std::move(ctx); + if (curl_multi_add_handle(_multi, easy) != CURLM_OK) { + // add_handle failed: deliver a transport failure and drop the handle/context so it + // doesn't leak in _easy or stall the loop (the easy handle was never added to multi). + ResultCallback cb; + auto it = _easy.find(easy); + if (it != _easy.end()) { + cb = it->second->on_done; + } + curl_easy_cleanup(easy); + _easy.erase(easy); // frees the EasyContext (+ header slist via ~EasyContext) + if (cb) { + HttpResult fail; + fail.transport_ok = false; + fail.curl_code = CURLE_FAILED_INIT; + cb(fail); + } + return; + } + // Kick the transfer. If we're inside a completion callback (check_multi_info() is draining), + // do NOT drive curl synchronously — that would re-enter the drain. Arm the timer for 0 ms so + // the kick runs on the next loop iteration, after the outer drain unwinds. Otherwise kick now + // (matches curl's multi-socket examples and avoids a stalled first request). + if (_processing) { + if (_timer && !_timer->closing()) { + _timer->start(uvw::timer_handle::time{0}, uvw::timer_handle::time{0}); + } + return; + } + int running = 0; + curl_multi_socket_action(_multi, CURL_SOCKET_TIMEOUT, 0, &running); + check_multi_info(); +} + +int HttpClient::timer_cb(CURLM *, long timeout_ms, void *userp) +{ + auto *self = static_cast(userp); + if (self->_closed) { + return 0; + } + if (!self->_timer) return 0; + if (timeout_ms < 0) { + self->_timer->stop(); + } else { + // time is duration; cast to avoid narrowing from signed long. + self->_timer->start(uvw::timer_handle::time{static_cast(timeout_ms == 0 ? 1 : timeout_ms)}, uvw::timer_handle::time{0}); + } + return 0; +} + +void HttpClient::on_timeout() +{ + if (!_multi) return; + int running = 0; + curl_multi_socket_action(_multi, CURL_SOCKET_TIMEOUT, 0, &running); + check_multi_info(); +} + +int HttpClient::socket_cb(CURL *, curl_socket_t s, int what, void *userp, void *socketp) +{ + auto *self = static_cast(userp); + auto *sctx = static_cast(socketp); + if (what == CURL_POLL_REMOVE) { + if (sctx) { + curl_multi_assign(self->_multi, s, nullptr); + if (sctx->poll && !sctx->poll->closing()) sctx->poll->close(); + self->_sockets.erase(s); // owns + drops the SocketContext (uvw keeps the poll alive until its close cb fires) + } + return 0; + } + if (!sctx) { + auto owned = std::make_unique(); + sctx = owned.get(); + sctx->sockfd = s; + sctx->poll = self->_loop->resource(static_cast(s)); + if (!sctx->poll) { + return -1; // uv_poll_init failed (EMFILE/ENOMEM); signal curl to abort this transfer + } + sctx->poll->on([self, s](const uvw::poll_event &ev, uvw::poll_handle &) { + using flags_t = uvw::poll_handle::poll_event_flags; + // NB: uvw's operator&(enum,enum) returns the enum (not bool), so compare explicitly. + int flags = 0; + if ((ev.flags & flags_t::READABLE) == flags_t::READABLE) flags |= CURL_CSELECT_IN; + if ((ev.flags & flags_t::WRITABLE) == flags_t::WRITABLE) flags |= CURL_CSELECT_OUT; + self->on_socket_event(s, flags); + }); + self->_sockets[s] = std::move(owned); + curl_multi_assign(self->_multi, s, sctx); + } + uvw::poll_handle::poll_event_flags ev{}; + if (what & CURL_POLL_IN) ev = ev | uvw::poll_handle::poll_event_flags::READABLE; + if (what & CURL_POLL_OUT) ev = ev | uvw::poll_handle::poll_event_flags::WRITABLE; + if (sctx->poll && !sctx->poll->closing()) { + if (what & (CURL_POLL_IN | CURL_POLL_OUT)) { + sctx->poll->start(ev); + } else { + // CURL_POLL_NONE: curl wants no events right now — stop polling rather than call + // start() with an empty (invalid) libuv event mask. + sctx->poll->stop(); + } + } + return 0; +} + +void HttpClient::on_socket_event(curl_socket_t sockfd, int events) +{ + if (!_multi) return; + int running = 0; + curl_multi_socket_action(_multi, sockfd, events, &running); + check_multi_info(); +} + +void HttpClient::check_multi_info() +{ + if (!_multi) return; + if (_processing) return; // re-entrancy guard: a callback issued work that re-entered the drain + _processing = true; + struct ProcessingReset { + bool &flag; + ~ProcessingReset() { flag = false; } + } processing_reset{_processing}; + // Drain + clean up FIRST, collecting (callback,result) pairs; invoke callbacks only + // afterwards so a callback that calls request()/close() can't mutate _easy mid-loop. + std::vector> done; + CURLMsg *msg = nullptr; + int pending = 0; + while ((msg = curl_multi_info_read(_multi, &pending))) { + if (msg->msg != CURLMSG_DONE) continue; + CURL *easy = msg->easy_handle; + auto it = _easy.find(easy); + ResultCallback cb = (it != _easy.end()) ? it->second->on_done : ResultCallback{}; + HttpResult result; + if (msg->data.result == CURLE_OK) { + result.transport_ok = true; + long code = 0; + curl_easy_getinfo(easy, CURLINFO_RESPONSE_CODE, &code); + result.status_code = code; + // *_TIME_T are cumulative-from-start microseconds; deltas are best-effort + // phase splits (can be 0 on plain HTTP / across redirects — guarded below). + curl_off_t total = 0, dns = 0, conn = 0, app = 0, ttfb = 0; + curl_easy_getinfo(easy, CURLINFO_TOTAL_TIME_T, &total); + curl_easy_getinfo(easy, CURLINFO_NAMELOOKUP_TIME_T, &dns); + curl_easy_getinfo(easy, CURLINFO_CONNECT_TIME_T, &conn); + curl_easy_getinfo(easy, CURLINFO_APPCONNECT_TIME_T, &app); + curl_easy_getinfo(easy, CURLINFO_STARTTRANSFER_TIME_T, &ttfb); + result.timings.total_us = static_cast(total); + result.timings.dns_us = static_cast(dns); + result.timings.connect_us = conn > dns ? static_cast(conn - dns) : 0; + result.timings.tls_us = app > conn ? static_cast(app - conn) : 0; + result.timings.ttfb_us = ttfb > (app ? app : conn) ? static_cast(ttfb - (app ? app : conn)) : 0; + if (it != _easy.end() && it->second->capture) { + result.response_body = std::move(it->second->response); + } + char *ct = nullptr; + curl_easy_getinfo(easy, CURLINFO_CONTENT_TYPE, &ct); // may be null (no Content-Type) + if (ct) { + result.content_type = ct; // raw header value; consumers compare case-insensitively + } + } else { + result.transport_ok = false; + result.curl_code = msg->data.result; + // Prefer curl's per-handle error buffer (set via CURLOPT_ERRORBUFFER); fall back to + // the generic code string. Surfaced for logging by the probes. + if (it != _easy.end() && it->second->errbuf[0] != '\0') { + result.error_msg = it->second->errbuf; + } else { + result.error_msg = curl_easy_strerror(msg->data.result); + } + } + curl_multi_remove_handle(_multi, easy); + curl_easy_cleanup(easy); + _easy.erase(easy); + if (cb) done.emplace_back(std::move(cb), result); + } + for (auto &d : done) { + d.first(d.second); + } +} +} diff --git a/libs/visor_http_client/HttpClient.h b/libs/visor_http_client/HttpClient.h new file mode 100644 index 000000000..9fb691c28 --- /dev/null +++ b/libs/visor_http_client/HttpClient.h @@ -0,0 +1,80 @@ +#pragma once +#include "HttpTypes.h" +#include +#include +#include +#include +#include +#include +#include +#include + +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; + explicit HttpClient(std::shared_ptr 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 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 _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 _timer; + std::unordered_map> _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> _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: ''") the caller can surface. +std::optional validate_http_url(const std::string &url); +} diff --git a/libs/visor_http_client/HttpTypes.h b/libs/visor_http_client/HttpTypes.h new file mode 100644 index 000000000..797f29e70 --- /dev/null +++ b/libs/visor_http_client/HttpTypes.h @@ -0,0 +1,34 @@ +#pragma once +#include +#include +#include + +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 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 +}; +} diff --git a/libs/visor_http_client/test_http_client.cpp b/libs/visor_http_client/test_http_client.cpp new file mode 100644 index 000000000..62365f11f --- /dev/null +++ b/libs/visor_http_client/test_http_client.cpp @@ -0,0 +1,234 @@ +#include "HttpClient.h" +#include +#include +#include +#include + +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 arm_watchdog(std::shared_ptr loop, uint64_t timeout_ms) +{ + auto wd = loop->resource(); + wd->on([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 loop, std::shared_ptr 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 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 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 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(); +} diff --git a/src/handlers/netprobe/CMakeLists.txt b/src/handlers/netprobe/CMakeLists.txt index 038685559..2acf775c6 100644 --- a/src/handlers/netprobe/CMakeLists.txt +++ b/src/handlers/netprobe/CMakeLists.txt @@ -15,6 +15,7 @@ target_link_libraries(VisorHandlerNetProbe PUBLIC Visor::Lib::Transaction Visor::Input::NetProbe + Visor::Lib::Dns ) set(VISOR_STATIC_PLUGINS ${VISOR_STATIC_PLUGINS} Visor::Handler::NetProbe PARENT_SCOPE) diff --git a/src/handlers/netprobe/NetProbeStreamHandler.cpp b/src/handlers/netprobe/NetProbeStreamHandler.cpp index ea01aff93..e11b13d0b 100644 --- a/src/handlers/netprobe/NetProbeStreamHandler.cpp +++ b/src/handlers/netprobe/NetProbeStreamHandler.cpp @@ -4,6 +4,7 @@ #include "NetProbeStreamHandler.h" #include "PrometheusSerializer.h" +#include "dns.h" namespace visor::handler::netprobe { @@ -51,6 +52,8 @@ void NetProbeStreamHandler::start() _probe_recv_connection = _netprobe_proxy->probe_recv_signal.connect(&NetProbeStreamHandler::probe_signal_recv, this); _probe_fail_connection = _netprobe_proxy->probe_fail_signal.connect(&NetProbeStreamHandler::probe_signal_fail, this); _heartbeat_connection = _netprobe_proxy->heartbeat_signal.connect(&NetProbeStreamHandler::check_period_shift, this); + _probe_http_result_connection = _netprobe_proxy->probe_http_result_signal.connect(&NetProbeStreamHandler::probe_signal_http_result, this); + _probe_doh_result_connection = _netprobe_proxy->probe_doh_result_signal.connect(&NetProbeStreamHandler::probe_signal_doh_result, this); } _running = true; @@ -67,6 +70,8 @@ void NetProbeStreamHandler::stop() _probe_recv_connection.disconnect(); _probe_fail_connection.disconnect(); _heartbeat_connection.disconnect(); + _probe_http_result_connection.disconnect(); + _probe_doh_result_connection.disconnect(); } _running = false; @@ -102,9 +107,25 @@ void NetProbeStreamHandler::probe_signal_recv(pcpp::Packet &payload, TestType ty } } -void NetProbeStreamHandler::probe_signal_fail(ErrorType error, [[maybe_unused]] TestType type, const std::string &name) +void NetProbeStreamHandler::probe_signal_fail(ErrorType error, TestType type, const std::string &name) { - _metrics->process_failure(error, name); + if (type == TestType::HTTP) { + _metrics->process_netprobe_http_failure(error, name); + } else if (type == TestType::DOH) { + _metrics->process_netprobe_doh_failure(error, name); + } else { + _metrics->process_failure(error, name); + } +} + +void NetProbeStreamHandler::probe_signal_http_result(uint16_t status, visor::http::HttpTimings timings, const std::string &name, timespec stamp) +{ + _metrics->process_netprobe_http_result(status, timings, name, stamp); +} + +void NetProbeStreamHandler::probe_signal_doh_result(uint16_t http_status, uint8_t rcode, bool parse_ok, visor::http::HttpTimings timings, const std::string &name, timespec stamp) +{ + _metrics->process_netprobe_doh_result(http_status, rcode, parse_ok, timings, name, stamp); } void NetProbeMetricsBucket::specialized_merge(const AbstractMetricsBucket &o, Metric::Aggregate agg_operator) @@ -117,9 +138,7 @@ void NetProbeMetricsBucket::specialized_merge(const AbstractMetricsBucket &o, Me for (const auto &target : other._targets_metrics) { const auto &targetId = target.first; - if (!_targets_metrics.count(targetId)) { - _targets_metrics[targetId] = std::make_unique(); - } + get_or_create_target(targetId); if (group_enabled(group::NetProbeMetrics::Counters)) { _targets_metrics[targetId]->attempts += target.second->attempts; @@ -127,6 +146,10 @@ void NetProbeMetricsBucket::specialized_merge(const AbstractMetricsBucket &o, Me _targets_metrics[targetId]->connect_failures += target.second->connect_failures; _targets_metrics[targetId]->dns_failures += target.second->dns_failures; _targets_metrics[targetId]->timed_out += target.second->timed_out; + _targets_metrics[targetId]->http_status_failures += target.second->http_status_failures; + _targets_metrics[targetId]->top_status_codes.merge(target.second->top_status_codes); + _targets_metrics[targetId]->dns_response_failures += target.second->dns_response_failures; + _targets_metrics[targetId]->top_rcodes.merge(target.second->top_rcodes); } if (group_enabled(group::NetProbeMetrics::Histograms)) { _targets_metrics[targetId]->h_time_us.merge(target.second->h_time_us); @@ -134,6 +157,12 @@ void NetProbeMetricsBucket::specialized_merge(const AbstractMetricsBucket &o, Me if (group_enabled(group::NetProbeMetrics::Quantiles)) { _targets_metrics[targetId]->q_time_us.merge(target.second->q_time_us, agg_operator); } + if (group_enabled(group::NetProbeMetrics::HttpResponsePhases)) { + _targets_metrics[targetId]->q_dns_us.merge(target.second->q_dns_us, agg_operator); + _targets_metrics[targetId]->q_connect_us.merge(target.second->q_connect_us, agg_operator); + _targets_metrics[targetId]->q_tls_us.merge(target.second->q_tls_us, agg_operator); + _targets_metrics[targetId]->q_ttfb_us.merge(target.second->q_ttfb_us, agg_operator); + } } } @@ -152,6 +181,10 @@ void NetProbeMetricsBucket::to_prometheus(PrometheusSerializer &ser, Metric::Lab target.second->connect_failures.to_prometheus(ser, target_labels); target.second->dns_failures.to_prometheus(ser, target_labels); target.second->timed_out.to_prometheus(ser, target_labels); + target.second->http_status_failures.to_prometheus(ser, target_labels); + target.second->top_status_codes.to_prometheus(ser, target_labels); + target.second->dns_response_failures.to_prometheus(ser, target_labels); + target.second->top_rcodes.to_prometheus(ser, target_labels); } bool h_max_min{true}; @@ -190,6 +223,16 @@ void NetProbeMetricsBucket::to_prometheus(PrometheusSerializer &ser, Metric::Lab } catch (const std::exception &) { } } + + if (group_enabled(group::NetProbeMetrics::HttpResponsePhases)) { + try { + target.second->q_dns_us.to_prometheus(ser, target_labels); + target.second->q_connect_us.to_prometheus(ser, target_labels); + target.second->q_tls_us.to_prometheus(ser, target_labels); + target.second->q_ttfb_us.to_prometheus(ser, target_labels); + } catch (const std::exception &) { + } + } } } @@ -208,6 +251,10 @@ void NetProbeMetricsBucket::to_opentelemetry(metrics::v1::ScopeMetrics &scope, t target.second->connect_failures.to_opentelemetry(scope, start_ts, end_ts, target_labels); target.second->dns_failures.to_opentelemetry(scope, start_ts, end_ts, target_labels); target.second->timed_out.to_opentelemetry(scope, start_ts, end_ts, target_labels); + target.second->http_status_failures.to_opentelemetry(scope, start_ts, end_ts, target_labels); + target.second->top_status_codes.to_opentelemetry(scope, start_ts, end_ts, target_labels); + target.second->dns_response_failures.to_opentelemetry(scope, start_ts, end_ts, target_labels); + target.second->top_rcodes.to_opentelemetry(scope, start_ts, end_ts, target_labels); } bool h_max_min{true}; @@ -246,6 +293,16 @@ void NetProbeMetricsBucket::to_opentelemetry(metrics::v1::ScopeMetrics &scope, t } catch (const std::exception &) { } } + + if (group_enabled(group::NetProbeMetrics::HttpResponsePhases)) { + try { + target.second->q_dns_us.to_opentelemetry(scope, start_ts, end_ts, target_labels); + target.second->q_connect_us.to_opentelemetry(scope, start_ts, end_ts, target_labels); + target.second->q_tls_us.to_opentelemetry(scope, start_ts, end_ts, target_labels); + target.second->q_ttfb_us.to_opentelemetry(scope, start_ts, end_ts, target_labels); + } catch (const std::exception &) { + } + } } } @@ -263,6 +320,10 @@ void NetProbeMetricsBucket::to_json(json &j) const target.second->connect_failures.to_json(j["targets"][targetId]); target.second->dns_failures.to_json(j["targets"][targetId]); target.second->timed_out.to_json(j["targets"][targetId]); + target.second->http_status_failures.to_json(j["targets"][targetId]); + target.second->top_status_codes.to_json(j["targets"][targetId]); + target.second->dns_response_failures.to_json(j["targets"][targetId]); + target.second->top_rcodes.to_json(j["targets"][targetId]); } bool h_max_min{true}; @@ -301,6 +362,16 @@ void NetProbeMetricsBucket::to_json(json &j) const } catch (const std::exception &) { } } + + if (group_enabled(group::NetProbeMetrics::HttpResponsePhases)) { + try { + target.second->q_dns_us.to_json(j["targets"][targetId]); + target.second->q_connect_us.to_json(j["targets"][targetId]); + target.second->q_tls_us.to_json(j["targets"][targetId]); + target.second->q_ttfb_us.to_json(j["targets"][targetId]); + } catch (const std::exception &) { + } + } } } @@ -308,11 +379,23 @@ void NetProbeMetricsBucket::process_filtered() { } -void NetProbeMetricsBucket::process_failure(ErrorType error, const std::string &target) +Target &NetProbeMetricsBucket::get_or_create_target(const std::string &target) { - if (!_targets_metrics.count(target)) { - _targets_metrics[target] = std::make_unique(); + auto it = _targets_metrics.find(target); + if (it == _targets_metrics.end()) { + auto t = std::make_unique(); + // Honor topn_count / topn_percentile_threshold on the per-target TopN metrics. + t->top_status_codes.set_settings(_topn_count, _topn_percentile_threshold); + t->top_rcodes.set_settings(_topn_count, _topn_percentile_threshold); + it = _targets_metrics.emplace(target, std::move(t)).first; } + return *it->second; +} + +void NetProbeMetricsBucket::process_failure(ErrorType error, const std::string &target) +{ + std::unique_lock lock(_mutex); + get_or_create_target(target); if (group_enabled(group::NetProbeMetrics::Counters)) { switch (error) { case ErrorType::DnsLookupFailure: @@ -340,9 +423,8 @@ bool NetProbeStreamHandler::_filtering([[maybe_unused]] pcpp::Packet *payload) void NetProbeMetricsBucket::process_attempts([[maybe_unused]] bool deep, const std::string &target) { - if (!_targets_metrics.count(target)) { - _targets_metrics[target] = std::make_unique(); - } + std::unique_lock lock(_mutex); + get_or_create_target(target); if (group_enabled(group::NetProbeMetrics::Counters)) { ++_targets_metrics[target]->attempts; } @@ -352,9 +434,7 @@ void NetProbeMetricsBucket::new_transaction(bool deep, NetProbeTransaction xact) { std::unique_lock lock(_mutex); - if (!_targets_metrics.count(xact.target)) { - _targets_metrics[xact.target] = std::make_unique(); - } + get_or_create_target(xact.target); if (group_enabled(group::NetProbeMetrics::Counters)) { ++_targets_metrics[xact.target]->successes; } @@ -448,4 +528,116 @@ void NetProbeMetricsManager::process_filtered(timespec stamp) live_bucket()->process_filtered(); } -} \ No newline at end of file +void NetProbeMetricsBucket::process_netprobe_http(bool deep, uint16_t status, const visor::http::HttpTimings &timings, const std::string &target) +{ + // Take _mutex like new_transaction — both mutate q_time_us/h_time_us sketches + // and the TopN which the scrape thread reads under shared_lock. + // process_failure/process_attempts also lock _mutex (for their map insert), but + // their callers always invoke them sequentially — never while this lock is held. + std::unique_lock lock(_mutex); + + auto &t = get_or_create_target(target); + + // Counters (status outcome) are always recorded when the group is on — + // like new_transaction's successes++ (not gated on `deep`). + if (group_enabled(group::NetProbeMetrics::Counters)) { + t.top_status_codes.update(std::to_string(status)); + if (status >= 200 && status < 400) { + ++t.successes; + } else { + ++t.http_status_failures; + } + } + + // Sketches are gated on `deep` (deep sampling) exactly like new_transaction. + // Histograms is default-ON and drives response_min_us/max_us via h_time_us. + if (deep && group_enabled(group::NetProbeMetrics::Histograms)) { + t.h_time_us.update(timings.total_us); + } + if (deep && group_enabled(group::NetProbeMetrics::Quantiles)) { + t.q_time_us.update(timings.total_us); + } + if (deep && group_enabled(group::NetProbeMetrics::HttpResponsePhases)) { + t.q_dns_us.update(timings.dns_us); + t.q_connect_us.update(timings.connect_us); + t.q_tls_us.update(timings.tls_us); + t.q_ttfb_us.update(timings.ttfb_us); + } +} + +void NetProbeMetricsManager::process_netprobe_http_result(uint16_t status, const visor::http::HttpTimings &timings, const std::string &target, timespec stamp) +{ + new_event(stamp); + live_bucket()->process_attempts(_deep_sampling_now, target); + live_bucket()->process_netprobe_http(_deep_sampling_now, status, timings, target); +} + +void NetProbeMetricsManager::process_netprobe_http_failure(ErrorType error, const std::string &target) +{ + timespec stamp; + std::timespec_get(&stamp, TIME_UTC); + new_event(stamp); + live_bucket()->process_attempts(_deep_sampling_now, target); + live_bucket()->process_failure(error, target); +} + +void NetProbeMetricsBucket::process_netprobe_doh(bool deep, uint16_t http_status, uint8_t rcode, bool parse_ok, const visor::http::HttpTimings &timings, const std::string &target) +{ + std::unique_lock lock(_mutex); + + auto &t = get_or_create_target(target); + + if (group_enabled(group::NetProbeMetrics::Counters)) { + // DoH responses are HTTP responses too: record the HTTP status breakdown (like the HTTP + // probe) in addition to the DNS rcode breakdown below. + t.top_status_codes.update(std::to_string(http_status)); + if (http_status >= 200 && http_status < 400) { + std::string rname; + if (!parse_ok) { + rname = "PARSE_ERROR"; + } else { + auto it = visor::lib::dns::RCodeNames.find(rcode); + rname = (it != visor::lib::dns::RCodeNames.end()) ? it->second : std::to_string(rcode); + } + t.top_rcodes.update(rname); + if (parse_ok && rcode == 0) { + ++t.successes; + } else { + ++t.dns_response_failures; + } + } else { + ++t.http_status_failures; + } + } + + if (deep && group_enabled(group::NetProbeMetrics::Histograms)) { + t.h_time_us.update(timings.total_us); + } + if (deep && group_enabled(group::NetProbeMetrics::Quantiles)) { + t.q_time_us.update(timings.total_us); + } + if (deep && group_enabled(group::NetProbeMetrics::HttpResponsePhases)) { + t.q_dns_us.update(timings.dns_us); + t.q_connect_us.update(timings.connect_us); + t.q_tls_us.update(timings.tls_us); + t.q_ttfb_us.update(timings.ttfb_us); + } +} + +void NetProbeMetricsManager::process_netprobe_doh_result(uint16_t http_status, uint8_t rcode, bool parse_ok, const visor::http::HttpTimings &timings, const std::string &target, timespec stamp) +{ + new_event(stamp); + live_bucket()->process_attempts(_deep_sampling_now, target); + live_bucket()->process_netprobe_doh(_deep_sampling_now, http_status, rcode, parse_ok, timings, target); +} + +void NetProbeMetricsManager::process_netprobe_doh_failure(ErrorType error, const std::string &target) +{ + timespec stamp; + std::timespec_get(&stamp, TIME_UTC); + new_event(stamp); + live_bucket()->process_attempts(_deep_sampling_now, target); + live_bucket()->process_failure(error, target); +} + +} diff --git a/src/handlers/netprobe/NetProbeStreamHandler.h b/src/handlers/netprobe/NetProbeStreamHandler.h index c46dc61c9..b8ff5c716 100644 --- a/src/handlers/netprobe/NetProbeStreamHandler.h +++ b/src/handlers/netprobe/NetProbeStreamHandler.h @@ -5,6 +5,7 @@ #pragma once #include "AbstractMetricsManager.h" +#include "HttpTypes.h" #include "PrometheusSerializer.h" #include "NetProbeInputStream.h" #include "StreamHandler.h" @@ -35,7 +36,8 @@ namespace group { enum NetProbeMetrics : visor::MetricGroupIntType { Counters, Quantiles, - Histograms + Histograms, + HttpResponsePhases }; } @@ -53,6 +55,14 @@ struct Target { Counter connect_failures; Counter dns_failures; Counter timed_out; + Counter http_status_failures; + TopN top_status_codes; + Counter dns_response_failures; + TopN top_rcodes; + Quantile q_dns_us; + Quantile q_connect_us; + Quantile q_tls_us; + Quantile q_ttfb_us; Target() : q_time_us(NET_PROBE_SCHEMA, {"response_quantiles_us"}, "Net Probe quantile in microseconds") @@ -64,6 +74,14 @@ struct Target { , connect_failures(NET_PROBE_SCHEMA, {"connect_failures"}, "Total Net Probe failures when performing a TCP socket connection") , dns_failures(NET_PROBE_SCHEMA, {"dns_lookup_failures"}, "Total Net Probe failures when performing a DNS lookup") , timed_out(NET_PROBE_SCHEMA, {"packets_timeout"}, "Total Net Probe timeout transactions") + , http_status_failures(NET_PROBE_SCHEMA, {"http_status_failures"}, "Total HTTP/DoH responses with a non-success status (any HTTP status outside 2xx/3xx, e.g. 4xx/5xx)") + , top_status_codes(NET_PROBE_SCHEMA, "status_code", {"top_status_codes"}, "Top HTTP status codes") + , dns_response_failures(NET_PROBE_SCHEMA, {"dns_response_failures"}, "Total DoH responses with a success HTTP status (2xx/3xx) but a non-NOERROR or unparseable DNS response") + , top_rcodes(NET_PROBE_SCHEMA, "rcode", {"top_rcodes"}, "Top DNS response codes observed") + , q_dns_us(NET_PROBE_SCHEMA, {"response_dns_us"}, "DNS resolution time quantiles in microseconds") + , q_connect_us(NET_PROBE_SCHEMA, {"response_connect_us"}, "TCP connect time quantiles in microseconds") + , q_tls_us(NET_PROBE_SCHEMA, {"response_tls_us"}, "TLS handshake time quantiles in microseconds") + , q_ttfb_us(NET_PROBE_SCHEMA, {"response_ttfb_us"}, "Time-to-first-byte quantiles in microseconds") { } }; @@ -74,6 +92,14 @@ class NetProbeMetricsBucket final : public visor::AbstractMetricsBucket protected: mutable std::shared_mutex _mutex; std::map> _targets_metrics; + // Top-N settings (from topn_count / topn_percentile_threshold) applied to each target's TopN + // metrics at creation — see get_or_create_target()/update_topn_metrics(). + size_t _topn_count{10}; + uint64_t _topn_percentile_threshold{0}; + + // Return the metrics for `target`, creating it (with the current top-N settings applied to its + // TopN metrics) if it doesn't exist yet. Callers must already hold _mutex. + Target &get_or_create_target(const std::string &target); public: NetProbeMetricsBucket() @@ -85,8 +111,17 @@ class NetProbeMetricsBucket final : public visor::AbstractMetricsBucket void to_json(json &j) const override; void to_prometheus(PrometheusSerializer &ser, Metric::LabelMap add_labels = {}) const override; void to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels = {}) const override; - void update_topn_metrics(size_t, uint64_t) override + void update_topn_metrics(size_t count, uint64_t percentile_threshold) override { + // Called on a fresh bucket at period-shift (before any targets exist), so store the + // settings and apply them per target at creation (get_or_create_target). Also apply to + // any targets already present, to be safe. + _topn_count = count; + _topn_percentile_threshold = percentile_threshold; + for (auto &[name, t] : _targets_metrics) { + t->top_status_codes.set_settings(count, percentile_threshold); + t->top_rcodes.set_settings(count, percentile_threshold); + } } void on_set_read_only() override @@ -97,6 +132,8 @@ class NetProbeMetricsBucket final : public visor::AbstractMetricsBucket void process_failure(ErrorType error, const std::string &target); void process_attempts(bool deep, const std::string &target); void new_transaction(bool deep, NetProbeTransaction xact); + void process_netprobe_http(bool deep, uint16_t status, const visor::http::HttpTimings &timings, const std::string &target); + void process_netprobe_doh(bool deep, uint16_t http_status, uint8_t rcode, bool parse_ok, const visor::http::HttpTimings &timings, const std::string &target); }; class NetProbeMetricsManager final : public visor::AbstractMetricsManager @@ -127,6 +164,10 @@ class NetProbeMetricsManager final : public visor::AbstractMetricsManager @@ -138,6 +179,8 @@ class NetProbeStreamHandler final : public visor::StreamMetricsHandler.target` | string | — | Full URL to probe, e.g. `http://example.com/health` | +| `interval_msec` | uint64 | 5000 | How often to issue a probe, in milliseconds | +| `timeout_msec` | uint64 | 2000 | Per-request timeout in milliseconds (must not exceed `interval_msec`). `0` disables the per-request timeout — not recommended for HTTP, where a slow/stalled server could leave a transfer pending; keep the default. | +| `http_method` | string | `"GET"` | HTTP method to use for all targets (`GET`, `HEAD`, `POST`, …) | + +#### Success semantics + +HTTP probes classify results by status code: + +- **2xx / 3xx** → counted as a `success` +- **4xx / 5xx** (or any other non-2xx/3xx status including `0` / `1xx`) → counted as an `http_status_failures` +- Transport errors (DNS resolution failure, TCP connect failure, timeout) → counted in the corresponding existing failure counter (`dns_lookup_failures`, `connect_failures`, `packets_timeout`) + +### doh + +Issues DNS-over-HTTPS (DoH) queries to one or more URL targets using a libcurl-backed async transport (RFC 8484). +Like `http`, targets are specified as full URLs (e.g. `https://1.1.1.1/dns-query`). + +#### Configuration + +| Key | Type | Default | Description | +|-----|------|---------|-------------| +| `test_type` | string | — | Set to `"doh"` to enable DoH probing | +| `targets..target` | string | — | DoH URL to probe, e.g. `https://1.1.1.1/dns-query` | +| `qname` | string | **required** | DNS name to query (e.g. `example.com`) | +| `qtype` | string | `"A"` | DNS query type (e.g. `A`, `AAAA`, `MX`, `TXT`) | +| `interval_msec` | uint64 | 5000 | How often to issue a probe, in milliseconds | +| `timeout_msec` | uint64 | 2000 | Per-request timeout in milliseconds (must not exceed `interval_msec`) | +| `http_method` | string | `"POST"` | HTTP method to use for the DoH wire-format query (`POST` or `GET`) | + +#### Success semantics + +A DoH probe is counted as a **success** only when both of the following are true: + +1. The HTTP response status is **2xx or 3xx** +2. The DNS response is parseable (QR=1, size ≥ 12 bytes) **and** the DNS rcode is **NOERROR** (0) + +Other outcomes are classified as: + +- **HTTP 4xx/5xx** → counted as `http_status_failures` +- **HTTP 2xx/3xx but non-NOERROR or unparseable DNS response** → counted as `dns_response_failures` +- **Transport errors** (DNS resolution failure, TCP connect failure, timeout) → counted in the corresponding failure counter (`dns_lookup_failures`, `connect_failures`, `packets_timeout`) + +The `top_rcodes` TopN metric records the DNS rcode name (e.g. `NOERROR`, `NXDOMAIN`, `SRVFAIL`) for every response with a 2xx/3xx HTTP status. Unparseable responses are recorded as `PARSE_ERROR`. + +Response-time metrics (`response_histogram_us`, `response_quantiles_us`, `response_min_us`, `response_max_us`) and HTTP response phase metrics (`response_dns_us`, `response_connect_us`, `response_tls_us`, `response_ttfb_us`) apply to DoH probes in exactly the same way as for `http` probes. + +#### Example configuration (YAML policy) + +```yaml +handlers: + modules: + netprobe_doh: + type: netprobe + +input: + module: netprobe + config: + test_type: doh + qname: example.com + qtype: A + http_method: POST + interval_msec: 5000 + timeout_msec: 2000 + targets: + cloudflare_doh: + target: "https://1.1.1.1/dns-query" + google_doh: + target: "https://8.8.8.8/dns-query" +``` + +To enable HTTP response phase timing for DoH: + +```yaml +handlers: + modules: + netprobe_doh: + type: netprobe + config: + enable: + - http_response_phases +``` + +--- + +## Metrics + +All metrics are per-target (keyed by the name given in the `targets` config map). + +### Counters (group: `counters`, default ON) + +| Metric | Description | +|--------|-------------| +| `attempts` | Total probe attempts | +| `successes` | Total successful probes | +| `connect_failures` | TCP/socket connection failures | +| `dns_lookup_failures` | DNS resolution failures | +| `packets_timeout` | Probes that timed out | +| `http_status_failures` | HTTP/DoH responses with any HTTP status outside 2xx/3xx (e.g. 4xx/5xx, and also 1xx or 0) | +| `top_status_codes` | Top HTTP status codes observed (e.g. `"200"`, `"404"`, `"503"`) | +| `dns_response_failures` | DoH responses with HTTP 2xx/3xx but a non-NOERROR or unparseable DNS rcode | +| `top_rcodes` | Top DNS response codes observed in DoH probes (e.g. `"NOERROR"`, `"NXDOMAIN"`, `"SRVFAIL"`, `"PARSE_ERROR"`) | + +### Histograms (group: `histograms`, default ON) + +| Metric | Description | +|--------|-------------| +| `response_histogram_us` | Histogram of total response times in microseconds | +| `response_min_us` | Minimum total response time in microseconds (within the reporting interval); derived from the histogram | +| `response_max_us` | Maximum total response time in microseconds; derived from the histogram | + +### Quantiles (group: `quantiles`) + +| Metric | Description | +|--------|-------------| +| `response_quantiles_us` | Quantiles of total response times in microseconds | + +### HTTP response phases (group: `http_response_phases`, opt-in) + +These metrics are only emitted when the `http_response_phases` group is explicitly enabled +(e.g. `enable: [http_response_phases]` in the handler config). + +| Metric | Description | +|--------|-------------| +| `response_dns_us` | DNS resolution time quantiles in microseconds | +| `response_connect_us` | TCP connect time quantiles in microseconds | +| `response_tls_us` | TLS handshake time quantiles in microseconds | +| `response_ttfb_us` | Time-to-first-byte quantiles in microseconds | + +--- + +## Example configuration (YAML policy) + +```yaml +handlers: + modules: + netprobe_http: + type: netprobe + +input: + module: netprobe + config: + test_type: http + http_method: GET + interval_msec: 5000 + timeout_msec: 2000 + targets: + health_check: + target: "http://my-service:8080/healthz" + api_endpoint: + target: "https://api.example.com/ping" +``` + +To enable HTTP response phase timing: + +```yaml +handlers: + modules: + netprobe_http: + type: netprobe + config: + enable: + - http_response_phases +``` diff --git a/src/handlers/netprobe/test_net_probe.cpp b/src/handlers/netprobe/test_net_probe.cpp index 6b9a6a322..be02f5a3a 100644 --- a/src/handlers/netprobe/test_net_probe.cpp +++ b/src/handlers/netprobe/test_net_probe.cpp @@ -417,6 +417,276 @@ TEST_CASE("NetProbe specialized_merge aggregates targets across buckets", "[netp CHECK(j["targets"]["only-in-b"]["dns_lookup_failures"] == 1); } +TEST_CASE("NetProbe HTTP status-aware metrics: counters and top_status_codes", "[netprobe][http][unit]") +{ + UnitFixture fx; + + timespec stamp{1000, 0}; + auto timings = visor::http::HttpTimings{/*total_us*/ 1234, /*dns_us*/ 100, /*connect_us*/ 200, /*tls_us*/ 300, /*ttfb_us*/ 400}; + + // 200 → success; 404 → http_status_failures; ConnectFailure → connect_failures + fx.manager()->process_netprobe_http_result(200, timings, "t1", stamp); + fx.manager()->process_netprobe_http_result(404, timings, "t1", stamp); + fx.manager()->process_netprobe_http_failure(ErrorType::ConnectFailure, "t1"); + + json j; + fx.manager()->bucket(0)->to_json(j); + + CHECK(j["targets"]["t1"]["attempts"] == 3); + CHECK(j["targets"]["t1"]["successes"] == 1); + CHECK(j["targets"]["t1"]["http_status_failures"] == 1); + CHECK(j["targets"]["t1"]["connect_failures"] == 1); + + // top_status_codes should contain entries for "200" and "404" + REQUIRE(j["targets"]["t1"].contains("top_status_codes")); + auto &top = j["targets"]["t1"]["top_status_codes"]; + bool found_200 = false; + bool found_404 = false; + for (const auto &entry : top) { + if (entry.contains("name") && entry["name"] == "200") { + found_200 = true; + } + if (entry.contains("name") && entry["name"] == "404") { + found_404 = true; + } + } + CHECK(found_200); + CHECK(found_404); + + // Histograms is default-ON so response_min_us and response_max_us should appear + CHECK(j["targets"]["t1"].contains("response_min_us")); + CHECK(j["targets"]["t1"].contains("response_max_us")); +} + +TEST_CASE("NetProbe HTTP status boundary: 3xx is success, 1xx and 0 are failures", "[netprobe][http][unit]") +{ + UnitFixture fx; + + timespec stamp{2000, 0}; + auto timings = visor::http::HttpTimings{500, 50, 100, 0, 200}; + + fx.manager()->process_netprobe_http_result(301, timings, "redir", stamp); // 3xx → success + fx.manager()->process_netprobe_http_result(500, timings, "redir", stamp); // 5xx → http_status_failures + fx.manager()->process_netprobe_http_result(100, timings, "redir", stamp); // 1xx → http_status_failures + fx.manager()->process_netprobe_http_result(0, timings, "redir", stamp); // 0 → http_status_failures + + json j; + fx.manager()->bucket(0)->to_json(j); + + CHECK(j["targets"]["redir"]["attempts"] == 4); + CHECK(j["targets"]["redir"]["successes"] == 1); + CHECK(j["targets"]["redir"]["http_status_failures"] == 3); +} + +TEST_CASE("NetProbe HTTP http_response_phases group: quantiles present when enabled", "[netprobe][http][unit]") +{ + visor::Config c; + c.config_set("num_periods", 1); + NetProbeInputStream stream{"netprobe-http-phases"}; + auto *proxy = stream.add_event_proxy(c); + auto handler = std::make_unique("netprobe-http-phases", proxy, &c); + // Enable http_response_phases group before starting (via "enable" config key) + handler->config_set("enable", {"http_response_phases"}); + handler->start(); + + auto *mgr = const_cast(handler->metrics()); + + timespec stamp{3000, 0}; + auto timings = visor::http::HttpTimings{2000, 150, 300, 400, 600}; + mgr->process_netprobe_http_result(200, timings, "phases-tgt", stamp); + + json j; + mgr->bucket(0)->to_json(j); + + // response_ttfb_us should appear when http_response_phases is enabled + CHECK(j["targets"]["phases-tgt"].contains("response_ttfb_us")); + CHECK(j["targets"]["phases-tgt"].contains("response_dns_us")); + CHECK(j["targets"]["phases-tgt"].contains("response_connect_us")); + CHECK(j["targets"]["phases-tgt"].contains("response_tls_us")); + + handler->stop(); +} + +TEST_CASE("NetProbe HTTP http_response_phases group: quantiles absent when not enabled", "[netprobe][http][unit]") +{ + // Default fixture has Counters + Histograms enabled, NOT HttpResponsePhases + UnitFixture fx; + + timespec stamp{4000, 0}; + auto timings = visor::http::HttpTimings{2000, 150, 300, 400, 600}; + fx.manager()->process_netprobe_http_result(200, timings, "no-phases", stamp); + + json j; + fx.manager()->bucket(0)->to_json(j); + + CHECK(!j["targets"]["no-phases"].contains("response_ttfb_us")); + CHECK(!j["targets"]["no-phases"].contains("response_dns_us")); + CHECK(!j["targets"]["no-phases"].contains("response_connect_us")); + CHECK(!j["targets"]["no-phases"].contains("response_tls_us")); +} + +TEST_CASE("NetProbe HTTP metrics merge across buckets", "[netprobe][http][unit]") +{ + UnitFixture fx_a(2); + UnitFixture fx_b(2); + + timespec stamp{5000, 0}; + auto timings = visor::http::HttpTimings{1000, 50, 100, 0, 300}; + + fx_a.manager()->process_netprobe_http_result(200, timings, "shared", stamp); + fx_a.manager()->process_netprobe_http_result(404, timings, "shared", stamp); + fx_b.manager()->process_netprobe_http_result(500, timings, "shared", stamp); + + UnitFixture fx_merged(2); + auto *merged = const_cast(fx_merged.manager()->bucket(0)); + merged->specialized_merge(*fx_a.manager()->bucket(0), visor::Metric::Aggregate::DEFAULT); + merged->specialized_merge(*fx_b.manager()->bucket(0), visor::Metric::Aggregate::DEFAULT); + + json j; + merged->to_json(j); + + CHECK(j["targets"]["shared"]["successes"] == 1); + CHECK(j["targets"]["shared"]["http_status_failures"] == 2); + + // top_status_codes should reflect merged counts + REQUIRE(j["targets"]["shared"].contains("top_status_codes")); +} + +TEST_CASE("NetProbe DoH DNS-aware metrics: counters and top_rcodes", "[netprobe][doh][unit]") +{ + UnitFixture fx; + + timespec stamp{6000, 0}; + auto timings = visor::http::HttpTimings{/*total_us*/ 1234, /*dns_us*/ 100, /*connect_us*/ 200, /*tls_us*/ 300, /*ttfb_us*/ 400}; + + // 200 + rcode 0 + parse_ok=true → success + fx.manager()->process_netprobe_doh_result(200, 0, true, timings, "t1", stamp); + // 200 + rcode 2 + parse_ok=true → dns_response_failures (SRVFAIL) + fx.manager()->process_netprobe_doh_result(200, 2, true, timings, "t1", stamp); + // 200 + rcode 0 + parse_ok=false → dns_response_failures (PARSE_ERROR) + fx.manager()->process_netprobe_doh_result(200, 0, false, timings, "t1", stamp); + // 503 + parse_ok=false → http_status_failures + fx.manager()->process_netprobe_doh_result(503, 0, false, timings, "t1", stamp); + // transport failure → connect_failures + fx.manager()->process_netprobe_doh_failure(ErrorType::ConnectFailure, "t1"); + + json j; + fx.manager()->bucket(0)->to_json(j); + + CHECK(j["targets"]["t1"]["attempts"] == 5); + CHECK(j["targets"]["t1"]["successes"] == 1); + CHECK(j["targets"]["t1"]["dns_response_failures"] == 2); + CHECK(j["targets"]["t1"]["http_status_failures"] == 1); + CHECK(j["targets"]["t1"]["connect_failures"] == 1); + + // top_rcodes should contain entries for "NOERROR", "SRVFAIL", "PARSE_ERROR" + REQUIRE(j["targets"]["t1"].contains("top_rcodes")); + auto &top = j["targets"]["t1"]["top_rcodes"]; + bool found_noerror = false; + bool found_srvfail = false; + bool found_parse_error = false; + for (const auto &entry : top) { + if (entry.contains("name") && entry["name"] == "NOERROR") { + found_noerror = true; + CHECK(entry["estimate"] == 1); // fed exactly once + } + if (entry.contains("name") && entry["name"] == "SRVFAIL") { + found_srvfail = true; + CHECK(entry["estimate"] == 1); + } + if (entry.contains("name") && entry["name"] == "PARSE_ERROR") { + found_parse_error = true; + CHECK(entry["estimate"] == 1); + } + } + CHECK(found_noerror); + CHECK(found_srvfail); + CHECK(found_parse_error); + // The 503 (HTTP-failure) response must NOT add an rcode entry — top_rcodes is only + // recorded for HTTP-success responses, so exactly the three above are present. + CHECK(top.size() == 3); + + // Histograms is default-ON so response_min_us and response_max_us should appear + CHECK(j["targets"]["t1"].contains("response_min_us")); + CHECK(j["targets"]["t1"].contains("response_max_us")); +} + +TEST_CASE("NetProbe DoH http_response_phases group: quantiles present when enabled", "[netprobe][doh][unit]") +{ + visor::Config c; + c.config_set("num_periods", 1); + NetProbeInputStream stream{"netprobe-doh-phases"}; + auto *proxy = stream.add_event_proxy(c); + auto handler = std::make_unique("netprobe-doh-phases", proxy, &c); + handler->config_set("enable", {"http_response_phases"}); + handler->start(); + + auto *mgr = const_cast(handler->metrics()); + + timespec stamp{7000, 0}; + auto timings = visor::http::HttpTimings{2000, 150, 300, 400, 600}; + mgr->process_netprobe_doh_result(200, 0, true, timings, "doh-phases-tgt", stamp); + + json j; + mgr->bucket(0)->to_json(j); + + CHECK(j["targets"]["doh-phases-tgt"].contains("response_ttfb_us")); + CHECK(j["targets"]["doh-phases-tgt"].contains("response_dns_us")); + CHECK(j["targets"]["doh-phases-tgt"].contains("response_connect_us")); + CHECK(j["targets"]["doh-phases-tgt"].contains("response_tls_us")); + // Value assertions (single update => p50 is the fed value): proves each phase field is + // wired to the right HttpTimings member (dns_us=150, ttfb_us=600), not swapped. + CHECK(j["targets"]["doh-phases-tgt"]["response_dns_us"]["p50"] == 150); + CHECK(j["targets"]["doh-phases-tgt"]["response_ttfb_us"]["p50"] == 600); + + handler->stop(); +} + +TEST_CASE("NetProbe DoH top_rcodes honors topn_count", "[netprobe][doh][unit]") +{ + visor::Config c; + c.config_set("num_periods", 1); + c.config_set("topn_count", 2); // cap TopN output at 2 entries + NetProbeInputStream stream{"netprobe-doh-topn"}; + auto *proxy = stream.add_event_proxy(c); + auto handler = std::make_unique("netprobe-doh-topn", proxy, &c); + handler->start(); + + auto *mgr = const_cast(handler->metrics()); + timespec stamp{7000, 0}; + auto timings = visor::http::HttpTimings{1000, 0, 0, 0, 0}; + // Feed 3 distinct rcodes; with topn_count=2 the settings must be applied to the per-target + // TopN, so top_rcodes emits at most 2 entries (without the fix it would emit all 3). + mgr->process_netprobe_doh_result(200, 0, true, timings, "t1", stamp); // NOERROR + mgr->process_netprobe_doh_result(200, 2, true, timings, "t1", stamp); // SRVFAIL + mgr->process_netprobe_doh_result(200, 3, true, timings, "t1", stamp); // NXDOMAIN + + json j; + mgr->bucket(0)->to_json(j); + REQUIRE(j["targets"]["t1"].contains("top_rcodes")); + CHECK(j["targets"]["t1"]["top_rcodes"].size() <= 2); + + handler->stop(); +} + +TEST_CASE("NetProbe DoH http_response_phases group: quantiles absent when not enabled", "[netprobe][doh][unit]") +{ + // Default fixture has Counters + Histograms enabled, NOT HttpResponsePhases + UnitFixture fx; + + timespec stamp{8000, 0}; + auto timings = visor::http::HttpTimings{2000, 150, 300, 400, 600}; + fx.manager()->process_netprobe_doh_result(200, 0, true, timings, "doh-no-phases", stamp); + + json j; + fx.manager()->bucket(0)->to_json(j); + + CHECK(!j["targets"]["doh-no-phases"].contains("response_ttfb_us")); + CHECK(!j["targets"]["doh-no-phases"].contains("response_dns_us")); + CHECK(!j["targets"]["doh-no-phases"].contains("response_connect_us")); + CHECK(!j["targets"]["doh-no-phases"].contains("response_tls_us")); +} + TEST_CASE("Net Probe TCP tests", "[netprobe][tcp]") { // Requires external network (www.google.com:80) and only asserts diff --git a/src/inputs/netprobe/CMakeLists.txt b/src/inputs/netprobe/CMakeLists.txt index a009e0c22..533c4b0dd 100644 --- a/src/inputs/netprobe/CMakeLists.txt +++ b/src/inputs/netprobe/CMakeLists.txt @@ -3,6 +3,8 @@ message(STATUS "Input Module: Net Probe") add_library(VisorInputNetProbe STATIC NetProbeInputModulePlugin.cpp NetProbeInputStream.cpp + DohProbe.cpp + HttpProbe.cpp PingProbe.cpp TcpProbe.cpp ) @@ -19,6 +21,8 @@ target_link_libraries(VisorInputNetProbe PUBLIC Visor::Core Visor::Lib::Tcp + Visor::Lib::Dns + Visor::Lib::Http uvw::uvw ) @@ -29,8 +33,9 @@ add_executable(unit-tests-input-netprobe test_netprobe.cpp) target_link_libraries(unit-tests-input-netprobe PRIVATE - Visor::Input::NetProbe - Visor::Lib::Test) + Visor::Handler::NetProbe + Visor::Lib::Test + httplib::httplib) add_test(NAME unit-tests-input-netprobe WORKING_DIRECTORY ${PROJECT_SOURCE_DIR}/src diff --git a/src/inputs/netprobe/DohProbe.cpp b/src/inputs/netprobe/DohProbe.cpp new file mode 100644 index 000000000..75bbf2a87 --- /dev/null +++ b/src/inputs/netprobe/DohProbe.cpp @@ -0,0 +1,232 @@ +/* This Source Code Form is subject to the terms of the Mozilla Public + * License, v. 2.0. If a copy of the MPL was not distributed with this + * file, You can obtain one at https://mozilla.org/MPL/2.0/. */ + +#include "DohProbe.h" +#include "NetProbeException.h" +#include "dns.h" +#ifdef __GNUC__ +#pragma GCC diagnostic push +#pragma GCC diagnostic ignored "-Wold-style-cast" +#pragma GCC diagnostic ignored "-Wunused-parameter" +#pragma GCC diagnostic ignored "-Wzero-as-null-pointer-constant" +#endif +#include +#ifdef __GNUC__ +#pragma GCC diagnostic pop +#endif +#include +#include +#include +#include + +namespace visor::input::netprobe { + +// Case-insensitive string equality (DNS names compare case-insensitively). +static bool iequals(const std::string &a, const std::string &b) +{ + if (a.size() != b.size()) { + return false; + } + for (size_t i = 0; i < a.size(); ++i) { + if (std::tolower(static_cast(a[i])) != std::tolower(static_cast(b[i]))) { + return false; + } + } + return true; +} + +// RFC 8484: a DoH response must be Content-Type application/dns-message. Compare the media type +// case-insensitively, ignoring any parameters (e.g. "; charset=...") and surrounding whitespace. +static bool doh_content_type_ok(const std::string &ct) +{ + std::string base = ct.substr(0, ct.find(';')); + auto first = base.find_first_not_of(" \t"); + if (first == std::string::npos) { + return false; + } + auto last = base.find_last_not_of(" \t"); + base = base.substr(first, last - first + 1); + return iequals(base, "application/dns-message"); +} + +// Verify the response's answer + authority sections fully parsed. pcpp's getAnswerCount()/ +// getAuthorityCount() return the HEADER-DECLARED counts, while parseResources() silently stops at +// the first record that runs past the buffer — so a truncated/malformed section leaves the parsed +// record list shorter than declared. Comparing parsed vs declared catches that. A valid NODATA +// response (0 answers) passes (declared==parsed==0). The additional section is intentionally NOT +// checked: it commonly carries an EDNS OPT record, and we don't want any OPT-parsing edge case to +// false-reject a healthy resolver. +static bool doh_response_fully_parsed(pcpp::DnsLayer &dns) +{ + size_t answers_parsed = 0; + for (auto *r = dns.getFirstAnswer(); r != nullptr; r = dns.getNextAnswer(r)) { + ++answers_parsed; + } + if (answers_parsed < dns.getAnswerCount()) { + return false; + } + size_t authority_parsed = 0; + for (auto *r = dns.getFirstAuthority(); r != nullptr; r = dns.getNextAuthority(r)) { + ++authority_parsed; + } + return authority_parsed >= dns.getAuthorityCount(); +} + +// RFC 4648 §5 base64url, no padding (for DoH GET ?dns=). +static std::string base64url(const std::string &in) +{ + static const char tbl[] = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_"; + std::string out; + out.reserve((in.size() + 2) / 3 * 4); + size_t i = 0; + while (i + 3 <= in.size()) { + uint32_t n = (uint8_t(in[i]) << 16) | (uint8_t(in[i + 1]) << 8) | uint8_t(in[i + 2]); + out.push_back(tbl[(n >> 18) & 63]); out.push_back(tbl[(n >> 12) & 63]); + out.push_back(tbl[(n >> 6) & 63]); out.push_back(tbl[n & 63]); + i += 3; + } + if (i + 1 == in.size()) { + uint32_t n = uint8_t(in[i]) << 16; + out.push_back(tbl[(n >> 18) & 63]); out.push_back(tbl[(n >> 12) & 63]); + } else if (i + 2 == in.size()) { + uint32_t n = (uint8_t(in[i]) << 16) | (uint8_t(in[i + 1]) << 8); + out.push_back(tbl[(n >> 18) & 63]); out.push_back(tbl[(n >> 12) & 63]); out.push_back(tbl[(n >> 6) & 63]); + } + return out; +} + +bool DohProbe::start(std::shared_ptr io_loop) +{ + if (_init || _url.empty()) { + return false; + } + // Build the DNS query once (qname/qtype are stream-fixed). qname/qtype were validated at config + // time, but guard addQuery()'s nullable return so a bad name can never yield a zero-question + // probe or use a half-built buffer. + pcpp::DnsLayer q; + _qtype_code = visor::lib::dns::QTypeNumbers.at(_qtype); + // pcpp encodes/decodes the DNS root as an empty name (a single 0x00), not "." — so build the + // query (and later compare the echoed question) using "" for the root. + _wire_qname = (_qname == ".") ? std::string() : _qname; + if (!q.addQuery(_wire_qname, static_cast(_qtype_code), pcpp::DNS_CLASS_IN)) { + throw NetProbeException("netprobe doh: failed to build DNS query for qname '" + _qname + "'"); + } + q.getDnsHeader()->recursionDesired = 1; + q.getDnsHeader()->transactionID = 0; // RFC 8484 §4.1 + _query_wire.assign(reinterpret_cast(q.getData()), q.getDataLen()); + if (_method == "GET") { + _get_url = _url + (_url.find('?') == std::string::npos ? "?" : "&") + "dns=" + base64url(_query_wire); + } + + _io_loop = io_loop; + _interval_timer = _io_loop->resource(); + if (!_interval_timer) { + throw NetProbeException("Netprobe - unable to initialize interval TimerHandle"); + } + _interval_timer->on([this](const auto &, auto &) { + // Same capture contract as HttpProbe: outer lambda captures `this` (timer stopped on the + // loop thread before destruction); inner completion captures BY VALUE. + visor::http::HttpRequest req; + req.timeout_ms = _config.timeout_msec; + req.capture_response = true; + if (_method == "GET") { + req.url = _get_url; + req.method = "GET"; + req.headers = {"Accept: application/dns-message"}; + } else { + req.url = _url; + req.method = "POST"; + req.body = _query_wire; + req.headers = {"Content-Type: application/dns-message", "Accept: application/dns-message"}; + } + const std::string name = _name; + const std::string qname = _wire_qname; // "" for the root; matches what pcpp getName() returns + const uint16_t qtype_code = _qtype_code; + auto doh_result = _doh_result; + auto fail = _fail; + _client->request(req, [doh_result, fail, name, qname, qtype_code](const visor::http::HttpResult &r) { + timespec stamp; + std::timespec_get(&stamp, TIME_UTC); + auto logger = spdlog::get("visor"); + if (r.transport_ok) { + uint8_t rcode = 0; + bool parse_ok = false; + if (!doh_content_type_ok(r.content_type)) { + // RFC 8484: a valid DoH reply is application/dns-message. A wrong content type + // (e.g. an HTML/JSON error page returned with HTTP 200) is a DNS-level failure. + if (logger) { + logger->debug("netprobe doh[{}]: unexpected response Content-Type '{}' (want application/dns-message)", name, r.content_type); + } + } else if (r.response_body.size() >= sizeof(pcpp::dnshdr)) { + // pcpp::DnsLayer, when not attached to a Packet, OWNS its buffer and + // delete[]s it in ~Layer(). So it MUST be given a new[]-allocated buffer + // (a std::string's internal buffer is NOT new[] => delete[] on it is UB / + // heap corruption). Mirror DnsStreamHandler's TCP path: hand the Layer a + // new[] buffer and release ownership to it. + auto rawbuf = std::make_unique(r.response_body.size()); + std::memcpy(rawbuf.get(), r.response_body.data(), r.response_body.size()); + pcpp::DnsLayer dns(rawbuf.release(), r.response_body.size(), nullptr, nullptr); + auto *h = dns.getDnsHeader(); + if (h->queryOrResponse != 1) { + if (logger) { + logger->debug("netprobe doh[{}]: response is not a DNS reply (QR=0)", name); + } + } else { + // Validate the response echoes our question (qname + qtype) per RFC 8484, + // so a mismatched/unrelated answer isn't counted as a success. + auto *query = dns.getFirstQuery(); + if (!(query && static_cast(query->getDnsType()) == qtype_code && iequals(query->getName(), qname))) { + if (logger) { + logger->debug("netprobe doh[{}]: response question mismatch (got '{}' type {})", name, + query ? query->getName() : std::string(""), query ? static_cast(query->getDnsType()) : -1); + } + } else if (h->truncation) { + // Server set the TC (truncated) bit (RFC 1035 §4.1.1): the response is incomplete. + if (logger) { + logger->debug("netprobe doh[{}]: response has the TC (truncated) bit set", name); + } + } else if (!doh_response_fully_parsed(dns)) { + // A declared answer/authority record failed to parse (truncated/malformed). + if (logger) { + logger->debug("netprobe doh[{}]: response truncated/malformed (a declared record did not parse)", name); + } + } else { + parse_ok = true; + rcode = h->responseCode; + } + } + } else if (logger) { + logger->debug("netprobe doh[{}]: response too short for a DNS message ({} bytes)", name, r.response_body.size()); + } + doh_result(static_cast(r.status_code), rcode, parse_ok, r.timings, name, stamp); + } else { + if (logger) { + logger->debug("netprobe doh[{}]: transport error: {} (curl code {})", name, r.error_msg, r.curl_code); + } + ErrorType err = ErrorType::SocketError; + if (r.curl_code == CURLE_COULDNT_RESOLVE_HOST || r.curl_code == CURLE_COULDNT_RESOLVE_PROXY) { + err = ErrorType::DnsLookupFailure; + } else if (r.curl_code == CURLE_COULDNT_CONNECT) { + err = ErrorType::ConnectFailure; + } else if (r.curl_code == CURLE_OPERATION_TIMEDOUT) { + err = ErrorType::Timeout; + } + fail(err, TestType::DOH, name); + } + }); + }); + _interval_timer->start(uvw::timer_handle::time{0}, uvw::timer_handle::time{_config.interval_msec}); + _init = true; + return true; +} + +bool DohProbe::stop() +{ + if (_interval_timer && !_interval_timer->closing()) { + _interval_timer->stop(); + _interval_timer->close(); + } + return true; +} +} diff --git a/src/inputs/netprobe/DohProbe.h b/src/inputs/netprobe/DohProbe.h new file mode 100644 index 000000000..144aad0b2 --- /dev/null +++ b/src/inputs/netprobe/DohProbe.h @@ -0,0 +1,46 @@ +/* This Source Code Form is subject to the terms of the Mozilla Public + * License, v. 2.0. If a copy of the MPL was not distributed with this + * file, You can obtain one at https://mozilla.org/MPL/2.0/. */ + +#pragma once +#include "HttpClient.h" +#include "NetProbe.h" +#include +#include + +namespace visor::input::netprobe { + +// http_status, rcode, parse_ok, timings, name, stamp +using DohResultCallback = std::function; + +class DohProbe final : public NetProbe +{ + std::string _url; + std::string _method; // "POST" or "GET" + std::string _qname; + std::string _qtype; // e.g. "A" + std::shared_ptr _client; + DohResultCallback _doh_result; + std::shared_ptr _interval_timer; + std::string _query_wire; // pre-built DNS query (wire format), built in start() + std::string _get_url; // pre-built URL with ?dns= for GET + uint16_t _qtype_code{0}; // numeric DNS qtype (from QTypeNumbers), for response question validation + std::string _wire_qname; // qname as pcpp encodes/decodes it: "" for the root ("."), else _qname + bool _init{false}; + +public: + DohProbe(uint16_t id, const std::string &name, std::string url, std::string method, + std::string qname, std::string qtype, + std::shared_ptr client, DohResultCallback doh_result) + : NetProbe(id, name, pcpp::IPAddress(), std::string()) + , _url(std::move(url)) + , _method(std::move(method)) + , _qname(std::move(qname)) + , _qtype(std::move(qtype)) + , _client(std::move(client)) + , _doh_result(std::move(doh_result)) {} + ~DohProbe() = default; + bool start(std::shared_ptr io_loop) override; + bool stop() override; +}; +} diff --git a/src/inputs/netprobe/HttpProbe.cpp b/src/inputs/netprobe/HttpProbe.cpp new file mode 100644 index 000000000..f055933a1 --- /dev/null +++ b/src/inputs/netprobe/HttpProbe.cpp @@ -0,0 +1,69 @@ +/* This Source Code Form is subject to the terms of the Mozilla Public + * License, v. 2.0. If a copy of the MPL was not distributed with this + * file, You can obtain one at https://mozilla.org/MPL/2.0/. */ + +#include "HttpProbe.h" +#include "NetProbeException.h" +#include + +namespace visor::input::netprobe { + +bool HttpProbe::start(std::shared_ptr io_loop) +{ + if (_init || _url.empty()) { + return false; + } + _io_loop = io_loop; + _interval_timer = _io_loop->resource(); + if (!_interval_timer) { + throw NetProbeException("Netprobe - unable to initialize interval TimerHandle"); + } + _interval_timer->on([this](const auto &, auto &) { + // The outer timer lambda captures `this` — safe because the interval timer is + // stopped and closed on the loop thread (HttpProbe::stop()) before the probe is + // destroyed, and the loop is single-threaded, so no timer event fires after stop(). + // The inner completion callback captures _http_result/_fail BY VALUE so it can run + // after this timer tick returns without referencing the probe. + visor::http::HttpRequest req; + req.url = _url; + req.method = _method; + req.timeout_ms = _config.timeout_msec; + const std::string name = _name; + auto http_result = _http_result; + auto fail = _fail; + _client->request(req, [http_result, fail, name](const visor::http::HttpResult &r) { + timespec stamp; + std::timespec_get(&stamp, TIME_UTC); + if (r.transport_ok) { + http_result(static_cast(r.status_code), r.timings, name, stamp); + } else { + if (auto logger = spdlog::get("visor")) { + logger->debug("netprobe http[{}]: transport error: {} (curl code {})", name, r.error_msg, r.curl_code); + } + ErrorType err = ErrorType::SocketError; + if (r.curl_code == CURLE_COULDNT_RESOLVE_HOST || r.curl_code == CURLE_COULDNT_RESOLVE_PROXY) { + err = ErrorType::DnsLookupFailure; + } else if (r.curl_code == CURLE_COULDNT_CONNECT) { + err = ErrorType::ConnectFailure; + } else if (r.curl_code == CURLE_OPERATION_TIMEDOUT) { + err = ErrorType::Timeout; + } + fail(err, TestType::HTTP, name); + } + }); + }); + _interval_timer->start(uvw::timer_handle::time{0}, uvw::timer_handle::time{_config.interval_msec}); + _init = true; + return true; +} + +bool HttpProbe::stop() +{ + // Called on the loop thread (matches the netprobe loop-quiescent teardown). + if (_interval_timer && !_interval_timer->closing()) { + _interval_timer->stop(); + _interval_timer->close(); + } + return true; +} +} diff --git a/src/inputs/netprobe/HttpProbe.h b/src/inputs/netprobe/HttpProbe.h new file mode 100644 index 000000000..d9c0b096c --- /dev/null +++ b/src/inputs/netprobe/HttpProbe.h @@ -0,0 +1,36 @@ +/* This Source Code Form is subject to the terms of the Mozilla Public + * License, v. 2.0. If a copy of the MPL was not distributed with this + * file, You can obtain one at https://mozilla.org/MPL/2.0/. */ + +#pragma once +#include "HttpClient.h" +#include "NetProbe.h" +#include +#include + +namespace visor::input::netprobe { + +using HttpResultCallback = std::function; + +class HttpProbe final : public NetProbe +{ + std::string _url; + std::string _method; + std::shared_ptr _client; + HttpResultCallback _http_result; + std::shared_ptr _interval_timer; + bool _init{false}; + +public: + HttpProbe(uint16_t id, const std::string &name, std::string url, std::string method, + std::shared_ptr client, HttpResultCallback http_result) + : NetProbe(id, name, pcpp::IPAddress(), std::string()) + , _url(std::move(url)) + , _method(std::move(method)) + , _client(std::move(client)) + , _http_result(std::move(http_result)) {} + ~HttpProbe() = default; + bool start(std::shared_ptr io_loop) override; + bool stop() override; +}; +} diff --git a/src/inputs/netprobe/NetProbe.h b/src/inputs/netprobe/NetProbe.h index 9d314e239..d6c685787 100644 --- a/src/inputs/netprobe/NetProbe.h +++ b/src/inputs/netprobe/NetProbe.h @@ -32,7 +32,8 @@ enum class TestType { Ping, HTTP, UDP, - TCP + TCP, + DOH }; typedef std::function SendCallback; diff --git a/src/inputs/netprobe/NetProbeInputStream.cpp b/src/inputs/netprobe/NetProbeInputStream.cpp index 0710e40ec..a79654dd0 100644 --- a/src/inputs/netprobe/NetProbeInputStream.cpp +++ b/src/inputs/netprobe/NetProbeInputStream.cpp @@ -3,10 +3,14 @@ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */ #include "NetProbeInputStream.h" +#include "DohProbe.h" +#include "HttpClient.h" +#include "HttpProbe.h" #include "NetProbeException.h" #include "PingProbe.h" #include "TcpProbe.h" #include "ThreadName.h" +#include "dns.h" #include #ifdef __GNUC__ #pragma GCC diagnostic push @@ -93,6 +97,38 @@ void NetProbeInputStream::start() } } + if (config_exists("http_method")) { + _http_method = config_get("http_method"); + } + + if (config_exists("qname")) { + _doh_qname = config_get("qname"); + // Normalize a trailing dot (FQDN form, e.g. "example.com."): pcpp encodes/decodes names + // without it, so keeping it would both corrupt the query wire and break the response + // question-echo comparison in DohProbe. Guard size>1 so the root "." isn't emptied. + if (_doh_qname.size() > 1 && _doh_qname.back() == '.') { + _doh_qname.pop_back(); + } + } + if (config_exists("qtype")) { + _doh_qtype = config_get("qtype"); + // Normalize to uppercase before validation/lookup: QTypeNumbers is keyed by uppercase + // names (e.g. "AAAA"), and the DNS handler uppercases qtype inputs too — so "aaaa" + // should be accepted, consistent with the rest of the codebase. + for (auto &ch : _doh_qtype) { + if (ch >= 'a' && ch <= 'z') { + ch = static_cast(ch - 'a' + 'A'); + } + } + } + // DoH method defaults to POST; reuse the http_method key if set. (Only meaningful for DoH streams.) + if (_type == TestType::DOH) { + _doh_method = config_exists("http_method") ? config_get("http_method") : "POST"; + if (_doh_method != "GET" && _doh_method != "POST") { + throw NetProbeException(fmt::format("unsupported http_method '{}' for doh (use GET or POST)", _doh_method)); + } + } + if (!config_exists("targets")) { throw NetProbeException("no targets specified"); } else { @@ -103,6 +139,22 @@ void NetProbeInputStream::start() if (!config->config_exists("target")) { throw NetProbeException(fmt::format("'{}' does not have key 'target' which is required", key)); } + if (_type == TestType::HTTP) { + auto url = config->config_get("target"); + if (auto err = visor::http::validate_http_url(url)) { + throw NetProbeException(fmt::format("target '{}' {}", key, *err)); + } + _http_targets[key] = url; + continue; + } + if (_type == TestType::DOH) { + auto url = config->config_get("target"); + if (auto err = visor::http::validate_http_url(url)) { + throw NetProbeException(fmt::format("target '{}' {}", key, *err)); + } + _doh_targets[key] = url; + continue; + } uint32_t port{0}; if (!config->config_exists("port") && _type == TestType::TCP) { throw NetProbeException(fmt::format("'{}' does not have key 'port' which is required", key)); @@ -137,6 +189,38 @@ void NetProbeInputStream::start() } } + if (_type == TestType::DOH) { + if (_doh_qname.empty()) { + throw NetProbeException("netprobe: 'qname' is required when test_type is 'doh'"); + } + // Reject names outside DNS limits (RFC 1035): total presentation length <= 253, each label + // 1..63 chars. An invalid name would otherwise make DnsLayer::addQuery() fail at probe start. + if (_doh_qname.size() > 253) { + throw NetProbeException(fmt::format("netprobe: qname '{}' exceeds the 253-character DNS name limit", _doh_qname)); + } + // "." is the DNS root — a valid name with no labels (e.g. probing the root NS set). Skip the + // per-label checks for it; DohProbe builds the root query from an empty name. + if (_doh_qname != ".") { + size_t label_len = 0; + for (char ch : _doh_qname) { + if (ch == '.') { + if (label_len == 0) { + throw NetProbeException(fmt::format("netprobe: qname '{}' has an empty DNS label", _doh_qname)); + } + label_len = 0; + } else if (++label_len > 63) { + throw NetProbeException(fmt::format("netprobe: qname '{}' has a DNS label longer than 63 characters", _doh_qname)); + } + } + if (label_len == 0) { + throw NetProbeException(fmt::format("netprobe: qname '{}' has an empty DNS label", _doh_qname)); + } + } + if (visor::lib::dns::QTypeNumbers.find(_doh_qtype) == visor::lib::dns::QTypeNumbers.end()) { + throw NetProbeException(fmt::format("netprobe: unknown qtype '{}'", _doh_qtype)); + } + } + _create_netprobe_loop(); _running = true; @@ -166,6 +250,22 @@ void NetProbeInputStream::_fail_cb(ErrorType error, TestType type, const std::st } } +void NetProbeInputStream::_http_result_cb(uint16_t status, visor::http::HttpTimings t, const std::string &name, timespec stamp) +{ + std::shared_lock lock(_input_mutex); + for (auto &proxy : _event_proxies) { + static_cast(proxy.get())->probe_http_result_cb(status, t, name, stamp); + } +} + +void NetProbeInputStream::_doh_result_cb(uint16_t http_status, uint8_t rcode, bool parse_ok, visor::http::HttpTimings t, const std::string &name, timespec stamp) +{ + std::shared_lock lock(_input_mutex); + for (auto &proxy : _event_proxies) { + static_cast(proxy.get())->probe_doh_result_cb(http_status, rcode, parse_ok, t, name, stamp); + } +} + void NetProbeInputStream::_create_netprobe_loop() { // main io loop, run in its own thread @@ -193,6 +293,9 @@ void NetProbeInputStream::_create_netprobe_loop() probe->stop(); } } + if ((_type == TestType::HTTP || _type == TestType::DOH) && _http_client) { + _http_client->close(); + } _io_loop->stop(); handle.close(); }); @@ -201,6 +304,10 @@ void NetProbeInputStream::_create_netprobe_loop() handle.close(); }); + if (_type == TestType::HTTP || _type == TestType::DOH) { + _http_client = std::make_shared(_io_loop); + } + _timer = _io_loop->resource(); if (!_timer) { throw NetProbeException("unable to initialize TimerHandle"); @@ -255,6 +362,32 @@ void NetProbeInputStream::_create_netprobe_loop() _probes.push_back(std::move(probe)); } + for (const auto &[key, url] : _http_targets) { + auto probe = std::make_unique(_id, key, url, _http_method, _http_client, + [this](uint16_t status, visor::http::HttpTimings t, const std::string &name, timespec stamp) { _http_result_cb(status, t, name, stamp); }); + ++_id; + probe->set_configs(_interval_msec, _timeout_msec, _packets_per_test, _packets_interval_msec, _packet_payload_size); + probe->set_callbacks([this](pcpp::Packet &payload, TestType type, const std::string &name, timespec stamp) { _send_cb(payload, type, name, stamp); }, + [this](pcpp::Packet &payload, TestType type, const std::string &name, timespec stamp) { _recv_cb(payload, type, name, stamp); }, + [this](ErrorType error, TestType type, const std::string &name) { _fail_cb(error, type, name); }); + probe->start(_io_loop); + _probes.push_back(std::move(probe)); + } + + for (const auto &[key, url] : _doh_targets) { + auto probe = std::make_unique(_id, key, url, _doh_method, _doh_qname, _doh_qtype, _http_client, + [this](uint16_t http_status, uint8_t rcode, bool parse_ok, visor::http::HttpTimings t, const std::string &name, timespec stamp) { + _doh_result_cb(http_status, rcode, parse_ok, t, name, stamp); + }); + ++_id; + probe->set_configs(_interval_msec, _timeout_msec, _packets_per_test, _packets_interval_msec, _packet_payload_size); + probe->set_callbacks([this](pcpp::Packet &payload, TestType type, const std::string &name, timespec stamp) { _send_cb(payload, type, name, stamp); }, + [this](pcpp::Packet &payload, TestType type, const std::string &name, timespec stamp) { _recv_cb(payload, type, name, stamp); }, + [this](ErrorType error, TestType type, const std::string &name) { _fail_cb(error, type, name); }); + probe->start(_io_loop); + _probes.push_back(std::move(probe)); + } + // spawn the loop _io_thread = std::make_unique([this] { _timer->start(uvw::timer_handle::time{1000}, uvw::timer_handle::time{HEARTBEAT_INTERVAL * 1000}); @@ -296,7 +429,7 @@ void NetProbeInputStream::stop() void NetProbeInputStream::info_json(json &j) const { common_info_json(j); - j[schema_key()]["current_targets_total"] = _dns_list.size() + _ip_list.size(); + j[schema_key()]["current_targets_total"] = _dns_list.size() + _ip_list.size() + _http_targets.size() + _doh_targets.size(); // Report ping socket usage only for ping streams. Derive the probe count from _probes — a stable // per-stream member after start() — rather than a thread_local counter, which info_json (invoked // on an HTTP worker thread) could never read. The two addends are the shared receiver's v4 socket diff --git a/src/inputs/netprobe/NetProbeInputStream.h b/src/inputs/netprobe/NetProbeInputStream.h index c13f5ebf3..62c24d7c8 100644 --- a/src/inputs/netprobe/NetProbeInputStream.h +++ b/src/inputs/netprobe/NetProbeInputStream.h @@ -4,6 +4,7 @@ #pragma once +#include "HttpTypes.h" #include "InputStream.h" #include "NetProbe.h" #include @@ -16,6 +17,10 @@ class AsyncHandle; class TimerHandle; } +namespace visor::http { +class HttpClient; +} + namespace visor::input::netprobe { class NetProbeInputStream : public visor::InputStream @@ -32,6 +37,13 @@ class NetProbeInputStream : public visor::InputStream struct DnsEntry { std::string dns; uint32_t port{0}; std::optional force_ipv6; }; std::map _ip_list; std::map _dns_list; + std::map _http_targets; + std::string _http_method{"GET"}; + std::map _doh_targets; + std::string _doh_qname; + std::string _doh_qtype{"A"}; + std::string _doh_method{"POST"}; + std::shared_ptr _http_client; std::shared_ptr _logger; std::unique_ptr _io_thread; @@ -45,7 +57,8 @@ class NetProbeInputStream : public visor::InputStream {"ping", TestType::Ping}, {"http", TestType::HTTP}, {"udp", TestType::UDP}, - {"tcp", TestType::TCP}}; + {"tcp", TestType::TCP}, + {"doh", TestType::DOH}}; static const inline ConfigsDefType _config_defs = { "test_type", @@ -54,12 +67,17 @@ class NetProbeInputStream : public visor::InputStream "packets_per_test", "packets_interval_msec", "packet_payload_size", - "targets"}; + "targets", + "http_method", + "qname", + "qtype"}; void _create_netprobe_loop(); void _send_cb(pcpp::Packet &, TestType, const std::string &, timespec); void _recv_cb(pcpp::Packet &, TestType, const std::string &, timespec); void _fail_cb(ErrorType, TestType, const std::string &); + void _http_result_cb(uint16_t status, visor::http::HttpTimings t, const std::string &name, timespec stamp); + void _doh_result_cb(uint16_t http_status, uint8_t rcode, bool parse_ok, visor::http::HttpTimings t, const std::string &name, timespec stamp); public: NetProbeInputStream(const std::string &name); @@ -88,7 +106,7 @@ class NetProbeInputEventProxy : public visor::InputEventProxy size_t consumer_count() const override { - return policy_signal.slot_count() + heartbeat_signal.slot_count() + probe_recv_signal.slot_count() + probe_send_signal.slot_count() + probe_fail_signal.slot_count(); + return policy_signal.slot_count() + heartbeat_signal.slot_count() + probe_recv_signal.slot_count() + probe_send_signal.slot_count() + probe_fail_signal.slot_count() + probe_http_result_signal.slot_count() + probe_doh_result_signal.slot_count(); } void probe_send_cb(pcpp::Packet &p, TestType t, const std::string &n, timespec s) @@ -106,12 +124,24 @@ class NetProbeInputEventProxy : public visor::InputEventProxy probe_fail_signal(e, t, n); } + void probe_http_result_cb(uint16_t status, visor::http::HttpTimings t, const std::string &n, timespec s) + { + probe_http_result_signal(status, t, n, s); + } + + void probe_doh_result_cb(uint16_t http_status, uint8_t rcode, bool parse_ok, visor::http::HttpTimings t, const std::string &n, timespec s) + { + probe_doh_result_signal(http_status, rcode, parse_ok, t, n, s); + } + // handler functionality // IF THIS changes, see consumer_count() // note: these are mutable because consumer_count() calls slot_count() which is not const (unclear if it could/should be) mutable sigslot::signal probe_send_signal; mutable sigslot::signal probe_recv_signal; mutable sigslot::signal probe_fail_signal; + mutable sigslot::signal probe_http_result_signal; + mutable sigslot::signal probe_doh_result_signal; }; } diff --git a/src/inputs/netprobe/test_netprobe.cpp b/src/inputs/netprobe/test_netprobe.cpp index 88b41c379..8f6625a99 100644 --- a/src/inputs/netprobe/test_netprobe.cpp +++ b/src/inputs/netprobe/test_netprobe.cpp @@ -1,11 +1,28 @@ #include "NetProbeInputStream.h" +#include "NetProbeStreamHandler.h" #include "PingProbe.h" #include #include #include +#include +#include +#include +#ifdef __GNUC__ +#pragma GCC diagnostic push +#pragma GCC diagnostic ignored "-Wold-style-cast" +#pragma GCC diagnostic ignored "-Wunused-parameter" +#pragma GCC diagnostic ignored "-Wzero-as-null-pointer-constant" +#endif +#include +#include +#ifdef __GNUC__ +#pragma GCC diagnostic pop +#endif using namespace visor::input::netprobe; +using namespace visor::handler::netprobe; +using namespace nlohmann; using namespace std::chrono; TEST_CASE("NetProbe Configs", "[netprobe][ping]") @@ -107,7 +124,7 @@ TEST_CASE("Netprobe invalid config", "[netprobe][config]") NetProbeInputStream stream{"net-probe-test"}; stream.config_set("invalid_config", true); - CHECK_THROWS_WITH(stream.start(), "invalid_config is an invalid/unsupported config or filter. The valid configs/filters are: test_type, interval_msec, timeout_msec, packets_per_test, packets_interval_msec, packet_payload_size, targets"); + CHECK_THROWS_WITH(stream.start(), "invalid_config is an invalid/unsupported config or filter. The valid configs/filters are: test_type, interval_msec, timeout_msec, packets_per_test, packets_interval_msec, packet_payload_size, targets, http_method, qname, qtype"); } TEST_CASE("NetProbe ip_version config", "[netprobe][config][ipv6]") @@ -142,10 +159,111 @@ TEST_CASE("NetProbe ip_version config", "[netprobe][config][ipv6]") SECTION("top-level valid-keys string unchanged") { NetProbeInputStream s{"net-probe-test"}; s.config_set("invalid_config", true); - CHECK_THROWS_WITH(s.start(), "invalid_config is an invalid/unsupported config or filter. The valid configs/filters are: test_type, interval_msec, timeout_msec, packets_per_test, packets_interval_msec, packet_payload_size, targets"); + CHECK_THROWS_WITH(s.start(), "invalid_config is an invalid/unsupported config or filter. The valid configs/filters are: test_type, interval_msec, timeout_msec, packets_per_test, packets_interval_msec, packet_payload_size, targets, http_method, qname, qtype"); } } +TEST_CASE("NetProbe http_method config validates", "[netprobe][config][http]") +{ + // Validates that the http_method key is accepted by validate_configs (no throw before + // targets-missing is hit). This exercises the _config_defs registration path without + // requiring network access or curl. + NetProbeInputStream stream{"net-probe-test"}; + stream.config_set("test_type", "http"); + stream.config_set("http_method", std::string("GET")); + // No targets → throws "no targets specified", NOT a config-validation error. + // That means http_method was accepted as a valid key. + CHECK_THROWS_WITH(stream.start(), "no targets specified"); +} + +TEST_CASE("NetProbe DoH config: qname required", "[netprobe][config][doh]") +{ + // test_type=doh without qname must throw the 'qname' is required error. + NetProbeInputStream stream{"net-probe-test-doh-noqname"}; + stream.config_set("test_type", "doh"); + auto targets = std::make_shared(); + auto target = std::make_shared(); + target->config_set("target", "https://1.1.1.1/dns-query"); + targets->config_set>("cf_doh", target); + stream.config_set>("targets", targets); + // No qname → must throw. + CHECK_THROWS_WITH(stream.start(), "netprobe: 'qname' is required when test_type is 'doh'"); +} + +TEST_CASE("NetProbe DoH config: unsupported http_method rejected", "[netprobe][config][doh]") +{ + // test_type=doh only supports GET/POST; any other method must throw a clear error. + NetProbeInputStream stream{"net-probe-test-doh-badmethod"}; + stream.config_set("test_type", "doh"); + stream.config_set("qname", std::string("example.com")); + stream.config_set("http_method", std::string("PUT")); + auto targets = std::make_shared(); + auto target = std::make_shared(); + target->config_set("target", "https://1.1.1.1/dns-query"); + targets->config_set>("cf_doh", target); + stream.config_set>("targets", targets); + CHECK_THROWS_WITH(stream.start(), "unsupported http_method 'PUT' for doh (use GET or POST)"); +} + +TEST_CASE("NetProbe DoH config: qname accepted", "[netprobe][config][doh]") +{ + // test_type=doh with qname set must pass config validation + // (throws "no targets specified" only when targets aren't set — here we omit targets + // to confirm that qname/qtype keys are accepted before the targets check fires). + NetProbeInputStream stream{"net-probe-test-doh-valid"}; + stream.config_set("test_type", "doh"); + stream.config_set("qname", std::string("example.com")); + stream.config_set("qtype", std::string("A")); + // No targets → throws "no targets specified", meaning qname/qtype were accepted as valid keys. + CHECK_THROWS_WITH(stream.start(), "no targets specified"); +} + +TEST_CASE("NetProbe DoH config: qtype is normalized to uppercase", "[netprobe][config][doh]") +{ + // A lowercase qtype must be accepted (normalized to uppercase before lookup), matching the + // rest of the codebase. Verify via the rejection message: an invalid lowercase "zzz" is + // uppercased to "ZZZ" before the not-found error fires (which happens before any stream start). + NetProbeInputStream stream{"net-probe-test-doh-qtype-case"}; + stream.config_set("test_type", "doh"); + stream.config_set("qname", std::string("example.com")); + stream.config_set("qtype", std::string("zzz")); + auto targets = std::make_shared(); + auto target = std::make_shared(); + target->config_set("target", "https://1.1.1.1/dns-query"); + targets->config_set>("cf_doh", target); + stream.config_set>("targets", targets); + CHECK_THROWS_WITH(stream.start(), "netprobe: unknown qtype 'ZZZ'"); +} + +TEST_CASE("NetProbe DoH config: qname exceeding DNS label limit rejected", "[netprobe][config][doh]") +{ + // A label longer than 63 chars is outside DNS limits and must be rejected at config time + // (before it reaches DnsLayer::addQuery()). + NetProbeInputStream stream{"net-probe-test-doh-badqname"}; + stream.config_set("test_type", "doh"); + std::string long_label(64, 'a'); // 64 > 63 + stream.config_set("qname", long_label + ".example.com"); + auto targets = std::make_shared(); + auto target = std::make_shared(); + target->config_set("target", "https://1.1.1.1/dns-query"); + targets->config_set>("cf_doh", target); + stream.config_set>("targets", targets); + CHECK_THROWS_WITH(stream.start(), "netprobe: qname '" + long_label + ".example.com' has a DNS label longer than 63 characters"); +} + +TEST_CASE("NetProbe http/doh config: invalid target URL rejected", "[netprobe][config]") +{ + // A target whose URL has a non-http(s) scheme must be rejected at config time with a clear error. + NetProbeInputStream stream{"net-probe-test-badurl"}; + stream.config_set("test_type", "http"); + auto targets = std::make_shared(); + auto target = std::make_shared(); + target->config_set("target", std::string("ftp://example.com/x")); + targets->config_set>("bad", target); + stream.config_set>("targets", targets); + CHECK_THROWS_WITH(stream.start(), "target 'bad' is not a valid http(s) URL: 'ftp://example.com/x'"); +} + TEST_CASE("ICMPv6 reply carrier survives the fan-out Packet deep-copy", "[netprobe][ipv6]") { // Wire bytes of an ICMPv6 echo REPLY: type=129, code=0, checksum=0, id=0xBEEF, seq=0x0102 (network order). @@ -171,3 +289,544 @@ TEST_CASE("ICMPv6 reply carrier survives the fan-out Packet deep-copy", "[netpro const uint8_t too_short[4] = {129, 0, 0, 0}; CHECK_FALSE(build_icmpv6_reply_carrier(too_short, sizeof(too_short)).has_value()); } + +// --------------------------------------------------------------------------- +// End-to-end HTTP probe tests: real NetProbeInputStream + NetProbeStreamHandler +// against an in-process httplib server bound to an ephemeral port. +// --------------------------------------------------------------------------- + +// RAII guard: ensures httplib server is stopped and thread joined even if a +// Catch2 REQUIRE macro throws an exception that unwinds the test frame. +struct ServerGuard { + httplib::Server &svr; + std::thread &t; + ~ServerGuard() + { + svr.stop(); + if (t.joinable()) t.join(); + } +}; + +TEST_CASE("NetProbe HTTP e2e: success path records attempt, success, and 200 in top_status_codes", "[netprobe][http][e2e]") +{ + // Start an in-process httplib server on an ephemeral port. + httplib::Server svr; + svr.Get("/ok", [](const httplib::Request &, httplib::Response &res) { + res.set_content("ok", "text/plain"); + }); + int port = svr.bind_to_any_port("127.0.0.1"); + REQUIRE(port > 0); + std::thread server_thread([&svr] { svr.listen_after_bind(); }); + ServerGuard guard{svr, server_thread}; + svr.wait_until_ready(); + + std::string url = "http://127.0.0.1:" + std::to_string(port) + "/ok"; + + // Configure stream: interval=500ms, timeout=400ms (timeout must not exceed interval). + // A 500ms interval gives ≥1 tick in the 1s sleep window. + NetProbeInputStream stream{"netprobe-http-e2e"}; + stream.config_set("test_type", "http"); + stream.config_set("interval_msec", 500); + stream.config_set("timeout_msec", 400); + auto targets = std::make_shared(); + auto target = std::make_shared(); + target->config_set("target", url); + targets->config_set>("ok_target", target); + stream.config_set>("targets", targets); + + // Wire up the handler (mirrors the ping/tcp test pattern). + visor::Config c; + c.config_set("num_periods", 1); + auto *proxy = stream.add_event_proxy(c); + NetProbeStreamHandler handler{"netprobe-http-e2e", proxy, &c}; + + handler.start(); + stream.start(); + // Sleep generously: 1.5s covers 2+ interval ticks + request round-trip. + std::this_thread::sleep_for(1500ms); + stream.stop(); + handler.stop(); + + json j; + handler.metrics()->bucket(0)->to_json(j); + + REQUIRE(j.contains("targets")); + REQUIRE(j["targets"].contains("ok_target")); + auto &tgt = j["targets"]["ok_target"]; + + CHECK(tgt["attempts"].get() >= 1); + CHECK(tgt["successes"].get() >= 1); + + // top_status_codes must have a "200" entry. + REQUIRE(tgt.contains("top_status_codes")); + bool found_200 = false; + for (const auto &entry : tgt["top_status_codes"]) { + if (entry.contains("name") && entry["name"] == "200") { + found_200 = true; + } + } + CHECK(found_200); +} + +TEST_CASE("NetProbe HTTP e2e: stop while request in flight does not crash or hang", "[netprobe][http][e2e]") +{ + // The slow handler sleeps 500ms — longer than our start/stop window. + // This test verifies that stop() returns cleanly even with a curl request in flight, + // exercising the loop-quiescent teardown of curl poll handles (_http_client->close()). + // interval=300ms, timeout=250ms (timeout must not exceed interval); the /slow handler sleeps 500ms + // so the request will still be in flight when we call stop() after 200ms. + httplib::Server svr; + svr.Get("/slow", [](const httplib::Request &, httplib::Response &res) { + std::this_thread::sleep_for(500ms); + res.set_content("late", "text/plain"); + }); + int port = svr.bind_to_any_port("127.0.0.1"); + REQUIRE(port > 0); + std::thread server_thread([&svr] { svr.listen_after_bind(); }); + ServerGuard guard{svr, server_thread}; + svr.wait_until_ready(); + + std::string url = "http://127.0.0.1:" + std::to_string(port) + "/slow"; + + NetProbeInputStream stream{"netprobe-http-e2e-inflight"}; + stream.config_set("test_type", "http"); + // interval=300ms, timeout=250ms; timeout must not exceed interval. + stream.config_set("interval_msec", 300); + stream.config_set("timeout_msec", 250); + auto targets = std::make_shared(); + auto target = std::make_shared(); + target->config_set("target", url); + targets->config_set>("slow_target", target); + stream.config_set>("targets", targets); + + visor::Config c; + c.config_set("num_periods", 1); + auto *proxy = stream.add_event_proxy(c); + NetProbeStreamHandler handler{"netprobe-http-e2e-inflight", proxy, &c}; + + handler.start(); + stream.start(); + // Sleep just long enough for the first request to be in flight (interval fired, curl running). + std::this_thread::sleep_for(350ms); + // stop() must return without hanging or crashing even though the curl request is still in flight. + CHECK_NOTHROW(stream.stop()); + CHECK_NOTHROW(handler.stop()); +} + +// --------------------------------------------------------------------------- +// End-to-end DoH probe tests: real NetProbeInputStream (test_type=doh) + +// NetProbeStreamHandler against an in-process httplib server that returns a +// canned NOERROR DNS response as application/dns-message. +// --------------------------------------------------------------------------- + +// Build a minimal but valid NOERROR DNS response wire-format using pcpp::DnsLayer. +// The probe only requires: size >= 12, QR==1, rcode==0. +// The EMPTY DnsLayer constructor manages its own buffer (no heap-ownership hazard here +// unlike the parse path); addAnswer with IPv4DnsResourceData is straightforward. +static std::string make_doh_response() +{ + pcpp::DnsLayer resp; + resp.getDnsHeader()->queryOrResponse = 1; + resp.getDnsHeader()->responseCode = 0; // NOERROR + resp.addQuery("example.com", pcpp::DNS_TYPE_A, pcpp::DNS_CLASS_IN); + pcpp::IPv4DnsResourceData answerData(std::string("1.2.3.4")); + resp.addAnswer("example.com", pcpp::DNS_TYPE_A, pcpp::DNS_CLASS_IN, 60, &answerData); + return std::string(reinterpret_cast(resp.getData()), resp.getDataLen()); +} + +TEST_CASE("NetProbe DoH e2e POST: success path records attempt, success, and NOERROR in top_rcodes", "[netprobe][doh][e2e]") +{ + httplib::Server svr; + const std::string doh_body = make_doh_response(); + svr.Post("/dns-query", [&](const httplib::Request &, httplib::Response &res) { + res.set_content(doh_body, "application/dns-message"); + }); + svr.Get("/dns-query", [&](const httplib::Request &, httplib::Response &res) { + res.set_content(doh_body, "application/dns-message"); + }); + int port = svr.bind_to_any_port("127.0.0.1"); + REQUIRE(port > 0); + std::thread server_thread([&svr] { svr.listen_after_bind(); }); + ServerGuard guard{svr, server_thread}; + svr.wait_until_ready(); + + std::string url = "http://127.0.0.1:" + std::to_string(port) + "/dns-query"; + + NetProbeInputStream stream{"netprobe-doh-e2e-post"}; + stream.config_set("test_type", "doh"); + stream.config_set("qname", std::string("example.com")); + stream.config_set("qtype", std::string("A")); + stream.config_set("interval_msec", 200); + stream.config_set("timeout_msec", 150); + auto targets = std::make_shared(); + auto target = std::make_shared(); + target->config_set("target", url); + targets->config_set>("doh_target", target); + stream.config_set>("targets", targets); + + visor::Config c; + c.config_set("num_periods", 1); + auto *proxy = stream.add_event_proxy(c); + NetProbeStreamHandler handler{"netprobe-doh-e2e-post", proxy, &c}; + + handler.start(); + stream.start(); + std::this_thread::sleep_for(750ms); + stream.stop(); + handler.stop(); + + json j; + handler.metrics()->bucket(0)->to_json(j); + + REQUIRE(j.contains("targets")); + REQUIRE(j["targets"].contains("doh_target")); + auto &tgt = j["targets"]["doh_target"]; + + CHECK(tgt["attempts"].get() >= 1); + CHECK(tgt["successes"].get() >= 1); + + REQUIRE(tgt.contains("top_rcodes")); + bool found_noerror = false; + for (const auto &entry : tgt["top_rcodes"]) { + if (entry.contains("name") && entry["name"] == "NOERROR") { + found_noerror = true; + } + } + CHECK(found_noerror); + + // DoH records the HTTP status breakdown (top_status_codes) too, like the HTTP probe. + REQUIRE(tgt.contains("top_status_codes")); + bool found_200 = false; + for (const auto &entry : tgt["top_status_codes"]) { + if (entry.contains("name") && entry["name"] == "200") { + found_200 = true; + } + } + CHECK(found_200); +} + +TEST_CASE("NetProbe DoH e2e GET: success path records attempt, success, and NOERROR in top_rcodes", "[netprobe][doh][e2e]") +{ + httplib::Server svr; + const std::string doh_body = make_doh_response(); + svr.Post("/dns-query", [&](const httplib::Request &, httplib::Response &res) { + res.set_content(doh_body, "application/dns-message"); + }); + svr.Get("/dns-query", [&](const httplib::Request &, httplib::Response &res) { + res.set_content(doh_body, "application/dns-message"); + }); + int port = svr.bind_to_any_port("127.0.0.1"); + REQUIRE(port > 0); + std::thread server_thread([&svr] { svr.listen_after_bind(); }); + ServerGuard guard{svr, server_thread}; + svr.wait_until_ready(); + + std::string url = "http://127.0.0.1:" + std::to_string(port) + "/dns-query"; + + NetProbeInputStream stream{"netprobe-doh-e2e-get"}; + stream.config_set("test_type", "doh"); + stream.config_set("qname", std::string("example.com")); + stream.config_set("qtype", std::string("A")); + stream.config_set("http_method", std::string("GET")); + stream.config_set("interval_msec", 200); + stream.config_set("timeout_msec", 150); + auto targets = std::make_shared(); + auto target = std::make_shared(); + target->config_set("target", url); + targets->config_set>("doh_get_target", target); + stream.config_set>("targets", targets); + + visor::Config c; + c.config_set("num_periods", 1); + auto *proxy = stream.add_event_proxy(c); + NetProbeStreamHandler handler{"netprobe-doh-e2e-get", proxy, &c}; + + handler.start(); + stream.start(); + std::this_thread::sleep_for(750ms); + stream.stop(); + handler.stop(); + + json j; + handler.metrics()->bucket(0)->to_json(j); + + REQUIRE(j.contains("targets")); + REQUIRE(j["targets"].contains("doh_get_target")); + auto &tgt = j["targets"]["doh_get_target"]; + + CHECK(tgt["attempts"].get() >= 1); + CHECK(tgt["successes"].get() >= 1); + + REQUIRE(tgt.contains("top_rcodes")); + bool found_noerror = false; + for (const auto &entry : tgt["top_rcodes"]) { + if (entry.contains("name") && entry["name"] == "NOERROR") { + found_noerror = true; + } + } + CHECK(found_noerror); + + // DoH records the HTTP status breakdown (top_status_codes) too, like the HTTP probe. + REQUIRE(tgt.contains("top_status_codes")); + bool found_200 = false; + for (const auto &entry : tgt["top_status_codes"]) { + if (entry.contains("name") && entry["name"] == "200") { + found_200 = true; + } + } + CHECK(found_200); +} + +TEST_CASE("NetProbe DoH e2e: stop while request in flight does not crash or hang", "[netprobe][doh][e2e]") +{ + // The slow handler sleeps 500ms — longer than our start/stop window. + // Mirrors the HTTP in-flight-stop test to exercise DohProbe teardown. + httplib::Server svr; + svr.Post("/dns-query", [](const httplib::Request &, httplib::Response &res) { + std::this_thread::sleep_for(500ms); + const std::string body = make_doh_response(); + res.set_content(body, "application/dns-message"); + }); + svr.Get("/dns-query", [](const httplib::Request &, httplib::Response &res) { + std::this_thread::sleep_for(500ms); + const std::string body = make_doh_response(); + res.set_content(body, "application/dns-message"); + }); + int port = svr.bind_to_any_port("127.0.0.1"); + REQUIRE(port > 0); + std::thread server_thread([&svr] { svr.listen_after_bind(); }); + ServerGuard guard{svr, server_thread}; + svr.wait_until_ready(); + + std::string url = "http://127.0.0.1:" + std::to_string(port) + "/dns-query"; + + NetProbeInputStream stream{"netprobe-doh-e2e-inflight"}; + stream.config_set("test_type", "doh"); + stream.config_set("qname", std::string("example.com")); + stream.config_set("interval_msec", 300); + stream.config_set("timeout_msec", 250); + auto targets = std::make_shared(); + auto target = std::make_shared(); + target->config_set("target", url); + targets->config_set>("doh_slow_target", target); + stream.config_set>("targets", targets); + + visor::Config c; + c.config_set("num_periods", 1); + auto *proxy = stream.add_event_proxy(c); + NetProbeStreamHandler handler{"netprobe-doh-e2e-inflight", proxy, &c}; + + handler.start(); + stream.start(); + std::this_thread::sleep_for(350ms); + CHECK_NOTHROW(stream.stop()); + CHECK_NOTHROW(handler.stop()); +} + +TEST_CASE("NetProbe DoH e2e: wrong response Content-Type is a DNS failure (not a success)", "[netprobe][doh][e2e]") +{ + // A valid DNS body returned with the WRONG Content-Type (not application/dns-message) must be + // rejected per RFC 8484 — counted as dns_response_failures, never as a success. + httplib::Server svr; + const std::string doh_body = make_doh_response(); + svr.Post("/dns-query", [&](const httplib::Request &, httplib::Response &res) { + res.set_content(doh_body, "text/html"); + }); + svr.Get("/dns-query", [&](const httplib::Request &, httplib::Response &res) { + res.set_content(doh_body, "text/html"); + }); + int port = svr.bind_to_any_port("127.0.0.1"); + REQUIRE(port > 0); + std::thread server_thread([&svr] { svr.listen_after_bind(); }); + ServerGuard guard{svr, server_thread}; + svr.wait_until_ready(); + + std::string url = "http://127.0.0.1:" + std::to_string(port) + "/dns-query"; + NetProbeInputStream stream{"netprobe-doh-e2e-badct"}; + stream.config_set("test_type", "doh"); + stream.config_set("qname", std::string("example.com")); + stream.config_set("interval_msec", 200); + stream.config_set("timeout_msec", 150); + auto targets = std::make_shared(); + auto target = std::make_shared(); + target->config_set("target", url); + targets->config_set>("doh_target", target); + stream.config_set>("targets", targets); + + visor::Config c; + c.config_set("num_periods", 1); + auto *proxy = stream.add_event_proxy(c); + NetProbeStreamHandler handler{"netprobe-doh-e2e-badct", proxy, &c}; + + handler.start(); + stream.start(); + std::this_thread::sleep_for(750ms); + stream.stop(); + handler.stop(); + + json j; + handler.metrics()->bucket(0)->to_json(j); + REQUIRE(j["targets"].contains("doh_target")); + auto &tgt = j["targets"]["doh_target"]; + CHECK(tgt["attempts"].get() >= 1); + CHECK(tgt["successes"].get() == 0); + CHECK(tgt["dns_response_failures"].get() >= 1); +} + +TEST_CASE("NetProbe DoH e2e: malformed DNS response body is a DNS failure (not a success)", "[netprobe][doh][e2e]") +{ + // Correct Content-Type but a body that is not a valid DNS message (too short to be a header) + // must be rejected — counted as dns_response_failures, never as a success. + httplib::Server svr; + svr.Post("/dns-query", [](const httplib::Request &, httplib::Response &res) { + res.set_content(std::string("garbage", 7), "application/dns-message"); + }); + svr.Get("/dns-query", [](const httplib::Request &, httplib::Response &res) { + res.set_content(std::string("garbage", 7), "application/dns-message"); + }); + int port = svr.bind_to_any_port("127.0.0.1"); + REQUIRE(port > 0); + std::thread server_thread([&svr] { svr.listen_after_bind(); }); + ServerGuard guard{svr, server_thread}; + svr.wait_until_ready(); + + std::string url = "http://127.0.0.1:" + std::to_string(port) + "/dns-query"; + NetProbeInputStream stream{"netprobe-doh-e2e-malformed"}; + stream.config_set("test_type", "doh"); + stream.config_set("qname", std::string("example.com")); + stream.config_set("interval_msec", 200); + stream.config_set("timeout_msec", 150); + auto targets = std::make_shared(); + auto target = std::make_shared(); + target->config_set("target", url); + targets->config_set>("doh_target", target); + stream.config_set>("targets", targets); + + visor::Config c; + c.config_set("num_periods", 1); + auto *proxy = stream.add_event_proxy(c); + NetProbeStreamHandler handler{"netprobe-doh-e2e-malformed", proxy, &c}; + + handler.start(); + stream.start(); + std::this_thread::sleep_for(750ms); + stream.stop(); + handler.stop(); + + json j; + handler.metrics()->bucket(0)->to_json(j); + REQUIRE(j["targets"].contains("doh_target")); + auto &tgt = j["targets"]["doh_target"]; + CHECK(tgt["attempts"].get() >= 1); + CHECK(tgt["successes"].get() == 0); + CHECK(tgt["dns_response_failures"].get() >= 1); +} + +TEST_CASE("NetProbe DoH e2e: response declaring more answers than present is a DNS failure", "[netprobe][doh][e2e]") +{ + // NOERROR with a valid echoed question, but the header declares more answers than the body + // contains (a truncated/incomplete answer section). pcpp parses fewer records than declared, + // so this must be rejected, never counted as a success. + httplib::Server svr; + std::string body = make_doh_response(); // valid: 1 answer, ANCOUNT=1 + // Corrupt ANCOUNT (DNS header bytes 6-7, network byte order) to claim 2 answers while only 1 is present. + REQUIRE(body.size() > 7); + body[6] = 0x00; + body[7] = 0x02; + svr.Post("/dns-query", [&](const httplib::Request &, httplib::Response &res) { + res.set_content(body, "application/dns-message"); + }); + svr.Get("/dns-query", [&](const httplib::Request &, httplib::Response &res) { + res.set_content(body, "application/dns-message"); + }); + int port = svr.bind_to_any_port("127.0.0.1"); + REQUIRE(port > 0); + std::thread server_thread([&svr] { svr.listen_after_bind(); }); + ServerGuard guard{svr, server_thread}; + svr.wait_until_ready(); + + std::string url = "http://127.0.0.1:" + std::to_string(port) + "/dns-query"; + NetProbeInputStream stream{"netprobe-doh-e2e-truncated"}; + stream.config_set("test_type", "doh"); + stream.config_set("qname", std::string("example.com")); + stream.config_set("interval_msec", 200); + stream.config_set("timeout_msec", 150); + auto targets = std::make_shared(); + auto target = std::make_shared(); + target->config_set("target", url); + targets->config_set>("doh_target", target); + stream.config_set>("targets", targets); + + visor::Config c; + c.config_set("num_periods", 1); + auto *proxy = stream.add_event_proxy(c); + NetProbeStreamHandler handler{"netprobe-doh-e2e-truncated", proxy, &c}; + + handler.start(); + stream.start(); + std::this_thread::sleep_for(750ms); + stream.stop(); + handler.stop(); + + json j; + handler.metrics()->bucket(0)->to_json(j); + REQUIRE(j["targets"].contains("doh_target")); + auto &tgt = j["targets"]["doh_target"]; + CHECK(tgt["attempts"].get() >= 1); + CHECK(tgt["successes"].get() == 0); + CHECK(tgt["dns_response_failures"].get() >= 1); +} + +TEST_CASE("NetProbe DoH e2e: root qname (dot) probe succeeds", "[netprobe][doh][e2e]") +{ + // End-to-end root probe: qname "." must build a valid root query and match the echoed root + // question in the response (both encode/decode as the empty name in pcpp). + httplib::Server svr; + // Canned NOERROR response echoing a ROOT question of type A (the probe's default qtype). + pcpp::DnsLayer resp; + resp.getDnsHeader()->queryOrResponse = 1; + resp.getDnsHeader()->responseCode = 0; // NOERROR + REQUIRE(resp.addQuery("", pcpp::DNS_TYPE_A, pcpp::DNS_CLASS_IN) != nullptr); + const std::string body(reinterpret_cast(resp.getData()), resp.getDataLen()); + svr.Post("/dns-query", [&](const httplib::Request &, httplib::Response &res) { + res.set_content(body, "application/dns-message"); + }); + svr.Get("/dns-query", [&](const httplib::Request &, httplib::Response &res) { + res.set_content(body, "application/dns-message"); + }); + int port = svr.bind_to_any_port("127.0.0.1"); + REQUIRE(port > 0); + std::thread server_thread([&svr] { svr.listen_after_bind(); }); + ServerGuard guard{svr, server_thread}; + svr.wait_until_ready(); + + std::string url = "http://127.0.0.1:" + std::to_string(port) + "/dns-query"; + NetProbeInputStream stream{"netprobe-doh-e2e-root"}; + stream.config_set("test_type", "doh"); + stream.config_set("qname", std::string(".")); + stream.config_set("interval_msec", 200); + stream.config_set("timeout_msec", 150); + auto targets = std::make_shared(); + auto target = std::make_shared(); + target->config_set("target", url); + targets->config_set>("doh_target", target); + stream.config_set>("targets", targets); + + visor::Config c; + c.config_set("num_periods", 1); + auto *proxy = stream.add_event_proxy(c); + NetProbeStreamHandler handler{"netprobe-doh-e2e-root", proxy, &c}; + + handler.start(); + stream.start(); + std::this_thread::sleep_for(750ms); + stream.stop(); + handler.stop(); + + json j; + handler.metrics()->bucket(0)->to_json(j); + REQUIRE(j["targets"].contains("doh_target")); + auto &tgt = j["targets"]["doh_target"]; + CHECK(tgt["attempts"].get() >= 1); + CHECK(tgt["successes"].get() >= 1); +}