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:35:52 +01:00
parent cbd7dacd24
commit 7db2816660
1 changed files with 120 additions and 339 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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 <cmd> -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 <family> <table> <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 <family> <table> <chain> ...'.
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 <family> <table> <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 <family> <table> <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 <family> <table> <chain> position <n> <expr>
- Otherwise, uses: add rule <family> <table> <chain> <expr> (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 <n> ...' 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 <family> <table> <chain> handle <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)