PREPARE/tracing/modules/proberesponseMitigation.js

/**
 * This module is part of the TraCINg CIDS and introduces mitigation methodologies
 * against the probe response attack by analyzing statistical anomalies. On identification
 * of an attack, countermeasures are introduced which modify the way attack events get
 * presented to the user (see postHandler.js).
 */

// load node modules
var Chance = require("chance");
var fs = require('fs');
var path = require('path');
var Map = require("collections/map");
var Set = require("collections/set");
var es = require("event-stream");

var debugInstance = 0;
var Debug = function(){
        var output = "";
        var instance = debugInstance++;

        this.get = function(){
                return output;
        };

        this.add = function(text){
                text = "Proberesponse["+instance+"]: " + text;
                console.log(text);
                output += text + "\n";
        };
};

var debug = new Debug();
var chance = new Chance();


// TBA: duration of one time slot in a window
// 10 fits for destination port, 6ms
var TIMESLOT_SECONDS = 60;

// TBA: amount of "TIMESLOT_SECONDS" for a time window to save, total seconds = WINDOW_MAX_TIMESLOTS_RATIO * TIMESLOT_SECONDS
var WINDOW_MAX_TIMESLOTS_RATIO = 60;
// Attack count for every source IP address: { ip_source : { ip_destination : count}, ... }
var ip_source_dest_count = {};
// List of ip_source_dest_count for every slot in the time window [ ip_count1, ip_count2, ... ]
// This is a round robbin buffer where the oldest entries get overwritten by the newest.
var window_ip_source_dest = [];

// TBA; Ratio of (attacks/unique monitors) for the whole time window which triggers an alarm.
// There should be a save margin upwards to a "normal" ratio.
// ratio can not be lower than 1 (at minimum 1 attack per monitor)
var THRESHOLD_ATTACKS_MONITOR_RATIO = 3;	// depends on TIMESLOT_SECONDS and WINDOW_MAX_TIMESLOTS_RATIO and amount of events/s

// >>> TBA: Alarm thresholds for the whole time window WINDOW_MAX_TIMESLOTS_RATIO
// depents on TIMESLOT_SECONDS * WINDOW_MAX_TIMESLOTS_RATIO
// TBA: Amount of unique monitors per source (adress:monitors -> 1:n -> check n)
var THRESHOLD_MONITORS_PER_SOURCE_IP = 50000;	// ~20% of all monitor nodes.
// TBA: Amount of unique sources for one monitors (monitor:address -> 1:n -> check n)
var THRESHOLD_SOURCE_IPS_PER_MONITOR = 100;
// TBA: Amount of attacks from one source to a different monitors (monitor:address -> 1:1 -> check amount of attacks)
var THRESHOLD_SOURCE_IPS_MONITOR_ATTACKS = 6000;
// TBA: Amount of total attacks for one monitors (monitor:attacks -> 1:n -> check n)
var THRESHOLD_ATTACKS_PER_MONITOR = 15000;


// >>> Alarm thresholds for destination port statistics
// TBA: amount of "TIMESLOT_SECONDS" for a time windowto save, total seconds = WINDOW_MAX_TIMESLOTS_RATIO * TIMESLOT_SECONDS
var WINDOW_MAX_TIMESLOTS_PORT_DST = 60;
// Attack count for every destination port: { port : count, ...}
var port_dest_count = new Map();
// List of dicts of ports [ port_dest_count1, port_dest_count2, ... ]
// This is a round robbin buffer where the oldest entries get overwritten by the newest.
var window_port_dest = [];
// TBA: max amount of attacked destination ports in TIMESLOT_SECONDS*WINDOW_MAX_TIMESLOTS_PORT_DST seconds
var THRESHOLD_PORT_DEST_ATTACKED = 7000;

// TBA: Ban duration AFTTER source count falls below SOURCE_IP_THRESHOLD
var SOURCE_IP_BAN_MINUTES = 1;
// { ip : ban_time_minutes }
var banned_ips = {}

// Sampling of events: just take 0% -> 100% of all events
var sampling_percent = 100;
var sampling_active = false;
// noise information
var events_dshield_total = 0;
var events_dshield_added = 0;
var events_external_total = 0;
var events_external_added = 0;

// handle mitigation if activated (does NOT influence metrics -> those are further calculated but external events are ignored)
var is_active = true;

var dshield_entries = [];
// 24 events per second
var noise_interval_ms = 41;
//var noise_interval_ms = 7;
var FILE_STATS = "attack_stats.txt";
var dshield_read_linenr = 0;
var init_started = false;


function init_stats_window() {
	// Amount of events for: TIMESLOT_SECONDS * WINDOW_MAX_TIMESLOTS_RATIO
	// 27 events/s = 97200 events/window (60*60 seconds) = ~ ratio 5.8
	//var attacks_init_win = 97200;
	var attacks_init_win = Math.round(1000/noise_interval_ms)*TIMESLOT_SECONDS*WINDOW_MAX_TIMESLOTS_RATIO;

	var attacks_per_slot = Math.round(attacks_init_win / WINDOW_MAX_TIMESLOTS_RATIO);
	var port_count = new Map();

	debug.add("amount of dshield entries: "+ dshield_entries.length);
	debug.add("initiating window ip: "+WINDOW_MAX_TIMESLOTS_RATIO+"*"+TIMESLOT_SECONDS+
		" sec, window port: "+WINDOW_MAX_TIMESLOTS_PORT_DST+"*"+TIMESLOT_SECONDS+" sec");

	var cnt = 0;

	for(var x=0; x < WINDOW_MAX_TIMESLOTS_RATIO; ++x){
		//window_ip_source_dest[x] = {"0.0.0.0" : {"0.0.0.0" : 0}};
		window_ip_source_dest[x] = {};
		window_port_dest[x] = new Map();

		for (var y=0; y < attacks_per_slot; ++y) {
			entry = dshield_entries.pop();
			//entry = dshield_entries.shift();
			//l = [ipsrc, ipmon, chance.integer({min:0, max:65535})];
			l = [entry[0], entry[2], entry[3]];

			if (l in window_ip_source_dest[x]) {
				window_ip_source_dest[x][l] += 1;
				//debug.add("adding, total: "+ window_ip_source_dest[x][l]);
			}
			else
				window_ip_source_dest[x][l] = 1;

			if (port_count.get(entry[3]) == undefined) {
				port_count.set(entry[3], 1);
			} else {
				port_count.set(entry[3], port_count.get(entry[3]) + 1);
			}

			if (window_port_dest[x].get(entry[3]) == undefined) {
				window_port_dest[x].set(entry[3], 1);
			} else {
				window_port_dest[x].set(entry[3], window_port_dest[x].get(entry[3]) + 1);
			}
			cnt += 1;
		}
	}

	var port_cnt_check = 0;
	port_count.forEach( function merge(count, port, mapObj) {
		port_cnt_check += count;
	});


	debug.add("unique ports: "+ port_count.length);
	debug.add("finished initiating, count: "+attacks_init_win+"="+port_cnt_check+"=~"+cnt);
	// add noise using DShield data
	var noise_ref = setInterval(add_entry_dshield, noise_interval_ms);
	// mitigation check-logic
	//var check_attack_status_ref = setInterval(check_attack_status, TIMESLOT_SECONDS * 1000);
}

var FILE_DSHIELD_EVENTS = path.join(__dirname, 'dshield_report.csv');
debug.add("reading DShield events from "+FILE_DSHIELD_EVENTS+", standy by");

s = fs.createReadStream(FILE_DSHIELD_EVENTS)
    .pipe(es.split())
    .pipe(es.mapSync(function(line){
        // pause the readstream
        s.pause();
        dshield_read_linenr += 1;

        (function(){
                        var columns = line.toString().split("\t", -1);

                        if (columns.length == 6) {
                        try {
                                // 3 4 [8 -> IP] 5 8: ip_src port_src ip_dst port_dst id
                                // 0 1 [5 -> IP] 2 5: ip_src port_src ip_dst port_dst id
                                // convert monitor ID to an IP address
                                a = parseInt(columns[5].substring(2, 4), 16);
                                b = parseInt(columns[5].substring(4, 6), 16);
                                c = parseInt(columns[5].substring(6, 8), 16);
                                d = parseInt(columns[5].substring(8, 10), 16);
                                //debug.add("IP is: " + a+"."+b+"."+c+"."+d);

                                dshield_entries.push([columns[0],
                                                parseInt(columns[1]),
                                                a+"."+b+"."+c+"."+d,
                                                columns[2],
                                                columns[5]]);
                        } catch (err) { /* ignore malformed lines */ }
	               }

            // process line here and call s.resume() when rdy
            //logMemoryUsage(lineNr);
            if (dshield_read_linenr % 200000 == 0) debug.add("read DShield events: "+dshield_read_linenr);
	    //if (!init_started && dshield_read_linenr > 2000000) {
	    if (!init_started && dshield_read_linenr > 3000000) {
	    //if (!init_started && dshield_read_linenr > 5000000) {
	    //if (!init_started && dshield_read_linenr > 7000000) {
	    //if (!init_started && dshield_read_linenr > 6000000) {
	    //if (!init_started && dshield_read_linenr > 9000000) {
	    //if (!init_started && dshield_read_linenr > 12000000) {
		s.destroy(); init_stats_window(); init_started = true;
		}

            // resume the readstream
            s.resume();

        })();
    })
    .on('error', function(){ debug.add('Error while reading DShield report'); })
    .on('end', function(){ debug.add('Read entire DShield report') })
);


function update_slot(monitor_id, ip_source, ip_dest, port_dest) {
	//debug.add("updating window: "+monitor_id+" / "+ip_source+" / "+ip_dest+" / "+port_dest);
	//if (!is_active) { return; }

	/*
	if (!(ip_source === undefined) && (ip_source in banned_ips)) {
		// don't count banned IP addresses
		//debug.add("won't count banned ip: "+ ip_source)
		return;
	}
	*/

	//l = [ip_source, ip_dest, port_dest];
	l = [ip_source, monitor_id, port_dest];

	if (ip_source_dest_count[l] == undefined) { ip_source_dest_count[l] = 0; }

	ip_source_dest_count[l] += 1;

	if (port_dest_count.get(port_dest) == undefined) {
		port_dest_count.set(port_dest, 1);
	} else {
		port_dest_count.set(port_dest, port_dest_count.get(port_dest) + 1);
	}
}


// clear old attack stats file
fs.writeFile(FILE_STATS, "ratio\tports\talarm_ratio\talarm_port\tevents_dshield_added\tevents_dshield_total\tevents_external_added\tevents_extern_total\n", function (err) { if (err) throw err; });

// update ban status of any source IP address
function check_attack_status() {
	//debug.add("checking attack status");
	// >>> Update window with new slot
	var time_seconds_now = Math.round(new Date().getTime() / 1000);
	var time_minutes_now = Math.round(time_seconds_now / 60);
	var timeslot_ratio = Math.round(time_seconds_now / TIMESLOT_SECONDS) % WINDOW_MAX_TIMESLOTS_RATIO;
	// window_ip_source_dest[x] = [a, b, c]
	window_ip_source_dest[timeslot_ratio] = ip_source_dest_count;
	// reset count for next slot
	ip_source_dest_count = {};

	var timeslot_ports = Math.round(time_seconds_now / TIMESLOT_SECONDS) % WINDOW_MAX_TIMESLOTS_PORT_DST;
	window_port_dest[timeslot_ports] = port_dest_count;
	// reset count for next slot
	port_dest_count = new Map();

	var alarm_ratio_triggered = false;
	var alarm_portdest_triggered = false;
	var alarm_attacks_per_monitor_triggered = false;

	var sampling = 100;

	// >>> check threshold: amount of unique attacked destination ports
	//debug.add("counting ports...");
	var port_dest_count_merged = new Map();
	var top_ports = new Set();
	alarm_portdest_triggered = false;

	/**/
	// TODO: activate mitigation based on port frequency if needed
	for (var timeslot in window_port_dest) {
		window_port_dest[timeslot].forEach( function merge(count, port, mapObj) {
			if (port_dest_count_merged.get(port) == undefined) {
				port_dest_count_merged.set(port, count);
				//debug.add("setting port: "+port+" = count "+count);
			} else {
				port_dest_count_merged.set(port, port_dest_count_merged.get(port) + count);
			}
		});
	}

	// sort keys by value descending
	var top_ports_all = port_dest_count_merged.keys().sort(function(a,b){
		return port_dest_count_merged.get(b) - port_dest_count_merged.get(a)}
	);

	for (var x=0; x < 2 && x < top_ports_all.length; ++x) {
		top_ports.add(top_ports_all[x]);
	}

	debug.add("port: slot="+timeslot_ports+", unique dest ports: "+port_dest_count_merged.length);
		//", top 3 ports="+top_ports_all[0]+","+top_ports_all[1]+","+top_ports_all[2]);



	if (port_dest_count_merged.length > THRESHOLD_PORT_DEST_ATTACKED) {
		//debug.add("PRA-WARNING: too many unique destination ports: " +
		//" ("+ port_dest_count_merged.length +" > "+ THRESHOLD_PORT_DEST_ATTACKED +")");
		sampling = ((THRESHOLD_PORT_DEST_ATTACKED / port_dest_count_merged.length) * 100)
		alarm_portdest_triggered = true;
	}
	/**/
	// amount of attacks for one monitor {id : count}
	var attacks_per_monitor = new Map();
	var attacks_total = 0;
	//debug.add("counting monitors...");
	var skipped_top_ports = 0;

	/*
	// TODO: activate ratio mitigation if needed
	for (var slot_search in window_ip_source_dest) {
		for (var ipsrc_ipdst_portdst in window_ip_source_dest[slot_search]) {
			ipsrc_ipdst_portdst_split = ipsrc_ipdst_portdst.split(",");

			// TODO: activate for port normalizing (optional)
			//if (top_ports.contains(ipsrc_ipdst_portdst_split[2])) {
			//	// skip top x ports
			//	//debug.add("skipping port: "+ipsrc_ipdst_portdst_split[2]);
			//	skipped_top_ports += 1;
			//	continue;
			//}
			ip_dst = ipsrc_ipdst_portdst_split[1];
			count = window_ip_source_dest[slot_search][ipsrc_ipdst_portdst];

			//if (count > 100) {debug.add(ip_dst+"="+count);}
			if (attacks_per_monitor.get(ip_dst) == undefined) {
				//debug.add(ip_dst);
				attacks_per_monitor.set(ip_dst, count);
			} else {
				attacks_per_monitor.set(ip_dst, attacks_per_monitor.get(ip_dst) + count);
			}
			attacks_total += count;
		}
	}
	*/
	//debug.add("output...");
	// >>> check treshold:
	// ratio -> attacks / unique monitors
	var unique_monitors_total = attacks_per_monitor.length;
	var ratio = attacks_total / unique_monitors_total;
	var alarm_ratio_triggered = false;

	/**/
	if (isNaN(ratio)) {
		// ratio 0 means: no attacks and/or no monitors -> nothing happened -> no attack
		ratio = 0;
	}

	/*
	if (is_active) {
		debug.add("mon: slot="+timeslot_ratio+", attacks="+attacks_total+", mon="+unique_monitors_total+
		", ratio="+ratio+", top removed="+skipped_top_ports);
	}*/

	if (ratio != 0 && ratio <= THRESHOLD_ATTACKS_MONITOR_RATIO) {
		//debug.add("PRA-WARNING: Ratio attacks/monitors is too low: "+ratio+" <= "+THRESHOLD_ATTACKS_MONITOR_RATIO);
		// ratio is at minimum 1, so minimum is: 1/(threshold_ratio) -> set minimum to 0/(threshold_ratio-1)
		sampling_ratio = (((ratio-1)/(THRESHOLD_ATTACKS_MONITOR_RATIO-1)) * 100);

		if (sampling_ratio < sampling)  {
			sampling = sampling_ratio;
		}
		alarm_ratio_triggered = true;
	}
	/**/


	alarm_attacks_per_monitor_triggered = false;

	/*
	// >>> check threshold: attacks per monitor
	attacks_per_monitor.forEach(function each(amount, monitor, mapObj) {
		//debug.add(monitor+"="+amount);
		if (amount > THRESHOLD_ATTACKS_PER_MONITOR) {
			//debug.add("PRA-WARNING: Monitor "+monitor+" is attacked too many times("+
			//amount+" > "+THRESHOLD_ATTACKS_PER_MONITOR+")");
			sampling_attack_per_mon = ((THRESHOLD_ATTACKS_PER_MONITOR/amount) * 100);
			if (sampling_attack_per_mon < sampling)  {
				sampling = sampling_attack_per_mon;
			}
			alarm_attacks_per_monitor_triggered = true;
		}
	});
	*/

	if (is_active) {
		sampling_percent = sampling;

		if ((!alarm_ratio_triggered) && (!alarm_portdest_triggered) && (!alarm_attacks_per_monitor_triggered)) {
			sampling_percent = 100;
			sampling_active = false;
		} else if (sampling_percent < 100) {
			debug.add("sampling="+sampling_percent+"%, ratio="+alarm_ratio_triggered+
			", port_dest="+alarm_portdest_triggered+", atttacks_per_mon="+alarm_attacks_per_monitor_triggered+
			", noise added/total="+events_dshield_added+"/"+events_dshield_total);
			sampling_active = true;
		}
	} else {
		ratio = 0;
	}
	// unban IPs which ban timeout exceeded
	/*
	for (var ip in banned_ips) {
		if ( (time_minutes_now - banned_ips[ip]) >= SOURCE_IP_BAN_MINUTES) {
			debug.add("unbanning ip: "+ ip +" ("+ (time_minutes_now - banned_ips[ip]) +" minutes passed)");
			delete banned_ips[ip];
		}
	}
	*/
	fs.appendFile(FILE_STATS, ratio+"\t"+port_dest_count_merged.length+"\t"+alarm_ratio_triggered+"\t"
		+alarm_portdest_triggered+"\t"+events_dshield_added+"\t"
		+events_dshield_total+"\t"+events_external_added+"\t"+events_external_total+"\n", function (err) {});

	//debug.add("finished checking");
}

// this is exclusively used by addNoiseEvent()
var data = {
	sensorname: null,
	sensortype: "RandomType",
	src: {
		ip: null,
		port: null,
		ll: [0,0],
		country: "",
		cc: 0,
		city: ""
	},
	dst: {
		ip: null,
		port: null,
		ll: [0,0],
		country: "",
		cc: 0,
		city: ""
	},
	type: 11,
	log: null,
	md5sum: "0123456789",
	date: new Date(),
	authorized: false
};

function addNoiseEvent(ip_src, port_src, ip_dst, port_dst, monitorname, doEmit) {
	if (typeof io == 'undefined') {
		//debug.add("no io, no emit -> I quit");
		return;
	}

	//debug.add("adding, name/ip/port: " + data.sensorname + " / " + data.src.ip + " / " +
	//	data.dst.ip + " / " + data.src.port + " / " + data.dst.port);
	/*
	// not needed for PRA
	ipa = lookUpIp(ip_src);

	if(ipa){
		data.src["ll"] = [ipa.longitude, ipa.latitude];
		data.src["country"] = ipa.country_name;
		data.src["cc"] = ipa.country_code;
		data.src["city"] = ipa.city || "";
	}

	ipd = lookUpIp(ip_dst);

	if(ipd){
		data.dst["ll"] = [ipd.longitude, ipd.latitude];
		data.dst["country"] = ipd.country_name;
		data.dst["cc"] = ipd.country_code;
		data.dst["city"] = ipd.city || "";
	}
	*/
	//debug.add("> " + chance.integer({min: 0, max: 65535}));
	//debug.add("> " + chance.ip());
	//debug.add("Sending random entry: " + data["src"]["ip"]);

	update_slot(monitorname, ip_src, ip_dst, port_dst);

	// do emit data to client if sampled
	if (doEmit) {
		data.src.ip = ip_src;
		data.src.port = port_src;
		data.dst.ip = ip_dst;
		data.dst.port = port_dst;
		data.sensorname = monitorname;
		io.sockets.emit("markIncident", data);
	}
}

// add artificial noise to the report
function add_entry_random() {
	addNoiseEvent(chance.ip(),
		chance.integer({min: 0, max: 65535}),
		chance.ip(),
		chance.integer({min: 0, max: 65535}),
		"RandomMonitor_" + chance.integer({min: 0, max: 100}),
		true
	);
}

function add_entry_dshield() {
	e = dshield_entries.pop();
	//e = dshield_entries.shift();

	//debug.add("adding noise using dshield data: "+e);
	// we can not differentiate between PRA and non-PRA events so we have to sample both
	// accept entry if: sampling not active (sample 100%) or random value tells us so
	// add noise if: attack mitigation is not active, mitigation not triggered or random tells us to
	sample_entry = (!is_active || !sampling_active || chance.floating({min: 1, max: 100}) <= sampling_percent);

	try {
		addNoiseEvent(e[0], e[1], e[2], e[3], e[4], sample_entry);
		if (sample_entry) {
			events_dshield_added += 1;
		}
		events_dshield_total += 1;
	}
	catch (err) {
	// no events left
	}
}

function applyMitigation(ip_src, ip_dst, port_dst, monitor_id) {
	// praMitigation.applyMitigation(items[i].src.ip, items[i].dst.ip, items[i].dst.port
	events_external_total += 1;

	if (!is_active) {
		return false;
	} else {
		update_slot(monitor_id, ip_src, ip_dst, port_dst);

		// allow entry if sampling is not active or random tells us to
		is_sampled = (!sampling_active || chance.floating({min: 1, max: 100}) > sampling_percent);

		if (is_sampled) {
			events_external_added += 1;
		}

		return !is_sampled;
	}
}

function setActive(activeState) {
	debug.add("setting new PRA mitigation active state: " + activeState);
	is_active = activeState;
}

module.exports = {
	//init: init,
	setActive: setActive,
	applyMitigation: applyMitigation
};