from __future__ import annotations from govoplan_core.auth import ApiPrincipal, has_scope from govoplan_core.core.postbox import PostboxActorRef from govoplan_postbox.backend.permissions import ( ACKNOWLEDGE_SCOPE, BINDING_ADMIN_SCOPE, CONFIDENTIAL_SCOPE, READ_SCOPE, REPLY_SCOPE, RESTRICTED_SCOPE, SEND_SCOPE, TEMPLATE_ADMIN_SCOPE, ) class PostboxPrincipalError(ValueError): pass def actor_from_principal( principal: ApiPrincipal, *, assignment_context_id: str | None = None, ) -> PostboxActorRef: actions: set[str] = set() if has_scope(principal, READ_SCOPE): actions.update(("discover", "read")) if has_scope(principal, SEND_SCOPE): actions.add("send") if has_scope(principal, REPLY_SCOPE): actions.add("reply") if has_scope(principal, ACKNOWLEDGE_SCOPE): actions.add("acknowledge") if has_scope(principal, BINDING_ADMIN_SCOPE) or has_scope( principal, TEMPLATE_ADMIN_SCOPE, ): actions.add("administer") if ( assignment_context_id is not None and assignment_context_id not in principal.function_assignment_ids ): raise PostboxPrincipalError( "The selected assignment context is not active for this principal." ) selected = assignment_context_id or principal.acting_assignment_id if selected is None and len(principal.function_assignment_ids) == 1: selected = next(iter(principal.function_assignment_ids)) classifications = {"public", "internal"} if has_scope(principal, CONFIDENTIAL_SCOPE): classifications.add("confidential") if has_scope(principal, RESTRICTED_SCOPE): classifications.update(("confidential", "restricted")) return PostboxActorRef( account_id=principal.account_id, identity_id=principal.identity_id, selected_assignment_id=selected, acting_for_account_id=principal.acting_for_account_id, authorized_actions=frozenset(actions), # type: ignore[arg-type] authorized_classifications=frozenset(classifications), # type: ignore[arg-type] ) __all__ = ["PostboxPrincipalError", "actor_from_principal"]