rex/calendar_accounts.py
Rama master · réplica de lectura del repositorio original.
"""Canonical per-user calendar account authorization and routing (issue #303).
This module is the single source of truth for deciding **which calendar
accounts a user may touch** and **which account serves a request**. Every
real calendar surface (service, CLI, GUI/API routes, Electron bridge,
OpenClaw tools, local tool executor, scheduler) must route account selection
through :class:`CalendarAccountResolver`.
Ownership model
---------------
- ``calendar.accounts`` in ``config/rex_config.json`` is the canonical
non-secret account-definition list (provider, ICS source,
``credential_ref`` — an environment-variable *name*, never a secret value).
- ``users.{user_id}.calendar_accounts`` is the authoritative authorization
map assigning account IDs to users.
- ``users.{user_id}.default_calendar_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).
- ``calendar.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.
- The legacy global calendar configuration (``calendar.backend``/
``calendar.ics`` and ``calendar.provider`` with its global environment
token) is synthesized into reserved legacy accounts that belong only to
the ``default`` profile. Named users can never use them — not even via an
explicit assignment — so legacy global credentials are usable only by the
explicit ``default`` profile.
- Legacy ``calendar.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:`CalendarAccountAccessError` 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 dataclasses import dataclass
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 calendar 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 calendar
#: configuration and the legacy global default belong to this profile only.
DEFAULT_PROFILE = "default"
#: Reserved IDs for accounts synthesized from the legacy global calendar
#: configuration. These are never available to named users.
LEGACY_ICS_ACCOUNT_ID = "legacy-ics"
LEGACY_PROVIDER_ACCOUNT_PREFIX = "legacy-provider-"
_LEGACY_GOOGLE_TOKEN_ENV = "GOOGLE_CALENDAR_ACCESS_TOKEN"
#: Generic message for unauthorized-or-nonexistent accounts. Must not vary
#: with account existence so callers cannot enumerate other users' accounts.
ACCOUNT_UNAVAILABLE_MSG = "Calendar account {account_id!r} is not available for user {user_id!r}"
_IDENTITY_REQUIRED_MSG = (
"Calendar operations require an explicit valid user identity; "
"refusing to fall back to a default or shared account"
)
class CalendarAccountAccessError(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 CalendarIdentityError(PermissionError):
"""A real calendar operation was attempted without a valid user identity."""
def require_user_id(user_id: object) -> str:
"""Validate *user_id* for calendar operations, failing closed.
Returns:
The validated user ID.
Raises:
CalendarIdentityError: 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 CalendarIdentityError(_IDENTITY_REQUIRED_MSG)
try:
return validate_user_id(user_id)
except ValueError as exc:
raise CalendarIdentityError(_IDENTITY_REQUIRED_MSG) from exc
@dataclass(frozen=True)
class CalendarAccountDefinition:
"""Canonical non-secret calendar account definition.
``credential_ref`` names the logical credential-vault entry for this
account's token. It is a *reference*, never the
secret itself, and must not be echoed to CLI output, API responses,
events, or error messages.
"""
id: str
label: str = ""
provider: str = "ics" # "ics" | "google" | "outlook" | "stub"
credential_ref: str = ""
ics_source: str = ""
ics_url_timeout: int = 15
legacy: bool = False # synthesized from legacy global config
def _parse_account_definition(raw: dict[str, Any]) -> CalendarAccountDefinition | None:
account_id = str(raw.get("id") or "").strip()
if not account_id:
return None
ics_raw = raw.get("ics")
ics_cfg = ics_raw if isinstance(ics_raw, dict) else {}
try:
url_timeout = int(ics_cfg.get("url_timeout", 15))
except (TypeError, ValueError):
url_timeout = 15
return CalendarAccountDefinition(
id=account_id,
label=str(raw.get("label") or ""),
provider=str(raw.get("provider") or "ics").strip().lower(),
credential_ref=str(raw.get("credential_ref") or ""),
ics_source=str(ics_cfg.get("source") or raw.get("ics_source") or ""),
ics_url_timeout=url_timeout,
legacy=False,
)
def _synthesize_legacy_accounts(
calendar_section: dict[str, Any],
) -> list[CalendarAccountDefinition]:
"""Build reserved legacy accounts from the pre-#303 global configuration.
These represent the single-user global calendar setup (``backend: ics``
and/or ``provider: google``) and belong only to the ``default`` profile.
"""
legacy: list[CalendarAccountDefinition] = []
backend = str(calendar_section.get("backend") or "").strip().lower()
ics_raw = calendar_section.get("ics")
ics_cfg = ics_raw if isinstance(ics_raw, dict) else {}
source = str(ics_cfg.get("source") or "")
if backend == "ics" and source:
try:
url_timeout = int(ics_cfg.get("url_timeout", 15))
except (TypeError, ValueError):
url_timeout = 15
legacy.append(
CalendarAccountDefinition(
id=LEGACY_ICS_ACCOUNT_ID,
label="Legacy ICS calendar",
provider="ics",
ics_source=source,
ics_url_timeout=url_timeout,
legacy=True,
)
)
provider = str(calendar_section.get("provider") or "").strip().lower()
if provider == "gmail":
provider = "google"
if provider in ("google", "outlook"):
legacy.append(
CalendarAccountDefinition(
id=f"{LEGACY_PROVIDER_ACCOUNT_PREFIX}{provider}",
label=f"Legacy {provider} calendar",
provider=provider,
credential_ref=_LEGACY_GOOGLE_TOKEN_ENV if provider == "google" else "",
legacy=True,
)
)
return legacy
def _parse_user_assignments(users_block: object) -> dict[str, list[str]]:
"""Parse ``users.{user_id}.calendar_accounts`` into ``{user_id: [account_id]}``.
Accepts entries as ``{"account_id": "..."}`` dicts or bare strings.
"""
result: dict[str, list[str]] = {}
if not isinstance(users_block, dict):
return result
for user_id, user_data in users_block.items():
if not isinstance(user_data, dict):
continue
raw_accounts = user_data.get("calendar_accounts", [])
if not isinstance(raw_accounts, list):
continue
parsed: list[str] = []
for entry in raw_accounts:
if isinstance(entry, str) and entry.strip():
parsed.append(entry.strip())
elif isinstance(entry, dict):
account_id = str(entry.get("account_id") or "").strip()
if account_id:
parsed.append(account_id)
if parsed:
result[str(user_id)] = parsed
return result
def _parse_user_defaults(users_block: object) -> dict[str, str]:
"""Parse ``users.{user_id}.default_calendar_account_id`` into a per-user map."""
result: dict[str, str] = {}
if not isinstance(users_block, dict):
return result
for user_id, user_data in users_block.items():
if not isinstance(user_data, dict):
continue
default_id = user_data.get("default_calendar_account_id")
if isinstance(default_id, str) and default_id.strip():
result[str(user_id)] = default_id.strip()
return result
class CalendarAccountResolver:
"""Authorization and routing for per-user calendar accounts.
Args:
accounts: Canonical explicit account definitions.
legacy_accounts: Accounts synthesized from the legacy global config
(``default`` profile only).
user_accounts: ``{user_id: [account_id, ...]}`` authorization map.
user_defaults: ``{user_id: account_id}`` per-user default selection.
legacy_default_account_id: Legacy global ``calendar.default_account_id``
(``default`` profile only).
"""
def __init__(
self,
accounts: list[CalendarAccountDefinition] | None = None,
legacy_accounts: list[CalendarAccountDefinition] | None = None,
user_accounts: dict[str, list[str]] | None = None,
user_defaults: dict[str, str] | None = None,
legacy_default_account_id: str | None = None,
) -> None:
self._accounts: list[CalendarAccountDefinition] = list(accounts or [])
self._legacy_accounts: list[CalendarAccountDefinition] = list(legacy_accounts or [])
self._user_accounts: dict[str, list[str]] = dict(user_accounts or {})
self._user_defaults: dict[str, str] = dict(user_defaults or {})
self._legacy_default_account_id = legacy_default_account_id
# ------------------------------------------------------------------
# Construction
# ------------------------------------------------------------------
@classmethod
def from_raw_config(cls, raw_config: dict[str, Any] | None) -> CalendarAccountResolver:
"""Build a resolver from the merged runtime config dict."""
raw = raw_config or {}
calendar_section = raw.get("calendar")
if not isinstance(calendar_section, dict):
calendar_section = {}
accounts: list[CalendarAccountDefinition] = []
accounts_raw = calendar_section.get("accounts")
if isinstance(accounts_raw, list):
for item in accounts_raw:
if not isinstance(item, dict):
continue
definition = _parse_account_definition(item)
if definition is None:
logger.warning("Skipping malformed calendar account entry (missing id)")
continue
accounts.append(definition)
legacy_accounts = _synthesize_legacy_accounts(calendar_section)
users_block = raw.get("users")
user_accounts = _parse_user_assignments(users_block)
user_defaults = _parse_user_defaults(users_block)
legacy_default = calendar_section.get("default_account_id")
legacy_default_id = (
legacy_default.strip()
if isinstance(legacy_default, str) and legacy_default.strip()
else None
)
return cls(
accounts=accounts,
legacy_accounts=legacy_accounts,
user_accounts=user_accounts,
user_defaults=user_defaults,
legacy_default_account_id=legacy_default_id,
)
@classmethod
def load(cls) -> CalendarAccountResolver:
"""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 calendar account resolution: %s", exc)
raw = {}
return cls.from_raw_config(raw)
# ------------------------------------------------------------------
# Introspection
# ------------------------------------------------------------------
def has_configured_accounts(self) -> bool:
"""True when any real account definition exists (explicit or legacy)."""
return bool(self._accounts or self._legacy_accounts)
def get_account_definition(self, account_id: str) -> CalendarAccountDefinition | None:
"""Return the canonical account definition for *account_id*, if any.
This is an internal lookup used *after* ownership authorization;
surfaces must never expose foreign definitions obtained here.
"""
for acct in self._accounts:
if acct.id == account_id:
return acct
for acct in self._legacy_accounts:
if acct.id == account_id:
return acct
return None
# ------------------------------------------------------------------
# Authorization
# ------------------------------------------------------------------
def account_ids_for_user(self, user_id: str) -> list[str]:
"""Return the account IDs authorized for *user_id* (deterministic order).
Named users get only their explicit assignments, restricted to
accounts that actually exist in ``calendar.accounts``. Legacy
synthesized accounts are excluded for named users even when a config
entry tries to assign one (legacy global credentials are usable only
by the ``default`` profile).
The ``default`` profile additionally receives every explicit account
no user has claimed, plus the legacy synthesized accounts.
"""
validated = require_user_id(user_id)
explicit_ids = [acct.id for acct in self._accounts]
legacy_ids = [acct.id for acct in self._legacy_accounts]
owned: list[str] = []
for account_id in self._user_accounts.get(validated, []):
if account_id in legacy_ids:
logger.warning(
"Ignoring assignment of legacy calendar account to user %r: "
"legacy global calendar configuration is restricted to the "
"'default' profile",
validated,
)
continue
if account_id in explicit_ids and account_id not in owned:
owned.append(account_id)
if validated == DEFAULT_PROFILE:
assigned = {account_id for ids in self._user_accounts.values() for account_id in ids}
for account_id in explicit_ids:
if account_id not in assigned and account_id not in owned:
owned.append(account_id)
for account_id in legacy_ids:
if account_id not in owned:
owned.append(account_id)
return owned
def accounts_for_user(self, user_id: str) -> list[CalendarAccountDefinition]:
"""Return the canonical account definitions authorized for *user_id*."""
definitions = []
for account_id in self.account_ids_for_user(user_id):
definition = self.get_account_definition(account_id)
if definition is not None:
definitions.append(definition)
return definitions
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 unassigned or
legacy accounts exist. Order is deterministic.
"""
ids: list[str] = []
for user_id in self._user_accounts:
try:
validated = require_user_id(user_id)
except CalendarIdentityError:
logger.warning("Skipping calendar accounts assigned to an invalid user ID")
continue
if validated not in ids and self.account_ids_for_user(validated):
ids.append(validated)
if DEFAULT_PROFILE not in ids and self.account_ids_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:`CalendarAccountAccessError` unless *user_id* owns *account_id*."""
validated = require_user_id(user_id)
if not self.is_account_authorized(validated, account_id):
raise CalendarAccountAccessError(
ACCOUNT_UNAVAILABLE_MSG.format(account_id=account_id, user_id=validated)
)
# ------------------------------------------------------------------
# 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 calendar account for user %r: not owned by that user",
validated,
)
if validated == DEFAULT_PROFILE:
legacy = self._legacy_default_account_id
if legacy and legacy in owned:
return 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:
CalendarIdentityError: On missing/invalid identity.
CalendarAccountAccessError: 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 CalendarAccountAccessError(
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
return owned[0] if owned else None
def resolve_account(
self, user_id: str, account_id: str | None = None
) -> CalendarAccountDefinition | None:
"""Resolve the canonical account definition serving a request.
Same contract as :meth:`resolve_account_id`, but returns the account
definition.
"""
resolved_id = self.resolve_account_id(user_id, account_id)
if resolved_id is None:
return None
return self.get_account_definition(resolved_id)
def provider_account_for_user(self, user_id: str) -> CalendarAccountDefinition | None:
"""Return the user's first OAuth-provider (google/outlook) account.
Used by the GUI/bridge calendar surfaces, which speak the provider
API rather than ICS. Ownership is enforced by
:meth:`accounts_for_user`.
"""
for definition in self.accounts_for_user(user_id):
if definition.provider in ("google", "outlook"):
return definition
return None
__all__ = [
"ACCOUNT_UNAVAILABLE_MSG",
"DEFAULT_PROFILE",
"LEGACY_ICS_ACCOUNT_ID",
"LEGACY_PROVIDER_ACCOUNT_PREFIX",
"CalendarAccountAccessError",
"CalendarAccountDefinition",
"CalendarAccountResolver",
"CalendarIdentityError",
"config_stamp",
"require_user_id",
]