Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/scapy/utils.py: 38%
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
4# Copyright (C) Philippe Biondi <phil@secdev.org>
6"""
7General utility functions.
8"""
11from decimal import Decimal
12from io import StringIO
13from itertools import zip_longest
14from uuid import UUID
16import argparse
17import array
18import collections
19import decimal
20import difflib
21import enum
22import gzip
23import inspect
24import locale
25import math
26import os
27import random
28import re
29import shutil
30import socket
31import struct
32import subprocess
33import sys
34import tempfile
35import threading
36import time
37import traceback
38import warnings
39import zlib
41from scapy.config import conf
42from scapy.consts import DARWIN, OPENBSD, WINDOWS
43from scapy.data import MTU, DLT_EN10MB, DLT_RAW
44from scapy.compat import (
45 plain_str,
46 chb,
47 hex_bytes,
48 bytes_encode,
49)
50from scapy.error import (
51 log_interactive,
52 log_runtime,
53 Scapy_Exception,
54 warning,
55)
56from scapy.pton_ntop import inet_pton
58# Typing imports
59from typing import (
60 Any,
61 AnyStr,
62 Callable,
63 cast,
64 Dict,
65 IO,
66 Iterator,
67 List,
68 Optional,
69 overload,
70 Tuple,
71 TYPE_CHECKING,
72 Type,
73 Union,
74)
75from scapy.compat import (
76 DecoratorCallable,
77 Literal,
78)
80if TYPE_CHECKING:
81 from scapy.packet import Packet
82 from scapy.plist import _PacketIterable, PacketList
83 from scapy.supersocket import SuperSocket
84 import prompt_toolkit
86_ByteStream = Union[IO[bytes], gzip.GzipFile]
88###########
89# Tools #
90###########
93def issubtype(x, # type: Any
94 t, # type: Union[type, str]
95 ):
96 # type: (...) -> bool
97 """issubtype(C, B) -> bool
99 Return whether C is a class and if it is a subclass of class B.
100 When using a tuple as the second argument issubtype(X, (A, B, ...)),
101 is a shortcut for issubtype(X, A) or issubtype(X, B) or ... (etc.).
102 """
103 if isinstance(t, str):
104 return t in (z.__name__ for z in x.__bases__)
105 if isinstance(x, type) and issubclass(x, t):
106 return True
107 return False
110_Decimal = Union[Decimal, int]
113class EDecimal(Decimal):
114 """Extended Decimal
116 This implements arithmetic and comparison with float for
117 backward compatibility
118 """
120 def __add__(self, other, context=None):
121 # type: (_Decimal, Any) -> EDecimal
122 return EDecimal(Decimal.__add__(self, Decimal(other)))
124 def __radd__(self, other):
125 # type: (_Decimal) -> EDecimal
126 return EDecimal(Decimal.__add__(self, Decimal(other)))
128 def __sub__(self, other):
129 # type: (_Decimal) -> EDecimal
130 return EDecimal(Decimal.__sub__(self, Decimal(other)))
132 def __rsub__(self, other):
133 # type: (_Decimal) -> EDecimal
134 return EDecimal(Decimal.__rsub__(self, Decimal(other)))
136 def __mul__(self, other):
137 # type: (_Decimal) -> EDecimal
138 return EDecimal(Decimal.__mul__(self, Decimal(other)))
140 def __rmul__(self, other):
141 # type: (_Decimal) -> EDecimal
142 return EDecimal(Decimal.__mul__(self, Decimal(other)))
144 def __truediv__(self, other):
145 # type: (_Decimal) -> EDecimal
146 return EDecimal(Decimal.__truediv__(self, Decimal(other)))
148 def __floordiv__(self, other):
149 # type: (_Decimal) -> EDecimal
150 return EDecimal(Decimal.__floordiv__(self, Decimal(other)))
152 def __divmod__(self, other):
153 # type: (_Decimal) -> Tuple[EDecimal, EDecimal]
154 r = Decimal.__divmod__(self, Decimal(other))
155 return EDecimal(r[0]), EDecimal(r[1])
157 def __mod__(self, other):
158 # type: (_Decimal) -> EDecimal
159 return EDecimal(Decimal.__mod__(self, Decimal(other)))
161 def __rmod__(self, other):
162 # type: (_Decimal) -> EDecimal
163 return EDecimal(Decimal.__rmod__(self, Decimal(other)))
165 def __pow__(self, other, modulo=None):
166 # type: (_Decimal, Optional[_Decimal]) -> EDecimal
167 return EDecimal(Decimal.__pow__(self, Decimal(other), modulo))
169 def __eq__(self, other):
170 # type: (Any) -> bool
171 if isinstance(other, Decimal):
172 return super(EDecimal, self).__eq__(other)
173 else:
174 return bool(float(self) == other)
176 def normalize(self, precision): # type: ignore
177 # type: (int) -> EDecimal
178 with decimal.localcontext() as ctx:
179 ctx.prec = precision
180 return EDecimal(super(EDecimal, self).normalize(ctx))
183@overload
184def get_temp_file(keep, autoext, fd):
185 # type: (bool, str, Literal[True]) -> IO[bytes]
186 pass
189@overload
190def get_temp_file(keep=False, autoext="", fd=False):
191 # type: (bool, str, Literal[False]) -> str
192 pass
195def get_temp_file(keep=False, autoext="", fd=False):
196 # type: (bool, str, bool) -> Union[IO[bytes], str]
197 """Creates a temporary file.
199 :param keep: If False, automatically delete the file when Scapy exits.
200 :param autoext: Suffix to add to the generated file name.
201 :param fd: If True, this returns a file-like object with the temporary
202 file opened. If False (default), this returns a file path.
203 """
204 f = tempfile.NamedTemporaryFile(prefix="scapy", suffix=autoext,
205 delete=False)
206 if not keep:
207 conf.temp_files.append(f.name)
209 if fd:
210 return f
211 else:
212 # Close the file so something else can take it.
213 f.close()
214 return f.name
217def get_temp_dir(keep=False):
218 # type: (bool) -> str
219 """Creates a temporary file, and returns its name.
221 :param keep: If False (default), the directory will be recursively
222 deleted when Scapy exits.
223 :return: A full path to a temporary directory.
224 """
226 dname = tempfile.mkdtemp(prefix="scapy")
228 if not keep:
229 conf.temp_files.append(dname)
231 return dname
234def _create_fifo() -> Tuple[str, Any]:
235 """Creates a temporary fifo.
237 You must then use open_fifo() on the server_fd once
238 the client is connected to use it.
240 :returns: (client_file, server_fd)
241 """
242 if WINDOWS:
243 from scapy.arch.windows.structures import _get_win_fifo
244 return _get_win_fifo()
245 else:
246 f = get_temp_file()
247 os.unlink(f)
248 os.mkfifo(f, 0o600)
249 return f, f
252def _open_fifo(fd: Any, mode: str = "rb") -> IO[bytes]:
253 """Open the server_fd (see create_fifo)
254 """
255 if WINDOWS:
256 from scapy.arch.windows.structures import _win_fifo_open
257 return _win_fifo_open(fd)
258 else:
259 return open(fd, mode)
262def sane(x, color=False):
263 # type: (bytes, bool) -> str
264 r = ""
265 for i in x:
266 j = i
267 if (j < 32) or (j >= 127):
268 if color:
269 r += conf.color_theme.not_printable(".")
270 else:
271 r += "."
272 else:
273 r += chr(j)
274 return r
277@conf.commands.register
278def restart():
279 # type: () -> None
280 """Restarts scapy"""
281 if not conf.interactive or not os.path.isfile(sys.argv[0]):
282 raise OSError("Scapy was not started from console")
283 if WINDOWS:
284 res_code = 1
285 try:
286 res_code = subprocess.call([sys.executable] + sys.argv)
287 finally:
288 os._exit(res_code)
289 os.execv(sys.executable, [sys.executable] + sys.argv)
292def lhex(x):
293 # type: (Any) -> str
294 from scapy.volatile import VolatileValue
295 if isinstance(x, VolatileValue):
296 return repr(x)
297 if isinstance(x, int):
298 return hex(x)
299 if isinstance(x, tuple):
300 return "(%s)" % ", ".join(lhex(v) for v in x)
301 if isinstance(x, list):
302 return "[%s]" % ", ".join(lhex(v) for v in x)
303 return str(x)
306@conf.commands.register
307def hexdump(p, dump=False):
308 # type: (Union[Packet, AnyStr], bool) -> Optional[str]
309 """Build a tcpdump like hexadecimal view
311 :param p: a Packet
312 :param dump: define if the result must be printed or returned in a variable
313 :return: a String only when dump=True
314 """
315 s = ""
316 x = bytes_encode(p)
317 x_len = len(x)
318 i = 0
319 while i < x_len:
320 s += "%04x " % i
321 for j in range(16):
322 if i + j < x_len:
323 s += "%02X " % x[i + j]
324 else:
325 s += " "
326 s += " %s\n" % sane(x[i:i + 16], color=True)
327 i += 16
328 # remove trailing \n
329 s = s[:-1] if s.endswith("\n") else s
330 if dump:
331 return s
332 else:
333 print(s)
334 return None
337@conf.commands.register
338def linehexdump(p, onlyasc=0, onlyhex=0, dump=False):
339 # type: (Union[Packet, AnyStr], int, int, bool) -> Optional[str]
340 """Build an equivalent view of hexdump() on a single line
342 Note that setting both onlyasc and onlyhex to 1 results in a empty output
344 :param p: a Packet
345 :param onlyasc: 1 to display only the ascii view
346 :param onlyhex: 1 to display only the hexadecimal view
347 :param dump: print the view if False
348 :return: a String only when dump=True
349 """
350 s = ""
351 s = hexstr(p, onlyasc=onlyasc, onlyhex=onlyhex, color=not dump)
352 if dump:
353 return s
354 else:
355 print(s)
356 return None
359@conf.commands.register
360def chexdump(p, dump=False):
361 # type: (Union[Packet, AnyStr], bool) -> Optional[str]
362 """Build a per byte hexadecimal representation
364 Example:
365 >>> chexdump(IP())
366 0x45, 0x00, 0x00, 0x14, 0x00, 0x01, 0x00, 0x00, 0x40, 0x00, 0x7c, 0xe7, 0x7f, 0x00, 0x00, 0x01, 0x7f, 0x00, 0x00, 0x01 # noqa: E501
368 :param p: a Packet
369 :param dump: print the view if False
370 :return: a String only if dump=True
371 """
372 x = bytes_encode(p)
373 s = ", ".join("%#04x" % x for x in x)
374 if dump:
375 return s
376 else:
377 print(s)
378 return None
381@conf.commands.register
382def hexstr(p, onlyasc=0, onlyhex=0, color=False):
383 # type: (Union[Packet, AnyStr], int, int, bool) -> str
384 """Build a fancy tcpdump like hex from bytes."""
385 x = bytes_encode(p)
386 s = []
387 if not onlyasc:
388 s.append(" ".join("%02X" % b for b in x))
389 if not onlyhex:
390 s.append(sane(x, color=color))
391 return " ".join(s)
394def repr_hex(s):
395 # type: (bytes) -> str
396 """ Convert provided bitstring to a simple string of hex digits """
397 return "".join("%02x" % x for x in s)
400@conf.commands.register
401def hexdiff(
402 a: Union['Packet', AnyStr],
403 b: Union['Packet', AnyStr],
404 algo: Optional[str] = None,
405 autojunk: bool = False,
406) -> None:
407 """
408 Show differences between 2 binary strings, Packets...
410 Available algorithms:
411 - wagnerfischer: Use the Wagner and Fischer algorithm to compute the
412 Levenstein distance between the strings then backtrack.
413 - difflib: Use the difflib.SequenceMatcher implementation. This based on a
414 modified version of the Ratcliff and Obershelp algorithm.
415 This is much faster, but far less accurate.
416 https://docs.python.org/3.8/library/difflib.html#difflib.SequenceMatcher
418 :param a:
419 :param b: The binary strings, packets... to compare
420 :param algo: Force the algo to be 'wagnerfischer' or 'difflib'.
421 By default, this is chosen depending on the complexity, optimistically
422 preferring wagnerfischer unless really necessary.
423 :param autojunk: (difflib only) See difflib documentation.
424 """
425 xb = bytes_encode(a)
426 yb = bytes_encode(b)
428 if algo is None:
429 # Choose the best algorithm
430 complexity = len(xb) * len(yb)
431 if complexity < 1e7:
432 # Comparing two (non-jumbos) Ethernet packets is ~2e6 which is manageable.
433 # Anything much larger than this shouldn't be attempted by default.
434 algo = "wagnerfischer"
435 if complexity > 1e6:
436 log_interactive.info(
437 "Complexity is a bit high. hexdiff will take a few seconds."
438 )
439 else:
440 algo = "difflib"
442 backtrackx = []
443 backtracky = []
445 if algo == "wagnerfischer":
446 xb = xb[::-1]
447 yb = yb[::-1]
449 # costs for the 3 operations
450 INSERT = 1
451 DELETE = 1
452 SUBST = 1
454 # Typically, d[i,j] will hold the distance between
455 # the first i characters of xb and the first j characters of yb.
456 # We change the Wagner Fischer to also store pointers to all
457 # the intermediate steps taken while calculating the Levenstein distance.
458 d = {(-1, -1): (0, (-1, -1))}
459 for j in range(len(yb)):
460 d[-1, j] = (j + 1) * INSERT, (-1, j - 1)
461 for i in range(len(xb)):
462 d[i, -1] = (i + 1) * INSERT + 1, (i - 1, -1)
464 # Compute the Levenstein distance between the two strings, but
465 # store all the steps to be able to backtrack at the end.
466 for j in range(len(yb)):
467 for i in range(len(xb)):
468 d[i, j] = min(
469 (d[i - 1, j - 1][0] + SUBST * (xb[i] != yb[j]), (i - 1, j - 1)),
470 (d[i - 1, j][0] + DELETE, (i - 1, j)),
471 (d[i, j - 1][0] + INSERT, (i, j - 1)),
472 )
474 # Iterate through the steps backwards to create the diff
475 i = len(xb) - 1
476 j = len(yb) - 1
477 while not (i == j == -1):
478 i2, j2 = d[i, j][1]
479 backtrackx.append(xb[i2 + 1:i + 1])
480 backtracky.append(yb[j2 + 1:j + 1])
481 i, j = i2, j2
482 elif algo == "difflib":
483 sm = difflib.SequenceMatcher(a=xb, b=yb, autojunk=autojunk)
484 xarr = [xb[i:i + 1] for i in range(len(xb))]
485 yarr = [yb[i:i + 1] for i in range(len(yb))]
486 # Iterate through opcodes to build the backtrack
487 for opcode in sm.get_opcodes():
488 typ, x0, x1, y0, y1 = opcode
489 if typ == 'delete':
490 backtrackx += xarr[x0:x1]
491 backtracky += [b''] * (x1 - x0)
492 elif typ == 'insert':
493 backtrackx += [b''] * (y1 - y0)
494 backtracky += yarr[y0:y1]
495 elif typ in ['equal', 'replace']:
496 backtrackx += xarr[x0:x1]
497 backtracky += yarr[y0:y1]
498 # Some lines may have been considered as junk. Check the sizes
499 if autojunk:
500 lbx = len(backtrackx)
501 lby = len(backtracky)
502 backtrackx += [b''] * (max(lbx, lby) - lbx)
503 backtracky += [b''] * (max(lbx, lby) - lby)
504 else:
505 raise ValueError("Unknown algorithm '%s'" % algo)
507 # Print the diff
509 x = y = i = 0
510 colorize: Dict[int, Callable[[str], str]] = {
511 0: lambda x: x,
512 -1: conf.color_theme.left,
513 1: conf.color_theme.right
514 }
516 dox = 1
517 doy = 0
518 btx_len = len(backtrackx)
519 while i < btx_len:
520 linex = backtrackx[i:i + 16]
521 liney = backtracky[i:i + 16]
522 xx = sum(len(k) for k in linex)
523 yy = sum(len(k) for k in liney)
524 if dox and not xx:
525 dox = 0
526 doy = 1
527 if dox and linex == liney:
528 doy = 1
530 if dox:
531 xd = y
532 j = 0
533 while j < len(linex) and not linex[j]:
534 j += 1
535 xd -= 1
536 print(colorize[doy - dox]("%04x" % xd), end=' ')
537 x += xx
538 line = linex
539 else:
540 print(" ", end=' ')
541 if doy:
542 yd = y
543 j = 0
544 while j < len(liney) and not liney[j]:
545 j += 1
546 yd -= 1
547 print(colorize[doy - dox]("%04x" % yd), end=' ')
548 y += yy
549 line = liney
550 else:
551 print(" ", end=' ')
553 print(" ", end=' ')
555 cl = ""
556 for j in range(16):
557 if i + j < min(len(backtrackx), len(backtracky)):
558 if line[j]:
559 col = colorize[(linex[j] != liney[j]) * (doy - dox)]
560 print(col("%02X" % line[j][0]), end=' ')
561 if linex[j] == liney[j]:
562 cl += sane(line[j], color=True)
563 else:
564 cl += col(sane(line[j]))
565 else:
566 print(" ", end=' ')
567 cl += " "
568 else:
569 print(" ", end=' ')
570 if j == 7:
571 print("", end=' ')
573 print(" ", cl)
575 if doy or not yy:
576 doy = 0
577 dox = 1
578 i += 16
579 else:
580 if yy:
581 dox = 0
582 doy = 1
583 else:
584 i += 16
587if struct.pack("H", 1) == b"\x00\x01": # big endian
588 checksum_endian_transform = lambda chk: chk # type: Callable[[int], int]
589else:
590 checksum_endian_transform = lambda chk: ((chk >> 8) & 0xff) | chk << 8
593def checksum(pkt):
594 # type: (bytes) -> int
595 if len(pkt) % 2 == 1:
596 pkt += b"\0"
597 s = sum(array.array("H", pkt))
598 s = (s >> 16) + (s & 0xffff)
599 s += s >> 16
600 s = ~s
601 return checksum_endian_transform(s) & 0xffff
604def _fletcher16(charbuf):
605 # type: (bytes) -> Tuple[int, int]
606 # This is based on the GPLed C implementation in Zebra <http://www.zebra.org/> # noqa: E501
607 c0 = c1 = 0
608 for char in charbuf:
609 c0 += char
610 c1 += c0
612 c0 %= 255
613 c1 %= 255
614 return (c0, c1)
617@conf.commands.register
618def fletcher16_checksum(binbuf):
619 # type: (bytes) -> int
620 """Calculates Fletcher-16 checksum of the given buffer.
622 Note:
623 If the buffer contains the two checkbytes derived from the Fletcher-16 checksum # noqa: E501
624 the result of this function has to be 0. Otherwise the buffer has been corrupted. # noqa: E501
625 """
626 (c0, c1) = _fletcher16(binbuf)
627 return (c1 << 8) | c0
630@conf.commands.register
631def fletcher16_checkbytes(binbuf, offset):
632 # type: (bytes, int) -> bytes
633 """Calculates the Fletcher-16 checkbytes returned as 2 byte binary-string.
635 Including the bytes into the buffer (at the position marked by offset) the # noqa: E501
636 global Fletcher-16 checksum of the buffer will be 0. Thus it is easy to verify # noqa: E501
637 the integrity of the buffer on the receiver side.
639 For details on the algorithm, see RFC 2328 chapter 12.1.7 and RFC 905 Annex B. # noqa: E501
640 """
642 # This is based on the GPLed C implementation in Zebra <http://www.zebra.org/> # noqa: E501
643 if len(binbuf) < offset:
644 raise Exception("Packet too short for checkbytes %d" % len(binbuf))
646 binbuf = binbuf[:offset] + b"\x00\x00" + binbuf[offset + 2:]
647 (c0, c1) = _fletcher16(binbuf)
649 x = ((len(binbuf) - offset - 1) * c0 - c1) % 255
651 if (x <= 0):
652 x += 255
654 y = 510 - c0 - x
656 if (y > 255):
657 y -= 255
658 return chb(x) + chb(y)
661def mac2str(mac):
662 # type: (str) -> bytes
663 return b"".join(chb(int(x, 16)) for x in plain_str(mac).split(':'))
666def valid_mac(mac):
667 # type: (str) -> bool
668 try:
669 return len(mac2str(mac)) == 6
670 except ValueError:
671 pass
672 return False
675def str2mac(s):
676 # type: (bytes) -> str
677 if isinstance(s, str):
678 return ("%02x:" * len(s))[:-1] % tuple(map(ord, s))
679 return ("%02x:" * len(s))[:-1] % tuple(s)
682def randstring(length):
683 # type: (int) -> bytes
684 """
685 Returns a random string of length (length >= 0)
686 """
687 return b"".join(struct.pack('B', random.randint(0, 255))
688 for _ in range(length))
691def zerofree_randstring(length):
692 # type: (int) -> bytes
693 """
694 Returns a random string of length (length >= 0) without zero in it.
695 """
696 return b"".join(struct.pack('B', random.randint(1, 255))
697 for _ in range(length))
700def stror(s1, s2):
701 # type: (bytes, bytes) -> bytes
702 """
703 Returns the binary OR of the 2 provided strings s1 and s2. s1 and s2
704 must be of same length.
705 """
706 return b"".join(map(lambda x, y: struct.pack("!B", x | y), s1, s2))
709def strxor(s1, s2):
710 # type: (bytes, bytes) -> bytes
711 """
712 Returns the binary XOR of the 2 provided strings s1 and s2. s1 and s2
713 must be of same length.
714 """
715 return b"".join(map(lambda x, y: struct.pack("!B", x ^ y), s1, s2))
718def strand(s1, s2):
719 # type: (bytes, bytes) -> bytes
720 """
721 Returns the binary AND of the 2 provided strings s1 and s2. s1 and s2
722 must be of same length.
723 """
724 return b"".join(map(lambda x, y: struct.pack("!B", x & y), s1, s2))
727def strrot(s1, count, right=True):
728 # type: (bytes, int, bool) -> bytes
729 """
730 Rotate the binary by 'count' bytes
731 """
732 off = count % len(s1)
733 if right:
734 return s1[-off:] + s1[:-off]
735 else:
736 return s1[off:] + s1[:off]
739# Workaround bug 643005 : https://sourceforge.net/tracker/?func=detail&atid=105470&aid=643005&group_id=5470 # noqa: E501
740try:
741 socket.inet_aton("255.255.255.255")
742except socket.error:
743 def inet_aton(ip_string):
744 # type: (str) -> bytes
745 if ip_string == "255.255.255.255":
746 return b"\xff" * 4
747 else:
748 return socket.inet_aton(ip_string)
749else:
750 inet_aton = socket.inet_aton # type: ignore
752inet_ntoa = socket.inet_ntoa
755def atol(x):
756 # type: (str) -> int
757 try:
758 ip = inet_aton(x)
759 except socket.error:
760 raise ValueError("Bad IP format: %s" % x)
761 return cast(int, struct.unpack("!I", ip)[0])
764def valid_ip(addr):
765 # type: (str) -> bool
766 try:
767 addr = plain_str(addr)
768 except UnicodeDecodeError:
769 return False
770 try:
771 atol(addr)
772 except (OSError, ValueError, socket.error):
773 return False
774 return True
777def valid_net(addr):
778 # type: (str) -> bool
779 try:
780 addr = plain_str(addr)
781 except UnicodeDecodeError:
782 return False
783 if '/' in addr:
784 ip, mask = addr.split('/', 1)
785 return valid_ip(ip) and mask.isdigit() and 0 <= int(mask) <= 32
786 return valid_ip(addr)
789def valid_ip6(addr):
790 # type: (str) -> bool
791 try:
792 addr = plain_str(addr)
793 except UnicodeDecodeError:
794 return False
795 try:
796 inet_pton(socket.AF_INET6, addr)
797 except socket.error:
798 return False
799 return True
802def valid_net6(addr):
803 # type: (str) -> bool
804 try:
805 addr = plain_str(addr)
806 except UnicodeDecodeError:
807 return False
808 if '/' in addr:
809 ip, mask = addr.split('/', 1)
810 return valid_ip6(ip) and mask.isdigit() and 0 <= int(mask) <= 128
811 return valid_ip6(addr)
814def ltoa(x):
815 # type: (int) -> str
816 return inet_ntoa(struct.pack("!I", x & 0xffffffff))
819def itom(x):
820 # type: (int) -> int
821 return (0xffffffff00000000 >> x) & 0xffffffff
824def in4_cidr2mask(m):
825 # type: (int) -> bytes
826 """
827 Return the mask (bitstring) associated with provided length
828 value. For instance if function is called on 20, return value is
829 b'\xff\xff\xf0\x00'.
830 """
831 if m > 32 or m < 0:
832 raise Scapy_Exception("value provided to in4_cidr2mask outside [0, 32] domain (%d)" % m) # noqa: E501
834 return strxor(
835 b"\xff" * 4,
836 struct.pack(">I", 2**(32 - m) - 1)
837 )
840def in4_isincluded(addr, prefix, mask):
841 # type: (str, str, int) -> bool
842 """
843 Returns True when 'addr' belongs to prefix/mask. False otherwise.
844 """
845 temp = inet_pton(socket.AF_INET, addr)
846 pref = in4_cidr2mask(mask)
847 zero = inet_pton(socket.AF_INET, prefix)
848 return zero == strand(temp, pref)
851def in4_ismaddr(str):
852 # type: (str) -> bool
853 """
854 Returns True if provided address in printable format belongs to
855 allocated Multicast address space (224.0.0.0/4).
856 """
857 return in4_isincluded(str, "224.0.0.0", 4)
860def in4_ismlladdr(str):
861 # type: (str) -> bool
862 """
863 Returns True if address belongs to link-local multicast address
864 space (224.0.0.0/24)
865 """
866 return in4_isincluded(str, "224.0.0.0", 24)
869def in4_ismgladdr(str):
870 # type: (str) -> bool
871 """
872 Returns True if address belongs to global multicast address
873 space (224.0.1.0-238.255.255.255).
874 """
875 return (
876 in4_isincluded(str, "224.0.0.0", 4) and
877 not in4_isincluded(str, "224.0.0.0", 24) and
878 not in4_isincluded(str, "239.0.0.0", 8)
879 )
882def in4_ismlsaddr(str):
883 # type: (str) -> bool
884 """
885 Returns True if address belongs to limited scope multicast address
886 space (239.0.0.0/8).
887 """
888 return in4_isincluded(str, "239.0.0.0", 8)
891def in4_isaddrllallnodes(str):
892 # type: (str) -> bool
893 """
894 Returns True if address is the link-local all-nodes multicast
895 address (224.0.0.1).
896 """
897 return (inet_pton(socket.AF_INET, "224.0.0.1") ==
898 inet_pton(socket.AF_INET, str))
901def in4_getnsmac(a):
902 # type: (bytes) -> str
903 """
904 Return the multicast mac address associated with provided
905 IPv4 address. Passed address must be in network format.
906 """
908 return "01:00:5e:%.2x:%.2x:%.2x" % (a[1] & 0x7f, a[2], a[3])
911def decode_locale_str(x):
912 # type: (bytes) -> str
913 """
914 Decode bytes into a string using the system locale.
915 Useful on Windows where it can be unusual (e.g. cp1252)
916 """
917 return x.decode(encoding=locale.getlocale()[1] or "utf-8", errors="replace")
920class ContextManagerSubprocess(object):
921 """
922 Context manager that eases checking for unknown command, without
923 crashing.
925 Example:
926 >>> with ContextManagerSubprocess("tcpdump"):
927 >>> subprocess.Popen(["tcpdump", "--version"])
928 ERROR: Could not execute tcpdump, is it installed?
930 """
932 def __init__(self, prog, suppress=True):
933 # type: (str, bool) -> None
934 self.prog = prog
935 self.suppress = suppress
937 def __enter__(self):
938 # type: () -> None
939 pass
941 def __exit__(self,
942 exc_type, # type: Optional[type]
943 exc_value, # type: Optional[Exception]
944 traceback, # type: Optional[Any]
945 ):
946 # type: (...) -> Optional[bool]
947 if exc_value is None or exc_type is None:
948 return None
949 # Errored
950 if isinstance(exc_value, EnvironmentError):
951 msg = "Could not execute %s, is it installed?" % self.prog
952 else:
953 msg = "%s: execution failed (%s)" % (
954 self.prog,
955 exc_type.__class__.__name__
956 )
957 if not self.suppress:
958 raise exc_type(msg)
959 log_runtime.error(msg, exc_info=True)
960 return True # Suppress the exception
963class ContextManagerCaptureOutput(object):
964 """
965 Context manager that intercept the console's output.
967 Example:
968 >>> with ContextManagerCaptureOutput() as cmco:
969 ... print("hey")
970 ... assert cmco.get_output() == "hey"
971 """
973 def __init__(self):
974 # type: () -> None
975 self.result_export_object = ""
977 def __enter__(self):
978 # type: () -> ContextManagerCaptureOutput
979 from unittest import mock
981 def write(s, decorator=self):
982 # type: (str, ContextManagerCaptureOutput) -> None
983 decorator.result_export_object += s
984 mock_stdout = mock.Mock()
985 mock_stdout.write = write
986 self.bck_stdout = sys.stdout
987 sys.stdout = mock_stdout
988 return self
990 def __exit__(self, *exc):
991 # type: (*Any) -> Literal[False]
992 sys.stdout = self.bck_stdout
993 return False
995 def get_output(self, eval_bytes=False):
996 # type: (bool) -> str
997 if self.result_export_object.startswith("b'") and eval_bytes:
998 return plain_str(eval(self.result_export_object))
999 return self.result_export_object
1002def do_graph(
1003 graph, # type: str
1004 prog=None, # type: Optional[str]
1005 format=None, # type: Optional[str]
1006 target=None, # type: Optional[Union[IO[bytes], str]]
1007 type=None, # type: Optional[str]
1008 string=None, # type: Optional[bool]
1009 options=None # type: Optional[List[str]]
1010):
1011 # type: (...) -> Optional[str]
1012 """Processes graph description using an external software.
1013 This method is used to convert a graphviz format to an image.
1015 :param graph: GraphViz graph description
1016 :param prog: which graphviz program to use
1017 :param format: output type (svg, ps, gif, jpg, etc.), passed to dot's "-T"
1018 option
1019 :param string: if not None, simply return the graph string
1020 :param target: filename or redirect. Defaults pipe to Imagemagick's
1021 display program
1022 :param options: options to be passed to prog
1023 """
1025 if format is None:
1026 format = "svg"
1027 if string:
1028 return graph
1029 if type is not None:
1030 warnings.warn(
1031 "type is deprecated, and was renamed format",
1032 DeprecationWarning
1033 )
1034 format = type
1035 if prog is None:
1036 prog = conf.prog.dot
1037 start_viewer = False
1038 if target is None:
1039 if WINDOWS:
1040 target = get_temp_file(autoext="." + format)
1041 start_viewer = True
1042 else:
1043 with ContextManagerSubprocess(conf.prog.display):
1044 target = subprocess.Popen([conf.prog.display],
1045 stdin=subprocess.PIPE).stdin
1046 if format is not None:
1047 format = "-T%s" % format
1048 if isinstance(target, str):
1049 if target.startswith('|'):
1050 target = subprocess.Popen(target[1:].lstrip(), shell=True,
1051 stdin=subprocess.PIPE).stdin
1052 elif target.startswith('>'):
1053 target = open(target[1:].lstrip(), "wb")
1054 else:
1055 target = open(os.path.abspath(target), "wb")
1056 target = cast(IO[bytes], target)
1057 proc = subprocess.Popen(
1058 "\"%s\" %s %s" % (prog, options or "", format or ""),
1059 shell=True, stdin=subprocess.PIPE, stdout=target,
1060 stderr=subprocess.PIPE
1061 )
1062 _, stderr = proc.communicate(bytes_encode(graph))
1063 if proc.returncode != 0:
1064 raise OSError(
1065 "GraphViz call failed (is it installed?):\n" +
1066 plain_str(stderr)
1067 )
1068 try:
1069 target.close()
1070 except Exception:
1071 pass
1072 if start_viewer:
1073 # Workaround for file not found error: We wait until tempfile is written. # noqa: E501
1074 waiting_start = time.time()
1075 while not os.path.exists(target.name):
1076 time.sleep(0.1)
1077 if time.time() - waiting_start > 3:
1078 warning("Temporary file '%s' could not be written. Graphic will not be displayed.", tempfile) # noqa: E501
1079 break
1080 else:
1081 if WINDOWS and conf.prog.display == conf.prog._default:
1082 os.startfile(target.name)
1083 else:
1084 with ContextManagerSubprocess(conf.prog.display):
1085 subprocess.Popen([conf.prog.display, target.name])
1086 return None
1089_TEX_TR = {
1090 "{": "{\\tt\\char123}",
1091 "}": "{\\tt\\char125}",
1092 "\\": "{\\tt\\char92}",
1093 "^": "\\^{}",
1094 "$": "\\$",
1095 "#": "\\#",
1096 "_": "\\_",
1097 "&": "\\&",
1098 "%": "\\%",
1099 "|": "{\\tt\\char124}",
1100 "~": "{\\tt\\char126}",
1101 "<": "{\\tt\\char60}",
1102 ">": "{\\tt\\char62}",
1103}
1106def tex_escape(x):
1107 # type: (str) -> str
1108 s = ""
1109 for c in x:
1110 s += _TEX_TR.get(c, c)
1111 return s
1114def graphviz_escape(x):
1115 # type: (Any) -> str
1116 return str(x).replace("\\", "\\\\").replace('"', '\\"')
1119def colgen(*lstcol, # type: Any
1120 **kargs # type: Any
1121 ):
1122 # type: (...) -> Iterator[Any]
1123 """Returns a generator that mixes provided quantities forever
1124 trans: a function to convert the three arguments into a color. lambda x,y,z:(x,y,z) by default""" # noqa: E501
1125 if len(lstcol) < 2:
1126 lstcol *= 2
1127 trans = kargs.get("trans", lambda x, y, z: (x, y, z))
1128 while True:
1129 for i in range(len(lstcol)):
1130 for j in range(len(lstcol)):
1131 for k in range(len(lstcol)):
1132 if i != j or j != k or k != i:
1133 yield trans(lstcol[(i + j) % len(lstcol)], lstcol[(j + k) % len(lstcol)], lstcol[(k + i) % len(lstcol)]) # noqa: E501
1136def incremental_label(label="tag%05i", start=0):
1137 # type: (str, int) -> Iterator[str]
1138 while True:
1139 yield label % start
1140 start += 1
1143def binrepr(val):
1144 # type: (int) -> str
1145 return bin(val)[2:]
1148def long_converter(s):
1149 # type: (str) -> int
1150 return int(s.replace('\n', '').replace(' ', ''), 16)
1152#########################
1153# Enum management #
1154#########################
1157class EnumElement:
1158 def __init__(self, key, value):
1159 # type: (str, int) -> None
1160 self._key = key
1161 self._value = value
1163 def __repr__(self):
1164 # type: () -> str
1165 return "<%s %s[%r]>" % (self.__dict__.get("_name", self.__class__.__name__), self._key, self._value) # noqa: E501
1167 def __getattr__(self, attr):
1168 # type: (str) -> Any
1169 return getattr(self._value, attr)
1171 def __str__(self):
1172 # type: () -> str
1173 return self._key
1175 def __bytes__(self):
1176 # type: () -> bytes
1177 return bytes_encode(self.__str__())
1179 def __hash__(self):
1180 # type: () -> int
1181 return self._value
1183 def __int__(self):
1184 # type: () -> int
1185 return int(self._value)
1187 def __eq__(self, other):
1188 # type: (Any) -> bool
1189 return self._value == int(other)
1191 def __neq__(self, other):
1192 # type: (Any) -> bool
1193 return not self.__eq__(other)
1196class Enum_metaclass(type):
1197 element_class = EnumElement
1199 def __new__(cls, name, bases, dct):
1200 # type: (Any, str, Any, Dict[str, Any]) -> Any
1201 rdict = {}
1202 for k, v in dct.items():
1203 if isinstance(v, int):
1204 v = cls.element_class(k, v)
1205 dct[k] = v
1206 rdict[v] = k
1207 dct["__rdict__"] = rdict
1208 return super(Enum_metaclass, cls).__new__(cls, name, bases, dct)
1210 def __getitem__(self, attr):
1211 # type: (int) -> Any
1212 return self.__rdict__[attr] # type: ignore
1214 def __contains__(self, val):
1215 # type: (int) -> bool
1216 return val in self.__rdict__ # type: ignore
1218 def get(self, attr, val=None):
1219 # type: (str, Optional[Any]) -> Any
1220 return self.__rdict__.get(attr, val) # type: ignore
1222 def __repr__(self):
1223 # type: () -> str
1224 return "<%s>" % self.__dict__.get("name", self.__name__)
1227##################
1228# Corrupt data #
1229##################
1231@conf.commands.register
1232def corrupt_bytes(data, p=0.01, n=None):
1233 # type: (str, float, Optional[int]) -> bytes
1234 """
1235 Corrupt a given percentage (at least one byte) or number of bytes
1236 from a string
1237 """
1238 s = array.array("B", bytes_encode(data))
1239 s_len = len(s)
1240 if n is None:
1241 n = max(1, int(s_len * p))
1242 for i in random.sample(range(s_len), n):
1243 s[i] = (s[i] + random.randint(1, 255)) % 256
1244 return s.tobytes()
1247@conf.commands.register
1248def corrupt_bits(data, p=0.01, n=None):
1249 # type: (str, float, Optional[int]) -> bytes
1250 """
1251 Flip a given percentage (at least one bit) or number of bits
1252 from a string
1253 """
1254 s = array.array("B", bytes_encode(data))
1255 s_len = len(s) * 8
1256 if n is None:
1257 n = max(1, int(s_len * p))
1258 for i in random.sample(range(s_len), n):
1259 s[i // 8] ^= 1 << (i % 8)
1260 return s.tobytes()
1263#############################
1264# pcap capture file stuff #
1265#############################
1267@conf.commands.register
1268def wrpcap(filename, # type: Union[IO[bytes], str]
1269 pkt, # type: _PacketIterable
1270 *args, # type: Any
1271 **kargs # type: Any
1272 ):
1273 # type: (...) -> None
1274 """Write a list of packets to a pcap file
1276 :param filename: the name of the file to write packets to, or an open,
1277 writable file-like object. The file descriptor will be
1278 closed at the end of the call, so do not use an object you
1279 do not want to close (e.g., running wrpcap(sys.stdout, [])
1280 in interactive mode will crash Scapy).
1281 :param gz: set to 1 to save a gzipped capture
1282 :param linktype: force linktype value
1283 :param endianness: "<" or ">", force endianness
1284 :param sync: do not bufferize writes to the capture file
1285 """
1286 with PcapWriter(filename, *args, **kargs) as fdesc:
1287 fdesc.write(pkt)
1290@conf.commands.register
1291def wrpcapng(filename, # type: str
1292 pkt, # type: _PacketIterable
1293 ):
1294 # type: (...) -> None
1295 """Write a list of packets to a pcapng file
1297 :param filename: the name of the file to write packets to, or an open,
1298 writable file-like object. The file descriptor will be
1299 closed at the end of the call, so do not use an object you
1300 do not want to close (e.g., running wrpcapng(sys.stdout, [])
1301 in interactive mode will crash Scapy).
1302 :param pkt: packets to write
1303 """
1304 with PcapNgWriter(filename) as fdesc:
1305 fdesc.write(pkt)
1308@conf.commands.register
1309def rdpcap(filename, count=-1):
1310 # type: (Union[IO[bytes], str], int) -> PacketList
1311 """Read a pcap or pcapng file and return a packet list
1313 :param count: read only <count> packets
1314 """
1315 # Rant: Our complicated use of metaclasses and especially the
1316 # __call__ function is, of course, not supported by MyPy.
1317 # One day we should simplify this mess and use a much simpler
1318 # layout that will actually be supported and properly dissected.
1319 with PcapReader(filename) as fdesc: # type: ignore
1320 return fdesc.read_all(count=count)
1323# NOTE: Type hinting
1324# Mypy doesn't understand the following metaclass, and thinks each
1325# constructor (PcapReader...) needs 3 arguments each. To avoid this,
1326# we add a fake (=None) to the last 2 arguments then force the value
1327# to not be None in the signature and pack the whole thing in an ignore.
1328# This allows to not have # type: ignore every time we call those
1329# constructors.
1331class PcapReader_metaclass(type):
1332 """Metaclass for (Raw)Pcap(Ng)Readers"""
1334 def __new__(cls, name, bases, dct):
1335 # type: (Any, str, Any, Dict[str, Any]) -> Any
1336 """The `alternative` class attribute is declared in the PcapNg
1337 variant, and set here to the Pcap variant.
1339 """
1340 newcls = super(PcapReader_metaclass, cls).__new__(
1341 cls, name, bases, dct
1342 )
1343 if 'alternative' in dct:
1344 dct['alternative'].alternative = newcls
1345 return newcls
1347 def __call__(cls, filename):
1348 # type: (Union[IO[bytes], str]) -> Any
1349 """Creates a cls instance, use the `alternative` if that
1350 fails.
1352 """
1353 i = cls.__new__(
1354 cls,
1355 cls.__name__,
1356 cls.__bases__,
1357 cls.__dict__ # type: ignore
1358 )
1359 filename, fdesc, magic = cls.open(filename)
1360 if not magic:
1361 raise Scapy_Exception(
1362 "No data could be read!"
1363 )
1364 try:
1365 i.__init__(filename, fdesc, magic)
1366 return i
1367 except (Scapy_Exception, EOFError):
1368 pass
1370 if "alternative" in cls.__dict__:
1371 cls = cls.__dict__["alternative"]
1372 i = cls.__new__(
1373 cls,
1374 cls.__name__,
1375 cls.__bases__,
1376 cls.__dict__ # type: ignore
1377 )
1378 try:
1379 i.__init__(filename, fdesc, magic)
1380 return i
1381 except (Scapy_Exception, EOFError):
1382 pass
1384 raise Scapy_Exception("Not a supported capture file")
1386 @staticmethod
1387 def open(fname # type: Union[IO[bytes], str]
1388 ):
1389 # type: (...) -> Tuple[str, _ByteStream, bytes]
1390 """Open (if necessary) filename, and read the magic."""
1391 if isinstance(fname, str):
1392 filename = fname
1393 fdesc = open(filename, "rb") # type: _ByteStream
1394 magic = fdesc.peek(2)[:2] # type: ignore[union-attr] # BufferedReader
1395 if magic != b"\x1f\x8b":
1396 magic = fdesc.read(4)
1397 else:
1398 fdesc = fname
1399 filename = getattr(fdesc, "name", "No name")
1400 magic = fdesc.read(2)
1401 if magic == b"\x1f\x8b":
1402 # GZIP header detected.
1403 if not isinstance(fname, str):
1404 fdesc.seek(0)
1405 raw_fdesc = fdesc
1406 fdesc = gzip.GzipFile(fileobj=raw_fdesc)
1407 try:
1408 magic = fdesc.read(4)
1409 except (EOFError, OSError, zlib.error) as err:
1410 fdesc.close()
1411 if isinstance(fname, str):
1412 raw_fdesc.close()
1413 raise Scapy_Exception(
1414 "Not a supported capture file (invalid gzip file)"
1415 ) from err
1416 else:
1417 magic += fdesc.read(4 - len(magic))
1418 return filename, fdesc, magic
1421class RawPcapReader(metaclass=PcapReader_metaclass):
1422 """A stateful pcap reader. Each packet is returned as a string"""
1424 # TODO: use Generics to properly type the various readers.
1425 # As of right now, RawPcapReader is typed as if it returned packets
1426 # because all of its child do. Fix that
1428 nonblocking_socket = True
1429 PacketMetadata = collections.namedtuple("PacketMetadata",
1430 ["sec", "usec", "wirelen", "caplen"]) # noqa: E501
1431 # A helper subprocess (e.g. the tcpdump prefilter that sniff() spawns for
1432 # an offline capture with a filter) whose lifetime is bound to this reader.
1433 # It is reaped in close() so it does not linger as a zombie (#4512).
1434 subproc = None # type: Optional[subprocess.Popen[bytes]]
1436 def __init__(self, filename, fdesc=None, magic=None): # type: ignore
1437 # type: (str, _ByteStream, bytes) -> None
1438 self.filename = filename
1439 self.f = fdesc
1440 if magic == b"\xa1\xb2\xc3\xd4": # big endian
1441 self.endian = ">"
1442 self.nano = False
1443 elif magic == b"\xd4\xc3\xb2\xa1": # little endian
1444 self.endian = "<"
1445 self.nano = False
1446 elif magic == b"\xa1\xb2\x3c\x4d": # big endian, nanosecond-precision
1447 self.endian = ">"
1448 self.nano = True
1449 elif magic == b"\x4d\x3c\xb2\xa1": # little endian, nanosecond-precision # noqa: E501
1450 self.endian = "<"
1451 self.nano = True
1452 else:
1453 raise Scapy_Exception(
1454 "Not a pcap capture file (bad magic: %r)" % magic
1455 )
1456 try:
1457 hdr = self.f.read(20)
1458 except (OSError, OverflowError, zlib.error) as e:
1459 warning(f"Pcap: {e}")
1460 raise Scapy_Exception("Invalid pcap file (corrupted stream)")
1461 if len(hdr) < 20:
1462 raise Scapy_Exception("Invalid pcap file (too short)")
1463 vermaj, vermin, tz, sig, snaplen, linktype = struct.unpack(
1464 self.endian + "HHIIII", hdr
1465 )
1466 self.linktype = linktype
1467 self.snaplen = snaplen
1469 def __enter__(self):
1470 # type: () -> RawPcapReader
1471 return self
1473 def __iter__(self):
1474 # type: () -> RawPcapReader
1475 return self
1477 def __next__(self):
1478 # type: () -> Tuple[bytes, RawPcapReader.PacketMetadata]
1479 """
1480 implement the iterator protocol on a set of packets in a pcap file
1481 """
1482 try:
1483 return self._read_packet()
1484 except EOFError:
1485 raise StopIteration
1487 def _read_packet(self, size=MTU):
1488 # type: (int) -> Tuple[bytes, RawPcapReader.PacketMetadata]
1489 """return a single packet read from the file as a tuple containing
1490 (pkt_data, pkt_metadata)
1492 raise EOFError when no more packets are available
1493 """
1494 try:
1495 hdr = self.f.read(16)
1496 if len(hdr) < 16:
1497 raise EOFError
1498 sec, usec, caplen, wirelen = struct.unpack(self.endian + "IIII", hdr)
1499 # A malicious caplen can be up to 4 GiB: bound each read,
1500 # truncate the packet, then skip the rest of the record so the
1501 # next pcap header can still be parsed.
1502 read_size = min(caplen, MTU * 4)
1503 read_data = self.f.read(read_size)
1504 data = read_data[:size]
1505 remaining = caplen - len(read_data)
1506 if remaining > 0:
1507 warning("Pcap: packet has been truncated")
1508 while remaining > 0:
1509 skipped = self.f.read(min(remaining, MTU * 4))
1510 if not skipped:
1511 break
1512 remaining -= len(skipped)
1513 except (OSError, OverflowError, zlib.error) as e:
1514 warning(f"Pcap: {e}")
1515 raise EOFError
1517 return (data,
1518 RawPcapReader.PacketMetadata(sec=sec, usec=usec,
1519 wirelen=wirelen, caplen=caplen))
1521 def read_packet(self, size=MTU):
1522 # type: (int) -> Packet
1523 raise Exception(
1524 "Cannot call read_packet() in RawPcapReader. Use "
1525 "_read_packet()"
1526 )
1528 def dispatch(self,
1529 callback # type: Callable[[Tuple[bytes, RawPcapReader.PacketMetadata]], Any] # noqa: E501
1530 ):
1531 # type: (...) -> None
1532 """call the specified callback routine for each packet read
1534 This is just a convenience function for the main loop
1535 that allows for easy launching of packet processing in a
1536 thread.
1537 """
1538 for p in self:
1539 callback(p)
1541 def _read_all(self, count=-1):
1542 # type: (int) -> List[Packet]
1543 """return a list of all packets in the pcap file
1544 """
1545 res = [] # type: List[Packet]
1546 while count != 0:
1547 count -= 1
1548 try:
1549 p = self.read_packet() # type: Packet
1550 except EOFError:
1551 break
1552 res.append(p)
1553 return res
1555 def recv(self, size=MTU):
1556 # type: (int) -> bytes
1557 """ Emulate a socket
1558 """
1559 return self._read_packet(size=size)[0]
1561 def fileno(self):
1562 # type: () -> int
1563 return -1 if WINDOWS else self.f.fileno()
1565 def close(self):
1566 # type: () -> None
1567 if isinstance(self.f, gzip.GzipFile):
1568 self.f.fileobj.close() # type: ignore
1569 self.f.close()
1570 if self.subproc is not None:
1571 # Reap the prefilter subprocess. The read pipe is already closed
1572 # above, so a still-running tcpdump gets a SIGTERM and we then
1573 # wait() to avoid a zombie; an already-finished one is just reaped.
1574 self.subproc.terminate()
1575 self.subproc.wait()
1576 self.subproc = None
1578 def __exit__(self, exc_type, exc_value, tracback):
1579 # type: (Optional[Any], Optional[Any], Optional[Any]) -> None
1580 self.close()
1582 # emulate SuperSocket
1583 @staticmethod
1584 def select(sockets, # type: List[SuperSocket]
1585 remain=None, # type: Optional[float]
1586 ):
1587 # type: (...) -> List[SuperSocket]
1588 return sockets
1591class PcapReader(RawPcapReader):
1592 def __init__(self, filename, fdesc=None, magic=None): # type: ignore
1593 # type: (str, IO[bytes], bytes) -> None
1594 RawPcapReader.__init__(self, filename, fdesc, magic)
1595 try:
1596 self.LLcls = conf.l2types.num2layer[
1597 self.linktype
1598 ] # type: Type[Packet]
1599 except KeyError:
1600 warning("PcapReader: unknown LL type [%i]/[%#x]. Using Raw packets" % (self.linktype, self.linktype)) # noqa: E501
1601 if conf.raw_layer is None:
1602 # conf.raw_layer is set on import
1603 import scapy.packet # noqa: F401
1604 self.LLcls = conf.raw_layer
1606 def __enter__(self):
1607 # type: () -> PcapReader
1608 return self
1610 def read_packet(self, size=MTU, **kwargs):
1611 # type: (int, **Any) -> Packet
1612 rp = super(PcapReader, self)._read_packet(size=size)
1613 if rp is None:
1614 raise EOFError
1615 s, pkt_info = rp
1617 try:
1618 p = self.LLcls(s, **kwargs) # type: Packet
1619 except KeyboardInterrupt:
1620 raise
1621 except Exception:
1622 if conf.debug_dissector:
1623 from scapy.sendrecv import debug
1624 debug.crashed_on = (self.LLcls, s)
1625 raise
1626 if conf.raw_layer is None:
1627 # conf.raw_layer is set on import
1628 import scapy.packet # noqa: F401
1629 p = conf.raw_layer(s)
1630 power = Decimal(10) ** Decimal(-9 if self.nano else -6)
1631 p.time = EDecimal(pkt_info.sec + power * pkt_info.usec)
1632 p.wirelen = pkt_info.wirelen
1633 return p
1635 def recv(self, size=MTU, **kwargs): # type: ignore
1636 # type: (int, **Any) -> Packet
1637 return self.read_packet(size=size, **kwargs)
1639 def __iter__(self):
1640 # type: () -> PcapReader
1641 return self
1643 def __next__(self): # type: ignore
1644 # type: () -> Packet
1645 try:
1646 return self.read_packet()
1647 except EOFError:
1648 raise StopIteration
1650 def read_all(self, count=-1):
1651 # type: (int) -> PacketList
1652 res = self._read_all(count)
1653 from scapy import plist
1654 return plist.PacketList(res, name=os.path.basename(self.filename))
1657class RawPcapNgReader(RawPcapReader):
1658 """A stateful pcapng reader. Each packet is returned as
1659 bytes.
1661 """
1663 alternative = RawPcapReader # type: Type[Any]
1665 PacketMetadata = collections.namedtuple("PacketMetadataNg", # type: ignore
1666 ["linktype", "tsresol",
1667 "tshigh", "tslow", "wirelen",
1668 "comments", "ifname", "direction",
1669 "process_information"])
1671 def __init__(self, filename, fdesc=None, magic=None): # type: ignore
1672 # type: (str, IO[bytes], bytes) -> None
1673 self.filename = filename
1674 self.f = fdesc
1675 # A list of (linktype, snaplen, tsresol); will be populated by IDBs.
1676 self.interfaces = [] # type: List[Tuple[int, int, Dict[str, Any]]]
1677 self.default_options = {
1678 "tsresol": 1000000
1679 }
1680 self.blocktypes: Dict[
1681 int,
1682 Callable[
1683 [bytes, int],
1684 Optional[Tuple[bytes, RawPcapNgReader.PacketMetadata]]
1685 ]] = {
1686 1: self._read_block_idb,
1687 2: self._read_block_pkt,
1688 3: self._read_block_spb,
1689 6: self._read_block_epb,
1690 10: self._read_block_dsb,
1691 0x80000001: self._read_block_pib,
1692 }
1693 self.endian = "!" # Will be overwritten by first SHB
1694 self.process_information = [] # type: List[Dict[str, Any]]
1695 self._tls_state = None # type: Optional[Tuple[Dict[str, bytes], bool]]
1697 if magic != b"\x0a\x0d\x0d\x0a": # PcapNg:
1698 raise Scapy_Exception(
1699 "Not a pcapng capture file (bad magic: %r)" % magic
1700 )
1702 try:
1703 self._read_block_shb()
1704 except (EOFError, OSError, OverflowError, zlib.error) as e:
1705 warning(f"PcapNg: {e}")
1706 raise Scapy_Exception(
1707 "The first SHB of the pcapng file is malformed !"
1708 )
1710 def _read_block(self, size=MTU):
1711 # type: (int) -> Optional[Tuple[bytes, RawPcapNgReader.PacketMetadata]] # noqa: E501
1712 try:
1713 blocktype = struct.unpack(self.endian + "I", self.f.read(4))[0]
1714 except struct.error:
1715 raise EOFError
1716 if blocktype == 0x0A0D0D0A:
1717 # This function updates the endianness based on the block content.
1718 self._read_block_shb()
1719 return None
1720 try:
1721 blocklen = struct.unpack(self.endian + "I", self.f.read(4))[0]
1722 except struct.error:
1723 warning("PcapNg: Error reading blocklen before block body")
1724 raise EOFError
1725 if blocklen < 12:
1726 warning("PcapNg: Invalid block length !")
1727 raise EOFError
1729 _block_body_length = blocklen - 12
1730 block = self.f.read(_block_body_length)
1731 if len(block) != _block_body_length:
1732 warning("PcapNg: Invalid Block body length (too short)")
1733 raise EOFError
1734 self._read_block_tail(blocklen)
1735 if blocktype in self.blocktypes:
1736 return self.blocktypes[blocktype](block, size)
1737 return None
1739 def _read_block_tail(self, blocklen):
1740 # type: (int) -> None
1741 if blocklen % 4:
1742 pad = self.f.read(-blocklen % 4)
1743 warning("PcapNg: bad blocklen %d (MUST be a multiple of 4. "
1744 "Ignored padding %r" % (blocklen, pad))
1745 try:
1746 if blocklen != struct.unpack(self.endian + 'I',
1747 self.f.read(4))[0]:
1748 raise EOFError("PcapNg: Invalid pcapng block (bad blocklen)")
1749 except struct.error:
1750 warning("PcapNg: Could not read blocklen after block body")
1751 raise EOFError
1753 def _read_block_shb(self):
1754 # type: () -> None
1755 """Section Header Block"""
1756 _blocklen = self.f.read(4)
1757 endian = self.f.read(4)
1758 if endian == b"\x1a\x2b\x3c\x4d":
1759 self.endian = ">"
1760 elif endian == b"\x4d\x3c\x2b\x1a":
1761 self.endian = "<"
1762 else:
1763 warning("PcapNg: Bad magic in Section Header Block"
1764 " (not a pcapng file?)")
1765 raise EOFError
1767 try:
1768 blocklen = struct.unpack(self.endian + "I", _blocklen)[0]
1769 except struct.error:
1770 warning("PcapNg: Could not read blocklen")
1771 raise EOFError
1772 if blocklen < 28:
1773 warning(f"PcapNg: Invalid Section Header Block length ({blocklen})!") # noqa: E501
1774 raise EOFError
1776 # Major version must be 1
1777 _major = self.f.read(2)
1778 try:
1779 major = struct.unpack(self.endian + "H", _major)[0]
1780 except struct.error:
1781 warning("PcapNg: Could not read major value")
1782 raise EOFError
1783 if major != 1:
1784 warning(f"PcapNg: SHB Major version {major} unsupported !")
1785 raise EOFError
1787 # Skip minor version & section length
1788 skipped = self.f.read(10)
1789 if len(skipped) != 10:
1790 warning("PcapNg: Could not read minor value & section length")
1791 raise EOFError
1793 _options_len = blocklen - 28
1794 options = self.f.read(_options_len)
1795 if len(options) != _options_len:
1796 raise Scapy_Exception("PcapNg: Invalid Section Header Block "
1797 " options (too short)")
1798 self._read_block_tail(blocklen)
1799 self._read_options(options)
1801 def _read_packet(self, size=MTU): # type: ignore
1802 # type: (int) -> Tuple[bytes, RawPcapNgReader.PacketMetadata]
1803 """Read blocks until it reaches either EOF or a packet, and
1804 returns None or (packet, (linktype, sec, usec, wirelen)),
1805 where packet is a string.
1807 """
1808 while True:
1809 try:
1810 res = self._read_block(size=size)
1811 except (OSError, OverflowError, zlib.error) as e:
1812 warning(f"PcapNg: {e}")
1813 raise EOFError
1814 if res is not None:
1815 return res
1817 def _read_options(self, options):
1818 # type: (bytes) -> Dict[int, Union[bytes, List[bytes]]]
1819 opts = dict() # type: Dict[int, Union[bytes, List[bytes]]]
1820 offset = 0
1821 while len(options) - offset >= 4:
1822 try:
1823 code, length = struct.unpack_from(
1824 self.endian + "HH", options, offset
1825 )
1826 except struct.error:
1827 warning("PcapNg: options header is too small "
1828 "%d !" % (len(options) - offset))
1829 raise EOFError
1830 value_offset = offset + 4
1831 if code != 0 and value_offset + length <= len(options):
1832 # https://www.ietf.org/archive/id/draft-tuexen-opsawg-pcapng-05.html#name-options-format
1833 if code in [1, 2988, 2989, 19372, 19373]:
1834 if code not in opts:
1835 opts[code] = []
1836 opts[code].append( # type: ignore
1837 options[value_offset:value_offset + length]
1838 )
1839 else:
1840 opts[code] = options[value_offset:value_offset + length]
1841 if code == 0:
1842 if length != 0:
1843 warning("PcapNg: invalid option "
1844 "length %d for end-of-option" % length)
1845 break
1846 offset = value_offset + length + (-length % 4)
1847 return opts
1849 def _read_block_idb(self, block, _):
1850 # type: (bytes, int) -> None
1851 """Interface Description Block"""
1852 # 2 bytes LinkType + 2 bytes Reserved
1853 # 4 bytes Snaplen
1854 options_raw = self._read_options(block[8:])
1855 options = self.default_options.copy() # type: Dict[str, Any]
1856 for c, v in options_raw.items():
1857 if isinstance(v, list):
1858 # Spec allows multiple occurrences (see
1859 # https://www.ietf.org/archive/id/draft-tuexen-opsawg-pcapng-05.html#section-4.2-8.6)
1860 # but does not define which to use. We take the first for
1861 # backward compatibility.
1862 v = v[0]
1863 if c == 9:
1864 length = len(v)
1865 if length == 1:
1866 tsresol = v[0]
1867 options["tsresol"] = (2 if tsresol & 128 else 10) ** (
1868 tsresol & 127
1869 )
1870 else:
1871 warning("PcapNg: invalid options "
1872 "length %d for IDB tsresol" % length)
1873 elif c == 2:
1874 options["name"] = v
1875 elif c == 1:
1876 options["comment"] = v
1877 try:
1878 interface: Tuple[int, int, Dict[str, Any]] = struct.unpack(
1879 self.endian + "HxxI",
1880 block[:8]
1881 ) + (options,)
1882 except struct.error:
1883 warning("PcapNg: IDB is too small %d/8 !" % len(block))
1884 raise EOFError
1885 self.interfaces.append(interface)
1887 def _check_interface_id(self, intid):
1888 # type: (int) -> bool
1889 """Whether this block names an interface the file has described."""
1890 tmp_len = len(self.interfaces)
1891 if intid >= tmp_len:
1892 warning("PcapNg: invalid interface id %d/%d" % (intid, tmp_len))
1893 return False
1894 return True
1896 def _read_block_epb(self, block, size):
1897 # type: (bytes, int) -> Optional[Tuple[bytes, RawPcapNgReader.PacketMetadata]] # noqa: E501
1898 """Enhanced Packet Block"""
1899 try:
1900 intid, tshigh, tslow, caplen, wirelen = struct.unpack(
1901 self.endian + "5I",
1902 block[:20],
1903 )
1904 except struct.error:
1905 warning("PcapNg: EPB is too small %d/20 !" % len(block))
1906 raise EOFError
1908 # Compute the options offset taking padding into account
1909 if caplen % 4:
1910 opt_offset = 20 + caplen + (-caplen) % 4
1911 else:
1912 opt_offset = 20 + caplen
1914 # Parse options
1915 options = self._read_options(block[opt_offset:])
1917 process_information = {}
1918 for code, value in options.items():
1919 # PCAPNG_EPB_PIB_INDEX, PCAPNG_EPB_E_PIB_INDEX
1920 if code in [0x8001, 0x8003]:
1921 try:
1922 proc_index = struct.unpack(
1923 self.endian + "I", value)[0] # type: ignore
1924 except struct.error:
1925 warning("PcapNg: EPB invalid proc index "
1926 "(expected 4 bytes, got %d) !" % len(value))
1927 raise EOFError
1928 if proc_index < len(self.process_information):
1929 key = "proc" if code == 0x8001 else "eproc"
1930 process_information[key] = self.process_information[proc_index]
1931 else:
1932 warning("PcapNg: EPB invalid process information index "
1933 "(%d/%d) !" % (proc_index, len(self.process_information)))
1935 comments = options.get(1, None)
1936 epb_flags_raw = options.get(2, None)
1937 if epb_flags_raw and isinstance(epb_flags_raw, bytes):
1938 try:
1939 epb_flags, = struct.unpack(self.endian + "I", epb_flags_raw)
1940 except struct.error:
1941 warning("PcapNg: EPB invalid flags size"
1942 "(expected 4 bytes, got %d) !" % len(epb_flags_raw))
1943 raise EOFError
1944 direction = epb_flags & 3
1946 else:
1947 direction = None
1949 if not self._check_interface_id(intid):
1950 return None
1951 ifname = self.interfaces[intid][2].get('name', None)
1953 return (block[20:20 + caplen][:size],
1954 RawPcapNgReader.PacketMetadata(linktype=self.interfaces[intid][0], # noqa: E501
1955 tsresol=self.interfaces[intid][2]['tsresol'], # noqa: E501
1956 tshigh=tshigh,
1957 tslow=tslow,
1958 wirelen=wirelen,
1959 ifname=ifname,
1960 direction=direction,
1961 process_information=process_information,
1962 comments=comments))
1964 def _read_block_spb(self, block, size):
1965 # type: (bytes, int) -> Optional[Tuple[bytes, RawPcapNgReader.PacketMetadata]] # noqa: E501
1966 """Simple Packet Block"""
1967 # "it MUST be assumed that all the Simple Packet Blocks have
1968 # been captured on the interface previously specified in the
1969 # first Interface Description Block."
1970 intid = 0
1971 if not self._check_interface_id(intid):
1972 return None
1974 try:
1975 wirelen, = struct.unpack(self.endian + "I", block[:4])
1976 except struct.error:
1977 warning("PcapNg: SPB is too small %d/4 !" % len(block))
1978 raise EOFError
1980 caplen = min(wirelen, self.interfaces[intid][1])
1981 return (block[4:4 + caplen][:size],
1982 RawPcapNgReader.PacketMetadata(linktype=self.interfaces[intid][0], # noqa: E501
1983 tsresol=self.interfaces[intid][2]['tsresol'], # noqa: E501
1984 tshigh=None,
1985 tslow=None,
1986 wirelen=wirelen,
1987 ifname=None,
1988 direction=None,
1989 process_information={},
1990 comments=None))
1992 def _read_block_pkt(self, block, size):
1993 # type: (bytes, int) -> Optional[Tuple[bytes, RawPcapNgReader.PacketMetadata]] # noqa: E501
1994 """(Obsolete) Packet Block"""
1995 try:
1996 intid, drops, tshigh, tslow, caplen, wirelen = struct.unpack(
1997 self.endian + "HH4I",
1998 block[:20],
1999 )
2000 except struct.error:
2001 warning("PcapNg: PKT is too small %d/20 !" % len(block))
2002 raise EOFError
2004 if not self._check_interface_id(intid):
2005 return None
2006 return (block[20:20 + caplen][:size],
2007 RawPcapNgReader.PacketMetadata(linktype=self.interfaces[intid][0], # noqa: E501
2008 tsresol=self.interfaces[intid][2]['tsresol'], # noqa: E501
2009 tshigh=tshigh,
2010 tslow=tslow,
2011 wirelen=wirelen,
2012 ifname=None,
2013 direction=None,
2014 process_information={},
2015 comments=None))
2017 def _read_block_dsb(self, block, size):
2018 # type: (bytes, int) -> None
2019 """Decryption Secrets Block"""
2021 # Parse the secrets type and length fields
2022 try:
2023 secrets_type, secrets_length = struct.unpack(
2024 self.endian + "II",
2025 block[:8],
2026 )
2027 block = block[8:]
2028 except struct.error:
2029 warning("PcapNg: DSB is too small %d!", len(block))
2030 raise EOFError
2032 # Compute the secrets length including the padding
2033 padded_secrets_length = secrets_length + (-secrets_length) % 4
2034 if len(block) < padded_secrets_length:
2035 warning("PcapNg: invalid DSB secrets length!")
2036 raise EOFError
2038 # Extract secrets data and options
2039 secrets_data = block[:padded_secrets_length][:secrets_length]
2040 if block[padded_secrets_length:]:
2041 warning("PcapNg: DSB options are not supported!")
2043 # TLS Key Log
2044 if secrets_type == 0x544c534b:
2045 if getattr(conf, "tls_sessions", False) is False:
2046 warning("PcapNg: TLS Key Log available, but "
2047 "the TLS layer is not loaded! Scapy won't be able "
2048 "to decrypt the packets.")
2049 else:
2050 from scapy.layers.tls.session import parse_nss_keys
2052 try:
2053 keys = parse_nss_keys(secrets_data.decode())
2054 except UnicodeDecodeError as ex:
2055 warning("Cannot read NSS Key Log: %s", str(ex))
2056 keys = {}
2057 if not keys:
2058 warning("PcapNg: invalid TLS Key Log in DSB!")
2059 else:
2060 # Note: these attributes are only available when the TLS
2061 # layer is loaded.
2062 if self._tls_state is None:
2063 self._tls_state = (
2064 conf.tls_nss_keys,
2065 conf.tls_session_enable,
2066 )
2067 conf.tls_nss_keys = keys
2068 conf.tls_session_enable = True
2069 else:
2070 warning("PcapNg: Unknown DSB secrets type (0x%x)!", secrets_type)
2072 def close(self):
2073 # type: () -> None
2074 if self._tls_state is not None:
2075 conf.tls_nss_keys, conf.tls_session_enable = self._tls_state
2076 self._tls_state = None
2077 RawPcapReader.close(self)
2079 def _read_block_pib(self, block, _):
2080 # type: (bytes, int) -> None
2081 """Apple Process Information Block"""
2083 # Get the Process ID
2084 try:
2085 dpeb_pid = struct.unpack(self.endian + "I", block[:4])[0]
2086 process_information = {"id": dpeb_pid}
2087 block = block[4:]
2088 except struct.error:
2089 warning("PcapNg: DPEB is too small (%d). Cannot get PID!",
2090 len(block))
2091 raise EOFError
2093 # Get Options
2094 options = self._read_options(block)
2095 for code, value in options.items():
2096 if code == 2:
2097 process_information["name"] = value.decode( # type: ignore
2098 "ascii", "backslashreplace")
2099 elif code == 4:
2100 if len(value) == 16:
2101 process_information["uuid"] = str(UUID(bytes=value)) # type: ignore
2102 else:
2103 warning("PcapNg: DPEB UUID length is invalid (%d)!",
2104 len(value))
2106 # Store process information
2107 self.process_information.append(process_information)
2110class PcapNgReader(RawPcapNgReader, PcapReader):
2112 alternative = PcapReader
2114 def __init__(self, filename, fdesc=None, magic=None): # type: ignore
2115 # type: (str, IO[bytes], bytes) -> None
2116 RawPcapNgReader.__init__(self, filename, fdesc, magic)
2118 def __enter__(self):
2119 # type: () -> PcapNgReader
2120 return self
2122 def read_packet(self, size=MTU, **kwargs):
2123 # type: (int, **Any) -> Packet
2124 rp = super(PcapNgReader, self)._read_packet(size=size)
2125 if rp is None:
2126 raise EOFError
2127 s, (linktype, tsresol, tshigh, tslow, wirelen, comments, ifname, direction, process_information) = rp # noqa: E501
2128 try:
2129 cls = conf.l2types.num2layer[linktype] # type: Type[Packet]
2130 p = cls(s, **kwargs) # type: Packet
2131 except KeyboardInterrupt:
2132 raise
2133 except Exception:
2134 if conf.debug_dissector:
2135 raise
2136 if conf.raw_layer is None:
2137 # conf.raw_layer is set on import
2138 import scapy.packet # noqa: F401
2139 p = conf.raw_layer(s)
2140 if tshigh is not None:
2141 p.time = EDecimal((tshigh << 32) + tslow) / tsresol
2142 p.wirelen = wirelen
2143 p.comments = comments
2144 p.direction = direction
2145 p.process_information = process_information.copy()
2146 if ifname is not None:
2147 p.sniffed_on = ifname.decode('utf-8', 'backslashreplace')
2148 return p
2150 def recv(self, size: int = MTU, **kwargs: Any) -> 'Packet': # type: ignore
2151 return self.read_packet(size=size, **kwargs)
2154class GenericPcapWriter(object):
2155 nano = False
2156 linktype: int
2158 def _write_header(self, pkt):
2159 # type: (Optional[Union[Packet, bytes]]) -> None
2160 raise NotImplementedError
2162 def _write_packet(self,
2163 packet, # type: Union[bytes, Packet]
2164 linktype, # type: int
2165 sec=None, # type: Optional[float]
2166 usec=None, # type: Optional[int]
2167 caplen=None, # type: Optional[int]
2168 wirelen=None, # type: Optional[int]
2169 ifname=None, # type: Optional[bytes]
2170 direction=None, # type: Optional[int]
2171 comments=None, # type: Optional[List[bytes]]
2172 ):
2173 # type: (...) -> None
2174 raise NotImplementedError
2176 def _get_time(self,
2177 packet, # type: Union[bytes, Packet]
2178 sec, # type: Optional[float]
2179 usec # type: Optional[int]
2180 ):
2181 # type: (...) -> Tuple[float, int]
2182 if hasattr(packet, "time"):
2183 if sec is None:
2184 packet_time = packet.time
2185 tmp = int(packet_time)
2186 usec = int(round((packet_time - tmp) *
2187 (1000000000 if self.nano else 1000000)))
2188 sec = float(packet_time)
2189 if sec is not None and usec is None:
2190 usec = 0
2191 return sec, usec # type: ignore
2193 def write_header(self, pkt):
2194 # type: (Optional[Union[Packet, bytes]]) -> None
2195 if not hasattr(self, 'linktype'):
2196 try:
2197 if pkt is None or isinstance(pkt, bytes):
2198 # Can't guess LL
2199 raise KeyError
2200 self.linktype = conf.l2types.layer2num[
2201 pkt.__class__
2202 ]
2203 except KeyError:
2204 msg = "%s: unknown LL type for %s. Using type 1 (Ethernet)"
2205 warning(msg, self.__class__.__name__, pkt.__class__.__name__)
2206 self.linktype = DLT_EN10MB
2207 self._write_header(pkt)
2209 def write_packet(self,
2210 packet, # type: Union[bytes, Packet]
2211 sec=None, # type: Optional[float]
2212 usec=None, # type: Optional[int]
2213 caplen=None, # type: Optional[int]
2214 wirelen=None, # type: Optional[int]
2215 ):
2216 # type: (...) -> None
2217 """
2218 Writes a single packet to the pcap file.
2220 :param packet: Packet, or bytes for a single packet
2221 :type packet: scapy.packet.Packet or bytes
2222 :param sec: time the packet was captured, in seconds since epoch. If
2223 not supplied, defaults to now.
2224 :type sec: float
2225 :param usec: If ``nano=True``, then number of nanoseconds after the
2226 second that the packet was captured. If ``nano=False``,
2227 then the number of microseconds after the second the
2228 packet was captured. If ``sec`` is not specified,
2229 this value is ignored.
2230 :type usec: int or long
2231 :param caplen: The length of the packet in the capture file. If not
2232 specified, uses ``len(raw(packet))``.
2233 :type caplen: int
2234 :param wirelen: The length of the packet on the wire. If not
2235 specified, tries ``packet.wirelen``, otherwise uses
2236 ``caplen``.
2237 :type wirelen: int
2238 :return: None
2239 :rtype: None
2240 """
2241 f_sec, usec = self._get_time(packet, sec, usec)
2243 rawpkt = bytes_encode(packet)
2244 caplen = len(rawpkt) if caplen is None else caplen
2246 if wirelen is None:
2247 if hasattr(packet, "wirelen"):
2248 wirelen = packet.wirelen
2249 if wirelen is None:
2250 wirelen = caplen
2252 comments = getattr(packet, "comments", None)
2253 ifname = getattr(packet, "sniffed_on", None)
2254 direction = getattr(packet, "direction", None)
2255 if not isinstance(packet, bytes):
2256 # The class may not be bound to any linktype (e.g. conf.raw_layer),
2257 # in which case fall back to the one write_header() settled on.
2258 linktype: int = conf.l2types.layer2num.get(
2259 packet.__class__,
2260 self.linktype,
2261 )
2262 else:
2263 linktype = self.linktype
2264 if ifname is not None:
2265 ifname = str(ifname).encode('utf-8')
2266 self._write_packet(
2267 rawpkt,
2268 sec=f_sec, usec=usec,
2269 caplen=caplen, wirelen=wirelen,
2270 ifname=ifname,
2271 direction=direction,
2272 linktype=linktype,
2273 comments=comments,
2274 )
2277class GenericRawPcapWriter(GenericPcapWriter):
2278 header_present = False
2279 nano = False
2280 sync = False
2281 f = None # type: Union[IO[bytes], gzip.GzipFile]
2283 def fileno(self):
2284 # type: () -> int
2285 return -1 if WINDOWS else self.f.fileno()
2287 def flush(self):
2288 # type: () -> Optional[Any]
2289 return self.f.flush()
2291 def close(self):
2292 # type: () -> Optional[Any]
2293 if not self.header_present:
2294 self.write_header(None)
2295 return self.f.close()
2297 def __enter__(self):
2298 # type: () -> GenericRawPcapWriter
2299 return self
2301 def __exit__(self, exc_type, exc_value, tracback):
2302 # type: (Optional[Any], Optional[Any], Optional[Any]) -> None
2303 self.flush()
2304 self.close()
2306 def write(self, pkt):
2307 # type: (Union[_PacketIterable, bytes]) -> None
2308 """
2309 Writes a Packet, a SndRcvList object, or bytes to a pcap file.
2311 :param pkt: Packet(s) to write (one record for each Packet), or raw
2312 bytes to write (as one record).
2313 :type pkt: iterable[scapy.packet.Packet], scapy.packet.Packet or bytes
2314 """
2315 if isinstance(pkt, bytes):
2316 if not self.header_present:
2317 self.write_header(pkt)
2318 self.write_packet(pkt)
2319 else:
2320 # Import here to avoid circular dependency
2321 from scapy.supersocket import IterSocket
2322 for p in IterSocket(pkt).iter:
2323 if not self.header_present:
2324 self.write_header(p)
2326 if not isinstance(p, bytes) and \
2327 self.linktype != conf.l2types.get(type(p), None):
2328 warning("Inconsistent linktypes detected!"
2329 " The resulting file might contain"
2330 " invalid packets."
2331 )
2333 self.write_packet(p)
2336class RawPcapWriter(GenericRawPcapWriter):
2337 """A stream PCAP writer with more control than wrpcap()"""
2339 def __init__(self,
2340 filename, # type: Union[IO[bytes], str]
2341 linktype=None, # type: Optional[int]
2342 gz=False, # type: bool
2343 endianness="", # type: str
2344 append=False, # type: bool
2345 sync=False, # type: bool
2346 nano=False, # type: bool
2347 snaplen=MTU, # type: int
2348 bufsz=4096, # type: int
2349 ):
2350 # type: (...) -> None
2351 """
2352 :param filename: the name of the file to write packets to, or an open,
2353 writable file-like object.
2354 :param linktype: force linktype to a given value. If None, linktype is
2355 taken from the first writer packet
2356 :param gz: compress the capture on the fly
2357 :param endianness: force an endianness (little:"<", big:">").
2358 Default is native
2359 :param append: append packets to the capture file instead of
2360 truncating it
2361 :param sync: do not bufferize writes to the capture file
2362 :param nano: use nanosecond-precision (requires libpcap >= 1.5.0)
2364 """
2366 if linktype:
2367 self.linktype = linktype
2368 self.snaplen = snaplen
2369 self.append = append
2370 self.gz = gz
2371 self.endian = endianness
2372 self.sync = sync
2373 self.nano = nano
2374 if sync:
2375 bufsz = 0
2377 if isinstance(filename, str):
2378 self.filename = filename
2379 if gz:
2380 self.f = cast(_ByteStream, gzip.open(
2381 filename, append and "ab" or "wb", 9
2382 ))
2383 else:
2384 self.f = open(filename, append and "ab" or "wb", bufsz)
2385 else:
2386 self.f = filename
2387 self.filename = getattr(filename, "name", "No name")
2389 def _write_header(self, pkt):
2390 # type: (Optional[Union[Packet, bytes]]) -> None
2391 self.header_present = True
2393 if self.append:
2394 # Even if prone to race conditions, this seems to be
2395 # safest way to tell whether the header is already present
2396 # because we have to handle compressed streams that
2397 # are not as flexible as basic files
2398 if self.gz:
2399 g = gzip.open(self.filename, "rb") # type: _ByteStream
2400 else:
2401 g = open(self.filename, "rb")
2402 try:
2403 if g.read(16):
2404 return
2405 finally:
2406 g.close()
2408 if not hasattr(self, 'linktype'):
2409 raise ValueError(
2410 "linktype could not be guessed. "
2411 "Please pass a linktype while creating the writer"
2412 )
2414 self.f.write(struct.pack(self.endian + "IHHIIII", 0xa1b23c4d if self.nano else 0xa1b2c3d4, # noqa: E501
2415 2, 4, 0, 0, self.snaplen, self.linktype))
2416 self.f.flush()
2418 def _write_packet(self,
2419 packet, # type: Union[bytes, Packet]
2420 linktype, # type: int
2421 sec=None, # type: Optional[float]
2422 usec=None, # type: Optional[int]
2423 caplen=None, # type: Optional[int]
2424 wirelen=None, # type: Optional[int]
2425 ifname=None, # type: Optional[bytes]
2426 direction=None, # type: Optional[int]
2427 comments=None, # type: Optional[List[bytes]]
2428 ):
2429 # type: (...) -> None
2430 """
2431 Writes a single packet to the pcap file.
2433 :param packet: bytes for a single packet
2434 :type packet: bytes
2435 :param linktype: linktype value associated with the packet
2436 :type linktype: int
2437 :param sec: time the packet was captured, in seconds since epoch. If
2438 not supplied, defaults to now.
2439 :type sec: float
2440 :param usec: not used with pcapng
2441 packet was captured
2442 :type usec: int or long
2443 :param caplen: The length of the packet in the capture file. If not
2444 specified, uses ``len(packet)``.
2445 :type caplen: int
2446 :param wirelen: The length of the packet on the wire. If not
2447 specified, uses ``caplen``.
2448 :type wirelen: int
2449 :return: None
2450 :rtype: None
2451 """
2452 if caplen is None:
2453 caplen = len(packet)
2454 if wirelen is None:
2455 wirelen = caplen
2456 if sec is None or usec is None:
2457 t = time.time()
2458 it = int(t)
2459 if sec is None:
2460 sec = it
2461 usec = int(round((t - it) *
2462 (1000000000 if self.nano else 1000000)))
2463 elif usec is None:
2464 usec = 0
2466 self.f.write(struct.pack(self.endian + "IIII",
2467 int(sec), usec, caplen, wirelen))
2468 self.f.write(bytes(packet))
2469 if self.sync:
2470 self.f.flush()
2473class RawPcapNgWriter(GenericRawPcapWriter):
2474 """A stream pcapng writer with more control than wrpcapng()"""
2476 def __init__(self,
2477 filename, # type: str
2478 ):
2479 # type: (...) -> None
2481 self.header_present = False
2482 self.tsresol = 1000000
2483 # A dict to keep if_name to IDB id mapping.
2484 # unknown if_name(None) id=0
2485 self.interfaces2id: Dict[Optional[bytes], int] = {None: 0}
2487 # tcpdump only support little-endian in PCAPng files
2488 self.endian = "<"
2489 self.endian_magic = b"\x4d\x3c\x2b\x1a"
2491 self.filename = filename
2492 self.f = open(filename, "wb", 4096)
2494 def _get_time(self,
2495 packet, # type: Union[bytes, Packet]
2496 sec, # type: Optional[float]
2497 usec # type: Optional[int]
2498 ):
2499 # type: (...) -> Tuple[float, int]
2500 if hasattr(packet, "time"):
2501 if sec is None:
2502 sec = float(packet.time)
2504 if usec is None:
2505 usec = 0
2507 return sec, usec # type: ignore
2509 def _add_padding(self, raw_data):
2510 # type: (bytes) -> bytes
2511 raw_data += ((-len(raw_data)) % 4) * b"\x00"
2512 return raw_data
2514 def build_block(self, block_type, block_body, options=None):
2515 # type: (bytes, bytes, Optional[bytes]) -> bytes
2517 # Pad Block Body to 32 bits
2518 block_body = self._add_padding(block_body)
2520 if options:
2521 block_body += options
2523 # An empty block is 12 bytes long
2524 block_total_length = 12 + len(block_body)
2526 # Block Type
2527 block = block_type
2528 # Block Total Length$
2529 block += struct.pack(self.endian + "I", block_total_length)
2530 # Block Body
2531 block += block_body
2532 # Block Total Length$
2533 block += struct.pack(self.endian + "I", block_total_length)
2535 return block
2537 def _write_header(self, pkt):
2538 # type: (Optional[Union[Packet, bytes]]) -> None
2539 if not self.header_present:
2540 self.header_present = True
2541 self._write_block_shb()
2542 self._write_block_idb(linktype=self.linktype)
2544 def _write_block_shb(self):
2545 # type: () -> None
2547 # Block Type
2548 block_type = b"\x0A\x0D\x0D\x0A"
2549 # Byte-Order Magic
2550 block_shb = self.endian_magic
2551 # Major Version
2552 block_shb += struct.pack(self.endian + "H", 1)
2553 # Minor Version
2554 block_shb += struct.pack(self.endian + "H", 0)
2555 # Section Length
2556 block_shb += struct.pack(self.endian + "q", -1)
2558 self.f.write(self.build_block(block_type, block_shb))
2560 def _write_block_idb(self,
2561 linktype, # type: int
2562 ifname=None # type: Optional[bytes]
2563 ):
2564 # type: (...) -> None
2566 # Block Type
2567 block_type = struct.pack(self.endian + "I", 1)
2568 # LinkType
2569 block_idb = struct.pack(self.endian + "H", linktype)
2570 # Reserved
2571 block_idb += struct.pack(self.endian + "H", 0)
2572 # SnapLen
2573 block_idb += struct.pack(self.endian + "I", 262144)
2575 # if_name option
2576 opts = None
2577 if ifname is not None:
2578 opts = struct.pack(self.endian + "HH", 2, len(ifname))
2579 # Pad Option Value to 32 bits
2580 opts += self._add_padding(ifname)
2581 opts += struct.pack(self.endian + "HH", 0, 0)
2583 self.f.write(self.build_block(block_type, block_idb, options=opts))
2585 def _write_block_spb(self, raw_pkt):
2586 # type: (bytes) -> None
2588 # Block Type
2589 block_type = struct.pack(self.endian + "I", 3)
2590 # Original Packet Length
2591 block_spb = struct.pack(self.endian + "I", len(raw_pkt))
2592 # Packet Data
2593 block_spb += raw_pkt
2595 self.f.write(self.build_block(block_type, block_spb))
2597 def _write_block_epb(self,
2598 raw_pkt, # type: bytes
2599 ifid, # type: int
2600 timestamp=None, # type: Optional[Union[EDecimal, float]] # noqa: E501
2601 caplen=None, # type: Optional[int]
2602 orglen=None, # type: Optional[int]
2603 comments=None, # type: Optional[List[bytes]]
2604 flags=None, # type: Optional[int]
2605 ):
2606 # type: (...) -> None
2608 if timestamp:
2609 tmp_ts = int(timestamp * self.tsresol)
2610 ts_high = tmp_ts >> 32
2611 ts_low = tmp_ts & 0xFFFFFFFF
2612 else:
2613 ts_high = ts_low = 0
2615 if not caplen:
2616 caplen = len(raw_pkt)
2618 if not orglen:
2619 orglen = len(raw_pkt)
2621 # Block Type
2622 block_type = struct.pack(self.endian + "I", 6)
2623 # Interface ID
2624 block_epb = struct.pack(self.endian + "I", ifid)
2625 # Timestamp (High)
2626 block_epb += struct.pack(self.endian + "I", ts_high)
2627 # Timestamp (Low)
2628 block_epb += struct.pack(self.endian + "I", ts_low)
2629 # Captured Packet Length
2630 block_epb += struct.pack(self.endian + "I", caplen)
2631 # Original Packet Length
2632 block_epb += struct.pack(self.endian + "I", orglen)
2633 # Packet Data
2634 block_epb += raw_pkt
2636 # Options
2637 opts = b''
2638 if comments and len(comments):
2639 for c in comments:
2640 comment = bytes_encode(c)
2641 opts += struct.pack(self.endian + "HH", 1, len(comment))
2642 # Pad Option Value to 32 bits
2643 opts += self._add_padding(comment)
2644 if type(flags) == int:
2645 opts += struct.pack(self.endian + "HH", 2, 4)
2646 opts += struct.pack(self.endian + "I", flags)
2647 if opts:
2648 opts += struct.pack(self.endian + "HH", 0, 0)
2650 self.f.write(self.build_block(block_type, block_epb,
2651 options=opts))
2653 def _write_packet(self, # type: ignore
2654 packet, # type: bytes
2655 linktype, # type: int
2656 sec=None, # type: Optional[float]
2657 usec=None, # type: Optional[int]
2658 caplen=None, # type: Optional[int]
2659 wirelen=None, # type: Optional[int]
2660 ifname=None, # type: Optional[bytes]
2661 direction=None, # type: Optional[int]
2662 comments=None, # type: Optional[List[bytes]]
2663 ):
2664 # type: (...) -> None
2665 """
2666 Writes a single packet to the pcap file.
2668 :param packet: bytes for a single packet
2669 :type packet: bytes
2670 :param linktype: linktype value associated with the packet
2671 :type linktype: int
2672 :param sec: time the packet was captured, in seconds since epoch. If
2673 not supplied, defaults to now.
2674 :type sec: float
2675 :param caplen: The length of the packet in the capture file. If not
2676 specified, uses ``len(packet)``.
2677 :type caplen: int
2678 :param wirelen: The length of the packet on the wire. If not
2679 specified, uses ``caplen``.
2680 :type wirelen: int
2681 :param comment: UTF-8 string containing human-readable comment text
2682 that is associated to the current block. Line separators
2683 SHOULD be a carriage-return + linefeed ('\r\n') or
2684 just linefeed ('\n'); either form may appear and
2685 be considered a line separator. The string is not
2686 zero-terminated.
2687 :type bytes
2688 :param ifname: UTF-8 string containing the
2689 name of the device used to capture data.
2690 The string is not zero-terminated.
2691 :type bytes
2692 :param direction: 0 = information not available,
2693 1 = inbound,
2694 2 = outbound
2695 :type int
2696 :return: None
2697 :rtype: None
2698 """
2699 if caplen is None:
2700 caplen = len(packet)
2701 if wirelen is None:
2702 wirelen = caplen
2704 ifid = self.interfaces2id.get(ifname, None)
2705 if ifid is None:
2706 ifid = max(self.interfaces2id.values()) + 1
2707 self.interfaces2id[ifname] = ifid
2708 self._write_block_idb(linktype=linktype, ifname=ifname)
2710 # EPB flags (32 bits).
2711 # currently only direction is implemented (least 2 significant bits)
2712 if type(direction) == int:
2713 flags = direction & 0x3
2714 else:
2715 flags = None
2717 self._write_block_epb(packet, timestamp=sec, caplen=caplen,
2718 orglen=wirelen, comments=comments, ifid=ifid, flags=flags)
2719 if self.sync:
2720 self.f.flush()
2723class PcapWriter(RawPcapWriter):
2724 """A stream PCAP writer with more control than wrpcap()"""
2725 pass
2728class PcapNgWriter(RawPcapNgWriter):
2729 """A stream pcapng writer with more control than wrpcapng()"""
2731 def _get_time(self,
2732 packet, # type: Union[bytes, Packet]
2733 sec, # type: Optional[float]
2734 usec # type: Optional[int]
2735 ):
2736 # type: (...) -> Tuple[float, int]
2737 if hasattr(packet, "time"):
2738 if sec is None:
2739 sec = float(packet.time)
2741 if usec is None:
2742 usec = 0
2744 return sec, usec # type: ignore
2747@conf.commands.register
2748def rderf(filename, count=-1):
2749 # type: (Union[IO[bytes], str], int) -> PacketList
2750 """Read a ERF file and return a packet list
2752 :param count: read only <count> packets
2753 """
2754 with ERFEthernetReader(filename) as fdesc:
2755 return fdesc.read_all(count=count)
2758class ERFEthernetReader_metaclass(PcapReader_metaclass):
2759 def __call__(cls, filename):
2760 # type: (Union[IO[bytes], str]) -> Any
2761 i = cls.__new__(cls, cls.__name__, cls.__bases__, cls.__dict__) # type: ignore
2762 filename, fdesc = cls.open(filename)
2763 try:
2764 i.__init__(filename, fdesc)
2765 return i
2766 except (Scapy_Exception, EOFError):
2767 pass
2769 if "alternative" in cls.__dict__:
2770 cls = cls.__dict__["alternative"]
2771 i = cls.__new__(
2772 cls,
2773 cls.__name__,
2774 cls.__bases__,
2775 cls.__dict__ # type: ignore
2776 )
2777 try:
2778 i.__init__(filename, fdesc)
2779 return i
2780 except (Scapy_Exception, EOFError):
2781 pass
2783 raise Scapy_Exception("Not a supported capture file")
2785 @staticmethod
2786 def open(fname # type: ignore
2787 ):
2788 # type: (...) -> Tuple[str, _ByteStream]
2789 """Open (if necessary) filename"""
2790 if isinstance(fname, str):
2791 filename = fname
2792 try:
2793 with gzip.open(filename, "rb") as tmp:
2794 tmp.read(1)
2795 fdesc = gzip.open(filename, "rb") # type: _ByteStream
2796 except IOError:
2797 fdesc = open(filename, "rb")
2799 else:
2800 fdesc = fname
2801 filename = getattr(fdesc, "name", "No name")
2802 return filename, fdesc
2805class ERFEthernetReader(PcapReader,
2806 metaclass=ERFEthernetReader_metaclass):
2808 def __init__(self, filename, fdesc=None): # type: ignore
2809 # type: (Union[IO[bytes], str], IO[bytes]) -> None
2810 self.filename = filename # type: ignore
2811 self.f = fdesc
2812 self.power = Decimal(10) ** Decimal(-9)
2814 # time is in 64-bits Endace's format which can be see here:
2815 # https://www.endace.com/erf-extensible-record-format-types.pdf
2816 def _convert_erf_timestamp(self, t):
2817 # type: (int) -> EDecimal
2818 sec = t >> 32
2819 frac_sec = t & 0xffffffff
2820 frac_sec *= 10**9
2821 frac_sec += (frac_sec & 0x80000000) << 1
2822 frac_sec >>= 32
2823 return EDecimal(sec + self.power * frac_sec)
2825 # The details of ERF Packet format can be see here:
2826 # https://www.endace.com/erf-extensible-record-format-types.pdf
2827 def read_packet(self, size=MTU, **kwargs):
2828 # type: (int, **Any) -> Packet
2830 # General ERF Header have exactly 16 bytes
2831 hdr = self.f.read(16)
2832 if len(hdr) < 16:
2833 raise EOFError
2835 # The timestamp is in little-endian byte-order.
2836 time = struct.unpack('<Q', hdr[:8])[0]
2837 # The rest is in big-endian byte-order.
2838 # Ignoring flags and lctr (loss counter) since they are ERF specific
2839 # header fields which Packet object does not support.
2840 type, _, rlen, _, wlen = struct.unpack('>BBHHH', hdr[8:])
2841 # Check if the type != 0x02, type Ethernet
2842 if type & 0x02 == 0:
2843 raise Scapy_Exception("Invalid ERF Type (Not TYPE_ETH)")
2845 # If there are extended headers, ignore it because Packet object does
2846 # not support it. Extended headers size is 8 bytes before the payload.
2847 if type & 0x80:
2848 _ = self.f.read(8)
2849 s = self.f.read(max(0, rlen - 24))
2850 else:
2851 s = self.f.read(max(0, rlen - 16))
2853 # Ethernet has 2 bytes of padding containing `offset` and `pad`. Both
2854 # of the fields are disregarded by Endace.
2855 pb = s[2:size]
2856 from scapy.layers.l2 import Ether
2857 try:
2858 p = Ether(pb, **kwargs) # type: Packet
2859 except KeyboardInterrupt:
2860 raise
2861 except Exception:
2862 if conf.debug_dissector:
2863 from scapy.sendrecv import debug
2864 debug.crashed_on = (Ether, s)
2865 raise
2866 if conf.raw_layer is None:
2867 # conf.raw_layer is set on import
2868 import scapy.packet # noqa: F401
2869 p = conf.raw_layer(s)
2871 p.time = self._convert_erf_timestamp(time)
2872 p.wirelen = wlen
2874 return p
2877@conf.commands.register
2878def wrerf(filename, # type: Union[IO[bytes], str]
2879 pkt, # type: _PacketIterable
2880 *args, # type: Any
2881 **kargs # type: Any
2882 ):
2883 # type: (...) -> None
2884 """Write a list of packets to a ERF file
2886 :param filename: the name of the file to write packets to, or an open,
2887 writable file-like object. The file descriptor will be
2888 closed at the end of the call, so do not use an object you
2889 do not want to close (e.g., running wrerf(sys.stdout, [])
2890 in interactive mode will crash Scapy).
2891 :param gz: set to 1 to save a gzipped capture
2892 :param append: append packets to the capture file instead of
2893 truncating it
2894 :param sync: do not bufferize writes to the capture file
2895 """
2896 with ERFEthernetWriter(filename, *args, **kargs) as fdesc:
2897 fdesc.write(pkt)
2900class ERFEthernetWriter(PcapWriter):
2901 """A stream ERF Ethernet writer with more control than wrerf()"""
2903 def __init__(self,
2904 filename, # type: Union[IO[bytes], str]
2905 gz=False, # type: bool
2906 append=False, # type: bool
2907 sync=False, # type: bool
2908 ):
2909 # type: (...) -> None
2910 """
2911 :param filename: the name of the file to write packets to, or an open,
2912 writable file-like object.
2913 :param gz: compress the capture on the fly
2914 :param append: append packets to the capture file instead of
2915 truncating it
2916 :param sync: do not bufferize writes to the capture file
2917 """
2918 super(ERFEthernetWriter, self).__init__(filename,
2919 gz=gz,
2920 append=append,
2921 sync=sync)
2923 def write(self, pkt): # type: ignore
2924 # type: (_PacketIterable) -> None
2925 """
2926 Writes a Packet, a SndRcvList object, or bytes to a ERF file.
2928 :param pkt: Packet(s) to write (one record for each Packet)
2929 :type pkt: iterable[scapy.packet.Packet], scapy.packet.Packet
2930 """
2931 # Import here to avoid circular dependency
2932 from scapy.supersocket import IterSocket
2933 for p in IterSocket(pkt).iter:
2934 self.write_packet(p)
2936 def write_packet(self, pkt): # type: ignore
2937 # type: (Packet) -> None
2939 if hasattr(pkt, "time"):
2940 sec = int(pkt.time)
2941 usec = int((int(round((pkt.time - sec) * 10**9)) << 32) / 10**9)
2942 t = (sec << 32) + usec
2943 else:
2944 t = int(time.time()) << 32
2946 # There are 16 bytes of headers + 2 bytes of padding before the packets
2947 # payload.
2948 rlen = len(pkt) + 18
2950 if hasattr(pkt, "wirelen"):
2951 wirelen = pkt.wirelen
2952 if wirelen is None:
2953 wirelen = rlen
2955 self.f.write(struct.pack("<Q", t))
2956 self.f.write(struct.pack(">BBHHHH", 2, 0, rlen, 0, wirelen, 0))
2957 self.f.write(bytes(pkt))
2958 self.f.flush()
2960 def close(self):
2961 # type: () -> Optional[Any]
2962 return self.f.close()
2965@conf.commands.register
2966def import_hexcap(input_string=None):
2967 # type: (Optional[str]) -> bytes
2968 """Imports a tcpdump like hexadecimal view
2970 e.g: exported via hexdump() or tcpdump or wireshark's "export as hex"
2972 :param input_string: String containing the hexdump input to parse. If None,
2973 read from standard input.
2974 """
2975 re_extract_hexcap = re.compile(r"^((0x)?[0-9a-fA-F]{2,}[ :\t]{,3}|) *(([0-9a-fA-F]{2} {,2}){,16})") # noqa: E501
2976 p = ""
2977 try:
2978 if input_string:
2979 input_function = StringIO(input_string).readline
2980 else:
2981 input_function = input
2982 while True:
2983 line = input_function().strip()
2984 if not line:
2985 break
2986 try:
2987 p += re_extract_hexcap.match(line).groups()[2] # type: ignore
2988 except Exception:
2989 warning("Parsing error during hexcap")
2990 continue
2991 except EOFError:
2992 pass
2994 p = p.replace(" ", "")
2995 return hex_bytes(p)
2998@conf.commands.register
2999def wireshark(pktlist, wait=False, **kwargs):
3000 # type: (List[Packet], bool, **Any) -> Optional[Any]
3001 """
3002 Runs Wireshark on a list of packets.
3004 See :func:`tcpdump` for more parameter description.
3006 Note: this defaults to wait=False, to run Wireshark in the background.
3007 """
3008 return tcpdump(pktlist, prog=conf.prog.wireshark, wait=wait, **kwargs)
3011@conf.commands.register
3012def tdecode(
3013 pktlist, # type: Union[IO[bytes], None, str, _PacketIterable]
3014 args=None, # type: Optional[List[str]]
3015 **kwargs # type: Any
3016):
3017 # type: (...) -> Any
3018 """
3019 Run tshark on a list of packets.
3021 :param args: If not specified, defaults to ``tshark -V``.
3023 See :func:`tcpdump` for more parameters.
3024 """
3025 if args is None:
3026 args = ["-V"]
3027 return tcpdump(pktlist, prog=conf.prog.tshark, args=args, **kwargs)
3030def _guess_linktype_name(value):
3031 # type: (int) -> str
3032 """Guess the DLT name from its value."""
3033 from scapy.libs.winpcapy import pcap_datalink_val_to_name
3034 return cast(bytes, pcap_datalink_val_to_name(value)).decode()
3037def _guess_linktype_value(name):
3038 # type: (str) -> int
3039 """Guess the value of a DLT name."""
3040 from scapy.libs.winpcapy import pcap_datalink_name_to_val
3041 val = cast(int, pcap_datalink_name_to_val(name.encode()))
3042 if val == -1:
3043 warning("Unknown linktype: %s. Using EN10MB", name)
3044 return DLT_EN10MB
3045 return val
3048@conf.commands.register
3049def tcpdump(
3050 pktlist=None, # type: Union[IO[bytes], None, str, _PacketIterable]
3051 dump=False, # type: bool
3052 getfd=False, # type: bool
3053 args=None, # type: Optional[List[str]]
3054 flt=None, # type: Optional[str]
3055 prog=None, # type: Optional[Any]
3056 getproc=False, # type: bool
3057 quiet=False, # type: bool
3058 use_tempfile=None, # type: Optional[Any]
3059 read_stdin_opts=None, # type: Optional[Any]
3060 linktype=None, # type: Optional[Any]
3061 wait=True, # type: bool
3062 _suppress=False # type: bool
3063):
3064 # type: (...) -> Any
3065 """Run tcpdump or tshark on a list of packets.
3067 When using ``tcpdump`` on OSX (``prog == conf.prog.tcpdump``), this uses a
3068 temporary file to store the packets. This works around a bug in Apple's
3069 version of ``tcpdump``: http://apple.stackexchange.com/questions/152682/
3071 Otherwise, the packets are passed in stdin.
3073 This function can be explicitly enabled or disabled with the
3074 ``use_tempfile`` parameter.
3076 When using ``wireshark``, it will be called with ``-ki -`` to start
3077 immediately capturing packets from stdin.
3079 Otherwise, the command will be run with ``-r -`` (which is correct for
3080 ``tcpdump`` and ``tshark``).
3082 This can be overridden with ``read_stdin_opts``. This has no effect when
3083 ``use_tempfile=True``, or otherwise reading packets from a regular file.
3085 :param pktlist: a Packet instance, a PacketList instance or a list of
3086 Packet instances. Can also be a filename (as a string), an open
3087 file-like object that must be a file format readable by
3088 tshark (Pcap, PcapNg, etc.) or None (to sniff)
3089 :param flt: a filter to use with tcpdump
3090 :param dump: when set to True, returns a string instead of displaying it.
3091 :param getfd: when set to True, returns a file-like object to read data
3092 from tcpdump or tshark from.
3093 :param getproc: when set to True, the subprocess.Popen object is returned
3094 :param args: arguments (as a list) to pass to tshark (example for tshark:
3095 args=["-T", "json"]).
3096 :param prog: program to use (defaults to tcpdump, will work with tshark)
3097 :param quiet: when set to True, the process stderr is discarded
3098 :param use_tempfile: When set to True, always use a temporary file to store
3099 packets.
3100 When set to False, pipe packets through stdin.
3101 When set to None (default), only use a temporary file with
3102 ``tcpdump`` on OSX.
3103 :param read_stdin_opts: When set, a list of arguments needed to capture
3104 from stdin. Otherwise, attempts to guess.
3105 :param linktype: A custom DLT value or name, to overwrite the default
3106 values.
3107 :param wait: If True (default), waits for the process to terminate before
3108 returning to Scapy. If False, the process will be detached to the
3109 background. If dump, getproc or getfd is True, these have the same
3110 effect as ``wait=False``.
3112 Examples::
3114 >>> tcpdump([IP()/TCP(), IP()/UDP()])
3115 reading from file -, link-type RAW (Raw IP)
3116 16:46:00.474515 IP 127.0.0.1.20 > 127.0.0.1.80: Flags [S], seq 0, win 8192, length 0 # noqa: E501
3117 16:46:00.475019 IP 127.0.0.1.53 > 127.0.0.1.53: [|domain]
3119 >>> tcpdump([IP()/TCP(), IP()/UDP()], prog=conf.prog.tshark)
3120 1 0.000000 127.0.0.1 -> 127.0.0.1 TCP 40 20->80 [SYN] Seq=0 Win=8192 Len=0 # noqa: E501
3121 2 0.000459 127.0.0.1 -> 127.0.0.1 UDP 28 53->53 Len=0
3123 To get a JSON representation of a tshark-parsed PacketList(), one can::
3125 >>> import json, pprint
3126 >>> json_data = json.load(tcpdump(IP(src="217.25.178.5",
3127 ... dst="45.33.32.156"),
3128 ... prog=conf.prog.tshark,
3129 ... args=["-T", "json"],
3130 ... getfd=True))
3131 >>> pprint.pprint(json_data)
3132 [{u'_index': u'packets-2016-12-23',
3133 u'_score': None,
3134 u'_source': {u'layers': {u'frame': {u'frame.cap_len': u'20',
3135 u'frame.encap_type': u'7',
3136 [...]
3137 },
3138 u'ip': {u'ip.addr': u'45.33.32.156',
3139 u'ip.checksum': u'0x0000a20d',
3140 [...]
3141 u'ip.ttl': u'64',
3142 u'ip.version': u'4'},
3143 u'raw': u'Raw packet data'}},
3144 u'_type': u'pcap_file'}]
3145 >>> json_data[0]['_source']['layers']['ip']['ip.ttl']
3146 u'64'
3147 """
3148 getfd = getfd or getproc
3149 if prog is None:
3150 if not conf.prog.tcpdump:
3151 raise Scapy_Exception(
3152 "tcpdump is not available"
3153 )
3154 prog = [conf.prog.tcpdump]
3155 elif isinstance(prog, str):
3156 prog = [prog]
3157 else:
3158 raise ValueError("prog must be a string")
3160 if linktype is not None:
3161 if isinstance(linktype, int):
3162 # Guess name from value
3163 try:
3164 linktype_name = _guess_linktype_name(linktype)
3165 except StopIteration:
3166 linktype = -1
3167 else:
3168 # Guess value from name
3169 if linktype.startswith("DLT_"):
3170 linktype = linktype[4:]
3171 linktype_name = linktype
3172 try:
3173 linktype = _guess_linktype_value(linktype)
3174 except KeyError:
3175 linktype = -1
3176 if linktype == -1:
3177 raise ValueError(
3178 "Unknown linktype. Try passing its datalink name instead"
3179 )
3180 prog += ["-y", linktype_name]
3182 # Build Popen arguments
3183 if args is None:
3184 args = []
3185 else:
3186 # Make a copy of args
3187 args = list(args)
3189 if flt is not None:
3190 # Check the validity of the filter
3191 if linktype is None and isinstance(pktlist, str):
3192 # linktype is unknown but required. Read it from file
3193 with PcapReader(pktlist) as rd:
3194 if isinstance(rd, PcapNgReader):
3195 # Get the linktype from the first packet
3196 try:
3197 _, metadata = rd._read_packet()
3198 linktype = metadata.linktype
3199 if OPENBSD and linktype == 228:
3200 linktype = DLT_RAW
3201 except EOFError:
3202 raise ValueError(
3203 "Cannot get linktype from a PcapNg packet."
3204 )
3205 else:
3206 linktype = rd.linktype
3207 from scapy.arch.common import compile_filter
3208 compile_filter(flt, linktype=linktype)
3209 args.append(flt)
3211 stdout = subprocess.PIPE if dump or getfd else None
3212 stderr = open(os.devnull) if quiet else None
3213 proc = None
3215 if use_tempfile is None:
3216 # Apple's tcpdump cannot read from stdin, see:
3217 # http://apple.stackexchange.com/questions/152682/
3218 use_tempfile = DARWIN and prog[0] == conf.prog.tcpdump
3220 if read_stdin_opts is None:
3221 if prog[0] == conf.prog.wireshark:
3222 # Start capturing immediately (-k) from stdin (-i -)
3223 read_stdin_opts = ["-ki", "-"]
3224 elif prog[0] == conf.prog.tcpdump and not OPENBSD:
3225 # Capture in packet-buffered mode (-U) from stdin (-r -)
3226 read_stdin_opts = ["-U", "-r", "-"]
3227 else:
3228 read_stdin_opts = ["-r", "-"]
3229 else:
3230 # Make a copy of read_stdin_opts
3231 read_stdin_opts = list(read_stdin_opts)
3233 if pktlist is None:
3234 # sniff
3235 with ContextManagerSubprocess(prog[0], suppress=_suppress):
3236 proc = subprocess.Popen(
3237 prog + args,
3238 stdout=stdout,
3239 stderr=stderr,
3240 )
3241 elif isinstance(pktlist, str):
3242 # file
3243 with ContextManagerSubprocess(prog[0], suppress=_suppress):
3244 proc = subprocess.Popen(
3245 prog + ["-r", pktlist] + args,
3246 stdout=stdout,
3247 stderr=stderr,
3248 )
3249 elif use_tempfile:
3250 tmpfile = get_temp_file( # type: ignore
3251 autoext=".pcap",
3252 fd=True
3253 ) # type: IO[bytes]
3254 try:
3255 tmpfile.writelines(
3256 iter(lambda: pktlist.read(1048576), b"") # type: ignore
3257 )
3258 except AttributeError:
3259 pktlist = cast("_PacketIterable", pktlist)
3260 wrpcap(tmpfile, pktlist, linktype=linktype)
3261 else:
3262 tmpfile.close()
3263 with ContextManagerSubprocess(prog[0], suppress=_suppress):
3264 proc = subprocess.Popen(
3265 prog + ["-r", tmpfile.name] + args,
3266 stdout=stdout,
3267 stderr=stderr,
3268 )
3269 else:
3270 try:
3271 pktlist.fileno() # type: ignore
3272 # pass the packet stream
3273 with ContextManagerSubprocess(prog[0], suppress=_suppress):
3274 proc = subprocess.Popen(
3275 prog + read_stdin_opts + args,
3276 stdin=pktlist, # type: ignore
3277 stdout=stdout,
3278 stderr=stderr,
3279 )
3280 except (AttributeError, ValueError):
3281 # write the packet stream to stdin
3282 with ContextManagerSubprocess(prog[0], suppress=_suppress):
3283 proc = subprocess.Popen(
3284 prog + read_stdin_opts + args,
3285 stdin=subprocess.PIPE,
3286 stdout=stdout,
3287 stderr=stderr,
3288 )
3289 if proc is None:
3290 # An error has occurred
3291 return
3292 try:
3293 proc.stdin.writelines( # type: ignore
3294 iter(lambda: pktlist.read(1048576), b"") # type: ignore
3295 )
3296 except AttributeError:
3297 wrpcap(proc.stdin, pktlist, linktype=linktype) # type: ignore
3298 except UnboundLocalError:
3299 # The error was handled by ContextManagerSubprocess
3300 pass
3301 else:
3302 proc.stdin.close() # type: ignore
3303 if proc is None:
3304 # An error has occurred
3305 return
3306 if dump:
3307 data = b"".join(
3308 iter(lambda: proc.stdout.read(1048576), b"") # type: ignore
3309 )
3310 proc.terminate()
3311 return data
3312 if getproc:
3313 return proc
3314 if getfd:
3315 return proc.stdout
3316 if wait:
3317 proc.wait()
3320@conf.commands.register
3321def hexedit(pktlist):
3322 # type: (_PacketIterable) -> PacketList
3323 """Run hexedit on a list of packets, then return the edited packets."""
3324 f = get_temp_file()
3325 wrpcap(f, pktlist)
3326 with ContextManagerSubprocess(conf.prog.hexedit):
3327 subprocess.call([conf.prog.hexedit, f])
3328 rpktlist = rdpcap(f)
3329 os.unlink(f)
3330 return rpktlist
3333def get_terminal_width():
3334 # type: () -> Optional[int]
3335 """Get terminal width (number of characters) if in a window.
3337 Notice: this will try several methods in order to
3338 support as many terminals and OS as possible.
3339 """
3340 sizex = shutil.get_terminal_size(fallback=(0, 0))[0]
3341 if sizex != 0:
3342 return sizex
3343 # Backups
3344 if WINDOWS:
3345 from ctypes import windll, create_string_buffer
3346 # http://code.activestate.com/recipes/440694-determine-size-of-console-window-on-windows/
3347 h = windll.kernel32.GetStdHandle(-12)
3348 csbi = create_string_buffer(22)
3349 res = windll.kernel32.GetConsoleScreenBufferInfo(h, csbi)
3350 if res:
3351 (bufx, bufy, curx, cury, wattr,
3352 left, top, right, bottom, maxx, maxy) = struct.unpack("hhhhHhhhhhh", csbi.raw) # noqa: E501
3353 sizex = right - left + 1
3354 # sizey = bottom - top + 1
3355 return sizex
3356 return sizex
3357 # We have various methods
3358 # COLUMNS is set on some terminals
3359 try:
3360 sizex = int(os.environ['COLUMNS'])
3361 except Exception:
3362 pass
3363 if sizex:
3364 return sizex
3365 # We can query TIOCGWINSZ
3366 try:
3367 import fcntl
3368 import termios
3369 s = struct.pack('HHHH', 0, 0, 0, 0)
3370 x = fcntl.ioctl(1, termios.TIOCGWINSZ, s)
3371 sizex = struct.unpack('HHHH', x)[1]
3372 except (IOError, ModuleNotFoundError):
3373 # If everything failed, return default terminal size
3374 sizex = 79
3375 return sizex
3378def pretty_list(rtlst, # type: List[Tuple[Union[str, List[str]], ...]]
3379 header, # type: List[Tuple[str, ...]]
3380 sortBy=0, # type: Optional[int]
3381 borders=False, # type: bool
3382 ):
3383 # type: (...) -> str
3384 """
3385 Pretty list to fit the terminal, and add header.
3387 :param rtlst: a list of tuples. each tuple contains a value which can
3388 be either a string or a list of string.
3389 :param sortBy: the column id (starting with 0) which will be used for
3390 ordering
3391 :param borders: whether to put borders on the table or not
3392 """
3393 if borders:
3394 _space = "|"
3395 else:
3396 _space = " "
3397 cols = len(header[0])
3398 # Windows has a fat terminal border
3399 _spacelen = len(_space) * (cols - 1) + int(WINDOWS)
3400 _croped = False
3401 if sortBy is not None:
3402 # Sort correctly
3403 rtlst.sort(key=lambda x: x[sortBy])
3404 # Resolve multi-values
3405 for i, line in enumerate(rtlst):
3406 ids = [] # type: List[int]
3407 values = [] # type: List[Union[str, List[str]]]
3408 for j, val in enumerate(line):
3409 if isinstance(val, list):
3410 ids.append(j)
3411 values.append(val or " ")
3412 if values:
3413 del rtlst[i]
3414 k = 0
3415 for ex_vals in zip_longest(*values, fillvalue=" "):
3416 if k:
3417 extra_line = [" "] * cols
3418 else:
3419 extra_line = list(line) # type: ignore
3420 for j, h in enumerate(ids):
3421 extra_line[h] = ex_vals[j]
3422 rtlst.insert(i + k, tuple(extra_line))
3423 k += 1
3424 rtslst = cast(List[Tuple[str, ...]], rtlst)
3425 # Append tag
3426 rtslst = header + rtslst
3427 # Detect column's width
3428 colwidth = [max(len(y) for y in x) for x in zip(*rtslst)]
3429 # Make text fit in box (if required)
3430 width = get_terminal_width()
3431 if conf.auto_crop_tables and width:
3432 width = width - _spacelen
3433 while sum(colwidth) > width:
3434 _croped = True
3435 # Needs to be cropped
3436 # Get the longest row
3437 i = colwidth.index(max(colwidth))
3438 # Get all elements of this row
3439 row = [len(x[i]) for x in rtslst]
3440 # Get biggest element of this row: biggest of the array
3441 j = row.index(max(row))
3442 # Re-build column tuple with the edited element
3443 t = list(rtslst[j])
3444 t[i] = t[i][:-2] + "_"
3445 rtslst[j] = tuple(t)
3446 # Update max size
3447 row[j] = len(t[i])
3448 colwidth[i] = max(row)
3449 if _croped:
3450 log_runtime.info("Table cropped to fit the terminal (conf.auto_crop_tables==True)") # noqa: E501
3451 # Generate padding scheme
3452 fmt = _space.join(["%%-%ds" % x for x in colwidth])
3453 # Append separation line if needed
3454 if borders:
3455 rtslst.insert(1, tuple("-" * x for x in colwidth))
3456 # Compile
3457 return "\n".join(fmt % x for x in rtslst)
3460def human_size(x, fmt=".1f"):
3461 # type: (int, str) -> str
3462 """
3463 Convert a size in octets to a human string representation
3464 """
3465 units = ['K', 'M', 'G', 'T', 'P', 'E']
3466 if not x:
3467 return "0B"
3468 i = int(math.log(x, 2**10))
3469 if i and i < len(units):
3470 return format(x / 2**(10 * i), fmt) + units[i - 1]
3471 return str(x) + "B"
3474def __make_table(
3475 yfmtfunc, # type: Callable[[int], str]
3476 fmtfunc, # type: Callable[[int], str]
3477 endline, # type: str
3478 data, # type: List[Tuple[Packet, Packet]]
3479 fxyz, # type: Callable[[Packet, Packet], Tuple[Any, Any, Any]]
3480 sortx=None, # type: Optional[Callable[[str], Tuple[Any, ...]]]
3481 sorty=None, # type: Optional[Callable[[str], Tuple[Any, ...]]]
3482 seplinefunc=None, # type: Optional[Callable[[int, List[int]], str]]
3483 dump=False, # type: bool
3484 stringconv=str, # type: Callable[[str], str]
3485):
3486 # type: (...) -> Optional[str]
3487 """Core function of the make_table suite, which generates the table"""
3488 vx = {} # type: Dict[str, int]
3489 vy = {} # type: Dict[str, Optional[int]]
3490 vz = {} # type: Dict[Tuple[str, str], str]
3491 vxf = {} # type: Dict[str, str]
3493 tmp_len = 0
3494 for e in data:
3495 xx, yy, zz = [str(s) for s in fxyz(*e)]
3496 tmp_len = max(len(yy), tmp_len)
3497 vx[xx] = max(vx.get(xx, 0), len(xx), len(zz))
3498 vy[yy] = None
3499 vz[(xx, yy)] = zz
3501 vxk = list(vx)
3502 vyk = list(vy)
3503 if sortx:
3504 vxk.sort(key=sortx)
3505 else:
3506 try:
3507 vxk.sort(key=int)
3508 except Exception:
3509 try:
3510 vxk.sort(key=atol)
3511 except Exception:
3512 vxk.sort()
3513 if sorty:
3514 vyk.sort(key=sorty)
3515 else:
3516 try:
3517 vyk.sort(key=int)
3518 except Exception:
3519 try:
3520 vyk.sort(key=atol)
3521 except Exception:
3522 vyk.sort()
3524 s = ""
3525 if seplinefunc:
3526 sepline = seplinefunc(tmp_len, [vx[x] for x in vxk])
3527 s += sepline + "\n"
3529 fmt = yfmtfunc(tmp_len)
3530 s += fmt % ""
3531 s += ' '
3532 for x in vxk:
3533 vxf[x] = fmtfunc(vx[x])
3534 s += vxf[x] % stringconv(x)
3535 s += ' '
3536 s += endline + "\n"
3537 if seplinefunc:
3538 s += sepline + "\n"
3539 for y in vyk:
3540 s += fmt % stringconv(y)
3541 s += ' '
3542 for x in vxk:
3543 s += vxf[x] % stringconv(vz.get((x, y), "-"))
3544 s += ' '
3545 s += endline + "\n"
3546 if seplinefunc:
3547 s += sepline + "\n"
3549 if dump:
3550 return s
3551 else:
3552 print(s, end="")
3553 return None
3556def make_table(*args, **kargs):
3557 # type: (*Any, **Any) -> Optional[Any]
3558 return __make_table(
3559 lambda l: "%%-%is" % l,
3560 lambda l: "%%-%is" % l,
3561 "",
3562 *args,
3563 **kargs
3564 )
3567def make_lined_table(*args, **kargs):
3568 # type: (*Any, **Any) -> Optional[str]
3569 return __make_table( # type: ignore
3570 lambda l: "%%-%is |" % l,
3571 lambda l: "%%-%is |" % l,
3572 "",
3573 *args,
3574 seplinefunc=lambda a, x: "+".join(
3575 '-' * (y + 2) for y in [a - 1] + x + [-2]
3576 ),
3577 **kargs
3578 )
3581def make_tex_table(*args, **kargs):
3582 # type: (*Any, **Any) -> Optional[str]
3583 return __make_table( # type: ignore
3584 lambda l: "%s",
3585 lambda l: "& %s",
3586 "\\\\",
3587 *args,
3588 seplinefunc=lambda a, x: "\\hline",
3589 stringconv=tex_escape,
3590 **kargs
3591 )
3593####################
3594# WHOIS CLIENT #
3595####################
3598def whois(ip_address):
3599 # type: (str) -> bytes
3600 """Whois client for Python"""
3601 whois_ip = str(ip_address)
3602 try:
3603 query = socket.gethostbyname(whois_ip)
3604 except Exception:
3605 query = whois_ip
3606 s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
3607 s.connect(("whois.ripe.net", 43))
3608 s.send(query.encode("utf8") + b"\r\n")
3609 answer = b""
3610 while True:
3611 d = s.recv(4096)
3612 answer += d
3613 if not d:
3614 break
3615 s.close()
3616 ignore_tag = b"remarks:"
3617 # ignore all lines starting with the ignore_tag
3618 lines = [line for line in answer.split(b"\n") if not line or (line and not line.startswith(ignore_tag))] # noqa: E501
3619 # remove empty lines at the bottom
3620 for i in range(1, len(lines)):
3621 if not lines[-i].strip():
3622 del lines[-i]
3623 else:
3624 break
3625 return b"\n".join(lines[3:])
3627####################
3628# CLI utils #
3629####################
3632class _CLIUtilMetaclass(type):
3633 class TYPE(enum.Enum):
3634 COMMAND = 0
3635 OUTPUT = 1
3636 COMPLETE = 2
3638 def __new__(cls, # type: Type[_CLIUtilMetaclass]
3639 name, # type: str
3640 bases, # type: Tuple[type, ...]
3641 dct # type: Dict[str, Any]
3642 ):
3643 # type: (...) -> Type[CLIUtil]
3644 dct["commands"] = {
3645 x.__name__: x
3646 for x in dct.values()
3647 if getattr(x, "cliutil_type", None) == _CLIUtilMetaclass.TYPE.COMMAND
3648 }
3649 dct["commands_output"] = {
3650 x.cliutil_ref.__name__: x
3651 for x in dct.values()
3652 if getattr(x, "cliutil_type", None) == _CLIUtilMetaclass.TYPE.OUTPUT
3653 }
3654 dct["commands_complete"] = {
3655 x.cliutil_ref.__name__: x
3656 for x in dct.values()
3657 if getattr(x, "cliutil_type", None) == _CLIUtilMetaclass.TYPE.COMPLETE
3658 }
3659 newcls = cast(Type['CLIUtil'], type.__new__(cls, name, bases, dct))
3660 return newcls
3663class CLIUtil(metaclass=_CLIUtilMetaclass):
3664 """
3665 Provides a Util class to easily create simple CLI tools in Scapy,
3666 that can still be used as an API.
3668 Doc:
3669 - override the ps1() function
3670 - register commands with the @CLIUtil.addcomment decorator
3671 - call the loop() function when ready
3672 """
3674 def _depcheck(self) -> None:
3675 """
3676 Check that all dependencies are installed
3677 """
3678 try:
3679 import prompt_toolkit # noqa: F401
3680 except ImportError:
3681 # okay we lie but prompt_toolkit is a dependency...
3682 raise ImportError("You need to have IPython installed to use the CLI")
3684 # Okay let's do nice code
3685 commands: Dict[str, Callable[..., Any]] = {}
3686 # print output of command
3687 commands_output: Dict[str, Callable[..., str]] = {}
3688 # provides completion to command
3689 commands_complete: Dict[str, Callable[..., List[str]]] = {}
3691 def __init__(self, cli: bool = True, debug: bool = False) -> None:
3692 """
3693 DEV: overwrite
3694 """
3695 if cli:
3696 self._depcheck()
3697 self.loop(debug=debug)
3699 @staticmethod
3700 def _inspectkwargs(func: DecoratorCallable) -> None:
3701 """
3702 Internal function to parse arguments from the kwargs of the functions
3703 """
3704 func._flagnames = [ # type: ignore
3705 x.name for x in
3706 inspect.signature(func).parameters.values()
3707 if x.kind == inspect.Parameter.KEYWORD_ONLY
3708 ]
3709 func._flags = [ # type: ignore
3710 ("-%s" % x) if len(x) == 1 else ("--%s" % x)
3711 for x in func._flagnames # type: ignore
3712 ]
3714 @staticmethod
3715 def _parsekwargs(
3716 func: DecoratorCallable,
3717 args: List[str]
3718 ) -> Tuple[List[str], Dict[str, Literal[True]]]:
3719 """
3720 Internal function to parse CLI arguments of a function.
3721 """
3722 kwargs: Dict[str, Literal[True]] = {}
3723 if func._flags: # type: ignore
3724 i = 0
3725 for arg in args:
3726 if arg in func._flags: # type: ignore
3727 i += 1
3728 kwargs[func._flagnames[func._flags.index(arg)]] = True # type: ignore # noqa: E501
3729 continue
3730 break
3731 args = args[i:]
3732 return args, kwargs
3734 @classmethod
3735 def _parseallargs(
3736 cls,
3737 func: DecoratorCallable,
3738 cmd: str, args: List[str]
3739 ) -> Tuple[List[str], Dict[str, Literal[True]], Dict[str, Literal[True]]]:
3740 """
3741 Internal function to parse CLI arguments of both the function
3742 and its output function.
3743 """
3744 args, kwargs = cls._parsekwargs(func, args)
3745 outkwargs: Dict[str, Literal[True]] = {}
3746 if cmd in cls.commands_output:
3747 args, outkwargs = cls._parsekwargs(cls.commands_output[cmd], args)
3748 return args, kwargs, outkwargs
3750 @classmethod
3751 def addcommand(
3752 cls,
3753 mono: bool = False,
3754 globsupport: bool = False,
3755 ) -> Callable[[DecoratorCallable], DecoratorCallable]:
3756 """
3757 Decorator to register a command
3759 :param mono: if True, the command takes a single argument even
3760 if there are spaces.
3761 """
3762 def func(cmd: DecoratorCallable) -> DecoratorCallable:
3763 cmd.cliutil_type = _CLIUtilMetaclass.TYPE.COMMAND # type: ignore
3764 cmd._mono = mono # type: ignore
3765 cmd._globsupport = globsupport # type: ignore
3766 cls._inspectkwargs(cmd)
3767 if cmd._globsupport and not cmd._mono: # type: ignore
3768 raise ValueError("Cannot use globsupport without mono.")
3769 return cmd
3770 return func
3772 @classmethod
3773 def addoutput(cls, cmd: DecoratorCallable) -> Callable[[DecoratorCallable], DecoratorCallable]: # noqa: E501
3774 """
3775 Decorator to register a command output processor
3776 """
3777 def func(processor: DecoratorCallable) -> DecoratorCallable:
3778 processor.cliutil_type = _CLIUtilMetaclass.TYPE.OUTPUT # type: ignore
3779 processor.cliutil_ref = cmd # type: ignore
3780 cls._inspectkwargs(processor)
3781 return processor
3782 return func
3784 @classmethod
3785 def addcomplete(
3786 cls,
3787 cmd: DecoratorCallable,
3788 ) -> Callable[[DecoratorCallable], DecoratorCallable]:
3789 """
3790 Decorator to register a command completor
3791 """
3792 def func(processor: DecoratorCallable) -> DecoratorCallable:
3793 processor.cliutil_type = _CLIUtilMetaclass.TYPE.COMPLETE # type: ignore
3794 processor.cliutil_ref = cmd # type: ignore
3795 processor._mono = cmd._mono # type: ignore
3796 return processor
3797 return func
3799 def ps1(self) -> str:
3800 """
3801 Return the PS1 of the shell
3802 """
3803 return "> "
3805 def close(self) -> None:
3806 """
3807 Function called on exiting
3808 """
3809 print("Exited")
3811 def help(self, cmd: Optional[str] = None) -> None:
3812 """
3813 Return the help related to this CLI util
3814 """
3815 def _args(func: Any) -> str:
3816 flags = func._flags.copy()
3817 if func.__name__ in self.commands_output:
3818 flags += self.commands_output[func.__name__]._flags # type: ignore
3819 return " %s%s" % (
3820 (
3821 "%s " % " ".join("[%s]" % x for x in flags)
3822 if flags else ""
3823 ),
3824 " ".join(
3825 "<%s%s>" % (
3826 x.name,
3827 "?" if
3828 (x.default is None or x.default != inspect.Parameter.empty)
3829 else ""
3830 )
3831 for x in list(inspect.signature(func).parameters.values())[1:]
3832 if x.name not in func._flagnames and x.name[0] != "_"
3833 )
3834 )
3836 if cmd:
3837 if cmd not in self.commands:
3838 print("Unknown command '%s'" % cmd)
3839 return
3840 # help for one command
3841 func = self.commands[cmd]
3842 print("%s%s: %s" % (
3843 cmd,
3844 _args(func),
3845 func.__doc__ and func.__doc__.strip()
3846 ))
3847 else:
3848 header = "│ %s - Help │" % self.__class__.__name__
3849 print("┌" + "─" * (len(header) - 2) + "┐")
3850 print(header)
3851 print("└" + "─" * (len(header) - 2) + "┘")
3852 print(
3853 pretty_list(
3854 [
3855 (
3856 cmd,
3857 _args(func),
3858 func.__doc__ and func.__doc__.strip().split("\n")[0] or ""
3859 )
3860 for cmd, func in self.commands.items()
3861 ],
3862 [("Command", "Arguments", "Description")]
3863 )
3864 )
3866 def _split_cmd(self, cmd: str) -> Tuple[List[str], List[int]]:
3867 """
3868 Split the command in multiple arguments
3869 """
3870 quoted = None
3871 queue = [""]
3872 offsets = [0]
3873 for i, c in enumerate(cmd):
3874 if c == "'" or c == '"':
3875 # This is a quote.
3876 if quoted is not None and quoted == c:
3877 # We are closing the last quote
3878 quoted = None
3879 elif quoted:
3880 queue[-1] += c
3881 else:
3882 quoted = c
3883 elif c == " ":
3884 # This is a space.
3885 if quoted is not None:
3886 # We're in a quote, append it
3887 queue[-1] += c
3888 elif queue[-1]:
3889 # Not in a quote, this splits the argument.
3890 queue += [""]
3891 offsets.append(i)
3892 else:
3893 # Padding space, advance offset
3894 offsets[-1] += 1
3895 else:
3896 # This is a char
3897 queue[-1] += c
3898 return queue, offsets
3900 def _completer(self) -> 'prompt_toolkit.completion.Completer':
3901 """
3902 Returns a prompt_toolkit custom completer
3903 """
3904 from prompt_toolkit.completion import Completer, Completion
3906 class CLICompleter(Completer):
3907 def get_completions(cmpl, document, complete_event): # type: ignore
3908 if not complete_event.completion_requested:
3909 # Only activate when the user does <TAB>
3910 return
3911 parts, offsets = self._split_cmd(document.text)
3912 cmd = parts[0].lower()
3913 if cmd not in self.commands:
3914 # We are trying to complete the command
3915 for possible_cmd in (x for x in self.commands if x.startswith(cmd)):
3916 yield Completion(possible_cmd, start_position=-len(cmd))
3917 else:
3918 # We are trying to complete the command content
3919 if len(parts) == 1:
3920 return
3921 args, _, _ = self._parseallargs(self.commands[cmd], cmd, parts[1:])
3922 if cmd in self.commands_complete:
3923 completer = self.commands_complete[cmd]
3924 # If the completion is 'mono', it's a single argument with
3925 # spaces. Else we pass the list of arguments to complete,
3926 # and we only complete the last argument.
3927 if completer._mono: # type: ignore
3928 arg = " ".join(args)
3929 completions = completer(self, arg)
3930 startpos = offsets[1]
3931 else:
3932 completions = completer(self, args)
3933 startpos = offsets[-1]
3935 # For each possible completion
3936 for possible_arg in completions:
3937 # If there's a space in the completion, and we're
3938 # not in mono mode, add quotes.
3939 if " " in possible_arg and not completer._mono: # type: ignore # noqa: E501
3940 possible_arg = '"%s"' % possible_arg
3942 yield Completion(
3943 possible_arg,
3944 start_position=startpos - len(document.text) + 1
3945 )
3946 return
3947 return CLICompleter()
3949 def loop(self, debug: int = 0) -> None:
3950 """
3951 Main command handling loop
3952 """
3953 from prompt_toolkit import PromptSession
3954 session = PromptSession(completer=self._completer())
3956 while True:
3957 try:
3958 cmd = session.prompt(self.ps1()).strip()
3959 except KeyboardInterrupt:
3960 continue
3961 except EOFError:
3962 self.close()
3963 break
3964 parts, _ = self._split_cmd(cmd)
3965 args = parts[1:]
3966 cmd = parts[0].strip().lower()
3967 if not cmd:
3968 continue
3969 if cmd in ["help", "h", "?"]:
3970 self.help(" ".join(args))
3971 continue
3972 if cmd in "exit":
3973 break
3974 if cmd not in self.commands:
3975 print("Unknown command. Type help or ?")
3976 else:
3977 # check the number of arguments
3978 func = self.commands[cmd]
3979 args, kwargs, outkwargs = self._parseallargs(func, cmd, args)
3980 if func._mono: # type: ignore
3981 args = [" ".join(args)]
3982 # if globsupport is set, we might need to do several calls
3983 if func._globsupport and "*" in args[0]: # type: ignore
3984 if args[0].count("*") > 1:
3985 print("More than 1 glob star (*) is currently unsupported.")
3986 continue
3987 before, after = args[0].split("*", 1)
3988 reg = re.compile(re.escape(before) + r".*" + after)
3989 calls = [
3990 [x] for x in
3991 self.commands_complete[cmd](self, before)
3992 if reg.match(x)
3993 ]
3994 else:
3995 calls = [args]
3996 else:
3997 calls = [args]
3998 # now iterate if required, call the function and print its output
3999 res = None
4000 for args in calls:
4001 try:
4002 res = func(self, *args, **kwargs)
4003 except KeyboardInterrupt:
4004 print("Aborted.")
4005 except TypeError as ex:
4006 print("Bad number of arguments !")
4007 if debug:
4008 traceback.print_exception(ex)
4009 self.help(cmd=cmd)
4010 continue
4011 except Exception as ex:
4012 print("Command failed with error: %s" % ex)
4013 if debug:
4014 traceback.print_exception(ex)
4015 try:
4016 if res and cmd in self.commands_output:
4017 self.commands_output[cmd](self, res, **outkwargs)
4018 except KeyboardInterrupt:
4019 print("Aborted.")
4020 except Exception as ex:
4021 print("Output processor failed with error: %s" % ex)
4024def AutoArgparse(
4025 func: DecoratorCallable,
4026 _parseonly: bool = False,
4027) -> Optional[Tuple[List[str], List[str]]]:
4028 """
4029 Generate an Argparse call from a function, then call this function.
4031 Notes:
4033 - for the arguments to have a description, the sphinx docstring format
4034 must be used. See
4035 https://sphinx-rtd-tutorial.readthedocs.io/en/latest/docstrings.html
4036 - the arguments must be typed in Python (we ignore Sphinx-specific types)
4037 untyped arguments are ignored.
4038 - only types that would be supported by argparse are supported. The others
4039 are omitted.
4040 """
4041 argsdoc = {}
4042 if func.__doc__:
4043 # Sphinx doc format parser
4044 m = re.match(
4045 r"((?:.|\n)*?)(\n\s*:(?:param|type|raises|return|rtype)(?:.|\n)*)",
4046 func.__doc__.strip(),
4047 )
4048 if not m:
4049 desc = func.__doc__.strip()
4050 else:
4051 desc = m.group(1)
4052 sphinxargs = re.findall(
4053 r"\s*:(param|type|raises|return|rtype)\s*([^:]*):(.*)",
4054 m.group(2),
4055 )
4056 for argtype, argparam, argdesc in sphinxargs:
4057 argparam = argparam.strip()
4058 argdesc = argdesc.strip()
4059 if argtype == "param":
4060 if not argparam:
4061 raise ValueError(":param: without a name !")
4062 argsdoc[argparam] = argdesc
4063 else:
4064 desc = ""
4066 # Process the parameters
4067 positional = []
4068 noargument = []
4069 hexarguments = []
4070 parameters = {}
4071 for param in inspect.signature(func).parameters.values():
4072 if not param.annotation:
4073 continue
4074 noarg = False
4075 parname = param.name.replace("_", "-")
4076 paramkwargs: Dict[str, Any] = {}
4077 if param.annotation is bool:
4078 if param.default is True:
4079 parname = "no-" + parname
4080 paramkwargs["action"] = "store_false"
4081 else:
4082 paramkwargs["action"] = "store_true"
4083 noarg = True
4084 elif param.annotation is bytes:
4085 paramkwargs["type"] = str
4086 hexarguments.append(parname)
4087 elif param.annotation in [str, int, float]:
4088 paramkwargs["type"] = param.annotation
4089 elif (
4090 isinstance(param.annotation, type) and
4091 issubclass(param.annotation, enum.Enum)
4092 ):
4093 paramkwargs["type"] = param.annotation
4094 paramkwargs["choices"] = list(param.annotation)
4095 else:
4096 continue
4097 if param.default != inspect.Parameter.empty:
4098 if param.kind == inspect.Parameter.POSITIONAL_ONLY:
4099 positional.append(parname)
4100 paramkwargs["nargs"] = '?'
4101 else:
4102 parname = "--" + parname
4103 paramkwargs["default"] = param.default
4104 elif param.kind == inspect.Parameter.KEYWORD_ONLY:
4105 # Required but Keyword only
4106 parname = "--" + parname
4107 paramkwargs["required"] = True
4108 else:
4109 positional.append(parname)
4110 if param.kind == inspect.Parameter.VAR_POSITIONAL:
4111 paramkwargs["action"] = "append"
4112 if param.name in argsdoc:
4113 paramkwargs["help"] = argsdoc[param.name]
4114 if param.annotation is bytes:
4115 paramkwargs["help"] = "(hex) " + paramkwargs["help"]
4116 elif param.annotation is bool:
4117 paramkwargs["help"] = "(flag) " + paramkwargs["help"]
4118 else:
4119 paramkwargs["help"] = (
4120 "(%s) " % param.annotation.__name__ + paramkwargs["help"]
4121 )
4122 # Add to the parameter list
4123 parameters[parname] = paramkwargs
4124 if noarg:
4125 noargument.append(parname)
4127 if _parseonly:
4128 # An internal mode used to generate bash autocompletion, do it then exit.
4129 return (
4130 [x for x in parameters if x not in positional] + ["--help"],
4131 [x for x in noargument if x not in positional] + ["--help"],
4132 )
4134 # Now build the argparse.ArgumentParser
4135 parser = argparse.ArgumentParser(
4136 prog=func.__name__,
4137 description=desc,
4138 formatter_class=argparse.ArgumentDefaultsHelpFormatter,
4139 )
4141 # Add parameters to parser
4142 for parname, paramkwargs in parameters.items():
4143 parser.add_argument(parname, **paramkwargs)
4145 # Now parse the sys.argv parameters
4146 params = vars(parser.parse_args())
4148 # Convert hex parameters if provided
4149 for p in hexarguments:
4150 if params[p] is not None:
4151 try:
4152 params[p] = bytes.fromhex(params[p])
4153 except ValueError:
4154 print(
4155 conf.color_theme.fail(
4156 "ERROR: the value of parameter %s "
4157 "'%s' is not valid hexadecimal !" % (p, params[p])
4158 )
4159 )
4160 return None
4162 # Act as in interactive mode
4163 conf.logLevel = 20
4164 from scapy.themes import DefaultTheme
4165 conf.color_theme = DefaultTheme()
4166 # And call the function
4167 try:
4168 func(
4169 *[params.pop(x) for x in positional],
4170 **{
4171 (k[3:] if k.startswith("no_") else k): v
4172 for k, v in params.items()
4173 }
4174 )
4175 except AssertionError as ex:
4176 print(conf.color_theme.fail("ERROR: " + str(ex)))
4177 parser.print_help()
4178 return None
4181#######################
4182# PERIODIC SENDER #
4183#######################
4186class PeriodicSenderThread(threading.Thread):
4187 def __init__(self, sock, pkt, interval=0.5, ignore_exceptions=True):
4188 # type: (Any, _PacketIterable, float, bool) -> None
4189 """ Thread to send packets periodically
4191 Args:
4192 sock: socket where packet is sent periodically
4193 pkt: packet or list of packets to send
4194 interval: interval between two packets
4195 """
4196 if not isinstance(pkt, list):
4197 self._pkts = [cast("Packet", pkt)] # type: _PacketIterable
4198 else:
4199 self._pkts = pkt
4200 self._socket = sock
4201 self._stopped = threading.Event()
4202 self._enabled = threading.Event()
4203 self._enabled.set()
4204 self._interval = interval
4205 self._ignore_exceptions = ignore_exceptions
4206 threading.Thread.__init__(self)
4208 def enable(self):
4209 # type: () -> None
4210 self._enabled.set()
4212 def disable(self):
4213 # type: () -> None
4214 self._enabled.clear()
4216 def run(self):
4217 # type: () -> None
4218 while not self._stopped.is_set() and not self._socket.closed:
4219 for p in self._pkts:
4220 try:
4221 if self._enabled.is_set():
4222 self._socket.send(p)
4223 except (OSError, TimeoutError) as e:
4224 if self._ignore_exceptions:
4225 return
4226 else:
4227 raise e
4228 self._stopped.wait(timeout=self._interval)
4229 if self._stopped.is_set() or self._socket.closed:
4230 break
4232 def stop(self):
4233 # type: () -> None
4234 self._stopped.set()
4235 self.join(self._interval * 2)
4238class SingleConversationSocket(object):
4239 def __init__(self, o):
4240 # type: (Any) -> None
4241 self._inner = o
4242 self._tx_mutex = threading.RLock()
4244 @property
4245 def __dict__(self): # type: ignore
4246 return self._inner.__dict__
4248 def __getattr__(self, name):
4249 # type: (str) -> Any
4250 return getattr(self._inner, name)
4252 def sr1(self, *args, **kargs):
4253 # type: (*Any, **Any) -> Any
4254 with self._tx_mutex:
4255 return self._inner.sr1(*args, **kargs)
4257 def sr(self, *args, **kargs):
4258 # type: (*Any, **Any) -> Any
4259 with self._tx_mutex:
4260 return self._inner.sr(*args, **kargs)
4262 def send(self, x):
4263 # type: (Packet) -> Any
4264 with self._tx_mutex:
4265 try:
4266 return self._inner.send(x)
4267 except (ConnectionError, OSError) as e:
4268 self._inner.close()
4269 raise e