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

Update raw listener

Browse Source
This commit is contained in:
Leonid Bugaev
2015-07-13 16:19:24 +05:00
parent 32081b2d17
commit 0859642b38
3 changed files with 95 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
raw_socket_listener/listener.go
+55 -42
View File
@@ -1,4 +1,16 @@
package raw_socket
/*
Package rawSocket provides traffic sniffier using RAW sockets.
Capture traffic from socket using RAW_SOCKET's
http://en.wikipedia.org/wiki/Raw_socket
RAW_SOCKET allow you listen for traffic on any port (e.g. sniffing) because they operate on IP level.
Ports is TCP feature, same as flow control, reliable transmission and etc.
This package implements own TCP layer: TCP packets is parsed using tcp_packet.go, and flow control is managed by tcp_message.go
*/
package rawSocket
import (
"bytes"
@@ -8,41 +20,41 @@ import (
"strconv"
)
// Capture traffic from socket using RAW_SOCKET's
// http://en.wikipedia.org/wiki/Raw_socket
//
// RAW_SOCKET allow you listen for traffic on any port (e.g. sniffing) because they operate on IP level.
// Ports is TCP feature, same as flow control, reliable transmission and etc.
// Since we can't use default TCP libraries RAWTCPLitener implements own TCP layer
// TCP packets is parsed using tcp_packet.go, and flow control is managed by tcp_message.go
// Listener handle traffic capture
type Listener struct {
messages map[string]*TCPMessage // buffer of TCPMessages waiting to be send
// buffer of TCPMessages waiting to be send
messages map[string]*TCPMessage
// Expect: 100-continue request is send in 2 tcp messages
// We store ACK aliases to merge this packets together
ack_aliases map[uint32]uint32
seq_with_data map[uint32]uint32
ackAliases map[uint32]uint32
// To get ACK of second message we need to compute its Seq and wait for them message
seqWithData map[uint32]uint32
c_packets chan *TCPPacket
c_messages chan *TCPMessage // Messages ready to be send to client
// Messages ready to be send to client
packetsChan chan *TCPPacket
c_del_message chan *TCPMessage // Used for notifications about completed or expired messages
// Messages ready to be send to client
messagesChan chan *TCPMessage
// Used for notifications about completed or expired messages
messageDelChan chan *TCPMessage
addr string // IP to listen
port int // Port to listen
}
// RAWTCPListen creates a listener to capture traffic from RAW_SOCKET
// NewListener creates and initializes new Listener object
func NewListener(addr string, port string) (rawListener *Listener) {
rawListener = &Listener{}
rawListener.c_packets = make(chan *TCPPacket, 10000)
rawListener.c_messages = make(chan *TCPMessage, 10000)
rawListener.c_del_message = make(chan *TCPMessage, 10000)
rawListener.packetsChan = make(chan *TCPPacket, 10000)
rawListener.messagesChan = make(chan *TCPMessage, 10000)
rawListener.messageDelChan = make(chan *TCPMessage, 10000)
rawListener.messages = make(map[string]*TCPMessage)
rawListener.ack_aliases = make(map[uint32]uint32)
rawListener.seq_with_data = make(map[uint32]uint32)
rawListener.ackAliases = make(map[uint32]uint32)
rawListener.seqWithData = make(map[uint32]uint32)
rawListener.addr = addr
rawListener.port, _ = strconv.Atoi(port)
@@ -57,13 +69,14 @@ func (t *Listener) listen() {
for {
select {
// If message ready for deletion it means that its also complete or expired by timeout
case message := <-t.c_del_message:
t.c_messages <- message
delete(t.ack_aliases, message.Ack)
case message := <-t.messageDelChan:
log.Println("MESSAGE")
t.messagesChan <- message
delete(t.ackAliases, message.Ack)
delete(t.messages, message.ID)
// We need to use channels to process each packet to avoid data races
case packet := <-t.c_packets:
case packet := <-t.packetsChan:
t.processTCPPacket(packet)
}
}
@@ -94,18 +107,20 @@ func (t *Listener) readRAWSocket() {
}
func (t *Listener) parsePacket(addr net.Addr, buf []byte) {
log.Println("PACKET")
if t.isIncomingDataPacket(buf) {
t.c_packets <- ParseTCPPacket(addr, buf)
t.packetsChan <- ParseTCPPacket(addr, buf)
}
}
func (t *Listener) isIncomingDataPacket(buf []byte) bool {
// To avoid full packet parsing every time, we manually parsing values needed for packet filtering
// http://en.wikipedia.org/wiki/Transmission_Control_Protocol
dest_port := binary.BigEndian.Uint16(buf[2:4])
destPort := binary.BigEndian.Uint16(buf[2:4])
// Because RAW_SOCKET can't be bound to port, we have to control it by ourself
if int(dest_port) == t.port {
if int(destPort) == t.port {
// Get the 'data offset' (size of the TCP header in 32-bit words)
dataOffset := (buf[12] & 0xF0) >> 4
@@ -131,29 +146,27 @@ func (t *Listener) processTCPPacket(packet *TCPPacket) {
var message *TCPMessage
parent_message_ack, parent_ok := t.seq_with_data[packet.Seq]
if parent_ok {
t.ack_aliases[packet.Ack] = parent_message_ack
delete(t.seq_with_data, packet.Seq)
if parentAck, ok := t.seqWithData[packet.Seq]; ok {
t.ackAliases[packet.Ack] = parentAck
delete(t.seqWithData, packet.Seq)
}
ack_alias, alias_ok := t.ack_aliases[packet.Ack]
if alias_ok {
packet.Ack = ack_alias
if alias, ok := t.ackAliases[packet.Ack]; ok {
packet.Ack = alias
}
m_id := packet.Addr.String() + strconv.Itoa(int(packet.SrcPort)) + strconv.Itoa(int(packet.Ack))
message, ok := t.messages[m_id]
mID := packet.Addr.String() + strconv.Itoa(int(packet.SrcPort)) + strconv.Itoa(int(packet.Ack))
message, ok := t.messages[mID]
if !ok {
// We sending c_del_message channel, so message object can communicate with Listener and notify it if message completed
message = NewTCPMessage(m_id, t.c_del_message, packet.Ack)
t.messages[m_id] = message
// We sending messageDelChan channel, so message object can communicate with Listener and notify it if message completed
message = NewTCPMessage(mID, t.messageDelChan, packet.Ack)
t.messages[mID] = message
}
if bytes.Equal(packet.Data[0:4], bPOST) {
if bytes.Equal(packet.Data[len(packet.Data)-24:len(packet.Data)-4], bExpect100ContinueCheck) {
t.seq_with_data[packet.Seq+uint32(len(packet.Data))] = packet.Ack
t.seqWithData[packet.Seq+uint32(len(packet.Data))] = packet.Ack
// Removing `Expect: 100-continue` header
packet.Data = append(packet.Data[:len(packet.Data)-24], packet.Data[len(packet.Data)-2:]...)
@@ -161,10 +174,10 @@ func (t *Listener) processTCPPacket(packet *TCPPacket) {
}
// Adding packet to message
message.c_packets <- packet
message.packetsChan <- packet
}
// Receive TCP messages from the listener channel
func (t *Listener) Receive() *TCPMessage {
return <-t.c_messages
return <-t.messagesChan
}
raw_socket_listener/tcp_message.go
+15 -14
View File
@@ -1,4 +1,4 @@
package raw_socket
package rawSocket
import (
"log"
@@ -6,7 +6,8 @@ import (
"time"
)
const MSG_EXPIRE = 2000 * time.Millisecond
// MsgExpire specify period that message should wait before it considered as finished
const MsgExpire = 2000 * time.Millisecond
// 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
@@ -21,20 +22,20 @@ type TCPMessage struct {
timer *time.Timer // Used for expire check
c_packets chan *TCPPacket
packetsChan chan *TCPPacket
c_del_message chan *TCPMessage
delChan chan *TCPMessage
}
// NewTCPMessage pointer created from a Acknowledgment number and a channel of messages readuy to be deleted
func NewTCPMessage(ID string, c_del chan *TCPMessage, Ack uint32) (msg *TCPMessage) {
func NewTCPMessage(ID string, delChan chan *TCPMessage, Ack uint32) (msg *TCPMessage) {
msg = &TCPMessage{ID: ID, Ack: Ack}
msg.c_packets = make(chan *TCPPacket)
msg.c_del_message = c_del // used for notifying that message completed or expired
msg.packetsChan = make(chan *TCPPacket)
msg.delChan = delChan // used for notifying that message completed or expired
// Every time we receive packet we reset this timer
msg.timer = time.AfterFunc(MSG_EXPIRE, msg.Timeout)
msg.timer = time.AfterFunc(MsgExpire, msg.Timeout)
go msg.listen()
@@ -44,7 +45,7 @@ func NewTCPMessage(ID string, c_del chan *TCPMessage, Ack uint32) (msg *TCPMessa
func (t *TCPMessage) listen() {
for {
select {
case packet, more := <-t.c_packets:
case packet, more := <-t.packetsChan:
if more {
t.AddPacket(packet)
} else {
@@ -60,21 +61,21 @@ func (t *TCPMessage) Timeout() {
select {
// In some cases Timeout can be called multiple times (do not know how yet)
// Ensure that we did not close channel 2 times
case packet, ok := <-t.c_packets:
case packet, ok := <-t.packetsChan:
if ok {
t.AddPacket(packet)
} else {
return
}
default:
close(t.c_packets)
t.c_del_message <- t // Notify RAWListener that message is ready to be send to replay server
close(t.packetsChan)
t.delChan <- t // Notify RAWListener that message is ready to be send to replay server
}
}
// Bytes sorts packets in right orders and return message content
func (t *TCPMessage) Bytes() (output []byte) {
sort.Sort(BySeq(t.packets))
sort.Sort(sortBySeq(t.packets))
for _, v := range t.packets {
output = append(output, v.Data...)
@@ -102,5 +103,5 @@ func (t *TCPMessage) AddPacket(packet *TCPPacket) {
}
// Reset message timeout timer
t.timer.Reset(MSG_EXPIRE)
t.timer.Reset(MsgExpire)
}
raw_socket_listener/tcp_packet.go
+25 -25
View File
@@ -1,4 +1,4 @@
package raw_socket
package rawSocket
import (
"encoding/binary"
@@ -9,19 +9,18 @@ import (
// TCP Flags
const (
TCP_FIN = 1 << iota
TCP_SYN
TCP_RST
TCP_PSH
TCP_ACK
TCP_URG
TCP_ECE
TCP_CWR
TCP_NS
fFIN = 1 << iota
fSYN
fRST
fPSH
fACK
fURG
fECE
fCWR
fNS
)
// Simple TCP packet parser
//
// TCPPacket provides tcp packet parser
// Packet structure: http://en.wikipedia.org/wiki/Transmission_Control_Protocol
type TCPPacket struct {
SrcPort uint16
@@ -39,6 +38,7 @@ type TCPPacket struct {
Addr net.Addr
}
// ParseTCPPacket takes address and tcp payload and returns parsed TCPPacket
func ParseTCPPacket(addr net.Addr, b []byte) (p *TCPPacket) {
p = &TCPPacket{Data: b}
p.ParseBasic()
@@ -76,15 +76,15 @@ func (t *TCPPacket) String() string {
"Acknowledgment:" + strconv.Itoa(int(t.Ack)),
"Header len:" + strconv.Itoa(int(t.DataOffset)),
"Flag ns:" + strconv.FormatBool(t.Flags&TCP_NS != 0),
"Flag crw:" + strconv.FormatBool(t.Flags&TCP_CWR != 0),
"Flag ece:" + strconv.FormatBool(t.Flags&TCP_ECE != 0),
"Flag urg:" + strconv.FormatBool(t.Flags&TCP_URG != 0),
"Flag ack:" + strconv.FormatBool(t.Flags&TCP_ACK != 0),
"Flag psh:" + strconv.FormatBool(t.Flags&TCP_PSH != 0),
"Flag rst:" + strconv.FormatBool(t.Flags&TCP_RST != 0),
"Flag syn:" + strconv.FormatBool(t.Flags&TCP_SYN != 0),
"Flag fin:" + strconv.FormatBool(t.Flags&TCP_FIN != 0),
"Flag ns:" + strconv.FormatBool(t.Flags&fNS != 0),
"Flag crw:" + strconv.FormatBool(t.Flags&fCWR != 0),
"Flag ece:" + strconv.FormatBool(t.Flags&fECE != 0),
"Flag urg:" + strconv.FormatBool(t.Flags&fURG != 0),
"Flag ack:" + strconv.FormatBool(t.Flags&fACK != 0),
"Flag psh:" + strconv.FormatBool(t.Flags&fPSH != 0),
"Flag rst:" + strconv.FormatBool(t.Flags&fRST != 0),
"Flag syn:" + strconv.FormatBool(t.Flags&fSYN != 0),
"Flag fin:" + strconv.FormatBool(t.Flags&fFIN != 0),
"Window size:" + strconv.Itoa(int(t.Window)),
"Checksum:" + strconv.Itoa(int(t.Checksum)),
@@ -94,8 +94,8 @@ func (t *TCPPacket) String() string {
}, "\n")
}
type BySeq []*TCPPacket
type sortBySeq []*TCPPacket
func (a BySeq) Len() int { return len(a) }
func (a BySeq) Swap(i, j int) { a[i], a[j] = a[j], a[i] }
func (a BySeq) Less(i, j int) bool { return a[i].Seq < a[j].Seq }
func (a sortBySeq) Len() int { return len(a) }
func (a sortBySeq) Swap(i, j int) { a[i], a[j] = a[j], a[i] }
func (a sortBySeq) Less(i, j int) bool { return a[i].Seq < a[j].Seq }