Merge branch 'topic/bernhard/input' into topic/bernhard/input-threads

most stuff is inplace, logging framework needs a few changes merged before continuing here...

Conflicts:
	src/CMakeLists.txt
	src/LogMgr.h
	src/logging/Manager.cc
	src/main.cc

doc/index.rst

@@ -24,6 +24,7 @@ Frameworks
 
    notice
    logging
+   input
    cluster
    signatures
 

doc/input.rst

@@ -0,0 +1,190 @@
+=====================
+Loading Data into Bro
+=====================
+
+.. rst-class:: opening
+
+   Bro comes with a flexible input interface that allows to read
+   previously stored data. Data is either read into bro tables or
+   sent to scripts using events.
+   This document describes how the input framework can be used.
+
+.. contents::
+
+Terminology
+===========
+
+Bro's input framework is built around three main abstracts, that are 
+very similar to the abstracts used in the logging framework:
+
+        Input Streams
+                An input stream corresponds to a single input source 
+                (usually a textfile). It defined the information necessary
+                to find the source (e.g. the filename)
+
+        Filters
+                Each input stream has a set of filters attached to it, that
+                determine exaclty what kind of information is read.
+                There are two different kind of streams, event streams and table 
+                streams.
+                By default, event streams generate an event for each line read
+                from the input source.
+                Table streams on the other hand read the input source in a bro
+                table for easy later access.
+
+        Readers
+                A reader defines the input format for the specific input stream.
+                At the moment, Bro comes with only one type of reader, which can
+                read the tab seperated ASCII logfiles that were generated by the
+                logging framework.
+
+
+Basics
+======
+
+For examples, please look at the unit tests in
+``testing/btest/scripts/base/frameworks/input/``.
+
+A very basic example to open an input stream is:
+
+.. code:: bro
+        
+        module Foo;
+
+        export {
+                # Create an ID for our new stream
+                redef enum Input::ID += { INPUT };
+        }
+
+        event bro_init() {
+                Input::create_stream(FOO::INPUT, [$source="input.log"]);
+        }
+
+The fields that can be set when creating a stream are:
+
+        ``source``
+                A mandatory string identifying the source of the data.
+                For the ASCII reader this is the filename.
+
+        ``reader``
+                The reader used for this stream. Default is ``READER_ASCII``.
+        
+
+Filters
+=======
+
+Each filter defines the data fields that it wants to receive from the respective 
+input file. Depending on the type of filter, events or a table are created from 
+the data in the source file.
+
+Event Filters
+-------------
+
+Event filters are filters that generate an event for each line in of the input source.
+
+For example, a simple filter retrieving the fields ``i`` and ``b`` from an inputSource
+could be defined as follows:
+
+.. code:: bro
+
+        type Val: record {
+	        i: int;
+                b: bool;
+        };
+
+        event line(tpe: Input::Event, i: int, b: bool) {
+                # work with event data
+        }
+        
+        event bro_init {
+                # Input stream definition, etc
+                ...
+
+                Input::add_eventfilter(Foo::INPUT, [$name="input", $fields=Val, $ev=line]);
+
+                # read the file after all filters have been set
+                Input::force_update(Foo::INPUT); 
+        }
+
+The fields that can be set for an event filter are:
+
+        ``name``
+                A mandatory name for the filter that can later be used
+                to manipulate it further.
+
+        ``fields``
+                Name of a record type containing the fields, which should be retrieved from
+                the input stream.
+
+        ``ev``
+                The event which is fired, after a line has been read from the input source.
+                The first argument that is passed to the event is an Input::Event structure, 
+                followed by the data, either inside of a record (if ``want_record is set``) or as 
+                individual fields.
+                The Input::Event structure can contain information, if the received line is ``NEW``, has
+                been ``CHANGED`` or ``DELETED``. Singe the ascii reader cannot track this information
+                for event filters, the value is always ``NEW`` at the moment.
+
+        ``want_record``
+                Boolean value, that defines if the event wants to receive the fields inside of
+                a single record value, or individually (default).
+
+Table Filters
+-------------
+
+Table filters are the second, more complex type of filter.
+
+Table filters store the information they read from an input source in a bro table. For example,
+when reading a file that contains ip addresses and connection attemt information one could use
+an approach similar to this:
+
+.. code:: bro
+
+        type Idx: record {
+                a: addr;
+        };
+
+        type Val: record {
+                tries: count;
+        };
+
+        global conn_attempts: table[addr] of count = table();
+
+        event bro_init {
+                # Input stream definitions, etc.
+                ...
+
+                Input::add_tablefilter(Foo::INPUT, [$name="ssh", $idx=Idx, $val=Val, $destination=conn_attempts]);
+
+                # read the file after all filters have been set
+                Input::force_update(Foo::INPUT);
+        }
+
+The table conn_attempts will then contain the information about connection attemps.
+
+The possible fields that can be set for an table filter are:
+
+        ``name``
+                A mandatory name for the filter that can later be used
+                to manipulate it further.
+        
+        ``idx``
+                Record type that defines the index of the table
+
+        ``val``
+                Record type that defines the values of the table
+
+        ``want_record``
+                Defines if the values of the table should be stored as a record (default),
+                or as a simple value. Has to be set if Val contains more than one element.
+
+        ``destination``
+                The destination table
+
+        ``ev``
+                Optional event that is raised, when values are added to, changed in or deleted from the table.
+                Events are passed an Input::Event description as the first argument, the index record as the second argument
+                and the values as the third argument.
+
+        ``pred``
+                Optional predicate, that can prevent entries from being added to the table and events from being sent.

doc/scripts/DocSourcesList.cmake

@@ -19,6 +19,7 @@ rest_target(${psd} base/init-bare.bro internal)
 rest_target(${CMAKE_BINARY_DIR}/src base/bro.bif.bro)
 rest_target(${CMAKE_BINARY_DIR}/src base/const.bif.bro)
 rest_target(${CMAKE_BINARY_DIR}/src base/event.bif.bro)
+rest_target(${CMAKE_BINARY_DIR}/src base/input.bif.bro)
 rest_target(${CMAKE_BINARY_DIR}/src base/logging.bif.bro)
 rest_target(${CMAKE_BINARY_DIR}/src base/reporter.bif.bro)
 rest_target(${CMAKE_BINARY_DIR}/src base/strings.bif.bro)
@@ -31,6 +32,8 @@ rest_target(${psd} base/frameworks/cluster/setup-connections.bro)
 rest_target(${psd} base/frameworks/communication/main.bro)
 rest_target(${psd} base/frameworks/control/main.bro)
 rest_target(${psd} base/frameworks/dpd/main.bro)
+rest_target(${psd} base/frameworks/input/main.bro)
+rest_target(${psd} base/frameworks/input/readers/ascii.bro)
 rest_target(${psd} base/frameworks/intel/main.bro)
 rest_target(${psd} base/frameworks/logging/main.bro)
 rest_target(${psd} base/frameworks/logging/postprocessors/scp.bro)

scripts/base/frameworks/input/__load__.bro

@@ -0,0 +1,3 @@
+@load ./main
+@load ./readers/ascii
+

scripts/base/frameworks/input/main.bro

@@ -0,0 +1,192 @@
+##! The input framework provides a way to read previously stored data either
+##! as an event stream or into a bro table.
+
+module Input;
+
+export {
+	redef enum Input::ID += { TABLE_READ };
+	
+	## The default input reader used. Defaults to `READER_ASCII`.
+	const default_reader = READER_ASCII &redef;
+
+	## Stream decription type used for the `create_stream` method
+	type StreamDescription: record {
+		## String that allows the reader to find the source.
+		## For `READER_ASCII`, this is the filename.
+		source: string;
+		
+		## Reader to use for this steam
+		reader: Reader &default=default_reader;
+	};
+
+	## TableFilter description type used for the `add_tablefilter` method.
+	type TableFilter: record {
+		## Descriptive name. Used to remove a filter at a later time
+		name: string; 
+
+		## Table which will contain the data read by the input framework
+		destination: any;
+		## Record that defines the values used as the index of the table
+		idx: any;
+		## Record that defines the values used as the values of the table
+		## If val is undefined, destination has to be a set.
+		val: any &optional;
+		## Defines if the value of the table is a record (default), or a single value.
+		## Val can only contain one element when this is set to false.
+		want_record: bool &default=T;
+
+		## The event that is raised each time a value is added to, changed in or removed from the table.
+		## The event will receive an Input::Event enum as the first argument, the idx record as the second argument
+		## and the value (record) as the third argument.
+		ev: any &optional; # event containing idx, val as values.		
+
+		## Predicate function, that can decide if an insertion, update or removal should really be executed.
+		## Parameters are the same as for the event. If true is returned, the update is performed. If false
+		## is returned, it is skipped
+		pred: function(typ: Input::Event, left: any, right: any): bool &optional;
+	};
+
+	## EventFilter description type used for the `add_eventfilter` method.
+	type EventFilter: record {
+		## Descriptive name. Used to remove a filter at a later time
+		name: string;
+
+		## Record describing the fields to be retrieved from the source input.
+		fields: any;
+		## If want_record if false (default), the event receives each value in fields as a seperate argument.
+		## If it is set to true, the event receives all fields in a signle record value.
+		want_record: bool &default=F;
+
+		## The event that is rised each time a new line is received from the reader.
+		## The event will receive an Input::Event enum as the first element, and the fields as the following arguments.
+		ev: any; 
+
+	};
+
+	#const no_filter: Filter = [$name="<not found>", $idx="", $val="", $destination=""]; # Sentinel.
+
+	## Create a new input stream from a given source. Returns true on success.
+	##
+	## id: `Input::ID` enum value identifying this stream
+	## description: `StreamDescription` record describing the source.
+	global create_stream: function(id: Input::ID, description: Input::StreamDescription) : bool;
+
+	## Remove a current input stream. Returns true on success.
+	##
+	## id: `Input::ID` enum value identifying the stream to be removed
+	global remove_stream: function(id: Input::ID) : bool;
+
+	## Forces the current input to be checked for changes.
+	##
+	## id: `Input::ID` enum value identifying the stream
+	global force_update: function(id: Input::ID) : bool;
+
+	## Adds a table filter to a specific input stream. Returns true on success.
+	##
+	## id: `Input::ID` enum value identifying the stream
+	## filter: the `TableFilter` record describing the filter.
+	global add_tablefilter: function(id: Input::ID, filter: Input::TableFilter) : bool;
+
+	## Removes a named table filter to a specific input stream. Returns true on success.
+	##
+	## id: `Input::ID` enum value identifying the stream
+	## name: the name of the filter to be removed.
+	global remove_tablefilter: function(id: Input::ID, name: string) : bool;
+
+	## Adds an event filter to a specific input stream. Returns true on success.
+	##
+	## id: `Input::ID` enum value identifying the stream
+	## filter: the `EventFilter` record describing the filter.
+	global add_eventfilter: function(id: Input::ID, filter: Input::EventFilter) : bool;
+
+	## Removes a named event filter to a specific input stream. Returns true on success.
+	##
+	## id: `Input::ID` enum value identifying the stream
+	## name: the name of the filter to be removed.
+	global remove_eventfilter: function(id: Input::ID, name: string) : bool;
+	#global get_filter: function(id: ID, name: string) : Filter;
+	
+	## Convenience function for reading a specific input source exactly once using 
+	## exactly one tablefilter
+	##
+	## id: `Input::ID` enum value identifying the stream
+	## description: `StreamDescription` record describing the source.
+	## filter: the `TableFilter` record describing the filter.	
+	global read_table: function(description: Input::StreamDescription, filter: Input::TableFilter) : bool;
+
+}
+
+@load base/input.bif
+
+
+module Input;
+
+#global filters: table[ID, string] of Filter;
+
+function create_stream(id: Input::ID, description: Input::StreamDescription) : bool
+	{
+	return __create_stream(id, description);
+	}
+
+function remove_stream(id: Input::ID) : bool
+	{
+	return __remove_stream(id);
+	}
+
+function force_update(id: Input::ID) : bool
+	{
+	return __force_update(id);
+	}
+
+function add_tablefilter(id: Input::ID, filter: Input::TableFilter) : bool
+	{
+#	filters[id, filter$name] = filter;
+	return __add_tablefilter(id, filter);
+	}
+
+function remove_tablefilter(id: Input::ID, name: string) : bool
+	{
+#	delete filters[id, name];
+	return __remove_tablefilter(id, name);
+	}
+
+function add_eventfilter(id: Input::ID, filter: Input::EventFilter) : bool
+	{
+#	filters[id, filter$name] = filter;
+	return __add_eventfilter(id, filter);
+	}
+
+function remove_eventfilter(id: Input::ID, name: string) : bool
+	{
+#	delete filters[id, name];
+	return __remove_eventfilter(id, name);
+	}
+
+function read_table(description: Input::StreamDescription, filter: Input::TableFilter) : bool {
+	local ok: bool = T;
+	# since we create and delete it ourselves this should be ok... at least for singlethreaded operation
+	local id: Input::ID = Input::TABLE_READ;
+
+	ok = create_stream(id, description);
+	if ( ok ) {
+		ok = add_tablefilter(id, filter);
+	}
+	if ( ok ) {
+		ok = force_update(id);
+	}
+	if ( ok ) {
+		ok = remove_stream(id);
+	} else {
+		remove_stream(id);
+	}
+
+	return ok;
+}
+
+#function get_filter(id: ID, name: string) : Filter
+#	{
+#	if ( [id, name] in filters )
+#		return filters[id, name];
+#
+#	return no_filter;
+#	}

scripts/base/frameworks/input/readers/ascii.bro

@@ -0,0 +1,19 @@
+##! Interface for the ascii input reader.
+
+module InputAscii;
+
+export {
+	## Separator between fields.
+	## Please note that the separator has to be exactly one character long
+	const separator = "\t" &redef;
+
+	## Separator between set elements.
+	## Please note that the separator has to be exactly one character long
+	const set_separator = "," &redef;
+
+	## String to use for empty fields.
+	const empty_field = "(empty)" &redef;
+
+	## String to use for an unset &optional field.
+	const unset_field = "-" &redef;
+}

scripts/base/init-bare.bro

@@ -2337,3 +2337,6 @@ const snaplen = 8192 &redef;
 # Load the logging framework here because it uses fairly deep integration with 
 # BiFs and script-land defined types.
 @load base/frameworks/logging
+
+@load base/frameworks/input
+

src/Attr.cc

@@ -17,7 +17,7 @@ const char* attr_name(attr_tag t)
 		"&persistent", "&synchronized", "&postprocessor",
 		"&encrypt", "&match", "&disable_print_hook",
 		"&raw_output", "&mergeable", "&priority",
-		"&group", "&log", "&error_handler", "(&tracked)",
+		"&group", "&log", "&error_handler", "&type_column", "(&tracked)",
 	};
 
 	return attr_names[int(t)];
@@ -420,6 +420,26 @@ void Attributes::CheckAttr(Attr* a)
 			Error("&log applied to a type that cannot be logged");
 		break;
 
+	case ATTR_TYPE_COLUMN:
+		{
+		if ( type->Tag() != TYPE_PORT )
+			{
+			Error("type_column tag only applicable to ports");
+			break;
+			}
+
+		BroType* atype = a->AttrExpr()->Type();
+
+		if ( atype->Tag() != TYPE_STRING ) {
+			Error("type column needs to have a string argument");
+			break;
+		}
+
+
+		break;
+		}
+
+
 	default:
 		BadTag("Attributes::CheckAttr", attr_name(a->Tag()));
 	}

src/Attr.h

@@ -35,6 +35,7 @@ typedef enum {
 	ATTR_GROUP,
 	ATTR_LOG,
 	ATTR_ERROR_HANDLER,
+	ATTR_TYPE_COLUMN, // for input framework
 	ATTR_TRACKED,	// hidden attribute, tracked by NotifierRegistry
 #define NUM_ATTRS (int(ATTR_TRACKED) + 1)
 } attr_tag;

src/CMakeLists.txt

@@ -142,6 +142,7 @@ endmacro(GET_BIF_OUTPUT_FILES)
 set(BIF_SRCS
     bro.bif
     logging.bif
+    input.bif
     event.bif
     const.bif
     types.bif
@@ -419,6 +420,12 @@ set(bro_SRCS
     logging/writers/Ascii.cc
     logging/writers/None.cc
 
+    input/Manager.cc
+    input/ReaderBackend.cc
+    input/ReaderFrontend.cc
+    input/readers/Ascii.cc
+
+
     ${dns_SRCS}
     ${openssl_SRCS}
 )

src/Func.cc

@@ -523,11 +523,13 @@ void builtin_error(const char* msg, BroObj* arg)
 
 #include "bro.bif.func_h"
 #include "logging.bif.func_h"
+#include "input.bif.func_h"
 #include "reporter.bif.func_h"
 #include "strings.bif.func_h"
 
 #include "bro.bif.func_def"
 #include "logging.bif.func_def"
+#include "input.bif.func_def"
 #include "reporter.bif.func_def"
 #include "strings.bif.func_def"
 
@@ -542,6 +544,7 @@ void init_builtin_funcs()
 
 #include "bro.bif.func_init"
 #include "logging.bif.func_init"
+#include "input.bif.func_init"
 #include "reporter.bif.func_init"
 #include "strings.bif.func_init"
 

src/NetVar.cc

@@ -257,6 +257,7 @@ StringVal* cmd_line_bpf_filter;
 #include "types.bif.netvar_def"
 #include "event.bif.netvar_def"
 #include "logging.bif.netvar_def"
+#include "input.bif.netvar_def"
 #include "reporter.bif.netvar_def"
 
 void init_event_handlers()
@@ -317,6 +318,7 @@ void init_net_var()
 #include "const.bif.netvar_init"
 #include "types.bif.netvar_init"
 #include "logging.bif.netvar_init"
+#include "input.bif.netvar_init"
 #include "reporter.bif.netvar_init"
 
 	conn_id = internal_type("conn_id")->AsRecordType();

src/NetVar.h

@@ -266,6 +266,7 @@ extern void init_net_var();
 #include "types.bif.netvar_h"
 #include "event.bif.netvar_h"
 #include "logging.bif.netvar_h"
+#include "input.bif.netvar_h"
 #include "reporter.bif.netvar_h"
 
 #endif

src/Val.h

@@ -841,6 +841,9 @@ public:
 			timer = 0;
 		}
 
+	HashKey* ComputeHash(const Val* index) const
+		{ return table_hash->ComputeHash(index, 1); }
+
 protected:
 	friend class Val;
 	friend class StateAccess;
@@ -851,8 +854,6 @@ protected:
 	void CheckExpireAttr(attr_tag at);
 	int ExpandCompoundAndInit(val_list* vl, int k, Val* new_val);
 	int CheckAndAssign(Val* index, Val* new_val, Opcode op = OP_ASSIGN);
-	HashKey* ComputeHash(const Val* index) const
-		{ return table_hash->ComputeHash(index, 1); }
 
 	bool AddProperties(Properties arg_state);
 	bool RemoveProperties(Properties arg_state);

src/input.bif

@@ -0,0 +1,64 @@
+# functions and types for the input framework
+
+module Input;
+
+%%{
+#include "input/Manager.h"
+#include "NetVar.h"
+%%}
+
+type StreamDescription: record;
+type TableFilter: record;
+type EventFilter: record;
+
+function Input::__create_stream%(id: Input::ID, description: Input::StreamDescription%) : bool
+	%{
+	input::ReaderFrontend *the_reader = input_mgr->CreateStream(id->AsEnumVal(), description->AsRecordVal());
+	return new Val( the_reader != 0, TYPE_BOOL );
+	%}
+
+function Input::__remove_stream%(id: Input::ID%) : bool
+	%{
+	bool res = input_mgr->RemoveStream(id->AsEnumVal());
+	return new Val( res, TYPE_BOOL );
+	%}
+
+function Input::__force_update%(id: Input::ID%) : bool
+	%{
+	bool res = input_mgr->ForceUpdate(id->AsEnumVal());
+	return new Val( res, TYPE_BOOL );
+	%}
+
+function Input::__add_tablefilter%(id: Input::ID, filter: Input::TableFilter%) : bool
+	%{
+	bool res = input_mgr->AddTableFilter(id->AsEnumVal(), filter->AsRecordVal());
+	return new Val( res, TYPE_BOOL );
+	%}
+
+function Input::__remove_tablefilter%(id: Input::ID, name: string%) : bool
+	%{
+	bool res = input_mgr->RemoveTableFilter(id->AsEnumVal(), name->AsString()->CheckString());
+	return new Val( res, TYPE_BOOL);
+	%}
+
+function Input::__add_eventfilter%(id: Log::ID, filter: Input::EventFilter%) : bool
+	%{
+	bool res = input_mgr->AddEventFilter(id->AsEnumVal(), filter->AsRecordVal());
+	return new Val( res, TYPE_BOOL );
+	%}
+
+function Input::__remove_eventfilter%(id: Log::ID, name: string%) : bool
+	%{
+	bool res = input_mgr->RemoveEventFilter(id->AsEnumVal(), name->AsString()->CheckString());
+	return new Val( res, TYPE_BOOL);
+	%}
+
+# Options for Ascii Reader
+
+module InputAscii;
+
+const separator: string;
+const set_separator: string;
+const empty_field: string;
+const unset_field: string;
+

src/input/Manager.h

@@ -0,0 +1,90 @@
+// See the file "COPYING" in the main distribution directory for copyright.
+
+#ifndef INPUT_MANAGER_H
+#define INPUT_MANAGER_H
+
+#include "../BroString.h"
+
+#include "../Val.h"
+#include "../EventHandler.h"
+#include "../RemoteSerializer.h"
+
+#include <vector>
+
+namespace input {
+
+class ReaderFrontend;
+
+class Manager {
+public:
+	Manager();
+    
+    	ReaderFrontend* CreateStream(EnumVal* id, RecordVal* description);
+	bool ForceUpdate(const EnumVal* id);
+	bool RemoveStream(const EnumVal* id);	
+
+	bool AddTableFilter(EnumVal *id, RecordVal* filter);
+	bool RemoveTableFilter(EnumVal* id, const string &name);
+
+	bool AddEventFilter(EnumVal *id, RecordVal* filter);
+	bool RemoveEventFilter(EnumVal* id, const string &name);
+	
+protected:
+	
+	// Reports an error for the given reader.
+	void Error(ReaderFrontend* reader, const char* msg);
+
+	// for readers to write to input stream in direct mode (reporting new/deleted values directly)
+	void Put(const ReaderFrontend* reader, int id, const threading::Value* const *vals);
+	void Clear(const ReaderFrontend* reader, int id);
+	bool Delete(const ReaderFrontend* reader, int id, const threading::Value* const *vals);
+
+	// for readers to write to input stream in indirect mode (manager is monitoring new/deleted values)
+	void SendEntry(const ReaderFrontend* reader, int id, const threading::Value* const *vals);
+	void EndCurrentSend(const ReaderFrontend* reader, int id);
+	
+private:
+	struct ReaderInfo;
+
+	void SendEntryTable(const ReaderFrontend* reader, int id, const threading::Value* const *vals);	
+	void PutTable(const ReaderFrontend* reader, int id, const threading::Value* const *vals);	
+	void SendEventFilterEvent(const ReaderFrontend* reader, EnumVal* type, int id, const threading::Value* const *vals);
+
+	bool IsCompatibleType(BroType* t, bool atomic_only=false);
+
+	bool UnrollRecordType(vector<threading::Field*> *fields, const RecordType *rec, const string& nameprepend);
+
+	void SendEvent(EventHandlerPtr ev, const int numvals, ...);	
+	void SendEvent(EventHandlerPtr ev, list<Val*> events);	
+	bool SendEvent(const string& name, const int num_vals, const threading::Value* const *vals);
+
+	HashKey* HashValues(const int num_elements, const threading::Value* const *vals);
+	int GetValueLength(const threading::Value* val);
+	int CopyValue(char *data, const int startpos, const threading::Value* val);
+
+	Val* ValueToVal(const threading::Value* val, BroType* request_type);
+	Val* ValueToIndexVal(int num_fields, const RecordType* type, const threading::Value* const *vals);
+	RecordVal* ValueToRecordVal(const threading::Value* const *vals, RecordType *request_type, int* position);	
+	RecordVal* ListValToRecordVal(ListVal* list, RecordType *request_type, int* position);
+
+	ReaderInfo* FindReader(const ReaderFrontend* reader);
+	ReaderInfo* FindReader(const EnumVal* id);
+
+	vector<ReaderInfo*> readers;
+
+	string Hash(const string &input);	
+
+	class Filter;
+	class TableFilter;
+	class EventFilter;
+
+	enum FilterType { TABLE_FILTER, EVENT_FILTER };
+};
+
+
+}
+
+extern input::Manager* input_mgr;
+
+
+#endif /* INPUT_MANAGER_H */

src/input/ReaderBackend.cc

@@ -0,0 +1,124 @@
+// See the file "COPYING" in the main distribution directory for copyright.
+
+#include "InputReader.h"
+
+using threading::Value;
+using threading::Field;
+
+namespace logging {
+
+InputReader::InputReader(ReaderFrontend *arg_frontend) :MsgThread()
+{
+	buf = 0;
+	buf_len = 1024;
+	disabled = true; // disabled will be set correcty in init.
+
+	frontend = arg_frontend;
+
+	SetName(frontend->Name());
+}
+
+InputReader::~InputReader() 
+{
+	
+}
+
+void InputReader::Error(const char *msg)
+{
+	input_mgr->Error(this, msg);
+}
+
+void InputReader::Error(const string &msg)
+{
+	input_mgr->Error(this, msg.c_str());
+}
+
+void InputReader::Put(int id, const LogVal* const *val) 
+{
+	input_mgr->Put(this, id, val);
+}
+
+void InputReader::Clear(int id) 
+{
+	input_mgr->Clear(this, id);
+}
+
+void InputReader::Delete(int id, const LogVal* const *val) 
+{
+	input_mgr->Delete(this, id, val);
+}
+
+
+bool InputReader::Init(string arg_source) 
+{
+	source = arg_source;
+
+	// disable if DoInit returns error.
+	disabled = !DoInit(arg_source);
+	return !disabled;
+}
+
+bool InputReader::AddFilter(int id, int arg_num_fields,
+					   const LogField* const * arg_fields) 
+{
+	return DoAddFilter(id, arg_num_fields, arg_fields);
+}
+
+bool InputReader::RemoveFilter(int id) 
+{
+	return DoRemoveFilter(id);
+}
+
+void InputReader::Finish() 
+{
+	DoFinish();
+	disabled = true;
+}
+
+bool InputReader::Update() 
+{
+	return DoUpdate();
+}
+
+bool InputReader::SendEvent(const string& name, const int num_vals, const LogVal* const *vals) 
+{
+	return input_mgr->SendEvent(name, num_vals, vals);
+} 
+
+// stolen from logwriter
+const char* InputReader::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);
+	va_end(al);
+
+	if ( (unsigned int) n >= buf_len )
+		{ // Not enough room, grow the buffer.
+		buf_len = n + 32;
+		buf = (char*) realloc(buf, buf_len);
+
+		// Is it portable to restart?
+		va_start(al, format);
+		n = safe_vsnprintf(buf, buf_len, format, al);
+		va_end(al);
+		}
+
+	return buf;
+	}
+
+
+void InputReader::SendEntry(int id, const LogVal* const *vals)
+{
+	input_mgr->SendEntry(this, id, vals);
+}
+
+void InputReader::EndCurrentSend(int id) 
+{
+	input_mgr->EndCurrentSend(this, id);
+}
+
+}

src/input/ReaderBackend.h

@@ -0,0 +1,86 @@
+// See the file "COPYING" in the main distribution directory for copyright.
+// 
+// Same notes about thread safety as in LogWriter.h apply.
+
+
+#ifndef INPUT_READERBACKEND_H
+#define INPUT_READERBACKEND_H
+
+#include "InputMgr.h"
+#include "BroString.h"
+#include "LogMgr.h"
+
+namespace input {
+
+class ReaderBackend : public threading::MsgThread {
+public:
+	ReaderBackend(ReaderFrontend *frontend);
+    	
+	virtual ~ReaderBackend();
+	
+	bool Init(string arg_source);
+
+	bool AddFilter( int id, int arg_num_fields, const LogField* const* fields );
+
+	bool RemoveFilter ( int id );
+    
+	void Finish();
+
+	bool Update();
+	
+protected:
+    // Methods that have to be overwritten by the individual readers
+	virtual bool DoInit(string arg_sources) = 0;
+
+	virtual bool DoAddFilter( int id, int arg_num_fields, const LogField* const* fields ) = 0;
+
+	virtual bool DoRemoveFilter( int id ) = 0;
+
+	virtual void DoFinish() = 0;
+
+	// update file contents to logmgr
+	virtual bool DoUpdate() = 0;
+	
+	// Reports an error to the user.
+	void Error(const string &msg);
+	void Error(const char *msg);
+	
+	// The following methods return the information as passed to Init().
+	const string Source() const	{ return source; }
+
+	// A thread-safe version of fmt(). (stolen from logwriter)
+	const char* Fmt(const char* format, ...);
+
+	bool SendEvent(const string& name, const int num_vals, const LogVal* const *vals);
+
+	// Content-sendinf-functions (simple mode). Including table-specific stuff that simply is not used if we have no table
+	void Put(int id, const LogVal* const *val);
+	void Delete(int id, const LogVal* const *val);
+	void Clear(int id);
+
+	// Table-functions (tracking mode): Only changed lines are propagated.
+	void SendEntry(int id, const LogVal* const *vals);
+	void EndCurrentSend(int id);
+	
+
+private:
+	// Frontend that instantiated us. This object must not be access from
+	// this class, it's running in a different thread!
+	ReaderFrontend* frontend;	
+
+	string source;
+    
+    	// When an error occurs, this method is called to set a flag marking the 
+    	// writer as disabled.
+    
+    	bool disabled;
+    	bool Disabled() { return disabled; }
+
+	// For implementing Fmt().
+	char* buf;
+	unsigned int buf_len;    
+};
+
+}
+
+#endif /* INPUT_READERBACKEND_H */

src/input/ReaderFrontend.cc

@@ -0,0 +1,28 @@
+// See the file "COPYING" in the main distribution directory for copyright.
+
+#ifndef INPUT_READERFRONTEND_H
+#define INPUT_READERFRONTEND_H
+
+#include "Manager.h"
+
+#include "threading/MsgThread.h"
+
+namespace logging {
+
+class ReaderBackend;
+
+class ReaderFrontend {
+
+	ReaderFrontend(bro_int_t type);
+
+	virtual ~ReaderFrontend();
+
+
+protected:
+	friend class Manager;
+};
+
+}
+
+#endif /* INPUT_READERFRONTEND_H */
+

src/input/readers/Ascii.cc

@@ -0,0 +1,457 @@
+// See the file "COPYING" in the main distribution directory for copyright.
+
+#include "InputReaderAscii.h"
+#include "DebugLogger.h"
+#include "NetVar.h"
+
+#include <sstream>
+
+FieldMapping::FieldMapping(const string& arg_name, const TypeTag& arg_type, int arg_position) 
+	: name(arg_name), type(arg_type)
+{
+	position = arg_position;
+	secondary_position = -1;
+}
+
+FieldMapping::FieldMapping(const string& arg_name, const TypeTag& arg_type, const TypeTag& arg_subtype, int arg_position) 
+	: name(arg_name), type(arg_type), subtype(arg_subtype)
+{
+	position = arg_position;
+	secondary_position = -1;
+}
+
+FieldMapping::FieldMapping(const FieldMapping& arg) 
+	: name(arg.name), type(arg.type), subtype(arg.subtype)
+{
+	position = arg.position;
+	secondary_position = arg.secondary_position;
+}
+
+FieldMapping FieldMapping::subType() {
+	return FieldMapping(name, subtype, position);
+}
+
+InputReaderAscii::InputReaderAscii()
+{
+	file = 0;
+
+	//keyMap = new map<string, string>();
+	
+	separator.assign( (const char*) BifConst::InputAscii::separator->Bytes(), BifConst::InputAscii::separator->Len());
+	if ( separator.size() != 1 ) {
+		Error("separator length has to be 1. Separator will be truncated.");
+	}
+
+	set_separator.assign( (const char*) BifConst::InputAscii::set_separator->Bytes(), BifConst::InputAscii::set_separator->Len());
+	if ( set_separator.size() != 1 ) {
+		Error("set_separator length has to be 1. Separator will be truncated.");
+	}
+
+	empty_field.assign( (const char*) BifConst::InputAscii::empty_field->Bytes(), BifConst::InputAscii::empty_field->Len());
+	
+	unset_field.assign( (const char*) BifConst::InputAscii::unset_field->Bytes(), BifConst::InputAscii::unset_field->Len());
+	
+}
+
+InputReaderAscii::~InputReaderAscii()
+{
+	DoFinish();
+
+}
+
+void InputReaderAscii::DoFinish()
+{
+	filters.empty();
+	if ( file != 0 ) {
+		file->close();
+		delete(file);
+		file = 0;
+	}
+}
+
+bool InputReaderAscii::DoInit(string path)
+{
+	fname = path;
+	
+	file = new ifstream(path.c_str());
+	if ( !file->is_open() ) {
+		Error(Fmt("cannot open %s", fname.c_str()));
+		return false;
+	}
+
+	return true;
+}
+
+bool InputReaderAscii::DoAddFilter( int id, int arg_num_fields, const LogField* const* fields ) {
+	if ( HasFilter(id) ) {
+		return false; // no, we don't want to add this a second time
+	}
+
+	Filter f;
+	f.num_fields = arg_num_fields;
+	f.fields = fields;
+
+	filters[id] = f;
+
+	return true;
+}
+
+bool InputReaderAscii::DoRemoveFilter ( int id ) {
+	if (!HasFilter(id) ) {
+		return false;
+	}
+
+	assert ( filters.erase(id) == 1 );
+
+	return true;
+}	
+
+
+bool InputReaderAscii::HasFilter(int id) {
+	map<int, Filter>::iterator it = filters.find(id);	
+	if ( it == filters.end() ) {
+		return false;
+	}
+	return true;
+}
+
+
+bool InputReaderAscii::ReadHeader() {	 
+	// try to read the header line...
+	string line;
+	if ( !GetLine(line) ) {
+		Error("could not read first line");
+		return false;
+	}
+
+	map<string, uint32_t> fields;
+
+	// construcr list of field names.
+	istringstream splitstream(line);
+	int pos=0;
+	while ( splitstream ) {
+		string s;
+		if ( !getline(splitstream, s, separator[0]))
+			break;
+
+		fields[s] = pos;
+		pos++;
+	}
+
+
+	for ( map<int, Filter>::iterator it = filters.begin(); it != filters.end(); it++ ) {
+			
+		for ( unsigned int i = 0; i < (*it).second.num_fields; i++ ) {
+			const LogField* field = (*it).second.fields[i];
+			
+			map<string, uint32_t>::iterator fit = fields.find(field->name);	
+			if ( fit == fields.end() ) {
+				Error(Fmt("Did not find requested field %s in input data file.", field->name.c_str()));
+				return false;
+			}
+	
+
+			FieldMapping f(field->name, field->type, field->subtype, fields[field->name]);
+			if ( field->secondary_name != "" ) {
+				map<string, uint32_t>::iterator fit2 = fields.find(field->secondary_name);					
+				if ( fit2 == fields.end() ) {
+					Error(Fmt("Could not find requested port type field %s in input data file.", field->secondary_name.c_str()));
+					return false;
+				}
+				f.secondary_position = fields[field->secondary_name];
+			}
+			(*it).second.columnMap.push_back(f);
+		}
+
+	}
+	
+	// well, that seems to have worked...
+	return true;
+}
+
+bool InputReaderAscii::GetLine(string& str) {
+	while ( getline(*file, str) ) {
+		if ( str[0] != '#' ) {
+			return true;
+		}
+
+		if ( str.compare(0,8, "#fields\t") == 0 ) {
+			str = str.substr(8);
+			return true;
+		}
+	}
+
+	return false;
+}
+
+TransportProto InputReaderAscii::StringToProto(const string &proto) {
+	if ( proto == "unknown" ) {
+		return TRANSPORT_UNKNOWN;
+	} else if ( proto == "tcp" ) {
+		return TRANSPORT_TCP;
+	} else if ( proto == "udp" ) {
+		return TRANSPORT_UDP;
+	} else if ( proto == "icmp" ) {
+		return TRANSPORT_ICMP;
+	}
+
+	//assert(false);
+	
+	reporter->Error("Tried to parse invalid/unknown protocol: %s", proto.c_str());
+
+	return TRANSPORT_UNKNOWN;
+}
+
+LogVal* InputReaderAscii::EntryToVal(string s, FieldMapping field) {
+
+	LogVal* val = new LogVal(field.type, true);
+
+	if ( s.compare(unset_field) == 0 ) { // field is not set...
+		return new LogVal(field.type, false);
+	}
+
+	switch ( field.type ) {
+	case TYPE_ENUM:
+	case TYPE_STRING:
+		val->val.string_val = new string(s);
+		break;
+
+	case TYPE_BOOL:
+		if ( s == "T" ) {
+			val->val.int_val = 1;
+		} else if ( s == "F" ) {
+			val->val.int_val = 0;
+		} else {
+			Error(Fmt("Invalid value for boolean: %s", s.c_str()));
+			return false;
+		}
+		break;
+
+	case TYPE_INT:
+		val->val.int_val = atoi(s.c_str());
+		break;
+
+	case TYPE_DOUBLE:
+	case TYPE_TIME:
+	case TYPE_INTERVAL:
+		val->val.double_val = atof(s.c_str());
+		break;
+
+	case TYPE_COUNT:
+	case TYPE_COUNTER:
+		val->val.uint_val = atoi(s.c_str());
+		break;
+
+	case TYPE_PORT:
+		val->val.port_val.port = atoi(s.c_str());
+		val->val.port_val.proto = TRANSPORT_UNKNOWN;
+		break;
+
+	case TYPE_SUBNET: {
+		int pos = s.find("/");
+		string width = s.substr(pos+1);
+		val->val.subnet_val.width = atoi(width.c_str());
+		string addr = s.substr(0, pos);
+		s = addr;
+		// NOTE: dotted_to_addr BREAKS THREAD SAFETY! it uses reporter.
+		// Solve this some other time....
+#ifdef BROv6
+		if ( s.find(':') != s.npos ) {
+			uint32* addr = dotted_to_addr6(s.c_str());
+			copy_addr(val->val.subnet_val.net, addr);
+			delete addr;
+		} else {
+			val->val.subnet_val.net[0] = val->val.subnet_val.net[1] = val->val.subnet_val.net[2] = 0;
+		       	val->val.subnet_val.net[3] = dotted_to_addr(s.c_str());
+		}
+#else
+		val->val.subnet_val.net = dotted_to_addr(s.c_str());
+#endif
+		break;
+
+		}
+	case TYPE_ADDR: {
+		// NOTE: dottet_to_addr BREAKS THREAD SAFETY! it uses reporter.
+		// Solve this some other time....
+#ifdef BROv6
+		if ( s.find(':') != s.npos ) {
+			uint32* addr = dotted_to_addr6(s.c_str());
+			copy_addr(val->val.addr_val, addr);
+			delete addr;
+		} else {
+			val->val.addr_val[0] = val->val.addr_val[1] = val->val.addr_val[2] = 0;
+		       	val->val.addr_val[3] = dotted_to_addr(s.c_str());
+		}
+#else
+		uint32 t = dotted_to_addr(s.c_str());
+		copy_addr(&t, val->val.addr_val);
+#endif
+		break;
+		}
+
+	case TYPE_TABLE:
+	case TYPE_VECTOR:
+		// First - common initialization
+		// Then - initialization for table.
+		// Then - initialization for vector.
+		// Then - common stuff
+		{
+		// how many entries do we have...
+		unsigned int length = 1;
+		for ( unsigned int i = 0; i < s.size(); i++ )
+			if ( s[i] == ',') length++;
+
+		unsigned int pos = 0;
+		
+		if ( s.compare(empty_field) == 0 ) 
+			length = 0;
+
+		LogVal** lvals = new LogVal* [length];
+
+		if ( field.type == TYPE_TABLE ) {
+			val->val.set_val.vals = lvals;
+			val->val.set_val.size = length;
+		} else if ( field.type == TYPE_VECTOR ) {
+			val->val.vector_val.vals = lvals;
+			val->val.vector_val.size = length;
+		} else {
+			assert(false);
+		}
+
+		if ( length == 0 )
+			break; //empty
+
+		istringstream splitstream(s);
+		while ( splitstream ) {
+			string element;
+
+			if ( !getline(splitstream, element, set_separator[0]) )
+				break;
+
+			if ( pos >= length ) {
+				Error(Fmt("Internal error while parsing set. pos %d >= length %d. Element: %s", pos, length, element.c_str()));
+				break;
+			}
+
+			LogVal* newval = EntryToVal(element, field.subType());
+			if ( newval == 0 ) {
+				Error("Error while reading set");
+				return 0;
+			}
+			lvals[pos] = newval;
+
+			pos++;
+	
+		}
+
+
+		if ( pos != length ) {
+			Error("Internal error while parsing set: did not find all elements");
+			return 0;
+		}
+
+		break;
+		}
+
+
+	default:
+		Error(Fmt("unsupported field format %d for %s", field.type,
+		field.name.c_str()));
+		return 0;
+	}	
+
+	return val;
+
+}
+
+// read the entire file and send appropriate thingies back to InputMgr
+bool InputReaderAscii::DoUpdate() {
+	 
+	// dirty, fix me. (well, apparently after trying seeking, etc - this is not that bad)
+	if ( file && file->is_open() ) {
+		file->close();
+	}
+	file = new ifstream(fname.c_str());
+	if ( !file->is_open() ) {
+		Error(Fmt("cannot open %s", fname.c_str()));
+		return false;
+	}
+	// 
+	
+	// file->seekg(0, ios::beg); // do not forget clear.
+
+
+	if ( ReadHeader() == false ) {
+		return false;
+	}
+
+	string line;
+	while ( GetLine(line ) ) {
+		// split on tabs
+		istringstream splitstream(line);
+
+		map<int, string> stringfields;
+		int pos = 0;
+		while ( splitstream ) {
+			string s;
+			if ( !getline(splitstream, s, separator[0]) )
+				break;
+
+			stringfields[pos] = s;
+			pos++;
+		}
+
+		pos--; // for easy comparisons of max element.
+
+		for ( map<int, Filter>::iterator it = filters.begin(); it != filters.end(); it++ ) {
+		
+			LogVal** fields = new LogVal*[(*it).second.num_fields];
+
+			int fpos = 0;
+			for ( vector<FieldMapping>::iterator fit = (*it).second.columnMap.begin();
+				fit != (*it).second.columnMap.end();
+				fit++ ){
+
+				if ( (*fit).position > pos || (*fit).secondary_position > pos ) {
+					Error(Fmt("Not enough fields in line %s. Found %d fields, want positions %d and %d", line.c_str(), pos,  (*fit).position, (*fit).secondary_position));
+					return false;
+				}
+
+				LogVal* val = EntryToVal(stringfields[(*fit).position], *fit);
+				if ( val == 0 ) {
+					return false;
+				}
+				
+				if ( (*fit).secondary_position != -1 ) {
+					// we have a port definition :)
+					assert(val->type == TYPE_PORT ); 
+					//	Error(Fmt("Got type %d != PORT with secondary position!", val->type));
+
+					val->val.port_val.proto = StringToProto(stringfields[(*fit).secondary_position]);
+				}
+
+				fields[fpos] = val;
+
+				fpos++;
+			}
+
+			assert ( (unsigned int) fpos == (*it).second.num_fields );
+
+			SendEntry((*it).first, fields);
+
+			for ( unsigned int i = 0; i < (*it).second.num_fields; i++ ) {
+				delete fields[i];
+			}
+			delete [] fields;
+		}
+
+	}
+
+	//file->clear(); // remove end of file evil bits
+	//file->seekg(0, ios::beg); // and seek to start.
+
+	for ( map<int, Filter>::iterator it = filters.begin(); it != filters.end(); it++ ) {
+		EndCurrentSend((*it).first);
+	}
+	return true;
+}

src/input/readers/Ascii.h

@@ -0,0 +1,88 @@
+// See the file "COPYING" in the main distribution directory for copyright.
+
+#ifndef INPUTREADERASCII_H
+#define INPUTREADERASCII_H
+
+#include "InputReader.h"
+#include <fstream>
+#include <iostream>
+#include <vector>
+
+// Description for input field mapping
+struct FieldMapping {
+	string name;
+	TypeTag type;
+	// internal type for sets and vectors
+	TypeTag subtype;
+	int position;
+	// for ports: pos of the second field
+	int secondary_position;
+
+	FieldMapping(const string& arg_name, const TypeTag& arg_type, int arg_position); 
+	FieldMapping(const string& arg_name, const TypeTag& arg_type, const TypeTag& arg_subtype, int arg_position); 
+	FieldMapping(const FieldMapping& arg);
+	FieldMapping() { position = -1; secondary_position = -1; }
+
+	FieldMapping subType();
+	//bool IsEmpty() { return position == -1; }
+};
+
+
+class InputReaderAscii : public InputReader {
+public:
+    InputReaderAscii();
+    ~InputReaderAscii();
+    
+    static InputReader* Instantiate() { return new InputReaderAscii; }
+    
+protected:
+	
+	virtual bool DoInit(string path);
+
+	virtual bool DoAddFilter( int id, int arg_num_fields, const LogField* const* fields );
+
+	virtual bool DoRemoveFilter ( int id );	
+
+	virtual void DoFinish();
+
+	virtual bool DoUpdate();
+    
+private:
+
+	struct Filter {
+		unsigned int num_fields;
+
+		const LogField* const * fields; // raw mapping		
+
+		// map columns in the file to columns to send back to the manager
+		vector<FieldMapping> columnMap;		
+
+	};
+
+	bool HasFilter(int id);
+
+	TransportProto StringToProto(const string &proto);		
+
+	bool ReadHeader();
+	LogVal* EntryToVal(string s, FieldMapping type);
+
+	bool GetLine(string& str);
+	
+	ifstream* file;
+	string fname;
+
+	map<int, Filter> filters;
+
+	// Options set from the script-level.
+	string separator;
+
+	string set_separator;
+
+	string empty_field;
+
+	string unset_field;
+
+};
+
+
+#endif /* INPUTREADERASCII_H */

src/logging/Manager.cc

@@ -850,7 +850,8 @@ threading::Value* Manager::ValToLogVal(Val* val, BroType* ty)
 		break;
 
 	case TYPE_PORT:
-		lval->val.uint_val = val->AsPortVal()->Port();
+		lval->val.port_val.port = val->AsPortVal()->Port();
+		lval->val.port_val.proto = val->AsPortVal()->PortType();
 		break;
 
 	case TYPE_SUBNET:

src/logging/writers/Ascii.cc

@@ -169,10 +169,13 @@ bool Ascii::DoWriteOne(ODesc* desc, Value* val, const Field* field)
 
 	case TYPE_COUNT:
 	case TYPE_COUNTER:
-	case TYPE_PORT:
 		desc->Add(val->val.uint_val);
 		break;
 
+	case TYPE_PORT:
+		desc->Add(val->val.port_val.port);
+		break;
+
 	case TYPE_SUBNET:
 		desc->Add(dotted_addr(val->val.subnet_val.net));
 		desc->Add("/");

src/main.cc

@@ -52,6 +52,8 @@ extern "C" void OPENSSL_add_all_algorithms_conf(void);
 #include "logging/Manager.h"
 #include "logging/writers/Ascii.h"
 
+#include "input/Manager.h"
+
 #include "binpac_bro.h"
 
 Brofiler brofiler;
@@ -79,6 +81,7 @@ DNS_Mgr* dns_mgr;
 TimerMgr* timer_mgr;
 logging::Manager* log_mgr = 0;
 threading::Manager* thread_mgr = 0;
+input::Manager* input_mgr = 0;
 Stmt* stmts;
 EventHandlerPtr net_done = 0;
 RuleMatcher* rule_matcher = 0;
@@ -743,6 +746,7 @@ int main(int argc, char** argv)
 	remote_serializer = new RemoteSerializer();
 	event_registry = new EventRegistry();
 	log_mgr = new logging::Manager();
+    	input_mgr = new input::Manager();
 
 	if ( events_file )
 		event_player = new EventPlayer(events_file);

src/parse.y

@@ -2,7 +2,7 @@
 // See the file "COPYING" in the main distribution directory for copyright.
 %}
 
-%expect 87
+%expect 90
 
 %token TOK_ADD TOK_ADD_TO TOK_ADDR TOK_ANY
 %token TOK_ATENDIF TOK_ATELSE TOK_ATIF TOK_ATIFDEF TOK_ATIFNDEF
@@ -24,6 +24,7 @@
 %token TOK_ATTR_PERSISTENT TOK_ATTR_SYNCHRONIZED
 %token TOK_ATTR_DISABLE_PRINT_HOOK TOK_ATTR_RAW_OUTPUT TOK_ATTR_MERGEABLE
 %token TOK_ATTR_PRIORITY TOK_ATTR_GROUP TOK_ATTR_LOG TOK_ATTR_ERROR_HANDLER
+%token TOK_ATTR_TYPE_COLUMN
 
 %token TOK_DEBUG
 
@@ -1312,6 +1313,8 @@ attr:
 			{ $$ = new Attr(ATTR_PRIORITY, $3); }
 	|	TOK_ATTR_GROUP '=' expr
 			{ $$ = new Attr(ATTR_GROUP, $3); }
+	|	TOK_ATTR_TYPE_COLUMN '=' expr
+			{ $$ = new Attr(ATTR_TYPE_COLUMN, $3); }
 	|	TOK_ATTR_LOG
 			{ $$ = new Attr(ATTR_LOG); }
 	|	TOK_ATTR_ERROR_HANDLER

src/scan.l

@@ -315,6 +315,7 @@ when	return TOK_WHEN;
 &optional	return TOK_ATTR_OPTIONAL;
 &persistent	return TOK_ATTR_PERSISTENT;
 &priority	return TOK_ATTR_PRIORITY;
+&type_column	return TOK_ATTR_TYPE_COLUMN;
 &read_expire	return TOK_ATTR_EXPIRE_READ;
 &redef		return TOK_ATTR_REDEF;
 &rotate_interval	return TOK_ATTR_ROTATE_INTERVAL;

src/types.bif

@@ -168,4 +168,21 @@ enum ID %{
 	Unknown,
 %}
 
+module Input;
+
+enum Reader %{
+	READER_DEFAULT,
+	READER_ASCII,
+%}
+
+enum Event %{
+	EVENT_NEW,
+	EVENT_CHANGED,
+	EVENT_REMOVED,
+%}
+
+enum ID %{
+	Unknown,
+%}
+
 module GLOBAL;

testing/btest/Baseline/coverage.bare-load-baseline/canonified_loaded_scripts.log

@@ -19,4 +19,8 @@ scripts/base/init-bare.bro
       scripts/base/frameworks/logging/./postprocessors/./scp.bro
       scripts/base/frameworks/logging/./postprocessors/./sftp.bro
     scripts/base/frameworks/logging/./writers/ascii.bro
+  scripts/base/frameworks/input/__load__.bro
+    scripts/base/frameworks/input/./main.bro
+      build/src/base/input.bif.bro
+    scripts/base/frameworks/input/./readers/ascii.bro
 scripts/policy/misc/loaded-scripts.bro

testing/btest/Baseline/coverage.default-load-baseline/canonified_loaded_scripts.log

@@ -19,6 +19,10 @@ scripts/base/init-bare.bro
       scripts/base/frameworks/logging/./postprocessors/./scp.bro
       scripts/base/frameworks/logging/./postprocessors/./sftp.bro
     scripts/base/frameworks/logging/./writers/ascii.bro
+  scripts/base/frameworks/input/__load__.bro
+    scripts/base/frameworks/input/./main.bro
+      build/src/base/input.bif.bro
+    scripts/base/frameworks/input/./readers/ascii.bro
 scripts/base/init-default.bro
   scripts/base/utils/site.bro
     scripts/base/utils/./patterns.bro

testing/btest/Baseline/scripts.base.frameworks.input.basic/out

@@ -0,0 +1,14 @@
+{
+[-42] = [b=T, e=SSH::LOG, c=21, p=123/unknown, sn=10.0.0.0/24, a=1.2.3.4, d=3.14, t=1315801931.273616, iv=100.0, s=hurz, sc={
+2,
+4,
+1,
+3
+}, ss={
+CC,
+AA,
+BB
+}, se={
+
+}, vc=[10, 20, 30], ve=[]]
+}

testing/btest/Baseline/scripts.base.frameworks.input.event/out

@@ -0,0 +1,21 @@
+Input::EVENT_NEW
+1
+T
+Input::EVENT_NEW
+2
+T
+Input::EVENT_NEW
+3
+F
+Input::EVENT_NEW
+4
+F
+Input::EVENT_NEW
+5
+F
+Input::EVENT_NEW
+6
+F
+Input::EVENT_NEW
+7
+T

testing/btest/Baseline/scripts.base.frameworks.input.onecolumn-norecord/out

@@ -0,0 +1,3 @@
+{
+[-42] = T
+}

testing/btest/Baseline/scripts.base.frameworks.input.onecolumn-record/out

@@ -0,0 +1,3 @@
+{
+[-42] = [b=T]
+}

testing/btest/Baseline/scripts.base.frameworks.input.port/out

@@ -0,0 +1,3 @@
+[p=80/tcp]
+[p=52/udp]
+[p=30/unknown]

testing/btest/Baseline/scripts.base.frameworks.input.predicate/out

@@ -0,0 +1,7 @@
+VALID
+VALID
+VALID
+VALID
+VALID
+VALID
+VALID

testing/btest/Baseline/scripts.base.frameworks.input.tableevent/out

@@ -0,0 +1,21 @@
+Input::EVENT_NEW
+1
+T
+Input::EVENT_NEW
+2
+T
+Input::EVENT_NEW
+3
+F
+Input::EVENT_NEW
+4
+F
+Input::EVENT_NEW
+5
+F
+Input::EVENT_NEW
+6
+F
+Input::EVENT_NEW
+7
+T

testing/btest/Baseline/scripts.base.frameworks.input.twofilters/out

@@ -0,0 +1,15 @@
+VALID
+VALID
+VALID
+VALID
+VALID
+VALID
+VALID
+MARK
+VALID
+VALID
+VALID
+VALID
+VALID
+VALID
+VALID

testing/btest/btest.cfg

@@ -3,7 +3,7 @@ TestDirs    = doc bifs language core scripts istate coverage
 TmpDir      = %(testbase)s/.tmp
 BaselineDir = %(testbase)s/Baseline
 IgnoreDirs  = .svn CVS .tmp
-IgnoreFiles = *.tmp *.swp #* *.trace
+IgnoreFiles = *.tmp *.swp #* *.trace .DS_Store
 
 [environment]
 BROPATH=`bash -c %(testbase)s/../../build/bro-path-dev`

testing/btest/scripts/base/frameworks/input/basic.bro

@@ -0,0 +1,54 @@
+#
+# @TEST-EXEC: bro %INPUT >out
+# @TEST-EXEC: btest-diff out
+
+@TEST-START-FILE input.log
+#separator \x09
+#path	ssh
+#fields	b	i	e	c	p	sn	a	d	t	iv	s	sc	ss	se	vc	ve	f
+#types	bool	int	enum	count	port	subnet	addr	double	time	interval	string	table	table	table	vector	vector	func
+T	-42	SSH::LOG	21	123	10.0.0.0/24	1.2.3.4	3.14	1315801931.273616	100.000000	hurz	2,4,1,3	CC,AA,BB	EMPTY	10,20,30	EMPTY	SSH::foo\x0a{ \x0aif (0 < SSH::i) \x0a\x09return (Foo);\x0aelse\x0a\x09return (Bar);\x0a\x0a}
+@TEST-END-FILE
+
+redef InputAscii::empty_field = "EMPTY";
+
+module A;
+
+export {
+	redef enum Input::ID += { INPUT };
+}
+
+type Idx: record {
+	i: int;
+};
+
+type Val: record {
+	b: bool;
+	e: Log::ID;
+	c: count;
+	p: port;
+	sn: subnet;
+	a: addr;
+	d: double;
+	t: time;
+	iv: interval;
+	s: string;
+	sc: set[count];
+	ss: set[string];
+	se: set[string];
+	vc: vector of int;
+	ve: vector of int;
+};
+
+global servers: table[int] of Val = table();
+
+event bro_init()
+{
+	# first read in the old stuff into the table...
+	Input::create_stream(A::INPUT, [$source="input.log"]);
+	Input::add_tablefilter(A::INPUT, [$name="ssh", $idx=Idx, $val=Val, $destination=servers]);
+	Input::force_update(A::INPUT);
+	print servers;
+	Input::remove_tablefilter(A::INPUT, "ssh");
+	Input::remove_stream(A::INPUT);
+}

testing/btest/scripts/base/frameworks/input/event.bro

@@ -0,0 +1,42 @@
+#
+# @TEST-EXEC: bro %INPUT >out
+# @TEST-EXEC: btest-diff out
+
+@TEST-START-FILE input.log
+#separator \x09
+#path	ssh
+#fields	i	b	
+#types	int	bool
+1	T
+2	T
+3	F
+4	F
+5	F
+6	F
+7	T
+@TEST-END-FILE
+
+
+module A;
+
+export {
+	redef enum Input::ID += { INPUT };
+}
+
+type Val: record {
+	i: int;
+	b: bool;
+};
+
+event line(tpe: Input::Event, i: int, b: bool) {
+	print tpe;
+	print i;
+	print b;
+}
+
+event bro_init()
+{
+	Input::create_stream(A::INPUT, [$source="input.log"]);
+	Input::add_eventfilter(A::INPUT, [$name="input", $fields=Val, $ev=line]);
+	Input::force_update(A::INPUT);
+}

testing/btest/scripts/base/frameworks/input/onecolumn-norecord.bro

@@ -0,0 +1,38 @@
+#
+# @TEST-EXEC: bro %INPUT >out
+# @TEST-EXEC: btest-diff out
+
+@TEST-START-FILE input.log
+#separator \x09
+#path	ssh
+#fields	b	i	
+#types	bool	int
+T	-42	
+@TEST-END-FILE
+
+redef InputAscii::empty_field = "EMPTY";
+
+module A;
+
+export {
+	redef enum Input::ID += { INPUT };
+}
+
+type Idx: record {
+	i: int;
+};
+
+type Val: record {
+	b: bool;
+};
+
+global servers: table[int] of Val = table();
+
+event bro_init()
+{
+	# first read in the old stuff into the table...
+	Input::create_stream(A::INPUT, [$source="input.log"]);
+	Input::add_tablefilter(A::INPUT, [$name="input", $idx=Idx, $val=Val, $destination=servers, $want_record=F]);
+	Input::force_update(A::INPUT);
+	print servers;
+}

testing/btest/scripts/base/frameworks/input/onecolumn-record.bro

@@ -0,0 +1,38 @@
+#
+# @TEST-EXEC: bro %INPUT >out
+# @TEST-EXEC: btest-diff out
+
+@TEST-START-FILE input.log
+#separator \x09
+#path	ssh
+#fields	b	i	
+#types	bool	int
+T	-42	
+@TEST-END-FILE
+
+redef InputAscii::empty_field = "EMPTY";
+
+module A;
+
+export {
+	redef enum Input::ID += { INPUT };
+}
+
+type Idx: record {
+	i: int;
+};
+
+type Val: record {
+	b: bool;
+};
+
+global servers: table[int] of Val = table();
+
+event bro_init()
+{
+	# first read in the old stuff into the table...
+	Input::create_stream(A::INPUT, [$source="input.log"]);
+	Input::add_tablefilter(A::INPUT, [$name="input", $idx=Idx, $val=Val, $destination=servers]);
+	Input::force_update(A::INPUT);
+	print servers;
+}

testing/btest/scripts/base/frameworks/input/port.bro

@@ -0,0 +1,41 @@
+#
+# @TEST-EXEC: bro %INPUT >out
+# @TEST-EXEC: btest-diff out
+
+@TEST-START-FILE input.log
+#fields	i	p	t
+1.2.3.4	80	tcp
+1.2.3.5	52	udp
+1.2.3.6	30	unknown
+@TEST-END-FILE
+
+redef InputAscii::empty_field = "EMPTY";
+
+module A;
+
+export {
+	redef enum Input::ID += { INPUT };
+}
+
+type Idx: record {
+	i: addr;
+};
+
+type Val: record {
+	p: port &type_column="t";
+};
+
+global servers: table[addr] of Val = table();
+
+event bro_init()
+{
+	# first read in the old stuff into the table...
+	Input::create_stream(A::INPUT, [$source="input.log"]);
+	Input::add_tablefilter(A::INPUT, [$name="input", $idx=Idx, $val=Val, $destination=servers]);
+	Input::force_update(A::INPUT);
+	print servers[1.2.3.4];
+	print servers[1.2.3.5];
+	print servers[1.2.3.6];
+	Input::remove_tablefilter(A::INPUT, "input");
+	Input::remove_stream(A::INPUT);
+}

testing/btest/scripts/base/frameworks/input/predicate.bro

@@ -0,0 +1,66 @@
+#
+# @TEST-EXEC: bro %INPUT >out
+# @TEST-EXEC: btest-diff out
+
+@TEST-START-FILE input.log
+#separator \x09
+#path	ssh
+#fields	i	b	
+#types	int	bool
+1	T
+2	T
+3	F
+4	F
+5	F
+6	F
+7	T
+@TEST-END-FILE
+
+redef InputAscii::empty_field = "EMPTY";
+
+module A;
+
+export {
+	redef enum Input::ID += { INPUT };
+}
+
+type Idx: record {
+	i: int;
+};
+
+type Val: record {
+	b: bool;
+};
+
+global servers: table[int] of Val = table();
+
+event bro_init()
+{
+	# first read in the old stuff into the table...
+	Input::create_stream(A::INPUT, [$source="input.log"]);
+	Input::add_tablefilter(A::INPUT, [$name="input", $idx=Idx, $val=Val, $destination=servers, $want_record=F,
+				$pred(typ: Input::Event, left: Idx, right: bool) = { return right; }
+				]);
+	Input::force_update(A::INPUT);
+	if ( 1 in servers ) {
+		print "VALID";
+	}
+	if ( 2 in servers ) {
+		print "VALID";
+	}
+	if ( !(3 in servers) ) {
+		print "VALID";
+	}
+	if ( !(4 in servers) ) {
+		print "VALID";
+	}
+	if ( !(5 in servers) ) {
+		print "VALID";
+	}
+	if ( !(6 in servers) ) {
+		print "VALID";
+	}
+	if ( 7 in servers ) {
+		print "VALID";
+	}
+}

testing/btest/scripts/base/frameworks/input/tableevent.bro

@@ -0,0 +1,48 @@
+#
+# @TEST-EXEC: bro %INPUT >out
+# @TEST-EXEC: btest-diff out
+
+@TEST-START-FILE input.log
+#separator \x09
+#path	ssh
+#fields	i	b	
+#types	int	bool
+1	T
+2	T
+3	F
+4	F
+5	F
+6	F
+7	T
+@TEST-END-FILE
+
+redef InputAscii::empty_field = "EMPTY";
+
+module A;
+
+export {
+	redef enum Log::ID += { LOG };
+}
+
+type Idx: record {
+	i: int;
+};
+
+type Val: record {
+	b: bool;
+};
+
+global destination: table[int] of Val = table();
+
+event line(tpe: Input::Event, left: Idx, right: bool) {
+	print tpe;
+	print left;
+	print right;
+}
+
+event bro_init()
+{
+	Input::create_stream(A::LOG, [$source="input.log"]);
+	Input::add_tablefilter(A::LOG, [$name="input", $idx=Idx, $val=Val, $destination=destination, $want_record=F,$ev=line]);
+	Input::force_update(A::LOG);
+}

testing/btest/scripts/base/frameworks/input/twofilters.bro

@@ -0,0 +1,95 @@
+#
+# @TEST-EXEC: bro %INPUT >out
+# @TEST-EXEC: btest-diff out
+
+@TEST-START-FILE input.log
+#separator \x09
+#path	ssh
+#fields	i	b	
+#types	int	bool
+1	T
+2	T
+3	F
+4	F
+5	F
+6	F
+7	T
+@TEST-END-FILE
+
+redef InputAscii::empty_field = "EMPTY";
+
+module A;
+
+export {
+	redef enum Input::ID += { INPUT };
+}
+
+type Idx: record {
+	i: int;
+};
+
+type Val: record {
+	b: bool;
+};
+
+global destination1: table[int] of Val = table();
+global destination2: table[int] of Val = table();
+
+event bro_init()
+{
+	# first read in the old stuff into the table...
+	Input::create_stream(A::INPUT, [$source="input.log"]);
+	Input::add_tablefilter(A::INPUT, [$name="input", $idx=Idx, $val=Val, $destination=destination1, $want_record=F,
+				$pred(typ: Input::Event, left: Idx, right: bool) = { return right; }
+				]);
+	Input::add_tablefilter(A::INPUT, [$name="input2",$idx=Idx, $val=Val, $destination=destination2]);
+	
+	Input::force_update(A::INPUT);
+	if ( 1 in destination1 ) {
+		print "VALID";
+	}
+	if ( 2 in destination1 ) {
+		print "VALID";
+	}
+	if ( !(3 in destination1) ) {
+		print "VALID";
+	}
+	if ( !(4 in destination1) ) {
+		print "VALID";
+	}
+	if ( !(5 in destination1) ) {
+		print "VALID";
+	}
+	if ( !(6 in destination1) ) {
+		print "VALID";
+	}
+	if ( 7 in destination1 ) {
+		print "VALID";
+	}
+
+	print "MARK";
+
+	if ( 2 in destination2 ) {
+		print "VALID";
+	}
+	if ( 2 in destination2 ) {
+		print "VALID";
+	}
+	if ( 3 in destination2 ) {
+		print "VALID";
+	}
+	if ( 4 in destination2 ) {
+		print "VALID";
+	}
+	if ( 5 in destination2 ) {
+		print "VALID";
+	}
+	if ( 6 in destination2 ) {
+		print "VALID";
+	}
+	if ( 7 in destination2 ) {
+		print "VALID";
+	}
+
+
+}