Skip to content

Commit

Permalink
Add show domain alias and deleted alias
Browse files Browse the repository at this point in the history
  • Loading branch information
acasajus committed Aug 8, 2024
1 parent a8988cb commit d5869b8
Show file tree
Hide file tree
Showing 4 changed files with 221 additions and 58 deletions.
110 changes: 70 additions & 40 deletions app/admin_model.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
from __future__ import annotations
from typing import Optional

import arrow
Expand Down Expand Up @@ -30,6 +31,8 @@
Newsletter,
PADDLE_SUBSCRIPTION_GRACE_DAYS,
Mailbox,
DeletedAlias,
DomainDeletedAlias,
)
from app.newsletter_utils import send_newsletter_to_user, send_newsletter_to_address

Expand Down Expand Up @@ -729,6 +732,67 @@ class InvalidMailboxDomainAdmin(SLModelView):
can_delete = True


class EmailSearchResult:
no_match: bool = True
alias: Optional[Alias] = None
mailbox: Optional[Mailbox] = None
deleted_alias: Optional[DeletedAlias] = None
deleted_custom_alias: Optional[DomainDeletedAlias] = None
user: Optional[User] = None

@staticmethod
def from_email(email: str) -> EmailSearchResult:
output = EmailSearchResult()
alias = Alias.get_by(email=email)
if alias:
output.alias = alias
output.no_match = False
return output
user = User.get_by(email=email)
if user:
output.user = user
output.no_match = False
return output
mailbox = Mailbox.get_by(email=email)
if mailbox:
output.mailbox = mailbox
output.no_match = False
return output
deleted_alias = DeletedAlias.get_by(email=email)
if deleted_alias:
output.deleted_alias = deleted_alias
output.no_match = False
return output
domain_deleted_alias = DomainDeletedAlias.get_by(email=email)
if domain_deleted_alias:
output.domain_deleted_alias = domain_deleted_alias
output.no_match = False
return output


class EmailSearchHelpers:
@staticmethod
def mailbox_list(user: User) -> list[Mailbox]:
return (
Mailbox.filter_by(user_id=user.id)
.order_by(Mailbox.id.asc())
.limit(10)
.all()
)

@staticmethod
def mailbox_count(user: User) -> int:
return Mailbox.filter_by(user_id=user.id).order_by(Mailbox.id.asc()).count()

@staticmethod
def alias_list(user: User) -> list[Alias]:
return Alias.filter_by(user_id=user.id).order_by(Alias.id.asc()).limit(10).all()

@staticmethod
def alias_count(user: User) -> int:
return Alias.filter_by(user_id=user.id).count()


class EmailSearchAdmin(BaseView):
def is_accessible(self):
return current_user.is_authenticated and current_user.is_admin
Expand All @@ -740,50 +804,16 @@ def inaccessible_callback(self, name, **kwargs):

@expose("/", methods=["GET", "POST"])
def index(self):
alias = None
user = None
mailbox = None
no_match = False
email = None
search = EmailSearchResult()
email = ""
if request.form and request.form["email"]:
email = request.form["email"]
alias = Alias.get_by(email=email)
user = User.get_by(email=email)
mailbox = Mailbox.get_by(email=email)
if not alias and not user and not mailbox:
no_match = True

def user_mailboxes(user_id: int) -> list[Mailbox]:
return (
Mailbox.filter_by(user_id=user_id)
.order_by(Mailbox.id.asc())
.limit(10)
.all()
)

def user_mailboxes_count(user_id: int) -> int:
return Mailbox.filter_by(user_id=user_id).order_by(Mailbox.id.asc()).count()

def user_aliases(user_id: int) -> list[Alias]:
return (
Alias.filter_by(user_id=user_id)
.order_by(Alias.id.asc())
.limit(10)
.all()
)

def user_aliases_count(user_id: int) -> int:
return Alias.filter_by(user_id=user_id).count()
email = email.strip()
search = EmailSearchResult.from_email(email)

return self.render(
"admin/email_search.html",
email=email,
no_match=no_match,
alias=alias,
mailbox=mailbox,
user=user,
user_aliases=user_aliases,
user_aliases_count=user_aliases_count,
user_mailboxes=user_mailboxes,
user_mailboxes_count=user_mailboxes_count,
data=search,
helper=EmailSearchHelpers,
)
23 changes: 21 additions & 2 deletions app/alias_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -63,12 +63,16 @@ def get_user_if_alias_would_auto_create(
# Prevent addresses with unicode characters (🤯) in them for now.
validate_email(address, check_deliverability=False, allow_smtputf8=False)
except EmailNotValidError:
LOG.i(f"Not creating alias for {address} because email is invalid")
return None

domain_and_rule = check_if_alias_can_be_auto_created_for_custom_domain(
address, notify_user=notify_user
)
if DomainDeletedAlias.get_by(email=address):
LOG.i(
f"Not creating alias for {address} because it was previously deleted for this domain"
)
return None
if domain_and_rule:
return domain_and_rule[0].user
Expand All @@ -93,6 +97,9 @@ def check_if_alias_can_be_auto_created_for_custom_domain(
custom_domain: CustomDomain = CustomDomain.get_by(domain=alias_domain)

if not custom_domain:
LOG.i(
f"Cannot auto-create custom domain alias for {address} because there's no custom domain for {alias_domain}"
)
return None

user: User = custom_domain.user
Expand All @@ -108,6 +115,9 @@ def check_if_alias_can_be_auto_created_for_custom_domain(

if not custom_domain.catch_all:
if len(custom_domain.auto_create_rules) == 0:
LOG.i(
f"Cannot create alias {address} for domain {custom_domain} because it has no catch-all and no rules"
)
return None
local = get_email_local_part(address)

Expand All @@ -121,7 +131,7 @@ def check_if_alias_can_be_auto_created_for_custom_domain(
)
return custom_domain, rule
else: # no rule passes
LOG.d("no rule passed to create %s", local)
LOG.d(f"No rule matches auto-create {address} for domain {custom_domain}")
return None
LOG.d("Create alias via catchall")

Expand All @@ -148,13 +158,17 @@ def check_if_alias_can_be_auto_created_for_a_directory(
sep = "#"
else:
# if there's no directory separator in the alias, no way to auto-create it
LOG.info(f"Cannot auto-create {address} since it has no directory separator")
return None

directory_name = address[: address.find(sep)]
LOG.d("directory_name %s", directory_name)

directory = Directory.get_by(name=directory_name)
if not directory:
LOG.info(
f"Cannot auto-create {address} because there is no directory for {directory_name}"
)
return None

user: User = directory.user
Expand All @@ -163,12 +177,17 @@ def check_if_alias_can_be_auto_created_for_a_directory(
return None

if not user.can_create_new_alias():
LOG.d(f"{user} can't create new directory alias {address}")
LOG.d(
f"{user} can't create new directory alias {address} because user cannot create aliases"
)
if notify_user:
send_cannot_create_directory_alias(user, address, directory_name)
return None

if directory.disabled:
LOG.d(
f"{user} can't create new directory alias {address} bcause directory is disabled"
)
if notify_user:
send_cannot_create_directory_alias_disabled(user, address, directory_name)
return None
Expand Down
4 changes: 3 additions & 1 deletion app/email_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -548,7 +548,9 @@ def can_create_directory_for_address(email_address: str) -> bool:
for domain in config.ALIAS_DOMAINS:
if email_address.endswith("@" + domain):
return True

LOG.i(
f"Cannot create address in directory for {email_address} since it does not belong to a valid directory domain"
)
return False


Expand Down
Loading

0 comments on commit d5869b8

Please sign in to comment.