marcus/mitm-webserver
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

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

Browse Source
This commit is contained in:
malmert
2026-02-28 23:37:09 +01:00
parent 5a7bce647b
commit 1ad3eecee0
1 changed files with 37 additions and 322 deletions
Download Patch File Download Diff File Expand all files Collapse all files

357
backend/src/api/nft_manager.py
View File

@@ -20,17 +20,8 @@ class NftError(RuntimeError):
# ---------- NftManager (textual-only) ----------
class NftManager:
"""
Thin wrapper around python-nftables exposing:
- cmd execution (textual nft commands via Nftables.cmd())
- convenience list_rules_text / list_rules_json / list_chain_text
We prefer JSON globally, but for per-chain textual listing we temporarily disable JSON
so the output matches `nft list chain ...` textual rule lines.
"""
def __init__(self) -> None:
self.nft = Nftables()
# Try to prefer JSON for general listing; we'll toggle off for chain-list calls.
try:
self.nft.set_json_output(True)
self.nft.set_handle_output(True)
@@ -38,21 +29,12 @@ class NftManager:
logger.debug("set_json_output not available or ignored")
def cmd(self, text_cmd: str) -> Dict[str, Optional[str]]:
"""
Execute a textual nft command via Nftables.cmd().
Returns dict { "rc": rc, "stdout": out_str, "stderr": err_str }.
"""
rc, out, err = self.nft.cmd(text_cmd)
if rc != 0:
logger.warning("nft cmd rc=%s stderr=%s cmd=%s", rc, err, text_cmd)
return {"rc": rc, "stdout": out, "stderr": err}
def list_rules_json(self) -> Dict[str, Any]:
"""
Obtain nft -j list ruleset (JSON). Returns parsed JSON dict on success.
Raises NftError on failure or when output cannot be parsed as JSON.
"""
res = self.cmd("list ruleset")
if res["rc"] != 0:
raise NftError(f"nft list ruleset failed: {res['stderr']}")
@@ -65,11 +47,6 @@ class NftManager:
raise NftError(f"json decode error: {e}")
def list_rules_text(self) -> str:
"""
Obtain nft list ruleset in textual form. This is a fallback if JSON parsing fails.
Returns raw textual output on success, raises NftError on failure.
"""
# best-effort toggle JSON off for textual output
json_toggled = False
try:
if hasattr(self.nft, "set_json_output"):
@@ -91,28 +68,18 @@ class NftManager:
return res["stdout"] or ""
def list_chain_text(self, family: str, table: str, chain: str) -> str:
"""
Return textual output of `nft list chain <family> <table> <chain>`.
This tries to temporarily disable JSON output so the wrapper returns the textual
representation used by `nft list ruleset`. If disabling JSON is not possible,
we attempt to parse returned JSON (as a last resort), but the preferred path is
to get textual output.
"""
cmd = f"list chain {family} {table} {chain}"
# Attempt to temporarily disable JSON output on the wrapper (best-effort).
json_toggled = False
res = {"rc": -1, "stdout": "", "stderr": "unknown"}
try:
if hasattr(self.nft, "set_json_output"):
try:
# Turn off JSON output to force textual output for this call.
self.nft.set_json_output(False)
json_toggled = True
except Exception:
logger.debug("could not toggle set_json_output(False); will try command anyway")
res = self.cmd(cmd)
finally:
# Restore JSON output preference if we toggled it.
if json_toggled and hasattr(self.nft, "set_json_output"):
try:
self.nft.set_json_output(True)
@@ -123,15 +90,11 @@ class NftManager:
raise NftError(f"nft {cmd} failed: {res['stderr']}")
out = res["stdout"] or ""
# If the output looks like JSON (starts with '{' or '['), try a safe fallback:
s = out.strip()
if s.startswith("{") or s.startswith("["):
# Best-effort: parse JSON and attempt to extract rule textual forms if present.
try:
parsed = json.loads(s)
# parsed may be the whole ruleset (nftables list) or a list; find any "rule" objects
rule_lines: List[str] = []
# parsed might be dict with "nftables" or a list of records
records = parsed.get("nftables") if isinstance(parsed, dict) else parsed
if not isinstance(records, list):
records = []
@@ -166,7 +129,6 @@ class NftManager:
elif isinstance(part, dict) and "counter" in part:
tokens.append("counter")
elif isinstance(part, dict) and "queue" in part:
# handle fallback queue textualization
q = part["queue"]
if isinstance(q, dict):
num = q.get("num") or q.get("number") or q.get("queue_number") or q.get("from") or q.get("range")
@@ -177,13 +139,11 @@ class NftManager:
tok += " bypass"
tokens.append(tok)
else:
# numeric or string value
if isinstance(q, (int, float)):
tokens.append(f"queue num {int(q)}")
else:
tokens.append(f"queue num {q}")
else:
# fallback for unknown dict token
if isinstance(part, dict):
tokens.append("+".join(part.keys()))
else:
@@ -199,13 +159,8 @@ class NftManager:
return out
def delete_rule_by_handle_text(self, family: str, table: str, chain: str, handle: int) -> None:
"""
Delete a rule by handle using textual nft command:
delete rule <family> <table> <chain> handle <handle>
"""
if not isinstance(handle, int) or handle <= 0:
raise ValueError("handle must be a positive integer")
# construct textual command
cmd = f"delete rule {family} {table} {chain} handle {handle}"
res = self.cmd(cmd)
if res["rc"] != 0:
@@ -218,137 +173,93 @@ router = APIRouter(prefix="/firewall", tags=["firewall"])
mgr = NftManager()
# ---------- Request/Response models (strongly typed) ----------
# ---------- Models ----------
class RawCmdRequest(BaseModel):
cmd: str = Field(..., description="Textual nft command to execute", example="add rule inet filter input ip saddr 10.0.0.0/8 drop")
cmd: str = Field(...)
class ExecResult(BaseModel):
rc: int = Field(..., description="Return code from nft execution", example=0)
stdout: Optional[str] = Field(None, description="Standard output from nft", example="")
stderr: Optional[str] = Field(None, description="Standard error from nft", example="")
class Config:
schema_extra = {"example": {"rc": 0, "stdout": "ok", "stderr": ""}}
rc: int
stdout: Optional[str]
stderr: Optional[str]
# --- Strong models returned to frontend ---
class RuleOut(BaseModel):
handle: Optional[int] = Field(None, description="The rule handle (unique per rule), if available", example=3)
expr: Any = Field(..., description="Machine-readable nft expression (original nft JSON expr).")
text: str = Field(..., description="Deterministic short display string derived from expr", example="ip protocol icmp drop")
position: Optional[Any] = Field(None, description="Optional position metadata from nft if present")
comment: Optional[str] = Field(None, description="Optional comment attached to the rule")
handle: Optional[int]
expr: Any
text: str
position: Optional[Any]
comment: Optional[str]
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)")
name: str
type: Optional[str]
hook: Optional[str]
priority: Optional[int]
policy: Optional[str]
rules: List[RuleOut]
class TableOut(BaseModel):
family: str = Field(..., description="Table family (inet/bridge/ipv4/...)")
name: str = Field(..., description="Table name", example="filter")
chains: List[ChainOut] = Field(..., description="Chains in this table")
family: str
name: str
chains: List[ChainOut]
class RulesetModel(BaseModel):
tables: List[TableOut] = Field(..., description="Top-level tables list")
tables: List[TableOut]
# ---------- New: CreateRuleRequest (JSON, expr required) ----------
class CreateRuleRequest(BaseModel):
family: str = Field(..., description="Table family (e.g. inet, bridge, ip, ip6)", example="bridge")
table: str = Field(..., description="Table name (e.g. filter)", example="filter")
chain: str = Field(..., description="Chain name (e.g. forward)", example="forward")
expr: Any = Field(..., description="nft JSON expression (machine-readable). This field is required for JSON rule creation.")
position: Optional[int] = Field(None, description="Optional insertion position (zero-based). If provided endpoint will insert at that position.")
comment: Optional[str] = Field(None, description="Optional comment")
class Config:
schema_extra = {
"example": {
"family": "bridge",
"table": "filter",
"chain": "forward",
"expr": [{"match": {"left": {"payload": {"protocol": "ip", "field": "protocol"}}, "op": "==", "right": "icmp"}}, {"drop": None}],
"position": 0,
}
}
family: str
table: str
chain: str
expr: Any
position: Optional[int]
comment: Optional[str]
# ruleset may be typed RulesetModel or raw textual string (fallback)
RulesetValue = Optional[Union[RulesetModel, str]]
class RulesetOut(BaseModel):
ruleset: RulesetValue = Field(
None,
description="Parsed, strongly-typed ruleset (RulesetModel) or raw textual ruleset string if JSON is unavailable.",
)
ruleset: RulesetValue
# ---------- Helpers to convert to desired shape ----------
# ---------- Helpers ----------
_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()
# try integer parse
try:
return int(s)
except Exception:
try:
# sometimes it's "0.0" or similar
return int(float(s))
except Exception:
return None
# nested dicts sometimes appear
if isinstance(val, dict):
# look for common keys
for key in ("priority", "prio"):
if key in val:
return parse_priority(val.get(key))
# try nested dict values
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.
Covers common constructs; falls back to JSON dump for unknown constructs.
(Used for display in GET /rules).
"""
if expr is None:
return ""
if isinstance(expr, list):
tokens: List[str] = []
for part in expr:
if isinstance(part, dict):
# common tokens
if "match" in part:
m = part["match"]
left = m.get("left")
@@ -404,51 +315,31 @@ def rule_text_from_expr(expr: Any) -> str:
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": ..., "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", "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"]
# 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": {}})
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")
# try multiple keys for priority/prio shapes
ch_priority = parse_priority(ch.get("priority") if "priority" in ch else ch.get("prio") if "prio" in ch else ch.get("priority", 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,
@@ -459,7 +350,6 @@ def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]:
"rules": [],
}
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
@@ -469,7 +359,6 @@ def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]:
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"]
fam = r.get("family")
@@ -480,40 +369,25 @@ def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]:
if fam and table_name and chain_name:
tables.setdefault((fam, table_name), {"family": fam, "name": table_name, "chains": {}})
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,
"text": rule_text_from_expr(expr),
}
# include other useful metadata if present
rule_obj: Dict[str, Any] = {"handle": handle, "expr": expr, "text": rule_text_from_expr(expr)}
if "position" in r:
rule_obj["position"] = r["position"]
if "comment" in r:
rule_obj["comment"] = r["comment"]
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])):
tdata = tables[(fam, tname)]
chains_list: List[Dict[str, Any]] = []
@@ -530,122 +404,28 @@ def build_predictable_ruleset(nft_json: Dict[str, Any]) -> Dict[str, Any]:
}
)
result["tables"].append({"family": fam, "name": tname, "chains": chains_list})
return result
# ---------- Helpers to render expr -> textual nft (best-effort) ----------
def expr_to_text(expr: Any) -> Optional[str]:
"""
Best-effort renderer that converts a typical nft JSON expr (list) into a textual
fragment suitable to append to 'add rule <family> <table> <chain> ...'.
Returns None when it cannot deterministically render the provided expr.
"""
if expr is None:
return ""
if isinstance(expr, str):
return expr
if not isinstance(expr, list):
return None
parts: List[str] = []
for element in expr:
if isinstance(element, dict):
if "drop" in element:
parts.append("drop")
continue
if "accept" in element:
parts.append("accept")
continue
if "counter" in element:
parts.append("counter")
continue
if "queue" in element:
q = element["queue"]
token = "queue"
if isinstance(q, dict):
num = q.get("num") or q.get("number") or q.get("queue_number") or q.get("from") or q.get("range")
if num is not None:
token += f" num {num}"
if q.get("bypass"):
token += " bypass"
elif isinstance(q, (int, float)):
token += f" num {int(q)}"
elif isinstance(q, str):
token += f" num {q}"
parts.append(token)
continue
if "match" in element:
m = element["match"]
left = m.get("left")
right = m.get("right")
if isinstance(left, dict) and "payload" in left and isinstance(right, (str, int)):
p = left["payload"]
prot = p.get("protocol")
field = p.get("field")
if prot and field and isinstance(right, str):
if field == "protocol":
parts.append(f"{prot} {field} {right}")
continue
parts.append(f"payload({prot}.{field}) {right}")
continue
parts.append("match")
continue
if "payload" in element:
p = element["payload"]
prot = p.get("protocol")
field = p.get("field")
if prot and field:
parts.append(f"payload({prot}.{field})")
continue
parts.append("payload")
continue
if "tcp" in element or "udp" in element:
proto = "tcp" if "tcp" in element else "udp"
val = element.get(proto)
if isinstance(val, dict):
if "dport" in val:
parts.append(f"{proto} dport {val['dport']}")
continue
if "sport" in val:
parts.append(f"{proto} sport {val['sport']}")
continue
parts.append(proto)
continue
return None
else:
parts.append(str(element))
return " ".join(parts).strip()
# ---------- New helper: parse_ruleset_text ----------
# ---------- Text parsing helpers ----------
def parse_ruleset_text(nft_text: str) -> Dict[Tuple[str, str, str], List[Dict[str, Optional[Union[str, int]]]]]:
"""
Parse the full textual `nft list ruleset` output and return a mapping:
(family, table, chain) -> [ { "line": "<text line>", "handle": <int or None> }, ... ]
"""
result: Dict[Tuple[str, str, str], List[Dict[str, Optional[Union[str, int]]]]] = {}
if not nft_text:
return result
table_re = re.compile(r"^\s*table\s+(\S+)\s+(\S+)\s*\{")
# chain header line often: 'chain forward {', may include trailing comments
chain_re = re.compile(r"^\s*chain\s+(\S+)\s*\{")
handle_re = re.compile(r"#\s*handle\s*(\d+)\b")
# lines that indicate chain metadata (type/hook/priority/policy), we want to skip these
chain_meta_re = re.compile(r"^\s*(type\b|hook\b|priority\b|policy\b)\b.*;")
current_family = None
current_table = None
current_chain = None
# iterate line-by-line
for raw_ln in nft_text.splitlines():
ln = raw_ln.rstrip("\n")
s = ln.strip()
# detect table header
m_table = table_re.match(ln)
if m_table:
current_family = m_table.group(1)
@@ -653,7 +433,6 @@ def parse_ruleset_text(nft_text: str) -> Dict[Tuple[str, str, str], List[Dict[st
current_chain = None
continue
# detect chain header (works even if the same line contains trailing comment)
m_chain = chain_re.match(ln)
if m_chain and current_family and current_table:
current_chain = m_chain.group(1)
@@ -661,16 +440,12 @@ def parse_ruleset_text(nft_text: str) -> Dict[Tuple[str, str, str], List[Dict[st
result.setdefault(key, [])
continue
# if we're inside a chain, collect rule-like lines
if current_family and current_table and current_chain:
# skip empty / pure brace lines
if s == "" or s == "{" or s == "}":
continue
# skip chain metadata lines like 'type filter hook forward priority 0;'
if chain_meta_re.match(s):
continue
# likely a rule line — extract optional handle comment
m_handle = handle_re.search(s)
handle_val: Optional[int] = None
if m_handle:
@@ -685,16 +460,7 @@ def parse_ruleset_text(nft_text: str) -> Dict[Tuple[str, str, str], List[Dict[st
return result
# ---------- New helper: populate_text_from_ruleset_text ----------
def populate_text_from_ruleset_text(custom: Dict[str, Any], nft_text: str) -> None:
"""
Uses the parsed full ruleset textual output (nft_text) to update rule['text']
in the 'custom' structure in-place. Matching strategy:
1) handle -> line
2) explicit 'position' field if present
3) index-based mapping (best-effort)
4) substring probe match
"""
if not nft_text:
return
@@ -711,6 +477,10 @@ def populate_text_from_ruleset_text(custom: Dict[str, Any], nft_text: str) -> No
continue
key = (fam, tname, cname)
textual_entries = parsed.get(key, [])
if not textual_entries:
# no textual lines for this chain; skip
continue
handle_map: Dict[int, str] = {}
ordered_lines: List[str] = []
for ent in textual_entries:
@@ -746,18 +516,10 @@ def populate_text_from_ruleset_text(custom: Dict[str, Any], nft_text: str) -> No
rule["text"] = ln
replaced = True
break
# leave as-is if not replaced
# only update when a match was found; otherwise keep JSON-derived text
# ---------- Existing populate_text_from_chain_text (fallback) ----------
def populate_text_from_chain_text(custom: Dict[str, Any]) -> None:
"""
Replace rule['text'] in the 'custom' predictable ruleset with the exact textual
rule lines as produced by `nft list chain <family> <table> <chain>` when possible.
This modifies `custom` in-place. If textual listing for a chain fails, we fall
back to the existing rule['text'] that was produced from JSON.
"""
tables = custom.get("tables") or []
for t in tables:
fam = t.get("family")
@@ -771,7 +533,6 @@ def populate_text_from_chain_text(custom: Dict[str, Any]) -> None:
try:
chain_text = mgr.list_chain_text(fam, tname, cname) or ""
lines = [ln.rstrip() for ln in chain_text.splitlines() if ln.strip() != ""]
# build handle -> line map
handle_map: Dict[str, str] = {}
for ln in lines:
m = re.search(r"\bhandle\s+(\d+)\b", ln)
@@ -808,12 +569,8 @@ def populate_text_from_chain_text(custom: Dict[str, Any]) -> None:
# ---------- Routes ----------
@router.get("/rules", response_model=RulesetOut, summary="List ruleset")
def list_rules():
"""
Returns the ruleset in a stable, strongly-typed JSON shape derived from `nft -j list ruleset`.
"""
try:
try:
nft_json = mgr.list_rules_json()
@@ -821,7 +578,6 @@ def list_rules():
logger.debug("could not obtain nft JSON ruleset: %s", e)
raise HTTPException(status_code=500, detail=f"Failed to obtain nft JSON ruleset: {e}")
# try to obtain textual ruleset too (best-effort)
nft_text = ""
try:
nft_text = mgr.list_rules_text()
@@ -830,37 +586,6 @@ def list_rules():
custom = build_predictable_ruleset(nft_json)
# If no tables/rules were discovered from JSON, attempt to build from textual listing
try:
if nft_text and (not custom.get("tables")):
# Build minimal structure from textual parse if JSON did not produce tables
parsed = parse_ruleset_text(nft_text)
if parsed:
# convert parsed -> custom structure
tables_map: Dict[Tuple[str, str], Dict[str, Any]] = {}
for (fam, tname, cname), entries in parsed.items():
tables_map.setdefault((fam, tname), {"family": fam, "name": tname, "chains": {}})
chain_rules: List[Dict[str, Any]] = []
for ent in entries:
ln = ent.get("line") or ""
h = ent.get("handle")
# we don't have JSON expr; leave expr None and text the line
chain_rules.append({"handle": h, "expr": None, "text": ln})
tables_map[(fam, tname)]["chains"].setdefault(cname, {"name": cname, "type": None, "hook": None, "priority": None, "policy": None, "rules": []})
tables_map[(fam, tname)]["chains"][cname]["rules"].extend(chain_rules)
# convert map to custom shape
custom = {"tables": []}
for (fam, tname) in sorted(tables_map.keys(), key=lambda k: (k[0], k[1])):
t = tables_map[(fam, tname)]
chains_list = []
for cname in sorted(t["chains"].keys()):
ch = t["chains"][cname]
chains_list.append({"name": ch["name"], "type": None, "hook": None, "priority": None, "policy": None, "rules": ch["rules"]})
custom["tables"].append({"family": fam, "name": tname, "chains": chains_list})
except Exception as e:
logger.debug("fallback build from textual parse failed: %s", e)
# Enrich rule['text'] using the full textual ruleset (handle-first, position, index)
try:
if nft_text:
@@ -874,7 +599,6 @@ def list_rules():
except Exception as e:
logger.debug("list_rules: populate_text_from_chain_text failed: %s", e)
# debug: log counts so you can see what's being returned
num_tables = len(custom.get("tables", []))
num_rules = sum(len(ch.get("rules", [])) for t in custom.get("tables", []) for ch in t.get("chains", []))
logger.info("list_rules: returning tables=%d rules=%d", num_tables, num_rules)
@@ -895,9 +619,6 @@ def list_rules():
summary="Create rule (JSON, expr required; returns ExecResult with rc/stdout/stderr)",
)
def create_rule_json(req: CreateRuleRequest):
"""
Create a rule from JSON (expr required).
"""
try:
family = req.family
table = req.table
@@ -971,12 +692,8 @@ def create_rule_json(req: CreateRuleRequest):
raise HTTPException(status_code=500, detail=str(e))
@router.delete("/rules/{handle}", status_code=status.HTTP_204_NO_CONTENT, summary="Delete rule by handle")
def delete_rule(handle: int, family: str = "inet", table: str = "filter", chain: str = "input"):
"""
Delete a rule by handle using textual nft command.
"""
try:
mgr.delete_rule_by_handle_text(family=family, table=table, chain=chain, handle=handle)
except ValueError as e:
@@ -992,9 +709,6 @@ def delete_rule(handle: int, family: str = "inet", table: str = "filter", chain:
@router.post("/raw", response_model=ExecResult, summary="Execute raw textual nft command")
def exec_raw(req: RawCmdRequest):
"""
Execute an arbitrary textual nft command and return structured {rc, stdout, stderr}.
"""
try:
res = mgr.cmd(req.cmd)
rc = int(res.get("rc", -1) or -1)
@@ -1003,4 +717,5 @@ def exec_raw(req: RawCmdRequest):
logger.exception("exec_raw failed")
raise HTTPException(status_code=500, detail=str(e))
app.include_router(router)