Close rows after use

analysis/spagooper.go

@@ -333,6 +333,7 @@ func (analyzer *Analyzer) ScoopSNIConns(ctx context.Context, bars *tea.Program)
 		// return error and cancel all uconn analysis
 		return fmt.Errorf("could not retrieve unique SNI connections for analysis: %w", err)
 	}
+	defer rows.Close()
 	logger.Debug().Msg("successfully retrieved SNI connections")
 
 	i := uint64(0)
@@ -342,7 +343,6 @@ func (analyzer *Analyzer) ScoopSNIConns(ctx context.Context, bars *tea.Program)
 		// abort this function if the context was cancelled
 		case <-ctx.Done():
 			logger.Warn().Msg("cancelling SNI uconns query for analysis")
-			rows.Close()
 			return ctx.Err()
 		default:
 			var res AnalysisResult
@@ -358,7 +358,9 @@ func (analyzer *Analyzer) ScoopSNIConns(ctx context.Context, bars *tea.Program)
 			i++
 		}
 	}
-	rows.Close()
+	if err := rows.Err(); err != nil {
+		return fmt.Errorf("could not read unique SNI connections for analysis: %w", err)
+	}
 	bars.Send(progressbar.ProgressMsg{ID: 1, Percent: 1})
 	return nil
 }
@@ -553,14 +555,14 @@ func (analyzer *Analyzer) ScoopIPConns(ctx context.Context, bars *tea.Program) e
 		// return error and cancel all uconn analysis
 		return fmt.Errorf("could not retrieve unique IP connections for analysis: %w", err)
 	}
+	defer rows.Close()
 	logger.Debug().Msg("successsfully retrieved IP connections")
 	// loop over the rows
 	for rows.Next() {
 		select {
 		// abort this function if the context was cancelled
 		case <-ctx.Done():
 			logger.Warn().Msg("cancelling IP uconns query for analysis")
-			rows.Close()
 			return ctx.Err()
 		default:
 			var res AnalysisResult
@@ -573,7 +575,9 @@ func (analyzer *Analyzer) ScoopIPConns(ctx context.Context, bars *tea.Program) e
 			analyzer.UconnChan <- res
 		}
 	}
-	rows.Close()
+	if err := rows.Err(); err != nil {
+		return fmt.Errorf("could not read IP connections for analysis: %w", err)
+	}
 	return nil
 }
 
@@ -728,14 +732,14 @@ func (analyzer *Analyzer) ScoopDNS(ctx context.Context, bars *tea.Program) error
 		// return error and cancel all uconn analysis
 		return fmt.Errorf("could not retrieve unique exploded domains for analysis: %w", err)
 	}
+	defer rows.Close()
 	logger.Debug().Msg("successfully retrieved exploded dns")
 	// loop over the rows
 	for rows.Next() {
 		select {
 		// abort this function if the context was cancelled
 		case <-ctx.Done():
 			logger.Warn().Msg("cancelling exploded dns query for analysis")
-			rows.Close()
 			return ctx.Err()
 		default:
 			var res AnalysisResult
@@ -747,6 +751,8 @@ func (analyzer *Analyzer) ScoopDNS(ctx context.Context, bars *tea.Program) error
 			analyzer.UconnChan <- res
 		}
 	}
-	rows.Close()
+	if err := rows.Err(); err != nil {
+		return fmt.Errorf("could not read exploded dns for analysis: %w", err)
+	}
 	return nil
 }

database/server.go

@@ -260,6 +260,9 @@ func (server *ServerConn) DropMultipleSensorDatabases(dbName string, wildcardSta
 		// increment the number of databases deleted
 		numDeleted++
 	}
+	if err := rows.Err(); err != nil {
+		return numDeleted, err
+	}
 
 	return numDeleted, nil
 }

database/threat_intel.go

@@ -102,6 +102,7 @@ func (server *ServerConn) syncThreatIntelFeedsFromConfig(afs afero.Fs, cfg *conf
 	if err != nil {
 		return err
 	}
+	defer rows.Close()
 
 	// create a rate limiter to control the rate of writing to the database
 	limiter := rate.NewLimiter(5, 5)
@@ -187,6 +188,9 @@ func (server *ServerConn) syncThreatIntelFeedsFromConfig(afs afero.Fs, cfg *conf
 		}
 
 	}
+	if err := rows.Err(); err != nil {
+		return err
+	}
 
 	// iterate over each feed in the config that was not in the database
 	for path := range feeds {

importer/http.go

@@ -270,13 +270,13 @@ func (importer *Importer) writeLinkedHTTP(ctx context.Context, progress *tea.Pro
 	if err != nil {
 		return err
 	}
+	defer rows.Close()
 	i := 0
 	for rows.Next() {
 		select {
 		// abort this function if the context was cancelled
 		case <-ctx.Done():
 			logger.Warn().Msg("cancelling HTTP connection linking")
-			rows.Close()
 			return ctx.Err()
 		default:
 			var entry HTTPEntry
@@ -339,7 +339,9 @@ func (importer *Importer) writeLinkedHTTP(ctx context.Context, progress *tea.Pro
 			}
 		}
 	}
-	rows.Close()
+	if err := rows.Err(); err != nil {
+		return err
+	}
 	progress.Send(progressbar.ProgressMsg{ID: barID, Percent: 1})
 
 	return nil

importer/ssl.go

@@ -220,14 +220,14 @@ func (importer *Importer) writeLinkedSSL(ctx context.Context, progress *tea.Prog
 	if err != nil {
 		log.Panicln(err)
 	}
+	defer rows.Close()
 
 	i := 0
 	for rows.Next() {
 		select {
 		// abort this function if the context was cancelled
 		case <-ctx.Done():
 			logger.Warn().Msg("cancelling SSL connection linking")
-			rows.Close()
 			return ctx.Err()
 		default:
 			var entry SSLEntry
@@ -251,7 +251,9 @@ func (importer *Importer) writeLinkedSSL(ctx context.Context, progress *tea.Prog
 			sslWriter.WriteChannel <- &entry
 		}
 	}
-	rows.Close()
+	if err := rows.Err(); err != nil {
+		return err
+	}
 	progress.Send(progressbar.ProgressMsg{ID: barID, Percent: 1})
 
 	return nil

integration/missing_host_test.go

@@ -142,6 +142,7 @@ func (it *MissingHostSuite) TestThreat() {
 	ctx = it.db.QueryParameters(params)
 	rows, err := it.db.Conn.Query(ctx, query)
 	require.NoError(t, err, "getting the mixtape results shouldn't error")
+	defer rows.Close()
 
 	i := 0
 	for rows.Next() {
@@ -167,7 +168,7 @@ func (it *MissingHostSuite) TestThreat() {
 		require.ElementsMatch(t, expected.portProtoService, res.PortProtoService, "port:proto:service arrays should match")
 		i++
 	}
-	rows.Close()
+	require.NoError(t, rows.Err())
 	require.EqualValues(t, 1, i, "there should only be one row for an aggregated result in the mixtape: 73.54.23.243 -> 64.225.56.201")
 
 	// verify modifier for the threat
@@ -179,6 +180,7 @@ func (it *MissingHostSuite) TestThreat() {
 		AND import_id = (SELECT argMax(import_id, analyzed_at) FROM threat_mixtape)
 	`)
 	require.NoError(t, err, "retrieving the modifiers for this threat should not error")
+	defer rows.Close()
 
 	i = 0
 	for rows.Next() {
@@ -192,6 +194,7 @@ func (it *MissingHostSuite) TestThreat() {
 		require.InDelta(t, it.cfg.Modifiers.RareSignatureScoreIncrease, score, 0.001, "the rare signature score should match the config modifier value")
 		i++
 	}
+	require.NoError(t, rows.Err())
 	require.EqualValues(t, 1, i, "there should only be one modifier for 73.54.23.243 -> 64.225.56.201")
 
 	//  verify that the threat has the ICMP protocol in port:proto:service (in a previous import)

integration/networkid_test.go

@@ -94,6 +94,7 @@ func (it *NetworkIDSuite) TestNetworkIDSeparation() {
 		GROUP BY src_nuid
 	`)
 	require.NoError(t, err, "retrieving the unique src IPs for each src nuid should not error")
+	defer rows.Close()
 
 	i := 0
 	for rows.Next() {
@@ -104,7 +105,7 @@ func (it *NetworkIDSuite) TestNetworkIDSeparation() {
 		require.ElementsMatch(t, srcIPs, expectedSrcNUIDs[srcNUID])
 		i++
 	}
-	rows.Close()
+	require.NoError(t, rows.Err())
 	require.EqualValues(t, len(expectedSrcNUIDs), i, "there should be %d unique src_nuids", len(expectedSrcNUIDs))
 
 	// verify that all expected destination nuids exist and have the right destination IPs
@@ -119,6 +120,7 @@ func (it *NetworkIDSuite) TestNetworkIDSeparation() {
 		GROUP BY dst_nuid
 	`)
 	require.NoError(t, err)
+	defer rows.Close()
 	i = 0
 	hadExternalID := false
 	for rows.Next() {
@@ -134,7 +136,7 @@ func (it *NetworkIDSuite) TestNetworkIDSeparation() {
 		}
 		i++
 	}
-	rows.Close()
+	require.NoError(t, rows.Err())
 	require.EqualValues(t, 2, i, "there should be %d unique dst_nuids", 2)
 	require.True(t, hadExternalID, "external network ID should have appeared in results")
 

integration/open_sni_test.go

@@ -84,6 +84,7 @@ func (it *OpenSNITestSuite) TestThreats() {
 	ctx := it.db.QueryParameters(params)
 	rows, err := it.db.Conn.Query(ctx, query)
 	require.NoError(t, err)
+	defer rows.Close()
 
 	i := 0
 	for rows.Next() {
@@ -114,7 +115,7 @@ func (it *OpenSNITestSuite) TestThreats() {
 		require.InDelta(t, 0, res.FirstSeenScore, 0.001, "first seen score should equal 0 for a non-rolling dataset")
 		i++
 	}
-	rows.Close()
+	require.NoError(t, rows.Err())
 
 	require.EqualValues(t, len(expectedResults), i, "there should be an equal number of expected results")
 }

integration/proxy_rolling_test.go

@@ -116,6 +116,7 @@ func (it *ProxyRollingTestSuite) TestRollingThreats() {
 	ctx := it.db.QueryParameters(params)
 	rows, err := it.db.Conn.Query(ctx, query)
 	require.NoError(t, err)
+	defer rows.Close()
 
 	i := 0
 	for rows.Next() {
@@ -136,7 +137,7 @@ func (it *ProxyRollingTestSuite) TestRollingThreats() {
 		}
 		i++
 	}
-	rows.Close()
+	require.NoError(t, rows.Err())
 }
 
 func (it *ProxyRollingTestSuite) TestProxy() {

modifier/modifier.go

@@ -139,13 +139,13 @@ func (modifier *Modifier) detectRareSignature(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
+	defer rows.Close()
 
 	for rows.Next() {
 		select {
 		// abort this function if the context was cancelled
 		case <-ctx.Done():
 			logger.Warn().Msg("cancelling rare signature modifier query")
-			rows.Close()
 			return ctx.Err()
 		default:
 			var res analysis.ThreatMixtape
@@ -169,7 +169,9 @@ func (modifier *Modifier) detectRareSignature(ctx context.Context) error {
 			modifier.writer.WriteChannel <- &res
 		}
 	}
-	rows.Close()
+	if err := rows.Err(); err != nil {
+		return fmt.Errorf("could not read entries for rare signature modifier detection: %w", err)
+	}
 
 	return nil
 }
@@ -198,13 +200,13 @@ func (modifier *Modifier) detectMIMETypeMismatch(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
+	defer rows.Close()
 
 	for rows.Next() {
 		select {
 		// abort this function if the context was cancelled
 		case <-ctx.Done():
 			logger.Warn().Msg("cancelling MIME type/URI mismatch modifier query")
-			rows.Close()
 			return ctx.Err()
 		default:
 			var res analysis.ThreatMixtape
@@ -228,7 +230,9 @@ func (modifier *Modifier) detectMIMETypeMismatch(ctx context.Context) error {
 			modifier.writer.WriteChannel <- &res
 		}
 	}
-	rows.Close()
+	if err := rows.Err(); err != nil {
+		return fmt.Errorf("could not read entries for MIME type/URI mismatch modifier detection: %w", err)
+	}
 
 	return nil
 }

viewer/results.go

@@ -184,6 +184,7 @@ func GetResults(db *database.DB, filter *Filter, currentPage, pageSize int, minT
 	if err != nil {
 		return nil, false, err
 	}
+	defer rows.Close()
 
 	var items []list.Item
 	for rows.Next() {
@@ -194,7 +195,9 @@ func GetResults(db *database.DB, filter *Filter, currentPage, pageSize int, minT
 		items = append(items, list.Item(&res))
 	}
 
-	rows.Close()
+	if err := rows.Err(); err != nil {
+		return nil, false, fmt.Errorf("could not read mixtape results for viewer: %w", err)
+	}
 
 	return items, appliedFilter, nil
 }