fix: security and validation hardening for REST API management endpoints

backend/api/migrations/0123_merge_20260509_1024.py

@@ -0,0 +1,14 @@
+# Generated by Django 4.2.29 on 2026-05-09 10:24
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('api', '0120_invite_sa_inviter'),
+        ('api', '0122_sso_audit_fks_set_null'),
+    ]
+
+    operations = [
+    ]

backend/api/utils/access/permissions.py

@@ -24,10 +24,13 @@ def user_can_access_app(user_id, app_id):
     OrganisationMember = apps.get_model("api", "OrganisationMember")
     App = apps.get_model("api", "App")
 
-    app = App.objects.get(id=app_id)
-    org_member = OrganisationMember.objects.get(
-        user_id=user_id, organisation=app.organisation, deleted_at=None
-    )
+    try:
+        app = App.objects.get(id=app_id)
+        org_member = OrganisationMember.objects.get(
+            user_id=user_id, organisation=app.organisation, deleted_at=None
+        )
+    except (App.DoesNotExist, OrganisationMember.DoesNotExist):
+        return False
     return org_member in app.members.all()
 
 
@@ -36,10 +39,13 @@ def user_can_access_environment(user_id, env_id):
     Environment = apps.get_model("api", "Environment")
     EnvironmentKey = apps.get_model("api", "EnvironmentKey")
 
-    env = Environment.objects.get(id=env_id)
-    org_member = OrganisationMember.objects.get(
-        organisation=env.app.organisation, user_id=user_id, deleted_at=None
-    )
+    try:
+        env = Environment.objects.get(id=env_id)
+        org_member = OrganisationMember.objects.get(
+            organisation=env.app.organisation, user_id=user_id, deleted_at=None
+        )
+    except (Environment.DoesNotExist, OrganisationMember.DoesNotExist):
+        return False
     return EnvironmentKey.objects.filter(
         user_id=org_member, environment_id=env_id
     ).exists()
@@ -50,10 +56,13 @@ def service_account_can_access_environment(account_id, env_id):
     EnvironmentKey = apps.get_model("api", "EnvironmentKey")
     ServiceAccount = apps.get_model("api", "ServiceAccount")
 
-    env = Environment.objects.get(id=env_id)
-    service_account = ServiceAccount.objects.get(
-        organisation=env.app.organisation, id=account_id, deleted_at=None
-    )
+    try:
+        env = Environment.objects.get(id=env_id)
+        service_account = ServiceAccount.objects.get(
+            organisation=env.app.organisation, id=account_id, deleted_at=None
+        )
+    except (Environment.DoesNotExist, ServiceAccount.DoesNotExist):
+        return False
     return EnvironmentKey.objects.filter(
         service_account=service_account, environment_id=env_id
     ).exists()

backend/api/utils/rest.py

@@ -1,5 +1,8 @@
 from api.models import EnvironmentToken, ServiceAccountToken, ServiceToken, UserToken
 from django.utils import timezone
+from django.utils.html import strip_tags
+from django.core.validators import validate_email
+from django.core.exceptions import ValidationError as DjangoValidationError
 import base64
 from api.utils.access.ip import get_client_ip
 
@@ -19,51 +22,62 @@ def get_resolver_request_meta(request):
     return ip_address, user_agent
 
 
+def _parse_auth_token(auth_token):
+    """Split 'Bearer <Type> <Value>' into (token_type, token_value).
+    Returns (None, None) for any malformed input."""
+    if not auth_token:
+        return None, None
+    parts = auth_token.split(" ")
+    if len(parts) < 3 or not parts[1] or not parts[2]:
+        return None, None
+    return parts[1], parts[2]
+
+
 def get_token_type(auth_token):
-    return auth_token.split(" ")[1]
+    token_type, _ = _parse_auth_token(auth_token)
+    return token_type
 
 
 def get_env_from_service_token(auth_token):
-    token = auth_token.split(" ")[2]
-
+    _, token = _parse_auth_token(auth_token)
     if not token:
         return False
 
     try:
         env_token = EnvironmentToken.objects.get(token=token)
         return env_token.environment, env_token.user
-    except Exception as ex:
+    except Exception:
         return False
 
 
 def get_org_member_from_user_token(auth_token):
-    token = auth_token.split(" ")[2]
-
+    _, token = _parse_auth_token(auth_token)
     if not token:
         return False
 
     try:
         user_token = UserToken.objects.get(token=token)
         return user_token.user
-    except Exception as ex:
+    except Exception:
         return False
 
 
 def get_service_account_from_token(auth_token):
-    token = auth_token.split(" ")[2]
-
+    _, token = _parse_auth_token(auth_token)
     if not token:
         return False
 
     try:
         sa_token = ServiceAccountToken.objects.get(token=token)
         return sa_token.service_account
-    except Exception as ex:
+    except Exception:
         return False
 
 
 def get_service_token(auth_token):
-    prefix, token_type, token_value = auth_token.split(" ")
+    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
@@ -76,8 +90,11 @@ def get_service_token(auth_token):
 
 
 def token_is_expired_or_deleted(auth_token):
-    prefix, token_type, token_value = auth_token.split(" ")
+    token_type, token_value = _parse_auth_token(auth_token)
+    if not token_type or not token_value:
+        return True
 
+    token = None
     if token_type == "User":
         try:
             token = UserToken.objects.get(token=token_value)
@@ -96,11 +113,59 @@ def token_is_expired_or_deleted(auth_token):
         except ServiceAccountToken.DoesNotExist:
             return True
 
+    if token is None:
+        return True
+
     return token.deleted_at is not None or (
         token.expires_at is not None and token.expires_at < timezone.now()
     )
 
 
+def validate_text_field(value, field_name, max_length=None, required=True):
+    """Validate and sanitize a text field from request data.
+
+    Returns (cleaned_value, error_message).
+    If error_message is not None, the caller should return a 400 response.
+    """
+    if value is None or (isinstance(value, str) and not value.strip()):
+        if required:
+            return None, f"Missing required field: {field_name}"
+        return None, None
+
+    if not isinstance(value, str):
+        return None, f"'{field_name}' must be a string."
+
+    cleaned = strip_tags(value).strip()
+    if required and not cleaned:
+        return None, f"Missing required field: {field_name}"
+
+    if max_length and len(cleaned) > max_length:
+        return None, f"'{field_name}' cannot exceed {max_length} characters."
+
+    return cleaned, None
+
+
+def validate_email_address(email):
+    """Validate an email address using Django's built-in validator.
+
+    Returns (cleaned_email, error_message).
+    """
+    if email is None or (isinstance(email, str) and not email.strip()):
+        return None, "Missing required field: email"
+
+    if not isinstance(email, str):
+        return None, "'email' must be a string."
+
+    cleaned = email.strip().lower()
+
+    try:
+        validate_email(cleaned)
+    except DjangoValidationError:
+        return None, f"'{cleaned}' is not a valid email address."
+
+    return cleaned, None
+
+
 def encode_string_to_base64(s):
     # Convert string to bytes
     byte_representation = s.encode("utf-8")

backend/api/views/apps.py

@@ -14,14 +14,14 @@
 )
 from api.utils.audit_logging import log_audit_event, get_actor_info, build_change_values
 from api.utils.environments import create_environment
-from api.utils.rest import METHOD_TO_ACTION, get_resolver_request_meta
+from api.utils.rest import METHOD_TO_ACTION, get_resolver_request_meta, validate_text_field
 from api.throttling import PlanBasedRateThrottle
 from api.utils.access.middleware import IsIPAllowed
 from backend.quotas import can_add_app, can_use_custom_envs
 
 from rest_framework.views import APIView
 from rest_framework.permissions import IsAuthenticated
-from rest_framework.exceptions import PermissionDenied
+from rest_framework.exceptions import MethodNotAllowed, PermissionDenied
 from rest_framework.response import Response
 from rest_framework import status
 from djangorestframework_camel_case.render import CamelCaseJSONRenderer
@@ -53,7 +53,7 @@ def initial(self, request, *args, **kwargs):
 
         action = METHOD_TO_ACTION.get(request.method)
         if not action:
-            raise PermissionDenied(f"Unsupported HTTP method: {request.method}")
+            raise MethodNotAllowed(request.method)
 
         account = None
         is_sa = False
@@ -102,25 +102,15 @@ def post(self, request, *args, **kwargs):
         org = self._get_org(request)
 
         # --- Validate input ---
-        name = request.data.get("name")
-        if not name or not str(name).strip():
-            return Response(
-                {"error": "Missing required field: name"},
-                status=status.HTTP_400_BAD_REQUEST,
-            )
-        name = str(name).strip()
-        if len(name) > 64:
-            return Response(
-                {"error": "App name cannot exceed 64 characters."},
-                status=status.HTTP_400_BAD_REQUEST,
-            )
+        name, err = validate_text_field(request.data.get("name"), "name", max_length=64)
+        if err:
+            return Response({"error": err}, status=status.HTTP_400_BAD_REQUEST)
 
         description = request.data.get("description", None)
-        if description is not None and len(str(description)) > 10000:
-            return Response(
-                {"error": "App description cannot exceed 10,000 characters."},
-                status=status.HTTP_400_BAD_REQUEST,
-            )
+        if description is not None:
+            description, err = validate_text_field(description, "description", max_length=10000, required=False)
+            if err:
+                return Response({"error": err}, status=status.HTTP_400_BAD_REQUEST)
 
         # --- Validate optional environments list ---
         custom_envs = request.data.get("environments", None)
@@ -289,7 +279,7 @@ def initial(self, request, *args, **kwargs):
 
         action = METHOD_TO_ACTION.get(request.method)
         if not action:
-            raise PermissionDenied(f"Unsupported HTTP method: {request.method}")
+            raise MethodNotAllowed(request.method)
 
         account = None
         is_sa = False
@@ -321,10 +311,10 @@ def get(self, request, app_id, *args, **kwargs):
     def put(self, request, app_id, *args, **kwargs):
         app = request.auth["app"]
 
-        name = request.data.get("name")
-        description = request.data.get("description")
+        raw_name = request.data.get("name")
+        raw_desc = request.data.get("description")
 
-        if name is None and description is None:
+        if raw_name is None and raw_desc is None:
             return Response(
                 {"error": "At least one of 'name' or 'description' must be provided."},
                 status=status.HTTP_400_BAD_REQUEST,
@@ -334,25 +324,16 @@ def put(self, request, app_id, *args, **kwargs):
             app, ["name", "description"], request.data
         )
 
-        if name is not None:
-            if not name or str(name).strip() == "":
-                return Response(
-                    {"error": "App name cannot be blank."},
-                    status=status.HTTP_400_BAD_REQUEST,
-                )
-            if len(str(name)) > 64:
-                return Response(
-                    {"error": "App name cannot exceed 64 characters."},
-                    status=status.HTTP_400_BAD_REQUEST,
-                )
-            app.name = str(name).strip()
+        if raw_name is not None:
+            name, err = validate_text_field(raw_name, "name", max_length=64)
+            if err:
+                return Response({"error": err}, status=status.HTTP_400_BAD_REQUEST)
+            app.name = name
 
-        if description is not None:
-            if len(str(description)) > 10000:
-                return Response(
-                    {"error": "App description cannot exceed 10,000 characters."},
-                    status=status.HTTP_400_BAD_REQUEST,
-                )
+        if raw_desc is not None:
+            description, err = validate_text_field(raw_desc, "description", max_length=10000, required=False)
+            if err:
+                return Response({"error": err}, status=status.HTTP_400_BAD_REQUEST)
             app.description = description
 
         app.save()

backend/api/views/environments.py

@@ -17,7 +17,7 @@
 
 from rest_framework.views import APIView
 from rest_framework.permissions import IsAuthenticated
-from rest_framework.exceptions import PermissionDenied
+from rest_framework.exceptions import MethodNotAllowed, PermissionDenied
 from rest_framework.response import Response
 from rest_framework import status
 from djangorestframework_camel_case.render import CamelCaseJSONRenderer
@@ -43,7 +43,7 @@ def initial(self, request, *args, **kwargs):
 
         action = METHOD_TO_ACTION.get(request.method)
         if not action:
-            raise PermissionDenied(f"Unsupported HTTP method: {request.method}")
+            raise MethodNotAllowed(request.method)
 
         account = None
         is_sa = False
@@ -176,7 +176,7 @@ def initial(self, request, *args, **kwargs):
 
         action = METHOD_TO_ACTION.get(request.method)
         if not action:
-            raise PermissionDenied(f"Unsupported HTTP method: {request.method}")
+            raise MethodNotAllowed(request.method)
 
         account = None
         is_sa = False