0xb00b5/firegex-traffic-viewer
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

optional nfqueue fail-open option

Browse Source
This commit is contained in:
Domingo Dirutigliano
2025-02-18 17:36:15 +01:00
parent ece058d533
commit 59652fc697
11 changed files with 247 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -237,17 +237,15 @@ class NfQueue {
nlh = nfq_nlmsg_put(queue_msg_buffer, NFQNL_MSG_CONFIG, queue_num);
nfq_nlmsg_cfg_put_params(nlh, NFQNL_COPY_PACKET, 0xffff);
#ifdef NFQUEUE_FAIL_OPEN
char * enable_fail_open = getenv("FIREGEX_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
mnl_attr_put_u32(nlh, NFQA_CFG_FLAGS, htonl(NFQA_CFG_F_GSO));
mnl_attr_put_u32(nlh, NFQA_CFG_MASK, htonl(NFQA_CFG_F_GSO));
#endif
if (strcmp(enable_fail_open, "1") == 0){
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{
mnl_attr_put_u32(nlh, NFQA_CFG_FLAGS, htonl(NFQA_CFG_F_GSO));
mnl_attr_put_u32(nlh, NFQA_CFG_MASK, htonl(NFQA_CFG_F_GSO));
}
if (mnl_socket_sendto(nl, nlh, nlh->nlmsg_len) < 0) {
_clear();

9
backend/binsrc/nfregex.cpp
View File

@@ -9,10 +9,7 @@ using namespace Firegex::Regex;
using Firegex::NfQueue::MultiThreadQueue;
/*
Compile options:
NFQUEUE_FAIL_OPEN - enable fail-open option of nfqueueß
---
USE_PIPES_FOR_BLOKING_QUEUE - use pipes instead of conditional variable, queue and mutex for blocking queue
*/
@@ -63,14 +60,14 @@ int main(int argc, char *argv[]){
if (matchmode != nullptr && strcmp(matchmode, "block") == 0){
stream_mode = false;
}
bool fail_open = strcmp(getenv("FIREGEX_NFQUEUE_FAIL_OPEN"), "1") == 0;
regex_config.reset(new RegexRules(stream_mode));
MultiThreadQueue<RegexNfQueue> queue_manager(n_of_threads);
osyncstream(cout) << "QUEUE " << queue_manager.queue_num() << endl;
cerr << "[info] [main] Queue: " << queue_manager.queue_num() << " threads assigned: " << n_of_threads << " stream mode: " << stream_mode << endl;
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();

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

@@ -88,7 +88,11 @@ class FiregexInterceptor:
self.process = await asyncio.create_subprocess_exec(
proxy_binary_path,
stdout=asyncio.subprocess.PIPE, stdin=asyncio.subprocess.PIPE,
env={"MATCH_MODE": "stream" if self.srv.proto == "tcp" else "block", "NTHREADS": os.getenv("NTHREADS","1")},
env={
"MATCH_MODE": "stream" if self.srv.proto == "tcp" else "block",
"NTHREADS": os.getenv("NTHREADS","1"),
"FIREGEX_NFQUEUE_FAIL_OPEN": "1" if self.srv.fail_open else "0",
},
)
line_fut = self.process.stdout.readuntil()
try:

3
backend/modules/nfregex/models.py
View File

@@ -1,13 +1,14 @@
import base64
class Service:
def __init__(self, service_id: str, status: str, port: int, name: str, proto: str, ip_int: str, **other):
def __init__(self, service_id: str, status: str, port: int, name: str, proto: str, ip_int: str, fail_open: bool, **other):
self.id = service_id
self.status = status
self.port = port
self.name = name
self.proto = proto
self.ip_int = ip_int
self.fail_open = fail_open
@classmethod
def from_dict(cls, var: dict):

54
backend/routers/nfregex.py
View File

@@ -19,10 +19,17 @@ class ServiceModel(BaseModel):
ip_int: str
n_regex: int
n_packets: int
fail_open: bool
class RenameForm(BaseModel):
name:str
class SettingsForm(BaseModel):
port: PortType|None = None
proto: str|None = None
ip_int: str|None = None
fail_open: bool|None = None
class RegexModel(BaseModel):
regex:str
mode:str
@@ -44,6 +51,7 @@ class ServiceAddForm(BaseModel):
port: PortType
proto: str
ip_int: str
fail_open: bool = False
class ServiceAddResponse(BaseModel):
status:str
@@ -59,6 +67,7 @@ db = SQLite('db/nft-regex.db', {
'name': 'VARCHAR(100) NOT NULL UNIQUE',
'proto': 'VARCHAR(3) NOT NULL CHECK (proto IN ("tcp", "udp"))',
'ip_int': 'VARCHAR(100) NOT NULL',
'fail_open': 'BOOLEAN NOT NULL CHECK (fail_open IN (0, 1)) DEFAULT 1'
},
'regexes': {
'regex': 'TEXT NOT NULL',
@@ -128,6 +137,7 @@ async def get_service_list():
s.name name,
s.proto proto,
s.ip_int ip_int,
s.fail_open fail_open,
COUNT(r.regex_id) n_regex,
COALESCE(SUM(r.blocked_packets),0) n_packets
FROM services s LEFT JOIN regexes r ON s.service_id = r.service_id
@@ -145,6 +155,7 @@ async def get_service_by_id(service_id: str):
s.name name,
s.proto proto,
s.ip_int ip_int,
s.fail_open fail_open,
COUNT(r.regex_id) n_regex,
COALESCE(SUM(r.blocked_packets),0) n_packets
FROM services s LEFT JOIN regexes r ON s.service_id = r.service_id
@@ -190,6 +201,45 @@ async def service_rename(service_id: str, form: RenameForm):
await refresh_frontend()
return {'status': 'ok'}
@app.put('/services/{service_id}/settings', response_model=StatusMessageModel)
async def service_settings(service_id: str, form: SettingsForm):
"""Request to change the settings of a specific service (will cause a restart)"""
if form.proto is not None and form.proto not in ["tcp", "udp"]:
raise HTTPException(status_code=400, detail="Invalid protocol")
if form.port is not None and (form.port < 1 or form.port > 65535):
raise HTTPException(status_code=400, detail="Invalid port")
if form.ip_int is not None:
try:
form.ip_int = ip_parse(form.ip_int)
except ValueError:
raise HTTPException(status_code=400, detail="Invalid address")
keys = []
values = []
for key, value in form.model_dump(exclude_none=True).items():
keys.append(key)
values.append(value)
if len(keys) == 0:
raise HTTPException(status_code=400, detail="No settings to change provided")
try:
db.query(f'UPDATE services SET {", ".join([f"{key}=?" for key in keys])} WHERE service_id = ?;', *values, service_id)
except sqlite3.IntegrityError:
raise HTTPException(status_code=400, detail="A service with these settings already exists")
old_status = firewall.get(service_id).status
await firewall.remove(service_id)
await firewall.reload()
await firewall.get(service_id).next(old_status)
await refresh_frontend()
return {'status': 'ok'}
@app.get('/services/{service_id}/regexes', response_model=list[RegexModel])
async def get_service_regexe_list(service_id: str):
"""Get the list of the regexes of a service"""
@@ -275,8 +325,8 @@ async def add_new_service(form: ServiceAddForm):
srv_id = None
try:
srv_id = gen_service_id()
db.query("INSERT INTO services (service_id ,name, port, status, proto, ip_int) VALUES (?, ?, ?, ?, ?, ?)",
srv_id, refactor_name(form.name), form.port, STATUS.STOP, form.proto, form.ip_int)
db.query("INSERT INTO services (service_id ,name, port, status, proto, ip_int, fail_open) VALUES (?, ?, ?, ?, ?, ?, ?)",
srv_id, refactor_name(form.name), form.port, STATUS.STOP, form.proto, form.ip_int, form.fail_open)
except sqlite3.IntegrityError:
raise HTTPException(status_code=400, detail="This type of service already exists")
await firewall.reload()