234 lines
12 KiB
Python
234 lines
12 KiB
Python
"""Batch transport reuse must never become an authorization or evidence cache."""
|
|
from __future__ import annotations
|
|
|
|
import imaplib
|
|
from dataclasses import asdict
|
|
from types import SimpleNamespace
|
|
from unittest.mock import Mock, patch
|
|
|
|
import pytest
|
|
|
|
from govoplan_core.security.credential_envelopes import CredentialEnvelope
|
|
from govoplan_mail.backend import capabilities, server_hierarchy
|
|
from govoplan_mail.backend.db.models import MailProfilePolicy, MailServerEndpoint
|
|
from govoplan_mail.backend.mail_profiles import MailProfileError
|
|
from govoplan_mail.backend.sending import imap as transport
|
|
from govoplan_mail.backend.sending.imap import ImapAppendError, ImapConfigurationError
|
|
|
|
from test_campaign_protocol_authorization import hierarchy, imap_args # noqa: F401
|
|
|
|
|
|
def client():
|
|
return SimpleNamespace(
|
|
utf8_enabled=False,
|
|
append=Mock(return_value=("OK", [b"provider internal secret response"])),
|
|
logout=Mock(return_value=("BYE", [])),
|
|
noop=Mock(return_value=("OK", [])),
|
|
)
|
|
|
|
|
|
def recovery():
|
|
return SimpleNamespace(replayed=False, succeed_imap=Mock(), reject=Mock(), unknown=Mock())
|
|
|
|
|
|
def test_reuses_transport_but_decrypts_and_authorizes_each_message_and_records_each_effect(hierarchy):
|
|
connection = client()
|
|
effects = [recovery(), recovery()]
|
|
with (
|
|
patch.object(transport, "_open_imap", return_value=connection) as open_connection,
|
|
patch.object(capabilities, "begin_provider_effect_recovery", side_effect=effects) as begin,
|
|
patch.object(capabilities, "_authorized_campaign_profile", wraps=capabilities._authorized_campaign_profile) as authorize,
|
|
patch.object(capabilities, "assert_mail_policy_allows_send", wraps=capabilities.assert_mail_policy_allows_send) as policy,
|
|
patch.object(server_hierarchy, "resolve_credential_envelope", wraps=server_hierarchy.resolve_credential_envelope) as decrypt,
|
|
capabilities.campaign_imap_batch(tenant_id="tenant-1", campaign_id="campaign-1") as batch,
|
|
):
|
|
assert batch.connection_count == 0
|
|
open_connection.assert_not_called()
|
|
results = [capabilities.append_campaign_message_to_sent(
|
|
hierarchy.session, **imap_args(message_bytes=f"message-{index}".encode(), recovery_effect_id=f"effect-{index}"),
|
|
) for index in range(2)]
|
|
assert batch.connection_count == 1
|
|
assert batch.reconnect_count == 0
|
|
assert authorize.call_count == policy.call_count == 2
|
|
assert [call.kwargs["credential_id"] for call in decrypt.call_args_list] == ["imap-credential"] * 2
|
|
assert [call.kwargs["effect_id"] for call in begin.call_args_list] == ["effect-0", "effect-1"]
|
|
assert [item.session_reused for item in results] == [False, True]
|
|
assert [item.connection_sequence for item in results] == [1, 1]
|
|
assert "secret" not in repr([asdict(item) for item in results])
|
|
assert "imap.example" not in repr(results)
|
|
open_connection.assert_called_once()
|
|
connection.logout.assert_called_once()
|
|
for effect in effects:
|
|
effect.succeed_imap.assert_called_once_with(folder="Sent")
|
|
effect.reject.assert_not_called()
|
|
effect.unknown.assert_not_called()
|
|
|
|
|
|
@pytest.mark.parametrize("change", ["credential_revoked", "smtp_revision_changed", "imap_revision_changed"])
|
|
def test_next_message_rechecks_current_authority_and_frozen_revisions_before_decryption(hierarchy, change):
|
|
connection = client()
|
|
with (
|
|
patch.object(transport, "_open_imap", return_value=connection),
|
|
patch.object(server_hierarchy, "resolve_credential_envelope", wraps=server_hierarchy.resolve_credential_envelope) as decrypt,
|
|
capabilities.campaign_imap_batch(tenant_id="tenant-1", campaign_id="campaign-1"),
|
|
):
|
|
capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args())
|
|
if change == "credential_revoked":
|
|
hierarchy.session.get(CredentialEnvelope, "imap-credential").is_active = False
|
|
else:
|
|
protocol = change.split("_")[0]
|
|
hierarchy.session.get(MailServerEndpoint, f"{protocol}-server").transport_revision = "changed"
|
|
hierarchy.session.commit()
|
|
with pytest.raises(MailProfileError):
|
|
capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args(message_bytes=b"blocked"))
|
|
assert decrypt.call_count == 1
|
|
connection.append.assert_called_once()
|
|
|
|
|
|
def test_effective_credential_policy_is_not_cached_by_warm_batch(hierarchy):
|
|
for scope in ("system", "tenant"):
|
|
hierarchy.session.get(MailProfilePolicy, f"{scope}-policy").policy = {
|
|
"smtp_credentials": {"inherit": False}, "imap_credentials": {"inherit": True},
|
|
}
|
|
hierarchy.session.commit()
|
|
connection = client()
|
|
with (
|
|
patch.object(transport, "_open_imap", return_value=connection),
|
|
patch.object(server_hierarchy, "resolve_credential_envelope", wraps=server_hierarchy.resolve_credential_envelope) as decrypt,
|
|
capabilities.campaign_imap_batch(tenant_id="tenant-1", campaign_id="campaign-1"),
|
|
):
|
|
capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args(imap_credential_id=None))
|
|
hierarchy.session.get(MailProfilePolicy, "tenant-policy").policy = {
|
|
"imap_credentials": {"inherit": False},
|
|
}
|
|
hierarchy.session.commit()
|
|
with pytest.raises(MailProfileError, match="explicit credential"):
|
|
capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args(imap_credential_id=None))
|
|
assert decrypt.call_count == 1
|
|
connection.append.assert_called_once()
|
|
|
|
|
|
def test_copying_context_to_a_worker_cannot_share_the_batch_connection(hierarchy):
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from contextvars import copy_context
|
|
with (
|
|
patch.object(capabilities, "_authorized_campaign_profile") as authorize,
|
|
capabilities.campaign_imap_batch(tenant_id="tenant-1", campaign_id="campaign-1"),
|
|
ThreadPoolExecutor(max_workers=1) as executor,
|
|
):
|
|
future = executor.submit(copy_context().run, capabilities.append_campaign_message_to_sent, hierarchy.session, **imap_args())
|
|
with pytest.raises(ImapConfigurationError, match="scope"):
|
|
future.result(timeout=5)
|
|
authorize.assert_not_called()
|
|
|
|
|
|
@pytest.mark.parametrize("scope", [{"tenant_id": "other"}, {"campaign_id": "other"}])
|
|
def test_cross_scope_batch_is_rejected_before_authorization_decryption_evidence_or_network(hierarchy, scope):
|
|
with (
|
|
patch.object(capabilities, "_authorized_campaign_profile") as authorize,
|
|
patch.object(capabilities, "begin_provider_effect_recovery") as begin,
|
|
patch.object(transport, "_open_imap") as open_connection,
|
|
capabilities.campaign_imap_batch(tenant_id="tenant-1", campaign_id="campaign-1"),
|
|
pytest.raises(ImapConfigurationError, match="scope"),
|
|
):
|
|
capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args(**scope))
|
|
authorize.assert_not_called()
|
|
begin.assert_not_called()
|
|
open_connection.assert_not_called()
|
|
|
|
|
|
def test_changed_resolved_secret_does_not_reuse_previous_authenticated_connection(hierarchy):
|
|
from govoplan_core.security.secrets import encrypt_secret
|
|
connections = [client(), client()]
|
|
with (
|
|
patch.object(transport, "_open_imap", side_effect=connections) as open_connection,
|
|
capabilities.campaign_imap_batch(tenant_id="tenant-1", campaign_id="campaign-1") as batch,
|
|
):
|
|
capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args())
|
|
hierarchy.session.get(CredentialEnvelope, "imap-credential").secret_data_encrypted = encrypt_secret('{"password":"new fake password"}')
|
|
hierarchy.session.commit()
|
|
result = capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args(message_bytes=b"new credential"))
|
|
assert batch.connection_count == result.connection_sequence == 2
|
|
assert not result.session_reused
|
|
assert open_connection.call_count == 2
|
|
assert open_connection.call_args_list[0].args[0].password != open_connection.call_args_list[1].args[0].password
|
|
for connection in connections:
|
|
connection.logout.assert_called_once()
|
|
connection.append.assert_called_once()
|
|
|
|
|
|
def test_nested_context_restores_outer_connection_and_always_cleans_up(hierarchy):
|
|
outer_client, inner_client = client(), client()
|
|
with patch.object(transport, "_open_imap", side_effect=[outer_client, inner_client]) as open_connection:
|
|
with capabilities.campaign_imap_batch(tenant_id="tenant-1", campaign_id="campaign-1") as outer:
|
|
capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args())
|
|
with pytest.raises(ValueError, match="caller"):
|
|
with capabilities.campaign_imap_batch(tenant_id="tenant-1", campaign_id="campaign-1"):
|
|
capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args())
|
|
raise ValueError("caller failure")
|
|
inner_client.logout.assert_called_once()
|
|
assert capabilities._ACTIVE_IMAP_BATCH.get() is outer
|
|
result = capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args())
|
|
assert result.session_reused
|
|
assert capabilities._ACTIVE_IMAP_BATCH.get() is None
|
|
assert open_connection.call_count == 2
|
|
assert outer_client.append.call_count == 2
|
|
outer_client.logout.assert_called_once()
|
|
|
|
|
|
def test_known_success_evidence_does_not_replay_even_with_warm_connection(hierarchy):
|
|
connection = client()
|
|
first, replayed = recovery(), recovery()
|
|
replayed.replayed = True
|
|
with (
|
|
patch.object(transport, "_open_imap", return_value=connection),
|
|
patch.object(capabilities, "begin_provider_effect_recovery", side_effect=[first, replayed]),
|
|
capabilities.campaign_imap_batch(tenant_id="tenant-1", campaign_id="campaign-1"),
|
|
):
|
|
capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args(recovery_effect_id="same-effect"))
|
|
with pytest.raises(ImapAppendError, match="already succeeded") as caught:
|
|
capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args(recovery_effect_id="same-effect"))
|
|
assert caught.value.outcome_unknown
|
|
connection.append.assert_called_once()
|
|
replayed.succeed_imap.assert_not_called()
|
|
|
|
|
|
def test_ambiguous_effect_is_recorded_once_sanitized_and_never_replayed(hierarchy):
|
|
connection = client()
|
|
connection.append.side_effect = imaplib.IMAP4.abort("provider host and secret")
|
|
effect = recovery()
|
|
with (
|
|
patch.object(transport, "_open_imap", return_value=connection) as open_connection,
|
|
patch.object(capabilities, "begin_provider_effect_recovery", return_value=effect),
|
|
capabilities.campaign_imap_batch(tenant_id="tenant-1", campaign_id="campaign-1"),
|
|
pytest.raises(ImapAppendError) as caught,
|
|
):
|
|
capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args(recovery_effect_id="effect-1"))
|
|
assert caught.value.outcome_unknown
|
|
assert "secret" not in str(caught.value)
|
|
assert "secret" not in repr(effect.unknown.call_args)
|
|
effect.unknown.assert_called_once()
|
|
effect.succeed_imap.assert_not_called()
|
|
effect.reject.assert_not_called()
|
|
open_connection.assert_called_once()
|
|
connection.append.assert_called_once()
|
|
connection.logout.assert_called_once()
|
|
|
|
|
|
def test_evidence_finalization_failure_closes_batch_and_prevents_further_effects(hierarchy):
|
|
connection = client()
|
|
effect = recovery()
|
|
effect.succeed_imap.side_effect = OSError("evidence store unavailable")
|
|
with (
|
|
patch.object(transport, "_open_imap", return_value=connection),
|
|
patch.object(capabilities, "begin_provider_effect_recovery", return_value=effect),
|
|
capabilities.campaign_imap_batch(tenant_id="tenant-1", campaign_id="campaign-1"),
|
|
):
|
|
with pytest.raises(ImapAppendError) as caught:
|
|
capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args(recovery_effect_id="effect-1"))
|
|
assert caught.value.outcome_unknown
|
|
with pytest.raises(ImapConfigurationError):
|
|
capabilities.append_campaign_message_to_sent(hierarchy.session, **imap_args(message_bytes=b"must not run"))
|
|
connection.append.assert_called_once()
|
|
connection.logout.assert_called_once()
|