Base Agent

The BaseAgent class provides the foundational interface for all agents interacting with the NetSecGame environment. It handles TCP socket communication with the game server, agent registration, game reset requests, and the core action-observation loop.

All custom agents should extend this class and implement their decision-making logic by overriding a method like choose_action (see Getting Started for an example).

netsecgame.agents.base_agent.BaseAgent

Bases: ABC

Basic agent for the network based NetSecGame environment. Implemenets communication with the game server.

Initializes the BaseAgent and connects it to the game server.

Parameters:

Name	Type	Description	Default
host	str	The host where the game server is running.	required
port	int	The port where the game server is running.	required
role	AgentRole	The assigned role of the agent (e.g., Attacker, Defender).	required

Source code in netsecgame/agents/base_agent.py

def __init__(self, host, port, role:AgentRole)->None:
    """
    Initializes the BaseAgent and connects it to the game server.

    Args:
        host (str): The host where the game server is running.
        port (int): The port where the game server is running.
        role (AgentRole): The assigned role of the agent (e.g., Attacker, Defender).
    """
    self._connection_details = (host, port)
    self._logger = logging.getLogger(self.__class__.__name__)
    self._role = role
    try:
        self._socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self._socket.connect((host, port))
    except socket.error as e:
        self._logger.error(f"Socket error: {e}")
        self._socket = None
    self._logger.info("Agent created")

logger property

Returns the logger instance for this agent.

__del__

In case the extending class did not close the connection, terminate the socket when the object is deleted.

Source code in netsecgame/agents/base_agent.py

def __del__(self):
    "In case the extending class did not close the connection, terminate the socket when the object is deleted."
    if self._socket:
        try:
            self._socket.close()
            self._logger.info("Socket closed")
        except socket.error as e:
            self._logger.error(f"Error closing socket: {e}")

communicate

Exchanges data with the server and returns the server's response.

Sends an Action object to the server and waits for a response. The response is expected to be a JSON-encoded string containing status, observation, and message fields.

Parameters:

Name	Type	Description	Default
data	Action	The action to send to the server.	required

Returns:

Type	Description
Tuple[GameStatus, Dict[str, Any], Optional[str]]	Tuple[GameStatus, Dict[str, Any], Optional[str]]: A tuple containing: - status (GameStatus): The status parsed from the server response. - observation (Dict[str, Any]): The observation data from the server. - message (Optional[str]): An optional message from the server.

Raises:

Type	Description
ValueError	If data is not of type Action.
ConnectionError	If the server response is incomplete.
Exception	If there is an error during communication.

Source code in netsecgame/agents/base_agent.py

def communicate(self, data: Action) -> Tuple[GameStatus, Dict[str, Any], Optional[str]]:
    """
    Exchanges data with the server and returns the server's response.

    Sends an `Action` object to the server and waits for a response.
    The response is expected to be a JSON-encoded string containing status, observation, and message fields.

    Args:
        data (Action): The action to send to the server.

    Returns:
        Tuple[GameStatus, Dict[str, Any], Optional[str]]: A tuple containing:
            - status (GameStatus): The status parsed from the server response.
            - observation (Dict[str, Any]): The observation data from the server.
            - message (Optional[str]): An optional message from the server.

    Raises:
        ValueError: If `data` is not of type `Action`.
        ConnectionError: If the server response is incomplete.
        Exception: If there is an error during communication.
    """

    def _send_data(socket: socket.socket, msg: str) -> None:
        """Internal method to send data over the socket."""
        try:
            self._logger.debug(f'Sending: {msg}')
            socket.sendall(msg.encode())
        except Exception as e:
            self._logger.error(f'Exception in _send_data(): {e}')
            raise e

    def _receive_data(socket: socket.socket) -> Tuple[GameStatus, Dict[str, Any], Optional[str]]:
        """Internal method to receive data from the socket."""
        # Receive data from the server
        data = b""  # Initialize an empty byte string

        while True:
            chunk = socket.recv(ProtocolConfig.BUFFER_SIZE)  # Receive a chunk
            if not chunk:  # If no more data, break (connection closed)
                break
            data += chunk
            if ProtocolConfig.END_OF_MESSAGE in data:  # Check if EOF marker is present
                break
        if ProtocolConfig.END_OF_MESSAGE not in data:
            raise ConnectionError("Unfinished connection.")
        data = data.replace(ProtocolConfig.END_OF_MESSAGE, b"")  # Remove EOF marker
        data = data.decode() 
        self._logger.debug(f"Data received from env: {data}")
        # extract data from string representation
        data_dict = json.loads(data)
        # Add default values if dict keys are missing
        status = data_dict["status"] if "status" in data_dict else ""
        observation = data_dict["observation"] if "observation" in data_dict else {}
        message = data_dict["message"] if "message" in data_dict else None

        return GameStatus.from_string(str(status)), observation, message

    if isinstance(data, Action):
        data = data.to_json()
    else:
        raise ValueError("Incorrect data type! Data should be ONLY of type Action")

    _send_data(self._socket, data)
    return _receive_data(self._socket)

make_step

Executes a single step in the environment by sending the agent's action to the server and receiving the resulting observation.

Parameters:

Name	Type	Description	Default
action	Action	The action to be performed by the agent.	required

Returns:

Name	Type	Description
Observation	Optional[Observation]	The new observation received from the server, containing the updated game state, reward, end flag, and additional info.
None	Optional[Observation]	If no observation is received from the server.

Raises:

Type	Description
Exception	Any exceptions raised by the communicate method are propagated.

Source code in netsecgame/agents/base_agent.py

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.

    Args:
        action (Action): The action to be performed by the agent.

    Returns:
        Observation: The new observation received from the server, containing the updated game state, reward, end flag, and additional info.
        None: If no observation is received from the server.

    Raises:
        Exception: Any exceptions raised by the `communicate` method are propagated.
    """
    _, observation_dict, _ = self.communicate(action)
    if observation_dict:
        return Observation(GameState.from_dict(observation_dict["state"]), observation_dict["reward"], observation_dict["end"], observation_dict["info"])
    else:
        return None

register

Registers the agent with the game server.

Returns:

Type	Description
Optional[Observation]	Optional[Observation]: Initial observation if successful, None otherwise.

Source code in netsecgame/agents/base_agent.py

def register(self) -> Optional[Observation]:
    """
    Registers the agent with the game server.

    Returns:
        Optional[Observation]: Initial observation if successful, None otherwise.
    """
    try:
        self._logger.info(f'Registering agent as {self.role}')
        status, observation_dict, message = self.communicate(Action(ActionType.JoinGame,
                                                                     parameters={"agent_info":AgentInfo(self.__class__.__name__,self.role.value)}))
        if status is GameStatus.CREATED:
            self._logger.info(f"\tRegistration successful! {message}")
            return Observation(GameState.from_dict(observation_dict["state"]), observation_dict["reward"], observation_dict["end"], message)
        else:
            self._logger.error(f'\tRegistration failed! (status: {status}, msg:{message}')
            return None
    except Exception as e:
        self._logger.error(f'Exception in register(): {e}')

request_game_reset

Request a game reset from the server. Args: 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: The initial observation after the reset if successful, None otherwise.

Source code in netsecgame/agents/base_agent.py

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: 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:
        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, "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:
        self._logger.error(f'\rReset failed! (status: {status}, msg:{message}')
        return None

terminate_connection

Method for graceful termination of connection. Should be used by any class extending the BaseAgent.

Source code in netsecgame/agents/base_agent.py

def terminate_connection(self)->None:
    """Method for graceful termination of connection. Should be used by any class extending the BaseAgent."""
    if self._socket:
        try:
            self._socket.close()
            self._socket = None
            self._logger.info("Socket closed")
        except socket.error as e:
            self._logger.error(f"Error closing socket: {e}")