chore: bump to 1.1.0 (logging-discipline audit)

Signed-off-by: disqualifier <dev@disqualifier.me>

.gitignore

@@ -1,5 +1,5 @@
 # claude
-CLAUDE.md
+.claude/
 
 # python
 __pycache__/

README.md

@@ -11,21 +11,23 @@ This reads codes from email; it does not generate them (that is `pyotp`'s job).
 `requirements.txt`:
 
 ```
-aiomail @ git+ssh://git@git.rethinkstudios.io/rethink-public/aiomail.git@v0.1.1
+aiomail @ git+ssh://git@git.rethinkstudios.io/rethink-public/aiomail.git@v1.0.0
 # OAuth token providers (Microsoft / Google) need the extra:
-aiomail[oauth] @ git+ssh://git@git.rethinkstudios.io/rethink-public/aiomail.git@v0.1.1
+aiomail[oauth] @ git+ssh://git@git.rethinkstudios.io/rethink-public/aiomail.git@v1.0.0
 ```
 
 Direct:
 
 ```bash
-pip install "aiomail @ git+ssh://git@git.rethinkstudios.io/rethink-public/aiomail.git@v0.1.1"
+pip install "aiomail @ git+ssh://git@git.rethinkstudios.io/rethink-public/aiomail.git@v1.0.0"
-pip install "aiomail[oauth] @ git+ssh://git@git.rethinkstudios.io/rethink-public/aiomail.git@v0.1.1"
+pip install "aiomail[oauth] @ git+ssh://git@git.rethinkstudios.io/rethink-public/aiomail.git@v1.0.0"
 ```
 
 Requires `aioimaplib` and `beautifulsoup4` (pulled transitively). The `oauth`
 extra adds `aiohttp` for the refresh-token providers.
 
+Drop the `@v1.0.0` suffix from the line above to install the latest unpinned.
+
 ## Password auth
 
 ```python
@@ -59,11 +61,18 @@ Credentials are always supplied by you — nothing is hardcoded.
 
 ```python
 import re
+from email.utils import parseaddr
 await retrieve_otp(client, sender="uber.com")                      # substring
 await retrieve_otp(client, sender=re.compile(r"no-?reply@.*\.io")) # regex
-await retrieve_otp(client, sender=lambda f: f.endswith("@x.com"))  # callable
+await retrieve_otp(client, sender=lambda f: parseaddr(f)[1].endswith("@x.com"))  # callable
 ```
 
+A real `From` header is `Name <addr@x.com>`, not a bare address — `parseaddr` pulls
+the address out before matching (a plain `f.endswith(...)` would never match).
+
+Subject headers are RFC2047-decoded before matching/extraction, so providers
+that encode non-ASCII subjects (`=?utf-8?B?...?=`) still match on plain text.
+
 Code extraction is tunable too — `patterns` (regexes, first group wins) and
 `lengths` (standalone digit-run fallback):
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "aiomail"
-version = "0.1.1"
+version = "1.1.0"
 description = "async IMAP one-time-code retrieval with password/OAuth2 auth and dynamic matching"
 requires-python = ">=3.10"
 dependencies = [

src/aiomail/__init__.py

@@ -1,9 +1,15 @@
-"""aiomail — async IMAP one-time-code retrieval.
+"""aiomail - async IMAP one-time-code retrieval, password or OAuth2 auth, dynamic matching.
 
-reads OTP / login codes out of IMAP mailboxes (accounts you own). supports plain
+facade over auth (PasswordAuth/OAuth2Auth), IMAPClient (connection lifecycle), and
-password and OAuth2 (XOAUTH2) auth, and dynamic sender/subject/code matching via
+retrieve_otp (folder-scan orchestration); see each module's docstring for detail.
-substrings, regexes, or callables.
+
+footguns: an IMAPClient instance is not safe for concurrent callers beyond its internal
+connect/reconnect lock; sequence-number ids from before a reconnect are invalid after
+(pass use_uid=True if ids must survive one); credentials are always caller-supplied via
+an injected Auth, never read from config.
 """
+from importlib.metadata import version, PackageNotFoundError
+
 from .auth import Auth, OAuth2Auth, PasswordAuth
 from .client import IMAPClient
 from .extract import (
@@ -11,6 +17,7 @@ from .extract import (
     DEFAULT_PATTERNS,
     MatchSpec,
     as_predicate,
+    decode_header_value,
     extract_code,
 )
 from .retrieve import DEFAULT_FOLDERS, retrieve_otp
@@ -22,6 +29,7 @@ __all__ = [
     "IMAPClient",
     "extract_code",
     "as_predicate",
+    "decode_header_value",
     "retrieve_otp",
     "MatchSpec",
     "DEFAULT_PATTERNS",
@@ -29,4 +37,7 @@ __all__ = [
     "DEFAULT_FOLDERS",
 ]
 
-__version__ = "0.1.1"
+try:
+    __version__ = version("aiomail")
+except PackageNotFoundError:
+    __version__ = "0.0.0+unknown"

src/aiomail/auth.py

@@ -1,18 +1,11 @@
-"""authentication mechanisms for the IMAP client.
+"""authentication mechanisms for the IMAP client: `PasswordAuth` (LOGIN), `OAuth2Auth` (XOAUTH2)."""
-
-an `Auth` is anything that can authenticate an already-connected aioimaplib
-client. two ship: `PasswordAuth` (plain LOGIN) and `OAuth2Auth` (XOAUTH2 using an
-access token, either passed directly or pulled from a token provider callable).
-credentials are always injected by the caller, never hardcoded here.
-"""
 import base64
 import logging
 from typing import Awaitable, Callable, Optional, Protocol, Union, runtime_checkable
 
 log = logging.getLogger(__name__)
 
-# a token provider is any (optionally async) callable returning a fresh access
+# see aiomail.oauth for ready-made Microsoft / Google providers
-# token string; see aiomail.oauth for ready-made Microsoft / Google providers
 TokenProvider = Callable[[], Union[str, Awaitable[str]]]
 
 
@@ -39,11 +32,7 @@ class PasswordAuth:
 
 
 def _as_str(token) -> str:
-    """coerce a token to str (a provider may hand back bytes)
+    """coerce a token to str (a provider may hand back bytes)"""
-
-    both XOAUTH2 entrypoints downstream need a str (one .encode()s it, the SASL
-    builder interpolates it), so normalize here rather than crashing on bytes.
-    """
     return token.decode() if isinstance(token, bytes) else token
 
 
@@ -54,12 +43,7 @@ def _sasl_xoauth2(user: str, token: str) -> str:
 
 
 class OAuth2Auth:
-    """XOAUTH2 auth using an access token or a token provider
+    """XOAUTH2 auth: a static `token`, or a `token_provider` (sync/async callable) fetched fresh at connect time"""
-
-    pass `token` for a static token, or `token_provider` (sync or async callable)
-    to fetch a fresh one at connect time — the provider path is what makes refresh
-    flows "easy": hand it a provider from aiomail.oauth and forget about it.
-    """
 
     def __init__(
         self,
@@ -85,15 +69,14 @@ class OAuth2Auth:
 
     async def authenticate(self, mail) -> None:
         token = await self._resolve_token()
-        # aioimaplib's mail.xoauth2(user, token) builds the SASL string by f-string
+        # mail.xoauth2(user, token) f-string-interpolates token, so it MUST be str -
-        # interpolating the token, so token MUST be str — passing bytes interpolates
+        # bytes would interpolate the b'...' repr and corrupt the Bearer value.
-        # the b'...' repr and corrupts the Bearer value. _resolve_token already
-        # returns str (via _as_str). clients lacking .xoauth2 are driven via the
-        # SASL callback from _sasl_xoauth2.
         xoauth2 = getattr(mail, "xoauth2", None)
         if xoauth2 is not None:
             result, data = await xoauth2(self.user, token)
         elif hasattr(mail, "authenticate"):
+            # escape hatch for a non-aioimaplib client (aioimaplib's IMAP4 always
+            # has .xoauth2, never .authenticate); untested against any real driver
             result, data = await mail.authenticate(
                 "XOAUTH2", lambda _: _sasl_xoauth2(self.user, token)
             )

src/aiomail/client.py

@@ -1,14 +1,15 @@
-"""async IMAP client wrapping aioimaplib.
+"""async IMAP client wrapping aioimaplib: connect/retry/reconnect/close plus folders/search/fetch/mark-seen.
 
-a thin, provider-agnostic client: it owns the connection lifecycle (connect with
+auth is injected. reconnect-on-stale re-selects the prior folder, but sequence-number ids from
-retries, reconnect-on-stale, close) and exposes the handful of operations the OTP
+before a reconnect are not valid after (a fresh SELECT can renumber the mailbox) - pass
-flow needs (folders, search, fetch, mark-seen). auth is injected, so the same
+`use_uid=True` if ids need to survive a reconnect. one instance is not safe for concurrent
-client serves password and OAuth accounts.
+callers beyond the internal connect/reconnect lock.
 """
 import asyncio
 import email
 import email.message
 import logging
+import re
 from typing import List, Optional
 
 from aioimaplib import IMAP4, IMAP4_SSL
@@ -17,13 +18,23 @@ from .auth import Auth
 
 log = logging.getLogger(__name__)
 
+# IMAP LIST reply: (flags) "<delim>" <name> - delim is server-defined (often "/" or
+# "." or NIL); capture the trailing name regardless, quoted or bare
+_LIST_RE = re.compile(rb'^\([^)]*\)\s+(?:"[^"]*"|NIL)\s+(.+)$')
+
+
+def _folder_name(raw: bytes) -> Optional[str]:
+    """extract the folder name from a LIST reply line, or None on no match"""
+    match = _LIST_RE.match(raw.strip())
+    if not match:
+        return None
+    return match.group(1).decode().strip().strip('"')
+
 
 class IMAPClient:
     """connection-managing IMAP client driven by an injected auth mechanism
 
-    note: `use_uid` selects UID vs sequence-number addressing and is independent
+    `use_uid` (UID vs sequence-number addressing) is independent of `use_ssl` - unrelated concerns.
-    of `use_ssl` — the two were conflated in an earlier draft (`use_uid = use_ssl`),
-    which is a bug; they are unrelated concerns.
     """
 
     def __init__(
@@ -45,6 +56,8 @@ class IMAPClient:
         self.timeout = timeout
         self.max_retries = max_retries
         self._mail = None
+        self._selected_folder: Optional[str] = None
+        self._lock = asyncio.Lock()
 
     async def __aenter__(self) -> "IMAPClient":
         await self.ensure_connection()
@@ -54,8 +67,16 @@ class IMAPClient:
         await self.close()
 
     async def connect(self) -> bool:
-        """open a connection and authenticate, retrying with linear backoff"""
+        """open a connection and authenticate, retrying with linear backoff; serialized by an internal lock"""
-        await self.close()
+        async with self._lock:
+            return await self._connect_locked()
+
+    async def _connect_locked(self) -> bool:
+        """connect()'s body; caller must hold self._lock"""
+        superseded, self._mail = self._mail, None
+        self._selected_folder = None
+        if superseded is not None:
+            await self._discard_mail(superseded)
         for attempt in range(self.max_retries):
             try:
                 if self.use_ssl:
@@ -68,40 +89,84 @@ class IMAPClient:
             except Exception as exc:
                 log.warning("connect attempt %d/%d failed: %s", attempt + 1, self.max_retries, exc)
                 if self._mail is not None:
+                    await self._discard_mail(self._mail)
+                    self._mail = None
+                if attempt < self.max_retries - 1:
+                    await asyncio.sleep(2 * (attempt + 1))
+        # terminal signal: all attempts are exhausted and this SWALLOWS the failure into a
+        # False return the caller branches on. without this line the per-attempt WARNINGs are
+        # the only trace, so a genuinely dead connection could look like routine noise - log
+        # the terminal exhaustion loudly so real degradation is visible, then return False.
+        log.warning("connect to %s failed after %d attempts", self.host, self.max_retries)
+        return False
+
+    @staticmethod
+    async def _discard_mail(mail) -> None:
+        """tear down a half-built IMAP4 without leaking its fire-and-forget connect task"""
+        task = getattr(mail, "_client_task", None)
+        if task is not None and not task.done():
+            task.cancel()
+        if task is not None:
+            # shield: a bare `await task` would also swallow an external cancel; under
+            # shield, CancelledError here means external only.
             try:
-                        await self._mail.logout()
+                await asyncio.shield(task)
+            except asyncio.CancelledError:
+                if not task.cancelled():
+                    raise
+            except Exception:
+                pass
+        try:
+            await mail.logout()
         except Exception as teardown:
             log.debug("logout error ignored during failed connect: %s", teardown)
-                    self._mail = None
-                await asyncio.sleep(2 * (attempt + 1))
-        return False
 
     async def close(self) -> None:
         """log out and drop the connection, swallowing teardown errors"""
+        async with self._lock:
+            await self._close_locked()
+
+    async def _close_locked(self) -> None:
+        """close()'s body; caller must hold self._lock"""
         if self._mail is not None:
+            mail, self._mail = self._mail, None
             try:
-                await self._mail.logout()
+                await mail.logout()
             except Exception as exc:
                 log.debug("logout error ignored: %s", exc)
-            self._mail = None
+        self._selected_folder = None
 
     async def ensure_connection(self) -> bool:
-        """return a live connection, reconnecting if the link is stale"""
+        """return a live, SELECTED-if-applicable connection, reconnecting (and re-selecting the prior
-        if self._mail is None:
+        folder) if the link is stale; see module docstring for the sequence-number-vs-use_uid caveat"""
-            return await self.connect()
+        async with self._lock:
+            if self._mail is not None:
                 try:
                     await self._mail.noop()
                     return True
                 except Exception:
-            return await self.connect()
+                    pass
+            return await self._connect_and_reselect_locked()
 
-    def is_throttled(self) -> bool:
+    async def _connect_and_reselect_locked(self) -> bool:
-        """best-effort detection of a provider throttling response"""
+        """connect() then re-select the previously-selected folder; caller must hold self._lock"""
-        return bool(
+        folder = self._selected_folder
-            self._mail is not None
+        if not await self._connect_locked():
-            and getattr(self._mail, "resp", None)
+            return False
-            and "THROTTLED" in str(self._mail.resp)
+        if folder is None:
-        )
+            return True
+        try:
+            result, _ = await self._mail.select(f'"{folder}"')
+        except Exception as exc:
+            log.debug("re-select %s after reconnect failed: %s", folder, exc)
+            self._selected_folder = None
+            return False
+        if result != "OK":
+            log.debug("re-select %s after reconnect failed: %s", folder, result)
+            self._selected_folder = None
+            return False
+        self._selected_folder = folder
+        return True
 
     async def get_folders(self) -> List[str]:
         """list mailbox folder names"""
@@ -115,21 +180,32 @@ class IMAPClient:
         folders: List[str] = []
         for folder in folder_list or []:
             try:
-                folders.append(folder.decode().split(' "/" ')[-1].strip('"'))
+                name = _folder_name(folder)
-            except Exception:
+            except Exception as exc:
+                log.debug("skipping unparseable folder entry %r: %s", folder, exc)
                 continue
+            if name is not None:
+                folders.append(name)
         return folders
 
     async def select(self, folder: str) -> bool:
-        """select a folder, returning whether it succeeded"""
+        """select a folder, returning whether it succeeded
-        if not await self.ensure_connection():
+
+        connects/reconnects first if needed; a failed re-select of the *previously*
+        selected folder during reconnect does not block attempting this call's own
+        target folder, since a live connection can still select it.
+        """
+        if not await self.ensure_connection() and self._mail is None:
             return False
         try:
             result, _ = await self._mail.select(f'"{folder}"')
-            return result == "OK"
         except Exception as exc:
             log.debug("select %s failed: %s", folder, exc)
             return False
+        if result == "OK":
+            self._selected_folder = folder
+            return True
+        return False
 
     async def search(self, query: str) -> List[int]:
         """search the selected folder, returning ids newest-first"""
@@ -148,8 +224,7 @@ class IMAPClient:
             try:
                 ids.append(int(token))
             except (TypeError, ValueError):
-                # tolerate a malformed/non-numeric token in the SEARCH response
+                # tolerate a malformed/non-numeric SEARCH token instead of crashing
-                # instead of crashing the whole search
                 log.debug("skipping non-numeric search token: %r", token)
         return sorted(set(ids), reverse=True)
 
@@ -169,19 +244,30 @@ class IMAPClient:
         if result != "OK" or not data:
             return None
         for item in data:
-            if isinstance(item, (bytes, bytearray)) and len(item) > 20:
+            # select by structure, not length: aioimaplib stores the literal payload
+            # as the only bytearray in the response, so a length heuristic would
+            # mismatch the header line for a 2+ digit id or a BODY[]/UID fetch
+            if isinstance(item, bytearray):
                 return email.message_from_bytes(bytes(item))
+            # cross-version fallback for an imaplib-style (header, payload) tuple;
+            # aioimaplib 2.0.x never yields one here
             if isinstance(item, tuple) and len(item) > 1:
                 return email.message_from_bytes(item[1])
         return None
 
     async def mark_seen(self, email_id: int) -> bool:
-        """flag a message as read without deleting it"""
+        """flag a message as read without deleting it
+
+        aioimaplib does not apply its own timeout to the UID STORE command path, so the
+        use_uid=True call is wrapped here to bound it the same as the non-uid path.
+        """
         if not await self.ensure_connection():
             return False
         try:
             if self.use_uid:
-                result, _ = await self._mail.uid("store", str(email_id), "+FLAGS", "(\\Seen)")
+                result, _ = await asyncio.wait_for(
+                    self._mail.uid("store", str(email_id), "+FLAGS", "(\\Seen)"), self.timeout
+                )
             else:
                 result, _ = await self._mail.store(str(email_id), "+FLAGS", "(\\Seen)")
             return result == "OK"

src/aiomail/extract.py

@@ -1,13 +1,9 @@
-"""code extraction and dynamic matching for email messages.
+"""code extraction and dynamic matching for email messages, pure logic with no network IO."""
-
-this module is pure logic with no network IO, so it is the easiest part to
-unit-test. `extract_code` pulls a one-time code out of a message; `as_predicate`
-turns a string / compiled regex / callable into a uniform match function used to
-filter senders and subjects.
-"""
 import email.message
 import logging
 import re
+from email.errors import HeaderParseError
+from email.header import decode_header, make_header
 from typing import Callable, Iterable, Iterator, Optional, Pattern, Sequence, Union
 
 from bs4 import BeautifulSoup
@@ -39,6 +35,21 @@ def _compile(patterns: Sequence[Union[str, Pattern]]) -> list[Pattern]:
     return out
 
 
+def decode_header_value(raw: str) -> str:
+    """decode an RFC2047 encoded-word header (=?charset?B/Q?...?=) to text, falling back to raw on failure
+
+    messages are parsed with the compat32 policy, which leaves encoded-word
+    headers raw; decode before scanning so matching sees real text.
+    """
+    if not raw:
+        return raw
+    try:
+        return str(make_header(decode_header(raw)))
+    except (UnicodeDecodeError, LookupError, ValueError, HeaderParseError) as exc:
+        log.debug("header decode failed (%s): %s", raw, exc)
+        return raw
+
+
 def _decode_part(part: email.message.Message) -> Optional[str]:
     """decode a single message part to text, tolerating bad charsets"""
     payload = part.get_payload(decode=True)
@@ -47,7 +58,7 @@ def _decode_part(part: email.message.Message) -> Optional[str]:
     charset = part.get_content_charset() or "utf-8"
     try:
         return payload.decode(charset, errors="replace")
-    except (LookupError, TypeError):
+    except (LookupError, TypeError, ValueError):
         return payload.decode("utf-8", errors="replace")
 
 
@@ -74,10 +85,9 @@ def _scan(text: str, patterns: list[Pattern], lengths: set[int]) -> Optional[str
         m = pat.search(text)
         if m:
             return m.group(1) if m.groups() else m.group(0)
-    for token in re.split(r"\s+", text):
+    for run in re.findall(r"\d+", text):
-        digits = "".join(c for c in token if c.isdigit())
+        if len(run) in lengths:
-        if digits and len(digits) in lengths and digits.isdigit():
+            return run
-            return digits
     return None
 
 
@@ -89,15 +99,13 @@ def extract_code(
 ) -> Optional[str]:
     """extract a one-time code from a message, subject first then body parts
 
-    `patterns` are regexes tried in order; the first capturing group wins, or the
+    `patterns` are regexes tried in order (first capturing group wins, else the whole
-    whole match if a pattern has no groups. when no pattern matches a block, any
+    match); if none hit, a standalone digit run whose length is in `lengths` is returned.
-    standalone digit run whose length is in `lengths` is returned. both knobs are
-    parameters so callers can tune per provider without forking this function.
     """
     compiled = _compile(patterns)
     length_set = set(lengths)
 
-    subject = message.get("Subject", "") or ""
+    subject = decode_header_value(message.get("Subject", "") or "")
     hit = _scan(subject, compiled, length_set)
     if hit:
         return hit
@@ -110,17 +118,14 @@ def extract_code(
 
 
 def as_predicate(spec: MatchSpec) -> Callable[[Optional[str]], bool]:
-    """normalize a match spec into a predicate over an optional string
+    """normalize a match spec (None, regex, callable, or substring) into a predicate over an optional string"""
-
-    accepts None (matches everything), a precompiled regex (search), a callable
-    (used directly), or a plain string (case-insensitive substring). this is what
-    gives sender/subject filtering its "string or regex, caller's choice" behavior.
-    """
     if spec is None:
         return lambda value: True
     if isinstance(spec, re.Pattern):
         return lambda value: bool(spec.search(value or ""))
     if callable(spec):
-        return spec
+        # coalesce None like the other branches so the Optional[str] contract holds
+        # even if the caller's callable assumes a real string
+        return lambda value: bool(spec(value or ""))
     needle = str(spec).lower()
     return lambda value: needle in (value or "").lower()

src/aiomail/oauth.py

@@ -1,11 +1,6 @@
-"""optional OAuth2 token providers (refresh-token -> access-token).
+"""optional OAuth2 token providers (refresh-token -> access-token) for `OAuth2Auth`, credentials always
-
+caller-supplied. aiohttp is an optional extra; missing it raises a clear error only when a provider is
-these turn a refresh token into a fresh access token for `OAuth2Auth`. they need
+instantiated, not on import."""
-aiohttp, kept as an optional extra so the core stays light; importing this module
-without aiohttp raises a clear error only when a provider is instantiated.
-
-credentials (client_id, refresh_token) are always supplied by the caller.
-"""
 import asyncio
 import logging
 import time
@@ -76,12 +71,21 @@ class _RefreshTokenProvider:
                     async with aiohttp.ClientSession(timeout=timeout) as session:
                         async with session.post(endpoint, data=data) as resp:
                             if resp.status == 200:
-                                token = (await resp.json()).get("access_token")
+                                # content_type=None: some endpoints return 200 as text/plain
+                                # or text/javascript; default json() would raise ContentTypeError
+                                body_json = await resp.json(content_type=None)
+                                token = body_json.get("access_token")
                                 if token:
                                     self._failures = 0
                                     return token
+                                # truncated, never whole: the body may carry sensitive material
+                                log.warning(
+                                    "token endpoint %s -> 200 with no access_token: %s",
+                                    endpoint, str(body_json)[:200],
+                                )
                             else:
-                                body = await resp.text()
+                                # truncated only - the body may carry sensitive material
+                                body = (await resp.text())[:200]
                                 log.warning("token endpoint %s -> %s: %s", endpoint, resp.status, body)
                 except Exception as exc:
                     log.warning("token request to %s failed: %s", endpoint, exc)

src/aiomail/retrieve.py

@@ -1,18 +1,20 @@
-"""orchestration: find the most recent valid OTP across folders.
+"""orchestration: `retrieve_otp` ties the client and extractor together to find the most recent valid OTP."""
-
-`retrieve_otp` ties the client and extractor together. sender/subject accept the
-flexible match specs from `extract`, folders/age/patterns are all parameters with
-sane defaults, and provider quirks live in the arguments rather than hardcoded
-branches inside the function.
-"""
 import asyncio
 import logging
 import time
+from datetime import timezone
 from email.utils import parsedate_to_datetime
 from typing import Iterable, List, Optional, Pattern, Sequence, Union
 
 from .client import IMAPClient
-from .extract import DEFAULT_LENGTHS, DEFAULT_PATTERNS, MatchSpec, as_predicate, extract_code
+from .extract import (
+    DEFAULT_LENGTHS,
+    DEFAULT_PATTERNS,
+    MatchSpec,
+    as_predicate,
+    decode_header_value,
+    extract_code,
+)
 
 log = logging.getLogger(__name__)
 
@@ -20,15 +22,13 @@ log = logging.getLogger(__name__)
 DEFAULT_FOLDERS: Sequence[str] = ("INBOX", "Junk", "Spam", "Archive", "All Mail")
 
 
-def _server_query(sender: MatchSpec, subject: MatchSpec) -> str:
+def _server_query(sender: MatchSpec, subject: MatchSpec, match_field: str = "from") -> str:
-    """build a narrowing IMAP query from plain-string specs only
+    """build a narrowing IMAP query from plain-string specs, falling back to ALL for regex/callable specs"""
-
-    only plain strings translate to server-side FROM/SUBJECT filters; regex and
-    callable specs fall back to ALL and are filtered client-side, so dynamic
-    matching always works even when the server cannot express it.
-    """
     parts: List[str] = []
     if isinstance(sender, str):
+        if match_field == "to":
+            parts.append(f'OR TO "{sender}" FROM "{sender}"')
+        else:
             parts.append(f'FROM "{sender}"')
     if isinstance(subject, str):
         parts.append(f'SUBJECT "{subject}"')
@@ -41,7 +41,12 @@ def _age_seconds(message) -> Optional[float]:
     if not raw:
         return None
     try:
-        return time.time() - parsedate_to_datetime(raw).timestamp()
+        dt = parsedate_to_datetime(raw)
+        if dt.tzinfo is None:
+            # naive datetime means a "-0000" zone (RFC 2822: unspecified offset);
+            # force UTC rather than letting .timestamp() read it as local
+            dt = dt.replace(tzinfo=timezone.utc)
+        return time.time() - dt.timestamp()
     except (TypeError, ValueError) as exc:
         log.debug("date parse failed (%s): %s", raw, exc)
         return None
@@ -52,6 +57,7 @@ async def retrieve_otp(
     *,
     sender: MatchSpec = None,
     subject: MatchSpec = None,
+    match_field: str = "from",
     folders: Optional[Iterable[str]] = None,
     patterns: Sequence[Union[str, Pattern]] = DEFAULT_PATTERNS,
     lengths: Iterable[int] = DEFAULT_LENGTHS,
@@ -64,14 +70,15 @@ async def retrieve_otp(
 ) -> Optional[str]:
     """return the newest OTP matching the filters, or None
 
-    sender/subject accept a substring, a compiled regex, or a callable. folders,
+    sender/subject accept a substring, a compiled regex, or a callable.
-    patterns, code lengths, max age and retry behavior are all tunable. set
+    `match_field="to"` matches the recipient address and additionally accepts a
-    `max_age=None` to disable the freshness check.
+    forwarded match on From. set `max_age=None` to disable the freshness check.
     """
     folders = list(folders) if folders is not None else list(DEFAULT_FOLDERS)
+    lengths = list(lengths)
     sender_ok = as_predicate(sender)
     subject_ok = as_predicate(subject)
-    query = _server_query(sender, subject)
+    query = _server_query(sender, subject, match_field)
 
     for attempt in range(retries + 1):
         for folder in folders:
@@ -90,8 +97,13 @@ async def retrieve_otp(
                     continue
 
                 from_hdr = message.get("From", "")
-                subj_hdr = message.get("Subject", "")
+                subj_hdr = decode_header_value(message.get("Subject", ""))
-                if not sender_ok(from_hdr) or not subject_ok(subj_hdr):
+                if match_field == "to":
+                    to_hdr = message.get("To", "")
+                    matched = sender_ok(to_hdr) or sender_ok(from_hdr)
+                else:
+                    matched = sender_ok(from_hdr)
+                if not matched or not subject_ok(subj_hdr):
                     continue
 
                 code = extract_code(message, patterns=patterns, lengths=lengths)