154 lines
5.0 KiB
Python
154 lines
5.0 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import timedelta
|
|
from uuid import uuid4
|
|
|
|
from cryptography.fernet import Fernet, InvalidToken, MultiFernet
|
|
from django.conf import settings
|
|
from django.db import transaction
|
|
from django.utils import timezone
|
|
|
|
from apps.sources.models import MailboxConnection
|
|
|
|
|
|
class MailboxCredentialError(RuntimeError):
|
|
"""Safe credential configuration error without secret material."""
|
|
|
|
|
|
def _cipher() -> MultiFernet:
|
|
keys = settings.MAILBOX_CREDENTIAL_KEYS
|
|
if not keys:
|
|
raise MailboxCredentialError(
|
|
"Credentialopslag is niet geconfigureerd. Stel MAILBOX_CREDENTIAL_KEYS in."
|
|
)
|
|
try:
|
|
return MultiFernet([Fernet(key.encode("ascii")) for key in keys])
|
|
except (ValueError, TypeError) as exc:
|
|
raise MailboxCredentialError(
|
|
"MAILBOX_CREDENTIAL_KEYS bevat een ongeldige sleutel."
|
|
) from exc
|
|
|
|
|
|
def encrypt_mailbox_password(password: str) -> str:
|
|
if not password:
|
|
raise MailboxCredentialError("Een app-wachtwoord is verplicht.")
|
|
return _cipher().encrypt(password.encode("utf-8")).decode("ascii")
|
|
|
|
|
|
def decrypt_mailbox_password(connection: MailboxConnection) -> str:
|
|
try:
|
|
return _cipher().decrypt(connection.encrypted_password.encode("ascii")).decode("utf-8")
|
|
except (InvalidToken, ValueError, UnicodeError) as exc:
|
|
raise MailboxCredentialError(
|
|
"Het opgeslagen app-wachtwoord kan niet worden ontsleuteld. Sla het opnieuw op."
|
|
) from exc
|
|
|
|
|
|
@transaction.atomic
|
|
def rotate_mailbox_credentials() -> int:
|
|
"""Re-encrypt every mailbox password with the first configured key."""
|
|
rotated = 0
|
|
for connection in MailboxConnection.objects.select_for_update().exclude(
|
|
provider=MailboxConnection.Provider.OUTLOOK
|
|
):
|
|
password = decrypt_mailbox_password(connection)
|
|
connection.encrypted_password = encrypt_mailbox_password(password)
|
|
connection.save(update_fields=["encrypted_password", "updated_at"])
|
|
rotated += 1
|
|
return rotated
|
|
|
|
|
|
@transaction.atomic
|
|
def save_mailbox_connection(
|
|
*,
|
|
user,
|
|
cleaned_data: dict[str, object],
|
|
instance: MailboxConnection | None = None,
|
|
) -> MailboxConnection:
|
|
connection = instance or MailboxConnection(user=user)
|
|
if connection.pk and connection.user_id != user.pk:
|
|
raise PermissionError("Mailboxkoppeling behoort niet tot deze gebruiker.")
|
|
|
|
password = str(cleaned_data.pop("password", "") or "")
|
|
for field in (
|
|
"platform",
|
|
"provider",
|
|
"custom_host",
|
|
"port",
|
|
"username",
|
|
"mailbox",
|
|
"enabled",
|
|
"poll_interval_minutes",
|
|
):
|
|
setattr(connection, field, cleaned_data[field])
|
|
uses_oauth = connection.provider == MailboxConnection.Provider.OUTLOOK
|
|
if uses_oauth:
|
|
connection.encrypted_password = ""
|
|
elif password:
|
|
connection.encrypted_password = encrypt_mailbox_password(password)
|
|
elif not connection.encrypted_password:
|
|
raise MailboxCredentialError("Een app-wachtwoord is verplicht.")
|
|
connection.next_poll_at = timezone.now()
|
|
connection.last_error_category = ""
|
|
connection.last_error_message = ""
|
|
connection.full_clean()
|
|
connection.save()
|
|
return connection
|
|
|
|
|
|
@transaction.atomic
|
|
def claim_mailbox_connection(
|
|
*, connection_id: int, worker_token: str | None = None, force: bool = False
|
|
) -> tuple[MailboxConnection, str] | None:
|
|
now = timezone.now()
|
|
connection = MailboxConnection.objects.select_for_update().get(pk=connection_id)
|
|
if not connection.enabled:
|
|
return None
|
|
if connection.lease_expires_at and connection.lease_expires_at > now:
|
|
return None
|
|
if not force and connection.next_poll_at > now:
|
|
return None
|
|
token = worker_token or uuid4().hex
|
|
connection.lease_token = token
|
|
connection.lease_expires_at = now + timedelta(minutes=5)
|
|
connection.save(update_fields=["lease_token", "lease_expires_at", "updated_at"])
|
|
return connection, token
|
|
|
|
|
|
@transaction.atomic
|
|
def finish_mailbox_poll(
|
|
*,
|
|
connection_id: int,
|
|
worker_token: str,
|
|
success: bool,
|
|
error_category: str = "",
|
|
error_message: str = "",
|
|
) -> None:
|
|
connection = MailboxConnection.objects.select_for_update().get(pk=connection_id)
|
|
if connection.lease_token != worker_token:
|
|
return
|
|
now = timezone.now()
|
|
connection.last_polled_at = now
|
|
connection.next_poll_at = now + timedelta(minutes=connection.poll_interval_minutes)
|
|
connection.lease_token = ""
|
|
connection.lease_expires_at = None
|
|
if success:
|
|
connection.last_success_at = now
|
|
connection.last_error_category = ""
|
|
connection.last_error_message = ""
|
|
else:
|
|
connection.last_error_category = error_category[:80]
|
|
connection.last_error_message = error_message[:500]
|
|
connection.save(
|
|
update_fields=[
|
|
"last_polled_at",
|
|
"next_poll_at",
|
|
"lease_token",
|
|
"lease_expires_at",
|
|
"last_success_at",
|
|
"last_error_category",
|
|
"last_error_message",
|
|
"updated_at",
|
|
]
|
|
)
|