AF_PACKET terminology cleanup
All checks were successful
Build and Deploy MITM Webserver / traffic_target (push) Successful in 0s
Build and Deploy MITM Webserver / build (push) Successful in 10s

This commit is contained in:
2026-04-12 17:41:22 +02:00
parent cb4c6ff564
commit 9ee47284e0
11 changed files with 69 additions and 45 deletions

View File

@@ -1,4 +1,4 @@
"""HTTP API for starting, stopping, and inspecting sniffer sessions."""
"""HTTP API for starting, stopping, and inspecting packet capture sessions."""
from typing import Any, Dict, Optional
@@ -7,16 +7,16 @@ from pydantic import BaseModel, Field
from src.network_sniffer import (
get_internal_debug_state,
get_sniffer_status,
start_afpacket_sniffer,
stop_afpacket_sniffer,
get_capture_session_status,
start_capture_session,
stop_capture_session,
)
router = APIRouter()
class SnifferStartRequest(BaseModel):
"""Request payload for starting a sniffer session."""
"""Request payload for starting a packet capture session."""
bridge: Optional[str] = Field(
None,
@@ -60,7 +60,7 @@ class InterfaceSnifferStatus(BaseModel):
running: bool = Field(..., description="Whether a sniffer is currently active.")
exists: bool = Field(..., description="Whether the interface exists on the host.")
up: bool = Field(..., description="Whether the interface is operationally up.")
session_id: Optional[str] = Field(None, description="Owning sniffer session ID.")
session_id: Optional[str] = Field(None, description="Owning capture session ID.")
session_label: Optional[str] = Field(None, description="Human-readable session label.")
@@ -75,13 +75,13 @@ class SnifferStatusResponse(BaseModel):
@router.post("/start", response_model=SnifferStartResponse)
def sniffer_start(req: SnifferStartRequest) -> SnifferStartResponse:
"""Start one sniffer session for exactly one target."""
"""Start one packet capture session for exactly one target."""
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:
session_id = start_afpacket_sniffer(req.interface, target_is_interface=True)
session_id = start_capture_session(req.interface, target_is_interface=True)
return SnifferStartResponse(
started=True,
session_id=session_id,
@@ -89,7 +89,7 @@ def sniffer_start(req: SnifferStartRequest) -> SnifferStartResponse:
target_type="interface",
)
session_id = start_afpacket_sniffer(req.bridge, target_is_interface=False)
session_id = start_capture_session(req.bridge, target_is_interface=False)
return SnifferStartResponse(
started=True,
session_id=session_id,
@@ -117,7 +117,7 @@ def sniffer_stop(
"""Stop by session ID, target query, or globally when no selector is given."""
if body and body.session_id:
try:
stop_afpacket_sniffer(session_id=body.session_id)
stop_capture_session(session_id=body.session_id)
return SnifferStopResponse(stopped=True, session_id=body.session_id, target=None, target_type=None)
except Exception as exc:
raise HTTPException(status_code=500, detail=f"Failed to stop session {body.session_id}: {exc}") from exc
@@ -127,14 +127,14 @@ def sniffer_stop(
try:
if q_interface:
stop_afpacket_sniffer(target=q_interface, target_is_interface=True)
stop_capture_session(target=q_interface, target_is_interface=True)
return SnifferStopResponse(stopped=True, session_id=None, target=q_interface, target_type="interface")
if q_bridge:
stop_afpacket_sniffer(target=q_bridge, target_is_interface=False)
stop_capture_session(target=q_bridge, target_is_interface=False)
return SnifferStopResponse(stopped=True, session_id=None, target=q_bridge, target_type="bridge")
stop_afpacket_sniffer()
stop_capture_session()
return SnifferStopResponse(stopped=True, session_id=None, target=None, target_type=None)
except Exception as exc:
raise HTTPException(status_code=500, detail=f"Failed to stop sniffer: {exc}") from exc
@@ -142,9 +142,9 @@ def sniffer_stop(
@router.get("/status", response_model=SnifferStatusResponse)
def sniffer_status() -> SnifferStatusResponse:
"""Return current sniffer status per interface."""
"""Return current capture-session status per interface."""
try:
raw: Dict[str, Dict[str, Any]] = get_sniffer_status()
raw: Dict[str, Dict[str, Any]] = get_capture_session_status()
typed = {key: InterfaceSnifferStatus(**value) for key, value in raw.items()}
return SnifferStatusResponse(interfaces=typed)
except Exception as exc:
@@ -153,7 +153,7 @@ def sniffer_status() -> SnifferStatusResponse:
@router.get("/debug")
def sniffer_debug() -> Dict[str, Any]:
"""Return internal sniffer and packet-tracker debug state."""
"""Return internal capture-session and packet-tracker debug state."""
try:
return get_internal_debug_state()
except Exception as exc: