Skip to content
Draft
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 18 additions & 5 deletions agentex/src/domain/services/authorization_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,18 +31,31 @@ def __init__(
self.agent_identity = request.state.agent_identity
self.enabled = enabled

def _bypass(self) -> bool:
def _bypass_check(self) -> bool:
"""Bypass authorization checks for agent-to-agent calls or when disabled.

Used for check() and list_resources() operations where we trust agent identity.
"""
if self.agent_identity:
return True
return not self.is_enabled()

def _bypass_write(self) -> bool:
"""Bypass authorization writes only when authorization is disabled.

Used for grant() and revoke() operations. We do NOT bypass these for
agent_identity because permission records must still be created so that
users can access agent-created resources (tasks, etc.) from the UI.
"""
return not self.is_enabled()

def is_enabled(self) -> bool:
return self.enabled

async def grant(
self, resource: AgentexResource, *, commit: bool = True, principal_context=...
) -> None:
if self._bypass():
if self._bypass_write():
logger.info(
f"Authorization bypassed for grant operation on resource {resource}"
)
Expand All @@ -67,7 +80,7 @@ async def grant(
async def revoke(
self, resource: AgentexResource, *, commit: bool = True, principal_context=...
) -> None:
if self._bypass():
if self._bypass_write():
logger.info("Authorization bypassed for revoke operation")
return None

Expand Down Expand Up @@ -98,7 +111,7 @@ async def check(
*,
principal_context=...,
) -> bool:
if self._bypass():
if self._bypass_check():
logger.info("Authorization bypassed for check operation")
return True

Expand Down Expand Up @@ -166,7 +179,7 @@ async def list_resources(
) -> Iterable[str] | None:
"""List resource identifiers for which the current principal has *filter_operation* permission."""

if self._bypass():
if self._bypass_check():
logger.info("Authorization bypassed for list_resources operation")
return None

Expand Down
Loading