from __future__ import annotations import base64 import dataclasses import importlib from datetime import UTC, datetime import poplib import ssl from types import SimpleNamespace import unittest from unittest.mock import patch from alembic.migration import MigrationContext from alembic.operations import Operations from fastapi import HTTPException from pydantic import ValidationError from sqlalchemy import MetaData, Table, create_engine, inspect, select from sqlalchemy.orm import sessionmaker from govoplan_access.backend.db.models import Account, User from govoplan_core.auth import ApiPrincipal from govoplan_core.core.access import PrincipalRef from govoplan_core.db.base import Base from govoplan_core.security.secrets import decrypt_secret from govoplan_mail.backend.config import Pop3Config, TransportSecurity from govoplan_mail.backend.db.models import ( MailPop3Import, MailServerEndpoint, MailServerProfile, ) from govoplan_mail.backend.pop3_imports import ( Pop3ImportConflict, bind_legacy_maildrop, maildrop_identity, pop3_preview_revision, Pop3ImportResult, create_pop3_imports, list_pop3_imports, ) from govoplan_mail.backend.router import ( import_profile_pop3_messages, preview_profile_pop3_import, reconcile_pop3_import_maildrop, ) from govoplan_mail.backend.schemas import ( MailPop3BindMaildropRequest, MailPop3ImportRequest, MailPop3PreviewRequest, ) from govoplan_mail.backend.sending.pop3 import ( Pop3ConfigurationError, Pop3DownloadedMessage, Pop3MessageSummary, Pop3ProviderError, _open_pop3, delete_pop3_messages, download_pop3_messages, preview_pop3_messages, ) _RAW = ( b"Subject: Legacy notice\r\n" b"From: Office \r\n" b"To: Subject \r\n" b"Message-ID: \r\n" b"\r\n" b"A bounded legacy message.\r\n" ) class _Pop3Client: def __init__(self, *, quit_error: Exception | None = None) -> None: self.deletions: list[int] = [] self.quit_calls = 0 self.rset_calls = 0 self.close_calls = 0 self.quit_error = quit_error def stat(self): return 1, len(_RAW) def uidl(self): return b"+OK", [b"1 uid-1"], 1 def list(self): return b"+OK", [f"1 {len(_RAW)}".encode("ascii")], 1 def top(self, _number, _lines): return b"+OK", _RAW.rstrip(b"\r\n").split(b"\r\n"), len(_RAW) def retr(self, _number): return b"+OK", _RAW.rstrip(b"\r\n").split(b"\r\n"), len(_RAW) def dele(self, number): self.deletions.append(number) def quit(self): self.quit_calls += 1 if self.quit_error is not None: raise self.quit_error return b"+OK" def rset(self): self.rset_calls += 1 def close(self): self.close_calls += 1 def _config(**changes) -> Pop3Config: values = { "host": "pop3.example.test", "security": "tls", "username": "legacy-user", "password": "legacy-password", "legacy_import_enabled": True, } values.update(changes) return Pop3Config.model_validate(values) def _download(uidl: str = "uid-1") -> Pop3DownloadedMessage: summary = Pop3MessageSummary( message_number=1, uidl=uidl, subject="Legacy notice", from_header="Office ", to_header="Subject ", date="Sat, 22 Aug 2026 10:00:00 +0200", message_id="", size_bytes=len(_RAW), body_preview="A bounded legacy message.", ) import hashlib return Pop3DownloadedMessage( message_number=1, uidl=uidl, raw=_RAW, raw_sha256=hashlib.sha256(_RAW).hexdigest(), summary=summary, ) class Pop3TransportTests(unittest.TestCase): def test_legacy_import_is_disabled_until_explicitly_enabled(self) -> None: with self.assertRaisesRegex(Pop3ConfigurationError, "disabled"): preview_pop3_messages( pop3_config=_config(legacy_import_enabled=False), limit=10, ) with self.assertRaisesRegex(ValueError, "batch size limit"): _config(max_message_bytes=2 * 1024 * 1024, max_batch_bytes=1024 * 1024) def test_preview_and_download_are_non_destructive(self) -> None: preview_client = _Pop3Client() with patch( "govoplan_mail.backend.sending.pop3._open_pop3", return_value=preview_client, ): preview = preview_pop3_messages(pop3_config=_config(), limit=10) self.assertEqual(["uid-1"], [item.uidl for item in preview.messages]) self.assertEqual([], preview_client.deletions) self.assertEqual(1, preview_client.quit_calls) download_client = _Pop3Client() with patch( "govoplan_mail.backend.sending.pop3._open_pop3", return_value=download_client, ): downloaded = download_pop3_messages(pop3_config=_config(), uidls=("uid-1",)) self.assertEqual(_RAW, downloaded[0].raw) self.assertEqual([], download_client.deletions) self.assertEqual(1, download_client.quit_calls) def test_source_deletion_needs_policy_and_commits_with_quit(self) -> None: with self.assertRaisesRegex(Pop3ConfigurationError, "disabled"): delete_pop3_messages(pop3_config=_config(), uidls=("uid-1",)) client = _Pop3Client() with patch( "govoplan_mail.backend.sending.pop3._open_pop3", return_value=client ): result = delete_pop3_messages( pop3_config=_config(allow_delete_after_import=True), uidls=("uid-1",), ) self.assertEqual(("uid-1",), result.deleted_uidls) self.assertEqual([1], client.deletions) self.assertEqual(1, client.quit_calls) def test_quit_failure_marks_deletion_outcome_unknown(self) -> None: client = _Pop3Client(quit_error=poplib.error_proto("connection lost")) with ( patch( "govoplan_mail.backend.sending.pop3._open_pop3", return_value=client, ), self.assertRaises(Pop3ProviderError) as captured, ): delete_pop3_messages( pop3_config=_config(allow_delete_after_import=True), uidls=("uid-1",), ) self.assertTrue(captured.exception.outcome_unknown) self.assertEqual([1], client.deletions) def test_tls_and_authentication_failures_are_sanitized(self) -> None: with ( patch("govoplan_mail.backend.sending.pop3.validate_outbound_host"), patch( "govoplan_mail.backend.sending.pop3._OutboundPolicyPOP3SSL", side_effect=ssl.SSLError("private TLS detail"), ), self.assertRaisesRegex(Pop3ProviderError, "TLS negotiation failed"), ): _open_pop3(_config()) auth_client = _Pop3Client() auth_client.user = lambda _value: None # type: ignore[attr-defined] auth_client.pass_ = lambda _value: (_ for _ in ()).throw( # type: ignore[attr-defined] poplib.error_proto("private auth detail") ) with ( patch("govoplan_mail.backend.sending.pop3.validate_outbound_host"), patch( "govoplan_mail.backend.sending.pop3._OutboundPolicyPOP3", return_value=auth_client, ), self.assertRaisesRegex(Pop3ProviderError, "authentication failed"), ): _open_pop3(_config(security=TransportSecurity.PLAIN)) class Pop3PersistenceTests(unittest.TestCase): def setUp(self) -> None: self.engine = create_engine("sqlite+pysqlite:///:memory:", future=True) Base.metadata.create_all( self.engine, tables=( Account.__table__, User.__table__, MailServerProfile.__table__, MailServerEndpoint.__table__, MailPop3Import.__table__, ), ) self.session = sessionmaker(bind=self.engine, expire_on_commit=False)() self.profile = MailServerProfile( id="profile-1", tenant_id="tenant-1", scope_type="tenant", scope_id="tenant-1", name="Legacy source", slug="legacy-source", smtp_config={}, ) self.server = MailServerEndpoint( id="server-1", profile_id=self.profile.id, tenant_id="tenant-1", protocol="pop3", name="Legacy POP3", config={"legacy_import_enabled": True}, scope_type="tenant", scope_id="tenant-1", transport_revision="revision-1", ) self.session.add_all((self.profile, self.server)) self.session.commit() def tearDown(self) -> None: self.session.close() self.engine.dispose() def test_import_is_encrypted_and_duplicate_uidl_is_reused(self) -> None: first = create_pop3_imports( self.session, tenant_id="tenant-1", profile_id=self.profile.id, pop3_server_id=self.server.id, pop3_credential_id=None, transport_revision="revision-1", maildrop_key=maildrop_identity(_config()), messages=(_download(),), user_id=None, deletion_requested=False, ) self.session.commit() self.assertEqual(1, len(first.imported)) encrypted = first.imported[0].raw_message_encrypted self.assertNotIn("Legacy notice", encrypted) self.assertEqual( _RAW, base64.b64decode(decrypt_secret(encrypted) or ""), ) self.assertEqual("not_requested", first.imported[0].deletion_status) repeated = create_pop3_imports( self.session, tenant_id="tenant-1", profile_id=self.profile.id, pop3_server_id=self.server.id, pop3_credential_id=None, transport_revision="revision-1", maildrop_key=maildrop_identity(_config()), messages=(_download(),), user_id=None, deletion_requested=False, ) self.assertEqual((), repeated.imported) self.assertEqual( (first.imported[0].id,), tuple(row.id for row in repeated.duplicates) ) self.assertEqual( (first.imported[0].id,), tuple( row.id for row in list_pop3_imports( self.session, tenant_id="tenant-1", profile_ids=(self.profile.id,), ) ), ) self.assertEqual( (), list_pop3_imports( self.session, tenant_id="tenant-1", profile_ids=("unrelated-profile",), ), ) def _import( self, *, config=None, messages=None, credential="credential-1", tenant="tenant-1", ): return create_pop3_imports( self.session, tenant_id=tenant, profile_id=self.profile.id, pop3_server_id=self.server.id, pop3_credential_id=credential, transport_revision="revision-1", maildrop_key=maildrop_identity(config or _config()), messages=messages or (_download(),), user_id=None, deletion_requested=False, ) def test_mailbox_identity_survives_secret_rotation_but_not_username_or_host(self): original = maildrop_identity(_config()) self.assertEqual(original, maildrop_identity(_config(password="rotated"))) self.assertEqual(original, maildrop_identity(_config(host="POP3.EXAMPLE.TEST"))) for change in ( {"username": "Legacy-user"}, {"host": "other.example.test"}, {"port": 1995}, ): self.assertNotEqual(original, maildrop_identity(_config(**change))) first = self._import().imported[0] self.session.commit() self.assertEqual( first.id, self._import(config=_config(password="rotated"), credential="credential-2") .duplicates[0] .id, ) second = self._import(config=_config(username="mailbox-B")).imported[0] self.assertNotEqual(first.id, second.id) self.assertNotEqual(first.fingerprint, second.fingerprint) def test_tenant_boundary_remains_in_identity_lookup_and_constraint(self): self.server.tenant_id = None self.profile.tenant_id = None first = self._import().imported[0] self.session.commit() second = self._import(tenant="tenant-2").imported[0] self.assertNotEqual(first.id, second.id) def test_changed_uidl_content_rejects_the_whole_batch_before_insert(self): self._import() self.session.commit() import hashlib message = dataclasses.replace( _download(), raw=b"different", raw_sha256=hashlib.sha256(b"different").hexdigest(), ) with self.assertRaisesRegex(Pop3ImportConflict, "different content"): self._import(messages=(_download("new-uidl"), message)) self.assertEqual(1, self.session.query(MailPop3Import).count()) self.assertFalse(self.session.new) def test_legacy_conflict_is_explicit_and_binding_is_non_destructive_and_repeatable( self, ): row = self._import().imported[0] row.maildrop_identity = None self.session.commit() retained = ( row.id, row.raw_message_encrypted, row.fingerprint, row.transport_revision, ) with self.assertRaisesRegex(Pop3ImportConflict, "unbound legacy"): self._import() for _ in range(2): bound = bind_legacy_maildrop( self.session, tenant_id="tenant-1", profile_id=self.profile.id, server_id=self.server.id, import_id=row.id, identity=maildrop_identity(_config()), message=_download(), ) self.session.commit() self.assertEqual( retained, ( bound.id, bound.raw_message_encrypted, bound.fingerprint, bound.transport_revision, ), ) self.assertEqual(row.id, self._import().duplicates[0].id) def test_legacy_binding_rejects_different_content_and_collision(self): row = self._import().imported[0] self._import(config=_config(username="mailbox-B")) row.maildrop_identity = None self.session.commit() with self.assertRaisesRegex(Pop3ImportConflict, "does not match"): bind_legacy_maildrop( self.session, tenant_id="tenant-1", profile_id=self.profile.id, server_id=self.server.id, import_id=row.id, identity=maildrop_identity(_config()), message=dataclasses.replace(_download(), raw=b"different"), ) with self.assertRaisesRegex(Pop3ImportConflict, "already owns"): bind_legacy_maildrop( self.session, tenant_id="tenant-1", profile_id=self.profile.id, server_id=self.server.id, import_id=row.id, identity=maildrop_identity(_config(username="mailbox-B")), message=_download(), ) self.assertIsNone(row.maildrop_identity) def _principal(self, *, tenant="tenant-1", omitted=None): scopes = {"mail:profile:use", "mail:pop3:manage", "mail:pop3:import"} - { omitted } return ApiPrincipal( principal=PrincipalRef( account_id="account", membership_id="user", tenant_id=tenant, scopes=frozenset(scopes), ), account=SimpleNamespace(id="account"), user=SimpleNamespace(id="user"), ) def test_legacy_binding_rejects_unverifiable_retained_ciphertext(self): row = self._import().imported[0] row.maildrop_identity = None row.raw_message_encrypted = "invalid retained ciphertext" self.session.commit() with self.assertRaisesRegex(Pop3ImportConflict, "could not be verified"): bind_legacy_maildrop( self.session, tenant_id="tenant-1", profile_id=self.profile.id, server_id=self.server.id, import_id=row.id, identity=maildrop_identity(_config()), message=_download(), ) self.assertIsNone(row.maildrop_identity) self.assertEqual("invalid retained ciphertext", row.raw_message_encrypted) def _resolved(self, config=None): return ( self.profile, SimpleNamespace( config=config or _config(), server=self.server, credential=None, transport_revision="revision-1", ), ) def _binding_request(self): return MailPop3BindMaildropRequest( server_id=self.server.id, expected_transport_revision=pop3_preview_revision( "revision-1", maildrop_identity(_config()) ), confirm_binding=True, ) def test_preview_distinguishes_mailboxes_and_reports_unbound_legacy(self): row = self._import().imported[0] self.session.commit() preview = SimpleNamespace( messages=(_download().summary,), host="pop3.example.test", port=995, security="tls", message_count=1, mailbox_size_bytes=len(_RAW), ) for config, imported, conflict in ( (_config(), True, False), (_config(username="other"), False, False), (_config(), False, True), ): if conflict: row.maildrop_identity = None self.session.commit() with ( patch( "govoplan_mail.backend.router._resolve_profile_pop3_transport", return_value=self._resolved(config), ), patch( "govoplan_mail.backend.router.preview_pop3_messages", return_value=preview, ), ): result = preview_profile_pop3_import( self.profile.id, MailPop3PreviewRequest(server_id=self.server.id), principal=self._principal(), session=self.session, ) self.assertEqual(imported, result.messages[0].already_imported) self.assertEqual(conflict, result.messages[0].legacy_identity_conflict) self.assertEqual( pop3_preview_revision("revision-1", maildrop_identity(config)), result.transport_revision, ) def test_reconciliation_route_checks_each_scope_before_resolving_or_downloading( self, ): for scope in ("mail:profile:use", "mail:pop3:manage", "mail:pop3:import"): with ( patch( "govoplan_mail.backend.router._resolve_profile_pop3_transport" ) as resolve, self.assertRaises(HTTPException) as caught, ): reconcile_pop3_import_maildrop( self.profile.id, "unused", self._binding_request(), principal=self._principal(omitted=scope), session=self.session, ) self.assertEqual(403, caught.exception.status_code) resolve.assert_not_called() with self.assertRaises(ValidationError): MailPop3BindMaildropRequest( server_id=self.server.id, expected_transport_revision="revision", confirm_binding=False, ) def test_reconciliation_route_checks_tenant_and_preview_before_download(self): row = self._import().imported[0] row.maildrop_identity = None self.session.commit() with ( patch( "govoplan_mail.backend.router._resolve_profile_pop3_transport", return_value=self._resolved(), ), patch("govoplan_mail.backend.router.download_pop3_messages") as download, ): with self.assertRaises(HTTPException) as caught: reconcile_pop3_import_maildrop( self.profile.id, row.id, self._binding_request(), principal=self._principal(tenant="another"), session=self.session, ) self.assertEqual(404, caught.exception.status_code) stale = self._binding_request().model_copy( update={"expected_transport_revision": "old-preview"} ) with self.assertRaises(HTTPException) as caught: reconcile_pop3_import_maildrop( self.profile.id, row.id, stale, principal=self._principal(), session=self.session, ) self.assertEqual(409, caught.exception.status_code) download.assert_not_called() def test_reconciliation_route_audit_failure_rolls_back_and_success_never_deletes( self, ): row = self._import().imported[0] row.maildrop_identity = None self.session.commit() with ( patch( "govoplan_mail.backend.router._resolve_profile_pop3_transport", return_value=self._resolved(), ), patch( "govoplan_mail.backend.router.download_pop3_messages", return_value=(_download(),), ), patch("govoplan_mail.backend.router.delete_pop3_messages") as delete, patch( "govoplan_mail.backend.router.audit_event", side_effect=RuntimeError("audit unavailable"), ), ): with self.assertRaisesRegex(RuntimeError, "audit unavailable"): reconcile_pop3_import_maildrop( self.profile.id, row.id, self._binding_request(), principal=self._principal(), session=self.session, ) self.assertIsNone( self.session.get(MailPop3Import, row.id).maildrop_identity ) delete.assert_not_called() with ( patch( "govoplan_mail.backend.router._resolve_profile_pop3_transport", return_value=self._resolved(), ), patch( "govoplan_mail.backend.router.download_pop3_messages", return_value=(_download(),), ), patch("govoplan_mail.backend.router.delete_pop3_messages") as delete, patch("govoplan_mail.backend.router.audit_event") as audit, ): result = reconcile_pop3_import_maildrop( self.profile.id, row.id, self._binding_request(), principal=self._principal(), session=self.session, ) self.assertTrue(result.maildrop_identity_bound) self.assertEqual(row.id, result.id) self.assertEqual( "mail.pop3.maildrop_bound", audit.call_args.kwargs["action"] ) delete.assert_not_called() def test_migration_retains_unbound_records_and_refuses_lossy_downgrade(self): self.session.close() MailPop3Import.__table__.drop(self.engine) old = importlib.import_module( "govoplan_mail.backend.migrations.versions.a4c5d6e7f809_mail_pop3_imports" ) new = importlib.import_module( "govoplan_mail.backend.migrations.versions.b5d6e7f8091a_mail_pop3_maildrop_identity" ) with self.engine.begin() as connection: with Operations.context(MigrationContext.configure(connection)): old.upgrade() table = Table("mail_pop3_imports", MetaData(), autoload_with=connection) now = datetime.now(UTC) values = dict( id="legacy", tenant_id="tenant-1", profile_id=self.profile.id, pop3_server_id=self.server.id, transport_revision="historical", provider_uidl="uidl", fingerprint="original", raw_sha256="retained", raw_message_encrypted="retained ciphertext", size_bytes=7, status="pending_review", imported_at=now, deletion_requested=False, deletion_status="not_requested", created_at=now, updated_at=now, ) connection.execute(table.insert().values(**values)) with Operations.context(MigrationContext.configure(connection)): new.upgrade() current = Table("mail_pop3_imports", MetaData(), autoload_with=connection) retained = connection.execute(select(current)).mappings().one() self.assertIsNone(retained["maildrop_identity"]) for key in ( "id", "fingerprint", "raw_message_encrypted", "raw_sha256", "transport_revision", ): self.assertEqual(values[key], retained[key]) connection.execute( current.insert().values( **{**values, "id": "new-maildrop", "maildrop_identity": "a" * 64} ) ) with ( Operations.context(MigrationContext.configure(connection)), self.assertRaisesRegex(RuntimeError, "collapse different maildrops"), ): new.downgrade() self.assertIn( "maildrop_identity", { column["name"] for column in inspect(connection).get_columns("mail_pop3_imports") }, ) self.assertEqual(2, len(connection.execute(select(current.c.id)).all())) class _RouteSession: def __init__(self, events: list[str]) -> None: self.events = events def commit(self) -> None: self.events.append("commit") def rollback(self) -> None: self.events.append("rollback") class Pop3ImportRouteTests(unittest.TestCase): def test_local_import_and_audit_commit_before_source_deletion(self) -> None: events: list[str] = [] now = datetime.now(UTC) row = SimpleNamespace( id="import-1", profile_id="profile-1", pop3_server_id="server-1", transport_revision="revision-1", provider_uidl="uid-1", message_id="", subject="Legacy notice", from_header="office@example.test", to_header="subject@example.test", date="2026-08-22", body_preview="A bounded legacy message.", size_bytes=len(_RAW), raw_sha256=_download().raw_sha256, status="pending_review", imported_at=now, deletion_requested=True, deletion_status="pending", deletion_attempted_at=None, deletion_error=None, ) resolved = SimpleNamespace( config=_config(allow_delete_after_import=True), server=SimpleNamespace(id="server-1"), credential=None, transport_revision="revision-1", ) principal = ApiPrincipal( principal=PrincipalRef( account_id="account-1", membership_id="user-1", tenant_id="tenant-1", scopes=frozenset( { "mail:profile:use", "mail:pop3:import", "mail:pop3:delete", } ), ), account=SimpleNamespace(id="account-1"), user=SimpleNamespace(id="user-1"), ) payload = MailPop3ImportRequest( server_id="server-1", expected_transport_revision=pop3_preview_revision( "revision-1", maildrop_identity(_config()) ), uidls=["uid-1"], delete_after_import=True, ) def audit(*_args, **_kwargs) -> None: events.append("audit") def delete(**_kwargs): self.assertEqual(["audit", "commit"], events) events.append("delete") def mark(*_args, **_kwargs): events.append("mark") row.deletion_status = "succeeded" row.deletion_attempted_at = now return (row,) with ( patch( "govoplan_mail.backend.router._resolve_profile_pop3_transport", return_value=(SimpleNamespace(id="profile-1"), resolved), ), patch( "govoplan_mail.backend.router.download_pop3_messages", return_value=(_download(),), ), patch( "govoplan_mail.backend.router.create_pop3_imports", return_value=Pop3ImportResult(imported=(row,), duplicates=()), ), patch("govoplan_mail.backend.router.audit_event", side_effect=audit), patch( "govoplan_mail.backend.router.delete_pop3_messages", side_effect=delete ), patch( "govoplan_mail.backend.router.mark_pop3_deletion_result", side_effect=mark, ), ): result = import_profile_pop3_messages( "profile-1", payload, principal=principal, session=_RouteSession(events), # type: ignore[arg-type] ) self.assertEqual("succeeded", result.deletion_status) self.assertEqual( ["audit", "commit", "delete", "mark", "commit", "audit", "commit"], events, ) if __name__ == "__main__": unittest.main()