Repository: log2timeline/dftimewolf

Issue:
Add support for Turbinia recipes
Add support for specifying turbinia recipe names.

The user should be able to pass a new cli parameter (e.g. --turbinia_recipe) from any dfTimewolf recipe that utilizes Turbinia.

This will enable use of new Trubiania triage recipes.

Discussion:
jleaniz: Tests are failing because the version of Turbinia being used for the tests does not contain the new methods in the turbinia.client module
aarontp: FYI, a new release of Turbinia  (`20220216`) has been cut as of this afternoon, so I think we just need to update the dftimewolf dependencies appropriately.
ramo-j: I'm aware this PR is still in draft, so apologies if I'm jumping the gun, but I notice that in some of the recipe updates the `turbinia_recipe` option is required, and optional in some. Is this deliberate? Optional across the board makes sense to me, if there's a reasonable default recipe. 
jleaniz: hey @ramo-j thanks for pointing that out. Yeah, still WIP, the turbinia_recipe arg will be optional for the time being, the default recipe has not been modified as of now but this would just give the ability to specifcy a turbinia recipe if desired.
jleaniz: @tomchop I believe some checks are fialing due to using an older Turbinia version. The method in question was introduced in version [20220216](https://github.com/google/turbinia/releases/tag/20220216)

```
************* Module tests.lib.processors.turbinia_gcp
tests/lib/processors/turbinia_gcp.py:56:4: E1101: Instance of 'TurbiniaCeleryClient' has no 'create_request' member (no-member)
```
jleaniz: Merge.

Patch:
diff --git a/data/recipes/aws_turbinia_ts.json b/data/recipes/aws_turbinia_ts.json
index 8ea4134..c6f8248 100644
--- a/data/recipes/aws_turbinia_ts.json
+++ b/data/recipes/aws_turbinia_ts.json
@@ -66,8 +66,8 @@
         "turbinia_config_file": null,
         "disk_names": null,
         "project": "@gcp_project",
+        "turbinia_recipe": "@turbinia_recipe",
         "turbinia_zone": "@turbinia_zone",
-        "run_all_jobs": "@run_all_jobs",
         "sketch_id": "@sketch_id"
       }
     }, {
@@ -92,9 +92,9 @@
     ["--gcp_project", "Destination GCP project.", null],
     ["--aws_profile", "Source AWS profile.", null],
     ["--incident_id", "Incident ID (used for Timesketch description).", null],
-    ["--run_all_jobs", "Run all Turbinia processing jobs instead of a faster subset.", false],
     ["--sketch_id", "Timesketch sketch to which the timeline should be added.", null],
     ["--token_password", "Optional custom password to decrypt Timesketch credential file with.", ""],
+    ["--turbinia_recipe", "The Turbinia recipe name to use for evidence processing.", null],
     ["--turbinia_zone", "Zone Turbinia is located in", "us-central1-f"],
     ["--wait_for_timelines", "Whether to wait for Timesketch to finish processing all timelines.", true]
   ]
diff --git a/data/recipes/gcp_turbinia_disk_copy_ts.json b/data/recipes/gcp_turbinia_disk_copy_ts.json
index 34b581a..aed73c9 100644
--- a/data/recipes/gcp_turbinia_disk_copy_ts.json
+++ b/data/recipes/gcp_turbinia_disk_copy_ts.json
@@ -51,8 +51,8 @@
             "turbinia_config_file": null,
             "disk_names": null,
             "project": "@analysis_project_name",
+            "turbinia_recipe": "@turbinia_recipe",
             "turbinia_zone": "@turbinia_zone",
-            "run_all_jobs": "@run_all_jobs",
             "sketch_id": "@sketch_id"
         }
     }, {
@@ -69,9 +69,9 @@
     "args": [
         ["source_project_name", "Name of the project containing the instance / disks to copy.", null],
         ["analysis_project_name", "Name of the project containing the Turbinia instance.", null],
+        ["--turbinia_recipe", "The Turbinia recipe name to use for evidence processing.", null],
         ["--turbinia_zone", "The GCP zone the disk to process and Turbinia workers are in.", "us-central1-f"],
         ["--incident_id", "Incident ID (used for Timesketch description and to label the VM with).", null],
-        ["--run_all_jobs", "Run all Turbinia processing jobs instead of a faster subset.", false],
         ["--sketch_id", "Timesketch sketch to which the timeline should be added.", null],
         ["--token_password", "Optional custom password to decrypt Timesketch credential file with.", ""],
         ["--create_analysis_vm", "Create an analysis VM in the destination project.", true],
diff --git a/data/recipes/gcp_turbinia_ts.json b/data/recipes/gcp_turbinia_ts.json
index 918f9bd..18302d5 100644
--- a/data/recipes/gcp_turbinia_ts.json
+++ b/data/recipes/gcp_turbinia_ts.json
@@ -16,8 +16,8 @@
             "turbinia_config_file": null,
             "disk_names": "@disk_name",
             "project": "@analysis_project_name",
+            "turbinia_recipe": "@turbinia_recipe",
             "turbinia_zone": "@turbinia_zone",
-            "run_all_jobs": "@run_all_jobs",
             "sketch_id": "@sketch_id"
         }
     }, {
@@ -36,9 +36,9 @@
         ["turbinia_zone", "The GCP zone the disk to process (and Turbinia workers) are in.", null],
         ["disk_names", "Names of GCP persistent disks to process.", null],
         ["--incident_id", "Incident ID (used for Timesketch description).", null],
-        ["--run_all_jobs", "Run all Turbinia processing jobs instead of a faster subset.", false],
         ["--sketch_id", "Timesketch sketch to which the timeline should be added.", null],
         ["--token_password", "Optional custom password to decrypt Timesketch credential file with.", ""],
+        ["--turbinia_recipe", "The Turbinia recipe name to use for evidence processing.", null],
         ["--wait_for_timelines", "Whether to wait for Timesketch to finish processing all timelines.", true]
     ]
 }
diff --git a/data/recipes/upload_turbinia.json b/data/recipes/upload_turbinia.json
index 9430df8..6ee9867 100644
--- a/data/recipes/upload_turbinia.json
+++ b/data/recipes/upload_turbinia.json
@@ -36,10 +36,10 @@
         "args": {
             "turbinia_config_file": "@turbinia_config",
             "project": null,
+            "turbinia_recipe": "@turbinia_recipe",
             "turbinia_zone": "us-central1-f",
             "output_directory": null,
-            "sketch_id": null,
-            "run_all_jobs": false
+            "sketch_id": null
         }
     }, {
           "wants": ["TurbiniaArtifactProcessor"],
@@ -59,6 +59,7 @@
       }],
     "args": [
         ["files", "Paths to process.", null],
+        ["--turbinia_recipe", "The Turbinia recipe name to use for evidence processing.", null],
         ["--destination_turbinia_dir", "Destination path in Turbinia host to write the files to.", null],
         ["--hostname", "Remote host.", null],
         ["--directory", "Directory in which to copy and compress files.", null],
diff --git a/dftimewolf/lib/processors/turbinia_artifact.py b/dftimewolf/lib/processors/turbinia_artifact.py
index 68577f5..c3f5adc 100644
--- a/dftimewolf/lib/processors/turbinia_artifact.py
+++ b/dftimewolf/lib/processors/turbinia_artifact.py
@@ -47,19 +47,19 @@ class TurbiniaArtifactProcessor(TurbiniaProcessorBase,
   def SetUp(self,
             turbinia_config_file: str,
             project: str,
+            turbinia_recipe: str,
             turbinia_zone: str,
             output_directory: str,
-            sketch_id: int,
-            run_all_jobs: bool) -> None:
+            sketch_id: int) -> None:
     """Sets up the object attributes.
 
     Args:
       turbinia_config_file (str): Full path to the Turbinia config file to use.
       project (str): name of the GCP project containing the disk to process.
+      turbinia_recipe (str): Turbinia recipe name.
       turbinia_zone (str): GCP zone in which the Turbinia server is running.
       output_directory (str): Name of the directory to process.
       sketch_id (int): The Timesketch sketch ID.
-      run_all_jobs (bool): Whether to run all jobs instead of a faster subset.
     """
     self.turbinia_config_file = turbinia_config_file
     self.output_directory = output_directory
@@ -68,7 +68,7 @@ class TurbiniaArtifactProcessor(TurbiniaProcessorBase,
       self.logger.success('Turbinia results will be dumped to {0:s}'.format(
           self.output_directory))
     try:
-      self.TurbiniaSetUp(project, turbinia_zone, sketch_id, run_all_jobs)
+      self.TurbiniaSetUp(project, turbinia_recipe, turbinia_zone, sketch_id)
     except TurbiniaException as exception:
       self.ModuleError(str(exception), critical=True)
       return
diff --git a/dftimewolf/lib/processors/turbinia_base.py b/dftimewolf/lib/processors/turbinia_base.py
index d036b42..7369dae 100644
--- a/dftimewolf/lib/processors/turbinia_base.py
+++ b/dftimewolf/lib/processors/turbinia_base.py
@@ -11,7 +11,6 @@ from turbinia import TurbiniaException
 from turbinia import client as turbinia_client
 from turbinia import config as turbinia_config
 from turbinia import evidence, output_manager
-from turbinia.message import TurbiniaRequest
 from dftimewolf.lib.logging_utils import WolfLogger
 
 
@@ -24,9 +23,8 @@ class TurbiniaProcessorBase(object):
     client (TurbiniaClient): Turbinia client.
     instance (str): name of the Turbinia instance
     project (str): name of the GCP project containing the disk to process.
-    run_all_jobs (bool): Whether to run all jobs or to remove slow-running jobs:
-        'StringsJob', 'BinaryExtractorJob', 'BulkExtractorJob', 'PhotorecJob'.
     sketch_id (int): The Timesketch sketch id
+    turbinia_recipe (str): Turbinia recipe name.
     turbinia_region (str): GCP region in which the Turbinia server is running.
     turbinia_zone (str): GCP zone in which the Turbinia server is running.
   """
@@ -45,8 +43,8 @@ class TurbiniaProcessorBase(object):
     self.client = None  # type: turbinia_client.BaseTurbiniaClient
     self.instance = None
     self.project = str()
-    self.run_all_jobs = False
     self.sketch_id = int()
+    self.turbinia_recipe = str()
     self.turbinia_region = None
     self.turbinia_zone = str()
     self.parallel_count = 5  # Arbitrary, used by ThreadAwareModule
@@ -123,9 +121,9 @@ class TurbiniaProcessorBase(object):
 
   def TurbiniaSetUp(self,
                     project: str,
+                    turbinia_recipe: str,
                     turbinia_zone: str,
-                    sketch_id: int,
-                    run_all_jobs: bool) -> None:
+                    sketch_id: int) -> None:
     """Sets up the object attributes.
 
     Raises:
@@ -133,14 +131,14 @@ class TurbiniaProcessorBase(object):
 
     Args:
       project (str): name of the GCP project containing the disk to process.
+      turbinia_recipe (str): Turbinia recipe name.
       turbinia_zone (str): GCP zone in which the Turbinia server is running.
       sketch_id (int): The Timesketch sketch ID.
-      run_all_jobs (bool): Whether to run all jobs instead of a faster subset.
     """
     self.project = project
+    self.turbinia_recipe = turbinia_recipe
     self.turbinia_zone = turbinia_zone
     self.sketch_id = sketch_id
-    self.run_all_jobs = run_all_jobs
 
     turbinia_config.LoadConfig(config_file=self.turbinia_config_file)
     if not self.project:
@@ -150,7 +148,7 @@ class TurbiniaProcessorBase(object):
 
     if not self.project or not self.turbinia_zone:
       raise TurbiniaException(
-        'project or turbinia_zone are not all specified, bailing out')
+          'project or turbinia_zone are not all specified, bailing out')
 
     self.turbinia_region = turbinia_config.TURBINIA_REGION
     self.instance = turbinia_config.INSTANCE_ID
@@ -182,20 +180,33 @@ class TurbiniaProcessorBase(object):
     """
     evidence_.validate()
 
-    request = TurbiniaRequest(requester=getpass.getuser())
-    request.evidence.append(evidence_)
-    if self.sketch_id:
-      request.recipe['globals']['sketch_id'] = self.sketch_id
-    if not self.run_all_jobs:
-      request.recipe['globals']['jobs_denylist'] = [
-          'StringsJob', 'BinaryExtractorJob', 'BulkExtractorJob', 'PhotorecJob']
+    recipe = None
+    jobs_denylist = None
+    yara_text = None
 
-    if threat_intel_indicators:
-      request.recipe['globals']['filter_patterns'] = threat_intel_indicators
+    jobs_denylist = [
+        'StringsJob', 'BinaryExtractorJob', 'BulkExtractorJob', 'PhotorecJob'
+    ]
 
     if yara_rules:
       yara_text = '\n'.join(list(yara_rules))
-      request.recipe['globals']['yara_rules'] = yara_text
+
+    if self.turbinia_recipe:
+      # Use a pre-configured turbinia recipe
+      recipe = self.client.create_recipe(
+          recipe_name=self.turbinia_recipe,
+          sketch_id=self.sketch_id,
+          yara_rules=yara_text)
+    else:
+      # Use default recipe with custom parameters
+      recipe = self.client.create_recipe(
+          sketch_id=self.sketch_id,
+          filter_patterns=threat_intel_indicators,
+          jobs_denylist=jobs_denylist,
+          yara_rules=yara_text)
+
+    request = self.client.create_request(
+        requester=getpass.getuser(), recipe=recipe)
 
     request_dict = {
         'instance': self.instance,
@@ -209,8 +220,9 @@ class TurbiniaProcessorBase(object):
         'Creating Turbinia request {0:s} with Evidence {1!s}'.format(
             request.request_id, evidence_.name))
     self.client.send_request(request)
-    self.logger.info('Waiting for Turbinia request {0:s} to complete'.format(
-        request.request_id))
+    self.logger.info(
+        'Waiting for Turbinia request {0:s} to complete'.format(
+            request.request_id))
 
     # Workaround for rate limiting in turbinia when checking task status
     while True:
diff --git a/dftimewolf/lib/processors/turbinia_gcp.py b/dftimewolf/lib/processors/turbinia_gcp.py
index a73dbd9..b8ceccf 100644
--- a/dftimewolf/lib/processors/turbinia_gcp.py
+++ b/dftimewolf/lib/processors/turbinia_gcp.py
@@ -43,9 +43,9 @@ class TurbiniaGCPProcessor(TurbiniaProcessorBase, module.ThreadAwareModule):
   def SetUp(self,
             turbinia_config_file: str,
             project: str,
+            turbinia_recipe: str,
             turbinia_zone: str,
             sketch_id: int,
-            run_all_jobs: bool,
             disk_names: str = '') -> None:
     """Sets up the object attributes.
 
@@ -53,9 +53,9 @@ class TurbiniaGCPProcessor(TurbiniaProcessorBase, module.ThreadAwareModule):
       turbinia_config_file (str): Full path to the Turbinia config file to use.
       disk_names (str): names of the disks to process.
       project (str): name of the GCP project containing the disk to process.
+      turbinia_recipe (str): Turbinia recipe name.
       turbinia_zone (str): GCP zone in which the Turbinia server is running.
       sketch_id (int): The Timesketch sketch ID.
-      run_all_jobs (bool): Whether to run all jobs instead of a faster subset.
     """
     self.turbinia_config_file = turbinia_config_file
 
@@ -68,7 +68,7 @@ class TurbiniaGCPProcessor(TurbiniaProcessorBase, module.ThreadAwareModule):
             project=project))
 
     try:
-      self.TurbiniaSetUp(project, turbinia_zone, sketch_id, run_all_jobs)
+      self.TurbiniaSetUp(project, turbinia_recipe, turbinia_zone, sketch_id)
     except TurbiniaException as exception:
       self.ModuleError(str(exception), critical=True)
       return


Test patch:
diff --git a/tests/lib/processors/turbinia_artifact.py b/tests/lib/processors/turbinia_artifact.py
index f7ee092..46f0d82 100644
--- a/tests/lib/processors/turbinia_artifact.py
+++ b/tests/lib/processors/turbinia_artifact.py
@@ -22,8 +22,9 @@ from dftimewolf import config
 # Manually set TURBINIA_PROJECT to the value we expect.
 # pylint: disable=wrong-import-position, wrong-import-order
 from turbinia import config as turbinia_config
-turbinia_config.TURBINIA_PROJECT = 'turbinia-project'
+from turbinia import message as turbinia_message
 
+turbinia_config.TURBINIA_PROJECT = 'turbinia-project'
 YARA_RULE = """rule dummy { condition: false }"""
 
 
@@ -45,40 +46,42 @@ class TurbiniaArtifactProcessorTest(unittest.TestCase):
     turbinia_processor.SetUp(
         turbinia_config_file=None,
         project='turbinia-project',
+        turbinia_recipe=None,
         turbinia_zone='europe-west1',
         output_directory='/tmp/outputdir',
-        sketch_id=123,
-        run_all_jobs=False)
+        sketch_id=123)
+    # pylint: disable=line-too-long
+    turbinia_processor.client.create_request.return_value = turbinia_message.TurbiniaRequest()
     self.assertEqual(turbinia_processor.project, 'turbinia-project')
+    self.assertEqual(turbinia_processor.turbinia_recipe, None)
     self.assertEqual(turbinia_processor.turbinia_zone, 'europe-west1')
     self.assertEqual(turbinia_processor.sketch_id, 123)
-    self.assertEqual(turbinia_processor.run_all_jobs, False)
     self.assertEqual(turbinia_processor.output_directory, '/tmp/outputdir')
     self.assertEqual(test_state.errors, [])
 
     # pylint: disable=protected-access
-    six.assertRegex(self, turbinia_processor._output_path,
-                    '(/tmp/tmp|/var/folders).+')
+    six.assertRegex(
+        self, turbinia_processor._output_path, '(/tmp/tmp|/var/folders).+')
 
   @mock.patch('turbinia.client.get_turbinia_client')
   # pylint: disable=invalid-name
-  def testProcess(self,
-                  _mock_TurbiniaClient):
+  def testProcess(self, _mock_TurbiniaClient):
     """Tests that the processor processes data correctly when a GCEDisk is
-    received from the state.
+        received from the state.
     """
     test_state = state.DFTimewolfState(config.Config)
-    test_state.StoreContainer(containers.RemoteFSPath(hostname='remotehost',
-                                                      path='/tmp/file.ext'))
+    test_state.StoreContainer(
+        containers.RemoteFSPath(hostname='remotehost', path='/tmp/file.ext'))
     turbinia_processor = turbinia_artifact.TurbiniaArtifactProcessor(test_state)
     turbinia_processor.SetUp(
         turbinia_config_file=None,
         project='turbinia-project',
+        turbinia_recipe=None,
         turbinia_zone='europe-west1',
         output_directory='/tmp/outputdir',
-        sketch_id=123,
-        run_all_jobs=False)
-
+        sketch_id=123)
+    # pylint: disable=line-too-long
+    turbinia_processor.client.create_request.return_value = turbinia_message.TurbiniaRequest()
     turbinia_processor.client.get_task_data.return_value = [{
         'saved_paths': [
             '/fake/data.plaso',
@@ -100,8 +103,7 @@ class TurbiniaArtifactProcessorTest(unittest.TestCase):
     self.assertEqual(len(conts), 2)
     for c in conts:
       self.assertEqual(c.hostname, 'remotehost')
-      self.assertIn(c.path, ['/fake/data.plaso',
-                             '/fake/data2.plaso'])
+      self.assertIn(c.path, ['/fake/data.plaso', '/fake/data2.plaso'])
 
 
 if __name__ == '__main__':
diff --git a/tests/lib/processors/turbinia_gcp.py b/tests/lib/processors/turbinia_gcp.py
index 395c562..b3db81a 100644
--- a/tests/lib/processors/turbinia_gcp.py
+++ b/tests/lib/processors/turbinia_gcp.py
@@ -24,6 +24,8 @@ from dftimewolf import config
 # Manually set TURBINIA_PROJECT to the value we expect.
 # pylint: disable=wrong-import-position, wrong-import-order
 from turbinia import config as turbinia_config
+from turbinia import message as turbinia_message
+
 turbinia_config.TURBINIA_PROJECT = 'turbinia-project'
 
 YARA_RULE = """rule dummy { condition: false }"""
@@ -48,27 +50,29 @@ class TurbiniaGCPProcessorTest(unittest.TestCase):
         turbinia_config_file=None,
         disk_names='disk-1',
         project='turbinia-project',
+        turbinia_recipe=None,
         turbinia_zone='europe-west1',
-        sketch_id=123,
-        run_all_jobs=False)
+        sketch_id=123)
+    # pylint: disable=line-too-long
+    turbinia_processor.client.create_request.return_value = turbinia_message.TurbiniaRequest()
     self.assertEqual(turbinia_processor.project, 'turbinia-project')
     self.assertEqual(turbinia_processor.turbinia_zone, 'europe-west1')
+    self.assertEqual(turbinia_processor.turbinia_recipe, None)
     self.assertEqual(turbinia_processor.sketch_id, 123)
-    self.assertEqual(turbinia_processor.run_all_jobs, False)
     self.assertEqual(test_state.errors, [])
     self.assertEqual(
         'disk-1',
         test_state.GetContainers(
             turbinia_processor.GetThreadOnContainerType())[0].name)
 
-
     # TURBINIA_REGION is dynamically generated
     # pylint: disable=no-member
-    self.assertEqual(turbinia_processor.turbinia_region,
-                     turbinia_gcp.turbinia_config.TURBINIA_REGION)
+    self.assertEqual(
+        turbinia_processor.turbinia_region,
+        turbinia_gcp.turbinia_config.TURBINIA_REGION)
     # pylint: disable=protected-access
-    six.assertRegex(self, turbinia_processor._output_path,
-                    '(/tmp/tmp|/var/folders).+')
+    six.assertRegex(
+        self, turbinia_processor._output_path, '(/tmp/tmp|/var/folders).+')
 
   @mock.patch('turbinia.client.get_turbinia_client')
   # pylint: disable=invalid-name
@@ -81,18 +85,20 @@ class TurbiniaGCPProcessorTest(unittest.TestCase):
           turbinia_config_file=None,
           disk_names='disk-1',
           project='turbinia-wrong-project',
+          turbinia_recipe=None,
           turbinia_zone='europe-west1',
-          sketch_id=None,
-          run_all_jobs=False)
-
+          sketch_id=None)
+      # pylint: disable=line-too-long
+      turbinia_processor.client.create_request.return_value = turbinia_message.TurbiniaRequest()
     self.assertEqual(len(test_state.errors), 1)
     self.assertEqual(test_state.errors[0], error.exception)
     error_msg = error.exception.message
-    self.assertEqual(error_msg, 'Specified project turbinia-wrong-project does '
-                                'not match Turbinia configured project '
-                                'turbinia-project. Use '
-                                'gcp_turbinia_disk_copy_ts recipe to copy the '
-                                'disk into the same project.')
+    self.assertEqual(
+        error_msg, 'Specified project turbinia-wrong-project does '
+        'not match Turbinia configured project '
+        'turbinia-project. Use '
+        'gcp_turbinia_disk_copy_ts recipe to copy the '
+        'disk into the same project.')
     self.assertTrue(error.exception.critical)
 
   @mock.patch('dftimewolf.lib.processors.turbinia_base.turbinia_config')
@@ -100,26 +106,24 @@ class TurbiniaGCPProcessorTest(unittest.TestCase):
   # pylint: disable=invalid-name
   def testWrongSetup(self, _mock_TurbiniaClient, mock_turbinia_config):
     """Tests that invalid setup options generate errors."""
-    params = [
-        {
-            'turbinia_config_file': None,
-            'disk_names': 'disk-1',
-            'project': None,
-            'turbinia_zone': 'europe-west1',
-            'sketch_id': None,
-            'run_all_jobs': False,
-        },
-        {
-            'turbinia_config_file': None,
-            'disk_names': 'disk-1',
-            'project': 'turbinia-project',
-            'turbinia_zone': None,
-            'sketch_id': None,
-            'run_all_jobs': False
-        }
-    ]
-    expected_error = ('project or turbinia_zone are not all '
-                      'specified, bailing out')
+    params = [{
+        'turbinia_config_file': None,
+        'disk_names': 'disk-1',
+        'project': None,
+        'turbinia_recipe': None,
+        'turbinia_zone': 'europe-west1',
+        'sketch_id': None,
+    }, {
+        'turbinia_config_file': None,
+        'disk_names': 'disk-1',
+        'project': 'turbinia-project',
+        'turbinia_recipe': None,
+        'turbinia_zone': None,
+        'sketch_id': None,
+    }]
+    expected_error = (
+        'project or turbinia_zone are not all '
+        'specified, bailing out')
 
     for combination in params:
       mock_turbinia_config.TURBINIA_PROJECT = combination['project']
@@ -140,28 +144,40 @@ class TurbiniaGCPProcessorTest(unittest.TestCase):
   @mock.patch('turbinia.evidence.GoogleCloudDisk')
   @mock.patch('turbinia.client.get_turbinia_client')
   # pylint: disable=invalid-name
-  def testProcessFromParams(self,
-                  _mock_TurbiniaClient,
-                  mock_GoogleCloudDisk,
-                  mock_GCSOutputWriter,
-                  mock_exists):
+  def testProcessFromParams(
+      self, _mock_TurbiniaClient, mock_GoogleCloudDisk, mock_GCSOutputWriter,
+      mock_exists):
     """Tests that the processor processes data correctly when a disk name is
     passed in as a parameter."""
 
     test_state = state.DFTimewolfState(config.Config)
     test_state.StoreContainer(
-      containers.YaraRule(
-        name='dummy_yara', rule_text="rule dummy { condition: false }")
-    )
+        containers.YaraRule(
+            name='dummy_yara', rule_text="rule dummy { condition: false }"))
     turbinia_processor = turbinia_gcp.TurbiniaGCPProcessor(test_state)
     turbinia_processor.SetUp(
         turbinia_config_file=None,
         disk_names='disk-1',
         project='turbinia-project',
+        turbinia_recipe=None,
         turbinia_zone='europe-west1',
-        sketch_id=4567,
-        run_all_jobs=False)
-
+        sketch_id=4567)
+
+    recipe = {
+        'globals': {
+            'sketch_id':
+                4567,
+            'yara_rules':
+                YARA_RULE,
+            'jobs_denylist': [
+                'StringsJob', 'BinaryExtractorJob', 'BulkExtractorJob',
+                'PhotorecJob'
+            ]
+        }
+    }
+    # pylint: disable=line-too-long
+    turbinia_processor.client.create_request.return_value = turbinia_message.TurbiniaRequest(
+        recipe=recipe)
     turbinia_processor.client.get_task_data.return_value = [{
         'saved_paths': [
             '/fake/data.plaso',
@@ -187,9 +203,7 @@ class TurbiniaGCPProcessorTest(unittest.TestCase):
     turbinia_processor.PostProcess()
 
     mock_GoogleCloudDisk.assert_called_with(
-        disk_name='disk-1',
-        project='turbinia-project',
-        zone='europe-west1')
+        disk_name='disk-1', project='turbinia-project', zone='europe-west1')
 
     # These are mock classes, so there is a member
     # pylint: disable=no-member
@@ -201,15 +215,13 @@ class TurbiniaGCPProcessorTest(unittest.TestCase):
         ['StringsJob', 'BinaryExtractorJob', 'BulkExtractorJob', 'PhotorecJob'])
     turbinia_processor.client.get_task_data.assert_called()
     self.assertEqual(
-      request.recipe['globals']['yara_rules'],
-      "rule dummy { condition: false }"
-    )
+        request.recipe['globals']['yara_rules'],
+        "rule dummy { condition: false }")
 
     # pylint: disable=protected-access
     mock_GCSOutputWriter.assert_any_call(
         'gs://BinaryExtractorTask.tar.gz',
-        local_output_dir=turbinia_processor._output_path
-    )
+        local_output_dir=turbinia_processor._output_path)
     self.assertEqual(test_state.errors, [])
     ti_containers = test_state.GetContainers(containers.ThreatIntelligence)
     file_containers = test_state.GetContainers(containers.File)
@@ -230,11 +242,9 @@ class TurbiniaGCPProcessorTest(unittest.TestCase):
   @mock.patch('turbinia.evidence.GoogleCloudDisk')
   @mock.patch('turbinia.client.get_turbinia_client')
   # pylint: disable=invalid-name
-  def testProcessFromState(self,
-                  _mock_TurbiniaClient,
-                  mock_GoogleCloudDisk,
-                  mock_GCSOutputWriter,
-                  mock_exists):
+  def testProcessFromState(
+      self, _mock_TurbiniaClient, mock_GoogleCloudDisk, mock_GCSOutputWriter,
+      mock_exists):
     """Tests that the processor processes data correctly when a GCEDisk is
     received from the state.
     """
@@ -249,10 +259,25 @@ class TurbiniaGCPProcessorTest(unittest.TestCase):
     turbinia_processor.SetUp(
         turbinia_config_file=None,
         project='turbinia-project',
+        turbinia_recipe=None,
         turbinia_zone='europe-west1',
-        sketch_id=4567,
-        run_all_jobs=False)
-
+        sketch_id=4567)
+
+    recipe = {
+        'globals': {
+            'sketch_id':
+                4567,
+            'yara_rules':
+                YARA_RULE,
+            'jobs_denylist': [
+                'StringsJob', 'BinaryExtractorJob', 'BulkExtractorJob',
+                'PhotorecJob'
+            ]
+        }
+    }
+    # pylint: disable=line-too-long
+    turbinia_processor.client.create_request.return_value = turbinia_message.TurbiniaRequest(
+        recipe=recipe)
     turbinia_processor.client.get_task_data.return_value = [{
         'saved_paths': [
             '/fake/data.plaso',
@@ -278,9 +303,7 @@ class TurbiniaGCPProcessorTest(unittest.TestCase):
     turbinia_processor.PostProcess()
 
     mock_GoogleCloudDisk.assert_called_with(
-        disk_name='disk-1',
-        project='turbinia-project',
-        zone='europe-west1')
+        disk_name='disk-1', project='turbinia-project', zone='europe-west1')
 
     # These are mock classes, so there is a member
     # pylint: disable=no-member
@@ -292,15 +315,13 @@ class TurbiniaGCPProcessorTest(unittest.TestCase):
         ['StringsJob', 'BinaryExtractorJob', 'BulkExtractorJob', 'PhotorecJob'])
     turbinia_processor.client.get_task_data.assert_called()
     self.assertEqual(
-      request.recipe['globals']['yara_rules'],
-      "rule dummy { condition: false }"
-    )
+        request.recipe['globals']['yara_rules'],
+        "rule dummy { condition: false }")
 
     # pylint: disable=protected-access
     mock_GCSOutputWriter.assert_any_call(
         'gs://BinaryExtractorTask.tar.gz',
-        local_output_dir=turbinia_processor._output_path
-    )
+        local_output_dir=turbinia_processor._output_path)
     self.assertEqual(test_state.errors, [])
     ti_containers = test_state.GetContainers(containers.ThreatIntelligence)
     file_containers = test_state.GetContainers(containers.File)
@@ -331,8 +352,8 @@ class TurbiniaGCPProcessorTest(unittest.TestCase):
         turbinia_config_file=None,
         project='turbinia-project',
         turbinia_zone='europe-west1',
-        sketch_id=4567,
-        run_all_jobs=False)
+        turbinia_recipe=None,
+        sketch_id=4567)
 
     turbinia_processor.PreProcess()
     in_containers = test_state.GetContainers(
@@ -352,6 +373,7 @@ class TurbiniaGCPProcessorTest(unittest.TestCase):
   # pylint: disable=invalid-name
   def testDownloadFilesFromGCS(self, mock_GCSOutputWriter):
     """Tests _DownloadFilesFromGCS"""
+
     def _fake_copy(filename):
       return '/fake/local/' + filename.rsplit('/')[-1]
 
@@ -361,10 +383,9 @@ class TurbiniaGCPProcessorTest(unittest.TestCase):
     fake_paths = ['gs://hashes.json', 'gs://results.plaso']
     # pylint: disable=protected-access
     local_paths = turbinia_processor._DownloadFilesFromGCS('fake', fake_paths)
-    self.assertEqual(local_paths, [
-        ('fake', '/fake/local/hashes.json'),
-        ('fake', '/fake/local/results.plaso')
-    ])
+    self.assertEqual(
+        local_paths, [('fake', '/fake/local/hashes.json'),
+                      ('fake', '/fake/local/results.plaso')])
 
   def testDeterminePaths(self):
     """Tests _DeterminePaths"""
@@ -389,21 +410,24 @@ class TurbiniaGCPProcessorTest(unittest.TestCase):
     We store a ForensicsVM container in the state, run PreProcess, then check
     that the state contains a GCEDisk as expected."""
     test_state = state.DFTimewolfState(config.Config)
-    test_state.StoreContainer(containers.ForensicsVM(
-         name='ForensicsVM',
-         evidence_disk=GoogleComputeDisk(
-           name='disk-1',
-           project_id='turbinia-project',
-           zone='europe-west1'),
-         platform = 'gcp'))
+    test_state.StoreContainer(
+        containers.ForensicsVM(
+            name='ForensicsVM',
+            evidence_disk=GoogleComputeDisk(
+                name='disk-1',
+                project_id='turbinia-project',
+                zone='europe-west1'),
+            platform='gcp'))
 
     turbinia_processor = turbinia_gcp.TurbiniaGCPProcessor(test_state)
     turbinia_processor.SetUp(
         turbinia_config_file=None,
         project='turbinia-project',
+        turbinia_recipe=None,
         turbinia_zone='europe-west1',
-        sketch_id=4567,
-        run_all_jobs=False)
+        sketch_id=4567)
+    # pylint: disable=line-too-long
+    turbinia_processor.client.create_request.return_value = turbinia_message.TurbiniaRequest()
     turbinia_processor.PreProcess()
 
     out_containers = test_state.GetContainers(containers.GCEDiskEvidence)