Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/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 

298 # Skip arch* modules 

299 if layer.__module__.startswith("scapy.arch."): 

300 return 

301 

302 # Register in module 

303 if layer.__module__ not in self.ldict: 

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

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

306 

307 def layers(self): 

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

309 result = [] 

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

311 import scapy # noqa: F401 

312 try: 

313 import builtins # noqa: F401 

314 except ImportError: 

315 import __builtin__ # noqa: F401 

316 for lay in self.ldict: 

317 doc = eval(lay).__doc__ 

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

319 return result 

320 

321 def filter(self, items): 

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

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

324 if self.filtered: 

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

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

327 for cls in lay: 

328 if cls not in self._backup_dict: 

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

330 cls.payload_guess = [ 

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

332 ] 

333 self.filtered = True 

334 

335 def unfilter(self): 

336 # type: () -> None 

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

338 if not self.filtered: 

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

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

341 for cls in lay: 

342 cls.payload_guess = self._backup_dict[cls] 

343 self._backup_dict.clear() 

344 self.filtered = False 

345 

346 

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

348 def __repr__(self): 

349 # type: () -> str 

350 s = [] 

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

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

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

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

355 return "\n".join(s) 

356 

357 def register(self, cmd): 

358 # type: (DecoratorCallable) -> DecoratorCallable 

359 self.append(cmd) 

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

361 

362 

363def lsc(): 

364 # type: () -> None 

365 """Displays Scapy's default commands""" 

366 print(repr(conf.commands)) 

367 

368 

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

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

371 

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

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

374 self.timeout = timeout 

375 self.name = name 

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

377 

378 def flush(self): 

379 # type: () -> None 

380 self._timetable.clear() 

381 self.clear() 

382 

383 def __getitem__(self, item): 

384 # type: (str) -> Any 

385 if item in self.__slots__: 

386 return object.__getattribute__(self, item) 

387 if not self.__contains__(item): 

388 raise KeyError(item) 

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

390 

391 def __contains__(self, item): 

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

393 return False 

394 if self.timeout is not None: 

395 t = self._timetable[item] 

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

397 return False 

398 return True 

399 

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

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

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

403 # the timetable check 

404 try: 

405 return self[item] 

406 except KeyError: 

407 return default 

408 

409 def __setitem__(self, item, v): 

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

411 if item in self.__slots__: 

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

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

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

415 

416 def update(self, 

417 other, # type: Any 

418 **kwargs # type: Any 

419 ): 

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

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

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

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

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

425 dict.__setitem__(self, key, value) 

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

427 

428 def iteritems(self): 

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

430 if self.timeout is None: 

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

432 t0 = time.time() 

433 return ( 

434 (k, v) 

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

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

437 ) 

438 

439 def iterkeys(self): 

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

441 if self.timeout is None: 

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

443 t0 = time.time() 

444 return ( 

445 k 

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

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

448 ) 

449 

450 def __iter__(self): 

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

452 return self.iterkeys() 

453 

454 def itervalues(self): 

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

456 if self.timeout is None: 

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

458 t0 = time.time() 

459 return ( 

460 v 

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

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

463 ) 

464 

465 def items(self): 

466 # type: () -> Any 

467 return list(self.iteritems()) 

468 

469 def keys(self): 

470 # type: () -> Any 

471 return list(self.iterkeys()) 

472 

473 def values(self): 

474 # type: () -> Any 

475 return list(self.itervalues()) 

476 

477 def __len__(self): 

478 # type: () -> int 

479 if self.timeout is None: 

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

481 return len(self.keys()) 

482 

483 def summary(self): 

484 # type: () -> str 

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

486 

487 def __repr__(self): 

488 # type: () -> str 

489 s = [] 

490 if self: 

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

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

493 for item in self.items(): 

494 s.append(fmt % item) 

495 return "\n".join(s) 

496 

497 def copy(self): 

498 # type: () -> CacheInstance 

499 return copy.copy(self) 

500 

501 

502class NetCache: 

503 def __init__(self): 

504 # type: () -> None 

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

506 

507 def add_cache(self, cache): 

508 # type: (CacheInstance) -> None 

509 self._caches_list.append(cache) 

510 setattr(self, cache.name, cache) 

511 

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

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

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

515 self.add_cache(c) 

516 return c 

517 

518 def __delattr__(self, attr): 

519 # type: (str) -> NoReturn 

520 raise AttributeError("Cannot delete attributes") 

521 

522 def update(self, other): 

523 # type: (NetCache) -> None 

524 for co in other._caches_list: 

525 if hasattr(self, co.name): 

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

527 else: 

528 self.add_cache(co.copy()) 

529 

530 def flush(self): 

531 # type: () -> None 

532 for c in self._caches_list: 

533 c.flush() 

534 

535 def __repr__(self): 

536 # type: () -> str 

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

538 

539 

540class ScapyExt: 

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

542 

543 class MODE(Enum): 

544 LAYERS = "layers" 

545 CONTRIB = "contrib" 

546 MODULES = "modules" 

547 

548 @dataclass 

549 class ScapyExtSpec: 

550 fullname: str 

551 mode: 'ScapyExt.MODE' 

552 spec: Any 

553 default: bool 

554 

555 def __init__(self): 

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

557 

558 def config(self, name, version): 

559 self.name = name 

560 self.version = version 

561 

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

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

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

565 spec = importlib.util.spec_from_file_location( 

566 fullname, 

567 str(path), 

568 ) 

569 spec = self.ScapyExtSpec( 

570 fullname=fullname, 

571 mode=mode, 

572 spec=spec, 

573 default=default or False, 

574 ) 

575 if default is None: 

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

577 self.specs[fullname] = spec 

578 

579 def __repr__(self): 

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

581 self.name, 

582 self.version, 

583 len(self.specs), 

584 ) 

585 

586 

587class ExtsManager(importlib.abc.MetaPathFinder): 

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

589 

590 SCAPY_PLUGIN_CLASSIFIER = 'Framework :: Scapy' 

591 GPLV2_CLASSIFIERS = [ 

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

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

594 ] 

595 

596 def __init__(self): 

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

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

599 self._loaded = [] 

600 

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

602 if fullname in self.all_specs: 

603 return self.all_specs[fullname].spec 

604 

605 def invalidate_caches(self): 

606 pass 

607 

608 def _register_spec(self, spec): 

609 self.all_specs[spec.fullname] = spec 

610 if spec.default: 

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

612 spec.spec.loader = loader 

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

614 sys.modules[spec.fullname] = module 

615 loader.exec_module(module) 

616 

617 def load(self): 

618 try: 

619 import importlib.metadata 

620 except ImportError: 

621 return 

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

623 if any( 

624 v == self.SCAPY_PLUGIN_CLASSIFIER 

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

626 ): 

627 try: 

628 # Python 3.13 raises an internal warning when calling this 

629 with warnings.catch_warnings(): 

630 warnings.filterwarnings("ignore", category=DeprecationWarning) 

631 pkg = next( 

632 k 

633 for k, v in 

634 importlib.metadata.packages_distributions().items() 

635 if distr.name in v 

636 ) 

637 except KeyError: 

638 pkg = distr.name 

639 if pkg in self._loaded: 

640 continue 

641 if not any( 

642 v in self.GPLV2_CLASSIFIERS 

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

644 ): 

645 log_loading.warning( 

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

647 ) 

648 continue 

649 self._loaded.append(pkg) 

650 ext = ScapyExt() 

651 try: 

652 scapy_ext = importlib.import_module(pkg) 

653 except Exception as ex: 

654 log_loading.warning( 

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

656 pkg, 

657 ex 

658 ) 

659 ) 

660 continue 

661 try: 

662 scapy_ext_func = scapy_ext.scapy_ext 

663 except AttributeError: 

664 log_loading.info( 

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

666 "but did not include a scapy_ext" % pkg 

667 ) 

668 continue 

669 try: 

670 scapy_ext_func(ext) 

671 except Exception as ex: 

672 log_loading.warning( 

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

674 pkg, 

675 ex 

676 ) 

677 ) 

678 continue 

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

680 self._register_spec(spec) 

681 self.exts.append(ext) 

682 if self not in sys.meta_path: 

683 sys.meta_path.append(self) 

684 

685 def __repr__(self): 

686 from scapy.utils import pretty_list 

687 return pretty_list( 

688 [ 

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

690 for x in self.exts 

691 ], 

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

693 sortBy=0, 

694 ) 

695 

696 

697def _version_checker(module, minver): 

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

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

700 

701 params: 

702 - module: a module to test 

703 - minver: a tuple of versions 

704 """ 

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

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

707 version_tags_r = re.match( 

708 version_regexp, 

709 getattr(module, "__version__", "") 

710 ) 

711 if not version_tags_r: 

712 return False 

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

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

715 return bool(version_tags >= minver) 

716 

717 

718def isCryptographyValid(): 

719 # type: () -> bool 

720 """ 

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

722 version for most usages in Scapy. 

723 """ 

724 try: 

725 import cryptography 

726 except ImportError: 

727 return False 

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

729 

730 

731def isCryptographyAdvanced(): 

732 # type: () -> bool 

733 """ 

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

735 ChaCha20Poly1305 and such. 

736 

737 Notes: 

738 - cryptography >= 2.0 is required 

739 - OpenSSL >= 1.1.0 is required 

740 """ 

741 try: 

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

743 X25519PrivateKey.generate() 

744 except Exception: 

745 return False 

746 else: 

747 return True 

748 

749 

750def isPyPy(): 

751 # type: () -> bool 

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

753 try: 

754 import __pypy__ # noqa: F401 

755 return True 

756 except ImportError: 

757 return False 

758 

759 

760def _prompt_changer(attr, val, old): 

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

762 """Change the current prompt theme""" 

763 Interceptor.set_from_hook(conf, attr, val) 

764 try: 

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

766 except Exception: 

767 pass 

768 try: 

769 apply_ipython_style( 

770 get_ipython() # type: ignore 

771 ) 

772 except NameError: 

773 pass 

774 return getattr(conf, attr, old) 

775 

776 

777def _set_conf_sockets(): 

778 # type: () -> None 

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

780 according to the various use_* parameters 

781 """ 

782 if conf.use_bpf and not BSD: 

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

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

785 if not conf.use_pcap and SOLARIS: 

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

787 raise ScapyInvalidPlatformException( 

788 "Scapy only supports libpcap on Solaris !" 

789 ) 

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

791 if conf.use_pcap: 

792 try: 

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

794 L3pcapSocket 

795 except (OSError, ImportError): 

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

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

798 else: 

799 conf.L3socket = L3pcapSocket 

800 conf.L3socket6 = functools.partial( 

801 L3pcapSocket, filter="ip6") 

802 conf.L2socket = L2pcapSocket 

803 conf.L2listen = L2pcapListenSocket 

804 elif conf.use_bpf: 

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

806 L2bpfSocket, L3bpfSocket 

807 conf.L3socket = L3bpfSocket 

808 conf.L3socket6 = functools.partial( 

809 L3bpfSocket, filter="ip6") 

810 conf.L2socket = L2bpfSocket 

811 conf.L2listen = L2bpfListenSocket 

812 elif LINUX: 

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

814 conf.L3socket = L3PacketSocket 

815 conf.L3socket6 = cast( 

816 "Type[SuperSocket]", 

817 functools.partial( 

818 L3PacketSocket, 

819 filter="ip6" 

820 ) 

821 ) 

822 conf.L2socket = L2Socket 

823 conf.L2listen = L2ListenSocket 

824 elif WINDOWS: 

825 from scapy.arch.windows import _NotAvailableSocket 

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

827 conf.L3socket = L3WinSocket 

828 conf.L3socket6 = L3WinSocket6