fix: don't error out immediately on invalid meta tag

tests: add tests for validatemeta tag
tests: fix a bad package declaration

cmd/api/src/daemons/datapipe/fileops.go

@@ -1,3 +1,19 @@
+// Copyright 2024 Specter Ops, Inc.
+//
+// Licensed under the Apache License, Version 2.0
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// SPDX-License-Identifier: Apache-2.0
+
 package datapipe
 
 import (
@@ -20,7 +36,7 @@ func SeekToDataTag(decoder *json.Decoder) error {
 	for {
 		if token, err := decoder.Token(); err != nil {
 			if errors.Is(err, io.EOF) {
-				return ErrDataNotFound
+				return ErrDataTagNotFound
 			}
 
 			return err

cmd/api/src/daemons/datapipe/ingest.go

@@ -36,10 +36,10 @@ const (
 )
 
 var (
-	ErrMetaNotFound   = errors.New("no valid meta tag found")
-	ErrDataNotFound   = errors.New("no data tag found")
-	ErrNoTagFound     = errors.New("no valid meta tag or data tag found")
-	ErrInvalidDataTag = errors.New("invalid data tag found")
+	ErrMetaTagNotFound = errors.New("no valid meta tag found")
+	ErrDataTagNotFound = errors.New("no data tag found")
+	ErrNoTagFound      = errors.New("no valid meta tag or data tag found")
+	ErrInvalidDataTag  = errors.New("invalid data tag found")
 )
 
 func ReadFileForIngest(batch graph.Batch, reader io.ReadSeeker) error {

cmd/api/src/daemons/datapipe/util_test.go

@@ -1,22 +1,23 @@
 // Copyright 2023 Specter Ops, Inc.
-// 
+//
 // Licensed under the Apache License, Version 2.0
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
-// 
+//
 //     http://www.apache.org/licenses/LICENSE-2.0
-// 
+//
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.
-// 
+//
 // SPDX-License-Identifier: Apache-2.0
 
-package datapipe
+package datapipe_test
 
 import (
+	"github.com/specterops/bloodhound/src/daemons/datapipe"
 	"github.com/stretchr/testify/require"
 	"testing"
 	"time"
@@ -29,7 +30,7 @@ func TestRandomDurationBetween(t *testing.T) {
 	)
 
 	for i := 0; i < 1000; i++ {
-		randomDuration := RandomDurationBetween(min, max)
+		randomDuration := datapipe.RandomDurationBetween(min, max)
 
 		require.True(t, min < randomDuration)
 		require.True(t, max > randomDuration)

cmd/api/src/daemons/datapipe/validation.go

@@ -1,9 +1,26 @@
+// Copyright 2024 Specter Ops, Inc.
+//
+// Licensed under the Apache License, Version 2.0
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// SPDX-License-Identifier: Apache-2.0
+
 package datapipe
 
 import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"github.com/specterops/bloodhound/log"
 	"io"
 )
 
@@ -25,10 +42,10 @@ func ValidateMetaTag(reader io.ReadSeeker) (Metadata, error) {
 			if errors.Is(err, io.EOF) {
 				if !metaTagFound && !dataTagFound {
 					return Metadata{}, ErrNoTagFound
-				} else if !metaTagFound {
-					return Metadata{}, ErrDataNotFound
+				} else if !dataTagFound {
+					return Metadata{}, ErrDataTagNotFound
 				} else {
-					return Metadata{}, ErrMetaNotFound
+					return Metadata{}, ErrMetaTagNotFound
 				}
 			}
 			return Metadata{}, err
@@ -44,7 +61,7 @@ func ValidateMetaTag(reader io.ReadSeeker) (Metadata, error) {
 			case string:
 				if !metaTagFound && depth == 1 && typed == "meta" {
 					if err := decoder.Decode(&meta); err != nil {
-						return Metadata{}, err
+						log.Warnf("Found invalid metatag, skipping")
 					} else if meta.Type.IsValid() {
 						metaTagFound = true
 					}

cmd/api/src/daemons/datapipe/validation_test.go

@@ -0,0 +1,94 @@
+// Copyright 2024 Specter Ops, Inc.
+//
+// Licensed under the Apache License, Version 2.0
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// SPDX-License-Identifier: Apache-2.0
+
+package datapipe_test
+
+import (
+	"github.com/specterops/bloodhound/src/daemons/datapipe"
+	"github.com/stretchr/testify/assert"
+	"strings"
+	"testing"
+)
+
+func Test_ValidateMetaTag(t *testing.T) {
+	validJson := `{
+    "meta": {
+        "methods": 0,
+        "type": "sessions",
+        "count": 0,
+        "version": 5
+    },
+    "data": []
+	}`
+
+	meta, err := datapipe.ValidateMetaTag(strings.NewReader(validJson))
+	assert.Nil(t, err)
+	assert.Equal(t, datapipe.DataTypeSession, meta.Type)
+
+	missingDataTag := `{
+    "meta": {
+        "methods": 0,
+        "type": "sessions",
+        "count": 0,
+        "version": 5
+    }
+	}`
+
+	meta, err = datapipe.ValidateMetaTag(strings.NewReader(missingDataTag))
+	assert.Equal(t, err, datapipe.ErrDataTagNotFound)
+
+	missingMetaTag := `{
+    "data": []
+	}`
+
+	meta, err = datapipe.ValidateMetaTag(strings.NewReader(missingMetaTag))
+	assert.Equal(t, err, datapipe.ErrMetaTagNotFound)
+
+	missingBothTag := `{
+	}`
+
+	meta, err = datapipe.ValidateMetaTag(strings.NewReader(missingBothTag))
+	assert.Equal(t, err, datapipe.ErrNoTagFound)
+
+	ignoreInvalidTag := `{
+	"meta": 0,
+    "meta": {
+        "methods": 0,
+        "type": "sessions",
+        "count": 0,
+        "version": 5
+    },
+    "data": []
+	}`
+
+	meta, err = datapipe.ValidateMetaTag(strings.NewReader(ignoreInvalidTag))
+	assert.Nil(t, err)
+	assert.Equal(t, datapipe.DataTypeSession, meta.Type)
+
+	swapOrder := `{
+	"data": [],
+    "meta": {
+        "methods": 0,
+        "type": "sessions",
+        "count": 0,
+        "version": 5
+    }
+	}`
+
+	meta, err = datapipe.ValidateMetaTag(strings.NewReader(swapOrder))
+	assert.Nil(t, err)
+	assert.Equal(t, datapipe.DataTypeSession, meta.Type)
+}