Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/scapy/sessions.py: 14%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# SPDX-License-Identifier: GPL-2.0-only
2# This file is part of Scapy
3# See https://scapy.net/ for more information
5"""
6Sessions: decode flow of packets when sniffing
7"""
9from collections import defaultdict
10import socket
11import struct
13from scapy.config import conf
14from scapy.data import MTU
15from scapy.error import log_runtime
16from scapy.packet import Packet
17from scapy.pton_ntop import inet_pton
19# Typing imports
20from typing import (
21 Any,
22 Callable,
23 DefaultDict,
24 Dict,
25 Iterator,
26 List,
27 Optional,
28 Tuple,
29 Type,
30 cast,
31 TYPE_CHECKING,
32)
33from scapy.compat import Self
34if TYPE_CHECKING:
35 from scapy.supersocket import SuperSocket
38class DefaultSession(object):
39 """Default session: no stream decoding"""
41 def __init__(self, supersession: Optional[Self] = None):
42 if supersession and not isinstance(supersession, DefaultSession):
43 supersession = supersession()
44 self.supersession = supersession
46 def process(self, pkt: Packet) -> Optional[Packet]:
47 """
48 Called to pre-process the packet
49 """
50 # Optionally handle supersession
51 if self.supersession:
52 return self.supersession.process(pkt)
53 return pkt
55 def recv(self, sock: 'SuperSocket') -> Iterator[Packet]:
56 """
57 Will be called by sniff() to ask for a packet
58 """
59 pkt = sock.recv()
60 if not pkt:
61 return
62 try:
63 pkt = self.process(pkt)
64 except Exception as ex:
65 log_runtime.warning(
66 "%s processing failed with '%s'. Skipping." % (
67 type(self),
68 ex,
69 )
70 )
71 if conf.debug_dissector:
72 raise
73 if pkt is not None:
74 yield pkt
75 return
76 if pkt:
77 yield pkt
80class IPSession(DefaultSession):
81 """Defragment IP packets 'on-the-flow'.
83 Usage:
84 >>> sniff(session=IPSession)
85 """
87 def __init__(self, *args, **kwargs):
88 # type: (*Any, **Any) -> None
89 DefaultSession.__init__(self, *args, **kwargs)
90 self.fragments = defaultdict(list) # type: DefaultDict[Tuple[Any, ...], List[Packet]] # noqa: E501
92 def process(self, packet: Packet) -> Optional[Packet]:
93 from scapy.layers.inet import BadFragments, IP, _defrag_ip_pkt
94 if not packet:
95 return None
96 if IP not in packet:
97 return packet
98 try:
99 return _defrag_ip_pkt(packet, self.fragments)[1] # type: ignore
100 except BadFragments:
101 return None
104class StringBuffer(object):
105 """StringBuffer is an object used to re-order data received during
106 a TCP transmission.
108 Each TCP fragment contains a sequence number, which marks
109 (relatively to the first sequence number) the index of the data contained
110 in the fragment.
112 If a TCP fragment is missed, this class will fill the missing space with
113 zeros.
115 :param max_gap: the largest missing range that will be zero-filled, in bytes.
116 A sequence number far from the data already buffered would otherwise
117 allocate the whole distance. Defaults to MTU.
118 """
120 def __init__(self, max_gap: int = MTU) -> None:
121 self.content = bytearray(b"")
122 self.content_len = 0
123 self.noff = 0 # negative offset
124 self.incomplete = [] # type: List[Tuple[int, int]]
125 self.max_gap = max_gap
127 def append(self, data: bytes, seq: Optional[int] = None) -> None:
128 if not data:
129 return
130 data_len = len(data)
131 if seq is None:
132 seq = self.content_len
133 seq = seq - 1 - self.noff
134 if seq < 0:
135 # Data is located before the start of the current buffer
136 # (e.g. the first fragment was missing)
137 if -seq - data_len > self.max_gap:
138 log_runtime.warning(
139 "Dropped data further than allowed per 'max_gap'."
140 )
141 return
142 self.content = bytearray(b"\x00" * (-seq)) + self.content
143 self.content_len += (-seq)
144 self.noff += seq
145 seq = 0
146 if seq + data_len > self.content_len:
147 # Data is located after the end of the current buffer
148 if seq - self.content_len > self.max_gap:
149 log_runtime.warning(
150 "Dropped data further than allowed per 'max_gap'."
151 )
152 return
153 self.content += b"\x00" * (seq - self.content_len + data_len)
154 # As data was missing, mark it.
155 # self.incomplete.append((self.content_len, seq))
156 self.content_len = seq + data_len
157 assert len(self.content) == self.content_len
158 # XXX removes empty space marker.
159 # for ifrag in self.incomplete:
160 # if [???]:
161 # self.incomplete.remove([???])
162 memoryview(self.content)[seq:seq + data_len] = data
164 def shiftleft(self, i: int) -> None:
165 self.content = self.content[i:]
166 self.content_len -= i
168 def full(self):
169 # type: () -> bool
170 # Should only be true when all missing data was filled up,
171 # (or there never was missing data)
172 return bool(self)
174 def clear(self):
175 # type: () -> None
176 self.__init__() # type: ignore
178 def __bool__(self):
179 # type: () -> bool
180 return bool(self.content_len)
181 __nonzero__ = __bool__
183 def __len__(self):
184 # type: () -> int
185 return self.content_len
187 def __bytes__(self):
188 # type: () -> bytes
189 return bytes(self.content)
191 def __str__(self):
192 # type: () -> str
193 return cast(str, self.__bytes__())
196def streamcls(cls: Type[Packet]) -> Callable[
197 [bytes, Dict[str, Any], Dict[str, Any]],
198 Optional[Packet],
199]:
200 """
201 Wraps a class for use when dissecting streams.
202 """
203 if hasattr(cls, "tcp_reassemble"):
204 return cls.tcp_reassemble # type: ignore
205 else:
206 # There is no tcp_reassemble. Just dissect the packet
207 return lambda data, *_: data and cls(data)
210class TCPSession(IPSession):
211 """A Session that reconstructs TCP streams.
213 NOTE: this has the same effect as wrapping a real socket.socket into StreamSocket,
214 but for all concurrent TCP streams (can be used on pcaps or sniffed sessions).
216 NOTE: only protocols that implement a ``tcp_reassemble`` function will be processed
217 by this session. Other protocols will not be reconstructed.
219 DEV: implement a class-function `tcp_reassemble` in your Packet class::
221 @classmethod
222 def tcp_reassemble(cls, data, metadata, session):
223 # data = the reassembled data from the same request/flow
224 # metadata = empty dictionary, that can be used to store data
225 # during TCP reassembly
226 # session = a dictionary proper to the bidirectional TCP session,
227 # that can be used to store anything
228 [...]
229 # If the packet is available, return it. Otherwise don't.
230 # Whenever you return a packet, the buffer will be discarded.
231 return pkt
232 # Otherwise, maybe store stuff in metadata, and return None,
233 # as you need additional data.
234 return None
236 For more details and a real example, see:
237 https://scapy.readthedocs.io/en/latest/usage.html#how-to-use-tcpsession-to-defragment-tcp-packets
239 :param app: Whether the socket is on application layer = has no TCP
240 layer. This is identical to StreamSocket so only use this if your
241 underlying source of data isn't a socket.socket.
242 """
244 def __init__(self, app=False, *args, **kwargs):
245 # type: (bool, *Any, **Any) -> None
246 super(TCPSession, self).__init__(*args, **kwargs)
247 self.app = app
248 if app:
249 self.data = StringBuffer()
250 self.metadata = {} # type: Dict[str, Any]
251 self.session = {} # type: Dict[str, Any]
252 else:
253 # The StringBuffer() is used to build a global
254 # string from fragments and their seq nulber
255 self.tcp_frags = defaultdict(
256 lambda: (StringBuffer(), {})
257 ) # type: DefaultDict[bytes, Tuple[StringBuffer, Dict[str, Any]]]
258 self.tcp_sessions = defaultdict(
259 dict
260 ) # type: DefaultDict[bytes, Dict[str, Any]]
261 # Setup stopping dissection condition
262 from scapy.layers.inet import TCP
263 self.stop_dissection_after = TCP
265 def _get_ident(self, pkt, session=False):
266 # type: (Packet, bool) -> bytes
267 underlayer = pkt["TCP"].underlayer
268 af = socket.AF_INET6 if "IPv6" in pkt else socket.AF_INET
269 src = underlayer and inet_pton(af, underlayer.src) or b""
270 dst = underlayer and inet_pton(af, underlayer.dst) or b""
271 if session:
272 # Bidirectional
273 def xor(x, y):
274 # type: (bytes, bytes) -> bytes
275 return bytes(a ^ b for a, b in zip(x, y))
276 return struct.pack("!4sH", xor(src, dst), pkt.dport ^ pkt.sport)
277 else:
278 # Uni-directional
279 return src + dst + struct.pack("!HH", pkt.dport, pkt.sport)
281 def _strip_padding(self, pkt: Packet) -> Optional[bytes]:
282 """Strip the packet of any padding, and return the padding.
283 """
284 if isinstance(pkt, conf.padding_layer):
285 return cast(bytes, pkt.load)
286 pad = pkt.getlayer(conf.padding_layer)
287 if pad is not None and pad.underlayer is not None:
288 # strip padding
289 del pad.underlayer.payload
290 return cast(bytes, pad.load)
291 return None
293 def process(self,
294 pkt: Packet,
295 cls: Optional[Type[Packet]] = None) -> Optional[Packet]:
296 """Process each packet: matches the TCP seq/ack numbers
297 to follow the TCP streams, and orders the fragments.
298 """
299 packet = None # type: Optional[Packet]
300 if self.app:
301 # Special mode: Application layer. Use on top of TCP
302 self.data.append(bytes(pkt))
303 if cls is None and not isinstance(pkt, bytes):
304 cls = pkt.__class__
305 if "tcp_reassemble" in self.metadata:
306 tcp_reassemble = self.metadata["tcp_reassemble"]
307 elif cls is not None:
308 self.metadata["tcp_reassemble"] = tcp_reassemble = streamcls(cls)
309 else:
310 return None
311 if self.data.full():
312 min_len = self.metadata.get("tcp_min_len")
313 if min_len is None or len(self.data) >= min_len:
314 self.metadata.pop("tcp_min_len", None)
315 packet = tcp_reassemble(
316 bytes(self.data),
317 self.metadata,
318 self.session,
319 )
320 if packet:
321 padding = self._strip_padding(packet)
322 if padding:
323 # There is remaining data for the next payload.
324 self.data.shiftleft(len(self.data) - len(padding))
325 # Skip full-padding
326 if isinstance(packet, conf.padding_layer):
327 return None
328 else:
329 # No padding (data) left. Clear
330 self.data.clear()
331 self.metadata.clear()
332 return packet
333 return None
335 _pkt = super(TCPSession, self).process(pkt)
336 if _pkt is None:
337 return None
338 else: # Python 3.8 := would be nice
339 pkt = _pkt
341 from scapy.layers.inet import IP, TCP
342 if not pkt:
343 return None
344 if TCP not in pkt:
345 return pkt
346 pay = pkt[TCP].payload
347 new_data = pay.original
348 # Match packets by a unique TCP identifier
349 ident = self._get_ident(pkt)
350 data, metadata = self.tcp_frags[ident]
351 tcp_session = self.tcp_sessions[self._get_ident(pkt, True)]
352 # Handle TCP sequence numbers
353 seq = pkt[TCP].seq
354 if "seq" not in metadata:
355 metadata["seq"] = seq
356 if "next_seq" in metadata and seq < metadata["next_seq"]:
357 # Retransmitted data (that we already returned)
358 new_data = new_data[metadata["next_seq"] - seq:]
359 if not new_data:
360 return None
361 seq = metadata["next_seq"]
362 # Let's guess which class is going to be used
363 if "pay_class" not in metadata:
364 metadata["pay_class"] = pay_class = pkt[TCP].guess_payload_class(new_data)
365 metadata["tcp_reassemble"] = tcp_reassemble = streamcls(pay_class)
366 else:
367 tcp_reassemble = metadata["tcp_reassemble"]
369 if pay:
370 # Get a relative sequence number for a storage purpose
371 relative_seq = metadata.get("relative_seq", None)
372 if relative_seq is None:
373 relative_seq = metadata["relative_seq"] = seq - 1
374 seq = seq - relative_seq
375 # Add the data to the buffer
376 noff = data.noff
377 data.append(new_data, seq)
378 if data.noff != noff:
379 # The negative offset has changed => data was prepended.
380 # this means that our estimate min length may be wrong.
381 metadata.pop("tcp_min_len", None)
383 # Check TCP FIN or TCP RESET
384 if pkt[TCP].flags.F or pkt[TCP].flags.R:
385 metadata["tcp_end"] = True
386 elif not pay:
387 # If there's no payload and the stream isn't ending, ignore.
388 return pkt
390 # In case any app layer protocol requires it,
391 # allow the parser to inspect TCP PSH flag
392 if pkt[TCP].flags.P:
393 metadata["tcp_psh"] = True
394 # XXX TODO: check that no empty space is missing in the buffer.
395 # XXX Currently, if a TCP fragment was missing, we won't notice it.
396 if data.full():
397 min_len = metadata.get("tcp_min_len")
398 if min_len is None or len(data) >= min_len:
399 # Reassemble using all previous packets
400 metadata["original"] = pkt
401 metadata["ident"] = ident
402 metadata.pop("tcp_min_len", None)
403 packet = tcp_reassemble(
404 bytes(data),
405 metadata,
406 tcp_session
407 )
408 # Stack the result on top of the previous frames
409 if packet:
410 if "seq" in metadata:
411 pkt[TCP].seq = metadata["seq"]
412 # Clear TCP reassembly metadata
413 metadata.clear()
414 # Check for padding
415 padding = self._strip_padding(packet)
416 while padding:
417 # There is remaining data for the next payload.
418 full_length = data.content_len - len(padding)
419 metadata["relative_seq"] = relative_seq + full_length
420 data.shiftleft(full_length)
421 # There might be a sub-payload hidden in the padding
422 sub_packet = tcp_reassemble(
423 bytes(data),
424 metadata,
425 tcp_session
426 )
427 if sub_packet:
428 packet.add_payload(sub_packet)
429 padding = self._strip_padding(sub_packet)
430 else:
431 break
432 else:
433 # No padding (data) left. Clear
434 data.clear()
435 del self.tcp_frags[ident]
436 # Minimum next seq
437 metadata["next_seq"] = pkt[TCP].seq + len(new_data)
438 # Skip full-padding
439 if isinstance(packet, conf.padding_layer):
440 return None
441 # Rebuild resulting packet
442 if pay:
443 pay.underlayer.remove_payload()
444 if IP in pkt:
445 pkt[IP].len = None
446 pkt[IP].chksum = None
447 pkt = pkt / packet
448 pkt.wirelen = None
449 return pkt
450 return None
452 def recv(self, sock: 'SuperSocket') -> Iterator[Packet]:
453 """
454 Will be called by sniff() to ask for a packet
455 """
456 pkt = sock.recv(stop_dissection_after=self.stop_dissection_after)
457 _orig = pkt
458 # Now handle TCP reassembly
459 if self.app:
460 while pkt is not None:
461 try:
462 pkt = self.process(pkt)
463 except Exception as ex:
464 log_runtime.warning(
465 "%s processing failed with '%s'. Aborting." % (
466 type(self),
467 ex,
468 )
469 )
470 if conf.debug_dissector:
471 raise
472 return None
473 if pkt:
474 yield pkt
475 # keep calling process as there might be more
476 pkt = b"" # type: ignore
477 else:
478 try:
479 pkt = self.process(pkt) # type: ignore
480 except Exception as ex:
481 log_runtime.warning(
482 "%s processing failed with '%s'. Skipping." % (
483 type(self),
484 ex,
485 )
486 )
487 if conf.debug_dissector:
488 raise
489 if _orig is not None:
490 yield _orig
491 return None
492 if pkt:
493 yield pkt
494 return None