diff --git a/backend/src/api/nft_manager.py b/backend/src/api/nft_manager.py index 572602c..06ef32c 100644 --- a/backend/src/api/nft_manager.py +++ b/backend/src/api/nft_manager.py @@ -1,7 +1,7 @@ # app.py from typing import Any, Dict, List, Optional, Tuple, Union from fastapi import FastAPI, APIRouter, HTTPException, status -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, ValidationError import logging import json import re @@ -23,10 +23,11 @@ class NftManager: def __init__(self) -> None: self.nft = Nftables() try: + # prefer JSON output globally where available self.nft.set_json_output(True) self.nft.set_handle_output(True) except Exception: - logger.debug("set_json_output not available or ignored") + logger.debug("set_json_output/set_handle_output not available") def cmd(self, text_cmd: str) -> Dict[str, Optional[str]]: rc, out, err = self.nft.cmd(text_cmd) @@ -47,6 +48,7 @@ class NftManager: raise NftError(f"json decode error: {e}") def list_rules_text(self) -> str: + # best-effort: temporarily disable JSON output so we get textual form json_toggled = False try: if hasattr(self.nft, "set_json_output"): @@ -54,14 +56,14 @@ class NftManager: self.nft.set_json_output(False) json_toggled = True except Exception: - logger.debug("could not toggle set_json_output(False); will try command anyway") + logger.debug("could not toggle set_json_output(False)") 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)") + logger.debug("could not restore set_json_output(True)") if res["rc"] != 0: raise NftError(f"nft list ruleset failed: {res['stderr']}") @@ -77,24 +79,26 @@ class NftManager: self.nft.set_json_output(False) json_toggled = True except Exception: - logger.debug("could not toggle set_json_output(False); will try command anyway") + logger.debug("could not toggle set_json_output(False)") res = self.cmd(cmd) 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)") + logger.debug("could not restore set_json_output(True)") if res["rc"] != 0: raise NftError(f"nft {cmd} failed: {res['stderr']}") + # If returned output is JSON-formatted (rare), we try to derive textual lines, + # otherwise return raw textual output. out = res["stdout"] or "" s = out.strip() if s.startswith("{") or s.startswith("["): try: parsed = json.loads(s) - rule_lines: List[str] = [] + lines: List[str] = [] records = parsed.get("nftables") if isinstance(parsed, dict) else parsed if not isinstance(records, list): records = [] @@ -103,58 +107,39 @@ class NftManager: r = rec["rule"] expr = r.get("expr") if isinstance(expr, list): - tokens: List[str] = [] + # summarized tokenization + toks: List[str] = [] for part in expr: if isinstance(part, dict) and "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): + if isinstance(left, dict) and "payload" in left and isinstance(right, (str, int)): p = left["payload"] prot = p.get("protocol") field = p.get("field") if prot and field: - tokens.append(f"{prot} {field} {right}") + toks.append(f"{prot} {field} {right}") continue - tokens.append("match") - elif isinstance(part, dict) and "payload" in part: - p = part["payload"] - prot = p.get("protocol") - field = p.get("field") - tokens.append(f"payload({prot}.{field})") + toks.append("match") elif isinstance(part, dict) and "drop" in part: - tokens.append("drop") + toks.append("drop") elif isinstance(part, dict) and "accept" in part: - tokens.append("accept") + toks.append("accept") elif isinstance(part, dict) and "counter" in part: - tokens.append("counter") - elif isinstance(part, dict) and "queue" in part: - q = part["queue"] - if isinstance(q, dict): - num = q.get("num") or q.get("number") or q.get("queue_number") or q.get("from") or q.get("range") - tok = "queue" - if num is not None: - tok += f" num {num}" - if q.get("bypass"): - tok += " bypass" - tokens.append(tok) - else: - if isinstance(q, (int, float)): - tokens.append(f"queue num {int(q)}") - else: - tokens.append(f"queue num {q}") + toks.append("counter") else: if isinstance(part, dict): - tokens.append("+".join(part.keys())) + toks.append("+".join(part.keys())) else: - tokens.append(str(part)) - rule_lines.append(" ".join(tokens)) + toks.append(str(part)) + lines.append(" ".join(toks)) else: - rule_lines.append(json.dumps(r)) - if rule_lines: - return "\n".join(rule_lines) + lines.append(json.dumps(r)) + if lines: + return "\n".join(lines) except Exception: - logger.debug("fallback JSON parsing of chain output failed; returning raw output") + logger.debug("fallback json->text derivation failed; returning raw output") return out @@ -258,6 +243,7 @@ def parse_priority(val: Any) -> Optional[int]: def rule_text_from_expr(expr: Any) -> str: if expr is None: return "" + # try list first (common) if isinstance(expr, list): tokens: List[str] = [] for part in expr: @@ -266,7 +252,7 @@ def rule_text_from_expr(expr: Any) -> str: m = part["match"] left = m.get("left") right = m.get("right") - if isinstance(left, dict) and "payload" in left and isinstance(right, str): + if isinstance(left, dict) and "payload" in left and isinstance(right, (str, int)): p = left["payload"] prot = p.get("protocol") field = p.get("field") @@ -282,8 +268,6 @@ def rule_text_from_expr(expr: Any) -> str: 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: @@ -313,12 +297,20 @@ def rule_text_from_expr(expr: Any) -> str: else: tokens.append(str(part)) return " ".join(tokens) - return str(expr) + # if expr is a dict or other type, give a compact repr + try: + return str(expr) + except Exception: + return "" def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]: + """ + Convert nft -j list ruleset parsed JSON into a deterministic, predictable JSON: + """ result: Dict[str, Any] = {"tables": []} items = nft_json.get("nftables", []) if isinstance(nft_json, dict) else (nft_json or []) + tables: Dict[Tuple[str, str], Dict[str, Any]] = {} for rec in items: if "table" in rec: @@ -372,13 +364,16 @@ def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]: tables.setdefault((fam, table_name), {"family": fam, "name": table_name, "chains": {}}) chains_map = tables[(fam, table_name)]["chains"] chains_map.setdefault(chain_name, {"name": chain_name, "type": None, "hook": None, "priority": None, "policy": None, "rules": []}) - # we keep expr exactly as provided from JSON; do not replace it + + # do NOT change expr shape here; keep it exactly as NFT JSON provided rule_obj: Dict[str, Any] = {"handle": handle, "expr": expr, "text": rule_text_from_expr(expr)} if "position" in r: rule_obj["position"] = r["position"] if "comment" in r: rule_obj["comment"] = r["comment"] chains_map[chain_name]["rules"].append(rule_obj) + + # salvage chain-level metadata from rule record if present if isinstance(r.get("chain"), dict): csub = r.get("chain") if chains_map[chain_name].get("priority") is None: @@ -391,6 +386,8 @@ def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]: chains_map[chain_name]["hook"] = csub.get("hook") if chains_map[chain_name].get("policy") is None and csub.get("policy") is not None: chains_map[chain_name]["policy"] = csub.get("policy") + + # convert to lists (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]] = [] @@ -407,6 +404,7 @@ def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]: } ) result["tables"].append({"family": fam, "name": tname, "chains": chains_list}) + return result @@ -419,6 +417,7 @@ def parse_ruleset_text(nft_text: str) -> Dict[Tuple[str, str, str], List[Dict[st table_re = re.compile(r"^\s*table\s+(\S+)\s+(\S+)\s*\{") chain_re = re.compile(r"^\s*chain\s+(\S+)\s*\{") handle_re = re.compile(r"#\s*handle\s*(\d+)\b") + # only skip semicolon-terminated chain metadata lines (type/hook/priority/policy) chain_meta_re = re.compile(r"^\s*(type\b|hook\b|priority\b|policy\b)\b.*;") current_family = None @@ -447,6 +446,7 @@ def parse_ruleset_text(nft_text: str) -> Dict[Tuple[str, str, str], List[Dict[st if s == "" or s == "{" or s == "}": continue if chain_meta_re.match(s): + # skip chain metadata lines only continue m_handle = handle_re.search(s) @@ -464,11 +464,14 @@ 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: + """ + Enrich JSON-derived 'custom' structure in-place by setting only rule['text'] + when a reliable textual mapping is found. Do not change expr or other types. + """ if not nft_text: return parsed = parse_ruleset_text(nft_text) - for table in custom.get("tables", []): fam = table.get("family") tname = table.get("name") @@ -481,7 +484,7 @@ def populate_text_from_ruleset_text(custom: Dict[str, Any], nft_text: str) -> No key = (fam, tname, cname) textual_entries = parsed.get(key, []) if not textual_entries: - continue # nothing to enrich for this chain + continue handle_map: Dict[int, str] = {} ordered_lines: List[str] = [] @@ -494,37 +497,41 @@ def populate_text_from_ruleset_text(custom: Dict[str, Any], nft_text: str) -> No rules = chain.get("rules", []) for idx, rule in enumerate(rules): - replaced = False - # ensure we don't remove/overwrite expr + # ONLY update 'text' when we can map a textual line h = rule.get("handle") + mapped: Optional[str] = None if isinstance(h, int) and h in handle_map: - rule["text"] = handle_map[h] - replaced = True - - if not replaced: + mapped = handle_map[h] + else: pos = rule.get("position") if isinstance(pos, int) and 0 <= pos < len(ordered_lines): - rule["text"] = ordered_lines[pos] - replaced = True + mapped = ordered_lines[pos] + elif idx < len(ordered_lines): + mapped = ordered_lines[idx] - if not replaced and idx < len(ordered_lines): - rule["text"] = ordered_lines[idx] - replaced = True - - if not replaced: + # final substring probe (safe) + if mapped is None: probe = rule.get("text") or rule_text_from_expr(rule.get("expr")) if probe: for ln in ordered_lines: if probe in ln: - rule["text"] = ln - replaced = True + mapped = ln break - # only update 'text' when a match was found; otherwise leave existing rule['text'] + + if mapped is not None: + # ensure we only write a str into 'text' + try: + rule["text"] = str(mapped) + except Exception: + rule["text"] = mapped # should be str already def populate_text_from_chain_text(custom: Dict[str, Any]) -> None: - tables = custom.get("tables") or [] - for t in tables: + """ + Per-chain fallback that queries nft for the chain's textual listing and + updates only rule['text'] when a reliable mapping is found. + """ + for t in custom.get("tables", []): fam = t.get("family") tname = t.get("name") if not fam or not tname: @@ -545,88 +552,84 @@ def populate_text_from_chain_text(custom: Dict[str, Any]) -> None: except Exception: continue - for rule in ch.get("rules", []): - replaced = False + rules = ch.get("rules", []) + for idx, rule in enumerate(rules): + mapped: Optional[str] = None h = rule.get("handle") if isinstance(h, int) and h in handle_map: - rule["text"] = handle_map[h] - replaced = True - - if not replaced: - expr = rule.get("expr") - probe = rule.get("text") or rule_text_from_expr(expr) + mapped = handle_map[h] + else: + probe = rule.get("text") or rule_text_from_expr(rule.get("expr")) if probe: for ln in lines: if probe in ln: - rule["text"] = ln.strip() - replaced = True + mapped = ln.strip() break + + if mapped is not None: + try: + rule["text"] = str(mapped) + except Exception: + rule["text"] = mapped except Exception as e: - logger.debug( - "populate_text_from_chain_text: failed to get textual chain for %s %s %s: %s", - fam, - tname, - cname, - e, - ) + logger.debug("populate_text_from_chain_text failed for %s %s %s: %s", fam, tname, cname, e) continue -# ---------- Normalization helper to satisfy Pydantic model ---------- +# ---------- Normalization helper (lightweight and safe) ---------- def normalize_custom_for_model(custom: Dict[str, Any]) -> None: """ - Ensure every rule has required keys/types for RulesetModel parsing: - - 'expr' must exist (if missing/None -> set to []) - - 'text' must exist and be a str (if missing -> derive via rule_text_from_expr or "") - - 'handle' coerced to int or None - This mutates `custom` in-place. + Make minimal, safe guarantees required by Pydantic: + - rule['expr'] must exist (if None -> set to empty list) + - rule['text'] must be a str (if missing -> derived string) + - rule['handle'] coerced to int or None + Do NOT change any other shapes. """ - tables = custom.get("tables") or [] - for t in tables: + for t in custom.get("tables", []): for ch in t.get("chains", []): rules = ch.get("rules", []) or [] for r in rules: - # expr: required by model; prefer existing value, but don't leave None + # expr: if missing or None => set to [] (preserves Any) if "expr" not in r or r.get("expr") is None: - # default to empty list (acceptable Any value and consistent with prior shape) r["expr"] = [] - # text: must be str - txt = r.get("text") - if txt is None: + # text: ensure string + if "text" not in r or r.get("text") is None: try: r["text"] = rule_text_from_expr(r.get("expr")) except Exception: r["text"] = "" else: - # coerce to str if not string - if not isinstance(txt, str): + if not isinstance(r["text"], str): try: - r["text"] = str(txt) + r["text"] = str(r["text"]) except Exception: r["text"] = "" - # handle: ensure int or None + # handle: coerce to int or None h = r.get("handle") if isinstance(h, str): try: r["handle"] = int(h) except Exception: r["handle"] = None - elif not isinstance(h, int): - # if it's e.g. float, bool, etc. convert if reasonable + elif isinstance(h, float): try: - if isinstance(h, (float,)): - r["handle"] = int(h) - else: - r["handle"] = None + r["handle"] = int(h) except Exception: r["handle"] = None + elif not isinstance(h, int): + r["handle"] = None # ---------- Routes ---------- @router.get("/rules", response_model=RulesetOut, summary="List ruleset") def list_rules(): + """ + Returns JSON-derived ruleset (RulesetModel) and enriches each rule['text'] + with the textual nft rule line when possible. This function will not replace + JSON-derived 'expr' or other data types — enrichment is additive only. + """ try: try: nft_json = mgr.list_rules_json() @@ -634,41 +637,51 @@ def list_rules(): logger.debug("could not obtain nft JSON ruleset: %s", e) raise HTTPException(status_code=500, detail=f"Failed to obtain nft JSON ruleset: {e}") - # get textual snapshot (best-effort), used only to enrich 'text' values + # best-effort textual snapshot for enrichment nft_text = "" try: nft_text = mgr.list_rules_text() except Exception: - logger.debug("could not obtain textual nft ruleset (list_rules_text failed)") + logger.debug("could not obtain textual nft ruleset snapshot") - # build predictable JSON-derived structure (must be the source of truth) + # Build canonical JSON-derived shape (source of truth) custom = build_predictable_ruleset(nft_json) - # enrich existing JSON rules with textual lines (handle -> line; position/index fallback) + # Enrich only the 'text' field in-place using the textual snapshot 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) + logger.debug("populate_text_from_ruleset_text failed: %s", e) - # per-chain fallback that queries nft for chain-level textual lines + # Per-chain fallback using `nft list chain ...` try: populate_text_from_chain_text(custom) except Exception as e: - logger.debug("list_rules: populate_text_from_chain_text failed: %s", e) + logger.debug("populate_text_from_chain_text failed: %s", e) - # Normalize to satisfy Pydantic model (do this last) + # Normalize minimally for model validation normalize_custom_for_model(custom) - # Debug: log counts + # Debug counts 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) + # RETURN a shape matching response_model=RulesetOut + try: + ruleset_model = RulesetModel.parse_obj(custom) + except ValidationError as ve: + # log full validation error for debugging and return 500 with message + logger.exception("RulesetModel validation failed: %s", ve) + raise HTTPException(status_code=500, detail=f"Internal: ruleset validation failed: {ve}") + + return {"ruleset": ruleset_model} except NftError as e: logger.exception("list_rules failed") raise HTTPException(status_code=500, detail=str(e)) + except HTTPException: + raise except Exception as e: logger.exception("list_rules internal error") raise HTTPException(status_code=500, detail=str(e)) @@ -689,16 +702,10 @@ def create_rule_json(req: CreateRuleRequest): if req.expr is None: raise NftError("field 'expr' is required for JSON rule creation") - rendered = None - try: - rendered = rule_text_from_expr(req.expr) - except Exception: - rendered = None - - if rendered is None: + rendered = rule_text_from_expr(req.expr) + if rendered is None or rendered == "": raise NftError( - "cannot render provided 'expr' to textual nft syntax. " - "Please use POST /firewall/raw to execute the textual nft command." + "cannot render provided 'expr' to textual nft syntax. Use POST /firewall/raw" ) expr_text = rendered.strip() @@ -714,8 +721,7 @@ def create_rule_json(req: CreateRuleRequest): else: cmd = f"add rule {family} {table} {chain} {expr_text}" - logger.info("create_rule_json executing command: %s", cmd) - + logger.info("create_rule_json executing: %s", cmd) res = mgr.cmd(cmd) raw_rc = res.get("rc") stdout = res.get("stdout") or "" @@ -733,22 +739,19 @@ def create_rule_json(req: CreateRuleRequest): if rc == 0: return exec_res + # If non-zero rc but empty stderr, try to verify presence if (rc < 0 or rc != 0) and stderr.strip() == "": - logger.debug("create_rule_json: rc indicates failure but stderr empty; verifying rule presence") try: chain_text = mgr.list_chain_text(family, table, chain) or "" if expr_text and expr_text in chain_text: - logger.info("create_rule_json: detected rule in chain after add; treating as success") + logger.info("create_rule_json: rule detected after add; treating as success") 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) + logger.warning("create_rule_json verification failed: %s", e_chain) 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) - except NftError as e: logger.warning("create_rule_json NftError: %s", e) raise HTTPException(status_code=400, detail=str(e))