Initial integration with test workflow in Cinder staging environment

.env.example

@@ -40,3 +40,7 @@ GCS_IMAGE_BUCKET=favicon-bucket-2
 GOOGLE_CLOUD_PROJECT=niklas-test-fx-sharing
 # Path to a service-account key file (mounted at /app inside the container).
 GOOGLE_APPLICATION_CREDENTIALS=/app/.gcloud_credentials
+
+CINDER_URL=''
+CINDER_API_TOKEN=''
+CINDER_WEBHOOK_TOKEN=''

fxsharing/settings.py

@@ -233,3 +233,8 @@
         send_default_pii=False,
         traces_sample_rate=0,
     )
+
+CINDER_URL = env("CINDER_URL", default="")
+CINDER_API_ENDPOINT = CINDER_URL.rstrip("/") + "/api/v2/workflows/event/"
+CINDER_API_TOKEN = env("CINDER_API_TOKEN", default="")
+CINDER_WEBHOOK_TOKEN = env("CINDER_WEBHOOK_TOKEN", default="")

fxsharing/shares/tests.py

@@ -1,9 +1,12 @@
+import hashlib
+import hmac
 import importlib
 import json
 from datetime import timedelta
 from io import StringIO
 from unittest.mock import MagicMock
 
+from django.conf import settings
 from django.contrib.auth import get_user_model
 from django.contrib.auth.models import AnonymousUser
 from django.contrib.messages import get_messages
@@ -806,3 +809,81 @@ def test_500_returns_500_status(self):
     def test_500_contains_expected_copy(self):
         response = server_error(self._request())
         assert b"problem with this page" in response.content
+
+
+@override_settings(CINDER_WEBHOOK_TOKEN="test-webhook-token")  # noqa: S106
+class TestTsWebhook(TestCase):
+    @classmethod
+    def setUpTestData(cls):
+        cls.user = User.objects.create_user(fxa_id="a1b2c3d4e5f6webhook")
+        cls.share = Share.objects.create(title="Sample", user=cls.user)
+
+    def _signed_post(self, payload):
+        body = json.dumps(payload).encode("utf-8")
+        sig = hmac.new(
+            settings.CINDER_WEBHOOK_TOKEN.encode("utf-8"),
+            msg=body,
+            digestmod=hashlib.sha256,
+        ).hexdigest()
+        return self.client.post(
+            reverse("ts_webhook"),
+            data=body,
+            content_type="application/json",
+            HTTP_X_CINDER_SIGNATURE=sig,
+        )
+
+    def _decision_payload(self, share, enforcement_actions):
+        return {
+            "event": "decision.created",
+            "payload": {
+                "enforcement_actions": enforcement_actions,
+                "entity": {
+                    "entity_schema": "fxsharing",
+                    "attributes": {
+                        "id": str(share.id),
+                        "shortcode": share.shortcode,
+                        "title": share.title,
+                        "reason": "test",
+                    },
+                },
+            },
+        }
+
+    def test_rejects_invalid_signature(self):
+        payload = self._decision_payload(
+            self.share, ["link-collections-dont-publish-collection"]
+        )
+        response = self.client.post(
+            reverse("ts_webhook"),
+            data=json.dumps(payload),
+            content_type="application/json",
+            HTTP_X_CINDER_SIGNATURE="not-a-real-signature",
+        )
+        assert response.status_code == 400
+        self.share.refresh_from_db()
+        assert self.share.status == ShareStatus.ACTIVE
+
+    def test_dont_publish_blocks_share(self):
+        payload = self._decision_payload(
+            self.share, ["link-collections-dont-publish-collection"]
+        )
+        response = self._signed_post(payload)
+        assert response.status_code == 201
+        self.share.refresh_from_db()
+        assert self.share.status == ShareStatus.BLOCKED
+
+    def test_ban_user_blocks_all_shares_for_that_user(self):
+        other_share = Share.objects.create(title="Other", user=self.user)
+        bystander_user = User.objects.create_user(fxa_id="a1b2c3d4e5f6bystand")
+        bystander_share = Share.objects.create(title="Bystander", user=bystander_user)
+
+        payload = self._decision_payload(self.share, ["link-collections-ban-user"])
+        response = self._signed_post(payload)
+        assert response.status_code == 201
+
+        self.share.refresh_from_db()
+        other_share.refresh_from_db()
+        bystander_share.refresh_from_db()
+        assert self.share.status == ShareStatus.BLOCKED
+        assert other_share.status == ShareStatus.BLOCKED
+        assert bystander_share.status == ShareStatus.ACTIVE

fxsharing/shares/urls.py

@@ -6,6 +6,7 @@
 urlpatterns = [
     path("", views.landing, name="landing"),
     path("api/v1/create", views.create_share, name="create_share"),
+    path("api/v1/ts_response", views.ts_webhook, name="ts_webhook"),
     path("event", views.record_client_event, name="record_client_event"),
     path("report/<str:shortcode>", views.report_share, name="report_share"),
     path("auth-complete", views.auth_complete, name="auth_complete"),

fxsharing/shares/views.py

@@ -1,4 +1,7 @@
+import hashlib
+import hmac
 import json
+import logging
 from datetime import timedelta
 
 from django.conf import settings
@@ -9,10 +12,12 @@
 from django.shortcuts import get_object_or_404, redirect, render
 from django.urls import reverse
 from django.utils import timezone
+from django.utils.encoding import force_bytes
 from django.utils.html import escape
 from django.views.decorators.csrf import csrf_exempt
 from django.views.decorators.http import require_POST
 
+import requests
 from jsonschema import ValidationError, validate
 from modern_csrf.decorators import csrf_protect
 from opentelemetry import trace
@@ -23,6 +28,20 @@
 
 tracer = trace.get_tracer(__name__)
 
+log = logging.getLogger(__name__)
+
+
+class CinderWebhookError(ValidationError):
+    """Validation error from Cinder webhook payload. Returned to Cinder as 400 error."""
+
+    reportable = True
+
+
+class CinderWebhookIgnoredError(CinderWebhookError):
+    """Not an error, a decision we ignore because we already took action, or it's for an entity we don't need to track."""
+
+    reportable = False
+
 
 def shares(request):
     shares = Share.objects.filter(parent_share__isnull=True)
@@ -104,6 +123,48 @@ def create_share_from_data(data, user, parent_share=None):
     return share
 
 
+def report_link_sharing_quality(share):
+    token = settings.CINDER_API_TOKEN
+    endpoint = settings.CINDER_API_ENDPOINT
+    reason = "link_sharing"  # TODO figure out what reason is for
+
+    payload = {
+        "event_name": "link_sharing_quality",
+        "entity": {
+            "entity_schema": "fxsharing",
+            "attributes": {
+                "id": str(share.id),
+                "shortcode": share.shortcode,
+                "title": share.title,
+                "reason": reason,
+            },
+        },
+        "subgraph": {
+            "entities": [
+                {
+                    "entity_schema": "fxsharing_url",
+                    "attributes": {
+                        "id": str(link.id),
+                        "url": link.url,
+                        "title": link.title,
+                    },
+                }
+                for link in share.links.all()[:30]
+            ],
+            "relationships": [],
+        },
+    }
+
+    response = requests.post(
+        endpoint,
+        json=payload,
+        headers={"Authorization": f"Bearer {token}"},
+        timeout=10,
+    )
+    response.raise_for_status()
+    return response
+
+
 @require_POST
 @csrf_protect
 def create_share(request):
@@ -134,6 +195,13 @@ def create_share(request):
         span.set_attribute("share.link_count", len(data.get("links", [])))
         share = create_share_from_data(data=data, user=request.user)
         span.set_attribute("share.shortcode", share.shortcode)
+        # TODO - enqueue a job instead, so we don't block the response on Cinder responding
+        try:
+            report_link_sharing_quality(share)
+        except requests.RequestException:
+            log.exception("Cinder quality report failed for share %s", share.id)
+
+        url = request.build_absolute_uri(f"/{share.shortcode}")
         url = request.build_absolute_uri(f"/{share.shortcode}")
         return JsonResponse({"url": url}, status=201)
 
@@ -248,3 +316,92 @@ def page_not_found(request, exception):
 
 def server_error(request):
     return render(request, "shares/500.html", status=500)
+
+
+# First-iteration pass at webhook listener.
+@require_POST
+@csrf_exempt
+def ts_webhook(request):
+    # Loosely based on the AMO webhook handler at:
+    # https://github.com/mozilla/addons-server/blob/165b73f1/src/olympia/abuse/views.py#L355
+
+    # Verify the webhook signature matches the token.
+    header = request.headers.get("X-Cinder-Signature", "")
+    key = force_bytes(settings.CINDER_WEBHOOK_TOKEN)
+    digest = hmac.new(key, msg=request.body, digestmod=hashlib.sha256).hexdigest()
+    if not hmac.compare_digest(header, digest):
+        log.error("Invalid webhook signature")
+        return HttpResponseBadRequest("Invalid webhook token")
+
+    try:
+        data = json.loads(request.body)
+    except json.JSONDecodeError:
+        return HttpResponseBadRequest("Invalid JSON in request body")
+
+    event = data.get("event")
+    payload = data.get("payload") or {}
+
+    try:
+        match event:
+            case "decision.created":
+                log.info("Valid payload from fxsharing queue: %s", payload)
+
+                share_id = payload.get("entity", {}).get("attributes", {}).get("id")
+                try:
+                    share = Share.objects.get(id=share_id)
+                except Share.DoesNotExist:
+                    log.warning("Webhook for unknown share id %s; ignoring", share_id)
+                    return JsonResponse(
+                        {
+                            "fxsharing": {
+                                "received": True,
+                                "handled": False,
+                                "not_handled_reason": "unknown share id",
+                            }
+                        },
+                        status=200,
+                    )
+
+                for action in payload.get("enforcement_actions") or []:
+                    log.info("Received enforcement action: %s", action)
+                    if action == "link-collections-ban-user":
+                        # TODO actually implement user banning.
+                        Share.objects.filter(user=share.user).update(
+                            status=ShareStatus.BLOCKED
+                        )
+                        break
+                    elif action == "link-collections-dont-publish-collection":
+                        Share.objects.filter(pk=share.pk).update(
+                            status=ShareStatus.BLOCKED
+                        )
+                    elif action == "link-collections-publish-collection":
+                        # TODO add a "passed review" ShareStatus so we can
+                        # audit published shares for any that never got
+                        # approved.
+                        pass
+
+            case "job.actioned":
+                # For now, we just ignore these. Even in cases where a share
+                # was enqueued for processing, we should also get a decision
+                # created event.
+                pass
+            case _:
+                log.info("Unsupported payload received: %s", str(data)[:255])
+                raise CinderWebhookError(f"{event} is not supported")
+    except CinderWebhookError as exc:
+        return JsonResponse(
+            data={
+                "fxsharing": {
+                    "received": True,
+                    "handled": False,
+                    "not_handled_reason": exc.message,
+                }
+            },
+            # Differentiate errors we want exposed in Cinder's logs, and
+            # known cases where we can safely ignore the error.
+            status=(400 if exc.reportable else 200),
+        )
+    return JsonResponse(
+        data={"fxsharing": {"received": True, "handled": True}},
+        status=201,
+    )