From 3a5eb0a46dce3ef704d72b875c4f1f776aee604d Mon Sep 17 00:00:00 2001 From: malmert Date: Sat, 28 Feb 2026 22:11:23 +0100 Subject: [PATCH] qdfwd --- backend/src/api/nft_manager.py | 508 ++++++++++++++++----------------- 1 file changed, 239 insertions(+), 269 deletions(-) diff --git a/backend/src/api/nft_manager.py b/backend/src/api/nft_manager.py index 4fe30ea..e0cd803 100644 --- a/backend/src/api/nft_manager.py +++ b/backend/src/api/nft_manager.py @@ -91,6 +91,7 @@ class NftManager: cmd = f"list chain {family} {table} {chain}" # Attempt to temporarily disable JSON output on the wrapper (best-effort). json_toggled = False + res = {"rc": -1, "stdout": "", "stderr": "unknown"} try: if hasattr(self.nft, "set_json_output"): try: @@ -325,285 +326,73 @@ def parse_priority(val: Any) -> Optional[int]: return None -# ------------------------- -# Expr serializer helpers -# ------------------------- -def _compact_json_fragment(obj: Any) -> str: +def rule_text_from_expr(expr: Any) -> str: """ - Return a very compact JSON fragment for unknown tokens to include inline in text - (keeps text deterministic and safe). - """ - try: - return json.dumps(obj, separators=(",", ":"), ensure_ascii=False) - except Exception: - return str(obj) - - -def _stringify_value(v: Any) -> str: - """ - Convert RHS values to a deterministic textual form: - - strings -> raw - - numbers -> str - - list/sets -> "{a,b,c}" - - dict with 'start'/'end' -> "start-end" (range style) - - boolean -> "true"/"false" - """ - if v is None: - return "None" - if isinstance(v, bool): - return "true" if v else "false" - if isinstance(v, (int, float)): - # preserve integer appearance if possible - if isinstance(v, int) or float(v).is_integer(): - return str(int(v)) - return str(v) - if isinstance(v, str): - return v - if isinstance(v, (list, tuple, set)): - inner = ",".join(sorted(map(str, v))) if not isinstance(v, set) else ",".join(sorted(map(str, v))) - return "{" + inner + "}" - if isinstance(v, dict): - # common NFT range shape: {"start": "10.0.0.1", "end":"10.0.0.255"} or numeric equivalent - if "start" in v and "end" in v: - return f"{_stringify_value(v['start'])}-{_stringify_value(v['end'])}" - # fallback: compact fragment - return _compact_json_fragment(v) - return str(v) - - -def _render_match(m: Dict[str, Any]) -> Optional[str]: - """ - Render a 'match' dict into textual piece, best-effort. - Supports: - - payload left/right equality: {'left': {'payload': {'protocol':'ip','field':'protocol'}}, 'op':'==', 'right': 'icmp'} - - payload field equals port or address - - IN / not in via op 'in' or 'not in' - - ranges expressed as dict or as right 'range' - Returns None if completely unknown. - """ - if not isinstance(m, dict): - return None - - left = m.get("left") - right = m.get("right") - op = m.get("op") or m.get("operator") or m.get("type") or "==" - - # Helper: payload left - if isinstance(left, dict) and "payload" in left: - p = left["payload"] - prot = p.get("protocol") - field = p.get("field") - # if right is dict/list/str/number, stringify deterministically - rtxt = _stringify_value(right) - if prot and field: - # typical: protocol field - if field == "protocol": - # e.g. "ip protocol icmp" (right usually string) - return f"{prot} {field} {rtxt}" - # address/port fields - return f"payload({prot}.{field}) {op} {rtxt}" - - # left could be dict with 'meta' or 'ct' selectors or direct field names - if isinstance(left, dict) and "meta" in left: - # e.g. meta l4proto - mdata = left["meta"] - if isinstance(mdata, dict): - key = next(iter(mdata.keys()), None) - val = mdata.get(key) if key else None - return f"meta {key} {_stringify_value(val)}" - - # left as simple string (rare) or numeric field name - if isinstance(left, str): - return f"{left} {op} {_stringify_value(right)}" - - # Left may be a two-sided cmp e.g., {'left': {'payload':...}, 'right': {'payload':...}} - if isinstance(left, dict) and isinstance(right, dict): - # try to render both sides if they contain payloads - if "payload" in left and "payload" in right: - lp = left["payload"] - rp = right["payload"] - ltxt = f"{lp.get('protocol')}.{lp.get('field')}" if lp else _compact_json_fragment(left) - rtxt = f"{rp.get('protocol')}.{rp.get('field')}" if rp else _compact_json_fragment(right) - return f"{ltxt} {op} {rtxt}" - - # Fallback: include compact JSON fragment if we cannot deterministically render - return f"match {op} {_compact_json_fragment({'left': left, 'right': right})}" - - -def _serialize_expr(expr: Any) -> Optional[str]: - """ - Robust serializer for an nft JSON expr (list) -> textual fragment. - Returns a string (possibly verbose) or None if totally unsupported. - The intent is to produce deterministic, readable text for the UI. + Deterministic serializer to produce a compact UI-friendly string from expr list. + Covers common constructs; falls back to JSON dump for unknown constructs. + (Used for display in GET /rules). """ if expr is None: return "" - if isinstance(expr, str): - return expr - if not isinstance(expr, list): - # unsupported top-level type -> pretty-print compact - return _compact_json_fragment(expr) - - tokens: List[str] = [] - for el in expr: - # handle simple dict tokens - if isinstance(el, dict): - # direct known keywords - if "drop" in el: - tokens.append("drop") - continue - if "accept" in el: - tokens.append("accept") - continue - if "counter" in el: - tokens.append("counter") - continue - if "return" in el: - tokens.append("return") - continue - if "reject" in el: - # reject may be a string reason or dict - rv = el.get("reject") - if isinstance(rv, str): - tokens.append(f"reject {rv}") - elif isinstance(rv, dict): - tokens.append(f"reject {_compact_json_fragment(rv)}") - else: - tokens.append("reject") - continue - - # queue may be int/string or dict - if "queue" in el: - q = el["queue"] - tok = "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") - if num is not None: - tok += f" num {num}" - if q.get("bypass"): - tok += " bypass" - elif isinstance(q, (int, float)): - tok += f" num {int(q)}" - elif isinstance(q, str): - tok += f" num {q}" - tokens.append(tok) - continue - - # match token - if "match" in el: - try: - rendered = _render_match(el["match"]) - if rendered is None: - tokens.append(_compact_json_fragment(el)) - else: - tokens.append(rendered) - except Exception: - tokens.append(_compact_json_fragment(el)) - continue - - # payload shorthand - if "payload" in el: - p = el["payload"] - if isinstance(p, dict): + if isinstance(expr, list): + tokens: List[str] = [] + for part in expr: + if isinstance(part, dict): + # common tokens + if "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): + p = left["payload"] + prot = p.get("protocol") + field = p.get("field") + if prot and field: + tokens.append(f"{prot} {field} {right}") + continue + tokens.append("match") + elif "payload" in part: + p = part["payload"] prot = p.get("protocol") field = p.get("field") if prot and field: tokens.append(f"payload({prot}.{field})") continue - tokens.append(_compact_json_fragment(el)) - continue - - # tcp/udp nested objects e.g. {"tcp": {"dport": 22}} or {"tcp": {"flags":{"syn": True}}} - if "tcp" in el or "udp" in el: - proto = "tcp" if "tcp" in el else "udp" - val = el.get(proto) - if isinstance(val, dict): - # dport/sport - if "dport" in val: - tokens.append(f"{proto} dport {_stringify_value(val['dport'])}") - continue - if "sport" in val: - tokens.append(f"{proto} sport {_stringify_value(val['sport'])}") - continue - # flags - if "flags" in val: - flags = val.get("flags") - if isinstance(flags, (list, tuple)): - tokens.append(f"{proto} flags {{{','.join(map(str, flags))}}}") - else: - tokens.append(f"{proto} { _compact_json_fragment(val) }") - continue - tokens.append(proto) - continue - - # cmp / binary / bitwise — attempt to render if shape known - if "cmp" in el or "binary" in el or "bitwise" in el: - # try to compose a readable fragment - tokens.append(_compact_json_fragment(el)) - continue - - # named set membership e.g. {"in": {"left": ..., "right": ...}} or op in match - # fallback: include compact JSON fragment - tokens.append(_compact_json_fragment(el)) - continue - - # non-dict tokens (strings/numbers) - tokens.append(str(el)) - - return " ".join(tokens).strip() + 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: + tokens.append("accept") + elif "counter" in part: + tokens.append("counter") + elif "tcp" in part or "udp" in part: + proto = "tcp" if "tcp" in part else "udp" + tokens.append(proto) + elif "queue" in part: + q = part["queue"] + token = "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") + if num is not None: + token += f" num {num}" + if q.get("bypass"): + token += " bypass" + elif isinstance(q, (int, float)): + token += f" num {int(q)}" + elif isinstance(q, str): + token += f" num {q}" + tokens.append(token) + else: + keys = "+".join(sorted(part.keys())) + tokens.append(keys) + else: + tokens.append(str(part)) + return " ".join(tokens) + return str(expr) -# ------------------------- -# Existing helpers (now use _serialize_expr) -# ------------------------- -def rule_text_from_expr(expr: Any) -> str: - """ - Deterministic serializer to produce a compact UI-friendly string from expr list. - Uses the robust _serialize_expr and guarantees a string result (never None). - """ - try: - rendered = _serialize_expr(expr) - if rendered is None: - # as a last resort, dump compact JSON - return _compact_json_fragment(expr) - return rendered - except Exception: - return _compact_json_fragment(expr) - - -def expr_to_text(expr: Any) -> Optional[str]: - """ - Best-effort renderer that converts a typical nft JSON expr (list) into a textual - fragment suitable to append to 'add rule ...'. - Returns None only when the expr is clearly unsupported for textual insertion. - """ - # For create_rule_json we should be slightly stricter: if serialization produces - # a totally opaque fragment (compact JSON), we prefer to return None to force - # the caller to use the raw textual endpoint. - try: - rendered = _serialize_expr(expr) - if rendered is None: - return None - # Heuristic: if our rendering is just a compact JSON object (meaning we couldn't parse it), - # consider it unsupported (return None). - if isinstance(rendered, str) and rendered.startswith("{") and rendered.endswith("}"): - # try to be conservative: maybe it's a queue/counter JSON we can accept; allow specific tokens - try: - parsed = json.loads(rendered) - # if parsed is dict with single known action, allow it: - if any(k in parsed for k in ("drop", "accept", "queue", "counter")): - return rendered - except Exception: - pass - return None - return rendered - except Exception: - return None - - -# ------------------------- -# Build predictable ruleset (unchanged except it still uses rule_text_from_expr) -# ------------------------- def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]: """ Convert nft -j list ruleset parsed JSON into a deterministic, predictable JSON: @@ -735,6 +524,179 @@ def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]: return result +# ---------- Helpers to render expr -> textual nft (best-effort) ---------- +def expr_to_text(expr: Any) -> Optional[str]: + """ + Best-effort renderer that converts a typical nft JSON expr (list) into a textual + fragment suitable to append to 'add rule
...'. + Returns None when it cannot deterministically render the provided expr. + Supported cases (common): + - [{'match': {'left': {'payload': {'protocol':'ip','field':'protocol'}}, 'op':'==', 'right':'icmp'}}, {'drop': None}] + -> 'ip protocol icmp drop' + - payload / tcp / udp / counter / accept + - queue tokens and optional bypass support + This intentionally does not attempt to support every nft JSON construct. + """ + if expr is None: + return "" + if isinstance(expr, str): + return expr + if not isinstance(expr, list): + # unsupported top-level type + return None + + parts: List[str] = [] + for element in expr: + if isinstance(element, dict): + # handle drop/accept/counter directly + if "drop" in element: + parts.append("drop") + continue + if "accept" in element: + parts.append("accept") + continue + if "counter" in element: + parts.append("counter") + continue + + # queue support: allow {"queue": 1} or {"queue": {"num":1, "bypass": True}} etc. + if "queue" in element: + q = element["queue"] + token = "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") + if num is not None: + token += f" num {num}" + if q.get("bypass"): + token += " bypass" + elif isinstance(q, (int, float)): + token += f" num {int(q)}" + elif isinstance(q, str): + token += f" num {q}" + parts.append(token) + continue + + # match left/right payload equals -> ip protocol icmp, or ip saddr/daddr + if "match" in element: + m = element["match"] + left = m.get("left") + right = m.get("right") + # payload matches + if isinstance(left, dict) and "payload" in left and isinstance(right, (str, int)): + p = left["payload"] + prot = p.get("protocol") + field = p.get("field") + # common: protocol field match (protocol == icmp) + if prot and field and isinstance(right, str): + # ip vs ip6 decision is left to the frontend; here we render 'ip protocol icmp' (works for many setups) + if field == "protocol": + parts.append(f"{prot} {field} {right}") + continue + # payload might be l4 ports etc; produce generic payload(...) token + parts.append(f"payload({prot}.{field}) {right}") + continue + # fallback for match: try to stringify right + parts.append("match") + continue + + # payload shorthand + if "payload" in element: + p = element["payload"] + prot = p.get("protocol") + field = p.get("field") + if prot and field: + parts.append(f"payload({prot}.{field})") + continue + parts.append("payload") + continue + + # tcp/udp as nested dicts sometimes appear + if "tcp" in element or "udp" in element: + proto = "tcp" if "tcp" in element else "udp" + val = element.get(proto) + # attempt to detect dport/sport keys + if isinstance(val, dict): + if "dport" in val: + parts.append(f"{proto} dport {val['dport']}") + continue + if "sport" in val: + parts.append(f"{proto} sport {val['sport']}") + continue + parts.append(proto) + continue + + # cmp/binary operators etc — not supported deterministically + # return None to indicate we can't safely render this expr + return None + else: + # non-dict token (string/number) + parts.append(str(element)) + + # join tokens + return " ".join(parts).strip() + + +# ---------- New helper: populate_text_from_chain_text ---------- +def populate_text_from_chain_text(custom: Dict[str, Any]) -> None: + """ + Replace rule['text'] in the 'custom' predictable ruleset with the exact textual + rule lines as produced by `nft list chain
` when possible. + + This modifies `custom` in-place. If textual listing for a chain fails, we fall + back to the existing rule['text'] that was produced from JSON. + """ + tables = custom.get("tables") or [] + for t in tables: + fam = t.get("family") + tname = t.get("name") + if not fam or not tname: + continue + for ch in t.get("chains", []): + cname = ch.get("name") + if not cname: + continue + try: + chain_text = mgr.list_chain_text(fam, tname, cname) or "" + lines = [ln.rstrip() for ln in chain_text.splitlines() if ln.strip() != ""] + # build handle -> line map + handle_map: Dict[str, str] = {} + for ln in lines: + m = re.search(r"\bhandle\s+(\d+)\b", ln) + if m: + handle_map[m.group(1)] = ln.strip() + + for rule in ch.get("rules", []): + replaced = False + h = rule.get("handle") + if h is not None: + key = str(h) + if key in handle_map: + rule["text"] = handle_map[key] + replaced = True + + if not replaced: + # fallback: try to find a line that contains the JSON-derived compact text fragment + expr = rule.get("expr") + probe = rule.get("text") or rule_text_from_expr(expr) + if probe: + # try longest-first strategy (not strictly necessary here) — simple substring match + for ln in lines: + if probe in ln: + rule["text"] = ln.strip() + replaced = True + break + # if still not replaced, keep existing rule["text"] + 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, + ) + continue + + # ---------- Routes ---------- @router.get("/rules", response_model=RulesetOut, summary="List ruleset") @@ -757,6 +719,15 @@ def list_rules(): return RulesetOut(ruleset=text.strip() if text is not None else None) 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. + try: + populate_text_from_chain_text(custom) + except Exception as e: + logger.debug("list_rules: populate_text_from_chain_text failed: %s", e) + ruleset_model = RulesetModel.parse_obj(custom) return RulesetOut(ruleset=ruleset_model) except NftError as e: @@ -861,7 +832,6 @@ def create_rule_json(req: CreateRuleRequest): # with detail that includes stderr and the executed cmd. detail = f"nft command failed rc={rc}. stderr: {stderr!r}. cmd: {cmd}" logger.warning("create_rule_json failed: %s", detail) - # Return an HTTPException with the detail (frontend can still inspect error.response.data if ExecResult was included) raise HTTPException(status_code=400, detail=detail) except NftError as e: