json approach

backend/src/api/nft_manager.py

@@ -91,7 +91,6 @@ class NftManager:
         cmd = f"list chain {family} {table} {chain}"
         # Attempt to temporarily disable JSON output on the wrapper (best-effort).
         json_toggled = False
-        prev_state_set = False
         try:
             if hasattr(self.nft, "set_json_output"):
                 try:
@@ -99,7 +98,6 @@ class NftManager:
                     self.nft.set_json_output(False)
                     json_toggled = True
                 except Exception:
-                    # If toggling fails, continue and try the cmd anyway.
                     logger.debug("could not toggle set_json_output(False); will try command anyway")
             res = self.cmd(cmd)
         finally:
@@ -129,27 +127,21 @@ class NftManager:
                 for rec in records:
                     if "rule" in rec:
                         r = rec["rule"]
-                        # Try to extract a concise textual representation:
-                        # If expr present and is a list, build a short textual form. This is heuristic.
                         expr = r.get("expr")
                         if isinstance(expr, list):
                             tokens: List[str] = []
                             for part in expr:
-                                # common forms: {"match": {...}}, {"payload": ...}, {"cmp": ...}, {"drop": null}, {"accept": null}
                                 if "match" in part:
                                     m = part["match"]
-                                    # try to extract 'left' payload protocol match to 'ip protocol icmp' form
                                     left = m.get("left")
                                     right = m.get("right")
-                                    # try payload -> protocol -> ip / field -> protocol
+                                    if isinstance(left, dict) and "payload" in left and isinstance(right, str):
-                                    if isinstance(left, dict) and "payload" in left:
                                         p = left["payload"]
                                         prot = p.get("protocol")
                                         field = p.get("field")
-                                        if prot and field and isinstance(right, str):
+                                        if prot and field:
                                             tokens.append(f"{prot} {field} {right}")
                                             continue
-                                    # fallback to rough match string
                                     tokens.append("match")
                                 elif "payload" in part:
                                     p = part["payload"]
@@ -163,19 +155,15 @@ class NftManager:
                                 elif "counter" in part:
                                     tokens.append("counter")
                                 else:
-                                    # generic fallback: include the keys present
                                     tokens.append("+".join(part.keys()))
                             rule_lines.append(" ".join(tokens))
                         else:
-                            # No expr list we can interpret; fallback to the raw JSON string (but not the entire nftables block)
                             rule_lines.append(json.dumps(r))
-                # Join into a pseudo-text block similar to `nft list chain` output (one rule per line)
                 if rule_lines:
                     return "\n".join(rule_lines)
             except Exception:
                 logger.debug("fallback JSON parsing of chain output failed; returning raw output")
 
-        # Prefer returning the raw textual output if we have it (lines etc.)
         return out
 
     def delete_rule_by_handle_text(self, family: str, table: str, chain: str, handle: int) -> None:
@@ -225,92 +213,117 @@ class RulesetOut(BaseModel):
 # ---------- Helpers to convert to desired shape ----------
 _handle_re = re.compile(r"\s+#\s*handle\s+\d+\s*$")
 
-def extract_rule_lines_from_chain_text(text: str) -> List[str]:
+def rule_text_from_expr(expr: Any) -> str:
     """
-    Given output of `nft list chain fam table chain`, extract the rule lines as strings.
+    Deterministic serializer to produce a compact UI-friendly string from expr list.
-    - Only consider indented lines (rules are indented inside the chain block).
+    Covers common constructs; falls back to JSON dump for unknown constructs.
-    - Skip chain header metadata lines that typically end with ';' (e.g. "type ...; policy ...;").
-    - Skip closing brace lines ('}').
-    - Remove trailing '# handle N' fragments.
-    Returns cleaned rule strings like "ip protocol icmp drop".
     """
-    lines: List[str] = []
+    if expr is None:
-    if not text:
+        return ""
-        return lines
+    if isinstance(expr, list):
-
+        tokens: List[str] = []
-    for raw in text.splitlines():
+        for part in expr:
-        # preserve the original raw to check indentation
+            if isinstance(part, dict):
-        if raw is None:
+                # common tokens
-            continue
+                if "match" in part:
-        # ignore empty lines
+                    m = part["match"]
-        if raw.strip() == "":
+                    left = m.get("left")
-            continue
+                    right = m.get("right")
-        # ignore closing braces (possibly with indentation)
+                    if isinstance(left, dict) and "payload" in left and isinstance(right, str):
-        if raw.strip() == "}":
+                        p = left["payload"]
-            continue
+                        prot = p.get("protocol")
-        # Only accept lines that are indented (start with whitespace).
+                        field = p.get("field")
-        # This filters out top-level "table ..." and "chain ..." header lines.
+                        if prot and field:
-        if not raw.startswith((" ", "\t")):
+                            tokens.append(f"{prot} {field} {right}")
-            # not indented => likely header/footer, skip
+                            continue
-            continue
+                    tokens.append("match")
-        # Now we have an indented line. Remove leading whitespace to get the content.
+                elif "payload" in part:
-        line = raw.lstrip().rstrip()
+                    p = part["payload"]
-        # skip chain metadata lines that end with ';' (e.g. "type filter hook ...; policy accept;")
+                    prot = p.get("protocol")
-        if line.endswith(";"):
+                    field = p.get("field")
-            continue
+                    if prot and field:
-        # remove trailing " # handle N" if present
+                        tokens.append(f"payload({prot}.{field})")
-        line = _handle_re.sub("", line).rstrip()
+                        continue
-        if line:
+                    tokens.append("payload")
-            lines.append(line)
+                elif "cmp" in part or "binary" in part:
-    return lines
+                    tokens.append("cmp")
+                elif "drop" in part:
+                    tokens.append("drop")
+                elif "accept" in part:
+                    tokens.append("accept")
+                elif "counter" in part:
+                    tokens.append("counter")
+                elif "tcp" in part or "udp" in part:
+                    proto = "tcp" if "tcp" in part else "udp"
+                    tokens.append(proto)
+                else:
+                    keys = "+".join(sorted(part.keys()))
+                    tokens.append(keys)
+            else:
+                tokens.append(str(part))
+        return " ".join(tokens)
+    return str(expr)
 
 
-def build_custom_ruleset_from_nft_json(nft_json: Dict[str, Any]) -> Dict[str, Any]:
+def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]:
     """
-    Build the desired structure:
+    Convert nft -j list ruleset parsed JSON into a deterministic, predictable JSON:
-    { "table": [ { "name": <table>, "family": <family>, "chains": [ { "name": <chain>, "rules": [<rule strings>] } ] } ] }
+    {
-    Uses nft_json only to discover families/tables/chains, then fetches textual chain listing for exact rule strings.
+      "tables": [
+         { "family": ..., "name": ..., "chains": [ { "name": ..., "rules": [ { handle, expr, text } ] } ] }
+      ]
+    }
     """
-    result = {"table": []}
+    result: Dict[str, Any] = {"tables": []}
-    # nft_json is expected to be the parsed output of `nft -j list ruleset` which contains "nftables": [ ... ]
+    items = nft_json.get("nftables", []) if isinstance(nft_json, dict) else (nft_json or [])
-    items = nft_json.get("nftables", [])
+
-    # discover tables and associated family/name
+    # Build intermediate map: (family, table) -> {family, name, chains: {chain_name: {"name", "rules":[]}}}
     tables: Dict[tuple, Dict[str, Any]] = {}
-    # items can contain separate objects for table/chain/rule entries
+    for rec in items:
-    for item in items:
+        if "table" in rec:
-        if "table" in item:
+            t = rec["table"]
-            t = item["table"]
             fam = t.get("family")
             name = t.get("name")
             if fam and name:
-                key = (fam, name)
+                tables.setdefault((fam, name), {"family": fam, "name": name, "chains": {}})
-                if key not in tables:
+        elif "chain" in rec:
-                    tables[key] = {"name": name, "family": fam, "chains": {}}
+            ch = rec["chain"]
-        elif "chain" in item:
+            fam = ch.get("family") or (ch.get("table", {}) or {}).get("family")
-            ch = item["chain"]
+            table_name = ch.get("table") or (ch.get("table", {}) or {}).get("name")
-            fam = ch.get("family") or ch.get("table", {}).get("family")  # defensive
+            cname = ch.get("name")
-            table_name = ch.get("table") or ch.get("table", {}).get("name")  # defensive
+            if fam and table_name and cname:
-            chain_name = ch.get("name")
+                tables.setdefault((fam, table_name), {"family": fam, "name": table_name, "chains": {}})
+                tables[(fam, table_name)]["chains"].setdefault(cname, {"name": cname, "rules": []})
+        elif "rule" in rec:
+            r = rec["rule"]
+            fam = r.get("family")
+            table_name = r.get("table")
+            chain_name = r.get("chain")
+            handle = r.get("handle")
+            expr = r.get("expr")
             if fam and table_name and chain_name:
-                key = (fam, table_name)
+                tables.setdefault((fam, table_name), {"family": fam, "name": table_name, "chains": {}})
-                if key not in tables:
+                tables[(fam, table_name)]["chains"].setdefault(chain_name, {"name": chain_name, "rules": []})
-                    tables[key] = {"name": table_name, "family": fam, "chains": {}}
+                rule_obj: Dict[str, Any] = {
-                # register chain placeholder
+                    "handle": handle,
-                tables[key]["chains"].setdefault(chain_name, {"name": chain_name, "rules": []})
+                    "expr": expr,
+                    "text": rule_text_from_expr(expr),
+                }
+                # include other useful metadata if present
+                if "position" in r:
+                    rule_obj["position"] = r["position"]
+                if "comment" in r:
+                    rule_obj["comment"] = r["comment"]
+                tables[(fam, table_name)]["chains"][chain_name]["rules"].append(rule_obj)
 
-    # Now for each discovered table+chain call textual `nft list chain ...` to get actual rule lines
+    # Convert map to sorted lists for deterministic order
-    for (fam, tname), tdata in tables.items():
+    for (fam, tname) in sorted(tables.keys(), key=lambda k: (k[0], k[1])):
-        chains_out: List[Dict[str, Any]] = []
+        tdata = tables[(fam, tname)]
+        chains_list: List[Dict[str, Any]] = []
         for cname in sorted(tdata["chains"].keys()):
-            try:
+            chains_list.append({"name": cname, "rules": tdata["chains"][cname]["rules"]})
-                chain_text = mgr.list_chain_text(fam, tname, cname)
+        result["tables"].append({"family": fam, "name": tname, "chains": chains_list})
-                rules_lines = extract_rule_lines_from_chain_text(chain_text)
+
-            except NftError as e:
-                logger.warning("failed to list chain text for %s %s %s: %s", fam, tname, cname, e)
-                # fallback to empty rules list on error for that chain
-                rules_lines = []
-            chains_out.append({"name": cname, "rules": rules_lines})
-        result["table"].append({"name": tname, "family": fam, "chains": chains_out})
     return result
 
 
@@ -319,28 +332,22 @@ def build_custom_ruleset_from_nft_json(nft_json: Dict[str, Any]) -> Dict[str, An
 @router.get("/rules", response_model=RulesetOut, summary="List ruleset")
 def list_rules():
     """
-    Returns the ruleset in the custom JSON shape:
+    Returns the ruleset in a stable, predictable JSON shape derived from `nft -j list ruleset`.
-      { "table": [ { "name": ..., "family": ..., "chains": [ { "name": ..., "rules": [ "<rule text>", ... ] } ] } ] }
+    Structure:
+      { "ruleset": { "tables": [ { "family": ..., "name": ..., "chains": [ { "name": ..., "rules": [ { "handle", "expr", "text" } ] } ] } ] } }
 
-    Implementation:
+    Fallback:
-      1. Try to get JSON ruleset via nft -j list ruleset
+      - If nft JSON is unavailable, falls back to returning the raw textual ruleset string.
-      2. Use JSON to discover tables & chains
-      3. For each chain fetch textual `nft list chain fam table chain` and extract rule lines
-      4. Return the composed structure
-    If JSON isn't available or an error occurs, fall back to returning the raw textual ruleset string (existing behavior).
     """
     try:
-        # Try to obtain JSON ruleset
         try:
             nft_json = mgr.list_rules_json()
         except NftError as e:
             logger.debug("could not obtain nft JSON ruleset: %s", e)
-            # fallback to returning raw textual ruleset (existing behavior)
             text = mgr.list_rules()
             return {"ruleset": text.strip() if text is not None else None}
 
-        # Build custom structure using the JSON to find tables/chains, and textual listing to obtain rule lines
+        custom = build_predictable_ruleset(nft_json)
-        custom = build_custom_ruleset_from_nft_json(nft_json)
         return {"ruleset": custom}
     except NftError as e:
         logger.exception("list_rules failed")