Repository: google/turbinia
Pull request #984 (2022-01-28 06:13:15)

Changes to CI configuration to use Ubuntu 20.04

joachimmetz: This is failing because dfImageTools pip install pulls in more recent pre-release of dfVFS. Changes to install dfImageTools from the GIFT PPA https://github.com/google/turbinia/pull/985
joachimmetz: @hacktobeer Hold off with merging this one for now, need to get a more recent version of dfimagetools into GIFT PPA stable first
joachimmetz: @aarontp PR is ready for review
joachimmetz: @aarontp if you want stable to pass, we'll have to wait until the next plaso release
aarontp: I think the changes to the `file_system_timeline.py` have been updated in https://github.com/google/turbinia/pull/1051/files .  Should we remove the changes to that file from this PR and keep just the changes to `actions.yaml` (and update the description as appropriate)?
joachimmetz: would need to take a look when time permits
joachimmetz: @aarontp removed the changes to file_system_timeline would need to take a look of them if the issue in https://github.com/google/turbinia/issues/983 has been addressed 

Patch:
diff --git a/.github/workflows/actions.yml b/.github/workflows/actions.yml
index abd0138..a4c722a 100644
--- a/.github/workflows/actions.yml
+++ b/.github/workflows/actions.yml
@@ -6,14 +6,21 @@ jobs:
   install-n-test:
     name: Install Turbinia dependencies and run tests.
     runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        include:
+        - docker_base_image: 'ubuntu:20.04'
+          gift_ppa_track: 'staging'
+        - docker_base_image: 'ubuntu:20.04'
+          gift_ppa_track: 'stable'
     steps:
       - uses: actions/checkout@v2
       - name: Pull latest docker image for cache
         run: |
-          docker pull ubuntu:18.04
+          docker pull ${{ matrix.docker_base_image }}
           docker pull us-docker.pkg.dev/osdfir-registry/turbinia/release/turbinia-worker-dev:latest
       - name: Build Turbinia worker docker image
-        run: docker build --cache-from=ubuntu:18.04,us-docker.pkg.dev/osdfir-registry/turbinia/release/turbinia-worker-dev:latest -t turbinia-worker-dev -f docker/worker/Dockerfile .
+        run: docker build --build-arg PPA_TRACK=${{ matrix.gift_ppa_track }} --cache-from=${{ matrix.docker_base_image }},us-docker.pkg.dev/osdfir-registry/turbinia/release/turbinia-worker-dev:latest -t turbinia-worker-dev -f docker/worker/Dockerfile .
       - name: Run test (turbinia-worker) container
         run: |
           docker run --name turbinia-worker --entrypoint "/bin/bash" -it -d -t turbinia-worker-dev:latest
diff --git a/tools/gcf_init/deploy_gcf.py b/tools/gcf_init/deploy_gcf.py
index f0c009b..a4afe82 100755
--- a/tools/gcf_init/deploy_gcf.py
+++ b/tools/gcf_init/deploy_gcf.py
@@ -5,6 +5,7 @@
 from __future__ import print_function
 
 import subprocess
+import os
 import sys
 
 from turbinia import config
@@ -17,18 +18,19 @@ else:
   function_names = ['gettasks', 'closetasks']
 
 config.LoadConfig()
+current_dir = os.path.dirname(os.path.realpath(__file__))
 
 for cloud_function in function_names:
   print('Deploying function {0:s}'.format(cloud_function))
   cmd = (
       'gcloud --project {0:s} functions deploy {1:s} --stage-bucket {2:s} '
       '--region {3:s} --runtime nodejs14 --trigger-http --memory 256MB '
-      '--timeout 60s'.format(
+      '--timeout 60s --source {4:s}'.format(
           config.TURBINIA_PROJECT, cloud_function, config.BUCKET_NAME,
-          config.TURBINIA_REGION))
+          config.TURBINIA_REGION, current_dir))
   print(subprocess.check_call(cmd, shell=True))
 
 print('/nCreating Datastore index from {0:s}'.format(index_file))
 cmd = 'gcloud --quiet --project {0:s} datastore indexes create {1:s}'.format(
-    config.TURBINIA_PROJECT, index_file)
+    config.TURBINIA_PROJECT, '{0:s}/{1:s}'.format(current_dir, index_file))
 subprocess.check_call(cmd, shell=True)
diff --git a/turbinia/pubsub.py b/turbinia/pubsub.py
index c2f2023..5fae4b2 100644
--- a/turbinia/pubsub.py
+++ b/turbinia/pubsub.py
@@ -16,6 +16,7 @@
 
 from __future__ import unicode_literals
 
+import base64
 import codecs
 import logging
 
@@ -24,8 +25,11 @@ from six.moves import xrange
 
 from google.cloud import exceptions
 from google.cloud import pubsub
+from googleapiclient.errors import HttpError
+import libcloudforensics.providers.gcp.internal.common as gcp_common
 
 from turbinia import config
+from turbinia import TurbiniaException
 from turbinia.message import TurbiniaMessageBase
 
 log = logging.getLogger('turbinia')
@@ -36,7 +40,7 @@ class TurbiniaPubSub(TurbiniaMessageBase):
 
   Attributes:
     _queue: A Queue object for storing pubsub messages
-    publisher: The pubsub publisher client object
+    pubsub_api_client: The pubsub API client object
     subscriber: The pubsub subscriber client object
     subscription: The pubsub subscription object
     topic_name (str): The pubsub topic name
@@ -46,7 +50,7 @@ class TurbiniaPubSub(TurbiniaMessageBase):
   def __init__(self, topic_name):
     """Initialization for PubSubClient."""
     self._queue = queue.Queue()
-    self.publisher = None
+    self.pubsub_api_client = None
     self.subscriber = None
     self.subscription = None
     self.topic_name = topic_name
@@ -60,14 +64,26 @@ class TurbiniaPubSub(TurbiniaMessageBase):
   def setup_publisher(self):
     """Set up the pubsub publisher."""
     config.LoadConfig()
-    self.publisher = pubsub.PublisherClient()
-    self.topic_path = self.publisher.topic_path(
+    # the publisher we will use the pubsub client in googleapiclient.discovery
+    # for more information on using the APIs, see
+    # https://cloud.google.com/pubsub/docs/reference/rest
+    self.pubsub_api_client = gcp_common.CreateService('pubsub', 'v1')
+    self.topic_path = 'projects/{0:s}/topics/{1:s}'.format(
         config.TURBINIA_PROJECT, self.topic_name)
     try:
       log.debug('Trying to create pubsub topic {0:s}'.format(self.topic_path))
-      self.publisher.create_topic(self.topic_path)
-    except exceptions.Conflict:
-      log.debug('PubSub topic {0:s} already exists.'.format(self.topic_path))
+      topics_client = self.pubsub_api_client.projects().topics()
+      # the ExecuteRequest takes API URI, method name as string and parameters
+      # as a dict, it executes the API call, handles paging and return response.
+      gcp_common.ExecuteRequest(
+          topics_client, 'create', {'name': self.topic_path})
+    except HttpError as exception:
+      if exception.resp.status == 409:
+        log.debug('PubSub topic {0:s} already exists.'.format(self.topic_path))
+      else:
+        raise TurbiniaException(
+            'Unknown error occurred when creating Topic:'
+            ' {0!s}'.format(exception), __name__) from exception
     log.debug('Setup PubSub publisher at {0:s}'.format(self.topic_path))
 
   def setup_subscriber(self):
@@ -127,9 +143,25 @@ class TurbiniaPubSub(TurbiniaMessageBase):
 
     message: The message to send.
     """
-    data = message.encode('utf-8')
-    future = self.publisher.publish(self.topic_path, data)
-    msg_id = future.result()
+    base64_data = base64.b64encode(message.encode('utf-8'))
+    request_body = {
+        "messages": [{
+            "data":
+                base64_data.decode('utf-8')  # base64 encoded string
+        }]
+    }
+    publish_client = self.pubsub_api_client.projects().topics()
+    response = gcp_common.ExecuteRequest(
+        publish_client, 'publish', {
+            'topic': self.topic_path,
+            'body': request_body
+        })
+    # Safe to unpack since response is unpaged.
+    if not response[0]['messageIds']:
+      raise TurbiniaException(
+          'Message {0:s} was not published to topic {1:s}'.format(
+              message, self.topic_path))
+    msg_id = response[0]['messageIds'][0]
     log.info(
         'Published message {0!s} to topic {1!s}'.format(
             msg_id, self.topic_name))


Test patch:
diff --git a/turbinia/pubsub_test.py b/turbinia/pubsub_test.py
index bc8a185..360b0ad 100644
--- a/turbinia/pubsub_test.py
+++ b/turbinia/pubsub_test.py
@@ -18,6 +18,7 @@ from __future__ import unicode_literals
 
 import unittest
 
+import base64
 import mock
 
 from six.moves import queue
@@ -50,6 +51,22 @@ class MockPubSubMessage:
     self.message_id = message_id
 
 
+class MockPubSubProject:
+  """This is a mock of a PubSub project API object."""
+
+  def topics(cls):
+    "Mock class method topics."
+    return 'fake-topic-api-object'
+
+
+class MockPubSubClient:
+  """This is a mock of PubSub API client."""
+
+  def projects(cls):
+    "Mock class method projects."
+    return MockPubSubProject()
+
+
 class TestTurbiniaRequest(unittest.TestCase):
   """Test TurbiniaRequest class."""
 
@@ -120,12 +137,23 @@ class TestTurbiniaPubSub(unittest.TestCase):
 
     self.assertListEqual(self.pubsub.check_messages(), [])
 
-  def testSendMessage(self):
+  @mock.patch('libcloudforensics.providers.gcp.internal.common.ExecuteRequest')
+  def testSendMessage(self, gcp_execute):
     """Test sending a message."""
-    self.pubsub.publisher = mock.MagicMock()
+    gcp_execute.return_value = [{'messageIds': ['fake-id']}]
+    gcp_execute.side_effect = None
+    self.pubsub.pubsub_api_client = MockPubSubClient()
     self.pubsub.send_message('test message text')
-    self.pubsub.publisher.publish.assert_called_with(
-        'faketopicpath', b'test message text')
+    base64_data = base64.b64encode('test message text'.encode('utf-8'))
+    gcp_execute.assert_called_with(
+        'fake-topic-api-object', 'publish', {
+            'topic': 'faketopicpath',
+            'body': {
+                'messages': [{
+                    'data': base64_data.decode('utf-8')
+                }]
+            }
+        })
 
 
 class TestTurbiniaKombu(unittest.TestCase):