Repository: google/turbinia
Pull request #381 (2019-04-19 07:07:39)

 Fix setting and querying for user/requester
This makes it so that `turbiniactl status -u user` actually works to query for all Tasks for that user.  Previously the user was being set server-side, so it wasn't actually setting the original requester.

This also changes most of the code/storage references from 'user' to 'requester' to be a little clearer in the code what is being referenced (because in some of the Google Cloud Functions both are used).

aarontp: @ericzinnikas This PR changes the stored 'user' attribute to be 'requester'.  Is there any schema update or anything else that needs to happen to make sure this is compatible with Redis?   

Patch:
diff --git a/CONTRIBUTING b/CONTRIBUTING
deleted file mode 100644
index 21e4c71..0000000
--- a/CONTRIBUTING
+++ /dev/null
@@ -1,25 +0,0 @@
-Want to contribute? Great! First, read this page (including the small print at the end).
-
-### Before you contribute
-Before we can use your code, you must sign the
-[Google Individual Contributor License Agreement](https://cla.developers.google.com/about/google-individual)
-(CLA), which you can do online. The CLA is necessary mainly because you own the
-copyright to your changes, even after your contribution becomes part of our
-codebase, so we need your permission to use and distribute your code. We also
-need to be sure of various other things—for instance that you'll tell us if you
-know that your code infringes on other people's patents. You don't have to sign
-the CLA until after you've submitted your code for review and a member has
-approved it, but you must do it before we can put your code into our codebase.
-Before you start working on a larger contribution, you should get in touch with
-us first through the issue tracker with your idea so that we can help out and
-possibly guide you. Coordinating up front makes it much easier to avoid
-frustration later on.
-
-### Code reviews
-All submissions, including submissions by project members, require review. We
-use Github pull requests for this purpose.
-
-### The small print
-Contributions made by corporations are covered by a different agreement than
-the one above, the
-[Software Grant and Corporate Contributor License Agreement](https://cla.developers.google.com/about/google-corporate).
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
new file mode 100644
index 0000000..bd3f271
--- /dev/null
+++ b/CONTRIBUTING.md
@@ -0,0 +1,2 @@
+# Want to contribute? 
+Great! First read this [page](https://github.com/google/turbinia/blob/master/docs/contributing.md).
diff --git a/tools/gcf_init/index.js b/tools/gcf_init/index.js
index 2919aca..2c07363 100644
--- a/tools/gcf_init/index.js
+++ b/tools/gcf_init/index.js
@@ -67,7 +67,7 @@ exports.gettasks = function gettasks(req, res) {
   }
   if (req.body.user) {
     console.log('Getting Turbinia Tasks by user: ' + req.body.user);
-    query = query.filter('user', '=', req.body.user);
+    query = query.filter('requester', '=', req.body.user);
   }
   if (req.body.start_time) {
     try {
@@ -150,8 +150,8 @@ exports.closetasks = function closetasks(req, res) {
         '__key__', '=', datastore.key([turbiniaKind, req.body.task_id]))
   }
   if (req.body.user) {
-    console.log('Adding filter - user: ' + req.body.user);
-    query = query.filter('user', '=', req.body.user)
+    console.log('Adding filter - requester: ' + req.body.user);
+    query = query.filter('requester', '=', req.body.user)
   }
 
   console.log(query);
diff --git a/tools/gcf_init/index.yaml b/tools/gcf_init/index.yaml
index c609e9e..6d617f8 100644
--- a/tools/gcf_init/index.yaml
+++ b/tools/gcf_init/index.yaml
@@ -45,7 +45,7 @@ indexes:
 - kind: TurbiniaTask
   ancestor: no
   properties:
-  - name: user
+  - name: requester
   - name: last_update
     direction: desc
 
@@ -54,7 +54,7 @@ indexes:
   properties:
   - name: instance
   - name: id
-  - name: user
+  - name: requester
   - name: last_update
     direction: desc
 
@@ -63,7 +63,7 @@ indexes:
   properties:
   - name: instance
   - name: request_id
-  - name: user
+  - name: requester
   - name: last_update
     direction: desc
 
diff --git a/turbinia/client.py b/turbinia/client.py
index d736023..6458ebe 100644
--- a/turbinia/client.py
+++ b/turbinia/client.py
@@ -31,6 +31,7 @@ from turbinia.config import logger
 from turbinia import task_manager
 from turbinia import TurbiniaException
 from turbinia.lib import text_formatter as fmt
+from turbinia.workers import Priority
 from turbinia.workers.artifact import FileArtifactExtractionTask
 from turbinia.workers.analysis.wordpress import WordpressAccessLogAnalysisTask
 from turbinia.workers.analysis.jenkins import JenkinsAnalysisTask
@@ -161,12 +162,12 @@ class TurbiniaClient(object):
       user (string): The user of the request we want tasks for.
       poll_interval (int): Interval of seconds between polling cycles.
     """
+    last_count = 0
     while True:
       task_results = self.get_task_data(
           instance, project, region, request_id=request_id, user=user)
       completed_tasks = []
       uncompleted_tasks = []
-      last_count = 0
       for task in task_results:
         if task.get('successful') is not None:
           completed_tasks.append(task)
@@ -304,7 +305,8 @@ class TurbiniaClient(object):
 
   def format_task_status(
       self, instance, project, region, days=0, task_id=None, request_id=None,
-      user=None, all_fields=False, full_report=False, priority_filter=20):
+      user=None, all_fields=False, full_report=False,
+      priority_filter=Priority.HIGH):
     """Formats the recent history for Turbinia Tasks.
 
     Args:
@@ -328,8 +330,14 @@ class TurbiniaClient(object):
     """
     task_results = self.get_task_data(
         instance, project, region, days, task_id, request_id, user)
+    if not task_results:
+      return ''
     # Sort all tasks by the report_priority so that tasks with a higher
     # priority are listed first in the report.
+    for result in task_results:
+      # 0 is a valid value, so checking against specific values
+      if result.get('report_priority') in (None, ''):
+        result['report_priority'] = Priority.LOW
     task_results = sorted(task_results, key=itemgetter('report_priority'))
     num_results = len(task_results)
     if not num_results:
@@ -339,7 +347,7 @@ class TurbiniaClient(object):
 
     # Build up data
     report = []
-    user = task_results[0].get('user')
+    requester = task_results[0].get('requester')
     request_id = task_results[0].get('request_id')
     success_types = ['Successful', 'Failed', 'Scheduled or Running']
     success_values = [True, False, None]
@@ -358,7 +366,8 @@ class TurbiniaClient(object):
     report.append(fmt.heading1('Turbinia report {0:s}'.format(request_id)))
     report.append(
         fmt.bullet(
-            'Processed {0:d} Tasks for user {1:s}'.format(num_results, user)))
+            'Processed {0:d} Tasks for user {1:s}'.format(
+                num_results, requester)))
 
     # Print report data for tasks
     for success_type in success_types:
diff --git a/turbinia/config/turbinia_config.py b/turbinia/config/turbinia_config.py
index d5e9522..2ea8426 100644
--- a/turbinia/config/turbinia_config.py
+++ b/turbinia/config/turbinia_config.py
@@ -26,10 +26,14 @@ from __future__ import unicode_literals
 # separate when running with the same Cloud projects or backend servers.
 INSTANCE_ID = 'turbinia-instance1'
 
-# Which state manager to use. Valid options are 'Datastore' or 'Redis'.
+# Which state manager to use. Valid options are 'Datastore' or 'Redis'.  Use
+# 'Datastore' for Cloud (GCP) or hybrid installations, and 'Redis' for local
+# installations.
 STATE_MANAGER = 'Datastore'
 
-# Which Task manager to use. Valid options are 'PSQ' and 'Celery'.
+# Which Task manager to use. Valid options are 'PSQ' and 'Celery'.  Use 'PSQ'
+# for Cloud (GCP) or hybrid installations, and 'Celery' for local
+# installations.
 TASK_MANAGER = 'PSQ'
 
 # Default base output directory for worker results and evidence.
@@ -98,7 +102,8 @@ PUBSUB_TOPIC = INSTANCE_ID
 
 # GCS Path to copy worker results and Evidence output to.
 # Otherwise, set this as 'None' if output will be stored in shared storage.
-GCS_OUTPUT_PATH = 'gs://%s/output' % BUCKET_NAME
+# GCS_OUTPUT_PATH = 'gs://%s/output' % BUCKET_NAME
+GCS_OUTPUT_PATH = None
 
 ################################################################################
 #                           Celery / Redis / Kombu
diff --git a/turbinia/evidence.py b/turbinia/evidence.py
index e5326a4..815d33f 100644
--- a/turbinia/evidence.py
+++ b/turbinia/evidence.py
@@ -265,6 +265,24 @@ class BitlockerDisk(EncryptedDisk):
     self.encryption_type = self.__class__.__name__
 
 
+class APFSEncryptedDisk(EncryptedDisk):
+  """APFS encrypted disk file evidence.
+
+  Attributes:
+    recovery_key: A string of the recovery key for this disk
+    password: A string of the password used for this disk. If recovery key
+        is used, this argument is ignored
+    unencrypted_path: A string to the unencrypted local path
+  """
+
+  def __init__(self, recovery_key=None, password=None, *args, **kwargs):
+    """Initialization for Bitlocker disk evidence object"""
+    self.recovery_key = recovery_key
+    self.password = password
+    super(APFSEncryptedDisk, self).__init__(*args, **kwargs)
+    self.encryption_type = self.__class__.__name__
+
+
 class GoogleCloudDisk(RawDisk):
   """Evidence object for Google Cloud Disks.
 
diff --git a/turbinia/jobs/plaso.py b/turbinia/jobs/plaso.py
index c5aa4ca..7f1dd56 100644
--- a/turbinia/jobs/plaso.py
+++ b/turbinia/jobs/plaso.py
@@ -16,6 +16,7 @@
 
 from __future__ import unicode_literals
 
+from turbinia.evidence import APFSEncryptedDisk
 from turbinia.evidence import BitlockerDisk
 from turbinia.evidence import Directory
 from turbinia.evidence import GoogleCloudDisk
@@ -32,7 +33,7 @@ class PlasoJob(interface.TurbiniaJob):
   # The types of evidence that this Job will process
   evidence_input = [
       Directory, RawDisk, GoogleCloudDisk, GoogleCloudDiskRawEmbedded,
-      BitlockerDisk
+      BitlockerDisk, APFSEncryptedDisk
   ]
   evidence_output = [PlasoFile]
 
diff --git a/turbinia/lib/libcloudforensics.py b/turbinia/lib/libcloudforensics.py
index bd5a7b4..612dcd9 100644
--- a/turbinia/lib/libcloudforensics.py
+++ b/turbinia/lib/libcloudforensics.py
@@ -270,7 +270,8 @@ class GoogleCloudProject(object):
     # Max length of disk names in GCP is 63 characters
     project_id = snapshot.project.project_id
     disk_id = project_id + snapshot.disk.name
-    disk_id_crc32 = '{0:08x}'.format(binascii.crc32(disk_id) & 0xffffffff)
+    disk_id_crc32 = '{0:08x}'.format(
+        binascii.crc32(disk_id.encode()) & 0xffffffff)
     truncate_at = 62 - len(disk_id_crc32) - len('-copy') - len(project_id)
     if disk_name_prefix:
       disk_name_prefix += '-'
@@ -330,8 +331,9 @@ class GoogleCloudProject(object):
 
     # Analysis software to install.
     packages_to_install = [
-        'python-plaso', 'xmount', 'libfvde-tools', 'libbde-tools',
-        'plaso-tools', 'jq', 'ncdu', 'htop', 'binutils', 'upx-ucl'
+        'python-plaso', 'xmount', 'sleuthkit', 'libfvde-tools', 'libbde-tools',
+        'plaso-tools', 'jq', 'ncdu', 'htop', 'binutils', 'upx-ucl',
+        'docker-explorer-tools'
     ]
 
     startup_script = """
diff --git a/turbinia/message.py b/turbinia/message.py
index 3ee736e..f9b3708 100644
--- a/turbinia/message.py
+++ b/turbinia/message.py
@@ -35,16 +35,19 @@ class TurbiniaRequest(object):
   """An object to request evidence to be processed.
 
   Attributes:
-    request_id: A client specified ID for this request.
-    recipe: Recipe to use when processing this request.
-    context: A Dict of context data to be passed around with this request.
-    evidence: A list of Evidence objects.
+    request_id(str): A client specified ID for this request.
+    requestor(str): The username of who made the request.
+    recipe(dict): Recipe to use when processing this request.
+    context(dict): A Dict of context data to be passed around with this request.
+    evidence(list): A list of Evidence objects.
   """
 
   def __init__(
-      self, request_id=None, recipe=None, context=None, evidence_=None):
+      self, request_id=None, requester=None, recipe=None, context=None,
+      evidence_=None):
     """Initialization for TurbiniaRequest."""
     self.request_id = request_id if request_id else uuid.uuid4().hex
+    self.requester = requester if requester else 'user_unspecified'
     self.recipe = recipe if recipe else {}
     self.context = context if context else {}
     self.evidence = evidence_ if evidence_ else []
diff --git a/turbinia/output_manager.py b/turbinia/output_manager.py
index 8ab388d..0c1c9e4 100644
--- a/turbinia/output_manager.py
+++ b/turbinia/output_manager.py
@@ -27,8 +27,9 @@ from turbinia import config
 from turbinia import TurbiniaException
 
 config.LoadConfig()
-if config.TASK_MANAGER.lower() == 'psq':
+if config.GCS_OUTPUT_PATH and config.GCS_OUTPUT_PATH.lower() is not 'none':
   from google.cloud import storage
+  from google.cloud import exceptions
 
 log = logging.getLogger('turbinia')
 
@@ -251,6 +252,10 @@ class OutputWriter(object):
 
     Returns:
       The path the file was saved to, or None if file was not written.
+
+    Raises:
+      TurbiniaException: When the source file is empty or there are problems
+          saving the file.
     """
     raise NotImplementedError
 
@@ -265,6 +270,9 @@ class OutputWriter(object):
 
     Returns:
       The path the file was saved to, or None if file was not written.
+
+    Raises:
+      TurbiniaException: When there are problems copying from storage.
     """
     raise NotImplementedError
 
@@ -292,12 +300,12 @@ class LocalOutputWriter(OutputWriter):
       try:
         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))
+      except OSError as exception:
+        if exception.errno == errno.EACCES:
+          message = 'Permission error ({0:s})'.format(str(exception))
         else:
-          msg = str(e)
-        raise TurbiniaException(msg)
+          message = str(exception)
+        raise TurbiniaException(message)
 
     return output_dir
 
@@ -383,13 +391,25 @@ class GCSOutputWriter(OutputWriter):
     pass
 
   def copy_to(self, source_path):
+    if os.path.getsize(source_path) == 0:
+      message = (
+          'Local source file {0:s} is empty.  Not uploading to GCS'.format(
+              source_path))
+      log.error(message)
+      raise TurbiniaException(message)
+
     bucket = self.client.get_bucket(self.bucket)
     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)
+    try:
+      blob = storage.Blob(destination_path, bucket, chunk_size=self.CHUNK_SIZE)
+      blob.upload_from_filename(source_path, client=self.client)
+    except exceptions.GoogleCloudError as exception:
+      message = 'File upload to GCS failed: {0!s}'.format(exception)
+      log.error(message)
+      raise TurbiniaException(message)
     return os.path.join('gs://', self.bucket, destination_path)
 
   def copy_from(self, source_path):
@@ -414,10 +434,24 @@ class GCSOutputWriter(OutputWriter):
     log.info(
         'Writing GCS file {0:s} to local path {1:s}'.format(
             source_path, destination_path))
-    blob = storage.Blob(gcs_path, bucket, chunk_size=self.CHUNK_SIZE)
-    blob.download_to_filename(destination_path, client=self.client)
+    try:
+      blob = storage.Blob(gcs_path, bucket, chunk_size=self.CHUNK_SIZE)
+      blob.download_to_filename(destination_path, client=self.client)
+    except exceptions.RequestRangeNotSatisfiable as exception:
+      message = (
+          'File retrieval from GCS failed, file may be empty: {0!s}'.format(
+              exception))
+      log.error(message)
+      raise TurbiniaException(message)
+    except exceptions.GoogleCloudError as exception:
+      message = 'File retrieval from GCS failed: {0!s}'.format(exception)
+      log.error(message)
+      raise TurbiniaException(message)
+
     if not os.path.exists(destination_path):
-      raise TurbiniaException(
+      message = (
           'File retrieval from GCS failed: Local file {0:s} does not '
           'exist'.format(destination_path))
+      log.error(message)
+      raise TurbiniaException(message)
     return destination_path
diff --git a/turbinia/task_manager.py b/turbinia/task_manager.py
index af6ecd1..df22a5c 100644
--- a/turbinia/task_manager.py
+++ b/turbinia/task_manager.py
@@ -168,6 +168,7 @@ class BaseTaskManager(object):
         job_count += 1
         for task in job.create_tasks([evidence_]):
           task.base_output_dir = config.OUTPUT_DIR
+          task.requester = evidence_.config.get('requester')
           self.add_task(task, evidence_)
 
     if not job_count:
@@ -352,6 +353,7 @@ class CeleryTaskManager(BaseTaskManager):
         if not evidence_.request_id:
           evidence_.request_id = request.request_id
         evidence_.config = request.recipe
+        evidence_.config['requester'] = request.requester
         log.info(
             'Received evidence [{0:s}] from Kombu message.'.format(
                 str(evidence_)))
@@ -436,6 +438,7 @@ class PSQTaskManager(BaseTaskManager):
         if not evidence_.request_id:
           evidence_.request_id = request.request_id
         evidence_.config = request.recipe
+        evidence_.config['requester'] = request.requester
         log.info(
             'Received evidence [{0:s}] from PubSub message.'.format(
                 str(evidence_)))
diff --git a/turbinia/turbiniactl.py b/turbinia/turbiniactl.py
index 3e8e900..02acd3a 100644
--- a/turbinia/turbiniactl.py
+++ b/turbinia/turbiniactl.py
@@ -35,6 +35,7 @@ from turbinia.config import logger
 from turbinia import evidence
 from turbinia import __version__
 from turbinia.message import TurbiniaRequest
+from turbinia.workers import Priority
 
 log = logging.getLogger('turbinia')
 logger.setup()
@@ -85,10 +86,6 @@ def main():
   parser.add_argument(
       '-S', '--server', action='store_true',
       help='Run Turbinia Server indefinitely')
-  parser.add_argument(
-      '-C', '--use_celery', action='store_true',
-      help='Pass this flag when using Celery/Kombu for task queuing and '
-      'messaging (instead of Google PSQ/pubsub)')
   parser.add_argument(
       '-V', '--version', action='version', version=__version__,
       help='Show the version')
@@ -145,6 +142,25 @@ def main():
   parser_rawdisk.add_argument(
       '-n', '--name', help='Descriptive name of the evidence', required=False)
 
+  # Parser options for APFS Disk Evidence type
+  parser_apfs = subparsers.add_parser(
+      'apfs', help='Process APFSEncryptedDisk as Evidence')
+  parser_apfs.add_argument(
+      '-l', '--local_path', help='Local path to the encrypted APFS evidence',
+      required=True)
+  parser_apfs.add_argument(
+      '-r', '--recovery_key', help='Recovery key for the APFS evidence.  '
+      'Either recovery key or password must be specified.', required=False)
+  parser_apfs.add_argument(
+      '-p', '--password', help='Password for the APFS evidence.  '
+      'If a recovery key is specified concurrently, password will be ignored.',
+      required=False)
+  parser_apfs.add_argument(
+      '-s', '--source', help='Description of the source of the evidence',
+      required=False)
+  parser_apfs.add_argument(
+      '-n', '--name', help='Descriptive name of the evidence', required=False)
+
   # Parser options for Bitlocker Disk Evidence type
   parser_bitlocker = subparsers.add_parser(
       'bitlocker', help='Process Bitlocker Disk as Evidence')
@@ -255,7 +271,8 @@ def main():
       '-r', '--request_id', help='Show tasks with this Request ID',
       required=False)
   parser_status.add_argument(
-      '-p', '--priority_filter', default=20, type=int, required=False,
+      '-p', '--priority_filter', default=Priority.HIGH, type=int,
+      required=False,
       help='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 '
@@ -304,7 +321,7 @@ def main():
   # Client
   config.LoadConfig()
   if args.command not in ('psqworker', 'server'):
-    if args.use_celery:
+    if config.TASK_MANAGER.lower() == 'celery':
       client = TurbiniaCeleryClient()
     elif args.run_local:
       client = TurbiniaClient(run_local=True)
@@ -313,9 +330,8 @@ def main():
   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'))
+  server_flags_set = args.server or args.command == 'server'
+  worker_flags_set = 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)
@@ -337,6 +353,15 @@ def main():
     evidence_ = evidence.RawDisk(
         name=args.name, local_path=local_path,
         mount_partition=args.mount_partition, source=args.source)
+  elif args.command == 'apfs':
+    if not args.password and not args.recovery_key:
+      log.error('Neither recovery key nor password is specified.')
+      sys.exit(1)
+    args.name = args.name if args.name else args.local_path
+    local_path = os.path.abspath(args.local_path)
+    evidence_ = evidence.APFSEncryptedDisk(
+        name=args.name, local_path=local_path, recovery_key=args.recovery_key,
+        password=args.password, source=args.source)
   elif args.command == 'bitlocker':
     if not args.password and not args.recovery_key:
       log.error('Neither recovery key nor password is specified.')
@@ -455,7 +480,8 @@ def main():
     server.add_evidence(evidence_)
     server.start()
   elif evidence_:
-    request = TurbiniaRequest(request_id=args.request_id)
+    request = TurbiniaRequest(
+        request_id=args.request_id, requester=getpass.getuser())
     request.evidence.append(evidence_)
     if filter_patterns:
       request.recipe['filter_patterns'] = filter_patterns
@@ -465,14 +491,14 @@ def main():
       request.recipe['jobs_whitelist'] = args.jobs_whitelist
     if args.dump_json:
       print(request.to_json().encode('utf-8'))
+      sys.exit(0)
     else:
       log.info(
           'Creating request {0:s} with evidence {1:s}'.format(
               request.request_id, evidence_.name))
       log.info(
-          'Run command "turbiniactl {0:s} status -r {1:s}" to see the status of'
-          ' this request and associated tasks'.format(
-              '-C' if args.use_celery else '', request.request_id))
+          'Run command "turbiniactl status -r {0:s}" to see the status of'
+          ' this request and associated tasks'.format(request.request_id))
       if not args.run_local:
         client.send_request(request)
       else:
diff --git a/turbinia/workers/__init__.py b/turbinia/workers/__init__.py
index fbfe6ff..ce65023 100644
--- a/turbinia/workers/__init__.py
+++ b/turbinia/workers/__init__.py
@@ -18,6 +18,7 @@ from __future__ import unicode_literals
 
 from copy import deepcopy
 from datetime import datetime, timedelta
+from enum import IntEnum
 import getpass
 import logging
 import os
@@ -40,6 +41,18 @@ from turbinia import TurbiniaException
 log = logging.getLogger('turbinia')
 
 
+class Priority(IntEnum):
+  """Reporting priority enum to store common values.
+
+  Priorities can be anything in the range of 0-100, where 0 is the highest
+  priority.
+  """
+  LOW = 80
+  MEDIUM = 50
+  HIGH = 20
+  CRITICAL = 10
+
+
 class TurbiniaTaskResult(object):
   """Object to store task results to be returned by a TurbiniaTask.
 
@@ -62,7 +75,7 @@ class TurbiniaTaskResult(object):
       successful: Bool indicating success status.
       task_id: Task ID of the parent task.
       task_name: Name of parent task.
-      user: The user who requested the task.
+      requester: The user who requested the task.
       worker_name: Name of worker task executed on.
       _log: A list of log messages
   """
@@ -87,11 +100,11 @@ class TurbiniaTaskResult(object):
 
     self.task_id = None
     self.task_name = None
-    self.user = None
+    self.requester = None
     self.output_dir = None
 
     self.report_data = None
-    self.report_priority = 50
+    self.report_priority = Priority.MEDIUM
     self.start_time = datetime.now()
     self.run_time = None
     self.saved_paths = []
@@ -117,7 +130,7 @@ class TurbiniaTaskResult(object):
 
     self.task_id = task.id
     self.task_name = task.name
-    self.user = task.user
+    self.requester = task.requester
     if task.output_manager.is_setup:
       _, self.output_dir = task.output_manager.get_local_output_dirs()
     else:
@@ -151,11 +164,16 @@ class TurbiniaTaskResult(object):
     self.status = status
 
     for evidence in self.evidence:
-      if evidence.local_path:
+      if evidence.local_path and os.path.exists(evidence.local_path):
         self.saved_paths.append(evidence.local_path)
         if not task.run_local:
           if evidence.copyable and not config.SHARED_FILESYSTEM:
             task.output_manager.save_evidence(evidence, self)
+      else:
+        self.log(
+            'Evidence {0:s} has empty or missing file at local_path {1:s} so '
+            'not saving.'.format(evidence.name, evidence.local_path))
+
       if not evidence.request_id:
         evidence.request_id = self.request_id
 
@@ -165,13 +183,17 @@ class TurbiniaTaskResult(object):
     # to clean things up even after other failures in the task, so this could
     # also fail.
     # pylint: disable=broad-except
-    except Exception as e:
-      msg = 'Evidence post-processing for {0:s} failed: {1!s}'.format(
-          self.input_evidence.name, e)
-      self.log(msg, level=logging.ERROR)
+    except Exception as exception:
+      message = 'Evidence post-processing for {0:s} failed: {1!s}'.format(
+          self.input_evidence.name, exception)
+      self.log(message, level=logging.ERROR)
 
     # Write result log info to file
     logfile = os.path.join(self.output_dir, 'worker-log.txt')
+    # Create default log text just so that the worker log is created to
+    # avoid confusion if it doesn't exist.
+    if not self._log:
+      self._log.append('No worker messages were logged.')
     if self.output_dir and os.path.exists(self.output_dir):
       with open(logfile, 'w') as f:
         f.write('\n'.join(self._log))
@@ -293,16 +315,16 @@ class TurbiniaTask(object):
           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.
+      requester: The user who requested the task.
       _evidence_config (dict): The config that we want to pass to all new
             evidence created from this task.
   """
 
   # The list of attributes that we will persist into storage
-  STORED_ATTRIBUTES = ['id', 'last_update', 'name', 'request_id', 'user']
+  STORED_ATTRIBUTES = ['id', 'last_update', 'name', 'request_id', 'requester']
 
   def __init__(
-      self, name=None, base_output_dir=None, request_id=None, user=None):
+      self, name=None, base_output_dir=None, request_id=None, requester=None):
     """Initialization for TurbiniaTask."""
     if base_output_dir:
       self.base_output_dir = base_output_dir
@@ -320,7 +342,7 @@ class TurbiniaTask(object):
     self.stub = None
     self.tmp_dir = None
     self.turbinia_version = turbinia.__version__
-    self.user = user if user else getpass.getuser()
+    self.requester = requester if requester else 'user_unspecified'
     self._evidence_config = {}
 
   def serialize(self):
@@ -350,11 +372,11 @@ class TurbiniaTask(object):
     try:
       task = getattr(sys.modules['turbinia.client'], type_)()
     except AttributeError:
-      msg = (
+      message = (
           "Could not import {0:s} object! Make sure it is imported where "
           "this method is defined.".format(type_))
-      log.error(msg)
-      raise TurbiniaException(msg)
+      log.error(message)
+      raise TurbiniaException(message)
     task.__dict__.update(input_dict)
     task.output_manager = output_manager.OutputManager()
     task.output_manager.__dict__.update(input_dict['output_manager'])
@@ -401,35 +423,46 @@ class TurbiniaTask(object):
             'Log file {0:s} does not exist to save'.format(file_),
             level=logging.DEBUG)
         continue
+      if os.path.getsize(file_) == 0:
+        result.log(
+            'Log file {0:s} is empty. Not saving'.format(file_),
+            level=logging.DEBUG)
+        continue
       result.log('Output file at {0:s}'.format(file_))
       if not self.run_local:
         self.output_manager.save_local_file(file_, result)
 
     if ret not in success_codes:
-      msg = 'Execution of [{0!s}] failed with status {1:d}'.format(cmd, ret)
-      result.log(msg)
+      message = 'Execution of [{0!s}] failed with status {1:d}'.format(cmd, ret)
+      result.log(message)
       if close:
-        result.close(self, success=False, status=msg)
+        result.close(self, success=False, status=message)
     else:
       result.log('Execution of [{0!s}] succeeded'.format(cmd))
       for file_ in save_files:
+        if os.path.getsize(file_) == 0:
+          result.log(
+              'Output file {0:s} is empty. Not saving'.format(file_),
+              level=logging.DEBUG)
+          continue
         result.log('Output file at {0:s}'.format(file_))
         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.
         if evidence.local_path and not os.path.exists(evidence.local_path):
-          msg = (
+          message = (
               'Evidence {0:s} local_path {1:s} does not exist. Not returning '
               'empty Evidence.'.format(evidence.name, evidence.local_path))
-          result.log(msg, level=logging.WARN)
+          result.log(message, level=logging.WARN)
         elif (evidence.local_path and os.path.exists(evidence.local_path) and
               os.path.getsize(evidence.local_path) == 0):
-          msg = (
+          message = (
               'Evidence {0:s} local_path {1:s} is empty. Not returning '
               'empty new Evidence.'.format(evidence.name, evidence.local_path))
-          result.log(msg, level=logging.WARN)
+          result.log(message, level=logging.WARN)
         else:
           result.add_evidence(evidence, self._evidence_config)
 
@@ -504,11 +537,11 @@ class TurbiniaTask(object):
       try:
         log.debug('Checking TurbiniaTaskResult for serializability')
         pickle.dumps(result)
-      except (TypeError, pickle.PicklingError) as e:
+      except (TypeError, pickle.PicklingError) as exception:
         bad_message = (
             'Error pickling TurbiniaTaskResult object. Returning a new result '
             'with the pickling error, and all previous result data will be '
-            'lost. Pickle Error: {0!s}'.format(e))
+            'lost. Pickle Error: {0!s}'.format(exception))
 
     if bad_message:
       log.error(bad_message)
@@ -561,30 +594,32 @@ class TurbiniaTask(object):
         original_result_id = self.result.id
 
         if self.turbinia_version != turbinia.__version__:
-          msg = (
+          message = (
               'Worker and Server versions do not match: {0:s} != {1:s}'.format(
                   self.turbinia_version, turbinia.__version__))
-          self.result.log(msg, level=logging.ERROR)
-          self.result.status = msg
+          self.result.log(message, level=logging.ERROR)
+          self.result.status = message
           return self.result
 
         self._evidence_config = evidence.config
         self.result = self.run(evidence, self.result)
       # pylint: disable=broad-except
-      except Exception as e:
-        msg = '{0:s} Task failed with exception: [{1!s}]'.format(self.name, e)
+      except Exception as exception:
+        message = (
+            '{0:s} Task failed with exception: [{1!s}]'.format(
+                self.name, exception))
         # Logging explicitly here because the result is in an unknown state
         trace = traceback.format_exc()
-        log.error(msg)
+        log.error(message)
         log.error(trace)
         if self.result:
-          self.result.log(msg, level=logging.ERROR)
+          self.result.log(message, level=logging.ERROR)
           self.result.log(trace)
-          if hasattr(e, 'message'):
-            self.result.set_error(e.message, traceback.format_exc())
+          if hasattr(exception, 'message'):
+            self.result.set_error(exception.message, traceback.format_exc())
           else:
-            self.result.set_error(e.__class__, traceback.format_exc())
-          self.result.status = msg
+            self.result.set_error(exception.__class__, traceback.format_exc())
+          self.result.status = message
         else:
           log.error('No TurbiniaTaskResult object found after task execution.')
 
@@ -594,28 +629,28 @@ class TurbiniaTask(object):
       # This has a higher likelihood of failing because something must have gone
       # wrong as the Task should have already closed this.
       if self.result and not self.result.closed:
-        msg = 'Trying last ditch attempt to close result'
-        log.warning(msg)
-        self.result.log(msg)
+        message = 'Trying last ditch attempt to close result'
+        log.warning(message)
+        self.result.log(message)
 
         if self.result.status:
           status = self.result.status
         else:
           status = 'No previous status'
-        msg = (
+        message = (
             'Task Result was auto-closed from task executor on {0:s} likely '
             'due to previous failures.  Previous status: [{1:s}]'.format(
                 self.result.worker_name, status))
-        self.result.log(msg)
+        self.result.log(message)
         try:
-          self.result.close(self, False, msg)
+          self.result.close(self, False, message)
         # Using broad except here because lots can go wrong due to the reasons
         # listed above.
         # pylint: disable=broad-except
-        except Exception as e:
-          log.error('TurbiniaTaskResult close failed: {0!s}'.format(e))
+        except Exception as exception:
+          log.error('TurbiniaTaskResult close failed: {0!s}'.format(exception))
           if not self.result.status:
-            self.result.status = msg
+            self.result.status = message
         # Check the result again after closing to make sure it's still good.
         self.result = self.validate_result(self.result)
 
diff --git a/turbinia/workers/analysis/jenkins.py b/turbinia/workers/analysis/jenkins.py
index a7bb3f2..e9a2be3 100644
--- a/turbinia/workers/analysis/jenkins.py
+++ b/turbinia/workers/analysis/jenkins.py
@@ -23,6 +23,7 @@ from turbinia import TurbiniaException
 from turbinia.evidence import ReportText
 from turbinia.lib import text_formatter as fmt
 from turbinia.workers import TurbiniaTask
+from turbinia.workers import Priority
 from turbinia.lib.utils import extract_artifacts
 from turbinia.lib.utils import bruteforce_password_hashes
 
@@ -148,7 +149,7 @@ class JenkinsAnalysisTask(TurbiniaTask):
     """
     report = []
     summary = ''
-    priority = 50
+    priority = Priority.LOW
     credentials_registry = {hash: username for username, hash in credentials}
     # TODO: Add timeout parameter when dynamic configuration is ready.
     # Ref: https://github.com/google/turbinia/issues/244
@@ -159,7 +160,7 @@ class JenkinsAnalysisTask(TurbiniaTask):
     report.append(fmt.bullet('Jenkins version: {0:s}'.format(version)))
 
     if weak_passwords:
-      priority = 10
+      priority = Priority.CRITICAL
       summary = 'Jenkins analysis found potential issues'
       report.insert(0, fmt.heading4(fmt.bold(summary)))
       line = '{0:n} weak password(s) found:'.format(len(weak_passwords))
@@ -170,7 +171,7 @@ class JenkinsAnalysisTask(TurbiniaTask):
         report.append(fmt.bullet(line, level=2))
     else:
       summary = 'Jenkins analysis found no issues'
-      priority = 80
+      priority = Priority.LOW
       report.insert(0, fmt.heading4(summary))
 
     report = '\n'.join(report)
diff --git a/turbinia/workers/analysis/wordpress.py b/turbinia/workers/analysis/wordpress.py
index 74c723e..b8395b1 100644
--- a/turbinia/workers/analysis/wordpress.py
+++ b/turbinia/workers/analysis/wordpress.py
@@ -23,6 +23,7 @@ import re
 from turbinia.evidence import ReportText
 from turbinia.lib import text_formatter as fmt
 from turbinia.workers import TurbiniaTask
+from turbinia.workers import Priority
 
 
 class WordpressAccessLogAnalysisTask(TurbiniaTask):
@@ -123,7 +124,7 @@ class WordpressAccessLogAnalysisTask(TurbiniaTask):
 
       report.insert(0, fmt.heading4(fmt.bold(summary)))
       report_text = '\n'.join(report)
-      return (report_text, 20, summary)
+      return (report_text, Priority.HIGH, summary)
 
     report_text = 'No Wordpress install or theme editing found in access logs'
-    return (fmt.heading4(report_text), 80, report_text)
+    return (fmt.heading4(report_text), Priority.LOW, report_text)
diff --git a/turbinia/workers/hadoop.py b/turbinia/workers/hadoop.py
index 95cecf8..0324d8f 100644
--- a/turbinia/workers/hadoop.py
+++ b/turbinia/workers/hadoop.py
@@ -27,6 +27,7 @@ from turbinia.lib import text_formatter as fmt
 from turbinia.evidence import ReportText
 from turbinia.lib.utils import extract_artifacts
 from turbinia.workers import TurbiniaTask
+from turbinia.workers import Priority
 
 log = logging.getLogger('turbinia')
 
@@ -57,7 +58,7 @@ class HadoopAnalysisTask(TurbiniaTask):
     report = []
     evil_commands = []
     strings_count = 0
-    priority = 50
+    priority = Priority.MEDIUM
     summary = ''
     for filepath in collected_artifacts:
       relpath = os.path.relpath(filepath, output_dir)
@@ -76,7 +77,7 @@ class HadoopAnalysisTask(TurbiniaTask):
       msg = 'Found suspicious commands!'
       report.append(fmt.heading4(fmt.bold(msg)))
       summary = msg
-      priority = 10
+      priority = Priority.CRITICAL
     else:
       msg = 'Did not find any suspicious commands.'
       report.append(fmt.heading4(msg))
diff --git a/turbinia/workers/plaso.py b/turbinia/workers/plaso.py
index 5e42f05..0215f09 100644
--- a/turbinia/workers/plaso.py
+++ b/turbinia/workers/plaso.py
@@ -19,6 +19,7 @@ from __future__ import unicode_literals
 import os
 
 from turbinia import config
+from turbinia.evidence import APFSEncryptedDisk
 from turbinia.evidence import BitlockerDisk
 from turbinia.evidence import PlasoFile
 from turbinia.workers import TurbiniaTask
@@ -53,7 +54,7 @@ class PlasoTask(TurbiniaTask):
     if config.DEBUG_TASKS:
       cmd.append('-d')
 
-    if isinstance(evidence, BitlockerDisk):
+    if isinstance(evidence, (APFSEncryptedDisk, BitlockerDisk)):
       if evidence.recovery_key:
         cmd.extend([
             '--credential', 'recovery_password:{0:s}'.format(
diff --git a/turbinia/workers/sshd.py b/turbinia/workers/sshd.py
index 3317adc..c2822e8 100644
--- a/turbinia/workers/sshd.py
+++ b/turbinia/workers/sshd.py
@@ -22,6 +22,7 @@ import re
 from turbinia.evidence import ReportText
 from turbinia.lib import text_formatter as fmt
 from turbinia.workers import TurbiniaTask
+from turbinia.workers import Priority
 
 
 class SSHDAnalysisTask(TurbiniaTask):
@@ -99,7 +100,7 @@ class SSHDAnalysisTask(TurbiniaTask):
       summary = 'Insecure SSH configuration found.'
       findings.insert(0, fmt.heading4(fmt.bold(summary)))
       report = '\n'.join(findings)
-      return (report, 20, summary)
+      return (report, Priority.HIGH, summary)
 
     report = 'No issues found in SSH configuration'
-    return (report, 60, report)
+    return (report, Priority.LOW, report)
diff --git a/turbinia/workers/tomcat.py b/turbinia/workers/tomcat.py
index 2daeb9e..e2e6281 100644
--- a/turbinia/workers/tomcat.py
+++ b/turbinia/workers/tomcat.py
@@ -22,6 +22,7 @@ import re
 from turbinia.evidence import ReportText
 from turbinia.lib import text_formatter as fmt
 from turbinia.workers import TurbiniaTask
+from turbinia.workers import Priority
 
 
 class TomcatAnalysisTask(TurbiniaTask):
@@ -104,7 +105,7 @@ class TomcatAnalysisTask(TurbiniaTask):
       msg = 'Tomcat analysis found {0:d} results'.format(count)
       findings.insert(0, fmt.heading4(fmt.bold(msg)))
       report = '\n'.join(findings)
-      return (report, 20, msg)
+      return (report, Priority.HIGH, msg)
 
     report = 'No Tomcat findings to report'
-    return (report, 80, report)
+    return (report, Priority.LOW, report)


Test patch:
diff --git a/turbinia/client_test.py b/turbinia/client_test.py
index d57cb34..e2bd8d4 100644
--- a/turbinia/client_test.py
+++ b/turbinia/client_test.py
@@ -127,7 +127,7 @@ class TestTurbiniaClient(unittest.TestCase):
             'saved_paths': ['/no/path/', '/fake/path'],
             'status': 'This fake task executed',
             'successful': True,
-            'user': 'myuser',
+            'requester': 'myuser',
             'worker_name': 'fake_worker'
         }, {
             'id': '0xfakeTaskId2',
@@ -140,7 +140,7 @@ class TestTurbiniaClient(unittest.TestCase):
             'saved_paths': ['/no/path/2', '/fake/path/2'],
             'status': 'This second fake task executed',
             'successful': True,
-            'user': 'myuser',
+            'requester': 'myuser',
             'worker_name': 'fake_worker'
         }, {
             'id': '0xfakeTaskId3',
@@ -153,7 +153,7 @@ class TestTurbiniaClient(unittest.TestCase):
             'saved_paths': ['/no/path/3', '/fake/path/3'],
             'status': 'Third Task Failed...',
             'successful': False,
-            'user': 'myuser',
+            'requester': 'myuser',
             'worker_name': 'fake_worker'
         }
     ] # yapf: disable
@@ -200,6 +200,20 @@ class TestTurbiniaClient(unittest.TestCase):
     self.assertRaises(
         TurbiniaException, client.get_task_data, "inst", "proj", "reg")
 
+  @mock.patch('turbinia.client.GoogleCloudFunction.ExecuteFunction')
+  @mock.patch('turbinia.client.task_manager.PSQTaskManager._backend_setup')
+  @mock.patch('turbinia.state_manager.get_state_manager')
+  def testClientFormatTaskStatus(self, _, __, ___):
+    """Tests format_task_status() with empty report_priority."""
+    client = TurbiniaClient()
+    client.get_task_data = mock.MagicMock()
+    self.task_data[0]['report_priority'] = None
+    self.task_data[1]['report_priority'] = ''
+    self.task_data[2].pop('report_priority')
+    client.get_task_data.return_value = self.task_data
+    result = client.format_task_status('inst', 'proj', 'reg')
+    self.assertIn('Processed 3 Tasks', result.strip())
+
   @mock.patch('turbinia.client.GoogleCloudFunction.ExecuteFunction')
   @mock.patch('turbinia.client.task_manager.PSQTaskManager._backend_setup')
   @mock.patch('turbinia.state_manager.get_state_manager')
diff --git a/turbinia/output_manager_test.py b/turbinia/output_manager_test.py
index 65dfaac..caec1e8 100644
--- a/turbinia/output_manager_test.py
+++ b/turbinia/output_manager_test.py
@@ -59,6 +59,7 @@ class TestLocalOutputManager(unittest.TestCase):
 
   def testGetOutputWriters(self):
     """Tests get_output_writers function for valid response."""
+    output_manager.storage = mock.MagicMock()
     writers = output_manager.OutputManager.get_output_writers(self.task)
     self.assertEquals(len(writers), 2)
     for writer in writers:
@@ -66,6 +67,7 @@ class TestLocalOutputManager(unittest.TestCase):
 
   def testGetLocalOutputDirs(self):
     """Tests get_local_output_dirs function for valid response."""
+    output_manager.storage = mock.MagicMock()
     self.task.output_manager.setup(self.task)
     tmp_dir, local_dir = self.task.output_manager.get_local_output_dirs()