145 lines
5.0 KiB
Python
145 lines
5.0 KiB
Python
"""Gate 1 contract tests for authoritative TenantRouter isolation."""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
from tempfile import TemporaryDirectory
|
|
import unittest
|
|
|
|
from aios_v2r1.router import TenantRouter
|
|
from aios_v2r1.tenant_context import CanonicalMessage, ConflictError, ScopeError
|
|
|
|
|
|
class MutableClock:
|
|
def __init__(self, now: datetime) -> None:
|
|
self.now = now
|
|
|
|
def __call__(self) -> datetime:
|
|
return self.now
|
|
|
|
|
|
def canonical_message(*, tenant_id: str = "tenant-a", cell_id: str = "cell-a") -> CanonicalMessage:
|
|
return CanonicalMessage(
|
|
tenant_id=tenant_id,
|
|
cell_id=cell_id,
|
|
principal_id="user-a",
|
|
principal_type="authenticated_user",
|
|
origin_type="channel_user",
|
|
channel="weixin",
|
|
channel_account_id="account-a",
|
|
peer_id="peer-a",
|
|
conversation_id="conversation-a",
|
|
session_id="session-a",
|
|
task_id=None,
|
|
event_id="event-a",
|
|
message_id="message-a",
|
|
reply_to_message_id=None,
|
|
direction="inbound",
|
|
command_capable=True,
|
|
occurred_at="2026-07-13T00:00:00+00:00",
|
|
received_at="2026-07-13T00:00:01+00:00",
|
|
body="hello",
|
|
quoted_body=None,
|
|
adapter_name="weixin",
|
|
adapter_version="v2",
|
|
)
|
|
|
|
|
|
class TenantRouterIsolationContractTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.directory = TemporaryDirectory()
|
|
self.addCleanup(self.directory.cleanup)
|
|
self.database = f"{self.directory.name}/router.sqlite3"
|
|
self.router = TenantRouter(self.database)
|
|
self.router.register_tenant("tenant-a", "cell-a")
|
|
self.router.register_tenant("tenant-b", "cell-b")
|
|
self.router.bind_channel(
|
|
tenant_id="tenant-a", channel="weixin", channel_account_id="account-a"
|
|
)
|
|
self.router.bind_channel(
|
|
tenant_id="tenant-b", channel="weixin", channel_account_id="account-b"
|
|
)
|
|
|
|
def test_channel_account_cannot_bind_two_tenants(self) -> None:
|
|
with self.assertRaises(ConflictError):
|
|
self.router.bind_channel(
|
|
tenant_id="tenant-b", channel="weixin", channel_account_id="account-a"
|
|
)
|
|
|
|
def test_cell_cannot_bind_two_tenants(self) -> None:
|
|
with self.assertRaises(ConflictError):
|
|
self.router.register_tenant("tenant-c", "cell-a")
|
|
|
|
def test_principal_cannot_bind_across_tenant_boundary(self) -> None:
|
|
with self.assertRaises(ScopeError):
|
|
self.router.bind_principal(
|
|
tenant_id="tenant-b",
|
|
channel="weixin",
|
|
channel_account_id="account-a",
|
|
principal_id="user-a",
|
|
)
|
|
|
|
def test_session_cannot_cross_cell(self) -> None:
|
|
with self.assertRaises(ScopeError):
|
|
self.router.bind_session(
|
|
tenant_id="tenant-a",
|
|
session_id="session-a",
|
|
cell_id="cell-b",
|
|
principal_id="user-a",
|
|
)
|
|
|
|
def test_session_cannot_cross_principal(self) -> None:
|
|
self.router.bind_session(
|
|
tenant_id="tenant-a",
|
|
session_id="session-a",
|
|
cell_id="cell-a",
|
|
principal_id="user-a",
|
|
)
|
|
|
|
with self.assertRaises(ConflictError):
|
|
self.router.bind_session(
|
|
tenant_id="tenant-a",
|
|
session_id="session-a",
|
|
cell_id="cell-a",
|
|
principal_id="user-b",
|
|
)
|
|
|
|
def test_expired_session_is_rejected_without_private_storage_access(self) -> None:
|
|
clock = MutableClock(datetime(2026, 7, 13, tzinfo=timezone.utc))
|
|
router = TenantRouter(f"{self.directory.name}/expiry.sqlite3", clock=clock)
|
|
router.register_tenant("tenant-a", "cell-a")
|
|
router.bind_channel(
|
|
tenant_id="tenant-a", channel="weixin", channel_account_id="account-a"
|
|
)
|
|
router.bind_session(
|
|
tenant_id="tenant-a",
|
|
session_id="session-a",
|
|
cell_id="cell-a",
|
|
principal_id="user-a",
|
|
)
|
|
clock.now += timedelta(days=2)
|
|
|
|
with self.assertRaises(ScopeError):
|
|
router.resolve(
|
|
channel="weixin",
|
|
channel_account_id="account-a",
|
|
principal_id="user-a",
|
|
session_id="session-a",
|
|
)
|
|
|
|
def test_self_reported_tenant_or_cell_mismatch_fails_against_authority(self) -> None:
|
|
self.router.bind_session(
|
|
tenant_id="tenant-a",
|
|
session_id="session-a",
|
|
cell_id="cell-a",
|
|
principal_id="user-a",
|
|
)
|
|
expected = "message claims do not match authoritative route"
|
|
with self.assertRaisesRegex(ScopeError, expected):
|
|
self.router.assert_message_route(canonical_message(tenant_id="tenant-b"))
|
|
with self.assertRaisesRegex(ScopeError, expected):
|
|
self.router.assert_message_route(canonical_message(cell_id="cell-b"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|