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> 

5 

6""" 

7SuperSocket. 

8""" 

9 

10from select import select, error as select_error 

11import ctypes 

12import errno 

13import socket 

14import struct 

15import time 

16 

17from scapy.config import conf 

18from scapy.consts import DARWIN, WINDOWS 

19from scapy.data import ( 

20 MTU, 

21 ETH_P_IP, 

22 ETH_P_IPV6, 

23 SOL_PACKET, 

24 SO_TIMESTAMPNS, 

25) 

26from scapy.compat import raw 

27from scapy.error import warning, log_runtime 

28from scapy.interfaces import network_name 

29from scapy.packet import Packet, NoPayload 

30from scapy.plist import ( 

31 PacketList, 

32 SndRcvList, 

33 _PacketIterable, 

34) 

35from scapy.utils import PcapReader, tcpdump 

36 

37# Typing imports 

38from scapy.interfaces import _GlobInterfaceType 

39from typing import ( 

40 Any, 

41 Dict, 

42 Iterator, 

43 List, 

44 Optional, 

45 Tuple, 

46 Type, 

47 TypeVar, 

48 cast, 

49 TYPE_CHECKING, 

50) 

51from scapy.compat import Self 

52 

53if TYPE_CHECKING: 

54 from scapy.ansmachine import AnsweringMachine 

55 

56 

57# Utils 

58 

59 

60class _SuperSocket_metaclass(type): 

61 desc = None # type: Optional[str] 

62 

63 def __repr__(self): 

64 # type: () -> str 

65 if self.desc is not None: 

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

67 else: 

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

69 

70 

71# Used to get ancillary data 

72PACKET_AUXDATA = 8 

73ETH_P_8021Q = 0x8100 

74TP_STATUS_VLAN_VALID = 1 << 4 

75TP_STATUS_VLAN_TPID_VALID = 1 << 6 

76 

77 

78class tpacket_auxdata(ctypes.Structure): 

79 _fields_ = [ 

80 ("tp_status", ctypes.c_uint), 

81 ("tp_len", ctypes.c_uint), 

82 ("tp_snaplen", ctypes.c_uint), 

83 ("tp_mac", ctypes.c_ushort), 

84 ("tp_net", ctypes.c_ushort), 

85 ("tp_vlan_tci", ctypes.c_ushort), 

86 ("tp_vlan_tpid", ctypes.c_ushort), 

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

88 

89 

90# SuperSocket 

91 

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

93 

94 

95class SuperSocket(metaclass=_SuperSocket_metaclass): 

96 closed = False # type: bool 

97 nonblocking_socket = False # type: bool 

98 auxdata_available = False # type: bool 

99 

100 def __init__(self, 

101 family=socket.AF_INET, # type: int 

102 type=socket.SOCK_STREAM, # type: int 

103 proto=0, # type: int 

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

105 **kwargs # type: Any 

106 ): 

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

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

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

110 self.promisc = conf.sniff_promisc 

111 self.iface = iface or conf.iface 

112 

113 def send(self, x): 

114 # type: (Packet) -> int 

115 """Sends a `Packet` object 

116 

117 :param x: `Packet` to be send 

118 :return: Number of bytes that have been sent 

119 """ 

120 sx = raw(x) 

121 try: 

122 x.sent_time = time.time() 

123 except AttributeError: 

124 pass 

125 

126 if self.outs: 

127 return self.outs.send(sx) 

128 else: 

129 return 0 

130 

131 if WINDOWS: 

132 def _recv_raw(self, sock, x): 

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

134 """Internal function to receive a Packet. 

135 

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

137 :param x: Number of bytes to be received 

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

139 """ 

140 pkt, sa_ll = sock.recvfrom(x) 

141 return pkt, sa_ll, None 

142 else: 

143 def _recv_raw(self, sock, x): 

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

145 """Internal function to receive a Packet, 

146 and process ancillary data. 

147 

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

149 :param x: Number of bytes to be received 

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

151 """ 

152 timestamp = None 

153 if not self.auxdata_available: 

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

155 return pkt, sa_ll, timestamp 

156 flags_len = socket.CMSG_LEN(4096) 

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

158 if not pkt: 

159 return pkt, sa_ll, timestamp 

160 for cmsg_lvl, cmsg_type, cmsg_data in ancdata: 

161 # Check available ancillary data 

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

163 # Parse AUXDATA 

164 try: 

165 auxdata = tpacket_auxdata.from_buffer_copy(cmsg_data) 

166 except ValueError: 

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

168 # can return a truncated message. A ValueError 

169 # exception likely indicates that Auxiliary 

170 # Data is not supported by the Linux kernel. 

171 return pkt, sa_ll, timestamp 

172 if auxdata.tp_vlan_tci != 0 or \ 

173 auxdata.tp_status & TP_STATUS_VLAN_VALID: 

174 # Insert VLAN tag 

175 tpid = ETH_P_8021Q 

176 if auxdata.tp_status & TP_STATUS_VLAN_TPID_VALID: 

177 tpid = auxdata.tp_vlan_tpid 

178 tag = struct.pack( 

179 "!HH", 

180 tpid, 

181 auxdata.tp_vlan_tci 

182 ) 

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

184 elif cmsg_lvl == socket.SOL_SOCKET and \ 

185 cmsg_type == SO_TIMESTAMPNS: 

186 length = len(cmsg_data) 

187 if length == 16: # __kernel_timespec 

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

189 elif length == 8: # timespec 

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

191 else: 

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

193 continue 

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

195 return pkt, sa_ll, timestamp 

196 

197 def recv_raw(self, x=MTU): 

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

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

200 

201 

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

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

204 and a timestamp 

205 """ 

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

207 

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

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

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

211 

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

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

214 """ 

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

216 if not val or not cls: 

217 return None 

218 try: 

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

220 except KeyboardInterrupt: 

221 raise 

222 except Exception: 

223 if conf.debug_dissector: 

224 from scapy.sendrecv import debug 

225 debug.crashed_on = (cls, val) 

226 raise 

227 pkt = conf.raw_layer(val) 

228 if ts: 

229 pkt.time = ts 

230 return pkt 

231 

232 def fileno(self): 

233 # type: () -> int 

234 return self.ins.fileno() 

235 

236 def close(self): 

237 # type: () -> None 

238 """Gracefully close this socket 

239 """ 

240 if self.closed: 

241 return 

242 self.closed = True 

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

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

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

246 self.outs.close() 

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

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

249 self.ins.close() 

250 

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

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

253 """Send and Receive multiple packets 

254 """ 

255 from scapy import sendrecv 

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

257 

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

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

260 """Send one packet and receive one answer 

261 """ 

262 from scapy import sendrecv 

263 # if not explicitly specified by the user, 

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

265 # for a Thread creation 

266 kargs.setdefault("threaded", False) 

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

268 if len(ans) > 0: 

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

270 return pkt 

271 else: 

272 return None 

273 

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

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

276 from scapy import sendrecv 

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

278 

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

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

281 from scapy import sendrecv 

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

283 

284 def am(self, 

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

286 **kwargs # type: Any 

287 ): 

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

289 """ 

290 Creates an AnsweringMachine associated with this socket. 

291 

292 :param cls: A subclass of AnsweringMachine to instantiate 

293 """ 

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

295 

296 @staticmethod 

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

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

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

300 the available sockets. 

301 

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

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

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

305 """ 

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

307 try: 

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

309 except (IOError, select_error) as exc: 

310 # select.error has no .errno attribute 

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

312 raise 

313 return inp 

314 

315 def __del__(self): 

316 # type: () -> None 

317 """Close the socket""" 

318 self.close() 

319 

320 def __enter__(self): 

321 # type: () -> Self 

322 return self 

323 

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

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

326 """Close the socket""" 

327 self.close() 

328 

329 

330if not WINDOWS: 

331 class L3RawSocket(SuperSocket): 

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

333 

334 def __init__(self, 

335 type=ETH_P_IP, # type: int 

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

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

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

339 nofilter=0 # type: int 

340 ): 

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

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

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

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

345 if iface is not None: 

346 iface = network_name(iface) 

347 self.iface = iface 

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

349 else: 

350 self.iface = "any" 

351 try: 

352 # Receive Auxiliary Data (VLAN tags) 

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

354 self.ins.setsockopt( 

355 socket.SOL_SOCKET, 

356 SO_TIMESTAMPNS, 

357 1 

358 ) 

359 self.auxdata_available = True 

360 except OSError: 

361 # Note: Auxiliary Data is only supported since 

362 # Linux 2.6.21 

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

364 log_runtime.info(msg) 

365 

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

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

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

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

370 return None 

371 if sa_ll[3] in conf.l2types: 

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

373 lvl = 2 

374 elif sa_ll[1] in conf.l3types: 

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

376 lvl = 3 

377 else: 

378 cls = conf.default_l2 

379 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 

380 lvl = 3 

381 

382 try: 

383 pkt = cls(data, **kwargs) 

384 except KeyboardInterrupt: 

385 raise 

386 except Exception: 

387 if conf.debug_dissector: 

388 raise 

389 pkt = conf.raw_layer(data) 

390 

391 if lvl == 2: 

392 pkt = pkt.payload 

393 

394 if pkt is not None: 

395 if ts is None: 

396 from scapy.arch.linux import get_last_packet_timestamp 

397 ts = get_last_packet_timestamp(self.ins) 

398 pkt.time = ts 

399 return pkt 

400 

401 def send(self, x): 

402 # type: (Packet) -> int 

403 try: 

404 sx = raw(x) 

405 if self.outs: 

406 x.sent_time = time.time() 

407 return self.outs.sendto( 

408 sx, 

409 (x.dst, 0) 

410 ) 

411 except AttributeError: 

412 raise ValueError( 

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

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

415 "IP layer)" 

416 ) 

417 except socket.error as msg: 

418 log_runtime.error(msg) 

419 return 0 

420 

421 class L3RawSocket6(L3RawSocket): 

422 def __init__(self, 

423 type: int = ETH_P_IPV6, 

424 filter: Optional[str] = None, 

425 iface: Optional[_GlobInterfaceType] = None, 

426 promisc: Optional[bool] = None, 

427 nofilter: bool = False) -> None: 

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

429 self.outs = socket.socket( 

430 socket.AF_INET6, 

431 socket.SOCK_RAW, 

432 socket.IPPROTO_RAW 

433 ) 

434 self.ins = socket.socket( 

435 socket.AF_PACKET, 

436 socket.SOCK_RAW, 

437 socket.htons(type) 

438 ) 

439 self.iface = cast(_GlobInterfaceType, iface) 

440 

441 

442class SimpleSocket(SuperSocket): 

443 desc = "wrapper around a classic socket" 

444 __selectable_force_select__ = True 

445 

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

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

448 self.ins = sock 

449 self.outs = sock 

450 if basecls is None: 

451 basecls = conf.raw_layer 

452 self.basecls = basecls 

453 

454 def recv_raw(self, x=MTU): 

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

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

457 

458 if WINDOWS: 

459 @staticmethod 

460 def select(sockets, remain=None): 

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

462 from scapy.automaton import select_objects 

463 return select_objects(sockets, remain) 

464 

465 

466class StreamSocket(SimpleSocket): 

467 """ 

468 Wrap a stream socket into a layer 2 SuperSocket 

469 

470 :param sock: the socket to wrap 

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

472 """ 

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

474 

475 def __init__(self, 

476 sock, # type: socket.socket 

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

478 ): 

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

480 from scapy.sessions import streamcls 

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

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

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

484 self._buf = b"" 

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

486 

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

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

489 if x is None: 

490 x = MTU 

491 

492 while True: 

493 # Block but in PEEK mode 

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

495 if data == b"": 

496 raise EOFError 

497 x = len(data) 

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

499 if pkt is None: # Incomplete packet. 

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

501 else: 

502 break 

503 

504 self.metadata.clear() 

505 # Strip any madding 

506 pad = pkt.getlayer(conf.padding_layer) 

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

508 del pad.underlayer.payload 

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

510 x -= len(pad.load) 

511 pad = pad.payload 

512 # Only receive the packet length 

513 self.ins.recv(x) 

514 self._buf = b"" 

515 return pkt 

516 

517 

518class StreamSocketPeekless(StreamSocket): 

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

520 

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

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

523 from scapy.sessions import TCPSession 

524 self.sess = TCPSession(app=True) 

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

526 

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

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

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

530 if x is None: 

531 x = MTU 

532 # Block 

533 try: 

534 data = self.ins.recv(x) 

535 except OSError: 

536 raise EOFError 

537 try: 

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

539 except struct.error: 

540 # Buffer underflow 

541 pkt = None 

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

543 raise EOFError 

544 if not pkt: 

545 return self.recv(x) 

546 return pkt 

547 

548 @staticmethod 

549 def select(sockets, remain=None): 

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

551 queued = [ 

552 x 

553 for x in sockets 

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

555 ] 

556 if queued: 

557 return queued # type: ignore 

558 return super(StreamSocketPeekless, StreamSocketPeekless).select( 

559 sockets, 

560 remain=remain, 

561 ) 

562 

563 

564# Old name: SSLStreamSocket 

565SSLStreamSocket = StreamSocketPeekless 

566 

567 

568class L2ListenTcpdump(SuperSocket): 

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

570 

571 def __init__(self, 

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

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

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

575 nofilter=False, # type: bool 

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

577 quiet=False, # type: bool 

578 *arg, # type: Any 

579 **karg # type: Any 

580 ): 

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

582 self.outs = None 

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

584 self.iface = "any" 

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

586 self.iface = iface = conf.iface 

587 if promisc is None: 

588 promisc = conf.sniff_promisc 

589 if iface is not None: 

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

591 if not promisc: 

592 args.append('-p') 

593 if not nofilter: 

594 if conf.except_filter: 

595 if filter: 

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

597 else: 

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

599 if filter is not None: 

600 args.append(filter) 

601 self.tcpdump_proc = tcpdump( 

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

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

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

605 

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

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

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

609 

610 def close(self): 

611 # type: () -> None 

612 SuperSocket.close(self) 

613 self.tcpdump_proc.kill() 

614 

615 @staticmethod 

616 def select(sockets, remain=None): 

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

618 if (WINDOWS or DARWIN): 

619 return sockets 

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

621 

622 

623# More abstract objects 

624 

625class IterSocket(SuperSocket): 

626 desc = "wrapper around an iterable" 

627 nonblocking_socket = True 

628 

629 def __init__(self, obj): 

630 # type: (_PacketIterable) -> None 

631 if not obj: 

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

633 elif isinstance(obj, IterSocket): 

634 self.iter = obj.iter 

635 elif isinstance(obj, SndRcvList): 

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

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

638 for s, r in obj: 

639 if s.sent_time: 

640 s.time = s.sent_time 

641 yield s 

642 yield r 

643 

644 self.iter = _iter() 

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

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

647 self.iter = iter(obj) 

648 else: 

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

650 else: 

651 self.iter = obj.__iter__() 

652 

653 @staticmethod 

654 def select(sockets, remain=None): 

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

656 return sockets 

657 

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

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

660 try: 

661 pkt = next(self.iter) 

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

663 except StopIteration: 

664 raise EOFError 

665 

666 def close(self): 

667 # type: () -> None 

668 pass