feat: add endpoints for creating and removing network bridges with request models
All checks were successful
Build and Deploy MITM Webserver / build (push) Successful in 8s

This commit is contained in:
2025-11-24 20:09:31 +01:00
parent 3488ec12a6
commit cdebdde29c

View File

@@ -1,4 +1,4 @@
from fastapi import APIRouter, Depends from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
from typing import List, Optional from typing import List, Optional
from pyroute2 import IPRoute, NDB from pyroute2 import IPRoute, NDB
@@ -98,6 +98,13 @@ class BridgeInfo(BaseModel):
stp_state: Optional[int] = Field(None, description="STP (Spanning Tree Protocol) state of the bridge") stp_state: Optional[int] = Field(None, description="STP (Spanning Tree Protocol) state of the bridge")
members: List[BridgeInterfaceInfo] = Field(default_factory=list, description="List of member interfaces of the bridge") members: List[BridgeInterfaceInfo] = Field(default_factory=list, description="List of member interfaces of the bridge")
class BridgeCreateRequest(BaseModel):
name: str
interfaces: List[str]
class BridgeRemoveRequest(BaseModel):
name: str
# ------------------------------ # ------------------------------
# Lazy Init Functions # Lazy Init Functions
# ------------------------------ # ------------------------------
@@ -176,6 +183,17 @@ def parse_flags(flags_int: int) -> list[str]:
result.append(name) result.append(name)
return result return result
def iface_index(name: str, ip: IPRoute = Depends(get_iproute)) -> int:
idx = ip.link_lookup(ifname=name)
if not idx:
raise HTTPException(status_code=404, detail=f"Interface {name} not found")
return idx[0]
def bridge_exists(name: str, ip: IPRoute = Depends(get_iproute)) -> bool:
return bool(ip.link_lookup(ifname=name))
# ------------------------------ # ------------------------------
# Endpoints # Endpoints
# ------------------------------ # ------------------------------
@@ -330,4 +348,46 @@ def full_state(
"interfaces": get_interfaces(ip), "interfaces": get_interfaces(ip),
"routes": get_routes(ip), "routes": get_routes(ip),
"bridges": get_bridges(), # uses NDB internally "bridges": get_bridges(), # uses NDB internally
}
@router.post("/create")
def create_bridge(req: BridgeCreateRequest, ip: IPRoute = Depends(get_iproute)):
if bridge_exists(req.name):
raise HTTPException(400, f"Bridge {req.name} already exists")
# Bridge erzeugen
ip.link("add", ifname=req.name, kind="bridge")
br_idx = iface_index(req.name)
# Interfaces hinzufügen
for iface in req.interfaces:
idx = iface_index(iface)
ip.link("set", index=idx, master=br_idx)
# Bridge aktivieren
ip.link("set", index=br_idx, state="up")
return {
"status": "ok",
"bridge": req.name,
"interfaces": req.interfaces
}
@router.post("/remove")
def remove_bridge(req: BridgeRemoveRequest, ip: IPRoute = Depends(get_iproute)):
if not bridge_exists(req.name):
raise HTTPException(404, f"Bridge {req.name} not found")
br_idx = iface_index(req.name)
# Bridge runterfahren
ip.link("set", index=br_idx, state="down")
# Bridge löschen
ip.link("del", index=br_idx)
return {
"status": "ok",
"deleted": req.name
} }