Merge remote-tracking branch 'origin/topic/robin/log-threads' into topic/bernhard/log-threads-configureable-heartbeat

Conflicts:
	src/threading/Manager.cc

CHANGES

@@ -1,4 +1,38 @@
 
+2.0-139 | 2012-03-02 09:33:04 -0800
+
+  * Changes to how script coverage integrates with test suites. (Jon Siwek)
+
+    - BRO_PROFILER_FILE now passes .X* templated filenames to mkstemp
+      for generating unique coverage state files.
+
+    - Rearranging Makefile targets.  The general rule is that if the
+      all/brief target fails out due to a test failure, then the dependent
+      coverage target won't run, but can still be invoked directly later.
+      (e.g. make brief || make coverage) 
+
+  * Standardized on the &default function for SSL constants. (Seth
+    Hall)
+
+  * Adding btest group "leaks" to leak tests. (Robin Sommer)
+    
+  * Adding btest group "comm" to communication tests for parallelizing
+    execution with new btest version. (Robin Sommer)
+
+  * Sorting all output for diffing in the external tests. (Robin
+    Sommer)
+
+  * Cleaned up dead code from the old SSL analyzers. Reported by
+    Julien Sentier. (Seth Hall)
+
+  * Update/add tests for broccoli IPv6 addr/subnet support. Addresses
+    #448. (Jon Siwek)
+
+  * Remove connection compressor. Addresses #559. (Jon Siwek)
+    
+  * Refactor IP_Hdr class ctors. Addresses #532. (Jon Siwek)
+    
+
 2.0-121 | 2012-02-24 16:34:17 -0800
 
   * A number of smaller memory fixes and code cleanups. (Julien

NEWS

@@ -21,6 +21,9 @@ Bro 2.1
   such that at the scripting layer, the name resolution can yield a
   set with both IPv4 and IPv6 addresses.
 
+- The connection compressor was already deprecated in 2.0 and has now
+  been removed from the code base.
+
 TODO: Extend.
 
 Bro 2.0

VERSION

@@ -1 +1 @@
-2.0-121
+2.0-139

aux/binpac

@@ -1 +1 @@
-Subproject commit 43308aab47a3357ca1885e1b6954154a2744d821
+Subproject commit 3034da8f082b61157e234237993ffd7a95be6e62

aux/bro-aux

@@ -1 +1 @@
-Subproject commit 139cc2e1e049c4e1cc7e95f20866102be1d3d599
+Subproject commit f53bcb2b492cb0db3dd288384040abc2ab711767

aux/broccoli

@@ -1 +1 @@
-Subproject commit 930e7c78221929849086a578308e2fdc99ac3fb8
+Subproject commit 2602eb53e70d7f0afae8fac58d7636b9291974a4

aux/broctl

@@ -1 +1 @@
-Subproject commit e908ba686dceb56065bdf569c18dd0f67f662f6b
+Subproject commit 954538514d71983e7ef3f0e109960466096e1c1d

scripts/base/init-bare.bro

@@ -1727,13 +1727,6 @@ global dns_skip_all_addl = T &redef;
 ## traffic and do not process it.  Set to 0 to turn off this functionality. 
 global dns_max_queries = 5;
 
-## The maxiumum size in bytes for an SSL cipher specifcation.  If we see a packet
-## that has bigger cipherspecs, we won't do a comparisons of cipherspecs.
-const ssl_max_cipherspec_size = 68 &redef;
-
-# todo::Is this still used?
-# type X509_extensions: table[count] of string;
-
 ## An X509 certificate.
 ##
 ## .. bro:see:: x509_certificate 
@@ -1746,10 +1739,6 @@ type X509: record {
 	not_valid_after: time;	##< Timestamp after when certificate is not valid.
 };
 
-# This is indexed with the CA's name and yields a DER (binary) encoded certificate.
-# todo::Is this still used?
-# const root_ca_certs: table[string] of string = {} &redef;
-
 ## HTTP session statistics.
 ##
 ## .. bro:see:: http_stats 
@@ -2167,26 +2156,6 @@ const forward_remote_state_changes = F &redef;
 ## Place-holder constant indicating "no peer".
 const PEER_ID_NONE = 0;
 
-## Deprecated.
-## 
-## .. todo:: The connection compressor is scheduled to be removed from Bro.
-const use_connection_compressor = F &redef;
-
-## Deprecated.
-## 
-## .. todo:: The connection compressor is scheduled to be removed from Bro.
-const cc_handle_resets = F &redef;
-
-## Deprecated.
-## 
-## .. todo:: The connection compressor is scheduled to be removed from Bro.
-const cc_handle_only_syns = T &redef;
-
-## Deprecated.
-## 
-## .. todo:: The connection compressor is scheduled to be removed from Bro.
-const cc_instantiate_on_data = F &redef;
-
 # Signature payload pattern types.
 # todo::use enum to help autodoc
 # todo::Still used?

scripts/base/protocols/ssl/consts.bro

@@ -13,7 +13,7 @@ export {
 		[TLSv10] = "TLSv10",
 		[TLSv11] = "TLSv11",
 		[TLSv12] = "TLSv12",
-	} &default="UNKNOWN";
+	} &default=function(i: count):string { return fmt("unknown-%d", i); };
 	
 	## Mapping between numeric codes and human readable strings for alert 
 	## levels.
@@ -535,7 +535,7 @@ export {
 		[SSL_RSA_FIPS_WITH_3DES_EDE_CBC_SHA] = "SSL_RSA_FIPS_WITH_3DES_EDE_CBC_SHA",
 		[SSL_RSA_FIPS_WITH_DES_CBC_SHA_2] = "SSL_RSA_FIPS_WITH_DES_CBC_SHA_2",
 		[SSL_RSA_FIPS_WITH_3DES_EDE_CBC_SHA_2] = "SSL_RSA_FIPS_WITH_3DES_EDE_CBC_SHA_2",
-	} &default="UNKNOWN";
+	} &default=function(i: count):string { return fmt("unknown-%d", i); };
 	
 	## Mapping between the constants and string values for SSL/TLS errors.
 	const x509_errors: table[count] of string = {
@@ -573,6 +573,6 @@ export {
 		[31] = "keyusage no certsign",
 		[32] = "unable to get crl issuer",
 		[33] = "unhandled critical extension",
-	};
+	} &default=function(i: count):string { return fmt("unknown-%d", i); };
 
 }

src/Anon.cc

@@ -154,7 +154,7 @@ void AnonymizeIPAddr_A50::init()
 int AnonymizeIPAddr_A50::PreservePrefix(ipaddr32_t input, int num_bits)
 	{
 	DEBUG_MSG("%s/%d\n",
-			IPAddr(IPAddr::IPv4, &input, IPAddr::Network).AsString().c_str(),
+			IPAddr(IPv4, &input, IPAddr::Network).AsString().c_str(),
 			num_bits);
 
 	if ( ! before_anonymization )

src/Brofiler.cc

@@ -1,4 +1,5 @@
 #include <cstdio>
+#include <cstring>
 #include <utility>
 #include <algorithm>
 #include "Brofiler.h"
@@ -48,10 +49,27 @@ bool Brofiler::WriteStats()
 	char* bf = getenv("BRO_PROFILER_FILE");
 	if ( ! bf ) return false;
 
-	FILE* f = fopen(bf, "w");
+	FILE* f;
+	const char* p = strstr(bf, ".XXXXXX");
+
+	if ( p && ! p[7] )
+		{
+		int fd = mkstemp(bf);
+		if ( fd == -1 )
+			{
+			reporter->Error("Failed to generate unique file name from BRO_PROFILER_FILE: %s", bf);
+			return false;
+			}
+		f = fdopen(fd, "w");
+		}
+	else
+		{
+		f = fopen(bf, "w");
+		}
+
 	if ( ! f )
 		{
-		reporter->Error("Failed to open BRO_PROFILER_FILE destination '%s' for writing\n", bf);
+		reporter->Error("Failed to open BRO_PROFILER_FILE destination '%s' for writing", bf);
 		return false;
 		}
 

src/Brofiler.h

@@ -26,7 +26,9 @@ public:
 	/**
 	 * Combines usage stats from current run with any read from ReadStats(),
 	 * then writes information to file pointed to by environment variable
-	 * BRO_PROFILER_FILE.
+	 * BRO_PROFILER_FILE.  If the value of that env. variable ends with
+	 * ".XXXXXX" (exactly 6 X's), then it is first passed through mkstemp
+	 * to get a unique file.
 	 *
 	 * @return: true when usage info is written, otherwise false.
 	 */

src/CMakeLists.txt

@@ -290,7 +290,6 @@ set(bro_SRCS
     ChunkedIO.cc
     CompHash.cc
     Conn.cc
-    ConnCompressor.cc
     ConnSizeAnalyzer.cc
     ContentLine.cc
     DCE_RPC.cc

src/CompHash.cc

@@ -709,7 +709,7 @@ const char* CompositeHash::RecoverOneVal(const HashKey* k, const char* kp0,
 		const uint32* const kp = AlignType<uint32>(kp0);
 		kp1 = reinterpret_cast<const char*>(kp+4);
 
-		IPAddr addr(IPAddr::IPv6, kp, IPAddr::Network);
+		IPAddr addr(IPv6, kp, IPAddr::Network);
 
 		switch ( tag ) {
 		case TYPE_ADDR:

src/Conn.h

@@ -239,30 +239,6 @@ public:
 	// Sets the transport protocol in use.
 	void SetTransport(TransportProto arg_proto)	{ proto = arg_proto; }
 
-	// If the connection compressor is activated, we need a special memory
-	// layout for connections. (See ConnCompressor.h)
-	void* operator new(size_t size)
-		{
-		if ( ! use_connection_compressor )
-			return ::operator new(size);
-
-		void* c = ::operator new(size + 4);
-
-		// We have to turn off the is_pending bit.  By setting the
-		// first four bytes to zero, we'll achieve this.
-		*((uint32*) c) = 0;
-
-		return ((char *) c) + 4;
-		}
-
-	void operator delete(void* ptr)
-		{
-		if ( ! use_connection_compressor )
-			::operator delete(ptr);
-		else
-			::operator delete(((char*) ptr) - 4);
-		}
-
 	void SetUID(uint64 arg_uid)	 { uid = arg_uid; }
 
 protected:

src/ConnCompressor.h

@@ -1,240 +0,0 @@
-// The ConnCompressor keeps track of the first packet seen for a conn_id using
-// only a minimal amount of memory. This helps us to avoid instantiating
-// full Connection objects for never-established sessions.
-//
-// TCP only.
-
-#ifndef CONNCOMPRESSOR_H
-#define CONNCOMPRESSOR_H
-
-#include "Conn.h"
-#include "Dict.h"
-#include "NetVar.h"
-#include "TCP.h"
-
-class ConnCompressor {
-public:
-	ConnCompressor();
-	~ConnCompressor();
-
-	// Handle next packet.  Returns 0 if packet in handled internally.
-	// Takes ownership of key.
-	Connection* NextPacket(double t, HashKey* k, const IP_Hdr* ip_hdr,
-			const struct pcap_pkthdr* hdr, const u_char* const pkt);
-
-	// Look up a connection.  Returns non-nil for connections for
-	// which a Connection object has already been instantiated.
-	Connection* Lookup(HashKey* k)
-		{
-		ConnData* c = conns.Lookup(k);
-		return c && IsConnPtr(c) ? MakeConnPtr(c) : 0;
-		}
-
-	// Inserts connection into compressor.  If another entry with this key
-	// already exists, it's replaced.  If that was a full connection, it is
-	// also returned.
-	Connection* Insert(Connection* c);
-
-	// Remove all state belonging to the given connection.  Returns
-	// true if the connection was found in the compressor's table,
-	// false if not.
-	bool Remove(HashKey* k);
-
-	// Flush state.
-	void Drain();
-
-	struct Sizes {
-		// Current number of already fully instantiated connections.
-		unsigned int connections;
-
-		// Total number of fully instantiated connections.
-		unsigned int connections_total;
-
-		// Current number of seen but non-yet instantiated connections.
-		unsigned int pending_valid;
-
-		// Total number of seen but non-yet instantiated connections.
-		unsigned int pending_total;
-
-		// Total number of all entries in pending list (some a which
-		// may already been invalid, but not yet removed from memory).
-		unsigned int pending_in_mem;
-
-		// Total number of hash table entires
-		// (should equal connections + pending_valid)
-		unsigned int hash_table_size;
-
-		// Total memory usage;
-		unsigned int memory;
-	};
-
-	const Sizes& Size()
-		{ sizes.hash_table_size = conns.Length(); return sizes; }
-
-	unsigned int MemoryAllocation() const	{ return sizes.memory; }
-
-	// As long as we have only seen packets from one side, we just
-	// store a PendingConn.
-	struct PendingConn {
-		// True if the block is indeed a PendingConn (see below).
-		unsigned int is_pending:1;
-
-		// Whether roles in key are flipped.
-		unsigned int ip1_is_src:1;
-
-		unsigned int invalid:1;	// deleted
-		int window_scale:4;
-		unsigned int SYN:1;
-		unsigned int FIN:1;
-		unsigned int RST:1;
-		unsigned int ACK:1;
-
-		double time;
-		struct Key {
-			uint32 ip1[4];
-			uint32 ip2[4];
-			uint16 port1;
-			uint16 port2;
-		} key;
-		uint32 seq;
-		uint32 ack;
-		hash_t hash;
-		uint16 window;
-		uint64 uid;
-
-		// The following are set if use_conn_size_analyzer is T.
-		uint16 num_pkts;
-		uint16 num_bytes_ip;
-	};
-
-private:
-	// Helpers to extract addrs/ports from PendingConn.
-
-	const uint32* SrcAddr(const PendingConn* c)
-		{ return c->ip1_is_src ? c->key.ip1 : c->key.ip2; }
-	const uint32* DstAddr(const PendingConn* c)
-		{ return c->ip1_is_src ? c->key.ip2 : c->key.ip1; }
-
-	uint16 SrcPort(const PendingConn* c)
-		{ return c->ip1_is_src ? c->key.port1 : c->key.port2; }
-	uint16 DstPort(const PendingConn* c)
-		{ return c->ip1_is_src ? c->key.port2 : c->key.port1; }
-
-
-	// Called for the first packet in a connection.
-	Connection* FirstFromOrig(double t, HashKey* key,
-					const IP_Hdr* ip, const tcphdr* tp);
-
-	// Called for more packets from the orginator w/o seeing a response.
-	Connection* NextFromOrig(PendingConn* pending, double t, HashKey* key,
-					const IP_Hdr* ip, const tcphdr* tp);
-
-	// Called for the first response packet. Instantiates a Connection.
-	Connection* Response(PendingConn* pending, double t, HashKey* key,
-					const IP_Hdr* ip, const tcphdr* tp);
-
-	// Instantiates a full TCP connection (invalidates pending connection).
-	Connection* Instantiate(HashKey* key, PendingConn* pending);
-
-	// Same but based on packet.
-	Connection* Instantiate(double t, HashKey* key, const IP_Hdr* ip);
-
-	// Fills the attributes of a PendingConn based on the given arguments.
-	void PktHdrToPendingConn(double time, const HashKey* key,
-		const IP_Hdr* ip, const struct tcphdr* tp, PendingConn* c);
-
-	// Fakes a TCP packet based on the available information.
-	const IP_Hdr* PendingConnToPacket(const PendingConn* c);
-
-	// Construct a TCP-flags byte.
-	uint8 MakeFlags(const PendingConn* c) const;
-
-	// Allocate room for a new (Ext)PendingConn.
-	PendingConn* MakeNewState(double t);
-
-	// Expire PendingConns.
-	void DoExpire(double t);
-
-	// Remove all state belonging to the given connection.
-	void Invalidate(HashKey* k);
-
-	// Sends the given connection_* event.  If orig_state is
-	// TCP_ENDPOINT__INACTIVE, tries to guess a better one based
-	// on pending.  If arg in non-nil, it will be used as the
-	// *first* argument of the event call (this is for conn_weird()).
-	void Event(const PendingConn* pending, double t,
-			const EventHandlerPtr& event, int orig_state,
-			int orig_size, int resp_state, Val* arg = 0);
-
-	void Weird(const PendingConn* pending, double t, const char* msg)
-		{
-		// This will actually go through the Reporter; Event() takes
-		// care of that.
-		Event(pending, t, conn_weird, TCP_ENDPOINT_INACTIVE, 0,
-		      TCP_ENDPOINT_INACTIVE, new StringVal(msg));
-		}
-
-	static const int BLOCK_SIZE = 16 * 1024;
-
-	// The memory managment for PendConns.
-	struct Block {
-		double time;
-		Block* prev;
-		Block* next;
-		int bytes_used;
-		unsigned char data[BLOCK_SIZE];
-	};
-
-	// In the connection hash table, we store pointers to both PendingConns
-	// and Connections. Thus, we need a way to differentiate between
-	// these two types. To avoid an additional indirection, we use a little
-	// hack: a pointer retrieved from the table is interpreted as a
-	// PendingConn first. However, if is_pending is false, it's in fact a
-	// Connection which starts at offset 4. The methods below help to
-	// implement this scheme transparently. An "operator new" in
-	// Connection takes care of building Connection's accordingly.
-	typedef PendingConn ConnData;
-	declare(PDict, ConnData);
-	typedef PDict(ConnData) ConnMap;
-	ConnMap conns;
-
-	static ConnData* MakeMapPtr(PendingConn* c)
-		{ assert(c->is_pending); return c; }
-
-	static ConnData* MakeMapPtr(Connection* c)
-		{
-		ConnData* p = (ConnData*) (((char*) c) - 4);
-		assert(!p->is_pending);
-		return p;
-		}
-
-	static PendingConn* MakePendingConnPtr(ConnData* c)
-		{ assert(c->is_pending); return c; }
-
-	static Connection* MakeConnPtr(ConnData* c)
-		{
-		assert(!c->is_pending);
-		return (Connection*) (((char*) c) + 4);
-		}
-
-	static bool IsConnPtr(ConnData* c)
-		{ return ! c->is_pending; }
-
-	// New blocks are inserted at the end.
-	Block* first_block;
-	Block* last_block;
-
-	// If we have already expired some entries in a block,
-	// this points to the first non-expired.
-	unsigned char* first_non_expired;
-
-	// Last "connection" that we have build.
-	RecordVal* conn_val;
-
-	// Statistics.
-	Sizes sizes;
-	};
-
-extern ConnCompressor* conn_compressor;
-
-#endif

src/DCE_RPC.cc

@@ -137,7 +137,7 @@ static bool is_mapped_dce_rpc_endpoint(const dce_rpc_endpoint_addr& addr)
 
 bool is_mapped_dce_rpc_endpoint(const ConnID* id, TransportProto proto)
 	{
-	if ( id->dst_addr.GetFamily() == IPAddr::IPv6 )
+	if ( id->dst_addr.GetFamily() == IPv6 )
 		// TODO: Does the protocol support v6 addresses? #773
 		return false;
 
@@ -414,7 +414,7 @@ void DCE_RPC_Session::DeliverEpmapperMapResponse(
 
 				case binpac::DCE_RPC_Simple::EPM_PROTOCOL_IP:
 					uint32 hostip = floor->rhs()->data()->ip();
-					mapped.addr.addr = IPAddr(IPAddr::IPv4, &hostip, IPAddr::Host);
+					mapped.addr.addr = IPAddr(IPv4, &hostip, IPAddr::Host);
 					break;
 				}
 				}

src/DNS_Mgr.cc

@@ -321,10 +321,10 @@ void DNS_Mapping::Init(struct hostent* h)
 		addrs = new IPAddr[num_addrs];
 		for ( int i = 0; i < num_addrs; ++i )
 			if ( h->h_addrtype == AF_INET )
-				addrs[i] = IPAddr(IPAddr::IPv4, (uint32*)h->h_addr_list[i],
+				addrs[i] = IPAddr(IPv4, (uint32*)h->h_addr_list[i],
 				                  IPAddr::Network);
 			else if ( h->h_addrtype == AF_INET6 )
-				addrs[i] = IPAddr(IPAddr::IPv6, (uint32*)h->h_addr_list[i],
+				addrs[i] = IPAddr(IPv6, (uint32*)h->h_addr_list[i],
 				                  IPAddr::Network);
 		}
 	else

src/Desc.cc

@@ -157,6 +157,16 @@ void ODesc::Add(double d)
 		}
 	}
 
+void ODesc::Add(const IPAddr& addr)
+	{
+	Add(addr.AsString());
+	}
+
+void ODesc::Add(const IPPrefix& prefix)
+	{
+	Add(prefix.AsString());
+	}
+
 void ODesc::AddCS(const char* s)
 	{
 	int n = strlen(s);

src/Desc.h

@@ -8,7 +8,6 @@
 #include <utility>
 
 #include "BroString.h"
-#include "IPAddr.h"
 
 typedef enum {
 	DESC_READABLE,
@@ -23,6 +22,8 @@ typedef enum {
 } desc_style;
 
 class BroFile;
+class IPAddr;
+class IPPrefix;
 
 class ODesc {
 public:
@@ -76,8 +77,8 @@ public:
 	void Add(int64 i);
 	void Add(uint64 u);
 	void Add(double d);
-	void Add(const IPAddr& addr)	{ Add(addr.AsString()); }
+	void Add(const IPAddr& addr);
-	void Add(const IPPrefix& prefix)	{ Add(prefix.AsString()); }
+	void Add(const IPPrefix& prefix);
 
 	// Add s as a counted string.
 	void AddCS(const char* s);

src/Frag.cc

@@ -125,7 +125,7 @@ void FragReassembler::AddFragment(double t, const IP_Hdr* ip, const u_char* pkt,
 
 void FragReassembler::Overlap(const u_char* b1, const u_char* b2, int n)
 	{
-	IP_Hdr proto_h((const struct ip*) proto_hdr);
+	IP_Hdr proto_h(proto_hdr, false);
 
 	if ( memcmp((const void*) b1, (const void*) b2, n) )
 		s->Weird("fragment_inconsistency", &proto_h);
@@ -157,7 +157,7 @@ void FragReassembler::BlockInserted(DataBlock* /* start_block */)
 			// can happen for benign reasons when we're
 			// intermingling parts of two fragmented packets.
 
-			IP_Hdr proto_h((const struct ip*) proto_hdr);
+			IP_Hdr proto_h(proto_hdr, false);
 			s->Weird("fragment_size_inconsistency", &proto_h);
 
 			// We decide to analyze the contiguous portion now.
@@ -171,7 +171,7 @@ void FragReassembler::BlockInserted(DataBlock* /* start_block */)
 
 	else if ( last_block->upper > frag_size )
 		{
-		IP_Hdr proto_h((const struct ip*) proto_hdr);
+		IP_Hdr proto_h(proto_hdr, false);
 		s->Weird("fragment_size_inconsistency", &proto_h);
 		frag_size = last_block->upper;
 		}
@@ -214,7 +214,7 @@ void FragReassembler::BlockInserted(DataBlock* /* start_block */)
 		}
 
 	delete reassembled_pkt;
-	reassembled_pkt = new IP_Hdr(reassem4);
+	reassembled_pkt = new IP_Hdr(reassem4, true);
 
 	DeleteTimer();
 	}

src/IP.h

@@ -9,23 +9,13 @@
 
 class IP_Hdr {
 public:
-	IP_Hdr(struct ip* arg_ip4)
+	IP_Hdr(const struct ip* arg_ip4, bool arg_del)
-		: ip4(arg_ip4), ip6(0), del(1)
+		: ip4(arg_ip4), ip6(0), del(arg_del)
 		{
 		}
 
-	IP_Hdr(const struct ip* arg_ip4)
+	IP_Hdr(const struct ip6_hdr* arg_ip6, bool arg_del)
-		: ip4(arg_ip4), ip6(0), del(0)
+		: ip4(0), ip6(arg_ip6), del(arg_del)
-		{
-		}
-
-	IP_Hdr(struct ip6_hdr* arg_ip6)
-		: ip4(0), ip6(arg_ip6), del(1)
-		{
-		}
-
-	IP_Hdr(const struct ip6_hdr* arg_ip6)
-		: ip4(0), ip6(arg_ip6), del(0)
 		{
 		}
 
@@ -90,7 +80,7 @@ public:
 private:
 	const struct ip* ip4;
 	const struct ip6_hdr* ip6;
-	int del;
+	bool del;
 };
 
 #endif

src/IPAddr.cc

@@ -250,7 +250,7 @@ IPPrefix::IPPrefix(const in6_addr& in6, uint8_t length)
 IPPrefix::IPPrefix(const IPAddr& addr, uint8_t length)
 	: prefix(addr)
 	{
-	if ( prefix.GetFamily() == IPAddr::IPv4 )
+	if ( prefix.GetFamily() == IPv4 )
 		{
 		if ( length > 32 )
 			reporter->InternalError("Bad IPAddr(v4) IPPrefix length : %d",
@@ -275,7 +275,7 @@ string IPPrefix::AsString() const
 	{
 	char l[16];
 
-	if ( prefix.GetFamily() == IPAddr::IPv4 )
+	if ( prefix.GetFamily() == IPv4 )
 		modp_uitoa10(length - 96, l);
 	else
 		modp_uitoa10(length, l);

src/IPAddr.h

@@ -10,6 +10,8 @@
 #include "BroString.h"
 #include "Hash.h"
 #include "util.h"
+#include "Type.h"
+#include "threading/SerialTypes.h"
 
 struct ConnID;
 class ExpectedConn;
@@ -25,7 +27,7 @@ public:
 	/**
 	 * Address family.
 	 */
-	enum Family { IPv4, IPv6 };
+	typedef IPFamily Family;
 
 	/**
 	 * Byte order.
@@ -45,7 +47,7 @@ public:
 	 *
 	 * @param in6 The IPv6 address.
 	 */
-	IPAddr(const in4_addr& in4)
+	explicit IPAddr(const in4_addr& in4)
 		{
 		memcpy(in6.s6_addr, v4_mapped_prefix, sizeof(v4_mapped_prefix));
 		memcpy(&in6.s6_addr[12], &in4.s_addr, sizeof(in4.s_addr));
@@ -56,7 +58,7 @@ public:
 	 *
 	 * @param in6 The IPv6 address.
 	 */
-	IPAddr(const in6_addr& arg_in6) : in6(arg_in6) { }
+	explicit IPAddr(const in6_addr& arg_in6) : in6(arg_in6) { }
 
 	/**
 	 * Constructs an address instance from a string representation.
@@ -318,14 +320,19 @@ public:
 		return memcmp(&addr1.in6, &addr2.in6, sizeof(in6_addr)) < 0;
 		}
 
+	/** Converts the address into the type used internally by the
+	  * inter-thread communication.
+	  */
+	void ConvertToThreadingValue(threading::Value::addr_t* v) const;
+
 	friend HashKey* BuildConnIDHashKey(const ConnID& id);
 	friend HashKey* BuildExpectedConnHashKey(const ExpectedConn& c);
 
-	friend class IPPrefix;
-
 	unsigned int MemoryAllocation() const { return padded_sizeof(*this); }
 
 private:
+	friend class IPPrefix;
+
 	/**
 	 * Initializes an address instance from a string representation.
 	 *
@@ -384,6 +391,25 @@ inline bool IPAddr::IsLoopback() const
 			&& (in6.s6_addr[14] == 0) && (in6.s6_addr[15] == 1));
 	}
 
+inline void IPAddr::ConvertToThreadingValue(threading::Value::addr_t* v) const
+	{
+	v->family = GetFamily();
+
+	switch ( v->family ) {
+
+	case IPv4:
+		CopyIPv4(&v->in.in4);
+		return;
+
+	case IPv6:
+		CopyIPv6(&v->in.in6);
+		return;
+
+	// Can't be reached.
+	abort();
+	}
+	}
+
 /**
   * Returns a hash key for a given ConnID. Passes ownership to caller.
   */
@@ -459,7 +485,7 @@ public:
 	 */
 	uint8_t Length() const
 		{
-		return prefix.GetFamily() == IPAddr::IPv4 ? length - 96 : length;
+		return prefix.GetFamily() == IPv4 ? length - 96 : length;
 		}
 
 	/**
@@ -516,6 +542,15 @@ public:
 		return new HashKey(&key, sizeof(key));
 		}
 
+	/** Converts the prefix into the type used internally by the
+	  * inter-thread communication.
+	  */
+	void ConvertToThreadingValue(threading::Value::subnet_t* v) const
+		{
+		v->length = length;
+		prefix.ConvertToThreadingValue(&v->prefix);
+		}
+
 	unsigned int MemoryAllocation() const { return padded_sizeof(*this); }
 
 	/**

src/Net.cc

@@ -486,6 +486,8 @@ void net_run()
 			// since Bro timers are not high-precision anyway.)
 			if ( ! using_communication )
 				usleep(100000);
+			else
+				usleep(1000);
 
 			// Flawfinder says about usleep:
 			//

src/NetVar.cc

@@ -47,15 +47,6 @@ int tcp_max_initial_window;
 int tcp_max_above_hole_without_any_acks;
 int tcp_excessive_data_without_further_acks;
 
-int ssl_compare_cipherspecs;
-int ssl_analyze_certificates;
-int ssl_store_certificates;
-int ssl_verify_certificates;
-int ssl_store_key_material;
-int ssl_max_cipherspec_size;
-StringVal* ssl_store_cert_path;
-StringVal* x509_trusted_cert_path;
-TableType* cipher_suites_list;
 RecordType* x509_type;
 
 double non_analyzed_lifetime;
@@ -192,8 +183,6 @@ StringVal* ssl_ca_certificate;
 StringVal* ssl_private_key;
 StringVal* ssl_passphrase;
 
-StringVal* x509_crl_file;
-
 Val* profiling_file;
 double profiling_interval;
 int expensive_profiling_multiple;
@@ -213,11 +202,6 @@ int sig_max_group_size;
 
 int enable_syslog;
 
-int use_connection_compressor;
-int cc_handle_resets;
-int cc_handle_only_syns;
-int cc_instantiate_on_data;
-
 TableType* irc_join_list;
 RecordType* irc_join_info;
 TableVal* irc_servers;
@@ -358,17 +342,7 @@ void init_net_var()
 	tcp_excessive_data_without_further_acks =
 		opt_internal_int("tcp_excessive_data_without_further_acks");
 
-	ssl_compare_cipherspecs  = opt_internal_int("ssl_compare_cipherspecs");
-	ssl_analyze_certificates = opt_internal_int("ssl_analyze_certificates");
-	ssl_store_certificates   = opt_internal_int("ssl_store_certificates");
-	ssl_verify_certificates  = opt_internal_int("ssl_verify_certificates");
-	ssl_store_key_material = opt_internal_int("ssl_store_key_material");
-	ssl_max_cipherspec_size  = opt_internal_int("ssl_max_cipherspec_size");
-
-	x509_trusted_cert_path = opt_internal_string("X509_trusted_cert_path");
-	ssl_store_cert_path = opt_internal_string("ssl_store_cert_path");
 	x509_type = internal_type("X509")->AsRecordType();
-	x509_crl_file = opt_internal_string("X509_crl_file");
 
 	non_analyzed_lifetime = opt_internal_double("non_analyzed_lifetime");
 	tcp_inactivity_timeout = opt_internal_double("tcp_inactivity_timeout");
@@ -525,12 +499,6 @@ void init_net_var()
 
 	gap_report_freq = opt_internal_double("gap_report_freq");
 
-	use_connection_compressor =
-		opt_internal_int("use_connection_compressor");
-	cc_handle_resets = opt_internal_int("cc_handle_resets");
-	cc_handle_only_syns = opt_internal_int("cc_handle_only_syns");
-	cc_instantiate_on_data = opt_internal_int("cc_instantiate_on_data");
-
 	irc_join_info = internal_type("irc_join_info")->AsRecordType();
 	irc_join_list = internal_type("irc_join_list")->AsTableType();
 	irc_servers = internal_val("irc_servers")->AsTableVal();

src/NetVar.h

@@ -50,17 +50,7 @@ extern int tcp_max_initial_window;
 extern int tcp_max_above_hole_without_any_acks;
 extern int tcp_excessive_data_without_further_acks;
 
-// see policy/ssl.bro for details
-extern int ssl_compare_cipherspecs;
-extern int ssl_analyze_certificates;
-extern int ssl_store_certificates;
-extern int ssl_verify_certificates;
-extern int ssl_store_key_material;
-extern int ssl_max_cipherspec_size;
-extern StringVal* ssl_store_cert_path;
-extern StringVal* x509_trusted_cert_path;
 extern RecordType* x509_type;
-extern StringVal* x509_crl_file;
 
 extern double non_analyzed_lifetime;
 extern double tcp_inactivity_timeout;
@@ -216,11 +206,6 @@ extern int sig_max_group_size;
 
 extern int enable_syslog;
 
-extern int use_connection_compressor;
-extern int cc_handle_resets;
-extern int cc_handle_only_syns;
-extern int cc_instantiate_on_data;
-
 extern TableType* irc_join_list;
 extern RecordType* irc_join_info;
 extern TableVal* irc_servers;

src/PIA.cc

@@ -196,7 +196,7 @@ void PIA_TCP::FirstPacket(bool is_orig, const IP_Hdr* ip)
 			ip4->ip_p = IPPROTO_TCP;
 
 			// Cast to const so that it doesn't delete it.
-			ip4_hdr = new IP_Hdr((const struct ip*) ip4);
+			ip4_hdr = new IP_Hdr(ip4, false);
 			}
 
 		if ( is_orig )

src/PacketSort.cc

@@ -27,9 +27,9 @@ PacketSortElement::PacketSortElement(PktSrc* arg_src,
 		{
 		const struct ip* ip = (const struct ip*) (pkt + hdr_size);
 		if ( ip->ip_v == 4 )
-			ip_hdr = new IP_Hdr(ip);
+			ip_hdr = new IP_Hdr(ip, false);
 		else
-			ip_hdr = new IP_Hdr((const struct ip6_hdr*) ip);
+			ip_hdr = new IP_Hdr((const struct ip6_hdr*) ip, false);
 
 		if ( ip_hdr->NextProto() == IPPROTO_TCP &&
 		      // Note: can't sort fragmented packets

src/RemoteSerializer.cc

@@ -234,7 +234,7 @@ static const int PRINT_BUFFER_SIZE = 10 * 1024;
 static const int SOCKBUF_SIZE = 1024 * 1024;
 
 // Buffer size for remote-log data.
-static const int LOG_BUFFER_SIZE = 50 * 1024;
+static const int LOG_BUFFER_SIZE = 512;
 
 struct ping_args {
 	uint32 seq;
@@ -532,6 +532,7 @@ RemoteSerializer::RemoteSerializer()
 	terminating = false;
 	in_sync = 0;
 	last_flush = 0;
+	received_logs = 0;
 	}
 
 RemoteSerializer::~RemoteSerializer()
@@ -681,7 +682,7 @@ RemoteSerializer::PeerID RemoteSerializer::Connect(const IPAddr& ip,
 	if ( ! initialized )
 		reporter->InternalError("remote serializer not initialized");
 
-	if ( ip.GetFamily() == IPAddr::IPv6 )
+	if ( ip.GetFamily() == IPv6 )
 		Error("inter-Bro communication not supported over IPv6");
 
 	const uint32* bytes;
@@ -1238,7 +1239,7 @@ bool RemoteSerializer::Listen(const IPAddr& ip, uint16 port, bool expect_ssl)
 	if ( ! initialized )
 		reporter->InternalError("remote serializer not initialized");
 
-	if ( ip.GetFamily() == IPAddr::IPv6 )
+	if ( ip.GetFamily() == IPv6 )
 		Error("inter-Bro communication not supported over IPv6");
 
 	const uint32* bytes;
@@ -1353,6 +1354,14 @@ double RemoteSerializer::NextTimestamp(double* local_network_time)
 	{
 	Poll(false);
 
+	if ( received_logs > 0 )
+		{
+		// If we processed logs last time, assume there's more.
+		idle = false;
+		received_logs = 0;
+		return timer_mgr->Time();
+		}
+
 	double et = events.length() ? events[0]->time : -1;
 	double pt = packets.length() ? packets[0]->time : -1;
 
@@ -2552,7 +2561,9 @@ bool RemoteSerializer::SendLogWrite(Peer* peer, EnumVal* id, EnumVal* writer, st
 	if ( ! peer->logs_requested )
 		return false;
 
-	assert(peer->log_buffer);
+	if ( ! peer->log_buffer )
+		// Peer shutting down.
+		return false;
 
 	// Serialize the log record entry.
 
@@ -2587,7 +2598,10 @@ bool RemoteSerializer::SendLogWrite(Peer* peer, EnumVal* id, EnumVal* writer, st
 	if ( len > (LOG_BUFFER_SIZE - peer->log_buffer_used) || (network_time - last_flush > 1.0) )
 		{
 		if ( ! FlushLogBuffer(peer) )
+			{
+			delete [] data;
 			return false;
+			}
 		}
 
 	// If the data is actually larger than our complete buffer, just send it out.
@@ -2631,6 +2645,12 @@ bool RemoteSerializer::ProcessLogCreateWriter()
 	if ( current_peer->state == Peer::CLOSING )
 		return false;
 
+#ifdef USE_PERFTOOLS
+	// Don't track allocations here, they'll be released only after the
+	// main loop exists. And it's just a tiny amount anyway.
+	HeapLeakChecker::Disabler disabler;
+#endif
+
 	assert(current_args);
 
 	EnumVal* id_val = 0;
@@ -2666,7 +2686,7 @@ bool RemoteSerializer::ProcessLogCreateWriter()
 	id_val = new EnumVal(id, BifType::Enum::Log::ID);
 	writer_val = new EnumVal(writer, BifType::Enum::Log::Writer);
 
-	if ( ! log_mgr->CreateWriter(id_val, writer_val, path, num_fields, fields) )
+	if ( ! log_mgr->CreateWriter(id_val, writer_val, path, num_fields, fields, true, false) )
 		goto error;
 
 	Unref(id_val);
@@ -2735,6 +2755,8 @@ bool RemoteSerializer::ProcessLogWrite()
 
 	fmt.EndRead();
 
+	++received_logs;
+
 	return true;
 
 error:
@@ -3376,6 +3398,9 @@ void SocketComm::Run()
 		small_timeout.tv_usec =
 			io->CanWrite() || io->CanRead() ? 1 : 10;
 
+		if ( ! io->CanWrite() )
+			usleep(10);
+
 		int a = select(max_fd + 1, &fd_read, &fd_write, &fd_except,
 				&small_timeout);
 

src/RemoteSerializer.h

@@ -338,6 +338,7 @@ private:
 	int propagate_accesses;
 	bool ignore_accesses;
 	bool terminating;
+	int received_logs;
 	Peer* source_peer;
 	PeerID id_counter;	// Keeps track of assigned IDs.
 	uint32 current_sync_point;

src/RuleMatcher.cc

@@ -1082,7 +1082,7 @@ static bool val_to_maskedval(Val* v, maskedvalue_list* append_to)
 			bool is_v4_mask = m[0] == 0xffffffff &&
 						m[1] == m[0] && m[2] == m[0];
 
-			if ( v->AsSubNet().Prefix().GetFamily() == IPAddr::IPv4 &&
+			if ( v->AsSubNet().Prefix().GetFamily() == IPv4 &&
 			     is_v4_mask )
 				{
 				mval->val = ntohl(*n);

src/SerializationFormat.cc

@@ -250,9 +250,9 @@ bool BinarySerializationFormat::Read(IPAddr* addr, const char* tag)
 		}
 
 	if ( n == 1 )
-		*addr = IPAddr(IPAddr::IPv4, raw, IPAddr::Network);
+		*addr = IPAddr(IPv4, raw, IPAddr::Network);
 	else
-		*addr = IPAddr(IPAddr::IPv6, raw, IPAddr::Network);
+		*addr = IPAddr(IPv6, raw, IPAddr::Network);
 
 	return true;
 	}
@@ -269,6 +269,32 @@ bool BinarySerializationFormat::Read(IPPrefix* prefix, const char* tag)
 	return true;
 	}
 
+bool BinarySerializationFormat::Read(struct in_addr* addr, const char* tag)
+	{
+	uint32_t* bytes = (uint32_t*) &addr->s_addr;
+
+	if ( ! Read(&bytes[0], "addr4") )
+		return false;
+
+	bytes[0] = htonl(bytes[0]);
+	return true;
+	}
+
+bool BinarySerializationFormat::Read(struct in6_addr* addr, const char* tag)
+	{
+	uint32_t* bytes = (uint32_t*) &addr->s6_addr;
+
+	for ( int i = 0; i < 4; ++i )
+		{
+		if ( ! Read(&bytes[i], "addr6-part") )
+			return false;
+
+		bytes[i] = htonl(bytes[i]);
+		}
+
+	return true;
+	}
+
 bool BinarySerializationFormat::Write(char v, const char* tag)
 	{
 	DBG_LOG(DBG_SERIAL, "Write char %s [%s]", fmt_bytes(&v, 1), tag);
@@ -362,6 +388,29 @@ bool BinarySerializationFormat::Write(const IPPrefix& prefix, const char* tag)
 	return Write(prefix.Prefix(), "prefix") && Write(prefix.Length(), "width");
 	}
 
+bool BinarySerializationFormat::Write(const struct in_addr& addr, const char* tag)
+	{
+	const uint32_t* bytes = (uint32_t*) &addr.s_addr;
+
+	if ( ! Write(ntohl(bytes[0]), "addr4") )
+		return false;
+
+	return true;
+	}
+
+bool BinarySerializationFormat::Write(const struct in6_addr& addr, const char* tag)
+	{
+	const uint32_t* bytes = (uint32_t*) &addr.s6_addr;
+
+	for ( int i = 0; i < 4; ++i )
+		{
+		if ( ! Write(ntohl(bytes[i]), "addr6-part") )
+			return false;
+		}
+
+	return true;
+	}
+
 bool BinarySerializationFormat::WriteOpenTag(const char* tag)
 	{
 	return true;
@@ -464,6 +513,18 @@ bool XMLSerializationFormat::Read(IPPrefix* prefix, const char* tag)
 	return false;
 	}
 
+bool XMLSerializationFormat::Read(struct in_addr* addr, const char* tag)
+	{
+	reporter->InternalError("no reading of xml");
+	return false;
+	}
+
+bool XMLSerializationFormat::Read(struct in6_addr* addr, const char* tag)
+	{
+	reporter->InternalError("no reading of xml");
+	return false;
+	}
+
 bool XMLSerializationFormat::Write(char v, const char* tag)
 	{
 	return WriteElem(tag, "char", &v, 1);
@@ -556,6 +617,18 @@ bool XMLSerializationFormat::Write(const IPPrefix& prefix, const char* tag)
 	return false;
 	}
 
+bool XMLSerializationFormat::Write(const struct in_addr& addr, const char* tag)
+	{
+	reporter->InternalError("XML output of in_addr not implemented");
+	return false;
+	}
+
+bool XMLSerializationFormat::Write(const struct in6_addr& addr, const char* tag)
+	{
+	reporter->InternalError("XML output of in6_addr not implemented");
+	return false;
+	}
+
 bool XMLSerializationFormat::WriteEncodedString(const char* s, int len)
 	{
 	while ( len-- )

src/SerializationFormat.h

@@ -9,6 +9,9 @@ using namespace std;
 
 #include "util.h"
 
+class IPAddr;
+class IPPrefix;
+
 // Abstract base class.
 class SerializationFormat {
 public:
@@ -30,6 +33,8 @@ public:
 	virtual bool Read(string* s, const char* tag) = 0;
 	virtual bool Read(IPAddr* addr, const char* tag) = 0;
 	virtual bool Read(IPPrefix* prefix, const char* tag) = 0;
+	virtual bool Read(struct in_addr* addr, const char* tag) = 0;
+	virtual bool Read(struct in6_addr* addr, const char* tag) = 0;
 
 	// Returns number of raw bytes read since last call to StartRead().
 	int BytesRead() const	{ return bytes_read; }
@@ -54,6 +59,8 @@ public:
 	virtual bool Write(const string& s, const char* tag) = 0;
 	virtual bool Write(const IPAddr& addr, const char* tag) = 0;
 	virtual bool Write(const IPPrefix& prefix, const char* tag) = 0;
+	virtual bool Write(const struct in_addr& addr, const char* tag) = 0;
+	virtual bool Write(const struct in6_addr& addr, const char* tag) = 0;
 
 	virtual bool WriteOpenTag(const char* tag) = 0;
 	virtual bool WriteCloseTag(const char* tag) = 0;
@@ -96,6 +103,8 @@ public:
 	virtual bool Read(string* s, const char* tag);
 	virtual bool Read(IPAddr* addr, const char* tag);
 	virtual bool Read(IPPrefix* prefix, const char* tag);
+	virtual bool Read(struct in_addr* addr, const char* tag);
+	virtual bool Read(struct in6_addr* addr, const char* tag);
 	virtual bool Write(int v, const char* tag);
 	virtual bool Write(uint16 v, const char* tag);
 	virtual bool Write(uint32 v, const char* tag);
@@ -109,6 +118,8 @@ public:
 	virtual bool Write(const string& s, const char* tag);
 	virtual bool Write(const IPAddr& addr, const char* tag);
 	virtual bool Write(const IPPrefix& prefix, const char* tag);
+	virtual bool Write(const struct in_addr& addr, const char* tag);
+	virtual bool Write(const struct in6_addr& addr, const char* tag);
 	virtual bool WriteOpenTag(const char* tag);
 	virtual bool WriteCloseTag(const char* tag);
 	virtual bool WriteSeparator();
@@ -133,6 +144,8 @@ public:
 	virtual bool Write(const string& s, const char* tag);
 	virtual bool Write(const IPAddr& addr, const char* tag);
 	virtual bool Write(const IPPrefix& prefix, const char* tag);
+	virtual bool Write(const struct in_addr& addr, const char* tag);
+	virtual bool Write(const struct in6_addr& addr, const char* tag);
 	virtual bool WriteOpenTag(const char* tag);
 	virtual bool WriteCloseTag(const char* tag);
 	virtual bool WriteSeparator();
@@ -150,6 +163,8 @@ public:
 	virtual bool Read(string* s, const char* tag);
 	virtual bool Read(IPAddr* addr, const char* tag);
 	virtual bool Read(IPPrefix* prefix, const char* tag);
+	virtual bool Read(struct in_addr* addr, const char* tag);
+	virtual bool Read(struct in6_addr* addr, const char* tag);
 
 private:
 	// Encodes non-printable characters.

src/Serializer.h

@@ -415,7 +415,7 @@ public:
 		}
 
 	const IP_Hdr IP() const
-		{ return IP_Hdr((struct ip *) (pkt + hdr_size)); }
+		{ return IP_Hdr((struct ip *) (pkt + hdr_size), true); }
 
 	void Describe(ODesc* d) const;
 

src/Sessions.cc

@@ -27,7 +27,6 @@
 #include "InterConn.h"
 #include "Discard.h"
 #include "RuleMatcher.h"
-#include "ConnCompressor.h"
 #include "DPM.h"
 
 #include "PacketSort.h"
@@ -275,13 +274,13 @@ void NetSessions::NextPacket(double t, const struct pcap_pkthdr* hdr,
 		const struct ip* ip = (const struct ip*) (pkt + hdr_size);
 		if ( ip->ip_v == 4 )
 			{
-			IP_Hdr ip_hdr(ip);
+			IP_Hdr ip_hdr(ip, false);
 			DoNextPacket(t, hdr, &ip_hdr, pkt, hdr_size);
 			}
 
 		else if ( ip->ip_v == 6 )
 			{
-			IP_Hdr ip_hdr((const struct ip6_hdr*) (pkt + hdr_size));
+			IP_Hdr ip_hdr((const struct ip6_hdr*) (pkt + hdr_size), false);
 			DoNextPacket(t, hdr, &ip_hdr, pkt, hdr_size);
 			}
 
@@ -510,7 +509,6 @@ void NetSessions::DoNextPacket(double t, const struct pcap_pkthdr* hdr,
 	id.src_addr = ip_hdr->SrcAddr();
 	id.dst_addr = ip_hdr->DstAddr();
 	Dictionary* d = 0;
-	bool pass_to_conn_compressor = false;
 
 	switch ( proto ) {
 	case IPPROTO_TCP:
@@ -520,7 +518,6 @@ void NetSessions::DoNextPacket(double t, const struct pcap_pkthdr* hdr,
 		id.dst_port = tp->th_dport;
 		id.is_one_way = 0;
 		d = &tcp_conns;
-		pass_to_conn_compressor = ip4 && use_connection_compressor;
 		break;
 		}
 
@@ -563,45 +560,40 @@ void NetSessions::DoNextPacket(double t, const struct pcap_pkthdr* hdr,
 
 	// FIXME: The following is getting pretty complex. Need to split up
 	// into separate functions.
-	if ( pass_to_conn_compressor )
+	conn = (Connection*) d->Lookup(h);
-		conn = conn_compressor->NextPacket(t, h, ip_hdr, hdr, pkt);
+	if ( ! conn )
+		{
+		conn = NewConn(h, t, &id, data, proto);
+		if ( conn )
+			d->Insert(h, conn);
+		}
 	else
 		{
-		conn = (Connection*) d->Lookup(h);
+		// We already know that connection.
-		if ( ! conn )
+		int consistent = CheckConnectionTag(conn);
+		if ( consistent < 0 )
 			{
+			delete h;
+			return;
+			}
+
+		if ( ! consistent || conn->IsReuse(t, data) )
+			{
+			if ( consistent )
+				conn->Event(connection_reused, 0);
+
+			Remove(conn);
 			conn = NewConn(h, t, &id, data, proto);
 			if ( conn )
 				d->Insert(h, conn);
 			}
 		else
-			{
-			// We already know that connection.
-			int consistent = CheckConnectionTag(conn);
-			if ( consistent < 0 )
-				{
-				delete h;
-				return;
-				}
-
-			if ( ! consistent || conn->IsReuse(t, data) )
-				{
-				if ( consistent )
-					conn->Event(connection_reused, 0);
-
-				Remove(conn);
-				conn = NewConn(h, t, &id, data, proto);
-				if ( conn )
-					d->Insert(h, conn);
-				}
-			else
-				delete h;
-			}
-
-		if ( ! conn )
 			delete h;
 		}
 
+	if ( ! conn )
+		delete h;
+
 	if ( ! conn )
 		return;
 
@@ -838,16 +830,7 @@ Connection* NetSessions::FindConnection(Val* v)
 	Dictionary* d;
 
 	if ( orig_portv->IsTCP() )
-		{
+		d = &tcp_conns;
-		if ( use_connection_compressor )
-			{
-			Connection* conn = conn_compressor->Lookup(h);
-			delete h;
-			return conn;
-			}
-		else
-			d = &tcp_conns;
-		}
 	else if ( orig_portv->IsUDP() )
 		d = &udp_conns;
 	else if ( orig_portv->IsICMP() )
@@ -900,17 +883,7 @@ void NetSessions::Remove(Connection* c)
 
 		switch ( c->ConnTransport() ) {
 		case TRANSPORT_TCP:
-			if ( use_connection_compressor &&
+			if ( ! tcp_conns.RemoveEntry(k) )
-			     conn_compressor->Remove(k) )
-				// Note, if the Remove() returned false
-				// then the compressor doesn't know about
-				// this connection, which *should* mean that
-				// we never gave it the connection in the
-				// first place, and thus we should check
-				// the regular TCP table instead.
-				;
-
-			else if ( ! tcp_conns.RemoveEntry(k) )
 				reporter->InternalError("connection missing");
 			break;
 
@@ -957,13 +930,8 @@ void NetSessions::Insert(Connection* c)
 	// reference the old key for already existing connections.
 
 	case TRANSPORT_TCP:
-		if ( use_connection_compressor )
+		old = (Connection*) tcp_conns.Remove(c->Key());
-			old = conn_compressor->Insert(c);
+		tcp_conns.Insert(c->Key(), c);
-		else
-			{
-			old = (Connection*) tcp_conns.Remove(c->Key());
-			tcp_conns.Insert(c->Key(), c);
-			}
 		break;
 
 	case TRANSPORT_UDP:
@@ -995,9 +963,6 @@ void NetSessions::Insert(Connection* c)
 
 void NetSessions::Drain()
 	{
-	if ( use_connection_compressor )
-		conn_compressor->Drain();
-
 	IterCookie* cookie = tcp_conns.InitForIteration();
 	Connection* tc;
 
@@ -1110,10 +1075,7 @@ Connection* NetSessions::NewConn(HashKey* k, double t, const ConnID* id,
 		conn->AppendAddl(fmt("tag=%s",
 					conn->GetTimerMgr()->GetTag().c_str()));
 
-	// If the connection compressor is active, it takes care of the
+	if ( new_connection )
-	// new_connection/connection_external events for TCP connections.
-	if ( new_connection &&
-	     (tproto != TRANSPORT_TCP || ! use_connection_compressor) )
 		{
 		conn->Event(new_connection, 0);
 

src/Stats.cc

@@ -6,7 +6,6 @@
 #include "Stats.h"
 #include "Scope.h"
 #include "cq.h"
-#include "ConnCompressor.h"
 #include "DNS_Mgr.h"
 #include "Trigger.h"
 #include "threading/Manager.h"
@@ -129,19 +128,6 @@ void ProfileLogger::Log()
 		expensive ? sessions->ConnectionMemoryUsageConnVals() / 1024 : 0
 		));
 
-	const ConnCompressor::Sizes& cs = conn_compressor->Size();
-
-	file->Write(fmt("%.6f ConnCompressor: pending=%d pending_in_mem=%d full_conns=%d pending+real=%d mem=%dK avg=%.1f/%.1f\n",
-		network_time,
-		cs.pending_valid,
-		cs.pending_in_mem,
-		cs.connections,
-		cs.hash_table_size,
-		cs.memory / 1024,
-		cs.memory / double(cs.pending_valid),
-		cs.memory / double(cs.pending_in_mem)
-		));
-
 	SessionStats s;
 	sessions->GetStats(s);
 

src/Val.cc

@@ -606,7 +606,7 @@ ID* MutableVal::Bind() const
 			ip = htonl(0x7f000001);	// 127.0.0.1
 
 		safe_snprintf(name, MAX_NAME_SIZE, "#%s#%d#",
-			      IPAddr(IPAddr::IPv4, &ip, IPAddr::Network)->AsString().c_str(),
+			      IPAddr(IPv4, &ip, IPAddr::Network)->AsString().c_str(),
 			      getpid());
 #else
 		safe_snprintf(name, MAX_NAME_SIZE, "#%s#%d#", host, getpid());
@@ -864,12 +864,12 @@ AddrVal::AddrVal(const char* text) : Val(TYPE_ADDR)
 AddrVal::AddrVal(uint32 addr) : Val(TYPE_ADDR)
 	{
 	// ### perhaps do gethostbyaddr here?
-	val.addr_val = new IPAddr(IPAddr::IPv4, &addr, IPAddr::Network);
+	val.addr_val = new IPAddr(IPv4, &addr, IPAddr::Network);
 	}
 
 AddrVal::AddrVal(const uint32 addr[4]) : Val(TYPE_ADDR)
 	{
-	val.addr_val = new IPAddr(IPAddr::IPv6, addr, IPAddr::Network);
+	val.addr_val = new IPAddr(IPv6, addr, IPAddr::Network);
 	}
 
 AddrVal::AddrVal(const IPAddr& addr) : Val(TYPE_ADDR)
@@ -889,7 +889,7 @@ unsigned int AddrVal::MemoryAllocation() const
 
 Val* AddrVal::SizeVal() const
 	{
-	if ( val.addr_val->GetFamily() == IPAddr::IPv4 )
+	if ( val.addr_val->GetFamily() == IPv4 )
 		return new Val(32, TYPE_COUNT);
 	else
 		return new Val(128, TYPE_COUNT);
@@ -933,13 +933,13 @@ SubNetVal::SubNetVal(const char* text, int width) : Val(TYPE_SUBNET)
 
 SubNetVal::SubNetVal(uint32 addr, int width) : Val(TYPE_SUBNET)
 	{
-	IPAddr a(IPAddr::IPv4, &addr, IPAddr::Network);
+	IPAddr a(IPv4, &addr, IPAddr::Network);
 	val.subnet_val = new IPPrefix(a, width);
 	}
 
 SubNetVal::SubNetVal(const uint32* addr, int width) : Val(TYPE_SUBNET)
 	{
-	IPAddr a(IPAddr::IPv6, addr, IPAddr::Network);
+	IPAddr a(IPv6, addr, IPAddr::Network);
 	val.subnet_val = new IPPrefix(a, width);
 	}
 
@@ -953,6 +953,16 @@ SubNetVal::~SubNetVal()
 	delete val.subnet_val;
 	}
 
+const IPAddr& SubNetVal::Prefix() const
+	{
+	return val.subnet_val->Prefix();
+	}
+
+int SubNetVal::Width() const
+	{
+	return val.subnet_val->Length();
+	}
+
 unsigned int SubNetVal::MemoryAllocation() const
 	{
 	return padded_sizeof(*this) + val.subnet_val->MemoryAllocation();
@@ -978,7 +988,7 @@ IPAddr SubNetVal::Mask() const
 		uint32 m[4];
 		for ( unsigned int i = 0; i < 4; ++i )
 			m[i] = 0;
-		IPAddr rval(IPAddr::IPv6, m, IPAddr::Host);
+		IPAddr rval(IPv6, m, IPAddr::Host);
 		return rval;
 		}
 
@@ -994,7 +1004,7 @@ IPAddr SubNetVal::Mask() const
 	while ( ++mp < m + 4 )
 		   *mp = 0;
 
-	IPAddr rval(IPAddr::IPv6, m, IPAddr::Host);
+	IPAddr rval(IPv6, m, IPAddr::Host);
 	return rval;
 	}
 

src/Val.h

@@ -513,10 +513,6 @@ protected:
 #define UDP_PORT_MASK  0x20000
 #define ICMP_PORT_MASK 0x30000
 
-typedef enum {
-	TRANSPORT_UNKNOWN, TRANSPORT_TCP, TRANSPORT_UDP, TRANSPORT_ICMP,
-} TransportProto;
-
 class PortVal : public Val {
 public:
 	// Constructors - both take the port number in host order.
@@ -588,8 +584,8 @@ public:
 
 	Val* SizeVal() const;
 
-	const IPAddr& Prefix() const { return val.subnet_val->Prefix(); }
+	const IPAddr& Prefix() const;
-	int Width() const	{ return val.subnet_val->Length(); }
+	int Width() const;
 	IPAddr Mask() const;
 
 	bool Contains(const IPAddr& addr) const;

src/X509.cc

@@ -1,263 +0,0 @@
-#include <openssl/err.h>
-
-#include "X509.h"
-#include "config.h"
-
-// ### NOTE: while d2i_X509 does not take a const u_char** pointer,
-// here we assume d2i_X509 does not write to <data>, so it is safe to
-// convert data to a non-const pointer.  Could some X509 guru verify
-// this?
-
-X509* d2i_X509_(X509** px, const u_char** in, int len)
-	{
-#ifdef OPENSSL_D2I_X509_USES_CONST_CHAR
-	  return d2i_X509(px, in, len);
-#else
-	  return d2i_X509(px, (u_char**)in, len);
-#endif
-	}
-
-X509_STORE* X509_Cert::ctx = 0;
-X509_LOOKUP* X509_Cert::lookup = 0;
-X509_STORE_CTX X509_Cert::csc;
-bool X509_Cert::bInited = false;
-
-// TODO: Check if Key < 768 Bits => Weakness!
-// FIXME: Merge verify and verifyChain.
-
-void X509_Cert::sslCertificateEvent(Contents_SSL* e, X509* pCert)
-	{
-	EventHandlerPtr event = ssl_certificate;
-	if ( ! event )
-		return;
-
-	char tmp[256];
-	RecordVal* pX509Cert = new RecordVal(x509_type);
-
-	X509_NAME_oneline(X509_get_issuer_name(pCert), tmp, sizeof tmp);
-	pX509Cert->Assign(0, new StringVal(tmp));
-	X509_NAME_oneline(X509_get_subject_name(pCert), tmp, sizeof tmp);
-	pX509Cert->Assign(1, new StringVal(tmp));
-	pX509Cert->Assign(2, new AddrVal(e->Conn()->OrigAddr()));
-
-	val_list* vl = new val_list;
-	vl->append(e->BuildConnVal());
-	vl->append(pX509Cert);
-	vl->append(new Val(e->IsOrig(), TYPE_BOOL));
-
-	e->Conn()->ConnectionEvent(event, e, vl);
-	}
-
-void X509_Cert::sslCertificateError(Contents_SSL* e, int error_numbe)
-	{
-	Val* err_str = new StringVal(X509_verify_cert_error_string(csc.error));
-	val_list* vl = new val_list;
-
-	vl->append(e->BuildConnVal());
-	vl->append(new Val(csc.error, TYPE_INT));
-	vl->append(err_str);
-
-	e->Conn()->ConnectionEvent(ssl_X509_error, e, vl);
-	}
-
-int X509_Cert::init()
-	{
-#if 0
-	OpenSSL_add_all_algorithms();
-#endif
-
-	ctx = X509_STORE_new();
-	int flag = 0;
-	int ret = 0;
-
-	if ( x509_trusted_cert_path &&
-	     x509_trusted_cert_path->AsString()->Len() > 0 )
-		{ // add the path(s) for the local CA's certificates
-		const BroString* pString = x509_trusted_cert_path->AsString();
-
-		lookup = X509_STORE_add_lookup(ctx, X509_LOOKUP_hash_dir());
-		if ( ! lookup )
-			{
-			reporter->Error("X509_Cert::init(): initing lookup failed\n");
-			flag = 1;
-			}
-
-		int i = X509_LOOKUP_add_dir(lookup,
-				(const char*) pString->Bytes(),
-				X509_FILETYPE_PEM);
-		if ( ! i )
-			{
-			reporter->Error("X509_Cert::init(): error adding lookup directory\n");
-			ret = 0;
-			}
-		}
-	else
-		{
-		printf("X509: Using the default trusted cert path.\n");
-		X509_STORE_set_default_paths(ctx);
-		}
-
-	// Add crl functionality - will only add if defined and
-	// X509_STORE_add_lookup was successful.
-	if ( ! flag && x509_crl_file && x509_crl_file->AsString()->Len() > 0 )
-		{
-		const BroString* rString = x509_crl_file->AsString();
-
-		if ( X509_load_crl_file(lookup, (const char*) rString->Bytes(),
-					X509_FILETYPE_PEM) != 1 )
-			{
-			reporter->Error("X509_Cert::init(): error reading CRL file\n");
-			ret = 1;
-			}
-
-#if 0
-		// Note, openssl version must be > 0.9.7(a).
-		X509_STORE_set_flags(ctx,
-			X509_V_FLAG_CRL_CHECK | X509_V_FLAG_CRL_CHECK_ALL);
-#endif
-		}
-
-	bInited = true;
-	return ret;
-	}
-
-int X509_Cert::verify(Contents_SSL* e, const u_char* data, uint32 len)
-	{
-	if ( ! bInited )
-		init();
-
-	X509* pCert = d2i_X509_(NULL, &data, len);
-	if ( ! pCert )
-		{
-		// 5 = X509_V_ERR_UNABLE_TO_DECODE_ISSUER_PUBLIC_KEY
-		sslCertificateError(e, 5);
-		return -1;
-		}
-
-	sslCertificateEvent(e, pCert);
-
-	X509_STORE_CTX_init(&csc, ctx, pCert, 0);
-	X509_STORE_CTX_set_time(&csc, 0, (time_t) network_time);
-	int i = X509_verify_cert(&csc);
-	X509_STORE_CTX_cleanup(&csc);
-	int ret = 0;
-
-	int ext = X509_get_ext_count(pCert);
-
-	if ( ext > 0 )
-		{
-		TableVal* x509ex = new TableVal(x509_extension);
-		val_list* vl = new val_list;
-		char buf[256];
-
-		for ( int k = 0; k < ext; ++k )
-			{
-			X509_EXTENSION* ex = X509_get_ext(pCert, k);
-			ASN1_OBJECT* obj = X509_EXTENSION_get_object(ex);
-			i2t_ASN1_OBJECT(buf, sizeof(buf), obj);
-
-			Val* index = new Val(k+1, TYPE_COUNT);
-			Val* value = new StringVal(strlen(buf), buf);
-			x509ex->Assign(index, value);
-			Unref(index);
-			// later we can do critical extensions like:
-			// X509_EXTENSION_get_critical(ex);
-			}
-
-		vl->append(e->BuildConnVal());
-		vl->append(x509ex);
-		e->Conn()->ConnectionEvent(process_X509_extensions, e, vl);
-		}
-
-	if ( ! i )
-		{
-		sslCertificateError(e, csc.error);
-		ret = csc.error;
-		}
-	else
-		ret = 0;
-
-	delete pCert;
-	return ret;
-	}
-
-int X509_Cert::verifyChain(Contents_SSL* e, const u_char* data, uint32 len)
-	{
-	if ( ! bInited )
-		init();
-
-	// Gets an ssl3x cert chain (could be one single cert, too,
-	// but in chain format).
-
-	// Init the stack.
-	STACK_OF(X509)* untrustedCerts = sk_X509_new_null();
-	if ( ! untrustedCerts )
-		{
-		// Internal error allocating stack of untrusted certs.
-		// 11 = X509_V_ERR_OUT_OF_MEM
-		sslCertificateError(e, 11);
-		return -1;
-		}
-
-	// NOT AGAIN!!!
-	// Extract certificates and put them into an OpenSSL Stack.
-	uint tempLength = 0;
-	int certCount = 0;
-	X509* pCert = 0; // base cert, this one is to be verified
-
-	while ( tempLength < len )
-		{
-		++certCount;
-		uint32 certLength =
-			uint32((data[tempLength + 0] << 16) |
-			       data[tempLength + 1] << 8) |
-			data[tempLength + 2];
-
-		// Points to current cert.
-		const u_char* pCurrentCert = &data[tempLength+3];
-
-		X509* pTemp = d2i_X509_(0, &pCurrentCert, certLength);
-		if ( ! pTemp )
-			{ // error is somewhat of a misnomer
-			// 5 = X509_V_ERR_UNABLE_TO_DECODE_ISSUER_PUBLIC_KEY
-			sslCertificateError(e, 5);
-			//FIXME: free ptrs
-			return -1;
-			}
-
-		if ( certCount == 1 )
-			// The first certificate goes directly into the ctx.
-			pCert = pTemp;
-		else
-			// The remaining certificates (if any) are put into
-			// the list of untrusted certificates
-			sk_X509_push(untrustedCerts, pTemp);
-
-		tempLength += certLength + 3;
-		}
-
-	sslCertificateEvent(e, pCert);
-
-	X509_STORE_CTX_init(&csc, ctx, pCert, untrustedCerts);
-	X509_STORE_CTX_set_time(&csc, 0, (time_t) network_time);
-	int i = X509_verify_cert(&csc);
-	X509_STORE_CTX_cleanup(&csc);
-	//X509_STORE_CTX_free(&csc);
-	int ret = 0;
-
-	if ( ! i )
-		{
-		sslCertificateError(e, csc.error);
-		ret = csc.error;
-		}
-	else
-		ret = 0;
-
-	delete pCert;
-	// Free the stack, incuding. contents.
-
-	// FIXME: could this break Bro's memory tracking?
-	sk_X509_pop_free(untrustedCerts, X509_free);
-
-	return ret;
-	}

src/bro.bif

@@ -1949,7 +1949,7 @@ function is_local_interface%(ip: addr%) : bool
 	if ( ent )
 		{
 		for ( unsigned int len = 0; ent->h_addr_list[len]; ++len )
-			addrs.push_back(IPAddr(IPAddr::IPv4, (uint32*)ent->h_addr_list[len],
+			addrs.push_back(IPAddr(IPv4, (uint32*)ent->h_addr_list[len],
 			                       IPAddr::Network));
 		}
 
@@ -1958,7 +1958,7 @@ function is_local_interface%(ip: addr%) : bool
 	if ( ent )
 		{
 		for ( unsigned int len = 0; ent->h_addr_list[len]; ++len )
-			addrs.push_back(IPAddr(IPAddr::IPv6, (uint32*)ent->h_addr_list[len],
+			addrs.push_back(IPAddr(IPv6, (uint32*)ent->h_addr_list[len],
 			                       IPAddr::Network));
 		}
 
@@ -2024,7 +2024,7 @@ function gethostname%(%) : string
 ## Returns: true if *a* is an IPv4 address, else false.
 function is_v4_addr%(a: addr%): bool
 	%{
-	if ( a->AsAddr().GetFamily() == IPAddr::IPv4 )
+	if ( a->AsAddr().GetFamily() == IPv4 )
 		return new Val(1, TYPE_BOOL);
 	else
 		return new Val(0, TYPE_BOOL);
@@ -2037,7 +2037,7 @@ function is_v4_addr%(a: addr%): bool
 ## Returns: true if *a* is an IPv6 address, else false.
 function is_v6_addr%(a: addr%): bool
 	%{
-	if ( a->AsAddr().GetFamily() == IPAddr::IPv6 )
+	if ( a->AsAddr().GetFamily() == IPv6 )
 		return new Val(1, TYPE_BOOL);
 	else
 		return new Val(0, TYPE_BOOL);
@@ -3522,7 +3522,7 @@ function lookup_location%(a: addr%) : geo_location
 		}
 
 #ifdef HAVE_GEOIP_COUNTRY_EDITION_V6
-	if ( geoip_v6 && a->AsAddr().GetFamily() == IPAddr::IPv6 )
+	if ( geoip_v6 && a->AsAddr().GetFamily() == IPv6 )
 		{
 		geoipv6_t ga;
 		a->AsAddr().CopyIPv6(&ga);
@@ -3534,7 +3534,7 @@ function lookup_location%(a: addr%) : geo_location
 	else
 #endif
 
-	if ( geoip && a->AsAddr().GetFamily() == IPAddr::IPv4 )
+	if ( geoip && a->AsAddr().GetFamily() == IPv4 )
 		{
 		const uint32* bytes;
 		a->AsAddr().GetBytes(&bytes);
@@ -3617,7 +3617,7 @@ function lookup_asn%(a: addr%) : count
 		{
 // IPv6 support showed up in 1.4.5.
 #ifdef HAVE_GEOIP_COUNTRY_EDITION_V6
-		if ( a->AsAddr().GetFamily() == IPAddr::IPv6 )
+		if ( a->AsAddr().GetFamily() == IPv6 )
 			{
 			geoipv6_t ga;
 			a->AsAddr().CopyIPv6(&ga);
@@ -3626,7 +3626,7 @@ function lookup_asn%(a: addr%) : count
 		else
 #endif
 
-		if ( a->AsAddr().GetFamily() == IPAddr::IPv4 )
+		if ( a->AsAddr().GetFamily() == IPv4 )
 			{
 			const uint32* bytes;
 			a->AsAddr().GetBytes(&bytes);
@@ -5353,7 +5353,7 @@ function preserve_prefix%(a: addr, width: count%): any
 	AnonymizeIPAddr* ip_anon = ip_anonymizer[PREFIX_PRESERVING_A50];
 	if ( ip_anon )
 		{
-		if ( a->AsAddr().GetFamily() == IPAddr::IPv6 )
+		if ( a->AsAddr().GetFamily() == IPv6 )
 			builtin_error("preserve_prefix() not supported for IPv6 addresses");
 		else
 			{
@@ -5382,7 +5382,7 @@ function preserve_subnet%(a: subnet%): any
 	AnonymizeIPAddr* ip_anon = ip_anonymizer[PREFIX_PRESERVING_A50];
 	if ( ip_anon )
 		{
-		if ( a->AsSubNet().Prefix().GetFamily() == IPAddr::IPv6 )
+		if ( a->AsSubNet().Prefix().GetFamily() == IPv6 )
 			builtin_error("preserve_subnet() not supported for IPv6 addresses");
 		else
 			{
@@ -5418,7 +5418,7 @@ function anonymize_addr%(a: addr, cl: IPAddrAnonymizationClass%): addr
 	if ( anon_class < 0 || anon_class >= NUM_ADDR_ANONYMIZATION_CLASSES )
 		builtin_error("anonymize_addr(): invalid ip addr anonymization class");
 
-	if ( a->AsAddr().GetFamily() == IPAddr::IPv6 )
+	if ( a->AsAddr().GetFamily() == IPv6 )
 		{
 		builtin_error("anonymize_addr() not supported for IPv6 addresses");
 		return 0;

src/event.bif

@@ -4594,7 +4594,6 @@ event ssh_server_version%(c: connection, version: string%);
 ##
 ## .. bro:see:: ssl_alert  ssl_established ssl_extension ssl_server_hello
 ##    ssl_session_ticket_handshake x509_certificate x509_error x509_extension
-##    ssl_max_cipherspec_size
 event ssl_client_hello%(c: connection, version: count, possible_ts: time, session_id: string, ciphers: count_set%);
 
 ## Generated for an SSL/TLS servers's initial *hello* message. SSL/TLS sessions
@@ -4625,7 +4624,6 @@ event ssl_client_hello%(c: connection, version: count, possible_ts: time, sessio
 ##
 ## .. bro:see:: ssl_alert ssl_client_hello ssl_established ssl_extension
 ##    ssl_session_ticket_handshake x509_certificate x509_error x509_extension
-##    ssl_max_cipherspec_size
 event ssl_server_hello%(c: connection, version: count, possible_ts: time, session_id: string, cipher: count, comp_method: count%);
 
 ## Generated for SSL/TLS extensions seen in an initial handshake.  SSL/TLS sessions

src/logging/Manager.cc

@@ -105,9 +105,6 @@ Manager::Stream::~Stream()
 		{
 		WriterInfo* winfo = i->second;
 
-		if ( ! winfo )
-			continue;
-
 		if ( winfo->rotation_timer )
 			timer_mgr->Cancel(winfo->rotation_timer);
 
@@ -207,7 +204,7 @@ Manager::WriterInfo* Manager::FindWriter(WriterFrontend* writer)
 			{
 			WriterInfo* winfo = i->second;
 
-			if ( winfo && winfo->writer == writer )
+			if ( winfo->writer == writer )
 				return winfo;
 			}
 		}
@@ -221,7 +218,7 @@ void Manager::RemoveDisabledWriters(Stream* stream)
 
 	for ( Stream::WriterMap::iterator j = stream->writers.begin(); j != stream->writers.end(); j++ )
 		{
-		if ( j->second && j->second->writer->Disabled() )
+		if ( j->second->writer->Disabled() )
 			{
 			j->second->writer->Stop();
 			delete j->second;
@@ -680,11 +677,11 @@ bool Manager::Write(EnumVal* id, RecordVal* columns)
 
 			Val* path_arg;
 			if ( filter->path_val )
-				path_arg = filter->path_val;
+				path_arg = filter->path_val->Ref();
 			else
 				path_arg = new StringVal("");
 
-			vl.append(path_arg->Ref());
+			vl.append(path_arg);
 
 			Val* rec_arg;
 			BroType* rt = filter->path_func->FType()->Args()->FieldType("rec");
@@ -718,7 +715,6 @@ bool Manager::Write(EnumVal* id, RecordVal* columns)
 
 			if ( ! filter->path_val )
 				{
-				Unref(path_arg);
 				filter->path = v->AsString()->CheckString();
 				filter->path_val = v->Ref();
 				}
@@ -740,7 +736,7 @@ bool Manager::Write(EnumVal* id, RecordVal* columns)
 
 		if ( w != stream->writers.end() )
 			// We know this writer already.
-			writer = w->second ? w->second->writer : 0;
+			writer = w->second->writer;
 
 		else
 			{
@@ -753,64 +749,25 @@ bool Manager::Write(EnumVal* id, RecordVal* columns)
 			for ( int j = 0; j < filter->num_fields; ++j )
 				arg_fields[j] = new Field(*filter->fields[j]);
 
-			if ( filter->remote )
+			writer = CreateWriter(stream->id, filter->writer,
-				remote_serializer->SendLogCreateWriter(stream->id,
+					      path, filter->num_fields,
-								       filter->writer,
+					      arg_fields, filter->local, filter->remote);
-								       path,
-								       filter->num_fields,
-								       arg_fields);
 
-			if ( filter->local )
+			if ( ! writer )
 				{
-				writer = CreateWriter(stream->id, filter->writer,
+				Unref(columns);
-						      path, filter->num_fields,
+				return false;
-						      arg_fields);
-
-				if ( ! writer )
-					{
-					Unref(columns);
-					return false;
-					}
 				}
-			else
-				{
-				// Insert a null pointer into the map to make
-				// sure we don't try creating it again.
-				stream->writers.insert(Stream::WriterMap::value_type(
-				Stream::WriterPathPair(filter->writer->AsEnum(), path), 0));
 
-				for( int i = 0; i < filter->num_fields; ++i)
-					delete arg_fields[i];
-
-				delete [] arg_fields;
-				}
 			}
 
 		// Alright, can do the write now.
 
-		if ( filter->local || filter->remote )
+		threading::Value** vals = RecordToFilterVals(stream, filter, columns);
-			{
-				threading::Value** vals = RecordToFilterVals(stream, filter, columns);
-
-			if ( filter->remote )
-				remote_serializer->SendLogWrite(stream->id,
-								filter->writer,
-								path,
-								filter->num_fields,
-								vals);
-
-			if ( filter->local )
-				{
-				// Write takes ownership of vals.
-				assert(writer);
-				writer->Write(filter->num_fields, vals);
-				}
-
-			else
-				DeleteVals(filter->num_fields, vals);
-
-			}
 
+		// Write takes ownership of vals.
+		assert(writer);
+		writer->Write(filter->num_fields, vals);
 
 #ifdef DEBUG
 		DBG_LOG(DBG_LOGGING, "Wrote record to filter '%s' on stream '%s'",
@@ -862,11 +819,11 @@ threading::Value* Manager::ValToLogVal(Val* val, BroType* ty)
 		break;
 
 	case TYPE_SUBNET:
-		lval->val.subnet_val = new IPPrefix(val->AsSubNet());
+		val->AsSubNet().ConvertToThreadingValue(&lval->val.subnet_val);
 		break;
 
 	case TYPE_ADDR:
-		lval->val.addr_val = new IPAddr(val->AsAddr());
+		val->AsAddr().ConvertToThreadingValue(&lval->val.addr_val);
 		break;
 
 	case TYPE_DOUBLE:
@@ -976,7 +933,7 @@ Value** Manager::RecordToFilterVals(Stream* stream, Filter* filter,
 	}
 
 WriterFrontend* Manager::CreateWriter(EnumVal* id, EnumVal* writer, string path,
-				int num_fields, const Field* const*  fields)
+				int num_fields, const Field* const*  fields, bool local, bool remote)
 	{
 	Stream* stream = FindStream(id);
 
@@ -987,12 +944,12 @@ WriterFrontend* Manager::CreateWriter(EnumVal* id, EnumVal* writer, string path,
 	Stream::WriterMap::iterator w =
 		stream->writers.find(Stream::WriterPathPair(writer->AsEnum(), path));
 
-	if ( w != stream->writers.end() && w->second )
+	if ( w != stream->writers.end() )
 		// If we already have a writer for this. That's fine, we just
 		// return it.
 		return w->second->writer;
 
-	WriterFrontend* writer_obj = new WriterFrontend(writer->AsEnum());
+	WriterFrontend* writer_obj = new WriterFrontend(id, writer, local, remote);
 	assert(writer_obj);
 
 	writer_obj->Init(path, num_fields, fields);
@@ -1089,8 +1046,7 @@ bool Manager::Write(EnumVal* id, EnumVal* writer, string path, int num_fields,
 		return false;
 		}
 
-	if ( w->second )
+	w->second->writer->Write(num_fields, vals);
-		w->second->writer->Write(num_fields, vals);
 
 	DBG_LOG(DBG_LOGGING,
 		"Wrote pre-filtered record to path '%s' on stream '%s'",
@@ -1111,9 +1067,6 @@ void Manager::SendAllWritersTo(RemoteSerializer::PeerID peer)
 		for ( Stream::WriterMap::iterator i = stream->writers.begin();
 		      i != stream->writers.end(); i++ )
 			{
-			if ( ! i->second )
-				continue;
-
 			WriterFrontend* writer = i->second->writer;
 
 			EnumVal writer_val(i->first.first, BifType::Enum::Log::Writer);
@@ -1134,10 +1087,7 @@ bool Manager::SetBuf(EnumVal* id, bool enabled)
 
 	for ( Stream::WriterMap::iterator i = stream->writers.begin();
 	      i != stream->writers.end(); i++ )
-		{
+		i->second->writer->SetBuf(enabled);
-		if ( i->second )
-			i->second->writer->SetBuf(enabled);
-		}
 
 	RemoveDisabledWriters(stream);
 
@@ -1155,10 +1105,7 @@ bool Manager::Flush(EnumVal* id)
 
 	for ( Stream::WriterMap::iterator i = stream->writers.begin();
 	      i != stream->writers.end(); i++ )
-		{
+		i->second->writer->Flush();
-		if ( i->second )
-			i->second->writer->Flush();
-		}
 
 	RemoveDisabledWriters(stream);
 

src/logging/Manager.h

@@ -159,7 +159,8 @@ protected:
 
 	// Takes ownership of fields.
 	WriterFrontend* CreateWriter(EnumVal* id, EnumVal* writer, string path,
-				int num_fields, const threading::Field* const* fields);
+				int num_fields, const threading::Field* const* fields,
+				bool local, bool remote);
 
 	// Takes ownership of values..
 	bool Write(EnumVal* id, EnumVal* writer, string path,

src/logging/WriterBackend.cc

@@ -242,4 +242,40 @@ bool WriterBackend::DoHeartbeat(double network_time, double current_time)
 	return true;
 	}
 
+string WriterBackend::Render(const threading::Value::addr_t& addr) const
+	{
+	if ( addr.family == IPv4 )
+		{
+		char s[INET_ADDRSTRLEN];
+
+		if ( inet_ntop(AF_INET, &addr.in.in4, s, INET_ADDRSTRLEN) == NULL )
+			return "<bad IPv4 address conversion>";
+		else
+			return s;
+		}
+	else
+		{
+		char s[INET6_ADDRSTRLEN];
+
+		if ( inet_ntop(AF_INET6, &addr.in.in6, s, INET6_ADDRSTRLEN) == NULL )
+			return "<bad IPv6 address conversion>";
+		else
+			return s;
+		}
+	}
+
+string WriterBackend::Render(const threading::Value::subnet_t& subnet) const
+	{
+	char l[16];
+
+	if ( subnet.prefix.family == IPv4 )
+		modp_uitoa10(subnet.length - 96, l);
+	else
+		modp_uitoa10(subnet.length, l);
+
+	string s = Render(subnet.prefix) + "/" + l;
+
+	return s;
+	}
+
 

src/logging/WriterBackend.h

@@ -158,6 +158,21 @@ public:
 	bool FinishedRotation(string new_name, string old_name,
 			      double open, double close, bool terminating);
 
+	/** Helper method to render an IP address as a string.
+	  *
+	  * @param addr The address.
+	  *
+	  * @return An ASCII representation of the address.
+	  */
+	string Render(const threading::Value::addr_t& addr) const;
+
+	/** Helper method to render an subnet value as a string.
+	  *
+	  * @param addr The address.
+	  *
+	  * @return An ASCII representation of the address.
+	  */
+	string Render(const threading::Value::subnet_t& subnet) const;
 
 protected:
 	/**

src/logging/WriterFrontend.cc

@@ -99,21 +99,36 @@ public:
 
 using namespace logging;
 
-WriterFrontend::WriterFrontend(bro_int_t type)
+WriterFrontend::WriterFrontend(EnumVal* arg_stream, EnumVal* arg_writer, bool arg_local, bool arg_remote)
 	{
+	stream = arg_stream;
+	writer = arg_writer;
+	Ref(stream);
+	Ref(writer);
+
 	disabled = initialized = false;
 	buf = true;
+	local = arg_local;
+	remote = arg_remote;
 	write_buffer = 0;
 	write_buffer_pos = 0;
 	ty_name = "<not set>";
-	backend = log_mgr->CreateBackend(this, type);
 
-	assert(backend);
+	if ( local )
-	backend->Start();
+		{
+		backend = log_mgr->CreateBackend(this, writer->AsEnum());
+		assert(backend);
+		backend->Start();
+		}
+
+	else
+		backend = 0;
 	}
 
 WriterFrontend::~WriterFrontend()
 	{
+	Unref(stream);
+	Unref(writer);
 	}
 
 string WriterFrontend::Name() const
@@ -128,7 +143,9 @@ void WriterFrontend::Stop()
 	{
 	FlushWriteBuffer();
 	SetDisable();
-	backend->Stop();
+
+	if ( backend )
+		backend->Stop();
 	}
 
 void WriterFrontend::Init(string arg_path, int arg_num_fields, const Field* const * arg_fields)
@@ -144,7 +161,17 @@ void WriterFrontend::Init(string arg_path, int arg_num_fields, const Field* cons
 	fields = arg_fields;
 
 	initialized = true;
-	backend->SendIn(new InitMessage(backend, arg_path, arg_num_fields, arg_fields));
+
+	if ( backend )
+		backend->SendIn(new InitMessage(backend, arg_path, arg_num_fields, arg_fields));
+
+	if ( remote )
+		remote_serializer->SendLogCreateWriter(stream,
+						       writer,
+						       arg_path,
+						       arg_num_fields,
+						       arg_fields);
+
 	}
 
 void WriterFrontend::Write(int num_fields, Value** vals)
@@ -152,6 +179,19 @@ void WriterFrontend::Write(int num_fields, Value** vals)
 	if ( disabled )
 		return;
 
+	if ( remote )
+		remote_serializer->SendLogWrite(stream,
+						writer,
+						path,
+						num_fields,
+						vals);
+
+	if ( ! backend )
+		{
+		DeleteVals(vals);
+		return;
+		}
+
 	if ( ! write_buffer )
 		{
 		// Need new buffer.
@@ -173,7 +213,8 @@ void WriterFrontend::FlushWriteBuffer()
 		// Nothing to do.
 		return;
 
-	backend->SendIn(new WriteMessage(backend, num_fields, write_buffer_pos, write_buffer));
+	if ( backend )
+		backend->SendIn(new WriteMessage(backend, num_fields, write_buffer_pos, write_buffer));
 
 	// Clear buffer (no delete, we pass ownership to child thread.)
 	write_buffer = 0;
@@ -187,7 +228,8 @@ void WriterFrontend::SetBuf(bool enabled)
 
 	buf = enabled;
 
-	backend->SendIn(new SetBufMessage(backend, enabled));
+	if ( backend )
+		backend->SendIn(new SetBufMessage(backend, enabled));
 
 	if ( ! buf )
 		// Make sure no longer buffer any still queued data.
@@ -200,7 +242,9 @@ void WriterFrontend::Flush()
 		return;
 
 	FlushWriteBuffer();
-	backend->SendIn(new FlushMessage(backend));
+
+	if ( backend )
+		backend->SendIn(new FlushMessage(backend));
 	}
 
 void WriterFrontend::Rotate(string rotated_path, double open, double close, bool terminating)
@@ -209,7 +253,9 @@ void WriterFrontend::Rotate(string rotated_path, double open, double close, bool
 		return;
 
 	FlushWriteBuffer();
-	backend->SendIn(new RotateMessage(backend, this, rotated_path, open, close, terminating));
+
+	if ( backend )
+		backend->SendIn(new RotateMessage(backend, this, rotated_path, open, close, terminating));
 	}
 
 void WriterFrontend::Finish()
@@ -218,7 +264,18 @@ void WriterFrontend::Finish()
 		return;
 
 	FlushWriteBuffer();
-	backend->SendIn(new FinishMessage(backend));
+
+	if ( backend )
+		backend->SendIn(new FinishMessage(backend));
+	}
+
+void WriterFrontend::DeleteVals(Value** vals)
+	{
+	// Note this code is duplicated in Manager::DeleteVals().
+	for ( int i = 0; i < num_fields; i++ )
+		delete vals[i];
+
+	delete [] vals;
 	}
 
 

src/logging/WriterFrontend.h

@@ -25,14 +25,21 @@ public:
 	/**
 	 * Constructor.
 	 *
-	 * type: The backend writer type, with the value corresponding to the
+	 * stream: The logging stream.
+	 *
+	 * writer: The backend writer type, with the value corresponding to the
 	 * script-level \c Log::Writer enum (e.g., \a WRITER_ASCII). The
 	 * frontend will internally instantiate a WriterBackend of the
 	 * corresponding type.
+	 * 
+	 * local: If true, the writer will instantiate a local backend.
+	 *
+	 * remote: If true, the writer will forward all data to remote
+	 * clients.
 	 *
 	 * Frontends must only be instantiated by the main thread.
 	 */
-	WriterFrontend(bro_int_t type);
+	WriterFrontend(EnumVal* stream, EnumVal* writer, bool local, bool remote);
 
 	/**
 	 * Destructor.
@@ -187,10 +194,17 @@ public:
 protected:
 	friend class Manager;
 
+	void DeleteVals(threading::Value** vals);
+
+	EnumVal* stream;
+	EnumVal* writer;
+
 	WriterBackend* backend;	// The backend we have instanatiated.
 	bool disabled;	// True if disabled.
 	bool initialized;	// True if initialized.
 	bool buf;	// True if buffering is enabled (default).
+	bool local;	// True if logging locally.
+	bool remote;	// True if loggin remotely.
 
 	string ty_name;	// Name of the backend type. Set by the manager.
 	string path;	// The log path.

src/logging/writers/Ascii.cc

@@ -177,11 +177,11 @@ bool Ascii::DoWriteOne(ODesc* desc, Value* val, const Field* field)
 		break;
 
 	case TYPE_SUBNET:
-		desc->Add(*val->val.subnet_val);
+		desc->Add(Render(val->val.subnet_val));
 		break;
 
 	case TYPE_ADDR:
-		desc->Add(*val->val.addr_val);
+		desc->Add(Render(val->val.addr_val));
 		break;
 
 	case TYPE_TIME:

src/main.cc

@@ -43,7 +43,6 @@ extern "C" void OPENSSL_add_all_algorithms_conf(void);
 #include "PersistenceSerializer.h"
 #include "EventRegistry.h"
 #include "Stats.h"
-#include "ConnCompressor.h"
 #include "DPM.h"
 #include "BroDoc.h"
 #include "Brofiler.h"
@@ -98,7 +97,6 @@ int do_notice_analysis = 0;
 int rule_bench = 0;
 int generate_documentation = 0;
 SecondaryPath* secondary_path = 0;
-ConnCompressor* conn_compressor = 0;
 extern char version[];
 char* command_line_policy = 0;
 vector<string> params;
@@ -301,7 +299,6 @@ void terminate_bro()
 	delete state_serializer;
 	delete event_registry;
 	delete secondary_path;
-	delete conn_compressor;
 	delete remote_serializer;
 	delete dpm;
 	delete log_mgr;
@@ -829,8 +826,6 @@ int main(int argc, char** argv)
 
 	delete [] script_rule_files;
 
-	conn_compressor = new ConnCompressor();
-
 	if ( g_policy_debug )
 		// ### Add support for debug command file.
 		dbg_init_debugger(0);

src/net_util.cc

@@ -147,6 +147,26 @@ char addr_to_class(uint32 addr)
 		return 'A';
 	}
 
+const char* fmt_conn_id(const IPAddr& src_addr, uint32 src_port,
+			const IPAddr& dst_addr, uint32 dst_port)
+	{
+	static char buffer[512];
+
+	safe_snprintf(buffer, sizeof(buffer), "%s:%d > %s:%d",
+			string(src_addr).c_str(), src_port,
+			string(dst_addr).c_str(), dst_port);
+
+	return buffer;
+	}
+
+const char* fmt_conn_id(const uint32* src_addr, uint32 src_port,
+			const uint32* dst_addr, uint32 dst_port)
+	{
+	IPAddr src(IPv6, src_addr, IPAddr::Network);
+	IPAddr dst(IPv6, dst_addr, IPAddr::Network);
+	return fmt_conn_id(src, src_port, dst, dst_port);
+	}
+
 uint32 extract_uint32(const u_char* data)
 	{
 	uint32 val;

src/net_util.h

@@ -5,6 +5,13 @@
 
 #include "config.h"
 
+// Define first.
+typedef enum {
+	TRANSPORT_UNKNOWN, TRANSPORT_TCP, TRANSPORT_UDP, TRANSPORT_ICMP,
+} TransportProto;
+
+typedef enum { IPv4, IPv6 } IPFamily;
+
 #include <assert.h>
 #include <sys/types.h>
 
@@ -21,7 +28,6 @@
 #include <netinet/ip_icmp.h>
 
 #include "util.h"
-#include "IPAddr.h"
 
 #ifdef HAVE_NETINET_IP6_H
 #include <netinet/ip6.h>
@@ -58,6 +64,8 @@ inline int seq_delta(uint32 a, uint32 b)
 	return int(a-b);
 	}
 
+class IPAddr;
+
 // Returns the ones-complement checksum of a chunk of b short-aligned bytes.
 extern int ones_complement_checksum(const void* p, int b, uint32 sum);
 extern int ones_complement_checksum(const IPAddr& a, uint32 sum);

src/threading/BasicThread.cc

@@ -20,8 +20,8 @@ BasicThread::BasicThread()
 	terminating = false;
 	pthread = 0;
 
-	buf = 0;
+	buf_len = 2048;
-	buf_len = 1024;
+	buf = (char*) malloc(buf_len);
 
 	name = Fmt("thread-%d", ++thread_counter);
 
@@ -57,9 +57,6 @@ void BasicThread::SetOSName(const string& name)
 
 const char* BasicThread::Fmt(const char* format, ...)
 	{
-	if ( ! buf )
-		buf = (char*) malloc(buf_len);
-
 	va_list al;
 	va_start(al, format);
 	int n = safe_vsnprintf(buf, buf_len, format, al);
@@ -67,13 +64,15 @@ const char* BasicThread::Fmt(const char* format, ...)
 
 	if ( (unsigned int) n >= buf_len )
 		{ // Not enough room, grow the buffer.
-		buf_len = n + 32;
+		int tmp_len = n + 32;
-		buf = (char*) realloc(buf, buf_len);
+		char* tmp = (char*) malloc(tmp_len);
 
 		// Is it portable to restart?
 		va_start(al, format);
-		n = safe_vsnprintf(buf, buf_len, format, al);
+		n = safe_vsnprintf(tmp, tmp_len, format, al);
 		va_end(al);
+
+		free(tmp);
 		}
 
 	return buf;

src/threading/Manager.cc

@@ -11,7 +11,7 @@ Manager::Manager()
 	did_process = true;
 	next_beat = 0;
 	terminating = false;
-	idle = false;
+	idle = true;
 
 	heart_beat_interval = double(BifConst::Threading::heart_beat_interval);
 	DBG_LOG(DBG_THREADING, "Heart beat interval set to %f", heart_beat_interval);
@@ -62,6 +62,7 @@ void Manager::AddThread(BasicThread* thread)
 	{
 	DBG_LOG(DBG_THREADING, "Adding thread %s ...", thread->Name().c_str());
 	all_threads.push_back(thread);
+	idle = false;
 	}
 
 void Manager::AddMsgThread(MsgThread* thread)
@@ -105,22 +106,25 @@ void Manager::Process()
 			next_beat = 0;
 			}
 
-		if ( ! t->HasOut() )
+		while ( t->HasOut() )
-			continue;
-
-		Message* msg = t->RetrieveOut();
-
-		if ( msg->Process() && network_time )
-			did_process = true;
-
-		else
 			{
-			string s = msg->Name() + " failed, terminating thread";
+			Message* msg = t->RetrieveOut();
-			reporter->Error("%s", s.c_str());
+
-			t->Stop();
+			if ( msg->Process() )
+				{
+				if ( network_time )
+					did_process = true;
+				}
+
+			else
+				{
+				string s = msg->Name() + " failed, terminating thread";
+				reporter->Error("%s", s.c_str());
+				t->Stop();
 			}
 
-		delete msg;
+			delete msg;
+			}
 		}
 
 //	fprintf(stderr, "P %.6f %.6f do_beat=%d did_process=%d next_next=%.6f\n", network_time, timer_mgr->Time(), do_beat, (int)did_process, next_beat);

src/threading/MsgThread.cc

@@ -281,7 +281,7 @@ void MsgThread::GetStats(Stats* stats)
 	{
 	stats->sent_in = cnt_sent_in;
 	stats->sent_out = cnt_sent_out;
-	stats->pending_in = cnt_sent_in - queue_in.Size();
+	stats->pending_in = queue_in.Size();
-	stats->pending_out = cnt_sent_out - queue_out.Size();
+	stats->pending_out = queue_out.Size();
 	}
 

src/threading/SerialTypes.cc

@@ -30,12 +30,6 @@ Value::~Value()
 	     && present )
 		delete val.string_val;
 
-	if ( type == TYPE_ADDR && present )
-		delete val.addr_val;
-
-	if ( type == TYPE_SUBNET && present )
-		delete val.subnet_val;
-
 	if ( type == TYPE_TABLE && present )
 		{
 		for ( int i = 0; i < val.set_val.size; i++ )
@@ -130,8 +124,8 @@ bool Value::Read(SerializationFormat* fmt)
 		if ( ! (fmt->Read(&val.port_val.port, "port") && fmt->Read(&proto, "proto") ) ) {
 			return false;
 		}
-		
+
-		switch (proto) {
+		switch ( proto ) {
 			case 0:
 				val.port_val.proto = TRANSPORT_UNKNOWN;
 				break;
@@ -147,20 +141,55 @@ bool Value::Read(SerializationFormat* fmt)
 			default:
 				return false;
 		}
+
 		return true;
 		}
 
-
-	case TYPE_SUBNET:
-		{
-		val.subnet_val = new IPPrefix;
-		return fmt->Read(val.subnet_val, "subnet");
-		}
-
 	case TYPE_ADDR:
 		{
-		val.addr_val = new IPAddr;
+		char family;
-		return fmt->Read(val.addr_val, "addr");
+
+		if ( ! fmt->Read(&family, "addr-family") )
+			return false;
+
+		switch ( family ) {
+		case 4:
+			val.addr_val.family = IPv4;
+			return fmt->Read(&val.addr_val.in.in4, "addr-in4");
+
+		case 6:
+			val.addr_val.family = IPv6;
+			return fmt->Read(&val.addr_val.in.in6, "addr-in6");
+
+		}
+
+		// Can't be reached.
+		abort();
+		}
+
+	case TYPE_SUBNET:
+		{
+		char length;
+		char family;
+
+		if ( ! (fmt->Read(&length, "subnet-len") && fmt->Read(&family, "subnet-family")) ) 
+			return false;
+
+		switch ( family ) {
+		case 4:
+			val.subnet_val.length = (uint8_t)length;
+			val.subnet_val.prefix.family = IPv4;
+			return fmt->Read(&val.subnet_val.prefix.in.in4, "subnet-in4");
+
+		case 6:
+			val.subnet_val.length = (uint8_t)length;
+			val.subnet_val.prefix.family = IPv6;
+			return fmt->Read(&val.subnet_val.prefix.in.in6, "subnet-in6");
+
+		}
+
+		// Can't be reached.
+		abort();
 		}
 
 	case TYPE_DOUBLE:
@@ -239,13 +268,44 @@ bool Value::Write(SerializationFormat* fmt) const
 		return fmt->Write(val.uint_val, "uint");
 
 	case TYPE_PORT:
-		return fmt->Write(val.port_val.port, "port") && fmt->Write(val.port_val.proto, "proto");	
+		return fmt->Write(val.port_val.port, "port") && fmt->Write(val.port_val.proto, "proto");
-
-	case TYPE_SUBNET:
-		return fmt->Write(*val.subnet_val, "subnet");
 
 	case TYPE_ADDR:
-		return fmt->Write(*val.addr_val, "addr");
+		{
+		switch ( val.addr_val.family ) {
+		case IPv4:
+			return fmt->Write((char)4, "addr-family")
+				&& fmt->Write(val.addr_val.in.in4, "addr-in4");
+
+		case IPv6:
+			return fmt->Write((char)6, "addr-family")
+				&& fmt->Write(val.addr_val.in.in6, "addr-in6");
+			break;
+		}
+
+		// Can't be reached.
+		abort();
+		}
+
+	case TYPE_SUBNET:
+		{
+		if ( ! fmt->Write((char)val.subnet_val.length, "subnet-length") )
+			return false;
+
+		switch ( val.subnet_val.prefix.family ) {
+		case IPv4:
+			return fmt->Write((char)4, "subnet-family")
+				&& fmt->Write(val.subnet_val.prefix.in.in4, "subnet-in4");
+
+		case IPv6:
+			return fmt->Write((char)6, "subnet-family")
+				&& fmt->Write(val.subnet_val.prefix.in.in6, "subnet-in6");
+			break;
+		}
+
+		// Can't be reached.
+		abort();
+		}
 
 	case TYPE_DOUBLE:
 	case TYPE_TIME:

src/threading/SerialTypes.h

@@ -2,10 +2,17 @@
 #ifndef THREADING_SERIALIZATIONTYPES_H
 #define THREADING_SERIALIZATIONTYPES_H
 
-#include "../RemoteSerializer.h"
-
 using namespace std;
 
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <arpa/inet.h>
+
+#include "Type.h"
+#include "net_util.h"
+
+class SerializationFormat;
+
 namespace threading {
 
 /**
@@ -62,6 +69,16 @@ struct Value {
 	typedef set_t vec_t;
 	struct port_t { bro_uint_t port; TransportProto proto; };
 
+        struct addr_t {
+		IPFamily family;
+		union {
+			struct in_addr in4;
+			struct in6_addr in6;
+		} in;
+	};
+
+	struct subnet_t { addr_t prefix; uint8_t length; };
+
 	/**
 	 * This union is a subset of BroValUnion, including only the types we
 	 * can log directly. See IsCompatibleType().
@@ -73,8 +90,8 @@ struct Value {
 		double double_val;
 		set_t set_val;
 		vec_t vector_val;
-		IPAddr* addr_val;
+		addr_t addr_val;
-		IPPrefix* subnet_val;
+		subnet_t subnet_val;
 		string* string_val;
 	} val;
 
@@ -120,6 +137,7 @@ struct Value {
 	static bool IsCompatibleType(BroType* t, bool atomic_only=false);
 
 private:
+friend class ::IPAddr;
 	Value(const Value& other)	{ } // Disabled.
 };
 

testing/Makefile

@@ -12,6 +12,7 @@ make-brief:
 	@for repo in $(DIRS); do (cd $$repo && make brief ); done
 
 coverage:
+	@for repo in $(DIRS); do (cd $$repo && echo "Coverage for '$$repo' dir:" && make coverage); done
 	@test -f btest/coverage.log && cp btest/coverage.log `mktemp brocov.tmp.XXX` || true
 	@for f in external/*/coverage.log; do test -f $$f && cp $$f `mktemp brocov.tmp.XXX` || true; done
 	@echo "Complete test suite code coverage:"

testing/btest/Baseline/core.conn-uid/output.cc

@@ -1,43 +0,0 @@
-[orig_h=141.142.220.202, orig_p=5353/udp, resp_h=224.0.0.251, resp_p=5353/udp], UWkUyAuUGXf
-[orig_h=fe80::217:f2ff:fed7:cf65, orig_p=5353/udp, resp_h=ff02::fb, resp_p=5353/udp], arKYeMETxOg
-[orig_h=141.142.220.50, orig_p=5353/udp, resp_h=224.0.0.251, resp_p=5353/udp], k6kgXLOoSKl
-[orig_h=141.142.220.118, orig_p=35634/tcp, resp_h=208.80.152.2, resp_p=80/tcp], nQcgTWjvg4c
-[orig_h=141.142.220.118, orig_p=48649/tcp, resp_h=208.80.152.118, resp_p=80/tcp], j4u32Pc5bif
-[orig_h=141.142.220.118, orig_p=48649/tcp, resp_h=208.80.152.118, resp_p=80/tcp], j4u32Pc5bif
-[orig_h=141.142.220.118, orig_p=43927/udp, resp_h=141.142.2.2, resp_p=53/udp], TEfuqmmG4bh
-[orig_h=141.142.220.118, orig_p=37676/udp, resp_h=141.142.2.2, resp_p=53/udp], FrJExwHcSal
-[orig_h=141.142.220.118, orig_p=40526/udp, resp_h=141.142.2.2, resp_p=53/udp], 5OKnoww6xl4
-[orig_h=141.142.220.118, orig_p=49996/tcp, resp_h=208.80.152.3, resp_p=80/tcp], 3PKsZ2Uye21
-[orig_h=141.142.220.118, orig_p=49997/tcp, resp_h=208.80.152.3, resp_p=80/tcp], VW0XPVINV8a
-[orig_h=141.142.220.118, orig_p=32902/udp, resp_h=141.142.2.2, resp_p=53/udp], fRFu0wcOle6
-[orig_h=141.142.220.118, orig_p=59816/udp, resp_h=141.142.2.2, resp_p=53/udp], qSsw6ESzHV4
-[orig_h=141.142.220.118, orig_p=59714/udp, resp_h=141.142.2.2, resp_p=53/udp], iE6yhOq3SF
-[orig_h=141.142.220.118, orig_p=49998/tcp, resp_h=208.80.152.3, resp_p=80/tcp], GSxOnSLghOa
-[orig_h=141.142.220.118, orig_p=58206/udp, resp_h=141.142.2.2, resp_p=53/udp], qCaWGmzFtM5
-[orig_h=141.142.220.118, orig_p=38911/udp, resp_h=141.142.2.2, resp_p=53/udp], 70MGiRM1Qf4
-[orig_h=141.142.220.118, orig_p=59746/udp, resp_h=141.142.2.2, resp_p=53/udp], h5DsfNtYzi1
-[orig_h=141.142.220.118, orig_p=49999/tcp, resp_h=208.80.152.3, resp_p=80/tcp], P654jzLoe3a
-[orig_h=141.142.220.118, orig_p=50000/tcp, resp_h=208.80.152.3, resp_p=80/tcp], Tw8jXtpTGu6
-[orig_h=141.142.220.118, orig_p=45000/udp, resp_h=141.142.2.2, resp_p=53/udp], c4Zw9TmAE05
-[orig_h=141.142.220.118, orig_p=48479/udp, resp_h=141.142.2.2, resp_p=53/udp], EAr0uf4mhq
-[orig_h=141.142.220.118, orig_p=48128/udp, resp_h=141.142.2.2, resp_p=53/udp], GvmoxJFXdTa
-[orig_h=141.142.220.118, orig_p=50001/tcp, resp_h=208.80.152.3, resp_p=80/tcp], 0Q4FH8sESw5
-[orig_h=141.142.220.118, orig_p=56056/udp, resp_h=141.142.2.2, resp_p=53/udp], slFea8xwSmb
-[orig_h=141.142.220.118, orig_p=55092/udp, resp_h=141.142.2.2, resp_p=53/udp], UfGkYA2HI2g
-[orig_h=141.142.220.118, orig_p=35642/tcp, resp_h=208.80.152.2, resp_p=80/tcp], i2rO3KD1Syg
-[orig_h=141.142.220.118, orig_p=49997/tcp, resp_h=208.80.152.3, resp_p=80/tcp], VW0XPVINV8a
-[orig_h=141.142.220.118, orig_p=49996/tcp, resp_h=208.80.152.3, resp_p=80/tcp], 3PKsZ2Uye21
-[orig_h=141.142.220.118, orig_p=49998/tcp, resp_h=208.80.152.3, resp_p=80/tcp], GSxOnSLghOa
-[orig_h=141.142.220.118, orig_p=50000/tcp, resp_h=208.80.152.3, resp_p=80/tcp], Tw8jXtpTGu6
-[orig_h=141.142.220.118, orig_p=49999/tcp, resp_h=208.80.152.3, resp_p=80/tcp], P654jzLoe3a
-[orig_h=141.142.220.118, orig_p=50001/tcp, resp_h=208.80.152.3, resp_p=80/tcp], 0Q4FH8sESw5
-[orig_h=141.142.220.118, orig_p=35642/tcp, resp_h=208.80.152.2, resp_p=80/tcp], i2rO3KD1Syg
-[orig_h=141.142.220.235, orig_p=6705/tcp, resp_h=173.192.163.128, resp_p=80/tcp], 2cx26uAvUPl
-[orig_h=141.142.220.235, orig_p=6705/tcp, resp_h=173.192.163.128, resp_p=80/tcp], 2cx26uAvUPl
-[orig_h=141.142.220.44, orig_p=5353/udp, resp_h=224.0.0.251, resp_p=5353/udp], BWaU4aSuwkc
-[orig_h=141.142.220.226, orig_p=137/udp, resp_h=141.142.220.255, resp_p=137/udp], 10XodEwRycf
-[orig_h=fe80::3074:17d5:2052:c324, orig_p=65373/udp, resp_h=ff02::1:3, resp_p=5355/udp], zno26fFZkrh
-[orig_h=141.142.220.226, orig_p=55131/udp, resp_h=224.0.0.252, resp_p=5355/udp], v5rgkJBig5l
-[orig_h=fe80::3074:17d5:2052:c324, orig_p=54213/udp, resp_h=ff02::1:3, resp_p=5355/udp], eWZCH7OONC1
-[orig_h=141.142.220.226, orig_p=55671/udp, resp_h=224.0.0.252, resp_p=5355/udp], 0Pwk3ntf8O3
-[orig_h=141.142.220.238, orig_p=56641/udp, resp_h=141.142.220.255, resp_p=137/udp], 0HKorjr8Zp7

testing/btest/Baseline/core.conn-uid/output.cc2

@@ -1,43 +0,0 @@
-[orig_h=141.142.220.202, orig_p=5353/udp, resp_h=224.0.0.251, resp_p=5353/udp], UWkUyAuUGXf
-[orig_h=fe80::217:f2ff:fed7:cf65, orig_p=5353/udp, resp_h=ff02::fb, resp_p=5353/udp], arKYeMETxOg
-[orig_h=141.142.220.50, orig_p=5353/udp, resp_h=224.0.0.251, resp_p=5353/udp], k6kgXLOoSKl
-[orig_h=141.142.220.118, orig_p=35634/tcp, resp_h=208.80.152.2, resp_p=80/tcp], nQcgTWjvg4c
-[orig_h=141.142.220.118, orig_p=48649/tcp, resp_h=208.80.152.118, resp_p=80/tcp], j4u32Pc5bif
-[orig_h=141.142.220.118, orig_p=48649/tcp, resp_h=208.80.152.118, resp_p=80/tcp], j4u32Pc5bif
-[orig_h=141.142.220.118, orig_p=43927/udp, resp_h=141.142.2.2, resp_p=53/udp], TEfuqmmG4bh
-[orig_h=141.142.220.118, orig_p=37676/udp, resp_h=141.142.2.2, resp_p=53/udp], FrJExwHcSal
-[orig_h=141.142.220.118, orig_p=40526/udp, resp_h=141.142.2.2, resp_p=53/udp], 5OKnoww6xl4
-[orig_h=141.142.220.118, orig_p=49996/tcp, resp_h=208.80.152.3, resp_p=80/tcp], 3PKsZ2Uye21
-[orig_h=141.142.220.118, orig_p=49997/tcp, resp_h=208.80.152.3, resp_p=80/tcp], VW0XPVINV8a
-[orig_h=141.142.220.118, orig_p=32902/udp, resp_h=141.142.2.2, resp_p=53/udp], fRFu0wcOle6
-[orig_h=141.142.220.118, orig_p=59816/udp, resp_h=141.142.2.2, resp_p=53/udp], qSsw6ESzHV4
-[orig_h=141.142.220.118, orig_p=59714/udp, resp_h=141.142.2.2, resp_p=53/udp], iE6yhOq3SF
-[orig_h=141.142.220.118, orig_p=49998/tcp, resp_h=208.80.152.3, resp_p=80/tcp], GSxOnSLghOa
-[orig_h=141.142.220.118, orig_p=58206/udp, resp_h=141.142.2.2, resp_p=53/udp], qCaWGmzFtM5
-[orig_h=141.142.220.118, orig_p=38911/udp, resp_h=141.142.2.2, resp_p=53/udp], 70MGiRM1Qf4
-[orig_h=141.142.220.118, orig_p=59746/udp, resp_h=141.142.2.2, resp_p=53/udp], h5DsfNtYzi1
-[orig_h=141.142.220.118, orig_p=49999/tcp, resp_h=208.80.152.3, resp_p=80/tcp], P654jzLoe3a
-[orig_h=141.142.220.118, orig_p=50000/tcp, resp_h=208.80.152.3, resp_p=80/tcp], Tw8jXtpTGu6
-[orig_h=141.142.220.118, orig_p=45000/udp, resp_h=141.142.2.2, resp_p=53/udp], c4Zw9TmAE05
-[orig_h=141.142.220.118, orig_p=48479/udp, resp_h=141.142.2.2, resp_p=53/udp], EAr0uf4mhq
-[orig_h=141.142.220.118, orig_p=48128/udp, resp_h=141.142.2.2, resp_p=53/udp], GvmoxJFXdTa
-[orig_h=141.142.220.118, orig_p=50001/tcp, resp_h=208.80.152.3, resp_p=80/tcp], 0Q4FH8sESw5
-[orig_h=141.142.220.118, orig_p=56056/udp, resp_h=141.142.2.2, resp_p=53/udp], slFea8xwSmb
-[orig_h=141.142.220.118, orig_p=55092/udp, resp_h=141.142.2.2, resp_p=53/udp], UfGkYA2HI2g
-[orig_h=141.142.220.118, orig_p=35642/tcp, resp_h=208.80.152.2, resp_p=80/tcp], i2rO3KD1Syg
-[orig_h=141.142.220.118, orig_p=49997/tcp, resp_h=208.80.152.3, resp_p=80/tcp], VW0XPVINV8a
-[orig_h=141.142.220.118, orig_p=49996/tcp, resp_h=208.80.152.3, resp_p=80/tcp], 3PKsZ2Uye21
-[orig_h=141.142.220.118, orig_p=49998/tcp, resp_h=208.80.152.3, resp_p=80/tcp], GSxOnSLghOa
-[orig_h=141.142.220.118, orig_p=50000/tcp, resp_h=208.80.152.3, resp_p=80/tcp], Tw8jXtpTGu6
-[orig_h=141.142.220.118, orig_p=49999/tcp, resp_h=208.80.152.3, resp_p=80/tcp], P654jzLoe3a
-[orig_h=141.142.220.118, orig_p=50001/tcp, resp_h=208.80.152.3, resp_p=80/tcp], 0Q4FH8sESw5
-[orig_h=141.142.220.118, orig_p=35642/tcp, resp_h=208.80.152.2, resp_p=80/tcp], i2rO3KD1Syg
-[orig_h=141.142.220.235, orig_p=6705/tcp, resp_h=173.192.163.128, resp_p=80/tcp], 2cx26uAvUPl
-[orig_h=141.142.220.235, orig_p=6705/tcp, resp_h=173.192.163.128, resp_p=80/tcp], 2cx26uAvUPl
-[orig_h=141.142.220.44, orig_p=5353/udp, resp_h=224.0.0.251, resp_p=5353/udp], BWaU4aSuwkc
-[orig_h=141.142.220.226, orig_p=137/udp, resp_h=141.142.220.255, resp_p=137/udp], 10XodEwRycf
-[orig_h=fe80::3074:17d5:2052:c324, orig_p=65373/udp, resp_h=ff02::1:3, resp_p=5355/udp], zno26fFZkrh
-[orig_h=141.142.220.226, orig_p=55131/udp, resp_h=224.0.0.252, resp_p=5355/udp], v5rgkJBig5l
-[orig_h=fe80::3074:17d5:2052:c324, orig_p=54213/udp, resp_h=ff02::1:3, resp_p=5355/udp], eWZCH7OONC1
-[orig_h=141.142.220.226, orig_p=55671/udp, resp_h=224.0.0.252, resp_p=5355/udp], 0Pwk3ntf8O3
-[orig_h=141.142.220.238, orig_p=56641/udp, resp_h=141.142.220.255, resp_p=137/udp], 0HKorjr8Zp7

testing/btest/Baseline/core.leaks.basic-cluster/manager-1.metrics.log

@@ -0,0 +1,10 @@
+#separator \x09
+#set_separator	,
+#empty_field	(empty)
+#unset_field	-
+#path	metrics
+#fields	ts	metric_id	filter_name	index.host	index.str	index.network	value
+#types	time	enum	string	addr	string	subnet	count
+1331256494.591966	TEST_METRIC	foo-bar	6.5.4.3	-	-	4
+1331256494.591966	TEST_METRIC	foo-bar	7.2.1.5	-	-	2
+1331256494.591966	TEST_METRIC	foo-bar	1.2.3.4	-	-	6

testing/btest/Baseline/core.leaks.remote/sender.test.failure.log

@@ -0,0 +1,10 @@
+#separator \x09
+#set_separator	,
+#empty_field	(empty)
+#unset_field	-
+#path	test.failure
+#fields	t	id.orig_h	id.orig_p	id.resp_h	id.resp_p	status	country
+#types	time	addr	port	addr	port	string	string
+1331256472.375609	1.2.3.4	1234	2.3.4.5	80	failure	US
+1331256472.375609	1.2.3.4	1234	2.3.4.5	80	failure	UK
+1331256472.375609	1.2.3.4	1234	2.3.4.5	80	failure	MX

testing/btest/Baseline/core.leaks.remote/sender.test.log

@@ -0,0 +1,12 @@
+#separator \x09
+#set_separator	,
+#empty_field	(empty)
+#unset_field	-
+#path	test
+#fields	t	id.orig_h	id.orig_p	id.resp_h	id.resp_p	status	country
+#types	time	addr	port	addr	port	string	string
+1331256472.375609	1.2.3.4	1234	2.3.4.5	80	success	unknown
+1331256472.375609	1.2.3.4	1234	2.3.4.5	80	failure	US
+1331256472.375609	1.2.3.4	1234	2.3.4.5	80	failure	UK
+1331256472.375609	1.2.3.4	1234	2.3.4.5	80	success	BR
+1331256472.375609	1.2.3.4	1234	2.3.4.5	80	failure	MX

testing/btest/Baseline/core.leaks.remote/sender.test.success.log

@@ -0,0 +1,9 @@
+#separator \x09
+#set_separator	,
+#empty_field	(empty)
+#unset_field	-
+#path	test.success
+#fields	t	id.orig_h	id.orig_p	id.resp_h	id.resp_p	status	country
+#types	time	addr	port	addr	port	string	string
+1331256472.375609	1.2.3.4	1234	2.3.4.5	80	success	unknown
+1331256472.375609	1.2.3.4	1234	2.3.4.5	80	success	BR

testing/btest/Baseline/istate.broccoli-ipv6/bro..stdout

@@ -0,0 +1,9 @@
+handshake done with peer
+bro_addr(1.2.3.4)
+bro_subnet(10.0.0.0/16)
+bro_addr(2607:f8b0:4009:802::1014)
+bro_subnet(2607:f8b0::/32)
+broccoli_addr(1.2.3.4)
+broccoli_subnet(10.0.0.0/16)
+broccoli_addr(2607:f8b0:4009:802::1014)
+broccoli_subnet(2607:f8b0::/32)

testing/btest/Baseline/istate.broccoli-ipv6/broccoli..stdout

@@ -0,0 +1,6 @@
+Connected to Bro instance at: localhost:47757
+Received bro_addr(1.2.3.4)
+Received bro_subnet(10.0.0.0/16)
+Received bro_addr(2607:f8b0:4009:802::1014)
+Received bro_subnet(2607:f8b0::/32)
+Terminating

testing/btest/Baseline/istate.pybroccoli/bro..stdout

@@ -1,14 +1,16 @@
 ==== atomic
 -10
 2
-1313624487.48817
+1330035434.516896
 2.0 mins
 F
 1.5
 Servus
 5555/tcp
 6.7.6.5
+2001:db8:85a3::8a2e:370:7334
 192.168.0.0/16
+2001:db8:85a3::/48
 ==== record
 [a=42, b=6.6.7.7]
 42, 6.6.7.7

testing/btest/Baseline/istate.pybroccoli/python..stdout.filtered

@@ -1,7 +1,7 @@
 ==== atomic a 1 ====
 -4L -4
 42 42
-1313624487.4889
+1330035434.5180
 60.0
 True True
 3.14
@@ -9,10 +9,12 @@ True True
 '12345/udp' 12345/udp
 '1.2.3.4' 1.2.3.4
 '22.33.44.0/24' 22.33.44.0/24
+'2607:f8b0:4009:802::1014' 2607:f8b0:4009:802::1014
+'2607:f8b0::/32' 2607:f8b0::/32
 ==== atomic a 2 ====
 -10L -10
 2 2
-1313624487.4882
+1330035434.5169
 120.0
 False False
 1.5
@@ -20,10 +22,12 @@ False False
 '5555/tcp' 5555/tcp
 '6.7.6.5' 6.7.6.5
 '192.168.0.0/16' 192.168.0.0/16
+'2001:db8:85a3::8a2e:370:7334' 2001:db8:85a3::8a2e:370:7334
+'2001:db8:85a3::/48' 2001:db8:85a3::/48
 ==== atomic b 2 ====
 -10L -10
 <broccoli.count instance at > 2
-<broccoli.time instance at > 1313624487.4882
+<broccoli.time instance at > 1330035434.5169
 <broccoli.interval instance at > 120.0
 False False
 1.5
@@ -31,6 +35,8 @@ False False
 <broccoli.port instance at > 5555/tcp
 <broccoli.addr instance at > 6.7.6.5
 <broccoli.subnet instance at > 192.168.0.0/16
+<broccoli.addr instance at > 2001:db8:85a3::8a2e:370:7334
+<broccoli.subnet instance at > 2001:db8:85a3::/48
 ==== record 1 ====
 <broccoli.record instance at >
 42L 42

testing/btest/Makefile

@@ -2,16 +2,23 @@
 DIAG=diag.log
 BTEST=../../aux/btest/btest
 
-all: cleanup
+all: cleanup btest-verbose coverage
-    # Showing all tests.
-	@$(BTEST) -f $(DIAG)
-	@../scripts/coverage-calc ".tmp/script-coverage*" coverage.log `pwd`/../../scripts
 
-brief: cleanup
+# Showing all tests.
-    # Brief output showing only failed tests.
+btest-verbose:
+	@$(BTEST) -f $(DIAG)
+
+brief: cleanup btest-brief coverage
+
+# Brief output showing only failed tests.
+btest-brief:
 	@$(BTEST) -b -f $(DIAG)
+
+coverage:
 	@../scripts/coverage-calc ".tmp/script-coverage*" coverage.log `pwd`/../../scripts
 
 cleanup:
 	@rm -f $(DIAG)
 	@rm -f .tmp/script-coverage*
+
+.PHONY: all btest-verbose brief btest-brief coverage cleanup

testing/btest/analyzers/conn-size.bro

@@ -1,2 +1,2 @@
-# @TEST-EXEC: bro -C -r ${TRACES}/conn-size.trace tcp udp icmp report_conn_size_analyzer=T use_connection_compressor=F
+# @TEST-EXEC: bro -C -r ${TRACES}/conn-size.trace tcp udp icmp report_conn_size_analyzer=T
 # @TEST-EXEC: btest-diff conn.log

testing/btest/btest.cfg

@@ -18,4 +18,4 @@ DIST=%(testbase)s/../..
 BUILD=%(testbase)s/../../build
 TEST_DIFF_CANONIFIER=$SCRIPTS/diff-canonifier
 TMPDIR=%(testbase)s/.tmp
-BRO_PROFILER_FILE=%(testbase)s/.tmp/script-coverage
+BRO_PROFILER_FILE=%(testbase)s/.tmp/script-coverage.XXXXXX

testing/btest/core/conn-uid.bro

@@ -9,17 +9,6 @@
 # @TEST-EXEC: unset BRO_SEED_FILE &&  bro -C -r $TRACES/wikipedia.trace %INPUT >output2
 # @TEST-EXEC: cat output output2 | sort | uniq -c | wc -l | sed 's/ //g' >counts
 # @TEST-EXEC: btest-diff counts
-#
-# Make sure it works without the connection compressor as well.
-#
-# @TEST-EXEC: bro -C -r $TRACES/wikipedia.trace %INPUT use_connection_compressor=F >output.cc
-# @TEST-EXEC: btest-diff output.cc
-#
-# Make sure it works with the full connection compressor as well.
-#
-# @TEST-EXEC: bro -C -r $TRACES/wikipedia.trace %INPUT cc_handle_only_syns=F >output.cc2
-# @TEST-EXEC: btest-diff output.cc2
-
 
 event new_connection(c: connection)
 	{

testing/btest/core/leaks/basic-cluster.bro

@@ -0,0 +1,39 @@
+# Needs perftools support.
+#
+# @TEST-REQUIRES: bro  --help 2>&1 | grep -q mem-leaks
+# @TEST-EXEC: btest-bg-run manager-1 HEAP_CHECK_DUMP_DIRECTORY=. HEAPCHECK=local BROPATH=$BROPATH:.. CLUSTER_NODE=manager-1 bro -m %INPUT
+# @TEST-EXEC: btest-bg-run proxy-1   BROPATH=$BROPATH:.. CLUSTER_NODE=proxy-1 bro %INPUT
+# @TEST-EXEC: sleep 1
+# @TEST-EXEC: btest-bg-run worker-1  HEAP_CHECK_DUMP_DIRECTORY=. HEAPCHECK=local BROPATH=$BROPATH:.. CLUSTER_NODE=worker-1 bro -m -r $TRACES/web.trace --pseudo-realtime %INPUT
+# @TEST-EXEC: btest-bg-run worker-2  HEAP_CHECK_DUMP_DIRECTORY=. HEAPCHECK=local BROPATH=$BROPATH:.. CLUSTER_NODE=worker-2 bro -m -r $TRACES/web.trace --pseudo-realtime %INPUT
+# @TEST-EXEC: btest-bg-wait -k 30
+# @TEST-EXEC: btest-diff manager-1/metrics.log
+
+@TEST-START-FILE cluster-layout.bro
+redef Cluster::nodes = {
+	["manager-1"] = [$node_type=Cluster::MANAGER, $ip=127.0.0.1, $p=37757/tcp, $workers=set("worker-1")],
+	["proxy-1"] = [$node_type=Cluster::PROXY,     $ip=127.0.0.1, $p=37758/tcp, $manager="manager-1", $workers=set("worker-1")],
+	["worker-1"] = [$node_type=Cluster::WORKER,   $ip=127.0.0.1, $p=37760/tcp, $manager="manager-1", $proxy="proxy-1", $interface="eth0"],
+	["worker-2"] = [$node_type=Cluster::WORKER,   $ip=127.0.0.1, $p=37761/tcp, $manager="manager-1", $proxy="proxy-1", $interface="eth1"],
+};
+@TEST-END-FILE
+
+redef Log::default_rotation_interval = 0secs;
+
+redef enum Metrics::ID += {
+	TEST_METRIC,
+};
+
+event bro_init() &priority=5
+	{
+	Metrics::add_filter(TEST_METRIC, 
+		[$name="foo-bar",
+		 $break_interval=3secs]);
+		
+	if ( Cluster::local_node_type() == Cluster::WORKER )
+		{
+		Metrics::add_data(TEST_METRIC, [$host=1.2.3.4], 3);
+		Metrics::add_data(TEST_METRIC, [$host=6.5.4.3], 2);
+		Metrics::add_data(TEST_METRIC, [$host=7.2.1.5], 1);
+		}
+	}

testing/btest/core/leaks/dns.bro

@@ -1,5 +1,7 @@
 # Needs perftools support.
 #
+# @TEST-GROUP: leaks
+#
 # @TEST-REQUIRES: bro  --help 2>&1 | grep -q mem-leaks
 #
 # @TEST-EXEC: HEAP_CHECK_DUMP_DIRECTORY=. HEAPCHECK=local bro -m -r $TRACES/wikipedia.trace %INPUT

testing/btest/core/leaks/remote.bro

@@ -0,0 +1,79 @@
+# Needs perftools support.
+#
+# @TEST-REQUIRES: bro  --help 2>&1 | grep -q mem-leaks
+#
+# @TEST-EXEC: btest-bg-run sender HEAP_CHECK_DUMP_DIRECTORY=. HEAPCHECK=local bro -m --pseudo-realtime %INPUT ../sender.bro
+# @TEST-EXEC: sleep 1
+# @TEST-EXEC: btest-bg-run receiver HEAP_CHECK_DUMP_DIRECTORY=. HEAPCHECK=local bro -m --pseudo-realtime %INPUT ../receiver.bro
+# @TEST-EXEC: sleep 1
+# @TEST-EXEC: btest-bg-wait -k 10
+# @TEST-EXEC: btest-diff sender/test.log
+# @TEST-EXEC: btest-diff sender/test.failure.log
+# @TEST-EXEC: btest-diff sender/test.success.log
+# @TEST-EXEC: cmp receiver/test.log sender/test.log
+# @TEST-EXEC: cmp receiver/test.failure.log sender/test.failure.log
+# @TEST-EXEC: cmp receiver/test.success.log sender/test.success.log
+
+# This is the common part loaded by both sender and receiver.
+module Test;
+
+export {
+	# Create a new ID for our log stream
+	redef enum Log::ID += { LOG };
+
+	# Define a record with all the columns the log file can have.
+	# (I'm using a subset of fields from ssh-ext for demonstration.)
+	type Log: record {
+		t: time;
+		id: conn_id; # Will be rolled out into individual columns.
+		status: string &optional;
+		country: string &default="unknown";
+	} &log;
+}
+
+event bro_init()
+{
+	Log::create_stream(Test::LOG, [$columns=Log]);
+	Log::add_filter(Test::LOG, [$name="f1", $path="test.success", $pred=function(rec: Log): bool { return rec$status == "success"; }]);
+}
+
+#####
+
+@TEST-START-FILE sender.bro
+
+module Test;
+
+@load frameworks/communication/listen
+
+function fail(rec: Log): bool
+	{
+	return rec$status != "success";
+	}
+
+event remote_connection_handshake_done(p: event_peer)
+	{
+	Log::add_filter(Test::LOG, [$name="f2", $path="test.failure", $pred=fail]);
+
+	local cid = [$orig_h=1.2.3.4, $orig_p=1234/tcp, $resp_h=2.3.4.5, $resp_p=80/tcp];
+
+	local r: Log = [$t=network_time(), $id=cid, $status="success"];
+
+	# Log something.
+	Log::write(Test::LOG, r);
+	Log::write(Test::LOG, [$t=network_time(), $id=cid, $status="failure", $country="US"]);
+	Log::write(Test::LOG, [$t=network_time(), $id=cid, $status="failure", $country="UK"]);
+	Log::write(Test::LOG, [$t=network_time(), $id=cid, $status="success", $country="BR"]);
+	Log::write(Test::LOG, [$t=network_time(), $id=cid, $status="failure", $country="MX"]);
+	disconnect(p);
+	}
+@TEST-END-FILE
+
+@TEST-START-FILE receiver.bro
+
+#####
+
+redef Communication::nodes += {
+    ["foo"] = [$host = 127.0.0.1, $connect=T, $request_logs=T]
+};
+
+@TEST-END-FILE

testing/btest/core/leaks/test-all.bro

@@ -1,5 +1,7 @@
 # Needs perftools support.
 #
+# @TEST-GROUP: leaks
+#
 # @TEST-REQUIRES: bro  --help 2>&1 | grep -q mem-leaks
 #
 # @TEST-EXEC: HEAP_CHECK_DUMP_DIRECTORY=. HEAPCHECK=local bro -m -r $TRACES/wikipedia.trace test-all-policy

testing/btest/istate/broccoli-ipv6.bro

@@ -0,0 +1,15 @@
+# @TEST-GROUP: comm
+#
+# @TEST-REQUIRES: test -e $BUILD/aux/broccoli/src/libbroccoli.so || test -e $BUILD/aux/broccoli/src/libbroccoli.dylib
+#
+# @TEST-EXEC: btest-bg-run bro bro %INPUT $DIST/aux/broccoli/test/broccoli-v6addrs.bro
+# @TEST-EXEC: btest-bg-run broccoli $BUILD/aux/broccoli/test/broccoli-v6addrs
+# @TEST-EXEC: btest-bg-wait -k 20
+# @TEST-EXEC: btest-diff bro/.stdout
+# @TEST-EXEC: btest-diff broccoli/.stdout
+
+event remote_connection_closed(p: event_peer)
+	{
+	terminate();
+	}
+

testing/btest/istate/broccoli.bro

@@ -1,3 +1,5 @@
+# @TEST-GROUP: comm
+#
 # @TEST-REQUIRES: test -e $BUILD/aux/broccoli/src/libbroccoli.so || test -e $BUILD/aux/broccoli/src/libbroccoli.dylib
 #
 # @TEST-EXEC: btest-bg-run bro bro %INPUT $DIST/aux/broccoli/test/broping-record.bro

testing/btest/istate/events-ssl.bro

@@ -1,4 +1,5 @@
-# 
+# @TEST-GROUP: comm
+#
 # @TEST-EXEC: btest-bg-run sender   bro -C -r $TRACES/web.trace --pseudo-realtime ../sender.bro
 # @TEST-EXEC: btest-bg-run receiver bro ../receiver.bro
 # @TEST-EXEC: btest-bg-wait -k 20

testing/btest/istate/events.bro

@@ -1,4 +1,5 @@
-# 
+# @TEST-GROUP: comm
+#
 # @TEST-EXEC: btest-bg-run sender   bro -C -r $TRACES/web.trace --pseudo-realtime ../sender.bro
 # @TEST-EXEC: btest-bg-run receiver bro ../receiver.bro
 # @TEST-EXEC: btest-bg-wait -k 20

testing/btest/istate/pybroccoli.py

@@ -1,3 +1,5 @@
+# @TEST-GROUP: comm
+#
 # @TEST-REQUIRES: test -e $BUILD/aux/broccoli/src/libbroccoli.so || test -e $BUILD/aux/broccoli/src/libbroccoli.dylib
 # @TEST-REQUIRES: test -e $BUILD/aux/broccoli/bindings/broccoli-python/_broccoli_intern.so
 #

testing/btest/istate/sync.bro

@@ -1,3 +1,4 @@
+# @TEST-GROUP: comm
 #
 # @TEST-EXEC: btest-bg-run sender   bro %INPUT ../sender.bro
 # @TEST-EXEC: btest-bg-run receiver bro %INPUT ../receiver.bro

testing/btest/scripts/base/frameworks/cluster/start-it-up.bro

@@ -1,3 +1,5 @@
+# @TEST-GROUP: comm
+#
 # @TEST-EXEC: btest-bg-run manager-1 BROPATH=$BROPATH:.. CLUSTER_NODE=manager-1 bro %INPUT
 # @TEST-EXEC: btest-bg-run proxy-1   BROPATH=$BROPATH:.. CLUSTER_NODE=proxy-1 bro %INPUT
 # @TEST-EXEC: btest-bg-run proxy-2   BROPATH=$BROPATH:.. CLUSTER_NODE=proxy-2 bro %INPUT

testing/btest/scripts/base/frameworks/communication/communication_log_baseline.bro

@@ -1,4 +1,5 @@
-# 
+# @TEST-GROUP: comm
+#
 # @TEST-EXEC: btest-bg-run receiver bro -b ../receiver.bro
 # @TEST-EXEC: btest-bg-run sender   bro -b ../sender.bro
 # @TEST-EXEC: btest-bg-wait -k 10

testing/btest/scripts/base/frameworks/control/configuration_update.bro

@@ -1,3 +1,5 @@
+# @TEST-GROUP: comm
+#
 # @TEST-EXEC: btest-bg-run controllee  BROPATH=$BROPATH:.. bro %INPUT frameworks/control/controllee Communication::listen_port=65531/tcp 
 # @TEST-EXEC: btest-bg-run controller  BROPATH=$BROPATH:.. bro %INPUT test-redef frameworks/control/controller Control::host=127.0.0.1 Control::host_port=65531/tcp Control::cmd=configuration_update
 # @TEST-EXEC: btest-bg-run controller2 BROPATH=$BROPATH:.. bro %INPUT frameworks/control/controller Control::host=127.0.0.1 Control::host_port=65531/tcp Control::cmd=shutdown

testing/btest/scripts/base/frameworks/control/id_value.bro

@@ -1,3 +1,5 @@
+# @TEST-GROUP: comm
+#
 # @TEST-EXEC: btest-bg-run controllee  BROPATH=$BROPATH:.. bro %INPUT only-for-controllee frameworks/control/controllee Communication::listen_port=65532/tcp 
 # @TEST-EXEC: btest-bg-run controller  BROPATH=$BROPATH:.. bro %INPUT frameworks/control/controller Control::host=127.0.0.1 Control::host_port=65532/tcp Control::cmd=id_value Control::arg=test_var
 # @TEST-EXEC: btest-bg-wait -k 10

testing/btest/scripts/base/frameworks/control/shutdown.bro

@@ -1,3 +1,5 @@
+# @TEST-GROUP: comm
+#
 # @TEST-EXEC: btest-bg-run controllee BROPATH=$BROPATH:.. bro %INPUT frameworks/control/controllee Communication::listen_port=65530/tcp 
 # @TEST-EXEC: btest-bg-run controller BROPATH=$BROPATH:.. bro %INPUT frameworks/control/controller Control::host=127.0.0.1 Control::host_port=65530/tcp Control::cmd=shutdown
 # @TEST-EXEC: btest-bg-wait 10

testing/btest/scripts/base/frameworks/logging/remote-types.bro

@@ -1,3 +1,4 @@
+# @TEST-GROUP: comm
 #
 # @TEST-EXEC: btest-bg-run sender bro --pseudo-realtime %INPUT ../sender.bro
 # @TEST-EXEC: btest-bg-run receiver bro --pseudo-realtime %INPUT ../receiver.bro

testing/btest/scripts/base/frameworks/logging/remote.bro

@@ -1,3 +1,4 @@
+# @TEST-GROUP: comm
 #
 # @TEST-EXEC: btest-bg-run sender bro --pseudo-realtime %INPUT ../sender.bro
 # @TEST-EXEC: sleep 1

testing/btest/scripts/base/frameworks/metrics/basic-cluster.bro

@@ -1,3 +1,5 @@
+# @TEST-GROUP: comm
+#
 # @TEST-EXEC: btest-bg-run manager-1 BROPATH=$BROPATH:.. CLUSTER_NODE=manager-1 bro %INPUT
 # @TEST-EXEC: btest-bg-run proxy-1   BROPATH=$BROPATH:.. CLUSTER_NODE=proxy-1 bro %INPUT
 # @TEST-EXEC: sleep 1

testing/btest/scripts/base/frameworks/metrics/cluster-intermediate-update.bro

@@ -1,3 +1,5 @@
+# @TEST-GROUP: comm
+#
 # @TEST-EXEC: btest-bg-run manager-1 BROPATH=$BROPATH:.. CLUSTER_NODE=manager-1 bro %INPUT
 # @TEST-EXEC: btest-bg-run proxy-1   BROPATH=$BROPATH:.. CLUSTER_NODE=proxy-1 bro %INPUT
 # @TEST-EXEC: sleep 1

testing/btest/scripts/base/frameworks/notice/cluster.bro

@@ -1,3 +1,5 @@
+# @TEST-GROUP: comm
+#
 # @TEST-EXEC: btest-bg-run manager-1 BROPATH=$BROPATH:.. CLUSTER_NODE=manager-1 bro %INPUT
 # @TEST-EXEC: btest-bg-run proxy-1   BROPATH=$BROPATH:.. CLUSTER_NODE=proxy-1 bro %INPUT
 # @TEST-EXEC: sleep 1

testing/btest/scripts/base/frameworks/notice/suppression-cluster.bro

@@ -1,3 +1,5 @@
+# @TEST-GROUP: comm
+#
 # @TEST-EXEC: btest-bg-run manager-1 BROPATH=$BROPATH:.. CLUSTER_NODE=manager-1 bro %INPUT
 # @TEST-EXEC: btest-bg-run proxy-1   BROPATH=$BROPATH:.. CLUSTER_NODE=proxy-1 bro %INPUT
 # @TEST-EXEC: sleep 1

testing/external/Makefile

@@ -24,3 +24,7 @@ push:
 status:
 	@for repo in $(REPOS); do ( cd $$repo && echo '>>' $$repo && git status -bs && echo ); done
 
+coverage:
+	@for repo in $(REPOS); do (cd $$repo && echo "Coverage for '$$repo' repo:" && make coverage); done
+
+.PHONY: all brief init pull push status coverage

testing/external/scripts/perftools-adapt-paths

@@ -7,4 +7,4 @@
 
 cat $1 | sed "s#bro *\"\./#../../../build/src/bro \".tmp/$TEST_NAME/#g" | sed 's/ *--gv//g' >$1.tmp && mv $1.tmp $1
 
-grep -q "No leaks found" $1
+grep -qv "detected leaks of" $1

testing/external/subdir-btest.cfg

@@ -17,4 +17,4 @@ TRACES=%(testbase)s/Traces
 SCRIPTS=%(testbase)s/../scripts
 DIST=%(testbase)s/../../..
 BUILD=%(testbase)s/../../../build
-BRO_PROFILER_FILE=%(testbase)s/.tmp/script-coverage
+BRO_PROFILER_FILE=%(testbase)s/.tmp/script-coverage.XXXXXX

testing/scripts/btest-bg-run

@@ -1,7 +0,0 @@
-#! /usr/bin/env bash
-
-# This is a wrapper script to btest's real btest-bg-run.  It's used
-# when collecting Bro script coverage statistics so that two independent
-# Bro processing don't try to write those usage statistics to the same file.
-
-BRO_PROFILER_FILE=`mktemp $TMPDIR/script-coverage.XXXX` $BTEST_PATH/btest-bg-run $@

testing/scripts/diff-canonifier-external

@@ -6,3 +6,4 @@
     | `dirname $0`/diff-remove-uids \
     | `dirname $0`/diff-remove-mime-types \
     | `dirname $0`/diff-remove-x509-names \
+    | `dirname $0`/diff-sort

testing/scripts/diff-sort

@@ -0,0 +1,19 @@
+#! /usr/bin/env bash
+#
+# A diff canonifier that sorts all lines but keeps all comments
+# at the top. It also adds a note at the beginning as a reminder
+# that the output has been sorted.
+
+if [ "$TMP" == "" ]; then
+    TMP=/tmp
+fi
+
+tmp=$TMP/`basename $0`.$$.tmp
+
+cat >$tmp
+
+echo "### NOTE: This file has been sorted with `basename $0`."
+cat $tmp | grep ^#
+cat $tmp | grep -v ^# | sort -s
+
+rm -f $tmp