Ondra add topology randomization seed

.dockerignore

@@ -1,3 +1,4 @@
+.coverage
 .git
 .github
 .gitignore
@@ -16,10 +17,9 @@ netsecgame.egg-info/
 notebooks/
 NetSecGameAgents/
 site/
-tests/
 trajectories/
 readme_images/
 tests/
 *trajectories*.json
-README.md
+README*.md
 mkdocs.yml

Dockerfile

@@ -1,26 +1,20 @@
 # Use an official Python 3.12 runtime as a parent image
-FROM python:3.12.10-slim
+FROM python:3.12.12-slim-bookworm
 
 # Set the working directory in the container
 ENV DESTINATION_DIR=/netsecgame
+WORKDIR ${DESTINATION_DIR}
 
+# Copy the source code FIRST so pip has access to pyproject.toml
+COPY . ${DESTINATION_DIR}/
 
-# Install system dependencies
+# The "Single Layer" Trick: Install tools, build app, purge tools
 RUN apt-get update && \
-    apt-get install -y --no-install-recommends \
-    git \
-    build-essential \
-    && rm -rf /var/lib/apt/lists/*
-RUN pip install --upgrade pip
-
-COPY .  ${DESTINATION_DIR}/
-
-# Set the working directory in the container
-WORKDIR  ${DESTINATION_DIR}
-
-# Install any necessary Python dependencies
-# If a requirements.txt file is in the repository
-RUN if [ -f pyproject.toml ]; then pip install .[server] ; fi
+    apt-get install -y --no-install-recommends build-essential && \
+    pip install --no-cache-dir --upgrade pip && \
+    if [ -f pyproject.toml ]; then pip install --no-cache-dir .[server] ; fi && \
+    apt-get purge -y --auto-remove build-essential && \
+    rm -rf /var/lib/apt/lists/*
 
 ARG GAME_MODULE="netsecgame.game.worlds.NetSecGame"
 # Pass the build argument to an environment variable so CMD can use it
@@ -29,8 +23,8 @@ ENV ENV_GAME_MODULE=$GAME_MODULE
 # Expose the port the coordinator will run on
 EXPOSE 9000 
 
-# Run the Python script when the container launches (with default arguments --task_config=netsecenv_conf.yaml --game_port=9000 --game_host=0.0.0.0)
+# Run the Python script when the container launches
 ENTRYPOINT ["sh", "-c", "exec python3 -m ${ENV_GAME_MODULE} --task_config=netsecenv_conf.yaml --game_port=9000 --game_host=0.0.0.0 \"$@\"", "--"]
 
 # Default command arguments (can be overridden at runtime)
-CMD ["--debug_level=INFO"]
+CMD ["--debug_level=INFO"]

examples/example_task_configuration.yaml

@@ -10,7 +10,6 @@ coordinator:
       max_steps: 50
       goal:
         description: "Exfiltrate data from Samba server to remote C&C server (213.47.23.195)."
-        is_any_part_of_goal_random: True
         known_networks: []
         known_hosts: []
         controlled_hosts: []
@@ -20,15 +19,14 @@ coordinator:
       start_position: # Defined starting position of the attacker
         known_networks: []
         known_hosts: []
-        controlled_hosts: [213.47.23.195, random] #
+        controlled_hosts: [213.47.23.195, 192.168.1.1] #
         known_services: {}
         known_data: {}
         known_blocks: {}
 
     Defender:
       goal:
         description: "Block all attackers"
-        is_any_part_of_goal_random: False
         known_networks: []
         known_hosts: []
         controlled_hosts: []
@@ -48,7 +46,7 @@ coordinator:
 env:
   scenario: 'two_networks_tiny' # use the smallest topology for this example
   use_global_defender: False # Do not use global SIEM Defender
-  use_dynamic_addresses: False # Do not randomize IP addresses
+  use_dynamic_addresses: True # Do not randomize IP addresses
   use_firewall: True # Use firewall
   save_trajectories: False # Do not store trajectories
   required_players: 1

netsecgame/agents/base_agent.py

@@ -3,7 +3,8 @@
 import logging
 import socket
 import json
-from abc import ABC 
+from abc import ABC
+from typing import Optional, Tuple, Dict, Any
 
 from netsecgame.game_components import Action, GameState, Observation, ActionType, GameStatus, AgentInfo, ProtocolConfig, AgentRole
 
@@ -55,7 +56,7 @@ def role(self)->str:
     def logger(self)->logging.Logger:
         return self._logger
 
-    def make_step(self, action: Action) -> Observation | None:
+    def make_step(self, action: Action) -> Optional[Observation]:
         """
         Executes a single step in the environment by sending the agent's action to the server and receiving the resulting observation.
 
@@ -75,7 +76,7 @@ def make_step(self, action: Action) -> Observation | None:
         else:
             return None
 
-    def communicate(self, data:Action)-> tuple:
+    def communicate(self, data:Action)-> Tuple[GameStatus, Dict[str, Any], Optional[str]]:
         """
         Exchanges data with the server and returns the server's response.
         This method sends an `Action` object to the server and waits for a response.
@@ -102,7 +103,7 @@ def _send_data(socket, msg:str)->None:
                 self._logger.error(f'Exception in _send_data(): {e}')
                 raise e
 
-        def _receive_data(socket)->tuple:
+        def _receive_data(socket)->Tuple[GameStatus, Dict[str, Any], Optional[str]]:
             """
             Receive data from server
             """
@@ -138,7 +139,7 @@ def _receive_data(socket)->tuple:
         _send_data(self._socket, data)
         return _receive_data(self._socket)
 
-    def register(self)->Observation | None:
+    def register(self)->Optional[Observation]:
         """
         Method for registering agent to the game server.
         Classname is used as agent name and the role is based on the 'role' argument.
@@ -162,19 +163,28 @@ def register(self)->Observation | None:
         except Exception as e:
             self._logger.error(f'Exception in register(): {e}')
 
-    def request_game_reset(self, request_trajectory=False, randomize_topology=True, randomize_topology_seed=None) -> Observation|None:
-        """
-        Requests a game reset from the server. Optionally requests a trajectory and/or topology randomization.
+    def request_game_reset(
+        self, 
+        request_trajectory: bool = False, 
+        randomize_topology: bool = False, 
+        seed: Optional[int] = None
+    ) -> Optional[Observation]:
+        """Request a game reset from the server.
         Args:
-            request_trajectory (bool): If True, requests the server to provide a trajectory of the last episode.
-            randomize_topology (bool): If True, requests the server to randomize the network topology for the next episode. Defaults to True.
-            randomize_topology_seed (int): If provided, requests the server to use this seed for randomizing the network topology. Defaults to None.
+            request_trajectory: If True, requests the server to provide a 
+                trajectory of the last episode.
+            randomize_topology: If True, requests the server to randomize the 
+                network topology for the next episode. Defaults to False.
+            seed: If provided, requests the server to use this seed for 
+                randomizing the environment. Required if randomize_topology is True.
         Returns:
-            Observation: The initial observation after the reset if successful, None otherwise.
+            The initial observation after the reset if successful, None otherwise.
         """
+        if seed is None and randomize_topology:
+            raise ValueError("Topology randomization without seed is not supported.")
         self._logger.debug("Requesting game reset")
-        status, observation_dict, message = self.communicate(Action(ActionType.ResetGame, parameters={"request_trajectory": request_trajectory, "randomize_topology": randomize_topology}))
-        if status:
+        status, observation_dict, message = self.communicate(Action(ActionType.ResetGame, parameters={"request_trajectory": request_trajectory, "randomize_topology": randomize_topology, "seed": seed}))
+        if status is GameStatus.RESET_DONE:
             self._logger.debug('\tReset successful')
             return Observation(GameState.from_dict(observation_dict["state"]), observation_dict["reward"], observation_dict["end"], message)
         else:

netsecgame/game/config_parser.py

@@ -3,13 +3,12 @@
 # Author: Ondrej Lukas, ondrej.lukas@aic.fel.cvut.cz
 
 import yaml
-# This is used so the agent can see the environment and game components
-import importlib
-from netsecgame.game_components import IP, Data, Network, Service
 import netaddr
 import logging
 from random import randint
 from  typing import Optional
+from netsecgame.game_components import IP, Data, Network, Service
+from netsecgame.game.scenarios import SCENARIO_REGISTRY
 
 class ConfigParser():
     """
@@ -386,25 +385,15 @@ def get_scenario(self):
         """
         Get the scenario config objects based on the configuration. Only import objects that are selected via importlib.
         """
-        allowed_names = {
-            "scenario1" : "netsecgame.game.scenarios.scenario_configuration",
-            "scenario1_small" : "netsecgame.game.scenarios.smaller_scenario_configuration",
-            "scenario1_tiny" : "netsecgame.game.scenarios.tiny_scenario_configuration",
-            "one_network": "netsecgame.game.scenarios.one_net",
-            "three_net_scenario": "netsecgame.game.scenarios.three_net_scenario",
-            "two_networks": "netsecgame.game.scenarios.two_nets", # same as scenario1
-            "two_networks_small": "netsecgame.game.scenarios.two_nets_small", # same as scenario1_small
-            "two_networks_tiny": "netsecgame.game.scenarios.two_nets_tiny", # same as scenario1_small
-
-        }
         scenario_name = self.config['env']['scenario']
         # make sure to validate the input
-        if scenario_name not in allowed_names:
-            raise ValueError(f"Unsupported scenario: {scenario_name}")
+        if scenario_name not in SCENARIO_REGISTRY:
+            raise ValueError(
+                f"Unsupported scenario: {scenario_name}. "
+                f"Available scenarios: {list(SCENARIO_REGISTRY.keys())}"
+            )
 
-        # import the correct module
-        module = importlib.import_module(allowed_names[scenario_name])
-        return module.configuration_objects
+        return SCENARIO_REGISTRY[scenario_name]
 
     def get_seed(self, whom):
         """
@@ -419,11 +408,13 @@ def get_randomize_goal_every_episode(self, default_value: bool = False) -> bool:
         """
         Get if the randomization should be done only once or at the beginning of every episode
         """
+        # TODO Remove in future
         try:
             randomize_goal_every_episode = self.config["coordinator"]["agents"]["attackers"]["goal"]["is_any_part_of_goal_random"]
         except KeyError:
             # Option is not in the configuration - default to FALSE
             randomize_goal_every_episode = default_value
+        raise DeprecationWarning("This function is deprecated.")
         return randomize_goal_every_episode
 
     def get_use_firewall(self, default_value: bool = False)->bool: