JWTIdentity raises common error JWTIdentityError

Author: Maillol

aiohttp_security/jwt_identity.py

@@ -19,6 +19,17 @@
 AUTH_SCHEME = 'Bearer '
 
 
+if HAS_JWT:
+    # This class inherits from ValueError to maintain backward compatibility
+    # with previous versions of aiohttp-security.
+    class JWTIdentityError(jwt.exceptions.PyJWTError, ValueError):
+        pass
+
+else:
+    class JWTIdentityError(ValueError):
+        pass
+
+
 class JWTIdentityPolicy(AbstractIdentityPolicy):
     def __init__(self, secret: str, algorithm: str = "HS256", key: str = "login"):
         if not HAS_JWT:
@@ -34,14 +45,15 @@ async def identify(self, request: web.Request) -> Optional[str]:
             return None
 
         if not header_identity.startswith(AUTH_SCHEME):
-            raise ValueError("Invalid authorization scheme. "
-                             + "Should be `{}<token>`".format(AUTH_SCHEME))
+            raise JWTIdentityError("Invalid authorization scheme. "
+                                   + "Should be `{}<token>`".format(AUTH_SCHEME))
 
         token = header_identity.split(' ')[1].strip()
 
         identity = jwt.decode(token,
                               self.secret,
                               algorithms=[self.algorithm])
+
         return identity.get(self.key)  # type: ignore[no-any-return]
 
     async def remember(self, request: web.Request, response: web.StreamResponse,

tests/test_jwt_identity.py

@@ -80,3 +80,28 @@ async def check(request):
     resp = await client.get('/', headers=headers)
     assert 400 == resp.status
     assert 'Invalid authorization scheme' in resp.reason
+
+
+async def test_identify_expired_signature(make_token, aiohttp_client):
+    kwt_secret_key = "Key"  # noqa: S105
+
+    token = make_token({'login': 'Andrew', 'exp': 0}, kwt_secret_key)
+
+    async def check(request):
+        policy = request.app[IDENTITY_KEY]
+        try:
+            await policy.identify(request)
+        except jwt.exceptions.PyJWTError as exc:
+            raise web.HTTPBadRequest(reason=str(exc))
+
+        return web.Response()
+
+    app = web.Application()
+    _setup(app, JWTIdentityPolicy(kwt_secret_key), Autz())
+    app.router.add_route('GET', '/', check)
+
+    client = await aiohttp_client(app)
+    headers = {"Authorization": "Bearer {}".format(token)}
+    resp = await client.get('/', headers=headers)
+    assert 400 == resp.status
+    assert 'Signature has expired' in resp.reason