LokiStack/backend/services/k8s_client.go

package services

import (
	"encoding/base64"
	"fmt"
	"os"
	"regexp"
	"time"

	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"
	"lokistack/models"
)

// resolveKubeconfig 取集群最终的 kubeconfig 文本（优先内存字段，其次磁盘文件）
func resolveKubeconfig(cluster *models.Cluster) (string, error) {
	if cluster.KubeConfig != "" {
		if cluster.IsKubeconfigEncrypted() {
			plain := cluster.DecryptKubeconfig()
			if plain == "" {
				return "", fmt.Errorf("kubeconfig 解密失败（ENCRYPTION_KEY 是否变更？）")
			}
			return plain, nil
		}
		return cluster.KubeConfig, nil
	}
	if cluster.KubeConfigPath != "" {
		data, err := os.ReadFile(cluster.KubeConfigPath)
		if err != nil {
			return "", fmt.Errorf("读取 kubeconfig 文件失败: %w", err)
		}
		return string(data), nil
	}
	return "", fmt.Errorf("集群未配置 kubeconfig")
}

// clusterRESTConfig 使用集群保存的 kubeconfig 构造原生 API Server 客户端配置。
func clusterRESTConfig(cluster *models.Cluster, timeout time.Duration) (*rest.Config, error) {
	source, err := resolveKubeconfig(cluster)
	if err != nil {
		return nil, err
	}
	config, err := clientcmd.Load([]byte(source))
	if err != nil {
		return nil, fmt.Errorf("解析集群 kubeconfig 失败: %w", err)
	}
	overrides := &clientcmd.ConfigOverrides{}
	if cluster.Context != "" {
		overrides.CurrentContext = cluster.Context
	}
	result, err := clientcmd.NewNonInteractiveClientConfig(*config, config.CurrentContext, overrides, nil).ClientConfig()
	if err != nil {
		return nil, fmt.Errorf("构造 Kubernetes 客户端失败: %w", err)
	}
	result.Timeout = timeout
	result.UserAgent = "lokistack"
	return result, nil
}

func clusterClient(cluster *models.Cluster, timeout time.Duration) (kubernetes.Interface, error) {
	config, err := clusterRESTConfig(cluster, timeout)
	if err != nil {
		return nil, err
	}
	return kubernetes.NewForConfig(config)
}

// BuildKubeconfigFromToken 用 API Server + Token 构造最小 kubeconfig
// 优先使用 CA 证书验证 TLS，仅在用户显式要求时跳过证书验证
// Parameters:
//   - apiServer: Kubernetes API Server 地址，如 https://192.168.1.10:6443
//   - token: 认证 Token
//   - caCert: CA 证书 PEM 内容（可选）。若提供则嵌入证书数据进行 TLS 验证；
//     若为空且 skipTLSVerify=true 则跳过 TLS 验证；
//     若为空且 skipTLSVerify=false 则依赖系统 CA 池
//   - skipTLSVerify: 是否跳过 TLS 证书验证（不建议在生产环境使用）
func BuildKubeconfigFromToken(apiServer, token, caCert string, skipTLSVerify bool) string {
	var clusterBlock string

	if caCert != "" {
		// 嵌入 CA 证书进行安全的 TLS 验证
		encodedCA := base64.StdEncoding.EncodeToString([]byte(caCert))
		clusterBlock = fmt.Sprintf(`clusters:
- cluster:
    certificate-authority-data: %s
    server: %s
  name: loki-cluster`, encodedCA, apiServer)
	} else if skipTLSVerify {
		// 用户显式要求跳过 TLS 验证（内网/自签证书场景）
		clusterBlock = fmt.Sprintf(`clusters:
- cluster:
    insecure-skip-tls-verify: true
    server: %s
  name: loki-cluster`, apiServer)
	} else {
		// 依赖系统 CA 池（不嵌入证书也不跳过验证）
		clusterBlock = fmt.Sprintf(`clusters:
- cluster:
    server: %s
  name: loki-cluster`, apiServer)
	}

	return fmt.Sprintf(`apiVersion: v1
kind: Config
%s
contexts:
- context:
    cluster: loki-cluster
    user: loki-user
  name: loki-context
current-context: loki-context
users:
- name: loki-user
  user:
    token: %s
`, clusterBlock, token)
}

var apiServerRe = regexp.MustCompile(`(?m)^\s*server:\s*(\S+)`)

// ExtractAPIServer 从 kubeconfig 文本提取 server 地址（展示用）
func ExtractAPIServer(kubeconfig string) string {
	m := apiServerRe.FindStringSubmatch(kubeconfig)
	if len(m) >= 2 {
		return m[1]
	}
	return ""
}