fix(buzz): let the requirement gate see externally managed secrets

check_requirements() runs at gateway startup before any per-profile
secret scope is installed, and the scope-less get_secret path reads
only os.environ -- so a Bitwarden-managed BUZZ_PRIVATE_KEY (only
BWS_ACCESS_TOKEN in .env) was invisible to the platform gate and Buzz
was silently skipped with a misleading install hint (#95216). When no
scope is active and the process env has no value, consult a cached
one-shot build of the profile secret mapping (build_profile_secret_scope
resolves external secret sources); an active scope still shadows this
rung entirely, so multiplexed cross-profile isolation is unchanged.
BUZZ_RELAY_URL reads in the gate now go through the same helper so an
externally managed relay passes too.
This commit is contained in:
liuhao1024
2026-08-26 11:56:56 +08:00
committed by Teknium
parent c83121cf5f
commit a684d154bc
2 changed files with 186 additions and 5 deletions

View File

@@ -50,6 +50,7 @@ from typing import Any, Dict, List, Optional, Tuple
from urllib.parse import urlsplit, urlunsplit
from agent.secret_scope import UnscopedSecretError as _UnscopedSecretError
from agent.secret_scope import current_secret_scope as _current_secret_scope
from agent.secret_scope import get_secret as _scoped_get_secret
@@ -65,11 +66,58 @@ def _get_scoped_secret(name, default=None):
profile's own value, so fall back to it. Same pattern as the Slack
``SLACK_APP_TOKEN`` read (#59739) and
``gateway/platforms/whatsapp_common.py::_get_wsecret``.
The no-scope path has one more rung for the platform requirement gate:
``check_requirements()`` runs at gateway startup BEFORE any per-profile
secret scope is installed, and ``get_secret`` without a scope simply
reads ``os.environ`` — so a Bitwarden-managed ``BUZZ_PRIVATE_KEY``
(only ``BWS_ACCESS_TOKEN`` in ``.env``) was invisible to the check and
Buzz was silently skipped (#95216). When no scope is active and the
process env has no value, consult a one-shot build of the profile's
secret mapping (``build_profile_secret_scope`` resolves external secret
sources) so externally managed credentials pass the gate. An ACTIVE
scope still shadows this rung entirely — it never runs under
multiplexing, so cross-profile isolation is unchanged.
"""
try:
val = _scoped_get_secret(name, default)
val = _scoped_get_secret(name, None)
except _UnscopedSecretError:
val = os.getenv(name)
if val is None and _current_secret_scope() is None:
val = _unscoped_profile_secrets().get(name)
return val if val is not None else default
_UNSCOPED_PROFILE_SECRETS: Optional[Dict[str, str]] = None
def _unscoped_profile_secrets() -> Dict[str, str]:
"""One-shot build of the active profile's secret mapping.
Cached for the process: the build shells out to external secret
resolvers (Bitwarden via ``BWS_ACCESS_TOKEN``), and the requirement
gate / validate / is_connected probes all want the same snapshot. Any
failure degrades to an empty mapping — callers then simply report the
platform as not configured, which is the pre-fix behavior.
"""
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(
"Buzz requirement probe could not build the profile secret "
"scope; Bitwarden-managed credentials will not be visible "
"to the startup gate (#95216)",
exc_info=True,
)
_UNSCOPED_PROFILE_SECRETS = {}
return _UNSCOPED_PROFILE_SECRETS
return val if val is not None else default
@@ -1358,16 +1406,24 @@ def check_requirements() -> bool:
extra = _profile_buzz_extra()
relay = str(extra.get("relay_url") or "").strip()
return bool(relay and _resolve_private_key(extra))
if not os.getenv("BUZZ_RELAY_URL", "").strip():
# Scope-aware read: the gate runs before per-profile scopes install, and
# BUZZ_RELAY_URL can be externally managed just like the key (#95216).
if not (_get_scoped_secret("BUZZ_RELAY_URL", "") or "").strip():
return False
return bool(_resolve_private_key())
def validate_config(config) -> bool:
"""Validate that the platform config has enough info to connect."""
"""Validate that the platform config has enough information to connect."""
extra = getattr(config, "extra", {}) or {}
relay = _scoped_platform_setting("BUZZ_RELAY_URL", extra, "relay_url")
relay = relay if relay is not None else extra.get("relay_url", "")
# Inside a secondary profile scope, extra is authoritative (#98738);
# unscoped, the env read gains the external-secret rung so a managed
# relay passes too (#95216).
if _profile_scoped():
relay = _scoped_platform_setting("BUZZ_RELAY_URL", extra, "relay_url")
relay = relay if relay is not None else extra.get("relay_url", "")
else:
relay = _get_scoped_secret("BUZZ_RELAY_URL", "") or extra.get("relay_url", "")
return bool(relay and _resolve_private_key(extra))

View File

@@ -0,0 +1,125 @@
"""Regression tests for the Buzz requirement gate vs external secrets (#95216).
``check_requirements()`` runs at gateway startup BEFORE the per-profile
secret scope is installed, so a Bitwarden-managed ``BUZZ_PRIVATE_KEY`` (only
``BWS_ACCESS_TOKEN`` in ``.env``) was invisible to the bare env read and Buzz
was silently skipped. The fix adds a one-shot ``build_profile_secret_scope``
consultation to the unscoped fallback of ``_get_scoped_secret``.
The key values below are synthesized placeholders (never a usable secret).
"""
import os
import pytest
# Synthesized, non-credential placeholder (any non-empty string exercises
# the gate; no real key material is ever embedded here).
_STUB_KEY = os.environ.get("BUZZ_TEST_STUB_KEY") or ("k" * 8)
@pytest.fixture(autouse=True)
def _reset_unscoped_cache():
import plugins.platforms.buzz.adapter as adapter
prev = adapter._UNSCOPED_PROFILE_SECRETS
adapter._UNSCOPED_PROFILE_SECRETS = None
yield
adapter._UNSCOPED_PROFILE_SECRETS = prev
def _install_fake_scope(monkeypatch, secrets):
"""Point build_profile_secret_scope at a fake external-secret snapshot."""
import agent.secret_scope as secret_scope
calls = []
def fake_build(home): # noqa: ANN001 - test double
calls.append(home)
return dict(secrets)
monkeypatch.setattr(secret_scope, "build_profile_secret_scope", fake_build)
return calls
class TestRequirementGateSeesExternalSecrets:
def test_externally_managed_key_passes_gate(self, monkeypatch):
import plugins.platforms.buzz.adapter as adapter
monkeypatch.delenv("BUZZ_RELAY_URL", raising=False)
monkeypatch.delenv("BUZZ_PRIVATE_KEY", raising=False)
monkeypatch.setenv("BWS_ACCESS_TOKEN", "stub-token")
_install_fake_scope(
monkeypatch,
{"BUZZ_RELAY_URL": "wss://relay.example", "BUZZ_PRIVATE_KEY": _STUB_KEY},
)
assert adapter.check_requirements() is True
def test_gate_fails_cleanly_when_nothing_resolves(self, monkeypatch):
import plugins.platforms.buzz.adapter as adapter
monkeypatch.delenv("BUZZ_RELAY_URL", raising=False)
monkeypatch.delenv("BUZZ_PRIVATE_KEY", raising=False)
_install_fake_scope(monkeypatch, {})
assert adapter.check_requirements() is False
def test_relay_from_env_still_passes_with_external_key(self, monkeypatch):
import plugins.platforms.buzz.adapter as adapter
monkeypatch.setenv("BUZZ_RELAY_URL", "wss://relay.example")
monkeypatch.delenv("BUZZ_PRIVATE_KEY", raising=False)
_install_fake_scope(monkeypatch, {"BUZZ_PRIVATE_KEY": _STUB_KEY})
assert adapter.check_requirements() is True
def test_profile_scope_build_failure_degrades_to_not_configured(
self, monkeypatch
):
import agent.secret_scope as secret_scope
import plugins.platforms.buzz.adapter as adapter
monkeypatch.delenv("BUZZ_RELAY_URL", raising=False)
monkeypatch.delenv("BUZZ_PRIVATE_KEY", raising=False)
def boom(home): # noqa: ANN001 - test double
raise RuntimeError("external secret resolver unavailable")
monkeypatch.setattr(secret_scope, "build_profile_secret_scope", boom)
assert adapter.check_requirements() is False
def test_scope_snapshot_is_built_once_and_cached(self, monkeypatch):
import plugins.platforms.buzz.adapter as adapter
monkeypatch.setenv("BUZZ_RELAY_URL", "wss://relay.example")
monkeypatch.delenv("BUZZ_PRIVATE_KEY", raising=False)
calls = _install_fake_scope(monkeypatch, {"BUZZ_PRIVATE_KEY": _STUB_KEY})
assert adapter.check_requirements() is True
assert adapter.check_requirements() is True
assert adapter.validate_config(type("Cfg", (), {"extra": {}})()) is True
assert len(calls) == 1, "the external-secret snapshot must be cached"
class TestScopedSemanticsUnchanged:
def test_active_scope_miss_does_not_fall_through_to_unscoped_build(
self, monkeypatch
):
"""A scoped miss must keep returning default: the unscoped build is
only for the no-scope startup gate, never a cross-profile borrow."""
import agent.secret_scope as secret_scope
import plugins.platforms.buzz.adapter as adapter
monkeypatch.setenv("BUZZ_RELAY_URL", "wss://relay.example")
monkeypatch.delenv("BUZZ_PRIVATE_KEY", raising=False)
calls = _install_fake_scope(
monkeypatch, {"BUZZ_PRIVATE_KEY": _STUB_KEY * 2}
)
token = secret_scope.set_secret_scope({}) # active, empty scope
try:
assert adapter.check_requirements() is False
assert calls == [], "an active scope must shadow the unscoped build"
finally:
secret_scope.reset_secret_scope(token)