webmail-client/backend/app/providers/imap_smtp/parsing.py

"""Pure parsing helpers for IMAP responses (unit-testable, no I/O).

aioimaplib returns responses as lists of ``bytes``/``bytearray`` lines. A FETCH
with ``BODY[]`` yields, per message, a header line ending ``... {<size>}`` followed
by a ``bytearray`` literal, then a lone ``)``. LIST lines carry SPECIAL-USE flags.
"""

from __future__ import annotations

import re
from dataclasses import dataclass
from datetime import datetime

from imapclient.imap_utf7 import decode as imap_utf7_decode
from imapclient.imap_utf7 import encode as imap_utf7_encode

# --- SPECIAL-USE mapping (SPEC §4 special_use values) -----------------------

_SPECIAL_USE_FLAGS = {
    "\\sent": "sent",
    "\\drafts": "drafts",
    "\\trash": "trash",
    "\\junk": "junk",
    "\\archive": "archive",
    "\\all": "archive",  # Gmail "All Mail" acts as archive
}

_LIST_RE = re.compile(rb'^\(([^)]*)\)\s+(?:"([^"]*)"|NIL)\s+(.+)$')
_UID_RE = re.compile(rb"UID\s+(\d+)")
_FLAGS_RE = re.compile(rb"FLAGS\s+\(([^)]*)\)")
_GM_LABELS_RE = re.compile(rb"X-GM-LABELS\s+\(([^)]*)\)")
_INTERNALDATE_RE = re.compile(rb'INTERNALDATE\s+"([^"]+)"')
_LITERAL_RE = re.compile(rb"\{(\d+)\}\s*$")
_FETCH_HEADER_RE = re.compile(rb"^\d+\s+FETCH\s+\(", re.IGNORECASE)


@dataclass(frozen=True)
class ParsedMailbox:
    path: str
    special_use: str | None


@dataclass(frozen=True)
class ParsedFetch:
    uid: int
    flags: frozenset[str]
    internal_date: datetime | None
    raw: bytes | None  # None when the fetch did not request the body
    gmail_labels: tuple[str, ...] = ()  # custom Gmail labels (X-GM-LABELS), system ones dropped


def _decode_mailbox_name(raw: bytes) -> str:
    name = raw.strip()
    if name.startswith(b'"') and name.endswith(b'"'):
        name = name[1:-1]
    # IMAP mailbox names are modified UTF-7 (SPEC §4: store UTF-7 decoded).
    return str(imap_utf7_decode(bytes(name)))


def encode_mailbox_name(name: str) -> str:
    """Modified UTF-7 wire form of a mailbox name (inverse of ``_decode_mailbox_name``).

    Mailbox names are stored/displayed as Unicode but IMAP commands
    (SELECT/STATUS/APPEND) require them mUTF-7 encoded. ASCII names are unchanged;
    non-ASCII names — e.g. Gmail's localized ``[Gmail]/Wysłane`` — get encoded so the
    server can find them."""
    return str(imap_utf7_encode(name).decode("ascii"))


def parse_list_line(line: bytes) -> ParsedMailbox | None:
    """Parse one LIST response line into a :class:`ParsedMailbox` (or None).

    Non-selectable mailboxes are skipped (return None): ``\\Noselect`` /
    ``\\NonExistent`` names are pure hierarchy containers that hold no messages and
    cannot be SELECTed — e.g. Gmail's ``[Gmail]`` root or a parent label like
    ``Serial numbers`` that only groups child labels."""
    m = _LIST_RE.match(line.strip())
    if not m:
        return None
    flags = m.group(1).lower().split()
    if b"\\noselect" in flags or b"\\nonexistent" in flags:
        return None
    name = _decode_mailbox_name(m.group(3))
    special: str | None = None
    for flag in flags:
        mapped = _SPECIAL_USE_FLAGS.get(flag.decode("ascii", "ignore"))
        if mapped:
            special = mapped
            break
    if special is None and name.upper() == "INBOX":
        special = "inbox"
    return ParsedMailbox(path=name, special_use=special)


def parse_internaldate(raw: bytes | str) -> datetime | None:
    """Parse an IMAP INTERNALDATE (``d-Mon-Y H:M:S +ZZZZ``) to an aware datetime."""
    text = raw.decode("ascii", "ignore") if isinstance(raw, bytes) else raw
    text = text.strip()
    try:
        return datetime.strptime(text, "%d-%b-%Y %H:%M:%S %z")
    except ValueError:
        return None


def _parse_flag_token(tok: bytes) -> str:
    return tok.decode("ascii", "ignore")


def parse_fetch_header(line: bytes) -> tuple[int | None, frozenset[str], datetime | None]:
    """Extract (uid, flags, internaldate) from a FETCH header line."""
    uid_m = _UID_RE.search(line)
    uid = int(uid_m.group(1)) if uid_m else None
    flags_m = _FLAGS_RE.search(line)
    if flags_m:
        flags = frozenset(_parse_flag_token(t) for t in flags_m.group(1).split())
    else:
        flags = frozenset()
    date_m = _INTERNALDATE_RE.search(line)
    internal = parse_internaldate(date_m.group(1)) if date_m else None
    return uid, flags, internal


def _tokenize_gm_labels(blob: bytes) -> list[bytes]:
    """Split an ``X-GM-LABELS (...)`` body into raw tokens, keeping quoted strings
    (which may contain spaces) whole. Escaped chars inside quotes are unescaped."""
    tokens: list[bytes] = []
    i, n = 0, len(blob)
    while i < n:
        ch = blob[i : i + 1]
        if ch in (b" ", b"\t"):
            i += 1
            continue
        if ch == b'"':
            j = i + 1
            buf = bytearray()
            while j < n and blob[j : j + 1] != b'"':
                if blob[j : j + 1] == b"\\" and j + 1 < n:
                    buf += blob[j + 1 : j + 2]
                    j += 2
                    continue
                buf += blob[j : j + 1]
                j += 1
            tokens.append(bytes(buf))
            i = j + 1
        else:
            j = i
            while j < n and blob[j : j + 1] not in (b" ", b"\t"):
                j += 1
            tokens.append(blob[i:j])
            i = j
    return tokens


def parse_gm_labels(header: bytes) -> tuple[str, ...]:
    """Custom Gmail labels from a FETCH header's ``X-GM-LABELS`` (SPEC: Gmail sync).

    System labels (``\\Inbox``, ``\\Sent``, ``\\Important``, ``\\Starred``, …) are
    dropped — only user labels become chips. Names are modified-UTF-7 decoded, like
    mailbox names, so e.g. ``Zg&AAI...`` → the real Unicode label."""
    m = _GM_LABELS_RE.search(header)
    if not m:
        return ()
    out: list[str] = []
    for tok in _tokenize_gm_labels(m.group(1)):
        if not tok or tok.startswith(b"\\"):  # system label
            continue
        try:
            out.append(str(imap_utf7_decode(tok)))
        except Exception:  # noqa: BLE001 - fall back to a best-effort ASCII decode
            out.append(tok.decode("utf-8", "ignore"))
    return tuple(out)


def parse_fetch_lines(lines: list[bytes | bytearray]) -> list[ParsedFetch]:
    """Walk a full FETCH response, returning one :class:`ParsedFetch` per message.

    Handles both metadata-only fetches (FLAGS/INTERNALDATE) and body fetches
    (a ``{size}`` literal announced on the header line, body on the next line).
    """
    results: list[ParsedFetch] = []
    i = 0
    n = len(lines)
    while i < n:
        line = lines[i]
        if not isinstance(line, (bytes, bytearray)) or not _FETCH_HEADER_RE.match(bytes(line)):
            i += 1
            continue
        header = bytes(line)
        uid, flags, internal = parse_fetch_header(header)
        raw: bytes | None = None
        if _LITERAL_RE.search(header) and i + 1 < n:
            body = lines[i + 1]
            raw = bytes(body)
            i += 2
            # Skip the trailing ")" continuation line if present.
            if (
                i < n
                and isinstance(lines[i], (bytes, bytearray))
                and bytes(lines[i]).strip() == b")"
            ):
                i += 1
        else:
            i += 1
        if uid is not None:
            results.append(
                ParsedFetch(
                    uid=uid,
                    flags=flags,
                    internal_date=internal,
                    raw=raw,
                    gmail_labels=parse_gm_labels(header),
                )
            )
    return results


_VANISHED_RE = re.compile(rb"^VANISHED\s+(?:\(EARLIER\)\s+)?(\S+)", re.IGNORECASE)


def _expand_uid_set(token: bytes) -> list[int]:
    """Expand an IMAP uid-set (``1,3:5,9``) to a flat list of ints (``*`` skipped)."""
    out: list[int] = []
    for part in token.split(b","):
        part = part.strip()
        lo_b, sep, hi_b = part.partition(b":")
        if sep:
            if lo_b.isdigit() and hi_b.isdigit():
                lo, hi = int(lo_b), int(hi_b)
                out.extend(range(min(lo, hi), max(lo, hi) + 1))
        elif part.isdigit():
            out.append(int(part))
    return out


def parse_vanished(lines: list[bytes | bytearray]) -> list[int]:
    """UIDs from QRESYNC ``VANISHED [(EARLIER)] <uid-set>`` responses (RFC 7162).

    aioimaplib strips the leading ``* `` before buffering the line, so we match on
    the bare ``VANISHED`` token. Both the SELECT-time ``(EARLIER)`` form and the
    live-expunge form are accepted.
    """
    out: list[int] = []
    for line in lines:
        if not isinstance(line, (bytes, bytearray)):
            continue
        m = _VANISHED_RE.match(bytes(line).strip())
        if m:
            out.extend(_expand_uid_set(m.group(1)))
    return out


def parse_search_uids(lines: list[bytes | bytearray]) -> list[int]:
    """Parse a UID SEARCH response's first line into a list of ints."""
    if not lines:
        return []
    first = lines[0]
    if not isinstance(first, (bytes, bytearray)):
        return []
    return [int(tok) for tok in bytes(first).split() if tok.isdigit()]


def parse_select_untagged(lines: list[bytes | bytearray]) -> dict[str, int]:
    """Extract UIDVALIDITY / UIDNEXT / HIGHESTMODSEQ / EXISTS from SELECT lines."""
    out: dict[str, int] = {}
    patterns = {
        "uidvalidity": re.compile(rb"UIDVALIDITY\s+(\d+)"),
        "uidnext": re.compile(rb"UIDNEXT\s+(\d+)"),
        "highestmodseq": re.compile(rb"HIGHESTMODSEQ\s+(\d+)"),
        "exists": re.compile(rb"^(\d+)\s+EXISTS"),
    }
    for line in lines:
        if not isinstance(line, (bytes, bytearray)):
            continue
        b = bytes(line)
        for key, pat in patterns.items():
            m = pat.search(b)
            if m:
                out[key] = int(m.group(1))
    return out