137 lines
4.1 KiB
Python
137 lines
4.1 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.auth import ApiPrincipal
|
|
from govoplan_core.core.access import PrincipalRef
|
|
from govoplan_core.core.search import (
|
|
SearchAuthorizationRequest,
|
|
SearchBackfillRequest,
|
|
SearchResourceReference,
|
|
)
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_postbox.backend.db.models import (
|
|
Postbox,
|
|
PostboxAddress,
|
|
PostboxMessage,
|
|
)
|
|
from govoplan_postbox.backend.search_source import (
|
|
PROVIDER_ID,
|
|
RESOURCE_TYPE,
|
|
PostboxSearchSource,
|
|
)
|
|
|
|
|
|
class _AccessService:
|
|
def __init__(self, allowed: bool = True) -> None:
|
|
self.allowed = allowed
|
|
|
|
def can_read_message(self, session, **kwargs):
|
|
del session, kwargs
|
|
return self.allowed
|
|
|
|
|
|
class PostboxSearchSourceTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite://")
|
|
Base.metadata.create_all(
|
|
self.engine,
|
|
tables=(
|
|
PostboxAddress.__table__,
|
|
Postbox.__table__,
|
|
PostboxMessage.__table__,
|
|
),
|
|
)
|
|
self.session = Session(self.engine)
|
|
self.session.add_all(
|
|
(
|
|
PostboxAddress(
|
|
id="address-1",
|
|
tenant_id="tenant-1",
|
|
address_key="permits",
|
|
address="permits@example.test",
|
|
),
|
|
Postbox(
|
|
id="postbox-1",
|
|
tenant_id="tenant-1",
|
|
address_id="address-1",
|
|
name="Permit office",
|
|
),
|
|
PostboxMessage(
|
|
id="message-1",
|
|
tenant_id="tenant-1",
|
|
postbox_id="postbox-1",
|
|
subject="Permit decision",
|
|
body_text=None,
|
|
body_ciphertext=b"ciphertext",
|
|
encryption_profile="server_envelope_v1",
|
|
delivered_at=datetime(2026, 8, 5, tzinfo=timezone.utc),
|
|
),
|
|
)
|
|
)
|
|
self.session.commit()
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
self.engine.dispose()
|
|
|
|
def test_encrypted_content_is_not_indexed_and_access_is_rechecked(self) -> None:
|
|
source = PostboxSearchSource(_AccessService()) # type: ignore[arg-type]
|
|
page = source.backfill(
|
|
self.session,
|
|
request=SearchBackfillRequest(
|
|
tenant_id="tenant-1",
|
|
provider_id=PROVIDER_ID,
|
|
resource_type=RESOURCE_TYPE,
|
|
rebuild_id="rebuild-1",
|
|
),
|
|
)
|
|
self.assertEqual(1, len(page.documents))
|
|
self.assertIsNone(page.documents[0].body)
|
|
self.assertNotIn("ciphertext", str(page.documents[0].metadata).casefold())
|
|
reference = SearchResourceReference(
|
|
tenant_id="tenant-1",
|
|
module_id="postbox",
|
|
resource_type=RESOURCE_TYPE,
|
|
resource_id="message-1",
|
|
)
|
|
request = SearchAuthorizationRequest(reference=reference, source_revision="1")
|
|
self.assertTrue(
|
|
source.authorize(
|
|
self.session,
|
|
_principal(),
|
|
requests=(request,),
|
|
)[reference.key]
|
|
)
|
|
denied = PostboxSearchSource(_AccessService(False)) # type: ignore[arg-type]
|
|
self.assertFalse(
|
|
denied.authorize(
|
|
self.session,
|
|
_principal(),
|
|
requests=(request,),
|
|
)[reference.key]
|
|
)
|
|
|
|
|
|
def _principal() -> ApiPrincipal:
|
|
return ApiPrincipal(
|
|
principal=PrincipalRef(
|
|
account_id="account-1",
|
|
membership_id="user-1",
|
|
identity_id="identity-1",
|
|
tenant_id="tenant-1",
|
|
scopes=frozenset({"postbox:postbox:read"}),
|
|
),
|
|
account=SimpleNamespace(id="account-1"),
|
|
user=SimpleNamespace(id="user-1"),
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|