diff --git a/backend/src/api/nft_manager.py b/backend/src/api/nft_manager.py index 26c24c1..366ca32 100644 --- a/backend/src/api/nft_manager.py +++ b/backend/src/api/nft_manager.py @@ -1,19 +1,16 @@ # app.py """ -Unrestricted nftables FastAPI service (libnftables only) with separate endpoints for -textual commands (/firewall/raw) and JSON commands (/firewall/jsoncmd and /firewall/rules), -plus a preview endpoint that does NOT execute. +Unrestricted nftables FastAPI service using textual nft commands only. Endpoints (high level): - GET /firewall/rules -> list rules (kernel-provided rules with handles) - POST /firewall/rules -> add/execute libnftables JSON command (executes) - DELETE /firewall/rules/{handle} -> delete rule by handle (executes) - POST /firewall/raw -> execute textual nft command (executes) - POST /firewall/jsoncmd -> execute libnftables JSON command (executes) - POST /firewall/preview -> PREVIEW what would be executed (no changes) + GET /firewall/rules -> return textual ruleset (raw nft output) + POST /firewall/rules -> execute a textual nft command (convenience) + DELETE /firewall/rules/{handle} -> delete rule by handle via textual nft command + POST /firewall/raw -> execute arbitrary textual nft command (executes) + POST /firewall/preview -> PREVIEW what textual command would do (no changes) Requirements: - - python-nftables installed + - python-nftables must be installed for the Nftables wrapper (we still use python-nftables to call .cmd()) - CAP_NET_ADMIN or root required to modify nftables WARNING: @@ -24,263 +21,135 @@ from typing import Any, Dict, List, Optional from fastapi import FastAPI, APIRouter, HTTPException, status from pydantic import BaseModel, Field import logging -import json import re -# libnftables (must be installed) +# 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") +logger = logging.getLogger("nft_api_raw_only") # ---------- Exceptions ---------- class NftError(RuntimeError): pass -# ---------- NftManager (unrestricted) ---------- + +# ---------- NftManager (textual-only) ---------- class NftManager: """ Thin wrapper around python-nftables exposing: - - json_cmd execution (accepts libnftables JSON dict) - - cmd execution (accepts textual nft commands) - - convenience list_rules + delete_by_handle - This class intentionally does NOT validate or restrict nft commands. + - cmd execution (textual nft commands via Nftables.cmd()) + - convenience list_rules_text + This class intentionally avoids json transactions: everything is textual 'nft' commands. """ def __init__(self) -> None: self.nft = Nftables() - # request json output where available try: + # prefer seeing JSON when python-nftables prints it, but we only use .cmd() below self.nft.set_json_output(True) except Exception: - logger.debug("set_json_output not available") - - def json_cmd(self, cmd_obj: Dict[str, Any]) -> Any: - """ - Execute a libnftables JSON command object via json_cmd and return parsed output. - """ - rc, out, err = self.nft.json_cmd(cmd_obj) - if rc != 0: - logger.error("json_cmd failed: %s", err) - raise NftError(err if err else f"json_cmd exit {rc}") - return out + 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 { "stdout": out_str, "stderr": err_str, "rc": rc }. + Returns dict { "rc": rc, "stdout": out_str, "stderr": err_str }. """ rc, out, err = self.nft.cmd(text_cmd) if rc != 0: - logger.warning("cmd returned rc=%s, err=%s", rc, err) + 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) -> List[Dict[str, Any]]: + def list_rules_text(self) -> str: """ - List the full ruleset and return collected 'rule' dicts (each includes handle when present). + Return the textual ruleset as produced by 'nft list ruleset'. + Uses the textual command path. """ - out = self.json_cmd({"nftables": [{"list": {"ruleset": None}}]}) - rules: List[Dict[str, Any]] = [] - for item in out.get("nftables", []): - if "rule" in item: - rules.append(item["rule"]) - return rules + res = self.cmd("list ruleset") + if res["rc"] != 0: + raise NftError(f"nft list ruleset failed: {res['stderr']}") + return res["stdout"] - def delete_by_handle(self, family: str, table: str, chain: str, handle: int) -> None: + def delete_rule_by_handle_text(self, family: str, table: str, chain: str, handle: int) -> None: """ - Delete rule by handle using JSON delete rule. No validation performed. + Delete a rule by handle using textual nft command: + delete rule handle """ if not isinstance(handle, int) or handle <= 0: - raise ValueError("handle must be positive integer") - payload = { - "nftables": [{ - "delete": { - "rule": { - "family": family, - "table": table, - "chain": chain, - "handle": handle - } - } - }] - } - self.json_cmd(payload) + 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 (libnftables only)") +app = FastAPI(title="Unrestricted nftables API (textual only)") router = APIRouter(prefix="/firewall", tags=["firewall"]) mgr = NftManager() -# Request models -class CreateRuleJsonRequest(BaseModel): - json: Dict[str, Any] = Field(..., description="libnftables JSON command object") - - +# ---------- Request/Response models ---------- class RawCmdRequest(BaseModel): cmd: str = Field(..., description="Textual nft command to execute") -class JsonCmdRequest(BaseModel): - json: Dict[str, Any] = Field(..., description="libnftables JSON command object") - - -class PreviewRequest(BaseModel): - # preview accepts either textual cmd or json object - cmd: Optional[str] = Field(None, description="Textual nft command (preview only)") - json: Optional[Dict[str, Any]] = Field(None, description="libnftables JSON command object (preview only)") - - -class RuleOut(BaseModel): - family: Optional[str] - table: Optional[str] - chain: Optional[str] - handle: Optional[int] - expr: Optional[Any] - - -# ---------- PREVIEW helpers (unchanged) ---------- -def parse_text_cmd_summary(cmd: str) -> Dict[str, Any]: - summary: Dict[str, Any] = {"operation": None, "family": None, "table": None, "chain": None, "remainder": cmd} - s = cmd.strip() - tokens = s.split() - if len(tokens) >= 1: - summary["operation"] = tokens[0].lower() - m = re.match(r'^(add|insert|delete|replace)\s+rule\s+(\S+)\s+(\S+)\s+(\S+)\b', s, flags=re.I) - if m: - summary["operation"] = m.group(1).lower() - summary["family"] = m.group(2) - summary["table"] = m.group(3) - summary["chain"] = m.group(4) - summary["remainder"] = s[m.end():].strip() - return summary - m2 = re.match(r'^(delete)\s+rule\s+(\S+)\s+(\S+)\s+(\S+)\s+handle\s+(\d+)', s, flags=re.I) - if m2: - summary["operation"] = m2.group(1).lower() - summary["family"] = m2.group(2) - summary["table"] = m2.group(3) - summary["chain"] = m2.group(4) - summary["remainder"] = f"handle {m2.group(5)}" - return summary - try: - idx = next(i for i,t in enumerate(tokens) if t.lower() == "rule") - if len(tokens) > idx + 3: - summary["family"] = tokens[idx+1] - summary["table"] = tokens[idx+2] - summary["chain"] = tokens[idx+3] - summary["remainder"] = " ".join(tokens[idx+4:]) if len(tokens) > idx+4 else "" - except StopIteration: - pass - return summary - - -def parse_json_cmd_summary(obj: Dict[str, Any]) -> Dict[str, Any]: - summary = {"entries": []} - nft_entries = obj.get("nftables") if isinstance(obj, dict) else None - if not isinstance(nft_entries, list): - return {"error": "not a libnftables JSON object with 'nftables' list"} - for item in nft_entries: - if not isinstance(item, dict): - continue - for k, v in item.items(): - entry = {"op": k} - if isinstance(v, dict) and "rule" in v and isinstance(v["rule"], dict): - r = v["rule"] - entry["family"] = r.get("family") - entry["table"] = r.get("table") - entry["chain"] = r.get("chain") - if "handle" in r: - entry["handle"] = r.get("handle") - else: - if isinstance(v, dict): - entry["info"] = {kk: vv for kk, vv in v.items() if kk in ("table", "chain", "family")} - summary["entries"].append(entry) - return summary - +class RulesetOut(BaseModel): + ruleset: str # ---------- Routes ---------- -# List rules -@router.get("/rules", response_model=List[RuleOut]) +# GET /firewall/rules -> textual nft ruleset output +@router.get("/rules", response_model=RulesetOut) def list_rules(): + """ + Returns the textual nft ruleset output (as a single string). + Use clients to parse as needed. + """ try: - rules = mgr.list_rules() - out = [] - for r in rules: - out.append({ - "family": r.get("family"), - "table": r.get("table"), - "chain": r.get("chain"), - "handle": r.get("handle"), - "expr": r.get("expr"), - }) - return out + text = mgr.list_rules_text() + return {"ruleset": text} 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)) -# Create/execute rule via JSON only (structured) +# POST /firewall/rules -> execute a textual nft command (convenience) @router.post("/rules", status_code=status.HTTP_201_CREATED) -def create_rule_json(req: CreateRuleJsonRequest): +def create_rule_text(req: RawCmdRequest): """ - Execute a libnftables JSON command object. - Use this endpoint to add structured rules or multi-op transactions. + Execute a textual nft command (convenience, returns raw stdout). + Example: add rule inet filter input ip saddr 10.0.0.0/8 drop """ try: - out = mgr.json_cmd(req.json) - return {"status": "ok", "output": out} + res = mgr.cmd(req.cmd) + if res["rc"] != 0: + raise NftError(f"cmd failed rc={res['rc']}: {res.get('stderr')}") + return {"status": "ok", "stdout": res.get("stdout")} except NftError as e: - logger.warning("create_rule_json failed: %s", e) + logger.warning("create_rule_text failed: %s", e) raise HTTPException(status_code=400, detail=str(e)) except Exception as e: - logger.exception("create_rule_json internal error") + logger.exception("create_rule_text internal error") raise HTTPException(status_code=500, detail=str(e)) -# PREVIEW endpoint (does NOT execute anything) -@router.post("/preview") -def preview_rule(req: PreviewRequest): - """ - Preview what would be applied if you executed the given 'cmd' or 'json'. - This endpoint only analyzes and returns: textual command (if present), pretty JSON (if present), - and a parsed summary (operation, family, table, chain, etc.). Nothing is executed. - """ - try: - if req.cmd: - text = req.cmd.strip() - summary = parse_text_cmd_summary(text) - return { - "type": "text", - "text_cmd": text, - "summary": summary, - "note": "This is a preview only. Nothing was executed." - } - if req.json: - pretty = json.dumps(req.json, indent=2, sort_keys=True) - summary = parse_json_cmd_summary(req.json) - return { - "type": "json", - "json_pretty": pretty, - "summary": summary, - "note": "This is a preview only. Nothing was executed." - } - raise HTTPException(status_code=400, detail="either 'cmd' or 'json' must be provided for preview") - except Exception as e: - logger.exception("preview_rule internal error") - raise HTTPException(status_code=500, detail=str(e)) - - -# Delete by handle (authoritative) +# DELETE /firewall/rules/{handle} -> delete by handle using textual nft command @router.delete("/rules/{handle}", status_code=status.HTTP_204_NO_CONTENT) def delete_rule(handle: int, family: str = "inet", table: str = "filter", chain: str = "input"): """ - Delete a rule by handle. family/table/chain are passed through to the delete JSON. + Delete a rule by handle using textual nft command. + Command executed: + delete rule
handle """ try: - mgr.delete_by_handle(family=family, table=table, chain=chain, handle=handle) + 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)) @@ -292,9 +161,12 @@ def delete_rule(handle: int, family: str = "inet", table: str = "filter", chain: raise HTTPException(status_code=500, detail=str(e)) -# Execute arbitrary textual nft command (convenience endpoint) +# Execute arbitrary textual nft command @router.post("/raw") def exec_raw(req: RawCmdRequest): + """ + Execute an arbitrary textual nft command and return {rc, stdout, stderr}. + """ try: res = mgr.cmd(req.cmd) return {"rc": res["rc"], "stdout": res["stdout"], "stderr": res["stderr"]} @@ -303,3 +175,5 @@ def exec_raw(req: RawCmdRequest): raise HTTPException(status_code=500, detail=str(e)) +# include router +app.include_router(router)