Saving edges and node from CertStream traffic.

This work extracts fingerprints and domains from CertStream data
structure received through the stream. It builds nodes and edges and
saves them to the relevant files. It sends this data to Kafka but no
logic is implemented at the exit of the broker yet.

broker/main.go

@@ -54,7 +54,6 @@ func ReadEventFromKafka() {
 		json.Unmarshal(m.Value, &node)
 		if len(node.ID) != 0 {
 			fmt.Println(node)
-			models.SaveNode(&node)
 		}
 	}
 }

main.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/CaliDog/certstream-go"
 	"github.com/sirupsen/logrus"
+	"gitlab.dcso.lolcat/LABS/styx/broker"
 	"gitlab.dcso.lolcat/LABS/styx/models"
 )
 
@@ -12,30 +13,32 @@ func main() {
 	// The false flag specifies that we want heartbeat messages.
 	stream, errStream := certstream.CertStreamEventStream(false)
 	fmt.Println("Starting to get data from CertStream...")
-	// Conn, err := broker.SetUpKafkaConnecter()
-	// if err != nil {
-	// panic(err)
-	// }
-	// go broker.ReadEventFromKafka()
+	Conn, err := broker.SetUpKafkaConnecter()
+	if err != nil {
+		panic(err)
+	}
+	go broker.ReadEventFromKafka()
 	for {
 		select {
 		case jq := <-stream:
 			if data, err := models.ExtractCertFromStream(jq); err == nil {
 
-				fingerprintNode := models.BuildNode("certstream", data.Data.LeafCert.Fingerprint)
+				// rawNode := models.WrapCertStreamData(*data)
+				// models.SaveNode(rawNode)
+				fingerprintNode := models.BuildNode("certstream", "fingerprint", data.Data.LeafCert.Fingerprint)
 				models.SaveNode(fingerprintNode)
 				fmt.Println(fingerprintNode)
 				allDomains := data.Data.LeafCert.AllDomains
 				var edge *models.Edge
 				for _, domain := range allDomains {
-					domainNode := models.BuildNode("certstream", domain)
+					domainNode := models.BuildNode("certstream", "domain", domain)
 					models.SaveNode(domainNode)
 					edge = models.BuildEdge("certstream", fingerprintNode.ID, domainNode.ID)
 					fmt.Println(edge)
 					models.SaveEdge(edge)
 
 				}
-				// broker.SendEventToKafka(Conn, *fingerprintNode)
+				broker.SendEventToKafka(Conn, *fingerprintNode)
 			}
 
 		case err := <-errStream:

models/main.go

@@ -37,6 +37,7 @@ func BuildEdge(source string, nodeOneUUID string, nodeTwoUUID string) *Edge {
 	}
 }
 
+// SaveNode saves a node to a file.
 func SaveNode(node *Node) {
 	err := utils.FileExists("nodes.json")
 	if err != nil {
@@ -65,6 +66,7 @@ func SaveNode(node *Node) {
 	}
 }
 
+// SaveEdge saves an edge to a file.
 func SaveEdge(edge *Edge) {
 	err := utils.FileExists("edges.json")
 	if err != nil {

models/types.go

@@ -19,3 +19,13 @@ type Edge struct {
 	Timestamp string `json:"timestamp"`
 	Source    string `json:"source"`
 }
+
+// CertStreamWrapper is a wrapper around the stream function to unmarshall the
+// data receive in a Go structure.
+type CertStreamWrapper struct {
+	ID       string           `json:"id"`
+	Type     string           `json:"type"`
+	Data     CertStreamStruct `json:"data"`
+	Created  string           `json:"created"`
+	Modified string           `json:"modified"`
+}