Fix custom response

README.md

@@ -180,6 +180,8 @@ We use GitHub pull requests. If your PR should produce a new release of authoriz
 
 Changelog
 ---------
+- v2.0.1
+  * Fix allow custom responses and preserve original responses
 - v2.0.0
   * Allow for custom (JSON) response after raised exception instead of direct 4** response.
 - v1.8.0

authorization_django/exceptions.py

@@ -44,7 +44,7 @@ def __init__(
         self,
         status_code=401,
         code="invalid_token",
-        msg="Unauthorized",
+        msg="Unauthorized. Invalid token.",
         www_authenticate='Bearer realm="datapunt", error="invalid_token"',
     ):
         super().__init__(status_code, code, msg, www_authenticate)

authorization_django/middleware.py

@@ -83,6 +83,11 @@ def authorize_forced_anonymous(_):
         raise RuntimeError("Should not call is_authorized_for in anonymous routes")
 
     def handle_exception(self, request, exception):
+        settings = get_settings()
+        if exception_handler := settings["EXCEPTION_HANDLER"]:
+            return exception_handler(
+                request, exception
+            )  # other application takes care of exception handling
         if not request.accepts("text/html"):
             if isinstance(exception, AuthorizationError):
                 payload = {
@@ -93,19 +98,11 @@ def handle_exception(self, request, exception):
                 if exception.www_authenticate:
                     response["WWW-Authenticate"] = exception.www_authenticate
         else:
-            msg = 'Bearer realm="datapunt", error={exception.code}'
+            msg = f'Bearer realm="datapunt", error={exception.code}'
             response = HttpResponse(exception.message, status=exception.status_code)
             response["WWW-Authenticate"] = msg
-        return response
 
-    def process_exception(self, request, exception):
-        settings = get_settings()
-        if exception_handler := settings["EXCEPTION_HANDLER"]:
-            return exception_handler(
-                request, exception
-            )  # other application takes care of exception handling
-        else:
-            return self.handle_exception(request, exception)
+        return response
 
     def parse_token(self, authz_header):
         """Get the token data present in the given authorization header."""
@@ -265,12 +262,14 @@ def __call__(self, request: HttpRequest):
 
         x_unique_id = request.headers.get("x-unique-id")
         authz_header = request.headers.get("authorization")
+        try:
+            if authz_header:
+                scopes, token_signature, subject, claims = self.parse_token(authz_header)
 
-        if authz_header:
-            scopes, token_signature, subject, claims = self.parse_token(authz_header)
-
-        authz_func = self.authorize_function(scopes, token_signature, x_unique_id)
-        self.handle_scope(authz_func, request)
+            authz_func = self.authorize_function(scopes, token_signature, x_unique_id)
+            self.handle_scope(authz_func, request)
+        except AuthorizationError as e:
+            return self.handle_exception(request, e)
 
         request.is_authorized_for = authz_func
         request.get_token_subject = subject

setup.py

@@ -5,7 +5,7 @@
 with open("README.md", encoding="utf-8") as f:
     long_description = f.read()
 
-version = "2.0.0"
+version = "2.0.1"
 packages = [
     "authorization_django",
     "authorization_django.extensions",

tests/test_authorization_django.py

@@ -14,8 +14,7 @@
 from jwcrypto.jwt import JWT
 
 from authorization_django import authorization_middleware, config, jwks
-from authorization_django.exceptions import AuthorizationError, InsufficientScopeError
-from authorization_django.middleware import AuthorizationMiddleware
+from authorization_django.exceptions import AuthorizationError
 
 JWKS1 = {
     "keys": [
@@ -140,7 +139,7 @@ def create_request_no_auth_header(path="/", method="GET"):
     return RequestFactory().generic(method, path)
 
 
-def custom_handler(exception):
+def custom_handler(request, exception):
     if isinstance(exception, AuthorizationError):
         return JsonResponse({"message": "Unauthorized"}, status=401)
     return None
@@ -391,11 +390,10 @@ def test_reload_jwks_from_url(requests_mock, tokendata_two_scopes):
     - still not recognize the kid
     - respond with an invalid_token response
     """
-    with pytest.raises(AuthorizationError) as e:
-        middleware(request)
+    response = middleware(request)
     assert requests_mock.call_count == 3
-    assert e.value.status_code == 401
-    assert e.value.code == "invalid_token"
+    assert response.status_code == 401
+    assert response.content == b"Unauthorized. Invalid token."
     """
     Mock requests so jwks_url returns JWKS2 and do the same request again.
     The middleware should now:
@@ -455,9 +453,8 @@ def test_entra_id_token_no_aud(middleware, tokendata_entra_id_two_scopes):
     # Remove aud claim
     tokendata_entra_id_two_scopes.pop("aud", None)
     request = create_request(tokendata_entra_id_two_scopes, "1")
-    with pytest.raises(AuthorizationError) as e:
-        middleware(request)
-    assert e.value.status_code == 401
+    response = middleware(request)
+    assert response.status_code == 401
 
 
 @pytest.mark.xfail(reason="AD Token not supported for now")
@@ -506,20 +503,18 @@ def test_invalid_token_requests(middleware, tokendata_missing_scopes, tokendata_
         create_request(tokendata_two_scopes),  # unsigned token
     )
     for request in reqs:
-        with pytest.raises(AuthorizationError) as e:
-            middleware(request)
-        assert e.value.status_code == 401
-        assert e.value.code == "invalid_token"
-        assert e.value.message == "Unauthorized"
-        assert "invalid_token" in e.value.www_authenticate
+        response = middleware(request)
+        assert response.status_code == 401
+        assert "WWW-Authenticate" in response
+        assert "invalid_token" in response["WWW-Authenticate"]
 
 
 def test_expired_token_request(middleware, tokendata_expired):
-    with pytest.raises(AuthorizationError) as e:
-        middleware(create_request(tokendata_expired, "4"))
-    assert e.value.status_code == 401
-    assert e.value.message == "Unauthorized. Token expired."
-    assert "expired_token" in e.value.www_authenticate
+    response = middleware(create_request(tokendata_expired, "4"))
+    assert response.status_code == 401
+    assert "WWW-Authenticate" in response
+    assert "expired_token" in response["WWW-Authenticate"]
+    assert response.content == b"Unauthorized. Token expired."
 
 
 def test_unknown_kid(tokendata_two_scopes):
@@ -540,10 +535,10 @@ def test_unknown_kid(tokendata_two_scopes):
         }
     )
     middleware = authorization_middleware(_ok_view)
-    with pytest.raises(AuthorizationError) as e:
-        middleware(request)
-    assert e.value.status_code == 401
-    assert "invalid_token" in e.value.www_authenticate
+    response = middleware(request)
+    assert response.status_code == 401
+    assert "WWW-Authenticate" in response
+    assert "invalid_token" in response["WWW-Authenticate"]
 
 
 def test_malformed_requests(middleware, tokendata_two_scopes):
@@ -552,11 +547,11 @@ def test_malformed_requests(middleware, tokendata_two_scopes):
         create_request(tokendata_two_scopes, "2", prefix="Even Worse"),
     )
     for request in reqs:
-        with pytest.raises(AuthorizationError) as e:
-            middleware(request)
-        assert e.value.status_code == 400
-        assert "invalid_request" in e.value.www_authenticate
-        assert e.value.message == "Invalid Authorization header format"
+        response = middleware(request)
+        assert response.status_code == 400
+        assert "WWW-Authenticate" in response
+        assert "invalid_request" in response["WWW-Authenticate"]
+        assert response.content == b"Invalid Authorization header format"
 
 
 def test_no_authorization_header(middleware):
@@ -574,9 +569,8 @@ def test_check_missing_iss(tokendata_scope1):
     reload_settings(testsettings)
     middleware = authorization_middleware(_ok_view)
     request = create_request(tokendata_scope1, "4")
-    with pytest.raises(AuthorizationError) as e:
-        middleware(request)
-    assert e.value.status_code == 401
+    response = middleware(request)
+    assert response.status_code == 401
 
 
 @pytest.mark.parametrize(["issuer", "expect_code"], [("NOT_FOOBAR", 401), ("FOOBAR", 200)])
@@ -587,12 +581,8 @@ def test_check_issuer(tokendata_issuer, issuer, expect_code):
     reload_settings(testsettings)
     middleware = authorization_middleware(_ok_view)
     request = create_request(tokendata_issuer, "4")
-    try:
-        response = middleware(request)
-    except AuthorizationError as e:
-        assert e.status_code == expect_code
-    else:
-        assert response.status_code == expect_code
+    response = middleware(request)
+    assert response.status_code == expect_code
 
 
 def test_check_correct_issuer_expired(tokendata_issuer_expired):
@@ -604,9 +594,8 @@ def test_check_correct_issuer_expired(tokendata_issuer_expired):
     reload_settings(testsettings)
     middleware = authorization_middleware(_ok_view)
     request = create_request(tokendata_issuer_expired, "4")
-    with pytest.raises(AuthorizationError) as e:
-        middleware(request)
-    assert e.value.status_code == 401
+    response = middleware(request)
+    assert response.status_code == 401
 
 
 def test_check_iss_aud_present_for_entra(tokendata_issuer_expired):
@@ -615,9 +604,8 @@ def test_check_iss_aud_present_for_entra(tokendata_issuer_expired):
     reload_settings(testsettings)
     middleware = authorization_middleware(_ok_view)
     request = create_request(tokendata_issuer_expired, "4")
-    with pytest.raises(AuthorizationError) as e:
-        middleware(request)
-    assert e.value.status_code == 401
+    response = middleware(request)
+    assert response.status_code == 401
 
 
 def test_min_scope_sufficient(tokendata_scope1):
@@ -638,36 +626,9 @@ def test_min_scope_insufficient():
     reload_settings(testsettings)
     middleware = authorization_middleware(_ok_view)
     request = create_request_no_auth_header()
-    with pytest.raises(AuthorizationError) as e:
-        middleware(request)
-    assert e.value.status_code == 401
-    assert e.value.code == "insufficient_scope"
-
-
-@pytest.mark.parametrize(
-    "request_accepts, expected_response",
-    [
-        (False, JsonResponse),
-        (True, HttpResponse),
-    ],
-)
-def test_min_scope_insufficient_response_type(
-    request_accepts,
-    expected_response,
-):
-    """if request.accepts("text/html"), an HttpResponse should be returned"""
-    testsettings = TESTSETTINGS.copy()
-    testsettings["MIN_SCOPE"] = ("scope1",)
-    reload_settings(testsettings)
-    middleware = AuthorizationMiddleware(_ok_view)
-    request = create_request_no_auth_header()
-    request.accepts = lambda _: request_accepts
-    exception = AuthorizationError(
-        401, "Unauthorized", 'Bearer realm="datapunt", error="insufficient_scope"'
-    )
-    response = middleware.process_exception(request, exception)
+    response = middleware(request)
     assert response.status_code == 401
-    assert isinstance(response, expected_response)
+    assert "insufficient_scope" in response["WWW-Authenticate"]
 
 
 def test_min_scope_as_string_sufficient(tokendata_scope1):
@@ -688,9 +649,8 @@ def test_min_scope_as_string_insufficient(tokendata_scope1):
     reload_settings(testsettings)
     middleware = authorization_middleware(_ok_view)
     request = create_request_no_auth_header()
-    with pytest.raises(AuthorizationError) as e:
-        middleware(request)
-    assert e.value.status_code == 401
+    response = middleware(request)
+    assert response.status_code == 401
 
 
 def test_min_scope_multiple_sufficient(tokendata_two_scopes):
@@ -711,10 +671,9 @@ def test_min_scope_multiple_insufficient(tokendata_scope1):
     reload_settings(testsettings)
     middleware = authorization_middleware(_ok_view)
     request = create_request(tokendata_scope1, "4")
-    with pytest.raises(AuthorizationError) as e:
-        middleware(request)
-    assert e.value.status_code == 401
-    assert "insufficient_scope" in e.value.www_authenticate
+    response = middleware(request)
+    assert response.status_code == 401
+    assert "insufficient_scope" in response["WWW-Authenticate"]
 
 
 def test_forced_anonymous_routes(rf):
@@ -761,10 +720,9 @@ def test_protected_resources_all_methods(tokendata_scope1, tokendata_two_scopes)
 
     # a token with only scope1 does not give access to two_scopes_required route
     request = create_request(tokendata_scope1, "4", "Bearer", "/two_scopes_required", "GET")
-    with pytest.raises(AuthorizationError) as e:
-        middleware(request)
-    assert e.value.status_code == 401
-    assert "insufficient_scope" in e.value.www_authenticate
+    response = middleware(request)
+    assert response.status_code == 401
+    assert "insufficient_scope" in response["WWW-Authenticate"]
 
     # a token with scope1 and scope2 gives access to two_scopes_required route
     request = create_request(tokendata_two_scopes, "4", "Bearer", "/two_scopes_required", "GET")
@@ -791,10 +749,9 @@ def test_protected_resource_read_write_distinction(tokendata_scope1, tokendata_s
     assert response.status_code == 200
 
     request = create_request(tokendata_scope1, "4", "Bearer", "/read_write_distinction", "POST")
-    with pytest.raises(AuthorizationError) as e:
-        middleware(request)
-    assert e.value.status_code == 401
-    assert "insufficient_scope" in e.value.www_authenticate
+    response = middleware(request)
+    assert response.status_code == 401
+    assert "insufficient_scope" in response["WWW-Authenticate"]
 
     request = create_request(tokendata_scope2, "4", "Bearer", "/read_write_distinction", "POST")
     response = middleware(request)
@@ -845,14 +802,29 @@ def test_protected_route_overruled_error():
         authorization_middleware(None)
 
 
-def test_custom_exception():
+def test_invalid_request_request_type(middleware, tokendata_expired):
+    """Assert a json response is returned when settings the accept header"""
+    request = create_request(tokendata_expired, "4")
+    request.META["HTTP_ACCEPT"] = "application/json"
+    response = middleware(request)
+    assert response.status_code == 401
+    assert "WWW-Authenticate" in response
+    assert "expired_token" in response["WWW-Authenticate"]
+    assert (
+        response.content
+        == b'{"error": "expired_token", "message": "Unauthorized. Token expired."}'
+    )
+
+
+def test_custom_exception(middleware):
     """test custom handler"""
     testsettings = TESTSETTINGS.copy()
     testsettings["MIN_SCOPE"] = ("scope1",)
     testsettings["EXCEPTION_HANDLER"] = custom_handler
     reload_settings(testsettings)
-    middleware = authorization_middleware(_ok_view)
     request = create_request_no_auth_header()
-    with pytest.raises(InsufficientScopeError) as e:
-        middleware(request)
-    assert e.value.status_code == 401
+    middleware = authorization_middleware(_ok_view)
+    response = middleware(request)
+    assert isinstance(response, JsonResponse)
+    assert response.status_code == 401
+    assert response.content == b'{"message": "Unauthorized"}'