Repository: google/turbinia
Pull request #442 (2019-09-11 17:15:50)

Bulk Extractor Task
Created bulk extractor job, task, a processor module to compress and decompress archives, and the appropriate unit tests.

wajihyassine: All latest comments have been addressed and a new PR has been pushed for review.
wajihyassine: All latest comments have been addressed and a new PR has been pushed for review.


wajihyassine: New CL has been pushed and is ready for review!

Patch:
diff --git a/turbinia/client.py b/turbinia/client.py
index abe01b5..024d25c 100644
--- a/turbinia/client.py
+++ b/turbinia/client.py
@@ -49,6 +49,7 @@ from turbinia.workers.strings import StringsUnicodeTask
 from turbinia.workers.tomcat import TomcatAnalysisTask
 from turbinia.workers.volatility import VolatilityTask
 from turbinia.workers.worker_stat import StatTask
+from turbinia.workers.bulk_extractor import BulkExtractorTask
 
 # TODO(aarontp): Remove this map after
 # https://github.com/google/turbinia/issues/278 is fixed.
@@ -68,6 +69,7 @@ TASK_MAP = {
     'tomcatanalysistask': TomcatAnalysisTask,
     'volatilitytask': VolatilityTask,
     'stattask': StatTask,
+    'bulkextractortask': BulkExtractorTask
 }
 
 config.LoadConfig()
diff --git a/turbinia/evidence.py b/turbinia/evidence.py
index 62e0a31..e814ae2 100644
--- a/turbinia/evidence.py
+++ b/turbinia/evidence.py
@@ -24,6 +24,7 @@ import sys
 from turbinia import config
 from turbinia import TurbiniaException
 from turbinia.processors import mount_local
+from turbinia.processors import archive
 
 # pylint: disable=keyword-arg-before-vararg
 
@@ -166,7 +167,7 @@ class Evidence(object):
 
     return serialized
 
-  def _preprocess(self):
+  def _preprocess(self, _):
     """Preprocess this evidence prior to task running.
 
     This gets run in the context of the local task execution on the worker
@@ -184,16 +185,21 @@ class Evidence(object):
     """
     pass
 
-  def preprocess(self):
+  def preprocess(self, tmp_dir=None):
     """Runs the possible parent's evidence preprocessing code, then ours.
 
     This is a wrapper function that will call the chain of pre-processors
     starting with the most distant ancestor.  After all of the ancestors have
     been processed, then we run our pre-processor.
+
+    Args:
+      tmp_dir(str): The path to the temporary directory that the
+                       Task will write to.
+
     """
     if self.parent_evidence:
       self.parent_evidence.preprocess()
-    self._preprocess()
+    self._preprocess(tmp_dir)
 
   def postprocess(self):
     """Runs our postprocessing code, then our possible parent's evidence.
@@ -259,6 +265,41 @@ class Directory(Evidence):
   pass
 
 
+class CompressedDirectory(Evidence):
+  """CompressedDirectory based evidence.
+
+  Attributes:
+    compressed_directory: The path to the compressed directory.
+    uncompressed_directory: The path to the uncompressed directory.
+  """
+
+  def __init__(
+      self, compressed_directory=None, uncompressed_directory=None, *args,
+      **kwargs):
+    """Initialization for CompressedDirectory evidence object."""
+    super(CompressedDirectory, self).__init__(*args, **kwargs)
+    self.compressed_directory = compressed_directory
+    self.uncompressed_directory = uncompressed_directory
+    self.copyable = True
+
+  def _preprocess(self, tmp_dir):
+    # Uncompress a given tar file and return the uncompressed path.
+    self.uncompressed_directory = archive.UncompressTarFile(
+        self.local_path, tmp_dir)
+    self.local_path = self.uncompressed_directory
+
+  def compress(self):
+    """ Compresses a file or directory."""
+    # Compress a given directory and return the compressed path.
+    self.compressed_directory = archive.CompressDirectory(self.local_path)
+    self.local_path = self.compressed_directory
+
+
+class BulkExtractorOutput(CompressedDirectory):
+  """Bulk Extractor based evidence."""
+  pass
+
+
 class ChromiumProfile(Evidence):
   """Chromium based browser profile evidence.
 
@@ -297,7 +338,7 @@ class RawDisk(Evidence):
     self.size = size
     super(RawDisk, self).__init__(*args, **kwargs)
 
-  def _preprocess(self):
+  def _preprocess(self, _):
     self.loopdevice_path = mount_local.PreprocessLosetup(self.local_path)
 
   def _postprocess(self):
@@ -385,7 +426,7 @@ class GoogleCloudDisk(RawDisk):
     super(GoogleCloudDisk, self).__init__(*args, **kwargs)
     self.cloud_only = True
 
-  def _preprocess(self):
+  def _preprocess(self, _):
     self.local_path = google_cloud.PreprocessAttachDisk(self.disk_name)
 
   def _postprocess(self):
@@ -412,7 +453,7 @@ class GoogleCloudDiskRawEmbedded(GoogleCloudDisk):
     self.embedded_path = embedded_path
     super(GoogleCloudDiskRawEmbedded, self).__init__(*args, **kwargs)
 
-  def _preprocess(self):
+  def _preprocess(self, _):
     self.local_path = google_cloud.PreprocessAttachDisk(self.disk_name)
     self.loopdevice_path = mount_local.PreprocessLosetup(self.local_path)
     self.mount_path = mount_local.PreprocessMountDisk(
diff --git a/turbinia/jobs/__init__.py b/turbinia/jobs/__init__.py
index e916d72..8e51312 100644
--- a/turbinia/jobs/__init__.py
+++ b/turbinia/jobs/__init__.py
@@ -26,3 +26,4 @@ from turbinia.jobs import strings
 from turbinia.jobs import tomcat
 from turbinia.jobs import volatility
 from turbinia.jobs import worker_stat
+from turbinia.jobs import bulk_extractor
diff --git a/turbinia/jobs/bulk_extractor.py b/turbinia/jobs/bulk_extractor.py
new file mode 100644
index 0000000..6c2e34a
--- /dev/null
+++ b/turbinia/jobs/bulk_extractor.py
@@ -0,0 +1,55 @@
+# -*- coding: utf-8 -*-
+# Copyright 2019 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.
+"""Job to execute Bulk Extractor task."""
+
+from __future__ import unicode_literals
+
+from turbinia.evidence import GoogleCloudDisk
+from turbinia.evidence import GoogleCloudDiskRawEmbedded
+from turbinia.evidence import RawDisk
+from turbinia.evidence import BulkExtractorOutput
+from turbinia.jobs import interface
+from turbinia.jobs import manager
+from turbinia.workers.bulk_extractor import BulkExtractorTask
+
+
+class BulkExtractorJob(interface.TurbiniaJob):
+  """Bulk Extractor Job.
+
+  This will generate a compressed directory containing the resulting
+  Bulk Extractor output.
+  """
+
+  # The types of evidence that this Job will process
+  evidence_input = [RawDisk, GoogleCloudDisk, GoogleCloudDiskRawEmbedded]
+  evidence_output = [BulkExtractorOutput]
+
+  NAME = 'BulkExtractorJob'
+
+  def create_tasks(self, evidence):
+    """Create task for Bulk Extractor.
+
+    Args:
+      evidence: List of evidence objects to process
+
+    Returns:
+        A list of tasks to schedule.
+    """
+    # Generate tasks for Bulk Extractor job
+    tasks = [BulkExtractorTask() for _ in evidence]
+    return tasks
+
+
+manager.JobsManager.RegisterJob(BulkExtractorJob)
diff --git a/turbinia/jobs/plaso.py b/turbinia/jobs/plaso.py
index 7f1dd56..d51a778 100644
--- a/turbinia/jobs/plaso.py
+++ b/turbinia/jobs/plaso.py
@@ -18,6 +18,7 @@ from __future__ import unicode_literals
 
 from turbinia.evidence import APFSEncryptedDisk
 from turbinia.evidence import BitlockerDisk
+from turbinia.evidence import CompressedDirectory
 from turbinia.evidence import Directory
 from turbinia.evidence import GoogleCloudDisk
 from turbinia.evidence import GoogleCloudDiskRawEmbedded
@@ -33,7 +34,7 @@ class PlasoJob(interface.TurbiniaJob):
   # The types of evidence that this Job will process
   evidence_input = [
       Directory, RawDisk, GoogleCloudDisk, GoogleCloudDiskRawEmbedded,
-      BitlockerDisk, APFSEncryptedDisk
+      BitlockerDisk, APFSEncryptedDisk, CompressedDirectory
   ]
   evidence_output = [PlasoFile]
 
diff --git a/turbinia/processors/archive.py b/turbinia/processors/archive.py
new file mode 100644
index 0000000..d7f142c
--- /dev/null
+++ b/turbinia/processors/archive.py
@@ -0,0 +1,120 @@
+# -*- coding: utf-8 -*-
+# Copyright 2019 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.
+"""File archiving processor"""
+
+from __future__ import unicode_literals
+
+import os
+import tarfile
+import logging
+
+from time import time
+from turbinia import TurbiniaException
+
+log = logging.getLogger('turbinia')
+
+
+def ValidateTarFile(compressed_directory):
+  """ Validates a given compressed directory path.
+
+  Performs a check to determine if the path exists and if
+  the file extension is in the list of accepted extensions.
+
+  Args:
+    compressed_directory(str): The path to the compressed tar file.
+
+  Raises:
+    TurbiniaException: If validation fails.
+  """
+  if not os.path.exists(compressed_directory):
+    raise TurbiniaException(
+        'The File or Directory does not exist: {0:s}'.format(
+            compressed_directory))
+
+  # TODO(wyassine): rewrite this check so it is not dependant
+  # on a list of hard coded extensions and instead have a
+  # check to determine whether or not it is a tar file format.
+  if not (compressed_directory.endswith('.tgz') or
+          compressed_directory.endswith('.tar.gz')):
+    raise TurbiniaException(
+        'The file is not a supported format. The list of '
+        'acceptable exensions are: .tgz or .tar.gz')
+
+
+def CompressDirectory(uncompressed_directory):
+  """Compress a given directory into a tar file.
+
+  Args:
+    uncompressed_directory(str): The path to the uncompressed directory.
+
+  Returns:
+    str: The path to the tar file.
+  """
+  # Error handling check for a non-existent file or directory.
+  if not os.path.exists(uncompressed_directory):
+    raise TurbiniaException(
+        'The File or Directory does not exist: {0:s}'.format(
+            uncompressed_directory))
+
+  # Iterate through a given list of files and compress them.
+  compressed_directory = uncompressed_directory + '.tar.gz'
+  try:
+    with tarfile.TarFile.open(compressed_directory, 'w:gz') as tar:
+      tar.add(uncompressed_directory, arcname='')
+      tar.close()
+      log.info(
+          'The tar file has been created and '
+          'can be found at: {0:s}'.format(compressed_directory))
+  except IOError as exception:
+    raise TurbiniaException('An error has occured: {0:s}'.format(exception))
+  except tarfile.TarError as exception:
+    raise TurbiniaException(
+        'An error has while compressing the directory: {0:s}'.format(exception))
+  return compressed_directory
+
+
+def UncompressTarFile(compressed_directory, output_tmp):
+  """Uncompress a provided tar file.
+
+  Args:
+    compressed_directory(str): The path to the tar file.
+    output_tmp(str): The path to the temporary directory that the
+                      uncompressed tar file will be placed into.
+
+  Returns:
+    str: The path to the uncompressed directory.
+  """
+  # Tar file validation check
+  ValidateTarFile(compressed_directory)
+
+  # Generate the uncompressed directory path
+  uncompressed_file = 'uncompressed-' + str(int(time()))
+  uncompressed_directory = os.path.join(output_tmp, uncompressed_file)
+
+  # Uncompress the tar file into the uncompressed directory.
+  try:
+    tar = tarfile.TarFile.open(compressed_directory)
+    tar.extractall(path=uncompressed_directory)
+    tar.close()
+    log.info(
+        'The tar file has been uncompressed to the following directory: {0:s}'
+        .format(uncompressed_directory))
+  except IOError as exception:
+    raise TurbiniaException('An error has occured: {0:s}'.format(exception))
+  except tarfile.TarError as exception:
+    raise TurbiniaException(
+        'An error has occured while uncompressing the tar '
+        'file: {0:s}'.format(exception))
+  return uncompressed_directory
diff --git a/turbinia/turbiniactl.py b/turbinia/turbiniactl.py
index 19cb58c..40f5700 100644
--- a/turbinia/turbiniactl.py
+++ b/turbinia/turbiniactl.py
@@ -30,6 +30,7 @@ from turbinia import TurbiniaException
 from turbinia.config import logger
 from turbinia.lib import libcloudforensics
 from turbinia import __version__
+from turbinia.processors import archive
 
 log = logging.getLogger('turbinia')
 # We set up the logger first without the file handler, and we will set up the
@@ -275,6 +276,17 @@ def main():
   parser_directory.add_argument(
       '-n', '--name', help='Descriptive name of the evidence', required=False)
 
+  # Parser options for CompressedDirectory evidence type
+  parser_directory = subparsers.add_parser(
+      'compressedirectory', help='Process a compressed tar file as Evidence')
+  parser_directory.add_argument(
+      '-l', '--local_path', help='Local path to the evidence', required=True)
+  parser_directory.add_argument(
+      '-s', '--source', help='Description of the source of the evidence',
+      required=False)
+  parser_directory.add_argument(
+      '-n', '--name', help='Descriptive name of the evidence', required=False)
+
   # Parser options for ChromiumProfile evidence type
   parser_hindsight = subparsers.add_parser(
       'hindsight', help='Process ChromiumProfile as Evidence')
@@ -503,6 +515,12 @@ def main():
     local_path = os.path.abspath(args.local_path)
     evidence_ = evidence.Directory(
         name=args.name, local_path=local_path, source=args.source)
+  elif args.command == 'compressedirectory':
+    archive.ValidateTarFile(args.local_path)
+    args.name = args.name if args.name else args.local_path
+    local_path = os.path.abspath(args.local_path)
+    evidence_ = evidence.CompressedDirectory(
+        name=args.name, local_path=local_path, source=args.source)
   elif args.command == 'googleclouddisk':
     args.name = args.name if args.name else args.disk_name
     evidence_ = evidence.GoogleCloudDisk(
diff --git a/turbinia/workers/__init__.py b/turbinia/workers/__init__.py
index 8610b71..138e1fd 100644
--- a/turbinia/workers/__init__.py
+++ b/turbinia/workers/__init__.py
@@ -152,6 +152,7 @@ class TurbiniaTaskResult(object):
       success: Bool indicating task success
       status: One line descriptive task status.
     """
+
     if self.closed:
       # Don't try to close twice.
       return
@@ -514,7 +515,7 @@ class TurbiniaTask(object):
       raise TurbiniaException(
           'Evidence local path {0:s} does not exist'.format(
               evidence.local_path))
-    evidence.preprocess()
+    evidence.preprocess(self.tmp_dir)
     return self.result
 
   def touch(self):
diff --git a/turbinia/workers/bulk_extractor.py b/turbinia/workers/bulk_extractor.py
new file mode 100644
index 0000000..4817e46
--- /dev/null
+++ b/turbinia/workers/bulk_extractor.py
@@ -0,0 +1,141 @@
+# -*- coding: utf-8 -*-
+# Copyright 2019 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.
+"""Task for running Bulk Extractor."""
+
+import os
+import logging
+import xml.etree.ElementTree as xml_tree
+
+from turbinia import TurbiniaException
+
+from turbinia.evidence import BulkExtractorOutput
+from turbinia.workers import TurbiniaTask
+from turbinia.lib import text_formatter as fmt
+
+log = logging.getLogger('turbinia')
+
+
+class BulkExtractorTask(TurbiniaTask):
+  """Task to generate Bulk Extractor output."""
+
+  def run(self, evidence, result):
+    """Run Bulk Extractor binary.
+
+    Args:
+        evidence (Evidence object): The evidence we will process.
+        result (TurbiniaTaskResult): The object to place task results into.
+
+    Returns:
+        TurbiniaTaskResult object.
+    """
+    # TODO(wyassine): Research whether bulk extractor has an option to
+    # generate a summary report to stdout so that it could be used for
+    # a report in this task.
+    # Create the new Evidence object that will be generated by this Task.
+    output_evidence = BulkExtractorOutput()
+    # Create a path that we can write the new file to.
+    base_name = os.path.basename(evidence.local_path)
+    output_file_path = os.path.join(self.output_dir, base_name)
+    # Add the output path to the evidence so we can automatically save it
+    # later.
+    output_evidence.local_path = output_file_path
+
+    try:
+      # Generate the command we want to run then execute.
+      cmd = 'bulk_extractor {0:s} -o {1:s}'.format(
+          evidence.local_path, output_file_path)
+      result.log('Running Bulk Extractor as [{0:s}]'.format(cmd))
+      self.execute(cmd, result, new_evidence=[output_evidence], shell=True)
+
+      # Generate bulk extractor report
+      (report, summary) = self.generate_summary_report(output_file_path)
+      output_evidence.text_data = report
+      result.report_data = output_evidence.text_data
+
+      # Compress the bulk extractor output directory.
+      output_evidence.compress()
+      result.close(self, success=True, status=summary)
+    except TurbiniaException as exception:
+      result.close(self, success=False, status=str(exception))
+      return result
+    return result
+
+  def generate_summary_report(self, output_file_path):
+    """Generate a summary report from the resulting bulk extractor run.
+
+    Args:
+      output_file_path(str): the path to the bulk extractor output.
+
+    Returns:
+      tuple: containing:
+        report_test(str): The report data
+        summary(str): A summary of the report (used for task status)
+    """
+    findings = []
+    features_count = 0
+    report_path = os.path.join(output_file_path, 'report.xml')
+
+    # Check if report.xml was not generated by bulk extractor.
+    if not os.path.exists(report_path):
+      report = 'Execution successful, but the report is not available.'
+      return (report, report)
+
+    # Parse existing XML file.
+    xml = xml_tree.parse(report_path)
+
+    # Place in try/except statement to continue execution when
+    # an attribute is not found and NoneType is returned.
+    try:
+      # Retrieve summary related results.
+      findings.append(fmt.heading4('Bulk Extractor Results'))
+      findings.append(fmt.heading5('Run Summary'))
+      findings.append(
+          fmt.bullet(
+              'Program: {0} - {1}'.format(
+                  xml.find('creator/program').text,
+                  xml.find('creator/version').text)))
+      findings.append(
+          fmt.bullet(
+              'Command Line: {0}'.format(
+                  xml.find('creator/execution_environment/command_line').text)))
+      findings.append(
+          fmt.bullet(
+              'Start Time: {0}'.format(
+                  xml.find('creator/execution_environment/start_time').text)))
+      findings.append(
+          fmt.bullet(
+              'Elapsed Time: {0}'.format(
+                  xml.find('report/elapsed_seconds').text)))
+
+      # Retrieve results from each of the scanner runs
+      feature_files = xml.find('feature_files')
+      if feature_files is not None:
+        feature_iter = feature_files.iter()
+        findings.append(fmt.heading5('Scanner Results'))
+        for f in feature_iter:
+          if f.tag == 'feature_file':
+            name = next(feature_iter)
+            count = next(feature_iter)
+            findings.append(fmt.bullet('{0}:{1}'.format(name.text, count.text)))
+            features_count += int(count.text)
+      else:
+        findings.append(fmt.heading5("There are no findings to report."))
+    except AttributeError as exception:
+      log.warning(
+          'Error parsing feature from Bulk Extractor report: {0!s}'.format(
+              exception))
+    summary = '{0} artifacts have been extracted.'.format(features_count)
+    report = '\n'.join(findings)
+    return (report, summary)


Test patch:
diff --git a/turbinia/processors/archive_test.py b/turbinia/processors/archive_test.py
new file mode 100644
index 0000000..4258ec1
--- /dev/null
+++ b/turbinia/processors/archive_test.py
@@ -0,0 +1,88 @@
+# -*- coding: utf-8 -*-
+# Copyright 2019 Google Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Tests for the Archive processor to compress and decompress folders."""
+
+from __future__ import unicode_literals
+
+import os
+import tarfile
+import unittest
+import tempfile
+
+from random import randint
+from shutil import rmtree
+from turbinia.processors import archive
+from turbinia import TurbiniaException
+
+
+class ArchiveProcessorTest(unittest.TestCase):
+  """Tests for Archive Processor."""
+
+  def setUp(self):
+    # Setup testing directories/variables.
+    self.test_files = []
+    self.base_output_dir = tempfile.mkdtemp(prefix='turbinia-test-local')
+    self.tmp_files_dir = os.path.join(self.base_output_dir, 'files')
+    self.tmp_archive = os.path.join(self.base_output_dir, 'files.tar.gz')
+    if not os.path.exists(self.tmp_files_dir):
+      os.makedirs(self.tmp_files_dir)
+
+    # Generate text files containing random numbers.
+    file_max = 10
+    counter = 0
+    while counter <= file_max:
+      file_name = 'file{0:s}.txt'.format(str(counter))
+      file_path = os.path.join(self.tmp_files_dir, file_name)
+      file_open = open(file_path, 'w+')
+      rand_nums = [randint(0, 1000) for i in range(50)]
+      for i in rand_nums:
+        file_open.write('%s\n' % str(i))
+      file_open.close()
+      counter += 1
+      self.test_files.append(file_name)
+    archive.CompressDirectory(self.tmp_files_dir)
+
+  def tearDown(self):
+    # Remove testing directory for this unit test.
+    if os.path.exists(self.base_output_dir):
+      rmtree(self.base_output_dir)
+
+  def test_compressed_dir(self):
+    """Tests the compression function"""
+    # Check if compressed directory matches expected output path.
+    self.assertEqual(
+        archive.CompressDirectory(self.tmp_files_dir), self.tmp_archive)
+
+    # Check to confirm that the archive is gzip format.
+    self.assertEqual(tarfile.is_tarfile(self.tmp_archive), True)
+
+    # Raise assertion if folder does not exist.
+    with self.assertRaises(TurbiniaException):
+      archive.CompressDirectory('blah')
+
+  def test_validate_tarfile(self):
+    """Tests the validate function used to decompress tar files"""
+
+    # Raise exception for file that does not exist.
+    with self.assertRaises(TurbiniaException):
+      archive.ValidateTarFile('blah.no')
+
+    # Raise exception for a file with unsupported extension.
+    with self.assertRaises(TurbiniaException):
+      archive.ValidateTarFile(self.tmp_files_dir)
+
+
+if __name__ == '__main__':
+  unittest.main()
diff --git a/turbinia/workers/bulk_extractor_test.py b/turbinia/workers/bulk_extractor_test.py
new file mode 100644
index 0000000..df81094
--- /dev/null
+++ b/turbinia/workers/bulk_extractor_test.py
@@ -0,0 +1,99 @@
+# -*- coding: utf-8 -*-
+# Copyright 2019 Google Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Tests for the Bulk Extractor job."""
+
+from __future__ import unicode_literals
+from shutil import rmtree
+from io import StringIO
+
+import os
+import unittest
+import mock
+import textwrap
+
+from turbinia.evidence import BulkExtractorOutput
+from turbinia.workers import bulk_extractor
+from turbinia.workers.workers_test import TestTurbiniaTaskBase
+from turbinia.workers import TurbiniaTaskResult
+
+
+class BulkExtractorTaskTest(TestTurbiniaTaskBase):
+  """Tests for BulkExtractorTask."""
+
+  def setUp(self):
+    # pylint: disable=arguments-differ
+    super(BulkExtractorTaskTest, self).setUp(
+        task_class=bulk_extractor.BulkExtractorTask,
+        evidence_class=BulkExtractorOutput)
+    self.task.output_dir = self.task.base_output_dir
+    self.setResults(mock_run=False)
+
+  def tearDown(self):
+    # Remove testing directory for this unit test.
+    if os.path.exists(self.base_output_dir):
+      rmtree(self.base_output_dir)
+
+  def testBulkExtractorRun(self):
+    """Test BulkExtractor task run."""
+    self.task.execute = mock.MagicMock(return_value=0)
+    result = self.task.run(self.evidence, self.result)
+
+    # Ensure execute method is being called.
+    self.task.execute.assert_called_once()
+    # Ensure run method returns a TurbiniaTaskResult instance.
+    self.assertIsInstance(result, TurbiniaTaskResult)
+
+  @mock.patch('os.path')
+  def test_generate_report(self, mock_path):
+    """Tests Bulk Extractor report generation."""
+    # pylint: disable=line-too-long
+    empty_report_sample = textwrap.dedent(
+        """\
+        #### Bulk Extractor Results
+        ##### Run Summary
+        * Program: BULK_EXTRACTOR - 1.6.0-dev
+        * Command Line: bulk_extractor /tmp/test-small.img -o /output/test-small.img
+        * Start Time: 2019-09-27T16:34:48Z
+        * Elapsed Time: 2.389155
+        ##### There are no findings to report.""")
+
+    summary_sample = "0 artifacts have been extracted."
+
+    xml_sample = textwrap.dedent(
+        """\
+        <dfxml xmloutputversion="1.0">
+        <creator version="1.0">
+          <program>BULK_EXTRACTOR</program>
+          <version>1.6.0-dev</version>
+          <execution_environment>
+            <command_line>bulk_extractor /tmp/test-small.img -o /output/test-small.img</command_line>
+            <start_time>2019-09-27T16:34:48Z</start_time>
+          </execution_environment>
+        </creator>
+        <report>
+          <elapsed_seconds>2.389155</elapsed_seconds>
+        </report>
+        </dfxml>""")
+
+    str_io = StringIO(xml_sample)
+    mock_path.join.return_value = str_io
+    mock_path.exists.return_value = True
+    (report, summary) = self.task.generate_summary_report(str_io)
+    self.assertEqual(empty_report_sample, report)
+    self.assertEqual(summary_sample, summary)
+
+
+if __name__ == '__main__':
+  unittest.main()