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

test3
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-27 17:52:44 +01:00
parent ae80e967ed
commit 86cc1e6e50
1 changed files with 31 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

75
backend/src/api/nft_manager.py
View File

@@ -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 <family> <table> <chain> ...
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,16 +177,11 @@ 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():
@@ -193,19 +191,16 @@ def parse_json_cmd_summary(obj: Dict[str, Any]) -> Dict[str, Any]:
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 ----------
# ---------- 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")
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))