Fix ssh secrets for den launcher with org

runhouse/resources/hardware/cluster.py

@@ -97,6 +97,7 @@ def __init__(
         ssh_properties: Dict = None,
         den_auth: bool = False,
         dryrun: bool = False,
+        skip_creds: bool = False,
         **kwargs,  # We have this here to ignore extra arguments when calling from from_config
     ):
         """
@@ -131,7 +132,10 @@ def __init__(
         if self._default_env and not self._default_env.name:
             self._default_env.name = _unnamed_default_env_name(self.name)
 
-        self._setup_creds(creds)
+        if skip_creds and not creds:
+            self._creds = None
+        else:
+            self._setup_creds(creds)
 
     @property
     def head_ip(self):
@@ -341,8 +345,13 @@ def _should_save_creds(self, folder: str = None) -> bool:
     def _save_sub_resources(self, folder: str = None):
         from runhouse.resources.envs import Env
 
-        if self._should_save_creds(folder):
-            self._creds.save(folder=folder)
+        creds_folder = (
+            folder if (not self._creds or not self._creds._rns_folder) else None
+        )
+        if self._should_save_creds(creds_folder):
+            # Only automatically set the creds folder if it doesn't have one yet
+            # allows for org SSH keys to be associated with the user.
+            self._creds.save(folder=creds_folder)
 
         if self._default_env and isinstance(self._default_env, Env):
             if not self._default_env.name:

runhouse/resources/hardware/launcher_utils.py

@@ -192,7 +192,8 @@ def _update_from_den_response(cls, cluster, config: dict):
     def up(cls, cluster, verbose: bool = True, force: bool = False):
         """Launch the cluster via Den."""
         sky_secret = cls.sky_secret()
-        cluster._creds = sky_secret
+        cluster._setup_creds(sky_secret)
+        cluster.save()
 
         payload = {
             "cluster_config": {

runhouse/resources/hardware/on_demand_cluster.py

@@ -75,6 +75,9 @@ def __init__(
         .. note::
             To build a cluster, please use the factory method :func:`cluster`.
         """
+        cluster_launcher_type = launcher_type or configs.launcher_type
+        skip_creds = cluster_launcher_type == LauncherType.DEN
+
         super().__init__(
             name=name,
             default_env=default_env,
@@ -86,6 +89,7 @@ def __init__(
             domain=domain,
             den_auth=den_auth,
             dryrun=dryrun,
+            skip_creds=skip_creds,
             **kwargs,
         )
 
@@ -109,7 +113,7 @@ def __init__(
         self.memory = memory
         self.disk_size = disk_size
         self.sky_kwargs = sky_kwargs or {}
-        self.launcher_type = launcher_type or configs.launcher_type
+        self.launcher_type = cluster_launcher_type
 
         self.stable_internal_external_ips = kwargs.get(
             "stable_internal_external_ips", None