castlecraft_engineer.authorization.base_service

castlecraft_engineer.authorization.base_service

AuthorizationError

Bases: Exception

Custom exception for authorization failures.

Source code in src/castlecraft_engineer/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}"
        )

AuthorizationService

Bases: ABC

Abstract interface for authorization checks. Implementations connect to engines like Casbin, OPA, SpiceDB, etc.

Source code in src/castlecraft_engineer/authorization/base_service.py

class AuthorizationService(abc.ABC):
    """
    Abstract interface for authorization checks.
    Implementations connect to engines like Casbin,
    OPA, SpiceDB, etc.
    """

    @abc.abstractmethod
    async def check_permission(
        self,
        subject_id: Optional[str],
        required_permissions: List[Permission],
        # Optional: Pass permissions/attributes associated with the subject
        # (e.g., from token claims like roles, groups, or other string-based attributes)
        # if the engine needs them directly for evaluation.
        provided_permissions: Optional[List[str]] = None,
        # Optional: Pass additional context relevant to the decision
        # (e.g., resource owner, tenant id, command/query data)
        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.
                        Can be None for anonymous
                        checks if supported by the policy.
            required_permissions: A list of Permission objects
                                  declared by the handler
                                  via the @ctx decorator.
            provided_permissions: Optional list of permissions
                                  the subject possesses.
            context: Optional dictionary containing additional
                     data for policy evaluation.

        Returns:
            True if authorized.

        Raises:
            AuthorizationError: If the check fails.
                                This is often preferred over
                                returning False to halt execution clearly.
            NotImplementedError: If the method is not implemented.
        """
        raise NotImplementedError

check_permission(subject_id, required_permissions, provided_permissions=None, context=None) abstractmethod async

Checks if the subject has the required permissions.

Parameters:

Name	Type	Description	Default
subject_id	Optional[str]	Identifier of the user/service performing the action. Can be None for anonymous checks if supported by the policy.	required
required_permissions	List[Permission]	A list of Permission objects declared by the handler via the @ctx decorator.	required
provided_permissions	Optional[List[str]]	Optional list of permissions the subject possesses.	None
context	Optional[Dict[str, Any]]	Optional dictionary containing additional data for policy evaluation.	None

Returns:

Type	Description
bool	True if authorized.

Raises:

Type	Description
AuthorizationError	If the check fails. This is often preferred over returning False to halt execution clearly.
NotImplementedError	If the method is not implemented.

Source code in src/castlecraft_engineer/authorization/base_service.py

@abc.abstractmethod
async def check_permission(
    self,
    subject_id: Optional[str],
    required_permissions: List[Permission],
    # Optional: Pass permissions/attributes associated with the subject
    # (e.g., from token claims like roles, groups, or other string-based attributes)
    # if the engine needs them directly for evaluation.
    provided_permissions: Optional[List[str]] = None,
    # Optional: Pass additional context relevant to the decision
    # (e.g., resource owner, tenant id, command/query data)
    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.
                    Can be None for anonymous
                    checks if supported by the policy.
        required_permissions: A list of Permission objects
                              declared by the handler
                              via the @ctx decorator.
        provided_permissions: Optional list of permissions
                              the subject possesses.
        context: Optional dictionary containing additional
                 data for policy evaluation.

    Returns:
        True if authorized.

    Raises:
        AuthorizationError: If the check fails.
                            This is often preferred over
                            returning False to halt execution clearly.
        NotImplementedError: If the method is not implemented.
    """
    raise NotImplementedError