diff --git a/error_monitoring.txt b/error_monitoring.txt
new file mode 100644
index 0000000000000000000000000000000000000000..6c7df6929c86ff2734e640e52a22bceedacd5e11
--- /dev/null
+++ b/error_monitoring.txt
@@ -0,0 +1,23 @@
+[2022-09-30 10:06:58,753] {/var/teraflow/monitoring/service/MonitoringServiceServicerImpl.py:151} INFO - IncludeKpi
+INFO:monitoringservice-server:IncludeKpi
+[2022-09-30 10:06:58,754] {/var/teraflow/monitoring/service/MonitoringServiceServicerImpl.py:194} INFO - getting Kpi by KpiID
+INFO:monitoringservice-server:getting Kpi by KpiID
+[2022-09-30 10:06:58,764] {/var/teraflow/monitoring/service/MonitoringServiceServicerImpl.py:215} ERROR - GetKpiDescriptor exception
+Traceback (most recent call last):
+ File "/var/teraflow/monitoring/service/MonitoringServiceServicerImpl.py", line 196, in GetKpiDescriptor
+ kpi_db = self.sql_db.get_KPI(int(request.kpi_id.uuid))
+ValueError: invalid literal for int() with base 10: 'kpi_id {\n uuid: "17"\n}\n'
+ERROR:monitoringservice-server:GetKpiDescriptor exception
+Traceback (most recent call last):
+ File "/var/teraflow/monitoring/service/MonitoringServiceServicerImpl.py", line 196, in GetKpiDescriptor
+ kpi_db = self.sql_db.get_KPI(int(request.kpi_id.uuid))
+ValueError: invalid literal for int() with base 10: 'kpi_id {\n uuid: "17"\n}\n'
+[2022-09-30 10:06:58,780] {/var/teraflow/monitoring/service/MonitoringServiceServicerImpl.py:156} WARNING - Ignoring sample with KPIId(kpi_id {
+ uuid: "kpi_id {\n uuid: \"17\"\n}\n"
+}
+): not found in database
+WARNING:monitoringservice-server:Ignoring sample with KPIId(kpi_id {
+ uuid: "kpi_id {\n uuid: \"17\"\n}\n"
+}
+): not found in database
+[2022-09-30 10:06:58,807] {/var/teraflow/monitoring/service/MonitoringServiceServicerImpl.py:151} INFO - IncludeKpi
diff --git a/scripts/show_logs_am.sh b/scripts/show_logs_am.sh
new file mode 100755
index 0000000000000000000000000000000000000000..1cbd1994d2732f07a46ec8ec0bb1ffbad9b33a28
--- /dev/null
+++ b/scripts/show_logs_am.sh
@@ -0,0 +1,27 @@
+#!/bin/bash
+# Copyright 2021-2023 H2020 TeraFlow (https://www.teraflow-h2020.eu/)
+#
+# 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.
+
+########################################################################################################################
+# Define your deployment settings here
+########################################################################################################################
+
+# If not already set, set the name of the Kubernetes namespace to deploy to.
+export TFS_K8S_NAMESPACE=${TFS_K8S_NAMESPACE:-"tfs"}
+
+########################################################################################################################
+# Automated steps start here
+########################################################################################################################
+
+kubectl --namespace $TFS_K8S_NAMESPACE logs deployment/l3-attackmitigatorservice -c server
diff --git a/scripts/show_logs_cad.sh b/scripts/show_logs_cad.sh
new file mode 100755
index 0000000000000000000000000000000000000000..acfdad79eebf630b8e3d2e4a3f03053fe7c6e660
--- /dev/null
+++ b/scripts/show_logs_cad.sh
@@ -0,0 +1,27 @@
+#!/bin/bash
+# Copyright 2021-2023 H2020 TeraFlow (https://www.teraflow-h2020.eu/)
+#
+# 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.
+
+########################################################################################################################
+# Define your deployment settings here
+########################################################################################################################
+
+# If not already set, set the name of the Kubernetes namespace to deploy to.
+export TFS_K8S_NAMESPACE=${TFS_K8S_NAMESPACE:-"tfs"}
+
+########################################################################################################################
+# Automated steps start here
+########################################################################################################################
+
+kubectl --namespace $TFS_K8S_NAMESPACE logs deployment/l3-centralizedattackdetectorservice -c server
diff --git a/src/l3_centralizedattackdetector/.gitlab-ci.yml b/src/l3_centralizedattackdetector/.gitlab-ci.yml
index 073b7925c7f7c39e1a44766fc21167236875877d..f0e40a383e9077d35c5abf674bbb5ca98d348cfe 100644
--- a/src/l3_centralizedattackdetector/.gitlab-ci.yml
+++ b/src/l3_centralizedattackdetector/.gitlab-ci.yml
@@ -52,7 +52,7 @@ unit test l3_centralizedattackdetector:
- 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 $IMAGE_NAME -d -p 10001:10001 -v "$PWD/src/$IMAGE_NAME/tests:/opt/results" --network=teraflowbridge $CI_REGISTRY_IMAGE/$IMAGE_NAME:$IMAGE_TAG
+ - docker run --name $IMAGE_NAME -d -p 10001:10001 --env CAD_CLASSIFICATION_THRESHOLD=0.5 -v "$PWD/src/$IMAGE_NAME/tests:/opt/results" --network=teraflowbridge $CI_REGISTRY_IMAGE/$IMAGE_NAME:$IMAGE_TAG
- sleep 5
- docker ps -a
- docker logs $IMAGE_NAME
diff --git a/src/l3_centralizedattackdetector/service/l3_centralizedattackdetectorServiceServicerImpl.py b/src/l3_centralizedattackdetector/service/l3_centralizedattackdetectorServiceServicerImpl.py
index 0b55eaa956e46f7473f3b1e6770f779d08f7362f..50151efb5f0909df5c90da2c01799dd1f0688b4d 100644
--- a/src/l3_centralizedattackdetector/service/l3_centralizedattackdetectorServiceServicerImpl.py
+++ b/src/l3_centralizedattackdetector/service/l3_centralizedattackdetectorServiceServicerImpl.py
@@ -14,22 +14,22 @@
from __future__ import print_function
from datetime import datetime
+
import os
import grpc
import numpy as np
import onnxruntime as rt
import logging
+
from common.proto.l3_centralizedattackdetector_pb2 import Empty
from common.proto.l3_centralizedattackdetector_pb2_grpc import L3CentralizedattackdetectorServicer
from common.proto.l3_attackmitigator_pb2 import L3AttackmitigatorOutput
from common.proto.l3_attackmitigator_pb2_grpc import L3AttackmitigatorStub
-# KPIs and Monitoring
from common.proto.monitoring_pb2 import KpiDescriptor
from common.proto.kpi_sample_types_pb2 import KpiSampleType
-# from monitoring.client.MonitoringClient import MonitoringClient
from monitoring.client.MonitoringClient import MonitoringClient
from common.proto.monitoring_pb2 import Kpi
@@ -37,15 +37,16 @@ from common.tools.timestamp.Converters import timestamp_utcnow_to_float
from common.proto.context_pb2 import Timestamp
LOGGER = logging.getLogger(__name__)
-here = os.path.dirname(os.path.abspath(__file__))
-MODEL_FILE = os.path.join(here, "ml_model/crypto_5g_rf_spider_features.onnx")
+current_dir = os.path.dirname(os.path.abspath(__file__))
+MODEL_FILE = os.path.join(current_dir, "ml_model/crypto_5g_rf_spider_features.onnx")
classification_threshold = os.getenv("CAD_CLASSIFICATION_THRESHOLD", 0.5)
class l3_centralizedattackdetectorServiceServicerImpl(L3CentralizedattackdetectorServicer):
def __init__(self):
- LOGGER.debug("Creating Servicer...")
+ LOGGER.info("Creating Centralized Attack Detector Service")
+
self.inference_values = []
self.model = rt.InferenceSession(MODEL_FILE)
self.input_name = self.model.get_inputs()[0].name
@@ -56,11 +57,9 @@ class l3_centralizedattackdetectorServiceServicerImpl(L3Centralizedattackdetecto
self.class_probability_kpi_id = None
def create_predicted_class_kpi(self, client: MonitoringClient, service_id):
- # create kpi
kpi_description: KpiDescriptor = KpiDescriptor()
kpi_description.kpi_description = "L3 security status of service {}".format(service_id)
- # kpi_description.service_id.service_uuid.uuid = service_id
- kpi_description.service_id.service_uuid.uuid = str(service_id)
+ kpi_description.service_id.service_uuid.uuid = service_id.service_uuid.uuid
kpi_description.kpi_sample_type = KpiSampleType.KPISAMPLETYPE_UNKNOWN
new_kpi = client.SetKpi(kpi_description)
@@ -69,10 +68,9 @@ class l3_centralizedattackdetectorServiceServicerImpl(L3Centralizedattackdetecto
return new_kpi
def create_class_prob_kpi(self, client: MonitoringClient, service_id):
- # create kpi
kpi_description: KpiDescriptor = KpiDescriptor()
kpi_description.kpi_description = "L3 security status of service {}".format(service_id)
- kpi_description.service_id.service_uuid.uuid = str(service_id)
+ kpi_description.service_id.service_uuid.uuid = service_id.service_uuid.uuid
kpi_description.kpi_sample_type = KpiSampleType.KPISAMPLETYPE_UNKNOWN
new_kpi = client.SetKpi(kpi_description)
@@ -81,7 +79,6 @@ class l3_centralizedattackdetectorServiceServicerImpl(L3Centralizedattackdetecto
return new_kpi
def make_inference(self, request):
- # ML MODEL
x_data = np.array(
[
[
@@ -100,7 +97,8 @@ class l3_centralizedattackdetectorServiceServicerImpl(L3Centralizedattackdetecto
)
predictions = self.model.run([self.prob_name], {self.input_name: x_data.astype(np.float32)})[0]
- # Output format
+
+ # Gather the predicted class, the probability of that class and other relevant information required to block the attack
output_message = {
"confidence": None,
"timestamp": datetime.now().strftime("%d/%m/%Y %H:%M:%S"),
@@ -117,6 +115,7 @@ class l3_centralizedattackdetectorServiceServicerImpl(L3Centralizedattackdetecto
"time_start": request.time_start,
"time_end": request.time_end,
}
+
if predictions[0][1] >= classification_threshold:
output_message["confidence"] = predictions[0][1]
output_message["tag_name"] = "Crypto"
@@ -129,17 +128,15 @@ class l3_centralizedattackdetectorServiceServicerImpl(L3Centralizedattackdetecto
return L3AttackmitigatorOutput(**output_message)
def SendInput(self, request, context):
- # PERFORM INFERENCE WITH SENT INPUTS
- logging.debug("")
- print("Inferencing ...", flush=True)
-
- # STORE VALUES
+ # Store the data sent in the request
self.inference_values.append(request)
- # MAKE INFERENCE
+ # Perform inference with the data sent in the request
+ logging.info("Performing inference...")
output = self.make_inference(request)
+ logging.info("Inference performed correctly")
- # Monitoring
+ # Include monitored KPIs values
service_id = request.service_id
if self.predicted_class_kpi_id is None:
@@ -148,20 +145,21 @@ class l3_centralizedattackdetectorServiceServicerImpl(L3Centralizedattackdetecto
if self.class_probability_kpi_id is None:
self.class_probability_kpi_id = self.create_class_prob_kpi(self.monitoring_client, service_id)
- # Packet -> DAD -> CAD -> ML -> (2 Instantaneous Value: higher class probability, predicted class) -> Monitoring
+ # Packet Aggregation Features -> DAD -> CAD -> ML -> (2 Instantaneous Value: higher class probability, predicted class) -> Monitoring
# In addition, two counters:
# Counter 1: Total number of crypto attack connections
# Counter 2: Rate of crypto attack connections with respect to the total number of connections
+ # Predicted class KPI
kpi_class = Kpi()
- kpi_class.kpi_id.kpi_id.uuid = str(self.predicted_class_kpi_id)
+ kpi_class.kpi_id.kpi_id.CopyFrom(self.predicted_class_kpi_id.kpi_id)
kpi_class.kpi_value.int32Val = 1 if output.tag_name == "Crypto" else 0
+ # Class probability KPI
kpi_prob = Kpi()
- kpi_prob.kpi_id.kpi_id.uuid = str(self.class_probability_kpi_id)
+ kpi_prob.kpi_id.kpi_id.CopyFrom(self.class_probability_kpi_id.kpi_id)
kpi_prob.kpi_value.floatVal = output.confidence
- # timestamp = timestamp_utcnow_to_float()
timestamp = Timestamp()
timestamp.timestamp = timestamp_utcnow_to_float()
@@ -172,29 +170,37 @@ class l3_centralizedattackdetectorServiceServicerImpl(L3Centralizedattackdetecto
self.monitoring_client.IncludeKpi(kpi_prob)
if output.tag_name == "Crypto":
- # SEND INFO TO MITIGATION SERVER
+ logging.info("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:
with grpc.insecure_channel("192.168.165.78:10002") as channel:
stub = L3AttackmitigatorStub(channel)
- print("Sending to mitigator...", flush=True)
+ logging.info("Sending the connection information to the Attack Mitigator component...")
response = stub.SendOutput(output)
- # print("Response received", response, "Hola", flush=True)
- # print("Sent output to mitigator and received: ", response.message) #FIX No message received
+ logging.info(
+ "Attack Mitigator notified and received response: ", response.message
+ ) # FIX No message received
- # RETURN "OK" TO THE CALLER
return Empty(message="OK, information received and mitigator notified abou the attack")
except Exception as e:
- print("This is an exception", repr(e), flush=True)
- print("Couldnt find l3_attackmitigator", flush=True)
- return Empty(message="Mitigator Not found")
+ 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:
- print("No attack detected", flush=True)
- return Empty(message="OK, information received (no attack detected)")
+ logging.info("No attack detected")
+
+ return Empty(message="Ok, information received (no attack detected)")
def GetOutput(self, request, context):
- logging.debug("")
- print("Returing inference output...")
+ logging.info("Returning inference output...")
k = np.multiply(self.inference_values, [2])
k = np.sum(k)
+
return self.make_inference(k)
diff --git a/src/monitoring/service/ManagementDBTools.py b/src/monitoring/service/ManagementDBTools.py
new file mode 100644
index 0000000000000000000000000000000000000000..63bdb1893499ff560ff03866f31e3619070d3201
--- /dev/null
+++ b/src/monitoring/service/ManagementDBTools.py
@@ -0,0 +1,154 @@
+# Copyright 2021-2023 H2020 TeraFlow (https://www.teraflow-h2020.eu/)
+#
+# 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.
+
+import sqlite3 as sl
+
+class ManagementDB():
+ def __init__(self, database):
+ self.client = sl.connect(database, check_same_thread=False)
+ self.create_monitoring_table()
+ self.create_subscription_table()
+ self.create_alarm_table()
+
+ def create_monitoring_table(self):
+ self.client.execute("""
+ CREATE TABLE IF NOT EXISTS kpi(
+ kpi_id INTEGER PRIMARY KEY AUTOINCREMENT,
+ kpi_description TEXT,
+ kpi_sample_type INTEGER,
+ device_id INTEGER,
+ endpoint_id INTEGER,
+ service_id INTEGER
+ );
+ """)
+
+ def create_subscription_table(self):
+ self.client.execute("""
+ CREATE TABLE IF NOT EXISTS subscription(
+ subs_id INTEGER PRIMARY KEY AUTOINCREMENT,
+ kpi_id INTEGER,
+ subscriber TEXT,
+ sampling_duration_s REAL,
+ sampling_interval_s REAL,
+ start_timestamp REAL,
+ end_timestamp REAL
+ );
+ """)
+
+ def create_alarm_table(self):
+ self.client.execute("""
+ CREATE TABLE IF NOT EXISTS alarm(
+ alarm_id INTEGER PRIMARY KEY AUTOINCREMENT,
+ alarm_description TEXT,
+ alarm_name TEXT,
+ kpi_id INTEGER,
+ kpi_min_value REAL,
+ kpi_max_value REAL,
+ in_range INTEGER,
+ include_min_value INTEGER,
+ include_max_value INTEGER
+ );
+ """)
+
+ def insert_KPI(self,kpi_description,kpi_sample_type,device_id,endpoint_id,service_id):
+ c = self.client.cursor()
+ c.execute("SELECT kpi_id FROM kpi WHERE device_id is ? AND kpi_sample_type is ? AND endpoint_id is ? AND service_id is ?",(device_id,kpi_sample_type,endpoint_id,service_id))
+ data=c.fetchone()
+ if data is None:
+ c.execute("INSERT INTO kpi (kpi_description,kpi_sample_type,device_id,endpoint_id,service_id) VALUES (?,?,?,?,?)", (kpi_description,kpi_sample_type,device_id,endpoint_id,service_id))
+ self.client.commit()
+ return c.lastrowid
+ else:
+ return data[0]
+
+ def insert_subscription(self,kpi_id,subscriber,sampling_duration_s,sampling_interval_s,start_timestamp, end_timestamp):
+ c = self.client.cursor()
+ c.execute("SELECT subs_id FROM subscription WHERE kpi_id is ? AND subscriber is ? AND sampling_duration_s is ? AND sampling_interval_s is ? AND start_timestamp is ? AND end_timestamp is ?",(kpi_id,subscriber,sampling_duration_s,sampling_interval_s,start_timestamp, end_timestamp))
+ data=c.fetchone()
+ if data is None:
+ c.execute("INSERT INTO subscription (kpi_id,subscriber,sampling_duration_s,sampling_interval_s,start_timestamp, end_timestamp) VALUES (?,?,?,?,?,?)", (kpi_id,subscriber,sampling_duration_s,sampling_interval_s,start_timestamp, end_timestamp))
+ self.client.commit()
+ return c.lastrowid
+ else:
+ print("already exists")
+ return data[0]
+
+ def insert_alarm(self,alarm_description,alarm_name,kpi_id,kpi_min_value,kpi_max_value,in_range,include_min_value,include_max_value):
+ c = self.client.cursor()
+ c.execute("SELECT alarm_id FROM alarm WHERE alarm_description is ? AND alarm_name is ? AND kpi_id is ? AND kpi_min_value is ? AND kpi_max_value is ? AND in_range is ? AND include_min_value is ? AND include_max_value is ?",(alarm_description,alarm_name,kpi_id,kpi_min_value,kpi_max_value,in_range,include_min_value,include_max_value))
+ data=c.fetchone()
+ if data is None:
+ c.execute("INSERT INTO alarm (alarm_description, alarm_name, kpi_id, kpi_min_value, kpi_max_value, in_range, include_min_value, include_max_value) VALUES (?,?,?,?,?,?,?,?)", (alarm_description,alarm_name,kpi_id,kpi_min_value,kpi_max_value,in_range,include_min_value,include_max_value))
+ self.client.commit()
+ return c.lastrowid
+ else:
+ print("already exists")
+ return data[0]
+
+ def delete_KPI(self,kpi_id):
+ c = self.client.cursor()
+ c.execute("SELECT * FROM kpi WHERE kpi_id is ?",(kpi_id,))
+ data=c.fetchone()
+ if data is None:
+ return False
+ else:
+ c.execute("DELETE FROM kpi WHERE kpi_id is ?",(kpi_id,))
+ self.client.commit()
+ return True
+
+ def delete_subscription(self,subs_id):
+ c = self.client.cursor()
+ c.execute("SELECT * FROM subscription WHERE subs_id is ?",(subs_id,))
+ data=c.fetchone()
+ if data is None:
+ return False
+ else:
+ c.execute("DELETE FROM subscription WHERE subs_id is ?",(subs_id,))
+ self.client.commit()
+ return True
+
+ def delete_alarm(self,alarm_id):
+ c = self.client.cursor()
+ c.execute("SELECT * FROM alarm WHERE alarm_id is ?",(alarm_id,))
+ data=c.fetchone()
+ if data is None:
+ return False
+ else:
+ c.execute("DELETE FROM alarm WHERE alarm_id is ?",(alarm_id,))
+ self.client.commit()
+ return True
+
+ def get_KPI(self,kpi_id):
+ data = self.client.execute("SELECT * FROM kpi WHERE kpi_id is ?",(kpi_id,))
+ return data.fetchone()
+
+ def get_subscription(self,subs_id):
+ data = self.client.execute("SELECT * FROM subscription WHERE subs_id is ?",(subs_id,))
+ return data.fetchone()
+
+ def get_alarm(self,alarm_id):
+ data = self.client.execute("SELECT * FROM alarm WHERE alarm_id is ?",(alarm_id,))
+ return data.fetchone()
+
+ def get_KPIS(self):
+ data = self.client.execute("SELECT * FROM kpi")
+ return data.fetchall()
+
+ def get_subscriptions(self):
+ data = self.client.execute("SELECT * FROM subscription")
+ return data.fetchall()
+
+ def get_alarms(self):
+ data = self.client.execute("SELECT * FROM alarm")
+ return data.fetchall()