This commit is contained in:
@@ -1,5 +1,5 @@
|
|||||||
# src/routers/sniffer.py
|
# src/routers/sniffer.py
|
||||||
from fastapi import APIRouter, HTTPException, Query
|
from fastapi import APIRouter, HTTPException, Query, Body
|
||||||
from pydantic import BaseModel, Field
|
from pydantic import BaseModel, Field
|
||||||
from typing import Dict, Any, Optional
|
from typing import Dict, Any, Optional
|
||||||
|
|
||||||
@@ -18,20 +18,33 @@ router = APIRouter()
|
|||||||
class SnifferStartRequest(BaseModel):
|
class SnifferStartRequest(BaseModel):
|
||||||
"""
|
"""
|
||||||
Request model for starting the sniffer on a specific bridge OR interface.
|
Request model for starting the sniffer on a specific bridge OR interface.
|
||||||
Both fields are optional here; the endpoint will validate that exactly one is present.
|
Exactly one of `bridge` or `interface` must be provided.
|
||||||
"""
|
"""
|
||||||
bridge: Optional[str] = Field(None, example="br0", description="Name of the Linux bridge to sniff on")
|
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")
|
interface: Optional[str] = Field(None, example="eth0", description="Name of the network interface to sniff on")
|
||||||
|
|
||||||
|
|
||||||
class SnifferStartResponse(BaseModel):
|
class SnifferStartResponse(BaseModel):
|
||||||
|
"""
|
||||||
|
Response model returned when sniffer starts successfully.
|
||||||
|
"""
|
||||||
started: bool = Field(..., description="Whether the sniffer was started successfully")
|
started: bool = Field(..., description="Whether the sniffer was started successfully")
|
||||||
|
session_id: str = Field(..., description="Session identifier for this sniffer instance")
|
||||||
target: str = Field(..., description="Target that was started (bridge or interface)")
|
target: str = Field(..., description="Target that was started (bridge or interface)")
|
||||||
target_type: str = Field(..., description="Either 'bridge' or 'interface'")
|
target_type: str = Field(..., description="Either 'bridge' or 'interface'")
|
||||||
|
|
||||||
|
|
||||||
|
class SnifferStopRequest(BaseModel):
|
||||||
|
"""
|
||||||
|
Optional body for stop — prefer session_id if you want to stop a specific session.
|
||||||
|
If omitted, stopping behavior will be determined by query params (bridge/interface) or global stop.
|
||||||
|
"""
|
||||||
|
session_id: Optional[str] = Field(None, description="Session id to stop")
|
||||||
|
|
||||||
|
|
||||||
class SnifferStopResponse(BaseModel):
|
class SnifferStopResponse(BaseModel):
|
||||||
stopped: bool = Field(..., description="Whether the sniffer was stopped successfully")
|
stopped: bool = Field(..., description="Whether the sniffer was stopped successfully")
|
||||||
|
session_id: Optional[str] = Field(None, description="Session id stopped (if any)")
|
||||||
target: Optional[str] = Field(None, description="Target stopped; null if global stop")
|
target: Optional[str] = Field(None, description="Target stopped; null if global stop")
|
||||||
target_type: Optional[str] = Field(None, description="'bridge' or 'interface' or None")
|
target_type: Optional[str] = Field(None, description="'bridge' or 'interface' or None")
|
||||||
|
|
||||||
@@ -40,6 +53,8 @@ class InterfaceSnifferStatus(BaseModel):
|
|||||||
running: bool = Field(..., description="Whether the sniffer thread/socket is active")
|
running: bool = Field(..., description="Whether the sniffer thread/socket is active")
|
||||||
exists: bool = Field(..., description="Whether the interface exists in /sys/class/net")
|
exists: bool = Field(..., description="Whether the interface exists in /sys/class/net")
|
||||||
up: bool = Field(..., description="Whether the interface is operationally UP")
|
up: bool = Field(..., description="Whether the interface is operationally UP")
|
||||||
|
session_id: Optional[str] = Field(None, description="Session id owning this interface")
|
||||||
|
session_label: Optional[str] = Field(None, description="Human label for the session")
|
||||||
|
|
||||||
|
|
||||||
class SnifferStatusResponse(BaseModel):
|
class SnifferStatusResponse(BaseModel):
|
||||||
@@ -54,54 +69,60 @@ class SnifferStatusResponse(BaseModel):
|
|||||||
@router.post("/start", response_model=SnifferStartResponse)
|
@router.post("/start", response_model=SnifferStartResponse)
|
||||||
def sniffer_start(req: SnifferStartRequest):
|
def sniffer_start(req: SnifferStartRequest):
|
||||||
"""
|
"""
|
||||||
Start the AF_PACKET sniffer for the given bridge OR interface.
|
Start a sniffer session for the given bridge OR interface.
|
||||||
|
Exactly one of `bridge` or `interface` must be provided.
|
||||||
Exactly one of `bridge` or `interface` must be provided; validate here explicitly so this
|
Returns a session_id to manage the session.
|
||||||
is compatible across Pydantic versions.
|
|
||||||
"""
|
"""
|
||||||
# explicit validation (works with pydantic v1 & v2)
|
|
||||||
if bool(req.bridge) == bool(req.interface):
|
if bool(req.bridge) == bool(req.interface):
|
||||||
raise HTTPException(status_code=400, detail="Exactly one of 'bridge' or 'interface' must be provided")
|
raise HTTPException(status_code=400, detail="Exactly one of 'bridge' or 'interface' must be provided")
|
||||||
|
|
||||||
try:
|
try:
|
||||||
if req.interface:
|
if req.interface:
|
||||||
start_afpacket_sniffer(req.interface, target_is_interface=True)
|
session_id = start_afpacket_sniffer(req.interface, target_is_interface=True)
|
||||||
return SnifferStartResponse(started=True, target=req.interface, target_type="interface")
|
return SnifferStartResponse(started=True, session_id=session_id, target=req.interface, target_type="interface")
|
||||||
else:
|
else:
|
||||||
start_afpacket_sniffer(req.bridge, target_is_interface=False)
|
session_id = start_afpacket_sniffer(req.bridge, target_is_interface=False)
|
||||||
return SnifferStartResponse(started=True, target=req.bridge, target_type="bridge")
|
return SnifferStartResponse(started=True, session_id=session_id, target=req.bridge, target_type="bridge")
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
raise HTTPException(status_code=500, detail=f"Failed to start sniffer: {exc}")
|
raise HTTPException(status_code=500, detail=f"Failed to start sniffer: {exc}")
|
||||||
|
|
||||||
|
|
||||||
@router.post("/stop", response_model=SnifferStopResponse)
|
@router.post("/stop", response_model=SnifferStopResponse)
|
||||||
def sniffer_stop(
|
def sniffer_stop(
|
||||||
bridge: Optional[str] = Query(None, description="If provided, stop sniffer sockets for this bridge"),
|
q_bridge: Optional[str] = Query(None, alias="bridge", description="If provided, stop sniffer sockets for this bridge"),
|
||||||
interface: Optional[str] = Query(None, description="If provided, stop sniffer socket for this interface"),
|
q_interface: Optional[str] = Query(None, alias="interface", description="If provided, stop sniffer socket for this interface"),
|
||||||
|
body: SnifferStopRequest = Body(...),
|
||||||
):
|
):
|
||||||
"""
|
"""
|
||||||
Stop the AF_PACKET sniffer.
|
Stop sniffer sessions.
|
||||||
|
|
||||||
- If no query params provided: stop the global sniffer thread and clear snapshot.
|
- If body.session_id is provided: stop that session (preferred).
|
||||||
- If `?interface=eth0` provided: stop sniffing on that interface only (close socket).
|
- Else if query param `interface` provided: close socket for that interface across sessions.
|
||||||
- If `?bridge=br0` provided: stop snapshot/sockets associated with that bridge.
|
- Else if query param `bridge` provided: remove snapshot / close sockets for that bridge across sessions.
|
||||||
|
- Else: stop all sessions (global stop).
|
||||||
"""
|
"""
|
||||||
if bridge and interface:
|
if body and body.session_id:
|
||||||
|
try:
|
||||||
|
stop_afpacket_sniffer(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}")
|
||||||
|
|
||||||
|
# validate query params
|
||||||
|
if q_bridge and q_interface:
|
||||||
raise HTTPException(status_code=400, detail="Only one of 'bridge' or 'interface' may be provided")
|
raise HTTPException(status_code=400, detail="Only one of 'bridge' or 'interface' may be provided")
|
||||||
|
|
||||||
try:
|
try:
|
||||||
if interface:
|
if q_interface:
|
||||||
stop_afpacket_sniffer(target=interface, target_is_interface=True)
|
stop_afpacket_sniffer(target=q_interface, target_is_interface=True)
|
||||||
return SnifferStopResponse(stopped=True, target=interface, target_type="interface")
|
return SnifferStopResponse(stopped=True, session_id=None, target=q_interface, target_type="interface")
|
||||||
if bridge:
|
if q_bridge:
|
||||||
stop_afpacket_sniffer(target=bridge, target_is_interface=False)
|
stop_afpacket_sniffer(target=q_bridge, target_is_interface=False)
|
||||||
return SnifferStopResponse(stopped=True, target=bridge, target_type="bridge")
|
return SnifferStopResponse(stopped=True, session_id=None, target=q_bridge, target_type="bridge")
|
||||||
|
|
||||||
# global stop
|
# global stop
|
||||||
stop_afpacket_sniffer()
|
stop_afpacket_sniffer()
|
||||||
return SnifferStopResponse(stopped=True, target=None, target_type=None)
|
return SnifferStopResponse(stopped=True, session_id=None, target=None, target_type=None)
|
||||||
except HTTPException:
|
|
||||||
raise
|
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
raise HTTPException(status_code=500, detail=f"Failed to stop sniffer: {exc}")
|
raise HTTPException(status_code=500, detail=f"Failed to stop sniffer: {exc}")
|
||||||
|
|
||||||
|
|||||||
@@ -8,6 +8,7 @@ import selectors
|
|||||||
import errno
|
import errno
|
||||||
import struct
|
import struct
|
||||||
from typing import Dict, List, Optional, Any
|
from typing import Dict, List, Optional, Any
|
||||||
|
from uuid import uuid4
|
||||||
|
|
||||||
# NOTE: ensure this path points to your shared runtime module
|
# NOTE: ensure this path points to your shared runtime module
|
||||||
from src import shared_objects
|
from src import shared_objects
|
||||||
@@ -36,14 +37,18 @@ from src.Models.ip_protocol import IPProtocolEnum, protocol_from_number
|
|||||||
# ---- Logging ----------------------------------------------------------
|
# ---- Logging ----------------------------------------------------------
|
||||||
logger = logging.getLogger("af_packet_sniffer")
|
logger = logging.getLogger("af_packet_sniffer")
|
||||||
|
|
||||||
# ---- Globals (kept minimal) -------------------------------------------
|
# ---- Session model -------------------------------------------
|
||||||
af_sockets: Dict[str, socket.socket] = {}
|
# sessions: session_id -> session dict
|
||||||
af_thread: Optional[threading.Thread] = None
|
sessions: Dict[str, Dict[str, Any]] = {}
|
||||||
af_stop_event: Optional[threading.Event] = None
|
# session schema:
|
||||||
|
# {
|
||||||
fixed_bridge_ports: Dict[str, List[str]] = {}
|
# "thread": threading.Thread,
|
||||||
current_bridge: Optional[str] = None
|
# "stop_event": threading.Event,
|
||||||
current_bridge_label_for_thread: Optional[str] = None
|
# "sockets": Dict[iface, socket.socket],
|
||||||
|
# "label": str,
|
||||||
|
# "is_bridge": bool,
|
||||||
|
# "ports": List[str],
|
||||||
|
# }
|
||||||
|
|
||||||
# small bounded buffer for packets produced before shared_objects is ready
|
# small bounded buffer for packets produced before shared_objects is ready
|
||||||
_PACKET_BUFFER: List[Dict[str, Any]] = []
|
_PACKET_BUFFER: List[Dict[str, Any]] = []
|
||||||
@@ -135,26 +140,142 @@ def parse_packet(pkt, bridge_label: str) -> None:
|
|||||||
"dst_port": None,
|
"dst_port": None,
|
||||||
}
|
}
|
||||||
|
|
||||||
# (packet parsing unchanged — omitted here for brevity in this snippet)
|
# Ethernet layer
|
||||||
# --- copy the full packet parsing logic from your original file here ---
|
if Ether in pkt:
|
||||||
# For brevity in this repo patch, the original parse_packet body remains unchanged.
|
eth = pkt[Ether]
|
||||||
# Keep everything from "Ether in pkt" through the Raw fallback and DB scheduling exactly as you had.
|
pkt_info["src_mac"] = _safe_get_attr(eth, "src")
|
||||||
# (In your codebase, ensure the same body is present, unchanged.)
|
pkt_info["dst_mac"] = _safe_get_attr(eth, "dst")
|
||||||
# -----------------------------
|
|
||||||
|
# Base ethertype
|
||||||
|
try:
|
||||||
|
eth_type_raw = int(eth.type)
|
||||||
|
except Exception:
|
||||||
|
eth_type_raw = None
|
||||||
|
|
||||||
|
# VLAN inner ethertype and vlan id if Dot1Q exists
|
||||||
|
if Dot1Q in pkt:
|
||||||
|
try:
|
||||||
|
inner = int(pkt[Dot1Q].type)
|
||||||
|
if inner:
|
||||||
|
eth_type_raw = inner
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
try:
|
||||||
|
pkt_info["vlan_id"] = int(pkt[Dot1Q].vlan)
|
||||||
|
except Exception:
|
||||||
|
pkt_info["vlan_id"] = None
|
||||||
|
|
||||||
|
if eth_type_raw is not None:
|
||||||
|
pkt_info["eth_type_raw"] = eth_type_raw
|
||||||
|
try:
|
||||||
|
pkt_info["eth_type"] = ethertype_from_int(eth_type_raw)
|
||||||
|
except Exception:
|
||||||
|
pkt_info["eth_type"] = EtherTypeEnum.UNKNOWN
|
||||||
|
|
||||||
|
# ARP
|
||||||
|
if ARP in pkt:
|
||||||
|
arp = pkt[ARP]
|
||||||
|
pkt_info["protocol_name"] = "ARP"
|
||||||
|
pkt_info["src_ip"] = _safe_get_attr(arp, "psrc")
|
||||||
|
pkt_info["dst_ip"] = _safe_get_attr(arp, "pdst")
|
||||||
|
pkt_info["src_port"] = None
|
||||||
|
pkt_info["dst_port"] = None
|
||||||
|
|
||||||
|
# IPv4
|
||||||
|
if IP in pkt:
|
||||||
|
ip = pkt[IP]
|
||||||
|
pkt_info["src_ip"] = _safe_get_attr(ip, "src")
|
||||||
|
pkt_info["dst_ip"] = _safe_get_attr(ip, "dst")
|
||||||
|
|
||||||
|
try:
|
||||||
|
proto_num = int(_safe_get_attr(ip, "proto"))
|
||||||
|
except Exception:
|
||||||
|
proto_num = None
|
||||||
|
|
||||||
|
if proto_num is not None:
|
||||||
|
pkt_info["protocol_raw"] = proto_num
|
||||||
|
try:
|
||||||
|
pkt_info["protocol"] = protocol_from_number(proto_num)
|
||||||
|
except Exception:
|
||||||
|
pkt_info["protocol"] = IPProtocolEnum.UNKNOWN
|
||||||
|
|
||||||
|
if proto_num == 6 and TCP in pkt:
|
||||||
|
pkt_info["protocol_name"] = "TCP"
|
||||||
|
pkt_info["src_port"] = _safe_get_attr(pkt[TCP], "sport")
|
||||||
|
pkt_info["dst_port"] = _safe_get_attr(pkt[TCP], "dport")
|
||||||
|
elif proto_num == 17 and UDP in pkt:
|
||||||
|
pkt_info["protocol_name"] = "UDP"
|
||||||
|
pkt_info["src_port"] = _safe_get_attr(pkt[UDP], "sport")
|
||||||
|
pkt_info["dst_port"] = _safe_get_attr(pkt[UDP], "dport")
|
||||||
|
elif proto_num == 1 and ICMP in pkt:
|
||||||
|
pkt_info["protocol_name"] = "ICMP"
|
||||||
|
else:
|
||||||
|
if pkt_info["protocol_name"] is None:
|
||||||
|
pkt_info["protocol_name"] = f"IP_PROTO_{proto_num}" if proto_num is not None else None
|
||||||
|
|
||||||
|
# IPv6
|
||||||
|
if IPv6 in pkt:
|
||||||
|
ip6 = pkt[IPv6]
|
||||||
|
pkt_info["src_ip"] = _safe_get_attr(ip6, "src")
|
||||||
|
pkt_info["dst_ip"] = _safe_get_attr(ip6, "dst")
|
||||||
|
|
||||||
|
try:
|
||||||
|
nh = int(_safe_get_attr(ip6, "nh"))
|
||||||
|
except Exception:
|
||||||
|
nh = None
|
||||||
|
|
||||||
|
if nh is not None:
|
||||||
|
pkt_info["protocol_raw"] = nh
|
||||||
|
try:
|
||||||
|
pkt_info["protocol"] = protocol_from_number(nh)
|
||||||
|
except Exception:
|
||||||
|
pkt_info["protocol"] = IPProtocolEnum.UNKNOWN
|
||||||
|
|
||||||
|
if nh == 6 and TCP in pkt:
|
||||||
|
pkt_info["protocol_name"] = "TCP"
|
||||||
|
pkt_info["src_port"] = _safe_get_attr(pkt[TCP], "sport")
|
||||||
|
pkt_info["dst_port"] = _safe_get_attr(pkt[TCP], "dport")
|
||||||
|
elif nh == 17 and UDP in pkt:
|
||||||
|
pkt_info["protocol_name"] = "UDP"
|
||||||
|
pkt_info["src_port"] = _safe_get_attr(pkt[UDP], "sport")
|
||||||
|
pkt_info["dst_port"] = _safe_get_attr(pkt[UDP], "dport")
|
||||||
|
elif ICMPv6Unknown in pkt:
|
||||||
|
pkt_info["protocol_name"] = "ICMPv6"
|
||||||
|
else:
|
||||||
|
if pkt_info["protocol_name"] is None:
|
||||||
|
pkt_info["protocol_name"] = f"IPV6_PROTO_{nh}" if nh is not None else None
|
||||||
|
|
||||||
|
# Raw fallback label
|
||||||
|
if Raw in pkt and not pkt_info["protocol_name"]:
|
||||||
|
pkt_info["protocol_name"] = "RAW"
|
||||||
|
|
||||||
# Submit DB insert to shared web loop if available, otherwise buffer
|
# Submit DB insert to shared web loop if available, otherwise buffer
|
||||||
try:
|
try:
|
||||||
web_loop = getattr(shared_objects, "web_loop", None)
|
web_loop = getattr(shared_objects, "web_loop", None)
|
||||||
web_db = getattr(shared_objects, "db", None)
|
web_db = getattr(shared_objects, "db", None)
|
||||||
if web_db is not None and web_loop is not None:
|
if web_db is not None and web_loop is not None:
|
||||||
asyncio.run_coroutine_threadsafe(web_db.insert_packet(pkt_info), web_loop)
|
try:
|
||||||
else:
|
fut = asyncio.run_coroutine_threadsafe(web_db.insert_packet(pkt_info), web_loop)
|
||||||
# buffer (bounded) until the web app initializes
|
# best-effort non-blocking check for immediate errors
|
||||||
|
try:
|
||||||
|
fut.result(timeout=0.005)
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
logger.debug("Scheduled insert for packet on %s (len=%d)", pkt_info.get("iface"), pkt_info.get("length"))
|
||||||
|
except Exception as e:
|
||||||
|
logger.exception("Failed to schedule insert_packet for %s — buffering: %s", pkt_info.get("iface"), e)
|
||||||
_PACKET_BUFFER.append(pkt_info)
|
_PACKET_BUFFER.append(pkt_info)
|
||||||
if len(_PACKET_BUFFER) > _BUFFER_CAPACITY:
|
if len(_PACKET_BUFFER) > _BUFFER_CAPACITY:
|
||||||
# drop oldest packet
|
|
||||||
_PACKET_BUFFER.pop(0)
|
_PACKET_BUFFER.pop(0)
|
||||||
|
else:
|
||||||
|
_PACKET_BUFFER.append(pkt_info)
|
||||||
|
if len(_PACKET_BUFFER) == 1:
|
||||||
|
logger.info("Buffering packets; web DB not ready yet (first buffered packet on %s)", pkt_info.get("iface"))
|
||||||
|
if len(_PACKET_BUFFER) > _BUFFER_CAPACITY:
|
||||||
|
_PACKET_BUFFER.pop(0)
|
||||||
|
logger.warning("Packet buffer full: dropped oldest packet")
|
||||||
except Exception:
|
except Exception:
|
||||||
logger.exception("Failed to schedule DB insert")
|
logger.exception("Failed to schedule DB insert (unexpected)")
|
||||||
|
|
||||||
|
|
||||||
# -------------------------
|
# -------------------------
|
||||||
@@ -197,11 +318,12 @@ def _create_af_packet_socket(ifname: str, rx_buf: int = 4 * 1024 * 1024) -> Opti
|
|||||||
return None
|
return None
|
||||||
|
|
||||||
s.setblocking(False)
|
s.setblocking(False)
|
||||||
|
logger.info("AF_PACKET socket bound to %s (fd=%s)", ifname, getattr(s, "fileno", lambda: "?")())
|
||||||
return s
|
return s
|
||||||
|
|
||||||
|
|
||||||
def _close_socket(ifname: str) -> None:
|
def _close_socket_in_dict(sockets: Dict[str, socket.socket], ifname: str) -> None:
|
||||||
s = af_sockets.pop(ifname, None)
|
s = sockets.pop(ifname, None)
|
||||||
if s:
|
if s:
|
||||||
try:
|
try:
|
||||||
s.close()
|
s.close()
|
||||||
@@ -209,52 +331,52 @@ def _close_socket(ifname: str) -> None:
|
|||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
def _ensure_socket_for_iface(iface: str, bridge_label: str) -> None:
|
def _ensure_socket_for_session(sockets: Dict[str, socket.socket], iface: str, bridge_label: str) -> None:
|
||||||
"""
|
"""
|
||||||
Ensure we have an AF_PACKET socket in af_sockets for iface.
|
Ensure we have an AF_PACKET socket in the given sockets dict for iface.
|
||||||
If socket doesn't exist, attempt to create it. bridge_label is used for logging/context.
|
If socket doesn't exist, attempt to create it. bridge_label is used for logging/context.
|
||||||
"""
|
"""
|
||||||
if iface in af_sockets:
|
if iface in sockets:
|
||||||
return
|
return
|
||||||
if not check_interface_exists(iface):
|
if not check_interface_exists(iface):
|
||||||
logger.warning("Interface %s does not exist; skipping socket creation", iface)
|
logger.warning("Interface %s does not exist; skipping socket creation", iface)
|
||||||
return
|
return
|
||||||
s = _create_af_packet_socket(iface)
|
s = _create_af_packet_socket(iface)
|
||||||
if s:
|
if s:
|
||||||
af_sockets[iface] = s
|
sockets[iface] = s
|
||||||
logger.info("AF_PACKET socket created for %s (label=%s)", iface, bridge_label)
|
logger.info("AF_PACKET socket created for %s (label=%s)", iface, bridge_label)
|
||||||
else:
|
else:
|
||||||
logger.warning("Failed to create AF_PACKET socket for %s (label=%s)", iface, bridge_label)
|
logger.warning("Failed to create AF_PACKET socket for %s (label=%s)", iface, bridge_label)
|
||||||
|
|
||||||
|
|
||||||
# -------------------------
|
# -------------------------
|
||||||
# Main reader thread
|
# Per-session reader loop
|
||||||
# -------------------------
|
# -------------------------
|
||||||
def afpacket_reader_loop(bridge_label: str, stop_event: threading.Event) -> None:
|
def _session_reader_loop(session_id: str) -> None:
|
||||||
"""
|
session = sessions.get(session_id)
|
||||||
Multiplex AF_PACKET sockets using a selector and hand packets to parse_packet.
|
if not session:
|
||||||
Uses the fixed_bridge_ports snapshot to decide which interfaces to open when bridge-mode.
|
logger.warning("Session %s not found at reader start", session_id)
|
||||||
If started in interface-mode, fixed_bridge_ports may contain a single entry mapping the
|
return
|
||||||
interface name to itself (or None), but af_sockets drives which sockets are active.
|
stop_event: threading.Event = session["stop_event"]
|
||||||
"""
|
sockets: Dict[str, socket.socket] = session["sockets"]
|
||||||
logger.info("AF_PACKET reader starting (label=%s)", bridge_label)
|
label: str = session["label"]
|
||||||
|
|
||||||
|
logger.info("Session %s reader starting (label=%s)", session_id, label)
|
||||||
sel = selectors.DefaultSelector()
|
sel = selectors.DefaultSelector()
|
||||||
|
|
||||||
# If there is a snapshot for this label, create sockets for snapshot ports.
|
|
||||||
for iface in fixed_bridge_ports.get(bridge_label, []):
|
|
||||||
_ensure_socket_for_iface(iface, bridge_label)
|
|
||||||
|
|
||||||
# register existing sockets
|
# register existing sockets
|
||||||
for iface, s in list(af_sockets.items()):
|
for iface, s in list(sockets.items()):
|
||||||
try:
|
try:
|
||||||
sel.register(s, selectors.EVENT_READ, data=iface)
|
sel.register(s, selectors.EVENT_READ, data=iface)
|
||||||
except Exception:
|
except Exception:
|
||||||
logger.debug("Register failed for %s (continuing)", iface)
|
logger.debug("Register failed for %s in session %s (continuing)", iface, session_id)
|
||||||
|
|
||||||
|
last_drain_ts = time.time()
|
||||||
|
|
||||||
try:
|
try:
|
||||||
while not stop_event.is_set():
|
while not stop_event.is_set():
|
||||||
# ensure newly created sockets are registered
|
# ensure newly created sockets are registered
|
||||||
for iface, s in list(af_sockets.items()):
|
for iface, s in list(sockets.items()):
|
||||||
try:
|
try:
|
||||||
if not any(k.fileobj is s for k in sel.get_map().values()):
|
if not any(k.fileobj is s for k in sel.get_map().values()):
|
||||||
sel.register(s, selectors.EVENT_READ, data=iface)
|
sel.register(s, selectors.EVENT_READ, data=iface)
|
||||||
@@ -264,11 +386,18 @@ def afpacket_reader_loop(bridge_label: str, stop_event: threading.Event) -> None
|
|||||||
try:
|
try:
|
||||||
events = sel.select(timeout=1.0)
|
events = sel.select(timeout=1.0)
|
||||||
except Exception:
|
except Exception:
|
||||||
logger.exception("Selector error")
|
logger.exception("Selector error in session %s", session_id)
|
||||||
time.sleep(0.1)
|
time.sleep(0.1)
|
||||||
continue
|
continue
|
||||||
|
|
||||||
if not events:
|
if not events:
|
||||||
|
# occasionally attempt to drain buffer if DB becomes available
|
||||||
|
if time.time() - last_drain_ts > 5:
|
||||||
|
try:
|
||||||
|
drain_buffer_to_shared_db()
|
||||||
|
except Exception:
|
||||||
|
logger.exception("Periodic drain failed in session %s", session_id)
|
||||||
|
last_drain_ts = time.time()
|
||||||
continue
|
continue
|
||||||
|
|
||||||
for key, _ in events:
|
for key, _ in events:
|
||||||
@@ -282,28 +411,29 @@ def afpacket_reader_loop(bridge_label: str, stop_event: threading.Event) -> None
|
|||||||
continue
|
continue
|
||||||
except OSError as e:
|
except OSError as e:
|
||||||
if e.errno in (errno.ENODEV, errno.ENETDOWN, errno.EBADF):
|
if e.errno in (errno.ENODEV, errno.ENETDOWN, errno.EBADF):
|
||||||
logger.warning("Socket error on %s: %s — closing", iface, e)
|
logger.warning("Socket error on %s in session %s: %s — closing", iface, session_id, e)
|
||||||
try:
|
try:
|
||||||
sel.unregister(sock)
|
sel.unregister(sock)
|
||||||
except Exception:
|
except Exception:
|
||||||
pass
|
pass
|
||||||
_close_socket(iface)
|
_close_socket_in_dict(sockets, iface)
|
||||||
continue
|
continue
|
||||||
else:
|
else:
|
||||||
logger.exception("Recv error on %s", iface)
|
logger.exception("Recv error on %s in session %s", iface, session_id)
|
||||||
continue
|
continue
|
||||||
|
|
||||||
# parse with scapy
|
# parse with scapy
|
||||||
try:
|
try:
|
||||||
pkt = Ether(raw)
|
pkt = Ether(raw)
|
||||||
pkt.sniffed_on = iface
|
pkt.sniffed_on = iface
|
||||||
parse_packet(pkt, bridge_label)
|
parse_packet(pkt, label)
|
||||||
|
logger.debug("Captured packet on %s in session %s (len=%d)", iface, session_id, len(raw))
|
||||||
except Exception:
|
except Exception:
|
||||||
logger.exception("Failed to parse/process packet from %s", iface)
|
logger.exception("Failed to parse/process packet from %s in session %s", iface, session_id)
|
||||||
continue
|
continue
|
||||||
finally:
|
finally:
|
||||||
logger.info("AF_PACKET reader stopping; cleaning up sockets")
|
logger.info("Session %s reader stopping; cleaning up sockets", session_id)
|
||||||
# unregister and close
|
# unregister and close sockets
|
||||||
try:
|
try:
|
||||||
for key in list(sel.get_map().values()):
|
for key in list(sel.get_map().values()):
|
||||||
try:
|
try:
|
||||||
@@ -313,123 +443,106 @@ def afpacket_reader_loop(bridge_label: str, stop_event: threading.Event) -> None
|
|||||||
except Exception:
|
except Exception:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
for iface in list(af_sockets.keys()):
|
for iface in list(sockets.keys()):
|
||||||
_close_socket(iface)
|
_close_socket_in_dict(sockets, iface)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
sel.close()
|
sel.close()
|
||||||
except Exception:
|
except Exception:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
logger.info("AF_PACKET reader stopped")
|
logger.info("Session %s reader stopped", session_id)
|
||||||
|
|
||||||
|
|
||||||
# -------------------------
|
# -------------------------
|
||||||
# Public API: start/stop/status
|
# Public API: start/stop/status
|
||||||
# -------------------------
|
# -------------------------
|
||||||
def start_afpacket_sniffer(target: str, target_is_interface: bool = False) -> None:
|
def start_afpacket_sniffer(target: str, target_is_interface: bool = False) -> str:
|
||||||
"""
|
"""
|
||||||
Start the sniffer.
|
Start a sniffer session. Returns session_id string.
|
||||||
|
|
||||||
If target_is_interface == False: target is treated as bridge name; snapshot ports and start reader
|
If target_is_interface == False: target is treated as bridge name and ports are snapshotted.
|
||||||
for that bridge (existing behavior).
|
If target_is_interface == True: target is treated as interface name.
|
||||||
If target_is_interface == True: target is treated as interface name; create socket for interface
|
|
||||||
and start reader if not running.
|
|
||||||
"""
|
"""
|
||||||
global af_thread, af_stop_event, current_bridge, current_bridge_label_for_thread
|
session_id = str(uuid4())
|
||||||
if af_thread and af_thread.is_alive():
|
session: Dict[str, Any] = {
|
||||||
logger.info("Sniffer thread already running — adding sockets if needed")
|
"stop_event": threading.Event(),
|
||||||
|
"sockets": {},
|
||||||
|
"label": target,
|
||||||
|
"is_bridge": not target_is_interface,
|
||||||
|
"ports": [],
|
||||||
|
}
|
||||||
|
sessions[session_id] = session
|
||||||
|
|
||||||
# If thread already running and target is interface -> try to add socket only
|
# determine ports for this session
|
||||||
if target_is_interface:
|
if target_is_interface:
|
||||||
_ensure_socket_for_iface(target, target)
|
ports = [target]
|
||||||
else:
|
|
||||||
# bridge-mode: snapshot ports and ensure sockets for those ports
|
|
||||||
ports = get_bridge_ports_once(target)
|
|
||||||
fixed_bridge_ports[target] = ports
|
|
||||||
logger.info("Fixed bridge ports updated for %s: %s", target, ports)
|
|
||||||
for p in ports:
|
|
||||||
_ensure_socket_for_iface(p, target)
|
|
||||||
return
|
|
||||||
|
|
||||||
# Thread not running — set up snapshot or single interface and start thread
|
|
||||||
if target_is_interface:
|
|
||||||
# ensure socket for the single interface
|
|
||||||
_ensure_socket_for_iface(target, target)
|
|
||||||
current_bridge = target # use target as label
|
|
||||||
current_bridge_label_for_thread = target
|
|
||||||
fixed_bridge_ports[target] = [target] # optional snapshot entry
|
|
||||||
else:
|
else:
|
||||||
ports = get_bridge_ports_once(target)
|
ports = get_bridge_ports_once(target)
|
||||||
fixed_bridge_ports[target] = ports
|
|
||||||
current_bridge = target
|
session["ports"] = ports
|
||||||
current_bridge_label_for_thread = target
|
|
||||||
logger.info("Fixed bridge ports for %s: %s", target, ports)
|
# create sockets for this session only
|
||||||
# create sockets now
|
|
||||||
for iface in ports:
|
for iface in ports:
|
||||||
_ensure_socket_for_iface(iface, target)
|
if not check_interface_exists(iface):
|
||||||
|
logger.warning("Snapshot port %s missing for session %s, skipping", iface, session_id)
|
||||||
|
continue
|
||||||
|
s = _create_af_packet_socket(iface)
|
||||||
|
if s:
|
||||||
|
session["sockets"][iface] = s
|
||||||
|
|
||||||
af_stop_event = threading.Event()
|
# start dedicated thread
|
||||||
af_thread = threading.Thread(target=afpacket_reader_loop, args=(current_bridge_label_for_thread or target, af_stop_event), daemon=True)
|
t = threading.Thread(target=_session_reader_loop, args=(session_id,), daemon=True)
|
||||||
af_thread.start()
|
session["thread"] = t
|
||||||
logger.info("AF_PACKET sniffer started (label=%s)", current_bridge_label_for_thread or target)
|
t.start()
|
||||||
|
logger.info("Started sniffer session %s label=%s ports=%s", session_id, target, ports)
|
||||||
|
return session_id
|
||||||
|
|
||||||
|
|
||||||
def stop_afpacket_sniffer(target: Optional[str] = None, target_is_interface: bool = False) -> None:
|
def stop_afpacket_sniffer(session_id: Optional[str] = None, target: Optional[str] = None, target_is_interface: bool = False) -> None:
|
||||||
"""
|
"""
|
||||||
Stop the sniffer or a subset:
|
Stop either a specific session by session_id (preferred), or stop sockets associated with target.
|
||||||
|
- If session_id provided: stop that session and clean up.
|
||||||
- If no target: stop the whole reader thread and clear snapshot (existing behavior).
|
- Else if target provided: remove sockets associated with target from sessions (close sockets).
|
||||||
- If target_is_interface True: close the socket for that interface (do not stop thread).
|
- Else if no args: stop all sessions (global stop).
|
||||||
- If target provided and target_is_interface False: remove snapshot for the bridge and close sockets for ports in that snapshot.
|
|
||||||
"""
|
"""
|
||||||
global af_thread, af_stop_event, current_bridge, current_bridge_label_for_thread
|
# Stop by session_id
|
||||||
# Stop a single interface
|
if session_id:
|
||||||
if target is not None and target_is_interface:
|
s = sessions.pop(session_id, None)
|
||||||
_close_socket(target)
|
if not s:
|
||||||
# Also remove from any fixed_bridge_ports entries where present
|
logger.warning("Requested stop for unknown session %s", session_id)
|
||||||
for br, ports in list(fixed_bridge_ports.items()):
|
return
|
||||||
if target in ports:
|
s["stop_event"].set()
|
||||||
|
t = s.get("thread")
|
||||||
|
if t and isinstance(t, threading.Thread):
|
||||||
|
t.join(timeout=2)
|
||||||
|
logger.info("Stopped session %s", session_id)
|
||||||
|
return
|
||||||
|
|
||||||
|
# Stop by target (interface or bridge)
|
||||||
|
if target:
|
||||||
|
# For interface-mode: close that iface in any session that has it
|
||||||
|
for sid, s in list(sessions.items()):
|
||||||
|
if target in s.get("ports", []):
|
||||||
|
# close only that iface socket in that session
|
||||||
|
sock = s["sockets"].pop(target, None)
|
||||||
|
if sock:
|
||||||
try:
|
try:
|
||||||
fixed_bridge_ports[br].remove(target)
|
sock.close()
|
||||||
except Exception:
|
except Exception:
|
||||||
pass
|
pass
|
||||||
logger.info("Stopped sniffer on interface %s (socket closed)", target)
|
|
||||||
return
|
|
||||||
|
|
||||||
# Stop a bridge snapshot: close sockets for snapshot ports and remove snapshot
|
|
||||||
if target is not None and not target_is_interface:
|
|
||||||
ports = fixed_bridge_ports.pop(target, [])
|
|
||||||
for p in ports:
|
|
||||||
_close_socket(p)
|
|
||||||
# If we stopped the bridge that the running thread was for, also clear current_bridge
|
|
||||||
if current_bridge == target:
|
|
||||||
current_bridge = None
|
|
||||||
current_bridge_label_for_thread = None
|
|
||||||
logger.info("Stopped sniffer for bridge %s (closed ports: %s)", target, ports)
|
|
||||||
return
|
|
||||||
|
|
||||||
# Global stop (existing behavior)
|
|
||||||
if not af_thread:
|
|
||||||
return
|
|
||||||
|
|
||||||
if af_stop_event:
|
|
||||||
af_stop_event.set()
|
|
||||||
|
|
||||||
af_thread.join(timeout=2)
|
|
||||||
af_thread = None
|
|
||||||
af_stop_event = None
|
|
||||||
|
|
||||||
# clear all snapshots
|
|
||||||
try:
|
try:
|
||||||
for br in list(fixed_bridge_ports.keys()):
|
s["ports"].remove(target)
|
||||||
fixed_bridge_ports.pop(br, None)
|
|
||||||
except Exception:
|
except Exception:
|
||||||
pass
|
pass
|
||||||
current_bridge = None
|
logger.info("Removed target %s from session %s", target, sid)
|
||||||
current_bridge_label_for_thread = None
|
return
|
||||||
|
|
||||||
# close db pool if shared.web_loop is available; otherwise leave to main
|
# Global stop: stop all sessions
|
||||||
|
for sid in list(sessions.keys()):
|
||||||
|
stop_afpacket_sniffer(session_id=sid)
|
||||||
|
|
||||||
|
# attempt to close db pool if shared.web_loop is available; otherwise leave to main
|
||||||
try:
|
try:
|
||||||
web_loop = getattr(shared_objects, "web_loop", None)
|
web_loop = getattr(shared_objects, "web_loop", None)
|
||||||
web_db = getattr(shared_objects, "db", None)
|
web_db = getattr(shared_objects, "db", None)
|
||||||
@@ -438,19 +551,46 @@ def stop_afpacket_sniffer(target: Optional[str] = None, target_is_interface: boo
|
|||||||
except Exception:
|
except Exception:
|
||||||
logger.exception("Failed to schedule DB pool close")
|
logger.exception("Failed to schedule DB pool close")
|
||||||
|
|
||||||
logger.info("AF_PACKET sniffer stopped")
|
logger.info("All sniffer sessions stopped")
|
||||||
|
|
||||||
|
|
||||||
def get_sniffer_status() -> Dict[str, Dict[str, object]]:
|
def get_sniffer_status() -> Dict[str, Dict[str, object]]:
|
||||||
"""
|
"""
|
||||||
Return simple status per managed interface.
|
Return simple status per managed interface, including which session owns it.
|
||||||
|
Output format:
|
||||||
|
{
|
||||||
|
"<iface>": {"running": True, "exists": True, "up": True, "session_id": "<uuid>", "session_label": "..."}
|
||||||
|
}
|
||||||
"""
|
"""
|
||||||
running = af_thread.is_alive() if af_thread else False
|
|
||||||
out: Dict[str, Dict[str, object]] = {}
|
out: Dict[str, Dict[str, object]] = {}
|
||||||
for iface in list(af_sockets.keys()):
|
# build map iface -> session
|
||||||
|
for sid, s in sessions.items():
|
||||||
|
running = s.get("thread") is not None and s.get("thread").is_alive()
|
||||||
|
for iface in s.get("sockets", {}).keys():
|
||||||
out[iface] = {
|
out[iface] = {
|
||||||
"running": running,
|
"running": running,
|
||||||
"exists": check_interface_exists(iface),
|
"exists": check_interface_exists(iface),
|
||||||
"up": check_interface_up(iface),
|
"up": check_interface_up(iface),
|
||||||
|
"session_id": sid,
|
||||||
|
"session_label": s.get("label"),
|
||||||
}
|
}
|
||||||
return out
|
return out
|
||||||
|
|
||||||
|
|
||||||
|
def get_internal_debug_state() -> dict:
|
||||||
|
"""
|
||||||
|
Return internal debug information.
|
||||||
|
"""
|
||||||
|
return {
|
||||||
|
"sessions": {
|
||||||
|
sid: {
|
||||||
|
"label": s.get("label"),
|
||||||
|
"is_bridge": s.get("is_bridge"),
|
||||||
|
"ports": list(s.get("ports", [])),
|
||||||
|
"sockets": list(s.get("sockets", {}).keys()),
|
||||||
|
"thread_alive": bool(s.get("thread") and s.get("thread").is_alive()),
|
||||||
|
}
|
||||||
|
for sid, s in sessions.items()
|
||||||
|
},
|
||||||
|
"buffer_len": len(_PACKET_BUFFER),
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user