diff --git a/core/elastic/cluster_secrets.go b/core/elastic/cluster_secrets.go new file mode 100644 index 000000000..156f9a671 --- /dev/null +++ b/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 +} diff --git a/core/elastic/cluster_secrets_test.go b/core/elastic/cluster_secrets_test.go new file mode 100644 index 000000000..33836208c --- /dev/null +++ b/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 +} diff --git a/core/keystore/keystore.go b/core/keystore/keystore.go index c009b31d2..785165926 100644 --- a/core/keystore/keystore.go +++ b/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.") { diff --git a/core/keystore/keystore_test.go b/core/keystore/keystore_test.go index 5029c3473..3ee991561 100644 --- a/core/keystore/keystore_test.go +++ b/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) diff --git a/core/pipeline/config.go b/core/pipeline/config.go index ceec5f42d..efa439e8b 100644 --- a/core/pipeline/config.go +++ b/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) } diff --git a/core/pipeline/equals_nested_test.go b/core/pipeline/equals_nested_test.go new file mode 100644 index 000000000..bbe50a42d --- /dev/null +++ b/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") +} diff --git a/modules/easysearch/cluster_api.go b/modules/easysearch/cluster_api.go index 7b8845737..fb1715753 100644 --- a/modules/easysearch/cluster_api.go +++ b/modules/easysearch/cluster_api.go @@ -91,6 +91,16 @@ func registerClusterAPI() { // loop persists status for it (the loop keys off this source value). cfg.Source = elastic.ElasticsearchConfigSourceElasticsearch cfg.Enabled = true + // SecretString round-trips through the ORM as the marshal mask, + // so the real credential must be stashed in the keystore before + // the record is saved (see core/elastic/cluster_secrets.go). + elastic.StashClusterSecrets(cfg) + return nil + }, + PrepareUpdate: func(cfg *elastic.ElasticsearchConfig, delta util.MapStr) error { + // Partial-update mode hands this hook a sparse object: the new + // credential (when retyped) only exists in the raw request body. + elastic.StashClusterSecretsFromDelta(cfg.ID, delta) return nil }, GuardDelete: func(cfg *elastic.ElasticsearchConfig) error { @@ -100,23 +110,29 @@ func registerClusterAPI() { return nil }, // Live registration: writing a cluster record takes effect - // immediately — no restart or boot-time ORM reload needed. This is + // immediately - no restart or boot-time ORM reload needed. This is // what lets a manager (e.g. LogPilot) push sink clusters to gateways // dynamically; pipelines referencing the cluster id resolve on the // next use. PostCreate: func(cfg *elastic.ElasticsearchConfig) error { + elastic.HydrateClusterSecrets(cfg) if _, err := common.InitElasticInstance(*cfg); err != nil { return fmt.Errorf("cluster %s saved but live registration failed: %w", cfg.ID, err) } return nil }, PostUpdate: func(cfg *elastic.ElasticsearchConfig) error { + // crud reloads the record from the ORM before post hooks run: + // hydrate so live registration sees the real credential, not + // the mask persisted in the record. + elastic.HydrateClusterSecrets(cfg) if _, err := common.InitElasticInstance(*cfg); err != nil { return fmt.Errorf("cluster %s updated but live re-registration failed: %w", cfg.ID, err) } return nil }, PostDelete: func(cfg *elastic.ElasticsearchConfig) error { + elastic.RemoveClusterSecrets(cfg.ID) elastic.RemoveInstance(cfg.ID) return nil }, diff --git a/modules/elastic/cluster_hook.go b/modules/elastic/cluster_hook.go index 45d6b0c33..4bd843ed2 100644 --- a/modules/elastic/cluster_hook.go +++ b/modules/elastic/cluster_hook.go @@ -51,6 +51,12 @@ func handleClusterChange(ctx *orm.Context, op orm.Operation, o interface{}) (*or return ctx, o, nil } + // ORM-loaded records carry the marshal mask instead of the real + // credential (SecretString json round-trip): hydrate from the keystore + // so identity comparison and live re-initialization below see real + // credentials. A no-op when the caller's copy already carries them. + elastic.HydrateClusterSecrets(cfg) + switch op { case orm.OpCreate: if _, err := common.InitElasticInstance(*cfg); err != nil { @@ -86,6 +92,7 @@ func handleClusterChange(ctx *orm.Context, op orm.Operation, o interface{}) (*or } case orm.OpDelete: + elastic.RemoveClusterSecrets(cfg.ID) elastic.RemoveInstance(cfg.ID) elastic.InvalidateClient(*cfg) log.Debugf("cluster %s (%s): live client removed after delete", cfg.ID, cfg.Name) diff --git a/modules/elastic/cluster_loader.go b/modules/elastic/cluster_loader.go index ab77236fc..f1db9bd06 100644 --- a/modules/elastic/cluster_loader.go +++ b/modules/elastic/cluster_loader.go @@ -36,6 +36,9 @@ func LoadClustersFromORM() { return } for _, cfg := range clusters { + // ORM records persist the marshal mask, not the real credential: + // hydrate from the keystore before building the live client. + elastic.HydrateClusterSecrets(&cfg) if _, err := common.InitElasticInstance(cfg); err != nil { log.Warnf("cluster %s (%s): init failed: %v", cfg.ID, cfg.Name, err) }