diff --git a/README.md b/README.md index 6600e6c..61445ed 100644 --- a/README.md +++ b/README.md @@ -151,7 +151,11 @@ tokens: scopes: "*" # "*" for all scopes, or comma-separated list storage: - type: "vault" - path: "secret/data/linode/tokens/my-api-token" + # Path is relative to vault.mount_path (omit mount name and "data/"). + # With mount_path "secret", this writes secret/data/linode/tokens/my-api-token + path: "linode/tokens/my-api-token" + # key: "token" # optional KV data key (default: token) + # action: "replace" # optional: replace (default) or append - label: "backup-token" team: "sre-team" @@ -160,9 +164,30 @@ tokens: rotation_threshold: 15 # Override global threshold for this token storage: - type: "vault" - path: "secret/data/linode/tokens/backup" + path: "linode/tokens/backup" + key: "api_token" # for consumers that do not read key "token" + action: "append" # preserve other keys already on this secret ``` +### Vault storage fields + +| Field | Default | Description | +|-------|---------|-------------| +| `type` | (required) | Only `vault` is supported today | +| `path` | (required) | Secret path **relative to** `vault.mount_path`. Do **not** include a leading `data/` or `{mount}/data/` prefix (a later path segment named `data` is fine). latr writes `{mount_path}/data/{path}` and state to `{mount_path}/metadata/{path}`. | +| `key` | `token` | KV v2 **data** map key for the token string (custom metadata for rotation state is separate). | +| `action` | `replace` | How to write the data map (case-insensitive): | + +**`action: replace` (default)** — data map becomes only `{key: }`. Other data keys on that secret are removed. Safe when latr owns the whole secret (typical). Custom metadata is not cleared. + +**`action: append`** — read-modify-write with KV v2 check-and-set (CAS) retries: set/overwrite `key`, keep other data keys. Use when the secret is shared with non-latr fields (e.g. multi-key consumer secrets). + +AppRole policy implications: + +- Both actions need `create`/`update` on `…/data/…` and `…/metadata/…` (for rotation state). +- **`append` also needs `read` on `…/data/…`** so latr can merge existing keys. +- Prefer least-privilege paths (no mount-wide wildcards). + ## Usage ### One-Shot Mode @@ -351,7 +376,23 @@ latr supports OpenTelemetry for observability: - `latr_rotations_total{status,label,team}` - Rotation attempts (per token) - `latr_rotation_duration_seconds{label,team}` - Rotation operation duration - `latr_token_validity_remaining_seconds{label,team}` - Time until rotation needed -- `latr_vault_storage_errors_total{path}` - Vault write failures +- `latr_vault_storage_errors_total{path,action}` - Vault write failures (`action` is `replace` or `append`) +- `latr_vault_writes_total{action,result}` - Vault KV data writes (`action=replace|append`, `result=success|error`) +- `latr_vault_write_duration_seconds{action}` - Vault KV write latency histogram +- `latr_vault_append_cas_conflicts_total` - Check-and-set version mismatches during `action: append` (retries) + +Useful PromQL: + +```promql +# Write success rate by action +sum by (action, result) (rate(latr_vault_writes_total[5m])) + +# p99 Vault write latency +histogram_quantile(0.99, sum by (le, action) (rate(latr_vault_write_duration_seconds_bucket[5m]))) + +# CAS contention on shared secrets (append only) +sum(rate(latr_vault_append_cas_conflicts_total[5m])) +``` ### Grafana dashboard & alerts (mixin) diff --git a/examples/config.yaml b/examples/config.yaml index c45d530..4255c35 100644 --- a/examples/config.yaml +++ b/examples/config.yaml @@ -31,7 +31,12 @@ tokens: scopes: "*" # "*" for all scopes, or comma-separated list storage: - type: "vault" - path: "secret/data/linode/tokens/my-api-token" + # Path is relative to vault.mount_path (NOT including mount or "data/"). + # Client writes to: {mount_path}/data/{path} + # Example below → secret/data/linode/tokens/my-api-token + path: "linode/tokens/my-api-token" + # key defaults to "token" when omitted + # action defaults to "replace" when omitted - label: "backup-token" team: "sre-team" @@ -40,7 +45,7 @@ tokens: rotation_threshold: 15 # Override global threshold for this token storage: - type: "vault" - path: "secret/data/linode/tokens/backup" + path: "linode/tokens/backup" - label: "short-lived-token" team: "dev-team" @@ -48,4 +53,10 @@ tokens: scopes: "linodes:read_write" storage: - type: "vault" - path: "secret/data/linode/tokens/dev" + path: "linode/tokens/dev" + # Override KV data key for consumers that do not read "token" + # (e.g. legacy Salt / IPService-style secrets) + key: "api_token" + # action: "replace" (default) — secret data becomes only this key + # action: "append" — merge this key into existing secret data (needs AppRole read) + action: "append" diff --git a/helm/latr/README.md b/helm/latr/README.md index d1fbace..6c8a849 100644 --- a/helm/latr/README.md +++ b/helm/latr/README.md @@ -45,7 +45,8 @@ config: scopes: "*" storage: - type: "vault" - path: "secret/data/linode/tokens/production" + # relative to config.vault.mountPath (no "data/" prefix) + path: "linode/tokens/production" secrets: linodeToken: "your-linode-token" @@ -194,7 +195,7 @@ config: scopes: "*" storage: - type: "vault" - path: "secret/data/linode/tokens/my-token" + path: "linode/tokens/my-token" secrets: linodeToken: "your-linode-token" @@ -217,7 +218,7 @@ config: rotationThreshold: 10 storage: - type: "vault" - path: "secret/data/linode/tokens/prod-full" + path: "linode/tokens/prod-full" - label: "dev-limited-access" team: "development" @@ -226,7 +227,7 @@ config: rotationThreshold: 20 storage: - type: "vault" - path: "secret/data/linode/tokens/dev-limited" + path: "linode/tokens/dev-limited" ``` ### Example 3: With OpenTelemetry @@ -247,7 +248,7 @@ config: scopes: "*" storage: - type: "vault" - path: "secret/data/linode/tokens/monitored" + path: "linode/tokens/monitored" ``` ### Example 4: High Availability Setup diff --git a/helm/latr/values.yaml b/helm/latr/values.yaml index 93bb88b..7103c4f 100644 --- a/helm/latr/values.yaml +++ b/helm/latr/values.yaml @@ -123,7 +123,10 @@ config: # rotationThreshold: 10 # storage: # - type: vault - # path: secret/data/linode/tokens/my-api-token + # # path is relative to config.vault.mountPath (no "data/" prefix) + # path: linode/tokens/my-api-token + # key: token # optional; default "token" + # action: replace # optional; "replace" (default) or "append" (needs AppRole read) # Secrets configuration # These values should be provided via a separate values file or via --set flags diff --git a/internal/config/config.go b/internal/config/config.go index 61f98ee..1ac554c 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -5,6 +5,7 @@ import ( "os" "regexp" "strconv" + "strings" "time" "gopkg.in/yaml.v3" @@ -55,10 +56,48 @@ type TokenConfig struct { Storage []StorageConfig `yaml:"storage"` } +// Storage write actions for Vault KV data maps. +const ( + // StorageActionReplace replaces the entire secret data map with a single key + // (historical default behavior). + StorageActionReplace = "replace" + // StorageActionAppend merges the token key into existing secret data, preserving + // other keys. If the secret does not exist, behaves like replace for that write. + StorageActionAppend = "append" +) + // StorageConfig represents where to store the rotated token type StorageConfig struct { Type string `yaml:"type"` Path string `yaml:"path"` + // Key is the KV v2 data key written/read for the token value. + // Defaults to "token" when empty (ApplyDefaults). Override for consumers + // that expect a different key (e.g. IPService / Salt cutovers). + Key string `yaml:"key,omitempty"` + // Action controls how the key is written into the secret data map: + // - "replace" (default): data map becomes only {key: token} + // (destructive to other data keys on that secret) + // - "append": merge key into existing data with CAS retries; AppRole + // needs read+create+update on the data path + Action string `yaml:"action,omitempty"` +} + +// NormalizeStorageAction trims and lowercases action; empty becomes DefaultStorageAction. +func NormalizeStorageAction(action string) string { + a := strings.ToLower(strings.TrimSpace(action)) + if a == "" { + return DefaultStorageAction + } + return a +} + +// NormalizeStorageKey trims key; empty becomes DefaultStorageKey. +func NormalizeStorageKey(key string) string { + k := strings.TrimSpace(key) + if k == "" { + return DefaultStorageKey + } + return k } // Parse parses YAML configuration data into a Config struct. @@ -88,6 +127,12 @@ func Parse(data []byte) (*Config, error) { return &cfg, nil } +// DefaultStorageKey is the KV v2 data key used when storage.key is omitted. +const DefaultStorageKey = "token" + +// DefaultStorageAction is used when storage.action is omitted. +const DefaultStorageAction = StorageActionReplace + // ApplyDefaults sets default values for optional configuration fields func (c *Config) ApplyDefaults() { if c.Daemon.Mode == "" { @@ -105,6 +150,12 @@ func (c *Config) ApplyDefaults() { if c.Observability.LogLevel == "" { c.Observability.LogLevel = "info" } + for i := range c.Tokens { + for j := range c.Tokens[i].Storage { + c.Tokens[i].Storage[j].Key = NormalizeStorageKey(c.Tokens[i].Storage[j].Key) + c.Tokens[i].Storage[j].Action = NormalizeStorageAction(c.Tokens[i].Storage[j].Action) + } + } } // Validate checks that the configuration is valid @@ -148,6 +199,12 @@ func (c *Config) validateToken(token *TokenConfig, index int) error { return fmt.Errorf("token[%d]: at least one storage backend is required", index) } + for j, storage := range token.Storage { + if err := validateStorage(&storage, index, j, c.Vault.MountPath); err != nil { + return err + } + } + // Validate validity period duration, err := ParseValidityDuration(token.Validity) if err != nil { @@ -163,6 +220,56 @@ func (c *Config) validateToken(token *TokenConfig, index int) error { return nil } +func validateStorage(storage *StorageConfig, tokenIndex, storageIndex int, mountPath string) error { + if storage.Type == "" { + return fmt.Errorf("token[%d].storage[%d]: type is required", tokenIndex, storageIndex) + } + if strings.TrimSpace(storage.Path) == "" { + return fmt.Errorf("token[%d].storage[%d]: path is required", tokenIndex, storageIndex) + } + + // Path is relative to vault.mount_path; client writes {mount}/data/{path}. + // Reject mistaken API-style prefixes, but allow a path segment named "data" + // that is not the first segment (e.g. "team/data/token" is valid). + trimmedPath := strings.Trim(strings.TrimSpace(storage.Path), "/") + if trimmedPath == "data" || strings.HasPrefix(trimmedPath, "data/") { + return fmt.Errorf( + "token[%d].storage[%d]: path %q must be relative to vault.mount_path without a leading \"data/\" prefix (example: \"shared-all/team/token\", not \"data/team/token\")", + tokenIndex, + storageIndex, + storage.Path, + ) + } + mount := strings.Trim(strings.TrimSpace(mountPath), "/") + if mount != "" { + // Catch "infra/data/..." when mount_path is "infra" (double data/ in Vault). + if trimmedPath == mount+"/data" || strings.HasPrefix(trimmedPath, mount+"/data/") { + return fmt.Errorf( + "token[%d].storage[%d]: path %q must not include vault.mount_path %q or the \"data/\" API prefix (use the path under the mount only)", + tokenIndex, + storageIndex, + storage.Path, + mount, + ) + } + } + + switch NormalizeStorageAction(storage.Action) { + case StorageActionReplace, StorageActionAppend: + // ok (empty normalizes to replace) + default: + return fmt.Errorf( + "token[%d].storage[%d]: invalid action %q (want %q or %q)", + tokenIndex, + storageIndex, + storage.Action, + StorageActionReplace, + StorageActionAppend, + ) + } + return nil +} + // ParseValidityDuration parses a validity string (e.g., "90d", "6mo") into a time.Duration func ParseValidityDuration(validity string) (time.Duration, error) { // Support formats: 90d, 6mo, 1h, 30m diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 5198ab1..bcc6384 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -37,7 +37,7 @@ tokens: scopes: "*" storage: - type: "vault" - path: "secret/data/linode/tokens/test" + path: "linode/tokens/test" ` cfg, err := Parse([]byte(yamlContent)) @@ -71,7 +71,7 @@ tokens: assert.Equal(t, "*", token.Scopes) require.Len(t, token.Storage, 1) assert.Equal(t, "vault", token.Storage[0].Type) - assert.Equal(t, "secret/data/linode/tokens/test", token.Storage[0].Path) + assert.Equal(t, "linode/tokens/test", token.Storage[0].Path) } func TestParseConfigWithDefaults(t *testing.T) { @@ -88,7 +88,7 @@ tokens: scopes: "*" storage: - type: "vault" - path: "secret/data/linode/tokens/test" + path: "linode/tokens/test" ` cfg, err := Parse([]byte(yamlContent)) @@ -105,6 +105,164 @@ tokens: assert.Equal(t, 10, cfg.Rotation.ThresholdPercent) assert.Equal(t, "secret", cfg.Vault.MountPath) assert.Equal(t, "info", cfg.Observability.LogLevel) + require.Len(t, cfg.Tokens, 1) + require.Len(t, cfg.Tokens[0].Storage, 1) + assert.Equal(t, DefaultStorageKey, cfg.Tokens[0].Storage[0].Key) + assert.Equal(t, DefaultStorageAction, cfg.Tokens[0].Storage[0].Action) +} + +func TestParseConfig_StorageKey(t *testing.T) { + yamlContent := ` +vault: + address: "https://vault.example.com" + role_id: "test-role-id" + secret_id: "test-secret-id" + +tokens: + - label: "test-token" + team: "platform-team" + validity: "90d" + scopes: "*" + storage: + - type: "vault" + path: "shared-all/sre-compute/ipservice" + key: "api_token" +` + + cfg, err := Parse([]byte(yamlContent)) + require.NoError(t, err) + require.Len(t, cfg.Tokens, 1) + require.Len(t, cfg.Tokens[0].Storage, 1) + assert.Equal(t, "api_token", cfg.Tokens[0].Storage[0].Key) + + cfg.ApplyDefaults() + // Custom key must not be overwritten by defaults + assert.Equal(t, "api_token", cfg.Tokens[0].Storage[0].Key) + assert.Equal(t, DefaultStorageAction, cfg.Tokens[0].Storage[0].Action) +} + +func TestParseConfig_StorageActionAppend(t *testing.T) { + yamlContent := ` +vault: + address: "https://vault.example.com" + role_id: "test-role-id" + secret_id: "test-secret-id" + +tokens: + - label: "test-token" + team: "platform-team" + validity: "90d" + scopes: "*" + storage: + - type: "vault" + path: "shared-all/sre/mixed-secret" + key: "linode_token" + action: "append" +` + + cfg, err := Parse([]byte(yamlContent)) + require.NoError(t, err) + assert.Equal(t, StorageActionAppend, cfg.Tokens[0].Storage[0].Action) + + cfg.ApplyDefaults() + assert.Equal(t, StorageActionAppend, cfg.Tokens[0].Storage[0].Action) + + err = cfg.Validate() + require.NoError(t, err) +} + +func TestValidateConfig_InvalidStorageAction(t *testing.T) { + cfg := &Config{ + Vault: VaultConfig{ + Address: "https://vault.example.com", + RoleID: "role", + SecretID: "secret", + }, + Tokens: []TokenConfig{ + { + Label: "t", + Team: "team", + Validity: "90d", + Scopes: "*", + Storage: []StorageConfig{ + {Type: "vault", Path: "path", Action: "merge"}, + }, + }, + }, + } + + err := cfg.Validate() + require.Error(t, err) + assert.Contains(t, err.Error(), "invalid action") +} + +func TestValidateConfig_StoragePathRules(t *testing.T) { + base := func(path, mount string) *Config { + return &Config{ + Vault: VaultConfig{ + Address: "https://vault.example.com", + RoleID: "role", + SecretID: "secret", + MountPath: mount, + }, + Tokens: []TokenConfig{ + { + Label: "t", + Team: "team", + Validity: "90d", + Scopes: "*", + Storage: []StorageConfig{ + {Type: "vault", Path: path}, + }, + }, + }, + } + } + + t.Run("rejects leading data prefix", func(t *testing.T) { + err := base("data/linode/tokens/x", "secret").Validate() + require.Error(t, err) + assert.Contains(t, err.Error(), "leading") + }) + + t.Run("rejects mount plus data prefix", func(t *testing.T) { + err := base("secret/data/linode/tokens/x", "secret").Validate() + require.Error(t, err) + assert.Contains(t, err.Error(), "must not include vault.mount_path") + }) + + t.Run("allows data as a non-leading path segment", func(t *testing.T) { + err := base("team/data/token", "secret").Validate() + require.NoError(t, err) + }) + + t.Run("allows normal relative path", func(t *testing.T) { + err := base("shared-all/sre-o11y/linode-api", "infra").Validate() + require.NoError(t, err) + }) +} + +func TestNormalizeStorageAction_CaseAndWhitespace(t *testing.T) { + assert.Equal(t, StorageActionReplace, NormalizeStorageAction("")) + assert.Equal(t, StorageActionReplace, NormalizeStorageAction(" Replace ")) + assert.Equal(t, StorageActionAppend, NormalizeStorageAction("APPEND")) +} + +func TestApplyDefaults_NormalizesKeyAndAction(t *testing.T) { + cfg := &Config{ + Tokens: []TokenConfig{ + { + Storage: []StorageConfig{ + {Type: "vault", Path: "p", Key: " api ", Action: " APPEND "}, + }, + }, + }, + } + cfg.ApplyDefaults() + // Key is only trimmed by NormalizeStorageKey; embedded spaces preserved after trim ends + assert.Equal(t, "api", NormalizeStorageKey(" api ")) + assert.Equal(t, "api", cfg.Tokens[0].Storage[0].Key) + assert.Equal(t, StorageActionAppend, cfg.Tokens[0].Storage[0].Action) } func TestValidateConfig_ValidityPeriodTooLong(t *testing.T) { @@ -122,7 +280,7 @@ func TestValidateConfig_ValidityPeriodTooLong(t *testing.T) { Validity: "7mo", // More than 6 months Scopes: "*", Storage: []StorageConfig{ - {Type: "vault", Path: "secret/data/linode/tokens/test"}, + {Type: "vault", Path: "linode/tokens/test"}, }, }, }, @@ -148,7 +306,7 @@ func TestValidateConfig_ValidityPeriodExactly6Months(t *testing.T) { Validity: "180d", // Exactly 6 months Scopes: "*", Storage: []StorageConfig{ - {Type: "vault", Path: "secret/data/linode/tokens/test"}, + {Type: "vault", Path: "linode/tokens/test"}, }, }, }, @@ -324,7 +482,7 @@ tokens: rotation_threshold: 15 storage: - type: "vault" - path: "secret/data/linode/tokens/test" + path: "linode/tokens/test" ` cfg, err := Parse([]byte(yamlContent)) @@ -366,7 +524,7 @@ tokens: scopes: "*" storage: - type: "vault" - path: "secret/data/linode/tokens/test" + path: "linode/tokens/test" ` cfg, err := Parse([]byte(yamlContent)) @@ -394,7 +552,7 @@ tokens: scopes: "*" storage: - type: "vault" - path: "secret/data/linode/tokens/test" + path: "linode/tokens/test" ` cfg, err := Parse([]byte(yamlContent)) diff --git a/internal/config/loader_test.go b/internal/config/loader_test.go index fa06842..5cee3fe 100644 --- a/internal/config/loader_test.go +++ b/internal/config/loader_test.go @@ -28,7 +28,7 @@ tokens: scopes: "*" storage: - type: "vault" - path: "secret/data/linode/tokens/test" + path: "linode/tokens/test" ` err := os.WriteFile(configPath, []byte(configContent), 0644) @@ -70,7 +70,7 @@ tokens: scopes: "*" storage: - type: "vault" - path: "secret/data/linode/tokens/token1" + path: "linode/tokens/token1" ` config2 := ` @@ -81,7 +81,7 @@ tokens: scopes: "linodes:read_only" storage: - type: "vault" - path: "secret/data/linode/tokens/token2" + path: "linode/tokens/token2" ` err := os.WriteFile(filepath.Join(tmpDir, "config1.yaml"), []byte(config1), 0644) @@ -200,7 +200,7 @@ tokens: scopes: "*" storage: - type: "vault" - path: "secret/data/linode/tokens/test" + path: "linode/tokens/test" ` err := os.WriteFile(configPath, []byte(configContent), 0644) @@ -230,7 +230,7 @@ tokens: scopes: "*" storage: - type: "vault" - path: "secret/data/linode/tokens/test" + path: "linode/tokens/test" ` err := os.WriteFile(configPath, []byte(configContent), 0644) diff --git a/internal/observability/telemetry.go b/internal/observability/telemetry.go index 5ff3fd2..f7d9808 100644 --- a/internal/observability/telemetry.go +++ b/internal/observability/telemetry.go @@ -40,6 +40,17 @@ type Metrics struct { RotationDuration metric.Float64Histogram TokenValidityRemaining metric.Float64Gauge VaultStorageErrorsTotal metric.Int64Counter + // VaultWritesTotal counts KV data writes by action (replace|append) and result. + // PromQL: sum by (action, result) (rate(latr_vault_writes_total[5m])) + VaultWritesTotal metric.Int64Counter + // VaultWriteDuration records latency of Vault KV data writes by action. + // PromQL: histogram_quantile(0.99, sum by (le, action) (rate(latr_vault_write_duration_seconds_bucket[5m]))) + VaultWriteDuration metric.Float64Histogram + // VaultAppendCASConflictsTotal counts CAS version mismatches during append + // (each conflict before a successful retry or exhaustion). + // PromQL: sum(rate(latr_vault_append_cas_conflicts_total[5m])) + // Alert when sustained: increase(...[15m]) > N under concurrent writers on shared secrets. + VaultAppendCASConflictsTotal metric.Int64Counter } var ( @@ -180,12 +191,43 @@ func createMetrics(meter metric.Meter) (*Metrics, error) { return nil, fmt.Errorf("failed to create vault_storage_errors_total counter: %w", err) } + // sum by (action, result) (rate(latr_vault_writes_total[5m])) + vaultWritesTotal, err := meter.Int64Counter( + "latr_vault_writes_total", + metric.WithDescription("Total Vault KV data writes by storage action and result"), + ) + if err != nil { + return nil, fmt.Errorf("failed to create vault_writes_total counter: %w", err) + } + + // histogram_quantile(0.99, sum by (le, action) (rate(latr_vault_write_duration_seconds_bucket[5m]))) + vaultWriteDuration, err := meter.Float64Histogram( + "latr_vault_write_duration_seconds", + metric.WithDescription("Duration of Vault KV data write operations"), + metric.WithUnit("s"), + ) + if err != nil { + return nil, fmt.Errorf("failed to create vault_write_duration histogram: %w", err) + } + + // sum(rate(latr_vault_append_cas_conflicts_total[5m])) + vaultAppendCASConflictsTotal, err := meter.Int64Counter( + "latr_vault_append_cas_conflicts_total", + metric.WithDescription("Vault KV check-and-set conflicts while appending a token key"), + ) + if err != nil { + return nil, fmt.Errorf("failed to create vault_append_cas_conflicts_total counter: %w", err) + } + return &Metrics{ - TokensTotal: tokensTotal, - RotationsTotal: rotationsTotal, - RotationDuration: rotationDuration, - TokenValidityRemaining: tokenValidityRemaining, - VaultStorageErrorsTotal: vaultStorageErrorsTotal, + TokensTotal: tokensTotal, + RotationsTotal: rotationsTotal, + RotationDuration: rotationDuration, + TokenValidityRemaining: tokenValidityRemaining, + VaultStorageErrorsTotal: vaultStorageErrorsTotal, + VaultWritesTotal: vaultWritesTotal, + VaultWriteDuration: vaultWriteDuration, + VaultAppendCASConflictsTotal: vaultAppendCASConflictsTotal, }, nil } @@ -259,16 +301,59 @@ func RecordTokenValidityRemaining(ctx context.Context, label, team string, secon ) } -// RecordVaultStorageError records a Vault storage error -func RecordVaultStorageError(ctx context.Context, path string) { +// RecordVaultStorageError records a Vault storage error. +// action should be a low-cardinality value (replace|append|state|unknown). +func RecordVaultStorageError(ctx context.Context, path, action string) { if globalMetrics == nil { return } + if action == "" { + action = "unknown" + } globalMetrics.VaultStorageErrorsTotal.Add(ctx, 1, - metric.WithAttributes(attribute.String("path", path)), + metric.WithAttributes( + attribute.String("path", path), + attribute.String("action", action), + ), ) } +// RecordVaultWrite records a Vault KV data write attempt. +// action: replace|append; success selects result=success|error. +// +// sum by (action, result) (rate(latr_vault_writes_total[5m])) +// histogram_quantile(0.99, sum by (le, action) (rate(latr_vault_write_duration_seconds_bucket[5m]))) +func RecordVaultWrite(ctx context.Context, action string, success bool, duration time.Duration) { + if globalMetrics == nil { + return + } + if action == "" { + action = "unknown" + } + result := "success" + if !success { + result = "error" + } + attrs := metric.WithAttributes( + attribute.String("action", action), + attribute.String("result", result), + ) + globalMetrics.VaultWritesTotal.Add(ctx, 1, attrs) + globalMetrics.VaultWriteDuration.Record(ctx, duration.Seconds(), + metric.WithAttributes(attribute.String("action", action)), + ) +} + +// RecordVaultAppendCASConflict records one CAS version mismatch during append. +// +// sum(rate(latr_vault_append_cas_conflicts_total[5m])) +func RecordVaultAppendCASConflict(ctx context.Context) { + if globalMetrics == nil { + return + } + globalMetrics.VaultAppendCASConflictsTotal.Add(ctx, 1) +} + // TraceAttrs extracts OpenTelemetry trace context attributes for structured logging // Returns attributes as []any for use with slog methods func TraceAttrs(ctx context.Context) []any { diff --git a/internal/rotation/engine.go b/internal/rotation/engine.go index d180843..0afc302 100644 --- a/internal/rotation/engine.go +++ b/internal/rotation/engine.go @@ -21,8 +21,11 @@ type LinodeClient interface { // VaultClient defines the interface for Vault operations type VaultClient interface { - WriteToken(ctx context.Context, path, token string) error - ReadToken(ctx context.Context, path string) (string, error) + // WriteToken stores token under the given KV data key (empty key means "token"). + // action is "replace" (default) or "append" (merge into existing secret data). + WriteToken(ctx context.Context, path, token, key, action string) error + // ReadToken reads token from the given KV data key (empty key means "token"). + ReadToken(ctx context.Context, path, key string) (string, error) WriteTokenState(ctx context.Context, path string, state *models.TokenState) error ReadTokenState(ctx context.Context, path string) (*models.TokenState, error) } @@ -55,11 +58,13 @@ func (e *Engine) ProcessToken(ctx context.Context, tokenConfig config.TokenConfi span.SetAttributes( attribute.String("token.label", tokenConfig.Label), attribute.String("token.team", tokenConfig.Team), + attribute.Int("token.storage_count", len(tokenConfig.Storage)), ) attrs := append([]any{ slog.String("token_label", tokenConfig.Label), slog.String("team", tokenConfig.Team), + slog.Int("storage_backends", len(tokenConfig.Storage)), }, observability.TraceAttrs(ctx)...) logger.InfoContext(ctx, "Processing token", attrs...) @@ -183,7 +188,7 @@ func (e *Engine) createNewToken(ctx context.Context, tokenConfig config.TokenCon span.SetStatus(codes.Error, "failed to store token") observability.RecordRotation(ctx, tokenConfig.Label, tokenConfig.Team, false) observability.RecordRotationDuration(ctx, tokenConfig.Label, tokenConfig.Team, time.Since(startTime)) - observability.RecordVaultStorageError(ctx, storagePath) + // Vault write metrics/errors recorded inside vault.Client.WriteToken return fmt.Errorf("failed to store token in vault: %w", err) } @@ -275,7 +280,7 @@ func (e *Engine) rotateToken(ctx context.Context, tokenConfig config.TokenConfig span.SetStatus(codes.Error, "failed to store token") observability.RecordRotation(ctx, tokenConfig.Label, tokenConfig.Team, false) observability.RecordRotationDuration(ctx, tokenConfig.Label, tokenConfig.Team, time.Since(startTime)) - observability.RecordVaultStorageError(ctx, storagePath) + // Vault write metrics/errors recorded inside vault.Client.WriteToken return fmt.Errorf("failed to store token in vault: %w", err) } @@ -301,16 +306,23 @@ func (e *Engine) storeTokenInBackends(ctx context.Context, storageConfigs []conf logger := observability.GetLogger() for _, storage := range storageConfigs { - if storage.Type == "vault" { - if err := e.vaultClient.WriteToken(ctx, storage.Path, token); err != nil { - return err - } - attrs := append([]any{ - slog.String("storage_type", "vault"), - slog.String("vault_path", storage.Path), - }, observability.TraceAttrs(ctx)...) - logger.InfoContext(ctx, "Stored token in Vault", attrs...) + if storage.Type != "vault" { + continue } + + key := config.NormalizeStorageKey(storage.Key) + action := config.NormalizeStorageAction(storage.Action) + + if err := e.vaultClient.WriteToken(ctx, storage.Path, token, key, action); err != nil { + return err + } + attrs := append([]any{ + slog.String("storage_type", "vault"), + slog.String("vault_path", storage.Path), + slog.String("vault_key", key), + slog.String("vault_action", action), + }, observability.TraceAttrs(ctx)...) + logger.InfoContext(ctx, "Stored token in Vault", attrs...) } return nil } diff --git a/internal/rotation/engine_test.go b/internal/rotation/engine_test.go index 7891df1..9e2184d 100644 --- a/internal/rotation/engine_test.go +++ b/internal/rotation/engine_test.go @@ -6,11 +6,11 @@ import ( "testing" "time" + "github.com/linode-obs/latr/internal/config" + "github.com/linode-obs/latr/pkg/models" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" - "github.com/linode-obs/latr/internal/config" - "github.com/linode-obs/latr/pkg/models" ) // MockLinodeClient is a mock implementation of the Linode client @@ -39,13 +39,13 @@ type MockVaultClient struct { mock.Mock } -func (m *MockVaultClient) WriteToken(ctx context.Context, path, token string) error { - args := m.Called(ctx, path, token) +func (m *MockVaultClient) WriteToken(ctx context.Context, path, token, key, action string) error { + args := m.Called(ctx, path, token, key, action) return args.Error(0) } -func (m *MockVaultClient) ReadToken(ctx context.Context, path string) (string, error) { - args := m.Called(ctx, path) +func (m *MockVaultClient) ReadToken(ctx context.Context, path, key string) (string, error) { + args := m.Called(ctx, path, key) return args.String(0), args.Error(1) } @@ -72,7 +72,7 @@ func TestEngine_ProcessToken_NewToken(t *testing.T) { Validity: "90d", Scopes: "*", Storage: []config.StorageConfig{ - {Type: "vault", Path: "secret/data/test/new-token"}, + {Type: "vault", Path: "test/new-token"}, }, } @@ -92,9 +92,9 @@ func TestEngine_ProcessToken_NewToken(t *testing.T) { mockLinode.On("CreateToken", mock.Anything, "new-token", "*", mock.Anything).Return(createdToken, nil) // Vault operations - mockVault.On("ReadTokenState", mock.Anything, "secret/data/test/new-token").Return(nil, nil) - mockVault.On("WriteToken", mock.Anything, "secret/data/test/new-token", "new-secret-token").Return(nil) - mockVault.On("WriteTokenState", mock.Anything, "secret/data/test/new-token", mock.Anything).Return(nil) + mockVault.On("ReadTokenState", mock.Anything, "test/new-token").Return(nil, nil) + mockVault.On("WriteToken", mock.Anything, "test/new-token", "new-secret-token", "token", "replace").Return(nil) + mockVault.On("WriteTokenState", mock.Anything, "test/new-token", mock.Anything).Return(nil) engine := &Engine{ linodeClient: mockLinode, @@ -120,7 +120,7 @@ func TestEngine_ProcessToken_ExistingToken_NoRotationNeeded(t *testing.T) { Validity: "90d", Scopes: "*", Storage: []config.StorageConfig{ - {Type: "vault", Path: "secret/data/test/existing-token"}, + {Type: "vault", Path: "test/existing-token"}, }, } @@ -161,7 +161,7 @@ func TestEngine_ProcessToken_ExistingToken_NeedsRotation(t *testing.T) { Validity: "90d", Scopes: "*", Storage: []config.StorageConfig{ - {Type: "vault", Path: "secret/data/test/existing-token"}, + {Type: "vault", Path: "test/existing-token"}, }, } @@ -195,9 +195,9 @@ func TestEngine_ProcessToken_ExistingToken_NeedsRotation(t *testing.T) { mockLinode.On("FindTokenByLabel", mock.Anything, "existing-token").Return(existingToken, nil) mockLinode.On("CreateToken", mock.Anything, "existing-token", "*", mock.Anything).Return(newToken, nil) - mockVault.On("ReadTokenState", mock.Anything, "secret/data/test/existing-token").Return(existingState, nil) - mockVault.On("WriteToken", mock.Anything, "secret/data/test/existing-token", "new-rotated-token").Return(nil) - mockVault.On("WriteTokenState", mock.Anything, "secret/data/test/existing-token", mock.MatchedBy(func(state *models.TokenState) bool { + mockVault.On("ReadTokenState", mock.Anything, "test/existing-token").Return(existingState, nil) + mockVault.On("WriteToken", mock.Anything, "test/existing-token", "new-rotated-token", "token", "replace").Return(nil) + mockVault.On("WriteTokenState", mock.Anything, "test/existing-token", mock.MatchedBy(func(state *models.TokenState) bool { return state.CurrentLinodeID == 456 && state.PreviousLinodeID == 123 && state.RotationCount == 1 @@ -227,7 +227,7 @@ func TestEngine_ProcessToken_DryRunMode(t *testing.T) { Validity: "90d", Scopes: "*", Storage: []config.StorageConfig{ - {Type: "vault", Path: "secret/data/test/dry-run-token"}, + {Type: "vault", Path: "test/dry-run-token"}, }, } @@ -260,7 +260,7 @@ func TestEngine_ProcessToken_LinodeCreateFails(t *testing.T) { Validity: "90d", Scopes: "*", Storage: []config.StorageConfig{ - {Type: "vault", Path: "secret/data/test/new-token"}, + {Type: "vault", Path: "test/new-token"}, }, } @@ -268,7 +268,7 @@ func TestEngine_ProcessToken_LinodeCreateFails(t *testing.T) { mockLinode.On("FindTokenByLabel", mock.Anything, "new-token").Return(nil, nil) mockLinode.On("CreateToken", mock.Anything, "new-token", "*", mock.Anything).Return(nil, errors.New("API error")) - mockVault.On("ReadTokenState", mock.Anything, "secret/data/test/new-token").Return(nil, nil) + mockVault.On("ReadTokenState", mock.Anything, "test/new-token").Return(nil, nil) engine := &Engine{ linodeClient: mockLinode, @@ -296,7 +296,7 @@ func TestEngine_ProcessToken_VaultWriteFails_StateTracked(t *testing.T) { Validity: "90d", Scopes: "*", Storage: []config.StorageConfig{ - {Type: "vault", Path: "secret/data/test/new-token"}, + {Type: "vault", Path: "test/new-token"}, }, } @@ -314,10 +314,10 @@ func TestEngine_ProcessToken_VaultWriteFails_StateTracked(t *testing.T) { mockLinode.On("FindTokenByLabel", mock.Anything, "new-token").Return(nil, nil) mockLinode.On("CreateToken", mock.Anything, "new-token", "*", mock.Anything).Return(createdToken, nil) - mockVault.On("ReadTokenState", mock.Anything, "secret/data/test/new-token").Return(nil, nil) - mockVault.On("WriteToken", mock.Anything, "secret/data/test/new-token", "new-secret-token").Return(errors.New("vault error")) + mockVault.On("ReadTokenState", mock.Anything, "test/new-token").Return(nil, nil) + mockVault.On("WriteToken", mock.Anything, "test/new-token", "new-secret-token", "token", "replace").Return(errors.New("vault error")) // State should still be written to track that we need to retry Vault write - mockVault.On("WriteTokenState", mock.Anything, "secret/data/test/new-token", mock.Anything).Return(nil) + mockVault.On("WriteTokenState", mock.Anything, "test/new-token", mock.Anything).Return(nil) engine := &Engine{ linodeClient: mockLinode, @@ -333,3 +333,92 @@ func TestEngine_ProcessToken_VaultWriteFails_StateTracked(t *testing.T) { mockLinode.AssertExpectations(t) mockVault.AssertExpectations(t) } + +func TestEngine_ProcessToken_CustomStorageKey(t *testing.T) { + mockLinode := new(MockLinodeClient) + mockVault := new(MockVaultClient) + + tokenConfig := config.TokenConfig{ + Label: "custom-key-token", + Team: "platform", + Validity: "90d", + Scopes: "*", + Storage: []config.StorageConfig{ + {Type: "vault", Path: "shared-all/sre/ipservice", Key: "api_token"}, + }, + } + + now := time.Now() + createdToken := &models.Token{ + ID: 456, + Label: "custom-key-token", + Token: "secret-value", + CreatedAt: now, + ExpiresAt: now.Add(90 * 24 * time.Hour), + Scopes: "*", + Validity: 90 * 24 * time.Hour, + } + + mockLinode.On("FindTokenByLabel", mock.Anything, "custom-key-token").Return(nil, nil) + mockLinode.On("CreateToken", mock.Anything, "custom-key-token", "*", mock.Anything).Return(createdToken, nil) + mockVault.On("ReadTokenState", mock.Anything, "shared-all/sre/ipservice").Return(nil, nil) + mockVault.On("WriteToken", mock.Anything, "shared-all/sre/ipservice", "secret-value", "api_token", "replace").Return(nil) + mockVault.On("WriteTokenState", mock.Anything, "shared-all/sre/ipservice", mock.Anything).Return(nil) + + engine := &Engine{ + linodeClient: mockLinode, + vaultClient: mockVault, + dryRun: false, + } + + err := engine.ProcessToken(context.Background(), tokenConfig, 10) + require.NoError(t, err) + mockVault.AssertExpectations(t) +} + +func TestEngine_ProcessToken_AppendAction(t *testing.T) { + mockLinode := new(MockLinodeClient) + mockVault := new(MockVaultClient) + + tokenConfig := config.TokenConfig{ + Label: "append-token", + Team: "platform", + Validity: "90d", + Scopes: "*", + Storage: []config.StorageConfig{ + { + Type: "vault", + Path: "shared-all/sre/mixed", + Key: "linode_token", + Action: config.StorageActionAppend, + }, + }, + } + + now := time.Now() + createdToken := &models.Token{ + ID: 789, + Label: "append-token", + Token: "appended-value", + CreatedAt: now, + ExpiresAt: now.Add(90 * 24 * time.Hour), + Scopes: "*", + Validity: 90 * 24 * time.Hour, + } + + mockLinode.On("FindTokenByLabel", mock.Anything, "append-token").Return(nil, nil) + mockLinode.On("CreateToken", mock.Anything, "append-token", "*", mock.Anything).Return(createdToken, nil) + mockVault.On("ReadTokenState", mock.Anything, "shared-all/sre/mixed").Return(nil, nil) + mockVault.On("WriteToken", mock.Anything, "shared-all/sre/mixed", "appended-value", "linode_token", "append").Return(nil) + mockVault.On("WriteTokenState", mock.Anything, "shared-all/sre/mixed", mock.Anything).Return(nil) + + engine := &Engine{ + linodeClient: mockLinode, + vaultClient: mockVault, + dryRun: false, + } + + err := engine.ProcessToken(context.Background(), tokenConfig, 10) + require.NoError(t, err) + mockVault.AssertExpectations(t) +} diff --git a/internal/vault/client.go b/internal/vault/client.go index 0cb635a..18105e7 100644 --- a/internal/vault/client.go +++ b/internal/vault/client.go @@ -2,12 +2,19 @@ package vault import ( "context" + "encoding/json" "fmt" + "log/slog" "strconv" + "strings" "time" "github.com/hashicorp/vault/api" + "github.com/linode-obs/latr/internal/observability" "github.com/linode-obs/latr/pkg/models" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" + "go.opentelemetry.io/otel/trace" ) // Config holds Vault client configuration @@ -65,27 +72,287 @@ func authenticateAppRole(client *api.Client, roleID, secretID string) error { return nil } -// WriteToken writes a token value to a KV v2 path -func (c *Client) WriteToken(ctx context.Context, path string, token string) error { +// defaultDataKey is used when WriteToken/ReadToken are called with an empty key. +const defaultDataKey = "token" + +// Write actions for KV secret data maps. +const ( + WriteActionReplace = "replace" + WriteActionAppend = "append" +) + +// maxAppendCASAttempts bounds read-modify-write retries when check-and-set conflicts. +const maxAppendCASAttempts = 5 + +// resolveDataKey returns key, or defaultDataKey when key is empty after trim. +func resolveDataKey(key string) string { + key = strings.TrimSpace(key) + if key == "" { + return defaultDataKey + } + return key +} + +// resolveWriteAction returns a valid write action; empty defaults to replace. +// Comparison is case-insensitive after trim. +func resolveWriteAction(action string) (string, error) { + switch strings.ToLower(strings.TrimSpace(action)) { + case "", WriteActionReplace: + return WriteActionReplace, nil + case WriteActionAppend: + return WriteActionAppend, nil + default: + return "", fmt.Errorf("invalid write action %q (want %q or %q)", action, WriteActionReplace, WriteActionAppend) + } +} + +// WriteToken writes a token value to a KV v2 path under the given data key. +// +// Path is relative to mount_path (do not include the mount name or a "data/" +// prefix). The client writes to "{mount_path}/data/{path}". +// +// - key empty → stored under "token" +// - action "replace" (default): secret data map becomes only {key: token}. +// Other data keys on that secret are removed. Custom metadata is untouched. +// - action "append": merge key into existing data via read-modify-write with +// KV v2 check-and-set (CAS). Other keys are preserved. Missing secret is +// created with just this key. AppRole policy must allow read+create+update +// on the data path. +// +// Append retries on CAS conflict up to maxAppendCASAttempts times. +func (c *Client) WriteToken(ctx context.Context, path, token, key, action string) error { + start := time.Now() + tracer := observability.GetTracer() + ctx, span := tracer.Start(ctx, "Vault.WriteToken") + defer span.End() + + dataKey := resolveDataKey(key) + + writeAction, err := resolveWriteAction(action) + if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, "invalid action") + return err + } + + span.SetAttributes( + attribute.String("vault.mount_path", c.mountPath), + attribute.String("vault.path", path), + attribute.String("vault.key", dataKey), + attribute.String("vault.action", writeAction), + ) + fullPath := fmt.Sprintf("%s/data/%s", c.mountPath, path) - data := map[string]interface{}{ - "data": map[string]interface{}{ - "token": token, - }, + var writeErr error + switch writeAction { + case WriteActionReplace: + writeErr = c.writeSecretData(ctx, fullPath, map[string]interface{}{ + dataKey: token, + }, nil) + case WriteActionAppend: + writeErr = c.writeTokenAppend(ctx, fullPath, path, dataKey, token) + default: + // Unreachable when resolveWriteAction is used; keep for exhaustiveness. + writeErr = fmt.Errorf("unsupported write action %q", writeAction) + } + + observability.RecordVaultWrite(ctx, writeAction, writeErr == nil, time.Since(start)) + if writeErr != nil { + span.RecordError(writeErr) + span.SetStatus(codes.Error, "vault write failed") + // path label matches existing storage-error series; action is bounded. + observability.RecordVaultStorageError(ctx, path, writeAction) + return writeErr + } + + span.SetStatus(codes.Ok, "ok") + return nil +} + +// writeTokenAppend merges token into an existing KV v2 secret using CAS retries. +// configPath is the mount-relative path for logs (not the full /data/ API path). +func (c *Client) writeTokenAppend(ctx context.Context, fullPath, configPath, dataKey, token string) error { + logger := observability.GetLogger() + var lastErr error + + for attempt := 0; attempt < maxAppendCASAttempts; attempt++ { + snap, err := c.readSecretSnapshot(ctx, fullPath) + if err != nil { + return err + } + + secretData := make(map[string]interface{}, len(snap.data)+1) + for k, v := range snap.data { + secretData[k] = v + } + secretData[dataKey] = token + + var cas *int + if snap.exists { + v := snap.version + cas = &v + } else { + // cas=0: only create if the secret does not already exist + zero := 0 + cas = &zero + } + + err = c.writeSecretData(ctx, fullPath, secretData, cas) + if err == nil { + if attempt > 0 { + attrs := append([]any{ + slog.String("vault_path", configPath), + slog.String("vault_key", dataKey), + slog.Int("cas_attempts", attempt+1), + slog.Int("cas_version", snap.version), + }, observability.TraceAttrs(ctx)...) + logger.InfoContext(ctx, "Vault append succeeded after CAS retry", attrs...) + } + return nil + } + if !isCASConflict(err) { + return err + } + + observability.RecordVaultAppendCASConflict(ctx) + if span := trace.SpanFromContext(ctx); span.IsRecording() { + span.AddEvent("vault.append.cas_conflict", + trace.WithAttributes( + attribute.Int("attempt", attempt+1), + attribute.Int("cas_version", snap.version), + ), + ) + } + + attrs := append([]any{ + slog.String("vault_path", configPath), + slog.String("vault_key", dataKey), + slog.Int("attempt", attempt+1), + slog.Int("max_attempts", maxAppendCASAttempts), + slog.Int("cas_version", snap.version), + slog.Any("error", err), + }, observability.TraceAttrs(ctx)...) + logger.WarnContext(ctx, "Vault append CAS conflict; retrying", attrs...) + + lastErr = err + } + + return fmt.Errorf("failed to append token after %d CAS attempts: %w", maxAppendCASAttempts, lastErr) +} + +// writeSecretData writes a KV v2 data map. When cas is non-nil, sets options.cas. +func (c *Client) writeSecretData(ctx context.Context, fullPath string, secretData map[string]interface{}, cas *int) error { + payload := map[string]interface{}{ + "data": secretData, + } + if cas != nil { + payload["options"] = map[string]interface{}{ + "cas": *cas, + } } - _, err := c.client.Logical().WriteWithContext(ctx, fullPath, data) + _, err := c.client.Logical().WriteWithContext(ctx, fullPath, payload) if err != nil { return fmt.Errorf("failed to write token to vault: %w", err) } - return nil } -// ReadToken reads a token value from a KV v2 path -func (c *Client) ReadToken(ctx context.Context, path string) (string, error) { +// secretSnapshot is the current KV v2 data map and version for CAS writes. +type secretSnapshot struct { + data map[string]interface{} + version int + exists bool +} + +// readSecretSnapshot returns the current secret data and version, or exists=false if missing. +func (c *Client) readSecretSnapshot(ctx context.Context, fullPath string) (*secretSnapshot, error) { + secret, err := c.client.Logical().ReadWithContext(ctx, fullPath) + if err != nil { + return nil, fmt.Errorf("failed to read vault secret for append: %w", err) + } + if secret == nil || secret.Data == nil { + return &secretSnapshot{exists: false, version: 0, data: map[string]interface{}{}}, nil + } + + // KV v2 read response: { "data": { ...keys }, "metadata": { "version": N, ... } } + // When the secret is missing, Logical.Read typically returns nil, nil. + rawData, hasData := secret.Data["data"] + if !hasData || rawData == nil { + return &secretSnapshot{exists: false, version: 0, data: map[string]interface{}{}}, nil + } + + data, ok := rawData.(map[string]interface{}) + if !ok { + return nil, fmt.Errorf("invalid vault secret data structure at %s", fullPath) + } + + version := 0 + if meta, ok := secret.Data["metadata"].(map[string]interface{}); ok { + version = parseMetadataVersion(meta["version"]) + } + + // Copy so we do not mutate the map owned by the API response. + out := make(map[string]interface{}, len(data)+1) + for k, v := range data { + out[k] = v + } + + return &secretSnapshot{ + data: out, + version: version, + exists: true, + }, nil +} + +// parseMetadataVersion normalizes Vault metadata version values from JSON decode. +func parseMetadataVersion(v interface{}) int { + switch n := v.(type) { + case json.Number: + i, err := n.Int64() + if err != nil { + return 0 + } + return int(i) + case float64: + return int(n) + case int: + return n + case int64: + return int(n) + case string: + i, err := strconv.Atoi(n) + if err != nil { + return 0 + } + return i + default: + return 0 + } +} + +// isCASConflict reports whether err looks like a KV v2 check-and-set version +// mismatch (safe to re-read and retry). It does not treat "cas required" as a +// conflict: that means CAS was omitted when the mount requires it, and retrying +// the same request will not help (append always sends options.cas). +func isCASConflict(err error) bool { + if err == nil { + return false + } + msg := strings.ToLower(err.Error()) + // Vault KV v2 version mismatch, e.g.: + // "check-and-set parameter did not match the current version" + return strings.Contains(msg, "check-and-set") || + strings.Contains(msg, "did not match the current version") +} + +// ReadToken reads a token value from a KV v2 path using the given data key. +// If key is empty, the value is read from "token" (backward compatible). +// Path is relative to mount_path (same rules as WriteToken). +func (c *Client) ReadToken(ctx context.Context, path string, key string) (string, error) { fullPath := fmt.Sprintf("%s/data/%s", c.mountPath, path) + dataKey := resolveDataKey(key) secret, err := c.client.Logical().ReadWithContext(ctx, fullPath) if err != nil { @@ -101,9 +368,9 @@ func (c *Client) ReadToken(ctx context.Context, path string) (string, error) { return "", fmt.Errorf("invalid data structure at path: %s", path) } - tokenValue, ok := data["token"].(string) + tokenValue, ok := data[dataKey].(string) if !ok { - return "", fmt.Errorf("token value not found at path: %s", path) + return "", fmt.Errorf("token value not found at path %s key %q", path, dataKey) } return tokenValue, nil diff --git a/internal/vault/client_test.go b/internal/vault/client_test.go index f0206db..c3754b6 100644 --- a/internal/vault/client_test.go +++ b/internal/vault/client_test.go @@ -9,9 +9,9 @@ import ( "testing" "time" + "github.com/linode-obs/latr/pkg/models" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" - "github.com/linode-obs/latr/pkg/models" ) func TestNewClient_AppRoleAuth(t *testing.T) { @@ -23,7 +23,7 @@ func TestNewClient_AppRoleAuth(t *testing.T) { w.WriteHeader(http.StatusOK) response := map[string]interface{}{ "auth": map[string]interface{}{ - "client_token": "test-token", + "client_token": "test-token", "lease_duration": 3600, }, } @@ -76,7 +76,7 @@ func TestWriteToken(t *testing.T) { w.WriteHeader(http.StatusOK) response := map[string]interface{}{ "auth": map[string]interface{}{ - "client_token": "test-token", + "client_token": "test-token", "lease_duration": 3600, }, } @@ -113,16 +113,370 @@ func TestWriteToken(t *testing.T) { require.NoError(t, err) ctx := context.Background() - err = client.WriteToken(ctx, "test/path", "my-secret-token") + err = client.WriteToken(ctx, "test/path", "my-secret-token", "", "") require.NoError(t, err) assert.Equal(t, 1, writeCount) assert.NotNil(t, lastWrittenData) - // Verify the data structure + // Verify the data structure (empty key defaults to "token"; empty action = replace) data, ok := lastWrittenData["data"].(map[string]interface{}) require.True(t, ok) assert.Equal(t, "my-secret-token", data["token"]) + assert.Len(t, data, 1) +} + +func TestWriteToken_CustomKey(t *testing.T) { + var lastWrittenData map[string]interface{} + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/v1/auth/approle/login" { + w.WriteHeader(http.StatusOK) + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "auth": map[string]interface{}{ + "client_token": "test-token", + "lease_duration": 3600, + }, + }) + return + } + + if r.URL.Path == "/v1/secret/data/test/path" && (r.Method == "POST" || r.Method == "PUT") { + var payload map[string]interface{} + _ = json.NewDecoder(r.Body).Decode(&payload) + lastWrittenData = payload + w.WriteHeader(http.StatusOK) + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "data": map[string]interface{}{"version": 1}, + }) + return + } + + w.WriteHeader(http.StatusNotFound) + })) + defer server.Close() + + client, err := NewClient(&Config{ + Address: server.URL, + RoleID: "test-role-id", + SecretID: "test-secret-id", + MountPath: "secret", + }) + require.NoError(t, err) + + err = client.WriteToken(context.Background(), "test/path", "custom-value", "linode-token", WriteActionReplace) + require.NoError(t, err) + + data, ok := lastWrittenData["data"].(map[string]interface{}) + require.True(t, ok) + assert.Equal(t, "custom-value", data["linode-token"]) + assert.Nil(t, data["token"]) +} + +func TestWriteToken_AppendPreservesOtherKeys(t *testing.T) { + var lastWrittenData map[string]interface{} + readCount := 0 + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/v1/auth/approle/login" { + w.WriteHeader(http.StatusOK) + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "auth": map[string]interface{}{ + "client_token": "test-token", + "lease_duration": 3600, + }, + }) + return + } + + if r.URL.Path == "/v1/secret/data/test/path" && r.Method == "GET" { + readCount++ + w.WriteHeader(http.StatusOK) + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "data": map[string]interface{}{ + "data": map[string]interface{}{ + "other": "keep-me", + "token": "old-token", + }, + "metadata": map[string]interface{}{"version": 3}, + }, + }) + return + } + + if r.URL.Path == "/v1/secret/data/test/path" && (r.Method == "POST" || r.Method == "PUT") { + var payload map[string]interface{} + _ = json.NewDecoder(r.Body).Decode(&payload) + lastWrittenData = payload + w.WriteHeader(http.StatusOK) + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "data": map[string]interface{}{"version": 4}, + }) + return + } + + w.WriteHeader(http.StatusNotFound) + })) + defer server.Close() + + client, err := NewClient(&Config{ + Address: server.URL, + RoleID: "test-role-id", + SecretID: "test-secret-id", + MountPath: "secret", + }) + require.NoError(t, err) + + err = client.WriteToken(context.Background(), "test/path", "new-token", "token", WriteActionAppend) + require.NoError(t, err) + + assert.Equal(t, 1, readCount) + data, ok := lastWrittenData["data"].(map[string]interface{}) + require.True(t, ok) + assert.Equal(t, "new-token", data["token"]) + assert.Equal(t, "keep-me", data["other"]) + // CAS must pin the version observed on read + opts, ok := lastWrittenData["options"].(map[string]interface{}) + require.True(t, ok, "append write should include options.cas") + assert.EqualValues(t, 3, opts["cas"]) +} + +func TestWriteToken_ReplaceDropsOtherKeys(t *testing.T) { + var lastWrittenData map[string]interface{} + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/v1/auth/approle/login" { + w.WriteHeader(http.StatusOK) + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "auth": map[string]interface{}{ + "client_token": "test-token", + "lease_duration": 3600, + }, + }) + return + } + + // replace must NOT read first + if r.URL.Path == "/v1/secret/data/test/path" && r.Method == "GET" { + t.Error("replace action should not read existing secret") + w.WriteHeader(http.StatusInternalServerError) + return + } + + if r.URL.Path == "/v1/secret/data/test/path" && (r.Method == "POST" || r.Method == "PUT") { + var payload map[string]interface{} + _ = json.NewDecoder(r.Body).Decode(&payload) + lastWrittenData = payload + w.WriteHeader(http.StatusOK) + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "data": map[string]interface{}{"version": 1}, + }) + return + } + + w.WriteHeader(http.StatusNotFound) + })) + defer server.Close() + + client, err := NewClient(&Config{ + Address: server.URL, + RoleID: "test-role-id", + SecretID: "test-secret-id", + MountPath: "secret", + }) + require.NoError(t, err) + + err = client.WriteToken(context.Background(), "test/path", "only", "token", WriteActionReplace) + require.NoError(t, err) + + data, ok := lastWrittenData["data"].(map[string]interface{}) + require.True(t, ok) + assert.Equal(t, "only", data["token"]) + assert.Len(t, data, 1) + assert.Nil(t, lastWrittenData["options"], "replace should not set CAS options") +} + +func TestWriteToken_AppendCASConflictRetries(t *testing.T) { + writeAttempts := 0 + version := 1 + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/v1/auth/approle/login" { + w.WriteHeader(http.StatusOK) + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "auth": map[string]interface{}{ + "client_token": "test-token", + "lease_duration": 3600, + }, + }) + return + } + + if r.URL.Path == "/v1/secret/data/test/path" && r.Method == "GET" { + w.WriteHeader(http.StatusOK) + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "data": map[string]interface{}{ + "data": map[string]interface{}{ + "other": "keep", + "token": "old", + }, + "metadata": map[string]interface{}{"version": version}, + }, + }) + return + } + + if r.URL.Path == "/v1/secret/data/test/path" && (r.Method == "POST" || r.Method == "PUT") { + writeAttempts++ + if writeAttempts == 1 { + // Simulate concurrent writer winning first CAS + version = 2 + w.WriteHeader(http.StatusBadRequest) + _, _ = fmt.Fprint(w, `{"errors":["check-and-set parameter did not match the current version"]}`) + return + } + w.WriteHeader(http.StatusOK) + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "data": map[string]interface{}{"version": version + 1}, + }) + return + } + + w.WriteHeader(http.StatusNotFound) + })) + defer server.Close() + + client, err := NewClient(&Config{ + Address: server.URL, + RoleID: "test-role-id", + SecretID: "test-secret-id", + MountPath: "secret", + }) + require.NoError(t, err) + + err = client.WriteToken(context.Background(), "test/path", "new", "token", "APPEND") + require.NoError(t, err) + assert.Equal(t, 2, writeAttempts) +} + +func TestWriteToken_ActionCaseInsensitive(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/v1/auth/approle/login" { + w.WriteHeader(http.StatusOK) + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "auth": map[string]interface{}{ + "client_token": "test-token", + "lease_duration": 3600, + }, + }) + return + } + if r.URL.Path == "/v1/secret/data/p" && (r.Method == "POST" || r.Method == "PUT") { + w.WriteHeader(http.StatusOK) + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "data": map[string]interface{}{"version": 1}, + }) + return + } + w.WriteHeader(http.StatusNotFound) + })) + defer server.Close() + + client, err := NewClient(&Config{ + Address: server.URL, RoleID: "r", SecretID: "s", MountPath: "secret", + }) + require.NoError(t, err) + + err = client.WriteToken(context.Background(), "p", "v", "token", " Replace ") + require.NoError(t, err) +} + +func TestWriteToken_AppendWhenMissingCreatesKey(t *testing.T) { + var lastWrittenData map[string]interface{} + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/v1/auth/approle/login" { + w.WriteHeader(http.StatusOK) + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "auth": map[string]interface{}{ + "client_token": "test-token", + "lease_duration": 3600, + }, + }) + return + } + + // No secret yet + if r.URL.Path == "/v1/secret/data/test/path" && r.Method == "GET" { + w.WriteHeader(http.StatusOK) + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "data": nil, + }) + return + } + + if r.URL.Path == "/v1/secret/data/test/path" && (r.Method == "POST" || r.Method == "PUT") { + var payload map[string]interface{} + _ = json.NewDecoder(r.Body).Decode(&payload) + lastWrittenData = payload + w.WriteHeader(http.StatusOK) + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "data": map[string]interface{}{"version": 1}, + }) + return + } + + w.WriteHeader(http.StatusNotFound) + })) + defer server.Close() + + client, err := NewClient(&Config{ + Address: server.URL, + RoleID: "test-role-id", + SecretID: "test-secret-id", + MountPath: "secret", + }) + require.NoError(t, err) + + err = client.WriteToken(context.Background(), "test/path", "only-token", "token", WriteActionAppend) + require.NoError(t, err) + + data, ok := lastWrittenData["data"].(map[string]interface{}) + require.True(t, ok) + assert.Equal(t, "only-token", data["token"]) + assert.Len(t, data, 1) + opts, ok := lastWrittenData["options"].(map[string]interface{}) + require.True(t, ok) + assert.EqualValues(t, 0, opts["cas"], "create should use cas=0") +} + +func TestWriteToken_InvalidAction(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/v1/auth/approle/login" { + w.WriteHeader(http.StatusOK) + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "auth": map[string]interface{}{ + "client_token": "test-token", + "lease_duration": 3600, + }, + }) + return + } + w.WriteHeader(http.StatusNotFound) + })) + defer server.Close() + + client, err := NewClient(&Config{ + Address: server.URL, + RoleID: "test-role-id", + SecretID: "test-secret-id", + MountPath: "secret", + }) + require.NoError(t, err) + + err = client.WriteToken(context.Background(), "test/path", "v", "token", "merge") + require.Error(t, err) + assert.Contains(t, err.Error(), "invalid write action") } func TestReadToken(t *testing.T) { @@ -131,7 +485,7 @@ func TestReadToken(t *testing.T) { w.WriteHeader(http.StatusOK) response := map[string]interface{}{ "auth": map[string]interface{}{ - "client_token": "test-token", + "client_token": "test-token", "lease_duration": 3600, }, } @@ -170,18 +524,61 @@ func TestReadToken(t *testing.T) { require.NoError(t, err) ctx := context.Background() - token, err := client.ReadToken(ctx, "test/path") + token, err := client.ReadToken(ctx, "test/path", "") require.NoError(t, err) assert.Equal(t, "retrieved-secret-token", token) } +func TestReadToken_CustomKey(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/v1/auth/approle/login" { + w.WriteHeader(http.StatusOK) + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "auth": map[string]interface{}{ + "client_token": "test-token", + "lease_duration": 3600, + }, + }) + return + } + + if r.URL.Path == "/v1/secret/data/test/path" && r.Method == "GET" { + w.WriteHeader(http.StatusOK) + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "data": map[string]interface{}{ + "data": map[string]interface{}{ + "api_token": "custom-key-token", + }, + "metadata": map[string]interface{}{"version": 1}, + }, + }) + return + } + + w.WriteHeader(http.StatusNotFound) + })) + defer server.Close() + + client, err := NewClient(&Config{ + Address: server.URL, + RoleID: "test-role-id", + SecretID: "test-secret-id", + MountPath: "secret", + }) + require.NoError(t, err) + + token, err := client.ReadToken(context.Background(), "test/path", "api_token") + require.NoError(t, err) + assert.Equal(t, "custom-key-token", token) +} + func TestReadToken_NotFound(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.URL.Path == "/v1/auth/approle/login" { w.WriteHeader(http.StatusOK) response := map[string]interface{}{ "auth": map[string]interface{}{ - "client_token": "test-token", + "client_token": "test-token", "lease_duration": 3600, }, } @@ -204,7 +601,7 @@ func TestReadToken_NotFound(t *testing.T) { require.NoError(t, err) ctx := context.Background() - token, err := client.ReadToken(ctx, "nonexistent/path") + token, err := client.ReadToken(ctx, "nonexistent/path", "") require.Error(t, err) assert.Empty(t, token) } @@ -217,7 +614,7 @@ func TestWriteTokenState(t *testing.T) { w.WriteHeader(http.StatusOK) response := map[string]interface{}{ "auth": map[string]interface{}{ - "client_token": "test-token", + "client_token": "test-token", "lease_duration": 3600, }, } @@ -248,13 +645,13 @@ func TestWriteTokenState(t *testing.T) { require.NoError(t, err) state := &models.TokenState{ - Label: "test-token", - CurrentLinodeID: 123, - CurrentTokenValue: "secret-value", - LastRotatedAt: time.Now(), - PreviousLinodeID: 100, - PreviousExpiresAt: time.Now().Add(60 * 24 * time.Hour), - RotationCount: 5, + Label: "test-token", + CurrentLinodeID: 123, + CurrentTokenValue: "secret-value", + LastRotatedAt: time.Now(), + PreviousLinodeID: 100, + PreviousExpiresAt: time.Now().Add(60 * 24 * time.Hour), + RotationCount: 5, } ctx := context.Background() @@ -276,7 +673,7 @@ func TestReadTokenState(t *testing.T) { w.WriteHeader(http.StatusOK) response := map[string]interface{}{ "auth": map[string]interface{}{ - "client_token": "test-token", + "client_token": "test-token", "lease_duration": 3600, }, } @@ -289,12 +686,12 @@ func TestReadTokenState(t *testing.T) { response := map[string]interface{}{ "data": map[string]interface{}{ "custom_metadata": map[string]interface{}{ - "label": "test-token", - "current_linode_id": "123", - "last_rotated_at": now.Format(time.RFC3339), - "previous_linode_id": "100", - "previous_expires_at": now.Add(60 * 24 * time.Hour).Format(time.RFC3339), - "rotation_count": "5", + "label": "test-token", + "current_linode_id": "123", + "last_rotated_at": now.Format(time.RFC3339), + "previous_linode_id": "100", + "previous_expires_at": now.Add(60 * 24 * time.Hour).Format(time.RFC3339), + "rotation_count": "5", }, }, } @@ -333,7 +730,7 @@ func TestReadTokenState_NotFound(t *testing.T) { w.WriteHeader(http.StatusOK) response := map[string]interface{}{ "auth": map[string]interface{}{ - "client_token": "test-token", + "client_token": "test-token", "lease_duration": 3600, }, } diff --git a/latr-mixin/README.md b/latr-mixin/README.md index fa6013a..3fd89f6 100644 --- a/latr-mixin/README.md +++ b/latr-mixin/README.md @@ -5,7 +5,7 @@ A [Prometheus monitoring mixin](https://monitoring.mixins.dev/) for [latr](https It packages: - A Grafana dashboard for rotations, token validity, duration, and Vault errors -- Prometheus alerting rules for failed rotations, Vault storage failures, low remaining validity, and missing metrics +- Prometheus alerting rules for failed rotations, Vault storage failures, append CAS contention, low remaining validity, and missing metrics Mixins are written in [Jsonnet](https://jsonnet.org/) and are typically installed with [jsonnet-bundler](https://github.com/jsonnet-bundler/jsonnet-bundler). diff --git a/latr-mixin/alerts/alerts.libsonnet b/latr-mixin/alerts/alerts.libsonnet index a5edaaa..9204c96 100644 --- a/latr-mixin/alerts/alerts.libsonnet +++ b/latr-mixin/alerts/alerts.libsonnet @@ -42,7 +42,7 @@ { alert: 'LatrVaultStorageErrors', expr: ||| - sum by (path) ( + sum by (path, action) ( increase(latr_vault_storage_errors_total%(sel)s[%(window)s]) ) > 0 ||| % { @@ -56,11 +56,41 @@ annotations: { summary: 'latr failed to write a rotated token to Vault.', description: ||| - Vault path {{ $labels.path }} had {{ $value | humanize }} storage error(s) in the last %(window)s. + Vault path {{ $labels.path }} (action={{ $labels.action }}) had {{ $value | humanize }} storage error(s) in the last %(window)s. Linode and Vault may be out of sync — do not revoke tokens until storage succeeds and consumers are updated. + For action=append, check AppRole has read+write on the data path and that concurrent writers are not thrashing CAS. ||| % { window: cfg.alertWindow }, }, }, + { + // Sustained CAS conflicts mean shared secrets have concurrent writers + // racing latr appends (or CAS misconfiguration). Occasional conflicts are OK. + alert: 'LatrVaultAppendCASContention', + expr: ||| + sum( + increase(latr_vault_append_cas_conflicts_total%(sel)s[%(window)s]) + ) > %(casThreshold)s + ||| % { + sel: sel, + window: cfg.alertWindow, + casThreshold: cfg.appendCASConflictAlertThreshold, + }, + 'for': cfg.alertFor, + labels: { + severity: 'warning', + }, + annotations: { + summary: 'latr Vault append is hitting frequent check-and-set conflicts.', + description: ||| + latr observed {{ $value | humanize }} Vault KV CAS conflict(s) during append writes in the last %(window)s (threshold %(casThreshold)s). + Another process may be updating the same secret, or multiple latr pods may be active. + Check vault_path in latr logs ("Vault append CAS conflict") and ensure a single active latr writer per secret. + ||| % { + window: cfg.alertWindow, + casThreshold: cfg.appendCASConflictAlertThreshold, + }, + }, + }, { alert: 'LatrTokenValidityLow', expr: ||| diff --git a/latr-mixin/config.libsonnet b/latr-mixin/config.libsonnet index c60c463..73ca1f3 100644 --- a/latr-mixin/config.libsonnet +++ b/latr-mixin/config.libsonnet @@ -26,5 +26,8 @@ alertWindow: '15m', // Pending duration before alerts fire. alertFor: '5m', + // Fire when append CAS conflicts exceed this count in alertWindow + // (default: more than a few retries — occasional conflicts are normal). + appendCASConflictAlertThreshold: 10, }, } diff --git a/test/e2e/testdata/config-create.yaml b/test/e2e/testdata/config-create.yaml index 53dde9e..a8ae9fc 100644 --- a/test/e2e/testdata/config-create.yaml +++ b/test/e2e/testdata/config-create.yaml @@ -21,4 +21,4 @@ tokens: scopes: "*" storage: - type: "vault" - path: "secret/data/e2e/test-create" + path: "e2e/test-create" diff --git a/test/e2e/testdata/config-daemon.yaml b/test/e2e/testdata/config-daemon.yaml index ce8e34b..0d475e1 100644 --- a/test/e2e/testdata/config-daemon.yaml +++ b/test/e2e/testdata/config-daemon.yaml @@ -22,4 +22,4 @@ tokens: scopes: "*" storage: - type: "vault" - path: "secret/data/e2e/test-daemon" + path: "e2e/test-daemon" diff --git a/test/e2e/testdata/config-dryrun.yaml b/test/e2e/testdata/config-dryrun.yaml index 2e26276..6f91bbc 100644 --- a/test/e2e/testdata/config-dryrun.yaml +++ b/test/e2e/testdata/config-dryrun.yaml @@ -21,4 +21,4 @@ tokens: scopes: "*" storage: - type: "vault" - path: "secret/data/e2e/test-dryrun" + path: "e2e/test-dryrun" diff --git a/test/e2e/testdata/config-rotate.yaml b/test/e2e/testdata/config-rotate.yaml index 4cb6763..4941131 100644 --- a/test/e2e/testdata/config-rotate.yaml +++ b/test/e2e/testdata/config-rotate.yaml @@ -21,4 +21,4 @@ tokens: scopes: "*" storage: - type: "vault" - path: "secret/data/e2e/test-rotate" + path: "e2e/test-rotate"