l3_centralizedattackdetectorServiceServicerImpl.py

# Copyright 2022-2023 ETSI TeraFlowSDN - TFS OSG (https://tfs.etsi.org/)
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from __future__ import print_function
from datetime import datetime
from datetime import timedelta

import os
import numpy as np
import onnxruntime as rt
import logging
import time

import csv
from multiprocessing import Process
from multiprocessing import Value

from common.proto.l3_centralizedattackdetector_pb2 import Empty, AutoFeatures
from common.proto.l3_centralizedattackdetector_pb2_grpc import L3CentralizedattackdetectorServicer

from common.proto.l3_attackmitigator_pb2 import L3AttackmitigatorOutput

from common.proto.monitoring_pb2 import KpiDescriptor
from common.proto.kpi_sample_types_pb2 import KpiSampleType

from monitoring.client.MonitoringClient import MonitoringClient
from common.proto.monitoring_pb2 import Kpi

from common.tools.timestamp.Converters import timestamp_utcnow_to_float
from common.proto.context_pb2 import Timestamp, SliceId, ConnectionId

from l3_attackmitigator.client.l3_attackmitigatorClient import l3_attackmitigatorClient

import uuid

import sklearn.metrics as metrics
import numpy as np

from common.method_wrappers.Decorator import MetricsPool, safe_and_metered_rpc_method

LOGGER = logging.getLogger(__name__)
current_dir = os.path.dirname(os.path.abspath(__file__))

# Demo constants
DEMO_MODE = True
ATTACK_IPS = ["37.187.95.110", "91.121.140.167", "94.23.23.52", "94.23.247.226", "149.202.83.171"]
BATCH_SIZE= 10
TIME_START = time.time()
METRICS_POOL = MetricsPool('l3_centralizedattackdetector', 'RPC')


class ConnectionInfo:
    def __init__(self, ip_o, port_o, ip_d, port_d):
        self.ip_o = ip_o
        self.port_o = port_o
        self.ip_d = ip_d
        self.port_d = port_d

    def __eq__(self, other):
        return (
            self.ip_o == other.ip_o
            and self.port_o == other.port_o
            and self.ip_d == other.ip_d
            and self.port_d == other.port_d
        )

    def __str__(self):
        return "ip_o: " + self.ip_o + "\nport_o: " + self.port_o + "\nip_d: " + self.ip_d + "\nport_d: " + self.port_d


class l3_centralizedattackdetectorServiceServicerImpl(L3CentralizedattackdetectorServicer):

    """
    Initialize variables, prediction model and clients of components used by CAD
    """

    def __init__(self):
        LOGGER.info("Creating Centralized Attack Detector Service - Scalability Experiment 2")

        self.inference_values = []
        self.inference_results = []
        self.cryptomining_detector_path = os.path.join(current_dir, "ml_model/cryptomining_detector/")
        self.cryptomining_detector_file_name = os.listdir(self.cryptomining_detector_path)[0]
        self.cryptomining_detector_model_path = os.path.join(
            self.cryptomining_detector_path, self.cryptomining_detector_file_name
        )
        self.cryptomining_detector_model = rt.InferenceSession(self.cryptomining_detector_model_path)

        # Load cryptomining detector features metadata from ONNX file
        self.cryptomining_detector_features_metadata = list(
            self.cryptomining_detector_model.get_modelmeta().custom_metadata_map.values()
        )
        self.cryptomining_detector_features_metadata = [float(x) for x in self.cryptomining_detector_features_metadata]
        self.cryptomining_detector_features_metadata.sort()
        LOGGER.info("Cryptomining Detector Features: " + str(self.cryptomining_detector_features_metadata))

        self.input_name = self.cryptomining_detector_model.get_inputs()[0].name
        self.label_name = self.cryptomining_detector_model.get_outputs()[0].name
        self.prob_name = self.cryptomining_detector_model.get_outputs()[1].name

        # Kpi values
        self.l3_security_status = 0  # unnecessary
        self.l3_ml_model_confidence = 0
        self.l3_inferences_in_interval_counter = 0

        self.l3_ml_model_confidence_normal = 0
        self.l3_inferences_in_interval_counter_normal = 0

        self.l3_ml_model_confidence_crypto = 0
        self.l3_inferences_in_interval_counter_crypto = 0

        self.l3_attacks = []
        self.l3_unique_attack_conns = 0
        self.l3_unique_compromised_clients = 0
        self.l3_unique_attackers = 0

        self.l3_non_empty_time_interval = False
        
        self.active_requests = []

        self.monitoring_client = MonitoringClient()
        self.service_ids = []
        self.monitored_kpis = {
            "l3_security_status": {
                "kpi_id": None,
                "description": "L3 - Confidence of the cryptomining detector in the security status in the last time interval of the service {service_id}",
                "kpi_sample_type": KpiSampleType.KPISAMPLETYPE_L3_SECURITY_STATUS_CRYPTO,
                "service_ids": [],
            },
            "l3_ml_model_confidence": {
                "kpi_id": None,
                "description": "L3 - Security status of the service in a time interval of the service {service_id} (“0” if no attack has been detected on the service and “1” if a cryptomining attack has been detected)",
                "kpi_sample_type": KpiSampleType.KPISAMPLETYPE_ML_CONFIDENCE,
                "service_ids": [],
            },
            "l3_unique_attack_conns": {
                "kpi_id": None,
                "description": "L3 - Number of attack connections detected in a time interval of the service {service_id} (attacks of the same connection [origin IP, origin port, destination IP and destination port] are only considered once)",
                "kpi_sample_type": KpiSampleType.KPISAMPLETYPE_L3_UNIQUE_ATTACK_CONNS,
                "service_ids": [],
            },
            "l3_unique_compromised_clients": {
                "kpi_id": None,
                "description": "L3 - Number of unique compromised clients of the service in a time interval of the service {service_id} (attacks from the same origin IP are only considered once)",
                "kpi_sample_type": KpiSampleType.KPISAMPLETYPE_L3_UNIQUE_COMPROMISED_CLIENTS,
                "service_ids": [],
            },
            "l3_unique_attackers": {
                "kpi_id": None,
                "description": "L3 - number of unique attackers of the service in a time interval of the service {service_id} (attacks from the same destination IP are only considered once)",
                "kpi_sample_type": KpiSampleType.KPISAMPLETYPE_L3_UNIQUE_ATTACKERS,
                "service_ids": [],
            },
        }
        self.attackmitigator_client = l3_attackmitigatorClient()

        # Environment variables
        self.CLASSIFICATION_THRESHOLD = os.getenv("CAD_CLASSIFICATION_THRESHOLD", 0.5)
        self.MONITORED_KPIS_TIME_INTERVAL_AGG = os.getenv("MONITORED_KPIS_TIME_INTERVAL_AGG", 60)

        # Constants
        self.NORMAL_CLASS = 0
        self.CRYPTO_CLASS = 1

        self.kpi_test = None
        self.time_interval_start = None
        self.time_interval_end = None

        # CAD evaluation tests
        self.cad_inference_times = []
        self.cad_num_inference_measurements = 100

        # AM evaluation tests
        self.am_notification_times = []

        # List of attack connections
        self.attack_connections = []

        # Accuracy metrics
        self.correct_attack_conns = 0
        self.correct_predictions = 0
        self.attack_connections_len = Value('f', 0) # Must use multiprocessing.Value to share values between processes
        self.total_predictions = Value('f', 0)
        self.false_positives = Value('f', 0)
        self.false_negatives = Value('f', 0)
        self.overall_detection_acc = Value('f', 0)
        self.cryptomining_attack_detection_acc = Value('f', 0)
        self.confidence = Value('f', 0)
        
        self.f1_score_macro = Value('f', 0)
        self.f1_score_weighted = Value('f', 0)
        self.balanced_accuracy = Value('f', 0)
        self.precision_score = Value('f', 0)
        self.recall_score = Value('f', 0)
        
        self.replica_uuid = uuid.uuid4()
        
        self.y_true = []
        self.y_pred = []
        
        self.max_connection_time = 60
        self.time_to_stabilize = 5

    """
    Create a monitored KPI for a specific service and add it to the Monitoring Client
        -input: 
            + service_id: service ID where the KPI will be monitored
            + kpi_name: name of the KPI
            + kpi_description: description of the KPI
            + kpi_sample_type: KPI sample type of the KPI (it must be defined in the kpi_sample_types.proto file)
        -output: KPI identifier representing the KPI
    """

    def create_kpi(
        self,
        service_id,
        kpi_name,
        kpi_description,
        kpi_sample_type,
    ):
        kpidescriptor = KpiDescriptor()
        kpidescriptor.kpi_description = kpi_description
        kpidescriptor.service_id.service_uuid.uuid = service_id.service_uuid.uuid
        kpidescriptor.kpi_sample_type = kpi_sample_type
        new_kpi = self.monitoring_client.SetKpi(kpidescriptor)

        LOGGER.info("Created KPI {}".format(kpi_name))

        return new_kpi

    """
    Create the monitored KPIs for a specific service, add them to the Monitoring Client and store their identifiers in the monitored_kpis dictionary
        -input:
            + service_id: service ID where the KPIs will be monitored
        -output: None
    """

    def create_kpis(self, service_id, device_id, endpoint_id):
        LOGGER.info("Creating KPIs for service {}".format(service_id))

        # for now, all the KPIs are created for all the services from which requests are received
        for kpi in self.monitored_kpis:
            # generate random slice_id
            slice_id = SliceId()
            slice_id.slice_uuid.uuid = str(uuid.uuid4())

            # generate random connection_id
            connection_id = ConnectionId()
            connection_id.connection_uuid.uuid = str(uuid.uuid4())

            created_kpi = self.create_kpi(
                service_id,
                kpi,
                self.monitored_kpis[kpi]["description"].format(service_id=service_id.service_uuid.uuid),
                self.monitored_kpis[kpi]["kpi_sample_type"],
            )
            self.monitored_kpis[kpi]["kpi_id"] = created_kpi.kpi_id
            self.monitored_kpis[kpi]["service_ids"].append(service_id.service_uuid.uuid)

        LOGGER.info("Created KPIs for service {}\n".format(service_id))

    def monitor_kpis(self):
        monitor_inference_results = self.inference_results
        monitor_service_ids = self.service_ids

        self.assign_timestamp(monitor_inference_results)

        non_empty_time_interval = self.l3_non_empty_time_interval

        if non_empty_time_interval:
            for service_id in monitor_service_ids:
                #LOGGER.debug("service_id: {}".format(service_id))

                self.monitor_compute_l3_kpi(service_id, monitor_inference_results)

                # Demo mode inference results are erased
                """if DEMO_MODE:
                    # Delete fist half of the inference results
                    LOGGER.debug("inference_results len: {}".format(len(self.inference_results)))
                    self.inference_results = self.inference_results[len(self.inference_results)//2:]
                    LOGGER.debug("inference_results len after erase: {}".format(len(self.inference_results)))"""
                # end = time.time()
                # LOGGER.debug("Time to process inference results with erase: {}".format(end - start))
                LOGGER.debug("KPIs sent to monitoring server\n")
        else:
            LOGGER.debug("No KPIs sent to monitoring server")

    def assign_timestamp(self, monitor_inference_results):
        time_interval = self.MONITORED_KPIS_TIME_INTERVAL_AGG

        # assign the timestamp of the first inference result to the time_interval_start
        if self.time_interval_start is None:
            self.time_interval_start = monitor_inference_results[0]["timestamp"]
            #LOGGER.debug("self.time_interval_start: {}".format(self.time_interval_start))

            # add time_interval to the current time to get the time interval end
            #LOGGER.debug("time_interval: {}".format(time_interval))
            #LOGGER.debug(timedelta(seconds=time_interval))
            self.time_interval_end = self.time_interval_start + timedelta(seconds=time_interval)

        current_time = datetime.utcnow()

        #LOGGER.debug("current_time: {}".format(current_time))

        if current_time >= self.time_interval_end:
            self.time_interval_start = self.time_interval_end
            self.time_interval_end = self.time_interval_start + timedelta(seconds=time_interval)
            self.l3_security_status = 0  # unnecessary
            self.l3_ml_model_confidence = 0
            self.l3_inferences_in_interval_counter = 0

            self.l3_ml_model_confidence_normal = 0
            self.l3_inferences_in_interval_counter_normal = 0

            self.l3_ml_model_confidence_crypto = 0
            self.l3_inferences_in_interval_counter_crypto = 0

            self.l3_attacks = []
            self.l3_unique_attack_conns = 0
            self.l3_unique_compromised_clients = 0
            self.l3_unique_attackers = 0

            self.l3_non_empty_time_interval = False

        #LOGGER.debug("time_interval_start: {}".format(self.time_interval_start))
        #LOGGER.debug("time_interval_end: {}".format(self.time_interval_end))

    def monitor_compute_l3_kpi(self, service_id, monitor_inference_results):
        # L3 security status
        kpi_security_status = Kpi()
        kpi_security_status.kpi_id.kpi_id.CopyFrom(self.monitored_kpis["l3_security_status"]["kpi_id"])
        kpi_security_status.kpi_value.int32Val = self.l3_security_status

        # L3 ML model confidence
        kpi_conf = Kpi()
        kpi_conf.kpi_id.kpi_id.CopyFrom(self.monitored_kpis["l3_ml_model_confidence"]["kpi_id"])
        kpi_conf.kpi_value.floatVal = self.monitor_ml_model_confidence()

        # L3 unique attack connections
        kpi_unique_attack_conns = Kpi()
        kpi_unique_attack_conns.kpi_id.kpi_id.CopyFrom(self.monitored_kpis["l3_unique_attack_conns"]["kpi_id"])
        kpi_unique_attack_conns.kpi_value.int32Val = self.l3_unique_attack_conns

        # L3 unique compromised clients
        kpi_unique_compromised_clients = Kpi()
        kpi_unique_compromised_clients.kpi_id.kpi_id.CopyFrom(
            self.monitored_kpis["l3_unique_compromised_clients"]["kpi_id"]
        )
        kpi_unique_compromised_clients.kpi_value.int32Val = self.l3_unique_compromised_clients

        # L3 unique attackers
        kpi_unique_attackers = Kpi()
        kpi_unique_attackers.kpi_id.kpi_id.CopyFrom(self.monitored_kpis["l3_unique_attackers"]["kpi_id"])
        kpi_unique_attackers.kpi_value.int32Val = self.l3_unique_attackers

        timestamp = Timestamp()
        timestamp.timestamp = timestamp_utcnow_to_float()

        kpi_security_status.timestamp.CopyFrom(timestamp)
        kpi_conf.timestamp.CopyFrom(timestamp)
        kpi_unique_attack_conns.timestamp.CopyFrom(timestamp)
        kpi_unique_compromised_clients.timestamp.CopyFrom(timestamp)
        kpi_unique_attackers.timestamp.CopyFrom(timestamp)

        LOGGER.debug("Sending KPIs to monitoring server")

        '''LOGGER.debug("kpi_security_status: {}".format(kpi_security_status))
        LOGGER.debug("kpi_conf: {}".format(kpi_conf))
        LOGGER.debug("kpi_unique_attack_conns: {}".format(kpi_unique_attack_conns))
        LOGGER.debug("kpi_unique_compromised_clients: {}".format(kpi_unique_compromised_clients))
        LOGGER.debug("kpi_unique_attackers: {}".format(kpi_unique_attackers))'''

        try:
            self.monitoring_client.IncludeKpi(kpi_security_status)
            self.monitoring_client.IncludeKpi(kpi_conf)
            self.monitoring_client.IncludeKpi(kpi_unique_attack_conns)
            self.monitoring_client.IncludeKpi(kpi_unique_compromised_clients)
            self.monitoring_client.IncludeKpi(kpi_unique_attackers)
        except Exception as e:
            LOGGER.debug("Error sending KPIs to monitoring server: {}".format(e))

    def monitor_ml_model_confidence(self):
        if self.l3_security_status == 0:
            return self.l3_ml_model_confidence_normal

        return self.l3_ml_model_confidence_crypto

    """
    Classify connection as standard traffic or cryptomining attack and return results
        -input: 
            + request: L3CentralizedattackdetectorMetrics object with connection features information
        -output: L3AttackmitigatorOutput object with information about the assigned class and prediction confidence
    """

    def perform_distributed_inference(self, requests):
        batch_size = len(requests)

        # Create an empty array to hold the input data
        x_data = np.empty((batch_size, len(requests[0].features)))

        # Fill in the input data array with features from each request
        for i, request in enumerate(requests):
            x_data[i] = [feature.feature for feature in request.features]

        # Print input data shape
        LOGGER.debug("x_data.shape: {}".format(x_data.shape))

        inference_time_start = time.perf_counter()

        # Perform inference
        predictions = self.cryptomining_detector_model.run(
            [self.prob_name], {self.input_name: x_data.astype(np.float32)}
        )[0]

        inference_time_end = time.perf_counter()

        # Measure inference time
        inference_time = inference_time_end - inference_time_start
        self.cad_inference_times.append(inference_time)

        if len(self.cad_inference_times) > self.cad_num_inference_measurements:
            inference_times_np_array = np.array(self.cad_inference_times)
            np.save(f"inference_times_{batch_size}.npy", inference_times_np_array)

            avg_inference_time = np.mean(inference_times_np_array)
            max_inference_time = np.max(inference_times_np_array)
            min_inference_time = np.min(inference_times_np_array)
            std_inference_time = np.std(inference_times_np_array)
            median_inference_time = np.median(inference_times_np_array)

            LOGGER.debug("Average inference time: {}".format(avg_inference_time))
            LOGGER.debug("Max inference time: {}".format(max_inference_time))
            LOGGER.debug("Min inference time: {}".format(min_inference_time))
            LOGGER.debug("Standard deviation inference time: {}".format(std_inference_time))
            LOGGER.debug("Median inference time: {}".format(median_inference_time))

            with open(f"inference_times_stats_{batch_size}.txt", "w") as f:
                f.write("Average inference time: {}\n".format(avg_inference_time))
                f.write("Max inference time: {}\n".format(max_inference_time))
                f.write("Min inference time: {}\n".format(min_inference_time))
                f.write("Standard deviation inference time: {}\n".format(std_inference_time))
                f.write("Median inference time: {}\n".format(median_inference_time))

        # Gather the predicted class, the probability of that class and other relevant information required to block the attack
        output_messages = []
        for i, request in enumerate(requests):
            output_messages.append({
                "confidence": None,
                "timestamp": datetime.now().strftime("%d/%m/%Y %H:%M:%S"),
                "ip_o": request.connection_metadata.ip_o,
                "ip_d": request.connection_metadata.ip_d,
                "tag_name": None,
                "tag": None,
                "flow_id": request.connection_metadata.flow_id,
                "protocol": request.connection_metadata.protocol,
                "port_o": request.connection_metadata.port_o,
                "port_d": request.connection_metadata.port_d,
                "ml_id": self.cryptomining_detector_file_name,
                "service_id": request.connection_metadata.service_id,
                "endpoint_id": request.connection_metadata.endpoint_id,
                "time_start": request.connection_metadata.time_start,
                "time_end": request.connection_metadata.time_end,
            })

            if predictions[i][1] >= self.CLASSIFICATION_THRESHOLD:
                output_messages[i]["confidence"] = predictions[i][1]
                output_messages[i]["tag_name"] = "Crypto"
                output_messages[i]["tag"] = self.CRYPTO_CLASS
            else:
                output_messages[i]["confidence"] = predictions[i][0]
                output_messages[i]["tag_name"] = "Normal"
                output_messages[i]["tag"] = self.NORMAL_CLASS

        return output_messages

    """
    Receive features from Attack Mitigator, predict attack and communicate with Attack Mitigator
        -input: 
            + request: L3CentralizedattackdetectorMetrics object with connection features information
        -output: Empty object with a message about the execution of the function
    """
    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
    def AnalyzeConnectionStatistics(self, request, context):
        # Perform inference with the data sent in the request
        self.active_requests.append(request)
        
        if len(self.active_requests) == BATCH_SIZE:
            csv_file_path = 'hola_mundo.csv'
        
            col_values = [1, 2, 3]
            
            with open(csv_file_path, 'a', newline='') as file:
                writer = csv.writer(file)
                writer.writerow(col_values)
            
            logging.debug("Performing inference... {}".format(self.replica_uuid))
            
            inference_time_start = time.time()
            cryptomining_detector_output = self.perform_distributed_inference(self.active_requests)
            inference_time_end = time.time()
            
            LOGGER.debug("Inference performed in {} seconds".format(inference_time_end - inference_time_start))
            logging.info("Inference performed correctly")

            self.inference_results.append({"output": cryptomining_detector_output, "timestamp": datetime.now()})
            LOGGER.debug("inference_results length: {}".format(len(self.inference_results)))

            for i, req in enumerate(self.active_requests):
                service_id = req.connection_metadata.service_id
                device_id = req.connection_metadata.endpoint_id.device_id
                endpoint_id = req.connection_metadata.endpoint_id

                # Check if a request of a new service has been received and, if so, create the monitored KPIs for that service
                if service_id not in self.service_ids:
                    self.create_kpis(service_id, device_id, endpoint_id)
                    self.service_ids.append(service_id)

                monitor_kpis_start = time.time()
                self.monitor_kpis()
                monitor_kpis_end = time.time()

                LOGGER.debug("Monitoring KPIs performed in {} seconds".format(monitor_kpis_end - monitor_kpis_start))
                LOGGER.debug("cryptomining_detector_output: {}".format(cryptomining_detector_output[i]))

                if DEMO_MODE:
                    self.confidence.value = cryptomining_detector_output[i]["confidence"]
                    self.analyze_prediction_accuracy()

                connection_info = ConnectionInfo(
                    req.connection_metadata.ip_o,
                    req.connection_metadata.port_o,
                    req.connection_metadata.ip_d,
                    req.connection_metadata.port_d,
                )

                self.l3_non_empty_time_interval = True

                if cryptomining_detector_output[i]["tag_name"] == "Crypto":
                    self.l3_security_status = 1

                    self.l3_inferences_in_interval_counter_crypto += 1
                    self.l3_ml_model_confidence_crypto = (
                        self.l3_ml_model_confidence_crypto * (self.l3_inferences_in_interval_counter_crypto - 1)
                        + cryptomining_detector_output[i]["confidence"]
                    ) / self.l3_inferences_in_interval_counter_crypto

                    if connection_info not in self.l3_attacks:
                        self.l3_attacks.append(connection_info)
                        self.l3_unique_attack_conns += 1

                    self.l3_unique_compromised_clients = len(set([conn.ip_o for conn in self.l3_attacks]))
                    self.l3_unique_attackers = len(set([conn.ip_d for conn in self.l3_attacks]))

                else:
                    self.l3_inferences_in_interval_counter_normal += 1
                    self.l3_ml_model_confidence_normal = (
                        self.l3_ml_model_confidence_normal * (self.l3_inferences_in_interval_counter_normal - 1)
                        + cryptomining_detector_output[i]["confidence"]
                    ) / self.l3_inferences_in_interval_counter_normal

                # Only notify Attack Mitigator when a cryptomining connection has been detected
                if cryptomining_detector_output[i]["tag_name"] == "Crypto":
                    if DEMO_MODE:
                        self.attack_connections.append(connection_info)
                        
                    # Calculate F1 score
                    self.y_pred.append(1)

                    if connection_info.ip_o in ATTACK_IPS or connection_info.ip_d in ATTACK_IPS:
                        self.correct_attack_conns += 1
                        self.correct_predictions += 1
                        self.y_true.append(1)
                    else:
                        LOGGER.debug("False positive: {}".format(connection_info))
                        self.false_positives.value = self.false_positives.value + 1
                        self.y_true.append(0)

                    self.total_predictions.value = self.total_predictions.value + 1

                    # if False:
                    notification_time_start = time.perf_counter()

                    LOGGER.debug("Crypto attack detected")

                    # Notify the Attack Mitigator component about the attack
                    logging.info(
                        "Notifying the Attack Mitigator component about the attack in order to block the connection..."
                    )

                    try:
                        logging.info("Sending the connection information to the Attack Mitigator component...")
                        message = L3AttackmitigatorOutput(**cryptomining_detector_output[i])
                        response = self.attackmitigator_client.PerformMitigation(message)
                        notification_time_end = time.perf_counter()

                        self.am_notification_times.append(notification_time_end - notification_time_start)

                        LOGGER.debug(f"am_notification_times length: {len(self.am_notification_times)}")
                        LOGGER.debug(f"last am_notification_time: {self.am_notification_times[-1]}")

                        if len(self.am_notification_times) > 100:
                            am_notification_times_np_array = np.array(self.am_notification_times)
                            np.save("am_notification_times.npy", am_notification_times_np_array)

                            avg_notification_time = np.mean(am_notification_times_np_array)
                            max_notification_time = np.max(am_notification_times_np_array)
                            min_notification_time = np.min(am_notification_times_np_array)
                            std_notification_time = np.std(am_notification_times_np_array)
                            median_notification_time = np.median(am_notification_times_np_array)

                            LOGGER.debug("Average notification time: {}".format(avg_notification_time))
                            LOGGER.debug("Max notification time: {}".format(max_notification_time))
                            LOGGER.debug("Min notification time: {}".format(min_notification_time))
                            LOGGER.debug("Std notification time: {}".format(std_notification_time))
                            LOGGER.debug("Median notification time: {}".format(median_notification_time))

                            with open("am_notification_times_stats.txt", "w") as f:
                                f.write("Average notification time: {}\n".format(avg_notification_time))
                                f.write("Max notification time: {}\n".format(max_notification_time))
                                f.write("Min notification time: {}\n".format(min_notification_time))
                                f.write("Std notification time: {}\n".format(std_notification_time))
                                f.write("Median notification time: {}\n".format(median_notification_time))

                        # logging.info("Attack Mitigator notified and received response: ", response.message)  # FIX No message received
                        logging.info("Attack Mitigator notified")

                        #return Empty(message="OK, information received and mitigator notified abou the attack")
                    
                    except Exception as e:
                        logging.error("Error notifying the Attack Mitigator component about the attack: ", e)
                        logging.error("Couldn't find l3_attackmitigator")

                        return Empty(message="Attack Mitigator not found")
                else:
                    logging.info("No attack detected")

                    if cryptomining_detector_output[i]["tag_name"] != "Crypto":
                        self.y_pred.append(0)
                        if connection_info.ip_o not in ATTACK_IPS and connection_info.ip_d not in ATTACK_IPS:
                            self.correct_predictions += 1
                            self.y_true.append(0)
                        else:
                            LOGGER.debug("False negative: {}".format(connection_info))
                            self.false_negatives.value = self.false_negatives.value + 1
                            self.y_true.append(1)

                        self.total_predictions.value = self.total_predictions.value + 1

                    # return Empty(message="Ok, information received (no attack detected)")
            
            self.active_requests = []
            return Empty(message="Ok, metrics processed")
            
        return Empty(message="Ok, information received")

    def analyze_prediction_accuracy(self):
        #LOGGER.info("Number of Attack Connections Correctly Classified: {}".format(self.correct_attack_conns))
        #LOGGER.info("Number of Attack Connections: {}".format(len(self.attack_connections)))

        if self.total_predictions.value > 0:
            self.overall_detection_acc.value = self.correct_predictions / self.total_predictions.value
        else:
            self.overall_detection_acc.value = 0

        #LOGGER.info("Overall Detection Accuracy: {}\n".format(self.overall_detection_acc.value))

        if len(self.attack_connections) > 0:
            self.cryptomining_attack_detection_acc.value = self.correct_attack_conns / len(self.attack_connections)
        else:
            self.cryptomining_attack_detection_acc.value = 0

        #LOGGER.info("Cryptomining Attack Detection Accuracy: {}".format(self.cryptomining_attack_detection_acc.value))
        #LOGGER.info("Cryptomining Detector Confidence: {}".format(self.confidence.value))
        
        #LOGGER.info("Time elapsed: {}".format(time.time() - TIME_START))
        
        self.attack_connections_len.value = len(self.attack_connections)
        
        self.f1_score_macro.value = metrics.f1_score(self.y_true, self.y_pred, average="macro")
        self.f1_score_weighted.value = metrics.f1_score(self.y_true, self.y_pred, average="weighted")
        self.balanced_accuracy.value = metrics.balanced_accuracy_score(self.y_true, self.y_pred)
        self.precision_score.value = metrics.precision_score(self.y_true, self.y_pred, average="macro")
        self.recall_score.value = metrics.recall_score(self.y_true, self.y_pred, average="macro")
        
        with open("prediction_accuracy.txt", "a") as f:
            #LOGGER.debug("Exporting prediction accuracy and confidence")

            f.write("Overall Detection Accuracy: {}\n".format(self.overall_detection_acc.value))
            f.write("Cryptomining Attack Detection Accuracy: {}\n".format(self.cryptomining_attack_detection_acc.value))
            f.write("Total Predictions: {}\n".format(self.total_predictions.value))
            f.write("Total Positives: {}\n".format(len(self.attack_connections)))
            f.write("False Positives: {}\n".format(self.false_positives.value))
            f.write("True Negatives: {}\n".format(self.total_predictions.value - len(self.attack_connections)))
            f.write("False Negatives: {}\n".format(self.false_negatives.value))
            f.write("Cryptomining Detector Confidence: {}\n\n".format(self.confidence.value))
            f.write("Timestamp: {}\n".format(datetime.now().strftime("%d/%m/%Y %H:%M:%S")))
            f.close()        
            
    def generate_accuracy_scalability_csv(self):
        LOGGER.debug("Starting async prediction accuracy analysis")
        LOGGER.debug("Correct csv load: {}".format(os.path.exists("/var/teraflow/scalability_accuracy.csv")))
        
        timestamp_start = datetime.now().strftime("%d/%m/%Y-%H:%M:%S")
        
        # Wait for the system to stabilize
        time.sleep(self.time_to_stabilize * 60)
        
        np.save("y_true_max_conn_{}_time_to_stabilize_{}_exp2.npy".format(self.max_connection_time, self.time_to_stabilize), self.y_true)
        np.save("y_pred_max_conn_{}_time_to_stabilize_{}_exp2.npy".format(self.max_connection_time, self.time_to_stabilize), self.y_pred)
        
        LOGGER.debug("Scalability csv started for Exp 2\n")
        with open("/var/teraflow/scalability_accuracy.csv", 'a', newline='') as f:
            spamwriter = csv.writer(f, delimiter=' ', quotechar='|', quoting=csv.QUOTE_MINIMAL)
            '''spamwriter.writerow(['TIME_CONS', 'OVERALL_ACCURACY', 'F1_SCORE_MACRO', 'F1_SCORE_WEIGHTED',
                                'BALANCED_ACCURACY', 'PRECISION_SCORE', 'RECALL_SCORE','OVERALL_ACCURACY'
                                'CRYPTO_ACCURACY', 'TOTAL_PREDICTIONS', 'TOTAL_POSITIVES', 'F_POSITIVES', 
                                'T_NEGATIVES', 'F_NEGATIVES', 'CONFIDENCE', 'TIMESTAMP_START', 'TIMESTAMP_FINISH', 'TIME_TO_STABILIZE'])'''
            
            spamwriter.writerow([self.max_connection_time, self.f1_score_macro.value, self.f1_score_weighted.value,
                                self.balanced_accuracy.value, self.precision_score.value, self.recall_score.value,
                                self.overall_detection_acc.value, self.cryptomining_attack_detection_acc.value,
                                self.total_predictions.value, self.attack_connections_len.value, self.false_positives.value,
                                self.total_predictions.value - self.attack_connections_len.value, self.false_negatives.value,
                                self.confidence.value, timestamp_start, datetime.now().strftime("%d/%m/%Y-%H:%M:%S"), self.time_to_stabilize])
            
            f.close()
        
    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
    def GetScalabilityConfig(self, request, context):
        LOGGER.info("Received scalability config request for Exp 2")
        
        self.max_connection_time = request.max_connection_time
        self.time_to_stabilize = request.time_to_stabilize
        
        # Start process to generate accuracy scalability csv asynchronically
        p = Process(target=self.generate_accuracy_scalability_csv)
        p.start()
        
        return Empty(message="CSV generated")

    def AnalyzeBatchConnectionStatistics(self, request, context):
        batch_time_start = time.time()

        for metric in request.metrics:
            self.AnalyzeConnectionStatistics(metric, context)
            
        batch_time_end = time.time()

        with open("batch_time.txt", "a") as f:
            f.write(str(len(request.metrics)) + "\n")
            f.write(str(batch_time_end - batch_time_start) + "\n\n")
            f.close()

        #logging.debug("Metrics: " + str(len(request.metrics)))
        #logging.debug("Batch time: " + str(batch_time_end - batch_time_start))

        return Empty(message="OK, information received.")

    """
    Send features allocated in the metadata of the onnx file to the DAD
        -output: ONNX metadata as a list of integers
    """
    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
    def GetFeaturesIds(self, request: Empty, context):
        features = AutoFeatures()

        for feature in self.cryptomining_detector_features_metadata:
            features.auto_features.append(feature)

        return features