Fix #643: preserve user-defined handlers for authorization exceptions

CHANGELOG.md

@@ -20,6 +20,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   crashed with `AttributeError` when a handler returned `AsyncIterable[ServerSentEvent]`
   (or any `collections.abc` async-iterable generic). `_try_get_schema_for_generic` now
   skips types whose origin has no `__parameters__`.
+- Fix [#643](https://github.com/Neoteroi/BlackSheep/issues/643): `use_authorization()`
+  no longer overrides exception handlers that the user has already registered for
+  `UnauthorizedError`, `ForbiddenError`, `AuthenticateChallenge`, or
+  `RateLimitExceededError`. Framework defaults are now installed via `setdefault`
+  so user-defined handlers win regardless of registration order.
 
 ## [2.6.2] - 2026-02-25 :gift:
 

blacksheep/server/application.py

@@ -528,14 +528,18 @@ def use_authorization(
             strategy.add(Policy("authenticated", AuthenticatedRequirement()))
 
         self._authorization_strategy = strategy
-        self.exceptions_handlers.update(
-            {  # type: ignore
-                AuthenticateChallenge: handle_authentication_challenge,
-                UnauthorizedError: handle_unauthorized,
-                ForbiddenError: handle_forbidden,
-                RateLimitExceededError: handle_rate_limited_auth,
-            }
-        )
+        # Install framework defaults for authorization exceptions, but only for
+        # types that the user has not already registered a handler for. This
+        # lets users register their own handlers either before or after the
+        # call to ``use_authorization()`` and keep them in effect — see #643.
+        framework_defaults = {  # type: ignore
+            AuthenticateChallenge: handle_authentication_challenge,
+            UnauthorizedError: handle_unauthorized,
+            ForbiddenError: handle_forbidden,
+            RateLimitExceededError: handle_rate_limited_auth,
+        }
+        for exception_type, default_handler in framework_defaults.items():
+            self.exceptions_handlers.setdefault(exception_type, default_handler)
 
         self.middlewares.append(
             get_authorization_middleware(strategy), MiddlewareCategory.AUTHZ, 0

tests/test_auth.py

@@ -5,17 +5,20 @@
 import pytest
 from essentials.secrets import Secret
 from guardpost import AuthorizationContext, Identity, Policy, UnauthorizedError
+from guardpost.authorization import ForbiddenError
 from guardpost.common import AuthenticatedRequirement
 from guardpost.jwks import JWKS, InMemoryKeysProvider, KeysProvider
 from guardpost.jwts import SymmetricJWTValidator
 from pytest import raises
 from rodi import Container
 
+from blacksheep import Response, TextContent
 from blacksheep.messages import Request
 from blacksheep.server.application import Application
 from blacksheep.server.authentication import (
     AuthenticateChallenge,
     AuthenticationHandler,
+    handle_authentication_challenge,
 )
 from blacksheep.server.authentication.jwt import JWTBearerAuthentication
 from blacksheep.server.authorization import (
@@ -24,6 +27,8 @@
     allow_anonymous,
     auth,
     get_www_authenticated_header_from_generic_unauthorized_error,
+    handle_forbidden,
+    handle_unauthorized,
 )
 from blacksheep.server.di import di_scope_middleware, register_http_context
 from blacksheep.server.resources import get_resource_file_path
@@ -1138,3 +1143,118 @@ async def home(request):
 
 
 # endregion
+
+
+# region user-defined exception handlers vs framework defaults — issue #643
+
+
+async def test_user_unauthorized_handler_wins_when_registered_before_use_authorization(
+    app,
+):
+    """Regression for #643.
+
+    Registering an exception handler for ``UnauthorizedError`` BEFORE
+    ``use_authorization()`` must not be silently overridden by the framework
+    defaults installed inside ``use_authorization()``.
+    """
+
+    @app.exception_handler(UnauthorizedError)
+    async def custom_unauthorized(self, request, exc):
+        return Response(401, content=TextContent("Custom unauthorized page"))
+
+    app.use_authentication().add(MockNotAuthHandler())
+    app.use_authorization().default_policy += AuthenticatedRequirement()
+
+    @auth()
+    @app.router.get("/")
+    async def home():
+        return None
+
+    await app(get_example_scope("GET", "/"), MockReceive(), MockSend())
+
+    assert app.response is not None
+    assert app.response.status == 401
+    body = await app.response.text()
+    assert body == "Custom unauthorized page"
+
+
+async def test_user_forbidden_handler_wins_when_registered_before_use_authorization(
+    app,
+):
+    """Same as above but for ``ForbiddenError`` (HTTP 403)."""
+
+    @app.exception_handler(ForbiddenError)
+    async def custom_forbidden(self, request, exc):
+        return Response(403, content=TextContent("Custom forbidden page"))
+
+    admin = Identity({"id": "001", "role": "user"}, "JWT")
+    app.use_authentication().add(MockAuthHandler(admin))
+    app.use_authorization().add(AdminsPolicy())
+
+    @auth("admin")
+    @app.router.get("/")
+    async def home():
+        return None
+
+    await app(get_example_scope("GET", "/"), MockReceive(), MockSend())
+
+    assert app.response is not None
+    assert app.response.status == 403
+    body = await app.response.text()
+    assert body == "Custom forbidden page"
+
+
+async def test_user_unauthorized_handler_wins_when_registered_after_use_authorization(
+    app,
+):
+    """Sanity: registering AFTER ``use_authorization()`` already worked before
+    #643. This test guards the existing behaviour."""
+    app.use_authentication().add(MockNotAuthHandler())
+    app.use_authorization().default_policy += AuthenticatedRequirement()
+
+    @app.exception_handler(UnauthorizedError)
+    async def custom_unauthorized(self, request, exc):
+        return Response(401, content=TextContent("Custom unauthorized page"))
+
+    @auth()
+    @app.router.get("/")
+    async def home():
+        return None
+
+    await app(get_example_scope("GET", "/"), MockReceive(), MockSend())
+
+    assert app.response is not None
+    assert app.response.status == 401
+    body = await app.response.text()
+    assert body == "Custom unauthorized page"
+
+
+async def test_framework_defaults_still_apply_when_no_user_handler(app):
+    """Sanity: without a user-defined handler, the framework defaults installed
+    by ``use_authorization()`` must remain in effect."""
+    app.use_authentication().add(MockNotAuthHandler())
+    app.use_authorization().default_policy += AuthenticatedRequirement()
+
+    @auth()
+    @app.router.get("/")
+    async def home():
+        return None
+
+    await app(get_example_scope("GET", "/"), MockReceive(), MockSend())
+
+    assert app.response is not None
+    assert app.response.status == 401
+    # Framework default body for handle_unauthorized
+    body = await app.response.text()
+    assert body == "Unauthorized"
+
+    # And the registered handler is the framework's default
+    assert app.exceptions_handlers[UnauthorizedError] is handle_unauthorized
+    assert app.exceptions_handlers[ForbiddenError] is handle_forbidden
+    assert (
+        app.exceptions_handlers[AuthenticateChallenge]
+        is handle_authentication_challenge
+    )
+
+
+# endregion