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
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(