Add lossy events to lossy feeds

Adds lossy feed checking for npm, pypi, crates and rubygems feeds using the event framework.
ossf · May 4, 2021 · cae2d39 · cae2d39
1 parent 6439e92
commit cae2d39
Show file tree

Hide file tree

Showing 10 changed files with 96 additions and 12 deletions.
diff --git a/config/scheduledfeed.go b/config/scheduledfeed.go
@@ -87,22 +87,26 @@ func AddTo(ls *[]int, value int) {
 func (config *ScheduledFeedConfig) GetScheduledFeeds() (map[string]feeds.ScheduledFeed, error) {
 	var err error
 	scheduledFeeds := map[string]feeds.ScheduledFeed{}
+	eventHandler, err := config.GetEventHandler()
+	if err != nil {
+		return nil, err
+	}
 	for _, entry := range config.EnabledFeeds {
 		switch entry {
 		case crates.FeedName:
-			scheduledFeeds[entry] = crates.Feed{}
+			scheduledFeeds[entry] = crates.New(eventHandler)
 		case goproxy.FeedName:
 			scheduledFeeds[entry] = goproxy.Feed{}
 		case npm.FeedName:
-			scheduledFeeds[entry] = npm.Feed{}
+			scheduledFeeds[entry] = npm.New(eventHandler)
 		case nuget.FeedName:
 			scheduledFeeds[entry] = nuget.Feed{}
 		case pypi.FeedName:
-			scheduledFeeds[entry] = pypi.Feed{}
+			scheduledFeeds[entry] = pypi.New(eventHandler)
 		case packagist.FeedName:
 			scheduledFeeds[entry] = packagist.Feed{}
 		case rubygems.FeedName:
-			scheduledFeeds[entry] = rubygems.Feed{}
+			scheduledFeeds[entry] = rubygems.New(eventHandler)
 		default:
 			err = fmt.Errorf("%w : %v", errUnknownFeed, entry)
 		}

diff --git a/feeds/crates/crates.go b/feeds/crates/crates.go
@@ -5,6 +5,7 @@ import (
 	"net/http"
 	"time"
 
+	"github.com/ossf/package-feeds/events"
 	"github.com/ossf/package-feeds/feeds"
 )
 
@@ -50,7 +51,15 @@ func fetchPackages() ([]*Package, error) {
 	return v.JustUpdated, nil
 }
 
-type Feed struct{}
+type Feed struct {
+	lossyFeedAlerter *feeds.LossyFeedAlerter
+}
+
+func New(eventHandler *events.Handler) *Feed {
+	return &Feed{
+		lossyFeedAlerter: feeds.NewLossyFeedAlerter(eventHandler),
+	}
+}
 
 func (feed Feed) Latest(cutoff time.Time) ([]*feeds.Package, error) {
 	pkgs := []*feeds.Package{}
@@ -62,6 +71,8 @@ func (feed Feed) Latest(cutoff time.Time) ([]*feeds.Package, error) {
 		pkg := feeds.NewPackage(pkg.UpdatedAt, pkg.Name, pkg.NewestVersion, FeedName)
 		pkgs = append(pkgs, pkg)
 	}
+	feed.lossyFeedAlerter.ProcessPackages(FeedName, pkgs)
+
 	pkgs = feeds.ApplyCutoff(pkgs, cutoff)
 	return pkgs, nil
 }
diff --git a/feeds/crates/crates_test.go b/feeds/crates/crates_test.go
@@ -5,6 +5,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/ossf/package-feeds/events"
 	"github.com/ossf/package-feeds/testutils"
 )
 
@@ -17,7 +18,7 @@ func TestCratesLatest(t *testing.T) {
 	srv := testutils.HTTPServerMock(handlers)
 
 	baseURL = srv.URL + "/api/v1/summary"
-	feed := Feed{}
+	feed := New(events.NewNullHandler())
 
 	cutoff := time.Date(1970, 1, 1, 0, 0, 0, 0, time.UTC)
 	pkgs, err := feed.Latest(cutoff)

diff --git a/feeds/lossy_logging_test.go b/feeds/lossy_logging_test.go
@@ -41,3 +41,35 @@ func TestProcessPackagesNoOverlap(t *testing.T) {
 		t.Errorf("ProcessPackages did not produce a lossy feed event")
 	}
 }
+
+func TestProcessPackagesWithOverlap(t *testing.T) {
+	t.Parallel()
+	feedName := "foo-feed"
+
+	mockSink := &events.MockSink{}
+	allowLossyFeedEventsFilter := events.NewFilter([]string{events.LOSSY_FEED_EVENT}, nil, nil)
+	eventHandler := events.NewHandler(mockSink, *allowLossyFeedEventsFilter)
+	lossyFeedAlerter := NewLossyFeedAlerter(eventHandler)
+
+	baseTime := time.Date(2021, 4, 20, 14, 30, 0, 0, time.UTC)
+	pkgs1 := []*Package{
+		NewPackage(baseTime.Add(-time.Minute*2), "foopkg", "1.0", feedName),
+		NewPackage(baseTime.Add(-time.Minute*3), "barpkg", "1.0", feedName),
+	}
+	// Populate previous packages
+	lossyFeedAlerter.ProcessPackages(feedName, pkgs1)
+
+	pkgs2 := []*Package{
+		NewPackage(baseTime, "bazpkg", "1.0", feedName),
+		NewPackage(baseTime.Add(-time.Minute*1), "quxpkg", "2.0", feedName),
+		NewPackage(baseTime.Add(-time.Minute*2), "foopkg", "1.0", feedName),
+	}
+	// Trigger overlap
+	lossyFeedAlerter.ProcessPackages(feedName, pkgs2)
+
+	evs := mockSink.GetEvents()
+
+	if len(evs) != 0 {
+		t.Fatalf("ProcessPackages failed to identify an overlap when one existed")
+	}
+}
diff --git a/feeds/npm/npm.go b/feeds/npm/npm.go
@@ -7,6 +7,7 @@ import (
 	"net/http"
 	"time"
 
+	"github.com/ossf/package-feeds/events"
 	"github.com/ossf/package-feeds/feeds"
 )
 
@@ -89,7 +90,15 @@ func fetchVersionInformation(packageName string) (string, error) {
 	return v.DistTags.Latest, nil
 }
 
-type Feed struct{}
+type Feed struct {
+	lossyFeedAlerter *feeds.LossyFeedAlerter
+}
+
+func New(eventHandler *events.Handler) *Feed {
+	return &Feed{
+		lossyFeedAlerter: feeds.NewLossyFeedAlerter(eventHandler),
+	}
+}
 
 func (feed Feed) Latest(cutoff time.Time) ([]*feeds.Package, error) {
 	pkgs := []*feeds.Package{}
@@ -105,6 +114,8 @@ func (feed Feed) Latest(cutoff time.Time) ([]*feeds.Package, error) {
 		pkg := feeds.NewPackage(pkg.CreatedDate.Time, pkg.Title, v, FeedName)
 		pkgs = append(pkgs, pkg)
 	}
+	feed.lossyFeedAlerter.ProcessPackages(FeedName, pkgs)
+
 	pkgs = feeds.ApplyCutoff(pkgs, cutoff)
 	return pkgs, nil
 }
diff --git a/feeds/npm/npm_test.go b/feeds/npm/npm_test.go
@@ -6,6 +6,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/ossf/package-feeds/events"
 	"github.com/ossf/package-feeds/testutils"
 )
 
@@ -21,7 +22,7 @@ func TestNpmLatest(t *testing.T) {
 
 	baseURL = srv.URL + "/-/rss/"
 	versionURL = srv.URL + "/"
-	feed := Feed{}
+	feed := New(events.NewNullHandler())
 
 	cutoff := time.Date(1970, 1, 1, 0, 0, 0, 0, time.UTC)
 	pkgs, err := feed.Latest(cutoff)

diff --git a/feeds/pypi/pypi.go b/feeds/pypi/pypi.go
@@ -6,6 +6,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/ossf/package-feeds/events"
 	"github.com/ossf/package-feeds/feeds"
 )
 
@@ -72,7 +73,15 @@ func fetchPackages() ([]*Package, error) {
 	return rssResponse.Packages, nil
 }
 
-type Feed struct{}
+type Feed struct {
+	lossyFeedAlerter *feeds.LossyFeedAlerter
+}
+
+func New(eventHandler *events.Handler) *Feed {
+	return &Feed{
+		lossyFeedAlerter: feeds.NewLossyFeedAlerter(eventHandler),
+	}
+}
 
 func (feed Feed) Latest(cutoff time.Time) ([]*feeds.Package, error) {
 	pkgs := []*feeds.Package{}
@@ -84,6 +93,8 @@ func (feed Feed) Latest(cutoff time.Time) ([]*feeds.Package, error) {
 		pkg := feeds.NewPackage(pkg.CreatedDate.Time, pkg.Name(), pkg.Version(), FeedName)
 		pkgs = append(pkgs, pkg)
 	}
+	feed.lossyFeedAlerter.ProcessPackages(FeedName, pkgs)
+
 	pkgs = feeds.ApplyCutoff(pkgs, cutoff)
 	return pkgs, nil
 }
diff --git a/feeds/pypi/pypi_test.go b/feeds/pypi/pypi_test.go
@@ -5,6 +5,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/ossf/package-feeds/events"
 	"github.com/ossf/package-feeds/testutils"
 )
 
@@ -17,7 +18,7 @@ func TestPypiLatest(t *testing.T) {
 	srv := testutils.HTTPServerMock(handlers)
 
 	baseURL = srv.URL + "/rss/updates.xml"
-	feed := Feed{}
+	feed := New(events.NewNullHandler())
 
 	cutoff := time.Date(1970, 1, 1, 0, 0, 0, 0, time.UTC)
 	pkgs, err := feed.Latest(cutoff)

diff --git a/feeds/rubygems/rubygems.go b/feeds/rubygems/rubygems.go
@@ -6,6 +6,7 @@ import (
 	"net/http"
 	"time"
 
+	"github.com/ossf/package-feeds/events"
 	"github.com/ossf/package-feeds/feeds"
 )
 
@@ -37,7 +38,15 @@ func fetchPackages(url string) ([]*Package, error) {
 	return response, err
 }
 
-type Feed struct{}
+type Feed struct {
+	lossyFeedAlerter *feeds.LossyFeedAlerter
+}
+
+func New(eventHandler *events.Handler) *Feed {
+	return &Feed{
+		lossyFeedAlerter: feeds.NewLossyFeedAlerter(eventHandler),
+	}
+}
 
 func (feed Feed) Latest(cutoff time.Time) ([]*feeds.Package, error) {
 	pkgs := []*feeds.Package{}
@@ -61,6 +70,8 @@ func (feed Feed) Latest(cutoff time.Time) ([]*feeds.Package, error) {
 		pkg := feeds.NewPackage(pkg.CreatedDate, pkg.Name, pkg.Version, FeedName)
 		pkgs = append(pkgs, pkg)
 	}
+	feed.lossyFeedAlerter.ProcessPackages(FeedName, pkgs)
+
 	pkgs = feeds.ApplyCutoff(pkgs, cutoff)
 	return pkgs, nil
 }
diff --git a/feeds/rubygems/rubygems_test.go b/feeds/rubygems/rubygems_test.go
@@ -5,6 +5,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/ossf/package-feeds/events"
 	"github.com/ossf/package-feeds/feeds"
 	"github.com/ossf/package-feeds/testutils"
 )
@@ -19,7 +20,7 @@ func TestRubyLatest(t *testing.T) {
 	srv := testutils.HTTPServerMock(handlers)
 
 	baseURL = srv.URL + "/api/v1/activity"
-	feed := Feed{}
+	feed := New(events.NewNullHandler())
 
 	cutoff := time.Date(1970, 1, 1, 0, 0, 0, 0, time.UTC)
 	pkgs, err := feed.Latest(cutoff)