Fix jwt optional edge cases and default to more helpful error messages

refs #421

Author: Landon Gilbert-Bland

flask_jwt_extended/view_decorators.py

@@ -67,7 +67,7 @@ def verify_jwt_in_request(optional=False, fresh=False, refresh=False, locations=
             jwt_data, jwt_header, jwt_location = _decode_jwt_from_request(
                 locations, fresh
             )
-    except (NoAuthorizationError, InvalidHeaderError):
+    except NoAuthorizationError:
         if not optional:
             raise
         _request_ctx_stack.top.jwt = {}
@@ -143,34 +143,39 @@ def _decode_jwt_from_headers():
     # Verify we have the auth header
     auth_header = request.headers.get(header_name, "").strip().strip(",")
     if not auth_header:
-        raise NoAuthorizationError("Missing {} Header".format(header_name))
+        raise NoAuthorizationError(f"Missing {header_name} Header")
 
     # Make sure the header is in a valid format that we are expecting, ie
-    # <HeaderName>: <HeaderType(optional)> <JWT>
-    jwt_header = None
-
-    # Check if header is comma delimited, ie
+    # <HeaderName>: <HeaderType(optional)> <JWT>.
+    #
+    # Also handle the fact that the header that can be comma delimited, ie
     # <HeaderName>: <field> <value>, <field> <value>, etc...
     if header_type:
         field_values = split(r",\s*", auth_header)
-        jwt_header = [s for s in field_values if s.split()[0] == header_type]
-        if len(jwt_header) < 1 or len(jwt_header[0].split()) != 2:
-            msg = "Bad {} header. Expected value '{} <JWT>'".format(
-                header_name, header_type
+        jwt_headers = [s for s in field_values if s.split()[0] == header_type]
+        if len(jwt_headers) != 1:
+            msg = (
+                f"Missing '{header_type}' type in '{header_name}' header. "
+                f"Expected '{header_name}: {header_type} <JWT>'"
+            )
+            raise NoAuthorizationError(msg)
+
+        parts = jwt_headers[0].split()
+        if len(parts) != 2:
+            msg = (
+                f"Bad {header_name} header. "
+                f"Expected '{header_name}: {header_type} <JWT>'"
             )
             raise InvalidHeaderError(msg)
-        jwt_header = jwt_header[0]
-    else:
-        jwt_header = auth_header
 
-    parts = jwt_header.split()
-    if not header_type:
+        encoded_token = parts[1]
+    else:
+        parts = auth_header.split()
         if len(parts) != 1:
-            msg = "Bad {} header. Expected value '<JWT>'".format(header_name)
+            msg = f"Bad {header_name} header. Expected '{header_name}: <JWT>'"
             raise InvalidHeaderError(msg)
+
         encoded_token = parts[0]
-    else:
-        encoded_token = parts[1]
 
     return encoded_token, None
 

tests/test_headers.py

@@ -31,9 +31,12 @@ def test_default_headers(app):
     # Ensure other authorization types don't work
     access_headers = {"Authorization": "Basic basiccreds"}
     response = test_client.get("/protected", headers=access_headers)
-    expected_json = {"msg": "Bad Authorization header. Expected value 'Bearer <JWT>'"}
-    assert response.status_code == 422
-    assert response.get_json() == expected_json
+    error_msg = (
+        "Missing 'Bearer' type in 'Authorization' header. "
+        "Expected 'Authorization: Bearer <JWT>'"
+    )
+    assert response.status_code == 401
+    assert response.get_json() == {"msg": error_msg}
 
     # Ensure default headers work
     access_headers = {"Authorization": "Bearer {}".format(access_token)}
@@ -108,9 +111,12 @@ def test_custom_header_type(app):
     # Insure 'default' headers no longer work
     access_headers = {"Authorization": "Bearer {}".format(access_token)}
     response = test_client.get("/protected", headers=access_headers)
-    expected_json = {"msg": "Bad Authorization header. Expected value 'JWT <JWT>'"}
-    assert response.status_code == 422
-    assert response.get_json() == expected_json
+    error_msg = (
+        "Missing 'JWT' type in 'Authorization' header. "
+        "Expected 'Authorization: JWT <JWT>'"
+    )
+    assert response.status_code == 401
+    assert response.get_json() == {"msg": error_msg}
 
     # Insure new headers do work
     access_headers = {"Authorization": "JWT {}".format(access_token)}
@@ -141,7 +147,7 @@ def test_custom_header_type(app):
     app.config["JWT_HEADER_TYPE"] = ""
     access_headers = {"Authorization": "Bearer {}".format(access_token)}
     response = test_client.get("/protected", headers=access_headers)
-    expected_json = {"msg": "Bad Authorization header. Expected value '<JWT>'"}
+    expected_json = {"msg": "Bad Authorization header. Expected 'Authorization: <JWT>'"}
     assert response.get_json() == expected_json
     assert response.status_code == 422
 
@@ -172,7 +178,7 @@ def test_header_without_jwt(app):
     response = test_client.get("/protected", headers=access_headers)
     assert response.status_code == 422
     assert response.get_json() == {
-        "msg": "Bad Authorization header. Expected value 'Bearer <JWT>'"
+        "msg": "Bad Authorization header. Expected 'Authorization: Bearer <JWT>'"
     }
 
 

tests/test_view_decorators.py

@@ -178,13 +178,55 @@ def test_jwt_optional(app, delta_func):
     assert response.status_code == 422
     assert response.get_json() == {"msg": "Only non-refresh tokens are allowed"}
 
+    response = test_client.get(url, headers=make_headers(expired_token))
+    assert response.status_code == 401
+    assert response.get_json() == {"msg": "Token has expired"}
+
+
+def test_jwt_optional_with_no_valid_jwt(app):
+    url = "/optional_protected"
+    test_client = app.test_client()
+
+    # No auth headers
     response = test_client.get(url, headers=None)
     assert response.status_code == 200
     assert response.get_json() == {"foo": "bar"}
 
-    response = test_client.get(url, headers=make_headers(expired_token))
-    assert response.status_code == 401
-    assert response.get_json() == {"msg": "Token has expired"}
+    # auth header with type that isn't configured to be checked
+    response = test_client.get(url, headers={"Authorization": "basic creds"})
+    assert response.status_code == 200
+    assert response.get_json() == {"foo": "bar"}
+
+    # auth header with Bearer type but no JWT
+    response = test_client.get(url, headers={"Authorization": "Bearer "})
+    assert response.status_code == 422
+    assert response.get_json() == {
+        "msg": "Bad Authorization header. Expected 'Authorization: Bearer <JWT>'"
+    }
+
+    # Bearer token malformed
+    response = test_client.get(url, headers={"Authorization": "Bearer xxx"})
+    assert response.status_code == 422
+    assert response.get_json() == {"msg": "Not enough segments"}
+
+    # auth header comma seperated with no bearer token
+    response = test_client.get(url, headers={"Authorization": "Foo 1, Bar 2, Baz, 3"})
+    assert response.status_code == 200
+    assert response.get_json() == {"foo": "bar"}
+
+    # auth header comma seperated with missing bearer token
+    response = test_client.get(url, headers={"Authorization": "Foo 1, Bearer, Baz, 3"})
+    assert response.status_code == 422
+    assert response.get_json() == {
+        "msg": "Bad Authorization header. Expected 'Authorization: Bearer <JWT>'"
+    }
+
+    # Bearer token comma seperated with malformed bearer token
+    response = test_client.get(
+        url, headers={"Authorization": "Foo 1, Bearer 2, Baz, 3"}
+    )
+    assert response.status_code == 422
+    assert response.get_json() == {"msg": "Not enough segments"}
 
 
 def test_override_jwt_location(app):