fix 3

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.
     """
-    try:
-        if interface and bridge:
+    if bridge and interface:
         raise HTTPException(status_code=400, detail="Only one of 'bridge' or 'interface' may be provided")
+
+    try:
         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)