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

This commit is contained in:
2026-01-11 17:31:11 +01:00
parent 2c5523d3fe
commit 6347bcafdf

View File

@@ -333,30 +333,78 @@ def ensure_table_and_chain_exist(family: str, table: str, chain: str) -> None:
logger.error("Failed to create chain '%s' in table '%s': %s", chain, table, getattr(e, "detail", str(e))) logger.error("Failed to create chain '%s' in table '%s': %s", chain, table, getattr(e, "detail", str(e)))
raise raise
def nft_list_chain_text(
family: str,
table: str,
chain: str,
) -> Dict[int, str]:
"""
Return a mapping: handle -> textual nft rule line
extracted from `nft list chain <family> <table> <chain>`.
Example return:
{
12: 'meta iifname "eth0" accept comment "allow lan"',
13: 'ip saddr 10.0.0.0/24 drop'
}
"""
ensure_nft_available()
cmd = [NFT_BIN, "list", "chain", family, table, chain]
logger.debug("Listing chain in text mode: %s", " ".join(cmd))
try:
out = subprocess.check_output(cmd, stderr=subprocess.PIPE).decode()
except subprocess.CalledProcessError as e:
logger.error("Failed to list chain text: %s", e.stderr.decode())
raise HTTPException(status_code=500, detail=e.stderr.decode())
rules: Dict[int, str] = {}
for line in out.splitlines():
line = line.strip()
# Typical rule line contains: "handle <n>"
# Example:
# meta iifname "eth0" accept comment "foo" handle 7
if " handle " not in line:
continue
try:
rule_part, handle_part = line.rsplit(" handle ", 1)
handle = int(handle_part.strip())
rules[handle] = rule_part.strip()
except Exception:
logger.debug("Could not parse rule line: %s", line)
return rules
def nft_list_rules(table: str = "filter", chain: str = "forward") -> Dict[str, Any]: def nft_list_rules(table: str = "filter", chain: str = "forward") -> Dict[str, Any]:
""" """
Parse `nft --json list ruleset` and return a list of rules with metadata. Parse nft rules using JSON mode for structure AND text mode for readability.
Each returned rule includes:
- family, table, chain
- handle (if present)
- position (1-based within its chain)
- comment (if present, best-effort)
- verdict (best-effort string: 'accept'/'drop'/'reject' or None)
- verdict_details (raw nested object for reject or other complex verdicts)
- exprs (original expression list from nft JSON)
- nft_rule (the original rule dict from nft JSON)
This function is conservative and aims to give the UI enough info to Returned fields per rule:
display and edit rules precisely (by handle or position). - family, table, chain
- handle
- position (1-based)
- comment (best-effort)
- verdict (best-effort)
- verdict_details
- exprs (JSON expressions)
- nft_rule (TEXTUAL rule line, exactly as nft prints it)
""" """
ensure_nft_available() ensure_nft_available()
try: try:
out = subprocess.check_output([NFT_BIN, "--json", "list", "ruleset"], stderr=subprocess.PIPE) out = subprocess.check_output(
[NFT_BIN, "--json", "list", "ruleset"],
stderr=subprocess.PIPE,
)
parsed = json.loads(out) parsed = json.loads(out)
except subprocess.CalledProcessError as e: except subprocess.CalledProcessError as e:
logger.error("Failed to list ruleset: %s", e.stderr.decode()) logger.error("Failed to list ruleset (json): %s", e.stderr.decode())
raise HTTPException(status_code=500, detail=f"nft failed: {e.stderr.decode()}") raise HTTPException(status_code=500, detail=e.stderr.decode())
# 2) Text rules (human-readable)
text_rules = nft_list_chain_text(DEFAULT_FAMILY, table, chain)
results: List[Dict[str, Any]] = [] results: List[Dict[str, Any]] = []
counters: Dict[str, int] = {} counters: Dict[str, int] = {}
@@ -373,46 +421,47 @@ def nft_list_rules(table: str = "filter", chain: str = "forward") -> Dict[str, A
family = r.get("family") family = r.get("family")
table_name = r.get("table") table_name = r.get("table")
chain_name = r.get("chain") chain_name = r.get("chain")
if table_name != table or chain_name != chain:
continue
key = f"{family}:{table_name}:{chain_name}" key = f"{family}:{table_name}:{chain_name}"
counters.setdefault(key, 0) counters.setdefault(key, 0)
counters[key] += 1 counters[key] += 1
position = counters[key] position = counters[key]
handle = r.get("handle")
exprs = r.get("expr", []) # original expression list handle = r.get("handle")
exprs = r.get("expr", [])
comment: Optional[str] = None comment: Optional[str] = None
verdict: Optional[str] = None verdict: Optional[str] = None
verdict_details: Optional[Any] = None verdict_details: Optional[Any] = None
# scan expressions to extract comment and verdict/action
for expr in exprs: for expr in exprs:
if not isinstance(expr, dict): if not isinstance(expr, dict):
continue continue
if "comment" in expr: if "comment" in expr:
c = expr.get("comment") c = expr["comment"]
if isinstance(c, str): if isinstance(c, str):
comment = c comment = c
elif isinstance(c, dict): elif isinstance(c, dict):
comment = c.get("text") or c.get("str") or comment comment = c.get("text") or c.get("str")
# common verdict shapes: {"verdict": {"accept": null}} or {"drop": null}
if "verdict" in expr: if "verdict" in expr:
v = expr["verdict"] v = expr["verdict"]
if isinstance(v, dict): if isinstance(v, dict):
k = next(iter(v.keys()), None) verdict = next(iter(v.keys()), None)
verdict = k verdict_details = v.get(verdict)
verdict_details = v.get(k)
else: else:
verdict = str(v) verdict = str(v)
if "drop" in expr and verdict is None: if "drop" in expr and verdict is None:
verdict = "drop" verdict = "drop"
verdict_details = expr.get("drop")
if "accept" in expr and verdict is None: if "accept" in expr and verdict is None:
verdict = "accept" verdict = "accept"
verdict_details = expr.get("accept")
if "reject" in expr and verdict is None: if "reject" in expr and verdict is None:
verdict = "reject" verdict = "reject"
verdict_details = expr.get("reject")
results.append( results.append(
{ {
@@ -425,16 +474,15 @@ def nft_list_rules(table: str = "filter", chain: str = "forward") -> Dict[str, A
"verdict": verdict, "verdict": verdict,
"verdict_details": verdict_details, "verdict_details": verdict_details,
"exprs": exprs, "exprs": exprs,
"nft_rule": r, # 👇 THIS IS THE IMPORTANT CHANGE
"nft_rule": text_rules.get(handle),
} }
) )
# 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]
return {"rules": results} return {"rules": results}
def expr_to_nft_snippet(e: Expr) -> str: def expr_to_nft_snippet(e: Expr) -> str:
"""Convert a typed Expr into a short nft syntax snippet (used for preview/add).""" """Convert a typed Expr into a short nft syntax snippet (used for preview/add)."""
if isinstance(e, MetaExpr): if isinstance(e, MetaExpr):