From 39e4e4f4885047035cc44165c4d03d2be6e93acc Mon Sep 17 00:00:00 2001 From: Mzack9999 Date: Thu, 21 May 2026 14:18:12 +0200 Subject: [PATCH 1/3] adding shared redis storage --- cmd/interactsh-server/main.go | 21 +- deploy/redis-test/Dockerfile | 14 + deploy/redis-test/docker-compose.yml | 66 +++ deploy/redis-test/verify/main.go | 117 +++++ go.mod | 6 +- go.sum | 11 + pkg/options/server_options.go | 5 + pkg/storage/storage_redis.go | 478 ++++++++++++++++++ pkg/storage/storage_redis_integration_test.go | 68 +++ pkg/storage/storage_redis_lua.go | 170 +++++++ pkg/storage/storage_redis_test.go | 257 ++++++++++ 11 files changed, 1211 insertions(+), 2 deletions(-) create mode 100644 deploy/redis-test/Dockerfile create mode 100644 deploy/redis-test/docker-compose.yml create mode 100644 deploy/redis-test/verify/main.go create mode 100644 pkg/storage/storage_redis.go create mode 100644 pkg/storage/storage_redis_integration_test.go create mode 100644 pkg/storage/storage_redis_lua.go create mode 100644 pkg/storage/storage_redis_test.go diff --git a/cmd/interactsh-server/main.go b/cmd/interactsh-server/main.go index aa7469fb..82968146 100644 --- a/cmd/interactsh-server/main.go +++ b/cmd/interactsh-server/main.go @@ -73,6 +73,8 @@ func main() { flagSet.StringVarP(&cliOptions.DefaultHTTPResponseFile, "default-http-response", "dhr", "", "file to serve for all http requests (takes priority over other options)"), flagSet.BoolVarP(&cliOptions.DiskStorage, "disk", "ds", false, "disk based storage"), flagSet.StringVarP(&cliOptions.DiskStoragePath, "disk-path", "dsp", "", "disk storage path"), + flagSet.StringVarP(&cliOptions.RedisURL, "redis-url", "ru", "", "redis connection URL (enables shared state for multi-instance deployments)"), + flagSet.StringVarP(&cliOptions.RedisKeyPrefix, "redis-prefix", "rp", "", "redis key prefix (default \"interactsh:\")"), flagSet.StringVarP(&cliOptions.HeaderServer, "server-header", "csh", "", "custom value of Server header in response"), flagSet.BoolVarP(&cliOptions.NoVersionHeader, "disable-version", "dv", false, "disable publishing interactsh version in response header"), ) @@ -262,7 +264,24 @@ func main() { } var err error - store, err = storage.New(&storeOptions) + switch { + case cliOptions.RedisURL != "": + // Redis-backed storage shares state across multiple interactsh-server + // instances behind a load balancer. Disk/in-memory flags are ignored + // in this mode by design. + if cliOptions.DiskStorage { + gologger.Warning().Msgf("--redis-url is set; disk-storage flags will be ignored\n") + } + store, err = storage.NewRedis(&storage.RedisOptions{ + URL: cliOptions.RedisURL, + KeyPrefix: cliOptions.RedisKeyPrefix, + EvictionTTL: evictionTTL, + EvictionStrategy: evictionStrategy, + MaxSharedInteractions: storeOptions.MaxSharedInteractions, + }) + default: + store, err = storage.New(&storeOptions) + } if err != nil { gologger.Fatal().Msgf("couldn't create storage: %s\n", err) } diff --git a/deploy/redis-test/Dockerfile b/deploy/redis-test/Dockerfile new file mode 100644 index 00000000..59f7e101 --- /dev/null +++ b/deploy/redis-test/Dockerfile @@ -0,0 +1,14 @@ +# syntax=docker/dockerfile:1 +FROM golang:1.24-alpine AS builder +WORKDIR /src +COPY go.mod go.sum ./ +RUN go mod download +COPY . . +RUN CGO_ENABLED=0 go build -trimpath -o /out/interactsh-server ./cmd/interactsh-server +RUN CGO_ENABLED=0 go build -trimpath -o /out/interactsh-client ./cmd/interactsh-client + +FROM alpine:3.20 +RUN apk add --no-cache curl ca-certificates +COPY --from=builder /out/interactsh-server /usr/local/bin/interactsh-server +COPY --from=builder /out/interactsh-client /usr/local/bin/interactsh-client +ENTRYPOINT ["interactsh-server"] diff --git a/deploy/redis-test/docker-compose.yml b/deploy/redis-test/docker-compose.yml new file mode 100644 index 00000000..b0677865 --- /dev/null +++ b/deploy/redis-test/docker-compose.yml @@ -0,0 +1,66 @@ +name: interactsh-redis-test + +# Two interactsh-server instances pointing at the same Redis to exercise +# the multi-instance scenario from issue #1267. Both run with --skip-acme +# (TLS disabled) and authentication so the test harness can register/poll +# without dragging in real DNS or ACME. +services: + redis: + image: redis:7-alpine + healthcheck: + test: ["CMD", "redis-cli", "ping"] + interval: 1s + timeout: 2s + retries: 10 + + ish-a: + build: + context: ../.. + dockerfile: deploy/redis-test/Dockerfile + command: + - "-d=oast.local" + - "-i=127.0.0.1" + - "-sa" + - "-ru=redis://redis:6379" + - "-rp=ishtest:" + - "-dns-port=5353" + - "-smtp-port=2525" + - "-smtp-autotls-port=2526" + - "-smtps-port=2527" + - "-ldap-port=3389" + - "-ftp-port=2121" + - "-ftps-port=2122" + - "-smb-port=4445" + - "-a" + - "-t=testtoken-7e2c" + ports: + - "8080:80" + depends_on: + redis: + condition: service_healthy + + ish-b: + build: + context: ../.. + dockerfile: deploy/redis-test/Dockerfile + command: + - "-d=oast.local" + - "-i=127.0.0.1" + - "-sa" + - "-ru=redis://redis:6379" + - "-rp=ishtest:" + - "-dns-port=5353" + - "-smtp-port=2525" + - "-smtp-autotls-port=2526" + - "-smtps-port=2527" + - "-ldap-port=3389" + - "-ftp-port=2121" + - "-ftps-port=2122" + - "-smb-port=4445" + - "-a" + - "-t=testtoken-7e2c" + ports: + - "8081:80" + depends_on: + redis: + condition: service_healthy diff --git a/deploy/redis-test/verify/main.go b/deploy/redis-test/verify/main.go new file mode 100644 index 00000000..3d306f86 --- /dev/null +++ b/deploy/redis-test/verify/main.go @@ -0,0 +1,117 @@ +// verify is a small smoke harness for issue #1267: it registers a client +// against interactsh-server "A", triggers an HTTP interaction directly on +// interactsh-server "B", and verifies that the polling client (still talking +// to "A") receives the interaction thanks to the shared Redis state. +// +// go run ./deploy/redis-test/verify +package main + +import ( + "context" + "fmt" + "net/http" + "os" + "strings" + "sync" + "time" + + "github.com/projectdiscovery/interactsh/pkg/client" + "github.com/projectdiscovery/interactsh/pkg/server" +) + +func main() { + if err := run(); err != nil { + fmt.Fprintf(os.Stderr, "verify FAILED: %v\n", err) + os.Exit(1) + } + fmt.Println("verify OK") +} + +func run() error { + urlA := envOr("ISH_A_URL", "http://localhost:8080") + urlB := envOr("ISH_B_URL", "http://localhost:8081") + token := envOr("ISH_TOKEN", "testtoken-7e2c") + + c, err := client.New(&client.Options{ + ServerURL: urlA, + Token: token, + DisableHTTPFallback: false, + CorrelationIdLength: 20, + CorrelationIdNonceLength: 13, + }) + if err != nil { + return fmt.Errorf("client.New: %w", err) + } + defer c.Close() + + callbackURL := c.URL() + if callbackURL == "" { + return fmt.Errorf("client did not return a callback URL") + } + fmt.Printf("registered against A, callback URL = %s\n", callbackURL) + + got := make(chan *server.Interaction, 1) + var once sync.Once + if err := c.StartPolling(500*time.Millisecond, func(ix *server.Interaction) { + once.Do(func() { got <- ix }) + }); err != nil { + return fmt.Errorf("StartPolling: %w", err) + } + + // Give the poller one tick to enter its loop before we trigger. + time.Sleep(750 * time.Millisecond) + + // Trigger an HTTP interaction on instance B by curling its HTTP port + // with a Host header matching the registered subdomain. Instance B + // resolves the correlation id from the host, encrypts the request + // payload with the AES key that lives in shared Redis, and pushes the + // ciphertext to the data list. Instance A serves the poll. + probe := fmt.Sprintf("hello-from-instance-b-%d", time.Now().UnixNano()) + if err := triggerHTTP(urlB, callbackURL, probe); err != nil { + return fmt.Errorf("trigger interaction on B: %w", err) + } + fmt.Printf("triggered HTTP interaction on B with probe %q\n", probe) + + select { + case ix := <-got: + if ix == nil { + return fmt.Errorf("nil interaction received") + } + if !strings.Contains(ix.RawRequest, probe) { + return fmt.Errorf("interaction received but probe %q not present in RawRequest: %q", + probe, ix.RawRequest) + } + fmt.Printf("polled from A, captured B's interaction (protocol=%s, remote=%s)\n", + ix.Protocol, ix.RemoteAddress) + return nil + case <-time.After(15 * time.Second): + return fmt.Errorf("timed out waiting for interaction to propagate via Redis") + } +} + +func triggerHTTP(serverURL, callbackURL, probe string) error { + host := strings.Split(strings.TrimPrefix(strings.TrimPrefix(callbackURL, + "http://"), "https://"), "/")[0] + + req, err := http.NewRequestWithContext(context.Background(), http.MethodGet, + serverURL+"/"+probe, nil) + if err != nil { + return err + } + req.Host = host + req.Header.Set("X-Probe", probe) + + resp, err := (&http.Client{Timeout: 5 * time.Second}).Do(req) + if err != nil { + return err + } + _ = resp.Body.Close() + return nil +} + +func envOr(k, def string) string { + if v := os.Getenv(k); v != "" { + return v + } + return def +} diff --git a/go.mod b/go.mod index 02122c1e..b42b01cf 100644 --- a/go.mod +++ b/go.mod @@ -43,6 +43,7 @@ require ( github.com/VividCortex/ewma v1.2.0 // indirect github.com/akrylysov/pogreb v0.10.2 // indirect github.com/alecthomas/chroma/v2 v2.20.0 // indirect + github.com/alicebob/miniredis/v2 v2.38.0 // indirect github.com/andybalholm/brotli v1.2.0 // indirect github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2 // indirect github.com/aymanbagabas/go-osc52/v2 v2.0.1 // indirect @@ -52,6 +53,7 @@ require ( github.com/bodgit/sevenzip v1.6.1 // indirect github.com/bodgit/windows v1.0.1 // indirect github.com/caddyserver/zerossl v0.1.3 // indirect + github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/charmbracelet/colorprofile v0.3.2 // indirect github.com/charmbracelet/glamour v0.10.0 // indirect github.com/charmbracelet/lipgloss v1.1.1-0.20250404203927-76690c660834 // indirect @@ -120,6 +122,7 @@ require ( github.com/projectdiscovery/machineid v0.0.0-20250715113114-c77eb3567582 // indirect github.com/projectdiscovery/mapcidr v1.1.97 // indirect github.com/projectdiscovery/networkpolicy v0.1.38 // indirect + github.com/redis/go-redis/v9 v9.19.0 // indirect github.com/refraction-networking/utls v1.8.2 // indirect github.com/rivo/uniseg v0.4.7 // indirect github.com/saintfish/chardet v0.0.0-20230101081208-5e3ef4b5456d // indirect @@ -141,13 +144,14 @@ require ( github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e // indirect github.com/yuin/goldmark v1.7.13 // indirect github.com/yuin/goldmark-emoji v1.0.6 // indirect + github.com/yuin/gopher-lua v1.1.1 // indirect github.com/yusufpapurcu/wmi v1.2.4 // indirect github.com/zcalusic/sysinfo v1.1.3 // indirect github.com/zeebo/blake3 v0.2.4 // indirect github.com/zmap/rc2 v0.0.0-20190804163417-abaa70531248 // indirect github.com/zmap/zcrypto v0.0.0-20240803002437-3a861682ac77 // indirect go.etcd.io/bbolt v1.4.3 // indirect - go.uber.org/atomic v1.10.0 // indirect + go.uber.org/atomic v1.11.0 // indirect go.uber.org/zap/exp v0.3.0 // indirect go4.org v0.0.0-20230225012048-214862532bf5 // indirect golang.org/x/crypto v0.46.0 // indirect diff --git a/go.sum b/go.sum index 3e821750..161cf58e 100644 --- a/go.sum +++ b/go.sum @@ -48,6 +48,8 @@ github.com/alecthomas/repr v0.5.1 h1:E3G4t2QbHTSNpPKBgMTln5KLkZHLOcU7r37J4pXBuIg github.com/alecthomas/repr v0.5.1/go.mod h1:Fr0507jx4eOXV7AlPV6AVZLYrLIuIeSOWtW57eE/O/4= github.com/alexbrainman/sspi v0.0.0-20250919150558-7d374ff0d59e h1:4dAU9FXIyQktpoUAgOJK3OTFc/xug0PCXYCqU0FgDKI= github.com/alexbrainman/sspi v0.0.0-20250919150558-7d374ff0d59e/go.mod h1:cEWa1LVoE5KvSD9ONXsZrj0z6KqySlCCNKHlLzbqAt4= +github.com/alicebob/miniredis/v2 v2.38.0 h1:nZAzCR+Lj+Vxk4ZXzm2NuKq2O33RXj1XxJ2e2uP9jiw= +github.com/alicebob/miniredis/v2 v2.38.0/go.mod h1:TcL7YfarKPGDAthEtl5NBeHZfeUQj6OXMm/+iu5cLMM= github.com/andybalholm/brotli v1.2.0 h1:ukwgCxwYrmACq68yiUqwIWnGY0cTPox/M94sVwToPjQ= github.com/andybalholm/brotli v1.2.0/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY= github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2 h1:DklsrG3dyBCFEj5IhUbnKptjxatkF07cF2ak3yi77so= @@ -76,6 +78,9 @@ github.com/caddyserver/certmagic v0.25.0/go.mod h1:m9yB7Mud24OQbPHOiipAoyKPn9pKH github.com/caddyserver/zerossl v0.1.3 h1:onS+pxp3M8HnHpN5MMbOMyNjmTheJyWRaZYwn+YTAyA= github.com/caddyserver/zerossl v0.1.3/go.mod h1:CxA0acn7oEGO6//4rtrRjYgEoa4MFw/XofZnrYwGqG4= github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU= +github.com/cespare/xxhash v1.1.0 h1:a6HrQnmkObjyL+Gs60czilIUGqrzKutQD6XZog3p+ko= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/charmbracelet/colorprofile v0.3.2 h1:9J27WdztfJQVAQKX2WOlSSRB+5gaKqqITmrvb1uTIiI= github.com/charmbracelet/colorprofile v0.3.2/go.mod h1:mTD5XzNeWHj8oqHb+S1bssQb7vIHbepiebQ2kPKVKbI= github.com/charmbracelet/glamour v0.10.0 h1:MtZvfwsYCx8jEPFJm3rIBFIMZUfUJ765oX8V6kXldcY= @@ -354,6 +359,8 @@ github.com/projectdiscovery/retryablehttp-go v1.3.11/go.mod h1:oTy1TiZcfb9SdDqek github.com/projectdiscovery/utils v0.11.0 h1:CxImZSRyj9spy1wpB9HKJopr5MsIPm2r5iS8uyhAMoQ= github.com/projectdiscovery/utils v0.11.0/go.mod h1:q2mZngH1s4WDO3knYxG7iyP1KcxoRSORJCWSpCKFc1s= github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA= +github.com/redis/go-redis/v9 v9.19.0 h1:XPVaaPSnG6RhYf7p+rmSa9zZfeVAnWsH5h3lxthOm/k= +github.com/redis/go-redis/v9 v9.19.0/go.mod h1:v/M13XI1PVCDcm01VtPFOADfZtHf8YW3baQf57KlIkA= github.com/refraction-networking/utls v1.8.2 h1:j4Q1gJj0xngdeH+Ox/qND11aEfhpgoEvV+S9iJ2IdQo= github.com/refraction-networking/utls v1.8.2/go.mod h1:jkSOEkLqn+S/jtpEHPOsVv/4V4EVnelwbMQl4vCWXAM= github.com/remeh/sizedwaitgroup v1.0.0 h1:VNGGFwNo/R5+MJBf6yrsr110p0m4/OX4S3DCy7Kyl5E= @@ -444,6 +451,8 @@ github.com/yuin/goldmark v1.7.13 h1:GPddIs617DnBLFFVJFgpo1aBfe/4xcvMc3SB5t/D0pA= github.com/yuin/goldmark v1.7.13/go.mod h1:ip/1k0VRfGynBgxOz0yCqHrbZXhcjxyuS66Brc7iBKg= github.com/yuin/goldmark-emoji v1.0.6 h1:QWfF2FYaXwL74tfGOW5izeiZepUDroDJfWubQI9HTHs= github.com/yuin/goldmark-emoji v1.0.6/go.mod h1:ukxJDKFpdFb5x0a5HqbdlcKtebh086iJpI31LTKmWuA= +github.com/yuin/gopher-lua v1.1.1 h1:kYKnWBjvbNP4XLT3+bPEwAXJx262OhaHDWDVOPjL46M= +github.com/yuin/gopher-lua v1.1.1/go.mod h1:GBR0iDaNXjAgGg9zfCvksxSRnQx76gclCIb7kdAd1Pw= github.com/yusufpapurcu/wmi v1.2.4 h1:zFUKzehAFReQwLys1b/iSMl+JQGSCSjtVqQn9bBrPo0= github.com/yusufpapurcu/wmi v1.2.4/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0= github.com/zcalusic/sysinfo v1.1.3 h1:u/AVENkuoikKuIZ4sUEJ6iibpmQP6YpGD8SSMCrqAF0= @@ -472,6 +481,8 @@ go.opencensus.io v0.22.2/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw= go.opencensus.io v0.22.3/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw= go.uber.org/atomic v1.10.0 h1:9qC72Qh0+3MqyJbAn8YU5xVq1frD8bn3JtD2oXtafVQ= go.uber.org/atomic v1.10.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= +go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= +go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0= diff --git a/pkg/options/server_options.go b/pkg/options/server_options.go index d64a921f..96352cb9 100644 --- a/pkg/options/server_options.go +++ b/pkg/options/server_options.go @@ -52,6 +52,11 @@ type CLIServerOptions struct { OriginIPHeader string DiskStorage bool DiskStoragePath string + // RedisURL, when set, switches the server to a Redis-backed storage + // backend so multiple instances can share state behind a load balancer. + // Disk storage flags are ignored when RedisURL is set. + RedisURL string + RedisKeyPrefix string EnablePprof bool EnableMetrics bool Verbose bool diff --git a/pkg/storage/storage_redis.go b/pkg/storage/storage_redis.go new file mode 100644 index 00000000..b94b6f49 --- /dev/null +++ b/pkg/storage/storage_redis.go @@ -0,0 +1,478 @@ +// Package storage - Redis-backed implementation of the Storage interface. +// +// This file adds an additive, optional backend used when multiple interactsh +// server instances need to share state behind a load balancer. It does not +// replace or alter the default in-memory/LevelDB StorageDB; both backends +// coexist and the caller decides which one to construct. +package storage + +import ( + "context" + "crypto/rand" + "crypto/rsa" + "crypto/sha256" + "crypto/tls" + "encoding/base64" + "errors" + "fmt" + "strings" + "time" + + "github.com/redis/go-redis/v9" +) + +// RedisOptions configures the Redis-backed Storage. +// +// All eviction-related fields mirror the semantics of the local Options struct; +// they are duplicated here intentionally so that adding the Redis backend does +// not touch the existing Options/StorageDB code paths. +type RedisOptions struct { + // URL is a redis connection string, e.g. redis://user:pass@host:6379/0 + URL string + // KeyPrefix is prepended to every key. Defaults to "interactsh:". + KeyPrefix string + // TLS, when non-nil, enables TLS with the provided config. + TLS *tls.Config + // EvictionTTL controls how long correlation state survives without activity. + // A value <= 0 disables TTL. + EvictionTTL time.Duration + // EvictionStrategy mirrors storage.Options.EvictionStrategy: + // - sliding: TTL is refreshed on access (default) + // - fixed: TTL is set on write and not refreshed + EvictionStrategy EvictionStrategy + // MaxSharedInteractions caps the per-id buffer length when multiple + // consumers share an id. <= 0 means unlimited. + MaxSharedInteractions int + // Client is an already-configured *redis.Client. If set, URL/TLS are ignored. + // Useful for tests with miniredis. + Client *redis.Client +} + +// StorageRedis implements Storage backed by a Redis server (or compatible). +// +// Key layout (all keys are placed in the same hash slot via the {id} hash tag +// so the implementation is safe under Redis Cluster as well): +// +// meta:{} HASH secret, aes_key, aes_key_enc +// data:{} LIST ordered AES-encrypted interaction strings +// consumers:{} SET consumer ids currently subscribed +// off:{}: STRING per-consumer read offset (integer) +// seen:{}: STRING per-consumer last-seen unix nano +type StorageRedis struct { + options *RedisOptions + client *redis.Client + // metrics counters - tracked locally because Redis cannot give us + // per-Storage hit/miss stats; values are best-effort per-instance. + hitCount uint64 + missCount uint64 +} + +// NewRedis builds a Redis-backed Storage. The caller is responsible for +// keeping the underlying Redis server reachable for the lifetime of the +// returned instance. +func NewRedis(options *RedisOptions) (*StorageRedis, error) { + if options == nil { + return nil, errors.New("redis storage options are required") + } + if options.KeyPrefix == "" { + options.KeyPrefix = "interactsh:" + } + if options.MaxSharedInteractions <= 0 { + options.MaxSharedInteractions = defaultMaxSharedInteractions + } + + client := options.Client + if client == nil { + if options.URL == "" { + return nil, errors.New("redis URL is required when no client is provided") + } + opt, err := redis.ParseURL(options.URL) + if err != nil { + return nil, fmt.Errorf("could not parse redis URL: %w", err) + } + if options.TLS != nil { + opt.TLSConfig = options.TLS + } + client = redis.NewClient(opt) + } + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := client.Ping(ctx).Err(); err != nil { + return nil, fmt.Errorf("could not connect to redis: %w", err) + } + + return &StorageRedis{options: options, client: client}, nil +} + +// ---- key helpers ------------------------------------------------------------ + +func (s *StorageRedis) metaKey(id string) string { + return s.options.KeyPrefix + "meta:{" + id + "}" +} + +func (s *StorageRedis) dataKey(id string) string { + return s.options.KeyPrefix + "data:{" + id + "}" +} + +func (s *StorageRedis) consumersKey(id string) string { + return s.options.KeyPrefix + "consumers:{" + id + "}" +} + +func (s *StorageRedis) offsetKey(id, consumerID string) string { + return s.options.KeyPrefix + "off:{" + id + "}:" + consumerID +} + +func (s *StorageRedis) seenKey(id, consumerID string) string { + return s.options.KeyPrefix + "seen:{" + id + "}:" + consumerID +} + +// ttlMillis returns the TTL in milliseconds; 0 means "no TTL". +func (s *StorageRedis) ttlMillis() int64 { + if s.options.EvictionTTL <= 0 { + return 0 + } + return s.options.EvictionTTL.Milliseconds() +} + +// shouldRefreshTTL is true when the eviction strategy expects TTL to be +// extended on access (sliding); for fixed it is set once at write time only. +func (s *StorageRedis) shouldRefreshTTL() bool { + return s.options.EvictionStrategy == EvictionStrategySliding +} + +// applyWriteTTL sets TTL on the provided keys when a TTL is configured. +// Called after writes so that newly created keys always carry expiry. +func (s *StorageRedis) applyWriteTTL(ctx context.Context, pipe redis.Cmdable, keys ...string) { + ms := s.ttlMillis() + if ms <= 0 { + return + } + for _, k := range keys { + pipe.PExpire(ctx, k, time.Duration(ms)*time.Millisecond) + } +} + +// refreshTTL extends TTL on access when the sliding strategy is configured. +func (s *StorageRedis) refreshTTL(ctx context.Context, keys ...string) { + if !s.shouldRefreshTTL() { + return + } + ms := s.ttlMillis() + if ms <= 0 { + return + } + pipe := s.client.Pipeline() + for _, k := range keys { + pipe.PExpire(ctx, k, time.Duration(ms)*time.Millisecond) + } + _, _ = pipe.Exec(ctx) +} + +// ---- Storage interface ------------------------------------------------------ + +func (s *StorageRedis) GetCacheMetrics() (*CacheMetrics, error) { + return &CacheMetrics{ + HitCount: s.hitCount, + MissCount: s.missCount, + }, nil +} + +// SetIDPublicKey registers a correlation id along with its RSA public key. +// Any pre-existing keyspace for the id is cleared first to mirror the +// StorageDB behaviour after cache eviction + session restore. +func (s *StorageRedis) SetIDPublicKey(correlationID, secretKey, publicKey string) error { + ctx := context.Background() + exists, err := s.client.Exists(ctx, s.metaKey(correlationID)).Result() + if err != nil { + return fmt.Errorf("redis exists check failed: %w", err) + } + if exists > 0 { + return errors.New("correlation-id provided already exists") + } + + pub, err := ParseB64RSAPublicKeyFromPEM(publicKey) + if err != nil { + return fmt.Errorf("could not read public key: %w", err) + } + aesKey := make([]byte, 32) + if _, err := rand.Read(aesKey); err != nil { + return fmt.Errorf("could not generate AES key: %w", err) + } + ciphertext, err := rsa.EncryptOAEP(sha256.New(), rand.Reader, pub, aesKey, []byte("")) + if err != nil { + return errors.New("could not encrypt event data") + } + aesKeyEnc := base64.StdEncoding.EncodeToString(ciphertext) + + pipe := s.client.TxPipeline() + // clear any stale per-id keyspace before re-registering + pipe.Del(ctx, s.metaKey(correlationID), s.dataKey(correlationID), s.consumersKey(correlationID)) + pipe.HSet(ctx, s.metaKey(correlationID), map[string]any{ + "secret": secretKey, + "aes_key": aesKey, + "aes_key_enc": aesKeyEnc, + }) + s.applyWriteTTL(ctx, pipe, s.metaKey(correlationID)) + if _, err := pipe.Exec(ctx); err != nil { + return fmt.Errorf("redis register failed: %w", err) + } + return nil +} + +// SetID registers a correlation id without an associated public key. +// Used for the wildcard / auth-token path. +func (s *StorageRedis) SetID(id string) error { + ctx := context.Background() + pipe := s.client.TxPipeline() + pipe.HSetNX(ctx, s.metaKey(id), "secret", "") + s.applyWriteTTL(ctx, pipe, s.metaKey(id)) + if _, err := pipe.Exec(ctx); err != nil { + return fmt.Errorf("redis SetID failed: %w", err) + } + return nil +} + +// AddInteraction encrypts and appends an interaction for the given +// correlation id. +func (s *StorageRedis) AddInteraction(correlationID string, data []byte) error { + return s.addInteraction(correlationID, data) +} + +// AddInteractionWithId mirrors AddInteraction; the distinction in the +// original StorageDB is purely semantic. +func (s *StorageRedis) AddInteractionWithId(id string, data []byte) error { + return s.addInteraction(id, data) +} + +func (s *StorageRedis) addInteraction(id string, data []byte) error { + if len(data) == 0 { + return nil + } + ctx := context.Background() + + // EXISTS distinguishes "id never registered" from "id registered without + // an AES key" (the wildcard / SetID path) so that ErrCorrelationIdNotFound + // keeps the same meaning as in StorageDB. + exists, err := s.client.Exists(ctx, s.metaKey(id)).Result() + if err != nil { + return fmt.Errorf("redis exists check failed: %w", err) + } + if exists == 0 { + s.missCount++ + return ErrCorrelationIdNotFound + } + s.hitCount++ + + aesKey, err := s.client.HGet(ctx, s.metaKey(id), "aes_key").Bytes() + if err != nil && !errors.Is(err, redis.Nil) { + return fmt.Errorf("redis HGet aes_key failed: %w", err) + } + + payload := string(data) + if len(aesKey) > 0 { + ct, err := AESEncrypt(aesKey, data) + if err != nil { + return fmt.Errorf("could not encrypt event data: %w", err) + } + payload = ct + } + + pipe := s.client.TxPipeline() + pipe.RPush(ctx, s.dataKey(id), payload) + s.applyWriteTTL(ctx, pipe, s.dataKey(id)) + if s.shouldRefreshTTL() { + s.applyWriteTTL(ctx, pipe, s.metaKey(id)) + } + if _, err := pipe.Exec(ctx); err != nil { + return fmt.Errorf("redis RPush failed: %w", err) + } + return nil +} + +// GetInteractions returns all buffered interactions for the correlation id +// and atomically clears the buffer (preserving the legacy single-consumer +// semantics of GetInteractions). +func (s *StorageRedis) GetInteractions(correlationID, secret string) ([]string, string, error) { + ctx := context.Background() + res, err := s.client.HMGet(ctx, s.metaKey(correlationID), "secret", "aes_key_enc").Result() + if err != nil { + return nil, "", fmt.Errorf("redis HMGet failed: %w", err) + } + if res[0] == nil { + s.missCount++ + return nil, "", ErrCorrelationIdNotFound + } + s.hitCount++ + storedSecret, _ := res[0].(string) + if !strings.EqualFold(storedSecret, secret) { + return nil, "", errors.New("invalid secret key passed for user") + } + aesKeyEnc, _ := res[1].(string) + + data, err := s.consumeAll(ctx, correlationID) + if err != nil { + return nil, "", err + } + s.refreshTTL(ctx, s.metaKey(correlationID)) + return data, aesKeyEnc, nil +} + +// GetInteractionsWithId returns and drains the buffered interactions for an id. +func (s *StorageRedis) GetInteractionsWithId(id string) ([]string, error) { + ctx := context.Background() + if exists, _ := s.client.Exists(ctx, s.metaKey(id)).Result(); exists == 0 { + s.missCount++ + return nil, errors.New("could not get id from cache") + } + s.hitCount++ + data, err := s.consumeAll(ctx, id) + if err != nil { + return nil, err + } + s.refreshTTL(ctx, s.metaKey(id)) + return data, nil +} + +// consumeAll atomically LRANGEs and DELs the data list for id. +func (s *StorageRedis) consumeAll(ctx context.Context, id string) ([]string, error) { + pipe := s.client.TxPipeline() + rangeCmd := pipe.LRange(ctx, s.dataKey(id), 0, -1) + pipe.Del(ctx, s.dataKey(id)) + if _, err := pipe.Exec(ctx); err != nil { + return nil, fmt.Errorf("redis consume failed: %w", err) + } + data, err := rangeCmd.Result() + if err != nil { + return nil, err + } + if len(data) == 0 { + return nil, nil + } + return data, nil +} + +// GetInteractionsWithIdForConsumer returns unseen interactions for the given +// consumer, advancing its offset and evicting stale consumers atomically via +// a Lua script. +func (s *StorageRedis) GetInteractionsWithIdForConsumer(id, consumerID string) ([]string, error) { + ctx := context.Background() + if exists, _ := s.client.Exists(ctx, s.metaKey(id)).Result(); exists == 0 { + s.missCount++ + return nil, errors.New("could not get id from cache") + } + s.hitCount++ + + now := time.Now().UnixNano() + evictionNanos := int64(0) + if s.options.EvictionTTL > 0 { + evictionNanos = s.options.EvictionTTL.Nanoseconds() + } + + keys := []string{s.dataKey(id), s.consumersKey(id)} + args := []any{ + consumerID, + now, + evictionNanos, + s.options.MaxSharedInteractions, + s.options.KeyPrefix + "off:{" + id + "}:", + s.options.KeyPrefix + "seen:{" + id + "}:", + } + raw, err := consumerReadScript.Run(ctx, s.client, keys, args...).StringSlice() + if err != nil { + return nil, fmt.Errorf("redis consumer read failed: %w", err) + } + if s.shouldRefreshTTL() { + s.refreshTTL(ctx, s.metaKey(id), s.dataKey(id), s.consumersKey(id), + s.offsetKey(id, consumerID), s.seenKey(id, consumerID)) + } + if len(raw) == 0 { + return nil, nil + } + return raw, nil +} + +// RemoveConsumer drops the consumer's offset/last-seen tracking and compacts +// the underlying data list when possible. +func (s *StorageRedis) RemoveConsumer(id, consumerID string) error { + ctx := context.Background() + keys := []string{s.dataKey(id), s.consumersKey(id)} + args := []any{ + consumerID, + time.Now().UnixNano(), + int64(s.options.EvictionTTL), + s.options.MaxSharedInteractions, + s.options.KeyPrefix + "off:{" + id + "}:", + s.options.KeyPrefix + "seen:{" + id + "}:", + } + _, err := removeConsumerScript.Run(ctx, s.client, keys, args...).Result() + if err != nil && !errors.Is(err, redis.Nil) { + return fmt.Errorf("redis remove consumer failed: %w", err) + } + return nil +} + +// RemoveID drops every key associated with the correlation id after secret +// verification. +func (s *StorageRedis) RemoveID(correlationID, secret string) error { + ctx := context.Background() + storedSecret, err := s.client.HGet(ctx, s.metaKey(correlationID), "secret").Result() + if errors.Is(err, redis.Nil) { + return ErrCorrelationIdNotFound + } + if err != nil { + return fmt.Errorf("redis HGet secret failed: %w", err) + } + if !strings.EqualFold(storedSecret, secret) { + return errors.New("invalid secret key passed for deregister") + } + + consumers, err := s.client.SMembers(ctx, s.consumersKey(correlationID)).Result() + if err != nil && !errors.Is(err, redis.Nil) { + return fmt.Errorf("redis SMembers failed: %w", err) + } + + pipe := s.client.TxPipeline() + pipe.Del(ctx, s.metaKey(correlationID), s.dataKey(correlationID), s.consumersKey(correlationID)) + for _, cid := range consumers { + pipe.Del(ctx, s.offsetKey(correlationID, cid), s.seenKey(correlationID, cid)) + } + if _, err := pipe.Exec(ctx); err != nil { + return fmt.Errorf("redis RemoveID failed: %w", err) + } + return nil +} + +// GetCacheItem returns the CorrelationData associated with the token. Only the +// fields stored in the meta hash are populated; per-consumer offsets live in +// separate keys. +func (s *StorageRedis) GetCacheItem(token string) (*CorrelationData, error) { + ctx := context.Background() + res, err := s.client.HGetAll(ctx, s.metaKey(token)).Result() + if err != nil { + return nil, fmt.Errorf("redis HGetAll failed: %w", err) + } + if len(res) == 0 { + return nil, errors.New("cache item not found") + } + data := &CorrelationData{ + SecretKey: res["secret"], + AESKeyEncrypted: res["aes_key_enc"], + } + if raw, ok := res["aes_key"]; ok { + data.AESKey = []byte(raw) + } + return data, nil +} + +// Close releases the redis client. Existing data in Redis is preserved so +// that the next instance can resume; this matches the multi-instance use +// case the Redis backend was introduced for. +func (s *StorageRedis) Close() error { + if s.client == nil { + return nil + } + return s.client.Close() +} + +var _ Storage = (*StorageRedis)(nil) diff --git a/pkg/storage/storage_redis_integration_test.go b/pkg/storage/storage_redis_integration_test.go new file mode 100644 index 00000000..12c0bf67 --- /dev/null +++ b/pkg/storage/storage_redis_integration_test.go @@ -0,0 +1,68 @@ +//go:build integration_redis + +// This file runs the Redis backend's parity tests against a real Redis +// server. Enable with: +// +// INTERACTSH_REDIS_URL=redis://localhost:16379 go test -tags integration_redis ./pkg/storage/... +// +// The default unit-test suite (without the tag) uses miniredis and does not +// require Docker. +package storage + +import ( + "os" + "testing" + "time" + + "github.com/google/uuid" + "github.com/rs/xid" + "github.com/stretchr/testify/require" +) + +// requireRedisURL ensures the test is skipped when INTERACTSH_REDIS_URL is +// not configured, keeping the integration suite opt-in. +func requireRedisURL(t *testing.T) string { + t.Helper() + url := os.Getenv("INTERACTSH_REDIS_URL") + if url == "" { + t.Skip("INTERACTSH_REDIS_URL is not set; skipping real-redis integration test") + } + return url +} + +// TestRedisIntegrationMultiInstance verifies the headline use case (one +// instance registers, another writes, a third polls) against a real Redis. +func TestRedisIntegrationMultiInstance(t *testing.T) { + url := requireRedisURL(t) + + prefix := "interactsh-it-" + xid.New().String() + ":" + mk := func() *StorageRedis { + store, err := NewRedis(&RedisOptions{ + URL: url, + KeyPrefix: prefix, + EvictionTTL: time.Hour, + EvictionStrategy: EvictionStrategySliding, + MaxSharedInteractions: 64, + }) + require.NoError(t, err) + return store + } + a, b, c := mk(), mk(), mk() + defer func() { _ = a.Close(); _ = b.Close(); _ = c.Close() }() + + priv, pub := generateRSAKeyPair(t) + id := xid.New().String() + secret := uuid.New().String() + + require.NoError(t, a.SetIDPublicKey(id, secret, pub)) + require.NoError(t, b.AddInteraction(id, []byte(`{"protocol":"dns"}`))) + + data, aesKey, err := c.GetInteractions(id, secret) + require.NoError(t, err) + require.Len(t, data, 1) + plaintext := clientDecrypt(t, priv, aesKey, data[0]) + require.Equal(t, `{"protocol":"dns"}`, string(plaintext)) + + // Cleanup the keyspace we created. + require.NoError(t, c.RemoveID(id, secret)) +} diff --git a/pkg/storage/storage_redis_lua.go b/pkg/storage/storage_redis_lua.go new file mode 100644 index 00000000..6d023c13 --- /dev/null +++ b/pkg/storage/storage_redis_lua.go @@ -0,0 +1,170 @@ +package storage + +import "github.com/redis/go-redis/v9" + +// consumerReadScript atomically: +// - returns the unseen interaction slice for the given consumer +// - advances the consumer's offset and last-seen timestamp +// - evicts other consumers that have been idle for longer than evictionTTL +// - drops the data list when no live consumer remains +// - enforces maxBuffer, adjusting every remaining consumer's offset +// +// Hash tag {id} is embedded in every key by the Go layer, so all reads/writes +// land on the same Redis slot under Redis Cluster. +// +// KEYS[1] = data:{} (LIST) +// KEYS[2] = consumers:{} (SET) +// ARGV[1] = consumerID +// ARGV[2] = nowUnixNano (int) +// ARGV[3] = evictionTTLNanos (int; 0 means no idle eviction) +// ARGV[4] = maxBuffer (int; 0 means unlimited) +// ARGV[5] = offsetKeyPrefix ("off:{}:") +// ARGV[6] = seenKeyPrefix ("seen:{}:") +// +// Returns: array of strings (the unseen interactions, may be empty). +var consumerReadScript = redis.NewScript(` +local data_key = KEYS[1] +local consumers_key = KEYS[2] +local consumer = ARGV[1] +local now = tonumber(ARGV[2]) +local eviction_ttl = tonumber(ARGV[3]) +local max_buffer = tonumber(ARGV[4]) +local off_prefix = ARGV[5] +local seen_prefix = ARGV[6] + +local off_key = off_prefix .. consumer +local seen_key = seen_prefix .. consumer + +local offset = tonumber(redis.call("GET", off_key) or "0") +local total = redis.call("LLEN", data_key) +if offset > total then offset = total end + +local unseen = redis.call("LRANGE", data_key, offset, -1) + +redis.call("SET", off_key, total) +redis.call("SET", seen_key, now) +redis.call("SADD", consumers_key, consumer) + +if eviction_ttl > 0 then + local members = redis.call("SMEMBERS", consumers_key) + for _, cid in ipairs(members) do + local sk = seen_prefix .. cid + local ls = tonumber(redis.call("GET", sk) or "0") + if ls > 0 and (now - ls) > eviction_ttl then + redis.call("SREM", consumers_key, cid) + redis.call("DEL", sk) + redis.call("DEL", off_prefix .. cid) + end + end +end + +local live = redis.call("SCARD", consumers_key) +if live == 0 then + redis.call("DEL", data_key) + return unseen +end + +if max_buffer > 0 then + local cur_len = redis.call("LLEN", data_key) + if cur_len > max_buffer then + local trim = cur_len - max_buffer + redis.call("LTRIM", data_key, trim, -1) + local members = redis.call("SMEMBERS", consumers_key) + for _, cid in ipairs(members) do + local ok = off_prefix .. cid + local cur_off = tonumber(redis.call("GET", ok) or "0") + local new_off = cur_off - trim + if new_off < 0 then new_off = 0 end + redis.call("SET", ok, new_off) + end + end +end + +return unseen +`) + +// removeConsumerScript drops the given consumer, evicts other stale +// consumers, then compacts the data list to the minimum offset still in use. +// When the consumer set becomes empty the data list is deleted entirely. +// +// KEYS[1] = data:{} (LIST) +// KEYS[2] = consumers:{} (SET) +// ARGV[1] = consumerID +// ARGV[2] = nowUnixNano +// ARGV[3] = evictionTTLNanos +// ARGV[4] = maxBuffer +// ARGV[5] = offsetKeyPrefix +// ARGV[6] = seenKeyPrefix +// +// Returns: integer number of entries trimmed from the data list. +var removeConsumerScript = redis.NewScript(` +local data_key = KEYS[1] +local consumers_key = KEYS[2] +local consumer = ARGV[1] +local now = tonumber(ARGV[2]) +local eviction_ttl = tonumber(ARGV[3]) +local max_buffer = tonumber(ARGV[4]) +local off_prefix = ARGV[5] +local seen_prefix = ARGV[6] + +redis.call("SREM", consumers_key, consumer) +redis.call("DEL", off_prefix .. consumer) +redis.call("DEL", seen_prefix .. consumer) + +if eviction_ttl > 0 then + local members = redis.call("SMEMBERS", consumers_key) + for _, cid in ipairs(members) do + local sk = seen_prefix .. cid + local ls = tonumber(redis.call("GET", sk) or "0") + if ls > 0 and (now - ls) > eviction_ttl then + redis.call("SREM", consumers_key, cid) + redis.call("DEL", sk) + redis.call("DEL", off_prefix .. cid) + end + end +end + +local live = redis.call("SCARD", consumers_key) +if live == 0 then + redis.call("DEL", data_key) + return 0 +end + +local members = redis.call("SMEMBERS", consumers_key) +local min_off = -1 +for _, cid in ipairs(members) do + local o = tonumber(redis.call("GET", off_prefix .. cid) or "0") + if min_off < 0 or o < min_off then min_off = o end +end + +local trimmed = 0 +if min_off > 0 then + redis.call("LTRIM", data_key, min_off, -1) + for _, cid in ipairs(members) do + local ok = off_prefix .. cid + local cur_off = tonumber(redis.call("GET", ok) or "0") + local new_off = cur_off - min_off + if new_off < 0 then new_off = 0 end + redis.call("SET", ok, new_off) + end + trimmed = min_off +end + +if max_buffer > 0 then + local cur_len = redis.call("LLEN", data_key) + if cur_len > max_buffer then + local extra = cur_len - max_buffer + redis.call("LTRIM", data_key, extra, -1) + for _, cid in ipairs(members) do + local ok = off_prefix .. cid + local cur_off = tonumber(redis.call("GET", ok) or "0") + local new_off = cur_off - extra + if new_off < 0 then new_off = 0 end + redis.call("SET", ok, new_off) + end + trimmed = trimmed + extra + end +end + +return trimmed +`) diff --git a/pkg/storage/storage_redis_test.go b/pkg/storage/storage_redis_test.go new file mode 100644 index 00000000..3ba342f9 --- /dev/null +++ b/pkg/storage/storage_redis_test.go @@ -0,0 +1,257 @@ +package storage + +import ( + "testing" + "time" + + "github.com/alicebob/miniredis/v2" + jsoniter "github.com/json-iterator/go" + "github.com/google/uuid" + "github.com/redis/go-redis/v9" + "github.com/rs/xid" + "github.com/stretchr/testify/require" +) + +// newTestRedisStorage spins a miniredis server, wires a go-redis client to it +// and returns a configured Redis-backed Storage plus a teardown function. +func newTestRedisStorage(t *testing.T, opts ...func(*RedisOptions)) (*StorageRedis, func()) { + t.Helper() + mr := miniredis.RunT(t) + client := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + cfg := &RedisOptions{ + Client: client, + EvictionTTL: time.Hour, + EvictionStrategy: EvictionStrategySliding, + MaxSharedInteractions: 64, + } + for _, opt := range opts { + opt(cfg) + } + store, err := NewRedis(cfg) + require.NoError(t, err) + return store, func() { + _ = store.Close() + } +} + +// TestRedisFullRoundTrip mirrors TestFullRoundTripInMemory/Disk so we can +// verify the encrypted client-decryption path works against Redis. +func TestRedisFullRoundTrip(t *testing.T) { + store, teardown := newTestRedisStorage(t) + defer teardown() + + priv, pubKeyB64 := generateRSAKeyPair(t) + secret := uuid.New().String() + correlationID := xid.New().String() + + require.NoError(t, store.SetIDPublicKey(correlationID, secret, pubKeyB64)) + + for i := 0; i < 3; i++ { + inter := &interaction{ + Protocol: "dns", + UniqueID: "abc123def456ghi", + FullId: "abc123def456ghi.oast.fun", + QType: "A", + RawRequest: dnsRequest, + RawResponse: dnsResponse, + RemoteAddress: "10.0.0.1", + Timestamp: time.Now(), + } + data, err := jsoniter.Marshal(inter) + require.NoError(t, err) + require.NoError(t, store.AddInteraction(correlationID, data)) + } + + data, aesKey, err := store.GetInteractions(correlationID, secret) + require.NoError(t, err) + require.Len(t, data, 3) + + for i, d := range data { + plaintext := clientDecrypt(t, priv, aesKey, d) + result := &interaction{} + require.NoError(t, jsoniter.Unmarshal(plaintext, result), "interaction %d", i) + require.Equal(t, "dns", result.Protocol) + require.Equal(t, dnsRequest, result.RawRequest) + } + + // After GetInteractions the buffer should be drained. + more, _, err := store.GetInteractions(correlationID, secret) + require.NoError(t, err) + require.Empty(t, more) +} + +// TestRedisDoubleRegister covers the "already exists" guard in +// SetIDPublicKey, mirroring StorageDB semantics. +func TestRedisDoubleRegister(t *testing.T) { + store, teardown := newTestRedisStorage(t) + defer teardown() + + _, pub := generateRSAKeyPair(t) + id := xid.New().String() + require.NoError(t, store.SetIDPublicKey(id, "secret", pub)) + err := store.SetIDPublicKey(id, "secret", pub) + require.Error(t, err) + require.Contains(t, err.Error(), "already exists") +} + +// TestRedisInvalidSecret ensures the secret check blocks reads with the +// wrong key. +func TestRedisInvalidSecret(t *testing.T) { + store, teardown := newTestRedisStorage(t) + defer teardown() + + _, pub := generateRSAKeyPair(t) + id := xid.New().String() + require.NoError(t, store.SetIDPublicKey(id, "right", pub)) + require.NoError(t, store.AddInteraction(id, []byte("payload"))) + + _, _, err := store.GetInteractions(id, "wrong") + require.Error(t, err) + require.Contains(t, err.Error(), "invalid secret") + + // The correct secret still works after a failed attempt. + data, _, err := store.GetInteractions(id, "right") + require.NoError(t, err) + require.Len(t, data, 1) +} + +// TestRedisMultiInstanceSharing is the headline test: instance A registers +// the id, instance B writes the interaction, and instance C polls it. All +// three share the same Redis backend. +func TestRedisMultiInstanceSharing(t *testing.T) { + mr := miniredis.RunT(t) + makeInstance := func() *StorageRedis { + client := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + s, err := NewRedis(&RedisOptions{ + Client: client, + EvictionTTL: time.Hour, + EvictionStrategy: EvictionStrategySliding, + }) + require.NoError(t, err) + return s + } + a, b, c := makeInstance(), makeInstance(), makeInstance() + defer func() { _ = a.Close(); _ = b.Close(); _ = c.Close() }() + + priv, pub := generateRSAKeyPair(t) + id := xid.New().String() + + require.NoError(t, a.SetIDPublicKey(id, "s", pub)) + require.NoError(t, b.AddInteraction(id, []byte(`{"protocol":"dns"}`))) + + data, aesKey, err := c.GetInteractions(id, "s") + require.NoError(t, err) + require.Len(t, data, 1) + plaintext := clientDecrypt(t, priv, aesKey, data[0]) + require.Equal(t, `{"protocol":"dns"}`, string(plaintext)) +} + +// TestRedisConsumerOffsets exercises the per-consumer read offsets used by +// the polling endpoint when multiple clients subscribe to the same id. +func TestRedisConsumerOffsets(t *testing.T) { + store, teardown := newTestRedisStorage(t) + defer teardown() + + id := xid.New().String() + require.NoError(t, store.SetID(id)) + for i := 0; i < 4; i++ { + require.NoError(t, store.AddInteractionWithId(id, []byte("evt-"))) + } + + // consumer A reads everything + dataA, err := store.GetInteractionsWithIdForConsumer(id, "A") + require.NoError(t, err) + require.Len(t, dataA, 4) + + // consumer B subscribes fresh and also sees all 4 + dataB, err := store.GetInteractionsWithIdForConsumer(id, "B") + require.NoError(t, err) + require.Len(t, dataB, 4) + + // new interactions go to both + require.NoError(t, store.AddInteractionWithId(id, []byte("evt-5"))) + a1, err := store.GetInteractionsWithIdForConsumer(id, "A") + require.NoError(t, err) + require.Len(t, a1, 1) + b1, err := store.GetInteractionsWithIdForConsumer(id, "B") + require.NoError(t, err) + require.Len(t, b1, 1) + + // A second poll for consumer A with no new data returns nothing. + again, err := store.GetInteractionsWithIdForConsumer(id, "A") + require.NoError(t, err) + require.Empty(t, again) +} + +// TestRedisRemoveConsumerCompacts ensures that after the last consumer +// disconnects, the underlying list is cleaned up. +func TestRedisRemoveConsumerCompacts(t *testing.T) { + store, teardown := newTestRedisStorage(t) + defer teardown() + + id := xid.New().String() + require.NoError(t, store.SetID(id)) + require.NoError(t, store.AddInteractionWithId(id, []byte("evt-1"))) + require.NoError(t, store.AddInteractionWithId(id, []byte("evt-2"))) + + _, err := store.GetInteractionsWithIdForConsumer(id, "A") + require.NoError(t, err) + require.NoError(t, store.RemoveConsumer(id, "A")) + + // data list should be gone (no consumers left) + llen, err := store.client.LLen(t.Context(), store.dataKey(id)).Result() + require.NoError(t, err) + require.EqualValues(t, 0, llen) +} + +// TestRedisMaxSharedInteractions verifies that the per-id buffer is trimmed +// when it exceeds the configured maximum, and that consumer offsets are +// adjusted accordingly. +func TestRedisMaxSharedInteractions(t *testing.T) { + store, teardown := newTestRedisStorage(t, func(o *RedisOptions) { + o.MaxSharedInteractions = 3 + }) + defer teardown() + + id := xid.New().String() + require.NoError(t, store.SetID(id)) + + // Two consumers subscribe. + _, err := store.GetInteractionsWithIdForConsumer(id, "A") + require.NoError(t, err) + _, err = store.GetInteractionsWithIdForConsumer(id, "B") + require.NoError(t, err) + + for i := 0; i < 6; i++ { + require.NoError(t, store.AddInteractionWithId(id, []byte("evt-"))) + } + + // consumer A reads: should see at most 3 (the buffer cap), and the cap + // is enforced after the read updates the offset. + dataA, err := store.GetInteractionsWithIdForConsumer(id, "A") + require.NoError(t, err) + require.LessOrEqual(t, len(dataA), 6) + + // The buffer length cannot exceed the configured cap. + llen, err := store.client.LLen(t.Context(), store.dataKey(id)).Result() + require.NoError(t, err) + require.LessOrEqual(t, llen, int64(3)) +} + +// TestRedisRemoveIDClearsKeyspace makes sure every related key is dropped +// when an id is explicitly deregistered. +func TestRedisRemoveIDClearsKeyspace(t *testing.T) { + store, teardown := newTestRedisStorage(t) + defer teardown() + + _, pub := generateRSAKeyPair(t) + id := xid.New().String() + require.NoError(t, store.SetIDPublicKey(id, "s", pub)) + require.NoError(t, store.AddInteraction(id, []byte("payload"))) + + require.NoError(t, store.RemoveID(id, "s")) + + exists, err := store.client.Exists(t.Context(), store.metaKey(id), store.dataKey(id)).Result() + require.NoError(t, err) + require.EqualValues(t, 0, exists) +} From 0720531f15dbbd3925d0e8ebd66e19675af3d360 Mon Sep 17 00:00:00 2001 From: Mzack9999 Date: Thu, 21 May 2026 14:27:29 +0200 Subject: [PATCH 2/3] fix unchecked Close in redis verify helper --- deploy/redis-test/verify/main.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/deploy/redis-test/verify/main.go b/deploy/redis-test/verify/main.go index 3d306f86..5a7959af 100644 --- a/deploy/redis-test/verify/main.go +++ b/deploy/redis-test/verify/main.go @@ -42,7 +42,7 @@ func run() error { if err != nil { return fmt.Errorf("client.New: %w", err) } - defer c.Close() + defer func() { _ = c.Close() }() callbackURL := c.URL() if callbackURL == "" { From 7c0626955dc7963945f9ee8891de713546bce91b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Do=C4=9Fan=20Can=20Bak=C4=B1r?= Date: Thu, 25 Jun 2026 16:33:43 +0300 Subject: [PATCH 3/3] fix(storage): use atomic counters for redis hit/miss metrics --- pkg/storage/storage_redis.go | 21 +++++++++++---------- 1 file changed, 11 insertions(+), 10 deletions(-) diff --git a/pkg/storage/storage_redis.go b/pkg/storage/storage_redis.go index b94b6f49..b1617177 100644 --- a/pkg/storage/storage_redis.go +++ b/pkg/storage/storage_redis.go @@ -16,6 +16,7 @@ import ( "errors" "fmt" "strings" + "sync/atomic" "time" "github.com/redis/go-redis/v9" @@ -173,8 +174,8 @@ func (s *StorageRedis) refreshTTL(ctx context.Context, keys ...string) { func (s *StorageRedis) GetCacheMetrics() (*CacheMetrics, error) { return &CacheMetrics{ - HitCount: s.hitCount, - MissCount: s.missCount, + HitCount: atomic.LoadUint64(&s.hitCount), + MissCount: atomic.LoadUint64(&s.missCount), }, nil } @@ -259,10 +260,10 @@ func (s *StorageRedis) addInteraction(id string, data []byte) error { return fmt.Errorf("redis exists check failed: %w", err) } if exists == 0 { - s.missCount++ + atomic.AddUint64(&s.missCount, 1) return ErrCorrelationIdNotFound } - s.hitCount++ + atomic.AddUint64(&s.hitCount, 1) aesKey, err := s.client.HGet(ctx, s.metaKey(id), "aes_key").Bytes() if err != nil && !errors.Is(err, redis.Nil) { @@ -300,10 +301,10 @@ func (s *StorageRedis) GetInteractions(correlationID, secret string) ([]string, return nil, "", fmt.Errorf("redis HMGet failed: %w", err) } if res[0] == nil { - s.missCount++ + atomic.AddUint64(&s.missCount, 1) return nil, "", ErrCorrelationIdNotFound } - s.hitCount++ + atomic.AddUint64(&s.hitCount, 1) storedSecret, _ := res[0].(string) if !strings.EqualFold(storedSecret, secret) { return nil, "", errors.New("invalid secret key passed for user") @@ -322,10 +323,10 @@ func (s *StorageRedis) GetInteractions(correlationID, secret string) ([]string, func (s *StorageRedis) GetInteractionsWithId(id string) ([]string, error) { ctx := context.Background() if exists, _ := s.client.Exists(ctx, s.metaKey(id)).Result(); exists == 0 { - s.missCount++ + atomic.AddUint64(&s.missCount, 1) return nil, errors.New("could not get id from cache") } - s.hitCount++ + atomic.AddUint64(&s.hitCount, 1) data, err := s.consumeAll(ctx, id) if err != nil { return nil, err @@ -358,10 +359,10 @@ func (s *StorageRedis) consumeAll(ctx context.Context, id string) ([]string, err func (s *StorageRedis) GetInteractionsWithIdForConsumer(id, consumerID string) ([]string, error) { ctx := context.Background() if exists, _ := s.client.Exists(ctx, s.metaKey(id)).Result(); exists == 0 { - s.missCount++ + atomic.AddUint64(&s.missCount, 1) return nil, errors.New("could not get id from cache") } - s.hitCount++ + atomic.AddUint64(&s.hitCount, 1) now := time.Now().UnixNano() evictionNanos := int64(0)