Authorization in Castlecraft Engineer

Authorization is a critical aspect of most applications, determining whether a user or service (the 'subject') has the necessary rights to perform a specific action on a particular resource. castlecraft-engineer provides a flexible and extensible framework for integrating authorization checks into your command and query handlers.

Core Concepts & Components

The authorization system revolves around a few key abstractions:

1. AuthorizationService (Interface)

This is the central piece, defined in castlecraft_engineer.abstractions.authorization.base_service.AuthorizationService. It's an abstract base class that defines the contract for all authorization services.

# Excerpt from castlecraft_engineer.abstractions.authorization.base_service.py
class AuthorizationService(abc.ABC):
    @abc.abstractmethod
    async def check_permission(
        self,
        subject_id: Optional[str],
        required_permissions: List[Permission],
        provided_permissions: Optional[List[str]] = None,
        context: Optional[Dict[str, Any]] = None,
    ) -> bool:
        """
        Checks if the subject has the required permissions.

        Args:
            subject_id: Identifier of the user/service performing the action.
            required_permissions: A list of Permission objects declared by the handler.
            provided_permissions: Optional list of permissions/attributes the subject possesses (e.g., from token claims).
            context: Optional dictionary containing additional data for policy evaluation (e.g., resource details).

        Returns:
            True if authorized.

        Raises:
            AuthorizationError: If the check fails.
            NotImplementedError: If the method is not implemented.
        """
        raise NotImplementedError

Implementations of this interface would typically connect to external authorization engines like Casbin, Open Policy Agent (OPA), SpiceDB, or implement custom logic. The check_permission method is expected to raise an AuthorizationError if the subject is not authorized, rather than just returning False. This allows for a clear interruption of the execution flow.

2. Permission Dataclass

Defined in castlecraft_engineer.authorization.permission.Permission, this dataclass represents a single permission required for an operation.

# Excerpt from castlecraft_engineer.authorization.permission.py
@dataclass(frozen=True)
class Permission:
    action: BaseStringEnum  # e.g., Action.READ, Action.UPDATE
    resource: BaseStringEnum # e.g., Resource.WIDGET, 'product'
    scope: Optional[BaseStringEnum] = None # e.g., Scope.OWN, 'tenant:123'

The action, resource, and scope fields typically use string-based enums (inheriting from BaseStringEnum) for type safety and discoverability, but can also be plain strings.

3. AuthorizationError Exception

Defined in castlecraft_engineer.abstractions.authorization.base_service.AuthorizationError, this custom exception is raised by AuthorizationService.check_permission when an authorization check fails.

# Excerpt from castlecraft_engineer.abstractions.authorization.base_service.py
class AuthorizationError(Exception):
    """Custom exception for authorization failures."""
    def __init__(
        self,
        subject_id: Optional[str],
        required_permissions: List[Permission],
        message: str = "Forbidden",
    ):
        self.subject_id = subject_id
        self.required_permissions = required_permissions
        super().__init__(
            f"Subject '{subject_id}' is not authorized for required "
            f"permissions: {required_permissions}. Reason: {message}"
        )

Declaring Required Permissions with @ctx

To specify the permissions needed for a particular command or query handler's execute method, castlecraft-engineer provides the @ctx decorator (from castlecraft_engineer.authorization.permission).

from castlecraft_engineer.authorization.permission import Permission, ctx
from castlecraft_engineer.authorization.types import Action, Resource # Example enums
from castlecraft_engineer.abstractions.command_handler import CommandHandler
from your_app.commands import UpdateWidgetCommand # Example command

class UpdateWidgetCommandHandler(CommandHandler[UpdateWidgetCommand]):
    # ... (init method with AuthorizationService)

    @ctx(Permission(action=Action.UPDATE, resource=Resource.WIDGET))
    async def execute(self, command: UpdateWidgetCommand, **kwargs):
        # ... authorization check logic ...
        # ... command handling logic ...
        pass

The @ctx decorator injects a required_permissions key (as a list of Permission objects) into the kwargs of the decorated execute method. This allows the handler to access the declared permissions.

You can provide a single Permission or a list of Permission objects to @ctx if an operation requires multiple permissions to be satisfied (e.g., an AND condition, or if the policy engine handles OR conditions internally).

Performing Authorization Checks in Handlers

Inside your handler's execute method, you'll typically perform the following steps:

1. Inject AuthorizationService: Your handler should receive an instance of AuthorizationService via dependency injection.
2. Retrieve required_permissions: Get the list of Permission objects from kwargs['required_permissions'].
3. Gather Context: Collect any necessary information for the authorization decision. This might include:
   • subject_id: The ID of the user or service making the request (often passed to the handler).
   • provided_permissions: A list of permissions or attributes the subject possesses (e.g., from JWT claims, user roles).
   • context: A dictionary of additional data relevant to the policy, such as resource IDs, owner IDs, tenant information, or data from the command/query itself.
4. Call check_permission: Invoke await self._auth_service.check_permission(...) with the gathered information. If the check fails, AuthorizationError will be raised, halting further execution.

import uuid
from typing import Any, List, Optional, Dict
from castlecraft_engineer.authorization.base_service import AuthorizationService, AuthorizationError
from castlecraft_engineer.authorization.permission import Permission, ctx
from castlecraft_engineer.authorization.types import Action, Resource
from castlecraft_engineer.abstractions.command_handler import CommandHandler
from castlecraft_engineer.abstractions.command import Command # Assuming Command base class

@dataclass(frozen=True)
class UpdateWidgetCommand(Command): # Example command
    widget_id: uuid.UUID
    new_name: str
    owner_id: str # Example field for context

class UpdateWidgetCommandHandler(CommandHandler[UpdateWidgetCommand]):
    def __init__(self, auth_service: AuthorizationService):
        self._auth_service = auth_service

    @ctx(Permission(action=Action.UPDATE, resource=Resource.WIDGET))
    async def execute(
        self,
        command: UpdateWidgetCommand,
        subject_id: Optional[str] = None,
        # 'permissions' here would be the subject's possessed permissions/attributes
        permissions: Optional[List[str]] = None,
        **kwargs,
    ) -> Any:
        required: Optional[List[Permission]] = kwargs.get('required_permissions')

        if required:
            auth_context: Dict[str, Any] = {
                "resource_id": str(command.widget_id),
                "resource_owner": command.owner_id,
                # Add other relevant data from the command or environment
            }
            try:
                await self._auth_service.check_permission(
                    subject_id=subject_id,
                    required_permissions=required,
                    provided_permissions=permissions,
                    context=auth_context,
                )
            except AuthorizationError as e:
                # Log the error, re-raise, or handle as appropriate
                print(f"Authorization failed: {e}")
                raise

        # If check_permission passes (or if no permissions were required), proceed:
        print(f"Executing UpdateWidgetCommand for {command.widget_id} by {subject_id}")
        # ... actual command logic ...
        return {"status": "updated", "widget_id": command.widget_id}

Default AuthorizationService Implementations

castlecraft-engineer provides two basic implementations in castlecraft_engineer.authorization.default_services:

• DenyAllAuthorizationService: This service always denies access by raising an AuthorizationError. It's often used as a secure default if no other authorization engine is configured.
• AllowAllAuthorizationService: This service always returns True, effectively bypassing authorization checks. It should be used with extreme caution, typically only in development or testing environments.

Setup and Configuration

The setup_authorization function in castlecraft_engineer.authorization.setup handles the registration of an AuthorizationService instance into the dependency injection (DI) container (punq.Container).

This is typically called via the ContainerBuilder.with_authorization() method. You can programmatically override the configured engine by passing an engine_name to this method (e.g., builder.with_authorization(engine_name="allow_all")).

It determines which service to use based on the ENV_AUTHORIZATION_ENGINE environment variable (defined in castlecraft_engineer.common.env).

• If ENV_AUTHORIZATION_ENGINE is set to AUTH_ENGINE_ALLOW (e.g., "allow_all"), it registers AllowAllAuthorizationService.
• If ENV_AUTHORIZATION_ENGINE is set to AUTH_ENGINE_DENY (e.g., "deny_all") or is not set, it registers DenyAllAuthorizationService (default).
• If ENV_AUTHORIZATION_ENGINE is set to any other value, setup_authorization attempts to resolve an AuthorizationService instance that is assumed to have been pre-registered in the DI container by the application. This allows for custom service integration.

The chosen service is registered as a singleton in the DI container.

Extensibility: Implementing Custom AuthorizationService

To integrate with a specific authorization engine (like Casbin, OPA, a custom database-backed ACL, etc.):

1. Create a new class that inherits from AuthorizationService.
2. Implement the async def check_permission(...) method. This method will contain the logic to interact with your chosen authorization engine, evaluate the policies, and either return True or raise AuthorizationError.
3. Register an instance of your custom service in the DI container before setup_authorization is called (if you want setup_authorization to pick it up when ENV_AUTHORIZATION_ENGINE is set to a custom value), or register it directly as the AuthorizationService implementation if you are managing DI setup fully.

This approach allows castlecraft-engineer to remain unopinionated about the specific authorization technology used, providing a standard interface for handlers to request authorization checks.