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-03-01 00:04:58 +01:00
parent 2033d2fce2
commit 87bad13729
1 changed files with 2 additions and 200 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -240,70 +240,6 @@ def parse_priority(val: Any) -> Optional[int]:
return None return None
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:
if isinstance(part, dict):
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, int)):
p = left["payload"]
prot = p.get("protocol")
field = p.get("field")
if prot and field:
tokens.append(f"{prot} {field} {right}")
continue
tokens.append("match")
elif "payload" in part:
p = part["payload"]
prot = p.get("protocol")
field = p.get("field")
if prot and field:
tokens.append(f"payload({prot}.{field})")
continue
tokens.append("payload")
elif "drop" in part:
tokens.append("drop")
elif "accept" in part:
tokens.append("accept")
elif "counter" in part:
tokens.append("counter")
elif "tcp" in part or "udp" in part:
proto = "tcp" if "tcp" in part else "udp"
tokens.append(proto)
elif "queue" in part:
q = part["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}"
tokens.append(token)
else:
keys = "+".join(sorted(part.keys()))
tokens.append(keys)
else:
tokens.append(str(part))
return " ".join(tokens)
# 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]: 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:
@@ -366,7 +302,7 @@ def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]:
chains_map.setdefault(chain_name, {"name": chain_name, "type": None, "hook": None, "priority": None, "policy": None, "rules": []}) chains_map.setdefault(chain_name, {"name": chain_name, "type": None, "hook": None, "priority": None, "policy": None, "rules": []})
# do NOT change expr shape here; keep it exactly as NFT JSON provided # 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)} rule_obj: Dict[str, Any] = {"handle": handle, "expr": expr, "text": ""}
if "position" in r: if "position" in r:
rule_obj["position"] = r["position"] rule_obj["position"] = r["position"]
if "comment" in r: if "comment" in r:
@@ -511,7 +447,7 @@ def populate_text_from_ruleset_text(custom: Dict[str, Any], nft_text: str) -> No
# final substring probe (safe) # final substring probe (safe)
if mapped is None: if mapped is None:
probe = rule.get("text") or rule_text_from_expr(rule.get("expr")) probe = rule.get("text")
if probe: if probe:
for ln in ordered_lines: for ln in ordered_lines:
if probe in ln: if probe in ln:
@@ -526,56 +462,6 @@ def populate_text_from_ruleset_text(custom: Dict[str, Any], nft_text: str) -> No
rule["text"] = mapped # should be str already rule["text"] = mapped # should be str already
def populate_text_from_chain_text(custom: Dict[str, Any]) -> None:
"""
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:
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() != ""]
handle_map: Dict[int, str] = {}
for ln in lines:
m = re.search(r"\bhandle\s+(\d+)\b", ln)
if m:
try:
handle_map[int(m.group(1))] = ln.strip()
except Exception:
continue
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:
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:
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 for %s %s %s: %s", fam, tname, cname, e)
continue
# ---------- Normalization helper (lightweight and safe) ---------- # ---------- Normalization helper (lightweight and safe) ----------
def normalize_custom_for_model(custom: Dict[str, Any]) -> None: def normalize_custom_for_model(custom: Dict[str, Any]) -> None:
""" """
@@ -595,9 +481,6 @@ def normalize_custom_for_model(custom: Dict[str, Any]) -> None:
# text: ensure string # text: ensure string
if "text" not in r or r.get("text") is None: 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"] = "" r["text"] = ""
else: else:
if not isinstance(r["text"], str): if not isinstance(r["text"], str):
@@ -654,12 +537,6 @@ def list_rules():
except Exception as e: except Exception as e:
logger.debug("populate_text_from_ruleset_text failed: %s", e) logger.debug("populate_text_from_ruleset_text failed: %s", e)
# Per-chain fallback using `nft list chain ...`
try:
populate_text_from_chain_text(custom)
except Exception as e:
logger.debug("populate_text_from_chain_text failed: %s", e)
# Normalize minimally for model validation # Normalize minimally for model validation
normalize_custom_for_model(custom) normalize_custom_for_model(custom)
@@ -687,81 +564,6 @@ def list_rules():
raise HTTPException(status_code=500, detail=str(e)) raise HTTPException(status_code=500, detail=str(e))
@router.post(
"/rules",
response_model=ExecResult,
status_code=status.HTTP_201_CREATED,
summary="Create rule (JSON, expr required; returns ExecResult with rc/stdout/stderr)",
)
def create_rule_json(req: CreateRuleRequest):
try:
family = req.family
table = req.table
chain = req.chain
if req.expr is None:
raise NftError("field 'expr' is required for JSON rule creation")
rendered = rule_text_from_expr(req.expr)
if rendered is None or rendered == "":
raise NftError(
"cannot render provided 'expr' to textual nft syntax. Use POST /firewall/raw"
)
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)
raw_rc = res.get("rc")
stdout = res.get("stdout") or ""
stderr = res.get("stderr") or ""
try:
rc = int(raw_rc)
except Exception:
rc = -1
logger.info("nft cmd rc=%s stdout=%r stderr=%r cmd=%s", rc, stdout, stderr, cmd)
exec_res = ExecResult(rc=rc, stdout=stdout or None, stderr=stderr or None)
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() == "":
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: rule detected 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("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))
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") @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: try: