commit 034ab45abb6e0aab345659c5c1345def2232524b Author: Leonid Bugaev Date: Thu May 30 15:38:18 2013 +0600 First commit diff --git a/README.md b/README.md new file mode 100644 index 0000000..e69de29 diff --git a/src/gor/gor b/src/gor/gor new file mode 100755 index 0000000..ef8af36 Binary files /dev/null and b/src/gor/gor differ diff --git a/src/gor/gor.go b/src/gor/gor.go new file mode 100644 index 0000000..1918064 --- /dev/null +++ b/src/gor/gor.go @@ -0,0 +1,32 @@ +package main + +import ( + "flag" + "fmt" + "gor/listener" + "gor/replay" + "os" +) + +var mode string + +func main() { + if len(os.Args) > 1 { + mode = os.Args[1] + } + + // Remove mode attr + os.Args = append(os.Args[:1], os.Args[2:]...) + + flag.Parse() + + switch mode { + case "listen": + listener.Run() + case "replay": + replay.Run() + default: + fmt.Println("Usage: gor listen") + } + +} diff --git a/src/gor/listener/listener.go b/src/gor/listener/listener.go new file mode 100644 index 0000000..07179ad --- /dev/null +++ b/src/gor/listener/listener.go @@ -0,0 +1,161 @@ +package listener + +import ( + "fmt" + "io" + "log" + "os/exec" + //"time" + "bufio" + "bytes" + "encoding/gob" + //"errors" + "flag" + "net" + "os" + "regexp" + "strconv" + "strings" +) + +type ListenerSettings struct { + networkInterface string + port int + + replayAddress string +} + +var settings ListenerSettings = ListenerSettings{} + +type HttpRequest struct { + Tag string + Method string + Url string + Headers map[string]string +} + +func readOutput(pipe io.ReadCloser, c chan *HttpRequest, err chan int) { + re := regexp.MustCompile("(GET) (/.*) HTTP/1.1") + headers_re := regexp.MustCompile("([^ ]*): (.*)") + + reader := bufio.NewScanner(pipe) + + var request *HttpRequest + + var requestStarted = false + + for reader.Scan() { + line := reader.Text() + + if strings.Index(line, "HTTP/1.1") != -1 { + match := re.FindAllString(line, -1) + + if len(match) > 0 { + info := strings.Split(match[0], " ") + + request = &HttpRequest{ + Method: info[0], + Url: info[1], + Headers: make(map[string]string), + } + + requestStarted = true + } + } + + if requestStarted { + if line == "" { + c <- request + requestStarted = false + } else { + match := headers_re.FindAllString(line, -1) + + if len(match) > 0 { + header := strings.Split(match[0], ": ") + + request.Headers[header[0]] = header[1] + } + } + } + } +} + +func sendOutput(c chan *HttpRequest, quite chan int) { + serverAddr, err := net.ResolveUDPAddr("udp4", settings.replayAddress) + conn, err := net.DialUDP("udp", nil, serverAddr) + + defer conn.Close() + + if err != nil { + log.Fatal("Connection error", err) + } + + for { + select { + case request := <-c: + fmt.Println("Request:", request.Url) + + msg := bytes.Buffer{} + + enc := gob.NewEncoder(&msg) + err := enc.Encode(request) + + conn.Write(msg.Bytes()) + + if err != nil { + log.Println("encode error:", err) + } + + case <-quite: + conn.Close() + return + } + } +} + +func Run() { + fmt.Println("Settings:", settings) + + if !strings.Contains(settings.replayAddress, ":") { + settings.replayAddress = settings.replayAddress + ":28020" + } + + cmd := exec.Command("tcpdump", "-vv", "-A", "-i", settings.networkInterface, "port "+strconv.Itoa(settings.port)) + //cmd := exec.Command("ls", "-al") + + stdout, _ := cmd.StdoutPipe() + cmd.Stderr = os.Stderr + + if err := cmd.Start(); err != nil { + log.Fatal(err) + } + + c := make(chan *HttpRequest) + err := make(chan int) + + go readOutput(stdout, c, err) + go sendOutput(c, err) + + if err := cmd.Wait(); err != nil { + flag.Usage() + } +} + +func init() { + if len(os.Args) > 1 && os.Args[1] != "listen" { + return + } + + const ( + defaultPort = 80 + defaultNetworkInterface = "any" + + defaultReplayAddress = "localhost:28020" + ) + + flag.IntVar(&settings.port, "p", defaultPort, "Specify the http server port whose traffic you want to capture") + + flag.StringVar(&settings.networkInterface, "i", defaultNetworkInterface, "By default it try to listen on all network interfaces.To get list of interfaces run `ifconfig`") + + flag.StringVar(&settings.replayAddress, "r", defaultReplayAddress, "Address of replay server.") +} diff --git a/src/gor/replay/replay.go b/src/gor/replay/replay.go new file mode 100644 index 0000000..f7257b5 --- /dev/null +++ b/src/gor/replay/replay.go @@ -0,0 +1,172 @@ +package replay + +import ( + "bytes" + "encoding/gob" + "flag" + "fmt" + "log" + "net" + "net/http" + "os" + "strconv" +) + +const ( + bufSize = 1024 * 10 +) + +type ReplaySettings struct { + port int + host string + + limit int + + address string +} + +var settings ReplaySettings = ReplaySettings{} + +type HttpRequest struct { + Tag string + Method string + Url string + Headers map[string]string +} + +type Response struct { + req *HttpRequest + resp *http.Response + err error +} + +func sendRequest(request *HttpRequest, responses chan *Response) { + var req *http.Request + + client := &http.Client{} + + req, err := http.NewRequest("GET", settings.address+request.Url, nil) + + if err != nil { + responses <- &Response{err: err} + return + } + + for key, value := range request.Headers { + req.Header.Add(key, value) + } + + resp, err := client.Do(req) + + responses <- &Response{request, resp, err} +} + +func handleRequests(requests chan *HttpRequest) { + stat := &RequestStat{} + stat.reset() + + responses := make(chan *Response) + + for { + select { + case req := <-requests: + go sendRequest(req, responses) + + case resp := <-responses: + stat.inc(resp) + + if resp.err != nil { + log.Println("Request err:", resp.err) + } else { + log.Println("Request ok:", resp.req.Url) + } + } + } +} + +func decodeRequest(enc []byte) (request *HttpRequest, err error) { + var buf bytes.Buffer + buf.Write(enc) + + request = &HttpRequest{} + + encoder := gob.NewDecoder(&buf) + err = encoder.Decode(request) + + return +} + +func Run() { + var buf [bufSize]byte + + serverAddress := settings.host + ":" + strconv.Itoa(settings.port) + + addr, err := net.ResolveUDPAddr("udp", serverAddress) + if err != nil { + log.Fatal("Can't start:", err) + } + + conn, err := net.ListenUDP("udp", addr) + if err != nil { + log.Fatal("Can't start:", err) + } + + defer conn.Close() + + if settings.address == "" { + + } + + fmt.Println("Starting replay server at:", serverAddress) + fmt.Println("Forwarding incoming requests to:", settings.address) + fmt.Println("Limiting concurrent request count to:", settings.limit) + + requests := make(chan *HttpRequest) + + go handleRequests(requests) + + for { + rlen, _, err := conn.ReadFromUDP(buf[0:]) + + if err != nil { + continue + } + + if rlen > 0 { + if rlen > bufSize { + log.Fatal("Too large udp packet", bufSize) + } + + request, err := decodeRequest(buf[0:rlen]) + + if err != nil { + log.Println("Decode error:", err) + } else { + requests <- request + } + } + } + +} + +func init() { + if len(os.Args) > 1 && os.Args[1] != "replay" { + return + } + + const ( + defaultPort = 28020 + defaultHost = "0.0.0.0" + + defaultLimit = 10 + defaultAddress = "http://localhost:8080" + ) + + flag.IntVar(&settings.port, "p", defaultPort, "specify port number") + + flag.StringVar(&settings.host, "ip", defaultHost, "ip addresses to listen on") + + flag.IntVar(&settings.limit, "l", defaultLimit, "limit number for requests per second. It will start dropping packets.") + + flag.StringVar(&settings.address, "h", defaultAddress, "http address to forward traffic ") +} diff --git a/src/gor/replay/request_stats.go b/src/gor/replay/request_stats.go new file mode 100644 index 0000000..94881fb --- /dev/null +++ b/src/gor/replay/request_stats.go @@ -0,0 +1,39 @@ +package replay + +import ( + "log" + "time" +) + +type RequestStat struct { + timestamp int64 + + codes map[int]int + + count int + errors int +} + +func (s *RequestStat) inc(resp *Response) { + if s.timestamp != time.Now().Unix() { + s.reset() + } + + if resp.err != nil { + s.errors++ + return + } + + s.count++ + s.codes[resp.resp.StatusCode]++ +} + +func (s *RequestStat) reset() { + log.Println("reseting stats", s) + + s.timestamp = time.Now().Unix() + + s.codes = make(map[int]int) + s.count = 0 + s.errors = 0 +}