rex/credentials.py
Rama master · réplica de lectura del repositorio original.
"""Credential vault for secure token handling.
This module provides a centralized credential management system that:
- Loads credentials from the OS-backed vault by opaque contextual reference
- Permits legacy plaintext environment/config reads only by explicit opt-in
- Provides methods to get, set, and refresh tokens
- Never returns secret-derived previews from status APIs
- Checks token expiry and supports refresh stubs
Credentials are loaded lazily and can be refreshed at runtime.
"""
from __future__ import annotations
import json
import logging
import os
from collections.abc import Callable
from dataclasses import dataclass, field
from datetime import UTC, datetime
from pathlib import Path
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from rex.credential_vault import CredentialVaultBackend
logger = logging.getLogger(__name__)
# Default environment variable prefix for Rex credentials
ENV_PREFIX = "REX_"
# Default mapping of service names to environment variable names
# Format: service_name -> env_var_name (without prefix)
DEFAULT_CREDENTIAL_MAPPING: dict[str, str] = {
"email": "EMAIL_TOKEN",
"calendar": "CALENDAR_TOKEN",
"home_assistant": "HA_TOKEN",
"brave": "BRAVE_API_KEY",
"openai": "OPENAI_API_KEY",
"openrouter": "OPENROUTER_API_KEY",
"anthropic": "ANTHROPIC_API_KEY",
"ollama": "OLLAMA_API_KEY",
"serpapi": "SERPAPI_API_KEY",
"github": "GITHUB_TOKEN",
"speak": "SPEAK_API_KEY",
"openweathermap": "OPENWEATHERMAP_API_KEY",
}
_CREDENTIAL_CONTEXTS: dict[str, tuple[str, str | None, str]] = {
"REX_JWT_SECRET": ("authentication", None, "jwt_signing_secret"),
"REX_PROXY_TOKEN": ("gui_proxy", None, "token"),
"REX_AGENT_TOKEN": ("computer_agent", None, "token"),
"HA_TOKEN": ("home_assistant", None, "token"),
"HA_SECRET": ("home_assistant", None, "secret"),
"TWILIO_ACCOUNT_SID": ("twilio", "sms", "account_sid"),
"TWILIO_AUTH_TOKEN": ("twilio", "sms", "auth_token"),
"TWILIO_FROM_NUMBER": ("twilio", "sms", "from_number"),
"TWILIO_PHONE_ACCOUNT_SID": ("twilio", "phone", "account_sid"),
"TWILIO_PHONE_AUTH_TOKEN": ("twilio", "phone", "auth_token"),
"TWILIO_PHONE_NUMBER": ("twilio", "phone", "phone_number"),
"TWILIO_TRANSFER_NUMBER": ("twilio", "phone", "transfer_number"),
"TELEGRAM_BOT_TOKEN": ("telegram", None, "token"),
"OPENCLAW_GATEWAY_TOKEN": ("openclaw_gateway", None, "token"),
"PUSH_TOKEN": ("push", None, "token"),
}
def credential_context_for_name(name: str) -> tuple[str, str | None, str]:
"""Return the fixed authorization context for a logical credential name."""
explicit = _CREDENTIAL_CONTEXTS.get(name.upper())
if explicit is not None:
return explicit
return credential_integration_for_name(name), None, credential_slot_for_name(name)
def credential_slot_for_name(name: str) -> str:
explicit = _CREDENTIAL_CONTEXTS.get(name.upper())
if explicit is not None:
return explicit[2]
upper = name.upper()
if upper.endswith("API_KEY") or upper.endswith("_SID") or upper.endswith("_NUMBER"):
return "api_key" if upper.endswith("API_KEY") else upper.lower()
if upper.endswith("_SECRET"):
return "secret"
if upper.endswith("_PASSWORD"):
return "password"
return "token"
def credential_integration_for_name(name: str) -> str:
explicit = _CREDENTIAL_CONTEXTS.get(name.upper())
if explicit is not None:
return explicit[0]
reverse = {env_name: service for service, env_name in DEFAULT_CREDENTIAL_MAPPING.items()}
if name in reverse:
return reverse[name]
upper = name.upper()
for suffix in ("_API_KEY", "_KEY"):
if upper.endswith(suffix):
return name[: -len(suffix)].lower()
return name.lower()
# Default path for credential config file
DEFAULT_CONFIG_PATH = Path(__file__).resolve().parent.parent / "config" / "credentials.json"
# Explicit, clearly-named opt-in for reading secrets from plaintext
# env/.env/config.json sources. Left unset (the default), normal production
# operation is vault-only: a secret not in the vault is simply absent, it is
# never read from plaintext as an automatic fallback. Packaged Electron
# actively removes this variable from every bridge child's environment so an
# operator shell cannot accidentally weaken the packaged application. Python
# also rejects the flag whenever ASKREX_PACKAGED=1 as defense in depth.
LEGACY_PLAINTEXT_FALLBACK_ENV_VAR = "REX_ALLOW_PLAINTEXT_CREDENTIAL_FALLBACK"
def legacy_plaintext_fallback_enabled() -> bool:
"""Whether plaintext env/.env/config.json may be read as a credential source.
Returns:
True only outside packaged mode when
``REX_ALLOW_PLAINTEXT_CREDENTIAL_FALLBACK`` is explicitly set to one
of the documented truthy values. False (fail closed) by default.
"""
if os.getenv("ASKREX_PACKAGED", "").strip().lower() in {"1", "true", "yes", "on"}:
return False
return os.getenv(LEGACY_PLAINTEXT_FALLBACK_ENV_VAR, "").strip().lower() in {
"1",
"true",
"yes",
"on",
}
REDACTED_MARKER = "[redacted]"
def mask_token(token: str | None, *, visible_chars: int = 4) -> str:
"""Return a constant redaction marker for a token, revealing nothing about it.
Args:
token: The token to mask. Present only to keep the call signature at
existing call sites; its value never influences the output.
visible_chars: Unused. Retained for call-site compatibility; no
characters of the token are ever revealed.
Returns:
``"[empty]"`` if the token is None or empty, otherwise a constant
redaction marker that leaks no prefix, suffix, length, or hash of
the underlying secret.
"""
del visible_chars
if not token:
return "[empty]"
return REDACTED_MARKER
@dataclass
class Credential:
"""A credential containing a token and optional metadata.
Attributes:
name: The service name this credential is for.
token: The secret token value.
expires_at: Optional expiration datetime (UTC).
scopes: Optional list of permission scopes.
source: Where the credential was loaded from (env, config, runtime).
"""
name: str
token: str
expires_at: datetime | None = None
scopes: list[str] = field(default_factory=list)
source: str = "unknown"
def is_expired(self) -> bool:
"""Check if the credential has expired.
Returns:
True if expires_at is set and is in the past, False otherwise.
"""
if self.expires_at is None:
return False
return datetime.now(UTC) > self.expires_at
def __repr__(self) -> str:
"""Safe representation that masks the token."""
return (
f"Credential(name={self.name!r}, token={mask_token(self.token)!r}, "
f"expires_at={self.expires_at!r}, scopes={self.scopes!r}, source={self.source!r})"
)
class CredentialRefreshError(Exception):
"""Raised when credential refresh fails or is not implemented."""
def __init__(self, service: str, message: str) -> None:
self.service = service
self.message = message
super().__init__(f"Failed to refresh credential for '{service}': {message}")
class CredentialManager:
"""Central manager for loading and accessing credentials.
Production reads resolve contextual opaque references through the OS-backed
vault. Plaintext environment and JSON sources are considered only when the
explicit legacy/operator flag is enabled. Runtime-only values and refresh
handlers remain supported without adding persistence authority.
Example:
>>> manager = CredentialManager()
>>> token = manager.get_token("email")
>>> if token is None:
... print("Email token not configured")
"""
def __init__(
self,
*,
credential_mapping: dict[str, str] | None = None,
config_path: Path | str | None = None,
env_prefix: str = ENV_PREFIX,
refresh_handlers: dict[str, Callable[[str], str]] | None = None,
vault: CredentialVaultBackend | None = None,
use_vault: bool = True,
scope: str = "household",
user_id: str | None = None,
vault_refs: dict[str, dict[str, Any]] | None = None,
) -> None:
"""Initialize the credential manager.
Args:
credential_mapping: Mapping of service names to env var names.
Defaults to DEFAULT_CREDENTIAL_MAPPING.
config_path: Path to JSON config file for overrides.
Defaults to config/credentials.json.
env_prefix: Prefix for environment variables (default "REX_").
refresh_handlers: Optional dict of service_name -> refresh callable.
Each callable takes the current token and returns a new one.
vault: Inject a specific credential vault backend (mainly for
tests). When omitted and `use_vault` is True, one is
resolved lazily via `rex.credential_vault.get_credential_vault`.
use_vault: Set False to never consult the vault (e.g. callers
that intentionally only want config/env sources).
scope: Vault scope to resolve when no `vault` is injected -
`"household"` (default, matches pre-vault global behavior)
or `"user"` (requires `user_id`).
user_id: Validated Rex user id, required when `scope="user"`.
"""
self._mapping = credential_mapping or DEFAULT_CREDENTIAL_MAPPING.copy()
self._config_path = Path(config_path) if config_path else DEFAULT_CONFIG_PATH
self._env_prefix = env_prefix
self._refresh_handlers = refresh_handlers or {}
self._credentials: dict[str, Credential] = {}
self._loaded = False
self._config_invalid = False
self._vault: CredentialVaultBackend | None = vault
self._use_vault = use_vault
self._vault_scope = scope
self._vault_user_id = user_id
self._vault_resolved = vault is not None
self._vault_refs = dict(vault_refs) if vault_refs is not None else None
def _load_from_env(self) -> None:
"""Load credentials from environment variables."""
for service_name, env_var in self._mapping.items():
# Try with prefix first, then without
full_var = f"{self._env_prefix}{env_var}"
token = os.getenv(full_var)
if token is None:
# Try without prefix (for standard vars like OPENAI_API_KEY)
token = os.getenv(env_var)
if token:
self._credentials[service_name] = Credential(
name=service_name,
token=token,
source="env",
)
logger.debug("Loaded credential for %s from environment", service_name)
def _load_from_config(self) -> None:
"""Load credentials from JSON config file if it exists."""
self._config_invalid = False
if not self._config_path.exists():
logger.debug("Credential config file not found at %s", self._config_path)
return
try:
with open(self._config_path, encoding="utf-8") as f:
config = json.load(f)
except (json.JSONDecodeError, OSError) as e:
logger.warning("Failed to load credential config: %s", e)
self._config_invalid = True
self._credentials.clear()
return
if not isinstance(config, dict):
logger.warning("Invalid credential config format (expected dict)")
self._config_invalid = True
self._credentials.clear()
return
credentials_section = config.get("credentials", config)
if not isinstance(credentials_section, dict):
logger.warning("Invalid credentials section format")
self._config_invalid = True
self._credentials.clear()
return
for service_name, cred_data in credentials_section.items():
if isinstance(cred_data, str):
# Simple format: just the token
token = cred_data
expires_at = None
scopes: list[str] = []
elif isinstance(cred_data, dict):
# Full format with metadata
token = cred_data.get("token") # type: ignore[assignment]
if not token:
continue
expires_at_str = cred_data.get("expires_at")
expires_at = None
if expires_at_str:
try:
expires_at = datetime.fromisoformat(expires_at_str.replace("Z", "+00:00"))
except ValueError:
logger.warning("Invalid expires_at format for %s", service_name)
scopes = cred_data.get("scopes", [])
if not isinstance(scopes, list):
scopes = []
else:
continue
# Config overrides env
self._credentials[service_name] = Credential(
name=service_name,
token=token,
expires_at=expires_at,
scopes=scopes,
source="config",
)
logger.debug("Loaded credential for %s from config", service_name)
def _get_vault(self) -> CredentialVaultBackend | None:
"""Lazily resolve and cache the vault backend for this manager.
A known platform/backend availability failure is cached as "no vault
available". Unexpected construction failures are surfaced so a
packaged process cannot silently downgrade or hide corruption.
"""
if self._vault is not None:
return self._vault
if not self._use_vault or self._vault_resolved:
return self._vault
self._vault_resolved = True
try:
from rex.credential_vault import VaultUnavailableError, get_credential_vault
self._vault = get_credential_vault(scope=self._vault_scope, user_id=self._vault_user_id)
except VaultUnavailableError as exc:
logger.debug("Credential vault unavailable: %s", exc)
self._vault = None
return self._vault
def _vault_key_for_service(self, service_name: str) -> str:
"""Map a service name to its vault key (mapped env var name, or itself)."""
return self._mapping.get(service_name, service_name)
def _configured_vault_refs(self) -> dict[str, dict[str, Any]]:
if self._vault_refs is not None:
section: object = self._vault_refs
else:
try:
from rex.config_manager import load_config
raw = load_config() or {}
except Exception as exc:
from rex.credential_vault import VaultCorruptedError
raise VaultCorruptedError(
"Credential reference registry could not be loaded"
) from exc
root = raw.get("credential_refs")
if root is None:
return {}
if not isinstance(root, dict):
from rex.credential_vault import VaultCorruptedError
raise VaultCorruptedError("Credential reference registry is invalid")
if self._vault_scope == "household":
section = root.get("household")
else:
users = root.get("users")
if users is not None and not isinstance(users, dict):
from rex.credential_vault import VaultCorruptedError
raise VaultCorruptedError("User credential reference registry is invalid")
section = users.get(self._vault_user_id) if isinstance(users, dict) else None
if section is None:
return {}
if not isinstance(section, dict) or any(
not isinstance(key, str) or not isinstance(value, dict)
for key, value in section.items()
):
from rex.credential_vault import VaultCorruptedError
raise VaultCorruptedError("Credential reference registry section is invalid")
return section
def _vault_context_for_service(
self,
service_name: str,
*,
integration: str | None = None,
account: str | None = None,
slot: str | None = None,
) -> tuple[str, str, str | None, str] | None:
logical_name = self._vault_key_for_service(service_name)
record = self._configured_vault_refs().get(logical_name)
if record is None:
return None
ref = record.get("ref")
allowed_fields = {"ref", "integration", "account", "slot"}
if "migrated_from" in record:
allowed_fields.add("migrated_from")
if (
set(record) != allowed_fields
or not isinstance(ref, str)
or (
"migrated_from" in record
and record.get("migrated_from") not in {"env", "credentials.json"}
)
):
from rex.credential_vault import VaultCorruptedError
raise VaultCorruptedError("Credential reference metadata is invalid")
from rex.credential_vault import VaultCorruptedError, validate_credential_ref
try:
validate_credential_ref(ref)
except ValueError as exc:
raise VaultCorruptedError("Credential reference metadata is invalid") from exc
derived_integration, derived_account, derived_slot = credential_context_for_name(
logical_name
)
expected_integration = integration or derived_integration
expected_slot = slot or derived_slot
expected_account = account if account is not None else derived_account
if (
record.get("integration") != expected_integration
or record.get("account") != expected_account
or record.get("slot") != expected_slot
):
from rex.credential_vault import VaultCorruptedError
raise VaultCorruptedError("Credential reference context metadata is invalid")
return ref, expected_integration, expected_account, expected_slot
def _load_from_vault(self) -> None:
"""Load credentials from the vault, if one is available.
Runs last in `_ensure_loaded()` so vault entries take priority over
both config file and environment sources, and independently of
whether the config file itself was valid.
"""
contexts: list[tuple[str, tuple[str, str, str | None, str]]] = []
for service_name in self._mapping:
context = self._vault_context_for_service(service_name)
if context is not None:
contexts.append((service_name, context))
if not contexts:
return
vault = self._get_vault()
if vault is None:
return
for service_name, context in contexts:
ref, integration, account, slot = context
try:
token = vault.get_secret(ref, integration=integration, account=account, slot=slot)
except Exception:
logger.warning("Failed to read configured vault credential for %s", service_name)
raise
if not token:
continue
self._credentials[service_name] = Credential(
name=service_name,
token=token,
source="vault",
)
logger.debug("Loaded credential for %s from vault", service_name)
def _ensure_loaded(self) -> None:
"""Ensure credentials are loaded (lazy loading).
Plaintext env/.env/config.json sources are only consulted when
`legacy_plaintext_fallback_enabled()` is true (an explicit,
non-production opt-in - see its docstring). By default this manager
is vault-only: a secret not present in the vault is simply absent,
never silently read from plaintext. The vault is always consulted
last regardless, so it wins when both a legacy source and the vault
have a value for the same service.
"""
if not self._loaded:
if legacy_plaintext_fallback_enabled():
self._load_from_env()
self._load_from_config()
if self._config_invalid:
self._credentials.clear()
self._load_from_vault()
self._loaded = True
def reload(self) -> None:
"""Reload all credentials from environment and config.
This clears the cache and reloads from all sources.
Runtime credentials set via set_token() will be preserved.
"""
runtime_creds = {
name: cred for name, cred in self._credentials.items() if cred.source == "runtime"
}
self._credentials.clear()
self._loaded = False
self._ensure_loaded()
# Restore runtime credentials
self._credentials.update(runtime_creds)
logger.info("Reloaded credentials")
def get_credential(
self,
service_name: str,
*,
integration: str | None = None,
account: str | None = None,
slot: str | None = None,
) -> Credential | None:
"""Get the full credential object for a service.
Args:
service_name: Name of the service (e.g., "email", "openai").
Returns:
Credential object or None if not found.
"""
self._ensure_loaded()
if service_name not in self._mapping and service_name not in self._credentials:
context = self._vault_context_for_service(
service_name, integration=integration, account=account, slot=slot
)
vault = self._get_vault() if context is not None else None
if vault is not None and context is not None:
ref, expected_integration, expected_account, expected_slot = context
token = vault.get_secret(
ref,
integration=expected_integration,
account=expected_account,
slot=expected_slot,
)
if token:
self._credentials[service_name] = Credential(
name=service_name, token=token, source="vault"
)
return self._credentials.get(service_name)
def get_token(
self,
service_name: str,
*,
auto_refresh: bool = True,
integration: str | None = None,
account: str | None = None,
slot: str | None = None,
) -> str | None:
"""Get the token for a service.
If the token is expired and auto_refresh is True, attempts to refresh it.
Args:
service_name: Name of the service (e.g., "email", "openai").
auto_refresh: Whether to attempt refresh if token is expired.
Returns:
Token string or None if not found or refresh failed.
"""
credential = self.get_credential(
service_name, integration=integration, account=account, slot=slot
)
if credential is None:
return None
if credential.is_expired() and auto_refresh:
try:
new_token = self.refresh_token(service_name)
return new_token
except CredentialRefreshError:
logger.warning("Token for %s is expired and refresh failed", service_name)
return None
return credential.token
def set_token(
self,
service_name: str,
token: str,
*,
expires_at: datetime | None = None,
scopes: list[str] | None = None,
persist: bool = False,
integration: str | None = None,
account: str | None = None,
slot: str | None = None,
credential_ref: str | None = None,
) -> str | None:
"""Set or update a token at runtime, optionally persisting it to the vault.
Args:
service_name: Name of the service.
token: The new token value.
expires_at: Optional expiration datetime (UTC).
scopes: Optional list of permission scopes.
persist: When True, write the token through to the credential
vault (in addition to updating the in-memory value). Raises
`rex.credential_vault.VaultUnavailableError` if no vault is
available - this method never falls back to writing
plaintext config/env on a persist failure.
"""
self._ensure_loaded()
logical_name = self._vault_key_for_service(service_name)
persisted_ref: str | None = None
if persist:
from rex.credential_vault import VaultUnavailableError, generate_credential_ref
vault = self._get_vault()
if vault is None:
raise VaultUnavailableError(
f"Cannot persist credential for {service_name!r}: "
"no credential vault is available on this platform."
)
derived_integration, derived_account, derived_slot = credential_context_for_name(
logical_name
)
expected_integration = integration or derived_integration
expected_slot = slot or derived_slot
expected_account = account if account is not None else derived_account
persisted_ref = credential_ref or generate_credential_ref()
vault.set_secret(
persisted_ref,
token,
integration=expected_integration,
account=expected_account,
slot=expected_slot,
)
if (
vault.get_secret(
persisted_ref,
integration=expected_integration,
account=expected_account,
slot=expected_slot,
)
!= token
):
raise VaultUnavailableError("Credential vault readback verification failed")
self._credentials[service_name] = Credential(
name=service_name,
token=token,
expires_at=expires_at,
scopes=scopes or [],
source="vault" if persist else "runtime",
)
logger.debug(
"Set credential for %s (%s)",
service_name,
"persisted to vault" if persist else "runtime",
)
return persisted_ref
def refresh_token(self, service_name: str) -> str:
"""Refresh a token for a service.
If a refresh handler is registered for the service, it will be called.
Otherwise, raises CredentialRefreshError.
Args:
service_name: Name of the service to refresh.
Returns:
The new token value.
Raises:
CredentialRefreshError: If refresh fails or no handler is registered.
"""
if service_name not in self._refresh_handlers:
raise CredentialRefreshError(
service_name,
"No refresh handler registered. Token refresh not implemented for this service.",
)
credential = self.get_credential(service_name)
current_token = credential.token if credential else ""
try:
handler = self._refresh_handlers[service_name]
new_token = handler(current_token)
except Exception as e:
raise CredentialRefreshError(service_name, str(e)) from e
# Update stored credential
self.set_token(
service_name,
new_token,
expires_at=credential.expires_at if credential else None,
scopes=credential.scopes if credential else None,
)
logger.info("Refreshed token for %s", service_name)
return new_token
def register_refresh_handler(
self,
service_name: str,
handler: Callable[[str], str],
) -> None:
"""Register a refresh handler for a service.
Args:
service_name: Name of the service.
handler: Callable that takes current token and returns new token.
"""
self._refresh_handlers[service_name] = handler
logger.debug("Registered refresh handler for %s", service_name)
def list_services(self) -> list[str]:
"""List all services that have credentials configured.
Returns:
List of service names with available credentials.
"""
self._ensure_loaded()
return list(self._credentials.keys())
def has_token(self, service_name: str) -> bool:
"""Check if a token is available for a service.
Args:
service_name: Name of the service.
Returns:
True if token exists and is not expired.
"""
credential = self.get_credential(service_name)
if credential is None:
return False
return not credential.is_expired()
def add_credential_mapping(self, service_name: str, env_var: str) -> None:
"""Add a custom credential mapping.
Args:
service_name: Name of the service.
env_var: Environment variable name (without prefix).
"""
self._mapping[service_name] = env_var
# If already loaded, check for this new mapping
if self._loaded:
token = None
if legacy_plaintext_fallback_enabled():
token = os.getenv(f"{self._env_prefix}{env_var}") or os.getenv(env_var)
if token and service_name not in self._credentials:
self._credentials[service_name] = Credential(
name=service_name,
token=token,
source="env",
)
def get_credential_info(self, service_name: str) -> dict[str, Any] | None:
"""Get non-secret credential state without deriving output from the token.
Args:
service_name: Name of the service.
Returns:
Dict with credential metadata or None.
"""
credential = self.get_credential(service_name)
if credential is None:
return None
return {
"name": credential.name,
"has_credential": True,
"expires_at": credential.expires_at.isoformat() if credential.expires_at else None,
"scopes": credential.scopes,
"source": credential.source,
"is_expired": credential.is_expired(),
}
# Global credential manager instance
_credential_manager: CredentialManager | None = None
def get_credential_manager() -> CredentialManager:
"""Get the global credential manager instance.
Creates a new instance if one doesn't exist.
Returns:
The global CredentialManager instance.
"""
global _credential_manager
if _credential_manager is None:
_credential_manager = CredentialManager()
return _credential_manager
def set_credential_manager(manager: CredentialManager | None) -> None:
"""Set the global credential manager instance.
The global singleton must remain household/system-scoped. Installing a
per-user-scoped manager here would make one user's credentials the
process-wide default, racing across overlapping requests for other
users (the same global-mutable-identity hazard documented for
`Assistant` in CLAUDE.md). Per-user credential access must construct
and use a local `CredentialManager(scope="user", user_id=...)` instead
of routing through this global.
Args:
manager: The CredentialManager instance to use globally, or None to
clear it.
Raises:
ValueError: If `manager` is scoped to a specific user.
"""
global _credential_manager
if manager is not None and getattr(manager, "_vault_scope", "household") == "user":
raise ValueError(
"The global credential manager must remain household-scoped; "
"construct a per-user CredentialManager(scope='user', user_id=...) "
"locally instead of installing one globally."
)
_credential_manager = manager
def get_persisted_credential(
logical_name: str,
*,
scope: str = "household",
user_id: str | None = None,
integration: str | None = None,
account: str | None = None,
slot: str | None = None,
) -> str | None:
"""Resolve one credential with a request-local, scope-bound manager.
This helper deliberately never uses the global manager, so user-scoped
secrets cannot survive in process-global cache across requests.
"""
manager = CredentialManager(
credential_mapping={logical_name: logical_name},
scope=scope,
user_id=user_id,
)
return manager.get_token(
logical_name,
integration=integration,
account=account,
slot=slot,
)
__all__ = [
"Credential",
"CredentialManager",
"CredentialRefreshError",
"LEGACY_PLAINTEXT_FALLBACK_ENV_VAR",
"get_credential_manager",
"legacy_plaintext_fallback_enabled",
"set_credential_manager",
"mask_token",
"REDACTED_MARKER",
"DEFAULT_CREDENTIAL_MAPPING",
"credential_context_for_name",
"get_persisted_credential",
]