nft improv 3
All checks were successful
Build and Deploy MITM Webserver / build (push) Successful in 9s
All checks were successful
Build and Deploy MITM Webserver / build (push) Successful in 9s
This commit is contained in:
@@ -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 <family> <table> <chain> handle <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 <family> <table> <chain> handle <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)
|
||||
|
||||
Reference in New Issue
Block a user