Repository: google/turbinia
Pull request #281 (2018-10-26 23:11:59)

Add serverless mode for independent Task execution
This will allow us to execute one-off Tasks without any Turbinia or Cloud infrastructure.  This can be used to make the Tasks more reusable outside of the framework or just for testing.  This is what it looks like:

```
$ turbiniactl -f --run_local --task PlasoTask rawdisk -l ~/tmp/turbinia/images/test-small.img
[INFO] Creating request c954257c28c7481ab1a784a4e5efdaae with evidence /home/aaronpeterson/tmp/turbinia/images/test-small.img
[DEBUG] --run_local specified so not sending request to server
[DEBUG] Looking up Task PlasoTask by name
[INFO] Running Task PlasoTask locally
[INFO] Starting Task PlasoTask 3e1c1a7d08c040269a56d6fcddd077de
[INFO] Creating new directory /var/tmp/1540595266-3e1c1a7d08c040269a56d6fcddd077de-PlasoTask
[INFO] Running plaso as [log2timeline.py --status_view none --hashers all --partition all --vss_stores all --logfile /var/tmp/1540595266-3e1c1a7d08c040269a56d6fcddd077de-PlasoTask/3e1c1a7d08c040269a56d6fcddd077de.log /var/tmp/1540595266-3e1c1a7d08c040269a56d6fcddd077de-Pl
asoTask/3e1c1a7d08c040269a56d6fcddd077de.plaso /usr/local/google/home/aaronpeterson/tmp/turbinia/images/test-small.img]
Checking availability and versions of dependencies.
[OPTIONAL]      missing: lzma.
[OK]

Source path     : /home/aaronpeterson/tmp/turbinia/images/test-small.img
Source type     : storage media image
Processing time : 00:00:00

Processing started.
Processing completed.

[INFO] Output file at /var/tmp/1540595266-3e1c1a7d08c040269a56d6fcddd077de-PlasoTask/3e1c1a7d08c040269a56d6fcddd077de.log
[INFO] Completed successfully in 0:00:11.005304 on aaronpeterson
[DEBUG] Result close successful. Status is [Completed successfully in 0:00:11.005304 on aaronpeterson]
[DEBUG] Checking TurbiniaTaskResult for serializability
[INFO] Result check: Successful
[DEBUG] Returning original result object 0c7cf1ac454e413eaf7c25f3087e4506 after task execution
[INFO] Task execution result: {'_log': [u'Running plaso as [log2timeline.py --status_view none --hashers all --partition all --vss_stores all --logfile /var/tmp/1540595266-3e1c1a7d08c040269a56d6fcddd077de-PlasoTask/3e1c1a7d08c040269a56d6fcddd077de.log /var/tmp/1540595266-
3e1c1a7d08c040269a56d6fcddd077de-PlasoTask/3e1c1a7d08c040269a56d6fcddd077de.plaso /home/aaronpeterson/tmp/turbinia/images/test-small.img]',
          u'Output file at /var/tmp/1540595266-3e1c1a7d08c040269a56d6fcddd077de-PlasoTask/3e1c1a7d08c040269a56d6fcddd077de.log',
          u'Completed successfully in 0:00:11.005304 on aaronpeterson'],
 'base_output_dir': u'/var/tmp',
 'closed': True,
 'error': {u'stderr': None, u'stdout': None},
 'evidence': [PlasoFile:PlasoFile:/var/tmp/1540595266-3e1c1a7d08c040269a56d6fcddd077de-PlasoTask/3e1c1a7d08c040269a56d6fcddd077de.plaso],
 'id': '0c7cf1ac454e413eaf7c25f3087e4506',
 'input_evidence': [RawDisk:/home/aaronpeterson/tmp/turbinia/images/test-small.img:/home/aaronpeterson/tmp/turbinia/images/test-small.img],
 'output_dir': u'/var/tmp/1540595266-3e1c1a7d08c040269a56d6fcddd077de-PlasoTask',
 'request_id': 'c954257c28c7481ab1a784a4e5efdaae',
 'run_time': datetime.timedelta(0, 11, 5304),
 'saved_paths': [u'/var/tmp/1540595266-3e1c1a7d08c040269a56d6fcddd077de-PlasoTask/3e1c1a7d08c040269a56d6fcddd077de.plaso'],
 'start_time': datetime.datetime(2018, 10, 26, 16, 7, 47, 344438),
 'status': u'Completed successfully in 0:00:11.005304 on aaronpeterson.svl.corp.google.com',
 'successful': True,
 'task_id': '3e1c1a7d08c040269a56d6fcddd077de',
 'task_name': 'PlasoTask',
 'user': 'aaronpeterson',
 'worker_name': 'aaronpeterson'}
[INFO] Done.
```


Patch:
diff --git a/README.md b/README.md
index 98306f8..06de463 100644
--- a/README.md
+++ b/README.md
@@ -29,9 +29,9 @@ turbiniactl can be used to start the different components, and here is the basic
 ``` 
 $ turbiniactl --help
 usage: turbiniactl [-h] [-q] [-v] [-d] [-a] [-f] [-o OUTPUT_DIR] [-L LOG_FILE]
-                   [-r REQUEST_ID] [-S] [-C] [-V] [-D]
+                   [-r REQUEST_ID] [-R] [-S] [-C] [-V] [-D]
                    [-F FILTER_PATTERNS_FILE] [-j JOBS_WHITELIST]
-                   [-J JOBS_BLACKLIST] [-p POLL_INTERVAL] [-w]
+                   [-J JOBS_BLACKLIST] [-p POLL_INTERVAL] [-t TASK] [-w]
                    <command> ...
 
 optional arguments:
@@ -48,6 +48,9 @@ optional arguments:
                         Log file
   -r REQUEST_ID, --request_id REQUEST_ID
                         Create new requests with this Request ID
+  -R, --run_local       Run completely locally without any server or other
+                        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)
@@ -67,6 +70,8 @@ optional arguments:
   -p POLL_INTERVAL, --poll_interval POLL_INTERVAL
                         Number of seconds to wait between polling for task
                         state info
+  -t TASK, --task TASK  The name of a single Task to run locally (must be used
+                        with --run_local.
   -w, --wait            Wait to exit until all tasks for the given request
                         have completed
 
@@ -82,7 +87,7 @@ Commands:
     psqworker           Run PSQ worker
     celeryworker        Run Celery worker
     status              Get Turbinia Task status
-    server              Run Turbinia Server    
+    server              Run Turbinia Server
 ```
 
 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:
diff --git a/requirements.txt b/requirements.txt
index b4b29d8..0aa3216 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,5 +1,6 @@
 filelock
 google-api-core>=1.2.1,<=1.4.0
+google-cloud-datastore==1.7.0
 google-api-python-client>=1.6.2
 google-cloud-pubsub==0.35.4
 google-cloud-storage
diff --git a/turbinia/client.py b/turbinia/client.py
index 67113ba..39f93b2 100644
--- a/turbinia/client.py
+++ b/turbinia/client.py
@@ -27,7 +27,39 @@ import time
 from turbinia import config
 from turbinia.config import logger
 from turbinia import task_manager
+from turbinia import workers
 from turbinia import TurbiniaException
+from turbinia.workers.artifact import FileArtifactExtractionTask
+from turbinia.workers.analysis.wordpress import WordpressAccessLogAnalysisTask
+from turbinia.workers.analysis.jenkins import JenkinsAnalysisTask
+from turbinia.workers.grep import GrepTask
+from turbinia.workers.hadoop import HadoopAnalysisTask
+from turbinia.workers.plaso import PlasoTask
+from turbinia.workers.psort import PsortTask
+from turbinia.workers.sshd import SSHDAnalysisTask
+from turbinia.workers.strings import StringsAsciiTask
+from turbinia.workers.strings import StringsUnicodeTask
+from turbinia.workers.tomcat import TomcatAnalysisTask
+from turbinia.workers.worker_stat import StatTask
+
+
+# TODO(aarontp): Remove this map after
+# https://github.com/google/turbinia/issues/278 is fixed.
+TASK_MAP = {
+    'fileartifactextractiontask': FileArtifactExtractionTask,
+    'wordpressaccessloganalysistask': WordpressAccessLogAnalysisTask,
+    'jenkinsanalysistask': JenkinsAnalysisTask,
+    'greptask': GrepTask,
+    'hadoopanalysistask': HadoopAnalysisTask,
+    'plasotask': PlasoTask,
+    'psorttask': PsortTask,
+    'sshdanalysistask': SSHDAnalysisTask,
+    'stringsasciitask': StringsAsciiTask,
+    'stringsunicodetask': StringsUnicodeTask,
+    'tomcatanalysistask': TomcatAnalysisTask,
+    'stattask': StatTask,
+}
+
 
 config.LoadConfig()
 if config.TASK_MANAGER == 'PSQ':
@@ -81,10 +113,32 @@ class TurbiniaClient(object):
     task_manager (TaskManager): Turbinia task manager
   """
 
-  def __init__(self):
+  def __init__(self, run_local=False):
     config.LoadConfig()
-    self.task_manager = task_manager.get_task_manager()
-    self.task_manager.setup(server=False)
+    if run_local:
+      self.task_manager = None
+    else:
+      self.task_manager = task_manager.get_task_manager()
+      self.task_manager.setup(server=False)
+
+  def create_task(self, task_name):
+    """Creates a Turbinia Task by name.
+
+    Args:
+      task_name(string): Name of the Task we are going to run.
+
+    Returns:
+      TurbiniaTask: An instantiated Task object.
+
+    Raises:
+      TurbiniaException: When no Task object matching task_name is found.
+    """
+    task_obj = TASK_MAP.get(task_name.lower())
+    log.debug('Looking up Task {0:s} by name'.format(task_name))
+    if not task_obj:
+      raise TurbiniaException('No Task named {0:s} found'.format(task_name))
+    return task_obj()
+
 
   def list_jobs(self):
     """List the available jobs."""
@@ -255,6 +309,26 @@ class TurbiniaClient(object):
 
     return '\n'.join(results)
 
+  def run_local_task(self, task_name, request):
+    """Runs a Turbinia Task locally.
+
+    Args:
+      task_name(string): Name of the Task we are going to run.
+      request (TurbiniaRequest): Object containing request and evidence info.
+
+    Returns:
+      TurbiniaTaskResult: The result returned by the Task Execution.
+    """
+    task = self.create_task(task_name)
+    task.request_id = request.request_id
+    task.base_output_dir = config.OUTPUT_DIR
+    task.run_local = True
+    if not request.evidence:
+      raise TurbiniaException('TurbiniaRequest does not contain evidence.')
+    log.info('Running Task {0:s} locally'.format(task_name))
+    result = task.run_wrapper(request.evidence[0])
+    return result
+
   def send_request(self, request):
     """Sends a TurbiniaRequest message.
 
@@ -380,6 +454,7 @@ class TurbiniaCeleryWorker(TurbiniaClient):
     super(TurbiniaCeleryWorker, self).__init__()
     check_directory(config.MOUNT_DIR_PREFIX)
     check_directory(config.OUTPUT_DIR)
+    check_directory(config.TMP_DIR)
     self.worker = self.task_manager.celery.app
 
   def start(self):
@@ -420,6 +495,7 @@ class TurbiniaPsqWorker(object):
 
     check_directory(config.MOUNT_DIR_PREFIX)
     check_directory(config.OUTPUT_DIR)
+    check_directory(config.TMP_DIR)
 
     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 f5cd54a..f696aed 100644
--- a/turbinia/config/__init__.py
+++ b/turbinia/config/__init__.py
@@ -39,6 +39,7 @@ CONFIGVARS = [
     'LOG_FILE',
     'LOCK_FILE',
     'OUTPUT_DIR',
+    'TMP_DIR',
     'SLEEP_TIME',
     'SINGLE_RUN',
     'MOUNT_DIR_PREFIX',
@@ -85,12 +86,16 @@ def LoadConfig():
   if CONFIG:
     return CONFIG
 
+  # If the environment variable is set, take precedence over the pre-defined
+  # CONFIGPATHs.
+  configpath = CONFIGPATH
   if ENVCONFIGVAR in os.environ:
-    CONFIGPATH.extend(os.environ[ENVCONFIGVAR].split(':'))
+    configpath = os.environ[ENVCONFIGVAR].split(':')
+
 
   config_file = None
   # Load first file found
-  for _dir, _file in itertools.product(CONFIGPATH, CONFIGFILES):
+  for _dir, _file in itertools.product(configpath, CONFIGFILES):
     if os.path.exists(os.path.join(_dir, _file)):
       config_file = os.path.join(_dir, _file)
       break
diff --git a/turbinia/config/turbinia_config.py b/turbinia/config/turbinia_config.py
index 8f52ee5..81238a7 100644
--- a/turbinia/config/turbinia_config.py
+++ b/turbinia/config/turbinia_config.py
@@ -23,16 +23,22 @@ ROLE = 'server'
 # Which user account Turbinia runs as
 USER = 'turbinia'
 
-# Turbinia's installation directory
-TURBINIA_DIR = '/opt/turbinia'
-
 # 'PSQ' is currently the only valid option as
 # a distributed task queue using Google Cloud Pub/Sub
 TASK_MANAGER = 'PSQ'
 
+# Turbinia's installation directory
+TURBINIA_DIR = '/opt/turbinia'
+
 # Default base output directory for worker results and evidence.
 OUTPUT_DIR = '/var/tmp'
 
+# Directory for temporary files.  Some temporary files can be quite large (e.g.
+# Plaso files can easily be multiple gigabytes), so make sure there is enough
+# space.  Nothing from this directory will be saved.  This directory should be
+# different from the OUTPUT_DIR.
+TMP_DIR = '/tmp'
+
 # File to log to; set this as None if log file is not desired
 # By default, Turbinia logs are written to a directory (GCS_OUTPUT_DIR)
 # in the GCS mount
diff --git a/turbinia/evidence.py b/turbinia/evidence.py
index b14cfa3..933e4f2 100644
--- a/turbinia/evidence.py
+++ b/turbinia/evidence.py
@@ -114,6 +114,9 @@ class Evidence(object):
   def __str__(self):
     return '{0:s}:{1:s}:{2!s}'.format(self.type, self.name, self.local_path)
 
+  def __repr__(self):
+    return self.__str__()
+
   def serialize(self):
     """Return JSON serializable object."""
     return self.__dict__
diff --git a/turbinia/output_manager.py b/turbinia/output_manager.py
index 1417e44..e853ecc 100644
--- a/turbinia/output_manager.py
+++ b/turbinia/output_manager.py
@@ -41,10 +41,12 @@ class OutputManager(object):
 
   Attributes:
     _output_writers (list): The configured output writers
+    is_setup (bool): Whether this object has been setup or not.
   """
 
   def __init__(self):
     self._output_writers = None
+    self.is_setup = False
 
   @staticmethod
   def get_output_writers(task):
@@ -71,11 +73,11 @@ class OutputManager(object):
       writers.append(writer)
     return writers
 
-  def get_local_output_dir(self):
-    """Gets the local output dir from the local output writer.
+  def get_local_output_dirs(self):
+    """Gets the local output directories from the local output writer.
 
     Returns:
-      String to locally created output directory.
+      Tuple(string): (Path to temp directory, path to local output directory)
 
     Raises:
       TurbiniaException: If no local output writer with output_dir is found.
@@ -93,7 +95,15 @@ class OutputManager(object):
       raise TurbiniaException(
           'Local output writer attribute local_output_dir is not set')
 
-    return writer.local_output_dir
+    if not hasattr(writer, 'tmp_dir'):
+      raise TurbiniaException(
+          'Local output writer does not have tmp_dir attribute.')
+
+    if not writer.tmp_dir:
+      raise TurbiniaException(
+          'Local output writer attribute tmp_dir is not set')
+
+    return (writer.tmp_dir, writer.local_output_dir)
 
   def retrieve_evidence(self, evidence_):
     """Retrieves evidence data from remote location.
@@ -129,7 +139,11 @@ class OutputManager(object):
     return evidence_
 
   def save_local_file(self, file_, result):
-    """Saves local file by writing to all non-local output writers.
+    """Saves local file by writing to all output writers.
+
+    Most local files will already be in the local output directory and won't
+    need to be copied by the LocalOutputWriter, but any files outside of this
+    directory (e.g. files in the tmp_dir) will still be copied locally.
 
     Args:
       file_ (string): Path to file to save.
@@ -142,18 +156,18 @@ class OutputManager(object):
     saved_path = None
     saved_path_type = None
     for writer in self._output_writers:
-      if writer.name != 'LocalOutputWriter':
-        new_path = writer.copy_to(file_)
-        if new_path:
-          result.saved_paths.append(new_path)
-          saved_path = new_path
-          saved_path_type = writer.name
+      new_path = writer.copy_to(file_)
+      if new_path:
+        result.saved_paths.append(new_path)
+        saved_path = new_path
+        saved_path_type = writer.name
 
     return saved_path, saved_path_type
 
   def setup(self, task):
     """Setup OutputManager object."""
     self._output_writers = self.get_output_writers(task)
+    self.is_setup = True
 
 
 class OutputWriter(object):
@@ -185,20 +199,28 @@ class OutputWriter(object):
           will be generated automatically if not set.
     """
     self.unique_dir = unique_dir
-    self.local_output_dir = local_output_dir
     self.name = self.NAME
+
     if base_output_dir:
       self.base_output_dir = base_output_dir
     else:
       config.LoadConfig()
       self.base_output_dir = config.OUTPUT_DIR
-    self.create_output_dir()
 
-  def create_output_dir(self):
+    if local_output_dir:
+      self.local_output_dir = local_output_dir
+    else:
+      self.local_output_dir = self.create_output_dir()
+
+
+  def create_output_dir(self, base_path=None):
     """Creates a unique output path for this task and creates directories.
 
     Needs to be run at runtime so that the task creates the directory locally.
 
+    Args:
+      base_path(string): Base directory output directory will be created in.
+
     Returns:
       A local output path string.
 
@@ -207,22 +229,28 @@ class OutputWriter(object):
     """
     raise NotImplementedError
 
-  def copy_to(self, file_):
-    """Copies output file to the managed location.
+  def copy_to(self, source_file):
+    """Copies file to the managed location.
+
+    Files will be copied into base_output_dir with a filename set to the
+    basename of the source file.
 
     Args:
-      file_: A string path to a source file.
+      source_file (string): A path to a local source file.
 
     Returns:
       The path the file was saved to, or None if file was not written.
     """
     raise NotImplementedError
 
-  def copy_from(self, file_):
-    """Copies output file from the managed location.
+  def copy_from(self, source_file):
+    """Copies output file from the managed location to the local output dir.
 
     Args:
-      file_: A string path to a source file.
+      source_file (string): A path to a source file in the managed storage
+          location.  This path should be in a format matching the storage type
+          (e.g. GCS paths are formatted like 'gs://bucketfoo/' and local paths
+          are like '/foo/bar'.
 
     Returns:
       The path the file was saved to, or None if file was not written.
@@ -231,7 +259,11 @@ class OutputWriter(object):
 
 
 class LocalOutputWriter(OutputWriter):
-  """Class for writing to local filesystem output."""
+  """Class for writing to local filesystem output.
+
+  Attributes:
+    tmp_dir (string): Path to temp directory
+  """
 
   NAME = 'LocalWriter'
 
@@ -239,13 +271,16 @@ class LocalOutputWriter(OutputWriter):
   def __init__(self, base_output_dir=None, *args, **kwargs):
     super(LocalOutputWriter, self).__init__(base_output_dir=base_output_dir,
                                             *args, **kwargs)
+    config.LoadConfig()
+    self.tmp_dir = self.create_output_dir(base_path=config.TMP_DIR)
 
-  def create_output_dir(self):
-    self.local_output_dir = os.path.join(self.base_output_dir, self.unique_dir)
-    if not os.path.exists(self.local_output_dir):
+  def create_output_dir(self, base_path=None):
+    base_path = base_path if base_path else self.base_output_dir
+    output_dir = os.path.join(base_path, self.unique_dir)
+    if not os.path.exists(output_dir):
       try:
-        log.info('Creating new directory {0:s}'.format(self.local_output_dir))
-        os.makedirs(self.local_output_dir)
+        log.debug('Creating new directory {0:s}'.format(output_dir))
+        os.makedirs(output_dir)
       except OSError as e:
         if e.errno == errno.EACCES:
           msg = 'Permission error ({0:s})'.format(str(e))
@@ -253,34 +288,41 @@ class LocalOutputWriter(OutputWriter):
           msg = str(e)
         raise TurbiniaException(msg)
 
-    return self.local_output_dir
+    return output_dir
 
   def _copy(self, file_path):
     """Copies file to local output dir.
 
     Args:
-      file_path: A string path to a source file.
+      file_path(string): Source path to the file to copy.
 
     Returns:
       The path the file was saved to, or None if file was not written.
     """
-    output_file = os.path.join(self.local_output_dir,
-                               os.path.basename(file_path))
+    destination_file = os.path.join(
+        self.local_output_dir, os.path.basename(file_path))
+
+    if self.local_output_dir in os.path.commonprefix(
+        [file_path, destination_file]):
+      log.debug('Not copying file {0:s} in output dir {1:s}'.format(
+          file_path, self.local_output_dir))
+      return None
     if not os.path.exists(file_path):
       log.warning('File [{0:s}] does not exist.'.format(file_path))
       return None
-    if os.path.exists(output_file):
-      log.warning('New file path [{0:s}] already exists.'.format(output_file))
+    if os.path.exists(destination_file):
+      log.warning('New file path [{0:s}] already exists.'.format(
+          destination_file))
       return None
 
-    shutil.copy(file_path, output_file)
-    return output_file
+    shutil.copy(file_path, destination_file)
+    return destination_file
 
-  def copy_to(self, file_path):
-    return self._copy(file_path)
+  def copy_to(self, source_file):
+    return self._copy(source_file)
 
-  def copy_from(self, file_path):
-    return self._copy(file_path)
+  def copy_from(self, source_file):
+    return self._copy(source_file)
 
 
 class GCSOutputWriter(OutputWriter):
@@ -323,30 +365,46 @@ class GCSOutputWriter(OutputWriter):
           'Cannot find bucket and path from GCS config {0:s}'.format(file_))
     return match.group(1), match.group(2)
 
-  def create_output_dir(self):
+  def create_output_dir(self, base_path=None):
     # Directories in GCS are artificial, so any path can be written as part of
     # the object name.
     pass
 
-  def copy_to(self, file_):
+  def copy_to(self, source_path):
     bucket = self.client.get_bucket(self.bucket)
-    full_path = os.path.join(
-        self.base_output_dir, self.unique_dir, os.path.basename(file_))
-    log.info('Writing {0:s} to GCS path {1:s}'.format(file_, full_path))
-    blob = storage.Blob(full_path, bucket, chunk_size=self.CHUNK_SIZE)
-    blob.upload_from_filename(file_, client=self.client)
-    return os.path.join('gs://', self.bucket, full_path)
-
-  def copy_from(self, file_):
+    destination_path = os.path.join(
+        self.base_output_dir, self.unique_dir, os.path.basename(source_path))
+    log.info(
+        'Writing {0:s} to GCS path {1:s}'.format(source_path, destination_path))
+    blob = storage.Blob(destination_path, bucket, chunk_size=self.CHUNK_SIZE)
+    blob.upload_from_filename(source_path, client=self.client)
+    return os.path.join('gs://', self.bucket, destination_path)
+
+  def copy_from(self, source_path):
+    """Copies output file from the managed location to the local output dir.
+
+    Args:
+      source_file (string): A path to a source file in the managed storage
+          location.  This path should be in a format matching the storage type
+          (e.g. GCS paths are formatted like 'gs://bucketfoo/' and local paths
+          are like '/foo/bar'.
+
+    Returns:
+      The path the file was saved to, or None if file was not written.
+
+    Raises:
+      TurbiniaException: If file retrieval fails.
+    """
     bucket = self.client.get_bucket(self.bucket)
-    gcs_path = self._parse_gcs_path(file_)[1]
-    full_path = os.path.join(self.local_output_dir, os.path.basename(file_))
+    gcs_path = self._parse_gcs_path(source_path)[1]
+    destination_path = os.path.join(
+        self.local_output_dir, os.path.basename(source_path))
     log.info('Writing GCS file {0:s} to local path {1:s}'.format(
-        file_, full_path))
+        source_path, destination_path))
     blob = storage.Blob(gcs_path, bucket, chunk_size=self.CHUNK_SIZE)
-    blob.download_to_filename(full_path, client=self.client)
-    if not os.path.exists(full_path):
+    blob.download_to_filename(destination_path, client=self.client)
+    if not os.path.exists(destination_path):
       raise TurbiniaException(
           'File retrieval from GCS failed: Local file {0:s} does not '
-          'exist'.format(full_path))
-    return full_path
+          'exist'.format(destination_path))
+    return destination_path
diff --git a/turbinia/turbiniactl.py b/turbinia/turbiniactl.py
index b744ed0..8d19f82 100755
--- a/turbinia/turbiniactl.py
+++ b/turbinia/turbiniactl.py
@@ -76,6 +76,12 @@ def main():
       '--request_id',
       help='Create new requests with this Request ID',
       required=False)
+  parser.add_argument(
+      '-R',
+      '--run_local',
+      action='store_true',
+      help='Run completely locally without any server or other infrastructure. '
+      'This can be used to run one-off Tasks to process data locally.')
   parser.add_argument(
       '-S',
       '--server',
@@ -119,6 +125,11 @@ def main():
       default=60,
       type=int,
       help='Number of seconds to wait between polling for task state info')
+  parser.add_argument(
+      '-t',
+      '--task',
+      help='The name of a single Task to run locally (must be used with '
+      '--run_local.')
   parser.add_argument(
       '-w',
       '--wait',
@@ -290,14 +301,14 @@ def main():
     log.setLevel(logging.WARNING)
 
   if args.jobs_whitelist and args.jobs_blacklist:
-    log.warning('A Job filter whitelist and blacklist cannot be specified '
-                'at the same time')
+    log.error('A Job filter whitelist and blacklist cannot be specified '
+              'at the same time')
     sys.exit(1)
 
   filter_patterns = None
   if (args.filter_patterns_file and
       not os.path.exists(args.filter_patterns_file)):
-    log.warning('Filter patterns file {0:s} does not exist.')
+    log.error('Filter patterns file {0:s} does not exist.')
     sys.exit(1)
   elif args.filter_patterns_file:
     try:
@@ -311,11 +322,24 @@ def main():
   if args.command not in ('psqworker', 'server'):
     if args.use_celery:
       client = TurbiniaCeleryClient()
+    elif args.run_local:
+      client = TurbiniaClient(run_local=True)
     else:
       client = TurbiniaClient()
   else:
     client = None
 
+  server_flags_set = args.server or args.use_celery or args.command == 'server'
+  worker_flags_set = (
+      args.use_celery or args.command in ('psqworker', 'celeryworker'))
+  if args.run_local and (server_flags_set or worker_flags_set):
+    log.error('--run_local flag is not compatible with server/worker flags')
+    sys.exit(1)
+
+  if args.run_local and not args.task:
+    log.error('--run_local flag requires --task flag')
+    sys.exit(1)
+
   if args.output_dir:
     config.OUTPUT_DIR = args.output_dir
   if args.log_file:
@@ -437,13 +461,14 @@ def main():
       log.warning(msg)
     else:
       msg += ' Use --force_evidence if you are sure you want to do this.'
-      log.warning(msg)
+      log.error(msg)
       sys.exit(1)
 
   # If we have evidence to process and we also want to run as a server, then
   # we'll just process the evidence directly rather than send it through the
   # PubSub frontend interface.  If we're not running as a server then we will
   # create a new TurbiniaRequest and send it over PubSub.
+  request = None
   if evidence_ and args.server:
     server = TurbiniaServer()
     server.add_evidence(evidence_)
@@ -458,7 +483,10 @@ def main():
     else:
       log.info('Creating request {0:s} with evidence {1:s}'.format(
           request.request_id, evidence_.name))
-      client.send_request(request)
+      if not args.run_local:
+        client.send_request(request)
+      else:
+        log.debug('--run_local specified so not sending request to server')
 
     if args.wait:
       log.info('Waiting for request {0:s} to complete'.format(
@@ -477,6 +505,16 @@ def main():
           request_id=request.request_id,
           all_fields=args.all_fields))
 
+  if args.run_local and not evidence_:
+    log.error('Evidence must be specified if using --run_local')
+    sys.exit(1)
+  if args.run_local and evidence_.cloud_only:
+    log.error('--run_local cannot be used with Cloud only Evidence types')
+    sys.exit(1)
+  if args.run_local and evidence_:
+    result = client.run_local_task(args.task, request)
+    log.info('Task execution result: {0:s}'.format(result))
+
   log.info('Done.')
   sys.exit(0)
 
diff --git a/turbinia/workers/__init__.py b/turbinia/workers/__init__.py
index be9f852..d18a770 100644
--- a/turbinia/workers/__init__.py
+++ b/turbinia/workers/__init__.py
@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-# Copyright 2015 Google Inc.
+# Copyright 2018 Google Inc.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -22,6 +22,7 @@ import logging
 import os
 import pickle
 import platform
+import pprint
 import subprocess
 import traceback
 import uuid
@@ -72,6 +73,9 @@ class TurbiniaTaskResult(object):
 
     Args:
       task (TurbiniaTask): The calling Task object
+
+    Raises:
+      TurbiniaException: If the Output Manager is not setup.
     """
 
     self.closed = False
@@ -93,7 +97,15 @@ class TurbiniaTaskResult(object):
     self.worker_name = platform.node()
     # TODO(aarontp): Create mechanism to grab actual python logging data.
     self._log = []
-    self.output_dir = task.output_manager.get_local_output_dir()
+    if task.output_manager.is_setup:
+      _, self.output_dir = task.output_manager.get_local_output_dirs()
+    else:
+      raise TurbiniaException('Output Manager is not setup yet.')
+
+
+  def __str__(self):
+    return pprint.pformat(vars(self), depth=3)
+
 
   def close(self, task, success, status=None):
     """Handles closing of this result and writing logs.
@@ -108,6 +120,9 @@ class TurbiniaTaskResult(object):
       success: Bool indicating task success
       status: One line descriptive task status.
     """
+    if self.closed:
+      # Don't try to close twice.
+      return
     self.successful = success
     self.run_time = datetime.now() - self.start_time
     if not status:
@@ -119,8 +134,9 @@ class TurbiniaTaskResult(object):
     for evidence in self.evidence:
       if evidence.local_path:
         self.saved_paths.append(evidence.local_path)
-        if evidence.copyable and not config.SHARED_FILESYSTEM:
-          task.output_manager.save_evidence(evidence, self)
+        if not task.run_local:
+          if evidence.copyable and not config.SHARED_FILESYSTEM:
+            task.output_manager.save_evidence(evidence, self)
       if not evidence.request_id:
         evidence.request_id = self.request_id
 
@@ -143,7 +159,8 @@ class TurbiniaTaskResult(object):
       with open(logfile, 'w') as f:
         f.write('\n'.join(self._log))
         f.write('\n')
-      task.output_manager.save_local_file(logfile, self)
+      if not task.run_local:
+        task.output_manager.save_local_file(logfile, self)
 
     self.closed = True
     log.debug('Result close successful. Status is [{0:s}]'.format(self.status))
@@ -199,11 +216,13 @@ class TurbiniaTask(object):
       output_manager: An output manager object
       result: A TurbiniaTaskResult object.
       request_id: The id of the initial request to process this evidence.
+      run_local: Whether we are running locally without a Worker or not.
       state_key: A key used to manage task state
       stub: The task manager implementation specific task stub that exists
             server side to keep a reference to the remote task objects.  For PSQ
             this is a task result object, but other implementations have their
             own stub objects.
+      tmp_dir: Temporary directory for Task to write to.
       user: The user who requested the task.
       _evidence_config (dict): The config that we want to pass to all new
             evidence created from this task.
@@ -229,8 +248,10 @@ class TurbiniaTask(object):
     self.output_manager = output_manager.OutputManager()
     self.result = None
     self.request_id = request_id
+    self.run_local = False
     self.state_key = None
     self.stub = None
+    self.tmp_dir = None
     self.user = user if user else getpass.getuser()
     self._evidence_config = {}
 
@@ -275,7 +296,8 @@ class TurbiniaTask(object):
     else:
       for file_ in save_files:
         result.log('Output file at {0:s}'.format(file_))
-        self.output_manager.save_local_file(file_, result)
+        if not self.run_local:
+          self.output_manager.save_local_file(file_, result)
       for evidence in new_evidence:
         # If the local path is set in the Evidence, we check to make sure that
         # the path exists and is not empty before adding it.
@@ -317,16 +339,17 @@ class TurbiniaTask(object):
       TurbiniaException: If the evidence can not be found.
     """
     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(
           task=self,
           input_evidence=[evidence],
           base_output_dir=self.base_output_dir,
           request_id=self.request_id)
-    self.output_dir = self.result.output_dir
 
-    if evidence.copyable and not config.SHARED_FILESYSTEM:
-      self.output_manager.retrieve_evidence(evidence)
+    if not self.run_local:
+      if evidence.copyable and not config.SHARED_FILESYSTEM:
+        self.output_manager.retrieve_evidence(evidence)
 
     if evidence.local_path and not os.path.exists(evidence.local_path):
       raise TurbiniaException('Evidence local path {0:s} does not exist'.format(
@@ -391,6 +414,7 @@ class TurbiniaTask(object):
     log.info('Result check: {0:s}'.format(check_status))
     return result
 
+
   def run_wrapper(self, evidence):
     """Wrapper to manage TurbiniaTaskResults and exception handling.
 
diff --git a/turbinia/workers/artifact.py b/turbinia/workers/artifact.py
index 1ff1049..a6c8835 100644
--- a/turbinia/workers/artifact.py
+++ b/turbinia/workers/artifact.py
@@ -66,6 +66,7 @@ class FileArtifactExtractionTask(TurbiniaTask):
     ret, _ = self.execute(cmd, result, save_files=[image_export_log])
     if ret:
       result.close(self, False, 'image_export.py failed.')
+      return result
 
     for dirpath, _, filenames in os.walk(export_directory):
       for filename in filenames:
diff --git a/turbinia/workers/plaso.py b/turbinia/workers/plaso.py
index c96ab56..2b58a01 100644
--- a/turbinia/workers/plaso.py
+++ b/turbinia/workers/plaso.py
@@ -39,7 +39,9 @@ class PlasoTask(TurbiniaTask):
     config.LoadConfig()
     plaso_evidence = PlasoFile()
 
-    plaso_file = os.path.join(self.output_dir, '{0:s}.plaso'.format(self.id))
+    # Write plaso file into tmp_dir because sqlite has issues with some shared
+    # filesystems (e.g NFS).
+    plaso_file = os.path.join(self.tmp_dir, '{0:s}.plaso'.format(self.id))
     plaso_evidence.local_path = plaso_file
     plaso_log = os.path.join(self.output_dir, '{0:s}.log'.format(self.id))
 


Test patch:
diff --git a/turbinia/output_manager_test.py b/turbinia/output_manager_test.py
index fba099b..5cf8db3 100644
--- a/turbinia/output_manager_test.py
+++ b/turbinia/output_manager_test.py
@@ -18,9 +18,108 @@ from __future__ import unicode_literals
 
 import unittest
 import os
+import shutil
 import tempfile
 
+import mock
+
+from turbinia import config
+from turbinia import evidence
 from turbinia import output_manager
+from turbinia import workers
+
+
+class TestLocalOutputManager(unittest.TestCase):
+  """Test LocalOutputManager module."""
+
+  @classmethod
+  def setUpClass(cls):
+    """Sets up the TestLocalOoutputManager class."""
+    config.LoadConfig()
+
+  def setUp(self):
+    """Test setup."""
+    # Set up TurbiniaTask
+    self.base_output_dir = tempfile.mkdtemp(prefix='turbinia-test-local')
+    self.tmp_dir = tempfile.mkdtemp(prefix='turbinia-test-tmp')
+    self.gcs_save = config.GCS_OUTPUT_PATH
+    self.tmp_dir_save = config.TMP_DIR
+    config.GCS_OUTPUT_PATH = 'gs://fake/path'
+    config.TMP_DIR = self.tmp_dir
+    self.task = workers.TurbiniaTask(base_output_dir=self.base_output_dir)
+
+  def tearDown(self):
+    """Tears Down class."""
+    config.GCS_OUTPUT_PATH = self.gcs_save
+    config.TMP_DIR = self.tmp_dir_save
+    if 'turbinia-test-local' in self.base_output_dir:
+      shutil.rmtree(self.base_output_dir)
+    if 'turbinia-test-tmp' in self.tmp_dir:
+      shutil.rmtree(self.tmp_dir)
+
+  def testGetOutputWriters(self):
+    """Tests get_output_writers function for valid response."""
+    writers = output_manager.OutputManager.get_output_writers(self.task)
+    self.assertEquals(len(writers), 2)
+    for writer in writers:
+      self.assertIsInstance(writer, output_manager.OutputWriter)
+
+  def testGetLocalOutputDirs(self):
+    """Tests get_local_output_dirs function for valid response."""
+    self.task.output_manager.setup(self.task)
+    tmp_dir, local_dir = self.task.output_manager.get_local_output_dirs()
+
+    self.assertTrue(self.task.output_manager.is_setup)
+    self.assertTrue(os.path.isdir(tmp_dir))
+    self.assertTrue(os.path.isdir(local_dir))
+    self.assertTrue(tmp_dir.startswith(self.tmp_dir))
+    self.assertTrue(local_dir.startswith(self.base_output_dir))
+
+  def testSaveLocalFile(self):
+    """Test the save_local_file method."""
+    # Set path to None so we don't try to initialize GCS outout writer.
+    config.GCS_OUTPUT_PATH = None
+    self.task.output_manager.setup(self.task)
+    tmp_dir, local_dir = self.task.output_manager.get_local_output_dirs()
+    self.task.result = mock.MagicMock()
+    self.task.result.saved_paths = []
+    test_contents = 'test_contents'
+    test_file = 'test-file.out'
+    src_file = os.path.join(tmp_dir, test_file)
+    dst_file = os.path.join(local_dir, test_file)
+
+    with open(src_file, 'w') as fh:
+      fh.write(test_contents)
+
+    self.assertFalse(os.path.exists(dst_file))
+    self.task.output_manager.save_local_file(src_file, self.task.result)
+    self.assertTrue(os.path.exists(dst_file))
+    self.assertIn(dst_file, self.task.result.saved_paths)
+
+  def testSaveEvidence(self):
+    """Test the save_evidence method."""
+    # Set path to None so we don't try to initialize GCS outout writer.
+    config.GCS_OUTPUT_PATH = None
+    self.task.output_manager.setup(self.task)
+    tmp_dir, local_dir = self.task.output_manager.get_local_output_dirs()
+    self.task.result = mock.MagicMock()
+    self.task.result.saved_paths = []
+    test_contents = 'test_contents'
+    test_file = 'test-file.out'
+    src_file = os.path.join(tmp_dir, test_file)
+    dst_file = os.path.join(local_dir, test_file)
+    test_evidence = evidence.Evidence()
+
+    with open(src_file, 'w') as fh:
+      fh.write(test_contents)
+    test_evidence.local_path = src_file
+
+    self.assertFalse(os.path.exists(dst_file))
+    return_evidence = self.task.output_manager.save_evidence(
+        test_evidence, self.task.result)
+    self.assertTrue(os.path.exists(dst_file))
+    self.assertIsInstance(return_evidence, evidence.Evidence)
+    self.assertIn(dst_file, return_evidence.saved_path)
 
 
 class TestLocalOutputWriter(unittest.TestCase):
diff --git a/turbinia/state_manager_test.py b/turbinia/state_manager_test.py
index 900f56f..5bb3663 100644
--- a/turbinia/state_manager_test.py
+++ b/turbinia/state_manager_test.py
@@ -58,6 +58,8 @@ class TestPSQStateManager(unittest.TestCase):
         base_output_dir=self.base_output_dir, name=self.test_data['name'],
         request_id=self.test_data['request_id'])
     self.task.output_manager = mock.MagicMock()
+    self.task.output_manager.get_local_output_dirs.return_value = (
+        '/fake/tmp/dir', self.base_output_dir)
 
     # Set up TurbiniaTaskResult
     self.result = TurbiniaTaskResult(
@@ -66,8 +68,6 @@ class TestPSQStateManager(unittest.TestCase):
     self.result.saved_paths = self.test_data['saved_paths']
     self.task.result = self.result
 
-    self.result.output_dir = self.base_output_dir
-
   def tearDown(self):
     config.STATE_MANAGER = self.state_manager_save
     [os.remove(f) for f in self.remove_files if os.path.exists(f)]
diff --git a/turbinia/workers/workers_test.py b/turbinia/workers/workers_test.py
index ff438ef..43dfc6f 100644
--- a/turbinia/workers/workers_test.py
+++ b/turbinia/workers/workers_test.py
@@ -39,6 +39,7 @@ class TestTurbiniaTask(unittest.TestCase):
     self.base_output_dir = tempfile.mkdtemp()
     self.task = TurbiniaTask(base_output_dir=self.base_output_dir)
     self.task.output_manager = mock.MagicMock()
+    self.task.output_manager.get_local_output_dirs.return_value = (None, None)
 
     # Set up RawDisk Evidence
     test_disk_path = tempfile.mkstemp(dir=self.base_output_dir)[1]