diff --git a/app/app/admin_model.py b/app/app/admin_model.py index 5d6bf66..2123e36 100644 --- a/app/app/admin_model.py +++ b/app/app/admin_model.py @@ -8,14 +8,16 @@ from flask_admin.form import SecureForm from flask_admin.model.template import EndpointLinkRowAction from markupsafe import Markup -from app import models, s3 +from app import models, s3, config from flask import redirect, url_for, request, flash, Response from flask_admin import expose, AdminIndexView from flask_admin.actions import action from flask_admin.contrib import sqla from flask_login import current_user +from app.custom_domain_validation import CustomDomainValidation, DomainValidationResult from app.db import Session +from app.dns_utils import get_network_dns_client from app.events.event_dispatcher import EventDispatcher from app.events.generated.event_pb2 import EventContent, UserPlanChanged from app.models import ( @@ -39,6 +41,7 @@ from app.models import ( AliasMailbox, AliasAuditLog, UserAuditLog, + CustomDomain, ) from app.newsletter_utils import send_newsletter_to_user, send_newsletter_to_address from app.user_audit_log_utils import emit_user_audit_log, UserAuditLogAction @@ -773,18 +776,19 @@ class InvalidMailboxDomainAdmin(SLModelView): class EmailSearchResult: - no_match: bool = True - alias: Optional[Alias] = None - alias_audit_log: Optional[List[AliasAuditLog]] = None - mailbox: List[Mailbox] = [] - mailbox_count: int = 0 - deleted_alias: Optional[DeletedAlias] = None - deleted_alias_audit_log: Optional[List[AliasAuditLog]] = None - domain_deleted_alias: Optional[DomainDeletedAlias] = None - domain_deleted_alias_audit_log: Optional[List[AliasAuditLog]] = None - user: Optional[User] = None - user_audit_log: Optional[List[UserAuditLog]] = None - query: str + def __init__(self): + self.no_match: bool = True + self.alias: Optional[Alias] = None + self.alias_audit_log: Optional[List[AliasAuditLog]] = None + self.mailbox: List[Mailbox] = [] + self.mailbox_count: int = 0 + self.deleted_alias: Optional[DeletedAlias] = None + self.deleted_alias_audit_log: Optional[List[AliasAuditLog]] = None + self.domain_deleted_alias: Optional[DomainDeletedAlias] = None + self.domain_deleted_alias_audit_log: Optional[List[AliasAuditLog]] = None + self.user: Optional[User] = None + self.user_audit_log: Optional[List[UserAuditLog]] = None + self.query: str @staticmethod def from_email(email: str) -> EmailSearchResult: @@ -916,3 +920,106 @@ class EmailSearchAdmin(BaseView): data=search, helper=EmailSearchHelpers, ) + + +class CustomDomainWithValidationData: + def __init__(self, domain: CustomDomain): + self.domain: CustomDomain = domain + self.ownership_expected: Optional[str] = None + self.ownership_validation: Optional[DomainValidationResult] = None + self.mx_expected: Optional[str] = None + self.mx_validation: Optional[DomainValidationResult] = None + self.spf_expected: Optional[str] = None + self.spf_validation: Optional[DomainValidationResult] = None + self.dkim_expected: {str: str} = {} + self.dkim_validation: {str: str} = {} + + +class CustomDomainSearchResult: + def __init__(self): + self.no_match: bool = False + self.user: Optional[User] = None + self.domains: list[CustomDomainWithValidationData] = [] + + @staticmethod + def from_user(user: Optional[User]) -> CustomDomainSearchResult: + out = CustomDomainSearchResult() + if user is None: + out.no_match = True + return out + out.user = user + dns_client = get_network_dns_client() + validator = CustomDomainValidation( + dkim_domain=config.EMAIL_DOMAIN, + partner_domains=config.PARTNER_DNS_CUSTOM_DOMAINS, + partner_domains_validation_prefixes=config.PARTNER_CUSTOM_DOMAIN_VALIDATION_PREFIXES, + dns_client=dns_client, + ) + for custom_domain in user.custom_domains: + validation_data = CustomDomainWithValidationData(custom_domain) + if not custom_domain.ownership_verified: + validation_data.ownership_expected = ( + validator.get_ownership_verification_record(custom_domain) + ) + validation_data.ownership_validation = ( + validator.validate_domain_ownership(custom_domain) + ) + if not custom_domain.verified: + validation_data.mx_expected = validator.get_expected_mx_records( + custom_domain + ) + validation_data.mx_validation = validator.validate_mx_records( + custom_domain + ) + if not custom_domain.spf_verified: + validation_data.spf_expected = validator.get_expected_spf_record( + custom_domain + ) + validation_data.spf_validation = validator.validate_spf_records( + custom_domain + ) + if not custom_domain.dkim_verified: + validation_data.dkim_expected = validator.get_dkim_records( + custom_domain + ) + validation_data.dkim_validation = validator.validate_dkim_records( + custom_domain + ) + out.domains.append(validation_data) + print(validation_data.dkim_expected, validation_data.dkim_validation) + + return out + + +class CustomDomainSearchAdmin(BaseView): + def is_accessible(self): + return current_user.is_authenticated and current_user.is_admin + + def inaccessible_callback(self, name, **kwargs): + # redirect to login page if user doesn't have access + flash("You don't have access to the admin page", "error") + return redirect(url_for("dashboard.index", next=request.url)) + + @expose("/", methods=["GET", "POST"]) + def index(self): + query = request.args.get("user") + if query is None: + search = CustomDomainSearchResult() + else: + try: + user_id = int(query) + user = User.get_by(id=user_id) + except ValueError: + user = User.get_by(email=query) + if user is None: + cd = CustomDomain.get_by(domain=query) + if cd is not None: + user = cd.user + search = CustomDomainSearchResult.from_user(user) + print("NEW", search.domains) + + return self.render( + "admin/custom_domain_search.html", + data=search, + query=query, + ) diff --git a/app/app/api/views/alias.py b/app/app/api/views/alias.py index 7630b4b..a2f4bdb 100644 --- a/app/app/api/views/alias.py +++ b/app/app/api/views/alias.py @@ -299,7 +299,10 @@ def update_alias(alias_id): changed = True if "mailbox_ids" in data: - mailbox_ids = [int(m_id) for m_id in data.get("mailbox_ids")] + try: + mailbox_ids = [int(m_id) for m_id in data.get("mailbox_ids")] + except ValueError: + return jsonify(error="Invalid mailbox_id"), 400 err = set_mailboxes_for_alias( user_id=user.id, alias=alias, mailbox_ids=mailbox_ids ) diff --git a/app/app/api/views/new_custom_alias.py b/app/app/api/views/new_custom_alias.py index 6cb78c2..94d48a9 100644 --- a/app/app/api/views/new_custom_alias.py +++ b/app/app/api/views/new_custom_alias.py @@ -1,3 +1,4 @@ +from email_validator import EmailNotValidError from flask import g from flask import jsonify, request @@ -93,12 +94,15 @@ def new_custom_alias_v2(): 400, ) - alias = Alias.create( - user_id=user.id, - email=full_alias, - mailbox_id=user.default_mailbox_id, - note=note, - ) + try: + alias = Alias.create( + user_id=user.id, + email=full_alias, + mailbox_id=user.default_mailbox_id, + note=note, + ) + except EmailNotValidError: + return jsonify(error="Email is not valid"), 400 Session.commit() @@ -154,8 +158,16 @@ def new_custom_alias_v3(): return jsonify(error="request body does not follow the required format"), 400 alias_prefix_data = data.get("alias_prefix", "") or "" + + if not isinstance(alias_prefix_data, str): + return jsonify(error="request body does not follow the required format"), 400 + alias_prefix = alias_prefix_data.strip().lower().replace(" ", "") signed_suffix = data.get("signed_suffix", "") or "" + + if not isinstance(signed_suffix, str): + return jsonify(error="request body does not follow the required format"), 400 + signed_suffix = signed_suffix.strip() mailbox_ids = data.get("mailbox_ids") diff --git a/app/app/models.py b/app/app/models.py index 8186a7e..e5f6453 100644 --- a/app/app/models.py +++ b/app/app/models.py @@ -1659,7 +1659,7 @@ class Alias(Base, ModelMixin): return False @staticmethod - def get_custom_domain(alias_address) -> Optional["CustomDomain"]: + def get_custom_domain(alias_address: str) -> Optional["CustomDomain"]: alias_domain = validate_email( alias_address, check_deliverability=False, allow_smtputf8=False ).domain diff --git a/app/crontab.yml b/app/crontab.yml index 416e13b..69fc9d9 100644 --- a/app/crontab.yml +++ b/app/crontab.yml @@ -27,7 +27,7 @@ jobs: - name: SimpleLogin HIBP check command: python /code/cron.py -j check_hibp shell: /bin/bash - schedule: "*/5 * * * *" + schedule: "13 */4 * * *" captureStderr: true concurrencyPolicy: Forbid onFailure: diff --git a/app/server.py b/app/server.py index 6018b7b..e707b03 100644 --- a/app/server.py +++ b/app/server.py @@ -44,6 +44,7 @@ from app.admin_model import ( MetricAdmin, InvalidMailboxDomainAdmin, EmailSearchAdmin, + CustomDomainSearchAdmin, ) from app.api.base import api_bp from app.auth.base import auth_bp @@ -443,6 +444,11 @@ def init_admin(app): admin.init_app(app, index_view=SLAdminIndexView()) admin.add_view(EmailSearchAdmin(name="Email Search", endpoint="email_search")) + admin.add_view( + CustomDomainSearchAdmin( + name="Custom domain search", endpoint="custom_domain_search" + ) + ) admin.add_view(UserAdmin(User, Session)) admin.add_view(AliasAdmin(Alias, Session)) admin.add_view(MailboxAdmin(Mailbox, Session)) diff --git a/app/templates/admin/custom_domain_search.html b/app/templates/admin/custom_domain_search.html new file mode 100644 index 0000000..0ab0665 --- /dev/null +++ b/app/templates/admin/custom_domain_search.html @@ -0,0 +1,118 @@ +{% extends 'admin/master.html' %} + +{% macro show_user(user) -%} +

User {{ user.email }} with ID {{ user.id }}.

+ + + + + + + + + + + + + + + + {% if user.activated %} + + + {% else %} + + {% endif %} + {% if user.disabled %} + + + {% else %} + + {% endif %} + + + + +
User IDEmailVerifiedStatusPaidPremium
{{ user.id }} + {{ user.email }} + ActivatedPendingDisabledEnabled{{ "yes" if user.is_paid() else "No" }}{{ "yes" if user.is_premium() else "No" }}
+{%- endmacro %} + + +{% macro show_verification(title, expected, errors) -%} + {% if not expected %} +

{{ title }} Verified

+ {% else %} +

{{ title }}

+

Expected

+

{{expected}}

+

Current response

+ + {% endif %} +{%- endmacro %} + + +{% macro show_domain(domain_with_data) -%} +

Domain {{ domain_with_data.domain.domain }}

+ {% set domain = domain_with_data.domain %} + + +{%- endmacro %} + + +{% block body %} + +
+
+
+ + +
+ +
+
+ {% if data.no_match and query %} + + {% endif %} + {% if data.user %} +
+

Found User {{ data.user.email }}

+ {{ show_user(data.user) }} +
+ {% endif %} +
+ + {% for domain_with_data in data.domains %} +
+
+ {{ show_domain(domain_with_data) }} +
+
+ {% endfor %} +
+ + +{% endblock %}