Saleor is vulnerable to staff-authenticated error message information disclosure vulnerability via Python exceptions

@@ -5278,6 +5278,7 @@ def test_address_validation_rules_fields_in_camel_case(user_api_client):
             errors {
                 field
                 message
+                code
             }
         }
     }
@@ -5516,12 +5517,10 @@ def test_account_reset_password_user_is_inactive(
         "channel": channel_USD.slug,
     }
     response = user_api_client.post_graphql(REQUEST_PASSWORD_RESET_MUTATION, variables)
-    results = response.json()
-    assert "errors" in results
-    assert (
-        results["errors"][0]["message"]
-        == "Invalid token. User does not exist or is inactive."
-    )
+    content = get_graphql_content(response)
+    data = content["data"]["requestPasswordReset"]
+    assert len(data["errors"]) == 1
+    assert data["errors"][0]["code"] == AccountErrorCode.JWT_INVALID_TOKEN.name
     assert not mocked_notify.called
 
 

@@ -19,6 +19,7 @@
 from uuid import UUID
 
 import graphene
+import jwt
 from django.core.exceptions import (
     NON_FIELD_ERRORS,
     ImproperlyConfigured,
@@ -32,6 +33,7 @@
 from graphene.types.mutation import MutationOptions
 from graphql.error import GraphQLError
 
+from ...account.error_codes import AccountErrorCode
 from ...core.error_codes import MetadataErrorCode
 from ...core.exceptions import PermissionDenied
 from ...core.utils.events import call_event
@@ -502,7 +504,14 @@ def check_permissions(cls, context, permissions=None, **data):
     @classmethod
     def mutate(cls, root, info: ResolveInfo, **data):
         disallow_replica_in_context(info.context)
-        setup_context_user(info.context)
+        try:
+            setup_context_user(info.context)
+        except jwt.InvalidTokenError:
+            return cls.handle_errors(
+                ValidationError(
+                    "Invalid token", code=AccountErrorCode.JWT_INVALID_TOKEN.value
+                )
+            )
 
         if not cls.check_permissions(info.context, data=data):
             raise PermissionDenied(permissions=cls._meta.permissions)
@@ -938,7 +947,14 @@ def perform_mutation(  # type: ignore[override]
     @classmethod
     def mutate(cls, root, info: ResolveInfo, **data):
         disallow_replica_in_context(info.context)
-        setup_context_user(info.context)
+        try:
+            setup_context_user(info.context)
+        except jwt.InvalidTokenError:
+            return cls.handle_errors(
+                ValidationError(
+                    "Invalid token", code=AccountErrorCode.JWT_INVALID_TOKEN.value
+                )
+            )
 
         if not cls.check_permissions(info.context):
             raise PermissionDenied(permissions=cls._meta.permissions)

@@ -7,6 +7,7 @@
 
 from .... import __version__ as saleor_version
 from ....demo.views import EXAMPLE_QUERY
+from ....graphql.utils import INTERNAL_ERROR_MESSAGE
 from ...tests.fixtures import API_PATH
 from ...tests.utils import get_graphql_content, get_graphql_content_from_response
 from ...views import generate_cache_key
@@ -145,7 +146,7 @@ def mocked_execute(*args, **kwargs):
     response = api_client.post_graphql("{ shop { name }}")
     assert response.status_code == 400
     content = get_graphql_content_from_response(response)
-    assert content["errors"][0]["message"] == "Spanish inquisition"
+    assert content["errors"][0]["message"] == INTERNAL_ERROR_MESSAGE
 
 
 def test_invalid_query_graphql_errors_are_logged_in_another_logger(

@@ -1,7 +1,9 @@
 import re
 
+from django.test import override_settings
 from graphql.utils import schema_printer
 
+from ..utils import ALLOWED_ERRORS, INTERNAL_ERROR_MESSAGE, format_error
 from .utils import get_graphql_content
 
 
@@ -25,3 +27,25 @@ def test_multiple_interface_separator_in_schema(api_client):
 
 def test_graphql_core_contains_patched_function():
     assert hasattr(schema_printer, "_print_object")
+
+
+@override_settings(DEBUG=False)
+def test_format_error_hides_internal_error_msg_in_production_mode():
+    error = ValueError("Example error")
+    result = format_error(error, ())
+    assert result["message"] == INTERNAL_ERROR_MESSAGE
+
+
+@override_settings(DEBUG=False)
+def test_format_error_prints_allowed_errors():
+    error_cls = ALLOWED_ERRORS[0]
+    error = error_cls("Example error")
+    result = format_error(error, ())
+    assert result["message"] == str(error)
+
+
+@override_settings(DEBUG=True)
+def test_format_error_prints_internal_error_msg_in_debug_mode():
+    error = ValueError("Example error")
+    result = format_error(error, ())
+    assert result["message"] == str(error)

@@ -6,6 +6,7 @@
 
 import graphene
 from django.conf import settings
+from django.core.exceptions import ValidationError
 from django.db.models import Q, Value
 from django.db.models.functions import Concat
 from graphql import GraphQLDocument
@@ -14,9 +15,15 @@
 
 from ...account.models import User
 from ...app.models import App
+from ...core.exceptions import (
+    CircularSubscriptionSyncEvent,
+    PermissionDenied,
+    ReadOnlyException,
+)
 from ..core.enums import PermissionEnum
 from ..core.types import TYPES_WITH_DOUBLE_ID_AVAILABLE, Permission
 from ..core.utils import from_global_id_or_error
+from ..core.validators.query_cost import QueryCostError
 
 if TYPE_CHECKING:
     from ..core import SaleorContext
@@ -33,6 +40,18 @@
     "": "-",
 }
 
+# List of error types of which messages can be returned in the GraphQL API.
+ALLOWED_ERRORS = [
+    CircularSubscriptionSyncEvent,
+    GraphQLError,
+    PermissionDenied,
+    ReadOnlyException,
+    ValidationError,
+    QueryCostError,
+]
+
+INTERNAL_ERROR_MESSAGE = "Internal Server Error"
+
 
 def resolve_global_ids_to_primary_keys(
     ids: Iterable[str], graphene_type=None, raise_error: bool = False
@@ -268,6 +287,12 @@ def format_error(error, handled_exceptions):
     else:
         unhandled_errors_logger.error("A query failed unexpectedly", exc_info=exc)
 
+    # If DEBUG mode is disabled we allow only certain error messages to be returned in
+    # the API. This prevents from leaking internals that might be included in Python
+    # exceptions' error messages.
+    if type(exc) not in ALLOWED_ERRORS and not settings.DEBUG:
+        result["message"] = INTERNAL_ERROR_MESSAGE
+
     result["extensions"]["exception"] = {"code": type(exc).__name__}
     if settings.DEBUG:
         lines = []