diff --git a/backend/src/api/nft_api.py b/backend/src/api/nft_api.py index aac022e..96ce6f5 100644 --- a/backend/src/api/nft_api.py +++ b/backend/src/api/nft_api.py @@ -333,30 +333,78 @@ def ensure_table_and_chain_exist(family: str, table: str, chain: str) -> None: logger.error("Failed to create chain '%s' in table '%s': %s", chain, table, getattr(e, "detail", str(e))) raise +def nft_list_chain_text( + family: str, + table: str, + chain: str, +) -> Dict[int, str]: + """ + Return a mapping: handle -> textual nft rule line + extracted from `nft list chain `. + + Example return: + { + 12: 'meta iifname "eth0" accept comment "allow lan"', + 13: 'ip saddr 10.0.0.0/24 drop' + } + """ + ensure_nft_available() + + cmd = [NFT_BIN, "list", "chain", family, table, chain] + logger.debug("Listing chain in text mode: %s", " ".join(cmd)) + + try: + out = subprocess.check_output(cmd, stderr=subprocess.PIPE).decode() + except subprocess.CalledProcessError as e: + logger.error("Failed to list chain text: %s", e.stderr.decode()) + raise HTTPException(status_code=500, detail=e.stderr.decode()) + + rules: Dict[int, str] = {} + + for line in out.splitlines(): + line = line.strip() + # Typical rule line contains: "handle " + # Example: + # meta iifname "eth0" accept comment "foo" handle 7 + if " handle " not in line: + continue + + try: + rule_part, handle_part = line.rsplit(" handle ", 1) + handle = int(handle_part.strip()) + rules[handle] = rule_part.strip() + except Exception: + logger.debug("Could not parse rule line: %s", line) + + return rules def nft_list_rules(table: str = "filter", chain: str = "forward") -> Dict[str, Any]: """ - Parse `nft --json list ruleset` and return a list of rules with metadata. - Each returned rule includes: - - family, table, chain - - handle (if present) - - position (1-based within its chain) - - comment (if present, best-effort) - - verdict (best-effort string: 'accept'/'drop'/'reject' or None) - - verdict_details (raw nested object for reject or other complex verdicts) - - exprs (original expression list from nft JSON) - - nft_rule (the original rule dict from nft JSON) + Parse nft rules using JSON mode for structure AND text mode for readability. - This function is conservative and aims to give the UI enough info to - display and edit rules precisely (by handle or position). + Returned fields per rule: + - family, table, chain + - handle + - position (1-based) + - comment (best-effort) + - verdict (best-effort) + - verdict_details + - exprs (JSON expressions) + - nft_rule (TEXTUAL rule line, exactly as nft prints it) """ ensure_nft_available() try: - out = subprocess.check_output([NFT_BIN, "--json", "list", "ruleset"], stderr=subprocess.PIPE) + out = subprocess.check_output( + [NFT_BIN, "--json", "list", "ruleset"], + stderr=subprocess.PIPE, + ) parsed = json.loads(out) except subprocess.CalledProcessError as e: - logger.error("Failed to list ruleset: %s", e.stderr.decode()) - raise HTTPException(status_code=500, detail=f"nft failed: {e.stderr.decode()}") + logger.error("Failed to list ruleset (json): %s", e.stderr.decode()) + raise HTTPException(status_code=500, detail=e.stderr.decode()) + + # 2) Text rules (human-readable) + text_rules = nft_list_chain_text(DEFAULT_FAMILY, table, chain) results: List[Dict[str, Any]] = [] counters: Dict[str, int] = {} @@ -373,46 +421,47 @@ def nft_list_rules(table: str = "filter", chain: str = "forward") -> Dict[str, A family = r.get("family") table_name = r.get("table") chain_name = r.get("chain") + + if table_name != table or chain_name != chain: + continue + key = f"{family}:{table_name}:{chain_name}" counters.setdefault(key, 0) counters[key] += 1 position = counters[key] - handle = r.get("handle") - exprs = r.get("expr", []) # original expression list + handle = r.get("handle") + exprs = r.get("expr", []) + comment: Optional[str] = None verdict: Optional[str] = None verdict_details: Optional[Any] = None - # scan expressions to extract comment and verdict/action for expr in exprs: if not isinstance(expr, dict): continue + if "comment" in expr: - c = expr.get("comment") + c = expr["comment"] if isinstance(c, str): comment = c elif isinstance(c, dict): - comment = c.get("text") or c.get("str") or comment + comment = c.get("text") or c.get("str") - # common verdict shapes: {"verdict": {"accept": null}} or {"drop": null} if "verdict" in expr: v = expr["verdict"] if isinstance(v, dict): - k = next(iter(v.keys()), None) - verdict = k - verdict_details = v.get(k) + verdict = next(iter(v.keys()), None) + verdict_details = v.get(verdict) else: verdict = str(v) + if "drop" in expr and verdict is None: verdict = "drop" - verdict_details = expr.get("drop") if "accept" in expr and verdict is None: verdict = "accept" - verdict_details = expr.get("accept") if "reject" in expr and verdict is None: verdict = "reject" - verdict_details = expr.get("reject") results.append( { @@ -425,16 +474,15 @@ def nft_list_rules(table: str = "filter", chain: str = "forward") -> Dict[str, A "verdict": verdict, "verdict_details": verdict_details, "exprs": exprs, - "nft_rule": r, + # 👇 THIS IS THE IMPORTANT CHANGE + "nft_rule": text_rules.get(handle), } ) - # filter by requested table/chain if provided - if table or chain: - results = [x for x in results if x["table"] == table and x["chain"] == chain] return {"rules": results} + def expr_to_nft_snippet(e: Expr) -> str: """Convert a typed Expr into a short nft syntax snippet (used for preview/add).""" if isinstance(e, MetaExpr):