ad/firegex
1
0
Fork 0
mirror of https://github.com/Av0idD4rk/firegex.git synced 2026-10-23 14:31:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

catching exception on queue crashing on packet handling, implementation of a watchdog for queue processes

Browse Source
This commit is contained in:
Domingo Dirutigliano 2026-09-18 09:36:20 +02:00
parent 6b34de4371
commit f0821afe72
No known key found for this signature in database
13 changed files with 655 additions and 192 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
Dockerfile
View File

@ -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

249
backend/binsrc/classes/netfilter.cpp
View File

@ -1,5 +1,7 @@
#include <vector>
#include <thread>
#include <chrono>
#include <iostream>
#include <type_traits>
#include "../utils.cpp"
#include "nfqueue.cpp"
@ -7,111 +9,154 @@
#ifndef NETFILTER_CLASS_CPP
#define NETFILTER_CLASS_CPP
namespace Firegex {
namespace NfQueue {
template <typename Derived>
class ThreadNfQueue {
public:
ThreadNfQueue() = default;
virtual ~ThreadNfQueue() = default;
std::thread thr;
BlockingQueue<PktRequest<Derived>*> queue;
virtual void before_loop() {}
virtual void handle_next_packet(PktRequest<Derived>* pkt){}
void loop() {
static_cast<Derived*>(this)->before_loop();
PktRequest<Derived>* pkt;
for(;;) {
queue.take(pkt);
try {
static_cast<Derived*>(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 <typename Worker, typename = is_base_of<ThreadNfQueue<Worker>, Worker>>
void __real_handler(PktRequest<std::vector<Worker>>* pkt) {
const size_t idx = hash_stream_id(pkt->sid) % pkt->ctx->size();
auto* converted_pkt = reinterpret_cast<PktRequest<Worker>*>(pkt);
converted_pkt->ctx = &((*pkt->ctx)[idx]);
converted_pkt->ctx->queue.put(converted_pkt);
}
template <typename Worker, typename = is_base_of<ThreadNfQueue<Worker>, Worker>>
class MultiThreadQueue {
static_assert(std::is_base_of_v<ThreadNfQueue<Worker>, Worker>,
"Worker must inherit from ThreadNfQueue<Worker>");
private:
std::vector<Worker> workers;
NfQueue<std::vector<Worker>, __real_handler<Worker>> * 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<std::vector<Worker>, __real_handler<Worker>>(qnum);
queue_num_ = qnum;
break;
template <typename Derived>
class ThreadNfQueue
{
public:
ThreadNfQueue() = default;
virtual ~ThreadNfQueue() = default;
std::thread thr;
BlockingQueue<PktRequest<Derived> *> queue;
virtual void before_loop() {}
virtual void handle_next_packet(PktRequest<Derived> *pkt) {}
void loop()
{
static_cast<Derived *>(this)->before_loop();
PktRequest<Derived> *pkt;
for (;;)
{
queue.take(pkt);
try
{
static_cast<Derived *>(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<uint16_t>::max())
throw std::runtime_error("No available queue numbers");
void run_thread_loop()
{
thr = std::thread([this]()
{ this->loop(); });
}
};
template <typename Worker, typename = is_base_of<ThreadNfQueue<Worker>, Worker>>
void __real_handler(PktRequest<std::vector<Worker>> *pkt)
{
const size_t idx = hash_stream_id(pkt->sid) % pkt->ctx->size();
auto *converted_pkt = reinterpret_cast<PktRequest<Worker> *>(pkt);
converted_pkt->ctx = &((*pkt->ctx)[idx]);
converted_pkt->ctx->queue.put(converted_pkt);
}
template <typename Worker, typename = is_base_of<ThreadNfQueue<Worker>, Worker>>
class MultiThreadQueue
{
static_assert(std::is_base_of_v<ThreadNfQueue<Worker>, Worker>,
"Worker must inherit from ThreadNfQueue<Worker>");
private:
std::vector<Worker> workers;
NfQueue<std::vector<Worker>, __real_handler<Worker>> *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<std::vector<Worker>, __real_handler<Worker>>(qnum);
queue_num_ = qnum;
break;
}
catch (const std::invalid_argument &)
{
if (qnum == std::numeric_limits<uint16_t>::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

190
backend/binsrc/classes/nfqueue.cpp
View File

@ -8,12 +8,37 @@
#include <libmnl/libmnl.h>
#include <tins/tins.h>
#include <map>
#include <vector>
#include <iostream>
#include <cerrno>
#include <cstdlib>
#include <cstring>
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<Tins::TCP>();
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<Tins::TCP>();
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<char> 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<D>(
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<D>* pkt = nullptr;
try {
pkt = new PktRequest<D>(
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;
}
}
}
};

40
backend/binsrc/nfproxy.cpp
View File

@ -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();

18
backend/binsrc/nfregex.cpp
View File

@ -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();

45
backend/binsrc/regex/regex_rules.cpp
View File

@ -80,7 +80,7 @@ class RegexRules{
private:
static inline uint16_t glob_seq = 0;
uint16_t version;
uint16_t version = 0;
vector<pair<string, decoded_regex>> decoded_input_rules;
vector<pair<string, decoded_regex>> decoded_output_rules;
bool is_stream = true;
@ -104,7 +104,7 @@ class RegexRules{
vector<const char*> regex_match_rules(n_of_regex);
vector<unsigned int> regex_array_ids(n_of_regex);
vector<unsigned int> 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<string>(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<string> 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<string> 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<string>(), is_stream) {}
bool stream_mode(){
return is_stream;
@ -184,9 +187,7 @@ class RegexRules{
RegexRules(){
RegexRules(true);
}
RegexRules(): RegexRules(true) {}
~RegexRules(){
free_dbs();

65
backend/binsrc/utils.cpp
View File

@ -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<size_t>(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<size_t>(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::ostream& (*)(std::ostream&)>(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::ostream& (*)(std::ostream&)>(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::ostream& (*)(std::ostream&)>(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::ostream& (*)(std::ostream&)>(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<T> 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)

51
backend/modules/nfproxy/firegex.py
View File

@ -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()

50
backend/modules/nfproxy/firewall.py
View File

@ -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)

20
backend/modules/nfproxy/nftables.py
View File

@ -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)
]
}}}
)

47
backend/modules/nfregex/firegex.py
View File

@ -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()

48
backend/modules/nfregex/firewall.py
View File

@ -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)

20
backend/modules/nfregex/nftables.py
View File

@ -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)
]
}}}
)