Add support for image formats when exporting to GCS

Also, command line / reference example to invoke the exporting of a disk image to GCS.

Author: ramo-j

libcloudforensics/providers/gcp/forensics.py

@@ -252,6 +252,50 @@ def CreateDiskFromGCSImage(
   return result
 
 
+def CopyDisksToGCS(source_project: str,
+                   source_disk: str,
+                   destination_bucket: str,
+                   destination_directory: str,
+                   image_format: str) -> str:
+  """Given a VM, copy the disks to a GCS bucket.
+
+  Args:
+    source_project: The project containing the disk to copy
+    source_disk: The name of the disk to copy
+    destination_bucket: The destination bucket to store the disk copy
+    destination_directory: The directory in the bucket in which to store the
+        disk image
+    image_format: The image format to use. Supported formats documented at
+        https://github.com/GoogleCloudPlatform/compute-image-import/blob/edee48bddbe159100da9ad961131a4beb0f12158/cli_tools/gce_vm_image_export/README.md?plain=1#L3
+  """
+  try:
+    src_project = gcp_project.GoogleCloudProject(source_project)
+    disk_to_copy = src_project.compute.GetDisk(source_disk)
+    copied_image = src_project.compute.CreateImageFromDisk(disk_to_copy)
+    return copied_image.ExportImage(
+      gcs_output_folder=f'gs://{destination_bucket}/{destination_directory}',
+      image_format=image_format,
+      output_name=disk_to_copy.name)
+  except (RefreshError, DefaultCredentialsError) as exception:
+    raise errors.CredentialsConfigurationError(
+        'Something is wrong with your Application Default Credentials. Try '
+        'running: $ gcloud auth application-default login: {0!s}'.format(
+            exception),
+        __name__) from exception
+  except HttpError as exception:
+    if exception.resp.status == 403:
+      raise errors.CredentialsConfigurationError(
+          'Make sure you have the appropriate permissions on the project: '
+          '{0!s}'.format(exception),
+          __name__) from exception
+    if exception.resp.status == 404:
+      raise errors.ResourceNotFoundError(
+          'GCP resource not found. Maybe a typo in the project / instance / '
+          'disk name?',
+          __name__) from exception
+    raise RuntimeError(exception) from exception
+
+
 def AddDenyAllFirewallRules(
     project_id: str,
     network: str,

libcloudforensics/providers/gcp/internal/compute.py

@@ -2253,39 +2253,49 @@ def GetOperation(self) -> Dict[str, Any]:
     return response
 
   def ExportImage(
-      self, gcs_output_folder: str, output_name: Optional[str] = None) -> None:
-    """Export compute image to Google Cloud storage.
-
+      self,
+      gcs_output_folder: str,
+      image_format: str,
+      output_name: Optional[str]) -> str:
+    """Export compute image to Google Cloud Storage.
+    
     Exported image is compressed and stored in .tar.gz format.
 
     Args:
       gcs_output_folder (str): Folder path of the exported image.
+      image_format (str): The image format to use for the export.
       output_name (str): Optional. Name of the output file. Name will be
           appended with .tar.gz. Default is [image_name].tar.gz.
-
+    Returns:
+      str: The full path of the exported image.
     Raises:
       InvalidNameError: If exported image name is invalid.
     """
-
     if output_name:
       if not common.REGEX_DISK_NAME.match(output_name):
         raise errors.InvalidNameError(
             'Exported image name {0:s} does not comply with {1:s}'.format(
                 output_name, common.REGEX_DISK_NAME.pattern),
             __name__)
-      full_path = '{0:s}.tar.gz'.format(
-          os.path.join(gcs_output_folder, output_name))
+      full_path = '{0:s}'.format(os.path.join(gcs_output_folder, output_name))
     else:
-      full_path = '{0:s}.tar.gz'.format(
-          os.path.join(gcs_output_folder, self.name))
+      full_path = '{0:s}'.format(os.path.join(gcs_output_folder, self.name))
+    if not image_format:
+      full_path = '{0:s}.tar.gz'.format(full_path)
+    else:
+      full_path = '{0:s}.{1:s}'.format(full_path, image_format)
+			
+    build_args = [
+        '-source_image={0:s}'.format(self.name),
+        '-destination_uri={0:s}'.format(full_path),
+        '-client_id=api',
+    ]
+    if image_format:
+      build_args.append('-format={0:s}'.format(image_format))
     build_body = {
         'timeout': '86400s',
-        'steps': [{
-            'args': [
-                '-source_image={0:s}'.format(self.name),
-                '-destination_uri={0:s}'.format(full_path),
-                '-client_id=api',
-            ],
+        'steps': [{		
+            'args': build_args,
             'name': 'gcr.io/compute-image-tools/gce_vm_image_export:release',
             'env': []
         }],
@@ -2295,6 +2305,7 @@ def ExportImage(
     response = cloud_build.CreateBuild(build_body)
     cloud_build.BlockOperation(response)
     logger.info('Image {0:s} exported to {1:s}.'.format(self.name, full_path))
+    return full_path
 
   def Delete(self) -> None:
     """Delete Compute Disk Image from a project."""

tools/cli.py

@@ -52,6 +52,7 @@
         'bucketacls': gcp_cli.GetBucketACLs,
         'bucketsize': gcp_cli.GetBucketSize,
         'copydisk': gcp_cli.CreateDiskCopy,
+        'copydisktogcs': gcp_cli.CopyDiskToGCS,
         'creatediskgcs': gcp_cli.CreateDiskFromGCSImage,
         'deleteinstance': gcp_cli.DeleteInstance,
         'deleteobject': gcp_cli.DeleteObject,
@@ -387,6 +388,16 @@ def Main() -> None:
                                 'The default behavior is to use the same disk '
                                 'type as the source disk.', None)
             ])
+  AddParser('gcp', gcp_subparsers, 'copydisktogcs',
+            'Copy a disk content into GCS.',
+            args=[
+                ('project', 'Source GCP project containing the disk to copy',
+                 ''),
+                ('disk_name', 'Name of the disk to copy.', ''),
+                ('bucket', 'Name of the destination bucket.', ''),
+                ('directory', 'Destination directory path in the GCS bucket.',
+                 ''),
+                ('image_format', 'Image format.', '')])
   AddParser('gcp', gcp_subparsers, 'startvm', 'Start a forensic analysis VM.',
             args=[
                 ('instance_name', 'Name of the GCE instance to create.',

tools/gcp_cli.py

@@ -113,6 +113,21 @@ def CreateDiskCopy(args: 'argparse.Namespace') -> None:
   logger.info('Name: {0:s}'.format(disk.name))
 
 
+def CopyDiskToGCS(args):
+  """Make a copy of a GCE disk into GCS storage.
+
+  Args:
+    args (argparse.Namespace): Arguments from ArgumentParser.
+  """
+  gcs_path = forensics.CopyDisksToGCS(source_project=args.project,
+                                      source_disk=args.disk_name,
+                                      destination_bucket=args.bucket,
+                                      destination_directory=args.directory,
+                                      image_format=args.image_format)
+  logger.info('Disk copy completed.')
+  logger.info('Location: {0:s}'.format(gcs_path))
+
+
 def DeleteInstance(args: 'argparse.Namespace') -> None:
   """Deletes a GCE instance.
 
@@ -541,6 +556,7 @@ def GKEWorkloadQuarantine(args: 'argparse.Namespace') -> None:
                                   args.namespace, args.workload,
                                   exempted_src_ips=exempted_src_ips)
 
+
 def GKEEnumerate(args: 'argparse.Namespace') -> None:
   """Enumerate GKE cluster objects.