Add files using upload-large-folder tool

f76c374 verified about 2 months ago

11.3 kB

"""Cross-adapter helpers shared by gateway/platforms/* and plugins/platforms/*.
Kept dependency-light (stdlib + ``agent.secret_scope``) so every adapter can
import it at module top level without cycles.
"""

from __future__ import annotations

import contextlib
import json
import logging
import os
from typing import Any, Callable, Iterable, Optional

# Profile-scoped secret reader for multiplexing support (PR #50094)
from agent.secret_scope import UnscopedSecretError as _UnscopedSecretError
from agent.secret_scope import get_secret as _scoped_get_secret

logger = logging.getLogger(__name__)


def get_scoped_secret(name: str, default: Any = None, *, external_fallback: bool = False) -> Any:
    """Scope-aware credential read with the default-profile startup fallback.
    An installed profile secret scope is authoritative: a scoped miss returns
    ``default`` (never borrow another profile's value from ``os.environ``).
    The DEFAULT profile constructs and sends *unscoped* under multiplexing,
    where a bare ``get_secret`` raises ``UnscopedSecretError``; there
    ``os.environ`` is that profile's own value, so fall back to it.
    ``external_fallback`` adds one rung for startup gates that run before any
    scope exists: an unscoped miss consults a one-shot build of the profile's
    own secret scope, so externally managed credentials (Bitwarden etc., only
    ``BWS_ACCESS_TOKEN`` in ``.env``) are visible to ``check_requirements``.
    """
    try:
        val = _scoped_get_secret(name, None)
    except _UnscopedSecretError:
        val = os.getenv(name)
    if val is None and external_fallback and _current_scope() is None:
        val = _unscoped_profile_secrets().get(name)
    return val if val is not None else default


def _current_scope():
    from agent.secret_scope import current_secret_scope
    return current_secret_scope()


_UNSCOPED_PROFILE_SECRETS: Optional[dict] = None


def _unscoped_profile_secrets() -> dict:
    """Process-cached profile secret mapping (external resolvers are slow); failures degrade to {}."""
    global _UNSCOPED_PROFILE_SECRETS
    if _UNSCOPED_PROFILE_SECRETS is None:
        try:
            from agent.secret_scope import build_profile_secret_scope
            from hermes_constants import get_hermes_home
            _UNSCOPED_PROFILE_SECRETS = dict(build_profile_secret_scope(get_hermes_home()))
        except Exception:
            logger.warning(
                "Could not build the profile secret scope; externally managed credentials will not be "
                "visible to the startup gate (#95216)", exc_info=True)
            _UNSCOPED_PROFILE_SECRETS = {}
    return _UNSCOPED_PROFILE_SECRETS


def platform_gate_env(name: str, default: str = "") -> str:
    """Allow/deny gate env read with per-profile isolation, always stripped.
    With a profile secret scope installed AND multiplexing active, a scoped miss returns ``default``
    instead of falling through to ``os.environ``, which may hold ANOTHER profile's first-writer
    bridged value (the YAML→env bridges are first-writer-wins; allowlist leak, #72348).
    Single-profile deployments behave exactly like ``os.getenv``.
    """
    if not name:
        return default
    with contextlib.suppress(Exception):
        from agent.secret_scope import current_secret_scope, is_multiplex_active

        scope = current_secret_scope()
        if scope is not None and is_multiplex_active():
            val = scope.get(name)
            return default if val is None else str(val).strip()
    return (os.getenv(name) or default).strip()


def decode_json_list_literal(raw):
    """Decode a JSON-encoded allowlist written by ``hermes config set``.
    String-typed defaults keep list literals verbatim on write (``allowed_chats`` is
    declared as ``""``), so the config can hold ``'["-100","-200"]'`` as a string.
    Malformed JSON passes through unchanged and keeps the legacy comma-split path.
    """
    if isinstance(raw, str) and raw.lstrip()[:1] == "[":
        try:
            loaded = json.loads(raw)
        except ValueError:
            return raw
        if isinstance(loaded, list):
            return loaded
    return raw


def extra_or_secret(extra: Optional[dict], key: str, env: str, default: Any = "",
                    *, blank_is_unset: bool = True) -> Any:
    """The ONE per-profile setting reader: explicit env ``env`` → the profile's YAML
    ``config.extra[key]`` → ``default``.
    The env rung is the owning profile's, read through ``get_scoped_secret``: a secondary
    multiplex profile sees its own ``.env`` and a miss falls to ITS YAML, never to the launch
    process's ``os.environ`` (which holds the default profile's bridged values); single-profile
    and default-profile installs read ``os.environ`` there, keeping the documented env-over-YAML
    contract (an explicit ``DISCORD_ALLOW_MENTION_EVERYONE=false`` beats ``everyone: true``,
    #108440; ``TELEGRAM_REACTIONS=true`` beats the stock ``reactions: false``, #109032). A blank
    env value is unset. An explicit ``False``/``0`` in YAML is a real value (``require_mention:
    false`` must not fall to ``default``). A blank YAML string is unset by default; readers whose
    YAML key means "clear it" (``allowed_channels: "" `` = no whitelist) pass
    ``blank_is_unset=False`` so only a missing/``None`` key falls through.
    """
    if env:
        env_value = get_scoped_secret(env, None)
        if env_value is not None and str(env_value).strip():
            return env_value
    value = (extra or {}).get(key)
    if value is None or (blank_is_unset and isinstance(value, str) and not value.strip()):
        return default
    return value


def profile_scoped() -> bool:
    """True when running inside a multiplexed secondary profile's scope.
    Secondary-profile adapters are constructed/connected inside
    ``_profile_runtime_scope`` (secret scope installed + multiplex active).
    The DEFAULT profile under multiplexing runs unscoped and keeps the legacy
    ``os.environ`` precedence, so YAML->env bridges must skip only when True.
    """
    try:
        from agent.secret_scope import current_secret_scope, is_multiplex_active
        return bool(is_multiplex_active() and current_secret_scope() is not None)
    except Exception:
        return False


# --------------------------------------------------------------------------- YAML → env config bridge
# (apply_yaml_config_fn, #25443)
# ---------------------------------------------------------------------------

def yaml_env_setter() -> Callable[[str, Any], None]:
    """``set_env(name, value)`` for ``apply_yaml_config_fn`` hooks: writes ``os.environ[name]`` only
    when the var is unset (explicit env wins over YAML) and NEVER while a multiplexed secondary
    profile's scope is active — the gateway loads every secondary's config inside
    ``_profile_runtime_scope``, so a write there would pin that profile's policy process-wide and the
    default profile's adapters would read it as their own (first-writer-wins poisoning, #80099).
    Hooks seed the same values into the returned ``extra`` so each profile's adapter reads its own.
    Lists are comma-joined; ``None`` is skipped.
    """
    skip = profile_scoped()

    def set_env(name: str, value: Any) -> None:
        if value is None or skip or os.getenv(name):
            return
        os.environ[name] = ",".join(str(v) for v in value) if isinstance(value, list) else str(value)

    return set_env


def send_error(message: Any) -> dict:
    """Standalone-sender failure envelope with vendor exception text redacted (the same helper
    ``send_message`` uses), so a token or signed URL in an httpx/aiohttp error never reaches the
    model transcript."""
    from tools.send_message_senders import _error
    return _error(str(message))


# kind -> (applies-when predicate over (cfg, key), env encoder). "lower"/"json" bridge whenever the key
# is present (YAML ``none`` still writes "none"); "str" skips null/blank, "csv" skips null and leaves
# lists to the setter's comma join.
_YAML_KINDS: dict[str, tuple[Callable[[dict, str], bool], Callable[[Any], Any]]] = {
    "lower": (lambda cfg, key: key in cfg, lambda v: str(v).lower()),
    "str": (lambda cfg, key: cfg.get(key) not in (None, ""), str),
    "csv": (lambda cfg, key: cfg.get(key) is not None, lambda v: v),
    "json": (lambda cfg, key: key in cfg, json.dumps),
}


def apply_yaml_bridge(cfg: dict, spec: Iterable[tuple[str, str, str]]) -> dict | None:
    """Table-driven ``apply_yaml_config_fn`` body: for each ``(yaml_key, ENV_VAR, kind)`` row seed the
    original YAML value into the returned ``extra`` and bridge it to env through ``yaml_env_setter``
    (env wins; skipped under a secondary profile's scope). ``None`` when nothing matched.
    """
    set_env = yaml_env_setter()
    seeded: dict = {}
    for key, env, kind in spec:
        applies, encode = _YAML_KINDS[kind]
        if applies(cfg, key):
            seeded[key] = cfg[key]
            set_env(env, encode(cfg[key]))
    return seeded or None


# --------------------------------------------------------------------------- env → extra seeding
# (env_enablement_fn)
# ---------------------------------------------------------------------------

def seed_extra_from_env(spec: Iterable[tuple[str, str, Callable[[str], Any] | None]], *,
                        home_env: str | None = None, home_default: str = "") -> dict:
    """Table-driven ``env_enablement_fn`` body: for each ``(ENV_VAR, extra_key, conv)`` row read the
    profile-scoped value, skip it when blank or when ``conv`` raises ``ValueError``, else seed
    ``conv(value)`` (``None`` conv keeps the stripped string).
    ``home_env`` seeds ``home_channel`` as ``{"chat_id", "name"}`` from ``<home_env>`` (else
    ``home_default``) with ``<home_env>_NAME`` naming it — the core hook lifts that dict into a
    ``HomeChannel``. The name falls back to the literal ``"Home"``, the same rule the built-in
    platforms use.
    """
    seed: dict = {}
    for env, key, conv in spec:
        raw = str(get_scoped_secret(env, "") or "").strip()
        if not raw:
            continue
        with contextlib.suppress(ValueError):
            seed[key] = conv(raw) if conv else raw
    if home_env:
        home = str(get_scoped_secret(home_env, "") or "").strip() or home_default
        if home:
            seed["home_channel"] = {"chat_id": home, "name": get_scoped_secret(f"{home_env}_NAME", "Home")}
    return seed


def env_is_connected(*names: str) -> Callable[[Any], bool]:
    """``is_connected`` for platforms whose only configuration is env: True when every ``names`` var is
    non-blank. Resolves ``hermes_cli.gateway.get_env_value`` at call time (scope-aware, .env-backed) so
    setup-status tests that patch it see the same value."""

    def is_connected(config: Any) -> bool:
        import hermes_cli.gateway as gateway_mod
        return all((gateway_mod.get_env_value(name) or "").strip() for name in names)

    return is_connected


def coerce_port(value: Any, default: int) -> int:
    """``int(value)`` or ``default`` when unparseable."""
    try:
        return int(value)
    except (TypeError, ValueError):
        return default

	"""Cross-adapter helpers shared by gateway/platforms/* and plugins/platforms/*.

	Kept dependency-light (stdlib + ``agent.secret_scope``) so every adapter can
	import it at module top level without cycles.
	"""

	from __future__ import annotations

	import contextlib
	import json
	import logging
	import os
	from typing import Any, Callable, Iterable, Optional

	# Profile-scoped secret reader for multiplexing support (PR #50094)
	from agent.secret_scope import UnscopedSecretError as _UnscopedSecretError
	from agent.secret_scope import get_secret as _scoped_get_secret

	logger = logging.getLogger(__name__)


	def get_scoped_secret(name: str, default: Any = None, *, external_fallback: bool = False) -> Any:
	"""Scope-aware credential read with the default-profile startup fallback.

	An installed profile secret scope is authoritative: a scoped miss returns
	``default`` (never borrow another profile's value from ``os.environ``).
	The DEFAULT profile constructs and sends unscoped under multiplexing,
	where a bare ``get_secret`` raises ``UnscopedSecretError``; there
	``os.environ`` is that profile's own value, so fall back to it.

	``external_fallback`` adds one rung for startup gates that run before any
	scope exists: an unscoped miss consults a one-shot build of the profile's
	own secret scope, so externally managed credentials (Bitwarden etc., only
	``BWS_ACCESS_TOKEN`` in ``.env``) are visible to ``check_requirements``.
	"""
	try:
	val = _scoped_get_secret(name, None)
	except _UnscopedSecretError:
	val = os.getenv(name)
	if val is None and external_fallback and _current_scope() is None:
	val = _unscoped_profile_secrets().get(name)
	return val if val is not None else default


	def _current_scope():
	from agent.secret_scope import current_secret_scope
	return current_secret_scope()


	_UNSCOPED_PROFILE_SECRETS: Optional[dict] = None


	def _unscoped_profile_secrets() -> dict:
	"""Process-cached profile secret mapping (external resolvers are slow); failures degrade to {}."""
	global _UNSCOPED_PROFILE_SECRETS
	if _UNSCOPED_PROFILE_SECRETS is None:
	try:
	from agent.secret_scope import build_profile_secret_scope
	from hermes_constants import get_hermes_home
	_UNSCOPED_PROFILE_SECRETS = dict(build_profile_secret_scope(get_hermes_home()))
	except Exception:
	logger.warning(
	"Could not build the profile secret scope; externally managed credentials will not be "
	"visible to the startup gate (#95216)", exc_info=True)
	_UNSCOPED_PROFILE_SECRETS = {}
	return _UNSCOPED_PROFILE_SECRETS


	def platform_gate_env(name: str, default: str = "") -> str:
	"""Allow/deny gate env read with per-profile isolation, always stripped.

	With a profile secret scope installed AND multiplexing active, a scoped miss returns ``default``
	instead of falling through to ``os.environ``, which may hold ANOTHER profile's first-writer
	bridged value (the YAML→env bridges are first-writer-wins; allowlist leak, #72348).
	Single-profile deployments behave exactly like ``os.getenv``.
	"""
	if not name:
	return default
	with contextlib.suppress(Exception):
	from agent.secret_scope import current_secret_scope, is_multiplex_active

	scope = current_secret_scope()
	if scope is not None and is_multiplex_active():
	val = scope.get(name)
	return default if val is None else str(val).strip()
	return (os.getenv(name) or default).strip()


	def decode_json_list_literal(raw):
	"""Decode a JSON-encoded allowlist written by ``hermes config set``.

	String-typed defaults keep list literals verbatim on write (``allowed_chats`` is
	declared as ``""``), so the config can hold ``'["-100","-200"]'`` as a string.
	Malformed JSON passes through unchanged and keeps the legacy comma-split path.
	"""
	if isinstance(raw, str) and raw.lstrip()[:1] == "[":
	try:
	loaded = json.loads(raw)
	except ValueError:
	return raw
	if isinstance(loaded, list):
	return loaded
	return raw


	def extra_or_secret(extra: Optional[dict], key: str, env: str, default: Any = "",
	*, blank_is_unset: bool = True) -> Any:
	"""The ONE per-profile setting reader: explicit env ``env`` → the profile's YAML
	``config.extra[key]`` → ``default``.

	The env rung is the owning profile's, read through ``get_scoped_secret``: a secondary
	multiplex profile sees its own ``.env`` and a miss falls to ITS YAML, never to the launch
	process's ``os.environ`` (which holds the default profile's bridged values); single-profile
	and default-profile installs read ``os.environ`` there, keeping the documented env-over-YAML
	contract (an explicit ``DISCORD_ALLOW_MENTION_EVERYONE=false`` beats ``everyone: true``,
	#108440; ``TELEGRAM_REACTIONS=true`` beats the stock ``reactions: false``, #109032). A blank
	env value is unset. An explicit ``False``/``0`` in YAML is a real value (``require_mention:
	false`` must not fall to ``default``). A blank YAML string is unset by default; readers whose
	YAML key means "clear it" (``allowed_channels: "" `` = no whitelist) pass
	``blank_is_unset=False`` so only a missing/``None`` key falls through.
	"""
	if env:
	env_value = get_scoped_secret(env, None)
	if env_value is not None and str(env_value).strip():
	return env_value
	value = (extra or {}).get(key)
	if value is None or (blank_is_unset and isinstance(value, str) and not value.strip()):
	return default
	return value


	def profile_scoped() -> bool:
	"""True when running inside a multiplexed secondary profile's scope.

	Secondary-profile adapters are constructed/connected inside
	``_profile_runtime_scope`` (secret scope installed + multiplex active).
	The DEFAULT profile under multiplexing runs unscoped and keeps the legacy
	``os.environ`` precedence, so YAML->env bridges must skip only when True.
	"""
	try:
	from agent.secret_scope import current_secret_scope, is_multiplex_active
	return bool(is_multiplex_active() and current_secret_scope() is not None)
	except Exception:
	return False


	# --------------------------------------------------------------------------- YAML → env config bridge
	# (apply_yaml_config_fn, #25443)
	# ---------------------------------------------------------------------------

	def yaml_env_setter() -> Callable[[str, Any], None]:
	"""``set_env(name, value)`` for ``apply_yaml_config_fn`` hooks: writes ``os.environ[name]`` only
	when the var is unset (explicit env wins over YAML) and NEVER while a multiplexed secondary
	profile's scope is active — the gateway loads every secondary's config inside
	``_profile_runtime_scope``, so a write there would pin that profile's policy process-wide and the
	default profile's adapters would read it as their own (first-writer-wins poisoning, #80099).
	Hooks seed the same values into the returned ``extra`` so each profile's adapter reads its own.
	Lists are comma-joined; ``None`` is skipped.
	"""
	skip = profile_scoped()

	def set_env(name: str, value: Any) -> None:
	if value is None or skip or os.getenv(name):
	return
	os.environ[name] = ",".join(str(v) for v in value) if isinstance(value, list) else str(value)

	return set_env


	def send_error(message: Any) -> dict:
	"""Standalone-sender failure envelope with vendor exception text redacted (the same helper
	``send_message`` uses), so a token or signed URL in an httpx/aiohttp error never reaches the
	model transcript."""
	from tools.send_message_senders import _error
	return _error(str(message))


	# kind -> (applies-when predicate over (cfg, key), env encoder). "lower"/"json" bridge whenever the key
	# is present (YAML ``none`` still writes "none"); "str" skips null/blank, "csv" skips null and leaves
	# lists to the setter's comma join.
	_YAML_KINDS: dict[str, tuple[Callable[[dict, str], bool], Callable[[Any], Any]]] = {
	"lower": (lambda cfg, key: key in cfg, lambda v: str(v).lower()),
	"str": (lambda cfg, key: cfg.get(key) not in (None, ""), str),
	"csv": (lambda cfg, key: cfg.get(key) is not None, lambda v: v),
	"json": (lambda cfg, key: key in cfg, json.dumps),
	}


	def apply_yaml_bridge(cfg: dict, spec: Iterable[tuple[str, str, str]]) -> dict \| None:
	"""Table-driven ``apply_yaml_config_fn`` body: for each ``(yaml_key, ENV_VAR, kind)`` row seed the
	original YAML value into the returned ``extra`` and bridge it to env through ``yaml_env_setter``
	(env wins; skipped under a secondary profile's scope). ``None`` when nothing matched.
	"""
	set_env = yaml_env_setter()
	seeded: dict = {}
	for key, env, kind in spec:
	applies, encode = _YAML_KINDS[kind]
	if applies(cfg, key):
	seeded[key] = cfg[key]
	set_env(env, encode(cfg[key]))
	return seeded or None


	# --------------------------------------------------------------------------- env → extra seeding
	# (env_enablement_fn)
	# ---------------------------------------------------------------------------

	def seed_extra_from_env(spec: Iterable[tuple[str, str, Callable[[str], Any] \| None]], *,
	home_env: str \| None = None, home_default: str = "") -> dict:
	"""Table-driven ``env_enablement_fn`` body: for each ``(ENV_VAR, extra_key, conv)`` row read the
	profile-scoped value, skip it when blank or when ``conv`` raises ``ValueError``, else seed
	``conv(value)`` (``None`` conv keeps the stripped string).

	``home_env`` seeds ``home_channel`` as ``{"chat_id", "name"}`` from ``<home_env>`` (else
	``home_default``) with ``<home_env>_NAME`` naming it — the core hook lifts that dict into a
	``HomeChannel``. The name falls back to the literal ``"Home"``, the same rule the built-in
	platforms use.
	"""
	seed: dict = {}
	for env, key, conv in spec:
	raw = str(get_scoped_secret(env, "") or "").strip()
	if not raw:
	continue
	with contextlib.suppress(ValueError):
	seed[key] = conv(raw) if conv else raw
	if home_env:
	home = str(get_scoped_secret(home_env, "") or "").strip() or home_default
	if home:
	seed["home_channel"] = {"chat_id": home, "name": get_scoped_secret(f"{home_env}_NAME", "Home")}
	return seed


	def env_is_connected(*names: str) -> Callable[[Any], bool]:
	"""``is_connected`` for platforms whose only configuration is env: True when every ``names`` var is
	non-blank. Resolves ``hermes_cli.gateway.get_env_value`` at call time (scope-aware, .env-backed) so
	setup-status tests that patch it see the same value."""

	def is_connected(config: Any) -> bool:
	import hermes_cli.gateway as gateway_mod
	return all((gateway_mod.get_env_value(name) or "").strip() for name in names)

	return is_connected


	def coerce_port(value: Any, default: int) -> int:
	"""``int(value)`` or ``default`` when unparseable."""
	try:
	return int(value)
	except (TypeError, ValueError):
	return default