using brotli 1.2 from pypi, fixed tests to py 3.14, removed from experimental pyproxy

Fix caching problems on fresh restart

Dockerfile

@@ -13,8 +13,8 @@ COPY ./frontend/ .
 RUN bun run build
 
 # Base fedora container
-FROM --platform=$TARGETARCH quay.io/fedora/fedora:42 AS base
+FROM --platform=$TARGETARCH quay.io/fedora/fedora:43 AS base
-RUN dnf -y update && dnf install -y python3.13 libnetfilter_queue \
+RUN dnf -y update && dnf install -y python3.14 libnetfilter_queue \
     libnfnetlink libmnl libcap-ng-utils nftables \
     vectorscan libtins python3-nftables libpcap && dnf clean all
 
@@ -23,13 +23,13 @@ WORKDIR /execute
 
 FROM --platform=$TARGETARCH base AS compiler
 
-RUN dnf -y update && dnf install -y python3.13-devel @development-tools gcc-c++ \
+RUN dnf -y update && dnf install -y python3.14-devel @development-tools gcc-c++ \
     libnetfilter_queue-devel libnfnetlink-devel libmnl-devel \
     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.13 -pthread -lnfnetlink $(pkg-config --cflags --libs libtins libmnl python3)
+RUN g++ binsrc/nfproxy.cpp -o cpproxy -std=c++23 -O3 -lnetfilter_queue -lpython3.14 -pthread -lnfnetlink $(pkg-config --cflags --libs libtins libmnl python3)
 
 #Building main conteiner
 FROM --platform=$TARGETARCH base AS final
@@ -37,10 +37,10 @@ FROM --platform=$TARGETARCH base AS final
 COPY ./backend/requirements.txt /execute/requirements.txt
 COPY ./fgex-lib /execute/fgex-lib
 
-RUN dnf -y update && dnf install -y gcc-c++ python3.13-devel uv git &&\
+RUN dnf -y update && dnf install -y gcc-c++ python3.14-devel uv git &&\
     uv pip install --no-cache --system ./fgex-lib &&\
     uv pip install --no-cache --system -r /execute/requirements.txt &&\
-    uv cache clean && dnf remove -y gcc-c++ python3.13-devel uv git && dnf clean all
+    uv cache clean && dnf remove -y gcc-c++ python3.14-devel uv git && dnf clean all
 
 COPY ./backend/ /execute/
 COPY --from=compiler /execute/cppregex /execute/cpproxy /execute/modules/

backend/app.py

@@ -227,7 +227,7 @@ if __name__ == '__main__':
     uvicorn.run(
         "app:app",
         # None allows to bind also on ipv6, and is selected if FIREGEX_HOST is any
-        host=None if FIREGEX_HOST == "any" else FIREGEX_HOST,
+        host="" if FIREGEX_HOST == "any" else FIREGEX_HOST,
         port=FIREGEX_PORT,
         uds=FIREGEX_SOCKET,
         reload=DEBUG and not NORELOAD,

backend/docker-entrypoint.sh

@@ -2,6 +2,13 @@
 
 chown nobody -R /execute/
 
+# Create socket directory if SOCKET_DIR is set
+if [ -n "$SOCKET_DIR" ]; then
+    mkdir -p "$SOCKET_DIR"
+    chown nobody:nobody "$SOCKET_DIR"
+    chmod 755 "$SOCKET_DIR"
+fi
+
 echo "[*] Attempting to start with capabilities..."
 
 if capsh --caps="cap_net_admin,cap_setpcap,cap_setuid,cap_setgid,cap_sys_nice+eip" \

backend/requirements.txt

@@ -4,5 +4,5 @@ uvicorn[standard]
 psutil
 python-jose[cryptography]
 python-socketio
-git+https://github.com/google/brotli.git@35d4992ac8eb1eca3b6c5f220e76cfc8b7e470aa
+brotli
 #git+https://salsa.debian.org/pkg-netfilter-team/pkg-nftables#egg=nftables&subdirectory=py

fgex-lib/firegex/nfproxy/models/http.py

@@ -10,7 +10,7 @@ from firegex.nfproxy.internals.exceptions import (
 from firegex.nfproxy.internals.models import FullStreamAction, ExceptionAction
 from dataclasses import dataclass, field
 from collections import deque
-from zstd import ZSTD_uncompress
+from compression import zstd
 import gzip
 import io
 import zlib
@@ -200,7 +200,7 @@ class InternalCallbackHandler:
                     break
             elif enc == "zstd":
                 try:
-                    decoding_body = ZSTD_uncompress(decoding_body)
+                    decoding_body = zstd.decompress(decoding_body)
                 except Exception as e:
                     print(f"Error decompressing zstd: {e}: skipping", flush=True)
                     decode_success = False

fgex-lib/requirements.txt

@@ -2,7 +2,6 @@ typer
 pydantic>=2
 typing-extensions>=4.7.1
 pycryptodome
-zstd
 watchfiles
 fgex
 websockets

frontend/src/App.tsx

@@ -13,7 +13,7 @@ import { Firewall } from './pages/Firewall';
 import { useQueryClient } from '@tanstack/react-query';
 import NFProxy from './pages/NFProxy';
 import ServiceDetailsNFProxy from './pages/NFProxy/ServiceDetails';
-import { useAuth } from './js/store';
+import { useAuthStore } from './js/store';
 
 function App() {
 
@@ -23,7 +23,7 @@ function App() {
   const [error, setError] = useState<string|null>()
   const [loadinBtn, setLoadingBtn] = useState(false);
   const queryClient = useQueryClient()
-  const { isAuthenticated, access_token } = useAuth()
+  const { access_token } = useAuthStore()
 
   useEffect(()=>{
     socketio.auth = { token: access_token || "" }
@@ -43,7 +43,7 @@ function App() {
     socketio.off("connect_error")
     socketio.disconnect()
   }
-},[isAuthenticated])
+  },[access_token])
 
   const getStatus = () =>{
     getstatus().then( res =>{
@@ -92,6 +92,7 @@ function App() {
         }
       }).catch( err => setError(err.toString()))
       setLoadingBtn(false)
+      form.reset()
     }
     
 
@@ -119,12 +120,14 @@ function App() {
         setLoadingBtn(true)
         await login(values).then(res => {
           if(!res){
+            queryClient.invalidateQueries()
             setSystemStatus({...systemStatus, loggined:true})
           }else{
             setError("Login failed")
           }
         }).catch( err => setError(err.toString()))
         setLoadingBtn(false)
+        form.reset()
       }
     
 

frontend/src/components/NavBar/index.tsx

@@ -39,12 +39,12 @@ export default function NavBar() {
             <NavBarButton navigate="nfregex" closeNav={closeNav} name="Netfilter Regex" color="grape" icon={<BsRegex size={19} />} />
             <NavBarButton navigate="firewall" closeNav={closeNav} name="Firewall Rules" color="red" icon={<PiWallLight size={19} />} />
             <NavBarButton navigate="porthijack" closeNav={closeNav} name="Hijack Port to Proxy" color="blue" icon={<GrDirections size={19} />} />
-            <Box px="xs" mt="lg">
+            <NavBarButton navigate="nfproxy" closeNav={closeNav} name="Netfilter Proxy" color="lime" icon={<TbPlugConnected size={19} />} />
+            {/*             <Box px="xs" mt="lg">
                 <Title order={5}>Experimental Features 🧪</Title>
             </Box>
             <Text></Text>
-            <Divider my="xs" />
+            <Divider my="xs" /> */}
-            <NavBarButton navigate="nfproxy" closeNav={closeNav} name="Netfilter Proxy" color="lime" icon={<TbPlugConnected size={19} />} />
         </Box>
 
     </AppShell.Navbar>

frontend/src/js/store.ts

@@ -42,21 +42,6 @@ export const useAuthStore = create<AuthState>()(
   )
 );
 
-// Hook personalizzati per un uso più facile nei componenti
-export const useAuth = () => {
-  const { access_token, setAccessToken, clearAccessToken, getAccessToken } = useAuthStore();
-  
-  const isAuthenticated = !!access_token;
-  
-  return {
-    access_token,
-    isAuthenticated,
-    setAccessToken,
-    clearAccessToken,
-    getAccessToken,
-  };
-};
-
 interface SessionState {
   home_section: string | null;
   setHomeSection: (section: string | null) => void;

run.py

@@ -255,8 +255,6 @@ def get_web_interface_url():
     if args.socket_dir:
         return os.path.join(args.socket_dir, "firegex.sock")
     
-    # Per altre piattaforme, usiamo l'host configurato se non è 0.0.0.0
-    # altrimenti usiamo localhost per evitare confusione
     display_host = "localhost" if args.host == "0.0.0.0" else args.host
     return f"http://{display_host}:{args.port}"
 
@@ -277,7 +275,7 @@ def write_compose(skip_password = True):
                             f"HOST={args.host}",
                             f"NTHREADS={args.threads}",
                             *([f"PSW_HASH_SET={hash_psw(psw_set)}"] if psw_set else []),
-                            *([f"SOCKET_DIR=/run/firegex"] if args.socket_dir else [])
+                            *(["SOCKET_DIR=/run/firegex"] if args.socket_dir else [])
                         ],
                         "volumes": [
                             "firegex_data:/execute/db",
@@ -325,7 +323,7 @@ def write_compose(skip_password = True):
                         "container_name": "firegex",
                         "build" if g.build else "image": "." if g.build else f"ghcr.io/pwnzer0tt1/firegex:{args.version}",
                         "ports": [
-                            f"{args.host}:{args.port}:{args.port}"
+                            f"{'' if args.host == 'any' else args.host+':'}{args.port}:{args.port}"
                         ],
                         "environment": [
                             f"PORT={args.port}",
@@ -600,6 +598,10 @@ def cleanup_standalone_mounts():
         f"{g.rootfs_path}/sys_host/net.ipv6.conf.all.forwarding"
     ]
     
+    # Add socket directory mount point if configured
+    if args.socket_dir:
+        mount_points.append(f"{g.rootfs_path}/run/firegex")
+    
     # Create umount commands (with || true to ignore errors)
     umount_commands = [f"umount -l {mount_point} || true" for mount_point in mount_points]
     
@@ -754,6 +756,18 @@ def setup_standalone_mounts():
         f"mount --bind /proc/sys/net/ipv6/conf/all/forwarding {g.rootfs_path}/sys_host/net.ipv6.conf.all.forwarding"
     ])
     
+    # Add socket directory bind mount if configured
+    if args.socket_dir:
+        # Create socket directory on host if it doesn't exist
+        # Create mount point in rootfs and bind mount the socket directory
+        privileged_commands.extend([
+            f"mkdir -p {args.socket_dir}",
+            f"chmod 755 {args.socket_dir}",
+            f"mkdir -p {g.rootfs_path}/run/firegex",
+            f"chmod 755 {g.rootfs_path}/run/firegex",
+            f"mount --bind {args.socket_dir} {g.rootfs_path}/run/firegex"
+        ])
+
     # Run all privileged commands in one batch
     if not run_privileged_commands(privileged_commands, "setup bind mounts"):
         puts("Failed to set up bind mounts", color=colors.red)
@@ -784,9 +798,9 @@ def run_standalone():
     if psw_set:
         env_vars.append(f"PSW_HASH_SET={hash_psw(psw_set)}")
     
-    # Add socket dir if set
+    # Add socket dir if set (use path inside chroot)
     if args.socket_dir:
-        env_vars.append(f"SOCKET_DIR={args.socket_dir}")
+        env_vars.append("SOCKET_DIR=/run/firegex")
     
     # Prepare environment string for chroot
     env_string = " ".join([f"{var}" for var in env_vars])

tests/api_test.py

@@ -4,9 +4,19 @@ from utils.firegexapi import FiregexAPI
 import argparse
 import secrets
 
+if __name__ == "__main__":
     parser = argparse.ArgumentParser()
-parser.add_argument("--address", "-a", type=str , required=False, help='Address of firegex backend', default="http://127.0.0.1:4444/")
+    parser.add_argument(
-parser.add_argument("--password", "-p", type=str, required=True, help='Firegex password')
+        "--address",
+        "-a",
+        type=str,
+        required=False,
+        help="Address of firegex backend",
+        default="http://127.0.0.1:4444/",
+    )
+    parser.add_argument(
+        "--password", "-p", type=str, required=True, help="Firegex password"
+    )
     args = parser.parse_args()
     sep()
     puts("Testing will start on ", color=colors.cyan, end="")
@@ -16,19 +26,22 @@ firegex = FiregexAPI(args.address)
 
     # Connect to Firegex
     if firegex.status()["status"] == "init":
-    if (firegex.set_password(args.password)):
+        if firegex.set_password(args.password):
             puts(f"Sucessfully set password to {args.password} ✔", color=colors.green)
         else:
-        puts("Test Failed: Unknown response or password already put ✗", color=colors.red)
+            puts(
+                "Test Failed: Unknown response or password already put ✗",
+                color=colors.red,
+            )
             exit(1)
     else:
-    if (firegex.login(args.password)):
+        if firegex.login(args.password):
             puts("Sucessfully logged in ✔", color=colors.green)
         else:
             puts("Test Failed: Unknown response or wrong passowrd ✗", color=colors.red)
             exit(1)
 
-if(firegex.status()["loggined"]):
+    if firegex.status()["loggined"]:
         puts("Correctly received status ✔", color=colors.green)
     else:
         puts("Test Failed: Unknown response or not logged in✗", color=colors.red)
@@ -36,60 +49,81 @@ else:
 
     # Prepare second instance
     firegex2 = FiregexAPI(args.address)
-if (firegex2.login(args.password)):
+    if firegex2.login(args.password):
         puts("Sucessfully logged in on second instance ✔", color=colors.green)
     else:
-    puts("Test Failed: Unknown response or wrong passowrd on second instance ✗", color=colors.red)
+        puts(
+            "Test Failed: Unknown response or wrong passowrd on second instance ✗",
+            color=colors.red,
+        )
         exit(1)
 
-if(firegex2.status()["loggined"]):
+    if firegex2.status()["loggined"]:
         puts("Correctly received status on second instance✔", color=colors.green)
     else:
-    puts("Test Failed: Unknown response or not logged in on second instance✗", color=colors.red)
+        puts(
+            "Test Failed: Unknown response or not logged in on second instance✗",
+            color=colors.red,
+        )
         exit(1)
 
     # Change password
     new_password = secrets.token_hex(10)
-if (firegex.change_password(new_password,expire=True)):
+    if firegex.change_password(new_password, expire=True):
         puts(f"Sucessfully changed password to {new_password} ✔", color=colors.green)
     else:
         puts("Test Failed: Coundl't change the password ✗", color=colors.red)
         exit(1)
 
     # Check if we are still logged in
-if(firegex.status()["loggined"]):
+    if firegex.status()["loggined"]:
         puts("Correctly received status after password change ✔", color=colors.green)
     else:
-    puts("Test Failed: Unknown response or not logged after password change ✗", color=colors.red)
+        puts(
+            "Test Failed: Unknown response or not logged after password change ✗",
+            color=colors.red,
+        )
         exit(1)
 
     # Check if second session expired and relog
 
-if(not firegex2.status()["loggined"]):
+    if not firegex2.status()["loggined"]:
         puts("Second instance was expired currectly ✔", color=colors.green)
     else:
-    puts("Test Failed: Still logged in on second instance, expire expected ✗", color=colors.red)
+        puts(
+            "Test Failed: Still logged in on second instance, expire expected ✗",
+            color=colors.red,
+        )
         exit(1)
-if (firegex2.login(new_password)):
+    if firegex2.login(new_password):
         puts("Sucessfully logged in on second instance ✔", color=colors.green)
     else:
-    puts("Test Failed: Unknown response or wrong passowrd on second instance ✗", color=colors.red)
+        puts(
+            "Test Failed: Unknown response or wrong passowrd on second instance ✗",
+            color=colors.red,
+        )
         exit(1)
 
     # Change it back
-if (firegex.change_password(args.password,expire=False)):
+    if firegex.change_password(args.password, expire=False):
         puts("Sucessfully restored the password ✔", color=colors.green)
     else:
         puts("Test Failed: Coundl't change the password ✗", color=colors.red)
         exit(1)
 
     # Check if we are still logged in
-if(firegex2.status()["loggined"]):
+    if firegex2.status()["loggined"]:
         puts("Correctly received status after password change ✔", color=colors.green)
     else:
-    puts("Test Failed: Unknown response or not logged after password change ✗", color=colors.red)
+        puts(
+            "Test Failed: Unknown response or not logged after password change ✗",
+            color=colors.red,
+        )
         exit(1)
 
     puts("List of available interfaces:", color=colors.yellow)
     for interface in firegex.get_interfaces():
-    puts("name: {}, address: {}".format(interface["name"], interface["addr"]), color=colors.yellow)
+        puts(
+            "name: {}, address: {}".format(interface["name"], interface["addr"]),
+            color=colors.yellow,
+        )

tests/nfproxy_test.py

@@ -5,7 +5,9 @@ from utils.tcpserver import TcpServer
 import argparse
 import secrets
 import time
+import textwrap
 
+if __name__ == "__main__":
     parser = argparse.ArgumentParser()
     parser.add_argument(
         "--address",
@@ -15,7 +17,9 @@ parser.add_argument(
         help="Address of firegex backend",
         default="http://127.0.0.1:4444/",
     )
-parser.add_argument("--password", "-p", type=str, required=True, help="Firegex password")
+    parser.add_argument(
+        "--password", "-p", type=str, required=True, help="Firegex password"
+    )
     parser.add_argument(
         "--service_name",
         "-n",
@@ -32,7 +36,9 @@ parser.add_argument(
         help="Port of the test service",
         default=1337,
     )
-parser.add_argument("--ipv6", "-6", action="store_true", help="Test Ipv6", default=False)
+    parser.add_argument(
+        "--ipv6", "-6", action="store_true", help="Test Ipv6", default=False
+    )
     parser.add_argument(
         "--verbose", "-V", action="store_true", help="Verbose output", default=False
     )
@@ -68,19 +74,15 @@ else:
         puts("Test Failed: Failed to create service ✗", color=colors.red)
         exit(1)
 
-
     def exit_test(code):
         if service_id:
             server.stop()
-        """
             if firegex.nfproxy_delete_service(service_id):
                 puts("Sucessfully deleted service ✔", color=colors.green)
             else:
                 puts("Test Failed: Coulnd't delete serivce ✗", color=colors.red)
-        """
         exit(code)
 
-
     if firegex.nfproxy_start_service(service_id):
         puts("Sucessfully started service ✔", color=colors.green)
     else:
@@ -91,7 +93,9 @@ server.start()
     time.sleep(0.5)
     try:
         if server.sendCheckData(secrets.token_bytes(432)):
-        puts("Successfully tested first proxy with no filters ✔", color=colors.green)
+            puts(
+                "Successfully tested first proxy with no filters ✔", color=colors.green
+            )
         else:
             puts("Test Failed: Data was corrupted ", color=colors.red)
             exit_test(1)
@@ -99,7 +103,7 @@ except Exception:
         puts("Test Failed: Couldn't send data to the server ", color=colors.red)
         exit_test(1)
 
-BASE_FILTER_VERDICT_TEST = """
+    BASE_FILTER_VERDICT_TEST = textwrap.dedent("""
     from firegex.nfproxy.models import RawPacket
     from firegex.nfproxy import pyfilter, ACCEPT, UNSTABLE_MANGLE, DROP, REJECT
 
@@ -108,11 +112,10 @@ def verdict_test(packet:RawPacket):
         if b"%%TEST%%" in packet.data:
             packet.l4_data = packet.l4_data.replace(b"%%TEST%%", b"%%MANGLE%%")
             return %%ACTION%%
-"""
+    """)
 
     BASE_FILTER_VERDICT_NAME = "verdict_test"
 
-
     def get_vedict_test(to_match: str, action: str, mangle_to: str = "REDACTED"):
         return (
             BASE_FILTER_VERDICT_TEST.replace("%%TEST%%", to_match)
@@ -120,12 +123,10 @@ def get_vedict_test(to_match: str, action: str, mangle_to: str = "REDACTED"):
             .replace("%%MANGLE%%", mangle_to)
         )
 
-
     # Check if filter is present in the service
     n_blocked = 0
     n_mangled = 0
 
-
     def checkFilter(match_bytes, filter_name, should_work=True, mangle_with=None):
         if mangle_with:
             if should_work:
@@ -176,7 +177,8 @@ def checkFilter(match_bytes, filter_name, should_work=True, mangle_with=None):
                                 exit_test(1)
                         else:
                             puts(
-                            "Test Failed: The request wasn't mangled ✗", color=colors.red
+                                "Test Failed: The request wasn't mangled ✗",
+                                color=colors.red,
                             )
                             exit_test(1)
                         server.close_client()
@@ -200,7 +202,9 @@ def checkFilter(match_bytes, filter_name, should_work=True, mangle_with=None):
                     if r["name"] == filter_name:
                         # Test the filter
                         if not server.sendCheckData(
-                        secrets.token_bytes(40) + match_bytes + secrets.token_bytes(40)
+                            secrets.token_bytes(40)
+                            + match_bytes
+                            + secrets.token_bytes(40)
                         ):
                             puts(
                                 "The malicious request was successfully blocked ✔",
@@ -226,7 +230,8 @@ def checkFilter(match_bytes, filter_name, should_work=True, mangle_with=None):
                                 exit_test(1)
                         else:
                             puts(
-                            "Test Failed: The request wasn't blocked ✗", color=colors.red
+                                "Test Failed: The request wasn't blocked ✗",
+                                color=colors.red,
                             )
                             exit_test(1)
                         return
@@ -244,7 +249,6 @@ def checkFilter(match_bytes, filter_name, should_work=True, mangle_with=None):
                     )
                     exit_test(1)
 
-
     # Add new filter
     secret = bytes(secrets.token_hex(16).encode())
 
@@ -280,7 +284,10 @@ checkFilter(secret, BASE_FILTER_VERDICT_NAME)
 
     # Disable filter
     if firegex.nfproxy_disable_pyfilter(service_id, BASE_FILTER_VERDICT_NAME):
-    puts(f"Sucessfully disabled filter {BASE_FILTER_VERDICT_NAME} ✔", color=colors.green)
+        puts(
+            f"Sucessfully disabled filter {BASE_FILTER_VERDICT_NAME} ✔",
+            color=colors.green,
+        )
     else:
         puts("Test Failed: Coulnd't disable the filter ✗", color=colors.red)
         exit_test(1)
@@ -290,25 +297,27 @@ checkFilter(secret, BASE_FILTER_VERDICT_NAME, should_work=False)
 
     # Enable filter
     if firegex.nfproxy_enable_pyfilter(service_id, BASE_FILTER_VERDICT_NAME):
-    puts(f"Sucessfully enabled filter {BASE_FILTER_VERDICT_NAME} ✔", color=colors.green)
+        puts(
+            f"Sucessfully enabled filter {BASE_FILTER_VERDICT_NAME} ✔",
+            color=colors.green,
+        )
     else:
         puts("Test Failed: Coulnd't enable the regex ✗", color=colors.red)
         exit_test(1)
 
     checkFilter(secret, BASE_FILTER_VERDICT_NAME)
 
-
     def remove_filters():
         global n_blocked, n_mangled
         server.stop()
         server.start()
+        time.sleep(0.5)  # Wait for server to fully start
         if not firegex.nfproxy_set_code(service_id, ""):
             puts("Test Failed: Couldn't remove the filter ✗", color=colors.red)
             exit_test(1)
         n_blocked = 0
         n_mangled = 0
 
-
     remove_filters()
 
     # Check if it's actually deleted
@@ -317,7 +326,8 @@ checkFilter(secret, BASE_FILTER_VERDICT_NAME, should_work=False)
     # Check if DROP works
     if firegex.nfproxy_set_code(service_id, get_vedict_test(secret.decode(), "DROP")):
         puts(
-        f"Sucessfully added filter for {str(secret)} in DROP mode ✔", color=colors.green
+            f"Sucessfully added filter for {str(secret)} in DROP mode ✔",
+            color=colors.green,
         )
     else:
         puts(f"Test Failed: Couldn't add the filter {str(secret)} ✗", color=colors.red)
@@ -364,7 +374,7 @@ checkFilter(secret, BASE_FILTER_VERDICT_NAME, mangle_with=mangle_result)
     remove_filters()
 
     secret = b"8331ee1bf75893dd7fa3d34f29bac7fc8935aa3ef6c565fe8b395ef7f485"
-TCP_INPUT_STREAM_TEST = f"""
+    TCP_INPUT_STREAM_TEST = textwrap.dedent(f"""
     from firegex.nfproxy.models import TCPInputStream
     from firegex.nfproxy import pyfilter, ACCEPT, UNSTABLE_MANGLE, DROP, REJECT
 
@@ -373,7 +383,7 @@ def data_type_test(packet:TCPInputStream):
         if {repr(secret)} in packet.data:
             return REJECT
 
-"""
+    """)
 
     if firegex.nfproxy_set_code(service_id, TCP_INPUT_STREAM_TEST):
         puts(
@@ -403,7 +413,7 @@ server.close_client()
     remove_filters()
 
     secret = b"8331ee1bf75893dd7fa3d34f29bac7fc8935aa3ef6c565fe8b395ef7f485"
-TCP_OUTPUT_STREAM_TEST = f"""
+    TCP_OUTPUT_STREAM_TEST = textwrap.dedent(f"""
     from firegex.nfproxy.models import TCPOutputStream
     from firegex.nfproxy import pyfilter, ACCEPT, UNSTABLE_MANGLE, DROP, REJECT
 
@@ -412,7 +422,7 @@ def data_type_test(packet:TCPOutputStream):
         if {repr(secret)} in packet.data:
             return REJECT
 
-"""
+    """)
 
     if firegex.nfproxy_set_code(service_id, TCP_OUTPUT_STREAM_TEST):
         puts(
@@ -443,21 +453,21 @@ remove_filters()
 
     secret = b"8331ee1bf75893dd7fa3d34f29bac7fc8935aa3ef6c565fe8b395ef7f485"
 
-REQUEST_HEADER_TEST = f"""POST / HTTP/1.1
+    REQUEST_HEADER_TEST = textwrap.dedent(f"""POST / HTTP/1.1
     Host: localhost
     X-TeSt: {secret.decode()}
     Content-Length: 15
 
-A Friendly Body""".replace("\n", "\r\n")
+    A Friendly Body""").replace("\n", "\r\n")
 
-REQUEST_BODY_TEST = f"""POST / HTTP/1.1
+    REQUEST_BODY_TEST = textwrap.dedent(f"""POST / HTTP/1.1
     Host: localhost
     X-TeSt: NotTheSecret
     Content-Length: {len(secret.decode())}
 
-{secret.decode()}""".replace("\n", "\r\n")
+    {secret.decode()}""").replace("\n", "\r\n")
 
-HTTP_REQUEST_STREAM_TEST = f"""
+    HTTP_REQUEST_STREAM_TEST = textwrap.dedent(f"""
     from firegex.nfproxy.models import HttpRequest
     from firegex.nfproxy import pyfilter, ACCEPT, UNSTABLE_MANGLE, DROP, REJECT
 
@@ -469,7 +479,7 @@ def data_type_test(req:HttpRequest):
             if {repr(secret)} in req.body:
                 return REJECT
 
-"""
+    """)
 
     if firegex.nfproxy_set_code(service_id, HTTP_REQUEST_STREAM_TEST):
         puts(
@@ -512,7 +522,7 @@ server.close_client()
 
     remove_filters()
 
-HTTP_FULL_REQUEST_STREAM_TEST = f"""
+    HTTP_FULL_REQUEST_STREAM_TEST = textwrap.dedent(f"""
     from firegex.nfproxy.models import HttpFullRequest
     from firegex.nfproxy import pyfilter, ACCEPT, UNSTABLE_MANGLE, DROP, REJECT
 
@@ -527,7 +537,7 @@ def data_type_test(req:HttpFullRequest):
         if {repr(secret)} in req.body:
             return REJECT
 
-"""
+    """)
 
     if firegex.nfproxy_set_code(service_id, HTTP_FULL_REQUEST_STREAM_TEST):
         puts(
@@ -570,8 +580,7 @@ server.close_client()
 
     remove_filters()
 
-
+    HTTP_REQUEST_HEADER_STREAM_TEST = textwrap.dedent(f"""
-HTTP_REQUEST_HEADER_STREAM_TEST = f"""
     from firegex.nfproxy.models import HttpRequestHeader
     from firegex.nfproxy import pyfilter, ACCEPT, UNSTABLE_MANGLE, DROP, REJECT
 
@@ -580,7 +589,7 @@ def data_type_test(req:HttpRequestHeader):
         if {repr(secret.decode())} in req.get_header("x-test"):
             return REJECT
 
-"""
+    """)
 
     if firegex.nfproxy_set_code(service_id, HTTP_REQUEST_HEADER_STREAM_TEST):
         puts(
@@ -610,21 +619,21 @@ remove_filters()
 
     secret = b"8331ee1bf75893dd7fa3d34f29bac7fc8935aa3ef6c565fe8b395ef7f485"
 
-RESPONSE_HEADER_TEST = f"""HTTP/1.1 200 OK
+    RESPONSE_HEADER_TEST = textwrap.dedent(f"""HTTP/1.1 200 OK
     Host: localhost
     X-TeSt: {secret.decode()}
     Content-Length: 15
 
-A Friendly Body""".replace("\n", "\r\n")
+    A Friendly Body""").replace("\n", "\r\n")
 
-RESPONSE_BODY_TEST = f"""HTTP/1.1 200 OK
+    RESPONSE_BODY_TEST = textwrap.dedent(f"""HTTP/1.1 200 OK
     Host: localhost
     X-TeSt: NotTheSecret
     Content-Length: {len(secret.decode())}
 
-{secret.decode()}""".replace("\n", "\r\n")
+    {secret.decode()}""").replace("\n", "\r\n")
 
-HTTP_RESPONSE_STREAM_TEST = f"""
+    HTTP_RESPONSE_STREAM_TEST = textwrap.dedent(f"""
     from firegex.nfproxy.models import HttpResponse
     from firegex.nfproxy import pyfilter, ACCEPT, UNSTABLE_MANGLE, DROP, REJECT
 
@@ -636,7 +645,7 @@ def data_type_test(req:HttpResponse):
             if {repr(secret)} in req.body:
                 return REJECT
 
-"""
+    """)
 
     if firegex.nfproxy_set_code(service_id, HTTP_RESPONSE_STREAM_TEST):
         puts(
@@ -679,8 +688,7 @@ server.close_client()
 
     remove_filters()
 
-
+    HTTP_FULL_RESPONSE_STREAM_TEST = textwrap.dedent(f"""
-HTTP_FULL_RESPONSE_STREAM_TEST = f"""
     from firegex.nfproxy.models import HttpFullResponse
     from firegex.nfproxy import pyfilter, ACCEPT, UNSTABLE_MANGLE, DROP, REJECT
 
@@ -695,7 +703,7 @@ def data_type_test(req:HttpFullResponse):
         if {repr(secret)} in req.body:
             return REJECT
 
-"""
+    """)
 
     if firegex.nfproxy_set_code(service_id, HTTP_FULL_RESPONSE_STREAM_TEST):
         puts(
@@ -738,8 +746,7 @@ server.close_client()
 
     remove_filters()
 
-
+    HTTP_RESPONSE_HEADER_STREAM_TEST = textwrap.dedent(f"""
-HTTP_RESPONSE_HEADER_STREAM_TEST = f"""
     from firegex.nfproxy.models import HttpResponseHeader
     from firegex.nfproxy import pyfilter, ACCEPT, UNSTABLE_MANGLE, DROP, REJECT
 
@@ -748,7 +755,7 @@ def data_type_test(req:HttpResponseHeader):
         if {repr(secret.decode())} in req.get_header("x-test"):
             return REJECT
 
-"""
+    """)
 
     if firegex.nfproxy_set_code(service_id, HTTP_RESPONSE_HEADER_STREAM_TEST):
         puts(
@@ -781,7 +788,7 @@ remove_filters()
     WS_REQUEST_PARSING_TEST = b"GET /sock/?EIO=4&transport=websocket HTTP/1.1\r\nHost: localhost:8080\r\nConnection: Upgrade\r\nPragma: no-cache\r\nCache-Control: no-cache\r\nUser-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)\xac AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36\r\nUpgrade: websocket\r\nOrigin: http://localhost:8080\r\nSec-WebSocket-Version: 13\r\nAccept-Encoding: gzip, deflate, br, zstd\r\nAccept-Language: it-IT,it;q=0.9,en-US;q=0.8,en;q=0.7,zh-CN;q=0.6,zh;q=0.5\r\nCookie: cookie-consent=true; _iub_cs-86405163=%7B%22timestamp%22%3A%222024-09-12T18%3A20%3A18.627Z%22%2C%22version%22%3A%221.65.1%22%2C%22purposes%22%3A%7B%221%22%3Atrue%2C%224%22%3Atrue%7D%2C%22id%22%3A86405163%2C%22cons%22%3A%7B%22rand%22%3A%222b09e6%22%7D%7D\r\nSec-WebSocket-Key: eE01O3/ZShPKsrykACLAaA==\r\nSec-WebSocket-Extensions: permessage-deflate; client_max_window_bits\r\n\r\n\xc1\x84#\x8a\xb2\xbb\x11\xbb\xb2\xbb"
     WS_RESPONSE_PARSING_TEST = b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: eGnJqUSoSKE3wOfKD2M3G82RsS8=\r\nSec-WebSocket-Extensions: permessage-deflate\r\ndate: Sat, 15 Mar 2025 12:04:19 GMT\r\nserver: uvicorn\r\n\r\n\xc1_2\xa8V*\xceLQ\xb2Rr1\xb4\xc8\xf6r\x0c\xf3\xaf\xd25\xf7\x8e\xf4\xb3LsttrW\xd2Q*-H/JLI-V\xb2\x8a\x8e\xd5Q*\xc8\xccK\x0f\xc9\xccM\xcd/-Q\xb222\x00\x02\x88\x98g^IjQYb\x0eP\xd0\x14,\x98\x9bX\x11\x90X\x99\x93\x9f\x084\xda\xd0\x00\x0cj\x01\x00\xc1\x1b21\x80\xd9e\xe1n\x19\x9e\xe3RP\x9a[Z\x99\x93j\xea\x15\x00\xb4\xcbC\xa9\x16\x00"
 
-HTTP_REQUEST_WS_PARSING_TEST = """
+    HTTP_REQUEST_WS_PARSING_TEST = textwrap.dedent("""
     from firegex.nfproxy.models import HttpRequest, HttpResponse
     from firegex.nfproxy import pyfilter, ACCEPT, UNSTABLE_MANGLE, DROP, REJECT
 
@@ -793,7 +800,7 @@ def data_type_test(req:HttpRequest):
     def data_type_test(req:HttpResponse):
         print(req)
 
-"""
+    """)
 
     if firegex.nfproxy_set_code(service_id, HTTP_REQUEST_WS_PARSING_TEST):
         puts(
@@ -801,7 +808,9 @@ if firegex.nfproxy_set_code(service_id, HTTP_REQUEST_WS_PARSING_TEST):
             color=colors.green,
         )
     else:
-    puts("Test Failed: Couldn't add the websocket parsing filter ✗", color=colors.red)
+        puts(
+            "Test Failed: Couldn't add the websocket parsing filter ✗", color=colors.red
+        )
         exit_test(1)
 
     server.connect_client()
@@ -823,7 +832,9 @@ remove_filters()
 
     # Rename service
     if firegex.nfproxy_rename_service(service_id, f"{args.service_name}2"):
-    puts(f"Sucessfully renamed service to {args.service_name}2 ✔", color=colors.green)
+        puts(
+            f"Sucessfully renamed service to {args.service_name}2 ✔", color=colors.green
+        )
     else:
         puts("Test Failed: Coulnd't rename service ✗", color=colors.red)
         exit_test(1)
@@ -838,7 +849,9 @@ else:
 
     # Rename back service
     if firegex.nfproxy_rename_service(service_id, f"{args.service_name}"):
-    puts(f"Sucessfully renamed service to {args.service_name} ✔", color=colors.green)
+        puts(
+            f"Sucessfully renamed service to {args.service_name} ✔", color=colors.green
+        )
     else:
         puts("Test Failed: Coulnd't rename service ✗", color=colors.red)
         exit_test(1)
@@ -850,13 +863,15 @@ if firegex.nfproxy_settings_service(
         srv_updated = firegex.nfproxy_get_service(service_id)
         if (
             srv_updated["port"] == 1338
-        and ("::dead:beef" if args.ipv6 else "123.123.123.123") in srv_updated["ip_int"]
+            and ("::dead:beef" if args.ipv6 else "123.123.123.123")
+            in srv_updated["ip_int"]
             and srv_updated["fail_open"]
         ):
             puts("Sucessfully changed service settings ✔", color=colors.green)
         else:
             puts(
-            "Test Failed: Service settings weren't updated correctly ✗", color=colors.red
+                "Test Failed: Service settings weren't updated correctly ✗",
+                color=colors.red,
             )
             exit_test(1)
     else:

tests/nfregex_test.py

@@ -8,13 +8,47 @@ import secrets
 import base64
 import time
 
+if __name__ == "__main__":
     parser = argparse.ArgumentParser()
-parser.add_argument("--address", "-a", type=str , required=False, help='Address of firegex backend', default="http://127.0.0.1:4444/")
+    parser.add_argument(
-parser.add_argument("--password", "-p", type=str, required=True, help='Firegex password')
+        "--address",
-parser.add_argument("--service_name", "-n", type=str , required=False, help='Name of the test service', default="Test Service")
+        "-a",
-parser.add_argument("--port", "-P", type=int , required=False, help='Port of the test service', default=1337)
+        type=str,
-parser.add_argument("--ipv6", "-6" , action="store_true", help='Test Ipv6', default=False)
+        required=False,
-parser.add_argument("--proto", "-m" , type=str, required=False, choices=["tcp","udp"], help='Select the protocol', default="tcp")
+        help="Address of firegex backend",
+        default="http://127.0.0.1:4444/",
+    )
+    parser.add_argument(
+        "--password", "-p", type=str, required=True, help="Firegex password"
+    )
+    parser.add_argument(
+        "--service_name",
+        "-n",
+        type=str,
+        required=False,
+        help="Name of the test service",
+        default="Test Service",
+    )
+    parser.add_argument(
+        "--port",
+        "-P",
+        type=int,
+        required=False,
+        help="Port of the test service",
+        default=1337,
+    )
+    parser.add_argument(
+        "--ipv6", "-6", action="store_true", help="Test Ipv6", default=False
+    )
+    parser.add_argument(
+        "--proto",
+        "-m",
+        type=str,
+        required=False,
+        choices=["tcp", "udp"],
+        help="Select the protocol",
+        default="tcp",
+    )
 
     args = parser.parse_args()
     sep()
@@ -24,19 +58,21 @@ puts(f"{args.address}", color=colors.yellow)
     firegex = FiregexAPI(args.address)
 
     # Login
-if (firegex.login(args.password)):
+    if firegex.login(args.password):
         puts("Sucessfully logged in ✔", color=colors.green)
     else:
         puts("Test Failed: Unknown response or wrong passowrd ✗", color=colors.red)
         exit(1)
 
     # Create server
-server = (TcpServer if args.proto == "tcp" else UdpServer)(args.port,ipv6=args.ipv6)
+    server = (TcpServer if args.proto == "tcp" else UdpServer)(
+        args.port, ipv6=args.ipv6
+    )
 
     def exit_test(code):
         if service_id:
             server.stop()
-        if(firegex.nfregex_delete_service(service_id)):
+            if firegex.nfregex_delete_service(service_id):
                 puts("Sucessfully deleted service ✔", color=colors.green)
             else:
                 puts("Test Failed: Coulnd't delete serivce ✗", color=colors.red)
@@ -45,17 +81,19 @@ def exit_test(code):
 
     srvs = firegex.nfregex_get_services()
     for ele in srvs:
-    if ele['name'] == args.service_name:
+        if ele["name"] == args.service_name:
-        firegex.nfregex_delete_service(ele['service_id'])
+            firegex.nfregex_delete_service(ele["service_id"])
 
-service_id = firegex.nfregex_add_service(args.service_name, args.port, args.proto , "::1" if args.ipv6 else "127.0.0.1" )
+    service_id = firegex.nfregex_add_service(
+        args.service_name, args.port, args.proto, "::1" if args.ipv6 else "127.0.0.1"
+    )
     if service_id:
         puts(f"Sucessfully created service {service_id} ✔", color=colors.green)
     else:
         puts("Test Failed: Failed to create service ✗", color=colors.red)
         exit(1)
 
-if(firegex.nfregex_start_service(service_id)):
+    if firegex.nfregex_start_service(service_id):
         puts("Sucessfully started service ✔", color=colors.green)
     else:
         puts("Test Failed: Failed to start service ✗", color=colors.red)
@@ -75,13 +113,14 @@ except Exception:
     # Add new regex
     secret = bytes(secrets.token_hex(16).encode())
 
-if firegex.nfregex_add_regex(service_id,secret,"B",active=True,is_case_sensitive=True): 
+    if firegex.nfregex_add_regex(
+        service_id, secret, "B", active=True, is_case_sensitive=True
+    ):
         puts(f"Sucessfully added regex {str(secret)} ✔", color=colors.green)
     else:
         puts(f"Test Failed: Couldn't add the regex {str(secret)} ✗", color=colors.red)
         exit_test(1)
 
-
     # Check if regex is present in the service
     n_blocked = 0
 
@@ -97,42 +136,84 @@ def checkRegex(regex, should_work=True, upper=False, deleted=False):
                 if r["regex"] == secret:
                     # Test the regex
                     s = regex.upper() if upper else regex
-                if not server.sendCheckData(secrets.token_bytes(40) + s +  secrets.token_bytes(40)):
+                    if not server.sendCheckData(
-                    puts("The malicious request was successfully blocked ✔", color=colors.green)
+                        secrets.token_bytes(40) + s + secrets.token_bytes(40)
+                    ):
+                        puts(
+                            "The malicious request was successfully blocked ✔",
+                            color=colors.green,
+                        )
                         n_blocked += 1
                         time.sleep(1)
                         if firegex.nfregex_get_regex(r["id"])["n_packets"] == n_blocked:
-                        puts("The packet was reported as blocked in the API ✔", color=colors.green)
+                            puts(
+                                "The packet was reported as blocked in the API ✔",
+                                color=colors.green,
+                            )
                         else:
-                        puts("Test Failed: The packet wasn't reported as blocked in the API ✗", color=colors.red)
+                            puts(
+                                "Test Failed: The packet wasn't reported as blocked in the API ✗",
+                                color=colors.red,
+                            )
                             exit_test(1)
-                    if getMetric("firegex_blocked_packets", secret.decode()) == n_blocked:
+                        if (
-                        puts("The packet was reported as blocked in the metrics ✔", color=colors.green)
+                            getMetric("firegex_blocked_packets", secret.decode())
+                            == n_blocked
+                        ):
+                            puts(
+                                "The packet was reported as blocked in the metrics ✔",
+                                color=colors.green,
+                            )
                         else:
-                        puts("Test Failed: The packet wasn't reported as blocked in the metrics ✗", color=colors.red)
+                            puts(
+                                "Test Failed: The packet wasn't reported as blocked in the metrics ✗",
+                                color=colors.red,
+                            )
                             exit_test(1)
                         if getMetric("firegex_active", secret.decode()) == 1:
-                        puts("The regex was reported as active in the metrics ✔", color=colors.green)
+                            puts(
+                                "The regex was reported as active in the metrics ✔",
+                                color=colors.green,
+                            )
                         else:
-                        puts("Test Failed: The regex wasn't reported as active in the metrics ✗", color=colors.red)
+                            puts(
+                                "Test Failed: The regex wasn't reported as active in the metrics ✗",
+                                color=colors.red,
+                            )
                             exit_test(1)
                     else:
-                    puts("Test Failed: The request wasn't blocked ✗", color=colors.red)
+                        puts(
+                            "Test Failed: The request wasn't blocked ✗",
+                            color=colors.red,
+                        )
                         exit_test(1)
                     return
             puts("Test Failed: The regex wasn't found ✗", color=colors.red)
             exit_test(1)
         else:
-        if server.sendCheckData(secrets.token_bytes(40) + base64.b64decode(regex) +  secrets.token_bytes(40)):
+            if server.sendCheckData(
+                secrets.token_bytes(40)
+                + base64.b64decode(regex)
+                + secrets.token_bytes(40)
+            ):
                 puts("The request wasn't blocked ✔", color=colors.green)
             else:
-            puts("Test Failed: The request was blocked when it shouldn't have", color=colors.red)
+                puts(
+                    "Test Failed: The request was blocked when it shouldn't have",
+                    color=colors.red,
+                )
                 exit_test(1)
             if not deleted:
                 if getMetric("firegex_active", secret.decode()) == 0:
-                puts("The regex was reported as inactive in the metrics ✔", color=colors.green)
+                    puts(
+                        "The regex was reported as inactive in the metrics ✔",
+                        color=colors.green,
+                    )
                 else:
-                puts("Test Failed: The regex wasn't reported as inactive in the metrics ✗", color=colors.red)
+                    puts(
+                        "Test Failed: The regex wasn't reported as inactive in the metrics ✗",
+                        color=colors.red,
+                    )
                     exit_test(1)
 
     def clear_regexes():
@@ -140,8 +221,11 @@ def clear_regexes():
         n_blocked = 0
         for r in firegex.nfregex_get_service_regexes(service_id):
             if r["regex"] == secret:
-            if(firegex.nfregex_delete_regex(r["id"])): 
+                if firegex.nfregex_delete_regex(r["id"]):
-                puts(f"Sucessfully deleted regex with id {r['id']} ✔", color=colors.green)
+                    puts(
+                        f"Sucessfully deleted regex with id {r['id']} ✔",
+                        color=colors.green,
+                    )
                 else:
                     puts("Test Failed: Coulnd't delete the regex ✗", color=colors.red)
                     exit_test(1)
@@ -149,13 +233,16 @@ def clear_regexes():
         if f'regex="{secret.decode()}"' not in firegex.nfregex_get_metrics():
             puts("No regex metrics after deletion ✔", color=colors.green)
         else:
-        puts("Test Failed: Metrics found after deleting the regex ✗", color=colors.red)
+            puts(
+                "Test Failed: Metrics found after deleting the regex ✗",
+                color=colors.red,
+            )
             exit_test(1)
 
     checkRegex(secret)
 
     # Pause the proxy
-if(firegex.nfregex_stop_service(service_id)):
+    if firegex.nfregex_stop_service(service_id):
         puts(f"Sucessfully paused service with id {service_id} ✔", color=colors.green)
     else:
         puts("Test Failed: Coulnd't pause the service ✗", color=colors.red)
@@ -165,7 +252,7 @@ else:
     checkRegex(secret, should_work=False)
 
     # Start firewall
-if(firegex.nfregex_start_service(service_id)):
+    if firegex.nfregex_start_service(service_id):
         puts(f"Sucessfully started service with id {service_id} ✔", color=colors.green)
     else:
         puts("Test Failed: Coulnd't start the service ✗", color=colors.red)
@@ -176,8 +263,11 @@ checkRegex(secret)
     # Disable regex
     for r in firegex.nfregex_get_service_regexes(service_id):
         if r["regex"] == secret:
-        if(firegex.nfregex_disable_regex(r["id"])): 
+            if firegex.nfregex_disable_regex(r["id"]):
-            puts(f"Sucessfully disabled regex with id {r['id']} ✔", color=colors.green)
+                puts(
+                    f"Sucessfully disabled regex with id {r['id']} ✔",
+                    color=colors.green,
+                )
             else:
                 puts("Test Failed: Coulnd't disable the regex ✗", color=colors.red)
                 exit_test(1)
@@ -189,8 +279,10 @@ checkRegex(secret,should_work=False)
     # Enable regex
     for r in firegex.nfregex_get_service_regexes(service_id):
         if r["regex"] == secret:
-        if(firegex.nfregex_enable_regex(r["id"])): 
+            if firegex.nfregex_enable_regex(r["id"]):
-            puts(f"Sucessfully enabled regex with id {r['id']} ✔", color=colors.green)
+                puts(
+                    f"Sucessfully enabled regex with id {r['id']} ✔", color=colors.green
+                )
             else:
                 puts("Test Failed: Coulnd't enable the regex ✗", color=colors.red)
                 exit_test(1)
@@ -205,10 +297,18 @@ clear_regexes()
     checkRegex(secret, should_work=False, deleted=True)
 
     # Add case insensitive regex
-if(firegex.nfregex_add_regex(service_id,secret,"B",active=True, is_case_sensitive=False)): 
+    if firegex.nfregex_add_regex(
-    puts(f"Sucessfully added case insensitive regex {str(secret)} ✔", color=colors.green)
+        service_id, secret, "B", active=True, is_case_sensitive=False
+    ):
+        puts(
+            f"Sucessfully added case insensitive regex {str(secret)} ✔",
+            color=colors.green,
+        )
     else:
-    puts(f"Test Failed: Coulnd't add the case insensitive regex {str(secret)} ✗", color=colors.red)
+        puts(
+            f"Test Failed: Coulnd't add the case insensitive regex {str(secret)} ✗",
+            color=colors.red,
+        )
         exit_test(1)
 
     checkRegex(secret, upper=True)
@@ -217,8 +317,10 @@ checkRegex(secret)
     clear_regexes()
 
     # Rename service
-if(firegex.nfregex_rename_service(service_id,f"{args.service_name}2")):
+    if firegex.nfregex_rename_service(service_id, f"{args.service_name}2"):
-    puts(f"Sucessfully renamed service to {args.service_name}2 ✔", color=colors.green)
+        puts(
+            f"Sucessfully renamed service to {args.service_name}2 ✔", color=colors.green
+        )
     else:
         puts("Test Failed: Coulnd't rename service ✗", color=colors.red)
         exit_test(1)
@@ -232,20 +334,37 @@ else:
         exit_test(1)
 
     # Rename back service
-if(firegex.nfregex_rename_service(service_id,f"{args.service_name}")):
+    if firegex.nfregex_rename_service(service_id, f"{args.service_name}"):
-    puts(f"Sucessfully renamed service to {args.service_name} ✔", color=colors.green)
+        puts(
+            f"Sucessfully renamed service to {args.service_name} ✔", color=colors.green
+        )
     else:
         puts("Test Failed: Coulnd't rename service ✗", color=colors.red)
         exit_test(1)
 
     # Change settings
     opposite_proto = "udp" if args.proto == "tcp" else "tcp"
-if(firegex.nfregex_settings_service(service_id, 1338, opposite_proto, "::dead:beef" if args.ipv6 else "123.123.123.123", True)):
+    if firegex.nfregex_settings_service(
+        service_id,
+        1338,
+        opposite_proto,
+        "::dead:beef" if args.ipv6 else "123.123.123.123",
+        True,
+    ):
         srv_updated = firegex.nfregex_get_service(service_id)
-    if srv_updated["port"] == 1338 and srv_updated["proto"] == opposite_proto and ("::dead:beef" if args.ipv6 else "123.123.123.123") in srv_updated["ip_int"] and srv_updated["fail_open"]:
+        if (
+            srv_updated["port"] == 1338
+            and srv_updated["proto"] == opposite_proto
+            and ("::dead:beef" if args.ipv6 else "123.123.123.123")
+            in srv_updated["ip_int"]
+            and srv_updated["fail_open"]
+        ):
             puts("Sucessfully changed service settings ✔", color=colors.green)
         else:
-        puts("Test Failed: Service settings weren't updated correctly ✗", color=colors.red)
+            puts(
+                "Test Failed: Service settings weren't updated correctly ✗",
+                color=colors.red,
+            )
             exit_test(1)
     else:
         puts("Test Failed: Coulnd't change service settings ✗", color=colors.red)

tests/ph_test.py

@@ -7,13 +7,47 @@ import argparse
 import secrets
 import time
 
+if __name__ == "__main__":
     parser = argparse.ArgumentParser()
-parser.add_argument("--address", "-a", type=str , required=False, help='Address of firegex backend', default="http://127.0.0.1:4444/")
+    parser.add_argument(
-parser.add_argument("--password", "-p", type=str, required=True, help='Firegex password')
+        "--address",
-parser.add_argument("--service_name", "-n", type=str , required=False, help='Name of the test service', default="Test Service")
+        "-a",
-parser.add_argument("--port", "-P", type=int , required=False, help='Port of the test service', default=1337)
+        type=str,
-parser.add_argument("--ipv6", "-6" , action="store_true", help='Test Ipv6', default=False)
+        required=False,
-parser.add_argument("--proto", "-m" , type=str, required=False, choices=["tcp","udp"], help='Select the protocol', default="tcp")
+        help="Address of firegex backend",
+        default="http://127.0.0.1:4444/",
+    )
+    parser.add_argument(
+        "--password", "-p", type=str, required=True, help="Firegex password"
+    )
+    parser.add_argument(
+        "--service_name",
+        "-n",
+        type=str,
+        required=False,
+        help="Name of the test service",
+        default="Test Service",
+    )
+    parser.add_argument(
+        "--port",
+        "-P",
+        type=int,
+        required=False,
+        help="Port of the test service",
+        default=1337,
+    )
+    parser.add_argument(
+        "--ipv6", "-6", action="store_true", help="Test Ipv6", default=False
+    )
+    parser.add_argument(
+        "--proto",
+        "-m",
+        type=str,
+        required=False,
+        choices=["tcp", "udp"],
+        help="Select the protocol",
+        default="tcp",
+    )
 
     args = parser.parse_args()
     sep()
@@ -23,19 +57,21 @@ puts(f"{args.address}", color=colors.yellow)
     firegex = FiregexAPI(args.address)
 
     # Login
-if (firegex.login(args.password)):
+    if firegex.login(args.password):
         puts("Sucessfully logged in ✔", color=colors.green)
     else:
         puts("Test Failed: Unknown response or wrong passowrd ✗", color=colors.red)
         exit(1)
 
     # Create server
-server = (TcpServer if args.proto == "tcp" else UdpServer)(args.port+1,ipv6=args.ipv6,proxy_port=args.port)
+    server = (TcpServer if args.proto == "tcp" else UdpServer)(
+        args.port + 1, ipv6=args.ipv6, proxy_port=args.port
+    )
 
     def exit_test(code):
         if service_id:
             server.stop()
-        if(firegex.ph_delete_service(service_id)):
+            if firegex.ph_delete_service(service_id):
                 puts("Sucessfully deleted service ✔", color=colors.green)
             else:
                 puts("Test Failed: Coulnd't delete serivce ✗", color=colors.red)
@@ -44,18 +80,25 @@ def exit_test(code):
 
     srvs = firegex.ph_get_services()
     for ele in srvs:
-    if ele['name'] == args.service_name:
+        if ele["name"] == args.service_name:
-        firegex.ph_delete_service(ele['service_id'])
+            firegex.ph_delete_service(ele["service_id"])
 
     # Create and start serivce
-service_id = firegex.ph_add_service(args.service_name, args.port, args.port+1, args.proto , "::1" if args.ipv6 else "127.0.0.1",  "::1" if args.ipv6 else "127.0.0.1")
+    service_id = firegex.ph_add_service(
+        args.service_name,
+        args.port,
+        args.port + 1,
+        args.proto,
+        "::1" if args.ipv6 else "127.0.0.1",
+        "::1" if args.ipv6 else "127.0.0.1",
+    )
     if service_id:
         puts(f"Sucessfully created service {service_id} ✔", color=colors.green)
     else:
         puts("Test Failed: Failed to create service ✗", color=colors.red)
         exit(1)
 
-if(firegex.ph_start_service(service_id)):
+    if firegex.ph_start_service(service_id):
         puts("Sucessfully started service ✔", color=colors.green)
     else:
         puts("Test Failed: Failed to start service ✗", color=colors.red)
@@ -87,7 +130,7 @@ def checkData(should_work):
     checkData(True)
 
     # Pause the proxy
-if(firegex.ph_stop_service(service_id)):
+    if firegex.ph_stop_service(service_id):
         puts(f"Sucessfully paused service with id {service_id} ✔", color=colors.green)
     else:
         puts("Test Failed: Coulnd't pause the service ✗", color=colors.red)
@@ -96,7 +139,7 @@ else:
     checkData(False)
 
     # Start firewall
-if(firegex.ph_start_service(service_id)):
+    if firegex.ph_start_service(service_id):
         puts(f"Sucessfully started service with id {service_id} ✔", color=colors.green)
     else:
         puts("Test Failed: Coulnd't start the service ✗", color=colors.red)
@@ -105,7 +148,9 @@ else:
     checkData(True)
 
     # Change port
-if(firegex.ph_change_destination(service_id, "::1" if args.ipv6 else "127.0.0.1", args.port+2)): 
+    if firegex.ph_change_destination(
+        service_id, "::1" if args.ipv6 else "127.0.0.1", args.port + 2
+    ):
         puts("Sucessfully changed port ✔", color=colors.green)
     else:
         puts("Test Failed: Coulnd't change destination ✗", color=colors.red)
@@ -114,15 +159,19 @@ else:
     checkData(False)
 
     server.stop()
-server = (TcpServer if args.proto == "tcp" else UdpServer)(args.port+2,ipv6=args.ipv6,proxy_port=args.port)
+    server = (TcpServer if args.proto == "tcp" else UdpServer)(
+        args.port + 2, ipv6=args.ipv6, proxy_port=args.port
+    )
     server.start()
     time.sleep(0.5)
 
     checkData(True)
 
     # Rename service
-if(firegex.ph_rename_service(service_id,f"{args.service_name}2")):
+    if firegex.ph_rename_service(service_id, f"{args.service_name}2"):
-    puts(f"Sucessfully renamed service to {args.service_name}2 ✔", color=colors.green)
+        puts(
+            f"Sucessfully renamed service to {args.service_name}2 ✔", color=colors.green
+        )
     else:
         puts("Test Failed: Coulnd't rename service ✗", color=colors.red)
         exit_test(1)

tests/run_tests.sh

@@ -8,7 +8,7 @@ ERROR=0
 
 pip3 install -r requirements.txt
 
-until curl --output /dev/null --silent --fail http://localhost:4444/api/status; do
+until curl --output /dev/null --silent --fail http://127.0.0.1:4444/api/status; do
     printf '.'
     sleep 1
 done

tests/utils/tcpserver.py

@@ -1,7 +1,44 @@
+import queue
 from multiprocessing import Process, Queue
 import socket
 import traceback
 
+
+def _start_tcp_server(port, server_queue: Queue, ipv6, verbose):
+    sock = socket.socket(
+        socket.AF_INET6 if ipv6 else socket.AF_INET, socket.SOCK_STREAM
+    )
+    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    sock.bind(("::1" if ipv6 else "127.0.0.1", port))
+    sock.listen(8)
+    while True:
+        connection, address = sock.accept()
+        while True:
+            try:
+                buf = connection.recv(4096)
+                if buf == b"":
+                    break
+
+                reply = buf  # Default to echo
+                try:
+                    # See if there is a custom reply, but don't block
+                    custom_reply = server_queue.get(block=False)
+                    reply = custom_reply
+                except queue.Empty:
+                    pass  # No custom reply, just echo
+
+                if verbose:
+                    print("SERVER: ", reply)
+                connection.sendall(reply)
+            except (ConnectionResetError, BrokenPipeError):
+                break  # Client closed connection
+            except Exception:
+                if verbose:
+                    traceback.print_exc()
+                break  # Exit on other errors
+        connection.close()
+
+
 class TcpServer:
     def __init__(self, port, ipv6, proxy_port=None, verbose=False):
         self.proxy_port = proxy_port
@@ -12,30 +49,10 @@ class TcpServer:
         self._regen_process()
 
     def _regen_process(self):
-        def _startServer(port, server_queue:Queue):
+        self.server = Process(
-            sock = socket.socket(socket.AF_INET6 if self.ipv6 else socket.AF_INET, socket.SOCK_STREAM)
+            target=_start_tcp_server,
-            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+            args=[self.port, self._server_data_queue, self.ipv6, self.verbose],
-            sock.bind(('::1' if self.ipv6 else '127.0.0.1', port))
+        )
-            sock.listen(8)
-            while True:
-                connection,address = sock.accept()
-                while True:
-                    try:
-                        buf = connection.recv(4096)
-                        if buf == b'':
-                            break
-                        try:
-                            buf = server_queue.get(block=False)
-                        except Exception:
-                            pass
-                        if self.verbose:
-                            print("SERVER: ", buf)
-                        connection.sendall(buf)
-                    except Exception:
-                        if self.verbose:
-                            traceback.print_exc()
-                connection.close()
-        self.server = Process(target=_startServer,args=[self.port, self._server_data_queue])
 
     def start(self):
         self.server.start()
@@ -46,9 +63,16 @@ class TcpServer:
         self._regen_process()
 
     def connect_client(self):
-        self.client_sock = socket.socket(socket.AF_INET6 if self.ipv6 else socket.AF_INET, socket.SOCK_STREAM)
+        self.client_sock = socket.socket(
+            socket.AF_INET6 if self.ipv6 else socket.AF_INET, socket.SOCK_STREAM
+        )
         self.client_sock.settimeout(1)
-        self.client_sock.connect(('::1' if self.ipv6 else '127.0.0.1', self.proxy_port if self.proxy_port else self.port))      
+        self.client_sock.connect(
+            (
+                "::1" if self.ipv6 else "127.0.0.1",
+                self.proxy_port if self.proxy_port else self.port,
+            )
+        )
 
     def close_client(self):
         if self.client_sock:

tests/utils/udpserver.py

@@ -1,35 +1,94 @@
-from multiprocessing import Process
+from multiprocessing import Process, Queue
 import socket
+import queue
+import traceback
 
-class UdpServer:
+
-    def __init__(self,port,ipv6, proxy_port = None):
+def _start_udp_server(port, server_queue: Queue, ipv6, verbose):
-        def _startServer(port):
     sock = socket.socket(socket.AF_INET6 if ipv6 else socket.AF_INET, socket.SOCK_DGRAM)
     sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-            sock.bind(('::1' if ipv6 else '127.0.0.1', port))
+    sock.bind(("::1" if ipv6 else "127.0.0.1", port))
     while True:
-                bytesAddressPair = sock.recvfrom(432)
+        try:
+            bytesAddressPair = sock.recvfrom(4096)
             message = bytesAddressPair[0]
             address = bytesAddressPair[1]
-                sock.sendto(message, address)
 
-        self.ipv6 = ipv6
+            reply = message  # Default to echo
+            try:
+                # See if there is a custom reply, but don't block
+                custom_reply = server_queue.get(block=False)
+                reply = custom_reply
+            except queue.Empty:
+                pass  # No custom reply, just echo
+
+            if verbose:
+                print(f"SERVER: sending {reply} to {address}")
+            sock.sendto(reply, address)
+        except Exception:
+            if verbose:
+                traceback.print_exc()
+
+
+class UdpServer:
+    def __init__(self, port, ipv6, proxy_port=None, verbose=False):
         self.port = port
+        self.ipv6 = ipv6
         self.proxy_port = proxy_port
-        self.server = Process(target=_startServer,args=[port])
+        self.verbose = verbose
+        self._server_data_queue = Queue()
+        self._regen_process()
+
+    def _regen_process(self):
+        self.server = Process(
+            target=_start_udp_server,
+            args=[self.port, self._server_data_queue, self.ipv6, self.verbose],
+        )
 
     def start(self):
         self.server.start()
 
     def stop(self):
         self.server.terminate()
+        self.server.join()
+        self._regen_process()
 
-    def sendCheckData(self,data):
+    def connect_client(self):
-        s = socket.socket(socket.AF_INET6 if self.ipv6 else socket.AF_INET, socket.SOCK_DGRAM)
+        self.client_sock = socket.socket(
-        s.settimeout(2)
+            socket.AF_INET6 if self.ipv6 else socket.AF_INET, socket.SOCK_DGRAM
-        s.sendto(data, ('::1' if self.ipv6 else '127.0.0.1', self.proxy_port if self.proxy_port else self.port))
+        )
+        self.client_sock.settimeout(1)
+        self.client_sock.connect(
+            (
+                "::1" if self.ipv6 else "127.0.0.1",
+                self.proxy_port if self.proxy_port else self.port,
+            )
+        )
+
+    def close_client(self):
+        if self.client_sock:
+            self.client_sock.close()
+
+    def send_packet(self, packet, server_reply=None):
+        if self.verbose:
+            print("CLIENT: ", packet)
+        if server_reply:
+            self._server_data_queue.put(server_reply)
+        self.client_sock.sendall(packet)
+
+    def recv_packet(self):
         try:
-            received_data = s.recvfrom(432)
+            return self.client_sock.recv(4096)
-        except Exception:
+        except (TimeoutError, ConnectionResetError):
+            if self.verbose:
+                traceback.print_exc()
             return False
-        return received_data[0] == data
+
+    def sendCheckData(self, data, get_data=False):
+        self.connect_client()
+        self.send_packet(data)
+        received_data = self.recv_packet()
+        self.close_client()
+        if get_data:
+            return received_data
+        return received_data == data