v0.2.17 add parallel mode

HISTORY.md

@@ -1,3 +1,9 @@
+## v0.2.17 add parallel mode
+
+-   such as ```fofa -f ip "is_ipv6=false && port=22" | fofa -f ip -uniqByIP -template "port=8443 && ip={}"```
+-   host mode raise error if data.error is true
+-   fixed bug of: ```fofa -f ip -uniqByIP 'port=22 && ip=154.19.247.29'``` return multiple lines of ip even if set uniqByIP
+
 ## v0.1.16 add domains mode
 
 -   add domains mode to extend domains from domain, through certs ```./fofa domains -s 1000 -withCount baidu.com```

README.md

@@ -105,6 +105,13 @@ redis://152.136.145.87:6379
 ./fofa --fixUrl --size 1000 --fields host --uniqByIP 'host="edu.cn"'
 ```
 
+-   pipeline with parallel mode
+
+```shell
+fofa -f ip "is_ipv6=false && port=22" | fofa -f ip -uniqByIP -template "port=8443 && ip={}" 
+```
+can use `-rate 3` to increase rate limit, default is 2
+
 ### Stats
 
 -   stats subcommand

cmd/fofa/cmd/search.go

@@ -1,30 +1,39 @@
 package cmd
 
 import (
+	"bufio"
+	"context"
 	"errors"
 	"fmt"
 	"github.com/LubyRuffy/gofofa"
 	"github.com/LubyRuffy/gofofa/pkg/outformats"
 	"github.com/sirupsen/logrus"
 	"github.com/urfave/cli/v2"
+	"golang.org/x/time/rate"
 	"io"
+	"log"
 	"os"
+	"strconv"
 	"strings"
+	"sync"
 )
 
 var (
-	fieldString string // fieldString
-	size        int    // fetch size
-	format      string // out format
-	outFile     string // out file
-	inFile      string // in file
-	deductMode  string // deduct Mode
-	fixUrl      bool   // each host fix as url, like 1.1.1.1,80 will change to http://1.1.1.1
-	urlPrefix   string // each host fix as url, like 1.1.1.1,80 will change to http://1.1.1.1
-	full        bool   // search result for over a year
-	batchSize   int    // amount of data contained in each batch, only for dump
-	json        bool   // out format as json for short
-	uniqByIP    bool   // group by ip
+	fieldString   string // fieldString
+	size          int    // fetch size
+	format        string // out format
+	outFile       string // out file
+	inFile        string // in file
+	deductMode    string // deduct Mode
+	fixUrl        bool   // each host fix as url, like 1.1.1.1,80 will change to http://1.1.1.1
+	urlPrefix     string // each host fix as url, like 1.1.1.1,80 will change to http://1.1.1.1
+	full          bool   // search result for over a year
+	batchSize     int    // amount of data contained in each batch, only for dump
+	json          bool   // out format as json for short
+	uniqByIP      bool   // group by ip
+	workers       int    // number of workers
+	ratePerSecond int    // fofa request per second
+	template      string // template in pipeline mode
 )
 
 // search subcommand
@@ -89,6 +98,24 @@ var searchCmd = &cli.Command{
 			Usage:       "uniq by ip",
 			Destination: &uniqByIP,
 		},
+		&cli.IntFlag{
+			Name:        "workers",
+			Value:       10,
+			Usage:       "number of workers",
+			Destination: &workers,
+		},
+		&cli.IntFlag{
+			Name:        "rate",
+			Value:       2,
+			Usage:       "fofa query per second",
+			Destination: &ratePerSecond,
+		},
+		&cli.StringFlag{
+			Name:        "template",
+			Value:       "ip={}",
+			Usage:       "template in pipeline mode",
+			Destination: &template,
+		},
 	},
 	Action: SearchAction,
 }
@@ -126,8 +153,10 @@ func SearchAction(ctx *cli.Context) error {
 
 	query := ctx.Args().First()
 	if len(query) == 0 {
-		return errors.New("fofa query cannot be empty")
+		//return errors.New("fofa query cannot be empty")
+		log.Println("not set fofa query, now in pipeline mode....")
 	}
+
 	fields := strings.Split(fieldString, ",")
 	if len(fields) == 0 {
 		return errors.New("fofa fields cannot be empty")
@@ -165,20 +194,65 @@ func SearchAction(ctx *cli.Context) error {
 		}
 	}
 
-	// do search
-	res, err := fofaCli.HostSearch(query, size, fields, gofofa.SearchOptions{
-		FixUrl:    fixUrl,
-		UrlPrefix: urlPrefix,
-		Full:      full,
-		UniqByIP:  uniqByIP,
-	})
-	if err != nil {
-		return err
+	writeQuery := func(query string) error {
+		log.Println("query fofa of:", query)
+		// do search
+		res, err := fofaCli.HostSearch(query, size, fields, gofofa.SearchOptions{
+			FixUrl:    fixUrl,
+			UrlPrefix: urlPrefix,
+			Full:      full,
+			UniqByIP:  uniqByIP,
+		})
+		if err != nil {
+			return err
+		}
+
+		// output
+		if err = writer.WriteAll(res); err != nil {
+			return err
+		}
+
+		return nil
 	}
 
-	// output
-	if err = writer.WriteAll(res); err != nil {
-		return err
+	if query != "" {
+		return writeQuery(query)
+	} else {
+		// 并发模式
+		wg := sync.WaitGroup{}
+		queries := make(chan string, workers)
+		limiter := rate.NewLimiter(rate.Limit(ratePerSecond), 5)
+
+		worker := func(queries <-chan string, wg *sync.WaitGroup) {
+			for q := range queries {
+				tmpQuery := strings.ReplaceAll(template, "{}",
+					strconv.Quote(q))
+				if err := limiter.Wait(context.Background()); err != nil {
+					fmt.Println("Error: ", err)
+				}
+				if err := writeQuery(tmpQuery); err != nil {
+					log.Println(err)
+				}
+				wg.Done()
+			}
+		}
+		for w := 0; w < workers; w++ {
+			go worker(queries, &wg)
+		}
+
+		scanner := bufio.NewScanner(os.Stdin)
+		for scanner.Scan() { // internally, it advances token based on sperator
+			line := scanner.Text()
+			wg.Add(1)
+			queries <- line
+		}
+
+		if err := scanner.Err(); err != nil {
+			log.Println(err)
+		}
+
+		wg.Wait()
 	}
+
 	return nil
 }

cmd/fofa/main.go

@@ -9,7 +9,7 @@ import (
 )
 
 var (
-	version = "v0.1.16"
+	version = "v0.2.17"
 	commit  = "none"
 	date    = "unknown"
 	builtBy = "unknown" // goreleaser fill

go.mod

@@ -25,6 +25,7 @@ require (
 	github.com/russross/blackfriday/v2 v2.1.0 // indirect
 	golang.org/x/sys v0.10.0 // indirect
 	golang.org/x/text v0.11.0 // indirect
+	golang.org/x/time v0.5.0 // indirect
 	gopkg.in/check.v1 v1.0.0-20200902074654-038fdea0a05b // indirect
 	gopkg.in/yaml.v3 v3.0.0-20210106172901-c476de37821d // indirect
 )

go.sum

@@ -101,6 +101,8 @@ golang.org/x/text v0.8.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
 golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
 golang.org/x/text v0.11.0 h1:LAntKIrcmeSKERyiOh0XMV39LXS8IE9UL2yP7+f5ij4=
 golang.org/x/text v0.11.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
+golang.org/x/time v0.5.0 h1:o7cqy6amK/52YcAKIPlM3a+Fpj35zvRj2TP+e1xFSfk=
+golang.org/x/time v0.5.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=

host.go

@@ -264,6 +264,13 @@ func (c *Client) HostSearch(query string, size int, fields []string, options ...
 					}
 					results = append(results, newSlice)
 				} else if vStr, ok := result.(string); ok {
+					// 确定第一个就是ip
+					if uniqByIP && ipIndex == 0 {
+						if _, ok := uniqIPMap[vStr]; ok {
+							continue
+						}
+						uniqIPMap[vStr] = true
+					}
 					results = append(results, []string{vStr})
 				}
 			}
@@ -320,6 +327,9 @@ func (c *Client) HostStats(host string) (data HostStatsData, err error) {
 	if err != nil {
 		return
 	}
+	if data.Error {
+		err = errors.New(data.Errmsg)
+	}
 	return
 }