🐛(dimail) authorize alias and mailbox with same local part

a mailbox and alias can coexist despite having the same local part
This commit is contained in:
Marie PUPO JEAMMET
2026-01-14 16:56:24 +01:00
committed by Marie
parent d2ef9e0beb
commit 9c62efc9f8
6 changed files with 71 additions and 94 deletions

View File

@@ -8,6 +8,7 @@ and this project adheres to
## [Unreleased] ## [Unreleased]
- 🐛(dimail) allow mailboxes and aliases to have the same local part #986
- 🐛(dimail) ignore oxadmin when importing mailboxes from dimail #986 - 🐛(dimail) ignore oxadmin when importing mailboxes from dimail #986
- ✨(aliases) delete all aliases in one call #1002 - ✨(aliases) delete all aliases in one call #1002
- ✨(aliases) fix deleting single aliases #1002 - ✨(aliases) fix deleting single aliases #1002

View File

@@ -79,17 +79,6 @@ class MailboxSerializer(serializers.ModelSerializer):
return mailbox return mailbox
def validate_local_part(self, value):
"""Validate this local part does not match a mailbox."""
if models.Alias.objects.filter(
local_part=value, domain__slug=self.context["domain_slug"]
):
raise exceptions.ValidationError(
f'Local part "{value}" already used by an alias.'
)
return value
class MailboxUpdateSerializer(MailboxSerializer): class MailboxUpdateSerializer(MailboxSerializer):
"""A more restrictive serializer when updating mailboxes""" """A more restrictive serializer when updating mailboxes"""
@@ -338,14 +327,3 @@ class AliasSerializer(serializers.ModelSerializer):
return super().create(validated_data) return super().create(validated_data)
return None return None
def validate_local_part(self, value):
"""Validate this local part does not match a mailbox."""
if models.Mailbox.objects.filter(
local_part=value, domain__slug=self.context["domain_slug"]
).exists():
raise exceptions.ValidationError(
f'Local part "{value}" already used by a mailbox.'
)
return value

View File

@@ -81,26 +81,6 @@ def test_api_aliases_create__duplicate_forbidden():
assert models.Alias.objects.filter(domain=access.domain).count() == 1 assert models.Alias.objects.filter(domain=access.domain).count() == 1
def test_api_aliases_create__existing_mailbox_bad_request():
"""Cannot create alias if local_part is already used by a mailbox."""
access = factories.MailDomainAccessFactory(
role="owner", domain=factories.MailDomainEnabledFactory()
)
mailbox = factories.MailboxFactory(domain=access.domain)
client = APIClient()
client.force_login(access.user)
response = client.post(
f"/api/v1.0/mail-domains/{access.domain.slug}/aliases/",
{"local_part": mailbox.local_part, "destination": "someone@outsidedomain.com"},
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
assert response.json() == {
"local_part": [f'Local part "{mailbox.local_part}" already used by a mailbox.']
}
assert not models.Alias.objects.exists()
@responses.activate @responses.activate
def test_api_aliases_create__async_alias_bad_request(dimail_token_ok): def test_api_aliases_create__async_alias_bad_request(dimail_token_ok):
""" """
@@ -151,17 +131,14 @@ def test_api_aliases_create__admins_ok(role, dimail_token_ok):
client.force_login(access.user) client.force_login(access.user)
# Prepare responses # Prepare responses
# token response in fixtures # token response in fixtures
responses.add( responses.post(
responses.POST,
re.compile(rf".*/domains/{access.domain.name}/aliases/"), re.compile(rf".*/domains/{access.domain.name}/aliases/"),
body=json.dumps( json={
{ "username": "contact",
"username": "contact", "domain": access.domain.name,
"domain": access.domain.name, "destination": "someone@outsidedomain.com",
"destination": "someone@outsidedomain.com", "allow_to_send": True,
"allow_to_send": True, },
}
),
status=status.HTTP_201_CREATED, status=status.HTTP_201_CREATED,
content_type="application/json", content_type="application/json",
) )
@@ -174,3 +151,34 @@ def test_api_aliases_create__admins_ok(role, dimail_token_ok):
alias = models.Alias.objects.get() alias = models.Alias.objects.get()
assert alias.local_part == "contact" assert alias.local_part == "contact"
assert alias.destination == "someone@outsidedomain.com" assert alias.destination == "someone@outsidedomain.com"
@responses.activate
def test_api_aliases_create__existing_mailbox_ok(dimail_token_ok):
"""Can create alias even if local_part is already used by a mailbox."""
access = factories.MailDomainAccessFactory(
role="owner", domain=factories.MailDomainEnabledFactory()
)
mailbox = factories.MailboxFactory(domain=access.domain)
client = APIClient()
client.force_login(access.user)
responses.post(
re.compile(rf".*/domains/{access.domain.name}/aliases/"),
json={
"username": mailbox.local_part,
"domain": access.domain.name,
"destination": "someone@outsidedomain.com",
"allow_to_send": False,
},
status=status.HTTP_201_CREATED,
content_type="application/json",
)
response = client.post(
f"/api/v1.0/mail-domains/{access.domain.slug}/aliases/",
{"local_part": mailbox.local_part, "destination": "someone@outsidedomain.com"},
)
assert response.status_code == status.HTTP_201_CREATED
assert models.Alias.objects.exists()

View File

@@ -439,10 +439,9 @@ def test_api_mailboxes__create_pending_mailboxes(domain_status, mailbox_data):
assert mailbox.status == "pending" assert mailbox.status == "pending"
def test_api_mailboxes__existing_alias_bad_request(mailbox_data): @responses.activate
""" def test_api_mailboxes__existing_alias_ok(mailbox_data, dimail_token_ok):
Cannot create mailbox if local_part is already used by an alias. """Can create mailbox even if local_part is already used by an alias."""
"""
alias = factories.AliasFactory() alias = factories.AliasFactory()
access = factories.MailDomainAccessFactory( access = factories.MailDomainAccessFactory(
role=enums.MailDomainRoleChoices.ADMIN, domain=alias.domain role=enums.MailDomainRoleChoices.ADMIN, domain=alias.domain
@@ -450,7 +449,16 @@ def test_api_mailboxes__existing_alias_bad_request(mailbox_data):
client = APIClient() client = APIClient()
client.force_login(access.user) client.force_login(access.user)
# No response because we expect no outside calls to be made
responses.post(
re.compile(rf".*/domains/{access.domain.name}/mailboxes/"),
body=response_mailbox_created(
f"{mailbox_data['local_part']}@{access.domain.name}"
),
status=status.HTTP_201_CREATED,
content_type="application/json",
)
response = client.post( response = client.post(
f"/api/v1.0/mail-domains/{access.domain.slug}/mailboxes/", f"/api/v1.0/mail-domains/{access.domain.slug}/mailboxes/",
{ {
@@ -461,11 +469,8 @@ def test_api_mailboxes__existing_alias_bad_request(mailbox_data):
}, },
format="json", format="json",
) )
assert response.status_code == status.HTTP_400_BAD_REQUEST assert response.status_code == status.HTTP_201_CREATED
assert response.json() == { assert models.Mailbox.objects.exists()
"local_part": [f'Local part "{alias.local_part}" already used by an alias.']
}
assert not models.Mailbox.objects.exists()
### REACTING TO DIMAIL-API ### REACTING TO DIMAIL-API

View File

@@ -66,8 +66,8 @@ def test_dimail_synchronization__already_sync(dimail_token_ok):
@responses.activate @responses.activate
def test_dimail_synchronization__synchronize_mailboxes(caplog, dimail_token_ok): # pylint: disable=W0613, R0914 def test_dimail_synchronization__synchronize_mailboxes(caplog, dimail_token_ok): # pylint: disable=W0613, R0914
"""A mailbox existing solely on dimail should be synchronized """Importing mailboxes from dimail should synchronize valid mailboxes
upon calling sync function on its domain""" and log errors for invalid ones."""
caplog.set_level(logging.INFO) caplog.set_level(logging.INFO)
domain = factories.MailDomainEnabledFactory() domain = factories.MailDomainEnabledFactory()
@@ -143,7 +143,7 @@ def test_dimail_synchronization__synchronize_mailboxes(caplog, dimail_token_ok):
imported_mailboxes = dimail_client.import_mailboxes(domain) imported_mailboxes = dimail_client.import_mailboxes(domain)
# 4 imports failed: oxadmin, wrong domain, HeaderParseError, NonASCIILocalPartDefect # 4 imports failed: oxadmin, wrong domain, HeaderParseError, NonASCIILocalPartDefect
assert len(caplog.records) == 6 assert len(caplog.records) == 5
log_messages = [record.message for record in caplog.records] log_messages = [record.message for record in caplog.records]
expected_messages = [ expected_messages = [
@@ -152,21 +152,21 @@ def test_dimail_synchronization__synchronize_mailboxes(caplog, dimail_token_ok):
f"Import of email {mailbox_with_invalid_domain['email']} failed with error Invalid Domain", f"Import of email {mailbox_with_invalid_domain['email']} failed with error Invalid Domain",
f"Import of email {mailbox_with_invalid_local_part['email']} failed with error local-part \ f"Import of email {mailbox_with_invalid_local_part['email']} failed with error local-part \
contains non-ASCII characters)", contains non-ASCII characters)",
f"{existing_alias.local_part} already used in an existing alias.",
] ]
for message in expected_messages: for message in expected_messages:
assert message in log_messages assert message in log_messages
assert imported_mailboxes == [mailbox_valid["email"]] assert models.Mailbox.objects.count() == 2
mailbox = models.Mailbox.objects.get() assert imported_mailboxes == [
assert mailbox.local_part == "validmailbox" mailbox_valid["email"],
assert mailbox.status == enums.MailboxStatusChoices.ENABLED mailbox_existing_alias["email"],
]
@responses.activate @responses.activate
def test_dimail_synchronization__synchronize_aliases(dimail_token_ok): # pylint: disable=unused-argument def test_dimail_synchronization__synchronize_aliases(dimail_token_ok): # pylint: disable=unused-argument
"""Should import aliases from dimail if they don't already exist """Importing aliases from dimail should synchronize valid aliases
and if username is not already used for mailbox""" and log errors for invalid ones."""
alias = factories.AliasFactory() alias = factories.AliasFactory()
dimail_client = DimailAPIClient() dimail_client = DimailAPIClient()
@@ -178,11 +178,11 @@ def test_dimail_synchronization__synchronize_aliases(dimail_token_ok): # pylint
{ {
"username": "contact", "username": "contact",
"domain": alias.domain.name, "domain": alias.domain.name,
"destination": alias.destination, # same destination "destination": alias.destination, # same destination = ok
"allow_to_send": False, "allow_to_send": False,
}, },
{ {
"username": alias.local_part, # same username "username": alias.local_part, # same username = ok
"domain": alias.domain.name, "domain": alias.domain.name,
"destination": "maheius.endorecles@somethingelse.com", "destination": "maheius.endorecles@somethingelse.com",
"allow_to_send": False, "allow_to_send": False,
@@ -193,7 +193,7 @@ def test_dimail_synchronization__synchronize_aliases(dimail_token_ok): # pylint
"destination": alias.destination, "destination": alias.destination,
"allow_to_send": False, "allow_to_send": False,
}, },
{ # username already used for a mailbox { # mailbox with same username = ok
"username": existing_mailbox.local_part, "username": existing_mailbox.local_part,
"domain": alias.domain.name, "domain": alias.domain.name,
"destination": existing_mailbox.secondary_email, "destination": existing_mailbox.secondary_email,
@@ -209,8 +209,8 @@ def test_dimail_synchronization__synchronize_aliases(dimail_token_ok): # pylint
imported_aliases = dimail_client.import_aliases(alias.domain) imported_aliases = dimail_client.import_aliases(alias.domain)
assert len(imported_aliases) == 2 assert len(imported_aliases) == 3
assert models.Alias.objects.count() == 3 assert models.Alias.objects.count() == 4
@pytest.mark.parametrize( @pytest.mark.parametrize(

View File

@@ -394,16 +394,6 @@ class DimailAPIClient:
) )
continue continue
if address.username in [
alias_.local_part
for alias_ in models.Alias.objects.filter(domain=domain)
]:
logger.warning(
"%s already used in an existing alias.",
address.username,
)
continue
if str(address) not in [ if str(address) not in [
str(people_mailbox) for people_mailbox in people_mailboxes str(people_mailbox) for people_mailbox in people_mailboxes
]: ]:
@@ -842,18 +832,13 @@ class DimailAPIClient:
(known_alias.local_part, known_alias.destination) (known_alias.local_part, known_alias.destination)
for known_alias in models.Alias.objects.filter(domain=domain) for known_alias in models.Alias.objects.filter(domain=domain)
] ]
known_mailboxes = [
known_mailbox.local_part
for known_mailbox in models.Mailbox.objects.filter(domain=domain)
]
imported_aliases = [] imported_aliases = []
for incoming_alias in incoming_aliases: for incoming_alias in incoming_aliases:
if ( if (
incoming_alias["username"], incoming_alias["username"],
incoming_alias["destination"], incoming_alias["destination"],
) not in known_aliases and incoming_alias[ ) not in known_aliases:
"username"
] not in known_mailboxes:
try: try:
new_alias = models.Alias.objects.create( new_alias = models.Alias.objects.create(
local_part=incoming_alias["username"], local_part=incoming_alias["username"],