Added import stream object for more flexible data upload

Added import stream object for more flexible data upload

README.md

@@ -28,6 +28,7 @@ Timesketch is an open source tool for collaborative forensic timeline analysis.
 * [Create timeline from JSON/JSONL/CSV file](docs/CreateTimelineFromJSONorCSV.md)
 * [Create timeline from Plaso file](docs/CreateTimelineFromPlaso.md)
 * [Enable Plaso upload via HTTP](docs/EnablePlasoUpload.md)
+* [Use the API client to upload data](docs/UploadDataViaAPI.md)
 * [Import other data](docs/CreateTimeLineFromOtherData.md)
 
 #### Using Timesketch

api_client/python/setup.py

@@ -21,7 +21,7 @@
 
 setup(
     name='timesketch-api-client',
-    version='20191018',
+    version='20191028',
     description='Timesketch API client',
     license='Apache License, Version 2.0',
     url='http://www.timesketch.org/',
@@ -39,5 +39,6 @@
     install_requires=frozenset([
         'pandas',
         'requests',
+        'xlrd',
         'beautifulsoup4']),
     )

api_client/python/timesketch_api_client/client.py

@@ -26,6 +26,7 @@
 import altair
 import pandas
 from .definitions import HTTP_STATUS_CODE_20X
+from . import importer
 
 
 class TimesketchApi(object):
@@ -395,7 +396,7 @@ def list_aggregations(self):
         """
         data = self.lazyload_data()
         aggregations = []
-        for aggregation in data['aggregations']:
+        for aggregation in data.get('aggregations', []):
             aggregation_obj = Aggregation(
                 sketch=self, api=self.api)
             aggregation_obj.from_store(aggregation_id=aggregation['id'])
@@ -413,7 +414,7 @@ def get_aggregation(self, aggregation_id):
             otherwise None object.
         """
         sketch = self.lazyload_data()
-        for aggregation in sketch['aggregations']:
+        for aggregation in sketch.get('aggregations', []):
             if aggregation['id'] != aggregation_id:
                 continue
             aggregation_obj = Aggregation(sketch=self, api=self.api)
@@ -503,6 +504,70 @@ def upload(self, timeline_name, file_path):
             searchindex=timeline['searchindex']['index_name'])
         return timeline_obj
 
+    def upload_data_frame(
+            self, data_frame, timeline_name, format_message_string=''):
+        """Upload a data frame to the server for indexing.
+
+        In order for a data frame to be uploaded to Timesketch it requires the
+        following columns:
+            + message
+            + datetime
+            + timestamp_desc
+
+        See more details here: https://github.com/google/timesketch/blob/\
+            master/docs/CreateTimelineFromJSONorCSV.md
+
+        Args:
+            data_frame: the pandas dataframe object containing the data to
+                upload.
+            timeline_name: the name of the timeline in Timesketch.
+            format_message_string: optional format string that will be used
+                to generate a message column to the data frame. An example
+                would be: '{src_ip:s} to {dst_ip:s}, {bytes:d} bytes
+                transferred'. Each variable name in the format strings maps
+                to column names in the data frame.
+
+        Returns:
+            A string with the upload status.
+
+        Raises:
+            ValueError: if the dataframe cannot be uploaded to Timesketch.
+        """
+        # TODO: Explore the option of supporting YAML files for loading
+        # configs for formatting strings. (this would be implemented
+        # in the streamer itself, but accepted here as a parameter).
+        if not format_message_string:
+            string_items = []
+            for column in data_frame.columns:
+                if 'time' in column:
+                    continue
+                elif 'timestamp_desc' in column:
+                    continue
+                string_items.append('{0:s} = {{0!s}}'.format(column))
+            format_message_string = ' '.join(string_items)
+
+        streamer_response = None
+        with importer.ImportStreamer() as streamer:
+            streamer.set_sketch(self)
+            streamer.set_timeline_name(timeline_name)
+            streamer.set_timestamp_description('LOG')
+            streamer.set_message_format_string(format_message_string)
+
+            streamer.add_data_frame(data_frame)
+            streamer_response = streamer.response
+
+        if not streamer_response:
+            return 'No return value.'
+
+        return_lines = []
+        for sketch_object in streamer_response.get('objects', []):
+            return_lines.append('Timeline: {0:s}\nStatus: {1:s}'.format(
+                sketch_object.get('description'),
+                ','.join([x.get(
+                    'status') for x in sketch_object.get('status')])))
+
+        return '\n'.join(return_lines)
+
     def add_timeline(self, searchindex):
         """Add timeline to sketch.
 
@@ -637,6 +702,19 @@ def explore(self,
 
         return response_json
 
+    def list_available_analyzers(self):
+        """Returns a list of available analyzers."""
+        resource_url = '{0:s}/sketches/{1:d}/analyzer/'.format(
+            self.api.api_root, self.id)
+
+        response = self.api.session.get(resource_url)
+
+        if response.status_code == 200:
+            return response.json()
+
+        return '[{0:d}] {1:s} {2:s}'.format(
+            response.status_code, response.reason, response.text)
+
     def run_analyzer(
             self, analyzer_name, analyzer_kwargs=None, timeline_id=None,
             timeline_name=None):
@@ -687,8 +765,8 @@ def run_analyzer(
             timeline_id = timelines[0]
 
         data = {
-            'timeline_id': timeline_id,
-            'analyzer_name': analyzer_name,
+            'index_name': timeline_id,
+            'analyzer_names': [analyzer_name],
             'analyzer_kwargs': analyzer_kwargs,
         }
 
@@ -776,6 +854,9 @@ def store_aggregation(
             aggregator_parameters: parameters of the aggregator.
             chart_type: string representing the chart type.
 
+        Raises:
+            RuntimeError: if the client is unable to store the aggregation.
+
         Returns:
           A stored aggregation object or None if not stored.
         """
@@ -792,6 +873,12 @@ def store_aggregation(
         }
 
         response = self.api.session.post(resource_url, json=form_data)
+        if response.status_code not in HTTP_STATUS_CODE_20X:
+            raise RuntimeError(
+                'Error storing the aggregation, Error message: '
+                '[{0:d}] {1:s} {2:s}'.format(
+                    response.status_code, response.reason, response.text))
+
         response_dict = response.json()
 
         objects = response_dict.get('objects', [])