diff --git a/backend/src/api/nft_manager.py b/backend/src/api/nft_manager.py index 6671bb0..b355dc5 100644 --- a/backend/src/api/nft_manager.py +++ b/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": "", "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_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 - # as printed by `nft list chain `. This is best-effort and - # will not fail the overall listing if textual retrieval fails for some chains. + # First: try to enrich using the full textual ruleset we already fetched + 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) + + # If some chains still look unpopulated, attempt per-chain listing fallback try: populate_text_from_chain_text(custom) except Exception as e: