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

Now on IPv6

Browse Source
This commit is contained in:
DomySh
2022-07-10 15:05:56 +02:00
parent 8cd1b69752
commit ada3a7212b
13 changed files with 208 additions and 163 deletions
Download Patch File Download Diff File Expand all files Collapse all files

107
backend/app.py
View File

@@ -1,5 +1,6 @@
from base64 import b64decode
import sqlite3, uvicorn, sys, secrets, re, os, asyncio, httpx, urllib, websockets
import sqlite3, uvicorn, sys, secrets, re, os, asyncio
import httpx, urllib, websockets
from typing import List, Union
from fastapi import FastAPI, HTTPException, WebSocket, Depends
from pydantic import BaseModel, BaseSettings
@@ -19,12 +20,12 @@ db = SQLite('db/firegex.db')
conf = KeyValueStorage(db)
firewall = ProxyManager(db)
class Settings(BaseSettings):
JWT_ALGORITHM: str = "HS256"
REACT_BUILD_DIR: str = "../frontend/build/" if not ON_DOCKER else "frontend/"
REACT_HTML_PATH: str = os.path.join(REACT_BUILD_DIR,"index.html")
VERSION = "1.3.0"
settings = Settings()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/login", auto_error=False)
@@ -158,8 +159,10 @@ async def get_general_stats(auth: bool = Depends(is_loggined)):
class ServiceModel(BaseModel):
status: str
service_id: str
port: int
name: str
ipv6: bool
n_regex: int
n_packets: int
@@ -167,66 +170,74 @@ class ServiceModel(BaseModel):
async def get_service_list(auth: bool = Depends(is_loggined)):
"""Get the list of existent firegex services"""
return db.query("""
SELECT
SELECT
s.service_id service_id,
s.status status,
s.port port,
s.name name,
s.ipv6 ipv6,
COUNT(r.regex_id) n_regex,
COALESCE(SUM(r.blocked_packets),0) n_packets
FROM services s LEFT JOIN regexes r ON r.service_port = s.port
GROUP BY s.port;
FROM services s LEFT JOIN regexes r
GROUP BY s.service_id;
""")
@app.get('/api/service/{service_port}', response_model=ServiceModel)
async def get_service_by_id(service_port: int, auth: bool = Depends(is_loggined)):
@app.get('/api/service/{service_id}', response_model=ServiceModel)
async def get_service_by_id(service_id: str, auth: bool = Depends(is_loggined)):
"""Get info about a specific service using his id"""
res = db.query("""
SELECT
s.service_id service_id,
s.status status,
s.port port,
s.name name,
s.ipv6 ipv6,
COUNT(r.regex_id) n_regex,
COALESCE(SUM(r.blocked_packets),0) n_packets
FROM services s LEFT JOIN regexes r ON r.service_port = s.port WHERE s.port = ?
GROUP BY s.port;
""", service_port)
FROM services s LEFT JOIN regexes r WHERE s.service_id = ?
GROUP BY s.service_id;
""", service_id)
if len(res) == 0: raise HTTPException(status_code=400, detail="This service does not exists!")
return res[0]
class StatusMessageModel(BaseModel):
status:str
@app.get('/api/service/{service_port}/stop', response_model=StatusMessageModel)
async def service_stop(service_port: int, auth: bool = Depends(is_loggined)):
@app.get('/api/service/{service_id}/stop', response_model=StatusMessageModel)
async def service_stop(service_id: str, auth: bool = Depends(is_loggined)):
"""Request the stop of a specific service"""
await firewall.get(service_port).next(STATUS.STOP)
await firewall.get(service_id).next(STATUS.STOP)
await refresh_frontend()
return {'status': 'ok'}
@app.get('/api/service/{service_port}/start', response_model=StatusMessageModel)
async def service_start(service_port: int, auth: bool = Depends(is_loggined)):
@app.get('/api/service/{service_id}/start', response_model=StatusMessageModel)
async def service_start(service_id: str, auth: bool = Depends(is_loggined)):
"""Request the start of a specific service"""
await firewall.get(service_port).next(STATUS.ACTIVE)
await firewall.get(service_id).next(STATUS.ACTIVE)
await refresh_frontend()
return {'status': 'ok'}
@app.get('/api/service/{service_port}/delete', response_model=StatusMessageModel)
async def service_delete(service_port: int, auth: bool = Depends(is_loggined)):
@app.get('/api/service/{service_id}/delete', response_model=StatusMessageModel)
async def service_delete(service_id: str, auth: bool = Depends(is_loggined)):
"""Request the deletion of a specific service"""
db.query('DELETE FROM services WHERE port = ?;', service_port)
db.query('DELETE FROM regexes WHERE service_port = ?;', service_port)
await firewall.remove(service_port)
db.query('DELETE FROM services WHERE service_id = ?;', service_id)
db.query('DELETE FROM regexes WHERE service_id = ?;', service_id)
await firewall.remove(service_id)
await refresh_frontend()
return {'status': 'ok'}
class RenameForm(BaseModel):
name:str
@app.post('/api/service/{service_port}/rename', response_model=StatusMessageModel)
async def service_rename(service_port: int, form: RenameForm, auth: bool = Depends(is_loggined)):
@app.post('/api/service/{service_id}/rename', response_model=StatusMessageModel)
async def service_rename(service_id: str, form: RenameForm, auth: bool = Depends(is_loggined)):
"""Request to change the name of a specific service"""
form.name = refactor_name(form.name)
if not form.name: return {'status': 'The name cannot be empty!'}
db.query('UPDATE services SET name=? WHERE port = ?;', form.name, service_port)
try:
db.query('UPDATE services SET name=? WHERE service_id = ?;', form.name, service_id)
except sqlite3.IntegrityError:
return {'status': 'This name is already used'}
await refresh_frontend()
return {'status': 'ok'}
@@ -234,28 +245,28 @@ class RegexModel(BaseModel):
regex:str
mode:str
id:int
service_port:int
service_id:str
is_blacklist: bool
n_packets:int
is_case_sensitive:bool
active:bool
@app.get('/api/service/{service_port}/regexes', response_model=List[RegexModel])
async def get_service_regexe_list(service_port: int, auth: bool = Depends(is_loggined)):
@app.get('/api/service/{service_id}/regexes', response_model=List[RegexModel])
async def get_service_regexe_list(service_id: str, auth: bool = Depends(is_loggined)):
"""Get the list of the regexes of a service"""
return db.query("""
SELECT
regex, mode, regex_id `id`, service_port, is_blacklist,
regex, mode, regex_id `id`, service_id, is_blacklist,
blocked_packets n_packets, is_case_sensitive, active
FROM regexes WHERE service_port = ?;
""", service_port)
FROM regexes WHERE service_id = ?;
""", service_id)
@app.get('/api/regex/{regex_id}', response_model=RegexModel)
async def get_regex_by_id(regex_id: int, auth: bool = Depends(is_loggined)):
"""Get regex info using his id"""
res = db.query("""
SELECT
regex, mode, regex_id `id`, service_port, is_blacklist,
regex, mode, regex_id `id`, service_id, is_blacklist,
blocked_packets n_packets, is_case_sensitive, active
FROM regexes WHERE `id` = ?;
""", regex_id)
@@ -268,7 +279,7 @@ async def regex_delete(regex_id: int, auth: bool = Depends(is_loggined)):
res = db.query('SELECT * FROM regexes WHERE regex_id = ?;', regex_id)
if len(res) != 0:
db.query('DELETE FROM regexes WHERE regex_id = ?;', regex_id)
await firewall.get(res[0]["service_port"]).update_filters()
await firewall.get(res[0]["service_id"]).update_filters()
await refresh_frontend()
return {'status': 'ok'}
@@ -279,7 +290,7 @@ async def regex_enable(regex_id: int, auth: bool = Depends(is_loggined)):
res = db.query('SELECT * FROM regexes WHERE regex_id = ?;', regex_id)
if len(res) != 0:
db.query('UPDATE regexes SET active=1 WHERE regex_id = ?;', regex_id)
await firewall.get(res[0]["service_port"]).update_filters()
await firewall.get(res[0]["service_id"]).update_filters()
await refresh_frontend()
return {'status': 'ok'}
@@ -289,12 +300,12 @@ async def regex_disable(regex_id: int, auth: bool = Depends(is_loggined)):
res = db.query('SELECT * FROM regexes WHERE regex_id = ?;', regex_id)
if len(res) != 0:
db.query('UPDATE regexes SET active=0 WHERE regex_id = ?;', regex_id)
await firewall.get(res[0]["service_port"]).update_filters()
await firewall.get(res[0]["service_id"]).update_filters()
await refresh_frontend()
return {'status': 'ok'}
class RegexAddForm(BaseModel):
service_port: int
service_id: str
regex: str
mode: str
active: Union[bool,None]
@@ -309,31 +320,39 @@ async def add_new_regex(form: RegexAddForm, auth: bool = Depends(is_loggined)):
except Exception:
return {"status":"Invalid regex"}
try:
db.query("INSERT INTO regexes (service_port, regex, is_blacklist, mode, is_case_sensitive, active ) VALUES (?, ?, ?, ?, ?, ?);",
form.service_port, form.regex, form.is_blacklist, form.mode, form.is_case_sensitive, True if form.active is None else form.active )
db.query("INSERT INTO regexes (service_id, regex, is_blacklist, mode, is_case_sensitive, active ) VALUES (?, ?, ?, ?, ?, ?);",
form.service_id, form.regex, form.is_blacklist, form.mode, form.is_case_sensitive, True if form.active is None else form.active )
except sqlite3.IntegrityError:
return {'status': 'An identical regex already exists'}
await firewall.get(form.service_port).update_filters()
await firewall.get(form.service_id).update_filters()
await refresh_frontend()
return {'status': 'ok'}
class ServiceAddForm(BaseModel):
name: str
port: int
ipv6: bool
@app.post('/api/services/add', response_model=StatusMessageModel)
class ServiceAddResponse(BaseModel):
status:str
service_id: Union[None,str]
@app.post('/api/services/add', response_model=ServiceAddResponse)
async def add_new_service(form: ServiceAddForm, auth: bool = Depends(is_loggined)):
"""Add a new service"""
import time
srv_id = None
try:
db.query("INSERT INTO services (name, port, status) VALUES (?, ?, ?)",
form.name, form.port, STATUS.STOP)
srv_id = str(form.port)+"::"+("ipv6" if form.ipv6 else "ipv4")
db.query("INSERT INTO services (service_id ,name, port, ipv6, status) VALUES (?, ?, ?, ?, ?)",
srv_id, refactor_name(form.name), form.port, form.ipv6, STATUS.STOP)
except sqlite3.IntegrityError:
return {'status': 'Name or/and ports of the service has been already assigned to another service'}
return {'status': 'Name or/and ports of the service has been already assigned'}
await firewall.reload()
init_t = time.time()
await refresh_frontend()
return {'status': 'ok'}
return {'status': 'ok', 'service_id': srv_id}
async def frontend_debug_proxy(path):
httpc = httpx.AsyncClient()

105
backend/proxy.py
View File

@@ -1,24 +1,25 @@
import multiprocessing
from threading import Thread
from typing import List
from netfilterqueue import NetfilterQueue
from multiprocessing import Manager, Process
from scapy.all import IP, TCP, UDP
from subprocess import Popen, PIPE
import os, traceback, pcre, re
QUEUE_BASE_NUM = 1000
def bind_queues(func, len_list=1):
def bind_queues(func, ipv6, len_list=1):
from scapy.all import IP, TCP, UDP, IPv6
if len_list <= 0: raise Exception("len must be >= 1")
queue_list = []
starts = QUEUE_BASE_NUM
end = starts
def func_wrap(pkt):
pkt_parsed = IP(pkt.get_payload())
pkt_parsed = IPv6(pkt.get_payload()) if ipv6 else IP(pkt.get_payload())
try:
if pkt_parsed[UDP if UDP in pkt_parsed else TCP].payload: func(pkt, pkt_parsed)
payload = None
if UDP in pkt_parsed: payload = pkt_parsed[UDP].payload
if TCP in pkt_parsed: payload = pkt_parsed[TCP].payload
if payload: func(pkt, pkt_parsed, bytes(payload))
else: pkt.accept()
except Exception:
traceback.print_exc()
@@ -52,15 +53,16 @@ class ProtoTypes:
class IPTables:
@staticmethod
def command(params):
def __init__(self, ipv6=False):
self.ipv6 = ipv6
def command(self, params):
if os.geteuid() != 0:
exit("You need to have root privileges to run this script.\nPlease try again, this time using 'sudo'. Exiting.")
return Popen(["iptables"]+params, stdout=PIPE, stderr=PIPE).communicate()
return Popen(["ip6tables"]+params if self.ipv6 else ["iptables"]+params, stdout=PIPE, stderr=PIPE).communicate()
@staticmethod
def list_filters(param):
stdout, strerr = IPTables.command(["-L", str(param), "--line-number", "-n"])
def list_filters(self, param):
stdout, strerr = self.command(["-L", str(param), "--line-number", "-n"])
output = [ele.split() for ele in stdout.decode().split("\n")]
return [{
"id": ele[0],
@@ -72,42 +74,35 @@ class IPTables:
"details": " ".join(ele[6:]) if len(ele) >= 7 else "",
} for ele in output if len(ele) >= 6 and ele[0].isnumeric()]
@staticmethod
def delete_command(param, id):
IPTables.command(["-R", str(param), str(id)])
def delete_command(self, param, id):
self.command(["-R", str(param), str(id)])
@staticmethod
def create_chain(name):
IPTables.command(["-N", str(name)])
def create_chain(self, name):
self.command(["-N", str(name)])
@staticmethod
def flush_chain(name):
IPTables.command(["-F", str(name)])
def flush_chain(self, name):
self.command(["-F", str(name)])
@staticmethod
def add_chain_to_input(name):
IPTables.command(["-I", "INPUT", "-j", str(name)])
def add_chain_to_input(self, name):
self.command(["-I", "INPUT", "-j", str(name)])
@staticmethod
def add_chain_to_output(name):
IPTables.command(["-I", "OUTPUT", "-j", str(name)])
def add_chain_to_output(self, name):
self.command(["-I", "OUTPUT", "-j", str(name)])
@staticmethod
def add_s_to_c(proto, port, queue_range):
def add_s_to_c(self, proto, port, queue_range):
init, end = queue_range
if init > end: init, end = end, init
IPTables.command([
self.command([
"-A", FilterTypes.OUTPUT, "-p", str(proto),
"--sport", str(port), "-j", "NFQUEUE",
"--queue-num" if init == end else "--queue-balance",
f"{init}" if init == end else f"{init}:{end}", "--queue-bypass"
])
@staticmethod
def add_c_to_s(proto, port, queue_range):
def add_c_to_s(self, proto, port, queue_range):
init, end = queue_range
if init > end: init, end = end, init
IPTables.command([
self.command([
"-A", FilterTypes.INPUT, "-p", str(proto),
"--dport", str(port), "-j", "NFQUEUE",
"--queue-num" if init == end else "--queue-balance",
@@ -115,41 +110,45 @@ class IPTables:
])
class FiregexFilter():
def __init__(self, type, number, queue, proto, port):
def __init__(self, type, number, queue, proto, port, ipv6):
self.type = type
self.id = int(number)
self.queue = queue
self.proto = proto
self.port = int(port)
self.iptable = IPTables(ipv6)
def __repr__(self) -> str:
return f"<FiregexFilter type={self.type} id={self.id} port={self.port} proto={self.proto} queue={self.queue}>"
def delete(self):
IPTables.delete_command(self.type, self.id)
self.iptable.delete_command(self.type, self.id)
class FiregexFilterManager:
def __init__(self):
IPTables.create_chain(FilterTypes.INPUT)
IPTables.create_chain(FilterTypes.OUTPUT)
def __init__(self, ipv6):
self.ipv6 = ipv6
self.iptables = IPTables(ipv6)
self.iptables.create_chain(FilterTypes.INPUT)
self.iptables.create_chain(FilterTypes.OUTPUT)
input_found = False
output_found = False
for filter in IPTables.list_filters("INPUT"):
for filter in self.iptables.list_filters("INPUT"):
if filter["target"] == FilterTypes.INPUT:
input_found = True
break
for filter in IPTables.list_filters("OUTPUT"):
for filter in self.iptables.list_filters("OUTPUT"):
if filter["target"] == FilterTypes.OUTPUT:
output_found = True
break
if not input_found: IPTables.add_chain_to_input(FilterTypes.INPUT)
if not output_found: IPTables.add_chain_to_output(FilterTypes.OUTPUT)
if not input_found: self.iptables.add_chain_to_input(FilterTypes.INPUT)
if not output_found: self.iptables.add_chain_to_output(FilterTypes.OUTPUT)
def get(self) -> List[FiregexFilter]:
res = []
for filter_type in [FilterTypes.INPUT, FilterTypes.OUTPUT]:
for filter in IPTables.list_filters(filter_type):
for filter in self.iptables.list_filters(filter_type):
queue_num = None
balanced = re.findall(r"NFQUEUE balance ([0-9]+):([0-9]+)", filter["details"])
numbered = re.findall(r"NFQUEUE num ([0-9]+)", filter["details"])
@@ -162,7 +161,8 @@ class FiregexFilterManager:
number=filter["id"],
queue=queue_num,
proto=filter["prot"],
port=int(port[0])
port=int(port[0]),
ipv6=self.ipv6
))
return res
@@ -170,18 +170,18 @@ class FiregexFilterManager:
for ele in self.get():
if int(port) == ele.port: return None
def c_to_s(pkt, data): return func(pkt, data, True)
def s_to_c(pkt, data): return func(pkt, data, False)
def c_to_s(pkt, data, payload): return func(pkt, data, payload, True)
def s_to_c(pkt, data, payload): return func(pkt, data, payload, False)
queues_c_to_s, codes = bind_queues(c_to_s, n_threads)
IPTables.add_c_to_s(proto, port, codes)
self.iptables.add_c_to_s(proto, port, codes)
queues_s_to_c, codes = bind_queues(s_to_c, n_threads)
IPTables.add_s_to_c(proto, port, codes)
self.iptables.add_s_to_c(proto, port, codes)
return queues_c_to_s + queues_s_to_c
def delete_all(self):
for filter_type in [FilterTypes.INPUT, FilterTypes.OUTPUT]:
IPTables.flush_chain(filter_type)
self.iptables.flush_chain(filter_type)
def delete_by_port(self, port):
for filter in self.get():
@@ -209,8 +209,8 @@ class Filter:
return True if self.compiled_regex.search(data) else False
class Proxy:
def __init__(self, port, filters=None):
self.manager = FiregexFilterManager()
def __init__(self, port, ipv6, filters=None):
self.manager = FiregexFilterManager(ipv6)
self.port = port
self.filters = Manager().list(filters) if filters else Manager().list([])
self.process = None
@@ -224,8 +224,7 @@ class Proxy:
def _starter(self):
self.manager.delete_by_port(self.port)
def regex_filter(pkt, data, by_client):
packet = bytes(data[TCP if TCP in data else UDP].payload)
def regex_filter(pkt, data, packet, by_client):
try:
for i, filter in enumerate(self.filters):
if (by_client and filter.c_to_s) or (not by_client and filter.s_to_c):

57
backend/utils.py
View File

@@ -49,27 +49,30 @@ class SQLite():
self.connect()
self.create_schema({
'services': {
'service_id': 'VARCHAR(100) PRIMARY KEY',
'status': 'VARCHAR(100) NOT NULL',
'port': 'INT NOT NULL CHECK(port > 0 and port < 65536) UNIQUE PRIMARY KEY',
'name': 'VARCHAR(100) NOT NULL'
'port': 'INT NOT NULL CHECK(port > 0 and port < 65536)',
'name': 'VARCHAR(100) NOT NULL UNIQUE',
'ipv6': 'BOOLEAN NOT NULL CHECK (ipv6 IN (0, 1)) DEFAULT 0',
},
'regexes': {
'regex': 'TEXT NOT NULL',
'mode': 'VARCHAR(1) NOT NULL',
'service_port': 'INT NOT NULL',
'service_id': 'VARCHAR(100) NOT NULL',
'is_blacklist': 'BOOLEAN NOT NULL CHECK (is_blacklist IN (0, 1))',
'blocked_packets': 'INTEGER UNSIGNED NOT NULL DEFAULT 0',
'regex_id': 'INTEGER PRIMARY KEY',
'is_case_sensitive' : 'BOOLEAN NOT NULL CHECK (is_case_sensitive IN (0, 1))',
'active' : 'BOOLEAN NOT NULL CHECK (is_case_sensitive IN (0, 1)) DEFAULT 1',
'FOREIGN KEY (service_port)':'REFERENCES services (port)',
'active' : 'BOOLEAN NOT NULL CHECK (active IN (0, 1)) DEFAULT 1',
'FOREIGN KEY (service_id)':'REFERENCES services (service_id)',
},
'keys_values': {
'key': 'VARCHAR(100) PRIMARY KEY',
'value': 'VARCHAR(100) NOT NULL',
},
})
self.query("CREATE UNIQUE INDEX IF NOT EXISTS unique_regex_service ON regexes (regex,service_port,is_blacklist,mode,is_case_sensitive);")
self.query("CREATE UNIQUE INDEX IF NOT EXISTS unique_services ON services (ipv6,port);")
self.query("CREATE UNIQUE INDEX IF NOT EXISTS unique_regex_service ON regexes (regex,service_id,is_blacklist,mode,is_case_sensitive);")
class KeyValueStorage:
def __init__(self, db):
@@ -95,10 +98,11 @@ class STATUS:
class ServiceNotFoundException(Exception): pass
class ServiceManager:
def __init__(self, port, db):
def __init__(self, id, port, ipv6, db):
self.id = id
self.port = port
self.db = db
self.proxy = Proxy(port)
self.proxy = Proxy(port, ipv6)
self.status = STATUS.STOP
self.filters = {}
self._update_filters_from_db()
@@ -110,8 +114,8 @@ class ServiceManager:
SELECT
regex, mode, regex_id `id`, is_blacklist,
blocked_packets n_packets, is_case_sensitive
FROM regexes WHERE service_port = ? AND active=1;
""", self.port)
FROM regexes WHERE service_id = ? AND active=1;
""", self.id)
#Filter check
old_filters = set(self.filters.keys())
@@ -137,7 +141,7 @@ class ServiceManager:
self.proxy.set_filters(self.filters.values())
def __update_status_db(self, status):
self.db.query("UPDATE services SET status = ? WHERE port = ?;", status, self.port)
self.db.query("UPDATE services SET status = ? WHERE service_id = ?;", status, self.id)
async def next(self,to):
async with self.lock:
@@ -190,11 +194,11 @@ class ProxyManager:
for key in list(self.proxy_table.keys()):
await self.remove(key)
async def remove(self,port):
async def remove(self,srv_id):
async with self.lock:
if port in self.proxy_table:
await self.proxy_table[port].next(STATUS.STOP)
del self.proxy_table[port]
if srv_id in self.proxy_table:
await self.proxy_table[srv_id].next(STATUS.STOP)
del self.proxy_table[srv_id]
async def init(self, callback = None):
self.init_updater(callback)
@@ -202,13 +206,13 @@ class ProxyManager:
async def reload(self):
async with self.lock:
for srv in self.db.query('SELECT port, status FROM services;'):
srv_port, req_status = srv["port"], srv["status"]
for srv in self.db.query('SELECT service_id, port, status, ipv6 FROM services;'):
srv_id, srv_port, req_status, srv_ipv6 = srv["service_id"], srv["port"], srv["status"], srv["ipv6"]
if srv_port in self.proxy_table:
continue
self.proxy_table[srv_port] = ServiceManager(srv_port,self.db)
await self.proxy_table[srv_port].next(req_status)
self.proxy_table[srv_id] = ServiceManager(srv_id, srv_port, srv_ipv6, self.db)
await self.proxy_table[srv_id].next(req_status)
async def _stats_updater(self, callback):
try:
@@ -226,10 +230,13 @@ class ProxyManager:
self.updater_task = None
return
def get(self,port):
if port in self.proxy_table:
return self.proxy_table[port]
def get(self,srv_id):
if srv_id in self.proxy_table:
return self.proxy_table[srv_id]
else:
raise ServiceNotFoundException()
raise ServiceNotFoundException()
def refactor_name(name:str):
name = name.strip()
while " " in name: name = name.replace(" "," ")
return name