rex/email_accounts.py
Rama master · réplica de lectura del repositorio original.
"""Canonical per-user email account authorization and routing (issue #303).
This module is the single source of truth for deciding **which email accounts
a user may touch** and **which account serves a request**. Every real email
surface (service, CLI, Electron bridge, OpenClaw tools, scheduler, notification
channel) must route account selection through :class:`EmailAccountResolver`.
Ownership model
---------------
- ``email.accounts`` in ``config/rex_config.json`` is the canonical non-secret
connection-metadata list (host/port/address/``credential_ref``).
- ``users.{user_id}.email_accounts`` is the authoritative authorization map
assigning account IDs to users (parsed into
``AppConfig.user_email_accounts``).
- ``users.{user_id}.default_email_account_id`` selects that user's default
account. It must reference an account the user owns; a foreign or unknown
default is ignored (fail closed to the user's own fallback).
- Legacy ``email.accounts`` entries not assigned to any user belong **only**
to the distinct ``default`` profile. They are never shared with or
silently reassigned to named users.
- Legacy ``email.default_account_id`` applies only to the ``default`` profile.
Routing order (always within the requesting user's authorized set):
1. Explicit requested account ID (after ownership validation).
2. The user's configured default account.
3. Legacy global default — only for the explicit ``default`` profile.
4. The first account assigned to the user (deterministic, config order).
5. Otherwise: no account (callers fail closed / report not-configured).
Unauthorized and nonexistent accounts are indistinguishable to callers:
both raise :class:`EmailAccountAccessError` with the same generic message.
Credential lookup happens only *after* ownership validation and only with the
authorized account definition's own ``credential_ref``.
"""
from __future__ import annotations
import logging
from pathlib import Path
from typing import Any
from rex.identity import validate_user_id
logger = logging.getLogger(__name__)
#: Runtime config file backing the authorization map.
_CONFIG_PATH = Path("config/rex_config.json")
def config_stamp() -> int | None:
"""Modification stamp of the runtime config file.
Used by long-lived services to invalidate cached resolvers so that
revoking or reassigning a user's email accounts takes effect without a
process restart.
"""
try:
return _CONFIG_PATH.stat().st_mtime_ns
except OSError:
return None
#: The distinct legacy single-user profile. Legacy global accounts and the
#: legacy global default belong to this profile only.
DEFAULT_PROFILE = "default"
#: Generic message for unauthorized-or-nonexistent accounts. Must not vary
#: with account existence so callers cannot enumerate other users' accounts.
ACCOUNT_UNAVAILABLE_MSG = "Email account {account_id!r} is not available for user {user_id!r}"
_IDENTITY_REQUIRED_MSG = (
"Email operations require an explicit valid user identity; "
"refusing to fall back to a default or shared account"
)
class EmailAccountAccessError(PermissionError):
"""The requested account is not available to the requesting user.
Raised both for accounts owned by another user and for accounts that do
not exist, with an identical message shape, so the two cases cannot be
distinguished by the caller.
"""
class EmailIdentityError(PermissionError):
"""A real email operation was attempted without a valid user identity."""
def require_user_id(user_id: object) -> str:
"""Validate *user_id* for email operations, failing closed.
Returns:
The validated user ID.
Raises:
EmailIdentityError: If *user_id* is missing, blank, malformed, or
traversal-style. Raised before any account or credential lookup.
"""
if not isinstance(user_id, str) or not user_id.strip():
raise EmailIdentityError(_IDENTITY_REQUIRED_MSG)
try:
return validate_user_id(user_id)
except ValueError as exc:
raise EmailIdentityError(_IDENTITY_REQUIRED_MSG) from exc
class EmailAccountResolver:
"""Authorization and routing for per-user email accounts.
Args:
email_config: Parsed ``EmailConfig`` (canonical account definitions).
user_accounts: ``{user_id: [UserEmailAccount, ...]}`` authorization map.
user_defaults: ``{user_id: account_id}`` per-user default selection.
"""
def __init__(
self,
email_config: Any,
user_accounts: dict[str, list[Any]] | None = None,
user_defaults: dict[str, str] | None = None,
) -> None:
self._email_config = email_config
self._user_accounts: dict[str, list[Any]] = dict(user_accounts or {})
self._user_defaults: dict[str, str] = dict(user_defaults or {})
# ------------------------------------------------------------------
# Construction
# ------------------------------------------------------------------
@classmethod
def from_raw_config(cls, raw_config: dict[str, Any] | None) -> EmailAccountResolver:
"""Build a resolver from the merged runtime config dict."""
from rex.config import _parse_user_default_email_accounts, _parse_user_email_accounts
from rex.email_backends.account_config import EmailConfig
raw = raw_config or {}
email_section = raw.get("email")
if not isinstance(email_section, dict):
email_section = {}
# Only pass the keys EmailConfig knows about; the email section may
# also carry unrelated keys such as ``provider``.
try:
email_config = EmailConfig.model_validate(
{
"default_account_id": email_section.get("default_account_id"),
"accounts": email_section.get("accounts") or [],
}
)
except Exception as exc:
logger.warning("Invalid email account configuration ignored: %s", exc)
email_config = EmailConfig()
users_block = raw.get("users")
user_accounts = _parse_user_email_accounts(users_block, email_section.get("accounts") or [])
user_defaults = _parse_user_default_email_accounts(users_block)
return cls(email_config, user_accounts, user_defaults)
@classmethod
def load(cls) -> EmailAccountResolver:
"""Build a resolver from ``config/rex_config.json``."""
from rex.config_manager import load_config
try:
raw = load_config()
except Exception as exc:
logger.warning("Failed to load config for email account resolution: %s", exc)
raw = {}
return cls.from_raw_config(raw)
# ------------------------------------------------------------------
# Introspection
# ------------------------------------------------------------------
@property
def email_config(self) -> Any:
return self._email_config
def has_configured_accounts(self) -> bool:
"""True when any real account definition exists."""
return bool(getattr(self._email_config, "accounts", None))
def get_account_definition(self, account_id: str) -> Any | None:
"""Return the canonical account definition for *account_id*, if any."""
for acct in getattr(self._email_config, "accounts", []) or []:
if acct.id == account_id:
return acct
return None
# ------------------------------------------------------------------
# Authorization
# ------------------------------------------------------------------
def entries_for_user(self, user_id: str) -> list[Any]:
"""Return the ``UserEmailAccount`` entries authorized for *user_id*.
For the ``default`` profile this includes legacy account definitions
that no user has claimed. Named users get only explicit assignments.
"""
validated = require_user_id(user_id)
entries = list(self._user_accounts.get(validated, []))
if validated == DEFAULT_PROFILE:
assigned = {
entry.account_id
for user_entries in self._user_accounts.values()
for entry in user_entries
}
have = {entry.account_id for entry in entries}
from rex.config import UserEmailAccount
for acct in getattr(self._email_config, "accounts", []) or []:
if acct.id not in assigned and acct.id not in have:
entries.append(
UserEmailAccount(
account_id=acct.id,
display_name=acct.label or acct.address,
backend="imap",
credentials_key=acct.credential_ref,
)
)
return entries
def account_ids_for_user(self, user_id: str) -> list[str]:
"""Return account IDs authorized for *user_id* (deterministic order)."""
return [entry.account_id for entry in self.entries_for_user(user_id)]
def configured_user_ids(self) -> list[str]:
"""Return the users with at least one authorized account.
Only valid user IDs are returned (invalid persisted IDs are skipped,
fail closed). The ``default`` profile is included when legacy
unassigned accounts exist. Order is deterministic.
"""
ids: list[str] = []
for user_id in self._user_accounts:
try:
validated = require_user_id(user_id)
except EmailIdentityError:
logger.warning("Skipping email accounts assigned to an invalid user ID")
continue
if self._user_accounts.get(validated) and validated not in ids:
ids.append(validated)
if DEFAULT_PROFILE not in ids and self.entries_for_user(DEFAULT_PROFILE):
ids.append(DEFAULT_PROFILE)
return ids
def is_account_authorized(self, user_id: str, account_id: str) -> bool:
return account_id in self.account_ids_for_user(user_id)
def check_account_access(self, user_id: str, account_id: str) -> None:
"""Raise :class:`EmailAccountAccessError` unless *user_id* owns *account_id*."""
validated = require_user_id(user_id)
if not self.is_account_authorized(validated, account_id):
raise EmailAccountAccessError(
ACCOUNT_UNAVAILABLE_MSG.format(account_id=account_id, user_id=validated)
)
def accounts_for_user(self, user_id: str) -> list[Any]:
"""Return the canonical account definitions authorized for *user_id*."""
definitions = []
for entry in self.entries_for_user(user_id):
definition = self.get_account_definition(entry.account_id)
if definition is not None:
definitions.append(definition)
return definitions
def provider_entry_for_user(self, user_id: str) -> Any | None:
"""Return the user's first OAuth-provider account entry, if any.
Used by the GUI inbox surfaces: a ``gmail``/``outlook`` backend entry
carries its own ``credentials_key`` and does not reference an
``email.accounts`` IMAP definition.
"""
for entry in self.entries_for_user(user_id):
if getattr(entry, "backend", "") in ("gmail", "outlook"):
return entry
return None
# ------------------------------------------------------------------
# Routing
# ------------------------------------------------------------------
def default_account_id_for_user(self, user_id: str) -> str | None:
"""Return the user's effective default account ID, or ``None``.
A configured default that the user does not own is ignored (fail
closed) rather than granting access or aborting.
"""
validated = require_user_id(user_id)
owned = self.account_ids_for_user(validated)
configured = self._user_defaults.get(validated)
if configured:
if configured in owned:
return configured
logger.warning(
"Ignoring default email account for user %r: not owned by that user",
validated,
)
if validated == DEFAULT_PROFILE:
legacy = getattr(self._email_config, "default_account_id", None)
if legacy and legacy in owned:
return str(legacy)
return None
def resolve_account_id(self, user_id: str, account_id: str | None = None) -> str | None:
"""Resolve the account that serves a request for *user_id*.
Returns:
The resolved account ID, or ``None`` when the user has no
authorized accounts (callers fail closed or report
not-configured).
Raises:
EmailIdentityError: On missing/invalid identity.
EmailAccountAccessError: When *account_id* is explicitly requested
but is not available to this user (unauthorized or
nonexistent — indistinguishable).
"""
validated = require_user_id(user_id)
owned = self.account_ids_for_user(validated)
if account_id:
if account_id not in owned:
raise EmailAccountAccessError(
ACCOUNT_UNAVAILABLE_MSG.format(account_id=account_id, user_id=validated)
)
return account_id
default_id = self.default_account_id_for_user(validated)
if default_id:
return default_id
# Deterministic fallback: first authorized account that has a real
# definition; else the first authorized entry (stub-mode setups).
for owned_id in owned:
if self.get_account_definition(owned_id) is not None:
return owned_id
return owned[0] if owned else None
def resolve_account(self, user_id: str, account_id: str | None = None) -> Any | None:
"""Resolve the canonical account definition serving a request.
Same contract as :meth:`resolve_account_id`, but returns the account
definition (or ``None`` if the resolved account has no definition,
e.g. stub-only setups).
"""
resolved_id = self.resolve_account_id(user_id, account_id)
if resolved_id is None:
return None
return self.get_account_definition(resolved_id)
__all__ = [
"ACCOUNT_UNAVAILABLE_MSG",
"DEFAULT_PROFILE",
"EmailAccountAccessError",
"EmailAccountResolver",
"EmailIdentityError",
"require_user_id",
]