Commit c269153b authored by Carlos Natalino's avatar Carlos Natalino
Browse files

Adjusting and updating the CI file.

parent bb58b02c

src/opticalcentralizedattackdetector/.gitlab-ci.yml

+8 −13
Original line number Diff line number Diff line
@@ -10,8 +10,8 @@ build opticalcentralizedattackdetector: 
    - docker build -t "$IMAGE_NAME:$IMAGE_TAG" -f ./src/$IMAGE_NAME/Dockerfile ./src/

    - docker tag "$IMAGE_NAME:$IMAGE_TAG" "$CI_REGISTRY_IMAGE/$IMAGE_NAME:$IMAGE_TAG"

    - docker push "$CI_REGISTRY_IMAGE/$IMAGE_NAME:$IMAGE_TAG"

  after_script:

    - docker rmi $(docker images --quiet --filter=dangling=true)

  # after_script:

  #   - docker rmi $(docker images --quiet --filter=dangling=true)

  rules:

    - if: '$CI_PIPELINE_SOURCE == "merge_request_event" && ($CI_MERGE_REQUEST_TARGET_BRANCH_NAME == "develop" || $CI_MERGE_REQUEST_TARGET_BRANCH_NAME == $CI_DEFAULT_BRANCH)'

    - if: '$CI_PIPELINE_SOURCE == "push" && $CI_COMMIT_BRANCH == "develop"'
@@ -34,17 +34,15 @@ unit test opticalcentralizedattackdetector: 
  before_script:

    - docker login -u "$CI_REGISTRY_USER" -p "$CI_REGISTRY_PASSWORD" $CI_REGISTRY

    - if docker network list | grep teraflowbridge; then echo "teraflowbridge is already created"; else docker network create -d bridge teraflowbridge; fi

    - if docker container ls | grep influxdb; then docker rm -f influxdb; else echo "influxdb image is not in the system"; fi

    - if docker container ls | grep $IMAGE_NAME; then docker rm -f $IMAGE_NAME; else echo "$IMAGE_NAME image is not in the system"; fi

  script:

    - docker pull "$CI_REGISTRY_IMAGE/$IMAGE_NAME:$IMAGE_TAG"

    - docker run --name influxdb -d -p 8086:8086 -e INFLUXDB_DB=$INFLUXDB_DATABASE -e INFLUXDB_ADMIN_USER=$INFLUXDB_USER -e INFLUXDB_ADMIN_PASSWORD=$INFLUXDB_PASSWORD -e INXLUXDB_HTTP_AUTH_ENABLED=True --network=teraflowbridge --rm influxdb:1.8

    - docker run --name $IMAGE_NAME -d -p 7070:7070 --env INFLUXDB_USER=$INFLUXDB_USER --env INFLUXDB_PASSWORD=$INFLUXDB_PASSWORD --env INFLUXDB_DATABASE=$INFLUXDB_DATABASE --env INFLUXDB_HOSTNAME=influxdb -v "$PWD/src/$IMAGE_NAME/tests:/opt/results" --network=teraflowbridge  --rm $CI_REGISTRY_IMAGE/$IMAGE_NAME:$IMAGE_TAG

    - docker run --name $IMAGE_NAME -d -p 10005:10005 --network=teraflowbridge --rm $CI_REGISTRY_IMAGE/$IMAGE_NAME:$IMAGE_TAG

    - sleep 5

    - docker ps -a

    - docker exec -i $IMAGE_NAME bash -c "pytest --junitxml=/opt/results/report.xml"

    - docker exec -i $IMAGE_NAME bash -c "pytest --log-level=DEBUG --verbose $IMAGE_NAME/tests/test_unitary.py"

  after_script:

    - docker rm -f $IMAGE_NAME

    - docker rm -f  influxdb

    - docker network rm teraflowbridge

  rules:

    - if: '$CI_PIPELINE_SOURCE == "merge_request_event" && ($CI_MERGE_REQUEST_TARGET_BRANCH_NAME == "develop" || $CI_MERGE_REQUEST_TARGET_BRANCH_NAME == $CI_DEFAULT_BRANCH)'
@@ -54,12 +52,9 @@ unit test opticalcentralizedattackdetector: 
      - src/$IMAGE_NAME/Dockerfile

      - src/$IMAGE_NAME/tests/*.py

      - src/$IMAGE_NAME/tests/Dockerfile

      - manifests/$IMAGE_NAME.yaml

      - manifests/$IMAGE_NAMEservice.yaml

      - .gitlab-ci.yml

  artifacts:

      when: always

      reports:

        junit: src/$IMAGE_NAME/tests/report.xml





# Deployment of the opticalcentralizedattackdetector service in Kubernetes Cluster

deploy opticalcentralizedattackdetector:
@@ -71,7 +66,7 @@ deploy opticalcentralizedattackdetector: 
    - unit test opticalcentralizedattackdetector

    # - integ_test execute

  script:

    - 'sed -i "s/$IMAGE_NAME:.*/$IMAGE_NAME:$IMAGE_TAG/" manifests/$IMAGE_NAME.yaml'

    - 'sed -i "s/$IMAGE_NAME:.*/$IMAGE_NAME:$IMAGE_TAG/" manifests/${IMAGE_NAME}service.yaml'

    - kubectl version

    - kubectl get all

    - kubectl apply -f "manifests/$IMAGE_NAME.yaml"

src/opticalcentralizedattackdetector/Config.py

+6 −6
Original line number Diff line number Diff line
@@ -11,12 +11,12 @@ GRPC_GRACE_PERIOD = 60 
# service settings

MONITORING_INTERVAL = 2  # monitoring interval in seconds

#TODO: adjust the addresses below for the specific case

MONITORING_SERVICE_ADDRESS = 'monitoring'  # address/name of the monitoring service

# MONITORING_SERVICE_ADDRESS = '10.98.180.171'  # address/name of the monitoring service

CONTEXT_SERVICE_ADDRESS = 'contextservice'  # address/name of the monitoring service

# CONTEXT_SERVICE_ADDRESS = '10.97.61.5'  # address/name of the monitoring service

SERVICE_SERVICE_ADDRESS = 'serviceservice'  # address/name of the service service

# SERVICE_SERVICE_ADDRESS = '10.97.59.81'  # address/name of the service service

# MONITORING_SERVICE_ADDRESS = 'monitoring'  # address/name of the monitoring service

MONITORING_SERVICE_ADDRESS = '10.103.245.101'  # address/name of the monitoring service

# CONTEXT_SERVICE_ADDRESS = 'contextservice'  # address/name of the monitoring service

CONTEXT_SERVICE_ADDRESS = '10.108.81.135'  # address/name of the monitoring service

# SERVICE_SERVICE_ADDRESS = 'serviceservice'  # address/name of the service service

SERVICE_SERVICE_ADDRESS = '10.97.31.9'  # address/name of the service service

INFERENCE_SERVICE_ADDRESS = 'dbscanserving'  # address/name of the inference service

# INFERENCE_SERVICE_ADDRESS = 'localhost'  # address/name of the inference service

src/opticalcentralizedattackdetector/service/OpticalCentralizedAttackDetectorServiceServicerImpl.py

+56 −2
Original line number Diff line number Diff line 
import grpc, logging

import os, grpc, logging

from influxdb import InfluxDBClient

from common.rpc_method_wrapper.Decorator import create_metrics, safe_and_metered_rpc_method

from opticalcentralizedattackdetector.proto.context_pb2 import Empty, Service

from context.client.ContextClient import ContextClient

from context.Config import GRPC_SERVICE_PORT as CONTEXT_GRPC_SERVICE_PORT

from monitoring.client.monitoring_client import MonitoringClient

from monitoring.Config import GRPC_SERVICE_PORT as MONITORING_GRPC_SERVICE_PORT

from service.client.ServiceClient import ServiceClient

from service.Config import GRPC_SERVICE_PORT as SERVICE_GRPC_SERVICE_PORT

from opticalcentralizedattackdetector.proto.context_pb2 import (Empty,

    Context,  ContextId,  ContextIdList,  ContextList,

    Service,  ServiceId,  ServiceIdList,  ServiceList

)

from opticalcentralizedattackdetector.proto.monitoring_pb2 import KpiList

from opticalcentralizedattackdetector.proto.optical_centralized_attack_detector_pb2_grpc import (

    OpticalCentralizedAttackDetectorServiceServicer)

from opticalcentralizedattackdetector.Config import (

    CONTEXT_SERVICE_ADDRESS, SERVICE_SERVICE_ADDRESS, INFERENCE_SERVICE_ADDRESS, MONITORING_SERVICE_ADDRESS)





LOGGER = logging.getLogger(__name__)
@@ -11,11 +24,22 @@ SERVICE_NAME = 'OpticalCentralizedAttackDetector' 
METHOD_NAMES = ['NotifyServiceUpdate', 'DetectAttack', 'ReportSummarizedKpi', 'ReportKpi']

METRICS = create_metrics(SERVICE_NAME, METHOD_NAMES)



INFLUXDB_HOSTNAME = os.environ.get("INFLUXDB_HOSTNAME")

INFLUXDB_USER = os.environ.get("INFLUXDB_USER")

INFLUXDB_PASSWORD = os.environ.get("INFLUXDB_PASSWORD")

INFLUXDB_DATABASE = os.environ.get("INFLUXDB_DATABASE")





class OpticalCentralizedAttackDetectorServiceServicerImpl(OpticalCentralizedAttackDetectorServiceServicer):



    def __init__(self):

        LOGGER.debug('Creating Servicer...')

        LOGGER.debug('Servicer Created')

        self.context_client: ContextClient = ContextClient(address=CONTEXT_SERVICE_ADDRESS, port=CONTEXT_GRPC_SERVICE_PORT)

        self.monitoring_client: MonitoringClient = MonitoringClient(server=MONITORING_SERVICE_ADDRESS, port=MONITORING_GRPC_SERVICE_PORT)

        self.service_client: ServiceClient = ServiceClient(SERVICE_SERVICE_ADDRESS, SERVICE_GRPC_SERVICE_PORT)

        self.influxdb_client: InfluxDBClient = InfluxDBClient(host=MONITORING_SERVICE_ADDRESS, port=8086, username=INFLUXDB_USER, password=INFLUXDB_PASSWORD, database=INFLUXDB_DATABASE)

        



    @safe_and_metered_rpc_method(METRICS, LOGGER)

    def NotifyServiceUpdate(self, request : Service, context : grpc.ServicerContext) -> Empty:
@@ -23,6 +47,36 @@ class OpticalCentralizedAttackDetectorServiceServicerImpl(OpticalCentralizedAtta 


    @safe_and_metered_rpc_method(METRICS, LOGGER)

    def DetectAttack(self, request : Empty, context : grpc.ServicerContext) -> Empty:

        

        # retrieve list with current contexts

        # import pdb; pdb.set_trace()

        context_ids: ContextIdList = self.context_client.ListContextIds(Empty())



        # for each context, retrieve list of current services

        services = []

        for context_id in context_ids.context_ids:



            context_services: ServiceIdList = self.context_client.ListServices(context_id)

            for service in context_services.services:

                services.append(service)



        # get monitoring data for each of the current services

        results = self.influxdb_client.query('select * from samples;')

        print(results.raw)



        for service in services:

            for endpoint in service.service_endpoint_ids:

                # get instant KPI for this endpoint

                LOGGER.warning(f'service: {service.service_id.service_uuid.uuid}\t endpoint: {endpoint.endpoint_uuid.uuid}\tdevice: {endpoint.device_id.device_uuid.uuid}')

                # how to get all KPIs for a particular device?

                points = results.get_points(tags={'device_id': endpoint.device_id.device_uuid.uuid})

                print('points:', points)

                for point in points:

                    print('\t', point)



        # run attack detection for every service



        # if attack is detected, run the attack mitigator

        return Empty()



    @safe_and_metered_rpc_method(METRICS, LOGGER)

src/opticalcentralizedattackdetector/service/__main__.py

+7 −47
Original line number Diff line number Diff line 
import os, logging, signal, sys, time, threading, multiprocessing

from prometheus_client import start_http_server

from influxdb import InfluxDBClient



from common.Settings import get_setting

from opticalcentralizedattackdetector.Config import (

    GRPC_SERVICE_PORT, GRPC_MAX_WORKERS, GRPC_GRACE_PERIOD, LOG_LEVEL, METRICS_PORT,

    MONITORING_INTERVAL, CONTEXT_SERVICE_ADDRESS, SERVICE_SERVICE_ADDRESS, INFERENCE_SERVICE_ADDRESS, MONITORING_SERVICE_ADDRESS)

from context.Config import GRPC_SERVICE_PORT as CONTEXT_GRPC_SERVICE_PORT

from context.client.ContextClient import ContextClient

    MONITORING_INTERVAL)

from opticalcentralizedattackdetector.proto.context_pb2 import (Empty,

    Context,  ContextId,  ContextIdList,  ContextList,

    Service,  ServiceId,  ServiceIdList,  ServiceList

)

from monitoring.Config import GRPC_SERVICE_PORT as MONITORING_GRPC_SERVICE_PORT

from monitoring.client.monitoring_client import MonitoringClient

from service.Config import GRPC_SERVICE_PORT as SERVICE_GRPC_SERVICE_PORT

from service.client.ServiceClient import ServiceClient

from opticalcentralizedattackdetector.service.OpticalCentralizedAttackDetectorService import OpticalCentralizedAttackDetectorService

from opticalcentralizedattackdetector.client.OpticalCentralizedAttackDetectorClient import OpticalCentralizedAttackDetectorClient



terminate = threading.Event()

LOGGER = None



INFLUXDB_HOSTNAME = os.environ.get("INFLUXDB_HOSTNAME")

INFLUXDB_USER = os.environ.get("INFLUXDB_USER")

INFLUXDB_PASSWORD = os.environ.get("INFLUXDB_PASSWORD")

INFLUXDB_DATABASE = os.environ.get("INFLUXDB_DATABASE")

client: OpticalCentralizedAttackDetectorClient = None



def signal_handler(signal, frame): # pylint: disable=redefined-outer-name

    LOGGER.warning('Terminate signal received')
@@ -32,45 +23,14 @@ def signal_handler(signal, frame): # pylint: disable=redefined-outer-name 


def detect_attack(monitoring_interval):

    LOGGER.info("Starting the attack detection loop")

    context_client: ContextClient = ContextClient(address=CONTEXT_SERVICE_ADDRESS, port=CONTEXT_GRPC_SERVICE_PORT)

    monitoring_client: MonitoringClient = MonitoringClient(server=MONITORING_SERVICE_ADDRESS, port=MONITORING_GRPC_SERVICE_PORT)

    service_client: ServiceClient = ServiceClient(SERVICE_SERVICE_ADDRESS, SERVICE_GRPC_SERVICE_PORT)

    influxdb_client: InfluxDBClient = InfluxDBClient(host=MONITORING_SERVICE_ADDRESS, port=8086, username=INFLUXDB_USER, password=INFLUXDB_PASSWORD, database=INFLUXDB_DATABASE)

    client = OpticalCentralizedAttackDetectorClient(address='localhost', port=GRPC_SERVICE_PORT)

    client.connect()

    while True:  # infinite loop that runs until the terminate is set

        if terminate.is_set():  # if terminate is set

            LOGGER.warning("Stopping execution...")

            context_client.close()

            service_client.close()

            client.close()

            break  # break the while and stop execution

        

        # retrieve list with current contexts

        # import pdb; pdb.set_trace()

        context_ids: ContextIdList = context_client.ListContextIds(Empty())



        # for each context, retrieve list of current services

        services = []

        for context_id in context_ids.context_ids:



            context_services: ServiceIdList = context_client.ListServices(context_id)

            for service in context_services.services:

                services.append(service)



        # get monitoring data for each of the current services

        results = influxdb_client.query('select * from samples;')

        print(results.raw)



        for service in services:

            for endpoint in service.service_endpoint_ids:

                # get instant KPI for this endpoint

                LOGGER.warning(f'service: {service.service_id.service_uuid.uuid}\t endpoint: {endpoint.endpoint_uuid.uuid}\tdevice: {endpoint.device_id.device_uuid.uuid}')

                # how to get all KPIs for a particular device?

                points = results.get_points(tags={'device_id': endpoint.device_id.device_uuid.uuid})

                print(points)



        # run attack detection for every service



        # if attack is detected, run the attack mitigator



        client.DetectAttack(Empty())

        # sleep

        LOGGER.debug("Sleeping for {} seconds...".format(monitoring_interval))

        time.sleep(monitoring_interval)