Require sudo for account changes

app/api/views/auth.py

@@ -11,7 +11,7 @@
 from app import email_utils
 from app.api.base import api_bp
 from app.config import FLASK_SECRET, DISABLE_REGISTRATION
-from app.dashboard.views.setting import send_reset_password_email
+from app.dashboard.views.account_setting import send_reset_password_email
 from app.db import Session
 from app.email_utils import (
     email_can_be_used_as_mailbox,

app/auth/views/change_email.py

@@ -3,6 +3,7 @@
 
 from app.auth.base import auth_bp
 from app.db import Session
+from app.log import LOG
 from app.models import EmailChange, ResetPasswordCode
 
 
@@ -22,12 +23,14 @@ def change_email():
         return render_template("auth/change_email.html")
 
     user = email_change.user
+    old_email = user.email
     user.email = email_change.new_email
 
     EmailChange.delete(email_change.id)
     ResetPasswordCode.filter_by(user_id=user.id).delete()
     Session.commit()
 
+    LOG.i(f"User {user} has changed their email from {old_email} to {user.email}")
     flash("Your new email has been updated", "success")
 
     login_user(user)

app/auth/views/forgot_password.py

@@ -3,7 +3,7 @@
 from wtforms import StringField, validators
 
 from app.auth.base import auth_bp
-from app.dashboard.views.setting import send_reset_password_email
+from app.dashboard.views.account_setting import send_reset_password_email
 from app.extensions import limiter
 from app.log import LOG
 from app.models import User

app/dashboard/__init__.py

@@ -32,6 +32,7 @@
     delete_account,
     notification,
     support,
+    account_setting,
 )
 
 __all__ = [
@@ -68,4 +69,5 @@
     "delete_account",
     "notification",
     "support",
+    "account_setting",
 ]

app/dashboard/views/account_setting.py

@@ -0,0 +1,242 @@
+import arrow
+from flask import (
+    render_template,
+    request,
+    redirect,
+    url_for,
+    flash,
+)
+from flask_login import login_required, current_user
+
+from app import email_utils
+from app.config import (
+    URL,
+    FIRST_ALIAS_DOMAIN,
+    ALIAS_RANDOM_SUFFIX_LENGTH,
+    CONNECT_WITH_PROTON,
+)
+from app.dashboard.base import dashboard_bp
+from app.dashboard.views.enter_sudo import sudo_required
+from app.dashboard.views.mailbox_detail import ChangeEmailForm
+from app.db import Session
+from app.email_utils import (
+    email_can_be_used_as_mailbox,
+    personal_email_already_used,
+)
+from app.extensions import limiter
+from app.jobs.export_user_data_job import ExportUserDataJob
+from app.log import LOG
+from app.models import (
+    BlockBehaviourEnum,
+    PlanEnum,
+    ResetPasswordCode,
+    EmailChange,
+    User,
+    Alias,
+    AliasGeneratorEnum,
+    SenderFormatEnum,
+    UnsubscribeBehaviourEnum,
+)
+from app.proton.utils import perform_proton_account_unlink
+from app.utils import (
+    random_string,
+    CSRFValidationForm,
+    canonicalize_email,
+)
+
+
+@dashboard_bp.route("/account_setting", methods=["GET", "POST"])
+@login_required
+@sudo_required
+@limiter.limit("5/minute", methods=["POST"])
+def account_setting():
+    change_email_form = ChangeEmailForm()
+    csrf_form = CSRFValidationForm()
+
+    email_change = EmailChange.get_by(user_id=current_user.id)
+    if email_change:
+        pending_email = email_change.new_email
+    else:
+        pending_email = None
+
+    if request.method == "POST":
+        if not csrf_form.validate():
+            flash("Invalid request", "warning")
+            return redirect(url_for("dashboard.setting"))
+        if request.form.get("form-name") == "update-email":
+            if change_email_form.validate():
+                # whether user can proceed with the email update
+                new_email_valid = True
+                new_email = canonicalize_email(change_email_form.email.data)
+                if new_email != current_user.email and not pending_email:
+                    # check if this email is not already used
+                    if personal_email_already_used(new_email) or Alias.get_by(
+                        email=new_email
+                    ):
+                        flash(f"Email {new_email} already used", "error")
+                        new_email_valid = False
+                    elif not email_can_be_used_as_mailbox(new_email):
+                        flash(
+                            "You cannot use this email address as your personal inbox.",
+                            "error",
+                        )
+                        new_email_valid = False
+                    # a pending email change with the same email exists from another user
+                    elif EmailChange.get_by(new_email=new_email):
+                        other_email_change: EmailChange = EmailChange.get_by(
+                            new_email=new_email
+                        )
+                        LOG.w(
+                            "Another user has a pending %s with the same email address. Current user:%s",
+                            other_email_change,
+                            current_user,
+                        )
+
+                        if other_email_change.is_expired():
+                            LOG.d(
+                                "delete the expired email change %s", other_email_change
+                            )
+                            EmailChange.delete(other_email_change.id)
+                            Session.commit()
+                        else:
+                            flash(
+                                "You cannot use this email address as your personal inbox.",
+                                "error",
+                            )
+                            new_email_valid = False
+
+                    if new_email_valid:
+                        email_change = EmailChange.create(
+                            user_id=current_user.id,
+                            code=random_string(
+                                60
+                            ),  # todo: make sure the code is unique
+                            new_email=new_email,
+                        )
+                        Session.commit()
+                        send_change_email_confirmation(current_user, email_change)
+                        flash(
+                            "A confirmation email is on the way, please check your inbox",
+                            "success",
+                        )
+                        return redirect(url_for("dashboard.account_setting"))
+        elif request.form.get("form-name") == "change-password":
+            flash(
+                "You are going to receive an email containing instructions to change your password",
+                "success",
+            )
+            send_reset_password_email(current_user)
+            return redirect(url_for("dashboard.account_setting"))
+        elif request.form.get("form-name") == "send-full-user-report":
+            if ExportUserDataJob(current_user).store_job_in_db():
+                flash(
+                    "You will receive your SimpleLogin data via email shortly",
+                    "success",
+                )
+            else:
+                flash("An export of your data is currently in progress", "error")
+
+    partner_sub = None
+    partner_name = None
+
+    return render_template(
+        "dashboard/account_setting.html",
+        csrf_form=csrf_form,
+        PlanEnum=PlanEnum,
+        SenderFormatEnum=SenderFormatEnum,
+        BlockBehaviourEnum=BlockBehaviourEnum,
+        change_email_form=change_email_form,
+        pending_email=pending_email,
+        AliasGeneratorEnum=AliasGeneratorEnum,
+        UnsubscribeBehaviourEnum=UnsubscribeBehaviourEnum,
+        partner_sub=partner_sub,
+        partner_name=partner_name,
+        FIRST_ALIAS_DOMAIN=FIRST_ALIAS_DOMAIN,
+        ALIAS_RAND_SUFFIX_LENGTH=ALIAS_RANDOM_SUFFIX_LENGTH,
+        connect_with_proton=CONNECT_WITH_PROTON,
+    )
+
+
+def send_reset_password_email(user):
+    """
+    generate a new ResetPasswordCode and send it over email to user
+    """
+    # the activation code is valid for 1h
+    reset_password_code = ResetPasswordCode.create(
+        user_id=user.id, code=random_string(60)
+    )
+    Session.commit()
+
+    reset_password_link = f"{URL}/auth/reset_password?code={reset_password_code.code}"
+
+    email_utils.send_reset_password_email(user.email, reset_password_link)
+
+
+def send_change_email_confirmation(user: User, email_change: EmailChange):
+    """
+    send confirmation email to the new email address
+    """
+
+    link = f"{URL}/auth/change_email?code={email_change.code}"
+
+    email_utils.send_change_email(email_change.new_email, user.email, link)
+
+
+@dashboard_bp.route("/resend_email_change", methods=["GET", "POST"])
+@limiter.limit("5/hour")
+@login_required
+@sudo_required
+def resend_email_change():
+    form = CSRFValidationForm()
+    if not form.validate():
+        flash("Invalid request. Please try again", "warning")
+        return redirect(url_for("dashboard.setting"))
+    email_change = EmailChange.get_by(user_id=current_user.id)
+    if email_change:
+        # extend email change expiration
+        email_change.expired = arrow.now().shift(hours=12)
+        Session.commit()
+
+        send_change_email_confirmation(current_user, email_change)
+        flash("A confirmation email is on the way, please check your inbox", "success")
+        return redirect(url_for("dashboard.setting"))
+    else:
+        flash(
+            "You have no pending email change. Redirect back to Setting page", "warning"
+        )
+        return redirect(url_for("dashboard.setting"))
+
+
+@dashboard_bp.route("/cancel_email_change", methods=["GET", "POST"])
+@login_required
+@sudo_required
+def cancel_email_change():
+    form = CSRFValidationForm()
+    if not form.validate():
+        flash("Invalid request. Please try again", "warning")
+        return redirect(url_for("dashboard.setting"))
+    email_change = EmailChange.get_by(user_id=current_user.id)
+    if email_change:
+        EmailChange.delete(email_change.id)
+        Session.commit()
+        flash("Your email change is cancelled", "success")
+        return redirect(url_for("dashboard.setting"))
+    else:
+        flash(
+            "You have no pending email change. Redirect back to Setting page", "warning"
+        )
+        return redirect(url_for("dashboard.setting"))
+
+
+@dashboard_bp.route("/unlink_proton_account", methods=["POST"])
+@login_required
+@sudo_required
+def unlink_proton_account():
+    csrf_form = CSRFValidationForm()
+    if not csrf_form.validate():
+        flash("Invalid request", "warning")
+        return redirect(url_for("dashboard.setting"))
+
+    perform_proton_account_unlink(current_user)
+    flash("Your Proton account has been unlinked", "success")
+    return redirect(url_for("dashboard.setting"))

app/dashboard/views/alias_export.py

@@ -1,9 +1,11 @@
 from app.dashboard.base import dashboard_bp
 from flask_login import login_required, current_user
 from app.alias_utils import alias_export_csv
+from app.dashboard.views.enter_sudo import sudo_required
 
 
 @dashboard_bp.route("/alias_export", methods=["GET"])
 @login_required
+@sudo_required
 def alias_export_route():
     return alias_export_csv(current_user)

app/dashboard/views/batch_import.py

@@ -5,6 +5,7 @@
 from app import s3
 from app.config import JOB_BATCH_IMPORT
 from app.dashboard.base import dashboard_bp
+from app.dashboard.views.enter_sudo import sudo_required
 from app.db import Session
 from app.log import LOG
 from app.models import File, BatchImport, Job
@@ -13,6 +14,7 @@
 
 @dashboard_bp.route("/batch_import", methods=["GET", "POST"])
 @login_required
+@sudo_required
 def batch_import_route():
     # only for users who have custom domains
     if not current_user.verified_custom_domains():