PREPARE/tracing/modules/db/database.js

/**
 * TraCINg-Server - Gathering and visualizing cyber incidents on the world
 *
 * Copyright 2013 Matthias Gazzari, Annemarie Mattmann, André Wolski
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
var MongoClient = require('mongodb').Db,
	ObjectId = require('mongodb').ObjectID,
	config      = require("../../config.json"),
	assert      = require('assert'),
	_           = require('underscore');

var fields = require("../fields.js");
var db = null;
var collection = null;
var collectionName = "incident";

function safeCallback(callback){
	if(callback){
		callback.apply(this, Array.prototype.splice.call(arguments, 1));
	}
}


function connect(callback){
	MongoClient.connect(config.db, function(err, conn){
		assert.equal(null, err);
		console.log("Connected successfully to the database");

		db = conn;
		collection = conn.collection(collectionName);

		collection.ensureIndex({"src.ll" : "2dsphere"}, function(err, result) {
			collection.ensureIndex({"date": 1}, function(err, result){
				collection.ensureIndex({"sync_id": 1, "device": 1}, {
					unique: true,
					sparse: true
				}, function (err, result) {
					assert.equal(null, err);
					safeCallback(callback);
				});
			});
		});
	});
}

// CONSTANTS
var DB_NOT_READY = 0;

function createError(number){
	var msg = {errno: number, code: ""};

	switch(number){
		case DB_NOT_READY:
			msg.code = "The database is not ready";
			break;
	}

	return msg;
}

function ensureDBReady(callback){
	if(!db){
		console.log("db is not ready!");

		safeCallback(callback, createError(DB_NOT_READY));
		return false;
	}

	return true;
}

function getQueryCriteriaFromFilter(filter){
	if(filter.hasOwnProperty("raw") && filter.raw){
		return filter.criteria;
	}

	var criteria = {};

	if(filter.hasOwnProperty("start") && filter.hasOwnProperty("end")){
		criteria["date"] = {"$gte": new Date(filter["start"]), "$lte": new Date(filter["end"])};
	} else if(filter.hasOwnProperty("start")){
		criteria["date"] = {"$gte": new Date(filter["start"])};
	} else if(filter.hasOwnProperty("end")){
		criteria["date"] = {"$lte": new Date(filter["end"])};
	}

	if(filter.hasOwnProperty("countries")){
		var countries = filter["countries"];
		if(!_.isArray(countries)) countries = [countries];

		criteria["src.cc"] = {"$in": countries};
	}

	if(filter.hasOwnProperty("countries_dest")){
		var countries_dst = filter["countries"];
		if(!_.isArray(countries_dst)) countries_dst = [countries_dst];

		criteria["dst.cc"] = {"$in": countries_dst};
	}

	if(filter.hasOwnProperty("location")){
		criteria["dst.ll"] = {"$nearSphere": filter.location};

		if(filter.hasOwnProperty("distance")){
			criteria["dst.ll"]["$maxDistance"] = parseInt(filter.distance, 10) / 6371;
		}
	}

	if(filter.hasOwnProperty("authorized")){
		criteria["authorized"] = true;
	}

	if(filter.hasOwnProperty("types")){
		var types = filter["types"];
		if(!_.isArray(types)) types = [types];

		criteria["type"] = {"$in": _.map(types, _.partial(parseInt, _, 10))};
	}

	if(filter.hasOwnProperty("sensors")){
		var sensors = filter["sensors"];
		if(!_.isArray(sensors)) sensors = [sensors];

		criteria["sensortype"] = {"$in": sensors};
	}

	if(filter.hasOwnProperty("query") && filter.query && filter.query.length > 0) {
		var regexp = new RegExp('.*' + filter.query + '.*', 'i');

		criteria["$or"] = [{
			sensorname: regexp
		}, {
			log: regexp
		}, {
			md5sum: regexp
		}]
	}

	return criteria;
}

// ------------------------------ PUBLIC / EXPORT ------------------------------

function getSensorTypes(callback){
	console.log("called getSensorTypes");
	if(!ensureDBReady(callback)) return;

	collection.group({'sensortype': 1}, {}, {}, function(){}, function(err, docs){
		if(err){
			safeCallback(callback, err);
			return;
		}

		var sensors = [];
		for(var k in docs){
			sensors.push(docs[k]["sensortype"]);
		}

		safeCallback(callback, null, sensors);
	});
}

function insert(items, callback){
	//console.log("called insert");
	if(!ensureDBReady(callback)) return;

	collection.insert(items, {continueOnError: true}, function(err, result){
		if(err){
			if(err.code !== 11000) {
				console.error(err);
				safeCallback(callback, err);
				return;
			}
		}

		safeCallback(callback, null, result);
	});
}

function requestAttacks(filter, callback, options){
	console.log("called requestAttacks", filter);
	if(!ensureDBReady(callback)) return;

	console.log(getQueryCriteriaFromFilter(filter));

	if(filter.page && filter.page === 'all'){
		var pipeline = [];

		var group = {
			_id: {
				city: "$src.city",
				country: "$src.country",
				cc: "$src.cc"
			},
			count: {"$sum": 1},
			date: {"$first": "$date"},
			location: {$first: "$src.ll"},
			destination: {$first: {$concat: ["$dst.city", ", ", "$dst.country"]}},
			type: {$first: "$type"},
			authorized: {$first: "$authorized"}
		};

		var criteria = getQueryCriteriaFromFilter(filter);

		if(criteria){
			pipeline.push({"$match": criteria});
		}

		pipeline.push({"$group": group});

		// we need to aggregate the data
		collection.aggregate(pipeline, {allowDiskUse: true}, function(err, res){
			if (err) {
				console.error(err);
				safeCallback(callback, err);
				return;
			}

			safeCallback(callback, null, res);
		});

		return;
	}

	var result = collection.find(getQueryCriteriaFromFilter(filter)).sort({date: -1});

	if(filter.page >= 0){
		var perpage = filter.perpage ? (filter.perpage > 50 ? 50 : filter.perpage) : 10;

		result.skip(filter.page).limit(perpage);
	}

	var defaultOptions = {"type": "array"};
	if(!options){
		options = defaultOptions;
	}

	if(options.type === "array") {
		result.count(function(error, total){
			result.toArray(function (err, docs) {
				if (err) {
					console.error(err);
					safeCallback(callback, err);
					return;
				}

				safeCallback(callback, null, options.sync ? docs : {total: total, data: docs});
			});
		});
	} else if(options.type === "stream") {
		var stream = result.stream();
		options.stream.pipe(stream);
	} else {
		var stream = result.stream();
		var str = "[";
		var first = true;
		var response = options.response || null;
		if(response){
			response.write("[");
		}

		stream.on('data', function(item){
			var prefix = first ? '' : ', ';

			if(response){
				response.write(prefix + JSON.stringify(item));
			} else {
				str += prefix + JSON.stringify(item);
			}

			first = false;
		});

		stream.on('end', function() {
			if(response){
				response.write("]");
				response.end();
			} else {
				str += "]";
				safeCallback(callback, false, str);
			}
		});

		stream.on('error', function(err){
			console.error(err);
			safeCallback(callback, err);
		})
	}
}

function getLog(id, callback){
	console.log("called getLog");
	if(!ensureDBReady(callback)) return;

	collection.findOne({_id: ObjectId(id)}, {log: 1}, function(err, res){
		if(err){
			console.error(err);
			safeCallback(callback, err);
			return;
		}

		console.log(id, err, res);
		safeCallback(callback, null, (res && res.log) || "");
	});
}

function extractSeries(rows, serieName, serieDisplay){
	var series = [];

	var convertRow = function(row){
		var timestamp = Date.UTC(
			row.datum.getUTCFullYear(),
			row.datum.getUTCMonth(),
			row.datum.getUTCDate()
		);

		return [
			timestamp,
			row.count
		];
	};

	if (!serieName) {
		series.push({
			name: "total",
			data: _.map(rows, convertRow)
		});
		console.log("total", series);

	} else {
		var serie = {};

		for (var k in rows) {
			if(!rows.hasOwnProperty(k)) continue;

			var row = rows[k];
			var rowSerieValue = row._id[serieName] + "";

			if (!serie.hasOwnProperty(rowSerieValue)) {
				serie[rowSerieValue] = {data: [], name: ""};

				if (serieDisplay)
					serie[rowSerieValue].name = row[serieDisplay];
				else
					serie[rowSerieValue].name = fields.translate(serieName, rowSerieValue);
			}

			// push this row to the current serie
			serie[rowSerieValue].data.push(convertRow(row));
		}

		series = _.values(serie);
	}

	return series;
}


/*
 converts the dates to unix time in all data of all series

 determines minDate and maxDate if not given

 fills all series with 0-data on each date between minDate and maxDate if it has no data on that day
 */
function convertDates(series, minTimestamp, maxTimestamp){
	var checkMinMax = !minTimestamp || !maxTimestamp; // no logical need for this check, but should give performance as in most cases min and max are given

	// convert the dates to UTS timestamps for each serie
	for(var i = 0; i < series.length; i++){
		// convert it for each data entry
		for (var j = 0; j < series[i].data.length; j++) {
			var timestamp = series[i].data[j][0];

			if(!minTimestamp || minTimestamp > timestamp)
				minTimestamp = timestamp;
			if(!maxTimestamp || maxTimestamp < timestamp)
				maxTimestamp = timestamp;
		}
	}

	for(var jj = 0; jj < series.length; jj++){
		var currentTimestamp = minTimestamp;
		var ii = 0;
		var maxI = series[jj].data.length;
		var newData = [];
		while(currentTimestamp <= maxTimestamp){
			//console.log("convertDates.currentTimestamp:", currentTimestamp, new Date(currentTimestamp));

			if(ii >= maxI || currentTimestamp < series[jj].data[ii][0]){
				newData.push([currentTimestamp, 0]);
			}
			else{
				newData.push([currentTimestamp, series[jj].data[ii][1]]);
				ii++;
			}

			// next day
			currentTimestamp += 24 * 60 * 60 * 1000;
		}

		series[jj].data = newData;
	}
}

function getStatistics(options, callback){
	console.log("called getStatistics");
	var pipeline = [];

	var criteria = null;
	if(options.filter){
		criteria = getQueryCriteriaFromFilter(options.filter);
	}

	var group = {
		_id: {
			year: {"$year": "$date"},
			month: {"$month": "$date"},
			day: {"$dayOfMonth": "$date"}
		},
		count: {"$sum": 1},
		datum: {"$first": "$date"}
	};

	var project = null;
	var order = {date: 1};
	var serieName = null;
	var serieDisplay = null;

	if(options.detail){
		switch(options.detailChartType){
			case "typeDate":
				group._id["type"] = "$type";
				order["type"] = 1;
				serieName = "type";

				break;

			case "countryDate":
				if(!project){
					project = {};
				}

				group._id["source_cc"] = "$src.cc";
				group["source_country"] = {"$first": "$src.country"};
				order["src.cc"] = 1;

				serieName = "source_cc";
				serieDisplay = "source_country";
				break;

			default:
				callback("getStatistics: unknown detailChartType: " + detailChartType);
				return;
		}
	}

	pipeline.push({"$sort": order});

	//if(project) pipeline.push({"$project": project});
	if(criteria){
		pipeline.push({"$match": criteria});
	}

	pipeline.push({"$group": group});
	pipeline.push({"$sort": {datum: 1, count: 1}});

	collection.aggregate(pipeline, {allowDiskUse: true}, function(err, res){
		if(err){
			safeCallback(callback, err);
		}

		if(!res || res.length == 0){
			safeCallback(callback, null, []);
			return;
		}

		var series = extractSeries(res, serieName, serieDisplay);

		if(options.filter)
			convertDates(series, options.filter.start, options.filter.end);
		else
			convertDates(series);

		safeCallback(callback, null, series);
	});
}

function getAllCountries(callback){
	collection.aggregate([{
		$group: {
			"_id": {"cc": "$dst.cc", "country": "$dst.country"}
		}
	}], null, function(err, res){
		if(err){
			safeCallback(callback, err);
		}

		safeCallback(callback, null, res.filter(function(d){ return d._id.cc != 0; }).map(function(d){
			return d._id;
		}));
	});
}

function getEntriesFromSyncInfo(sync_request, callback) {
	if(!sync_request.hasOwnProperty("condition") || !sync_request.hasOwnProperty("info")){
		throw new Error("Invalid sync request");
	}

	var sync_condition = sync_request.condition;
	var sync_info = sync_request.info;
	var sync_device_condition = [];
	var device_list = [];

	for(var i=0; i<sync_info.length; i++){
		var entry = sync_info[i];

		sync_device_condition.push({
			"$and": [
				{"sync_id": {"$gt": entry.sync_id}},
				{"device": entry.device}
			]
		});

		device_list.push(entry.device);
	}

	sync_device_condition.push({"device": {"$nin": device_list}});

	var criteria = {
		"sensorname": "HosTaGe",
		"date": {"$gte": new Date(sync_condition.date || 0)},
		"$or": sync_device_condition
	};

	if(sync_condition.hasOwnProperty('location')){
		criteria["dst.ll"] = {"$nearSphere": sync_condition.location};

		if(sync_condition.hasOwnProperty("distance")){
			criteria["dst.ll"]["$maxDistance"] = parseInt(sync_condition.distance, 10) / 6371;
		} else {
			criteria["dst.ll"]["$maxDistance"] = 500 / 6371;
		}
	}

	if(sync_condition.hasOwnProperty('country')){
		criteria["dst.cc"] = sync_condition.country;
	}

	requestAttacks({
		raw: true,
		criteria: criteria
	}, function (error, data) {
		if(error){
			safeCallback(callback, error);
		}
		safeCallback(callback, null, data);
	}, {type: 'array', sync: true});
}

function getDatabase(){
	return db;
}

function closeDatabase(){
	db.close();
}

module.exports = {
	connect: connect,
	close: closeDatabase,
	getDB:  getDatabase,
	getSensorTypes: getSensorTypes,
	insert: insert,
	requestAttacks: requestAttacks,
	getLog: getLog,
	getStatistics: getStatistics,
	getAllCountries: getAllCountries,
	getEntriesFromSyncInfo: getEntriesFromSyncInfo
};