diff --git a/backend/src/api/nft_manager.py b/backend/src/api/nft_manager.py index a76736d..0f70b1b 100644 --- a/backend/src/api/nft_manager.py +++ b/backend/src/api/nft_manager.py @@ -1,54 +1,44 @@ # app.py -from typing import Any, Dict, List, Optional, Tuple, Union +from typing import Any, Dict, List, Optional, Tuple from fastapi import FastAPI, APIRouter, HTTPException, status from fastapi.responses import JSONResponse -from pydantic import BaseModel, Field import logging import json import re -# libnftables (we call textual commands through its .cmd() method) +# python-nftables wrapper from nftables import Nftables # type: ignore -# ---------- logging ---------- logging.basicConfig(level=logging.INFO) -logger = logging.getLogger("nft_api_raw_only") +logger = logging.getLogger("nft_api") -# ---------- Exceptions ---------- +# ---------- Errors ---------- class NftError(RuntimeError): pass - -# ---------- NftManager (textual vs JSON) ---------- +# ---------- NftManager (small, explicit) ---------- class NftManager: - """ - Thin wrapper around python-nftables exposing: - - cmd execution via Nftables.cmd() for textual output - - json_cmd execution when available for JSON output (or fallback to cmd with -j) - - helpers to list rules / chain text - """ - def __init__(self) -> None: self.nft = Nftables() - # best-effort: don't strictly rely on set_json_output for JSON path + # best-effort: don't rely solely on set_json_output globally try: if hasattr(self.nft, "set_json_output"): self.nft.set_json_output(True) except Exception: - logger.debug("set_json_output not available or ignored") + logger.debug("set_json_output not available") - def cmd(self, text_cmd: str) -> Dict[str, Optional[str]]: - """Execute textual nft command via Nftables.cmd().""" + def cmd(self, text_cmd: str) -> Tuple[int, str, str]: + """ + Run textual nft command via Nftables.cmd() and return (rc, stdout, stderr). + """ 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": int(rc), "stdout": out or "", "stderr": err or ""} + return int(rc), out or "", err or "" def json_cmd(self, text_cmd: str) -> Tuple[int, str, str]: """ - Execute nft command expecting JSON output. - Prefer Nftables.json_cmd when available (returns (rc, out, err)). - Otherwise call cmd() with a '-j' suffix and return a similar tuple. + Run nft expecting JSON output. + Prefer Nftables.json_cmd() when present; otherwise call cmd() with -j appended. + Returns (rc, stdout, stderr). """ if hasattr(self.nft, "json_cmd"): try: @@ -56,79 +46,61 @@ class NftManager: if isinstance(res, (list, tuple)) and len(res) >= 3: return int(res[0]), res[1] or "", res[2] or "" except Exception as e: - logger.debug("nft.json_cmd failed, falling back to cmd -j: %s", e) + logger.debug("nft.json_cmd failed: %s (falling back to -j)", e) - # fallback: append -j if not present and call textual cmd() + # fallback: append -j and use cmd() cmd_with_j = text_cmd if "-j" in text_cmd else f"{text_cmd} -j" - r = self.cmd(cmd_with_j) - return int(r.get("rc", -1) or -1), r.get("stdout", "") or "", r.get("stderr", "") or "" + return self.cmd(cmd_with_j) - def list_rules(self) -> str: - """Return textual ruleset from `nft list ruleset`.""" - res = self.cmd("list ruleset") - if res["rc"] != 0: - raise NftError(f"nft list ruleset failed: {res['stderr']}") - return res["stdout"] - - def list_rules_json(self) -> Dict[str, Any]: - """Return parsed JSON from `nft -j list ruleset`.""" - rc, out, err = self.json_cmd("list ruleset") + def list_rules_text(self) -> str: + rc, out, err = self.cmd("list ruleset") if rc != 0: raise NftError(f"nft list ruleset failed: {err}") - if not out: - raise NftError("empty JSON output from nft") + return out + + def list_rules_json(self) -> Dict[str, Any]: + rc, out, err = self.json_cmd("list ruleset") + if rc != 0: + raise NftError(f"nft list ruleset failed (json): {err}") try: - parsed = json.loads(out) - return parsed + return json.loads(out) except json.JSONDecodeError as e: - raise NftError(f"unable to parse JSON output from nft: {e}") + raise NftError(f"invalid JSON from nft: {e}") def list_chain_text(self, family: str, table: str, chain: str) -> str: """ - Return textual output of `nft list chain `. - Prefer textual cmd(); if it fails and JSON is returned, attempt a best-effort - reconstruction of textual lines from JSON. + Prefer textual listing via cmd(). If that fails, use JSON fallback to reconstruct reasonable lines. """ cmd = f"list chain {family} {table} {chain}" - res = self.cmd(cmd) - if res["rc"] == 0: - return res["stdout"] or "" - - # textual call failed -> try JSON fallback and reconstruct - logger.debug("list_chain_text: textual cmd failed (%s), attempting JSON fallback", res["stderr"]) - rc, out, err = self.json_cmd(cmd) - if rc != 0: - # raise original textual error if JSON fallback doesn't work either - raise NftError(f"nft {cmd} failed: {res['stderr'] or err}") + rc, out, err = self.cmd(cmd) + if rc == 0: + return out + # fallback to JSON path and try to reconstruct + rcj, outj, errj = self.json_cmd(cmd) + if rcj != 0: + # prefer original textual error message + raise NftError(f"nft {cmd} failed: {err or errj}") try: - parsed = json.loads(out) + parsed = json.loads(outj) except Exception: - # give up and return textual stdout (maybe empty) - return res.get("stdout") or "" + return out # return whatever textual output we had (maybe empty) lines: List[str] = [] - records = parsed.get("nftables") if isinstance(parsed, dict) else (parsed or []) - if not isinstance(records, list): - records = [] - for rec in records: + recs = parsed.get("nftables") if isinstance(parsed, dict) else (parsed or []) + if not isinstance(recs, list): + return out + for rec in recs: if "rule" in rec: r = rec["rule"] handle = r.get("handle") expr = r.get("expr") parts: List[str] = [] if isinstance(expr, list): - 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: - parts.append(f"payload({prot}.{field})") - continue - if "match" in part: - m = part["match"] + for el in expr: + if isinstance(el, dict): + if "match" in el: + m = el["match"] left = m.get("left") right = m.get("right") if isinstance(left, dict) and "payload" in left and isinstance(right, (str, int)): @@ -140,88 +112,54 @@ class NftManager: continue parts.append("match") continue - if "drop" in part: - parts.append("drop") + if "payload" in el: + p = el["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 "accept" in part: - parts.append("accept") - continue - if "counter" in part: - parts.append("counter") - continue - if "queue" in part: - q = part["queue"] + if "drop" in el: + parts.append("drop"); continue + if "accept" in el: + parts.append("accept"); continue + if "counter" in el: + parts.append("counter"); continue + if "queue" in el: + q = el["queue"] tok = "queue" if isinstance(q, dict): - num = q.get("num") or q.get("number") or q.get("queue_number") + num = q.get("num") or q.get("number") if num is not None: tok += f" num {num}" if q.get("bypass") or q.get("flags") == "bypass": tok += " bypass" - parts.append(tok) - continue - if isinstance(q, (int, float)): - parts.append(f"queue num {int(q)}") - continue - parts.append("queue") - continue - parts.append("+".join(sorted(part.keys()))) + elif isinstance(q, (int, float)): + tok += f" num {int(q)}" + parts.append(tok); continue + parts.append("+".join(sorted(el.keys()))) else: - parts.append(str(part)) + parts.append(str(el)) else: parts.append(json.dumps(r)) - text_line = " ".join(parts).strip() + txt = " ".join([p for p in parts if p]).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 "") - - def delete_rule_by_handle_text(self, family: str, table: str, chain: str, handle: int) -> None: - if not isinstance(handle, int) or handle <= 0: - raise ValueError("handle must be a positive integer") - cmd = f"delete rule {family} {table} {chain} handle {handle}" - res = self.cmd(cmd) - if res["rc"] != 0: - raise NftError(f"delete rule failed: {res['stderr']}") + txt = f"{txt} # handle {handle}" + lines.append(txt) + return "\n".join(lines) if lines else out -# ---------- FastAPI + Router ---------- -app = FastAPI(title="Unrestricted nftables API (json create)") -router = APIRouter(prefix="/firewall", tags=["firewall"]) -mgr = NftManager() +# ---------- utilities for predictable representation ---------- +_handle_re = re.compile(r"\bhandle\s+(\d+)\b") - -# ---------- Request/Response models (only used for validation / docs) ---------- -class RawCmdRequest(BaseModel): - cmd: str = Field(..., example="add rule inet filter input ip saddr 10.0.0.0/8 drop") - - -class ExecResult(BaseModel): - rc: int - stdout: Optional[str] = None - stderr: Optional[str] = None - - -class CreateRuleRequest(BaseModel): - family: str - table: str - chain: str - expr: Any - position: Optional[int] = None - comment: Optional[str] = None - - -# ---------- Helpers to make predictable output ---------- -_handle_re = re.compile(r"\s+#\s*handle\s+\d+\s*$") - - -def parse_priority(val: Any) -> Optional[int]: - if val is None: +def parse_priority(v: Any) -> Optional[int]: + if v is None: return None - if isinstance(val, int): - return val - if isinstance(val, str): - s = val.strip() + if isinstance(v, int): + return v + if isinstance(v, str): + s = v.strip() try: return int(s) except Exception: @@ -229,196 +167,177 @@ def parse_priority(val: Any) -> Optional[int]: return int(float(s)) except Exception: return None - if isinstance(val, dict): - for key in ("priority", "prio"): - if key in val: - return parse_priority(val.get(key)) - for v in val.values(): - p = parse_priority(v) + if isinstance(v, dict): + for k in ("priority", "prio"): + if k in v: + return parse_priority(v.get(k)) + for val in v.values(): + p = parse_priority(val) if p is not None: return p return None - def rule_text_from_expr(expr: Any) -> str: + """ + Conservative, compact textualifier from nft JSON expr. + If unknown tokens appear this returns a reasonable fallback (json dump). + """ if expr is None: return "" 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): - p = left["payload"] - prot = p.get("protocol") - field = p.get("field") + out: List[str] = [] + for el in expr: + if isinstance(el, dict): + if "match" in el: + m = el["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") + out.append(f"{prot} {field} {right}"); continue + out.append("match"); continue + if "payload" in el: + p = el["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" + out.append(f"payload({prot}.{field})"); continue + out.append("payload"); continue + if "drop" in el: + out.append("drop"); continue + if "accept" in el: + out.append("accept"); continue + if "counter" in el: + out.append("counter"); continue + if "queue" in el: + q = el["queue"]; tok = "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") + num = q.get("num") or q.get("number") if num is not None: - token += f" num {num}" + tok += f" num {num}" if q.get("bypass") or q.get("flags") == "bypass": - token += " bypass" + tok += " bypass" elif isinstance(q, (int, float)): - token += f" num {int(q)}" - elif isinstance(q, str): - token += f" num {q}" - tokens.append(token) - else: - tokens.append("+".join(sorted(part.keys()))) + tok += f" num {int(q)}" + out.append(tok); continue + # unknown dict -> show keys + out.append("+".join(sorted(el.keys()))) else: - tokens.append(str(part)) - return " ".join(tokens) - return str(expr) - + out.append(str(el)) + return " ".join(out) + # fallback + try: + return str(expr) + except Exception: + return json.dumps(expr) def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]: + """ + Convert nft -j ruleset into a deterministic dict: + { "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 []) tables: Dict[Tuple[str, str], Dict[str, Any]] = {} for rec in items: if "table" in rec: - t = rec["table"] - fam = t.get("family") - name = t.get("name") + t = rec["table"]; fam = t.get("family"); name = t.get("name") if fam and name: tables.setdefault((fam, name), {"family": fam, "name": name, "chains": {}}) elif "chain" in rec: - ch = rec["chain"] - 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 = chains_map.get(cname) - ch_type = ch.get("type") - ch_hook = ch.get("hook") - ch_priority = parse_priority(ch.get("priority") if "priority" in ch else ch.get("prio") if "prio" in ch else ch.get("priority", None)) - 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") + c = rec["chain"] + fam = c.get("family") or (c.get("table") or {}).get("family") + tname = c.get("table") or (c.get("table") or {}).get("name") + cname = c.get("name") + if fam and tname and cname: + tables.setdefault((fam, tname), {"family": fam, "name": tname, "chains": {}}) + chains = tables[(fam, tname)]["chains"] + existing = chains.get(cname) + ch_type = c.get("type"); ch_hook = c.get("hook") + ch_prio = parse_priority(c.get("priority") if "priority" in c else c.get("prio") if "prio" in c else c.get("priority")) + ch_policy = c.get("policy") if existing is None: - chains_map[cname] = { - "name": cname, - "type": ch_type, - "hook": ch_hook, - "priority": ch_priority, - "policy": ch_policy, - "rules": [], - } + chains[cname] = {"name": cname, "type": ch_type, "hook": ch_hook, "priority": ch_prio, "policy": ch_policy, "rules": []} else: - if isinstance(existing, dict): - if existing.get("type") is None and ch_type is not None: - existing["type"] = ch_type - if existing.get("hook") is None and ch_hook is not None: - existing["hook"] = ch_hook - if existing.get("priority") is None and ch_priority is not None: - existing["priority"] = ch_priority - if existing.get("policy") is None and ch_policy is not None: - existing["policy"] = ch_policy + if existing.get("type") is None and ch_type is not None: + existing["type"] = ch_type + if existing.get("hook") is None and ch_hook is not None: + existing["hook"] = ch_hook + if existing.get("priority") is None and ch_prio is not None: + existing["priority"] = ch_prio + if existing.get("policy") is None and ch_policy is not None: + existing["policy"] = ch_policy elif "rule" in rec: r = rec["rule"] - fam = r.get("family") - table_name = r.get("table") - chain_name = r.get("chain") - handle = r.get("handle") - expr = r.get("expr") - 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"] - 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), - } - 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) + fam = r.get("family"); tname = r.get("table"); cname = r.get("chain") + if not (fam and tname and cname): + continue + tables.setdefault((fam, tname), {"family": fam, "name": tname, "chains": {}}) + chains = tables[(fam, tname)]["chains"] + chains.setdefault(cname, {"name": cname, "type": None, "hook": None, "priority": None, "policy": None, "rules": []}) + rule_obj = { + "handle": r.get("handle"), + "expr": r.get("expr"), + # initial text from expr serializer (may be replaced later by exact textual line) + "text": rule_text_from_expr(r.get("expr")), + } + if "position" in r: + rule_obj["position"] = r["position"] + if "comment" in r: + rule_obj["comment"] = r["comment"] + chains[cname]["rules"].append(rule_obj) - if isinstance(r.get("chain"), dict): - csub = r.get("chain") - 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 - 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") + # salvage chain metadata if nested under rule (some nft outputs) + if isinstance(r.get("chain"), dict): + csub = r.get("chain") + if chains[cname].get("priority") is None: + p = parse_priority(csub.get("priority") if "priority" in csub else csub.get("prio")) + if p is not None: + chains[cname]["priority"] = p + if chains[cname].get("type") is None and csub.get("type") is not None: + chains[cname]["type"] = csub.get("type") + if chains[cname].get("hook") is None and csub.get("hook") is not None: + chains[cname]["hook"] = csub.get("hook") + if chains[cname].get("policy") is None and csub.get("policy") is not None: + chains[cname]["policy"] = csub.get("policy") + # materialize into lists (sorted for deterministic order) for (fam, tname) in sorted(tables.keys(), key=lambda k: (k[0], k[1])): tdata = tables[(fam, tname)] chains_list: List[Dict[str, Any]] = [] for cname in sorted(tdata["chains"].keys()): - chdata = tdata["chains"][cname] - chains_list.append( - { - "name": chdata.get("name"), - "type": chdata.get("type"), - "hook": chdata.get("hook"), - "priority": chdata.get("priority"), - "policy": chdata.get("policy"), - "rules": chdata.get("rules", []), - } - ) + ch = tdata["chains"][cname] + chains_list.append({ + "name": ch.get("name"), + "type": ch.get("type"), + "hook": ch.get("hook"), + "priority": ch.get("priority"), + "policy": ch.get("policy"), + "rules": ch.get("rules", []), + }) result["tables"].append({"family": fam, "name": tname, "chains": chains_list}) - return result - -def populate_text_from_chain_text(custom: Dict[str, Any]) -> None: - tables = custom.get("tables") or [] - for t in tables: - fam = t.get("family") - tname = t.get("name") +def enrich_text_from_chain(custom: Dict[str, Any], mgr: NftManager) -> None: + """ + Replace rule['text'] with exact textual lines from `nft list chain` when possible. + Looks up rule handles; if not found it tries substring matching on the compact probe. + """ + for table in custom.get("tables", []): + fam = table.get("family"); tname = table.get("name") if not fam or not tname: continue - for ch in t.get("chains", []): + for ch in table.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[str, str] = {} + txt = mgr.list_chain_text(fam, tname, cname) or "" + lines = [ln.strip() for ln in txt.splitlines() if ln.strip()] + # map handle -> line + handle_map = {} for ln in lines: - m = re.search(r"\bhandle\s+(\d+)\b", ln) + m = _handle_re.search(ln) if m: - handle_map[m.group(1)] = ln.strip() - + handle_map[m.group(1)] = ln for rule in ch.get("rules", []): replaced = False h = rule.get("handle") @@ -427,143 +346,112 @@ def populate_text_from_chain_text(custom: Dict[str, Any]) -> None: if key in handle_map: rule["text"] = handle_map[key] replaced = True - if not replaced: - expr = rule.get("expr") - probe = rule.get("text") or rule_text_from_expr(expr) + probe = rule.get("text") or rule_text_from_expr(rule.get("expr")) if probe: for ln in lines: if probe in ln: - rule["text"] = ln.strip() + rule["text"] = ln replaced = True break except Exception as e: - logger.debug("populate_text_from_chain_text failed for %s %s %s: %s", fam, tname, cname, e) + logger.debug("failed to enrich chain text for %s %s %s: %s", fam, tname, cname, e) continue +# ---------- FastAPI app ---------- +app = FastAPI(title="nft API (clean)") +router = APIRouter(prefix="/firewall", tags=["firewall"]) +mgr = NftManager() -# ---------- Routes ---------- -@router.get("/rules", summary="List ruleset") -def list_rules(): +@router.get("/rules") +def get_rules(): """ - Returns JSON: { "ruleset": } - Ensures `ruleset` is a native dict when JSON is available and parsed. + Return: JSONResponse({"ruleset": }) + - If nft -j is available -> returns native dict under "ruleset" + - If not, returns the textual ruleset string under "ruleset" """ try: try: nft_json = mgr.list_rules_json() except NftError as e: - logger.debug("could not obtain nft JSON ruleset: %s", e) - # textual fallback - return plain textual ruleset string - text = mgr.list_rules() + logger.debug("json listing unavailable: %s", e) + # fallback to textual listing (return string; not JSON-encoded) + text = mgr.list_rules_text() return JSONResponse(content={"ruleset": text.strip() if text is not None else None}, status_code=200) - # build structured representation (native Python) custom = build_predictable_ruleset(nft_json) - - # Try to replace rule['text'] with exact textual lines from nft list chain ... + # best-effort: replace compact text with exact lines try: - populate_text_from_chain_text(custom) + enrich_text_from_chain(custom, mgr) except Exception as e: - logger.debug("populate_text_from_chain_text failed: %s", e) + logger.debug("enrich_text_from_chain failed: %s", e) - # Return native structure (do NOT json.dumps) + # Return native dict (no json.dumps) return JSONResponse(content={"ruleset": custom}, status_code=200) except NftError as e: - logger.exception("list_rules failed") + logger.exception("get_rules: nft error") raise HTTPException(status_code=500, detail=str(e)) except Exception as e: - logger.exception("list_rules internal error") + logger.exception("get_rules internal") raise HTTPException(status_code=500, detail=str(e)) +@router.post("/raw") +def exec_raw(cmd_body: Dict[str, Any]): + """ + Run raw textual nft command. Body: {"cmd": "add rule ..."} + """ + cmd = cmd_body.get("cmd") + if not isinstance(cmd, str) or not cmd.strip(): + raise HTTPException(status_code=400, detail="field 'cmd' required") + rc, out, err = mgr.cmd(cmd) + return JSONResponse(content={"rc": rc, "stdout": out or None, "stderr": err or None}, status_code=200 if rc == 0 else 400) -@router.post( - "/rules", - response_model=ExecResult, - status_code=status.HTTP_201_CREATED, - summary="Create rule (JSON, expr required; returns ExecResult)", -) -def create_rule_json(req: CreateRuleRequest): - try: - family = req.family - table = req.table - chain = req.chain +@router.post("/rules") +def create_rule(req: Dict[str, Any]): + """ + Minimal JSON-create endpoint: expects {family, table, chain, expr, [position]} + Tries to render to textual fragment via rule_text_from_expr and runs add/insert. + """ + family = req.get("family"); table = req.get("table"); chain = req.get("chain"); expr = req.get("expr") + pos = req.get("position") + if not (family and table and chain and expr is not None): + raise HTTPException(status_code=400, detail="family,table,chain,expr required") + rendered = rule_text_from_expr(expr) + if rendered is None: + raise HTTPException(status_code=400, detail="cannot render expr to textual rule; use /raw") + expr_text = rendered.strip() + if pos is not None: + try: + pos_i = int(pos) + if pos_i < 0: + pos_i = 0 + except Exception: + pos_i = 0 + cmd = f"insert rule {family} {table} {chain} position {pos_i} {expr_text}" + else: + cmd = f"add rule {family} {table} {chain} {expr_text}" + rc, out, err = mgr.cmd(cmd) + if rc == 0: + return JSONResponse(content={"rc": rc, "stdout": out or None, "stderr": err or None}, status_code=201) + # sometimes nft returns non-zero rc but no stderr while rule exists -> verify + if (rc != 0) and not (err or "").strip(): + try: + chain_text = mgr.list_chain_text(family, table, chain) or "" + if expr_text and expr_text in chain_text: + return JSONResponse(content={"rc": 0, "stdout": out or None, "stderr": err or None}, status_code=201) + except Exception: + pass + # else error + raise HTTPException(status_code=400, detail=f"nft failed rc={rc} stderr={err!r} cmd={cmd}") - if req.expr is None: - raise NftError("field 'expr' is required for JSON rule creation") - - rendered = expr_to_text(req.expr) - if rendered is None: - raise NftError("cannot render provided 'expr' to textual nft syntax; use /raw to run text command") - - 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) - rc = int(res.get("rc", -1) or -1) - stdout = res.get("stdout") or "" - stderr = res.get("stderr") or "" - - if rc == 0: - return ExecResult(rc=rc, stdout=stdout or None, stderr=stderr or None) - - # If rc indicates failure but stderr empty, check chain 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: detected rule 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(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") +@router.delete("/rules/{handle}", status_code=status.HTTP_204_NO_CONTENT) def delete_rule(handle: int, family: str = "inet", table: str = "filter", chain: str = "input"): - try: - mgr.delete_rule_by_handle_text(family=family, table=table, chain=chain, handle=handle) - except ValueError as e: - logger.warning("delete_rule client error: %s", e) - raise HTTPException(status_code=400, detail=str(e)) - except NftError as e: - logger.exception("delete_rule failed") - raise HTTPException(status_code=500, detail=str(e)) - except Exception as e: - logger.exception("delete_rule internal error") - raise HTTPException(status_code=500, detail=str(e)) - - -@router.post("/raw", response_model=ExecResult, summary="Execute raw textual nft command") -def exec_raw(req: RawCmdRequest): - try: - res = mgr.cmd(req.cmd) - rc = int(res.get("rc", -1) or -1) - return ExecResult(rc=rc, stdout=res.get("stdout") or None, stderr=res.get("stderr") or None) - except Exception as e: - logger.exception("exec_raw failed") - raise HTTPException(status_code=500, detail=str(e)) - + if not isinstance(handle, int) or handle <= 0: + raise HTTPException(status_code=400, detail="invalid handle") + cmd = f"delete rule {family} {table} {chain} handle {handle}" + rc, out, err = mgr.cmd(cmd) + if rc != 0: + raise HTTPException(status_code=500, detail=f"delete failed rc={rc} stderr={err}") + return JSONResponse(status_code=204, content={}) app.include_router(router) \ No newline at end of file