Push finding overwrite events to asyncapi channel

Author: Daniel Jimenez

README.md

@@ -99,7 +99,7 @@ Those are the variables you have to setup:
 |KAFKA_USER||user|
 |KAFKA_PASS||supersecret|
 |KAFKA_BROKER|if set to empty the Async API will be disabled|kafka.example.com:9094|
-|KAFKA_TOPICS|Contains a map, using toml format, mapping entities in the Vulcan async API to the kafka topics they wil be pushed to, by now the only available entity is ``assets`` |[assets = "assets-topic"]| 
+|KAFKA_TOPICS|Contains a map, using toml format, mapping entities in the Vulcan async API to the kafka topics they wil be pushed to, by now the only available entity is ``assets`` |{assets = "assets-topic", findings = "findings-topic"}| 
 First we have to build the `vulcan-api` because the build only copies the file.
 
 We need to provide `linux` compiled binary to the docker build command. This won't be necessary when this component has been open sourced.

_resources/config/local.toml.example

@@ -58,6 +58,12 @@ key = "a key"
 retries = 4
 retry_interval = 2 # seconds
 
+[kafka]
+user = "user"
+pass = "supersecret"
+broker = "kafka.example.com:9094"
+topics = '{assets = "assets-topic", findings = "findings-topic"}'
+
 [globalpolicy]
 # This config policy emulates code policy definition.
 [globalpolicy.web-scanning-global]

docs/asyncapi.yaml

@@ -15,6 +15,10 @@ channels:
     subscribe:
       message:
         $ref: '#/components/messages/asset'
+  findings:
+    subscribe:
+      message:
+        $ref: '#/components/messages/FindingPayload'
 
 components:
   messages:
@@ -29,6 +33,21 @@ components:
       contentType: application/json
       payload:
         $ref: "#/components/schemas/assetPayload"
+    FindingPayload:
+      contentType: application/json
+      headers:
+        properties:
+          version:
+            description: schema version header
+            type: string
+        required:
+        - version
+        type: object
+      payload:
+        $ref: '#/components/schemas/FindingPayload'
+      summary: Events generated from Vulnerability DB findings state changes
+      name: Finding
+      title: Findings state
 
   schemas:
     assetMetadata:
@@ -123,3 +142,113 @@ components:
         - name
         - description
         - tag
+    
+    FindingPayload:
+      properties:
+        affected_resource:
+          type: string
+        current_exposure:
+          type: integer
+        details:
+          type: string
+        id:
+          type: string
+        impact_details:
+          type: string
+        issue:
+          $ref: '#/components/schemas/StoreIssueLabels'
+        resources:
+          $ref: '#/components/schemas/StoreResources'
+        score:
+          type: number
+        source:
+          $ref: '#/components/schemas/StoreSource'
+        status:
+          type: string
+        target:
+          $ref: '#/components/schemas/StoreTargetTeams'
+        total_exposure:
+          type: integer
+      type: object
+    PqStringArray:
+      items:
+        type: string
+      type:
+      - array
+      - "null"
+    StoreIssueLabels:
+      properties:
+        cwe_id:
+          minimum: 0
+          type: integer
+        description:
+          type: string
+        id:
+          type: string
+        labels:
+          items:
+            type: string
+          type:
+          - array
+          - "null"
+        recommendations:
+          $ref: '#/components/schemas/PqStringArray'
+        reference_links:
+          $ref: '#/components/schemas/PqStringArray'
+        summary:
+          type: string
+      type: object
+    StoreResourceGroup:
+      properties:
+        attributes:
+          items:
+            type: string
+          type:
+          - array
+          - "null"
+        name:
+          type: string
+        resources:
+          items:
+            additionalProperties:
+              type: string
+            type: object
+          type:
+          - array
+          - "null"
+      type: object
+    StoreResources:
+      items:
+        $ref: '#/components/schemas/StoreResourceGroup'
+      type:
+      - array
+      - "null"
+    StoreSource:
+      properties:
+        component:
+          type: string
+        id:
+          type: string
+        instance:
+          type: string
+        name:
+          type: string
+        options:
+          type: string
+        time:
+          format: date-time
+          type: string
+      type: object
+    StoreTargetTeams:
+      properties:
+        id:
+          type: string
+        identifier:
+          type: string
+        teams:
+          items:
+            type: string
+          type:
+          - array
+          - "null"
+      type: object

local.env.example

@@ -41,4 +41,4 @@ GPC_1_NAME=web-scanning-global
 KAFKA_USER=user
 KAFKA_PASS=supersecret
 KAFKA_BROKER=kafka.example.com:9094
-KAFKA_TOPICS={assets = "assets-topic"}
+KAFKA_TOPICS={assets = "assets-topic", findings = "findings-topic"}

pkg/api/store/cdc/parser.go

@@ -58,6 +58,7 @@ type AsyncTxParser struct {
 type AsyncAPI interface {
 	PushAsset(asset asyncapi.AssetPayload) error
 	DeleteAsset(asset asyncapi.AssetPayload) error
+	PushFinding(finding asyncapi.FindingPayload) error
 }
 
 // NewAsyncTxParser builds a new CDC log parser to handle distributed
@@ -241,12 +242,7 @@ func (p *AsyncTxParser) processUpdateAsset(data []byte) error {
 		return errInvalidData
 	}
 	asyncAsset := assetToAsyncAsset(dto.NewAsset)
-	err = p.asyncAPI.PushAsset(asyncAsset)
-	if err != nil {
-		return err
-	}
-
-	return nil
+	return p.asyncAPI.PushAsset(asyncAsset)
 }
 
 func (p *AsyncTxParser) processDeleteAllAssets(data []byte) error {
@@ -276,6 +272,7 @@ func (p *AsyncTxParser) processFindingOverwrite(data []byte) error {
 		return errInvalidData
 	}
 
+	// Update finding in vulndb
 	_, err = p.VulnDBClient.UpdateFinding(
 		context.Background(),
 		dto.FindingOverwrite.FindingID,
@@ -289,7 +286,20 @@ func (p *AsyncTxParser) processFindingOverwrite(data []byte) error {
 		}
 		return err
 	}
-	return nil
+
+	// Retrieve current finding status and push event
+	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?
+	asyncFinding := findingToAsyncFinding(f)
+	return p.asyncAPI.PushFinding(asyncFinding)
 }
 
 // processMergeDiscoveredAssets performs the following actions:
@@ -407,3 +417,39 @@ func assetToAsyncAsset(a api.Asset) asyncapi.AssetPayload {
 	}
 	return asyncAsset
 }
+
+func findingToAsyncFinding(f *api.Finding) asyncapi.FindingPayload {
+	return 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{
+			CweId:           f.Finding.Issue.CWEID,
+			Description:     f.Finding.Issue.Description,
+			Id:              f.Finding.Issue.ID,
+			Labels:          []interface{}{f.Finding.Issue.Labels},
+			Recommendations: []interface{}{f.Finding.Issue.Recommendations},
+			ReferenceLinks:  []interface{}{f.Finding.Issue.ReferenceLinks},
+			Summary:         f.Finding.Issue.Summary,
+		},
+		Resources: []interface{}{f.Finding.Resources},
+		Score:     float64(f.Finding.Score),
+		Source: &asyncapi.StoreSource{
+			Component: f.Finding.Source.Component,
+			Id:        f.Finding.Source.ID,
+			Instance:  f.Finding.Source.Instance,
+			Name:      f.Finding.Source.Name,
+			Options:   f.Finding.Source.Options,
+			Time:      f.Finding.Source.Time,
+		},
+		Status: f.Finding.Status,
+		Target: &asyncapi.StoreTargetTeams{
+			Id:         f.Finding.Target.ID,
+			Identifier: f.Finding.Target.Identifier,
+			Teams:      []interface{}{f.Finding.Target.Teams},
+		},
+		TotalExposure: int(f.Finding.TotalExposure),
+	}
+}

pkg/asyncapi/models.go

@@ -43,3 +43,64 @@ type Annotation struct {
 	Key   string
 	Value string
 }
+
+// FindingPayload represents a FindingPayload model.
+type FindingPayload struct {
+	AffectedResource     string
+	CurrentExposure      int
+	Details              string
+	Id                   string
+	ImpactDetails        string
+	Issue                *StoreIssueLabels
+	Resources            []interface{}
+	Score                float64
+	Source               *StoreSource
+	Status               string
+	Target               *StoreTargetTeams
+	TotalExposure        int
+	AdditionalProperties map[string][]interface{}
+}
+
+// StoreIssueLabels represents a StoreIssueLabels model.
+type StoreIssueLabels struct {
+	CweId                int
+	Description          string
+	Id                   string
+	Labels               []interface{}
+	Recommendations      []interface{}
+	ReferenceLinks       []interface{}
+	Summary              string
+	AdditionalProperties map[string][]interface{}
+}
+
+// StoreResourceGroup represents a StoreResourceGroup model.
+type StoreResourceGroup struct {
+	Attributes           []interface{}
+	Name                 string
+	Resources            []interface{}
+	AdditionalProperties map[string][]interface{}
+}
+
+// AnonymousSchema33 represents a AnonymousSchema33 model.
+type AnonymousSchema33 struct {
+	AdditionalProperties map[string]string
+}
+
+// StoreSource represents a StoreSource model.
+type StoreSource struct {
+	Component            string
+	Id                   string
+	Instance             string
+	Name                 string
+	Options              string
+	Time                 string
+	AdditionalProperties map[string][]interface{}
+}
+
+// StoreTargetTeams represents a StoreTargetTeams model.
+type StoreTargetTeams struct {
+	Id                   string
+	Identifier           string
+	Teams                []interface{}
+	AdditionalProperties map[string][]interface{}
+}

pkg/asyncapi/vulcan.go

@@ -39,9 +39,15 @@ func (l LevelLogger) Debugf(s string, params ...any) {
 	level.Debug(l.Logger).Log("log", v)
 }
 
-// AssetsEntityName defines the key for the assets entity used by an [EventStreamClient] to
-// determine the topic where the assets are send.
-const AssetsEntityName = "assets"
+const (
+	// AssetsEntityName defines the key for the assets entity used by an [EventStreamClient] to
+	// determine the topic where the assets are sent.
+	AssetsEntityName = "assets"
+
+	// FindingsEntityName defines the key for the findings entity used by an [EventStreamClient] to
+	// determine the topic where the findings are sent.
+	FindingsEntityName = "findings"
+)
 
 // Vulcan implements the asynchorus API of Vulcan.
 type Vulcan struct {
@@ -104,6 +110,25 @@ func (v *Vulcan) DeleteAsset(asset AssetPayload) error {
 	return err
 }
 
+// PushFinding publishes the state of a finding in the current point of time
+// to the underlying [EventStreamClient].
+func (v *Vulcan) PushFinding(finding FindingPayload) error {
+	v.logger.Debugf("pushing finding %+v", finding)
+	payload, err := json.Marshal(finding)
+	if err != nil {
+		return fmt.Errorf("error marshaling to json: %w", err)
+	}
+	metadata := map[string][]byte{
+		"version": []byte(Version),
+	}
+	err = v.client.Push(FindingsEntityName, finding.Id, payload, metadata)
+	if err != nil {
+		return fmt.Errorf("error pushing finding %v: %w", finding, err)
+	}
+	v.logger.Debugf("finding pushed %+v", finding)
+	return nil
+}
+
 // NullVulcan implements an Async Vulcan API interface that does not send the
 // events to any [EventStreamClient]. It's intended to be used when the async
 // API is disabled but other components still need to fullfill a dependency
@@ -123,6 +148,12 @@ func (v *NullVulcan) PushAsset(asset AssetPayload) error {
 	return nil
 }
 
+// PushFinding acepts an event indicating that a finding has been modified or
+// created and just ignores it.
+func (v *NullVulcan) PushFinding(finding FindingPayload) error {
+	return nil
+}
+
 func metadata(asset AssetPayload) map[string][]byte {
 	// The asset type can't be nil.
 	return map[string][]byte{