Add keyed hmac256 mask strategy (CDC_MASK_HMAC_KEY)

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.
 

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}`.
 

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
 		}

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

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 {

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),

internal/infrastructure/transform/sink.go

@@ -27,22 +27,23 @@ 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,
 ) (domain.Sink, error) {
 	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)
 		}
 	}
 
 	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)
 		}

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
 			}

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)
 	}