qwfq

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
+FastAPI router for nftables (bridge family). Produces both JSON and
-human-readable textual representations for rules so the frontend can
+human-readable textual representations for rules.
-display exact nft textual lines and programmatically safe metadata.
 
-Features:
+This version improves the JSON->text fallback so common shapes like
-- nft_list_rules returns nft_rule_text_full, nft_rule_text (no handle), and add_command per rule.
+'ip protocol icmp drop' are rendered as nft-style clauses.
-- 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.
 """
 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 <family> <table> <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.
+    Best-effort conversion of nft JSON exprs to a human-readable nft-style
-    Used as a fallback when textual chain dump doesn't include the handle mapping.
+    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:
+        # payload/protocol detection -> render 'ip protocol icmp'
-            meta = ex["meta"]
+        proto = extract_protocol_from_expr(ex)
-            key = meta.get("key") or meta.get("name")
+        if proto:
-            op = meta.get("op", "==")
+            # only render once per expr; protocol is a match, not action
-            val = meta.get("value")
+            matches.append(f"ip protocol {proto}")
-            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"]))
             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")}'
+                    piece += f" group {lg.get('group')}"
-            parts.append(piece)
+            matches.append(piece)
             continue
-        # fallback: compact JSON
+        # payload fallback: compact it
-        parts.append(json.dumps(ex))
+        if "payload" in ex:
-    return " ".join(parts)
+            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]:
+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."""
-    Return structured rules with both JSON exprs and textual representations.
+    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:
+def nft_list_chain_text(family: str, table: str, chain: str) -> Dict[int, str]:
-      - family, table, chain
+    ensure_nft_available()
-      - handle
+    cmd = [NFT_BIN, "list", "chain", family, table, chain]
-      - position (1-based)
+    logger.debug("Running textual chain list: %s", " ".join(cmd))
-      - comment (best-effort)
+    try:
-      - verdict (best-effort)
+        out = subprocess.check_output(cmd, stderr=subprocess.PIPE).decode()
-      - verdict_details
+    except subprocess.CalledProcessError as e:
-      - exprs (JSON)
+        logger.error("Failed textual chain list: %s", e.stderr.decode())
-      - nft_rule_text_full (exact textual line from `nft list chain ...`, includes 'handle N' if present)
+        raise HTTPException(status_code=500, detail=e.stderr.decode())
-      - nft_rule_text (textual clause without trailing 'handle N')
+
-      - add_command (best-effort `add rule <table> <chain> ...` command)
+    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)
+    # Try to obtain text map, but do not fail if unavailable
-    text_map: Dict[int, str] = {}
+    text_map = nft_list_chain_text_map(DEFAULT_FAMILY, table, chain)
-    try:
+    logger.debug("Text map entries: %d", len(text_map))
-        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)))
 
     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
+        # textual resolution: prefer exact mapping by handle, fallback to JSON->text
-        full_text: Optional[str] = None
+        nft_rule_text_full: Optional[str] = None
-        if handle is not None:
+        nft_rule_text: Optional[str] = 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)
 
-        # build an add_command (best-effort) - uses add rule <table> <chain> <clause>
+        if handle is not None and handle in text_map:
-        add_cmd_clause = no_handle_text or ""
+            nft_rule_text_full = text_map[handle]
-        add_command = f"add rule {table_name} {chain_name} {add_cmd_clause}".strip()
+            # 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 <table> <chain> <clause>
+        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_full": nft_rule_text_full,   # includes handle token if present
-            "nft_rule_text": no_handle_text,   # e.g. 'meta iifname "eth0" accept comment "x"'
+            "nft_rule_text": nft_rule_text,             # no-handle clause
-            "add_command": add_command,        # e.g. 'add rule mitm_tbl forward meta iifname "eth0" accept ...'
+            "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