diff --git a/backend/src/api/nft_manager.py b/backend/src/api/nft_manager.py index 82c5073..bd5d978 100644 --- a/backend/src/api/nft_manager.py +++ b/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 - This class intentionally avoids json transactions: everything is textual 'nft' commands. + - convenience list_rules_text / list_rules_json / list_chain_text + 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(False) + self.nft.set_json_output(True) 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 `. - 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}" - res = self.cmd(cmd) + # 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: """