Update environment variables for a VM.
(
self, vm_id: str, envs: Dict[str, str], kms_urls: Optional[List[str]] = None
)
| 907 | return response.get("id") |
| 908 | |
| 909 | def update_vm_env( |
| 910 | self, vm_id: str, envs: Dict[str, str], kms_urls: Optional[List[str]] = None |
| 911 | ) -> None: |
| 912 | """Update environment variables for a VM.""" |
| 913 | # Validate: requires --kms-url |
| 914 | if not kms_urls: |
| 915 | raise Exception("--kms-url is required to encrypt environment variables") |
| 916 | |
| 917 | envs = envs or {} |
| 918 | # First get the VM info to retrieve the app_id |
| 919 | vm_info_response = self.rpc_call("GetInfo", {"id": vm_id}) |
| 920 | |
| 921 | if not vm_info_response.get("found", False) or "info" not in vm_info_response: |
| 922 | raise Exception(f"VM with ID {vm_id} not found") |
| 923 | |
| 924 | app_id = vm_info_response["info"]["app_id"] |
| 925 | print(f"Retrieved app ID: {app_id}") |
| 926 | vm_configuration = vm_info_response["info"].get("configuration") or {} |
| 927 | compose_file = vm_configuration.get("compose_file") |
| 928 | |
| 929 | # Now get the encryption key for the app |
| 930 | encrypt_pubkey = self.get_app_env_encrypt_pub_key( |
| 931 | app_id, kms_urls[0] if kms_urls else None |
| 932 | ) |
| 933 | print(f"Encrypting environment variables with key: {encrypt_pubkey}") |
| 934 | envs_list = [{"key": k, "value": v} for k, v in envs.items()] |
| 935 | encrypted_env = encrypt_env(envs_list, encrypt_pubkey) |
| 936 | |
| 937 | # Use UpdateApp with the VM ID |
| 938 | payload = {"id": vm_id, "encrypted_env": encrypted_env} |
| 939 | |
| 940 | if compose_file: |
| 941 | try: |
| 942 | app_compose = json.loads(compose_file) |
| 943 | except json.JSONDecodeError: |
| 944 | app_compose = {} |
| 945 | compose_changed = False |
| 946 | allowed_envs = list(envs.keys()) |
| 947 | if app_compose.get("allowed_envs") != allowed_envs: |
| 948 | app_compose["allowed_envs"] = allowed_envs |
| 949 | compose_changed = True |
| 950 | launch_token_value = envs.get("APP_LAUNCH_TOKEN") |
| 951 | if launch_token_value is not None: |
| 952 | launch_token_hash = hashlib.sha256( |
| 953 | launch_token_value.encode("utf-8") |
| 954 | ).hexdigest() |
| 955 | if app_compose.get("launch_token_hash") != launch_token_hash: |
| 956 | app_compose["launch_token_hash"] = launch_token_hash |
| 957 | compose_changed = True |
| 958 | if compose_changed: |
| 959 | payload["compose_file"] = json.dumps( |
| 960 | app_compose, indent=4, ensure_ascii=False |
| 961 | ) |
| 962 | |
| 963 | self.rpc_call("UpgradeApp", payload) |
| 964 | print(f"Environment variables updated for VM {vm_id}") |
| 965 | |
| 966 | def update_vm_user_config(self, vm_id: str, user_config: str) -> None: |
no test coverage detected