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