feat(secrets): orchestrator-level preserve_existing + profile aliasing (#69058)

Fixes the profile-clobber bug cluster at the apply_all() chokepoint so
every secret source — bundled and plugin — gets both behaviors for free:

- secrets.preserve_existing (#58073): env var names whose existing .env /
  shell value always wins, even against a source with
  override_existing: true.  Escape hatch for per-profile platform
  secrets while everything else rotates centrally.
- Profile aliasing (#51447): under a named profile, an applied
  FOO_<PROFILE> var (credential-shaped suffixes only) also hydrates the
  canonical FOO, so adapters/plugins that read fixed env names see the
  profile's value.  Direct supply beats alias; protected/claimed/
  override guards all apply; secrets.profile_alias: false disables.

Reimplements the intent of PR #58085 (tianma-if, preserve_existing on the
legacy Bitwarden apply shim) and PR #51616 (LeonSGP43, profile aliasing
inside the Bitwarden backend) on the SecretSource orchestrator that
superseded those code paths.

Fixes #58073.  Fixes #51447.

Co-authored-by: tianma-if <5895871+tianma-if@users.noreply.github.com>
Co-authored-by: LeonSGP43 <154585401+LeonSGP43@users.noreply.github.com>

agent/secret_sources/registry.py

@@ -29,6 +29,7 @@ from __future__ import annotations
 
 import concurrent.futures
 import logging
+import os
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Dict, List, Optional
@@ -275,6 +276,43 @@ def _ordered_enabled_sources(secrets_cfg: dict) -> List[SecretSource]:
     return enabled
 
 
+def _active_profile_name(home_path: Optional[Path]) -> str:
+    """Best-effort active profile name for profile-scoped secret aliases.
+
+    A named profile's HERMES_HOME is ``~/.hermes/profiles/<name>``; the
+    default profile (``~/.hermes``) returns "".
+    """
+    if home_path is not None:
+        resolved = Path(home_path)
+        if resolved.parent.name == "profiles" and resolved.name:
+            return resolved.name
+    for env_name in ("HERMES_PROFILE_NAME", "HERMES_PROFILE"):
+        value = os.environ.get(env_name, "").strip()
+        if value and value != "default":
+            return value
+    return ""
+
+
+# Only credential-shaped names get auto-aliased — a random profile-suffixed
+# var should not silently hydrate an unsuffixed name.
+_ALIAS_SUFFIXES = ("_API_KEY", "_TOKEN", "_SECRET", "_KEY", "_PASSWORD")
+
+
+def _profile_alias_target(var: str, profile: str) -> Optional[str]:
+    """Map ``FOO_<PROFILE>`` to ``FOO`` for the active profile when safe."""
+    if not profile:
+        return None
+    suffix = "_" + profile.replace("-", "_").upper()
+    if not var.endswith(suffix):
+        return None
+    alias = var[: -len(suffix)]
+    if not alias or not is_valid_env_name(alias):
+        return None
+    if not any(alias.endswith(s) for s in _ALIAS_SUFFIXES):
+        return None
+    return alias
+
+
 def apply_all(secrets_cfg: dict, home_path: Path,
               environ: Optional[Dict[str, str]] = None) -> ApplyReport:
     """Fetch from every enabled source and apply the merged result to env.
@@ -283,14 +321,24 @@ def apply_all(secrets_cfg: dict, home_path: Path,
 
     Precedence per env var (most-specific intent wins):
 
-    1. Pre-existing env (.env / shell) — unless the winning source has
+    1. ``secrets.preserve_existing`` names — a pre-existing env value always
+       wins for these, even against a source with ``override_existing: true``
+       (escape hatch for profile-local platform secrets, #58073).
+    2. Pre-existing env (.env / shell) — unless the winning source has
        ``override_existing: true``.
-    2. Mapped sources, in configured order.
-    3. Bulk sources, in configured order.
+    3. Mapped sources, in configured order.
+    4. Bulk sources, in configured order.
 
     First claim wins.  A later source that also carries the var gets a
     ``skipped_claimed`` entry and a conflict warning — never a silent
     clobber, and ``override_existing`` never applies across sources.
+
+    Profile aliasing (#51447): when running under a named profile, an applied
+    var ``FOO_<PROFILE>`` (credential-shaped suffixes only) also hydrates the
+    canonical ``FOO`` so platform adapters and plugins that read fixed env
+    names see the profile's value.  The alias obeys the same protected /
+    preserve / claimed / override guards and is disabled with
+    ``secrets.profile_alias: false``.
     """
     import os as _os
 
@@ -302,6 +350,14 @@ def apply_all(secrets_cfg: dict, home_path: Path,
     if not enabled:
         return report
 
+    preserve_raw = secrets_cfg.get("preserve_existing")
+    preserve: frozenset = frozenset(
+        n.strip() for n in preserve_raw if isinstance(n, str) and n.strip()
+    ) if isinstance(preserve_raw, list) else frozenset()
+
+    alias_enabled = bool(secrets_cfg.get("profile_alias", True))
+    profile = _active_profile_name(home_path) if alias_enabled else ""
+
     # Mapped sources outrank bulk sources regardless of list order:
     # an explicit VAR→ref binding is stronger intent than a project dump.
     ordered = ([s for s in enabled if s.shape == "mapped"]
@@ -321,6 +377,15 @@ def apply_all(secrets_cfg: dict, home_path: Path,
         except Exception:  # noqa: BLE001
             pass
 
+    # Every var any source supplies directly — an alias never shadows a
+    # var that some source will (or tried to) claim by its real name.
+    supplied_directly: set = set()
+    for _, _, result in fetches:
+        if result.ok:
+            supplied_directly.update(
+                v for v in result.secrets if isinstance(v, str)
+            )
+
     # Apply phase — sequential, first-wins, fully attributed.
     claimed: Dict[str, str] = {}  # var → source name that won it
     for source, cfg, result in fetches:
@@ -336,15 +401,14 @@ def apply_all(secrets_cfg: dict, home_path: Path,
         except Exception:  # noqa: BLE001
             override = False
 
-        for var, value in result.secrets.items():
-            if not isinstance(var, str) or not isinstance(value, str):
-                continue
+        def _try_apply(var: str, value: str, *, is_alias: bool = False) -> bool:
+            """Apply one var through the shared guard chain. True = applied."""
             if not is_valid_env_name(var):
                 sr.skipped_invalid.append(var)
-                continue
+                return False
             if var in protected:
                 sr.skipped_protected.append(var)
-                continue
+                return False
             if var in claimed:
                 sr.skipped_claimed.append(var)
                 report.conflicts.append(
@@ -352,11 +416,14 @@ def apply_all(secrets_cfg: dict, home_path: Path,
                     f"{source.name} also supplies it (first source wins — "
                     "remove one binding or reorder secrets.sources)"
                 )
-                continue
+                return False
             existed = bool(env.get(var))
+            if existed and var in preserve:
+                sr.skipped_existing.append(var)
+                return False
             if existed and not override:
                 sr.skipped_existing.append(var)
-                continue
+                return False
             env[var] = value
             claimed[var] = source.name
             sr.applied.append(var)
@@ -366,5 +433,21 @@ def apply_all(secrets_cfg: dict, home_path: Path,
                 shape=source.shape,
                 overrode_env=existed,
             )
+            return True
+
+        for var, value in result.secrets.items():
+            if not isinstance(var, str) or not isinstance(value, str):
+                continue
+            applied = _try_apply(var, value)
+
+            if not applied or not profile:
+                continue
+            alias = _profile_alias_target(var, profile)
+            if alias and alias not in supplied_directly and alias not in claimed:
+                if _try_apply(alias, value, is_alias=True):
+                    result.warnings.append(
+                        f"applied profile-scoped {var} as {alias} "
+                        f"(active profile {profile!r})"
+                    )
 
     return report

tests/secret_sources/test_profile_secrets.py

@@ -0,0 +1,183 @@
+"""Orchestrator-level profile secret handling.
+
+Covers the two halves of the profile-clobber bug cluster:
+
+- ``secrets.preserve_existing`` (#58073): named env vars keep their existing
+  value even against a source with ``override_existing: true``.
+- Profile aliasing (#51447): under a named profile, an applied
+  ``FOO_<PROFILE>`` var also hydrates the canonical ``FOO`` so adapters and
+  plugins that read fixed env names see the profile's value.
+
+Both are implemented ONCE in ``apply_all()`` so every backend — bundled or
+plugin — gets them for free.
+"""
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from agent.secret_sources import registry
+from agent.secret_sources.base import ErrorKind, FetchResult, SecretSource
+
+
+class _FakeBulk(SecretSource):
+    name = "fakebulk"
+    label = "Fake Bulk"
+    shape = "bulk"
+
+    def __init__(self, secrets):
+        self._secrets = secrets
+
+    def override_existing(self, cfg):
+        return bool(cfg.get("override_existing", True))
+
+    def fetch(self, cfg, home_path):
+        res = FetchResult()
+        res.secrets = dict(self._secrets)
+        return res
+
+
+@pytest.fixture(autouse=True)
+def _clean_registry():
+    registry._reset_registry_for_tests()
+    registry._BUILTINS_LOADED = True  # keep real builtins out
+    yield
+    registry._reset_registry_for_tests()
+
+
+def _apply(secrets, cfg_extra=None, home=Path("/tmp/x/.hermes"), env=None):
+    registry.register_source(_FakeBulk(secrets), replace=True)
+    cfg = {"fakebulk": {"enabled": True}}
+    cfg.update(cfg_extra or {})
+    env = env if env is not None else {}
+    report = registry.apply_all(cfg, home, environ=env)
+    return report, env
+
+
+PROFILE_HOME = Path("/home/u/.hermes/profiles/milla")
+
+
+# ---------------------------------------------------------------------------
+# preserve_existing
+# ---------------------------------------------------------------------------
+
+
+def test_preserve_existing_beats_override():
+    report, env = _apply(
+        {"FEISHU_APP_SECRET": "shared", "OPENAI_API_KEY": "fresh"},
+        cfg_extra={"preserve_existing": ["FEISHU_APP_SECRET"]},
+        env={"FEISHU_APP_SECRET": "profile-local", "OPENAI_API_KEY": "stale"},
+    )
+    assert env["FEISHU_APP_SECRET"] == "profile-local"   # preserved
+    assert env["OPENAI_API_KEY"] == "fresh"              # override still works
+    sr = report.sources[0]
+    assert "FEISHU_APP_SECRET" in sr.skipped_existing
+    assert "OPENAI_API_KEY" in sr.applied
+
+
+def test_preserve_existing_only_guards_set_vars():
+    """A preserve-listed var with NO existing value still gets applied."""
+    _, env = _apply(
+        {"FEISHU_APP_SECRET": "shared"},
+        cfg_extra={"preserve_existing": ["FEISHU_APP_SECRET"]},
+        env={},
+    )
+    assert env["FEISHU_APP_SECRET"] == "shared"
+
+
+def test_preserve_existing_junk_config_ignored():
+    for junk in ("notalist", 42, {"a": 1}, [1, 2], None):
+        _, env = _apply(
+            {"K": "v"}, cfg_extra={"preserve_existing": junk}, env={"K": "old"}
+        )
+        assert env["K"] == "v"  # falls back to normal override semantics
+
+
+# ---------------------------------------------------------------------------
+# profile aliasing
+# ---------------------------------------------------------------------------
+
+
+def test_profile_suffixed_var_hydrates_canonical():
+    report, env = _apply(
+        {"TELEGRAM_BOT_TOKEN_MILLA": "123:tok"},
+        home=PROFILE_HOME,
+    )
+    assert env["TELEGRAM_BOT_TOKEN_MILLA"] == "123:tok"
+    assert env["TELEGRAM_BOT_TOKEN"] == "123:tok"
+    assert "TELEGRAM_BOT_TOKEN" in report.provenance
+    assert any("applied profile-scoped" in w
+               for w in report.sources[0].result.warnings)
+
+
+def test_alias_requires_credential_suffix():
+    _, env = _apply({"RANDOM_SETTING_MILLA": "x"}, home=PROFILE_HOME)
+    assert "RANDOM_SETTING" not in env
+
+
+def test_alias_never_shadows_directly_supplied_var():
+    """If the project also carries the canonical name, the alias must not
+    fight it — direct supply wins."""
+    _, env = _apply(
+        {"TELEGRAM_BOT_TOKEN_MILLA": "profile-tok",
+         "TELEGRAM_BOT_TOKEN": "canonical-tok"},
+        home=PROFILE_HOME,
+    )
+    assert env["TELEGRAM_BOT_TOKEN"] == "canonical-tok"
+
+
+def test_alias_respects_existing_env_without_override():
+    class _NoOverride(_FakeBulk):
+        def override_existing(self, cfg):
+            return False
+
+    registry.register_source(
+        _NoOverride({"TELEGRAM_BOT_TOKEN_MILLA": "123:tok"}), replace=True
+    )
+    env = {"TELEGRAM_BOT_TOKEN": "existing"}
+    registry.apply_all({"fakebulk": {"enabled": True}}, PROFILE_HOME, environ=env)
+    assert env["TELEGRAM_BOT_TOKEN"] == "existing"
+
+
+def test_alias_disabled_by_config():
+    _, env = _apply(
+        {"TELEGRAM_BOT_TOKEN_MILLA": "123:tok"},
+        cfg_extra={"profile_alias": False},
+        home=PROFILE_HOME,
+    )
+    assert "TELEGRAM_BOT_TOKEN" not in env
+
+
+def test_default_profile_never_aliases():
+    _, env = _apply(
+        {"TELEGRAM_BOT_TOKEN_MILLA": "123:tok"},
+        home=Path("/home/u/.hermes"),
+    )
+    assert "TELEGRAM_BOT_TOKEN" not in env
+
+
+def test_hyphenated_profile_name_matches_underscore_suffix():
+    _, env = _apply(
+        {"SLACK_APP_TOKEN_MY_BOT": "xapp-1"},
+        home=Path("/home/u/.hermes/profiles/my-bot"),
+    )
+    assert env["SLACK_APP_TOKEN"] == "xapp-1"
+
+
+def test_alias_never_touches_protected_vars():
+    class _Protecting(_FakeBulk):
+        def protected_env_vars(self, cfg):
+            return frozenset({"BWS_ACCESS_TOKEN"})
+
+    registry.register_source(
+        _Protecting({"BWS_ACCESS_TOKEN_MILLA": "0.evil"}), replace=True
+    )
+    env = {"BWS_ACCESS_TOKEN": "0.real"}
+    registry.apply_all({"fakebulk": {"enabled": True}}, PROFILE_HOME, environ=env)
+    assert env["BWS_ACCESS_TOKEN"] == "0.real"
+
+
+def test_alias_provenance_recorded():
+    report, _ = _apply({"NOTION_TOKEN_MILLA": "sec"}, home=PROFILE_HOME)
+    assert report.provenance["NOTION_TOKEN"].source == "fakebulk"

website/docs/user-guide/secrets/index.md

@@ -27,6 +27,21 @@ secrets:
 
 Every credential injected by a source is labelled with its origin — setup flows and `hermes model` show `(from Bitwarden)` next to detected keys so you always know where a value came from.
 
+## Profiles and shared vaults
+
+Two orchestrator-level knobs make one shared vault safe across [profiles](../features/profiles):
+
+- **`secrets.preserve_existing`** — a list of env var names whose existing `.env` / shell value always wins, even against a source with `override_existing: true`. Use it for per-profile platform secrets (e.g. `FEISHU_APP_SECRET`) that intentionally differ across profiles while everything else rotates centrally:
+
+  ```yaml
+  secrets:
+    preserve_existing: [FEISHU_APP_SECRET, TELEGRAM_BOT_TOKEN]
+  ```
+
+- **Profile aliasing** (on by default, `secrets.profile_alias: false` to disable) — when Hermes runs under a named profile, a vault secret named `FOO_<PROFILE>` (credential-shaped suffixes only: `*_API_KEY`, `*_TOKEN`, `*_SECRET`, `*_KEY`, `*_PASSWORD`) also hydrates the canonical `FOO`. Store `TELEGRAM_BOT_TOKEN_MILLA` in the shared project and the `milla` profile's adapters — which read the fixed name `TELEGRAM_BOT_TOKEN` — get the right value automatically. A var the vault supplies directly under its canonical name always beats an alias.
+
+Both apply to every source — bundled and plugin — because they live in the orchestrator, not the backends.
+
 ## Adding your own backend
 
 Third-party secret managers ship as standalone plugins, not core PRs. A backend subclasses `agent.secret_sources.base.SecretSource` (one required method: `fetch(cfg, home_path) -> FetchResult`) and registers via `ctx.register_secret_source(MySource())` in the plugin's `register(ctx)`. The orchestrator owns precedence, conflict handling, timeouts, and provenance — your source only fetches. Full guide with the contract rules, subprocess-safety helper, and conformance kit: [Building a Secret Source Plugin](/developer-guide/secret-source-plugin).