Fixes for the intel subcommand

Committer: caffix <caffix@users.noreply.github.com>

cmd/amass/intel.go

@@ -260,7 +260,9 @@ func runIntelCommand(clArgs []string) {
 		go func() { _ = ic.HostedDomains(ctx) }()
 	}
 
-	processIntelOutput(ic, &args)
+	if !processIntelOutput(ic, &args) {
+		os.Exit(1)
+	}
 }
 
 func printNetblocks(asns []int, cfg *config.Config, sys systems.System) {
@@ -279,7 +281,7 @@ func printNetblocks(asns []int, cfg *config.Config, sys systems.System) {
 	}
 }
 
-func processIntelOutput(ic *intel.Collection, args *intelArgs) {
+func processIntelOutput(ic *intel.Collection, args *intelArgs) bool {
 	var err error
 	dir := config.OutputDirectory(ic.Config.Dir)
 
@@ -303,6 +305,7 @@ func processIntelOutput(ic *intel.Collection, args *intelArgs) {
 		_, _ = outptr.Seek(0, 0)
 	}
 
+	var found bool
 	// Collect all the names returned by the intelligence collection
 	for out := range ic.Output {
 		source, _, ips := format.OutputLineParts(out, args.Options.Sources,
@@ -317,7 +320,9 @@ func processIntelOutput(ic *intel.Collection, args *intelArgs) {
 		if outptr != nil {
 			fmt.Fprintf(outptr, "%s%s%s\n", source, out.Domain, ips)
 		}
+		found = true
 	}
+	return found
 }
 
 // Obtain parameters from provided input files

intel/input.go

@@ -57,24 +57,7 @@ func (r *intelSource) Next(ctx context.Context) bool {
 	default:
 	}
 
-	if !r.queue.Empty() {
-		return true
-	}
-
-	t := time.NewTimer(r.timeout)
-	defer t.Stop()
-
-	for {
-		select {
-		case <-t.C:
-			close(r.done)
-			return false
-		case <-r.queue.Signal():
-			if !r.queue.Empty() {
-				return true
-			}
-		}
-	}
+	return !r.queue.Empty()
 }
 
 // Data implements the pipeline InputSource interface.

intel/intel.go

@@ -26,8 +26,8 @@ import (
 )
 
 const (
-	maxDnsPipelineTasks    int = 15000
-	maxActivePipelineTasks int = 25
+	maxDnsPipelineTasks    int = 2000
+	maxActivePipelineTasks int = 50
 )
 
 // Collection is the object type used to execute a open source information gathering with Amass.
@@ -76,16 +76,12 @@ func (c *Collection) HostedDomains(ctx context.Context) error {
 		return err
 	}
 
+	defer close(c.Output)
 	// Setup the context used throughout the collection
 	var cancel context.CancelFunc
 	c.ctx, cancel = context.WithCancel(ctx)
 	defer cancel()
 
-	go func() {
-		<-ctx.Done()
-		close(c.Output)
-	}()
-
 	var stages []pipeline.Stage
 	stages = append(stages, pipeline.DynamicPool("", c.makeDNSTaskFunc(), maxDnsPipelineTasks))
 	if c.Config.Active {
@@ -104,11 +100,9 @@ func (c *Collection) HostedDomains(ctx context.Context) error {
 			continue
 		}
 
-		go func(n *net.IPNet) {
-			for _, addr := range amassnet.AllHosts(n) {
-				source.InputAddress(&requests.AddrRequest{Address: addr.String()})
-			}
-		}(cidr)
+		for _, addr := range amassnet.AllHosts(cidr) {
+			source.InputAddress(&requests.AddrRequest{Address: addr.String()})
+		}
 	}
 
 	return pipeline.NewPipeline(stages...).Execute(ctx, source, c.makeOutputSink())