Configuration manager

Configuration manager is a component of the game coordinator that handles the configuration of the game. It is responsible for loading the configuration from the YAML file and providing it to the game coordinator.

netsecgame.game.configuration_manager.ConfigurationManager

Manages the loading and access of game configuration.

Handles fetching configuration from efficient sources (local file or remote service) and provides structured access to configuration data.

Source code in netsecgame/game/configuration_manager.py

def __init__(self, task_config_file: Optional[str] = None, service_host: Optional[str] = None, service_port: Optional[int] = None):
    self.logger = logging.getLogger("ConfigurationManager")
    self._task_config_file = task_config_file
    self._service_host = service_host
    self._service_port = service_port

    self._parser: Optional[ConfigParser] = None
    self._cyst_objects = None
    self._config_file_hash: Optional[str] = None

    # Cache for parsed values
    self._starting_positions: Dict[str, Any] = {}
    self._win_conditions: Dict[str, Any] = {}
    self._goal_descriptions: Dict[str, str] = {}
    self._max_steps: Dict[str, Optional[int]] = {}

_fetch_remote_configuration async

Fetches initialization objects from the remote service.

Source code in netsecgame/game/configuration_manager.py

async def _fetch_remote_configuration(self) -> None:
    """Fetches initialization objects from the remote service."""
    url = f"http://{self._service_host}:{self._service_port}/cyst_init_objects"
    async with ClientSession() as session:
        try:
            async with session.get(url) as response:
                if response.status == 200:
                    config_data = await response.json()
                    self.logger.debug(f"Received config data: {config_data}")

                    # Initialize CYST environment
                    env = Environment.create()
                    self._config_file_hash = get_str_hash(config_data)
                    self._cyst_objects = env.configuration.general.load_configuration(config_data)
                    self.logger.debug(f"Initialization objects received: {self._cyst_objects}")

                    # Initialize parser with the fetched dict (assuming it contains task_configuration or similar structure)
                    # Note: The original coordinator code for remote fetch commented out creating ConfigParser:
                    # #self.task_config = ConfigParser(config_dict=response["task_configuration"])
                    # usage of self.task_config in original code fell back to loading from file even if remote fetch happened?
                    # "Temporary fix" comment in original code suggests fallback.
                    # For this implementation, we should try to use the fetched config if possible.
                    # If the API returns the same structure as the YAML file, we can pass it to ConfigParser(config_dict=...)
                    # If not, we might need to rely on the file as the original code did for the parser part.

                    # Let's assume for now we try to use the dictionary if available, otherwise fallback logic might be needed
                    # derived from how the response is structured.
                    # Looking at original code: response seems to be the full config.
                    self._parser = ConfigParser(config_dict=config_data)

                else:
                    self.logger.error(f"Failed to fetch initialization objects. Status: {response.status}")
                    raise RuntimeError(f"Remote configuration fetch failed with status {response.status}")
        except Exception as e:
            self.logger.error(f"Error fetching initialization objects: {e}")
            # Fallback to local file if remote fails? The original code did:
            # self.task_config = ConfigParser(self._task_config_file)
            # We can implement similar fallback behavior here if desired, or just raise.
            if self._task_config_file:
                self.logger.warning("Falling back to local configuration file.")
                self._load_local_configuration()
            else:
                raise e

_load_local_configuration

Loads configuration from the local file.

Source code in netsecgame/game/configuration_manager.py

def _load_local_configuration(self) -> None:
    """Loads configuration from the local file."""
    self._parser = ConfigParser(task_config_file=self._task_config_file)
    self._cyst_objects = self._parser.get_scenario()
    # Original code does str(self._cyst_objects) for hash
    self._config_file_hash = get_str_hash(str(self._cyst_objects))

get_all_goal_descriptions

Returns goal descriptions for all roles.

Returns:

Type	Description
Dict[str, str]	Dict[str, str]: The goal descriptions for all roles.

Source code in netsecgame/game/configuration_manager.py

def get_all_goal_descriptions(self) -> Dict[str, str]:
    """Returns goal descriptions for all roles.

    Returns:
        Dict[str, str]: The goal descriptions for all roles.
    """
    goal_descriptions = {}
    for agent_role in AgentRole:
        try:
            goal_descriptions[agent_role] = self.get_goal_description(role=agent_role)
        except KeyError:
            goal_descriptions[agent_role] = ""
        self.logger.info(f"Goal description for role '{agent_role}': {goal_descriptions[agent_role]}")
    return goal_descriptions

get_all_max_steps

Returns max steps for all roles.

Returns:

Type	Description
Dict[str, Optional[int]]	Dict[str, Optional[int]]: A dictionary mapping roles to their maximum steps.

Source code in netsecgame/game/configuration_manager.py

def get_all_max_steps(self) -> Dict[str, Optional[int]]:
    """
    Returns max steps for all roles.

    Returns:
        Dict[str, Optional[int]]: A dictionary mapping roles to their maximum steps.
    """
    # Using self.get_max_steps might raise RuntimeError if checks are there, 
    # but simpler to just call parser directly or the single accessor since we are inside the class.
    # However, the single accessor has the check.
    # But wait, self.get_max_steps(role) does `self._parser.get_max_steps(role)` already.
    # Iterating over AgentRole is correct.
    return {role: self.get_max_steps(role) for role in AgentRole}

get_all_starting_positions

Returns starting positions for all roles.

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: The starting positions for all roles.

Source code in netsecgame/game/configuration_manager.py

def get_all_starting_positions(self) -> Dict[str, Any]:
    """Returns starting positions for all roles.

    Returns:
        Dict[str, Any]: The starting positions for all roles.
    """
    starting_positions = {}
    for agent_role in AgentRole:
        try:
            starting_positions[agent_role] = self.get_starting_position(role=agent_role)
            self.logger.info(f"Starting position for role '{agent_role}': {starting_positions[agent_role]}")
        except KeyError:
            starting_positions[agent_role] = {}
    return starting_positions

get_all_win_conditions

Returns win conditions for all roles.

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: The win conditions for all roles.

Source code in netsecgame/game/configuration_manager.py

def get_all_win_conditions(self) -> Dict[str, Any]:
    """Returns win conditions for all roles.

    Returns:
        Dict[str, Any]: The win conditions for all roles.
    """
    win_conditions = {}
    for agent_role in AgentRole:
        try:
            win_conditions[agent_role] = self.get_win_conditions(role=agent_role)
        except KeyError:
            win_conditions[agent_role] = {}
        self.logger.info(f"Win condition for role '{agent_role}': {win_conditions[agent_role]}")
    return win_conditions

get_config_hash

Returns the hash of the loaded configuration.

Returns:

Type	Description
Optional[str]	Optional[str]: The hexadecimal hash of the configuration, or None if not loaded.

Source code in netsecgame/game/configuration_manager.py

def get_config_hash(self) -> Optional[str]:
    """
    Returns the hash of the loaded configuration.

    Returns:
        Optional[str]: The hexadecimal hash of the configuration, or None if not loaded.
    """
    return self._config_file_hash

get_cyst_objects

Returns the loaded CYST configuration objects.

Returns:

Name	Type	Description
Any	Iterable[Any]	The CYST configuration objects (usually a list of NodeConfig, etc.).

Source code in netsecgame/game/configuration_manager.py

def get_cyst_objects(self) -> Iterable[Any]:
    """
    Returns the loaded CYST configuration objects.

    Returns:
        Any: The CYST configuration objects (usually a list of NodeConfig, etc.).
    """
    return self._cyst_objects

get_goal_description

Returns the goal description for a specific role. Args: role (str): The role of the agent.

Returns:

Name	Type	Description
str	str	The goal description for the specified role.

Source code in netsecgame/game/configuration_manager.py

def get_goal_description(self, role: str) -> str:
    """Returns the goal description for a specific role.
    Args:
        role (str): The role of the agent.

    Returns:
        str: The goal description for the specified role.
    """
    if not self._parser:
        raise RuntimeError("Configuration not loaded.")
    return self._parser.get_goal_description(agent_role=role)

get_max_steps

Returns the max steps for a specific role. Args: role (str): The role of the agent.

Returns:

Type	Description
Optional[int]	Optional[int]: The max steps for the specified role.

Source code in netsecgame/game/configuration_manager.py

def get_max_steps(self, role: str) -> Optional[int]:
    """Returns the max steps for a specific role.
    Args:
        role (str): The role of the agent.

    Returns:
        Optional[int]: The max steps for the specified role.
    """
    if not self._parser:
        raise RuntimeError("Configuration not loaded.")
    return self._parser.get_max_steps(role)

get_required_num_players

Returns the required number of players configuration.

Parameters:

Name	Type	Description	Default
default_value	int	The default value for the required number of players.	1

Returns:

Name	Type	Description
int	int	The required number of players configuration.

Source code in netsecgame/game/configuration_manager.py

def get_required_num_players(self, default_value: int = 1) -> int:
    """Returns the required number of players configuration.

    Args:
        default_value (int): The default value for the required number of players.

    Returns:
        int: The required number of players configuration.
    """
    if not self._parser:
        raise RuntimeError("Configuration not loaded.")
    return self._parser.get_required_num_players(default_value)

get_rewards

Returns the rewards configuration. Args: reward_names (List[str]): The names of the rewards. default_value (int): The default value for the rewards.

Returns:

Name	Type	Description
dict	Dict[str, Any]	The rewards configuration.

Source code in netsecgame/game/configuration_manager.py

def get_rewards(self, reward_names: List[str] = ["step", "success", "fail", "false_positive"], default_value: int = 0) -> Dict[str, Any]:
    """Returns the rewards configuration.
    Args:
        reward_names (List[str]): The names of the rewards.
        default_value (int): The default value for the rewards.

    Returns:
        dict: The rewards configuration.
    """
    if not self._parser:
        raise RuntimeError("Configuration not loaded.")
    rewards = self._parser.get_rewards(reward_names, default_value)
    if rewards.get("fail") > 0:
        self.logger.warning("Fail reward is positive. This is not recommended.")
    if rewards.get("false_positive") > 0:
        self.logger.warning("False positive reward is positive. This is not recommended.")
    if rewards.get("success") < 0:
        self.logger.warning("Success reward is negative. This is not recommended.")
    return rewards

get_starting_position

Returns the starting position configuration for a specific role. Args: role (str): The role of the agent.

Returns:

Name	Type	Description
dict	Dict[str, Any]	The starting position configuration for the specified role.

Source code in netsecgame/game/configuration_manager.py

def get_starting_position(self, role: str) -> Dict[str, Any]:
    """Returns the starting position configuration for a specific role.
    Args:
        role (str): The role of the agent.

    Returns:
        dict: The starting position configuration for the specified role.
    """
    if not self._parser:
        raise RuntimeError("Configuration not loaded.")
    return self._parser.get_start_position(agent_role=role)

get_store_trajectories

Returns the store trajectories configuration.

Parameters:

Name	Type	Description	Default
default_value	bool	The default value for the store trajectories.	False

Returns:

Name	Type	Description
bool	bool	True if trajectories should be stored, False otherwise.

Source code in netsecgame/game/configuration_manager.py

def get_store_trajectories(self, default_value: bool = False) -> bool:  
    """Returns the store trajectories configuration.

    Args:
        default_value (bool): The default value for the store trajectories.

    Returns:
        bool: True if trajectories should be stored, False otherwise.
    """
    if not self._parser:
        raise RuntimeError("Configuration not loaded.")
    return self._parser.get_store_trajectories(default_value)

get_use_dynamic_addresses

Returns the use dynamic addresses configuration. Args: default_value (bool): The default value for the use dynamic addresses.

Returns:

Name	Type	Description
bool	bool	The use dynamic addresses configuration.

Source code in netsecgame/game/configuration_manager.py

def get_use_dynamic_addresses(self, default_value: bool = False) -> bool:
    """Returns the use dynamic addresses configuration.
    Args:
        default_value (bool): The default value for the use dynamic addresses.

    Returns:
        bool: The use dynamic addresses configuration.
    """
    if not self._parser:
        raise RuntimeError("Configuration not loaded.")
    return self._parser.get_use_dynamic_addresses(default_value)

get_use_firewall

Returns the use firewall configuration.

Parameters:

Name	Type	Description	Default
default_value	bool	The default value for the use firewall.	True

Returns:

Name	Type	Description
bool	bool	The use firewall configuration.

Source code in netsecgame/game/configuration_manager.py

def get_use_firewall(self, default_value: bool = True) -> bool:
    """Returns the use firewall configuration.

    Args:
        default_value (bool): The default value for the use firewall.

    Returns:
        bool: The use firewall configuration.
    """
    if not self._parser:
        raise RuntimeError("Configuration not loaded.")
    return self._parser.get_use_firewall(default_value)

get_use_global_defender

Returns the use global defender configuration. Args: default_value (bool): The default value for the use global defender.

Returns:

Name	Type	Description
bool	bool	The use global defender configuration.

Source code in netsecgame/game/configuration_manager.py

def get_use_global_defender(self, default_value: bool = False) -> bool:
    """Returns the use global defender configuration.
    Args:
        default_value (bool): The default value for the use global defender.

    Returns:
        bool: The use global defender configuration.
    """
    if not self._parser:
        raise RuntimeError("Configuration not loaded.")
    return self._parser.get_use_global_defender(default_value)

get_win_conditions

Returns the win conditions for a specific role. Args: role (str): The role of the agent.

Returns:

Name	Type	Description
dict	Dict[str, Any]	The win conditions for the specified role.

Source code in netsecgame/game/configuration_manager.py

def get_win_conditions(self, role: str) -> Dict[str, Any]:
    """Returns the win conditions for a specific role.
    Args:
        role (str): The role of the agent.

    Returns:
        dict: The win conditions for the specified role.
    """
    if not self._parser:
        raise RuntimeError("Configuration not loaded.")
    return self._parser.get_win_conditions(agent_role=role)

load async

Determines the source and loads the configuration. Prioritizes remote service if configured, otherwise falls back to local file.

Source code in netsecgame/game/configuration_manager.py

async def load(self) -> None:
    """
    Determines the source and loads the configuration.
    Prioritizes remote service if configured, otherwise falls back to local file.
    """
    if self._service_host and self._service_port:
        self.logger.info(f"Fetching task configuration from {self._service_host}:{self._service_port}")
        await self._fetch_remote_configuration()
    elif self._task_config_file:
        self.logger.info(f"Loading task configuration from file: {self._task_config_file}")
        self._load_local_configuration()
    else:
        raise ValueError("Task configuration source not specified (neither file nor service)")