4.56.3

app/SECURITY.md

@@ -7,8 +7,4 @@ If you want be up to date on security patches, make sure your SimpleLogin image
 
 ## Reporting a Vulnerability
 
-If you've found a security vulnerability, you can disclose it responsibly by sending a summary to security@simplelogin.io.
-We will review the potential threat and fix it as fast as we can.
-
-We are incredibly thankful for people who disclose vulnerabilities, unfortunately we do not have a bounty program in place yet.
-
+If you want to report a vulnerability, please take a look at our bug bounty program at https://proton.me/security/bug-bounty.

app/app/api/views/alias.py

@@ -419,9 +419,8 @@ def create_contact_route(alias_id):
     if not data:
         return jsonify(error="request body cannot be empty"), 400
 
-    alias: Alias = Alias.get(alias_id)
-
-    if alias.user_id != g.user.id:
+    alias: Optional[Alias] = Alias.get_by(id=alias_id, user_id=g.user.id)
+    if not alias:
         return jsonify(error="Forbidden"), 403
 
     contact_address = data.get("contact")

app/app/auth/views/login.py

@@ -10,6 +10,7 @@ from app.events.auth_event import LoginEvent
 from app.extensions import limiter
 from app.log import LOG
 from app.models import User
+from app.pw_models import PasswordOracle
 from app.utils import sanitize_email, sanitize_next_url, canonicalize_email
 
 
@@ -43,6 +44,13 @@ def login():
         user = User.get_by(email=email) or User.get_by(email=canonical_email)
 
         if not user or not user.check_password(form.password.data):
+            if not user:
+                # Do the hash to avoid timing attacks nevertheless
+                dummy_pw = PasswordOracle()
+                dummy_pw.password = (
+                    "$2b$12$ZWqpL73h4rGNfLkJohAFAu0isqSw/bX9p/tzpbWRz/To5FAftaW8u"
+                )
+                dummy_pw.check_password(form.password.data)
             # Trigger rate limiter
             g.deduct_limit = True
             form.password.data = None

app/app/config.py

@@ -309,6 +309,7 @@ JOB_DELETE_DOMAIN = "delete-domain"
 JOB_SEND_USER_REPORT = "send-user-report"
 JOB_SEND_PROTON_WELCOME_1 = "proton-welcome-1"
 JOB_SEND_ALIAS_CREATION_EVENTS = "send-alias-creation-events"
+JOB_SEND_EVENT_TO_WEBHOOK = "send-event-to-webhook"
 
 # for pagination
 PAGE_LIMIT = 20

app/app/contact_utils.py

@@ -16,6 +16,7 @@ from app.utils import sanitize_email
 class ContactCreateError(Enum):
     InvalidEmail = "Invalid email"
     NotAllowed = "Your plan does not allow to create contacts"
+    Unknown = "Unknown error when trying to create contact"
 
 
 @dataclass
@@ -87,8 +88,10 @@ def create_contact(
         return __update_contact_if_needed(contact, name, mail_from)
     # Create the contact
     reply_email = generate_reply_email(email, alias)
+    alias_id = alias.id
     try:
         flags = Contact.FLAG_PARTNER_CREATED if from_partner else 0
+        is_invalid_email = email == ""
         contact = Contact.create(
             user_id=alias.user_id,
             alias_id=alias.id,
@@ -98,9 +101,10 @@ def create_contact(
             mail_from=mail_from,
             automatic_created=automatic_created,
             flags=flags,
-            invalid_email=email == "",
+            invalid_email=is_invalid_email,
             commit=True,
         )
+        contact_id = contact.id
         if automatic_created:
             trail = ". Automatically created"
         else:
@@ -108,17 +112,27 @@ def create_contact(
         emit_alias_audit_log(
             alias=alias,
             action=AliasAuditLogAction.CreateContact,
-            message=f"Created contact {contact.id} ({contact.email}){trail}",
+            message=f"Created contact {contact_id} ({email}){trail}",
             commit=True,
         )
         LOG.d(
-            f"Created contact {contact} for alias {alias} with email {email} invalid_email={contact.invalid_email}"
+            f"Created contact {contact} for alias {alias} with email {email} invalid_email={is_invalid_email}"
         )
+        return ContactCreateResult(contact, created=True, error=None)
     except IntegrityError:
         Session.rollback()
         LOG.info(
-            f"Contact with email {email} for alias_id {alias.id} already existed, fetching from DB"
+            f"Contact with email {email} for alias_id {alias_id} already existed, fetching from DB"
         )
-        contact = Contact.get_by(alias_id=alias.id, website_email=email)
-        return __update_contact_if_needed(contact, name, mail_from)
-    return ContactCreateResult(contact, created=True, error=None)
+        contact: Optional[Contact] = Contact.get_by(
+            alias_id=alias_id, website_email=email
+        )
+        if contact:
+            return __update_contact_if_needed(contact, name, mail_from)
+        else:
+            LOG.warning(
+                f"Could not find contact with email {email} for alias_id {alias_id} and it should exist"
+            )
+            return ContactCreateResult(
+                None, created=False, error=ContactCreateError.Unknown
+            )

app/app/dashboard/views/account_setting.py

@@ -1,3 +1,5 @@
+import secrets
+
 import arrow
 from flask import (
     render_template,
@@ -163,7 +165,7 @@ def send_reset_password_email(user):
     """
     # the activation code is valid for 1h
     reset_password_code = ResetPasswordCode.create(
-        user_id=user.id, code=random_string(60)
+        user_id=user.id, code=secrets.token_urlsafe(32)
     )
     Session.commit()
 

app/app/dashboard/views/lifetime_licence.py

@@ -1,3 +1,4 @@
+import arrow
 from flask import render_template, flash, redirect, url_for
 from flask_login import login_required, current_user
 from flask_wtf import FlaskForm
@@ -7,6 +8,8 @@ from app.config import ADMIN_EMAIL
 from app.dashboard.base import dashboard_bp
 from app.db import Session
 from app.email_utils import send_email
+from app.events.event_dispatcher import EventDispatcher
+from app.events.generated.event_pb2 import UserPlanChanged, EventContent
 from app.models import LifetimeCoupon
 
 
@@ -40,6 +43,14 @@ def lifetime_licence():
             current_user.lifetime_coupon_id = coupon.id
             if coupon.paid:
                 current_user.paid_lifetime = True
+            EventDispatcher.send_event(
+                user=current_user,
+                content=EventContent(
+                    user_plan_change=UserPlanChanged(
+                        plan_end_time=arrow.get("2100-01-01").timestamp
+                    )
+                ),
+            )
             Session.commit()
 
             # notify admin

app/app/jobs/send_event_job.py

@@ -0,0 +1,70 @@
+from __future__ import annotations
+
+import base64
+from typing import Optional
+
+import arrow
+
+from app import config
+from app.errors import ProtonPartnerNotSetUp
+from app.events.generated import event_pb2
+from app.events.generated.event_pb2 import EventContent
+from app.models import (
+    User,
+    Job,
+    PartnerUser,
+)
+from app.proton.utils import get_proton_partner
+from events.event_sink import EventSink
+
+
+class SendEventToWebhookJob:
+    def __init__(self, user: User, event: EventContent):
+        self._user: User = user
+        self._event: EventContent = event
+
+    def run(self, sink: EventSink) -> bool:
+        # Check if the current user has a partner_id
+        try:
+            proton_partner_id = get_proton_partner().id
+        except ProtonPartnerNotSetUp:
+            return False
+
+        # It has. Retrieve the information for the PartnerUser
+        partner_user = PartnerUser.get_by(
+            user_id=self._user.id, partner_id=proton_partner_id
+        )
+        if partner_user is None:
+            return True
+        event = event_pb2.Event(
+            user_id=self._user.id,
+            external_user_id=partner_user.external_user_id,
+            partner_id=partner_user.partner_id,
+            content=self._event,
+        )
+
+        serialized = event.SerializeToString()
+        return sink.send_data_to_webhook(serialized)
+
+    @staticmethod
+    def create_from_job(job: Job) -> Optional[SendEventToWebhookJob]:
+        user = User.get(job.payload["user_id"])
+        if not user:
+            return None
+        event_data = base64.b64decode(job.payload["event"])
+        event = event_pb2.EventContent()
+        event.ParseFromString(event_data)
+
+        return SendEventToWebhookJob(user=user, event=event)
+
+    def store_job_in_db(self, run_at: Optional[arrow.Arrow]) -> Job:
+        stub = self._event.SerializeToString()
+        return Job.create(
+            name=config.JOB_SEND_EVENT_TO_WEBHOOK,
+            payload={
+                "user_id": self._user.id,
+                "event": base64.b64encode(stub).decode("utf-8"),
+            },
+            run_at=run_at if run_at is not None else arrow.now(),
+            commit=True,
+        )

app/app/mailbox_utils.py

@@ -1,6 +1,5 @@
 import dataclasses
 import secrets
-import random
 from enum import Enum
 from typing import Optional
 import arrow
@@ -37,8 +36,9 @@ class OnlyPaidError(MailboxError):
 
 
 class CannotVerifyError(MailboxError):
-    def __init__(self, msg: str):
+    def __init__(self, msg: str, deleted_activation_code: bool = False):
         self.msg = msg
+        self.deleted_activation_code = deleted_activation_code
 
 
 MAX_ACTIVATION_TRIES = 3
@@ -196,7 +196,10 @@ def verify_mailbox_code(user: User, mailbox_id: int, code: str) -> Mailbox:
     if activation.tries >= MAX_ACTIVATION_TRIES:
         LOG.i(f"User {user} failed to verify mailbox {mailbox_id} more than 3 times")
         clear_activation_codes_for_mailbox(mailbox)
-        raise CannotVerifyError("Invalid activation code. Please request another code.")
+        raise CannotVerifyError(
+            "Invalid activation code. Please request another code.",
+            deleted_activation_code=True,
+        )
     if activation.created_at < arrow.now().shift(minutes=-15):
         LOG.i(
             f"User {user} failed to verify mailbox {mailbox_id} because code is too old"
@@ -229,7 +232,7 @@ def generate_activation_code(
         if config.MAILBOX_VERIFICATION_OVERRIDE_CODE:
             code = config.MAILBOX_VERIFICATION_OVERRIDE_CODE
         else:
-            code = "{:06d}".format(random.randint(1, 999999))
+            code = "{:06d}".format(secrets.randbelow(1000000))[:6]
     else:
         code = secrets.token_urlsafe(16)
     return MailboxActivation.create(

app/app/utils.py

@@ -1,4 +1,3 @@
-import random
 import re
 import secrets
 import string
@@ -32,8 +31,9 @@ def random_words(words: int = 2, numbers: int = 0):
     fields = [secrets.choice(_words) for i in range(words)]
 
     if numbers > 0:
-        digits = "".join([str(random.randint(0, 9)) for i in range(numbers)])
-        return "_".join(fields) + digits
+        digits = [n for n in range(10)]
+        suffix = "".join([str(secrets.choice(digits)) for i in range(numbers)])
+        return "_".join(fields) + suffix
     else:
         return "_".join(fields)
 

app/cron.py

@@ -286,8 +286,16 @@ def notify_manual_sub_end():
 
 def poll_apple_subscription():
     """Poll Apple API to update AppleSubscription"""
-    # todo: only near the end of the subscription
-    for apple_sub in AppleSubscription.all():
+    for apple_sub in (
+        AppleSubscription.filter(
+            AppleSubscription.expires_date < arrow.now().shift(days=15)
+        )
+        .enable_eagerloads(False)
+        .yield_per(100)
+    ):
+        if not apple_sub.is_valid():
+            # Subscription is not valid anymore and hasn't been renewed
+            continue
         if not apple_sub.product_id:
             LOG.d("Ignore %s", apple_sub)
             continue
@@ -900,6 +908,24 @@ def check_mailbox_valid_pgp_keys():
 
 
 def check_custom_domain():
+    # Delete custom domains that haven't been verified in a month
+    for custom_domain in (
+        CustomDomain.filter(
+            CustomDomain.verified == False,  # noqa: E712
+            CustomDomain.created_at < arrow.now().shift(months=-1),
+        )
+        .enable_eagerloads(False)
+        .yield_per(100)
+    ):
+        alias_count = Alias.filter(Alias.custom_domain_id == custom_domain.id).count()
+        if alias_count > 0:
+            LOG.warn(
+                f"Custom Domain {custom_domain} has {alias_count} aliases. Won't delete"
+            )
+        else:
+            LOG.i(f"Deleting unverified old custom domain {custom_domain}")
+            CustomDomain.delete(custom_domain.id)
+
     LOG.d("Check verified domain for DNS issues")
 
     for custom_domain in CustomDomain.filter_by(verified=True):  # type: CustomDomain
@@ -971,7 +997,7 @@ def delete_expired_tokens():
     LOG.d("Delete api to cookie tokens older than %s, nb row %s", max_time, nb_row)
 
 
-async def _hibp_check(api_key, queue):
+async def _hibp_check(api_key: str, queue: asyncio.Queue):
     """
     Uses a single API key to check the queue as fast as possible.
 
@@ -990,11 +1016,16 @@ async def _hibp_check(api_key, queue):
         if not alias:
             continue
         user = alias.user
-        if user.disabled or not user.is_paid():
+        if user.disabled or not user.is_premium():
             # Mark it as hibp done to skip it as if it had been checked
             alias.hibp_last_check = arrow.utcnow()
             Session.commit()
             continue
+        if alias.flags & Alias.FLAG_PARTNER_CREATED > 0:
+            # Mark as hibp done
+            alias.hibp_last_check = arrow.utcnow()
+            Session.commit()
+            continue
 
         LOG.d("Checking HIBP for %s", alias)
 

app/crontab.yml

@@ -16,13 +16,25 @@ jobs:
     shell: /bin/bash
     schedule: "15 2 * * *"
     captureStderr: true
+    onFailure:
+      retry:
+        maximumRetries: 10
+        initialDelay: 1
+        maximumDelay: 30
+        backoffMultiplier: 2
 
   - name: SimpleLogin HIBP check
     command: python /code/cron.py -j check_hibp
     shell: /bin/bash
-    schedule: "15 3 * * *"
+    schedule: "16 */4 * * *"
     captureStderr: true
     concurrencyPolicy: Forbid
+    onFailure:
+      retry:
+        maximumRetries: 10
+        initialDelay: 1
+        maximumDelay: 30
+        backoffMultiplier: 2
 
   - name: SimpleLogin Notify HIBP breaches
     command: python /code/cron.py -j notify_hibp
@@ -31,6 +43,7 @@ jobs:
     captureStderr: true
     concurrencyPolicy: Forbid
 
+
   - name: SimpleLogin Delete Logs
     command: python /code/cron.py -j delete_logs
     shell: /bin/bash

app/email_handler.py

@@ -177,7 +177,9 @@ from init_app import load_pgp_public_keys
 from server import create_light_app
 
 
-def get_or_create_contact(from_header: str, mail_from: str, alias: Alias) -> Contact:
+def get_or_create_contact(
+    from_header: str, mail_from: str, alias: Alias
+) -> Optional[Contact]:
     """
     contact_from_header is the RFC 2047 format FROM header
     """
@@ -208,6 +210,8 @@ def get_or_create_contact(from_header: str, mail_from: str, alias: Alias) -> Con
         automatic_created=True,
         from_partner=False,
     )
+    if contact_result.error:
+        LOG.w(f"Error creating contact: {contact_result.error.value}")
     return contact_result.contact
 
 
@@ -558,7 +562,7 @@ def handle_forward(envelope, msg: Message, rcpt_to: str) -> List[Tuple[bool, str
 
     if not user.is_active():
         LOG.w(f"User {user} has been soft deleted")
-        return False, status.E502
+        return [(False, status.E502)]
 
     if not user.can_send_or_receive():
         LOG.i(f"User {user} cannot receive emails")
@@ -579,6 +583,8 @@ def handle_forward(envelope, msg: Message, rcpt_to: str) -> List[Tuple[bool, str
     from_header = get_header_unicode(msg[headers.FROM])
     LOG.d("Create or get contact for from_header:%s", from_header)
     contact = get_or_create_contact(from_header, envelope.mail_from, alias)
+    if not contact:
+        return [(False, status.E504)]
     alias = (
         contact.alias
     )  # In case the Session was closed in the get_or_create we re-fetch the alias

app/events/event_sink.py

@@ -12,6 +12,10 @@ class EventSink(ABC):
     def process(self, event: SyncEvent) -> bool:
         pass
 
+    @abstractmethod
+    def send_data_to_webhook(self, data: bytes) -> bool:
+        pass
+
 
 class HttpEventSink(EventSink):
     def process(self, event: SyncEvent) -> bool:
@@ -21,9 +25,16 @@ class HttpEventSink(EventSink):
 
         LOG.info(f"Sending event {event.id} to {EVENT_WEBHOOK}")
 
+        if self.send_data_to_webhook(event.content):
+            LOG.info(f"Event {event.id} sent successfully to webhook")
+            return True
+
+        return False
+
+    def send_data_to_webhook(self, data: bytes) -> bool:
         res = requests.post(
             url=EVENT_WEBHOOK,
-            data=event.content,
+            data=data,
             headers={"Content-Type": "application/x-protobuf"},
             verify=not EVENT_WEBHOOK_SKIP_VERIFY_SSL,
         )
@@ -36,7 +47,6 @@ class HttpEventSink(EventSink):
             )
             return False
         else:
-            LOG.info(f"Event {event.id} sent successfully to webhook")
             return True
 
 
@@ -44,3 +54,7 @@ class ConsoleEventSink(EventSink):
     def process(self, event: SyncEvent) -> bool:
         LOG.info(f"Handling event {event.id}")
         return True
+
+    def send_data_to_webhook(self, data: bytes) -> bool:
+        LOG.info(f"Sending {len(data)} bytes to webhook")
+        return True

app/job_runner.py

@@ -18,6 +18,7 @@ from app.events.event_dispatcher import PostgresDispatcher
 from app.import_utils import handle_batch_import
 from app.jobs.event_jobs import send_alias_creation_events_for_user
 from app.jobs.export_user_data_job import ExportUserDataJob
+from app.jobs.send_event_job import SendEventToWebhookJob
 from app.log import LOG
 from app.models import User, Job, BatchImport, Mailbox, CustomDomain, JobState
 from app.user_audit_log_utils import emit_user_audit_log, UserAuditLogAction
@@ -300,6 +301,10 @@ def process_job(job: Job):
             send_alias_creation_events_for_user(
                 user, dispatcher=PostgresDispatcher.get()
             )
+    elif job.name == config.JOB_SEND_EVENT_TO_WEBHOOK:
+        send_job = SendEventToWebhookJob.create_from_job(job)
+        if send_job:
+            send_job.run()
     else:
         LOG.e("Unknown job name %s", job.name)
 

app/oneshot/send_lifetime_user_events.py

@@ -0,0 +1,60 @@
+#!/usr/bin/env python3
+import argparse
+import time
+
+import arrow
+from sqlalchemy import func
+
+from app.events.event_dispatcher import EventDispatcher
+from app.events.generated.event_pb2 import UserPlanChanged, EventContent
+from app.models import PartnerUser
+from app.db import Session
+
+parser = argparse.ArgumentParser(
+    prog="Backfill alias", description="Send lifetime users to proton"
+)
+parser.add_argument(
+    "-s", "--start_pu_id", default=0, type=int, help="Initial partner_user_id"
+)
+parser.add_argument(
+    "-e", "--end_pu_id", default=0, type=int, help="Last partner_user_id"
+)
+
+args = parser.parse_args()
+pu_id_start = args.start_pu_id
+max_pu_id = args.end_pu_id
+if max_pu_id == 0:
+    max_pu_id = Session.query(func.max(PartnerUser.id)).scalar()
+
+print(f"Checking partner user {pu_id_start} to {max_pu_id}")
+step = 100
+done = 0
+start_time = time.time()
+with_lifetime = 0
+for batch_start in range(pu_id_start, max_pu_id, step):
+    partner_users = (
+        Session.query(PartnerUser).filter(
+            PartnerUser.id >= batch_start, PartnerUser.id < batch_start + step
+        )
+    ).all()
+    for partner_user in partner_users:
+        done += 1
+        if not partner_user.user.lifetime:
+            continue
+        with_lifetime += 1
+        event = UserPlanChanged(plan_end_time=arrow.get("2100-01-01").timestamp)
+        EventDispatcher.send_event(
+            partner_user.user, EventContent(user_plan_change=event)
+        )
+        Session.flush()
+    Session.commit()
+    elapsed = time.time() - start_time
+    last_batch_id = batch_start + step
+    time_per_alias = elapsed / (last_batch_id)
+    remaining = max_pu_id - last_batch_id
+    time_remaining = remaining / time_per_alias
+    hours_remaining = time_remaining / 60.0
+    print(
+        f"\PartnerUser {batch_start}/{max_pu_id} {done} {hours_remaining:.2f} mins remaining"
+    )
+print(f"With SL lifetime {with_lifetime}")

app/oneshot/send_plan_change_events.py

@@ -2,6 +2,7 @@
 import argparse
 import time
 
+import arrow
 from sqlalchemy import func
 
 from app.events.event_dispatcher import EventDispatcher
@@ -30,6 +31,7 @@ step = 100
 updated = 0
 start_time = time.time()
 with_premium = 0
+with_lifetime = 0
 for batch_start in range(pu_id_start, max_pu_id, step):
     partner_users = (
         Session.query(PartnerUser).filter(
@@ -41,7 +43,10 @@ for batch_start in range(pu_id_start, max_pu_id, step):
             include_partner_subscription=False
         )
         end_timestamp = None
-        if subscription_end:
+        if partner_user.user.lifetime:
+            with_lifetime += 1
+            end_timestamp = arrow.get("2100-01-01").timestamp
+        elif subscription_end:
             with_premium += 1
             end_timestamp = subscription_end.timestamp
         event = UserPlanChanged(plan_end_time=end_timestamp)
@@ -60,4 +65,4 @@ for batch_start in range(pu_id_start, max_pu_id, step):
     print(
         f"\PartnerUser {batch_start}/{max_pu_id} {updated} {hours_remaining:.2f} mins remaining"
     )
-print(f"With SL premium {with_premium}")
+print(f"With SL premium {with_premium} lifetime {with_lifetime}")

app/templates/admin/email_search.html

@@ -11,6 +11,7 @@
             <th scope="col">Verified</th>
             <th scope="col">Status</th>
             <th scope="col">Paid</th>
+            <th scope="col">Premium</th>
             <th>Subscription</th>
             <th>Created At</th>
             <th>Updated At</th>
@@ -32,6 +33,7 @@
                 <td class="text-success">Enabled</td>
             {% endif %}
             <td>{{ "yes" if user.is_paid() else "No" }}</td>
+            <td>{{ "yes" if user.is_premium() else "No" }}</td>
             <td>{{ user.get_active_subscription() }}</td>
             <td>{{ user.created_at }}</td>
             <td>{{ user.updated_at }}</td>

app/tests/api/test_alias.py

@@ -511,6 +511,19 @@ def test_create_contact_route_invalid_alias(flask_client):
     assert r.status_code == 403
 
 
+def test_create_contact_route_non_existing_alias(flask_client):
+    user, api_key = get_new_user_and_api_key()
+    Session.commit()
+
+    r = flask_client.post(
+        url_for("api.create_contact_route", alias_id=99999999),
+        headers={"Authentication": api_key.code},
+        json={"contact": "First Last <first@example.com>"},
+    )
+
+    assert r.status_code == 403
+
+
 def test_create_contact_route_free_users(flask_client):
     user, api_key = get_new_user_and_api_key()
 

app/tests/jobs/test_send_event_to_webhook.py

@@ -0,0 +1,40 @@
+import arrow
+
+from app import config
+from app.events.generated.event_pb2 import EventContent, AliasDeleted
+from app.jobs.send_event_job import SendEventToWebhookJob
+from app.models import PartnerUser
+from app.proton.utils import get_proton_partner
+from events.event_sink import ConsoleEventSink
+from tests.utils import create_new_user, random_token
+
+
+def test_serialize_and_deserialize_job():
+    user = create_new_user()
+    alias_id = 34
+    alias_email = "a@b.c"
+    event = EventContent(alias_deleted=AliasDeleted(id=alias_id, email=alias_email))
+    run_at = arrow.now().shift(hours=10)
+    db_job = SendEventToWebhookJob(user, event).store_job_in_db(run_at=run_at)
+    assert db_job.run_at == run_at
+    assert db_job.name == config.JOB_SEND_EVENT_TO_WEBHOOK
+    job = SendEventToWebhookJob.create_from_job(db_job)
+    assert job._user.id == user.id
+    assert job._event.alias_deleted.id == alias_id
+    assert job._event.alias_deleted.email == alias_email
+
+
+def test_send_event_to_webhook():
+    user = create_new_user()
+    PartnerUser.create(
+        user_id=user.id,
+        partner_id=get_proton_partner().id,
+        external_user_id=random_token(10),
+        flush=True,
+    )
+    alias_id = 34
+    alias_email = "a@b.c"
+    event = EventContent(alias_deleted=AliasDeleted(id=alias_id, email=alias_email))
+    job = SendEventToWebhookJob(user, event)
+    sink = ConsoleEventSink()
+    assert job.run(sink)

app/tests/test_mailbox_utils.py

@@ -314,10 +314,13 @@ def test_verify_too_may():
     output = mailbox_utils.create_mailbox(user, random_email())
     output.activation.tries = mailbox_utils.MAX_ACTIVATION_TRIES
     Session.commit()
-    with pytest.raises(mailbox_utils.CannotVerifyError):
+    try:
         mailbox_utils.verify_mailbox_code(
             user, output.mailbox.id, output.activation.code
         )
+        assert False
+    except mailbox_utils.CannotVerifyError as e:
+        assert e.deleted_activation_code
 
 
 @mail_sender.store_emails_test_decorator