jester1177/cloud-native-debug-env

1"""Kubernetes manifest validator and simulator — deterministic, rule-based."""2 3import re4from typing import Any, Dict, List, Optional5 6import yaml7 8from server.models import FileContent9 10 11# Valid top-level K8s resource kinds we recognise12VALID_KINDS = {13    "Deployment", "StatefulSet", "DaemonSet", "ReplicaSet",14    "Pod", "Service", "Ingress", "ConfigMap", "Secret",15    "PersistentVolumeClaim", "PersistentVolume",16    "Job", "CronJob", "Namespace", "ServiceAccount",17    "Role", "RoleBinding", "ClusterRole", "ClusterRoleBinding",18    "HorizontalPodAutoscaler", "NetworkPolicy",19}20 21VALID_API_VERSIONS = {22    "v1", "apps/v1", "batch/v1", "networking.k8s.io/v1",23    "rbac.authorization.k8s.io/v1", "autoscaling/v2",24    "autoscaling/v1", "policy/v1",25}26 27 28def _parse_memory(mem_str: str) -> int:29    """Parse K8s memory string to bytes."""30    mem_str = str(mem_str).strip()31    multipliers = {32        "Ki": 1024, "Mi": 1024**2, "Gi": 1024**3, "Ti": 1024**4,33        "K": 1000, "M": 1000**2, "G": 1000**3, "T": 1000**4,34    }35    for suffix, mult in multipliers.items():36        if mem_str.endswith(suffix):37            return int(mem_str[:-len(suffix)]) * mult38    if mem_str.isdigit():39        return int(mem_str)40    return 041 42 43class KubernetesSimulator:44    """Simulates kubectl apply / kubectl get output.45 46    Validates K8s manifests without a real cluster.47    """48 49    def validate(self, manifests: Dict[str, FileContent]) -> Dict[str, Any]:50        """Validate all Kubernetes manifests in the file set.51 52        Returns dict with keys:53            valid: bool54            errors: list of error strings55            pod_status: simulated pod status56            service_status: simulated service endpoint status57        """58        k8s_files: Dict[str, Any] = {}59        errors: List[str] = []60 61        # Parse all K8s YAML files62        for path, fc in manifests.items():63            if fc.file_type.value != "kubernetes":64                continue65            try:66                docs = list(yaml.safe_load_all(fc.content))67                for doc in docs:68                    if doc and isinstance(doc, dict):69                        k8s_files[path] = doc70            except yaml.YAMLError as exc:71                errors.append(f"YAML parse error in {path}: {exc}")72 73        if not k8s_files and not errors:74            return {"valid": True, "errors": [], "pod_status": "N/A", "service_status": "N/A"}75 76        if errors:77            return {"valid": False, "errors": errors, "pod_status": "Error", "service_status": "Error"}78 79        # Validate each manifest80        all_resources: List[Dict[str, Any]] = []81        for path, doc in k8s_files.items():82            resource_errors = self._validate_resource(path, doc)83            errors.extend(resource_errors)84            all_resources.append({"path": path, "doc": doc})85 86        # Cross-resource validation87        cross_errors = self._validate_cross_resources(all_resources)88        errors.extend(cross_errors)89 90        # Simulate pod status91        pod_status = self._simulate_pod_status(all_resources)92        service_status = self._simulate_service_status(all_resources)93 94        return {95            "valid": len(errors) == 0,96            "errors": errors,97            "pod_status": pod_status,98            "service_status": service_status,99        }100 101    def _validate_resource(self, path: str, doc: Dict[str, Any]) -> List[str]:102        """Validate a single K8s resource document."""103        errors: List[str] = []104 105        kind = doc.get("kind", "")106        api_version = doc.get("apiVersion", "")107 108        if not kind:109            errors.append(f"{path}: missing 'kind' field")110        elif kind not in VALID_KINDS:111            errors.append(f"{path}: unknown kind '{kind}'")112 113        if not api_version:114            errors.append(f"{path}: missing 'apiVersion' field")115        elif api_version not in VALID_API_VERSIONS:116            errors.append(f"{path}: unknown apiVersion '{api_version}'")117 118        metadata = doc.get("metadata", {})119        if not isinstance(metadata, dict) or not metadata.get("name"):120            errors.append(f"{path}: metadata.name is required")121 122        # Kind-specific validation123        if kind == "Deployment":124            errors.extend(self._validate_deployment(path, doc))125        elif kind == "Service":126            errors.extend(self._validate_service(path, doc))127        elif kind == "Ingress":128            errors.extend(self._validate_ingress(path, doc))129 130        return errors131 132    def _validate_deployment(self, path: str, doc: Dict[str, Any]) -> List[str]:133        errors: List[str] = []134        spec = doc.get("spec", {})135        if not isinstance(spec, dict):136            errors.append(f"{path}: Deployment spec must be a mapping")137            return errors138 139        selector = spec.get("selector", {})140        template = spec.get("template", {})141 142        if not selector or not selector.get("matchLabels"):143            errors.append(f"{path}: Deployment must have spec.selector.matchLabels")144            return errors145 146        tmpl_labels = template.get("metadata", {}).get("labels", {})147        sel_labels = selector.get("matchLabels", {})148 149        # selector must match template labels150        for k, v in sel_labels.items():151            if tmpl_labels.get(k) != v:152                errors.append(153                    f"{path}: selector matchLabels ({k}={v}) does not match template labels"154                )155 156        # Validate containers157        containers = template.get("spec", {}).get("containers", [])158        if not containers:159            errors.append(f"{path}: Deployment must have at least one container")160 161        for c in containers:162            if not c.get("image"):163                errors.append(f"{path}: container '{c.get('name', '?')}' missing image")164 165        return errors166 167    def _validate_service(self, path: str, doc: Dict[str, Any]) -> List[str]:168        errors: List[str] = []169        spec = doc.get("spec", {})170        if not isinstance(spec, dict):171            errors.append(f"{path}: Service spec must be a mapping")172            return errors173 174        if not spec.get("selector"):175            errors.append(f"{path}: Service must have spec.selector")176 177        ports = spec.get("ports", [])178        if not ports:179            errors.append(f"{path}: Service must define at least one port")180 181        for p in ports:182            if not p.get("port"):183                errors.append(f"{path}: Service port entry missing 'port' field")184 185        return errors186 187    def _validate_ingress(self, path: str, doc: Dict[str, Any]) -> List[str]:188        errors: List[str] = []189        spec = doc.get("spec", {})190        rules = spec.get("rules", [])191        if not rules:192            errors.append(f"{path}: Ingress must define at least one rule")193        return errors194 195    def _validate_cross_resources(self, resources: List[Dict[str, Any]]) -> List[str]:196        """Validate cross-resource dependencies (e.g. Service selector matches Deployment labels)."""197        errors: List[str] = []198 199        # Collect all pod labels from Deployments/StatefulSets200        pod_labels_by_name: Dict[str, Dict[str, str]] = {}201        for r in resources:202            doc = r["doc"]203            kind = doc.get("kind", "")204            if kind in ("Deployment", "StatefulSet", "DaemonSet"):205                tmpl = doc.get("spec", {}).get("template", {})206                labels = tmpl.get("metadata", {}).get("labels", {})207                name = doc.get("metadata", {}).get("name", "?")208                pod_labels_by_name[name] = labels209 210        # Check Service selectors match some pod labels211        for r in resources:212            doc = r["doc"]213            if doc.get("kind") != "Service":214                continue215            svc_name = doc.get("metadata", {}).get("name", "?")216            selector = doc.get("spec", {}).get("selector", {})217            if not selector:218                continue219 220            matched = False221            for dep_name, labels in pod_labels_by_name.items():222                if all(labels.get(k) == v for k, v in selector.items()):223                    matched = True224                    break225            if not matched and pod_labels_by_name:226                errors.append(227                    f"Service '{svc_name}' selector {selector} does not match any pod labels"228                )229 230        return errors231 232    def _simulate_pod_status(self, resources: List[Dict[str, Any]]) -> str:233        """Simulate what pod status would be."""234        for r in resources:235            doc = r["doc"]236            kind = doc.get("kind", "")237            if kind not in ("Deployment", "StatefulSet", "DaemonSet", "Pod"):238                continue239 240            if kind == "Pod":241                containers = doc.get("spec", {}).get("containers", [])242            else:243                containers = doc.get("spec", {}).get("template", {}).get("spec", {}).get("containers", [])244 245            for c in containers:246                image = c.get("image", "")247 248                # Check for image typos (common: latset, lates, etc.)249                if image and ":" in image:250                    tag = image.split(":")[-1]251                    if tag in ("latset", "lates", "latets"):252                        return "ImagePullBackOff"253 254                # Check for hardcoded placeholder images255                if "OWNER/REPO" in image or "TAG" in image:256                    return "ImagePullBackOff"257 258                # Check memory limits259                resources_spec = c.get("resources", {})260                limits = resources_spec.get("limits", {})261                mem_limit = limits.get("memory", "")262                if mem_limit:263                    mem_bytes = _parse_memory(str(mem_limit))264                    # Simulate OOM if memory limit is very low265                    if 0 < mem_bytes < 128 * 1024 * 1024:  # < 128Mi266                        return "CrashLoopBackOff (OOMKilled)"267 268                # Check command269                command = c.get("command", [])270                if command and isinstance(command, list):271                    if any("wrong" in str(cmd).lower() or "typo" in str(cmd).lower() for cmd in command):272                        return "CrashLoopBackOff"273 274                # Check env refs to missing configmaps275                env_from = c.get("envFrom", [])276                for ef in env_from:277                    cm_ref = ef.get("configMapRef", {})278                    if cm_ref and cm_ref.get("name"):279                        # Check if configmap exists in resources280                        cm_exists = any(281                            res["doc"].get("kind") == "ConfigMap"282                            and res["doc"].get("metadata", {}).get("name") == cm_ref["name"]283                            for res in resources284                        )285                        if not cm_exists:286                            return f"CreateContainerConfigError (ConfigMap '{cm_ref['name']}' not found)"287 288        return "Running"289 290    def _simulate_service_status(self, resources: List[Dict[str, Any]]) -> str:291        """Simulate service endpoint status."""292        services = [r for r in resources if r["doc"].get("kind") == "Service"]293        deployments = [r for r in resources if r["doc"].get("kind") in ("Deployment", "StatefulSet")]294 295        if not services:296            return "N/A"297 298        for svc_r in services:299            svc = svc_r["doc"]300            selector = svc.get("spec", {}).get("selector", {})301            if not selector:302                continue303 304            matched = False305            for dep_r in deployments:306                dep = dep_r["doc"]307                tmpl_labels = dep.get("spec", {}).get("template", {}).get("metadata", {}).get("labels", {})308                if all(tmpl_labels.get(k) == v for k, v in selector.items()):309                    matched = True310 311                    # Check port matching312                    svc_ports = svc.get("spec", {}).get("ports", [])313                    container_ports = []314                    for c in dep.get("spec", {}).get("template", {}).get("spec", {}).get("containers", []):315                        for p in c.get("ports", []):316                            container_ports.append(p.get("containerPort"))317 318                    for sp in svc_ports:319                        tp = sp.get("targetPort")320                        if tp and tp not in container_ports and container_ports:321                            return f"Service port mismatch: targetPort {tp} not in container ports {container_ports}"322                    break323 324            if not matched:325                svc_name = svc.get("metadata", {}).get("name", "?")326                return f"No endpoints (selector {selector} matches no pods)"327 328        return "Endpoints active"329