evaluator.py

"""
The Evaluator is charged with evaluating a given strategy and assigning a numerical fitness metric to it.
"""
import argparse
import copy
import logging
import multiprocessing
import os
import random
import string
import socket
import subprocess
import sys
import threading
import time
import re
import warnings
#import ast
#import pandas as pd
import numpy as np
import joblib
#from scapy.all import sniff

import requests
import urllib3

import actions.utils
import actions.strategy
#import actions.sniffer
import censors.censor_driver



# Suppress unfixed Paramiko warnings (see Paramiko issue #1386)
warnings.filterwarnings(action='ignore',module='.*paramiko.*')

# Placeholder for a docker import (see below why we cannot import docker here)
docker = None
BASEPATH = os.path.dirname(os.path.abspath(__file__))
PROJECT_ROOT = BASEPATH


class Evaluator():
    def __init__(self, command, logger):
        """
        Initialize the global evaluator for this evolution.

        Args:
            command (list): sys.argv or list of arguments
            logger (:obj:`logging.Logger`): logger passed in from the main driver to log from
        """
        self.args = get_args(command)
        self.test_plugin = self.args["test_type"]
        assert self.test_plugin, "Cannot import an empty plugin"

        self.public_ip = self.args.get("public_ip", "")
        
        
        
        self.external_client = self.args["external_client"]
        self.censor = self.args.get("censor")
        # If there is no external client defined and no internal test setup, default --external-server to True
        if not self.external_client and not self.censor:
            self.args["external_server"] = True
            s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            s.connect(('10.0.0.0', 0)) 
            self.client_ip = s.getsockname()[0]
        self.external_server = self.args["external_server"]

        # If there is an external client connecting to us, override the server with our public ip
        if not self.external_server and self.external_client:
            assert self.args.get("public_ip", ""), "Cannot use an external client to this server without specifying the public IP."
            self.public_ip = self.args.get("public_ip", "")
            worker = actions.utils.get_worker(self.public_ip, logger)
            if worker:
                self.public_ip = worker["ip"]
            self.args.update({'server': self.public_ip})
            command += ["--server", self.public_ip]

        self.run_canary_phase = True

        self.client_args = copy.deepcopy(self.args)
        self.server_args = copy.deepcopy(self.args)

        self.client_cls = None
        self.server_cls = None
        self.plugin = None

        self.override_evaluation = False

        # Plugin may optionally override the strategy evaluation for a single individual or the entire evaluation
        try:
            _, plugin_cls = actions.utils.import_plugin(self.test_plugin, "plugin")
            parsed_args = plugin_cls.get_args(command)
            self.args.update({k:v for k,v in parsed_args.items() if v or (not v and k not in self.args)})
            self.plugin = plugin_cls(self.args)
            # Disable the canary phase if the plugin will override the default evaluation logic
            self.run_canary_phase = not self.plugin.override_evaluation
            self.override_evaluation = self.plugin.override_evaluation
        except ImportError:
            pass

        self.client_cls = collect_plugin(self.test_plugin, "client", command, self.args, self.client_args)
        #print(type(collect_plugin))
        self.server_cls = collect_plugin(self.test_plugin, "server", command, self.args, self.server_args)

        self.workers = self.args["workers"]
        self.stop = False
        self.skip_empty = not self.args["no_skip_empty"]
        self.output_directory = self.args["output_directory"]

        self.routing_ip = self.args.get("routing_ip", None)
        self.iface = self.args.get("iface", None)
        self.runs = self.args.get("runs", 1)
        self.trw = self.args["real_time_detection"]
        self.Geneva = self.args["Geneva"]
        if self.Geneva==False:
            self.alpha =0.5
            self.beta=0.5
        else:
            self.alpha =0
            self.beta=1
        print('TRW evaluation is :',self.trw)
        self.fitness_by = self.args.get("fitness_by", "avg")

        self.forwarder = {}
        # If NAT options were specified to train as a middle box, set up the engine's
        # NAT configuration
        self.act_as_middlebox = self.args.get("act_as_middlebox")
        if self.act_as_middlebox:
            assert self.args.get("forward_ip")
            assert self.args.get("sender_ip")
            assert self.args.get("routing_ip")
            self.forwarder["forward_ip"] = self.args["forward_ip"]
            self.forwarder["sender_ip"] = self.args["sender_ip"]
            self.forwarder["routing_ip"] = self.args["routing_ip"]

        # Legacy environments storage
        self.environments = []
        if not os.path.exists(self.output_directory):
            os.mkdir(self.output_directory)

        # Only enable docker if we're going to use an internal censor
        self.use_docker = False
        if self.args["censor"]:
            import docker
            self.use_docker = True
            self.docker_client = docker.from_env()
            self.apiclient = docker.APIClient()

        self.logger = logger
        self.local_model_namee = self.args.get("local_model")
        self.censor_model_namee = self.args.get("censor_model")
        #self.detector = joblib.load(os.path.join("ML detectors","rfc.joblib"),mmap_mode=None) ## to update later on according to the user's model
        self.train_detector = joblib.load(os.path.join("ML detectors",self.local_model_namee),mmap_mode=None)
        self.eval_detector = joblib.load(os.path.join("ML detectors",self.censor_model_namee),mmap_mode=None)
        self.detection_count = 0
        self.detectability=[]
        self.y_preds = []
        self.jump = self.args["jump"]
        self.random_jump = self.args["random_jump"]
       
        if self.random_jump:
            self.max_jump=self.jump
        #self.alpha = 0.5
        #self.beta = 0.5
        if self.trw:
            self.discarded=0
            self.to_discard=0
            self.eta1 = 0.99/0.01
            self.eta0 = (1-0.99)/(1-0.01)
            self.trw_ratio = (self.eta0+self.eta1)/2 # likelyhood ratio
            self.ratio_all=[]
            # theta0 and theta1 are defined with respect to the confusion matrix of the pre-trained ML model
            self.theta0 = 3146/(3146+10)
            self.theta1 = 10/(3146+9)  
            self.history = 0

    def evaluate(self, ind_list):
        """
        Perform the overall fitness evaluation driving.

        Args:
            ind_list (list): list of individuals to evaluate

        Returns:
            list: Population list after evaluation
            list: Detection probabilities
        """
        # Setup environment ids for each individual
        self.assign_ids(ind_list)
        self.detectability=[]
        

        # If the plugin has overridden default evaluation, call that here
        if self.override_evaluation:
            self.logger.debug("Beginning evaluation in plugin")
            return self.plugin.evaluate(self.args, self, ind_list, self.logger)

        if self.workers > 1 and self.use_docker:
            # Chunk the population and test sites into smaller lists to hand to each worker
            split = [ind_list[i::self.workers] for i in range(0, self.workers)]

            procs = []
            # Create workers
            for i in range(0, len(split)):
                if not split[i]:
                    continue
                if self.use_docker:
                    try:
                        # Due to limitations in docker-py, it is not safe to build the containers in a multiprocessed
                        # setting. To handle this, build the environments ahead of time, and pass them to the workers to use.
                        environment = self.create_test_environment(i)
                    except (docker.errors.APIError, requests.exceptions.ConnectionError, urllib3.exceptions.ProtocolError):
                        self.logger.exception("Failed to create evaluator environment - is docker running?")
                        return

                proc = multiprocessing.Process(target=self.worker, args=(split[i], str(i), environment))
                proc.start()
                procs.append(proc)

            try:
                # Join all the processes
                for proc in procs:
                    proc.join()
            except KeyboardInterrupt:
                self.shutdown()
        else:
            environment = {}
            if self.use_docker:
                try:
                    print('Creating a test envirnment\n\n')
                    environment = self.create_test_environment("main")
                except (docker.errors.APIError, requests.exceptions.ConnectionError, urllib3.exceptions.ProtocolError):
                    self.logger.exception("Failed to create evaluator environment - is docker running?")
                    return

            self.worker(ind_list, "main", environment)

        for ind in ind_list:
            self.read_fitness(ind)
        self.terminate_docker()
        
        

        return ind_list#, self.detectability

    def run_test(self, environment, ind, benign=0):
        """
        Conducts a test of a given individual in the environment.

        Args:
            environment (dict): Dictionary of environment variables
            ind (:obj:`actions.strategy.Strategy`): A strategy object to test with

        Returns:
            tuple: (ind.environment_id, ind.fitness) environment ID of strategy and fitness
        """
        # If skip_empty is enabled, this is not the canary, and the individual is empty,
        # skip it
        if len(ind) == 0 and ind.environment_id != "canary" and ind.environment_id.find('normal') == -1 and self.skip_empty:
            self.logger.info("[skipped] Fitness %d: %s" % (-1000, str(ind)))
            ind.fitness = -1000
            return "skipped", -1000

        fitnesses = []
        
        
        
        
        # Run the strategy multiple times if requested
        for run in range(0, self.runs):
         
            self.logger.debug("Launching %s plugin (run %d) for %s" % (self.test_plugin, run + 1, str(ind)))

            environment["id"] = ind.environment_id
            self.client_args.update({"environment_id": ind.environment_id})
            self.server_args.update({"environment_id": ind.environment_id})

            if not self.args["server_side"]:
                self.client_args.update({"strategy" : str(ind)})
                self.server_args.update({"no_engine" : True})
            else:
                self.server_args.update({"strategy" : str(ind)})
                self.client_args.update({"no_engine" : True})

            # If we're using an internal censor, make sure the client is pointed at the server
            if self.args["censor"]:
                self.client_args.update({"server": environment["server"]["ip"]})
                self.client_args.update({"wait_for_censor": True})
                self.server_args.update({"wait_for_shutdown": True})
                self.update_ports(environment)

            try:
                # If the plugin has overridden the below logic, run that plugin's version directly
                if self.plugin:
                    self.logger.debug("Running standalone plugin.")
                    self.args.update({"strategy": str(ind)})
                    self.plugin.start(self.args, self, environment, ind, self.logger)
                    self.read_fitness(ind)
                else:
                    self.logger.debug("Launching client and server directly.")
                    # If we're given a server to start, start it now
                    if self.server_cls and not self.external_server and not self.act_as_middlebox:
                        server = self.start_server(self.server_args, environment, self.logger)

                    fitness = self.run_client(self.client_args, environment, self.logger)

                    if self.server_cls and not self.external_server and not self.act_as_middlebox:
                        self.stop_server(environment, server)

                    self.read_fitness(ind)

                    # If the engine ran on the server side, ask that it punish fitness
                    if self.args["server_side"]:
                        ind.fitness = server.punish_fitness(ind.fitness, self.logger)
                        actions.utils.write_fitness(ind.fitness, self.output_directory, environment["id"])
            except actions.utils.SkipStrategyException as exc:
                self.logger.debug("Strategy evaluation ending.")
                ind.fitness = exc.fitness
                fitnesses.append(ind.fitness)
                break
        
            ### Added code ###
            print('Run {} is completed'.format(run))
            ## Reading pcap tracks to check detection
            df = actions.utils.parse_pcap(ind.environment_id,self.args,self.client_ip)
            if df is not None and '{}' not in df['flags'] and "{'S':1}" not in df['flags']:
                print(df)
                # define flags to check for according to pretrained ML detector
                new_cols = ['A', 'PA', 'SA', 'FA', 'RA', 'FPA', 'S', 'R', 'U', 'SEC', 'SRPUEC', 'FSRPAUE', 'FPCN', 'P', 'SAE', 'SPUE', 'F', 'RUE', 'FPUC', 'FUCN', 'SUC', 'FSRPAC', 'FSAN', 'SRAECN', 'AEC', 'FSRPEC', 'FUECN', 'FRA', 'PAU', 'SRACN', 'SE', 'SCN', 'FRPAUE', 'FSRPAEN', 'FRPEN', 'FSREC', 'SRPAUC', 'FRAECN', 'FSRUECN', 'SAC', 'RPAU', 'FPAE', 'SPAUECN', 'FRAUEN', 'FRAUN', 'PUN', 'FRPACN', 'SRPAUE', 'FSUE', 'FSU', 'SP', 'RPA', 'FSPAUECN', 'SPC', 'SRPACN', 'RPAUECN', 'SU', 'FSRAE', 'FRPAU', 'RAECN', 'FPEN', 'SRPECN', 'SPUN', 'SRAUE', 'FPUECN', 'RPAE', 'SUEC', 'RPUN', 'FPAUEN', 'RAUE', 'FR', 'SRUCN', 'FSPAE', 'FSRAECN', 'SR', 'FSPEN', 'FSRPUEN', 'FAE', 'SRAUECN', 'SPEN', 'FRUC', 'FRAN', 'FSPE', 'FSPUEC', 'SRPAN', 'FUEN', 'FRPAECN', 'SRPUCN', 'SRAE', 'FSPAN', 'SRA', 'PUE', 'FSUN', 'RAE', 'CN', 'FRPAC', 'SRAC', 'RN', 'SRPAUECN', 'FSRC', 'FRAEC', 'FSPUEN', 'SRUE', 'FSUC', 'SRPUECN', 'FRAE', 'FUN', 'FRPAUEC', 'FSPA', 'SRPUC', 'SAUE', 'RPAUE', 'FRPAEC', 'AUECN', 'SPAEN', 'RPU', 'FSPUECN', 'SRPEC', 'FSRPAUN', 'FAEN', 'AU', 'FSRPECN', 'PUC', 'FPAUN', 'FRUEN', 'FSPAUEC', 'SPUCN', 'SRPAEC', 'FUEC', 'FRUN', 'SEN', 'FSPAUE', 'FSRPAU', 'RPEN', 'FRPAUECN', 'FSAUCN', 'FRPUEC', 'EC', 'FRUCN', 'RAUECN', 'SUECN', 'PUCN', 'FPAEN', 'FSRAC', 'PAUN', 'FPACN', 'FRAUEC', 'RUEN', 'FAUECN', 'FSRAUEC', 'FPUEC', 'SRAUEN', 'FSUEC', 'FAUN', 'SPE', 'FRPN', 'RPECN', 'RAN', 'SACN', 'FSRUC', 'FS', 'FRPA', 'SPUC', 'SRPUN', 'RP', 'FSRPN', 'SPAN', 'SRAUC', 'PCN', 'RUECN', 'SPN']
                for col in new_cols:
                    values = []
                    for elt in df['flags']:
                        #dictt = ast.literal_eval(elt)
                        dictt = elt
                        if col in dictt.keys():
                            values.append(dictt[col])
                        else:
                            values.append(0)
                    df[col] = values
    
    
                del df['flags']
                X = df.values
                
                # Check if strategy is detectable
                train_y_prob = self.train_detector.predict_proba(X)
                # Compute the avg probability that the source is running Geneva
                P_g = np.mean(train_y_prob[:,0])
                print('\n\n train: ',P_g)
                
                # get evaluation detection
                eval_y_prob = self.eval_detector.predict_proba(X)
                print('\n\n Eval: ',np.mean(eval_y_prob[:,0]))
                #print('\n\n',y_prob)
                
                y_pred = [y.argmax() for y in eval_y_prob]
                self.y_preds.append(y_pred[0])
                # update detection count
                for y in y_pred:
                    if y==0: # Geneva detected
                        self.detection_count+=1
                print('\n\n',y_pred)
                print('Evaluation Predictions so far: ',self.y_preds)
                if self.trw:
                    print('TRW history patamater ',self.history)
                    print('Ratio changes so far: ',self.ratio_all)
                print('Number of Evaluation detections so far: ',self.detection_count)
                
                # capturing training detectability to check for pauses
                for prob in train_y_prob[:,0]:
                    self.detectability.append(prob)
                
                
                ### TRW code for real-time evaluation
                if self.trw:
                    #if self.discarded==self.to_discard:
                    print("\nUpdating TRW ratio...\n")
                    # attempting to discard estimated normal traffic
                    ## update the TRW ratio
                   
                    y=self.y_preds[-1]
                    if y==0: # predicted as geneva traffic
                        self.trw_ratio = self.trw_ratio*(self.theta1/self.theta0)
                        self.history+=1
                        self.ratio_all.append(self.trw_ratio)
                    if y==1: # predicted as legitimate traffic
                        self.trw_ratio = self.trw_ratio*((1-self.theta1)/(1-self.theta0))
                        self.ratio_all.append(self.trw_ratio)
                    # Checking for final decision
                    if self.trw_ratio <= self.eta0:
                        raise Exception('Geneva is detected with an accuracy of {} after {} observations. Check ML detection history: {} \n {}'.format(0.99,len(self.y_preds),self.y_preds,self.ratio_all))
                    elif self.trw_ratio >= self.eta1:
                        # reset ratio for currently benign sources
                        #self.trw_ratio = (self.eta0+self.eta1)/2
                        #'''
                        if self.history == 0:
                            self.trw_ratio  = (self.eta0+self.eta1)/2 # averaging both thresholds
                        else:
                            self.trw_ratio = (self.history/len(self.y_preds)) * self.eta0 + (1-(self.history/len(self.y_preds))) * self.eta1
                            if self.trw_ratio<= self.eta0:
                                print('Reset caused an automatic detection')
                            
                   # else:
                   #     self.discarded+=1
                   #     print("\n{} flows are discarded by TRW. {} discards are remaining\n".format(self.discarded,self.to_discard-self.discarded))
        
                #####
                detection_rate = self.detection_count/len(self.y_preds)
                
                print('Detection rate :',detection_rate)
                
                # update fitness value as f = beta. g - alpha. P_g
                print('alpha=',self.alpha)
                ind.fitness = self.beta*ind.fitness - self.alpha * P_g*1000# *800
                
            else:
                print('No traffic traces are found, testing the strategy again!')
                
        
        
        fitnesses.append(ind.fitness)
        

        if self.runs > 1:
            self.logger.debug("\t(%d/%d) Fitness %s: %s" % (run + 1, self.runs, str(ind.fitness), str(ind)))
        
        
        
        if ind.environment_id.find('normal') == -1:
            
            # We sore only relevent evaluations
            self.logger.debug("Storing fitness of %s by: %s" % (fitnesses, self.fitness_by))
            if self.fitness_by == "min":
                ind.fitness = min(fitnesses)
            elif self.fitness_by == "max":
                ind.fitness = max(fitnesses)
            elif self.fitness_by == "avg":
                ind.fitness = round(sum(fitnesses)/len(fitnesses), 2)
            actions.utils.write_fitness(ind.fitness, self.output_directory, environment["id"])
    
            # Log the fitness
            self.logger.info("[%s] Fitness %s: %s" % (ind.environment_id, str(ind.fitness), str(ind)))
        
        
        
        return ind.environment_id, ind.fitness

    def run_client(self, args, environment, logger, benign=0):
        """
        Runs the plugin client given the current configuration

        Args:
            args (dict): Dictionary of arguments
            environment (dict): Dictionary describing environment configuration for this evaluation
            logger (:obj:`logging.Logger`): A logger to log with

        Returns:
            float: Fitness of individual
        """
        fitness = None
        if environment.get("remote"):
            fitness = self.run_remote_client(args, environment, logger)
        elif environment.get("docker"):
            self.run_docker_client(args, environment, logger)
        else:
            self.run_local_client(args, environment, logger)

        fitpath = os.path.join(BASEPATH, self.output_directory, actions.utils.FLAGFOLDER, environment["id"]) + ".fitness"
        # Do not overwrite the fitness if it already exists
        if not os.path.exists(fitpath):
            actions.utils.write_fitness(fitness, self.output_directory, environment["id"])
        return fitness

    def run_docker_client(self, args, environment, logger, benign=0):
        """
        Runs client within the docker container. Does not return fitness; instead
        fitness is written via the flags directory and read back in later.

        Args:
            args (dict): Dictionary of arguments
            environment (dict): Dictionary describing environment configuration for this evaluation
            logger (:obj:`logging.Logger`): A logger to log with
        """
        command = ["docker", "exec", "--privileged", environment["client"]["container"].name, "python", "code/plugins/plugin_client.py", "--server", environment["server"]["ip"]]
        base_cmd = actions.utils.build_command(args)
        command += base_cmd
        self.exec_cmd(command)
        
     
    def run_benign_traffic(self, args, environment, logger):
        """
        Runs client within the docker container with benign traffic. 

        Args:
            args (dict): Dictionary of arguments
            environment (dict): Dictionary describing environment configuration for this evaluation
            logger (:obj:`logging.Logger`): A logger to log with
        """
        
        command = ["docker", "exec", "--privileged","--detach",environment["benign_server"]["container"].name, "code/docker/harpoon/src/harpoon", "-f", "code/docker/harpoon/examples/docker_server.xml", "-c"]
        self.exec_cmd(command)
        
        command = ["docker", "exec", "--privileged","--detach",environment["client"]["container"].name, "code/docker/harpoon/src/harpoon", "-f", "code/docker/harpoon/examples/docker_client.xml", "-c"]
        self.exec_cmd(command)
        
    
        
        
        
        

    def update_ports(self, environment):
        """
        Checks that the chosen port is open inside the docker container - if not, it chooses a new port.

        Args:
            environment (dict): Dictionary describing docker environment
        """
        command = ["docker", "exec", "--privileged", environment["server"]["container"].name, "netstat", "-ano"]
        output = self.exec_cmd_output(command)
        requested_port = self.args.get("port")
        self.logger.debug("Testing if port %s is open in the docker container" % requested_port)
        while (":%s" % requested_port) in output:
            self.logger.warn("Port %s is in use, choosing a new port" % requested_port)
            requested_port = random.randint(1000, 65000)
            output = self.exec_cmd_output(command)
        self.logger.debug("Using port %s" % requested_port)
        self.args.update({"port": requested_port})
        self.client_args.update({"port": requested_port})
        self.server_args.update({"port": requested_port})

    def run_remote_client(self, args, environment, logger):
        """
        Runs client remotely over SSH, using the given SSH channel

        Args:
            args (dict): Dictionary of arguments
            environment (dict): Dictionary describing environment configuration for this evaluation
            logger (:obj:`logging.Logger`): A logger to log with

        Returns:
            float: Fitness of individual
        """
        worker = environment["worker"]
        remote = environment["remote"]
        command = []
        if worker["username"] != "root":
            command = ["sudo"]
        command += [worker["python"], os.path.join(worker["geneva_path"], "plugins/plugin_client.py")]
        base_cmd = actions.utils.build_command(args)
        command += base_cmd
        command = " ".join(command)

        self.remote_exec_cmd(remote, command, logger, timeout=20)

        # Get the logs from the run
        self.get_log(remote, worker, "%s.client.log" % environment["id"], logger)
        if not args["server_side"]:
            self.get_log(remote, worker, "%s.engine.log" % environment["id"], logger)

        # Get the individual's fitness
        command = 'cat %s/%s/%s/%s.fitness' % (worker["geneva_path"], self.output_directory, actions.utils.FLAGFOLDER, environment["id"])
        remote_fitness, error_lines = self.remote_exec_cmd(remote, command, logger)
        fitness = -1000
        try:
            fitness = int(remote_fitness[0])
        except Exception:
            logger.exception("Failed to parse remote fitness.")
            return None

        return fitness

    def remote_exec_cmd(self, remote, command, logger, timeout=15, verbose=True):
        """
        Given a remote SSH session, executes a string command. Blocks until
        command completes, and returns the stdout and stderr. If the SSH
        connection is broken, it will try again.

        Args:
            remote: Paramiko SSH channel to execute commands over
            command (str): Command to execute
            logger (:obj:`logging.Logger`): A logger to log with
            timeout (int, optional): Timeout for the command
            verbose (bool, optional): Whether the output should be printed

        Returns:
            tuple: (stdout, stderr) of command, each is a list
        """
        i, max_tries = 0, 3
        lines = []
        error_lines = []
        stdin_, stdout_, stderr_ = None, None, None
        while i < max_tries:
            try:
                if verbose:
                    logger.debug(command)
                stdin_, stdout_, stderr_ = remote.exec_command(command, timeout=timeout)
                # Block until the client finishes
                stdout_.channel.recv_exit_status()
                error_lines = stderr_.readlines()
                lines = stdout_.readlines()
                break
            # We would like to catch paramiko.SSHException here, but because of issues with importing paramiko
            # at the top of the file in the main namespace, we catch Exception instead as a broader exception.
            except Exception:
                logger.error("Failed to execute \"%s\" on remote host. Re-creating SSH tunnel." % command)
                # Note that at this point, our remote object still has a valid channel as far as paramiko is
                # concerned, but the channel is no longer responding. If we tried to do remote.close() here,
                # it would hang our process. Instead, we'll set up a new remote channel, and let Python's garbage
                # collection handle destroying the original remote object for us.
                try:
                    remote = self.setup_remote()
                except Exception:
                    logger.error("Failed to re-connect remote - trying again.")
                i += 1

        if verbose:
            for line in error_lines:
                logger.debug("ERROR: %s", line.strip())
        # Close the channels
        if stdin_:
            stdin_.close()
        if stdout_:
            stdout_.close()
        if stderr_:
            stderr_.close()
        return lines, error_lines

    def get_log(self, remote, worker, log_name, logger):
        """
        Retrieves a log from the remote server and writes it to disk.

        Args:
            remote: A Paramiko SSH channel to execute over
            worker (dict): Dictionary describing external client worker
            log_name (str): Log name to retrieve
            logger (:obj:`logging.Logger`): A logger to log with
        """
        # Get the client.log
        log_path = os.path.join(self.output_directory, "logs", log_name)
        command = "cat %s" % os.path.join(worker["geneva_path"], log_path)
        client_log, error_lines = self.remote_exec_cmd(remote, command, logger, verbose=False)
        # If something goes wrong, we don't necessarily want to dump the entire client_log to the screen
        # a second time, so just disable verbosity and display the stderr.
        for line in error_lines:
            logger.error(line.strip())

        # Write the client log out to disk
        with open(log_path, "w") as fd:
            for line in client_log:
                fd.write(line)

    def run_local_client(self, args, environment, logger):
        """
        Runs client locally. Does not return fitness.

        Args:
            args (dict): Dictionary of arguments
            environment (dict): Dictionary describing environment configuration for this evaluation
            logger (:obj:`logging.Logger`): A logger to log with
        """
        # Launch the plugin client
        command = [sys.executable, "plugins/plugin_client.py", "--plugin", self.client_cls.name]
        base_cmd = actions.utils.build_command(args)
        command += base_cmd
        # Replace strings of empty strings "''" with empty strings "", as subprocess will handle this correctly
        command = [x if x != "''" else "" for x in command]
        logger.debug(" ".join(command))
        self.exec_cmd(command)

    def exec_cmd(self, command, timeout=60):
        """
        Runs a subprocess command at the correct log level.

        Args:
            command (list): Command to execute.
            timeout (int, optional): Timeout for execution
        """
        self.logger.debug(" ".join(command))
        try:
            if actions.utils.get_console_log_level() == "debug":
                subprocess.check_call(command, timeout=60)
            else:
                subprocess.check_call(command, stderr=subprocess.DEVNULL, stdout=subprocess.DEVNULL, timeout=60)
        except subprocess.CalledProcessError as exc:
            # Code 137 is for SIGKILL, which is how docker containers are shutdown by the evaluator.
            # Ignore these exceptions, raise all others
            if exc.returncode != 137:
                raise

    def exec_cmd_output(self, command, timeout=60):
        """
        Runs a subprocess command at the correct log level. This is a separate method from the above exec_cmd,
        since that is used to stream output to the screen (so check_output is not appropriate).

        Args:
            command (list): Command to execute.
            timeout (int, optional): Timeout for execution

        Returns:
            str: Output of command
        """
        self.logger.debug(" ".join(command))
        output = ""
        try:
            output = subprocess.check_output(command, timeout=60, stderr=subprocess.PIPE).decode('utf-8', 'ignore')
            if actions.utils.get_console_log_level() == "debug":
                self.logger.debug(output)
        except subprocess.CalledProcessError as exc:
            # Code 137 is for SIGKILL, which is how docker containers are shutdown by the evaluator.
            # Ignore these exceptions, raise all others
            if exc.returncode != 137:
                raise
        return output

    def start_server(self, args, environment, logger):
        """
        Launches the server.

        Args:
            args (dict): Dictionary of arguments
            environment (dict): Dictionary describing environment configuration for this evaluation
            logger (:obj:`logging.Logger`): A logger to log with

        Return:
            float: fitness of individual (if one is provided)
        """
        if environment.get("docker"):
            logger.debug("Evaluator: running server inside docker")
            return self.run_docker_server(args, environment, logger)
        else:
            logger.debug("Evaluator: running server")
            return self.run_local_server(args, environment, logger)

    def run_docker_server(self, args, environment, logger):
        """
        Runs server and censor in their respective docker containers.

        Args:
            args (dict): Dictionary of arguments
            environment (dict): Dictionary describing environment configuration for this evaluation
            logger (:obj:`logging.Logger`): A logger to log with
        """
        command = ["docker", "exec", "--privileged", environment["server"]["container"].name, "python", "code/plugins/plugin_server.py", "--test-type", self.server_cls.name]
        base_cmd = actions.utils.build_command(args)
        command += base_cmd
        # Replace strings of empty strings "''" with empty strings "", as subprocess will handle this correctly
        command = [x if x != "''" else "" for x in command]

        port = args.get("port")
        queue_num = random.randint(1, 1000)
        environment["port"] = port
        environment["queue_num"] = queue_num
        server_thread = threading.Thread(target=self.exec_cmd, args=(command, ))
        censor_thread = threading.Thread(target=self.start_censor, args=(environment, environment["id"]))
        censor_thread.start()
        server_thread.start()
        max_wait = 30
        count = 0
        flag_file = os.path.join(args["output_directory"], "flags", "%s.server_ready" % args["environment_id"])

        while count < max_wait:
            if os.path.exists(flag_file):
                break
            if count % 15 == 0:
                logger.debug("Evaluator waiting for confirmation of server startup")
            count += 1
            time.sleep(0.5)
        else:
            logger.warn("Evaluator: Server did not startup within window")
            return
        logger.debug("Evaluator: Server ready.")

    def stop_server(self, environment, server):
        """
        Stops server.

        Args:
            environment (dict): Environment dictionary
            server (:obj:`plugins.plugin_server.ServerPlugin`): A plugin server to stop
        """
        # If the server is running inside a docker container, we don't have access to it directly
        # to shut it down. Instead, write a shutdown flag to instruct it to shut down.
        self.logger.debug("Evaluator shutting down server.")
        if environment.get("docker"):
            flag_file = os.path.join(self.args["output_directory"], "flags", "%s.server_shutdown" % self.server_args["environment_id"])
            # Touch shutdown file to instruct the server to shutdown
            open(flag_file, 'a').close()
            self.stop_censor(environment)
        else:
            # Shut down the server
            server.stop()
            # Shut down the server's logger, now that we are done with it
            actions.utils.close_logger(environment["server_logger"])

    def run_local_server(self, args, environment, logger):
        """
        Runs local server.

        Args:
            args (dict): Dictionary of arguments
            environment (dict): Dictionary describing environment configuration for this evaluation
            logger (:obj:`logging.Logger`): A logger to log with
        """
        server = self.server_cls(args)
        logger.debug("Starting local server with args: %s" % str(args))
        server_logger = actions.utils.get_logger(PROJECT_ROOT, args["output_directory"], "server", "server", environment["id"], log_level=actions.utils.get_console_log_level())
        environment["server_logger"] = server_logger
        args.update({"test_type": self.server_cls.name})
        if not args.get("server_side"):
            args.update({"no_engine" : True})
        server.start(args, server_logger)
        return server

    def canary_phase(self, canary):
        """
        Learning phase runs the client against the censor to collect packets.

        Args:
            canary (:obj:`actions.strategy.Strategy`): A (usually empty) strategy object to evaluate

        Returns:
            str: canary id used ("canary")
        """
        if not self.run_canary_phase:
            return None

        self.logger.info("Starting collection phase")
        environment = {}
        canary.environment_id = "canary"
        if self.use_docker:
            try:
                environment = self.create_test_environment("canary")
            except (docker.errors.APIError, requests.exceptions.ConnectionError, urllib3.exceptions.ProtocolError):
                self.logger.error("Failed to create evaluator environment - is docker running?")
                return None

        self.worker([canary], canary.environment_id, environment)

        self.logger.info("Collected packets under %s" % canary)
        return "canary"
    
    
    

    def get_ip(self):
        """
        Gets IP of evaluator computer.

        Returns:
            str: Public IP provided
        """
        if self.public_ip:
            return self.public_ip
        return None

    def create_test_environment(self, worker_id):
        """
        Creates a test environment in docker.

        Args:
            worker_id (int): Worker ID of this worker

        Returns:
            dict: Environment dictionary to use
        """
        self.logger.debug("Initializing docker environment.")

        # We can't have an environment with an intenral test server and no censor
        # with the current set up. To be addressed later to allow for no censor testing
        assert not (not self.censor and not self.external_server), "Can't create internal server w/o censor"
        assert not (self.censor and self.external_server), "Can't create a censor without an internal training server"

        # Create a dict to hold the environment we're about to create
        environment = {}

        # Create the client container
        environment["client"] = self.initialize_base_container("client_%s" % worker_id)
        environment["client"]["ip"] = self.parse_ip(environment["client"]["container"], "eth0")
        self.client_ip = environment["client"]["ip"]

        # If a training censor is requested, create a censor container
        if self.censor:
            environment["censor"] = self.initialize_base_container("censor_%s" % worker_id)
            environment["server"] = self.initialize_base_container("server_%s" % worker_id)
            environment["benign_server"] = self.initialize_base_container("benign_server_%s" % worker_id)
            # Set up the routing
            environment["server"]["ip"] = self.parse_ip(environment["server"]["container"], "eth0")
            environment["benign_server"]["ip"] = self.parse_ip(environment["benign_server"]["container"], "eth0")
            environment["censor"]["ip"] = self.parse_ip(environment["censor"]["container"], "eth0")
            self._add_route(environment["server"]["container"], environment["censor"]["ip"])
            self._add_route(environment["client"]["container"], environment["censor"]["ip"])

            # Calculate the network base ("172.17.0.0")
            network_base = ".".join(environment["server"]["ip"].split(".")[:2]) + ".0.0"

            # Delete all other routes for the server and client to force communication through the censor
            environment["server"]["container"].exec_run(["route", "del", "-net", network_base, "gw", "0.0.0.0", "netmask", "255.255.0.0", "dev", "eth0"], privileged=True)
            environment["client"]["container"].exec_run(["route", "del", "-net", network_base, "gw", "0.0.0.0", "netmask", "255.255.0.0", "dev", "eth0"], privileged=True)

            # Set up NAT on the censor
            environment["censor"]["container"].exec_run(["iptables", "-t", "nat", "-A", "POSTROUTING", "-j", "MASQUERADE"], privileged=True)

        self.environments.append(environment)
        # Flag that this environment is a docker environment
        environment["docker"] = True
        
        ### Launch benign background traffic using harpoon
        if self.censor:
            self.run_benign_traffic(self.args, environment, self.logger)
        # Return the configured environment for use
        return environment

    def _add_route(self, container, via):
        """
        Helper method to take down an interface on a container

        Args:
            container: Docker container object to execute within
            via (str): IP address to route through
        """
        exit_code, _output = container.exec_run(["ip", "route", "del", "default"], privileged=True)
        exit_code, _output = container.exec_run(["ip", "route", "add", "default", "via", via], privileged=True)
        return exit_code

    def parse_ip(self, container, iface):
        """
        Helper method to parse an IP address from ifconfig.

        Args:
            container: Docker container object to execute within
            iface (str): Interface to parse from

        Returns:
            str: IP address
        """
        _exit_code, output = container.exec_run(["ifconfig", iface], privileged=True)
        #print(output)
        ip = re.findall(r'[0-9]+(?:\.[0-9]+){3}', output.decode("utf-8"))[0]
        return ip

    def setup_remote(self):
        """
        Opens an SSH tunnel to the remote client worker.
        """
        # Import paramiko here instead of at the top of the file. This is done intentionally. When
        # paramiko is imported, pynacl is loaded, which polls /dev/random for entropy to setup crypto
        # keys. However, if the evaluator is run on a relatively blank VM (or AWS instance) with little
        # network traffic before it starts (as will often be the case), there may be insufficient entropy
        # available in the system. This will cause pynacl to block on entropy, and since the only thing
        # running on the system is now blocking, it is liable to block indefinitely. Instead, the import
        # is performed here so that the system interaction of running the evaluator this far collects
        # enough entropy to not block paramiko. The pynacl team is aware of this issue: see issue #503
        # (https://github.com/pyca/pynacl/issues/503) and #327 (https://github.com/pyca/pynacl/issues/327)
        import paramiko
        paramiko_logger = paramiko.util.logging.getLogger()
        paramiko_logger.setLevel(logging.WARN)
        worker = actions.utils.get_worker(self.external_client, self.logger)

        if self.use_docker:
            worker["ip"] = "0.0.0.0"

        # Pull the destination to connect to this worker. Preference hostnames over IP addresses.
        destination = worker["hostname"]
        if not destination:
            destination = worker["ip"]

        self.logger.debug("Connecting to worker %s@%s" % (worker["username"], destination))
        remote = paramiko.SSHClient()
        remote.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        max_tries = 5
        i = 0
        while i < max_tries:
            try:
                if "keyfile" in worker:
                    k = paramiko.RSAKey.from_private_key_file(worker["keyfile"])
                    remote.connect(destination, username=worker["username"], pkey=k, port=worker["port"], timeout=60)
                else:
                    remote.connect(destination, username=worker["username"], password=worker["password"], port=worker["port"], timeout=60)
                break
            except socket.timeout:
                self.logger.error("Could not connect to worker %s" % destination)
                i += 1
        return remote

    def worker(self, ind_list, worker_id, environment):
        """
        Perform the actual fitness evaluation as a multithreaded worker. The
        worker pops off an individual from the list and evaluates it.

        Args:
            ind_list (list): List of strategy objects to evaluate
            worker_id (int): ID of this worker
            environment (dict): Environment dictionary
        """
        environment["remote"] = None

        if self.external_client:
            environment["remote"] = self.setup_remote()
            environment["worker"] = actions.utils.get_worker(self.external_client, self.logger)
        
        
        for ind in ind_list:
            if self.stop:
                break
            
            
            # Run a test
            if self.trw:
                if self.discarded==self.to_discard:
                    # reset discarding process
                    self.to_discard=random.choice(range(3))
                    self.discarded=0
                    #print("The TRW will discard {} flows".format(self.to_discard))
            eid, fitness = self.run_test(environment, ind)
            
            
            ## Uncomment to make regular periodic pauses instead ##
            if self.detectability != [] and self.detectability[-1]>=0.5 and self.Geneva==False:
            
            ## Uncomment for random jump ##
            # Selecting a random jump size
#            if self.random_jump:
#                self.jump=random.choice(range(1,self.max_jump))
#                
                print('***\nPausing Geneva, and switching to normal traffic for j={}\n***'.format(self.jump))
                output_path = os.path.join(PROJECT_ROOT, self.args.get("output_directory"))
                eid_normal = "normal_"+"".join(random.choice(string.ascii_letters+string.digits) for i in range(10))
                environment["id"] = eid
                self.client_args.update({"environment_id": eid_normal})
                self.server_args.update({"environment_id": eid_normal})
                pcap_filename = os.path.join(output_path, "packets", eid_normal + "_client.pcap") 
                #time.sleep(5)
                if self.args.get("censor"):
                    for ind in range(1,self.jump+1):
                        location = pcap_filename.split('.')[0]+str(ind)+'.pcap'
                        with actions.sniffer.Sniffer(location, 10000, self.logger,jump=self.jump, host=environment["client"]["ip"], iface="docker0",censor=self.censor) as sniff:
                            time.sleep(3)
                else:
                    
                    for ind in range(1,self.jump+1):
                        location = pcap_filename.split('.')[0]+str(ind)+'.pcap'
                        print('performing jump {}'.format(ind))
                        
                        if self.iface==None:
                            self.iface='eth0'
                        with actions.sniffer.Sniffer(location, 80, self.logger,jump=ind,iface=self.iface,censor=self.censor) as sniff:
                            try:
                                res = requests.get("http://www.baidu.com:80",allow_redirects=False, timeout=5)
                            except:
                                self.logger.debug('Connection to uncensored server {} is timed out. please use another website for normal traffic.'.format("http://www.baidu.com:80"))
                                pass
                            time.sleep(3)
                            #try:
                            #    res = requests.get("http://www.sina.com:80",allow_redirects=False, timeout=5)
                            #except:
                            #    self.logger.debug('Connection to uncensored server {} is timed out. please use another website for normal traffic.'.format("http://www.sina.com:80"))
                            #    pass
                            #time.sleep(3)
                            sniff.stop_sniffing()
                        
                        
                    #print('\n\n completed Geneva pause! \n\n')
                        #logger.debug(res.text)
                ## Reading pcap tracks to check detection
                    
                for index in range(1,self.jump+1):
                    df = actions.utils.parse_pcap(eid_normal, self.args, self.client_ip, index)
                    
                    if not self.args.get("censor"):
                        repeat=0
                        while type(df) == type(None) and repeat<5:
                            ## repeat background traffic
                            print('\n\nExtending pause ...\n\n')
                            location = pcap_filename.split('.')[0]+str(index)+'.pcap'
                            with actions.sniffer.Sniffer(location, 80, self.logger,jump=index,iface=self.iface,censor=self.censor) as sniff:
                                try:
                                    res = requests.get("http://www.baidu.com:80",allow_redirects=False, timeout=5)
                                except:
                                    self.logger.debug('Connection to uncensored server {} is timed out. please use another website for normal traffic.'.format("http://www.baidu.com:80"))
                                    pass
                                time.sleep(3)
                                try:
                                    res = requests.get("http://www.sina.com:80",allow_redirects=False, timeout=5)
                                except:
                                    self.logger.debug('Connection to uncensored server {} is timed out. please use another website for normal traffic.'.format("http://www.sina.com:80"))
                                    pass
                                time.sleep(3)
                                sniff.stop_sniffing()
                            df = actions.utils.parse_pcap(eid_normal, self.args, self.client_ip, index)
                            repeat+=1
                    #else:
                    #    df = actions.utils.parse_pcap(eid_normal, self.args, self.client_ip)
                    #if df is not None and '{}' not in df['flags'].tolist():
                    print(df)
                    # define flags to check for according to pretrained ML detector
                    new_cols = ['A', 'PA', 'SA', 'FA', 'RA', 'FPA', 'S', 'R', 'U', 'SEC', 'SRPUEC', 'FSRPAUE', 'FPCN', 'P', 'SAE', 'SPUE', 'F', 'RUE', 'FPUC', 'FUCN', 'SUC', 'FSRPAC', 'FSAN', 'SRAECN', 'AEC', 'FSRPEC', 'FUECN', 'FRA', 'PAU', 'SRACN', 'SE', 'SCN', 'FRPAUE', 'FSRPAEN', 'FRPEN', 'FSREC', 'SRPAUC', 'FRAECN', 'FSRUECN', 'SAC', 'RPAU', 'FPAE', 'SPAUECN', 'FRAUEN', 'FRAUN', 'PUN', 'FRPACN', 'SRPAUE', 'FSUE', 'FSU', 'SP', 'RPA', 'FSPAUECN', 'SPC', 'SRPACN', 'RPAUECN', 'SU', 'FSRAE', 'FRPAU', 'RAECN', 'FPEN', 'SRPECN', 'SPUN', 'SRAUE', 'FPUECN', 'RPAE', 'SUEC', 'RPUN', 'FPAUEN', 'RAUE', 'FR', 'SRUCN', 'FSPAE', 'FSRAECN', 'SR', 'FSPEN', 'FSRPUEN', 'FAE', 'SRAUECN', 'SPEN', 'FRUC', 'FRAN', 'FSPE', 'FSPUEC', 'SRPAN', 'FUEN', 'FRPAECN', 'SRPUCN', 'SRAE', 'FSPAN', 'SRA', 'PUE', 'FSUN', 'RAE', 'CN', 'FRPAC', 'SRAC', 'RN', 'SRPAUECN', 'FSRC', 'FRAEC', 'FSPUEN', 'SRUE', 'FSUC', 'SRPUECN', 'FRAE', 'FUN', 'FRPAUEC', 'FSPA', 'SRPUC', 'SAUE', 'RPAUE', 'FRPAEC', 'AUECN', 'SPAEN', 'RPU', 'FSPUECN', 'SRPEC', 'FSRPAUN', 'FAEN', 'AU', 'FSRPECN', 'PUC', 'FPAUN', 'FRUEN', 'FSPAUEC', 'SPUCN', 'SRPAEC', 'FUEC', 'FRUN', 'SEN', 'FSPAUE', 'FSRPAU', 'RPEN', 'FRPAUECN', 'FSAUCN', 'FRPUEC', 'EC', 'FRUCN', 'RAUECN', 'SUECN', 'PUCN', 'FPAEN', 'FSRAC', 'PAUN', 'FPACN', 'FRAUEC', 'RUEN', 'FAUECN', 'FSRAUEC', 'FPUEC', 'SRAUEN', 'FSUEC', 'FAUN', 'SPE', 'FRPN', 'RPECN', 'RAN', 'SACN', 'FSRUC', 'FS', 'FRPA', 'SPUC', 'SRPUN', 'RP', 'FSRPN', 'SPAN', 'SRAUC', 'PCN', 'RUECN', 'SPN']
                    for col in new_cols:
                        values = []
                        for elt in df['flags']:
                            #dictt = ast.literal_eval(elt)
                            dictt = elt
                            if col in dictt.keys():
                                values.append(dictt[col])
                            else:
                                values.append(0)
                        df[col] = values
        
        
                    del df['flags']
                    X = df.values
                    
                    # Check if flow is detectable
                    train_y_prob = self.train_detector.predict_proba(X)
                    eval_y_prob = self.eval_detector.predict_proba(X)
                    # Compute the avg probability that the source is running Geneva
                    P_g = np.mean(train_y_prob[:,0])
                    
                    #print('\n\n',y_prob)
                    print('\n\n',P_g)
                    y_pred = [y.argmax() for y in eval_y_prob]
                    self.y_preds.append(y_pred[0])
                    # update detection count
                    for y in y_pred:
                        if y==0: # Geneva detected
                            self.detection_count+=1
                    print('\n\n',y_pred)
                    print('Evaluation Predictions so far: ',self.y_preds)
                    #print('TRW history patamater ',self.history)
                    #print('Ratio changes so far: ',self.ratio_all)
                    print('Number of Evaluation detection so far: ',self.detection_count)
                    
                    for prob in eval_y_prob[:,0]:
                        self.detectability.append(prob)
    #                    
    #                    
                    ## TRW code for real-time evaluation
                    if self.trw:
                        #if self.discarded==self.to_discard:
                        print("\nUpdating TRW ratio...\n")
                        #if self.detectability != [] and self.detectability[-1]>=0.5 and self.discarded==0:
                        #    self.discarded+=1
                            
                        #if self.discarded==self.to_discard:
                        y=self.y_preds[-1]
                        if y==0: # predicted as geneva traffic
                            self.trw_ratio = self.trw_ratio*(self.theta1/self.theta0)
                            self.history+=1
                            self.ratio_all.append(self.trw_ratio)
                        if y==1: # predicted as legitimate traffic
                            self.trw_ratio = self.trw_ratio*((1-self.theta1)/(1-self.theta0))
                            self.ratio_all.append(self.trw_ratio)
                        # Checking for final decision
                        if self.trw_ratio <= self.eta0:
                            raise Exception('A normal flow is mistakingly detected as Geneva flow. Please improve your evaluation model.')
                        elif self.trw_ratio >= self.eta1:
                            # reset ratio for currently benign sources
                            if self.history == 0:
                                self.trw_ratio  = (self.eta0+self.eta1)/2 # averaging both thresholds
                            else:
                                self.trw_ratio = (self.history/len(self.y_preds)) * self.eta0 + (1-(self.history/len(self.y_preds))) * self.eta1
                                if self.trw_ratio<= self.eta0:
                                    print('Reset caused an automatic detection')
                            
                        #else:
                        #    self.discarded+=1
                        #    print("\n{} flows are discarded by TRW. {} discards are remaining\n".format(self.discarded,self.to_discard-self.discarded))
                detection_rate = self.detection_count/len(self.y_preds)
                
                print('Detection rate :',detection_rate)
                    
            
            
            
            if not fitness:
                fitness = -1000

            # Dump logs if requested
            if fitness < 0 and self.args.get("log_on_fail"):
                self.dump_logs(eid)
            elif fitness > 0 and self.args.get("log_on_success"):
                self.dump_logs(eid)

        # store detectability to disk
        print("count: ",self.detection_count)
        print("Polpulation detectability: ",self.detectability)

        actions.utils.write_detectability(self.detectability,self.detection_count,self.y_preds)
            
        # Clean up the test environment
        self.shutdown_environment(environment)

    def assign_ids(self, ind_list):
        """
        Assigns random environment ids to each individual to be evaluated.

        Args:
            ind_list (list): List of individuals to assign random IDs to
        """
        for ind in ind_list:
            ind.environment_id = actions.utils.get_id()
            ind.fitness = None

    def dump_logs(self, environment_id):
        """
        Dumps client, engine, server, and censor logs, to be called on test failure
        at ERROR level.

        Args:
            environment_id (str): Environment ID of a strategy to dump
        """
        log_files = ["client.log", "engine.log", "censor.log", "server.log"]
        for log_file in log_files:
            log = ""
            log_path = os.path.join(BASEPATH,
                                    self.output_directory,
                                    "logs",
                                    "%s.%s" % (environment_id, log_file))
            try:
                if not os.path.exists(log_path):
                    continue
                with open(log_path, "rb") as logfd:
                    log = logfd.read().decode('utf-8')
            except Exception:
                self.logger.exception("Failed to open log file")
                continue
            self.logger.error("%s: %s", log_file, log)

    def terminate_docker(self):
        """
        Terminates any hanging running containers.
        """
        if not self.use_docker:
            return
        # First, stop all the docker containers that match the given names
        # If a previous run was cut off in between container creation and startup,
        # we must also remove the container ('docker rm <container>')
        for operation in ["stop", "rm"]:
            try:
                output = subprocess.check_output(['docker', 'ps', '--format', "'{{.Names}}'"]).decode('utf-8')
            except subprocess.CalledProcessError:
                self.logger.error("Failed to list container names -- is docker running?")
                return
            if output.strip():
                self.logger.debug("Cleaning up docker (%s)" % operation)
            for name in output.splitlines():
                if any(key in name for key in ["client", "censor", "server", "benign"]):
                    try:
                        subprocess.check_output(['docker', operation, name])
                    except subprocess.CalledProcessError:
                        pass

    def initialize_base_container(self, name):
        """
        Builds a base container with a given name and connects it to a given network.
        Also retrieves lower level settings and the IP address of the container.

        Args:
            name (str): Name of this docker container

        Returns:
            dict: Dictionary containing docker container object and relevant information
        """
        try:
            container = {}
            container["name"] = name
            # Note that this is _not_ safe to do in a multiprocessed context - must be run single threaded.
            container["container"] = self.docker_client.containers.run('base', detach=True, privileged=True, volumes={os.path.abspath(os.getcwd()): {"bind" : "/code", "mode" : "rw"}}, environment=["LD_LIBRARY_PATH=$PATH:code/docker/harpoon/src/plugins"],tty=True, remove=True, name=name)
            
            container["settings"] = self.apiclient.inspect_container(name)
        except docker.errors.NotFound:
            self.logger.error("Could not run container \"base\". Is docker not running, or does the base container not exist?")
            return None

        return container

    def get_pid(self, container):
        """
        Returns PID of first actively running python process.

        Args:
            container: Docker container object to query

        Returns:
            int: PID of Python process
        """
        pid = None
        try:
            output = subprocess.check_output(["docker", "exec", container.name, "ps", "aux"], stderr=subprocess.PIPE).decode('utf-8')
        except subprocess.CalledProcessError:
            return None
        for line in output.splitlines():
            if "root" not in line or "python" not in line:
                continue
            parts = line.split()
            # Try to parse out the pid to confirm we found it
            try:
                pid = int(parts[1])
                break
            except ValueError:
                raise
        return pid

    def stop_censor(self, environment):
        """
        Send SIGKILL to all remaining python processes in the censor container.
        This is done intentionally over a SIGINT or a graceful shutdown mecahnism - due to
        dynamics with signal handling in nfqueue callbacks (threads), SIGINTs can be ignored
        and graceful shutdown mechanisms may not be picked up (or be fast enough).

        The output this method parses is below:

        .. code-block:: bash

            # ps aux
            USER        PID %CPU %MEM    VSZ   RSS TTY      STAT START   TIME COMMAND
            root          1  0.1  0.0  21944  3376 pts/0    Ss+  13:30   0:00 /bin/bash
            root         14 24.0  0.4 200376 38564 ?        Ss   13:30   0:00 python code/censor_driver.py censor2 jgsko1rf trials/2018-10-30_06:30:48 60181
            root         32  0.0  0.0  19188  2400 ?        Rs   13:30   0:00 ps aux

        Args:
            environment (dict): Environment dictionary
        """
        port = environment["port"]
        queue_num = environment["queue_num"]
        if environment:
            pid = self.get_pid(environment["censor"]["container"])
            while pid:
                #self.logger.info("%s killing process %s in %s" % (environment["id"], str(pid), environment["censor"]["container"].name))
                try:
                    subprocess.check_call(["docker", "exec", "--privileged", environment["censor"]["container"].name, "kill", "-9", str(pid)])
                except subprocess.CalledProcessError:
                    pass
                pid = self.get_pid(environment["censor"]["container"])
                time.sleep(0.25)

            try:
                subprocess.check_call(["docker", "exec", "--privileged", environment["censor"]["container"].name, "iptables", "-D", "FORWARD", "-j", "NFQUEUE", "-p", "tcp", "--sport", str(port), "--queue-num", str(queue_num)])
            except subprocess.CalledProcessError:
                pass
            try:
                subprocess.check_call(["docker", "exec",  "--privileged",environment["censor"]["container"].name, "iptables", "-D", "FORWARD", "-j", "NFQUEUE", "-p", "tcp", "--dport", str(port), "--queue-num", str(queue_num)])
            except subprocess.CalledProcessError:
                pass

    def start_censor(self, environment, environment_id):
        """
        Starts the censor in the server environment container.

        Args:
            environment (dict): Environment dictionary
            environment_id (str): Environment ID of the censor to stop
        """
        assert self.use_docker, "Cannot start censor without enabling docker"
        port = environment["port"]
        queue_num = environment["queue_num"]
        try:
            self.logger.debug(" Starting censor %s with driver" % self.censor)
            command = ["docker", "exec",  "--privileged", environment["censor"]["container"].name,
                       "python", "code/censors/censor_driver.py",
                       "--censor", self.censor,
                       "--environment-id", environment_id,
                       "--output-directory", self.output_directory,
                       "--port", str(port),
                       "--log", "debug",
                       "--forbidden", self.args.get("bad_word", "ultrasurf"),
                       "--queue", str(queue_num)]
            self.exec_cmd(command)
        except (subprocess.CalledProcessError, subprocess.TimeoutExpired):
            # Docker containers were killed out from under us - likely means
            # user forced a shutdown. Bail gracefully.
            return False
        except Exception:
            self.logger.exception("Failed out of start_censor")
        finally:
            self.logger.debug("Dockerized censor thread exiting")

    def read_fitness(self, ind):
        """
        Looks for this individual's fitness file on disk, opens it, and stores the fitness in the given individual.

        Args:
            ind (:obj:`actions.strategy.Strategy`): Individual to read fitness for
        """
        fitness_path = os.path.join(BASEPATH, self.output_directory, actions.utils.FLAGFOLDER, ind.environment_id + ".fitness")
        try:
            if os.path.exists(fitness_path):
                with open(fitness_path, "r") as fd:
                    ind.fitness = float(fd.read())
            elif not ind.fitness:
                self.logger.warning("Could not find fitness file for %s" % fitness_path)
                ind.fitness = -1000
        except:
            self.logger.exception("[%s] Failed to read fitness file" % ind.environment_id)
            ind.fitness = -1000

    def shutdown_container(self, container):
        """
        Tries to shutdown a given container and eats a NotFound exception if the container
        has already exited.

        Args:
            container: docker container object to call stop() on
        """
        try:
            container.stop()
        except docker.errors.NotFound:
            pass

    def shutdown_environment(self, environment):
        """
        Shuts down the evaluation environment.
        If Docker, shuts down server and client container.
        If a remote SSH connection, the connection is shut down.
        """
        if environment.get("docker"):
            self.shutdown_container(environment["client"]["container"])
            if self.censor:
                self.shutdown_container(environment["censor"]["container"])
                self.shutdown_container(environment["server"]["container"])
                self.shutdown_container(environment["benign_server"]["container"])
        elif environment.get("remote"):
            environment["remote"].close()

    def shutdown(self):
        """
        Shuts down all active environments
        """
        self.terminate_docker()


def collect_plugin(test_plugin, plugin_type, command, full_args, plugin_args):
    """
    Import a given plugin

    Args:
        test_plugin (str): Plugin name to import ("http")
        plugin_type (str): Component of plugin to import ("client")
        command (list): sys.argv or list of arguments
        full_args (dict): Parsed full list of arguments already maintained by the parent plugin
        plugin_args (dict): Dictionary of args specific to this plugin component

    Returns:
        Imported plugin class for instantiation later
    """
    cls = None
    try:
        _, cls = actions.utils.import_plugin(test_plugin, plugin_type)
        parsed_args = cls.get_args(command)
        # Only override the args if the plugin successfully parsed something; this allows
        # defaults from the evaluator or plugin to propagate.
        parsed_args = {k:v for k,v in parsed_args.items() if v or (not v and k not in full_args) }
        full_args.update(parsed_args)
        plugin_args.update(parsed_args)
    except ImportError as exc:
        pass
    return cls


def get_random_open_port():
    """
    Selects a random ephemeral port that is open.

    Returns:
        int: Open port
    """
    while True:
        port = random.randint(1024, 65000)
        # Bind TCP socket
        try:
            with socket.socket() as sock:
                # If we can bind, nothing is listening
                sock.bind(('', port))
                break
        except OSError:
            continue
    return port


def get_arg_parser(single_use=False):
    """
    Sets up argparse. This is done separately to enable collection of help messages.

    Args:
        single_use (bool, optional): whether this evaluator will only be used for one strategy, used to configure sane defaults
    """
    # Disable prefix matching to avoid prefix collisions for unseen plugin arguments
    parser = argparse.ArgumentParser(description='Evaluate a given strategy a given number of times.', allow_abbrev=False, prog="evaluator.py")
    # Type of evaluation
    parser.add_argument('--test-type', action='store', choices=actions.utils.get_plugins(), default="http", help="plugin to launch")
    parser.add_argument('--strategy', action='store', default="", required=single_use, help='strategy to evaluate')

    logging_group = parser.add_argument_group('control aspects of evaluator logging and storage')
    logging_group.add_argument('--log', action='store', choices=("debug", "info", "warning", "critical", "error"), help="Sets the log level")
    logging_group.add_argument('--output-directory', action='store', help="where to output results")
    logging_group.add_argument('--log-on-fail', action='store_true', help="dump the logs associated with each individual on strategy failure")
    logging_group.add_argument('--log-on-success', action='store_true', help="dump the logs associated with each individual on strategy success")

    external_group = parser.add_argument_group('control aspects of external resource usage')
    external_group.add_argument('--external-server', action='store_true', help="use an external server for testing.")
    external_group.add_argument('--external-client', action='store', help="use the given external client for testing.")

    networking_group = parser.add_argument_group('control aspects of evaluator networking configuration')
    networking_group.add_argument('--server-side', action="store_true", help="run the Geneva engine on the server side, not the client")
    networking_group.add_argument('--public-ip', action='store', help="public facing IP for this computer for server-side evaluation.")
    networking_group.add_argument('--routing-ip', action='store', help="locally facing IP for this computer, used for NAT")
    networking_group.add_argument('--sender-ip', action='store', help="IP address of sending machine, used for NAT")
    networking_group.add_argument('--forward-ip', action='store', help="IP address to forward traffic to")
    networking_group.add_argument('--act-as-middlebox', action='store_true', help="enables NAT mode. Requires --routing-ip, --sender-ip, and --forward-ip")
    networking_group.add_argument('--port', action='store', type=int, default=get_random_open_port(), help='default port to use')
    networking_group.add_argument('--iface', action='store', help="Network interface name on your device to use for sniffing packets")

    docker_group = parser.add_argument_group('control aspects of docker-specific options')
    docker_group.add_argument('--censor', action='store', help='censor to test against.', choices=censors.censor_driver.get_censors())
    docker_group.add_argument('--workers', action='store', default=1, type=int, help='controls the number of docker containers the evaluator will use.')
    docker_group.add_argument('--bad-word', action='store', help="forbidden word to test with", default="ultrasurf")
    

    evaluation_group = parser.add_argument_group('control aspects of evaluation')
    evaluation_group.add_argument('--runs', type=int, default=1, action='store', help="number of times each individual should be run per evaluation")
    evaluation_group.add_argument("--fitness-by", action='store', choices=('min', 'avg', 'max'), default='avg', help="if each individual is run multiple times, control how fitness is assigned.")
    evaluation_group.add_argument('--no-skip-empty', action='store_true', help="evaluate empty strategies (default: False).")
    evaluation_group.add_argument('--real-time-detection', action='store_true', help="enables [TRW] for real-time detection evaluation (default: False")
    evaluation_group.add_argument('--random-jump', action='store_true', help="enables randomizing the jump size of background normal traffic")
    evaluation_group.add_argument('--jump', type=int, default=0, action='store', help="number of times dummy evaluations after a detected strategy")
    evaluation_group.add_argument("--local-model", action='store', default='rfc.joblib', help="Specify DeResistor's training local model name")
    evaluation_group.add_argument("--censor-model", action='store', default='rfc.joblib', help="Specify the censor's testing model name")
    evaluation_group.add_argument('--Geneva', action='store_true', help="Disavles all DeResistor functionalities and run normal Geneva")
    
    return parser


def get_args(cmd, single_use=False):
    """
    Creates an argparser and collects arguments.

    Args:
        single_use (bool, optional): whether this evaluator will only be used for one strategy, used to configure sane defaults

    Returns:
        dict: parsed args
    """
    parser = get_arg_parser(single_use=single_use)

    args, _ = parser.parse_known_args(cmd)
    return vars(args)