nft tables api pretty text
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:
@@ -325,73 +325,285 @@ def parse_priority(val: Any) -> Optional[int]:
|
||||
return None
|
||||
|
||||
|
||||
def rule_text_from_expr(expr: Any) -> str:
|
||||
# -------------------------
|
||||
# Expr serializer helpers
|
||||
# -------------------------
|
||||
def _compact_json_fragment(obj: Any) -> str:
|
||||
"""
|
||||
Deterministic serializer to produce a compact UI-friendly string from expr list.
|
||||
Covers common constructs; falls back to JSON dump for unknown constructs.
|
||||
(Used for display in GET /rules).
|
||||
Return a very compact JSON fragment for unknown tokens to include inline in text
|
||||
(keeps text deterministic and safe).
|
||||
"""
|
||||
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:
|
||||
return ""
|
||||
if isinstance(expr, list):
|
||||
tokens: List[str] = []
|
||||
for part in expr:
|
||||
if isinstance(part, dict):
|
||||
# common tokens
|
||||
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):
|
||||
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"]
|
||||
if isinstance(expr, str):
|
||||
return expr
|
||||
if not isinstance(expr, list):
|
||||
# unsupported top-level type -> pretty-print compact
|
||||
return _compact_json_fragment(expr)
|
||||
|
||||
tokens: List[str] = []
|
||||
for el in expr:
|
||||
# handle simple dict tokens
|
||||
if isinstance(el, dict):
|
||||
# direct known keywords
|
||||
if "drop" in el:
|
||||
tokens.append("drop")
|
||||
continue
|
||||
if "accept" in el:
|
||||
tokens.append("accept")
|
||||
continue
|
||||
if "counter" in el:
|
||||
tokens.append("counter")
|
||||
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")
|
||||
field = p.get("field")
|
||||
if prot and field:
|
||||
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:
|
||||
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)
|
||||
return str(expr)
|
||||
tokens.append(_compact_json_fragment(el))
|
||||
continue
|
||||
|
||||
# tcp/udp nested objects e.g. {"tcp": {"dport": 22}} or {"tcp": {"flags":{"syn": True}}}
|
||||
if "tcp" in el or "udp" in el:
|
||||
proto = "tcp" if "tcp" in el else "udp"
|
||||
val = el.get(proto)
|
||||
if isinstance(val, dict):
|
||||
# dport/sport
|
||||
if "dport" in val:
|
||||
tokens.append(f"{proto} dport {_stringify_value(val['dport'])}")
|
||||
continue
|
||||
if "sport" in val:
|
||||
tokens.append(f"{proto} sport {_stringify_value(val['sport'])}")
|
||||
continue
|
||||
# flags
|
||||
if "flags" in val:
|
||||
flags = val.get("flags")
|
||||
if isinstance(flags, (list, tuple)):
|
||||
tokens.append(f"{proto} flags {{{','.join(map(str, flags))}}}")
|
||||
else:
|
||||
tokens.append(f"{proto} { _compact_json_fragment(val) }")
|
||||
continue
|
||||
tokens.append(proto)
|
||||
continue
|
||||
|
||||
# cmp / binary / bitwise — attempt to render if shape known
|
||||
if "cmp" in el or "binary" in el or "bitwise" in el:
|
||||
# try to compose a readable fragment
|
||||
tokens.append(_compact_json_fragment(el))
|
||||
continue
|
||||
|
||||
# named set membership e.g. {"in": {"left": ..., "right": ...}} or op in match
|
||||
# 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]:
|
||||
"""
|
||||
Convert nft -j list ruleset parsed JSON into a deterministic, predictable JSON:
|
||||
@@ -523,118 +735,6 @@ def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]:
|
||||
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()
|
||||
|
||||
|
||||
# ---------- Routes ----------
|
||||
|
||||
@router.get("/rules", response_model=RulesetOut, summary="List ruleset")
|
||||
|
||||
@@ -350,7 +350,7 @@ export default function FirewallTables({ tables, error, refreshRules: refresh }:
|
||||
},
|
||||
{
|
||||
title: 'Rule',
|
||||
dataIndex: 'raw',
|
||||
dataIndex: 'frontendParsed',
|
||||
render: (v) => <Paragraph ellipsis={{ rows: 2, expandable: false }}>{v}</Paragraph>,
|
||||
},
|
||||
{
|
||||
@@ -377,7 +377,8 @@ export default function FirewallTables({ tables, error, refreshRules: refresh }:
|
||||
key: `${chain.name}:${idx}`,
|
||||
idx: idx + 1,
|
||||
handle: r.handle ?? null,
|
||||
raw: renderRuleFriendly(r),
|
||||
frontendParsed: renderRuleFriendly(r),
|
||||
backendtext: r.text,
|
||||
}));
|
||||
|
||||
return (
|
||||
|
||||
Reference in New Issue
Block a user