Merge branch '20-admin-webapp-anomaly-explorer' into 'main'

Resolve "Admin WebApp Anomaly Explorer"

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

Author: DlieBG

setup.py

@@ -2,7 +2,7 @@
 
 setup(
     name='ZettaBGP',
-    version='0.2.2',
+    version='0.2.3',
     description='',
     url='https://git.univ.leitwert.net/imprj/01-bgp-testbed/zettabgp',
     author='Benedikt Schwering & Sebastian Forstner',

src/controllers/message_replay.py

@@ -0,0 +1,52 @@
+# -*- 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.models.message_replay import MessageReplayRequest, MessageReplayResult
+from src.services.message_replay import message_replay
+from fastapi import APIRouter
+import json, os
+
+message_replay_router = APIRouter()
+
+@message_replay_router.post('/')
+def start_message_replay(message_replay_request: MessageReplayRequest) -> MessageReplayResult:
+    '''
+    This function starts a message replay.
+
+    Author:
+        Benedikt Schwering <bes9584@thi.de>
+
+    Args:
+        message_replay_request (MessageReplayRequest): The message replay request.
+
+    Returns:
+        MessageReplayResult: The message replay result.
+    '''
+    message_replay_result = message_replay(
+        no_rabbitmq_direct=message_replay_request.no_rabbitmq_direct,
+        rabbitmq_grouped=message_replay_request.rabbitmq_grouped,
+        no_mongodb_log=message_replay_request.no_mongodb_log,
+        no_mongodb_state=message_replay_request.no_mongodb_state,
+        no_mongodb_statistics=message_replay_request.no_mongodb_statistics,
+        clear_mongodb=message_replay_request.clear_mongodb,
+        playback_speed=message_replay_request.playback_speed,
+        playback_interval=None,
+        start_timestamp=None,
+        end_timestamp=None,
+        start_time=message_replay_request.start_time + ':00',
+        end_time=message_replay_request.end_time + ':00',
+    )
+
+    return MessageReplayResult(
+        count_announce=message_replay_result.count_announce,
+        count_withdraw=message_replay_result.count_withdraw,
+    )

src/main.py

@@ -10,18 +10,12 @@
     Benedikt Schwering <bes9584@thi.de>
     Sebastian Forstner <sef9869@thi.de>
 '''
-from src.adapters.mongodb import MongoDBAdapter, MongoDBLogLoader
 import src.services.mrt_simulation as mrt_simulation_service
-from src.adapters.rabbitmq import RabbitMQAdapter
-from src.parsers.reverse import ReverseParser
+import src.services.message_replay as message_replay_service
+import src.services.rib_load as rib_load_service
 import src.services.exabgp as exabgp_service
-from datetime import timedelta, datetime
-from src.parsers.rib import RibParser
-from collections import OrderedDict
 from src.webapp import start_webapp
-from mrtparse import Reader
-from rich import print
-import click, time
+import click
 
 @click.group()
 def cli():
@@ -275,33 +269,15 @@ def rib_load(no_rabbitmq_direct: bool, rabbitmq_grouped: int, no_mongodb_log: bo
         clear_mongodb (bool): Clear MongoDB collections.
         rib_file (str): RIB file to process.
     '''
-    parser = RibParser()
-
-    if not no_rabbitmq_direct or rabbitmq_grouped:
-        RabbitMQAdapter(
-            parser=parser,
-            no_direct=no_rabbitmq_direct,
-            queue_interval=rabbitmq_grouped,
-        )
-
-    if not no_mongodb_log or not no_mongodb_state or not no_mongodb_statistics:
-        MongoDBAdapter(
-            parser=parser,
-            no_mongodb_log=no_mongodb_log,
-            no_mongodb_state=no_mongodb_state,
-            no_mongodb_statistics=no_mongodb_statistics,
-            clear_mongodb=clear_mongodb,
-        )
-
-    for message in Reader(rib_file):
-        if message.data['type'] != {13: 'TABLE_DUMP_V2'}:
-            print('[dark_orange]\[WARN][/] Skipping unsupported MRT type: ', end='')
-            print(message.data['type'])
-            continue
-
-        parser.parse(
-            statement=message.data,
-        )
+    rib_load_service.rib_load(
+        no_rabbitmq_direct=no_rabbitmq_direct,
+        rabbitmq_grouped=rabbitmq_grouped,
+        no_mongodb_log=no_mongodb_log,
+        no_mongodb_state=no_mongodb_state,
+        no_mongodb_statistics=no_mongodb_statistics,
+        clear_mongodb=clear_mongodb,
+        rib_file=rib_file,
+    )
 
 @cli.command(
     name='message-replay',
@@ -409,60 +385,17 @@ def message_replay(no_rabbitmq_direct: bool, rabbitmq_grouped: int, no_mongodb_l
         start_time (str): Starttime of replay as time; in format (T is a set character): YYYY-MM-DDThh:mm:ss.
         end_time (str): Endtime of replay as time; in format (T is a set character): YYYY-MM-DDThh:mm:ss.
     '''
-    parser = ReverseParser()
-
-    if not no_rabbitmq_direct or rabbitmq_grouped:
-        RabbitMQAdapter(
-            parser=parser,
-            no_direct=no_rabbitmq_direct,
-            queue_interval=rabbitmq_grouped,
-        )
-    
-    playback_speed_reference: datetime = None
-    playback_interval_stop: datetime = None
-
-    # Check if start and end are given and in which format; no time given results in replaying whole db
-    if start_timestamp and end_timestamp:
-        start_time = datetime.fromtimestamp(start_timestamp)
-        end_time = datetime.fromtimestamp(end_timestamp)
-        new_messages = MongoDBLogLoader.load_messages(timestamp_start = start_time, timestamp_end = end_time)
-    elif start_time and end_time: 
-        time_start = datetime.fromisoformat(start_time)
-        time_end = datetime.fromisoformat(end_time)
-        new_messages = MongoDBLogLoader.load_messages(timestamp_start = time_start, timestamp_end = time_end)
-    else:
-        new_messages = MongoDBLogLoader.load_messages(timestamp_start = None, timestamp_end = None)
-
-    # Copy messages in local list to avoid deleting them if -c is set; new_messages is corsor pointing to db
-    all_messages: list[OrderedDict] = []
-    for message in new_messages:
-        all_messages.append(message)
-
-    # Init for MongoDBAdapter to avouid deleting messages before loading
-    if not no_mongodb_log or not no_mongodb_state or not no_mongodb_statistics:
-        MongoDBAdapter(
-            parser=parser,
-            no_mongodb_log=no_mongodb_log,
-            no_mongodb_state=no_mongodb_state,
-            no_mongodb_statistics=no_mongodb_statistics,
-            clear_mongodb=clear_mongodb,
-        )
-
-    for message in all_messages:
-        current_timestamp: datetime = message['timestamp']
-
-        if playback_speed:
-            if playback_speed_reference:
-                time.sleep((current_timestamp - playback_speed_reference).seconds / playback_speed)
-
-            playback_speed_reference = current_timestamp
-
-        if playback_interval:
-            if playback_interval_stop:
-                if current_timestamp > playback_interval_stop:
-                    input('Enter for next interval...')
-                    playback_interval_stop = playback_interval_stop + timedelta(minutes=playback_interval)
-            else:
-                playback_interval_stop = current_timestamp + timedelta(minutes=playback_interval)
-
-        parser.parse(message)
+    message_replay_service.message_replay(
+        no_rabbitmq_direct=no_rabbitmq_direct,
+        rabbitmq_grouped=rabbitmq_grouped,
+        no_mongodb_log=no_mongodb_log,
+        no_mongodb_state=no_mongodb_state,
+        no_mongodb_statistics=no_mongodb_statistics,
+        clear_mongodb=clear_mongodb,
+        playback_speed=playback_speed,
+        playback_interval=playback_interval,
+        start_timestamp=start_timestamp,
+        end_timestamp=end_timestamp,
+        start_time=start_time,
+        end_time=end_time,
+    )

src/models/message_replay.py

@@ -0,0 +1,41 @@
+# -*- 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 pydantic import BaseModel
+from typing import Optional
+
+class MessageReplayRequest(BaseModel):
+    '''
+    This class represents a request to replay messages.
+
+    Author:
+        Benedikt Schwering <bes9584@thi.de>
+    '''
+    no_rabbitmq_direct: bool
+    rabbitmq_grouped: Optional[int]
+    no_mongodb_log: bool
+    no_mongodb_state: bool
+    no_mongodb_statistics: bool
+    clear_mongodb: bool
+    playback_speed: Optional[int]
+    start_time: str
+    end_time: str
+
+class MessageReplayResult(BaseModel):
+    '''
+    This class represents the result of replay messages.
+
+    Author:
+        Benedikt Schwering <bes9584@thi.de>
+    '''
+    count_withdraw: int
+    count_announce: int

src/services/message_replay.py

@@ -0,0 +1,120 @@
+# -*- 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.adapters.mongodb import MongoDBAdapter, MongoDBLogLoader
+from src.adapters.rabbitmq import RabbitMQAdapter
+from src.models.route_update import ChangeType
+from src.parsers.reverse import ReverseParser
+from datetime import timedelta, datetime
+from collections import OrderedDict
+from pydantic import BaseModel
+import time
+
+class MessageReplayResult(BaseModel):
+    count_announce: int
+    count_withdraw: int
+
+def message_replay(no_rabbitmq_direct: bool, rabbitmq_grouped: int, no_mongodb_log: bool, no_mongodb_state: bool, no_mongodb_statistics: bool, clear_mongodb: bool, playback_speed: int, playback_interval: int, start_timestamp: float, end_timestamp: float, start_time: str, end_time: str) -> MessageReplayResult:
+    '''
+    Message replay service for replaying BGP messages from Database log.
+
+    Author:
+        Sebastian Forstner <sef9869@thi.de>
+
+    Args:
+        no_rabbitmq_direct (bool): Disable direct RabbitMQ direct queue..
+        rabbitmq_grouped (int): Queue group interval in minutes.
+        no_mongodb_log (bool): Disable logging to MongoDB.
+        no_mongodb_state (bool): Disable state storage to MongoDB.
+        no_mongodb_statistics (bool): Disable statistics storage to MongoDB.
+        clear_mongodb (bool): Clear MongoDB collections.
+        playback_speed (int): Playback speed in multiples of real time.
+        playback_interval (int): Playback interval in minutes.
+        start_timestamp (float): Starttime of replay as timestamp.
+        end_timestamp (float): Endtime of replay as timestamp.
+        start_time (str): Starttime of replay as time; in format (T is a set character): YYYY-MM-DDThh:mm:ss.
+        end_time (str): Endtime of replay as time; in format (T is a set character): YYYY-MM-DDThh:mm:ss.
+
+    Returns:
+        MessageReplayResult: The message replay result.
+    '''
+    message_replay_result = MessageReplayResult(
+        count_announce=0,
+        count_withdraw=0,
+    )
+
+    parser = ReverseParser()
+
+    if not no_rabbitmq_direct or rabbitmq_grouped:
+        RabbitMQAdapter(
+            parser=parser,
+            no_direct=no_rabbitmq_direct,
+            queue_interval=rabbitmq_grouped,
+        )
+    
+    playback_speed_reference: datetime = None
+    playback_interval_stop: datetime = None
+
+    # Check if start and end are given and in which format; no time given results in replaying whole db
+    if start_timestamp and end_timestamp:
+        start_time = datetime.fromtimestamp(start_timestamp)
+        end_time = datetime.fromtimestamp(end_timestamp)
+        new_messages = MongoDBLogLoader.load_messages(timestamp_start = start_time, timestamp_end = end_time)
+    elif start_time and end_time: 
+        time_start = datetime.fromisoformat(start_time)
+        time_end = datetime.fromisoformat(end_time)
+        new_messages = MongoDBLogLoader.load_messages(timestamp_start = time_start, timestamp_end = time_end)
+    else:
+        new_messages = MongoDBLogLoader.load_messages(timestamp_start = None, timestamp_end = None)
+
+    # Copy messages in local list to avoid deleting them if -c is set; new_messages is corsor pointing to db
+    all_messages: list[OrderedDict] = []
+    for message in new_messages:
+        all_messages.append(message)
+
+    # Init for MongoDBAdapter to avouid deleting messages before loading
+    if not no_mongodb_log or not no_mongodb_state or not no_mongodb_statistics:
+        MongoDBAdapter(
+            parser=parser,
+            no_mongodb_log=no_mongodb_log,
+            no_mongodb_state=no_mongodb_state,
+            no_mongodb_statistics=no_mongodb_statistics,
+            clear_mongodb=clear_mongodb,
+        )
+
+    for message in all_messages:
+        current_timestamp: datetime = message['timestamp']
+
+        if playback_speed:
+            if playback_speed_reference:
+                time.sleep((current_timestamp - playback_speed_reference).seconds / playback_speed)
+
+            playback_speed_reference = current_timestamp
+
+        if playback_interval:
+            if playback_interval_stop:
+                if current_timestamp > playback_interval_stop:
+                    input('Enter for next interval...')
+                    playback_interval_stop = playback_interval_stop + timedelta(minutes=playback_interval)
+            else:
+                playback_interval_stop = current_timestamp + timedelta(minutes=playback_interval)
+
+        updates = parser.parse(message)
+
+        if updates:
+            for update in updates:
+                if update.change_type == ChangeType.ANNOUNCE:
+                    message_replay_result.count_announce += 1
+                elif update.change_type == ChangeType.WITHDRAW:
+                    message_replay_result.count_withdraw += 1
+
+    return message_replay_result