adding firewall function to firegex!

backend/utils/__init__.py

@@ -2,6 +2,8 @@ import asyncio
 from ipaddress import ip_address, ip_interface
 import os, socket, psutil, sys, nftables
 from fastapi_socketio import SocketManager
+from fastapi import Path
+from typing import Annotated
 
 LOCALHOST_IP = socket.gethostbyname(os.getenv("LOCALHOST_IP","127.0.0.1"))
 
@@ -15,6 +17,8 @@ FIREGEX_PORT = int(os.getenv("PORT","4444"))
 JWT_ALGORITHM: str = "HS256"
 API_VERSION = "2.0.0"
 
+PortType = Annotated[int, Path(gt=0, lt=65536)]   
+
 async def run_func(func, *args, **kwargs):
     if asyncio.iscoroutinefunction(func): 
         return await func(*args, **kwargs)
@@ -133,4 +137,4 @@ class NFTableManager(Singleton):
     def raw_list(self):
         return self.cmd({"list": {"ruleset": None}})["nftables"]
 
-            
+   

backend/utils/loader.py

@@ -1,7 +1,6 @@
 
-import os, httpx, websockets
-from sys import prefix
-from typing import Callable, List, Union
+import os, httpx
+from typing import Callable
 from fastapi import APIRouter, WebSocket
 import asyncio
 from starlette.responses import StreamingResponse
@@ -49,10 +48,10 @@ def list_routers():
     return [ele[:-3] for ele in list_files(ROUTERS_DIR) if ele != "__init__.py" and " " not in ele and ele.endswith(".py")]
 
 class RouterModule():
-    router: Union[None, APIRouter]
-    reset: Union[None, Callable]
-    startup: Union[None, Callable]
-    shutdown: Union[None, Callable]
+    router: None|APIRouter
+    reset: None|Callable
+    startup: None|Callable
+    shutdown: None|Callable
     name: str
     
     def __init__(self, router: APIRouter, reset: Callable, startup: Callable, shutdown: Callable, name:str):
@@ -66,7 +65,7 @@ class RouterModule():
         return f"RouterModule(router={self.router}, reset={self.reset}, startup={self.startup}, shutdown={self.shutdown})"
 
 def get_router_modules():
-    res: List[RouterModule] = []
+    res: list[RouterModule] = []
     for route in list_routers():
         module = getattr(__import__(f"routers.{route}"), route, None)
         if module:

backend/utils/models.py

@@ -1,4 +1,3 @@
-from typing import Union
 from pydantic import BaseModel
 
 class StatusMessageModel(BaseModel):
@@ -18,7 +17,7 @@ class PasswordChangeForm(BaseModel):
 
 class ChangePasswordModel(BaseModel):
     status: str
-    access_token: Union[str,None]
+    access_token: str|None
 
 class IpInterface(BaseModel):
     addr: str

backend/utils/sqlite.py

@@ -1,11 +1,9 @@
-from typing import Union
 import json, sqlite3, os
 from hashlib import md5
-import base64
 
 class SQLite():
     def __init__(self, db_name: str, schema:dict = None) -> None:
-        self.conn: Union[None, sqlite3.Connection] = None
+        self.conn: sqlite3.Connection|None = None
         self.cur = None
         self.db_name = db_name
         self.__backup = None
@@ -58,10 +56,25 @@ class SQLite():
             cur.close()
     
     def query(self, query, *values):
+        return self.queries([(query, *values)])[0]
+
+    def queries(self, queries: list[tuple[str, ...]]):
+        return list(self.queries_iter(queries))
+
+    def queries_iter(self, queries: list[tuple[str, ...]]):
         cur = self.conn.cursor()
         try:
-            cur.execute(query, values)
-            return cur.fetchall()
+            for query_data in queries:
+                values = []
+                str_query = None
+                if isinstance(query_data, str):
+                    str_query = query_data
+                elif (isinstance(query_data, tuple) or isinstance(query_data, list)) and len(query_data) > 0 and isinstance(query_data[0], str):
+                    str_query = query_data[0]
+                    values = query_data[1:]
+                if str_query:
+                    cur.execute(str_query, values)
+                yield cur.fetchall()
         finally:
             cur.close()
             try: self.conn.commit()