fix(authz): check authorization through gateway

Author: deepthi-rao-scale

agentex/src/domain/services/authorization_service.py

@@ -6,7 +6,6 @@
 from src.adapters.authorization.adapter_agentex_authz_proxy import (
     DAgentexAuthorization,
 )
-from src.api.authentication_cache import get_auth_cache
 from src.api.authentication_middleware import DAuthorizationEnabled
 from src.api.schemas.authorization_types import (
     AgentexResource,
@@ -107,29 +106,6 @@ async def check(
             else self.principal_context
         )
 
-        use_authorization_cache = resource.type == AgentexResourceType.agent
-        cached_result = None
-        if use_authorization_cache:
-            # Try to get cached result first
-            auth_cache = await get_auth_cache()
-            cached_result = await auth_cache.get_authorization_check(
-                resource_type=str(resource.type),
-                resource_selector=resource.selector,
-                operation=str(operation),
-                principal_context=effective_principal,
-            )
-
-        if cached_result is not None:
-            logger.info(
-                "[authorization_service] Using cached result for %s permission on %s:%s: %s",
-                operation,
-                resource.type,
-                resource.selector,
-                "allowed" if cached_result else "denied",
-            )
-            return cached_result
-
-        # Not in cache, perform actual check
         logger.info(
             "[authorization_service] Checking %s permission on %s:%s",
             operation,
@@ -142,16 +118,6 @@ async def check(
             operation,
         )
 
-        if use_authorization_cache:
-            # Cache the result
-            await auth_cache.set_authorization_check(
-                resource_type=str(resource.type),
-                resource_selector=resource.selector,
-                operation=str(operation),
-                principal_context=effective_principal,
-                allowed=result,
-            )
-
         logger.info(
             f"Authorization check for {operation} on {resource.type}:{resource.selector}: {'allowed' if result else 'denied'}"
         )

agentex/tests/unit/services/test_authorization_service_cache.py

@@ -2,7 +2,6 @@
 from unittest.mock import AsyncMock
 
 import pytest
-from src.api.authentication_cache import reset_auth_cache
 from src.api.schemas.authorization_types import AgentexResource, AuthorizedOperationType
 from src.domain.services.authorization_service import AuthorizationService
 
@@ -24,53 +23,23 @@ def _service(principal_context, gateway):
     )
 
 
-@pytest.mark.unit
-@pytest.mark.asyncio
-async def test_agent_authorization_check_uses_cache():
-    await reset_auth_cache()
-    try:
-        gateway = AsyncMock()
-        gateway.check.return_value = True
-        service = _service({"user_id": "user-1", "account_id": "acct-1"}, gateway)
-
-        assert (
-            await service.check(
-                AgentexResource.agent("agent-1"), AuthorizedOperationType.read
-            )
-            is True
-        )
-        assert (
-            await service.check(
-                AgentexResource.agent("agent-1"), AuthorizedOperationType.read
-            )
-            is True
-        )
-
-        assert gateway.check.await_count == 1
-    finally:
-        await reset_auth_cache()
-
-
 @pytest.mark.unit
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     "resource",
     [
+        AgentexResource.agent("agent-1"),
         AgentexResource.task("task-1"),
         AgentexResource.api_key("api-key-1"),
         AgentexResource.schedule("agent-1/schedule-1"),
     ],
 )
-async def test_subresource_authorization_checks_call_gateway_each_time(resource):
-    await reset_auth_cache()
-    try:
-        gateway = AsyncMock()
-        gateway.check.return_value = True
-        service = _service({"user_id": "user-1", "account_id": "acct-1"}, gateway)
+async def test_authorization_checks_call_gateway_each_time(resource):
+    gateway = AsyncMock()
+    gateway.check.return_value = True
+    service = _service({"user_id": "user-1", "account_id": "acct-1"}, gateway)
 
-        assert await service.check(resource, AuthorizedOperationType.read) is True
-        assert await service.check(resource, AuthorizedOperationType.read) is True
+    assert await service.check(resource, AuthorizedOperationType.read) is True
+    assert await service.check(resource, AuthorizedOperationType.read) is True
 
-        assert gateway.check.await_count == 2
-    finally:
-        await reset_auth_cache()
+    assert gateway.check.await_count == 2