Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/scapy/config.py: 54%

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

7Implementation of the configuration object. 

8""" 

9 

10 

11import atexit 

12import copy 

13import functools 

14import os 

15import re 

16import socket 

17import sys 

18import time 

19import warnings 

20 

21from dataclasses import dataclass 

22from enum import Enum 

23 

24import importlib 

25import importlib.abc 

26import importlib.util 

27 

28import scapy 

29from scapy import VERSION 

30from scapy.base_classes import BasePacket 

31from scapy.consts import DARWIN, WINDOWS, LINUX, BSD, SOLARIS 

32from scapy.error import ( 

33 log_loading, 

34 log_scapy, 

35 ScapyInvalidPlatformException, 

36 warning, 

37) 

38from scapy.themes import ColorTheme, NoTheme, apply_ipython_style 

39 

40# Typing imports 

41from typing import ( 

42 cast, 

43 Any, 

44 Callable, 

45 Dict, 

46 Iterator, 

47 List, 

48 NoReturn, 

49 Optional, 

50 Set, 

51 Tuple, 

52 Type, 

53 Union, 

54 overload, 

55 TYPE_CHECKING, 

56) 

57from types import ModuleType 

58from scapy.compat import DecoratorCallable 

59 

60if TYPE_CHECKING: 

61 # Do not import at runtime 

62 import scapy.as_resolvers 

63 from scapy.modules.nmap import NmapKnowledgeBase 

64 from scapy.packet import Packet 

65 from scapy.supersocket import SuperSocket # noqa: F401 

66 import scapy.asn1.asn1 

67 import scapy.asn1.mib 

68 

69############ 

70# Config # 

71############ 

72 

73 

74class ConfClass(object): 

75 def configure(self, cnf): 

76 # type: (ConfClass) -> None 

77 self.__dict__ = cnf.__dict__.copy() 

78 

79 def __repr__(self): 

80 # type: () -> str 

81 return str(self) 

82 

83 def __str__(self): 

84 # type: () -> str 

85 s = "" 

86 dkeys = self.__class__.__dict__.copy() 

87 dkeys.update(self.__dict__) 

88 keys = sorted(dkeys) 

89 for i in keys: 

90 if i[0] != "_": 

91 r = repr(getattr(self, i)) 

92 r = " ".join(r.split()) 

93 wlen = 76 - max(len(i), 10) 

94 if len(r) > wlen: 

95 r = r[:wlen - 3] + "..." 

96 s += "%-10s = %s\n" % (i, r) 

97 return s[:-1] 

98 

99 

100class Interceptor(object): 

101 def __init__(self, 

102 name, # type: str 

103 default, # type: Any 

104 hook, # type: Callable[..., Any] 

105 args=None, # type: Optional[List[Any]] 

106 kargs=None # type: Optional[Dict[str, Any]] 

107 ): 

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

109 self.name = name 

110 self.intname = "_intercepted_%s" % name 

111 self.default = default 

112 self.hook = hook 

113 self.args = args if args is not None else [] 

114 self.kargs = kargs if kargs is not None else {} 

115 

116 def __get__(self, obj, typ=None): 

117 # type: (Conf, Optional[type]) -> Any 

118 if not hasattr(obj, self.intname): 

119 setattr(obj, self.intname, self.default) 

120 return getattr(obj, self.intname) 

121 

122 @staticmethod 

123 def set_from_hook(obj, name, val): 

124 # type: (Conf, str, bool) -> None 

125 int_name = "_intercepted_%s" % name 

126 setattr(obj, int_name, val) 

127 

128 def __set__(self, obj, val): 

129 # type: (Conf, Any) -> None 

130 old = getattr(obj, self.intname, self.default) 

131 val = self.hook(self.name, val, old, *self.args, **self.kargs) 

132 setattr(obj, self.intname, val) 

133 

134 

135def _readonly(name): 

136 # type: (str) -> NoReturn 

137 default = Conf.__dict__[name].default 

138 Interceptor.set_from_hook(conf, name, default) 

139 raise ValueError("Read-only value !") 

140 

141 

142ReadOnlyAttribute = functools.partial( 

143 Interceptor, 

144 hook=(lambda name, *args, **kwargs: _readonly(name)) 

145) 

146ReadOnlyAttribute.__doc__ = "Read-only class attribute" 

147 

148 

149class ProgPath(ConfClass): 

150 _default: str = "<System default>" 

151 universal_open: str = "open" if DARWIN else "xdg-open" 

152 pdfreader: str = universal_open 

153 psreader: str = universal_open 

154 svgreader: str = universal_open 

155 dot: str = "dot" 

156 display: str = "display" 

157 tcpdump: str = "tcpdump" 

158 tcpreplay: str = "tcpreplay" 

159 hexedit: str = "hexer" 

160 tshark: str = "tshark" 

161 wireshark: str = "wireshark" 

162 ifconfig: str = "ifconfig" 

163 extcap_folders: List[str] = [ 

164 os.path.join(os.path.expanduser("~"), ".config", "wireshark", "extcap"), 

165 "/usr/lib/x86_64-linux-gnu/wireshark/extcap", 

166 ] 

167 

168 

169class ConfigFieldList: 

170 def __init__(self): 

171 # type: () -> None 

172 self.fields = set() # type: Set[Any] 

173 self.layers = set() # type: Set[Any] 

174 

175 @staticmethod 

176 def _is_field(f): 

177 # type: (Any) -> bool 

178 return hasattr(f, "owners") 

179 

180 def _recalc_layer_list(self): 

181 # type: () -> None 

182 self.layers = {owner for f in self.fields for owner in f.owners} 

183 

184 def add(self, *flds): 

185 # type: (*Any) -> None 

186 self.fields |= {f for f in flds if self._is_field(f)} 

187 self._recalc_layer_list() 

188 

189 def remove(self, *flds): 

190 # type: (*Any) -> None 

191 self.fields -= set(flds) 

192 self._recalc_layer_list() 

193 

194 def __contains__(self, elt): 

195 # type: (Any) -> bool 

196 if isinstance(elt, BasePacket): 

197 return elt in self.layers 

198 return elt in self.fields 

199 

200 def __repr__(self): 

201 # type: () -> str 

202 return "<%s [%s]>" % (self.__class__.__name__, " ".join(str(x) for x in self.fields)) # noqa: E501 

203 

204 

205class Emphasize(ConfigFieldList): 

206 pass 

207 

208 

209class Resolve(ConfigFieldList): 

210 pass 

211 

212 

213class Num2Layer: 

214 def __init__(self): 

215 # type: () -> None 

216 self.num2layer = {} # type: Dict[int, Type[Packet]] 

217 self.layer2num = {} # type: Dict[Type[Packet], int] 

218 

219 def register(self, num, layer): 

220 # type: (int, Type[Packet]) -> None 

221 self.register_num2layer(num, layer) 

222 self.register_layer2num(num, layer) 

223 

224 def register_num2layer(self, num, layer): 

225 # type: (int, Type[Packet]) -> None 

226 self.num2layer[num] = layer 

227 

228 def register_layer2num(self, num, layer): 

229 # type: (int, Type[Packet]) -> None 

230 self.layer2num[layer] = num 

231 

232 @overload 

233 def __getitem__(self, item): 

234 # type: (Type[Packet]) -> int 

235 pass 

236 

237 @overload 

238 def __getitem__(self, item): # noqa: F811 

239 # type: (int) -> Type[Packet] 

240 pass 

241 

242 def __getitem__(self, item): # noqa: F811 

243 # type: (Union[int, Type[Packet]]) -> Union[int, Type[Packet]] 

244 if isinstance(item, int): 

245 return self.num2layer[item] 

246 else: 

247 return self.layer2num[item] 

248 

249 def __contains__(self, item): 

250 # type: (Union[int, Type[Packet]]) -> bool 

251 if isinstance(item, int): 

252 return item in self.num2layer 

253 else: 

254 return item in self.layer2num 

255 

256 def get(self, 

257 item, # type: Union[int, Type[Packet]] 

258 default=None, # type: Optional[Type[Packet]] 

259 ): 

260 # type: (...) -> Optional[Union[int, Type[Packet]]] 

261 return self[item] if item in self else default 

262 

263 def __repr__(self): 

264 # type: () -> str 

265 lst = [] 

266 for num, layer in self.num2layer.items(): 

267 if layer in self.layer2num and self.layer2num[layer] == num: 

268 dir = "<->" 

269 else: 

270 dir = " ->" 

271 lst.append((num, "%#6x %s %-20s (%s)" % (num, dir, layer.__name__, 

272 layer._name))) 

273 for layer, num in self.layer2num.items(): 

274 if num not in self.num2layer or self.num2layer[num] != layer: 

275 lst.append((num, "%#6x <- %-20s (%s)" % (num, layer.__name__, 

276 layer._name))) 

277 lst.sort() 

278 return "\n".join(y for x, y in lst) 

279 

280 

281class LayersList(List[Type['scapy.packet.Packet']]): 

282 def __init__(self): 

283 # type: () -> None 

284 list.__init__(self) 

285 self.ldict = {} # type: Dict[str, List[Type[Packet]]] 

286 self.filtered = False 

287 self._backup_dict = {} # type: Dict[Type[Packet], List[Tuple[Dict[str, Any], Type[Packet]]]] # noqa: E501 

288 

289 def __repr__(self): 

290 # type: () -> str 

291 return "\n".join("%-20s: %s" % (layer.__name__, layer.name) 

292 for layer in self) 

293 

294 def register(self, layer): 

295 # type: (Type[Packet]) -> None 

296 self.append(layer) 

297 if layer.__module__ not in self.ldict: 

298 self.ldict[layer.__module__] = [] 

299 self.ldict[layer.__module__].append(layer) 

300 

301 def layers(self): 

302 # type: () -> List[Tuple[str, str]] 

303 result = [] 

304 # This import may feel useless, but it is required for the eval below 

305 import scapy # noqa: F401 

306 try: 

307 import builtins # noqa: F401 

308 except ImportError: 

309 import __builtin__ # noqa: F401 

310 for lay in self.ldict: 

311 doc = eval(lay).__doc__ 

312 result.append((lay, doc.strip().split("\n")[0] if doc else lay)) 

313 return result 

314 

315 def filter(self, items): 

316 # type: (List[Type[Packet]]) -> None 

317 """Disable dissection of unused layers to speed up dissection""" 

318 if self.filtered: 

319 raise ValueError("Already filtered. Please disable it first") 

320 for lay in self.ldict.values(): 

321 for cls in lay: 

322 if cls not in self._backup_dict: 

323 self._backup_dict[cls] = cls.payload_guess[:] 

324 cls.payload_guess = [ 

325 y for y in cls.payload_guess if y[1] in items 

326 ] 

327 self.filtered = True 

328 

329 def unfilter(self): 

330 # type: () -> None 

331 """Re-enable dissection for all layers""" 

332 if not self.filtered: 

333 raise ValueError("Not filtered. Please filter first") 

334 for lay in self.ldict.values(): 

335 for cls in lay: 

336 cls.payload_guess = self._backup_dict[cls] 

337 self._backup_dict.clear() 

338 self.filtered = False 

339 

340 

341class CommandsList(List[Callable[..., Any]]): 

342 def __repr__(self): 

343 # type: () -> str 

344 s = [] 

345 for li in sorted(self, key=lambda x: x.__name__): 

346 doc = li.__doc__ if li.__doc__ else "--" 

347 doc = doc.lstrip().split('\n', 1)[0] 

348 s.append("%-22s: %s" % (li.__name__, doc)) 

349 return "\n".join(s) 

350 

351 def register(self, cmd): 

352 # type: (DecoratorCallable) -> DecoratorCallable 

353 self.append(cmd) 

354 return cmd # return cmd so that method can be used as a decorator 

355 

356 

357def lsc(): 

358 # type: () -> None 

359 """Displays Scapy's default commands""" 

360 print(repr(conf.commands)) 

361 

362 

363class CacheInstance(Dict[str, Any]): 

364 __slots__ = ["timeout", "name", "_timetable"] 

365 

366 def __init__(self, name="noname", timeout=None): 

367 # type: (str, Optional[int]) -> None 

368 self.timeout = timeout 

369 self.name = name 

370 self._timetable = {} # type: Dict[str, float] 

371 

372 def flush(self): 

373 # type: () -> None 

374 self._timetable.clear() 

375 self.clear() 

376 

377 def __getitem__(self, item): 

378 # type: (str) -> Any 

379 if item in self.__slots__: 

380 return object.__getattribute__(self, item) 

381 if not self.__contains__(item): 

382 raise KeyError(item) 

383 return super(CacheInstance, self).__getitem__(item) 

384 

385 def __contains__(self, item): 

386 if not super(CacheInstance, self).__contains__(item): 

387 return False 

388 if self.timeout is not None: 

389 t = self._timetable[item] 

390 if time.time() - t > self.timeout: 

391 return False 

392 return True 

393 

394 def get(self, item, default=None): 

395 # type: (str, Optional[Any]) -> Any 

396 # overloading this method is needed to force the dict to go through 

397 # the timetable check 

398 try: 

399 return self[item] 

400 except KeyError: 

401 return default 

402 

403 def __setitem__(self, item, v): 

404 # type: (str, str) -> None 

405 if item in self.__slots__: 

406 return object.__setattr__(self, item, v) 

407 self._timetable[item] = time.time() 

408 super(CacheInstance, self).__setitem__(item, v) 

409 

410 def update(self, 

411 other, # type: Any 

412 **kwargs # type: Any 

413 ): 

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

415 for key, value in other.items(): 

416 # We only update an element from `other` either if it does 

417 # not exist in `self` or if the entry in `self` is older. 

418 if key not in self or self._timetable[key] < other._timetable[key]: 

419 dict.__setitem__(self, key, value) 

420 self._timetable[key] = other._timetable[key] 

421 

422 def iteritems(self): 

423 # type: () -> Iterator[Tuple[str, Any]] 

424 if self.timeout is None: 

425 return super(CacheInstance, self).items() 

426 t0 = time.time() 

427 return ( 

428 (k, v) 

429 for (k, v) in super(CacheInstance, self).items() 

430 if t0 - self._timetable[k] < self.timeout 

431 ) 

432 

433 def iterkeys(self): 

434 # type: () -> Iterator[str] 

435 if self.timeout is None: 

436 return super(CacheInstance, self).keys() 

437 t0 = time.time() 

438 return ( 

439 k 

440 for k in super(CacheInstance, self).keys() 

441 if t0 - self._timetable[k] < self.timeout 

442 ) 

443 

444 def __iter__(self): 

445 # type: () -> Iterator[str] 

446 return self.iterkeys() 

447 

448 def itervalues(self): 

449 # type: () -> Iterator[Tuple[str, Any]] 

450 if self.timeout is None: 

451 return super(CacheInstance, self).values() 

452 t0 = time.time() 

453 return ( 

454 v 

455 for (k, v) in super(CacheInstance, self).items() 

456 if t0 - self._timetable[k] < self.timeout 

457 ) 

458 

459 def items(self): 

460 # type: () -> Any 

461 return list(self.iteritems()) 

462 

463 def keys(self): 

464 # type: () -> Any 

465 return list(self.iterkeys()) 

466 

467 def values(self): 

468 # type: () -> Any 

469 return list(self.itervalues()) 

470 

471 def __len__(self): 

472 # type: () -> int 

473 if self.timeout is None: 

474 return super(CacheInstance, self).__len__() 

475 return len(self.keys()) 

476 

477 def summary(self): 

478 # type: () -> str 

479 return "%s: %i valid items. Timeout=%rs" % (self.name, len(self), self.timeout) # noqa: E501 

480 

481 def __repr__(self): 

482 # type: () -> str 

483 s = [] 

484 if self: 

485 mk = max(len(k) for k in self) 

486 fmt = "%%-%is %%s" % (mk + 1) 

487 for item in self.items(): 

488 s.append(fmt % item) 

489 return "\n".join(s) 

490 

491 def copy(self): 

492 # type: () -> CacheInstance 

493 return copy.copy(self) 

494 

495 

496class NetCache: 

497 def __init__(self): 

498 # type: () -> None 

499 self._caches_list = [] # type: List[CacheInstance] 

500 

501 def add_cache(self, cache): 

502 # type: (CacheInstance) -> None 

503 self._caches_list.append(cache) 

504 setattr(self, cache.name, cache) 

505 

506 def new_cache(self, name, timeout=None): 

507 # type: (str, Optional[int]) -> CacheInstance 

508 c = CacheInstance(name=name, timeout=timeout) 

509 self.add_cache(c) 

510 return c 

511 

512 def __delattr__(self, attr): 

513 # type: (str) -> NoReturn 

514 raise AttributeError("Cannot delete attributes") 

515 

516 def update(self, other): 

517 # type: (NetCache) -> None 

518 for co in other._caches_list: 

519 if hasattr(self, co.name): 

520 getattr(self, co.name).update(co) 

521 else: 

522 self.add_cache(co.copy()) 

523 

524 def flush(self): 

525 # type: () -> None 

526 for c in self._caches_list: 

527 c.flush() 

528 

529 def __repr__(self): 

530 # type: () -> str 

531 return "\n".join(c.summary() for c in self._caches_list) 

532 

533 

534class ScapyExt: 

535 __slots__ = ["specs", "name", "version"] 

536 

537 class MODE(Enum): 

538 LAYERS = "layers" 

539 CONTRIB = "contrib" 

540 MODULES = "modules" 

541 

542 @dataclass 

543 class ScapyExtSpec: 

544 fullname: str 

545 mode: 'ScapyExt.MODE' 

546 spec: Any 

547 default: bool 

548 

549 def __init__(self): 

550 self.specs: Dict[str, 'ScapyExt.ScapyExtSpec'] = {} 

551 

552 def config(self, name, version): 

553 self.name = name 

554 self.version = version 

555 

556 def register(self, name, mode, path, default=None): 

557 assert mode in self.MODE, "mode must be one of ScapyExt.MODE !" 

558 fullname = f"scapy.{mode.value}.{name}" 

559 spec = importlib.util.spec_from_file_location( 

560 fullname, 

561 str(path), 

562 ) 

563 spec = self.ScapyExtSpec( 

564 fullname=fullname, 

565 mode=mode, 

566 spec=spec, 

567 default=default or False, 

568 ) 

569 if default is None: 

570 spec.default = bool(importlib.util.find_spec(spec.fullname)) 

571 self.specs[fullname] = spec 

572 

573 def __repr__(self): 

574 return "<ScapyExt %s %s (%s specs)>" % ( 

575 self.name, 

576 self.version, 

577 len(self.specs), 

578 ) 

579 

580 

581class ExtsManager(importlib.abc.MetaPathFinder): 

582 __slots__ = ["exts", "_loaded", "all_specs"] 

583 

584 SCAPY_PLUGIN_CLASSIFIER = 'Framework :: Scapy' 

585 GPLV2_CLASSIFIERS = [ 

586 'License :: OSI Approved :: GNU General Public License v2 (GPLv2)', 

587 'License :: OSI Approved :: GNU General Public License v2 or later (GPLv2+)', 

588 ] 

589 

590 def __init__(self): 

591 self.exts: List[ScapyExt] = [] 

592 self.all_specs: Dict[str, ScapyExt.ScapyExtSpec] = {} 

593 self._loaded = [] 

594 

595 def find_spec(self, fullname, path, target=None): 

596 if fullname in self.all_specs: 

597 return self.all_specs[fullname].spec 

598 

599 def invalidate_caches(self): 

600 pass 

601 

602 def _register_spec(self, spec): 

603 self.all_specs[spec.fullname] = spec 

604 if spec.default: 

605 loader = importlib.util.LazyLoader(spec.spec.loader) 

606 spec.spec.loader = loader 

607 module = importlib.util.module_from_spec(spec.spec) 

608 sys.modules[spec.fullname] = module 

609 loader.exec_module(module) 

610 

611 def load(self): 

612 try: 

613 import importlib.metadata 

614 except ImportError: 

615 return 

616 for distr in importlib.metadata.distributions(): 

617 if any( 

618 v == self.SCAPY_PLUGIN_CLASSIFIER 

619 for k, v in distr.metadata.items() if k == 'Classifier' 

620 ): 

621 try: 

622 pkg = next( 

623 k 

624 for k, v in importlib.metadata.packages_distributions().items() 

625 if distr.name in v 

626 ) 

627 except KeyError: 

628 pkg = distr.name 

629 if pkg in self._loaded: 

630 continue 

631 if not any( 

632 v in self.GPLV2_CLASSIFIERS 

633 for k, v in distr.metadata.items() if k == 'Classifier' 

634 ): 

635 log_loading.warning( 

636 "'%s' has no GPLv2 classifier therefore cannot be loaded." % pkg # noqa: E501 

637 ) 

638 continue 

639 self._loaded.append(pkg) 

640 ext = ScapyExt() 

641 try: 

642 scapy_ext = importlib.import_module(pkg) 

643 except Exception as ex: 

644 log_loading.warning( 

645 "'%s' failed during import with %s" % ( 

646 pkg, 

647 ex 

648 ) 

649 ) 

650 continue 

651 try: 

652 scapy_ext_func = scapy_ext.scapy_ext 

653 except AttributeError: 

654 log_loading.info( 

655 "'%s' included the Scapy Framework specifier " 

656 "but did not include a scapy_ext" % pkg 

657 ) 

658 continue 

659 try: 

660 scapy_ext_func(ext) 

661 except Exception as ex: 

662 log_loading.warning( 

663 "'%s' failed during initialization with %s" % ( 

664 pkg, 

665 ex 

666 ) 

667 ) 

668 continue 

669 for spec in ext.specs.values(): 

670 self._register_spec(spec) 

671 self.exts.append(ext) 

672 if self not in sys.meta_path: 

673 sys.meta_path.append(self) 

674 

675 def __repr__(self): 

676 from scapy.utils import pretty_list 

677 return pretty_list( 

678 [ 

679 (x.name, x.version, [y.fullname for y in x.specs.values()]) 

680 for x in self.exts 

681 ], 

682 [("Name", "Version", "Specs")], 

683 sortBy=0, 

684 ) 

685 

686 

687def _version_checker(module, minver): 

688 # type: (ModuleType, Tuple[int, ...]) -> bool 

689 """Checks that module has a higher version that minver. 

690 

691 params: 

692 - module: a module to test 

693 - minver: a tuple of versions 

694 """ 

695 # We could use LooseVersion, but distutils imports imp which is deprecated 

696 version_regexp = r'[a-z]?((?:\d|\.)+\d+)(?:\.dev[0-9]+)?' 

697 version_tags_r = re.match( 

698 version_regexp, 

699 getattr(module, "__version__", "") 

700 ) 

701 if not version_tags_r: 

702 return False 

703 version_tags_i = version_tags_r.group(1).split(".") 

704 version_tags = tuple(int(x) for x in version_tags_i) 

705 return bool(version_tags >= minver) 

706 

707 

708def isCryptographyValid(): 

709 # type: () -> bool 

710 """ 

711 Check if the cryptography module >= 2.0.0 is present. This is the minimum 

712 version for most usages in Scapy. 

713 """ 

714 try: 

715 import cryptography 

716 except ImportError: 

717 return False 

718 return _version_checker(cryptography, (2, 0, 0)) 

719 

720 

721def isCryptographyAdvanced(): 

722 # type: () -> bool 

723 """ 

724 Check if the cryptography module is present, and if it supports X25519, 

725 ChaCha20Poly1305 and such. 

726 

727 Notes: 

728 - cryptography >= 2.0 is required 

729 - OpenSSL >= 1.1.0 is required 

730 """ 

731 try: 

732 from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey # noqa: E501 

733 X25519PrivateKey.generate() 

734 except Exception: 

735 return False 

736 else: 

737 return True 

738 

739 

740def isPyPy(): 

741 # type: () -> bool 

742 """Returns either scapy is running under PyPy or not""" 

743 try: 

744 import __pypy__ # noqa: F401 

745 return True 

746 except ImportError: 

747 return False 

748 

749 

750def _prompt_changer(attr, val, old): 

751 # type: (str, Any, Any) -> Any 

752 """Change the current prompt theme""" 

753 Interceptor.set_from_hook(conf, attr, val) 

754 try: 

755 sys.ps1 = conf.color_theme.prompt(conf.prompt) 

756 except Exception: 

757 pass 

758 try: 

759 apply_ipython_style( 

760 get_ipython() # type: ignore 

761 ) 

762 except NameError: 

763 pass 

764 return getattr(conf, attr, old) 

765 

766 

767def _set_conf_sockets(): 

768 # type: () -> None 

769 """Populate the conf.L2Socket and conf.L3Socket 

770 according to the various use_* parameters 

771 """ 

772 if conf.use_bpf and not BSD: 

773 Interceptor.set_from_hook(conf, "use_bpf", False) 

774 raise ScapyInvalidPlatformException("BSD-like (OSX, *BSD...) only !") 

775 if not conf.use_pcap and SOLARIS: 

776 Interceptor.set_from_hook(conf, "use_pcap", True) 

777 raise ScapyInvalidPlatformException( 

778 "Scapy only supports libpcap on Solaris !" 

779 ) 

780 # we are already in an Interceptor hook, use Interceptor.set_from_hook 

781 if conf.use_pcap: 

782 try: 

783 from scapy.arch.libpcap import L2pcapListenSocket, L2pcapSocket, \ 

784 L3pcapSocket 

785 except (OSError, ImportError): 

786 log_loading.warning("No libpcap provider available ! pcap won't be used") 

787 Interceptor.set_from_hook(conf, "use_pcap", False) 

788 else: 

789 conf.L3socket = L3pcapSocket 

790 conf.L3socket6 = functools.partial( 

791 L3pcapSocket, filter="ip6") 

792 conf.L2socket = L2pcapSocket 

793 conf.L2listen = L2pcapListenSocket 

794 elif conf.use_bpf: 

795 from scapy.arch.bpf.supersocket import L2bpfListenSocket, \ 

796 L2bpfSocket, L3bpfSocket 

797 conf.L3socket = L3bpfSocket 

798 conf.L3socket6 = functools.partial( 

799 L3bpfSocket, filter="ip6") 

800 conf.L2socket = L2bpfSocket 

801 conf.L2listen = L2bpfListenSocket 

802 elif LINUX: 

803 from scapy.arch.linux import L3PacketSocket, L2Socket, L2ListenSocket 

804 conf.L3socket = L3PacketSocket 

805 conf.L3socket6 = cast( 

806 "Type[SuperSocket]", 

807 functools.partial( 

808 L3PacketSocket, 

809 filter="ip6" 

810 ) 

811 ) 

812 conf.L2socket = L2Socket 

813 conf.L2listen = L2ListenSocket 

814 elif WINDOWS: 

815 from scapy.arch.windows import _NotAvailableSocket 

816 from scapy.arch.windows.native import L3WinSocket, L3WinSocket6 

817 conf.L3socket = L3WinSocket 

818 conf.L3socket6 = L3WinSocket6 

819 conf.L2socket = _NotAvailableSocket 

820 conf.L2listen = _NotAvailableSocket 

821 else: 

822 from scapy.supersocket import L3RawSocket, L3RawSocket6 

823 conf.L3socket = L3RawSocket 

824 conf.L3socket6 = L3RawSocket6 

825 # Reload the interfaces 

826 conf.ifaces.reload()