Output database support (for MySQL) now complete

README

@@ -158,7 +158,8 @@ preprocessor ai: \
 	correlation_rules_dir "/your/snort/dir/etc/corr_rules" \
 	correlated_alerts_dir "/your/snort/dir/log/correlated_alerts" \
 	correlation_threshold_coefficient 0.5 \
-	database ( type="mysql", name="snort", user="snortusr", password="snortpass", host="dbhost" ) \
+	database ( type="dbtype", name="snort", user="snortusr", password="snortpass", host="dbhost" ) \
+	output_database ( type="dbtype", name="snort", user="snortusr", password="snortpass", host="dbhost" ) \
 	database_parsing_interval 30 \
 	cluster_max_alert_interval 14400 \
 	clusterfile "/your/snort/dir/log/clustered_alerts" \
@@ -173,36 +174,33 @@ preprocessor ai: \
 
 The options are the following:
 
-- hashtable_cleanup_interval: The interval that should occur from the cleanup of
-the  hashtable  of  TCP  streams and the next one (default if not specified: 300
- seconds)
-
-- tcp_stream_expire_interval:  The  interval that should occur for marking a TCP
-stream as "expired", if no more packets are received inside of that and it's not
-"marked"    as    suspicious   (default   if   not   specified:   300   seconds)
-
 - alertfile:  The file where Snort saves its alerts, if they are saved to a file
 and   not  to  a  database  (default  if  not  specified:  /var/log/snort/alert)
 
+
 - alert_history_file:  The   file keeping track of the history, in binary format,
 of   all  the  alerts  received  by  the  IDS, so that the module can build some
 statistical        correlation        inferences       over       the       past
 
+
 - alert_serialization_interval:   The   interval   that   should  occur  from  a
 serialization  of  a  buffer  of  alerts  on  the  history file and the next one
 (default if not specified: 1 hour, as it is a quite expensive operation in terms
 of       resources      if      the     system     received     many     alerts)
 
+
 - alert_bufsize:  Size of the buffer containing the alerts to be sent, in group,
 to   the  serializer  thread.  The  buffer is sent when full and made empty even
 when   the  alert_serialization_interval  parameter  is  not  expired  yet,  for
 avoiding   overflows,  other  memory  problems  or  deadlocks  (default value if
 not                                 specified:                               30)
 
+
 - alert_clustering_interval:  The interval that should occur from the clustering
 of  the  alerts  in the log according to the provided clustering hierarchies and
 the     next     one     (default     if    not    specified:    300    seconds)
 
+
 - bayesian_correlation_interval: Interval, in seconds, that should occur between
 two  alerts  in  the  history  for  considering  them as, more or less strongly,
 correlated (default: 1200 seconds). NOTE: A value of 0 will disable the bayesian
@@ -210,23 +208,28 @@ correlation.  This  setting  is  strongly suggested when your alert log is still
 "learning",  i.e.  when you don't have enough alerts yet. After this period, you
 can      set      the      correlation      interval      to      any     value.
 
+
 - bayesian_correlation_cache_validity:  interval, in seconds, for which an entry
 in  the  bayesian  correlation  hash  table  (i.e.  a  pair  of  alerts with the
  associated   historical   bayesian   correlation)   is   considered   as  valid
 before         being         updated        (default:        600        seconds)
 
+
 - correlation_graph_interval:  The  interval that should occur from the building
 of  the correlation graph between the clustered alerts and the next one (default
  if              not             specified:             300             seconds)
 
+
 - correlation_rules_dir: Directory where the correlation rules are saved, as XML
 files      (default      if      not      specified:      /etc/snort/corr_rules)
 
+
 - correlated_alerts_dir:  Directory  where  the  information  between correlated
 alerts  will  be  saved,  as  .dot  files ready to be rendered as graphs and, if
 libgraphviz  support  is  enabled, as .png and .ps files as well (default if not
  specified: /var/log/snort/clustered_alerts)
 
+
 - correlation_threshold_coefficient: The threshold the software uses for stating
 two   alerts   are   correlated   is   avg(correlation   coefficient)   +   k  *
 std_deviation(correlation_coefficient). The value of k is specified through this
@@ -239,26 +242,17 @@ where  no correlation exists). When the value of k raises also the threshold for
 two alerts for being considered as correlated raises. A high value of k may just
 lead           to          an          empty          correlation          graph
 
-- database:  If Snort saves its alerts to a database and the module was compiled
- with   database   support   (e.g.   --with-mysql)  this  option  specifies  the
- information   for   accessing   that   database.   The   fields   in  side  are
-	-- type: DBMS to be used (so far MySQL and PostgreSQL are supported)
-	-- name: Database name
-	-- user: Username for accessing the database
-	-- password: Password for accessing the database
-	-- host: Host holding the database
-
-- database_parsing_interval:  The  interval  that should occur between a read of
-the alerts from database and the next one (default if not specified: 30 seconds)
 
 - clusterfile:  File  where  the  clustered  alerts  will be saved by the module
 (default       if      not      specified:      /var/log/snort/clustered_alerts)
 
+
 - cluster_max_alert_interval:   Maximum  time  interval,  in  seconds,  occurred
 between  two  alerts  for considering them as part of the same cluster (default:
 14400   seconds,  i.e.  4  hours).  Specify  0  for  this  option if you want to
 cluster   alerts   regardlessly   of   how   much  time  occurred  between  them
 
+
 - cluster:  Clustering  hierarchy  or  list  of  hierarchies  to  be applied for
 grouping     similar     alerts.     This     option     needs    to    specify:
 	-- class: Class of the cluster node. It may be src_addr, dst_addr, src_port
@@ -269,6 +263,41 @@ grouping     similar     alerts.     This     option     needs    to    specify:
 			range (specified as xxx-xxx)
 
 
+- database:  If Snort saves its alerts to a database and the module was compiled
+ with   database   support   (e.g.   --with-mysql)  this  option  specifies  the
+ information   for   accessing   that   database.   The   fields   in  side  are
+	-- type: DBMS to be used (so far MySQL and PostgreSQL are supported)
+	-- name: Database name
+	-- user: Username for accessing the database
+	-- password: Password for accessing the database
+	-- host: Host holding the database
+
+
+- database_parsing_interval:  The  interval  that should occur between a read of
+the alerts from database and the next one (default if not specified: 30 seconds)
+
+
+- hashtable_cleanup_interval: The interval that should occur from the cleanup of
+the  hashtable  of  TCP  streams and the next one (default if not specified: 300
+ seconds)
+
+
+- output_database:  Specify this option if you want to save the outputs from the
+module  (correlated  alerts,  clustered  alerts,  alerts  information  and their
+associated    packets   streams,   and  so  on)  to  a  relational  database  as
+well  (by  default  the module only saves the alerts on static plain files). The
+options    here   are   the   same   specified   for   the   'database'   option.
+The  structure  of this database can be seen in the files schemas/*.sql (replace
+to  * the name of your DBMS). If you want to initialize the tables needed by the
+module,   just   give   the   right  file  to  your  database,  e.g.  for  MySQL
+$ mysql -uusername -ppassword dbname < schemas/mysql.sql
+
+
+- tcp_stream_expire_interval:  The  interval that should occur for marking a TCP
+stream as "expired", if no more packets are received inside of that and it's not
+"marked"    as    suspicious   (default   if   not   specified:   300   seconds)
+
+
 ====================
 5. Correlation rules
 ====================

TODO

@@ -2,7 +2,9 @@
 AVERAGE/HIGH PRIORITY:
 ======================
 
-- Save clusters and correlations to db
+- Full PostgreSQL support for output db
+- Redefine function names
+- Errno
 - Web interface
 - Code profiling
 - Comment all the code!!!
@@ -31,4 +33,5 @@ DONE:
 + Bayesian learning among alerts in alert log
 + Split bayesian correlation out of correlation.c
 + Clustering alerts with time constraints
++ Save clusters and correlations to db
 

cluster.c

@@ -23,6 +23,7 @@
 #include	<unistd.h>
 #include	<math.h>
 #include	<limits.h>
+#include	<errno.h>
 #include 	<pthread.h>
 
 /** \defgroup cluster Manage the clustering of alarms
@@ -262,6 +263,8 @@ PRIVATE int
 _AI_merge_alerts ( AI_snort_alert **log )
 {
 	AI_snort_alert *tmp, *tmp2, *tmp3;
+	AI_alerts_couple *alerts_couple;
+	pthread_t db_thread;
 	int count = 0;
 
 	for ( tmp = *log; tmp; tmp = tmp->next )
@@ -279,6 +282,27 @@ _AI_merge_alerts ( AI_snort_alert **log )
 						/* If the two alerts are equal... */
 						if ( _AI_equal_alerts ( tmp, tmp2->next ))
 						{
+							/* If we are storing the outputs of the module to a database, save the cluster containing the two alerts */
+							if ( config->outdbtype != outdb_none )
+							{
+								if ( !( alerts_couple = (AI_alerts_couple*) malloc ( sizeof ( AI_alerts_couple ))))
+									_dpd.fatalMsg ( "AIPreproc: Fatal dynamic memory allocation error at %s:%d\n", __FILE__, __LINE__ );
+
+								alerts_couple->alert1 = tmp;
+								alerts_couple->alert2 = tmp2->next;
+
+								if ( pthread_create ( &db_thread, NULL, AI_store_cluster_to_db_thread, alerts_couple ) != 0 )
+								{
+									_dpd.fatalMsg ( "AIPreproc: Failed to create the cluster-to-database thread: %s\n", strerror(errno) );
+								}
+
+								if ( pthread_join ( db_thread, NULL ) != 0 )
+								{
+									_dpd.fatalMsg ( "AIPreproc: Could not join the cluster-to-database thread: %s\n", strerror(errno) );
+								}
+							}
+
+							/* Merge the two alerts */
 							if ( !( tmp->grouped_alerts = ( AI_snort_alert** ) realloc ( tmp->grouped_alerts, (++(tmp->grouped_alerts_count)) * sizeof ( AI_snort_alert* ))))
 								_dpd.fatalMsg ( "AIPreproc: Fatal dynamic memory allocation error at %s:%d\n", __FILE__, __LINE__ );
 
@@ -685,7 +709,7 @@ AI_hierarchies_build ( hierarchy_node **nodes, int n_nodes )
 
 	if ( pthread_create ( &cluster_thread, NULL, _AI_cluster_thread, NULL ) != 0 )
 	{
-		_dpd.fatalMsg ( "Failed to create the hash cleanup thread\n" );
+		_dpd.fatalMsg ( "AIPreproc: Failed to create the hash cleanup thread\n" );
 	}
 }		/* -----  end of function AI_hierarchies_build  ----- */
 

correlation.c

@@ -25,6 +25,7 @@
 #include	<unistd.h>
 #include	<time.h>
 #include	<math.h>
+#include	<errno.h>
 #include	<alloca.h>
 #include	<sys/stat.h>
 #include 	<pthread.h>
@@ -44,35 +45,11 @@
 /** Enumeration for the types of XML tags */
 enum  { inHyperAlert, inSnortIdTag, inPreTag, inPostTag, TAG_NUM };
 
-/** Key for the correlation hash table */
-typedef struct  {
-	/** First alert */
-	AI_snort_alert *a;
-
-	/** Second alert */
-	AI_snort_alert *b;
-} AI_alert_correlation_key;
-
-
-/** Struct representing the correlation between all the couples of alerts */
-typedef struct  {
-	/** Hash key */
-	AI_alert_correlation_key  key;
-
-	/** Correlation coefficient */
-	double                    correlation;
-
-	/** Make the struct 'hashable' */
-	UT_hash_handle            hh;
-} AI_alert_correlation;
-
-
 PRIVATE AI_hyperalert_info       *hyperalerts       = NULL;
 PRIVATE AI_snort_alert           *alerts            = NULL;
 PRIVATE AI_alert_correlation     *correlation_table = NULL;
 PRIVATE pthread_mutex_t          mutex;
 
-
 /**
  * \brief  Clean up the correlation hash table
  */
@@ -706,6 +683,8 @@ AI_alert_correlation_thread ( void *arg )
 	AI_snort_alert            *alert_iterator       = NULL,
 					      *alert_iterator2      = NULL;
 
+	pthread_t                 db_thread;
+
 	#ifdef                    HAVE_LIBGVC
 	char                      corr_png_file[4096]   = { 0 };
 	GVC_t                     *gvc                  = NULL;
@@ -873,6 +852,19 @@ AI_alert_correlation_thread ( void *arg )
 					corr->key.a->derived_alerts[ corr->key.a->n_derived_alerts - 1 ] = corr->key.b;
 					corr->key.b->parent_alerts [ corr->key.b->n_parent_alerts  - 1 ] = corr->key.a;
 					_AI_print_correlated_alerts ( corr, fp );
+
+					if ( config->outdbtype != outdb_none )
+					{
+						if ( pthread_create ( &db_thread, NULL, AI_store_correlation_to_db_thread, corr ) != 0 )
+						{
+							_dpd.fatalMsg ( "AIPreproc: Failed to create the correlation-to-database storing thread: %s\n", strerror ( errno ));
+						}
+
+						if ( pthread_join ( db_thread, NULL ) != 0 )
+						{
+							_dpd.fatalMsg ( "AIPreproc: Failed to join the correlation-to-database storing thread: %s\n", strerror ( errno ));
+						}
+					}
 				}
 			}
 

outdb.c

@@ -27,17 +27,40 @@
 #ifdef 	HAVE_DB
 
 #include	"db.h"
+#include	"uthash.h"
 #include	<alloca.h>
 #include	<pthread.h>
 
+/** Enumeration for describing the table in the output database */
 enum  { ALERTS_TABLE, IPV4_HEADERS_TABLE, TCP_HEADERS_TABLE, PACKET_STREAMS_TABLE, CLUSTERED_ALERTS_TABLE, CORRELATED_ALERTS_TABLE, N_TABLES };
 
+/** Tables in the output database */
 static const char *outdb_config[] = {
 	"ca_alerts", "ca_ipv4_headers", "ca_tcp_headers",
 	"ca_packet_streams", "ca_clustered_alerts", "ca_correlated_alerts"
 };
 
+/** Hash table built as cache for the couple of alerts already belonging to the same cluster,
+ * for avoiding more queries on the database*/
+typedef struct  {
+	AI_alerts_couple *alerts_couple;
+	unsigned long    cluster_id;
+	UT_hash_handle   hh;
+} AI_couples_cache;
+
+/** Mutex object, for managing concurrent thread access to the database */
 PRIVATE pthread_mutex_t  mutex;
+PRIVATE AI_couples_cache *couples_cache = NULL;
+
+/**
+ * \brief  Initialize the mutex on the output database
+ */
+
+void
+AI_outdb_mutex_initialize ()
+{
+	pthread_mutex_init ( &mutex, NULL );
+}		/* -----  end of function AI_outdb_mutex_initialize  ----- */
 
 /**
  * \brief  Thread for storing an alert to the database
@@ -47,8 +70,10 @@ PRIVATE pthread_mutex_t  mutex;
 void*
 AI_store_alert_to_db_thread ( void *arg )
 {
-	char srcip[INET_ADDRSTRLEN], dstip[INET_ADDRSTRLEN];
 	char query[65535] = { 0 };
+	char srcip[INET_ADDRSTRLEN],
+		dstip[INET_ADDRSTRLEN];
+
 	unsigned char *pkt_data = NULL;
 	unsigned long latest_ip_hdr_id  = 0,
 			    latest_tcp_hdr_id = 0,
@@ -61,13 +86,11 @@ AI_store_alert_to_db_thread ( void *arg )
 	DB_row    row;
 	AI_snort_alert *alert = (AI_snort_alert*) arg;
 
-	pthread_mutex_init ( &mutex, NULL );
+	pthread_mutex_lock ( &mutex );
 
 	if ( !DB_out_init() )
 		_dpd.fatalMsg ( "AIPreproc: Unable to connect to output database '%s'\n", config->outdbname );
 
-	pthread_mutex_lock ( &mutex );
-
 	inet_ntop ( AF_INET, &(alert->ip_src_addr), srcip, INET_ADDRSTRLEN );
 	inet_ntop ( AF_INET, &(alert->ip_dst_addr), dstip, INET_ADDRSTRLEN );
 
@@ -84,7 +107,7 @@ AI_store_alert_to_db_thread ( void *arg )
 		srcip,
 		dstip );
 
-	DB_out_query ( query );
+	DB_free_result ((DB_result) DB_out_query ( query ));
 
 	memset ( query, 0, sizeof ( query ));
 	snprintf ( query, sizeof ( query ), "SELECT MAX(ip_hdr_id) FROM %s", outdb_config[IPV4_HEADERS_TABLE] );
@@ -92,11 +115,13 @@ AI_store_alert_to_db_thread ( void *arg )
 	if ( !( res = (DB_result) DB_out_query ( query )))
 	{
 		_dpd.logMsg ( "AIPreproc: Warning: error in executing query: '%s'\n", query );
+		pthread_mutex_unlock ( &mutex );
 		pthread_exit ((void*) 0);
 	}
 
 	if ( !( row = (DB_row) DB_fetch_row ( res )))
 	{
+		pthread_mutex_unlock ( &mutex );
 		pthread_exit ((void*) 0);
 	}
 
@@ -118,7 +143,7 @@ AI_store_alert_to_db_thread ( void *arg )
 				ntohs (alert->tcp_window ),
 				ntohs (alert->tcp_len ));
 
-		DB_out_query ( query );
+		DB_free_result ((DB_result) DB_out_query ( query ));
 
 		memset ( query, 0, sizeof ( query ));
 		snprintf ( query, sizeof ( query ), "SELECT MAX(tcp_hdr_id) FROM %s", outdb_config[TCP_HEADERS_TABLE] );
@@ -126,11 +151,13 @@ AI_store_alert_to_db_thread ( void *arg )
 		if ( !( res = (DB_result) DB_out_query ( query )))
 		{
 			_dpd.logMsg ( "AIPreproc: Warning: error in executing query: '%s'\n", query );
+			pthread_mutex_unlock ( &mutex );
 			pthread_exit ((void*) 0);
 		}
 
 		if ( !( row = (DB_row) DB_fetch_row ( res )))
 		{
+			pthread_mutex_unlock ( &mutex );
 			pthread_exit ((void*) 0);
 		}
 
@@ -152,7 +179,7 @@ AI_store_alert_to_db_thread ( void *arg )
 		latest_ip_hdr_id,
 		((alert->ip_proto == IPPROTO_TCP || alert->ip_proto == IPPROTO_UDP) ? latest_tcp_hdr_id : 0));
 
-	DB_out_query ( query );
+	DB_free_result ((DB_result) DB_out_query ( query ));
 
 	memset ( query, 0, sizeof ( query ));
 	snprintf ( query, sizeof ( query ), "SELECT MAX(alert_id) FROM %s", outdb_config[ALERTS_TABLE] );
@@ -160,11 +187,13 @@ AI_store_alert_to_db_thread ( void *arg )
 	if ( !( res = (DB_result) DB_out_query ( query )))
 	{
 		_dpd.logMsg ( "AIPreproc: Warning: error in executing query: '%s'\n", query );
+		pthread_mutex_unlock ( &mutex );
 		pthread_exit ((void*) 0);
 	}
 
 	if ( !( row = (DB_row) DB_fetch_row ( res )))
 	{
+		pthread_mutex_unlock ( &mutex );
 		pthread_exit ((void*) 0);
 	}
 
@@ -196,14 +225,241 @@ AI_store_alert_to_db_thread ( void *arg )
 				pkt->timestamp,
 				pkt_data );
 
-			DB_out_query ( query );
+			DB_free_result ((DB_result) DB_out_query ( query ));
 		}
 	}
 
 	pthread_mutex_unlock ( &mutex );
 	pthread_exit ((void*) 0);
 	return (void*) 0;
-}
+}		/* -----  end of function AI_store_alert_to_db_thread  ----- */
+
+/**
+ * \brief  Store an alert cluster to database
+ * \param  arg 	Struct pointer containing the couple of alerts to be clustered together
+ */
+
+void*
+AI_store_cluster_to_db_thread ( void *arg )
+{
+	unsigned long cluster1 = 0,
+			    cluster2 = 0,
+			    latest_cluster_id = 0;
+
+	char query[1024] = { 0 },
+		srcip[INET_ADDRSTRLEN] = { 0 },
+		dstip[INET_ADDRSTRLEN] = { 0 },
+		srcport[10] = { 0 },
+		dstport[10] = { 0 };
+
+	AI_alerts_couple *alerts_couple = (AI_alerts_couple*) arg;
+	AI_couples_cache *found         = NULL;
+	DB_result res;
+	DB_row    row;
+	BOOL      new_cluster = false;
+
+	pthread_mutex_lock ( &mutex );
+
+	/* Check if the couple of alerts is already in our cache, so it already
+	 * belongs to the same cluster. If so, just return */
+	HASH_FIND ( hh, couples_cache, alerts_couple, sizeof ( AI_alerts_couple ), found );
+
+	if ( found )
+	{
+		pthread_mutex_unlock ( &mutex );
+		pthread_exit ((void*) 0);
+		return (void*) 0;
+	}
+
+	/* Initialize the database (it just does nothing if it is already initialized) */
+	if ( !DB_out_init() )
+		_dpd.fatalMsg ( "AIPreproc: Unable to connect to output database '%s'\n", config->outdbname );
+
+	/* If one of the two alerts has no alert_id, simply return */
+	if ( !alerts_couple->alert1->alert_id || !alerts_couple->alert2->alert_id )
+	{
+		pthread_mutex_unlock ( &mutex );
+		pthread_exit ((void*) 0);
+		return (void*) 0;
+	}
+
+	/* Check if there already exist a cluster containing one of them */
+	memset ( query, 0, sizeof ( query ));
+	snprintf ( query, sizeof ( query ),
+		"SELECT cluster_id FROM %s WHERE alert_id=%lu OR alert_id=%lu",
+		outdb_config[ALERTS_TABLE], alerts_couple->alert1->alert_id, alerts_couple->alert2->alert_id );
+
+	if ( !( res = (DB_result) DB_out_query ( query )))
+	{
+		_dpd.logMsg ( "AIPreproc: Warning: error in executing query: '%s'\n", query );
+		pthread_mutex_unlock ( &mutex );
+		pthread_exit ((void*) 0);
+		return (void*) 0;
+	}
+
+	if ( !( row = (DB_row) DB_fetch_row ( res )))
+	{
+		pthread_mutex_unlock ( &mutex );
+		pthread_exit ((void*) 0);
+		return (void*) 0;
+	}
+
+	/* If no cluster exists containing at least of them, create it */
+	new_cluster = false;
+
+	if ( !row[0] && !row[1] )
+	{
+		new_cluster = true;
+	} else {
+		if ( row[0] )
+		{
+			cluster1 = strtoul ( row[0], NULL, 10 );
+		}
+
+		if ( row[1] )
+		{
+			cluster2 = strtoul ( row[1], NULL, 10 );
+		}
+
+		if ( cluster1 == 0 && cluster2 == 0 )
+		{
+			new_cluster = true;
+		}
+	}
+
+	DB_free_result ( res );
+
+	/* If both the alerts already belong to the same cluster (but they're not in the cache yet),
+	 * insert them in the cache */
+	if ( cluster1 != 0 && cluster2 != 0 && cluster1 == cluster2 )
+	{
+		if ( !( found = ( AI_couples_cache* ) malloc ( sizeof ( AI_couples_cache ))))
+			_dpd.fatalMsg ( "AIPreproc: Fatal dynamic allocation memory at %s:%d\n", __FILE__, __LINE__ );
+		
+		found->alerts_couple = alerts_couple;
+		found->cluster_id = cluster1;
+		HASH_ADD ( hh, couples_cache, alerts_couple, sizeof ( AI_alerts_couple ), found );
+		pthread_mutex_unlock ( &mutex );
+		pthread_exit ((void*) 0);
+		return (void*) 0;
+	}
+
+	if ( new_cluster )
+	{
+		/* Insert a new cluster containing alert1 and alert2 for now */
+		inet_ntop ( AF_INET, &(alerts_couple->alert1->ip_src_addr), srcip, INET_ADDRSTRLEN );
+		inet_ntop ( AF_INET, &(alerts_couple->alert1->ip_dst_addr), dstip, INET_ADDRSTRLEN );
+		snprintf ( srcport, sizeof ( srcport ), "%u", ntohs( alerts_couple->alert1->tcp_src_port ));
+		snprintf ( dstport, sizeof ( dstport ), "%u", ntohs( alerts_couple->alert1->tcp_dst_port ));
+
+		memset ( query, 0, sizeof ( query ));
+		snprintf ( query, sizeof ( query ),
+			"INSERT INTO %s ( clustered_srcip, clustered_dstip, clustered_srcport, clustered_dstport ) "
+			"VALUES ( '%s', '%s', '%s', '%s' )",
+			outdb_config[CLUSTERED_ALERTS_TABLE],
+			((alerts_couple->alert1->h_node[src_addr]) ? alerts_couple->alert1->h_node[src_addr]->label : srcip),
+			((alerts_couple->alert1->h_node[dst_addr]) ? alerts_couple->alert1->h_node[dst_addr]->label : dstip),
+			((alerts_couple->alert1->h_node[src_port]) ? alerts_couple->alert1->h_node[src_port]->label : srcport),
+			((alerts_couple->alert1->h_node[dst_port]) ? alerts_couple->alert1->h_node[dst_port]->label : dstport)
+		);
+
+		DB_free_result ((DB_result) DB_out_query ( query ));
+
+		memset ( query, 0, sizeof ( query ));
+		snprintf ( query, sizeof ( query ),
+			"SELECT MAX(cluster_id) FROM %s", outdb_config[CLUSTERED_ALERTS_TABLE] );
+
+		if ( !( res = (DB_result) DB_out_query ( query )))
+		{
+			_dpd.logMsg ( "AIPreproc: Warning: error in executing query: '%s'\n", query );
+			pthread_mutex_unlock ( &mutex );
+			pthread_exit ((void*) 0);
+			return (void*) 0;
+		}
+
+		if ( !( row = (DB_row) DB_fetch_row ( res )))
+		{
+			pthread_mutex_unlock ( &mutex );
+			pthread_exit ((void*) 0);
+			return (void*) 0;
+		}
+
+		latest_cluster_id = strtoul ( row[0], NULL, 10 );
+		DB_free_result ( res );
+
+		/* Update the two alerts, setting them as belonging to the new cluster */
+		memset ( query, 0, sizeof ( query ));
+		snprintf ( query, sizeof ( query ),
+			"UPDATE %s SET cluster_id=%lu WHERE alert_id=%lu OR alert_id=%lu",
+			outdb_config[ALERTS_TABLE], latest_cluster_id,
+			alerts_couple->alert1->alert_id, alerts_couple->alert2->alert_id );
+
+		DB_free_result ((DB_result) DB_out_query ( query ));
+	} else {
+		/* Update the alert marked as 'not clustered' */
+		if ( !cluster1 )
+		{
+			memset ( query, 0, sizeof ( query ));
+			snprintf ( query, sizeof ( query ),
+				"UPDATE %s SET cluster_id=%lu WHERE alert_id=%lu",
+				outdb_config[ALERTS_TABLE], cluster2, alerts_couple->alert1->alert_id );
+
+			DB_free_result ((DB_result) DB_out_query ( query ));
+		} else {
+			memset ( query, 0, sizeof ( query ));
+			snprintf ( query, sizeof ( query ),
+				"UPDATE %s SET cluster_id=%lu WHERE alert_id=%lu",
+				outdb_config[ALERTS_TABLE], cluster1, alerts_couple->alert2->alert_id );
+
+			DB_free_result ((DB_result) DB_out_query ( query ));
+		}
+	}
+
+	/* Add the couple to the cache */
+	if ( !( found = ( AI_couples_cache* ) malloc ( sizeof ( AI_couples_cache ))))
+		_dpd.fatalMsg ( "AIPreproc: Fatal dynamic allocation memory at %s:%d\n", __FILE__, __LINE__ );
+
+	found->alerts_couple = alerts_couple;
+	found->cluster_id = cluster1;
+	HASH_ADD ( hh, couples_cache, alerts_couple, sizeof ( AI_alerts_couple ), found );
+
+	pthread_mutex_unlock ( &mutex );
+	pthread_exit ((void*) 0);
+	return (void*) 0;
+}		/* -----  end of function AI_store_cluster_to_db_thread  ----- */
+
+
+/**
+ * \brief  Store the correlation between two alerts to the output database
+ * \param  arg 	Structure containing the two alerts to be saved and their correlation
+ */
+
+void*
+AI_store_correlation_to_db_thread ( void *arg )
+{
+	char query[1024] = { 0 };
+	AI_alert_correlation *corr = (AI_alert_correlation*) arg;
+
+	pthread_mutex_lock ( &mutex );
+
+	/* Initialize the database (it just does nothing if it is already initialized) */
+	if ( !DB_out_init() )
+		_dpd.fatalMsg ( "AIPreproc: Unable to connect to output database '%s'\n", config->outdbname );
+
+	memset ( query, 0, sizeof ( query ));
+	snprintf ( query, sizeof ( query ),
+		"INSERT INTO %s ( alert1, alert2, correlation_coeff ) "
+		"VALUES ( %lu, %lu, %f )",
+		outdb_config[CORRELATED_ALERTS_TABLE],
+		corr->key.a->alert_id,
+		corr->key.b->alert_id,
+		corr->correlation );
+	DB_free_result ((DB_result) DB_out_query ( query ));
+
+	pthread_mutex_unlock ( &mutex );
+	pthread_exit ((void*) 0);
+	return 0;
+}		/* -----  end of function AI_store_correlation_to_db_thread  ----- */
 
 #endif
 

schemas/mysql.sql

@@ -50,7 +50,7 @@ CREATE TABLE ca_alerts (
 	timestamp      datetime,
 	ip_hdr         integer,
 	tcp_hdr        integer,
-	cluster_id     integer,
+	cluster_id     integer default 0,
 
 	primary key(alert_id),
 	foreign key(ip_hdr) references ca_ip_headers(ip_hdr_id),
@@ -71,12 +71,12 @@ CREATE TABLE ca_clustered_alerts (
 
 DROP TABLE IF EXISTS ca_correlated_alerts;
 CREATE TABLE ca_correlated_alerts (
-	cluster1          integer,
+	alert1            integer,
-	cluster2          integer,
+	alert2            integer,
 	correlation_coeff double,
 
-	primary key(cluster1, cluster2),
+	primary key(alert1, alert2),
-	foreign key(cluster1) references ca_clustered_alerts(cluster_id),
+	foreign key(alert1) references ca_alerts(alert_id),
-	foreign key(cluster2) references ca_clustered_alerts(cluster_id)
+	foreign key(alert2) references ca_alerts(alert_id)
 );
 

spp_ai.c

@@ -747,6 +747,7 @@ static AI_config * AI_parse(char *args)
 			_dpd.fatalMsg ( "AIPreproc: Output database option used in config, but missing configuration option (all 'host', 'type', 'name', 'user', and 'password' options must be used)\n" );
 		}
 
+		AI_outdb_mutex_initialize();
 		_dpd.logMsg("    Saving output alerts to the database %s\n", config->outdbname );
 	}
 

spp_ai.h

@@ -359,6 +359,34 @@ typedef struct _AI_alert_event  {
 	struct _AI_alert_event  *next;
 	UT_hash_handle          hh;
 } AI_alert_event;
+/*****************************************************************/
+/** Simple structure for holding a couple of alerts to be merged, to be passed to the outdb thread */
+typedef struct  {
+	AI_snort_alert *alert1;
+	AI_snort_alert *alert2;
+} AI_alerts_couple;
+/*****************************************************************/
+/** Key for the correlation hash table */
+typedef struct  {
+	/** First alert */
+	AI_snort_alert *a;
+
+	/** Second alert */
+	AI_snort_alert *b;
+} AI_alert_correlation_key;
+/*****************************************************************/
+/** Struct representing the correlation between all the couples of alerts */
+typedef struct  {
+	/** Hash key */
+	AI_alert_correlation_key  key;
+
+	/** Correlation coefficient */
+	double                    correlation;
+
+	/** Make the struct 'hashable' */
+	UT_hash_handle            hh;
+} AI_alert_correlation;
+/*****************************************************************/
 
 int                preg_match ( const char*, char*, char***, int* );
 char*              str_replace ( char*, char*, char *);
@@ -391,7 +419,10 @@ const AI_alert_event*  AI_get_alert_events_by_key ( AI_alert_event_key );
 unsigned int           AI_get_history_alert_number ();
 double                 AI_alert_bayesian_correlation ( AI_snort_alert *a, AI_snort_alert *b );
 
+void                   AI_outdb_mutex_initialize ();
 void*                  AI_store_alert_to_db_thread ( void* );
+void*                  AI_store_cluster_to_db_thread ( void* );
+void*                  AI_store_correlation_to_db_thread ( void* );
 
 /** Function pointer to the function used for getting the alert list (from log file, db, ...) */
 extern AI_snort_alert* (*get_alerts)(void);