add chain properties to api
All checks were successful
Build and Deploy MITM Webserver / build (push) Successful in 11s

This commit is contained in:
2026-02-28 13:49:35 +01:00
parent 6636f72f11
commit 8b1160dff5

View File

@@ -211,6 +211,10 @@ class RuleOut(BaseModel):
class ChainOut(BaseModel): class ChainOut(BaseModel):
name: str = Field(..., description="Chain name", example="forward") 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)") 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: Convert nft -j list ruleset parsed JSON into a deterministic, predictable JSON:
{ {
"tables": [ "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": []} result: Dict[str, Any] = {"tables": []}
items = nft_json.get("nftables", []) if isinstance(nft_json, dict) else (nft_json or []) 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]] = {} tables: Dict[Tuple[str, str], Dict[str, Any]] = {}
for rec in items: for rec in items:
# table records
if "table" in rec: if "table" in rec:
t = rec["table"] t = rec["table"]
fam = t.get("family") fam = t.get("family")
name = t.get("name") name = t.get("name")
if fam and name: if fam and name:
tables.setdefault((fam, name), {"family": fam, "name": name, "chains": {}}) tables.setdefault((fam, name), {"family": fam, "name": name, "chains": {}})
# chain records: capture chain metadata
elif "chain" in rec: elif "chain" in rec:
ch = rec["chain"] ch = rec["chain"]
fam = ch.get("family") or (ch.get("table", {}) or {}).get("family") # chain may include family/table or nested table reference
table_name = ch.get("table") or (ch.get("table", {}) or {}).get("name") 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") cname = ch.get("name")
if fam and table_name and cname: if fam and table_name and cname:
tables.setdefault((fam, table_name), {"family": fam, "name": table_name, "chains": {}}) 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: elif "rule" in rec:
r = rec["rule"] r = rec["rule"]
fam = r.get("family") fam = r.get("family")
@@ -350,7 +379,11 @@ def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]:
expr = r.get("expr") expr = r.get("expr")
if fam and table_name and chain_name: if fam and table_name and chain_name:
tables.setdefault((fam, table_name), {"family": fam, "name": table_name, "chains": {}}) 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] = { rule_obj: Dict[str, Any] = {
"handle": handle, "handle": handle,
"expr": expr, "expr": expr,
@@ -363,12 +396,22 @@ def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]:
rule_obj["comment"] = r["comment"] rule_obj["comment"] = r["comment"]
tables[(fam, table_name)]["chains"][chain_name]["rules"].append(rule_obj) 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])): for (fam, tname) in sorted(tables.keys(), key=lambda k: (k[0], k[1])):
tdata = tables[(fam, tname)] tdata = tables[(fam, tname)]
chains_list: List[Dict[str, Any]] = [] chains_list: List[Dict[str, Any]] = []
for cname in sorted(tdata["chains"].keys()): 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}) result["tables"].append({"family": fam, "name": tname, "chains": chains_list})
return result 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`. Returns the ruleset in a stable, strongly-typed JSON shape derived from `nft -j list ruleset`.
Structure: 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: Fallback:
- If nft JSON is unavailable, falls back to returning the raw textual ruleset string. - If nft JSON is unavailable, falls back to returning the raw textual ruleset string.