Authorization Framework

castlecraft-engineer provides a flexible framework for handling authorization, ensuring that users or processes have the necessary permissions to perform actions or access data. This is primarily managed through an AuthorizationService, permission definitions, and integration with Command and Query Handlers.

1. AuthorizationService Abstract Base Class

The core of the authorization mechanism is the AuthorizationService abstract base class, located in castlecraft_engineer.authorization.base_service.AuthorizationService.

• Purpose: To define a contract for services that can check permissions.
• Key Method:
  • async check_permission(self, required_permissions: List[Permission], context: Optional[Dict[str, Any]] = None) -> None (or a synchronous version if your service is sync):
    • This method is responsible for verifying if the current actor (e.g., user, system process) possesses all the required_permissions.
    • required_permissions: A list of Permission objects that the actor must have.
    • context: An optional dictionary providing additional context for the permission check (e.g., resource IDs, user attributes).
    • If any permission is not granted, this method must raise an AuthorizationError.

You will typically implement a concrete version of this service tailored to your application's authorization logic (e.g., checking against user roles, ACLs, or a policy engine).

2. AuthorizationError Exception

When a permission check fails within the check_permission method of an AuthorizationService, it should raise an AuthorizationError. This exception is defined in castlecraft_engineer.authorization.base_service.AuthorizationError and re-exported in castlecraft_engineer.exc.

The CommandBus and QueryBus are designed to catch this specific exception if it's raised during the authorize phase of a handler, preventing the execute method from being called.

3. Default AuthorizationService Implementations

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

• DenyAllAuthorizationService: This service, as its name suggests, denies all permission checks by always raising an AuthorizationError. It's useful as a secure default if no specific authorization service is configured.
• AllowAllAuthorizationService: This service allows all permission checks by doing nothing in its check_permission method. It's useful for development, testing, or in scenarios where authorization is handled externally or not required.

4. The Permission Class and Defining Permissions

Permissions are defined using the Permission class from castlecraft_engineer.authorization.permission.Permission. A Permission object typically encapsulates an action, a resource, and a scope.

from castlecraft_engineer.authorization.permission import Permission
from castlecraft_engineer.authorization.types import Action, Resource, Scope

# Example permission definitions
CAN_CREATE_PRODUCT = Permission(action=Action.CREATE, resource=Resource.PRODUCT, scope=Scope.ANY)
CAN_READ_OWN_ORDER = Permission(action=Action.READ, resource=Resource.ORDER, scope=Scope.OWN)
CAN_UPDATE_ANY_USER_PROFILE = Permission(action=Action.UPDATE, resource=Resource.USER_PROFILE, scope=Scope.ANY)

The Action, Resource, and Scope are enums defined in castlecraft_engineer.authorization.types. These enums provide a structured way to define common permission elements.

Extending Action, Resource, Scope

For application-specific needs, you will often need to extend these enums. You can do this by creating your own enum classes that inherit from the base enums or by defining new string constants if preferred (though enums offer better type safety).

Example: Extending Enums

from enum import Enum
from castlecraft_engineer.authorization.types import Action as BaseAction, Resource as BaseResource, Scope as BaseScope

class AppAction(BaseAction):
    APPROVE = "approve"
    REJECT = "reject"

class AppResource(BaseResource):
    INVOICE = "invoice"
    TIMESHEET = "timesheet"

class AppScope(BaseScope):
    DEPARTMENT = "department"

# Using extended enums to define a permission
CAN_APPROVE_DEPARTMENT_TIMESHEET = Permission(action=AppAction.APPROVE, resource=AppResource.TIMESHEET, scope=AppScope.DEPARTMENT)

When using extended enums, ensure your custom AuthorizationService is aware of these new values.

5. The @ctx Decorator for Handlers

To associate required permissions directly with Command or Query Handlers, castlecraft-engineer provides the @ctx decorator from castlecraft_engineer.authorization.permission.ctx.

• Purpose: To declaratively specify the permissions needed to execute a handler.
• Usage: @ctx(required_permissions=[Permission(...)])

This decorator attaches the list of Permission objects to the handler class itself. This metadata can then be accessed within the handler's authorize method.

# Example: Using @ctx with a Command Handler
from castlecraft_engineer.abstractions.command import Command
from castlecraft_engineer.abstractions.command_handler import CommandHandler
from castlecraft_engineer.authorization.permission import ctx, Permission
from castlecraft_engineer.authorization.types import Action, Resource, Scope
from castlecraft_engineer.authorization.base_service import AuthorizationService
from dataclasses import dataclass
import uuid

CAN_CREATE_PRODUCT = Permission(action=Action.CREATE, resource=Resource.PRODUCT, scope=Scope.ANY)

@dataclass(frozen=True)
class CreateProductCommand(Command):
    name: str
    price: float

@ctx(required_permissions=[CAN_CREATE_PRODUCT])
class CreateProductHandler(CommandHandler[CreateProductCommand, uuid.UUID]):
    def __init__(self, auth_service: AuthorizationService): # Other deps like session, repo omitted for brevity
        self.auth_service = auth_service

    async def authorize(self, command: CreateProductCommand) -> None:
        print(f"Authorizing CreateProductCommand for '{command.name}'")
        # Access permissions attached by @ctx
        required_perms = getattr(self, 'required_permissions', [])
        await self.auth_service.check_permission(required_permissions=required_perms)
        print("Authorization successful for CreateProductCommand")

    async def execute(self, command: CreateProductCommand) -> uuid.UUID:
        product_id = uuid.uuid4()
        print(f"Executing CreateProductCommand: Product '{command.name}' (ID: {product_id}) would be created.")
        return product_id

6. Implementing the authorize Method in Handlers

The authorize method within your Command and Query Handlers is where you invoke the AuthorizationService.

1. The handler receives an instance of your AuthorizationService via dependency injection.
2. It retrieves the required_permissions (often set by the @ctx decorator or defined directly within the handler).
3. It calls auth_service.check_permission(required_permissions=...), potentially passing additional context derived from the command or query (e.g., the ID of the resource being accessed for scope checks like Scope.OWN).

Example authorize method (continuation from above):

# Inside CreateProductHandler
# async def authorize(self, command: CreateProductCommand) -> None:
#     print(f"Authorizing CreateProductCommand for '{command.name}'")
#     required_perms = getattr(self, 'required_permissions', [])
#     # Example of adding context if needed for more granular checks
#     # context = {"user_department": "sales", "resource_owner_id": command.some_owner_id}
#     await self.auth_service.check_permission(required_permissions=required_perms) #, context=context)
#     print("Authorization successful for CreateProductCommand")

7. Registering AuthorizationService with DI

Your chosen AuthorizationService implementation (custom or default) must be registered with the Dependency Injection container so it can be injected into handlers.

Using ContainerBuilder from castlecraft_engineer.common.di:

from castlecraft_engineer.common.di import ContainerBuilder
from castlecraft_engineer.authorization.base_service import AuthorizationService
from castlecraft_engineer.authorization.default_services import AllowAllAuthorizationService # Or your custom service
# from my_app.custom_auth import MyCustomAuthorizationService

builder = ContainerBuilder()

# Register your chosen authorization service
builder.with_authorization(service=AllowAllAuthorizationService) # For development/testing
# builder.with_authorization(service=MyCustomAuthorizationService) # For production

# ... other registrations (database, command/query buses, handlers, etc.)

container = builder.build()

The with_authorization method ensures that the specified service is registered against the AuthorizationService interface type.

Conclusion

The authorization framework in castlecraft-engineer provides a robust and extensible way to manage permissions. By defining Permission objects, using the @ctx decorator, implementing custom AuthorizationService logic, and integrating with Command/Query Handlers, you can build secure applications where actions and data access are appropriately controlled.