Fix software ingestion when fields are larger than supported (#13741)

Should fix the issue reported in #12230. For Wireshark, osquery was
reporting a `vendor` value larger than what we allowed storing in the
`vendor` column (32 bytes). But recently we enlarged the `vendor` column
to fit `114` chars. The direct software ingestion routine was inserting
a new software entry every time because the incoming software vendor was
different to what Fleet had stored in the previous ingestion (`vendor`
column trimmed from `The Wireshark developer community,
https://www.wireshark.org/` to `The Wireshark developer communit`).

I've now made sure that all fields are trimmed as soon as they are
received by osquery thus not triggering any re-inserts when any field is
larger than what Fleet supports.

- [X] Changes file added for user-visible changes in `changes/` or
`orbit/changes/`.
See [Changes
files](https://fleetdm.com/docs/contributing/committing-changes#changes-files)
for more information.
- ~[ ] Documented any API changes (docs/Using-Fleet/REST-API.md or
docs/Contributing/API-for-contributors.md)~
- ~[ ] Documented any permissions changes (docs/Using
Fleet/manage-access.md)~
- ~[ ] Input data is properly validated, `SELECT *` is avoided, SQL
injection is prevented (using placeholders for values in statements)~
- ~[ ] Added support on fleet's osquery simulator `cmd/osquery-perf` for
new osquery data ingestion features.~
- [X] Added/updated tests
- [X] Manual QA for all new/changed functionality
  - ~For Orbit and Fleet Desktop changes:~
- ~[ ] Manual QA must be performed in the three main OSs, macOS, Windows
and Linux.~
- ~[ ] Auto-update manual QA, from released version of component to new
version (see [tools/tuf/test](../tools/tuf/test/README.md)).~

changes/fix-software-reinserts-when-fields-are-too-long

@@ -0,0 +1,2 @@
+* Fixed software ingestion to not re-insert software when incoming fields from hosts are longer than what Fleet supports. This bug caused some CVEs to be reported every time the vulnerability cron ran.
+IMPORTANT: After deploying this fix, the vulnerability cron will report the CVEs one last time, and subsequent cron runs will not report the CVE (as expected).

server/datastore/mysql/software.go

@@ -16,48 +16,6 @@ import (
 	"github.com/jmoiron/sqlx"
 )
 
-const (
-	maxSoftwareNameLen             = 255
-	maxSoftwareVersionLen          = 255
-	maxSoftwareSourceLen           = 64
-	maxSoftwareBundleIdentifierLen = 255
-
-	maxSoftwareReleaseLen = 64
-	maxSoftwareVendorLen  = 32
-	maxSoftwareArchLen    = 16
-)
-
-func truncateString(str string, length int) string {
-	if len(str) > length {
-		return str[:length]
-	}
-	return str
-}
-
-func uniqueStringToSoftware(s string) fleet.Software {
-	parts := strings.Split(s, fleet.SoftwareFieldSeparator)
-
-	// Release, Vendor and Arch fields were added on a migration,
-	// If one of them is defined, then they are included in the string.
-	var release, vendor, arch string
-	if len(parts) > 4 {
-		release = truncateString(parts[4], maxSoftwareReleaseLen)
-		vendor = truncateString(parts[5], maxSoftwareVendorLen)
-		arch = truncateString(parts[6], maxSoftwareArchLen)
-	}
-
-	return fleet.Software{
-		Name:             truncateString(parts[0], maxSoftwareNameLen),
-		Version:          truncateString(parts[1], maxSoftwareVersionLen),
-		Source:           truncateString(parts[2], maxSoftwareSourceLen),
-		BundleIdentifier: truncateString(parts[3], maxSoftwareBundleIdentifierLen),
-
-		Release: release,
-		Vendor:  vendor,
-		Arch:    arch,
-	}
-}
-
 func softwareSliceToMap(softwares []fleet.Software) map[string]fleet.Software {
 	result := make(map[string]fleet.Software)
 	for _, s := range softwares {
@@ -493,24 +451,31 @@ func insertNewInstalledHostSoftwareDB(
 	var insertsHostSoftware []interface{}
 	var insertedSoftware []fleet.Software
 
-	incomingOrdered := make([]string, 0, len(incomingMap))
-	for s := range incomingMap {
-		incomingOrdered = append(incomingOrdered, s)
+	type softwareWithUniqueName struct {
+		uniqueName string
+		software   fleet.Software
 	}
-	sort.Strings(incomingOrdered)
+	incomingOrdered := make([]softwareWithUniqueName, 0, len(incomingMap))
+	for uniqueName, software := range incomingMap {
+		incomingOrdered = append(incomingOrdered, softwareWithUniqueName{
+			uniqueName: uniqueName,
+			software:   software,
+		})
+	}
+	sort.Slice(incomingOrdered, func(i, j int) bool {
+		return incomingOrdered[i].uniqueName < incomingOrdered[j].uniqueName
+	})
 
 	for _, s := range incomingOrdered {
-		if _, ok := currentMap[s]; !ok {
-			swToInsert := uniqueStringToSoftware(s)
-			id, err := getOrGenerateSoftwareIdDB(ctx, tx, swToInsert)
+		if _, ok := currentMap[s.uniqueName]; !ok {
+			id, err := getOrGenerateSoftwareIdDB(ctx, tx, s.software)
 			if err != nil {
 				return nil, err
 			}
-			sw := incomingMap[s]
-			insertsHostSoftware = append(insertsHostSoftware, hostID, id, sw.LastOpenedAt)
+			insertsHostSoftware = append(insertsHostSoftware, hostID, id, s.software.LastOpenedAt)
 
-			swToInsert.ID = id
-			insertedSoftware = append(insertedSoftware, swToInsert)
+			s.software.ID = id
+			insertedSoftware = append(insertedSoftware, s.software)
 		}
 	}
 

server/datastore/mysql/software_test.go

@@ -208,16 +208,13 @@ func testSoftwareCPE(t *testing.T, ds *Datastore) {
 func testSoftwareHostDuplicates(t *testing.T, ds *Datastore) {
 	host1 := test.NewHost(t, ds, "host1", "", "host1key", "host1uuid", time.Now())
 
-	longName := strings.Repeat("a", 260)
+	longName := strings.Repeat("a", fleet.SoftwareNameMaxLength+5)
 
 	incoming := make(map[string]fleet.Software)
-	sw := fleet.Software{
-		Name:    longName + "b",
-		Version: "0.0.1",
-		Source:  "chrome_extension",
-	}
+	sw, err := fleet.SoftwareFromOsqueryRow(longName+"b", "0.0.1", "chrome_extension", "", "", "", "", "", "")
+	require.NoError(t, err)
 	soft2Key := sw.ToUniqueStr()
-	incoming[soft2Key] = sw
+	incoming[soft2Key] = *sw
 
 	tx, err := ds.writer(context.Background()).Beginx()
 	require.NoError(t, err)
@@ -225,20 +222,39 @@ func testSoftwareHostDuplicates(t *testing.T, ds *Datastore) {
 	require.NoError(t, err)
 	require.NoError(t, tx.Commit())
 
+	// Check that the software entry was stored for the host.
+	var software []fleet.Software
+	err = sqlx.SelectContext(context.Background(), ds.reader(context.Background()),
+		&software, `SELECT s.id, s.name FROM software s JOIN host_software hs WHERE hs.host_id = ?`,
+		host1.ID,
+	)
+	require.NoError(t, err)
+	require.Len(t, software, 1)
+	require.NotZero(t, software[0].ID)
+	require.Equal(t, strings.Repeat("a", fleet.SoftwareNameMaxLength), software[0].Name)
+
 	incoming = make(map[string]fleet.Software)
-	sw = fleet.Software{
-		Name:    longName + "c",
-		Version: "0.0.1",
-		Source:  "chrome_extension",
-	}
+	sw, err = fleet.SoftwareFromOsqueryRow(longName+"c", "0.0.1", "chrome_extension", "", "", "", "", "", "")
+	require.NoError(t, err)
 	soft3Key := sw.ToUniqueStr()
-	incoming[soft3Key] = sw
+	incoming[soft3Key] = *sw
 
 	tx, err = ds.writer(context.Background()).Beginx()
 	require.NoError(t, err)
 	_, err = insertNewInstalledHostSoftwareDB(context.Background(), tx, host1.ID, make(map[string]fleet.Software), incoming)
 	require.NoError(t, err)
 	require.NoError(t, tx.Commit())
+
+	// Check that the software entry was not modified with the new insert because of the name trimming.
+	var software2 []fleet.Software
+	err = sqlx.SelectContext(context.Background(), ds.reader(context.Background()),
+		&software2, `SELECT s.id, s.name FROM software s JOIN host_software hs WHERE hs.host_id = ?`,
+		host1.ID,
+	)
+	require.NoError(t, err)
+	require.Len(t, software2, 1)
+	require.Equal(t, strings.Repeat("a", fleet.SoftwareNameMaxLength), software2[0].Name)
+	require.Equal(t, software[0].ID, software2[0].ID)
 }
 
 func testSoftwareLoadVulnerabilities(t *testing.T, ds *Datastore) {

server/fleet/software.go

@@ -1,15 +1,30 @@
 package fleet
 
 import (
+	"errors"
+	"fmt"
+	"strconv"
 	"strings"
 	"time"
+	"unicode/utf8"
 )
 
-// Must be kept in sync with the vendor column definition.
 const (
-	SoftwareVendorMaxLength    = 114
 	SoftwareVendorMaxLengthFmt = "%.111s..."
 	SoftwareFieldSeparator     = "\u0000"
+
+	//
+	// The following length values must be kept in sync with the DB column definitions.
+	//
+
+	SoftwareNameMaxLength             = 255
+	SoftwareVersionMaxLength          = 255
+	SoftwareSourceMaxLength           = 64
+	SoftwareBundleIdentifierMaxLength = 255
+
+	SoftwareReleaseMaxLength = 64
+	SoftwareVendorMaxLength  = 114
+	SoftwareArchMaxLength    = 16
 )
 
 type Vulnerabilities []CVE
@@ -170,3 +185,68 @@ func (uhsdbr *UpdateHostSoftwareDBResult) CurrInstalled() []Software {
 
 	return r
 }
+
+// ParseSoftwareLastOpenedAtRowValue attempts to parse the last_opened_at
+// software column value. If the value is empty or if the parsed value is
+// less or equal than 0 it returns (time.Time{}, nil). We do this because
+// some macOS apps return "-1.0" when the app was never opened and we hardcode
+// to 0 for some tables that don't have such info.
+func ParseSoftwareLastOpenedAtRowValue(value string) (time.Time, error) {
+	if value == "" {
+		return time.Time{}, nil
+	}
+	lastOpenedEpoch, err := strconv.ParseFloat(value, 64)
+	if err != nil {
+		return time.Time{}, err
+	}
+	if lastOpenedEpoch <= 0 {
+		return time.Time{}, nil
+	}
+	return time.Unix(int64(lastOpenedEpoch), 0).UTC(), nil
+}
+
+// SoftwareFromOsqueryRow creates a fleet.Software from the values reported by osquery.
+// Arguments name and source must be defined, all other fields are optional.
+// This method doesn't fail if lastOpenedAt is empty or cannot be parsed.
+//
+// All fields are trimmed to fit on Fleet's database.
+// The vendor field is currently trimmed by removing the extra characters and adding `...` at the end.
+func SoftwareFromOsqueryRow(name, version, source, vendor, installedPath, release, arch, bundleIdentifier, lastOpenedAt string) (*Software, error) {
+	if name == "" {
+		return nil, errors.New("host reported software with empty name")
+	}
+	if source == "" {
+		return nil, errors.New("host reported software with empty source")
+	}
+
+	// We don't fail if only the last_opened_at cannot be parsed.
+	lastOpenedAtTime, _ := ParseSoftwareLastOpenedAtRowValue(lastOpenedAt)
+
+	// Check whether the vendor is longer than the max allowed width and if so, truncate it.
+	if utf8.RuneCountInString(vendor) >= SoftwareVendorMaxLength {
+		vendor = fmt.Sprintf(SoftwareVendorMaxLengthFmt, vendor)
+	}
+
+	truncateString := func(str string, length int) string {
+		runes := []rune(str)
+		if len(runes) > length {
+			return string(runes[:length])
+		}
+		return str
+	}
+
+	software := Software{
+		Name:             truncateString(name, SoftwareNameMaxLength),
+		Version:          truncateString(version, SoftwareVersionMaxLength),
+		Source:           truncateString(source, SoftwareSourceMaxLength),
+		BundleIdentifier: truncateString(bundleIdentifier, SoftwareBundleIdentifierMaxLength),
+
+		Release: truncateString(release, SoftwareReleaseMaxLength),
+		Vendor:  vendor,
+		Arch:    truncateString(arch, SoftwareArchMaxLength),
+	}
+	if !lastOpenedAtTime.IsZero() {
+		software.LastOpenedAt = &lastOpenedAtTime
+	}
+	return &software, nil
+}

server/fleet/software_test.go

@@ -54,3 +54,24 @@ func TestSoftwareIterQueryOptionsIsValid(t *testing.T) {
 		}
 	}
 }
+
+func TestParseSoftwareLastOpenedAtRowValue(t *testing.T) {
+	// Some macOS apps return last_opened_at=-1.0 on apps
+	// that were never opened.
+	lastOpenedAt, err := ParseSoftwareLastOpenedAtRowValue("-1.0")
+	require.NoError(t, err)
+	require.Zero(t, lastOpenedAt)
+
+	// Our software queries hardcode to 0 if such info is not available.
+	lastOpenedAt, err = ParseSoftwareLastOpenedAtRowValue("0")
+	require.NoError(t, err)
+	require.Zero(t, lastOpenedAt)
+
+	lastOpenedAt, err = ParseSoftwareLastOpenedAtRowValue("foobar")
+	require.Error(t, err)
+	require.Zero(t, lastOpenedAt)
+
+	lastOpenedAt, err = ParseSoftwareLastOpenedAtRowValue("1694026958")
+	require.NoError(t, err)
+	require.NotZero(t, lastOpenedAt)
+}

server/service/integration_core_test.go

@@ -4,9 +4,11 @@ import (
 	"bytes"
 	"context"
 	"crypto/rand"
+	"database/sql"
 	"encoding/base64"
 	"encoding/csv"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -7298,3 +7300,271 @@ func (s *integrationTestSuite) TestDirectIngestScheduledQueryStats() {
 	require.Equal(t, strconv.FormatInt(int64(sqs.UserTime), 10), row["user_time"])
 	require.Equal(t, strconv.FormatInt(int64(sqs.WallTime), 10), row["wall_time"])
 }
+
+// TestDirectIngestSoftwareWithLongFields tests that software with reported long fields
+// are inserted properly and subsequent reports of the same software do not generate new
+// entries in the `software` table. (It mainly tests the comparison between the currenly
+// inserted software and the incoming software from a host.)
+func (s *integrationTestSuite) TestDirectIngestSoftwareWithLongFields() {
+	t := s.T()
+
+	appConfig, err := s.ds.AppConfig(context.Background())
+	require.NoError(t, err)
+	appConfig.Features.EnableSoftwareInventory = true
+
+	globalHost, err := s.ds.NewHost(context.Background(), &fleet.Host{
+		DetailUpdatedAt: time.Now(),
+		LabelUpdatedAt:  time.Now(),
+		PolicyUpdatedAt: time.Now(),
+		SeenTime:        time.Now().Add(-1 * time.Minute),
+		OsqueryHostID:   ptr.String(uuid.New().String()),
+		NodeKey:         ptr.String(uuid.New().String()),
+		UUID:            uuid.New().String(),
+		Hostname:        fmt.Sprintf("%sfoo.global", t.Name()),
+		Platform:        "darwin",
+	})
+	require.NoError(t, err)
+
+	// Simulate a osquery agent on Windows reporting a software row for Wireshark.
+	rows := []map[string]string{
+		{
+			"name":           "Wireshark 4.0.8 64-bit",
+			"version":        "4.0.8",
+			"type":           "Program (Windows)",
+			"source":         "programs",
+			"vendor":         "The Wireshark developer community, https://www.wireshark.org",
+			"installed_path": "C:\\Program Files\\Wireshark",
+		},
+	}
+	detailQueries := osquery_utils.GetDetailQueries(context.Background(), config.FleetConfig{}, appConfig, &appConfig.Features)
+	err = detailQueries["software_windows"].DirectIngestFunc(
+		context.Background(),
+		log.NewNopLogger(),
+		globalHost,
+		s.ds,
+		rows,
+	)
+	require.NoError(t, err)
+
+	// Check that the software was properly ingested.
+	softwareQueryByName := "SELECT id, name, version, source, bundle_identifier, `release`, arch, vendor FROM software WHERE name = ?;"
+	var wiresharkSoftware fleet.Software
+	mysql.ExecAdhocSQL(t, s.ds, func(q sqlx.ExtContext) error {
+		return sqlx.GetContext(context.Background(), q, &wiresharkSoftware, softwareQueryByName, "Wireshark 4.0.8 64-bit")
+	})
+	require.NotZero(t, wiresharkSoftware.ID)
+	require.Equal(t, "Wireshark 4.0.8 64-bit", wiresharkSoftware.Name)
+	require.Equal(t, "4.0.8", wiresharkSoftware.Version)
+	require.Equal(t, "programs", wiresharkSoftware.Source)
+	require.Empty(t, wiresharkSoftware.BundleIdentifier)
+	require.Empty(t, wiresharkSoftware.Release)
+	require.Empty(t, wiresharkSoftware.Arch)
+	require.Equal(t, "The Wireshark developer community, https://www.wireshark.org", wiresharkSoftware.Vendor)
+	hostSoftwareInstalledPathsQuery := `SELECT installed_path FROM host_software_installed_paths WHERE software_id = ?;`
+	var wiresharkSoftwareInstalledPath string
+	mysql.ExecAdhocSQL(t, s.ds, func(q sqlx.ExtContext) error {
+		return sqlx.GetContext(context.Background(), q, &wiresharkSoftwareInstalledPath, hostSoftwareInstalledPathsQuery, wiresharkSoftware.ID)
+	})
+	require.Equal(t, "C:\\Program Files\\Wireshark", wiresharkSoftwareInstalledPath)
+
+	// We now check that the same software is not created again as a new row when it is received again during software ingestion.
+	err = detailQueries["software_windows"].DirectIngestFunc(
+		context.Background(),
+		log.NewNopLogger(),
+		globalHost,
+		s.ds,
+		rows,
+	)
+	require.NoError(t, err)
+	var wiresharkSoftware2 fleet.Software
+	mysql.ExecAdhocSQL(t, s.ds, func(q sqlx.ExtContext) error {
+		return sqlx.GetContext(context.Background(), q, &wiresharkSoftware2, softwareQueryByName, "Wireshark 4.0.8 64-bit")
+	})
+	require.NotZero(t, wiresharkSoftware2.ID)
+	require.Equal(t, wiresharkSoftware.ID, wiresharkSoftware2.ID)
+
+	// Simulate a osquery agent on Windows reporting a software row with a longer than 114 chars vendor field.
+	rows = []map[string]string{
+		{
+			"name":           "Foobar" + strings.Repeat("A", fleet.SoftwareNameMaxLength),
+			"version":        "4.0.8" + strings.Repeat("B", fleet.SoftwareVersionMaxLength),
+			"type":           "Program (Windows)",
+			"source":         "programs" + strings.Repeat("C", fleet.SoftwareSourceMaxLength),
+			"vendor":         strings.Repeat("D", fleet.SoftwareVendorMaxLength+1),
+			"installed_path": "C:\\Program Files\\Foobar",
+			// Test UTF-8 encoded strings.
+			"bundle_identifier": strings.Repeat("⌘", fleet.SoftwareBundleIdentifierMaxLength+1),
+			"release":           strings.Repeat("F", fleet.SoftwareReleaseMaxLength-1) + "⌘⌘",
+			"arch":              strings.Repeat("G", fleet.SoftwareArchMaxLength+1),
+		},
+	}
+
+	err = detailQueries["software_windows"].DirectIngestFunc(
+		context.Background(),
+		log.NewNopLogger(),
+		globalHost,
+		s.ds,
+		rows,
+	)
+	require.NoError(t, err)
+
+	var foobarSoftware fleet.Software
+	mysql.ExecAdhocSQL(t, s.ds, func(q sqlx.ExtContext) error {
+		return sqlx.GetContext(context.Background(), q, &foobarSoftware, softwareQueryByName, "Foobar"+strings.Repeat("A", fleet.SoftwareNameMaxLength-6))
+	})
+	require.NotZero(t, foobarSoftware.ID)
+	require.Equal(t, "Foobar"+strings.Repeat("A", fleet.SoftwareNameMaxLength-6), foobarSoftware.Name)
+	require.Equal(t, "4.0.8"+strings.Repeat("B", fleet.SoftwareNameMaxLength-5), foobarSoftware.Version)
+	require.Equal(t, "programs"+strings.Repeat("C", fleet.SoftwareSourceMaxLength-8), foobarSoftware.Source)
+	// Vendor field is currenty trimmed with a different method (... appended at the end)
+	require.Equal(t, strings.Repeat("D", fleet.SoftwareVendorMaxLength-3)+"...", foobarSoftware.Vendor)
+	require.Equal(t, strings.Repeat("⌘", fleet.SoftwareBundleIdentifierMaxLength), foobarSoftware.BundleIdentifier)
+	require.Equal(t, strings.Repeat("F", fleet.SoftwareReleaseMaxLength-1)+"⌘", foobarSoftware.Release)
+	require.Equal(t, strings.Repeat("G", fleet.SoftwareArchMaxLength), foobarSoftware.Arch)
+
+	// We now check that the same software with long (to be trimmed) fields is not created again as a new row.
+	err = detailQueries["software_windows"].DirectIngestFunc(
+		context.Background(),
+		log.NewNopLogger(),
+		globalHost,
+		s.ds,
+		rows,
+	)
+	require.NoError(t, err)
+
+	var foobarSoftware2 fleet.Software
+	mysql.ExecAdhocSQL(t, s.ds, func(q sqlx.ExtContext) error {
+		return sqlx.GetContext(context.Background(), q, &foobarSoftware2, softwareQueryByName, "Foobar"+strings.Repeat("A", fleet.SoftwareNameMaxLength-6))
+	})
+	require.Equal(t, foobarSoftware.ID, foobarSoftware2.ID)
+}
+
+func (s *integrationTestSuite) TestDirectIngestSoftwareWithInvalidFields() {
+	t := s.T()
+
+	appConfig, err := s.ds.AppConfig(context.Background())
+	require.NoError(t, err)
+	appConfig.Features.EnableSoftwareInventory = true
+
+	globalHost, err := s.ds.NewHost(context.Background(), &fleet.Host{
+		DetailUpdatedAt: time.Now(),
+		LabelUpdatedAt:  time.Now(),
+		PolicyUpdatedAt: time.Now(),
+		SeenTime:        time.Now().Add(-1 * time.Minute),
+		OsqueryHostID:   ptr.String(uuid.New().String()),
+		NodeKey:         ptr.String(uuid.New().String()),
+		UUID:            uuid.New().String(),
+		Hostname:        fmt.Sprintf("%sfoo.global", t.Name()),
+		Platform:        "darwin",
+	})
+	require.NoError(t, err)
+
+	// Ingesting software without name should not fail, but the software won't be inserted.
+	rows := []map[string]string{
+		{
+			"version":        "4.0.8",
+			"type":           "Program (Windows)",
+			"source":         "programs",
+			"vendor":         "The Wireshark developer community, https://www.wireshark.org",
+			"installed_path": "C:\\Program Files\\Wireshark",
+			"last_opened_at": "foobar",
+		},
+	}
+	var w1 bytes.Buffer
+	logger1 := log.NewJSONLogger(&w1)
+	detailQueries := osquery_utils.GetDetailQueries(context.Background(), config.FleetConfig{}, appConfig, &appConfig.Features)
+	err = detailQueries["software_windows"].DirectIngestFunc(
+		context.Background(),
+		logger1,
+		globalHost,
+		s.ds,
+		rows,
+	)
+	require.NoError(t, err)
+	logs1, err := io.ReadAll(&w1)
+	require.NoError(t, err)
+	require.Contains(t, string(logs1), "host reported software with empty name", fmt.Sprintf("%s", logs1))
+	require.Contains(t, string(logs1), "debug")
+
+	// Check that the software was not ingested.
+	softwareQueryByVendor := "SELECT id, name, version, source, bundle_identifier, `release`, arch, vendor FROM software WHERE vendor = ?;"
+	mysql.ExecAdhocSQL(t, s.ds, func(q sqlx.ExtContext) error {
+		var wiresharkSoftware fleet.Software
+		if sqlx.GetContext(context.Background(), q, &wiresharkSoftware, softwareQueryByVendor, "The Wireshark developer community, https://www.wireshark.org") != sql.ErrNoRows {
+			return errors.New("expected no results")
+		}
+		return nil
+	})
+
+	// Ingesting software without source should not fail, but the software won't be inserted.
+	rows = []map[string]string{
+		{
+			"name":           "Wireshark 4.0.8 64-bit",
+			"version":        "4.0.8",
+			"type":           "Program (Windows)",
+			"vendor":         "The Wireshark developer community, https://www.wireshark.org",
+			"installed_path": "C:\\Program Files\\Wireshark",
+			"last_opened_at": "foobar",
+		},
+	}
+	detailQueries = osquery_utils.GetDetailQueries(context.Background(), config.FleetConfig{}, appConfig, &appConfig.Features)
+	var w2 bytes.Buffer
+	logger2 := log.NewJSONLogger(&w2)
+	err = detailQueries["software_windows"].DirectIngestFunc(
+		context.Background(),
+		logger2,
+		globalHost,
+		s.ds,
+		rows,
+	)
+	require.NoError(t, err)
+	logs2, err := io.ReadAll(&w2)
+	require.NoError(t, err)
+	require.Contains(t, string(logs2), "host reported software with empty source")
+	require.Contains(t, string(logs2), "debug")
+
+	// Check that the software was not ingested.
+	softwareQueryByName := "SELECT id, name, version, source, bundle_identifier, `release`, arch, vendor FROM software WHERE name = ?;"
+	mysql.ExecAdhocSQL(t, s.ds, func(q sqlx.ExtContext) error {
+		var wiresharkSoftware fleet.Software
+		if sqlx.GetContext(context.Background(), q, &wiresharkSoftware, softwareQueryByName, "Wireshark 4.0.8 64-bit") != sql.ErrNoRows {
+			return errors.New("expected no results")
+		}
+		return nil
+	})
+
+	// Ingesting software with invalid last_opened_at should not fail (only log a debug error)
+	rows = []map[string]string{
+		{
+			"name":           "Wireshark 4.0.8 64-bit",
+			"version":        "4.0.8",
+			"type":           "Program (Windows)",
+			"source":         "programs",
+			"vendor":         "The Wireshark developer community, https://www.wireshark.org",
+			"installed_path": "C:\\Program Files\\Wireshark",
+			"last_opened_at": "foobar",
+		},
+	}
+	var w3 bytes.Buffer
+	logger3 := log.NewJSONLogger(&w3)
+	detailQueries = osquery_utils.GetDetailQueries(context.Background(), config.FleetConfig{}, appConfig, &appConfig.Features)
+	err = detailQueries["software_windows"].DirectIngestFunc(
+		context.Background(),
+		logger3,
+		globalHost,
+		s.ds,
+		rows,
+	)
+	require.NoError(t, err)
+	logs3, err := io.ReadAll(&w3)
+	require.NoError(t, err)
+	require.Contains(t, string(logs3), "host reported software with invalid last opened timestamp")
+	require.Contains(t, string(logs3), "debug")
+
+	// Check that the software was properly ingested.
+	var wiresharkSoftware fleet.Software
+	mysql.ExecAdhocSQL(t, s.ds, func(q sqlx.ExtContext) error {
+		return sqlx.GetContext(context.Background(), q, &wiresharkSoftware, softwareQueryByName, "Wireshark 4.0.8 64-bit")
+	})
+	require.NotZero(t, wiresharkSoftware.ID)
+}

server/service/osquery_utils/queries.go

@@ -11,7 +11,6 @@ import (
 	"strconv"
 	"strings"
 	"time"
-	"unicode/utf8"
 
 	"github.com/fleetdm/fleet/v4/server/config"
 	"github.com/fleetdm/fleet/v4/server/contexts/ctxerr"
@@ -1163,65 +1162,36 @@ func directIngestSoftware(ctx context.Context, logger log.Logger, host *fleet.Ho
 	sPaths := map[string]struct{}{}
 
 	for _, row := range rows {
-		name := row["name"]
-		version := row["version"]
-		source := row["source"]
-		bundleIdentifier := row["bundle_identifier"]
-		vendor := row["vendor"]
-
-		if name == "" {
+		// Attempt to parse the last_opened_at and emit a debug log if it fails.
+		if _, err := fleet.ParseSoftwareLastOpenedAtRowValue(row["last_opened_at"]); err != nil {
 			level.Debug(logger).Log(
-				"msg", "host reported software with empty name",
-				"host", host.Hostname,
-				"version", version,
-				"source", source,
+				"msg", "host reported software with invalid last opened timestamp",
+				"host_id", host.ID,
+				"row", row,
+			)
+		}
+
+		s, err := fleet.SoftwareFromOsqueryRow(
+			row["name"],
+			row["version"],
+			row["source"],
+			row["vendor"],
+			row["installed_path"],
+			row["release"],
+			row["arch"],
+			row["bundle_identifier"],
+			row["last_opened_at"],
+		)
+		if err != nil {
+			level.Debug(logger).Log(
+				"msg", "failed to parse software row",
+				"host_id", host.ID,
+				"row", row,
+				"err", err,
 			)
 			continue
 		}
-		if source == "" {
-			level.Debug(logger).Log(
-				"msg", "host reported software with empty name",
-				"host", host.Hostname,
-				"version", version,
-				"name", name,
-			)
-			continue
-		}
-
-		var lastOpenedAt time.Time
-		if lastOpenedRaw := row["last_opened_at"]; lastOpenedRaw != "" {
-			if lastOpenedEpoch, err := strconv.ParseFloat(lastOpenedRaw, 64); err != nil {
-				level.Debug(logger).Log(
-					"msg", "host reported software with invalid last opened timestamp",
-					"host", host.Hostname,
-					"version", version,
-					"name", name,
-					"last_opened_at", lastOpenedRaw,
-				)
-			} else if lastOpenedEpoch > 0 {
-				lastOpenedAt = time.Unix(int64(lastOpenedEpoch), 0).UTC()
-			}
-		}
-
-		// Check whether the vendor is longer than the max allowed width and if so, truncate it.
-		if utf8.RuneCountInString(vendor) >= fleet.SoftwareVendorMaxLength {
-			vendor = fmt.Sprintf(fleet.SoftwareVendorMaxLengthFmt, vendor)
-		}
-
-		s := fleet.Software{
-			Name:             name,
-			Version:          version,
-			Source:           source,
-			BundleIdentifier: bundleIdentifier,
-
-			Release: row["release"],
-			Vendor:  vendor,
-			Arch:    row["arch"],
-		}
-		if !lastOpenedAt.IsZero() {
-			s.LastOpenedAt = &lastOpenedAt
-		}
-		software = append(software, s)
+		software = append(software, *s)
 
 		installedPath := strings.TrimSpace(row["installed_path"])
 		if installedPath != "" &&