test2

backend/src/api/nft_manager.py

@@ -23,15 +23,16 @@ class NftManager:
     """
     Thin wrapper around python-nftables exposing:
       - cmd execution (textual nft commands via Nftables.cmd())
-      - convenience list_rules_text
+      - convenience list_rules_text / list_rules_json / list_chain_text
-    This class intentionally avoids json transactions: everything is textual 'nft' commands.
+    We prefer JSON globally, but for per-chain textual listing we temporarily disable JSON
+    so the output matches `nft list chain ...` textual rule lines.
     """
 
     def __init__(self) -> None:
         self.nft = Nftables()
+        # Try to prefer JSON for general listing; we'll toggle off for chain-list calls.
         try:
-            # Prefer JSON output where python-nftables may produce it, but we still treat stdout textually.
+            self.nft.set_json_output(True)
-            self.nft.set_json_output(False)
         except Exception:
             logger.debug("set_json_output not available or ignored")
 
@@ -60,8 +61,6 @@ class NftManager:
         Try to obtain nft -j list ruleset (JSON). Returns parsed JSON dict on success.
         Raises NftError on failure or when output cannot be parsed as JSON.
         """
-        # Try passing -j explicitly; depending on python-nftables wrapper, `set_json_output(True)` might already do it.
-        # If -j is not accepted by the wrapper, we still attempt a plain "list ruleset" which can be JSON depending on set_json_output.
         cmd_variants = ["list ruleset -j", "list ruleset"]
         last_err = None
         for c in cmd_variants:
@@ -71,14 +70,12 @@ class NftManager:
                 continue
             out = res["stdout"]
             if not out:
-                # empty output is treated as error here
                 last_err = "empty output"
                 continue
             try:
                 parsed = json.loads(out)
                 return parsed
             except json.JSONDecodeError as e:
-                # Not JSON for this invocation; continue to next variant
                 last_err = f"json decode error: {e}"
                 continue
         raise NftError(f"unable to get JSON ruleset: {last_err}")
@@ -86,13 +83,100 @@ class NftManager:
     def list_chain_text(self, family: str, table: str, chain: str) -> str:
         """
         Return textual output of `nft list chain <family> <table> <chain>`.
-        This output includes the chain header and rule lines. Caller should parse it.
+        This tries to temporarily disable JSON output so the wrapper returns the textual
+        representation used by `nft list ruleset`. If disabling JSON is not possible,
+        we attempt to parse returned JSON (as a last resort), but the preferred path is
+        to get textual output.
         """
         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:
+                    # Turn off JSON output to force textual output for this call.
+                    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:
+            # Restore JSON output preference if we toggled it.
+            if json_toggled and hasattr(self.nft, "set_json_output"):
+                try:
+                    self.nft.set_json_output(True)
+                except Exception:
+                    logger.debug("failed to restore set_json_output(True)")
+
         if res["rc"] != 0:
             raise NftError(f"nft {cmd} failed: {res['stderr']}")
-        return res["stdout"]
+
+        out = res["stdout"] or ""
+        # If the output looks like JSON (starts with '{' or '['), try a safe fallback:
+        s = out.strip()
+        if s.startswith("{") or s.startswith("["):
+            # Best-effort: parse JSON and attempt to extract rule textual forms if present.
+            try:
+                parsed = json.loads(s)
+                # parsed may be the whole ruleset (nftables list) or a list; find any "rule" objects
+                rule_lines = []
+                # 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):
+                    records = []
+                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:
+                                        p = left["payload"]
+                                        prot = p.get("protocol")
+                                        field = p.get("field")
+                                        if prot and field and isinstance(right, str):
+                                            tokens.append(f"{prot} {field} {right}")
+                                            continue
+                                    # fallback to rough match string
+                                    tokens.append("match")
+                                elif "payload" in part:
+                                    p = part["payload"]
+                                    prot = p.get("protocol")
+                                    field = p.get("field")
+                                    tokens.append(f"payload({prot}.{field})")
+                                elif "drop" in part:
+                                    tokens.append("drop")
+                                elif "accept" in part:
+                                    tokens.append("accept")
+                                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:
         """