Mirror
/
app-MAIL-temp
mirror of https://github.com/simple-login/app.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Use the alias domain for contacts

This commit is contained in:
Adrià Casajús 2023-04-11 17:04:42 +02:00
parent ec666aee87
commit 1e07cd2365
No known key found for this signature in database
GPG Key ID: F0033226A5AFC9B9
6 changed files with 41 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
app/dashboard/views/alias_contact_manager.py
View File

@ -90,7 +90,7 @@ def create_contact(user: User, alias: Alias, contact_address: str) -> Contact:
alias_id=alias.id,
website_email=contact_email,
name=contact_name,
reply_email=generate_reply_email(contact_email, user),
reply_email=generate_reply_email(contact_email, alias),
)
LOG.d(

12
app/email_utils.py
View File

@ -1043,7 +1043,7 @@ def replace(msg: Union[Message, str], old, new) -> Union[Message, str]:
return msg
def generate_reply_email(contact_email: str, user: User) -> str:
def generate_reply_email(contact_email: str, alias: Alias) -> str:
"""
generate a reply_email (aka reverse-alias), make sure it isn't used by any contact
"""
@ -1054,6 +1054,7 @@ def generate_reply_email(contact_email: str, user: User) -> str:
include_sender_in_reverse_alias = False
user = alias.user
# user has set this option explicitly
if user.include_sender_in_reverse_alias is not None:
include_sender_in_reverse_alias = user.include_sender_in_reverse_alias
@ -1068,6 +1069,7 @@ def generate_reply_email(contact_email: str, user: User) -> str:
contact_email = contact_email.replace(".", "_")
contact_email = convert_to_alphanumeric(contact_email)
reply_domain = alias.get_domain()
# not use while to avoid infinite loop
for _ in range(1000):
if include_sender_in_reverse_alias and contact_email:
@ -1075,15 +1077,17 @@ def generate_reply_email(contact_email: str, user: User) -> str:
reply_email = (
# do not use the ra+ anymore
# f"ra+{contact_email}+{random_string(random_length)}@{config.EMAIL_DOMAIN}"
f"{contact_email}_{random_string(random_length)}@{config.EMAIL_DOMAIN}"
f"{contact_email}_{random_string(random_length)}@{reply_domain}"
)
else:
random_length = random.randint(20, 50)
# do not use the ra+ anymore
# reply_email = f"ra+{random_string(random_length)}@{config.EMAIL_DOMAIN}"
reply_email = f"{random_string(random_length)}@{config.EMAIL_DOMAIN}"
reply_email = f"{random_string(random_length)}@{reply_domain}"
if not Contact.get_by(reply_email=reply_email):
if not Contact.get_by(reply_email=reply_email) and not Alias.get_by(
email=reply_email
):
return reply_email
raise Exception("Cannot generate reply email")

15
app/models.py
View File

@ -1297,12 +1297,15 @@ def generate_email(
scheme: int = AliasGeneratorEnum.word.value,
in_hex: bool = False,
alias_domain=config.FIRST_ALIAS_DOMAIN,
retries: int = 5,
) -> str:
"""generate an email address that does not exist before
:param alias_domain: the domain used to generate the alias.
:param scheme: int, value of AliasGeneratorEnum, indicate how the email is generated
:type in_hex: bool, if the generate scheme is uuid, is hex favorable?
"""
if retries <= 0:
raise Exception("Cannot generate alias after many retries")
if scheme == AliasGeneratorEnum.uuid.value:
name = uuid.uuid4().hex if in_hex else uuid.uuid4().__str__()
random_email = name + "@" + alias_domain
@ -1312,15 +1315,17 @@ def generate_email(
random_email = random_email.lower().strip()
# check that the client does not exist yet
if not Alias.get_by(email=random_email) and not DeletedAlias.get_by(
email=random_email
if (
not Alias.get_by(email=random_email)
and not DeletedAlias.get_by(email=random_email)
and not Contact.get_by(reply_email=random_email)
):
LOG.d("generate email %s", random_email)
return random_email
# Rerun the function
LOG.w("email %s already exists, generate a new email", random_email)
return generate_email(scheme=scheme, in_hex=in_hex)
return generate_email(scheme=scheme, in_hex=in_hex, retries=retries - 1)
class Alias(Base, ModelMixin):
@ -1581,6 +1586,10 @@ class Alias(Base, ModelMixin):
else:
return self.user.email
def get_domain(self) -> str:
splitPos = self.email.find("@")
return self.email[splitPos + 1 :]
def __repr__(self):
return f"<Alias {self.id} {self.email}>"

6
email_handler.py
View File

@ -243,7 +243,7 @@ def get_or_create_contact(from_header: str, mail_from: str, alias: Alias) -> Con
website_email=contact_email,
name=contact_name,
mail_from=mail_from,
reply_email=generate_reply_email(contact_email, alias.user)
reply_email=generate_reply_email(contact_email, alias)
if is_valid_email(contact_email)
else NOREPLY,
automatic_created=True,
@ -304,7 +304,7 @@ def get_or_create_reply_to_contact(
alias_id=alias.id,
website_email=contact_address,
name=contact_name,
reply_email=generate_reply_email(contact_address, alias.user),
reply_email=generate_reply_email(contact_address, alias),
automatic_created=True,
)
Session.commit()
@ -372,7 +372,7 @@ def replace_header_when_forward(msg: Message, alias: Alias, header: str):
alias_id=alias.id,
website_email=contact_email,
name=full_address.display_name,
reply_email=generate_reply_email(contact_email, alias.user),
reply_email=generate_reply_email(contact_email, alias),
is_cc=header.lower() == "cc",
automatic_created=True,
)

27
tests/test_email_utils.py
View File

@ -48,6 +48,7 @@ from app.models import (
IgnoreBounceSender,
InvalidMailboxDomain,
VerpType,
AliasGeneratorEnum,
)
# flake8: noqa: E101, W191
@ -469,33 +470,37 @@ def test_replace_str():
def test_generate_reply_email(flask_client):
user = create_new_user()
reply_email = generate_reply_email("test@example.org", user)
assert reply_email.endswith(EMAIL_DOMAIN)
alias = Alias.create_new_random(user, AliasGeneratorEnum.uuid.value)
Session.commit()
reply_email = generate_reply_email("test@example.org", alias)
assert reply_email.endswith(alias.get_domain())
reply_email = generate_reply_email("", user)
assert reply_email.endswith(EMAIL_DOMAIN)
reply_email = generate_reply_email("", alias)
assert reply_email.endswith(alias.get_domain())
def test_generate_reply_email_include_sender_in_reverse_alias(flask_client):
# user enables include_sender_in_reverse_alias
user = create_new_user()
alias = Alias.create_new_random(user, AliasGeneratorEnum.uuid.value)
Session.commit()
user.include_sender_in_reverse_alias = True
reply_email = generate_reply_email("test@example.org", user)
reply_email = generate_reply_email("test@example.org", alias)
assert reply_email.startswith("test_at_example_org")
assert reply_email.endswith(EMAIL_DOMAIN)
assert reply_email.endswith(alias.get_domain())
reply_email = generate_reply_email("", user)
assert reply_email.endswith(EMAIL_DOMAIN)
reply_email = generate_reply_email("", alias)
assert reply_email.endswith(alias.get_domain())
reply_email = generate_reply_email("👌汉字@example.org", user)
reply_email = generate_reply_email("👌汉字@example.org", alias)
assert reply_email.startswith("yizi_at_example_org")
# make sure reply_email only contain lowercase
reply_email = generate_reply_email("TEST@example.org", user)
reply_email = generate_reply_email("TEST@example.org", alias)
assert reply_email.startswith("test_at_example_org")
reply_email = generate_reply_email("test.dot@example.org", user)
reply_email = generate_reply_email("test.dot@example.org", alias)
assert reply_email.startswith("test_dot_at_example_org")

2
tests/test_models.py
View File

@ -312,6 +312,6 @@ def test_create_contact_for_noreply(flask_client):
user_id=user.id,
alias_id=alias.id,
website_email=NOREPLY,
reply_email=generate_reply_email(NOREPLY, user),
reply_email=generate_reply_email(NOREPLY, alias),
)
assert contact.website_email == NOREPLY