Event-Driven Temporal Graph Networks for Asynchronous Multi-Agent Cyber Defense in NetForge_RL

1. 1
   Identify Simulation Limitations

   Recognize that legacy cyber simulators abstract network physics, operate synchronously, and provide idealized state information, making them inadequate for realistic MARL training in cyber defense.
2. 2
   Adopt Event-Driven Temporal Graph Networks

   Integrate Event-Driven Temporal Graph Networks to model complex, asynchronous interactions inherent in real-world cyber defense scenarios, moving beyond simplistic synchronous models.
3. 3
   Utilize NetForge_RL Framework

   Employ NetForge_RL as the specific platform for developing and testing advanced multi-agent cyber defense strategies based on temporal graph networks.
4. 4
   Prioritize Asynchronous Interactions

   Design MARL policies to accurately reflect real-world attack and defense dynamics, emphasizing asynchronous multi-agent interactions over idealized, step-by-step processes.
5. 5
   Develop Deployable AI Agents

   Shift MARL policy design, training, and evaluation to account for dynamic, real-time interactions, aiming for robust and adaptable AI agents deployable in operational SOCs.

import networkx as nx
import time

class CyberEvent:
    def __init__(self, event_id, event_type, timestamp, source, target=None):
        self.event_id = event_id
        self.event_type = event_type
        self.timestamp = timestamp
        self.source = source
        self.target = target

    def __repr__(self):
        return f"Event(ID:{self.event_id}, Type:{self.event_type}, Time:{self.timestamp}, Src:{self.source})"

class TemporalGraph:
    def __init__(self):
        self.graph = nx.DiGraph()
        self.events = []

    def add_event(self, event: CyberEvent):
        self.events.append(event)
        # Add nodes if they don't exist
        if event.source not in self.graph:
            self.graph.add_node(event.source, type='agent')
        if event.target and event.target not in self.graph:
            self.graph.add_node(event.target, type='resource')
        
        # Add a temporal edge based on the event
        if event.target:
            self.graph.add_edge(event.source, event.target, event_type=event.event_type, timestamp=event.timestamp)

    def get_events_in_time_window(self, start_time, end_time):
        return [e for e in self.events if start_time <= e.timestamp <= end_time]

# Example Usage:
tg = TemporalGraph()

# Simulate some asynchronous events
tg.add_event(CyberEvent(1, 'scan', time.time(), 'attacker_agent', 'target_server_A'))
time.sleep(0.1)
tg.add_event(CyberEvent(2, 'detect', time.time(), 'defender_agent_1', 'target_server_A'))
time.sleep(0.05)
tg.add_event(CyberEvent(3, 'exploit_attempt', time.time(), 'attacker_agent', 'target_server_A'))

print("Current Graph Nodes:", tg.graph.nodes)
print("Current Graph Edges:", tg.graph.edges(data=True))
print("All Events:", tg.events)

• Paperarxiv.org