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:45:20 +01:00
parent 131d0eabe8
commit 2033d2fce2
1 changed files with 136 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,7 +1,7 @@
# app.py
from typing import Any, Dict, List, Optional, Tuple, Union
from fastapi import FastAPI, APIRouter, HTTPException, status
from pydantic import BaseModel, Field
from pydantic import BaseModel, Field, ValidationError
import logging
import json
import re
@@ -23,10 +23,11 @@ class NftManager:
def __init__(self) -> None:
self.nft = Nftables()
try:
# prefer JSON output globally where available
self.nft.set_json_output(True)
self.nft.set_handle_output(True)
except Exception:
logger.debug("set_json_output not available or ignored")
logger.debug("set_json_output/set_handle_output not available")
def cmd(self, text_cmd: str) -> Dict[str, Optional[str]]:
rc, out, err = self.nft.cmd(text_cmd)
@@ -47,6 +48,7 @@ class NftManager:
raise NftError(f"json decode error: {e}")
def list_rules_text(self) -> str:
# best-effort: temporarily disable JSON output so we get textual form
json_toggled = False
try:
if hasattr(self.nft, "set_json_output"):
@@ -54,14 +56,14 @@ class NftManager:
self.nft.set_json_output(False)
json_toggled = True
except Exception:
logger.debug("could not toggle set_json_output(False); will try command anyway")
logger.debug("could not toggle set_json_output(False)")
res = self.cmd("list ruleset")
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)")
logger.debug("could not restore set_json_output(True)")
if res["rc"] != 0:
raise NftError(f"nft list ruleset failed: {res['stderr']}")
@@ -77,24 +79,26 @@ class NftManager:
self.nft.set_json_output(False)
json_toggled = True
except Exception:
logger.debug("could not toggle set_json_output(False); will try command anyway")
logger.debug("could not toggle set_json_output(False)")
res = self.cmd(cmd)
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)")
logger.debug("could not restore set_json_output(True)")
if res["rc"] != 0:
raise NftError(f"nft {cmd} failed: {res['stderr']}")
# If returned output is JSON-formatted (rare), we try to derive textual lines,
# otherwise return raw textual output.
out = res["stdout"] or ""
s = out.strip()
if s.startswith("{") or s.startswith("["):
try:
parsed = json.loads(s)
rule_lines: List[str] = []
lines: List[str] = []
records = parsed.get("nftables") if isinstance(parsed, dict) else parsed
if not isinstance(records, list):
records = []
@@ -103,58 +107,39 @@ class NftManager:
r = rec["rule"]
expr = r.get("expr")
if isinstance(expr, list):
tokens: List[str] = []
# summarized tokenization
toks: List[str] = []
for part in expr:
if isinstance(part, dict) and "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}")
toks.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})")
toks.append("match")
elif isinstance(part, dict) and "drop" in part:
tokens.append("drop")
toks.append("drop")
elif isinstance(part, dict) and "accept" in part:
tokens.append("accept")
toks.append("accept")
elif isinstance(part, dict) and "counter" in part:
tokens.append("counter")
elif isinstance(part, dict) and "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"
if num is not None:
tok += f" num {num}"
if q.get("bypass"):
tok += " bypass"
tokens.append(tok)
else:
if isinstance(q, (int, float)):
tokens.append(f"queue num {int(q)}")
else:
tokens.append(f"queue num {q}")
toks.append("counter")
else:
if isinstance(part, dict):
tokens.append("+".join(part.keys()))
toks.append("+".join(part.keys()))
else:
tokens.append(str(part))
rule_lines.append(" ".join(tokens))
toks.append(str(part))
lines.append(" ".join(toks))
else:
rule_lines.append(json.dumps(r))
if rule_lines:
return "\n".join(rule_lines)
lines.append(json.dumps(r))
if lines:
return "\n".join(lines)
except Exception:
logger.debug("fallback JSON parsing of chain output failed; returning raw output")
logger.debug("fallback json->text derivation failed; returning raw output")
return out
@@ -258,6 +243,7 @@ def parse_priority(val: Any) -> Optional[int]:
def rule_text_from_expr(expr: Any) -> str:
if expr is None:
return ""
# try list first (common)
if isinstance(expr, list):
tokens: List[str] = []
for part in expr:
@@ -266,7 +252,7 @@ def rule_text_from_expr(expr: Any) -> str:
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")
@@ -282,8 +268,6 @@ def rule_text_from_expr(expr: Any) -> str:
tokens.append(f"payload({prot}.{field})")
continue
tokens.append("payload")
elif "cmp" in part or "binary" in part:
tokens.append("cmp")
elif "drop" in part:
tokens.append("drop")
elif "accept" in part:
@@ -313,12 +297,20 @@ def rule_text_from_expr(expr: Any) -> str:
else:
tokens.append(str(part))
return " ".join(tokens)
return str(expr)
# if expr is a dict or other type, give a compact repr
try:
return str(expr)
except Exception:
return ""
def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]:
"""
Convert nft -j list ruleset parsed JSON into a deterministic, predictable JSON:
"""
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:
@@ -372,13 +364,16 @@ def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]:
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": []})
# we keep expr exactly as provided from JSON; do not replace it
# do NOT change expr shape here; keep it exactly as NFT JSON provided
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)
# salvage chain-level metadata from rule record if present
if isinstance(r.get("chain"), dict):
csub = r.get("chain")
if chains_map[chain_name].get("priority") is None:
@@ -391,6 +386,8 @@ def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]:
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")
# convert to lists (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]] = []
@@ -407,6 +404,7 @@ def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]:
}
)
result["tables"].append({"family": fam, "name": tname, "chains": chains_list})
return result
@@ -419,6 +417,7 @@ def parse_ruleset_text(nft_text: str) -> Dict[Tuple[str, str, str], List[Dict[st
table_re = re.compile(r"^\s*table\s+(\S+)\s+(\S+)\s*\{")
chain_re = re.compile(r"^\s*chain\s+(\S+)\s*\{")
handle_re = re.compile(r"#\s*handle\s*(\d+)\b")
# only skip semicolon-terminated chain metadata lines (type/hook/priority/policy)
chain_meta_re = re.compile(r"^\s*(type\b|hook\b|priority\b|policy\b)\b.*;")
current_family = None
@@ -447,6 +446,7 @@ def parse_ruleset_text(nft_text: str) -> Dict[Tuple[str, str, str], List[Dict[st
if s == "" or s == "{" or s == "}":
continue
if chain_meta_re.match(s):
# skip chain metadata lines only
continue
m_handle = handle_re.search(s)
@@ -464,11 +464,14 @@ 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:
"""
Enrich JSON-derived 'custom' structure in-place by setting only rule['text']
when a reliable textual mapping is found. Do not change expr or other types.
"""
if not nft_text:
return
parsed = parse_ruleset_text(nft_text)
for table in custom.get("tables", []):
fam = table.get("family")
tname = table.get("name")
@@ -481,7 +484,7 @@ def populate_text_from_ruleset_text(custom: Dict[str, Any], nft_text: str) -> No
key = (fam, tname, cname)
textual_entries = parsed.get(key, [])
if not textual_entries:
continue # nothing to enrich for this chain
continue
handle_map: Dict[int, str] = {}
ordered_lines: List[str] = []
@@ -494,37 +497,41 @@ def populate_text_from_ruleset_text(custom: Dict[str, Any], nft_text: str) -> No
rules = chain.get("rules", [])
for idx, rule in enumerate(rules):
replaced = False
# ensure we don't remove/overwrite expr
# ONLY update 'text' when we can map a textual line
h = rule.get("handle")
mapped: Optional[str] = None
if isinstance(h, int) and h in handle_map:
rule["text"] = handle_map[h]
replaced = True
if not replaced:
mapped = handle_map[h]
else:
pos = rule.get("position")
if isinstance(pos, int) and 0 <= pos < len(ordered_lines):
rule["text"] = ordered_lines[pos]
replaced = True
mapped = ordered_lines[pos]
elif idx < len(ordered_lines):
mapped = ordered_lines[idx]
if not replaced and idx < len(ordered_lines):
rule["text"] = ordered_lines[idx]
replaced = True
if not replaced:
# final substring probe (safe)
if mapped is None:
probe = rule.get("text") or rule_text_from_expr(rule.get("expr"))
if probe:
for ln in ordered_lines:
if probe in ln:
rule["text"] = ln
replaced = True
mapped = ln
break
# only update 'text' when a match was found; otherwise leave existing rule['text']
if mapped is not None:
# ensure we only write a str into 'text'
try:
rule["text"] = str(mapped)
except Exception:
rule["text"] = mapped # should be str already
def populate_text_from_chain_text(custom: Dict[str, Any]) -> None:
tables = custom.get("tables") or []
for t in tables:
"""
Per-chain fallback that queries nft for the chain's textual listing and
updates only rule['text'] when a reliable mapping is found.
"""
for t in custom.get("tables", []):
fam = t.get("family")
tname = t.get("name")
if not fam or not tname:
@@ -545,88 +552,84 @@ def populate_text_from_chain_text(custom: Dict[str, Any]) -> None:
except Exception:
continue
for rule in ch.get("rules", []):
replaced = False
rules = ch.get("rules", [])
for idx, rule in enumerate(rules):
mapped: Optional[str] = None
h = rule.get("handle")
if isinstance(h, int) and h in handle_map:
rule["text"] = handle_map[h]
replaced = True
if not replaced:
expr = rule.get("expr")
probe = rule.get("text") or rule_text_from_expr(expr)
mapped = handle_map[h]
else:
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()
replaced = True
mapped = ln.strip()
break
if mapped is not None:
try:
rule["text"] = str(mapped)
except Exception:
rule["text"] = mapped
except Exception as e:
logger.debug(
"populate_text_from_chain_text: failed to get textual chain for %s %s %s: %s",
fam,
tname,
cname,
e,
)
logger.debug("populate_text_from_chain_text failed for %s %s %s: %s", fam, tname, cname, e)
continue
# ---------- Normalization helper to satisfy Pydantic model ----------
# ---------- Normalization helper (lightweight and safe) ----------
def normalize_custom_for_model(custom: Dict[str, Any]) -> None:
"""
Ensure every rule has required keys/types for RulesetModel parsing:
- 'expr' must exist (if missing/None -> set to [])
- 'text' must exist and be a str (if missing -> derive via rule_text_from_expr or "")
- 'handle' coerced to int or None
This mutates `custom` in-place.
Make minimal, safe guarantees required by Pydantic:
- rule['expr'] must exist (if None -> set to empty list)
- rule['text'] must be a str (if missing -> derived string)
- rule['handle'] coerced to int or None
Do NOT change any other shapes.
"""
tables = custom.get("tables") or []
for t in tables:
for t in custom.get("tables", []):
for ch in t.get("chains", []):
rules = ch.get("rules", []) or []
for r in rules:
# expr: required by model; prefer existing value, but don't leave None
# expr: if missing or None => set to [] (preserves Any)
if "expr" not in r or r.get("expr") is None:
# default to empty list (acceptable Any value and consistent with prior shape)
r["expr"] = []
# text: must be str
txt = r.get("text")
if txt is None:
# text: ensure string
if "text" not in r or r.get("text") is None:
try:
r["text"] = rule_text_from_expr(r.get("expr"))
except Exception:
r["text"] = ""
else:
# coerce to str if not string
if not isinstance(txt, str):
if not isinstance(r["text"], str):
try:
r["text"] = str(txt)
r["text"] = str(r["text"])
except Exception:
r["text"] = ""
# handle: ensure int or None
# handle: coerce to int or None
h = r.get("handle")
if isinstance(h, str):
try:
r["handle"] = int(h)
except Exception:
r["handle"] = None
elif not isinstance(h, int):
# if it's e.g. float, bool, etc. convert if reasonable
elif isinstance(h, float):
try:
if isinstance(h, (float,)):
r["handle"] = int(h)
else:
r["handle"] = None
r["handle"] = int(h)
except Exception:
r["handle"] = None
elif not isinstance(h, int):
r["handle"] = None
# ---------- Routes ----------
@router.get("/rules", response_model=RulesetOut, summary="List ruleset")
def list_rules():
"""
Returns JSON-derived ruleset (RulesetModel) and enriches each rule['text']
with the textual nft rule line when possible. This function will not replace
JSON-derived 'expr' or other data types — enrichment is additive only.
"""
try:
try:
nft_json = mgr.list_rules_json()
@@ -634,41 +637,51 @@ def list_rules():
logger.debug("could not obtain nft JSON ruleset: %s", e)
raise HTTPException(status_code=500, detail=f"Failed to obtain nft JSON ruleset: {e}")
# get textual snapshot (best-effort), used only to enrich 'text' values
# best-effort textual snapshot for enrichment
nft_text = ""
try:
nft_text = mgr.list_rules_text()
except Exception:
logger.debug("could not obtain textual nft ruleset (list_rules_text failed)")
logger.debug("could not obtain textual nft ruleset snapshot")
# build predictable JSON-derived structure (must be the source of truth)
# Build canonical JSON-derived shape (source of truth)
custom = build_predictable_ruleset(nft_json)
# enrich existing JSON rules with textual lines (handle -> line; position/index fallback)
# Enrich only the 'text' field in-place using the textual snapshot
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)
logger.debug("populate_text_from_ruleset_text failed: %s", e)
# per-chain fallback that queries nft for chain-level textual lines
# Per-chain fallback using `nft list chain ...`
try:
populate_text_from_chain_text(custom)
except Exception as e:
logger.debug("list_rules: populate_text_from_chain_text failed: %s", e)
logger.debug("populate_text_from_chain_text failed: %s", e)
# Normalize to satisfy Pydantic model (do this last)
# Normalize minimally for model validation
normalize_custom_for_model(custom)
# Debug: log counts
# Debug counts
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)
# RETURN a shape matching response_model=RulesetOut
try:
ruleset_model = RulesetModel.parse_obj(custom)
except ValidationError as ve:
# log full validation error for debugging and return 500 with message
logger.exception("RulesetModel validation failed: %s", ve)
raise HTTPException(status_code=500, detail=f"Internal: ruleset validation failed: {ve}")
return {"ruleset": ruleset_model}
except NftError as e:
logger.exception("list_rules failed")
raise HTTPException(status_code=500, detail=str(e))
except HTTPException:
raise
except Exception as e:
logger.exception("list_rules internal error")
raise HTTPException(status_code=500, detail=str(e))
@@ -689,16 +702,10 @@ def create_rule_json(req: CreateRuleRequest):
if req.expr is None:
raise NftError("field 'expr' is required for JSON rule creation")
rendered = None
try:
rendered = rule_text_from_expr(req.expr)
except Exception:
rendered = None
if rendered is None:
rendered = rule_text_from_expr(req.expr)
if rendered is None or rendered == "":
raise NftError(
"cannot render provided 'expr' to textual nft syntax. "
"Please use POST /firewall/raw to execute the textual nft command."
"cannot render provided 'expr' to textual nft syntax. Use POST /firewall/raw"
)
expr_text = rendered.strip()
@@ -714,8 +721,7 @@ def create_rule_json(req: CreateRuleRequest):
else:
cmd = f"add rule {family} {table} {chain} {expr_text}"
logger.info("create_rule_json executing command: %s", cmd)
logger.info("create_rule_json executing: %s", cmd)
res = mgr.cmd(cmd)
raw_rc = res.get("rc")
stdout = res.get("stdout") or ""
@@ -733,22 +739,19 @@ def create_rule_json(req: CreateRuleRequest):
if rc == 0:
return exec_res
# If non-zero rc but empty stderr, try to verify presence
if (rc < 0 or rc != 0) and stderr.strip() == "":
logger.debug("create_rule_json: rc indicates failure but stderr empty; verifying rule presence")
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 in chain after add; treating as success")
logger.info("create_rule_json: rule detected after add; treating as success")
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)
logger.warning("create_rule_json verification failed: %s", e_chain)
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)
except NftError as e:
logger.warning("create_rule_json NftError: %s", e)
raise HTTPException(status_code=400, detail=str(e))