Repository: google/turbinia
Pull request #504 (2020-02-14 21:16:49)

Refactored some of the code and removed some code duplication.
Fixes for #502 


Patch:
diff --git a/out.txt b/out.txt
deleted file mode 100644
index e69de29..0000000
diff --git a/requirements.txt b/requirements.txt
index 4441358..97b5e2e 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -9,3 +9,4 @@ oauth2client
 psq
 six
 urllib3[secure]
+docker
\ No newline at end of file
diff --git a/turbinia/client.py b/turbinia/client.py
index cc298dc..b282c67 100644
--- a/turbinia/client.py
+++ b/turbinia/client.py
@@ -27,7 +27,8 @@ from operator import attrgetter
 import os
 import stat
 import time
-import shutil
+import subprocess
+import codecs
 
 from turbinia import config
 from turbinia.config import logger
@@ -35,6 +36,7 @@ from turbinia.config import DATETIME_FORMAT
 from turbinia import task_manager
 from turbinia import TurbiniaException
 from turbinia.lib import text_formatter as fmt
+from turbinia.lib import docker_manager
 from turbinia.jobs import manager as job_manager
 from turbinia.workers import Priority
 from turbinia.workers.artifact import FileArtifactExtractionTask
@@ -113,41 +115,76 @@ def get_turbinia_client(run_local=False):
     raise TurbiniaException(msg)
 
 
-def check_dependencies(dependencies):
-  """Checks system dependencies.
+def check_docker_dependencies(dependencies):
+  """Checks docker dependencies.
 
   Args:
     dependencies(dict): dictionary of dependencies to check for.
 
   Raises:
-    TurbiniaException: if dependency is not met or a bad config file.
+    TurbiniaException: If dependency is not met.
   """
+  #TODO(wyassine): may run into issues down the line when a docker image
+  # does not have bash or which installed. (no linux fs layer).
+  log.info('Performing docker dependency check.')
+  job_names = list(job_manager.JobsManager.GetJobNames())
+  images = docker_manager.DockerManager().list_images(return_filter='short_id')
+
+  # Iterate through list of jobs
+  for job, values in dependencies.items():
+    if job not in job_names:
+      log.warning(
+          'The job {0:s} was not found or has been disabled. Skipping '
+          'dependency check...'.format(job))
+      continue
+    elif values.get('docker_image') in images:
+      for program in values['programs']:
+        cmd = 'type {0:s}'.format(program)
+        stdout, stderr, ret = docker_manager.ContainerManager(
+            values['docker_image']).execute_container(cmd, shell=True)
+        if ret != 0:
+          raise TurbiniaException(
+              'Job dependency {0:s} not found for job {1:s}. Please install '
+              'the dependency for the container or disable the job.'.format(
+                  program, job))
+      job_manager.JobsManager.RegisterDockerImage(job, values['docker_image'])
+    elif values.get('docker_image'):
+      raise TurbiniaException(
+          'Docker image {0:s} was not found for the job {1:s}. Please '
+          'update the config with the correct image id'.format(
+              values['docker_image'], job))
+
+
+def check_system_dependencies(dependencies):
+  """Checks system dependencies.
+
+  Args:
+    dependencies(dict): dictionary of dependencies to check for.
 
+  Raises:
+    TurbiniaException: If dependency is not met.
+  """
   log.info('Performing system dependency check.')
   job_names = list(job_manager.JobsManager.GetJobNames())
-  # Iterate through dependency config and perform a system check.
-  try:
-    for dep in dependencies:
-      if dep['job'].lower() in job_names:
-        for p in dep['programs']:
-          # Program can not be found.
-          if shutil.which(p) is None:
-            raise TurbiniaException(
-                'Job dependency {0:s} not found for job: {1:s}. Please install '
-                'the dependency or disable the job.'.format(p, dep['job']))
-      # If job is not found in list of registered jobs.
-      else:
-        log.warning(
-            'The job: {0:s} was not found or has been disabled. Skipping '
-            'dependency check...'.format(dep['job']))
-  except (KeyError, TypeError) as exception:
-    raise TurbiniaException(
-        'An issue has occured while parsing the '
-        'dependency config:{0!s} '.format(exception))
 
-  log.info(
-      'Dependency check complete. The following jobs will be active:'
-      'for this worker: {0:s}'.format(', '.join(job_names)))
+  # Iterate through list of jobs
+  for job, values in dependencies.items():
+    if job not in job_names:
+      log.warning(
+          'The job {0:s} was not found or has been disabled. Skipping '
+          'dependency check...'.format(job))
+      continue
+    elif not values.get('docker_image'):
+      for program in values['programs']:
+        cmd = 'type {0:s}'.format(program)
+        proc = subprocess.Popen(cmd, shell=True)
+        proc.communicate()
+        ret = proc.returncode
+        if ret != 0:
+          raise TurbiniaException(
+              'Job dependency {0:s} not found in $PATH for the job {1:s}. '
+              'Please install the dependency or disable the job.'.format(
+                  program, job))
 
 
 def check_directory(directory):
@@ -861,10 +898,18 @@ class TurbiniaCeleryWorker(BaseTurbiniaClient):
       job_manager.JobsManager.DeregisterJobs(jobs_blacklist=disabled_jobs)
 
     # Check for valid dependencies/directories.
-    check_dependencies(config.DEPENDENCIES)
+    dependencies = config.ParseDependencies()
+    if config.DOCKER_ENABLED:
+      check_docker_dependencies(dependencies)
+    check_system_dependencies(config.DEPENDENCIES)
     check_directory(config.MOUNT_DIR_PREFIX)
     check_directory(config.OUTPUT_DIR)
     check_directory(config.TMP_DIR)
+
+    jobs = job_manager.JobsManager.GetJobNames()
+    log.info(
+        'Dependency check complete. The following jobs will be enabled '
+        'for this worker: {0:s}'.format(','.join(jobs)))
     self.worker = self.task_manager.celery.app
 
   def start(self):
@@ -916,11 +961,18 @@ class TurbiniaPsqWorker(object):
       job_manager.JobsManager.DeregisterJobs(jobs_blacklist=disabled_jobs)
 
     # Check for valid dependencies/directories.
-    check_dependencies(config.DEPENDENCIES)
+    dependencies = config.ParseDependencies()
+    if config.DOCKER_ENABLED:
+      check_docker_dependencies(dependencies)
+    check_system_dependencies(dependencies)
     check_directory(config.MOUNT_DIR_PREFIX)
     check_directory(config.OUTPUT_DIR)
     check_directory(config.TMP_DIR)
 
+    jobs = job_manager.JobsManager.GetJobNames()
+    log.info(
+        'Dependency check complete. The following jobs will be enabled '
+        'for this worker: {0:s}'.format(','.join(jobs)))
     log.info('Starting PSQ listener on queue {0:s}'.format(self.psq.name))
     self.worker = psq.Worker(queue=self.psq)
 
diff --git a/turbinia/config/__init__.py b/turbinia/config/__init__.py
index acac734..59450af 100644
--- a/turbinia/config/__init__.py
+++ b/turbinia/config/__init__.py
@@ -187,3 +187,26 @@ def ValidateAndSetConfig(_config):
       setattr(sys.modules[__name__], var, None)
     else:
       setattr(sys.modules[__name__], var, getattr(_config, var))
+
+
+def ParseDependencies():
+  """Parses the config file DEPENDENCIES variable.
+
+  Raises:
+    TurbiniaException: If bad config file.
+
+  Returns:
+   dependencies(dict): The parsed dependency values.
+  """
+  dependencies = {}
+  try:
+    for values in CONFIG.DEPENDENCIES:
+      job = values['job'].lower()
+      dependencies[job] = {}
+      dependencies[job]['programs'] = values['programs']
+      dependencies[job]['docker_image'] = values.get('docker_image')
+  except (KeyError, TypeError) as exception:
+    raise TurbiniaException(
+        'An issue has occurred while parsing the '
+        'dependency config: {0!s}'.format(exception))
+  return dependencies
diff --git a/turbinia/jobs/interface.py b/turbinia/jobs/interface.py
index a2a29ba..83aa8ea 100644
--- a/turbinia/jobs/interface.py
+++ b/turbinia/jobs/interface.py
@@ -33,6 +33,7 @@ class TurbiniaJob(object):
     request_id (str): ID of the request
     priority (int): Job priority from 0-100, lowest == highest priority
     tasks (list): A list of Tasks generated by this Job
+    docker_image (str): Docker image to run for this job or None if unavailable.
     completed_task_count (int): The number of Tasks that have been removed.
     evidence (EvidenceCollection): The Evidence returned by Tasks from this Job.
   """
@@ -47,6 +48,7 @@ class TurbiniaJob(object):
     self.priority = 100
     self.request_id = request_id
     self.tasks = []
+    self.docker_image = None
     self.completed_task_count = 0
     self.evidence = EvidenceCollection()
     self.evidence.request_id = request_id
diff --git a/turbinia/jobs/manager.py b/turbinia/jobs/manager.py
index 1f86f63..607c446 100644
--- a/turbinia/jobs/manager.py
+++ b/turbinia/jobs/manager.py
@@ -225,3 +225,30 @@ class JobsManager(object):
     """
     for job_class in job_classes:
       cls.RegisterJob(job_class)
+
+  @classmethod
+  def RegisterDockerImage(cls, job_name, docker_image):
+    """Registers a Docker image for the job.
+
+    Args:
+      job_name(str): name of the job.
+      docker_image(str): name of the Docker image to be registered.
+    """
+    job_name = job_name.lower()
+    cls._job_classes[job_name].docker_image = docker_image
+
+  @classmethod
+  def GetDockerImage(cls, job_name):
+    """Retrieves the Docker image associated with the job.
+
+    Args:
+      job_name(str): name of the job.
+
+    Returns:
+      docker_image(str): The Docker image if available.
+    """
+    docker_image = None
+    job_class = cls._job_classes.get(job_name.lower())
+    if hasattr(job_class, 'docker_image') and job_class:
+      docker_image = job_class.docker_image
+    return docker_image
diff --git a/turbinia/lib/docker_manager.py b/turbinia/lib/docker_manager.py
new file mode 100644
index 0000000..ed47180
--- /dev/null
+++ b/turbinia/lib/docker_manager.py
@@ -0,0 +1,249 @@
+# -*- coding: utf-8 -*-
+# Copyright 2020 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.
+"""Library to handle docker related queries."""
+
+from __future__ import unicode_literals
+
+import logging
+import codecs
+import os
+import stat
+import docker
+
+from turbinia import TurbiniaException
+
+log = logging.getLogger('turbinia')
+
+
+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)
+
+
+class DockerManager(object):
+  """Class handling Docker management."""
+
+  def __init__(self):
+    """Initialize the Docker manager client."""
+    self.client = self._create_client()
+
+  def _create_client(self):
+    """Creates a Docker client.
+
+    Initializes a connection to the Docker daemon using
+    preconfigured Docker environment variables.
+
+    Returns:
+      docker_client(DockerClient): The Docker daemon client.
+
+    Raises:
+      TurbiniaException: If the Docker daemon could not be connected to.
+    """
+    try:
+      docker_client = docker.from_env()
+    except docker.errors.APIError as exception:
+      raise TurbiniaException(
+          'An issue has occurred connecting to the Docker daemon: {0!s}'.format(
+              exception))
+    return docker_client
+
+  def get_image(self, image_id):
+    """Retrieve the Docker Image object.
+
+    Args:
+      image_id(str): The short image id.
+
+    Returns:
+      image(Image): The Image object.
+
+    Raises:
+      TurbiniaException: If the Docker Image is not found.
+    """
+    try:
+      image = self.client.images.get(image_id)
+    except docker.errors.ImageNotFound as exception:
+      message = 'The Docker image {0!s} could not be found: {1!s}'
+      raise TurbiniaException(message.format(image_id, exception))
+    return image
+
+  def list_images(self, return_filter=None):
+    """Lists all available Docker images.
+
+    Args:
+     return_filter(str): If provided, will return a subset of the Images data.
+        Allowed values are 'short_id' and 'id'.
+
+    Returns:
+      list: containing:
+        Images: The Image objects
+        str: The Image ids if a filter was specified.
+
+    Raises:
+      TurbiniaException: If an error occurred retrieving the images.
+    """
+    accepted_vars = ['short_id', 'id']
+    try:
+      images = self.client.images.list()
+      if return_filter in accepted_vars:
+        # short_id and id will always start with sha256:
+        images = [
+            getattr(img, return_filter).replace('sha256:', '') for img in images
+        ]
+    except docker.errors.APIError as exception:
+      raise TurbiniaException(
+          'An error occurred retrieving the images: {0!s}'.format(exception))
+    return images
+
+
+class ContainerManager(DockerManager):
+  """Class representing Docker containers.
+
+  Attributes:
+    image_id(str): Docker image id.
+  """
+
+  def __init__(self, image_id):
+    """Initialize the ContainerManager object.
+
+    Args:
+      image_id(str): The image id to create a container from.
+    """
+    super(ContainerManager, self).__init__()
+    self.image = self.get_image(image_id)
+
+  def _create_mount_points(self, mount_paths, mode='rw'):
+    """Creates file and device mounting arguments.
+
+    The arguments will be passed into the container with the appropiate
+    mounting parameters. All device blocks will be mounted as read only,
+    regardless of the specified mode.
+
+    Attributes:
+      mount_paths(list): The paths on the host system to be mounted.
+      mode(str): The mode the path will be mounted in. The acceptable
+                 parameters are rw for read write and ro for read only.
+
+    Returns:
+      tuple: containing:
+        list: The device blocks that will be mounted.
+        dict: The file paths that will be mounted.
+
+    Raises:
+      TurbiniaException: If an incorrect mode was passed.
+    """
+    accepted_vars = ['rw', 'ro']
+    device_paths = []
+    file_paths = {}
+
+    if mode in accepted_vars:
+      for mpath in mount_paths:
+        device_mpath = '{0:s}:{0:s}:{1:s}'.format(mpath, 'r')
+        if mpath not in file_paths.keys() and device_mpath not in device_paths:
+          if IsBlockDevice(mpath):
+            device_paths.append(device_mpath)
+          else:
+            file_paths[mpath] = {'bind': mpath, 'mode': mode}
+    else:
+      raise TurbiniaException(
+          'An incorrect mode was passed: {0:s}. Unable to create the correct '
+          'mount points for the Docker container.'.format(mode))
+
+    return device_paths, file_paths
+
+  def execute_container(
+      self, cmd, shell=False, ro_paths=None, rw_paths=None, **kwargs):
+    """Executes a Docker container.
+
+    A new Docker container will be created from the image id,
+    executed, and then removed.
+
+    Attributes:
+      cmd(str|list): command to be executed.
+      shell (bool): Whether the cmd is in the form of a string or a list.
+      mount_paths(list): A list of paths to mount to the container.
+      **kwargs: Any additional keywords to pass to the container.
+
+    Returns:
+      stdout(str): stdout of the container.
+      stderr(str): stderr of the container.
+      ret(int): the return code of the process run.
+
+    Raises:
+      TurbiniaException: If an error occurred with the Docker container.
+    """
+    container = None
+    args = {}
+    stdout = ''
+
+    # Override the entrypoint to /bin/sh
+    kwargs['entrypoint'] = '/bin/sh'
+    if shell:
+      cmd = '-c ' + '\"{0:s}\"'.format(cmd)
+    else:
+      cmd = ' '.join(cmd)
+      cmd = '-c ' + '\"{0:s}\"'.format(cmd)
+
+    # Create the device and file mount paths
+    device_paths = []
+    file_paths = {}
+    if rw_paths:
+      dwpath, fwpath = self._create_mount_points(rw_paths)
+      device_paths.extend(dwpath)
+      file_paths.update(fwpath)
+    if ro_paths:
+      drpath, frpath = self._create_mount_points(ro_paths, mode='ro')
+      device_paths.extend(drpath)
+      file_paths.update(frpath)
+
+    args['devices'] = device_paths
+    args['volumes'] = file_paths
+
+    # Add any additional arguments
+    for key, value in kwargs.items():
+      args[key] = value
+
+    try:
+      container = self.client.containers.create(self.image, cmd, **args)
+      container.start()
+      # Stream program stdout from container
+      stdstream = container.logs(stream=True)
+      for stdo in stdstream:
+        stdo = codecs.decode(stdo, 'utf-8').strip()
+        log.debug(stdo)
+        stdout += stdo
+      results = container.wait()
+    except docker.errors.APIError as exception:
+      if container:
+        container.remove(v=True)
+      message = (
+          'An error has occurred with the container: {0!s}'.format(exception))
+      log.error(message)
+      raise TurbiniaException(message)
+
+    stderr, ret = results['Error'], results['StatusCode']
+    if container:
+      container.remove(v=True)
+
+    return stdout, stderr, ret
diff --git a/turbinia/processors/archive.py b/turbinia/processors/archive.py
index d7f142c..45b454a 100644
--- a/turbinia/processors/archive.py
+++ b/turbinia/processors/archive.py
@@ -78,7 +78,7 @@ def CompressDirectory(uncompressed_directory):
           'The tar file has been created and '
           'can be found at: {0:s}'.format(compressed_directory))
   except IOError as exception:
-    raise TurbiniaException('An error has occured: {0:s}'.format(exception))
+    raise TurbiniaException('An error has occurred: {0:s}'.format(exception))
   except tarfile.TarError as exception:
     raise TurbiniaException(
         'An error has while compressing the directory: {0:s}'.format(exception))
@@ -112,9 +112,9 @@ def UncompressTarFile(compressed_directory, output_tmp):
         'The tar file has been uncompressed to the following directory: {0:s}'
         .format(uncompressed_directory))
   except IOError as exception:
-    raise TurbiniaException('An error has occured: {0:s}'.format(exception))
+    raise TurbiniaException('An error has occurred: {0:s}'.format(exception))
   except tarfile.TarError as exception:
     raise TurbiniaException(
-        'An error has occured while uncompressing the tar '
+        'An error has occurred while uncompressing the tar '
         'file: {0:s}'.format(exception))
   return uncompressed_directory
diff --git a/turbinia/workers/__init__.py b/turbinia/workers/__init__.py
index d0406f3..51241aa 100644
--- a/turbinia/workers/__init__.py
+++ b/turbinia/workers/__init__.py
@@ -39,6 +39,7 @@ from turbinia.evidence import evidence_decode
 from turbinia import output_manager
 from turbinia import state_manager
 from turbinia import TurbiniaException
+from turbinia.lib import docker_manager
 
 log = logging.getLogger('turbinia')
 
@@ -454,19 +455,37 @@ class TurbiniaTask(object):
     Returns:
       Tuple of the return code, and the TurbiniaTaskResult object
     """
+    # Avoid circular dependency.
+    from turbinia.jobs import manager as job_manager
+
     save_files = save_files if save_files else []
     log_files = log_files if log_files else []
     new_evidence = new_evidence if new_evidence else []
     success_codes = success_codes if success_codes else [0]
 
-    if shell:
-      proc = subprocess.Popen(cmd, shell=True)
+    # Execute the job via docker.
+    docker_image = job_manager.JobsManager.GetDockerImage(self.job_name)
+    if docker_image:
+      ro_paths = [
+          result.input_evidence.local_path, result.input_evidence.source_path,
+          result.input_evidence.device_path, result.input_evidence.mount_path
+      ]
+      rw_paths = [self.output_dir, self.tmp_dir]
+      container_manager = docker_manager.ContainerManager(docker_image)
+      stdout, stderr, ret = container_manager.execute_container(
+          cmd, shell, ro_paths=ro_paths, rw_paths=rw_paths)
+
+    # Execute the job on the host system.
     else:
-      proc = subprocess.Popen(cmd)
-    stdout, stderr = proc.communicate()
+      if shell:
+        proc = subprocess.Popen(cmd, shell=True)
+      else:
+        proc = subprocess.Popen(cmd)
+      stdout, stderr = proc.communicate()
+      ret = proc.returncode
+
     result.error['stdout'] = stdout
     result.error['stderr'] = stderr
-    ret = proc.returncode
 
     for file_ in log_files:
       if not os.path.exists(file_):
@@ -541,10 +560,7 @@ class TurbiniaTask(object):
     self.output_manager.setup(self)
     self.tmp_dir, self.output_dir = self.output_manager.get_local_output_dirs()
     if not self.result:
-      self.result = TurbiniaTaskResult(
-          input_evidence=evidence, base_output_dir=self.base_output_dir,
-          request_id=self.request_id, job_id=self.job_id)
-      self.result.setup(self)
+      self.result = self.create_result()
 
     if not self.run_local:
       if evidence.copyable and not config.SHARED_FILESYSTEM:
@@ -560,6 +576,27 @@ class TurbiniaTask(object):
     """Updates the last_update time of the task."""
     self.last_update = datetime.now()
 
+  def create_result(self, status=None, message=None, trace=None):
+    """Creates a new TurbiniaTaskResults and instantiates the result.
+
+    Args:
+      status: A one line descriptive task status.
+      message: An error message to show when returning the result.
+      trace: Stack traceback for errors.
+    """
+    result = TurbiniaTaskResult(
+        base_output_dir=self.base_output_dir, request_id=self.request_id,
+        job_id=self.job_id)
+    result.setup(self)
+    if message:
+      if status:
+        result.status = '{0:s}. Previous status: [{1:s}]'.format(
+            message, status)
+      else:
+        result.status = message
+      result.set_error(message, trace)
+    return result
+
   def validate_result(self, result):
     """Checks to make sure that the result is valid.
 
@@ -576,38 +613,33 @@ class TurbiniaTask(object):
       The original result object if it is OK, otherwise an empty result object
       indicating a failure.
     """
-    bad_message = None
+    message = None
     check_status = 'Successful'
 
-    if not isinstance(result, TurbiniaTaskResult):
-      bad_message = (
-          'Task returned type [{0!s}] instead of TurbiniaTaskResult.').format(
-              type(result))
-    else:
+    if isinstance(result, TurbiniaTaskResult):
       try:
         log.debug('Checking TurbiniaTaskResult for serializability')
         pickle.dumps(result.serialize())
       except (TypeError, pickle.PicklingError) as exception:
-        bad_message = (
+        message = (
             'Error pickling TurbiniaTaskResult object. Returning a new result '
             'with the pickling error, and all previous result data will be '
             'lost. Pickle Error: {0!s}'.format(exception))
+    else:
+      message = (
+          'Task returned type [{0!s}] instead of TurbiniaTaskResult.').format(
+              type(result))
 
-    if bad_message:
-      log.error(bad_message)
+    if message:
+      log.error(message)
       if result and hasattr(result, 'status') and result.status:
-        old_status = result.status
+        status = result.status
       else:
-        old_status = 'No previous status'
-
-      result = TurbiniaTaskResult(
-          base_output_dir=self.base_output_dir, request_id=self.request_id,
-          job_id=self.job_id)
-      result.setup(self)
-      result.status = '{0:s}. Previous status: [{1:s}]'.format(
-          bad_message, old_status)
-      result.set_error(bad_message, traceback.format_exc())
-      result.close(self, success=False, status=bad_message)
+        status = 'No previous status'
+
+      result = self.create_result(
+          status=status, message=message, trace=traceback.format_exc())
+      result.close(self, success=False, status=message)
       check_status = 'Failed, but replaced with empty result'
 
     log.info('Result check: {0:s}'.format(check_status))
@@ -658,12 +690,8 @@ class TurbiniaTask(object):
           self.result.set_error(exception.__class__, traceback.format_exc())
         self.result.status = message
       else:
-        self.result = TurbiniaTaskResult(
-            base_output_dir=self.base_output_dir, request_id=self.request_id,
-            job_id=self.job_id)
-        self.result.setup(self)
-        self.result.status = message
-        self.result.set_error(message, traceback.format_exc())
+        self.result = self.create_result(
+            message=message, trace=traceback.format_exc())
       return self.result.serialize()
     with filelock.FileLock(config.LOCK_FILE):
       log.info('Starting Task {0:s} {1:s}'.format(self.name, self.id))


Test patch:
diff --git a/turbinia/client_test.py b/turbinia/client_test.py
index ebd2b2d..da8ee19 100644
--- a/turbinia/client_test.py
+++ b/turbinia/client_test.py
@@ -32,7 +32,8 @@ from turbinia import client as TurbiniaClientProvider
 from turbinia.client import TurbiniaServer
 from turbinia.client import TurbiniaStats
 from turbinia.client import TurbiniaPsqWorker
-from turbinia.client import check_dependencies
+from turbinia.client import check_system_dependencies
+from turbinia.client import check_docker_dependencies
 from turbinia import TurbiniaException
 
 SHORT_REPORT = textwrap.dedent(
@@ -441,7 +442,7 @@ class TestTurbiniaPsqWorker(unittest.TestCase):
     config.LoadConfig()
     config.OUTPUT_DIR = self.tmp_dir
     config.MOUNT_DIR_PREFIX = self.tmp_dir
-    config.DEPENDENCIES = []
+    config.ParseDependencies = mock.MagicMock(return_value={})
 
   def tearDown(self):
     if 'turbinia-test' in self.tmp_dir:
@@ -450,7 +451,8 @@ class TestTurbiniaPsqWorker(unittest.TestCase):
   @mock.patch('turbinia.client.pubsub')
   @mock.patch('turbinia.client.datastore.Client')
   @mock.patch('turbinia.client.psq.Worker')
-  def testTurbiniaPsqWorkerInit(self, _, __, ___):
+  @mock.patch('turbinia.lib.docker_manager.DockerManager')
+  def testTurbiniaPsqWorkerInit(self, _, __, ___, ____):
     """Basic test for PSQ worker."""
     worker = TurbiniaPsqWorker([], [])
     self.assertTrue(hasattr(worker, 'worker'))
@@ -458,7 +460,8 @@ class TestTurbiniaPsqWorker(unittest.TestCase):
   @mock.patch('turbinia.client.pubsub')
   @mock.patch('turbinia.client.datastore.Client')
   @mock.patch('turbinia.client.psq.Worker')
-  def testTurbiniaClientNoDir(self, _, __, ___):
+  @mock.patch('turbinia.lib.docker_manager.DockerManager')
+  def testTurbiniaClientNoDir(self, _, __, ___, ____):
     """Test that OUTPUT_DIR path is created."""
     config.OUTPUT_DIR = os.path.join(self.tmp_dir, 'no_such_dir')
     TurbiniaPsqWorker([], [])
@@ -467,36 +470,76 @@ class TestTurbiniaPsqWorker(unittest.TestCase):
   @mock.patch('turbinia.client.pubsub')
   @mock.patch('turbinia.client.datastore.Client')
   @mock.patch('turbinia.client.psq.Worker')
-  def testTurbiniaClientIsNonDir(self, _, __, ___):
+  @mock.patch('turbinia.lib.docker_manager.DockerManager')
+  def testTurbiniaClientIsNonDir(self, _, __, ___, ____):
     """Test that OUTPUT_DIR does not point to an existing non-directory."""
     config.OUTPUT_DIR = os.path.join(self.tmp_dir, 'empty_file')
     open(config.OUTPUT_DIR, 'a').close()
     self.assertRaises(TurbiniaException, TurbiniaPsqWorker)
 
-  @mock.patch('turbinia.client.shutil')
+  @mock.patch('turbinia.client.subprocess.Popen')
   @mock.patch('logging.Logger.warning')
-  def testDependencyCheck(self, mock_logger, mock_shutil):
+  def testSystemDependencyCheck(self, mock_logger, peopen_mock):
     """Test system dependency check."""
-    dependencies = [{
-        'job': 'PlasoJob',
-        'programs': ['non_exist'],
-        'docker_image': None
-    }]
-
+    dependencies = {
+        'plasojob': {
+            'programs': ['non_exist'],
+            'docker_image': None
+        }
+    }
     # Dependency not found.
-    mock_shutil.which.return_value = None
-    self.assertRaises(TurbiniaException, check_dependencies, dependencies)
+    proc_mock = mock.MagicMock()
+    proc_mock.returncode = 1
+    peopen_mock.return_value = proc_mock
+    self.assertRaises(
+        TurbiniaException, check_system_dependencies, dependencies)
 
     # Normal run.
-    mock_shutil.which.return_value = True
-    check_dependencies(dependencies)
+    proc_mock.returncode = 0
+    check_system_dependencies(dependencies)
 
     # Job not found.
-    dependencies[0]['job'] = 'non_exist'
-    check_dependencies(dependencies)
+    dependencies['non_exist'] = dependencies.pop('plasojob')
+    check_system_dependencies(dependencies)
     mock_logger.assert_called_with(
-        'The job: non_exist was not found or has been disabled. '
+        'The job non_exist was not found or has been disabled. '
         'Skipping dependency check...')
 
-    # Bad dependency config.
-    self.assertRaises(TurbiniaException, check_dependencies, [{'test': 'test'}])
+  @mock.patch('turbinia.lib.docker_manager.DockerManager')
+  @mock.patch('turbinia.lib.docker_manager.ContainerManager')
+  @mock.patch('logging.Logger.warning')
+  def testDockerDependencyCheck(
+      self, mock_logger, mock_contmgr, mock_dockermgr):
+    """Test Docker dependency check."""
+    dependencies = {
+        'plasojob': {
+            'programs': ['non_exist'],
+            'docker_image': 'test_img'
+        }
+    }
+
+    # Set up mock objects
+    mock_dm = mock_dockermgr.return_value
+    mock_dm.list_images.return_value = ['test_img']
+    mock_cm = mock_contmgr.return_value
+
+    # Dependency not found.
+    mock_cm.execute_container.return_value = ['non_exist', None, 1]
+    self.assertRaises(
+        TurbiniaException, check_docker_dependencies, dependencies)
+
+    # Normal run
+    mock_cm.execute_container.return_value = ['exists', None, 0]
+    check_docker_dependencies(dependencies)
+
+    # Docker image not found
+    mock_dm.list_images.return_value = ['non_exist']
+    self.assertRaises(
+        TurbiniaException, check_docker_dependencies, dependencies)
+
+    # Job not found.
+    dependencies['non_exist'] = dependencies.pop('plasojob')
+    check_docker_dependencies(dependencies)
+    mock_logger.assert_called_with(
+        'The job non_exist was not found or has been disabled. '
+        'Skipping dependency check...')
diff --git a/turbinia/config/turbinia_config_test.py b/turbinia/config/turbinia_config_test.py
index 51e1079..f5d94d3 100644
--- a/turbinia/config/turbinia_config_test.py
+++ b/turbinia/config/turbinia_config_test.py
@@ -154,3 +154,19 @@ class TestTurbiniaConfig(unittest.TestCase):
     self.assertEqual(config.TURBINIA_PROJECT, 'foo')
     self.assertEqual(config.TURBINIA_ZONE, 'bar')
     os.environ.pop(config.ENVCONFIGVAR)
+
+  def testParseDependencies(self):
+    """Tests a valid config for the ParseDependencies() method."""
+    smpl_depends = 'DEPENDENCIES = [{"job": "PlasoJob","programs": ["test"], "docker_image": "test"}]'
+    self.WriteConfig(smpl_depends)
+    config.LoadConfig()
+    smpl_out = {'plasojob': {'programs': ['test'], 'docker_image': 'test'}}
+    smpl_test = config.ParseDependencies()
+    self.assertEqual(smpl_out, smpl_test)
+
+  def testBadDependenciesConfig(self):
+    """Tests a bad config for the ParseDependencies() method."""
+    bad_config = 'DEPENDENCIES = [{"bad_config"}]'
+    self.WriteConfig(bad_config)
+    config.LoadConfig()
+    self.assertRaises(TurbiniaException, config.ParseDependencies)
diff --git a/turbinia/lib/dockermanager_test.py b/turbinia/lib/dockermanager_test.py
new file mode 100644
index 0000000..0875df1
--- /dev/null
+++ b/turbinia/lib/dockermanager_test.py
@@ -0,0 +1,186 @@
+# -*- coding: utf-8 -*-
+# Copyright 2020 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.
+"""Tests for docker_manager module."""
+
+from __future__ import unicode_literals
+
+import unittest
+import codecs
+import mock
+import docker
+
+from turbinia.lib import docker_manager
+from turbinia import TurbiniaException
+
+
+class MockImage(object):
+  """Mock class for a Docker image.
+
+  Attributes:
+    short_id(str): The short id of the image.
+    id(str): The full id of the image.
+  """
+
+  def __init__(self, id, short_id):
+    """Initialization of the MockImage class."""
+    self.id = 'sha256:{0:s}'.format(id)
+    self.short_id = 'sha256:{0:s}'.format(short_id)
+
+
+class MockContainer(object):
+  """Mock class for a Docker container.
+
+  Attributes:
+    s_start(str): sample output for start method.
+    s_logs(list): sample output for logs method.
+    s_wait(str): sample output for wait method.
+    s_remove(str): sample output for remove method.
+  """
+
+  def __init__(self, s_start=None, s_logs=None, s_wait=None, s_remove=None):
+    """Initialization of the MockContainer class."""
+    self.s_start = s_start
+    self.s_logs = s_logs
+    self.s_wait = s_wait
+    self.s_remove = s_remove
+
+  def start(self):
+    """Mock method of container.start()"""
+    return self.s_start
+
+  def logs(self, stream=None):
+    """Mock method of container.logs()"""
+    return self.s_logs
+
+  def wait(self):
+    """Mock method of container.wait()"""
+    return self.s_wait
+
+  def remove(self, v=None):
+    """Mock method of container.remove()"""
+    return self.s_remove
+
+
+class TestDockerManager(unittest.TestCase):
+  """Test DockerManager class."""
+
+  @mock.patch('turbinia.lib.docker_manager.docker')
+  def setUp(self, mock_docker):
+    mock_docker.from_env.return_value = docker.client.DockerClient
+    self.docker_mgr = docker_manager.DockerManager()
+
+  def testDockerManagerInit(self):
+    """Tests __init__ method of DockerManager."""
+    # Test that a DockerClient is being returned
+    assert self.docker_mgr.client == docker.client.DockerClient
+
+  def testVerifyImages(self):
+    """Tests DockerManager.get_image() method."""
+    # Test normal execution.
+    test_img = '1234'
+    self.docker_mgr.client = mock.MagicMock()
+    self.docker_mgr.client.images.get.return_value = test_img
+    assert test_img == self.docker_mgr.get_image(test_img)
+
+    # Ensure exception is being handled when image doesn't exist.
+    self.docker_mgr.client.images.get.side_effect = docker.errors.ImageNotFound(
+        'mock test fail.')
+    self.assertRaises(TurbiniaException, self.docker_mgr.get_image, test_img)
+
+  def testListImages(self):
+    """Tests DockerManager.list_images() method."""
+    id_smpl = ['123456', '234567']
+    sid_smpl = ['123', '234']
+    img_sample = [
+        MockImage(id_smpl[0], sid_smpl[0]),
+        MockImage(id_smpl[1], sid_smpl[1])
+    ]
+
+    # Ensure list is being returned.
+    self.docker_mgr.client = mock.MagicMock()
+    self.docker_mgr.client.images.list.return_value = img_sample
+    assert isinstance(self.docker_mgr.list_images(), list)
+
+    # Ensure values are properly being stripped off.
+    assert self.docker_mgr.list_images(return_filter='short_id') == sid_smpl
+    assert self.docker_mgr.list_images(return_filter='id') == id_smpl
+
+    # Ensure img object is being returned if incorrect filter is provided.
+    img = self.docker_mgr.list_images(return_filter='non_exist')
+    assert isinstance(img[0], MockImage)
+
+    # Ensure exception is being handled.
+    self.docker_mgr.client.images.list.side_effect = docker.errors.APIError(
+        'mock test fail.')
+    self.assertRaises(TurbiniaException, self.docker_mgr.list_images)
+
+
+class TestContainerManager(unittest.TestCase):
+  """Test ContainerManager class."""
+
+  @mock.patch('turbinia.lib.docker_manager.docker')
+  @mock.patch('turbinia.lib.docker_manager.DockerManager.get_image')
+  def setUp(self, mock_get, mock_docker):
+    self.test_img = '1234'
+    mock_docker.from_env.return_value = docker.client.DockerClient
+    mock_get.return_value = self.test_img
+    self.container_mgr = docker_manager.ContainerManager(self.test_img)
+
+  def testContainerManagerInit(self):
+    """Tests __init__ method of ContainerManager."""
+    # Ensure correct instance and image id
+    assert self.container_mgr.client == docker.client.DockerClient
+    assert self.container_mgr.image == self.test_img
+
+  @mock.patch('turbinia.lib.docker_manager.IsBlockDevice')
+  def testCreateMountPoints(self, mock_blockcheck):
+    """Tests ContainerManager._create_mount_points() method."""
+    # Ensure correct device formatting.
+    mock_blockcheck.return_value = True
+    device_smpl = ['/path/to/device']
+    device_formatted = ['{0:s}:{0:s}:r'.format(device_smpl[0])]
+    device_paths, _ = self.container_mgr._create_mount_points(device_smpl)
+    assert device_formatted == device_paths
+
+    # Ensure correct file path formmating.
+    mock_blockcheck.return_value = False
+    file_smpl = ['/path/to/file']
+    file_formatted = {}
+    file_formatted[file_smpl[0]] = {'bind': file_smpl[0], 'mode': 'rw'}
+    _, file_paths = self.container_mgr._create_mount_points(file_smpl)
+    assert file_formatted == file_paths
+
+  def testExecuteContainer(self):
+    """Tests ContainerManager.execute_container() method."""
+    # sample output
+    s_logs = [b'this', b'is', b'a', b'mock', b'test']
+    s_wait = {'Error': None, 'StatusCode': 0}
+    self.container_mgr.client = mock.MagicMock()
+    self.container_mgr.client.containers.create.return_value = MockContainer(
+        s_logs=s_logs, s_wait=s_wait)
+    stdout, stderr, ret = self.container_mgr.execute_container('cm', shell=True)
+
+    # Ensure correct output
+    out = [codecs.decode(w, 'utf-8') for w in s_logs]
+    assert stdout == ''.join(out)
+    assert stderr == s_wait['Error']
+    assert ret == s_wait['StatusCode']
+
+    # Ensure exception is being handled
+    exception = docker.errors.APIError('mock test fail.')
+    self.container_mgr.client.containers.create.side_effect = exception
+    self.assertRaises(
+        TurbiniaException, self.container_mgr.execute_container, 'cmd',
+        shell=True)
\ No newline at end of file