malmert
82012e9640
Some checks failed
Build and Deploy MITM Webserver / build (push) Failing after 5s
- Updated API routes to remove redundant prefixes. - Introduced sniffer status fetching in the backend API. - Added Sniffing page and integrated it into the routing. - Created a new hook for backend API interactions. - Implemented sniffer status type definitions.
100 lines
3.0 KiB
Python
100 lines
3.0 KiB
Python
from fastapi import APIRouter, HTTPException
|
|
from pydantic import BaseModel, Field
|
|
from typing import Dict, Any, List, Optional
|
|
|
|
from src.network_sniffer import (
|
|
get_sniffer_status,
|
|
start_afpacket_sniffer,
|
|
stop_afpacket_sniffer,
|
|
)
|
|
|
|
router = APIRouter()
|
|
|
|
# ------------------------------
|
|
# Pydantic Models
|
|
# ------------------------------
|
|
|
|
class SnifferStartRequest(BaseModel):
|
|
"""
|
|
Request model for starting the sniffer on a specific bridge.
|
|
"""
|
|
bridge: str = Field(..., example="br0", description="Name of the Linux bridge to sniff on")
|
|
|
|
|
|
class SnifferStartResponse(BaseModel):
|
|
"""
|
|
Response model returned when sniffer starts successfully.
|
|
"""
|
|
started: bool = Field(..., description="Whether the sniffer was started successfully")
|
|
bridge: str = Field(..., description="Bridge where the sniffer was started")
|
|
|
|
|
|
class SnifferStopResponse(BaseModel):
|
|
"""
|
|
Response model returned when the sniffer stops successfully.
|
|
"""
|
|
stopped: bool = Field(..., description="Whether the sniffer was stopped successfully")
|
|
|
|
|
|
class InterfaceSnifferStatus(BaseModel):
|
|
"""
|
|
Status of an individual interface monitored by the AF_PACKET sniffer.
|
|
"""
|
|
running: bool = Field(..., description="Whether the sniffer thread 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"
|
|
)
|
|
|
|
|
|
# ------------------------------
|
|
# Endpoints
|
|
# ------------------------------
|
|
|
|
@router.post("/start", response_model=SnifferStartResponse)
|
|
def sniffer_start(req: SnifferStartRequest):
|
|
"""
|
|
Start the AF_PACKET sniffer for the given bridge.
|
|
"""
|
|
try:
|
|
start_afpacket_sniffer(req.bridge)
|
|
return SnifferStartResponse(started=True, bridge=req.bridge)
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=500, detail=f"Failed to start sniffer: {exc}")
|
|
|
|
|
|
@router.post("/stop", response_model=SnifferStopResponse)
|
|
def sniffer_stop():
|
|
"""
|
|
Stop the AF_PACKET sniffer (if running).
|
|
"""
|
|
try:
|
|
stop_afpacket_sniffer()
|
|
return SnifferStopResponse(stopped=True)
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=500, detail=f"Failed to stop sniffer: {exc}")
|
|
|
|
|
|
@router.get("/status", response_model=SnifferStatusResponse)
|
|
def sniffer_status():
|
|
"""
|
|
Return the sniffer status information.
|
|
"""
|
|
try:
|
|
raw = get_sniffer_status()
|
|
# Convert raw dict → typed model
|
|
typed = {
|
|
k: InterfaceSnifferStatus(**v)
|
|
for k, v in raw.items()
|
|
}
|
|
return SnifferStatusResponse(interfaces=typed)
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=500, detail=f"Failed to query sniffer status: {exc}")
|