From 86cc1e6e5084bf0a72e5f524ed8ed326bcd5164c Mon Sep 17 00:00:00 2001 From: malmert Date: Tue, 27 Jan 2026 17:52:44 +0100 Subject: [PATCH] test3 --- backend/src/api/nft_manager.py | 83 +++++++++++++--------------------- 1 file changed, 31 insertions(+), 52 deletions(-) diff --git a/backend/src/api/nft_manager.py b/backend/src/api/nft_manager.py index 2d8658a..26c24c1 100644 --- a/backend/src/api/nft_manager.py +++ b/backend/src/api/nft_manager.py @@ -1,22 +1,26 @@ # app.py """ -Unrestricted nftables FastAPI service (libnftables only) with a preview endpoint. +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. Endpoints (high level): GET /firewall/rules -> list rules (kernel-provided rules with handles) - POST /firewall/rules -> add rule (executes; accepts 'cmd' or 'json') + 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) + Requirements: - 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, Union +from typing import Any, Dict, List, Optional from fastapi import FastAPI, APIRouter, HTTPException, status from pydantic import BaseModel, Field import logging @@ -105,22 +109,30 @@ class NftManager: # ---------- FastAPI + Router ---------- +app = FastAPI(title="Unrestricted nftables API (libnftables only)") router = APIRouter(prefix="/firewall", tags=["firewall"]) mgr = NftManager() # Request models -class CreateRuleRequest(BaseModel): - 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 CreateRuleJsonRequest(BaseModel): + json: Dict[str, Any] = Field(..., 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 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] @@ -129,20 +141,13 @@ class RuleOut(BaseModel): expr: Optional[Any] -# ---------- PREVIEW helpers ---------- +# ---------- PREVIEW helpers (unchanged) ---------- 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() @@ -151,7 +156,6 @@ def parse_text_cmd_summary(cmd: str) -> Dict[str, Any]: 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() @@ -160,7 +164,6 @@ def parse_text_cmd_summary(cmd: str) -> Dict[str, Any]: 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: @@ -174,38 +177,30 @@ def parse_text_cmd_summary(cmd: str) -> Dict[str, Any]: 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(): + 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")} + 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 ---------- +# ---------- Routes ---------- # List rules @router.get("/rules", response_model=List[RuleOut]) @@ -227,34 +222,27 @@ def list_rules(): raise HTTPException(status_code=500, detail=str(e)) -# Create rule: executes +# Create/execute rule via JSON only (structured) @router.post("/rules", status_code=status.HTTP_201_CREATED) -def create_rule(req: CreateRuleRequest): +def create_rule_json(req: CreateRuleJsonRequest): """ - Create a rule by sending either textual 'cmd' or libnftables 'json' object. - This endpoint executes the command. + Execute a libnftables JSON command object. + Use this endpoint to add structured rules or multi-op transactions. """ try: - 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") + out = mgr.json_cmd(req.json) + return {"status": "ok", "output": out} except NftError as e: - logger.warning("create_rule failed: %s", e) + logger.warning("create_rule_json failed: %s", e) raise HTTPException(status_code=400, detail=str(e)) except Exception as e: - logger.exception("create_rule internal error") + logger.exception("create_rule_json internal error") raise HTTPException(status_code=500, detail=str(e)) # PREVIEW endpoint (does NOT execute anything) @router.post("/preview") -def preview_rule(req: CreateRuleRequest): +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), @@ -315,12 +303,3 @@ def exec_raw(req: RawCmdRequest): 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))