fix: Sublime adapter default base URL and unbounded pagination

sublime/client.go

@@ -8,6 +8,9 @@ import (
 	"io/ioutil"
 	"net"
 	"net/http"
+	"net/url"
+	"strconv"
+	"strings"
 	"sync"
 	"time"
 
@@ -67,13 +70,17 @@ func (c *SublimeConfig) Validate() error {
 	if c.BaseURL == "" {
 		c.BaseURL = defaultBaseURL
 	}
+	c.BaseURL = strings.TrimRight(c.BaseURL, "/")
 	if c.PollInterval <= 0 {
 		c.PollInterval = defaultPollInterval
 	}
 	return nil
 }
 
 func NewSublimeAdapter(ctx context.Context, conf SublimeConfig) (*SublimeAdapter, chan struct{}, error) {
+	if err := conf.Validate(); err != nil {
+		return nil, nil, err
+	}
 	return newSublimeAdapter(ctx, conf, nil)
 }
 
@@ -88,9 +95,9 @@ func newSublimeAdapter(ctx context.Context, conf SublimeConfig, sink uspSink) (*
 		dedupe: make(map[string]int64),
 	}
 
-	// The general adapter runner constructs the adapter without calling
-	// Validate(), so backfill the poll interval default here as well --
-	// otherwise the poll loop would spin on WaitFor(0).
+	// Tests construct the adapter through here without calling Validate(),
+	// so backfill the poll interval default as well -- otherwise the poll
+	// loop would spin on WaitFor(0).
 	if a.conf.PollInterval <= 0 {
 		a.conf.PollInterval = defaultPollInterval
 	}
@@ -177,11 +184,20 @@ func (a *SublimeAdapter) makeOneRequest(since time.Time) ([]utils.Dict, time.Tim
 	var offset int
 	lastDetectionTime := since
 
+	// Only request events at or after the watermark (minus a small overlap
+	// for clock skew): without the filter every poll pages through the
+	// tenant's entire audit log history, which for high-volume tenants never
+	// completes before the HTTP timeout.
+	query := url.Values{}
+	query.Set("limit", strconv.Itoa(pageLimit))
+	query.Set("created_at[gte]", since.Add(-overlapPeriod).UTC().Format(time.RFC3339))
+
 	for {
-		url := fmt.Sprintf("%s%s?limit=%d&offset=%d", a.conf.BaseURL, logsPath, pageLimit, offset)
-		a.conf.ClientOptions.DebugLog(fmt.Sprintf("requesting from %s", url))
+		query.Set("offset", strconv.Itoa(offset))
+		reqURL := fmt.Sprintf("%s%s?%s", a.conf.BaseURL, logsPath, query.Encode())
+		a.conf.ClientOptions.DebugLog(fmt.Sprintf("requesting from %s", reqURL))
 
-		req, err := http.NewRequest("GET", url, nil)
+		req, err := http.NewRequest("GET", reqURL, nil)
 		if err != nil {
 			a.doStop.Set()
 			return nil, lastDetectionTime, err
@@ -195,15 +211,17 @@ func (a *SublimeAdapter) makeOneRequest(since time.Time) ([]utils.Dict, time.Tim
 			a.conf.ClientOptions.OnError(fmt.Errorf("http.Client.Do(): %v", err))
 			return nil, lastDetectionTime, err
 		}
-		defer resp.Body.Close()
 
 		if resp.StatusCode != http.StatusOK {
 			body, _ := ioutil.ReadAll(resp.Body)
-			a.conf.ClientOptions.OnError(fmt.Errorf("sublime api non-200: %s\nRESPONSE: %s", resp.Status, string(body)))
+			resp.Body.Close()
+			err = fmt.Errorf("sublime api non-200: %s\nRESPONSE: %s", resp.Status, string(body))
+			a.conf.ClientOptions.OnError(err)
 			return nil, lastDetectionTime, err
 		}
 
 		body, err := ioutil.ReadAll(resp.Body)
+		resp.Body.Close()
 		if err != nil {
 			a.conf.ClientOptions.OnError(fmt.Errorf("read body error: %v", err))
 			return nil, lastDetectionTime, err

sublime/client_test.go

@@ -1,8 +1,11 @@
 package usp_sublime
 
 import (
+	"context"
+	"fmt"
 	"net/http"
 	"net/http/httptest"
+	"strings"
 	"testing"
 	"time"
 
@@ -69,6 +72,41 @@ func TestValidate(t *testing.T) {
 		assert.Equal(t, "https://sublime.example.com", c.BaseURL)
 		assert.Equal(t, 5*time.Second, c.PollInterval)
 	})
+
+	t.Run("trims trailing slash from base_url", func(t *testing.T) {
+		c := SublimeConfig{
+			ClientOptions: testClientOptions(t),
+			ApiKey:        "k",
+			BaseURL:       "https://platform.sublime.security/",
+		}
+		require.NoError(t, c.Validate())
+		assert.Equal(t, "https://platform.sublime.security", c.BaseURL)
+	})
+}
+
+// TestNewSublimeAdapterValidates verifies the public constructor runs
+// Validate(): the base URL default is applied when the config omits it (the
+// bug where an empty base_url produced requests with no scheme/host), and a
+// config with no api key is rejected.
+func TestNewSublimeAdapterValidates(t *testing.T) {
+	t.Run("applies defaults", func(t *testing.T) {
+		a, chStopped, err := NewSublimeAdapter(context.Background(), SublimeConfig{
+			ClientOptions: testClientOptions(t),
+			ApiKey:        "k",
+		})
+		require.NoError(t, err)
+		assert.Equal(t, defaultBaseURL, a.conf.BaseURL)
+		assert.Equal(t, defaultPollInterval, a.conf.PollInterval)
+		require.NoError(t, a.Close())
+		<-chStopped
+	})
+
+	t.Run("rejects missing api key", func(t *testing.T) {
+		_, _, err := NewSublimeAdapter(context.Background(), SublimeConfig{
+			ClientOptions: testClientOptions(t),
+		})
+		assert.Error(t, err)
+	})
 }
 
 // TestMakeOneRequestFiltersAndAdvancesSince verifies one poll: events at or
@@ -146,9 +184,67 @@ func TestMakeOneRequestInvalidJSON(t *testing.T) {
 	assert.True(t, newSince.Equal(since), "since must not advance on a bad response")
 }
 
-// TestMakeOneRequestNon200 pins the adapter's behavior on a non-200: no items,
-// the watermark is preserved, and (a long-standing quirk) no error is returned
-// -- the failure is only reported through OnError.
+// TestMakeOneRequestSendsTimeFilter verifies each poll carries a
+// created_at[gte] filter of the watermark minus the overlap period, so the
+// API only returns new events instead of the entire audit log history.
+func TestMakeOneRequestSendsTimeFilter(t *testing.T) {
+	var gotGte string
+	now := time.Now().UTC()
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		gotGte = r.URL.Query().Get("created_at[gte]")
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(http.StatusOK)
+		_, _ = w.Write([]byte(`{"events": [], "count": 0, "total": 0}`))
+	}))
+	defer server.Close()
+
+	a := newDirectAdapter(t, server.URL)
+	_, _, err := a.makeOneRequest(now)
+	require.NoError(t, err)
+
+	require.NotEmpty(t, gotGte, "poll requests must carry a created_at[gte] filter")
+	gte, err := time.Parse(time.RFC3339, gotGte)
+	require.NoError(t, err)
+	want := now.Add(-overlapPeriod).Truncate(time.Second)
+	assert.True(t, gte.Equal(want), "created_at[gte] must be since minus the overlap period, got %v want %v", gte, want)
+}
+
+// TestMakeOneRequestPaginates verifies a full page triggers a follow-up
+// request at the next offset and the results are combined.
+func TestMakeOneRequestPaginates(t *testing.T) {
+	now := time.Now().UTC()
+	eventTime := now.Add(time.Minute).Format(time.RFC3339Nano)
+	var offsets []string
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		offsets = append(offsets, r.URL.Query().Get("offset"))
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(http.StatusOK)
+		if r.URL.Query().Get("offset") == "0" {
+			var b strings.Builder
+			b.WriteString(`{"events": [`)
+			for i := 0; i < pageLimit; i++ {
+				if i > 0 {
+					b.WriteString(",")
+				}
+				fmt.Fprintf(&b, `{"id": "evt-%d", "created_at": "%s"}`, i, eventTime)
+			}
+			fmt.Fprintf(&b, `], "count": %d, "total": %d}`, pageLimit, pageLimit+1)
+			_, _ = w.Write([]byte(b.String()))
+			return
+		}
+		_, _ = w.Write([]byte(`{"events": [{"id": "evt-last", "created_at": "` + eventTime + `"}], "count": 1, "total": 501}`))
+	}))
+	defer server.Close()
+
+	a := newDirectAdapter(t, server.URL)
+	items, _, err := a.makeOneRequest(now)
+	require.NoError(t, err)
+	assert.Len(t, items, pageLimit+1, "events from all pages must be combined")
+	assert.Equal(t, []string{"0", "500"}, offsets, "a full page must trigger a request at the next offset")
+}
+
+// TestMakeOneRequestNon200 verifies a non-200 returns no items, preserves the
+// watermark, and surfaces the failure both through OnError and as an error.
 func TestMakeOneRequestNon200(t *testing.T) {
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		w.WriteHeader(http.StatusInternalServerError)
@@ -165,7 +261,5 @@ func TestMakeOneRequestNon200(t *testing.T) {
 	assert.Nil(t, items)
 	assert.True(t, newSince.Equal(since), "since must not advance on an error response")
 	assert.Equal(t, 1, errs, "a non-200 must be reported via OnError")
-	// Pin the current behavior: the non-200 path returns a nil error (the
-	// error variable it returns belongs to the preceding, successful, Do call).
-	assert.NoError(t, err)
+	assert.Error(t, err, "a non-200 must also be returned as an error")
 }