6 Commits
Author SHA1 Message Date
dsql d478ed0d4c fix: warn on dropped OAuth token, skip wasted final backoff, honor external cancel, fix digit-run split (v0.1.8)
aiomail-7: log a truncated body when a 200 token response has no access_token
instead of silently discarding it. aiomail-8: connect() no longer sleeps after
the last failed retry attempt. aiomail-9: _discard_mail now distinguishes its
own task.cancel() from an external cancellation via asyncio.shield, so
connect() no longer resists cancellation. aiomail-10: the digit-run fallback
now matches contiguous \d+ runs instead of mashing digits across punctuation
within a whitespace token (e.g. a date). aiomail-11: README's dynamic-matching
example uses email.utils.parseaddr since a real From header is not a bare
address. Also compresses essay docstrings to one-line-plus-nuance with zero
behavior change (re-verified against the same negative controls).

Signed-off-by: disqualifier <dev@disqualifier.me>
2026-07-02 23:30:26 -04:00
dsql fd789c2ba2 fix: decode RFC2047 subjects, OR TO/FROM in to-mode query, tz-aware Date, drop LIST completion line (v0.1.7)
Subject headers were matched/scanned raw (never RFC2047-decoded), so any
provider that base64/quoted-printable encodes non-ASCII subjects silently
failed subject filtering and code extraction. match_field='to' only
server-queried TO, so an older TO-matching message permanently hid a
forwarded-From match the client-side check was documented to accept.
_age_seconds treated a naive datetime (from a "-0000" Date header) as local
time instead of UTC, corrupting max_age freshness on non-UTC hosts. get_folders
appended aioimaplib's tagged LIST completion text as a phantom folder name on
every call against any RFC-compliant server.

Also corrects __init__.py's __version__, which was left at 0.1.5 through the
0.1.6 release.

Signed-off-by: disqualifier <dev@disqualifier.me>
2026-07-02 17:02:56 -04:00
dsql a340067048 fix: reconnect-on-stale re-selects folder; lock connect() against races
ensure_connection() reconnected via connect(), which only reaches IMAP
state AUTH, not SELECTED - every search/fetch/store after a mid-pass
link drop silently failed (swallowed into []/None/False) because
aioimaplib rejects those commands outside SELECTED. Track the
currently-selected folder and re-select it after reconnecting; note
pre-reconnect sequence-number ids are invalid post-reconnect unless
use_uid=True.

Also add an asyncio.Lock around connect()/close()/ensure_connection():
concurrent callers on one instance used to race inside connect(),
where task B's `await self.close()` tore down task A's mid-handshake
connection, leaking sockets and orphaning server sessions. Superseded
connections are now always logged out instead of silently overwritten.

Bump to 0.1.6, update README install pins.

Signed-off-by: disqualifier <dev@disqualifier.me>
2026-07-02 16:40:50 -04:00
dsql b00f122b74 chore: ignore .claude/ dir (CLAUDE.md now lives under .claude/)
Signed-off-by: disqualifier <dev@disqualifier.me>
2026-06-29 21:55:13 -04:00
dsql 3da833f2fc fix: revert OTP-in-logs (spent on arrival, not a secret); F1 ContentTypeError, F5 callable None-guard
revert the M-1 log change — a single-use OTP is consumed on arrival, not a live secret,
so log the code value again. keep the oauth error-body truncate.

F1: oauth token fetch uses resp.json(content_type=None) so a 200 with text/plain doesn't
ContentTypeError and discard a valid token. F5: as_predicate coalesces None for the
callable branch like the string/regex branches. drop a redundant digits.isdigit().

Signed-off-by: disqualifier <dev@disqualifier.me>
2026-06-29 21:34:35 -04:00
dsql f940641a5a fix: never log the OTP code value (secret-in-logs); correct false test claim (v0.1.5)
M-1: retrieve.py logged the live single-use code at INFO ('found code %s', 'code %s
skipped too old'), shipping the secret to any aggregation/retention sink the host wires
(our /srv/logs -> loki/grafana path). drop the code value from both lines — log that a
code was found/retrieved and where, never the value. also truncate the oauth token-endpoint
error body to 200 chars so a token response can't be dumped whole.

aiomail-F3: CLAUDE.md claimed an '8-case tested' suite that does not exist in the repo;
corrected to describe the manual throwaway-venv exercise + the real flake8 check.

verified by execution: code retrieved, value absent from logs; control confirms the old
line carried it.

Signed-off-by: disqualifier <dev@disqualifier.me>
2026-06-29 20:46:51 -04:00
9 changed files with 200 additions and 125 deletions
+1 -1
View File
@@ -1,5 +1,5 @@
# claude # claude
CLAUDE.md .claude/
# python # python
__pycache__/ __pycache__/
+13 -6
View File
@@ -11,22 +11,22 @@ This reads codes from email; it does not generate them (that is `pyotp`'s job).
`requirements.txt`: `requirements.txt`:
``` ```
aiomail @ git+ssh://git@git.rethinkstudios.io/rethink-public/aiomail.git@v0.1.4 aiomail @ git+ssh://git@git.rethinkstudios.io/rethink-public/aiomail.git@v0.1.8
# OAuth token providers (Microsoft / Google) need the extra: # OAuth token providers (Microsoft / Google) need the extra:
aiomail[oauth] @ git+ssh://git@git.rethinkstudios.io/rethink-public/aiomail.git@v0.1.4 aiomail[oauth] @ git+ssh://git@git.rethinkstudios.io/rethink-public/aiomail.git@v0.1.8
``` ```
Direct: Direct:
```bash ```bash
pip install "aiomail @ git+ssh://git@git.rethinkstudios.io/rethink-public/aiomail.git@v0.1.4" pip install "aiomail @ git+ssh://git@git.rethinkstudios.io/rethink-public/aiomail.git@v0.1.8"
pip install "aiomail[oauth] @ git+ssh://git@git.rethinkstudios.io/rethink-public/aiomail.git@v0.1.4" pip install "aiomail[oauth] @ git+ssh://git@git.rethinkstudios.io/rethink-public/aiomail.git@v0.1.8"
``` ```
Requires `aioimaplib` and `beautifulsoup4` (pulled transitively). The `oauth` Requires `aioimaplib` and `beautifulsoup4` (pulled transitively). The `oauth`
extra adds `aiohttp` for the refresh-token providers. extra adds `aiohttp` for the refresh-token providers.
Drop the `@v0.1.4` suffix from the line above to install the latest unpinned. Drop the `@v0.1.8` suffix from the line above to install the latest unpinned.
## Password auth ## Password auth
@@ -61,11 +61,18 @@ Credentials are always supplied by you — nothing is hardcoded.
```python ```python
import re import re
from email.utils import parseaddr
await retrieve_otp(client, sender="uber.com") # substring 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=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 Code extraction is tunable too — `patterns` (regexes, first group wins) and
`lengths` (standalone digit-run fallback): `lengths` (standalone digit-run fallback):
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project] [project]
name = "aiomail" name = "aiomail"
version = "0.1.4" version = "0.1.8"
description = "async IMAP one-time-code retrieval with password/OAuth2 auth and dynamic matching" description = "async IMAP one-time-code retrieval with password/OAuth2 auth and dynamic matching"
requires-python = ">=3.10" requires-python = ">=3.10"
dependencies = [ dependencies = [
+3 -1
View File
@@ -11,6 +11,7 @@ from .extract import (
DEFAULT_PATTERNS, DEFAULT_PATTERNS,
MatchSpec, MatchSpec,
as_predicate, as_predicate,
decode_header_value,
extract_code, extract_code,
) )
from .retrieve import DEFAULT_FOLDERS, retrieve_otp from .retrieve import DEFAULT_FOLDERS, retrieve_otp
@@ -22,6 +23,7 @@ __all__ = [
"IMAPClient", "IMAPClient",
"extract_code", "extract_code",
"as_predicate", "as_predicate",
"decode_header_value",
"retrieve_otp", "retrieve_otp",
"MatchSpec", "MatchSpec",
"DEFAULT_PATTERNS", "DEFAULT_PATTERNS",
@@ -29,4 +31,4 @@ __all__ = [
"DEFAULT_FOLDERS", "DEFAULT_FOLDERS",
] ]
__version__ = "0.1.4" __version__ = "0.1.8"
+8 -23
View File
@@ -1,9 +1,6 @@
"""authentication mechanisms for the IMAP client. """authentication mechanisms for the IMAP client.
an `Auth` is anything that can authenticate an already-connected aioimaplib `PasswordAuth` (LOGIN) and `OAuth2Auth` (XOAUTH2); credentials always injected.
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 base64
import logging import logging
@@ -39,11 +36,7 @@ class PasswordAuth:
def _as_str(token) -> str: 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 str"""
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 return token.decode() if isinstance(token, bytes) else token
@@ -54,12 +47,7 @@ def _sasl_xoauth2(user: str, token: str) -> str:
class OAuth2Auth: 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__( def __init__(
self, self,
@@ -85,18 +73,15 @@ class OAuth2Auth:
async def authenticate(self, mail) -> None: async def authenticate(self, mail) -> None:
token = await self._resolve_token() 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 # _resolve_token already guarantees str via _as_str.
# returns str (via _as_str). clients lacking .xoauth2 are driven via the
# SASL callback from _sasl_xoauth2.
xoauth2 = getattr(mail, "xoauth2", None) xoauth2 = getattr(mail, "xoauth2", None)
if xoauth2 is not None: if xoauth2 is not None:
result, data = await xoauth2(self.user, token) result, data = await xoauth2(self.user, token)
elif hasattr(mail, "authenticate"): elif hasattr(mail, "authenticate"):
# escape hatch for a non-aioimaplib client: the shipped aioimaplib IMAP4 # escape hatch for a non-aioimaplib client (aioimaplib's IMAP4 always
# always has .xoauth2 and never .authenticate, so this branch never runs # has .xoauth2, never .authenticate); untested against any real driver
# for it; the SASL-callback signature here is untested against any driver
result, data = await mail.authenticate( result, data = await mail.authenticate(
"XOAUTH2", lambda _: _sasl_xoauth2(self.user, token) "XOAUTH2", lambda _: _sasl_xoauth2(self.user, token)
) )
+98 -42
View File
@@ -1,9 +1,9 @@
"""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)
flow needs (folders, search, fetch, mark-seen). auth is injected, so the same — pass `use_uid=True` if ids need to survive a reconnect. one instance is not
client serves password and OAuth accounts. safe for concurrent callers beyond the internal connect/reconnect lock.
""" """
import asyncio import asyncio
import email import email
@@ -23,24 +23,24 @@ log = logging.getLogger(__name__)
_LIST_RE = re.compile(rb'^\([^)]*\)\s+(?:"[^"]*"|NIL)\s+(.+)$') _LIST_RE = re.compile(rb'^\([^)]*\)\s+(?:"[^"]*"|NIL)\s+(.+)$')
def _folder_name(raw: bytes) -> str: def _folder_name(raw: bytes) -> Optional[str]:
"""extract the folder name from a LIST reply line, delimiter-agnostic """extract the folder name from a LIST reply line, delimiter-agnostic
parses the real reply form `(flags) "<delim>" <name>` so any server hierarchy returns None on a non-matching line instead of a last-token rsplit fallback,
delimiter works (not just "/"); falls back to the last quoted/space token if the so the tagged completion line aioimaplib appends to the same response list
line doesn't match the canonical shape. (e.g. `b"LIST completed."`) gets dropped instead of read as a phantom folder.
""" """
match = _LIST_RE.match(raw.strip()) match = _LIST_RE.match(raw.strip())
name = match.group(1).decode() if match else raw.decode().rsplit(" ", 1)[-1] if not match:
return name.strip().strip('"') return None
return match.group(1).decode().strip().strip('"')
class IMAPClient: class IMAPClient:
"""connection-managing IMAP client driven by an injected auth mechanism """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` —
of `use_ssl` — the two were conflated in an earlier draft (`use_uid = use_ssl`), an earlier draft conflated them (`use_uid = use_ssl`); unrelated concerns.
which is a bug; they are unrelated concerns.
""" """
def __init__( def __init__(
@@ -62,6 +62,8 @@ class IMAPClient:
self.timeout = timeout self.timeout = timeout
self.max_retries = max_retries self.max_retries = max_retries
self._mail = None self._mail = None
self._selected_folder: Optional[str] = None
self._lock = asyncio.Lock()
async def __aenter__(self) -> "IMAPClient": async def __aenter__(self) -> "IMAPClient":
await self.ensure_connection() await self.ensure_connection()
@@ -71,8 +73,20 @@ class IMAPClient:
await self.close() await self.close()
async def connect(self) -> bool: async def connect(self) -> bool:
"""open a connection and authenticate, retrying with linear backoff""" """open a connection and authenticate, retrying with linear backoff
await self.close()
serialized by an internal lock: queued callers never tear down each
other's in-progress handshake, and a superseded connection is logged out.
"""
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): for attempt in range(self.max_retries):
try: try:
if self.use_ssl: if self.use_ssl:
@@ -87,25 +101,31 @@ class IMAPClient:
if self._mail is not None: if self._mail is not None:
await self._discard_mail(self._mail) await self._discard_mail(self._mail)
self._mail = None self._mail = None
await asyncio.sleep(2 * (attempt + 1)) if attempt < self.max_retries - 1:
await asyncio.sleep(2 * (attempt + 1))
return False return False
@staticmethod @staticmethod
async def _discard_mail(mail) -> None: async def _discard_mail(mail) -> None:
"""tear down a half-built IMAP4 without leaking its connect task """tear down a half-built IMAP4 without leaking its connect task
aioimaplib's IMAP4 schedules `create_connection` as a fire-and-forget task it aioimaplib schedules `create_connection` as a fire-and-forget task; on a
never retrieves; on a refused connection that task raises and asyncio logs a refused connection it raises and asyncio logs a noisy "Task exception was
noisy "Task exception was never retrieved" traceback. cancel/await it here (and never retrieved" traceback unless retrieved here first.
retrieve its exception) before discarding, so a failed connect stays quiet.
""" """
task = getattr(mail, "_client_task", None) task = getattr(mail, "_client_task", None)
if task is not None and not task.done(): if task is not None and not task.done():
task.cancel() task.cancel()
if task is not None: if task is not None:
# shield distinguishes our own task.cancel() from an external cancel of
# this coroutine: a bare `await task` swallowed both, resisting
# cancellation. under shield, CancelledError here means external only.
try: try:
await task await asyncio.shield(task)
except (asyncio.CancelledError, Exception): except asyncio.CancelledError:
if not task.cancelled():
raise
except Exception:
pass pass
try: try:
await mail.logout() await mail.logout()
@@ -114,22 +134,55 @@ class IMAPClient:
async def close(self) -> None: async def close(self) -> None:
"""log out and drop the connection, swallowing teardown errors""" """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: if self._mail is not None:
mail, self._mail = self._mail, None
try: try:
await self._mail.logout() await mail.logout()
except Exception as exc: except Exception as exc:
log.debug("logout error ignored: %s", exc) log.debug("logout error ignored: %s", exc)
self._mail = None self._selected_folder = None
async def ensure_connection(self) -> bool: async def ensure_connection(self) -> bool:
"""return a live connection, reconnecting if the link is stale""" """return a live, SELECTED-if-applicable connection, reconnecting if the link is stale
if self._mail is None:
return await self.connect() re-selects the previously-selected folder after a reconnect; sequence-number
try: ids from before the reconnect are NOT valid against the new session (a fresh
await self._mail.noop() SELECT can renumber the mailbox) unless use_uid=True. serialized by the
internal lock so a queued caller rechecks liveness before reconnecting.
"""
async with self._lock:
if self._mail is not None:
try:
await self._mail.noop()
return True
except Exception:
pass
return await self._connect_and_reselect_locked()
async def _connect_and_reselect_locked(self) -> bool:
"""connect() then re-select the previously-selected folder; caller must hold self._lock"""
folder = self._selected_folder
if not await self._connect_locked():
return False
if folder is None:
return True return True
except Exception: try:
return await self.connect() 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]: async def get_folders(self) -> List[str]:
"""list mailbox folder names""" """list mailbox folder names"""
@@ -143,9 +196,11 @@ class IMAPClient:
folders: List[str] = [] folders: List[str] = []
for folder in folder_list or []: for folder in folder_list or []:
try: try:
folders.append(_folder_name(folder)) name = _folder_name(folder)
except Exception: except Exception:
continue continue
if name is not None:
folders.append(name)
return folders return folders
async def select(self, folder: str) -> bool: async def select(self, folder: str) -> bool:
@@ -154,10 +209,13 @@ class IMAPClient:
return False return False
try: try:
result, _ = await self._mail.select(f'"{folder}"') result, _ = await self._mail.select(f'"{folder}"')
return result == "OK"
except Exception as exc: except Exception as exc:
log.debug("select %s failed: %s", folder, exc) log.debug("select %s failed: %s", folder, exc)
return False return False
if result == "OK":
self._selected_folder = folder
return True
return False
async def search(self, query: str) -> List[int]: async def search(self, query: str) -> List[int]:
"""search the selected folder, returning ids newest-first""" """search the selected folder, returning ids newest-first"""
@@ -176,8 +234,7 @@ class IMAPClient:
try: try:
ids.append(int(token)) ids.append(int(token))
except (TypeError, ValueError): 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) log.debug("skipping non-numeric search token: %r", token)
return sorted(set(ids), reverse=True) return sorted(set(ids), reverse=True)
@@ -197,14 +254,13 @@ class IMAPClient:
if result != "OK" or not data: if result != "OK" or not data:
return None return None
for item in data: for item in data:
# aioimaplib stores the literal message payload as the only bytearray in # select by structure, not length: aioimaplib stores the literal payload
# the response; every other line (including the `<id> FETCH (...` header) # as the only bytearray in the response, so a length heuristic would
# is plain bytes. select by structure, not length — a length heuristic # mismatch the header line for a 2+ digit id or a BODY[]/UID fetch
# mismatches the header line for any 2+ digit id or a BODY[]/UID fetch.
if isinstance(item, bytearray): if isinstance(item, bytearray):
return email.message_from_bytes(bytes(item)) return email.message_from_bytes(bytes(item))
# cross-version fallback: aioimaplib 2.0.x never yields tuples here, but an # cross-version fallback for an imaplib-style (header, payload) tuple;
# imaplib-style (header, payload) tuple is handled if a future/alt driver does # aioimaplib 2.0.x never yields one here
if isinstance(item, tuple) and len(item) > 1: if isinstance(item, tuple) and len(item) > 1:
return email.message_from_bytes(item[1]) return email.message_from_bytes(item[1])
return None return None
+31 -21
View File
@@ -1,13 +1,13 @@
"""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 `extract_code` pulls a one-time code out of a message; `as_predicate` turns a
unit-test. `extract_code` pulls a one-time code out of a message; `as_predicate` string / compiled regex / callable into a uniform match function for filtering
turns a string / compiled regex / callable into a uniform match function used to senders and subjects.
filter senders and subjects.
""" """
import email.message import email.message
import logging import logging
import re import re
from email.header import decode_header, make_header
from typing import Callable, Iterable, Iterator, Optional, Pattern, Sequence, Union from typing import Callable, Iterable, Iterator, Optional, Pattern, Sequence, Union
from bs4 import BeautifulSoup from bs4 import BeautifulSoup
@@ -39,6 +39,21 @@ def _compile(patterns: Sequence[Union[str, Pattern]]) -> list[Pattern]:
return out 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) as exc:
log.debug("header decode failed (%s): %s", raw, exc)
return raw
def _decode_part(part: email.message.Message) -> Optional[str]: def _decode_part(part: email.message.Message) -> Optional[str]:
"""decode a single message part to text, tolerating bad charsets""" """decode a single message part to text, tolerating bad charsets"""
payload = part.get_payload(decode=True) payload = part.get_payload(decode=True)
@@ -74,10 +89,9 @@ def _scan(text: str, patterns: list[Pattern], lengths: set[int]) -> Optional[str
m = pat.search(text) m = pat.search(text)
if m: if m:
return m.group(1) if m.groups() else m.group(0) 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 return None
@@ -89,15 +103,14 @@ def extract_code(
) -> Optional[str]: ) -> Optional[str]:
"""extract a one-time code from a message, subject first then body parts """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 match if a pattern has no groups. when no pattern matches a block, any whole match); if none hit, a standalone digit run whose length is in
standalone digit run whose length is in `lengths` is returned. both knobs are `lengths` is returned. both are parameters so callers tune per provider.
parameters so callers can tune per provider without forking this function.
""" """
compiled = _compile(patterns) compiled = _compile(patterns)
length_set = set(lengths) length_set = set(lengths)
subject = message.get("Subject", "") or "" subject = decode_header_value(message.get("Subject", "") or "")
hit = _scan(subject, compiled, length_set) hit = _scan(subject, compiled, length_set)
if hit: if hit:
return hit return hit
@@ -110,17 +123,14 @@ def extract_code(
def as_predicate(spec: MatchSpec) -> Callable[[Optional[str]], bool]: 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: if spec is None:
return lambda value: True return lambda value: True
if isinstance(spec, re.Pattern): if isinstance(spec, re.Pattern):
return lambda value: bool(spec.search(value or "")) return lambda value: bool(spec.search(value or ""))
if callable(spec): 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() needle = str(spec).lower()
return lambda value: needle in (value or "").lower() return lambda value: needle in (value or "").lower()
+15 -8
View File
@@ -1,10 +1,7 @@
"""optional OAuth2 token providers (refresh-token -> access-token). """optional OAuth2 token providers (refresh-token -> access-token) for `OAuth2Auth`, credentials always caller-supplied.
these turn a refresh token into a fresh access token for `OAuth2Auth`. they need aiohttp is an optional extra so the core stays light; missing it raises a clear
aiohttp, kept as an optional extra so the core stays light; importing this module error only when a provider is instantiated, not on import.
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 asyncio
import logging import logging
@@ -76,12 +73,22 @@ class _RefreshTokenProvider:
async with aiohttp.ClientSession(timeout=timeout) as session: async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.post(endpoint, data=data) as resp: async with session.post(endpoint, data=data) as resp:
if resp.status == 200: 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: if token:
self._failures = 0 self._failures = 0
return token return token
# log a truncated body (never whole, may carry sensitive
# material) so a 200-with-no-token isn't a silent drop
log.warning(
"token endpoint %s -> 200 with no access_token: %s",
endpoint, str(body_json)[:200],
)
else: 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) log.warning("token endpoint %s -> %s: %s", endpoint, resp.status, body)
except Exception as exc: except Exception as exc:
log.warning("token request to %s failed: %s", endpoint, exc) log.warning("token request to %s failed: %s", endpoint, exc)
+30 -22
View File
@@ -1,18 +1,24 @@
"""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 sender/subject accept the flexible match specs from `extract`; provider quirks
flexible match specs from `extract`, folders/age/patterns are all parameters with (folders, age, patterns) live in the arguments, not hardcoded branches.
sane defaults, and provider quirks live in the arguments rather than hardcoded
branches inside the function.
""" """
import asyncio import asyncio
import logging import logging
import time import time
from datetime import timezone
from email.utils import parsedate_to_datetime from email.utils import parsedate_to_datetime
from typing import Iterable, List, Optional, Pattern, Sequence, Union from typing import Iterable, List, Optional, Pattern, Sequence, Union
from .client import IMAPClient 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__) log = logging.getLogger(__name__)
@@ -21,17 +27,18 @@ DEFAULT_FOLDERS: Sequence[str] = ("INBOX", "Junk", "Spam", "Archive", "All Mail"
def _server_query(sender: MatchSpec, subject: MatchSpec, match_field: str = "from") -> 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 only, falling back to ALL for regex/callable specs
only plain strings translate to server-side filters; regex and callable specs `match_field="to"` searches TO OR FROM (a forwarded code may keep the
fall back to ALL and are filtered client-side, so dynamic matching always works original From), matching the client-side forwarded-From fallback so the
even when the server cannot express it. `match_field` selects which header the server query never narrows out a result the client would have accepted.
`sender` spec searches: "from" filters by the sender address (default), "to"
filters by the recipient address (the per-user alias the code was sent to).
""" """
parts: List[str] = [] parts: List[str] = []
if isinstance(sender, str): if isinstance(sender, str):
parts.append(f'TO "{sender}"' if match_field == "to" else f'FROM "{sender}"') if match_field == "to":
parts.append(f'OR TO "{sender}" FROM "{sender}"')
else:
parts.append(f'FROM "{sender}"')
if isinstance(subject, str): if isinstance(subject, str):
parts.append(f'SUBJECT "{subject}"') parts.append(f'SUBJECT "{subject}"')
return f"({' '.join(parts)})" if parts else "ALL" return f"({' '.join(parts)})" if parts else "ALL"
@@ -43,7 +50,12 @@ def _age_seconds(message) -> Optional[float]:
if not raw: if not raw:
return None return None
try: 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: except (TypeError, ValueError) as exc:
log.debug("date parse failed (%s): %s", raw, exc) log.debug("date parse failed (%s): %s", raw, exc)
return None return None
@@ -67,13 +79,9 @@ async def retrieve_otp(
) -> Optional[str]: ) -> Optional[str]:
"""return the newest OTP matching the filters, or None """return the newest OTP matching the filters, or None
sender/subject accept a substring, a compiled regex, or a callable. `match_field` sender/subject accept a substring, a compiled regex, or a callable.
selects which header the `sender` spec is matched against: "from" (default) `match_field="to"` matches the recipient address and additionally accepts a
matches the sender address; "to" matches the recipient address (the per-user forwarded match on From. set `max_age=None` to disable the freshness check.
alias the code was sent to) and additionally accepts a forwarded match on the
From header, so a forwarded code still resolves. folders, patterns, code lengths,
max age and retry behavior are all tunable. set `max_age=None` to disable the
freshness check.
""" """
folders = list(folders) if folders is not None else list(DEFAULT_FOLDERS) folders = list(folders) if folders is not None else list(DEFAULT_FOLDERS)
sender_ok = as_predicate(sender) sender_ok = as_predicate(sender)
@@ -97,7 +105,7 @@ async def retrieve_otp(
continue continue
from_hdr = message.get("From", "") from_hdr = message.get("From", "")
subj_hdr = message.get("Subject", "") subj_hdr = decode_header_value(message.get("Subject", ""))
if match_field == "to": if match_field == "to":
to_hdr = message.get("To", "") to_hdr = message.get("To", "")
matched = sender_ok(to_hdr) or sender_ok(from_hdr) matched = sender_ok(to_hdr) or sender_ok(from_hdr)