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

qdfwd
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:11:23 +01:00
parent 5930794a4e
commit 3a5eb0a46d
1 changed files with 239 additions and 269 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -91,6 +91,7 @@ class NftManager:
cmd = f"list chain {family} {table} {chain}" cmd = f"list chain {family} {table} {chain}"
# Attempt to temporarily disable JSON output on the wrapper (best-effort). # Attempt to temporarily disable JSON output on the wrapper (best-effort).
json_toggled = False json_toggled = False
res = {"rc": -1, "stdout": "", "stderr": "unknown"}
try: try:
if hasattr(self.nft, "set_json_output"): if hasattr(self.nft, "set_json_output"):
try: try:
@@ -325,285 +326,73 @@ def parse_priority(val: Any) -> Optional[int]:
return None return None
# ------------------------- def rule_text_from_expr(expr: Any) -> str:
# Expr serializer helpers
# -------------------------
def _compact_json_fragment(obj: Any) -> str:
""" """
Return a very compact JSON fragment for unknown tokens to include inline in text Deterministic serializer to produce a compact UI-friendly string from expr list.
(keeps text deterministic and safe). Covers common constructs; falls back to JSON dump for unknown constructs.
""" (Used for display in GET /rules).
try:
return json.dumps(obj, separators=(",", ":"), ensure_ascii=False)
except Exception:
return str(obj)
def _stringify_value(v: Any) -> str:
"""
Convert RHS values to a deterministic textual form:
- strings -> raw
- numbers -> str
- list/sets -> "{a,b,c}"
- dict with 'start'/'end' -> "start-end" (range style)
- boolean -> "true"/"false"
"""
if v is None:
return "None"
if isinstance(v, bool):
return "true" if v else "false"
if isinstance(v, (int, float)):
# preserve integer appearance if possible
if isinstance(v, int) or float(v).is_integer():
return str(int(v))
return str(v)
if isinstance(v, str):
return v
if isinstance(v, (list, tuple, set)):
inner = ",".join(sorted(map(str, v))) if not isinstance(v, set) else ",".join(sorted(map(str, v)))
return "{" + inner + "}"
if isinstance(v, dict):
# common NFT range shape: {"start": "10.0.0.1", "end":"10.0.0.255"} or numeric equivalent
if "start" in v and "end" in v:
return f"{_stringify_value(v['start'])}-{_stringify_value(v['end'])}"
# fallback: compact fragment
return _compact_json_fragment(v)
return str(v)
def _render_match(m: Dict[str, Any]) -> Optional[str]:
"""
Render a 'match' dict into textual piece, best-effort.
Supports:
- payload left/right equality: {'left': {'payload': {'protocol':'ip','field':'protocol'}}, 'op':'==', 'right': 'icmp'}
- payload field equals port or address
- IN / not in via op 'in' or 'not in'
- ranges expressed as dict or as right 'range'
Returns None if completely unknown.
"""
if not isinstance(m, dict):
return None
left = m.get("left")
right = m.get("right")
op = m.get("op") or m.get("operator") or m.get("type") or "=="
# Helper: payload left
if isinstance(left, dict) and "payload" in left:
p = left["payload"]
prot = p.get("protocol")
field = p.get("field")
# if right is dict/list/str/number, stringify deterministically
rtxt = _stringify_value(right)
if prot and field:
# typical: protocol field
if field == "protocol":
# e.g. "ip protocol icmp" (right usually string)
return f"{prot} {field} {rtxt}"
# address/port fields
return f"payload({prot}.{field}) {op} {rtxt}"
# left could be dict with 'meta' or 'ct' selectors or direct field names
if isinstance(left, dict) and "meta" in left:
# e.g. meta l4proto
mdata = left["meta"]
if isinstance(mdata, dict):
key = next(iter(mdata.keys()), None)
val = mdata.get(key) if key else None
return f"meta {key} {_stringify_value(val)}"
# left as simple string (rare) or numeric field name
if isinstance(left, str):
return f"{left} {op} {_stringify_value(right)}"
# Left may be a two-sided cmp e.g., {'left': {'payload':...}, 'right': {'payload':...}}
if isinstance(left, dict) and isinstance(right, dict):
# try to render both sides if they contain payloads
if "payload" in left and "payload" in right:
lp = left["payload"]
rp = right["payload"]
ltxt = f"{lp.get('protocol')}.{lp.get('field')}" if lp else _compact_json_fragment(left)
rtxt = f"{rp.get('protocol')}.{rp.get('field')}" if rp else _compact_json_fragment(right)
return f"{ltxt} {op} {rtxt}"
# Fallback: include compact JSON fragment if we cannot deterministically render
return f"match {op} {_compact_json_fragment({'left': left, 'right': right})}"
def _serialize_expr(expr: Any) -> Optional[str]:
"""
Robust serializer for an nft JSON expr (list) -> textual fragment.
Returns a string (possibly verbose) or None if totally unsupported.
The intent is to produce deterministic, readable text for the UI.
""" """
if expr is None: if expr is None:
return "" return ""
if isinstance(expr, str): if isinstance(expr, list):
return expr tokens: List[str] = []
if not isinstance(expr, list): for part in expr:
# unsupported top-level type -> pretty-print compact if isinstance(part, dict):
return _compact_json_fragment(expr) # common tokens
if "match" in part:
tokens: List[str] = [] m = part["match"]
for el in expr: left = m.get("left")
# handle simple dict tokens right = m.get("right")
if isinstance(el, dict): if isinstance(left, dict) and "payload" in left and isinstance(right, str):
# direct known keywords p = left["payload"]
if "drop" in el: prot = p.get("protocol")
tokens.append("drop") field = p.get("field")
continue if prot and field:
if "accept" in el: tokens.append(f"{prot} {field} {right}")
tokens.append("accept") continue
continue tokens.append("match")
if "counter" in el: elif "payload" in part:
tokens.append("counter") p = part["payload"]
continue
if "return" in el:
tokens.append("return")
continue
if "reject" in el:
# reject may be a string reason or dict
rv = el.get("reject")
if isinstance(rv, str):
tokens.append(f"reject {rv}")
elif isinstance(rv, dict):
tokens.append(f"reject {_compact_json_fragment(rv)}")
else:
tokens.append("reject")
continue
# queue may be int/string or dict
if "queue" in el:
q = el["queue"]
tok = "queue"
if isinstance(q, dict):
num = q.get("num") or q.get("number") or q.get("queue_number") or q.get("from") or q.get("range")
if num is not None:
tok += f" num {num}"
if q.get("bypass"):
tok += " bypass"
elif isinstance(q, (int, float)):
tok += f" num {int(q)}"
elif isinstance(q, str):
tok += f" num {q}"
tokens.append(tok)
continue
# match token
if "match" in el:
try:
rendered = _render_match(el["match"])
if rendered is None:
tokens.append(_compact_json_fragment(el))
else:
tokens.append(rendered)
except Exception:
tokens.append(_compact_json_fragment(el))
continue
# payload shorthand
if "payload" in el:
p = el["payload"]
if isinstance(p, dict):
prot = p.get("protocol") prot = p.get("protocol")
field = p.get("field") field = p.get("field")
if prot and field: if prot and field:
tokens.append(f"payload({prot}.{field})") tokens.append(f"payload({prot}.{field})")
continue continue
tokens.append(_compact_json_fragment(el)) tokens.append("payload")
continue elif "cmp" in part or "binary" in part:
tokens.append("cmp")
# tcp/udp nested objects e.g. {"tcp": {"dport": 22}} or {"tcp": {"flags":{"syn": True}}} elif "drop" in part:
if "tcp" in el or "udp" in el: tokens.append("drop")
proto = "tcp" if "tcp" in el else "udp" elif "accept" in part:
val = el.get(proto) tokens.append("accept")
if isinstance(val, dict): elif "counter" in part:
# dport/sport tokens.append("counter")
if "dport" in val: elif "tcp" in part or "udp" in part:
tokens.append(f"{proto} dport {_stringify_value(val['dport'])}") proto = "tcp" if "tcp" in part else "udp"
continue tokens.append(proto)
if "sport" in val: elif "queue" in part:
tokens.append(f"{proto} sport {_stringify_value(val['sport'])}") q = part["queue"]
continue token = "queue"
# flags if isinstance(q, dict):
if "flags" in val: num = q.get("num") or q.get("number") or q.get("queue_number") or q.get("from") or q.get("range")
flags = val.get("flags") if num is not None:
if isinstance(flags, (list, tuple)): token += f" num {num}"
tokens.append(f"{proto} flags {{{','.join(map(str, flags))}}}") if q.get("bypass"):
else: token += " bypass"
tokens.append(f"{proto} { _compact_json_fragment(val) }") elif isinstance(q, (int, float)):
continue token += f" num {int(q)}"
tokens.append(proto) elif isinstance(q, str):
continue token += f" num {q}"
tokens.append(token)
# cmp / binary / bitwise — attempt to render if shape known else:
if "cmp" in el or "binary" in el or "bitwise" in el: keys = "+".join(sorted(part.keys()))
# try to compose a readable fragment tokens.append(keys)
tokens.append(_compact_json_fragment(el)) else:
continue tokens.append(str(part))
return " ".join(tokens)
# named set membership e.g. {"in": {"left": ..., "right": ...}} or op in match return str(expr)
# fallback: include compact JSON fragment
tokens.append(_compact_json_fragment(el))
continue
# non-dict tokens (strings/numbers)
tokens.append(str(el))
return " ".join(tokens).strip()
# -------------------------
# Existing helpers (now use _serialize_expr)
# -------------------------
def rule_text_from_expr(expr: Any) -> str:
"""
Deterministic serializer to produce a compact UI-friendly string from expr list.
Uses the robust _serialize_expr and guarantees a string result (never None).
"""
try:
rendered = _serialize_expr(expr)
if rendered is None:
# as a last resort, dump compact JSON
return _compact_json_fragment(expr)
return rendered
except Exception:
return _compact_json_fragment(expr)
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 only when the expr is clearly unsupported for textual insertion.
"""
# For create_rule_json we should be slightly stricter: if serialization produces
# a totally opaque fragment (compact JSON), we prefer to return None to force
# the caller to use the raw textual endpoint.
try:
rendered = _serialize_expr(expr)
if rendered is None:
return None
# Heuristic: if our rendering is just a compact JSON object (meaning we couldn't parse it),
# consider it unsupported (return None).
if isinstance(rendered, str) and rendered.startswith("{") and rendered.endswith("}"):
# try to be conservative: maybe it's a queue/counter JSON we can accept; allow specific tokens
try:
parsed = json.loads(rendered)
# if parsed is dict with single known action, allow it:
if any(k in parsed for k in ("drop", "accept", "queue", "counter")):
return rendered
except Exception:
pass
return None
return rendered
except Exception:
return None
# -------------------------
# Build predictable ruleset (unchanged except it still uses rule_text_from_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 list ruleset parsed JSON into a deterministic, predictable JSON: Convert nft -j list ruleset parsed JSON into a deterministic, predictable JSON:
@@ -735,6 +524,179 @@ def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]:
return result 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")
tname = t.get("name")
if not fam or not tname:
continue
for ch in t.get("chains", []):
cname = ch.get("name")
if not cname:
continue
try:
chain_text = mgr.list_chain_text(fam, tname, cname) or ""
lines = [ln.rstrip() for ln in chain_text.splitlines() if ln.strip() != ""]
# build handle -> line map
handle_map: Dict[str, str] = {}
for ln in lines:
m = re.search(r"\bhandle\s+(\d+)\b", ln)
if m:
handle_map[m.group(1)] = ln.strip()
for rule in ch.get("rules", []):
replaced = False
h = rule.get("handle")
if h is not None:
key = str(h)
if key in handle_map:
rule["text"] = handle_map[key]
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",
fam,
tname,
cname,
e,
)
continue
# ---------- Routes ---------- # ---------- Routes ----------
@router.get("/rules", response_model=RulesetOut, summary="List ruleset") @router.get("/rules", response_model=RulesetOut, summary="List ruleset")
@@ -757,6 +719,15 @@ def list_rules():
return RulesetOut(ruleset=text.strip() if text is not None else None) return RulesetOut(ruleset=text.strip() if text is not None else None)
custom = build_predictable_ruleset(nft_json) 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)
ruleset_model = RulesetModel.parse_obj(custom) ruleset_model = RulesetModel.parse_obj(custom)
return RulesetOut(ruleset=ruleset_model) return RulesetOut(ruleset=ruleset_model)
except NftError as e: except NftError as e:
@@ -861,7 +832,6 @@ def create_rule_json(req: CreateRuleRequest):
# with detail that includes stderr and the executed cmd. # with detail that includes stderr and the executed cmd.
detail = f"nft command failed rc={rc}. stderr: {stderr!r}. cmd: {cmd}" detail = f"nft command failed rc={rc}. stderr: {stderr!r}. cmd: {cmd}"
logger.warning("create_rule_json failed: %s", detail) logger.warning("create_rule_json failed: %s", detail)
# Return an HTTPException with the detail (frontend can still inspect error.response.data if ExecResult was included)
raise HTTPException(status_code=400, detail=detail) raise HTTPException(status_code=400, detail=detail)
except NftError as e: except NftError as e: