diff --git a/sublime/client.go b/sublime/client.go index 8b3e3663..8ee63c01 100755 --- a/sublime/client.go +++ b/sublime/client.go @@ -8,6 +8,9 @@ import ( "io/ioutil" "net" "net/http" + "net/url" + "strconv" + "strings" "sync" "time" @@ -67,6 +70,7 @@ func (c *SublimeConfig) Validate() error { if c.BaseURL == "" { c.BaseURL = defaultBaseURL } + c.BaseURL = strings.TrimRight(c.BaseURL, "/") if c.PollInterval <= 0 { c.PollInterval = defaultPollInterval } @@ -74,6 +78,9 @@ func (c *SublimeConfig) Validate() error { } 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 diff --git a/sublime/client_test.go b/sublime/client_test.go index 2f96cd83..72f06545 100644 --- a/sublime/client_test.go +++ b/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") }