added more types

backend/src/api/nft_manager.py

@@ -1,27 +1,8 @@
 # app.py
-"""
-Unrestricted nftables FastAPI service using textual nft commands only.
-
-Endpoints (high level):
-  GET  /firewall/rules            -> return textual ruleset (raw nft output or parsed JSON when nft returned JSON)
-  POST /firewall/rules           -> execute a textual nft command (convenience)
-  DELETE /firewall/rules/{handle} -> delete rule by handle via textual nft command
-  POST /firewall/raw             -> execute arbitrary textual nft command (executes)
-  POST /firewall/preview         -> PREVIEW what textual command would do (no changes)
-
-Requirements:
-  - python-nftables must be installed for the Nftables wrapper (we still use python-nftables to call .cmd())
-  - CAP_NET_ADMIN or root required to modify nftables
-
-WARNING:
-  This service can run arbitrary nft commands. Run only in a trusted environment.
-"""
-
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Union
 from fastapi import FastAPI, APIRouter, HTTPException, status
 from pydantic import BaseModel, Field
 import logging
-import re
 import json
 
 # libnftables (we call textual commands through its .cmd() method)
@@ -63,7 +44,7 @@ class NftManager:
             logger.warning("nft cmd rc=%s stderr=%s cmd=%s", rc, err, text_cmd)
         return {"rc": rc, "stdout": out, "stderr": err}
 
-    def list_rules_text(self) -> str:
+    def list_rules(self) -> str:
         """
         Return the textual ruleset as produced by 'nft list ruleset'.
         Uses the textual command path.
@@ -95,71 +76,51 @@ mgr = NftManager()
 
 # ---------- Request/Response models ----------
 class RawCmdRequest(BaseModel):
-    cmd: str = Field(..., description="Textual nft command to execute")
+    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="") 
+    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 = Dict
 
 
 class RulesetOut(BaseModel):
-    # ruleset can be either a parsed JSON object (dict/list) or a string
-    ruleset: Any
-
-
-# ---------- PREVIEW helpers (unchanged) ----------
-def parse_text_cmd_summary(cmd: str) -> Dict[str, Any]:
-    summary: Dict[str, Any] = {"operation": None, "family": None, "table": None, "chain": None, "remainder": cmd}
-    s = cmd.strip()
-    tokens = s.split()
-    if len(tokens) >= 1:
-        summary["operation"] = tokens[0].lower()
-    m = re.match(r'^(add|insert|delete|replace)\s+rule\s+(\S+)\s+(\S+)\s+(\S+)\b', s, flags=re.I)
-    if m:
-        summary["operation"] = m.group(1).lower()
-        summary["family"] = m.group(2)
-        summary["table"] = m.group(3)
-        summary["chain"] = m.group(4)
-        summary["remainder"] = s[m.end():].strip()
-        return summary
-    m2 = re.match(r'^(delete)\s+rule\s+(\S+)\s+(\S+)\s+(\S+)\s+handle\s+(\d+)', s, flags=re.I)
-    if m2:
-        summary["operation"] = m2.group(1).lower()
-        summary["family"] = m2.group(2)
-        summary["table"] = m2.group(3)
-        summary["chain"] = m2.group(4)
-        summary["remainder"] = f"handle {m2.group(5)}"
-        return summary
-    try:
-        idx = next(i for i,t in enumerate(tokens) if t.lower() == "rule")
-        if len(tokens) > idx + 3:
-            summary["family"] = tokens[idx+1]
-            summary["table"] = tokens[idx+2]
-            summary["chain"] = tokens[idx+3]
-            summary["remainder"] = " ".join(tokens[idx+4:]) if len(tokens) > idx+4 else ""
-    except StopIteration:
-        pass
-    return summary
+    ruleset: RulesetValue = Field(
+        description="Parsed JSON ruleset (dict/list).",
+    )
 
 
 # ---------- Routes ----------
 
-# GET /firewall/rules -> textual nft ruleset output or parsed JSON when applicable
-@router.get("/rules", response_model=RulesetOut)
+@router.get("/rules", response_model=RulesetOut, summary="List ruleset")
 def list_rules():
     """
     Returns the textual nft ruleset output (as a string) or native JSON if nft returned JSON.
     Clients should handle both cases.
     """
     try:
-        text = mgr.list_rules_text()
+        text = mgr.list_rules()
         if text is None:
             return {"ruleset": None}
         s = text.strip()
-        # Try to interpret as JSON. nft will produce JSON when set_json_output(True) is active.
+        # Try to interpret as JSON. nft may produce JSON when set_json_output(True).
         try:
             parsed = json.loads(s)
-            # Return parsed JSON (will be serialized as JSON by FastAPI)
+            # Return parsed JSON (FastAPI will serialize)
             return {"ruleset": parsed}
         except json.JSONDecodeError:
             # Not JSON — return raw text
-            return {"ruleset": text}
+            # Important: we do NOT treat non-json as an internal error; return raw string.
+            logger.debug("nft output is not JSON; returning raw text")
+            return {"ruleset": s}
     except NftError as e:
         logger.exception("list_rules failed")
         raise HTTPException(status_code=500, detail=str(e))
@@ -168,18 +129,20 @@ def list_rules():
         raise HTTPException(status_code=500, detail=str(e))
 
 
-# POST /firewall/rules -> execute a textual nft command (convenience)
-@router.post("/rules", status_code=status.HTTP_201_CREATED)
+@router.post("/rules", response_model=ExecResult, status_code=status.HTTP_201_CREATED, summary="Execute textual rule command")
 def create_rule_text(req: RawCmdRequest):
     """
-    Execute a textual nft command (convenience, returns raw stdout).
+    Execute a textual nft command (convenience, returns structured exec result).
     Example: add rule inet filter input ip saddr 10.0.0.0/8 drop
     """
     try:
         res = mgr.cmd(req.cmd)
-        if res["rc"] != 0:
-            raise NftError(f"cmd failed rc={res['rc']}: {res.get('stderr')}")
-        return {"status": "ok", "stdout": res.get("stdout")}
+        rc = int(res.get("rc", -1) or -1)
+        if rc != 0:
+            # Bad request: command failed (client-provided textual command)
+            # Return 400 with the stderr detail
+            raise NftError(f"cmd failed rc={rc}: {res.get('stderr')}")
+        return ExecResult(rc=rc, stdout=res.get("stdout"), stderr=res.get("stderr"))
     except NftError as e:
         logger.warning("create_rule_text failed: %s", e)
         raise HTTPException(status_code=400, detail=str(e))
@@ -188,8 +151,7 @@ def create_rule_text(req: RawCmdRequest):
         raise HTTPException(status_code=500, detail=str(e))
 
 
-# DELETE /firewall/rules/{handle} -> delete by handle using textual nft command
-@router.delete("/rules/{handle}", status_code=status.HTTP_204_NO_CONTENT)
+@router.delete("/rules/{handle}", status_code=status.HTTP_204_NO_CONTENT, summary="Delete rule by handle")
 def delete_rule(handle: int, family: str = "inet", table: str = "filter", chain: str = "input"):
     """
     Delete a rule by handle using textual nft command.
@@ -198,6 +160,7 @@ def delete_rule(handle: int, family: str = "inet", table: str = "filter", chain:
     """
     try:
         mgr.delete_rule_by_handle_text(family=family, table=table, chain=chain, handle=handle)
+        # 204 No Content — nothing to return
     except ValueError as e:
         logger.warning("delete_rule client error: %s", e)
         raise HTTPException(status_code=400, detail=str(e))
@@ -209,19 +172,18 @@ def delete_rule(handle: int, family: str = "inet", table: str = "filter", chain:
         raise HTTPException(status_code=500, detail=str(e))
 
 
-# Execute arbitrary textual nft command
-@router.post("/raw")
+@router.post("/raw", response_model=ExecResult, summary="Execute raw textual nft command")
 def exec_raw(req: RawCmdRequest):
     """
-    Execute an arbitrary textual nft command and return {rc, stdout, stderr}.
+    Execute an arbitrary textual nft command and return structured {rc, stdout, stderr}.
     """
     try:
         res = mgr.cmd(req.cmd)
-        return {"rc": res["rc"], "stdout": res["stdout"], "stderr": res["stderr"]}
+        rc = int(res.get("rc", -1) or -1)
+        return ExecResult(rc=rc, stdout=res.get("stdout"), stderr=res.get("stderr"))
     except Exception as e:
         logger.exception("exec_raw failed")
         raise HTTPException(status_code=500, detail=str(e))
 
 
-# include router
 app.include_router(router)