Get traffic analysis summary with security-focused metrics
(self)
| 1582 | 'nick': session.nick, |
| 1583 | 'user': session.user, |
| 1584 | 'channels': sorted(session.channels), |
| 1585 | 'server_port': session.server_port, |
| 1586 | 'server_banner': session.server_banner[:3], |
| 1587 | 'mitre': ['T1071.001', 'T1102'], |
| 1588 | }, |
| 1589 | ) |
| 1590 | |
| 1591 | def get_tls_fingerprints(self, limit: int = 100) -> List[Dict[str, Any]]: |
| 1592 | if self._ja3_collector is None: |
| 1593 | return [] |
| 1594 | return self._ja3_collector.get_records(limit=limit) |
| 1595 | |
| 1596 | def get_irc_sessions(self, limit: int = 50) -> List[Dict[str, Any]]: |
| 1597 | if self._irc_parser is None: |
| 1598 | return [] |
| 1599 | return self._irc_parser.get_sessions(limit=limit) |
| 1600 | |
| 1601 | def _check_dns_tunneling(self, src_ip: str): |
| 1602 | """Detect potential DNS tunneling based on query frequency""" |
| 1603 | current_time = time.time() |
| 1604 | |
| 1605 | # Track this query timestamp |
| 1606 | self._dns_query_times[src_ip].append(current_time) |
| 1607 | |
| 1608 | # Count queries in the tracking window |
| 1609 | window_start = current_time - self.DNS_QUERY_TRACKING_WINDOW |
| 1610 | recent_queries = sum(1 for t in self._dns_query_times[src_ip] if t > window_start) |
| 1611 | |
| 1612 | if recent_queries > self.DNS_TUNNEL_THRESHOLD: |
| 1613 | self._create_alert( |
| 1614 | level=TrafficAlertLevel.HIGH, |
| 1615 | category=AlertCategory.DNS_TUNNELING.value, |
| 1616 | message=f"Potential DNS tunneling: {src_ip} made {recent_queries} queries in {self.DNS_QUERY_TRACKING_WINDOW}s", |
| 1617 | src_ip=src_ip, |
| 1618 | details={ |
| 1619 | 'queries_per_minute': recent_queries, |
| 1620 | 'threshold': self.DNS_TUNNEL_THRESHOLD, |
| 1621 | 'window_seconds': self.DNS_QUERY_TRACKING_WINDOW |
| 1622 | } |
| 1623 | ) |
| 1624 | |
| 1625 | def _create_alert(self, level: TrafficAlertLevel, category: str, |
| 1626 | message: str, src_ip: str = None, dst_ip: str = None, |
| 1627 | details: Dict = None, dedup_key: str = None): |
| 1628 | """Create a traffic alert with rate limiting and deduplication. |
| 1629 | |
| 1630 | dedup_key: optional override for the dedup hash. Use when the default |
no test coverage detected