Coverage for gco_mcp/resources/k8s.py: 93.15%

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``. Live 

6reads require ``gco://k8s/{region}/{namespace}/{kind}/{name}`` so kubectl is 

7always pinned to an account-qualified regional EKS context. 

8""" 

9 

10from __future__ import annotations 

11 

12import json 

13import re 

14from typing import Any 

15 

16import cli_runner 

17from cli_runner import PROJECT_ROOT # runtime-resolved checkout root (uvx-safe) 

18from server import mcp 

19 

20from resources._eks import eks_context_for_region, is_valid_region 

21 

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

23 

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

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

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

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

28 

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

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

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

32# into a flag. 

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

34 

35_KUBECTL_TIMEOUT_SECONDS = 30 

36 

37 

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

39def k8s_manifests_index() -> str: 

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

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

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

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

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

45 readme = MANIFESTS_DIR / "README.md" 

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

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

48 return "\n".join(lines) 

49 

50 

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

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

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

54 path = MANIFESTS_DIR / filename 

55 if not path.is_file(): 

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

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

58 return path.read_text() 

59 

60 

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

62 """Fail closed for the legacy URI that did not identify a cluster.""" 

63 return json.dumps( 

64 { 

65 "error": "explicit region required", 

66 "code": "eks_region_required", 

67 "use": f"gco://k8s/{{region}}/{namespace}/{kind}/{name}", 

68 } 

69 ) 

70 

71 

72def _k8s_live_resource_for_region(region: str, namespace: str, kind: str, name: str) -> str: 

73 """Return live YAML from an explicitly selected regional EKS cluster.""" 

74 if not is_valid_region(region): 74 ↛ 75line 74 didn't jump to line 75 because the condition on line 74 was never true

75 return json.dumps({"error": "invalid region", "value": region}) 

76 if not _K8S_NAME_RE.fullmatch(namespace): 

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

78 if not _K8S_KIND_RE.fullmatch(kind): 

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

80 if not _K8S_NAME_RE.fullmatch(name): 

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

82 try: 

83 context_arn = eks_context_for_region(region) 

84 except Exception as exc: # AWS credential/session failures become resource errors 

85 return json.dumps({"error": "unable to resolve EKS context", "detail": str(exc)[:200]}) 

86 try: 

87 result = cli_runner.subprocess.run( # type: ignore[attr-defined] # nosemgrep: dangerous-subprocess-use-audit - shell=False; every caller-controlled argv element is regex-validated and context is constructed internally 

88 [ 

89 "kubectl", 

90 "get", 

91 kind, 

92 name, 

93 "-n", 

94 namespace, 

95 "-o", 

96 "yaml", 

97 "--context", 

98 context_arn, 

99 ], 

100 capture_output=True, 

101 text=True, 

102 timeout=_KUBECTL_TIMEOUT_SECONDS, 

103 ) 

104 except FileNotFoundError: 

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

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

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

108 if result.returncode != 0: 

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

110 return json.dumps( 

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

112 ) 

113 return str(result.stdout) 

114 

115 

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

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

118 

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

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

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

122 alongside the rest of the live-state modules. 

123 """ 

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

125 mcp_instance.resource("gco://k8s/{region}/{namespace}/{kind}/{name}")( 

126 _k8s_live_resource_for_region 

127 )