Skip to content

Commit

Permalink
Support split payloads
Browse files Browse the repository at this point in the history 
The offset was not managed how it should have been:
* TCP decode Offset is now managed in the function decodeDnsData()
* Add a unit test calling the Parse() method on a split query

Also:
* remove assignements and use pointers to alter DnsStream objects in
Parse()
  • Loading branch information
@McStork
McStork committed Dec 16, 2015
1 parent bef7e71 commit 7b8861b
Show file tree
Hide file tree
Showing 2 changed files with 83 additions and 40 deletions.
73 changes: 38 additions & 35 deletions packetbeat/protos/dns/dns.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,7 @@ const (
NonDnsPacketMsg = "Packet's data could not be decoded as DNS."
NonDnsCompleteMsg = "Message's data could not be decoded as DNS."
NonDnsResponsePacketMsg = "Response packet's data could not be decoded as DNS."
EmptyPacket = "Packet's data is null."
EmptyMsg = "Message's data is empty."
DuplicateQueryMsg = "Another query with the same DNS ID from this client " +
"was received so this query was closed without receiving a response."
OrphanedResponseMsg = "Response was received without an associated query."
Expand Down Expand Up @@ -323,7 +323,7 @@ func (dns *Dns) ParseUdp(pkt *protos.Packet) {
logp.Debug("dns", "Parsing packet addressed with %s of length %d.",
pkt.Tuple.String(), len(pkt.Payload))

dnsPkt, err := decodeDnsData(pkt.Payload)
dnsPkt, err := decodeDnsData(TransportUdp, pkt.Payload)
if err != nil {
// This means that malformed requests or responses are being sent or
// that someone is attempting to the DNS port for non-DNS traffic. Both
Expand Down Expand Up @@ -388,11 +388,6 @@ func (dns *Dns) receivedDnsResponse(tuple *DnsTuple, msg *DnsMessage) {
}

func (dns *Dns) publishTransaction(t *DnsTransaction) {
var offset int
if t.Transport == TransportTcp {
offset = DecodeOffset
}

if dns.results == nil {
return
}
Expand All @@ -416,8 +411,8 @@ func (dns *Dns) publishTransaction(t *DnsTransaction) {
event["dns"] = dnsEvent

if t.Request != nil && t.Response != nil {
event["bytes_in"] = t.Request.Length + offset
event["bytes_out"] = t.Response.Length + offset
event["bytes_in"] = t.Request.Length
event["bytes_out"] = t.Response.Length
event["responsetime"] = int32(t.Response.Ts.Sub(t.ts).Nanoseconds() / 1e6)
event["method"] = dnsOpCodeToString(t.Request.Data.OpCode)
if len(t.Request.Data.Questions) > 0 {
Expand All @@ -438,7 +433,7 @@ func (dns *Dns) publishTransaction(t *DnsTransaction) {
event["response"] = dnsToString(t.Response.Data)
}
} else if t.Request != nil {
event["bytes_in"] = t.Request.Length + offset
event["bytes_in"] = t.Request.Length
event["method"] = dnsOpCodeToString(t.Request.Data.OpCode)
if len(t.Request.Data.Questions) > 0 {
event["query"] = dnsQuestionToString(t.Request.Data.Questions[0])
Expand All @@ -451,7 +446,7 @@ func (dns *Dns) publishTransaction(t *DnsTransaction) {
event["request"] = dnsToString(t.Request.Data)
}
} else if t.Response != nil {
event["bytes_out"] = t.Response.Length + offset
event["bytes_out"] = t.Response.Length
event["method"] = dnsOpCodeToString(t.Response.Data.OpCode)
if len(t.Response.Data.Questions) > 0 {
event["query"] = dnsQuestionToString(t.Response.Data.Questions[0])
Expand Down Expand Up @@ -707,7 +702,12 @@ func nameToString(name []byte) string {
// decodeDnsData decodes a byte array into a DNS struct. If an error occurs
// then the returnd dns pointer will be nil. This method recovers from panics
// and is concurrency-safe.
func decodeDnsData(data []byte) (dns *layers.DNS, err error) {
func decodeDnsData(transport Transport, data []byte) (dns *layers.DNS, err error) {
var offset int
if transport == TransportTcp {
offset = DecodeOffset
}

// Recover from any panics that occur while parsing a packet.
defer func() {
if r := recover(); r != nil {
Expand All @@ -716,7 +716,7 @@ func decodeDnsData(data []byte) (dns *layers.DNS, err error) {
}()

d := &layers.DNS{}
err = d.DecodeFromBytes(data, gopacket.NilDecodeFeedback)
err = d.DecodeFromBytes(data[offset:], gopacket.NilDecodeFeedback)
if err != nil {
return nil, err
}
Expand All @@ -741,45 +741,47 @@ func (dns *Dns) Parse(pkt *protos.Packet, tcpTuple *common.TcpTuple, dir uint8,
}
}

var payload []byte
payload := pkt.Payload

// Offset is critical
if len(pkt.Payload) > DecodeOffset {
payload = pkt.Payload[DecodeOffset:]
} else {
logp.Debug("dns", EmptyPacket+" addresses %s",
tcpTuple.String())
return priv
}
stream := &priv.Data[dir]

stream := priv.Data[dir]
if stream == nil {
stream = &DnsStream{
if *stream == nil {
*stream = &DnsStream{
tcpTuple: tcpTuple,
data: payload,
message: &DnsMessage{Ts: pkt.Ts, Tuple: pkt.Tuple},
}
if len(payload) <= DecodeOffset {
logp.Debug("dns", EmptyMsg+" addresses %s",
tcpTuple.String())

return priv
}
} else {
stream.data = append(stream.data, payload...)
if len(stream.data) > tcp.TCP_MAX_DATA_IN_STREAM {
(*stream).data = append((*stream).data, payload...)
dataLength := len((*stream).data)
if dataLength > tcp.TCP_MAX_DATA_IN_STREAM {
logp.Debug("dns", "Stream data too large, dropping DNS stream")
stream = nil
return priv
}
if dataLength <= DecodeOffset {
logp.Debug("dns", EmptyMsg+" addresses %s",
tcpTuple.String())
return priv
}
}

priv.Data[dir] = stream
data, err := decodeDnsData(stream.data)
data, err := decodeDnsData(TransportTcp, (*stream).data)

if err != nil {
logp.Debug("dns", NonDnsCompleteMsg+" addresses %s, length %d",
tcpTuple.String(), len(stream.data))
tcpTuple.String(), len((*stream).data))

// wait for decoding with the next segment
return priv
}
dns.messageComplete(tcpTuple, dir, stream, data)

dns.messageComplete(tcpTuple, dir, *stream, data)
return priv
}

Expand Down Expand Up @@ -819,7 +821,8 @@ func (dns *Dns) ReceivedFin(tcpTuple *common.TcpTuple, dir uint8, private protos
}
stream := dnsData.Data[dir]
if stream.message != nil {
decodedData, err := decodeDnsData(stream.data)
decodedData, err := decodeDnsData(TransportTcp, stream.data)

if err == nil {
dns.messageComplete(tcpTuple, dir, stream, decodedData)
} else /*Failed decode */ {
Expand Down Expand Up @@ -848,7 +851,7 @@ func (dns *Dns) GapInStream(tcpTuple *common.TcpTuple, dir uint8, nbytes int, pr
return private, false
}

decodedData, err := decodeDnsData(stream.data)
decodedData, err := decodeDnsData(TransportTcp, stream.data)

// Add Notes if the failed stream is the response
if err != nil {
Expand Down Expand Up @@ -876,7 +879,7 @@ func (dns *Dns) publishDecodeFailureNotes(dnsData dnsPrivateData) {
return
}

dataOrigin, err := decodeDnsData(streamOrigin.data)
dataOrigin, err := decodeDnsData(TransportTcp, streamOrigin.data)
tupleReverse := streamReverse.message.Tuple

if err == nil {
Expand Down
50 changes: 45 additions & 5 deletions packetbeat/protos/dns/dns_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -916,19 +916,59 @@ func parseTcpRequestResponse(t testing.TB, dns *Dns, q DnsTestMessage) {
assertMapStrData(t, m, q)
}

// Verify that the split lone request packet is parsed.
func TestParseTcpSplitRequest(t *testing.T) {
stream := &DnsStream{data: sophosTxtTcp.request[2:10], message: new(DnsMessage)}
_, err := decodeDnsData(stream.data)
// Verify that the split lone request packet is decoded.
func TestDecodeTcpSplitRequest(t *testing.T) {
stream := &DnsStream{data: sophosTxtTcp.request[:10], message: new(DnsMessage)}
_, err := decodeDnsData(TransportTcp, stream.data)

assert.NotNil(t, err, "Not expecting a complete message yet")

stream.data = append(stream.data, sophosTxtTcp.request[10:]...)
_, err = decodeDnsData(stream.data)
_, err = decodeDnsData(TransportTcp, stream.data)

assert.Nil(t, err, "Message should be complete")
}

// Verify that the split lone request packet is parsed.
func TestParseTcpSplitResponse(t *testing.T) {
dns := newDns(testing.Verbose())
tcpQuery := elasticATcp

q := tcpQuery.request
r0 := tcpQuery.response[:10]
r1 := tcpQuery.response[10:]

tcptuple := testTcpTuple()
private := protos.ProtocolData(new(dnsPrivateData))

packet := newPacket(forward, q)
private = dns.Parse(packet, tcptuple, tcp.TcpDirectionOriginal, private)
assert.Equal(t, 1, dns.transactions.Size(), "There should be one transaction.")

packet = newPacket(reverse, r0)
private = dns.Parse(packet, tcptuple, tcp.TcpDirectionReverse, private)
assert.Equal(t, 1, dns.transactions.Size(), "There should be one transaction.")

packet = newPacket(reverse, r1)
private = dns.Parse(packet, tcptuple, tcp.TcpDirectionReverse, private)
assert.Empty(t, dns.transactions.Size(), "There should be no transaction.")

m := expectResult(t, dns)
assert.Equal(t, "tcp", mapValue(t, m, "transport"))
assert.Equal(t, len(tcpQuery.request), mapValue(t, m, "bytes_in"))
assert.Equal(t, len(tcpQuery.response), mapValue(t, m, "bytes_out"))
assert.NotNil(t, mapValue(t, m, "responsetime"))

if assert.ObjectsAreEqual("NOERROR", mapValue(t, m, "dns.response_code")) {
assert.Equal(t, common.OK_STATUS, mapValue(t, m, "status"))
} else {
assert.Equal(t, common.ERROR_STATUS, mapValue(t, m, "status"))
}

assert.Nil(t, mapValue(t, m, "notes"))
assertMapStrData(t, m, tcpQuery)
}

func TestGapRequestDrop(t *testing.T) {
dns := newDns(testing.Verbose())
q := sophosTxtTcp.request[:10]
Expand Down

0 comments on commit 7b8861b

Please sign in to comment.