Add support for JSONL formatted feeds (#10)

Author: ebarkie

go.mod

@@ -5,7 +5,7 @@ go 1.22.7
 toolchain go1.23.4
 
 require (
-	github.com/ebarkie/netaggr v1.3.3
+	github.com/ebarkie/netaggr v1.3.9
 	github.com/golang/protobuf v1.5.3
 	github.com/osrg/gobgp/v3 v3.32.0
 	github.com/sirupsen/logrus v1.9.3

go.sum

@@ -55,8 +55,8 @@ github.com/eapache/channels v1.1.0 h1:F1taHcn7/F0i8DYqKXJnyhJcVpp2kgFcNePxXtnyu4
 github.com/eapache/channels v1.1.0/go.mod h1:jMm2qB5Ubtg9zLd+inMZd2/NUvXgzmWXsDaLyQIGfH0=
 github.com/eapache/queue v1.1.0 h1:YOEu7KNc61ntiQlcEeUIoDTJ2o8mQznoNvUhiigpIqc=
 github.com/eapache/queue v1.1.0/go.mod h1:6eCeP0CKFpHLu8blIFXhExK/dRa7WDZfr6jVFPTqq+I=
-github.com/ebarkie/netaggr v1.3.3 h1:YXvsoSokMhh51iMHdexgGWDU/cE/OFz//b9ZOQ4Og/E=
-github.com/ebarkie/netaggr v1.3.3/go.mod h1:y1qSgnqXXH+UHFRcwgEYEt6yDK8yNc3r6uiX4KAdIAc=
+github.com/ebarkie/netaggr v1.3.9 h1:wtBE88jWyDrWERl4NvieV3+bwOHhOM0vJLAACrF8VRw=
+github.com/ebarkie/netaggr v1.3.9/go.mod h1:o7CZu4wVQE+jVIFIuRRsW+IC4fc4n1ZkwsP7HBNGJyk=
 github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98=

gobgp.go

@@ -11,13 +11,13 @@ import (
 	"slices"
 	"time"
 
-	"google.golang.org/protobuf/types/known/anypb"
 	"github.com/golang/protobuf/ptypes/any"
 	api "github.com/osrg/gobgp/v3/api"
 	gobgpconfig "github.com/osrg/gobgp/v3/pkg/config"
 	"github.com/osrg/gobgp/v3/pkg/config/oc"
 	"github.com/osrg/gobgp/v3/pkg/server"
 	log "github.com/sirupsen/logrus"
+	"google.golang.org/protobuf/types/known/anypb"
 )
 
 // NewServer creates, starts, and configures a new BGP server with configFile.

main.go

@@ -63,7 +63,6 @@ func main() {
 	defer stop()
 	signal.Notify(bh.SigC, os.Signal(syscall.SIGUSR1))
 
-	log.WithFields(log.Fields{
-		"err": bh.UpdateRoutes(ctx),
-	}).Info("Server stopped")
+	log.WithError(bh.UpdateRoutes(ctx)).Info("Server stopped")
+
 }

parse.go

@@ -5,6 +5,7 @@
 package main
 
 import (
+	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
@@ -13,6 +14,7 @@ import (
 	"net/url"
 	"os"
 	"sort"
+	"strings"
 	"sync"
 
 	"github.com/ebarkie/netaggr/pkg/netcalc"
@@ -22,30 +24,45 @@ import (
 
 var ErrUnhandledScheme = errors.New("unhandled scheme")
 
-// readFeed determines a feed's scheme and creates an appropriate io.ReadCloser.
-func readFeed(feed string) (io.ReadCloser, error) {
+// readFeed reads a feed and returns the number of networks.
+func readFeed(feed string, netC chan<- *net.IPNet) (int64, error) {
 	u, err := url.Parse(feed)
 	if err != nil {
-		return nil, err
+		return 0, err
 	}
 
 	switch u.Scheme {
 	case "":
-		return os.Open(feed)
+		r, err := os.Open(feed)
+		if err != nil {
+			return 0, err
+		}
+		defer r.Close()
+
+		if strings.HasSuffix(feed, ".json") {
+			return readFromJSON(r, netC)
+		} else {
+			return netcalc.ReadFrom(r, netC)
+		}
 	case "http", "https":
 		resp, err := http.Get(feed)
 		if err != nil {
-			return nil, err
+			return 0, err
 		}
+		defer resp.Body.Close()
 
 		if resp.StatusCode != http.StatusOK {
-			return nil, fmt.Errorf("non-OK status code: %d", resp.StatusCode)
+			return 0, fmt.Errorf("non-OK status code: %d", resp.StatusCode)
 		}
 
-		return resp.Body, nil
+		if strings.Contains(resp.Header.Get("Content-Type"), "/json") {
+			return readFromJSON(resp.Body, netC)
+		} else {
+			return netcalc.ReadFrom(resp.Body, netC)
+		}
 	}
 
-	return nil, ErrUnhandledScheme
+	return 0, fmt.Errorf("%w: %s", ErrUnhandledScheme, u.Scheme)
 }
 
 // parseFeeds parses feeds concurrently and returns summarized nets and the
@@ -64,22 +81,19 @@ func parseFeeds(feeds ...string) (nets netcalc.Nets, totalNets int) {
 		go func(feed string) {
 			defer wg.Done()
 
-			r, err := readFeed(feed)
+			i, err := readFeed(feed, netC)
 			if err != nil {
 				log.WithFields(log.Fields{
-					"err": err,
-				}).Error("Feed read error")
+					"feed": feed,
+				}).WithError(err).Error("Feed read error")
 				return
 			}
-			defer r.Close()
 
-			i, err := netcalc.ReadFrom(r, netC)
 			if err != nil {
 				log.WithFields(log.Fields{
 					"feed": feed,
 					"nets": i,
-					"err":  err,
-				}).Error("Feed parse error")
+				}).WithError(err).Error("Feed parse error")
 				return
 			}
 
@@ -102,3 +116,44 @@ func parseFeeds(feeds ...string) (nets netcalc.Nets, totalNets int) {
 
 	return
 }
+
+// sblEntry represents a Spamhaus Block List entry.
+type sblEntry struct {
+	// IP network to drop, in CIDR format.
+	CIDR string `json:"cidr"`
+
+	// Regional Internet Registry that manages the network.
+	RIR string `json:"rir"`
+
+	// Spamhaus Block List identifier.
+	SBLID string `json:"sblid"`
+}
+
+// readFromJSON parses the io.Reader as Spamhaus formatted JSONL and sends the
+// resulting IPNets to the NetC channel.
+func readFromJSON(r io.Reader, netC chan<- *net.IPNet) (int64, error) {
+	d := json.NewDecoder(r)
+	var i int64
+	for ; ; i++ {
+		var e sblEntry
+		err := d.Decode(&e)
+		if err == io.EOF {
+			break
+		} else if err != nil {
+			return i, fmt.Errorf("line %d decode: %w", i, err)
+		}
+		if e.CIDR == "" {
+			// Probably the footer line
+			continue
+		}
+
+		_, n, err := netcalc.ParseNet(e.CIDR)
+		if err != nil {
+			return i, fmt.Errorf("line %d parse %q: %w", i, e.CIDR, err)
+		}
+
+		netC <- n
+	}
+
+	return i, nil
+}