Coverage for mcp/resources/k8s.py: 98%

1"""Kubernetes manifest and live-state resources (k8s:// + gco://k8s/...) for 

2the GCO MCP server. 

3 

4The static ``k8s://gco/manifests/...`` paths surface the cluster-bootstrap 

5manifests that ship under ``lambda/kubectl-applier-simple/manifests``. The 

6live ``gco://k8s/{namespace}/{kind}/{name}`` template wraps ``kubectl get`` 

7so an LLM can pin any cluster resource for inspection across turns. 

8""" 

9 

10from __future__ import annotations 

11 

12import json 

13import re 

14from pathlib import Path 

15from typing import Any 

16 

17import cli_runner 

18from server import mcp 

19 

20PROJECT_ROOT = Path(__file__).parent.parent.parent 

21MANIFESTS_DIR = PROJECT_ROOT / "lambda" / "kubectl-applier-simple" / "manifests" 

22 

23# Permissive RFC 1123 label rule for namespace and resource names. 

24# Bounded length plus the alphanumeric+hyphen alphabet rules out 

25# command-injection vectors when the value is forwarded to ``kubectl``. 

26_K8S_NAME_RE = re.compile(r"^[a-z0-9](?:[-a-z0-9]{0,251}[a-z0-9])?$") 

27 

28# Kubernetes resource kinds — alphanumeric only, including dotted CRD 

29# group forms like ``deployments.apps`` and ``ingresses.networking.k8s.io``. 

30# Kept deliberately tight so ``kubectl get <kind>`` cannot be coerced 

31# into a flag. 

32_K8S_KIND_RE = re.compile(r"^[A-Za-z][A-Za-z0-9]*(?:\.[A-Za-z0-9-]+)*$") 

33 

34_KUBECTL_TIMEOUT_SECONDS = 30 

35 

36 

37@mcp.resource("k8s://gco/manifests/index") 

38def k8s_manifests_index() -> str: 

39 """List all Kubernetes manifests deployed to the EKS cluster.""" 

40 lines = ["# Kubernetes Cluster Manifests\n"] 

41 lines.append("Applied in order during `gco stacks deploy`:\n") 

42 for f in sorted(MANIFESTS_DIR.glob("*.yaml")): 

43 lines.append(f"- `k8s://gco/manifests/{f.name}` — {f.stem}") 

44 readme = MANIFESTS_DIR / "README.md" 

45 if readme.is_file(): 45 ↛ 47line 45 didn't jump to line 47 because the condition on line 45 was always true

46 lines.append("\n- `k8s://gco/manifests/README.md` — manifest documentation") 

47 return "\n".join(lines) 

48 

49 

50@mcp.resource("k8s://gco/manifests/{filename}") 

51def k8s_manifest_resource(filename: str) -> str: 

52 """Read a Kubernetes manifest that gets applied to the EKS cluster.""" 

53 path = MANIFESTS_DIR / filename 

54 if not path.is_file(): 

55 available = sorted(f.name for f in MANIFESTS_DIR.glob("*") if f.is_file()) 

56 return f"Manifest '{filename}' not found. Available:\n" + "\n".join(available) 

57 return path.read_text() 

58 

59 

60def _k8s_live_resource(namespace: str, kind: str, name: str) -> str: 

61 """Return the live YAML for ``<kind>/<name>`` in ``<namespace>``.""" 

62 if not _K8S_NAME_RE.match(namespace): 

63 return json.dumps({"error": "invalid namespace", "value": namespace}) 

64 if not _K8S_KIND_RE.match(kind): 

65 return json.dumps({"error": "invalid kind", "value": kind}) 

66 if not _K8S_NAME_RE.match(name): 

67 return json.dumps({"error": "invalid name", "value": name}) 

68 try: 

69 result = cli_runner.subprocess.run( # type: ignore[attr-defined] # nosemgrep: dangerous-subprocess-use-audit - shell=False; every argv element is regex-validated above 

70 ["kubectl", "get", kind, name, "-n", namespace, "-o", "yaml"], 

71 capture_output=True, 

72 text=True, 

73 timeout=_KUBECTL_TIMEOUT_SECONDS, 

74 ) 

75 except FileNotFoundError: 

76 return json.dumps({"error": "kubectl not found"}) 

77 except cli_runner.subprocess.TimeoutExpired: # type: ignore[attr-defined] 

78 return json.dumps({"error": f"kubectl timed out after {_KUBECTL_TIMEOUT_SECONDS}s"}) 

79 if result.returncode != 0: 

80 err = (result.stderr or result.stdout or "").strip() 

81 return json.dumps( 

82 {"error": err or "kubectl command failed", "exit_code": result.returncode} 

83 ) 

84 return str(result.stdout) 

85 

86 

87def register(mcp_instance: Any) -> None: 

88 """Register the live Kubernetes resource template against ``mcp_instance``. 

89 

90 The static ``k8s://gco/manifests/...`` resources are decorated at 

91 import time and don't need re-registration; this function exists 

92 so ``register_all_resources()`` can wire the live template in 

93 alongside the rest of the live-state modules. 

94 """ 

95 mcp_instance.resource("gco://k8s/{namespace}/{kind}/{name}")(_k8s_live_resource)