From c81494bb3770874402df10af9a3e9950bddd2da5 Mon Sep 17 00:00:00 2001 From: karamchandan Date: Thu, 13 Jul 2023 12:30:23 +0200 Subject: [PATCH 01/12] Fix scenario3/l3 deployment script --- src/tests/scenario3/l3/deploy.sh | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/tests/scenario3/l3/deploy.sh b/src/tests/scenario3/l3/deploy.sh index e8e02b026..be6c43ef7 100755 --- a/src/tests/scenario3/l3/deploy.sh +++ b/src/tests/scenario3/l3/deploy.sh @@ -13,8 +13,8 @@ # See the License for the specific language governing permissions and # limitations under the License. -source deploy_specs.sh -./deploy/all.sh +source src/tests/scenario3/l3/deploy_specs.sh +deploy/all.sh source tfs_runtime_env_vars.sh ofc22/run_test_01_bootstrap.sh ofc22/run_test_02_create_service.sh -- GitLab From 3f081d8d565517b163394532e7f09d45aa66e1f0 Mon Sep 17 00:00:00 2001 From: karamchandan Date: Thu, 13 Jul 2023 12:47:23 +0200 Subject: [PATCH 02/12] Fixed missing import in L3 Centralized Attack Detector --- .../service/l3_centralizedattackdetectorServiceServicerImpl.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/l3_centralizedattackdetector/service/l3_centralizedattackdetectorServiceServicerImpl.py b/src/l3_centralizedattackdetector/service/l3_centralizedattackdetectorServiceServicerImpl.py index 91793230d..a5c38151c 100644 --- a/src/l3_centralizedattackdetector/service/l3_centralizedattackdetectorServiceServicerImpl.py +++ b/src/l3_centralizedattackdetector/service/l3_centralizedattackdetectorServiceServicerImpl.py @@ -28,7 +28,7 @@ from common.method_wrappers.Decorator import MetricsPool, safe_and_metered_rpc_m 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 AttackIPs, AutoFeatures, L3CentralizedattackdetectorMetrics, StatusMessage +from common.proto.l3_centralizedattackdetector_pb2 import AttackIPs, AutoFeatures, L3CentralizedattackdetectorMetrics, L3CentralizedattackdetectorBatchInput, 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 -- GitLab From ce46899545c3a097a388c2756e6eec04b10641ae Mon Sep 17 00:00:00 2001 From: karamchandan Date: Thu, 13 Jul 2023 13:32:06 +0200 Subject: [PATCH 03/12] Cleanup of the DAD service implementation --- .../service/known_attack_ips.csv | 1 + .../service/l3_distributedattackdetector.py | 113 ++++++++---------- 2 files changed, 54 insertions(+), 60 deletions(-) create mode 100644 src/l3_distributedattackdetector/service/known_attack_ips.csv diff --git a/src/l3_distributedattackdetector/service/known_attack_ips.csv b/src/l3_distributedattackdetector/service/known_attack_ips.csv new file mode 100644 index 000000000..254dad83d --- /dev/null +++ b/src/l3_distributedattackdetector/service/known_attack_ips.csv @@ -0,0 +1 @@ +37.187.95.110,91.121.140.167,94.23.23.52,94.23.247.226,149.202.83.171 \ No newline at end of file diff --git a/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py b/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py index 357f44a9a..32c7c190c 100644 --- a/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py +++ b/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py @@ -13,28 +13,26 @@ # limitations under the License. import asyncio -import grpc import logging -import numpy as np import os import signal import time from sys import stdout -from common.proto.context_pb2 import ( - Empty, - ServiceTypeEnum, - ContextId, -) + +import grpc +import numpy as np + +from common.proto.context_pb2 import ContextId, Empty, ServiceTypeEnum from common.proto.context_pb2_grpc import ContextServiceStub from common.proto.l3_centralizedattackdetector_pb2 import ( - L3CentralizedattackdetectorMetrics, - L3CentralizedattackdetectorBatchInput, ConnectionMetadata, Feature, + L3CentralizedattackdetectorBatchInput, + L3CentralizedattackdetectorMetrics, ) from common.proto.l3_centralizedattackdetector_pb2_grpc import L3CentralizedattackdetectorStub -# Setup LOGGER +# Setup LOGGER LOGGER = logging.getLogger("dad_LOGGER") LOGGER.setLevel(logging.INFO) logFormatter = logging.Formatter(fmt="%(levelname)-8s %(message)s") @@ -42,8 +40,12 @@ consoleHandler = logging.StreamHandler(stdout) consoleHandler.setFormatter(logFormatter) LOGGER.addHandler(consoleHandler) +# Define constants TSTAT_DIR_NAME = "piped/" -CENTRALIZED_ATTACK_DETECTOR = "192.168.165.78:10001" +CONTROLLER_IP = "192.168.165.78" # Change this to the IP of the controller +CONTEXT_ID = "admin" # Change this to the context ID to be used +CONTEXT_CHANNEL = f"{CONTROLLER_IP}:1010" # Change this to the IP of the controller +CENTRALIZED_ATTACK_DETECTOR = f"{CONTROLLER_IP}:10001" JSON_BLANK = { "ip_o": "", # Client IP "port_o": "", # Client port @@ -54,31 +56,29 @@ JSON_BLANK = { "time_start": 0.0, # Start of connection "time_end": 0.0, # Time of last packet } - STOP = False IGNORE_FIRST_LINE_TSTAT = True - -CONTEXT_ID = "admin" -CONTEXT_CHANNEL = "192.168.165.78:1010" PROFILING = False SEND_DATA_IN_BATCHES = False BATCH_SIZE = 10 -ATTACK_IPS = ["37.187.95.110", "91.121.140.167", "94.23.23.52", "94.23.247.226", "149.202.83.171"] -class l3_distributedattackdetector(): + +class l3_distributedattackdetector: def __init__(self): LOGGER.info("Creating Distributed Attack Detector") self.feature_ids = [] - + self.cad_features = {} self.conn_id = () - - self.connections_dict = {} # Dict for storing ALL data - self.new_connections = {} # Dict for storing NEW data + + self.connections_dict = {} # Dict for storing ALL data + self.new_connections = {} # Dict for storing NEW data + + self.known_attack_ips = self.read_kwnown_attack_ips() signal.signal(signal.SIGINT, self.handler) - + with grpc.insecure_channel(CENTRALIZED_ATTACK_DETECTOR) as channel: self.cad = L3CentralizedattackdetectorStub(channel) LOGGER.info("Connected to the centralized attack detector") @@ -86,10 +86,18 @@ class l3_distributedattackdetector(): LOGGER.info("Obtaining features...") self.feature_ids = self.get_features_ids() LOGGER.info("Features Ids.: {:s}".format(str(self.feature_ids))) - + asyncio.run(self.process_traffic()) - + + def read_kwnown_attack_ips(self): + known_attack_ips = [] + # open known attack ips csv file + with open("known_attack_ips.csv", "r") as f: + known_attack_ips = f.read().splitlines() + + return known_attack_ips + def handler(self): if STOP: exit() @@ -97,34 +105,35 @@ class l3_distributedattackdetector(): STOP = True LOGGER.info("Gracefully stopping...") - - def follow(self, thefile, time_sleep): + + def follow(self, file, time_sleep): """ Generator function that yields new lines in a file - It reads the logfie (the opened file) """ + # seek the end of the file - # thefile.seek(0, os.SEEK_END) + # file.seek(0, os.SEEK_END) - trozo = "" + chunk = "" - # start infinite loop + # start an infinite loop while True: # read last line of file - line = thefile.readline() + line = file.readline() # sleep if file hasn't been updated if not line: time.sleep(time_sleep) continue + if line[-1] != "\n": - trozo += line + chunk += line else: - if trozo != "": - line = trozo + line - trozo = "" - yield line + if chunk != "": + line = chunk + line + chunk = "" + yield line def load_file(self, dirname=TSTAT_DIR_NAME): # - Client side - while True: @@ -141,7 +150,6 @@ class l3_distributedattackdetector(): LOGGER.info("No Tstat directory!") time.sleep(5) - def process_line(self, line): """ - Preprocessing before a message per line @@ -161,7 +169,6 @@ class l3_distributedattackdetector(): return values - def get_service_ids(self, context_id_str): with grpc.insecure_channel(CONTEXT_CHANNEL) as channel: stub = ContextServiceStub(channel) @@ -169,7 +176,6 @@ class l3_distributedattackdetector(): context_id.context_uuid.uuid = context_id_str return stub.ListServiceIds(context_id) - def get_services(self, context_id_str): with grpc.insecure_channel(CONTEXT_CHANNEL) as channel: stub = ContextServiceStub(channel) @@ -177,7 +183,6 @@ class l3_distributedattackdetector(): context_id.context_uuid.uuid = context_id_str return stub.ListServices(context_id) - def get_service_id(self, context_id): service_id_list = self.get_service_ids(context_id) service_id = None @@ -190,7 +195,6 @@ class l3_distributedattackdetector(): return service_id - def get_service_id2(self, context_id): service_list = self.get_services(context_id) service_id = None @@ -202,7 +206,6 @@ class l3_distributedattackdetector(): pass return service_id - def get_endpoint_id(self, context_id): service_list = self.get_services(context_id) endpoint_id = None @@ -212,11 +215,9 @@ class l3_distributedattackdetector(): break return endpoint_id - def get_features_ids(self): return self.cad.GetFeaturesIds(Empty()).auto_features - def check_types(self): for feature in self.cad_features["features"]: assert isinstance(feature, float) @@ -230,7 +231,6 @@ class l3_distributedattackdetector(): assert isinstance(self.cad_features["connection_metadata"]["time_start"], float) assert isinstance(self.cad_features["connection_metadata"]["time_end"], float) - def insert_connection(self): try: self.connections_dict[self.conn_id]["time_end"] = time.time() @@ -247,12 +247,10 @@ class l3_distributedattackdetector(): self.connections_dict[self.conn_id]["ip_d"] = self.conn_id[2] self.connections_dict[self.conn_id]["port_d"] = self.conn_id[3] - def check_if_connection_is_attack(self): - if self.conn_id[0] in ATTACK_IPS or self.conn_id[2] in ATTACK_IPS: + if self.conn_id[0] in self.known_attack_ips or self.conn_id[2] in self.known_attack_ips: LOGGER.info("Attack detected. Origin: {0}, destination: {1}".format(self.conn_id[0], self.conn_id[2])) - def create_cad_features(self): self.cad_features = { "features": self.new_connections[self.conn_id][0:10], @@ -270,7 +268,6 @@ class l3_distributedattackdetector(): }, } - async def send_batch_async(self, metrics_list_pb): loop = asyncio.get_running_loop() @@ -279,16 +276,13 @@ class l3_distributedattackdetector(): metrics_batch.metrics.extend(metrics_list_pb) # Send batch - future = loop.run_in_executor( - None, self.cad.AnalyzeBatchConnectionStatistics, metrics_batch - ) + future = loop.run_in_executor(None, self.cad.AnalyzeBatchConnectionStatistics, metrics_batch) try: await future except Exception as e: LOGGER.error(f"Error sending batch: {e}") - async def send_data(self, metrics_list_pb, send_data_times): # Send to CAD if SEND_DATA_IN_BATCHES: @@ -311,7 +305,6 @@ class l3_distributedattackdetector(): return metrics_list_pb, send_data_times - async def process_traffic(self): LOGGER.info("Loading Tstat log file...") logfile = open(self.load_file(), "r") @@ -326,7 +319,7 @@ class l3_distributedattackdetector(): metrics_list_pb = [] LOGGER.info("Starting to process data...") - + index = 0 while True: line = next(loglines, None) @@ -352,9 +345,9 @@ class l3_distributedattackdetector(): self.insert_connection() self.create_cad_features() - + self.check_types() - + connection_metadata = ConnectionMetadata(**self.cad_features["connection_metadata"]) metrics = L3CentralizedattackdetectorMetrics() @@ -369,8 +362,8 @@ class l3_distributedattackdetector(): metrics_list_pb, send_data_times = await self.send_data(metrics_list_pb, send_data_times) index = index + 1 - + process_time.append(time.time() - start) - + if num_lines % 10 == 0: - LOGGER.info(f"Number of lines: {num_lines} - Average processing time: {sum(process_time) / num_lines}") \ No newline at end of file + LOGGER.info(f"Number of lines: {num_lines} - Average processing time: {sum(process_time) / num_lines}") -- GitLab From 5967330e1219c3aef7d286da13346267ea417ade Mon Sep 17 00:00:00 2001 From: karamchandan Date: Thu, 13 Jul 2023 13:45:00 +0200 Subject: [PATCH 04/12] Cleanup of the DAD service implementation --- .../service/l3_distributedattackdetector.py | 16 ++-------------- 1 file changed, 2 insertions(+), 14 deletions(-) diff --git a/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py b/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py index 32c7c190c..52cdb0c09 100644 --- a/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py +++ b/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py @@ -94,7 +94,7 @@ class l3_distributedattackdetector: # open known attack ips csv file with open("known_attack_ips.csv", "r") as f: - known_attack_ips = f.read().splitlines() + known_attack_ips = f.read().split(",") return known_attack_ips @@ -184,18 +184,6 @@ class l3_distributedattackdetector: return stub.ListServices(context_id) def get_service_id(self, context_id): - service_id_list = self.get_service_ids(context_id) - service_id = None - for s_id in service_id_list.service_ids: - if ( - s_id.service_uuid.uuid == "0eaa0752-c7b6-4c2e-97da-317fbfee5112" - ): # TODO: Change this identifier to the L3VPN service identifier with the real router for the demo v2 - service_id = s_id - break - - return service_id - - def get_service_id2(self, context_id): service_list = self.get_services(context_id) service_id = None for s in service_list.services: @@ -241,7 +229,7 @@ class l3_distributedattackdetector: self.connections_dict[self.conn_id]["ip_o"] = self.conn_id[0] self.connections_dict[self.conn_id]["port_o"] = self.conn_id[1] self.connections_dict[self.conn_id]["flow_id"] = ":".join(self.conn_id) - self.connections_dict[self.conn_id]["service_id"] = self.get_service_id2(CONTEXT_ID) + self.connections_dict[self.conn_id]["service_id"] = self.get_service_id(CONTEXT_ID) self.connections_dict[self.conn_id]["endpoint_id"] = self.get_endpoint_id(CONTEXT_ID) self.connections_dict[self.conn_id]["protocol"] = "TCP" self.connections_dict[self.conn_id]["ip_d"] = self.conn_id[2] -- GitLab From b3c0e922bbd99be4b1dab57ec3402a08162b34c2 Mon Sep 17 00:00:00 2001 From: karamchandan Date: Thu, 13 Jul 2023 13:47:19 +0200 Subject: [PATCH 05/12] Cleanup of the DAD service implementation --- .../service/l3_distributedattackdetector.py | 17 ++++++++++------- 1 file changed, 10 insertions(+), 7 deletions(-) diff --git a/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py b/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py index 52cdb0c09..e87636952 100644 --- a/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py +++ b/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py @@ -81,9 +81,9 @@ class l3_distributedattackdetector: with grpc.insecure_channel(CENTRALIZED_ATTACK_DETECTOR) as channel: self.cad = L3CentralizedattackdetectorStub(channel) - LOGGER.info("Connected to the centralized attack detector") - - LOGGER.info("Obtaining features...") + LOGGER.info("Connected to the Centralized Attack Detector") + + LOGGER.info("Obtaining features Ids. from the Centralized Attack Detector...") self.feature_ids = self.get_features_ids() LOGGER.info("Features Ids.: {:s}".format(str(self.feature_ids))) @@ -118,10 +118,10 @@ class l3_distributedattackdetector: # start an infinite loop while True: - # read last line of file + # read last line of the file line = file.readline() - # sleep if file hasn't been updated + # sleep if the file hasn't been updated if not line: time.sleep(time_sleep) continue @@ -135,16 +135,19 @@ class l3_distributedattackdetector: yield line - def load_file(self, dirname=TSTAT_DIR_NAME): # - Client side - + def load_file(self, dirname=TSTAT_DIR_NAME): while True: here = os.path.dirname(os.path.abspath(__file__)) tstat_piped = os.path.join(here, dirname) tstat_dirs = os.listdir(tstat_piped) + if len(tstat_dirs) > 0: tstat_dirs.sort() new_dir = tstat_dirs[-1] tstat_file = tstat_piped + new_dir + "/log_tcp_temp_complete" + LOGGER.info("Following: {:s}".format(str(tstat_file))) + return tstat_file else: LOGGER.info("No Tstat directory!") @@ -165,7 +168,7 @@ class l3_distributedattackdetector: feature = feature_id - 1 values.append(float(line[feature])) except IndexError: - print("IndexError: {0}".format(line)) + LOGGER.error("IndexError: {0}".format(line)) return values -- GitLab From b81596f5f91552efbda8372b67cd4422d1e59fa6 Mon Sep 17 00:00:00 2001 From: karamchandan Date: Thu, 13 Jul 2023 13:50:07 +0200 Subject: [PATCH 06/12] Cleanup of the DAD service implementation --- .../service/l3_distributedattackdetector.py | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py b/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py index e87636952..1ed5182b9 100644 --- a/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py +++ b/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py @@ -177,6 +177,7 @@ class l3_distributedattackdetector: stub = ContextServiceStub(channel) context_id = ContextId() context_id.context_uuid.uuid = context_id_str + return stub.ListServiceIds(context_id) def get_services(self, context_id_str): @@ -184,26 +185,31 @@ class l3_distributedattackdetector: stub = ContextServiceStub(channel) context_id = ContextId() context_id.context_uuid.uuid = context_id_str + return stub.ListServices(context_id) def get_service_id(self, context_id): service_list = self.get_services(context_id) service_id = None + for s in service_list.services: if s.service_type == ServiceTypeEnum.SERVICETYPE_L3NM: service_id = s.service_id break else: pass + return service_id def get_endpoint_id(self, context_id): service_list = self.get_services(context_id) endpoint_id = None + for s in service_list.services: if s.service_type == ServiceTypeEnum.SERVICETYPE_L3NM: endpoint_id = s.service_endpoint_ids[0] break + return endpoint_id def get_features_ids(self): @@ -275,7 +281,7 @@ class l3_distributedattackdetector: LOGGER.error(f"Error sending batch: {e}") async def send_data(self, metrics_list_pb, send_data_times): - # Send to CAD + # Send data to CAD if SEND_DATA_IN_BATCHES: if len(metrics_list_pb) == BATCH_SIZE: send_data_time_start = time.time() @@ -319,24 +325,24 @@ class l3_distributedattackdetector: LOGGER.info("Waiting for new data...") time.sleep(1 / 100) line = next(loglines, None) + if index == 0 and IGNORE_FIRST_LINE_TSTAT: index = index + 1 continue + if STOP: break num_lines += 1 start = time.time() line_id = line.split(" ") + self.conn_id = (line_id[0], line_id[1], line_id[14], line_id[15]) self.new_connections[self.conn_id] = self.process_line(line) self.check_if_connection_is_attack() - self.insert_connection() - self.create_cad_features() - self.check_types() connection_metadata = ConnectionMetadata(**self.cad_features["connection_metadata"]) -- GitLab From f50c0ee6e0858cde90906623e9a9106b636b919f Mon Sep 17 00:00:00 2001 From: karamchandan Date: Thu, 13 Jul 2023 14:25:41 +0200 Subject: [PATCH 07/12] Cleanup of DAD service implementation --- .../service/l3_distributedattackdetector.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py b/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py index 1ed5182b9..14acdadec 100644 --- a/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py +++ b/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py @@ -44,7 +44,7 @@ LOGGER.addHandler(consoleHandler) TSTAT_DIR_NAME = "piped/" CONTROLLER_IP = "192.168.165.78" # Change this to the IP of the controller CONTEXT_ID = "admin" # Change this to the context ID to be used -CONTEXT_CHANNEL = f"{CONTROLLER_IP}:1010" # Change this to the IP of the controller +CONTEXT_CHANNEL = f"{CONTROLLER_IP}:1010" CENTRALIZED_ATTACK_DETECTOR = f"{CONTROLLER_IP}:10001" JSON_BLANK = { "ip_o": "", # Client IP -- GitLab From bc1d7a582dcf94501237ba79f47c3c952f0333e8 Mon Sep 17 00:00:00 2001 From: karamchandan Date: Thu, 13 Jul 2023 14:27:50 +0200 Subject: [PATCH 08/12] Cleanup of the DAD service implementation --- .../service/l3_distributedattackdetector.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py b/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py index 14acdadec..bb2607ac0 100644 --- a/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py +++ b/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py @@ -51,7 +51,7 @@ JSON_BLANK = { "port_o": "", # Client port "ip_d": "", # Server ip "port_d": "", # Server port - "flow_id": "", # Identifier:c_ip,c_port,s_ip,s_port,time_start + "flow_id": "", # Identifier: c_ip,c_port, s_ip,s_port, time_start "protocol": "", # Connection protocol "time_start": 0.0, # Start of connection "time_end": 0.0, # Time of last packet @@ -72,8 +72,8 @@ class l3_distributedattackdetector: self.cad_features = {} self.conn_id = () - self.connections_dict = {} # Dict for storing ALL data - self.new_connections = {} # Dict for storing NEW data + self.connections_dict = {} # Dictionary for storing all connections data + self.new_connections = {} # Dictionary for storing new connections data self.known_attack_ips = self.read_kwnown_attack_ips() @@ -150,7 +150,7 @@ class l3_distributedattackdetector: return tstat_file else: - LOGGER.info("No Tstat directory!") + LOGGER.info("No Tstat directory found. Waiting...") time.sleep(5) def process_line(self, line): -- GitLab From ef2960a14c3eced458d8a1cfa740dde331fa73ca Mon Sep 17 00:00:00 2001 From: karamchandan Date: Thu, 13 Jul 2023 14:30:05 +0200 Subject: [PATCH 09/12] Cleanup of the DAD service implementation --- .../service/l3_distributedattackdetector.py | 41 ++++++++++--------- 1 file changed, 21 insertions(+), 20 deletions(-) diff --git a/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py b/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py index bb2607ac0..269640538 100644 --- a/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py +++ b/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py @@ -66,7 +66,7 @@ BATCH_SIZE = 10 class l3_distributedattackdetector: def __init__(self): LOGGER.info("Creating Distributed Attack Detector") - + self.feature_ids = [] self.cad_features = {} @@ -76,26 +76,26 @@ class l3_distributedattackdetector: self.new_connections = {} # Dictionary for storing new connections data self.known_attack_ips = self.read_kwnown_attack_ips() - + signal.signal(signal.SIGINT, self.handler) with grpc.insecure_channel(CENTRALIZED_ATTACK_DETECTOR) as channel: self.cad = L3CentralizedattackdetectorStub(channel) LOGGER.info("Connected to the Centralized Attack Detector") - + LOGGER.info("Obtaining features Ids. from the Centralized Attack Detector...") self.feature_ids = self.get_features_ids() LOGGER.info("Features Ids.: {:s}".format(str(self.feature_ids))) asyncio.run(self.process_traffic()) - + def read_kwnown_attack_ips(self): known_attack_ips = [] - + # open known attack ips csv file with open("known_attack_ips.csv", "r") as f: known_attack_ips = f.read().split(",") - + return known_attack_ips def handler(self): @@ -140,14 +140,14 @@ class l3_distributedattackdetector: here = os.path.dirname(os.path.abspath(__file__)) tstat_piped = os.path.join(here, dirname) tstat_dirs = os.listdir(tstat_piped) - + if len(tstat_dirs) > 0: tstat_dirs.sort() new_dir = tstat_dirs[-1] tstat_file = tstat_piped + new_dir + "/log_tcp_temp_complete" - + LOGGER.info("Following: {:s}".format(str(tstat_file))) - + return tstat_file else: LOGGER.info("No Tstat directory found. Waiting...") @@ -177,7 +177,7 @@ class l3_distributedattackdetector: stub = ContextServiceStub(channel) context_id = ContextId() context_id.context_uuid.uuid = context_id_str - + return stub.ListServiceIds(context_id) def get_services(self, context_id_str): @@ -185,31 +185,31 @@ class l3_distributedattackdetector: stub = ContextServiceStub(channel) context_id = ContextId() context_id.context_uuid.uuid = context_id_str - + return stub.ListServices(context_id) def get_service_id(self, context_id): service_list = self.get_services(context_id) service_id = None - + for s in service_list.services: if s.service_type == ServiceTypeEnum.SERVICETYPE_L3NM: service_id = s.service_id break else: pass - + return service_id def get_endpoint_id(self, context_id): service_list = self.get_services(context_id) endpoint_id = None - + for s in service_list.services: if s.service_type == ServiceTypeEnum.SERVICETYPE_L3NM: endpoint_id = s.service_endpoint_ids[0] break - + return endpoint_id def get_features_ids(self): @@ -246,7 +246,7 @@ class l3_distributedattackdetector: def check_if_connection_is_attack(self): if self.conn_id[0] in self.known_attack_ips or self.conn_id[2] in self.known_attack_ips: - LOGGER.info("Attack detected. Origin: {0}, destination: {1}".format(self.conn_id[0], self.conn_id[2])) + LOGGER.info("Attack detected. Origin IP address: {0}, destination IP address: {1}".format(self.conn_id[0], self.conn_id[2])) def create_cad_features(self): self.cad_features = { @@ -323,20 +323,21 @@ class l3_distributedattackdetector: while line is None: LOGGER.info("Waiting for new data...") + time.sleep(1 / 100) line = next(loglines, None) - + if index == 0 and IGNORE_FIRST_LINE_TSTAT: index = index + 1 continue - + if STOP: break num_lines += 1 start = time.time() line_id = line.split(" ") - + self.conn_id = (line_id[0], line_id[1], line_id[14], line_id[15]) self.new_connections[self.conn_id] = self.process_line(line) @@ -358,7 +359,7 @@ class l3_distributedattackdetector: metrics_list_pb, send_data_times = await self.send_data(metrics_list_pb, send_data_times) - index = index + 1 + index += 1 process_time.append(time.time() - start) -- GitLab From 4c195451c5eae176a88f0ccc97d44fe123f3f984 Mon Sep 17 00:00:00 2001 From: Lluis Gifre Renom Date: Thu, 13 Jul 2023 12:38:18 +0000 Subject: [PATCH 10/12] Update l3_distributedattackdetector.py --- .../service/l3_distributedattackdetector.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py b/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py index 269640538..2cee9d20c 100644 --- a/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py +++ b/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py @@ -106,20 +106,20 @@ class l3_distributedattackdetector: LOGGER.info("Gracefully stopping...") - def follow(self, file, time_sleep): + def follow(self, logfile, time_sleep): """ - Generator function that yields new lines in a file + Generator function that yields new lines in a log file """ # seek the end of the file - # file.seek(0, os.SEEK_END) + # logfile.seek(0, os.SEEK_END) chunk = "" # start an infinite loop while True: # read last line of the file - line = file.readline() + line = logfile.readline() # sleep if the file hasn't been updated if not line: -- GitLab From 768960f6631912dd430ce22d46bf137464c2c235 Mon Sep 17 00:00:00 2001 From: Lluis Gifre Renom Date: Thu, 13 Jul 2023 12:40:03 +0000 Subject: [PATCH 11/12] Update deploy.sh --- src/tests/scenario3/l3/deploy.sh | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/tests/scenario3/l3/deploy.sh b/src/tests/scenario3/l3/deploy.sh index be6c43ef7..4f4f8baed 100755 --- a/src/tests/scenario3/l3/deploy.sh +++ b/src/tests/scenario3/l3/deploy.sh @@ -14,7 +14,7 @@ # limitations under the License. source src/tests/scenario3/l3/deploy_specs.sh -deploy/all.sh +./deploy/all.sh source tfs_runtime_env_vars.sh ofc22/run_test_01_bootstrap.sh ofc22/run_test_02_create_service.sh -- GitLab From 8e04db4e6978831d23aff3a6877d680894cca992 Mon Sep 17 00:00:00 2001 From: karamchandan Date: Thu, 13 Jul 2023 14:51:04 +0200 Subject: [PATCH 12/12] Cleanup of DAD service implementation --- .../service/l3_distributedattackdetector.py | 252 ++++++++++++++++-- 1 file changed, 236 insertions(+), 16 deletions(-) diff --git a/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py b/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py index 269640538..bd9e490a9 100644 --- a/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py +++ b/src/l3_distributedattackdetector/service/l3_distributedattackdetector.py @@ -65,6 +65,18 @@ BATCH_SIZE = 10 class l3_distributedattackdetector: def __init__(self): + """ + Initializes a Distributed Attack Detector. + + This method initializes a Distributed Attack Detector by setting up instance variables, connecting to the Centralized Attack Detector, obtaining feature IDs, and starting the process traffic loop. It also sets up a signal handler for handling keyboard interrupts. + + Args: + None. + + Returns: + None. + """ + LOGGER.info("Creating Distributed Attack Detector") self.feature_ids = [] @@ -88,27 +100,67 @@ class l3_distributedattackdetector: LOGGER.info("Features Ids.: {:s}".format(str(self.feature_ids))) asyncio.run(self.process_traffic()) - + def read_kwnown_attack_ips(self): + """ + Reads a list of known attack IPs from a CSV file. + + This method reads a list of known attack IPs from a CSV file named "known_attack_ips.csv". The method returns a list of strings containing the IP addresses. + + Args: + None. + + Returns: + List[str]: A list of strings containing the IP addresses of known attack sources. + """ + + # Initialize an empty list to store the known attack IPs known_attack_ips = [] - # open known attack ips csv file + # Open the known attack IPs CSV file with open("known_attack_ips.csv", "r") as f: + # Read the contents of the file and split it into a list of strings known_attack_ips = f.read().split(",") + # Return the list of known attack IPs return known_attack_ips - + def handler(self): + """ + Handles a keyboard interrupt signal. + + This method handles a keyboard interrupt signal by setting the `STOP` flag to `True` and logging a message indicating that the program is stopping gracefully. + + Args: + None. + + Returns: + None. + """ + + # Set the STOP flag to True if STOP: exit() - STOP = True + # Log a message indicating that the program is stopping gracefully LOGGER.info("Gracefully stopping...") def follow(self, file, time_sleep): """ - Generator function that yields new lines in a file + Generator function that yields new lines in a file. + + This method reads a file object and yields new lines as they are added to the file. The method uses an infinite loop to continuously read the last line of the file. If the file hasn't been updated, the method sleeps for the specified `time_sleep` interval. If the last line of the file doesn't end with a newline character, the method appends the line to a `chunk` variable. When a newline character is encountered, the method yields the line, possibly appending the `chunk` variable to the line if it contains a partial line. The method returns an iterator that yields lines from the file. + + Args: + file (TextIO): The file object to read from. + time_sleep (float): The time to sleep if the file hasn't been updated. + + Yields: + str: The next line in the file. + + Returns: + None. """ # seek the end of the file @@ -136,11 +188,27 @@ class l3_distributedattackdetector: yield line def load_file(self, dirname=TSTAT_DIR_NAME): + """ + Loads the latest Tstat log file. + + This method loads the latest Tstat log file by searching for the most recent directory in the specified `dirname` directory. If a directory is found, the method returns the path to the `log_tcp_temp_complete` file in that directory. If no directory is found, the method logs a message and waits for 5 seconds before trying again. + + Args: + dirname (str): The name of the directory to search for Tstat log files. Defaults to `TSTAT_DIR_NAME`. + + Returns: + str: The path to the latest Tstat log file. + """ + while True: + # Get the path to the Tstat directory here = os.path.dirname(os.path.abspath(__file__)) tstat_piped = os.path.join(here, dirname) + + # Get a list of all directories in the Tstat directory tstat_dirs = os.listdir(tstat_piped) + # If there are directories in the Tstat directory, find the most recent one and return the path to the log file if len(tstat_dirs) > 0: tstat_dirs.sort() new_dir = tstat_dirs[-1] @@ -149,16 +217,25 @@ class l3_distributedattackdetector: LOGGER.info("Following: {:s}".format(str(tstat_file))) return tstat_file + # If there are no directories in the Tstat directory, log a message and wait for 5 seconds before trying again else: LOGGER.info("No Tstat directory found. Waiting...") time.sleep(5) def process_line(self, line): """ - - Preprocessing before a message per line - - Avoids crash when nan are found by generating a 0s array - - Returns a list of values + Processes a single line of input data and returns a list of feature values. + + Args: + line (str): A single line of input data containing feature values separated by spaces. + + Returns: + List[float]: A list of feature values extracted from the input line. + + Raises: + IndexError: If the input line does not contain enough feature values. """ + line = line.split(" ") try: @@ -172,15 +249,19 @@ class l3_distributedattackdetector: return values - def get_service_ids(self, context_id_str): - with grpc.insecure_channel(CONTEXT_CHANNEL) as channel: - stub = ContextServiceStub(channel) - context_id = ContextId() - context_id.context_uuid.uuid = context_id_str + def get_services(self, context_id_str): + """ + Gets the services for a given context ID. - return stub.ListServiceIds(context_id) + This method gets the services for a given context ID by calling the ListServices method of the ContextService gRPC stub. - def get_services(self, context_id_str): + Args: + context_id_str (str): The context ID to get services for. + + Returns: + ListServicesResponse: A response object containing the services. + """ + with grpc.insecure_channel(CONTEXT_CHANNEL) as channel: stub = ContextServiceStub(channel) context_id = ContextId() @@ -189,6 +270,18 @@ class l3_distributedattackdetector: return stub.ListServices(context_id) def get_service_id(self, context_id): + """ + Gets the service ID for a given context ID. + + This method gets the service ID for a given context ID by calling the get_services method and searching for the first service in the list with the service type of SERVICETYPE_L3NM. + + Args: + context_id (str): The context ID to get the service ID for. + + Returns: + str: The service ID. + """ + service_list = self.get_services(context_id) service_id = None @@ -202,6 +295,18 @@ class l3_distributedattackdetector: return service_id def get_endpoint_id(self, context_id): + """ + Gets the endpoint ID for a given context ID. + + This method gets the endpoint ID for a given context ID by calling the get_services method and searching for the first service in the list with the service type of SERVICETYPE_L3NM. + + Args: + context_id (str): The context ID to get the endpoint ID for. + + Returns: + str: The endpoint ID. + """ + service_list = self.get_services(context_id) endpoint_id = None @@ -213,9 +318,33 @@ class l3_distributedattackdetector: return endpoint_id def get_features_ids(self): + """ + Gets the feature IDs used by the Centralized Attack Detector model. + + This method gets the feature IDs used by the Centralized Attack Detector model by calling the GetFeaturesIds method of the Centralized Attack Detector gRPC stub. + + Args: + None. + + Returns: + list: A list of feature IDs. + """ + return self.cad.GetFeaturesIds(Empty()).auto_features def check_types(self): + """ + Checks the types of the features that will be sent to the Centralized Attack Detector. + + This method checks the types of the Centralized Attack Detector features to ensure that they are of the correct type. If any of the types are incorrect, the method raises an AssertionError. + + Args: + None. + + Returns: + None. + """ + for feature in self.cad_features["features"]: assert isinstance(feature, float) @@ -229,6 +358,28 @@ class l3_distributedattackdetector: assert isinstance(self.cad_features["connection_metadata"]["time_end"], float) def insert_connection(self): + """ + Inserts a new connection into the `connections_dict` instance variable. + + This method inserts a new connection into the `connections_dict` instance variable. The method uses the `conn_id` instance variable to create a new dictionary entry for the connection. If the connection already exists in the dictionary, the method updates the `time_end` value of the existing entry. If the connection doesn't exist in the dictionary, the method creates a new entry with the following keys: + - "ip_o": The source IP address of the connection. + - "port_o": The source port of the connection. + - "ip_d": The destination IP address of the connection. + - "port_d": The destination port of the connection. + - "flow_id": The flow ID of the connection. + - "service_id": The service ID of the connection. + - "endpoint_id": The endpoint ID of the connection. + - "protocol": The protocol of the connection. + - "time_start": The start time of the connection. + - "time_end": The end time of the connection. + + Args: + None. + + Returns: + None. + """ + try: self.connections_dict[self.conn_id]["time_end"] = time.time() except KeyError: @@ -245,10 +396,46 @@ class l3_distributedattackdetector: self.connections_dict[self.conn_id]["port_d"] = self.conn_id[3] def check_if_connection_is_attack(self): + """ + Checks if a connection is an attack based on known attack IP addresses. + + This method checks if a connection is an attack based on known attack IP addresses. The method uses the `conn_id` and `known_attack_ips` instance variables to determine if the source or destination IP address of the connection is in the list of known attack IP addresses. If either IP address is in the list, the method logs a message indicating that an attack has been detected. + + Args: + None. + + Returns: + None. + """ + if self.conn_id[0] in self.known_attack_ips or self.conn_id[2] in self.known_attack_ips: LOGGER.info("Attack detected. Origin IP address: {0}, destination IP address: {1}".format(self.conn_id[0], self.conn_id[2])) def create_cad_features(self): + """ + Creates a dictionary of features and connection metadata for the Centralized Attack Detector. + + This method creates a dictionary of features and connection metadata for the Centralized Attack Detector. The method uses the `new_connections` and `connections_dict` instance variables to obtain the necessary data. The resulting dictionary contains the following keys: + - "features": A list of the first 10 features of the connection. + - "connection_metadata": A dictionary containing the following keys: + - "ip_o": The source IP address of the connection. + - "port_o": The source port of the connection. + - "ip_d": The destination IP address of the connection. + - "port_d": The destination port of the connection. + - "flow_id": The flow ID of the connection. + - "service_id": The service ID of the connection. + - "endpoint_id": The endpoint ID of the connection. + - "protocol": The protocol of the connection. + - "time_start": The start time of the connection. + - "time_end": The end time of the connection. + + Args: + None. + + Returns: + None. + """ + self.cad_features = { "features": self.new_connections[self.conn_id][0:10], "connection_metadata": { @@ -266,6 +453,18 @@ class l3_distributedattackdetector: } async def send_batch_async(self, metrics_list_pb): + """ + Sends a batch of traffic data to a Centralized Attack Detector. + + This method sends a batch of traffic data to a Centralized Attack Detector for analysis. The method creates a `L3CentralizedattackdetectorBatchInput` object from the provided `metrics_list_pb`, and sends the batch using an executor to run the `AnalyzeBatchConnectionStatistics` method in a separate thread. The method returns `None`. + + Args: + metrics_list_pb (List[L3CentralizedattackdetectorMetrics]): A list of traffic metrics to send to the Centralized Attack Detector. + + Returns: + None. + """ + loop = asyncio.get_running_loop() # Create metrics batch @@ -281,7 +480,19 @@ class l3_distributedattackdetector: LOGGER.error(f"Error sending batch: {e}") async def send_data(self, metrics_list_pb, send_data_times): - # Send data to CAD + """ + Sends traffic data to a Centralized Attack Detector. + + This method sends traffic data to a Centralized Attack Detector for analysis. If the `SEND_DATA_IN_BATCHES` flag is set to `True`, the data is sent in batches of size `BATCH_SIZE`. Otherwise, the data is sent one metric at a time. The method returns the updated `metrics_list_pb` and `send_data_times` arrays. + + Args: + metrics_list_pb (List[L3CentralizedattackdetectorMetrics]): A list of traffic metrics to send to the Centralized Attack Detector. + send_data_times (np.ndarray): An array of times it took to send each batch of data. + + Returns: + Tuple[List[L3CentralizedattackdetectorMetrics], np.ndarray]: A tuple containing the updated `metrics_list_pb` and `send_data_times` arrays. + """ + if SEND_DATA_IN_BATCHES: if len(metrics_list_pb) == BATCH_SIZE: send_data_time_start = time.time() @@ -303,6 +514,15 @@ class l3_distributedattackdetector: return metrics_list_pb, send_data_times async def process_traffic(self): + """ Processes traffic data from a Tstat log file. + This method reads traffic data from a Tstat log file, processes each line of data, and sends the resulting metrics to a Centralized Attack Detector. It runs indefinitely until the `STOP` flag is set to `True`. + + Args: + None. + Returns: + None. + """ + LOGGER.info("Loading Tstat log file...") logfile = open(self.load_file(), "r") -- GitLab