This commit is contained in:
@@ -202,19 +202,6 @@ def ensure_nft_available():
|
||||
|
||||
|
||||
def nft_list_rules(table: str = "filter", chain: str = "forward") -> Dict[str, Any]:
|
||||
"""Return nft rules parsed from `nft --json list ruleset`.
|
||||
|
||||
For each rule we return a dict with at least:
|
||||
- family, table, chain
|
||||
- handle (if present)
|
||||
- position (1-based index within chain)
|
||||
- comment (if present in expr)
|
||||
- verdict (best-effort string like 'accept'/'drop'/'reject')
|
||||
- raw_exprs: the original expr list from nft JSON
|
||||
|
||||
This makes it easier for the frontend to show position and edit
|
||||
or replace a specific rule by handle/position.
|
||||
"""
|
||||
ensure_nft_available()
|
||||
cmd = [NFT_BIN, "--json", "list", "ruleset"]
|
||||
try:
|
||||
@@ -224,69 +211,88 @@ def nft_list_rules(table: str = "filter", chain: str = "forward") -> Dict[str, A
|
||||
raise HTTPException(status_code=500, detail=f"nft failed: {e.stderr.decode()}")
|
||||
|
||||
results: List[Dict[str, Any]] = []
|
||||
# counters per chain to compute position
|
||||
counters: Dict[str, int] = {}
|
||||
|
||||
# The JSON usually contains a top-level dict with key 'nftables' -> list
|
||||
items = parsed.get("nftables") if isinstance(parsed, dict) else parsed
|
||||
if not isinstance(items, list):
|
||||
items = []
|
||||
|
||||
for item in items:
|
||||
# Keep a simple mapping of current table/chain context
|
||||
if 'table' in item:
|
||||
tbl = item['table']
|
||||
# nothing to do for context; continue
|
||||
# skip table/chain metadata entries (we only need rule entries)
|
||||
if "rule" not in item:
|
||||
continue
|
||||
if 'chain' in item:
|
||||
ch = item['chain']
|
||||
# continue; chain metadata present
|
||||
continue
|
||||
if 'rule' in item:
|
||||
r = item['rule']
|
||||
family = r.get('family')
|
||||
table_name = r.get('table')
|
||||
chain_name = r.get('chain')
|
||||
|
||||
r = item["rule"]
|
||||
family = r.get("family")
|
||||
table_name = r.get("table")
|
||||
chain_name = r.get("chain")
|
||||
key = f"{family}:{table_name}:{chain_name}"
|
||||
counters.setdefault(key, 0)
|
||||
counters[key] += 1
|
||||
position = counters[key]
|
||||
handle = r.get('handle')
|
||||
# extract comment and verdict best-effort
|
||||
raw_exprs = r.get('expr', [])
|
||||
handle = r.get("handle")
|
||||
|
||||
exprs = r.get("expr", []) # original expression list from nft JSON
|
||||
comment = None
|
||||
verdict = None
|
||||
for expr in raw_exprs:
|
||||
if isinstance(expr, dict):
|
||||
if 'comment' in expr:
|
||||
comment = expr.get('comment')
|
||||
if 'verdict' in expr:
|
||||
# verdict might be dict like {'verdict': {'kind': 'accept'}}
|
||||
v = expr['verdict']
|
||||
verdict_details = None
|
||||
|
||||
# scan expressions to extract comment and verdict/action
|
||||
for expr in exprs:
|
||||
if not isinstance(expr, dict):
|
||||
continue
|
||||
# comment can appear as {"comment":"text"} or {"comment": {"text": "..."}} depending on nft json variant
|
||||
if "comment" in expr:
|
||||
# handle both simple and nested forms
|
||||
c = expr.get("comment")
|
||||
if isinstance(c, str):
|
||||
comment = c
|
||||
elif isinstance(c, dict):
|
||||
# some representations: {"comment": {"text": "..."}} or {"comment": {"str": "..."}}
|
||||
comment = c.get("text") or c.get("str") or comment
|
||||
|
||||
# verdict forms
|
||||
if "verdict" in expr:
|
||||
v = expr["verdict"]
|
||||
# v often looks like {"accept": None} or {"drop": None} or {"reject": {...}}
|
||||
if isinstance(v, dict):
|
||||
verdict = list(v.keys())[0]
|
||||
# take the first key as the action
|
||||
k = next(iter(v.keys()), None)
|
||||
verdict = k
|
||||
verdict_details = v.get(k)
|
||||
else:
|
||||
verdict = str(v)
|
||||
# some JSON formats have 'match' or 'payload' etc. look for 'type' fields
|
||||
if 'reject' in expr:
|
||||
verdict = 'reject'
|
||||
# older/alternate forms
|
||||
if "drop" in expr and verdict is None:
|
||||
verdict = "drop"
|
||||
verdict_details = expr.get("drop")
|
||||
if "accept" in expr and verdict is None:
|
||||
verdict = "accept"
|
||||
verdict_details = expr.get("accept")
|
||||
if "reject" in expr and verdict is None:
|
||||
verdict = "reject"
|
||||
verdict_details = expr.get("reject")
|
||||
|
||||
results.append({
|
||||
'family': family,
|
||||
'table': table_name,
|
||||
'chain': chain_name,
|
||||
'handle': handle,
|
||||
'position': position,
|
||||
'comment': comment,
|
||||
'verdict': verdict,
|
||||
'raw_exprs': raw_exprs,
|
||||
'raw_rule': r,
|
||||
"family": family,
|
||||
"table": table_name,
|
||||
"chain": chain_name,
|
||||
"handle": handle,
|
||||
"position": position,
|
||||
"comment": comment,
|
||||
"verdict": verdict, # e.g. "accept", "drop", "reject", or None
|
||||
"verdict_details": verdict_details, # raw details for rejects/other actions
|
||||
"exprs": exprs, # original expression list (renamed, clearer)
|
||||
"nft_rule": r, # the original nft JSON dict for this rule
|
||||
})
|
||||
# Optionally filter by table/chain if requested
|
||||
|
||||
# filter by requested table/chain if provided
|
||||
if table or chain:
|
||||
results = [x for x in results if x['table'] == table and x['chain'] == chain]
|
||||
results = [x for x in results if x["table"] == table and x["chain"] == chain]
|
||||
return {"rules": results}
|
||||
|
||||
|
||||
|
||||
def expr_to_nft_snippet(e: Expr) -> str:
|
||||
# runtime dispatch via model type
|
||||
if isinstance(e, MetaExpr):
|
||||
|
||||
Reference in New Issue
Block a user