Enable service account token auth for public API

engine/apps/alerts/migrations/0065_alertreceivechannel_service_account.py

@@ -0,0 +1,20 @@
+# Generated by Django 4.2.15 on 2024-11-12 13:13
+
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('user_management', '0027_serviceaccount'),
+        ('alerts', '0064_migrate_resolutionnoteslackmessage_slack_channel_id'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='alertreceivechannel',
+            name='service_account',
+            field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='alert_receive_channels', to='user_management.serviceaccount'),
+        ),
+    ]

engine/apps/alerts/models/alert_receive_channel.py

@@ -234,6 +234,13 @@ class AlertReceiveChannel(IntegrationOptionsMixin, MaintainableObject):
     author = models.ForeignKey(
         "user_management.User", on_delete=models.SET_NULL, related_name="alert_receive_channels", blank=True, null=True
     )
+    service_account = models.ForeignKey(
+        "user_management.ServiceAccount",
+        on_delete=models.SET_NULL,
+        related_name="alert_receive_channels",
+        blank=True,
+        null=True,
+    )
     team = models.ForeignKey(
         "user_management.Team",
         on_delete=models.SET_NULL,
@@ -764,15 +771,16 @@ def listen_for_alertreceivechannel_model_save(
     from apps.heartbeat.models import IntegrationHeartBeat
 
     if created:
-        write_resource_insight_log(instance=instance, author=instance.author, event=EntityEvent.CREATED)
+        author = instance.author or instance.service_account
+        write_resource_insight_log(instance=instance, author=author, event=EntityEvent.CREATED)
         default_filter = ChannelFilter(alert_receive_channel=instance, filtering_term=None, is_default=True)
         default_filter.save()
-        write_resource_insight_log(instance=default_filter, author=instance.author, event=EntityEvent.CREATED)
+        write_resource_insight_log(instance=default_filter, author=author, event=EntityEvent.CREATED)
 
         TEN_MINUTES = 600  # this is timeout for cloud heartbeats
         if instance.is_available_for_integration_heartbeat:
             heartbeat = IntegrationHeartBeat.objects.create(alert_receive_channel=instance, timeout_seconds=TEN_MINUTES)
-            write_resource_insight_log(instance=heartbeat, author=instance.author, event=EntityEvent.CREATED)
+            write_resource_insight_log(instance=heartbeat, author=author, event=EntityEvent.CREATED)
 
         metrics_add_integrations_to_cache([instance], instance.organization)
 

engine/apps/api/permissions.py

@@ -18,6 +18,7 @@
 RBAC_PERMISSIONS_ATTR = "rbac_permissions"
 RBAC_OBJECT_PERMISSIONS_ATTR = "rbac_object_permissions"
 
+
 ViewSetOrAPIView = typing.Union[ViewSet, APIView]
 
 

engine/apps/auth_token/auth.py

@@ -9,7 +9,6 @@
 from rest_framework.authentication import BaseAuthentication, get_authorization_header
 from rest_framework.request import Request
 
-from apps.api.permissions import GrafanaAPIPermissions, LegacyAccessControlRole
 from apps.grafana_plugin.helpers.gcom import check_token
 from apps.grafana_plugin.sync_data import SyncPermission, SyncUser
 from apps.user_management.exceptions import OrganizationDeletedException, OrganizationMovedException
@@ -20,13 +19,13 @@
 
 from .constants import GOOGLE_OAUTH2_AUTH_TOKEN_NAME, SCHEDULE_EXPORT_TOKEN_NAME, SLACK_AUTH_TOKEN_NAME
 from .exceptions import InvalidToken
-from .grafana.grafana_auth_token import get_service_account_token_permissions
 from .models import (
     ApiAuthToken,
     GoogleOAuth2Token,
     IntegrationBacksyncAuthToken,
     PluginAuthToken,
     ScheduleExportAuthToken,
+    ServiceAccountToken,
     SlackAuthToken,
     UserScheduleExportAuthToken,
 )
@@ -336,16 +335,16 @@ def authenticate_credentials(
         return auth_token.user, auth_token
 
 
+X_GRAFANA_URL = "X-Grafana-URL"
 X_GRAFANA_INSTANCE_ID = "X-Grafana-Instance-ID"
-GRAFANA_SA_PREFIX = "glsa_"
 
 
 class GrafanaServiceAccountAuthentication(BaseAuthentication):
     def authenticate(self, request):
         auth = get_authorization_header(request).decode("utf-8")
         if not auth:
             raise exceptions.AuthenticationFailed("Invalid token.")
-        if not auth.startswith(GRAFANA_SA_PREFIX):
+        if not auth.startswith(ServiceAccountToken.GRAFANA_SA_PREFIX):
             return None
 
         organization = self.get_organization(request)
@@ -359,6 +358,13 @@ def authenticate(self, request):
         return self.authenticate_credentials(organization, auth)
 
     def get_organization(self, request):
+        grafana_url = request.headers.get(X_GRAFANA_URL)
+        if grafana_url:
+            organization = Organization.objects.filter(grafana_url=grafana_url).first()
+            if not organization:
+                raise exceptions.AuthenticationFailed("Invalid Grafana URL.")
+            return organization
+
         if settings.LICENSE == settings.CLOUD_LICENSE_NAME:
             instance_id = request.headers.get(X_GRAFANA_INSTANCE_ID)
             if not instance_id:
@@ -370,36 +376,13 @@ def get_organization(self, request):
             return Organization.objects.filter(org_slug=org_slug, stack_slug=instance_slug).first()
 
     def authenticate_credentials(self, organization, token):
-        permissions = get_service_account_token_permissions(organization, token)
-        if not permissions:
+        try:
+            user, auth_token = ServiceAccountToken.validate_token(organization, token)
+        except InvalidToken:
             raise exceptions.AuthenticationFailed("Invalid token.")
 
-        role = LegacyAccessControlRole.NONE
-        if not organization.is_rbac_permissions_enabled:
-            role = self.determine_role_from_permissions(permissions)
-
-        user = User(
-            organization_id=organization.pk,
-            name="Grafana Service Account",
-            username="grafana_service_account",
-            role=role,
-            permissions=GrafanaAPIPermissions.construct_permissions(permissions.keys()),
-        )
-
-        auth_token = ApiAuthToken(organization=organization, user=user, name="Grafana Service Account")
-
         return user, auth_token
 
-    # Using default permissions as proxies for roles since we cannot explicitly get role from the service account token
-    def determine_role_from_permissions(self, permissions):
-        if "plugins:write" in permissions:
-            return LegacyAccessControlRole.ADMIN
-        if "dashboards:write" in permissions:
-            return LegacyAccessControlRole.EDITOR
-        if "dashboards:read" in permissions:
-            return LegacyAccessControlRole.VIEWER
-        return LegacyAccessControlRole.NONE
-
 
 class IntegrationBacksyncAuthentication(BaseAuthentication):
     model = IntegrationBacksyncAuthToken

engine/apps/auth_token/grafana/grafana_auth_token.py

@@ -46,3 +46,9 @@ def get_service_account_token_permissions(organization: Organization, token: str
     grafana_api_client = GrafanaAPIClient(api_url=organization.grafana_url, api_token=token)
     permissions, _ = grafana_api_client.get_service_account_token_permissions()
     return permissions
+
+
+def get_service_account_details(organization: Organization, token: str) -> typing.Dict[str, typing.List[str]]:
+    grafana_api_client = GrafanaAPIClient(api_url=organization.grafana_url, api_token=token)
+    user_data, _ = grafana_api_client.get_current_user()
+    return user_data

engine/apps/auth_token/migrations/0007_serviceaccounttoken.py

@@ -0,0 +1,29 @@
+# Generated by Django 4.2.15 on 2024-11-12 13:13
+
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('user_management', '0027_serviceaccount'),
+        ('auth_token', '0006_googleoauth2token'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='ServiceAccountToken',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('token_key', models.CharField(db_index=True, max_length=8)),
+                ('digest', models.CharField(max_length=128)),
+                ('created_at', models.DateTimeField(auto_now_add=True)),
+                ('revoked_at', models.DateTimeField(null=True)),
+                ('service_account', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='tokens', to='user_management.serviceaccount')),
+            ],
+            options={
+                'unique_together': {('token_key', 'service_account', 'digest')},
+            },
+        ),
+    ]

engine/apps/auth_token/models/__init__.py

@@ -4,5 +4,6 @@
 from .integration_backsync_auth_token import IntegrationBacksyncAuthToken  # noqa: F401
 from .plugin_auth_token import PluginAuthToken  # noqa: F401
 from .schedule_export_auth_token import ScheduleExportAuthToken  # noqa: F401
+from .service_account_token import ServiceAccountToken  # noqa: F401
 from .slack_auth_token import SlackAuthToken  # noqa: F401
 from .user_schedule_export_auth_token import UserScheduleExportAuthToken  # noqa: F401

engine/apps/auth_token/models/service_account_token.py

@@ -0,0 +1,110 @@
+import binascii
+from hmac import compare_digest
+
+from django.db import models
+
+from apps.api.permissions import GrafanaAPIPermissions, LegacyAccessControlRole
+from apps.auth_token import constants
+from apps.auth_token.crypto import hash_token_string
+from apps.auth_token.exceptions import InvalidToken
+from apps.auth_token.grafana.grafana_auth_token import (
+    get_service_account_details,
+    get_service_account_token_permissions,
+)
+from apps.auth_token.models import BaseAuthToken
+from apps.user_management.models import ServiceAccount, ServiceAccountUser
+
+
+class ServiceAccountTokenManager(models.Manager):
+    def get_queryset(self):
+        return super().get_queryset().select_related("service_account__organization")
+
+
+class ServiceAccountToken(BaseAuthToken):
+    GRAFANA_SA_PREFIX = "glsa_"
+
+    objects = ServiceAccountTokenManager()
+
+    service_account: "ServiceAccount"
+    service_account = models.ForeignKey(ServiceAccount, on_delete=models.CASCADE, related_name="tokens")
+
+    class Meta:
+        unique_together = ("token_key", "service_account", "digest")
+
+    @property
+    def organization(self):
+        return self.service_account.organization
+
+    @classmethod
+    def validate_token(cls, organization, token):
+        # require RBAC enabled to allow service account auth
+        if not organization.is_rbac_permissions_enabled:
+            raise InvalidToken
+
+        # Grafana API request: get permissions and confirm token is valid
+        permissions = get_service_account_token_permissions(organization, token)
+        if not permissions:
+            # NOTE: a token can be disabled/re-enabled (not setting as revoked in oncall DB for now)
+            raise InvalidToken
+
+        # check if we have already seen this token
+        validated_token = None
+        service_account = None
+        prefix_length = len(cls.GRAFANA_SA_PREFIX)
+        token_key = token[prefix_length : prefix_length + constants.TOKEN_KEY_LENGTH]
+        try:
+            hashable_token = binascii.hexlify(token.encode()).decode()
+            digest = hash_token_string(hashable_token)
+        except (TypeError, binascii.Error):
+            raise InvalidToken
+        for existing_token in cls.objects.filter(service_account__organization=organization, token_key=token_key):
+            if compare_digest(digest, existing_token.digest):
+                validated_token = existing_token
+                service_account = existing_token.service_account
+                break
+
+        if not validated_token:
+            # if it didn't match an existing token, create a new one
+            # make request to Grafana API api/user using token
+            service_account_data = get_service_account_details(organization, token)
+            if not service_account_data:
+                # Grafana versions < 11.3 return 403 trying to get user details with service account token
+                # use some default values
+                service_account_data = {
+                    "login": "grafana_service_account",
+                    "uid": None,  # "service-account:7"
+                }
+
+            grafana_id = 0  # default to zero for old Grafana versions (to keep service account unique)
+            if service_account_data["uid"] is not None:
+                # extract service account Grafana ID
+                try:
+                    grafana_id = int(service_account_data["uid"].split(":")[-1])
+                except ValueError:
+                    pass
+
+            # get or create service account
+            service_account, _ = ServiceAccount.objects.get_or_create(
+                organization=organization,
+                grafana_id=grafana_id,
+                defaults={
+                    "login": service_account_data["login"],
+                },
+            )
+            # create token
+            validated_token, _ = cls.objects.get_or_create(
+                service_account=service_account,
+                token_key=token_key,
+                digest=digest,
+            )
+
+        user = ServiceAccountUser(
+            organization=organization,
+            service_account=service_account,
+            username=service_account.username,
+            public_primary_key=service_account.public_primary_key,
+            role=LegacyAccessControlRole.NONE,
+            permissions=GrafanaAPIPermissions.construct_permissions(permissions.keys()),
+        )
+
+        return user, validated_token

engine/apps/auth_token/tests/helpers.py

@@ -0,0 +1,18 @@
+import json
+
+import httpretty
+
+
+def setup_service_account_api_mocks(organization, perms=None, user_data=None, perms_status=200, user_status=200):
+    # requires enabling httpretty
+    if perms is None:
+        perms = {}
+    mock_response = httpretty.Response(status=perms_status, body=json.dumps(perms))
+    perms_url = f"{organization.grafana_url}/api/access-control/user/permissions"
+    httpretty.register_uri(httpretty.GET, perms_url, responses=[mock_response])
+
+    if user_data is None:
+        user_data = {"login": "some-login", "uid": "service-account:42"}
+    mock_response = httpretty.Response(status=user_status, body=json.dumps(user_data))
+    user_url = f"{organization.grafana_url}/api/user"
+    httpretty.register_uri(httpretty.GET, user_url, responses=[mock_response])