diff --git a/backend/src/api/nft_manager.py b/backend/src/api/nft_manager.py index 06ef32c..614ea26 100644 --- a/backend/src/api/nft_manager.py +++ b/backend/src/api/nft_manager.py @@ -240,70 +240,6 @@ def parse_priority(val: Any) -> Optional[int]: return None -def rule_text_from_expr(expr: Any) -> str: - if expr is None: - return "" - # try list first (common) - if isinstance(expr, list): - tokens: List[str] = [] - for part in expr: - if isinstance(part, dict): - 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, int)): - 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("payload") - 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) - # if expr is a dict or other type, give a compact repr - try: - return str(expr) - except Exception: - return "" - - def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]: """ Convert nft -j list ruleset parsed JSON into a deterministic, predictable JSON: @@ -366,7 +302,7 @@ def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]: chains_map.setdefault(chain_name, {"name": chain_name, "type": None, "hook": None, "priority": None, "policy": None, "rules": []}) # do NOT change expr shape here; keep it exactly as NFT JSON provided - rule_obj: Dict[str, Any] = {"handle": handle, "expr": expr, "text": rule_text_from_expr(expr)} + rule_obj: Dict[str, Any] = {"handle": handle, "expr": expr, "text": ""} if "position" in r: rule_obj["position"] = r["position"] if "comment" in r: @@ -511,7 +447,7 @@ def populate_text_from_ruleset_text(custom: Dict[str, Any], nft_text: str) -> No # final substring probe (safe) if mapped is None: - probe = rule.get("text") or rule_text_from_expr(rule.get("expr")) + probe = rule.get("text") if probe: for ln in ordered_lines: if probe in ln: @@ -526,56 +462,6 @@ def populate_text_from_ruleset_text(custom: Dict[str, Any], nft_text: str) -> No rule["text"] = mapped # should be str already -def populate_text_from_chain_text(custom: Dict[str, Any]) -> None: - """ - Per-chain fallback that queries nft for the chain's textual listing and - updates only rule['text'] when a reliable mapping is found. - """ - for t in custom.get("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() != ""] - handle_map: Dict[int, str] = {} - for ln in lines: - m = re.search(r"\bhandle\s+(\d+)\b", ln) - if m: - try: - handle_map[int(m.group(1))] = ln.strip() - except Exception: - continue - - rules = ch.get("rules", []) - for idx, rule in enumerate(rules): - mapped: Optional[str] = None - h = rule.get("handle") - if isinstance(h, int) and h in handle_map: - mapped = handle_map[h] - else: - probe = rule.get("text") or rule_text_from_expr(rule.get("expr")) - if probe: - for ln in lines: - if probe in ln: - mapped = ln.strip() - break - - if mapped is not None: - try: - rule["text"] = str(mapped) - except Exception: - rule["text"] = mapped - except Exception as e: - logger.debug("populate_text_from_chain_text failed for %s %s %s: %s", fam, tname, cname, e) - continue - - # ---------- Normalization helper (lightweight and safe) ---------- def normalize_custom_for_model(custom: Dict[str, Any]) -> None: """ @@ -595,9 +481,6 @@ def normalize_custom_for_model(custom: Dict[str, Any]) -> None: # text: ensure string if "text" not in r or r.get("text") is None: - try: - r["text"] = rule_text_from_expr(r.get("expr")) - except Exception: r["text"] = "" else: if not isinstance(r["text"], str): @@ -654,12 +537,6 @@ def list_rules(): except Exception as e: logger.debug("populate_text_from_ruleset_text failed: %s", e) - # Per-chain fallback using `nft list chain ...` - try: - populate_text_from_chain_text(custom) - except Exception as e: - logger.debug("populate_text_from_chain_text failed: %s", e) - # Normalize minimally for model validation normalize_custom_for_model(custom) @@ -687,81 +564,6 @@ def list_rules(): raise HTTPException(status_code=500, detail=str(e)) -@router.post( - "/rules", - response_model=ExecResult, - status_code=status.HTTP_201_CREATED, - summary="Create rule (JSON, expr required; returns ExecResult with rc/stdout/stderr)", -) -def create_rule_json(req: CreateRuleRequest): - try: - family = req.family - table = req.table - chain = req.chain - - if req.expr is None: - raise NftError("field 'expr' is required for JSON rule creation") - - rendered = rule_text_from_expr(req.expr) - if rendered is None or rendered == "": - raise NftError( - "cannot render provided 'expr' to textual nft syntax. Use POST /firewall/raw" - ) - - expr_text = rendered.strip() - - if req.position is not None: - try: - pos = int(req.position) - if pos < 0: - pos = 0 - except Exception: - pos = 0 - cmd = f"insert rule {family} {table} {chain} position {pos} {expr_text}" - else: - cmd = f"add rule {family} {table} {chain} {expr_text}" - - logger.info("create_rule_json executing: %s", cmd) - res = mgr.cmd(cmd) - raw_rc = res.get("rc") - stdout = res.get("stdout") or "" - stderr = res.get("stderr") or "" - - try: - rc = int(raw_rc) - except Exception: - rc = -1 - - logger.info("nft cmd rc=%s stdout=%r stderr=%r cmd=%s", rc, stdout, stderr, cmd) - - exec_res = ExecResult(rc=rc, stdout=stdout or None, stderr=stderr or None) - - if rc == 0: - return exec_res - - # If non-zero rc but empty stderr, try to verify presence - if (rc < 0 or rc != 0) and stderr.strip() == "": - try: - chain_text = mgr.list_chain_text(family, table, chain) or "" - if expr_text and expr_text in chain_text: - logger.info("create_rule_json: rule detected after add; treating as success") - return ExecResult(rc=0, stdout=stdout or None, stderr=stderr or None) - except Exception as e_chain: - logger.warning("create_rule_json verification failed: %s", e_chain) - - detail = f"nft command failed rc={rc}. stderr: {stderr!r}. cmd: {cmd}" - logger.warning("create_rule_json failed: %s", detail) - raise HTTPException(status_code=400, detail=detail) - except NftError as e: - logger.warning("create_rule_json NftError: %s", e) - raise HTTPException(status_code=400, detail=str(e)) - except HTTPException: - raise - except Exception as e: - logger.exception("create_rule_json internal error") - raise HTTPException(status_code=500, detail=str(e)) - - @router.delete("/rules/{handle}", status_code=status.HTTP_204_NO_CONTENT, summary="Delete rule by handle") def delete_rule(handle: int, family: str = "inet", table: str = "filter", chain: str = "input"): try: