test

backend/src/api/nft_manager.py

@@ -632,7 +632,172 @@ def expr_to_text(expr: Any) -> Optional[str]:
     return " ".join(parts).strip()
 
 
-# ---------- New helper: populate_text_from_chain_text ----------
+# ---------- New helper: parse_ruleset_text ----------
+def parse_ruleset_text(nft_text: str) -> Dict[Tuple[str, str, str], List[Dict[str, Optional[Union[str, int]]]]]:
+    """
+    Parse the full textual `nft list ruleset` output and return a mapping:
+      (family, table, chain) -> [ { "line": "<text line>", "handle": <int or None> }, ... ]
+
+    This is a best-effort parser that:
+      - detects table headers like: 'table <family> <name> {'
+      - detects chain headers like: 'chain <name> {'
+      - 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_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).*;")
+
+    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()
+        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
+        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')
+        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 current_family and current_table and current_chain:
+            # skip empty or purely-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
+                continue
+
+            # likely a rule line — extract handle if present
+            m_handle = handle_re.search(s)
+            handle_val: Optional[int] = None
+            if m_handle:
+                try:
+                    handle_val = int(m_handle.group(1))
+                except Exception:
+                    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
+
+
+# ---------- New helper: populate_text_from_ruleset_text ----------
+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.
+    """
+    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")
+        if not fam or not tname:
+            continue
+        for chain in table.get("chains", []):
+            cname = chain.get("name")
+            if not cname:
+                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:
+                ln = ent.get("line") or ""
+                h = ent.get("handle")
+                ordered_lines.append(ln)
+                if isinstance(h, int):
+                    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
+
+                # 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:
+                        for ln in ordered_lines:
+                            if probe in ln:
+                                rule["text"] = ln
+                                replaced = True
+                                break
+                # if nothing matched, keep existing rule['text'] (from JSON serializer)
+
+
+# ---------- 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) ----------
 def populate_text_from_chain_text(custom: Dict[str, Any]) -> None:
     """
     Replace rule['text'] in the 'custom' predictable ruleset with the exact textual
@@ -717,9 +882,14 @@ def list_rules():
 
         custom = build_predictable_ruleset(nft_json)
 
-        # Enrich rule['text'] by attempting to fetch the exact textual nft rule lines
+        # First: try to enrich using the full textual ruleset we already fetched
-        # as printed by `nft list chain <family> <table> <chain>`. This is best-effort and
+        try:
-        # will not fail the overall listing if textual retrieval fails for some chains.
+            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)
+
+        # If some chains still look unpopulated, attempt per-chain listing fallback
         try:
             populate_text_from_chain_text(custom)
         except Exception as e: