4.53.1

app/tests/dashboard/test_alias_transfer.py

@@ -1,38 +1,72 @@
 import app.alias_utils
+from app import config
 from app.db import Session
+from app.events.event_dispatcher import GlobalDispatcher
 from app.models import (
     Alias,
     Mailbox,
-    User,
     AliasMailbox,
 )
+from tests.events.event_test_utils import (
+    OnMemoryDispatcher,
+    _get_event_from_string,
+    _create_linked_user,
+)
 from tests.utils import login
 
+on_memory_dispatcher = OnMemoryDispatcher()
+
+
+def setup_module():
+    GlobalDispatcher.set_dispatcher(on_memory_dispatcher)
+    config.EVENT_WEBHOOK = "http://test"
+
+
+def teardown_module():
+    GlobalDispatcher.set_dispatcher(None)
+    config.EVENT_WEBHOOK = None
+
 
 def test_alias_transfer(flask_client):
-    user = login(flask_client)
-    mb = Mailbox.create(user_id=user.id, email="mb@gmail.com", commit=True)
+    (source_user, source_user_pu) = _create_linked_user()
+    source_user = login(flask_client, source_user)
+    mb = Mailbox.create(user_id=source_user.id, email="mb@gmail.com", commit=True)
 
-    alias = Alias.create_new_random(user)
+    alias = Alias.create_new_random(source_user)
     Session.commit()
 
     AliasMailbox.create(alias_id=alias.id, mailbox_id=mb.id, commit=True)
 
-    new_user = User.create(
-        email="hey@example.com",
-        password="password",
-        activated=True,
-        commit=True,
-    )
+    (target_user, target_user_pu) = _create_linked_user()
 
     Mailbox.create(
-        user_id=new_user.id, email="hey2@example.com", verified=True, commit=True
+        user_id=target_user.id, email="hey2@example.com", verified=True, commit=True
     )
 
-    app.alias_utils.transfer_alias(alias, new_user, new_user.mailboxes())
+    on_memory_dispatcher.clear()
+    app.alias_utils.transfer_alias(alias, target_user, target_user.mailboxes())
 
     # refresh from db
     alias = Alias.get(alias.id)
-    assert alias.user == new_user
-    assert set(alias.mailboxes) == set(new_user.mailboxes())
+    assert alias.user == target_user
+    assert set(alias.mailboxes) == set(target_user.mailboxes())
     assert len(alias.mailboxes) == 2
+
+    # Check events
+    assert len(on_memory_dispatcher.memory) == 2
+    # 1st delete event
+    event_data = on_memory_dispatcher.memory[0]
+    event_content = _get_event_from_string(event_data, source_user, source_user_pu)
+    assert event_content.alias_deleted is not None
+    alias_deleted = event_content.alias_deleted
+    assert alias_deleted.id == alias.id
+    assert alias_deleted.email == alias.email
+    # 2nd create event
+    event_data = on_memory_dispatcher.memory[1]
+    event_content = _get_event_from_string(event_data, target_user, target_user_pu)
+    assert event_content.alias_created is not None
+    alias_created = event_content.alias_created
+    assert alias.id == alias_created.id
+    assert alias.email == alias_created.email
+    assert alias.note or "" == alias_created.note
+    assert alias.enabled == alias_created.enabled

app/tests/events/event_test_utils.py

@@ -1,4 +1,5 @@
 from app.events.event_dispatcher import Dispatcher
+from app.events.generated import event_pb2
 from app.models import PartnerUser, User
 from app.proton.utils import get_proton_partner
 from tests.utils import create_new_user, random_token
@@ -30,3 +31,14 @@ def _create_linked_user() -> Tuple[User, PartnerUser]:
     )
 
     return user, partner_user
+
+
+def _get_event_from_string(
+    data: str, user: User, pu: PartnerUser
+) -> event_pb2.EventContent:
+    event = event_pb2.Event()
+    event.ParseFromString(data)
+    assert user.id == event.user_id
+    assert pu.external_user_id == event.external_user_id
+    assert pu.partner_id == event.partner_id
+    return event.content

app/tests/events/test_sent_events.py

@@ -1,12 +1,12 @@
 from app import config, alias_utils
 from app.db import Session
 from app.events.event_dispatcher import GlobalDispatcher
-from app.events.generated import event_pb2
-from app.models import Alias, User, PartnerUser
+from app.models import Alias
 from tests.utils import random_token
 from .event_test_utils import (
     OnMemoryDispatcher,
     _create_linked_user,
+    _get_event_from_string,
 )
 
 on_memory_dispatcher = OnMemoryDispatcher()
@@ -26,17 +26,6 @@ def setup_function(func):
     on_memory_dispatcher.clear()
 
 
-def _get_event_from_string(
-    data: str, user: User, pu: PartnerUser
-) -> event_pb2.EventContent:
-    event = event_pb2.Event()
-    event.ParseFromString(data)
-    assert user.id == event.user_id
-    assert pu.external_user_id == event.external_user_id
-    assert pu.partner_id == event.partner_id
-    return event.content
-
-
 def test_fire_event_on_alias_creation():
     (user, pu) = _create_linked_user()
     alias = Alias.create_new_random(user)

app/tests/test_alias_utils.py

@@ -4,6 +4,7 @@ from app.alias_utils import (
     delete_alias,
     check_alias_prefix,
     get_user_if_alias_would_auto_create,
+    get_alias_recipient_name,
     try_auto_create,
 )
 from app.config import ALIAS_DOMAINS
@@ -18,7 +19,8 @@ from app.models import (
     User,
     DomainDeletedAlias,
 )
-from tests.utils import create_new_user, random_domain, random_token
+from app.utils import random_string
+from tests.utils import create_new_user, random_domain, random_token, random_email
 
 
 def test_delete_alias(flask_client):
@@ -131,3 +133,91 @@ def test_auto_create_alias(flask_client):
             assert result, f"Case {test_id} - Failed address {address}"
         else:
             assert result is None, f"Case {test_id} - Failed address {address}"
+
+
+# get_alias_recipient_name
+def test_get_alias_recipient_name_no_overrides():
+    user = create_new_user()
+    alias = Alias.create(
+        user_id=user.id,
+        email=random_email(),
+        mailbox_id=user.default_mailbox_id,
+        commit=True,
+    )
+    res = get_alias_recipient_name(alias)
+    assert res.message is None
+    assert res.name == alias.email
+
+
+def test_get_alias_recipient_name_alias_name():
+    user = create_new_user()
+    alias = Alias.create(
+        user_id=user.id,
+        email=random_email(),
+        mailbox_id=user.default_mailbox_id,
+        name=random_string(),
+        commit=True,
+    )
+    res = get_alias_recipient_name(alias)
+    assert res.message is not None
+    assert res.name == f"{alias.name} <{alias.email}>"
+
+
+def test_get_alias_recipient_alias_with_name_and_custom_domain_name():
+    user = create_new_user()
+    custom_domain = CustomDomain.create(
+        user_id=user.id,
+        domain=random_domain(),
+        name=random_string(),
+        verified=True,
+    )
+    alias = Alias.create(
+        user_id=user.id,
+        email=random_email(),
+        mailbox_id=user.default_mailbox_id,
+        name=random_string(),
+        custom_domain_id=custom_domain.id,
+        commit=True,
+    )
+    res = get_alias_recipient_name(alias)
+    assert res.message is not None
+    assert res.name == f"{alias.name} <{alias.email}>"
+
+
+def test_get_alias_recipient_alias_without_name_and_custom_domain_without_name():
+    user = create_new_user()
+    custom_domain = CustomDomain.create(
+        user_id=user.id,
+        domain=random_domain(),
+        verified=True,
+    )
+    alias = Alias.create(
+        user_id=user.id,
+        email=random_email(),
+        mailbox_id=user.default_mailbox_id,
+        custom_domain_id=custom_domain.id,
+        commit=True,
+    )
+    res = get_alias_recipient_name(alias)
+    assert res.message is None
+    assert res.name == alias.email
+
+
+def test_get_alias_recipient_alias_without_name_and_custom_domain_name():
+    user = create_new_user()
+    custom_domain = CustomDomain.create(
+        user_id=user.id,
+        domain=random_domain(),
+        name=random_string(),
+        verified=True,
+    )
+    alias = Alias.create(
+        user_id=user.id,
+        email=random_email(),
+        mailbox_id=user.default_mailbox_id,
+        custom_domain_id=custom_domain.id,
+        commit=True,
+    )
+    res = get_alias_recipient_name(alias)
+    assert res.message is not None
+    assert res.name == f"{custom_domain.name} <{alias.email}>"

app/tests/test_custom_domain_validation.py

@@ -5,7 +5,7 @@ from app.constants import DMARC_RECORD
 from app.custom_domain_validation import CustomDomainValidation
 from app.db import Session
 from app.models import CustomDomain, User
-from app.dns_utils import InMemoryDNSClient
+from app.dns_utils import InMemoryDNSClient, MxRecord
 from app.proton.utils import get_proton_partner
 from app.utils import random_string
 from tests.utils import create_new_user, random_domain
@@ -58,6 +58,123 @@ def test_custom_domain_validation_get_dkim_records_for_partner():
     assert records["dkim._domainkey"] == f"dkim._domainkey.{dkim_domain}"
 
 
+# get_expected_mx_records
+def test_custom_domain_validation_get_expected_mx_records_regular_domain():
+    domain = random_domain()
+    custom_domain = create_custom_domain(domain)
+
+    partner_id = get_proton_partner().id
+
+    dkim_domain = random_domain()
+    validator = CustomDomainValidation(
+        domain, partner_domains={partner_id: dkim_domain}
+    )
+    records = validator.get_expected_mx_records(custom_domain)
+    # As the domain is not a partner_domain,default records should be used even if
+    # there is a config for the partner
+    assert len(records) == len(config.EMAIL_SERVERS_WITH_PRIORITY)
+    for i in range(len(config.EMAIL_SERVERS_WITH_PRIORITY)):
+        config_record = config.EMAIL_SERVERS_WITH_PRIORITY[i]
+        assert records[i].priority == config_record[0]
+        assert records[i].domain == config_record[1]
+
+
+def test_custom_domain_validation_get_expected_mx_records_domain_from_partner():
+    domain = random_domain()
+    custom_domain = create_custom_domain(domain)
+
+    partner_id = get_proton_partner().id
+    custom_domain.partner_id = partner_id
+    Session.commit()
+
+    dkim_domain = random_domain()
+    validator = CustomDomainValidation(dkim_domain)
+    records = validator.get_expected_mx_records(custom_domain)
+    # As the domain is a partner_domain but there is no custom config for partner, default records
+    # should be used
+    assert len(records) == len(config.EMAIL_SERVERS_WITH_PRIORITY)
+    for i in range(len(config.EMAIL_SERVERS_WITH_PRIORITY)):
+        config_record = config.EMAIL_SERVERS_WITH_PRIORITY[i]
+        assert records[i].priority == config_record[0]
+        assert records[i].domain == config_record[1]
+
+
+def test_custom_domain_validation_get_expected_mx_records_domain_from_partner_with_custom_config():
+    domain = random_domain()
+    custom_domain = create_custom_domain(domain)
+
+    partner_id = get_proton_partner().id
+    custom_domain.partner_id = partner_id
+    Session.commit()
+
+    dkim_domain = random_domain()
+    expected_mx_domain = random_domain()
+    validator = CustomDomainValidation(
+        dkim_domain, partner_domains={partner_id: expected_mx_domain}
+    )
+    records = validator.get_expected_mx_records(custom_domain)
+    # As the domain is a partner_domain and there is a custom config for partner, partner records
+    # should be used
+    assert len(records) == 2
+
+    assert records[0].priority == 10
+    assert records[0].domain == f"mx1.{expected_mx_domain}."
+    assert records[1].priority == 20
+    assert records[1].domain == f"mx2.{expected_mx_domain}."
+
+
+# get_expected_spf_records
+def test_custom_domain_validation_get_expected_spf_record_regular_domain():
+    domain = random_domain()
+    custom_domain = create_custom_domain(domain)
+
+    partner_id = get_proton_partner().id
+
+    dkim_domain = random_domain()
+    validator = CustomDomainValidation(
+        domain, partner_domains={partner_id: dkim_domain}
+    )
+    record = validator.get_expected_spf_record(custom_domain)
+    # As the domain is not a partner_domain, default records should be used even if
+    # there is a config for the partner
+    assert record == f"v=spf1 include:{config.EMAIL_DOMAIN} ~all"
+
+
+def test_custom_domain_validation_get_expected_spf_record_domain_from_partner():
+    domain = random_domain()
+    custom_domain = create_custom_domain(domain)
+
+    partner_id = get_proton_partner().id
+    custom_domain.partner_id = partner_id
+    Session.commit()
+
+    dkim_domain = random_domain()
+    validator = CustomDomainValidation(dkim_domain)
+    record = validator.get_expected_spf_record(custom_domain)
+    # As the domain is a partner_domain but there is no custom config for partner, default records
+    # should be used
+    assert record == f"v=spf1 include:{config.EMAIL_DOMAIN} ~all"
+
+
+def test_custom_domain_validation_get_expected_spf_record_domain_from_partner_with_custom_config():
+    domain = random_domain()
+    custom_domain = create_custom_domain(domain)
+
+    partner_id = get_proton_partner().id
+    custom_domain.partner_id = partner_id
+    Session.commit()
+
+    dkim_domain = random_domain()
+    expected_mx_domain = random_domain()
+    validator = CustomDomainValidation(
+        dkim_domain, partner_domains={partner_id: expected_mx_domain}
+    )
+    record = validator.get_expected_spf_record(custom_domain)
+    # As the domain is a partner_domain and there is a custom config for partner, partner records
+    # should be used
+    assert record == f"v=spf1 include:{expected_mx_domain} ~all"
+
+
 # validate_dkim_records
 def test_custom_domain_validation_validate_dkim_records_empty_records_failure():
     dns_client = InMemoryDNSClient()
@@ -253,7 +370,7 @@ def test_custom_domain_validation_validate_mx_records_wrong_records_failure():
 
     wrong_record_1 = random_string()
     wrong_record_2 = random_string()
-    wrong_records = [(10, wrong_record_1), (20, wrong_record_2)]
+    wrong_records = [MxRecord(10, wrong_record_1), MxRecord(20, wrong_record_2)]
     dns_client.set_mx_records(domain.domain, wrong_records)
     res = validator.validate_mx_records(domain)
 
@@ -270,7 +387,7 @@ def test_custom_domain_validation_validate_mx_records_success():
 
     domain = create_custom_domain(random_domain())
 
-    dns_client.set_mx_records(domain.domain, config.EMAIL_SERVERS_WITH_PRIORITY)
+    dns_client.set_mx_records(domain.domain, validator.get_expected_mx_records(domain))
     res = validator.validate_mx_records(domain)
 
     assert res.success is True
@@ -328,6 +445,58 @@ def test_custom_domain_validation_validate_spf_records_success():
     assert db_domain.spf_verified is True
 
 
+def test_custom_domain_validation_validate_spf_records_partner_domain_success():
+    dns_client = InMemoryDNSClient()
+    proton_partner_id = get_proton_partner().id
+
+    expected_domain = random_domain()
+    validator = CustomDomainValidation(
+        dkim_domain=random_domain(),
+        dns_client=dns_client,
+        partner_domains={proton_partner_id: expected_domain},
+    )
+
+    domain = create_custom_domain(random_domain())
+    domain.partner_id = proton_partner_id
+    Session.commit()
+
+    dns_client.set_txt_record(domain.domain, [f"v=spf1 include:{expected_domain}"])
+    res = validator.validate_spf_records(domain)
+
+    assert res.success is True
+    assert len(res.errors) == 0
+
+    db_domain = CustomDomain.get_by(id=domain.id)
+    assert db_domain.spf_verified is True
+
+
+def test_custom_domain_validation_validate_spf_cleans_verification_record():
+    dns_client = InMemoryDNSClient()
+    proton_partner_id = get_proton_partner().id
+
+    expected_domain = random_domain()
+    validator = CustomDomainValidation(
+        dkim_domain=random_domain(),
+        dns_client=dns_client,
+        partner_domains={proton_partner_id: expected_domain},
+    )
+
+    domain = create_custom_domain(random_domain())
+    domain.partner_id = proton_partner_id
+    Session.commit()
+
+    wrong_record = random_string()
+    dns_client.set_txt_record(
+        hostname=domain.domain,
+        txt_list=[wrong_record, validator.get_ownership_verification_record(domain)],
+    )
+    res = validator.validate_spf_records(domain)
+
+    assert res.success is False
+    assert len(res.errors) == 1
+    assert res.errors[0] == wrong_record
+
+
 # validate_dmarc_records
 def test_custom_domain_validation_validate_dmarc_records_empty_failure():
     dns_client = InMemoryDNSClient()

app/tests/test_dns_utils.py

@@ -3,6 +3,7 @@ from app.dns_utils import (
     get_network_dns_client,
     is_mx_equivalent,
     InMemoryDNSClient,
+    MxRecord,
 )
 
 from tests.utils import random_domain
@@ -17,8 +18,8 @@ def test_get_mx_domains():
     assert len(r) > 0
 
     for x in r:
-        assert x[0] > 0
-        assert x[1]
+        assert x.priority > 0
+        assert x.domain
 
 
 def test_get_spf_domain():
@@ -33,20 +34,32 @@ def test_get_txt_record():
 
 def test_is_mx_equivalent():
     assert is_mx_equivalent([], [])
-    assert is_mx_equivalent([(1, "domain")], [(1, "domain")])
     assert is_mx_equivalent(
-        [(10, "domain1"), (20, "domain2")], [(10, "domain1"), (20, "domain2")]
+        mx_domains=[MxRecord(1, "domain")], ref_mx_domains=[MxRecord(1, "domain")]
     )
     assert is_mx_equivalent(
-        [(5, "domain1"), (10, "domain2")], [(10, "domain1"), (20, "domain2")]
+        mx_domains=[MxRecord(10, "domain1"), MxRecord(20, "domain2")],
+        ref_mx_domains=[MxRecord(10, "domain1"), MxRecord(20, "domain2")],
     )
     assert is_mx_equivalent(
-        [(5, "domain1"), (10, "domain2"), (20, "domain3")],
-        [(10, "domain1"), (20, "domain2")],
+        mx_domains=[MxRecord(5, "domain1"), MxRecord(10, "domain2")],
+        ref_mx_domains=[MxRecord(10, "domain1"), MxRecord(20, "domain2")],
+    )
+    assert is_mx_equivalent(
+        mx_domains=[
+            MxRecord(5, "domain1"),
+            MxRecord(10, "domain2"),
+            MxRecord(20, "domain3"),
+        ],
+        ref_mx_domains=[MxRecord(10, "domain1"), MxRecord(20, "domain2")],
     )
     assert not is_mx_equivalent(
-        [(5, "domain1"), (10, "domain2")],
-        [(10, "domain1"), (20, "domain2"), (20, "domain3")],
+        mx_domains=[MxRecord(5, "domain1"), MxRecord(10, "domain2")],
+        ref_mx_domains=[
+            MxRecord(10, "domain1"),
+            MxRecord(20, "domain2"),
+            MxRecord(20, "domain3"),
+        ],
     )
 
 

app/tests/test_email_utils.py

@@ -90,12 +90,19 @@ def test_can_be_used_as_personal_email(flask_client):
     assert not email_can_be_used_as_mailbox("ab@sl.local")
     assert not email_can_be_used_as_mailbox("hey@d1.test")
 
-    # custom domain
+    # custom domain as SL domain
     domain = random_domain()
     user = create_new_user()
-    CustomDomain.create(user_id=user.id, domain=domain, verified=True, commit=True)
+    domain_obj = CustomDomain.create(
+        user_id=user.id, domain=domain, verified=True, is_sl_subdomain=True, flush=True
+    )
     assert not email_can_be_used_as_mailbox(f"hey@{domain}")
 
+    # custom domain is NOT SL domain
+    domain_obj.is_sl_subdomain = False
+    Session.flush()
+    assert email_can_be_used_as_mailbox(f"hey@{domain}")
+
     # disposable domain
     disposable_domain = random_domain()
     InvalidMailboxDomain.create(domain=disposable_domain, commit=True)