Rename findings asyncAPI spec and generate JSON tags for models

Author: Daniel Jimenez

docs/asyncapi.yaml

@@ -14,11 +14,11 @@ channels:
     description: CDC Events of the assets stored in Vulcan.
     subscribe:
       message:
-        $ref: '#/components/messages/asset'
+        $ref: "#/components/messages/asset"
   findings:
     subscribe:
       message:
-        $ref: '#/components/messages/FindingPayload'
+        $ref: "#/components/messages/finding"
 
 components:
   messages:
@@ -33,18 +33,12 @@ components:
       contentType: application/json
       payload:
         $ref: "#/components/schemas/assetPayload"
-    FindingPayload:
+    finding:
+      headers: 
+        $ref: "#/components/schemas/findingMetadata"
       contentType: application/json
-      headers:
-        properties:
-          version:
-            description: schema version header
-            type: string
-        required:
-        - version
-        type: object
       payload:
-        $ref: '#/components/schemas/FindingPayload'
+        $ref: "#/components/schemas/findingPayload"
       summary: Events generated from Vulnerability DB findings state changes
       name: Finding
       title: Findings state
@@ -143,7 +137,7 @@ components:
         - description
         - tag
 
-    FindingPayload:
+    findingPayload:
       properties:
         affected_resource:
           type: string
@@ -156,27 +150,40 @@ components:
         impact_details:
           type: string
         issue:
-          $ref: '#/components/schemas/StoreIssueLabels'
+          $ref: '#/components/schemas/issue'
         resources:
-          $ref: '#/components/schemas/StoreResources'
+          $ref: '#/components/schemas/resources'
         score:
           type: number
         source:
-          $ref: '#/components/schemas/StoreSource'
+          $ref: '#/components/schemas/source'
         status:
           type: string
         target:
-          $ref: '#/components/schemas/StoreTargetTeams'
+          $ref: '#/components/schemas/target'
         total_exposure:
           type: integer
       type: object
-    PqStringArray:
+      additionalProperties: false
+    
+    findingMetadata:
+      type: object
+      additionalProperties: false
+      properties:
+        version:
+          type: string
+          description: The value of this field is equal to the value of the field info.version of this document.
+      required:
+        - version
+    
+    stringArray:
       items:
         type: string
       type:
       - array
       - "null"
-    StoreIssueLabels:
+    
+    issue:
       properties:
         cwe_id:
           minimum: 0
@@ -192,13 +199,15 @@ components:
           - array
           - "null"
         recommendations:
-          $ref: '#/components/schemas/PqStringArray'
+          $ref: '#/components/schemas/stringArray'
         reference_links:
-          $ref: '#/components/schemas/PqStringArray'
+          $ref: '#/components/schemas/stringArray'
         summary:
           type: string
       type: object
-    StoreResourceGroup:
+      additionalProperties: false
+    
+    resourceGroup:
       properties:
         attributes:
           items:
@@ -217,13 +226,16 @@ components:
           - array
           - "null"
       type: object
-    StoreResources:
+      additionalProperties: false
+
+    resources:
       items:
-        $ref: '#/components/schemas/StoreResourceGroup'
+        $ref: '#/components/schemas/resourceGroup'
       type:
       - array
       - "null"
-    StoreSource:
+
+    source:
       properties:
         component:
           type: string
@@ -239,7 +251,9 @@ components:
           format: date-time
           type: string
       type: object
-    StoreTargetTeams:
+      additionalProperties: false
+
+    target:
       properties:
         id:
           type: string
@@ -252,3 +266,4 @@ components:
           - array
           - "null"
       type: object
+      additionalProperties: false

pkg/api/store/cdc/parser.go

@@ -291,13 +291,21 @@ func (p *AsyncTxParser) processFindingOverwrite(data []byte) error {
 	f, err := p.VulnDBClient.Finding(context.Background(), dto.FindingOverwrite.FindingID)
 	if err != nil {
 		if errors.IsKind(err, errors.ErrNotFound) {
-			// This should not happen as distributed txs from API to VulnDB are single
-			// threaded and we already verified the finding existence in the previous step
 			return nil
 		}
 	}
-	// TODO: Can we have a conflict in relation with possible concurrent finding state
-	// changes from API finding overwrite and check processing from vulndb consumer?
+	// TODO: There can be a race condition here between two concurrent state changes for a
+	// finding between this finding overwrite and a related check processing from vulndb side.
+	// Currently this can only generate a conflict if the two concurrent events are a "mark as
+	// false positive" action initiated from Vulcan API and a finding detection event from the
+	// vulndb side which contains a fingerprint variation, as in that situation the FALSE POSITIVE
+	// state "preference" does not apply.
+	// Example:
+	// API 		-> 		mark as false positive 		-> VulnDB API
+	// API		-> 		retrieve finding state		-> VulnDB API
+	// VulnDB 	-> 		check processing 			-> Reopen false positive finding
+	// VulnDB 	-> 		push reopened finding		-> Kafka
+	// API		-> 		push false positive finding -> Kafka
 	asyncFinding := findingToAsyncFinding(f)
 	return p.asyncAPI.PushFinding(asyncFinding)
 }
@@ -419,13 +427,12 @@ func assetToAsyncAsset(a api.Asset) asyncapi.AssetPayload {
 }
 
 func findingToAsyncFinding(f *api.Finding) asyncapi.FindingPayload {
-	return asyncapi.FindingPayload{
+	findingPayload := asyncapi.FindingPayload{
 		AffectedResource: f.Finding.AffectedResource,
-		CurrentExposure:  int(f.Finding.CurrentExposure),
 		Details:          f.Finding.Details,
 		Id:               f.Finding.ID,
 		ImpactDetails:    f.Finding.ImpactDetails,
-		Issue: &asyncapi.StoreIssueLabels{
+		Issue: &asyncapi.Issue{
 			CweId:           f.Finding.Issue.CWEID,
 			Description:     f.Finding.Issue.Description,
 			Id:              f.Finding.Issue.ID,
@@ -436,7 +443,7 @@ func findingToAsyncFinding(f *api.Finding) asyncapi.FindingPayload {
 		},
 		Resources: []interface{}{f.Finding.Resources},
 		Score:     float64(f.Finding.Score),
-		Source: &asyncapi.StoreSource{
+		Source: &asyncapi.Source{
 			Component: f.Finding.Source.Component,
 			Id:        f.Finding.Source.ID,
 			Instance:  f.Finding.Source.Instance,
@@ -445,11 +452,15 @@ func findingToAsyncFinding(f *api.Finding) asyncapi.FindingPayload {
 			Time:      f.Finding.Source.Time,
 		},
 		Status: f.Finding.Status,
-		Target: &asyncapi.StoreTargetTeams{
+		Target: &asyncapi.Target{
 			Id:         f.Finding.Target.ID,
 			Identifier: f.Finding.Target.Identifier,
 			Teams:      []interface{}{f.Finding.Target.Teams},
 		},
 		TotalExposure: int(f.Finding.TotalExposure),
 	}
+	if f.Finding.OpenFinding != nil {
+		findingPayload.CurrentExposure = int(f.Finding.OpenFinding.CurrentExposure)
+	}
+	return findingPayload
 }

pkg/api/store/cdc/parser_test.go

@@ -141,6 +141,7 @@ type mockVulnDBClient struct {
 	deleteTeamTagF    func(ctx context.Context, authTeam, teamID, tag string) error
 	deleteTargetTeamF func(ctx context.Context, authTeam, targetID, teamID string) error
 	deleteTargetTagF  func(ctx context.Context, authTeam, targetID, tag string) error
+	getFindingF       func(ctx context.Context, findingID string) (*api.Finding, error)
 	updateFindingF    func(ctx context.Context, findingID string, payload *api.UpdateFinding, tag string) (*api.Finding, error)
 }
 
@@ -162,6 +163,9 @@ func (m *mockVulnDBClient) DeleteTargetTeam(ctx context.Context, authTeam, targe
 func (m *mockVulnDBClient) DeleteTargetTag(ctx context.Context, authTeam, targetID, tag string) error {
 	return m.deleteTargetTagF(ctx, authTeam, targetID, tag)
 }
+func (m *mockVulnDBClient) Finding(ctx context.Context, findingID string) (*api.Finding, error) {
+	return m.getFindingF(ctx, findingID)
+}
 func (m *mockVulnDBClient) UpdateFinding(ctx context.Context, findingID string, payload *api.UpdateFinding, tag string) (*api.Finding, error) {
 	return m.updateFindingF(ctx, findingID, payload, tag)
 }
@@ -196,14 +200,15 @@ func init() {
 
 func TestParse(t *testing.T) {
 	testCases := []struct {
-		name            string
-		log             []Event
-		vulnDBClient    *mockVulnDBClient
-		asyncAPI        func() (*asyncapi.Vulcan, kafka.Client, error)
-		loggr           *mockLoggr
-		wantNParsed     uint
-		wantAsyncAssets []testutil.AssetTopicData
-		wantErr         error
+		name              string
+		log               []Event
+		vulnDBClient      *mockVulnDBClient
+		asyncAPI          func() (*asyncapi.Vulcan, kafka.Client, error)
+		loggr             *mockLoggr
+		wantNParsed       uint
+		wantAsyncAssets   []testutil.AssetTopicData
+		wantAsyncFindings []testutil.FindingTopicData
+		wantErr           error
 	}{
 		{
 			name: "Happy path",
@@ -257,6 +262,12 @@ func TestParse(t *testing.T) {
 				deleteTargetTagF: func(ctx context.Context, authTeam, targetID, tag string) error {
 					return nil
 				},
+				getFindingF: func(ctx context.Context, findingID string) (*api.Finding, error) {
+					return &api.Finding{
+						Finding: vulndb.FindingExpanded{
+							Finding: vulndb.Finding{ID: "1"}},
+					}, nil
+				},
 				updateFindingF: func(ctx context.Context, findingID string, payload *api.UpdateFinding, tag string) (*api.Finding, error) {
 					var f = &api.Finding{}
 					return f, nil
@@ -307,6 +318,26 @@ func TestParse(t *testing.T) {
 					},
 				},
 			},
+			wantAsyncFindings: []testutil.FindingTopicData{
+				{
+					Payload: asyncapi.FindingPayload{
+						Id: "1",
+						Issue: &asyncapi.Issue{
+							Recommendations: []any{nil},
+							ReferenceLinks:  []any{nil},
+							Labels:          []any{nil},
+						},
+						Source: &asyncapi.Source{},
+						Target: &asyncapi.Target{
+							Teams: []any{nil},
+						},
+						Resources: []any{nil},
+					},
+					Headers: map[string][]byte{
+						"version": []byte(asyncapi.Version),
+					},
+				},
+			},
 			wantNParsed: 6,
 		},
 		{
@@ -431,6 +462,8 @@ func TestParse(t *testing.T) {
 			if nParsed != tc.wantNParsed {
 				t.Fatalf("expected nParsed to be %d, but got %d", tc.wantNParsed, nParsed)
 			}
+
+			// Verify async assets
 			topic := kclient.Topics[asyncapi.AssetsEntityName]
 			gotAssets, err := testutil.ReadAllAssetsTopic(topic)
 			if err != nil {
@@ -445,6 +478,20 @@ func TestParse(t *testing.T) {
 				t.Fatalf("want!=got, diff: %s", diff)
 			}
 
+			// Verify async findings
+			topic = kclient.Topics[asyncapi.FindingsEntityName]
+			gotFindings, err := testutil.ReadAllFindingsTopic(topic)
+			if err != nil {
+				t.Fatalf("error reading findings from kafka %v", err)
+			}
+			wantFindings := tc.wantAsyncFindings
+			sortSlices = cmpopts.SortSlices(func(a, b testutil.FindingTopicData) bool {
+				return strings.Compare(a.Payload.Id, b.Payload.Id) < 0
+			})
+			diff = cmp.Diff(wantFindings, gotFindings, sortSlices)
+			if diff != "" {
+				t.Fatalf("want!=got, diff: %s", diff)
+			}
 		})
 	}
 }
@@ -464,7 +511,10 @@ func (n nullLogger) Debugf(s string, params ...any) {
 }
 
 func newTestAsyncAPI() (*asyncapi.Vulcan, kafka.Client, error) {
-	topics := map[string]string{asyncapi.AssetsEntityName: "assets"}
+	topics := map[string]string{
+		asyncapi.AssetsEntityName:   "assets",
+		asyncapi.FindingsEntityName: "findings",
+	}
 	testTopics, err := testutil.PrepareKafka(topics)
 	if err != nil {
 		return nil, kafka.Client{}, fmt.Errorf("error creating test topics: %v", err)

pkg/asyncapi/_gen/gen.sh

@@ -27,3 +27,7 @@ docker run \
 	/bin/sh -c "
     npm install --silent &&
     node gen.js "${SOURCE_FILE}" "${GO_PACKAGE_NAME}""
+
+# Add JSON tags
+go install github.com/betacraft/easytags@v1.0.2
+easytags models.go

pkg/asyncapi/kafka/client.go

@@ -7,6 +7,7 @@ package kafka
 import (
 	"errors"
 	"fmt"
+	"time"
 
 	"github.com/confluentinc/confluent-kafka-go/kafka"
 )
@@ -23,6 +24,8 @@ var (
 const (
 	kafkaSecurityProtocol = "sasl_ssl"
 	kafkaSaslMechanisms   = "SCRAM-SHA-256"
+
+	maxDeliveryWait = 30 * time.Second
 )
 
 // Client implements an EventStreamClient using Kafka as the event stream
@@ -63,15 +66,18 @@ func (c *Client) Push(entity string, id string, payload []byte, metadata map[str
 	if !ok {
 		return ErrUndefinedEntity
 	}
+
 	delivered := make(chan kafka.Event)
 	defer close(delivered)
+
 	var headers []kafka.Header
 	for k, v := range metadata {
 		headers = append(headers, kafka.Header{
 			Key:   k,
 			Value: v,
 		})
 	}
+
 	msg := kafka.Message{
 		TopicPartition: kafka.TopicPartition{
 			Topic:     &topic,
@@ -85,7 +91,14 @@ func (c *Client) Push(entity string, id string, payload []byte, metadata map[str
 	if err != nil {
 		return fmt.Errorf("error producing message: %w", err)
 	}
-	e := <-delivered
+
+	var e kafka.Event
+	select {
+	case <-time.After(maxDeliveryWait):
+		return fmt.Errorf("error time out waiting for mssg delivery confirmation")
+	case e = <-delivered:
+	}
+
 	m := e.(*kafka.Message)
 	if m.TopicPartition.Error != nil {
 		return fmt.Errorf("error delivering message: %w", m.TopicPartition.Error)