Generate kubeconfig for human players only

resources/charts/namespaces/templates/serviceaccount.yaml

@@ -8,4 +8,6 @@ metadata:
   annotations:
     helm.sh/hook: post-install,post-upgrade
     helm.sh/hook-weight: "-5"
+  labels:
+    mission: user
 {{- end }}

src/warnet/admin.py

@@ -11,12 +11,12 @@
     K8sError,
     get_cluster_of_current_context,
     get_namespaces_by_type,
-    get_service_accounts_in_namespace,
+    get_token_for_service_acount,
+    get_warnet_user_service_accounts_in_namespace,
     open_kubeconfig,
 )
 from .namespaces import copy_namespaces_defaults, namespaces
 from .network import copy_network_defaults
-from .process import run_command
 
 
 @click.group(name="admin", hidden=True)
@@ -84,21 +84,21 @@ def create_kubeconfigs(kubeconfig_dir, token_duration):
     for v1namespace in warnet_namespaces:
         namespace = v1namespace.metadata.name
         click.echo(f"Processing namespace: {namespace}")
-        service_accounts = get_service_accounts_in_namespace(namespace)
+        service_accounts = get_warnet_user_service_accounts_in_namespace(namespace)
 
         for sa in service_accounts:
+            name = sa.metadata.name
             # Create a token for the ServiceAccount with specified duration
-            command = f"kubectl create token {sa} -n {namespace} --duration={token_duration}s"
             try:
-                token = run_command(command)
+                token = get_token_for_service_acount(sa, token_duration)
             except Exception as e:
                 click.echo(
-                    f"Failed to create token for ServiceAccount {sa} in namespace {namespace}. Error: {str(e)}. Skipping..."
+                    f"Failed to create token for ServiceAccount {name} in namespace {namespace}. Error: {str(e)}. Skipping..."
                 )
                 continue
 
             # Create a kubeconfig file for the user
-            kubeconfig_file = os.path.join(kubeconfig_dir, f"{sa}-{namespace}-kubeconfig")
+            kubeconfig_file = os.path.join(kubeconfig_dir, f"{name}-{namespace}-kubeconfig")
 
             # TODO: move yaml  out of python code to resources/manifests/
             #
@@ -109,21 +109,25 @@ def create_kubeconfigs(kubeconfig_dir, token_duration):
                 "apiVersion": "v1",
                 "kind": "Config",
                 "clusters": [cluster],
-                "users": [{"name": sa, "user": {"token": token}}],
+                "users": [{"name": name, "user": {"token": token}}],
                 "contexts": [
                     {
-                        "name": f"{sa}-{namespace}",
-                        "context": {"cluster": cluster["name"], "namespace": namespace, "user": sa},
+                        "name": f"{name}-{namespace}",
+                        "context": {
+                            "cluster": cluster["name"],
+                            "namespace": namespace,
+                            "user": name,
+                        },
                     }
                 ],
-                "current-context": f"{sa}-{namespace}",
+                "current-context": f"{name}-{namespace}",
             }
 
             # Write to a YAML file
             with open(kubeconfig_file, "w") as f:
                 yaml.dump(kubeconfig_dict, f, default_flow_style=False)
 
-            click.echo(f"    Created kubeconfig file for {sa}: {kubeconfig_file}")
+            click.echo(f"    Created kubeconfig file for {name}: {kubeconfig_file}")
 
     click.echo("---")
     click.echo(

src/warnet/k8s.py

@@ -4,13 +4,13 @@
 import tarfile
 import tempfile
 from pathlib import Path
-from time import time, sleep
+from time import sleep, time
 from typing import Optional
 
 import yaml
 from kubernetes import client, config, watch
 from kubernetes.client import CoreV1Api
-from kubernetes.client.models import V1Namespace, V1Pod, V1PodList
+from kubernetes.client.models import V1Namespace, V1Pod, V1PodList, V1TokenRequestSpec
 from kubernetes.client.rest import ApiException
 from kubernetes.dynamic import DynamicClient
 from kubernetes.stream import stream
@@ -516,14 +516,32 @@ def get_namespaces_by_type(namespace_type: str) -> list[V1Namespace]:
     return [ns for ns in namespaces if ns.metadata.name.startswith(namespace_type)]
 
 
-def get_service_accounts_in_namespace(namespace):
+def get_warnet_user_service_accounts_in_namespace(namespace):
     """
-    Get all service accounts in a namespace. Returns an empty list if no service accounts are found in the specified namespace.
+    Get all service accounts in a namespace that were created for human users
+    (not scenario commanders or other pods)
+    Returns an empty list if no applicable service accounts are found in the specified namespace.
     """
-    command = f"kubectl get serviceaccounts -n {namespace} -o jsonpath={{.items[*].metadata.name}}"
-    # skip the default service account created by k8s
-    service_accounts = run_command(command).split()
-    return [sa for sa in service_accounts if sa != "default"]
+    sclient = get_static_client()
+    sas = sclient.list_namespaced_service_account(namespace)
+    return [
+        sa
+        for sa in sas.items
+        if sa.metadata.labels
+        and "mission" in sa.metadata.labels
+        and sa.metadata.labels["mission"] == "user"
+    ]
+
+
+def get_token_for_service_acount(sa, duration):
+    sclient = get_static_client()
+    spec = V1TokenRequestSpec(
+        audiences=["https://kubernetes.default.svc"], expiration_seconds=duration
+    )
+    resp = sclient.create_namespaced_service_account_token(
+        name=sa.metadata.name, namespace=sa.metadata.namespace, body=spec
+    )
+    return resp.status.token
 
 
 def can_delete_pods(namespace: Optional[str] = None) -> bool:

test/wargames_test.py

@@ -87,6 +87,17 @@ def check_scenario_permissions(self):
         # Sanity check
         assert self.warnet("bitcoin rpc miner getblockcount") == "6"
 
+        self.log.info("Re-generate kubeconfigs after a scenario has been run")
+        self.log.info(self.warnet("admin create-kubeconfigs"))
+        kubeconfig_dir = Path("kubeconfigs/")
+        count = 0
+        for p in kubeconfig_dir.iterdir():
+            if p.is_file():
+                print(p)
+                assert "warnet-user" in str(p)
+                count += 1
+                assert count <= 1
+
 
 if __name__ == "__main__":
     test = WargamesTest()