Merge pull request #6 from SkyflowFoundry/IE-521

IE-521: updated pipeline workflow, added code comments

Author: skyflow-srivyshnavi

.github/workflows/migrate_pipelines.yml

@@ -16,41 +16,21 @@ on:
         description: "Source datastore JSON. Provide either an ftpServer or s3Bucket object."
         required: true
         default: |
-          {
-            "ftpServer": {
-              "transferProtocol": "FTPS",
-              "plainText": {
-                "hostname": "",
-                "port": "",
-                "username": "",
-                "password": "",
-              },
-              "skyflowHosted": false
-            }
-          }
+          {"ftpServer":{"transferProtocol": "FTPS","plainText": {"hostname": "","port": "","username": "","password": ""},"skyflowHosted": false}}
       target_datastore_config:
         description: "Destination datastore JSON. Provide either an ftpServer or s3Bucket object."
         required: true
         default: |
-          {
-            "s3Bucket": {
-              "name": "",
-              "region": "",
-              "assumedRoleARN": ""
-            }
-          }
-      source_vault_id:
-        description: "Source Vault ID."
-        required: false
+          {"s3Bucket": {"name": "","region": "","assumedRoleARN": ""}}
       pipeline_id:
         description: "PipelineID to be migrated."
-        required: false
+        required: true
         default: ""
       target_vault_id:
         description: "Target Vault ID"
         required: true
       source_account_access_token:
-        description: "Access token of the Source Account. (Not required, if config file is selected)"
+        description: "Access token of the Source Account"
         required: false
       target_account_access_token:
         description: "Access token of the Target Account"
@@ -108,7 +88,6 @@ jobs:
           PIPELINE_ID: ${{ github.event.inputs.pipeline_id }}
           SOURCE_DATASTORE_CONFIG: ${{ github.event.inputs.source_datastore_config }}
           TARGET_DATASTORE_CONFIG: ${{ github.event.inputs.target_datastore_config }}
-          SOURCE_VAULT_ID: ${{ github.event.inputs.source_vault_id }}
           TARGET_VAULT_ID: ${{ github.event.inputs.target_vault_id }}
           SOURCE_ACCOUNT_AUTH: ${{ github.event.inputs.source_account_access_token }}
           TARGET_ACCOUNT_AUTH: ${{ github.event.inputs.target_account_access_token }}

README.md

@@ -117,11 +117,12 @@ Note: Please note that if all values are provided `config_file` will take the pr
 
 ### Pipelines Migration
 
-Migrates a pipeline definition from the source vault to the target vault. 
+Migrates a pipeline from the source vault to the target vault. 
 
 ##### Parameters:
 - **`source_and_target_env`**: Source and Target Env's.
-- **`pipeline_id`**: Pipeline ID to migrate. Get the pipeline ID from Studio.
+- **`pipeline_id`**: Pipeline ID to migrate.
+- **`target_vault_id`**: Target Vault ID.
 - **`source_datastore_config`**: JSON object that replaces the source datastore configuration. Provide either an `ftpServer` or `s3Bucket` object with the required credentials.
 - **`target_datastore_config`**: JSON object that replaces the destination datastore configuration. Provide either an `ftpServer` or `s3Bucket` object with the required credentials.
 - **`source_account_access_token`**: Access token of the source account.

migrate_connections.py

@@ -29,6 +29,7 @@
 
 
 def list_connections(vault_id):
+    """Lists inbound + outbound connections for a vault."""
     connections = []
     response = requests.get(
         f"{SOURCE_ENV_URL}/v1/gateway/outboundRoutes?vaultID={vault_id}",
@@ -45,6 +46,7 @@ def list_connections(vault_id):
     return connections
 
 def get_connection(connection_id):
+    """Fetches a single connection"""
     # /inboundRoutes can also fetch outbound connection details
     response = requests.get(
         f"{SOURCE_ENV_URL}/v1/gateway/inboundRoutes/{connection_id}",
@@ -54,6 +56,7 @@ def get_connection(connection_id):
     return response.json()
 
 def create_connection(connection):
+    """Creates connection"""
     route = "outboundRoutes" if connection["mode"] == "EGRESS" else "inboundRoutes"
     response = requests.post(
         f"{TARGET_ENV_URL}/v1/gateway/{route}",
@@ -64,8 +67,10 @@ def create_connection(connection):
 
 
 def transform_connection_payload(source_resource):
+    """Transforms source connection payload to target payload."""
     transformed_resource = source_resource
     transformed_resource["vaultID"] = TARGET_VAULT_ID
+    # drop basic audit and invocation URL
     if "BasicAudit" in transformed_resource.keys():
         del transformed_resource["BasicAudit"]
     for route in transformed_resource["routes"]:
@@ -74,6 +79,7 @@ def transform_connection_payload(source_resource):
 
 
 def main(connection_ids=None):
+    """Migrates connections"""
     try:
         print("-- Initiating Connections migration --")
         connections = []

migrate_pipelines.py

@@ -4,9 +4,10 @@
 import requests
 from typing import Any, Dict, List, Optional
 
+from dotenv import load_dotenv
+load_dotenv()
 
 PIPELINE_ID = os.getenv("PIPELINE_ID")
-SOURCE_VAULT_ID = os.getenv("SOURCE_VAULT_ID")
 TARGET_VAULT_ID = os.getenv("TARGET_VAULT_ID")
 SOURCE_ACCOUNT_ID = os.getenv("SOURCE_ACCOUNT_ID")
 TARGET_ACCOUNT_ID = os.getenv("TARGET_ACCOUNT_ID")
@@ -35,7 +36,7 @@
 }
 
 def list_pipelines(vault_id: str) -> List[Dict[str, Any]]:
-    """Return all pipelines in the supplied vault."""
+    """Lists Pipelines"""
     pipelines = []
     response = requests.get(
         f"{SOURCE_ENV_URL}/v1/pipelines?vaultID={vault_id}",
@@ -46,7 +47,7 @@ def list_pipelines(vault_id: str) -> List[Dict[str, Any]]:
     return pipelines
 
 def get_pipeline(pipeline_id: str) -> Dict[str, Any]:
-    """Fetch a single pipeline definition from the source environment."""
+    """Fetches a single pipeline"""
     response = requests.get(
         f"{SOURCE_ENV_URL}/v1/pipelines/{pipeline_id}",
         headers=SOURCE_ACCOUNT_HEADERS,
@@ -55,7 +56,7 @@ def get_pipeline(pipeline_id: str) -> Dict[str, Any]:
     return response.json()[PIPELINE]
 
 def create_pipeline(pipeline: Dict[str, Any]) -> requests.Response:
-    """Create a pipeline in the target environment."""
+    """Creates a pipeline"""
     response = requests.post(
         f"{TARGET_ENV_URL}/v1/pipelines",
         json=pipeline,
@@ -170,7 +171,7 @@ def transform_pipeline_payload(
     source_datastore_input: Optional[Dict[str, Any]] = None,
     target_datastore_input: Optional[Dict[str, Any]] = None,
 ) -> Dict[str, Any]:
-    """Prepare the payload for the target pipeline."""
+    """Transforms source pipeline payload to target payload."""
     transformed_resource = copy.deepcopy(source_resource)
     if "ID" in transformed_resource:
         del transformed_resource["ID"]      # remove pipeline ID
@@ -187,7 +188,7 @@ def transform_pipeline_payload(
 
 
 def main(pipeline_id: str) -> None:
-    """pipeline migration"""
+    """Migrates pipeline"""
     try:
         print("-- Initiating Pipelines migration --")
         source_datastore_input = load_datastore_input(SOURCE_DATASTORE_CONFIG, "source")

migrate_policies.py

@@ -26,6 +26,7 @@
 
 
 def get_policy(policy_id):
+    """Fetches a policy"""
     response = requests.get(
         f"{SOURCE_ENV_URL}/v1/policies/{policy_id}", headers=SOURCE_ACCOUNT_HEADERS
     )
@@ -34,6 +35,7 @@ def get_policy(policy_id):
 
 
 def create_policy(policy_data):
+    """Creates a policy"""
     response = requests.post(
         f"{TARGET_ENV_URL}/v1/policies", json=policy_data, headers=TARGET_ACCOUNT_HEADERS
     )
@@ -42,6 +44,7 @@ def create_policy(policy_data):
 
 
 def transform_policy_payload(source_resource):
+    """Transforms source policy payload to target payload."""
     transformed_resource = source_resource["policy"]
     transformed_resource["resource"] = {"ID": TARGET_VAULT_ID, "type": "VAULT"}
     policy_rules = transformed_resource["rules"]
@@ -95,6 +98,7 @@ def transform_policy_payload(source_resource):
 
 
 def main(policy_ids=None):
+    """Migrates policies"""
     try:
         policy_ids = policy_ids if policy_ids else ast.literal_eval(POLICY_IDS)
         policies_created = []

migrate_roles.py

@@ -32,20 +32,23 @@
 
 
 def get_role(role_id):
+    """Fetch a single role definition from the source account."""
     response = requests.get(
         f"{SOURCE_ENV_URL}/v1/roles/{role_id}", headers=SOURCE_ACCOUNT_HEADERS
     )
     response.raise_for_status()
     return response.json()
 
 def get_system_role(role_name):
+    """Return a system role present in the target vault."""
     response = requests.get(
         f"{TARGET_ENV_URL}/v1/roles?name={role_name}&resource.type=VAULT&resource.ID={TARGET_VAULT_ID}", headers=TARGET_ACCOUNT_HEADERS
     )
     response.raise_for_status()
     return response.json()    
 
 def create_role(role):
+    """Create a custom role in the target vault."""
     response = requests.post(
         f"{TARGET_ENV_URL}/v1/roles", json=role, headers=TARGET_ACCOUNT_HEADERS
     )
@@ -54,6 +57,7 @@ def create_role(role):
 
 
 def get_role_policies(role_id):
+    """List all policies attached to the given role."""
     response = requests.get(
         f"{SOURCE_ENV_URL}/v1/roles/{role_id}/policies", headers=SOURCE_ACCOUNT_HEADERS
     )
@@ -62,6 +66,7 @@ def get_role_policies(role_id):
 
 
 def get_role_by_role_name(role_name):
+    """Search the target vault for an existing custom role by name."""
     response = requests.get(
         f"{TARGET_ENV_URL}/v1/roles?name={role_name}&resource.type=VAULT&resource.ID={TARGET_VAULT_ID}",
         headers=TARGET_ACCOUNT_HEADERS,
@@ -71,6 +76,7 @@ def get_role_by_role_name(role_name):
 
 
 def assign_policy_to_role(policy_ids, role_id: list):
+    """Assign the provided policies to the role."""
     for policy_id in policy_ids:
         assign_request = {"ID": policy_id, "roleIDs": role_id}
         response = requests.post(
@@ -82,6 +88,7 @@ def assign_policy_to_role(policy_ids, role_id: list):
     # return response.json()
 
 def list_all_roles() -> list:
+    """Lists custom roles"""
     response = requests.get(
         f"{SOURCE_ENV_URL}/v1/roles?type=CUSTOM&resource.ID={SOURCE_VAULT_ID}&resource.type=VAULT",
         headers=SOURCE_ACCOUNT_HEADERS,
@@ -91,6 +98,7 @@ def list_all_roles() -> list:
 
 
 def transform_role_payload(source_resource):
+    """Transforms source role payload to target payload."""
     transformed_resource = {}
     transformed_resource["roleDefinition"] = source_resource["role"]["definition"]
     permissions: list = source_resource["role"]["definition"]["permissions"]
@@ -104,11 +112,13 @@ def transform_role_payload(source_resource):
             "vaults.read:upstream",
         ]
     ]
+    # remove upstream read permissions that are implicitly granted in the target account
     transformed_resource["roleDefinition"]["permissions"] = new_permissions
     transformed_resource["resource"] = {"ID": TARGET_VAULT_ID, "type": "VAULT"}
     return transformed_resource
 
 def main(role_ids=None):
+    """Migrates roles and their associated policies."""
     try:
         print("-- Initializing Roles migration --")
         should_enable_custom_role_check = SKIP_ROLE_CREATION_IF_ROLE_EXISTS

migrate_service_accounts.py

@@ -27,6 +27,7 @@
 
 
 def list_service_account_roles(service_account_id):
+    """Return every role assigned to the given service account."""
     response = requests.get(
         f"{SOURCE_ENV_URL}/v1/members/{service_account_id}/roles?member.type=SERVICE_ACCOUNT",
         headers=SOURCE_ACCOUNT_HEADERS,
@@ -36,6 +37,7 @@ def list_service_account_roles(service_account_id):
 
 
 def get_service_account(service_account_id):
+    """Fetch a service account definition from the source account."""
     response = requests.get(
         f"{SOURCE_ENV_URL}/v1/serviceAccounts/{service_account_id}",
         headers=SOURCE_ACCOUNT_HEADERS,
@@ -45,6 +47,7 @@ def get_service_account(service_account_id):
 
 
 def create_service_account(service_account):
+    """Create the supplied service account in the target account."""
     response = requests.post(
         f"{TARGET_ENV_URL}/v1/serviceAccounts",
         json=service_account,
@@ -55,6 +58,7 @@ def create_service_account(service_account):
 
 
 def assign_roles_to_service_account(role_ids, service_account_id):
+    """Attach the specified roles to the new service account."""
     for role_id in role_ids:
         assign_request = {
             "ID": role_id,
@@ -69,15 +73,17 @@ def assign_roles_to_service_account(role_ids, service_account_id):
 
 
 def transform_service_account_payload(source_resource):
+    """Strip source-only metadata before creating the service account."""
     transformed_resource = source_resource
     # transformed_resource["resource"] = {"ID": TARGET_VAULT_ID, "type": "VAULT"} // not require due to SA flattening change
-    del transformed_resource["serviceAccount"]["ID"]
+    del transformed_resource["serviceAccount"]["ID"]  # IDs are regenerated per account
     del transformed_resource["serviceAccount"]["namespace"]
     del transformed_resource["serviceAccount"]["BasicAudit"]
     return transformed_resource
 
 
 def main(service_accounts_ids=None):
+    """Migrates service accounts and associated roles."""
     try:
         print("-- Initializing Service accounts migration --")
         service_accounts_ids = (

migrate_vault_roles_and_policies.py

@@ -16,6 +16,7 @@
 
 
 def list_all_vault_custom_roles() -> list:
+    """Return all custom roles of the source vault."""
     response = requests.get(
         f"{SOURCE_ENV_URL}/v1/roles?type=CUSTOM&resource.ID={SOURCE_VAULT_ID}&resource.type=VAULT",
         headers=SOURCE_ACCOUNT_HEADERS,
@@ -25,6 +26,7 @@ def list_all_vault_custom_roles() -> list:
 
 
 def main():
+    """Migrates all the custom roles and policies"""
     try:
         print(f"-- Fetching roles for the vault {SOURCE_VAULT_ID} --")
         roles = list_all_vault_custom_roles()

migrate_vault_schema.py

@@ -26,18 +26,20 @@
     "Authorization": f"Bearer {TARGET_ACCOUNT_AUTH}",
     "Content-Type": "application/json",
 }
-
 def get_vault_details(vaultID: str):
+    """Return the vault metadata and schema"""
     response = requests.get(f"{SOURCE_ENV_URL}/v1/vaults/{vaultID}", headers=SOURCE_ACCOUNT_HEADERS)
     response.raise_for_status()
     return response.json()
 
 def create_vault(create_vault_request_payload):
+    """Creates a vault"""
     response = requests.post(f"{TARGET_ENV_URL}/v1/vaults", json=create_vault_request_payload, headers=TARGET_ACCOUNT_HEADERS)
     response.raise_for_status()
     return response.json()
 
 def transform_payload(vault_details):
+    """Transforms source vault payload to target payload."""
     create_vault_payload = {
         "name": VAULT_NAME if VAULT_NAME else f"UntitledVault{random.randint(0,1000)}" if VAULT_SCHEMA_CONFIG else vault_details["name"],
         "description": VAULT_DESCRIPTION if VAULT_DESCRIPTION else "" if VAULT_SCHEMA_CONFIG else vault_details["description"],
@@ -50,6 +52,7 @@ def transform_payload(vault_details):
     return create_vault_payload
 
 def main():
+    """Migrates vault schema"""
     try:
         print("-- Initializing Vault migration --")
         if VAULT_SCHEMA_CONFIG is not None and VAULT_SCHEMA_CONFIG == "config_file":