diff --git a/backend/src/api/nft_manager.py b/backend/src/api/nft_manager.py index 164b9dc..d027089 100644 --- a/backend/src/api/nft_manager.py +++ b/backend/src/api/nft_manager.py @@ -20,17 +20,8 @@ class NftError(RuntimeError): # ---------- NftManager (textual-only) ---------- class NftManager: - """ - Thin wrapper around python-nftables exposing: - - cmd execution (textual nft commands via Nftables.cmd()) - - convenience list_rules_text / list_rules_json / list_chain_text - We prefer JSON globally, but for per-chain textual listing we temporarily disable JSON - so the output matches `nft list chain ...` textual rule lines. - """ - def __init__(self) -> None: self.nft = Nftables() - # Try to prefer JSON for general listing; we'll toggle off for chain-list calls. try: self.nft.set_json_output(True) self.nft.set_handle_output(True) @@ -38,21 +29,12 @@ class NftManager: logger.debug("set_json_output not available or ignored") def cmd(self, text_cmd: str) -> Dict[str, Optional[str]]: - """ - Execute a textual nft command via Nftables.cmd(). - Returns dict { "rc": rc, "stdout": out_str, "stderr": err_str }. - """ rc, out, err = self.nft.cmd(text_cmd) if rc != 0: logger.warning("nft cmd rc=%s stderr=%s cmd=%s", rc, err, text_cmd) return {"rc": rc, "stdout": out, "stderr": err} - def list_rules_json(self) -> Dict[str, Any]: - """ - Obtain nft -j list ruleset (JSON). Returns parsed JSON dict on success. - Raises NftError on failure or when output cannot be parsed as JSON. - """ res = self.cmd("list ruleset") if res["rc"] != 0: raise NftError(f"nft list ruleset failed: {res['stderr']}") @@ -63,13 +45,8 @@ class NftManager: return json.loads(out) except json.JSONDecodeError as e: raise NftError(f"json decode error: {e}") - + def list_rules_text(self) -> str: - """ - Obtain nft list ruleset in textual form. This is a fallback if JSON parsing fails. - Returns raw textual output on success, raises NftError on failure. - """ - # best-effort toggle JSON off for textual output json_toggled = False try: if hasattr(self.nft, "set_json_output"): @@ -91,28 +68,18 @@ class NftManager: return res["stdout"] or "" def list_chain_text(self, family: str, table: str, chain: str) -> str: - """ - Return textual output of `nft list chain `. - This tries to temporarily disable JSON output so the wrapper returns the textual - representation used by `nft list ruleset`. If disabling JSON is not possible, - we attempt to parse returned JSON (as a last resort), but the preferred path is - to get textual output. - """ 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: - # Turn off JSON output to force textual output for this call. self.nft.set_json_output(False) json_toggled = True except Exception: logger.debug("could not toggle set_json_output(False); will try command anyway") res = self.cmd(cmd) finally: - # Restore JSON output preference if we toggled it. if json_toggled and hasattr(self.nft, "set_json_output"): try: self.nft.set_json_output(True) @@ -123,15 +90,11 @@ class NftManager: raise NftError(f"nft {cmd} failed: {res['stderr']}") out = res["stdout"] or "" - # If the output looks like JSON (starts with '{' or '['), try a safe fallback: s = out.strip() if s.startswith("{") or s.startswith("["): - # Best-effort: parse JSON and attempt to extract rule textual forms if present. try: parsed = json.loads(s) - # parsed may be the whole ruleset (nftables list) or a list; find any "rule" objects rule_lines: List[str] = [] - # parsed might be dict with "nftables" or a list of records records = parsed.get("nftables") if isinstance(parsed, dict) else parsed if not isinstance(records, list): records = [] @@ -166,7 +129,6 @@ class NftManager: elif isinstance(part, dict) and "counter" in part: tokens.append("counter") elif isinstance(part, dict) and "queue" in part: - # handle fallback queue textualization q = part["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") @@ -177,13 +139,11 @@ class NftManager: tok += " bypass" tokens.append(tok) else: - # numeric or string value if isinstance(q, (int, float)): tokens.append(f"queue num {int(q)}") else: tokens.append(f"queue num {q}") else: - # fallback for unknown dict token if isinstance(part, dict): tokens.append("+".join(part.keys())) else: @@ -199,13 +159,8 @@ class NftManager: return out def delete_rule_by_handle_text(self, family: str, table: str, chain: str, handle: int) -> None: - """ - Delete a rule by handle using textual nft command: - delete rule
handle - """ if not isinstance(handle, int) or handle <= 0: raise ValueError("handle must be a positive integer") - # construct textual command cmd = f"delete rule {family} {table} {chain} handle {handle}" res = self.cmd(cmd) if res["rc"] != 0: @@ -218,137 +173,93 @@ router = APIRouter(prefix="/firewall", tags=["firewall"]) mgr = NftManager() -# ---------- Request/Response models (strongly typed) ---------- +# ---------- Models ---------- class RawCmdRequest(BaseModel): - cmd: str = Field(..., description="Textual nft command to execute", example="add rule inet filter input ip saddr 10.0.0.0/8 drop") - + cmd: str = Field(...) class ExecResult(BaseModel): - rc: int = Field(..., description="Return code from nft execution", example=0) - stdout: Optional[str] = Field(None, description="Standard output from nft", example="") - stderr: Optional[str] = Field(None, description="Standard error from nft", example="") - - class Config: - schema_extra = {"example": {"rc": 0, "stdout": "ok", "stderr": ""}} + rc: int + stdout: Optional[str] + stderr: Optional[str] -# --- Strong models returned to frontend --- class RuleOut(BaseModel): - handle: Optional[int] = Field(None, description="The rule handle (unique per rule), if available", example=3) - expr: Any = Field(..., description="Machine-readable nft expression (original nft JSON expr).") - text: str = Field(..., description="Deterministic short display string derived from expr", example="ip protocol icmp drop") - position: Optional[Any] = Field(None, description="Optional position metadata from nft if present") - comment: Optional[str] = Field(None, description="Optional comment attached to the rule") + handle: Optional[int] + expr: Any + text: str + position: Optional[Any] + comment: Optional[str] class ChainOut(BaseModel): - name: str = Field(..., description="Chain name", example="forward") - type: Optional[str] = Field(None, description="Chain type (e.g. filter, nat, route)") - hook: Optional[str] = Field(None, description="Hook (input/forward/output/ingress/egress) if present") - priority: Optional[int] = Field(None, description="Hook priority if present") - policy: Optional[str] = Field(None, description="Chain policy (accept/drop) if present") - rules: List[RuleOut] = Field(..., description="Rules in this chain (ordered)") + name: str + type: Optional[str] + hook: Optional[str] + priority: Optional[int] + policy: Optional[str] + rules: List[RuleOut] class TableOut(BaseModel): - family: str = Field(..., description="Table family (inet/bridge/ipv4/...)") - name: str = Field(..., description="Table name", example="filter") - chains: List[ChainOut] = Field(..., description="Chains in this table") + family: str + name: str + chains: List[ChainOut] class RulesetModel(BaseModel): - tables: List[TableOut] = Field(..., description="Top-level tables list") + tables: List[TableOut] -# ---------- New: CreateRuleRequest (JSON, expr required) ---------- class CreateRuleRequest(BaseModel): - family: str = Field(..., description="Table family (e.g. inet, bridge, ip, ip6)", example="bridge") - table: str = Field(..., description="Table name (e.g. filter)", example="filter") - chain: str = Field(..., description="Chain name (e.g. forward)", example="forward") - expr: Any = Field(..., description="nft JSON expression (machine-readable). This field is required for JSON rule creation.") - position: Optional[int] = Field(None, description="Optional insertion position (zero-based). If provided endpoint will insert at that position.") - comment: Optional[str] = Field(None, description="Optional comment") - - class Config: - schema_extra = { - "example": { - "family": "bridge", - "table": "filter", - "chain": "forward", - "expr": [{"match": {"left": {"payload": {"protocol": "ip", "field": "protocol"}}, "op": "==", "right": "icmp"}}, {"drop": None}], - "position": 0, - } - } + family: str + table: str + chain: str + expr: Any + position: Optional[int] + comment: Optional[str] -# ruleset may be typed RulesetModel or raw textual string (fallback) RulesetValue = Optional[Union[RulesetModel, str]] class RulesetOut(BaseModel): - ruleset: RulesetValue = Field( - None, - description="Parsed, strongly-typed ruleset (RulesetModel) or raw textual ruleset string if JSON is unavailable.", - ) + ruleset: RulesetValue -# ---------- Helpers to convert to desired shape ---------- +# ---------- Helpers ---------- _handle_re = re.compile(r"\s+#\s*handle\s+\d+\s*$") - def parse_priority(val: Any) -> Optional[int]: - """ - Robustly parse a priority value returned in various nft JSON shapes. - Accepts: - - int -> returns unchanged - - numeric string -> parsed int - - dict -> tries common nested keys ('priority', 'prio') - Returns None if not parseable. - """ if val is None: return None - # if it's already an int if isinstance(val, int): return val - # numeric string if isinstance(val, str): s = val.strip() - # try integer parse try: return int(s) except Exception: try: - # sometimes it's "0.0" or similar return int(float(s)) except Exception: return None - # nested dicts sometimes appear if isinstance(val, dict): - # look for common keys for key in ("priority", "prio"): if key in val: return parse_priority(val.get(key)) - # try nested dict values for v in val.values(): p = parse_priority(v) if p is not None: return p return None - def rule_text_from_expr(expr: Any) -> str: - """ - 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, 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") @@ -404,51 +315,31 @@ def rule_text_from_expr(expr: Any) -> str: def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]: - """ - Convert nft -j list ruleset parsed JSON into a deterministic, predictable JSON: - { - "tables": [ - { "family": ..., "name": ..., "chains": [ { "name": ..., "type": ..., "hook": ..., "priority": ..., "policy": ..., "rules": [ { handle, expr, text } ] } ] } - ] - } - """ result: Dict[str, Any] = {"tables": []} items = nft_json.get("nftables", []) if isinstance(nft_json, dict) else (nft_json or []) - - # Build intermediate map: (family, table) -> {family, name, chains: {chain_name: {"name", "type", "hook", "priority", "policy", "rules":[]}}} tables: Dict[Tuple[str, str], Dict[str, Any]] = {} for rec in items: - # table records if "table" in rec: t = rec["table"] fam = t.get("family") name = t.get("name") if fam and name: tables.setdefault((fam, name), {"family": fam, "name": name, "chains": {}}) - # chain records: capture chain metadata elif "chain" in rec: ch = rec["chain"] - # chain may include family/table or nested table reference fam = ch.get("family") or (ch.get("table") or {}).get("family") table_name = ch.get("table") or (ch.get("table") or {}).get("name") cname = ch.get("name") if fam and table_name and cname: tables.setdefault((fam, table_name), {"family": fam, "name": table_name, "chains": {}}) chains_map = tables[(fam, table_name)]["chains"] - - # existing chain (maybe created earlier by rule processing) existing = chains_map.get(cname) - # extract metadata robustly ch_type = ch.get("type") ch_hook = ch.get("hook") - # try multiple keys for priority/prio shapes ch_priority = parse_priority(ch.get("priority") if "priority" in ch else ch.get("prio") if "prio" in ch else ch.get("priority", None)) - # also attempt to parse nested shapes if present (some nft JSON variations) if ch_priority is None: ch_priority = parse_priority(ch.get("hook") if isinstance(ch.get("hook"), dict) else None) - ch_policy = ch.get("policy") - if existing is None: chains_map[cname] = { "name": cname, @@ -459,7 +350,6 @@ def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]: "rules": [], } else: - # merge into placeholder (do not overwrite existing rules) if isinstance(existing, dict): if existing.get("type") is None and ch_type is not None: existing["type"] = ch_type @@ -469,7 +359,6 @@ def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]: existing["priority"] = ch_priority if existing.get("policy") is None and ch_policy is not None: existing["policy"] = ch_policy - # rule records elif "rule" in rec: r = rec["rule"] fam = r.get("family") @@ -480,40 +369,25 @@ def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]: if fam and table_name and chain_name: tables.setdefault((fam, table_name), {"family": fam, "name": table_name, "chains": {}}) chains_map = tables[(fam, table_name)]["chains"] - # ensure chain placeholder exists, with possible metadata defaults chains_map.setdefault(chain_name, {"name": chain_name, "type": None, "hook": None, "priority": None, "policy": None, "rules": []}) - - rule_obj: Dict[str, Any] = { - "handle": handle, - "expr": expr, - "text": rule_text_from_expr(expr), - } - # include other useful metadata if present + rule_obj: Dict[str, Any] = {"handle": handle, "expr": expr, "text": rule_text_from_expr(expr)} if "position" in r: rule_obj["position"] = r["position"] if "comment" in r: rule_obj["comment"] = r["comment"] chains_map[chain_name]["rules"].append(rule_obj) - - # Attempt to salvage chain metadata from rule record if present - # some nft JSON may include 'chain' subfields inside rule record - # e.g. r.get('chain') might be an object - handle that defensively if isinstance(r.get("chain"), dict): csub = r.get("chain") - # try to parse nested priority if chains_map[chain_name].get("priority") is None: parsed_prio = parse_priority(csub.get("priority") if "priority" in csub else csub.get("prio")) if parsed_prio is not None: chains_map[chain_name]["priority"] = parsed_prio - # type/hook/policy from nested if present if chains_map[chain_name].get("type") is None and csub.get("type") is not None: chains_map[chain_name]["type"] = csub.get("type") if chains_map[chain_name].get("hook") is None and csub.get("hook") is not None: chains_map[chain_name]["hook"] = csub.get("hook") if chains_map[chain_name].get("policy") is None and csub.get("policy") is not None: chains_map[chain_name]["policy"] = csub.get("policy") - - # Convert map to sorted lists for deterministic order, and include chain metadata for (fam, tname) in sorted(tables.keys(), key=lambda k: (k[0], k[1])): tdata = tables[(fam, tname)] chains_list: List[Dict[str, Any]] = [] @@ -530,122 +404,28 @@ def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]: } ) result["tables"].append({"family": fam, "name": tname, "chains": chains_list}) - 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. - """ - if expr is None: - return "" - if isinstance(expr, str): - return expr - if not isinstance(expr, list): - return None - - parts: List[str] = [] - for element in expr: - if isinstance(element, dict): - if "drop" in element: - parts.append("drop") - continue - if "accept" in element: - parts.append("accept") - continue - if "counter" in element: - parts.append("counter") - continue - 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 - if "match" in element: - m = element["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 and isinstance(right, str): - if field == "protocol": - parts.append(f"{prot} {field} {right}") - continue - parts.append(f"payload({prot}.{field}) {right}") - continue - parts.append("match") - continue - 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 - if "tcp" in element or "udp" in element: - proto = "tcp" if "tcp" in element else "udp" - val = element.get(proto) - 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 - return None - else: - parts.append(str(element)) - - return " ".join(parts).strip() - - -# ---------- New helper: parse_ruleset_text ---------- +# ---------- Text parsing helpers ---------- def parse_ruleset_text(nft_text: str) -> Dict[Tuple[str, str, str], List[Dict[str, Optional[Union[str, int]]]]]: - """ - Parse the full textual `nft list ruleset` output and return a mapping: - (family, table, chain) -> [ { "line": "", "handle": }, ... ] - """ result: Dict[Tuple[str, str, str], List[Dict[str, Optional[Union[str, int]]]]] = {} if not nft_text: return result table_re = re.compile(r"^\s*table\s+(\S+)\s+(\S+)\s*\{") - # chain header line often: 'chain forward {', may include trailing comments chain_re = re.compile(r"^\s*chain\s+(\S+)\s*\{") handle_re = re.compile(r"#\s*handle\s*(\d+)\b") - # lines that indicate chain metadata (type/hook/priority/policy), we want to skip these chain_meta_re = re.compile(r"^\s*(type\b|hook\b|priority\b|policy\b)\b.*;") current_family = None current_table = None current_chain = None - # iterate line-by-line for raw_ln in nft_text.splitlines(): ln = raw_ln.rstrip("\n") s = ln.strip() - # detect table header m_table = table_re.match(ln) if m_table: current_family = m_table.group(1) @@ -653,7 +433,6 @@ def parse_ruleset_text(nft_text: str) -> Dict[Tuple[str, str, str], List[Dict[st current_chain = None continue - # detect chain header (works even if the same line contains trailing comment) m_chain = chain_re.match(ln) if m_chain and current_family and current_table: current_chain = m_chain.group(1) @@ -661,16 +440,12 @@ def parse_ruleset_text(nft_text: str) -> Dict[Tuple[str, str, str], List[Dict[st result.setdefault(key, []) continue - # if we're inside a chain, collect rule-like lines if current_family and current_table and current_chain: - # skip empty / pure brace lines if s == "" or s == "{" or s == "}": continue - # skip chain metadata lines like 'type filter hook forward priority 0;' if chain_meta_re.match(s): continue - # likely a rule line — extract optional handle comment m_handle = handle_re.search(s) handle_val: Optional[int] = None if m_handle: @@ -685,16 +460,7 @@ def parse_ruleset_text(nft_text: str) -> Dict[Tuple[str, str, str], List[Dict[st return result -# ---------- New helper: populate_text_from_ruleset_text ---------- def populate_text_from_ruleset_text(custom: Dict[str, Any], nft_text: str) -> None: - """ - Uses the parsed full ruleset textual output (nft_text) to update rule['text'] - in the 'custom' structure in-place. Matching strategy: - 1) handle -> line - 2) explicit 'position' field if present - 3) index-based mapping (best-effort) - 4) substring probe match - """ if not nft_text: return @@ -711,6 +477,10 @@ def populate_text_from_ruleset_text(custom: Dict[str, Any], nft_text: str) -> No continue key = (fam, tname, cname) textual_entries = parsed.get(key, []) + if not textual_entries: + # no textual lines for this chain; skip + continue + handle_map: Dict[int, str] = {} ordered_lines: List[str] = [] for ent in textual_entries: @@ -746,18 +516,10 @@ def populate_text_from_ruleset_text(custom: Dict[str, Any], nft_text: str) -> No rule["text"] = ln replaced = True break - # leave as-is if not replaced + # only update when a match was found; otherwise keep JSON-derived text -# ---------- Existing populate_text_from_chain_text (fallback) ---------- 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") @@ -771,7 +533,6 @@ 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() != ""] - # build handle -> line map handle_map: Dict[str, str] = {} for ln in lines: m = re.search(r"\bhandle\s+(\d+)\b", ln) @@ -808,12 +569,8 @@ def populate_text_from_chain_text(custom: Dict[str, Any]) -> None: # ---------- Routes ---------- - @router.get("/rules", response_model=RulesetOut, summary="List ruleset") def list_rules(): - """ - Returns the ruleset in a stable, strongly-typed JSON shape derived from `nft -j list ruleset`. - """ try: try: nft_json = mgr.list_rules_json() @@ -821,7 +578,6 @@ 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}") - # try to obtain textual ruleset too (best-effort) nft_text = "" try: nft_text = mgr.list_rules_text() @@ -830,37 +586,6 @@ def list_rules(): custom = build_predictable_ruleset(nft_json) - # If no tables/rules were discovered from JSON, attempt to build from textual listing - try: - if nft_text and (not custom.get("tables")): - # Build minimal structure from textual parse if JSON did not produce tables - parsed = parse_ruleset_text(nft_text) - if parsed: - # convert parsed -> custom structure - tables_map: Dict[Tuple[str, str], Dict[str, Any]] = {} - for (fam, tname, cname), entries in parsed.items(): - tables_map.setdefault((fam, tname), {"family": fam, "name": tname, "chains": {}}) - chain_rules: List[Dict[str, Any]] = [] - for ent in entries: - ln = ent.get("line") or "" - h = ent.get("handle") - # we don't have JSON expr; leave expr None and text the line - chain_rules.append({"handle": h, "expr": None, "text": ln}) - tables_map[(fam, tname)]["chains"].setdefault(cname, {"name": cname, "type": None, "hook": None, "priority": None, "policy": None, "rules": []}) - tables_map[(fam, tname)]["chains"][cname]["rules"].extend(chain_rules) - - # convert map to custom shape - custom = {"tables": []} - for (fam, tname) in sorted(tables_map.keys(), key=lambda k: (k[0], k[1])): - t = tables_map[(fam, tname)] - chains_list = [] - for cname in sorted(t["chains"].keys()): - ch = t["chains"][cname] - chains_list.append({"name": ch["name"], "type": None, "hook": None, "priority": None, "policy": None, "rules": ch["rules"]}) - custom["tables"].append({"family": fam, "name": tname, "chains": chains_list}) - except Exception as e: - logger.debug("fallback build from textual parse failed: %s", e) - # Enrich rule['text'] using the full textual ruleset (handle-first, position, index) try: if nft_text: @@ -874,7 +599,6 @@ def list_rules(): except Exception as e: logger.debug("list_rules: populate_text_from_chain_text failed: %s", e) - # debug: log counts so you can see what's being returned 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) @@ -895,9 +619,6 @@ def list_rules(): summary="Create rule (JSON, expr required; returns ExecResult with rc/stdout/stderr)", ) def create_rule_json(req: CreateRuleRequest): - """ - Create a rule from JSON (expr required). - """ try: family = req.family table = req.table @@ -971,12 +692,8 @@ def create_rule_json(req: CreateRuleRequest): 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"): - """ - Delete a rule by handle using textual nft command. - """ try: mgr.delete_rule_by_handle_text(family=family, table=table, chain=chain, handle=handle) except ValueError as e: @@ -992,9 +709,6 @@ def delete_rule(handle: int, family: str = "inet", table: str = "filter", chain: @router.post("/raw", response_model=ExecResult, summary="Execute raw textual nft command") def exec_raw(req: RawCmdRequest): - """ - Execute an arbitrary textual nft command and return structured {rc, stdout, stderr}. - """ try: res = mgr.cmd(req.cmd) rc = int(res.get("rc", -1) or -1) @@ -1003,4 +717,5 @@ def exec_raw(req: RawCmdRequest): logger.exception("exec_raw failed") raise HTTPException(status_code=500, detail=str(e)) + app.include_router(router) \ No newline at end of file