Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/scapy/supersocket.py: 28%

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"""

7SuperSocket.

8"""

10from select import select, error as select_error

12try:

13 import ctypes

14except OSError:

15 ctypes = None # type: ignore[assignment]

16import errno

17import socket

18import struct

19import time

21from scapy.config import conf

22from scapy.consts import DARWIN, WINDOWS

23from scapy.data import (

24 MTU,

25 ETH_P_IP,

26 ETH_P_IPV6,

27 SOL_PACKET,

28 SO_TIMESTAMPNS,

29)

30from scapy.compat import raw

31from scapy.error import warning, log_runtime

32from scapy.interfaces import network_name

33from scapy.packet import Packet, NoPayload

34from scapy.plist import (

35 PacketList,

36 SndRcvList,

37 _PacketIterable,

38)

39from scapy.utils import PcapReader, tcpdump

41# Typing imports

42from scapy.interfaces import _GlobInterfaceType

43from typing import (

44 Any,

45 Dict,

46 Iterator,

47 List,

48 Optional,

49 Tuple,

50 Type,

51 TypeVar,

52 cast,

53 TYPE_CHECKING,

54)

55from scapy.compat import Self

57if TYPE_CHECKING:

58 from scapy.ansmachine import AnsweringMachine

61# Utils

64class _SuperSocket_metaclass(type):

65 desc = None # type: Optional[str]

67 def __repr__(self):

68 # type: () -> str

69 if self.desc is not None:

70 return "<%s: %s>" % (self.__name__, self.desc)

71 else:

72 return "<%s>" % self.__name__

75# Used to get ancillary data

76PACKET_AUXDATA = 8

77ETH_P_8021Q = 0x8100

78TP_STATUS_VLAN_VALID = 1 << 4

79TP_STATUS_VLAN_TPID_VALID = 1 << 6

82if ctypes is not None:

83 class tpacket_auxdata(ctypes.Structure):

84 _fields_ = [

85 ("tp_status", ctypes.c_uint),

86 ("tp_len", ctypes.c_uint),

87 ("tp_snaplen", ctypes.c_uint),

88 ("tp_mac", ctypes.c_ushort),

89 ("tp_net", ctypes.c_ushort),

90 ("tp_vlan_tci", ctypes.c_ushort),

91 ("tp_vlan_tpid", ctypes.c_ushort),

92 ] # type: List[Tuple[str, Any]]

95# SuperSocket

97_T = TypeVar("_T", Packet, PacketList)

100class SuperSocket(metaclass=_SuperSocket_metaclass):

101 closed = False # type: bool

102 nonblocking_socket = False # type: bool

103 auxdata_available = False # type: bool

104

105 def __init__(self,

106 family=socket.AF_INET, # type: int

107 type=socket.SOCK_STREAM, # type: int

108 proto=0, # type: int

109 iface=None, # type: Optional[_GlobInterfaceType]

110 **kwargs # type: Any

111 ):

112 # type: (...) -> None

113 self.ins = socket.socket(family, type, proto) # type: socket.socket

114 self.outs = self.ins # type: Optional[socket.socket]

115 self.promisc = conf.sniff_promisc

116 self.iface = iface or conf.iface

117

118 def send(self, x):

119 # type: (Packet) -> int

120 """Sends a `Packet` object

121

122 :param x: `Packet` to be send

123 :return: Number of bytes that have been sent

124 """

125 sx = raw(x)

126 try:

127 x.sent_time = time.time()

128 except AttributeError:

129 pass

130

131 if self.outs:

132 return self.outs.send(sx)

133 else:

134 return 0

135

136 if WINDOWS:

137 def _recv_raw(self, sock, x):

138 # type: (socket.socket, int) -> Tuple[bytes, Any, Optional[float]]

139 """Internal function to receive a Packet.

140

141 :param sock: Socket object from which data are received

142 :param x: Number of bytes to be received

143 :return: Received bytes, address information and no timestamp

144 """

145 pkt, sa_ll = sock.recvfrom(x)

146 return pkt, sa_ll, None

147 else:

148 def _recv_raw(self, sock, x):

149 # type: (socket.socket, int) -> Tuple[bytes, Any, Optional[float]]

150 """Internal function to receive a Packet,

151 and process ancillary data.

152

153 :param sock: Socket object from which data are received

154 :param x: Number of bytes to be received

155 :return: Received bytes, address information and an optional timestamp

156 """

157 timestamp = None

158 if not self.auxdata_available:

159 pkt, _, _, sa_ll = sock.recvmsg(x)

160 return pkt, sa_ll, timestamp

161 flags_len = socket.CMSG_LEN(4096)

162 pkt, ancdata, flags, sa_ll = sock.recvmsg(x, flags_len)

163 if not pkt:

164 return pkt, sa_ll, timestamp

165 for cmsg_lvl, cmsg_type, cmsg_data in ancdata:

166 # Check available ancillary data

167 if (cmsg_lvl == SOL_PACKET and cmsg_type == PACKET_AUXDATA):

168 if ctypes is None:

169 warning(

170 "ctypes is required to parse PACKET_AUXDATA"

171 )

172 continue

173 # Parse AUXDATA

174 try:

175 auxdata = tpacket_auxdata.from_buffer_copy(cmsg_data)

176 except ValueError:

177 # Note: according to Python documentation, recvmsg()

178 # can return a truncated message. A ValueError

179 # exception likely indicates that Auxiliary

180 # Data is not supported by the Linux kernel.

181 return pkt, sa_ll, timestamp

182 if auxdata.tp_vlan_tci != 0 or \

183 auxdata.tp_status & TP_STATUS_VLAN_VALID:

184 # Insert VLAN tag

185 tpid = ETH_P_8021Q

186 if auxdata.tp_status & TP_STATUS_VLAN_TPID_VALID:

187 tpid = auxdata.tp_vlan_tpid

188 tag = struct.pack(

189 "!HH",

190 tpid,

191 auxdata.tp_vlan_tci

192 )

193 pkt = pkt[:12] + tag + pkt[12:]

194 elif cmsg_lvl == socket.SOL_SOCKET and \

195 cmsg_type == SO_TIMESTAMPNS:

196 length = len(cmsg_data)

197 if length == 16: # __kernel_timespec

198 tmp = struct.unpack("ll", cmsg_data)

199 elif length == 8: # timespec

200 tmp = struct.unpack("ii", cmsg_data)

201 else:

202 log_runtime.warning("Unknown timespec format.. ?!")

203 continue

204 timestamp = tmp[0] + tmp[1] * 1e-9

205 return pkt, sa_ll, timestamp

206

207 def recv_raw(self, x=MTU):

208 # type: (int) -> Tuple[Optional[Type[Packet]], Optional[bytes], Optional[float]] # noqa: E501

209 """Returns a tuple containing (cls, pkt_data, time)

210

211

212 :param x: Maximum number of bytes to be received, defaults to MTU

213 :return: A tuple, consisting of a Packet type, the received data,

214 and a timestamp

215 """

216 return conf.raw_layer, self.ins.recv(x), None

217

218 def recv(self, x=MTU, **kwargs):

219 # type: (int, **Any) -> Optional[Packet]

220 """Receive a Packet according to the `basecls` of this socket

221

222 :param x: Maximum number of bytes to be received, defaults to MTU

223 :return: The received `Packet` object, or None

224 """

225 cls, val, ts = self.recv_raw(x)

226 if not val or not cls:

227 return None

228 try:

229 pkt = cls(val, **kwargs) # type: Packet

230 except KeyboardInterrupt:

231 raise

232 except Exception:

233 if conf.debug_dissector:

234 from scapy.sendrecv import debug

235 debug.crashed_on = (cls, val)

236 raise

237 pkt = conf.raw_layer(val)

238 if ts:

239 pkt.time = ts

240 return pkt

241

242 def fileno(self):

243 # type: () -> int

244 return self.ins.fileno()

245

246 def close(self):

247 # type: () -> None

248 """Gracefully close this socket

249 """

250 if self.closed:

251 return

252 self.closed = True

253 if getattr(self, "outs", None):

254 if getattr(self, "ins", None) != self.outs:

255 if self.outs and self.outs.fileno() != -1:

256 self.outs.close()

257 if getattr(self, "ins", None):

258 if self.ins.fileno() != -1:

259 self.ins.close()

260

261 def sr(self, *args, **kargs):

262 # type: (Any, Any) -> Tuple[SndRcvList, PacketList]

263 """Send and Receive multiple packets

264 """

265 from scapy import sendrecv

266 return sendrecv.sndrcv(self, *args, **kargs)

267

268 def sr1(self, *args, **kargs):

269 # type: (Any, Any) -> Optional[Packet]

270 """Send one packet and receive one answer

271 """

272 from scapy import sendrecv

273 # if not explicitly specified by the user,

274 # set threaded to False in sr1 to remove the overhead

275 # for a Thread creation

276 kargs.setdefault("threaded", False)

277 ans = sendrecv.sndrcv(self, *args, **kargs)[0] # type: SndRcvList

278 if len(ans) > 0:

279 pkt = ans[0][1] # type: Packet

280 return pkt

281 else:

282 return None

283

284 def sniff(self, *args, **kargs):

285 # type: (Any, Any) -> PacketList

286 from scapy import sendrecv

287 return sendrecv.sniff(opened_socket=self, *args, **kargs)

288

289 def tshark(self, *args, **kargs):

290 # type: (Any, Any) -> None

291 from scapy import sendrecv

292 sendrecv.tshark(opened_socket=self, *args, **kargs)

293

294 def am(self,

295 cls, # type: Type[AnsweringMachine[_T]]

296 **kwargs # type: Any

297 ):

298 # type: (...) -> AnsweringMachine[_T]

299 """

300 Creates an AnsweringMachine associated with this socket.

301

302 :param cls: A subclass of AnsweringMachine to instantiate

303 """

304 return cls(opened_socket=self, socket=self, **kwargs)

305

306 @staticmethod

307 def select(sockets, remain=conf.recv_poll_rate):

308 # type: (List[SuperSocket], Optional[float]) -> List[SuperSocket]

309 """This function is called during sendrecv() routine to select

310 the available sockets.

311

312 :param sockets: an array of sockets that need to be selected

313 :returns: an array of sockets that were selected and

314 the function to be called next to get the packets (i.g. recv)

315 """

316 inp = [] # type: List[SuperSocket]

317 try:

318 inp, _, _ = select(sockets, [], [], remain)

319 except (IOError, select_error) as exc:

320 # select.error has no .errno attribute

321 if not exc.args or exc.args[0] != errno.EINTR:

322 raise

323 return inp

324

325 def __del__(self):

326 # type: () -> None

327 """Close the socket"""

328 self.close()

329

330 def __enter__(self):

331 # type: () -> Self

332 return self

333

334 def __exit__(self, exc_type, exc_value, traceback):

335 # type: (Optional[Type[BaseException]], Optional[BaseException], Optional[Any]) -> None # noqa: E501

336 """Close the socket"""

337 self.close()

338

339

340if not WINDOWS:

341 class L3RawSocket(SuperSocket):

342 desc = "Layer 3 using Raw sockets (PF_INET/SOCK_RAW)"

343

344 def __init__(self,

345 type=ETH_P_IP, # type: int

346 filter=None, # type: Optional[str]

347 iface=None, # type: Optional[_GlobInterfaceType]

348 promisc=None, # type: Optional[bool]

349 nofilter=0 # type: int

350 ):

351 # type: (...) -> None

352 self.outs = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_RAW) # noqa: E501

353 self.outs.setsockopt(socket.SOL_IP, socket.IP_HDRINCL, 1)

354 self.ins = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(type)) # noqa: E501

355 if iface is not None:

356 iface = network_name(iface)

357 self.iface = iface

358 self.ins.bind((iface, type))

359 else:

360 self.iface = "any"

361 try:

362 # Receive Auxiliary Data (VLAN tags)

363 self.ins.setsockopt(SOL_PACKET, PACKET_AUXDATA, 1)

364 self.ins.setsockopt(

365 socket.SOL_SOCKET,

366 SO_TIMESTAMPNS,

367 1

368 )

369 self.auxdata_available = True

370 except OSError:

371 # Note: Auxiliary Data is only supported since

372 # Linux 2.6.21

373 msg = "Your Linux Kernel does not support Auxiliary Data!"

374 log_runtime.info(msg)

375

376 def recv(self, x=MTU, **kwargs):

377 # type: (int, **Any) -> Optional[Packet]

378 data, sa_ll, ts = self._recv_raw(self.ins, x)

379 if sa_ll[2] == socket.PACKET_OUTGOING:

380 return None

381 if sa_ll[3] in conf.l2types:

382 cls = conf.l2types.num2layer[sa_ll[3]] # type: Type[Packet]

383 lvl = 2

384 elif sa_ll[1] in conf.l3types:

385 cls = conf.l3types.num2layer[sa_ll[1]]

386 lvl = 3

387 else:

388 cls = conf.default_l2

389 warning("Unable to guess type (interface=%s protocol=%#x family=%i). Using %s", sa_ll[0], sa_ll[1], sa_ll[3], cls.name) # noqa: E501

390 lvl = 3

391

392 try:

393 pkt = cls(data, **kwargs)

394 except KeyboardInterrupt:

395 raise

396 except Exception:

397 if conf.debug_dissector:

398 raise

399 pkt = conf.raw_layer(data)

400

401 if lvl == 2:

402 pkt = pkt.payload

403

404 if pkt is not None:

405 if ts is None:

406 from scapy.arch.linux import get_last_packet_timestamp

407 ts = get_last_packet_timestamp(self.ins)

408 pkt.time = ts

409 return pkt

410

411 def send(self, x):

412 # type: (Packet) -> int

413 try:

414 sx = raw(x)

415 if self.outs:

416 x.sent_time = time.time()

417 return self.outs.sendto(

418 sx,

419 (x.dst, 0)

420 )

421 except AttributeError:

422 raise ValueError(

423 "Missing 'dst' attribute in the first layer to be "

424 "sent using a native L3 socket ! (make sure you passed the "

425 "IP layer)"

426 )

427 except socket.error as msg:

428 log_runtime.error(msg)

429 return 0

430

431 class L3RawSocket6(L3RawSocket):

432 def __init__(self,

433 type: int = ETH_P_IPV6,

434 filter: Optional[str] = None,

435 iface: Optional[_GlobInterfaceType] = None,

436 promisc: Optional[bool] = None,

437 nofilter: bool = False) -> None:

438 # NOTE: if fragmentation is needed, it will be done by the kernel (RFC 2292) # noqa: E501

439 self.outs = socket.socket(

440 socket.AF_INET6,

441 socket.SOCK_RAW,

442 socket.IPPROTO_RAW

443 )

444 self.ins = socket.socket(

445 socket.AF_PACKET,

446 socket.SOCK_RAW,

447 socket.htons(type)

448 )

449 self.iface = cast(_GlobInterfaceType, iface)

450

451

452class SimpleSocket(SuperSocket):

453 desc = "wrapper around a classic socket"

454 __selectable_force_select__ = True

455

456 def __init__(self, sock, basecls=None):

457 # type: (socket.socket, Optional[Type[Packet]]) -> None

458 self.ins = sock

459 self.outs = sock

460 if basecls is None:

461 basecls = conf.raw_layer

462 self.basecls = basecls

463

464 def recv_raw(self, x=MTU):

465 # type: (int) -> Tuple[Optional[Type[Packet]], Optional[bytes], Optional[float]]

466 return self.basecls, self.ins.recv(x), None

467

468 if WINDOWS:

469 @staticmethod

470 def select(sockets, remain=None):

471 # type: (List[SuperSocket], Optional[float]) -> List[SuperSocket]

472 from scapy.automaton import select_objects

473 return select_objects(sockets, remain)

474

475

476class StreamSocket(SimpleSocket):

477 """

478 Wrap a stream socket into a layer 2 SuperSocket

479

480 :param sock: the socket to wrap

481 :param basecls: the base class packet to use to dissect the packet

482 """

483 desc = "transforms a stream socket into a layer 2"

484

485 def __init__(self,

486 sock, # type: socket.socket

487 basecls=None, # type: Optional[Type[Packet]]

488 ):

489 # type: (...) -> None

490 from scapy.sessions import streamcls

491 self.rcvcls = streamcls(basecls or conf.raw_layer)

492 self.metadata: Dict[str, Any] = {}

493 self.streamsession: Dict[str, Any] = {}

494 self._buf = b""

495 super(StreamSocket, self).__init__(sock, basecls=basecls)

496

497 def recv(self, x=None, **kwargs):

498 # type: (Optional[int], Any) -> Optional[Packet]

499 if x is None:

500 x = MTU

501

502 while True:

503 # Block but in PEEK mode

504 data = self.ins.recv(x, socket.MSG_PEEK)

505 if data == b"":

506 raise EOFError

507 x = len(data)

508 pkt = self.rcvcls(self._buf + data, self.metadata, self.streamsession)

509 if pkt is None: # Incomplete packet.

510 self._buf += self.ins.recv(x)

511 else:

512 break

513

514 self.metadata.clear()

515 # Strip any madding

516 pad = pkt.getlayer(conf.padding_layer)

517 if pad is not None and pad.underlayer is not None:

518 del pad.underlayer.payload

519 while pad is not None and not isinstance(pad, NoPayload):

520 x -= len(pad.load)

521 pad = pad.payload

522 # Only receive the packet length

523 self.ins.recv(x)

524 self._buf = b""

525 return pkt

526

527

528class StreamSocketPeekless(StreamSocket):

529 desc = "StreamSocket that doesn't use MSG_PEEK"

530

531 def __init__(self, sock, basecls=None):

532 # type: (socket.socket, Optional[Type[Packet]]) -> None

533 from scapy.sessions import TCPSession

534 self.sess = TCPSession(app=True)

535 super(StreamSocketPeekless, self).__init__(sock, basecls)

536

537 # 65535, the default value of x is the maximum length of a TLS record

538 def recv(self, x=None, **kwargs):

539 # type: (Optional[int], **Any) -> Optional[Packet]

540 if x is None:

541 x = MTU

542 while True:

543 # Block

544 try:

545 data = self.ins.recv(x)

546 except OSError:

547 raise EOFError

548 try:

549 pkt = self.sess.process(data, cls=self.basecls) # type: ignore

550 except struct.error:

551 # Buffer underflow

552 pkt = None

553 if data == b"" and not pkt:

554 raise EOFError

555 if pkt:

556 return pkt

557

558 @property

559 def streamsession(self) -> Dict[Any, Any]:

560 return self.sess.session

561

562 @streamsession.setter

563 def streamsession(self, session: Dict[Any, Any]) -> None:

564 self.sess.session = session

565

566 @staticmethod

567 def select(sockets, remain=None):

568 # type: (List[SuperSocket], Optional[float]) -> List[SuperSocket]

569 queued = [

570 x

571 for x in sockets

572 if isinstance(x, StreamSocketPeekless) and x.sess.data

573 ]

574 if queued:

575 return queued # type: ignore

576 return super(StreamSocketPeekless, StreamSocketPeekless).select(

577 sockets,

578 remain=remain,

579 )

580

581

582# Old name: SSLStreamSocket

583SSLStreamSocket = StreamSocketPeekless

584

585

586class L2ListenTcpdump(SuperSocket):

587 desc = "read packets at layer 2 using tcpdump"

588

589 def __init__(self,

590 iface=None, # type: Optional[_GlobInterfaceType]

591 promisc=None, # type: Optional[bool]

592 filter=None, # type: Optional[str]

593 nofilter=False, # type: bool

594 prog=None, # type: Optional[str]

595 quiet=False, # type: bool

596 *arg, # type: Any

597 **karg # type: Any

598 ):

599 # type: (...) -> None

600 self.outs = None

601 args = ['-w', '-', '-s', '65535']

602 self.iface = "any"

603 if iface is None and (WINDOWS or DARWIN):

604 self.iface = iface = conf.iface

605 if promisc is None:

606 promisc = conf.sniff_promisc

607 if iface is not None:

608 args.extend(['-i', network_name(iface)])

609 if not promisc:

610 args.append('-p')

611 if not nofilter:

612 if conf.except_filter:

613 if filter:

614 filter = "(%s) and not (%s)" % (filter, conf.except_filter)

615 else:

616 filter = "not (%s)" % conf.except_filter

617 if filter is not None:

618 args.append(filter)

619 self.tcpdump_proc = tcpdump(

620 None, prog=prog, args=args, getproc=True, quiet=quiet)

621 self.reader = PcapReader(self.tcpdump_proc.stdout)

622 self.ins = self.reader # type: ignore

623

624 def recv(self, x=MTU, **kwargs):

625 # type: (int, **Any) -> Optional[Packet]

626 return self.reader.recv(x, **kwargs)

627

628 def close(self):

629 # type: () -> None

630 SuperSocket.close(self)

631 self.tcpdump_proc.kill()

632

633 @staticmethod

634 def select(sockets, remain=None):

635 # type: (List[SuperSocket], Optional[float]) -> List[SuperSocket]

636 if (WINDOWS or DARWIN):

637 return sockets

638 return SuperSocket.select(sockets, remain=remain)

639

640

641# More abstract objects

642

643class IterSocket(SuperSocket):

644 desc = "wrapper around an iterable"

645 nonblocking_socket = True

646

647 def __init__(self, obj):

648 # type: (_PacketIterable) -> None

649 if not obj:

650 self.iter = iter([]) # type: Iterator[Packet]

651 elif isinstance(obj, IterSocket):

652 self.iter = obj.iter

653 elif isinstance(obj, SndRcvList):

654 def _iter(obj=cast(SndRcvList, obj)):

655 # type: (SndRcvList) -> Iterator[Packet]

656 for s, r in obj:

657 if s.sent_time:

658 s.time = s.sent_time

659 yield s

660 yield r

661

662 self.iter = _iter()

663 elif isinstance(obj, (list, PacketList)):

664 if isinstance(obj[0], bytes):

665 self.iter = iter(obj)

666 else:

667 self.iter = (y for x in obj for y in x)

668 else:

669 self.iter = obj.__iter__()

670

671 @staticmethod

672 def select(sockets, remain=None):

673 # type: (List[SuperSocket], Any) -> List[SuperSocket]

674 return sockets

675

676 def recv(self, x=None, **kwargs):

677 # type: (Optional[int], Any) -> Optional[Packet]

678 try:

679 pkt = next(self.iter)

680 if isinstance(pkt, bytes):

681 # Raw bytes carry no link-layer information

682 return conf.raw_layer(pkt, **kwargs)

683 return pkt.__class__(bytes(pkt), **kwargs)

684 except StopIteration:

685 raise EOFError

686

687 def close(self):

688 # type: () -> None

689 pass