L3 CyberSecurity:

- Replace custom Empty message by StatusMessage
- Pre-merge code cleanup

proto/l3_attackmitigator.proto

@@ -13,15 +13,14 @@
 // limitations under the License.
 
 syntax = "proto3";
+package l3_attackmitigator;
 
 import "context.proto";
+import "l3_centralizedattackdetector.proto";
 
 service L3Attackmitigator{
-  // Perform Mitigation
-  rpc PerformMitigation (L3AttackmitigatorOutput) returns (context.Empty) {}
-  // Get Mitigation
+  rpc PerformMitigation (L3AttackmitigatorOutput) returns (l3_centralizedattackdetector.StatusMessage) {}
   rpc GetMitigation (context.Empty) returns (context.Empty) {}
-  // Get Configured ACL Rules
   rpc GetConfiguredACLRules (context.Empty) returns (ACLRules) {}
 }
 

proto/l3_centralizedattackdetector.proto

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 syntax = "proto3";
+package l3_centralizedattackdetector;
 
 import "context.proto";
 
@@ -26,8 +27,9 @@ service L3Centralizedattackdetector {
   // Get the list of features used by the ML model in the CAD component
   rpc GetFeaturesIds (context.Empty) returns (AutoFeatures) {}
 
-  // Sets the list of attack IPs in order to be used to compute the prediction accuracy of the ML model in the CAD component in case of testing the ML model
-  rpc SetAttackIPs (AttackIPs) returns (Empty) {}
+  // Sets the list of attack IPs in order to be used to compute the prediction accuracy of the
+  // ML model in the CAD component in case of testing the ML model.
+  rpc SetAttackIPs (AttackIPs) returns (context.Empty) {}
 }
 
 message Feature {

src/l3_attackmitigator/README.md

-# l3_attackmitigator
-- Receives detected attacks from the Centralized Attack Detector component and performs the necessary mitigations.
-- Functions: 
-  - PerformMitigation(self, request: L3AttackmitigatorOutput)
-  - GetMitigation(self, request: Empty)
-  - GetConfiguredACLRules(self, request: Empty)
+# L3 Attack Mitigator
+
+Receives detected attacks from the Centralized Attack Detector component and performs the necessary mitigations.
+
+## Functions:
+- PerformMitigation(L3AttackmitigatorOutput) -> StatusMessage
+- GetMitigation(Empty) -> Empty
+- GetConfiguredACLRules(Empty) -> ACLRules

src/l3_attackmitigator/client/l3_attackmitigatorClient.py

@@ -15,17 +15,12 @@
 import grpc, logging
 from common.Constants import ServiceNameEnum
 from common.Settings import get_service_host, get_service_port_grpc
+from common.proto.context_pb2 import Empty
+from common.proto.l3_attackmitigator_pb2 import L3AttackmitigatorOutput, ACLRules
+from common.proto.l3_attackmitigator_pb2_grpc import L3AttackmitigatorStub
+from common.proto.l3_centralizedattackdetector_pb2 import StatusMessage
 from common.tools.client.RetryDecorator import retry, delay_exponential
-from common.proto.l3_attackmitigator_pb2_grpc import (
-    L3AttackmitigatorStub,
-)
-from common.proto.l3_attackmitigator_pb2 import (
-    L3AttackmitigatorOutput, ACLRules
-)
-
-from common.proto.context_pb2 import (
-    Empty
-)
+from common.tools.grpc.Tools import grpc_message_to_json_string
 
 LOGGER = logging.getLogger(__name__)
 MAX_RETRIES = 15
@@ -37,7 +32,7 @@ class l3_attackmitigatorClient:
         if not host: host = get_service_host(ServiceNameEnum.L3_AM)
         if not port: port = get_service_port_grpc(ServiceNameEnum.L3_AM)
         self.endpoint = "{}:{}".format(host, port)
-        LOGGER.debug("Creating channel to {}...".format(self.endpoint))
+        LOGGER.debug("Creating channel to {:s}...".format(self.endpoint))
         self.channel = None
         self.stub = None
         self.connect()
@@ -54,23 +49,22 @@ class l3_attackmitigatorClient:
         self.stub = None
 
     @RETRY_DECORATOR
-    def PerformMitigation(self, request: L3AttackmitigatorOutput) -> Empty:
-        LOGGER.debug('PerformMitigation request: {}'.format(request))
+    def PerformMitigation(self, request: L3AttackmitigatorOutput) -> StatusMessage:
+        LOGGER.debug('PerformMitigation request: {:s}'.format(grpc_message_to_json_string(request)))
         response = self.stub.PerformMitigation(request)
-        LOGGER.debug('PerformMitigation result: {}'.format(response))
+        LOGGER.debug('PerformMitigation result: {:s}'.format(grpc_message_to_json_string(response)))
         return response
     
     @RETRY_DECORATOR
     def GetMitigation(self, request: Empty) -> Empty:
-        LOGGER.debug('GetMitigation request: {}'.format(request))
+        LOGGER.debug('GetMitigation request: {:s}'.format(grpc_message_to_json_string(request)))
         response = self.stub.GetMitigation(request)
-        LOGGER.debug('GetMitigation result: {}'.format(response))
+        LOGGER.debug('GetMitigation result: {:s}'.format(grpc_message_to_json_string(response)))
         return response
     
     @RETRY_DECORATOR
     def GetConfiguredACLRules(self, request: Empty) -> ACLRules:
-        LOGGER.debug('GetConfiguredACLRules request: {}'.format(request))
+        LOGGER.debug('GetConfiguredACLRules request: {:s}'.format(grpc_message_to_json_string(request)))
         response = self.stub.GetConfiguredACLRules(request)
-        LOGGER.debug('GetConfiguredACLRules result: {}'.format(response))
+        LOGGER.debug('GetConfiguredACLRules result: {:s}'.format(grpc_message_to_json_string(response)))
         return response
-

src/l3_attackmitigator/service/l3_attackmitigatorServiceServicerImpl.py

@@ -13,24 +13,21 @@
 # limitations under the License.
 
 from __future__ import print_function
+
+import grpc
 import logging
 import time
 
-from common.proto.l3_centralizedattackdetector_pb2 import Empty
-from common.proto.l3_attackmitigator_pb2_grpc import L3AttackmitigatorServicer
-from common.proto.l3_attackmitigator_pb2 import ACLRules
-from common.proto.context_pb2 import (
-    ServiceId,
-    ConfigActionEnum,
-)
-
+from common.method_wrappers.Decorator import MetricsPool, safe_and_metered_rpc_method
 from common.proto.acl_pb2 import AclForwardActionEnum, AclLogActionEnum, AclRuleTypeEnum
-from common.proto.context_pb2 import ConfigActionEnum, Service, ServiceId, ConfigRule
+from common.proto.context_pb2 import ConfigActionEnum, Empty, Service, ServiceId
+from common.proto.l3_attackmitigator_pb2 import ACLRules, L3AttackmitigatorOutput
+from common.proto.l3_attackmitigator_pb2_grpc import L3AttackmitigatorServicer
+from common.proto.l3_centralizedattackdetector_pb2 import StatusMessage
 from common.tools.grpc.Tools import grpc_message_to_json_string
 from context.client.ContextClient import ContextClient
 from service.client.ServiceClient import ServiceClient
 
-from common.method_wrappers.Decorator import MetricsPool, safe_and_metered_rpc_method
 
 LOGGER = logging.getLogger(__name__)
 METRICS_POOL = MetricsPool("l3_attackmitigator", "RPC")
@@ -150,7 +147,7 @@ class l3_attackmitigatorServiceServicerImpl(L3AttackmitigatorServicer):
             raise Exception("Service update failed. Wrong ServiceId was returned")
 
     @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
-    def PerformMitigation(self, request, context):
+    def PerformMitigation(self, request : L3AttackmitigatorOutput, context : grpc.ServicerContext) -> StatusMessage:
         """
         Performs mitigation on an attack by configuring an ACL rule to block undesired TCP traffic.
 
@@ -159,7 +156,8 @@ class l3_attackmitigatorServiceServicerImpl(L3AttackmitigatorServicer):
             context (Empty): The context of the request.
 
         Returns:
-            Empty: An empty response indicating that the attack mitigation information was received and processed.
+            StatusMessage: A response with a message indicating that the attack mitigation information
+                was received and processed.
         """
 
         last_value = request.confidence
@@ -217,10 +215,10 @@ class l3_attackmitigatorServiceServicerImpl(L3AttackmitigatorServicer):
             )
         )
 
-        return Empty(message=f"OK, received values: {last_tag} with confidence {last_value}.")
+        return StatusMessage(message=f"OK, received values: {last_tag} with confidence {last_value}.")
 
     @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
-    def GetConfiguredACLRules(self, request, context):
+    def GetConfiguredACLRules(self, request : Empty, context : grpc.ServicerContext) -> ACLRules:
         """
         Returns the configured ACL rules.
 

src/l3_centralizedattackdetector/README.md

-# l3_centralizedattackdetector
-- Receives snapshot statistics from Distributed Attack Detector component and performs an inference to detect attacks. It then sends the detected attacks to the Attack Mitigator component for them to be mitigated.
-- Functions: 
-  - AnalyzeConnectionStatistics(self, request: L3CentralizedattackdetectorMetrics)
-  - AnalyzeBatchConnectionStatistics(self, request: L3CentralizedattackdetectorBatchInput)
-  - GetFeaturesIds(self, request: Empty)
+# L3 Centralized Attack Detector
+
+Receives snapshot statistics from Distributed Attack Detector component and performs an inference to detect attacks.
+It then sends the detected attacks to the Attack Mitigator component for them to be mitigated.
+
+## Functions: 
+- AnalyzeConnectionStatistics(L3CentralizedattackdetectorMetrics) -> StatusMessage
+- AnalyzeBatchConnectionStatistics(L3CentralizedattackdetectorBatchInput) -> StatusMessage
+- GetFeaturesIds(Empty) -> AutoFeatures
+- SetAttackIPs(AttackIPs) -> Empty

src/l3_centralizedattackdetector/client/l3_centralizedattackdetectorClient.py

@@ -13,18 +13,17 @@
 # limitations under the License.
 
 import grpc, logging
-from common.tools.client.RetryDecorator import retry, delay_exponential
-from common.proto.l3_centralizedattackdetector_pb2_grpc import (
-    L3CentralizedattackdetectorStub,
-)
+from common.proto.context_pb2 import Empty
+from common.proto.l3_centralizedattackdetector_pb2_grpc import L3CentralizedattackdetectorStub
 from common.proto.l3_centralizedattackdetector_pb2 import (
+    AttackIPs,
     AutoFeatures,
-    Empty,
     L3CentralizedattackdetectorBatchInput,
     L3CentralizedattackdetectorMetrics,
-    ModelInput,
-    ModelOutput
+    StatusMessage
 )
+from common.tools.client.RetryDecorator import retry, delay_exponential
+from common.tools.grpc.Tools import grpc_message_to_json_string
 
 LOGGER = logging.getLogger(__name__)
 MAX_RETRIES = 15
@@ -34,7 +33,7 @@ RETRY_DECORATOR = retry(max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION,
 class l3_centralizedattackdetectorClient:
     def __init__(self, address, port):
         self.endpoint = "{}:{}".format(address, port)
-        LOGGER.debug("Creating channel to {}...".format(self.endpoint))
+        LOGGER.debug("Creating channel to {:s}...".format(self.endpoint))
         self.channel = None
         self.stub = None
         self.connect()
@@ -51,24 +50,29 @@ class l3_centralizedattackdetectorClient:
         self.stub = None
 
     @RETRY_DECORATOR
-    def AnalyzeConnectionStatistics(self, request: L3CentralizedattackdetectorMetrics) -> Empty:
-        LOGGER.debug('AnalyzeConnectionStatistics request: {}'.format(request))
+    def AnalyzeConnectionStatistics(self, request : L3CentralizedattackdetectorMetrics) -> StatusMessage:
+        LOGGER.debug('AnalyzeConnectionStatistics request: {:s}'.format(grpc_message_to_json_string(request)))
         response = self.stub.AnalyzeConnectionStatistics(request)
-        LOGGER.debug('AnalyzeConnectionStatistics result: {}'.format(response))
+        LOGGER.debug('AnalyzeConnectionStatistics result: {:s}'.format(grpc_message_to_json_string(response)))
         return response
-    
+
     @RETRY_DECORATOR
-    def AnalyzeBatchConnectionStatistics(self, request: L3CentralizedattackdetectorBatchInput) -> Empty:
-        LOGGER.debug('AnalyzeBatchConnectionStatistics request: {}'.format(request))
-        response = self.stub.GetOutput(request)
-        LOGGER.debug('AnalyzeBatchConnectionStatistics result: {}'.format(response))
+    def AnalyzeBatchConnectionStatistics(self, request: L3CentralizedattackdetectorBatchInput) -> StatusMessage:
+        LOGGER.debug('AnalyzeBatchConnectionStatistics request: {:s}'.format(grpc_message_to_json_string(request)))
+        response = self.stub.AnalyzeBatchConnectionStatistics(request)
+        LOGGER.debug('AnalyzeBatchConnectionStatistics result: {:s}'.format(grpc_message_to_json_string(response)))
         return response
-    
+
     @RETRY_DECORATOR
-    def GetFeaturesIds(self, request: Empty) -> AutoFeatures:
-        LOGGER.debug('GetFeaturesIds request: {}'.format(request))
-        response = self.stub.GetOutput(request)
-        LOGGER.debug('GetFeaturesIds result: {}'.format(response))
+    def GetFeaturesIds(self, request : Empty) -> AutoFeatures:
+        LOGGER.debug('GetFeaturesIds request: {:s}'.format(grpc_message_to_json_string(request)))
+        response = self.stub.GetFeaturesIds(request)
+        LOGGER.debug('GetFeaturesIds result: {:s}'.format(grpc_message_to_json_string(response)))
         return response
 
-
+    @RETRY_DECORATOR
+    def SetAttackIPs(self, request : AttackIPs) -> Empty:
+        LOGGER.debug('SetAttackIPs request: {:s}'.format(grpc_message_to_json_string(request)))
+        response = self.stub.SetAttackIPs(request)
+        LOGGER.debug('SetAttackIPs result: {:s}'.format(grpc_message_to_json_string(response)))
+        return response

src/l3_centralizedattackdetector/service/l3_centralizedattackdetectorServiceServicerImpl.py

@@ -13,26 +13,27 @@
 # limitations under the License.
 
 from __future__ import print_function
-from datetime import datetime, timedelta
 
 import csv
-import os
+import grpc
+import logging
 import numpy as np
 import onnxruntime as rt
-import logging
+import os
 import time
 import uuid
 
+from datetime import datetime, timedelta
 from common.method_wrappers.Decorator import MetricsPool, safe_and_metered_rpc_method
-from common.proto.context_pb2 import Timestamp, SliceId, ConnectionId
+from common.proto.context_pb2 import Empty, Timestamp
 from common.proto.kpi_sample_types_pb2 import KpiSampleType
 from common.proto.l3_attackmitigator_pb2 import L3AttackmitigatorOutput
-from common.proto.l3_centralizedattackdetector_pb2 import Empty, AutoFeatures
+from common.proto.l3_centralizedattackdetector_pb2 import AttackIPs, AutoFeatures, L3CentralizedattackdetectorMetrics, StatusMessage
 from common.proto.l3_centralizedattackdetector_pb2_grpc import L3CentralizedattackdetectorServicer
 from common.proto.monitoring_pb2 import Kpi, KpiDescriptor
 from common.tools.timestamp.Converters import timestamp_utcnow_to_float
-from monitoring.client.MonitoringClient import MonitoringClient
 from l3_attackmitigator.client.l3_attackmitigatorClient import l3_attackmitigatorClient
+from monitoring.client.MonitoringClient import MonitoringClient
 
 
 LOGGER = logging.getLogger(__name__)
@@ -597,16 +598,22 @@ class l3_centralizedattackdetectorServiceServicerImpl(L3Centralizedattackdetecto
         return output_messages
 
     @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
-    def AnalyzeConnectionStatistics(self, request, context):
+    def AnalyzeConnectionStatistics(
+        self, request : L3CentralizedattackdetectorMetrics, context : grpc.ServicerContext
+    ) -> StatusMessage:
         """
-        Analyzes the connection statistics sent in the request, performs batch inference on the input data using the Cryptomining Attack Detector model to classify the connection as standard traffic or cryptomining attack, and notifies the Attack Mitigator component in case of attack.
+        Analyzes the connection statistics sent in the request, performs batch inference on the
+        input data using the Cryptomining Attack Detector model to classify the connection as
+        standard traffic or cryptomining attack, and notifies the Attack Mitigator component in
+        case of attack.
 
         Args:
-            request (L3CentralizedattackdetectorMetrics): A L3CentralizedattackdetectorMetrics object with connection features information.
-            context (Empty): The context of the request.
+            request (L3CentralizedattackdetectorMetrics): A L3CentralizedattackdetectorMetrics
+                object with connection features information.
+            context (grpc.ServicerContext): The context of the request.
 
         Returns:
-            Empty: An empty response indicating that the information was received and processed.
+            StatusMessage: An response indicating that the information was received and processed.
         """
 
         # Perform inference with the data sent in the request
@@ -631,7 +638,8 @@ class l3_centralizedattackdetectorServiceServicerImpl(L3Centralizedattackdetecto
             for i, req in enumerate(self.active_requests):
                 service_id = req.connection_metadata.service_id
 
-                # Check if a request of a new service has been received and, if so, create the monitored KPIs for that service
+                # 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)
                     self.service_ids.append(service_id)
@@ -743,7 +751,7 @@ class l3_centralizedattackdetectorServiceServicerImpl(L3Centralizedattackdetecto
                         LOGGER.error("Error notifying the Attack Mitigator component about the attack: ", e)
                         LOGGER.error("Couldn't find l3_attackmitigator")
 
-                        return Empty(message="Attack Mitigator not found")
+                        return StatusMessage(message="Attack Mitigator not found")
                 else:
                     LOGGER.info("No attack detected")
 
@@ -772,9 +780,9 @@ class l3_centralizedattackdetectorServiceServicerImpl(L3Centralizedattackdetecto
                 writer = csv.writer(file)
                 writer.writerow(col_values)
 
-            return Empty(message="Ok, metrics processed")
+            return StatusMessage(message="Ok, metrics processed")
 
-        return Empty(message="Ok, information received")
+        return StatusMessage(message="Ok, information received")
 
     def analyze_prediction_accuracy(self, confidence):
         """
@@ -820,16 +828,21 @@ class l3_centralizedattackdetectorServiceServicerImpl(L3Centralizedattackdetecto
             f.close()
 
     @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
-    def AnalyzeBatchConnectionStatistics(self, request, context):
+    def AnalyzeBatchConnectionStatistics(
+        self, request : L3CentralizedattackdetectorBatchInput, context : grpc.ServicerContext
+    ) -> StatusMessage:
         """
-        Analyzes a batch of connection statistics sent in the request, performs batch inference on the input data using the Cryptomining Attack Detector model to classify the connection as standard traffic or cryptomining attack, and notifies the Attack Mitigator component in case of attack.
+        Analyzes a batch of connection statistics sent in the request, performs batch inference on the
+        input data using the Cryptomining Attack Detector model to classify the connection as standard
+        traffic or cryptomining attack, and notifies the Attack Mitigator component in case of attack.
 
         Args:
-            request (L3CentralizedattackdetectorBatchMetrics): A L3CentralizedattackdetectorBatchMetrics object with connection features information.
-            context (Empty): The context of the request.
+            request (L3CentralizedattackdetectorBatchInput): A L3CentralizedattackdetectorBatchInput
+                object with connection features information.
+            context (grpc.ServicerContext): The context of the request.
 
         Returns:
-            Empty: An empty response indicating that the information was received and processed.
+            StatusMessage: An StatusMessage indicating that the information was received and processed.
         """
 
         batch_time_start = time.time()
@@ -846,16 +859,16 @@ class l3_centralizedattackdetectorServiceServicerImpl(L3Centralizedattackdetecto
         LOGGER.debug(f"Batch time: {batch_time_end - batch_time_start}")
         LOGGER.debug("Batch time: {}".format(batch_time_end - batch_time_start))
 
-        return Empty(message="OK, information received.")
+        return StatusMessage(message="OK, information received.")
 
     @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
-    def GetFeaturesIds(self, request, context):
+    def GetFeaturesIds(self, request : Empty, context : grpc.ServicerContext) -> AutoFeatures:
         """
         Returns a list of feature IDs used by the Cryptomining Attack Detector model.
 
         Args:
             request (Empty): An empty request object.
-            context (Empty): The context of the request.
+            context (grpc.ServicerContext): The context of the request.
 
         Returns:
             features_ids (AutoFeatures): A list of feature IDs used by the Cryptomining Attack Detector model.
@@ -869,19 +882,20 @@ class l3_centralizedattackdetectorServiceServicerImpl(L3Centralizedattackdetecto
         return features_ids
 
     @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
-    def SetAttackIPs(self, request, context):
+    def SetAttackIPs(self, request : AttackIPs, context : grpc.ServicerContext) -> Empty:
         """
-        Sets the list of attack IPs in order to be used to compute the prediction accuracy of the Centralized Attack Detector in case of testing the ML model.
+        Sets the list of attack IPs in order to be used to compute the prediction accuracy of the
+        Centralized Attack Detector in case of testing the ML model.
 
         Args:
             request (AttackIPs): A list of attack IPs.
-            context (Empty): The context of the request.
+            context (grpc.ServicerContext): The context of the request.
 
         Returns:
-            None
+            empty (Empty): An empty response object.
         """
 
         self.attack_ips = request.attack_ips
         LOGGER.debug(f"Succesfully set attack IPs: {self.attack_ips}")
 
-        return Empty(message="Attack IPs set.")
+        return Empty()

src/l3_distributedattackdetector/service/l3_distributedattackdetector.py

@@ -12,29 +12,27 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import asyncio
+import grpc
 import logging
-from sys import stdout
+import numpy as np
 import os
-import time
 import signal
-import grpc
-import numpy as np
-import asyncio
-from common.proto import L3CentralizedattackdetectorStub
+import time
+from sys import stdout
+from common.proto.context_pb2 import (
+    Empty,
+    ServiceTypeEnum,
+    ContextId,
+)
+from common.proto.context_pb2_grpc import ContextServiceStub
 from common.proto.l3_centralizedattackdetector_pb2 import (
     L3CentralizedattackdetectorMetrics,
     L3CentralizedattackdetectorBatchInput,
     ConnectionMetadata,
     Feature,
-    Empty,
-)
-
-from common.proto.context_pb2 import (
-    ServiceTypeEnum,
-    ContextId,
 )
-
-from common.proto.context_pb2_grpc import ContextServiceStub
+from common.proto.l3_centralizedattackdetector_pb2_grpc import L3CentralizedattackdetectorStub
 
 #  Setup LOGGER
 LOGGER = logging.getLogger("dad_LOGGER")
@@ -87,7 +85,7 @@ class l3_distributedattackdetector():
 
             LOGGER.info("Obtaining features...")
             self.feature_ids = self.get_features_ids()
-            LOGGER.info("Features Ids.: {0}".format(self.feature_ids))
+            LOGGER.info("Features Ids.: {:s}".format(str(self.feature_ids)))
             
             asyncio.run(self.process_traffic())
             
@@ -137,7 +135,7 @@ class l3_distributedattackdetector():
                 tstat_dirs.sort()
                 new_dir = tstat_dirs[-1]
                 tstat_file = tstat_piped + new_dir + "/log_tcp_temp_complete"
-                LOGGER.info("Following: {0}".format(tstat_file))
+                LOGGER.info("Following: {:s}".format(str(tstat_file)))
                 return tstat_file
             else:
                 LOGGER.info("No Tstat directory!")
@@ -333,7 +331,7 @@ class l3_distributedattackdetector():
         while True:
             line = next(loglines, None)
 
-            while line == None:
+            while line is None:
                 LOGGER.info("Waiting for new data...")
                 time.sleep(1 / 100)
                 line = next(loglines, None)