refactor: retire legacy service tokens

backend/api/auth.py

@@ -2,7 +2,7 @@
 from api.utils.rest import (
     get_org_member_from_user_token,
     get_service_account_from_token,
-    get_service_token,
+    get_service_account_token,
     get_token_type,
     token_is_expired_or_deleted,
 )
@@ -30,20 +30,6 @@ def __init__(self, service_account):
         self.service_account = service_account
 
 
-class ServiceTokenUser:
-    """Synthetic principal for a Service token whose creator was deleted
-    (created_by SET_NULL). Only needs to present as authenticated so the
-    token keeps working; requests read request.auth["service_token"]."""
-
-    def __init__(self, service_token):
-        self.userId = service_token.id
-        self.id = service_token.id
-        self.is_authenticated = True
-        self.is_active = True
-        self.username = service_token.name
-        self.service_token = service_token
-
-
 def _resolve_caller_org(token_type, auth_token):
     """Best-effort lookup of the calling principal's organisation. Used
     to scope subsequent Secret/Environment/App lookups so an unrelated
@@ -59,10 +45,6 @@ def _resolve_caller_org(token_type, auth_token):
             sa = get_service_account_from_token(auth_token)
             if sa and getattr(sa, "deleted_at", None) is None:
                 return sa.organisation
-        elif token_type == "Service":
-            st = get_service_token(auth_token)
-            if st is not None:
-                return st.app.organisation
     except Exception:
         pass
     return None
@@ -78,7 +60,9 @@ def authenticate_header(self, request):
 
     def authenticate(self, request):
 
-        token_types = ["User", "Service", "ServiceAccount"]
+        # Legacy Service tokens are retired, including bootstrap requests.
+        # Reject them before loading any persisted token or target resource.
+        token_types = ["User", "ServiceAccount"]
 
         parser_context = getattr(request, "parser_context", {}) or {}
         view = parser_context.get("view")
@@ -100,7 +84,6 @@ def authenticate(self, request):
             "token": auth_token,
             "auth_type": token_type,
             "org_member": None,
-            "service_token": None,
             "service_account": None,
             "service_account_token": None,
         }
@@ -292,55 +275,28 @@ def _env_qs():
                         "User cannot access this app"
                     )
 
-        elif token_type == "Service":
-            service_token = get_service_token(auth_token)
-            if env is None:
-                if not contextless_token_bootstrap:
-                    raise exceptions.AuthenticationFailed(
-                        "Service tokens require an environment context"
-                    )
-                # The bootstrap response is token-scoped. Derive its plan and
-                # network-policy context from the token itself rather than any
-                # attacker-supplied app/environment selector.
-                auth["app"] = service_token.app
-                auth["organisation"] = service_token.app.organisation
-            elif (
-                env.app_id != service_token.app_id
-                or not service_token.keys.filter(
-                    environment_id=env.id, deleted_at=None
-                ).exists()
-            ):
-                raise exceptions.AuthenticationFailed(
-                    "Service token cannot access this environment"
-                )
-            auth["service_token"] = service_token
-            # created_by is SET_NULL: fall back to a synthetic principal so a
-            # token whose creator was deleted keeps authenticating.
-            creator = service_token.created_by
-            user = creator.user if creator else ServiceTokenUser(service_token)
-
         if token_type == "ServiceAccount":
 
             try:
-                service_token = get_service_token(auth_token)
+                service_account_token = get_service_account_token(auth_token)
                 service_account = get_service_account_from_token(auth_token)
 
-                creator = getattr(service_token, "created_by", None)
+                creator = getattr(service_account_token, "created_by", None)
                 if creator:
                     user = creator.user
                 else:
                     user = ServiceAccountUser(service_account)
 
                 auth["service_account"] = service_account
-                auth["service_account_token"] = service_token
+                auth["service_account_token"] = service_account_token
 
                 # Track last-used timestamp for SA tokens used against the
                 # REST/management API. Without this the GraphQL resolver
                 # falls back to SecretEvent history which only records
                 # E2EE secret operations, so tokens actively hitting
                 # management endpoints showed "never used".
                 from django.utils import timezone as _tz
-                ServiceAccountToken.objects.filter(id=service_token.id).update(
+                ServiceAccountToken.objects.filter(id=service_account_token.id).update(
                     last_used_at=_tz.now()
                 )
 

backend/api/models.py

@@ -16,7 +16,6 @@
     can_add_account,
     can_add_app,
     can_add_environment,
-    can_add_service_token,
 )
 from django.core.exceptions import ValidationError
 from api.utils.access.roles import MANAGED_ROLE_CHOICES
@@ -691,15 +690,9 @@ class EnvironmentToken(models.Model):
     deleted_at = models.DateTimeField(blank=True, null=True)
 
 
-class ServiceTokenManager(models.Manager):
-    def create(self, *args, **kwargs):
-        app = kwargs.get("app")
-        if not can_add_service_token(app):
-            raise ValueError("Cannot add more service tokens to this app.")
-        return super().create(*args, **kwargs)
-
-
 class ServiceToken(models.Model):
+    """Retained for historical data and audit foreign keys, not authentication."""
+
     id = models.TextField(default=uuid4, primary_key=True, editable=False)
     app = models.ForeignKey(App, on_delete=models.CASCADE)
     keys = models.ManyToManyField(EnvironmentKey)
@@ -716,7 +709,6 @@ class ServiceToken(models.Model):
     updated_at = models.DateTimeField(auto_now=True)
     deleted_at = models.DateTimeField(blank=True, null=True)
     expires_at = models.DateTimeField(null=True)
-    objects = ServiceTokenManager()
 
 
 class ServiceAccountToken(models.Model):

backend/api/serializers.py

@@ -18,7 +18,6 @@
     OrganisationMemberInvite,
     Secret,
     ServiceAccount,
-    ServiceToken,
     ServiceAccountToken,
     UserToken,
     PersonalSecret,
@@ -347,41 +346,6 @@ def to_representation(self, instance):
         return representation
 
 
-class ServiceTokenSerializer(serializers.ModelSerializer):
-    apps = EnvironmentKeySerializer(many=True, read_only=True)
-
-    organisation = OrganisationSerializer(source="app.organisation", read_only=True)
-
-    class Meta:
-        model = ServiceToken
-        fields = ["wrapped_key_share", "apps", "organisation"]
-
-    def to_representation(self, instance):
-        representation = super().to_representation(instance)
-
-        environment_keys = instance.keys.all()
-        apps = []
-        for key in environment_keys:
-            serializer = EnvironmentKeySerializer(key)
-            index = find_index_by_id(apps, key.environment.app.id)
-
-            app_data = {
-                "id": key.environment.app.id,
-                "name": key.environment.app.name,
-                "encryption": "E2E",  # Adding encryption to each app
-            }
-
-            if index == -1:
-                app_data["environment_keys"] = [serializer.data]
-                apps.append(app_data)
-            else:
-                apps[index]["environment_keys"].append(serializer.data)
-
-        representation["apps"] = apps
-
-        return representation
-
-
 class ServiceAccountTokenSerializer(serializers.ModelSerializer):
     apps = EnvironmentKeySerializer(many=True, read_only=True)
 

backend/api/throttling.py

@@ -21,8 +21,6 @@ def get_cache_key(self, request, view):
                 ident = f"user_{request.auth['org_member'].id}"
             elif request.auth.get("service_account"):
                 ident = f"sa_{request.auth['service_account'].id}"
-            elif request.auth.get("service_token"):
-                ident = f"st_{request.auth['service_token'].id}"
         else:
             ident = f"anon_{ident}"
 

backend/api/utils/access/middleware.py

@@ -23,7 +23,6 @@ def has_permission(self, request, view):
 
         org_member = request.auth.get("org_member", None)
         service_account = request.auth.get("service_account", None)
-        service_token = request.auth.get("service_token")
 
         org = None
         account_policies = NetworkAccessPolicy.objects.none()
@@ -34,8 +33,6 @@ def has_permission(self, request, view):
         elif service_account:
             account_policies = service_account.network_policies.all()
             org = service_account.organisation
-        elif service_token:
-            org = service_token.app.organisation
 
         if org is None or org.plan == Organisation.FREE_PLAN:
             return True

backend/api/utils/rest.py

@@ -1,5 +1,5 @@
 import re
-from api.models import EnvironmentToken, ServiceAccountToken, ServiceToken, UserToken
+from api.models import EnvironmentToken, ServiceAccountToken, UserToken
 from django.utils import timezone
 from django.utils.html import strip_tags
 from django.core.validators import validate_email
@@ -78,18 +78,12 @@ def get_service_account_from_token(auth_token):
         return False
 
 
-def get_service_token(auth_token):
+def get_service_account_token(auth_token):
     token_type, token_value = _parse_auth_token(auth_token)
     if not token_type or not token_value:
         return None
 
-    if token_type == "User":
-        return None
-
-    elif token_type == "Service":
-        return ServiceToken.objects.get(token=token_value)
-
-    elif token_type == "ServiceAccount":
+    if token_type == "ServiceAccount":
         return ServiceAccountToken.objects.get(token=token_value)
 
 
@@ -105,12 +99,6 @@ def token_is_expired_or_deleted(auth_token):
         except UserToken.DoesNotExist:
             return True
 
-    elif token_type == "Service":
-        try:
-            token = ServiceToken.objects.get(token=token_value)
-        except ServiceToken.DoesNotExist:
-            return True
-
     elif token_type == "ServiceAccount":
         try:
             token = ServiceAccountToken.objects.get(token=token_value)

backend/api/views/auth.py

@@ -15,7 +15,6 @@
 from backend.utils.secrets import get_secret
 from api.serializers import (
     ServiceAccountTokenSerializer,
-    ServiceTokenSerializer,
     UserTokenSerializer,
 )
 from api.auth import PhaseTokenAuthentication
@@ -92,9 +91,6 @@ def get(self, request):
         if token_type == "User":
             token = request.auth["user_token"]
             serializer = UserTokenSerializer(token)
-        elif token_type == "Service":
-            token = request.auth["service_token"]
-            serializer = ServiceTokenSerializer(token)
         else:
             token = request.auth["service_account_token"]
             serializer = ServiceAccountTokenSerializer(token)

backend/api/views/secrets.py

@@ -218,7 +218,7 @@ def get(self, request, *args, **kwargs):
             secrets,
             SecretEvent.READ,
             request.auth["org_member"],
-            request.auth["service_token"],
+            None,
             request.auth["service_account_token"],
             ip_address,
             user_agent,
@@ -430,7 +430,7 @@ def post(self, request, *args, **kwargs):
             created_secrets,
             SecretEvent.CREATE,
             request.auth["org_member"],
-            request.auth["service_token"],
+            None,
             request.auth["service_account_token"],
             ip_address,
             user_agent,
@@ -582,7 +582,7 @@ def put(self, request, *args, **kwargs):
             updated_secrets,
             SecretEvent.UPDATE,
             request.auth["org_member"],
-            request.auth["service_token"],
+            None,
             request.auth["service_account_token"],
             ip_address,
             user_agent,
@@ -637,7 +637,7 @@ def delete(self, request, *args, **kwargs):
             deleted_secrets,
             SecretEvent.DELETE,
             request.auth["org_member"],
-            request.auth["service_token"],
+            None,
             request.auth["service_account_token"],
             ip_address,
             user_agent,
@@ -739,7 +739,7 @@ def get(self, request, *args, **kwargs):
             secrets,
             SecretEvent.READ,
             request.auth["org_member"],
-            request.auth["service_token"],
+            None,
             request.auth["service_account_token"],
             ip_address,
             user_agent,
@@ -1026,7 +1026,7 @@ def post(self, request, *args, **kwargs):
             created_secrets,
             SecretEvent.CREATE,
             request.auth["org_member"],
-            request.auth["service_token"],
+            None,
             request.auth["service_account_token"],
             ip_address,
             user_agent,
@@ -1220,7 +1220,7 @@ def put(self, request, *args, **kwargs):
             updated_secrets,
             SecretEvent.UPDATE,
             request.auth["org_member"],
-            request.auth["service_token"],
+            None,
             request.auth["service_account_token"],
             ip_address,
             user_agent,
@@ -1313,7 +1313,7 @@ def delete(self, request, *args, **kwargs):
             deleted_secrets,
             SecretEvent.DELETE,
             request.auth["org_member"],
-            request.auth["service_token"],
+            None,
             request.auth["service_account_token"],
             ip_address,
             user_agent,

backend/backend/graphene/middleware.py

@@ -349,14 +349,13 @@ def _resolve_from_input_value(cls, value, request_cache):
 
     @classmethod
     def _lookup_token_org(cls, request, token_id):
-        """token_id spans four models (UserToken / ServiceToken /
-        ServiceAccountToken / EnvironmentToken); probe in order, stop
+        """token_id spans UserToken / ServiceAccountToken /
+        EnvironmentToken; probe in order, stop
         on first hit. UUIDs are globally unique so collisions can't
         happen."""
         from api.models import (
             EnvironmentToken,
             ServiceAccountToken,
-            ServiceToken,
             UserToken,
         )
         request_cache = getattr(request, cls._ID_CACHE_ATTR, {})
@@ -381,13 +380,6 @@ def _lookup_token_org(cls, request, token_id):
         except UserToken.DoesNotExist:
             pass
 
-        if not org_id:
-            try:
-                st = ServiceToken.objects.only("app_id").get(id=token_id)
-                org_id = resolve_org_id("app_id", st.app_id, request_cache)
-            except ServiceToken.DoesNotExist:
-                pass
-
         if not org_id:
             try:
                 sat = ServiceAccountToken.objects.only(