fix(mail): bind POP3 imports to stable maildrop identity

Release v0.1.28. Coordinated integrity review: GovOPlaN/govoplan-core#298.

README.md

@@ -6,6 +6,17 @@
 
 GovOPlaN Mail is the mail transport module. It owns reusable SMTP/IMAP/JMAP profile management, an explicitly enabled legacy POP3 import path, mail profile policy enforcement, mock mail infrastructure, and the mail WebUI package.
 
+POP3 is part of this same `mail` module, not a separately installed module.
+Its **Legacy import** navigation entry opens a dedicated Mail workflow because
+retrieving historical messages requires explicit permissions, duplicate checks,
+and retained-source evidence. It shares Mail's profiles, server definitions,
+credential ownership, migrations, and release package.
+
+Deutsch: POP3 gehört zum selben Modul **Mail**. Der eigene Navigationseintrag
+**Altbestand importieren** öffnet lediglich einen gesonderten Mail-Arbeitsablauf
+mit ausdrücklichen Berechtigungen, Dublettenprüfung und Herkunftsnachweisen;
+es gibt kein separat zu installierendes POP3-Modul.
+
 ## Ownership
 
 This repository owns:

package-lock.json

@@ -1,12 +1,12 @@
 {
   "name": "@govoplan/mail-webui",
-  "version": "0.1.27",
+  "version": "0.1.28",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "@govoplan/mail-webui",
-      "version": "0.1.27",
+      "version": "0.1.28",
       "peerDependencies": {
         "@govoplan/core-webui": "^0.1.45",
         "lucide-react": "^1.23.0",

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@govoplan/mail-webui",
-  "version": "0.1.27",
+  "version": "0.1.28",
   "private": true,
   "type": "module",
   "main": "webui/src/index.ts",

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "govoplan-mail"
-version = "0.1.27"
+version = "0.1.28"
 description = "GovOPlaN mail module with backend and WebUI integration."
 readme = "README.md"
 requires-python = ">=3.12"

src/govoplan_mail/backend/db/models.py

@@ -398,8 +398,9 @@ class MailPop3Import(Base, TimestampMixin):
             "tenant_id",
             "profile_id",
             "pop3_server_id",
+            "maildrop_identity",
             "provider_uidl",
-            name="uq_mail_pop3_imports_source_uidl",
+            name="uq_mail_pop3_imports_maildrop_uidl",
         ),
         Index(
             "ix_mail_pop3_imports_review",
@@ -422,6 +423,7 @@ class MailPop3Import(Base, TimestampMixin):
         index=True,
     )
     pop3_credential_id: Mapped[str | None] = mapped_column(String(36), nullable=True)
+    maildrop_identity: Mapped[str | None] = mapped_column(String(64), nullable=True)
     transport_revision: Mapped[str] = mapped_column(String(120), nullable=False)
     provider_uidl: Mapped[str] = mapped_column(String(500), nullable=False)
     provider_message_number: Mapped[int | None] = mapped_column(Integer, nullable=True)

src/govoplan_mail/backend/manifest.py

@@ -450,7 +450,7 @@ POP3_PROVIDER = ExternalProviderDeclaration(
 manifest = ModuleManifest(
     id="mail",
     name="Mail",
-    version="0.1.27",
+    version="0.1.28",
     required_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR),
     optional_dependencies=("campaigns", "addresses", "calendar", "postbox", "search"),
     provides_interfaces=(
@@ -675,6 +675,21 @@ manifest = ModuleManifest(
         ),
     },
     documentation=(
+        DocumentationTopic(
+            id="mail.admin.pop3-maildrop-reconciliation",
+            title="Reconcile historical POP3 mailbox identities",
+            summary="Keep mailbox identity stable across secret rotation and reconcile legacy ambiguity without deleting evidence.",
+            body="Duplicate identity includes the effective canonical host, port and exact case-sensitive mailbox username within the tenant/profile/server. Password rotation or replacement credentials for the same account retain deduplication; another mailbox or endpoint is distinct. Preview revisions are bound to that identity. A reused UIDL with different bytes stops the entire selected batch with a conflict. Historical imports are retained without guessing their former account. A legacy identity conflict is not selectable: an administrator with mail:profile:use, mail:pop3:manage and mail:pop3:import must review the retained import, refresh the current mailbox preview, then POST /api/v1/mail/profiles/{profile_id}/pop3/imports/{import_id}/bind-maildrop with server_id, optional credential_id, expected_transport_revision from that preview, and confirm_binding=true. The operation re-downloads only that UIDL, verifies exact retained bytes and SHA-256, denies an existing competing binding, and audits the binding without changing content, IDs, historical fingerprints or source mail. Refresh the preview afterward. If content differs or the old source is unavailable, leave the record retained and the conflict unresolved; never delete evidence to force an import.",
+            layer="available", documentation_types=("user", "admin"), audience=("mail_admin", "mail_import_operator"), order=35,
+            conditions=(DocumentationCondition(required_modules=("mail",), any_scopes=("mail:pop3:import", "mail:pop3:manage")),),
+            links=(DocumentationLink(label="Legacy POP3 import", href="/mail/legacy-import", kind="runtime"), DocumentationLink(label="Bind a verified legacy maildrop", href="/api/v1/mail/profiles/{profile_id}/pop3/imports/{import_id}/bind-maildrop", kind="api")),
+            metadata={"kind": "reference", "related_topic_ids": ["mail.workflow.legacy-pop3-import"]},
+            translations={"de": {
+                "title": "Historische POP3-Postfachidentitäten abstimmen",
+                "summary": "Die Identität bei Passwortrotation erhalten und Altdatenkonflikte ohne Löschen von Nachweisen klären.",
+                "body": "Die Duplikaterkennung berücksichtigt innerhalb von Mandant, Profil und Server den kanonischen Host, Port und den exakten, groß-/kleinschreibungsabhängigen Postfachnamen. Passwortrotation oder neue Zugangsdaten desselben Kontos behalten die Zuordnung; andere Postfächer oder Endpunkte sind getrennt. Die Vorschaurevision ist daran gebunden. Dieselbe UIDL mit anderem Inhalt stoppt den gesamten ausgewählten Stapel als Konflikt. Historische Importe bleiben ohne erratene Kontozuordnung erhalten. Bei einem nicht auswählbaren Altdatenkonflikt prüft eine Administration mit mail:profile:use, mail:pop3:manage und mail:pop3:import den gespeicherten Import und aktualisiert die Postfachvorschau. Anschließend ruft sie POST /api/v1/mail/profiles/{profile_id}/pop3/imports/{import_id}/bind-maildrop mit server_id, optional credential_id, expected_transport_revision aus dieser Vorschau und confirm_binding=true auf. Nur diese UIDL wird erneut gelesen; exakte gespeicherte Bytes und SHA-256 müssen übereinstimmen, eine konkurrierende Zuordnung wird abgelehnt. Das Audit dokumentiert die Bindung; Inhalt, Kennungen, historische Fingerprints und Quellpostfach bleiben unverändert. Danach die Vorschau aktualisieren. Bei abweichendem Inhalt oder nicht verfügbarer Altquelle bleiben Nachweis und Konflikt erhalten; niemals Nachweise löschen, um einen Import zu erzwingen.",
+            }},
+        ),
         DocumentationTopic(
             id="mail.workflow.legacy-pop3-import",
             title="Import messages from a legacy POP3 mailbox",
@@ -723,6 +738,7 @@ manifest = ModuleManifest(
                 "kind": "workflow",
                 "route": "/mail/legacy-import",
                 "screen": "Legacy POP3 import",
+                "related_topic_ids": ["mail.admin.pop3-maildrop-reconciliation"],
                 "help_contexts": [
                     "mail.pop3",
                     "mail.pop3.source-editor",

src/govoplan_mail/backend/migrations/versions/b5d6e7f8091a_mail_pop3_maildrop_identity.py

@@ -0,0 +1,60 @@
+"""Scope POP3 imports to a stable effective maildrop.
+
+Revision ID: b5d6e7f8091a
+Revises: a4c5d6e7f809
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+revision = "b5d6e7f8091a"
+down_revision = "a4c5d6e7f809"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # Historical effective credentials may have changed. NULL is deliberate:
+    # never backfill from today's account or guess from the old credential ID.
+    with op.batch_alter_table("mail_pop3_imports") as batch:
+        batch.add_column(sa.Column("maildrop_identity", sa.String(64), nullable=True))
+        batch.drop_constraint("uq_mail_pop3_imports_source_uidl", type_="unique")
+        batch.create_unique_constraint(
+            "uq_mail_pop3_imports_maildrop_uidl",
+            [
+                "tenant_id",
+                "profile_id",
+                "pop3_server_id",
+                "maildrop_identity",
+                "provider_uidl",
+            ],
+        )
+
+
+def downgrade() -> None:
+    connection = op.get_bind()
+    table = sa.table(
+        "mail_pop3_imports",
+        *(
+            sa.column(name)
+            for name in ("tenant_id", "profile_id", "pop3_server_id", "provider_uidl")
+        ),
+    )
+    collision = connection.execute(
+        sa.select(sa.func.count())
+        .select_from(table)
+        .group_by(*table.c)
+        .having(sa.func.count() > 1)
+        .limit(1)
+    ).first()
+    if collision is not None:
+        raise RuntimeError(
+            "POP3 downgrade would collapse different maildrops; preserve all records and resolve this explicitly before downgrading"
+        )
+    with op.batch_alter_table("mail_pop3_imports") as batch:
+        batch.drop_constraint("uq_mail_pop3_imports_maildrop_uidl", type_="unique")
+        batch.create_unique_constraint(
+            "uq_mail_pop3_imports_source_uidl",
+            ["tenant_id", "profile_id", "pop3_server_id", "provider_uidl"],
+        )
+        batch.drop_column("maildrop_identity")

src/govoplan_mail/backend/pop3_imports.py

@@ -2,6 +2,8 @@ from __future__ import annotations
 
 import base64
 import hashlib
+import ipaddress
+import json
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from typing import Iterable
@@ -9,7 +11,12 @@ from typing import Iterable
 from sqlalchemy import or_, select
 from sqlalchemy.orm import Session
 
-from govoplan_core.security.secrets import encrypt_secret
+from govoplan_core.security.secrets import (
+    SecretDecryptionError,
+    decrypt_secret,
+    encrypt_secret,
+)
+from govoplan_mail.backend.config import Pop3Config
 from govoplan_mail.backend.db.models import MailPop3Import, MailServerEndpoint
 from govoplan_mail.backend.sending.pop3 import Pop3DownloadedMessage
 
@@ -18,6 +25,38 @@ class Pop3ImportError(RuntimeError):
     pass
 
 
+class Pop3ImportConflict(Pop3ImportError):
+    pass
+
+
+def maildrop_identity(config: Pop3Config) -> str:
+    """Stable mailbox key, excluding credentials and secret rotation revisions."""
+    if not config.host or not config.port or not config.username:
+        raise Pop3ImportError(
+            "POP3 maildrop identity requires a host, port and username"
+        )
+    host = config.host.strip().rstrip(".")
+    try:
+        host = ipaddress.ip_address(host).compressed
+    except ValueError:
+        try:
+            host = host.encode("idna").decode("ascii").casefold()
+        except UnicodeError as exc:
+            raise Pop3ImportError("The POP3 hostname is invalid") from exc
+    if not host:
+        raise Pop3ImportError("The POP3 hostname is invalid")
+    material = ("pop3-maildrop-v1", host, config.port, config.username)
+    return hashlib.sha256(
+        json.dumps(material, ensure_ascii=True, separators=(",", ":")).encode("ascii")
+    ).hexdigest()
+
+
+def pop3_preview_revision(transport_revision: str, identity: str) -> str:
+    return hashlib.sha256(
+        json.dumps((transport_revision, identity)).encode("ascii")
+    ).hexdigest()
+
+
 @dataclass(frozen=True, slots=True)
 class Pop3ImportResult:
     imported: tuple[MailPop3Import, ...]
@@ -32,11 +71,16 @@ def create_pop3_imports(
     pop3_server_id: str,
     pop3_credential_id: str | None,
     transport_revision: str,
+    maildrop_key: str,
     messages: Iterable[Pop3DownloadedMessage],
     user_id: str | None,
     deletion_requested: bool,
 ) -> Pop3ImportResult:
     downloaded = tuple(messages)
+    if len(maildrop_key) != 64 or any(
+        character not in "0123456789abcdef" for character in maildrop_key
+    ):
+        raise Pop3ImportError("A resolved POP3 maildrop identity is required")
     if not downloaded:
         raise Pop3ImportError("No POP3 messages were downloaded for import")
     uidls = [item.uidl for item in downloaded]
@@ -63,17 +107,36 @@ def create_pop3_imports(
     if source is None:
         raise Pop3ImportError("The selected POP3 source is unavailable")
 
-    existing = {
+    rows = tuple(
-        row.provider_uidl: row
+        session.scalars(
-        for row in session.scalars(
             select(MailPop3Import).where(
                 MailPop3Import.tenant_id == tenant_id,
                 MailPop3Import.profile_id == profile_id,
                 MailPop3Import.pop3_server_id == pop3_server_id,
                 MailPop3Import.provider_uidl.in_(uidls),
+                or_(
+                    MailPop3Import.maildrop_identity == maildrop_key,
+                    MailPop3Import.maildrop_identity.is_(None),
+                ),
             )
         )
-    }
+    )
+    # Validate the entire batch before adding records, including legacy ambiguity.
+    if any(row.maildrop_identity is None for row in rows):
+        raise Pop3ImportConflict(
+            "A selected UIDL has an unbound legacy import. An administrator must reconcile its maildrop identity before importing; no messages were imported or deleted."
+        )
+    existing = {row.provider_uidl: row for row in rows}
+    for message in downloaded:
+        if hashlib.sha256(message.raw).hexdigest() != message.raw_sha256:
+            raise Pop3ImportConflict(
+                "The downloaded POP3 content digest is inconsistent"
+            )
+        duplicate = existing.get(message.uidl)
+        if duplicate is not None and duplicate.raw_sha256 != message.raw_sha256:
+            raise Pop3ImportConflict(
+                "The maildrop reused a UIDL for different content. Review the retained and source messages; no messages were imported or deleted."
+            )
     imported: list[MailPop3Import] = []
     duplicates: list[MailPop3Import] = []
     now = datetime.now(timezone.utc)
@@ -92,6 +155,7 @@ def create_pop3_imports(
             pop3_server_id=pop3_server_id,
             pop3_credential_id=pop3_credential_id,
             transport_revision=_required_revision(transport_revision),
+            maildrop_identity=maildrop_key,
             provider_uidl=message.uidl,
             provider_message_number=message.message_number,
             fingerprint=_fingerprint(
@@ -100,6 +164,7 @@ def create_pop3_imports(
                 pop3_server_id=pop3_server_id,
                 uidl=message.uidl,
                 raw_sha256=message.raw_sha256,
+                maildrop_key=maildrop_key,
             ),
             raw_sha256=message.raw_sha256,
             raw_message_encrypted=encrypted,
@@ -122,6 +187,87 @@ def create_pop3_imports(
     return Pop3ImportResult(imported=tuple(imported), duplicates=tuple(duplicates))
 
 
+def bind_legacy_maildrop(
+    session: Session,
+    *,
+    tenant_id: str,
+    profile_id: str,
+    server_id: str,
+    import_id: str,
+    identity: str,
+    message: Pop3DownloadedMessage,
+) -> MailPop3Import:
+    # Same lock order as import, including when two administrators reconcile.
+    source = session.scalar(
+        select(MailServerEndpoint)
+        .where(
+            MailServerEndpoint.id == server_id,
+            MailServerEndpoint.profile_id == profile_id,
+            MailServerEndpoint.protocol == "pop3",
+            or_(
+                MailServerEndpoint.tenant_id == tenant_id,
+                MailServerEndpoint.tenant_id.is_(None),
+            ),
+        )
+        .with_for_update()
+    )
+    if source is None:
+        raise Pop3ImportError("The selected POP3 source is unavailable")
+    row = session.scalar(
+        select(MailPop3Import)
+        .where(
+            MailPop3Import.id == import_id,
+            MailPop3Import.tenant_id == tenant_id,
+            MailPop3Import.profile_id == profile_id,
+            MailPop3Import.pop3_server_id == server_id,
+        )
+        .with_for_update()
+        .execution_options(populate_existing=True)
+    )
+    if row is None:
+        raise Pop3ImportError("The legacy POP3 import is unavailable")
+    if row.maildrop_identity not in (None, identity):
+        raise Pop3ImportConflict(
+            "This POP3 import is already bound to another maildrop"
+        )
+    try:
+        retained = base64.b64decode(
+            decrypt_secret(row.raw_message_encrypted) or "", validate=True
+        )
+    except (ValueError, TypeError, SecretDecryptionError) as exc:
+        raise Pop3ImportConflict(
+            "The retained POP3 content could not be verified"
+        ) from exc
+    if (
+        message.uidl != row.provider_uidl
+        or retained != message.raw
+        or hashlib.sha256(retained).hexdigest() != row.raw_sha256
+        or message.raw_sha256 != row.raw_sha256
+    ):
+        raise Pop3ImportConflict(
+            "The current UIDL content does not match the retained legacy message; binding was refused"
+        )
+    collision = session.scalar(
+        select(MailPop3Import.id)
+        .where(
+            MailPop3Import.tenant_id == tenant_id,
+            MailPop3Import.profile_id == profile_id,
+            MailPop3Import.pop3_server_id == server_id,
+            MailPop3Import.provider_uidl == row.provider_uidl,
+            MailPop3Import.maildrop_identity == identity,
+            MailPop3Import.id != row.id,
+        )
+        .limit(1)
+    )
+    if collision is not None:
+        raise Pop3ImportConflict(
+            "Another import already owns this maildrop UIDL; no records were merged"
+        )
+    row.maildrop_identity = identity
+    session.flush()
+    return row
+
+
 def list_pop3_imports(
     session: Session,
     *,
@@ -130,9 +276,7 @@ def list_pop3_imports(
     profile_ids: Iterable[str] | None = None,
     limit: int = 100,
 ) -> tuple[MailPop3Import, ...]:
-    statement = select(MailPop3Import).where(
+    statement = select(MailPop3Import).where(MailPop3Import.tenant_id == tenant_id)
-        MailPop3Import.tenant_id == tenant_id
-    )
     if profile_id:
         statement = statement.where(MailPop3Import.profile_id == profile_id)
     elif profile_ids is not None:
@@ -160,7 +304,9 @@ def mark_pop3_deletion_result(
     clean_status = str(status or "").strip().casefold()
     if clean_status not in {"succeeded", "failed", "outcome_unknown"}:
         raise Pop3ImportError("Unsupported POP3 deletion result")
-    ids = tuple(dict.fromkeys(str(value).strip() for value in import_ids if str(value).strip()))
+    ids = tuple(
+        dict.fromkeys(str(value).strip() for value in import_ids if str(value).strip())
+    )
     if not ids:
         return ()
     rows = tuple(
@@ -192,6 +338,7 @@ def pop3_import_payload(row: MailPop3Import) -> dict[str, object]:
         "profile_id": row.profile_id,
         "pop3_server_id": row.pop3_server_id,
         "transport_revision": row.transport_revision,
+        "maildrop_identity_bound": bool(getattr(row, "maildrop_identity", None)),
         "provider_uidl": row.provider_uidl,
         "message_id": row.message_id,
         "subject": row.subject,
@@ -217,9 +364,10 @@ def _fingerprint(
     pop3_server_id: str,
     uidl: str,
     raw_sha256: str,
+    maildrop_key: str,
 ) -> str:
     material = "\x1f".join(
-        (tenant_id, profile_id, pop3_server_id, uidl, raw_sha256)
+        (tenant_id, profile_id, pop3_server_id, maildrop_key, uidl, raw_sha256)
     ).encode("utf-8")
     return hashlib.sha256(material).hexdigest()
 

src/govoplan_mail/backend/router.py

@@ -10,6 +10,7 @@ from sqlalchemy import func, or_, select
 from sqlalchemy.orm import Session
 
 from govoplan_mail.backend.schemas import (
+    MailPop3BindMaildropRequest,
     MailAddressLookupCandidate,
     MailAddressLookupResponse,
     MailAddressWriteTarget,
@@ -105,6 +106,10 @@ from govoplan_mail.backend.config import ImapConfig, JmapConfig, Pop3Config, Smt
 from govoplan_mail.backend.db.models import MailPop3Import
 from govoplan_mail.backend.pop3_imports import (
     Pop3ImportError,
+    Pop3ImportConflict,
+    maildrop_identity,
+    pop3_preview_revision,
+    bind_legacy_maildrop,
     create_pop3_imports,
     list_pop3_imports,
     mark_pop3_deletion_result,
@@ -3079,20 +3084,24 @@ def preview_profile_pop3_import(
             limit=payload.limit,
         )
         uidls = [message.uidl for message in result.messages]
-        imported_uidls = set(
+        identity = maildrop_identity(resolved.config)
-            session.scalars(
+        prior = list(
-                select(MailPop3Import.provider_uidl).where(
+            session.execute(
+                select(MailPop3Import.provider_uidl, MailPop3Import.maildrop_identity).where(
                     MailPop3Import.tenant_id == principal.tenant_id,
                     MailPop3Import.profile_id == profile_id,
                     MailPop3Import.pop3_server_id == resolved.server.id,
                     MailPop3Import.provider_uidl.in_(uidls),
+                    or_(MailPop3Import.maildrop_identity == identity, MailPop3Import.maildrop_identity.is_(None)),
                 )
             )
-        ) if uidls else set()
+        ) if uidls else []
+        legacy_uidls = {row.provider_uidl for row in prior if row.maildrop_identity is None}
+        imported_uidls = {row.provider_uidl for row in prior if row.maildrop_identity == identity}
         return MailPop3PreviewResponse(
             profile_id=profile_id,
             server_id=resolved.server.id,
-            transport_revision=resolved.transport_revision,
+            transport_revision=pop3_preview_revision(resolved.transport_revision, identity),
             host=result.host,
             port=result.port,
             security=result.security,
@@ -3111,6 +3120,7 @@ def preview_profile_pop3_import(
                     size_bytes=message.size_bytes,
                     body_preview=message.body_preview,
                     already_imported=message.uidl in imported_uidls,
+                    legacy_identity_conflict=message.uidl in legacy_uidls,
                 )
                 for message in result.messages
             ],
@@ -3154,7 +3164,8 @@ def import_profile_pop3_messages(
             server_id=payload.server_id,
             credential_id=payload.credential_id,
         )
-        if resolved.transport_revision != payload.expected_transport_revision:
+        identity = maildrop_identity(resolved.config)
+        if pop3_preview_revision(resolved.transport_revision, identity) != payload.expected_transport_revision:
             raise HTTPException(
                 status_code=status.HTTP_409_CONFLICT,
                 detail="The POP3 server or credential selection changed; refresh the preview before importing",
@@ -3177,6 +3188,7 @@ def import_profile_pop3_messages(
                 resolved.credential.id if resolved.credential is not None else None
             ),
             transport_revision=resolved.transport_revision,
+            maildrop_key=identity,
             messages=downloaded,
             user_id=principal.user.id,
             deletion_requested=payload.delete_after_import,
@@ -3272,6 +3284,11 @@ def import_profile_pop3_messages(
             status_code=status.HTTP_404_NOT_FOUND,
             detail=str(exc),
         ) from exc
+    except Pop3ImportConflict as exc:
+        session.rollback()
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT, detail=str(exc)
+        ) from exc
     except (Pop3ConfigurationError, Pop3ImportError) as exc:
         session.rollback()
         raise HTTPException(
@@ -3289,6 +3306,97 @@ def import_profile_pop3_messages(
         raise
 
 
+@router.post(
+    "/profiles/{profile_id}/pop3/imports/{import_id}/bind-maildrop",
+    response_model=MailPop3ImportRecordResponse,
+)
+def reconcile_pop3_import_maildrop(
+    profile_id: str,
+    import_id: str,
+    payload: MailPop3BindMaildropRequest,
+    principal: ApiPrincipal = Depends(get_api_principal),
+    session: Session = Depends(get_session),
+):
+    _require_scope(principal, "mail:profile:use")
+    _require_scope(principal, "mail:pop3:manage")
+    _require_scope(principal, "mail:pop3:import")
+    try:
+        _profile, resolved = _resolve_profile_pop3_transport(
+            session,
+            principal=principal,
+            profile_id=profile_id,
+            server_id=payload.server_id,
+            credential_id=payload.credential_id,
+        )
+        identity = maildrop_identity(resolved.config)
+        if (
+            pop3_preview_revision(resolved.transport_revision, identity)
+            != payload.expected_transport_revision
+        ):
+            raise Pop3ImportConflict(
+                "The POP3 selection changed; refresh the preview before reconciling"
+            )
+        row = session.scalar(
+            select(MailPop3Import).where(
+                MailPop3Import.id == import_id,
+                MailPop3Import.tenant_id == principal.tenant_id,
+                MailPop3Import.profile_id == profile_id,
+                MailPop3Import.pop3_server_id == resolved.server.id,
+            )
+        )
+        if row is None:
+            raise HTTPException(
+                status_code=404, detail="The legacy POP3 import is unavailable"
+            )
+        messages = download_pop3_messages(
+            pop3_config=resolved.config, uidls=(row.provider_uidl,)
+        )
+        if len(messages) != 1:
+            raise Pop3ImportConflict(
+                "The source did not return exactly the selected legacy message"
+            )
+        bound = bind_legacy_maildrop(
+            session,
+            tenant_id=principal.tenant_id,
+            profile_id=profile_id,
+            server_id=resolved.server.id,
+            import_id=import_id,
+            identity=identity,
+            message=messages[0],
+        )
+        audit_event(
+            session,
+            tenant_id=principal.tenant_id,
+            user_id=principal.user.id,
+            action="mail.pop3.maildrop_bound",
+            object_type="mail_pop3_import",
+            object_id=bound.id,
+            details={
+                "profile_id": profile_id,
+                "server_id": resolved.server.id,
+                "content_verified": True,
+                "source_deleted": False,
+            },
+        )
+        session.commit()
+        return MailPop3ImportRecordResponse.model_validate(pop3_import_payload(bound))
+    except Pop3ImportConflict as exc:
+        session.rollback()
+        raise HTTPException(status_code=409, detail=str(exc)) from exc
+    except (MailProfileError, MailServerHierarchyError) as exc:
+        session.rollback()
+        raise HTTPException(status_code=404, detail=str(exc)) from exc
+    except (Pop3ConfigurationError, Pop3ImportError) as exc:
+        session.rollback()
+        raise HTTPException(status_code=422, detail=str(exc)) from exc
+    except Pop3ProviderError as exc:
+        session.rollback()
+        raise HTTPException(status_code=502, detail=str(exc)) from exc
+    except Exception:
+        session.rollback()
+        raise
+
+
 @router.get("/pop3/imports", response_model=MailPop3ImportListResponse)
 def get_pop3_imports(
     profile_id: str | None = Query(default=None),

src/govoplan_mail/backend/schemas.py

@@ -410,6 +410,7 @@ class MailPop3MessagePreviewResponse(BaseModel):
     size_bytes: int = 0
     body_preview: str | None = None
     already_imported: bool = False
+    legacy_identity_conflict: bool = False
 
 
 class MailPop3PreviewResponse(BaseModel):
@@ -440,6 +441,7 @@ class MailPop3ImportRecordResponse(BaseModel):
     profile_id: str
     pop3_server_id: str
     transport_revision: str
+    maildrop_identity_bound: bool = False
     provider_uidl: str
     message_id: str | None = None
     subject: str | None = None
@@ -467,6 +469,15 @@ class MailPop3ImportListResponse(BaseModel):
     imports: list[MailPop3ImportRecordResponse] = Field(default_factory=list)
 
 
+class MailPop3BindMaildropRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    server_id: str = Field(min_length=1, max_length=36)
+    credential_id: str | None = Field(default=None, max_length=36)
+    expected_transport_revision: str = Field(min_length=1, max_length=120)
+    confirm_binding: Literal[True]
+
+
 class MailImapFolderResponse(BaseModel):
     name: str
     flags: list[str] = Field(default_factory=list)

tests/test_pop3_imports.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 import base64
+import dataclasses
+import importlib
 from datetime import UTC, datetime
 import poplib
 import ssl
@@ -8,7 +10,12 @@ from types import SimpleNamespace
 import unittest
 from unittest.mock import patch
 
-from sqlalchemy import create_engine
+from alembic.migration import MigrationContext
+from alembic.operations import Operations
+from fastapi import HTTPException
+from pydantic import ValidationError
+
+from sqlalchemy import MetaData, Table, create_engine, inspect, select
 from sqlalchemy.orm import sessionmaker
 
 from govoplan_access.backend.db.models import Account, User
@@ -23,12 +30,24 @@ from govoplan_mail.backend.db.models import (
     MailServerProfile,
 )
 from govoplan_mail.backend.pop3_imports import (
+    Pop3ImportConflict,
+    bind_legacy_maildrop,
+    maildrop_identity,
+    pop3_preview_revision,
     Pop3ImportResult,
     create_pop3_imports,
     list_pop3_imports,
 )
-from govoplan_mail.backend.router import import_profile_pop3_messages
+from govoplan_mail.backend.router import (
-from govoplan_mail.backend.schemas import MailPop3ImportRequest
+    import_profile_pop3_messages,
+    preview_profile_pop3_import,
+    reconcile_pop3_import_maildrop,
+)
+from govoplan_mail.backend.schemas import (
+    MailPop3BindMaildropRequest,
+    MailPop3ImportRequest,
+    MailPop3PreviewRequest,
+)
 from govoplan_mail.backend.sending.pop3 import (
     Pop3ConfigurationError,
     Pop3DownloadedMessage,
@@ -152,9 +171,7 @@ class Pop3TransportTests(unittest.TestCase):
             "govoplan_mail.backend.sending.pop3._open_pop3",
             return_value=download_client,
         ):
-            downloaded = download_pop3_messages(
+            downloaded = download_pop3_messages(pop3_config=_config(), uidls=("uid-1",))
-                pop3_config=_config(), uidls=("uid-1",)
-            )
 
         self.assertEqual(_RAW, downloaded[0].raw)
         self.assertEqual([], download_client.deletions)
@@ -196,9 +213,7 @@ class Pop3TransportTests(unittest.TestCase):
 
     def test_tls_and_authentication_failures_are_sanitized(self) -> None:
         with (
-            patch(
+            patch("govoplan_mail.backend.sending.pop3.validate_outbound_host"),
-                "govoplan_mail.backend.sending.pop3.validate_outbound_host"
-            ),
             patch(
                 "govoplan_mail.backend.sending.pop3._OutboundPolicyPOP3SSL",
                 side_effect=ssl.SSLError("private TLS detail"),
@@ -213,9 +228,7 @@ class Pop3TransportTests(unittest.TestCase):
             poplib.error_proto("private auth detail")
         )
         with (
-            patch(
+            patch("govoplan_mail.backend.sending.pop3.validate_outbound_host"),
-                "govoplan_mail.backend.sending.pop3.validate_outbound_host"
-            ),
             patch(
                 "govoplan_mail.backend.sending.pop3._OutboundPolicyPOP3",
                 return_value=auth_client,
@@ -274,6 +287,7 @@ class Pop3PersistenceTests(unittest.TestCase):
             pop3_server_id=self.server.id,
             pop3_credential_id=None,
             transport_revision="revision-1",
+            maildrop_key=maildrop_identity(_config()),
             messages=(_download(),),
             user_id=None,
             deletion_requested=False,
@@ -296,13 +310,16 @@ class Pop3PersistenceTests(unittest.TestCase):
             pop3_server_id=self.server.id,
             pop3_credential_id=None,
             transport_revision="revision-1",
+            maildrop_key=maildrop_identity(_config()),
             messages=(_download(),),
             user_id=None,
             deletion_requested=False,
         )
 
         self.assertEqual((), repeated.imported)
-        self.assertEqual((first.imported[0].id,), tuple(row.id for row in repeated.duplicates))
+        self.assertEqual(
+            (first.imported[0].id,), tuple(row.id for row in repeated.duplicates)
+        )
         self.assertEqual(
             (first.imported[0].id,),
             tuple(
@@ -323,6 +340,414 @@ class Pop3PersistenceTests(unittest.TestCase):
             ),
         )
 
+    def _import(
+        self,
+        *,
+        config=None,
+        messages=None,
+        credential="credential-1",
+        tenant="tenant-1",
+    ):
+        return create_pop3_imports(
+            self.session,
+            tenant_id=tenant,
+            profile_id=self.profile.id,
+            pop3_server_id=self.server.id,
+            pop3_credential_id=credential,
+            transport_revision="revision-1",
+            maildrop_key=maildrop_identity(config or _config()),
+            messages=messages or (_download(),),
+            user_id=None,
+            deletion_requested=False,
+        )
+
+    def test_mailbox_identity_survives_secret_rotation_but_not_username_or_host(self):
+        original = maildrop_identity(_config())
+        self.assertEqual(original, maildrop_identity(_config(password="rotated")))
+        self.assertEqual(original, maildrop_identity(_config(host="POP3.EXAMPLE.TEST")))
+        for change in (
+            {"username": "Legacy-user"},
+            {"host": "other.example.test"},
+            {"port": 1995},
+        ):
+            self.assertNotEqual(original, maildrop_identity(_config(**change)))
+        first = self._import().imported[0]
+        self.session.commit()
+        self.assertEqual(
+            first.id,
+            self._import(config=_config(password="rotated"), credential="credential-2")
+            .duplicates[0]
+            .id,
+        )
+        second = self._import(config=_config(username="mailbox-B")).imported[0]
+        self.assertNotEqual(first.id, second.id)
+        self.assertNotEqual(first.fingerprint, second.fingerprint)
+
+    def test_tenant_boundary_remains_in_identity_lookup_and_constraint(self):
+        self.server.tenant_id = None
+        self.profile.tenant_id = None
+        first = self._import().imported[0]
+        self.session.commit()
+        second = self._import(tenant="tenant-2").imported[0]
+        self.assertNotEqual(first.id, second.id)
+
+    def test_changed_uidl_content_rejects_the_whole_batch_before_insert(self):
+        self._import()
+        self.session.commit()
+        import hashlib
+
+        message = dataclasses.replace(
+            _download(),
+            raw=b"different",
+            raw_sha256=hashlib.sha256(b"different").hexdigest(),
+        )
+        with self.assertRaisesRegex(Pop3ImportConflict, "different content"):
+            self._import(messages=(_download("new-uidl"), message))
+        self.assertEqual(1, self.session.query(MailPop3Import).count())
+        self.assertFalse(self.session.new)
+
+    def test_legacy_conflict_is_explicit_and_binding_is_non_destructive_and_repeatable(
+        self,
+    ):
+        row = self._import().imported[0]
+        row.maildrop_identity = None
+        self.session.commit()
+        retained = (
+            row.id,
+            row.raw_message_encrypted,
+            row.fingerprint,
+            row.transport_revision,
+        )
+        with self.assertRaisesRegex(Pop3ImportConflict, "unbound legacy"):
+            self._import()
+        for _ in range(2):
+            bound = bind_legacy_maildrop(
+                self.session,
+                tenant_id="tenant-1",
+                profile_id=self.profile.id,
+                server_id=self.server.id,
+                import_id=row.id,
+                identity=maildrop_identity(_config()),
+                message=_download(),
+            )
+            self.session.commit()
+            self.assertEqual(
+                retained,
+                (
+                    bound.id,
+                    bound.raw_message_encrypted,
+                    bound.fingerprint,
+                    bound.transport_revision,
+                ),
+            )
+        self.assertEqual(row.id, self._import().duplicates[0].id)
+
+    def test_legacy_binding_rejects_different_content_and_collision(self):
+        row = self._import().imported[0]
+        self._import(config=_config(username="mailbox-B"))
+        row.maildrop_identity = None
+        self.session.commit()
+        with self.assertRaisesRegex(Pop3ImportConflict, "does not match"):
+            bind_legacy_maildrop(
+                self.session,
+                tenant_id="tenant-1",
+                profile_id=self.profile.id,
+                server_id=self.server.id,
+                import_id=row.id,
+                identity=maildrop_identity(_config()),
+                message=dataclasses.replace(_download(), raw=b"different"),
+            )
+        with self.assertRaisesRegex(Pop3ImportConflict, "already owns"):
+            bind_legacy_maildrop(
+                self.session,
+                tenant_id="tenant-1",
+                profile_id=self.profile.id,
+                server_id=self.server.id,
+                import_id=row.id,
+                identity=maildrop_identity(_config(username="mailbox-B")),
+                message=_download(),
+            )
+        self.assertIsNone(row.maildrop_identity)
+
+    def _principal(self, *, tenant="tenant-1", omitted=None):
+        scopes = {"mail:profile:use", "mail:pop3:manage", "mail:pop3:import"} - {
+            omitted
+        }
+        return ApiPrincipal(
+            principal=PrincipalRef(
+                account_id="account",
+                membership_id="user",
+                tenant_id=tenant,
+                scopes=frozenset(scopes),
+            ),
+            account=SimpleNamespace(id="account"),
+            user=SimpleNamespace(id="user"),
+        )
+
+    def test_legacy_binding_rejects_unverifiable_retained_ciphertext(self):
+        row = self._import().imported[0]
+        row.maildrop_identity = None
+        row.raw_message_encrypted = "invalid retained ciphertext"
+        self.session.commit()
+        with self.assertRaisesRegex(Pop3ImportConflict, "could not be verified"):
+            bind_legacy_maildrop(
+                self.session,
+                tenant_id="tenant-1",
+                profile_id=self.profile.id,
+                server_id=self.server.id,
+                import_id=row.id,
+                identity=maildrop_identity(_config()),
+                message=_download(),
+            )
+        self.assertIsNone(row.maildrop_identity)
+        self.assertEqual("invalid retained ciphertext", row.raw_message_encrypted)
+
+    def _resolved(self, config=None):
+        return (
+            self.profile,
+            SimpleNamespace(
+                config=config or _config(),
+                server=self.server,
+                credential=None,
+                transport_revision="revision-1",
+            ),
+        )
+
+    def _binding_request(self):
+        return MailPop3BindMaildropRequest(
+            server_id=self.server.id,
+            expected_transport_revision=pop3_preview_revision(
+                "revision-1", maildrop_identity(_config())
+            ),
+            confirm_binding=True,
+        )
+
+    def test_preview_distinguishes_mailboxes_and_reports_unbound_legacy(self):
+        row = self._import().imported[0]
+        self.session.commit()
+        preview = SimpleNamespace(
+            messages=(_download().summary,),
+            host="pop3.example.test",
+            port=995,
+            security="tls",
+            message_count=1,
+            mailbox_size_bytes=len(_RAW),
+        )
+        for config, imported, conflict in (
+            (_config(), True, False),
+            (_config(username="other"), False, False),
+            (_config(), False, True),
+        ):
+            if conflict:
+                row.maildrop_identity = None
+                self.session.commit()
+            with (
+                patch(
+                    "govoplan_mail.backend.router._resolve_profile_pop3_transport",
+                    return_value=self._resolved(config),
+                ),
+                patch(
+                    "govoplan_mail.backend.router.preview_pop3_messages",
+                    return_value=preview,
+                ),
+            ):
+                result = preview_profile_pop3_import(
+                    self.profile.id,
+                    MailPop3PreviewRequest(server_id=self.server.id),
+                    principal=self._principal(),
+                    session=self.session,
+                )
+            self.assertEqual(imported, result.messages[0].already_imported)
+            self.assertEqual(conflict, result.messages[0].legacy_identity_conflict)
+            self.assertEqual(
+                pop3_preview_revision("revision-1", maildrop_identity(config)),
+                result.transport_revision,
+            )
+
+    def test_reconciliation_route_checks_each_scope_before_resolving_or_downloading(
+        self,
+    ):
+        for scope in ("mail:profile:use", "mail:pop3:manage", "mail:pop3:import"):
+            with (
+                patch(
+                    "govoplan_mail.backend.router._resolve_profile_pop3_transport"
+                ) as resolve,
+                self.assertRaises(HTTPException) as caught,
+            ):
+                reconcile_pop3_import_maildrop(
+                    self.profile.id,
+                    "unused",
+                    self._binding_request(),
+                    principal=self._principal(omitted=scope),
+                    session=self.session,
+                )
+            self.assertEqual(403, caught.exception.status_code)
+            resolve.assert_not_called()
+        with self.assertRaises(ValidationError):
+            MailPop3BindMaildropRequest(
+                server_id=self.server.id,
+                expected_transport_revision="revision",
+                confirm_binding=False,
+            )
+
+    def test_reconciliation_route_checks_tenant_and_preview_before_download(self):
+        row = self._import().imported[0]
+        row.maildrop_identity = None
+        self.session.commit()
+        with (
+            patch(
+                "govoplan_mail.backend.router._resolve_profile_pop3_transport",
+                return_value=self._resolved(),
+            ),
+            patch("govoplan_mail.backend.router.download_pop3_messages") as download,
+        ):
+            with self.assertRaises(HTTPException) as caught:
+                reconcile_pop3_import_maildrop(
+                    self.profile.id,
+                    row.id,
+                    self._binding_request(),
+                    principal=self._principal(tenant="another"),
+                    session=self.session,
+                )
+            self.assertEqual(404, caught.exception.status_code)
+            stale = self._binding_request().model_copy(
+                update={"expected_transport_revision": "old-preview"}
+            )
+            with self.assertRaises(HTTPException) as caught:
+                reconcile_pop3_import_maildrop(
+                    self.profile.id,
+                    row.id,
+                    stale,
+                    principal=self._principal(),
+                    session=self.session,
+                )
+            self.assertEqual(409, caught.exception.status_code)
+            download.assert_not_called()
+
+    def test_reconciliation_route_audit_failure_rolls_back_and_success_never_deletes(
+        self,
+    ):
+        row = self._import().imported[0]
+        row.maildrop_identity = None
+        self.session.commit()
+        with (
+            patch(
+                "govoplan_mail.backend.router._resolve_profile_pop3_transport",
+                return_value=self._resolved(),
+            ),
+            patch(
+                "govoplan_mail.backend.router.download_pop3_messages",
+                return_value=(_download(),),
+            ),
+            patch("govoplan_mail.backend.router.delete_pop3_messages") as delete,
+            patch(
+                "govoplan_mail.backend.router.audit_event",
+                side_effect=RuntimeError("audit unavailable"),
+            ),
+        ):
+            with self.assertRaisesRegex(RuntimeError, "audit unavailable"):
+                reconcile_pop3_import_maildrop(
+                    self.profile.id,
+                    row.id,
+                    self._binding_request(),
+                    principal=self._principal(),
+                    session=self.session,
+                )
+            self.assertIsNone(
+                self.session.get(MailPop3Import, row.id).maildrop_identity
+            )
+            delete.assert_not_called()
+        with (
+            patch(
+                "govoplan_mail.backend.router._resolve_profile_pop3_transport",
+                return_value=self._resolved(),
+            ),
+            patch(
+                "govoplan_mail.backend.router.download_pop3_messages",
+                return_value=(_download(),),
+            ),
+            patch("govoplan_mail.backend.router.delete_pop3_messages") as delete,
+            patch("govoplan_mail.backend.router.audit_event") as audit,
+        ):
+            result = reconcile_pop3_import_maildrop(
+                self.profile.id,
+                row.id,
+                self._binding_request(),
+                principal=self._principal(),
+                session=self.session,
+            )
+            self.assertTrue(result.maildrop_identity_bound)
+            self.assertEqual(row.id, result.id)
+            self.assertEqual(
+                "mail.pop3.maildrop_bound", audit.call_args.kwargs["action"]
+            )
+            delete.assert_not_called()
+
+    def test_migration_retains_unbound_records_and_refuses_lossy_downgrade(self):
+        self.session.close()
+        MailPop3Import.__table__.drop(self.engine)
+        old = importlib.import_module(
+            "govoplan_mail.backend.migrations.versions.a4c5d6e7f809_mail_pop3_imports"
+        )
+        new = importlib.import_module(
+            "govoplan_mail.backend.migrations.versions.b5d6e7f8091a_mail_pop3_maildrop_identity"
+        )
+        with self.engine.begin() as connection:
+            with Operations.context(MigrationContext.configure(connection)):
+                old.upgrade()
+            table = Table("mail_pop3_imports", MetaData(), autoload_with=connection)
+            now = datetime.now(UTC)
+            values = dict(
+                id="legacy",
+                tenant_id="tenant-1",
+                profile_id=self.profile.id,
+                pop3_server_id=self.server.id,
+                transport_revision="historical",
+                provider_uidl="uidl",
+                fingerprint="original",
+                raw_sha256="retained",
+                raw_message_encrypted="retained ciphertext",
+                size_bytes=7,
+                status="pending_review",
+                imported_at=now,
+                deletion_requested=False,
+                deletion_status="not_requested",
+                created_at=now,
+                updated_at=now,
+            )
+            connection.execute(table.insert().values(**values))
+            with Operations.context(MigrationContext.configure(connection)):
+                new.upgrade()
+            current = Table("mail_pop3_imports", MetaData(), autoload_with=connection)
+            retained = connection.execute(select(current)).mappings().one()
+            self.assertIsNone(retained["maildrop_identity"])
+            for key in (
+                "id",
+                "fingerprint",
+                "raw_message_encrypted",
+                "raw_sha256",
+                "transport_revision",
+            ):
+                self.assertEqual(values[key], retained[key])
+            connection.execute(
+                current.insert().values(
+                    **{**values, "id": "new-maildrop", "maildrop_identity": "a" * 64}
+                )
+            )
+            with (
+                Operations.context(MigrationContext.configure(connection)),
+                self.assertRaisesRegex(RuntimeError, "collapse different maildrops"),
+            ):
+                new.downgrade()
+            self.assertIn(
+                "maildrop_identity",
+                {
+                    column["name"]
+                    for column in inspect(connection).get_columns("mail_pop3_imports")
+                },
+            )
+            self.assertEqual(2, len(connection.execute(select(current.c.id)).all()))
+
 
 class _RouteSession:
     def __init__(self, events: list[str]) -> None:
@@ -384,7 +809,9 @@ class Pop3ImportRouteTests(unittest.TestCase):
         )
         payload = MailPop3ImportRequest(
             server_id="server-1",
-            expected_transport_revision="revision-1",
+            expected_transport_revision=pop3_preview_revision(
+                "revision-1", maildrop_identity(_config())
+            ),
             uidls=["uid-1"],
             delete_after_import=True,
         )
@@ -416,7 +843,9 @@ class Pop3ImportRouteTests(unittest.TestCase):
                 return_value=Pop3ImportResult(imported=(row,), duplicates=()),
             ),
             patch("govoplan_mail.backend.router.audit_event", side_effect=audit),
-            patch("govoplan_mail.backend.router.delete_pop3_messages", side_effect=delete),
+            patch(
+                "govoplan_mail.backend.router.delete_pop3_messages", side_effect=delete
+            ),
             patch(
                 "govoplan_mail.backend.router.mark_pop3_deletion_result",
                 side_effect=mark,

webui/package-lock.json

@@ -1,12 +1,12 @@
 {
   "name": "@govoplan/mail-webui",
-  "version": "0.1.27",
+  "version": "0.1.28",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "@govoplan/mail-webui",
-      "version": "0.1.27",
+      "version": "0.1.28",
       "devDependencies": {
         "typescript": "^5.7.2"
       },

webui/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@govoplan/mail-webui",
-  "version": "0.1.27",
+  "version": "0.1.28",
   "private": true,
   "type": "module",
   "main": "src/index.ts",

webui/src/api/mail.ts

@@ -349,6 +349,7 @@ export type MailPop3MessagePreview = {
   size_bytes: number;
   body_preview?: string | null;
   already_imported: boolean;
+  legacy_identity_conflict?: boolean;
 };
 
 export type MailPop3PreviewResponse = {

webui/src/features/mail/MailLegacyImportPage.tsx

@@ -189,7 +189,7 @@ export default function MailLegacyImportPage({
       });
       setPreview(next);
       setSelectedUidls((current) =>
-        current.filter((uidl) => next.messages.some((item) => item.uidl === uidl && !item.already_imported))
+        current.filter((uidl) => next.messages.some((item) => item.uidl === uidl && !item.already_imported && !item.legacy_identity_conflict))
       );
     } catch (reason) {
       setError(adminErrorMessage(reason));
@@ -341,7 +341,7 @@ export default function MailLegacyImportPage({
         data-help-context-id="mail.pop3.field.message-selection"
         data-help-module-id="mail"
         checked={selectedUidls.includes(item.uidl)}
-        disabled={Boolean(busy) || item.already_imported}
+        disabled={Boolean(busy) || item.already_imported || item.legacy_identity_conflict}
         onChange={() => setSelectedUidls((current) =>
           current.includes(item.uidl)
             ? current.filter((uidl) => uidl !== item.uidl)
@@ -362,8 +362,8 @@ export default function MailLegacyImportPage({
       id: "state",
       header: "State",
       width: 130,
-      value: (item) => item.already_imported ? "imported" : "available",
+      value: (item) => item.legacy_identity_conflict ? "legacy identity conflict" : item.already_imported ? "imported" : "available",
-      render: (item) => <StatusBadge status={item.already_imported ? "inactive" : "success"} label={item.already_imported ? "imported" : "available"} />
+      render: (item) => <StatusBadge status={item.legacy_identity_conflict ? "warning" : item.already_imported ? "inactive" : "success"} label={item.legacy_identity_conflict ? "review legacy identity" : item.already_imported ? "imported" : "available"} />
     }
   ];