ref(flags): limit scopes for secret updates

src/sentry/flags/endpoints/secrets.py

@@ -83,18 +83,38 @@ def post(self, request: Request, organization: Organization) -> Response:
         if not validator.is_valid():
             return self.respond(validator.errors, status=400)
 
-        FlagWebHookSigningSecretModel.objects.create_or_update(
-            organization=organization,
-            provider=validator.validated_data["provider"],
-            values={
-                "created_by": request.user.id,
-                "date_added": datetime.now(tz=timezone.utc),
-                "provider": validator.validated_data["provider"],
-                "secret": validator.validated_data["secret"],
-            },
+        # these scopes can always update or post secrets
+        has_permission = request.access.has_scope("org:write") or request.access.has_scope(
+            "org:admin"
         )
 
-        return Response(status=201)
+        try:
+            secret = FlagWebHookSigningSecretModel.objects.filter(
+                organization_id=organization.id
+            ).get(provider=validator.validated_data["provider"])
+            # allow update access if the user created the secret
+            is_creator = request.user.id == secret.created_by
+        except FlagWebHookSigningSecretModel.DoesNotExist:
+            # anyone can post a new secret
+            is_creator = True
+
+        if is_creator or has_permission:
+            FlagWebHookSigningSecretModel.objects.create_or_update(
+                organization=organization,
+                provider=validator.validated_data["provider"],
+                values={
+                    "created_by": request.user.id,
+                    "date_added": datetime.now(tz=timezone.utc),
+                    "provider": validator.validated_data["provider"],
+                    "secret": validator.validated_data["secret"],
+                },
+            )
+            return Response(status=201)
+        else:
+            return Response(
+                "You must be an organization owner or manager, or the creator of this secret in order to perform this action.",
+                status=403,
+            )
 
 
 @region_silo_endpoint

tests/sentry/flags/endpoints/test_secrets.py

@@ -116,6 +116,96 @@ def test_post_other_organization(self):
             response = self.client.post(url, data={})
             assert response.status_code == 403, response.content
 
+    def test_update_same_creator(self):
+        new_user = self.create_user("[email protected]")
+        member = self.create_member(organization=self.organization, user=new_user)
+        self.login_as(user=member)
+
+        with self.feature(self.features):
+            response = self.client.post(
+                self.url,
+                data={"secret": "41271af8b9804cd99a4c787a28274991", "provider": "generic"},
+            )
+            assert response.status_code == 201, response.content
+
+        models = FlagWebHookSigningSecretModel.objects.filter(provider="generic").all()
+        assert len(models) == 1
+        assert models[0].secret == "41271af8b9804cd99a4c787a28274991"
+
+        # update secret should be allowed since the creator is the same
+        with self.feature(self.features):
+            response = self.client.post(
+                self.url,
+                data={"secret": "31271af8b9804cd99a4c787a28274993", "provider": "generic"},
+            )
+            assert response.status_code == 201, response.content
+
+        models = FlagWebHookSigningSecretModel.objects.filter(provider="generic").all()
+        assert len(models) == 1
+        assert models[0].secret == "31271af8b9804cd99a4c787a28274993"
+
+    def test_update_no_access(self):
+        FlagWebHookSigningSecretModel.objects.create(
+            created_by="12314124",
+            organization=self.organization,
+            provider="generic",
+            secret="41271af8b9804cd99a4c787a28274991",
+        )
+
+        models = FlagWebHookSigningSecretModel.objects.filter(provider="generic").all()
+        assert len(models) == 1
+        assert models[0].secret == "41271af8b9804cd99a4c787a28274991"
+
+        # update secret should not allowed since the creator is not the same
+        new_user = self.create_user("[email protected]")
+        member = self.create_member(organization=self.organization, user=new_user)
+        self.login_as(user=member)
+        with self.feature(self.features):
+            response = self.client.post(
+                self.url,
+                data={"secret": "31271af8b9804cd99a4c787a28274993", "provider": "generic"},
+            )
+            assert response.status_code == 403, response.content
+            assert (
+                response.data
+                == "You must be an organization owner or manager, or the creator of this secret in order to perform this action."
+            )
+
+        models = FlagWebHookSigningSecretModel.objects.filter(provider="generic").all()
+        assert len(models) == 1
+        assert models[0].secret == "41271af8b9804cd99a4c787a28274991"
+
+    def test_update_has_scope(self):
+        FlagWebHookSigningSecretModel.objects.create(
+            created_by="12314124",
+            organization=self.organization,
+            provider="generic",
+            secret="41271af8b9804cd99a4c787a28274991",
+        )
+
+        models = FlagWebHookSigningSecretModel.objects.filter(provider="generic").all()
+        assert len(models) == 1
+        assert models[0].secret == "41271af8b9804cd99a4c787a28274991"
+
+        # update secret should be allowed due to proper scope
+        new_user = self.create_user("[email protected]")
+        owner = self.create_member(
+            organization=self.organization,
+            user=new_user,
+            role="owner",
+        )
+        self.login_as(user=owner)
+        with self.feature(self.features):
+            response = self.client.post(
+                self.url,
+                data={"secret": "31271af8b9804cd99a4c787a28274993", "provider": "generic"},
+            )
+            assert response.status_code == 201, response.content
+
+        models = FlagWebHookSigningSecretModel.objects.filter(provider="generic").all()
+        assert len(models) == 1
+        assert models[0].secret == "31271af8b9804cd99a4c787a28274993"
+
 
 class OrganizationFlagsWebHookSigningSecretEndpointTestCase(APITestCase):
     endpoint = "sentry-api-0-organization-flag-hooks-signing-secret"