shorten random string for secret value (#3263)

skyvern/forge/sdk/workflow/context_manager.py

@@ -1,5 +1,4 @@
 import copy
-import uuid
 from typing import TYPE_CHECKING, Any, Self
 
 import structlog
@@ -38,6 +37,7 @@ from skyvern.forge.sdk.workflow.models.parameter import (
     WorkflowParameter,
     WorkflowParameterType,
 )
+from skyvern.utils.strings import generate_random_string
 
 if TYPE_CHECKING:
     from skyvern.forge.sdk.workflow.models.workflow import WorkflowRunParameter
@@ -193,7 +193,7 @@ class WorkflowRunContext:
 
     @staticmethod
     def generate_random_secret_id() -> str:
-        return f"secret_{uuid.uuid4()}"
+        return f"secret_{generate_random_string()}"
 
     async def _get_credential_vault_and_item_ids(self, credential_id: str) -> tuple[str, str]:
         """

skyvern/forge/sdk/workflow/models/block.py

@@ -6,10 +6,8 @@ import asyncio
 import csv
 import json
 import os
-import random
 import re
 import smtplib
-import string
 import textwrap
 import uuid
 from collections import defaultdict
@@ -83,6 +81,7 @@ from skyvern.forge.sdk.workflow.models.parameter import (
 )
 from skyvern.schemas.runs import RunEngine
 from skyvern.schemas.workflows import BlockResult, BlockStatus, BlockType, FileStorageType, FileType
+from skyvern.utils.strings import generate_random_string
 from skyvern.utils.url_validators import prepend_scheme_and_validate_url
 from skyvern.webeye.browser_factory import BrowserState
 from skyvern.webeye.utils.page import SkyvernFrame
@@ -91,11 +90,6 @@ LOG = structlog.get_logger()
 jinja_sandbox_env = SandboxedEnvironment()
 
 
-def _generate_random_string(length: int = 8) -> str:
-    """Generate a random string for unique identifiers."""
-    return "".join(random.choices(string.ascii_lowercase + string.digits, k=length))
-
-
 # Mapping from TaskV2Status to the corresponding BlockStatus. Declared once at
 # import time so it is not recreated on each block execution.
 TASKV2_TO_BLOCK_STATUS: dict[TaskV2Status, BlockStatus] = {
@@ -991,7 +985,7 @@ class ForLoopBlock(Block):
 
                     # Update the loop variable reference to point to the extracted loop values
                     # We'll use a temporary key that we can reference
-                    temp_key = f"extracted_loop_values_{_generate_random_string()}"
+                    temp_key = f"extracted_loop_values_{generate_random_string()}"
                     workflow_run_context.set_value(temp_key, loop_values)
                     self.loop_variable_reference = temp_key
 
@@ -1119,7 +1113,7 @@ class ForLoopBlock(Block):
 
         output_param = OutputParameter(
             output_parameter_id=str(uuid.uuid4()),
-            key=f"natural_lang_extraction_{_generate_random_string()}",
+            key=f"natural_lang_extraction_{generate_random_string()}",
             workflow_id=self.output_parameter.workflow_id,
             created_at=datetime.now(),
             modified_at=datetime.now(),
@@ -1128,7 +1122,7 @@ class ForLoopBlock(Block):
         )
 
         return ExtractionBlock(
-            label=f"natural_lang_extraction_{_generate_random_string()}",
+            label=f"natural_lang_extraction_{generate_random_string()}",
             data_extraction_goal=extraction_goal,
             data_schema=data_schema,
             output_parameter=output_param,

skyvern/services/task_v2_service.py

@@ -1,6 +1,4 @@
 import json
-import os
-import random
 import string
 from datetime import UTC, datetime
 from typing import Any
@@ -58,6 +56,7 @@ from skyvern.schemas.workflows import (
     WorkflowStatus,
 )
 from skyvern.utils.prompt_engine import load_prompt_with_elements
+from skyvern.utils.strings import generate_random_string
 from skyvern.webeye.browser_factory import BrowserState
 from skyvern.webeye.scraper.scraper import ScrapedPage, scrape_website
 from skyvern.webeye.utils.page import SkyvernFrame
@@ -1045,7 +1044,7 @@ async def _generate_loop_task(
                 artifact_type=ArtifactType.SCREENSHOT_LLM,
                 data=screenshot,
             )
-    loop_random_string = _generate_random_string()
+    loop_random_string = generate_random_string()
     label = f"extraction_task_for_loop_{loop_random_string}"
     loop_values_key = f"loop_values_{loop_random_string}"
     extraction_block_yaml = ExtractionBlockYAML(
@@ -1145,7 +1144,7 @@ async def _generate_loop_task(
     )
     app.WORKFLOW_CONTEXT_MANAGER.add_context_parameter(workflow_run_id, url_value_context_parameter)
 
-    task_in_loop_label = f"task_in_loop_{_generate_random_string()}"
+    task_in_loop_label = f"task_in_loop_{generate_random_string()}"
     context = skyvern_context.ensure_context()
     task_in_loop_metadata_prompt = prompt_engine.load_prompt(
         "task_v2_generate_task_block",
@@ -1215,7 +1214,7 @@ async def _generate_loop_task(
 
     # use the output parameter of the extraction block to create the for loop block
     for_loop_yaml = ForLoopBlockYAML(
-        label=f"loop_{_generate_random_string()}",
+        label=f"loop_{generate_random_string()}",
         loop_over_parameter_key=loop_for_context_parameter.key,
         loop_blocks=[block_yaml],
     )
@@ -1273,7 +1272,7 @@ async def _generate_extraction_task(
 
     # create OutputParameter for the data_extraction block
     data_schema: dict[str, Any] | list | None = generate_extraction_task_response.get("schema")
-    label = f"data_extraction_{_generate_random_string()}"
+    label = f"data_extraction_{generate_random_string()}"
     url: str | None = None
     if not task_history:
         # data extraction is the very first block
@@ -1312,7 +1311,7 @@ async def _generate_navigation_task(
     totp_identifier: str | None = None,
 ) -> tuple[NavigationBlock, list[BLOCK_YAML_TYPES], list[PARAMETER_YAML_TYPES]]:
     LOG.info("Generating navigation task", navigation_goal=navigation_goal, original_url=original_url)
-    label = f"navigation_{_generate_random_string()}"
+    label = f"navigation_{generate_random_string()}"
     navigation_block_yaml = NavigationBlockYAML(
         label=label,
         url=original_url,
@@ -1346,7 +1345,7 @@ async def _generate_goto_url_task(
 ) -> tuple[UrlBlock, list[BLOCK_YAML_TYPES], list[PARAMETER_YAML_TYPES]]:
     LOG.info("Generating goto url task", url=url)
     # create OutputParameter for the data_extraction block
-    label = f"goto_url_{_generate_random_string()}"
+    label = f"goto_url_{generate_random_string()}"
 
     url_block_yaml = UrlBlockYAML(
         label=label,
@@ -1368,12 +1367,6 @@ async def _generate_goto_url_task(
     )
 
 
-def _generate_random_string(length: int = 5) -> str:
-    # Use the current timestamp as the seed
-    random.seed(os.urandom(16))
-    return "".join(random.choices(RANDOM_STRING_POOL, k=length))
-
-
 async def get_thought_timelines(*, task_v2_id: str, organization_id: str) -> list[WorkflowRunTimeline]:
     thoughts = await app.DATABASE.get_thoughts(
         task_v2_id=task_v2_id,

skyvern/utils/strings.py

@@ -0,0 +1,11 @@
+import os
+import random
+import string
+
+RANDOM_STRING_POOL = string.ascii_letters + string.digits
+
+
+def generate_random_string(length: int = 5) -> str:
+    # Use the os.urandom(16) as the seed
+    random.seed(os.urandom(16))
+    return "".join(random.choices(RANDOM_STRING_POOL, k=length))