Merge pull request #8 from onna/workload-ids

.pre-commit-config.yaml

@@ -1,6 +1,5 @@
 repos:
 -   repo: https://github.com/ambv/black
-    rev: stable
     hooks:
     - id: black
       language_version: python3.7

CHANGELOG.rst

@@ -1,3 +1,7 @@
+6.0.6 (unreleased)
+------------------
+- Allow usage of Workload Identities for GCP authentication
+
 6.0.5 (unreleased)
 ------------------
 

Makefile

@@ -6,15 +6,13 @@ pre-checks-deps: lint-deps
 
 pre-checks: pre-checks-deps
 	flake8 guillotina_gcloudstorage --config=setup.cfg
-	isort -c -rc guillotina_gcloudstorage
 	black --check --verbose guillotina_gcloudstorage
 	mypy -p guillotina_gcloudstorage --ignore-missing-imports
 
 lint-deps:
 	pip install "isort>=4,<5" black
 
 lint:
-	isort -rc guillotina_gcloudstorage
 	black guillotina_gcloudstorage
 
 

VERSION

@@ -1 +1 @@
-6.0.5
+6.0.6

guillotina_gcloudstorage/interfaces.py

@@ -1,8 +1,7 @@
 # -*- coding: utf-8 -*-
-from zope.interface import Interface
-
 from guillotina.interfaces import IFile
 from guillotina.interfaces import IFileField
+from zope.interface import Interface
 
 
 class IGCloudFileField(IFileField):

guillotina_gcloudstorage/storage.py

@@ -1,18 +1,5 @@
 # -*- coding: utf-8 -*-
-import asyncio
-import json
-import logging
 from datetime import datetime
-from typing import AsyncIterator
-from urllib.parse import quote_plus
-
-from zope.interface import implementer
-
-import aiohttp
-import backoff
-import google.api_core.exceptions
-import google.cloud.exceptions
-import google.cloud.storage
 from guillotina import configure
 from guillotina import task_vars
 from guillotina.component import get_multi_adapter
@@ -33,12 +20,24 @@
 from guillotina.utils import apply_coroutine
 from guillotina.utils import get_authenticated_user_id
 from guillotina.utils import get_current_request
-from guillotina.utils import run_async
 from guillotina.utils import to_str
 from guillotina_gcloudstorage.interfaces import IGCloudBlobStore
 from guillotina_gcloudstorage.interfaces import IGCloudFile
 from guillotina_gcloudstorage.interfaces import IGCloudFileField
 from oauth2client.service_account import ServiceAccountCredentials
+from typing import AsyncIterator
+from urllib.parse import quote_plus
+from zope.interface import implementer
+
+import aiohttp
+import asyncio
+import backoff
+import google.api_core.exceptions
+import google.cloud.exceptions
+import google.cloud.storage
+import json
+import logging
+import os
 
 
 class IGCloudFileStorageManager(IExternalFileStorageManager):
@@ -49,6 +48,10 @@ class IGCloudFileStorageManager(IExternalFileStorageManager):
 
 MAX_SIZE = 1073741824
 
+METADATA_URL = "http://metadata.google.internal/computeMetadata/v1/"
+METADATA_HEADERS = {"Metadata-Flavor": "Google"}
+SERVICE_ACCOUNT = "default"
+
 SCOPES = ["https://www.googleapis.com/auth/devstorage.read_write"]
 UPLOAD_URL = "https://www.googleapis.com/upload/storage/v1/b/{bucket}/o?uploadType=resumable"  # noqa
 OBJECT_BASE_URL = "https://www.googleapis.com/storage/v1/b"
@@ -97,6 +100,17 @@ def should_clean(self, file):
         cleanup = IFileCleanup(self.context, None)
         return cleanup is None or cleanup.should_clean(file=file, field=self.field)
 
+    async def get_headers(self):
+        util = get_utility(IGCloudBlobStore)
+
+        access_token = await util.get_access_token()
+
+        headers = {}
+        if access_token:
+            headers["AUTHORIZATION"] = f"Bearer {access_token}"
+
+        return headers
+
     async def iter_data(self, uri=None, headers=None):
         if uri is None:
             file = self.field.get(self.field.context or self.context)
@@ -105,16 +119,13 @@ async def iter_data(self, uri=None, headers=None):
             else:
                 uri = file.uri
 
-        if headers is None:
-            headers = {}
-
         util = get_utility(IGCloudBlobStore)
         url = "{}/{}/o/{}".format(
             OBJECT_BASE_URL, await util.get_bucket_name(), quote_plus(uri)
         )
-        headers["AUTHORIZATION"] = "Bearer {}".format(await util.get_access_token())
+
         async with util.session.get(
-            url, headers=headers, params={"alt": "media"}, timeout=-1
+            url, headers=await self.get_headers(), params={"alt": "media"}, timeout=-1
         ) as api_resp:
             if api_resp.status not in (200, 206):
                 text = await api_resp.text()
@@ -180,15 +191,19 @@ async def start(self, dm):
             {"CREATOR": creator, "REQUEST": str(request), "NAME": dm.get("filename")}
         )
         call_size = len(metadata)
-        async with util.session.post(
-            init_url,
-            headers={
-                "AUTHORIZATION": "Bearer {}".format(await util.get_access_token()),
+
+        headers = await self.get_headers()
+        headers.update(
+            {
                 "X-Upload-Content-Type": to_str(dm.content_type),
                 "X-Upload-Content-Length": str(dm.size),
                 "Content-Type": "application/json; charset=UTF-8",
                 "Content-Length": str(call_size),
-            },
+            }
+        )
+        async with util.session.post(
+            init_url,
+            headers=headers,
             data=metadata,
         ) as call:
             if call.status != 200:
@@ -208,10 +223,7 @@ async def delete_upload(self, uri):
                 OBJECT_BASE_URL, await util.get_bucket_name(), quote_plus(uri)
             )
             async with util.session.delete(
-                url,
-                headers={
-                    "AUTHORIZATION": "Bearer {}".format(await util.get_access_token())
-                },
+                url, headers=await self.get_headers()
             ) as resp:
                 try:
                     data = await resp.json()
@@ -317,11 +329,10 @@ async def exists(self):
         url = "{}/{}/o/{}".format(
             OBJECT_BASE_URL, await util.get_bucket_name(), quote_plus(file.uri)
         )
+
         async with util.session.get(
             url,
-            headers={
-                "AUTHORIZATION": "Bearer {}".format(await util.get_access_token())
-            },
+            headers=await self.get_headers(),
         ) as api_resp:
             return api_resp.status == 200
 
@@ -344,12 +355,12 @@ async def copy(self, to_storage_manager, to_dm):
             bucket_name,
             quote_plus(new_uri),
         )
+
+        headers = await self.get_headers()
+        headers.update({"Content-Type": "application/json"})
         async with util.session.post(
             url,
-            headers={
-                "AUTHORIZATION": "Bearer {}".format(await util.get_access_token()),
-                "Content-Type": "application/json",
-            },
+            headers=headers,
         ) as resp:
             if resp.status == 404:
                 text = await resp.text()
@@ -395,9 +406,15 @@ class GCloudBlobStore(object):
     def __init__(self, settings, loop=None):
         self._loop = loop
         self._json_credentials = settings["json_credentials"]
-        self._credentials = ServiceAccountCredentials.from_json_keyfile_name(
-            self._json_credentials, SCOPES
-        )
+
+        if os.path.exists(self._json_credentials):
+            self._credentials = ServiceAccountCredentials.from_json_keyfile_name(
+                self._json_credentials, SCOPES
+            )
+        else:
+            self._credentials = None
+            self._json_credentials = None
+
         self._bucket_name = settings["bucket"]
         self._location = settings.get("location", None)
         self._project = settings.get("project", None)
@@ -420,21 +437,33 @@ def session(self):
             self._session = aiohttp.ClientSession()
         return self._session
 
-    def _get_access_token(self):
-        access_token = self._credentials.get_access_token()
-        self._creation_access_token = datetime.now()
-        return access_token.access_token
-
     async def get_access_token(self):
-        return await run_async(self._get_access_token)
+        # If not using json service credentials, get the access token based on pod rbac
+        if not self._credentials:
+            url = "{}instance/service-accounts/{}/token".format(
+                METADATA_URL, SERVICE_ACCOUNT
+            )
+
+            # Request an access token from the metadata server.
+            async with self.session.get(url, headers=METADATA_HEADERS) as resp:
+                assert resp.status == 200
+                data = await resp.json()
+                access_token = data["access_token"]
+        else:
+            access_token = self._credentials.get_access_token().access_token
+        self._creation_access_token = datetime.now()
+        return access_token
 
     def get_client(self):
         if self._client is None:
-            self._client = (
-                google.cloud.storage.Client.from_service_account_json(  # noqa
-                    self._json_credentials
+            if self._json_credentials:
+                self._client = (
+                    google.cloud.storage.Client.from_service_account_json(  # noqa
+                        self._json_credentials
+                    )
                 )
-            )
+            else:
+                self._client = google.cloud.storage.Client()
         return self._client
 
     def _create_bucket(self, bucket_name, client):
@@ -547,11 +576,14 @@ async def iterate_bucket_page(self, page_token=None, prefix=None):
         if page_token:
             params["pageToken"] = page_token
 
+        headers = {}
+        access_token = await self.get_access_token()
+        if access_token:
+            headers = {"AUTHORIZATION": f"Bearer {access_token}"}
+
         async with self.session.get(
             url,
-            headers={
-                "AUTHORIZATION": "Bearer {}".format(await self.get_access_token())
-            },
+            headers=headers,
             params=params,
         ) as resp:
             assert resp.status == 200

guillotina_gcloudstorage/tests/fixtures.py

@@ -1,7 +1,7 @@
-import os
-
 from guillotina import testing
 
+import os
+
 
 def base_settings_configurator(settings):
     if "applications" in settings:

guillotina_gcloudstorage/tests/test_storage.py

@@ -1,31 +1,30 @@
-import base64
 from functools import partial
-from hashlib import md5
-from urllib.parse import quote_plus
-
-from zope.interface import Interface
-
-import aiohttp
-import google.cloud.storage
-import pytest
 from guillotina import task_vars
 from guillotina.component import get_multi_adapter
 from guillotina.component import get_utility
 from guillotina.content import Container
 from guillotina.exceptions import UnRetryableRequestError
-from guillotina.files import MAX_REQUEST_CACHE_SIZE
 from guillotina.files import FileManager
+from guillotina.files import MAX_REQUEST_CACHE_SIZE
 from guillotina.files.adapter import DBDataManager
 from guillotina.interfaces import IFileNameGenerator
 from guillotina.tests.utils import create_content
 from guillotina.tests.utils import login
 from guillotina.utils import apply_coroutine
 from guillotina_gcloudstorage.interfaces import IGCloudBlobStore
 from guillotina_gcloudstorage.storage import CHUNK_SIZE
-from guillotina_gcloudstorage.storage import OBJECT_BASE_URL
-from guillotina_gcloudstorage.storage import UPLOAD_URL
 from guillotina_gcloudstorage.storage import GCloudFileField
 from guillotina_gcloudstorage.storage import GCloudFileManager
+from guillotina_gcloudstorage.storage import OBJECT_BASE_URL
+from guillotina_gcloudstorage.storage import UPLOAD_URL
+from hashlib import md5
+from urllib.parse import quote_plus
+from zope.interface import Interface
+
+import aiohttp
+import base64
+import google.cloud.storage
+import pytest
 
 
 _test_gif = base64.b64decode(