marcus/mitm-webserver
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test
All checks were successful
Build and Deploy MITM Webserver / build (push) Successful in 9s
Details

Browse Source
This commit is contained in:
malmert
2026-02-28 22:41:40 +01:00
parent 216f8464f2
commit 7c5acbf758
1 changed files with 280 additions and 392 deletions
Download Patch File Download Diff File Expand all files Collapse all files

672
backend/src/api/nft_manager.py
View File

@@ -1,54 +1,44 @@
# app.py # 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 import FastAPI, APIRouter, HTTPException, status
from fastapi.responses import JSONResponse from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
import logging import logging
import json import json
import re import re
# libnftables (we call textual commands through its .cmd() method) # python-nftables wrapper
from nftables import Nftables # type: ignore from nftables import Nftables # type: ignore
# ---------- logging ----------
logging.basicConfig(level=logging.INFO) logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("nft_api_raw_only") logger = logging.getLogger("nft_api")
# ---------- Exceptions ---------- # ---------- Errors ----------
class NftError(RuntimeError): class NftError(RuntimeError):
pass pass
# ---------- NftManager (small, explicit) ----------
# ---------- NftManager (textual vs JSON) ----------
class NftManager: 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: def __init__(self) -> None:
self.nft = Nftables() 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: try:
if hasattr(self.nft, "set_json_output"): if hasattr(self.nft, "set_json_output"):
self.nft.set_json_output(True) self.nft.set_json_output(True)
except Exception: 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]]: def cmd(self, text_cmd: str) -> Tuple[int, str, str]:
"""Execute textual nft command via Nftables.cmd().""" """
Run textual nft command via Nftables.cmd() and return (rc, stdout, stderr).
"""
rc, out, err = self.nft.cmd(text_cmd) rc, out, err = self.nft.cmd(text_cmd)
if rc != 0: return int(rc), out or "", err or ""
logger.warning("nft cmd rc=%s stderr=%s cmd=%s", rc, err, text_cmd)
return {"rc": int(rc), "stdout": out or "", "stderr": err or ""}
def json_cmd(self, text_cmd: str) -> Tuple[int, str, str]: def json_cmd(self, text_cmd: str) -> Tuple[int, str, str]:
""" """
Execute nft command expecting JSON output. Run nft expecting JSON output.
Prefer Nftables.json_cmd when available (returns (rc, out, err)). Prefer Nftables.json_cmd() when present; otherwise call cmd() with -j appended.
Otherwise call cmd() with a '-j' suffix and return a similar tuple. Returns (rc, stdout, stderr).
""" """
if hasattr(self.nft, "json_cmd"): if hasattr(self.nft, "json_cmd"):
try: try:
@@ -56,79 +46,61 @@ class NftManager:
if isinstance(res, (list, tuple)) and len(res) >= 3: if isinstance(res, (list, tuple)) and len(res) >= 3:
return int(res[0]), res[1] or "", res[2] or "" return int(res[0]), res[1] or "", res[2] or ""
except Exception as e: 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" cmd_with_j = text_cmd if "-j" in text_cmd else f"{text_cmd} -j"
r = self.cmd(cmd_with_j) return self.cmd(cmd_with_j)
return int(r.get("rc", -1) or -1), r.get("stdout", "") or "", r.get("stderr", "") or ""
def list_rules(self) -> str: def list_rules_text(self) -> str:
"""Return textual ruleset from `nft list ruleset`.""" rc, out, err = self.cmd("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")
if rc != 0: if rc != 0:
raise NftError(f"nft list ruleset failed: {err}") raise NftError(f"nft list ruleset failed: {err}")
if not out: return out
raise NftError("empty JSON output from nft")
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: try:
parsed = json.loads(out) return json.loads(out)
return parsed
except json.JSONDecodeError as e: 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: def list_chain_text(self, family: str, table: str, chain: str) -> str:
""" """
Return textual output of `nft list chain <family> <table> <chain>`. Prefer textual listing via cmd(). If that fails, use JSON fallback to reconstruct reasonable lines.
Prefer textual cmd(); if it fails and JSON is returned, attempt a best-effort
reconstruction of textual lines from JSON.
""" """
cmd = f"list chain {family} {table} {chain}" cmd = f"list chain {family} {table} {chain}"
res = self.cmd(cmd) rc, out, err = self.cmd(cmd)
if res["rc"] == 0: if rc == 0:
return res["stdout"] or "" return out
# 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}")
# 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: try:
parsed = json.loads(out) parsed = json.loads(outj)
except Exception: except Exception:
# give up and return textual stdout (maybe empty) return out # return whatever textual output we had (maybe empty)
return res.get("stdout") or ""
lines: List[str] = [] lines: List[str] = []
records = parsed.get("nftables") if isinstance(parsed, dict) else (parsed or []) recs = parsed.get("nftables") if isinstance(parsed, dict) else (parsed or [])
if not isinstance(records, list): if not isinstance(recs, list):
records = [] return out
for rec in records: for rec in recs:
if "rule" in rec: if "rule" in rec:
r = rec["rule"] r = rec["rule"]
handle = r.get("handle") handle = r.get("handle")
expr = r.get("expr") expr = r.get("expr")
parts: List[str] = [] parts: List[str] = []
if isinstance(expr, list): if isinstance(expr, list):
for part in expr: for el in expr:
if isinstance(part, dict): if isinstance(el, dict):
if "payload" in part: if "match" in el:
p = part["payload"] m = el["match"]
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"]
left = m.get("left") left = m.get("left")
right = m.get("right") right = m.get("right")
if isinstance(left, dict) and "payload" in left and isinstance(right, (str, int)): if isinstance(left, dict) and "payload" in left and isinstance(right, (str, int)):
@@ -140,88 +112,54 @@ class NftManager:
continue continue
parts.append("match") parts.append("match")
continue continue
if "drop" in part: if "payload" in el:
parts.append("drop") 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 continue
if "accept" in part: if "drop" in el:
parts.append("accept") parts.append("drop"); continue
continue if "accept" in el:
if "counter" in part: parts.append("accept"); continue
parts.append("counter") if "counter" in el:
continue parts.append("counter"); continue
if "queue" in part: if "queue" in el:
q = part["queue"] q = el["queue"]
tok = "queue" tok = "queue"
if isinstance(q, dict): 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: if num is not None:
tok += f" num {num}" tok += f" num {num}"
if q.get("bypass") or q.get("flags") == "bypass": if q.get("bypass") or q.get("flags") == "bypass":
tok += " bypass" tok += " bypass"
parts.append(tok) elif isinstance(q, (int, float)):
continue tok += f" num {int(q)}"
if isinstance(q, (int, float)): parts.append(tok); continue
parts.append(f"queue num {int(q)}") parts.append("+".join(sorted(el.keys())))
continue
parts.append("queue")
continue
parts.append("+".join(sorted(part.keys())))
else: else:
parts.append(str(part)) parts.append(str(el))
else: else:
parts.append(json.dumps(r)) 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: if handle is not None:
text_line = f"{text_line} # handle {handle}" txt = f"{txt} # handle {handle}"
lines.append(text_line) lines.append(txt)
return "\n".join(lines) if lines else (res.get("stdout") or "") return "\n".join(lines) if lines else out
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']}")
# ---------- FastAPI + Router ---------- # ---------- utilities for predictable representation ----------
app = FastAPI(title="Unrestricted nftables API (json create)") _handle_re = re.compile(r"\bhandle\s+(\d+)\b")
router = APIRouter(prefix="/firewall", tags=["firewall"])
mgr = NftManager()
def parse_priority(v: Any) -> Optional[int]:
# ---------- Request/Response models (only used for validation / docs) ---------- if v is None:
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:
return None return None
if isinstance(val, int): if isinstance(v, int):
return val return v
if isinstance(val, str): if isinstance(v, str):
s = val.strip() s = v.strip()
try: try:
return int(s) return int(s)
except Exception: except Exception:
@@ -229,196 +167,177 @@ def parse_priority(val: Any) -> Optional[int]:
return int(float(s)) return int(float(s))
except Exception: except Exception:
return None return None
if isinstance(val, dict): if isinstance(v, dict):
for key in ("priority", "prio"): for k in ("priority", "prio"):
if key in val: if k in v:
return parse_priority(val.get(key)) return parse_priority(v.get(k))
for v in val.values(): for val in v.values():
p = parse_priority(v) p = parse_priority(val)
if p is not None: if p is not None:
return p return p
return None return None
def rule_text_from_expr(expr: Any) -> str: 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: if expr is None:
return "" return ""
if isinstance(expr, list): if isinstance(expr, list):
tokens: List[str] = [] out: List[str] = []
for part in expr: for el in expr:
if isinstance(part, dict): if isinstance(el, dict):
if "match" in part: if "match" in el:
m = part["match"] m = el["match"]; left = m.get("left"); right = m.get("right")
left = m.get("left") if isinstance(left, dict) and "payload" in left and isinstance(right, (str, int)):
right = m.get("right") p = left["payload"]; prot = p.get("protocol"); field = p.get("field")
if isinstance(left, dict) and "payload" in left and isinstance(right, str):
p = left["payload"]
prot = p.get("protocol")
field = p.get("field")
if prot and field: if prot and field:
tokens.append(f"{prot} {field} {right}") out.append(f"{prot} {field} {right}"); continue
continue out.append("match"); continue
tokens.append("match") if "payload" in el:
elif "payload" in part: p = el["payload"]; prot = p.get("protocol"); field = p.get("field")
p = part["payload"]
prot = p.get("protocol")
field = p.get("field")
if prot and field: if prot and field:
tokens.append(f"payload({prot}.{field})") out.append(f"payload({prot}.{field})"); continue
continue out.append("payload"); continue
tokens.append("payload") if "drop" in el:
elif "drop" in part: out.append("drop"); continue
tokens.append("drop") if "accept" in el:
elif "accept" in part: out.append("accept"); continue
tokens.append("accept") if "counter" in el:
elif "counter" in part: out.append("counter"); continue
tokens.append("counter") if "queue" in el:
elif "tcp" in part or "udp" in part: q = el["queue"]; tok = "queue"
proto = "tcp" if "tcp" in part else "udp"
tokens.append(proto)
elif "queue" in part:
q = part["queue"]
token = "queue"
if isinstance(q, dict): 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: if num is not None:
token += f" num {num}" tok += f" num {num}"
if q.get("bypass") or q.get("flags") == "bypass": if q.get("bypass") or q.get("flags") == "bypass":
token += " bypass" tok += " bypass"
elif isinstance(q, (int, float)): elif isinstance(q, (int, float)):
token += f" num {int(q)}" tok += f" num {int(q)}"
elif isinstance(q, str): out.append(tok); continue
token += f" num {q}" # unknown dict -> show keys
tokens.append(token) out.append("+".join(sorted(el.keys())))
else:
tokens.append("+".join(sorted(part.keys())))
else: else:
tokens.append(str(part)) out.append(str(el))
return " ".join(tokens) return " ".join(out)
return str(expr) # fallback
try:
return str(expr)
except Exception:
return json.dumps(expr)
def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]: 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": []} result: Dict[str, Any] = {"tables": []}
items = nft_json.get("nftables", []) if isinstance(nft_json, dict) else (nft_json or []) items = nft_json.get("nftables", []) if isinstance(nft_json, dict) else (nft_json or [])
tables: Dict[Tuple[str, str], Dict[str, Any]] = {} tables: Dict[Tuple[str, str], Dict[str, Any]] = {}
for rec in items: for rec in items:
if "table" in rec: if "table" in rec:
t = rec["table"] t = rec["table"]; fam = t.get("family"); name = t.get("name")
fam = t.get("family")
name = t.get("name")
if fam and name: if fam and name:
tables.setdefault((fam, name), {"family": fam, "name": name, "chains": {}}) tables.setdefault((fam, name), {"family": fam, "name": name, "chains": {}})
elif "chain" in rec: elif "chain" in rec:
ch = rec["chain"] c = rec["chain"]
fam = ch.get("family") or (ch.get("table") or {}).get("family") fam = c.get("family") or (c.get("table") or {}).get("family")
table_name = ch.get("table") or (ch.get("table") or {}).get("name") tname = c.get("table") or (c.get("table") or {}).get("name")
cname = ch.get("name") cname = c.get("name")
if fam and table_name and cname: if fam and tname and cname:
tables.setdefault((fam, table_name), {"family": fam, "name": table_name, "chains": {}}) tables.setdefault((fam, tname), {"family": fam, "name": tname, "chains": {}})
chains_map = tables[(fam, table_name)]["chains"] chains = tables[(fam, tname)]["chains"]
existing = chains_map.get(cname) existing = chains.get(cname)
ch_type = ch.get("type") ch_type = c.get("type"); ch_hook = c.get("hook")
ch_hook = ch.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_priority = parse_priority(ch.get("priority") if "priority" in ch else ch.get("prio") if "prio" in ch else ch.get("priority", None)) ch_policy = c.get("policy")
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: if existing is None:
chains_map[cname] = { chains[cname] = {"name": cname, "type": ch_type, "hook": ch_hook, "priority": ch_prio, "policy": ch_policy, "rules": []}
"name": cname,
"type": ch_type,
"hook": ch_hook,
"priority": ch_priority,
"policy": ch_policy,
"rules": [],
}
else: else:
if isinstance(existing, dict): if existing.get("type") is None and ch_type is not None:
if existing.get("type") is None and ch_type is not None: existing["type"] = ch_type
existing["type"] = ch_type if existing.get("hook") is None and ch_hook is not None:
if existing.get("hook") is None and ch_hook is not None: existing["hook"] = ch_hook
existing["hook"] = ch_hook if existing.get("priority") is None and ch_prio is not None:
if existing.get("priority") is None and ch_priority is not None: existing["priority"] = ch_prio
existing["priority"] = ch_priority if existing.get("policy") is None and ch_policy is not None:
if existing.get("policy") is None and ch_policy is not None: existing["policy"] = ch_policy
existing["policy"] = ch_policy
elif "rule" in rec: elif "rule" in rec:
r = rec["rule"] r = rec["rule"]
fam = r.get("family") fam = r.get("family"); tname = r.get("table"); cname = r.get("chain")
table_name = r.get("table") if not (fam and tname and cname):
chain_name = r.get("chain") continue
handle = r.get("handle") tables.setdefault((fam, tname), {"family": fam, "name": tname, "chains": {}})
expr = r.get("expr") chains = tables[(fam, tname)]["chains"]
if fam and table_name and chain_name: chains.setdefault(cname, {"name": cname, "type": None, "hook": None, "priority": None, "policy": None, "rules": []})
tables.setdefault((fam, table_name), {"family": fam, "name": table_name, "chains": {}}) rule_obj = {
chains_map = tables[(fam, table_name)]["chains"] "handle": r.get("handle"),
chains_map.setdefault(chain_name, {"name": chain_name, "type": None, "hook": None, "priority": None, "policy": None, "rules": []}) "expr": r.get("expr"),
rule_obj: Dict[str, Any] = { # initial text from expr serializer (may be replaced later by exact textual line)
"handle": handle, "text": rule_text_from_expr(r.get("expr")),
"expr": expr, }
"text": rule_text_from_expr(expr), if "position" in r:
} rule_obj["position"] = r["position"]
if "position" in r: if "comment" in r:
rule_obj["position"] = r["position"] rule_obj["comment"] = r["comment"]
if "comment" in r: chains[cname]["rules"].append(rule_obj)
rule_obj["comment"] = r["comment"]
chains_map[chain_name]["rules"].append(rule_obj)
if isinstance(r.get("chain"), dict): # salvage chain metadata if nested under rule (some nft outputs)
csub = r.get("chain") if isinstance(r.get("chain"), dict):
if chains_map[chain_name].get("priority") is None: csub = r.get("chain")
parsed_prio = parse_priority(csub.get("priority") if "priority" in csub else csub.get("prio")) if chains[cname].get("priority") is None:
if parsed_prio is not None: p = parse_priority(csub.get("priority") if "priority" in csub else csub.get("prio"))
chains_map[chain_name]["priority"] = parsed_prio if p is not None:
if chains_map[chain_name].get("type") is None and csub.get("type") is not None: chains[cname]["priority"] = p
chains_map[chain_name]["type"] = csub.get("type") if chains[cname].get("type") is None and csub.get("type") is not None:
if chains_map[chain_name].get("hook") is None and csub.get("hook") is not None: chains[cname]["type"] = csub.get("type")
chains_map[chain_name]["hook"] = csub.get("hook") if chains[cname].get("hook") is None and csub.get("hook") is not None:
if chains_map[chain_name].get("policy") is None and csub.get("policy") is not None: chains[cname]["hook"] = csub.get("hook")
chains_map[chain_name]["policy"] = csub.get("policy") 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])): for (fam, tname) in sorted(tables.keys(), key=lambda k: (k[0], k[1])):
tdata = tables[(fam, tname)] tdata = tables[(fam, tname)]
chains_list: List[Dict[str, Any]] = [] chains_list: List[Dict[str, Any]] = []
for cname in sorted(tdata["chains"].keys()): for cname in sorted(tdata["chains"].keys()):
chdata = tdata["chains"][cname] ch = tdata["chains"][cname]
chains_list.append( chains_list.append({
{ "name": ch.get("name"),
"name": chdata.get("name"), "type": ch.get("type"),
"type": chdata.get("type"), "hook": ch.get("hook"),
"hook": chdata.get("hook"), "priority": ch.get("priority"),
"priority": chdata.get("priority"), "policy": ch.get("policy"),
"policy": chdata.get("policy"), "rules": ch.get("rules", []),
"rules": chdata.get("rules", []), })
}
)
result["tables"].append({"family": fam, "name": tname, "chains": chains_list}) result["tables"].append({"family": fam, "name": tname, "chains": chains_list})
return result return result
def enrich_text_from_chain(custom: Dict[str, Any], mgr: NftManager) -> None:
def populate_text_from_chain_text(custom: Dict[str, Any]) -> None: """
tables = custom.get("tables") or [] Replace rule['text'] with exact textual lines from `nft list chain` when possible.
for t in tables: Looks up rule handles; if not found it tries substring matching on the compact probe.
fam = t.get("family") """
tname = t.get("name") for table in custom.get("tables", []):
fam = table.get("family"); tname = table.get("name")
if not fam or not tname: if not fam or not tname:
continue continue
for ch in t.get("chains", []): for ch in table.get("chains", []):
cname = ch.get("name") cname = ch.get("name")
if not cname: if not cname:
continue continue
try: try:
chain_text = mgr.list_chain_text(fam, tname, cname) or "" txt = mgr.list_chain_text(fam, tname, cname) or ""
lines = [ln.rstrip() for ln in chain_text.splitlines() if ln.strip() != ""] lines = [ln.strip() for ln in txt.splitlines() if ln.strip()]
handle_map: Dict[str, str] = {} # map handle -> line
handle_map = {}
for ln in lines: for ln in lines:
m = re.search(r"\bhandle\s+(\d+)\b", ln) m = _handle_re.search(ln)
if m: if m:
handle_map[m.group(1)] = ln.strip() handle_map[m.group(1)] = ln
for rule in ch.get("rules", []): for rule in ch.get("rules", []):
replaced = False replaced = False
h = rule.get("handle") h = rule.get("handle")
@@ -427,143 +346,112 @@ def populate_text_from_chain_text(custom: Dict[str, Any]) -> None:
if key in handle_map: if key in handle_map:
rule["text"] = handle_map[key] rule["text"] = handle_map[key]
replaced = True replaced = True
if not replaced: if not replaced:
expr = rule.get("expr") probe = rule.get("text") or rule_text_from_expr(rule.get("expr"))
probe = rule.get("text") or rule_text_from_expr(expr)
if probe: if probe:
for ln in lines: for ln in lines:
if probe in ln: if probe in ln:
rule["text"] = ln.strip() rule["text"] = ln
replaced = True replaced = True
break break
except Exception as e: 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 continue
# ---------- FastAPI app ----------
app = FastAPI(title="nft API (clean)")
router = APIRouter(prefix="/firewall", tags=["firewall"])
mgr = NftManager()
# ---------- Routes ---------- @router.get("/rules")
@router.get("/rules", summary="List ruleset") def get_rules():
def list_rules():
""" """
Returns JSON: { "ruleset": <dict | textual-string-fallback> } Return: JSONResponse({"ruleset": <dict-or-string>})
Ensures `ruleset` is a native dict when JSON is available and parsed. - If nft -j is available -> returns native dict under "ruleset"
- If not, returns the textual ruleset string under "ruleset"
""" """
try: try:
try: try:
nft_json = mgr.list_rules_json() nft_json = mgr.list_rules_json()
except NftError as e: except NftError as e:
logger.debug("could not obtain nft JSON ruleset: %s", e) logger.debug("json listing unavailable: %s", e)
# textual fallback - return plain textual ruleset string # fallback to textual listing (return string; not JSON-encoded)
text = mgr.list_rules() text = mgr.list_rules_text()
return JSONResponse(content={"ruleset": text.strip() if text is not None else None}, status_code=200) 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) custom = build_predictable_ruleset(nft_json)
# best-effort: replace compact text with exact lines
# Try to replace rule['text'] with exact textual lines from nft list chain ...
try: try:
populate_text_from_chain_text(custom) enrich_text_from_chain(custom, mgr)
except Exception as e: 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) return JSONResponse(content={"ruleset": custom}, status_code=200)
except NftError as e: except NftError as e:
logger.exception("list_rules failed") logger.exception("get_rules: nft error")
raise HTTPException(status_code=500, detail=str(e)) raise HTTPException(status_code=500, detail=str(e))
except Exception as e: except Exception as e:
logger.exception("list_rules internal error") logger.exception("get_rules internal")
raise HTTPException(status_code=500, detail=str(e)) 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( @router.post("/rules")
"/rules", def create_rule(req: Dict[str, Any]):
response_model=ExecResult, """
status_code=status.HTTP_201_CREATED, Minimal JSON-create endpoint: expects {family, table, chain, expr, [position]}
summary="Create rule (JSON, expr required; returns ExecResult)", Tries to render to textual fragment via rule_text_from_expr and runs add/insert.
) """
def create_rule_json(req: CreateRuleRequest): family = req.get("family"); table = req.get("table"); chain = req.get("chain"); expr = req.get("expr")
try: pos = req.get("position")
family = req.family if not (family and table and chain and expr is not None):
table = req.table raise HTTPException(status_code=400, detail="family,table,chain,expr required")
chain = req.chain 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: @router.delete("/rules/{handle}", status_code=status.HTTP_204_NO_CONTENT)
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")
def delete_rule(handle: int, family: str = "inet", table: str = "filter", chain: str = "input"): def delete_rule(handle: int, family: str = "inet", table: str = "filter", chain: str = "input"):
try: if not isinstance(handle, int) or handle <= 0:
mgr.delete_rule_by_handle_text(family=family, table=table, chain=chain, handle=handle) raise HTTPException(status_code=400, detail="invalid handle")
except ValueError as e: cmd = f"delete rule {family} {table} {chain} handle {handle}"
logger.warning("delete_rule client error: %s", e) rc, out, err = mgr.cmd(cmd)
raise HTTPException(status_code=400, detail=str(e)) if rc != 0:
except NftError as e: raise HTTPException(status_code=500, detail=f"delete failed rc={rc} stderr={err}")
logger.exception("delete_rule failed") return JSONResponse(status_code=204, content={})
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))
app.include_router(router) app.include_router(router)