[[new-protocol]]
= Developer Guide: Adding a New Protocol


[partintro]
--
The following topics describe how to add a new protocol to Packetbeat:

* <<getting-ready-new-protocol>>
* <<protocol-modules>>
* <<testing>>

--

[[getting-ready-new-protocol]]
== Getting Ready

Packetbeat is written in http://golang.org/[Go], so having Go installed and knowing the basics are prerequisites for understanding this guide. But don't worry if you aren't a Go expert. Go is a relatively new language, and very few people are experts in it. In fact, several people learned Go by contributing to Packetbeat and libbeat, including the original Packetbeat authors.

You will also need a good understanding of the wire protocol that you want to
add support for. For standard protocols or protocols used in open source
projects, you can usually find detailed specifications and example source code.
Wireshark is a very useful tool for understanding the inner workings of the
protocols it supports.

In some cases you can even make use of existing libraries for doing the actual
parsing and decoding of the protocol. If the particular protocol has a Go
implementation with a liberal enough license, you might be able to use it to
parse and decode individual messages instead of writing your own parser.

Before starting, please also read the
https://github.com/elastic/beats/blob/master/packetbeat/CONTRIBUTING.md[CONTRIBUTING]
file on GitHub.

[float]
=== Cloning and Compiling

After you have https://golang.org/doc/install[installed Go] and set up the
https://golang.org/doc/code.html#GOPATH[GOPATH] environment variable to point to
your preferred workspace location, you can clone Packetbeat with the
following commands:

[source,shell]
----------------------------------------------------------------------
$ mkdir -p $GOPATH/src/github.com/elastic
$ cd $GOPATH/src/github.com/elastic
$ git clone https://github.com/elastic/beats.git
----------------------------------------------------------------------

Then you can compile it with:

[source,shell]
----------------------------------------------------------------------
$ cd beats
$ make
----------------------------------------------------------------------

Note that the location where you clone is important. If you prefer working
outside of the `GOPATH` environment, you can clone to another directory and only
create a symlink to the `$GOPATH/src/github.com/elastic/` directory.

[float]
=== Forking and Branching

We recommend the following work flow for contributing to Packetbeat:

* Fork Beats in GitHub to your own account

* In the `$GOPATH/src/github.com/elastic/beats` folder, add your fork
  as a new remote. For example (replace `tsg` with your GitHub account):

[source,shell]
----------------------------------------------------------------------
$ git remote add tsg git@github.com:tsg/beats.git
----------------------------------------------------------------------

* Create a new branch for your work:

[source,shell]
----------------------------------------------------------------------
$ git checkout -b cool_new_protocol
----------------------------------------------------------------------

* Commit as often as you like, and then push to your private fork with:

[source,shell]
----------------------------------------------------------------------
$ git push --set-upstream tsg cool_new_protocol
----------------------------------------------------------------------

* When you are ready to submit your PR, simply do so from the GitHub web
  interface. Feel free to submit your PR early. You can still add commits to
  the branch after creating the PR. Submitting the PR early gives us more time to
  provide feedback and perhaps help you with it.

[[protocol-modules]]
== Protocol Modules

Packetbeat's source code is split up in Go packages or modules. The protocol
modules can be found in individual folders in the `beats/packetbeat/protos` directory
in the https://github.com/elastic/beats[Beats GitHub repository].

Before starting, we recommend reading through the source code of some of the
existing modules. For TCP based protocols, the MySQL or HTTP ones are good
models to follow. For UDP protocols, you can look at the DNS module.

All protocol modules implement the `TcpProtocolPlugin` or the
`UdpProtocolPlugin` (or both) from the following listing (found in
`beats/packetbeat/protos/protos.go`).

[source,go]
----------------------------------------------------------------------
// Functions to be exported by a protocol plugin
type ProtocolPlugin interface {
	// Called to initialize the Plugin
	Init(test_mode bool, results publish.Transactions) error

	// Called to return the configured ports
	GetPorts() []int
}

type TcpProtocolPlugin interface {
	ProtocolPlugin

	// Called when TCP payload data is available for parsing.
	Parse(pkt *Packet, tcptuple *common.TcpTuple,
		dir uint8, private ProtocolData) ProtocolData

	// Called when the FIN flag is seen in the TCP stream.
	ReceivedFin(tcptuple *common.TcpTuple, dir uint8,
		private ProtocolData) ProtocolData

	// Called when a packets are missing from the tcp
	// stream.
	GapInStream(tcptuple *common.TcpTuple, dir uint8, nbytes int,
		private ProtocolData) (priv ProtocolData, drop bool)

	// ConnectionTimeout returns the per stream connection timeout.
	// Return <=0 to set default tcp module transaction timeout.
	ConnectionTimeout() time.Duration
}

type UdpProtocolPlugin interface {
	ProtocolPlugin

	// ParseUdp is invoked when UDP payload data is available for parsing.
	ParseUdp(pkt *Packet)
}
----------------------------------------------------------------------

At the high level, the protocols plugins receive raw packet data via the
`Parse()` or `ParseUdp()` methods and produce objects that will be indexed into
Elasticsearch.

The `Parse()` and `ParseUdp()` methods are called for every packet that is
sniffed and is using one of the ports of the protocol as defined by the
`GetPorts()` function. They receive the packet in a Packet object, which looks
like this:

[source,go]
----------------------------------------------------------------------
type Packet struct {
	Ts      time.Time <1>
	Tuple   common.IpPortTuple <2>
	Payload []byte <3>
}
----------------------------------------------------------------------

<1> The timestamp of the packet
<2> The source IP address, source port, destination IP address, destination port
combination.
<3> The application layer payload (that is, without the IP and TCP/UDP headers) as a
byte slice.

The objects are sent using the `publisher.Client` interface defined in the
`beats/libbeat/publisher` directory in the https://github.com/elastic/beats[Beats GitHub repository].

Besides the `Parse()` function, the TCP layer also calls the `ReceivedFin()` function
when a TCP stream is closed, and it calls the `GapInStream()` function when packet
loss is detected in a TCP stream. The protocol module can use these callbacks to make
decisions about what to do with partial data that it receives. For example, for the
HTTP/1.0 protocol, the end of connection is used to know when the message is
finished.


[float]
=== Registering Your Plugin

To configure your plugin, you need to add a configuration struct to the
Protocols struct in `config/config.go`. This struct will be filled by
https://gopkg.in/yaml.v2[goyaml] on startup.

[source,go]
----------------------------------------------------------------------
type Protocols struct {
	Icmp     Icmp
	Dns      Dns
	Http     Http
	Memcache Memcache
	Mysql    Mysql
	Mongodb  Mongodb
	Pgsql    Pgsql
	Redis    Redis
	Thrift   Thrift
}
----------------------------------------------------------------------

Next create an ID for the new plugin in `protos/protos.go`:

[source,go]
----------------------------------------------------------------------
// Protocol constants.
const (
	UnknownProtocol Protocol = iota
	HttpProtocol
	MysqlProtocol
	RedisProtocol
	PgsqlProtocol
	ThriftProtocol
	MongodbProtocol
	DnsProtocol
	MemcacheProtocol
)

// Protocol names
var ProtocolNames = []string{
	"unknown",
	"http",
	"mysql",
	"redis",
	"pgsql",
	"thrift",
	"mongodb",
	"dns",
	"memcache",
}
----------------------------------------------------------------------

The protocol names must be in the same order as their corresponding protocol IDs. Additionally the protocol name must match the configuration name.

Finally register your new protocol plugin in `packetbeat.go` EnabledProtocolPlugins:

[source,go]
----------------------------------------------------------------------

var EnabledProtocolPlugins map[protos.Protocol]protos.ProtocolPlugin = map[protos.Protocol]protos.ProtocolPlugin{
	protos.HttpProtocol:     new(http.Http),
	protos.MemcacheProtocol: new(memcache.Memcache),
	protos.MysqlProtocol:    new(mysql.Mysql),
	protos.PgsqlProtocol:    new(pgsql.Pgsql),
	protos.RedisProtocol:    new(redis.Redis),
	protos.ThriftProtocol:   new(thrift.Thrift),
	protos.MongodbProtocol:  new(mongodb.Mongodb),
	protos.DnsProtocol:      new(dns.Dns),
}

----------------------------------------------------------------------

Once the module is registered, it can be configured, and packets will be processed.

Before implementing all the logic for your new protocol module, it can be
helpful to first register the module and implement the minimal plugin interface
for printing a debug message on received packets. This way you can test the plugin registration to ensure that it's working correctly.

[float]
=== The TCP Parse Function

For TCP protocols, the `Parse()` function is the heart of the module. As
mentioned earlier, this function is called for every TCP packet
that contains data on the configured ports.

It is important to understand that because TCP is a stream-based protocol,
the packet boundaries don't necessarily match the application
layer message boundaries. For example, a packet can contain only a part of the
message, it can contain a complete message, or it can contain multiple messages.

If you see a packet in the middle of the stream, you have no guaranties that its
first byte is the beginning of a message. However, if the packet is the first
seen in a given TCP stream, then you can assume it is the beginning of the message.

The `Parse()` function needs to deal with these facts, which generally means that it
needs to keep state across multiple packets.

Let's have a look again at its signature:

[source,go]
----------------------------------------------------------------------
func Parse(pkt *protos.Packet, tcptuple *common.TcpTuple, dir uint8,
	private protos.ProtocolData) protos.ProtocolData
----------------------------------------------------------------------

We've already talked about the first parameter, which contains the packet data.
The rest of the parameters and the return value are used for maintaining state
inside the TCP stream.

The `tcptuple` is a unique identifier for the TCP stream that the packet
is part of. You can use the `tcptuple.Hashable()` function to get a value that
you can store in a map. The `dir` flag gives you the direction in which the
packet is flowing inside the TCP stream. The two possible values are
`TcpDirectionOriginal` if the packet goes in the same direction as the first
packet from the stream and `TcpDirectionReverse` if the packet goes in
the other direction.

The `private` parameter can be used by the module to store state in the TCP stream.
The module would typically cast this at run time to a
type of its choice, modify it as needed, and then return the modified value.
The next time the TCP layer calls `Parse()` or another function from the
`TcpProtocolPlugin` interface, it will call the function with the modified
private value.

Here is an example of how the MySQL module handles the private data:

[source,go]
----------------------------------------------------------------------
	priv := mysqlPrivateData{}
	if private != nil {
		var ok bool
		priv, ok = private.(mysqlPrivateData)
		if !ok {
			priv = mysqlPrivateData{}
		}
	}

	[ ... ]

	return priv
----------------------------------------------------------------------

Most modules then use a logic similar to the following to deal with incomplete
data (this example is also from MySQL):


[source,go]
----------------------------------------------------------------------
		ok, complete := mysqlMessageParser(priv.Data[dir])
		if !ok {
			// drop this tcp stream. Will retry parsing with the next
			// segment in it
			priv.Data[dir] = nil
			logp.Debug("mysql", "Ignore MySQL message. Drop tcp stream.")
			return priv
		}

		if complete {
			mysql.messageComplete(tcptuple, dir, stream)
		} else {
			// wait for more data
			break
		}
----------------------------------------------------------------------

The `mysqlMessageParser()` is the function that tries to parse a single MySQL
message. Its implementation is MySQL-specific, so it's not interesting to us for this
guide. It returns two values: `ok`, which is `false` if there was a parsing error
from which we cannot recover, and `complete`, which indicates whether a complete
and valid message was separated from the stream. These two values are used for
deciding what to do next. In case of errors, we drop the stream. If there are no
errors, but the message is not yet complete, we do nothing and wait for more
data. Finally, if the message is complete, we go to the next level.

This block of code is called in a loop so that it can separate multiple messages
found in the same packet.

[float]
=== The UDP ParseUdp Function

If the protocol you are working on is running on top of UDP, then all the
complexities that TCP parser/decoders need to deal with around extracting
messages from packets are no longer relevant.

For an example, see the `ParseUdp()` function from the DNS module.

[float]
=== Correlation

Most protocols that Packetbeat supports today are request-response oriented.
Packetbeat indexes into Elasticsearch a document for each request-response pair
(called a transaction). This way we can have data from the request and the
response in the same document and measure the response time.

But this can be different for your protocol. For example for an asynchronous
protocol like AMPQ, it makes more sense to index a document for every message,
and then no correlation is necessary. On the other hand, for a session-based
protocol like SIP, it might make sense to index a document for a SIP transaction
or for a full SIP dialog, which can have more than two messages.

The TCP stream or UDP ports are usually good indicators that two messages belong
to the same transactions. Therefore most protocol implementations in
Packetbeat use a map with `tcptuple` maps for correlating the requests with the
responses. One thing you should be careful about is to expire and remove from
this map incomplete transactions. For example, we might see the request that has
created an entry in the map, but if we never see the reply, we need to remove
the request from memory on a timer, otherwise we risk leaking memory.

[float]
=== Sending the Result

After the correlation step, you should have an JSON-like object that can be sent
to Elasticsearch for indexing. You send the object by publishing it
through the publisher client interface, which is received by the `Init`
function. The publisher client accepts structures of type `common.MapStr`, which
is essentially a `map[string]interface{}` with a few more convenience methods
added (see the `beats/libbeat/common` package in the https://github.com/elastic/beats[Beats GitHub repository]).
 
As an example, here is the relevant code from the Redis module:

[source,go]
----------------------------------------------------------------------
    event := common.MapStr{
		"@timestamp":   common.Time(requ.Ts),
		"type":         "redis",
		"status":       error,
		"responsetime": responseTime,
		"redis":        returnValue,
		"method":       common.NetString(bytes.ToUpper(requ.Method)),
		"resource":     requ.Path,
		"query":        requ.Message,
		"bytes_in":     uint64(requ.Size),
		"bytes_out":    uint64(resp.Size),
		"src":          src,
		"dst":          dst,
	}
	if redis.SendRequest {
		event["request"] = requ.Message
	}
	if redis.SendResponse {
		event["response"] = resp.Message
	}
    
    return event
----------------------------------------------------------------------

The following fields are required and their presence will be checked by
system tests:

 * `@timestamp`. Set this to the timestamp of the first packet from the message
   and cast it to `common.Time` like in the example.
 * `type`. Set this to the protocol name.
 * `status`. The status of the transactions. Use either `common.OK_STATUS` or
   `common.ERROR_STATUS`. If the protocol doesn't have responses or a meaning of
   status code, use OK.
 * `resource`. This should represent what is requested, with the exact meaning
   depending on the protocol. For HTTP, this is the URL.  For SQL databases,
   this is the table name. For key-value stores, this is the key. If nothing
   seems to make sense to put in this field, use the empty string.

[float]
=== Helpers

[float]
==== Parsing Helpers

In libbeat you also find some helpers for implementing parsers for binary and
text-based protocols. The `Bytes_*` functions are the most low-level helpers
for binary protocols that use network byte order. These functions can be found in the
`beats/libbeat/common` module in the https://github.com/elastic/beats[Beats GitHub repository].
In addition to these very low-level helpers, a stream
buffer for parsing TCP-based streams, or simply UDP packets with integrated
error handling, is provided by `beats/libbeat/common/streambuf`. The following example
demonstrates using the stream buffer for parsing the Memcache protocol UDP header:

[source,go]
----------------------------------------------------------------------
func parseUdpHeader(buf *streambuf.Buffer) (mcUdpHeader, error) {
    var h mcUdpHeader
    h.requestId, _ = buf.ReadNetUint16()
    h.seqNumber, _ = buf.ReadNetUint16()
    h.numDatagrams, _ = buf.ReadNetUint16()
    buf.Advance(2) // ignore reserved
    return h, buf.Err()
}
----------------------------------------------------------------------

The stream buffer is also used to implement the binary and text-based protocols
for memcache.

[source,go]
----------------------------------------------------------------------
	header := buf.Snapshot()
	buf.Advance(memcacheHeaderSize)

	msg := parser.message
	if msg.IsRequest {
		msg.vbucket, _ = header.ReadNetUint16At(6)
	} else {
		msg.status, _ = header.ReadNetUint16At(6)
	}

	cas, _ := header.ReadNetUint64At(16)
	if cas != 0 {
		setCasUnique(msg, cas)
	}
	msg.opaque, _ = header.ReadNetUint32At(12)

	// check message length

	extraLen, _ := header.ReadNetUint8At(4)
	keyLen, _ := header.ReadNetUint16At(2)
	totalLen, _ := header.ReadNetUint32At(8)

    [...]

	if extraLen > 0 {
		tmp, _ := buf.Collect(int(extraLen))
		extras := streambuf.NewFixed(tmp)
		var err error
		if msg.IsRequest && requestArgs != nil {
			err = parseBinaryArgs(parser, requestArgs, header, extras)
		} else if responseArgs != nil {
			err = parseBinaryArgs(parser, responseArgs, header, extras)
		}
		if err != nil {
			msg.AddNotes(err.Error())
		}
	}

	if keyLen > 0 {
		key, _ := buf.Collect(int(keyLen))
		keys := []memcacheString{memcacheString{key}}
		msg.keys = keys
	}

	if valueLen == 0 {
		return parser.yield(buf.BufferConsumed())
	}
----------------------------------------------------------------------

The stream buffer also implements a number of interfaces defined in the standard "io" package
and can easily be used to serialize some packets for testing parsers (see
`beats/packetbeat/protos/memcache/binary_test.go`).

[float]
==== Module Helpers

Packetbeat provides the module `beats/packetbeat/protos/applayer` with
common definitions among all application layer protocols. For example using the
Transaction type from `applayer` guarantees that the final document will have all common required fields defined. Just embed the `applayer.Transaction` with your own
application layer transaction type to make use of it. Here is an example from the memcache protocol:

[source,go]
----------------------------------------------------------------------
	type transaction struct {
		applayer.Transaction

		command *commandType

		request  *message
		response *message
	}

	func (t *transaction) Event(event common.MapStr) error { // use applayer.Transaction to write common required fields
		if err := t.Transaction.Event(event); err != nil {
			logp.Warn("error filling generic transaction fields: %v", err)
			return err
		}

		mc := common.MapStr{}
		event["memcache"] = mc

        [...]

		return nil
	}
----------------------------------------------------------------------

Use `applayer.Message` in conjunction with `applayer.Transaction` for creating the
transaction and `applayer.Stream` to manage your stream buffers for parsing.

[[testing]]
== Testing

[float]
=== Unit Tests

For unit tests, use only the Go standard library
http://golang.org/pkg/testing/[testing] package. To make comparing complex
structures less verbose, we use the assert package from the
https://github.com/stretchr/testify[testify] library.

For parser and decoder tests, it's a good practice to have an array with
test cases containing the inputs and expected outputs. For an example, see the
`Test_splitCookiesHeader` unit test in `beats/packetbeat/protos/http/http_test.go`
in the https://github.com/elastic/beats[Beats GitHub repository].

You can also have unit tests that treat the whole module as a black box, calling
its interface functions, then reading the result and checking it. This pattern 
is especially useful for checking corner cases related to packet boundaries or 
correlation issues. Here is an example from the HTTP module:

[source,go]
----------------------------------------------------------------------
func Test_gap_in_body_http1dot0_fin(t *testing.T) {
	if testing.Verbose() { <1>
		logp.LogInit(logp.LOG_DEBUG, "", false, true, []string{"http",
			"httpdetailed"})
	}
	http := HttpModForTests()

	data1 := []byte("GET / HTTP/1.0\r\n\r\n") <2>

	data2 := []byte("HTTP/1.0 200 OK\r\n" +
		"Date: Tue, 14 Aug 2012 22:31:45 GMT\r\n" +
		"Expires: -1\r\n" +
		"Cache-Control: private, max-age=0\r\n" +
		"Content-Type: text/html; charset=UTF-8\r\n" +
		"Content-Encoding: gzip\r\n" +
		"Server: gws\r\n" +
		"X-XSS-Protection: 1; mode=block\r\n" +
		"X-Frame-Options: SAMEORIGIN\r\n" +
		"\r\n" +
		"xxxxxxxxxxxxxxxxxxxx")

	tcptuple := testCreateTCPTuple()
	req := protos.Packet{Payload: data1}
	resp := protos.Packet{Payload: data2}

	private := protos.ProtocolData(new(httpConnectionData))

	private = http.Parse(&req, tcptuple, 0, private) <3>
	private = http.ReceivedFin(tcptuple, 0, private)

	private = http.Parse(&resp, tcptuple, 1, private)

	logp.Debug("http", "Now sending gap..")

	private, drop := http.GapInStream(tcptuple, 1, 10, private)
	assert.Equal(t, false, drop)

	private = http.ReceivedFin(tcptuple, 1, private)

	trans := expectTransaction(t, http) <4>
	assert.NotNil(t, trans)
	assert.Equal(t, trans["notes"], []string{"Packet loss while capturing the response"})
}
----------------------------------------------------------------------

<1> It's useful to initialize the logging system in case the `-v` flag is passed
to `go test`. This makes it easy to get the logs for a failing test while
keeping the output clean on a normal run.

<2> Define the data we'll be using in the test.

<3> Call the interface functions exported by the module. The `private` structure
is passed from one call to the next like the TCP layer would do.

<4> The `expectTransaction` function tries to read from the publisher queue and
causes errors in the test case if there's no transaction present.

To check the coverage of your unit tests, run the `make cover` command at the
top of the repository.

[float]
=== System Testing

Because the main input to Packetbeat are packets and the main output are JSON
objects, a convenient way of testing its functionality is by providing PCAP
files as input and checking the results in the files created by using the "file"
output plugin.

This is the approach taken by the tests in the `beats/packetbeat/tests/system` directory
in the https://github.com/elastic/beats[Beats GitHub repository]. The
tests are written in Python and executed using
https://nose.readthedocs.org/en/latest/[nose]. Here is a simple example test
from the MongoDB suite:


[source,python]
----------------------------------------------------------------------
    def test_mongodb_find(self):
        """
        Should correctly pass a simple MongoDB find query
        """
        self.render_config_template( <1>
            mongodb_ports=[27017]
        )
        self.run_packetbeat(pcap="mongodb_find.pcap", <2>
                            debug_selectors=["mongodb"])

        objs = self.read_output() <3>
        o = objs[0]
        assert o["type"] == "mongodb"
        assert o["method"] == "find"
        assert o["status"] == "OK"
----------------------------------------------------------------------

<1> The configuration file for each test run is generated from the template. If
your protocol plugin has options in the configuration file, you should add them
to the template.

<2> The `run_packetbeat` function receives the PCAP file to run. It looks for
the PCAP file in the `beats/packetbeat/tests/system/pcaps` folder. The `debug_selectors` array controls
which log lines to be included. You can use `debug_selectors=["*"]` to enable
all debug messages.

<3> After the run, the test reads the output files and checks the result.

TIP: To generate the PCAP files, you can use Packetbeat. The `-dump` CLI
flag will dump to disk all the packets sniffed from the network that match the
BPF filter.

To run the whole test suite, use:

[source,shell]
----------------------------------------------------------------------
$ make test
----------------------------------------------------------------------

This requires you to have Python and virtualenv installed, but it automatically
creates and uses the virtualenv.

To run an individual test, use the following steps:

[source,shell]
----------------------------------------------------------------------
$ cd tests
$ . env/bin/activate
$ nosetests test_0025_mongodb_basic.py:Test.test_write_errors
----------------------------------------------------------------------

After running the individual test, you can check the logs, the output, and the
configuration file manually by looking into the folder that the `last_run`
symlink points to:

[source,shell]
----------------------------------------------------------------------
$ cd last_run
$ ls
output packetbeat.log packetbeat.yml
----------------------------------------------------------------------
