fix(elastic+pipeline): keystore-backed cluster credentials + ProcessorsEquals self-compare

core/elastic/cluster_secrets.go

@@ -0,0 +1,182 @@
+/* Copyright © INFINI LTD. All rights reserved. */
+
+package elastic
+
+import (
+	"errors"
+	"strings"
+
+	log "github.com/cihub/seelog"
+
+	"infini.sh/framework/core/keystore"
+	"infini.sh/framework/core/util"
+	ucfg "infini.sh/framework/lib/go-ucfg"
+	keystore2 "infini.sh/framework/lib/keystore"
+)
+
+// ──────────────────────────────────────────────────────────────────────────
+// Cluster-scoped secrets in the keystore.
+//
+// ElasticsearchConfig credentials (basic_auth.password, token) are
+// ucfg.SecretString fields: json.Marshal emits only the raw part, and a
+// plain-text secret's raw part is the shadow text ("******"). A cluster
+// saved through the ORM therefore persists the MASK, never the real
+// credential - every consumer that later loads the record from the ORM
+// (boot-time live registration, the cluster-change hook, app-side
+// resolvers such as LogPilot's stream search) would authenticate with the
+// mask and get 401s from secured clusters.
+//
+// The keystore is the durable home for the real values:
+//
+//   - StashClusterSecrets / StashClusterSecretsFromDelta run on cluster
+//     create/update (the /easysearch/ CRUD), before the ORM write, and
+//     store any plain-text credential under a cluster-scoped key.
+//   - HydrateClusterSecrets runs wherever an ORM-loaded record is about to
+//     be used to build a client or authenticate a request, replacing a
+//     missing/masked secret with the keystore value.
+//   - RemoveClusterSecrets drops the keys when the cluster is deleted.
+//
+// The ORM record keeps the mask, so API responses (which marshal the ORM
+// object) never leak the real credential.
+// ──────────────────────────────────────────────────────────────────────────
+
+const clusterSecretKeyPrefix = "cluster_secret"
+
+// ClusterBasicAuthPasswordKey is the keystore key holding a cluster's
+// basic_auth password.
+func ClusterBasicAuthPasswordKey(clusterID string) string {
+	return clusterSecretKeyPrefix + "/" + clusterID + "/basic_auth_password"
+}
+
+// ClusterTokenKey is the keystore key holding a cluster's Easysearch API
+// token.
+func ClusterTokenKey(clusterID string) string {
+	return clusterSecretKeyPrefix + "/" + clusterID + "/token"
+}
+
+// StashClusterSecrets saves real (non-masked) credentials carried by cfg
+// into the keystore. Called on cluster create, before the ORM write.
+// Masked or empty values are skipped: the API masks secrets in responses,
+// so an edit that doesn't retype the password round-trips the mask and
+// must not clear the stored secret.
+func StashClusterSecrets(cfg *ElasticsearchConfig) {
+	if cfg == nil || cfg.ID == "" {
+		return
+	}
+	if cfg.BasicAuth != nil && cfg.BasicAuth.Username != "" {
+		if pw := cfg.BasicAuth.Password.Get(); isRealSecret(pw) {
+			stashClusterSecret(ClusterBasicAuthPasswordKey(cfg.ID), pw, ClusterTokenKey(cfg.ID))
+		}
+	}
+	if tok := cfg.Token.Get(); isRealSecret(tok) {
+		stashClusterSecret(ClusterTokenKey(cfg.ID), tok, ClusterBasicAuthPasswordKey(cfg.ID))
+	}
+}
+
+// StashClusterSecretsFromDelta is StashClusterSecrets for the raw
+// create/update request body (partial-update mode hands the hook a sparse
+// object; the credentials only exist in the delta map). Switching auth
+// mode (a new password vs a new token) drops the other key so hydration
+// cannot resurrect a stale credential.
+func StashClusterSecretsFromDelta(clusterID string, delta util.MapStr) {
+	if clusterID == "" || len(delta) == 0 {
+		return
+	}
+	if ba := asMapStr(delta["basic_auth"]); ba != nil {
+		username, _ := ba["username"].(string)
+		password, _ := ba["password"].(string)
+		if username != "" && isRealSecret(password) {
+			stashClusterSecret(ClusterBasicAuthPasswordKey(clusterID), password, ClusterTokenKey(clusterID))
+		}
+	}
+	if tok, ok := delta["token"].(string); ok && isRealSecret(tok) {
+		stashClusterSecret(ClusterTokenKey(clusterID), tok, ClusterBasicAuthPasswordKey(clusterID))
+	}
+}
+
+// HydrateClusterSecrets fills masked or missing credentials on cfg with
+// the real values from the keystore. Values already in memory (e.g. a
+// request that just carried the plain-text secret) are kept. Call this
+// wherever an ORM-loaded cluster record is about to be used to build a
+// client, compare connection identity, or authenticate a request.
+func HydrateClusterSecrets(cfg *ElasticsearchConfig) {
+	if cfg == nil || cfg.ID == "" {
+		return
+	}
+	if cfg.BasicAuth != nil && cfg.BasicAuth.Username != "" && !isRealSecret(cfg.BasicAuth.Password.Get()) {
+		v, ok := loadClusterSecret(ClusterBasicAuthPasswordKey(cfg.ID))
+		if ok {
+			if v == "" {
+				log.Warnf("cluster [%s]'s basic auth password is empty: %v", cfg.ID)
+			}
+			cfg.BasicAuth.Password = ucfg.SecretString(v)
+		} else {
+			log.Warnf("cluster [%s]'s basic auth password not found in the keystore: %v", cfg.ID)
+		}
+	}
+	if !isRealSecret(cfg.Token.Get()) {
+		v, ok := loadClusterSecret(ClusterTokenKey(cfg.ID))
+		if ok {
+			if v == "" {
+				log.Warnf("cluster [%s]'s token is empty: %v", cfg.ID)
+			}
+			cfg.Token = ucfg.SecretString(v)
+		} else {
+			log.Warnf("cluster [%s]'s token not found in the keystore: %v", cfg.ID)
+		}
+	}
+}
+
+// RemoveClusterSecrets drops the cluster's keystore entries. Called when
+// the cluster record is deleted.
+func RemoveClusterSecrets(clusterID string) {
+	if clusterID == "" {
+		return
+	}
+	for _, key := range []string{ClusterBasicAuthPasswordKey(clusterID), ClusterTokenKey(clusterID)} {
+		if err := keystore.DeleteValue(key); err != nil {
+			log.Debugf("cluster %s: remove keystore secret %s: %v", clusterID, key, err)
+		}
+	}
+}
+
+// stashClusterSecret stores value under key and clears otherKey (the
+// credential of the alternative auth mode), best-effort.
+func stashClusterSecret(key, value, otherKey string) {
+	if err := keystore.SetValue(key, []byte(value)); err != nil {
+		log.Warnf("keystore: stash cluster secret %s failed: %v", key, err)
+		return
+	}
+	if err := keystore.DeleteValue(otherKey); err != nil {
+		log.Debugf("keystore: drop stale cluster secret %s: %v", otherKey, err)
+	}
+}
+
+// Return value: (trimmed value, a bool indidcating if key eixsts)
+func loadClusterSecret(key string) (string, bool) {
+	v, err := keystore.GetValue(key)
+	if err != nil {
+		if !errors.Is(err, keystore2.ErrKeyDoesntExists) {
+			log.Debugf("keystore: load cluster secret %s: %v", key, err)
+		}
+		return "", false
+	}
+	s := strings.TrimSpace(string(v))
+	return s, true
+}
+
+// isRealSecret reports whether s carries an actual secret: not empty and
+// not the marshal mask.
+func isRealSecret(s string) bool {
+	return s != "" && s != ucfg.SecretShadowText
+}
+
+func asMapStr(v interface{}) map[string]interface{} {
+	switch m := v.(type) {
+	case map[string]interface{}:
+		return m
+	case util.MapStr:
+		return m
+	}
+	return nil
+}

core/elastic/cluster_secrets_test.go

@@ -0,0 +1,108 @@
+/* Copyright © INFINI Ltd. All rights reserved.
+ * Web: https://infinilabs.com
+ * Email: hello#infini.ltd */
+
+package elastic
+
+import (
+	"encoding/json"
+	"os"
+	"testing"
+
+	"infini.sh/framework/core/model"
+	"infini.sh/framework/core/keystore"
+	ucfg "infini.sh/framework/lib/go-ucfg"
+)
+
+// TestClusterSecretsORMRoundtrip reproduces the secured-cluster 401: a
+// plain-text password set on an ElasticsearchConfig is marshaled to the
+// shadow mask when the record is persisted, and only the keystore holds
+// the real value for HydrateClusterSecrets to restore.
+func TestClusterSecretsORMRoundtrip(t *testing.T) {
+	wd, _ := os.Getwd()
+	if err := os.Setenv(keystore.PathEnvKey, wd); err != nil {
+		t.Fatal(err)
+	}
+
+	// 1) Create: the request decodes a plain password; stash it, then
+	//    marshal the record the way the ORM does (mask must be stored).
+	cfg := ElasticsearchConfig{}
+	cfg.ID = "cs-test-1"
+	cfg.Name = "secured"
+	ba := model.BasicAuth{Username: "admin", Password: ucfg.SecretString("real-pass-123")}
+	cfg.BasicAuth = &ba
+	StashClusterSecrets(&cfg)
+
+	stored, err := json.Marshal(cfg)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if string(stored) == "" || !jsonContains(stored, `"password":"******"`) {
+		t.Fatalf("ORM marshal must persist the mask, got: %s", stored)
+	}
+	if jsonContains(stored, "real-pass-123") {
+		t.Fatalf("ORM marshal must not leak the real password: %s", stored)
+	}
+
+	// 2) Read back (what every ORM consumer sees) + hydrate.
+	loaded := ElasticsearchConfig{}
+	if err := json.Unmarshal(stored, &loaded); err != nil {
+		t.Fatal(err)
+	}
+	loaded.ID = cfg.ID
+	if got := loaded.BasicAuth.Password.Get(); got != ucfg.SecretShadowText {
+		t.Fatalf("loaded password should be the mask, got %q", got)
+	}
+	HydrateClusterSecrets(&loaded)
+	if got := loaded.BasicAuth.Password.Get(); got != "real-pass-123" {
+		t.Fatalf("hydrated password = %q, want real-pass-123", got)
+	}
+
+	// 3) Update that does NOT retype the password (mask round-trips):
+	//    stash must be skipped, hydration must still restore.
+	edited := ElasticsearchConfig{}
+	if err := json.Unmarshal(stored, &edited); err != nil {
+		t.Fatal(err)
+	}
+	edited.ID = cfg.ID
+	StashClusterSecrets(&edited) // password is the mask: no-op
+	HydrateClusterSecrets(&edited)
+	if got := edited.BasicAuth.Password.Get(); got != "real-pass-123" {
+		t.Fatalf("unretyped edit must keep the stashed secret, got %q", got)
+	}
+
+	// 4) Retyped password via the raw delta (partial-update mode).
+	StashClusterSecretsFromDelta(cfg.ID, map[string]interface{}{
+		"basic_auth": map[string]interface{}{"username": "admin", "password": "new-pass-456"},
+	})
+	rehydrated := ElasticsearchConfig{}
+	_ = json.Unmarshal(stored, &rehydrated)
+	rehydrated.ID = cfg.ID
+	HydrateClusterSecrets(&rehydrated)
+	if got := rehydrated.BasicAuth.Password.Get(); got != "new-pass-456" {
+		t.Fatalf("retyped password must replace the stashed secret, got %q", got)
+	}
+
+	// 5) Delete removes the secret; hydration then leaves the mask.
+	RemoveClusterSecrets(cfg.ID)
+	after := ElasticsearchConfig{}
+	_ = json.Unmarshal(stored, &after)
+	after.ID = cfg.ID
+	HydrateClusterSecrets(&after)
+	if got := after.BasicAuth.Password.Get(); got != ucfg.SecretShadowText {
+		t.Fatalf("after delete hydration must leave the mask, got %q", got)
+	}
+}
+
+func jsonContains(b []byte, sub string) bool {
+	return len(sub) == 0 || (len(b) >= len(sub) && stringContains(string(b), sub))
+}
+
+func stringContains(s, sub string) bool {
+	for i := 0; i+len(sub) <= len(s); i++ {
+		if s[i:i+len(sub)] == sub {
+			return true
+		}
+	}
+	return false
+}

core/keystore/keystore.go

@@ -140,6 +140,19 @@ func SetValue(key string, value []byte) error {
 	return ksw.Save()
 }
 
+// DeleteValue removes a key from the keystore and persists the change.
+// Removing a key that is not present is a no-op.
+func DeleteValue(key string) error {
+	ksw, err := GetWriteableKeystore()
+	if err != nil {
+		return err
+	}
+	if err := ksw.Delete(key); err != nil {
+		return err
+	}
+	return ksw.Save()
+}
+
 func GetVariableResolver() (ucfg.Option, error) {
 	return ucfg.Resolve(func(keyName string) (string, parse.Config, error) {
 		if strings.HasPrefix(keyName, "keystore.") {

core/keystore/keystore_test.go

@@ -30,7 +30,7 @@ package keystore
 import (
 	"github.com/stretchr/testify/assert"
 	"infini.sh/framework/core/config"
-	"infini.sh/framework/core/elastic"
+	ucfg "infini.sh/framework/lib/go-ucfg"
 	"os"
 	"path"
 	"testing"
@@ -56,7 +56,16 @@ func TestConfigVariable(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	esConfigs := []elastic.ElasticsearchConfig{}
+	// 本地同构结构体: 避免 import core/elastic (cluster_secrets 引用本包,
+	// 测试再引 elastic 会成环)。
+	type testBasicAuth struct {
+		Username string            `config:"username"`
+		Password ucfg.SecretString `config:"password"`
+	}
+	type testESConfig struct {
+		BasicAuth *testBasicAuth `config:"basic_auth"`
+	}
+	esConfigs := []testESConfig{}
 	esCfg, err := cfg.Child("elasticsearch", -1)
 	if err != nil {
 		t.Fatal(err)

core/pipeline/config.go

@@ -145,7 +145,7 @@ func (this PipelineConfigV2) ProcessorsEquals(target PipelineConfigV2) bool {
 	if err != nil {
 		panic(err)
 	}
-	targetCfg, err := this.GetProcessorsConfig()
+	targetCfg, err := target.GetProcessorsConfig()
 	if err != nil {
 		panic(err)
 	}

core/pipeline/equals_nested_test.go

@@ -0,0 +1,34 @@
+package pipeline
+
+import (
+	"testing"
+)
+
+func TestEqualsNestedChange(t *testing.T) {
+	old := PipelineConfigV2{
+		Name: "p1",
+		Processors: []map[string]interface{}{
+			{"logs_processor": map[string]interface{}{
+				"logs_path": "/x",
+				"ship_config": map[string]interface{}{
+					"endpoints": []interface{}{"192.168.43.62:4317"},
+				},
+			}},
+		},
+	}
+	neu := PipelineConfigV2{
+		Name: "p1",
+		Processors: []map[string]interface{}{
+			{"logs_processor": map[string]interface{}{
+				"logs_path": "/x",
+				"ship_config": map[string]interface{}{
+					"endpoints": []interface{}{"127.0.0.1:4317"},
+				},
+			}},
+		},
+	}
+	if old.Equals(neu) {
+		t.Fatal("nested endpoint change NOT detected by Equals")
+	}
+	t.Log("Equals detects nested change correctly")
+}