From 98865a965c7fa75ee0f3da042adb98d43c6b635d Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 26 May 2026 21:39:48 +0000 Subject: [PATCH 1/2] Add realtime session with RT-kernel mode --- doc/scapy/usage.rst | 33 ++++- scapy/sendrecv.py | 290 ++++++++++++++++++++++++++++++++++++++++++++ test/regression.uts | 95 +++++++++++++++ 3 files changed, 417 insertions(+), 1 deletion(-) diff --git a/doc/scapy/usage.rst b/doc/scapy/usage.rst index 1795d1058e2..da397a8e019 100644 --- a/doc/scapy/usage.rst +++ b/doc/scapy/usage.rst @@ -782,6 +782,38 @@ Another example: using ``prn`` and ``store=False`` >>> time.sleep(20) >>> t.stop() +Real-time session mode +---------------------- + +For interactive send/receive workflows, Scapy provides a context manager that +keeps a socket open across sends and receives: + +.. code-block:: python + + >>> with realtime_session(iface="eth0") as rt: + ... rt.send(Ether()/IP(dst="192.0.2.1")/ICMP()) + ... ans = rt.recv(timeout=1) + +To reduce scheduler jitter on real-time kernels, enable ``rt_kernel=True``. +In that mode, Scapy uses a single-threaded blocking receive loop on one socket. +In default mode, reception is asynchronous through a background thread. + +Optional tuning parameters are available: + +- ``socket_rcvbuf`` / ``socket_sndbuf`` to request socket buffer sizes +- ``poll_interval`` to tune receive polling granularity +- ``queue_size`` with ``strict_drop=True`` to keep freshest packets under load + +Scapy does **not** change process scheduling policy directly. Use +``setup_callback`` / ``teardown_callback`` to apply environment-specific process +settings (for instance CPU affinity or RT policy) in your own code. + +.. warning:: + Real-time scheduler policies typically require elevated privileges. + Incorrect settings can starve other system workloads. Prefer conservative + priorities and validate behavior under realistic traffic. Tight latency + targets often require dropping packets under load. + Advanced Sniffing - Sniffing Sessions ------------------------------------- @@ -1901,4 +1933,3 @@ For example to guess OS from a single captured packet: - diff --git a/scapy/sendrecv.py b/scapy/sendrecv.py index 2ab413eb2c7..73e27ad629b 100644 --- a/scapy/sendrecv.py +++ b/scapy/sendrecv.py @@ -8,6 +8,7 @@ """ import itertools +from collections import deque from threading import Thread, Event import os import re @@ -43,6 +44,7 @@ from typing import ( Any, Callable, + Deque, Dict, Iterator, List, @@ -104,6 +106,294 @@ class debug: _GlobSessionType = Union[Type[DefaultSession], DefaultSession] +class RealtimeSession(object): + """ + Context-managed real-time send/receive session. + + In default mode, packets are received asynchronously in a background thread. + With ``rt_kernel=True``, reception uses a single-threaded blocking loop to + reduce scheduler jitter on real-time kernels. + """ + + def __init__(self, + socket=None, # type: Optional[SuperSocket] + iface=None, # type: Optional[_GlobInterfaceType] + L2socket=None, # type: Optional[Type[SuperSocket]] + promisc=None, # type: Optional[bool] + filter=None, # type: Optional[str] + nofilter=0, # type: int + type=ETH_P_ALL, # type: int + session=None, # type: Optional[_GlobSessionType] + lfilter=None, # type: Optional[Callable[[Packet], bool]] + rt_kernel=False, # type: bool + poll_interval=None, # type: Optional[float] + queue_size=0, # type: int + strict_drop=False, # type: bool + socket_rcvbuf=None, # type: Optional[int] + socket_sndbuf=None, # type: Optional[int] + setup_callback=None, # type: Optional[Callable[["RealtimeSession"], Any]] # noqa: E501 + teardown_callback=None, # type: Optional[Callable[["RealtimeSession"], Any]] # noqa: E501 + **socket_kwargs # type: Any + ): + # type: (...) -> None + self.socket = socket + self._owns_socket = socket is None + self.iface = iface + self.L2socket = L2socket + self.promisc = promisc + self.filter = filter + self.nofilter = nofilter + self.type = type + self.session = session + self.lfilter = lfilter + self.rt_kernel = rt_kernel + self.poll_interval = poll_interval + self.queue_size = max(0, queue_size) + self.strict_drop = strict_drop + self.socket_rcvbuf = socket_rcvbuf + self.socket_sndbuf = socket_sndbuf + self.setup_callback = setup_callback + self.teardown_callback = teardown_callback + self.socket_kwargs = socket_kwargs + + self.running = False + self.closed = False + self._socket_label = None # type: Optional[_GlobInterfaceType] + self._session = None # type: Optional[DefaultSession] + self._pending = deque() # type: Deque[Packet] + self._queue = deque() # type: Deque[Packet] + self._queue_event = Event() + self._stop_event = Event() + self._rx_thread = None # type: Optional[Thread] + self._thread_exception = None # type: Optional[Exception] + + def __enter__(self): + # type: () -> "RealtimeSession" + if self.running: + return self + if self.socket is None: + iface = resolve_iface(self.iface or conf.iface) + l2socket = self.L2socket or iface.l2socket() + self.socket = l2socket( + promisc=self.promisc, + filter=self.filter, + iface=iface, + nofilter=self.nofilter, + type=self.type, + **self.socket_kwargs + ) + self._socket_label = iface + else: + self._socket_label = self.iface + self._configure_socket_buffers() + if not isinstance(self.session, DefaultSession): + klass = self.session or DefaultSession + self._session = klass() # type: ignore + else: + self._session = self.session + self.running = True + self.closed = False + if self.setup_callback is not None: + self.setup_callback(self) + if not self.rt_kernel: + self._start_rx_thread() + return self + + def __exit__(self, exc_type, exc_value, traceback): + # type: (Optional[Type[BaseException]], Optional[BaseException], Optional[Any]) -> None # noqa: E501 + self.close() + if exc_type is None and self._thread_exception is not None: + raise self._thread_exception + + def _configure_socket_buffers(self): + # type: () -> None + if not self.socket: + return + socks = [getattr(self.socket, "ins", None), getattr(self.socket, "outs", None)] # noqa: E501 + for sock in socks: + if sock is None or not hasattr(sock, "setsockopt"): + continue + try: + if self.socket_rcvbuf is not None: + sock.setsockopt( + socket.SOL_SOCKET, + socket.SO_RCVBUF, + self.socket_rcvbuf + ) + if self.socket_sndbuf is not None: + sock.setsockopt( + socket.SOL_SOCKET, + socket.SO_SNDBUF, + self.socket_sndbuf + ) + except OSError: + pass + + def _start_rx_thread(self): + # type: () -> None + self._rx_thread = Thread( + target=self._rx_loop, + name="RealtimeSession", + daemon=True + ) + self._rx_thread.start() + + def _enqueue(self, pkt): + # type: (Packet) -> None + if self.queue_size and len(self._queue) >= self.queue_size: + if self.strict_drop: + self._queue.popleft() + else: + return + self._queue.append(pkt) + self._queue_event.set() + + def _decorate(self, pkt): + # type: (Packet) -> Packet + if getattr(pkt, "sniffed_on", None) is None: + pkt.sniffed_on = self._socket_label + if getattr(pkt, "time", None) is None: + pkt.time = time.time() + return pkt + + def _iter_packets(self): + # type: () -> Iterator[Packet] + if not self.socket or not self._session: + return + packets = self._session.recv(self.socket) + for pkt in packets: + if self.lfilter and not self.lfilter(pkt): + continue + yield self._decorate(pkt) + + def _rx_loop(self): + # type: () -> None + while not self._stop_event.is_set(): + if not self.socket: + break + remain = self.poll_interval + if remain is None: + remain = conf.recv_poll_rate + sockets = self.socket.select([self.socket], remain) + if not sockets: + continue + try: + for pkt in self._iter_packets(): + self._enqueue(pkt) + except EOFError: + break + except Exception as ex: + self._thread_exception = ex + break + self._stop_event.set() + self._queue_event.set() + + def send(self, pkt): + # type: (Packet) -> int + if not self.socket: + raise Scapy_Exception("RealtimeSession is not started") + return self.socket.send(pkt) + + def _recv_rt(self, timeout=None): + # type: (Optional[float]) -> Optional[Packet] + if self._pending: + return self._pending.popleft() + if not self.socket: + return None + deadline = None + if timeout is not None: + deadline = time.monotonic() + timeout + while True: + remain = None + if deadline is not None: + remain = deadline - time.monotonic() + if remain <= 0: + return None + sockets = self.socket.select([self.socket], remain) + if not sockets: + return None + try: + found = False + for pkt in self._iter_packets(): + if not found: + found = True + first = pkt + else: + self._pending.append(pkt) + if found: + return first + except EOFError: + return None + + def _recv_bg(self, timeout=None): + # type: (Optional[float]) -> Optional[Packet] + deadline = None + if timeout is not None: + deadline = time.monotonic() + timeout + while True: + if self._queue: + pkt = self._queue.popleft() + if not self._queue: + self._queue_event.clear() + return pkt + if self._thread_exception is not None: + raise self._thread_exception + if self._stop_event.is_set(): + return None + wait_time = self.poll_interval + if wait_time is None: + wait_time = conf.recv_poll_rate + if deadline is not None: + remain = deadline - time.monotonic() + if remain <= 0: + return None + wait_time = min(wait_time, remain) + self._queue_event.wait(wait_time) + + def recv(self, timeout=None): + # type: (Optional[float]) -> Optional[Packet] + """ + Receive one packet from the session. + """ + if self.rt_kernel: + return self._recv_rt(timeout=timeout) + return self._recv_bg(timeout=timeout) + + def sendrecv(self, pkt, timeout=None): + # type: (Packet, Optional[float]) -> Optional[Packet] + """ + Send one packet then receive one packet from the same session socket. + """ + self.send(pkt) + return self.recv(timeout=timeout) + + def close(self): + # type: () -> None + if self.closed: + return + self.closed = True + self.running = False + self._stop_event.set() + self._queue_event.set() + if self._rx_thread is not None: + self._rx_thread.join() + self._rx_thread = None + self._queue.clear() + self._pending.clear() + if self.socket and self._owns_socket: + self.socket.close() + if self.teardown_callback is not None: + self.teardown_callback(self) + + +def realtime_session(*args, **kwargs): + # type: (*Any, **Any) -> RealtimeSession + """ + Return a context-managed real-time packet session. + """ + return RealtimeSession(*args, **kwargs) + + class SndRcvHandler(object): """ Util to send/receive packets, used by sr*(). diff --git a/test/regression.uts b/test/regression.uts index 8f0592b486d..c00aaa2c464 100644 --- a/test/regression.uts +++ b/test/regression.uts @@ -2058,6 +2058,101 @@ o.send(REFPACKET) pkts = sniff(opened_socket=[o], timeout=3) assert len(pkts) == 10 += Realtime session: RT kernel timeout and timestamps + +import time +from scapy.data import MTU +from scapy.automaton import select_objects +from scapy.supersocket import SuperSocket + +exec("""class RTTestSocket(SuperSocket): + def __init__(self, basecls=Ether): + self.basecls = basecls + self.ins = ObjectPipe(name="RTTestSocket") + self.outs = None + self.paired_sockets = [] + self._ts = 1000.0 + + def pair(self, sock): + self.paired_sockets.append(sock) + sock.paired_sockets.append(self) + + def send(self, x): + sx = bytes(x) + for r in self.paired_sockets: + r.ins.send(sx) + x.sent_time = time.time() + return len(sx) + + def recv_raw(self, x=MTU): + self._ts += 0.001 + return self.basecls, self.ins.recv(0), self._ts + + @staticmethod + def select(sockets, remain=conf.recv_poll_rate): + return select_objects(sockets, remain) +""") + + +rx = RTTestSocket(Ether) +tx = RTTestSocket(Ether) +rx.pair(tx) +pkt1 = Ether()/IP()/UDP(dport=1111) +pkt2 = Ether()/IP()/UDP(dport=2222) +tx.send(pkt1) +tx.send(pkt2) + +with realtime_session(socket=rx, rt_kernel=True) as rt: + p1 = rt.recv(timeout=0.5) + p2 = rt.recv(timeout=0.5) + assert p1 is not None and p2 is not None + assert p1.time <= p2.time + assert p1[UDP].dport == 1111 + assert p2[UDP].dport == 2222 + +rx.close() +tx.close() + += Realtime session: timeout and thread shutdown + +s1 = RTTestSocket(Ether) +s2 = RTTestSocket(Ether) +s1.pair(s2) + +with realtime_session(socket=s1, rt_kernel=True, poll_interval=0.01) as rt: + start = time.monotonic() + p = rt.recv(timeout=0.05) + elapsed = time.monotonic() - start + assert p is None + assert elapsed >= 0.03 + assert elapsed < 0.25 + +with realtime_session(socket=s1, rt_kernel=False, poll_interval=0.01) as rt: + assert rt._rx_thread is not None + assert rt._rx_thread.is_alive() + +assert rt._rx_thread is None or not rt._rx_thread.is_alive() +s1.close() +s2.close() + += Realtime session: strict drop keeps freshest packet + +rxd = RTTestSocket(Ether) +txd = RTTestSocket(Ether) +rxd.pair(txd) + +with realtime_session(socket=rxd, rt_kernel=False, poll_interval=0.01, + queue_size=1, strict_drop=True) as rt: + txd.send(Ether()/IP()/UDP(dport=3333)) + txd.send(Ether()/IP()/UDP(dport=4444)) + time.sleep(0.1) + p = rt.recv(timeout=0.5) + assert p is not None + assert p[UDP].dport == 4444 + +rxd.close() +txd.close() + = GH issue 3306 ~ netaccess needs_root From 92d7b48d0107e6a7b626447ded60762fef167d4f Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 26 May 2026 21:45:07 +0000 Subject: [PATCH 2/2] Refine realtime session behavior and tests --- scapy/sendrecv.py | 17 ++++++++++------- test/regression.uts | 22 ++++++++++++++++++++++ 2 files changed, 32 insertions(+), 7 deletions(-) diff --git a/scapy/sendrecv.py b/scapy/sendrecv.py index 73e27ad629b..e17e4b201df 100644 --- a/scapy/sendrecv.py +++ b/scapy/sendrecv.py @@ -238,6 +238,12 @@ def _start_rx_thread(self): ) self._rx_thread.start() + def _get_poll_interval(self): + # type: () -> float + if self.poll_interval is None: + return conf.recv_poll_rate + return self.poll_interval + def _enqueue(self, pkt): # type: (Packet) -> None if self.queue_size and len(self._queue) >= self.queue_size: @@ -271,9 +277,7 @@ def _rx_loop(self): while not self._stop_event.is_set(): if not self.socket: break - remain = self.poll_interval - if remain is None: - remain = conf.recv_poll_rate + remain = self._get_poll_interval() sockets = self.socket.select([self.socket], remain) if not sockets: continue @@ -290,7 +294,7 @@ def _rx_loop(self): def send(self, pkt): # type: (Packet) -> int - if not self.socket: + if not self.running or not self.socket: raise Scapy_Exception("RealtimeSession is not started") return self.socket.send(pkt) @@ -314,6 +318,7 @@ def _recv_rt(self, timeout=None): return None try: found = False + first = None # type: Optional[Packet] for pkt in self._iter_packets(): if not found: found = True @@ -340,9 +345,7 @@ def _recv_bg(self, timeout=None): raise self._thread_exception if self._stop_event.is_set(): return None - wait_time = self.poll_interval - if wait_time is None: - wait_time = conf.recv_poll_rate + wait_time = self._get_poll_interval() if deadline is not None: remain = deadline - time.monotonic() if remain <= 0: diff --git a/test/regression.uts b/test/regression.uts index c00aaa2c464..90a28804c11 100644 --- a/test/regression.uts +++ b/test/regression.uts @@ -2153,6 +2153,28 @@ with realtime_session(socket=rxd, rt_kernel=False, poll_interval=0.01, rxd.close() txd.close() += Realtime session: send lifecycle and sendrecv + +sa = RTTestSocket(Ether) +sb = RTTestSocket(Ether) +sa.pair(sb) + +rt = realtime_session(socket=sa, rt_kernel=True) +try: + rt.send(Ether()/IP()/UDP()) + assert False, "send() before __enter__ should fail" +except Scapy_Exception: + assert True + +with rt: + sb.send(Ether()/IP()/UDP(dport=5555)) + ans = rt.sendrecv(Ether()/IP()/UDP(dport=6666), timeout=0.5) + assert ans is not None + assert ans[UDP].dport == 5555 + +sa.close() +sb.close() + = GH issue 3306 ~ netaccess needs_root