Fix require_permission dependency

src/saleor_app/deps.py

@@ -4,12 +4,13 @@
 from typing import List, Optional
 
 import jwt
+from jwt.exceptions import JWTDecodeError
 from fastapi import Depends, Header, HTTPException, Query, Request
 
 from saleor_app.saleor.exceptions import GraphQLError
 from saleor_app.saleor.mutations import VERIFY_TOKEN
 from saleor_app.saleor.utils import get_client_for_app
-from saleor_app.schemas.core import DomainName
+from saleor_app.schemas.core import DomainName, SaleorPermissions
 
 logger = logging.getLogger(__name__)
 
@@ -121,7 +122,7 @@ async def verify_webhook_signature(
         )
 
 
-def require_permission(permissions: List):
+def require_permission(permissions: List[SaleorPermissions]):
     """
     Validates is the requesting principal is authorized for the specified action
 
@@ -133,11 +134,13 @@ def require_permission(permissions: List):
     """
 
     async def func(
-        saleor_domain=Depends(saleor_domain_header),
-        saleor_token=Depends(saleor_token),
+        token=Depends(saleor_token),
         _token_is_valid=Depends(verify_saleor_token),
     ):
-        jwt_payload = jwt.decode(saleor_token, verify=False)
+        try:
+            jwt_payload = jwt.JWT().decode(token, do_verify=False)
+        except JWTDecodeError as exc:
+            raise HTTPException(status_code=400, detail=f"JWT decode error: {exc}")
         user_permissions = set(jwt_payload.get("permissions", []))
         if not set([p.value for p in permissions]) - user_permissions:
             return True

src/saleor_app/tests/test_deps.py

@@ -1,5 +1,5 @@
 import hashlib
-from unittest.mock import AsyncMock
+from unittest.mock import AsyncMock, Mock
 
 import pytest
 from fastapi import HTTPException
@@ -10,10 +10,12 @@
     verify_saleor_domain,
     verify_saleor_token,
     verify_webhook_signature,
+    require_permission,
 )
 from saleor_app.saleor.client import SaleorClient
 from saleor_app.saleor.exceptions import GraphQLError
-from saleor_app.schemas.core import WebhookData
+from saleor_app.schemas.core import WebhookData, SaleorPermissions
+from jwt.exceptions import JWTDecodeError
 
 
 async def test_saleor_domain_header_missing():
@@ -119,3 +121,41 @@ async def test_verify_webhook_signature_invalid(
         await verify_webhook_signature(mock_request, "BAD_signature", "saleor_domain")
 
     assert excinfo.value.detail == "Invalid webhook signature for x-saleor-signature"
+
+
+async def test_require_permission(mocker):
+    permissions, token = [SaleorPermissions.MANAGE_PRODUCTS], "token"
+    mock_jwt_decode = Mock(return_value={"permissions": permissions})
+    mocker.patch("saleor_app.deps.jwt.JWT.decode", mock_jwt_decode)
+
+    assert await require_permission(permissions)(token)
+    mock_jwt_decode.assert_called_once_with(token, do_verify=False)
+
+
+async def test_require_permission_when_user_unauthorized(mocker):
+    mocker.patch(
+        "saleor_app.deps.jwt.JWT.decode",
+        return_value={"permissions": [SaleorPermissions.MANAGE_PRODUCTS]},
+    )
+
+    with pytest.raises(HTTPException) as excinfo:
+        assert await require_permission([SaleorPermissions.MANAGE_APPS])("token")
+    assert excinfo.value.status_code == 403
+
+
+async def test_require_permission_when_permissions_not_provided(mocker):
+    mocker.patch("saleor_app.deps.jwt.JWT.decode", return_value={})
+
+    with pytest.raises(HTTPException) as excinfo:
+        assert await require_permission([SaleorPermissions.MANAGE_APPS])("token")
+    assert excinfo.value.status_code == 403
+
+
+async def test_require_permission_jwt_decode_error(mocker):
+    mocker.patch(
+        "saleor_app.deps.jwt.JWT.decode", side_effect=JWTDecodeError("JWT Expired")
+    )
+
+    with pytest.raises(HTTPException) as excinfo:
+        assert await require_permission([SaleorPermissions.MANAGE_APPS])("token")
+    assert excinfo.value.status_code == 400