Create a traffic alert with rate limiting and deduplication. dedup_key: optional override for the dedup hash. Use when the default (category, src, dst) tuple would collapse distinct alerts — e.g. sweep alerts that vary by port rather than dst_ip.
(self, level: TrafficAlertLevel, category: str,
message: str, src_ip: str = None, dst_ip: str = None,
details: Dict = None, dedup_key: str = None)
| 1485 | a full tshark dissector (~290MB RSS plus a ~155MB dumpcap child), so the |
| 1486 | pair alone outweighs everything a Pi Zero has. The tcpdump core runs on |
| 1487 | any board; the sidecars need a big one, hence the separate capability. |
| 1488 | """ |
| 1489 | caps = get_server_capabilities(self.shared_data).capabilities |
| 1490 | if not caps.traffic_sidecars_enabled: |
| 1491 | logger.info( |
| 1492 | f"JA3/IRC sidecars skipped: they need tshark and a board with at " |
| 1493 | f"least {ServerCapabilities.TRAFFIC_SIDECAR_MIN_RAM_GB:.1f}GB RAM " |
| 1494 | f"(this one reports {caps.total_ram_gb:.2f}GB). Packet capture, " |
| 1495 | f"beacon and port-scan detection are unaffected.") |
| 1496 | return |
| 1497 | |
| 1498 | try: |
| 1499 | from tls_fingerprint import JA3Collector |
| 1500 | except Exception as exc: |
| 1501 | logger.debug(f"JA3 collector unavailable: {exc}") |
| 1502 | JA3Collector = None # type: ignore |
| 1503 | |
| 1504 | if JA3Collector is not None and self._ja3_collector is None: |
| 1505 | self._ja3_collector = JA3Collector( |
| 1506 | interface=self.interface, |
| 1507 | on_match=self._on_ja3_match, |
| 1508 | ) |
| 1509 | if not self._ja3_collector.start(): |
| 1510 | self._ja3_collector = None |
| 1511 | |
| 1512 | try: |
| 1513 | from irc_dpi import IRCDPIParser |
| 1514 | except Exception as exc: |
| 1515 | logger.debug(f"IRC DPI unavailable: {exc}") |
| 1516 | IRCDPIParser = None # type: ignore |
| 1517 | |
| 1518 | if IRCDPIParser is not None and self._irc_parser is None: |
| 1519 | self._irc_parser = IRCDPIParser( |
| 1520 | interface=self.interface, |
| 1521 | on_session_event=self._on_irc_event, |
| 1522 | ) |
| 1523 | if not self._irc_parser.start(): |
| 1524 | self._irc_parser = None |
| 1525 | |
| 1526 | def _stop_sidecars(self) -> None: |
| 1527 | for attr in ('_ja3_collector', '_irc_parser'): |
| 1528 | sidecar = getattr(self, attr, None) |
| 1529 | if sidecar is not None: |
| 1530 | try: |
| 1531 | sidecar.stop() |
| 1532 | except Exception: |
| 1533 | pass |
| 1534 | setattr(self, attr, None) |
| 1535 | |
| 1536 | def _on_ja3_match(self, record) -> None: |
| 1537 | """Callback: a TLS fingerprint matched a known signature.""" |
| 1538 | m = record.match |
| 1539 | if not m: |
| 1540 | return |
| 1541 | if m.category == 'malware' or m.confidence == 'high': |
| 1542 | level = TrafficAlertLevel.HIGH |
| 1543 | elif m.confidence == 'medium': |
no test coverage detected