Repository: google/turbinia
Pull request #459 (2019-11-13 15:44:14)

Updating the libcloudforensics.py lib with the GCP labels management functionality and adding missing modules to dev requirements in setup.py
##########################Changes to libcloudforensics.py ##########################
the GCP labels management functionalities will be used to add incident_id labels to the GCE vm and disks created and to tie them all to an incident. This change will enable label management on all GCE resources that extend that class (VMs, Disks, Snapshots) and will do the following changes to: 
******** GoogleComputeBaseResource class in the libcloudforensics.py ********
Add: get_labels 
Description: get all the labels assigned to a certain resource
Add: add_labels 
Description: add/update labels to a certain resource
Add: form_operation
Description: helper method to allow creating different GCE API operation object (ex: insert, setLabel, etc..)
Add: get_resource_type
Description: helper method to get the compute resource type from the resource key-value store (ex: vm, disk, snapshot, image, etc ..)
******** GoogleCloudProject class in the libcloudforensics.py ********
Add: list_instance_by_label method  to GoogleCloudProject class:
Description: Lists VMs in a project with one of the provided labels.
This method list compute engine instances if they have one of the supplied label
Add: list_disk_by_label method  to GoogleCloudProject class:
Description: Lists Disks in a project with one of the provided labels.
This method list compute engine disks if they have one of the supplied label
Add: __list_by_label method  to GoogleCloudProject class:
Description: Private helper method to list_instance_by_label and list_disk_by_label methods.
**********************************************************************************************
Tests are done in a separate setup against GCP APIs. Can be further discussed if needed
********************************************************************************************** 
##########################Changes to setup.py ##########################
Adding missing modules "celery" and "coverage" to the dev setup so that run_tests.py doesn't fail in dev mode

googlebot: We found a Contributor License Agreement for you (the sender of this pull request), but were unable to find agreements for all the commit author(s) or Co-authors.  If you authored these, maybe you used a different email address in the git commits than was used to sign the CLA ([login here](https://cla.developers.google.com/) to double check)?  If these were authored by someone else, then they will need to sign a CLA as well, and confirm that they're okay with these being contributed to Google.
In order to pass this check, please resolve this problem and then comment `@googlebot I fixed it.`. If the bot doesn't comment, it means it doesn't think anything has changed.

ℹ️ **Googlers: [Go here](https://goto.google.com/prinfo/https%3A%2F%2Fgithub.com%2Fgoogle%2Fturbinia%2Fpull%2F459) for more info**.

<!-- need_author_cla -->
tomchop: I pushed one commit (https://github.com/google/turbinia/pull/459/commits/9ce60cf894a251650584f6ec52ee67fc5c284fe2) so make sure to pull before you push 💪 

Patch:
diff --git a/README.md b/README.md
index 58178ed..f755bb6 100644
--- a/README.md
+++ b/README.md
@@ -1,35 +1,65 @@
 # Turbinia
 
 ## Summary
-Turbinia is an open-source framework for deploying, managing, and running distributed forensic workloads.  It is intended to automate running of common forensic processing tools (i.e. Plaso, TSK, strings, etc) to help with processing evidence in the Cloud, scaling the processing of large amounts of evidence, and decreasing response time by parallelizing processing where possible.
+
+Turbinia is an open-source framework for deploying, managing, and running
+distributed forensic workloads. It is intended to automate running of common
+forensic processing tools (i.e. Plaso, TSK, strings, etc) to help with
+processing evidence in the Cloud, scaling the processing of large amounts of
+evidence, and decreasing response time by parallelizing processing where
+possible.
 
 <img src="docs/images/turbinia-logo.jpg?raw=true" width=240>
 
 ## How it works
-Turbinia is composed of different components for the client, server and the workers.  These components can be run in the Cloud, on local machines, or as a hybrid of both.  The Turbinia client makes requests to process evidence to the Turbinia server.  The Turbinia server creates logical jobs from these incoming user requests, which creates and schedules forensic processing tasks to be run by the workers.  The evidence to be processed will be split up by the jobs when possible, and many tasks can be created in order to process the evidence in parallel.  One or more workers run continuously to process tasks from the server.  Any new evidence created or discovered by the tasks will be fed back into Turbinia for further processing.
-
-Communication from the client to the server is currently done with either Google Cloud PubSub or [Kombu](https://github.com/celery/kombu) messaging.  The worker implementation can use either [PSQ](https://github.com/GoogleCloudPlatform/psq) (a Google Cloud PubSub Task Queue) or [Celery](http://www.celeryproject.org/) for task scheduling.
 
-More information on Turbinia and how it works can be [found here](docs/how-it-works.md).
+Turbinia is composed of different components for the client, server and the
+workers. These components can be run in the Cloud, on local machines, or as a
+hybrid of both. The Turbinia client makes requests to process evidence to the
+Turbinia server. The Turbinia server creates logical jobs from these incoming
+user requests, which creates and schedules forensic processing tasks to be run
+by the workers. The evidence to be processed will be split up by the jobs when
+possible, and many tasks can be created in order to process the evidence in
+parallel. One or more workers run continuously to process tasks from the server.
+Any new evidence created or discovered by the tasks will be fed back into
+Turbinia for further processing.
+
+Communication from the client to the server is currently done with either Google
+Cloud PubSub or [Kombu](https://github.com/celery/kombu) messaging. The worker
+implementation can use either [PSQ](https://github.com/GoogleCloudPlatform/psq)
+(a Google Cloud PubSub Task Queue) or [Celery](http://www.celeryproject.org/)
+for task scheduling.
+
+More information on Turbinia and how it works can be
+[found here](docs/how-it-works.md).
 
 ## Status
+
 Turbinia is currently in Alpha release.
 
 ## Installation
+
 There is an [rough installation guide here](docs/install.md).
 
 ## Usage
-The basic steps to get things running after the initial installation and configuration are:
-* Start Turbinia server component with ```turbiniactl server``` command
-* Start one or more Turbinia workers with ```turbiniactl psqworker```
-* Send evidence to be processed from the turbinia client with ```turbiniactl ${evidencetype}```
-* Check status of running tasks with ```turbiniactl status```
-
-turbiniactl can be used to start the different components, and here is the basic usage:
-``` 
+
+The basic steps to get things running after the initial installation and
+configuration are:
+
+*   Start Turbinia server component with `turbiniactl server` command
+*   Start one or more Turbinia workers with `turbiniactl psqworker`
+*   Send evidence to be processed from the turbinia client with `turbiniactl
+    ${evidencetype}`
+*   Check status of running tasks with `turbiniactl status`
+
+turbiniactl can be used to start the different components, and here is the basic
+usage:
+
+```
 $ turbiniactl --help
-usage: turbiniactl [-h] [-q] [-v] [-d] [-a] [-f] [-o OUTPUT_DIR] [-L LOG_FILE]
-                   [-r REQUEST_ID] [-R] [-S] [-C] [-V] [-D]
+usage: turbiniactl [-h] [-q] [-v] [-d] [-a] [-c CONFIG_FILE]
+                   [-C RECIPE_CONFIG] [-f] [-o OUTPUT_DIR] [-L LOG_FILE]
+                   [-r REQUEST_ID] [-R] [-S] [-V] [-D]
                    [-F FILTER_PATTERNS_FILE] [-j JOBS_WHITELIST]
                    [-J JOBS_BLACKLIST] [-p POLL_INTERVAL] [-t TASK] [-w]
                    <command> ...
@@ -40,6 +70,19 @@ optional arguments:
   -v, --verbose         Show verbose output
   -d, --debug           Show debug output
   -a, --all_fields      Show all task status fields in output
+  -c CONFIG_FILE, --config_file CONFIG_FILE
+                        Load explicit config file. If specified it will ignore
+                        config files in other default locations
+                        (/etc/turbinia.conf, ~/.turbiniarc, or in paths
+                        referenced in environment variable
+                        TURBINIA_CONFIG_PATH)
+  -C RECIPE_CONFIG, --recipe_config RECIPE_CONFIG
+                        Recipe configuration data passed in as comma separated
+                        key=value pairs (e.g. "-C
+                        key=value,otherkey=othervalue"). These will get passed
+                        to tasks as evidence config, and will also be written
+                        to the metadata.json file for Evidence types that
+                        write it
   -f, --force_evidence  Force evidence processing request in potentially
                         unsafe conditions
   -o OUTPUT_DIR, --output_dir OUTPUT_DIR
@@ -52,8 +95,6 @@ optional arguments:
                         infrastructure. This can be used to run one-off Tasks
                         to process data locally.
   -S, --server          Run Turbinia Server indefinitely
-  -C, --use_celery      Pass this flag when using Celery/Kombu for task
-                        queuing and messaging (instead of Google PSQ/pubsub)
   -V, --version         Show the version
   -D, --dump_json       Dump JSON output of Turbinia Request instead of
                         sending it
@@ -63,10 +104,18 @@ optional arguments:
                         grep regex format). This filtered output will be in
                         addition to the complete output
   -j JOBS_WHITELIST, --jobs_whitelist JOBS_WHITELIST
-                        A whitelist for Jobs that we will allow to run (note
-                        that it will not force them to run).
+                        A whitelist for Jobs that will be allowed to run (in
+                        CSV format, no spaces). This will not force them to
+                        run if they are not configured to. This is applied
+                        both at server start time and when the client makes a
+                        processing request. When applied at server start time
+                        the change is persistent while the server is running.
+                        When applied by the client, it will only affect that
+                        processing request.
   -J JOBS_BLACKLIST, --jobs_blacklist JOBS_BLACKLIST
-                        A blacklist for Jobs we will not allow to run
+                        A blacklist for Jobs we will not allow to run. See
+                        --jobs_whitelist help for details on format and when
+                        it is applied.
   -p POLL_INTERVAL, --poll_interval POLL_INTERVAL
                         Number of seconds to wait between polling for task
                         state info
@@ -88,9 +137,17 @@ Commands:
     celeryworker        Run Celery worker
     status              Get Turbinia Task status
     server              Run Turbinia Server
+    config              Prints out config file
+    testnotify          Sends test notification
 ```
 
-The commands for processing the evidence types of rawdisk and directory specify information about evidence that Turbinia should process. By default, when adding new evidence to be processed, turbiniactl will act as a client and send a request to the configured Turbinia server, otherwise if ```--server``` is specified, it will start up its own Turbinia server process.  Here's the turbiniactl usage for adding a raw disk type of evidence to be processed by Turbinia:
+The commands for processing evidence specify the metadata about that evidence
+for Turbinia to process. By default, when adding new evidence to be processed,
+turbiniactl will act as a client and send a request to the configured Turbinia
+server, otherwise if `server` is specified, it will start up its own Turbinia
+server process. Here's the turbiniactl usage for adding a raw disk type of
+evidence to be processed by Turbinia:
+
 ```
 $ ./turbiniactl rawdisk -h
 usage: turbiniactl rawdisk [-h] -l LOCAL_PATH [-s SOURCE] [-n NAME]
@@ -104,19 +161,55 @@ optional arguments:
   -n NAME, --name NAME  Descriptive name of the evidence
 ```
 
-## Other documentation
-* [Installation](docs/install.md)
-* [How it works](docs/how-it-works.md)
-* [Contributing to Turbinia](docs/contributing.md)
-* [Developing new Tasks](docs/developing-new-tasks.md)
-* [FAQ](docs/faq.md)
-* [Debugging and Common Errors](docs/debugging.md)
+Status information about the requests that are being or have been processed can
+be viewed with the `turbiniactl status` command.  You can specify the request ID
+that was generated, or other filters like the username of the requester, or how
+many days of processing history you want to view.  You can also generate
+statistics and reports (in markdown format) with other flags.
+
+```
+$ turbiniactl status -h
+usage: turbiniactl status [-h] [-c] [-C] [-d DAYS_HISTORY] [-f]
+                          [-r REQUEST_ID] [-p PRIORITY_FILTER] [-R] [-s]
+                          [-t TASK_ID] [-u USER]
+
+optional arguments:
+  -h, --help            show this help message and exit
+  -c, --close_tasks     Close tasks based on Request ID or Task ID
+  -C, --csv             When used with --statistics, the output will be in CSV
+                        format
+  -d DAYS_HISTORY, --days_history DAYS_HISTORY
+                        Number of days of history to show
+  -f, --force           Gatekeeper for --close_tasks
+  -r REQUEST_ID, --request_id REQUEST_ID
+                        Show tasks with this Request ID
+  -p PRIORITY_FILTER, --priority_filter PRIORITY_FILTER
+                        This sets what report sections are shown in full
+                        detail in report output. Any tasks that have set a
+                        report_priority value equal to or lower than this
+                        setting will be shown in full detail, and tasks with a
+                        higher value will only have a summary shown. To see
+                        all tasks report output in full detail, set
+                        --priority_filter=100
+  -R, --full_report     Generate full markdown report instead of just a
+                        summary
+  -s, --statistics      Generate statistics only
+  -t TASK_ID, --task_id TASK_ID
+                        Show task for given Task ID
+  -u USER, --user USER  Show task for given user
+```
 
+## Other documentation
 
-## Notes
-* Turbinia currently assumes that Evidence is equally available to all worker nodes (e.g. through locally mapped storage, or through attachable persistent Google Cloud Disks, etc).
-* Not all evidence types are supported yet
-* Still only a small number of processing job types supported, but more are being developed.
+*   [Installation](docs/install.md)
+*   [How it works](docs/how-it-works.md)
+*   [Operational Details](docs/operational-details.md)
+*   [Contributing to Turbinia](docs/contributing.md)
+*   [Developing new Tasks](docs/developing-new-tasks.md)
+*   [FAQ](docs/faq.md)
+*   [Debugging and Common Errors](docs/debugging.md)
 
 ##### Obligatory Fine Print
-This is not an official Google product (experimental or otherwise), it is just code that happens to be owned by Google.
+
+This is not an official Google product (experimental or otherwise), it is just
+code that happens to be owned by Google.
diff --git a/docs/how-it-works.md b/docs/how-it-works.md
index 790c111..f43c356 100644
--- a/docs/how-it-works.md
+++ b/docs/how-it-works.md
@@ -87,12 +87,14 @@ that is available to all Workers.
 
 ## Task Manager
 
-The Task Manager Acts as a broker between the clients and workers and handles
-management of Evidence, Jobs, Tasks and Workers.
+The Task Manager runs in the server and acts as a broker between the clients and
+workers and handles management of Evidence, Jobs, Tasks and Workers.  It will keep
+state on these objects while the processing request that generated them is still
+ongoing.
 
 ## Task Manager Flow
 
-Jobs are configured to run for specific Evidence types, and if the Task
+Jobs are configured to process specific Evidence types, and if the Task
 Manager sees a new piece of Evidence (either from a new Turbinia request, or
 because another Task generated a new piece of Evidence), and there is a Job
 that is configured to run for that type of Evidence, then the Task Mananger will
diff --git a/docs/operational-details.md b/docs/operational-details.md
new file mode 100644
index 0000000..0a5a1ff
--- /dev/null
+++ b/docs/operational-details.md
@@ -0,0 +1,48 @@
+# Operational Details
+
+## Google Cloud Platform (GCP) Processing Details
+
+Turbinia can read Evidence from either cloud Persistent Disks, or from other
+Evidence types saved as GCS objects. Turbinia can also write output to GCS. Note
+that you can operate multiple Turbinia instances within the same GCP Project as
+long as you're careful to make sure your config options (Pub/Sub
+topics/subscriptions, output paths, instance name, etc) don't overlap.
+
+### Persistent Disks
+
+Persistent disks are the default when processing disks that come from the Cloud.
+The account you run Turbinia as must have access to the persistent disks that
+you want to process, and those disks must also be in the same zone as the
+Turbinia workers. If you process GoogleCloudDisk Evidence with Turbinia, the
+worker node will attach the disk automatically before it runs its tasks. If you
+already have access to a persistent disk in a separate project, Turbinia can
+copy this into the project where Turbinia is being run.
+
+### Processing non-cloud disks in GCP
+
+If you have raw disk images from physical machines or elsewhere that you want to
+process in the cloud, the best option is to create a new Cloud Persistent Disk,
+and then copy the image into the filesystem of the new disk. Then you can use
+the GoogleCloudDiskRawEmbedded Evidence type.
+
+Another option is to
+[convert the raw image to a cloud image](https://cloud.google.com/compute/docs/images/import-existing-image),
+and then create a Persistent Disk from that and process it as the
+GoogleCloudDisk Evidence type, but this is not generally recommended as it
+requires zero-padding the disk to a GB boundary which can change the hash of the
+disk, and isn't considered forensically sound.
+
+One last option is to copy the image into GCS and process them directly from
+there, but the GoogleCloudDiskRawEmbeded option is generally recommended because
+this method requires setting up
+[GCS FUSE](https://cloud.google.com/storage/docs/gcs-fuse), and this is less
+stable than using Persistent Disks. If you do choose this option you will need
+to configure all of your worker nodes to mounting your GCS bucket at a common
+path. Once your GCS bucket is mounted, you can process these images as the
+'rawdisk' Evidence type.
+
+## General Notes
+
+*   Turbinia currently assumes that Evidence is equally available to all worker
+    nodes (e.g. through locally mapped storage, or through attachable persistent
+    Google Cloud Disks, etc).
diff --git a/history.txt b/history.txt
new file mode 100644
index 0000000..e69de29
diff --git a/setup.py b/setup.py
index c13a91e..9b9d814 100644
--- a/setup.py
+++ b/setup.py
@@ -69,7 +69,7 @@ setup(
         'requirements.txt', session=PipSession())
     ],
     extras_require={
-        'dev': ['mock', 'nose', 'yapf'],
+        'dev': ['mock', 'nose', 'yapf', 'celery~=4.1', 'coverage'],
         'local': ['celery~=4.1', 'kombu~=4.1', 'redis~=3.0'],
         'worker': ['plaso>=20171118', 'pyhindsight>=2.2.0']
     }
diff --git a/tools/turbinia_job_graph.py b/tools/turbinia_job_graph.py
index ab0c2e2..dc577b9 100755
--- a/tools/turbinia_job_graph.py
+++ b/tools/turbinia_job_graph.py
@@ -1,3 +1,4 @@
+#!/usr/bin/env python
 # -*- coding: utf-8 -*-
 # Copyright 2018 Google Inc.
 #
@@ -14,10 +15,12 @@
 # limitations under the License.
 """Graph to visualise job/evidence relationships."""
 
+from __future__ import print_function
 from __future__ import unicode_literals
 
 import argparse
 import graphviz
+import sys
 
 from turbinia.jobs import manager as jobs_manager
 
@@ -49,14 +52,43 @@ def create_graph():
 if __name__ == '__main__':
   parser = argparse.ArgumentParser(
       description='Create Turbinia evidence graph.')
+  parser.add_argument(
+      '-f', '--format', default='png',
+      help='The format of the output file you wish to generate.  Specify '
+      '"list" to list out the available output types.  More info is here: '
+      'http://www.graphviz.org/doc/info/output.html')
+  parser.add_argument(
+      '-e', '--engine', default='dot',
+      help='The graphviz engine used to generate the graph layout.  Specify '
+      '"list" to list out the available engines.')
   parser.add_argument('filename', type=unicode, help='where to save the file')
   args = parser.parse_args()
 
+  if args.format == 'list':
+    formats = ' '.join(graphviz.FORMATS)
+    print('Available format types: {0:s}'.format(formats))
+    sys.exit(0)
+
+  if args.format not in graphviz.FORMATS:
+    print('Format type {0:s} is not supported'.format(args.format))
+    sys.exit(1)
+
+  if args.engine == 'list':
+    engines = ' '.join(graphviz.ENGINES)
+    print('Available graph layout engines: {0:s}'.format(engines))
+    sys.exit(0)
+
+  if args.engine not in graphviz.ENGINES:
+    print('Layout engine type {0:s} is not supported'.format(args.engine))
+    sys.exit(1)
+
   graph = create_graph()
+  graph.engine = args.engine
   output_file = args.filename.replace('.png', '')
 
   try:
-    rendered_graph = graph.render(filename=output_file, cleanup=True)
+    rendered_graph = graph.render(
+        filename=output_file, format=args.format, cleanup=True)
     print('Graph generated and saved to: {0}'.format(rendered_graph))
   except graphviz.ExecutableNotFound:
     print('Graphviz is not installed - Run: apt-get install graphviz')
diff --git a/turbinia/client.py b/turbinia/client.py
index 8c58e15..e5a3796 100644
--- a/turbinia/client.py
+++ b/turbinia/client.py
@@ -19,6 +19,7 @@ from __future__ import unicode_literals
 from collections import defaultdict
 from datetime import datetime
 from datetime import timedelta
+
 import json
 import logging
 from operator import itemgetter
@@ -26,6 +27,7 @@ from operator import attrgetter
 import os
 import stat
 import time
+import shutil
 
 from turbinia import config
 from turbinia.config import logger
@@ -33,6 +35,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.jobs import manager as job_manager
 from turbinia.workers import Priority
 from turbinia.workers.artifact import FileArtifactExtractionTask
 from turbinia.workers.analysis.wordpress import WordpressAccessLogAnalysisTask
@@ -90,6 +93,43 @@ log = logging.getLogger('turbinia')
 logger.setup()
 
 
+def check_dependencies(dependencies):
+  """Checks system dependencies.
+
+  Args:
+    dependencies(dict): dictionary of dependencies to check for.
+
+  Raises:
+    TurbiniaException: if dependency is not met or a bad config file.
+  """
+
+  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)))
+
+
 def check_directory(directory):
   """Checks directory to make sure it exists and is writable.
 
@@ -230,7 +270,7 @@ class TurbiniaClient(object):
     # the task manager from the client.
     log.info('Available Jobs:')
     for job in self.task_manager.jobs:
-      log.info('\t{0:s}'.format(job.name))
+      log.info('\t{0:s}'.format(job.NAME))
 
   def wait_for_request(
       self, instance, project, region, request_id=None, user=None,
@@ -284,7 +324,7 @@ class TurbiniaClient(object):
     log.info('All {0:d} Tasks completed'.format(len(task_results)))
 
   def get_task_data(
-      self, instance, project, region, days=0, task_id=None, request_id=None,
+      self, instance, project, region, days=1000, task_id=None, request_id=None,
       user=None, function_name='gettasks'):
     """Gets task data from Google Cloud Functions.
 
@@ -569,7 +609,7 @@ class TurbiniaClient(object):
     return '\n'.join(report)
 
   def format_task_status(
-      self, instance, project, region, days=0, task_id=None, request_id=None,
+      self, instance, project, region, days=1000, task_id=None, request_id=None,
       user=None, all_fields=False, full_report=False,
       priority_filter=Priority.HIGH):
     """Formats the recent history for Turbinia Tasks.
@@ -783,9 +823,25 @@ class TurbiniaCeleryWorker(TurbiniaClient):
     worker (celery.app): Celery worker app
   """
 
-  def __init__(self, *_, **__):
-    """Initialization for Celery worker."""
+  def __init__(self, jobs_blacklist=None, jobs_whitelist=None):
+    """Initialization for celery worker.
+
+    Args:
+      jobs_blacklist (Optional[list[str]]): Jobs we will exclude from running
+      jobs_whitelist (Optional[list[str]]): The only Jobs we will include to run
+    """
     super(TurbiniaCeleryWorker, self).__init__()
+    # Deregister jobs from blacklist/whitelist.
+    disabled_jobs = list(config.DISABLED_JOBS) if config.DISABLED_JOBS else []
+    job_manager.JobsManager.DeregisterJobs(jobs_blacklist, jobs_whitelist)
+    if disabled_jobs:
+      log.info(
+          'Disabling jobs that were configured to be disabled in the '
+          'config file: {0:s}'.format(', '.join(disabled_jobs)))
+      job_manager.JobsManager.DeregisterJobs(jobs_blacklist=disabled_jobs)
+
+    # Check for valid dependencies/directories.
+    check_dependencies(config.DEPENDENCIES)
     check_directory(config.MOUNT_DIR_PREFIX)
     check_directory(config.OUTPUT_DIR)
     check_directory(config.TMP_DIR)
@@ -810,8 +866,13 @@ class TurbiniaPsqWorker(object):
     TurbiniaException: When errors occur
   """
 
-  def __init__(self, *_, **__):
-    """Initialization for PSQ Worker."""
+  def __init__(self, jobs_blacklist=None, jobs_whitelist=None):
+    """Initialization for PSQ Worker.
+
+    Args:
+      jobs_blacklist (Optional[list[str]]): Jobs we will exclude from running
+      jobs_whitelist (Optional[list[str]]): The only Jobs we will include to run
+    """
     config.LoadConfig()
     psq_publisher = pubsub.PublisherClient()
     psq_subscriber = pubsub.SubscriberClient()
@@ -825,6 +886,17 @@ class TurbiniaPsqWorker(object):
       log.error(msg)
       raise TurbiniaException(msg)
 
+    # Deregister jobs from blacklist/whitelist.
+    disabled_jobs = list(config.DISABLED_JOBS) if config.DISABLED_JOBS else []
+    job_manager.JobsManager.DeregisterJobs(jobs_blacklist, jobs_whitelist)
+    if disabled_jobs:
+      log.info(
+          'Disabling jobs that were configured to be disabled in the '
+          'config file: {0:s}'.format(', '.join(disabled_jobs)))
+      job_manager.JobsManager.DeregisterJobs(jobs_blacklist=disabled_jobs)
+
+    # Check for valid dependencies/directories.
+    check_dependencies(config.DEPENDENCIES)
     check_directory(config.MOUNT_DIR_PREFIX)
     check_directory(config.OUTPUT_DIR)
     check_directory(config.TMP_DIR)
diff --git a/turbinia/config/__init__.py b/turbinia/config/__init__.py
index 3f0457e..acac734 100644
--- a/turbinia/config/__init__.py
+++ b/turbinia/config/__init__.py
@@ -55,6 +55,9 @@ REQUIRED_VARS = [
     'SHARED_FILESYSTEM',
     # TODO(aarontp): Move this to the recipe config when it's available.
     'DEBUG_TASKS',
+    'DEPENDENCIES',
+    'DOCKER_ENABLED',
+    'DISABLED_JOBS',
 ]
 
 # Optional config vars.  Some may be mandatory depending on the configuration
@@ -79,6 +82,12 @@ OPTIONAL_VARS = [
     'KOMBU_BROKER',
     'KOMBU_CHANNEL',
     'KOMBU_DURABLE',
+    # Email config
+    'EMAIL_NOTIFICATIONS',
+    'EMAIL_HOST_ADDRESS',
+    'EMAIL_PORT',
+    'EMAIL_ADDRESS',
+    'EMAIL_PASSWORD',
 ]
 
 # Environment variable to look for path data in
@@ -157,13 +166,24 @@ def ValidateAndSetConfig(_config):
 
   CONFIGVARS = REQUIRED_VARS + OPTIONAL_VARS
   for var in CONFIGVARS:
+    empty_value = False
     if not hasattr(_config, var):
-      raise TurbiniaException(
-          'No config attribute {0:s}:{1:s}'.format(_config.configSource, var))
+      if var in OPTIONAL_VARS:
+        log.debug(
+            'Setting non-existent but optional config variable {0:s} to '
+            'None'.format(var))
+        empty_value = True
+      else:
+        raise TurbiniaException(
+            'Required config attribute {0:s}:{1:s} not in config'.format(
+                _config.configSource, var))
     if var in REQUIRED_VARS and getattr(_config, var) is None:
       raise TurbiniaException(
           'Config attribute {0:s}:{1:s} is not set'.format(
               _config.configSource, var))
 
     # Set the attribute in the current module
-    setattr(sys.modules[__name__], var, getattr(_config, var))
+    if empty_value:
+      setattr(sys.modules[__name__], var, None)
+    else:
+      setattr(sys.modules[__name__], var, getattr(_config, var))
diff --git a/turbinia/config/turbinia_config_tmpl.py b/turbinia/config/turbinia_config_tmpl.py
index 2ea8426..b431f7d 100644
--- a/turbinia/config/turbinia_config_tmpl.py
+++ b/turbinia/config/turbinia_config_tmpl.py
@@ -73,6 +73,70 @@ SHARED_FILESYSTEM = False
 # problems.
 DEBUG_TASKS = False
 
+################################################################################
+#                         External Dependency Configurations
+#
+# Options in this section are used to configure system and docker dependencies.
+################################################################################
+
+# This will allow for the configuration of system dependencies and docker
+# containers. The following configuration will need to be set per job config.
+# Please use the example below as a guide to add additional job checks.
+# {
+#   'job': 'MyJob'
+#   'programs': ['list', 'of', 'required', 'programs']
+#   'docker_image': 'ImageID' # Or None if no image configured for this job.
+# }
+
+# This will enable the usage of docker containers for the worker.
+DOCKER_ENABLED = False
+
+# Any jobs added to this list will disable it from being used.
+DISABLED_JOBS = []
+
+# Configure additional job dependency checks below.
+DEPENDENCIES = [{
+    'job': 'BinaryExtractorJob',
+    'programs': ['image_export.py'],
+    'docker_image': None
+}, {
+    'job': 'BulkExtractorJob',
+    'programs': ['bulk_extractor'],
+    'docker_image': None
+}, {
+    'job': 'GrepJob',
+    'programs': ['grep'],
+    'docker_image': None
+}, {
+    'job': 'HadoopAnalysisJob',
+    'programs': ['strings'],
+    'docker_image': None
+}, {
+    'job': 'HindsightJob',
+    'programs': ['hindsight.py'],
+    'docker_image': None
+}, {
+    'job': 'JenkinsAnalysisJob',
+    'programs': ['john'],
+    'docker_image': None
+}, {
+    'job': 'PlasoJob',
+    'programs': ['log2timeline.py'],
+    'docker_image': None
+}, {
+    'job': 'PsortJob',
+    'programs': ['psort.py'],
+    'docker_image': None
+}, {
+    'job': 'StringsJob',
+    'programs': ['strings'],
+    'docker_image': None
+}, {
+    'job': 'VolatilityJob',
+    'programs': ['vol.py'],
+    'docker_image': None
+}]
+
 ################################################################################
 #                        Google Cloud Platform (GCP)
 #
@@ -131,3 +195,20 @@ KOMBU_DURABLE = True
 REDIS_HOST = 'localhost'
 REDIS_PORT = '6379'
 REDIS_DB = '0'
+
+################################################################################
+#                           Email Config
+#
+# These options are required if you wish to use email notifications
+################################################################################
+
+# Will emails notifcations be enabled
+EMAIL_NOTIFICATIONS = False
+
+# Host Address and port number(TLS)
+EMAIL_HOST_ADDRESS = 'example.address.com'
+EMAIL_PORT = 587
+
+# Email address and password
+EMAIL_ADDRESS = 'example@address.com'
+EMAIL_PASSWORD = 'Hunter2'
diff --git a/turbinia/evidence.py b/turbinia/evidence.py
index 6afc740..977d85b 100644
--- a/turbinia/evidence.py
+++ b/turbinia/evidence.py
@@ -540,7 +540,7 @@ class PlasoFile(Evidence):
     self.save_metadata = True
 
 
-class PlasoCsvFile(PlasoFile):
+class PlasoCsvFile(Evidence):
   """Psort output file evidence.  """
 
   def __init__(self, plaso_version=None, *args, **kwargs):
diff --git a/turbinia/jobs/manager.py b/turbinia/jobs/manager.py
index 53fd6b1..1f86f63 100644
--- a/turbinia/jobs/manager.py
+++ b/turbinia/jobs/manager.py
@@ -93,6 +93,39 @@ class JobsManager(object):
 
     del cls._job_classes[job_name]
 
+  @classmethod
+  def DeregisterJobs(cls, jobs_blacklist=None, jobs_whitelist=None):
+    """Deregisters a list of job names against white/black lists.
+
+    jobs_whitelist and jobs_blacklist must not be specified at the same time.
+
+    Args:
+      jobs_blacklist (Optional[list[str]]): Job names to deregister.
+      jobs_whitelist (Optional[list[str]]): Job names to register.
+
+    Raises:
+      TurbiniaException if both jobs_blacklist and jobs_whitelist are specified.
+    """
+    registered_jobs = list(cls.GetJobNames())
+    jobs_remove = []
+
+    # Create a list of jobs to deregister.
+    if jobs_whitelist and jobs_blacklist:
+      raise TurbiniaException(
+          'jobs_whitelist and jobs_blacklist cannot be specified at the same '
+          'time.')
+    elif jobs_whitelist:
+      jobs_whitelist = [j.lower() for j in jobs_whitelist]
+      jobs_remove = [j for j in registered_jobs if j not in jobs_whitelist]
+    elif jobs_blacklist:
+      jobs_blacklist = [j.lower() for j in jobs_blacklist]
+      jobs_remove = [j for j in jobs_blacklist if j in registered_jobs]
+
+    # Deregister the jobs.
+    jobs_remove = [j.lower() for j in jobs_remove]
+    for job_name in jobs_remove:
+      del cls._job_classes[job_name]
+
   @classmethod
   def GetJobInstance(cls, job_name):
     """Retrieves an instance of a specific job.
diff --git a/turbinia/lib/libcloudforensics.py b/turbinia/lib/libcloudforensics.py
index 7ec539f..16f1974 100644
--- a/turbinia/lib/libcloudforensics.py
+++ b/turbinia/lib/libcloudforensics.py
@@ -437,6 +437,99 @@ class GoogleCloudProject(object):
     created = True
     return instance, created
 
+  def list_instance_by_labels(self, labels_filter, filter_union=True):
+    """List VMs in a project with one/all of the provided labels.
+
+    This will call the __list_by_label on instances() API object
+    with the proper labels filter.
+
+    Args:
+      labels_filter: A dict of labels to find e.g. {'id': '123'}.
+      filter_union: A Boolean; True to get the union of all filters,
+          False to get the intersection.
+
+    Returns:
+      A dictionary with name and metadata (zone, labels) for each instance.
+      ex: {'instance-1': {'zone': 'us-central1-a', 'labels': {'id': '123'}}
+    """
+
+    instance_service_object = self.gce_api().instances()
+    return self.__list_by_label(
+        labels_filter, instance_service_object, filter_union)
+
+  def list_disk_by_labels(self, labels_filter, filter_union=True):
+    """List Disks in a project with one/all of the provided labels.
+
+    This will call the __list_by_label on disks() API object
+    with the proper labels filter.
+
+    Args:
+      labels_filter: A dict of labels to find e.g. {'id': '123'}.
+      filter_union: A Boolean; True to get the union of all filters,
+          False to get the intersection.
+
+    Returns:
+      A dictionary with name and metadata (zone, labels) for each disk.
+      ex: {'disk-1': {'zone': 'us-central1-a', 'labels': {'id': '123'}}
+    """
+
+    disk_service_object = self.gce_api().disks()
+    return self.__list_by_label(
+        labels_filter, disk_service_object, filter_union)
+
+  def __list_by_label(self, labels_filter, service_object, filter_union):
+    """List Disks/VMs in a project with one/all of the provided labels.
+
+    Private method used to select different compute resources by labels.
+
+    Args:
+      labels_filter:  A dict of labels to find e.g. {'id': '123'}.
+      service_object: Google Compute Engine (Disk | Instance) service object.
+      filter_union: A boolean; True to get the union of all filters,
+          False to get the intersection.
+
+    Returns:
+      Dictionary with name and metadata (zone, labels) for each instance/disk.
+      ex: {'instance-1': {'zone': 'us-central1-a', 'labels': {'id': '123'}}
+    """
+    if not isinstance(filter_union, bool):
+      raise RuntimeError((
+          'filter_union parameter must be of Type boolean'
+          ' {0:s} is an invalid argument.').format(filter_union))
+
+    resource_dict = dict()
+    filter_expression = ''
+    operation = 'AND' if filter_union else 'OR'
+    for key, value in labels_filter.items():
+      filter_expression += 'labels.{0:s}={1:s} {2:s} '.format(
+          key, value, operation)
+    filter_expression = filter_expression[:-(len(operation) + 1)]
+
+    request = service_object.aggregatedList(
+        project=self.project_id, filter=filter_expression)
+    while request is not None:
+      response = request.execute()
+      result = self.gce_operation(response, zone=self.default_zone)
+
+      for item in result['items'].items():
+        region_or_zone_string, resource_scoped_list = item
+
+        if 'warning' not in resource_scoped_list.keys():
+          _, zone = region_or_zone_string.rsplit('/', 1)
+          # Only one of the following loops will execute since the method is
+          # called either with a service object Instances or Disks
+          for resource in resource_scoped_list.get('instances', []):
+            resource_dict[resource['name']] = dict(
+                zone=zone, labels=resource['labels'])
+
+          for resource in resource_scoped_list.get('disks', []):
+            resource_dict[resource['name']] = dict(
+                zone=zone, labels=resource['labels'])
+
+      request = service_object.aggregatedList_next(
+          previous_request=request, previous_response=response)
+    return resource_dict
+
 
 class GoogleCloudFunction(GoogleCloudProject):
   """Class to call Google Cloud Functions.
@@ -535,15 +628,12 @@ class GoogleComputeBaseResource(object):
     """Get specific value from the resource key value store.
 
     Args:
-      key: Key to get value from.
+      key: A key of type String to get key's corresponding value.
 
     Returns:
       Value of key or None if key is missing.
     """
-    if not self._data:
-      operation = self.get_operation().execute()
-      self._data = self.project.gce_operation(
-          operation, zone=self.zone, block=False)
+    self._data = self.get_operation().execute()
     return self._data.get(key)
 
   def get_source_string(self):
@@ -552,8 +642,115 @@ class GoogleComputeBaseResource(object):
     Returns:
       The full API URL to the resource.
     """
+    if self._data:
+      return self._data['selfLink']
     return self.get_value('selfLink')
 
+  def get_resource_type(self):
+    """Get the resource type from the resource key-value store.
+
+    Returns:
+      Resource Type which is a string with one of the following values:
+        compute#instance
+        compute#disk
+        compute#snapshot
+    """
+    if self._data:
+      return self._data['kind']
+    return self.get_value('kind')
+
+  def form_operation(self, operation_name):
+    """Form an API operation object for the compute resource.
+
+    Example:[RESOURCE].form_operation('setLabels')(**kwargs)
+    [RESOURCE] can be type "instance", disk or "snapshot".
+
+    Args:
+      operation_name: The name of the API operation you need to perform.
+
+    Returns:
+      An API operation object for the referenced compute resource.
+
+    Raises RuntimeError:
+      If resource type is not defined as a type which extends the
+      GoogleGomputeBaseResource class.
+    """
+    resource_type = self.get_resource_type()
+    module = None
+    if resource_type not in ['compute#instance', 'compute#snapshot',
+                             'compute#disk']:
+      raise RuntimeError((
+          'Compute resource Type {0:s} is not one of the defined types in '
+          'libcloudforensics library (Instance, Disk or Snapshot) '
+      ).format(resource_type))
+    if resource_type == 'compute#instance':
+      module = self.project.gce_api().instances()
+    elif resource_type == 'compute#disk':
+      module = self.project.gce_api().disks()
+    elif resource_type == 'compute#snapshot':
+      module = self.project.gce_api().snapshots()
+
+    operation_func_to_call = getattr(module, operation_name)
+    return operation_func_to_call
+
+  def get_labels(self):
+    """Get all labels of a compute resource.
+
+    Returns:
+      A dictionary of all labels.
+    """
+
+    operation = self.get_operation().execute()
+
+    return operation.get('labels')
+
+  def add_labels(self, new_labels_dict, blocking_call=False):
+    """Add or update labels of a compute resource.
+
+    Args:
+      new_labels_dict: A dictionary containing the labels to be added.
+          ex: {"incident_id": "1234abcd"}.
+      blocking_call: A boolean to decide whether the API call should
+          be blocking or not, default is False.
+
+    Returns:
+      The response of the API operation.
+    """
+
+    get_operation = self.get_operation().execute()
+    labelFingerprint = get_operation['labelFingerprint']
+
+    exisitng_labels_dict = dict()
+    if self.get_labels() is not None:
+      exisitng_labels_dict = self.get_labels()
+    exisitng_labels_dict.update(new_labels_dict)
+    labels_dict = exisitng_labels_dict
+    request_body = {'labels': labels_dict, 'labelFingerprint': labelFingerprint}
+
+    resource_type = self.get_resource_type()
+    operation = None
+    if resource_type not in ['compute#instance', 'compute#snapshot',
+                             'compute#disk']:
+      raise RuntimeError((
+          'Compute resource Type {0:s} is not one of the defined types in '
+          'libcloudforensics library (Instance, Disk or Snapshot) '
+      ).format(resource_type))
+    if resource_type == 'compute#instance':
+      operation = self.form_operation('setLabels')(
+          instance=self.name, project=self.project.project_id, zone=self.zone,
+          body=request_body).execute()
+    elif resource_type == 'compute#disk':
+      operation = self.form_operation('setLabels')(
+          resource=self.name, project=self.project.project_id, zone=self.zone,
+          body=request_body).execute()
+    elif resource_type == 'compute#snapshot':
+      operation = self.form_operation('setLabels')(
+          resource=self.name, project=self.project.project_id,
+          body=request_body).execute()
+
+    return self.project.gce_operation(
+        operation, zone=self.zone, block=blocking_call)
+
 
 class GoogleComputeInstance(GoogleComputeBaseResource):
   """Class representing a Google Compute Engine virtual machine."""
diff --git a/turbinia/notify.py b/turbinia/notify.py
new file mode 100644
index 0000000..5f86908
--- /dev/null
+++ b/turbinia/notify.py
@@ -0,0 +1,87 @@
+# -*- coding: utf-8 -*-
+# Copyright 2017 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.
+"""Sends email notifications"""
+
+import logging
+import smtplib
+from email.mime.multipart import MIMEMultipart
+from email.mime.text import MIMEText
+from turbinia import config
+
+log = logging.getLogger('turbinia')
+
+
+#Sends notfications via email
+def sendmail(address, subject, message):
+  """Sends an Email notification via SMTP
+     Args:
+         address: This is the address that the email will be sent to
+         subject: This will be the subject of the email
+         message: This will be the message of the email
+  """
+  try:
+    if config.EMAIL_NOTIFICATIONS is True:
+
+      #Puts message in MIME format
+      msg = MIMEMultipart()
+      msg['From'] = config.EMAIL_ADDRESS
+      msg['To'] = address
+      msg['Subject'] = subject
+      msg.attach(MIMEText(message))
+
+      #start SMTP
+      server = smtplib.SMTP(config.EMAIL_HOST_ADDRESS, config.EMAIL_PORT)
+
+      # identify to mail server
+      server.ehlo()
+
+      # secure email using TLS
+      server.starttls()
+
+      # reidentify as using TLS
+      server.ehlo()
+
+      #Attempt to login
+      if config.EMAIL_PASSWORD != '':
+        server.login(config.EMAIL_ADDRESS, config.EMAIL_PASSWORD)
+
+      else:
+        log.info(
+            'Email password is blank, '
+            'attempting to continue without logging in')
+
+      server.sendmail(config.EMAIL_ADDRESS, address, msg.as_string())
+      log.info('Email notification sent to ' + address)
+    else:
+      log.info('Email notifications are disabled')
+
+  except smtplib.SMTPException as e:
+    log.error(e)
+    log.error(
+        'Email failed to send, SMTP has raised an error,'
+        ' this likely means that there is a problem with the config')
+  except TypeError:
+    log.error('Email failed to send, there is likely a problem with the config')
+  except NameError:
+    log.error(
+        'Email failed to send, A value which is required'
+        ' for email notifications is not defined in the config')
+
+  finally:
+    #Terminate connection to server if active
+    try:
+      server.quit()
+    except UnboundLocalError:
+      pass
diff --git a/turbinia/output_manager.py b/turbinia/output_manager.py
index d6b8d17..6e39f1b 100644
--- a/turbinia/output_manager.py
+++ b/turbinia/output_manager.py
@@ -170,7 +170,7 @@ class OutputManager(object):
     evidence_.saved_path = path
     evidence_.saved_path_type = path_type
     log.info(
-        'Saved copyable evidence data to {0:s}'.format(evidence_.saved_path))
+        'Saved copyable evidence data to {0!s}'.format(evidence_.saved_path))
     return evidence_
 
   def save_local_file(self, file_, result):
diff --git a/turbinia/task_manager.py b/turbinia/task_manager.py
index 7eddb60..24d65d3 100644
--- a/turbinia/task_manager.py
+++ b/turbinia/task_manager.py
@@ -297,24 +297,23 @@ class BaseTaskManager(object):
     Args:
       task: An instantiated Turbinia Task
       evidence_: An Evidence object to be processed.
-
-    Raises:
-      TurbiniaException: If the request_id cannot be found.
     """
     if evidence_.request_id:
       task.request_id = evidence_.request_id
     elif job and job.request_id:
       task.request_id = job.request_id
     else:
-      raise TurbiniaException(
-          'Request ID not found in Evidence {0!s} or Task {1!s}'.format(
-              evidence_, task))
+      log.error(
+          'Request ID not found in Evidence {0!s} or Task {1!s}. Not adding '
+          'new Task because of undefined state'.format(evidence_, task))
+      return
 
     evidence_.config = job.evidence.config
     task.base_output_dir = config.OUTPUT_DIR
     task.requester = evidence_.config.get('requester')
     if job:
       task.job_id = job.id
+      task.job_name = job.name
       job.tasks.append(task)
     self.state_manager.write_new_task(task)
     self.enqueue_task(task, evidence_)
diff --git a/turbinia/turbiniactl.py b/turbinia/turbiniactl.py
index 2a9d385..29faaa6 100644
--- a/turbinia/turbiniactl.py
+++ b/turbinia/turbiniactl.py
@@ -136,6 +136,10 @@ def main():
   parser_config.add_argument(
       '-f', '--file_only', action='store_true', help='Print out file path only')
 
+  #Sends Test Notification
+  parser_testnotify = subparsers.add_parser(
+      'testnotify', help='Sends test notification')
+
   # TODO(aarontp): Find better way to specify these that allows for multiple
   # pieces of evidence to be submitted. Maybe automagically create different
   # commands based on introspection of evidence objects?
@@ -331,7 +335,7 @@ def main():
       help='When used with --statistics, the output will be in CSV format',
       required=False)
   parser_status.add_argument(
-      '-d', '--days_history', default=0, type=int,
+      '-d', '--days_history', default=1000, type=int,
       help='Number of days of history to show', required=False)
   parser_status.add_argument(
       '-f', '--force', help='Gatekeeper for --close_tasks', action='store_true',
@@ -399,6 +403,7 @@ def main():
   # config is loaded by these modules at load time, and we want to wait to load
   # the config until after we parse the args so that we can use those arguments
   # to point to config paths.
+  from turbinia import notify
   from turbinia.client import TurbiniaClient
   from turbinia.client import TurbiniaCeleryClient
   from turbinia.client import TurbiniaServer
@@ -422,6 +427,12 @@ def main():
           "Failed to read config file {0:s}: {1!s}".format(
               config.configSource, exception))
       sys.exit(1)
+  #sends test notification
+  if args.command == 'testnotify':
+    notify.sendmail(
+        config.EMAIL_ADDRESS, 'Turbinia test notification',
+        'This is a test notification')
+    sys.exit(0)
 
   if args.jobs_whitelist and args.jobs_blacklist:
     log.error(
@@ -565,11 +576,13 @@ def main():
     # Set up root logger level which is normally set by the psqworker command
     # which we are bypassing.
     logger.setup()
-    worker = TurbiniaPsqWorker()
+    worker = TurbiniaPsqWorker(
+        jobs_blacklist=args.jobs_blacklist, jobs_whitelist=args.jobs_whitelist)
     worker.start()
   elif args.command == 'celeryworker':
     logger.setup()
-    worker = TurbiniaCeleryWorker()
+    worker = TurbiniaCeleryWorker(
+        jobs_blacklist=args.jobs_blacklist, jobs_whitelist=args.jobs_whitelist)
     worker.start()
   elif args.command == 'server':
     server = TurbiniaServer(
@@ -620,7 +633,7 @@ def main():
     log.info('Available Jobs:')
     client.list_jobs()
   else:
-    log.warning('Command {0:s} not implemented.'.format(args.command))
+    log.warning('Command {0!s} not implemented.'.format(args.command))
 
   if evidence_ and not args.force_evidence:
     if config.SHARED_FILESYSTEM and evidence_.cloud_only:
diff --git a/turbinia/workers/__init__.py b/turbinia/workers/__init__.py
index 68188b1..d5ea9db 100644
--- a/turbinia/workers/__init__.py
+++ b/turbinia/workers/__init__.py
@@ -174,7 +174,7 @@ class TurbiniaTaskResult(object):
           task.output_manager.save_evidence(evidence, self)
       else:
         self.log(
-            'Evidence {0:s} has empty or missing file at source_path {1:s} so '
+            'Evidence {0!s} has empty or missing file at source_path {1!s} so '
             'not saving.'.format(evidence.name, evidence.source_path))
 
       if not evidence.request_id:
@@ -306,6 +306,7 @@ class TurbiniaTask(object):
       id (str): Unique Id of task (string of hex)
       is_finalize_task (bool): Whether this is a finalize Task or not.
       job_id (str): Job ID the Task was created by.
+      job_name (str): The name of the Job.
       last_update (datetime): A datetime object with the last time the task was
           updated.
       name (str): Name of task
@@ -342,6 +343,7 @@ class TurbiniaTask(object):
     self.id = uuid.uuid4().hex
     self.is_finalize_task = False
     self.job_id = None
+    self.job_name = None
     self.last_update = datetime.now()
     self.name = name if name else self.__class__.__name__
     self.output_dir = None
@@ -597,6 +599,9 @@ class TurbiniaTask(object):
     Returns:
       A TurbiniaTaskResult object
     """
+    # Avoid circular dependency.
+    from turbinia.jobs import manager as job_manager
+
     log.debug('Task {0:s} {1:s} awaiting execution'.format(self.name, self.id))
     evidence = evidence_decode(evidence)
     with filelock.FileLock(config.LOCK_FILE):
@@ -607,13 +612,26 @@ class TurbiniaTask(object):
         original_result_id = self.result.id
         evidence.validate()
 
+        # TODO(wyassine): refactor it so the result task does not
+        # have to go through the preprocess stage. At the moment
+        # self.results.setup is required to be called to set its status.
+        # Check if Task's job is available for the worker.
+        active_jobs = list(job_manager.JobsManager.GetJobNames())
+        if self.job_name.lower() not in active_jobs:
+          message = (
+              'Task will not run due to the job: {0:s} being disabled '
+              'on the worker.'.format(self.job_name))
+          self.result.log(message, level=logging.ERROR)
+          self.result.status = message
+          return self.result.serialize()
+
         if self.turbinia_version != turbinia.__version__:
           message = (
               'Worker and Server versions do not match: {0:s} != {1:s}'.format(
                   self.turbinia_version, turbinia.__version__))
           self.result.log(message, level=logging.ERROR)
           self.result.status = message
-          return self.result
+          return self.result.serialize()
 
         self._evidence_config = evidence.config
         self.result = self.run(evidence, self.result)


Test patch:
diff --git a/turbinia/client_test.py b/turbinia/client_test.py
index 24c9a64..b27626f 100644
--- a/turbinia/client_test.py
+++ b/turbinia/client_test.py
@@ -32,6 +32,7 @@ from turbinia.client import TurbiniaClient
 from turbinia.client import TurbiniaServer
 from turbinia.client import TurbiniaStats
 from turbinia.client import TurbiniaPsqWorker
+from turbinia.client import check_dependencies
 from turbinia import TurbiniaException
 
 SHORT_REPORT = textwrap.dedent(
@@ -440,6 +441,7 @@ class TestTurbiniaPsqWorker(unittest.TestCase):
     config.LoadConfig()
     config.OUTPUT_DIR = self.tmp_dir
     config.MOUNT_DIR_PREFIX = self.tmp_dir
+    config.DEPENDENCIES = []
 
   def tearDown(self):
     if 'turbinia-test' in self.tmp_dir:
@@ -450,7 +452,7 @@ class TestTurbiniaPsqWorker(unittest.TestCase):
   @mock.patch('turbinia.client.psq.Worker')
   def testTurbiniaPsqWorkerInit(self, _, __, ___):
     """Basic test for PSQ worker."""
-    worker = TurbiniaPsqWorker()
+    worker = TurbiniaPsqWorker([], [])
     self.assertTrue(hasattr(worker, 'worker'))
 
   @mock.patch('turbinia.client.pubsub')
@@ -459,7 +461,7 @@ class TestTurbiniaPsqWorker(unittest.TestCase):
   def testTurbiniaClientNoDir(self, _, __, ___):
     """Test that OUTPUT_DIR path is created."""
     config.OUTPUT_DIR = os.path.join(self.tmp_dir, 'no_such_dir')
-    TurbiniaPsqWorker()
+    TurbiniaPsqWorker([], [])
     self.assertTrue(os.path.exists(config.OUTPUT_DIR))
 
   @mock.patch('turbinia.client.pubsub')
@@ -470,3 +472,31 @@ class TestTurbiniaPsqWorker(unittest.TestCase):
     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('logging.Logger.warning')
+  def testDependencyCheck(self, mock_logger, mock_shutil):
+    """Test system dependency check."""
+    dependencies = [{
+        'job': 'PlasoJob',
+        'programs': ['non_exist'],
+        'docker_image': None
+    }]
+
+    # Dependency not found.
+    mock_shutil.which.return_value = None
+    self.assertRaises(TurbiniaException, check_dependencies, dependencies)
+
+    # Normal run.
+    mock_shutil.which.return_value = True
+    check_dependencies(dependencies)
+
+    # Job not found.
+    dependencies[0]['job'] = 'non_exist'
+    check_dependencies(dependencies)
+    mock_logger.assert_called_with(
+        'The job: non_exist was not found or has been disabled. '
+        'Skipping dependency check...')
+
+    # Bad dependency config.
+    self.assertRaises(TurbiniaException, check_dependencies, [{'test': 'test'}])
diff --git a/turbinia/config/turbinia_config_test.py b/turbinia/config/turbinia_config_test.py
index 2a63456..51e1079 100644
--- a/turbinia/config/turbinia_config_test.py
+++ b/turbinia/config/turbinia_config_test.py
@@ -110,6 +110,15 @@ class TestTurbiniaConfig(unittest.TestCase):
     self.assertEqual(config.UNSETKEY, None)
     self.assertEqual(config.SETKEY, 'bar')
 
+  def testNonexistentOptionalKeyConfig(self):
+    """Test that optional vars don't need to exist."""
+    self.WriteConfig('SETKEY = "bar"\n')
+    config.REQUIRED_VARS = ['SETKEY']
+    config.OPTIONAL_VARS = ['UNSETKEY']
+    config.LoadConfig()
+    self.assertEqual(config.UNSETKEY, None)
+    self.assertEqual(config.SETKEY, 'bar')
+
   def testMissingConfigPath(self):
     """Test non-existent config path."""
     config.CONFIGPATH = ['DOESNOTEXIST']
diff --git a/turbinia/workers/workers_test.py b/turbinia/workers/workers_test.py
index b3cb350..53551d6 100644
--- a/turbinia/workers/workers_test.py
+++ b/turbinia/workers/workers_test.py
@@ -55,6 +55,7 @@ class TestTurbiniaTaskBase(unittest.TestCase):
     self.plaso_task.output_manager.get_local_output_dirs.return_value = (
         None, None)
     self.task = self.task_class(base_output_dir=self.base_output_dir)
+    self.task.job_name = 'PlasoJob'
     self.task.output_manager = mock.MagicMock()
     self.task.output_manager.get_local_output_dirs.return_value = (None, None)
 
@@ -150,6 +151,17 @@ class TestTurbiniaTask(TestTurbiniaTaskBase):
     self.assertEqual(type(new_result), TurbiniaTaskResult)
     self.assertIn('CheckedResult', new_result.status)
 
+  def testTurbiniaTaskJobUnavailable(self):
+    """Test that the run wrapper can fail if the job doesn't exist."""
+    self.setResults()
+    self.task.job_name = 'non_exist'
+    canary_status = (
+        'Task will not run due to the job: '
+        'non_exist being disabled on the worker.')
+    new_result = self.task.run_wrapper(self.evidence.__dict__)
+    new_result = TurbiniaTaskResult.deserialize(new_result)
+    self.assertEqual(new_result.status, canary_status)
+
   def testTurbiniaTaskRunWrapperExceptionThrown(self):
     """Test that the run wrapper recovers from run throwing an exception."""
     self.setResults()