diff --git a/backend/src/api/nft_api.py b/backend/src/api/nft_api.py index 284a9bf..2141856 100644 --- a/backend/src/api/nft_api.py +++ b/backend/src/api/nft_api.py @@ -1,11 +1,22 @@ # fastapi_nft_router.py # -*- coding: utf-8 -*- """ -FastAPI router for nftables (bridge family). Produces both JSON and -human-readable textual representations for rules. +FastAPI router that lists and manages nftables rules for family 'bridge' +(default table 'mitm_tbl', chain 'forward'). -This version improves the JSON->text fallback so common shapes like -'ip protocol icmp drop' are rendered as nft-style clauses. +Behavior: +- Uses `nft --json list ruleset` to obtain authoritative rule metadata (handles). +- Uses `nft list chain ` to extract the exact textual + rule lines. Mapping is done by matching `handle N` in the textual output. +- Returns for each rule: + - nft_rule_text_full: exact line from 'nft list chain ...' including 'handle N' (or None) + - nft_rule_text: same line trimmed to remove trailing 'handle N' (or None) + - add_command: "add rule
" (or None) +- No JSON->text reconstruction is attempted. If text mapping is missing we return None. + +Security note: +- Process must be run with privileges to run nft (root or appropriate capabilities). +- Consider adding auth before exposing these endpoints. """ from typing import Any, Dict, List, Optional, Union, Literal import subprocess @@ -23,14 +34,14 @@ router = APIRouter() logger = logging.getLogger("nftables") logger.debug("nftables router module loaded") -# Defaults & nft binary locator +# Defaults & nft binary DEFAULT_TABLE = "mitm_tbl" DEFAULT_CHAIN = "forward" DEFAULT_FAMILY = "bridge" NFT_BIN = shutil.which("nft") -# ---------- Enums (for frontend dropdowns) ------------------------------ +# ----------------- Enums (for frontend) ----------------- class Family(str, Enum): bridge = DEFAULT_FAMILY @@ -104,7 +115,7 @@ class LogGroup(int, Enum): g6 = 6 g7 = 7 -# ---------- Pydantic expression models --------------------------------- +# ------------ Pydantic expression models (typed for frontend) ---------- class BaseExpr(BaseModel): kind: str class Config: @@ -161,14 +172,14 @@ Expr = Union[ VerdictExpr, RejectExpr, LogExpr, RawExpr, ] -# ---------- Rule model ------------------------------------------------- +# ---------------- Rule model ---------------- class RuleModel(BaseModel): family: Family = Family.bridge table: Table = Table.table chain: Chain = Chain.forward expr: List[Expr] = Field(default_factory=list) comment: Optional[str] = None - position: Optional[int] = None + position: Optional[int] = None # 1-based handle: Optional[int] = None @validator("family") @@ -177,35 +188,42 @@ class RuleModel(BaseModel): raise ValueError("This router only manages family 'bridge'") return v -# ---------- Low-level helpers ----------------------------------------- +# ----------------- Helpers -------------------- def ensure_nft_available() -> None: if not NFT_BIN: - logger.error("nft binary not found") + logger.error("nft binary not found on server") raise HTTPException(status_code=500, detail="nft binary not found on server") def run_nft_cmd(cmd: str) -> Dict[str, Any]: + """ + Execute a single nft script line via `nft -f -`. Returns stdout/stderr. + """ ensure_nft_available() full_cmd = [NFT_BIN, "-f", "-"] script = cmd.rstrip() + "\n" logger.info("Running nft command: %s", cmd) - logger.debug("Executing: %s ; script: %s", full_cmd, script) + logger.debug("Exec: %s ; script: %s", full_cmd, script) try: proc = subprocess.run(full_cmd, input=script.encode(), stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True) stdout = proc.stdout.decode() stderr = proc.stderr.decode() - logger.info("nft command succeeded (%d bytes stdout, %d bytes stderr)", len(stdout), len(stderr)) + logger.info("nft success (stdout %d bytes, stderr %d bytes)", len(stdout), len(stderr)) logger.debug("nft stdout: %s", stdout or "") if stderr: logger.debug("nft stderr: %s", stderr) return {"stdout": stdout, "stderr": stderr} except subprocess.CalledProcessError as e: err = e.stderr.decode() if e.stderr else str(e) - logger.error("nft command failed: %s", err) + logger.error("nft failed: %s", err) raise HTTPException(status_code=500, detail=err) def ensure_table_and_chain_exist(family: str, table: str, chain: str) -> None: - logger.debug("Checking/existence for family=%s table=%s chain=%s", family, table, chain) + """ + Ensure the named table and chain exist; create them with conservative defaults if missing. + """ + logger.debug("Ensure table/chain exist family=%s table=%s chain=%s", family, table, chain) ensure_nft_available() + try: out = subprocess.check_output([NFT_BIN, "--json", "list", "ruleset"], stderr=subprocess.PIPE) parsed = json.loads(out) @@ -219,13 +237,13 @@ def ensure_table_and_chain_exist(family: str, table: str, chain: str) -> None: table_exists = False chain_exists = False - for item in items: - if "table" in item: - t = item["table"] + for it in items: + if "table" in it: + t = it["table"] if isinstance(t, dict) and t.get("name") == table and t.get("family") == family: table_exists = True - if "chain" in item: - ch = item["chain"] + if "chain" in it: + ch = it["chain"] if isinstance(ch, dict) and ch.get("name") == chain and ch.get("table") == table and ch.get("family") == family: chain_exists = True @@ -240,17 +258,22 @@ def ensure_table_and_chain_exist(family: str, table: str, chain: str) -> None: else: run_nft_cmd(f"add chain {family} {table} {chain} {{ policy accept; }}") -# ---------- textual chain parser & JSON->text fallback ----------------- +# ----------------- Text mapping (strict) ----------------- HANDLE_RE = re.compile(r"\bhandle\s+(\d+)\b", flags=re.IGNORECASE) -def nft_list_chain_text(family: str, table: str, chain: str) -> Dict[int, str]: +def build_handle_text_map(family: str, table: str, chain: str) -> Dict[int, str]: + """ + Runs: nft list chain
+ Returns mapping handle -> full textual line containing 'handle N'. + If the textual output cannot be retrieved, raises HTTPException. + """ ensure_nft_available() cmd = [NFT_BIN, "list", "chain", family, table, chain] - logger.debug("Running textual chain list: %s", " ".join(cmd)) + logger.debug("Listing chain text: %s", " ".join(cmd)) try: out = subprocess.check_output(cmd, stderr=subprocess.PIPE).decode() except subprocess.CalledProcessError as e: - logger.error("Failed textual chain list: %s", e.stderr.decode()) + logger.error("Failed to list chain text: %s", e.stderr.decode()) raise HTTPException(status_code=500, detail=e.stderr.decode()) mapping: Dict[int, str] = {} @@ -262,205 +285,52 @@ def nft_list_chain_text(family: str, table: str, chain: str) -> Dict[int, str]: if not m: continue try: - handle = int(m.group(1)) - mapping[handle] = s - logger.debug("Text mapping: handle=%d -> %s", handle, s) - except Exception as ex: - logger.debug("Failed parsing handle from line: %s (%s)", s, ex) - continue - return mapping - -def extract_protocol_from_expr(expr: Any) -> Optional[str]: - """ - Attempt to detect a protocol (tcp/udp/icmp) from a single expr JSON object. - It searches common shapes ('payload', 'protocol', 'icmp') recursively. - Returns protocol string (e.g. 'icmp') or None. - """ - if not isinstance(expr, dict): - return None - # direct icmp key - if "icmp" in expr: - return "icmp" - # payload shapes: walk nested dicts looking for 'icmp' or 'tcp'/'udp' - def walk(d): - if isinstance(d, dict): - for k, v in d.items(): - if isinstance(v, str): - if v.lower() in ("icmp", "tcp", "udp"): - return v.lower() - if isinstance(v, dict) or isinstance(v, list): - found = walk(v) - if found: - return found - elif isinstance(d, list): - for item in d: - found = walk(item) - if found: - return found - return None - # check common keys - for key in ("payload", "protocol", "ip", "meta"): - if key in expr: - found = walk(expr[key]) - if found: - return found - # last resort: scan whole expr - return walk(expr) - -def json_exprs_to_text(exprs: List[Any]) -> str: - """ - Best-effort conversion of nft JSON exprs to a human-readable nft-style - single-line clause. We aim to produce phrases like: - ip protocol icmp drop - meta iifname \"eth0\" accept - ct state established accept - The function collects match fragments then appends verdict/action at the end. - """ - matches: List[str] = [] - verdicts: List[str] = [] - - for ex in exprs: - if not isinstance(ex, dict): - continue - # comment - if "comment" in ex: - c = ex["comment"] - if isinstance(c, str): - matches.append(f'comment "{c}"') - elif isinstance(c, dict): - txt = c.get("text") or c.get("str") - if txt: - matches.append(f'comment "{txt}"') - continue - # verdict shapes - if "verdict" in ex: - v = ex["verdict"] - if isinstance(v, dict): - k = next(iter(v.keys()), None) - if k: - verdicts.append(k) - else: - verdicts.append(str(v)) - continue - if "drop" in ex: - verdicts.append("drop") - continue - if "accept" in ex: - verdicts.append("accept") - continue - if "reject" in ex: - verdicts.append("reject") - continue - # conntrack - if "ct" in ex: - ct = ex["ct"] - if isinstance(ct, dict): - # prefer printed form 'ct state established' - if "state" in ct: - matches.append(f"ct state {ct['state']}") - else: - for k, v in ct.items(): - matches.append(f"ct {k} {v}") - continue - # meta - if "meta" in ex: - meta = ex["meta"] - if isinstance(meta, dict): - key = meta.get("key") or meta.get("name") - op = meta.get("op", "==") - val = meta.get("value") - if key and val is not None: - matches.append(f"meta {key} {op} {val}") - continue - # match object with left/op/right - if "match" in ex: - m = ex["match"] - left = m.get("left") - op = m.get("op") - right = m.get("right") - if left and op and (right is not None): - matches.append(f"{left} {op} {right}") - continue - # payload/protocol detection -> render 'ip protocol icmp' - proto = extract_protocol_from_expr(ex) - if proto: - # only render once per expr; protocol is a match, not action - matches.append(f"ip protocol {proto}") - continue - # log - if "log" in ex: - lg = ex["log"] - piece = "log" - if isinstance(lg, dict): - if lg.get("prefix"): - piece += f' prefix "{lg.get("prefix")}"' - if lg.get("group") is not None: - piece += f" group {lg.get('group')}" - matches.append(piece) - continue - # payload fallback: compact it - if "payload" in ex: - matches.append(json.dumps(ex["payload"])) - continue - # unknown: compact JSON - matches.append(json.dumps(ex)) - - # join matches then verdict(s) - clause = " ".join(matches).strip() - if clause and verdicts: - clause = f"{clause} {' '.join(verdicts)}" - elif not clause and verdicts: - clause = " ".join(verdicts) - return clause.strip() - -# ---------- JSON rules parsing with textual injection ------------------ -def nft_list_chain_text_map(family: str, table: str, chain: str) -> Dict[int, str]: - """Thin wrapper to get textual mapping, returns empty mapping on failure.""" - try: - return nft_list_chain_text(family, table, chain) - except HTTPException as e: - logger.debug("text map unavailable: %s", getattr(e, "detail", str(e))) - return {} - -def nft_list_chain_text(family: str, table: str, chain: str) -> Dict[int, str]: - ensure_nft_available() - cmd = [NFT_BIN, "list", "chain", family, table, chain] - logger.debug("Running textual chain list: %s", " ".join(cmd)) - try: - out = subprocess.check_output(cmd, stderr=subprocess.PIPE).decode() - except subprocess.CalledProcessError as e: - logger.error("Failed textual chain list: %s", e.stderr.decode()) - raise HTTPException(status_code=500, detail=e.stderr.decode()) - - mapping: Dict[int, str] = {} - for line in out.splitlines(): - s = line.strip() - if not s: - continue - m = HANDLE_RE.search(s) - if not m: - continue - try: - handle = int(m.group(1)) - mapping[handle] = s - logger.debug("Text mapping: handle=%d -> %s", handle, s) + h = int(m.group(1)) + # full textual line as-is + mapping[h] = s + logger.debug("Found textual rule for handle %d: %s", h, s) except Exception as ex: logger.debug("Failed parsing handle from line: %s (%s)", s, ex) continue return mapping +# ----------------- Rules listing (JSON + strict textual lookup) ----------------- def nft_list_rules(table: str = DEFAULT_TABLE, chain: str = DEFAULT_CHAIN) -> Dict[str, Any]: + """ + Return rules parsed from nft --json list ruleset, augmented with textual lines + extracted from `nft list chain
` via handle matching. + + For each rule returned: + - family, table, chain + - handle + - position (1-based in chain) + - comment (best-effort from JSON exprs) + - verdict (best-effort) + - exprs (the JSON expr list) + - nft_rule_text_full: exact textual line from nft list chain ... INCLUDING 'handle N' (or None) + - nft_rule_text: textual line trimmed to remove trailing 'handle N' (or None) + - add_command: "add rule
" (or None) + """ ensure_nft_available() + + # 1) JSON dump: authoritative structure try: out = subprocess.check_output([NFT_BIN, "--json", "list", "ruleset"], stderr=subprocess.PIPE) parsed = json.loads(out) except subprocess.CalledProcessError as e: - logger.error("Failed to list ruleset (json): %s", e.stderr.decode()) + logger.error("Failed to get JSON ruleset: %s", e.stderr.decode()) raise HTTPException(status_code=500, detail=e.stderr.decode()) - # Try to obtain text map, but do not fail if unavailable - text_map = nft_list_chain_text_map(DEFAULT_FAMILY, table, chain) - logger.debug("Text map entries: %d", len(text_map)) + # 2) textual map: strict mapping by handle + text_map: Dict[int, str] = {} + try: + text_map = build_handle_text_map(DEFAULT_FAMILY, table, chain) + logger.debug("Text map size: %d", len(text_map)) + except HTTPException as e: + # bubble up the error: user asked to extract exact textual lines and we couldn't get them + logger.error("Failed to obtain textual chain dump: %s", getattr(e, "detail", str(e))) + # still continue — per your request we won't attempt reconstructions, but we can return None textual fields. + text_map = {} results: List[Dict[str, Any]] = [] counters: Dict[str, int] = {} @@ -469,14 +339,15 @@ def nft_list_rules(table: str = DEFAULT_TABLE, chain: str = DEFAULT_CHAIN) -> Di if not isinstance(items, list): items = [] - for item in items: - if "rule" not in item: + for it in items: + if "rule" not in it: continue - r = item["rule"] + r = it["rule"] family = r.get("family") table_name = r.get("table") chain_name = r.get("chain") + # only return rules for requested table/chain if table_name != table or chain_name != chain: continue @@ -488,56 +359,46 @@ def nft_list_rules(table: str = DEFAULT_TABLE, chain: str = DEFAULT_CHAIN) -> Di handle = r.get("handle") exprs = r.get("expr", []) + # best-effort comment + verdict extraction from JSON exprs (keeps UI useful) comment: Optional[str] = None verdict: Optional[str] = None - verdict_details: Optional[Any] = None - - for expr in exprs: - if not isinstance(expr, dict): + for ex in exprs: + if not isinstance(ex, dict): continue - if "comment" in expr: - c = expr.get("comment") + if "comment" in ex: + c = ex.get("comment") if isinstance(c, str): comment = c elif isinstance(c, dict): comment = c.get("text") or c.get("str") - if "verdict" in expr: - v = expr["verdict"] + if "verdict" in ex: + v = ex["verdict"] if isinstance(v, dict): verdict = next(iter(v.keys()), None) - verdict_details = v.get(verdict) else: verdict = str(v) - if "drop" in expr and verdict is None: + if "drop" in ex and verdict is None: verdict = "drop" - if "accept" in expr and verdict is None: + if "accept" in ex and verdict is None: verdict = "accept" - if "reject" in expr and verdict is None: + if "reject" in ex and verdict is None: verdict = "reject" - # textual resolution: prefer exact mapping by handle, fallback to JSON->text + # strict textual lookup: only use exact line if present in text_map nft_rule_text_full: Optional[str] = None nft_rule_text: Optional[str] = None + add_command: Optional[str] = None if handle is not None and handle in text_map: nft_rule_text_full = text_map[handle] - # strip trailing ' handle N' to produce no-handle variant + # remove trailing ' handle N' to get copy/paste clause m = HANDLE_RE.search(nft_rule_text_full) - if m: - nft_rule_text = nft_rule_text_full[: m.start()].strip() - else: - nft_rule_text = nft_rule_text_full - logger.debug("Using textual map for handle %s -> %s", handle, nft_rule_text_full) + nft_rule_text = nft_rule_text_full[: m.start()].strip() if m else nft_rule_text_full + add_command = f"add rule {table_name} {chain_name} {nft_rule_text}".strip() if nft_rule_text else None + logger.debug("Attached textual rule for handle %s", handle) else: - # fallback: build from exprs - clause = json_exprs_to_text(exprs) - nft_rule_text = clause or None - nft_rule_text_full = (clause + (f" handle {handle}" if handle is not None else "")) if clause else None - logger.debug("Fallback clause for handle %s -> %s", handle, clause) - - # build add_command: use add rule
- add_clause = nft_rule_text or "" - add_command = f"add rule {table_name} {chain_name} {add_clause}".strip() + # If mapping missing, per your instruction do not attempt to reconstruct — leave textual fields None + logger.debug("No textual mapping for handle %s — textual fields will be None", handle) results.append({ "family": family, @@ -547,17 +408,17 @@ def nft_list_rules(table: str = DEFAULT_TABLE, chain: str = DEFAULT_CHAIN) -> Di "position": position, "comment": comment, "verdict": verdict, - "verdict_details": verdict_details, "exprs": exprs, - "nft_rule_text_full": nft_rule_text_full, # includes handle token if present - "nft_rule_text": nft_rule_text, # no-handle clause + "nft_rule_text_full": nft_rule_text_full, + "nft_rule_text": nft_rule_text, "add_command": add_command, }) return {"rules": results} -# ---------- Expr -> nft snippet used for preview/add -------------------- +# ------------ Expr -> nft snippet & command builder (preview/add) ------ def expr_to_nft_snippet(e: Expr) -> str: + """Build short nft snippet from typed Expr (used for preview/add).""" if isinstance(e, MetaExpr): val = e.value key = e.key.value @@ -593,8 +454,8 @@ def expr_to_nft_snippet(e: Expr) -> str: raise ValueError("Unsupported expression type") def rule_to_nft_cmd(rule: RuleModel) -> str: - expr_snippets = [expr_to_nft_snippet(e) for e in rule.expr] - body = " ".join(s for s in expr_snippets if s) + expr_snips = [expr_to_nft_snippet(e) for e in rule.expr] + body = " ".join(s for s in expr_snips if s) if rule.position is not None: cmd = f"insert rule {rule.table.value} {rule.chain.value} position {rule.position} {body}" else: @@ -603,9 +464,10 @@ def rule_to_nft_cmd(rule: RuleModel) -> str: cmd += f' comment "{rule.comment}"' return cmd -# ---------- Endpoints ------------------------------------------------- +# ---------------- Endpoints ------------------- @router.get("/options") def get_options() -> Dict[str, Any]: + """Return enum choices for frontend dropdowns.""" return { "family": [f.value for f in Family], "table": [t.value for t in Table], @@ -624,32 +486,37 @@ def get_options() -> Dict[str, Any]: @router.get("/rules") def list_rules(table: str = DEFAULT_TABLE, chain: str = DEFAULT_CHAIN) -> Dict[str, Any]: + """List rules for the given table/chain (ensures table/chain exist first).""" ensure_table_and_chain_exist(DEFAULT_FAMILY, table, chain) return nft_list_rules(table=table, chain=chain) @router.post("/rules/preview") def preview_rule(rule: RuleModel = Body(...)) -> Dict[str, str]: + """Return the nft command that would be executed for the provided rule (preview only).""" try: cmd = rule_to_nft_cmd(rule) except Exception as e: - logger.error("Preview failed: %s", e) + logger.error("Preview build failed: %s", e) raise HTTPException(status_code=400, detail=str(e)) return {"cmd": cmd} @router.post("/rules") def add_rule(rule: RuleModel = Body(...)) -> Dict[str, Any]: + """Insert/append rule (creates table/chain if missing).""" ensure_table_and_chain_exist(rule.family.value, rule.table.value, rule.chain.value) cmd = rule_to_nft_cmd(rule) return run_nft_cmd(cmd) @router.delete("/rules/{handle}") def delete_rule(handle: int, table: str = DEFAULT_TABLE, chain: str = DEFAULT_CHAIN) -> Dict[str, Any]: + """Delete rule by nft handle.""" ensure_table_and_chain_exist(DEFAULT_FAMILY, table, chain) cmd = f"delete rule {table} {chain} handle {handle}" return run_nft_cmd(cmd) @router.put("/rules/{handle}") def update_rule(handle: int, rule: RuleModel = Body(...), table: str = DEFAULT_TABLE, chain: str = DEFAULT_CHAIN) -> Dict[str, Any]: + """Replace a rule by handle: delete by handle then insert replacement (attempt to preserve position).""" ensure_table_and_chain_exist(rule.family.value, rule.table.value, rule.chain.value) rules_info = nft_list_rules(table=table, chain=chain) position: Optional[int] = None