Fixed L3 Cybersecurity framework

Lluis Gifre Renomrequested to merge
feat/l3-components into develop
  • Renamed custom Empty message in L3 CentralizedAttackDetector proto to StatusMessage
  • Improved documentation
  • Added helper RPC methods for testing
Edited by Lluis Gifre Renom

Merge request reports

manifests/l3_centralizedattackdetectorservice.yaml

+2 −0
Changes for manifests/l3_centralizedattackdetectorservice.yaml: 2 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -42,6 +42,8 @@ spec: 
          value: "0.5"

        - name: MONITORED_KPIS_TIME_INTERVAL_AGG

          value: "60"

        - name: TEST_ML_MODEL

          value: "0"

        readinessProbe:

          exec:

            command: ["/bin/grpc_health_probe", "-addr=:10001"]

proto/l3_attackmitigator.proto

+3 −4
Changes for proto/l3_attackmitigator.proto: 3 added lines, 4 removed lines.
Original line number Diff line number Diff line
@@ -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

+9 −0
Changes for proto/l3_centralizedattackdetector.proto: 9 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -13,6 +13,7 @@ 
// limitations under the License.



syntax = "proto3";

package l3_centralizedattackdetector;



import "context.proto";
@@ -25,6 +26,10 @@ 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 (context.Empty) {}

}



message Feature {
@@ -66,3 +71,7 @@ message L3CentralizedattackdetectorBatchInput { 
message StatusMessage {

	string message = 1;

}



message AttackIPs {

	repeated string attack_ips = 1;

}
 
 No newline at end of file

src/l3_attackmitigator/client/l3_attackmitigatorClient.py

+13 −19
Changes for src/l3_attackmitigator/client/l3_attackmitigatorClient.py: 13 added lines, 19 removed lines.
Original line number Diff line number Diff line
@@ -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

+77 −29
Changes for src/l3_attackmitigator/service/l3_attackmitigatorServiceServicerImpl.py: 77 added lines, 29 removed lines.
Original line number Diff line number Diff line
@@ -13,33 +13,39 @@ 
# 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')

METRICS_POOL = MetricsPool("l3_attackmitigator", "RPC")





class l3_attackmitigatorServiceServicerImpl(L3AttackmitigatorServicer):

    def __init__(self):

        LOGGER.info("Creating Attack Mitigator Service")

        """

        Initializes the Attack Mitigator service.



        Args:

            None.



        Returns:

            None.

        """



        LOGGER.info("Creating Attack Mitigator service")



        self.last_value = -1

        self.last_tag = 0
@@ -60,6 +66,23 @@ class l3_attackmitigatorServiceServicerImpl(L3AttackmitigatorServicer): 
        src_port: str,

        dst_port: str,

    ) -> None:

        """

        Configures an ACL rule to block undesired TCP traffic.



        Args:

            context_uuid (str): The UUID of the context.

            service_uuid (str): The UUID of the service.

            device_uuid (str): The UUID of the device.

            endpoint_uuid (str): The UUID of the endpoint.

            src_ip (str): The source IP address.

            dst_ip (str): The destination IP address.

            src_port (str): The source port.

            dst_port (str): The destination port.



        Returns:

            None.

        """



        # Create ServiceId

        service_id = ServiceId()

        service_id.context_id.context_uuid.uuid = context_uuid
@@ -107,29 +130,41 @@ class l3_attackmitigatorServiceServicerImpl(L3AttackmitigatorServicer): 
        acl_entry.action.forward_action = AclForwardActionEnum.ACLFORWARDINGACTION_DROP

        acl_entry.action.log_action = AclLogActionEnum.ACLLOGACTION_NOLOG



        LOGGER.info("ACL Rule Set: %s", grpc_message_to_json_string(acl_rule_set))

        LOGGER.info("ACL Config Rule: %s", grpc_message_to_json_string(acl_config_rule))

        LOGGER.info(f"ACL Rule Set: {grpc_message_to_json_string(acl_rule_set)}")

        LOGGER.info(f"ACL Config Rule: {grpc_message_to_json_string(acl_config_rule)}")



        # Add the ACLRuleSet to the list of configured ACLRuleSets

        self.configured_acl_config_rules.append(acl_config_rule)

        

        LOGGER.info(service_request)



        # Update the Service with the new ACL RuleSet

        service_reply: ServiceId = self.service_client.UpdateService(service_request)

        service_reply = self.service_client.UpdateService(service_request)



        LOGGER.info("Service reply: %s", grpc_message_to_json_string(service_reply))

        LOGGER.info(f"Service reply: {grpc_message_to_json_string(service_reply)}")



        if service_reply != service_request.service_id:  # pylint: disable=no-member

            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.



        Args:

            request (L3AttackmitigatorOutput): The request message containing the attack mitigation information.

            context (Empty): The context of the request.



        Returns:

            StatusMessage: A response with a message indicating that the attack mitigation information

                was received and processed.

        """



        last_value = request.confidence

        last_tag = request.tag



        LOGGER.info(

            "Attack Mitigator received attack mitigation information. Prediction confidence: %s, Predicted class: %s",

            last_value,

            last_tag,

            f"Attack Mitigator received attack mitigation information. Prediction confidence: {last_value}, Predicted class: {last_tag}"

        )



        ip_o = request.ip_o
@@ -141,21 +176,23 @@ class l3_attackmitigatorServiceServicerImpl(L3AttackmitigatorServicer): 
        counter = 0

        service_id = request.service_id



        LOGGER.info("Service Id.:\n{}".format(grpc_message_to_json_string(service_id)))



        LOGGER.info(f"Service Id.: {grpc_message_to_json_string(service_id)}")

        LOGGER.info("Retrieving service from Context")



        while sentinel:

            try:

                service = self.context_client.GetService(service_id)

                sentinel = False

            except Exception as e:

                counter = counter + 1

                LOGGER.debug("Waiting 2 seconds", counter, e)

                LOGGER.debug(f"Waiting 2 seconds for service to be available (attempt: {counter})")

                time.sleep(2)



        LOGGER.info(f"Service with Service Id.: {grpc_message_to_json_string(service_id)}\n{grpc_message_to_json_string(service)}")



        LOGGER.info(

            f"Service with Service Id.: {grpc_message_to_json_string(service_id)}\n{grpc_message_to_json_string(service)}"

        )

        LOGGER.info("Adding new rule to the service to block the attack")



        self.configure_acl_rule(

            context_uuid=service_id.context_id.context_uuid.uuid,

            service_uuid=service_id.service_uuid.uuid,
@@ -167,8 +204,8 @@ class l3_attackmitigatorServiceServicerImpl(L3AttackmitigatorServicer): 
            dst_port=port_d,

        )

        LOGGER.info("Service with new rule:\n{}".format(grpc_message_to_json_string(service)))



        LOGGER.info("Updating service with the new rule")



        self.service_client.UpdateService(service)

        service = self.context_client.GetService(service_id)
@@ -178,10 +215,21 @@ 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.



        Args:

            request (Empty): The request message.

            context (Empty): The context of the RPC call.



        Returns:

            acl_rules (ACLRules): The configured ACL rules.

        """



        acl_rules = ACLRules()



        for acl_config_rule in self.configured_acl_config_rules: