Add support for background to CloudEvent conversion

.github/workflows/conformance.yml

@@ -24,25 +24,25 @@ jobs:
         go-version: '1.13'
 
     - name: Run HTTP conformance tests
-      uses: GoogleCloudPlatform/functions-framework-conformance/action@v0.3.2
+      uses: GoogleCloudPlatform/functions-framework-conformance/action@v0.3.7
       with:
         functionType: 'http'
         useBuildpacks: false
         validateMapping: false
         cmd: "'functions-framework --source tests/conformance/main.py --target write_http --signature-type http'"
 
     - name: Run event conformance tests
-      uses: GoogleCloudPlatform/functions-framework-conformance/action@v0.3.2
+      uses: GoogleCloudPlatform/functions-framework-conformance/action@v0.3.7
       with:
         functionType: 'legacyevent'
         useBuildpacks: false
         validateMapping: false
         cmd: "'functions-framework --source tests/conformance/main.py --target write_legacy_event --signature-type event'"
 
     - name: Run cloudevent conformance tests
-      uses: GoogleCloudPlatform/functions-framework-conformance/action@v0.3.2
+      uses: GoogleCloudPlatform/functions-framework-conformance/action@v0.3.7
       with:
         functionType: 'cloudevent'
         useBuildpacks: false
-        validateMapping: false
+        validateMapping: true
         cmd: "'functions-framework --source tests/conformance/main.py --target write_cloud_event --signature-type cloudevent'"

src/functions_framework/__init__.py

@@ -27,7 +27,10 @@
 
 from cloudevents.http import from_http, is_binary
 
+from functions_framework import convert
+from functions_framework.background import Event
 from functions_framework.exceptions import (
+    EventConversionException,
     FunctionsFrameworkException,
     InvalidConfigurationException,
     InvalidTargetTypeException,
@@ -43,30 +46,7 @@
 _FUNCTION_STATUS_HEADER_FIELD = "X-Google-Status"
 _CRASH = "crash"
 
-
-class _Event(object):
-    """Event passed to background functions."""
-
-    # Supports both v1beta1 and v1beta2 event formats.
-    def __init__(
-        self,
-        context=None,
-        data="",
-        eventId="",
-        timestamp="",
-        eventType="",
-        resource="",
-        **kwargs,
-    ):
-        self.context = context
-        if not self.context:
-            self.context = {
-                "eventId": eventId,
-                "timestamp": timestamp,
-                "eventType": eventType,
-                "resource": resource,
-            }
-        self.data = data
+_CLOUDEVENT_MIME_TYPE = "application/cloudevents+json"
 
 
 class _LoggingHandler(io.TextIOWrapper):
@@ -97,26 +77,32 @@ def _run_cloudevent(function, request):
 
 def _cloudevent_view_func_wrapper(function, request):
     def view_func(path):
+        ce_exception = None
+        event = None
         try:
-            _run_cloudevent(function, request)
-        except cloud_exceptions.MissingRequiredFields as e:
-            flask.abort(
-                400,
-                description=(
-                    "Function was defined with FUNCTION_SIGNATURE_TYPE=cloudevent but"
-                    " failed to find all required cloudevent fields. Found HTTP"
-                    f" headers: {request.headers} and data: {request.get_data()}. "
-                    f"cloudevents.exceptions.MissingRequiredFields: {e}"
-                ),
-            )
-        except cloud_exceptions.InvalidRequiredFields as e:
+            event = from_http(request.headers, request.get_data())
+        except (
+            cloud_exceptions.MissingRequiredFields,
+            cloud_exceptions.InvalidRequiredFields,
+        ) as e:
+            ce_exception = e
+
+        if not ce_exception:
+            function(event)
+            return "OK"
+
+        # Not a CloudEvent. Try converting to a CloudEvent.
+        try:
+            function(convert.background_event_to_cloudevent(request))
+        except EventConversionException as e:
             flask.abort(
                 400,
                 description=(
                     "Function was defined with FUNCTION_SIGNATURE_TYPE=cloudevent but"
-                    " found one or more invalid required cloudevent fields. Found HTTP"
-                    f" headers: {request.headers} and data: {request.get_data()}. "
-                    f"cloudevents.exceptions.InvalidRequiredFields: {e}"
+                    " parsing CloudEvent failed and converting from background event to"
+                    f" CloudEvent also failed.\nGot HTTP headers: {request.headers}\nGot"
+                    f" data: {request.get_data()}\nGot CloudEvent exception: {repr(ce_exception)}"
+                    f"\nGot background event conversion exception: {repr(e)}"
                 ),
             )
         return "OK"
@@ -143,7 +129,7 @@ def view_func(path):
             event_data = request.get_json()
             if not event_data:
                 flask.abort(400)
-            event_object = _Event(**event_data)
+            event_object = Event(**event_data)
             data = event_object.data
             context = Context(**event_object.context)
             function(data, context)

src/functions_framework/background.py

@@ -0,0 +1,38 @@
+# Copyright 2021 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+class Event(object):
+    """Event passed to background functions."""
+
+    # Supports both v1beta1 and v1beta2 event formats.
+    def __init__(
+        self,
+        context=None,
+        data="",
+        eventId="",
+        timestamp="",
+        eventType="",
+        resource="",
+        **kwargs,
+    ):
+        self.context = context
+        if not self.context:
+            self.context = {
+                "eventId": eventId,
+                "timestamp": timestamp,
+                "eventType": eventType,
+                "resource": resource,
+            }
+        self.data = data

src/functions_framework/convert.py

@@ -0,0 +1,160 @@
+# Copyright 2021 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import re
+
+from typing import Tuple
+
+from cloudevents.http import CloudEvent
+
+from functions_framework import background
+from functions_framework.exceptions import EventConversionException
+from google.cloud.functions.context import Context
+
+_CLOUDEVENT_SPEC_VERSION = "1.0"
+
+_BACKGROUND_TO_CE_TYPE = {
+    "google.pubsub.topic.publish": "google.cloud.pubsub.topic.v1.messagePublished",
+    "providers/cloud.pubsub/eventTypes/topic.publish": "google.cloud.pubsub.topic.v1.messagePublished",
+    "google.storage.object.finalize": "google.cloud.storage.object.v1.finalized",
+    "google.storage.object.delete": "google.cloud.storage.object.v1.deleted",
+    "google.storage.object.archive": "google.cloud.storage.object.v1.archived",
+    "google.storage.object.metadataUpdate": "google.cloud.storage.object.v1.metadataUpdated",
+    "providers/cloud.firestore/eventTypes/document.write": "google.cloud.firestore.document.v1.written",
+    "providers/cloud.firestore/eventTypes/document.create": "google.cloud.firestore.document.v1.created",
+    "providers/cloud.firestore/eventTypes/document.update": "google.cloud.firestore.document.v1.updated",
+    "providers/cloud.firestore/eventTypes/document.delete": "google.cloud.firestore.document.v1.deleted",
+    "providers/firebase.auth/eventTypes/user.create": "google.firebase.auth.user.v1.created",
+    "providers/firebase.auth/eventTypes/user.delete": "google.firebase.auth.user.v1.deleted",
+    "providers/google.firebase.analytics/eventTypes/event.log": "google.firebase.analytics.log.v1.written",
+    "providers/google.firebase.database/eventTypes/ref.create": "google.firebase.database.document.v1.created",
+    "providers/google.firebase.database/eventTypes/ref.write": "google.firebase.database.document.v1.written",
+    "providers/google.firebase.database/eventTypes/ref.update": "google.firebase.database.document.v1.updated",
+    "providers/google.firebase.database/eventTypes/ref.delete": "google.firebase.database.document.v1.deleted",
+    "providers/cloud.storage/eventTypes/object.change": "google.cloud.storage.object.v1.finalized",
+}
+
+_FIREBASE_AUTH_CE_SERVICE = "firebaseauth.googleapis.com"
+_FIREBASE_CE_SERVICE = "firebase.googleapis.com"
+_FIREBASE_DB_CE_SERVICE = "firebasedatabase.googleapis.com"
+_FIRESTORE_CE_SERVICE = "firestore.googleapis.com"
+_PUBSUB_CE_SERVICE = "pubsub.googleapis.com"
+_STORAGE_CE_SERVICE = "storage.googleapis.com"
+
+_SERVICE_BACKGROUND_TO_CE = {
+    "providers/cloud.firestore/": _FIRESTORE_CE_SERVICE,
+    "providers/google.firebase.analytics/": _FIREBASE_CE_SERVICE,
+    "providers/firebase.auth/": _FIREBASE_AUTH_CE_SERVICE,
+    "providers/google.firebase.database/": _FIREBASE_DB_CE_SERVICE,
+    "providers/cloud.pubsub/": _PUBSUB_CE_SERVICE,
+    "providers/cloud.storage/": _STORAGE_CE_SERVICE,
+    "google.pubsub": _PUBSUB_CE_SERVICE,
+    "google.storage": _STORAGE_CE_SERVICE,
+}
+
+_CE_SERVICE_TO_RESOURCE_RE = {
+    _FIREBASE_CE_SERVICE: re.compile(r"^(projects/[^/]+)/(events/[^/]+)$"),
+    _FIREBASE_DB_CE_SERVICE: re.compile(r"^(projects/[^/]/instances/[^/]+)/(refs/.+)$"),
+    _FIRESTORE_CE_SERVICE: re.compile(
+        r"^(projects/[^/]+/databases/\(default\))/(documents/.+)$"
+    ),
+    _STORAGE_CE_SERVICE: re.compile(r"^(projects/[^/]/buckets/[^/]+)/(objects/.+)$"),
+}
+
+# Maps Firebase Auth background event metadata field names to their equivalent
+# CloudEvent field names.
+_FIREBASE_AUTH_METADATA_FIELDS_BACKGROUND_TO_CE = {
+    "createdAt": "createTime",
+    "lastSignedInAt": "lastSignInTime",
+}
+
+
+def background_event_to_cloudevent(request) -> CloudEvent:
+    event_data = request.get_json()
+    if not event_data:
+        raise EventConversionException("Failed to parse JSON")
+
+    event_object = background.Event(**event_data)
+    data = event_object.data
+    context = Context(**event_object.context)
+
+    if context.event_type not in _BACKGROUND_TO_CE_TYPE:
+        raise EventConversionException(
+            f'Unable to find CloudEvent equivalent type for "{context.event_type}"'
+        )
+    new_type = _BACKGROUND_TO_CE_TYPE[context.event_type]
+
+    service, resource, subject = _split_resource(context)
+
+    # Handle Pub/Sub events.
+    if service == _PUBSUB_CE_SERVICE:
+        data = {"message": data}
+
+    # Handle Firebase Auth events.
+    if service == _FIREBASE_AUTH_CE_SERVICE:
+        if "metadata" in data:
+            for old, new in _FIREBASE_AUTH_METADATA_FIELDS_BACKGROUND_TO_CE.items():
+                if old in data["metadata"]:
+                    data["metadata"][new] = data["metadata"][old]
+                    del data["metadata"][old]
+        if "uid" in data:
+            uid = data["uid"]
+            subject = f"users/{uid}"
+
+    metadata = {
+        "id": context.event_id,
+        "time": context.timestamp,
+        "specversion": _CLOUDEVENT_SPEC_VERSION,
+        "datacontenttype": "application/json",
+        "type": new_type,
+        "source": f"//{service}/{resource}",
+    }
+
+    if subject:
+        metadata["subject"] = subject
+
+    return CloudEvent(metadata, data)
+
+
+# Splits a background event's resource into a CloudEvent service, resource, and subject.
+def _split_resource(context: Context) -> Tuple[str, str, str]:
+    service = ""
+    resource = ""
+    if isinstance(context.resource, dict):
+        service = context.resource.get("service", "")
+        resource = context.resource["name"]
+    else:
+        resource = context.resource
+
+    # If there's no service we'll choose an appropriate one based on the event type.
+    if not service:
+        for b_service, ce_service in _SERVICE_BACKGROUND_TO_CE.items():
+            if context.event_type.startswith(b_service):
+                service = ce_service
+                break
+        if not service:
+            raise EventConversionException(
+                "Unable to find CloudEvent equivalent service "
+                f"for {context.event_type}"
+            )
+
+    # If we don't need to split the resource string then we're done.
+    if service not in _CE_SERVICE_TO_RESOURCE_RE:
+        return service, resource, ""
+
+    # Split resource into resource and subject.
+    match = _CE_SERVICE_TO_RESOURCE_RE[service].fullmatch(resource)
+    if not match:
+        raise EventConversionException("Resource regex did not match")
+
+    return service, match.group(1), match.group(2)

src/functions_framework/exceptions.py

@@ -31,3 +31,7 @@ class MissingSourceException(FunctionsFrameworkException):
 
 class MissingTargetException(FunctionsFrameworkException):
     pass
+
+
+class EventConversionException(FunctionsFrameworkException):
+    pass