diff --git a/backend/src/api/nft_manager.py b/backend/src/api/nft_manager.py index 561d4ac..a8cf40b 100644 --- a/backend/src/api/nft_manager.py +++ b/backend/src/api/nft_manager.py @@ -263,6 +263,45 @@ class RulesetOut(BaseModel): _handle_re = re.compile(r"\s+#\s*handle\s+\d+\s*$") +def parse_priority(val: Any) -> Optional[int]: + """ + Robustly parse a priority value returned in various nft JSON shapes. + Accepts: + - int -> returns unchanged + - numeric string -> parsed int + - dict -> tries common nested keys ('priority', 'prio') + Returns None if not parseable. + """ + if val is None: + return None + # if it's already an int + if isinstance(val, int): + return val + # numeric string + if isinstance(val, str): + s = val.strip() + if s.isdigit() or (s.startswith("-") and s[1:].isdigit()): + try: + return int(s) + except Exception: + return None + # sometimes nft uses "0" etc with whitespace + try: + return int(float(s)) + except Exception: + return None + # nested dicts sometimes appear + if isinstance(val, dict): + for key in ("priority", "prio"): + if key in val: + return parse_priority(val.get(key)) + # sometimes structure like {'hook': {'priority': 0}} - try to dive in + for v in val.values(): + p = parse_priority(v) + if p is not None: + return p + return None + def rule_text_from_expr(expr: Any) -> str: """ Deterministic serializer to produce a compact UI-friendly string from expr list. @@ -347,28 +386,40 @@ def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]: cname = ch.get("name") if fam and table_name and cname: tables.setdefault((fam, table_name), {"family": fam, "name": table_name, "chains": {}}) - # create chain with metadata fields (if present) - chain_obj = tables[(fam, table_name)]["chains"].setdefault( - cname, - { + chains_map = tables[(fam, table_name)]["chains"] + + # existing chain (maybe created earlier by rule processing) + existing = chains_map.get(cname) + # extract metadata robustly + ch_type = ch.get("type") + ch_hook = ch.get("hook") + ch_priority = parse_priority(ch.get("priority") if "priority" in ch else ch.get("prio") if "prio" in ch else ch.get("prio", None)) + # also attempt to parse nested shapes if present (some nft JSON variations) + if ch_priority is None: + ch_priority = parse_priority(ch.get("hook") if isinstance(ch.get("hook"), dict) else None) + + ch_policy = ch.get("policy") + + if existing is None: + chains_map[cname] = { "name": cname, - "type": ch.get("type"), - "hook": ch.get("hook"), - "priority": ch.get("priority"), - "policy": ch.get("policy"), + "type": ch_type, + "hook": ch_hook, + "priority": ch_priority, + "policy": ch_policy, "rules": [], - }, - ) - # if the chain already existed (due to earlier rules), ensure we add missing metadata if present - if isinstance(chain_obj, dict): - if chain_obj.get("type") is None and ch.get("type") is not None: - chain_obj["type"] = ch.get("type") - if chain_obj.get("hook") is None and ch.get("hook") is not None: - chain_obj["hook"] = ch.get("hook") - if chain_obj.get("priority") is None and ch.get("priority") is not None: - chain_obj["priority"] = ch.get("priority") - if chain_obj.get("policy") is None and ch.get("policy") is not None: - chain_obj["policy"] = ch.get("policy") + } + else: + # merge into placeholder (do not overwrite existing rules) + if isinstance(existing, dict): + if existing.get("type") is None and ch_type is not None: + existing["type"] = ch_type + if existing.get("hook") is None and ch_hook is not None: + existing["hook"] = ch_hook + if existing.get("priority") is None and ch_priority is not None: + existing["priority"] = ch_priority + if existing.get("policy") is None and ch_policy is not None: + existing["policy"] = ch_policy # rule records elif "rule" in rec: r = rec["rule"] @@ -379,11 +430,10 @@ def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]: expr = r.get("expr") if fam and table_name and chain_name: tables.setdefault((fam, table_name), {"family": fam, "name": table_name, "chains": {}}) - # ensure chain record exists, preserve placeholders for metadata if not yet set - tables[(fam, table_name)]["chains"].setdefault( - chain_name, - {"name": chain_name, "type": None, "hook": None, "priority": None, "policy": None, "rules": []}, - ) + chains_map = tables[(fam, table_name)]["chains"] + # ensure chain placeholder exists, with possible metadata defaults + chains_map.setdefault(chain_name, {"name": chain_name, "type": None, "hook": None, "priority": None, "policy": None, "rules": []}) + rule_obj: Dict[str, Any] = { "handle": handle, "expr": expr, @@ -394,7 +444,25 @@ def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]: rule_obj["position"] = r["position"] if "comment" in r: rule_obj["comment"] = r["comment"] - tables[(fam, table_name)]["chains"][chain_name]["rules"].append(rule_obj) + chains_map[chain_name]["rules"].append(rule_obj) + + # Attempt to salvage chain metadata from rule record if present + # some nft JSON may include 'chain' subfields inside rule record + # e.g. r.get('chain') might be an object - handle that defensively + if isinstance(r.get("chain"), dict): + csub = r.get("chain") + # try to parse nested priority + if chains_map[chain_name].get("priority") is None: + parsed_prio = parse_priority(csub.get("priority") if "priority" in csub else csub.get("prio")) + if parsed_prio is not None: + chains_map[chain_name]["priority"] = parsed_prio + # type/hook/policy from nested if present + if chains_map[chain_name].get("type") is None and csub.get("type") is not None: + chains_map[chain_name]["type"] = csub.get("type") + if chains_map[chain_name].get("hook") is None and csub.get("hook") is not None: + chains_map[chain_name]["hook"] = csub.get("hook") + if chains_map[chain_name].get("policy") is None and csub.get("policy") is not None: + chains_map[chain_name]["policy"] = csub.get("policy") # Convert map to sorted lists for deterministic order, and include chain metadata for (fam, tname) in sorted(tables.keys(), key=lambda k: (k[0], k[1])):