Store importer authentication configuration in DB.

This avoids restarting the server when we want to add a new
importer.

Also cleanup some dead code and fix some linting issues.

Author: clems4ever

README.md

@@ -13,7 +13,7 @@ and results can be visualized in the UI as shown below.
 Run the following commands
 
     # Spin up GraphKB in few seconds with (wait 15 seconds for mariadb to start).
-    docker-compose up -d
+    source bootstrap.sh && docker-compose up -d
 
     # Insert the example data available in examples/ directory
     # with the following command:

cmd/go-graphkb/config.yml

@@ -1,12 +1,8 @@
-listen: :8080
-# tls_key: cmd/go-graphkb/keys/server.key
-# tls_cert: cmd/go-graphkb/keys/server.crt
+server_listen: :8080
+# server_tls_key: cmd/go-graphkb/keys/server.key
+# server_tls_cert: cmd/go-graphkb/keys/server.crt
 
-mariadb:
-  username: graphkb
-  password: password
-  host: db
-  database: graphkb
-
-sources:
-  csv-employees: token_auth_for_csv_importer
+mariadb_username: graphkb
+mariadb_password: password
+mariadb_host: db
+mariadb_database: graphkb

cmd/go-graphkb/main.go

@@ -80,14 +80,14 @@ func onInit() {
 
 	fmt.Println("Using config file:", viper.ConfigFileUsed())
 
-	dbName := viper.GetString("mariadb.database")
+	dbName := viper.GetString("mariadb_database")
 	if dbName == "" {
 		log.Fatal("Please provide database_name option in your configuration file")
 	}
 	Database = database.NewMariaDB(
-		viper.GetString("mariadb.username"),
-		viper.GetString("mariadb.password"),
-		viper.GetString("mariadb.host"),
+		viper.GetString("mariadb_username"),
+		viper.GetString("mariadb_password"),
+		viper.GetString("mariadb_host"),
 		dbName)
 }
 
@@ -121,9 +121,9 @@ func listen(cmd *cobra.Command, args []string) {
 		log.Fatal(err)
 	}
 
-	listenInterface := viper.GetString("listen")
+	listenInterface := viper.GetString("server_listen")
 
-	server.StartServer(listenInterface, Database, Database, eventBus)
+	server.StartServer(listenInterface, Database, Database, Database, eventBus)
 
 	close(eventBus)
 }

cmd/importer-csv/config.yml

@@ -1,6 +1,6 @@
 graphkb:
   url: "http://localhost:8080"
-  auth_token: "token_auth_for_csv_importer"
+  auth_token: "importer-csv"
   skip_verify: true
 
-path: "./examples/example-data.csv"
+path: "example.csv"

graphkb/importer.go

@@ -3,18 +3,20 @@ package graphkb
 import (
 	"fmt"
 
+	"github.com/clems4ever/go-graphkb/internal/importers"
 	"github.com/clems4ever/go-graphkb/internal/knowledge"
 	"github.com/clems4ever/go-graphkb/internal/schema"
-	"github.com/clems4ever/go-graphkb/internal/sources"
 )
 
+// ImporterOptions options for configuring importer
 type ImporterOptions struct {
 	URL        string
 	AuthToken  string
 	SkipVerify bool
 }
 
-func Start(source sources.Source, options ImporterOptions) error {
+// Start the importer with provided options
+func Start(source importers.Importer, options ImporterOptions) error {
 	if options.URL == "" {
 		return fmt.Errorf("Please provide graphkb URL in configuration file")
 	}
@@ -32,6 +34,7 @@ func Start(source sources.Source, options ImporterOptions) error {
 	return nil
 }
 
+// CreateRelation helper function for creating a relation
 func CreateRelation(fromType schema.AssetType, relation, toType schema.AssetType) RelationType {
 	return schema.RelationType{
 		FromType: fromType,
@@ -40,6 +43,7 @@ func CreateRelation(fromType schema.AssetType, relation, toType schema.AssetType
 	}
 }
 
+// CreateAsset helper function for creating an asset
 func CreateAsset(fromType string) AssetType {
 	return schema.AssetType(fromType)
 }

internal/database/mariadb.go

@@ -87,26 +87,44 @@ CREATE TABLE IF NOT EXISTS relations (
 	q, err = m.db.QueryContext(context.Background(), `
 CREATE TABLE IF NOT EXISTS graph_schema (
 	id INTEGER AUTO_INCREMENT NOT NULL,
-	source_name VARCHAR(64) NOT NULL,
+	importer VARCHAR(64) NOT NULL,
 	graph TEXT NOT NULL,
 	timestamp TIMESTAMP,
-CONSTRAINT pk_schema PRIMARY KEY (id))`)
+	
+	CONSTRAINT pk_schema PRIMARY KEY (id))`)
+	if err != nil {
+		return err
+	}
+	defer q.Close()
+
+	// Create the table storing importers tokens
+	q, err = m.db.QueryContext(context.Background(), `
+CREATE TABLE IF NOT EXISTS importers (
+	id INTEGER AUTO_INCREMENT NOT NULL,
+	name VARCHAR(64) NOT NULL,
+	auth_token VARCHAR(64) NOT NULL,
+
+	CONSTRAINT pk_importer PRIMARY KEY (id),
+	UNIQUE unique_importer_idx (name, auth_token)
+)`)
 	if err != nil {
 		return err
 	}
 	defer q.Close()
 	return nil
 }
 
-// AssetIDResolver store ID assets in a cache
+// AssetRegistry store ID of assets in a cache
 type AssetRegistry struct {
 	cache map[knowledge.AssetKey]int64
 }
 
+// Set id of an asset
 func (ar *AssetRegistry) Set(a knowledge.AssetKey, idx int64) {
 	ar.cache[a] = idx
 }
 
+// Get id of an asset
 func (ar *AssetRegistry) Get(a knowledge.AssetKey) (int64, bool) {
 	idx, ok := ar.cache[a]
 	return idx, ok
@@ -286,19 +304,15 @@ func (m *MariaDB) removeRelations(source string, relations []knowledge.Relation)
 DELETE r FROM relations r
 INNER JOIN assets a ON r.from_id = a.id
 INNER JOIN assets b ON r.to_id = b.id
-WHERE a.type = ? AND a.value = ? AND b.type = ? AND b.value = ? AND r.type = ?`)
+WHERE a.type = ? AND a.value = ? AND b.type = ? AND b.value = ? AND r.type = ? AND r.source = ?`)
 	if err != nil {
 		return 0, 0, err
 	}
 	defer stmt.Close()
 
 	for _, r := range relations {
-		rel := SourceRelation{
-			Relation: r,
-			Source:   source,
-		}
 		res, err := stmt.ExecContext(context.Background(),
-			rel.From.Type, rel.From.Key, rel.To.Type, rel.To.Key, rel.Type)
+			r.From.Type, r.From.Key, r.To.Type, r.To.Key, r.Type, source)
 		if err != nil {
 			return 0, 0, fmt.Errorf("Unable to detete relation %v: %v", r, err)
 		}
@@ -508,6 +522,7 @@ func (m *MariaDB) Query(ctx context.Context, query *query.QueryIL) (*knowledge.G
 	return res, nil
 }
 
+// SaveSchema save the schema graph in database
 func (m *MariaDB) SaveSchema(ctx context.Context, sourceName string, schema schema.SchemaGraph) error {
 	b, err := json.Marshal(schema)
 	if err != nil {
@@ -523,55 +538,60 @@ func (m *MariaDB) SaveSchema(ctx context.Context, sourceName string, schema sche
 	return nil
 }
 
+// LoadSchema load the schema graph of the source from DB
 func (m *MariaDB) LoadSchema(ctx context.Context, sourceName string) (schema.SchemaGraph, error) {
 	row := m.db.QueryRowContext(ctx, "SELECT graph FROM graph_schema WHERE source_name = ? ORDER BY id DESC LIMIT 1", sourceName)
-	var rawJson string
-	if err := row.Scan(&rawJson); err != nil {
+	var rawJSON string
+	if err := row.Scan(&rawJSON); err != nil {
 		if err == sql.ErrNoRows {
 			return schema.NewSchemaGraph(), nil
-		} else {
-			return schema.NewSchemaGraph(), err
 		}
+		return schema.NewSchemaGraph(), err
 	}
 
 	graph := schema.NewSchemaGraph()
-	err := json.Unmarshal([]byte(rawJson), &graph)
+	err := json.Unmarshal([]byte(rawJSON), &graph)
 	if err != nil {
 		return schema.NewSchemaGraph(), err
 	}
 
 	return graph, nil
 }
 
-func (m *MariaDB) ListSources(ctx context.Context) ([]string, error) {
-	rows, err := m.db.QueryContext(ctx, "SELECT DISTINCT source_name FROM graph_schema")
+// ListImporters list importers with their authentication tokens
+func (m *MariaDB) ListImporters(ctx context.Context) (map[string]string, error) {
+	rows, err := m.db.QueryContext(ctx, "SELECT name, auth_token FROM importers")
 
 	if err != nil {
 		return nil, fmt.Errorf("Unable to read sources from database: %v", err)
 	}
 	defer rows.Close()
 
-	sources := make([]string, 0)
+	importers := make(map[string]string)
 	for rows.Next() {
-		var source string
-		if err := rows.Scan(&source); err != nil {
+		var importerName string
+		var authToken string
+		if err := rows.Scan(&importerName, &authToken); err != nil {
 			return nil, err
 		}
-		sources = append(sources, source)
+		importers[importerName] = authToken
 	}
-	return sources, nil
+	return importers, nil
 }
 
+// MariaDBCursor is a cursor of data retrieved by MariaDB
 type MariaDBCursor struct {
 	*sql.Rows
 
 	Projections []knowledge.Projection
 }
 
+// HasMore tells whether there are more data to retrieve from the cursor
 func (mc *MariaDBCursor) HasMore() bool {
 	return mc.Rows.Next()
 }
 
+// Read read one more item from the cursor
 func (mc *MariaDBCursor) Read(ctx context.Context, doc interface{}) error {
 	var err error
 	var fArr []string
@@ -650,6 +670,7 @@ func (mc *MariaDBCursor) Read(ctx context.Context, doc interface{}) error {
 	return nil
 }
 
+// Close the cursor
 func (mc *MariaDBCursor) Close() error {
 	return mc.Rows.Close()
 }

internal/sources/source.go renamed to internal/importers/importer.go

@@ -1,9 +1,9 @@
-package sources
+package importers
 
 import "github.com/clems4ever/go-graphkb/internal/knowledge"
 
-// Source represent a source of data
-type Source interface {
+// Importer represent an importer of data
+type Importer interface {
 	Start(emitter *knowledge.GraphImporter) error
 	Stop() error
 }

internal/importers/registry.go

@@ -0,0 +1,9 @@
+package importers
+
+import "context"
+
+// Registry is a regostry of importers with their auth tokens
+type Registry interface {
+	// List importers with their authentication tokens
+	ListImporters(ctx context.Context) (map[string]string, error)
+}

internal/schema/graph.go

@@ -82,24 +82,26 @@ func (sg *SchemaGraph) Equal(other SchemaGraph) bool {
 	return true
 }
 
+// MarshalJSON marshal the schema graph into json format
 func (sg *SchemaGraph) MarshalJSON() ([]byte, error) {
-	schemaJson := SchemaGraphJSON{}
-	schemaJson.Vertices = []AssetType{}
-	schemaJson.Edges = []RelationType{}
+	schemaJSON := SchemaGraphJSON{}
+	schemaJSON.Vertices = []AssetType{}
+	schemaJSON.Edges = []RelationType{}
 
 	for v := range sg.Vertices.Iter() {
 		vertice := v.(AssetType)
-		schemaJson.Vertices = append(schemaJson.Vertices, vertice)
+		schemaJSON.Vertices = append(schemaJSON.Vertices, vertice)
 	}
 
 	for e := range sg.Edges.Iter() {
 		edge := e.(RelationType)
-		schemaJson.Edges = append(schemaJson.Edges, edge)
+		schemaJSON.Edges = append(schemaJSON.Edges, edge)
 	}
 
-	return json.Marshal(schemaJson)
+	return json.Marshal(schemaJSON)
 }
 
+// UnmarshalJSON unmarshal the schema graph from json payload
 func (sg *SchemaGraph) UnmarshalJSON(b []byte) error {
 	j := SchemaGraphJSON{}
 	if err := json.Unmarshal(b, &j); err != nil {

internal/schema/persistor.go

@@ -2,9 +2,8 @@ package schema
 
 import "context"
 
+// Persistor is a persistor of schema
 type Persistor interface {
-	ListSources(ctx context.Context) ([]string, error)
-
 	SaveSchema(ctx context.Context, sourceName string, sg SchemaGraph) error
 	LoadSchema(ctx context.Context, sourceName string) (SchemaGraph, error)
 }