diff --git a/backend/src/api/nft_manager.py b/backend/src/api/nft_manager.py index 88e11fa..164b9dc 100644 --- a/backend/src/api/nft_manager.py +++ b/backend/src/api/nft_manager.py @@ -69,9 +69,23 @@ class NftManager: Obtain nft list ruleset in textual form. This is a fallback if JSON parsing fails. Returns raw textual output on success, raises NftError on failure. """ - self.nft.set_json_output(False) # best-effort attempt to disable JSON output - res = self.cmd("list ruleset") - self.nft.set_json_output(True) # restore JSON output preference + # best-effort toggle JSON off for textual output + json_toggled = False + try: + if hasattr(self.nft, "set_json_output"): + try: + self.nft.set_json_output(False) + json_toggled = True + except Exception: + logger.debug("could not toggle set_json_output(False); will try command anyway") + res = self.cmd("list ruleset") + finally: + 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 list ruleset failed: {res['stderr']}") return res["stdout"] or "" @@ -526,25 +540,17 @@ def expr_to_text(expr: Any) -> Optional[str]: Best-effort renderer that converts a typical nft JSON expr (list) into a textual fragment suitable to append to 'add rule ...'. Returns None when it cannot deterministically render the provided expr. - Supported cases (common): - - [{'match': {'left': {'payload': {'protocol':'ip','field':'protocol'}}, 'op':'==', 'right':'icmp'}}, {'drop': None}] - -> 'ip protocol icmp drop' - - payload / tcp / udp / counter / accept - - queue tokens and optional bypass support - This intentionally does not attempt to support every nft JSON construct. """ if expr is None: return "" if isinstance(expr, str): return expr if not isinstance(expr, list): - # unsupported top-level type return None parts: List[str] = [] for element in expr: if isinstance(element, dict): - # handle drop/accept/counter directly if "drop" in element: parts.append("drop") continue @@ -554,8 +560,6 @@ def expr_to_text(expr: Any) -> Optional[str]: if "counter" in element: parts.append("counter") continue - - # queue support: allow {"queue": 1} or {"queue": {"num":1, "bypass": True}} etc. if "queue" in element: q = element["queue"] token = "queue" @@ -571,31 +575,22 @@ def expr_to_text(expr: Any) -> Optional[str]: token += f" num {q}" parts.append(token) continue - - # match left/right payload equals -> ip protocol icmp, or ip saddr/daddr if "match" in element: m = element["match"] left = m.get("left") right = m.get("right") - # payload matches if isinstance(left, dict) and "payload" in left and isinstance(right, (str, int)): p = left["payload"] prot = p.get("protocol") field = p.get("field") - # common: protocol field match (protocol == icmp) if prot and field and isinstance(right, str): - # ip vs ip6 decision is left to the frontend; here we render 'ip protocol icmp' (works for many setups) if field == "protocol": parts.append(f"{prot} {field} {right}") continue - # payload might be l4 ports etc; produce generic payload(...) token parts.append(f"payload({prot}.{field}) {right}") continue - # fallback for match: try to stringify right parts.append("match") continue - - # payload shorthand if "payload" in element: p = element["payload"] prot = p.get("protocol") @@ -605,12 +600,9 @@ def expr_to_text(expr: Any) -> Optional[str]: continue parts.append("payload") continue - - # tcp/udp as nested dicts sometimes appear if "tcp" in element or "udp" in element: proto = "tcp" if "tcp" in element else "udp" val = element.get(proto) - # attempt to detect dport/sport keys if isinstance(val, dict): if "dport" in val: parts.append(f"{proto} dport {val['dport']}") @@ -620,15 +612,10 @@ def expr_to_text(expr: Any) -> Optional[str]: continue parts.append(proto) continue - - # cmp/binary operators etc — not supported deterministically - # return None to indicate we can't safely render this expr return None else: - # non-dict token (string/number) parts.append(str(element)) - # join tokens return " ".join(parts).strip() @@ -637,71 +624,53 @@ def parse_ruleset_text(nft_text: str) -> Dict[Tuple[str, str, str], List[Dict[st """ Parse the full textual `nft list ruleset` output and return a mapping: (family, table, chain) -> [ { "line": "", "handle": }, ... ] - - This is a best-effort parser that: - - detects table headers like: 'table {' - - detects chain headers like: 'chain {' - - collects lines inside a chain that look like rule lines (not chain metadata like 'type ...; policy ...;') - - extracts '# handle N' when present """ result: Dict[Tuple[str, str, str], List[Dict[str, Optional[Union[str, int]]]]] = {} if not nft_text: return result table_re = re.compile(r"^\s*table\s+(\S+)\s+(\S+)\s*\{") + # chain header line often: 'chain forward {', may include trailing comments chain_re = re.compile(r"^\s*chain\s+(\S+)\s*\{") handle_re = re.compile(r"#\s*handle\s*(\d+)\b") - # lines that indicate chain metadata (not rules) - chain_meta_re = re.compile(r"\b(type\b|hook\b|priority\b|policy\b|counter\b).*;") + # lines that indicate chain metadata (type/hook/priority/policy), we want to skip these + chain_meta_re = re.compile(r"^\s*(type\b|hook\b|priority\b|policy\b)\b.*;") current_family = None current_table = None current_chain = None - # track depth to handle nested braces more robustly - brace_depth = 0 # iterate line-by-line for raw_ln in nft_text.splitlines(): - ln = raw_ln.rstrip() + ln = raw_ln.rstrip("\n") s = ln.strip() - # update brace depth counting '{' and '}' to keep context - # but also we rely on explicit table/chain headers - open_count = ln.count("{") - close_count = ln.count("}") - # check for table header + # detect table header m_table = table_re.match(ln) if m_table: current_family = m_table.group(1) current_table = m_table.group(2) current_chain = None - brace_depth += open_count - close_count continue - # chain header (may include additional metadata and optional '# handle N') + # detect chain header (works even if the same line contains trailing comment) m_chain = chain_re.match(ln) if m_chain and current_family and current_table: current_chain = m_chain.group(1) - # ensure mapping exists key = (current_family, current_table, current_chain) result.setdefault(key, []) - brace_depth += open_count - close_count continue - # adjust brace depth for other lines - brace_depth += open_count - close_count - - # if we're inside a chain, try to find rule-like lines + # if we're inside a chain, collect rule-like lines if current_family and current_table and current_chain: - # skip empty or purely-brace lines + # skip empty / pure brace lines if s == "" or s == "{" or s == "}": continue - # skip semicolon-terminated metadata lines inside chain (type/hook/policy) - if chain_meta_re.search(s) or s.endswith(";"): - # these are chain-level metadata, not rules + # skip chain metadata lines like 'type filter hook forward priority 0;' + if chain_meta_re.match(s): continue - # likely a rule line — extract handle if present + # likely a rule line — extract optional handle comment m_handle = handle_re.search(s) handle_val: Optional[int] = None if m_handle: @@ -711,7 +680,6 @@ def parse_ruleset_text(nft_text: str) -> Dict[Tuple[str, str, str], List[Dict[st handle_val = None key = (current_family, current_table, current_chain) - # store the line as-is (trim leading whitespace), and handle (if found) result.setdefault(key, []).append({"line": ln.strip(), "handle": handle_val}) return result @@ -721,16 +689,11 @@ def parse_ruleset_text(nft_text: str) -> Dict[Tuple[str, str, str], List[Dict[st def populate_text_from_ruleset_text(custom: Dict[str, Any], nft_text: str) -> None: """ Uses the parsed full ruleset textual output (nft_text) to update rule['text'] - in the 'custom' structure in-place. - - Matching strategy: - 1) For each chain, build mapping handle -> line. - 2) For each rule in the JSON-built custom structure: - - If rule.handle exists and a matching handle line is found, use that. - - Otherwise, attempt to match by order/position: assign the i-th textual rule line - to the i-th JSON rule for that chain (best-effort). - - If position field exists in the JSON rule, prefer that index. - - If neither works, leave rule['text'] untouched. + in the 'custom' structure in-place. Matching strategy: + 1) handle -> line + 2) explicit 'position' field if present + 3) index-based mapping (best-effort) + 4) substring probe match """ if not nft_text: return @@ -748,7 +711,6 @@ def populate_text_from_ruleset_text(custom: Dict[str, Any], nft_text: str) -> No continue key = (fam, tname, cname) textual_entries = parsed.get(key, []) - # build handle map and ordered lines list handle_map: Dict[int, str] = {} ordered_lines: List[str] = [] for ent in textual_entries: @@ -759,29 +721,23 @@ def populate_text_from_ruleset_text(custom: Dict[str, Any], nft_text: str) -> No handle_map[h] = ln rules = chain.get("rules", []) - # iterate rules and apply mapping for idx, rule in enumerate(rules): replaced = False h = rule.get("handle") - # 1) Try handle match if handle present if isinstance(h, int) and h in handle_map: rule["text"] = handle_map[h] replaced = True if not replaced: - # 2) Try explicit position if provided (numerical) pos = rule.get("position") if isinstance(pos, int) and 0 <= pos < len(ordered_lines): rule["text"] = ordered_lines[pos] replaced = True - if not replaced: - # 3) Try index-based match (best-effort); use idx in JSON rules order - if idx < len(ordered_lines): - rule["text"] = ordered_lines[idx] - replaced = True + if not replaced and idx < len(ordered_lines): + rule["text"] = ordered_lines[idx] + replaced = True - # 4) As a last-ditch, try substring match (probe) within ordered_lines if not replaced: probe = rule.get("text") or rule_text_from_expr(rule.get("expr")) if probe: @@ -790,14 +746,10 @@ def populate_text_from_ruleset_text(custom: Dict[str, Any], nft_text: str) -> No rule["text"] = ln replaced = True break - # if nothing matched, keep existing rule['text'] (from JSON serializer) + # leave as-is if not replaced -# ---------- Helpers to render expr -> textual nft (best-effort) ---------- -# (expr_to_text already defined above) - - -# ---------- Existing populate_text_from_chain_text (no change, used as fallback) ---------- +# ---------- Existing populate_text_from_chain_text (fallback) ---------- def populate_text_from_chain_text(custom: Dict[str, Any]) -> None: """ Replace rule['text'] in the 'custom' predictable ruleset with the exact textual @@ -836,17 +788,14 @@ def populate_text_from_chain_text(custom: Dict[str, Any]) -> None: replaced = True if not replaced: - # fallback: try to find a line that contains the JSON-derived compact text fragment expr = rule.get("expr") probe = rule.get("text") or rule_text_from_expr(expr) if probe: - # try longest-first strategy (not strictly necessary here) — simple substring match for ln in lines: if probe in ln: rule["text"] = ln.strip() replaced = True break - # if still not replaced, keep existing rule["text"] except Exception as e: logger.debug( "populate_text_from_chain_text: failed to get textual chain for %s %s %s: %s", @@ -864,30 +813,72 @@ def populate_text_from_chain_text(custom: Dict[str, Any]) -> None: def list_rules(): """ Returns the ruleset in a stable, strongly-typed JSON shape derived from `nft -j list ruleset`. - - Structure: - { "ruleset": { "tables": [ { "family": ..., "name": ..., "chains": [ { "name": ..., "type": ..., "hook": ..., "priority": ..., "policy": ..., "rules": [ { "handle", "expr", "text" } ] } ] } ] } } - - Fallback: - - If nft JSON is unavailable, falls back to returning the raw textual ruleset string. """ try: try: nft_json = mgr.list_rules_json() - nft_text = mgr.list_rules_text() - logger.error("Successfully obtained nft JSON ruleset, but also got textual output: %r", nft_text) except NftError as e: logger.debug("could not obtain nft JSON ruleset: %s", e) raise HTTPException(status_code=500, detail=f"Failed to obtain nft JSON ruleset: {e}") + # try to obtain textual ruleset too (best-effort) + nft_text = "" + try: + nft_text = mgr.list_rules_text() + except Exception: + logger.debug("could not obtain textual nft ruleset (list_rules_text failed)") + custom = build_predictable_ruleset(nft_json) - # First: try to enrich using the full textual ruleset we already fetched + # If no tables/rules were discovered from JSON, attempt to build from textual listing + try: + if nft_text and (not custom.get("tables")): + # Build minimal structure from textual parse if JSON did not produce tables + parsed = parse_ruleset_text(nft_text) + if parsed: + # convert parsed -> custom structure + tables_map: Dict[Tuple[str, str], Dict[str, Any]] = {} + for (fam, tname, cname), entries in parsed.items(): + tables_map.setdefault((fam, tname), {"family": fam, "name": tname, "chains": {}}) + chain_rules: List[Dict[str, Any]] = [] + for ent in entries: + ln = ent.get("line") or "" + h = ent.get("handle") + # we don't have JSON expr; leave expr None and text the line + chain_rules.append({"handle": h, "expr": None, "text": ln}) + tables_map[(fam, tname)]["chains"].setdefault(cname, {"name": cname, "type": None, "hook": None, "priority": None, "policy": None, "rules": []}) + tables_map[(fam, tname)]["chains"][cname]["rules"].extend(chain_rules) + + # convert map to custom shape + custom = {"tables": []} + for (fam, tname) in sorted(tables_map.keys(), key=lambda k: (k[0], k[1])): + t = tables_map[(fam, tname)] + chains_list = [] + for cname in sorted(t["chains"].keys()): + ch = t["chains"][cname] + chains_list.append({"name": ch["name"], "type": None, "hook": None, "priority": None, "policy": None, "rules": ch["rules"]}) + custom["tables"].append({"family": fam, "name": tname, "chains": chains_list}) + except Exception as e: + logger.debug("fallback build from textual parse failed: %s", e) + + # Enrich rule['text'] using the full textual ruleset (handle-first, position, index) try: if nft_text: populate_text_from_ruleset_text(custom, nft_text) except Exception as e: logger.debug("list_rules: populate_text_from_ruleset_text failed: %s", e) + + # per-chain fallback (will not fail overall listing) + try: + populate_text_from_chain_text(custom) + except Exception as e: + logger.debug("list_rules: populate_text_from_chain_text failed: %s", e) + + # debug: log counts so you can see what's being returned + num_tables = len(custom.get("tables", [])) + num_rules = sum(len(ch.get("rules", [])) for t in custom.get("tables", []) for ch in t.get("chains", [])) + logger.info("list_rules: returning tables=%d rules=%d", num_tables, num_rules) + return RulesetModel.parse_obj(custom) except NftError as e: logger.exception("list_rules failed") @@ -906,12 +897,6 @@ def list_rules(): def create_rule_json(req: CreateRuleRequest): """ Create a rule from JSON (expr required). - - If req.position is provided, uses: insert rule
position - - Otherwise, uses: add rule
(append) - - If rendering fails: 400 instructing the client to use POST /firewall/raw - - Returns ExecResult on success (201) or on error (400) with stdout/stderr in body. - - If nft wrapper returns an invalid rc but the command produced no stderr, we double-check the chain - to see if the new rule is present; if present we treat as success. """ try: family = req.family @@ -930,12 +915,9 @@ def create_rule_json(req: CreateRuleRequest): expr_text = rendered.strip() - # If a position is explicitly requested, use the 'insert rule ... position ...' form. - # 'add rule ... position ...' is not supported by some nft versions / syntaxes. if req.position is not None: try: pos = int(req.position) - # clamp pos to >= 0 if pos < 0: pos = 0 except Exception: @@ -947,12 +929,10 @@ def create_rule_json(req: CreateRuleRequest): logger.info("create_rule_json executing command: %s", cmd) res = mgr.cmd(cmd) - # res expected {"rc": rc, "stdout": out, "stderr": err} raw_rc = res.get("rc") stdout = res.get("stdout") or "" stderr = res.get("stderr") or "" - # Coerce rc to int safely; if not int-like, set -1 to indicate unknown. try: rc = int(raw_rc) except Exception: @@ -962,33 +942,21 @@ def create_rule_json(req: CreateRuleRequest): exec_res = ExecResult(rc=rc, stdout=stdout or None, stderr=stderr or None) - # If rc == 0 — success if rc == 0: return exec_res - # Handle the annoying case: wrapper returned invalid rc (<0) or non-zero, - # but stderr is empty. The command may have succeeded nevertheless. if (rc < 0 or rc != 0) and stderr.strip() == "": logger.debug("create_rule_json: rc indicates failure but stderr empty; verifying rule presence") - - # Attempt to verify the rule exists by listing the chain and searching for a textual match. - # We use list_chain_text because it returns textual rule lines we can search for the preview text. try: chain_text = mgr.list_chain_text(family, table, chain) or "" - # Simple presence check: the textual fragment we attempted to add should be present - # as a substring in the chain listing (e.g. "ip protocol icmp drop"). if expr_text and expr_text in chain_text: logger.info("create_rule_json: detected rule in chain after add; treating as success") - # return success ExecResult with rc=0 to indicate success to client return ExecResult(rc=0, stdout=stdout or None, stderr=stderr or None) else: logger.debug("create_rule_json: rule not found in chain text; chain_text=%r", chain_text) except Exception as e_chain: logger.warning("create_rule_json: failed to list chain for verification: %s", e_chain) - # If we reach here -> treat as error: return 400 with exec_res in body. - # FastAPI cannot both raise HTTPException and include ExecResult as body easily, so raise HTTPException - # with detail that includes stderr and the executed cmd. detail = f"nft command failed rc={rc}. stderr: {stderr!r}. cmd: {cmd}" logger.warning("create_rule_json failed: %s", detail) raise HTTPException(status_code=400, detail=detail) @@ -997,7 +965,6 @@ def create_rule_json(req: CreateRuleRequest): logger.warning("create_rule_json NftError: %s", e) raise HTTPException(status_code=400, detail=str(e)) except HTTPException: - # re-raise HTTPException so we don't wrap it again raise except Exception as e: logger.exception("create_rule_json internal error") @@ -1009,8 +976,6 @@ def create_rule_json(req: CreateRuleRequest): def delete_rule(handle: int, family: str = "inet", table: str = "filter", chain: str = "input"): """ Delete a rule by handle using textual nft command. - Command executed: - delete rule
handle """ try: mgr.delete_rule_by_handle_text(family=family, table=table, chain=chain, handle=handle)