Commit 0b991b39 authored by Luis de la Cal's avatar Luis de la Cal
Browse files

Fixed the implementation of the monitored KPIs in the Centralized Attack Detector component

parent c6a54c84

complete_deploy.sh

0 → 100755
+1 −0
Original line number Diff line number Diff line 
source my_deploy.sh; ./deploy.sh; source tfs_runtime_env_vars.sh; ofc22/run_test_01_bootstrap.sh; ofc22/run_test_02_create_service.sh
 
 No newline at end of file

deploy.sh

+1 −1
Original line number Diff line number Diff line
@@ -248,7 +248,7 @@ for COMPONENT in $TFS_COMPONENTS; do 
    echo "Waiting for '$COMPONENT' component..."

    COMPONENT_OBJNAME=$(echo "${COMPONENT}" | sed "s/\_/-/")

    kubectl wait --namespace $TFS_K8S_NAMESPACE \

        --for='condition=available' --timeout=300s deployment/${COMPONENT_OBJNAME}service

        --for='condition=available' --timeout=10s deployment/${COMPONENT_OBJNAME}service

    printf "\n"

done

src/l3_centralizedattackdetector/service/l3_centralizedattackdetectorServiceServicerImpl.py

+210 −186
Original line number Diff line number Diff line
@@ -35,12 +35,16 @@ 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

from common.proto.context_pb2 import Timestamp, ServiceId, EndPointId



from l3_attackmitigator.client.l3_attackmitigatorClient import l3_attackmitigatorClient



from multiprocessing import Process, Queue



from google.protobuf.json_format import MessageToJson, Parse

import copy





LOGGER = logging.getLogger(__name__)

current_dir = os.path.dirname(os.path.abspath(__file__))

MODEL_FILE = os.path.join(current_dir, "ml_model/crypto_5g_rf_spider_features.onnx")
@@ -106,9 +110,7 @@ class l3_centralizedattackdetectorServiceServicerImpl(L3Centralizedattackdetecto 
        self.CRYPTO_CLASS = 1



        # start monitoring process

        self.monitoring_process = Process(

            target=self.monitoring_process, args=(self.inference_values, self.inference_results)

        )

        self.monitoring_process = Process(target=self.monitor_kpis, args=(self.inference_results,))

        self.monitoring_process.start()



    """
@@ -116,17 +118,19 @@ class l3_centralizedattackdetectorServiceServicerImpl(L3Centralizedattackdetecto 
        -input: 

            + client: Monitoring Client object where the KPI will be tracked

            + service_id: service ID where the KPI will be monitored

            + 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, client: MonitoringClient, service_id, kpi_description, kpi_sample_type):

        kpi_description: KpiDescriptor = KpiDescriptor()

        kpi_description.kpi_description = kpi_description

        kpi_description.service_id.service_uuid.uuid = service_id.service_uuid.uuid

        kpi_description.kpi_sample_type = kpi_sample_type

        new_kpi = client.SetKpi(kpi_description)

    def create_kpi(self, client: MonitoringClient, 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 = client.SetKpi(kpidescriptor)



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

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



        return new_kpi
@@ -143,26 +147,31 @@ class l3_centralizedattackdetectorServiceServicerImpl(L3Centralizedattackdetecto 
            created_kpi = self.create_kpi(

                self.monitoring_client,

                service_id,

                kpi,

                self.monitored_kpis[kpi]["description"],

                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)



    def monitor_kpis(self):

    def monitor_kpis(self, inference_results):

        while True:

            # get all information from the inference_values queue

            monitor_inference_values = []



            for i in range(self.inference_values.qsize()):

                monitor_inference_values.append(self.inference_values.get())



            try:

                # get all information from the inference_results queue

                monitor_inference_results = []



            for i in range(self.inference_results.qsize()):

                for i in range(inference_results.qsize()):

                    monitor_inference_results.append(self.inference_results.get())



                # deserialize the inference results

                for i in range(len(monitor_inference_results)):

                    monitor_inference_results[i]["output"]["service_id"] = Parse(

                        monitor_inference_results[i]["output"]["service_id"], ServiceId()

                    )

                    monitor_inference_results[i]["output"]["endpoint_id"] = Parse(

                        monitor_inference_results[i]["output"]["endpoint_id"], EndPointId()

                    )



                for service_id in self.service_ids:

                    time_interval = self.MONITORED_KPIS_TIME_INTERVAL_AGG

                    time_interval_start = datetime.utcnow()
@@ -170,7 +179,9 @@ class l3_centralizedattackdetectorServiceServicerImpl(L3Centralizedattackdetecto 


                    # L3 security status

                    kpi_security_status = Kpi()

                kpi_security_status.kpi_id.kpi_id.CopyFrom(self.monitored_kpis["l3_security_status"]["kpi_id"].kpi_id)

                    kpi_security_status.kpi_id.kpi_id.CopyFrom(

                        self.monitored_kpis["l3_security_status"]["kpi_id"].kpi_id

                    )



                    # get the output.tag of the ML model of the last aggregation time interval as indicated by the self.MONITORED_KPIS_TIME_INTERVAL_AGG variable

                    outputs_last_time_interval = []
@@ -326,6 +337,9 @@ class l3_centralizedattackdetectorServiceServicerImpl(L3Centralizedattackdetecto 
                    self.monitoring_client.IncludeKpi(kpi_unique_attackers)



                sleep(self.MONITORED_KPIS_TIME_INTERVAL_AGG)

            except KeyboardInterrupt:

                print("Exiting...")

                break



    """

    Classify connection as standard traffic or cryptomining attack and return results
@@ -393,7 +407,8 @@ class l3_centralizedattackdetectorServiceServicerImpl(L3Centralizedattackdetecto 


    def SendInput(self, request, context):

        # Store the data sent in the request

        self.inference_values.put({"request": request, "timestamp": datetime.now()})

        # Protobuff messages are NOT pickable, so we need to serialize them first

        # self.inference_values.put({"request": request, "timestamp": datetime.now()})



        # Perform inference with the data sent in the request

        logging.info("Performing inference...")
@@ -401,7 +416,16 @@ class l3_centralizedattackdetectorServiceServicerImpl(L3Centralizedattackdetecto 
        logging.info("Inference performed correctly")



        # Store the results of the inference that will be later used to monitor the KPIs

        self.inference_results.put({"output": cryptomining_detector_output, "timestamp": datetime.now()})

        # Protobuff messages are NOT pickable, so we need to serialize them first

        cryptomining_detector_output_serialized = copy.deepcopy(cryptomining_detector_output)

        cryptomining_detector_output_serialized["service_id"] = MessageToJson(

            request.service_id, preserving_proto_field_name=True

        )

        cryptomining_detector_output_serialized["endpoint_id"] = MessageToJson(

            request.endpoint_id, preserving_proto_field_name=True

        )



        self.inference_results.put({"output": cryptomining_detector_output_serialized, "timestamp": datetime.now()})



        service_id = request.service_id