From ae80e967ed4c292e50477572de825080473bb7ba Mon Sep 17 00:00:00 2001 From: malmert Date: Tue, 27 Jan 2026 17:37:45 +0100 Subject: [PATCH] test2 --- backend/src/api/nft_manager.py | 411 +++++++++++++++++---------------- 1 file changed, 217 insertions(+), 194 deletions(-) diff --git a/backend/src/api/nft_manager.py b/backend/src/api/nft_manager.py index 70202f3..2d8658a 100644 --- a/backend/src/api/nft_manager.py +++ b/backend/src/api/nft_manager.py @@ -1,201 +1,94 @@ # app.py """ -Stateless nftables FastAPI service (libnftables only, no CLI fallback, no annotations). -Endpoints: +Unrestricted nftables FastAPI service (libnftables only) with a preview endpoint. + +Endpoints (high level): GET /firewall/rules -> list rules (kernel-provided rules with handles) - POST /firewall/rules -> add a constrained rule (returns 201) - DELETE /firewall/rules/{handle} -> delete rule by handle + POST /firewall/rules -> add rule (executes; accepts 'cmd' or 'json') + 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) Requirements: - python-nftables must be installed and usable: - pip install python-nftables fastapi uvicorn pydantic - The process must have CAP_NET_ADMIN (or be run as root) to modify nftables. + - python-nftables installed + - CAP_NET_ADMIN or root required to modify nftables +WARNING: + This service can run arbitrary nft commands. Run only in a trusted environment. """ -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Union from fastapi import FastAPI, APIRouter, HTTPException, status from pydantic import BaseModel, Field -import ipaddress import logging +import json +import re -# libnftables import (must be present) +# libnftables (must be installed) from nftables import Nftables # type: ignore # ---------- logging ---------- logging.basicConfig(level=logging.INFO) logger = logging.getLogger("nft_api") -# ---------- NftManager (libnftables only) ---------- +# ---------- Exceptions ---------- class NftError(RuntimeError): pass - +# ---------- NftManager (unrestricted) ---------- class NftManager: """ - Minimal libnftables wrapper that performs JSON transactions only via python-nftables. - No CLI fallback, no comments/annotations. + 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. """ - # Tighten these to your environment - ALLOWED_FAMILIES = {"inet"} - ALLOWED_TABLES = {"filter"} - ALLOWED_CHAINS = {"input", "output", "forward"} - ALLOWED_PROTOS = {"tcp", "udp"} - def __init__(self) -> None: self.nft = Nftables() - # Request JSON output where applicable + # request json output where available try: self.nft.set_json_output(True) except Exception: - # Some libnftables builds may ignore this; proceed anyway - logger.debug("set_json_output not available or failed") + logger.debug("set_json_output not available") - def _json_cmd(self, cmd: Dict[str, Any]) -> Any: - rc, out, err = self.nft.json_cmd(cmd) + 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("libnftables error: %s", err) - raise NftError(err) + logger.error("json_cmd failed: %s", err) + raise NftError(err if err else f"json_cmd exit {rc}") return out - def _validate_family_table_chain(self, family: str, table: str, chain: str) -> None: - if family not in self.ALLOWED_FAMILIES: - raise ValueError(f"family '{family}' not allowed") - if table not in self.ALLOWED_TABLES: - raise ValueError(f"table '{table}' not allowed") - if chain not in self.ALLOWED_CHAINS: - raise ValueError(f"chain '{chain}' not allowed") - - def _validate_proto(self, proto: Optional[str]) -> None: - if proto is None: - return - if proto not in self.ALLOWED_PROTOS: - raise ValueError(f"protocol '{proto}' not allowed") - - def _validate_src(self, src: Optional[str]) -> None: - if not src: - return - try: - # allow host or network - ipaddress.ip_network(src, strict=False) - except Exception as e: - raise ValueError(f"invalid src '{src}': {e}") - - def _validate_port(self, port: Optional[int]) -> None: - if port is None: - return - if not (1 <= port <= 65535): - raise ValueError(f"invalid port: {port}") + 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 }. + """ + rc, out, err = self.nft.cmd(text_cmd) + if rc != 0: + logger.warning("cmd returned rc=%s, err=%s", rc, err) + return {"rc": rc, "stdout": out, "stderr": err} def list_rules(self) -> List[Dict[str, Any]]: """ - Return raw rule dicts extracted from libnftables JSON output. - Each item is the inner 'rule' dict returned by nftables JSON. + List the full ruleset and return collected 'rule' dicts (each includes handle when present). """ - out = self._json_cmd({"nftables": [{"list": {"ruleset": None}}]}) + 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 - def add_rule( - self, - family: str, - table: str, - chain: str, - proto: Optional[str], - src: Optional[str], - dst_port: Optional[int], - verdict: str, - ) -> None: + def delete_by_handle(self, family: str, table: str, chain: str, handle: int) -> None: """ - Add a constrained rule via libnftables JSON. - This constructs an 'expr' list that covers: - - ip saddr match (if src provided) - - l4 proto match (if proto provided) - - dport match (if dst_port provided) - - final verdict (accept/drop) + Delete rule by handle using JSON delete rule. No validation performed. """ - # validate - self._validate_family_table_chain(family, table, chain) - self._validate_proto(proto) - self._validate_src(src) - self._validate_port(dst_port) - if verdict not in {"accept", "drop"}: - raise ValueError("verdict must be 'accept' or 'drop'") - - expr: List[Dict[str, Any]] = [] - - # match source address (IP) - if src: - # Construct prefix match using nft JSON structure (ip payload -> saddr) - # The json shape used here is compatible with python-nftables expectations. - # We use ip payload field and prefix match for networks. - try: - net = ipaddress.ip_network(src, strict=False) - addr = str(net.network_address) - prefix_len = net.prefixlen - except Exception: - # fallback to treating as single IP - addr = src - prefix_len = 32 if ":" not in src else 128 - expr.append({ - "match": { - "left": {"payload": {"protocol": "ip", "field": "saddr"}}, - "op": "==", - "right": {"prefix": {"addr": addr, "len": prefix_len}} - } - }) - - # match L4 protocol - if proto: - # meta.l4proto match is a common approach in JSON exprs - expr.append({ - "match": { - "left": {"meta": {"key": "l4proto"}}, - "op": "==", - "right": proto - } - }) - - # match destination port (only meaningful if proto provided) - if dst_port: - if not proto: - raise ValueError("dst_port requires proto to be set") - expr.append({ - "match": { - "left": {"payload": {"protocol": proto, "field": "dport"}}, - "op": "==", - "right": dst_port - } - }) - - # final verdict - expr.append({"verdict": verdict}) - - payload = { - "nftables": [{ - "add": { - "rule": { - "family": family, - "table": table, - "chain": chain, - "expr": expr - } - } - }] - } - - self._json_cmd(payload) - - def delete_rule(self, family: str, table: str, chain: str, handle: int) -> None: - """ - Delete a rule by handle (authoritative). - """ - self._validate_family_table_chain(family, table, chain) if not isinstance(handle, int) or handle <= 0: - raise ValueError("handle must be a positive integer") - + raise ValueError("handle must be positive integer") payload = { "nftables": [{ "delete": { @@ -208,22 +101,24 @@ class NftManager: } }] } - self._json_cmd(payload) + self.json_cmd(payload) -# ---------- FastAPI application (router combined) ---------- +# ---------- FastAPI + Router ---------- router = APIRouter(prefix="/firewall", tags=["firewall"]) -manager = NftManager() +mgr = NftManager() +# Request models class CreateRuleRequest(BaseModel): - family: str = Field("inet", description="family (whitelisted)") - table: str = Field("filter", description="table (whitelisted)") - chain: str = Field("input", description="chain (whitelisted)") - proto: Optional[str] = Field(None, description="tcp|udp") - src: Optional[str] = Field(None, description="source IP or CIDR") - dst_port: Optional[int] = Field(None, description="destination port") - verdict: str = Field("accept", description="accept|drop") + cmd: Optional[str] = Field(None, description="Textual nft command (e.g. 'add rule ...')") + json: Optional[Dict[str, Any]] = Field(None, description="libnftables JSON command object") + +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 RuleOut(BaseModel): @@ -234,15 +129,89 @@ class RuleOut(BaseModel): expr: Optional[Any] +# ---------- PREVIEW helpers ---------- +def parse_text_cmd_summary(cmd: str) -> Dict[str, Any]: + """ + Try to parse a short summary from a textual nft cmd. + We do NOT execute anything here — just regexp/token heuristics. + Returns a dict with possible keys: operation, family, table, chain, remainder. + """ + summary: Dict[str, Any] = {"operation": None, "family": None, "table": None, "chain": None, "remainder": cmd} + # normalize whitespace + s = cmd.strip() + tokens = s.split() + if len(tokens) >= 1: + summary["operation"] = tokens[0].lower() + # detect patterns like: add rule ... + 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 + # detect delete rule by handle: e.g. delete rule inet filter input handle 42 + 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 + # fallback: try to find family/table/chain tokens near 'rule' + 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]: + """ + Extract a short summary from a libnftables JSON command object. + Looks into top-level 'nftables' list for add/delete/replace/insert keys and extracts rule family/table/chain. + """ + 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: + # each item is like {"add": {"rule": {"family":...}}} or {"delete": {...}} + 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") + # include handle if present + if "handle" in r: + entry["handle"] = r.get("handle") + else: + # some commands are different shapes (e.g., add table ...). Try to capture names + # look for 'table' or 'chain' nested keys + 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 + + +# ---------- End preview helpers ---------- + +# List rules @router.get("/rules", response_model=List[RuleOut]) def list_rules(): - """ - Returns a list of rules (raw rule dicts from nftables) that include kernel handles. - Clients should save handles if they want to delete rules later. - """ try: - rules = manager.list_rules() - # Normalize fields for the response model: ensure expected keys exist + rules = mgr.list_rules() out = [] for r in rules: out.append({ @@ -250,7 +219,7 @@ def list_rules(): "table": r.get("table"), "chain": r.get("chain"), "handle": r.get("handle"), - "expr": r.get("expr") + "expr": r.get("expr"), }) return out except NftError as e: @@ -258,46 +227,100 @@ def list_rules(): raise HTTPException(status_code=500, detail=str(e)) +# Create rule: executes @router.post("/rules", status_code=status.HTTP_201_CREATED) def create_rule(req: CreateRuleRequest): """ - Create a constrained rule. Returns 201 on success. - Use GET /firewall/rules to obtain the kernel-assigned handle. + Create a rule by sending either textual 'cmd' or libnftables 'json' object. + This endpoint executes the command. """ try: - manager.add_rule( - family=req.family, - table=req.table, - chain=req.chain, - proto=req.proto, - src=req.src, - dst_port=req.dst_port, - verdict=req.verdict, - ) - return {"status": "created"} - except (ValueError, NftError) as e: - logger.warning("create_rule bad request: %s", e) + if req.cmd: + 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")} + if req.json: + out = mgr.json_cmd(req.json) + return {"status": "ok", "output": out} + raise HTTPException(status_code=400, detail="either 'cmd' or 'json' must be provided") + except NftError as e: + logger.warning("create_rule failed: %s", e) raise HTTPException(status_code=400, detail=str(e)) except Exception as e: logger.exception("create_rule internal error") raise HTTPException(status_code=500, detail=str(e)) -@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", -): +# PREVIEW endpoint (does NOT execute anything) +@router.post("/preview") +def preview_rule(req: CreateRuleRequest): """ - Delete a rule by handle. family/table/chain default to common values but must match. + 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: - manager.delete_rule(family=family, table=table, chain=chain, handle=handle) + 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) +@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. + """ + try: + mgr.delete_by_handle(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)) \ No newline at end of file + 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)) + + +# Execute arbitrary textual nft command (convenience endpoint) +@router.post("/raw") +def exec_raw(req: RawCmdRequest): + try: + res = mgr.cmd(req.cmd) + return {"rc": res["rc"], "stdout": res["stdout"], "stderr": res["stderr"]} + except Exception as e: + logger.exception("exec_raw failed") + raise HTTPException(status_code=500, detail=str(e)) + + +# Execute arbitrary JSON command +@router.post("/jsoncmd") +def exec_json(req: JsonCmdRequest): + try: + out = mgr.json_cmd(req.json) + return {"output": out} + except Exception as e: + logger.exception("exec_json failed") + raise HTTPException(status_code=400, detail=str(e))