changed how the IP sweep filtering works

Author: caffix

engine/plugins/dns/plugin.go

@@ -236,9 +236,17 @@ func (d *dnsPlugin) lookupWithinTTL(session et.Session, name string, atype oam.A
 }
 
 func sweepCallback(e *et.Event, ip *oamnet.IPAddress, src *et.Source) {
-	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
 
+	// ensure we do not work on an IP address that was processed previously
+	_, err := e.Session.DB().FindEntitiesByContent(ctx, oam.IPAddress, e.Session.StartTime(), 1, dbt.ContentFilters{
+		"address": ip.Address.String(),
+	})
+	if err == nil {
+		return
+	}
+
 	entity, err := e.Session.DB().CreateAsset(ctx, ip)
 	if err == nil && entity != nil {
 		_, _ = e.Session.DB().CreateEntityProperty(ctx, entity, &general.SourceProperty{

engine/plugins/horizontals/plugin.go

@@ -325,9 +325,17 @@ func (h *horizPlugin) sweepAroundIPs(ctx context.Context, e *et.Event, nb *dbt.E
 */
 
 func (h *horizPlugin) submitIPAddresses(e *et.Event, asset *oamnet.IPAddress, src *et.Source) {
-	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
 
+	// ensure we do not work on an IP address that was processed previously
+	_, err := e.Session.DB().FindEntitiesByContent(ctx, oam.IPAddress, e.Session.StartTime(), 1, dbt.ContentFilters{
+		"address": asset.Address.String(),
+	})
+	if err == nil {
+		return
+	}
+
 	addr, err := e.Session.DB().CreateAsset(ctx, asset)
 	if err == nil && addr != nil {
 		_, _ = e.Session.DB().CreateEntityProperty(ctx, addr, &oamgen.SourceProperty{

engine/plugins/support/support.go

@@ -170,20 +170,10 @@ func AddNetblock(session et.Session, cidr string, asn int, src *et.Source) error
 }
 
 func IPAddressSweep(e *et.Event, addr *oamnet.IPAddress, src *et.Source, size int, callback SweepCallback) {
-	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
-	defer cancel()
-
-	// ensure we do not work on an IP address that was processed previously
-	_, err := e.Session.DB().FindEntitiesByContent(ctx, oam.IPAddress, e.Session.StartTime(), 1, dbt.ContentFilters{
-		"address": addr.Address.String(),
-	})
-	if err == nil {
-		return
-	}
-
 	var mask net.IPMask
 	addrstr := addr.Address.String()
 	ip := net.ParseIP(addrstr)
+
 	if amassnet.IsIPv4(ip) {
 		mask = net.CIDRMask(18, 32)
 	} else if amassnet.IsIPv6(ip) {