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

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

Browse Source
This commit is contained in:
malmert
2026-01-11 17:51:28 +01:00
parent 96cf5d573c
commit 80030ec07c
1 changed files with 116 additions and 249 deletions
Download Patch File Download Diff File Expand all files Collapse all files

365
backend/src/api/nft_api.py
View File

@@ -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 <family> <table> <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 <table> <chain> <nft_rule_text>" (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 "<empty>")
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 <family> <table> <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 <family> <table> <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 <table> <chain> <nft_rule_text>" (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 <table> <chain> <clause>
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