diff --git a/turbinia/config/__init__.py b/turbinia/config/__init__.py
index 403981fbd..d0703f253 100644
--- a/turbinia/config/__init__.py
+++ b/turbinia/config/__init__.py
@@ -87,6 +87,8 @@
     'GCS_OUTPUT_PATH',
     'RECIPE_FILE_DIR',
     'STACKDRIVER_TRACEBACK',
+    # AWS config
+    'AWS_ZONE',
     # REDIS CONFIG
     'REDIS_HOST',
     'REDIS_PORT',
diff --git a/turbinia/config/turbinia_config_tmpl.py b/turbinia/config/turbinia_config_tmpl.py
index 3a6bac8fc..1de131365 100644
--- a/turbinia/config/turbinia_config_tmpl.py
+++ b/turbinia/config/turbinia_config_tmpl.py
@@ -24,8 +24,7 @@
 # separate when running with the same Cloud projects or backend servers.
 INSTANCE_ID = 'turbinia-instance1'
 
-# Which Cloud provider to use. Valid options are 'Local' and 'GCP'. Use 'GCP'
-# for GCP or hybrid installations, and 'Local' for local installations.
+# Which Cloud provider to use. Valid options are 'Local', 'GCP' or 'AWS'.
 CLOUD_PROVIDER = 'Local'
 
 # Task manager only supports 'Celery'.
@@ -278,13 +277,11 @@
 ################################################################################
 #                        Google Cloud Platform (GCP)
 #
-# Options in this section are required if the TASK_MANAGER is set to 'PSQ'.
+# Options in this section are required if the CLOUD_PROVIDER is set to 'GCP'.
 ################################################################################
 
 # GCP project, region and zone where Turbinia will run.  Note that Turbinia does
-# not currently support multi-zone operation.  Even if you are running Turbinia
-# in Hybrid mode (with the Server and Workers running on local machines), you
-# will still need to provide these three parameters.
+# not currently support multi-zone operation.
 TURBINIA_PROJECT = None
 TURBINIA_ZONE = None
 TURBINIA_REGION = None
@@ -301,6 +298,15 @@
 # Set this to True if you would like to enable Google Cloud Error Reporting.
 STACKDRIVER_TRACEBACK = False
 
+################################################################################
+#                        Amazon Web Services (AWS)
+#
+# Options in this section are required if the CLOUD_PROVIDER is set to 'AWS'.
+################################################################################
+
+# The default AWS zone being used.
+AWS_ZONE = None
+
 ################################################################################
 #                           Celery / Redis / Kombu
 #
diff --git a/turbinia/evidence.py b/turbinia/evidence.py
index 56a831b28..c72567ccd 100644
--- a/turbinia/evidence.py
+++ b/turbinia/evidence.py
@@ -42,6 +42,8 @@
 config.LoadConfig()
 if config.CLOUD_PROVIDER.lower() == 'gcp':
   from turbinia.processors import google_cloud
+elif config.CLOUD_PROVIDER.lower() == 'aws':
+  from turbinia.processors import aws
 
 log = logging.getLogger(__name__)
 
@@ -917,6 +919,56 @@ def _postprocess(self):
         self.state[EvidenceState.ATTACHED] = False
 
 
+class AwsEbsVolume(Evidence):
+  """Evidence object for an AWS EBS Disk.
+
+  Attributes:
+    zone: The geographic zone.
+    volume_id: The unique volume id for the disk.
+    disk_name: Optional name for the disk.
+  """
+
+  REQUIRED_ATTRIBUTES = ['volume_id', 'zone']
+  POSSIBLE_STATES = [EvidenceState.ATTACHED, EvidenceState.MOUNTED]
+
+  def __init__(
+      self, zone, volume_id, disk_name=None, mount_partition=1, *args,
+      **kwargs):
+    """Initialization for AWS EBS Disk."""
+    super(AwsEbsVolume, self).__init__(*args, **kwargs)
+    self.zone = zone
+    self.volume_id = volume_id
+    self.disk_name = disk_name
+    self.mount_partition = mount_partition
+    self.partition_paths = None
+    self.cloud_only = True
+    self.resource_tracked = True
+    self.resource_id = self.volume_id
+    self.device_path = None
+
+  @property
+  def name(self):
+    if self._name:
+      return self._name
+    elif self.disk_name:
+      return ':'.join((self.type, self.disk_name, self.volume_id))
+    else:
+      return ':'.join((self.type, self.volume_id))
+
+  def _preprocess(self, _, required_states):
+    if EvidenceState.ATTACHED in required_states:
+      self.device_path, partition_paths = aws.PreprocessAttachDisk(
+          self.volume_id)
+      self.partition_paths = partition_paths
+      self.local_path = self.device_path
+      self.state[EvidenceState.ATTACHED] = True
+
+  def _postprocess(self):
+    if self.state[EvidenceState.ATTACHED]:
+      aws.PostprocessDetachDisk(self.volume_id)
+      self.state[EvidenceState.ATTACHED] = False
+
+
 class GoogleCloudDisk(Evidence):
   """Evidence object for a Google Cloud Disk.
 
diff --git a/turbinia/lib/utils.py b/turbinia/lib/utils.py
index 36adde6c6..8bb87d6cd 100644
--- a/turbinia/lib/utils.py
+++ b/turbinia/lib/utils.py
@@ -19,6 +19,7 @@
 import logging
 import os
 import subprocess
+import stat
 import tempfile
 import threading
 
@@ -173,6 +174,21 @@ def get_exe_path(filename):
   return binary
 
 
+def is_block_device(path):
+  """Checks path to determine whether it is a block device.
+
+  Args:
+      path: String of path to check.
+
+  Returns:
+      Bool indicating success.
+  """
+  if not os.path.exists(path):
+    return False
+  mode = os.stat(path).st_mode
+  return stat.S_ISBLK(mode)
+
+
 def bruteforce_password_hashes(
     password_hashes, tmp_dir, timeout=300, extra_args=''):
   """Bruteforce password hashes using Hashcat or john.
diff --git a/turbinia/processors/aws.py b/turbinia/processors/aws.py
new file mode 100644
index 000000000..836e0b451
--- /dev/null
+++ b/turbinia/processors/aws.py
@@ -0,0 +1,234 @@
+# -*- coding: utf-8 -*-
+# Copyright 2024 Google Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Evidence processor for AWS resources."""
+
+import glob
+import json
+import logging
+import os
+import subprocess
+import time
+import urllib
+
+from libcloudforensics.providers.aws.internal import account
+from prometheus_client import Counter
+from turbinia import config
+from turbinia.lib import utils
+from turbinia import TurbiniaException
+
+log = logging.getLogger('turbinia')
+
+RETRY_MAX = 10
+ATTACH_SLEEP_TIME = 3
+DETACH_SLEEP_TIME = 5
+
+turbinia_nonexisting_disk_path = Counter(
+    'turbinia_nonexisting_disk_path',
+    'Total number of non existing disk paths after attempts to attach')
+
+
+def GetDevicePath():
+  """Gets the next free block device path from the local system.
+
+  Returns:
+    new_path(str|None): The new device path name if one is found, else None.
+  """
+  path_base = '/dev/sd'
+  # Recommended device names are /dev/sd[f-p] as per:
+  # https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/device_naming.html#available-ec2-device-names
+  have_path = False
+  for i in range(ord('f'), ord('p') + 1):
+    new_path = f'{path_base}{chr(i)}'
+    # Using `exists` instead of `is_block_device` because even if the file
+    # exists and isn't a block device we still won't be able to use it as a new
+    # device path.
+    if not os.path.exists(new_path):
+      have_path = True
+      break
+
+  if have_path:
+    return new_path
+
+  return None
+
+
+def CheckVolumeAttached(disk_id):
+  """Uses lsblk to determine if the disk is already attached.
+
+  AWS EBS puts the volume ID in the serial number for the device:
+  https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/nvme-ebs-volumes.html#identify-nvme-ebs-device
+
+  Returns:
+    device_name(str|None): The name of the device if it is attached, else None
+
+  Raises:
+    TurbiniaException: If the output from lsblk cannot be parsed.
+  """
+  # From testing the volume ID seems to have the dash removed from the volume ID listed
+  # in the AWS console.
+  serial_num = disk_id.replace('-', '')
+  command = ['lsblk', '-o', 'NAME,SERIAL', '-J']
+  result = subprocess.run(command, check=True, capture_output=True, text=True)
+  device_name = None
+
+  if result.returncode == 0:
+    try:
+      lsblk_results = json.loads(result.stdout)
+    except json.JSONDecodeError as exception:
+      raise TurbiniaException(
+          f'Unable to parse output from {command}: {exception}')
+
+    for device in lsblk_results.get('blockdevices', []):
+      if device.get('serial').lower() == serial_num.lower() and device.get(
+          'name'):
+        device_name = f'/dev/{device.get("name")}'
+        log.info(
+            f'Found device {device_name} attached with serial {serial_num}')
+        break
+  else:
+    log.info(
+        f'Received non-zero exit status {result.returncode} from {command}')
+
+  return device_name
+
+
+def GetLocalInstanceId():
+  """Gets the instance Id of the current machine.
+
+  Returns:
+    The instance Id as a string
+
+  Raises:
+    TurbiniaException: If instance name cannot be determined from metadata
+        server.
+  """
+  req = urllib.request.Request(
+      'http://169.254.169.254/latest/meta-data/instance-id')
+  try:
+    instance = urllib.request.urlopen(req).read().decode('utf-8')
+  except urllib.error.HTTPError as exception:
+    raise TurbiniaException(f'Could not get instance name: {exception}')
+
+  return instance
+
+
+def PreprocessAttachDisk(volume_id):
+  """Attaches AWS EBS volume to an instance.
+
+  Args:
+    disk_id(str): The name of volume to attach.
+
+  Returns:
+    (str, list(str)): a tuple consisting of the path to the 'disk' block device
+      and a list of paths to partition block devices. For example:
+      (
+       '/dev/sdf',
+       ['/dev/sdf1', '/dev/sdf2']
+      )
+
+  Raises:
+    TurbiniaException: If the device is not a block device or the config does
+        not have the required variables configured.
+  """
+  # Check if volume is already attached
+  attached_device = CheckVolumeAttached(volume_id)
+  if attached_device:
+    log.info(f'Disk {volume_id} already attached as {attached_device}')
+    # TODO: Fix globbing for partitions
+    return (attached_device, sorted(glob.glob(f'{attached_device}+')))
+
+  # Volume is not attached so need to attach it
+  config.LoadConfig()
+  if not config.AWS_ZONE:
+    msg = f'AWS_ZONE must be set in configuration file in order to attach AWS disks.'
+    log.error(msg)
+    raise TurbiniaException(msg)
+
+  aws_account = account.AWSAccount(config.AWS_ZONE)
+  instance_id = GetLocalInstanceId()
+  instance = aws_account.ec2.GetInstanceById(instance_id)
+  device_path = GetDevicePath()
+
+  instance.AttachVolume(aws_account.ebs.GetVolumeById(volume_id), device_path)
+
+  # Make sure we have a proper block device
+  for _ in range(RETRY_MAX):
+    # The device path is provided in the above attach volume command but that
+    # name/path is not guaranted to be the actual device name that is used by
+    # the host so we need to check the device names again here.  See here for
+    # more details:
+    # https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/nvme-ebs-volumes.html#identify-nvme-ebs-device
+    device_path = CheckVolumeAttached(volume_id)
+    if device_path and utils.is_block_device(device_path):
+      log.info(f'Block device {device_path:s} successfully attached')
+      break
+    if device_path and os.path.exists(device_path):
+      log.info(
+          f'Block device {device_path:s} mode is '
+          f'{os.stat(device_path).st_mode}')
+    time.sleep(ATTACH_SLEEP_TIME)
+
+  # Final sleep to allow time between API calls.
+  time.sleep(ATTACH_SLEEP_TIME)
+
+  message = None
+  if not device_path:
+    message = 'No valid device paths found after attaching'
+  elif not os.path.exists(device_path):
+    turbinia_nonexisting_disk_path.inc()
+    message = f'Device path {device_path:s} does not exist'
+  elif not util.is_block_device(device_path):
+    message = f'Device path {device_path:s} is not a block device'
+  if message:
+    log.error(message)
+    raise TurbiniaException(message)
+
+  # TODO: Fix globbing for partitions
+  return (device_path, sorted(glob.glob(f'{device_path}+')))
+
+
+def PostprocessDetachDisk(volume_id):
+  """Detaches AWS EBS volume from an instance.
+
+  Args:
+    volume_id(str): The name of the Cloud Disk to detach.
+  """
+  attached_device = CheckVolumeAttached(volume_id)
+  if not attached_device:
+    log.info(f'Disk {volume_id} no longer attached')
+    return
+
+  config.LoadConfig()
+  aws_account = account.AWSAccount(config.AWS_ZONE)
+  instance_id = GetLocalInstanceId()
+  instance = aws_account.ec2.GetInstanceById(instance_id)
+
+  log.info(f'Detaching disk {volume_id:s} from instance {instance_id:s}')
+  instance.DetachVolume(
+      aws_account.ebs.GetVolumeById(volume_id), attached_device)
+
+  # Make sure device is Detached
+  for _ in range(RETRY_MAX):
+    if not os.path.exists(attached_device):
+      log.info(f'Block device {attached_device:s} is no longer attached')
+      break
+    time.sleep(DETACH_SLEEP_TIME)
+
+  if os.path.exists(attached_device):
+    raise TurbiniaException(
+        f'Could not detach volume {volume_id} with device name '
+        f'{attached_device}')
+  else:
+    log.info(f'Detached volume {volume_id} with device name {attached_device}')
diff --git a/turbinia/processors/google_cloud.py b/turbinia/processors/google_cloud.py
index 2f7c26046..2be4cbc45 100644
--- a/turbinia/processors/google_cloud.py
+++ b/turbinia/processors/google_cloud.py
@@ -19,7 +19,6 @@
 import glob
 import logging
 import os
-import stat
 import time
 
 from six.moves import urllib
@@ -27,6 +26,7 @@
 from libcloudforensics.providers.gcp.internal import project as gcp_project
 from prometheus_client import Counter
 from turbinia import config
+from turbinia.lib import util
 from turbinia import TurbiniaException
 
 log = logging.getLogger(__name__)
@@ -40,21 +40,6 @@
     'Total number of non existing disk paths after attempts to attach')
 
 
-def IsBlockDevice(path):
-  """Checks path to determine whether it is a block device.
-
-  Args:
-      path: String of path to check.
-
-  Returns:
-      Bool indicating success.
-  """
-  if not os.path.exists(path):
-    return False
-  mode = os.stat(path).st_mode
-  return stat.S_ISBLK(mode)
-
-
 def GetLocalInstanceName():
   """Gets the instance name of the current machine.
 
@@ -97,7 +82,7 @@ def PreprocessAttachDisk(disk_name):
     TurbiniaException: If the device is not a block device.
   """
   path = f'/dev/disk/by-id/google-{disk_name:s}'
-  if IsBlockDevice(path):
+  if util.is_block_device(path):
     log.info(f'Disk {disk_name:s} already attached!')
     return (path, sorted(glob.glob(f'{path:s}-part*')))
 
@@ -113,7 +98,7 @@ def PreprocessAttachDisk(disk_name):
 
   # Make sure we have a proper block device
   for _ in range(RETRY_MAX):
-    if IsBlockDevice(path):
+    if util.is_block_device(path):
       log.info(f'Block device {path:s} successfully attached')
       break
     if os.path.exists(path):
@@ -127,7 +112,7 @@ def PreprocessAttachDisk(disk_name):
   if not os.path.exists(path):
     turbinia_nonexisting_disk_path.inc()
     message = f'Device path {path:s} does not exist'
-  elif not IsBlockDevice(path):
+  elif not util.is_block_device(path):
     message = f'Device path {path:s} is not a block device'
   if message:
     log.error(message)
@@ -149,7 +134,7 @@ def PostprocessDetachDisk(disk_name, local_path):
   else:
     path = f'/dev/disk/by-id/google-{disk_name:s}'
 
-  if not IsBlockDevice(path):
+  if not util.is_block_device(path):
     log.info(f'Disk {disk_name:s} already detached!')
     return