firewall api and FE

backend/src/api/nft_manager.py

@@ -1,5 +1,5 @@
 # app.py
-from typing import Any, Dict, List, Optional, Union
+from typing import Any, Dict, List, Optional, Tuple, Union
 from fastapi import FastAPI, APIRouter, HTTPException, status
 from pydantic import BaseModel, Field
 import logging
@@ -119,7 +119,7 @@ class NftManager:
             try:
                 parsed = json.loads(s)
                 # parsed may be the whole ruleset (nftables list) or a list; find any "rule" objects
-                rule_lines = []
+                rule_lines: List[str] = []
                 # parsed might be dict with "nftables" or a list of records
                 records = parsed.get("nftables") if isinstance(parsed, dict) else parsed
                 if not isinstance(records, list):
@@ -186,33 +186,79 @@ router = APIRouter(prefix="/firewall", tags=["firewall"])
 mgr = NftManager()
 
 
-# ---------- Request/Response models ----------
+# ---------- Request/Response models (strongly typed) ----------
 class RawCmdRequest(BaseModel):
-    cmd: str = Field(description="Textual nft command to execute", example="add rule inet filter input ip saddr 10.0.0.0/8 drop")
+    cmd: str = Field(..., description="Textual nft command to execute", example="add rule inet filter input ip saddr 10.0.0.0/8 drop")
 
 
 class ExecResult(BaseModel):
-    rc: int = Field(description="Return code from nft execution", example=0)
-    stdout: Optional[str] = Field(None, description="Standard output from nft", example="") 
+    rc: int = Field(..., description="Return code from nft execution", example=0)
+    stdout: Optional[str] = Field(None, description="Standard output from nft", example="")
     stderr: Optional[str] = Field(None, description="Standard error from nft", example="")
 
     class Config:
         schema_extra = {"example": {"rc": 0, "stdout": "ok", "stderr": ""}}
 
 
-# ruleset may be a parsed JSON object (dict/list) OR a raw string (text) OR null
-RulesetValue = Union[Dict[str, Any], List[Any], str, None]
+# --- Strong models returned to frontend ---
+class RuleOut(BaseModel):
+    handle: Optional[int] = Field(None, description="The rule handle (unique per rule), if available", example=3)
+    expr: Any = Field(..., description="Machine-readable nft expression (original nft JSON expr).")
+    text: str = Field(..., description="Deterministic short display string derived from expr", example="ip protocol icmp drop")
+    position: Optional[Any] = Field(None, description="Optional position metadata from nft if present")
+    comment: Optional[str] = Field(None, description="Optional comment attached to the rule")
+
+    class Config:
+        schema_extra = {
+            "example": {
+                "handle": 3,
+                "expr": [{"match": {"left": {"payload": {"protocol": "ip", "field": "protocol"}}, "op": "==", "right": "icmp"}}, {"drop": None}],
+                "text": "ip protocol icmp drop",
+                "position": None,
+                "comment": None,
+            }
+        }
+
+
+class ChainOut(BaseModel):
+    name: str = Field(..., description="Chain name", example="forward")
+    rules: List[RuleOut] = Field(..., description="Rules in this chain (ordered)")
+
+    class Config:
+        schema_extra = {"example": {"name": "forward", "rules": []}}
+
+
+class TableOut(BaseModel):
+    family: str = Field(..., description="Table family (inet/bridge/ipv4/...)")
+    name: str = Field(..., description="Table name", example="filter")
+    chains: List[ChainOut] = Field(..., description="Chains in this table")
+
+    class Config:
+        schema_extra = {"example": {"family": "bridge", "name": "filter", "chains": []}}
+
+
+class RulesetModel(BaseModel):
+    tables: List[TableOut] = Field(..., description="Top-level tables list")
+
+    class Config:
+        schema_extra = {"example": {"tables": []}}
+
+
+# ruleset may be typed RulesetModel or raw textual string (fallback)
+RulesetValue = Optional[Union[RulesetModel, str]]
 
 
 class RulesetOut(BaseModel):
     ruleset: RulesetValue = Field(
-        description="parsed JSON ruleset (dict/list), or the raw textual ruleset as a string.",
+        None,
+        description="Parsed, strongly-typed ruleset (RulesetModel) or raw textual ruleset string if JSON is unavailable.",
     )
 
 
 # ---------- Helpers to convert to desired shape ----------
 _handle_re = re.compile(r"\s+#\s*handle\s+\d+\s*$")
 
+
 def rule_text_from_expr(expr: Any) -> str:
     """
     Deterministic serializer to produce a compact UI-friendly string from expr list.
@@ -278,7 +324,7 @@ def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]:
     items = nft_json.get("nftables", []) if isinstance(nft_json, dict) else (nft_json or [])
 
     # Build intermediate map: (family, table) -> {family, name, chains: {chain_name: {"name", "rules":[]}}}
-    tables: Dict[tuple, Dict[str, Any]] = {}
+    tables: Dict[Tuple[str, str], Dict[str, Any]] = {}
     for rec in items:
         if "table" in rec:
             t = rec["table"]
@@ -332,7 +378,8 @@ def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]:
 @router.get("/rules", response_model=RulesetOut, summary="List ruleset")
 def list_rules():
     """
-    Returns the ruleset in a stable, predictable JSON shape derived from `nft -j list ruleset`.
+    Returns the ruleset in a stable, strongly-typed JSON shape derived from `nft -j list ruleset`.
+
     Structure:
       { "ruleset": { "tables": [ { "family": ..., "name": ..., "chains": [ { "name": ..., "rules": [ { "handle", "expr", "text" } ] } ] } ] } }
 
@@ -345,10 +392,12 @@ def list_rules():
         except NftError as e:
             logger.debug("could not obtain nft JSON ruleset: %s", e)
             text = mgr.list_rules()
-            return {"ruleset": text.strip() if text is not None else None}
+            return RulesetOut(ruleset=text.strip() if text is not None else None)
 
         custom = build_predictable_ruleset(nft_json)
-        return {"ruleset": custom}
+        # Validate/construct Pydantic model so OpenAPI + client libs get accurate typing
+        ruleset_model = RulesetModel.parse_obj(custom)
+        return RulesetOut(ruleset=ruleset_model)
     except NftError as e:
         logger.exception("list_rules failed")
         raise HTTPException(status_code=500, detail=str(e))