diff --git a/Dockerfile b/Dockerfile index 7c2473c..4d68095 100644 --- a/Dockerfile +++ b/Dockerfile @@ -29,8 +29,8 @@ RUN dnf -y update && dnf install -y python3.14-devel @development-tools gcc-c++ vectorscan-devel libtins-devel libpcap-devel boost-devel COPY ./backend/binsrc /execute/binsrc -RUN g++ binsrc/nfregex.cpp -o cppregex -std=c++23 -O3 -lnetfilter_queue -pthread -lnfnetlink $(pkg-config --cflags --libs libtins libhs libmnl) -RUN g++ binsrc/nfproxy.cpp -o cpproxy -std=c++23 -O3 -lnetfilter_queue -lpython3.14 -pthread -lnfnetlink $(pkg-config --cflags --libs libtins libmnl python3) +RUN g++ binsrc/nfregex.cpp -o cppregex -std=c++23 -O3 -Wall -lnetfilter_queue -pthread -lnfnetlink $(pkg-config --cflags --libs libtins libhs libmnl) +RUN g++ binsrc/nfproxy.cpp -o cpproxy -std=c++23 -O3 -Wall -lnetfilter_queue -lpython3.14 -pthread -lnfnetlink $(pkg-config --cflags --libs libtins libmnl python3) #Building main conteiner FROM --platform=$TARGETARCH base AS final diff --git a/backend/binsrc/classes/netfilter.cpp b/backend/binsrc/classes/netfilter.cpp index 3b7b278..d887eef 100644 --- a/backend/binsrc/classes/netfilter.cpp +++ b/backend/binsrc/classes/netfilter.cpp @@ -1,5 +1,7 @@ #include #include +#include +#include #include #include "../utils.cpp" #include "nfqueue.cpp" @@ -7,111 +9,154 @@ #ifndef NETFILTER_CLASS_CPP #define NETFILTER_CLASS_CPP -namespace Firegex { -namespace NfQueue { - -template -class ThreadNfQueue { -public: - ThreadNfQueue() = default; - virtual ~ThreadNfQueue() = default; - - std::thread thr; - BlockingQueue*> queue; - - virtual void before_loop() {} - virtual void handle_next_packet(PktRequest* pkt){} - - void loop() { - static_cast(this)->before_loop(); - PktRequest* pkt; - for(;;) { - queue.take(pkt); - try { - static_cast(this)->handle_next_packet(pkt); - } catch (const std::exception& e) { - std::cerr << "[error] Exception handling packet: " << e.what() << std::endl; - if (pkt->get_action() == FilterAction::NOACTION) { - try { - pkt->accept(); - } catch (...) {} - } - } catch (...) { - std::cerr << "[error] Unknown exception handling packet" << std::endl; - if (pkt->get_action() == FilterAction::NOACTION) { - try { - pkt->accept(); - } catch (...) {} - } - } - delete pkt; - } - } - - void run_thread_loop() { - thr = std::thread([this]() { this->loop(); }); - } -}; - -template , Worker>> -void __real_handler(PktRequest>* pkt) { - const size_t idx = hash_stream_id(pkt->sid) % pkt->ctx->size(); - - auto* converted_pkt = reinterpret_cast*>(pkt); - converted_pkt->ctx = &((*pkt->ctx)[idx]); - - converted_pkt->ctx->queue.put(converted_pkt); -} - - -template , Worker>> -class MultiThreadQueue { - static_assert(std::is_base_of_v, Worker>, - "Worker must inherit from ThreadNfQueue"); - -private: - std::vector workers; - NfQueue, __real_handler> * nfq; - uint16_t queue_num_; - - -public: - const size_t n_threads; - static constexpr int QUEUE_BASE_NUM = 1000; - - explicit MultiThreadQueue(size_t n_threads) - : n_threads(n_threads), workers(n_threads) +namespace Firegex +{ + namespace NfQueue { - if(n_threads == 0) throw std::invalid_argument("At least 1 thread required"); - - for(uint16_t qnum = QUEUE_BASE_NUM; ; qnum++) { - try { - nfq = new NfQueue, __real_handler>(qnum); - queue_num_ = qnum; - break; + + template + class ThreadNfQueue + { + public: + ThreadNfQueue() = default; + virtual ~ThreadNfQueue() = default; + + std::thread thr; + BlockingQueue *> queue; + + virtual void before_loop() {} + virtual void handle_next_packet(PktRequest *pkt) {} + + void loop() + { + static_cast(this)->before_loop(); + PktRequest *pkt; + for (;;) + { + queue.take(pkt); + try + { + static_cast(this)->handle_next_packet(pkt); + } + catch (const std::exception &e) + { + std::cerr << "[error] Exception handling packet: " << e.what() << std::endl; + if (pkt->get_action() == FilterAction::NOACTION) + { + try + { + pkt->accept(); + } + catch (...) + { + } + } + } + catch (...) + { + std::cerr << "[error] Unknown exception handling packet" << std::endl; + if (pkt->get_action() == FilterAction::NOACTION) + { + try + { + pkt->accept(); + } + catch (...) + { + } + } + } + delete pkt; + } } - catch(const std::invalid_argument&) { - if(qnum == std::numeric_limits::max()) - throw std::runtime_error("No available queue numbers"); + + void run_thread_loop() + { + thr = std::thread([this]() + { this->loop(); }); } + }; + + template , Worker>> + void __real_handler(PktRequest> *pkt) + { + const size_t idx = hash_stream_id(pkt->sid) % pkt->ctx->size(); + + auto *converted_pkt = reinterpret_cast *>(pkt); + converted_pkt->ctx = &((*pkt->ctx)[idx]); + + converted_pkt->ctx->queue.put(converted_pkt); } + + template , Worker>> + class MultiThreadQueue + { + static_assert(std::is_base_of_v, Worker>, + "Worker must inherit from ThreadNfQueue"); + + private: + std::vector workers; + NfQueue, __real_handler> *nfq; + uint16_t queue_num_; + + public: + const size_t n_threads; + static constexpr int QUEUE_BASE_NUM = 1000; + + // Listed in declaration order (that is the order they are really built in). + explicit MultiThreadQueue(size_t n_threads) + : workers(n_threads), n_threads(n_threads) + { + if (n_threads == 0) + throw std::invalid_argument("At least 1 thread required"); + + for (uint16_t qnum = QUEUE_BASE_NUM;; qnum++) + { + try + { + nfq = new NfQueue, __real_handler>(qnum); + queue_num_ = qnum; + break; + } + catch (const std::invalid_argument &) + { + if (qnum == std::numeric_limits::max()) + throw std::runtime_error("No available queue numbers"); + } + } + } + + ~MultiThreadQueue() + { + delete nfq; + } + + void start() + { + for (auto &worker : workers) + { + worker.run_thread_loop(); + } + for (;;) + { + try + { + nfq->handle_next_packet(&workers); + } + catch (const std::exception &e) + { + std::cerr << "[error] [MultiThreadQueue.start] " << e.what() << std::endl; + } + catch (...) + { + std::cerr << "[error] [MultiThreadQueue.start] unknown error while handling a packet" << std::endl; + } + } + } + + uint16_t queue_num() const { return queue_num_; } + }; + } - - ~MultiThreadQueue() { - delete nfq; - } - - void start() { - for(auto& worker : workers) { - worker.run_thread_loop(); - } - for (;;){ - nfq->handle_next_packet(&workers); - } - } - - uint16_t queue_num() const { return queue_num_; } -}; - -}} // namespace Firegex::NfQueue +} // namespace Firegex::NfQueue #endif // NETFILTER_CLASS_CPP \ No newline at end of file diff --git a/backend/binsrc/classes/nfqueue.cpp b/backend/binsrc/classes/nfqueue.cpp index 4db983c..9a70310 100644 --- a/backend/binsrc/classes/nfqueue.cpp +++ b/backend/binsrc/classes/nfqueue.cpp @@ -8,12 +8,37 @@ #include #include #include +#include +#include +#include +#include +#include using namespace std; namespace Firegex{ namespace NfQueue{ +/* Largest packet we are willing to hand back to the kernel in a verdict. + NFQUEUE payloads are carried in a netlink attribute whose length field is + 16 bit wide, so anything above this can never be re-injected anyway; the + cap exists to stop a python filter from making us allocate an arbitrary + amount of memory through mangle_custom_pkt(). */ +const size_t MAX_VERDICT_PACKET_SIZE = 0xffff; + +/* Reads the fail-open policy once, tolerating a missing variable. + Note that this only drives NFQA_CFG_F_FAIL_OPEN (what the kernel does when + the queue is full) and the fallback verdict used when a packet cannot be + parsed; the "what happens when this process is not running at all" half of + the policy lives in the nftables rules (the `bypass` queue flag). */ +inline bool nfqueue_fail_open(){ + static const bool value = [](){ + const char* env = getenv("FIREGEX_NFQUEUE_FAIL_OPEN"); + return env != nullptr && strcmp(env, "1") == 0; + }(); + return value; +} + enum class FilterAction{ DROP, ACCEPT, MANGLE, NOACTION }; enum class L4Proto { TCP, UDP, RAW }; typedef Tins::TCPIP::StreamIdentifier stream_id; @@ -115,12 +140,13 @@ class PktRequest { public: + // Listed in declaration order (that is the order they are really built in). PktRequest(const char* payload, size_t plen, T* ctx, mnl_socket* nl, nfgenmsg *nfg, nfqnl_msg_packet_hdr *ph, bool is_input): - ctx(ctx), nl(nl), res_id(nfg->res_id), - packet_id(ph->packet_id), is_input(is_input), + action(FilterAction::NOACTION), nl(nl), res_id(nfg->res_id), + packet_id(ph->packet_id), + is_ipv6((payload[0] & 0xf0) == 0x60), is_input(is_input), packet(string(payload, plen)), - action(FilterAction::NOACTION), - is_ipv6((payload[0] & 0xf0) == 0x60) + ctx(ctx) { if (is_ipv6){ ipv6 = new Tins::IPv6((uint8_t*)packet.c_str(), plen); @@ -187,15 +213,21 @@ class PktRequest { // Parsing only the header with libtins Tins::PDU *data_pdu = nullptr; size_t total_size; + // The ipv6 branch has to patch the payload length before handing the + // buffer to libtins, and for a mangled packet `data` points straight at + // the bytes object owned by the python filter: work on a private copy + // instead of mutating the caller's memory. + string raw_packet(data, data_size); + char* raw = raw_packet.data(); if (is_ipv6){ delete ipv6; ipv6 = nullptr; if (data_size >= 40){ // 40 == fixed size of ipv6 header // Resetting payload length before parsing to libtins uint16_t payload_len = htons(data_size-40); - memcpy(((uint8_t *)data)+4, &payload_len, 2); + memcpy(((uint8_t *)raw)+4, &payload_len, 2); } - ipv6 = new Tins::IPv6((uint8_t*)data, data_size); + ipv6 = new Tins::IPv6((uint8_t*)raw, data_size); if (tcp){ tcp = ipv6->find_pdu(); data_pdu = tcp; @@ -209,7 +241,7 @@ class PktRequest { }else{ delete ipv4; ipv4 = nullptr; - ipv4 = new Tins::IP((uint8_t*)data, data_size); + ipv4 = new Tins::IP((uint8_t*)raw, data_size); if (tcp){ tcp = ipv4->find_pdu(); data_pdu = tcp; @@ -222,8 +254,11 @@ class PktRequest { total_size = ipv4->size(); } _header_size = total_size - inner_data_size(data_pdu); + if (_header_size > data_size){ + throw invalid_argument("Parsed header is larger than the packet itself"); + } // Libtins can skip data if the lenght is changed to a bigger len (due to ip header total lenght), so we need to specify the data section manually - set_data(data+_header_size, data_size-_header_size); + set_data(raw+_header_size, data_size-_header_size); } void fix_tcp_ack(){ @@ -311,6 +346,9 @@ class PktRequest { void mangle_custom_pkt(const char* raw_pkt, size_t raw_pkt_size){ if (action == FilterAction::NOACTION){ try{ + if (raw_pkt_size > MAX_VERDICT_PACKET_SIZE){ + throw invalid_argument("Mangled packet is too big to be re-injected"); + } set_packet(raw_pkt, raw_pkt_size); reserialize(); action = FilterAction::MANGLE; @@ -345,7 +383,16 @@ class PktRequest { private: void perform_action(bool do_serialize = true){ - char buf[MNL_SOCKET_BUFFER_SIZE+packet.size()]; + /* This used to be a VLA sized on the (attacker influenced) packet + length, which overflows the worker stack as soon as a filter mangles + a packet into something large. A thread_local buffer keeps the + allocation off the stack while still avoiding a malloc per packet. */ + static thread_local vector verdict_buffer; + const size_t needed = MNL_SOCKET_BUFFER_SIZE + packet.size(); + if (verdict_buffer.size() < needed){ + verdict_buffer.resize(needed); + } + char* buf = verdict_buffer.data(); struct nlmsghdr *nlh_verdict = nfq_nlmsg_put(buf, NFQNL_MSG_VERDICT, ntohs(res_id)); switch (action) { @@ -462,9 +509,7 @@ class NfQueue { nlh = nfq_nlmsg_put(queue_msg_buffer, NFQNL_MSG_CONFIG, queue_num); nfq_nlmsg_cfg_put_params(nlh, NFQNL_COPY_PACKET, 0xffff); - char * enable_fail_open = getenv("FIREGEX_NFQUEUE_FAIL_OPEN"); - - if (strcmp(enable_fail_open, "1") == 0){ + if (nfqueue_fail_open()){ mnl_attr_put_u32(nlh, NFQA_CFG_FLAGS, htonl(NFQA_CFG_F_GSO|NFQA_CFG_F_FAIL_OPEN)); mnl_attr_put_u32(nlh, NFQA_CFG_MASK, htonl(NFQA_CFG_F_GSO|NFQA_CFG_F_FAIL_OPEN)); }else{ @@ -511,6 +556,18 @@ class NfQueue { private: + /* Issues a plain verdict for a packet we could not build a PktRequest for. + Every early return below has to go through this: a packet that is never + given a verdict stays pinned in the kernel queue until it fills up. */ + static void _send_raw_verdict(mnl_socket* nl, uint16_t res_id, uint32_t packet_id, int verdict) { + char buf[MNL_SOCKET_BUFFER_SIZE]; + struct nlmsghdr *nlh_verdict = nfq_nlmsg_put(buf, NFQNL_MSG_VERDICT, ntohs(res_id)); + nfq_nlmsg_verdict_put(nlh_verdict, ntohl(packet_id), verdict); + if (mnl_socket_sendto(nl, nlh_verdict, nlh_verdict->nlmsg_len) < 0) { + cerr << "[error] [NfQueue._send_raw_verdict] failed to send the verdict" << endl; + } + } + static int _real_queue_cb(const nlmsghdr *nlh, void *data_ptr) { internal_nfqueue_execution_data_tmp* info = (internal_nfqueue_execution_data_tmp*) data_ptr; @@ -518,28 +575,37 @@ class NfQueue { //Extract attributes from the nlmsghdr nlattr *attr[NFQA_MAX+1] = {}; + /* None of the paths below may return MNL_CB_ERROR: that aborts + mnl_cb_run(), which the caller turns into an exception. A single + unexpected message must never be able to take the interceptor down, + because the nftables rules outlive the process. */ if (nfq_nlmsg_parse(nlh, attr) < 0) { cerr << "[error] [NfQueue._real_queue_cb] problems parsing" << endl; - return MNL_CB_ERROR; + return MNL_CB_OK; } if (attr[NFQA_PACKET_HDR] == nullptr) { + // Without the header there is no packet id, so there is nothing to + // answer: just skip the message. cerr << "[error] [NfQueue._real_queue_cb] packet header not set" << endl; - return MNL_CB_ERROR; - } - if (attr[NFQA_MARK] == nullptr) { - cerr << "[error] [NfQueue._real_queue_cb] mark not set" << endl; - return MNL_CB_ERROR; + return MNL_CB_OK; } struct nfqnl_msg_packet_hdr *ph = (nfqnl_msg_packet_hdr*) mnl_attr_get_payload(attr[NFQA_PACKET_HDR]); struct nfgenmsg *nfg = (nfgenmsg *)mnl_nlmsg_get_payload(nlh); + // A packet we cannot inspect follows the service's fail-open policy + // instead of being silently let through. + const int fallback_verdict = nfqueue_fail_open() ? NF_ACCEPT : NF_DROP; + + if (attr[NFQA_MARK] == nullptr) { + cerr << "[error] [NfQueue._real_queue_cb] mark not set" << endl; + _send_raw_verdict(info->nl, nfg->res_id, ph->packet_id, fallback_verdict); + return MNL_CB_OK; + } + if (attr[NFQA_PAYLOAD] == nullptr) { cerr << "[error] [NfQueue._real_queue_cb] payload not set" << endl; - char buf[MNL_SOCKET_BUFFER_SIZE]; - struct nlmsghdr *nlh_verdict = nfq_nlmsg_put(buf, NFQNL_MSG_VERDICT, ntohs(nfg->res_id)); - nfq_nlmsg_verdict_put(nlh_verdict, ntohl(ph->packet_id), NF_ACCEPT); - mnl_socket_sendto(info->nl, nlh_verdict, nlh_verdict->nlmsg_len); + _send_raw_verdict(info->nl, nfg->res_id, ph->packet_id, NF_ACCEPT); return MNL_CB_OK; } @@ -547,10 +613,58 @@ class NfQueue { uint16_t plen = mnl_attr_get_payload_len(attr[NFQA_PAYLOAD]); char *payload = (char *)mnl_attr_get_payload(attr[NFQA_PAYLOAD]); + if (plen == 0) { + cerr << "[error] [NfQueue._real_queue_cb] empty payload" << endl; + _send_raw_verdict(info->nl, nfg->res_id, ph->packet_id, NF_ACCEPT); + return MNL_CB_OK; + } + + /* NFQA_CFG_F_GSO is enabled, so the kernel can hand us a super packet + bigger than the copy range; NFQA_CAP_LEN then reports the real + length. Parsing a truncated packet yields a wrong view of the stream + and re-injecting one would corrupt the connection, so treat it like + any other packet we cannot inspect. */ + if (attr[NFQA_CAP_LEN] != nullptr) { + uint32_t cap_len = ntohl(mnl_attr_get_u32(attr[NFQA_CAP_LEN])); + if (cap_len > plen) { + cerr << "[warning] [NfQueue._real_queue_cb] truncated packet (" + << plen << " of " << cap_len << " bytes), applying the fail-open policy" << endl; + _send_raw_verdict(info->nl, nfg->res_id, ph->packet_id, fallback_verdict); + return MNL_CB_OK; + } + } + bool is_input = ntohl(mnl_attr_get_u32(attr[NFQA_MARK])) & 0x1; // == 0x1337 that is odd - handle_func(new PktRequest( - payload, plen, (D*)info->data, info->nl, nfg, ph, is_input - )); + + /* libtins throws (Tins::malformed_packet & friends) on a malformed ip + header, truncated tcp options or bogus ipv6 extension headers. That + exception would otherwise unwind through libmnl's C frames and kill + the whole process, which is a remotely triggerable way of disabling + the firewall: one crafted packet is enough. */ + PktRequest* pkt = nullptr; + try { + pkt = new PktRequest( + payload, plen, (D*)info->data, info->nl, nfg, ph, is_input + ); + } catch (const std::exception& e) { + cerr << "[error] [NfQueue._real_queue_cb] cannot parse the packet: " << e.what() << endl; + _send_raw_verdict(info->nl, nfg->res_id, ph->packet_id, fallback_verdict); + return MNL_CB_OK; + } catch (...) { + cerr << "[error] [NfQueue._real_queue_cb] cannot parse the packet (unknown error)" << endl; + _send_raw_verdict(info->nl, nfg->res_id, ph->packet_id, fallback_verdict); + return MNL_CB_OK; + } + + try { + handle_func(pkt); // Takes ownership of pkt + } catch (const std::exception& e) { + cerr << "[error] [NfQueue._real_queue_cb] cannot enqueue the packet: " << e.what() << endl; + if (pkt->get_action() == FilterAction::NOACTION) { + _send_raw_verdict(info->nl, nfg->res_id, ph->packet_id, fallback_verdict); + } + delete pkt; + } return MNL_CB_OK; } @@ -561,6 +675,7 @@ class NfQueue { nl = nullptr; } delete[] queue_msg_buffer; + queue_msg_buffer = nullptr; } inline ssize_t _send_config_cmd(nfqnl_msg_config_cmds cmd){ @@ -569,8 +684,31 @@ class NfQueue { return mnl_socket_sendto(nl, nlh, nlh->nlmsg_len); } + /* Retries the errors that simply mean "this message is gone, carry on". + ENOBUFS signals kernel side packet loss, ENOSPC is what libmnl reports + for a netlink message too big for our buffer (reachable with GSO super + packets) and EINTR is just a signal. Treating any of them as fatal used + to abort the whole interceptor. */ inline ssize_t _recv_packet(){ - return mnl_socket_recvfrom(nl, queue_msg_buffer, NFQUEUE_BUFFER_SIZE); + for(;;){ + ssize_t ret = mnl_socket_recvfrom(nl, queue_msg_buffer, NFQUEUE_BUFFER_SIZE); + if (ret >= 0){ + return ret; + } + switch (errno) { + case EINTR: + continue; + case ENOBUFS: + cerr << "[warning] [NfQueue._recv_packet] the kernel dropped packets (ENOBUFS)" << endl; + continue; + case ENOSPC: + case EMSGSIZE: + cerr << "[warning] [NfQueue._recv_packet] netlink message too big for the buffer, skipped" << endl; + continue; + default: + return -1; + } + } } }; diff --git a/backend/binsrc/nfproxy.cpp b/backend/binsrc/nfproxy.cpp index 613b922..a8a83e3 100644 --- a/backend/binsrc/nfproxy.cpp +++ b/backend/binsrc/nfproxy.cpp @@ -85,13 +85,31 @@ Final note: is not raccomanded to use variables that starts with __firegex_ in y +/* Upper bound for a filter code blob. Without it a desynchronised or corrupt + length prefix makes us try to allocate an arbitrary amount of memory. */ +const uint32_t MAX_PYFILTER_CODE_SIZE = 64 * 1024 * 1024; + void config_updater (){ while (true){ PyThreadState* state = PyEval_SaveThread(); // Release GIL while doing IO operation uint32_t code_size; - memcpy(&code_size, control_socket.recv(4).c_str(), 4); - code_size = be32toh(code_size); - string code = control_socket.recv(code_size); + string code; + try{ + string raw_code_size = control_socket.recv(4); + memcpy(&code_size, raw_code_size.c_str(), 4); + code_size = be32toh(code_size); + if (code_size > MAX_PYFILTER_CODE_SIZE){ + throw runtime_error("Announced filter code size is out of range"); + } + code = control_socket.recv(code_size); + }catch(const std::exception& e){ + // The control socket is the only link with the backend: if it is + // gone there is nothing left to filter for, so exit cleanly instead + // of letting the exception terminate the process. + PyEval_AcquireThread(state); + cerr << "[fatal] [updater] Control socket error: " << e.what() << endl; + exit(EXIT_FAILURE); + } #ifdef DEBUG cerr << "[DEBUG] [updater] Received code: " << code << endl; #endif @@ -104,6 +122,9 @@ void config_updater (){ }catch(const std::exception& e){ cerr << "[error] [updater] Failed to build new configuration!" << endl; control_socket << "ACK FAIL " << e.what() << endl; + }catch(...){ + cerr << "[error] [updater] Failed to build new configuration!" << endl; + control_socket << "ACK FAIL unknown error" << endl; } } } @@ -133,7 +154,18 @@ int main(int argc, char *argv[]) { cerr << "[info] [main] Queue: " << queue.queue_num() << " threads assigned: " << n_of_threads << endl; thread qthr([&](){ - queue.start(); + /* An exception escaping a std::thread is std::terminate(): exit + explicitly instead, so the backend sees a failed process it can + react to rather than an abort. */ + try{ + queue.start(); + }catch(const std::exception& e){ + cerr << "[fatal] [main] Queue loop stopped: " << e.what() << endl; + exit(EXIT_FAILURE); + }catch(...){ + cerr << "[fatal] [main] Queue loop stopped with an unknown error" << endl; + exit(EXIT_FAILURE); + } }); config_updater(); qthr.join(); diff --git a/backend/binsrc/nfregex.cpp b/backend/binsrc/nfregex.cpp index a37e2b2..ec76b62 100644 --- a/backend/binsrc/nfregex.cpp +++ b/backend/binsrc/nfregex.cpp @@ -44,6 +44,9 @@ void config_updater (){ }catch(const std::exception& e){ cerr << "[error] [updater] Failed to build new configuration!" << endl; osyncstream(cout) << "ACK FAIL " << e.what() << endl; + }catch(...){ + cerr << "[error] [updater] Failed to build new configuration!" << endl; + osyncstream(cout) << "ACK FAIL unknown error" << endl; } } @@ -76,7 +79,7 @@ int main(int argc, char *argv[]){ stream_mode = false; } - bool fail_open = strcmp(getenv("FIREGEX_NFQUEUE_FAIL_OPEN"), "1") == 0; + bool fail_open = Firegex::NfQueue::nfqueue_fail_open(); regex_config.reset(new RegexRules(stream_mode)); @@ -85,7 +88,18 @@ int main(int argc, char *argv[]){ cerr << "[info] [main] Queue: " << queue_manager.queue_num() << " threads assigned: " << n_of_threads << " stream mode: " << stream_mode << " fail open: " << fail_open << endl; thread qthr([&](){ - queue_manager.start(); + /* An exception escaping a std::thread is std::terminate(): exit + explicitly instead, so the backend sees a failed process it can + react to rather than an abort. */ + try{ + queue_manager.start(); + }catch(const std::exception& e){ + cerr << "[fatal] [main] Queue loop stopped: " << e.what() << endl; + exit(EXIT_FAILURE); + }catch(...){ + cerr << "[fatal] [main] Queue loop stopped with an unknown error" << endl; + exit(EXIT_FAILURE); + } }); config_updater(); qthr.join(); diff --git a/backend/binsrc/regex/regex_rules.cpp b/backend/binsrc/regex/regex_rules.cpp index 83fd6dc..6d1d698 100644 --- a/backend/binsrc/regex/regex_rules.cpp +++ b/backend/binsrc/regex/regex_rules.cpp @@ -80,7 +80,7 @@ class RegexRules{ private: static inline uint16_t glob_seq = 0; - uint16_t version; + uint16_t version = 0; vector> decoded_input_rules; vector> decoded_output_rules; bool is_stream = true; @@ -104,7 +104,7 @@ class RegexRules{ vector regex_match_rules(n_of_regex); vector regex_array_ids(n_of_regex); vector regex_flags(n_of_regex); - for(int i = 0; i < n_of_regex; i++){ + for(size_t i = 0; i < n_of_regex; i++){ regex_match_rules[i] = decoded[i].second.regex.c_str(); regex_array_ids[i] = i; regex_flags[i] = HS_FLAG_SINGLEMATCH | HS_FLAG_ALLOWEMPTY; @@ -114,7 +114,7 @@ class RegexRules{ } #ifdef DEBUG cerr << "[DEBUG] [RegexRules.fill_ruleset] compiling " << n_of_regex << " regexes..." << endl; - for (int i = 0; i < n_of_regex; i++){ + for (size_t i = 0; i < n_of_regex; i++){ cerr << "[DEBUG] [RegexRules.fill_ruleset] regex[" << i << "]: " << decoded[i].first << " " << decoded[i].second.regex << endl; cerr << "[DEBUG] [RegexRules.fill_ruleset] regex_match_rules[" << i << "]: " << regex_match_rules[i] << endl; cerr << "[DEBUG] [RegexRules.fill_ruleset] regex_flags[" << i << "]: " << regex_flags[i] << endl; @@ -139,7 +139,7 @@ class RegexRules{ } ruleset.hs_db = rebuilt_db; ruleset.regexes = vector(n_of_regex); - for(int i = 0; i < n_of_regex; i++){ + for(size_t i = 0; i < n_of_regex; i++){ ruleset.regexes[i] = decoded[i].first; } } @@ -147,24 +147,25 @@ class RegexRules{ public: RegexRules(vector raw_rules, bool is_stream){ this->is_stream = is_stream; + /* `throw current_exception()` used to be used here: that throws a + std::exception_ptr *object*, which does not derive from + std::exception, so the `catch(const std::exception&)` in the + config updater never matched it and an invalid regex terminated + the whole process. A bare `throw;` rethrows the original. */ for(string ele : raw_rules){ - try{ - decoded_regex rule = decode_regex(ele); - if (rule.direction == FilterDirection::CTOS){ - decoded_input_rules.push_back(make_pair(ele, rule)); - }else{ - decoded_output_rules.push_back(make_pair(ele, rule)); - } - }catch(...){ - throw current_exception(); + decoded_regex rule = decode_regex(ele); + if (rule.direction == FilterDirection::CTOS){ + decoded_input_rules.push_back(make_pair(ele, rule)); + }else{ + decoded_output_rules.push_back(make_pair(ele, rule)); } } - fill_ruleset(decoded_input_rules, input_ruleset); try{ + fill_ruleset(decoded_input_rules, input_ruleset); fill_ruleset(decoded_output_rules, output_ruleset); }catch(...){ free_dbs(); - throw current_exception(); + throw; } this->version = ++glob_seq; // 0 version is the null version } @@ -173,10 +174,12 @@ class RegexRules{ return version; } - RegexRules(bool is_stream){ - vector no_rules; - RegexRules(no_rules, is_stream); - } + /* These two have to *delegate*: writing `RegexRules(...)` as a plain + statement in the body just builds a temporary and throws it away, + leaving this instance with an uninitialised version and with the + is_stream argument silently ignored (which is why MATCH_MODE=block, + used by every udp service, never took effect). */ + RegexRules(bool is_stream): RegexRules(vector(), is_stream) {} bool stream_mode(){ return is_stream; @@ -184,9 +187,7 @@ class RegexRules{ - RegexRules(){ - RegexRules(true); - } + RegexRules(): RegexRules(true) {} ~RegexRules(){ free_dbs(); diff --git a/backend/binsrc/utils.cpp b/backend/binsrc/utils.cpp index 59ca77b..e4a923c 100644 --- a/backend/binsrc/utils.cpp +++ b/backend/binsrc/utils.cpp @@ -76,19 +76,42 @@ public: return *this; } + // Stream sockets are free to accept fewer bytes than asked for, so a single + // write() can silently truncate a message and desynchronise the protocol. void send(const std::string& data) { - if (::write(sockfd, data.c_str(), data.size()) == -1) { - throw std::runtime_error(std::string("write error: ") + std::strerror(errno)); + size_t sent = 0; + while (sent < data.size()) { + ssize_t written = ::write(sockfd, data.c_str() + sent, data.size() - sent); + if (written < 0) { + if (errno == EINTR) continue; + throw std::runtime_error(std::string("write error: ") + std::strerror(errno)); + } + if (written == 0) { + throw std::runtime_error("write error: connection closed by peer"); + } + sent += static_cast(written); } } + /* Reads exactly `size` bytes. The previous single read() returned whatever + happened to be buffered, which for the nfproxy control protocol meant a + short read on the 4 byte length prefix (reading past the end of the + returned string) or on the filter code itself, leaving the leftover + bytes to be parsed as the next length prefix. */ std::string recv(size_t size) { std::string buffer(size, '\0'); - ssize_t bytesRead = ::read(sockfd, &buffer[0], size); - if (bytesRead <= 0) { - throw std::runtime_error(std::string("read error: ") + std::strerror(errno)); + size_t total = 0; + while (total < size) { + ssize_t bytesRead = ::read(sockfd, &buffer[total], size - total); + if (bytesRead < 0) { + if (errno == EINTR) continue; + throw std::runtime_error(std::string("read error: ") + std::strerror(errno)); + } + if (bytesRead == 0) { + throw std::runtime_error("read error: connection closed by peer"); + } + total += static_cast(bytesRead); } - buffer.resize(bytesRead); // resize to actual bytes read return buffer; } @@ -101,22 +124,21 @@ public: // Overload for manipulators (e.g., std::endl) UnixClientConnection& operator<<(std::ostream& (*manip)(std::ostream&)) { - // Check if the manipulator is std::endl (or equivalent flush) - if (manip == static_cast(std::endl)){ + // The flush test used to be `if (static_cast<...>(std::flush))`, i.e. the + // address of a function, which is always true: every manipulator flushed + // and std::endl flushed twice. + if (manip == static_cast(std::endl)) { streamBuffer << '\n'; // Add a newline - std::string packet = streamBuffer.str(); - streamBuffer.str(""); // Clear the buffer - // Send the accumulated data as one packet - send(packet); - } - if (static_cast(std::flush)) { - std::string packet = streamBuffer.str(); - streamBuffer.str(""); // Clear the buffer - // Send the accumulated data as one packet - send(packet); - } else { + } else if (manip != static_cast(std::flush)) { // For other manipulators, simply pass them to the buffer streamBuffer << manip; + return *this; + } + std::string packet = streamBuffer.str(); + streamBuffer.str(""); // Clear the buffer + // Send the accumulated data as one packet + if (!packet.empty()) { + send(packet); } return *this; } @@ -174,7 +196,10 @@ private: std::queue private_std_queue; std::condition_variable condNotEmpty; std::condition_variable condNotFull; - size_t count; // Guard with Mutex + // Must be initialised: an indeterminate value either disables the backpressure + // entirely (the queue then grows until the process is OOM killed) or, if it + // happens to equal MAX, blocks the netlink thread on the very first packet. + size_t count = 0; // Guard with Mutex public: void put(T new_value) diff --git a/backend/modules/nfproxy/firegex.py b/backend/modules/nfproxy/firegex.py index 74c4fbf..d3d5f10 100644 --- a/backend/modules/nfproxy/firegex.py +++ b/backend/modules/nfproxy/firegex.py @@ -10,6 +10,8 @@ from utils import nicenessify nft = FiregexTables() +MODULE = "nfproxy" + OUTSTREAM_BUFFER_SIZE = 1024*10 class FiregexInterceptor: @@ -36,11 +38,15 @@ class FiregexInterceptor: self.expection_function = None self.outstrem_task: asyncio.Task self.outstrem_buffer = "" + self.watchdog_task: asyncio.Task = None + self.exit_callback = None + self.stopped = False @classmethod - async def start(cls, srv: Service, outstream_func=None, exception_func=None): + async def start(cls, srv: Service, outstream_func=None, exception_func=None, on_exit=None): self = cls() self.srv = srv + self.exit_callback = on_exit self.filter_map_lock = asyncio.Lock() self.update_config_lock = asyncio.Lock() self.sock_conn_lock = asyncio.Lock() @@ -55,6 +61,7 @@ class FiregexInterceptor: self.server_task = asyncio.create_task(self.unix_sock.serve_forever()) queue_range = await self._start_binary() self.update_task = asyncio.create_task(self.update_stats()) + self.watchdog_task = asyncio.create_task(self._watch_process()) nft.add(self.srv, queue_range) if not self.ack_lock.locked(): await self.ack_lock.acquire() @@ -72,7 +79,7 @@ class FiregexInterceptor: self.ack_status = False self.ack_fail_what = "Can't read from nfq client" self.ack_lock.release() - await self.stop() + await self.stop(expected=False) traceback.print_exc() # Python can't print it alone? nope it's python... wasted 1 day :) raise HTTPException(status_code=500, detail="Can't read from nfq client") from e self.outstrem_buffer+=out_data @@ -130,7 +137,7 @@ class FiregexInterceptor: self.ack_status = False self.ack_fail_what = "Can't read from nfq client" self.ack_lock.release() - await self.stop() + await self.stop(expected=False) raise HTTPException(status_code=500, detail="Can't read from nfq client") from e if line.startswith("BLOCKED "): filter_name = line.split()[1] @@ -161,7 +168,43 @@ class FiregexInterceptor: except Exception: traceback.print_exc() - async def stop(self): + + async def _watch_process(self): + """Detects an unexpected death of the interceptor binary. + + The nftables rules are installed independently of this process and + survive it, so a crashed interceptor leaves the service completely + unfiltered (silently accepted when the service is fail_open, dropped + otherwise) while the UI keeps reporting it as active. Nothing used to + notice: the stdout reader just saw EOF and returned. + """ + try: + returncode = await self.process.wait() + except asyncio.CancelledError: + return + if self.stopped: + return # We killed it ourselves + print(f"[error] [{MODULE}] The interceptor of service {self.srv.id} exited unexpectedly with code {returncode}") + if self.exit_callback: + try: + await run_func(self.exit_callback, returncode) + except asyncio.CancelledError: + raise + except Exception: + traceback.print_exc() + + async def stop(self, expected: bool = True): + """Tears the interceptor down. + + `expected` tells the watchdog whether this shutdown was asked for: an + internal failure passes False so that the crash is still reported and + the service can be recovered, instead of looking like a clean stop. + """ + self.stopped = expected + # The watchdog itself calls stop() through the exit callback: cancelling + # the running task from inside would abort the recovery. + if self.watchdog_task and self.watchdog_task is not asyncio.current_task(): + self.watchdog_task.cancel() self.server_task.cancel() self.update_task.cancel() self.unix_sock.close() diff --git a/backend/modules/nfproxy/firewall.py b/backend/modules/nfproxy/firewall.py index 04bcdd6..89bea36 100644 --- a/backend/modules/nfproxy/firewall.py +++ b/backend/modules/nfproxy/firewall.py @@ -1,10 +1,11 @@ import asyncio +import time +import traceback from modules.nfproxy.firegex import FiregexInterceptor from modules.nfproxy.nftables import FiregexTables, FiregexFilter from modules.nfproxy.models import Service, PyFilter from utils.sqlite import SQLite -from utils.sqlite import SQLite -from utils import run_func +from utils import run_func, socketio_emit class STATUS: STOP = "stop" @@ -12,6 +13,12 @@ class STATUS: nft = FiregexTables() +MODULE = "nfproxy" +# How many times a service may be brought back up before it is declared broken, +# and after how long without a crash the counter goes back to zero. +MAX_RESTART_ATTEMPTS = 5 +CRASH_COUNTER_RESET_SECONDS = 60 + class ServiceManager: def __init__(self, srv: Service, db, outstream_func=None, exception_func=None): self.srv = srv @@ -20,6 +27,8 @@ class ServiceManager: self.filters: dict[str, FiregexFilter] = {} self.lock = asyncio.Lock() self.interceptor = None + self._crash_count = 0 + self._last_crash_time = 0.0 self.outstream_function = outstream_func self.last_exception_time = 0 async def excep_internal_handler(srv, exc_time): @@ -68,10 +77,45 @@ class ServiceManager: else: return "" + + async def _on_interceptor_exit(self, returncode: int): + """Recovers from an unexpected death of the interceptor binary. + + Restarting rebuilds both the process and its nftables rules; it is + attempted a bounded number of times so that a systematically crashing + interceptor does not turn into a restart loop. When the budget is + exhausted the service is stopped for good, which at least removes the + rules and makes the failure visible instead of leaving a service that + claims to be active while filtering nothing. + """ + async with self.lock: + if self.interceptor is None or self.status != STATUS.ACTIVE: + return # Already being stopped on purpose + # The process is already gone, but its sockets and reader tasks are + # not: release them before building a replacement. + await self.interceptor.stop() + self.interceptor = None + now = time.monotonic() + if now - self._last_crash_time > CRASH_COUNTER_RESET_SECONDS: + self._crash_count = 0 + self._last_crash_time = now + self._crash_count += 1 + if self._crash_count > MAX_RESTART_ATTEMPTS: + print(f"[error] [{MODULE}] Service {self.srv.id} crashed {self._crash_count} times (last exit code {returncode}), giving up and stopping it") + await self.stop() + else: + print(f"[warning] [{MODULE}] Restarting the interceptor of service {self.srv.id} after exit code {returncode} ({self._crash_count}/{MAX_RESTART_ATTEMPTS})") + try: + await self.start() + except Exception: + traceback.print_exc() + await self.stop() + await socketio_emit([MODULE]) + async def start(self): if not self.interceptor: nft.delete(self.srv) - self.interceptor = await FiregexInterceptor.start(self.srv, outstream_func=self.outstream_function, exception_func=self.exception_function) + self.interceptor = await FiregexInterceptor.start(self.srv, outstream_func=self.outstream_function, exception_func=self.exception_function, on_exit=self._on_interceptor_exit) await self._update_filters_from_db() self._set_status(STATUS.ACTIVE) diff --git a/backend/modules/nfproxy/nftables.py b/backend/modules/nfproxy/nftables.py index 9f80d47..711e879 100644 --- a/backend/modules/nfproxy/nftables.py +++ b/backend/modules/nfproxy/nftables.py @@ -24,6 +24,22 @@ def resolve_target(srv: Service) -> tuple[str, int] | None: return loopback_ip, clear_port return srv.ip_int, srv.port + +def build_queue_statement(srv: Service, init: int, end: int) -> dict: + """Builds the nftables `queue` statement for a service. + + The `bypass` flag tells the kernel to accept packets when no process is + bound to the queue, which is what happens after the interceptor binary + dies. Tying it to the service's fail_open setting keeps the two halves of + the policy consistent: without fail_open a dead interceptor must not + silently let the traffic through unfiltered. + """ + statement = {"num": str(init) if init == end else {"range": [init, end]}} + if srv.fail_open: + statement["flags"] = ["bypass"] + return {"queue": statement} + + class FiregexFilter: def __init__(self, proto:str, port:int, ip_int:str, target:str, id:int): self.id = id @@ -104,7 +120,7 @@ class FiregexTables(NFTableManager): {'match': {'left': {'payload': {'protocol': ip_family(target_ip), 'field': 'saddr'}}, 'op': '==', 'right': nftables_int_to_json(target_ip)}}, {'match': {"left": { "payload": {"protocol": convert_protocol_to_l4(str(srv.proto)), "field": "dport"}}, "op": "==", "right": int(target_port)}}, {"mangle": {"key": {"meta": {"key": "mark"}},"value": 0x1338}}, - {"queue": {"num": str(init) if init == end else {"range":[init, end] }, "flags": ["bypass"]}} + build_queue_statement(srv, init, end) ] }}}, {"insert":{"rule":{ # Send non-TLS inbound traffic to NFQUEUE (Server -> Client) @@ -115,7 +131,7 @@ class FiregexTables(NFTableManager): {'match': {'left': {'payload': {'protocol': ip_family(target_ip), 'field': 'saddr'}}, 'op': '==', 'right': nftables_int_to_json(target_ip)}}, {'match': {"left": { "payload": {"protocol": convert_protocol_to_l4(str(srv.proto)), "field": "sport"}}, "op": "==", "right": int(target_port)}}, {"mangle": {"key": {"meta": {"key": "mark"}},"value": 0x1337}}, - {"queue": {"num": str(init) if init == end else {"range":[init, end] }, "flags": ["bypass"]}} + build_queue_statement(srv, init, end) ] }}} ) diff --git a/backend/modules/nfregex/firegex.py b/backend/modules/nfregex/firegex.py index d73b80f..9dbd7d3 100644 --- a/backend/modules/nfregex/firegex.py +++ b/backend/modules/nfregex/firegex.py @@ -10,6 +10,8 @@ from utils import nicenessify nft = FiregexTables() +MODULE = "nfregex" + async def test_regex_validity(regex: str) -> bool: proxy_binary_path = os.path.join(os.path.dirname(os.path.abspath(__file__)),"../cppregex") process = await asyncio.create_subprocess_exec( @@ -82,15 +84,20 @@ class FiregexInterceptor: self.ack_status = None self.ack_fail_what = "Queue response timed-out" self.ack_lock = asyncio.Lock() + self.watchdog_task: asyncio.Task = None + self.exit_callback = None + self.stopped = False @classmethod - async def start(cls, srv: Service): + async def start(cls, srv: Service, on_exit=None): self = cls() self.srv = srv + self.exit_callback = on_exit self.filter_map_lock = asyncio.Lock() self.update_config_lock = asyncio.Lock() queue_range = await self._start_binary() self.update_task = asyncio.create_task(self.update_blocked()) + self.watchdog_task = asyncio.create_task(self._watch_process()) nft.add(self.srv, queue_range) if not self.ack_lock.locked(): await self.ack_lock.acquire() @@ -147,7 +154,43 @@ class FiregexInterceptor: except Exception: traceback.print_exc() - async def stop(self): + + async def _watch_process(self): + """Detects an unexpected death of the interceptor binary. + + The nftables rules are installed independently of this process and + survive it, so a crashed interceptor leaves the service completely + unfiltered (silently accepted when the service is fail_open, dropped + otherwise) while the UI keeps reporting it as active. Nothing used to + notice: the stdout reader just saw EOF and returned. + """ + try: + returncode = await self.process.wait() + except asyncio.CancelledError: + return + if self.stopped: + return # We killed it ourselves + print(f"[error] [{MODULE}] The interceptor of service {self.srv.id} exited unexpectedly with code {returncode}") + if self.exit_callback: + try: + await run_func(self.exit_callback, returncode) + except asyncio.CancelledError: + raise + except Exception: + traceback.print_exc() + + async def stop(self, expected: bool = True): + """Tears the interceptor down. + + `expected` tells the watchdog whether this shutdown was asked for: an + internal failure passes False so that the crash is still reported and + the service can be recovered, instead of looking like a clean stop. + """ + self.stopped = expected + # The watchdog itself calls stop() through the exit callback: cancelling + # the running task from inside would abort the recovery. + if self.watchdog_task and self.watchdog_task is not asyncio.current_task(): + self.watchdog_task.cancel() self.update_task.cancel() if self.process and self.process.returncode is None: self.process.kill() diff --git a/backend/modules/nfregex/firewall.py b/backend/modules/nfregex/firewall.py index 6ae3d58..08a0344 100644 --- a/backend/modules/nfregex/firewall.py +++ b/backend/modules/nfregex/firewall.py @@ -1,8 +1,11 @@ import asyncio +import time +import traceback from modules.nfregex.firegex import FiregexInterceptor, RegexFilter from modules.nfregex.nftables import FiregexTables, FiregexFilter from modules.nfregex.models import Regex, Service from utils.sqlite import SQLite +from utils import socketio_emit class STATUS: STOP = "stop" @@ -10,6 +13,12 @@ class STATUS: nft = FiregexTables() +MODULE = "nfregex" +# How many times a service may be brought back up before it is declared broken, +# and after how long without a crash the counter goes back to zero. +MAX_RESTART_ATTEMPTS = 5 +CRASH_COUNTER_RESET_SECONDS = 60 + class ServiceManager: def __init__(self, srv: Service, db): @@ -19,6 +28,8 @@ class ServiceManager: self.filters: dict[int, FiregexFilter] = {} self.lock = asyncio.Lock() self.interceptor = None + self._crash_count = 0 + self._last_crash_time = 0.0 async def _update_filters_from_db(self): regexes = [ @@ -58,10 +69,45 @@ class ServiceManager: if persist: self.__update_status_db(status) + + async def _on_interceptor_exit(self, returncode: int): + """Recovers from an unexpected death of the interceptor binary. + + Restarting rebuilds both the process and its nftables rules; it is + attempted a bounded number of times so that a systematically crashing + interceptor does not turn into a restart loop. When the budget is + exhausted the service is stopped for good, which at least removes the + rules and makes the failure visible instead of leaving a service that + claims to be active while filtering nothing. + """ + async with self.lock: + if self.interceptor is None or self.status != STATUS.ACTIVE: + return # Already being stopped on purpose + # The process is already gone, but its sockets and reader tasks are + # not: release them before building a replacement. + await self.interceptor.stop() + self.interceptor = None + now = time.monotonic() + if now - self._last_crash_time > CRASH_COUNTER_RESET_SECONDS: + self._crash_count = 0 + self._last_crash_time = now + self._crash_count += 1 + if self._crash_count > MAX_RESTART_ATTEMPTS: + print(f"[error] [{MODULE}] Service {self.srv.id} crashed {self._crash_count} times (last exit code {returncode}), giving up and stopping it") + await self.stop() + else: + print(f"[warning] [{MODULE}] Restarting the interceptor of service {self.srv.id} after exit code {returncode} ({self._crash_count}/{MAX_RESTART_ATTEMPTS})") + try: + await self.start() + except Exception: + traceback.print_exc() + await self.stop() + await socketio_emit([MODULE]) + async def start(self): if not self.interceptor: nft.delete(self.srv) - self.interceptor = await FiregexInterceptor.start(self.srv) + self.interceptor = await FiregexInterceptor.start(self.srv, on_exit=self._on_interceptor_exit) await self._update_filters_from_db() self._set_status(STATUS.ACTIVE) diff --git a/backend/modules/nfregex/nftables.py b/backend/modules/nfregex/nftables.py index 997fc8a..4653f23 100644 --- a/backend/modules/nfregex/nftables.py +++ b/backend/modules/nfregex/nftables.py @@ -16,6 +16,22 @@ def resolve_target(srv: Service) -> tuple[str, int] | None: return loopback_ip, clear_port return srv.ip_int, srv.port + +def build_queue_statement(srv: Service, init: int, end: int) -> dict: + """Builds the nftables `queue` statement for a service. + + The `bypass` flag tells the kernel to accept packets when no process is + bound to the queue, which is what happens after the interceptor binary + dies. Tying it to the service's fail_open setting keeps the two halves of + the policy consistent: without fail_open a dead interceptor must not + silently let the traffic through unfiltered. + """ + statement = {"num": str(init) if init == end else {"range": [init, end]}} + if srv.fail_open: + statement["flags"] = ["bypass"] + return {"queue": statement} + + class FiregexFilter: def __init__(self, proto:str, port:int, ip_int:str, target:str, id:int): self.id = id @@ -96,7 +112,7 @@ class FiregexTables(NFTableManager): {'match': {'left': {'payload': {'protocol': ip_family(target_ip), 'field': 'saddr'}}, 'op': '==', 'right': nftables_int_to_json(target_ip)}}, {'match': {"left": { "payload": {"protocol": str(srv.proto), "field": "sport"}}, "op": "==", "right": int(target_port)}}, {"mangle": {"key": {"meta": {"key": "mark"}},"value": 0x1338}}, - {"queue": {"num": str(init) if init == end else {"range":[init, end] }, "flags": ["bypass"]}} + build_queue_statement(srv, init, end) ] }}}, {"insert":{"rule":{ @@ -107,7 +123,7 @@ class FiregexTables(NFTableManager): {'match': {'left': {'payload': {'protocol': ip_family(target_ip), 'field': 'daddr'}}, 'op': '==', 'right': nftables_int_to_json(target_ip)}}, {'match': {"left": { "payload": {"protocol": str(srv.proto), "field": "dport"}}, "op": "==", "right": int(target_port)}}, {"mangle": {"key": {"meta": {"key": "mark"}},"value": 0x1337}}, - {"queue": {"num": str(init) if init == end else {"range":[init, end] }, "flags": ["bypass"]}} + build_queue_statement(srv, init, end) ] }}} )