Add new permission for internal api service checks (#2472)

Make naming more consistent with "users-backend" and not "users-service"

toolkit/pythonit_toolkit/api/graphql_test_client.py

@@ -74,9 +74,14 @@ def force_login(self, user: SimulatedUser):
             staff=user.is_staff,
         )
 
-    def force_service_login(self, key: Optional[str] = None):
+    def force_service_login(
+        self,
+        issuer: str = "gateway",
+        audience: str = "users-backend",
+        key: Optional[str] = None,
+    ):
         self.service_to_service_token = fake_service_to_service_token(
             str(key or self._service_to_service_secret),
-            issuer="gateway",
-            audience="users-service",
+            issuer=issuer,
+            audience=audience,
         )

toolkit/pythonit_toolkit/api/permissions.py

@@ -1,9 +1,52 @@
+from typing import Any
+
+import jwt
+from pythonit_toolkit.headers import SERVICE_JWT_HEADER
 from pythonit_toolkit.pastaporto.entities import Credential
+from pythonit_toolkit.pastaporto.tokens import decode_service_to_service_token
 from strawberry.permission import BasePermission
+from strawberry.types import Info
 
 
 class IsAuthenticated(BasePermission):
     message = "Not authenticated"
 
     def has_permission(self, source, info, **kwargs):
         return Credential.AUTHENTICATED in info.context.request.auth.scopes
+
+
+def IsService(allowed_callers: list[str], secret: str, service: str):
+    if not allowed_callers:
+        raise ValueError("No callers allowed specified")
+
+    if not secret:
+        raise ValueError("JWT secret cannot be empty")
+
+    if not service:
+        raise ValueError("Current service name cannot be empty")
+
+    secret = str(secret)
+
+    class _IsService(BasePermission):
+        message = "Forbidden"
+
+        def has_permission(self, source: Any, info: Info, **kwargs) -> bool:
+            token = info.context.request.headers.get(SERVICE_JWT_HEADER)
+
+            for caller in allowed_callers:
+                try:
+                    decode_service_to_service_token(
+                        token, secret, issuer=caller, audience=service
+                    )
+                    return True
+                except (
+                    jwt.DecodeError,
+                    jwt.InvalidIssuerError,
+                    jwt.ExpiredSignatureError,
+                    jwt.InvalidAudienceError,
+                ):
+                    pass
+
+            return False
+
+    return _IsService

toolkit/tests/api/test_permissions.py

@@ -0,0 +1,99 @@
+from unittest.mock import MagicMock
+
+from pythonit_toolkit.api.permissions import IsService
+from pythonit_toolkit.headers import SERVICE_JWT_HEADER
+from pythonit_toolkit.pastaporto.tokens import generate_service_to_service_token
+from ward import fixture, raises, test
+
+
+@fixture
+def fake_info():
+    def _fake_info(token):
+        headers = {SERVICE_JWT_HEADER: token}
+        mock_info = MagicMock()
+        mock_info.context.request.headers.get = headers.get
+        return mock_info
+
+    return _fake_info
+
+
+@test("Allowed callers pass the permission")
+async def _(fake_info=fake_info):
+    test_token = generate_service_to_service_token("test", "gateway", "users-backend")
+    mock_info = fake_info(test_token)
+
+    PermissionClass = IsService(["gateway"], "test", "users-backend")
+
+    assert PermissionClass().has_permission(None, mock_info) is True
+
+
+@test("Not allowed callers fail the permission")
+async def _(fake_info=fake_info):
+    test_token = generate_service_to_service_token(
+        "test", "pycon-backend", "users-backend"
+    )
+    mock_info = fake_info(test_token)
+
+    PermissionClass = IsService(["gateway"], "test", "users-backend")
+
+    assert PermissionClass().has_permission(None, mock_info) is False
+
+
+@test("Multiple allowed callers pass the permission")
+async def _(fake_info=fake_info):
+    test_token = generate_service_to_service_token(
+        "test", "association-backend", "users-backend"
+    )
+    mock_info = fake_info(test_token)
+
+    PermissionClass = IsService(
+        ["gateway", "association-backend"], "test", "users-backend"
+    )
+
+    assert PermissionClass().has_permission(None, mock_info) is True
+
+
+@test("Wrong secret fail permission")
+async def _(fake_info=fake_info):
+    test_token = generate_service_to_service_token(
+        "wrong-secret", "pycon-backend", "users-backend"
+    )
+    mock_info = fake_info(test_token)
+
+    PermissionClass = IsService(["gateway"], "test", "users-backend")
+
+    assert PermissionClass().has_permission(None, mock_info) is False
+
+
+@test("Token for another service fails permission")
+async def _(fake_info=fake_info):
+    test_token = generate_service_to_service_token("test", "pycon-backend", "gateway")
+    mock_info = fake_info(test_token)
+
+    PermissionClass = IsService(["pycon-backend"], "test", "users-backend")
+
+    assert PermissionClass().has_permission(None, mock_info) is False
+
+
+@test("Cannot create a permission not allowing any service")
+async def _():
+    with raises(ValueError) as exc:
+        IsService([], "test", "users-backend")
+
+    assert str(exc.raised) == "No callers allowed specified"
+
+
+@test("Cannot create a permission without any secret")
+async def _():
+    with raises(ValueError) as exc:
+        IsService(["service"], "", "users-backend")
+
+    assert str(exc.raised) == "JWT secret cannot be empty"
+
+
+@test("Cannot create a permission without the current service name")
+async def _():
+    with raises(ValueError) as exc:
+        IsService(["service"], "secret", "")
+
+    assert str(exc.raised) == "Current service name cannot be empty"

toolkit/tests/pastaporto/test_tokens.py

@@ -16,28 +16,28 @@ async def _():
             "iat": datetime.now(timezone.utc),
             "exp": datetime.now(timezone.utc) + timedelta(seconds=10),
             "iss": "test",
-            "aud": "users-service",
+            "aud": "users-backend",
         },
         "secret",
         algorithm="HS256",
     )
 
     decode_service_to_service_token(
-        test_token, "secret", issuer="test", audience="users-service"
+        test_token, "secret", issuer="test", audience="users-backend"
     )
 
 
 @test("reject tokens without expiration")
 async def _():
     test_token = jwt.encode(
-        {"iat": datetime.now(timezone.utc), "iss": "test", "aud": "users-service"},
+        {"iat": datetime.now(timezone.utc), "iss": "test", "aud": "users-backend"},
         "secret",
         algorithm="HS256",
     )
 
     with raises(jwt.MissingRequiredClaimError):
         decode_service_to_service_token(
-            test_token, "secret", issuer="test", audience="users-service"
+            test_token, "secret", issuer="test", audience="users-backend"
         )
 
 
@@ -55,7 +55,7 @@ async def _():
 
     with raises(jwt.MissingRequiredClaimError):
         decode_service_to_service_token(
-            test_token, "secret", issuer="test", audience="users-service"
+            test_token, "secret", issuer="test", audience="users-backend"
         )
 
 
@@ -66,7 +66,7 @@ async def _():
             "iat": datetime.now(timezone.utc),
             "exp": datetime.now(timezone.utc) + timedelta(seconds=10),
             "iss": "test",
-            "aud": "users-service",
+            "aud": "users-backend",
         },
         "secret",
         algorithm="HS256",

toolkit/tests/service_client/test_client.py

@@ -25,7 +25,7 @@ async def _():
         client = ServiceClient(
             url="http://localhost:8050/internal-api",
             caller="pycon",
-            service_name="users-service",
+            service_name="users-backend",
             jwt_secret="mysecret",
         )
 
@@ -56,7 +56,7 @@ async def _():
         client = ServiceClient(
             url="http://localhost:8050/internal-api",
             caller="pycon",
-            service_name="users-service",
+            service_name="users-backend",
             jwt_secret="mysecret",
         )
 
@@ -71,7 +71,7 @@ def _():
         ServiceClient(
             url="",
             caller="pycon",
-            service_name="users-service",
+            service_name="users-backend",
             jwt_secret="mysecret",
         )
 
@@ -84,7 +84,7 @@ def _():
         ServiceClient(
             url="http://localhost:8050",
             caller="",
-            service_name="users-service",
+            service_name="users-backend",
             jwt_secret="mysecret",
         )
 
@@ -110,7 +110,7 @@ def _():
         ServiceClient(
             url="http://localhost:8050",
             caller="pycon",
-            service_name="users-service",
+            service_name="users-backend",
             jwt_secret="",
         )
 

users-backend/users/internal_api/permissions.py

@@ -2,6 +2,7 @@
 from pythonit_toolkit.headers import SERVICE_JWT_HEADER
 from pythonit_toolkit.pastaporto.tokens import decode_service_to_service_token
 from starlette.requests import Request
+
 from users.settings import SERVICE_TO_SERVICE_SECRET