247 lines
8.0 KiB
Python
247 lines
8.0 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from typing import Iterable
|
|
|
|
from sqlalchemy import or_, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.security.secrets import encrypt_secret
|
|
from govoplan_mail.backend.db.models import MailPop3Import, MailServerEndpoint
|
|
from govoplan_mail.backend.sending.pop3 import Pop3DownloadedMessage
|
|
|
|
|
|
class Pop3ImportError(RuntimeError):
|
|
pass
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class Pop3ImportResult:
|
|
imported: tuple[MailPop3Import, ...]
|
|
duplicates: tuple[MailPop3Import, ...]
|
|
|
|
|
|
def create_pop3_imports(
|
|
session: Session,
|
|
*,
|
|
tenant_id: str,
|
|
profile_id: str,
|
|
pop3_server_id: str,
|
|
pop3_credential_id: str | None,
|
|
transport_revision: str,
|
|
messages: Iterable[Pop3DownloadedMessage],
|
|
user_id: str | None,
|
|
deletion_requested: bool,
|
|
) -> Pop3ImportResult:
|
|
downloaded = tuple(messages)
|
|
if not downloaded:
|
|
raise Pop3ImportError("No POP3 messages were downloaded for import")
|
|
uidls = [item.uidl for item in downloaded]
|
|
if len(uidls) != len(set(uidls)):
|
|
raise Pop3ImportError("The POP3 download contained duplicate UIDL identifiers")
|
|
|
|
# Serialize imports per source before checking UIDLs. The database unique
|
|
# constraint remains the last line of defense, while this lock lets a
|
|
# concurrent request observe the first request's committed rows and report
|
|
# them as duplicates instead of surfacing an integrity error.
|
|
source = session.scalar(
|
|
select(MailServerEndpoint)
|
|
.where(
|
|
MailServerEndpoint.id == pop3_server_id,
|
|
MailServerEndpoint.profile_id == profile_id,
|
|
or_(
|
|
MailServerEndpoint.tenant_id == tenant_id,
|
|
MailServerEndpoint.tenant_id.is_(None),
|
|
),
|
|
MailServerEndpoint.protocol == "pop3",
|
|
)
|
|
.with_for_update()
|
|
)
|
|
if source is None:
|
|
raise Pop3ImportError("The selected POP3 source is unavailable")
|
|
|
|
existing = {
|
|
row.provider_uidl: row
|
|
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),
|
|
)
|
|
)
|
|
}
|
|
imported: list[MailPop3Import] = []
|
|
duplicates: list[MailPop3Import] = []
|
|
now = datetime.now(timezone.utc)
|
|
for message in downloaded:
|
|
duplicate = existing.get(message.uidl)
|
|
if duplicate is not None:
|
|
duplicates.append(duplicate)
|
|
continue
|
|
encrypted = encrypt_secret(base64.b64encode(message.raw).decode("ascii"))
|
|
if not encrypted:
|
|
raise Pop3ImportError("The downloaded POP3 message could not be encrypted")
|
|
summary = message.summary
|
|
row = MailPop3Import(
|
|
tenant_id=tenant_id,
|
|
profile_id=profile_id,
|
|
pop3_server_id=pop3_server_id,
|
|
pop3_credential_id=pop3_credential_id,
|
|
transport_revision=_required_revision(transport_revision),
|
|
provider_uidl=message.uidl,
|
|
provider_message_number=message.message_number,
|
|
fingerprint=_fingerprint(
|
|
tenant_id=tenant_id,
|
|
profile_id=profile_id,
|
|
pop3_server_id=pop3_server_id,
|
|
uidl=message.uidl,
|
|
raw_sha256=message.raw_sha256,
|
|
),
|
|
raw_sha256=message.raw_sha256,
|
|
raw_message_encrypted=encrypted,
|
|
message_id=summary.message_id,
|
|
subject=summary.subject,
|
|
from_header=summary.from_header,
|
|
to_header=summary.to_header,
|
|
date=summary.date,
|
|
body_preview=summary.body_preview,
|
|
size_bytes=len(message.raw),
|
|
status="pending_review",
|
|
imported_at=now,
|
|
imported_by_user_id=user_id,
|
|
deletion_requested=bool(deletion_requested),
|
|
deletion_status=("pending" if deletion_requested else "not_requested"),
|
|
)
|
|
session.add(row)
|
|
imported.append(row)
|
|
session.flush()
|
|
return Pop3ImportResult(imported=tuple(imported), duplicates=tuple(duplicates))
|
|
|
|
|
|
def list_pop3_imports(
|
|
session: Session,
|
|
*,
|
|
tenant_id: str,
|
|
profile_id: str | None = None,
|
|
profile_ids: Iterable[str] | None = None,
|
|
limit: int = 100,
|
|
) -> tuple[MailPop3Import, ...]:
|
|
statement = select(MailPop3Import).where(
|
|
MailPop3Import.tenant_id == tenant_id
|
|
)
|
|
if profile_id:
|
|
statement = statement.where(MailPop3Import.profile_id == profile_id)
|
|
elif profile_ids is not None:
|
|
allowed = tuple(dict.fromkeys(str(value) for value in profile_ids if value))
|
|
if not allowed:
|
|
return ()
|
|
statement = statement.where(MailPop3Import.profile_id.in_(allowed))
|
|
rows = session.scalars(
|
|
statement.order_by(
|
|
MailPop3Import.imported_at.desc(),
|
|
MailPop3Import.id.desc(),
|
|
).limit(max(1, min(int(limit), 500)))
|
|
)
|
|
return tuple(rows)
|
|
|
|
|
|
def mark_pop3_deletion_result(
|
|
session: Session,
|
|
*,
|
|
tenant_id: str,
|
|
import_ids: Iterable[str],
|
|
status: str,
|
|
error: str | None = None,
|
|
) -> tuple[MailPop3Import, ...]:
|
|
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()))
|
|
if not ids:
|
|
return ()
|
|
rows = tuple(
|
|
session.scalars(
|
|
select(MailPop3Import)
|
|
.where(
|
|
MailPop3Import.tenant_id == tenant_id,
|
|
MailPop3Import.id.in_(ids),
|
|
MailPop3Import.deletion_requested.is_(True),
|
|
)
|
|
.with_for_update()
|
|
)
|
|
)
|
|
if len(rows) != len(ids):
|
|
raise Pop3ImportError("One or more POP3 import records are unavailable")
|
|
now = datetime.now(timezone.utc)
|
|
safe_error = _bounded_error(error)
|
|
for row in rows:
|
|
row.deletion_status = clean_status
|
|
row.deletion_attempted_at = now
|
|
row.deletion_error = safe_error
|
|
session.flush()
|
|
return rows
|
|
|
|
|
|
def pop3_import_payload(row: MailPop3Import) -> dict[str, object]:
|
|
return {
|
|
"id": row.id,
|
|
"profile_id": row.profile_id,
|
|
"pop3_server_id": row.pop3_server_id,
|
|
"transport_revision": row.transport_revision,
|
|
"provider_uidl": row.provider_uidl,
|
|
"message_id": row.message_id,
|
|
"subject": row.subject,
|
|
"from_header": row.from_header,
|
|
"to_header": row.to_header,
|
|
"date": row.date,
|
|
"body_preview": row.body_preview,
|
|
"size_bytes": row.size_bytes,
|
|
"raw_sha256": row.raw_sha256,
|
|
"status": row.status,
|
|
"imported_at": row.imported_at,
|
|
"deletion_requested": row.deletion_requested,
|
|
"deletion_status": row.deletion_status,
|
|
"deletion_attempted_at": row.deletion_attempted_at,
|
|
"deletion_error": row.deletion_error,
|
|
}
|
|
|
|
|
|
def _fingerprint(
|
|
*,
|
|
tenant_id: str,
|
|
profile_id: str,
|
|
pop3_server_id: str,
|
|
uidl: str,
|
|
raw_sha256: str,
|
|
) -> str:
|
|
material = "\x1f".join(
|
|
(tenant_id, profile_id, pop3_server_id, uidl, raw_sha256)
|
|
).encode("utf-8")
|
|
return hashlib.sha256(material).hexdigest()
|
|
|
|
|
|
def _required_revision(value: object) -> str:
|
|
clean = str(value or "").strip()
|
|
if not clean or len(clean) > 120:
|
|
raise Pop3ImportError("A valid POP3 transport revision is required")
|
|
return clean
|
|
|
|
|
|
def _bounded_error(value: str | None) -> str | None:
|
|
clean = " ".join(str(value or "").split())
|
|
return clean[:500] or None
|
|
|
|
|
|
__all__ = [
|
|
"Pop3ImportError",
|
|
"Pop3ImportResult",
|
|
"create_pop3_imports",
|
|
"list_pop3_imports",
|
|
"mark_pop3_deletion_result",
|
|
"pop3_import_payload",
|
|
]
|