github/goreplay
4
0
Fork 0
mirror of https://github.com/buger/goreplay.git synced 2024-04-21 12:32:02 +00:00
Code Issues Releases Wiki Activity

Do not track response by default (#279)

Browse Source 
* Do not track response by default

* Add track-response integration test

* Apply fmt
This commit is contained in:
Leonid Bugaev
2016-05-20 21:09:39 +06:00
parent 2836b555e3
commit a52f56df2b
9 changed files with 92 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Makefile
+1 -1
View File
@@ -59,7 +59,7 @@ profile_test:
# Used mainly for debugging, because docker container do not have access to parent machine ports
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) go run $(LDFLAGS) $(SOURCE) --input-dummy=0 --output-http="http://localhost:9000" --input-raw-track-response --input-raw 127.0.0.1:9000 --input-http 127.0.0.1:9000 --verbose --debug --middleware "./examples/middleware/echo.sh"
run-2:
sudo -E go run $(SOURCE) --input-raw :8000 --output-http "http://localhost:8001" --verbose --output-http-workers 1
input_raw.go
+10 -8
View File
@@ -9,12 +9,13 @@ import (
// RAWInput used for intercepting traffic for given address
type RAWInput struct {
data chan *raw.TCPMessage
address string
expire time.Duration
quit chan bool
engine int
listener *raw.Listener
data chan *raw.TCPMessage
address string
expire time.Duration
quit chan bool
engine int
trackResponse bool
listener *raw.Listener
}
// Available engines for intercepting traffic
@@ -24,13 +25,14 @@ const (
)
// NewRAWInput constructor for RAWInput. Accepts address with port as argument.
func NewRAWInput(address string, engine int, expire time.Duration) (i *RAWInput) {
func NewRAWInput(address string, engine int, trackResponse bool, expire time.Duration) (i *RAWInput) {
i = new(RAWInput)
i.data = make(chan *raw.TCPMessage)
i.address = address
i.expire = expire
i.engine = engine
i.quit = make(chan bool)
i.trackResponse = trackResponse
i.listen(address)
i.listener.IsReady()
@@ -65,7 +67,7 @@ func (i *RAWInput) listen(address string) {
log.Fatal("input-raw: error while parsing address", err)
}
i.listener = raw.NewListener(host, port, i.engine, i.expire)
i.listener = raw.NewListener(host, port, i.engine, i.trackResponse, i.expire)
ch := i.listener.Receiver()
input_raw_test.go
+6 -6
View File
@@ -42,7 +42,7 @@ func TestRAWInput(t *testing.T) {
var respCounter, reqCounter int64
input := NewRAWInput(originAddr, EnginePcap, testRawExpire)
input := NewRAWInput(originAddr, EnginePcap, true, testRawExpire)
defer input.Close()
output := NewTestOutput(func(data []byte) {
@@ -97,7 +97,7 @@ func TestRAWInputIPv6(t *testing.T) {
var respCounter, reqCounter int64
input := NewRAWInput(originAddr, EnginePcap, testRawExpire)
input := NewRAWInput(originAddr, EnginePcap, true, testRawExpire)
defer input.Close()
output := NewTestOutput(func(data []byte) {
@@ -148,7 +148,7 @@ func TestInputRAW100Expect(t *testing.T) {
originAddr := strings.Replace(origin.Listener.Addr().String(), "[::]", "127.0.0.1", -1)
input := NewRAWInput(originAddr, EnginePcap, time.Second)
input := NewRAWInput(originAddr, EnginePcap, true, time.Second)
defer input.Close()
// We will use it to get content of raw HTTP request
@@ -211,7 +211,7 @@ func TestInputRAWChunkedEncoding(t *testing.T) {
}))
originAddr := strings.Replace(origin.Listener.Addr().String(), "[::]", "127.0.0.1", -1)
input := NewRAWInput(originAddr, EnginePcap, time.Second)
input := NewRAWInput(originAddr, EnginePcap, true, time.Second)
defer input.Close()
replay := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -275,7 +275,7 @@ func TestInputRAWLargePayload(t *testing.T) {
}))
originAddr := strings.Replace(origin.Listener.Addr().String(), "[::]", "127.0.0.1", -1)
input := NewRAWInput(originAddr, EnginePcap, testRawExpire)
input := NewRAWInput(originAddr, EnginePcap, true, testRawExpire)
defer input.Close()
replay := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
@@ -320,7 +320,7 @@ func BenchmarkRAWInput(b *testing.B) {
var respCounter, reqCounter int64
input := NewRAWInput(originAddr, EnginePcap, testRawExpire)
input := NewRAWInput(originAddr, EnginePcap, true, testRawExpire)
defer input.Close()
output := NewTestOutput(func(data []byte) {
middleware_test.go
+2 -2
View File
@@ -117,7 +117,7 @@ func TestEchoMiddleware(t *testing.T) {
// Catch traffic from one service
fromAddr := strings.Replace(from.Listener.Addr().String(), "[::]", "127.0.0.1", -1)
input := NewRAWInput(fromAddr, EnginePcap, testRawExpire)
input := NewRAWInput(fromAddr, EnginePcap, true, testRawExpire)
defer input.Close()
// And redirect to another
@@ -179,7 +179,7 @@ func TestTokenMiddleware(t *testing.T) {
fromAddr := strings.Replace(from.Listener.Addr().String(), "[::]", "127.0.0.1", -1)
// Catch traffic from one service
input := NewRAWInput(fromAddr, EnginePcap, testRawExpire)
input := NewRAWInput(fromAddr, EnginePcap, true, testRawExpire)
defer input.Close()
// And redirect to another
plugins.go
+1 -1
View File
@@ -94,7 +94,7 @@ func InitPlugins() {
}
for _, options := range Settings.inputRAW {
registerPlugin(NewRAWInput, options, engine, time.Duration(0))
registerPlugin(NewRAWInput, options, engine, Settings.inputRAWTrackResponse, time.Duration(0))
}
for _, options := range Settings.inputTCP {
raw_socket_listener/listener.go
+35 -20
View File
@@ -60,6 +60,7 @@ type Listener struct {
addr string // IP to listen
port uint16 // Port to listen
trackResponse bool
messageExpire time.Duration
conn net.PacketConn
@@ -82,7 +83,7 @@ const (
)
// NewListener creates and initializes new Listener object
func NewListener(addr string, port string, engine int, expire time.Duration) (l *Listener) {
func NewListener(addr string, port string, engine int, trackResponse bool, expire time.Duration) (l *Listener) {
l = &Listener{}
l.packetsChan = make(chan []byte, 10000)
@@ -95,6 +96,7 @@ func NewListener(addr string, port string, engine int, expire time.Duration) (l
l.seqWithData = make(map[uint32]uint32)
l.respAliases = make(map[uint32]*TCPMessage)
l.respWithoutReq = make(map[uint32]tcpID)
l.trackResponse = trackResponse
l.addr = addr
_port, _ := strconv.Atoi(port)
@@ -175,22 +177,24 @@ func (t *Listener) dispatchMessage(message *TCPMessage) {
if message.IsIncoming {
// If there were response before request
// log.Println("Looking for Response: ", t.respWithoutReq, message.ResponseAck)
if respID, ok := t.respWithoutReq[message.ResponseAck]; ok {
if resp, rok := t.messages[respID]; rok {
// if resp.AssocMessage == nil {
// log.Println("FOUND RESPONSE")
resp.AssocMessage = message
message.AssocMessage = resp
if t.trackResponse {
if respID, ok := t.respWithoutReq[message.ResponseAck]; ok {
if resp, rok := t.messages[respID]; rok {
// if resp.AssocMessage == nil {
// log.Println("FOUND RESPONSE")
resp.AssocMessage = message
message.AssocMessage = resp
if resp.IsFinished() {
defer t.dispatchMessage(resp)
if resp.IsFinished() {
defer t.dispatchMessage(resp)
}
// }
}
// }
}
}
if resp, ok := t.messages[message.ResponseID]; ok {
resp.AssocMessage = message
if resp, ok := t.messages[message.ResponseID]; ok {
resp.AssocMessage = message
}
}
} else {
if message.AssocMessage == nil {
@@ -296,7 +300,13 @@ func (t *Listener) readPcap() {
t.mu.Unlock()
if bpfSupported {
bpf := "tcp port " + strconv.Itoa(int(t.port))
var bpf string
if t.trackResponse {
bpf = "tcp port " + strconv.Itoa(int(t.port))
} else {
bpf = "tcp dst port " + strconv.Itoa(int(t.port))
}
if err := handle.SetBPFFilter(bpf); err != nil {
log.Println("BPF filter error:", err, "Device:", device.Name)
wg.Done()
@@ -367,7 +377,7 @@ func (t *Listener) readPcap() {
// log.Println(t.port, destPort, srcPort, packet)
if destPort != t.port && srcPort != t.port {
if !(destPort == t.port || (t.trackResponse && srcPort == t.port)) {
continue
}
}
@@ -431,7 +441,7 @@ func (t *Listener) isValidPacket(buf []byte) bool {
srcPort := binary.BigEndian.Uint16(buf[0:2])
// Because RAW_SOCKET can't be bound to port, we have to control it by ourself
if destPort == t.port || srcPort == t.port {
if destPort == t.port || (t.trackResponse && srcPort == t.port) {
// Get the 'data offset' (size of the TCP header in 32-bit words)
dataOffset := (buf[12] & 0xF0) >> 4
@@ -566,11 +576,16 @@ func (t *Listener) processTCPPacket(packet *TCPPacket) {
// If message contains only single packet immediately dispatch it
if message.IsFinished() {
if isIncoming {
if resp, ok := t.messages[message.ResponseID]; ok {
t.dispatchMessage(message)
if resp.IsFinished() {
t.dispatchMessage(resp)
// log.Println("I'm finished", string(message.Bytes()), message.ResponseID, t.messages)
if t.trackResponse {
if resp, ok := t.messages[message.ResponseID]; ok {
t.dispatchMessage(message)
if resp.IsFinished() {
t.dispatchMessage(resp)
}
}
} else {
t.dispatchMessage(message)
}
} else {
if message.AssocMessage == nil {
raw_socket_listener/listener_test.go
+28 -6
View File
@@ -12,7 +12,7 @@ import (
func TestRawListenerInput(t *testing.T) {
var req, resp *TCPMessage
listener := NewListener("", "0", EnginePcap, 10*time.Millisecond)
listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond)
defer listener.Close()
reqPacket := buildPacket(true, 1, 1, []byte("GET / HTTP/1.1\r\n\r\n"))
@@ -46,10 +46,32 @@ func TestRawListenerInput(t *testing.T) {
}
}
func TestRawListenerInputWithoutResponse(t *testing.T) {
var req *TCPMessage
listener := NewListener("", "0", EnginePcap, false, 10*time.Millisecond)
defer listener.Close()
reqPacket := buildPacket(true, 1, 1, []byte("GET / HTTP/1.1\r\n\r\n"))
listener.packetsChan <- reqPacket.Dump()
select {
case req = <-listener.messagesChan:
case <-time.After(time.Millisecond):
t.Error("Should return request immediately")
return
}
if !req.IsIncoming {
t.Error("Should be request")
}
}
func TestRawListenerResponse(t *testing.T) {
var req, resp *TCPMessage
listener := NewListener("", "0", EnginePcap, 10*time.Millisecond)
listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond)
defer listener.Close()
reqPacket := buildPacket(true, 1, 1, []byte("GET / HTTP/1.1\r\n\r\n"))
@@ -89,7 +111,7 @@ func TestRawListenerResponse(t *testing.T) {
func TestRawListener100Continue(t *testing.T) {
var req, resp *TCPMessage
listener := NewListener("", "0", EnginePcap, 10*time.Millisecond)
listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond)
defer listener.Close()
reqPacket1 := buildPacket(true, 1, 1, []byte("POST / HTTP/1.1\r\nContent-Length: 2\r\nExpect: 100-continue\r\n\r\n"))
@@ -146,7 +168,7 @@ func TestRawListener100Continue(t *testing.T) {
func TestRawListener100ContinueWrongOrder(t *testing.T) {
var req, resp *TCPMessage
listener := NewListener("", "0", EnginePcap, 10*time.Millisecond)
listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond)
defer listener.Close()
reqPacket1 := buildPacket(true, 1, 1, []byte("POST / HTTP/1.1\r\nContent-Length: 2\r\nExpect: 100-continue\r\n\r\n"))
@@ -303,7 +325,7 @@ func permutation(n int, list []*TCPPacket) []*TCPPacket {
// Response comes before Request
func TestRawListenerChunkedWrongOrder(t *testing.T) {
listener := NewListener("", "0", EnginePcap, 10*time.Millisecond)
listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond)
defer listener.Close()
reqPacket1 := buildPacket(true, 1, 1, []byte("POST / HTTP/1.1\r\nTransfer-Encoding: chunked\r\nExpect: 100-continue\r\n\r\n"))
@@ -381,7 +403,7 @@ func getMessage() []*TCPPacket {
// Response comes before Request
func TestRawListenerBench(t *testing.T) {
l := NewListener("", "0", EnginePcap, 200*time.Millisecond)
l := NewListener("", "0", EnginePcap, true, 200*time.Millisecond)
defer l.Close()
// Should re-construct message from all possible combinations
raw_socket_listener/tcp_message.go
+4 -4
View File
@@ -206,10 +206,10 @@ func (t *TCPMessage) UpdateResponseAck() uint32 {
t.ResponseAck = lastPacket.Seq + uint32(len(lastPacket.Data))
// We swappwed src and dst port
copy(t.ResponseID[:4], lastPacket.Addr)
copy(t.ResponseID[4:], lastPacket.Raw[2:4]) // Src port
copy(t.ResponseID[6:], lastPacket.Raw[0:2]) // Dest port
binary.BigEndian.PutUint32(t.ResponseID[8:12], t.ResponseAck)
copy(t.ResponseID[:16], lastPacket.Addr)
copy(t.ResponseID[16:], lastPacket.Raw[2:4]) // Src port
copy(t.ResponseID[18:], lastPacket.Raw[0:2]) // Dest port
binary.BigEndian.PutUint32(t.ResponseID[20:24], t.ResponseAck)
}
return t.ResponseAck
settings.go
+5 -2
View File
@@ -41,8 +41,9 @@ type AppSettings struct {
inputFile MultiOption
outputFile MultiOption
inputRAW MultiOption
inputRAWEngine string
inputRAW MultiOption
inputRAWEngine string
inputRAWTrackResponse bool
middleware string
@@ -83,6 +84,8 @@ func init() {
flag.Var(&Settings.inputRAW, "input-raw", "Capture traffic from given port (use RAW sockets and require *sudo* access):\n\t# Capture traffic from 8080 port\n\tgor --input-raw :8080 --output-http staging.com")
flag.BoolVar(&Settings.inputRAWTrackResponse, "input-raw-track-response", false, "If turned on Gor will track responses in addition to requests, and they will be available to middleware and file output.")
flag.StringVar(&Settings.inputRAWEngine, "input-raw-engine", "libpcap", "Intercept traffic using `libpcap` (default), and `raw_socket`")
flag.StringVar(&Settings.middleware, "middleware", "", "Used for modifying traffic using external command")