test: cover scenarios run --admin

Author: pinheadmz

.github/workflows/test.yml

@@ -52,6 +52,7 @@ jobs:
           - simln_test.py
           - scenarios_test.py
           - namespace_admin_test.py
+          - wargames_test.py
     steps:
       - uses: actions/checkout@v4
       - uses: azure/[email protected]

resources/scenarios/test_scenarios/generate_one_allnodes.py

@@ -0,0 +1,33 @@
+#!/usr/bin/env python3
+
+# The base class exists inside the commander container
+try:
+    from commander import Commander
+except Exception:
+    from resources.scenarios.commander import Commander
+
+
+class GenOneAllNodes(Commander):
+    def set_test_params(self):
+        self.num_nodes = 1
+
+    def add_options(self, parser):
+        parser.description = (
+            "Attempt to generate one block on every node the scenario has access to"
+        )
+        parser.usage = "warnet run /path/to/generate_one_allnodes.py"
+
+    def run_test(self):
+        for node in self.nodes:
+            wallet = self.ensure_miner(node)
+            addr = wallet.getnewaddress("bech32")
+            self.log.info(f"node: {node.tank}")
+            self.log.info(self.generatetoaddress(node, 1, addr))
+
+
+def main():
+    GenOneAllNodes().main()
+
+
+if __name__ == "__main__":
+    main()

src/warnet/control.py

@@ -352,6 +352,7 @@ def filter(path):
     except subprocess.CalledProcessError as e:
         print(f"Failed to deploy scenario commander: {scenario_name}")
         print(f"Error: {e.stderr}")
+        return None
 
     # upload scenario files and network data to the init container
     wait_for_init(name, namespace=namespace)

test/data/wargames/namespaces/armies/namespace-defaults.yaml

@@ -0,0 +1,5 @@
+users:
+  - name: warnet-user
+    roles:
+      - pod-viewer
+      - pod-manager

test/data/wargames/namespaces/armies/namespaces.yaml

@@ -0,0 +1,2 @@
+namespaces:
+- name: wargames-red

test/data/wargames/networks/armada/network.yaml

@@ -0,0 +1,6 @@
+nodes:
+- name: armada
+  image:
+    tag: '27.0'
+  addnode:
+    - miner.default

test/data/wargames/networks/armada/node-defaults.yaml

@@ -0,0 +1,6 @@
+global:
+  chain: regtest
+image:
+  repository: bitcoindevproject/bitcoin
+  pullPolicy: IfNotPresent
+  tag: '27.0'

test/data/wargames/networks/battlefield/network.yaml

@@ -0,0 +1,10 @@
+nodes:
+- name: miner
+  image:
+    tag: '27.0'
+- name: target-red
+  addnode:
+    - miner
+  image:
+    tag: '27.0'
+    

test/data/wargames/networks/battlefield/node-defaults.yaml

@@ -0,0 +1,6 @@
+global:
+  chain: regtest
+image:
+  repository: bitcoindevproject/bitcoin
+  pullPolicy: IfNotPresent
+  tag: '27.0'

test/wargames_test.py

@@ -0,0 +1,93 @@
+#!/usr/bin/env python3
+
+import os
+from pathlib import Path
+
+import pexpect
+from test_base import TestBase
+
+from warnet.k8s import get_kubeconfig_value
+from warnet.process import stream_command
+
+
+class WargamesTest(TestBase):
+    def __init__(self):
+        super().__init__()
+        self.wargame_dir = Path(os.path.dirname(__file__)) / "data" / "wargames"
+        self.scen_src_dir = Path(os.path.dirname(__file__)).parent / "resources" / "scenarios"
+        self.scen_test_dir = (
+            Path(os.path.dirname(__file__)).parent / "resources" / "scenarios" / "test_scenarios"
+        )
+        self.initial_context = get_kubeconfig_value("{.current-context}")
+
+    def run_test(self):
+        try:
+            self.setup_battlefield()
+            self.setup_armies()
+            self.check_scenario_permissions()
+        finally:
+            self.log.info("Restoring initial_context")
+            stream_command(f"kubectl config use-context {self.initial_context}")
+            self.cleanup()
+
+    def setup_battlefield(self):
+        self.log.info("Setting up battlefield")
+        self.log.info(self.warnet(f"deploy {self.wargame_dir / 'networks' / 'battlefield'}"))
+        self.wait_for_all_tanks_status(target="running")
+        self.wait_for_all_edges()
+
+    def setup_armies(self):
+        self.log.info("Deploying namespaces and armadas")
+        self.log.info(self.warnet(f"deploy {self.wargame_dir / 'namespaces' / 'armies'}"))
+        self.log.info(
+            self.warnet(f"deploy {self.wargame_dir / 'networks' / 'armada'} --to-all-users")
+        )
+        self.wait_for_all_tanks_status(target="running")
+        self.wait_for_all_edges()
+
+    def check_scenario_permissions(self):
+        self.log.info("Admin without --admin can not command a node outside of default namespace")
+        stream_command(
+            f"warnet run {self.scen_test_dir / 'generate_one_allnodes.py'} --source_dir={self.scen_src_dir} --debug"
+        )
+        # Only miner.default and target-red.default were accesible
+        assert self.warnet("bitcoin rpc miner getblockcount") == "2"
+
+        self.log.info("Admin with --admin can command all nodes in any namespace")
+        stream_command(
+            f"warnet run {self.scen_test_dir / 'generate_one_allnodes.py'} --source_dir={self.scen_src_dir} --admin --debug"
+        )
+        # armada.wargames-red, miner.default and target-red.default were accesible
+        assert self.warnet("bitcoin rpc miner getblockcount") == "5"
+
+        self.log.info("Switch to wargames player context")
+        self.log.info(self.warnet("admin create-kubeconfigs"))
+        clicker = pexpect.spawn("warnet auth kubeconfigs/warnet-user-wargames-red-kubeconfig")
+        while clicker.expect(["Overwrite", "Updated kubeconfig"]) == 0:
+            print(clicker.before, clicker.after)
+            clicker.sendline("y")
+        print(clicker.before, clicker.after)
+
+        self.log.info("Player without --admin can only command the node inside their own namespace")
+        stream_command(
+            f"warnet run {self.scen_test_dir / 'generate_one_allnodes.py'} --source_dir={self.scen_src_dir} --debug"
+        )
+        # Only armada.wargames-red was (and is) accesible
+        assert self.warnet("bitcoin rpc armada getblockcount") == "6"
+
+        self.log.info("Player attempting to use --admin is gonna have a bad time")
+        stream_command(
+            f"warnet run {self.scen_test_dir / 'generate_one_allnodes.py'} --source_dir={self.scen_src_dir} --admin --debug"
+        )
+        # Nothing was accesible
+        assert self.warnet("bitcoin rpc armada getblockcount") == "6"
+
+        self.log.info("Restore admin context")
+        stream_command(f"kubectl config use-context {self.initial_context}")
+        # Sanity check
+        assert self.warnet("bitcoin rpc miner getblockcount") == "6"
+
+
+if __name__ == "__main__":
+    test = WargamesTest()
+    test.run_test()