""" Simulation Router - Impact simulation and network policy endpoints Provides APIs for simulating changes and generating network policies """ from fastapi import APIRouter, Depends, HTTPException, Query, status from fastapi.responses import StreamingResponse from typing import Optional, List from datetime import datetime import structlog import json import csv import io import uuid from utils.jwt_utils import get_current_user from services.network_policy_service import get_network_policy_service, NetworkPolicyService from schemas.simulation import ( NetworkPolicyPreviewRequest, NetworkPolicyPreviewResponse, NetworkPolicyGenerateRequest, NetworkPolicyGenerateResponse, ImpactSimulationRequest, ImpactSimulationResponse, ImpactSimulationExportReport, ExportMetadata, SimulationExportData, ImpactSummary, AffectedService, SimulationDetails, NoDependencyInfo, ImpactLevel, ImpactCategory, DependencyType, ChangeType, ChangeTypeCharacteristics, ) logger = structlog.get_logger(__name__) router = APIRouter() def get_service() -> NetworkPolicyService: """Dependency provider for NetworkPolicyService""" return get_network_policy_service() # ============================================================================= # Helper Functions for Impact Simulation # ============================================================================= def is_infrastructure_endpoint(name: str, namespace: str) -> bool: """ Check if an endpoint is infrastructure (SDN, cluster network) that should be de-prioritized or filtered from impact reports. These are typically: - SDN infrastructure IPs (10.128.x.x - 10.131.x.x with random ports) - Cluster network IPs that are not actual services - Health check endpoints """ import re # Infrastructure namespaces infra_namespaces = { 'sdn-infrastructure', 'cluster-network', 'internal-network', 'openshift-sdn', 'openshift-ovn-kubernetes' } if namespace in infra_namespaces: # Check if it's just an IP (not a real service name) ip_pattern = r'^(\d{1,3}\.){3}\d{1,3}$' if re.match(ip_pattern, name): return True return False def is_real_service_dependency(dep: dict) -> bool: """ Determine if a dependency is a real service vs infrastructure noise. Real services: - Have meaningful names (not just IPs) - Are in application namespaces - Use well-known ports - Have significant request counts Infrastructure noise: - Raw IP addresses in sdn-infrastructure namespace - Random high ports (ephemeral ports > 32000) - Very low request counts on random ports """ name = dep.get("name", "") namespace = dep.get("namespace", "") port = dep.get("port", 0) request_count = dep.get("request_count", 0) # Always include if it's in an application namespace app_namespace_patterns = ['internal-', 'prod-', 'dev-', 'staging-', 'test-'] is_app_namespace = any(namespace.startswith(p) for p in app_namespace_patterns) # Exclude infrastructure IPs with random ports if is_infrastructure_endpoint(name, namespace): # Unless it's a well-known port with traffic well_known_ports = {22, 80, 443, 3306, 5432, 6379, 6443, 8080, 8443, 9090, 9200, 27017} if port not in well_known_ports: return False # Include if it has a real service name (not just an IP) import re ip_pattern = r'^(\d{1,3}\.){3}\d{1,3}$' has_real_name = not re.match(ip_pattern, name) # Include if it's a well-known port well_known_ports = { 22, 25, 53, 80, 443, 3306, 5432, 5433, 6379, 6443, 8000, 8080, 8443, 9090, 9200, 9300, 27017, 5672, 15672 } is_well_known_port = port in well_known_ports # Include if it has significant traffic has_significant_traffic = request_count >= 5 # Decision logic if has_real_name: return True if is_well_known_port: return True if is_app_namespace and has_significant_traffic: return True return False def extract_target_dependencies(graph_data: dict, target_name: str, target_namespace: str) -> dict: """ Extract direct and indirect dependencies for a target from the full graph. Args: graph_data: Full graph response from graph-query service with nodes and edges target_name: Name of the target workload target_namespace: Namespace of the target workload Returns: Dictionary with direct_dependencies, indirect_dependencies, node_matches """ if not graph_data: return { "direct_dependencies": [], "indirect_dependencies": [], "node_matches": 0, "has_external_connections": False, "confidence": 0.0 } nodes = graph_data.get("nodes", []) edges = graph_data.get("edges", []) # Build node lookup node_map = {n.get("id"): n for n in nodes} # Extract actual workload name from target_name # target_name can be in format: "{kind}-{namespace}-{name}" (e.g., "deployment-flowfish-prod-frontend") # or just the workload name (e.g., "frontend") target_name_lower = target_name.lower() target_ns_lower = target_namespace.lower() # Try to extract the actual workload name if target_name is in kind-namespace-name format workload_name = target_name_lower kind_prefixes = ['deployment-', 'statefulset-', 'daemonset-', 'replicaset-', 'pod-', 'service-'] for prefix in kind_prefixes: if target_name_lower.startswith(prefix): # Remove kind prefix remaining = target_name_lower[len(prefix):] # Check if namespace is also in the name if remaining.startswith(target_ns_lower + '-'): workload_name = remaining[len(target_ns_lower) + 1:] elif '-' in remaining: # Try to find namespace in the remaining string parts = remaining.split('-') # Find where namespace ends and workload name begins for i in range(len(parts)): potential_ns = '-'.join(parts[:i+1]) if potential_ns == target_ns_lower: workload_name = '-'.join(parts[i+1:]) break break logger.info(f"Target matching: original='{target_name}', extracted_workload='{workload_name}', namespace='{target_namespace}'") # Find target node(s) - match by name containing target_name and namespace target_ids = set() for node in nodes: node_name = node.get("name", "").lower() node_ns = node.get("namespace", "").lower() # Check namespace match first ns_match = (target_ns_lower == node_ns or target_ns_lower in node_ns or node_ns in target_ns_lower) if not ns_match: continue # Match strategies: # 1. Exact workload name match (node starts with workload name) # 2. Original target_name in node_name or vice versa # 3. Workload name in node_name if node_name.startswith(workload_name + '-') or node_name == workload_name: target_ids.add(node.get("id")) elif target_name_lower in node_name or node_name in target_name_lower: target_ids.add(node.get("id")) elif workload_name in node_name: target_ids.add(node.get("id")) if not target_ids: # Fallback: try matching just by workload name across all namespaces for node in nodes: node_name = node.get("name", "").lower() if node_name.startswith(workload_name + '-') or workload_name in node_name: target_ids.add(node.get("id")) logger.info(f"Found {len(target_ids)} target nodes for {target_namespace}/{target_name} (workload: {workload_name})") # Find direct dependencies (1-hop) direct_deps = [] direct_ids = set() for edge in edges: source_id = edge.get("source_id") or edge.get("source", {}).get("id") target_id = edge.get("target_id") or edge.get("target", {}).get("id") # Check if this edge connects to target if source_id in target_ids and target_id not in target_ids: dep_node = node_map.get(target_id, {}) direct_deps.append({ "id": target_id, "name": dep_node.get("name", "unknown"), "namespace": dep_node.get("namespace", "default"), "kind": dep_node.get("kind", "Pod"), "port": edge.get("port", 0), "protocol": edge.get("protocol", "TCP"), "request_count": edge.get("request_count", 0), "direction": "outgoing" }) direct_ids.add(target_id) elif target_id in target_ids and source_id not in target_ids: dep_node = node_map.get(source_id, {}) direct_deps.append({ "id": source_id, "name": dep_node.get("name", "unknown"), "namespace": dep_node.get("namespace", "default"), "kind": dep_node.get("kind", "Pod"), "port": edge.get("port", 0), "protocol": edge.get("protocol", "TCP"), "request_count": edge.get("request_count", 0), "direction": "incoming" }) direct_ids.add(source_id) # Find indirect dependencies (2-hop) indirect_deps = [] for edge in edges: source_id = edge.get("source_id") or edge.get("source", {}).get("id") target_id = edge.get("target_id") or edge.get("target", {}).get("id") # Check if this edge connects from a direct dependency to something else if source_id in direct_ids and target_id not in target_ids and target_id not in direct_ids: dep_node = node_map.get(target_id, {}) indirect_deps.append({ "id": target_id, "name": dep_node.get("name", "unknown"), "namespace": dep_node.get("namespace", "default"), "kind": dep_node.get("kind", "Pod"), "port": edge.get("port", 0), "protocol": edge.get("protocol", "TCP"), "request_count": edge.get("request_count", 0) }) elif target_id in direct_ids and source_id not in target_ids and source_id not in direct_ids: dep_node = node_map.get(source_id, {}) indirect_deps.append({ "id": source_id, "name": dep_node.get("name", "unknown"), "namespace": dep_node.get("namespace", "default"), "kind": dep_node.get("kind", "Pod"), "port": edge.get("port", 0), "protocol": edge.get("protocol", "TCP"), "request_count": edge.get("request_count", 0) }) # Deduplicate and filter infrastructure noise seen_direct = set() unique_direct = [] filtered_infra_count = 0 for d in direct_deps: key = f"{d['namespace']}/{d['name']}" if key not in seen_direct: seen_direct.add(key) # Filter out infrastructure noise but keep real services if is_real_service_dependency(d): unique_direct.append(d) else: filtered_infra_count += 1 seen_indirect = set() unique_indirect = [] for d in indirect_deps: key = f"{d['namespace']}/{d['name']}" if key not in seen_indirect and key not in seen_direct: seen_indirect.add(key) if is_real_service_dependency(d): unique_indirect.append(d) else: filtered_infra_count += 1 if filtered_infra_count > 0: logger.info(f"Filtered {filtered_infra_count} infrastructure endpoints from impact report") # Check for external connections has_external = any( d.get("namespace") in ["external"] or "." in d.get("name", "") for d in unique_direct + unique_indirect ) # Sort by importance: real services first, then by request count unique_direct.sort(key=lambda x: (-x.get("request_count", 0))) unique_indirect.sort(key=lambda x: (-x.get("request_count", 0))) return { "direct_dependencies": unique_direct, "indirect_dependencies": unique_indirect, "node_matches": len(target_ids), "has_external_connections": has_external, "confidence": min(1.0, 0.5 + (len(unique_direct) * 0.1)) if unique_direct else 0.3, "filtered_infrastructure_count": filtered_infra_count } def classify_endpoint_kind(name: str, namespace: str, original_kind: str) -> str: """ Properly classify the kind of an endpoint based on its name and namespace. Returns: - "Pod" for actual Kubernetes pods (e.g., backend-7686dccc6b-x8bqm) - "Service" for Kubernetes services - "ExternalIP" for external IP addresses (10.180.x.x, 192.168.x.x) - "ClusterIP" for cluster-internal IPs (10.128.x.x, 10.129.x.x, 10.130.x.x, 10.131.x.x) - "ExternalDNS" for external DNS names (*.bank, api.*, etc.) - "ClusterService" for internal cluster services (*.svc.cluster.local) - "Localhost" for localhost """ import re # Check for localhost first if name.lower() == 'localhost' or name == '127.0.0.1': return "Localhost" # Check if it's an IP address ip_pattern = r'^(\d{1,3}\.){3}\d{1,3}$' if re.match(ip_pattern, name): # Parse IP to determine type parts = name.split('.') first_octet = int(parts[0]) second_octet = int(parts[1]) # Kubernetes pod network IPs (typically 10.128.x.x - 10.131.x.x in OpenShift) if first_octet == 10 and 128 <= second_octet <= 131: # These are pod IPs within the cluster if namespace in ['sdn-infrastructure']: return "SDN-IP" return "ClusterIP" # Service network IPs (typically 10.96.x.x or 10.106.x.x) if first_octet == 10 and (96 <= second_octet <= 111 or 104 <= second_octet <= 111): return "ServiceIP" # External/infrastructure IPs if first_octet == 10 and second_octet == 180: return "ExternalIP" if first_octet == 192 and second_octet == 168: return "ExternalIP" # Default IP classification based on namespace if namespace in ['external', 'sdn-infrastructure', 'internal-network', 'cluster-network']: return "ExternalIP" return "ClusterIP" # Check if it's a DNS-style name with IP prefix (e.g., 10-128-22-163.harbor-core...) ip_dns_pattern = r'^\d+-\d+-\d+-\d+\.' if re.match(ip_dns_pattern, name): if '.svc.cluster.local' in name: return "ClusterService" return "ClusterIP" # Check if it's a cluster service DNS name if '.svc.cluster.local' in name: return "ClusterService" # Check if it's an external DNS name if '.' in name and name.count('.') >= 2: # Common external DNS patterns if any(ext in name.lower() for ext in ['.bank', '.com', '.net', '.org', '.io', '.local']): if '.svc.cluster.local' not in name: return "ExternalDNS" # Check if it looks like a pod name (has random suffix like -7686dccc6b-x8bqm) pod_pattern = r'-[a-z0-9]{6,10}-[a-z0-9]{5}$' if re.search(pod_pattern, name): return "Pod" # Check if it looks like a StatefulSet pod (ends with -0, -1, etc.) statefulset_pattern = r'-\d+$' if re.search(statefulset_pattern, name): return "Pod" # Default to original kind or Pod return original_kind if original_kind else "Pod" def calculate_dynamic_risk_score( dependency_type: str, request_count: int, is_critical_namespace: bool, port: int, change_type: ChangeType ) -> tuple: """ Calculate a dynamic risk score based on multiple factors. Returns: tuple: (risk_score: float, risk_factors: list[str]) Factors: - Dependency type (direct = higher risk) - Request count (more traffic = higher risk) - Critical namespace (system namespaces = higher risk) - Port (well-known ports = higher risk) - Change type severity """ import math base_score = 0.3 risk_factors = [] # Dependency type factor if dependency_type == "direct": base_score += 0.3 risk_factors.append("Direct dependency") else: base_score += 0.1 # Request count factor (logarithmic scale) if request_count > 0: traffic_factor = min(0.2, math.log10(request_count + 1) * 0.05) base_score += traffic_factor if request_count >= 100: risk_factors.append(f"High traffic ({request_count} requests)") elif request_count >= 10: risk_factors.append(f"Moderate traffic ({request_count} requests)") # Critical namespace factor if is_critical_namespace: base_score += 0.15 risk_factors.append("System namespace") # Well-known port factor critical_ports = { 443: "HTTPS", 6443: "Kubernetes API", 8443: "Secure HTTP", 53: "DNS", 5432: "PostgreSQL", 3306: "MySQL", 6379: "Redis", 9090: "Prometheus", 9091: "Prometheus Pushgateway", 8080: "HTTP Alt", 8000: "HTTP Alt", 3000: "Dev Server", } if port in critical_ports: base_score += 0.1 risk_factors.append(f"Critical port ({critical_ports[port]})") # Change type severity factor severity_map = { ChangeType.DELETE: (0.15, "Destructive change"), ChangeType.SCALE_DOWN: (0.12, "Service unavailability"), ChangeType.NETWORK_ISOLATE: (0.12, "Network isolation"), ChangeType.NETWORK_POLICY_APPLY: (0.1, "Traffic filtering"), ChangeType.PORT_CHANGE: (0.1, "Port configuration change"), ChangeType.IMAGE_UPDATE: (0.08, "Image update"), ChangeType.CONFIG_CHANGE: (0.05, "Configuration change"), ChangeType.RESOURCE_CHANGE: (0.05, "Resource adjustment"), ChangeType.NETWORK_POLICY_REMOVE: (0.02, "Policy removal"), } if change_type in severity_map: score_add, factor = severity_map[change_type] base_score += score_add if score_add >= 0.1: risk_factors.append(factor) final_score = min(1.0, max(0.0, round(base_score, 2))) return final_score, risk_factors def generate_specific_recommendation( endpoint_kind: str, namespace: str, impact_level: ImpactLevel, change_type: ChangeType, request_count: int ) -> str: """ Generate specific recommendations based on endpoint characteristics. """ recommendations = [] # Kind-specific recommendations if endpoint_kind == "ExternalIP": recommendations.append("Verify external IP connectivity requirements") elif endpoint_kind == "ExternalDNS": recommendations.append("Check DNS resolution and external service availability") elif endpoint_kind == "ClusterService": recommendations.append("Review internal service mesh configuration") elif endpoint_kind == "Localhost": recommendations.append("Localhost connections typically indicate local processes") # Namespace-specific recommendations if 'openshift-' in namespace: recommendations.append("⚠️ OpenShift system component - proceed with caution") elif 'kube-system' in namespace: recommendations.append("⚠️ Kubernetes system component - may affect cluster stability") elif namespace == 'default': recommendations.append("Consider moving workloads out of default namespace") # Change-type specific recommendations change_recommendations = { ChangeType.DELETE: "Ensure graceful shutdown and data backup before deletion", ChangeType.SCALE_DOWN: "Verify HPA/VPA policies and set up alerts for scale-up", ChangeType.NETWORK_ISOLATE: "Test network policy in audit mode first", ChangeType.NETWORK_POLICY_APPLY: "Validate policy rules against observed traffic patterns", ChangeType.PORT_CHANGE: "Update all client configurations and service discovery", ChangeType.CONFIG_CHANGE: "Test configuration changes in staging environment", ChangeType.IMAGE_UPDATE: "Verify image compatibility and rollback strategy", ChangeType.RESOURCE_CHANGE: "Monitor resource utilization after change", ChangeType.NETWORK_POLICY_REMOVE: "Review security implications of removing policy", } if change_type in change_recommendations: recommendations.append(change_recommendations[change_type]) # Traffic-based recommendations if request_count > 100: recommendations.append(f"High traffic endpoint ({request_count} requests) - consider gradual rollout") elif request_count == 0: recommendations.append("No recent traffic observed - verify if endpoint is still in use") # Impact-level recommendations if impact_level == ImpactLevel.HIGH: recommendations.insert(0, "🔴 HIGH IMPACT: Coordinate with dependent teams before proceeding") elif impact_level == ImpactLevel.MEDIUM: recommendations.insert(0, "🟠 Monitor for cascading effects after change") return " | ".join(recommendations[:3]) if recommendations else "Review impact carefully before proceeding" def deduplicate_services(services: list) -> list: """ Remove duplicate services based on normalized endpoint identification. Keeps the entry with more information (e.g., pod name over IP, pod over DNS). Priority order (highest to lowest): 1. Pod names (e.g., backend-7686dccc6b-x8bqm) - priority 5 2. StatefulSet pods (e.g., postgresql-0) - priority 5 3. Service names (e.g., backend) - priority 4 4. External DNS (e.g., api.external-service.com) - priority 3 5. DNS names with IP prefix (e.g., 10-128-22-163.harbor-core...) - priority 2 6. IP addresses (e.g., 10.128.16.2) - priority 1 """ import re def get_name_priority(name: str) -> int: """Return priority score - higher is better""" # IP address - lowest priority if re.match(r'^(\d{1,3}\.){3}\d{1,3}$', name): return 1 # DNS-style with IP prefix (e.g., 10-128-22-163.harbor-core...) if re.match(r'^\d+-\d+-\d+-\d+\.', name): return 2 # External DNS (e.g., api.external-service.com) if '.' in name and name.count('.') >= 2: return 3 # Pod name with hash suffix if re.search(r'-[a-z0-9]{5,10}-[a-z0-9]{5}$', name): return 5 # StatefulSet pod (e.g., postgresql-0) if re.search(r'-\d+$', name): return 5 # Service name (no suffix) return 4 def extract_service_identifier(name: str, namespace: str, port: int) -> str: """Extract the base service name for deduplication""" import re # Handle DNS-style names like 10-128-22-163.harbor-core.prod-harbor-ha.svc.cluster.local if '.svc.cluster.local' in name: parts = name.split('.') if len(parts) >= 2: # Extract service name and namespace from DNS service_name = parts[1] dns_namespace = parts[2] if len(parts) > 2 else namespace return f"svc:{service_name}:{dns_namespace}:{port}" # Handle external DNS - keep unique by hostname if '.' in name and any(ext in name.lower() for ext in ['.bank', '.com', '.net', '.org', '.io']): return f"dns:{name}:{port}" # Handle IP addresses if re.match(r'^(\d{1,3}\.){3}\d{1,3}$', name): # For cluster IPs (10.128-131.x.x), try to group by service parts = name.split('.') if parts[0] == '10' and 128 <= int(parts[1]) <= 131: # These are likely pod IPs - group by namespace and port # They'll be deduplicated with their pod names return f"podip:{namespace}:{port}" # External IPs - keep as unique return f"extip:{name}:{port}" # Handle DNS with IP prefix (e.g., 10-128-22-163.harbor-core...) if re.match(r'^\d+-\d+-\d+-\d+\.', name): parts = name.split('.') if len(parts) >= 2: service_name = parts[1] dns_namespace = parts[2] if len(parts) > 2 else namespace return f"svc:{service_name}:{dns_namespace}:{port}" # Handle localhost if name.lower() == 'localhost': return f"localhost:{port}" # Handle pod names - extract base service name base_name = re.sub(r'-[a-z0-9]{5,10}-[a-z0-9]{5}$', '', name) base_name = re.sub(r'-\d+$', '', base_name) # Remove StatefulSet index return f"pod:{base_name}:{namespace}:{port}" seen = {} for svc in services: name = svc.get("name", "") namespace = svc.get("namespace", "") port = svc.get("connection_details", {}).get("port", 0) # Create a normalized key service_id = extract_service_identifier(name, namespace, port) current_priority = get_name_priority(name) if service_id in seen: existing = seen[service_id] existing_priority = get_name_priority(existing.get("name", "")) # Keep the one with higher priority (better name) if current_priority > existing_priority: seen[service_id] = svc # If same priority, keep the one with more request count elif current_priority == existing_priority: current_count = svc.get("connection_details", {}).get("request_count", 0) existing_count = existing.get("connection_details", {}).get("request_count", 0) if current_count > existing_count: seen[service_id] = svc else: seen[service_id] = svc return list(seen.values()) def filter_indirect_dependencies( indirect_deps: list, direct_deps: list, change_type: ChangeType, max_indirect: int = 50 ) -> list: """ Filter and limit indirect dependencies to meaningful ones. Rules: - Exclude infrastructure services (ingress, DNS, monitoring) unless directly relevant - Remove duplicate entries (same service via different paths) - Limit total indirect dependencies to max_indirect - Prioritize by request count and criticality """ import re # Get direct dependency names for reference direct_names = {d.get("name") for d in direct_deps} # Infrastructure namespaces to de-prioritize for most changes infra_namespaces = [ 'openshift-ingress', 'openshift-ingress-canary', 'openshift-dns', 'openshift-monitoring', 'openshift-network-diagnostics', 'kube-system', 'openshift-operators', 'openshift-authentication', 'openshift-console', 'openshift-ovn-kubernetes', 'openshift-migration' ] # For network policy changes, infrastructure IS relevant if change_type in [ChangeType.NETWORK_ISOLATE, ChangeType.NETWORK_POLICY_APPLY]: infra_namespaces = [] # Don't filter infrastructure # First pass: deduplicate by base service name seen_services = {} def get_base_service_name(name: str, namespace: str) -> str: """Extract base service name for deduplication""" # Handle DNS-style names if '.svc.cluster.local' in name: parts = name.split('.') return f"{parts[1]}:{parts[2] if len(parts) > 2 else namespace}" # Handle IP-prefixed DNS if re.match(r'^\d+-\d+-\d+-\d+\.', name): parts = name.split('.') if len(parts) >= 2: return f"{parts[1]}:{parts[2] if len(parts) > 2 else namespace}" # Handle IP addresses - group by namespace if re.match(r'^(\d{1,3}\.){3}\d{1,3}$', name): return f"ip:{namespace}" # Handle pod names base = re.sub(r'-[a-z0-9]{5,10}-[a-z0-9]{5}$', '', name) base = re.sub(r'-\d+$', '', base) return f"{base}:{namespace}" for dep in indirect_deps: name = dep.get("name", "") namespace = dep.get("namespace", "") # Skip if it's actually a direct dependency if name in direct_names: continue # Skip localhost and loopback if name.lower() == 'localhost' or name == '127.0.0.1': continue base_key = get_base_service_name(name, namespace) if base_key in seen_services: # Keep the one with higher request count existing = seen_services[base_key] if dep.get("request_count", 0) > existing.get("request_count", 0): seen_services[base_key] = dep else: seen_services[base_key] = dep # Second pass: separate infrastructure from application dependencies filtered = [] infra_deps = [] for dep in seen_services.values(): namespace = dep.get("namespace", "") is_infra = any(ns in namespace for ns in infra_namespaces) if is_infra: infra_deps.append(dep) else: filtered.append(dep) # Sort by request count (descending) to prioritize high-traffic dependencies filtered.sort(key=lambda x: x.get("request_count", 0), reverse=True) # Limit application dependencies app_limit = min(len(filtered), max_indirect - 5) # Reserve 5 slots for infra result = filtered[:app_limit] # Add a few infrastructure deps if we have room remaining_slots = max_indirect - len(result) if remaining_slots > 0 and infra_deps: infra_deps.sort(key=lambda x: x.get("request_count", 0), reverse=True) result.extend(infra_deps[:remaining_slots]) return result # ============================================================================= # Network Policy Endpoints # ============================================================================= @router.post( "/network-policy/generate", response_model=NetworkPolicyGenerateResponse, summary="Generate Network Policy", description="Generate a Kubernetes NetworkPolicy based on observed traffic patterns. " "Creates a least-privilege policy that allows only observed connections." ) async def generate_network_policy( request: NetworkPolicyGenerateRequest, current_user: dict = Depends(get_current_user), service: NetworkPolicyService = Depends(get_service) ): """ Generate a NetworkPolicy YAML based on observed traffic. The generated policy will: - Allow only traffic that was observed during the analysis period - Include DNS egress rules by default - Use namespace and pod selectors for internal traffic - Use IP blocks for external traffic """ try: logger.info( "Network policy generation requested", user=current_user.get("username"), cluster_id=request.cluster_id, target=f"{request.target_namespace}/{request.target_workload}" ) response = await service.generate_network_policy(request) logger.info( "Network policy generated successfully", policy_name=response.policy_name, ingress_sources=response.observed_ingress_sources, egress_destinations=response.observed_egress_destinations ) return response except Exception as e: logger.error("Network policy generation failed", error=str(e)) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to generate network policy: {str(e)}" ) @router.post( "/network-policy/preview", response_model=NetworkPolicyPreviewResponse, summary="Preview Network Policy Impact", description="Preview which existing connections would be affected by applying a network policy." ) async def preview_network_policy_impact( request: NetworkPolicyPreviewRequest, current_user: dict = Depends(get_current_user), service: NetworkPolicyService = Depends(get_service) ): """ Preview the impact of applying a network policy. Returns: - List of connections that would be blocked - List of connections that would be allowed - Warnings and recommendations """ try: logger.info( "Network policy preview requested", user=current_user.get("username"), cluster_id=request.cluster_id, target=f"{request.target_namespace}/{request.target_workload}" ) response = await service.preview_network_policy_impact(request) logger.info( "Network policy preview completed", policy_name=response.policy_name, blocked=response.blocked_connections, allowed=response.allowed_connections ) return response except Exception as e: logger.error("Network policy preview failed", error=str(e)) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to preview network policy: {str(e)}" ) # ============================================================================= # Impact Simulation Endpoints # ============================================================================= @router.post( "/impact", response_model=ImpactSimulationResponse, summary="Run Impact Simulation", description="Simulate the impact of a change on the target resource and its dependencies." ) async def run_impact_simulation( request: ImpactSimulationRequest, current_user: dict = Depends(get_current_user), service: NetworkPolicyService = Depends(get_service) ): """ Run an impact simulation for a proposed change. Analyzes: - Direct dependencies (1-hop) - Indirect dependencies (2-hop) - Impact severity for each affected service - Recommendations for safe change execution """ try: logger.info( "Impact simulation requested", user=current_user.get("username"), cluster_id=request.cluster_id, analysis_id=request.analysis_id, target=f"{request.target_namespace}/{request.target_name}", change_type=request.change_type.value ) # ========== DEBUG: IMPACT SIMULATION START ========== logger.warning("IMPACT_SIM_START", user=current_user.get('username'), cluster_id=request.cluster_id, analysis_id=request.analysis_id, target=f"{request.target_namespace}/{request.target_name}", target_kind=request.target_kind, change_type=request.change_type.value ) # Use graph-query service (same as communications endpoint) for dependency data # This ensures we use the same data source that powers the visualization from routers.communications import graph_query_client # Get full dependency graph from graph-query service logger.warning("IMPACT_SIM_GRAPH_FETCH: fetching graph from service") full_graph = await graph_query_client.get_dependency_graph( cluster_id=request.cluster_id, analysis_id=request.analysis_id, namespace=None, # Get all namespaces to find dependencies depth=2, search=request.target_name # Filter to target and its connections ) # DEBUG: Log full graph summary nodes_count = len(full_graph.get("nodes", [])) if full_graph else 0 edges_count = len(full_graph.get("edges", [])) if full_graph else 0 namespaces_in_graph = [] sample_nodes = [] if full_graph and full_graph.get("nodes"): namespaces_in_graph = sorted(set(n.get("namespace", "unknown") for n in full_graph["nodes"])) sample_nodes = [f"{n.get('namespace', '?')}/{n.get('name', '?')}" for n in full_graph["nodes"][:10]] logger.warning("IMPACT_SIM_GRAPH_RESPONSE: graph data received", total_nodes=nodes_count, total_edges=edges_count, namespaces=namespaces_in_graph, sample_nodes=sample_nodes ) # Extract dependencies for the target from the full graph graph_data = extract_target_dependencies( full_graph, request.target_name, request.target_namespace ) # DEBUG: Log extracted dependencies direct_deps_summary = [ f"{d.get('namespace', '?')}/{d.get('name', '?')}:{d.get('port', '?')}" for d in graph_data.get("direct_dependencies", [])[:10] ] indirect_deps_summary = [ f"{d.get('namespace', '?')}/{d.get('name', '?')}:{d.get('port', '?')}" for d in graph_data.get("indirect_dependencies", [])[:5] ] logger.warning("IMPACT_SIM_DEPENDENCIES: dependencies extracted", target=f"{request.target_namespace}/{request.target_name}", node_matches=graph_data.get('node_matches', 0), direct_count=len(graph_data.get('direct_dependencies', [])), indirect_count=len(graph_data.get('indirect_dependencies', [])), has_external=graph_data.get('has_external_connections', False), filtered_infra=graph_data.get('filtered_infrastructure_count', 0), direct_deps=direct_deps_summary, indirect_deps=indirect_deps_summary ) logger.info( "Graph data extracted", direct_count=len(graph_data.get("direct_dependencies", [])), indirect_count=len(graph_data.get("indirect_dependencies", [])), node_matches=graph_data.get("node_matches", 0) ) # Get change type characteristics change_chars = ChangeTypeCharacteristics.get(request.change_type) # Calculate affected services with change-type aware impact affected_services = [] direct_deps = graph_data.get("direct_dependencies", []) indirect_deps_raw = graph_data.get("indirect_dependencies", []) # Filter and limit indirect dependencies to prevent explosion indirect_deps = filter_indirect_dependencies( indirect_deps_raw, direct_deps, request.change_type, max_indirect=50 # Reasonable limit ) # Process direct dependencies for dep in direct_deps: dep_name = dep.get("name", "unknown") dep_namespace = dep.get("namespace", "default") dep_port = dep.get("port", 0) request_count = dep.get("request_count", 0) # Properly classify the endpoint kind original_kind = dep.get("kind", "Pod") endpoint_kind = classify_endpoint_kind(dep_name, dep_namespace, original_kind) # Use new change-type aware calculation impact, impact_category, impact_desc, risk_factors = service.calculate_impact_for_change_type( request.change_type, DependencyType.DIRECT, request_count, dep.get("is_critical", False) ) # Skip if this change type doesn't affect direct dependencies if impact == ImpactLevel.NONE and not change_chars["affects_direct"]: continue # Downgrade impact for infrastructure endpoints (SDN-IP, ExternalIP with random ports) is_infra_endpoint = endpoint_kind in ['SDN-IP', 'ServiceIP'] or \ (endpoint_kind == 'ExternalIP' and dep_port > 32000) if is_infra_endpoint and impact == ImpactLevel.HIGH: # Infrastructure endpoints are less critical for application impact impact = ImpactLevel.MEDIUM impact_desc = f"Infrastructure connectivity may be affected. {impact_desc}" # Calculate dynamic risk score and risk factors risk_score, calculated_risk_factors = calculate_dynamic_risk_score( "direct", request_count, 'openshift-' in dep_namespace or 'kube-system' in dep_namespace, dep_port, request.change_type ) # Reduce risk score for infrastructure endpoints if is_infra_endpoint: risk_score = min(risk_score, 0.5) # Merge calculated risk factors with change-type risk factors all_risk_factors = list(set(risk_factors + calculated_risk_factors)) # Generate specific recommendation recommendation = generate_specific_recommendation( endpoint_kind, dep_namespace, impact, request.change_type, request_count ) affected_services.append(AffectedService( id=dep.get("id", str(uuid.uuid4())), name=dep_name, namespace=dep_namespace, kind=endpoint_kind, impact=impact, impact_category=impact_category, impact_description=impact_desc, dependency=DependencyType.DIRECT, recommendation=recommendation, connection_details={ "protocol": dep.get("protocol", "TCP"), "port": dep_port, "request_count": request_count, "last_seen": dep.get("last_seen") }, risk_score=risk_score, risk_factors=all_risk_factors, recovery_info={ "recovery_time": change_chars["recovery_time"], "reversible": change_chars["reversible"] } )) # Process indirect dependencies for dep in indirect_deps: dep_name = dep.get("name", "unknown") dep_namespace = dep.get("namespace", "default") dep_port = dep.get("port", 0) request_count = dep.get("request_count", 0) hop_distance = dep.get("hop_distance", 2) # Properly classify the endpoint kind original_kind = dep.get("kind", "Pod") endpoint_kind = classify_endpoint_kind(dep_name, dep_namespace, original_kind) # Use new change-type aware calculation impact, impact_category, impact_desc, risk_factors = service.calculate_impact_for_change_type( request.change_type, DependencyType.INDIRECT, request_count, False ) # Skip if this change type doesn't affect indirect dependencies if impact == ImpactLevel.NONE and not change_chars["affects_indirect"]: continue # Downgrade impact for infrastructure endpoints is_infra_endpoint = endpoint_kind in ['SDN-IP', 'ServiceIP'] or \ (endpoint_kind == 'ExternalIP' and dep_port > 32000) if is_infra_endpoint and impact in [ImpactLevel.HIGH, ImpactLevel.MEDIUM]: impact = ImpactLevel.LOW impact_desc = f"Infrastructure connectivity may be affected. {impact_desc}" # Calculate dynamic risk score (reduced for indirect) risk_score, calculated_risk_factors = calculate_dynamic_risk_score( "indirect", request_count, 'openshift-' in dep_namespace or 'kube-system' in dep_namespace, dep_port, request.change_type ) risk_score = risk_score * 0.7 # 30% reduction for indirect # Further reduce risk score for infrastructure endpoints if is_infra_endpoint: risk_score = min(risk_score, 0.3) # Merge calculated risk factors with change-type risk factors all_risk_factors = list(set(risk_factors + calculated_risk_factors)) # Generate specific recommendation recommendation = generate_specific_recommendation( endpoint_kind, dep_namespace, impact, request.change_type, request_count ) affected_services.append(AffectedService( id=dep.get("id", str(uuid.uuid4())), name=dep_name, namespace=dep_namespace, kind=endpoint_kind, impact=impact, impact_category=impact_category, impact_description=impact_desc, dependency=DependencyType.INDIRECT, recommendation=recommendation, connection_details={ "protocol": dep.get("protocol", "TCP"), "port": dep_port, "request_count": request_count, "hop_distance": hop_distance }, risk_score=round(risk_score, 2), risk_factors=all_risk_factors, recovery_info={ "recovery_time": change_chars["recovery_time"], "reversible": change_chars["reversible"] } )) # Deduplicate services (remove duplicate IPs that map to same service) affected_services_dicts = [ { "id": s.id, "name": s.name, "namespace": s.namespace, "kind": s.kind, "impact": s.impact, "impact_category": s.impact_category, "impact_description": s.impact_description, "dependency": s.dependency, "recommendation": s.recommendation, "connection_details": s.connection_details, "risk_score": s.risk_score, "risk_factors": s.risk_factors, "recovery_info": s.recovery_info } for s in affected_services ] deduped_dicts = deduplicate_services(affected_services_dicts) # Rebuild affected_services from deduplicated list affected_services = [ AffectedService( id=d["id"], name=d["name"], namespace=d["namespace"], kind=d["kind"], impact=d["impact"], impact_category=d["impact_category"], impact_description=d["impact_description"], dependency=d["dependency"], recommendation=d["recommendation"], connection_details=d["connection_details"], risk_score=d["risk_score"], risk_factors=d["risk_factors"], recovery_info=d["recovery_info"] ) for d in deduped_dicts ] # Calculate summary with change-type specific information high_count = sum(1 for s in affected_services if s.impact == ImpactLevel.HIGH) medium_count = sum(1 for s in affected_services if s.impact == ImpactLevel.MEDIUM) low_count = sum(1 for s in affected_services if s.impact == ImpactLevel.LOW) # Generate expected behavior description based on change type expected_behaviors = { ChangeType.DELETE: f"Service will be completely removed. {high_count} services will lose connectivity immediately.", ChangeType.SCALE_DOWN: f"Service will be unavailable until scaled back. {high_count} services will experience connection failures.", ChangeType.NETWORK_ISOLATE: f"Network traffic will be blocked. {high_count} direct connections will be severed.", ChangeType.RESOURCE_CHANGE: f"Performance may degrade. {medium_count} services may experience slower responses.", ChangeType.PORT_CHANGE: f"Port change will break existing connections. {high_count} clients need configuration update.", ChangeType.CONFIG_CHANGE: f"Behavior may change. {medium_count} services may need to adapt to new configuration.", ChangeType.IMAGE_UPDATE: f"Brief disruption during rollout. {medium_count} services may experience temporary errors.", ChangeType.NETWORK_POLICY_APPLY: f"Traffic filtering active. {high_count} connections may be blocked if not in allow list.", ChangeType.NETWORK_POLICY_REMOVE: f"Security policy removed. No connectivity impact, but security posture changed.", } summary = ImpactSummary( total_affected=len(affected_services), high_impact=high_count, medium_impact=medium_count, low_impact=low_count, blast_radius=len(direct_deps) + len(indirect_deps), confidence_score=min(1.0, graph_data.get("confidence", 0.8)), # Change-type specific fields primary_impact_category=change_chars["primary_impact"], impact_description=change_chars["description"], expected_behavior=expected_behaviors.get(request.change_type, "Review impact carefully."), recovery_time=change_chars["recovery_time"], is_reversible=change_chars["reversible"] ) # Determine no-dependency scenario if applicable no_dep_info = None if len(affected_services) == 0: graph_matches = graph_data.get("node_matches", 0) has_external = graph_data.get("has_external_connections", False) no_dep_info = service.determine_no_dependency_scenario( graph_matches, has_external, request.target_kind ) # Generate network policy suggestion for network-related changes network_policy_suggestion = None if request.change_type in [ChangeType.NETWORK_ISOLATE, ChangeType.NETWORK_POLICY_APPLY]: try: policy_request = NetworkPolicyGenerateRequest( cluster_id=request.cluster_id, analysis_id=request.analysis_id, target_namespace=request.target_namespace, target_workload=request.target_name, target_kind=request.target_kind, policy_types=[], include_dns=True, strict_mode=False ) network_policy_suggestion = await service.generate_network_policy(policy_request) except Exception as e: logger.warning("Failed to generate network policy suggestion", error=str(e)) # Build simulation details change_descriptions = { ChangeType.DELETE: "Complete removal of the resource from the cluster", ChangeType.SCALE_DOWN: "Scale deployment replicas to zero", ChangeType.NETWORK_ISOLATE: "Apply network policy to isolate the target", ChangeType.RESOURCE_CHANGE: "Modify CPU/Memory resource limits", ChangeType.PORT_CHANGE: "Change exposed service ports", ChangeType.CONFIG_CHANGE: "Modify ConfigMap/Secret/Environment variables", ChangeType.IMAGE_UPDATE: "Update container image version", ChangeType.NETWORK_POLICY_APPLY: "Apply a new network policy", ChangeType.NETWORK_POLICY_REMOVE: "Remove an existing network policy" } details = SimulationDetails( target_name=request.target_name, target_namespace=request.target_namespace, target_kind=request.target_kind, change_type=request.change_type.value, change_description=change_descriptions.get(request.change_type, "Unknown change type"), graph_matches=graph_data.get("node_matches", 0), simulation_timestamp=datetime.utcnow() ) # Build timeline projection based on change type characteristics timeline_descriptions = { ChangeType.DELETE: { "immediate": "Services lose connectivity immediately upon deletion", "short_term": "Dependent services may fail health checks and restart", "long_term": "System stabilizes after failover or manual intervention" }, ChangeType.SCALE_DOWN: { "immediate": "Existing connections drain, new connections fail", "short_term": "Queued requests may timeout, clients retry", "long_term": "System stable once scaled back up" }, ChangeType.NETWORK_ISOLATE: { "immediate": "Network traffic blocked, connections timeout", "short_term": "Health checks may fail, pods may restart", "long_term": "System adapts to new network topology" }, ChangeType.RESOURCE_CHANGE: { "immediate": "Pod may restart if limits reduced significantly", "short_term": "Response times may increase, throttling possible", "long_term": "System adapts to new resource allocation" }, ChangeType.PORT_CHANGE: { "immediate": "Connections to old port fail immediately", "short_term": "Clients with cached port info continue failing", "long_term": "Stable after all clients update configuration" }, ChangeType.CONFIG_CHANGE: { "immediate": "New config takes effect (may require pod restart)", "short_term": "Behavior changes propagate through system", "long_term": "System operates with new configuration" }, ChangeType.IMAGE_UPDATE: { "immediate": "Rolling update begins, brief connection disruption", "short_term": "Old and new versions may coexist during rollout", "long_term": "All pods running new version, system stable" }, ChangeType.NETWORK_POLICY_APPLY: { "immediate": "Policy enforced, non-matching traffic blocked", "short_term": "Blocked services may fail health checks", "long_term": "System operates within policy constraints" }, ChangeType.NETWORK_POLICY_REMOVE: { "immediate": "Policy removed, all traffic allowed", "short_term": "No operational impact expected", "long_term": "Security posture changed, monitor for anomalies" }, } change_timeline = timeline_descriptions.get(request.change_type, { "immediate": "Immediate effects upon change application", "short_term": "Short-term cascading effects", "long_term": "Long-term system adaptation" }) timeline_projection = { "immediate": { "description": change_timeline["immediate"], "affected_count": high_count, "expected_duration": "0-5 minutes", "impact_category": change_chars["primary_impact"].value if hasattr(change_chars["primary_impact"], 'value') else str(change_chars["primary_impact"]) }, "short_term": { "description": change_timeline["short_term"], "affected_count": medium_count, "expected_duration": "5-30 minutes", "secondary_impacts": [imp.value if hasattr(imp, 'value') else str(imp) for imp in change_chars["secondary_impacts"]] }, "long_term": { "description": change_timeline["long_term"], "affected_count": low_count, "expected_duration": "30+ minutes", "recovery_time": change_chars["recovery_time"] } } # Build rollback scenario based on change type rollback_steps = { ChangeType.DELETE: [ "Restore from backup or redeploy from manifest", "Verify all ConfigMaps and Secrets are present", "Wait for pods to become ready", "Verify connectivity from dependent services" ], ChangeType.SCALE_DOWN: [ "Scale deployment back to original replica count", "Wait for pods to become ready", "Verify load balancing is working", "Check for any queued request backlog" ], ChangeType.NETWORK_ISOLATE: [ "Remove or modify network policy", "Verify traffic flow is restored", "Check health checks are passing", "Monitor for connection timeouts" ], ChangeType.RESOURCE_CHANGE: [ "Revert resource limits to previous values", "Pod will restart with new limits", "Monitor resource utilization", "Verify performance is restored" ], ChangeType.PORT_CHANGE: [ "Revert port configuration", "Update service selector if needed", "Notify dependent teams of port reversion", "Verify client connections restored" ], ChangeType.CONFIG_CHANGE: [ "Revert ConfigMap/Secret to previous version", "Restart pods to pick up old config", "Verify application behavior", "Check for cached configuration issues" ], ChangeType.IMAGE_UPDATE: [ "Rollback deployment to previous revision", "kubectl rollout undo deployment/", "Wait for rollout to complete", "Verify application functionality" ], ChangeType.NETWORK_POLICY_APPLY: [ "Delete the applied network policy", "Verify traffic flow is restored", "Review security implications", "Update documentation" ], ChangeType.NETWORK_POLICY_REMOVE: [ "Reapply the network policy", "Verify policy is enforced", "Check no legitimate traffic is blocked", "Update security documentation" ], } rollback_risks = { ChangeType.DELETE: ["Data loss if not backed up", "State inconsistency", "Dependency ordering issues"], ChangeType.SCALE_DOWN: ["Request queue overflow", "Brief unavailability during scale-up"], ChangeType.NETWORK_ISOLATE: ["Security exposure during rollback"], ChangeType.RESOURCE_CHANGE: ["Pod restart required", "Brief unavailability"], ChangeType.PORT_CHANGE: ["Client configuration sync issues"], ChangeType.CONFIG_CHANGE: ["Config cache invalidation", "Restart required"], ChangeType.IMAGE_UPDATE: ["Data migration issues if schema changed"], ChangeType.NETWORK_POLICY_APPLY: ["Security exposure"], ChangeType.NETWORK_POLICY_REMOVE: ["Traffic may be blocked again"], } rollback_times = { ChangeType.DELETE: "15-60 minutes (requires redeployment)", ChangeType.SCALE_DOWN: "1-5 minutes", ChangeType.NETWORK_ISOLATE: "< 1 minute", ChangeType.RESOURCE_CHANGE: "1-5 minutes (pod restart)", ChangeType.PORT_CHANGE: "1-5 minutes + client updates", ChangeType.CONFIG_CHANGE: "1-5 minutes", ChangeType.IMAGE_UPDATE: "5-15 minutes", ChangeType.NETWORK_POLICY_APPLY: "< 1 minute", ChangeType.NETWORK_POLICY_REMOVE: "< 1 minute", } rollback_scenario = { "feasibility": "high" if change_chars["reversible"] else "medium", "estimated_time": rollback_times.get(request.change_type, "5-15 minutes"), "steps": rollback_steps.get(request.change_type, [ "Identify affected services from this simulation", "Prepare rollback configuration/manifest", "Execute rollback in reverse order", "Verify service health and connectivity" ]), "risks": rollback_risks.get(request.change_type, []), "reversible": change_chars["reversible"] } response = ImpactSimulationResponse( success=True, simulation_id=str(uuid.uuid4()), details=details, summary=summary, affected_services=affected_services, no_dependency_info=no_dep_info, network_policy_suggestion=network_policy_suggestion, timeline_projection=timeline_projection, rollback_scenario=rollback_scenario ) # ========== DEBUG: IMPACT SIMULATION RESULTS ========== affected_services_summary = [ { "name": f"{svc.namespace}/{svc.name}", "kind": svc.kind, "impact": svc.impact.value, "port": svc.connection_details.get('port', 0), "requests": svc.connection_details.get('request_count', 0), "risk": svc.risk_score } for svc in affected_services[:15] ] logger.warning("IMPACT_SIM_RESULTS: simulation completed", simulation_id=response.simulation_id, total_affected=summary.total_affected, high_impact=summary.high_impact, medium_impact=summary.medium_impact, low_impact=summary.low_impact, blast_radius=summary.blast_radius, affected_services=affected_services_summary, no_dependency_scenario=no_dep_info.scenario if no_dep_info else None ) # ========== DEBUG END ========== logger.info( "Impact simulation completed", simulation_id=response.simulation_id, total_affected=summary.total_affected, high_impact=summary.high_impact ) # Save simulation to history (non-blocking) try: import time start_time = time.time() await save_simulation_to_history( response, request, current_user.get('username', 'unknown'), duration_ms=int((time.time() - start_time) * 1000) ) except Exception as hist_err: logger.warning("Failed to save simulation to history", error=str(hist_err)) return response except Exception as e: logger.error("Impact simulation failed", error=str(e)) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to run impact simulation: {str(e)}" ) # ============================================================================= # Export Endpoints # ============================================================================= @router.post( "/impact/export/json", summary="Export Impact Simulation to JSON", description="Export impact simulation results to JSON format." ) async def export_impact_simulation_json( request: ImpactSimulationRequest, cluster_name: Optional[str] = Query(None, description="Cluster name for report"), current_user: dict = Depends(get_current_user), service: NetworkPolicyService = Depends(get_service) ): """Export impact simulation results to JSON""" try: # Run simulation first simulation_response = await run_impact_simulation(request, current_user, service) # Build export report export_report = ImpactSimulationExportReport( metadata=ExportMetadata( generated_at=datetime.utcnow(), analysis_id=request.analysis_id, cluster_id=request.cluster_id, cluster_name=cluster_name, export_format="json" ), simulation=SimulationExportData( target_name=request.target_name, target_namespace=request.target_namespace, target_kind=request.target_kind, change_type=request.change_type.value, graph_matches=simulation_response.details.graph_matches ), impact_summary=simulation_response.summary, affected_services=simulation_response.affected_services, network_policy_suggestion=simulation_response.network_policy_suggestion.model_dump() if simulation_response.network_policy_suggestion else None, recommendations=[s.recommendation for s in simulation_response.affected_services] ) # Convert to JSON output = json.dumps(export_report.model_dump(), indent=2, default=str) # Generate filename timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S") target_safe = request.target_name.replace("/", "-").replace(":", "-") filename = f"impact_simulation_{target_safe}_{timestamp}.json" logger.info("Impact simulation exported to JSON", filename=filename) return StreamingResponse( iter([output]), media_type="application/json", headers={ "Content-Disposition": f"attachment; filename={filename}" } ) except HTTPException: raise except Exception as e: logger.error("Impact simulation JSON export failed", error=str(e)) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Export failed: {str(e)}" ) @router.post( "/impact/export/csv", summary="Export Impact Simulation to CSV", description="Export impact simulation results to CSV format." ) async def export_impact_simulation_csv( request: ImpactSimulationRequest, cluster_name: Optional[str] = Query(None, description="Cluster name for report"), current_user: dict = Depends(get_current_user), service: NetworkPolicyService = Depends(get_service) ): """Export impact simulation results to CSV""" try: # Run simulation first simulation_response = await run_impact_simulation(request, current_user, service) # Build CSV output = io.StringIO() writer = csv.writer(output) # Metadata section writer.writerow(["# Impact Simulation Report"]) writer.writerow(["Generated At", datetime.utcnow().isoformat()]) writer.writerow(["Cluster ID", request.cluster_id]) writer.writerow(["Cluster Name", cluster_name or ""]) writer.writerow(["Analysis ID", request.analysis_id or ""]) writer.writerow([]) # Target section writer.writerow(["# Simulation Target"]) writer.writerow(["Target Name", request.target_name]) writer.writerow(["Target Namespace", request.target_namespace]) writer.writerow(["Target Kind", request.target_kind]) writer.writerow(["Change Type", request.change_type.value]) writer.writerow(["Graph Matches", simulation_response.details.graph_matches]) writer.writerow([]) # Summary section writer.writerow(["# Impact Summary"]) writer.writerow(["Total Affected", simulation_response.summary.total_affected]) writer.writerow(["High Impact", simulation_response.summary.high_impact]) writer.writerow(["Medium Impact", simulation_response.summary.medium_impact]) writer.writerow(["Low Impact", simulation_response.summary.low_impact]) writer.writerow(["Blast Radius", simulation_response.summary.blast_radius]) writer.writerow(["Confidence Score", simulation_response.summary.confidence_score]) writer.writerow([]) # Affected services section writer.writerow(["# Affected Services"]) writer.writerow([ "Name", "Namespace", "Kind", "Impact", "Dependency Type", "Protocol", "Port", "Request Count", "Risk Score", "Recommendation" ]) for svc in simulation_response.affected_services: conn = svc.connection_details or {} writer.writerow([ svc.name, svc.namespace, svc.kind, svc.impact.value, svc.dependency.value, conn.get("protocol", ""), conn.get("port", ""), conn.get("request_count", ""), svc.risk_score, svc.recommendation ]) output.seek(0) # Generate filename timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S") target_safe = request.target_name.replace("/", "-").replace(":", "-") filename = f"impact_simulation_{target_safe}_{timestamp}.csv" logger.info("Impact simulation exported to CSV", filename=filename) return StreamingResponse( iter([output.getvalue()]), media_type="text/csv", headers={ "Content-Disposition": f"attachment; filename={filename}" } ) except HTTPException: raise except Exception as e: logger.error("Impact simulation CSV export failed", error=str(e)) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Export failed: {str(e)}" ) # ============================================================================= # Utility Endpoints # ============================================================================= @router.get( "/change-types", summary="Get Available Change Types", description="Get list of available change types for simulation." ) async def get_change_types( current_user: dict = Depends(get_current_user) ): """Get available simulation change types""" return { "change_types": [ { "key": "delete", "label": "Delete / Remove", "description": "Completely remove the target from cluster", "icon": "delete", "category": "destructive" }, { "key": "scale_down", "label": "Scale Down (replicas: 0)", "description": "Scale deployment to zero replicas", "icon": "arrow-down", "category": "scaling" }, { "key": "network_isolate", "label": "Network Isolation", "description": "Apply network policy to isolate target", "icon": "lock", "category": "network" }, { "key": "resource_change", "label": "Resource Limit Change", "description": "Modify CPU/Memory limits", "icon": "dashboard", "category": "resource" }, { "key": "port_change", "label": "Port Change", "description": "Change exposed ports", "icon": "api", "category": "network" }, { "key": "config_change", "label": "Configuration Change", "description": "Modify ConfigMap/Secret/Environment", "icon": "setting", "category": "configuration" }, { "key": "image_update", "label": "Image Update", "description": "Update container image version", "icon": "cloud-upload", "category": "deployment" }, { "key": "network_policy_apply", "label": "Apply Network Policy", "description": "Simulate applying a new network policy", "icon": "safety", "category": "network", "advanced": True }, { "key": "network_policy_remove", "label": "Remove Network Policy", "description": "Simulate removing an existing network policy", "icon": "unlock", "category": "network", "advanced": True } ] } # ============================================================================= # Scheduled Simulations Endpoints # ============================================================================= from pydantic import BaseModel, Field from typing import List, Optional as OptionalType from datetime import datetime as dt class ScheduledSimulationCreate(BaseModel): """Request to create a scheduled simulation""" name: str = Field(..., description="Name of the scheduled simulation") description: OptionalType[str] = None cluster_id: str analysis_id: OptionalType[str] = None target_name: str target_namespace: str target_kind: str = "Deployment" change_type: str schedule_type: str = Field("once", description="once, daily, weekly") scheduled_time: str = Field(..., description="ISO format datetime") notify_before_minutes: int = 15 auto_rollback: bool = False rollback_on_failure: bool = True class ScheduledSimulationResponse(BaseModel): """Response for a scheduled simulation""" id: str name: str description: OptionalType[str] = None cluster_id: str analysis_id: OptionalType[str] = None target_name: str target_namespace: str target_kind: str change_type: str schedule_type: str scheduled_time: str notify_before_minutes: int auto_rollback: bool rollback_on_failure: bool status: str = "scheduled" created_at: str created_by: OptionalType[str] = None last_run_at: OptionalType[str] = None last_run_result: OptionalType[str] = None class ScheduledSimulationListResponse(BaseModel): """Response for listing scheduled simulations""" simulations: List[ScheduledSimulationResponse] total: int @router.get( "/scheduled", response_model=ScheduledSimulationListResponse, summary="List Scheduled Simulations", description="Get all scheduled simulations for the current user" ) async def list_scheduled_simulations( cluster_id: OptionalType[str] = Query(None, description="Filter by cluster"), status: OptionalType[str] = Query(None, description="Filter by status"), current_user: dict = Depends(get_current_user) ): """List all scheduled simulations""" try: from database.postgresql import database # Check if table exists check_query = """ SELECT EXISTS ( SELECT FROM information_schema.tables WHERE table_name = 'scheduled_simulations' ) as exists """ exists = await database.fetch_one(check_query) if not exists or not exists.get('exists'): # Create table if it doesn't exist create_table = """ CREATE TABLE IF NOT EXISTS scheduled_simulations ( id VARCHAR(50) PRIMARY KEY, name VARCHAR(255) NOT NULL, description TEXT, cluster_id VARCHAR(50) NOT NULL, analysis_id VARCHAR(50), target_name VARCHAR(255) NOT NULL, target_namespace VARCHAR(255) NOT NULL, target_kind VARCHAR(50) DEFAULT 'Deployment', change_type VARCHAR(50) NOT NULL, schedule_type VARCHAR(20) DEFAULT 'once', scheduled_time TIMESTAMP NOT NULL, notify_before_minutes INT DEFAULT 15, auto_rollback BOOLEAN DEFAULT FALSE, rollback_on_failure BOOLEAN DEFAULT TRUE, status VARCHAR(20) DEFAULT 'scheduled', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, created_by VARCHAR(100), last_run_at TIMESTAMP, last_run_result TEXT ) """ await database.execute(create_table) return ScheduledSimulationListResponse(simulations=[], total=0) # Build query query = """ SELECT * FROM scheduled_simulations WHERE 1=1 """ params = {} if cluster_id: query += " AND cluster_id = :cluster_id" params['cluster_id'] = cluster_id if status: query += " AND status = :status" params['status'] = status query += " ORDER BY scheduled_time ASC" rows = await database.fetch_all(query, params) simulations = [ ScheduledSimulationResponse( id=row['id'], name=row['name'], description=row['description'], cluster_id=row['cluster_id'], analysis_id=row['analysis_id'], target_name=row['target_name'], target_namespace=row['target_namespace'], target_kind=row['target_kind'], change_type=row['change_type'], schedule_type=row['schedule_type'], scheduled_time=row['scheduled_time'].isoformat() if row['scheduled_time'] else '', notify_before_minutes=row['notify_before_minutes'] or 15, auto_rollback=row['auto_rollback'] or False, rollback_on_failure=row['rollback_on_failure'] if row['rollback_on_failure'] is not None else True, status=row['status'], created_at=row['created_at'].isoformat() if row['created_at'] else '', created_by=row['created_by'], last_run_at=row['last_run_at'].isoformat() if row['last_run_at'] else None, last_run_result=row['last_run_result'] ) for row in rows ] return ScheduledSimulationListResponse(simulations=simulations, total=len(simulations)) except Exception as e: logger.error("Failed to list scheduled simulations", error=str(e)) return ScheduledSimulationListResponse(simulations=[], total=0) @router.post( "/scheduled", response_model=ScheduledSimulationResponse, summary="Create Scheduled Simulation", description="Create a new scheduled simulation" ) async def create_scheduled_simulation( request: ScheduledSimulationCreate, current_user: dict = Depends(get_current_user) ): """Create a new scheduled simulation""" try: from database.postgresql import database # Ensure table exists create_table = """ CREATE TABLE IF NOT EXISTS scheduled_simulations ( id VARCHAR(50) PRIMARY KEY, name VARCHAR(255) NOT NULL, description TEXT, cluster_id VARCHAR(50) NOT NULL, analysis_id VARCHAR(50), target_name VARCHAR(255) NOT NULL, target_namespace VARCHAR(255) NOT NULL, target_kind VARCHAR(50) DEFAULT 'Deployment', change_type VARCHAR(50) NOT NULL, schedule_type VARCHAR(20) DEFAULT 'once', scheduled_time TIMESTAMP NOT NULL, notify_before_minutes INT DEFAULT 15, auto_rollback BOOLEAN DEFAULT FALSE, rollback_on_failure BOOLEAN DEFAULT TRUE, status VARCHAR(20) DEFAULT 'scheduled', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, created_by VARCHAR(100), last_run_at TIMESTAMP, last_run_result TEXT ) """ await database.execute(create_table) simulation_id = f"sched-{str(uuid.uuid4())[:8]}" insert_query = """ INSERT INTO scheduled_simulations ( id, name, description, cluster_id, analysis_id, target_name, target_namespace, target_kind, change_type, schedule_type, scheduled_time, notify_before_minutes, auto_rollback, rollback_on_failure, status, created_by ) VALUES ( :id, :name, :description, :cluster_id, :analysis_id, :target_name, :target_namespace, :target_kind, :change_type, :schedule_type, :scheduled_time, :notify_before_minutes, :auto_rollback, :rollback_on_failure, 'scheduled', :created_by ) RETURNING * """ # Parse scheduled_time from ISO string to datetime (naive, without timezone) from dateutil.parser import parse as parse_datetime from datetime import timezone try: scheduled_time_dt = parse_datetime(request.scheduled_time) # Convert timezone-aware datetime to UTC, then make naive for PostgreSQL TIMESTAMP column if scheduled_time_dt.tzinfo is not None: # Convert to UTC first to preserve the actual moment in time scheduled_time_dt = scheduled_time_dt.astimezone(timezone.utc).replace(tzinfo=None) except Exception as e: logger.error("Invalid scheduled_time format", error=str(e), scheduled_time=request.scheduled_time) raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Invalid scheduled_time format: {request.scheduled_time}" ) row = await database.fetch_one(insert_query, { 'id': simulation_id, 'name': request.name, 'description': request.description, 'cluster_id': request.cluster_id, 'analysis_id': request.analysis_id, 'target_name': request.target_name, 'target_namespace': request.target_namespace, 'target_kind': request.target_kind, 'change_type': request.change_type, 'schedule_type': request.schedule_type, 'scheduled_time': scheduled_time_dt, 'notify_before_minutes': request.notify_before_minutes, 'auto_rollback': request.auto_rollback, 'rollback_on_failure': request.rollback_on_failure, 'created_by': current_user.get('username', 'unknown') }) logger.info( "Scheduled simulation created", simulation_id=simulation_id, name=request.name, user=current_user.get('username') ) return ScheduledSimulationResponse( id=row['id'], name=row['name'], description=row['description'], cluster_id=row['cluster_id'], analysis_id=row['analysis_id'], target_name=row['target_name'], target_namespace=row['target_namespace'], target_kind=row['target_kind'], change_type=row['change_type'], schedule_type=row['schedule_type'], scheduled_time=row['scheduled_time'].isoformat() if row['scheduled_time'] else '', notify_before_minutes=row['notify_before_minutes'] or 15, auto_rollback=row['auto_rollback'] or False, rollback_on_failure=row['rollback_on_failure'] if row['rollback_on_failure'] is not None else True, status=row['status'], created_at=row['created_at'].isoformat() if row['created_at'] else '', created_by=row['created_by'] ) except Exception as e: logger.error("Failed to create scheduled simulation", error=str(e)) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to create scheduled simulation: {str(e)}" ) @router.delete( "/scheduled/{simulation_id}", summary="Cancel Scheduled Simulation", description="Cancel a scheduled simulation" ) async def cancel_scheduled_simulation( simulation_id: str, current_user: dict = Depends(get_current_user) ): """Cancel a scheduled simulation""" try: from database.postgresql import database # Check if simulation exists query = "SELECT * FROM scheduled_simulations WHERE id = :id" simulation = await database.fetch_one(query, {'id': simulation_id}) if not simulation: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Scheduled simulation not found" ) # Delete the simulation delete_query = "DELETE FROM scheduled_simulations WHERE id = :id" await database.execute(delete_query, {'id': simulation_id}) logger.info( "Scheduled simulation cancelled", simulation_id=simulation_id, user=current_user.get('username') ) return {"message": "Scheduled simulation cancelled", "id": simulation_id} except HTTPException: raise except Exception as e: logger.error("Failed to cancel scheduled simulation", error=str(e)) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to cancel scheduled simulation: {str(e)}" ) @router.post( "/scheduled/{simulation_id}/run", summary="Run Scheduled Simulation Now", description="Execute a scheduled simulation immediately" ) async def run_scheduled_simulation_now( simulation_id: str, current_user: dict = Depends(get_current_user), service: NetworkPolicyService = Depends(get_service) ): """Run a scheduled simulation immediately""" try: from database.postgresql import database # Get simulation details query = "SELECT * FROM scheduled_simulations WHERE id = :id" simulation = await database.fetch_one(query, {'id': simulation_id}) if not simulation: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Scheduled simulation not found" ) # Run the impact simulation # Construct target_id from stored fields (format: kind-namespace-name or just target_name) target_id = simulation['target_name'] request = ImpactSimulationRequest( cluster_id=simulation['cluster_id'], analysis_id=simulation['analysis_id'], target_id=target_id, target_name=simulation['target_name'], target_namespace=simulation['target_namespace'], target_kind=simulation['target_kind'], change_type=ChangeType(simulation['change_type']) ) result = await run_impact_simulation(request, current_user, service) # Update simulation status update_query = """ UPDATE scheduled_simulations SET last_run_at = NOW(), last_run_result = :result, status = CASE WHEN schedule_type = 'once' THEN 'completed' ELSE status END WHERE id = :id """ await database.execute(update_query, { 'id': simulation_id, 'result': 'success' if result.success else 'failed' }) logger.info( "Scheduled simulation executed", simulation_id=simulation_id, user=current_user.get('username') ) return { "message": "Simulation executed", "id": simulation_id, "result": result } except HTTPException: raise except Exception as e: logger.error("Failed to run scheduled simulation", error=str(e)) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to run scheduled simulation: {str(e)}" ) @router.get( "/scheduled/worker/status", summary="Get Scheduler Worker Status", description="Get the status of the scheduled simulation worker" ) async def get_scheduler_status( current_user: dict = Depends(get_current_user) ): """Get the status of the scheduled simulation worker""" try: from workers.scheduled_simulation_worker import scheduled_simulation_worker status_info = await scheduled_simulation_worker.get_status() return { "worker": "scheduled_simulation_worker", **status_info } except Exception as e: logger.error("Failed to get scheduler status", error=str(e)) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to get scheduler status: {str(e)}" ) # ============================================================================= # Simulation History Endpoints # ============================================================================= class SimulationHistoryResponse(BaseModel): """Response for a simulation history entry""" id: str simulation_id: str cluster_id: str analysis_id: OptionalType[str] = None target_name: str target_namespace: str target_kind: str change_type: str total_affected: int high_impact: int medium_impact: int low_impact: int blast_radius: int confidence_score: float status: str created_at: str created_by: OptionalType[str] = None duration_ms: OptionalType[int] = None result_summary: OptionalType[dict] = None class SimulationHistoryListResponse(BaseModel): """Response for listing simulation history""" history: List[SimulationHistoryResponse] total: int async def save_simulation_to_history( simulation_response: ImpactSimulationResponse, request: ImpactSimulationRequest, user: str, duration_ms: int = 0 ): """Save a simulation result to history""" try: from database.postgresql import database # Ensure table exists create_table = """ CREATE TABLE IF NOT EXISTS simulation_history ( id VARCHAR(50) PRIMARY KEY, simulation_id VARCHAR(50) NOT NULL, cluster_id VARCHAR(50) NOT NULL, analysis_id VARCHAR(50), target_name VARCHAR(255) NOT NULL, target_namespace VARCHAR(255) NOT NULL, target_kind VARCHAR(50) DEFAULT 'Deployment', change_type VARCHAR(50) NOT NULL, total_affected INT DEFAULT 0, high_impact INT DEFAULT 0, medium_impact INT DEFAULT 0, low_impact INT DEFAULT 0, blast_radius INT DEFAULT 0, confidence_score DECIMAL(5,4) DEFAULT 0, status VARCHAR(20) DEFAULT 'completed', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, created_by VARCHAR(100), duration_ms INT, result_summary JSONB ) """ await database.execute(create_table) # Create index if not exists try: await database.execute(""" CREATE INDEX IF NOT EXISTS idx_simulation_history_created ON simulation_history(created_at DESC) """) await database.execute(""" CREATE INDEX IF NOT EXISTS idx_simulation_history_cluster ON simulation_history(cluster_id) """) except: pass # Index might already exist history_id = f"hist-{str(uuid.uuid4())[:8]}" # Prepare result summary result_summary = { "affected_services": [ { "name": s.name, "namespace": s.namespace, "kind": s.kind, "impact": s.impact.value if hasattr(s.impact, 'value') else s.impact, "dependency": s.dependency.value if hasattr(s.dependency, 'value') else s.dependency, "risk_score": s.risk_score } for s in simulation_response.affected_services[:20] # Limit to first 20 ], "timeline_projection": simulation_response.timeline_projection, "rollback_scenario": simulation_response.rollback_scenario } insert_query = """ INSERT INTO simulation_history ( id, simulation_id, cluster_id, analysis_id, target_name, target_namespace, target_kind, change_type, total_affected, high_impact, medium_impact, low_impact, blast_radius, confidence_score, status, created_by, duration_ms, result_summary ) VALUES ( :id, :simulation_id, :cluster_id, :analysis_id, :target_name, :target_namespace, :target_kind, :change_type, :total_affected, :high_impact, :medium_impact, :low_impact, :blast_radius, :confidence_score, :status, :created_by, :duration_ms, :result_summary ) """ await database.execute(insert_query, { 'id': history_id, 'simulation_id': simulation_response.simulation_id, 'cluster_id': str(request.cluster_id), 'analysis_id': str(request.analysis_id) if request.analysis_id else None, 'target_name': request.target_name, 'target_namespace': request.target_namespace, 'target_kind': request.target_kind, 'change_type': request.change_type.value if hasattr(request.change_type, 'value') else request.change_type, 'total_affected': simulation_response.summary.total_affected, 'high_impact': simulation_response.summary.high_impact, 'medium_impact': simulation_response.summary.medium_impact, 'low_impact': simulation_response.summary.low_impact, 'blast_radius': simulation_response.summary.blast_radius, 'confidence_score': simulation_response.summary.confidence_score, 'status': 'completed' if simulation_response.success else 'failed', 'created_by': user, 'duration_ms': duration_ms, 'result_summary': json.dumps(result_summary, default=str) }) logger.info( "Simulation saved to history", history_id=history_id, simulation_id=simulation_response.simulation_id ) except Exception as e: logger.warning("Failed to save simulation to history", error=str(e)) # Don't raise - history save failure shouldn't fail the simulation @router.get( "/history", response_model=SimulationHistoryListResponse, summary="List Simulation History", description="Get simulation history for the current user" ) async def list_simulation_history( cluster_id: OptionalType[str] = Query(None, description="Filter by cluster"), analysis_id: OptionalType[str] = Query(None, description="Filter by analysis"), limit: int = Query(50, description="Maximum number of results"), offset: int = Query(0, description="Offset for pagination"), current_user: dict = Depends(get_current_user) ): """List simulation history""" try: from database.postgresql import database # Check if table exists check_query = """ SELECT EXISTS ( SELECT FROM information_schema.tables WHERE table_name = 'simulation_history' ) as exists """ exists = await database.fetch_one(check_query) if not exists or not exists.get('exists'): # Create table if it doesn't exist create_table = """ CREATE TABLE IF NOT EXISTS simulation_history ( id VARCHAR(50) PRIMARY KEY, simulation_id VARCHAR(50) NOT NULL, cluster_id VARCHAR(50) NOT NULL, analysis_id VARCHAR(50), target_name VARCHAR(255) NOT NULL, target_namespace VARCHAR(255) NOT NULL, target_kind VARCHAR(50) DEFAULT 'Deployment', change_type VARCHAR(50) NOT NULL, total_affected INT DEFAULT 0, high_impact INT DEFAULT 0, medium_impact INT DEFAULT 0, low_impact INT DEFAULT 0, blast_radius INT DEFAULT 0, confidence_score DECIMAL(5,4) DEFAULT 0, status VARCHAR(20) DEFAULT 'completed', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, created_by VARCHAR(100), duration_ms INT, result_summary JSONB ) """ await database.execute(create_table) return SimulationHistoryListResponse(history=[], total=0) # Build query query = """ SELECT * FROM simulation_history WHERE 1=1 """ count_query = """ SELECT COUNT(*) as total FROM simulation_history WHERE 1=1 """ params = {} if cluster_id: query += " AND cluster_id = :cluster_id" count_query += " AND cluster_id = :cluster_id" params['cluster_id'] = cluster_id if analysis_id: query += " AND analysis_id = :analysis_id" count_query += " AND analysis_id = :analysis_id" params['analysis_id'] = analysis_id query += " ORDER BY created_at DESC LIMIT :limit OFFSET :offset" params['limit'] = limit params['offset'] = offset rows = await database.fetch_all(query, params) total_row = await database.fetch_one(count_query, {k: v for k, v in params.items() if k not in ['limit', 'offset']}) total = total_row.get('total', 0) if total_row else 0 history = [] for row in rows: result_summary = None if row['result_summary']: try: result_summary = json.loads(row['result_summary']) if isinstance(row['result_summary'], str) else row['result_summary'] except: result_summary = None history.append(SimulationHistoryResponse( id=row['id'], simulation_id=row['simulation_id'], cluster_id=row['cluster_id'], analysis_id=row['analysis_id'], target_name=row['target_name'], target_namespace=row['target_namespace'], target_kind=row['target_kind'], change_type=row['change_type'], total_affected=row['total_affected'] or 0, high_impact=row['high_impact'] or 0, medium_impact=row['medium_impact'] or 0, low_impact=row['low_impact'] or 0, blast_radius=row['blast_radius'] or 0, confidence_score=float(row['confidence_score']) if row['confidence_score'] else 0.0, status=row['status'], created_at=row['created_at'].isoformat() if row['created_at'] else '', created_by=row['created_by'], duration_ms=row['duration_ms'], result_summary=result_summary )) return SimulationHistoryListResponse(history=history, total=total) except Exception as e: logger.error("Failed to list simulation history", error=str(e)) return SimulationHistoryListResponse(history=[], total=0) @router.get( "/history/{history_id}", response_model=SimulationHistoryResponse, summary="Get Simulation History Entry", description="Get a specific simulation history entry" ) async def get_simulation_history_entry( history_id: str, current_user: dict = Depends(get_current_user) ): """Get a specific simulation history entry""" try: from database.postgresql import database query = "SELECT * FROM simulation_history WHERE id = :id" row = await database.fetch_one(query, {'id': history_id}) if not row: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Simulation history entry not found" ) result_summary = None if row['result_summary']: try: result_summary = json.loads(row['result_summary']) if isinstance(row['result_summary'], str) else row['result_summary'] except: result_summary = None return SimulationHistoryResponse( id=row['id'], simulation_id=row['simulation_id'], cluster_id=row['cluster_id'], analysis_id=row['analysis_id'], target_name=row['target_name'], target_namespace=row['target_namespace'], target_kind=row['target_kind'], change_type=row['change_type'], total_affected=row['total_affected'] or 0, high_impact=row['high_impact'] or 0, medium_impact=row['medium_impact'] or 0, low_impact=row['low_impact'] or 0, blast_radius=row['blast_radius'] or 0, confidence_score=float(row['confidence_score']) if row['confidence_score'] else 0.0, status=row['status'], created_at=row['created_at'].isoformat() if row['created_at'] else '', created_by=row['created_by'], duration_ms=row['duration_ms'], result_summary=result_summary ) except HTTPException: raise except Exception as e: logger.error("Failed to get simulation history entry", error=str(e)) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to get simulation history entry: {str(e)}" ) @router.delete( "/history/{history_id}", summary="Delete Simulation History Entry", description="Delete a simulation history entry" ) async def delete_simulation_history_entry( history_id: str, current_user: dict = Depends(get_current_user) ): """Delete a simulation history entry""" try: from database.postgresql import database # Check if entry exists query = "SELECT * FROM simulation_history WHERE id = :id" entry = await database.fetch_one(query, {'id': history_id}) if not entry: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Simulation history entry not found" ) # Delete the entry delete_query = "DELETE FROM simulation_history WHERE id = :id" await database.execute(delete_query, {'id': history_id}) logger.info( "Simulation history entry deleted", history_id=history_id, user=current_user.get('username') ) return {"message": "History entry deleted", "id": history_id} except HTTPException: raise except Exception as e: logger.error("Failed to delete simulation history entry", error=str(e)) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to delete simulation history entry: {str(e)}" )