fix(vercel): Adds signature check to vercel delete endpoint

src/sentry/integrations/vercel/webhook.py

@@ -6,11 +6,14 @@
 from collections.abc import Mapping
 from typing import Any, TypedDict
 
+from django.contrib.auth.models import AnonymousUser
 from django.http.request import HttpRequest
 from django.http.response import HttpResponseBase
 from django.utils.crypto import constant_time_compare
 from django.views.decorators.csrf import csrf_exempt
 from requests.exceptions import RequestException
+from rest_framework.authentication import BaseAuthentication
+from rest_framework.exceptions import AuthenticationFailed
 from rest_framework.request import Request
 from rest_framework.response import Response
 
@@ -48,14 +51,12 @@
     refs: list[dict[str, str]]
 
 
-def verify_signature(request):
-    signature = request.META.get("HTTP_X_VERCEL_SIGNATURE")
+def verify_vercel_hmac(body: bytes, signature: str) -> bool:
+    """Verify a Vercel webhook HMAC-SHA1 signature."""
     secret = options.get("vercel.client-secret")
-
-    expected = hmac.new(
-        key=secret.encode("utf-8"), msg=bytes(request.body), digestmod=hashlib.sha1
-    ).hexdigest()
-
+    if not secret:
+        return False
+    expected = hmac.new(key=secret.encode("utf-8"), msg=body, digestmod=hashlib.sha1).hexdigest()
     return constant_time_compare(expected, signature)
 
 
@@ -135,14 +136,29 @@
     return release_payload, sentry_app_installation_token.api_token.token
 
 
+class VercelSignatureAuthentication(BaseAuthentication):
+    def authenticate_header(self, request: Request) -> str:
+        return "X-Vercel-Signature"
+
+    def authenticate(self, request: Request) -> tuple[Any, Any]:
+        signature = request.META.get("HTTP_X_VERCEL_SIGNATURE")
+        if not signature:
+            logger.warning("vercel.webhook.missing-signature")
+            raise AuthenticationFailed("Missing signature")
+        if not verify_vercel_hmac(request.body, signature):
+            logger.warning("vercel.webhook.invalid-signature")
+            raise AuthenticationFailed("Invalid signature")
+        return (AnonymousUser(), None)
+
+
 @control_silo_endpoint
 class VercelWebhookEndpoint(Endpoint):
     owner = ApiOwner.INTEGRATIONS
     publish_status = {
         "DELETE": ApiPublishStatus.PRIVATE,
         "POST": ApiPublishStatus.PRIVATE,
     }
-    authentication_classes = ()
+    authentication_classes = (VercelSignatureAuthentication,)
     permission_classes = ()
     provider = "vercel"
 
@@ -161,14 +177,6 @@
         return external_id
 
     def post(self, request: Request) -> Response | None:
-        if not request.META.get("HTTP_X_VERCEL_SIGNATURE"):
-            logger.warning("vercel.webhook.missing-signature")
-            return self.respond(status=401)
-        is_valid = verify_signature(request)
-        if not is_valid:
-            logger.warning("vercel.webhook.invalid-signature")
-            return self.respond(status=401)
-
         # Vercel's webhook allows you to subscribe to different events,
         # denoted by the `type` attribute. We currently subscribe to:
         #     * integration-configuration.removed (Configuration Removed)
@@ -187,10 +195,9 @@
             return self._deployment_created(external_id, request)
         return None
 
-    def delete(self, request: Request):
+    def delete(self, request: Request) -> Response:
         external_id = self.parse_external_id(request)
         configuration_id = request.data["payload"]["configuration"]["id"]
-
         return self._delete(external_id, configuration_id, request)
 
     def _delete(self, external_id, configuration_id, request):

tests/sentry/integrations/vercel/test_uninstall.py

@@ -1,3 +1,6 @@
+import hashlib
+import hmac
+
 import responses
 
 from fixtures.vercel import SECRET
@@ -102,6 +105,64 @@ def test_uninstall(self) -> None:
             ).exists()
 
 
+@control_silo_test
+class VercelDeleteSignatureTest(APITestCase):
+    def setUp(self) -> None:
+        self.url = "/extensions/vercel/delete/"
+        metadata = {
+            "access_token": "my_access_token",
+            "installation_id": "my_config_id",
+            "installation_type": "team",
+            "webhook_id": "my_webhook_id",
+        }
+        self.integration, _ = self.create_provider_integration_for(
+            self.organization,
+            user=None,
+            provider="vercel",
+            external_id="vercel_team_id",
+            name="My Vercel Team",
+            metadata=metadata,
+        )
+
+    def test_delete_without_signature(self) -> None:
+        with override_options({"vercel.client-secret": SECRET}):
+            response = self.client.delete(
+                path=self.url,
+                data=PRIMARY_UNINSTALL_RESPONSE,
+                content_type="application/json",
+            )
+        assert response.status_code == 401
+
+    def test_delete_with_invalid_signature(self) -> None:
+        with override_options({"vercel.client-secret": SECRET}):
+            response = self.client.delete(
+                path=self.url,
+                data=PRIMARY_UNINSTALL_RESPONSE,
+                content_type="application/json",
+                HTTP_X_VERCEL_SIGNATURE="deadbeefdeadbeefdeadbeefdeadbeefdeadbeef",
+            )
+        assert response.status_code == 401
+
+    def test_delete_with_valid_signature(self) -> None:
+        sig = hmac.new(
+            SECRET.encode("utf-8"),
+            PRIMARY_UNINSTALL_RESPONSE.encode("utf-8"),
+            hashlib.sha1,
+        ).hexdigest()
+        with override_options({"vercel.client-secret": SECRET}):
+            response = self.client.delete(
+                path=self.url,
+                data=PRIMARY_UNINSTALL_RESPONSE,
+                content_type="application/json",
+                HTTP_X_VERCEL_SIGNATURE=sig,
+            )
+        assert response.status_code == 204
+        assert not Integration.objects.filter(id=self.integration.id).exists()
+        assert not OrganizationIntegration.objects.filter(
+            integration_id=self.integration.id, organization_id=self.organization.id
+        ).exists()
+
+
 @control_silo_test
 class VercelUninstallWithConfigurationsTest(APITestCase):
     def setUp(self) -> None:
@@ -144,11 +205,16 @@ def test_uninstall_primary_configuration(self) -> None:
         """
 
         assert len(OrganizationIntegration.objects.all()) == 2
-        response = self.client.delete(
-            path=self.url,
-            data=PRIMARY_UNINSTALL_RESPONSE,
-            content_type="application/json",
-        )
+        sig = hmac.new(
+            SECRET.encode("utf-8"), PRIMARY_UNINSTALL_RESPONSE.encode("utf-8"), hashlib.sha1
+        ).hexdigest()
+        with override_options({"vercel.client-secret": SECRET}):
+            response = self.client.delete(
+                path=self.url,
+                data=PRIMARY_UNINSTALL_RESPONSE,
+                content_type="application/json",
+                HTTP_X_VERCEL_SIGNATURE=sig,
+            )
 
         assert response.status_code == 204
         assert len(OrganizationIntegration.objects.all()) == 1
@@ -175,12 +241,16 @@ def test_uninstall_non_primary_configuration(self) -> None:
         """
 
         assert len(OrganizationIntegration.objects.all()) == 2
-
-        response = self.client.delete(
-            path=self.url,
-            data=NONPRIMARY_UNINSTALL_RESPONSE,
-            content_type="application/json",
-        )
+        sig = hmac.new(
+            SECRET.encode("utf-8"), NONPRIMARY_UNINSTALL_RESPONSE.encode("utf-8"), hashlib.sha1
+        ).hexdigest()
+        with override_options({"vercel.client-secret": SECRET}):
+            response = self.client.delete(
+                path=self.url,
+                data=NONPRIMARY_UNINSTALL_RESPONSE,
+                content_type="application/json",
+                HTTP_X_VERCEL_SIGNATURE=sig,
+            )
 
         assert response.status_code == 204
         assert len(OrganizationIntegration.objects.all()) == 1
@@ -227,11 +297,16 @@ def test_uninstall_single_configuration(self) -> None:
         )
         integration.add_organization(org)
 
-        response = self.client.delete(
-            path=self.url,
-            data=USERID_UNINSTALL_RESPONSE,
-            content_type="application/json",
-        )
+        sig = hmac.new(
+            SECRET.encode("utf-8"), USERID_UNINSTALL_RESPONSE.encode("utf-8"), hashlib.sha1
+        ).hexdigest()
+        with override_options({"vercel.client-secret": SECRET}):
+            response = self.client.delete(
+                path=self.url,
+                data=USERID_UNINSTALL_RESPONSE,
+                content_type="application/json",
+                HTTP_X_VERCEL_SIGNATURE=sig,
+            )
 
         assert response.status_code == 204
         assert not Integration.objects.filter(id=integration.id).exists()
@@ -274,11 +349,16 @@ def test_uninstall_from_sentry(self) -> None:
             == 1
         )
 
-        response = self.client.delete(
-            path=self.url,
-            data=PRIMARY_UNINSTALL_RESPONSE,
-            content_type="application/json",
-        )
+        primary_sig = hmac.new(
+            SECRET.encode("utf-8"), PRIMARY_UNINSTALL_RESPONSE.encode("utf-8"), hashlib.sha1
+        ).hexdigest()
+        with override_options({"vercel.client-secret": SECRET}):
+            response = self.client.delete(
+                path=self.url,
+                data=PRIMARY_UNINSTALL_RESPONSE,
+                content_type="application/json",
+                HTTP_X_VERCEL_SIGNATURE=primary_sig,
+            )
         assert response.status_code == 204
 
         integration = Integration.objects.get(id=self.integration.id)
@@ -319,11 +399,16 @@ def test_uninstall_from_sentry(self) -> None:
             == 0
         )
 
-        response = self.client.delete(
-            path=self.url,
-            data=NONPRIMARY_UNINSTALL_RESPONSE,
-            content_type="application/json",
-        )
+        nonprimary_sig = hmac.new(
+            SECRET.encode("utf-8"), NONPRIMARY_UNINSTALL_RESPONSE.encode("utf-8"), hashlib.sha1
+        ).hexdigest()
+        with override_options({"vercel.client-secret": SECRET}):
+            response = self.client.delete(
+                path=self.url,
+                data=NONPRIMARY_UNINSTALL_RESPONSE,
+                content_type="application/json",
+                HTTP_X_VERCEL_SIGNATURE=nonprimary_sig,
+            )
         assert response.status_code == 204
         assert not Integration.objects.filter(id=self.integration.id).exists()
 

tests/sentry/integrations/vercel/test_webhook.py

@@ -27,7 +27,8 @@ class SignatureVercelTest(APITestCase):
 
     def test_get(self) -> None:
         response = self.client.get(self.webhook_url)
-        assert response.status_code == 405
+        # This will fail the signature verification
+        assert response.status_code == 401
 
     def test_invalid_signature(self) -> None:
         with override_options({"vercel.client-secret": SECRET}):