diff --git a/backend/src/api/nft_manager.py b/backend/src/api/nft_manager.py index d027089..572602c 100644 --- a/backend/src/api/nft_manager.py +++ b/backend/src/api/nft_manager.py @@ -175,39 +175,39 @@ mgr = NftManager() # ---------- Models ---------- class RawCmdRequest(BaseModel): - cmd: str = Field(...) + cmd: str = Field(..., description="Textual nft command to execute") class ExecResult(BaseModel): - rc: int - stdout: Optional[str] - stderr: Optional[str] + rc: int = Field(..., description="Return code from nft execution") + stdout: Optional[str] = Field(None) + stderr: Optional[str] = Field(None) class RuleOut(BaseModel): - handle: Optional[int] - expr: Any - text: str - position: Optional[Any] - comment: Optional[str] + handle: Optional[int] = Field(None) + expr: Any = Field(..., description="Machine-readable nft expression (original nft JSON expr).") + text: str = Field(..., description="Deterministic short display string derived from expr") + position: Optional[Any] = Field(None) + comment: Optional[str] = Field(None) class ChainOut(BaseModel): - name: str - type: Optional[str] - hook: Optional[str] - priority: Optional[int] - policy: Optional[str] - rules: List[RuleOut] + name: str = Field(...) + type: Optional[str] = Field(None) + hook: Optional[str] = Field(None) + priority: Optional[int] = Field(None) + policy: Optional[str] = Field(None) + rules: List[RuleOut] = Field(...) class TableOut(BaseModel): - family: str - name: str - chains: List[ChainOut] + family: str = Field(...) + name: str = Field(...) + chains: List[ChainOut] = Field(...) class RulesetModel(BaseModel): - tables: List[TableOut] + tables: List[TableOut] = Field(...) class CreateRuleRequest(BaseModel): @@ -229,6 +229,7 @@ class RulesetOut(BaseModel): # ---------- Helpers ---------- _handle_re = re.compile(r"\s+#\s*handle\s+\d+\s*$") + def parse_priority(val: Any) -> Optional[int]: if val is None: return None @@ -253,6 +254,7 @@ def parse_priority(val: Any) -> Optional[int]: return p return None + def rule_text_from_expr(expr: Any) -> str: if expr is None: return "" @@ -370,6 +372,7 @@ def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]: tables.setdefault((fam, table_name), {"family": fam, "name": table_name, "chains": {}}) chains_map = tables[(fam, table_name)]["chains"] chains_map.setdefault(chain_name, {"name": chain_name, "type": None, "hook": None, "priority": None, "policy": None, "rules": []}) + # we keep expr exactly as provided from JSON; do not replace it rule_obj: Dict[str, Any] = {"handle": handle, "expr": expr, "text": rule_text_from_expr(expr)} if "position" in r: rule_obj["position"] = r["position"] @@ -407,7 +410,7 @@ def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]: return result -# ---------- Text parsing helpers ---------- +# ---------- Text parsing helpers (enrichment only) ---------- def parse_ruleset_text(nft_text: str) -> Dict[Tuple[str, str, str], List[Dict[str, Optional[Union[str, int]]]]]: result: Dict[Tuple[str, str, str], List[Dict[str, Optional[Union[str, int]]]]] = {} if not nft_text: @@ -478,8 +481,7 @@ def populate_text_from_ruleset_text(custom: Dict[str, Any], nft_text: str) -> No key = (fam, tname, cname) textual_entries = parsed.get(key, []) if not textual_entries: - # no textual lines for this chain; skip - continue + continue # nothing to enrich for this chain handle_map: Dict[int, str] = {} ordered_lines: List[str] = [] @@ -493,6 +495,7 @@ def populate_text_from_ruleset_text(custom: Dict[str, Any], nft_text: str) -> No rules = chain.get("rules", []) for idx, rule in enumerate(rules): replaced = False + # ensure we don't remove/overwrite expr h = rule.get("handle") if isinstance(h, int) and h in handle_map: rule["text"] = handle_map[h] @@ -516,7 +519,7 @@ def populate_text_from_ruleset_text(custom: Dict[str, Any], nft_text: str) -> No rule["text"] = ln replaced = True break - # only update when a match was found; otherwise keep JSON-derived text + # only update 'text' when a match was found; otherwise leave existing rule['text'] def populate_text_from_chain_text(custom: Dict[str, Any]) -> None: @@ -533,20 +536,21 @@ def populate_text_from_chain_text(custom: Dict[str, Any]) -> None: 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[str, str] = {} + handle_map: Dict[int, str] = {} for ln in lines: m = re.search(r"\bhandle\s+(\d+)\b", ln) if m: - handle_map[m.group(1)] = ln.strip() + try: + handle_map[int(m.group(1))] = ln.strip() + except Exception: + continue 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 isinstance(h, int) and h in handle_map: + rule["text"] = handle_map[h] + replaced = True if not replaced: expr = rule.get("expr") @@ -568,6 +572,58 @@ def populate_text_from_chain_text(custom: Dict[str, Any]) -> None: continue +# ---------- Normalization helper to satisfy Pydantic model ---------- +def normalize_custom_for_model(custom: Dict[str, Any]) -> None: + """ + Ensure every rule has required keys/types for RulesetModel parsing: + - 'expr' must exist (if missing/None -> set to []) + - 'text' must exist and be a str (if missing -> derive via rule_text_from_expr or "") + - 'handle' coerced to int or None + This mutates `custom` in-place. + """ + tables = custom.get("tables") or [] + for t in tables: + for ch in t.get("chains", []): + rules = ch.get("rules", []) or [] + for r in rules: + # expr: required by model; prefer existing value, but don't leave None + if "expr" not in r or r.get("expr") is None: + # default to empty list (acceptable Any value and consistent with prior shape) + r["expr"] = [] + + # text: must be str + txt = r.get("text") + if txt is None: + try: + r["text"] = rule_text_from_expr(r.get("expr")) + except Exception: + r["text"] = "" + else: + # coerce to str if not string + if not isinstance(txt, str): + try: + r["text"] = str(txt) + except Exception: + r["text"] = "" + + # handle: ensure int or None + h = r.get("handle") + if isinstance(h, str): + try: + r["handle"] = int(h) + except Exception: + r["handle"] = None + elif not isinstance(h, int): + # if it's e.g. float, bool, etc. convert if reasonable + try: + if isinstance(h, (float,)): + r["handle"] = int(h) + else: + r["handle"] = None + except Exception: + r["handle"] = None + + # ---------- Routes ---------- @router.get("/rules", response_model=RulesetOut, summary="List ruleset") def list_rules(): @@ -578,27 +634,33 @@ def list_rules(): logger.debug("could not obtain nft JSON ruleset: %s", e) raise HTTPException(status_code=500, detail=f"Failed to obtain nft JSON ruleset: {e}") + # get textual snapshot (best-effort), used only to enrich 'text' values nft_text = "" try: nft_text = mgr.list_rules_text() except Exception: logger.debug("could not obtain textual nft ruleset (list_rules_text failed)") + # build predictable JSON-derived structure (must be the source of truth) custom = build_predictable_ruleset(nft_json) - # Enrich rule['text'] using the full textual ruleset (handle-first, position, index) + # enrich existing JSON rules with textual lines (handle -> line; position/index fallback) try: if nft_text: populate_text_from_ruleset_text(custom, nft_text) except Exception as e: logger.debug("list_rules: populate_text_from_ruleset_text failed: %s", e) - # per-chain fallback (will not fail overall listing) + # per-chain fallback that queries nft for chain-level textual lines try: populate_text_from_chain_text(custom) except Exception as e: logger.debug("list_rules: populate_text_from_chain_text failed: %s", e) + # Normalize to satisfy Pydantic model (do this last) + normalize_custom_for_model(custom) + + # Debug: log counts num_tables = len(custom.get("tables", [])) num_rules = sum(len(ch.get("rules", [])) for t in custom.get("tables", []) for ch in t.get("chains", [])) logger.info("list_rules: returning tables=%d rules=%d", num_tables, num_rules) @@ -627,7 +689,12 @@ def create_rule_json(req: CreateRuleRequest): if req.expr is None: raise NftError("field 'expr' is required for JSON rule creation") - rendered = expr_to_text(req.expr) + rendered = None + try: + rendered = rule_text_from_expr(req.expr) + except Exception: + rendered = None + if rendered is None: raise NftError( "cannot render provided 'expr' to textual nft syntax. "