Feature/filter dnstap by hostmask

src/handlers/dns/tests/test_dnstap.cpp

@@ -11,7 +11,7 @@ TEST_CASE("Parse DNSTAP", "[dnstap][dns]")
 {
     DnstapInputStream stream{"dnstap-test"};
     stream.config_set("dnstap_file", "inputs/dnstap/tests/fixtures/fixture.dnstap");
-
+    stream.config_set<visor::Configurable::StringList>("only_hosts", {"192.168.0.0/24", "2001:db8::/48"});
     visor::Config c;
     c.config_set<uint64_t>("num_periods", 1);
     DnsStreamHandler dns_handler{"dns-test", &stream, &c};

src/inputs/dnstap/DnstapException.h

@@ -3,6 +3,8 @@
  * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
 
 #pragma once
+#include <stdexcept>
+#include <string>
 
 namespace visor::input::dnstap {
 
@@ -13,6 +15,11 @@ class DnstapException : public std::runtime_error
         : std::runtime_error(msg)
     {
     }
+
+    DnstapException(std::string msg)
+        : std::runtime_error(msg)
+    {
+    }
 };
 
 }

src/inputs/dnstap/DnstapInputStream.cpp

@@ -60,8 +60,11 @@ void DnstapInputStream::_read_frame_stream_file()
                 _logger->warn("dnstap data is wrong type or has no message, skipping frame of size {}", len_data);
                 continue;
             }
+
             // Emit signal to handlers
-            dnstap_signal(d);
+            if (!_filtering(d)) {
+                dnstap_signal(d);
+            }
         } else if (result == fstrm_res_stop) {
             // Normal end of data stream
             break;
@@ -82,6 +85,11 @@ void DnstapInputStream::start()
         return;
     }
 
+    if (config_exists("only_hosts")) {
+        _parse_host_specs(config_get<StringList>("only_hosts"));
+        _f_enabled.set(Filters::OnlyHosts);
+    }
+
     if (config_exists("dnstap_file")) {
         // read from dnstap file. this is a special case from a command line utility
         _running = true;
@@ -168,8 +176,11 @@ void DnstapInputStream::_create_frame_stream_tcp_socket()
                 _logger->warn("dnstap data is wrong type or has no message, skipping frame of size {}", len_data);
                 return;
             }
+
             // Emit signal to handlers
-            dnstap_signal(d);
+            if (!_filtering(d)) {
+                dnstap_signal(d);
+            }
         };
 
         client->on<uvw::ErrorEvent>([this](const uvw::ErrorEvent &err, uvw::TCPHandle &c_sock) {
@@ -272,7 +283,9 @@ void DnstapInputStream::_create_frame_stream_unix_socket()
                 return;
             }
             // Emit signal to handlers
-            dnstap_signal(d);
+            if (!_filtering(d)) {
+                dnstap_signal(d);
+            }
         };
 
         client->on<uvw::ErrorEvent>([this](const uvw::ErrorEvent &err, uvw::PipeHandle &c_sock) {
@@ -323,6 +336,85 @@ void DnstapInputStream::_create_frame_stream_unix_socket()
     });
 }
 
+void DnstapInputStream::_parse_host_specs(const std::vector<std::string> &host_list)
+{
+    for (const auto &host : host_list) {
+        auto delimiter = host.find('/');
+        if (delimiter == host.npos) {
+            throw DnstapException(fmt::format("invalid CIDR: {}", host));
+        }
+        auto ip = host.substr(0, delimiter);
+        auto cidr = host.substr(++delimiter);
+        auto not_number = std::count_if(cidr.begin(), cidr.end(),
+            [](unsigned char c) { return !std::isdigit(c); });
+        if (not_number) {
+            throw DnstapException(fmt::format("invalid CIDR: {}", host));
+        }
+
+        auto cidr_number = std::stoi(cidr);
+        if (ip.find(':') != ip.npos) {
+            if (cidr_number < 0 || cidr_number > 128) {
+                throw DnstapException(fmt::format("invalid CIDR: {}", host));
+            }
+            in6_addr ipv6;
+            if (inet_pton(AF_INET6, ip.c_str(), &ipv6) != 1) {
+                throw DnstapException(fmt::format("invalid IPv6 address: {}", ip));
+            }
+            _IPv6_host_list.emplace_back(ipv6, cidr_number);
+        } else {
+            if (cidr_number < 0 || cidr_number > 32) {
+                throw DnstapException(fmt::format("invalid CIDR: {}", host));
+            }
+            in_addr ipv4;
+            if (inet_pton(AF_INET, ip.c_str(), &ipv4) != 1) {
+                throw DnstapException(fmt::format("invalid IPv4 address: {}", ip));
+            }
+            _IPv4_host_list.emplace_back(ipv4, cidr_number);
+        }
+    }
+}
+
+bool DnstapInputStream::_match_subnet(const std::string &dnstap_ip)
+{
+    if (dnstap_ip.size() == 16 && _IPv6_host_list.size() > 0) {
+        in6_addr ipv6;
+        std::memcpy(&ipv6, dnstap_ip.c_str(), sizeof(in6_addr));
+        for (const auto &net : _IPv6_host_list) {
+            uint8_t prefixLength = net.second;
+            auto network = net.first;
+            uint8_t compareByteCount = prefixLength / 8;
+            uint8_t compareBitCount = prefixLength % 8;
+            bool result = false;
+            if (compareByteCount > 0) {
+                result = std::memcmp(&network.s6_addr, &ipv6.s6_addr, compareByteCount) == 0;
+            }
+            if ((result || prefixLength < 8) && compareBitCount > 0) {
+                uint8_t subSubnetByte = network.s6_addr[compareByteCount] >> (8 - compareBitCount);
+                uint8_t subThisByte = ipv6.s6_addr[compareByteCount] >> (8 - compareBitCount);
+                result = subSubnetByte == subThisByte;
+            }
+            if (result) {
+                return true;
+            }
+        }
+    } else if (dnstap_ip.size() == 4 && _IPv4_host_list.size() > 0) {
+        in_addr ipv4;
+        std::memcpy(&ipv4, dnstap_ip.c_str(), sizeof(in_addr));
+        for (const auto &net : _IPv4_host_list) {
+            uint8_t cidr = net.second;
+            if (cidr == 0) {
+                return true;
+            }
+            uint32_t mask = htonl((0xFFFFFFFFu) << (32 - cidr));
+            if (!((ipv4.s_addr ^ net.first.s_addr) & mask)) {
+                return true;
+            }
+        }
+    }
+
+    return false;
+}
+
 void DnstapInputStream::stop()
 {
     if (!_running) {
@@ -343,5 +435,4 @@ void DnstapInputStream::info_json(json &j) const
 {
     common_info_json(j);
 }
-
 }

src/inputs/dnstap/DnstapInputStream.h

@@ -4,8 +4,8 @@
 
 #pragma once
 
-#include "InputStream.h"
 #include "FrameSession.h"
+#include "InputStream.h"
 #include "dnstap.pb.h"
 #include <DnsLayer.h>
 #include <sigslot/signal.hpp>
@@ -24,6 +24,9 @@ struct fstrm_reader;
 
 namespace visor::input::dnstap {
 
+typedef std::pair<in_addr, uint8_t> Ipv4Subnet;
+typedef std::pair<in6_addr, uint8_t> Ipv6Subnet;
+
 const static std::string CONTENT_TYPE = "protobuf:dnstap.Dnstap";
 
 class DnstapInputStream : public visor::InputStream
@@ -40,10 +43,45 @@ class DnstapInputStream : public visor::InputStream
     std::shared_ptr<uvw::TCPHandle> _tcp_server_h;
     std::unordered_map<uv_os_fd_t, std::unique_ptr<FrameSessionData<uvw::TCPHandle>>> _tcp_sessions;
 
+    std::vector<Ipv4Subnet> _IPv4_host_list;
+    std::vector<Ipv6Subnet> _IPv6_host_list;
+
+    enum Filters {
+        OnlyHosts,
+        FiltersMAX
+    };
+    std::bitset<Filters::FiltersMAX> _f_enabled;
+
     void _read_frame_stream_file();
     void _create_frame_stream_unix_socket();
     void _create_frame_stream_tcp_socket();
 
+    void _parse_host_specs(const std::vector<std::string> &host_list);
+    bool _match_subnet(const std::string &dnstap_ip);
+
+    inline bool _filtering(const ::dnstap::Dnstap &d)
+    {
+        if (_f_enabled[Filters::OnlyHosts]) {
+            if (d.message().has_query_address() && d.message().has_response_address()) {
+                if (!_match_subnet(d.message().query_address()) && !_match_subnet(d.message().response_address())) {
+                    // message had both query and response address, and neither matched, so filter
+                    return true;
+                }
+            } else if (d.message().has_query_address() && !_match_subnet(d.message().query_address())) {
+                // message had only query address and it didn't match, so filter
+                return true;
+            } else if (d.message().has_response_address() && !_match_subnet(d.message().response_address())) {
+                // message had only response address and it didn't match, so filter
+                return true;
+            } else {
+                // message had neither query nor response address, so filter
+                return true;
+            }
+        }
+
+        return false;
+    }
+
 public:
     DnstapInputStream(const std::string &name);
     ~DnstapInputStream() = default;

src/inputs/dnstap/FrameSession.h

@@ -6,7 +6,10 @@
 
 #include "DnstapException.h"
 #include <arpa/inet.h>
+#include <cstring>
 #include <fstrm/fstrm.h>
+#include <functional>
+#include <memory>
 
 namespace visor::input::dnstap {
 
@@ -138,7 +141,7 @@ bool FrameSessionData<C>::_decode_control_frame(const void *control_frame, size_
         if (res != fstrm_res_success) {
             throw DnstapException("unable to parse content type");
         }
-        if (len_content_type != _content_type.size() || memcmp(content_type, _content_type.data(), len_content_type) != 0) {
+        if (len_content_type != _content_type.size() || std::memcmp(content_type, _content_type.data(), len_content_type) != 0) {
             throw DnstapException("content type mismatch");
         }
     }

src/inputs/dnstap/tests/test_dnstap.cpp

@@ -93,7 +93,6 @@ TEST_CASE("bi-directional frame stream process", "[dnstap][frmstrm]")
 
 TEST_CASE("dnstap file", "[dnstap][file]")
 {
-
     DnstapInputStream stream{"dnstap-test"};
     stream.config_set("dnstap_file", "inputs/dnstap/tests/fixtures/fixture.dnstap");
 
@@ -103,20 +102,97 @@ TEST_CASE("dnstap file", "[dnstap][file]")
 
 TEST_CASE("dnstap tcp socket", "[dnstap][tcp]")
 {
-
     DnstapInputStream stream{"dnstap-test"};
     stream.config_set("tcp", "127.0.0.1:5353");
+    stream.config_set<visor::Configurable::StringList>("only_hosts", {"192.168.0.0/24", "2001:db8::/48"});
 
     stream.start();
     stream.stop();
 }
 
 TEST_CASE("dnstap unix socket", "[dnstap][unix]")
 {
-
     DnstapInputStream stream{"dnstap-test"};
     stream.config_set("socket", "/tmp/dnstap-test.sock");
 
     stream.start();
     stream.stop();
 }
+
+TEST_CASE("dnstap file filter by valid subnet", "[dnstap][file][filter]")
+{
+    DnstapInputStream stream{"dnstap-test"};
+    stream.config_set("dnstap_file", "inputs/dnstap/tests/fixtures/fixture.dnstap");
+    stream.config_set<visor::Configurable::StringList>("only_hosts", {"192.168.0.0/24"});
+    uint64_t count_callbacks = 0;
+
+    stream.dnstap_signal.connect([&]([[maybe_unused]] const ::dnstap::Dnstap &d) {
+        ++count_callbacks;
+        return;
+    });
+
+    stream.start();
+    stream.stop();
+
+    CHECK(count_callbacks == 153);
+}
+
+TEST_CASE("dnstap file filter by invalid subnet", "[dnstap][file][filter]")
+{
+    DnstapInputStream stream{"dnstap-test"};
+    stream.config_set("dnstap_file", "inputs/dnstap/tests/fixtures/fixture.dnstap");
+    stream.config_set<visor::Configurable::StringList>("only_hosts", {"192.168.0.12/32"});
+    uint64_t count_callbacks = 0;
+
+    stream.dnstap_signal.connect([&]([[maybe_unused]] const ::dnstap::Dnstap &d) {
+        ++count_callbacks;
+        return;
+    });
+
+    stream.start();
+    stream.stop();
+
+    CHECK(count_callbacks == 0);
+}
+
+TEST_CASE("dnstap invalid filters", "[dnstap][tcp][filter]")
+{
+    DnstapInputStream stream{"dnstap-test"};
+    stream.config_set("tcp", "127.0.0.1:5353");
+
+    SECTION("invalid ipv4 cidr")
+    {
+        stream.config_set<visor::Configurable::StringList>("only_hosts", {"192.168.0.0/24/12ac"});
+        REQUIRE_THROWS_WITH(stream.start(), "invalid CIDR: 192.168.0.0/24/12ac");
+    }
+
+    SECTION("ipv4 cidr over max value")
+    {
+        stream.config_set<visor::Configurable::StringList>("only_hosts", {"192.168.0.0/64"});
+        REQUIRE_THROWS_WITH(stream.start(), "invalid CIDR: 192.168.0.0/64");
+    }
+
+    SECTION("invalid ipv4 ip")
+    {
+        stream.config_set<visor::Configurable::StringList>("only_hosts", {"192.168.AE.0/24"});
+        REQUIRE_THROWS_WITH(stream.start(), "invalid IPv4 address: 192.168.AE.0");
+    }
+
+    SECTION("invalid ipv6 cidr")
+    {
+        stream.config_set<visor::Configurable::StringList>("only_hosts", {"2001:db8::/48/12ac"});
+        REQUIRE_THROWS_WITH(stream.start(), "invalid CIDR: 2001:db8::/48/12ac");
+    }
+
+    SECTION("ipv6 cidr over max value")
+    {
+        stream.config_set<visor::Configurable::StringList>("only_hosts", {"2001:db8::/256"});
+        REQUIRE_THROWS_WITH(stream.start(), "invalid CIDR: 2001:db8::/256");
+    }
+
+    SECTION("invalid ipv6 ip")
+    {
+        stream.config_set<visor::Configurable::StringList>("only_hosts", {"fe80:2030:31:24/12"});
+        REQUIRE_THROWS_WITH(stream.start(), "invalid IPv6 address: fe80:2030:31:24");
+    }
+}