From 753907792c0d9fec644300d6ba51277351f7f5a3 Mon Sep 17 00:00:00 2001 From: Guoxun Wei Date: Fri, 12 Jun 2026 00:36:04 +0800 Subject: [PATCH] fix(k8s): block kubectl auth reconcile in readonly mode The mcp-kubernetes validator classifies the entire "auth" verb as a read-only operation because the common subcommands ("auth can-i", "auth whoami") are queries. However, "kubectl auth reconcile" creates or updates RBAC Role / RoleBinding / ClusterRole / ClusterRoleBinding objects from a manifest, so the broad "auth" classification leaks a write capability into readonly access level. Add a defense-in-depth guard in the executor adapter that runs before delegation, so it covers both the default mcp-kubernetes path and the token-auth-only RunCommand path. The guard tokenizes via shlex (matching the executor) and walks positional tokens past flags, so quoting and tab-separated forms cannot bypass a literal substring scan, and content after a free-standing "--" (subprocess args of `kubectl exec ... --`) is correctly ignored. Tests cover the readonly block (plain, no-prefix, with leading flag, tab-separated), readonly allow for can-i/whoami, readwrite/admin allow for reconcile, and the tokenizer edge cases directly. --- internal/k8s/adapter.go | 56 +++++++++++++++++ internal/k8s/adapter_test.go | 117 +++++++++++++++++++++++++++++++++++ 2 files changed, 173 insertions(+) diff --git a/internal/k8s/adapter.go b/internal/k8s/adapter.go index 851c96e..76189a9 100644 --- a/internal/k8s/adapter.go +++ b/internal/k8s/adapter.go @@ -5,6 +5,8 @@ package k8s import ( "context" + "fmt" + "strings" "github.com/Azure/aks-mcp/internal/config" "github.com/Azure/aks-mcp/internal/tools" @@ -12,6 +14,7 @@ import ( k8ssecurity "github.com/Azure/mcp-kubernetes/pkg/security" k8stelemetry "github.com/Azure/mcp-kubernetes/pkg/telemetry" k8stools "github.com/Azure/mcp-kubernetes/pkg/tools" + "github.com/google/shlex" ) // ConvertConfig maps an aks-mcp ConfigData into the equivalent @@ -79,6 +82,20 @@ type executorAdapter struct { // Execute adapts aks-mcp execution by converting its config // and delegating to the wrapped mcp-kubernetes executor or RunCommand executor. func (a *executorAdapter) Execute(ctx context.Context, params map[string]interface{}, cfg *config.ConfigData) (string, error) { + // Defense-in-depth: reject "kubectl auth reconcile" in readonly mode + // regardless of which downstream validator runs. The mcp-kubernetes + // readonly classifier groups all "auth" subcommands as read-only, but + // "auth reconcile" creates/updates RBAC objects and must not be reachable + // from a readonly access level. This check fails closed even if the + // dependency is downgraded or its classification regresses. + if cfg.AccessLevel == "readonly" { + if cmd, ok := params["command"].(string); ok { + if isKubectlAuthReconcile(cmd) { + return "", fmt.Errorf("security validation failed: kubectl auth reconcile is a write operation and cannot be executed in read-only mode") + } + } + } + if a.tokenAuthOnly { k8sCfg := ConvertConfig(cfg) return a.runCommandExecutor.Execute(ctx, params, k8sCfg) @@ -86,3 +103,42 @@ func (a *executorAdapter) Execute(ctx context.Context, params map[string]interfa k8sCfg := ConvertConfig(cfg) return a.k8sExecutor.Execute(ctx, params, k8sCfg) } + +// isKubectlAuthReconcile reports whether the kubectl command string invokes +// "auth reconcile". Tokenizes via shlex (matching the executor) so quoted or +// tab-separated forms cannot bypass a literal substring check, then walks the +// positional tokens skipping flags. Returns true only when the first +// positional after "auth" is "reconcile". +func isKubectlAuthReconcile(command string) bool { + tokens, err := shlex.Split(command) + if err != nil { + tokens = strings.Fields(command) + } + // Drop everything after a free-standing "--" so subprocess args + // (e.g. `kubectl exec ... -- auth reconcile`) are not misclassified. + for i, t := range tokens { + if t == "--" { + tokens = tokens[:i] + break + } + } + seenAuth := false + for _, t := range tokens { + if strings.HasPrefix(t, "-") { + continue + } + if t == "kubectl" { + continue + } + if !seenAuth { + if t == "auth" { + seenAuth = true + continue + } + // First positional is not "auth" — not an auth subcommand. + return false + } + return t == "reconcile" + } + return false +} diff --git a/internal/k8s/adapter_test.go b/internal/k8s/adapter_test.go index 2509a35..bb2a7e8 100644 --- a/internal/k8s/adapter_test.go +++ b/internal/k8s/adapter_test.go @@ -233,3 +233,120 @@ func BenchmarkConvertConfig(b *testing.B) { benchOut = ConvertConfig(in) } } + +// TestExecutorAdapter_BlocksAuthReconcileInReadonly verifies the +// defense-in-depth check rejects "kubectl auth reconcile" when the access +// level is readonly, without delegating to the underlying executor. The +// "auth" verb is classified as read-only by the upstream validator (because +// "auth can-i" / "auth whoami" are read-only), but "auth reconcile" creates +// or updates RBAC objects and must not be reachable from readonly. +func TestExecutorAdapter_BlocksAuthReconcileInReadonly(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + command string + }{ + {"plain", "kubectl auth reconcile -f rbac.yaml"}, + {"no-prefix", "auth reconcile -f rbac.yaml"}, + {"with-leading-flag", "kubectl -v=2 auth reconcile -f rbac.yaml"}, + {"tab-separated", "kubectl\tauth\treconcile\t-f\trbac.yaml"}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + fe := &fakeExecutor{out: "should-not-run"} + adapter := WrapK8sExecutor(fe, false) + cfg := &config.ConfigData{AccessLevel: "readonly"} + params := map[string]interface{}{"command": tc.command} + + out, err := adapter.Execute(context.Background(), params, cfg) + if err == nil { + t.Fatalf("expected error for %q, got nil (out=%q)", tc.command, out) + } + if fe.lastParams != nil { + t.Fatalf("downstream executor should not have been invoked, got params=%v", fe.lastParams) + } + }) + } +} + +// TestExecutorAdapter_AllowsAuthReadInReadonly verifies the defense-in-depth +// check does not over-block: "auth can-i" and "auth whoami" remain allowed +// in readonly mode. +func TestExecutorAdapter_AllowsAuthReadInReadonly(t *testing.T) { + t.Parallel() + + cases := []string{ + "kubectl auth can-i create pods", + "kubectl auth whoami", + } + + for _, command := range cases { + t.Run(command, func(t *testing.T) { + fe := &fakeExecutor{out: "ok"} + adapter := WrapK8sExecutor(fe, false) + cfg := &config.ConfigData{AccessLevel: "readonly"} + params := map[string]interface{}{"command": command} + + out, err := adapter.Execute(context.Background(), params, cfg) + if err != nil { + t.Fatalf("unexpected error for %q: %v", command, err) + } + mustEqual(t, out, "ok", "adapter output") + }) + } +} + +// TestExecutorAdapter_AllowsAuthReconcileInReadwrite verifies the +// defense-in-depth check is gated on the readonly access level — reconcile +// is a legitimate write operation and must work in readwrite/admin. +func TestExecutorAdapter_AllowsAuthReconcileInReadwrite(t *testing.T) { + t.Parallel() + + for _, level := range []string{"readwrite", "admin"} { + t.Run(level, func(t *testing.T) { + fe := &fakeExecutor{out: "ok"} + adapter := WrapK8sExecutor(fe, false) + cfg := &config.ConfigData{AccessLevel: level} + params := map[string]interface{}{"command": "kubectl auth reconcile -f rbac.yaml"} + + out, err := adapter.Execute(context.Background(), params, cfg) + if err != nil { + t.Fatalf("unexpected error at access level %q: %v", level, err) + } + mustEqual(t, out, "ok", "adapter output") + }) + } +} + +// TestIsKubectlAuthReconcile covers the tokenizer edge cases directly. +func TestIsKubectlAuthReconcile(t *testing.T) { + t.Parallel() + + cases := []struct { + command string + want bool + }{ + {"kubectl auth reconcile -f rbac.yaml", true}, + {"auth reconcile -f rbac.yaml", true}, + {"kubectl auth reconcile", true}, + {"kubectl -v=2 auth reconcile", true}, + {"kubectl auth can-i create pods", false}, + {"kubectl auth whoami", false}, + {"kubectl get pods", false}, + {"kubectl exec mypod -- auth reconcile", false}, + {"kubectl apply -f rbac.yaml", false}, + {"", false}, + {"kubectl", false}, + } + + for _, tc := range cases { + t.Run(tc.command, func(t *testing.T) { + got := isKubectlAuthReconcile(tc.command) + if got != tc.want { + t.Fatalf("isKubectlAuthReconcile(%q) = %v, want %v", tc.command, got, tc.want) + } + }) + } +}