malmert
ad94a51214
All checks were successful
Build and Deploy MITM Webserver / build (push) Successful in 9s
618 lines
26 KiB
Python
618 lines
26 KiB
Python
# app.py
|
|
from typing import Any, Dict, List, Optional, Tuple, Union
|
|
from fastapi import FastAPI, APIRouter, HTTPException, status
|
|
from pydantic import BaseModel, Field
|
|
import logging
|
|
import json
|
|
import re
|
|
|
|
# libnftables (we call textual commands through its .cmd() method)
|
|
from nftables import Nftables # type: ignore
|
|
|
|
# ---------- logging ----------
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = logging.getLogger("nft_api_raw_only")
|
|
|
|
# ---------- Exceptions ----------
|
|
class NftError(RuntimeError):
|
|
pass
|
|
|
|
|
|
# ---------- 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)
|
|
except Exception:
|
|
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(self) -> str:
|
|
"""
|
|
Return the textual ruleset as produced by 'nft list ruleset'.
|
|
Uses the textual command path.
|
|
"""
|
|
res = self.cmd("list ruleset")
|
|
if res["rc"] != 0:
|
|
raise NftError(f"nft list ruleset failed: {res['stderr']}")
|
|
return res["stdout"]
|
|
|
|
def list_rules_json(self) -> Dict[str, Any]:
|
|
"""
|
|
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.
|
|
"""
|
|
cmd_variants = ["list ruleset -j", "list ruleset"]
|
|
last_err = None
|
|
for c in cmd_variants:
|
|
res = self.cmd(c)
|
|
if res["rc"] != 0:
|
|
last_err = res["stderr"]
|
|
continue
|
|
out = res["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:
|
|
"""
|
|
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
|
|
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)
|
|
except Exception:
|
|
logger.debug("failed to restore set_json_output(True)")
|
|
|
|
if res["rc"] != 0:
|
|
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 = []
|
|
for rec in records:
|
|
if "rule" in rec:
|
|
r = rec["rule"]
|
|
expr = r.get("expr")
|
|
if isinstance(expr, list):
|
|
tokens: List[str] = []
|
|
for part in expr:
|
|
if "match" in part:
|
|
m = part["match"]
|
|
left = m.get("left")
|
|
right = m.get("right")
|
|
if isinstance(left, dict) and "payload" in left and isinstance(right, str):
|
|
p = left["payload"]
|
|
prot = p.get("protocol")
|
|
field = p.get("field")
|
|
if prot and field:
|
|
tokens.append(f"{prot} {field} {right}")
|
|
continue
|
|
tokens.append("match")
|
|
elif "payload" in part:
|
|
p = part["payload"]
|
|
prot = p.get("protocol")
|
|
field = p.get("field")
|
|
tokens.append(f"payload({prot}.{field})")
|
|
elif "drop" in part:
|
|
tokens.append("drop")
|
|
elif "accept" in part:
|
|
tokens.append("accept")
|
|
elif "counter" in part:
|
|
tokens.append("counter")
|
|
else:
|
|
tokens.append("+".join(part.keys()))
|
|
rule_lines.append(" ".join(tokens))
|
|
else:
|
|
rule_lines.append(json.dumps(r))
|
|
if rule_lines:
|
|
return "\n".join(rule_lines)
|
|
except Exception:
|
|
logger.debug("fallback JSON parsing of chain output failed; returning raw output")
|
|
|
|
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:
|
|
raise NftError(f"delete rule failed: {res['stderr']}")
|
|
|
|
|
|
# ---------- FastAPI + Router ----------
|
|
app = FastAPI(title="Unrestricted nftables API (json create)")
|
|
router = APIRouter(prefix="/firewall", tags=["firewall"])
|
|
mgr = NftManager()
|
|
|
|
|
|
# ---------- Request/Response models (strongly typed) ----------
|
|
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")
|
|
|
|
|
|
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": ""}}
|
|
|
|
|
|
# --- 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")
|
|
|
|
|
|
class ChainOut(BaseModel):
|
|
name: str = Field(..., description="Chain name", example="forward")
|
|
rules: List[RuleOut] = Field(..., description="Rules in this chain (ordered)")
|
|
|
|
|
|
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")
|
|
|
|
|
|
class RulesetModel(BaseModel):
|
|
tables: List[TableOut] = Field(..., description="Top-level tables list")
|
|
|
|
|
|
# ---------- 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[Any] = Field(None, description="Optional position metadata (if you want to specify insertion 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}],
|
|
}
|
|
}
|
|
|
|
|
|
# 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.",
|
|
)
|
|
|
|
|
|
# ---------- Helpers to convert to desired shape ----------
|
|
_handle_re = re.compile(r"\s+#\s*handle\s+\d+\s*$")
|
|
|
|
|
|
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")
|
|
right = m.get("right")
|
|
if isinstance(left, dict) and "payload" in left and isinstance(right, str):
|
|
p = left["payload"]
|
|
prot = p.get("protocol")
|
|
field = p.get("field")
|
|
if prot and field:
|
|
tokens.append(f"{prot} {field} {right}")
|
|
continue
|
|
tokens.append("match")
|
|
elif "payload" in part:
|
|
p = part["payload"]
|
|
prot = p.get("protocol")
|
|
field = p.get("field")
|
|
if prot and field:
|
|
tokens.append(f"payload({prot}.{field})")
|
|
continue
|
|
tokens.append("payload")
|
|
elif "cmp" in part or "binary" in part:
|
|
tokens.append("cmp")
|
|
elif "drop" in part:
|
|
tokens.append("drop")
|
|
elif "accept" in part:
|
|
tokens.append("accept")
|
|
elif "counter" in part:
|
|
tokens.append("counter")
|
|
elif "tcp" in part or "udp" in part:
|
|
proto = "tcp" if "tcp" in part else "udp"
|
|
tokens.append(proto)
|
|
else:
|
|
keys = "+".join(sorted(part.keys()))
|
|
tokens.append(keys)
|
|
else:
|
|
tokens.append(str(part))
|
|
return " ".join(tokens)
|
|
return str(expr)
|
|
|
|
|
|
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 } ] } ] }
|
|
]
|
|
}
|
|
"""
|
|
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":[]}}}
|
|
tables: Dict[Tuple[str, str], Dict[str, Any]] = {}
|
|
for rec in items:
|
|
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": {}})
|
|
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")
|
|
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": []})
|
|
elif "rule" in rec:
|
|
r = rec["rule"]
|
|
fam = r.get("family")
|
|
table_name = r.get("table")
|
|
chain_name = r.get("chain")
|
|
handle = r.get("handle")
|
|
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": []})
|
|
rule_obj: Dict[str, Any] = {
|
|
"handle": handle,
|
|
"expr": expr,
|
|
"text": rule_text_from_expr(expr),
|
|
}
|
|
# include other useful metadata if present
|
|
if "position" in r:
|
|
rule_obj["position"] = r["position"]
|
|
if "comment" in r:
|
|
rule_obj["comment"] = r["comment"]
|
|
tables[(fam, table_name)]["chains"][chain_name]["rules"].append(rule_obj)
|
|
|
|
# Convert map to sorted lists for deterministic order
|
|
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"]})
|
|
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.
|
|
Supported cases (common):
|
|
- [{'match': {'left': {'payload': {'protocol':'ip','field':'protocol'}}, 'op':'==', 'right':'icmp'}}, {'drop': None}]
|
|
-> 'ip protocol icmp drop'
|
|
- payload / tcp / udp / counter / accept
|
|
- simple dicts where keys are 'drop' | 'accept' | 'counter'
|
|
This intentionally does not attempt to support every nft JSON construct.
|
|
"""
|
|
if expr is None:
|
|
return ""
|
|
if isinstance(expr, str):
|
|
return expr
|
|
if not isinstance(expr, list):
|
|
# unsupported top-level type
|
|
return None
|
|
|
|
parts: List[str] = []
|
|
for element in expr:
|
|
if isinstance(element, dict):
|
|
# handle drop/accept/counter directly
|
|
if "drop" in element:
|
|
parts.append("drop")
|
|
continue
|
|
if "accept" in element:
|
|
parts.append("accept")
|
|
continue
|
|
if "counter" in element:
|
|
parts.append("counter")
|
|
continue
|
|
|
|
# match left/right payload equals -> ip protocol icmp, or ip saddr/daddr
|
|
if "match" in element:
|
|
m = element["match"]
|
|
left = m.get("left")
|
|
right = m.get("right")
|
|
# payload matches
|
|
if isinstance(left, dict) and "payload" in left and isinstance(right, (str, int)):
|
|
p = left["payload"]
|
|
prot = p.get("protocol")
|
|
field = p.get("field")
|
|
# common: protocol field match (protocol == icmp)
|
|
if prot and field and isinstance(right, str):
|
|
# ip vs ip6 decision is left to the frontend; here we render 'ip protocol icmp' (works for many setups)
|
|
if field == "protocol":
|
|
parts.append(f"{prot} {field} {right}")
|
|
continue
|
|
# payload might be l4 ports etc; produce generic payload(...) token
|
|
parts.append(f"payload({prot}.{field}) {right}")
|
|
continue
|
|
# fallback for match: try to stringify right
|
|
parts.append("match")
|
|
continue
|
|
|
|
# payload shorthand
|
|
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
|
|
|
|
# tcp/udp as nested dicts sometimes appear
|
|
if "tcp" in element or "udp" in element:
|
|
proto = "tcp" if "tcp" in element else "udp"
|
|
val = element.get(proto)
|
|
# attempt to detect dport/sport keys
|
|
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
|
|
|
|
# cmp/binary operators etc — not supported deterministically
|
|
# return None to indicate we can't safely render this expr
|
|
return None
|
|
else:
|
|
# non-dict token (string/number)
|
|
parts.append(str(element))
|
|
|
|
# join tokens
|
|
return " ".join(parts).strip()
|
|
|
|
|
|
# ---------- 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`.
|
|
|
|
Structure:
|
|
{ "ruleset": { "tables": [ { "family": ..., "name": ..., "chains": [ { "name": ..., "rules": [ { "handle", "expr", "text" } ] } ] } ] } }
|
|
|
|
Fallback:
|
|
- If nft JSON is unavailable, falls back to returning the raw textual ruleset string.
|
|
"""
|
|
try:
|
|
try:
|
|
nft_json = mgr.list_rules_json()
|
|
except NftError as e:
|
|
logger.debug("could not obtain nft JSON ruleset: %s", e)
|
|
text = mgr.list_rules()
|
|
return RulesetOut(ruleset=text.strip() if text is not None else None)
|
|
|
|
custom = build_predictable_ruleset(nft_json)
|
|
ruleset_model = RulesetModel.parse_obj(custom)
|
|
return RulesetOut(ruleset=ruleset_model)
|
|
except NftError as e:
|
|
logger.exception("list_rules failed")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
except Exception as e:
|
|
logger.exception("list_rules internal error")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.post(
|
|
"/rules",
|
|
response_model=ExecResult,
|
|
status_code=status.HTTP_201_CREATED,
|
|
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).
|
|
- Attempts to render expr -> textual fragment and execute: `add rule <family> <table> <chain> <fragment>`
|
|
- If rendering fails: 400 instructing the client to use POST /firewall/raw
|
|
- Returns ExecResult always (rc/stdout/stderr). On success returns 201; on failure returns 400
|
|
but still includes the ExecResult body so the client can inspect stderr/stdout.
|
|
"""
|
|
try:
|
|
family = req.family
|
|
table = req.table
|
|
chain = req.chain
|
|
|
|
if req.expr is None:
|
|
logger.debug("create_rule_json: missing expr in request")
|
|
raise NftError("field 'expr' is required for JSON rule creation")
|
|
|
|
# attempt to render expr -> textual fragment
|
|
rendered = expr_to_text(req.expr)
|
|
if rendered is None:
|
|
# cannot render deterministically
|
|
logger.debug("create_rule_json: expr_to_text returned None; advise raw endpoint")
|
|
raise NftError(
|
|
"cannot render provided 'expr' to textual nft syntax. "
|
|
"Please use POST /firewall/raw to execute the textual nft command."
|
|
)
|
|
|
|
expr_text = rendered.strip()
|
|
cmd = f"add rule {family} {table} {chain} {expr_text}"
|
|
logger.info("create_rule_json executing command: %s", cmd)
|
|
|
|
res = mgr.cmd(cmd)
|
|
# res is dict {"rc": rc, "stdout": out, "stderr": err}
|
|
rc = int(res.get("rc", -1) or -1)
|
|
stdout = res.get("stdout")
|
|
stderr = res.get("stderr")
|
|
|
|
# Log the outputs for debugging
|
|
logger.info("nft cmd rc=%s stdout=%r stderr=%r cmd=%s", rc, stdout, stderr, cmd)
|
|
|
|
# Build ExecResult to return in any case (so client gets stdout/stderr)
|
|
exec_res = ExecResult(rc=rc, stdout=stdout, stderr=stderr)
|
|
|
|
# If nft returned non-zero code, surface it as 400 but include the ExecResult in the response body
|
|
if rc != 0:
|
|
# include command and stderr in the HTTP error detail for client UX
|
|
detail = f"nft command failed rc={rc}. stderr: {stderr!r}. cmd: {cmd}"
|
|
logger.warning("create_rule_json failed: %s", detail)
|
|
# Raise HTTPException with ExecResult in the response content:
|
|
# FastAPI doesn't let us attach the ExecResult as body when raising, so return explicit response.
|
|
# Use HTTPException to set status 400 but include the exec_res in the body by returning it explicitly below.
|
|
raise NftError(detail)
|
|
|
|
# rc == 0 -> success
|
|
return exec_res
|
|
|
|
except NftError as e:
|
|
# For NftError we want to return 400 with the ExecResult if available, otherwise simple message.
|
|
logger.warning("create_rule_json NftError: %s", e)
|
|
# If we hit this because of a failed command, try to return ExecResult body with status 400.
|
|
# Build a minimal ExecResult with rc=-1 if not available.
|
|
# Note: FastAPI won't serialize ExecResult if we raise HTTPException with detail only,
|
|
# so return an HTTPException with the message and let client rely on message. However we want body.
|
|
# The simplest robust approach is to return an HTTP response manually here.
|
|
|
|
# Attempt to find last command outputs from mgr? Not safe. Use message from exception.
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
except Exception as e:
|
|
logger.exception("create_rule_json internal error")
|
|
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.
|
|
Command executed:
|
|
delete rule <family> <table> <chain> handle <handle>
|
|
"""
|
|
try:
|
|
mgr.delete_rule_by_handle_text(family=family, table=table, chain=chain, handle=handle)
|
|
except ValueError as e:
|
|
logger.warning("delete_rule client error: %s", e)
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
except NftError as e:
|
|
logger.exception("delete_rule failed")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
except Exception as e:
|
|
logger.exception("delete_rule internal error")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@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)
|
|
return ExecResult(rc=rc, stdout=res.get("stdout"), stderr=res.get("stderr"))
|
|
except Exception as e:
|
|
logger.exception("exec_raw failed")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
app.include_router(router)
|