test
All checks were successful
Build and Deploy MITM Webserver / build (push) Successful in 8s

This commit is contained in:
2026-02-28 22:33:14 +01:00
parent e844d79ac0
commit cbd7dacd24

View File

@@ -61,10 +61,34 @@ class NftManager:
Try to obtain nft -j list ruleset (JSON). Returns parsed JSON dict on success. Try to obtain nft -j list ruleset (JSON). Returns parsed JSON dict on success.
Raises NftError on failure or when output cannot be parsed as JSON. Raises NftError on failure or when output cannot be parsed as JSON.
""" """
res = self.nft.json_cmd("list ruleset") # prefer using json_cmd when available for a clean tuple (rc, out, err)
if res[0] != 0: try:
raise NftError(f"nft list ruleset failed: {res[2]}") res = self.nft.json_cmd("list ruleset")
return json.loads(res[1]) if not isinstance(res, (list, tuple)) or len(res) < 3:
raise Exception("unexpected json_cmd result shape")
if res[0] != 0:
raise NftError(f"nft list ruleset failed: {res[2]}")
return json.loads(res[1])
except AttributeError:
# fallback to textual cmd + json.loads if json_cmd is not available
cmd_variants = ["list ruleset -j", "list ruleset"]
last_err = None
for c in cmd_variants:
r = self.cmd(c)
if r["rc"] != 0:
last_err = r["stderr"]
continue
out = r["stdout"]
if not out:
last_err = "empty output"
continue
try:
parsed = json.loads(out)
return parsed
except json.JSONDecodeError as e:
last_err = f"json decode error: {e}"
continue
raise NftError(f"unable to get JSON ruleset: {last_err}")
def list_chain_text(self, family: str, table: str, chain: str) -> str: def list_chain_text(self, family: str, table: str, chain: str) -> str:
""" """
@@ -714,7 +738,9 @@ def list_rules():
except Exception as e: except Exception as e:
logger.debug("list_rules: populate_text_from_chain_text failed: %s", e) logger.debug("list_rules: populate_text_from_chain_text failed: %s", e)
ruleset_model = json.dumps(RulesetModel.parse_obj(custom)) # IMPORTANT: return the parsed RulesetModel (not a JSON string) so FastAPI/Pydantic
# will serialize the structure properly without double-escaping.
ruleset_model = RulesetModel.parse_obj(custom)
return RulesetOut(ruleset=ruleset_model) return RulesetOut(ruleset=ruleset_model)
except NftError as e: except NftError as e:
logger.exception("list_rules failed") logger.exception("list_rules failed")