Add first draft of MAL Petting Zoo Simulator

.github/workflows/publish-to-pypi-and-test-pypi.yml

@@ -0,0 +1,74 @@
+name: Publish Python distributions to PyPI and TestPyPI
+
+on: push
+
+jobs:
+    build:
+        name: Build Python distribution
+        runs-on: ubuntu-latest
+
+        steps:
+        - uses: actions/checkout@v4
+        - name: Set up Python
+          uses: actions/setup-python@v4
+          with:
+            python-version: "3.x"
+
+        - name: Install pypa/build
+          run: >-
+            python3 -m
+            pip install
+            build
+            --user
+
+        - name: Build a binary wheel and a source tarball
+          run: python3 -m build
+        - name: Store the distribution packages
+          uses: actions/upload-artifact@v3
+          with:
+            name: python-package-distributions
+            path: dist/
+
+    publish-to-pypi:
+      name: Publish Python distribution to PyPI
+      if: startsWith(github.ref, 'refs/tags/')  # only publish to PyPI on tag pushes
+      needs:
+      - build
+      runs-on: ubuntu-latest
+      environment:
+        name: pypi
+        url: https://pypi.org/p/mal-petting-zoo-simulator
+      permissions:
+        id-token: write  # IMPORTANT: mandatory for trusted publishing
+
+      steps:
+      - name: Download all the dists
+        uses: actions/download-artifact@v3
+        with:
+          name: python-package-distributions
+          path: dist/
+      - name: Publish distribution to PyPI
+        uses: pypa/gh-action-pypi-publish@release/v1
+
+    publish-to-testpypi:
+      name: Publish Python distribution to TestPyPI
+      if: startsWith(github.ref, 'refs/tags/')  # only publish to PyPI on tag pushes
+      needs:
+      - build
+      runs-on: ubuntu-latest
+      environment:
+        name: testpypi
+        url: https://test.pypi.org/p/mal-petting-zoo-simulator
+      permissions:
+        id-token: write  # IMPORTANT: mandatory for trusted publishing
+
+      steps:
+      - name: Download all the dists
+        uses: actions/download-artifact@v3
+        with:
+          name: python-package-distributions
+          path: dist/
+      - name: Publish distribution to TestPyPI
+        uses: pypa/gh-action-pypi-publish@release/v1
+        with:
+          repository-url: https://test.pypi.org/legacy/

.gitignore

@@ -1,3 +1,14 @@
+# Project specific temp files
+*.json
+*.scad
+*.log
+*.yaml
+*.yml
+*.sCAD
+*.swp
+*.swo
+tmp/
+
 # Byte-compiled / optimized / DLL files
 __pycache__/
 *.py[cod]

AUTHORS

@@ -0,0 +1,2 @@
+Andrei Buhaiu <[email protected]>
+Jakob Nyberg <[email protected]>

README.md

@@ -1 +1,4 @@
-# mal-petting-zoo-simulator
+# Overview
+
+A MAL compliant Petting Zoo simulator.
+

malpzsim/__init__.py

@@ -0,0 +1,31 @@
+# -*- encoding: utf-8 -*-
+# MAL Petting Zoo Simulator v0.0.3
+# Copyright 2024, Andrei Buhaiu.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+
+"""
+MAL Petting Zoo Simulator
+"""
+
+__title__ = 'malpzsim'
+__version__ = '0.0.3'
+__authors__ = ['Andrei Buhaiu',
+    'Jakob Nyberg']
+__license__ = 'Apache 2.0'
+__docformat__ = 'restructuredtext en'
+
+__all__ = ()
+

malpzsim/agents/keyboard_input.py

@@ -0,0 +1,61 @@
+import numpy as np
+import logging
+
+AGENT_ATTACKER = 'attacker'
+AGENT_DEFENDER = 'defender'
+
+logger = logging.getLogger(__name__)
+
+null_action = (0, None)
+
+class KeyboardAgent():
+    def __init__(self, vocab):
+        logger.debug('Create Keyboard agent.')
+        self.vocab = vocab
+
+    def compute_action_from_dict(self, obs: dict, mask: tuple) -> tuple:
+        def valid_action(user_input: str) -> bool:
+            if user_input == "":
+                return True
+
+            try:
+                node = int(user_input)
+            except ValueError:
+                return False
+
+            try:
+                a = associated_action[action_strings[node]]
+            except IndexError:
+                return False
+
+            if a == 0:
+                return True  # wait is always valid
+            return node < len(available_actions) and node >= 0
+
+        def get_action_object(user_input: str) -> tuple:
+            node = int(user_input) if user_input != "" else None
+            action = associated_action[action_strings[node]] if user_input != "" else 0
+            return node, action
+
+        available_actions = np.flatnonzero(mask[1])
+
+        action_strings = [self.vocab[i] for i in available_actions]
+        associated_action = {i: 1 for i in action_strings}
+        action_strings += ["wait"]
+        associated_action["wait"] = 0
+
+        user_input = "xxx"
+        while not valid_action(user_input):
+            print("Available actions:")
+            print("\n".join([f"{i}. {a}" for i, a in enumerate(action_strings)]))
+            print("Enter action or leave empty to wait:")
+            user_input = input("> ")
+
+            if not valid_action(user_input):
+                print("Invalid action.")
+
+        node, a = get_action_object(user_input)
+        print(f"Selected action: {action_strings[node] if node is not None else 'wait'}")
+
+        return (a, available_actions[node] if a != 0 else -1)
+

malpzsim/agents/searchers.py

@@ -0,0 +1,103 @@
+import logging
+import copy
+
+from collections import deque
+from typing import Any, Deque, Dict, List, Set, Type, Union
+
+import numpy as np
+
+logger = logging.getLogger(__name__)
+
+def get_new_targets(observation: dict, discovered_targets: Set[int], mask: tuple) -> List[int]:
+    attack_surface = mask[1]
+    surface_indexes = list(np.flatnonzero(attack_surface))
+    new_targets = [idx for idx in surface_indexes if idx not in discovered_targets]
+    return new_targets, surface_indexes
+
+class BreadthFirstAttacker():
+    def __init__(self, agent_config: dict) -> None:
+        self.targets: Deque[int] = deque([])
+        self.current_target: int = None
+        seed = agent_config["seed"] if agent_config.get("seed", None) else np.random.SeedSequence().entropy
+        self.rng = np.random.default_rng(seed) if agent_config.get("randomize", False) else None
+
+    def compute_action_from_dict(self, observation: Dict[str, Any], mask: tuple):
+        new_targets, surface_indexes = get_new_targets(observation,
+            self.targets,
+            mask)
+
+        # Add new targets to the back of the queue
+        # if desired, shuffle the new targets to make the attacker more unpredictable
+        if self.rng:
+            self.rng.shuffle(new_targets)
+        for c in new_targets:
+            self.targets.appendleft(c)
+
+        self.current_target, done = self.select_next_target(
+            self.current_target, self.targets, surface_indexes
+        )
+
+        self.current_target = None if done else self.current_target
+        action = 0 if done else 1
+        if action == 0:
+            logger.debug('Attacker Breadth First agent does not have '
+            'any valid targets it will terminate')
+
+        return (action, self.current_target)
+
+    @staticmethod
+    def select_next_target(
+        current_target: int, targets: Union[List[int], Deque[int]], attack_surface: Set[int]
+    ) -> int:
+        # If the current target was not compromised, put it
+        # back, but on the bottom of the stack.
+        if current_target in attack_surface:
+            targets.appendleft(current_target)
+            current_target = targets.pop()
+
+        while current_target not in attack_surface:
+            if len(targets) == 0:
+                return None, True
+
+            current_target = targets.pop()
+
+        return current_target, False
+
+class DepthFirstAttacker():
+    def __init__(self, agent_config: dict) -> None:
+        self.current_target = -1
+        self.targets: List[int] = []
+        seed = agent_config["seed"] if agent_config.get("seed", None) else np.random.SeedSequence().entropy
+        self.rng = np.random.default_rng(seed) if agent_config.get("randomize", False) else None
+
+    def compute_action_from_dict(self, observation: Dict[str, Any], mask: tuple):
+        new_targets, surface_indexes = get_new_targets(observation, self.targets, mask)
+
+        # Add new targets to the top of the stack
+        if self.rng:
+            self.rng.shuffle(new_targets)
+        for c in new_targets:
+            self.targets.append(c)
+
+        self.current_target, done = self.select_next_target(
+            self.current_target, self.targets, surface_indexes
+        )
+
+        self.current_target = None if done else self.current_target
+        action = 0 if done else 1
+        return (action, self.current_target)
+
+    @staticmethod
+    def select_next_target(
+        current_target: int, targets: Union[List[int], Deque[int]], attack_surface: Set[int]
+    ) -> int:
+        if current_target in attack_surface:
+            return current_target, False
+
+        while current_target not in attack_surface:
+            if len(targets) == 0:
+                return None, True
+
+            current_target = targets.pop()
+
+        return current_target, False