Refactor vulndb events

.travis.yml

@@ -6,20 +6,17 @@ services:
   - docker
 env:
   global:
-    - CGO_ENABLED=0
     - FLYWAY_VERSION=9.4.0
     - INPUT_BUILDARGS=FLYWAY_VERSION=$FLYWAY_VERSION
 before_install:
   # Requirement for 'test-local-deployment'
   - pip install --user awscli
   - export PATH=$PATH:$HOME/.local/bin
-before_script:
-  - _script/start-pg
 gobuild_args: -a -tags netgo -ldflags '-w'
 go_import_path: github.com/adevinta/vulnerability-db
 script:
   - go install ./...
-  - go test -v -tags integration $(go list ./... | grep -v /vendor/) ./test
+  - _script/test
 after_success:
   - bash -c 'source <(curl -s https://raw.githubusercontent.com/adevinta/vulcan-cicd/master/docker.sh)'
   - cd local_deployment

Dockerfile

@@ -2,6 +2,10 @@
 
 FROM golang:1.19.3-alpine3.15 as builder
 
+# Required because the dependency
+# https://github.com/confluentinc/confluent-kafka-go requires the gcc compiler.
+RUN apk add gcc libc-dev
+
 WORKDIR /app
 
 COPY go.mod .
@@ -11,7 +15,9 @@ RUN go mod download
 
 COPY . .
 
-RUN cd cmd/vulnerability-db-consumer/ && GOOS=linux GOARCH=amd64 go build . && cd -
+# -tags musl argument is required for dependency github.com/confluentinc/confluent-kafka-go.
+# see documentation: https://github.com/confluentinc/confluent-kafka-go#using-go-modules
+RUN cd cmd/vulnerability-db-consumer/ && GOOS=linux GOARCH=amd64 go build -tags musl . && cd -
 
 FROM alpine:3.16.3
 

README.md

@@ -13,6 +13,12 @@ cd db && source flyway-migrate.sh && cd -
 vulnerability-db-consumer -c _resources/config/local.toml
 ```
 
+## How to generate AsyncAPI documentation
+Generated [AsyncAPI documentation](https://www.asyncapi.com/) can be found in `./docs` directory.
+```
+cd pkg/asyncapi/_gen && ./gen.sh && cd -
+```
+
 ## How to run the Vulnerability DB in development mode
 
 You can test the Vulnerability DB Consumer locally in your machine.
@@ -78,13 +84,19 @@ Those are the variables you have to use:
 |PG_PORT|Database port|5432|
 |PG_SSLMODE|One of these (disable,allow,prefer,require,verify-ca,verify-full)|disable|
 |PG_CA_B64|A base64 encoded CA certificate||
-|SQS_NUMBER_OF_PROCESSORS|Number of concurrent SQS processors|Default: 10|
-|SQS_QUEUE_ARN|Checks queueu ARN|arn:aws:sqs:xxx:123456789012:yyy|
-|SNS_TOPIC_ARN|ARN of topic to publish new vulnerabilities|arn:aws:sns:xxx:123456789012:yyy|
 |RESULTS_URL|External vulcan-results URL|https://results.vulcan.com|
 |RESULTS_INTERNAL_URL|Internal vulcan-results URL|http://vulcan-results|
+|SQS_QUEUE_ARN|Checks queueu ARN|arn:aws:sqs:xxx:123456789012:yyy|
+|SQS_NUMBER_OF_PROCESSORS|Number of concurrent SQS processors|Default: 10|
 |AWS_SQS_ENDPOINT|Endpoint for SQS creation queue (optional)|http://custom-aws-endpoint|
+|SNS_ENABLED|Enables/Disables notifications sent to SNS|false|
+|SNS_TOPIC_ARN|ARN of topic to publish new vulnerabilities|arn:aws:sns:xxx:123456789012:yyy|
 |AWS_SNS_ENDPOINT|Endpoint for SNS topic (optional)|http://custom-aws-endpoint|
+|KAFKA_ENABLED|Enables/Disables notifications sent to Kafka|false|
+|KAFKA_USER|Kafka user||
+|KAFKA_PASSWORD|Kafka password||
+|KAFKA_BROKER_URL|Kafka Broker URL|localhost:9092|
+|KAFKA_TOPIC|Kafka topic|findings|
 
 ```bash
 docker build . -t vdb

_resources/config/local.toml.example

@@ -18,8 +18,15 @@ timeout = 30
 queue_arn = "arn:aws:sqs:xxx:123456789012:yyy"
 
 [sns]
+enabled = false
 topic_arn = "arn:aws:sns:xxx:123456789012:yyy"
-enabled = true
+
+[kafka]
+enabled = false
+user = "user"
+password = "password"
+broker_url = "localhost:9092"
+topic = "findings"
 
 [report]
 url_replace = "https://results.vulcan.example.com|http://localhost:8081"

cmd/vulnerability-db-consumer/config.go

@@ -19,6 +19,7 @@ type config struct {
 	DB          dbConfig
 	SQS         sqsConfig
 	SNS         snsConfig
+	Kafka       kafkaConfig
 	Report      reportConfig
 	Maintenance maintenanceConfig
 }
@@ -38,19 +39,27 @@ type dbConfig struct {
 }
 
 type sqsConfig struct {
-	NProcessors uint8 `toml:"number_of_processors"`
-	WaitTime    uint8 `toml:"wait_time"`
-	Timeout     uint8
+	NProcessors uint `toml:"number_of_processors"`
+	WaitTime    uint `toml:"wait_time"`
+	Timeout     uint
 	QueueARN    string `toml:"queue_arn"`
 	Endpoint    string `toml:"endpoint"`
 }
 
 type snsConfig struct {
-	TopicARN string `toml:"topic_arn"`
 	Enabled  bool
+	TopicARN string `toml:"topic_arn"`
 	Endpoint string `toml:"endpoint"`
 }
 
+type kafkaConfig struct {
+	Enabled   bool   `toml:"enabled"`
+	User      string `toml:"user"`
+	Pass      string `toml:"password"`
+	BrokerURL string `toml:"broker_url"`
+	Topic     string `toml:"topic"`
+}
+
 type reportConfig struct {
 	URLReplace string `toml:"url_replace"`
 }

cmd/vulnerability-db-consumer/main.go

@@ -11,6 +11,7 @@ import (
 	"os"
 	"sync"
 
+	"github.com/adevinta/vulnerability-db/pkg/asyncapi/kafka"
 	"github.com/adevinta/vulnerability-db/pkg/maintenance"
 	"github.com/adevinta/vulnerability-db/pkg/notify"
 	"github.com/adevinta/vulnerability-db/pkg/processor"
@@ -43,14 +44,9 @@ func main() {
 	}
 
 	// Build notifier.
-	snsConf := notify.SNSConfig{
-		TopicArn: conf.SNS.TopicARN,
-		Enabled:  conf.SNS.Enabled,
-		Endpoint: conf.SNS.Endpoint,
-	}
-	snsNotifier, err := notify.NewSNSNotifier(snsConf, logger)
+	notifier, err := buildNotifier(conf, logger)
 	if err != nil {
-		log.Fatalf("Error creating notifier: %v", err)
+		log.Fatalf("Error building notifier: %v", err)
 	}
 
 	// Build processor.
@@ -59,7 +55,7 @@ func main() {
 		log.Fatalf("Error creating results client: %v", err)
 	}
 
-	processor, err := processor.NewCheckProcessor(snsNotifier, db, resultsClient, conf.Report.URLReplace, conf.MaxEventAge, logger)
+	processor, err := processor.NewCheckProcessor(notifier, db, resultsClient, conf.Report.URLReplace, conf.MaxEventAge, logger)
 	if err != nil {
 		log.Fatalf("Error creating queue processor: %v", err)
 	}
@@ -97,6 +93,52 @@ func main() {
 	wg.Wait()
 }
 
+// buildNotifier builds the appropiate notifier given the defined configuration.
+// TODO: Once the integrations dependent on the old notification format have been
+// deprecated or updated to comply with the new format through Kafka topic channel
+// we can get rid of SNS and multi implementations of notifier and just use Kafka.
+func buildNotifier(conf *config, logger *log.Logger) (notify.Notifier, error) {
+	if !conf.SNS.Enabled && !conf.Kafka.Enabled {
+		logger.Info("using noop notifier")
+		return notify.NewNoopNotifier(), nil
+	}
+	if conf.SNS.Enabled && !conf.Kafka.Enabled {
+		logger.Info("using SNS notifier")
+		return buildSNSNotifier(conf, logger)
+	}
+	if !conf.SNS.Enabled && conf.Kafka.Enabled {
+		logger.Info("using Kafka notifier")
+		return buildKafkaNotifier(conf, logger)
+	}
+	// Multi Notifier
+	logger.Info("using multi notifier")
+	k, err := buildKafkaNotifier(conf, logger)
+	if err != nil {
+		return nil, err
+	}
+	s, err := buildSNSNotifier(conf, logger)
+	if err != nil {
+		return nil, err
+	}
+	return notify.NewMultiNotifier(k, s), nil
+}
+
+func buildSNSNotifier(conf *config, logger *log.Logger) (*notify.SNSNotifier, error) {
+	return notify.NewSNSNotifier(notify.SNSConfig{
+		TopicArn: conf.SNS.TopicARN,
+		Endpoint: conf.SNS.Endpoint,
+	}, logger)
+}
+
+func buildKafkaNotifier(conf *config, logger *log.Logger) (*notify.KafkaNotifier, error) {
+	kafkaCli, err := kafka.NewClient(conf.Kafka.User, conf.Kafka.Pass,
+		conf.Kafka.BrokerURL, conf.Kafka.Topic)
+	if err != nil {
+		return nil, err
+	}
+	return notify.NewKafkaNotifier(kafkaCli, logger), nil
+}
+
 func setupLogger(cfg config) *log.Logger {
 	var logger = log.New()
 

config.toml

@@ -17,15 +17,22 @@ name = "$PG_NAME"
 [sqs]
 number_of_processors = $SQS_NUMBER_OF_PROCESSORS
 wait_time = 20
-timeout = 30
+timeout = 60
 queue_arn = "$SQS_QUEUE_ARN"
 endpoint = "$AWS_SQS_ENDPOINT"
 
 [sns]
+enabled = $SNS_ENABLED
 topic_arn = "$SNS_TOPIC_ARN"
-enabled = true
 endpoint = "$AWS_SNS_ENDPOINT"
 
+[kafka]
+enabled = $KAFKA_ENABLED
+user = "$KAFKA_USER"
+password = "$KAFKA_PASSWORD"
+broker_url = "$KAFKA_BROKER_URL"
+topic = "$KAFKA_TOPIC"
+
 [report]
 url_replace = "$RESULTS_URL|$RESULTS_INTERNAL_URL"