From 57de326910f450279271529649cf22c9bc3be448 Mon Sep 17 00:00:00 2001 From: malmert Date: Sun, 11 Jan 2026 17:12:19 +0100 Subject: [PATCH] test123 --- backend/src/api/nft_api.py | 134 +++++++++++++++++++------------------ 1 file changed, 70 insertions(+), 64 deletions(-) diff --git a/backend/src/api/nft_api.py b/backend/src/api/nft_api.py index 95d0ac1..526558e 100644 --- a/backend/src/api/nft_api.py +++ b/backend/src/api/nft_api.py @@ -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') - 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', []) - 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'] - if isinstance(v, dict): - verdict = list(v.keys())[0] - else: - verdict = str(v) - # some JSON formats have 'match' or 'payload' etc. look for 'type' fields - if 'reject' in expr: - verdict = '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, - }) - # Optionally filter by table/chain if requested + + 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") + + exprs = r.get("expr", []) # original expression list from nft JSON + comment = None + verdict = None + 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): + # take the first key as the action + k = next(iter(v.keys()), None) + verdict = k + verdict_details = v.get(k) + else: + verdict = str(v) + # 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, # 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 + }) + + # 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):