Build networkx graph + layout from current memory state. Returns (G, pos, node_colors, edge_colors) or (None, ...) if nothing to show. live=True uses spring_layout (~100ms); live=False tries graphviz first. For each (src, dst) pair only the highest-severity edge is kept so malicious/
(live: bool = False, filter_option: str = "All", node_filter: str = "")
| 33 | |
| 34 | |
| 35 | def _build_graph_data(live: bool = False, filter_option: str = "All", node_filter: str = ""): |
| 36 | """Build networkx graph + layout from current memory state. |
| 37 | |
| 38 | Returns (G, pos, node_colors, edge_colors) or (None, ...) if nothing to show. |
| 39 | live=True uses spring_layout (~100ms); live=False tries graphviz first. |
| 40 | For each (src, dst) pair only the highest-severity edge is kept so malicious/ |
| 41 | Tor edges are never overdrawn by lower-priority normal edges. |
| 42 | node_filter: if non-empty, keep only edges where src or dst label/IP contains it. |
| 43 | """ |
| 44 | try: |
| 45 | import networkx as nx |
| 46 | except ImportError: |
| 47 | return None, None, None, None |
| 48 | |
| 49 | # Collect highest-priority edge per (src_label, dst_label) pair. |
| 50 | # priority: malicious=4 > tor=3 > covert=2 > normal=1 |
| 51 | edge_best: dict[tuple, tuple] = {} |
| 52 | |
| 53 | for session_key, session in list(memory.packet_db.items()): |
| 54 | parts = session_key.split("/") |
| 55 | if len(parts) != 3: |
| 56 | continue |
| 57 | src_ip, dst_ip, port = parts |
| 58 | |
| 59 | # Apply traffic filter |
| 60 | if filter_option == "Malicious" and session_key not in memory.possible_mal_traffic: |
| 61 | continue |
| 62 | if filter_option == "Tor" and session_key not in memory.possible_tor_traffic: |
| 63 | continue |
| 64 | if filter_option == "Covert" and not session.covert: |
| 65 | continue |
| 66 | if filter_option == "HTTP" and port != "80": |
| 67 | continue |
| 68 | if filter_option == "HTTPS" and port != "443": |
| 69 | continue |
| 70 | if filter_option == "DNS" and port != "53": |
| 71 | continue |
| 72 | if filter_option == "ICMP" and port != "ICMP": |
| 73 | continue |
| 74 | |
| 75 | eth_src = session.Ethernet.get("src", "") |
| 76 | eth_dst = session.Ethernet.get("dst", "") |
| 77 | |
| 78 | if eth_src and eth_src in memory.lan_hosts: |
| 79 | src_label, src_kind = _mac_label(eth_src), "lan" |
| 80 | else: |
| 81 | src_label, src_kind = src_ip, "ext" |
| 82 | |
| 83 | if dst_ip in memory.destination_hosts: |
| 84 | dst_mac = memory.destination_hosts[dst_ip].mac |
| 85 | if dst_mac in memory.lan_hosts: |
| 86 | dst_label, dst_kind = _mac_label(dst_mac), "lan" |
| 87 | else: |
| 88 | gw_id = dst_mac.replace(":", "")[-6:] if dst_mac else dst_ip[-4:] |
| 89 | dst_label, dst_kind = f"GW:{gw_id}", "gw" |
| 90 | else: |
| 91 | if eth_dst and eth_dst in memory.lan_hosts: |
| 92 | dst_label, dst_kind = _mac_label(eth_dst), "lan" |
no test coverage detected