diff --git a/src/govoplan_postbox/backend/access_decisions.py b/src/govoplan_postbox/backend/access_decisions.py index 6fc69dc..a1d0216 100644 --- a/src/govoplan_postbox/backend/access_decisions.py +++ b/src/govoplan_postbox/backend/access_decisions.py @@ -8,6 +8,8 @@ from govoplan_core.core.postbox import ( PostboxAccessDecisionRef, PostboxAction, PostboxActorRef, + PostboxBindingStatus, + normalize_postbox_classification, ) @@ -20,9 +22,18 @@ class PostboxAccessContext: organization_unit_id: str | None function_id: str | None holder_count: int - binding_available: bool + binding_status: PostboxBindingStatus + classification: str binding_assignments: tuple[OrganizationFunctionAssignmentRef, ...] + @property + def classification_allowed(self) -> bool: + classification = normalize_postbox_classification(self.classification) + return ( + classification is not None + and classification in self.actor.authorized_classifications + ) + @property def base(self) -> dict[str, object]: return { @@ -32,6 +43,9 @@ class PostboxAccessContext: "function_id": self.function_id, "holder_count": self.holder_count, "vacant": self.holder_count == 0, + "classification": self.classification, + "classification_allowed": self.classification_allowed, + "binding_status": self.binding_status, } @@ -53,6 +67,8 @@ def evaluate_postbox_access( holder_count: int, binding_available: bool, binding_assignments: Sequence[OrganizationFunctionAssignmentRef], + binding_status: PostboxBindingStatus | None = None, + classification: str = "internal", ) -> PostboxAccessDecisionRef: context = PostboxAccessContext( postbox_id=postbox_id, @@ -62,7 +78,12 @@ def evaluate_postbox_access( organization_unit_id=organization_unit_id, function_id=function_id, holder_count=holder_count, - binding_available=binding_available, + binding_status=( + binding_status + if binding_status is not None + else "active" if binding_available else "missing" + ), + classification=classification, binding_assignments=tuple(binding_assignments), ) for rule in ACCESS_DECISION_TABLE: @@ -115,10 +136,74 @@ def _administrator(context: PostboxAccessContext) -> PostboxAccessDecisionRef: def _binding_missing(context: PostboxAccessContext) -> PostboxAccessDecisionRef: + reasons = { + "missing": ( + "function_binding_missing", + "This Postbox has no current organization-function binding.", + ), + "not_effective": ( + "function_binding_not_effective", + "The organization-function binding is not currently effective.", + ), + "unit_missing": ( + "organization_unit_missing", + "The organization unit bound to this Postbox no longer exists.", + ), + "unit_inactive": ( + "organization_unit_inactive", + "The organization unit bound to this Postbox is inactive.", + ), + "unit_tenant_mismatch": ( + "organization_unit_tenant_mismatch", + "The bound organization unit belongs to another tenant.", + ), + "function_missing": ( + "organization_function_missing", + "The organization function bound to this Postbox no longer exists.", + ), + "function_inactive": ( + "organization_function_inactive", + "The organization function bound to this Postbox is inactive.", + ), + "function_tenant_mismatch": ( + "organization_function_tenant_mismatch", + "The bound organization function belongs to another tenant.", + ), + "function_reassigned": ( + "organization_function_reassigned", + "The bound function no longer belongs to the bound organization unit.", + ), + "directory_unavailable": ( + "organization_directory_unavailable", + "The organization facts required for this access decision are unavailable.", + ), + } + reason_code, explanation = reasons.get( + context.binding_status, + reasons["missing"], + ) return _deny( context, - reason_code="function_binding_missing", - explanation="This Postbox has no active organization-function binding.", + reason_code=reason_code, + explanation=explanation, + ) + + +def _classification_denied( + context: PostboxAccessContext, +) -> PostboxAccessDecisionRef: + if normalize_postbox_classification(context.classification) is None: + return _deny( + context, + reason_code="classification_unsupported", + explanation="The Postbox uses an unsupported classification.", + ) + return _deny( + context, + reason_code="classification_clearance_missing", + explanation=( + "The account is not authorized for this Postbox classification." + ), ) @@ -215,10 +300,15 @@ ACCESS_DECISION_TABLE = ( matches=lambda context: context.action == "administer", decision=_administrator, ), + AccessRule( + name="classification_clearance", + matches=lambda context: not context.classification_allowed, + decision=_classification_denied, + ), AccessRule( name="active_function_binding", matches=lambda context: ( - not context.binding_available + context.binding_status != "active" or not context.function_id or not context.organization_unit_id ), diff --git a/src/govoplan_postbox/backend/manifest.py b/src/govoplan_postbox/backend/manifest.py index 6744bb9..270a445 100644 --- a/src/govoplan_postbox/backend/manifest.py +++ b/src/govoplan_postbox/backend/manifest.py @@ -52,10 +52,13 @@ MODULE_VERSION = "0.1.2" READ_SCOPE = "postbox:postbox:read" SEND_SCOPE = "postbox:message:write" +REPLY_SCOPE = "postbox:message:reply" ACKNOWLEDGE_SCOPE = "postbox:message:acknowledge" DELIVERY_SCOPE = "postbox:delivery:write" BINDING_ADMIN_SCOPE = "postbox:binding:admin" TEMPLATE_ADMIN_SCOPE = "postbox:template:admin" +CONFIDENTIAL_SCOPE = "postbox:classification:confidential" +RESTRICTED_SCOPE = "postbox:classification:restricted" def _permission(scope: str, label: str, description: str) -> PermissionDefinition: @@ -81,7 +84,12 @@ PERMISSIONS = ( _permission( SEND_SCOPE, "Send through assigned postboxes", - "Create replies and new messages in postboxes available through the current function context.", + "Create new messages in postboxes available through the current function context.", + ), + _permission( + REPLY_SCOPE, + "Reply through assigned postboxes", + "Reply to messages in postboxes available through the current function context.", ), _permission( ACKNOWLEDGE_SCOPE, @@ -103,6 +111,16 @@ PERMISSIONS = ( "Administer postbox templates", "Create, revise, publish, and retire reusable function-scoped postbox templates.", ), + _permission( + CONFIDENTIAL_SCOPE, + "Access confidential Postbox content", + "Discover and use confidential Postboxes and messages when function access also permits it.", + ), + _permission( + RESTRICTED_SCOPE, + "Access restricted Postbox content", + "Discover and use restricted Postboxes and messages when function access also permits it.", + ), ) ROLE_TEMPLATES = ( @@ -110,7 +128,7 @@ ROLE_TEMPLATES = ( slug="postbox_user", name="Postbox user", description="Use postboxes available through current function assignments.", - permissions=(READ_SCOPE, SEND_SCOPE, ACKNOWLEDGE_SCOPE), + permissions=(READ_SCOPE, SEND_SCOPE, REPLY_SCOPE, ACKNOWLEDGE_SCOPE), default_authenticated=True, ), RoleTemplate( @@ -120,10 +138,13 @@ ROLE_TEMPLATES = ( permissions=( READ_SCOPE, SEND_SCOPE, + REPLY_SCOPE, ACKNOWLEDGE_SCOPE, DELIVERY_SCOPE, BINDING_ADMIN_SCOPE, TEMPLATE_ADMIN_SCOPE, + CONFIDENTIAL_SCOPE, + RESTRICTED_SCOPE, ), ), ) diff --git a/src/govoplan_postbox/backend/router.py b/src/govoplan_postbox/backend/router.py index 8409872..e16b977 100644 --- a/src/govoplan_postbox/backend/router.py +++ b/src/govoplan_postbox/backend/router.py @@ -18,8 +18,11 @@ from govoplan_core.db.session import get_session from govoplan_postbox.backend.manifest import ( ACKNOWLEDGE_SCOPE, BINDING_ADMIN_SCOPE, + CONFIDENTIAL_SCOPE, DELIVERY_SCOPE, READ_SCOPE, + REPLY_SCOPE, + RESTRICTED_SCOPE, SEND_SCOPE, TEMPLATE_ADMIN_SCOPE, ) @@ -80,6 +83,8 @@ def _actor( 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( @@ -90,12 +95,18 @@ def _actor( selected = assignment_context_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] ) @@ -213,6 +224,7 @@ def api_postbox_access( "discover", "read", "send", + "reply", "acknowledge", "administer", ] = "read", diff --git a/src/govoplan_postbox/backend/schemas.py b/src/govoplan_postbox/backend/schemas.py index c8ef5ca..6e9c740 100644 --- a/src/govoplan_postbox/backend/schemas.py +++ b/src/govoplan_postbox/backend/schemas.py @@ -6,6 +6,14 @@ from typing import Any, Literal from pydantic import BaseModel, Field, model_validator +PostboxClassification = Literal[ + "public", + "internal", + "confidential", + "restricted", +] + + class PostboxAccessDecisionResponse(BaseModel): allowed: bool action: str @@ -19,6 +27,9 @@ class PostboxAccessDecisionResponse(BaseModel): selected_assignment_id: str | None = None holder_count: int = 0 vacant: bool = True + classification: str = "internal" + classification_allowed: bool = True + binding_status: str = "active" class PostboxDirectoryItem(BaseModel): @@ -132,7 +143,7 @@ class PostboxDeliveryCreateRequest(BaseModel): subject: str = Field(min_length=1, max_length=1000) body_text: str | None = None sender_label: str | None = Field(default=None, max_length=500) - classification: str = Field(default="internal", min_length=1, max_length=50) + classification: PostboxClassification = "internal" participants: list[PostboxParticipantPayload] = Field(default_factory=list) attachments: list[PostboxAttachmentPayload] = Field(default_factory=list) expires_at: datetime | None = None @@ -161,7 +172,7 @@ class PostboxLinkedCopyPolicyPayload(BaseModel): target_function_type_id: str | None = Field(default=None, max_length=36) target_template_id: str | None = Field(default=None, max_length=36) fanout: Literal["nearest", "all"] = "nearest" - allowed_classifications: list[str] = Field( + allowed_classifications: list[PostboxClassification] = Field( default_factory=lambda: ["internal"], max_length=20, ) @@ -275,7 +286,7 @@ class PostboxRoutePreviewTarget(BaseModel): class PostboxRouteDryRunRequest(BaseModel): target: PostboxTargetPayload producer_module: str = Field(min_length=1, max_length=100) - classification: str = Field(default="internal", min_length=1, max_length=50) + classification: PostboxClassification = "internal" expires_at: datetime | None = None @@ -295,7 +306,7 @@ class PostboxExactCreateRequest(BaseModel): organization_unit_id: str = Field(min_length=1, max_length=36) function_id: str = Field(min_length=1, max_length=36) address_key: str | None = Field(default=None, max_length=120) - classification: str = Field(default="internal", min_length=1, max_length=50) + classification: PostboxClassification = "internal" class PostboxTemplateRevisionPayload(BaseModel): @@ -312,7 +323,7 @@ class PostboxTemplateRevisionPayload(BaseModel): min_length=1, max_length=500, ) - classification: str = Field(default="internal", min_length=1, max_length=50) + classification: PostboxClassification = "internal" allow_vacant_delivery: bool = True routing_policy: PostboxRoutingPolicyPayload = Field( default_factory=PostboxRoutingPolicyPayload diff --git a/src/govoplan_postbox/backend/service.py b/src/govoplan_postbox/backend/service.py index 48bf499..e1c464c 100644 --- a/src/govoplan_postbox/backend/service.py +++ b/src/govoplan_postbox/backend/service.py @@ -48,6 +48,7 @@ from govoplan_core.core.postbox import ( PostboxAction, PostboxActorRef, PostboxAttachmentRef, + PostboxBindingStatus, PostboxDeliveryCatalogRef, PostboxDeliveryReceiptSummaryRef, PostboxDeliveryRequest, @@ -62,6 +63,8 @@ from govoplan_core.core.postbox import ( PostboxOrganizationUnitTargetRef, PostboxParticipantRef, PostboxTargetRef, + normalize_postbox_classification, + postbox_classification_allows, ) from govoplan_core.core.registry import PlatformRegistry from govoplan_core.security.time import utc_now @@ -501,6 +504,7 @@ class PostboxService: tenant_id=tenant_id, allowed_ids=allowed_ids, account_id=actor.account_id, + allowed_classifications=tuple(actor.authorized_classifications), query=query, state=state, ) @@ -548,6 +552,7 @@ class PostboxService: tenant_id=tenant_id, allowed_ids=allowed_ids, account_id=actor.account_id, + allowed_classifications=tuple(actor.authorized_classifications), query=query, state=state, ) @@ -563,12 +568,14 @@ class PostboxService: tenant_id: str, allowed_ids: Sequence[str], account_id: str, + allowed_classifications: Sequence[str], query: str | None, state: PostboxMessageListState, ): result = session.query(PostboxMessage).filter( PostboxMessage.tenant_id == tenant_id, PostboxMessage.postbox_id.in_(allowed_ids), + PostboxMessage.classification.in_(allowed_classifications), ) needle = (query or "").strip().casefold() if needle: @@ -628,14 +635,24 @@ class PostboxService: ) if message is None: return None - decision = self.explain_access( + decision = self._message_access_decision( db, - tenant_id=tenant_id, - postbox_id=message.postbox_id, + message=message, actor=actor, action="read", ) if not decision.allowed: + self._record_access_event( + db, + tenant_id=tenant_id, + postbox_id=message.postbox_id, + message_id=message.id, + actor=actor, + action="message.read", + outcome="denied", + reason_code=decision.reason_code, + assignment_id=decision.selected_assignment_id, + ) return None self._record_access_event( db, @@ -668,14 +685,24 @@ class PostboxService: action: PostboxAction = ( "acknowledge" if state == "acknowledged" else "read" ) - decision = self.explain_access( + decision = self._message_access_decision( db, - tenant_id=tenant_id, - postbox_id=message.postbox_id, + message=message, actor=actor, action=action, ) if not decision.allowed: + self._record_access_event( + db, + tenant_id=tenant_id, + postbox_id=message.postbox_id, + message_id=message.id, + actor=actor, + action=f"message.{state}", + outcome="denied", + reason_code=decision.reason_code, + assignment_id=decision.selected_assignment_id, + ) raise PostboxError("access_denied", decision.explanation) availability = _message_availability(message) if availability != "available": @@ -787,6 +814,15 @@ class PostboxService: ) if postbox.status != "active": raise PostboxError("postbox_inactive", "The target Postbox is not active.") + classification = self._validate_classification(request.classification) + if not postbox_classification_allows( + postbox.classification, + classification, + ): + raise PostboxError( + "classification_not_allowed", + "The message classification exceeds the target Postbox classification.", + ) holders = self._holders( request.tenant_id, @@ -806,7 +842,7 @@ class PostboxService: subject=request.subject.strip() or "(No subject)", body_text=request.body_text, status="delivered", - classification=request.classification, + classification=classification, sender_label=request.sender_label, producer_module=request.producer_module, producer_resource_type=request.producer_resource_type, @@ -947,6 +983,7 @@ class PostboxService: classification: str, expires_at: datetime | None, ) -> dict[str, object]: + classification = self._validate_classification(classification) entry = self.resolve_postbox( session, tenant_id=tenant_id, @@ -2175,6 +2212,7 @@ class PostboxService: classification: str, actor_id: str | None, ) -> Postbox: + classification = self._validate_classification(classification) unit, function = self._validate_function_target( tenant_id=tenant_id, organization_unit_id=organization_unit_id, @@ -2292,6 +2330,7 @@ class PostboxService: actor_id: str | None, routing_policy: Mapping[str, object] | None = None, ) -> PostboxTemplate: + classification = self._validate_classification(classification) clean_slug = _slug(slug or name, fallback="template") if ( session.query(PostboxTemplate) @@ -2368,6 +2407,7 @@ class PostboxService: actor_id: str | None, routing_policy: Mapping[str, object] | None = None, ) -> PostboxTemplate: + classification = self._validate_classification(classification) template = self._get_template( session, tenant_id=tenant_id, @@ -2879,6 +2919,31 @@ class PostboxService: raise PostboxError("message_not_found", "Postbox message not found.") return message + def _message_access_decision( + self, + session: Session, + *, + message: PostboxMessage, + actor: PostboxActorRef, + action: PostboxAction, + ) -> PostboxAccessDecisionRef: + postbox = self._get_postbox( + session, + tenant_id=message.tenant_id, + postbox_id=message.postbox_id, + ) + return self._access_decision( + postbox, + actor=actor, + action=action, + assignments=self._assignments_for_actor( + actor, + tenant_id=message.tenant_id, + ), + holder_cache={}, + classification=message.classification, + ) + def _get_template( self, session: Session, @@ -3007,7 +3072,14 @@ class PostboxService: } def _active_binding(self, postbox: Postbox) -> PostboxBinding | None: + return self._binding_resolution(postbox)[0] + + def _binding_resolution( + self, + postbox: Postbox, + ) -> tuple[PostboxBinding | None, PostboxBindingStatus]: now = utc_now() + current: PostboxBinding | None = None for binding in postbox.bindings: if not binding.is_active: continue @@ -3015,8 +3087,38 @@ class PostboxService: continue if binding.valid_until and binding.valid_until <= now: continue - return binding - return None + current = binding + break + if current is None: + return (None, "not_effective" if postbox.bindings else "missing") + if not current.organization_unit_id or not current.function_id: + return current, "missing" + try: + unit = self._organizations.get_organization_unit( + current.organization_unit_id + ) + function = self._organizations.get_function(current.function_id) + except Exception: + logger.exception( + "Postbox organization binding resolution failed", + extra={"postbox_id": postbox.id}, + ) + return current, "directory_unavailable" + if unit is None: + return current, "unit_missing" + if unit.tenant_id != postbox.tenant_id: + return current, "unit_tenant_mismatch" + if unit.status != "active": + return current, "unit_inactive" + if function is None: + return current, "function_missing" + if function.tenant_id != postbox.tenant_id: + return current, "function_tenant_mismatch" + if function.status != "active": + return current, "function_inactive" + if function.organization_unit_id != unit.id: + return current, "function_reassigned" + return current, "active" def _access_decision( self, @@ -3026,8 +3128,9 @@ class PostboxService: action: PostboxAction, assignments: Sequence[OrganizationFunctionAssignmentRef], holder_cache: dict[str, tuple[OrganizationFunctionAssignmentRef, ...]], + classification: str | None = None, ) -> PostboxAccessDecisionRef: - binding = self._active_binding(postbox) + binding, binding_status = self._binding_resolution(postbox) function_id = ( binding.function_id if binding is not None @@ -3041,7 +3144,7 @@ class PostboxService: holders = self._holders(postbox.tenant_id, function_id, holder_cache) holder_count = len({holder.identity_id for holder in holders}) binding_assignments: list[OrganizationFunctionAssignmentRef] = [] - if binding is not None: + if binding is not None and binding_status == "active": for assignment in assignments: if assignment.tenant_id != postbox.tenant_id: continue @@ -3056,8 +3159,10 @@ class PostboxService: organization_unit_id=unit_id, function_id=function_id, holder_count=holder_count, - binding_available=binding is not None, + binding_available=binding is not None and binding_status == "active", binding_assignments=binding_assignments, + binding_status=binding_status, + classification=classification or postbox.classification, ) def _assignment_matches_binding( @@ -3469,6 +3574,15 @@ class PostboxService: ) return unit, function + def _validate_classification(self, classification: str) -> str: + normalized = normalize_postbox_classification(classification) + if normalized is None: + raise PostboxError( + "classification_unsupported", + "Postbox classification must be public, internal, confidential, or restricted.", + ) + return normalized + def _validate_scope( self, *, diff --git a/tests/test_access_decisions.py b/tests/test_access_decisions.py index 029d162..dcc89e1 100644 --- a/tests/test_access_decisions.py +++ b/tests/test_access_decisions.py @@ -38,6 +38,11 @@ def decide( assignments=(), selected_assignment_id: str | None = None, acting_for_account_id: str | None = None, + classification: str = "internal", + authorized_classifications: frozenset[str] = frozenset( + {"public", "internal"} + ), + binding_status: str | None = None, ): return evaluate_postbox_access( postbox_id="postbox-1", @@ -49,12 +54,15 @@ def decide( selected_assignment_id=selected_assignment_id, acting_for_account_id=acting_for_account_id, authorized_actions=authorized_actions, # type: ignore[arg-type] + authorized_classifications=authorized_classifications, # type: ignore[arg-type] ), organization_unit_id="unit-1" if binding_available else None, function_id="function-1" if binding_available else None, holder_count=len(assignments), binding_available=binding_available, binding_assignments=assignments, + binding_status=binding_status, # type: ignore[arg-type] + classification=classification, ) @@ -66,6 +74,7 @@ class PostboxAccessDecisionTableTests(unittest.TestCase): "inactive_postbox", "generic_permission", "administrator", + "classification_clearance", "active_function_binding", ], ) @@ -102,6 +111,65 @@ class PostboxAccessDecisionTableTests(unittest.TestCase): self.assertTrue(decision.allowed) self.assertEqual(decision.reason_code, "generic_administrator") + def test_reply_is_a_distinct_permission_decision(self) -> None: + missing = decide( + action="reply", + authorized_actions=frozenset({"send"}), + assignments=(assignment(),), + ) + allowed = decide( + action="reply", + authorized_actions=frozenset({"reply"}), + assignments=(assignment(),), + ) + + self.assertEqual(missing.reason_code, "generic_permission_missing") + self.assertTrue(allowed.allowed) + + def test_classification_is_fail_closed_and_explained(self) -> None: + missing_clearance = decide( + classification="confidential", + assignments=(assignment(),), + ) + allowed = decide( + classification="confidential", + authorized_classifications=frozenset( + {"public", "internal", "confidential"} + ), + assignments=(assignment(),), + ) + unsupported = decide( + classification="secret", + authorized_classifications=frozenset( + {"public", "internal", "confidential", "restricted"} + ), + assignments=(assignment(),), + ) + + self.assertEqual( + missing_clearance.reason_code, + "classification_clearance_missing", + ) + self.assertFalse(missing_clearance.classification_allowed) + self.assertTrue(allowed.allowed) + self.assertEqual(unsupported.reason_code, "classification_unsupported") + + def test_binding_failures_have_stable_transition_reasons(self) -> None: + expected = { + "unit_inactive": "organization_unit_inactive", + "function_inactive": "organization_function_inactive", + "function_reassigned": "organization_function_reassigned", + "directory_unavailable": "organization_directory_unavailable", + } + for binding_status, reason_code in expected.items(): + with self.subTest(binding_status=binding_status): + decision = decide( + assignments=(assignment(),), + binding_status=binding_status, + ) + self.assertFalse(decision.allowed) + self.assertEqual(decision.reason_code, reason_code) + def test_direct_delegated_directory_governance_and_system_sources_are_allowed(self) -> None: for source in ( "direct", diff --git a/tests/test_real_directory_access.py b/tests/test_real_directory_access.py new file mode 100644 index 0000000..9646834 --- /dev/null +++ b/tests/test_real_directory_access.py @@ -0,0 +1,310 @@ +from __future__ import annotations + +import unittest +from datetime import timedelta + +from sqlalchemy import create_engine +from sqlalchemy.orm import Session + +from govoplan_core.core.postbox import PostboxActorRef +from govoplan_core.db.base import Base +from govoplan_core.db.session import ( + DatabaseHandle, + get_database, + reset_database, + set_database, +) +from govoplan_core.security.time import utc_now +from govoplan_identity.backend.db.models import Identity, IdentityAccountLink +from govoplan_identity.backend.directory import SqlIdentityDirectory +from govoplan_idm.backend.db.models import IdmOrganizationFunctionAssignment +from govoplan_idm.backend.directory import SqlIdmDirectory +from govoplan_organizations.backend.db.models import ( + OrganizationFunction, + OrganizationUnit, +) +from govoplan_organizations.backend.directory import SqlOrganizationDirectory +from govoplan_postbox.backend.db.models import ( + Postbox, + PostboxAccessEvent, + PostboxAddress, + PostboxAttachmentReference, + PostboxBinding, + PostboxDelivery, + PostboxGrouping, + PostboxGroupingSource, + PostboxMessage, + PostboxMessageReceipt, + PostboxParticipant, + PostboxRoute, + PostboxTemplate, + PostboxTemplateRevision, +) +from govoplan_postbox.backend.service import PostboxService + + +TABLES = ( + Identity.__table__, + IdentityAccountLink.__table__, + OrganizationUnit.__table__, + OrganizationFunction.__table__, + IdmOrganizationFunctionAssignment.__table__, + PostboxTemplate.__table__, + PostboxTemplateRevision.__table__, + PostboxAddress.__table__, + Postbox.__table__, + PostboxBinding.__table__, + PostboxMessage.__table__, + PostboxParticipant.__table__, + PostboxAttachmentReference.__table__, + PostboxDelivery.__table__, + PostboxRoute.__table__, + PostboxMessageReceipt.__table__, + PostboxGrouping.__table__, + PostboxGroupingSource.__table__, + PostboxAccessEvent.__table__, +) + + +class PostboxRealDirectoryAccessTests(unittest.TestCase): + def setUp(self) -> None: + try: + self.previous_database = get_database() + except RuntimeError: + self.previous_database = None + self.engine = create_engine("sqlite:///:memory:") + Base.metadata.create_all(self.engine, tables=TABLES) + self.database = DatabaseHandle("sqlite:///:memory:", engine=self.engine) + set_database(self.database) + self.organizations = SqlOrganizationDirectory( + session_factory=self.database.SessionLocal + ) + self.identities = SqlIdentityDirectory() + self.idm = SqlIdmDirectory( + identities=self.identities, + organizations=self.organizations, + ) + self.service = PostboxService( + identities=self.identities, + idm=self.idm, + incumbencies=self.idm, + organizations=self.organizations, + ) + with self.database.SessionLocal() as session: + session.add_all( + ( + Identity( + id="identity-owner", + display_name="Owner", + source="test", + is_active=True, + settings={}, + ), + IdentityAccountLink( + id="link-owner", + identity_id="identity-owner", + account_id="account-owner", + is_primary=True, + source="test", + ), + Identity( + id="identity-delegate", + display_name="Delegate", + source="test", + is_active=True, + settings={}, + ), + IdentityAccountLink( + id="link-delegate", + identity_id="identity-delegate", + account_id="account-delegate", + is_primary=True, + source="test", + ), + OrganizationUnit( + id="unit-one", + tenant_id="tenant-1", + slug="unit-one", + name="Unit One", + is_active=True, + settings={}, + ), + OrganizationUnit( + id="unit-two", + tenant_id="tenant-1", + slug="unit-two", + name="Unit Two", + is_active=True, + settings={}, + ), + OrganizationFunction( + id="function-one", + tenant_id="tenant-1", + organization_unit_id="unit-one", + slug="clerk", + name="Clerk", + delegable=True, + is_active=True, + settings={}, + ), + ) + ) + session.commit() + postbox = self.service.create_exact_postbox( + session, + tenant_id="tenant-1", + name="Unit One / Clerk", + organization_unit_id="unit-one", + function_id="function-one", + address_key=None, + description=None, + classification="internal", + actor_id="admin-1", + ) + session.commit() + self.postbox_id = postbox.id + + def tearDown(self) -> None: + if self.previous_database is None: + reset_database() + else: + set_database(self.previous_database) + self.database.dispose() + + def _actor(self, account_id: str) -> PostboxActorRef: + return PostboxActorRef( + account_id=account_id, + authorized_actions=frozenset({"discover", "read", "reply"}), + ) + + def _decision(self, account_id: str): + with Session(self.engine) as session: + return self.service.explain_access( + session, + tenant_id="tenant-1", + postbox_id=self.postbox_id, + actor=self._actor(account_id), + action="read", + ) + + def _add_assignment( + self, + *, + assignment_id: str, + identity_id: str, + account_id: str, + source: str = "direct", + delegated_from_assignment_id: str | None = None, + valid_until=None, + ) -> None: + with self.database.SessionLocal() as session: + session.add( + IdmOrganizationFunctionAssignment( + id=assignment_id, + tenant_id="tenant-1", + identity_id=identity_id, + account_id=account_id, + function_id="function-one", + organization_unit_id="unit-one", + source=source, + delegated_from_assignment_id=delegated_from_assignment_id, + valid_until=valid_until, + is_active=True, + settings={}, + ) + ) + session.commit() + + def test_real_directory_reassignment_changes_current_holder_only(self) -> None: + self._add_assignment( + assignment_id="owner-assignment", + identity_id="identity-owner", + account_id="account-owner", + ) + self.assertTrue(self._decision("account-owner").allowed) + + with self.database.SessionLocal() as session: + assignment = session.get( + IdmOrganizationFunctionAssignment, + "owner-assignment", + ) + assignment.is_active = False + session.commit() + self._add_assignment( + assignment_id="delegate-assignment", + identity_id="identity-delegate", + account_id="account-delegate", + ) + + self.assertFalse(self._decision("account-owner").allowed) + replacement = self._decision("account-delegate") + self.assertTrue(replacement.allowed) + self.assertEqual(replacement.assignment_ids, ("delegate-assignment",)) + + def test_real_directory_delegation_expires_with_its_source(self) -> None: + self._add_assignment( + assignment_id="owner-assignment", + identity_id="identity-owner", + account_id="account-owner", + ) + self._add_assignment( + assignment_id="delegated-assignment", + identity_id="identity-delegate", + account_id="account-delegate", + source="delegated", + delegated_from_assignment_id="owner-assignment", + valid_until=utc_now() + timedelta(hours=1), + ) + self.assertTrue(self._decision("account-delegate").allowed) + + with self.database.SessionLocal() as session: + delegated = session.get( + IdmOrganizationFunctionAssignment, + "delegated-assignment", + ) + delegated.valid_until = utc_now() - timedelta(seconds=1) + session.commit() + + expired = self._decision("account-delegate") + self.assertFalse(expired.allowed) + self.assertEqual(expired.reason_code, "effective_assignment_missing") + + def test_real_organization_state_and_function_move_fail_closed(self) -> None: + self._add_assignment( + assignment_id="owner-assignment", + identity_id="identity-owner", + account_id="account-owner", + ) + self.assertTrue(self._decision("account-owner").allowed) + + with self.database.SessionLocal() as session: + function = session.get(OrganizationFunction, "function-one") + function.is_active = False + session.commit() + inactive_function = self._decision("account-owner") + self.assertEqual( + inactive_function.reason_code, + "organization_function_inactive", + ) + + with self.database.SessionLocal() as session: + function = session.get(OrganizationFunction, "function-one") + function.is_active = True + unit = session.get(OrganizationUnit, "unit-one") + unit.is_active = False + session.commit() + inactive_unit = self._decision("account-owner") + self.assertEqual(inactive_unit.reason_code, "organization_unit_inactive") + + with self.database.SessionLocal() as session: + unit = session.get(OrganizationUnit, "unit-one") + unit.is_active = True + function = session.get(OrganizationFunction, "function-one") + function.organization_unit_id = "unit-two" + session.commit() + moved = self._decision("account-owner") + self.assertEqual(moved.reason_code, "organization_function_reassigned") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_service.py b/tests/test_service.py index e70b0c7..4b01bfe 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -735,6 +735,94 @@ class PostboxServiceTests(unittest.TestCase): self.assertTrue(allowed.allowed) self.assertEqual("effective_acting_for_assignment", allowed.reason_code) + def test_classification_clearance_controls_directory_and_delivery(self) -> None: + self.idm.assignments.append(self.assignment) + elevated_actor = PostboxActorRef( + account_id="account-1", + identity_id="identity-1", + authorized_actions=frozenset({"discover", "read", "send", "reply"}), + authorized_classifications=frozenset( + {"public", "internal", "confidential"} + ), + ) + with Session(self.engine) as session: + postbox = self.service.create_exact_postbox( + session, + tenant_id="tenant-1", + name="Confidential intake", + organization_unit_id="unit-1", + function_id="function-1", + address_key=None, + description=None, + classification="confidential", + actor_id="admin-1", + ) + + denied = self.service.explain_access( + session, + tenant_id="tenant-1", + postbox_id=postbox.id, + actor=self.actor, + action="read", + ) + visible = self.service.list_visible_postboxes( + session, + tenant_id="tenant-1", + actor=elevated_actor, + ) + delivered = self.service.deliver( + session, + PostboxDeliveryRequest( + tenant_id="tenant-1", + target=PostboxTargetRef(postbox_id=postbox.id), + producer_module="campaigns", + producer_resource_type="recipient", + producer_resource_id="recipient-1", + idempotency_key="confidential-message", + subject="Confidential notice", + classification="confidential", + ), + ) + + self.assertEqual( + denied.reason_code, + "classification_clearance_missing", + ) + self.assertEqual([postbox.id], [item.id for item in visible]) + self.assertIsNotNone( + self.service.get_message( + session, + tenant_id="tenant-1", + message_id=delivered.message_id, + actor=elevated_actor, + ) + ) + self.assertIsNone( + self.service.get_message( + session, + tenant_id="tenant-1", + message_id=delivered.message_id, + actor=self.actor, + ) + ) + with self.assertRaisesRegex( + PostboxError, + "exceeds the target Postbox classification", + ): + self.service.deliver( + session, + PostboxDeliveryRequest( + tenant_id="tenant-1", + target=PostboxTargetRef(postbox_id=postbox.id), + producer_module="campaigns", + producer_resource_type="recipient", + producer_resource_id="recipient-2", + idempotency_key="restricted-message", + subject="Restricted notice", + classification="restricted", + ), + ) + def test_template_revision_is_immutable_and_materialization_idempotent(self) -> None: with Session(self.engine) as session: template = self.service.create_template( diff --git a/webui/src/api/postbox.ts b/webui/src/api/postbox.ts index 426ade4..e3bf6f8 100644 --- a/webui/src/api/postbox.ts +++ b/webui/src/api/postbox.ts @@ -18,6 +18,9 @@ export type PostboxAccessDecision = { selected_assignment_id?: string | null; holder_count: number; vacant: boolean; + classification: string; + classification_allowed: boolean; + binding_status: string; }; export type PostboxDirectoryItem = {