[iris] Drop implicit service_account fallback from ssh_impersonate_service_account

Remove the service_account parameter from ssh_impersonate_service_account()
so it only returns a value when impersonate_service_account is explicitly set
in SshConfig. In metadata mode the controller's ambient VM identity is used
for gcloud SSH — no impersonation needed. OS Login setups that need
impersonation must set ssh.impersonate_service_account explicitly.

Author: rjpower

lib/iris/src/iris/cluster/providers/gcp/controller.py

@@ -117,7 +117,6 @@ def tunnel(
             project=self.worker_provider.project_id,
             label_prefix=self.worker_provider.label_prefix,
             ssh_config=self.worker_provider.ssh_config,
-            service_account=self.controller_service_account,
             local_port=local_port,
         )
 
@@ -228,7 +227,6 @@ def _gcp_tunnel(
     project: str,
     label_prefix: str,
     ssh_config: config_pb2.SshConfig | None,
-    service_account: str | None = None,
     local_port: int | None = None,
     timeout: float = 60.0,
 ) -> Iterator[str]:
@@ -238,7 +236,7 @@ def _gcp_tunnel(
     that may be listening on the same port on a different address family (IPv6).
     Picks a free port automatically if none is specified.
     """
-    effective_service_account = ssh_impersonate_service_account(ssh_config, service_account)
+    effective_service_account = ssh_impersonate_service_account(ssh_config)
     key_file = ssh_key_file(ssh_config, effective_service_account)
     _check_gcloud_ssh_key(key_file)
 

lib/iris/src/iris/cluster/providers/gcp/handles.py

@@ -65,15 +65,12 @@
 
 def _os_login_user(
     ssh_config: config_pb2.SshConfig | None,
-    service_account: str | None = None,
 ) -> str:
     if ssh_config and ssh_config.os_login_user:
         return ssh_config.os_login_user
     if ssh_config and ssh_config.user and ssh_config.user != "root":
         return ssh_config.user
-    return resolve_current_os_login_user(
-        impersonate_service_account=ssh_impersonate_service_account(ssh_config, service_account)
-    )
+    return resolve_current_os_login_user(impersonate_service_account=ssh_impersonate_service_account(ssh_config))
 
 
 def _vm_slice_metadata_user(ssh_config: config_pb2.SshConfig | None) -> str:
@@ -319,10 +316,10 @@ def _describe_cloud(self) -> SliceStatus:
                 direct_host = external_ip or internal_ip
                 remote_exec = DirectSshRemoteExec(
                     host=direct_host,
-                    user=_os_login_user(self._ssh_config, self._service_account),
+                    user=_os_login_user(self._ssh_config),
                     key_file=ssh_key_file(
                         self._ssh_config,
-                        ssh_impersonate_service_account(self._ssh_config, self._service_account),
+                        ssh_impersonate_service_account(self._ssh_config),
                     ),
                     connect_timeout=(
                         duration_from_proto(self._ssh_config.connect_timeout)
@@ -339,12 +336,9 @@ def _describe_cloud(self) -> SliceStatus:
                     ssh_user=_vm_slice_metadata_user(self._ssh_config),
                     ssh_key_file=ssh_key_file(
                         self._ssh_config,
-                        ssh_impersonate_service_account(self._ssh_config, self._service_account),
-                    ),
-                    impersonate_service_account=ssh_impersonate_service_account(
-                        self._ssh_config,
-                        self._service_account,
+                        ssh_impersonate_service_account(self._ssh_config),
                     ),
+                    impersonate_service_account=ssh_impersonate_service_account(self._ssh_config),
                     _address=internal_ip,
                 )
             workers.append(
@@ -449,12 +443,9 @@ def _describe_cloud(self) -> SliceStatus:
             ssh_user=None if uses_os_login(self._ssh_config) else _vm_slice_metadata_user(self._ssh_config),
             ssh_key_file=ssh_key_file(
                 self._ssh_config,
-                ssh_impersonate_service_account(self._ssh_config, self._service_account),
-            ),
-            impersonate_service_account=ssh_impersonate_service_account(
-                self._ssh_config,
-                self._service_account,
+                ssh_impersonate_service_account(self._ssh_config),
             ),
+            impersonate_service_account=ssh_impersonate_service_account(self._ssh_config),
         )
         worker = GcpStandaloneWorkerHandle(
             _vm_id=f"{self._slice_id}-worker-0",

lib/iris/src/iris/cluster/providers/gcp/ssh.py

@@ -98,10 +98,8 @@ def ssh_key_file(
 
 def ssh_impersonate_service_account(
     ssh_config: config_pb2.SshConfig | None,
-    service_account: str | None = None,
 ) -> str | None:
+    """Return the explicit impersonation SA from ssh_config, or None."""
     if ssh_config and ssh_config.impersonate_service_account:
         return ssh_config.impersonate_service_account
-    if service_account:
-        return service_account
     return None

lib/iris/src/iris/cluster/providers/gcp/workers.py

@@ -236,7 +236,7 @@ def create_vm(self, config: config_pb2.VmConfig) -> GcpStandaloneWorkerHandle:
             zone=zone,
             vm_name=config.name,
             ssh_key_file=ssh_key_file(self._ssh_config),
-            impersonate_service_account=ssh_impersonate_service_account(self._ssh_config, request.service_account),
+            impersonate_service_account=ssh_impersonate_service_account(self._ssh_config),
         )
 
         return GcpStandaloneWorkerHandle(
@@ -536,7 +536,7 @@ def list_vms(
                 zone=vm.zone,
                 vm_name=vm.name,
                 ssh_key_file=ssh_key_file(self._ssh_config),
-                impersonate_service_account=ssh_impersonate_service_account(self._ssh_config, vm.service_account),
+                impersonate_service_account=ssh_impersonate_service_account(self._ssh_config),
             )
             handles.append(
                 GcpStandaloneWorkerHandle(

lib/iris/tests/cluster/providers/gcp/test_platform.py

@@ -475,7 +475,7 @@ def test_gcp_vm_slice_os_login_uses_service_account_impersonation():
 
     handle = platform.create_slice(cfg)
     status = handle.describe()
-    assert status.workers[0]._remote_exec.impersonate_service_account == cfg.gcp.service_account
+    assert status.workers[0]._remote_exec.impersonate_service_account is None
 
 
 def test_gcp_vm_slice_os_login_prefers_explicit_ssh_impersonation_account():
@@ -857,7 +857,7 @@ def test_gcp_tpu_slice_os_login_resolves_user_from_service_account():
         handle = platform.create_slice(cfg)
         status = handle.describe()
 
-    resolve_user.assert_called_with(impersonate_service_account=cfg.gcp.service_account)
+    resolve_user.assert_called_with(impersonate_service_account=None)
     assert status.workers[0]._remote_exec.user == "svc-user"