diff --git a/CLAUDE.md b/CLAUDE.md index b487a7d..0b73e21 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -89,6 +89,7 @@ Consequences to keep in mind: | CDC_DLQ_ENABLED | true | Park poison events in Redis instead of stalling forever (needs REDIS_URL) | | CDC_DLQ_THRESHOLD | 3 | Failure cycles for the same event before it is parked | | CDC_DLQ_MAX_ENTRIES | 10000 | DLQ cap; when full, parking fails and the pipeline stalls | +| CDC_MASK_HMAC_KEY | - | Secret key for the hmac256 mask strategy (startup fails if a transform uses hmac256 without it) | | SINK_CONFIG_FILE | - | Path to per-table sink YAML (see config/sinks.example.yaml). Without it, Redis sinks mirror all tables | | REDIS_URL | - | Redis connection (enables stream sink if set) | | REDIS_STREAM_PREFIX | cdc | Stream key prefix | @@ -106,7 +107,8 @@ DELETE events, which would make the delete target the wrong key. Sink YAML also carries declarative transforms (per sink `transform:` block and per-table object entries): CEL `filter` expressions (variables op, schema, table, new, old), `drop_columns`, and `mask` strategies (redact, -null, sha256, last4). Transforms compile at startup (fail fast), are applied +null, sha256, hmac256 — keyed by CDC_MASK_HMAC_KEY — and last4). +Transforms compile at startup (fail fast), are applied per sink outside the resilience wrapper, and never mutate the shared event — see `internal/infrastructure/transform` and config/sinks.example.yaml. diff --git a/README.md b/README.md index e1e0523..a23bc85 100644 --- a/README.md +++ b/README.md @@ -42,9 +42,9 @@ PostgreSQL ──logical replication──▶ GTC ──▶ Redis Streams (eve - **Per-table routing** — an optional YAML file selects which tables go to which sink and how keys are built, using Go templates (`{{.Prefix}}:orders:{{.Field "id"}}`). -- **Declarative transforms** — PII masking (`sha256`, `last4`, `redact`, - `null`), column dropping, and CEL row filters, configured per sink and per - table in YAML. The stream can carry full rows while the search index gets +- **Declarative transforms** — PII masking (`sha256`, keyed `hmac256`, + `last4`, `redact`, `null`), column dropping, and CEL row filters, + configured per sink and per table in YAML. The stream can carry full rows while the search index gets masked ones — no plugin code, no SMT classes. - **First-class transactional outbox** — write domain events to an outbox table in the same transaction as your business writes, and GTC publishes @@ -193,6 +193,7 @@ lives in an optional YAML file. | `CDC_DLQ_ENABLED` | `true` | Park poison events in Redis instead of stalling forever (needs `REDIS_URL`) | | `CDC_DLQ_THRESHOLD` | `3` | Failure cycles for the same event before it is parked | | `CDC_DLQ_MAX_ENTRIES` | `10000` | DLQ cap; when full, parking fails and the pipeline stalls | +| `CDC_MASK_HMAC_KEY` | – | Secret key for the `hmac256` mask strategy (required when a transform uses it) | | `SINK_CONFIG_FILE` | – | Path to the per-table sink YAML (below) | | `REDIS_URL` | – | Enables the Redis Stream sink | | `REDIS_STREAM_PREFIX` | `cdc` | Stream key prefix | @@ -253,12 +254,18 @@ redis_json: that's absent from the event is an error (visible, not silent data loss) — guard optional columns with `"col" in new && new.col == ...`. - **`mask`** strategies: `redact` (`[REDACTED]`), `null`, `sha256` - (deterministic, so masked values still join/dedupe), `last4` + (deterministic, so masked values still join/dedupe), `hmac256` + (deterministic like `sha256`, but keyed with `CDC_MASK_HMAC_KEY` so the + original value can't be recovered by hashing guessed inputs — use this for + low-entropy PII like emails and phone numbers), `last4` (`************4242`). - **`drop_columns`** removes columns entirely. -Transforms are compiled at startup — a typo in a CEL expression or an unknown -mask strategy fails boot with a clear error instead of surfacing mid-stream. +Transforms are compiled at startup — a typo in a CEL expression, an unknown +mask strategy, or `hmac256` without `CDC_MASK_HMAC_KEY` fails boot with a +clear error instead of surfacing mid-stream. Note that rotating the HMAC key +changes every digest, so previously-emitted masked values stop matching new +ones. Masking runs before key generation, so don't build sink keys from masked columns. Filtered events count in `cdc_events_filtered_total{sink,schema,table}`. diff --git a/cmd/gateway/modules.go b/cmd/gateway/modules.go index ea5faf5..b0f5a64 100644 --- a/cmd/gateway/modules.go +++ b/cmd/gateway/modules.go @@ -187,12 +187,13 @@ func newSinks( return wrapped } + transformOpts := transform.Options{HMACKey: []byte(cfg.MaskHMACKey)} addSink := func( sink domain.Sink, global *config.TransformSpec, tables map[string]config.TransformSpec, ) error { - wrapped, err := transform.NewSink(wrapResilienceAndDLQ(sink), global, tables, logger) + wrapped, err := transform.NewSink(wrapResilienceAndDLQ(sink), transformOpts, global, tables, logger) if err != nil { return err } diff --git a/config/sinks.example.yaml b/config/sinks.example.yaml index 8cddcc6..b0f0356 100644 --- a/config/sinks.example.yaml +++ b/config/sinks.example.yaml @@ -12,7 +12,11 @@ # schema, table, new (row data), old (previous data). # Guard optional columns: '"status" in new && new.status == "live"' # drop_columns: removed from the event before the sink sees it -# mask: column -> strategy: redact | null | sha256 | last4 +# mask: column -> strategy: redact | null | sha256 | hmac256 | last4 +# hmac256 is a keyed sha256 (HMAC) using CDC_MASK_HMAC_KEY; +# unlike plain sha256 it cannot be reversed by hashing guessed +# inputs, while staying deterministic (joinable). Startup fails +# if hmac256 is used without the key set. # # Transforms are per sink: the stream sink can carry full rows while the # search index receives masked ones. Masking runs before key generation, so diff --git a/internal/infrastructure/config/config.go b/internal/infrastructure/config/config.go index fad8b6d..e043cd7 100644 --- a/internal/infrastructure/config/config.go +++ b/internal/infrastructure/config/config.go @@ -23,6 +23,9 @@ type Config struct { SlotRetryTimeout time.Duration Backfill BackfillConfig DLQ DLQConfig + // MaskHMACKey keys the hmac256 mask strategy in sink transforms. Empty + // means the strategy is unavailable (configs using it fail at startup). + MaskHMACKey string } type DLQConfig struct { diff --git a/internal/infrastructure/config/loader.go b/internal/infrastructure/config/loader.go index 5838886..4472544 100644 --- a/internal/infrastructure/config/loader.go +++ b/internal/infrastructure/config/loader.go @@ -33,6 +33,7 @@ func Load() (*Config, error) { ChunkDelay: getDuration("CDC_BACKFILL_CHUNK_DELAY", 0), StateTable: backfillStateTable(), }, + MaskHMACKey: os.Getenv("CDC_MASK_HMAC_KEY"), DLQ: DLQConfig{ Enabled: getBool("CDC_DLQ_ENABLED", true), Threshold: getInt("CDC_DLQ_THRESHOLD", 3), diff --git a/internal/infrastructure/transform/sink.go b/internal/infrastructure/transform/sink.go index 2c0cfb6..d2ed50a 100644 --- a/internal/infrastructure/transform/sink.go +++ b/internal/infrastructure/transform/sink.go @@ -27,6 +27,7 @@ var _ domain.Sink = (*Sink)(nil) // inner unchanged when no transforms are configured. func NewSink( inner domain.Sink, + opts Options, global *config.TransformSpec, tables map[string]config.TransformSpec, logger *slog.Logger, @@ -34,7 +35,7 @@ func NewSink( var globalChain *Chain if global != nil { var err error - globalChain, err = Compile(*global) + globalChain, err = CompileWithOptions(opts, *global) if err != nil { return nil, fmt.Errorf("sink %s transform: %w", inner.Name(), err) } @@ -42,7 +43,7 @@ func NewSink( tableChains := make(map[string]*Chain, len(tables)) for name, spec := range tables { - chain, err := Compile(spec) + chain, err := CompileWithOptions(opts, spec) if err != nil { return nil, fmt.Errorf("sink %s, table %s transform: %w", inner.Name(), name, err) } diff --git a/internal/infrastructure/transform/transform.go b/internal/infrastructure/transform/transform.go index e1d5679..8d6f7fd 100644 --- a/internal/infrastructure/transform/transform.go +++ b/internal/infrastructure/transform/transform.go @@ -1,6 +1,7 @@ package transform import ( + "crypto/hmac" "crypto/sha256" "encoding/hex" "fmt" @@ -41,6 +42,36 @@ var maskStrategies = map[string]maskFunc{ }, } +// Options carries deployment-level inputs that mask strategies need. +type Options struct { + // HMACKey keys the hmac256 strategy (CDC_MASK_HMAC_KEY). Unlike plain + // sha256, values masked with a keyed hash cannot be recovered by hashing + // candidate inputs (dictionary attack) without the key, while staying + // deterministic so masked columns remain joinable across events. + HMACKey []byte +} + +// strategyFunc resolves a mask strategy name, including the keyed hmac256 +// strategy which is only available when a key is configured. +func strategyFunc(name string, opts Options) (maskFunc, error) { + if fn, ok := maskStrategies[name]; ok { + return fn, nil + } + if name == "hmac256" { + if len(opts.HMACKey) == 0 { + return nil, fmt.Errorf("mask strategy hmac256 requires CDC_MASK_HMAC_KEY to be set") + } + key := opts.HMACKey + return func(v any) any { + mac := hmac.New(sha256.New, key) + mac.Write([]byte(fmt.Sprintf("%v", v))) + return hex.EncodeToString(mac.Sum(nil)) + }, nil + } + return nil, fmt.Errorf( + "unknown mask strategy %q (valid: redact, null, sha256, hmac256, last4)", name) +} + // Chain is a compiled sequence of transform specs applied to an event in // order. Compilation happens once at startup so bad expressions fail fast. type Chain struct { @@ -56,6 +87,13 @@ type compiledSpec struct { // Compile validates and compiles transform specs into a Chain. A nil Chain // (no specs) is valid and applies nothing. func Compile(specs ...config.TransformSpec) (*Chain, error) { + return CompileWithOptions(Options{}, specs...) +} + +// CompileWithOptions is Compile with deployment-level options (e.g. the +// hmac256 key). Compilation happens once at startup so a strategy whose +// prerequisites are missing fails fast instead of at event time. +func CompileWithOptions(opts Options, specs ...config.TransformSpec) (*Chain, error) { var steps []compiledSpec var env *cel.Env @@ -98,11 +136,9 @@ func Compile(specs ...config.TransformSpec) (*Chain, error) { if len(spec.Mask) > 0 { step.mask = make(map[string]maskFunc, len(spec.Mask)) for col, strategy := range spec.Mask { - fn, ok := maskStrategies[strategy] - if !ok { - return nil, fmt.Errorf( - "unknown mask strategy %q for column %q (valid: redact, null, sha256, last4)", - strategy, col) + fn, err := strategyFunc(strategy, opts) + if err != nil { + return nil, fmt.Errorf("mask for column %q: %w", col, err) } step.mask[col] = fn } diff --git a/internal/infrastructure/transform/transform_test.go b/internal/infrastructure/transform/transform_test.go index 549e0af..7006733 100644 --- a/internal/infrastructure/transform/transform_test.go +++ b/internal/infrastructure/transform/transform_test.go @@ -135,6 +135,56 @@ func TestMaskStrategies(t *testing.T) { } } +func TestHmac256Mask(t *testing.T) { + spec := config.TransformSpec{Mask: map[string]string{"email": "hmac256"}} + chain, err := CompileWithOptions(Options{HMACKey: []byte("test-key")}, spec) + if err != nil { + t.Fatal(err) + } + + mask := func(c *Chain, v string) string { + t.Helper() + out, _, err := c.Apply(event(domain.OperationInsert, map[string]any{"email": v}, nil)) + if err != nil { + t.Fatal(err) + } + return out.NewData["email"].(string) + } + + got := mask(chain, "ada@example.com") + if len(got) != 64 || got == "ada@example.com" { + t.Fatalf("hmac256 mask failed: %q", got) + } + if again := mask(chain, "ada@example.com"); again != got { + t.Errorf("hmac256 must be deterministic for the same key: %q != %q", again, got) + } + + // A plain (unkeyed) sha256 of the same value must differ, otherwise the + // key isn't being used and the mask is dictionary-attackable. + plain := mustCompile(t, config.TransformSpec{Mask: map[string]string{"email": "sha256"}}) + if mask(plain, "ada@example.com") == got { + t.Error("hmac256 output must differ from unkeyed sha256") + } + + // A different key must produce a different digest. + other, err := CompileWithOptions(Options{HMACKey: []byte("other-key")}, spec) + if err != nil { + t.Fatal(err) + } + if mask(other, "ada@example.com") == got { + t.Error("different keys must produce different digests") + } +} + +func TestHmac256RequiresKey(t *testing.T) { + spec := config.TransformSpec{Mask: map[string]string{"email": "hmac256"}} + if _, err := Compile(spec); err == nil { + t.Fatal("hmac256 without a key must fail compilation") + } else if !strings.Contains(err.Error(), "CDC_MASK_HMAC_KEY") { + t.Errorf("error must point at the missing env var, got: %v", err) + } +} + func TestMaskAppliesToOldData(t *testing.T) { chain := mustCompile(t, config.TransformSpec{Mask: map[string]string{"email": "redact"}}) @@ -190,7 +240,7 @@ func (s *captureSink) Process(_ context.Context, e domain.CDCEvent) error { func TestSinkAppliesGlobalThenTableChain(t *testing.T) { inner := &captureSink{} - sink, err := NewSink(inner, + sink, err := NewSink(inner, Options{}, &config.TransformSpec{Mask: map[string]string{"email": "redact"}}, map[string]config.TransformSpec{ "public.users": {DropColumns: []string{"password_hash"}}, @@ -222,7 +272,7 @@ func TestSinkAppliesGlobalThenTableChain(t *testing.T) { func TestSinkFiltersEvent(t *testing.T) { inner := &captureSink{} - sink, err := NewSink(inner, &config.TransformSpec{Filter: `op != "DELETE"`}, nil, slog.Default()) + sink, err := NewSink(inner, Options{}, &config.TransformSpec{Filter: `op != "DELETE"`}, nil, slog.Default()) if err != nil { t.Fatal(err) } @@ -238,7 +288,7 @@ func TestSinkFiltersEvent(t *testing.T) { func TestSinkWithoutTransformsIsUnwrapped(t *testing.T) { inner := &captureSink{} - sink, err := NewSink(inner, nil, nil, slog.Default()) + sink, err := NewSink(inner, Options{}, nil, nil, slog.Default()) if err != nil { t.Fatal(err) }