Big rewrite

* Use RAW_SOCKETS instead of tcpdump
* Own TCP stack
* All HTTP request types support
* Use standard http.ParseRequest for parsing requests

gor.go

@@ -14,7 +14,7 @@ import (
 )
 
 const (
-	VERSION = "0.1.1"
+	VERSION = "0.3"
 )
 
 func main() {

listener/listener.go

@@ -1,4 +1,4 @@
-// Listener capture TCP traffic right from given port using `tcpdump` utility.
+// Listener capture TCP traffic using RAW SOCKETS.
 // Note: it requires sudo or root access.
 //
 // Rigt now it suport only HTTP, and only GET requests.
@@ -7,16 +7,11 @@ package listener
 import (
 	"bufio"
 	"bytes"
-	"encoding/gob"
-	"flag"
 	"fmt"
-	"io"
 	"log"
 	"net"
+	"net/http"
 	"os"
-	"os/exec"
-	"regexp"
-	"strings"
 )
 
 type HttpRequest struct {
@@ -28,106 +23,15 @@ type HttpRequest struct {
 
 // Enable debug logging only if "--verbose" flag passed
 func Debug(v ...interface{}) {
-	if Settings.verbose { log.Println(v...) }
-}
-
-// Parse `tcpdump` output to find HTTP GET requests
-// When HttpRequest found it get send to `requests` channel
-func parseRequest(pipe io.ReadCloser, requests chan *HttpRequest) {
-	request_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()
-
-		// HTTP/1.1 match finds both requests and response
-		// Index is used instead of Regexp just for speed
-		if strings.Index(line, "HTTP/1.1") != -1 {
-			// Allow only requests
-			match := request_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 {
-			// We assume that empty line is end of request info
-			// This is true only for GET requests
-			if line == "" {
-				requests <- request
-				requestStarted = false
-			} else {
-				// All headers comes in this format:
-				//
-				//     User-Agent: Mozilla
-				//     Content-Type: text/html
-				//
-				match := headers_re.FindAllString(line, -1)
-
-				if len(match) > 0 {
-					header := strings.Split(match[0], ": ")
-
-					request.Headers[header[0]] = header[1]
-				}
-			}
-		}
+	if Settings.verbose {
+		log.Println(v...)
 	}
 }
 
-// Sends request to replay server via UDP
-// Before sending it encode request object using standard gob encoder
-func forwardRequest(requests chan *HttpRequest) {
-	serverAddr, err := net.ResolveUDPAddr("udp4", Settings.ReplayServer())
-	conn, err := net.DialUDP("udp", nil, serverAddr)
+func greeting() {
 
-	defer conn.Close()
-
-	if err != nil {
-		log.Fatal("Connection error", err)
-	}
-
-	for {
-		select {
-		case request := <-requests:
-			Debug("Forwarding:", request.Url, "to", Settings.ReplayServer())
-
-			msg := bytes.Buffer{}
-
-			enc := gob.NewEncoder(&msg)
-			err := enc.Encode(request)
-
-			conn.Write(msg.Bytes())
-
-			if err != nil {
-				log.Println("encode error:", err)
-			}
-		}
-	}
 }
 
-
-func greeting() {	
-	fmt.Println("Listening for HTTP traffic on", Settings.port, "port")
-	fmt.Println("Running: tcpdump "+strings.Join(Settings.TCPDumpConfig()," "))
-	fmt.Println("Forwarding requests to replay server:", Settings.ReplayServer())
-}
-
-
 // Because its sub-program, Run acts as `main`
 func Run() {
 	if os.Getuid() != 0 {
@@ -135,25 +39,42 @@ func Run() {
 		fmt.Println("This is required since listener sniff traffic on given port.")
 		os.Exit(1)
 	}
-	
-	// TODO: use RAW_SOCKETS instead of tcpdump
-	cmd := exec.Command("tcpdump", Settings.TCPDumpConfig()...)
 
-	greeting()
+	fmt.Println("Listening for HTTP traffic on", Settings.port, "port")
+	fmt.Println("Forwarding requests to replay server:", Settings.ReplayServer())
 
-	stdout, _ := cmd.StdoutPipe()
-	cmd.Stderr = os.Stderr
+	// Connection to reaplay server
+	serverAddr, err := net.ResolveUDPAddr("udp4", Settings.ReplayServer())
+	conn, err := net.DialUDP("udp", nil, serverAddr)
 
-	if err := cmd.Start(); err != nil {
-		log.Fatal(err)
+	if err != nil {
+		log.Fatal("Connection error", err)
 	}
 
-	requests := make(chan *HttpRequest)
+	// Sniffing traffic from given port
+	listener := RAWTCPListen("0.0.0.0", Settings.port)
 
-	go parseRequest(stdout, requests)
-	go forwardRequest(requests)
+	for {
+		message := listener.Receive()
 
-	if err := cmd.Wait(); err != nil {
-		flag.Usage()
+		go func(m *TCPMessage) {
+			if Settings.verbose {
+				buf := bytes.NewBuffer(m.Bytes())
+				reader := bufio.NewReader(buf)
+
+				request, err := http.ReadRequest(reader)
+
+				if err != nil {
+					Debug("Error while parsing request:", string(m.Bytes()))
+				} else {
+					request.ParseMultipartForm(32 << 20)
+					Debug("Forwarding request:", request)
+				}
+			}
+
+			conn.Write(m.Bytes())
+		}(message)
 	}
+
+	conn.Close()
 }

listener/raw_tcp_listener.go

@@ -0,0 +1,105 @@
+package listener
+
+import (
+	"log"
+	"net"
+	"time"
+)
+
+type RAWTCPListener struct {
+	messages map[uint32]*TCPMessage // buffer of TCPMessages waiting to be send
+
+	c_packets  chan *TCPPacket
+	c_messages chan *TCPMessage
+
+	addr string
+	port int
+}
+
+func RAWTCPListen(addr string, port int) (listener *RAWTCPListener) {
+	listener = &RAWTCPListener{}
+	listener.messages = make(map[uint32]*TCPMessage)
+
+	listener.c_packets = make(chan *TCPPacket)
+	listener.c_messages = make(chan *TCPMessage)
+
+	listener.addr = addr
+	listener.port = port
+
+	go listener.listen()
+	go listener.readTCPPackets()
+
+	return
+}
+
+func (t *RAWTCPListener) listen() {
+
+	for {
+		var messages chan *TCPMessage
+		var message *TCPMessage
+
+		for _, msg := range t.messages {
+			if msg.Complete() {
+				messages = t.c_messages
+				message = msg
+			}
+		}
+
+		select {
+		case messages <- message:
+			delete(t.messages, message.ask)
+		case packet := <-t.c_packets:
+			t.processTCPPacket(packet)
+
+		// Ensure that this will be run at least each 200 ms, to ensure that all messages will be send
+		// Without it last message may not be send (it will be send only on next TCP packets)
+		case <-time.After(200 * time.Millisecond):
+		}
+	}
+}
+
+func (t *RAWTCPListener) readTCPPackets() {
+	conn, e := net.ListenPacket("ip4:tcp", t.addr)
+	defer conn.Close()
+
+	if e != nil {
+		log.Fatal(e)
+	}
+
+	for {
+		buf := make([]byte, 65565)
+		n, _, err := conn.ReadFrom(buf)
+
+		if err != nil {
+			Debug("Error:", err)
+		}
+
+		if n > 0 {
+			packet := NewTCPPacket(buf[:n])
+
+			if int(packet.dest_port) == t.port {
+				t.c_packets <- packet
+			}
+		}
+	}
+}
+
+//
+func (t *RAWTCPListener) processTCPPacket(packet *TCPPacket) {
+	// We interested only in packets that contain some data
+	if !(packet.f_ask && packet.f_psh) {
+		return
+	}
+
+	ask := packet.asknowledgement
+
+	if _, ok := t.messages[ask]; !ok {
+		t.messages[ask] = NewTCPMessage(ask)
+	}
+
+	t.messages[ask].AddPacket(packet)
+}
+
+func (t *RAWTCPListener) Receive() *TCPMessage {
+	return <-t.c_messages
+}

listener/settings.go

@@ -1,54 +1,47 @@
 package listener
 
 import (
-    "strconv"
-    "strings"
-    "flag"
-    "os"
+	"flag"
+	"os"
+	"strings"
 )
 
 const (
-    defaultPort             = 80
-    defaultNetworkInterface = "any"
+	defaultPort             = 80
+	defaultNetworkInterface = "any"
 
-    defaultReplayAddress = "localhost:28020"
+	defaultReplayAddress = "localhost:28020"
 )
 
 type ListenerSettings struct {
-    networkInterface string
-    port             int
+	networkInterface string
+	port             int
 
-    replayAddress string
+	replayAddress string
 
-    verbose bool
+	verbose bool
 }
 
 var Settings ListenerSettings = ListenerSettings{}
 
 func (s *ListenerSettings) ReplayServer() string {
-    if !strings.Contains(s.replayAddress, ":") {
-        return s.replayAddress + ":28020"
-    }
+	if !strings.Contains(s.replayAddress, ":") {
+		return s.replayAddress + ":28020"
+	}
 
-    return s.replayAddress
+	return s.replayAddress
 }
 
-// tcpdump -vv -A -i all port 8080
-func (s *ListenerSettings) TCPDumpConfig() []string {
-    return []string{"-vv", "-A", "-i", Settings.networkInterface, "port "+strconv.Itoa(Settings.port)}
-}
-
-
 func init() {
-    if len(os.Args) < 2 || os.Args[1] != "listen" {
-        return
-    }
+	if len(os.Args) < 2 || os.Args[1] != "listen" {
+		return
+	}
 
-    flag.IntVar(&Settings.port, "p", defaultPort, "Specify the http server port whose traffic you want to capture")
+	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.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.")
+	flag.StringVar(&Settings.replayAddress, "r", defaultReplayAddress, "Address of replay server.")
 
-    flag.BoolVar(&Settings.verbose, "verbose", false, "Log requests")
+	flag.BoolVar(&Settings.verbose, "verbose", false, "Log requests")
 }

listener/tcp_message.go

@@ -0,0 +1,64 @@
+package listener
+
+import (
+	"sort"
+	"time"
+)
+
+// TCPMessage ensure that all TCP packets for given request is received, and processed in right sequence
+// Its needed because all TCP message can be fragmented or re-transmitted
+//
+// Each TCP Packet have 2 ids: asknowledgement - message_id, and sequence - packet_id
+// Message can be compiled from unique packets with same message_id which sorted by sequence
+// Message is received if we did't receive any packets for 200ms OR if we received packet with "fin" flag
+type TCPMessage struct {
+	ask     uint32             // Message ID
+	packets map[int]*TCPPacket // map[packet.sequence]*TCPPacket
+	updated int64              // time of last packet
+}
+
+func NewTCPMessage(ask uint32) (msg *TCPMessage) {
+	msg = &TCPMessage{}
+	msg.packets = make(map[int]*TCPPacket)
+	msg.updated = time.Now().UnixNano()
+	msg.ask = ask
+	return
+}
+
+// Sort packets in right orders and return message content
+func (t *TCPMessage) Bytes() (output []byte) {
+	mk := make([]int, len(t.packets))
+
+	i := 0
+	for k, _ := range t.packets {
+		mk[i] = k
+		i++
+	}
+
+	sort.Ints(mk)
+
+	for _, k := range mk {
+		output = append(output, t.packets[k].data...)
+	}
+
+	return
+}
+
+// Add packet to the message
+func (t *TCPMessage) AddPacket(packet *TCPPacket) {
+	seq := int(packet.sequence)
+
+	if _, ok := t.packets[seq]; !ok {
+		t.packets[seq] = packet
+	} else {
+		Debug("Received packet with same sequence")
+	}
+
+	t.updated = time.Now().UnixNano()
+}
+
+// TCP message is complete if we not received any packets for 200ms since last packet
+func (t *TCPMessage) Complete() bool {
+	ns := time.Now().UnixNano()
+	return (ns - t.updated) > int64(200*time.Millisecond)
+}

listener/tcp_packet.go

@@ -0,0 +1,106 @@
+package listener
+
+import (
+	"bytes"
+	"encoding/binary"
+	"strconv"
+	"strings"
+)
+
+// Simple TCP packet parser
+//
+// Packet structure: http://en.wikipedia.org/wiki/Transmission_Control_Protocol
+type TCPPacket struct {
+	packet []byte
+
+	buf *bytes.Buffer
+
+	source_port uint16
+	dest_port   uint16
+
+	sequence        uint32
+	asknowledgement uint32
+
+	doff_reserved uint16
+	tcph_length   uint16
+
+	f_ns  bool
+	f_crw bool
+	f_ece bool
+	f_urg bool
+	f_ask bool
+	f_psh bool
+	f_rst bool
+	f_syn bool
+	f_fin bool
+
+	window_size uint16
+	checksum    uint16
+
+	data []byte
+}
+
+func NewTCPPacket(b []byte) (p *TCPPacket) {
+	buf := bytes.NewBuffer(b)
+	p = &TCPPacket{packet: b, buf: buf}
+	p.Parse()
+
+	return p
+}
+
+// Helper for binary.Read
+func (t *TCPPacket) read(data interface{}) {
+	binary.Read(t.buf, binary.BigEndian, data)
+}
+
+// Inspired by: https://gist.github.com/clicube/4978853
+func (t *TCPPacket) Parse() {
+	t.read(&t.source_port)
+	t.read(&t.dest_port)
+
+	t.read(&t.sequence)
+	t.read(&t.asknowledgement)
+	t.read(&t.doff_reserved)
+
+	t.tcph_length = t.doff_reserved >> 12 * 4
+
+	t.f_ns = (t.doff_reserved & 256) != 0
+	t.f_crw = (t.doff_reserved & 128) != 0
+	t.f_ece = (t.doff_reserved & 64) != 0
+	t.f_urg = (t.doff_reserved & 32) != 0
+	t.f_ask = (t.doff_reserved & 16) != 0
+	t.f_psh = (t.doff_reserved & 8) != 0
+	t.f_rst = (t.doff_reserved & 4) != 0
+	t.f_syn = (t.doff_reserved & 2) != 0
+	t.f_fin = (t.doff_reserved & 1) != 0
+
+	t.read(&t.window_size)
+	t.read(&t.checksum)
+
+	t.data = t.packet[t.tcph_length:]
+}
+
+func (t *TCPPacket) String() string {
+	return strings.Join([]string{
+		"Source port: " + strconv.Itoa(int(t.source_port)),
+		"Dest port:" + strconv.Itoa(int(t.dest_port)),
+		"Sequence:" + strconv.Itoa(int(t.sequence)),
+		"Acknowledgement:" + strconv.Itoa(int(t.asknowledgement)),
+		"Header len:" + strconv.Itoa(int(t.tcph_length)),
+
+		"Flag ns:" + strconv.FormatBool(t.f_ns),
+		"Flag crw:" + strconv.FormatBool(t.f_crw),
+		"Flag ece:" + strconv.FormatBool(t.f_ece),
+		"Flag urg:" + strconv.FormatBool(t.f_urg),
+		"Flag ask:" + strconv.FormatBool(t.f_ask),
+		"Flag psh:" + strconv.FormatBool(t.f_psh),
+		"Flag rst:" + strconv.FormatBool(t.f_rst),
+		"Flag syn:" + strconv.FormatBool(t.f_syn),
+		"Flag fin:" + strconv.FormatBool(t.f_fin),
+
+		"Window size:" + strconv.Itoa(int(t.window_size)),
+		"Checksum:" + strconv.Itoa(int(t.checksum)),
+
+		"Data:" + string(t.data),
+	}, "\n")
+}

replay/replay.go

@@ -25,29 +25,27 @@
 package replay
 
 import (
+	"bufio"
 	"bytes"
-	"encoding/gob"
 	"log"
 	"net"
+	"net/http"
 )
 
 const bufSize = 1024 * 10
 
 // Enable debug logging only if "--verbose" flag passed
 func Debug(v ...interface{}) {
-	if Settings.verbose { log.Println(v...) }
+	if Settings.verbose {
+		log.Println(v...)
+	}
 }
 
-// Decode HttpRequest object using standard gob decoder
-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)
+func ParseRequest(data []byte) (request *http.Request, err error) {
+	buf := bytes.NewBuffer(data)
+	reader := bufio.NewReader(buf)
 
+	request, err = http.ReadRequest(reader)
 	return
 }
 
@@ -78,25 +76,23 @@ func Run() {
 	requestFactory := NewRequestFactory()
 
 	for {
-		rlen, _, err := conn.ReadFromUDP(buf[0:])
+		n, _, err := conn.ReadFromUDP(buf[0:])
 
 		if err != nil {
 			continue
 		}
 
-		if rlen > 0 {
-			if rlen > bufSize {
-				log.Fatal("Too large udp packet", bufSize)
+		if n > 0 {
+			if n > bufSize {
+				Debug("Too large udp packet", bufSize)
 			}
 
-			request, err := DecodeRequest(buf[0:rlen])
-
-			if err != nil {
-				log.Println("Decode error:", err)
+			if request, err := ParseRequest(buf[0:n]); err != nil {
+				Debug("Error while parsing request", err, buf[0:n])
 			} else {
 				requestFactory.Add(request)
 			}
 		}
 	}
 
-}
+}

replay/request_factory.go

@@ -2,19 +2,12 @@ package replay
 
 import (
 	"net/http"
+	"net/url"
 )
 
-// Userd for transfering Request info between Listener and Replay server
-type HttpRequest struct {
-	Tag     string            // Not used yet
-	Method  string            // Right now only 'GET'
-	Url     string            // Request URL
-	Headers map[string]string // Request Headers
-}
-
 type HttpResponse struct {
 	host *ForwardHost
-	req  *HttpRequest
+	req  *http.Request
 	resp *http.Response
 	err  error
 }
@@ -29,7 +22,7 @@ type HttpResponse struct {
 // 4. handleRequest() listen for `response` channel and updates stats
 type RequestFactory struct {
 	responses chan *HttpResponse
-	requests  chan *HttpRequest
+	requests  chan *http.Request
 }
 
 // RequestFactory contstuctor
@@ -37,7 +30,7 @@ type RequestFactory struct {
 func NewRequestFactory() (factory *RequestFactory) {
 	factory = &RequestFactory{}
 	factory.responses = make(chan *HttpResponse)
-	factory.requests = make(chan *HttpRequest)
+	factory.requests = make(chan *http.Request)
 
 	go factory.handleRequests()
 
@@ -45,22 +38,24 @@ func NewRequestFactory() (factory *RequestFactory) {
 }
 
 // Forward http request to given host
-func (f *RequestFactory) sendRequest(host *ForwardHost, request *HttpRequest) {
-	var req *http.Request
-
+func (f *RequestFactory) sendRequest(host *ForwardHost, request *http.Request) {
 	client := &http.Client{}
 
-	req, err := http.NewRequest("GET", host.Url+request.Url, nil)
+	URL := host.Url + request.URL.String()
 
-	// Forwarded request should have same headers
-	for key, value := range request.Headers {
-		req.Header.Add(key, value)
+	request.RequestURI = ""
+	request.URL, _ = url.ParseRequestURI(URL)
+
+	if Settings.verbose {
+		Debug("Sending request:", request)
 	}
 
-	resp, err := client.Do(req)
+	resp, err := client.Do(request)
 
 	if err == nil {
 		defer resp.Body.Close()
+	} else {
+		Debug("Request error:", err)
 	}
 
 	f.responses <- &HttpResponse{host, request, resp, err}
@@ -81,8 +76,6 @@ func (f *RequestFactory) handleRequests() {
 					// Increment Stat.Count
 					host.Stat.IncReq()
 
-                    Debug("GET ",host.Url + req.Url)
-
 					go f.sendRequest(host, req)
 				}
 			}
@@ -94,6 +87,6 @@ func (f *RequestFactory) handleRequests() {
 }
 
 // Add request to channel for further processing
-func (f *RequestFactory) Add(request *HttpRequest) {
+func (f *RequestFactory) Add(request *http.Request) {
 	f.requests <- request
 }