diff --git a/backend/src/api/nft_manager.py b/backend/src/api/nft_manager.py index 1a9438e..e6e0d44 100644 --- a/backend/src/api/nft_manager.py +++ b/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: + if isinstance(left, dict) and "payload" in left and isinstance(right, str): 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. - - Only consider indented lines (rules are indented inside the chain block). - - 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". + Deterministic serializer to produce a compact UI-friendly string from expr list. + Covers common constructs; falls back to JSON dump for unknown constructs. """ - lines: List[str] = [] - if not text: - return lines - - for raw in text.splitlines(): - # preserve the original raw to check indentation - if raw is None: - continue - # ignore empty lines - if raw.strip() == "": - continue - # ignore closing braces (possibly with indentation) - if raw.strip() == "}": - continue - # Only accept lines that are indented (start with whitespace). - # This filters out top-level "table ..." and "chain ..." header lines. - if not raw.startswith((" ", "\t")): - # not indented => likely header/footer, skip - continue - # Now we have an indented line. Remove leading whitespace to get the content. - line = raw.lstrip().rstrip() - # skip chain metadata lines that end with ';' (e.g. "type filter hook ...; policy accept;") - if line.endswith(";"): - continue - # remove trailing " # handle N" if present - line = _handle_re.sub("", line).rstrip() - if line: - lines.append(line) - return lines + if expr is None: + return "" + if isinstance(expr, list): + tokens: List[str] = [] + for part in expr: + if isinstance(part, dict): + # common tokens + if "match" in part: + m = part["match"] + left = m.get("left") + right = m.get("right") + if isinstance(left, dict) and "payload" in left and isinstance(right, str): + p = left["payload"] + prot = p.get("protocol") + field = p.get("field") + if prot and field: + tokens.append(f"{prot} {field} {right}") + continue + tokens.append("match") + elif "payload" in part: + p = part["payload"] + prot = p.get("protocol") + field = p.get("field") + if prot and field: + tokens.append(f"payload({prot}.{field})") + continue + tokens.append("payload") + elif "cmp" in part or "binary" in part: + 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: - { "table": [ { "name": , "family": , "chains": [ { "name": , "rules": [] } ] } ] } - Uses nft_json only to discover families/tables/chains, then fetches textual chain listing for exact rule strings. + Convert nft -j list ruleset parsed JSON into a deterministic, predictable JSON: + { + "tables": [ + { "family": ..., "name": ..., "chains": [ { "name": ..., "rules": [ { handle, expr, text } ] } ] } + ] + } """ - result = {"table": []} - # nft_json is expected to be the parsed output of `nft -j list ruleset` which contains "nftables": [ ... ] - items = nft_json.get("nftables", []) - # discover tables and associated family/name + result: Dict[str, Any] = {"tables": []} + items = nft_json.get("nftables", []) if isinstance(nft_json, dict) else (nft_json or []) + + # 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 item in items: - if "table" in item: - t = item["table"] + for rec in items: + if "table" in rec: + t = rec["table"] fam = t.get("family") name = t.get("name") if fam and name: - key = (fam, name) - if key not in tables: - tables[key] = {"name": name, "family": fam, "chains": {}} - elif "chain" in item: - ch = item["chain"] - fam = ch.get("family") or ch.get("table", {}).get("family") # defensive - table_name = ch.get("table") or ch.get("table", {}).get("name") # defensive - chain_name = ch.get("name") + tables.setdefault((fam, name), {"family": fam, "name": name, "chains": {}}) + elif "chain" in rec: + ch = rec["chain"] + fam = ch.get("family") or (ch.get("table", {}) or {}).get("family") + table_name = ch.get("table") or (ch.get("table", {}) or {}).get("name") + cname = ch.get("name") + if fam and table_name and cname: + 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) - if key not in tables: - tables[key] = {"name": table_name, "family": fam, "chains": {}} - # register chain placeholder - tables[key]["chains"].setdefault(chain_name, {"name": chain_name, "rules": []}) + tables.setdefault((fam, table_name), {"family": fam, "name": table_name, "chains": {}}) + tables[(fam, table_name)]["chains"].setdefault(chain_name, {"name": chain_name, "rules": []}) + rule_obj: Dict[str, Any] = { + "handle": handle, + "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 - for (fam, tname), tdata in tables.items(): - chains_out: List[Dict[str, Any]] = [] + # Convert map to sorted lists for deterministic order + for (fam, tname) in sorted(tables.keys(), key=lambda k: (k[0], k[1])): + tdata = tables[(fam, tname)] + chains_list: List[Dict[str, Any]] = [] for cname in sorted(tdata["chains"].keys()): - try: - chain_text = mgr.list_chain_text(fam, tname, cname) - 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}) + chains_list.append({"name": cname, "rules": tdata["chains"][cname]["rules"]}) + result["tables"].append({"family": fam, "name": tname, "chains": chains_list}) + 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: - { "table": [ { "name": ..., "family": ..., "chains": [ { "name": ..., "rules": [ "", ... ] } ] } ] } + Returns the ruleset in a stable, predictable JSON shape derived from `nft -j list ruleset`. + Structure: + { "ruleset": { "tables": [ { "family": ..., "name": ..., "chains": [ { "name": ..., "rules": [ { "handle", "expr", "text" } ] } ] } ] } } - Implementation: - 1. Try to get JSON ruleset via nft -j list ruleset - 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). + Fallback: + - If nft JSON is unavailable, falls back to returning the raw textual ruleset string. """ 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_custom_ruleset_from_nft_json(nft_json) + custom = build_predictable_ruleset(nft_json) return {"ruleset": custom} except NftError as e: logger.exception("list_rules failed")