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 23:34:15 +01:00
parent 1d5750dc00
commit 5a7bce647b
1 changed files with 89 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -69,9 +69,23 @@ class NftManager:
Obtain nft list ruleset in textual form. This is a fallback if JSON parsing fails.
Returns raw textual output on success, raises NftError on failure.
"""
self.nft.set_json_output(False) # best-effort attempt to disable JSON output
# best-effort toggle JSON off for textual output
json_toggled = False
try:
if hasattr(self.nft, "set_json_output"):
try:
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("list ruleset")
self.nft.set_json_output(True) # restore JSON output preference
finally:
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)")
if res["rc"] != 0:
raise NftError(f"nft list ruleset failed: {res['stderr']}")
return res["stdout"] or ""
@@ -526,25 +540,17 @@ 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
@@ -554,8 +560,6 @@ def expr_to_text(expr: Any) -> Optional[str]:
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"
@@ -571,31 +575,22 @@ def expr_to_text(expr: Any) -> Optional[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")
@@ -605,12 +600,9 @@ def expr_to_text(expr: Any) -> Optional[str]:
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']}")
@@ -620,15 +612,10 @@ def expr_to_text(expr: Any) -> Optional[str]:
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()
@@ -637,71 +624,53 @@ def parse_ruleset_text(nft_text: str) -> Dict[Tuple[str, str, str], List[Dict[st
"""
Parse the full textual `nft list ruleset` output and return a mapping:
(family, table, chain) -> [ { "line": "<text line>", "handle": <int or None> }, ... ]
This is a best-effort parser that:
- detects table headers like: 'table <family> <name> {'
- detects chain headers like: 'chain <name> {'
- collects lines inside a chain that look like rule lines (not chain metadata like 'type ...; policy ...;')
- extracts '# handle N' when present
"""
result: Dict[Tuple[str, str, str], List[Dict[str, Optional[Union[str, int]]]]] = {}
if not nft_text:
return result
table_re = re.compile(r"^\s*table\s+(\S+)\s+(\S+)\s*\{")
# chain header line often: 'chain forward {', may include trailing comments
chain_re = re.compile(r"^\s*chain\s+(\S+)\s*\{")
handle_re = re.compile(r"#\s*handle\s*(\d+)\b")
# lines that indicate chain metadata (not rules)
chain_meta_re = re.compile(r"\b(type\b|hook\b|priority\b|policy\b|counter\b).*;")
# lines that indicate chain metadata (type/hook/priority/policy), we want to skip these
chain_meta_re = re.compile(r"^\s*(type\b|hook\b|priority\b|policy\b)\b.*;")
current_family = None
current_table = None
current_chain = None
# track depth to handle nested braces more robustly
brace_depth = 0
# iterate line-by-line
for raw_ln in nft_text.splitlines():
ln = raw_ln.rstrip()
ln = raw_ln.rstrip("\n")
s = ln.strip()
# update brace depth counting '{' and '}' to keep context
# but also we rely on explicit table/chain headers
open_count = ln.count("{")
close_count = ln.count("}")
# check for table header
# detect table header
m_table = table_re.match(ln)
if m_table:
current_family = m_table.group(1)
current_table = m_table.group(2)
current_chain = None
brace_depth += open_count - close_count
continue
# chain header (may include additional metadata and optional '# handle N')
# detect chain header (works even if the same line contains trailing comment)
m_chain = chain_re.match(ln)
if m_chain and current_family and current_table:
current_chain = m_chain.group(1)
# ensure mapping exists
key = (current_family, current_table, current_chain)
result.setdefault(key, [])
brace_depth += open_count - close_count
continue
# adjust brace depth for other lines
brace_depth += open_count - close_count
# if we're inside a chain, try to find rule-like lines
# if we're inside a chain, collect rule-like lines
if current_family and current_table and current_chain:
# skip empty or purely-brace lines
# skip empty / pure brace lines
if s == "" or s == "{" or s == "}":
continue
# skip semicolon-terminated metadata lines inside chain (type/hook/policy)
if chain_meta_re.search(s) or s.endswith(";"):
# these are chain-level metadata, not rules
# skip chain metadata lines like 'type filter hook forward priority 0;'
if chain_meta_re.match(s):
continue
# likely a rule line — extract handle if present
# likely a rule line — extract optional handle comment
m_handle = handle_re.search(s)
handle_val: Optional[int] = None
if m_handle:
@@ -711,7 +680,6 @@ def parse_ruleset_text(nft_text: str) -> Dict[Tuple[str, str, str], List[Dict[st
handle_val = None
key = (current_family, current_table, current_chain)
# store the line as-is (trim leading whitespace), and handle (if found)
result.setdefault(key, []).append({"line": ln.strip(), "handle": handle_val})
return result
@@ -721,16 +689,11 @@ def parse_ruleset_text(nft_text: str) -> Dict[Tuple[str, str, str], List[Dict[st
def populate_text_from_ruleset_text(custom: Dict[str, Any], nft_text: str) -> None:
"""
Uses the parsed full ruleset textual output (nft_text) to update rule['text']
in the 'custom' structure in-place.
Matching strategy:
1) For each chain, build mapping handle -> line.
2) For each rule in the JSON-built custom structure:
- If rule.handle exists and a matching handle line is found, use that.
- Otherwise, attempt to match by order/position: assign the i-th textual rule line
to the i-th JSON rule for that chain (best-effort).
- If position field exists in the JSON rule, prefer that index.
- If neither works, leave rule['text'] untouched.
in the 'custom' structure in-place. Matching strategy:
1) handle -> line
2) explicit 'position' field if present
3) index-based mapping (best-effort)
4) substring probe match
"""
if not nft_text:
return
@@ -748,7 +711,6 @@ def populate_text_from_ruleset_text(custom: Dict[str, Any], nft_text: str) -> No
continue
key = (fam, tname, cname)
textual_entries = parsed.get(key, [])
# build handle map and ordered lines list
handle_map: Dict[int, str] = {}
ordered_lines: List[str] = []
for ent in textual_entries:
@@ -759,29 +721,23 @@ def populate_text_from_ruleset_text(custom: Dict[str, Any], nft_text: str) -> No
handle_map[h] = ln
rules = chain.get("rules", [])
# iterate rules and apply mapping
for idx, rule in enumerate(rules):
replaced = False
h = rule.get("handle")
# 1) Try handle match if handle present
if isinstance(h, int) and h in handle_map:
rule["text"] = handle_map[h]
replaced = True
if not replaced:
# 2) Try explicit position if provided (numerical)
pos = rule.get("position")
if isinstance(pos, int) and 0 <= pos < len(ordered_lines):
rule["text"] = ordered_lines[pos]
replaced = True
if not replaced:
# 3) Try index-based match (best-effort); use idx in JSON rules order
if idx < len(ordered_lines):
if not replaced and idx < len(ordered_lines):
rule["text"] = ordered_lines[idx]
replaced = True
# 4) As a last-ditch, try substring match (probe) within ordered_lines
if not replaced:
probe = rule.get("text") or rule_text_from_expr(rule.get("expr"))
if probe:
@@ -790,14 +746,10 @@ def populate_text_from_ruleset_text(custom: Dict[str, Any], nft_text: str) -> No
rule["text"] = ln
replaced = True
break
# if nothing matched, keep existing rule['text'] (from JSON serializer)
# leave as-is if not replaced
# ---------- Helpers to render expr -> textual nft (best-effort) ----------
# (expr_to_text already defined above)
# ---------- Existing populate_text_from_chain_text (no change, used as fallback) ----------
# ---------- Existing populate_text_from_chain_text (fallback) ----------
def populate_text_from_chain_text(custom: Dict[str, Any]) -> None:
"""
Replace rule['text'] in the 'custom' predictable ruleset with the exact textual
@@ -836,17 +788,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",
@@ -864,30 +813,72 @@ 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:
nft_json = mgr.list_rules_json()
nft_text = mgr.list_rules_text()
logger.error("Successfully obtained nft JSON ruleset, but also got textual output: %r", nft_text)
except NftError as e:
logger.debug("could not obtain nft JSON ruleset: %s", e)
raise HTTPException(status_code=500, detail=f"Failed to obtain nft JSON ruleset: {e}")
# try to obtain textual ruleset too (best-effort)
nft_text = ""
try:
nft_text = mgr.list_rules_text()
except Exception:
logger.debug("could not obtain textual nft ruleset (list_rules_text failed)")
custom = build_predictable_ruleset(nft_json)
# First: try to enrich using the full textual ruleset we already fetched
# If no tables/rules were discovered from JSON, attempt to build from textual listing
try:
if nft_text and (not custom.get("tables")):
# Build minimal structure from textual parse if JSON did not produce tables
parsed = parse_ruleset_text(nft_text)
if parsed:
# convert parsed -> custom structure
tables_map: Dict[Tuple[str, str], Dict[str, Any]] = {}
for (fam, tname, cname), entries in parsed.items():
tables_map.setdefault((fam, tname), {"family": fam, "name": tname, "chains": {}})
chain_rules: List[Dict[str, Any]] = []
for ent in entries:
ln = ent.get("line") or ""
h = ent.get("handle")
# we don't have JSON expr; leave expr None and text the line
chain_rules.append({"handle": h, "expr": None, "text": ln})
tables_map[(fam, tname)]["chains"].setdefault(cname, {"name": cname, "type": None, "hook": None, "priority": None, "policy": None, "rules": []})
tables_map[(fam, tname)]["chains"][cname]["rules"].extend(chain_rules)
# convert map to custom shape
custom = {"tables": []}
for (fam, tname) in sorted(tables_map.keys(), key=lambda k: (k[0], k[1])):
t = tables_map[(fam, tname)]
chains_list = []
for cname in sorted(t["chains"].keys()):
ch = t["chains"][cname]
chains_list.append({"name": ch["name"], "type": None, "hook": None, "priority": None, "policy": None, "rules": ch["rules"]})
custom["tables"].append({"family": fam, "name": tname, "chains": chains_list})
except Exception as e:
logger.debug("fallback build from textual parse failed: %s", e)
# Enrich rule['text'] using the full textual ruleset (handle-first, position, index)
try:
if nft_text:
populate_text_from_ruleset_text(custom, nft_text)
except Exception as e:
logger.debug("list_rules: populate_text_from_ruleset_text failed: %s", e)
# per-chain fallback (will not fail overall listing)
try:
populate_text_from_chain_text(custom)
except Exception as e:
logger.debug("list_rules: populate_text_from_chain_text failed: %s", e)
# debug: log counts so you can see what's being returned
num_tables = len(custom.get("tables", []))
num_rules = sum(len(ch.get("rules", [])) for t in custom.get("tables", []) for ch in t.get("chains", []))
logger.info("list_rules: returning tables=%d rules=%d", num_tables, num_rules)
return RulesetModel.parse_obj(custom)
except NftError as e:
logger.exception("list_rules failed")
@@ -906,12 +897,6 @@ def list_rules():
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
@@ -930,12 +915,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:
@@ -947,12 +929,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:
@@ -962,33 +942,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)
@@ -997,7 +965,6 @@ 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")
@@ -1009,8 +976,6 @@ def create_rule_json(req: CreateRuleRequest):
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)