Graphes a conversations between sources and destinations and display it (using graphviz and imagemagick) :param getsrcdst: a function that takes an element of the list and returns the source, the destination and optionally a label. By default, returns the IP
(self,
getsrcdst=None, # type: Optional[Callable[[Packet], Tuple[Any, ...]]] # noqa: E501
**kargs # type: Any
)
| 457 | ) |
| 458 | |
| 459 | def conversations(self, |
| 460 | getsrcdst=None, # type: Optional[Callable[[Packet], Tuple[Any, ...]]] # noqa: E501 |
| 461 | **kargs # type: Any |
| 462 | ): |
| 463 | # type: (...) -> Any |
| 464 | """Graphes a conversations between sources and destinations and display it |
| 465 | (using graphviz and imagemagick) |
| 466 | |
| 467 | :param getsrcdst: a function that takes an element of the list and |
| 468 | returns the source, the destination and optionally |
| 469 | a label. By default, returns the IP source and |
| 470 | destination from IP and ARP layers |
| 471 | :param type: output type (svg, ps, gif, jpg, etc.), passed to dot's |
| 472 | "-T" option |
| 473 | :param target: filename or redirect. Defaults pipe to Imagemagick's |
| 474 | display program |
| 475 | :param prog: which graphviz program to use |
| 476 | """ |
| 477 | if getsrcdst is None: |
| 478 | def _getsrcdst(pkt): |
| 479 | # type: (Packet) -> Tuple[str, str] |
| 480 | """Extract src and dst addresses""" |
| 481 | if 'IP' in pkt: |
| 482 | return (pkt['IP'].src, pkt['IP'].dst) |
| 483 | if 'IPv6' in pkt: |
| 484 | return (pkt['IPv6'].src, pkt['IPv6'].dst) |
| 485 | if 'ARP' in pkt: |
| 486 | return (pkt['ARP'].psrc, pkt['ARP'].pdst) |
| 487 | raise TypeError() |
| 488 | getsrcdst = _getsrcdst |
| 489 | conv = {} # type: Dict[Tuple[Any, ...], Any] |
| 490 | for elt in self.res: |
| 491 | p = self._elt2pkt(elt) |
| 492 | try: |
| 493 | c = getsrcdst(p) |
| 494 | except Exception: |
| 495 | # No warning here: it's OK that getsrcdst() raises an |
| 496 | # exception, since it might be, for example, a |
| 497 | # function that expects a specific layer in each |
| 498 | # packet. The try/except approach is faster and |
| 499 | # considered more Pythonic than adding tests. |
| 500 | continue |
| 501 | if len(c) == 3: |
| 502 | conv.setdefault(c[:2], set()).add(c[2]) |
| 503 | else: |
| 504 | conv[c] = conv.get(c, 0) + 1 |
| 505 | gr = 'digraph "conv" {\n' |
| 506 | for (s, d), l in conv.items(): |
| 507 | gr += '\t "%s" -> "%s" [label="%s"]\n' % ( |
| 508 | s, d, ', '.join(str(x) for x in l) if isinstance(l, set) else l |
| 509 | ) |
| 510 | gr += "}\n" |
| 511 | return do_graph(gr, **kargs) |
| 512 | |
| 513 | def afterglow(self, |
| 514 | src=None, # type: Optional[Callable[[_Inner], Any]] |