introduce pyotp and support generating totp within skyvern

pyproject.toml

@@ -50,6 +50,7 @@ temporalio = "^1.6.0"
 requests-toolbelt = "^1.0.0"
 posthog = "^3.7.0"
 aiofiles = "^24.1.0"
+pyotp = "^2.9.0"
 
 [tool.poetry.group.dev.dependencies]
 isort = "^5.13.2"

skyvern/forge/sdk/services/bitwarden.py

@@ -16,7 +16,6 @@
     BitwardenLoginError,
     BitwardenLogoutError,
     BitwardenSyncError,
-    BitwardenTOTPError,
     BitwardenUnlockError,
 )
 
@@ -211,38 +210,19 @@ async def _get_secret_value_from_url(
                 collection_id_str = f" in collection with ID: {collection_id}" if collection_id else ""
                 raise BitwardenListItemsError(f"No items found in Bitwarden for URL: {url}{collection_id_str}")
 
-            # TODO (kerem): To make this more robust, we need to store the item id of the totp login item
-            # and use it here to get the TOTP code for that specific item
-            totp_command = ["bw", "get", "totp", domain, "--session", session_key]
-            if bw_organization_id:
-                # We need to add this filter because the TOTP command fails if there are multiple results
-                # For now, we require that the bitwarden organization id has only one totp login item for the domain
-                totp_command.extend(["--organizationid", bw_organization_id])
-            totp_result = BitwardenService.run_command(totp_command)
-
-            if totp_result.stderr and "Event post failed" not in totp_result.stderr:
-                LOG.warning(
-                    "Bitwarden TOTP Error",
-                    error=totp_result.stderr,
-                    e=BitwardenTOTPError(totp_result.stderr),
-                )
-            totp_code = totp_result.stdout
             bitwarden_result: list[BitwardenQueryResult] = [
                 BitwardenQueryResult(
                     credential={
                         BitwardenConstants.USERNAME: item.get("login", {}).get("username", ""),
                         BitwardenConstants.PASSWORD: item.get("login", {}).get("password", ""),
+                        BitwardenConstants.TOTP: item.get("login", {}).get("totp", ""),
                     },
                     uris=[uri.get("uri") for uri in item.get("login", {}).get("uris", []) if "uri" in uri],
                 )
                 for item in items
                 if "login" in item
             ]
 
-            if totp_code:
-                for single_result in bitwarden_result:
-                    single_result.credential[BitwardenConstants.TOTP] = totp_code
-
             if len(bitwarden_result) == 0:
                 return {}
 

skyvern/forge/sdk/workflow/context_manager.py

@@ -199,6 +199,8 @@ async def register_parameter_value(
                     if BitwardenConstants.TOTP in secret_credentials and secret_credentials[BitwardenConstants.TOTP]:
                         totp_secret_id = f"{random_secret_id}_totp"
                         self.secrets[totp_secret_id] = BitwardenConstants.TOTP
+                        totp_secret_value = self.totp_secret_value_key(totp_secret_id)
+                        self.secrets[totp_secret_value] = secret_credentials[BitwardenConstants.TOTP]
                         self.values[parameter.key]["totp"] = totp_secret_id
 
             except BitwardenBaseError as e:
@@ -451,6 +453,9 @@ async def register_block_parameters(
             self.parameters[parameter.key] = parameter
             await self.register_parameter_value(aws_client, parameter, organization)
 
+    def totp_secret_value_key(self, totp_secret_id: str) -> str:
+        return f"{totp_secret_id}_value"
+
 
 class WorkflowContextManager:
     aws_client: AsyncAWSClient

skyvern/webeye/actions/handler.py

@@ -7,6 +7,7 @@
 from datetime import datetime, timedelta, timezone
 from typing import Any, Awaitable, Callable, List
 
+import pyotp
 import structlog
 from playwright.async_api import FileChooser, Frame, Locator, Page, TimeoutError
 from pydantic import BaseModel
@@ -1082,8 +1083,9 @@ async def get_actual_value_of_parameter_if_secret(task: Task, parameter: str) ->
     secret_value = workflow_run_context.get_original_secret_value_or_none(parameter)
 
     if secret_value == BitwardenConstants.TOTP:
-        secrets = await workflow_run_context.get_secrets_from_password_manager()
-        secret_value = secrets[BitwardenConstants.TOTP]
+        totp_secret_key = workflow_run_context.totp_secret_value_key(secret_value)
+        totp_secret = workflow_run_context.get_original_secret_value_or_none(totp_secret_key)
+        secret_value = pyotp.TOTP(totp_secret).now()
     return secret_value if secret_value is not None else parameter