diff --git a/Makefile b/Makefile index 10edd4b..1cc454a 100644 --- a/Makefile +++ b/Makefile @@ -62,7 +62,7 @@ run: $(RUN) go run $(LDFLAGS) $(SOURCE) --input-dummy=0 --output-http="http://localhost:9000" --input-raw 127.0.0.1:9000 --input-http 127.0.0.1:9000 --verbose --debug --middleware "./examples/middleware/echo.sh" run-2: - $(RUN) go run $(SOURCE) --input-file ./fixtures/requests.gor --output-dummy=0 + sudo -E go run $(SOURCE) --input-raw :8000 --output-http "http://localhost:8001" --verbose --output-http-workers 1 record: $(RUN) go run $(SOURCE) --input-dummy=0 --output-file=requests.gor --verbose --debug diff --git a/http_client.go b/http_client.go index b82da7f..ecfdccc 100644 --- a/http_client.go +++ b/http_client.go @@ -13,10 +13,17 @@ import ( "sync" "syscall" "time" + "bytes" ) var httpMu sync.Mutex +const ( + readChunkSize = 64*1024 + maxResponseSize = 1073741824 +) +var chunkedSuffix []byte = []byte("0\r\n\r\n") + var defaultPorts = map[string]string{ "http": "80", "https": "443", @@ -57,6 +64,8 @@ func NewHTTPClient(baseURL string, config *HTTPClientConfig) *HTTPClient { config.Timeout = 5 * time.Second } + config.ConnectionTimeout = time.Second + if config.ResponseBufferSize == 0 { config.ResponseBufferSize = 100 * 1024 // 100kb } @@ -160,27 +169,115 @@ func (c *HTTPClient) Send(data []byte) (response []byte, err error) { return } - c.conn.SetReadDeadline(timeout) - n, err := c.conn.Read(c.respBuf) + var readBytes, n int + var currentChunk []byte + timeout = time.Now().Add(c.config.Timeout) + chunked := false + contentLength := -1 + currentContentLength := 0 + chunks := 0 - // If response large then our buffer, we need to read all response buffer - // Otherwise it will corrupt response of next request - // Parsing response body is non trivial thing, especially with keep-alive - // Simples case is to to close connection if response too large - // - // See https://github.com/buger/gor/issues/184 - if n == len(c.respBuf) { - c.Disconnect() + for { + c.conn.SetReadDeadline(timeout) + + if readBytes < len(c.respBuf) { + n, err = c.conn.Read(c.respBuf[readBytes:]) + readBytes += n + chunks++ + + if err != nil { + if err == io.EOF { + err = nil + } + break + } + + // First chunk + if (chunked || contentLength != -1) { + currentContentLength += n + } else { + if bytes.Equal(proto.Header(c.respBuf, []byte("Transfer-Encoding")), []byte("chunked")) { + chunked = true + } else { + l := proto.Header(c.respBuf, []byte("Content-Length")) + if len(l) > 0 { + contentLength, _ = strconv.Atoi(string(l)) + } + } + + currentContentLength += len(proto.Body(c.respBuf)) + } + + if chunked { + // Check if chunked message finished + if bytes.HasSuffix(c.respBuf[:readBytes], chunkedSuffix) { + break + } + } else if contentLength != -1 { + if currentContentLength > contentLength { + c.Disconnect() + break + } else if currentContentLength == contentLength { + break + } + } + } else { + if currentChunk == nil { + currentChunk = make([]byte, readChunkSize) + } + + n, err = c.conn.Read(currentChunk) + + if err == io.EOF { + break + } else if err != nil { + Debug("[HTTPClient] Read the whole body error:", err, c.baseURL) + break + } + + readBytes += int(n) + chunks++ + currentContentLength += n + + if chunked { + // Check if chunked message finished + if bytes.HasSuffix(currentChunk[:n], chunkedSuffix) { + break + } + } else if contentLength != -1 { + if currentContentLength > contentLength { + c.Disconnect() + break + } else if currentContentLength == contentLength { + break + } + } else { + c.Disconnect() + break + } + } + + if readBytes >= maxResponseSize { + Debug("[HTTPClient] Body is more than the max size", maxResponseSize, + c.baseURL) + break + } + + // For following chunks expect less timeout + timeout = time.Now().Add(c.config.Timeout / 5) } if err != nil { - Debug("[HTTPClient] Response read error", err, c.conn) + Debug("[HTTPClient] Response read error", err, c.conn, readBytes) response = errorPayload(HTTP_TIMEOUT) return } - payload := make([]byte, n) - copy(payload, c.respBuf[:n]) + if readBytes > len(c.respBuf) { + readBytes = len(c.respBuf) + } + payload := make([]byte, readBytes) + copy(payload, c.respBuf[:readBytes]) if c.config.Debug { Debug("[HTTPClient] Received:", string(payload)) diff --git a/http_client_test.go b/http_client_test.go index 2ca2850..04fe7c7 100644 --- a/http_client_test.go +++ b/http_client_test.go @@ -14,7 +14,6 @@ import ( "sync" "testing" "time" - _ "time" ) func TestHTTPClientURLPort(t *testing.T) { @@ -90,33 +89,67 @@ func TestHTTPClientSend(t *testing.T) { // https://github.com/buger/gor/issues/184 func TestHTTPClientResponseBuffer(t *testing.T) { - wg := new(sync.WaitGroup) + testCases := []struct { + name string + responseSize int + buffserSize int + expectedSize int + timeout time.Duration + }{ + { "Chunked, buffer overflow", 10 * 1024, 1024, 1024, 50*time.Millisecond }, - payload := []byte("GET / HTTP/1.1\r\n\r\n") + { "Chunked, fits buffer", 10 * 1024, 64 * 1024, 10*1024 + 145 /* headers length + chunked meta */, 50*time.Millisecond }, - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + { "Content-Length, buffer overflow", 1024, 1000, 1000, 50*time.Millisecond }, - size := 10 * 1024 // 10kb - rb := make([]byte, size) - rand.Read(rb) + { "Content-Length, fits buffer", 1024, 64 * 1024, 1024 + 118, 50*time.Millisecond }, + } - w.Write(rb) + for _, tc := range testCases { + wg := new(sync.WaitGroup) - wg.Done() - })) - defer server.Close() + payload := []byte("GET / HTTP/1.1\r\n\r\n") - client := NewHTTPClient(server.URL, &HTTPClientConfig{Debug: false, ResponseBufferSize: 1024}) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - wg.Add(2) - client.Send(payload) - resp, _ := client.Send(payload) + size := tc.responseSize // 1kb + rb := make([]byte, size) + rand.Read(rb) - if !bytes.Equal(resp[0:8], []byte("HTTP/1.1")) { - t.Error("Response buffer contains data from previous request", string(resp[0:5])) + w.Write(rb[:size/2]) + w.Write(rb[size/2:]) + + wg.Done() + })) + + client := NewHTTPClient(server.URL, &HTTPClientConfig{Debug: true, ResponseBufferSize: tc.buffserSize, Timeout: 100 * time.Millisecond}) + + wg.Add(2) + + start := time.Now() + client.Send(payload) + resp, err := client.Send(payload) + stop := time.Now() + + if err != nil { + t.Error("Request error", err) + } + + if stop.Sub(start) > tc.timeout { + t.Error("Request took too long", stop.Sub(start), tc.timeout) + } + + if len(resp) != tc.expectedSize { + t.Error(tc.name, " - Wrong response size:", tc.expectedSize, len(resp)) + } else { + if !bytes.Equal(resp[0:8], []byte("HTTP/1.1")) { + t.Error(tc.name, " - Response buffer contains data from previous request", string(resp), len(resp)) + } + } + + wg.Wait() + server.Close() } - - wg.Wait() } func TestHTTPClientHTTPSSend(t *testing.T) { diff --git a/input_dummy.go b/input_dummy.go index 935c16a..a2b4674 100644 --- a/input_dummy.go +++ b/input_dummy.go @@ -35,7 +35,7 @@ func (i *DummyInput) emit() { case <-ticker.C: uuid := uuid() reqh := payloadHeader(RequestPayload, uuid, time.Now().UnixNano()) - i.data <- append(reqh, []byte("GET /HTTP/1.1\r\nHost: www.w3.org\r\nUser-Agent: Go 1.1 package http\r\nAccept-Encoding: gzip\r\n\r\n")...) + i.data <- append(reqh, []byte("GET / HTTP/1.1\r\nHost: www.w3.org\r\nUser-Agent: Go 1.1 package http\r\nAccept-Encoding: gzip\r\n\r\n")...) resh := payloadHeader(ResponsePayload, uuid, 1) i.data <- append(resh, []byte("HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n")...) diff --git a/raw_socket_listener/listener.go b/raw_socket_listener/listener.go index 0148fef..2e78be3 100644 --- a/raw_socket_listener/listener.go +++ b/raw_socket_listener/listener.go @@ -221,6 +221,10 @@ type DeviceNotFoundError struct { func (e *DeviceNotFoundError) Error() string { devices, _ := pcap.FindAllDevs() + if len(devices) == 0 { + return "Can't get list of network interfaces, ensure that you running Gor as root user or sudo.\nTo run as non-root users see this docs https://github.com/buger/gor/wiki/Running-as-non-root-user" + } + var msg string msg += "Can't find interfaces with addr: " + e.addr + ". Provide available IP for intercepting traffic: \n" for _, device := range devices {