Repository: GoogleCloudPlatform/kubectl-ai
File: pkg/tools/bash_tool.go (Go)

Code before:
	if kubeconfig != "" {
		kubeconfig, err := expandShellVar(kubeconfig)
		if err != nil {
			return nil, err
		}
		cmd.Env = append(cmd.Env, "KUBECONFIG="+kubeconfig)
	}

	return executeCommand(cmd)
}

type ExecResult struct {
	Error      string `json:"error,omitempty"`
	Stdout     string `json:"stdout,omitempty"`
	Stderr     string `json:"stderr,omitempty"`
	ExitCode   int    `json:"exit_code,omitempty"`
	StreamType string `json:"stream_type,omitempty"`
}

func executeCommand(cmd *exec.Cmd) (*ExecResult, error) {
	command := strings.Join(cmd.Args, " ")
	isWatch := strings.Contains(command, "kubectl get") && strings.Contains(command, "-w")
	isLogs := strings.Contains(command, "kubectl logs") && strings.Contains(command, "-f")
	isExec := strings.Contains(command, "kubectl exec") && strings.Contains(command, "-it")
	isAttach := strings.Contains(command, "kubectl attach")
	isPortForward := strings.Contains(command, "kubectl port-forward")

	// Block interactive commands
	if isExec || isPortForward {
		return &ExecResult{Error: "interactive mode not supported for kubectl, please use non-interactive commands"}, nil
	}

	// Handle streaming commands
	if isWatch || isLogs || isAttach {
		// Create a context with timeout
		ctx, cancel := context.WithTimeout(context.Background(), 7*time.Second)
		defer cancel()

		// Create pipes for stdout and stderr
		stdout, err := cmd.StdoutPipe()
		if err != nil {
			return nil, fmt.Errorf("creating stdout pipe: %w", err)
		}
		stderr, err := cmd.StderrPipe()
		if err != nil {
			return nil, fmt.Errorf("creating stderr pipe: %w", err)
		}

		if err := cmd.Start(); err != nil {
			return nil, fmt.Errorf("starting command: %w", err)
		}

Review comment (suggestion):
We have explicit check for `kubectl forward` and `kubectl edit` here as well https://github.com/GoogleCloudPlatform/kubectl-ai/blob/main/pkg/tools/kubectl_tool.go#L97

Since we invoke `executeCommand` in both of the tools (bash and kubectl), might make sense to remove those filtering from `kubectlTool` and add `kubectl edit` command filtering here.

Code after:
	if kubeconfig != "" {
		kubeconfig, err := expandShellVar(kubeconfig)
		if err != nil {
			return nil, err
		}
		cmd.Env = append(cmd.Env, "KUBECONFIG="+kubeconfig)
	}

	return executeCommand(cmd)
}

type ExecResult struct {
	Error      string `json:"error,omitempty"`
	Stdout     string `json:"stdout,omitempty"`
	Stderr     string `json:"stderr,omitempty"`
	ExitCode   int    `json:"exit_code,omitempty"`
	StreamType string `json:"stream_type,omitempty"`
}

func IsInteractiveCommand(command string) (bool, error) {
	// Inline isKubectlCommand logic
	words := strings.Fields(command)
	if len(words) == 0 {
		return false, nil
	}
	base := filepath.Base(words[0])
	if base != "kubectl" {
		return false, nil
	}

	isExec := strings.Contains(command, " exec ") && strings.Contains(command, " -it")
	isPortForward := strings.Contains(command, " port-forward ")
	isEdit := strings.Contains(command, " edit ")

	if isExec || isPortForward || isEdit {
		return true, fmt.Errorf("interactive mode not supported for kubectl, please use non-interactive commands")
	}
	return false, nil
}

func executeCommand(cmd *exec.Cmd) (*ExecResult, error) {
	command := strings.Join(cmd.Args, " ")

	if isInteractive, err := IsInteractiveCommand(command); isInteractive {
		return &ExecResult{Error: err.Error()}, nil
	}

	isWatch := strings.Contains(command, " get ") && strings.Contains(command, " -w")
	isLogs := strings.Contains(command, " logs ") && strings.Contains(command, " -f")
	isAttach := strings.Contains(command, " attach ")