From 051087fed300b30be8c3784ead2837be01e2fc67 Mon Sep 17 00:00:00 2001 From: malmert Date: Sat, 10 Jan 2026 21:10:59 +0100 Subject: [PATCH] improve nft parsing --- backend/src/api/nftables_api.py | 216 ++++++++++++++++++++++---------- 1 file changed, 150 insertions(+), 66 deletions(-) diff --git a/backend/src/api/nftables_api.py b/backend/src/api/nftables_api.py index aeb6d08..d43e850 100644 --- a/backend/src/api/nftables_api.py +++ b/backend/src/api/nftables_api.py @@ -154,71 +154,124 @@ def ensure_table_chain(family: str, table: str, chain: str) -> None: # ---------------------- reconstruct rules from nft JSON (best-effort) ---------------------- +# Replace your previous _reconstruct_rule_from_exprs and list_rules_from_nft with these. + +def _extract_immediate_value(node): + """Return an integer/string value from various immediate/data shapes, or None.""" + if not node or not isinstance(node, dict): + return None + # common keys used by libnftables JSON + for k in ("immediate", "value", "data", "s", "v"): + if k in node: + val = node[k] + # hex strings like "0x00000050" + if isinstance(val, str) and val.startswith("0x"): + try: + return int(val, 16) + except Exception: + return val + return val + # sometimes immediate sits under {"immediate": {"value": ...}} + if "right" in node or "left" in node: + # caller handles left/right structures + return None + return None + + def _reconstruct_rule_from_exprs(exprs: List[Dict[str, Any]]) -> Dict[str, Any]: """ - Heuristic mapping of nft expression JSON to our RuleModel fields. - Covers common shapes: meta (iif/oif/length), cmp/payload (ip proto and ports), verdict (action). - This is best-effort; complex expressions may not map perfectly. + Improved best-effort mapping from nft expression JSON to our RuleModel fields. + This function attempts to match many of the shapes produced by libnftables. """ match: Dict[str, Any] = {} action: Dict[str, Any] = {} - for e in exprs: + # Track last payload hint (if present) so that subsequent cmp can be interpreted + last_payload_hint: Optional[Dict[str, Any]] = None + + for idx, e in enumerate(exprs): + # handle meta (iif/oif/length) if "meta" in e: m = e["meta"] - key = m.get("key") or m.get("type") + key = m.get("key") or m.get("type") or m.get("field") v = m.get("v") or m.get("s") or m.get("value") - if key in ("iifname", "iif"): + # sometimes v is dict + if isinstance(v, dict): + v = _extract_immediate_value(v) or v.get("s") or v.get("v") + if key in ("iifname", "iif", "in", "iifname?"): if v: match["iif"] = v - elif key in ("oifname", "oif"): + elif key in ("oifname", "oif", "out"): if v: match["oif"] = v - elif key == "length": + elif key in ("length", "len"): if v is not None: match["meta_length"] = v + else: + logger.debug("meta with unknown key: %s value=%s", key, v) + + # store payload hints for later comparisons elif "payload" in e: - # payload describes read of header bytes; often next 'cmp' compares it - # we tag the payload on the expression to help cmp heuristics - e["_payload_hint"] = e["payload"] - elif "cmp" in e: - cmp = e["cmp"] - left = cmp.get("left") - right = cmp.get("right") - def _extract_immediate(node): - if not node or not isinstance(node, dict): - return None - for k in ("immediate", "value", "data", "s", "v"): - if k in node: - val = node[k] - if isinstance(val, str) and val.startswith("0x"): - try: - return int(val, 16) - except Exception: - return val - return val - return None - imm_left = _extract_immediate(left) - imm_right = _extract_immediate(right) - # ip proto numeric likely in 1..255 - for imm in (imm_left, imm_right): - if isinstance(imm, int) and 0 < imm < 256: - match["ip_proto"] = imm - break - # port heuristics (1..65535) + last_payload_hint = e["payload"] + # make it easier for cmp handling: include index + last_payload_hint["_idx"] = idx + + # cmp (compare) expressions: left/right might be payload / immediate structures + elif "cmp" in e or "match" in e: + cmp_obj = e.get("cmp") or e.get("match") or {} + left = cmp_obj.get("left") + right = cmp_obj.get("right") + # extract immediate numeric if present + imm_left = _extract_immediate_value(left) + imm_right = _extract_immediate_value(right) imm = imm_left if imm_left is not None else imm_right + + # If either immediate is a small int treat as ip_proto + if isinstance(imm, int) and 0 < imm < 256: + # prefer to store as number (frontend may show numeric) + match["ip_proto"] = imm + + # If immediate looks like port (1-65535), try to detect target (tcp/udp) if isinstance(imm, int) and 0 < imm <= 65535: - # best-effort assign to tcp_dport (common case) - # more advanced heuristics could inspect payload hints - match.setdefault("tcp_dport", imm) - elif "verdict" in e: - v = e["verdict"] - # shapes vary: dict or string + assigned = False + # heuristics: if left/right contains a payload hint referring to TCP/UDP or 'dport' strings: + for side in (left, right): + if isinstance(side, dict): + # payload form used by libnftables can include 'protocol' or 'field' + pl = side.get("payload") or side.get("left", {}).get("payload") + if isinstance(pl, dict): + protocol_hint = pl.get("protocol") or pl.get("proto") or pl.get("family") + field_hint = pl.get("field") or pl.get("meta") + sh = json.dumps(pl).lower() + if "tcp" in sh or "sport" in sh or "dport" in sh or "th" in sh: + match["tcp_dport"] = imm + assigned = True + break + if "udp" in sh or "udph" in sh or "udp." in sh: + match["udp_dport"] = imm + assigned = True + break + if not assigned: + # fallback: if ip_proto already indicates tcp(6) or udp(17), assign accordingly + proto = match.get("ip_proto") + if proto in (6, "tcp"): + match["tcp_dport"] = imm + elif proto in (17, "udp"): + match["udp_dport"] = imm + else: + # if we can't know, default to tcp_dport as most common case + match.setdefault("tcp_dport", imm) + + # verdict / immediate verdict expressions -> action + elif "verdict" in e or "immediate" in e or "return" in e: + # verdict may be string or dict + v = e.get("verdict") or e.get("return") or e.get("immediate") + # normalize dict forms if isinstance(v, dict): - t = v.get("type") or v.get("kind") + t = v.get("type") or v.get("kind") or v.get("verdict") if t: action["type"] = t - # redirect / queue extra fields vary + # redirect/queue shaped differently across versions if "to" in v: action["type"] = "redirect" action["redirect_port"] = v.get("to") @@ -226,49 +279,78 @@ def _reconstruct_rule_from_exprs(exprs: List[Dict[str, Any]]) -> Dict[str, Any]: action["type"] = "queue" action["queue_num"] = v.get("queue") elif isinstance(v, str): + # strings like "accept" or "drop" action["type"] = v - # other expression types intentionally ignored for stateless reconstruction + else: + # sometimes verdict is expressed as nested dict under 'verdict': {'kind':'accept'} + if isinstance(e.get("verdict"), dict): + vv = e["verdict"] + action["type"] = vv.get("kind") or vv.get("type") + # old-style 'match' entries with left/right payload/immediate + elif "match" in e: + m = e["match"] + left = m.get("left") + right = m.get("right") + imm_left = _extract_immediate_value(left) + imm_right = _extract_immediate_value(right) + if isinstance(imm_left, int) and 0 < imm_left < 256: + match["ip_proto"] = imm_left + if isinstance(imm_right, int) and 0 < imm_right < 256: + match["ip_proto"] = imm_right + + else: + # unknown expression type: log (DEBUG) for later tuning + logger.debug("unhandled nft expr type: %s", list(e.keys())) + + # default action to accept if kernel default or none found if "type" not in action: - # if kernel default, we assume accept action["type"] = "accept" + return {"match": match, "action": action} def list_rules_from_nft(family: str, table: str) -> List[Dict[str, Any]]: """ - Reconstruct an ordered list of rule dicts from `nft list table` JSON via the binding. - Returns list of dicts shaped like RuleModel (without id). + Reconstruct rules using libnftables JSON output with more robust parsing. """ - try: - out = nft_run_or_raise(f"list table {family} {table}") - except RuntimeError: - logger.debug("no table %s.%s found when listing rules", family, table) + # ensure binding present and request JSON output explicitly + _ensure_binding_available() + nft = Nftables() + nft.set_json_output(True) + rc, out, err = nft.cmd(f"list table {family} {table}") + if rc != 0: + logger.debug("nft list table returned rc=%s err=%s", rc, err) return [] - # attempt to parse JSON output (binding can return textual JSON) try: data = json.loads(out) - except Exception: - # force JSON mode in binding if previous parsing failed - _ensure_binding_available() - nft = Nftables() - nft.set_json_output(True) - rc, out_json, err = nft.cmd(f"list table {family} {table}") - if rc != 0: - logger.debug("nft JSON listing failed: %s", err) - return [] - data = json.loads(out_json) + except Exception as e: + logger.error("failed to parse nft JSON output: %s", e) + return [] results: List[Dict[str, Any]] = [] for item in data.get("nftables", []): if "rule" not in item: continue rule_obj = item["rule"] - exprs = rule_obj.get("expr", []) or rule_obj.get("expressions", []) or [] + # different versions might use 'expr', 'expressions' or 'expr' + exprs = rule_obj.get("expr") or rule_obj.get("exprs") or rule_obj.get("expressions") or [] + # if exprs is not a list but a dict (some shaped outputs) normalize + if isinstance(exprs, dict): + # sometimes expressions are nested inside a single dict; try to find array keys + for k in ("expr", "expressions", "exprs"): + v = exprs.get(k) + if isinstance(v, list): + exprs = v + break + else: + # give up and wrap + exprs = [exprs] + recon = _reconstruct_rule_from_exprs(exprs) - # the chain name may be in the rule metadata - chain_name = rule_obj.get("chain") or rule_obj.get("chain_name") or DEFAULT_CHAIN + chain_name = rule_obj.get("chain") or rule_obj.get("chain_name") or rule_obj.get("table") or DEFAULT_CHAIN + # make output match your RuleModel-ish structure (no id) rule_dict = { "family": family, "table": table, @@ -277,10 +359,12 @@ def list_rules_from_nft(family: str, table: str) -> List[Dict[str, Any]]: "action": recon.get("action", {}), } results.append(rule_dict) + logger.debug("reconstructed %d rules from %s.%s", len(results), family, table) return results + # ---------------------- API endpoints ---------------------- @router.get("/rules") def get_rules(family: Optional[str] = DEFAULT_FAMILY, table: Optional[str] = DEFAULT_TABLE, chain: Optional[str] = DEFAULT_CHAIN):