Vulndb events followup

.travis.yml

@@ -6,7 +6,7 @@ services:
   - docker
 env:
   global:
-    - CGO_ENABLED=0
+    - CGO_ENABLED=1
     - FLYWAY_VERSION=10.10.0
     - INPUT_BUILDARGS=FLYWAY_VERSION=$FLYWAY_VERSION
     - INPUT_PLATFORM=linux/amd64  # ,linux/arm64
@@ -15,13 +15,11 @@ 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/buildx.sh)'
   - cd local_deployment

Dockerfile

@@ -1,6 +1,10 @@
 # Copyright 2020 Adevinta
 
-FROM --platform=$BUILDPLATFORM golang:1.22-alpine3.18 as builder
+FROM --platform=linux/$TARGETARCH golang:1.22-alpine3.18 as builder
+
+# Required because the dependency
+# https://github.com/confluentinc/confluent-kafka-go requires the gcc compiler.
+RUN apk add --no-cache gcc musl-dev cyrus-sasl-dev mold
 
 WORKDIR /app
 
@@ -13,13 +17,21 @@ COPY . .
 
 ARG TARGETOS TARGETARCH
 
-RUN cd cmd/vulnerability-db-consumer/ && GOOS=$TARGETOS GOARCH=$TARGETARCH go build . && cd -
+WORKDIR /app/cmd/vulnerability-db-consumer
+
+# -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 CGO_ENABLED=1 GOOS=linux GOARCH=$TARGETARCH \
+    # explicitly link to libsasl2 installed as part of cyrus-sasl-dev
+    CGO_LDFLAGS="-fuse-ld=mold -lsasl2" \
+    go build -tags musl -ldflags "-w -s" .
 
 FROM alpine:3.19
 
 WORKDIR /flyway
 
-RUN apk add --no-cache --update openjdk17-jre bash gettext
+RUN apk add --no-cache --update openjdk17-jre bash gettext cyrus-sasl libgcc
 
 ARG FLYWAY_VERSION=10.10.0
 

README.md

@@ -13,10 +13,16 @@ 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.
-The commands bellow will launch the necessary components required by the 
+The commands bellow will launch the necessary components required by the
 application.
 
 ```bash
@@ -83,8 +89,18 @@ Those are the variables you have to use:
 |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|
 |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|
+|KAFKA_SECURITY_PROTOCOL|Security protocol|SASL_SSL|
+|KAFKA_SASL_MECHANISM|SALSL_mechanism|SCRAM-SHA-256|
 
 ```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,29 @@ type dbConfig struct {
 }
 
 type sqsConfig struct {
-	NProcessors uint8 `toml:"number_of_processors"`
-	WaitTime    int   `toml:"wait_time"`
-	Timeout     int
+	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"`
+	SecurityProtocol string `toml:"security_protocol"`
+	SASLMechanism    string `toml:"sasl_mechanism"`
+}
+
 type reportConfig struct {
 	URLReplace string `toml:"url_replace"`
 }
@@ -74,6 +85,9 @@ func parseConfig(cfgFilePath string) (*config, error) {
 	defer cfgFile.Close()
 
 	cfgData, err := io.ReadAll(cfgFile)
+	if err != nil {
+		return nil, err
+	}
 
 	var conf config
 	if _, err := toml.Decode(string(cfgData[:]), &conf); err != nil {

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, conf.Kafka.SecurityProtocol, conf.Kafka.SASLMechanism)
+	if err != nil {
+		return nil, err
+	}
+	return notify.NewKafkaNotifier(kafkaCli, logger), nil
+}
+
 func setupLogger(cfg config) *log.Logger {
 	var logger = log.New()
 

cmd/vulnerability-db-rstats/config.go

@@ -40,6 +40,9 @@ func parseConfig(cfgFilePath string) (*config, error) {
 	defer cfgFile.Close()
 
 	cfgData, err := io.ReadAll(cfgFile)
+	if err != nil {
+		return nil, err
+	}
 
 	var conf config
 	if _, err := toml.Decode(string(cfgData[:]), &conf); err != nil {

config.toml

@@ -22,10 +22,19 @@ 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"
+security_protocol = "$KAFKA_SECURITY_PROTOCOL"
+sasl_mechanism = "$KAFKA_SASL_MECHANISM"
+
 [report]
 url_replace = "$RESULTS_URL|$RESULTS_INTERNAL_URL"