test2

backend/src/api/nft_manager.py

@@ -1,201 +1,94 @@
 # app.py
 """
-Stateless nftables FastAPI service (libnftables only, no CLI fallback, no annotations).
+Unrestricted nftables FastAPI service (libnftables only) with a preview endpoint.
-Endpoints:
+
+Endpoints (high level):
   GET  /firewall/rules            -> list rules (kernel-provided rules with handles)
-  POST /firewall/rules            -> add a constrained rule (returns 201)
+  POST /firewall/rules            -> add rule (executes; accepts 'cmd' or 'json')
-  DELETE /firewall/rules/{handle} -> delete rule by handle
+  DELETE /firewall/rules/{handle} -> delete rule by handle (executes)
+  POST /firewall/raw              -> execute textual nft command (executes)
+  POST /firewall/jsoncmd          -> execute libnftables JSON command (executes)
+  POST /firewall/preview          -> PREVIEW what would be executed (no changes)
 Requirements:
-  python-nftables must be installed and usable:
+  - python-nftables installed
-    pip install python-nftables fastapi uvicorn pydantic
+  - CAP_NET_ADMIN or root required to modify nftables
-  The process must have CAP_NET_ADMIN (or be run as root) 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 ipaddress
 import logging
+import json
+import re
 
-# libnftables import (must be present)
+# libnftables (must be installed)
 from nftables import Nftables  # type: ignore
 
 # ---------- logging ----------
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger("nft_api")
 
-# ---------- NftManager (libnftables only) ----------
+# ---------- Exceptions ----------
 class NftError(RuntimeError):
     pass
 
-
+# ---------- NftManager (unrestricted) ----------
 class NftManager:
     """
-    Minimal libnftables wrapper that performs JSON transactions only via python-nftables.
+    Thin wrapper around python-nftables exposing:
-    No CLI fallback, no comments/annotations.
+      - json_cmd execution (accepts libnftables JSON dict)
+      - cmd execution (accepts textual nft commands)
+      - convenience list_rules + delete_by_handle
+    This class intentionally does NOT validate or restrict nft commands.
     """
 
-    # Tighten these to your environment
-    ALLOWED_FAMILIES = {"inet"}
-    ALLOWED_TABLES = {"filter"}
-    ALLOWED_CHAINS = {"input", "output", "forward"}
-    ALLOWED_PROTOS = {"tcp", "udp"}
-
     def __init__(self) -> None:
         self.nft = Nftables()
-        # Request JSON output where applicable
+        # request json output where available
         try:
             self.nft.set_json_output(True)
         except Exception:
-            # Some libnftables builds may ignore this; proceed anyway
+            logger.debug("set_json_output not available")
-            logger.debug("set_json_output not available or failed")
 
-    def _json_cmd(self, cmd: Dict[str, Any]) -> Any:
+    def json_cmd(self, cmd_obj: Dict[str, Any]) -> Any:
-        rc, out, err = self.nft.json_cmd(cmd)
+        """
+        Execute a libnftables JSON command object via json_cmd and return parsed output.
+        """
+        rc, out, err = self.nft.json_cmd(cmd_obj)
         if rc != 0:
-            logger.error("libnftables error: %s", err)
+            logger.error("json_cmd failed: %s", err)
-            raise NftError(err)
+            raise NftError(err if err else f"json_cmd exit {rc}")
         return out
 
-    def _validate_family_table_chain(self, family: str, table: str, chain: str) -> None:
+    def cmd(self, text_cmd: str) -> Dict[str, Optional[str]]:
-        if family not in self.ALLOWED_FAMILIES:
+        """
-            raise ValueError(f"family '{family}' not allowed")
+        Execute a textual nft command via Nftables.cmd().
-        if table not in self.ALLOWED_TABLES:
+        Returns dict { "stdout": out_str, "stderr": err_str, "rc": rc }.
-            raise ValueError(f"table '{table}' not allowed")
+        """
-        if chain not in self.ALLOWED_CHAINS:
+        rc, out, err = self.nft.cmd(text_cmd)
-            raise ValueError(f"chain '{chain}' not allowed")
+        if rc != 0:
-
+            logger.warning("cmd returned rc=%s, err=%s", rc, err)
-    def _validate_proto(self, proto: Optional[str]) -> None:
+        return {"rc": rc, "stdout": out, "stderr": err}
-        if proto is None:
-            return
-        if proto not in self.ALLOWED_PROTOS:
-            raise ValueError(f"protocol '{proto}' not allowed")
-
-    def _validate_src(self, src: Optional[str]) -> None:
-        if not src:
-            return
-        try:
-            # allow host or network
-            ipaddress.ip_network(src, strict=False)
-        except Exception as e:
-            raise ValueError(f"invalid src '{src}': {e}")
-
-    def _validate_port(self, port: Optional[int]) -> None:
-        if port is None:
-            return
-        if not (1 <= port <= 65535):
-            raise ValueError(f"invalid port: {port}")
 
     def list_rules(self) -> List[Dict[str, Any]]:
         """
-        Return raw rule dicts extracted from libnftables JSON output.
+        List the full ruleset and return collected 'rule' dicts (each includes handle when present).
-        Each item is the inner 'rule' dict returned by nftables JSON.
         """
-        out = self._json_cmd({"nftables": [{"list": {"ruleset": None}}]})
+        out = self.json_cmd({"nftables": [{"list": {"ruleset": None}}]})
         rules: List[Dict[str, Any]] = []
         for item in out.get("nftables", []):
             if "rule" in item:
                 rules.append(item["rule"])
         return rules
 
-    def add_rule(
+    def delete_by_handle(self, family: str, table: str, chain: str, handle: int) -> None:
-        self,
-        family: str,
-        table: str,
-        chain: str,
-        proto: Optional[str],
-        src: Optional[str],
-        dst_port: Optional[int],
-        verdict: str,
-    ) -> None:
         """
-        Add a constrained rule via libnftables JSON.
+        Delete rule by handle using JSON delete rule. No validation performed.
-        This constructs an 'expr' list that covers:
-          - ip saddr match (if src provided)
-          - l4 proto match (if proto provided)
-          - dport match (if dst_port provided)
-          - final verdict (accept/drop)
         """
-        # validate
-        self._validate_family_table_chain(family, table, chain)
-        self._validate_proto(proto)
-        self._validate_src(src)
-        self._validate_port(dst_port)
-        if verdict not in {"accept", "drop"}:
-            raise ValueError("verdict must be 'accept' or 'drop'")
-
-        expr: List[Dict[str, Any]] = []
-
-        # match source address (IP)
-        if src:
-            # Construct prefix match using nft JSON structure (ip payload -> saddr)
-            # The json shape used here is compatible with python-nftables expectations.
-            # We use ip payload field and prefix match for networks.
-            try:
-                net = ipaddress.ip_network(src, strict=False)
-                addr = str(net.network_address)
-                prefix_len = net.prefixlen
-            except Exception:
-                # fallback to treating as single IP
-                addr = src
-                prefix_len = 32 if ":" not in src else 128
-            expr.append({
-                "match": {
-                    "left": {"payload": {"protocol": "ip", "field": "saddr"}},
-                    "op": "==",
-                    "right": {"prefix": {"addr": addr, "len": prefix_len}}
-                }
-            })
-
-        # match L4 protocol
-        if proto:
-            # meta.l4proto match is a common approach in JSON exprs
-            expr.append({
-                "match": {
-                    "left": {"meta": {"key": "l4proto"}},
-                    "op": "==",
-                    "right": proto
-                }
-            })
-
-        # match destination port (only meaningful if proto provided)
-        if dst_port:
-            if not proto:
-                raise ValueError("dst_port requires proto to be set")
-            expr.append({
-                "match": {
-                    "left": {"payload": {"protocol": proto, "field": "dport"}},
-                    "op": "==",
-                    "right": dst_port
-                }
-            })
-
-        # final verdict
-        expr.append({"verdict": verdict})
-
-        payload = {
-            "nftables": [{
-                "add": {
-                    "rule": {
-                        "family": family,
-                        "table": table,
-                        "chain": chain,
-                        "expr": expr
-                    }
-                }
-            }]
-        }
-
-        self._json_cmd(payload)
-
-    def delete_rule(self, family: str, table: str, chain: str, handle: int) -> None:
-        """
-        Delete a rule by handle (authoritative).
-        """
-        self._validate_family_table_chain(family, table, chain)
         if not isinstance(handle, int) or handle <= 0:
-            raise ValueError("handle must be a positive integer")
+            raise ValueError("handle must be positive integer")
-
         payload = {
             "nftables": [{
                 "delete": {
@@ -208,22 +101,24 @@ class NftManager:
                 }
             }]
         }
-        self._json_cmd(payload)
+        self.json_cmd(payload)
 
 
-# ---------- FastAPI application (router combined) ----------
+# ---------- FastAPI + Router ----------
 router = APIRouter(prefix="/firewall", tags=["firewall"])
-manager = NftManager()
+mgr = NftManager()
 
 
+# Request models
 class CreateRuleRequest(BaseModel):
-    family: str = Field("inet", description="family (whitelisted)")
+    cmd: Optional[str] = Field(None, description="Textual nft command (e.g. 'add rule ...')")
-    table: str = Field("filter", description="table (whitelisted)")
+    json: Optional[Dict[str, Any]] = Field(None, description="libnftables JSON command object")
-    chain: str = Field("input", description="chain (whitelisted)")
+
-    proto: Optional[str] = Field(None, description="tcp|udp")
+class RawCmdRequest(BaseModel):
-    src: Optional[str] = Field(None, description="source IP or CIDR")
+    cmd: str = Field(..., description="Textual nft command to execute")
-    dst_port: Optional[int] = Field(None, description="destination port")
+
-    verdict: str = Field("accept", description="accept|drop")
+class JsonCmdRequest(BaseModel):
+    json: Dict[str, Any] = Field(..., description="libnftables JSON command object")
 
 
 class RuleOut(BaseModel):
@@ -234,15 +129,89 @@ class RuleOut(BaseModel):
     expr: Optional[Any]
 
 
+# ---------- PREVIEW helpers ----------
+def parse_text_cmd_summary(cmd: str) -> Dict[str, Any]:
+    """
+    Try to parse a short summary from a textual nft cmd.
+    We do NOT execute anything here — just regexp/token heuristics.
+    Returns a dict with possible keys: operation, family, table, chain, remainder.
+    """
+    summary: Dict[str, Any] = {"operation": None, "family": None, "table": None, "chain": None, "remainder": cmd}
+    # normalize whitespace
+    s = cmd.strip()
+    tokens = s.split()
+    if len(tokens) >= 1:
+        summary["operation"] = tokens[0].lower()
+    # detect patterns like: add rule <family> <table> <chain> ...
+    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
+    # detect delete rule by handle: e.g. delete rule inet filter input handle 42
+    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
+    # fallback: try to find family/table/chain tokens near 'rule'
+    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
+
+
+def parse_json_cmd_summary(obj: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    Extract a short summary from a libnftables JSON command object.
+    Looks into top-level 'nftables' list for add/delete/replace/insert keys and extracts rule family/table/chain.
+    """
+    summary = {"entries": []}
+    nft_entries = obj.get("nftables") if isinstance(obj, dict) else None
+    if not isinstance(nft_entries, list):
+        return {"error": "not a libnftables JSON object with 'nftables' list"}
+    for item in nft_entries:
+        # each item is like {"add": {"rule": {"family":...}}} or {"delete": {...}}
+        if not isinstance(item, dict):
+            continue
+        for k,v in item.items():
+            entry = {"op": k}
+            if isinstance(v, dict) and "rule" in v and isinstance(v["rule"], dict):
+                r = v["rule"]
+                entry["family"] = r.get("family")
+                entry["table"] = r.get("table")
+                entry["chain"] = r.get("chain")
+                # include handle if present
+                if "handle" in r:
+                    entry["handle"] = r.get("handle")
+            else:
+                # some commands are different shapes (e.g., add table ...). Try to capture names
+                # look for 'table' or 'chain' nested keys
+                if isinstance(v, dict):
+                    entry["info"] = {kk: vv for kk,vv in v.items() if kk in ("table","chain","family")}
+            summary["entries"].append(entry)
+    return summary
+
+
+# ---------- End preview helpers ----------
+
+# List rules
 @router.get("/rules", response_model=List[RuleOut])
 def list_rules():
-    """
-    Returns a list of rules (raw rule dicts from nftables) that include kernel handles.
-    Clients should save handles if they want to delete rules later.
-    """
     try:
-        rules = manager.list_rules()
+        rules = mgr.list_rules()
-        # Normalize fields for the response model: ensure expected keys exist
         out = []
         for r in rules:
             out.append({
@@ -250,7 +219,7 @@ def list_rules():
                 "table": r.get("table"),
                 "chain": r.get("chain"),
                 "handle": r.get("handle"),
-                "expr": r.get("expr")
+                "expr": r.get("expr"),
             })
         return out
     except NftError as e:
@@ -258,46 +227,100 @@ def list_rules():
         raise HTTPException(status_code=500, detail=str(e))
 
 
+# Create rule: executes
 @router.post("/rules", status_code=status.HTTP_201_CREATED)
 def create_rule(req: CreateRuleRequest):
     """
-    Create a constrained rule. Returns 201 on success.
+    Create a rule by sending either textual 'cmd' or libnftables 'json' object.
-    Use GET /firewall/rules to obtain the kernel-assigned handle.
+    This endpoint executes the command.
     """
     try:
-        manager.add_rule(
+        if req.cmd:
-            family=req.family,
+            res = mgr.cmd(req.cmd)
-            table=req.table,
+            if res["rc"] != 0:
-            chain=req.chain,
+                raise NftError(f"cmd failed rc={res['rc']}: {res.get('stderr')}")
-            proto=req.proto,
+            return {"status": "ok", "stdout": res.get("stdout")}
-            src=req.src,
+        if req.json:
-            dst_port=req.dst_port,
+            out = mgr.json_cmd(req.json)
-            verdict=req.verdict,
+            return {"status": "ok", "output": out}
-        )
+        raise HTTPException(status_code=400, detail="either 'cmd' or 'json' must be provided")
-        return {"status": "created"}
+    except NftError as e:
-    except (ValueError, NftError) as e:
+        logger.warning("create_rule failed: %s", e)
-        logger.warning("create_rule bad request: %s", e)
         raise HTTPException(status_code=400, detail=str(e))
     except Exception as e:
         logger.exception("create_rule internal error")
         raise HTTPException(status_code=500, detail=str(e))
 
 
-@router.delete("/rules/{handle}", status_code=status.HTTP_204_NO_CONTENT)
+# PREVIEW endpoint (does NOT execute anything)
-def delete_rule(
+@router.post("/preview")
-    handle: int,
+def preview_rule(req: CreateRuleRequest):
-    family: str = "inet",
-    table: str = "filter",
-    chain: str = "input",
-):
     """
-    Delete a rule by handle. family/table/chain default to common values but must match.
+    Preview what would be applied if you executed the given 'cmd' or 'json'.
+    This endpoint only analyzes and returns: textual command (if present), pretty JSON (if present),
+    and a parsed summary (operation, family, table, chain, etc.). Nothing is executed.
     """
     try:
-        manager.delete_rule(family=family, table=table, chain=chain, handle=handle)
+        if req.cmd:
+            text = req.cmd.strip()
+            summary = parse_text_cmd_summary(text)
+            return {
+                "type": "text",
+                "text_cmd": text,
+                "summary": summary,
+                "note": "This is a preview only. Nothing was executed."
+            }
+        if req.json:
+            pretty = json.dumps(req.json, indent=2, sort_keys=True)
+            summary = parse_json_cmd_summary(req.json)
+            return {
+                "type": "json",
+                "json_pretty": pretty,
+                "summary": summary,
+                "note": "This is a preview only. Nothing was executed."
+            }
+        raise HTTPException(status_code=400, detail="either 'cmd' or 'json' must be provided for preview")
+    except Exception as e:
+        logger.exception("preview_rule internal error")
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+# Delete by handle (authoritative)
+@router.delete("/rules/{handle}", status_code=status.HTTP_204_NO_CONTENT)
+def delete_rule(handle: int, family: str = "inet", table: str = "filter", chain: str = "input"):
+    """
+    Delete a rule by handle. family/table/chain are passed through to the delete JSON.
+    """
+    try:
+        mgr.delete_by_handle(family=family, table=table, chain=chain, handle=handle)
     except ValueError as e:
         logger.warning("delete_rule client error: %s", e)
         raise HTTPException(status_code=400, detail=str(e))
     except NftError as e:
         logger.exception("delete_rule failed")
         raise HTTPException(status_code=500, detail=str(e))
+    except Exception as e:
+        logger.exception("delete_rule internal error")
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+# Execute arbitrary textual nft command (convenience endpoint)
+@router.post("/raw")
+def exec_raw(req: RawCmdRequest):
+    try:
+        res = mgr.cmd(req.cmd)
+        return {"rc": res["rc"], "stdout": res["stdout"], "stderr": res["stderr"]}
+    except Exception as e:
+        logger.exception("exec_raw failed")
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+# Execute arbitrary JSON command
+@router.post("/jsoncmd")
+def exec_json(req: JsonCmdRequest):
+    try:
+        out = mgr.json_cmd(req.json)
+        return {"output": out}
+    except Exception as e:
+        logger.exception("exec_json failed")
+        raise HTTPException(status_code=400, detail=str(e))