From 7db2816660deea399eb24a7ab63e36edf1a72899 Mon Sep 17 00:00:00 2001 From: malmert Date: Sat, 28 Feb 2026 22:35:52 +0100 Subject: [PATCH] test --- backend/src/api/nft_manager.py | 459 +++++++++------------------------ 1 file changed, 120 insertions(+), 339 deletions(-) diff --git a/backend/src/api/nft_manager.py b/backend/src/api/nft_manager.py index d2beb50..03dc1d6 100644 --- a/backend/src/api/nft_manager.py +++ b/backend/src/api/nft_manager.py @@ -23,15 +23,15 @@ class NftManager: """ Thin wrapper around python-nftables exposing: - cmd execution (textual nft commands via Nftables.cmd()) + - json execution via Nftables.json_cmd() when available - 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: + # If set_json_output exists it's a convenience; we won't rely on it for JSON path. self.nft.set_json_output(True) except Exception: logger.debug("set_json_output not available or ignored") @@ -46,157 +46,155 @@ class NftManager: logger.warning("nft cmd rc=%s stderr=%s cmd=%s", rc, err, text_cmd) return {"rc": rc, "stdout": out, "stderr": err} + def json_cmd(self, text_cmd: str) -> Tuple[int, str, str]: + """ + Execute a JSON-output nft command. Prefer Nftables.json_cmd if available + (returns tuple (rc, stdout, stderr)). Otherwise, call `nft -j` using cmd() + and attempt to return the same tuple shape. + """ + # Prefer built-in json_cmd if present + if hasattr(self.nft, "json_cmd"): + try: + res = self.nft.json_cmd(text_cmd) + # Expecting (rc, out, err) + if isinstance(res, (list, tuple)) and len(res) >= 3: + return int(res[0]), res[1], res[2] + except Exception as e: + logger.debug("json_cmd failed, falling back to cmd with -j: %s", e) + + # Fallback: call cmd with -j variant and parse output + # Use 'list ruleset -j' or similar command suffixes as caller provides the whole command + cmd_with_j = f"{text_cmd} -j" if "-j" not in text_cmd else text_cmd + r = self.cmd(cmd_with_j) + rc = int(r.get("rc", -1) or -1) + out = r.get("stdout") or "" + err = r.get("stderr") or "" + return rc, out, err + def list_rules(self) -> str: """ Return the textual ruleset as produced by 'nft list ruleset'. - Uses the textual command path. """ res = self.cmd("list ruleset") if res["rc"] != 0: raise NftError(f"nft list ruleset failed: {res['stderr']}") - return res["stdout"] + return res["stdout"] or "" def list_rules_json(self) -> Dict[str, Any]: """ Try to obtain nft -j list ruleset (JSON). Returns parsed JSON dict on success. Raises NftError on failure or when output cannot be parsed as JSON. """ - # prefer using json_cmd when available for a clean tuple (rc, out, err) + rc, out, err = self.json_cmd("list ruleset") + if rc != 0: + raise NftError(f"nft list ruleset failed: {err}") + if not out: + raise NftError("empty JSON output from nft") try: - res = self.nft.json_cmd("list ruleset") - if not isinstance(res, (list, tuple)) or len(res) < 3: - raise Exception("unexpected json_cmd result shape") - if res[0] != 0: - raise NftError(f"nft list ruleset failed: {res[2]}") - return json.loads(res[1]) - except AttributeError: - # fallback to textual cmd + json.loads if json_cmd is not available - cmd_variants = ["list ruleset -j", "list ruleset"] - last_err = None - for c in cmd_variants: - r = self.cmd(c) - if r["rc"] != 0: - last_err = r["stderr"] - continue - out = r["stdout"] - if not out: - last_err = "empty output" - continue - try: - parsed = json.loads(out) - return parsed - except json.JSONDecodeError as e: - last_err = f"json decode error: {e}" - continue - raise NftError(f"unable to get JSON ruleset: {last_err}") + parsed = json.loads(out) + return parsed + except json.JSONDecodeError as e: + raise NftError(f"unable to parse JSON output from nft: {e}") 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. + We prefer calling nft in textual mode (nft.cmd). If the wrapper can't return textual + output and returns JSON, we attempt a best-effort reconstruction of the textual lines. """ 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) - except Exception: - logger.debug("failed to restore set_json_output(True)") - + # Attempt to call textual cmd (this uses self.nft.cmd) + res = self.cmd(cmd) if res["rc"] != 0: - 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. + # if the raw textual command failed, try JSON and attempt to reconstruct textual + logger.debug("list_chain_text textual cmd rc!=0; trying JSON fallback: %s", res["stderr"]) + # try json_cmd + rc, out, err = self.json_cmd(cmd) + if rc != 0: + raise NftError(f"nft {cmd} failed: {err}") + # attempt to build textual lines from JSON 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 = [] - for rec in records: - if "rule" in rec: - r = rec["rule"] - expr = r.get("expr") - if isinstance(expr, list): - tokens: List[str] = [] - for part in expr: - if isinstance(part, dict) and "match" in part: + parsed = json.loads(out) + except Exception: + # give up, return raw text (could be empty) + return res.get("stdout") or "" + # build textual representation from JSON (best-effort) + lines: List[str] = [] + records = parsed.get("nftables") if isinstance(parsed, dict) else (parsed or []) + if not isinstance(records, list): + records = [] + for rec in records: + if "rule" in rec: + # try to use 'line' if present + r = rec["rule"] + # sometimes json includes 'expr' or 'handle' — we try to recompose a line + line_parts: List[str] = [] + handle = r.get("handle") + # attempt to render expr to short textual form + expr = r.get("expr") + if isinstance(expr, list): + # best-effort: reuse simple tokens + for part in expr: + if isinstance(part, dict): + if "payload" in part: + p = part["payload"] + prot = p.get("protocol") + field = p.get("field") + if prot and field: + line_parts.append(f"payload({prot}.{field})") + continue + 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): + 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}") + line_parts.append(f"{prot} {field} {right}") continue - tokens.append("match") - elif isinstance(part, dict) and "payload" in part: - p = part["payload"] - prot = p.get("protocol") - field = p.get("field") - tokens.append(f"payload({prot}.{field})") - elif isinstance(part, dict) and "drop" in part: - tokens.append("drop") - elif isinstance(part, dict) and "accept" in part: - tokens.append("accept") - elif isinstance(part, dict) and "counter" in part: - tokens.append("counter") - elif isinstance(part, dict) and "queue" in part: - # handle fallback queue textualization + line_parts.append("match") + continue + if "drop" in part: + line_parts.append("drop") + continue + if "accept" in part: + line_parts.append("accept") + continue + if "counter" in part: + line_parts.append("counter") + continue + if "queue" in part: 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") - tok = "queue" + num = q.get("num") or q.get("number") or q.get("queue_number") + token = "queue" if num is not None: - tok += f" num {num}" + token += f" num {num}" if q.get("bypass"): - tok += " bypass" - tokens.append(tok) + token += " bypass" + line_parts.append(token) + elif isinstance(q, (int, float)): + line_parts.append(f"queue num {int(q)}") 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: - tokens.append(str(part)) - rule_lines.append(" ".join(tokens)) - else: - rule_lines.append(json.dumps(r)) - if rule_lines: - return "\n".join(rule_lines) - except Exception: - logger.debug("fallback JSON parsing of chain output failed; returning raw output") - - return out + line_parts.append("queue") + continue + # fallback to keys + line_parts.append("+".join(sorted(part.keys()))) + else: + line_parts.append(str(part)) + else: + # expr not list -> fallback to JSON dump of rule + line_parts.append(json.dumps(r)) + # combine into single textual line; append handle if present + text_line = " ".join(line_parts).strip() + if handle is not None: + text_line = f"{text_line} # handle {handle}" + lines.append(text_line) + return "\n".join(lines) if lines else (res.get("stdout") or "") + # textual cmd succeeded + return res.get("stdout") or "" def delete_rule_by_handle_text(self, family: str, table: str, chain: str, handle: int) -> None: """ @@ -205,7 +203,6 @@ class NftManager: """ 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: @@ -297,38 +294,23 @@ _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: @@ -339,8 +321,6 @@ def parse_priority(val: Any) -> Optional[int]: 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 "" @@ -348,7 +328,6 @@ def rule_text_from_expr(expr: Any) -> str: 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 +383,32 @@ 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 +419,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 +428,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,32 +438,24 @@ 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 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: @@ -513,7 +463,6 @@ def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]: 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]] = [] @@ -534,127 +483,8 @@ def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]: 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. - Supported cases (common): - - [{'match': {'left': {'payload': {'protocol':'ip','field':'protocol'}}, 'op':'==', 'right':'icmp'}}, {'drop': None}] - -> 'ip protocol icmp drop' - - payload / tcp / udp / counter / accept - - queue tokens and optional bypass support - This intentionally does not attempt to support every nft JSON construct. - """ - if expr is None: - return "" - if isinstance(expr, str): - return expr - if not isinstance(expr, list): - # unsupported top-level type - return None - - parts: List[str] = [] - for element in expr: - if isinstance(element, dict): - # handle drop/accept/counter directly - if "drop" in element: - parts.append("drop") - continue - if "accept" in element: - parts.append("accept") - continue - if "counter" in element: - parts.append("counter") - continue - - # queue support: allow {"queue": 1} or {"queue": {"num":1, "bypass": True}} etc. - 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 - - # match left/right payload equals -> ip protocol icmp, or ip saddr/daddr - if "match" in element: - m = element["match"] - left = m.get("left") - right = m.get("right") - # payload matches - if isinstance(left, dict) and "payload" in left and isinstance(right, (str, int)): - p = left["payload"] - prot = p.get("protocol") - field = p.get("field") - # common: protocol field match (protocol == icmp) - if prot and field and isinstance(right, str): - # ip vs ip6 decision is left to the frontend; here we render 'ip protocol icmp' (works for many setups) - if field == "protocol": - parts.append(f"{prot} {field} {right}") - continue - # payload might be l4 ports etc; produce generic payload(...) token - parts.append(f"payload({prot}.{field}) {right}") - continue - # fallback for match: try to stringify right - parts.append("match") - continue - - # payload shorthand - 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 - - # tcp/udp as nested dicts sometimes appear - if "tcp" in element or "udp" in element: - proto = "tcp" if "tcp" in element else "udp" - val = element.get(proto) - # attempt to detect dport/sport keys - 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 - - # cmp/binary operators etc — not supported deterministically - # return None to indicate we can't safely render this expr - return None - else: - # non-dict token (string/number) - parts.append(str(element)) - - # join tokens - return " ".join(parts).strip() - - # ---------- New helper: populate_text_from_chain_text ---------- 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") @@ -668,7 +498,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) @@ -685,17 +514,14 @@ def populate_text_from_chain_text(custom: Dict[str, Any]) -> None: replaced = True if not replaced: - # fallback: try to find a line that contains the JSON-derived compact text fragment expr = rule.get("expr") probe = rule.get("text") or rule_text_from_expr(expr) if probe: - # try longest-first strategy (not strictly necessary here) — simple substring match for ln in lines: if probe in ln: rule["text"] = ln.strip() replaced = True break - # if still not replaced, keep existing rule["text"] except Exception as e: logger.debug( "populate_text_from_chain_text: failed to get textual chain for %s %s %s: %s", @@ -713,12 +539,6 @@ def populate_text_from_chain_text(custom: Dict[str, Any]) -> None: def list_rules(): """ Returns the ruleset in a stable, strongly-typed JSON shape derived from `nft -j list ruleset`. - - Structure: - { "ruleset": { "tables": [ { "family": ..., "name": ..., "chains": [ { "name": ..., "type": ..., "hook": ..., "priority": ..., "policy": ..., "rules": [ { "handle", "expr", "text" } ] } ] } ] } } - - Fallback: - - If nft JSON is unavailable, falls back to returning the raw textual ruleset string. """ try: try: @@ -730,18 +550,14 @@ def list_rules(): custom = build_predictable_ruleset(nft_json) - # Enrich rule['text'] by attempting to fetch the exact textual nft rule lines - # as printed by `nft list chain
`. This is best-effort and - # will not fail the overall listing if textual retrieval fails for some chains. try: populate_text_from_chain_text(custom) except Exception as e: logger.debug("list_rules: populate_text_from_chain_text failed: %s", e) - # IMPORTANT: return the parsed RulesetModel (not a JSON string) so FastAPI/Pydantic - # will serialize the structure properly without double-escaping. + # IMPORTANT: return a plain dict (not a JSON string) so FastAPI/Pydantic serializes it naturally. ruleset_model = RulesetModel.parse_obj(custom) - return RulesetOut(ruleset=ruleset_model) + return RulesetOut(ruleset=ruleset_model.dict()) except NftError as e: logger.exception("list_rules failed") raise HTTPException(status_code=500, detail=str(e)) @@ -757,15 +573,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). - - If req.position is provided, uses: insert rule
position - - Otherwise, uses: add rule
(append) - - If rendering fails: 400 instructing the client to use POST /firewall/raw - - Returns ExecResult on success (201) or on error (400) with stdout/stderr in body. - - If nft wrapper returns an invalid rc but the command produced no stderr, we double-check the chain - to see if the new rule is present; if present we treat as success. - """ try: family = req.family table = req.table @@ -783,12 +590,9 @@ def create_rule_json(req: CreateRuleRequest): expr_text = rendered.strip() - # If a position is explicitly requested, use the 'insert rule ... position ...' form. - # 'add rule ... position ...' is not supported by some nft versions / syntaxes. if req.position is not None: try: pos = int(req.position) - # clamp pos to >= 0 if pos < 0: pos = 0 except Exception: @@ -800,12 +604,10 @@ def create_rule_json(req: CreateRuleRequest): logger.info("create_rule_json executing command: %s", cmd) res = mgr.cmd(cmd) - # res expected {"rc": rc, "stdout": out, "stderr": err} raw_rc = res.get("rc") stdout = res.get("stdout") or "" stderr = res.get("stderr") or "" - # Coerce rc to int safely; if not int-like, set -1 to indicate unknown. try: rc = int(raw_rc) except Exception: @@ -815,33 +617,21 @@ def create_rule_json(req: CreateRuleRequest): exec_res = ExecResult(rc=rc, stdout=stdout or None, stderr=stderr or None) - # If rc == 0 — success if rc == 0: return exec_res - # Handle the annoying case: wrapper returned invalid rc (<0) or non-zero, - # but stderr is empty. The command may have succeeded nevertheless. if (rc < 0 or rc != 0) and stderr.strip() == "": logger.debug("create_rule_json: rc indicates failure but stderr empty; verifying rule presence") - - # Attempt to verify the rule exists by listing the chain and searching for a textual match. - # We use list_chain_text because it returns textual rule lines we can search for the preview text. try: chain_text = mgr.list_chain_text(family, table, chain) or "" - # Simple presence check: the textual fragment we attempted to add should be present - # as a substring in the chain listing (e.g. "ip protocol icmp drop"). if expr_text and expr_text in chain_text: logger.info("create_rule_json: detected rule in chain after add; treating as success") - # return success ExecResult with rc=0 to indicate success to client return ExecResult(rc=0, stdout=stdout or None, stderr=stderr or None) else: logger.debug("create_rule_json: rule not found in chain text; chain_text=%r", chain_text) except Exception as e_chain: logger.warning("create_rule_json: failed to list chain for verification: %s", e_chain) - # If we reach here -> treat as error: return 400 with exec_res in body. - # FastAPI cannot both raise HTTPException and include ExecResult as body easily, so raise HTTPException - # with detail that includes stderr and the executed cmd. 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) @@ -850,21 +640,14 @@ def create_rule_json(req: CreateRuleRequest): logger.warning("create_rule_json NftError: %s", e) raise HTTPException(status_code=400, detail=str(e)) except HTTPException: - # re-raise HTTPException so we don't wrap it again 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"): - """ - Delete a rule by handle using textual nft command. - Command executed: - delete rule
handle - """ try: mgr.delete_rule_by_handle_text(family=family, table=table, chain=chain, handle=handle) except ValueError as e: @@ -880,9 +663,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) @@ -891,4 +671,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