Merge branch 'master' into topic/jsiwek/improve_comm_loop

Conflicts:
	src/CMakeLists.txt
	src/FlowSrc.cc
	src/FlowSrc.h
	src/IOSource.h
	src/PktSrc.cc
	src/PktSrc.h
	src/iosource/Manager.cc

src/iosource/BPF_Program.cc

@@ -0,0 +1,148 @@
+// See the file "COPYING" in the main distribution directory for copyright.
+
+#include "config.h"
+
+#include "util.h"
+#include "BPF_Program.h"
+
+#ifdef DONT_HAVE_LIBPCAP_PCAP_FREECODE
+extern "C" {
+#include "pcap-int.h"
+
+int pcap_freecode(pcap_t* unused, struct bpf_program* program)
+	{
+	program->bf_len = 0;
+
+	if ( program->bf_insns )
+		{
+		free((char*) program->bf_insns);	// copied from libpcap
+		program->bf_insns = 0;
+		}
+
+	return 0;
+	}
+
+pcap_t* pcap_open_dead(int linktype, int snaplen)
+	{
+	pcap_t* p;
+
+	p = (pcap_t*) malloc(sizeof(*p));
+	if ( ! p )
+		return 0;
+
+	memset(p, 0, sizeof(*p));
+
+	p->fd = -1;
+	p->snapshot = snaplen;
+	p->linktype = linktype;
+
+	return p;
+	}
+
+int pcap_compile_nopcap(int snaplen_arg, int linktype_arg,
+			struct bpf_program* program, char* buf,
+			int optimize, bpf_u_int32 mask)
+	{
+	pcap_t* p;
+	int ret;
+
+	p = pcap_open_dead(linktype_arg, snaplen_arg);
+	if ( ! p )
+		return -1;
+
+	ret = pcap_compile(p, program, buf, optimize, mask);
+	pcap_close(p);
+
+	return ret;
+	}
+}
+#endif
+
+// Simple heuristic to identify filters that always match, so that we can
+// skip the filtering in that case. "ip or not ip" is Bro's default filter.
+static bool filter_matches_anything(const char *filter)
+	{
+	return (! filter) || strlen(filter) == 0 || strcmp(filter, "ip or not ip") == 0;
+	}
+
+BPF_Program::BPF_Program() : m_compiled(), m_matches_anything(false), m_program()
+	{
+	}
+
+BPF_Program::~BPF_Program()
+	{
+	FreeCode();
+	}
+
+bool BPF_Program::Compile(pcap_t* pcap, const char* filter, uint32 netmask,
+			  char* errbuf, unsigned int errbuf_len, bool optimize)
+	{
+	if ( ! pcap )
+		return false;
+
+	FreeCode();
+
+	if ( pcap_compile(pcap, &m_program, (char *) filter, optimize, netmask) < 0 )
+		{
+		if ( errbuf )
+			safe_snprintf(errbuf, errbuf_len,
+				      "pcap_compile(%s): %s", filter,
+				      pcap_geterr(pcap));
+
+		return false;
+		}
+
+	m_compiled = true;
+	m_matches_anything = filter_matches_anything(filter);
+
+	return true;
+	}
+
+bool BPF_Program::Compile(int snaplen, int linktype, const char* filter,
+			  uint32 netmask, char* errbuf, unsigned int errbuf_len,
+			  bool optimize)
+	{
+	FreeCode();
+
+#ifdef LIBPCAP_PCAP_COMPILE_NOPCAP_HAS_ERROR_PARAMETER
+	char my_error[PCAP_ERRBUF_SIZE];
+
+	int err = pcap_compile_nopcap(snaplen, linktype, &m_program,
+				     (char *) filter, optimize, netmask, my_error);
+	if ( err < 0 && errbuf )
+		safe_strncpy(errbuf, my_error, errbuf_len);
+		*errbuf = '\0';
+#else
+	int err = pcap_compile_nopcap(snaplen, linktype, &m_program,
+				     (char*) filter, optimize, netmask);
+
+	if ( err < 0 && errbuf && errbuf_len )
+		*errbuf = '\0';
+#endif
+
+	if ( err == 0 )
+		{
+		m_compiled = true;
+		m_matches_anything = filter_matches_anything(filter);
+		}
+
+	return err == 0;
+	}
+
+bpf_program* BPF_Program::GetProgram()
+	{
+	return m_compiled ? &m_program : 0;
+	}
+
+void BPF_Program::FreeCode()
+	{
+	if ( m_compiled )
+		{
+#ifdef DONT_HAVE_LIBPCAP_PCAP_FREECODE
+		pcap_freecode(NULL, &m_program);
+#else
+		pcap_freecode(&m_program);
+#endif
+		m_compiled = false;
+		}
+	}

src/iosource/BPF_Program.h

@@ -0,0 +1,58 @@
+// See the file "COPYING" in the main distribution directory for copyright.
+
+#ifndef bpf_program_h
+#define bpf_program_h
+
+extern "C" {
+#include <pcap.h>
+}
+
+#include "util.h"
+
+// BPF_Programs are an abstraction around struct bpf_program,
+// to create a clean facility for creating, compiling, and
+// freeing such programs.
+
+class BPF_Program {
+public:
+	// Creates an empty, uncompiled BPF program.
+	BPF_Program();
+	~BPF_Program();
+
+	// Creates a BPF program for the given pcap handle.
+	// Parameters are like in pcap_compile(). Returns true
+	// for successful compilation, false otherwise.
+	bool Compile(pcap_t* pcap, const char* filter, uint32 netmask,
+		 char* errbuf = 0, unsigned int errbuf_len = 0,
+		 bool optimize = true);
+
+	// Creates a BPF program when no pcap handle is around,
+	// similarly to pcap_compile_nopcap(). Parameters are
+	// similar. Returns true on success.
+	bool Compile(int snaplen, int linktype, const char* filter,
+		uint32 netmask, char* errbuf = 0, unsigned int errbuf_len = 0,
+		bool optimize = true);
+
+	// Returns true if this program currently contains compiled
+	// code, false otherwise.
+	bool IsCompiled()	{ return m_compiled; }
+
+	// Returns true if this program matches any packets. This is not
+	// comprehensive, but can identify a few cases where it does.
+	bool MatchesAnything() 	{ return m_matches_anything; }
+
+	// Accessor to the compiled program. Returns nil when
+	// no program is currently compiled.
+	bpf_program* GetProgram();
+
+protected:
+	void FreeCode();
+
+	// (I like to prefix member variables with m_, makes it clear
+	// in the implementation whether it's a global or not. --ck)
+	bool m_compiled;
+	bool m_matches_anything;
+	struct bpf_program m_program;
+};
+
+#endif

src/iosource/CMakeLists.txt

@@ -0,0 +1,23 @@
+
+include(BroSubdir)
+
+include_directories(BEFORE
+                    ${CMAKE_CURRENT_SOURCE_DIR}
+                    ${CMAKE_CURRENT_BINARY_DIR}
+)
+
+add_subdirectory(pcap)
+
+set(iosource_SRCS
+    BPF_Program.cc
+    Component.cc
+    Manager.cc
+    PktDumper.cc
+    PktSrc.cc
+)
+
+bif_target(pcap.bif)
+
+bro_add_subdir_library(iosource ${iosource_SRCS})
+add_dependencies(bro_iosource generate_outputs)
+

src/iosource/Component.cc

@@ -0,0 +1,165 @@
+// See the file "COPYING" in the main distribution directory for copyright.
+
+#include "Component.h"
+
+#include "Desc.h"
+#include "Reporter.h"
+
+using namespace iosource;
+
+Component::Component(const std::string& name)
+	: plugin::Component(plugin::component::IOSOURCE, name)
+	{
+	}
+
+Component::Component(plugin::component::Type type, const std::string& name)
+	: plugin::Component(type, name)
+	{
+	}
+
+Component::~Component()
+	{
+	}
+
+PktSrcComponent::PktSrcComponent(const std::string& arg_name, const std::string& arg_prefix, InputType arg_type, factory_callback arg_factory)
+	: iosource::Component(plugin::component::PKTSRC, arg_name)
+	{
+	tokenize_string(arg_prefix, ":", &prefixes);
+	type = arg_type;
+	factory = arg_factory;
+	}
+
+PktSrcComponent::~PktSrcComponent()
+	{
+	}
+
+const std::vector<std::string>& PktSrcComponent::Prefixes() const
+	{
+	return prefixes;
+	}
+
+bool PktSrcComponent::HandlesPrefix(const string& prefix) const
+	{
+	for ( std::vector<std::string>::const_iterator i = prefixes.begin();
+	      i != prefixes.end(); i++ )
+		{
+		if ( *i == prefix )
+			return true;
+		}
+
+	return false;
+	}
+
+bool PktSrcComponent::DoesLive() const
+	{
+	return type == LIVE || type == BOTH;
+	}
+
+bool PktSrcComponent::DoesTrace() const
+	{
+	return type == TRACE || type == BOTH;
+	}
+
+PktSrcComponent::factory_callback PktSrcComponent::Factory() const
+	{
+	return factory;
+	}
+
+void PktSrcComponent::DoDescribe(ODesc* d) const
+	{
+	iosource::Component::DoDescribe(d);
+
+	string prefs;
+
+	for ( std::vector<std::string>::const_iterator i = prefixes.begin();
+	      i != prefixes.end(); i++ )
+		{
+		if ( prefs.size() )
+			prefs += ", ";
+
+		prefs += '"' + *i + '"';
+		}
+
+	d->Add("interface prefix");
+	if ( prefixes.size() > 1 )
+		d->Add("es");
+
+	d->Add(" ");
+	d->Add(prefs);
+	d->Add("; supports ");
+
+	switch ( type ) {
+	case LIVE:
+		d->Add("live input");
+		break;
+
+	case TRACE:
+		d->Add("trace input");
+		break;
+
+	case BOTH:
+		d->Add("live and trace input");
+		break;
+
+	default:
+		reporter->InternalError("unknown PkrSrc type");
+	}
+
+	}
+
+PktDumperComponent::PktDumperComponent(const std::string& name, const std::string& arg_prefix, factory_callback arg_factory)
+	: plugin::Component(plugin::component::PKTDUMPER, name)
+	{
+	tokenize_string(arg_prefix, ":", &prefixes);
+	factory = arg_factory;
+	}
+
+PktDumperComponent::~PktDumperComponent()
+	{
+	}
+
+PktDumperComponent::factory_callback PktDumperComponent::Factory() const
+	{
+	return factory;
+	}
+
+const std::vector<std::string>& PktDumperComponent::Prefixes() const
+	{
+	return prefixes;
+	}
+
+bool PktDumperComponent::HandlesPrefix(const string& prefix) const
+	{
+	for ( std::vector<std::string>::const_iterator i = prefixes.begin();
+	      i != prefixes.end(); i++ )
+		{
+		if ( *i == prefix )
+			return true;
+		}
+
+	return false;
+	}
+
+void PktDumperComponent::DoDescribe(ODesc* d) const
+	{
+	plugin::Component::DoDescribe(d);
+
+	string prefs;
+
+	for ( std::vector<std::string>::const_iterator i = prefixes.begin();
+	      i != prefixes.end(); i++ )
+		{
+		if ( prefs.size() )
+			prefs += ", ";
+
+		prefs += '"' + *i + '"';
+		}
+
+	d->Add("dumper prefix");
+
+	if ( prefixes.size() > 1 )
+		d->Add("es");
+
+	d->Add(": ");
+	d->Add(prefs);
+	}

src/iosource/Component.h

@@ -0,0 +1,177 @@
+// See the file "COPYING" in the main distribution directory for copyright.
+
+#ifndef IOSOURCE_PLUGIN_COMPONENT_H
+#define IOSOURCE_PLUGIN_COMPONENT_H
+
+#include <string>
+#include <vector>
+
+#include "plugin/Component.h"
+
+namespace iosource {
+
+class IOSource;
+class PktSrc;
+class PktDumper;
+
+/**
+ * Component description for plugins providing IOSources.
+ */
+class Component : public plugin::Component {
+public:
+	typedef IOSource* (*factory_callback)();
+
+	/**
+	 * Constructor.
+	 *
+	 * @param name A descriptive name for the component.  This name must
+	 * be unique across all components of this type.
+	 */
+	Component(const std::string& name);
+
+	/**
+	 * Copy constructor.
+	 */
+	Component(const Component& other);
+
+	/**
+	 * Destructor.
+	 */
+	~Component();
+
+protected:
+	/**
+	 * Constructor to use by derived classes.
+	 *
+	 * @param type The type of the componnent.
+	 *
+	 * @param name A descriptive name for the component.  This name must
+	 * be unique across all components of this type.
+	 */
+	Component(plugin::component::Type type, const std::string& name);
+};
+
+/**
+ * Component description for plugins providing a PktSrc for packet input.
+ */
+class PktSrcComponent : public iosource::Component {
+public:
+	/**
+	 * Type of input a packet source supports.
+	 */
+	enum InputType {
+		LIVE,	///< Live input.
+		TRACE,		///< Offline input from trace file.
+		BOTH	///< Live input as well as offline.
+	};
+
+	typedef PktSrc* (*factory_callback)(const std::string& path, bool is_live);
+
+	/**
+	 * Constructor.
+	 *
+	 * @param name A descriptive name for the component.  This name must
+	 * be unique across all components of this type.
+	 *
+	 * @param prefixes The list of interface/file prefixes associated
+	 * with this component.
+	 *
+	 * @param type Type of input the component supports.
+	 *
+	 * @param factor Factory function to instantiate component.
+	 */
+	PktSrcComponent(const std::string& name, const std::string& prefixes, InputType type, factory_callback factory);
+
+	/**
+	 * Destructor.
+	 */
+	virtual ~PktSrcComponent();
+
+	/**
+	 * Returns the prefix(es) passed to the constructor.
+	 */
+	const std::vector<std::string>& Prefixes() const;
+
+	/**
+	 * Returns true if the given prefix is among the one specified for the component.
+	 */
+	bool HandlesPrefix(const std::string& prefix) const;
+
+	/**
+	 * Returns true if packet source instantiated by the component handle
+	 * live traffic.
+	 */
+	bool DoesLive() const;
+
+	/**
+	 * Returns true if packet source instantiated by the component handle
+	 * offline traces.
+	 */
+	bool DoesTrace() const;
+
+	/**
+	 * Returns the source's factory function.
+	 */
+	factory_callback Factory() const;
+
+	/**
+	 * Generates a human-readable description of the component. This goes
+	 * into the output of \c "bro -NN".
+	 */
+	virtual void DoDescribe(ODesc* d) const;
+
+private:
+	std::vector<std::string> prefixes;
+	InputType type;
+	factory_callback factory;
+};
+
+/**
+ * Component description for plugins providing a PktDumper for packet output.
+ *
+ * PktDumpers aren't IOSurces but we locate them here to keep them along with
+ * the PktSrc.
+ */
+class PktDumperComponent : public plugin::Component  {
+public:
+	typedef PktDumper* (*factory_callback)(const std::string& path, bool append);
+
+	/**
+	 * XXX
+	 */
+	PktDumperComponent(const std::string& name, const std::string& prefixes, factory_callback factory);
+
+	/**
+	 * Destructor.
+	 */
+	~PktDumperComponent();
+
+	/**
+	 * Returns the prefix(es) passed to the constructor.
+	 */
+	const std::vector<std::string>& Prefixes() const;
+
+	/**
+	 * Returns true if the given prefix is among the one specified for the component.
+	 */
+	bool HandlesPrefix(const std::string& prefix) const;
+
+	/**
+	 * Returns the source's factory function.
+	 */
+	factory_callback Factory() const;
+
+	/**
+	 * Generates a human-readable description of the component. This goes
+	 * into the output of \c "bro -NN".
+	 */
+	virtual void DoDescribe(ODesc* d) const;
+
+private:
+	std::vector<std::string> prefixes;
+	factory_callback factory;
+};
+
+}
+
+#endif

src/iosource/IOSource.h

@@ -0,0 +1,142 @@
+// See the file "COPYING" in the main distribution directory for copyright.
+
+#ifndef IOSOURCE_IOSOURCE_H
+#define IOSOURCE_IOSOURCE_H
+
+extern "C" {
+#include <pcap.h>
+}
+
+#include <string>
+#include <vector>
+
+#include "Timer.h"
+
+namespace iosource {
+
+/**
+ * Interface class for components providing/consuming data inside Bro's main
+ * loop.
+ */
+class IOSource {
+public:
+	/**
+	 * Constructor.
+	 */
+	IOSource()	{ idle = false; closed = false; }
+
+	/**
+	 * Destructor.
+	 */
+	virtual ~IOSource()	{}
+
+	/**
+	 * Returns true if source has nothing ready to process.
+	 */
+	bool IsIdle() const	{ return idle; }
+
+	/**
+	 * Returns true if more data is to be expected in the future.
+	 * Otherwise, source may be removed.
+	 */
+	bool IsOpen() const	{ return ! closed; }
+
+	/**
+	 * Initializes the source. Can be overwritten by derived classes.
+	 */
+	virtual void Init()	{ }
+
+	/**
+	 * Finalizes the source when it's being closed. Can be overwritten by
+	 * derived classes.
+	 */
+	virtual void Done()	{ }
+
+	/**
+	 * Returns select'able file descriptors for this source. Leaves the
+	 * passed values untouched if not available.
+	 *
+	 * @param read Pointer to container where to insert a read descriptor.
+	 *
+	 * @param write Pointer to container where to insert a write descriptor.
+	 *
+	 * @param except Pointer to container where to insert a except descriptor.
+	 */
+	virtual void GetFds(std::vector<int>* read, std::vector<int>* write,
+	                    std::vector<int>* except) = 0;
+
+	/**
+	 * Returns the timestamp (in \a global network time) associated with
+	 * next data item from this source.  If the source wants the data
+	 * item to be processed with a local network time, it sets the
+	 * argument accordingly.
+	 *
+	 * This method will be called only when either IsIdle() returns
+	 * false, or select() on one of the fds returned by GetFDs()
+	 * indicates that there's data to process.
+	 *
+	 * Must be overridden by derived classes.
+	 *
+	 * @param network_time A pointer to store the \a local network time
+	 * associated with the next item (as opposed to global network time).
+	 *
+	 * @return The global network time of the next entry, or a value
+	 * smaller than zero if none is available currently.
+	 */
+	virtual double NextTimestamp(double* network_time) = 0;
+
+	/**
+	 * Processes and consumes next data item.
+	 *
+	 * This method will be called only when either IsIdle() returns
+	 * false, or select() on one of the fds returned by GetFDs()
+	 * indicates that there's data to process.
+	 *
+	 * Must be overridden by derived classes.
+	 */
+	virtual void Process() = 0;
+
+	/**
+	 * Returns the tag of the timer manafger associated with the last
+	 * procesees data item.
+	 *
+	 * Can be overridden by derived classes.
+	 *
+	 * @return The tag, or null for the global timer manager.
+	 * 
+	 */
+	virtual TimerMgr::Tag* GetCurrentTag()	{ return 0; }
+
+	/**
+	 * Returns a descriptual tag representing the source for debugging.
+	 *
+	 * Can be overridden by derived classes.
+	 *
+	 * @return The debugging name.
+	 */
+	virtual const char* Tag() = 0;
+
+protected:
+	/*
+	 * Callback for derived classes to call when they have gone dry
+	 * temporarily.
+	 *
+	 * @param is_idle True if the source is idle currently.
+	 */
+	void SetIdle(bool is_idle)	{ idle = is_idle; }
+
+	/*
+	 * Callback for derived class to call when they have shutdown.
+	 *
+	 * @param is_closed True if the source is now closed.
+	 */
+	void SetClosed(bool is_closed)	{ closed = is_closed; }
+
+private:
+	bool idle;
+	bool closed;
+};
+
+}
+
+#endif

src/iosource/Manager.cc

@@ -0,0 +1,336 @@
+// See the file "COPYING" in the main distribution directory for copyright.
+
+#include <sys/types.h>
+#include <sys/time.h>
+#include <unistd.h>
+#include <assert.h>
+
+#include <algorithm>
+
+#include "Manager.h"
+#include "IOSource.h"
+#include "PktSrc.h"
+#include "PktDumper.h"
+#include "plugin/Manager.h"
+
+#include "util.h"
+
+#define DEFAULT_PREFIX "pcap"
+
+using namespace iosource;
+
+Manager::~Manager()
+	{
+	for ( SourceList::iterator i = sources.begin(); i != sources.end(); ++i )
+		{
+		(*i)->src->Done();
+		delete *i;
+		}
+
+	sources.clear();
+
+	for ( PktDumperList::iterator i = pkt_dumpers.begin(); i != pkt_dumpers.end(); ++i )
+		{
+		(*i)->Done();
+		delete *i;
+		}
+
+	pkt_dumpers.clear();
+	}
+
+void Manager::RemoveAll()
+	{
+	// We're cheating a bit here ...
+	dont_counts = sources.size();
+	}
+
+static void fd_vector_set(const std::vector<int>& fds, fd_set* set, int* max)
+	{
+	for ( size_t i = 0; i < fds.size(); ++i )
+		{
+		FD_SET(fds[i], set);
+		*max = ::max(fds[i], *max);
+		}
+	}
+
+IOSource* Manager::FindSoonest(double* ts)
+	{
+	// Remove sources which have gone dry. For simplicity, we only
+	// remove at most one each time.
+	for ( SourceList::iterator i = sources.begin();
+	      i != sources.end(); ++i )
+		if ( ! (*i)->src->IsOpen() )
+			{
+			(*i)->src->Done();
+			delete *i;
+			sources.erase(i);
+			break;
+			}
+
+	// Ideally, we would always call select on the fds to see which
+	// are ready, and return the soonest. Unfortunately, that'd mean
+	// one select-call per packet, which we can't afford in high-volume
+	// environments.  Thus, we call select only every SELECT_FREQUENCY
+	// call (or if all sources report that they are dry).
+
+	++call_count;
+
+	IOSource* soonest_src = 0;
+	double soonest_ts = 1e20;
+	double soonest_local_network_time = 1e20;
+	bool all_idle = true;
+
+	// Find soonest source of those which tell us they have something to
+	// process.
+	for ( SourceList::iterator i = sources.begin(); i != sources.end(); ++i )
+		{
+		if ( ! (*i)->src->IsIdle() )
+			{
+			all_idle = false;
+			double local_network_time = 0;
+			double ts = (*i)->src->NextTimestamp(&local_network_time);
+			if ( ts > 0 && ts < soonest_ts )
+				{
+				soonest_ts = ts;
+				soonest_src = (*i)->src;
+				soonest_local_network_time =
+					local_network_time ?
+						local_network_time : ts;
+				}
+			}
+		}
+
+	// If we found one and aren't going to select this time,
+	// return it.
+	int maxx = 0;
+
+	if ( soonest_src && (call_count % SELECT_FREQUENCY) != 0 )
+		goto finished;
+
+	// Select on the join of all file descriptors.
+	fd_set fd_read, fd_write, fd_except;
+
+	FD_ZERO(&fd_read);
+	FD_ZERO(&fd_write);
+	FD_ZERO(&fd_except);
+
+	for ( SourceList::iterator i = sources.begin();
+	      i != sources.end(); ++i )
+		{
+		Source* src = (*i);
+
+		if ( ! src->src->IsIdle() )
+			// No need to select on sources which we know to
+			// be ready.
+			continue;
+
+		src->fd_read.clear();
+		src->fd_write.clear();
+		src->fd_except.clear();
+		src->src->GetFds(&src->fd_read, &src->fd_write, &src->fd_except);
+
+		fd_vector_set(src->fd_read, &fd_read, &maxx);
+		fd_vector_set(src->fd_write, &fd_write, &maxx);
+		fd_vector_set(src->fd_except, &fd_except, &maxx);
+		}
+
+	// We can't block indefinitely even when all sources are dry:
+	// we're doing some IOSource-independent stuff in the main loop,
+	// so we need to return from time to time. (Instead of no time-out
+	// at all, we use a very small one. This lets FreeBSD trigger a
+	// BPF buffer switch on the next read when the hold buffer is empty
+	// while the store buffer isn't filled yet.
+
+	struct timeval timeout;
+
+	if ( all_idle )
+		{
+		// Interesting: when all sources are dry, simply sleeping a
+		// bit *without* watching for any fd becoming ready may
+		// decrease CPU load. I guess that's because it allows
+		// the kernel's packet buffers to fill. - Robin
+		timeout.tv_sec = 0;
+		timeout.tv_usec = 20; // SELECT_TIMEOUT;
+		select(0, 0, 0, 0, &timeout);
+		}
+
+	if ( ! maxx )
+		// No selectable fd at all.
+		goto finished;
+
+	timeout.tv_sec = 0;
+	timeout.tv_usec = 0;
+
+	if ( select(maxx + 1, &fd_read, &fd_write, &fd_except, &timeout) > 0 )
+		{ // Find soonest.
+		for ( SourceList::iterator i = sources.begin();
+		      i != sources.end(); ++i )
+			{
+			Source* src = (*i);
+
+			if ( ! src->src->IsIdle() )
+				continue;
+
+			if ( src->Ready(&fd_read, &fd_write, &fd_except) )
+				{
+				double local_network_time = 0;
+				double ts = src->src->NextTimestamp(&local_network_time);
+				if ( ts > 0.0 && ts < soonest_ts )
+					{
+					soonest_ts = ts;
+					soonest_src = src->src;
+					soonest_local_network_time =
+						local_network_time ?
+							local_network_time : ts;
+					}
+				}
+			}
+		}
+
+finished:
+	*ts = soonest_local_network_time;
+	return soonest_src;
+	}
+
+void Manager::Register(IOSource* src, bool dont_count)
+	{
+	src->Init();
+	Source* s = new Source;
+	s->src = src;
+	if ( dont_count )
+		++dont_counts;
+
+	sources.push_back(s);
+	}
+
+void Manager::Register(PktSrc* src)
+	{
+	pkt_srcs.push_back(src);
+	Register(src, false);
+	}
+
+static std::pair<std::string, std::string> split_prefix(std::string path)
+	{
+	// See if the path comes with a prefix telling us which type of
+	// PktSrc to use. If not, choose default.
+	std::string prefix;
+
+	std::string::size_type i = path.find(":");
+	if ( i != std::string::npos )
+		{
+		prefix = path.substr(0, i);
+		path = path.substr(++i, std::string::npos);
+		}
+
+	else
+		prefix= DEFAULT_PREFIX;
+
+	return std::make_pair(prefix, path);
+	}
+
+PktSrc* Manager::OpenPktSrc(const std::string& path, bool is_live)
+	{
+	std::pair<std::string, std::string> t = split_prefix(path);
+	std::string prefix = t.first;
+	std::string npath = t.second;
+
+	// Find the component providing packet sources of the requested prefix.
+
+	PktSrcComponent* component = 0;
+
+	std::list<PktSrcComponent*> all_components = plugin_mgr->Components<PktSrcComponent>();
+
+	for ( std::list<PktSrcComponent*>::const_iterator i = all_components.begin();
+	      i != all_components.end(); i++ )
+		{
+		PktSrcComponent* c = *i;
+
+		if ( c->HandlesPrefix(prefix) &&
+		     ((  is_live && c->DoesLive() ) ||
+		      (! is_live && c->DoesTrace())) )
+			{
+			component = c;
+			break;
+			}
+		}
+
+
+	if ( ! component )
+		reporter->FatalError("type of packet source '%s' not recognized, or mode not supported", prefix.c_str());
+
+	// Instantiate packet source.
+
+	PktSrc* ps = (*component->Factory())(npath, is_live);
+	assert(ps);
+
+	if ( ! ps->IsOpen() && ps->IsError() )
+		// Set an error message if it didn't open successfully.
+		ps->Error("could not open");
+
+	DBG_LOG(DBG_PKTIO, "Created packet source of type %s for %s", component->Name().c_str(), npath.c_str());
+
+	Register(ps);
+	return ps;
+	}
+
+
+PktDumper* Manager::OpenPktDumper(const string& path, bool append)
+	{
+	std::pair<std::string, std::string> t = split_prefix(path);
+	std::string prefix = t.first;
+	std::string npath = t.second;
+
+	// Find the component providing packet dumpers of the requested prefix.
+
+	PktDumperComponent* component = 0;
+
+	std::list<PktDumperComponent*> all_components = plugin_mgr->Components<PktDumperComponent>();
+
+	for ( std::list<PktDumperComponent*>::const_iterator i = all_components.begin();
+	      i != all_components.end(); i++ )
+		{
+		if ( (*i)->HandlesPrefix(prefix) )
+			{
+			component = (*i);
+			break;
+			}
+		}
+
+	if ( ! component )
+		reporter->FatalError("type of packet dumper '%s' not recognized", prefix.c_str());
+
+	// Instantiate packet dumper.
+
+	PktDumper* pd = (*component->Factory())(npath, append);
+	assert(pd);
+
+	if ( ! pd->IsOpen() && pd->IsError() )
+		// Set an error message if it didn't open successfully.
+		pd->Error("could not open");
+
+	DBG_LOG(DBG_PKTIO, "Created packer dumper of type %s for %s", component->Name().c_str(), npath.c_str());
+
+	pd->Init();
+	pkt_dumpers.push_back(pd);
+
+	return pd;
+	}
+
+static bool fd_vector_ready(const std::vector<int>& fds, fd_set* set)
+	{
+	for ( size_t i = 0; i < fds.size(); ++i )
+		if ( FD_ISSET(fds[i], set) )
+			return true;
+
+	return false;
+	}
+
+bool Manager::Source::Ready(fd_set* read, fd_set* write, fd_set* except) const
+	{
+	if ( fd_vector_ready(fd_read, read) ||
+	     fd_vector_ready(fd_write, write) ||
+	     fd_vector_ready(fd_except, except) )
+		return true;
+
+	return false;
+	}

src/iosource/Manager.h

@@ -0,0 +1,139 @@
+// See the file "COPYING" in the main distribution directory for copyright.
+
+#ifndef IOSOURCE_MANAGER_H
+#define IOSOURCE_MANAGER_H
+
+#include <string>
+#include <list>
+#include <vector>
+#include <sys/select.h>
+
+namespace iosource {
+
+class IOSource;
+class PktSrc;
+class PktDumper;
+
+/**
+ * Singleton class managing all IOSources.
+ */
+class Manager {
+public:
+	/**
+	 * Constructor.
+	 */
+	Manager()	{ call_count = 0; dont_counts = 0; }
+
+	/**
+	 * Destructor.
+	 */
+	~Manager();
+
+	/**
+	 * Registers an IOSource with the manager.
+	 *
+	 * @param src The source. The manager takes ownership.
+	 *
+	 * @param dont_count If true, this source does not contribute to the
+	 * number of IOSources returned by Size().  The effect is that if all
+	 * sources except for the non-counting ones have gone dry, processing
+	 * will shut down.
+	 */
+	void Register(IOSource* src, bool dont_count = false);
+
+	/**
+	 * Returns the packet source with the soonest available input. This
+	 * may block for a little while if all are dry.
+	 *
+	 * @param ts A pointer where to store the timestamp of the input that
+	 * the soonest source has available next. 
+	 *
+	 * @return The source, or null if no source has input.
+	 */
+	IOSource* FindSoonest(double* ts);
+
+	/**
+	 * Returns the number of registered and still active sources,
+	 * excluding those that are registered as \a dont_cont.
+	 */
+	int Size() const	{ return sources.size() - dont_counts; }
+
+	typedef std::list<PktSrc *> PktSrcList;
+
+	/**
+	 * Returns a list of all registered PktSrc instances. This is a
+	 * subset of all registered IOSource instances.
+	 */
+	const PktSrcList& GetPktSrcs() const	{ return pkt_srcs; }
+
+	/**
+	 * Terminate all processing immediately by removing all sources (and
+	 * therefore now returning a Size() of zero).
+	 */
+	void Terminate()	{ RemoveAll(); }
+
+	/**
+	 * Opens a new packet source.
+	 *
+	 * @param path The interface or file name, as one would give to Bro \c -i.
+	 *
+	 * @param is_live True if \a path represents a live interface, false
+	 * for a file.
+	 *
+	 * @return The new packet source, or null if an error occured.
+	 */
+	PktSrc* OpenPktSrc(const std::string& path, bool is_live);
+
+	/**
+	 * Opens a new packet dumper.
+	 *
+	 * @param path The file name to dump into.
+	 *
+	 * @param append True to append if \a path already exists.
+ 	 *
+	 * @return The new packet dumper, or null if an error occured.
+	 */
+	PktDumper* OpenPktDumper(const std::string& path, bool append);
+
+private:
+	/**
+	 * When looking for a source with something to process, every
+	 * SELECT_FREQUENCY calls we will go ahead and block on a select().
+	 */
+	static const int SELECT_FREQUENCY = 25;
+
+	/**
+	 * Microseconds to wait in an empty select if no source is ready.
+	 */
+	static const int SELECT_TIMEOUT = 50;
+
+	void Register(PktSrc* src);
+	void RemoveAll();
+
+	unsigned int call_count;
+	int dont_counts;
+
+	struct Source {
+		IOSource* src;
+		std::vector<int> fd_read;
+		std::vector<int> fd_write;
+		std::vector<int> fd_except;
+
+		bool Ready(fd_set* read, fd_set* write, fd_set* except) const;
+	};
+
+	typedef std::list<Source*> SourceList;
+	SourceList sources;
+
+	typedef std::list<PktDumper *> PktDumperList;
+
+	PktSrcList pkt_srcs;
+	PktDumperList pkt_dumpers;
+};
+
+}
+
+extern iosource::Manager* iosource_mgr;
+
+#endif
+

src/iosource/PktDumper.cc

@@ -0,0 +1,84 @@
+
+// See the file "COPYING" in the main distribution directory for copyright.
+
+#include <errno.h>
+#include <sys/stat.h>
+
+#include "config.h"
+
+#include "PktDumper.h"
+
+using namespace iosource;
+
+PktDumper::PktDumper()
+	{
+	is_open = false;
+	errmsg = "";
+	}
+
+PktDumper::~PktDumper()
+	{
+	}
+
+void PktDumper::Init()
+	{
+	Open();
+	}
+
+void PktDumper::Done()
+	{
+	Close();
+	}
+
+const std::string& PktDumper::Path() const
+	{
+	return props.path;
+	}
+
+bool PktDumper::IsOpen() const
+	{
+	return is_open;
+	}
+
+double PktDumper::OpenTime() const
+	{
+	return is_open ? props.open_time : 0;
+	}
+
+bool PktDumper::IsError() const
+	{
+	return errmsg.size();
+	}
+
+const char* PktDumper::ErrorMsg() const
+	{
+	return errmsg.size() ? errmsg.c_str() : 0;
+	}
+
+int PktDumper::HdrSize() const
+	{
+	return is_open ? props.hdr_size : -1;
+	}
+
+void PktDumper::Opened(const Properties& arg_props)
+	{
+	is_open = true;
+	props = arg_props;
+	DBG_LOG(DBG_PKTIO, "Opened dumper %s", props.path.c_str());
+	}
+
+void PktDumper::Closed()
+	{
+	is_open = false;
+	DBG_LOG(DBG_PKTIO, "Closed dumper %s", props.path.c_str());
+	props.path = "";
+	}
+
+void PktDumper::Error(const std::string& msg)
+	{
+	errmsg = msg;
+
+	DBG_LOG(DBG_PKTIO, "Error with dumper %s: %s",
+		IsOpen() ? props.path.c_str() : "<not open>",
+		msg.c_str());
+	}

src/iosource/PktDumper.h

@@ -0,0 +1,163 @@
+// See the file "COPYING" in the main distribution directory for copyright.
+
+#ifndef IOSOURCE_PKTSRC_PKTDUMPER_H
+#define IOSOURCE_PKTSRC_PKTDUMPER_H
+
+#include "IOSource.h"
+
+namespace iosource {
+
+/**
+ * Base class for packet dumpers.
+ */
+class PktDumper {
+public:
+	/**
+	 * Structure describing a packet.
+	 */
+	struct Packet {
+		/**
+		 * The pcap header associated with the packet.
+		 */
+		const struct pcap_pkthdr* hdr;
+
+		/**
+		 * The full content of the packet.
+		 */
+		const unsigned char* data;
+	};
+
+	/**
+	 * Constructor.
+	 */
+	PktDumper();
+
+	/**
+	 * Destructor.
+	 */
+	virtual ~PktDumper();
+
+	/**
+	 * Returns the path associated with the dumper.
+	 */
+	const std::string& Path() const;
+
+	/**
+	 * Returns true if the dumper is open for writing.
+	 */
+	bool IsOpen() const;
+
+	/**
+	 * Returns the time when the dumper was opened for writing.
+	 */
+	double OpenTime() const;
+
+	/**
+	 * Returns returns true if the dumper has encountered an error.
+	 */
+	bool IsError() const;
+
+	/**
+	 * Returns if the dumper has encountered an error, returns a
+	 * corresponding error message. Returns an emoty string otherwise.
+	 */
+	const char* ErrorMsg() const;
+
+	/**
+	 * Returns the size of the link-layer headers with this dumper.
+	 */
+	int HdrSize() const;
+
+	/**
+	 * Writes a packet to the dumper.
+	 *
+	 * @param pkt The packet to record.
+	 */
+	bool Record(const Packet* pkt);
+
+	// PktDumper interface for derived classes to implement.
+
+	/**
+	 * Called by the manager system to open the source.
+	 *
+	 * Derived classes must implement this method. If successful, the
+	 * implementation must call \a Opened(); if not, it must call Error()
+	 * with a corresponding message.
+	 */
+	virtual void Open() = 0;
+
+	/**
+	 * Called by the manager system to close the dumper.
+	 *
+	 * Derived classes must implement this method. If successful, the
+	 * implementation must call \a Closed(); if not, it must call Error()
+	 * with a corresponding message.
+	 */
+	virtual void Close() = 0;
+
+	/**
+	 * Called to write a packet to the dumper.
+	 *
+	 * Derived classes must implement this method.
+	 *
+	 * @param pkt The packet to record.
+	 *
+	 * @return True if succesful, false otherwise (in which case \a
+	 * Error() must have been called.)
+	 */
+	virtual bool Dump(const Packet* pkt) = 0;
+
+protected:
+	friend class Manager;
+
+	/**
+	 * Structure to pass back information about the packet dumper to the
+	 * base class. Derived class pass an instance of this to \a Opened().
+	 */
+	struct Properties {
+		std::string path;
+		int hdr_size;
+		double open_time;
+	};
+
+	/**
+	 * Called from the implementations of \a Open() to signal that the
+	 * source has been successully opened.
+	 *
+	 * @param props A properties instance describing the now open source.
+	 */
+	void Opened(const Properties& props);
+
+	/**
+	 * Called from the implementations of \a Close() to signal that the
+	 * source has been closed.
+	 */
+	void Closed();
+
+	/**
+	 * Called from derived classes to signal an error.
+	 *
+	 * @param msg A corresponding error message.
+	 */
+	void Error(const std::string& msg);
+
+	/**
+	 * Called by the manager to initialize the dumper.
+	 */
+	void Init();
+
+	/**
+	 * Called by the manager to shutdown the dumper.
+	 */
+	void Done();
+
+private:
+	bool is_open;
+	Properties props;
+
+	std::string errmsg;
+};
+
+}
+
+#endif

src/iosource/PktSrc.cc

@@ -0,0 +1,524 @@
+// See the file "COPYING" in the main distribution directory for copyright.
+
+#include <errno.h>
+#include <sys/stat.h>
+
+#include "config.h"
+
+#include "util.h"
+#include "PktSrc.h"
+#include "Hash.h"
+#include "Net.h"
+#include "Sessions.h"
+
+using namespace iosource;
+
+PktSrc::PktSrc()
+	{
+	have_packet = false;
+	errbuf = "";
+	SetClosed(true);
+
+	next_sync_point = 0;
+	first_timestamp = 0.0;
+	first_wallclock = current_wallclock = 0;
+	}
+
+PktSrc::~PktSrc()
+	{
+	BPF_Program* code;
+	IterCookie* cookie = filters.InitForIteration();
+	while ( (code = filters.NextEntry(cookie)) )
+		delete code;
+	}
+
+const std::string& PktSrc::Path() const
+	{
+	static std::string not_open("not open");
+	return IsOpen() ? props.path : not_open;
+	}
+
+const char* PktSrc::ErrorMsg() const
+	{
+	return errbuf.size() ? errbuf.c_str() : 0;
+	}
+
+int PktSrc::LinkType() const
+	{
+	return IsOpen() ? props.link_type : -1;
+	}
+
+uint32 PktSrc::Netmask() const
+	{
+	return IsOpen() ? props.netmask : PCAP_NETMASK_UNKNOWN;
+	}
+
+bool PktSrc::IsError() const
+	{
+	return ErrorMsg();
+	}
+
+int PktSrc::HdrSize() const
+	{
+	return IsOpen() ? props.hdr_size : -1;
+	}
+
+int PktSrc::SnapLen() const
+	{
+	return snaplen; // That's a global. Change?
+	}
+
+bool PktSrc::IsLive() const
+	{
+	return props.is_live;
+	}
+
+double PktSrc::CurrentPacketTimestamp()
+	{
+	return current_pseudo;
+	}
+
+double PktSrc::CurrentPacketWallClock()
+	{
+	// We stop time when we are suspended.
+	if ( net_is_processing_suspended() )
+		current_wallclock = current_time(true);
+
+	return current_wallclock;
+	}
+
+void PktSrc::Opened(const Properties& arg_props)
+	{
+	if ( arg_props.hdr_size < 0 )
+		{
+		char buf[512];
+		safe_snprintf(buf, sizeof(buf),
+			 "unknown data link type 0x%x", props.link_type);
+		Error(buf);
+		Close();
+		return;
+		}
+
+	props = arg_props;
+	SetClosed(false);
+
+	if ( ! PrecompileFilter(0, "") || ! SetFilter(0) )
+		{
+		Close();
+		return;
+		}
+
+	if ( props.is_live )
+		Info(fmt("listening on %s, capture length %d bytes\n", props.path.c_str(), SnapLen()));
+
+	DBG_LOG(DBG_PKTIO, "Opened source %s", props.path.c_str());
+	}
+
+void PktSrc::Closed()
+	{
+	SetClosed(true);
+
+	DBG_LOG(DBG_PKTIO, "Closed source %s", props.path.c_str());
+	}
+
+void PktSrc::Error(const std::string& msg)
+	{
+	// We don't report this immediately, Bro will ask us for the error
+	// once it notices we aren't open.
+	errbuf = msg;
+	DBG_LOG(DBG_PKTIO, "Error with source %s: %s",
+		IsOpen() ? props.path.c_str() : "<not open>",
+		msg.c_str());
+	}
+
+void PktSrc::Info(const std::string& msg)
+	{
+	reporter->Info("%s", msg.c_str());
+	}
+
+void PktSrc::Weird(const std::string& msg, const Packet* p)
+	{
+	sessions->Weird(msg.c_str(), p->hdr, p->data, 0);
+	}
+
+void PktSrc::InternalError(const std::string& msg)
+	{
+	reporter->InternalError("%s", msg.c_str());
+	}
+
+void PktSrc::ContinueAfterSuspend()
+	{
+	current_wallclock = current_time(true);
+	}
+
+int PktSrc::GetLinkHeaderSize(int link_type)
+	{
+	switch ( link_type ) {
+	case DLT_NULL:
+		return 4;
+
+	case DLT_EN10MB:
+		return 14;
+
+	case DLT_FDDI:
+		return 13 + 8;	// fddi_header + LLC
+
+#ifdef DLT_LINUX_SLL
+	case DLT_LINUX_SLL:
+		return 16;
+#endif
+
+	case DLT_PPP_SERIAL:	// PPP_SERIAL
+		return 4;
+
+	case DLT_RAW:
+		return 0;
+	}
+
+	return -1;
+	}
+
+double PktSrc::CheckPseudoTime()
+	{
+	if ( ! IsOpen() )
+		return 0;
+
+	if ( ! ExtractNextPacketInternal() )
+		return 0;
+
+	if ( remote_trace_sync_interval )
+		{
+		if ( next_sync_point == 0 || current_packet.ts >= next_sync_point )
+			{
+			int n = remote_serializer->SendSyncPoint();
+			next_sync_point = first_timestamp +
+						n * remote_trace_sync_interval;
+			remote_serializer->Log(RemoteSerializer::LogInfo,
+				fmt("stopping at packet %.6f, next sync-point at %.6f",
+					current_packet.ts, next_sync_point));
+
+			return 0;
+			}
+		}
+
+	double pseudo_time = current_packet.ts - first_timestamp;
+	double ct = (current_time(true) - first_wallclock) * pseudo_realtime;
+
+	return pseudo_time <= ct ? bro_start_time + pseudo_time : 0;
+	}
+
+void PktSrc::Init()
+	{
+	Open();
+	}
+
+void PktSrc::Done()
+	{
+	if ( IsOpen() )
+		Close();
+	}
+
+void PktSrc::GetFds(std::vector<int>* read, std::vector<int>* write,
+                    std::vector<int>* except)
+	{
+	if ( pseudo_realtime )
+		{
+		// Select would give erroneous results. But we simulate it
+		// by setting idle accordingly.
+		SetIdle(CheckPseudoTime() == 0);
+		return;
+		}
+
+	if ( IsOpen() && props.selectable_fd >= 0 )
+		read->push_back(props.selectable_fd);
+	}
+
+double PktSrc::NextTimestamp(double* local_network_time)
+	{
+	if ( ! IsOpen() )
+		return -1.0;
+
+	if ( ! ExtractNextPacketInternal() )
+		return -1.0;
+
+	if ( pseudo_realtime )
+		{
+		// Delay packet if necessary.
+		double packet_time = CheckPseudoTime();
+		if ( packet_time )
+			return packet_time;
+
+		SetIdle(true);
+		return -1.0;
+		}
+
+	return current_packet.ts;
+	}
+
+void PktSrc::Process()
+	{
+	if ( ! IsOpen() )
+		return;
+
+	if ( ! ExtractNextPacketInternal() )
+		return;
+
+	int pkt_hdr_size = props.hdr_size;
+
+	// Unfortunately some packets on the link might have MPLS labels
+	// while others don't. That means we need to ask the link-layer if
+	// labels are in place.
+	bool have_mpls = false;
+
+	int protocol = 0;
+	const u_char* data = current_packet.data;
+
+	switch ( props.link_type ) {
+	case DLT_NULL:
+		{
+		protocol = (data[3] << 24) + (data[2] << 16) + (data[1] << 8) + data[0];
+
+		// From the Wireshark Wiki: "AF_INET6, unfortunately, has
+		// different values in {NetBSD,OpenBSD,BSD/OS},
+		// {FreeBSD,DragonFlyBSD}, and {Darwin/Mac OS X}, so an IPv6
+		// packet might have a link-layer header with 24, 28, or 30
+		// as the AF_ value." As we may be reading traces captured on
+		// platforms other than what we're running on, we accept them
+		// all here.
+		if ( protocol != AF_INET
+		     && protocol != AF_INET6
+		     && protocol != 24
+		     && protocol != 28
+		     && protocol != 30 )
+			{
+			Weird("non_ip_packet_in_null_transport", &current_packet);
+			goto done;
+			}
+
+		break;
+		}
+
+	case DLT_EN10MB:
+		{
+		// Get protocol being carried from the ethernet frame.
+		protocol = (data[12] << 8) + data[13];
+
+		switch ( protocol )
+			{
+			// MPLS carried over the ethernet frame.
+			case 0x8847:
+				// Remove the data link layer and denote a
+				// header size of zero before the IP header.
+				have_mpls = true;
+				data += GetLinkHeaderSize(props.link_type);
+				pkt_hdr_size = 0;
+				break;
+
+			// VLAN carried over the ethernet frame.
+			case 0x8100:
+				data += GetLinkHeaderSize(props.link_type);
+
+				// Check for MPLS in VLAN.
+				if ( ((data[2] << 8) + data[3]) == 0x8847 )
+					have_mpls = true;
+
+				data += 4; // Skip the vlan header
+				pkt_hdr_size = 0;
+
+				// Check for 802.1ah (Q-in-Q) containing IP.
+				// Only do a second layer of vlan tag
+				// stripping because there is no
+				// specification that allows for deeper
+				// nesting.
+				if ( ((data[2] << 8) + data[3]) == 0x0800 )
+					data += 4;
+
+				break;
+
+			// PPPoE carried over the ethernet frame.
+			case 0x8864:
+				data += GetLinkHeaderSize(props.link_type);
+				protocol = (data[6] << 8) + data[7];
+				data += 8; // Skip the PPPoE session and PPP header
+				pkt_hdr_size = 0;
+
+				if ( protocol != 0x0021 && protocol != 0x0057 )
+					{
+					// Neither IPv4 nor IPv6.
+					Weird("non_ip_packet_in_pppoe_encapsulation", &current_packet);
+					goto done;
+					}
+				break;
+			}
+
+		break;
+		}
+
+	case DLT_PPP_SERIAL:
+		{
+		// Get PPP protocol.
+		protocol = (data[2] << 8) + data[3];
+
+		if ( protocol == 0x0281 )
+			{
+			// MPLS Unicast. Remove the data link layer and
+			// denote a header size of zero before the IP header.
+			have_mpls = true;
+				data += GetLinkHeaderSize(props.link_type);
+			pkt_hdr_size = 0;
+			}
+
+		else if ( protocol != 0x0021 && protocol != 0x0057 )
+			{
+			// Neither IPv4 nor IPv6.
+			Weird("non_ip_packet_in_ppp_encapsulation", &current_packet);
+			goto done;
+			}
+		break;
+		}
+	}
+
+	if ( have_mpls )
+		{
+		// Skip the MPLS label stack.
+		bool end_of_stack = false;
+
+		while ( ! end_of_stack )
+			{
+			end_of_stack = *(data + 2) & 0x01;
+			data += 4;
+			}
+		}
+
+	if ( pseudo_realtime )
+		{
+		current_pseudo = CheckPseudoTime();
+		net_packet_dispatch(current_pseudo, current_packet.hdr, data, pkt_hdr_size, this);
+		if ( ! first_wallclock )
+			first_wallclock = current_time(true);
+		}
+
+	else
+		net_packet_dispatch(current_packet.ts, current_packet.hdr, data, pkt_hdr_size, this);
+
+done:
+	have_packet = 0;
+	DoneWithPacket();
+	}
+
+const char* PktSrc::Tag()
+	{
+	return "PktSrc";
+	}
+
+bool PktSrc::ExtractNextPacketInternal()
+	{
+	if ( have_packet )
+		return true;
+
+	have_packet = false;
+
+	// Don't return any packets if processing is suspended (except for the
+	// very first packet which we need to set up times).
+	if ( net_is_processing_suspended() && first_timestamp )
+		{
+		SetIdle(true);
+		return 0;
+		}
+
+	if ( pseudo_realtime )
+		current_wallclock = current_time(true);
+
+	if ( ExtractNextPacket(&current_packet) )
+		{
+		if ( ! first_timestamp )
+			first_timestamp = current_packet.ts;
+
+		have_packet = true;
+		return 1;
+		}
+
+	if ( pseudo_realtime && using_communication && ! IsOpen() )
+		{
+		// Source has gone dry, we're done.
+		if ( remote_trace_sync_interval )
+			remote_serializer->SendFinalSyncPoint();
+		else
+			remote_serializer->Terminate();
+		}
+
+	SetIdle(true);
+	return 0;
+	}
+
+bool PktSrc::PrecompileBPFFilter(int index, const std::string& filter)
+	{
+	if ( index < 0 )
+		return false;
+
+	char errbuf[PCAP_ERRBUF_SIZE];
+
+	// Compile filter.
+	BPF_Program* code = new BPF_Program();
+
+	if ( ! code->Compile(SnapLen(), LinkType(), filter.c_str(), Netmask(), errbuf, sizeof(errbuf)) )
+		{
+		string msg = fmt("cannot compile BPF filter \"%s\"", filter.c_str());
+
+		if ( *errbuf )
+			msg += ": " + string(errbuf);
+
+		Error(msg);
+
+		delete code;
+		return 0;
+		}
+
+	// Store it in hash.
+	HashKey* hash = new HashKey(HashKey(bro_int_t(index)));
+	BPF_Program* oldcode = filters.Lookup(hash);
+	if ( oldcode )
+		delete oldcode;
+
+	filters.Insert(hash, code);
+	delete hash;
+
+	return 1;
+	}
+
+BPF_Program* PktSrc::GetBPFFilter(int index)
+	{
+	if ( index < 0 )
+		return 0;
+
+	HashKey* hash = new HashKey(HashKey(bro_int_t(index)));
+	BPF_Program* code = filters.Lookup(hash);
+	delete hash;
+	return code;
+	}
+
+bool PktSrc::ApplyBPFFilter(int index, const struct pcap_pkthdr *hdr, const u_char *pkt)
+	{
+	BPF_Program* code = GetBPFFilter(index);
+
+	if ( ! code )
+		{
+		Error(fmt("BPF filter %d not compiled", index));
+		Close();
+		}
+
+	if ( code->MatchesAnything() )
+		return true;
+
+	return pcap_offline_filter(code->GetProgram(), hdr, pkt);
+	}
+
+bool PktSrc::GetCurrentPacket(const pcap_pkthdr** hdr, const u_char** pkt)
+	{
+	if ( ! have_packet )
+		return false;
+
+	*hdr = current_packet.hdr;
+	*pkt = current_packet.data;
+	return true;
+	}

src/iosource/PktSrc.h

@@ -0,0 +1,417 @@
+// See the file "COPYING" in the main distribution directory for copyright.
+
+#ifndef IOSOURCE_PKTSRC_PKTSRC_H
+#define IOSOURCE_PKTSRC_PKTSRC_H
+
+#include "IOSource.h"
+#include "BPF_Program.h"
+#include "Dict.h"
+
+declare(PDict,BPF_Program);
+
+namespace iosource {
+
+/**
+ * Base class for packet sources.
+ */
+class PktSrc : public IOSource {
+public:
+	/**
+	 * Struct for returning statistics on a packet source.
+	 */
+	struct Stats {
+		/**
+		 * Packets received by source after filtering (w/o drops).
+		 */
+		unsigned int received;
+
+		/**
+		 * Packets dropped by source.
+		 */
+		unsigned int dropped;	// pkts dropped
+
+		/**
+		 * Total number of packets on link before filtering.
+		 * Optional, can be left unset if not available.
+		 */
+		unsigned int link;
+
+		Stats()	{ received = dropped = link = 0; }
+	};
+
+	/**
+	 * Constructor.
+	 */
+	PktSrc();
+
+	/**
+	 * Destructor.
+	 */
+	virtual ~PktSrc();
+
+	/**
+	 * Returns the path associated with the source. This is the interface
+	 * name for live source, and a filename for offline sources.
+	 */
+	const std::string& Path() const;
+
+	/**
+	 * Returns true if this is a live source.
+	 */
+	bool IsLive() const;
+
+	/**
+	 * Returns the link type of the source.
+	 */
+	int LinkType() const;
+
+	/**
+	 * Returns the netmask associated with the source, or \c
+	 * PCAP_NETMASK_UNKNOWN if unknown.
+	 */
+	uint32 Netmask() const;
+
+	/**
+	 * Returns true if the source has flagged an error.
+	 */
+	bool IsError() const;
+
+	/**
+	 * If the source encountered an error, returns a corresponding error
+	 * message. Returns an empty string otherwise.
+	 */
+	const char* ErrorMsg() const;
+
+	/**
+	 * Returns the size of the link-layer header for this source.
+	 */
+	int HdrSize() const;
+
+	/**
+	 * Returns the snap length for this source.
+	 */
+	int SnapLen() const;
+
+	/**
+	 * In pseudo-realtime mode, returns the logical timestamp of the
+	 * current packet. Undefined if not running pseudo-realtime mode.
+	 */
+	double CurrentPacketTimestamp();
+
+	/**
+	 * In pseudo-realtime mode, returns the wall clock time associated
+	 * with current packet. Undefined if not running pseudo-realtime
+	 * mode.
+	 */
+	double CurrentPacketWallClock();
+
+	/**
+	 * Signals packet source that processing is going to be continued
+	 * after previous suspension.
+	 */
+	void ContinueAfterSuspend();
+
+	/**
+	 * Precompiles a BPF filter and associates the given index with it.
+	 * The compiled filter will be then available via \a GetBPFFilter().
+	 *
+	 * This is primarily a helper for packet source implementation that
+	 * want to apply BPF filtering to their packets.
+	 *
+	 * @param index The index to associate with the filter.
+	 *
+	 * @param BPF filter The filter string to precompile.
+	 *
+	 * @return True on success, false if a problem occurred.
+	 */
+	bool PrecompileBPFFilter(int index, const std::string& filter);
+
+	/**
+	 * Returns the precompiled BPF filter associated with a given index,
+	 * if any, as compiled by \a PrecompileBPFFilter().
+	 *
+	 * This is primarily a helper for packet source implementation that
+	 * want to apply BPF filtering to their packets.
+	 *
+	 * @return The BPF filter associated, or null if none has been
+	 * (successfully) compiled.
+	 */
+	BPF_Program* GetBPFFilter(int index);
+
+	/**
+	 * Applies a precompiled BPF filter to a packet. This will close the
+	 * source with an error message if no filter with that index has been
+	 * compiled.
+	 *
+	 * This is primarily a helper for packet source implementation that
+	 * want to apply BPF filtering to their packets.
+	 *
+	 * @param index The index of the filter to apply.
+	 *
+	 * @param hdr The header of the packet to filter.
+	 *
+	 * @param pkt The content of the packet to filter.
+	 *
+	 * @return True if it maches. 	 */
+	bool ApplyBPFFilter(int index, const struct pcap_pkthdr *hdr, const u_char *pkt);
+
+	/**
+	 * Returns the packet currently being processed, if available.
+	 *
+	 * @param hdr A pointer to pass the header of the current packet back.
+	 *
+	 * @param pkt A pointer to pass the content of the current packet
+	 * back.
+	 *
+	 * @return True if the current packet is available, or false if not.
+	 */
+	bool GetCurrentPacket(const pcap_pkthdr** hdr, const u_char** pkt);
+
+	// PacketSource interace for derived classes to override.
+
+	/**
+	 * Precompiles a filter and associates a given index with it. The
+	 * filter syntax is defined by the packet source's implenentation.
+	 *
+	 * Derived classes must implement this to implement their filtering.
+	 * If they want to use BPF but don't support it natively, they can
+	 * call the corresponding helper method provided by \a PktSrc.
+	 *
+	 * @param index The index to associate with the filter
+	 *
+	 * @param filter The filter string to precompile.
+	 *
+	 * @return True on success, false if a problem occurred or filtering
+	 * is not supported.
+	 */
+	virtual bool PrecompileFilter(int index, const std::string& filter) = 0;
+
+	/**
+	 * Activates a precompiled filter with the given index.
+	 *
+	 * Derived classes must implement this to implement their filtering.
+	 * If they want to use BPF but don't support it natively, they can
+	 * call the corresponding helper method provided by \a PktSrc.
+	 *
+	 * @param index The index of the filter to activate.
+	 *
+	 * @return True on success, false if a problem occurred or the
+	 * filtering is not supported.
+	 */
+	virtual bool SetFilter(int index) = 0;
+
+	/**
+	 * Returns current statistics about the source.
+	 *
+	 * Derived classes must implement this method.
+	 *
+	 * @param stats A statistics structure that the method fill out.
+	 */
+	virtual void Statistics(Stats* stats) = 0;
+
+	/**
+	 * Helper method to return the header size for a given link tyoe.
+	 *
+	 * @param link_type The link tyoe.
+	 *
+	 * @return The header size in bytes.
+	 */
+	static int GetLinkHeaderSize(int link_type);
+
+protected:
+	friend class Manager;
+
+	// Methods to use by derived classes.
+
+	/**
+	 * Structure to pass back information about the packet source to the
+	 * base class. Derived class pass an instance of this to \a Opened().
+	 */
+	struct Properties {
+		/**
+		 * The path associated with the source. This is the interface
+		 * name for live source, and a filename for offline sources.
+		 */
+		std::string path;
+
+		/**
+		 * A file descriptor suitable to use with \a select() for
+		 * determining if there's input available from this source.
+		 */
+		int selectable_fd;
+
+		/**
+		 * The link type for packets from this source.
+		 */
+		int link_type;
+
+		/**
+		 * The size of the link-layer header for packets from this
+		 * source. \a GetLinkHeaderSize() may be used to derive this
+		 * value.
+		 */
+		int hdr_size;
+
+		/**
+		 * The netmask associated with the source, or \c
+		 * PCAP_NETMASK_UNKNOWN if unknown.
+		 */
+		uint32 netmask;
+
+		/**
+		 * True if the source is reading live inout, false for
+		 * working offline.
+		 */
+		bool is_live;
+
+		Properties()
+			{
+			netmask = PCAP_NETMASK_UNKNOWN;
+			}
+	};
+
+	/**
+	 * Structure describing a packet.
+	 */
+	struct Packet {
+		/**
+		 *  Time associated with the packet.
+		 */
+		double ts;
+
+		/**
+		 *  The pcap header associated with the packet.
+		 */
+		const struct ::pcap_pkthdr* hdr;
+
+		/**
+		 * The full content of the packet.
+		 */
+		const u_char* data;
+	};
+
+	/**
+	 * Called from the implementations of \a Open() to signal that the
+	 * source has been successully opened.
+	 *
+	 * @param props A properties instance describing the now open source.
+	 */
+	void Opened(const Properties& props);
+
+	/**
+	 * Called from the implementations of \a Close() to signal that the
+	 * source has been closed.
+	 */
+	void Closed();
+
+	/**
+	 * Can be called from derived classes to send an informational
+	 * message to the user.
+	 *
+	 * @param msg The message to pass on.
+	 */
+	void Info(const std::string& msg);
+
+	/**
+	 * Can be called from derived classes to flag send an error.
+	 *
+	 * @param msg The message going with the error.
+	 */
+	void Error(const std::string& msg);
+
+	/**
+	 * Can be called from derived classes to flah a "weird" situation.
+	 *
+	 * @param msg The message to pass on.
+	 *
+	 * @param pkt The packet associated with the weird, or null if none.
+	 */
+	void Weird(const std::string& msg, const Packet* pkt);
+
+	/**
+	 * Can be called from derived classes to flag an internal error,
+	 * which will abort execution.
+	 *
+	 * @param msg The message to pass on.
+	 */
+	void InternalError(const std::string& msg);
+
+	// PktSrc interface for derived classes to implement.
+
+	/**
+	 * Called by the manager system to open the source.
+	 *
+	 * Derived classes must implement this method. If successful, the
+	 * implementation must call \a Opened(); if not, it must call Error()
+	 * with a corresponding message.
+	 */
+	virtual void Open() = 0;
+
+	/**
+	 * Called by the manager system to close the source.
+	 *
+	 * Derived classes must implement this method. If successful, the
+	 * implementation must call \a Closed(); if not, it must call Error()
+	 * with a corresponding message.
+	 */
+	virtual void Close() = 0;
+
+	/**
+	 * Provides the next packet from the source.
+	 *
+	 * @param pkt The packet structure to fill in with the packet's
+	 * information. The callee keep ownership of the data but must
+	 * guaranetee that it stays available at least until \a
+	 * DoneWithPacket() is called.  It is guaranteed that no two calls to
+	 * this method will hapen with \a DoneWithPacket() in between.
+	 *
+	 * @return True if a packet is available and *pkt* filled in. False
+	 * if not packet is available or an error occured (which must be
+	 * flageed via Error()).
+	 */
+	virtual bool ExtractNextPacket(Packet* pkt) = 0;
+
+	/**
+	 * Signals that the data of previously extracted packet will no
+	 * longer be needed.
+	 */
+	virtual void DoneWithPacket() = 0;
+
+private:
+	// Checks if the current packet has a pseudo-time <= current_time. If
+	// yes, returns pseudo-time, otherwise 0.
+	double CheckPseudoTime();
+
+	// Internal helper for ExtractNextPacket().
+	bool ExtractNextPacketInternal();
+
+	// IOSource interface implementation.
+	virtual void Init();
+	virtual void Done();
+	virtual void GetFds(std::vector<int>* read, std::vector<int>* write,
+	                    std::vector<int>* except);
+	virtual double NextTimestamp(double* local_network_time);
+	virtual void Process();
+	virtual const char* Tag();
+
+	Properties props;
+
+	bool have_packet;
+	Packet current_packet;
+
+	// For BPF filtering support.
+	PDict(BPF_Program) filters;
+
+	// Only set in pseudo-realtime mode.
+	double first_timestamp;
+	double first_wallclock;
+	double current_wallclock;
+	double current_pseudo;
+	double next_sync_point; // For trace synchronziation in pseudo-realtime
+
+	std::string errbuf;
+};
+
+}
+
+#endif

src/iosource/pcap.bif

@@ -0,0 +1,104 @@
+
+## Precompiles a PCAP filter and binds it to a given identifier.
+##
+## id: The PCAP identifier to reference the filter *s* later on.
+##
+## s: The PCAP filter. See ``man tcpdump`` for valid expressions.
+##
+## Returns: True if *s* is valid and precompiles successfully.
+##
+## .. bro:see:: install_pcap_filter
+##          install_src_addr_filter
+##          install_src_net_filter
+##          uninstall_src_addr_filter
+##          uninstall_src_net_filter
+##          install_dst_addr_filter
+##          install_dst_net_filter
+##          uninstall_dst_addr_filter
+##          uninstall_dst_net_filter
+##          pcap_error
+function precompile_pcap_filter%(id: PcapFilterID, s: string%): bool
+	%{
+	bool success = true;
+
+	const iosource::Manager::PktSrcList& pkt_srcs(iosource_mgr->GetPktSrcs());
+
+	for ( iosource::Manager::PktSrcList::const_iterator i = pkt_srcs.begin();
+	      i != pkt_srcs.end(); i++ )
+		{
+		iosource::PktSrc* ps = *i;
+
+		if ( ! ps->PrecompileFilter(id->ForceAsInt(),
+							s->CheckString()) )
+			success = false;
+		}
+
+	return new Val(success, TYPE_BOOL);
+	%}
+
+## Installs a PCAP filter that has been precompiled with
+## :bro:id:`precompile_pcap_filter`.
+##
+## id: The PCAP filter id of a precompiled filter.
+##
+## Returns: True if the filter associated with *id* has been installed
+##          successfully.
+##
+## .. bro:see:: precompile_pcap_filter
+##              install_src_addr_filter
+##              install_src_net_filter
+##              uninstall_src_addr_filter
+##              uninstall_src_net_filter
+##              install_dst_addr_filter
+##              install_dst_net_filter
+##              uninstall_dst_addr_filter
+##              uninstall_dst_net_filter
+##              pcap_error
+function install_pcap_filter%(id: PcapFilterID%): bool
+	%{
+	bool success = true;
+
+	const iosource::Manager::PktSrcList& pkt_srcs(iosource_mgr->GetPktSrcs());
+
+	for ( iosource::Manager::PktSrcList::const_iterator i = pkt_srcs.begin();
+	      i != pkt_srcs.end(); i++ )
+		{
+		iosource::PktSrc* ps = *i;
+
+		if ( ! ps->SetFilter(id->ForceAsInt()) )
+			success = false;
+		}
+
+	return new Val(success, TYPE_BOOL);
+	%}
+
+## Returns a string representation of the last PCAP error.
+##
+## Returns: A descriptive error message of the PCAP function that failed.
+##
+## .. bro:see:: precompile_pcap_filter
+##              install_pcap_filter
+##              install_src_addr_filter
+##              install_src_net_filter
+##              uninstall_src_addr_filter
+##              uninstall_src_net_filter
+##              install_dst_addr_filter
+##              install_dst_net_filter
+##              uninstall_dst_addr_filter
+##              uninstall_dst_net_filter
+function pcap_error%(%): string
+	%{
+	const iosource::Manager::PktSrcList& pkt_srcs(iosource_mgr->GetPktSrcs());
+
+	for ( iosource::Manager::PktSrcList::const_iterator i = pkt_srcs.begin();
+	      i != pkt_srcs.end(); i++ )
+		{
+		iosource::PktSrc* ps = *i;
+
+		const char* err = ps->ErrorMsg();
+		if ( *err )
+			return new StringVal(err);
+		}
+
+	return new StringVal("no error");
+	%}

src/iosource/pcap/CMakeLists.txt

@@ -0,0 +1,8 @@
+
+include(BroPlugin)
+
+include_directories(BEFORE ${CMAKE_CURRENT_SOURCE_DIR} ${CMAKE_CURRENT_BINARY_DIR})
+
+bro_plugin_begin(Bro Pcap)
+bro_plugin_cc(Source.cc Dumper.cc Plugin.cc)
+bro_plugin_end()

src/iosource/pcap/Dumper.cc

@@ -0,0 +1,112 @@
+// See the file "COPYING" in the main distribution directory for copyright.
+
+#include <sys/stat.h>
+#include <errno.h>
+
+#include "Dumper.h"
+#include "../PktSrc.h"
+#include "../../Net.h"
+
+using namespace iosource::pcap;
+
+PcapDumper::PcapDumper(const std::string& path, bool arg_append)
+	{
+	append = arg_append;
+	props.path = path;
+	dumper = 0;
+	pd = 0;
+	}
+
+PcapDumper::~PcapDumper()
+	{
+	}
+
+void PcapDumper::Open()
+	{
+	int linktype = -1;
+
+	pd = pcap_open_dead(DLT_EN10MB, snaplen);
+	if ( ! pd )
+		{
+		Error("error for pcap_open_dead");
+		return;
+		}
+
+	if ( props.path.empty() )
+		{
+		Error("no filename given");
+		return;
+		}
+
+	struct stat s;
+	int exists = 0;
+
+	if ( append )
+		{
+		// See if output file already exists (and is non-empty).
+		exists = stat(props.path.c_str(), &s); ;
+
+		if ( exists < 0 && errno != ENOENT )
+			{
+			Error(fmt("can't stat file %s: %s", props.path.c_str(), strerror(errno)));
+			return;
+			}
+		}
+
+	if ( ! append || exists < 0 || s.st_size == 0 )
+		{
+		// Open new file.
+		dumper = pcap_dump_open(pd, props.path.c_str());
+		if ( ! dumper )
+			{
+			Error(pcap_geterr(pd));
+			return;
+			}
+		}
+
+	else
+		{
+		// Old file and we need to append, which, unfortunately,
+		// is not supported by libpcap. So, we have to hack a
+		// little bit, knowing that pcap_dumpter_t is, in fact,
+		// a FILE ... :-(
+		dumper = (pcap_dumper_t*) fopen(props.path.c_str(), "a");
+		if ( ! dumper )
+			{
+			Error(fmt("can't open dump %s: %s", props.path.c_str(), strerror(errno)));
+			return;
+			}
+		}
+
+	props.open_time = network_time;
+	props.hdr_size = PktSrc::GetLinkHeaderSize(pcap_datalink(pd));
+	Opened(props);
+	}
+
+void PcapDumper::Close()
+	{
+	if ( ! dumper )
+		return;
+
+	pcap_dump_close(dumper);
+	pcap_close(pd);
+	dumper = 0;
+	pd = 0;
+
+	Closed();
+	}
+
+bool PcapDumper::Dump(const Packet* pkt)
+	{
+	if ( ! dumper )
+		return false;
+
+	pcap_dump((u_char*) dumper, pkt->hdr, pkt->data);
+
+	return true;
+	}
+
+iosource::PktDumper* PcapDumper::Instantiate(const std::string& path, bool append)
+	{
+	return new PcapDumper(path, append);
+	}

src/iosource/pcap/Dumper.h

@@ -0,0 +1,41 @@
+// See the file  in the main distribution directory for copyright.
+
+#ifndef IOSOURCE_PKTSRC_PCAP_DUMPER_H
+#define IOSOURCE_PKTSRC_PCAP_DUMPER_H
+
+extern "C" {
+#include <pcap.h>
+}
+
+#include "../PktDumper.h"
+
+namespace iosource {
+namespace pcap {
+
+class PcapDumper : public PktDumper {
+public:
+	PcapDumper(const std::string& path, bool append);
+	virtual ~PcapDumper();
+
+	static PktDumper* Instantiate(const std::string& path, bool appen);
+
+protected:
+	// PktDumper interface.
+	virtual void Open();
+	virtual void Close();
+	virtual bool Dump(const Packet* pkt);
+
+private:
+	Properties props;
+
+	bool append;
+	pcap_dumper_t* dumper;
+	pcap_t* pd;
+};
+
+}
+}
+
+#endif
+
+

src/iosource/pcap/Plugin.cc

@@ -0,0 +1,27 @@
+// See the file  in the main distribution directory for copyright.
+
+#include "plugin/Plugin.h"
+
+#include "Source.h"
+#include "Dumper.h"
+
+namespace plugin {
+namespace Bro_Pcap {
+
+class Plugin : public plugin::Plugin {
+public:
+	plugin::Configuration Configure()
+		{
+		AddComponent(new ::iosource::PktSrcComponent("PcapReader", "pcap", ::iosource::PktSrcComponent::BOTH, ::iosource::pcap::PcapSource::Instantiate));
+		AddComponent(new ::iosource::PktDumperComponent("PcapWriter", "pcap", ::iosource::pcap::PcapDumper::Instantiate));
+
+		plugin::Configuration config;
+		config.name = "Bro::Pcap";
+		config.description = "Packet aquisition via libpcap";
+		return config;
+		}
+} plugin;
+
+}
+}
+

src/iosource/pcap/Source.cc

@@ -0,0 +1,272 @@
+// See the file  in the main distribution directory for copyright.
+
+#include <assert.h>
+
+#include "config.h"
+
+#include "Source.h"
+
+#ifdef HAVE_PCAP_INT_H
+#include <pcap-int.h>
+#endif
+
+using namespace iosource::pcap;
+
+PcapSource::~PcapSource()
+	{
+	Close();
+	}
+
+PcapSource::PcapSource(const std::string& path, bool is_live)
+	{
+	props.path = path;
+	props.is_live = is_live;
+	last_data = 0;
+	}
+
+void PcapSource::Open()
+	{
+	if ( props.is_live )
+		OpenLive();
+	else
+		OpenOffline();
+	}
+
+void PcapSource::Close()
+	{
+	if ( ! pd )
+		return;
+
+	pcap_close(pd);
+	pd = 0;
+	last_data = 0;
+
+	Closed();
+	}
+
+void PcapSource::OpenLive()
+	{
+	char errbuf[PCAP_ERRBUF_SIZE];
+	char tmp_errbuf[PCAP_ERRBUF_SIZE];
+
+	// Determine interface if not specified.
+	if ( props.path.empty() )
+		props.path = pcap_lookupdev(tmp_errbuf);
+
+	if ( props.path.empty() )
+		{
+		safe_snprintf(errbuf, sizeof(errbuf),
+			 "pcap_lookupdev: %s", tmp_errbuf);
+		Error(errbuf);
+		return;
+		}
+
+	// Determine network and netmask.
+	uint32 net;
+	if ( pcap_lookupnet(props.path.c_str(), &net, &props.netmask, tmp_errbuf) < 0 )
+		{
+		// ### The lookup can fail if no address is assigned to
+		// the interface; and libpcap doesn't have any useful notion
+		// of error codes, just error std::strings - how bogus - so we
+		// just kludge around the error :-(.
+		// sprintf(errbuf, "pcap_lookupnet %s", tmp_errbuf);
+		// return;
+		props.netmask = 0xffffff00;
+		}
+
+	// We use the smallest time-out possible to return almost immediately if
+	// no packets are available. (We can't use set_nonblocking() as it's
+	// broken on FreeBSD: even when select() indicates that we can read
+	// something, we may get nothing if the store buffer hasn't filled up
+	// yet.)
+	pd = pcap_open_live(props.path.c_str(), SnapLen(), 1, 1, tmp_errbuf);
+
+	if ( ! pd )
+		{
+		Error(tmp_errbuf);
+		return;
+		}
+
+	// ### This needs autoconf'ing.
+#ifdef HAVE_PCAP_INT_H
+	Info("pcap bufsize = %d\n", ((struct pcap *) pd)->bufsize);
+#endif
+
+#ifdef HAVE_LINUX
+	if ( pcap_setnonblock(pd, 1, tmp_errbuf) < 0 )
+		{
+		PcapError();
+		return;
+		}
+#endif
+
+	props.selectable_fd = pcap_fileno(pd);
+
+	SetHdrSize();
+
+	if ( ! pd )
+		// Was closed, couldn't get header size.
+		return;
+
+	props.is_live = true;
+
+	Opened(props);
+	}
+
+void PcapSource::OpenOffline()
+	{
+	char errbuf[PCAP_ERRBUF_SIZE];
+
+	pd = pcap_open_offline(props.path.c_str(), errbuf);
+
+	if ( ! pd )
+		{
+		Error(errbuf);
+		return;
+		}
+
+	SetHdrSize();
+
+	if ( ! pd )
+		// Was closed, unknown link layer type.
+		return;
+
+	props.selectable_fd = fileno(pcap_file(pd));
+
+	if ( props.selectable_fd < 0 )
+		InternalError("OS does not support selectable pcap fd");
+
+	props.is_live = false;
+	Opened(props);
+	}
+
+bool PcapSource::ExtractNextPacket(Packet* pkt)
+	{
+	if ( ! pd )
+		return false;
+
+	const u_char* data = pcap_next(pd, &current_hdr);
+
+	if ( ! data )
+		{
+		// Source has gone dry.  If it's a network interface, this just means
+		// it's timed out. If it's a file, though, then the file has been
+		// exhausted.
+		if ( ! props.is_live )
+			Close();
+
+		return false;
+		}
+
+	pkt->ts = current_hdr.ts.tv_sec + double(current_hdr.ts.tv_usec) / 1e6;
+	pkt->hdr = &current_hdr;
+	pkt->data = last_data = data;
+
+	if ( current_hdr.len == 0 || current_hdr.caplen == 0 )
+		{
+		Weird("empty_pcap_header", pkt);
+		return false;
+		}
+
+	last_hdr = current_hdr;
+	last_data = data;
+	++stats.received;
+	return true;
+	}
+
+void PcapSource::DoneWithPacket()
+	{
+	// Nothing to do.
+	}
+
+bool PcapSource::PrecompileFilter(int index, const std::string& filter)
+	{
+	return PktSrc::PrecompileBPFFilter(index, filter);
+	}
+
+bool PcapSource::SetFilter(int index)
+	{
+	if ( ! pd )
+		return true; // Prevent error message
+
+	char errbuf[PCAP_ERRBUF_SIZE];
+
+	BPF_Program* code = GetBPFFilter(index);
+
+	if ( ! code )
+		{
+		safe_snprintf(errbuf, sizeof(errbuf),
+			      "No precompiled pcap filter for index %d",
+			      index);
+		Error(errbuf);
+		return false;
+		}
+
+	if ( pcap_setfilter(pd, code->GetProgram()) < 0 )
+		{
+		PcapError();
+		return false;
+		}
+
+#ifndef HAVE_LINUX
+	// Linux doesn't clear counters when resetting filter.
+	stats.received = stats.dropped = stats.link = 0;
+#endif
+
+	return true;
+	}
+
+void PcapSource::Statistics(Stats* s)
+	{
+	char errbuf[PCAP_ERRBUF_SIZE];
+
+	if ( ! (props.is_live && pd) )
+		s->received = s->dropped = s->link = 0;
+
+	else
+		{
+		struct pcap_stat pstat;
+		if ( pcap_stats(pd, &pstat) < 0 )
+			{
+			PcapError();
+			s->received = s->dropped = s->link = 0;
+			}
+
+		else
+			{
+			s->dropped = pstat.ps_drop;
+			s->link = pstat.ps_recv;
+			}
+		}
+
+	s->received = stats.received;
+
+	if ( ! props.is_live )
+		s->dropped = 0;
+	}
+
+void PcapSource::PcapError()
+	{
+	if ( pd )
+		Error(fmt("pcap_error: %s", pcap_geterr(pd)));
+	else
+		Error("pcap_error: not open");
+
+	Close();
+	}
+
+void PcapSource::SetHdrSize()
+	{
+	if ( ! pd )
+		return;
+
+	char errbuf[PCAP_ERRBUF_SIZE];
+
+	props.link_type = pcap_datalink(pd);
+	props.hdr_size = GetLinkHeaderSize(props.link_type);
+	}
+
+iosource::PktSrc* PcapSource::Instantiate(const std::string& path, bool is_live)
+	{
+	return new PcapSource(path, is_live);
+	}

src/iosource/pcap/Source.h

@@ -0,0 +1,47 @@
+// See the file "COPYING" in the main distribution directory for copyright.
+
+#ifndef IOSOURCE_PKTSRC_PCAP_SOURCE_H
+#define IOSOURCE_PKTSRC_PCAP_SOURCE_H
+
+#include "../PktSrc.h"
+
+namespace iosource {
+namespace pcap {
+
+class PcapSource : public iosource::PktSrc {
+public:
+	PcapSource(const std::string& path, bool is_live);
+	virtual ~PcapSource();
+
+	static PktSrc* Instantiate(const std::string& path, bool is_live);
+
+protected:
+	// PktSrc interface.
+	virtual void Open();
+	virtual void Close();
+	virtual bool ExtractNextPacket(Packet* pkt);
+	virtual void DoneWithPacket();
+	virtual bool PrecompileFilter(int index, const std::string& filter);
+	virtual bool SetFilter(int index);
+	virtual void Statistics(Stats* stats);
+
+private:
+	void OpenLive();
+	void OpenOffline();
+	void PcapError();
+	void SetHdrSize();
+
+	Properties props;
+	Stats stats;
+
+	pcap_t *pd;
+
+	struct pcap_pkthdr current_hdr;
+	struct pcap_pkthdr last_hdr;
+	const u_char* last_data;
+};
+
+}
+}
+
+#endif