GCS Output Plugin for FileFinderResults (#1128)

Author: daschwanden

.gitignore

@@ -33,3 +33,6 @@ installers/
 GRRlog.txt
 grr/server/grr_response_server/gui/static/third-party/
 grr/server/grr_response_server/gui/ui/.angular/
+docker_config_files/*.pem
+compose.watch.yaml
+Dockerfile.client

compose.yaml

@@ -145,6 +145,7 @@ services:
     container_name: grr-worker
     volumes:
       - ./docker_config_files:/configs
+      - ${HOME}/.config/gcloud:/root/.config/gcloud
     hostname: grr-worker
     depends_on:
       db:

grr/proto/grr_response_proto/output_plugin.proto

@@ -104,3 +104,13 @@ message ElasticsearchOutputPluginArgs {
     description: "The Elasticsearch index to place the events in"
   }];
 }
+
+message GcsOutputPluginArgs {
+  optional string gcs_bucket = 1 [(sem_type) = {
+    description: "The GCS bucket to which objects will be stored.",
+  }];
+  optional string project_id = 2 [(sem_type) = {
+    description: "The Project ID with the GCS bucket.",
+  }];
+}
+

grr/server/grr_response_server/output_plugins/__init__.py

@@ -15,3 +15,4 @@
 from grr_response_server.output_plugins import splunk_plugin
 from grr_response_server.output_plugins import sqlite_plugin
 from grr_response_server.output_plugins import yaml_plugin
+from grr_response_server.output_plugins import gcs_plugin

grr/server/grr_response_server/output_plugins/gcs_plugin.py

@@ -0,0 +1,72 @@
+#!/usr/bin/env python
+"""GCS output plugin for FileFinder flow results."""
+
+import jinja2
+import json
+import logging
+
+from typing import Any, Dict
+
+from google.cloud import storage
+from google.protobuf import json_format
+from grr_response_core import config
+from grr_response_core.lib import rdfvalue
+from grr_response_core.lib import utils
+from grr_response_core.lib.rdfvalues import standard as rdf_standard
+from grr_response_core.lib.rdfvalues import structs as rdf_structs
+from grr_response_proto import output_plugin_pb2
+from grr_response_server import data_store
+from grr_response_server import file_store
+from grr_response_server import output_plugin
+from grr_response_server.databases import db
+
+JsonDict = Dict[str, Any]
+
+class GcsOutputPluginArgs(rdf_structs.RDFProtoStruct):
+  protobuf = output_plugin_pb2.GcsOutputPluginArgs
+
+
+class GcsOutputPlugin(output_plugin.OutputPlugin):
+  """An output plugin that sends the object to GCS for each response received."""
+  name = "GCS Bucket"
+  description = "Send to GCS for each result."
+  args_type = GcsOutputPluginArgs
+  produces_output_streams = False
+  uploaded_files = 0
+
+
+  def UploadBlobFromStream(self, project_id, bucket_name, client_path, client_id, flow_id, destination_blob_name):
+    """Uploads bytes from a stream to a blob"""
+    storage_client = storage.Client(project=project_id)
+    bucket = storage_client.bucket(bucket_name)
+    blob = bucket.blob(client_id+"/"+flow_id+destination_blob_name)
+    fd = file_store.OpenFile(client_path)
+    fd.seek(0)
+    blob.upload_from_file(fd)
+
+  def ProcessResponse(self, state, response):
+    """Sends objects to GCS for each response."""
+
+    client_id = response.client_id
+    flow_id = response.flow_id
+
+    if response.HasField("payload"):
+        if response.payload.HasField("transferred_file"):
+            if response.payload.stat_entry.st_size > 0 :
+                client_path = db.ClientPath.FromPathSpec(client_id, response.payload.stat_entry.pathspec)
+                self.UploadBlobFromStream(self.args.project_id, self.args.gcs_bucket, client_path, client_id, flow_id, response.payload.stat_entry.pathspec.path)
+                logging.info("response client_id: %s, flow_id: %s, transferred_file: %s", client_id, flow_id, response.payload.stat_entry.pathspec.path)
+                self.uploaded_files += 1
+            else:
+                logging.warning("%s : file size is 0, nothing to push to GCS", response.payload.stat_entry.pathspec.path)
+
+
+  def ProcessResponses(self, state, responses):
+    if self.args.gcs_bucket == "" or self.args.project_id == "":
+        logging.error("both GCS bucket and Project ID must be set")
+        return
+    else:
+        logging.info("Project ID: %s, GCS bucket: %s", self.args.project_id, self.args.gcs_bucket)
+
+    for response in responses:
+      self.ProcessResponse(state, response)

grr/server/grr_response_server/output_plugins/gcs_plugin_test.py

@@ -0,0 +1,159 @@
+#!/usr/bin/env python
+"""Tests for GCS output plugin."""
+
+import json
+import requests
+
+from absl import app
+
+from unittest import mock
+from unittest.mock import MagicMock
+
+from grr_response_core.lib import rdfvalue
+from grr_response_core.lib.rdfvalues import client as rdf_client
+from grr_response_core.lib.rdfvalues import client_fs as rdf_client_fs
+from grr_response_core.lib.rdfvalues import file_finder as rdf_file_finder
+from grr_response_core.lib.rdfvalues import paths as rdf_paths
+
+from grr_response_server.output_plugins import gcs_plugin
+from grr_response_server.rdfvalues import flow_objects as rdf_flow_objects
+
+from grr.test_lib import flow_test_lib
+from grr.test_lib import test_lib
+
+
+class GcsOutputPluginTest(flow_test_lib.FlowTestsBaseclass):
+  """Tests GCS output plugin."""
+
+  def setUp(self):
+    super().setUp()
+
+    self.client_id = self.SetupClient(0)
+    self.flow_id = '12345678'
+    self.source_id = (
+        rdf_client.ClientURN(self.client_id)
+        .Add('Results')
+        .RelativeName('aff4:/')
+    )
+
+
+  def _CallPlugin(self, plugin_args=None, responses=None):
+    plugin_cls = gcs_plugin.GcsOutputPlugin
+    plugin, plugin_state = plugin_cls.CreatePluginAndDefaultState(
+        source_urn=self.source_id, args=plugin_args
+    )
+
+    plugin_cls.UploadBlobFromStream = MagicMock()
+
+    messages = []
+    for response in responses:
+      messages.append(
+          rdf_flow_objects.FlowResult(
+              client_id=self.client_id, flow_id=self.flow_id, payload=response
+          )
+      )
+
+    with test_lib.FakeTime(1445995873):
+        plugin.ProcessResponses(plugin_state, messages)
+        plugin.Flush(plugin_state)
+        plugin.UpdateState(plugin_state)
+
+    return plugin.uploaded_files
+
+
+  def testClientFileFinderResponseUploaded(self):
+    rdf_payload = rdf_file_finder.FileFinderResult()
+    rdf_payload.stat_entry.st_size = 1234
+    rdf_payload.stat_entry.pathspec.path = ("/var/log/test.log")
+    rdf_payload.stat_entry.pathspec.pathtype = "OS"
+    rdf_payload.transferred_file = rdf_client_fs.BlobImageDescriptor(chunks=[])
+
+    with test_lib.FakeTime(rdfvalue.RDFDatetime.FromSecondsSinceEpoch(15)):
+        uploaded_files = self._CallPlugin(
+            plugin_args=gcs_plugin.GcsOutputPluginArgs(
+              project_id="test-project-id",
+              gcs_bucket="text-gcs-bucket"
+            ),
+            responses=[rdf_payload],
+        )
+
+    self.assertEqual(uploaded_files, 1)
+
+
+  def testClientFileFinderResponseNoProjectId(self):
+    rdf_payload = rdf_file_finder.FileFinderResult()
+    rdf_payload.stat_entry.st_size = 1234
+    rdf_payload.stat_entry.pathspec.path = ("/var/log/test.log")
+    rdf_payload.stat_entry.pathspec.pathtype = "OS"
+    rdf_payload.transferred_file = rdf_client_fs.BlobImageDescriptor(chunks=[])
+
+    with test_lib.FakeTime(rdfvalue.RDFDatetime.FromSecondsSinceEpoch(15)):
+        uploaded_files = self._CallPlugin(
+            plugin_args=gcs_plugin.GcsOutputPluginArgs(
+              project_id="",
+              gcs_bucket="text-gcs-bucket"
+            ),
+            responses=[rdf_payload],
+        )
+
+    self.assertEqual(uploaded_files, 0)
+
+
+  def testClientFileFinderResponseNoGcsBucket(self):
+    rdf_payload = rdf_file_finder.FileFinderResult()
+    rdf_payload.stat_entry.st_size = 1234
+    rdf_payload.stat_entry.pathspec.path = ("/var/log/test.log")
+    rdf_payload.stat_entry.pathspec.pathtype = "OS"
+    rdf_payload.transferred_file = rdf_client_fs.BlobImageDescriptor(chunks=[])
+
+    with test_lib.FakeTime(rdfvalue.RDFDatetime.FromSecondsSinceEpoch(15)):
+        uploaded_files = self._CallPlugin(
+            plugin_args=gcs_plugin.GcsOutputPluginArgs(
+              project_id="test-project-id",
+              gcs_bucket=""
+            ),
+            responses=[rdf_payload],
+        )
+
+    self.assertEqual(uploaded_files, 0)
+
+
+  def testClientFileFinderResponseEmptyFile(self):
+    rdf_payload = rdf_file_finder.FileFinderResult()
+    rdf_payload.stat_entry.st_size = 0
+    rdf_payload.stat_entry.pathspec.path = ("/var/log/test.log")
+    rdf_payload.stat_entry.pathspec.pathtype = "OS"
+    rdf_payload.transferred_file = rdf_client_fs.BlobImageDescriptor(chunks=[])
+
+    with test_lib.FakeTime(rdfvalue.RDFDatetime.FromSecondsSinceEpoch(15)):
+        uploaded_files = self._CallPlugin(
+            plugin_args=gcs_plugin.GcsOutputPluginArgs(
+              project_id="test-project-id",
+              gcs_bucket="text-gcs-bucket"
+            ),
+            responses=[rdf_payload],
+        )
+
+    self.assertEqual(uploaded_files, 0)
+
+
+  def testClientFileFinderResponseNoTransferredFile(self):
+    rdf_payload = rdf_file_finder.FileFinderResult()
+    rdf_payload.stat_entry.st_size = 1234
+    rdf_payload.stat_entry.pathspec.path = ("/var/log/test.log")
+    rdf_payload.stat_entry.pathspec.pathtype = "OS"
+
+    with test_lib.FakeTime(rdfvalue.RDFDatetime.FromSecondsSinceEpoch(15)):
+        uploaded_files = self._CallPlugin(
+            plugin_args=gcs_plugin.GcsOutputPluginArgs(
+              project_id="test-project-id",
+              gcs_bucket="text-gcs-bucket"
+            ),
+            responses=[rdf_payload],
+        )
+
+    self.assertEqual(uploaded_files, 0)
+
+
+if __name__ == '__main__':
+  app.run(test_lib.main)