scenarios: add --admin flag to run command for ClusterRole access

Author: pinheadmz

resources/charts/commander/templates/rbac.yaml

@@ -33,3 +33,33 @@ subjects:
   - kind: ServiceAccount
     name: {{ include "commander.fullname" . }}
     namespace: {{ .Release.Namespace }}
+{{- if .Values.admin }}
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  name: {{ include "commander.fullname" . }}
+  namespace: {{ .Release.Namespace }}
+  labels:
+    app.kubernetes.io/name: {{ .Chart.Name }}
+rules:
+  - apiGroups: [""]
+    resources: ["pods", "namespaces", "configmaps"]
+    verbs: ["get", "list", "watch"]
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRoleBinding
+metadata:
+  name: {{ include "commander.fullname" . }}
+  namespace: {{ .Release.Namespace }}
+  labels:
+    app.kubernetes.io/name: {{ .Chart.Name }}
+roleRef:
+  kind: ClusterRole
+  name: {{ include "commander.fullname" . }}
+  apiGroup: rbac.authorization.k8s.io
+subjects:
+  - kind: ServiceAccount
+    name: {{ include "commander.fullname" . }}
+    namespace: {{ .Release.Namespace }}
+{{- end}}

resources/charts/commander/values.yaml

@@ -66,3 +66,5 @@ volumeMounts: []
 port:
 
 args: ""
+
+admin: false

resources/scenarios/commander.py

@@ -29,11 +29,19 @@
 with open("/var/run/secrets/kubernetes.io/serviceaccount/namespace") as f:
     NAMESPACE = f.read().strip()
 
-# Use the in-cluster k8s client to determine what pods we have access to
+# Get the in-cluster k8s client to determine what we have access to
 config.load_incluster_config()
 sclient = client.CoreV1Api()
-pods = sclient.list_namespaced_pod(namespace=NAMESPACE)
-cmaps = sclient.list_namespaced_config_map(namespace=NAMESPACE)
+
+try:
+    # An admin with cluster access can list everything.
+    # A wargames player with namespaced access will get a FORBIDDEN error here
+    pods = sclient.list_pod_for_all_namespaces()
+    cmaps = sclient.list_config_map_for_all_namespaces()
+except Exception:
+    # Just get whatever we have access to in this namespace only
+    pods = sclient.list_namespaced_pod(namespace=NAMESPACE)
+    cmaps = sclient.list_namespaced_config_map(namespace=NAMESPACE)
 
 WARNET = {"tanks": [], "lightning": [], "channels": []}
 for pod in pods.items:

src/warnet/control.py

@@ -240,26 +240,29 @@ def get_active_network(namespace):
     "--source_dir", type=click.Path(exists=True, file_okay=False, dir_okay=True), required=False
 )
 @click.argument("additional_args", nargs=-1, type=click.UNPROCESSED)
+@click.option("--admin", is_flag=True, default=False, show_default=False)
 @click.option("--namespace", default=None, show_default=True)
 def run(
     scenario_file: str,
     debug: bool,
     source_dir,
     additional_args: tuple[str],
+    admin: bool,
     namespace: Optional[str],
 ):
     """
     Run a scenario from a file.
     Pass `-- --help` to get individual scenario help
     """
-    return _run(scenario_file, debug, source_dir, additional_args, namespace)
+    return _run(scenario_file, debug, source_dir, additional_args, admin, namespace)
 
 
 def _run(
     scenario_file: str,
     debug: bool,
     source_dir,
     additional_args: tuple[str],
+    admin: bool,
     namespace: Optional[str],
 ) -> str:
     namespace = get_default_namespace_or(namespace)
@@ -329,6 +332,8 @@ def filter(path):
         ]
 
         # Add additional arguments
+        if admin:
+            helm_command.extend(["--set", "admin=true"])
         if additional_args:
             helm_command.extend(["--set", f"args={' '.join(additional_args)}"])
 

src/warnet/deploy.py

@@ -385,6 +385,7 @@ def deploy_network(directory: Path, debug: bool = False, namespace: Optional[str
             debug=False,
             source_dir=SCENARIOS_DIR,
             additional_args=None,
+            admin=False,
             namespace=namespace,
         )
         wait_for_pod(name, namespace=namespace)