diff --git a/README.md b/README.md index 6f31f0cc..b5b8f441 100644 --- a/README.md +++ b/README.md @@ -700,6 +700,13 @@ $ interactsh-server -domain hackwithautomation.com -wildcard [DNS] Listening on UDP 157.230.223.165:53 ``` +In wildcard mode, each connected client independently receives all interactions for the shared domain. The server keeps a buffer of recent interactions so that multiple clients can poll without missing data. By default, the buffer holds up to **10,000** interactions per shared key. This can be adjusted via the `INTERACTSH_MAX_SHARED_INTERACTIONS` environment variable: + +```console +$ export INTERACTSH_MAX_SHARED_INTERACTIONS=50000 +$ interactsh-server -domain hackwithautomation.com -wildcard +``` + ## LDAP Interaction As default, Interactsh server support LDAP interaction for the payload included in [search query](https://ldapwiki.com/wiki/LDAP%20Query%20Examples), additionally `ldap` flag can be used for complete logging. diff --git a/cmd/interactsh-server/main.go b/cmd/interactsh-server/main.go index 09bf0b51..8e832ce4 100644 --- a/cmd/interactsh-server/main.go +++ b/cmd/interactsh-server/main.go @@ -254,6 +254,11 @@ func main() { } storeOptions.DbPath = cliOptions.DiskStoragePath } + if v := os.Getenv("INTERACTSH_MAX_SHARED_INTERACTIONS"); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 { + storeOptions.MaxSharedInteractions = n + } + } var err error store, err = storage.New(&storeOptions) diff --git a/pkg/server/http_server.go b/pkg/server/http_server.go index a0a06d1e..6cbd29bc 100644 --- a/pkg/server/http_server.go +++ b/pkg/server/http_server.go @@ -424,6 +424,14 @@ func (h *HTTPServer) deregisterHandler(w http.ResponseWriter, req *http.Request) jsonError(w, fmt.Sprintf("could not remove id: %s", err), http.StatusBadRequest) return } + if h.options.RootTLD { + for _, domain := range h.options.Domains { + _ = h.options.Storage.RemoveConsumer(domain, r.CorrelationID) + } + } + if h.options.Token != "" { + _ = h.options.Storage.RemoveConsumer(h.options.Token, r.CorrelationID) + } jsonMsg(w, "deregistration successful", http.StatusOK) gologger.Debug().Msgf("Deregistered correlationID %s for key\n", r.CorrelationID) } @@ -460,14 +468,14 @@ func (h *HTTPServer) pollHandler(w http.ResponseWriter, req *http.Request) { var tlddata, extradata []string if h.options.RootTLD { for _, domain := range h.options.Domains { - interactions, _ := h.options.Storage.GetInteractionsWithId(domain) + interactions, _ := h.options.Storage.GetInteractionsWithIdForConsumer(domain, ID) // root domains interaction are not encrypted tlddata = append(tlddata, interactions...) } } if h.options.Token != "" { // auth token interactions are not encrypted - extradata, _ = h.options.Storage.GetInteractionsWithId(h.options.Token) + extradata, _ = h.options.Storage.GetInteractionsWithIdForConsumer(h.options.Token, ID) } response := &PollResponse{Data: data, AESKey: aesKey, TLDData: tlddata, Extra: extradata} diff --git a/pkg/storage/option.go b/pkg/storage/option.go index 04bd0f18..5d32a5ff 100644 --- a/pkg/storage/option.go +++ b/pkg/storage/option.go @@ -10,17 +10,21 @@ const ( ) type Options struct { - DbPath string - EvictionTTL time.Duration - MaxSize int - EvictionStrategy EvictionStrategy + DbPath string + EvictionTTL time.Duration + MaxSize int + MaxSharedInteractions int + EvictionStrategy EvictionStrategy } func (options *Options) UseDisk() bool { return options.DbPath != "" } +const defaultMaxSharedInteractions = 10000 + var DefaultOptions = Options{ - MaxSize: 2500000, - EvictionStrategy: EvictionStrategySliding, + MaxSize: 2500000, + MaxSharedInteractions: defaultMaxSharedInteractions, + EvictionStrategy: EvictionStrategySliding, } diff --git a/pkg/storage/storage.go b/pkg/storage/storage.go index d47dc502..60e582b4 100644 --- a/pkg/storage/storage.go +++ b/pkg/storage/storage.go @@ -9,6 +9,8 @@ type Storage interface { AddInteractionWithId(id string, data []byte) error GetInteractions(correlationID, secret string) ([]string, string, error) GetInteractionsWithId(id string) ([]string, error) + GetInteractionsWithIdForConsumer(id, consumerID string) ([]string, error) + RemoveConsumer(id, consumerID string) error RemoveID(correlationID, secret string) error GetCacheItem(token string) (*CorrelationData, error) Close() error diff --git a/pkg/storage/storagedb.go b/pkg/storage/storagedb.go index 019f9822..07ac2d46 100644 --- a/pkg/storage/storagedb.go +++ b/pkg/storage/storagedb.go @@ -10,6 +10,7 @@ import ( "os" "path/filepath" "strings" + "time" "github.com/goburrow/cache" "github.com/pkg/errors" @@ -32,6 +33,9 @@ type StorageDB struct { // New creates a new storage instance for interactsh data. func New(options *Options) (*StorageDB, error) { + if options.MaxSharedInteractions <= 0 { + options.MaxSharedInteractions = defaultMaxSharedInteractions + } storageDB := &StorageDB{Options: options} cacheOptions := []cache.Option{ cache.WithMaximumSize(options.MaxSize), @@ -240,6 +244,184 @@ func (s *StorageDB) GetInteractionsWithId(id string) ([]string, error) { return s.getInteractions(value, id) } +// GetInteractionsWithIdForConsumer returns unseen interactions for a consumer +// using per-consumer read offsets. +func (s *StorageDB) GetInteractionsWithIdForConsumer(id, consumerID string) ([]string, error) { + item, ok := s.cache.GetIfPresent(id) + if !ok { + return nil, errors.New("could not get id from cache") + } + value, ok := item.(*CorrelationData) + if !ok { + return nil, errors.New("invalid id cache value found") + } + + value.Lock() + defer value.Unlock() + + if value.ReadOffsets == nil { + value.ReadOffsets = make(map[string]int) + } + if value.LastSeen == nil { + value.LastSeen = make(map[string]time.Time) + } + + var allData []string + switch { + case s.Options.UseDisk(): + raw, err := s.db.Get([]byte(id), nil) + if err != nil { + if errors.Is(err, leveldb.ErrNotFound) { + return nil, nil + } + return nil, err + } + for _, d := range bytes.Split(raw, []byte("\n")) { + if len(d) > 0 { + allData = append(allData, string(d)) + } + } + default: + allData = value.Data + } + + offset := min(value.ReadOffsets[consumerID], len(allData)) + + var unseen []string + if offset < len(allData) { + unseen = make([]string, len(allData)-offset) + copy(unseen, allData[offset:]) + } + + value.ReadOffsets[consumerID] = len(allData) + value.LastSeen[consumerID] = time.Now() + + s.evictAndEnforceBuffer(value, id) + + return unseen, nil +} + +// RemoveConsumer removes a consumer's read offset and compacts consumed data. +func (s *StorageDB) RemoveConsumer(id, consumerID string) error { + item, ok := s.cache.GetIfPresent(id) + if !ok { + return nil + } + value, ok := item.(*CorrelationData) + if !ok { + return nil + } + + value.Lock() + defer value.Unlock() + + delete(value.ReadOffsets, consumerID) + delete(value.LastSeen, consumerID) + + s.compactConsumedData(value, id) + return nil +} + +func (s *StorageDB) evictAndEnforceBuffer(value *CorrelationData, id string) { + s.evictStaleConsumers(value, id) + s.enforceMaxBuffer(value, id) +} + +func (s *StorageDB) compactConsumedData(value *CorrelationData, id string) { + s.evictStaleConsumers(value, id) + if len(value.ReadOffsets) == 0 { + return + } + + minOffset := -1 + for _, off := range value.ReadOffsets { + if minOffset < 0 || off < minOffset { + minOffset = off + } + } + if minOffset > 0 { + s.applyTrim(value, id, minOffset) + } + s.enforceMaxBuffer(value, id) +} + +func (s *StorageDB) evictStaleConsumers(value *CorrelationData, id string) { + if s.Options.EvictionTTL > 0 { + now := time.Now() + for cid, lastSeen := range value.LastSeen { + if now.Sub(lastSeen) > s.Options.EvictionTTL { + delete(value.ReadOffsets, cid) + delete(value.LastSeen, cid) + } + } + } + + if len(value.ReadOffsets) == 0 { + value.Data = nil + if s.Options.UseDisk() { + _ = s.db.Delete([]byte(id), nil) + } + } +} + +func (s *StorageDB) enforceMaxBuffer(value *CorrelationData, id string) { + dataLen := s.dataLen(value, id) + if dataLen <= s.Options.MaxSharedInteractions { + return + } + excess := dataLen - s.Options.MaxSharedInteractions + s.applyTrim(value, id, excess) +} + +func (s *StorageDB) applyTrim(value *CorrelationData, id string, trimCount int) { + switch { + case s.Options.UseDisk(): + raw, err := s.db.Get([]byte(id), nil) + if err != nil { + return + } + var allData []string + for _, d := range bytes.Split(raw, []byte("\n")) { + if len(d) > 0 { + allData = append(allData, string(d)) + } + } + if trimCount >= len(allData) { + _ = s.db.Delete([]byte(id), nil) + } else { + remaining := allData[trimCount:] + _ = s.db.Put([]byte(id), []byte(strings.Join(remaining, "\n")), nil) + } + default: + if trimCount >= len(value.Data) { + value.Data = nil + } else { + value.Data = value.Data[trimCount:] + } + } + + for cid, off := range value.ReadOffsets { + value.ReadOffsets[cid] = max(off-trimCount, 0) + } +} + +func (s *StorageDB) dataLen(value *CorrelationData, id string) int { + if s.Options.UseDisk() { + raw, err := s.db.Get([]byte(id), nil) + if err != nil { + return 0 + } + count := 0 + for _, d := range bytes.Split(raw, []byte("\n")) { + if len(d) > 0 { + count++ + } + } + return count + } + return len(value.Data) +} + // RemoveID removes data for a correlation ID and data related to it. func (s *StorageDB) RemoveID(correlationID, secret string) error { item, ok := s.cache.GetIfPresent(correlationID) diff --git a/pkg/storage/storagedb_test.go b/pkg/storage/storagedb_test.go index 45e564ed..0daffdd7 100644 --- a/pkg/storage/storagedb_test.go +++ b/pkg/storage/storagedb_test.go @@ -134,6 +134,203 @@ func doStuffWithOtherCache(cache cache.Cache) { } } +func TestGetInteractionsWithIdForConsumer(t *testing.T) { + t.Run("two consumers independently receive all interactions", func(t *testing.T) { + mem, err := New(&Options{EvictionTTL: 1 * time.Hour}) + require.NoError(t, err) + defer mem.Close() + + require.NoError(t, mem.SetID("shared")) + require.NoError(t, mem.AddInteractionWithId("shared", []byte("interaction-1"))) + require.NoError(t, mem.AddInteractionWithId("shared", []byte("interaction-2"))) + + dataA, err := mem.GetInteractionsWithIdForConsumer("shared", "consumer-a") + require.NoError(t, err) + require.Equal(t, []string{"interaction-1", "interaction-2"}, dataA) + + dataB, err := mem.GetInteractionsWithIdForConsumer("shared", "consumer-b") + require.NoError(t, err) + require.Equal(t, []string{"interaction-1", "interaction-2"}, dataB) + }) + + t.Run("subsequent poll returns only unseen data", func(t *testing.T) { + mem, err := New(&Options{EvictionTTL: 1 * time.Hour}) + require.NoError(t, err) + defer mem.Close() + + require.NoError(t, mem.SetID("shared")) + require.NoError(t, mem.AddInteractionWithId("shared", []byte("msg-1"))) + + dataA, err := mem.GetInteractionsWithIdForConsumer("shared", "consumer-a") + require.NoError(t, err) + require.Equal(t, []string{"msg-1"}, dataA) + + require.NoError(t, mem.AddInteractionWithId("shared", []byte("msg-2"))) + require.NoError(t, mem.AddInteractionWithId("shared", []byte("msg-3"))) + + // consumer-a should only see new data + dataA2, err := mem.GetInteractionsWithIdForConsumer("shared", "consumer-a") + require.NoError(t, err) + require.Equal(t, []string{"msg-2", "msg-3"}, dataA2) + + // consumer-b sees everything on first poll + dataB, err := mem.GetInteractionsWithIdForConsumer("shared", "consumer-b") + require.NoError(t, err) + require.Equal(t, []string{"msg-1", "msg-2", "msg-3"}, dataB) + }) + + t.Run("empty poll returns nil", func(t *testing.T) { + mem, err := New(&Options{EvictionTTL: 1 * time.Hour}) + require.NoError(t, err) + defer mem.Close() + + require.NoError(t, mem.SetID("shared")) + + data, err := mem.GetInteractionsWithIdForConsumer("shared", "consumer-a") + require.NoError(t, err) + require.Nil(t, data) + }) + + t.Run("RemoveConsumer compacts data read by remaining consumers", func(t *testing.T) { + mem, err := New(&Options{EvictionTTL: 1 * time.Hour}) + require.NoError(t, err) + defer mem.Close() + + require.NoError(t, mem.SetID("shared")) + require.NoError(t, mem.AddInteractionWithId("shared", []byte("msg-1"))) + require.NoError(t, mem.AddInteractionWithId("shared", []byte("msg-2"))) + + // Both consumers read all data + _, err = mem.GetInteractionsWithIdForConsumer("shared", "consumer-a") + require.NoError(t, err) + _, err = mem.GetInteractionsWithIdForConsumer("shared", "consumer-b") + require.NoError(t, err) + + // Remove consumer-a — consumer-b already read all, so data is compacted + require.NoError(t, mem.RemoveConsumer("shared", "consumer-a")) + + item, _ := mem.cache.GetIfPresent("shared") + value := item.(*CorrelationData) + value.Lock() + require.Nil(t, value.Data) + _, hasA := value.ReadOffsets["consumer-a"] + require.False(t, hasA) + value.Unlock() + }) + + t.Run("RemoveConsumer partial compaction", func(t *testing.T) { + mem, err := New(&Options{EvictionTTL: 1 * time.Hour}) + require.NoError(t, err) + defer mem.Close() + + require.NoError(t, mem.SetID("shared")) + for i := range 5 { + require.NoError(t, mem.AddInteractionWithId("shared", []byte("msg-"+strconv.Itoa(i)))) + } + + // consumer-a reads all 5 + dataA, err := mem.GetInteractionsWithIdForConsumer("shared", "consumer-a") + require.NoError(t, err) + require.Len(t, dataA, 5) + + // consumer-b reads 3 (of the 5 total, offset advances to 5 as well since it reads all) + // Actually, consumer-b also reads all 5 + dataB, err := mem.GetInteractionsWithIdForConsumer("shared", "consumer-b") + require.NoError(t, err) + require.Len(t, dataB, 5) + + // Remove consumer-a — consumer-b at offset 5, trim data[:5] + require.NoError(t, mem.RemoveConsumer("shared", "consumer-a")) + + item, _ := mem.cache.GetIfPresent("shared") + value := item.(*CorrelationData) + value.Lock() + require.Nil(t, value.Data) + require.Equal(t, 0, value.ReadOffsets["consumer-b"]) + value.Unlock() + }) + + t.Run("stale consumer eviction", func(t *testing.T) { + // Use long cache TTL but short consumer staleness check + mem, err := New(&Options{EvictionTTL: 1 * time.Hour}) + require.NoError(t, err) + defer mem.Close() + + require.NoError(t, mem.SetID("shared")) + require.NoError(t, mem.AddInteractionWithId("shared", []byte("msg-1"))) + + // consumer-a reads data + _, err = mem.GetInteractionsWithIdForConsumer("shared", "consumer-a") + require.NoError(t, err) + + // Manually backdate consumer-a's LastSeen to simulate staleness + item, _ := mem.cache.GetIfPresent("shared") + value := item.(*CorrelationData) + value.Lock() + value.LastSeen["consumer-a"] = time.Now().Add(-2 * time.Hour) + value.Unlock() + + require.NoError(t, mem.AddInteractionWithId("shared", []byte("msg-2"))) + + // consumer-b polls — stale consumer-a should be evicted + dataB, err := mem.GetInteractionsWithIdForConsumer("shared", "consumer-b") + require.NoError(t, err) + require.Contains(t, dataB, "msg-2") + + value.Lock() + _, hasStale := value.ReadOffsets["consumer-a"] + require.False(t, hasStale, "stale consumer should be evicted") + value.Unlock() + }) + + t.Run("max buffer cap enforced", func(t *testing.T) { + bufferCap := 100 + mem, err := New(&Options{EvictionTTL: 1 * time.Hour, MaxSharedInteractions: bufferCap}) + require.NoError(t, err) + defer mem.Close() + + require.NoError(t, mem.SetID("shared")) + + total := bufferCap + 50 + for i := range total { + require.NoError(t, mem.AddInteractionWithId("shared", []byte("msg-"+strconv.Itoa(i)))) + } + + // First poll triggers buffer cap enforcement + _, err = mem.GetInteractionsWithIdForConsumer("shared", "consumer-a") + require.NoError(t, err) + + // Verify buffer was capped after the read + item, _ := mem.cache.GetIfPresent("shared") + value := item.(*CorrelationData) + value.Lock() + require.LessOrEqual(t, len(value.Data), bufferCap) + value.Unlock() + }) + + t.Run("RemoveConsumer last consumer discards all data", func(t *testing.T) { + mem, err := New(&Options{EvictionTTL: 1 * time.Hour}) + require.NoError(t, err) + defer mem.Close() + + require.NoError(t, mem.SetID("shared")) + require.NoError(t, mem.AddInteractionWithId("shared", []byte("msg-1"))) + + _, err = mem.GetInteractionsWithIdForConsumer("shared", "consumer-a") + require.NoError(t, err) + + // Remove the only consumer — all data should be discarded + require.NoError(t, mem.RemoveConsumer("shared", "consumer-a")) + + item, _ := mem.cache.GetIfPresent("shared") + value := item.(*CorrelationData) + value.Lock() + require.Nil(t, value.Data) + require.Empty(t, value.ReadOffsets) + value.Unlock() + }) +} + func TestSlidingEvictionStrategy(t *testing.T) { testTTL := 100 * time.Millisecond smallDelay := 10 * time.Millisecond diff --git a/pkg/storage/types.go b/pkg/storage/types.go index 312faa07..c7750cbc 100644 --- a/pkg/storage/types.go +++ b/pkg/storage/types.go @@ -27,4 +27,6 @@ type CorrelationData struct { AESKeyEncrypted string `json:"aes-key"` // decrypted AES key for signing AESKey []byte `json:"-"` + ReadOffsets map[string]int `json:"-"` + LastSeen map[string]time.Time `json:"-"` }