299 lines
12 KiB
Python
299 lines
12 KiB
Python
"""Cliente IMAP de bajo nivel usado por el migrador.
|
|
|
|
Se apoya únicamente en ``imaplib`` (librería estándar). Todas las
|
|
operaciones que pueden fallar por problemas de red pasan por
|
|
:meth:`ImapClient.with_retry`, que reintenta y reconecta automáticamente.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import dataclasses
|
|
import imaplib
|
|
import re
|
|
import socket
|
|
import ssl
|
|
import time
|
|
|
|
import utf7
|
|
|
|
|
|
class ImapClientError(Exception):
|
|
pass
|
|
|
|
|
|
class ImapConnectionError(ImapClientError):
|
|
pass
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class FolderInfo:
|
|
name: str
|
|
delimiter: str
|
|
flags: str
|
|
|
|
|
|
def build_ssl_context(verify: bool) -> ssl.SSLContext:
|
|
ctx = ssl.create_default_context()
|
|
if not verify:
|
|
ctx.check_hostname = False
|
|
ctx.verify_mode = ssl.CERT_NONE
|
|
return ctx
|
|
|
|
|
|
class _IMAP4SSLCustom(imaplib.IMAP4_SSL):
|
|
"""IMAP4_SSL que permite conectar a una IP validando el certificado contra un hostname distinto."""
|
|
|
|
def __init__(self, connect_host, tls_hostname, port, ssl_context, timeout):
|
|
self._tls_hostname = tls_hostname
|
|
super().__init__(host=connect_host, port=port, ssl_context=ssl_context, timeout=timeout)
|
|
|
|
def _create_socket(self, timeout):
|
|
address = (self.host, self.port)
|
|
if timeout is not None:
|
|
sock = socket.create_connection(address, timeout)
|
|
else:
|
|
sock = socket.create_connection(address)
|
|
return self.ssl_context.wrap_socket(sock, server_hostname=self._tls_hostname)
|
|
|
|
|
|
_LIST_RE = re.compile(r'^\((?P<flags>[^)]*)\)\s+(?:"(?P<delim>[^"]*)"|NIL)\s+(?P<name>.+)$')
|
|
_FETCH_META_RE = re.compile(
|
|
r"UID (?P<uid>\d+)"
|
|
r"|RFC822\.SIZE (?P<size>\d+)"
|
|
r'|INTERNALDATE "(?P<idate>[^"]*)"'
|
|
r"|FLAGS \((?P<flags>[^)]*)\)"
|
|
)
|
|
_MESSAGE_ID_RE = re.compile(rb"^Message-ID:\s*(.+)$", re.IGNORECASE | re.MULTILINE)
|
|
|
|
|
|
class ImapClient:
|
|
def __init__(self, cfg, timeout: int, logger, retries: int = 3):
|
|
self.cfg = cfg
|
|
self.timeout = timeout
|
|
self.logger = logger
|
|
self.retries = max(1, retries)
|
|
self.conn = None
|
|
self._selected_folder = None
|
|
|
|
# -- conexión -----------------------------------------------------
|
|
|
|
def connect(self) -> None:
|
|
connect_host = self.cfg.connect_host
|
|
tls_hostname = self.cfg.tls_hostname
|
|
if self.cfg.use_ssl:
|
|
ctx = build_ssl_context(self.cfg.verify_tls)
|
|
if not self.cfg.verify_tls:
|
|
self.logger.warning(
|
|
f"[{self.cfg.label}] Verificación de certificado TLS DESHABILITADA para "
|
|
f"{self.cfg.host}. Esto es inseguro y solo debe usarse con servidores de confianza."
|
|
)
|
|
self.conn = _IMAP4SSLCustom(connect_host, tls_hostname, self.cfg.port, ctx, self.timeout)
|
|
else:
|
|
self.logger.warning(
|
|
f"[{self.cfg.label}] Conexión SIN SSL/TLS a {self.cfg.host}:{self.cfg.port}. "
|
|
"Las credenciales y los datos viajarán sin cifrar."
|
|
)
|
|
self.conn = imaplib.IMAP4(connect_host, self.cfg.port, timeout=self.timeout)
|
|
self.conn.login(self.cfg.user, self.cfg.password)
|
|
self._selected_folder = None
|
|
self.logger.info(f"[{self.cfg.label}] Conectado y autenticado como {self.cfg.user} en {self.cfg.host}:{self.cfg.port}")
|
|
|
|
def close(self) -> None:
|
|
if self.conn is not None:
|
|
try:
|
|
self.conn.logout()
|
|
except Exception:
|
|
pass
|
|
self.conn = None
|
|
|
|
def _reconnect(self) -> None:
|
|
self.logger.info(f"[{self.cfg.label}] Reintentando conexión a {self.cfg.host}...")
|
|
try:
|
|
if self.conn is not None:
|
|
self.conn.shutdown() if hasattr(self.conn, "shutdown") else None
|
|
except Exception:
|
|
pass
|
|
self.conn = None
|
|
self.connect()
|
|
|
|
def with_retry(self, func, *args, **kwargs):
|
|
last_exc = None
|
|
for attempt in range(1, self.retries + 1):
|
|
try:
|
|
return func(*args, **kwargs)
|
|
except (imaplib.IMAP4.abort, imaplib.IMAP4.error, OSError, ssl.SSLError, EOFError) as exc:
|
|
last_exc = exc
|
|
self.logger.warning(f"[{self.cfg.label}] Error IMAP (intento {attempt}/{self.retries}): {exc}")
|
|
if attempt < self.retries:
|
|
try:
|
|
self._reconnect()
|
|
if self._selected_folder:
|
|
self.select(self._selected_folder, readonly=True)
|
|
except Exception as reconnect_exc:
|
|
last_exc = reconnect_exc
|
|
self.logger.warning(f"[{self.cfg.label}] Fallo al reconectar: {reconnect_exc}")
|
|
time.sleep(min(2 ** attempt, 30))
|
|
raise ImapConnectionError(f"[{self.cfg.label}] Fallo tras {self.retries} intento(s): {last_exc}")
|
|
|
|
@staticmethod
|
|
def _quote(s: str) -> str:
|
|
escaped = s.replace("\\", "\\\\").replace('"', '\\"')
|
|
return f'"{escaped}"'
|
|
|
|
# -- carpetas -------------------------------------------------------
|
|
|
|
def list_folders(self) -> list[FolderInfo]:
|
|
def _do():
|
|
typ, data = self.conn.list()
|
|
if typ != "OK":
|
|
raise ImapClientError(f"LIST falló: {data}")
|
|
folders = []
|
|
for line in data:
|
|
if line is None:
|
|
continue
|
|
decoded = line.decode("utf-8", "replace") if isinstance(line, bytes) else line
|
|
m = _LIST_RE.match(decoded.strip())
|
|
if not m:
|
|
continue
|
|
flags = m.group("flags") or ""
|
|
delim = m.group("delim") or "/"
|
|
raw_name = m.group("name").strip()
|
|
if raw_name.startswith('"') and raw_name.endswith('"'):
|
|
raw_name = raw_name[1:-1].replace('\\"', '"').replace("\\\\", "\\")
|
|
try:
|
|
name = utf7.decode(raw_name.encode("ascii"))
|
|
except Exception:
|
|
name = raw_name
|
|
folders.append(FolderInfo(name=name, delimiter=delim, flags=flags))
|
|
return folders
|
|
|
|
return self.with_retry(_do)
|
|
|
|
def create_folder(self, name: str) -> None:
|
|
def _do():
|
|
encoded = utf7.encode(name).decode("ascii")
|
|
typ, data = self.conn.create(self._quote(encoded))
|
|
if typ != "OK":
|
|
msg = b" ".join(x for x in data if x).decode("utf-8", "replace") if data else ""
|
|
if "exist" not in msg.lower() and "duplicate" not in msg.lower():
|
|
raise ImapClientError(f"No se pudo crear la carpeta {name!r}: {msg}")
|
|
|
|
return self.with_retry(_do)
|
|
|
|
def select(self, name: str, readonly: bool = True) -> int:
|
|
def _do():
|
|
encoded = utf7.encode(name).decode("ascii")
|
|
typ, data = self.conn.select(self._quote(encoded), readonly=readonly)
|
|
if typ != "OK":
|
|
raise ImapClientError(f"No se pudo seleccionar la carpeta {name!r}: {data}")
|
|
self._selected_folder = name
|
|
return int(data[0]) if data and data[0] else 0
|
|
|
|
return self.with_retry(_do)
|
|
|
|
def get_uidvalidity(self, name: str):
|
|
def _do():
|
|
encoded = utf7.encode(name).decode("ascii")
|
|
typ, data = self.conn.status(self._quote(encoded), "(UIDVALIDITY)")
|
|
if typ != "OK" or not data or data[0] is None:
|
|
return None
|
|
m = re.search(rb"UIDVALIDITY (\d+)", data[0])
|
|
return int(m.group(1)) if m else None
|
|
|
|
return self.with_retry(_do)
|
|
|
|
# -- mensajes ---------------------------------------------------------
|
|
|
|
def search_uids(self, since_uid: int | None = None) -> list[int]:
|
|
def _do():
|
|
if since_uid:
|
|
typ, data = self.conn.uid("search", None, "UID", f"{since_uid}:*")
|
|
else:
|
|
typ, data = self.conn.uid("search", None, "ALL")
|
|
if typ != "OK":
|
|
raise ImapClientError(f"UID SEARCH falló: {data}")
|
|
if not data or not data[0]:
|
|
return []
|
|
uids = [int(x) for x in data[0].split()]
|
|
if since_uid:
|
|
uids = [u for u in uids if u >= since_uid]
|
|
return uids
|
|
|
|
return self.with_retry(_do)
|
|
|
|
def fetch_headers(self, uids: list[int], batch_size: int = 500) -> dict[int, dict]:
|
|
results: dict[int, dict] = {}
|
|
for i in range(0, len(uids), batch_size):
|
|
batch = uids[i:i + batch_size]
|
|
uid_set = ",".join(str(u) for u in batch)
|
|
|
|
def _do(uid_set=uid_set):
|
|
typ, data = self.conn.uid(
|
|
"fetch",
|
|
uid_set,
|
|
"(UID FLAGS INTERNALDATE RFC822.SIZE BODY.PEEK[HEADER.FIELDS (MESSAGE-ID)])",
|
|
)
|
|
if typ != "OK":
|
|
raise ImapClientError(f"FETCH de encabezados falló: {data}")
|
|
return data
|
|
|
|
data = self.with_retry(_do)
|
|
idx = 0
|
|
while idx < len(data):
|
|
item = data[idx]
|
|
if isinstance(item, tuple):
|
|
meta_bytes, literal = item
|
|
meta = meta_bytes.decode("utf-8", "replace")
|
|
info: dict = {}
|
|
for m in _FETCH_META_RE.finditer(meta):
|
|
if m.group("uid"):
|
|
info["uid"] = int(m.group("uid"))
|
|
elif m.group("size"):
|
|
info["size"] = int(m.group("size"))
|
|
elif m.group("idate") is not None:
|
|
info["internaldate_raw"] = m.group("idate")
|
|
elif m.group("flags") is not None:
|
|
info["flags"] = m.group("flags")
|
|
mid_match = _MESSAGE_ID_RE.search(literal or b"")
|
|
info["message_id"] = mid_match.group(1).decode("utf-8", "replace").strip() if mid_match else None
|
|
if "uid" in info:
|
|
results[info["uid"]] = info
|
|
idx += 1
|
|
if idx < len(data) and isinstance(data[idx], (bytes, bytearray)):
|
|
idx += 1
|
|
else:
|
|
idx += 1
|
|
return results
|
|
|
|
def fetch_full(self, uid: int):
|
|
def _do():
|
|
typ, data = self.conn.uid("fetch", str(uid), "(FLAGS INTERNALDATE RFC822)")
|
|
if typ != "OK" or not data or data[0] is None:
|
|
raise ImapClientError(f"FETCH RFC822 falló para UID {uid}: {data}")
|
|
meta_bytes, literal = data[0]
|
|
meta = meta_bytes.decode("utf-8", "replace")
|
|
flags_m = re.search(r"FLAGS \(([^)]*)\)", meta)
|
|
idate_m = re.search(r'INTERNALDATE "([^"]*)"', meta)
|
|
flags = flags_m.group(1) if flags_m else ""
|
|
internaldate_raw = idate_m.group(1) if idate_m else None
|
|
return literal, flags, internaldate_raw
|
|
|
|
return self.with_retry(_do)
|
|
|
|
def append(self, folder: str, message_bytes: bytes, flags: str, internaldate_raw: str | None):
|
|
def _do():
|
|
encoded = utf7.encode(folder).decode("ascii")
|
|
flags_str = f"({flags})" if flags else None
|
|
date_arg = f'"{internaldate_raw}"' if internaldate_raw else None
|
|
typ, data = self.conn.append(self._quote(encoded), flags_str, date_arg, message_bytes)
|
|
if typ != "OK":
|
|
raise ImapClientError(f"APPEND falló en {folder!r}: {data}")
|
|
uid = None
|
|
if data and data[0]:
|
|
m = re.search(rb"APPENDUID \d+ (\d+)", data[0])
|
|
if m:
|
|
uid = int(m.group(1))
|
|
return uid
|
|
|
|
return self.with_retry(_do)
|