feat: admin can delete a user's mailbox from email search

app/admin/email_search.py

@@ -600,6 +600,10 @@ def action_name(action_id: int) -> str:
             14: "clear_delete_on",
             15: "update_subdomain_quota",
             16: "update_directory_quota",
+            17: "disable_mailbox",
+            18: "enable_mailbox",
+            19: "change_default_mailbox",
+            20: "delete_mailbox",
         }
         return action_names.get(action_id, f"unknown_{action_id}")
 
@@ -1423,6 +1427,44 @@ def admin_disable_mailbox(self):
             )
         )
 
+    @expose("/delete_mailbox", methods=["POST"])
+    def delete_mailbox(self):
+        from app.mailbox_utils import MailboxError, admin_delete_mailbox
+
+        try:
+            mailbox_id = int(request.form.get("mailbox_id", ""))
+        except ValueError:
+            flash("Invalid mailbox_id", "error")
+            return redirect(url_for("admin.email_search.index"))
+
+        mailbox = Mailbox.get(mailbox_id)
+        if not mailbox:
+            flash("Mailbox not found", "error")
+            return redirect(url_for("admin.email_search.index"))
+
+        back = redirect(
+            url_for(
+                "admin.email_search.index", query=mailbox.email, search_type="email"
+            )
+        )
+        if request.form.get("confirm_email", "").strip() != mailbox.email:
+            flash("Confirmation email does not match the mailbox email", "error")
+            return back
+
+        email = mailbox.email
+        try:
+            target = admin_delete_mailbox(mailbox, User.get(current_user.id))
+        except MailboxError as e:
+            flash(f"Cannot delete mailbox {email}: {e}", "error")
+            return back
+
+        moved = f"aliases move to {target.email}" if target else "no aliases to move"
+        flash(f"Mailbox {email} scheduled for deletion, {moved}", "success")
+        LOG.warning(
+            f"Admin {current_user.email} scheduled deletion of mailbox {email} (id={mailbox_id}) transfer to {target.id if target else None}"
+        )
+        return back
+
     @expose("/admin_reenable_mailbox", methods=["POST"])
     def admin_reenable_mailbox(self):
         from app.mailbox_utils import admin_reenable_mailbox

app/mailbox_utils.py

@@ -8,6 +8,7 @@
 
 import arrow
 from aiosmtpd.smtp import Envelope
+import sqlalchemy
 from sqlalchemy.exc import IntegrityError
 
 from app import config, s3
@@ -34,6 +35,7 @@
     User,
     Mailbox,
     Job,
+    JobState,
     MailboxActivation,
     Alias,
     AliasMailbox,
@@ -580,15 +582,116 @@ def _reset_default_mailbox_if_needed(disabled_mailbox: Mailbox) -> None:
     user = disabled_mailbox.user
     if user.default_mailbox_id != disabled_mailbox.id:
         return
+    replacement = _find_replacement_default(user, {disabled_mailbox.id})
+    if replacement:
+        user.default_mailbox_id = replacement.id
+        return
+    LOG.w(f"User {user} has no usable mailbox to replace disabled default mailbox")
+
+
+def _find_replacement_default(user: User, excluded_ids: set[int]) -> Optional[Mailbox]:
+    """Oldest verified mailbox of the user that is not disabled nor excluded."""
     for candidate in (
         Mailbox.filter_by(user_id=user.id, verified=True, disabled=False)
         .order_by(Mailbox.id.asc())
         .all()
     ):
-        if candidate.id != disabled_mailbox.id and not candidate.is_admin_disabled():
-            user.default_mailbox_id = candidate.id
-            return
-    LOG.w(f"User {user} has no usable mailbox to replace disabled default mailbox")
+        if candidate.id not in excluded_ids and not candidate.is_admin_disabled():
+            return candidate
+    return None
+
+
+def _pending_mailbox_deletions(user: User) -> Tuple[set[int], set[int]]:
+    """Return (mailbox ids awaiting deletion, ids of their transfer targets)."""
+    mailbox_ids = [m.id for m in Mailbox.filter_by(user_id=user.id).all()]
+    jobs = Session.query(Job).filter(
+        Job.name == JobType.DELETE_MAILBOX.value,
+        Job.state.in_([JobState.ready.value, JobState.taken.value]),
+        Job.payload.op("->")("mailbox_id")
+        .cast(sqlalchemy.TEXT)
+        .in_([str(i) for i in mailbox_ids]),
+    )
+    pending: set[int] = set()
+    targets: set[int] = set()
+    for job in jobs:
+        pending.add(job.payload["mailbox_id"])
+        if job.payload.get("transfer_mailbox_id"):
+            targets.add(job.payload["transfer_mailbox_id"])
+    return pending, targets
+
+
+def admin_delete_mailbox(mailbox: Mailbox, admin_user: User) -> Optional[Mailbox]:
+    """Queue the deletion of a mailbox, moving its aliases to the default mailbox.
+
+    If the mailbox is the default, the oldest usable mailbox becomes the default
+    first. Mailboxes already awaiting deletion are never chosen as the target.
+    Returns the mailbox receiving the aliases, or None when the default mailbox
+    has no alias and no other usable mailbox exists (the user is left without
+    a default mailbox).
+    """
+    user = mailbox.user
+    pending, targets = _pending_mailbox_deletions(user)
+    if mailbox.id in pending:
+        raise MailboxError("Mailbox is already scheduled for deletion")
+    if mailbox.id in targets:
+        raise MailboxError(
+            "Mailbox is the transfer target of a pending deletion, retry once it completes"
+        )
+
+    old_default_id = user.default_mailbox_id
+    default_changes = mailbox.id == old_default_id
+    if default_changes:
+        target = _find_replacement_default(user, {mailbox.id} | pending)
+        if not target and mailbox.aliases:
+            raise MailboxError("No other usable mailbox to become the default")
+    else:
+        target = user.default_mailbox
+        if target is None:
+            if mailbox.aliases:
+                raise MailboxError("User has no default mailbox to receive the aliases")
+        elif target.id in pending:
+            raise MailboxError("Default mailbox is scheduled for deletion")
+    if target:
+        if not target.verified:
+            raise MailboxError("Your new mailbox is not verified")
+        ensure_mailbox_assignable(target)
+    target_id = target.id if target else None
+
+    if default_changes:
+        user.default_mailbox_id = target_id
+        emit_user_audit_log(
+            user=user,
+            action=UserAuditLogAction.UpdateMailbox,
+            message=f"Admin changed default mailbox from {mailbox.id} to {target_id} before deleting {mailbox.id}",
+        )
+        AdminAuditLog.create(
+            admin_user_id=admin_user.id,
+            model=Mailbox.__class__.__name__,
+            model_id=user.id,
+            action=AuditLogActionEnum.change_default_mailbox.value,
+            data={
+                "default_mailbox_id": target_id,
+                "old_default_mailbox_id": old_default_id,
+                "message": f"Changed default mailbox from {mailbox.email} (id={old_default_id}) to {target.email if target else None} (id={target_id}) before deleting it",
+            },
+        )
+    AdminAuditLog.create(
+        admin_user_id=admin_user.id,
+        model=Mailbox.__class__.__name__,
+        model_id=user.id,
+        action=AuditLogActionEnum.delete_mailbox.value,
+        data={
+            "mailbox_id": mailbox.id,
+            "email": mailbox.email,
+            "transfer_mailbox_id": target_id,
+        },
+    )
+    try:
+        delete_mailbox(user, mailbox.id, target_id, send_mail=False)
+    except MailboxError:
+        Session.rollback()
+        raise
+    return target
 
 
 def admin_disable_mailbox(

app/models.py

@@ -249,6 +249,7 @@ class AuditLogActionEnum(EnumE):
     disable_mailbox = 17
     enable_mailbox = 18
     change_default_mailbox = 19
+    delete_mailbox = 20
 
 
 class Phase(EnumE):

templates/admin/email_search.html

@@ -1254,6 +1254,12 @@ <h6 class="mb-0">
                         <i class="fa fa-ban mr-1"></i>Disable
                       </button>
                     {% endif %}
+                    <button type="button"
+                            class="btn btn-sm btn-outline-danger btn-block mt-1"
+                            data-toggle="modal"
+                            data-target="#deleteMailboxModal{{ mailbox.id }}">
+                      <i class="fa fa-trash mr-1"></i>Delete
+                    </button>
                   </div>
                 </td>
               </tr>
@@ -1459,6 +1465,62 @@ <h5 class="modal-title" id="disableMailboxModalLabel{{ mailbox.id }}">
                   </div>
                 </div>
               {% endif %}
+              {# Delete Mailbox Modal #}
+                <div class="modal fade"
+                     id="deleteMailboxModal{{ mailbox.id }}"
+                     tabindex="-1"
+                     role="dialog"
+                     aria-labelledby="deleteMailboxModalLabel{{ mailbox.id }}"
+                     aria-hidden="true">
+                  <div class="modal-dialog" role="document">
+                    <div class="modal-content">
+                      <div class="modal-header modal-danger">
+                        <h5 class="modal-title" id="deleteMailboxModalLabel{{ mailbox.id }}">
+                          <i class="fa fa-trash mr-2"></i>Delete Mailbox
+                        </h5>
+                        <button type="button"
+                                class="close text-white"
+                                data-dismiss="modal"
+                                aria-label="Close">
+                          <span aria-hidden="true">&times;</span>
+                        </button>
+                      </div>
+                      <form action="{{ url_for('admin.email_search.delete_mailbox') }}"
+                            method="POST">
+                        {{ csrf_form.csrf_token }}
+                        <div class="modal-body">
+                          <input type="hidden" name="mailbox_id" value="{{ mailbox.id }}">
+                          <div class="alert alert-danger">
+                            <i class="fa fa-exclamation-circle mr-1"></i>
+                            This will delete the mailbox <strong>{{ mailbox.email }}</strong>.
+                          </div>
+                          <p class="text-muted small">
+                            Its aliases move to the user's default mailbox.
+                            {% if mailbox.user.default_mailbox_id == mailbox.id %}
+                              This is the default mailbox: the oldest other usable mailbox becomes the new default and receives the aliases.
+                            {% endif %}
+                          </p>
+                          <div class="form-group mb-0">
+                            <label for="deleteMailboxConfirm{{ mailbox.id }}" class="font-weight-bold">Type the mailbox email to confirm</label>
+                            <input type="text"
+                                   class="form-control"
+                                   id="deleteMailboxConfirm{{ mailbox.id }}"
+                                   name="confirm_email"
+                                   required
+                                   autocomplete="off"
+                                   placeholder="{{ mailbox.email }}">
+                          </div>
+                        </div>
+                        <div class="modal-footer">
+                          <button type="button" class="btn btn-outline-secondary" data-dismiss="modal">Cancel</button>
+                          <button type="submit" class="btn btn-danger-modern">
+                            <i class="fa fa-trash mr-1"></i>Delete Mailbox
+                          </button>
+                        </div>
+                      </form>
+                    </div>
+                  </div>
+                </div>
             {% endfor %}
           </tbody>
         </table>

tests/admin/test_email_search.py

@@ -3,6 +3,7 @@
 import arrow
 from flask import url_for
 
+from app.constants import JobType
 from app.db import Session
 from app.models import (
     User,
@@ -17,6 +18,7 @@
     AuditLogActionEnum,
     Subscription,
     PlanEnum,
+    Job,
 )
 from app.admin.email_search import EmailSearchHelpers, find_related_users
 from app.proton.proton_partner import get_proton_partner
@@ -1295,3 +1297,74 @@ def test_disable_2fa_missing_user_id(flask_client):
     )
     assert r.status_code == 200
     assert b"Missing user_id" in r.data
+
+
+def _delete_mailbox_setup(flask_client):
+    login_admin(flask_client)
+    user = create_new_user()
+    other = Mailbox.create(
+        user_id=user.id, email=random_token(10) + "@mailbox.lan", verified=True
+    )
+    Session.commit()
+    return user, other
+
+
+def test_delete_mailbox_button_and_modal_displayed(flask_client):
+    user, other = _delete_mailbox_setup(flask_client)
+    r = flask_client.get(
+        url_for("admin.email_search.index", query=other.email, search_type="email")
+    )
+    assert f"deleteMailboxModal{other.id}" in r.get_data(as_text=True)
+
+
+def test_delete_mailbox_requires_matching_confirmation(flask_client):
+    user, other = _delete_mailbox_setup(flask_client)
+    before = Session.query(Job).count()
+    flask_client.post(
+        url_for("admin.email_search.delete_mailbox"),
+        data={"mailbox_id": other.id, "confirm_email": "nope@example.com"},
+        follow_redirects=True,
+    )
+    assert Session.query(Job).count() == before
+
+
+def test_delete_mailbox_schedules_job_and_audit_log(flask_client):
+    user, other = _delete_mailbox_setup(flask_client)
+    r = flask_client.post(
+        url_for("admin.email_search.delete_mailbox"),
+        data={"mailbox_id": other.id, "confirm_email": other.email},
+        follow_redirects=True,
+    )
+    assert r.status_code == 200
+    job = (
+        Session.query(Job)
+        .filter(Job.name == JobType.DELETE_MAILBOX.value)
+        .order_by(Job.id.desc())
+        .first()
+    )
+    assert job.payload["mailbox_id"] == other.id
+    assert job.payload["transfer_mailbox_id"] == user.default_mailbox_id
+    assert (
+        Session.query(AdminAuditLog)
+        .filter_by(model_id=user.id, action=AuditLogActionEnum.delete_mailbox.value)
+        .count()
+        == 1
+    )
+
+
+def test_delete_default_mailbox_with_aliases_without_other_mailbox_refused(
+    flask_client,
+):
+    login_admin(flask_client)
+    user = create_new_user()
+    Alias.create_new_random(user)
+    Session.commit()
+    default = Mailbox.get(user.default_mailbox_id)
+    before = Session.query(Job).count()
+    flask_client.post(
+        url_for("admin.email_search.delete_mailbox"),
+        data={"mailbox_id": default.id, "confirm_email": default.email},
+        follow_redirects=True,
+    )
+    assert Session.query(Job).count() == before
+    assert user.default_mailbox_id == default.id