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

@@ -9,14 +9,14 @@ import { getMainPath } from "../../js/utils";
 import { BsRegex } from "react-icons/bs";
 
 function NavBarButton({ navigate, closeNav, name, icon, color, disabled, onClick }:
-    { navigate?: string, closeNav: () => void, name:string, icon:any, color:MantineColor, disabled?:boolean, onClick?:CallableFunction }) {
+    { navigate?: string, closeNav: () => void, name: string, icon: any, color: MantineColor, disabled?: boolean, onClick?: CallableFunction }) {
     const navigator = useNavigate()
     return <UnstyledButton
-        className={`firegex__navbar__unstyled_button${navigate==getMainPath()?" selected":""}${disabled?" disabled":""}`}
+        className={`firegex__navbar__unstyled_button${navigate == getMainPath() ? " selected" : ""}${disabled ? " disabled" : ""}`}
-        onClick={()=>{
+        onClick={() => {
-        if(navigate){navigator(`/${navigate}`);closeNav()}
+            if (navigate) { navigator(`/${navigate}`); closeNav() }
-        if (onClick) onClick()
+            if (onClick) onClick()
-    }} disabled={disabled}>
+        }} disabled={disabled}>
         <Group>
             <ThemeIcon color={color} variant="light">
                 {icon}
@@ -24,7 +24,7 @@ function NavBarButton({ navigate, closeNav, name, icon, color, disabled, onClick
             <Text size="sm">{name}</Text>
         </Group>
     </UnstyledButton>
-} 
+}
 
 export default function NavBar() {
     const [toggle, setToggleState] = useState(false);
@@ -35,17 +35,17 @@ export default function NavBar() {
             <Title order={4}>Options ⚙️</Title>
         </Box>
         <Divider my="xs" />
-        <Box style={{flexGrow: 1}} component={ScrollArea} px="xs" mt="xs">
+        <Box style={{ flexGrow: 1 }} component={ScrollArea} px="xs" mt="xs">
             <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

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

tests/nfregex_test.py

@@ -8,247 +8,366 @@ import secrets
 import base64
 import time
 
-parser = argparse.ArgumentParser()
+if __name__ == "__main__":
-parser.add_argument("--address", "-a", type=str , required=False, help='Address of firegex backend', default="http://127.0.0.1:4444/")
+    parser = argparse.ArgumentParser()
-parser.add_argument("--password", "-p", type=str, required=True, help='Firegex password')
+    parser.add_argument(
-parser.add_argument("--service_name", "-n", type=str , required=False, help='Name of the test service', default="Test Service")
+        "--address",
-parser.add_argument("--port", "-P", type=int , required=False, help='Port of the test service', default=1337)
+        "-a",
-parser.add_argument("--ipv6", "-6" , action="store_true", help='Test Ipv6', default=False)
+        type=str,
-parser.add_argument("--proto", "-m" , type=str, required=False, choices=["tcp","udp"], help='Select the protocol', default="tcp")
+        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"
+    )
+    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()
+    args = parser.parse_args()
-sep()
+    sep()
-puts("Testing will start on ", color=colors.cyan, end="")
+    puts("Testing will start on ", color=colors.cyan, end="")
-puts(f"{args.address}", color=colors.yellow)
+    puts(f"{args.address}", color=colors.yellow)
 
-firegex = FiregexAPI(args.address)
+    firegex = FiregexAPI(args.address)
 
-#Login
+    # Login
-if (firegex.login(args.password)):
+    if firegex.login(args.password):
-    puts("Sucessfully logged in ✔", color=colors.green)
+        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)
-
-def exit_test(code):
-    if service_id:
-        server.stop()
-        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)
-            exit_test(1)        
-    exit(code)
-
-srvs = firegex.nfregex_get_services()
-for ele in srvs:
-    if ele['name'] == args.service_name:
-        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" )
-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)):
-    puts("Sucessfully started service ✔", color=colors.green)
-else:
-    puts("Test Failed: Failed to start service ✗", color=colors.red)
-    exit_test(1)
-
-server.start()
-time.sleep(0.5)
-try:
-    if server.sendCheckData(secrets.token_bytes(432)):
-        puts("Successfully tested first proxy with no regex ✔", color=colors.green)
     else:
-        puts("Test Failed: Data was corrupted ", color=colors.red)
+        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
+    )
+
+    def exit_test(code):
+        if service_id:
+            server.stop()
+            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)
+                exit_test(1)
+        exit(code)
+
+    srvs = firegex.nfregex_get_services()
+    for ele in srvs:
+        if ele["name"] == args.service_name:
+            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"
+    )
+    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):
+        puts("Sucessfully started service ✔", color=colors.green)
+    else:
+        puts("Test Failed: Failed to start service ✗", color=colors.red)
         exit_test(1)
-except Exception:
-    puts("Test Failed: Couldn't send data to the server ", color=colors.red)
-    exit_test(1)
-#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): 
+    server.start()
-    puts(f"Sucessfully added regex {str(secret)} ✔", color=colors.green)
+    time.sleep(0.5)
-else:
+    try:
-    puts(f"Test Failed: Couldn't add the regex {str(secret)} ✗", color=colors.red)
+        if server.sendCheckData(secrets.token_bytes(432)):
-    exit_test(1)
+            puts("Successfully tested first proxy with no regex ✔", color=colors.green)
+        else:
+            puts("Test Failed: Data was corrupted ", color=colors.red)
+            exit_test(1)
+    except Exception:
+        puts("Test Failed: Couldn't send data to the server ", color=colors.red)
+        exit_test(1)
+    # 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
+    ):
+        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
+    # Check if regex is present in the service
-n_blocked = 0
+    n_blocked = 0
 
-def getMetric(metric_name, regex):
+    def getMetric(metric_name, regex):
-    for metric in firegex.nfregex_get_metrics().split("\n"):
+        for metric in firegex.nfregex_get_metrics().split("\n"):
-        if metric.startswith(metric_name + "{") and f'regex="{regex}"' in metric:
+            if metric.startswith(metric_name + "{") and f'regex="{regex}"' in metric:
-            return int(metric.split(" ")[-1])
+                return int(metric.split(" ")[-1])
 
-def checkRegex(regex, should_work=True, upper=False, deleted=False):
+    def checkRegex(regex, should_work=True, upper=False, deleted=False):
-    if should_work:
+        if should_work:
+            global n_blocked
+            for r in firegex.nfregex_get_service_regexes(service_id):
+                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)
+                    ):
+                        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,
+                            )
+                        else:
+                            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
+                        ):
+                            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,
+                            )
+                            exit_test(1)
+                        if getMetric("firegex_active", secret.decode()) == 1:
+                            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,
+                            )
+                            exit_test(1)
+                    else:
+                        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)
+            ):
+                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,
+                )
+                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,
+                    )
+                else:
+                    puts(
+                        "Test Failed: The regex wasn't reported as inactive in the metrics ✗",
+                        color=colors.red,
+                    )
+                    exit_test(1)
+
+    def clear_regexes():
         global n_blocked
+        n_blocked = 0
         for r in firegex.nfregex_get_service_regexes(service_id):
             if r["regex"] == secret:
-                #Test the regex
+                if firegex.nfregex_delete_regex(r["id"]):
-                s = regex.upper() if upper else regex
+                    puts(
-                if not server.sendCheckData(secrets.token_bytes(40) + s +  secrets.token_bytes(40)):
+                        f"Sucessfully deleted regex with id {r['id']} ✔",
-                    puts("The malicious request was successfully blocked ✔", color=colors.green)
+                        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)
-                    else:
-                        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:
-                        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)
-                        exit_test(1)
-                    if getMetric("firegex_active", secret.decode()) == 1:
-                        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)
-                        exit_test(1)
                 else:
-                    puts("Test Failed: The request wasn't blocked ✗", color=colors.red)
+                    puts("Test Failed: Coulnd't delete the regex ✗", color=colors.red)
                     exit_test(1)
-                return
+                break
-        puts("Test Failed: The regex wasn't found ✗", color=colors.red)
+        if f'regex="{secret.decode()}"' not in firegex.nfregex_get_metrics():
-        exit_test(1)
+            puts("No regex metrics after deletion ✔", color=colors.green)
-    else:
-        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: Metrics found after deleting the regex ✗",
+                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)
-            else:
-                puts("Test Failed: The regex wasn't reported as inactive in the metrics ✗", color=colors.red)
-                exit_test(1)
 
-def clear_regexes():
+    checkRegex(secret)
-    global n_blocked
+
-    n_blocked = 0
+    # Pause the proxy
+    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)
+        exit_test(1)
+
+    # Check if it's actually paused
+    checkRegex(secret, should_work=False)
+
+    # Start firewall
+    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)
+        exit_test(1)
+
+    checkRegex(secret)
+
+    # Disable regex
     for r in firegex.nfregex_get_service_regexes(service_id):
         if r["regex"] == secret:
-            if(firegex.nfregex_delete_regex(r["id"])): 
+            if firegex.nfregex_disable_regex(r["id"]):
-                puts(f"Sucessfully deleted regex with id {r['id']} ✔", color=colors.green)
+                puts(
-            else: 
+                    f"Sucessfully disabled regex with id {r['id']} ✔",
-                puts("Test Failed: Coulnd't delete the regex ✗", color=colors.red)
+                    color=colors.green,
+                )
+            else:
+                puts("Test Failed: Coulnd't disable the regex ✗", color=colors.red)
                 exit_test(1)
             break
-    if f'regex="{secret.decode()}"' not in firegex.nfregex_get_metrics():
+
-        puts("No regex metrics after deletion ✔", color=colors.green)
+    # Check if it's actually disabled
+    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"]):
+                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)
+            break
+
+    checkRegex(secret)
+
+    # Delete regex
+    clear_regexes()
+
+    # Check if it's actually deleted
+    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
+    ):
+        puts(
+            f"Sucessfully added case insensitive regex {str(secret)} ✔",
+            color=colors.green,
+        )
     else:
-        puts("Test Failed: Metrics found after deleting the regex ✗", color=colors.red)
+        puts(
+            f"Test Failed: Coulnd't add the case insensitive regex {str(secret)} ✗",
+            color=colors.red,
+        )
         exit_test(1)
 
-checkRegex(secret)
+    checkRegex(secret, upper=True)
+    checkRegex(secret)
 
-#Pause the proxy
+    clear_regexes()
-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)
-    exit_test(1)
 
-#Check if it's actually paused
+    # Rename service
-checkRegex(secret,should_work=False)
+    if firegex.nfregex_rename_service(service_id, f"{args.service_name}2"):
-
+        puts(
-#Start firewall
+            f"Sucessfully renamed service to {args.service_name}2 ✔", color=colors.green
-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)
-    exit_test(1)
-
-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"])): 
-            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)
-        break
-
-#Check if it's actually disabled
-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"])): 
-            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)
-        break
-
-checkRegex(secret)
-
-#Delete regex
-clear_regexes()
-
-#Check if it's actually deleted
-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)): 
-    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)
-    exit_test(1)
-
-checkRegex(secret, upper=True)
-checkRegex(secret)
-
-clear_regexes()
-
-#Rename service
-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)
-else:
-    puts("Test Failed: Coulnd't rename service ✗", color=colors.red)
-    exit_test(1)
-
-#Check if service was renamed correctly
-service = firegex.nfregex_get_service(service_id)
-if service["name"] == f"{args.service_name}2":
-    puts("Checked that service was renamed correctly  ✔", color=colors.green)
-else:
-    puts("Test Failed: Service wasn't renamed correctly ✗", color=colors.red)
-    exit_test(1)
-
-#Rename back service
-if(firegex.nfregex_rename_service(service_id,f"{args.service_name}")):
-    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)):
-    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"]:
-        puts("Sucessfully changed service settings ✔", color=colors.green)
     else:
-        puts("Test Failed: Service settings weren't updated correctly ✗", color=colors.red)
+        puts("Test Failed: Coulnd't rename service ✗", color=colors.red)
         exit_test(1)
-else:
-    puts("Test Failed: Coulnd't change service settings ✗", color=colors.red)
-    exit_test(1)
 
-exit_test(0)
+    # Check if service was renamed correctly
+    service = firegex.nfregex_get_service(service_id)
+    if service["name"] == f"{args.service_name}2":
+        puts("Checked that service was renamed correctly  ✔", color=colors.green)
+    else:
+        puts("Test Failed: Service wasn't renamed correctly ✗", color=colors.red)
+        exit_test(1)
+
+    # Rename back service
+    if firegex.nfregex_rename_service(service_id, f"{args.service_name}"):
+        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,
+    ):
+        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"]
+        ):
+            puts("Sucessfully changed service settings ✔", color=colors.green)
+        else:
+            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)
+        exit_test(1)
+
+    exit_test(0)

tests/ph_test.py

@@ -7,131 +7,180 @@ import argparse
 import secrets
 import time
 
-parser = argparse.ArgumentParser()
+if __name__ == "__main__":
-parser.add_argument("--address", "-a", type=str , required=False, help='Address of firegex backend', default="http://127.0.0.1:4444/")
+    parser = argparse.ArgumentParser()
-parser.add_argument("--password", "-p", type=str, required=True, help='Firegex password')
+    parser.add_argument(
-parser.add_argument("--service_name", "-n", type=str , required=False, help='Name of the test service', default="Test Service")
+        "--address",
-parser.add_argument("--port", "-P", type=int , required=False, help='Port of the test service', default=1337)
+        "-a",
-parser.add_argument("--ipv6", "-6" , action="store_true", help='Test Ipv6', default=False)
+        type=str,
-parser.add_argument("--proto", "-m" , type=str, required=False, choices=["tcp","udp"], help='Select the protocol', default="tcp")
+        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"
+    )
+    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()
+    args = parser.parse_args()
-sep()
+    sep()
-puts("Testing will start on ", color=colors.cyan, end="")
+    puts("Testing will start on ", color=colors.cyan, end="")
-puts(f"{args.address}", color=colors.yellow)
+    puts(f"{args.address}", color=colors.yellow)
 
-firegex = FiregexAPI(args.address)
+    firegex = FiregexAPI(args.address)
 
-#Login
+    # Login
-if (firegex.login(args.password)):
+    if firegex.login(args.password):
-    puts("Sucessfully logged in ✔", color=colors.green)
+        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)
-
-def exit_test(code):
-    if service_id:
-        server.stop()
-        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)
-            exit_test(1)        
-    exit(code)
-
-srvs = firegex.ph_get_services()
-for ele in srvs:
-    if ele['name'] == args.service_name:
-        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")
-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)):
-    puts("Sucessfully started service ✔", color=colors.green)
-else:
-    puts("Test Failed: Failed to start service ✗", color=colors.red)
-    exit_test(1)
-
-server.start()
-time.sleep(0.5)
-
-#Check if it started
-def checkData(should_work):
-    res = None
-    try:
-        res = server.sendCheckData(secrets.token_bytes(432))
-    except (ConnectionRefusedError, TimeoutError):
-        res = None
-    if res:
-        if should_work:
-            puts("Successfully received data ✔", color=colors.green)
-        else:
-            puts("Test Failed: Connection wasn't blocked ✗", color=colors.red)
-            exit_test(1)
     else:
-        if should_work:
+        puts("Test Failed: Unknown response or wrong passowrd ✗", color=colors.red)
-            puts("Test Failed: Data wans't received ✗", color=colors.red)
+        exit(1)
-            exit_test(1)
+
+    # Create server
+    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):
+                puts("Sucessfully deleted service ✔", color=colors.green)
+            else:
+                puts("Test Failed: Coulnd't delete serivce ✗", color=colors.red)
+                exit_test(1)
+        exit(code)
+
+    srvs = firegex.ph_get_services()
+    for ele in srvs:
+        if ele["name"] == args.service_name:
+            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",
+    )
+    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):
+        puts("Sucessfully started service ✔", color=colors.green)
+    else:
+        puts("Test Failed: Failed to start service ✗", color=colors.red)
+        exit_test(1)
+
+    server.start()
+    time.sleep(0.5)
+
+    # Check if it started
+    def checkData(should_work):
+        res = None
+        try:
+            res = server.sendCheckData(secrets.token_bytes(432))
+        except (ConnectionRefusedError, TimeoutError):
+            res = None
+        if res:
+            if should_work:
+                puts("Successfully received data ✔", color=colors.green)
+            else:
+                puts("Test Failed: Connection wasn't blocked ✗", color=colors.red)
+                exit_test(1)
         else:
-            puts("Successfully blocked connection ✔", color=colors.green)
+            if should_work:
+                puts("Test Failed: Data wans't received ✗", color=colors.red)
+                exit_test(1)
+            else:
+                puts("Successfully blocked connection ✔", color=colors.green)
 
-checkData(True)
+    checkData(True)
 
-#Pause the proxy
+    # 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)
+        puts(f"Sucessfully paused service with id {service_id} ✔", color=colors.green)
-else:
+    else:
-    puts("Test Failed: Coulnd't pause the service ✗", color=colors.red)
+        puts("Test Failed: Coulnd't pause the service ✗", color=colors.red)
+        exit_test(1)
+
+    checkData(False)
+
+    # Start firewall
+    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)
+        exit_test(1)
+
+    checkData(True)
+
+    # Change port
+    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)
+        exit_test(1)
+
+    checkData(False)
+
+    server.stop()
+    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"):
+        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)
+
+    # Check if service was renamed correctly
+    for services in firegex.ph_get_services():
+        if services["name"] == f"{args.service_name}2":
+            puts("Checked that service was renamed correctly  ✔", color=colors.green)
+            exit_test(0)
+
+    puts("Test Failed: Service wasn't renamed correctly ✗", color=colors.red)
     exit_test(1)
-
-checkData(False)
-
-#Start firewall
-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)
-    exit_test(1)
-
-checkData(True)
-
-#Change port
-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)
-    exit_test(1)
-
-checkData(False)
-
-server.stop()
-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")):
-    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)
-
-#Check if service was renamed correctly
-for services in firegex.ph_get_services():
-    if services["name"] == f"{args.service_name}2":
-        puts("Checked that service was renamed correctly  ✔", color=colors.green)
-        exit_test(0)
-
-puts("Test Failed: Service wasn't renamed correctly ✗", 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,9 +1,46 @@
+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):
+    def __init__(self, port, ipv6, proxy_port=None, verbose=False):
         self.proxy_port = proxy_port
         self.ipv6 = ipv6
         self.port = port
@@ -12,43 +49,30 @@ 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()
-    
+
     def stop(self):
         self.server.terminate()
         self.server.join()
         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:
@@ -60,7 +84,7 @@ class TcpServer:
         if server_reply:
             self._server_data_queue.put(server_reply)
         self.client_sock.sendall(packet)
-    
+
     def recv_packet(self):
         try:
             return self.client_sock.recv(4096)
@@ -68,7 +92,7 @@ class TcpServer:
             if self.verbose:
                 traceback.print_exc()
             return False
-    
+
     def sendCheckData(self, data, get_data=False):
         self.connect_client()
         self.send_packet(data)

tests/utils/udpserver.py

@@ -1,35 +1,94 @@
-from multiprocessing import Process
+from multiprocessing import Process, Queue
 import socket
+import queue
+import traceback
+
+
+def _start_udp_server(port, server_queue: Queue, ipv6, verbose):
+    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))
+    while True:
+        try:
+            bytesAddressPair = sock.recvfrom(4096)
+            message = bytesAddressPair[0]
+            address = bytesAddressPair[1]
+
+            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):
+    def __init__(self, port, ipv6, proxy_port=None, verbose=False):
-        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))
-            while True:
-                bytesAddressPair = sock.recvfrom(432)
-                message = bytesAddressPair[0]
-                address = bytesAddressPair[1]
-                sock.sendto(message, address)
-                
-        self.ipv6 = ipv6
         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