GovOPlaN/govoplan-postbox
2
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Complete Postbox access transition matrix

Browse Source
This commit is contained in:
zemion
2026-07-31 18:21:36 +02:00
parent cbeaca979d
commit 0b6ebc3c74
9 changed files with 741 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/govoplan_postbox/backend/access_decisions.py
+95 -5
View File
@@ -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
),
src/govoplan_postbox/backend/manifest.py
+23 -2
View File
@@ -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,
),
),
)
src/govoplan_postbox/backend/router.py
+12
View File
@@ -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",
src/govoplan_postbox/backend/schemas.py
+16 -5
View File
@@ -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
src/govoplan_postbox/backend/service.py
+126 -12
View File
@@ -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,
*,
tests/test_access_decisions.py
+68
View File
@@ -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",
tests/test_real_directory_access.py
+310
View File
@@ -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()
tests/test_service.py
+88
View File
@@ -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(
webui/src/api/postbox.ts
+3
View File
@@ -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 = {