✨ feat: create initial DataAccessGrant Service (#95095)

Author: iamrajjoshi

src/sentry/data_secrecy/data_access_grant_service/__init__.py

@@ -0,0 +1,35 @@
+from django.db import models
+from django.utils import timezone
+
+from sentry.data_secrecy.data_access_grant_service.model import RpcEffectiveGrantStatus
+from sentry.data_secrecy.data_access_grant_service.serial import serialize_effective_grant_status
+from sentry.data_secrecy.data_access_grant_service.service import DataAccessGrantService
+from sentry.data_secrecy.models.data_access_grant import DataAccessGrant
+
+
+class DatabaseBackedDataAccessGrantService(DataAccessGrantService):
+    def get_effective_grant_status(self, *, organization_id: int) -> RpcEffectiveGrantStatus | None:
+        """
+        Get the effective grant status for an organization.
+        """
+        now = timezone.now()
+        active_grants = DataAccessGrant.objects.filter(
+            organization_id=organization_id,
+            grant_start__lte=now,
+            grant_end__gt=now,
+            revocation_date__isnull=True,  # Not revoked
+        )
+
+        if not active_grants.exists():
+            return None
+
+        # Calculate aggregate grant status - only need the time window for access control
+        min_start = active_grants.aggregate(min_start=models.Min("grant_start"))["min_start"]
+        max_end = active_grants.aggregate(max_end=models.Max("grant_end"))["max_end"]
+
+        grant_status = {
+            "access_start": min_start.isoformat(),
+            "access_end": max_end.isoformat(),
+        }
+
+        return serialize_effective_grant_status(grant_status, organization_id)

src/sentry/data_secrecy/data_access_grant_service/impl.py

@@ -0,0 +1,13 @@
+from datetime import datetime
+
+from sentry.hybridcloud.rpc import RpcModel
+
+
+class RpcEffectiveGrantStatus(RpcModel):
+    """
+    Simplified model for access control - only contains essential, aggregated grant information.
+    """
+
+    organization_id: int
+    access_start: datetime
+    access_end: datetime

src/sentry/data_secrecy/data_access_grant_service/model.py

@@ -0,0 +1,17 @@
+from datetime import datetime
+
+from sentry.data_secrecy.data_access_grant_service.model import RpcEffectiveGrantStatus
+
+
+def serialize_effective_grant_status(
+    grant_status: dict, organization_id: int
+) -> RpcEffectiveGrantStatus:
+    """
+    Convert cached grant status to simplified RpcGrantStatus model for access control.
+    """
+
+    return RpcEffectiveGrantStatus(
+        organization_id=organization_id,
+        access_start=datetime.fromisoformat(grant_status["access_start"]),
+        access_end=datetime.fromisoformat(grant_status["access_end"]),
+    )

src/sentry/data_secrecy/data_access_grant_service/serial.py

@@ -0,0 +1,31 @@
+# Please do not use
+#     from __future__ import annotations
+# in modules such as this one where hybrid cloud data models or service classes are
+# defined, because we want to reflect on type annotations and avoid forward references.
+
+import abc
+
+from sentry.data_secrecy.data_access_grant_service.model import RpcEffectiveGrantStatus
+from sentry.hybridcloud.rpc.service import RpcService, rpc_method
+from sentry.silo.base import SiloMode
+
+
+class DataAccessGrantService(RpcService):
+    key = "data_access_grant"
+    local_mode = SiloMode.CONTROL
+
+    @classmethod
+    def get_local_implementation(cls) -> RpcService:
+        from sentry.data_secrecy.data_access_grant_service.impl import (
+            DatabaseBackedDataAccessGrantService,
+        )
+
+        return DatabaseBackedDataAccessGrantService()
+
+    @rpc_method
+    @abc.abstractmethod
+    def get_effective_grant_status(self, *, organization_id: int) -> RpcEffectiveGrantStatus | None:
+        pass
+
+
+data_access_grant_service = DataAccessGrantService.create_delegation()

src/sentry/data_secrecy/data_access_grant_service/service.py

@@ -30,6 +30,7 @@
 from sentry.auth.access import RpcBackedAccess
 from sentry.auth.services.auth.model import RpcAuthState, RpcMemberSsoState
 from sentry.constants import SentryAppInstallationStatus, SentryAppStatus
+from sentry.data_secrecy.models.data_access_grant import DataAccessGrant
 from sentry.event_manager import EventManager
 from sentry.eventstore.models import Event
 from sentry.hybridcloud.models.outbox import RegionOutbox, outbox_context
@@ -569,6 +570,11 @@ def create_project_template(project=None, organization=None, **kwargs) -> Projec
 
         return project_template
 
+    @staticmethod
+    @assume_test_silo_mode(SiloMode.CONTROL)
+    def create_data_access_grant(**kwargs):
+        return DataAccessGrant.objects.create(**kwargs)
+
     @staticmethod
     @assume_test_silo_mode(SiloMode.REGION)
     def create_project_bookmark(project, user):

src/sentry/testutils/factories.py

@@ -495,6 +495,9 @@ def create_external_team(self, team=None, integration=None, **kwargs):
             team=team, organization=team.organization, integration_id=integration.id, **kwargs
         )
 
+    def create_data_access_grant(self, **kwargs):
+        return Factories.create_data_access_grant(**kwargs)
+
     def create_codeowners(self, project=None, code_mapping=None, **kwargs):
         if not project:
             project = self.project

src/sentry/testutils/fixtures.py

@@ -0,0 +1,135 @@
+from datetime import datetime, timedelta, timezone
+
+from sentry.data_secrecy.data_access_grant_service.service import data_access_grant_service
+from sentry.data_secrecy.models.data_access_grant import DataAccessGrant
+from sentry.testutils.cases import TestCase
+from sentry.testutils.helpers.datetime import freeze_time
+from sentry.testutils.silo import all_silo_test, create_test_regions
+
+
+@all_silo_test(regions=create_test_regions("us"))
+@freeze_time("2025-07-08 00:00:00")
+class TestDataAccessGrantService(TestCase):
+    def setUp(self):
+        self.organization = self.create_organization()
+        self.organization_2 = self.create_organization()
+
+    def test_get_effective_waiver_status_with_active_grant(self):
+        now = datetime.now(tz=timezone.utc)
+        grant_start = now - timedelta(hours=1)
+        grant_end = now + timedelta(hours=1)
+
+        self.create_data_access_grant(
+            organization_id=self.organization.id,
+            grant_type=DataAccessGrant.GrantType.ZENDESK,
+            ticket_id="TICKET-123",
+            grant_start=grant_start,
+            grant_end=grant_end,
+        )
+
+        result = data_access_grant_service.get_effective_grant_status(
+            organization_id=self.organization.id
+        )
+
+        assert result is not None
+        assert result.organization_id == self.organization.id
+        assert result.access_start == grant_start
+        assert result.access_end == grant_end
+
+    def test_get_effective_waiver_status_with_no_grants(self):
+        result = data_access_grant_service.get_effective_grant_status(
+            organization_id=self.organization.id
+        )
+        assert result is None
+
+    def test_get_effective_waiver_status_with_expired_grant(self):
+        now = datetime.now(tz=timezone.utc)
+        grant_start = now - timedelta(hours=2)
+        grant_end = now - timedelta(hours=1)  # Expired
+
+        self.create_data_access_grant(
+            organization_id=self.organization.id,
+            grant_type=DataAccessGrant.GrantType.ZENDESK,
+            ticket_id="TICKET-123",
+            grant_start=grant_start,
+            grant_end=grant_end,
+        )
+
+        result = data_access_grant_service.get_effective_grant_status(
+            organization_id=self.organization.id
+        )
+        assert result is None
+
+    def test_get_effective_waiver_status_with_future_grant(self):
+        now = datetime.now(tz=timezone.utc)
+        grant_start = now + timedelta(hours=1)  # Future
+        grant_end = now + timedelta(hours=2)
+
+        self.create_data_access_grant(
+            organization_id=self.organization.id,
+            grant_type=DataAccessGrant.GrantType.ZENDESK,
+            ticket_id="TICKET-123",
+            grant_start=grant_start,
+            grant_end=grant_end,
+        )
+
+        result = data_access_grant_service.get_effective_grant_status(
+            organization_id=self.organization.id
+        )
+        assert result is None
+
+    def test_get_effective_waiver_status_with_revoked_grant(self):
+        now = datetime.now(tz=timezone.utc)
+        grant_start = now - timedelta(hours=1)
+        grant_end = now + timedelta(hours=1)
+
+        self.create_data_access_grant(
+            organization_id=self.organization.id,
+            grant_type=DataAccessGrant.GrantType.ZENDESK,
+            ticket_id="TICKET-123",
+            grant_start=grant_start,
+            grant_end=grant_end,
+            revocation_date=now,
+            revocation_reason=DataAccessGrant.RevocationReason.MANUAL_REVOCATION,
+        )
+
+        result = data_access_grant_service.get_effective_grant_status(
+            organization_id=self.organization.id
+        )
+        assert result is None
+
+    def test_get_effective_waiver_status_with_multiple_grants(self):
+        now = datetime.now(tz=timezone.utc)
+
+        # Grant 1: Earlier start, earlier end
+        grant1_start = now - timedelta(hours=2)
+        grant1_end = now + timedelta(hours=1)
+
+        # Grant 2: Later start, later end
+        grant2_start = now - timedelta(hours=1)
+        grant2_end = now + timedelta(hours=2)
+
+        self.create_data_access_grant(
+            organization_id=self.organization.id,
+            grant_type=DataAccessGrant.GrantType.ZENDESK,
+            ticket_id="TICKET-123",
+            grant_start=grant1_start,
+            grant_end=grant1_end,
+        )
+        self.create_data_access_grant(
+            organization_id=self.organization.id,
+            grant_type=DataAccessGrant.GrantType.MANUAL,
+            granted_by_user=self.user,
+            grant_start=grant2_start,
+            grant_end=grant2_end,
+        )
+
+        result = data_access_grant_service.get_effective_grant_status(
+            organization_id=self.organization.id
+        )
+
+        assert result is not None
+        assert result.organization_id == self.organization.id
+        # Should use earliest start and latest end
+        assert result.access_start == grant1_start
+        assert result.access_end == grant2_end