Drop custom Pulumi whoami

data_safe_haven/external/api/azure_cli.py

@@ -17,17 +17,28 @@ def login(self) -> None:
         """Force log in via the Azure CLI"""
         try:
             self.logger.debug("Attempting to login using Azure CLI.")
+            # We do not use `check` in subprocess as this raises a CalledProcessError
+            # which would break the loop. Instead we check the return code of
+            # `az account show` which will be 0 on success.
             while True:
-                process = subprocess.run(["az", "account", "show"], capture_output=True)
+                # Check whether we are already logged in
+                process = subprocess.run(
+                    ["az", "account", "show"], capture_output=True, check=False
+                )
                 if process.returncode == 0:
                     break
+                # Note that subprocess.run will block until the process terminates so
+                # we need to print the guidance first.
                 self.logger.info(
-                    "Please login in your web browser at https://login.microsoftonline.com/organizations/oauth2/v2.0/authorize."
+                    "Please login in your web browser at [bold]https://login.microsoftonline.com/organizations/oauth2/v2.0/authorize[/]."
                 )
                 self.logger.info(
-                    "If no web browser is available, please run `az login --use-device-code` in a command line window."
+                    "If no web browser is available, please run [bold]az login --use-device-code[/] in a command line window."
+                )
+                # Attempt to log in at the command line
+                process = subprocess.run(
+                    ["az", "login"], capture_output=True, check=False
                 )
-                subprocess.run(["az", "login"], capture_output=True)
-        except (FileNotFoundError, subprocess.CalledProcessError) as exc:
+        except FileNotFoundError as exc:
             msg = f"Please ensure that the Azure CLI is installed.\n{exc}"
             raise DataSafeHavenAzureError(msg) from exc

data_safe_haven/pulumi/pulumi_stack.py

@@ -197,27 +197,29 @@ def install_plugins(self) -> None:
     def login(self) -> None:
         """Login to Pulumi."""
         try:
+            # Ensure we are authenticated with the Azure CLI
+            # Without this, we cannot read the encryption key from the keyvault
+            AzureCli().login()
             # Check whether we're already logged in
+            # Note that we cannot retrieve self.stack without being logged in
             with suppress(DataSafeHavenPulumiError):
-                username = self.whoami()
-                self.logger.info(f"Logged into Pulumi as [green]{username}[/]")
-                return
+                result = self.stack.workspace.who_am_i()
+                if result.user:
+                    self.logger.info(f"Logged into Pulumi as [green]{result.user}[/]")
+                    return
             # Otherwise log in to Pulumi
             try:
-                # Ensure we are authenticated with the Azure CLI
-                # Without this, we cannot read the encryption key from the keyvault
-                AzureCli().login()
                 process = subprocess.run(
                     [
                         "pulumi",
                         "login",
-                        f"'azblob://{self.cfg.pulumi.storage_container_name}'",
+                        f"azblob://{self.cfg.pulumi.storage_container_name}",
                     ],
-                    env={**os.environ, **self.env},
-                    encoding="UTF-8",
                     capture_output=True,
                     check=True,
                     cwd=self.work_dir,
+                    encoding="UTF-8",
+                    env={**os.environ, **self.env},
                 )
                 self.logger.info(process.stdout)
             except (subprocess.CalledProcessError, FileNotFoundError) as exc:
@@ -298,28 +300,6 @@ def update(self) -> None:
             msg = f"Pulumi update failed.\n{exc}"
             raise DataSafeHavenPulumiError(msg) from exc
 
-    def whoami(self) -> str:
-        """Check current Pulumi user."""
-        try:
-            AzureCli().login()  # this is needed to read the encryption key from the keyvault
-            self.work_dir.mkdir(parents=True, exist_ok=True)
-            try:
-                process = subprocess.run(
-                    ["pulumi", "whoami"],
-                    capture_output=True,
-                    check=True,
-                    cwd=self.work_dir,
-                    encoding="UTF-8",
-                    env={**os.environ, **self.env},
-                )
-                return process.stdout.strip()
-            except (subprocess.CalledProcessError, FileNotFoundError) as exc:
-                msg = f"No Pulumi user found.\n{exc}."
-                raise DataSafeHavenPulumiError(msg) from exc
-        except Exception as exc:
-            msg = f"Pulumi user check failed.\n{exc}."
-            raise DataSafeHavenPulumiError(msg) from exc
-
 
 class PulumiSHMStack(PulumiStack):
     """Interact with an SHM using Pulumi"""