"""Graph Database Query Engine - Neo4j Implementation""" import json import logging from typing import Dict, Any, List, Optional from neo4j import GraphDatabase, Driver, Session, Result from neo4j.exceptions import Neo4jError, ServiceUnavailable from app.config import settings logger = logging.getLogger(__name__) class GraphQueryEngine: """Neo4j graph database query engine""" def __init__(self): self.driver: Optional[Driver] = None self.database = settings.neo4j_database try: self._connect() except Exception as e: logger.warning(f"⚠️ Neo4j connection failed (will retry on first query): {e}") def _connect(self): """Connect to Neo4j database""" try: self.driver = GraphDatabase.driver( settings.neo4j_bolt_uri, auth=(settings.neo4j_user, settings.neo4j_password), max_connection_pool_size=10, connection_timeout=settings.query_timeout, max_transaction_retry_time=settings.query_timeout ) # Verify connectivity self.driver.verify_connectivity() logger.info(f"✅ Connected to Neo4j: {settings.neo4j_bolt_uri} (database: {self.database})") except Exception as e: logger.error(f"❌ Failed to connect to Neo4j: {e}") raise def execute_query(self, query: str, parameters: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: """Execute a Cypher query and return results""" if not self.driver: try: self._connect() except Exception as e: return {"success": False, "error": f"Connection failed: {str(e)}"} try: with self.driver.session(database=self.database) as session: result = session.run(query, parameters or {}) # Convert result to list of dictionaries records = [] for record in result: records.append(dict(record)) return { "success": True, "data": records, "count": len(records) } except Neo4jError as e: logger.error(f"❌ Neo4j query error: {e}") return { "success": False, "error": f"Query failed: {e.message}", "code": e.code } except Exception as e: logger.error(f"❌ Unexpected error: {e}") return { "success": False, "error": str(e) } def get_workload_dependencies(self, workload_id: str, depth: int = 1) -> Dict[str, Any]: """Get dependencies of a workload (downstream)""" query = """ MATCH path = (w:Workload {id: $workload_id})-[r*1..$depth]->(dep) WHERE ALL(rel IN relationships(path) WHERE rel.is_active = true) RETURN w.id AS source_id, w.name AS source_name, w.kind AS source_kind, [node IN nodes(path)[1..] | { id: node.id, name: node.name, kind: node.kind, namespace: node.namespace }] AS dependencies, [rel IN relationships(path) | type(rel)] AS relationship_types, length(path) AS path_length ORDER BY path_length LIMIT 100 """ return self.execute_query(query, {"workload_id": workload_id, "depth": depth}) def get_workload_dependents(self, workload_id: str, depth: int = 1) -> Dict[str, Any]: """Get dependents of a workload (upstream)""" query = """ MATCH path = (dep)-[r*1..$depth]->(w:Workload {id: $workload_id}) WHERE ALL(rel IN relationships(path) WHERE rel.is_active = true) RETURN w.id AS target_id, w.name AS target_name, w.kind AS target_kind, [node IN nodes(path)[..-1] | { id: node.id, name: node.name, kind: node.kind, namespace: node.namespace }] AS dependents, [rel IN relationships(path) | type(rel)] AS relationship_types, length(path) AS path_length ORDER BY path_length LIMIT 100 """ return self.execute_query(query, {"workload_id": workload_id, "depth": depth}) def get_communications( self, source_id: Optional[str] = None, destination_id: Optional[str] = None, namespace: Optional[str] = None, protocol: Optional[str] = None, analysis_id: Optional[str] = None, cluster_id: Optional[str] = None, start_time: Optional[str] = None, end_time: Optional[str] = None, limit: int = 100 ) -> Dict[str, Any]: """Get communications between workloads""" conditions = [] params = {"limit": limit} # Multi-cluster support: analysis_id can be in format '{id}' or '{id}-{cluster_id}' # Filter by analysis_id if provided - match both single and multi-cluster formats if analysis_id: analysis_id_str = str(analysis_id) analysis_id_prefix = f"{analysis_id_str}-" conditions.append( "(comm.analysis_id = $analysis_id OR comm.analysis_id STARTS WITH $analysis_id_prefix OR " "src.analysis_id = $analysis_id OR src.analysis_id STARTS WITH $analysis_id_prefix)" ) params["analysis_id"] = analysis_id_str params["analysis_id_prefix"] = analysis_id_prefix # Filter by cluster_id if provided (optional for multi-cluster) if cluster_id: conditions.append("(src.cluster_id = $cluster_id OR comm.cluster_id = $cluster_id)") params["cluster_id"] = str(cluster_id) if source_id: conditions.append("src.id = $source_id") params["source_id"] = source_id if destination_id: conditions.append("dst.id = $destination_id") params["destination_id"] = destination_id if namespace: # Include edges where source is in namespace, OR destination is in namespace, # OR source is in namespace AND destination is external (ExternalEndpoint) conditions.append( "(src.namespace = $namespace OR dst.namespace = $namespace OR " "(src.namespace = $namespace AND (dst:ExternalEndpoint OR dst.namespace = 'external')))" ) params["namespace"] = namespace if protocol: conditions.append("comm.protocol = $protocol") params["protocol"] = protocol # Time range filtering - filter by last_seen timestamp # Note: last_seen is stored as epoch milliseconds (from Neo4j timestamp() function) # We need to convert ISO datetime string to epoch ms for comparison if start_time: conditions.append("comm.last_seen >= datetime($start_time).epochMillis") params["start_time"] = start_time if end_time: conditions.append("comm.last_seen <= datetime($end_time).epochMillis") params["end_time"] = end_time # Build WHERE clause where_clause = " AND ".join(conditions) if conditions else "true" # Query ALL communications including to ExternalEndpoints # Note: Use same pattern as get_communication_stats (no label constraint) # This works for both single-cluster and multi-cluster analyses query = f""" MATCH (src)-[comm:COMMUNICATES_WITH]->(dst) WHERE {where_clause} RETURN src.id AS source_id, src.name AS source_name, src.kind AS source_kind, src.namespace AS source_namespace, src.ip AS source_ip, src.node AS source_node, src.labels AS source_labels, src.annotations AS source_annotations, src.owner_kind AS source_owner_kind, src.owner_name AS source_owner_name, src.pod_uid AS source_pod_uid, src.host_ip AS source_host_ip, src.container AS source_container, src.image AS source_image, src.service_account AS source_service_account, src.phase AS source_phase, dst.id AS destination_id, dst.name AS destination_name, dst.kind AS destination_kind, dst.namespace AS destination_namespace, dst.ip AS destination_ip, dst.node AS destination_node, dst.labels AS destination_labels, dst.annotations AS destination_annotations, dst.owner_kind AS destination_owner_kind, dst.owner_name AS destination_owner_name, dst.pod_uid AS destination_pod_uid, dst.host_ip AS destination_host_ip, dst.container AS destination_container, dst.image AS destination_image, dst.service_account AS destination_service_account, dst.phase AS destination_phase, comm.protocol AS protocol, comm.destination_port AS destination_port, comm.port AS port, comm.request_count AS request_count, comm.bytes_transferred AS bytes_transferred, comm.avg_latency_ms AS avg_latency_ms, comm.risk_level AS risk_level, comm.risk_score AS risk_score, comm.first_seen AS first_seen, comm.last_seen AS last_seen, comm.analysis_id AS analysis_id, comm.error_count AS error_count, comm.retransmit_count AS retransmit_count, comm.last_error_type AS last_error_type ORDER BY comm.last_seen DESC LIMIT $limit """ logger.info(f"[GET_COMMS] Executing query with params: {params}") logger.info(f"[GET_COMMS] WHERE clause: {where_clause}") result = self.execute_query(query, params) data_count = len(result.get("data", [])) if result else 0 logger.info(f"[GET_COMMS] Query result: success={result.get('success')}, data_count={data_count}") return result def get_communication_count( self, analysis_id: Optional[str] = None, cluster_id: Optional[str] = None, namespace: Optional[str] = None ) -> int: """Get total count of communications without limit (for smart edge limit calculation)""" conditions = [] params = {} # Multi-cluster support: analysis_id can be in format '{id}' or '{id}-{cluster_id}' if analysis_id: analysis_id_str = str(analysis_id) analysis_id_prefix = f"{analysis_id_str}-" conditions.append( "(comm.analysis_id = $analysis_id OR comm.analysis_id STARTS WITH $analysis_id_prefix OR " "src.analysis_id = $analysis_id OR src.analysis_id STARTS WITH $analysis_id_prefix)" ) params["analysis_id"] = analysis_id_str params["analysis_id_prefix"] = analysis_id_prefix # Filter by cluster_id if provided if cluster_id: conditions.append("(src.cluster_id = $cluster_id OR comm.cluster_id = $cluster_id)") params["cluster_id"] = str(cluster_id) if namespace: conditions.append( "(src.namespace = $namespace OR dst.namespace = $namespace OR " "(src.namespace = $namespace AND (dst:ExternalEndpoint OR dst.namespace = 'external')))" ) params["namespace"] = namespace # Build WHERE clause where_clause = " AND ".join(conditions) if conditions else "true" # COUNT query - no limit needed # Use same pattern as get_communication_stats (no label constraint) query = f""" MATCH (src)-[comm:COMMUNICATES_WITH]->(dst) WHERE {where_clause} RETURN count(comm) AS total_count """ result = self.execute_query(query, params) if result.get("success") and result.get("data"): return result["data"][0].get("total_count", 0) return 0 def get_cross_namespace_communications( self, analysis_id: Optional[str] = None, cluster_id: Optional[str] = None, limit: int = 50 ) -> Dict[str, Any]: """Get cross-namespace communications (potential security risk)""" conditions = [ "src.namespace <> dst.namespace", "comm.is_active = true", "NOT src.namespace IN ['kube-system', 'kube-public']" ] params = {"limit": limit} # Multi-cluster support: match both single and multi-cluster analysis_id formats if analysis_id: analysis_id_str = str(analysis_id) analysis_id_prefix = f"{analysis_id_str}-" conditions.append("(comm.analysis_id = $analysis_id OR comm.analysis_id STARTS WITH $analysis_id_prefix)") params["analysis_id"] = analysis_id_str params["analysis_id_prefix"] = analysis_id_prefix if cluster_id: conditions.append("src.cluster_id = $cluster_id") params["cluster_id"] = str(cluster_id) where_clause = " AND ".join(conditions) query = f""" MATCH (src:Workload)-[comm:COMMUNICATES_WITH]->(dst) WHERE (dst:Workload OR dst:ExternalEndpoint) AND {where_clause} RETURN src.namespace AS source_namespace, src.name AS source_name, COALESCE(dst.namespace, 'external') AS destination_namespace, dst.name AS destination_name, comm.protocol AS protocol, COALESCE(comm.destination_port, comm.port, 0) AS port, comm.risk_score AS risk_score, comm.analysis_id AS analysis_id ORDER BY comm.risk_score DESC LIMIT $limit """ return self.execute_query(query, params) def get_external_communications( self, namespace: Optional[str] = None, analysis_id: Optional[str] = None, cluster_id: Optional[str] = None, limit: int = 50 ) -> Dict[str, Any]: """Get external communications""" conditions = ["comm.is_active = true"] params = {"limit": limit} if namespace: conditions.append("src.namespace = $namespace") params["namespace"] = namespace # Multi-cluster support: match both single and multi-cluster analysis_id formats if analysis_id: analysis_id_str = str(analysis_id) analysis_id_prefix = f"{analysis_id_str}-" conditions.append("(comm.analysis_id = $analysis_id OR comm.analysis_id STARTS WITH $analysis_id_prefix)") params["analysis_id"] = analysis_id_str params["analysis_id_prefix"] = analysis_id_prefix if cluster_id: conditions.append("src.cluster_id = $cluster_id") params["cluster_id"] = str(cluster_id) where_clause = " AND ".join(conditions) query = f""" MATCH (src:Workload)-[comm:COMMUNICATES_WITH]->(ext:ExternalEndpoint) WHERE {where_clause} RETURN src.name AS source_name, src.namespace AS source_namespace, ext.ip_address AS external_ip, ext.hostname AS external_hostname, comm.destination_port AS port, comm.protocol AS protocol, comm.request_count AS request_count, comm.analysis_id AS analysis_id ORDER BY comm.last_seen DESC LIMIT $limit """ return self.execute_query(query, params) def get_high_risk_communications( self, analysis_id: Optional[str] = None, cluster_id: Optional[str] = None, limit: int = 50 ) -> Dict[str, Any]: """Get high-risk communications""" conditions = [ "comm.risk_level IN ['high', 'critical']", "comm.is_active = true" ] params = {"limit": limit} # Multi-cluster support: match both single and multi-cluster analysis_id formats if analysis_id: analysis_id_str = str(analysis_id) analysis_id_prefix = f"{analysis_id_str}-" conditions.append("(comm.analysis_id = $analysis_id OR comm.analysis_id STARTS WITH $analysis_id_prefix)") params["analysis_id"] = analysis_id_str params["analysis_id_prefix"] = analysis_id_prefix if cluster_id: conditions.append("src.cluster_id = $cluster_id") params["cluster_id"] = str(cluster_id) where_clause = " AND ".join(conditions) query = f""" MATCH (src:Workload)-[comm:COMMUNICATES_WITH]->(dst) WHERE {where_clause} RETURN src.name AS source_name, src.namespace AS source_namespace, dst.name AS destination_name, comm.protocol AS protocol, comm.destination_port AS port, comm.risk_level AS risk_level, comm.risk_score AS risk_score, comm.risk_factors AS risk_factors, comm.analysis_id AS analysis_id ORDER BY comm.risk_score DESC LIMIT $limit """ return self.execute_query(query, params) def get_workload_by_id(self, workload_id: str) -> Dict[str, Any]: """Get workload details by ID""" query = """ MATCH (w:Workload {id: $workload_id}) RETURN w """ return self.execute_query(query, {"workload_id": workload_id}) def get_workloads_by_namespace( self, namespace: str, kind: Optional[str] = None, analysis_id: Optional[str] = None, cluster_id: Optional[str] = None ) -> Dict[str, Any]: """Get all workloads in a namespace""" conditions = [ "w.namespace = $namespace", "w.is_active = true" ] params = {"namespace": namespace} if kind: conditions.append("w.kind = $kind") params["kind"] = kind # Multi-cluster support: match both single and multi-cluster analysis_id formats if analysis_id: analysis_id_str = str(analysis_id) analysis_id_prefix = f"{analysis_id_str}-" conditions.append("(w.analysis_id = $analysis_id OR w.analysis_id STARTS WITH $analysis_id_prefix)") params["analysis_id"] = analysis_id_str params["analysis_id_prefix"] = analysis_id_prefix if cluster_id: conditions.append("w.cluster_id = $cluster_id") params["cluster_id"] = str(cluster_id) where_clause = " AND ".join(conditions) query = f""" MATCH (w:Workload) WHERE {where_clause} RETURN w ORDER BY w.name """ return self.execute_query(query, params) def get_workloads( self, analysis_id: Optional[str] = None, cluster_id: Optional[str] = None, namespace: Optional[str] = None, kind: Optional[str] = None, limit: int = 1000 ) -> Dict[str, Any]: """Get workloads with optional filters""" conditions = ["w.is_active = true"] params = {"limit": limit} # Multi-cluster support: match both single and multi-cluster analysis_id formats if analysis_id: analysis_id_str = str(analysis_id) analysis_id_prefix = f"{analysis_id_str}-" conditions.append("(w.analysis_id = $analysis_id OR w.analysis_id STARTS WITH $analysis_id_prefix)") params["analysis_id"] = analysis_id_str params["analysis_id_prefix"] = analysis_id_prefix if cluster_id: conditions.append("w.cluster_id = $cluster_id") params["cluster_id"] = str(cluster_id) if namespace: conditions.append("w.namespace = $namespace") params["namespace"] = namespace if kind: conditions.append("w.kind = $kind") params["kind"] = kind where_clause = " AND ".join(conditions) query = f""" MATCH (w:Workload) WHERE {where_clause} RETURN w.id AS id, w.name AS name, w.namespace AS namespace, w.kind AS kind, w.cluster_id AS cluster_id, w.analysis_id AS analysis_id, w.ip AS ip, w.status AS status, w.labels AS labels, w.annotations AS annotations, w.created_at AS created_at ORDER BY w.namespace, w.name LIMIT $limit """ result = self.execute_query(query, params) if result.get("success") and result.get("data"): for record in result["data"]: for field in ("labels", "annotations"): raw = record.get(field) if isinstance(raw, str): try: record[field] = json.loads(raw) except (json.JSONDecodeError, TypeError): record[field] = {} elif not raw: record[field] = {} return result def get_dependency_graph( self, cluster_id: Optional[str] = None, analysis_id: Optional[str] = None, namespace: Optional[str] = None, depth: int = 2, search: Optional[str] = None ) -> Dict[str, Any]: """ Get dependency graph with nodes and edges for visualization Args: search: Optional search term (min 3 chars) to filter nodes by name, namespace, id, ip, host_ip, or edge port. When provided, limit is increased to ensure all matching results are returned. Returns: Dict with 'nodes' and 'edges' lists """ conditions = [] params = {} if cluster_id: params["cluster_id"] = str(cluster_id) # Multi-cluster support: set up both analysis_id and prefix for pattern matching if analysis_id: analysis_id_str = str(analysis_id) analysis_id_prefix = f"{analysis_id_str}-" params["analysis_id"] = analysis_id_str params["analysis_id_prefix"] = analysis_id_prefix if namespace: params["namespace"] = namespace # Server-side search: filter by node name, namespace, id, ip, or edge port # Only active for 3+ character searches to avoid overly broad matches search_condition = "" if search and len(search) >= 3: params["search"] = search.lower() search_condition = """ AND ( toLower(src.name) CONTAINS $search OR toLower(src.namespace) CONTAINS $search OR toLower(src.id) CONTAINS $search OR toLower(coalesce(src.ip, '')) CONTAINS $search OR toLower(coalesce(src.host_ip, '')) CONTAINS $search OR toLower(dst.name) CONTAINS $search OR toLower(dst.namespace) CONTAINS $search OR toLower(dst.id) CONTAINS $search OR toLower(coalesce(dst.ip, '')) CONTAINS $search OR toLower(coalesce(dst.host_ip, '')) CONTAINS $search OR toString(coalesce(r.port, 0)) CONTAINS $search ) """ # Build WHERE clause for edges edge_conditions = [] if cluster_id: edge_conditions.append("(r.cluster_id = $cluster_id)") if analysis_id: # Multi-cluster support: match both single and multi-cluster analysis_id formats edge_conditions.append("(r.analysis_id = $analysis_id OR r.analysis_id STARTS WITH $analysis_id_prefix)") if namespace: # CRITICAL FIX: At least ONE endpoint must be in the selected namespace # This prevents external-to-external edges that have no connection to the filtered namespace # # Edge is included if: # - At least one endpoint is in the selected namespace (src OR dst) # - AND both endpoints are either in namespace OR external (prevents cross-namespace leaks) # # Examples (namespace = 'flowfish'): # - flowfish → flowfish: OK (both in namespace) # - flowfish → external: OK (one in namespace) # - external → flowfish: OK (one in namespace) # - external → external: BLOCKED (neither in namespace - causes floating edges!) edge_conditions.append( "(src.namespace = $namespace OR dst.namespace = $namespace)" ) edge_where = " AND ".join(edge_conditions) if edge_conditions else "true" # When search is active, increase limit to get all matching results # Normal: 5000 (performance), Search: 50000 (find all matches) effective_limit = 50000 if search and len(search) >= 3 else settings.max_results # ============================================================================ # EDGE-FIRST APPROACH: Derive nodes FROM edges # ============================================================================ # This ensures: # 1. Only pods with active communication in this analysis are shown # 2. No floating edges (every edge endpoint has a node by definition) # 3. No analysis_id filtering issues for nodes (pods from old analyses visible) # # Flow: # 1. Get all edges (filtered by analysis_id + namespace + search) # 2. Collect node IDs from edge endpoints # 3. Fetch node details for those IDs only # ============================================================================ # Step 1: Get all edges first # Get COMMUNICATES_WITH edges comm_edges_query = f""" MATCH (src)-[r:COMMUNICATES_WITH]->(dst) WHERE {edge_where} {search_condition} RETURN DISTINCT src.id AS source_id, dst.id AS target_id, 'COMMUNICATES_WITH' AS edge_type, COALESCE(r.protocol, 'TCP') AS protocol, COALESCE(r.app_protocol, r.protocol, 'TCP') AS app_protocol, COALESCE(r.port, r.destination_port, 0) AS port, COALESCE(r.request_count, 1) AS request_count, COALESCE(r.error_count, 0) AS error_count, COALESCE(r.retransmit_count, 0) AS retransmit_count, r.last_error_type AS last_error_type ORDER BY request_count DESC, source_id, target_id LIMIT {effective_limit} """ comm_result = self.execute_query(comm_edges_query, params) edges = comm_result.get("data", []) if comm_result.get("success") else [] logger.info(f"[EDGE_FETCH] COMMUNICATES_WITH: {len(edges)} edges") # Get DNS query edges (DNS targets are always external) # All DNS edges go to external endpoints, so fetch all with reasonable limit dns_edges_query = f""" MATCH (src)-[r:QUERIES_DNS]->(dst) WHERE {edge_where} {search_condition} RETURN DISTINCT src.id AS source_id, dst.id AS target_id, 'QUERIES_DNS' AS edge_type, 'DNS' AS protocol, 'DNS' AS app_protocol, 53 AS port, COALESCE(r.request_count, 1) AS request_count, 0 AS error_count, 0 AS retransmit_count, null AS last_error_type ORDER BY request_count DESC, source_id, target_id LIMIT {effective_limit} """ dns_result = self.execute_query(dns_edges_query, params) dns_edges = dns_result.get("data", []) if dns_result.get("success") else [] logger.info(f"[EDGE_FETCH] QUERIES_DNS: {len(dns_edges)} edges") # Get TLS connection edges tls_edges_query = f""" MATCH (src)-[r:TLS_CONNECTS]->(dst) WHERE {edge_where} {search_condition} RETURN DISTINCT src.id AS source_id, dst.id AS target_id, 'TLS_CONNECTS' AS edge_type, 'TLS' AS protocol, 'TLS' AS app_protocol, COALESCE(r.port, r.destination_port, 443) AS port, COALESCE(r.request_count, 1) AS request_count, 0 AS error_count, 0 AS retransmit_count, null AS last_error_type ORDER BY request_count DESC, source_id, target_id LIMIT {effective_limit} """ tls_result = self.execute_query(tls_edges_query, params) tls_edges = tls_result.get("data", []) if tls_result.get("success") else [] logger.info(f"[EDGE_FETCH] TLS_CONNECTS: {len(tls_edges)} edges") # Get LISTENS_ON edges (service endpoints) listen_edges_query = f""" MATCH (src)-[r:LISTENS_ON]->(dst) WHERE {edge_where} {search_condition} RETURN DISTINCT src.id AS source_id, dst.id AS target_id, 'LISTENS_ON' AS edge_type, COALESCE(r.protocol, 'TCP') AS protocol, COALESCE(r.app_protocol, r.protocol, 'TCP') AS app_protocol, COALESCE(r.port, r.bind_port, 0) AS port, COALESCE(r.request_count, 1) AS request_count, 0 AS error_count, 0 AS retransmit_count, null AS last_error_type ORDER BY request_count DESC, source_id, target_id LIMIT {effective_limit} """ listen_result = self.execute_query(listen_edges_query, params) listen_edges = listen_result.get("data", []) if listen_result.get("success") else [] logger.info(f"[EDGE_FETCH] LISTENS_ON: {len(listen_edges)} edges") # Combine all edges all_edges = edges + dns_edges + tls_edges + listen_edges # Step 2: Collect all node IDs from edges edge_source_ids = {e.get("source_id") for e in all_edges if e.get("source_id")} edge_target_ids = {e.get("target_id") for e in all_edges if e.get("target_id")} all_edge_node_ids = edge_source_ids | edge_target_ids logger.warning(f"[GRAPH_QUERY_DEBUG] Total edges: {len(all_edges)}, unique node IDs from edges: {len(all_edge_node_ids)}") logger.warning(f"[GRAPH_QUERY_DEBUG] Namespace filter: {namespace}, analysis_id: {analysis_id}, search: {search}, limit: {effective_limit}") # Step 3: Fetch node details for edge endpoints only # No analysis_id filter for nodes - we already filtered edges by analysis_id nodes = [] if all_edge_node_ids: node_ids_list = list(all_edge_node_ids) nodes_query = """ MATCH (w) WHERE w.id IN $node_ids RETURN DISTINCT w.id AS id, COALESCE(w.name, 'unknown') AS name, CASE WHEN w.owner_kind = 'Service' THEN 'Service' ELSE COALESCE(labels(w)[0], 'Workload') END AS kind, COALESCE(w.namespace, 'external') AS namespace, COALESCE(w.cluster_id, '1') AS cluster_id, COALESCE(w.status, 'unknown') AS status, w.labels AS labels, w.annotations AS annotations, COALESCE(w.is_external, false) AS is_external, w.ip AS ip, w.host_ip AS host_ip, w.owner_kind AS owner_kind, w.owner_name AS owner_name, w.node AS node, w.network_type AS network_type, w.resolution_source AS resolution_source, w.pod_uid AS pod_uid, w.container AS container, w.image AS image, w.service_account AS service_account, w.phase AS phase """ nodes_result = self.execute_query(nodes_query, {"node_ids": node_ids_list}) nodes = nodes_result.get("data", []) if nodes_result.get("success") else [] logger.warning(f"[GRAPH_QUERY_DEBUG] Fetched {len(nodes)} nodes for {len(node_ids_list)} edge endpoints") # Step 4: Create synthetic nodes for any missing endpoints (edge endpoints not in Neo4j) existing_node_ids = {n.get("id") for n in nodes if n.get("id")} missing_node_ids = all_edge_node_ids - existing_node_ids if missing_node_ids: logger.warning(f"[GRAPH_QUERY_DEBUG] Creating {len(missing_node_ids)} synthetic nodes for missing endpoints") # Create synthetic nodes for missing endpoints # Parse node ID format: cluster_id:namespace:name for node_id in missing_node_ids: parts = node_id.split(":", 2) # Split into max 3 parts if len(parts) >= 3: node_cluster, node_ns, node_name = parts[0], parts[1], parts[2] elif len(parts) == 2: node_cluster, node_ns, node_name = "1", parts[0], parts[1] else: node_cluster, node_ns, node_name = "1", "external", node_id # Determine network_type based on namespace # Must match frontend NETWORK_TYPE_INFO keys exactly network_type = None if node_ns == "external": network_type = "External-IP" elif node_ns == "cluster-network": # Could be Pod-Network or Service-Network, default to Service network_type = "Service-Network" elif node_ns == "internal-network": network_type = "Internal-Network" elif node_ns == "sdn-infrastructure": network_type = "SDN-Gateway" synthetic_node = { "id": node_id, "name": node_name, "kind": "Workload", "namespace": node_ns, "cluster_id": node_cluster, "status": "unknown", "labels": {}, "is_external": node_ns == "external", "ip": node_name if self._is_ip_address(node_name) else None, "host_ip": None, "owner_kind": None, "owner_name": None, "node": None, "network_type": network_type, "resolution_source": "synthetic" } nodes.append(synthetic_node) logger.warning(f"[GRAPH_QUERY_DEBUG] Created {len(missing_node_ids)} synthetic nodes") # Final verification - all edges should have valid endpoints now final_node_ids = {n.get("id") for n in nodes if n.get("id")} edges_before = len(all_edges) all_edges = [ e for e in all_edges if e.get("source_id") in final_node_ids and e.get("target_id") in final_node_ids ] edges_after = len(all_edges) if edges_before != edges_after: # This should not happen with edge-first approach, but log if it does logger.error(f"[GRAPH_QUERY_ERROR] Filtered {edges_before - edges_after} edges - this should not happen!") logger.warning(f"[GRAPH_QUERY_DEBUG] Final result: {len(nodes)} nodes, {len(all_edges)} edges") # Post-process nodes: parse JSON string fields # Neo4j stores labels/annotations as JSON string, but frontend expects object for node in nodes: # Parse labels from JSON string to dict labels_raw = node.get("labels") if labels_raw: if isinstance(labels_raw, str): try: node["labels"] = json.loads(labels_raw) except (json.JSONDecodeError, TypeError): node["labels"] = {} elif not isinstance(labels_raw, dict): node["labels"] = {} else: node["labels"] = {} # Parse annotations from JSON string to dict annotations_raw = node.get("annotations") if annotations_raw: if isinstance(annotations_raw, str): try: node["annotations"] = json.loads(annotations_raw) except (json.JSONDecodeError, TypeError): node["annotations"] = {} elif not isinstance(annotations_raw, dict): node["annotations"] = {} else: node["annotations"] = {} return { "nodes": nodes, "edges": all_edges, "total_nodes": len(nodes), "total_edges": len(all_edges) } def get_communication_stats( self, cluster_id: Optional[str] = None, analysis_id: Optional[str] = None ) -> Dict[str, Any]: """ Get communication statistics including both network and DNS communications Multi-cluster support: analysis_id can be in format '{id}' or '{id}-{cluster_id}' """ conditions = [] params = {} if cluster_id: conditions.append("r.cluster_id = $cluster_id") params["cluster_id"] = str(cluster_id) # Multi-cluster support: match both single and multi-cluster analysis_id formats if analysis_id: analysis_id_str = str(analysis_id) analysis_id_prefix = f"{analysis_id_str}-" conditions.append("(r.analysis_id = $analysis_id OR r.analysis_id STARTS WITH $analysis_id_prefix)") params["analysis_id"] = analysis_id_str params["analysis_id_prefix"] = analysis_id_prefix where_clause = " AND ".join(conditions) if conditions else "true" # Count COMMUNICATES_WITH edges comm_query = f""" MATCH (src)-[r:COMMUNICATES_WITH]->(dst) WHERE {where_clause} RETURN count(r) AS comm_count, sum(COALESCE(r.request_count, 1)) AS comm_requests, sum(COALESCE(r.bytes_transferred, 0)) AS bytes_transferred, sum(COALESCE(r.error_count, 0)) AS total_errors, sum(COALESCE(r.retransmit_count, 0)) AS total_retransmits, count(DISTINCT src.namespace) AS src_namespaces, count(DISTINCT dst.namespace) AS dst_namespaces """ comm_result = self.execute_query(comm_query, params) # Count QUERIES_DNS edges dns_query = f""" MATCH (src)-[r:QUERIES_DNS]->(dst) WHERE {where_clause} RETURN count(r) AS dns_count, sum(COALESCE(r.request_count, 1)) AS dns_requests """ dns_result = self.execute_query(dns_query, params) # Combine results comm_data = comm_result.get("data", [{}])[0] if comm_result.get("success") else {} dns_data = dns_result.get("data", [{}])[0] if dns_result.get("success") else {} total_communications = (comm_data.get("comm_count", 0) or 0) + (dns_data.get("dns_count", 0) or 0) total_requests = (comm_data.get("comm_requests", 0) or 0) + (dns_data.get("dns_requests", 0) or 0) return { "total_communications": total_communications, "total_request_count": total_requests, "total_bytes_transferred": comm_data.get("bytes_transferred", 0) or 0, "total_errors": comm_data.get("total_errors", 0) or 0, "total_retransmits": comm_data.get("total_retransmits", 0) or 0, "unique_namespaces": (comm_data.get("src_namespaces", 0) or 0) + (comm_data.get("dst_namespaces", 0) or 0), "network_communications": comm_data.get("comm_count", 0) or 0, "dns_queries": dns_data.get("dns_count", 0) or 0, "protocol_distribution": { "TCP": comm_data.get("comm_count", 0) or 0, "DNS": dns_data.get("dns_count", 0) or 0 }, "risk_distribution": {}, "cluster_id": cluster_id, "analysis_id": analysis_id } def health_check(self) -> Dict[str, Any]: """Check Neo4j connection health""" try: if not self.driver: return {"healthy": False, "error": "No driver"} self.driver.verify_connectivity() # Execute simple query result = self.execute_query("RETURN 1 AS test") if result.get("success"): return { "healthy": True, "database": self.database, "uri": settings.neo4j_bolt_uri } else: return { "healthy": False, "error": result.get("error") } except Exception as e: return { "healthy": False, "error": str(e) } def find_pod_dependencies( self, analysis_id: Optional[str] = None, analysis_ids: Optional[List[str]] = None, cluster_id: Optional[str] = None, pod_name: Optional[str] = None, namespace: Optional[str] = None, owner_name: Optional[str] = None, label_key: Optional[str] = None, label_value: Optional[str] = None, annotation_key: Optional[str] = None, annotation_value: Optional[str] = None, ip: Optional[str] = None, depth: int = 1, include_communication_details: bool = True ) -> Dict[str, Any]: """ Find a pod by any metadata and return its upstream/downstream dependencies. The matched pod is the "upstream" (source). All pods it communicates with are "downstream" (targets). Pods that communicate TO the matched pod are also returned as callers (reverse upstream). Any combination of search parameters can be used. At least one is required. Args: analysis_id: Single analysis ID for scope (backward compat) analysis_ids: Multiple analysis IDs for scope (takes precedence over analysis_id) cluster_id: Cluster ID for scope pod_name: Pod/workload name to search namespace: Namespace to narrow search owner_name: Deployment/StatefulSet/DaemonSet name to search label_key/label_value: Label key=value to match annotation_key/annotation_value: Annotation key=value to match ip: Pod IP to search depth: Traversal depth for dependencies (default 1) Returns: Dict with upstream pod info and downstream dependencies """ # Build match conditions to find the target pod match_conditions = [] params = {} # Consolidate analysis_ids (plural takes precedence) effective_ids = None if analysis_ids: effective_ids = [str(a) for a in analysis_ids] elif analysis_id: effective_ids = [str(analysis_id)] if cluster_id: params["cluster_id"] = str(cluster_id) if pod_name: match_conditions.append("toLower(w.name) CONTAINS toLower($pod_name)") params["pod_name"] = pod_name if namespace: match_conditions.append("w.namespace = $namespace") params["namespace"] = namespace if owner_name: match_conditions.append("toLower(w.owner_name) CONTAINS toLower($owner_name)") params["owner_name"] = owner_name if ip: match_conditions.append("w.ip = $ip") params["ip"] = ip if annotation_key: if '*' in annotation_key or '?' in annotation_key: ann_key_prefix = annotation_key.split('*')[0].split('?')[0] if ann_key_prefix: match_conditions.append( "w.annotations CONTAINS $annotation_key_search" ) params["annotation_key_search"] = f'"{ann_key_prefix}' else: match_conditions.append( "w.annotations CONTAINS $annotation_key_search" ) params["annotation_key_search"] = f'"{annotation_key}"' if label_key and label_value: match_conditions.append( "w.labels CONTAINS $label_search" ) params["label_search"] = f'"{label_key}"' params["label_value"] = label_value elif label_key: match_conditions.append( "w.labels CONTAINS $label_key_search" ) params["label_key_search"] = f'"{label_key}"' if not match_conditions: return {"success": False, "error": "At least one search parameter required (pod_name, namespace, owner_name, ip, annotation_key, label_key)", "count": 0, "results": []} # Add analysis scope filter if effective_ids: params["analysis_ids"] = effective_ids params["analysis_id_prefixes"] = [f"{a}-" for a in effective_ids] match_conditions.append( "(w.analysis_id IN $analysis_ids OR " "ANY(prefix IN $analysis_id_prefixes WHERE w.analysis_id STARTS WITH prefix))" ) if cluster_id: match_conditions.append("w.cluster_id = $cluster_id") where_clause = " AND ".join(match_conditions) # Step 1: Find the upstream pod(s) matching criteria find_query = f""" MATCH (w:Workload) WHERE {where_clause} RETURN w.id AS id, w.name AS name, w.namespace AS namespace, w.cluster_id AS cluster_id, w.ip AS ip, w.labels AS labels, w.annotations AS annotations, w.owner_kind AS owner_kind, w.owner_name AS owner_name, w.phase AS phase, w.image AS image, w.container AS container, w.service_account AS service_account, w.host_ip AS host_ip, w.pod_uid AS pod_uid, w.node AS node LIMIT 200 """ find_result = self.execute_query(find_query, params) if not find_result.get("success") or not find_result.get("data"): return { "success": False, "error": "No pod found matching the given criteria", "search_params": { k: v for k, v in { "pod_name": pod_name, "namespace": namespace, "owner_name": owner_name, "annotation_key": annotation_key, "annotation_value": annotation_value, "label_key": label_key, "label_value": label_value, "ip": ip }.items() if v } } # Post-filter for annotation_key/value (supports * glob pattern) matched_pods = find_result["data"] if annotation_key: from fnmatch import fnmatch ann_key_has_glob = '*' in annotation_key or '?' in annotation_key ann_val_any = not annotation_value or annotation_value == '*' ann_val_has_glob = not ann_val_any and ('*' in annotation_value or '?' in annotation_value) filtered = [] for pod in matched_pods: ann_raw = pod.get("annotations", "{}") if isinstance(ann_raw, str): try: ann = json.loads(ann_raw) except (json.JSONDecodeError, TypeError): ann = {} else: ann = ann_raw or {} hit_keys = [k for k in ann if fnmatch(k, annotation_key)] if ann_key_has_glob else ([annotation_key] if annotation_key in ann else []) if not hit_keys: continue if ann_val_any: filtered.append(pod) continue for k in hit_keys: v = str(ann[k]) if ann_val_has_glob: if fnmatch(v, annotation_value): filtered.append(pod) break else: if v == annotation_value: filtered.append(pod) break matched_pods = filtered if label_key and label_value: filtered = [] for pod in matched_pods: lbl_raw = pod.get("labels", "{}") if isinstance(lbl_raw, str): try: lbl = json.loads(lbl_raw) except (json.JSONDecodeError, TypeError): lbl = {} else: lbl = lbl_raw or {} if lbl.get(label_key) == label_value: filtered.append(pod) matched_pods = filtered if not matched_pods: return { "success": False, "error": "No pod found matching the given criteria after filtering", "search_params": { k: v for k, v in { "pod_name": pod_name, "namespace": namespace, "owner_name": owner_name, "annotation_key": annotation_key, "annotation_value": annotation_value, "label_key": label_key, "label_value": label_value, "ip": ip }.items() if v } } results = [] for upstream_pod in matched_pods: pod_id = upstream_pod["id"] # Parse JSON fields for field in ["labels", "annotations"]: raw = upstream_pod.get(field, "{}") if isinstance(raw, str): try: upstream_pod[field] = json.loads(raw) except (json.JSONDecodeError, TypeError): upstream_pod[field] = {} elif not isinstance(raw, dict): upstream_pod[field] = {} # Step 2: Get downstream (pods this upstream connects TO) depth_val = max(1, min(depth, 5)) downstream_query = f""" MATCH path = (src:Workload {{id: $pod_id}})-[:COMMUNICATES_WITH*1..{depth_val}]->(dst) WHERE dst.id <> $pod_id WITH dst, path, length(path) as hops ORDER BY hops ASC WITH dst, collect(path)[0] as sp WITH dst, length(sp) as hop_count, relationships(sp) as rels WITH dst, hop_count, rels[size(rels)-1] as r RETURN dst.id AS id, dst.name AS name, dst.namespace AS namespace, dst.cluster_id AS cluster_id, dst.ip AS ip, dst.labels AS labels, dst.annotations AS annotations, dst.owner_kind AS owner_kind, dst.owner_name AS owner_name, dst.phase AS phase, dst.image AS image, dst.container AS container, dst.service_account AS service_account, dst.host_ip AS host_ip, dst.pod_uid AS pod_uid, dst.node AS node, hop_count, r.protocol AS protocol, r.port AS port, r.destination_port AS destination_port, r.app_protocol AS app_protocol, r.request_count AS request_count, r.bytes_transferred AS bytes_transferred, r.error_count AS error_count, r.retransmit_count AS retransmit_count, r.avg_latency_ms AS avg_latency_ms, r.last_seen AS last_seen ORDER BY hop_count, dst.name LIMIT 200 """ downstream_result = self.execute_query(downstream_query, {"pod_id": pod_id}) downstream_pods = [] if downstream_result.get("success"): for d in downstream_result.get("data", []): for field in ["labels", "annotations"]: raw = d.get(field, "{}") if isinstance(raw, str): try: d[field] = json.loads(raw) except (json.JSONDecodeError, TypeError): d[field] = {} elif not isinstance(raw, dict): d[field] = {} port = d.get("destination_port") or d.get("port") request_count = d.get("request_count") or 0 error_count = d.get("error_count") or 0 retransmit_count = d.get("retransmit_count") or 0 avg_latency = d.get("avg_latency_ms") or 0 dep_entry = { "pod_name": d.get("name"), "namespace": d.get("namespace"), "cluster_id": d.get("cluster_id"), "ip": d.get("ip"), "labels": d.get("labels", {}), "annotations": d.get("annotations", {}), "owner_kind": d.get("owner_kind"), "owner_name": d.get("owner_name"), "phase": d.get("phase"), "image": d.get("image"), "container": d.get("container"), "service_account": d.get("service_account"), "host_ip": d.get("host_ip"), "node": d.get("node"), "hop_count": d.get("hop_count", 1), } if include_communication_details: dep_entry["communication"] = self._build_communication_contract( d.get("protocol"), d.get("app_protocol"), port, request_count, d.get("bytes_transferred"), error_count, retransmit_count, avg_latency, d.get("last_seen"), workload_name=d.get("name", "") ) dep_entry["health"] = self._calculate_dependency_health( request_count, error_count, retransmit_count, avg_latency ) downstream_pods.append(dep_entry) # Step 3: Get callers (pods that connect TO this upstream pod - reverse direction) callers_query = f""" MATCH path = (caller)-[:COMMUNICATES_WITH*1..{depth_val}]->(target:Workload {{id: $pod_id}}) WHERE caller.id <> $pod_id WITH caller, path, length(path) as hops ORDER BY hops ASC WITH caller, collect(path)[0] as sp WITH caller, length(sp) as hop_count, relationships(sp) as rels WITH caller, hop_count, rels[size(rels)-1] as r RETURN caller.id AS id, caller.name AS name, caller.namespace AS namespace, caller.cluster_id AS cluster_id, caller.ip AS ip, caller.labels AS labels, caller.annotations AS annotations, caller.owner_kind AS owner_kind, caller.owner_name AS owner_name, caller.phase AS phase, caller.image AS image, caller.container AS container, caller.service_account AS service_account, caller.host_ip AS host_ip, caller.pod_uid AS pod_uid, caller.node AS node, hop_count, r.protocol AS protocol, r.port AS port, r.destination_port AS destination_port, r.app_protocol AS app_protocol, r.request_count AS request_count, r.bytes_transferred AS bytes_transferred, r.error_count AS error_count, r.retransmit_count AS retransmit_count, r.avg_latency_ms AS avg_latency_ms, r.last_seen AS last_seen ORDER BY hop_count, caller.name LIMIT 200 """ callers_result = self.execute_query(callers_query, {"pod_id": pod_id}) caller_pods = [] if callers_result.get("success"): for c in callers_result.get("data", []): for field in ["labels", "annotations"]: raw = c.get(field, "{}") if isinstance(raw, str): try: c[field] = json.loads(raw) except (json.JSONDecodeError, TypeError): c[field] = {} elif not isinstance(raw, dict): c[field] = {} port = c.get("destination_port") or c.get("port") request_count = c.get("request_count") or 0 error_count = c.get("error_count") or 0 retransmit_count = c.get("retransmit_count") or 0 avg_latency = c.get("avg_latency_ms") or 0 caller_entry = { "pod_name": c.get("name"), "namespace": c.get("namespace"), "cluster_id": c.get("cluster_id"), "ip": c.get("ip"), "labels": c.get("labels", {}), "annotations": c.get("annotations", {}), "owner_kind": c.get("owner_kind"), "owner_name": c.get("owner_name"), "phase": c.get("phase"), "image": c.get("image"), "container": c.get("container"), "service_account": c.get("service_account"), "host_ip": c.get("host_ip"), "node": c.get("node"), "hop_count": c.get("hop_count", 1), } if include_communication_details: caller_entry["communication"] = self._build_communication_contract( c.get("protocol"), c.get("app_protocol"), port, request_count, c.get("bytes_transferred"), error_count, retransmit_count, avg_latency, c.get("last_seen"), workload_name=c.get("name", "") ) caller_entry["health"] = self._calculate_dependency_health( request_count, error_count, retransmit_count, avg_latency ) caller_pods.append(caller_entry) results.append({ "upstream": { "pod_name": upstream_pod.get("name"), "namespace": upstream_pod.get("namespace"), "cluster_id": upstream_pod.get("cluster_id"), "ip": upstream_pod.get("ip"), "labels": upstream_pod.get("labels", {}), "annotations": upstream_pod.get("annotations", {}), "owner_kind": upstream_pod.get("owner_kind"), "owner_name": upstream_pod.get("owner_name"), "phase": upstream_pod.get("phase"), "image": upstream_pod.get("image"), "container": upstream_pod.get("container"), "service_account": upstream_pod.get("service_account"), "host_ip": upstream_pod.get("host_ip"), "node": upstream_pod.get("node") }, "downstream": downstream_pods, "callers": caller_pods }) return { "success": True, "count": len(results), "results": results } def batch_find_dependencies( self, analysis_id: Optional[str] = None, cluster_id: Optional[str] = None, services: List[Dict[str, Any]] = None, depth: int = 1, include_communication_details: bool = True, ) -> Dict[str, Any]: """Batch find dependencies for multiple services in one call.""" if not services: return {"error": "services list is required"} all_results = [] all_downstream_ids: List[set] = [] for svc in services: result = self.find_pod_dependencies( analysis_id=analysis_id, cluster_id=cluster_id, pod_name=svc.get("pod_name"), namespace=svc.get("namespace"), owner_name=svc.get("owner_name"), label_key=svc.get("label_key"), label_value=svc.get("label_value"), annotation_key=svc.get("annotation_key"), annotation_value=svc.get("annotation_value"), ip=svc.get("ip"), depth=depth, include_communication_details=include_communication_details, ) all_results.append(result) ids = set() if result.get("success"): for r in result.get("results", []): for d in r.get("downstream", []): name = d.get("pod_name") or d.get("owner_name") or "" ns = d.get("namespace", "") ids.add(f"{ns}/{name}") all_downstream_ids.append(ids) shared = set() if len(all_downstream_ids) >= 2: shared = all_downstream_ids[0] for s in all_downstream_ids[1:]: shared = shared & s return { "success": True, "service_count": len(services), "results": all_results, "shared_dependencies": sorted(shared), } def format_dependency_summary( self, stream_result: Dict[str, Any], analysis_ids: List[str], ) -> Dict[str, Any]: """Transform find_pod_dependencies output into a compact, AI-agent-friendly grouped format. Dependencies are grouped by service_category with only the fields relevant for cross-project impact analysis. When multiple pods match (e.g. namespace-wide query), aggregates ALL downstream/caller entries (deduplicated) and exposes matched_services. Replica pods belonging to the same Deployment/StatefulSet are collapsed into a single logical workload entry. """ try: int_ids = [int(a) for a in analysis_ids] except (ValueError, TypeError): int_ids = analysis_ids if not stream_result.get("success") or not stream_result.get("results"): return { "success": False, "analysis_ids": int_ids, "error": stream_result.get("error", "No results"), } results = stream_result["results"] def _safe_labels(entry: dict) -> dict: lbl = entry.get("labels") or {} if isinstance(lbl, str): try: lbl = json.loads(lbl) except (json.JSONDecodeError, TypeError): lbl = {} return lbl def _strip_template_hash(name: str, pth: str) -> str: """Strip pod-template-hash from a name, handling both ReplicaSet names (name-HASH) and pod names (name-HASH-RANDOM).""" if not pth: return name if name.endswith(f"-{pth}"): return name[:-(len(pth) + 1)] marker = f"-{pth}-" idx = name.find(marker) if idx > 0: return name[:idx] return name def _workload_name(entry: dict) -> str: """Resolve the logical workload name from the richest source available.""" labels = _safe_labels(entry) name = labels.get("app.kubernetes.io/name") or labels.get("app") if name: return name pth = labels.get("pod-template-hash", "") owner = entry.get("owner_name") or "" if owner: return _strip_template_hash(owner, pth) return _strip_template_hash(entry.get("pod_name", ""), pth) def _workload_key(entry: dict) -> str: """Determine a stable identity key that collapses replica pods into their owning Deployment/StatefulSet. Resolution order: 1. namespace + app.kubernetes.io/name or app label (most reliable) 2. namespace + owner_name (stripped of template hash if present) 3. namespace + pod name (stripped of template hash if present) """ ns = entry.get("namespace", "") labels = _safe_labels(entry) name = labels.get("app.kubernetes.io/name") or labels.get("app") if name: return f"{ns}/{name}" pth = labels.get("pod-template-hash", "") owner = entry.get("owner_name") or "" if owner: return f"{ns}/{_strip_template_hash(owner, pth)}" pod = entry.get("pod_name") or "" return f"{ns}/{_strip_template_hash(pod, pth)}" def _is_noise_entry(entry: dict) -> bool: """Filter out noise: reverse DNS, bare IPs with no metadata, 0.0.0.0.""" name = entry.get("pod_name") or entry.get("owner_name") or "" if name.endswith(".in-addr.arpa.") or name.endswith(".in-addr.arpa"): return True if name in ("0.0.0.0", "0.0.0.0:0"): return True comm = entry.get("communication") or {} port = comm.get("port") or 0 ns = entry.get("namespace", "") ann = entry.get("annotations") or {} lbl = entry.get("labels") or {} owner_kind = entry.get("owner_kind") or "" if port == 0 and not ann and not lbl and not owner_kind and ns in ("external", "cluster-network", ""): return True return False _KIND_ALIASES = {"ReplicaSet": "Deployment"} _NOISE_ANNOTATION_PREFIXES = ( 'kubectl.kubernetes.io/', 'kubernetes.io/', 'openshift.io/', 'openshift.openshift.io/', 'k8s.v1.cni.cncf.io/', 'k8s.ovn.org/', 'seccomp.security.alpha.kubernetes.io/', ) def _resolve_kind(raw: str, labels: dict = None) -> str: if raw in _KIND_ALIASES: return _KIND_ALIASES[raw] if raw in ("Unknown", "") and labels: if labels.get("pod-template-hash"): return "Deployment" if labels.get("controller-revision-hash"): if labels.get("statefulset.kubernetes.io/pod-name"): return "StatefulSet" return "DaemonSet" return raw def _filter_summary_annotations(ann: dict) -> dict: if not ann or not isinstance(ann, dict): return ann or {} return { k: v for k, v in ann.items() if not any(k.startswith(p) for p in _NOISE_ANNOTATION_PREFIXES) and len(str(v)) < 500 } def _compact_service(entry: dict, direction: str = "downstream") -> dict: comm = entry.get("communication") or {} svc_type = comm.get("service_type", "unknown") svc_cat = comm.get("service_category", "") is_crit = comm.get("is_critical", False) if not svc_cat: svc_cat = self.classify_service_category(svc_type, entry.get("pod_name", "")) if not is_crit: is_crit = self.is_critical_service(svc_type, entry.get("pod_name", "")) labels = _safe_labels(entry) result = { "name": _workload_name(entry), "namespace": entry.get("namespace", ""), "kind": _resolve_kind(entry.get("owner_kind") or "", labels), "annotations": _filter_summary_annotations(entry.get("annotations", {})), "labels": labels, "is_critical": is_crit, "service_type": svc_type, "service_category": svc_cat, "port": comm.get("port"), } hop = entry.get("hop_count", 1) if hop > 1: result["hop_count"] = hop return result def _dedup_entries(entries: list) -> list: """Deduplicate dependency entries by logical workload identity, collapsing replica pods that share the same Deployment/StatefulSet. When duplicates exist, keeps the entry with the lowest hop_count so multi-replica merges preserve the shortest path.""" seen: Dict[str, dict] = {} for entry in entries: key = _workload_key(entry) if key not in seen: seen[key] = entry elif entry.get("hop_count", 1) < seen[key].get("hop_count", 1): seen[key] = entry return list(seen.values()) def _filter_and_dedup(entries: list) -> list: """Remove noise entries then deduplicate.""" return _dedup_entries([e for e in entries if not _is_noise_entry(e)]) def _group_by_category(entries: list, direction: str = "downstream") -> dict: by_cat: Dict[str, list] = {} crit_count = 0 for entry in entries: compact = _compact_service(entry, direction) cat = compact.pop("service_category", "") or "other" by_cat.setdefault(cat, []).append(compact) if compact.get("is_critical"): crit_count += 1 return { "total": len(entries), "critical_count": crit_count, "by_category": by_cat, } # --- Unified loop: works identically for single and multi-service --- workload_map: Dict[str, dict] = {} for res in results: up = res.get("upstream", {}) ds = res.get("downstream", []) cl = res.get("callers", []) up_key = _workload_key(up) up_labels = _safe_labels(up) if up_key in workload_map: workload_map[up_key]["_raw_downstream"].extend(ds) workload_map[up_key]["_raw_callers"].extend(cl) existing = workload_map[up_key] resolved = _resolve_kind(up.get("owner_kind") or "", up_labels) if (not existing["kind"] or existing["kind"] == "Unknown") and resolved not in ("", "Unknown"): existing["kind"] = resolved if not existing["annotations"] and up.get("annotations"): existing["annotations"] = _filter_summary_annotations(up["annotations"]) else: workload_map[up_key] = { "name": _workload_name(up), "namespace": up.get("namespace", ""), "kind": _resolve_kind(up.get("owner_kind") or "", up_labels), "annotations": _filter_summary_annotations(up.get("annotations", {})), "labels": up_labels, "_raw_downstream": list(ds), "_raw_callers": list(cl), } # Build per-service grouped downstream/callers and collect for global summary all_downstream_raw = [] all_callers_raw = [] matched_services = [] for wk_data in workload_map.values(): raw_ds = wk_data.pop("_raw_downstream") raw_cl = wk_data.pop("_raw_callers") all_downstream_raw.extend(raw_ds) all_callers_raw.extend(raw_cl) wk_data["downstream"] = _group_by_category(_filter_and_dedup(raw_ds), "downstream") wk_data["callers"] = _group_by_category(_filter_and_dedup(raw_cl), "caller") matched_services.append(wk_data) # Global summary with cross-service deduplication global_ds = _filter_and_dedup(all_downstream_raw) global_cl = _filter_and_dedup(all_callers_raw) global_ds_grouped = _group_by_category(global_ds, "downstream") global_cl_grouped = _group_by_category(global_cl, "caller") is_multi = len(matched_services) > 1 all_namespaces = sorted(set(s["namespace"] for s in matched_services if s["namespace"])) if is_multi: if len(all_namespaces) == 1: svc_label = f"{all_namespaces[0]} ({len(matched_services)} services)" svc_ns = all_namespaces[0] else: svc_label = f"{len(matched_services)} services across {len(all_namespaces)} namespaces" svc_ns = ", ".join(all_namespaces) service_info = { "name": svc_label, "namespace": svc_ns, "kind": "", "annotations": {}, "labels": {}, } else: ms = matched_services[0] service_info = { "name": ms["name"], "namespace": ms["namespace"], "kind": ms.get("kind", ""), "annotations": ms.get("annotations", {}), "labels": ms.get("labels", {}), } return { "success": True, "analysis_ids": int_ids, "multi_service": is_multi, "summary": { "total_matched": len(matched_services), "total_downstream_unique": global_ds_grouped["total"], "total_callers_unique": global_cl_grouped["total"], "downstream_critical_count": global_ds_grouped["critical_count"], "callers_critical_count": global_cl_grouped["critical_count"], }, "service": service_info, "matched_services": matched_services, } def diff_pod_dependencies( self, analysis_id_before: str, analysis_id_after: str, pod_name: Optional[str] = None, namespace: Optional[str] = None, owner_name: Optional[str] = None, cluster_id: Optional[str] = None, ) -> Dict[str, Any]: """Compare dependencies between two analysis runs.""" search_kwargs = dict( pod_name=pod_name, namespace=namespace, owner_name=owner_name, cluster_id=cluster_id, depth=1, ) before = self.find_pod_dependencies(analysis_id=analysis_id_before, **search_kwargs) after = self.find_pod_dependencies(analysis_id=analysis_id_after, **search_kwargs) def _extract_deps(result): deps = {} if result.get("success"): for r in result.get("results", []): for d in r.get("downstream", []): key = f"{d.get('namespace', '')}/{d.get('pod_name', '')}" deps[key] = d return deps before_deps = _extract_deps(before) after_deps = _extract_deps(after) before_keys = set(before_deps.keys()) after_keys = set(after_deps.keys()) added = [] for k in sorted(after_keys - before_keys): d = after_deps[k] comm = d.get("communication", {}) added.append({ "name": d.get("pod_name"), "namespace": d.get("namespace"), "port": comm.get("port"), "protocol": comm.get("protocol"), "service_type": comm.get("service_type"), }) removed = [] for k in sorted(before_keys - after_keys): d = before_deps[k] comm = d.get("communication", {}) removed.append({ "name": d.get("pod_name"), "namespace": d.get("namespace"), "port": comm.get("port"), "protocol": comm.get("protocol"), "service_type": comm.get("service_type"), }) changed = [] for k in sorted(before_keys & after_keys): b_comm = before_deps[k].get("communication", {}) a_comm = after_deps[k].get("communication", {}) changes = [] for field in ("port", "protocol", "app_protocol", "service_type"): bv = b_comm.get(field) av = a_comm.get(field) if bv != av: changes.append(field) if changes: changed.append({ "name": after_deps[k].get("pod_name"), "namespace": after_deps[k].get("namespace"), "change": "_".join(changes) + "_changed", "before": {f: b_comm.get(f) for f in changes}, "after": {f: a_comm.get(f) for f in changes}, }) unchanged_count = len(before_keys & after_keys) - len(changed) service_name = owner_name or pod_name or namespace or "unknown" return { "success": True, "service": service_name, "analysis_before": analysis_id_before, "analysis_after": analysis_id_after, "added_dependencies": added, "removed_dependencies": removed, "changed_dependencies": changed, "unchanged_count": unchanged_count, "summary": f"{len(added)} added, {len(removed)} removed, {len(changed)} changed, {unchanged_count} unchanged", } def format_dependency_graph(self, result: Dict[str, Any], fmt: str = "json") -> Any: """Format dependency stream result as Mermaid, DOT, or JSON.""" if fmt == "json" or not result.get("success"): return result lines = [] for r in result.get("results", []): upstream = r.get("upstream", {}) up_name = (upstream.get("owner_name") or upstream.get("pod_name") or "unknown").replace("-", "_") for d in r.get("downstream", []): name = (d.get("owner_name") or d.get("pod_name") or "unknown").replace("-", "_") comm = d.get("communication", {}) proto = comm.get("protocol") or "TCP" port = comm.get("port") or 0 req = comm.get("request_count") or 0 label = f"{proto}:{port} ({self._format_count(req)} req)" lines.append((up_name, name, label, "downstream")) for c in r.get("callers", []): name = (c.get("owner_name") or c.get("pod_name") or "unknown").replace("-", "_") comm = c.get("communication", {}) proto = comm.get("protocol") or "TCP" port = comm.get("port") or 0 req = comm.get("request_count") or 0 label = f"{proto}:{port} ({self._format_count(req)} req)" lines.append((name, up_name, label, "caller")) if fmt == "mermaid": out = ["graph LR"] for src, dst, label, _ in lines: out.append(f' {src} -->|"{label}"| {dst}') return "\n".join(out) if fmt == "dot": out = ["digraph dependencies {", " rankdir=LR;"] for src, dst, label, _ in lines: out.append(f' {src} -> {dst} [label="{label}"];') out.append("}") return "\n".join(out) return result @staticmethod def _format_count(n: int) -> str: if n >= 1_000_000: return f"{n / 1_000_000:.1f}M" if n >= 1_000: return f"{n / 1_000:.1f}K" return str(n) PORT_SERVICE_MAP = { # Relational databases 5432: "postgresql", 3306: "mysql", 33060: "mysql-x", 1433: "mssql", 1434: "mssql-browser", 1521: "oracle", 1830: "oracle-net", 50000: "db2", 26257: "cockroachdb", 4000: "tidb", 5433: "postgresql", # also used by YugabyteDB; name-based detection resolves ambiguity # NoSQL / Document 27017: "mongodb", 27018: "mongodb", 27019: "mongodb", 5984: "couchdb", 8091: "couchbase", 8092: "couchbase", 8093: "couchbase", 11210: "couchbase", 8529: "arangodb", 8086: "influxdb", # Key-value / Cache 6379: "redis", 6380: "redis", 16379: "redis-sentinel", 26379: "redis-sentinel", 11211: "memcached", 5701: "hazelcast", 3001: "aerospike", 6060: "dragonflydb", # Wide-column / Column-family 9042: "cassandra", 7000: "cassandra-inter", 7001: "cassandra-ssl", 9160: "cassandra-thrift", 19042: "scylladb", 16000: "hbase-master", 16020: "hbase-region", 8123: "clickhouse", 9440: "clickhouse-native", 8082: "druid", # Graph databases 7687: "neo4j", 7474: "neo4j-http", 8182: "janusgraph", 9080: "dgraph", # Time-series 8428: "victoriametrics", 4242: "opentsdb", # Message brokers / Streaming 9092: "kafka", 9093: "kafka-ssl", 9094: "kafka", 5672: "rabbitmq", 15672: "rabbitmq-mgmt", 25672: "rabbitmq-dist", 4222: "nats", 6222: "nats-cluster", 8222: "nats-monitor", 61616: "activemq", 5673: "activemq-amqp", 6650: "pulsar", 9876: "rocketmq", 1883: "mqtt", 8883: "mqtt-ssl", # Search engines 9200: "elasticsearch", 9300: "elasticsearch-transport", 7700: "meilisearch", 8983: "solr", 19530: "milvus", 6333: "qdrant", 6334: "qdrant-grpc", # Service discovery / Config 2181: "zookeeper", 8500: "consul", 8501: "consul-https", 2379: "etcd", 2380: "etcd-peer", 8848: "nacos", # Object storage 9000: "minio", # also used by ClickHouse native; name-based detection resolves ambiguity # LDAP / Identity 389: "ldap", 636: "ldaps", 88: "kerberos", # Monitoring / Observability 9090: "prometheus", 3100: "loki", 9411: "zipkin", 14268: "jaeger", 6831: "jaeger-thrift", 4317: "otlp-grpc", 4318: "otlp-http", # HTTP / API (generic -- name-based detection refines ambiguous ports) 80: "http-api", 8080: "http-api", 8081: "http-api", 443: "https-api", 8443: "https-api", 3000: "http-api", # gRPC 50051: "grpc", 50052: "grpc", # DNS 53: "dns", 5353: "dns", # SSH / FTP 22: "ssh", 21: "ftp", 990: "ftps", # SMTP / Mail 25: "smtp", 465: "smtps", 587: "smtp-submission", 143: "imap", 993: "imaps", } SERVICE_CATEGORY_MAP = { "database": { "postgresql", "mysql", "mysql-x", "mssql", "mssql-browser", "oracle", "oracle-net", "db2", "cockroachdb", "tidb", "yugabytedb", "mongodb", "couchdb", "couchbase", "arangodb", "cassandra", "cassandra-inter", "cassandra-ssl", "cassandra-thrift", "scylladb", "hbase-master", "hbase-region", "clickhouse", "clickhouse-native", "druid", "neo4j", "neo4j-http", "janusgraph", "dgraph", "influxdb", "opentsdb", "vitess", "percona", "mariadb", "singlestore", "timescaledb", "cratedb", "voltdb", "greenplum", "citusdb", "spanner", "cosmosdb", "dynamodb", "firestore", "fauna", }, "cache": { "redis", "redis-sentinel", "memcached", "hazelcast", "aerospike", "dragonflydb", "varnish", "keydb", }, "message_broker": { "kafka", "kafka-ssl", "rabbitmq", "rabbitmq-mgmt", "rabbitmq-dist", "nats", "nats-cluster", "nats-monitor", "activemq", "activemq-amqp", "pulsar", "pulsar-http", "rocketmq", "mqtt", "mqtt-ssl", "redpanda", "amazon-sqs", "azure-servicebus", "google-pubsub", }, "search_engine": { "elasticsearch", "elasticsearch-transport", "opensearch", "solr", "meilisearch", "milvus", "qdrant", "qdrant-grpc", "typesense", "algolia", "weaviate", "pinecone", }, "service_discovery": { "zookeeper", "consul", "consul-https", "etcd", "etcd-peer", "nacos", "eureka", }, "identity": { "ldap", "ldaps", "kerberos", "keycloak", "okta", "auth0", }, "object_storage": { "minio", "ceph", "swift", }, "observability": { "prometheus", "victoriametrics", "loki", "zipkin", "jaeger", "jaeger-thrift", "otlp-grpc", "otlp-http", "grafana", "datadog", "newrelic", "splunk", }, "api_gateway": { "http-api", "https-api", "grpc", }, "mail": { "smtp", "smtps", "smtp-submission", "imap", "imaps", }, "dns": {"dns"}, "file_transfer": {"ssh", "ftp", "ftps"}, } CRITICAL_CATEGORIES = frozenset({ "database", "cache", "message_broker", "search_engine", "service_discovery", "identity", "object_storage", }) NAME_CATEGORY_PATTERNS = { "database": [ "postgres", "mysql", "mariadb", "mssql", "sqlserver", "oracle", "mongo", "couch", "dynamo", "fauna", "cockroach", "tidb", "yugabyte", "cassandra", "scylla", "hbase", "clickhouse", "druid", "neo4j", "janusgraph", "dgraph", "arangodb", "influx", "timescale", "opentsdb", "crate", "voltdb", "greenplum", "citus", "spanner", "cosmos", "firestore", "vitess", "percona", "singlestore", "database", "rds", "-db-", "-db", "db-", ], "cache": [ "redis", "memcache", "hazelcast", "aerospike", "dragonfly", "varnish", "keydb", "cache", ], "message_broker": [ "kafka", "rabbitmq", "rabbit", "nats", "activemq", "pulsar", "rocketmq", "mqtt", "redpanda", "broker", "queue", "messaging", "eventbus", "servicebus", "pubsub", "stream", ], "search_engine": [ "elastic", "opensearch", "solr", "meilisearch", "milvus", "qdrant", "typesense", "weaviate", "pinecone", "algolia", "search", ], "service_discovery": [ "zookeeper", "consul", "etcd", "nacos", "eureka", "registry", "discovery", ], "identity": [ "ldap", "keycloak", "okta", "auth0", "identity", "iam", "sso", ], "object_storage": [ "minio", "ceph", "swift", "s3", "blob", "storage", ], } @classmethod def classify_service_category(cls, service_type: str, workload_name: str = "") -> str: if service_type and service_type != "unknown": if service_type in cls.SERVICE_CATEGORY_MAP: return service_type for category, types in cls.SERVICE_CATEGORY_MAP.items(): if service_type in types: return category name_lower = (workload_name or "").lower() if name_lower: for category, patterns in cls.NAME_CATEGORY_PATTERNS.items(): for pattern in patterns: if pattern in name_lower: return category return "service" @classmethod def is_critical_service(cls, service_type: str, workload_name: str = "") -> bool: return cls.classify_service_category(service_type, workload_name) in cls.CRITICAL_CATEGORIES def _detect_service_type(self, port: int, app_protocol: str = None, workload_name: str = None) -> str: if app_protocol: proto_lower = str(app_protocol).lower() if proto_lower in ("grpc", "http", "https", "dns"): return proto_lower if proto_lower != "http" else "http-api" if port and port in self.PORT_SERVICE_MAP: return self.PORT_SERVICE_MAP[port] if workload_name: name_lower = workload_name.lower() for category, patterns in self.NAME_CATEGORY_PATTERNS.items(): for pattern in patterns: if pattern in name_lower: for svc_type in self.SERVICE_CATEGORY_MAP.get(category, set()): if pattern in svc_type: return svc_type return category return "unknown" def _build_communication_contract( self, protocol, app_protocol, port, request_count, bytes_transferred, error_count, retransmit_count, avg_latency_ms, last_seen, workload_name: str = "" ) -> dict: port_val = int(port) if port else 0 req = int(request_count) if request_count else 0 err = int(error_count) if error_count else 0 error_rate = round((err / req) * 100, 4) if req > 0 else 0.0 svc_type = self._detect_service_type(port_val, app_protocol, workload_name) svc_category = self.classify_service_category(svc_type, workload_name) return { "protocol": protocol, "app_protocol": app_protocol, "port": port_val, "service_type": svc_type, "service_category": svc_category, "is_critical": svc_category in self.CRITICAL_CATEGORIES, "request_count": req, "bytes_transferred": int(bytes_transferred) if bytes_transferred else 0, "error_count": err, "error_rate_percent": error_rate, "retransmit_count": int(retransmit_count) if retransmit_count else 0, "avg_latency_ms": round(float(avg_latency_ms), 2) if avg_latency_ms else 0.0, "last_seen": last_seen, } def _calculate_dependency_health( self, request_count, error_count, retransmit_count, avg_latency_ms ) -> dict: req = int(request_count) if request_count else 0 err = int(error_count) if error_count else 0 retx = int(retransmit_count) if retransmit_count else 0 latency = float(avg_latency_ms) if avg_latency_ms else 0.0 error_rate = (err / req) * 100 if req > 0 else 0.0 retransmit_rate = (retx / req) * 100 if req > 0 else 0.0 score = 100 risk_factors = [] if error_rate > 5: score -= 30 risk_factors.append(f"high_error_rate:{error_rate:.2f}%") elif error_rate > 1: score -= 20 risk_factors.append(f"elevated_error_rate:{error_rate:.2f}%") if retransmit_rate > 10: score -= 20 risk_factors.append(f"high_retransmit_rate:{retransmit_rate:.2f}%") elif retransmit_rate > 5: score -= 10 risk_factors.append(f"elevated_retransmit_rate:{retransmit_rate:.2f}%") if latency > 500: score -= 25 risk_factors.append(f"very_high_latency:{latency:.1f}ms") elif latency > 100: score -= 15 risk_factors.append(f"high_latency:{latency:.1f}ms") if req == 0: score = 0 risk_factors.append("no_traffic") score = max(0, score) if score >= 80: status = "healthy" elif score >= 60: status = "degraded" elif score >= 30: status = "unhealthy" else: status = "critical" return { "score": score, "status": status, "error_rate_percent": round(error_rate, 4), "retransmit_rate_percent": round(retransmit_rate, 4), "avg_latency_ms": round(latency, 2), "risk_factors": risk_factors, } def _is_ip_address(self, value: str) -> bool: """Check if a string is a valid IP address""" if not value: return False import re # IPv4 pattern ipv4_pattern = r'^(\d{1,3}\.){3}\d{1,3}$' if re.match(ipv4_pattern, value): parts = value.split('.') return all(0 <= int(p) <= 255 for p in parts) return False def close(self): """Close Neo4j driver connection""" if self.driver: self.driver.close() logger.info("🔌 Neo4j connection closed") # Singleton instance _query_engine_instance: Optional[GraphQueryEngine] = None def get_query_engine() -> GraphQueryEngine: """Get singleton query engine instance""" global _query_engine_instance if _query_engine_instance is None: _query_engine_instance = GraphQueryEngine() return _query_engine_instance # Create singleton instance for direct import graph_query_engine = get_query_engine()