Source code for runhouse.rns.login

import logging
import os
from typing import Dict, Optional

import requests

import typer

from runhouse.globals import configs, rns_client

logger = logging.getLogger(__name__)


def is_interactive():
    import __main__ as main

    return not hasattr(main, "__file__")


[docs]def login( token: str = None, download_config: bool = None, upload_config: bool = None, download_secrets: bool = None, upload_secrets: bool = None, ret_token: bool = False, interactive: bool = None, from_cli: bool = False, sync_secrets: bool = False, ): """Login to Runhouse. Validates token provided, with options to upload or download stored secrets or config between local environment and Runhouse / Vault. Args: token (str): Runhouse token, can be found at https://www.run.house/account#token. If not provided, function will interactively prompt for the token to be entered manually. download_config (bool): Whether to download configs from your Runhouse account to local environment. upload_config (bool): Whether to upload local configs into your Runhouse account. download_secrets (bool): Whether to download secrets from your Runhouse account to local environment. upload_secrets (bool): Whether to upload local secrets to your Runhouse account. ret_token (bool): Whether to return your Runhouse token. (Default: False) interactive (bool): Whether to interactively go through the login flow. ``token`` must be provided to set this to False. Returns: Token if ``ret_token`` is set to True, otherwise nothing. """ all_options_set = token and not any( arg is None for arg in (download_config, upload_config, download_secrets, upload_secrets) ) if interactive is False and not token: raise Exception( "`interactive` can only be set to `False` if token is provided." ) if interactive or (interactive is None and not all_options_set): from getpass import getpass from rich.console import Console console = Console() console.print( """ ____ __ @ @ @ / __ \__ ______ / /_ ____ __ __________ []___ / /_/ / / / / __ \/ __ \/ __ \/ / / / ___/ _ \ / /\____ @@ / _, _/ /_/ / / / / / / / /_/ / /_/ (__ ) __/ /_/\_//____/\ @@@@ /_/ |_|\__,_/_/ /_/_/ /_/\____/\__,_/____/\___/ | || |||__||| || """ ) account_url = f"{configs.get('dashboard_url')}/account#token" link = ( f"[link={account_url}]{account_url}[/link]" if is_interactive() else account_url ) if not token: console.print( f"Retrieve your token :key: here to use :person_running: :house: Runhouse for " f"secrets and artifact management: {link}", style="bold yellow", ) token = getpass("Token: ") rh_config_exists = configs.CONFIG_PATH.exists() if not rh_config_exists: upload_config = False download_secrets = False # download the config automatically if no config.yaml exists download_config = ( download_config if download_config is not None or not rh_config_exists else typer.confirm( "Download your Runhouse config to your local .rh folder?", default=True, ) ) upload_config = ( upload_config if upload_config is not None else typer.confirm("Upload your local .rh config to Runhouse?") ) if sync_secrets: from runhouse import Secret if Secret.vault_secrets(rns_client.request_headers()): download_secrets = ( download_secrets if download_secrets is not None else typer.confirm( "Download secrets from Vault to your local Runhouse environment?" ) ) upload_secrets = ( upload_secrets if upload_secrets is not None else typer.confirm( "Upload your local enabled provider secrets to Vault?" ) ) if token: # Note, this is to explicitly add it to the config file, as opposed to setting in python # via configs.token = token configs.set("token", token) if download_config: configs.download_and_save_defaults() if upload_config: configs.load_defaults_from_file() configs.upload_defaults(defaults=configs.defaults_cache) if not (download_config or upload_config): # If we are not downloading or uploading config, we still want to make sure the token is valid # and also download the username and default folder try: defaults = configs.download_defaults() except: logger.error("Failed to validate token") return None configs.set("username", defaults["username"]) configs.set("default_folder", defaults["default_folder"]) if download_secrets: _login_download_secrets(from_cli=from_cli) if upload_secrets: _login_upload_secrets(interactive=interactive) logger.info("Successfully logged into Runhouse.") if ret_token: return token
def _login_download_secrets(headers: Optional[str] = None, from_cli=False): from runhouse import Secret secrets = Secret.vault_secrets(headers=headers or rns_client.request_headers()) env_secrets = {} for name in secrets: try: secret = Secret.from_name(name) if not (hasattr(secret, "path") or hasattr(secret, "env_vars")): continue download_path = ( secret.path if secret.path else f"{secret._DEFAULT_CREDENTIALS_PATH}/{name}" if hasattr(secret, "provider") and secret.provider == "ssh" else secret._DEFAULT_CREDENTIALS_PATH ) if download_path and not secret.env_vars: logger.info(f"Loading down secrets for {name} into {download_path}") secret.write(path=download_path) else: env_vars = secret.env_vars or secret._DEFAULT_ENV_VARS logger.info( f"Writing down env secrets for {name} into {env_vars.values()}" ) if not from_cli: secret.write(env=True) else: for key, val in secret.values.items(): if key in env_vars: env_secrets.update({env_vars[key]: val}) except ValueError as e: logger.warning( f"Encountered {e}. Was not able to load down secrets for {name}." ) if from_cli and env_secrets: folder = os.path.expanduser("~/.rh/secrets") os.makedirs(folder, exist_ok=True) with open(f"{folder}/login.env", "w") as f: for key, val in env_secrets.items(): f.write(f"{key}={val}\n") logger.info( "Env var secrets written down into ~/.rh/secrets/login.env. " "Please run `source ~/.rh/secrets/login.env` to set the environment variables." ) def _login_upload_secrets(interactive: bool, headers: Optional[Dict] = None): from runhouse import Secret local_secrets = Secret.local_secrets() provider_secrets = Secret.extract_provider_secrets() local_secrets.update(provider_secrets) names = list(local_secrets.keys()) for name in names: resource_uri = rns_client.resource_uri(name) resp = requests.get( f"{rns_client.api_server_url}/resource/{resource_uri}", headers=headers or rns_client.request_headers(), ) if resp.status_code == 200: local_secrets.pop(name, None) continue if interactive is not False: upload_secrets = typer.confirm(f"Upload secrets for {name}?") if not upload_secrets: local_secrets.pop(name, None) if local_secrets: logger.info(f"Uploading secrets for {list(local_secrets)} to Vault.") for _, secret in local_secrets.items(): secret.save(save_values=True)
[docs]def logout( delete_loaded_secrets: bool = None, delete_rh_config_file: bool = None, interactive: bool = None, ): """ Logout from Runhouse. Provides option to delete credentials from the Runhouse config and the underlying credentials file. Token is also deleted from the config. Args: delete_loaded_secrets (bool, optional): If True, deletes the provider credentials file. Defaults to None. delete_rh_config_file (bool, optional): If True, deletes the rh config file. Defaults to None. interactive (bool, optional): If True, runs the logout process in interactive mode. Defaults to None. Returns: None """ from runhouse.resources.secrets import Secret from runhouse.resources.secrets.provider_secrets.ssh_secret import SSHSecret interactive_session: bool = ( interactive if interactive is not None else is_interactive() ) config_secrets = list(configs.get("secrets", {}).items()) for (name, value) in config_secrets: try: secret = Secret.from_name(name) if isinstance(secret, SSHSecret): logger.info( "Automatic deletion for local SSH credentials file is not supported. " "Please manually delete it if you would like to remove it" ) configs.delete_provider(name) continue except ValueError: pass if interactive_session: delete_loaded_secrets = typer.confirm( f"Delete credentials in {value} for {name}?" ) if delete_loaded_secrets: if isinstance(value, str): path = os.path.expanduser(value) if os.path.exists(path): os.remove(path) else: # list of env variables set for key in value: del os.environ[key] configs.delete_provider(name) local_secrets = Secret.local_secrets() for _, secret in local_secrets.items(): secret.delete() # Delete token and username/default folder from rh config file configs.delete(key="token") configs.delete(key="username") configs.delete(key="default_folder") # Delete values in configs object configs.token = None configs.username = None configs.default_folder = None # Wipe env vars os.environ.pop("RH_TOKEN", None) os.environ.pop("RH_USERNAME", None) rh_config_path = configs.CONFIG_PATH if not delete_rh_config_file and interactive_session: delete_rh_config_file = typer.confirm("Delete your local Runhouse config file?") if delete_rh_config_file: # Delete the credentials file on the file system configs.delete_defaults(rh_config_path) logger.info(f"Deleted Runhouse config file from path: {rh_config_path}") logger.info("Successfully logged out of Runhouse.")