diff --git a/backend/src/api/nft_manager.py b/backend/src/api/nft_manager.py index 920a5ae..561d4ac 100644 --- a/backend/src/api/nft_manager.py +++ b/backend/src/api/nft_manager.py @@ -211,6 +211,10 @@ class RuleOut(BaseModel): class ChainOut(BaseModel): name: str = Field(..., description="Chain name", example="forward") + type: Optional[str] = Field(None, description="Chain type (e.g. filter, nat, route)") + hook: Optional[str] = Field(None, description="Hook (input/forward/output/ingress/egress) if present") + priority: Optional[int] = Field(None, description="Hook priority if present") + policy: Optional[str] = Field(None, description="Chain policy (accept/drop) if present") rules: List[RuleOut] = Field(..., description="Rules in this chain (ordered)") @@ -317,30 +321,55 @@ def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]: Convert nft -j list ruleset parsed JSON into a deterministic, predictable JSON: { "tables": [ - { "family": ..., "name": ..., "chains": [ { "name": ..., "rules": [ { handle, expr, text } ] } ] } + { "family": ..., "name": ..., "chains": [ { "name": ..., "type": ..., "hook": ..., "priority": ..., "policy": ..., "rules": [ { handle, expr, text } ] } ] } ] } """ result: Dict[str, Any] = {"tables": []} items = nft_json.get("nftables", []) if isinstance(nft_json, dict) else (nft_json or []) - # Build intermediate map: (family, table) -> {family, name, chains: {chain_name: {"name", "rules":[]}}} + # Build intermediate map: (family, table) -> {family, name, chains: {chain_name: {"name", "type", "hook", "priority", "policy", "rules":[]}}} tables: Dict[Tuple[str, str], Dict[str, Any]] = {} for rec in items: + # table records if "table" in rec: t = rec["table"] fam = t.get("family") name = t.get("name") if fam and name: tables.setdefault((fam, name), {"family": fam, "name": name, "chains": {}}) + # chain records: capture chain metadata elif "chain" in rec: ch = rec["chain"] - fam = ch.get("family") or (ch.get("table", {}) or {}).get("family") - table_name = ch.get("table") or (ch.get("table", {}) or {}).get("name") + # chain may include family/table or nested table reference + fam = ch.get("family") or (ch.get("table") or {}).get("family") + table_name = ch.get("table") or (ch.get("table") or {}).get("name") cname = ch.get("name") if fam and table_name and cname: tables.setdefault((fam, table_name), {"family": fam, "name": table_name, "chains": {}}) - tables[(fam, table_name)]["chains"].setdefault(cname, {"name": cname, "rules": []}) + # create chain with metadata fields (if present) + chain_obj = tables[(fam, table_name)]["chains"].setdefault( + cname, + { + "name": cname, + "type": ch.get("type"), + "hook": ch.get("hook"), + "priority": ch.get("priority"), + "policy": ch.get("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") + # rule records elif "rule" in rec: r = rec["rule"] fam = r.get("family") @@ -350,7 +379,11 @@ 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": {}}) - tables[(fam, table_name)]["chains"].setdefault(chain_name, {"name": chain_name, "rules": []}) + # 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": []}, + ) rule_obj: Dict[str, Any] = { "handle": handle, "expr": expr, @@ -363,12 +396,22 @@ def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]: rule_obj["comment"] = r["comment"] tables[(fam, table_name)]["chains"][chain_name]["rules"].append(rule_obj) - # Convert map to sorted lists for deterministic order + # 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])): tdata = tables[(fam, tname)] chains_list: List[Dict[str, Any]] = [] for cname in sorted(tdata["chains"].keys()): - chains_list.append({"name": cname, "rules": tdata["chains"][cname]["rules"]}) + chdata = tdata["chains"][cname] + chains_list.append( + { + "name": chdata.get("name"), + "type": chdata.get("type"), + "hook": chdata.get("hook"), + "priority": chdata.get("priority"), + "policy": chdata.get("policy"), + "rules": chdata.get("rules", []), + } + ) result["tables"].append({"family": fam, "name": tname, "chains": chains_list}) return result @@ -477,7 +520,7 @@ def list_rules(): Returns the ruleset in a stable, strongly-typed JSON shape derived from `nft -j list ruleset`. Structure: - { "ruleset": { "tables": [ { "family": ..., "name": ..., "chains": [ { "name": ..., "rules": [ { "handle", "expr", "text" } ] } ] } ] } } + { "ruleset": { "tables": [ { "family": ..., "name": ..., "chains": [ { "name": ..., "type": ..., "hook": ..., "priority": ..., "policy": ..., "rules": [ { "handle", "expr", "text" } ] } ] } ] } } Fallback: - If nft JSON is unavailable, falls back to returning the raw textual ruleset string. @@ -629,4 +672,4 @@ def exec_raw(req: RawCmdRequest): raise HTTPException(status_code=500, detail=str(e)) -app.include_router(router) +app.include_router(router) \ No newline at end of file