From e9ff417df1d135182e56d74b29d76657c47dcb4f Mon Sep 17 00:00:00 2001 From: malmert Date: Sun, 1 Mar 2026 14:43:59 +0100 Subject: [PATCH] fix 3 --- backend/src/api/sniffer_api.py | 38 ++++++++++++---------------------- 1 file changed, 13 insertions(+), 25 deletions(-) diff --git a/backend/src/api/sniffer_api.py b/backend/src/api/sniffer_api.py index 018b0b9..dc9a4be 100644 --- a/backend/src/api/sniffer_api.py +++ b/backend/src/api/sniffer_api.py @@ -1,6 +1,6 @@ # src/routers/sniffer.py from fastapi import APIRouter, HTTPException, Query -from pydantic import BaseModel, Field, root_validator +from pydantic import BaseModel, Field from typing import Dict, Any, Optional from src.network_sniffer import ( @@ -18,51 +18,31 @@ router = APIRouter() class SnifferStartRequest(BaseModel): """ Request model for starting the sniffer on a specific bridge OR interface. - Exactly one of `bridge` or `interface` must be provided. + Both fields are optional here; the endpoint will validate that exactly one is present. """ bridge: Optional[str] = Field(None, example="br0", description="Name of the Linux bridge to sniff on") interface: Optional[str] = Field(None, example="eth0", description="Name of the network interface to sniff on") - @root_validator - def exactly_one_target(cls, values): - b, i = values.get("bridge"), values.get("interface") - if bool(b) == bool(i): - # either both set or both unset -> error - raise ValueError("Exactly one of 'bridge' or 'interface' must be provided") - return values - class SnifferStartResponse(BaseModel): - """ - Response model returned when sniffer starts successfully. - """ started: bool = Field(..., description="Whether the sniffer was started successfully") target: str = Field(..., description="Target that was started (bridge or interface)") target_type: str = Field(..., description="Either 'bridge' or 'interface'") class SnifferStopResponse(BaseModel): - """ - Response model returned when the sniffer stops successfully. - """ stopped: bool = Field(..., description="Whether the sniffer was stopped successfully") target: Optional[str] = Field(None, description="Target stopped; null if global stop") target_type: Optional[str] = Field(None, description="'bridge' or 'interface' or None") class InterfaceSnifferStatus(BaseModel): - """ - Status of an individual interface monitored by the AF_PACKET sniffer. - """ running: bool = Field(..., description="Whether the sniffer thread/socket is active") exists: bool = Field(..., description="Whether the interface exists in /sys/class/net") up: bool = Field(..., description="Whether the interface is operationally UP") class SnifferStatusResponse(BaseModel): - """ - Response model for the sniffer status endpoint. - """ interfaces: Dict[str, InterfaceSnifferStatus] = Field( ..., description="Map of interface names to their sniffer status" ) @@ -75,13 +55,19 @@ class SnifferStatusResponse(BaseModel): def sniffer_start(req: SnifferStartRequest): """ Start the AF_PACKET sniffer for the given bridge OR interface. + + Exactly one of `bridge` or `interface` must be provided; validate here explicitly so this + is compatible across Pydantic versions. """ + # explicit validation (works with pydantic v1 & v2) + if bool(req.bridge) == bool(req.interface): + raise HTTPException(status_code=400, detail="Exactly one of 'bridge' or 'interface' must be provided") + try: if req.interface: start_afpacket_sniffer(req.interface, target_is_interface=True) return SnifferStartResponse(started=True, target=req.interface, target_type="interface") else: - # req.bridge is guaranteed present by validator start_afpacket_sniffer(req.bridge, target_is_interface=False) return SnifferStartResponse(started=True, target=req.bridge, target_type="bridge") except Exception as exc: @@ -100,15 +86,17 @@ def sniffer_stop( - If `?interface=eth0` provided: stop sniffing on that interface only (close socket). - If `?bridge=br0` provided: stop snapshot/sockets associated with that bridge. """ + if bridge and interface: + raise HTTPException(status_code=400, detail="Only one of 'bridge' or 'interface' may be provided") + try: - if interface and bridge: - raise HTTPException(status_code=400, detail="Only one of 'bridge' or 'interface' may be provided") if interface: stop_afpacket_sniffer(target=interface, target_is_interface=True) return SnifferStopResponse(stopped=True, target=interface, target_type="interface") if bridge: stop_afpacket_sniffer(target=bridge, target_is_interface=False) return SnifferStopResponse(stopped=True, target=bridge, target_type="bridge") + # global stop stop_afpacket_sniffer() return SnifferStopResponse(stopped=True, target=None, target_type=None)