From 96cf5d573ca472e1fbfc4682320a9836890360b1 Mon Sep 17 00:00:00 2001 From: malmert Date: Sun, 11 Jan 2026 17:48:06 +0100 Subject: [PATCH] qwfq --- backend/src/api/nft_api.py | 269 +++++++++++++++++++++++-------------- 1 file changed, 169 insertions(+), 100 deletions(-) diff --git a/backend/src/api/nft_api.py b/backend/src/api/nft_api.py index df141f6..284a9bf 100644 --- a/backend/src/api/nft_api.py +++ b/backend/src/api/nft_api.py @@ -1,15 +1,11 @@ # fastapi_nft_router.py # -*- coding: utf-8 -*- """ -FastAPI router for nftables (bridge family). Returns both JSON and -human-readable textual representations for rules so the frontend can -display exact nft textual lines and programmatically safe metadata. +FastAPI router for nftables (bridge family). Produces both JSON and +human-readable textual representations for rules. -Features: -- nft_list_rules returns nft_rule_text_full, nft_rule_text (no handle), and add_command per rule. -- Robust mapping by handle using both JSON and textual chain dump. -- Fallback JSON->text renderer for cases where textual mapping is missing. -- Automatic create table/chain if missing, logging, and typed models for frontend. +This version improves the JSON->text fallback so common shapes like +'ip protocol icmp drop' are rendered as nft-style clauses. """ from typing import Any, Dict, List, Optional, Union, Literal import subprocess @@ -188,7 +184,6 @@ def ensure_nft_available() -> None: raise HTTPException(status_code=500, detail="nft binary not found on server") def run_nft_cmd(cmd: str) -> Dict[str, Any]: - """Run `nft -f -` with the given single-line script (ensures newline).""" ensure_nft_available() full_cmd = [NFT_BIN, "-f", "-"] script = cmd.rstrip() + "\n" @@ -209,7 +204,6 @@ def run_nft_cmd(cmd: str) -> Dict[str, Any]: raise HTTPException(status_code=500, detail=err) def ensure_table_and_chain_exist(family: str, table: str, chain: str) -> None: - """Detect and create table/chain if missing (conservative defaults).""" logger.debug("Checking/existence for family=%s table=%s chain=%s", family, table, chain) ensure_nft_available() try: @@ -250,10 +244,6 @@ def ensure_table_and_chain_exist(family: str, table: str, chain: str) -> None: HANDLE_RE = re.compile(r"\bhandle\s+(\d+)\b", flags=re.IGNORECASE) def nft_list_chain_text(family: str, table: str, chain: str) -> Dict[int, str]: - """ - Return mapping handle -> full textual line from: - nft list chain - """ ensure_nft_available() cmd = [NFT_BIN, "list", "chain", family, table, chain] logger.debug("Running textual chain list: %s", " ".join(cmd)) @@ -261,7 +251,6 @@ def nft_list_chain_text(family: str, table: str, chain: str) -> Dict[int, str]: out = subprocess.check_output(cmd, stderr=subprocess.PIPE).decode() except subprocess.CalledProcessError as e: logger.error("Failed textual chain list: %s", e.stderr.decode()) - # bubble up - caller may fallback; raising is acceptable here raise HTTPException(status_code=500, detail=e.stderr.decode()) mapping: Dict[int, str] = {} @@ -274,7 +263,6 @@ def nft_list_chain_text(family: str, table: str, chain: str) -> Dict[int, str]: continue try: handle = int(m.group(1)) - # full line including handle token mapping[handle] = s logger.debug("Text mapping: handle=%d -> %s", handle, s) except Exception as ex: @@ -282,12 +270,55 @@ def nft_list_chain_text(family: str, table: str, chain: str) -> Dict[int, str]: continue return mapping +def extract_protocol_from_expr(expr: Any) -> Optional[str]: + """ + Attempt to detect a protocol (tcp/udp/icmp) from a single expr JSON object. + It searches common shapes ('payload', 'protocol', 'icmp') recursively. + Returns protocol string (e.g. 'icmp') or None. + """ + if not isinstance(expr, dict): + return None + # direct icmp key + if "icmp" in expr: + return "icmp" + # payload shapes: walk nested dicts looking for 'icmp' or 'tcp'/'udp' + def walk(d): + if isinstance(d, dict): + for k, v in d.items(): + if isinstance(v, str): + if v.lower() in ("icmp", "tcp", "udp"): + return v.lower() + if isinstance(v, dict) or isinstance(v, list): + found = walk(v) + if found: + return found + elif isinstance(d, list): + for item in d: + found = walk(item) + if found: + return found + return None + # check common keys + for key in ("payload", "protocol", "ip", "meta"): + if key in expr: + found = walk(expr[key]) + if found: + return found + # last resort: scan whole expr + return walk(expr) + def json_exprs_to_text(exprs: List[Any]) -> str: """ - Build a best-effort single-line textual rule clause from nft JSON exprs. - Used as a fallback when textual chain dump doesn't include the handle mapping. + Best-effort conversion of nft JSON exprs to a human-readable nft-style + single-line clause. We aim to produce phrases like: + ip protocol icmp drop + meta iifname \"eth0\" accept + ct state established accept + The function collects match fragments then appends verdict/action at the end. """ - parts: List[str] = [] + matches: List[str] = [] + verdicts: List[str] = [] + for ex in exprs: if not isinstance(ex, dict): continue @@ -295,53 +326,68 @@ def json_exprs_to_text(exprs: List[Any]) -> str: if "comment" in ex: c = ex["comment"] if isinstance(c, str): - parts.append(f'comment "{c}"') + matches.append(f'comment "{c}"') elif isinstance(c, dict): txt = c.get("text") or c.get("str") if txt: - parts.append(f'comment "{txt}"') + matches.append(f'comment "{txt}"') continue - # verdict + # verdict shapes if "verdict" in ex: v = ex["verdict"] if isinstance(v, dict): k = next(iter(v.keys()), None) - parts.append(k if k else "verdict") + if k: + verdicts.append(k) else: - parts.append(str(v)) + verdicts.append(str(v)) continue if "drop" in ex: - parts.append("drop") + verdicts.append("drop") continue if "accept" in ex: - parts.append("accept") + verdicts.append("accept") continue - # match shapes + if "reject" in ex: + verdicts.append("reject") + continue + # conntrack + if "ct" in ex: + ct = ex["ct"] + if isinstance(ct, dict): + # prefer printed form 'ct state established' + if "state" in ct: + matches.append(f"ct state {ct['state']}") + else: + for k, v in ct.items(): + matches.append(f"ct {k} {v}") + continue + # meta + if "meta" in ex: + meta = ex["meta"] + if isinstance(meta, dict): + key = meta.get("key") or meta.get("name") + op = meta.get("op", "==") + val = meta.get("value") + if key and val is not None: + matches.append(f"meta {key} {op} {val}") + continue + # match object with left/op/right if "match" in ex: m = ex["match"] left = m.get("left") op = m.get("op") right = m.get("right") if left and op and (right is not None): - parts.append(f"{left} {op} {right}") + matches.append(f"{left} {op} {right}") continue - if "meta" in ex: - meta = ex["meta"] - key = meta.get("key") or meta.get("name") - op = meta.get("op", "==") - val = meta.get("value") - if key and val is not None: - parts.append(f"meta {key} {op} {val}") - continue - if "ct" in ex: - ct = ex["ct"] - if isinstance(ct, dict): - for k, v in ct.items(): - parts.append(f"ct {k} {v}") - continue - if "payload" in ex: - parts.append(json.dumps(ex["payload"])) + # payload/protocol detection -> render 'ip protocol icmp' + proto = extract_protocol_from_expr(ex) + if proto: + # only render once per expr; protocol is a match, not action + matches.append(f"ip protocol {proto}") continue + # log if "log" in ex: lg = ex["log"] piece = "log" @@ -349,30 +395,61 @@ def json_exprs_to_text(exprs: List[Any]) -> str: if lg.get("prefix"): piece += f' prefix "{lg.get("prefix")}"' if lg.get("group") is not None: - piece += f' group {lg.get("group")}' - parts.append(piece) + piece += f" group {lg.get('group')}" + matches.append(piece) continue - # fallback: compact JSON - parts.append(json.dumps(ex)) - return " ".join(parts) + # payload fallback: compact it + if "payload" in ex: + matches.append(json.dumps(ex["payload"])) + continue + # unknown: compact JSON + matches.append(json.dumps(ex)) + + # join matches then verdict(s) + clause = " ".join(matches).strip() + if clause and verdicts: + clause = f"{clause} {' '.join(verdicts)}" + elif not clause and verdicts: + clause = " ".join(verdicts) + return clause.strip() # ---------- JSON rules parsing with textual injection ------------------ -def nft_list_rules(table: str = DEFAULT_TABLE, chain: str = DEFAULT_CHAIN) -> Dict[str, Any]: - """ - Return structured rules with both JSON exprs and textual representations. +def nft_list_chain_text_map(family: str, table: str, chain: str) -> Dict[int, str]: + """Thin wrapper to get textual mapping, returns empty mapping on failure.""" + try: + return nft_list_chain_text(family, table, chain) + except HTTPException as e: + logger.debug("text map unavailable: %s", getattr(e, "detail", str(e))) + return {} - Each rule entry contains: - - family, table, chain - - handle - - position (1-based) - - comment (best-effort) - - verdict (best-effort) - - verdict_details - - exprs (JSON) - - nft_rule_text_full (exact textual line from `nft list chain ...`, includes 'handle N' if present) - - nft_rule_text (textual clause without trailing 'handle N') - - add_command (best-effort `add rule
...` command) - """ +def nft_list_chain_text(family: str, table: str, chain: str) -> Dict[int, str]: + ensure_nft_available() + cmd = [NFT_BIN, "list", "chain", family, table, chain] + logger.debug("Running textual chain list: %s", " ".join(cmd)) + try: + out = subprocess.check_output(cmd, stderr=subprocess.PIPE).decode() + except subprocess.CalledProcessError as e: + logger.error("Failed textual chain list: %s", e.stderr.decode()) + raise HTTPException(status_code=500, detail=e.stderr.decode()) + + mapping: Dict[int, str] = {} + for line in out.splitlines(): + s = line.strip() + if not s: + continue + m = HANDLE_RE.search(s) + if not m: + continue + try: + handle = int(m.group(1)) + mapping[handle] = s + logger.debug("Text mapping: handle=%d -> %s", handle, s) + except Exception as ex: + logger.debug("Failed parsing handle from line: %s (%s)", s, ex) + continue + return mapping + +def nft_list_rules(table: str = DEFAULT_TABLE, chain: str = DEFAULT_CHAIN) -> Dict[str, Any]: ensure_nft_available() try: out = subprocess.check_output([NFT_BIN, "--json", "list", "ruleset"], stderr=subprocess.PIPE) @@ -381,13 +458,9 @@ def nft_list_rules(table: str = DEFAULT_TABLE, chain: str = DEFAULT_CHAIN) -> Di logger.error("Failed to list ruleset (json): %s", e.stderr.decode()) raise HTTPException(status_code=500, detail=e.stderr.decode()) - # get textual mapping for the chain (may raise; catch and fallback) - text_map: Dict[int, str] = {} - try: - text_map = nft_list_chain_text(DEFAULT_FAMILY, table, chain) - logger.debug("Obtained textual mapping with %d entries", len(text_map)) - except HTTPException as e: - logger.debug("Unable to get textual chain dump: %s; will fallback per-rule", getattr(e, "detail", str(e))) + # Try to obtain text map, but do not fail if unavailable + text_map = nft_list_chain_text_map(DEFAULT_FAMILY, table, chain) + logger.debug("Text map entries: %d", len(text_map)) results: List[Dict[str, Any]] = [] counters: Dict[str, int] = {} @@ -442,27 +515,29 @@ def nft_list_rules(table: str = DEFAULT_TABLE, chain: str = DEFAULT_CHAIN) -> Di if "reject" in expr and verdict is None: verdict = "reject" - # textual resolution: prefer exact mapping by handle - full_text: Optional[str] = None - if handle is not None: - full_text = text_map.get(handle) - if full_text: - # derive no-handle version by stripping final ' handle N' if present - m = HANDLE_RE.search(full_text) - if m: - no_handle_text = full_text[: m.start()].strip() - else: - no_handle_text = full_text - logger.debug("Mapped handle %s -> textual full='%s'", handle, full_text) - else: - # fallback: build readable text from exprs - no_handle_text = json_exprs_to_text(exprs) - full_text = (no_handle_text + f" handle {handle}") if handle is not None else no_handle_text - logger.debug("Fallback textual for handle %s: %s", handle, no_handle_text) + # textual resolution: prefer exact mapping by handle, fallback to JSON->text + nft_rule_text_full: Optional[str] = None + nft_rule_text: Optional[str] = None - # build an add_command (best-effort) - uses add rule
- add_cmd_clause = no_handle_text or "" - add_command = f"add rule {table_name} {chain_name} {add_cmd_clause}".strip() + if handle is not None and handle in text_map: + nft_rule_text_full = text_map[handle] + # strip trailing ' handle N' to produce no-handle variant + m = HANDLE_RE.search(nft_rule_text_full) + if m: + nft_rule_text = nft_rule_text_full[: m.start()].strip() + else: + nft_rule_text = nft_rule_text_full + logger.debug("Using textual map for handle %s -> %s", handle, nft_rule_text_full) + else: + # fallback: build from exprs + clause = json_exprs_to_text(exprs) + nft_rule_text = clause or None + nft_rule_text_full = (clause + (f" handle {handle}" if handle is not None else "")) if clause else None + logger.debug("Fallback clause for handle %s -> %s", handle, clause) + + # build add_command: use add rule
+ add_clause = nft_rule_text or "" + add_command = f"add rule {table_name} {chain_name} {add_clause}".strip() results.append({ "family": family, @@ -474,9 +549,9 @@ def nft_list_rules(table: str = DEFAULT_TABLE, chain: str = DEFAULT_CHAIN) -> Di "verdict": verdict, "verdict_details": verdict_details, "exprs": exprs, - "nft_rule_text_full": full_text, # e.g. 'meta iifname "eth0" accept comment "x" handle 7' - "nft_rule_text": no_handle_text, # e.g. 'meta iifname "eth0" accept comment "x"' - "add_command": add_command, # e.g. 'add rule mitm_tbl forward meta iifname "eth0" accept ...' + "nft_rule_text_full": nft_rule_text_full, # includes handle token if present + "nft_rule_text": nft_rule_text, # no-handle clause + "add_command": add_command, }) return {"rules": results} @@ -531,7 +606,6 @@ def rule_to_nft_cmd(rule: RuleModel) -> str: # ---------- Endpoints ------------------------------------------------- @router.get("/options") def get_options() -> Dict[str, Any]: - """Return allowed enum choices for frontend dropdowns.""" return { "family": [f.value for f in Family], "table": [t.value for t in Table], @@ -550,13 +624,11 @@ def get_options() -> Dict[str, Any]: @router.get("/rules") def list_rules(table: str = DEFAULT_TABLE, chain: str = DEFAULT_CHAIN) -> Dict[str, Any]: - """List rules for a table/chain (creates table/chain if missing).""" ensure_table_and_chain_exist(DEFAULT_FAMILY, table, chain) return nft_list_rules(table=table, chain=chain) @router.post("/rules/preview") def preview_rule(rule: RuleModel = Body(...)) -> Dict[str, str]: - """Return nft command that would be executed for the provided rule (no-op).""" try: cmd = rule_to_nft_cmd(rule) except Exception as e: @@ -566,21 +638,18 @@ def preview_rule(rule: RuleModel = Body(...)) -> Dict[str, str]: @router.post("/rules") def add_rule(rule: RuleModel = Body(...)) -> Dict[str, Any]: - """Insert or append the rule; creates table/chain if missing.""" ensure_table_and_chain_exist(rule.family.value, rule.table.value, rule.chain.value) cmd = rule_to_nft_cmd(rule) return run_nft_cmd(cmd) @router.delete("/rules/{handle}") def delete_rule(handle: int, table: str = DEFAULT_TABLE, chain: str = DEFAULT_CHAIN) -> Dict[str, Any]: - """Delete a rule by handle.""" ensure_table_and_chain_exist(DEFAULT_FAMILY, table, chain) cmd = f"delete rule {table} {chain} handle {handle}" return run_nft_cmd(cmd) @router.put("/rules/{handle}") def update_rule(handle: int, rule: RuleModel = Body(...), table: str = DEFAULT_TABLE, chain: str = DEFAULT_CHAIN) -> Dict[str, Any]: - """Replace a rule by handle: delete by handle then insert at same position (if known).""" ensure_table_and_chain_exist(rule.family.value, rule.table.value, rule.chain.value) rules_info = nft_list_rules(table=table, chain=chain) position: Optional[int] = None