improve nft parsing
All checks were successful
Build and Deploy MITM Webserver / build (push) Successful in 8s
All checks were successful
Build and Deploy MITM Webserver / build (push) Successful in 8s
This commit is contained in:
@@ -154,71 +154,124 @@ def ensure_table_chain(family: str, table: str, chain: str) -> None:
|
||||
|
||||
|
||||
# ---------------------- reconstruct rules from nft JSON (best-effort) ----------------------
|
||||
def _reconstruct_rule_from_exprs(exprs: List[Dict[str, Any]]) -> Dict[str, Any]:
|
||||
"""
|
||||
Heuristic mapping of nft expression JSON to our RuleModel fields.
|
||||
Covers common shapes: meta (iif/oif/length), cmp/payload (ip proto and ports), verdict (action).
|
||||
This is best-effort; complex expressions may not map perfectly.
|
||||
"""
|
||||
match: Dict[str, Any] = {}
|
||||
action: Dict[str, Any] = {}
|
||||
# Replace your previous _reconstruct_rule_from_exprs and list_rules_from_nft with these.
|
||||
|
||||
for e in exprs:
|
||||
if "meta" in e:
|
||||
m = e["meta"]
|
||||
key = m.get("key") or m.get("type")
|
||||
v = m.get("v") or m.get("s") or m.get("value")
|
||||
if key in ("iifname", "iif"):
|
||||
if v:
|
||||
match["iif"] = v
|
||||
elif key in ("oifname", "oif"):
|
||||
if v:
|
||||
match["oif"] = v
|
||||
elif key == "length":
|
||||
if v is not None:
|
||||
match["meta_length"] = v
|
||||
elif "payload" in e:
|
||||
# payload describes read of header bytes; often next 'cmp' compares it
|
||||
# we tag the payload on the expression to help cmp heuristics
|
||||
e["_payload_hint"] = e["payload"]
|
||||
elif "cmp" in e:
|
||||
cmp = e["cmp"]
|
||||
left = cmp.get("left")
|
||||
right = cmp.get("right")
|
||||
def _extract_immediate(node):
|
||||
def _extract_immediate_value(node):
|
||||
"""Return an integer/string value from various immediate/data shapes, or None."""
|
||||
if not node or not isinstance(node, dict):
|
||||
return None
|
||||
# common keys used by libnftables JSON
|
||||
for k in ("immediate", "value", "data", "s", "v"):
|
||||
if k in node:
|
||||
val = node[k]
|
||||
# hex strings like "0x00000050"
|
||||
if isinstance(val, str) and val.startswith("0x"):
|
||||
try:
|
||||
return int(val, 16)
|
||||
except Exception:
|
||||
return val
|
||||
return val
|
||||
# sometimes immediate sits under {"immediate": {"value": ...}}
|
||||
if "right" in node or "left" in node:
|
||||
# caller handles left/right structures
|
||||
return None
|
||||
imm_left = _extract_immediate(left)
|
||||
imm_right = _extract_immediate(right)
|
||||
# ip proto numeric likely in 1..255
|
||||
for imm in (imm_left, imm_right):
|
||||
if isinstance(imm, int) and 0 < imm < 256:
|
||||
match["ip_proto"] = imm
|
||||
break
|
||||
# port heuristics (1..65535)
|
||||
imm = imm_left if imm_left is not None else imm_right
|
||||
if isinstance(imm, int) and 0 < imm <= 65535:
|
||||
# best-effort assign to tcp_dport (common case)
|
||||
# more advanced heuristics could inspect payload hints
|
||||
match.setdefault("tcp_dport", imm)
|
||||
elif "verdict" in e:
|
||||
v = e["verdict"]
|
||||
# shapes vary: dict or string
|
||||
return None
|
||||
|
||||
|
||||
def _reconstruct_rule_from_exprs(exprs: List[Dict[str, Any]]) -> Dict[str, Any]:
|
||||
"""
|
||||
Improved best-effort mapping from nft expression JSON to our RuleModel fields.
|
||||
This function attempts to match many of the shapes produced by libnftables.
|
||||
"""
|
||||
match: Dict[str, Any] = {}
|
||||
action: Dict[str, Any] = {}
|
||||
|
||||
# Track last payload hint (if present) so that subsequent cmp can be interpreted
|
||||
last_payload_hint: Optional[Dict[str, Any]] = None
|
||||
|
||||
for idx, e in enumerate(exprs):
|
||||
# handle meta (iif/oif/length)
|
||||
if "meta" in e:
|
||||
m = e["meta"]
|
||||
key = m.get("key") or m.get("type") or m.get("field")
|
||||
v = m.get("v") or m.get("s") or m.get("value")
|
||||
# sometimes v is dict
|
||||
if isinstance(v, dict):
|
||||
t = v.get("type") or v.get("kind")
|
||||
v = _extract_immediate_value(v) or v.get("s") or v.get("v")
|
||||
if key in ("iifname", "iif", "in", "iifname?"):
|
||||
if v:
|
||||
match["iif"] = v
|
||||
elif key in ("oifname", "oif", "out"):
|
||||
if v:
|
||||
match["oif"] = v
|
||||
elif key in ("length", "len"):
|
||||
if v is not None:
|
||||
match["meta_length"] = v
|
||||
else:
|
||||
logger.debug("meta with unknown key: %s value=%s", key, v)
|
||||
|
||||
# store payload hints for later comparisons
|
||||
elif "payload" in e:
|
||||
last_payload_hint = e["payload"]
|
||||
# make it easier for cmp handling: include index
|
||||
last_payload_hint["_idx"] = idx
|
||||
|
||||
# cmp (compare) expressions: left/right might be payload / immediate structures
|
||||
elif "cmp" in e or "match" in e:
|
||||
cmp_obj = e.get("cmp") or e.get("match") or {}
|
||||
left = cmp_obj.get("left")
|
||||
right = cmp_obj.get("right")
|
||||
# extract immediate numeric if present
|
||||
imm_left = _extract_immediate_value(left)
|
||||
imm_right = _extract_immediate_value(right)
|
||||
imm = imm_left if imm_left is not None else imm_right
|
||||
|
||||
# If either immediate is a small int treat as ip_proto
|
||||
if isinstance(imm, int) and 0 < imm < 256:
|
||||
# prefer to store as number (frontend may show numeric)
|
||||
match["ip_proto"] = imm
|
||||
|
||||
# If immediate looks like port (1-65535), try to detect target (tcp/udp)
|
||||
if isinstance(imm, int) and 0 < imm <= 65535:
|
||||
assigned = False
|
||||
# heuristics: if left/right contains a payload hint referring to TCP/UDP or 'dport' strings:
|
||||
for side in (left, right):
|
||||
if isinstance(side, dict):
|
||||
# payload form used by libnftables can include 'protocol' or 'field'
|
||||
pl = side.get("payload") or side.get("left", {}).get("payload")
|
||||
if isinstance(pl, dict):
|
||||
protocol_hint = pl.get("protocol") or pl.get("proto") or pl.get("family")
|
||||
field_hint = pl.get("field") or pl.get("meta")
|
||||
sh = json.dumps(pl).lower()
|
||||
if "tcp" in sh or "sport" in sh or "dport" in sh or "th" in sh:
|
||||
match["tcp_dport"] = imm
|
||||
assigned = True
|
||||
break
|
||||
if "udp" in sh or "udph" in sh or "udp." in sh:
|
||||
match["udp_dport"] = imm
|
||||
assigned = True
|
||||
break
|
||||
if not assigned:
|
||||
# fallback: if ip_proto already indicates tcp(6) or udp(17), assign accordingly
|
||||
proto = match.get("ip_proto")
|
||||
if proto in (6, "tcp"):
|
||||
match["tcp_dport"] = imm
|
||||
elif proto in (17, "udp"):
|
||||
match["udp_dport"] = imm
|
||||
else:
|
||||
# if we can't know, default to tcp_dport as most common case
|
||||
match.setdefault("tcp_dport", imm)
|
||||
|
||||
# verdict / immediate verdict expressions -> action
|
||||
elif "verdict" in e or "immediate" in e or "return" in e:
|
||||
# verdict may be string or dict
|
||||
v = e.get("verdict") or e.get("return") or e.get("immediate")
|
||||
# normalize dict forms
|
||||
if isinstance(v, dict):
|
||||
t = v.get("type") or v.get("kind") or v.get("verdict")
|
||||
if t:
|
||||
action["type"] = t
|
||||
# redirect / queue extra fields vary
|
||||
# redirect/queue shaped differently across versions
|
||||
if "to" in v:
|
||||
action["type"] = "redirect"
|
||||
action["redirect_port"] = v.get("to")
|
||||
@@ -226,49 +279,78 @@ def _reconstruct_rule_from_exprs(exprs: List[Dict[str, Any]]) -> Dict[str, Any]:
|
||||
action["type"] = "queue"
|
||||
action["queue_num"] = v.get("queue")
|
||||
elif isinstance(v, str):
|
||||
# strings like "accept" or "drop"
|
||||
action["type"] = v
|
||||
# other expression types intentionally ignored for stateless reconstruction
|
||||
else:
|
||||
# sometimes verdict is expressed as nested dict under 'verdict': {'kind':'accept'}
|
||||
if isinstance(e.get("verdict"), dict):
|
||||
vv = e["verdict"]
|
||||
action["type"] = vv.get("kind") or vv.get("type")
|
||||
|
||||
# old-style 'match' entries with left/right payload/immediate
|
||||
elif "match" in e:
|
||||
m = e["match"]
|
||||
left = m.get("left")
|
||||
right = m.get("right")
|
||||
imm_left = _extract_immediate_value(left)
|
||||
imm_right = _extract_immediate_value(right)
|
||||
if isinstance(imm_left, int) and 0 < imm_left < 256:
|
||||
match["ip_proto"] = imm_left
|
||||
if isinstance(imm_right, int) and 0 < imm_right < 256:
|
||||
match["ip_proto"] = imm_right
|
||||
|
||||
else:
|
||||
# unknown expression type: log (DEBUG) for later tuning
|
||||
logger.debug("unhandled nft expr type: %s", list(e.keys()))
|
||||
|
||||
# default action to accept if kernel default or none found
|
||||
if "type" not in action:
|
||||
# if kernel default, we assume accept
|
||||
action["type"] = "accept"
|
||||
|
||||
return {"match": match, "action": action}
|
||||
|
||||
|
||||
def list_rules_from_nft(family: str, table: str) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Reconstruct an ordered list of rule dicts from `nft list table` JSON via the binding.
|
||||
Returns list of dicts shaped like RuleModel (without id).
|
||||
Reconstruct rules using libnftables JSON output with more robust parsing.
|
||||
"""
|
||||
try:
|
||||
out = nft_run_or_raise(f"list table {family} {table}")
|
||||
except RuntimeError:
|
||||
logger.debug("no table %s.%s found when listing rules", family, table)
|
||||
return []
|
||||
|
||||
# attempt to parse JSON output (binding can return textual JSON)
|
||||
try:
|
||||
data = json.loads(out)
|
||||
except Exception:
|
||||
# force JSON mode in binding if previous parsing failed
|
||||
# ensure binding present and request JSON output explicitly
|
||||
_ensure_binding_available()
|
||||
nft = Nftables()
|
||||
nft.set_json_output(True)
|
||||
rc, out_json, err = nft.cmd(f"list table {family} {table}")
|
||||
rc, out, err = nft.cmd(f"list table {family} {table}")
|
||||
if rc != 0:
|
||||
logger.debug("nft JSON listing failed: %s", err)
|
||||
logger.debug("nft list table returned rc=%s err=%s", rc, err)
|
||||
return []
|
||||
|
||||
try:
|
||||
data = json.loads(out)
|
||||
except Exception as e:
|
||||
logger.error("failed to parse nft JSON output: %s", e)
|
||||
return []
|
||||
data = json.loads(out_json)
|
||||
|
||||
results: List[Dict[str, Any]] = []
|
||||
for item in data.get("nftables", []):
|
||||
if "rule" not in item:
|
||||
continue
|
||||
rule_obj = item["rule"]
|
||||
exprs = rule_obj.get("expr", []) or rule_obj.get("expressions", []) or []
|
||||
# different versions might use 'expr', 'expressions' or 'expr'
|
||||
exprs = rule_obj.get("expr") or rule_obj.get("exprs") or rule_obj.get("expressions") or []
|
||||
# if exprs is not a list but a dict (some shaped outputs) normalize
|
||||
if isinstance(exprs, dict):
|
||||
# sometimes expressions are nested inside a single dict; try to find array keys
|
||||
for k in ("expr", "expressions", "exprs"):
|
||||
v = exprs.get(k)
|
||||
if isinstance(v, list):
|
||||
exprs = v
|
||||
break
|
||||
else:
|
||||
# give up and wrap
|
||||
exprs = [exprs]
|
||||
|
||||
recon = _reconstruct_rule_from_exprs(exprs)
|
||||
# the chain name may be in the rule metadata
|
||||
chain_name = rule_obj.get("chain") or rule_obj.get("chain_name") or DEFAULT_CHAIN
|
||||
chain_name = rule_obj.get("chain") or rule_obj.get("chain_name") or rule_obj.get("table") or DEFAULT_CHAIN
|
||||
# make output match your RuleModel-ish structure (no id)
|
||||
rule_dict = {
|
||||
"family": family,
|
||||
"table": table,
|
||||
@@ -277,10 +359,12 @@ def list_rules_from_nft(family: str, table: str) -> List[Dict[str, Any]]:
|
||||
"action": recon.get("action", {}),
|
||||
}
|
||||
results.append(rule_dict)
|
||||
|
||||
logger.debug("reconstructed %d rules from %s.%s", len(results), family, table)
|
||||
return results
|
||||
|
||||
|
||||
|
||||
# ---------------------- API endpoints ----------------------
|
||||
@router.get("/rules")
|
||||
def get_rules(family: Optional[str] = DEFAULT_FAMILY, table: Optional[str] = DEFAULT_TABLE, chain: Optional[str] = DEFAULT_CHAIN):
|
||||
|
||||
Reference in New Issue
Block a user