We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
gcs-importer fails to read username when importing Plaso db from GCS
**Describe the bug**
When the gcs-importer attempts to import a plaso file from GCS, it fails to read the username from the "requester" field in the .plaso.metadata.json file. This means the import fails.

Jul 14 07:51:58 timesketch-server python3[14864]: 2022-07-14 07:51:58,892 [INFO] Downloaded file from GCS: /tmp/381929d791d04ec0a12f85454cf08196.plaso.metadata.json
Jul 14 07:51:58 timesketch-server python3[14864]: 2022-07-14 07:51:58 INFO     Downloaded file from GCS: /tmp/381929d791d04ec0a12f85454cf08196.plaso.metadata.json
Jul 14 07:51:59 timesketch-server python3[14864]: 2022-07-14 07:51:59,923 [INFO] Downloaded file from GCS: /tmp/381929d791d04ec0a12f85454cf08196.plaso
Jul 14 07:51:59 timesketch-server python3[14864]: 2022-07-14 07:51:59 INFO     Downloaded file from GCS: /tmp/381929d791d04ec0a12f85454cf08196.plaso
Jul 14 07:51:59 timesketch-server python3[14864]: 2022-07-14 07:51:59,926 [ERROR] Missing username
Jul 14 07:51:59 timesketch-server python3[14864]: 2022-07-14 07:51:59 ERROR    Missing username

Since [this commit](https://github.com/google/turbinia/commit/125696c8c6bd784af435316a020a38fd608dd5e5) in Turbinia, the "requester" field is under "globals" in the .plaso.metadata.json file, while Timesketch attempts to [read it directly from the root level](https://github.com/google/timesketch/blob/e33056e17a10e961dcd4b897c2cd66a1fc12ebae/contrib/gcs_importer.py#L171)

**To Reproduce**
Steps to reproduce the behavior:
1. Install latest Turbinia
2. Install latest Timesketch
3. Set up GCS importer
4. Analyze something with Turbinia and have it drop results to GCS
5. Observe GCS importer fail.

**Expected behavior**
The plaso database is correctly imported and indexed. 

**Screenshots**
If applicable, add screenshots to help explain your problem.

**Desktop (please complete the following information):**
 - OS: [e.g. iOS]
 - Browser [e.g. chrome, safari]
 - Version [e.g. 22]

**Additional context**
Add any other context about the problem here.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `contrib/gcs_importer.py`
Content:
```
1 # Copyright 2020 Google Inc. All rights reserved.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #     http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 """Google Cloud Storage importer."""
15 
16 import argparse
17 import time
18 import os
19 import sys
20 import uuid
21 import json
22 import logging
23 
24 from werkzeug.exceptions import Forbidden
25 
26 from timesketch.app import create_app
27 from timesketch.lib import tasks
28 from timesketch.models import db_session
29 from timesketch.models.sketch import SearchIndex
30 from timesketch.models.sketch import Sketch
31 from timesketch.models.sketch import Timeline
32 from timesketch.models.user import User
33 
34 try:
35     from google.cloud import pubsub_v1
36     from google.cloud import storage
37 except ImportError:
38     sys.exit("ERROR: You are missing Google Cloud libraries")
39 
40 # Create logger
41 logger = logging.getLogger("gcs_importer")
42 logger.setLevel(logging.DEBUG)
43 handler = logging.StreamHandler()
44 handler.setLevel(logging.DEBUG)
45 formatter = logging.Formatter("%(asctime)s [%(levelname)s] %(message)s")
46 handler.setFormatter(formatter)
47 logger.addHandler(handler)
48 
49 
50 def download_from_gcs(gcs_base_path, filename):
51     """Download file from Google Cloud Storage (GCS).
52 
53     Args:
54         gcs_base_path: (str) GCS bucket path
55         filename: (str) Filename of the file to download
56 
57     Returns:
58         (str) Path to downloaded file
59     """
60     storage_client = storage.Client(args.project)
61     bucket = storage_client.get_bucket(args.bucket)
62     gcs_full_path = os.path.join(gcs_base_path, filename)
63     local_path = os.path.join(args.output, filename)
64     blob = bucket.blob(gcs_full_path)
65     blob.download_to_filename(local_path)
66     logger.info("Downloaded file from GCS: {}".format(local_path))
67     return local_path
68 
69 
70 def setup_sketch(timeline_name, index_name, username, sketch_id=None):
71     """Use existing sketch or create a new sketch.
72 
73     Args:
74         timeline_name: (str) Name of the Timeline
75         index_name: (str) Name of the index
76         username: (str) Who should own the timeline
77         sketch_id: (str) Optional sketch_id to add timeline to
78 
79     Returns:
80         (tuple) sketch ID and timeline ID as integers
81     """
82     with app.app_context():
83         user = User.get_or_create(username=username)
84         sketch = None
85 
86         if sketch_id:
87             try:
88                 sketch = Sketch.query.get_with_acl(sketch_id, user=user)
89                 logger.info(
90                     "Using existing sketch: {} ({})".format(sketch.name, sketch.id)
91                 )
92             except Forbidden:
93                 pass
94 
95         if not (sketch or sketch_id):
96             # Create a new sketch.
97             sketch_name = "Turbinia: {}".format(timeline_name)
98             sketch = Sketch(name=sketch_name, description=sketch_name, user=user)
99             # Need to commit here to be able to set permissions later.
100             db_session.add(sketch)
101             db_session.commit()
102             sketch.grant_permission(permission="read", user=user)
103             sketch.grant_permission(permission="write", user=user)
104             sketch.grant_permission(permission="delete", user=user)
105             sketch.status.append(sketch.Status(user=None, status="new"))
106             db_session.add(sketch)
107             db_session.commit()
108             logger.info("Created new sketch: {} ({})".format(sketch.name, sketch.id))
109 
110         searchindex = SearchIndex.get_or_create(
111             name=timeline_name,
112             description="Created by Turbinia.",
113             user=user,
114             index_name=index_name,
115         )
116         searchindex.grant_permission(permission="read", user=user)
117         searchindex.grant_permission(permission="write", user=user)
118         searchindex.grant_permission(permission="delete", user=user)
119         searchindex.set_status("processing")
120         db_session.add(searchindex)
121         db_session.commit()
122 
123         timeline = Timeline(
124             name=searchindex.name,
125             description=searchindex.description,
126             sketch=sketch,
127             user=user,
128             searchindex=searchindex,
129         )
130 
131         # If the user doesn't have write access to the sketch then create the
132         # timeline but don't attach it to the sketch.
133         if not sketch.has_permission(user, "write"):
134             timeline.sketch = None
135         else:
136             sketch.timelines.append(timeline)
137 
138         db_session.add(timeline)
139         db_session.commit()
140         timeline.set_status("processing")
141 
142         return sketch.id, timeline.id
143 
144 
145 def callback(message):
146     """Google PubSub callback.
147 
148     This function is called on all incoming messages on the configured topic.
149 
150     Args:
151         message: (dict) PubSub message
152     """
153     message.ack()
154     gcs_full_path = message.attributes.get("objectId")
155 
156     # Exit early if the file type is wrong.
157     if not gcs_full_path.endswith(".plaso.metadata.json"):
158         return
159 
160     gcs_base_path = os.path.dirname(gcs_full_path)
161     gcs_metadata_filename = os.path.basename(gcs_full_path)
162     gcs_base_filename = gcs_metadata_filename.replace(".metadata.json", "")
163     gcs_plaso_filename = gcs_base_filename
164 
165     # Download files from GCS
166     local_metadata_file = download_from_gcs(gcs_base_path, gcs_metadata_filename)
167     local_plaso_file = download_from_gcs(gcs_base_path, gcs_plaso_filename)
168 
169     with open(local_metadata_file, "r") as metadata_file:
170         metadata = json.load(metadata_file)
171         username = metadata.get("requester")
172         sketch_id_from_metadata = metadata.get("sketch_id")
173 
174     if not username:
175         logger.error("Missing username")
176         return
177 
178     timeline_name = os.path.splitext(gcs_plaso_filename)[0]
179     index_name = uuid.uuid4().hex
180     sketch_id, timeline_id = setup_sketch(
181         timeline_name, index_name, "admin", sketch_id_from_metadata
182     )
183 
184     # Start indexing
185     with app.app_context():
186         pipeline = tasks.build_index_pipeline(
187             file_path=local_plaso_file,
188             timeline_name=gcs_base_filename,
189             index_name=index_name,
190             file_extension="plaso",
191             sketch_id=sketch_id,
192             timeline_id=timeline_id,
193         )
194         pipeline.apply_async()
195         logger.info("File sent for indexing: {}".format(gcs_base_filename))
196 
197 
198 if __name__ == "__main__":
199     parser = argparse.ArgumentParser(description="GCS importer")
200     parser.add_argument("--project", help="Google Cloud Project ID")
201     parser.add_argument("--bucket", help="Google Cloud Storage bucket to monitor")
202     parser.add_argument("--subscription", help="Google Cloud PubSub subscription")
203     parser.add_argument("--output", default="/tmp", help="Directory for downloads")
204     args = parser.parse_args()
205 
206     # Create flask app
207     app = create_app()
208 
209     # Setup Google Cloud Pub/Sub
210     subscriber = pubsub_v1.SubscriberClient()
211     subscription_path = subscriber.subscription_path(args.project, args.subscription)
212     subscriber.subscribe(subscription_path, callback=callback)
213 
214     logger.info("Listening on PubSub queue: {}".format(args.subscription))
215     while True:
216         time.sleep(10)
217
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/contrib/gcs_importer.py b/contrib/gcs_importer.py
--- a/contrib/gcs_importer.py
+++ b/contrib/gcs_importer.py
@@ -168,7 +168,10 @@
 
     with open(local_metadata_file, "r") as metadata_file:
         metadata = json.load(metadata_file)
-        username = metadata.get("requester")
+        username = metadata.get("globals", {}).get("requester")
+        if not username:
+            # Backwards compatibility for old Turbinia versions.
+            username = metadata.get("requester")
         sketch_id_from_metadata = metadata.get("sketch_id")
 
     if not username:

{"golden_diff": "diff --git a/contrib/gcs_importer.py b/contrib/gcs_importer.py\n--- a/contrib/gcs_importer.py\n+++ b/contrib/gcs_importer.py\n@@ -168,7 +168,10 @@\n \n     with open(local_metadata_file, \"r\") as metadata_file:\n         metadata = json.load(metadata_file)\n-        username = metadata.get(\"requester\")\n+        username = metadata.get(\"globals\", {}).get(\"requester\")\n+        if not username:\n+            # Backwards compatibility for old Turbinia versions.\n+            username = metadata.get(\"requester\")\n         sketch_id_from_metadata = metadata.get(\"sketch_id\")\n \n     if not username:\n", "issue": "gcs-importer fails to read username when importing Plaso db from GCS\n**Describe the bug**\r\nWhen the gcs-importer attempts to import a plaso file from GCS, it fails to read the username from the \"requester\" field in the .plaso.metadata.json file. This means the import fails.\r\n\r\nJul 14 07:51:58 timesketch-server python3[14864]: 2022-07-14 07:51:58,892 [INFO] Downloaded file from GCS: /tmp/381929d791d04ec0a12f85454cf08196.plaso.metadata.json\r\nJul 14 07:51:58 timesketch-server python3[14864]: 2022-07-14 07:51:58 INFO     Downloaded file from GCS: /tmp/381929d791d04ec0a12f85454cf08196.plaso.metadata.json\r\nJul 14 07:51:59 timesketch-server python3[14864]: 2022-07-14 07:51:59,923 [INFO] Downloaded file from GCS: /tmp/381929d791d04ec0a12f85454cf08196.plaso\r\nJul 14 07:51:59 timesketch-server python3[14864]: 2022-07-14 07:51:59 INFO     Downloaded file from GCS: /tmp/381929d791d04ec0a12f85454cf08196.plaso\r\nJul 14 07:51:59 timesketch-server python3[14864]: 2022-07-14 07:51:59,926 [ERROR] Missing username\r\nJul 14 07:51:59 timesketch-server python3[14864]: 2022-07-14 07:51:59 ERROR    Missing username\r\n\r\nSince [this commit](https://github.com/google/turbinia/commit/125696c8c6bd784af435316a020a38fd608dd5e5) in Turbinia, the \"requester\" field is under \"globals\" in the .plaso.metadata.json file, while Timesketch attempts to [read it directly from the root level](https://github.com/google/timesketch/blob/e33056e17a10e961dcd4b897c2cd66a1fc12ebae/contrib/gcs_importer.py#L171)\r\n\r\n**To Reproduce**\r\nSteps to reproduce the behavior:\r\n1. Install latest Turbinia\r\n2. Install latest Timesketch\r\n3. Set up GCS importer\r\n4. Analyze something with Turbinia and have it drop results to GCS\r\n5. Observe GCS importer fail.\r\n\r\n**Expected behavior**\r\nThe plaso database is correctly imported and indexed. \r\n\r\n**Screenshots**\r\nIf applicable, add screenshots to help explain your problem.\r\n\r\n**Desktop (please complete the following information):**\r\n - OS: [e.g. iOS]\r\n - Browser [e.g. chrome, safari]\r\n - Version [e.g. 22]\r\n\r\n**Additional context**\r\nAdd any other context about the problem here.\r\n\n", "before_files": [{"content": "# Copyright 2020 Google Inc. All rights reserved.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\"\"\"Google Cloud Storage importer.\"\"\"\n\nimport argparse\nimport time\nimport os\nimport sys\nimport uuid\nimport json\nimport logging\n\nfrom werkzeug.exceptions import Forbidden\n\nfrom timesketch.app import create_app\nfrom timesketch.lib import tasks\nfrom timesketch.models import db_session\nfrom timesketch.models.sketch import SearchIndex\nfrom timesketch.models.sketch import Sketch\nfrom timesketch.models.sketch import Timeline\nfrom timesketch.models.user import User\n\ntry:\n    from google.cloud import pubsub_v1\n    from google.cloud import storage\nexcept ImportError:\n    sys.exit(\"ERROR: You are missing Google Cloud libraries\")\n\n# Create logger\nlogger = logging.getLogger(\"gcs_importer\")\nlogger.setLevel(logging.DEBUG)\nhandler = logging.StreamHandler()\nhandler.setLevel(logging.DEBUG)\nformatter = logging.Formatter(\"%(asctime)s [%(levelname)s] %(message)s\")\nhandler.setFormatter(formatter)\nlogger.addHandler(handler)\n\n\ndef download_from_gcs(gcs_base_path, filename):\n    \"\"\"Download file from Google Cloud Storage (GCS).\n\n    Args:\n        gcs_base_path: (str) GCS bucket path\n        filename: (str) Filename of the file to download\n\n    Returns:\n        (str) Path to downloaded file\n    \"\"\"\n    storage_client = storage.Client(args.project)\n    bucket = storage_client.get_bucket(args.bucket)\n    gcs_full_path = os.path.join(gcs_base_path, filename)\n    local_path = os.path.join(args.output, filename)\n    blob = bucket.blob(gcs_full_path)\n    blob.download_to_filename(local_path)\n    logger.info(\"Downloaded file from GCS: {}\".format(local_path))\n    return local_path\n\n\ndef setup_sketch(timeline_name, index_name, username, sketch_id=None):\n    \"\"\"Use existing sketch or create a new sketch.\n\n    Args:\n        timeline_name: (str) Name of the Timeline\n        index_name: (str) Name of the index\n        username: (str) Who should own the timeline\n        sketch_id: (str) Optional sketch_id to add timeline to\n\n    Returns:\n        (tuple) sketch ID and timeline ID as integers\n    \"\"\"\n    with app.app_context():\n        user = User.get_or_create(username=username)\n        sketch = None\n\n        if sketch_id:\n            try:\n                sketch = Sketch.query.get_with_acl(sketch_id, user=user)\n                logger.info(\n                    \"Using existing sketch: {} ({})\".format(sketch.name, sketch.id)\n                )\n            except Forbidden:\n                pass\n\n        if not (sketch or sketch_id):\n            # Create a new sketch.\n            sketch_name = \"Turbinia: {}\".format(timeline_name)\n            sketch = Sketch(name=sketch_name, description=sketch_name, user=user)\n            # Need to commit here to be able to set permissions later.\n            db_session.add(sketch)\n            db_session.commit()\n            sketch.grant_permission(permission=\"read\", user=user)\n            sketch.grant_permission(permission=\"write\", user=user)\n            sketch.grant_permission(permission=\"delete\", user=user)\n            sketch.status.append(sketch.Status(user=None, status=\"new\"))\n            db_session.add(sketch)\n            db_session.commit()\n            logger.info(\"Created new sketch: {} ({})\".format(sketch.name, sketch.id))\n\n        searchindex = SearchIndex.get_or_create(\n            name=timeline_name,\n            description=\"Created by Turbinia.\",\n            user=user,\n            index_name=index_name,\n        )\n        searchindex.grant_permission(permission=\"read\", user=user)\n        searchindex.grant_permission(permission=\"write\", user=user)\n        searchindex.grant_permission(permission=\"delete\", user=user)\n        searchindex.set_status(\"processing\")\n        db_session.add(searchindex)\n        db_session.commit()\n\n        timeline = Timeline(\n            name=searchindex.name,\n            description=searchindex.description,\n            sketch=sketch,\n            user=user,\n            searchindex=searchindex,\n        )\n\n        # If the user doesn't have write access to the sketch then create the\n        # timeline but don't attach it to the sketch.\n        if not sketch.has_permission(user, \"write\"):\n            timeline.sketch = None\n        else:\n            sketch.timelines.append(timeline)\n\n        db_session.add(timeline)\n        db_session.commit()\n        timeline.set_status(\"processing\")\n\n        return sketch.id, timeline.id\n\n\ndef callback(message):\n    \"\"\"Google PubSub callback.\n\n    This function is called on all incoming messages on the configured topic.\n\n    Args:\n        message: (dict) PubSub message\n    \"\"\"\n    message.ack()\n    gcs_full_path = message.attributes.get(\"objectId\")\n\n    # Exit early if the file type is wrong.\n    if not gcs_full_path.endswith(\".plaso.metadata.json\"):\n        return\n\n    gcs_base_path = os.path.dirname(gcs_full_path)\n    gcs_metadata_filename = os.path.basename(gcs_full_path)\n    gcs_base_filename = gcs_metadata_filename.replace(\".metadata.json\", \"\")\n    gcs_plaso_filename = gcs_base_filename\n\n    # Download files from GCS\n    local_metadata_file = download_from_gcs(gcs_base_path, gcs_metadata_filename)\n    local_plaso_file = download_from_gcs(gcs_base_path, gcs_plaso_filename)\n\n    with open(local_metadata_file, \"r\") as metadata_file:\n        metadata = json.load(metadata_file)\n        username = metadata.get(\"requester\")\n        sketch_id_from_metadata = metadata.get(\"sketch_id\")\n\n    if not username:\n        logger.error(\"Missing username\")\n        return\n\n    timeline_name = os.path.splitext(gcs_plaso_filename)[0]\n    index_name = uuid.uuid4().hex\n    sketch_id, timeline_id = setup_sketch(\n        timeline_name, index_name, \"admin\", sketch_id_from_metadata\n    )\n\n    # Start indexing\n    with app.app_context():\n        pipeline = tasks.build_index_pipeline(\n            file_path=local_plaso_file,\n            timeline_name=gcs_base_filename,\n            index_name=index_name,\n            file_extension=\"plaso\",\n            sketch_id=sketch_id,\n            timeline_id=timeline_id,\n        )\n        pipeline.apply_async()\n        logger.info(\"File sent for indexing: {}\".format(gcs_base_filename))\n\n\nif __name__ == \"__main__\":\n    parser = argparse.ArgumentParser(description=\"GCS importer\")\n    parser.add_argument(\"--project\", help=\"Google Cloud Project ID\")\n    parser.add_argument(\"--bucket\", help=\"Google Cloud Storage bucket to monitor\")\n    parser.add_argument(\"--subscription\", help=\"Google Cloud PubSub subscription\")\n    parser.add_argument(\"--output\", default=\"/tmp\", help=\"Directory for downloads\")\n    args = parser.parse_args()\n\n    # Create flask app\n    app = create_app()\n\n    # Setup Google Cloud Pub/Sub\n    subscriber = pubsub_v1.SubscriberClient()\n    subscription_path = subscriber.subscription_path(args.project, args.subscription)\n    subscriber.subscribe(subscription_path, callback=callback)\n\n    logger.info(\"Listening on PubSub queue: {}\".format(args.subscription))\n    while True:\n        time.sleep(10)\n", "path": "contrib/gcs_importer.py"}], "after_files": [{"content": "# Copyright 2020 Google Inc. All rights reserved.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\"\"\"Google Cloud Storage importer.\"\"\"\n\nimport argparse\nimport time\nimport os\nimport sys\nimport uuid\nimport json\nimport logging\n\nfrom werkzeug.exceptions import Forbidden\n\nfrom timesketch.app import create_app\nfrom timesketch.lib import tasks\nfrom timesketch.models import db_session\nfrom timesketch.models.sketch import SearchIndex\nfrom timesketch.models.sketch import Sketch\nfrom timesketch.models.sketch import Timeline\nfrom timesketch.models.user import User\n\ntry:\n    from google.cloud import pubsub_v1\n    from google.cloud import storage\nexcept ImportError:\n    sys.exit(\"ERROR: You are missing Google Cloud libraries\")\n\n# Create logger\nlogger = logging.getLogger(\"gcs_importer\")\nlogger.setLevel(logging.DEBUG)\nhandler = logging.StreamHandler()\nhandler.setLevel(logging.DEBUG)\nformatter = logging.Formatter(\"%(asctime)s [%(levelname)s] %(message)s\")\nhandler.setFormatter(formatter)\nlogger.addHandler(handler)\n\n\ndef download_from_gcs(gcs_base_path, filename):\n    \"\"\"Download file from Google Cloud Storage (GCS).\n\n    Args:\n        gcs_base_path: (str) GCS bucket path\n        filename: (str) Filename of the file to download\n\n    Returns:\n        (str) Path to downloaded file\n    \"\"\"\n    storage_client = storage.Client(args.project)\n    bucket = storage_client.get_bucket(args.bucket)\n    gcs_full_path = os.path.join(gcs_base_path, filename)\n    local_path = os.path.join(args.output, filename)\n    blob = bucket.blob(gcs_full_path)\n    blob.download_to_filename(local_path)\n    logger.info(\"Downloaded file from GCS: {}\".format(local_path))\n    return local_path\n\n\ndef setup_sketch(timeline_name, index_name, username, sketch_id=None):\n    \"\"\"Use existing sketch or create a new sketch.\n\n    Args:\n        timeline_name: (str) Name of the Timeline\n        index_name: (str) Name of the index\n        username: (str) Who should own the timeline\n        sketch_id: (str) Optional sketch_id to add timeline to\n\n    Returns:\n        (tuple) sketch ID and timeline ID as integers\n    \"\"\"\n    with app.app_context():\n        user = User.get_or_create(username=username)\n        sketch = None\n\n        if sketch_id:\n            try:\n                sketch = Sketch.query.get_with_acl(sketch_id, user=user)\n                logger.info(\n                    \"Using existing sketch: {} ({})\".format(sketch.name, sketch.id)\n                )\n            except Forbidden:\n                pass\n\n        if not (sketch or sketch_id):\n            # Create a new sketch.\n            sketch_name = \"Turbinia: {}\".format(timeline_name)\n            sketch = Sketch(name=sketch_name, description=sketch_name, user=user)\n            # Need to commit here to be able to set permissions later.\n            db_session.add(sketch)\n            db_session.commit()\n            sketch.grant_permission(permission=\"read\", user=user)\n            sketch.grant_permission(permission=\"write\", user=user)\n            sketch.grant_permission(permission=\"delete\", user=user)\n            sketch.status.append(sketch.Status(user=None, status=\"new\"))\n            db_session.add(sketch)\n            db_session.commit()\n            logger.info(\"Created new sketch: {} ({})\".format(sketch.name, sketch.id))\n\n        searchindex = SearchIndex.get_or_create(\n            name=timeline_name,\n            description=\"Created by Turbinia.\",\n            user=user,\n            index_name=index_name,\n        )\n        searchindex.grant_permission(permission=\"read\", user=user)\n        searchindex.grant_permission(permission=\"write\", user=user)\n        searchindex.grant_permission(permission=\"delete\", user=user)\n        searchindex.set_status(\"processing\")\n        db_session.add(searchindex)\n        db_session.commit()\n\n        timeline = Timeline(\n            name=searchindex.name,\n            description=searchindex.description,\n            sketch=sketch,\n            user=user,\n            searchindex=searchindex,\n        )\n\n        # If the user doesn't have write access to the sketch then create the\n        # timeline but don't attach it to the sketch.\n        if not sketch.has_permission(user, \"write\"):\n            timeline.sketch = None\n        else:\n            sketch.timelines.append(timeline)\n\n        db_session.add(timeline)\n        db_session.commit()\n        timeline.set_status(\"processing\")\n\n        return sketch.id, timeline.id\n\n\ndef callback(message):\n    \"\"\"Google PubSub callback.\n\n    This function is called on all incoming messages on the configured topic.\n\n    Args:\n        message: (dict) PubSub message\n    \"\"\"\n    message.ack()\n    gcs_full_path = message.attributes.get(\"objectId\")\n\n    # Exit early if the file type is wrong.\n    if not gcs_full_path.endswith(\".plaso.metadata.json\"):\n        return\n\n    gcs_base_path = os.path.dirname(gcs_full_path)\n    gcs_metadata_filename = os.path.basename(gcs_full_path)\n    gcs_base_filename = gcs_metadata_filename.replace(\".metadata.json\", \"\")\n    gcs_plaso_filename = gcs_base_filename\n\n    # Download files from GCS\n    local_metadata_file = download_from_gcs(gcs_base_path, gcs_metadata_filename)\n    local_plaso_file = download_from_gcs(gcs_base_path, gcs_plaso_filename)\n\n    with open(local_metadata_file, \"r\") as metadata_file:\n        metadata = json.load(metadata_file)\n        username = metadata.get(\"globals\", {}).get(\"requester\")\n        if not username:\n            # Backwards compatibility for old Turbinia versions.\n            username = metadata.get(\"requester\")\n        sketch_id_from_metadata = metadata.get(\"sketch_id\")\n\n    if not username:\n        logger.error(\"Missing username\")\n        return\n\n    timeline_name = os.path.splitext(gcs_plaso_filename)[0]\n    index_name = uuid.uuid4().hex\n    sketch_id, timeline_id = setup_sketch(\n        timeline_name, index_name, \"admin\", sketch_id_from_metadata\n    )\n\n    # Start indexing\n    with app.app_context():\n        pipeline = tasks.build_index_pipeline(\n            file_path=local_plaso_file,\n            timeline_name=gcs_base_filename,\n            index_name=index_name,\n            file_extension=\"plaso\",\n            sketch_id=sketch_id,\n            timeline_id=timeline_id,\n        )\n        pipeline.apply_async()\n        logger.info(\"File sent for indexing: {}\".format(gcs_base_filename))\n\n\nif __name__ == \"__main__\":\n    parser = argparse.ArgumentParser(description=\"GCS importer\")\n    parser.add_argument(\"--project\", help=\"Google Cloud Project ID\")\n    parser.add_argument(\"--bucket\", help=\"Google Cloud Storage bucket to monitor\")\n    parser.add_argument(\"--subscription\", help=\"Google Cloud PubSub subscription\")\n    parser.add_argument(\"--output\", default=\"/tmp\", help=\"Directory for downloads\")\n    args = parser.parse_args()\n\n    # Create flask app\n    app = create_app()\n\n    # Setup Google Cloud Pub/Sub\n    subscriber = pubsub_v1.SubscriberClient()\n    subscription_path = subscriber.subscription_path(args.project, args.subscription)\n    subscriber.subscribe(subscription_path, callback=callback)\n\n    logger.info(\"Listening on PubSub queue: {}\".format(args.subscription))\n    while True:\n        time.sleep(10)\n", "path": "contrib/gcs_importer.py"}]}