Merge branch '18-mrt-funktionen-refactoring' into 'main'

Resolve "MRT Funktionen Refactoring"

Closes #18

See merge request imprj/01-bgp-testbed/zettabgp!14

LICENSE

@@ -1,6 +1,6 @@
 MIT License
 
-Copyright (c) 2024 Benedikt Schwering
+Copyright (c) 2024 Benedikt Schwering and Sebastian Forstner
 
 Permission is hereby granted, free of charge, to any person obtaining a copy
 of this software and associated documentation files (the "Software"), to deal

README.md

@@ -84,7 +84,7 @@ ZETTABGP_WEBAPP_MRT_LIBRARY_PATH
 ## Usage
 ZettaBGP provides a CLI interface with some commands for testbed simulations as well for production use.
 
-### Comands
+### Commands
 #### `zettabgp exabgp`
 The `exabgp` subcommand is used for processing ExaBGP Messages.\
 The process can be started from within ExaBGP.\
@@ -112,7 +112,7 @@ When no `-g` option is present, no grouped updates will appear at all.
 
 #### `zettabgp mrt-simulation`
 The `mrt-simulation` subcommand is used for processing mrt files.\
-It is mendatory to provide a valid path to at least one mrt file.\
+It is mandatory to provide a valid path to at least one mrt file.\
 `mrt-simulation` also supports the handling of multiple mrt files.\
 But keep in mind to provide sequentially sorted mrt files based on the timeframe.\
 Otherwise the grouping feature will not work properly!
@@ -147,7 +147,7 @@ zettabgp mrt-simulation <mrt-file> -p 2
 ```
 
 ##### Playback Interval
-For debugging the timebased group update queue, it is very useful to playback all update messages that occur within an interval of for example 5 minutes.\
+For debugging the timebase group update queue, it is very useful to playback all update messages that occur within an interval of for example 5 minutes.\
 When you specify option `-o` you can set a playback interval in minutes that defaults to 5 minutes.\
 Between the intervals you have to press enter to continue with the replay of the next interval.\
 Of course you can combine this option with the playback speed option.\

src/adapters/mongodb.py

@@ -1,3 +1,15 @@
+# -*- coding: utf-8 -*-
+'''
+ZettaBGP - Advanced Anomaly Detection in Internet Routing
+Copyright (c) 2024 Benedikt Schwering and Sebastian Forstner
+
+This work is licensed under the terms of the MIT license.
+For a copy, see LICENSE in the project root.
+
+Author:
+    Benedikt Schwering <bes9584@thi.de>
+    Sebastian Forstner <sef9869@thi.de>
+'''
 from src.parsers.route_update import RouteUpdateParser
 from src.models.route_update import RouteUpdate
 from src.models.route_update import ChangeType
@@ -7,10 +19,28 @@
 import os
 
 class MongoDBAdapter:
-    '''This class is responsible for receiving the parsed messages and forwarding them to both MongoDB databases'''
+    '''
+    This class is responsible for receiving the parsed messages and forwarding them to both MongoDB databases.
+
+    Author:
+        Sebastian Forstner <sef9869@thi.de>
+    '''
     def __init__(self, parser: RouteUpdateParser, no_mongodb_log: bool, no_mongodb_state: bool, no_mongodb_statistics: bool, clear_mongodb: bool):
+        '''
+        Initializes the MongoDBAdapter.
+
+        Author:
+            Sebastian Forstner <sef9869@thi.de>
+
+        Args:
+            parser (RouteUpdateParser): The parser to receive the parsed messages from.
+            no_mongodb_log (bool): Whether to disable the log storage.
+            no_mongodb_state (bool): Whether to disable the state storage.
+            no_mongodb_statistics (bool): Whether to disable the statistics storage.
+            clear_mongodb (bool): Whether to clear the MongoDB databases.
+        '''
         try:
-            '''Connects to MongoDB-Container running with Docker'''
+            # Connects to MongoDB-Container running with Docker
             database_client = MongoClient(
                 host=os.getenv('MONGO_DB_HOST', 'localhost'),
                 port=int(os.getenv('MONGO_DB_PORT', 27017)),
@@ -21,21 +51,22 @@ def __init__(self, parser: RouteUpdateParser, no_mongodb_log: bool, no_mongodb_s
         log_flag = no_mongodb_log
         state_flag = no_mongodb_state
         statistics_flag = no_mongodb_statistics
-        '''Creates database and collection for log storrage'''
+
+        # Creates database and collection for log storage
         if not log_flag:
             log_db = database_client.message_log
             log_collection = log_db.storage
             if clear_mongodb:
                 log_collection.delete_many({})
 
-        '''Creates database and collection for state storrage'''
+        # Creates database and collection for state storage
         if not state_flag:
             state_db = database_client.message_state
             state_collection = state_db.storage
             if clear_mongodb:
                 state_collection.delete_many({})
 
-        '''Creates database and collection for statistics storrage'''
+        # Creates database and collection for statistics storage
         if not statistics_flag:
             statistics_db = database_client.message_statistics
             statistics_collection = statistics_db.storage
@@ -45,7 +76,7 @@ def __init__(self, parser: RouteUpdateParser, no_mongodb_log: bool, no_mongodb_s
         @parser.on_update
         def on_update(message: RouteUpdate):
 
-            '''saves optional, non-base-type attributes for later use; required to guarantee save use of mongodb'''
+            # Saves optional, non-base-type attributes for later use; required to guarantee save use of mongodb
             if message.path_attributes.origin:
                 origins = message.path_attributes.origin.value
             else:
@@ -75,7 +106,7 @@ def on_update(message: RouteUpdate):
                     else:
                         extended_community.append(str(ext_com)) 
 
-            '''creates dict for message with _id and other unique attributes, that dont change'''
+            # Creates dict for message with _id and other unique attributes, that don't change
             new_message_id = {
                 'timestamp' : message.timestamp,
                 'peer_ip' : message.peer_ip,
@@ -103,7 +134,8 @@ def on_update(message: RouteUpdate):
                 },
                 '_id' : ObjectId(),
             }
-            '''creates dict used for collection updates, MUST NOT contain _id and should not contain other non changing attributes'''
+
+            # Creates dict used for collection updates, MUST NOT contain _id and should not contain other non changing attributes
             set_message = { 
                 '$set': {
                     'timestamp' : message.timestamp,
@@ -127,7 +159,7 @@ def on_update(message: RouteUpdate):
                 }
             }
 
-            '''route got withdrawn, db actions accordingly'''
+            # Route got withdrawn, db actions accordingly
             if message.change_type == ChangeType.WITHDRAW:
                 if not log_flag:
                     log_announce = log_collection.insert_one(new_message_id)
@@ -159,7 +191,7 @@ def on_update(message: RouteUpdate):
                         }
                     statistics_announce = statistics_collection.update_one(statistics_filter, new_values, upsert=True)
 
-            '''route got announced, db actions accordingly'''
+            # Route got announced, db actions accordingly
             if message.change_type == ChangeType.ANNOUNCE:
                 if not log_flag:
                     log_announce = log_collection.insert_one(new_message_id)

src/adapters/rabbitmq.py

@@ -1,25 +1,53 @@
+# -*- coding: utf-8 -*-
+'''
+ZettaBGP - Advanced Anomaly Detection in Internet Routing
+Copyright (c) 2024 Benedikt Schwering and Sebastian Forstner
+
+This work is licensed under the terms of the MIT license.
+For a copy, see LICENSE in the project root.
+
+Author:
+    Benedikt Schwering <bes9584@thi.de>
+    Sebastian Forstner <sef9869@thi.de>
+'''
 from src.parsers.route_update import RouteUpdateParser
 from src.models.route_update import RouteUpdate
 from datetime import timedelta, datetime
 import pika, json, os
 
 class RabbitMQAdapter:
-    '''This class is responsible for receiving the parsed messages and forwarding them to the RabbitMQ message broker'''
+    '''
+    This class is responsible for receiving the parsed messages and forwarding them to the RabbitMQ message broker.
+
+    Author:
+        Benedikt Schwering <bes9584@thi.de>
+    '''
     def __init__(self, parser: RouteUpdateParser, no_direct: bool, queue_interval: int):
+        '''
+        Initializes the RabbitMQAdapter.
+
+        Author:
+            Benedikt Schwering <bes9584@thi.de>
+
+        Args:
+            parser (RouteUpdateParser): The parser to receive the parsed messages from.
+            no_direct (bool): Whether to disable the direct route updates.
+            queue_interval (int): The interval in minutes to group the route updates.
+        '''
         connection = pika.BlockingConnection(
             pika.ConnectionParameters(
                 host=os.getenv('RABBIT_MQ_HOST', 'localhost'),
             )
         )
 
-        '''Creates a channel for the connection and declares the zettabgp exchange'''
+        # Creates a channel for the connection and declares the zettabgp exchange
         channel = connection.channel()
         channel.exchange_declare(
             exchange='zettabgp',
             exchange_type='direct',
         )
 
-        '''Declares the test_bgp_updates queue and binds it to the zettabgp exchange'''
+        # Declares the test_bgp_updates queue and binds it to the zettabgp exchange
         def _declare_test_queue(queue_name: str, routing_key: str):
             channel.queue_declare(
                 queue=queue_name,