diff --git a/README.md b/README.md index 4317175..ebbb18e 100644 --- a/README.md +++ b/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 diff --git a/authorization_django/exceptions.py b/authorization_django/exceptions.py index dccf8fd..93a95ab 100644 --- a/authorization_django/exceptions.py +++ b/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) diff --git a/authorization_django/middleware.py b/authorization_django/middleware.py index 57989fd..88edf6a 100644 --- a/authorization_django/middleware.py +++ b/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 diff --git a/setup.py b/setup.py index 25831c0..72b4ee1 100644 --- a/setup.py +++ b/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", diff --git a/tests/test_authorization_django.py b/tests/test_authorization_django.py index 24de1d2..6d4d34e 100644 --- a/tests/test_authorization_django.py +++ b/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"}'