reset button interfaces
All checks were successful
Build and Deploy MITM Webserver / traffic_target (push) Successful in 1s
Build and Deploy MITM Webserver / build (push) Successful in 10s

This commit is contained in:
2026-03-09 17:23:00 +01:00
parent 650b92d2c5
commit b836bf3f02
4 changed files with 141 additions and 1 deletions

View File

@@ -2,6 +2,8 @@
import asyncio
import logging
import os
import subprocess
from typing import Any, Dict, List, Optional
from fastapi import APIRouter, Depends, HTTPException, WebSocket, WebSocketDisconnect
@@ -16,6 +18,8 @@ from src.utilities.interface_bridge_helpers import get_bridge_ports_once, read_i
router = APIRouter()
logger = logging.getLogger("network_router")
_ETHTOOL_BIN = "/usr/sbin/ethtool" if os.path.exists("/usr/sbin/ethtool") else "ethtool"
DEFAULT_INTERFACE_MTU = 1500
ip: IPRoute | None = None
ndb: NDB | None = None
@@ -106,6 +110,30 @@ class BridgeLinkStateEnableRequest(BaseModel):
)
class InterfaceResetDefaultsRequest(BaseModel):
"""Payload for resetting one or more interfaces to baseline settings."""
interfaces: List[str] = Field(..., min_length=1, description="Interface names to reset.")
class InterfaceResetDefaultsResult(BaseModel):
"""Outcome for one interface reset attempt."""
interface: str = Field(..., description="Interface name.")
mtu: Optional[int] = Field(None, description="Resulting MTU after reset.")
ethernet_profile: Optional[Dict[str, Any]] = Field(
None,
description="Resulting speed/duplex/autoneg profile when available.",
)
message: str = Field(..., description="Human-readable reset result.")
class InterfaceResetDefaultsResponse(BaseModel):
"""Batch reset response for one or more interfaces."""
results: List[InterfaceResetDefaultsResult] = Field(default_factory=list)
class BridgeMemberLinkStateInfo(BaseModel):
"""Current link-state snapshot for one bridge member."""
@@ -282,6 +310,38 @@ def publish_network_state_update(reason: str) -> None:
logger.exception("Failed to publish network state update")
def _reset_interface_defaults(ifname: str, ip_route: IPRoute) -> InterfaceResetDefaultsResult:
"""Reset one interface to a conservative baseline configuration."""
idx = iface_index(ifname, ip_route)
messages: list[str] = []
ip_route.link("set", index=idx, state="down")
ip_route.link("set", index=idx, mtu=DEFAULT_INTERFACE_MTU)
try:
subprocess.run(
[_ETHTOOL_BIN, "-s", ifname, "autoneg", "on"],
capture_output=True,
text=True,
check=True,
)
messages.append("autoneg on")
except (FileNotFoundError, subprocess.CalledProcessError) as exc:
logger.debug("Failed to reset ethtool defaults on %s: %s", ifname, exc)
messages.append("autoneg unchanged")
ip_route.link("set", index=idx, state="up")
messages.append(f"mtu {DEFAULT_INTERFACE_MTU}")
messages.append("admin up")
return InterfaceResetDefaultsResult(
interface=ifname,
mtu=DEFAULT_INTERFACE_MTU,
ethernet_profile=read_interface_ethernet_profile(ifname),
message=", ".join(messages),
)
@router.get("/interfaces", response_model=List[InterfaceInfo])
def get_interfaces(ip: IPRoute = Depends(get_iproute)) -> List[InterfaceInfo]:
"""List host interfaces with addresses and decoded flags."""
@@ -422,6 +482,18 @@ def full_state(ip: IPRoute = Depends(get_iproute)) -> FullStateResponse:
return _build_full_state_response(ip)
@router.post("/interfaces/reset-defaults", response_model=InterfaceResetDefaultsResponse)
def reset_interfaces_to_defaults(
req: InterfaceResetDefaultsRequest,
ip: IPRoute = Depends(get_iproute),
) -> InterfaceResetDefaultsResponse:
"""Reset listed interfaces to baseline MTU/autoneg/up settings."""
unique_ifaces = list(dict.fromkeys(req.interfaces))
results = [_reset_interface_defaults(ifname, ip) for ifname in unique_ifaces]
publish_network_state_update("interfaces_reset_defaults")
return InterfaceResetDefaultsResponse(results=results)
@router.post("/bridge/create")
def create_bridge(req: BridgeCreateRequest, ip: IPRoute = Depends(get_iproute)) -> dict:
"""Create a bridge and attach listed interfaces."""