refactor(doctor,gateway): share the migrate preflight's duplicate-credential check with doctor and status

Under multiplex-only the duplicated platform token is exactly what the migration
preflight refuses to fold on, so doctor and `gateway status` reuse that check
(`gateway_migrate.duplicate_credential_findings`) instead of a second scanner
in profile_channels. All three surfaces print one shared line naming the
profiles, the env key NAME (TELEGRAM_BOT_TOKEN, never a value or fingerprint),
why one token can serve only one gateway, and the remedy (own token / remove the
key from the non-owner / profile_routes) ending in `hermes gateway migrate
--multiplex`. Doctor folds the finding into its existing Profiles section.

Tests trimmed to two invariants: identical wording across doctor, status and
the preflight with the secret absent from output; distinct tokens and a shared
non-singleton key (model API key) produce no finding.
This commit is contained in:
teknium1
2026-09-21 15:28:45 -07:00
committed by Teknium
parent 54817f463b
commit ed69aa9e6e
6 changed files with 109 additions and 214 deletions

View File

@@ -49,7 +49,6 @@ from hermes_cli.doctor_tools import (
) )
from hermes_cli.doctor_state import ( from hermes_cli.doctor_state import (
_check_checkpoint_store, _check_checkpoint_store,
_check_cross_profile_gateway_credentials,
_check_directory_structure, _check_directory_structure,
_check_memory_provider, _check_memory_provider,
_check_profiles, _check_profiles,
@@ -120,7 +119,7 @@ DOCTOR_CHECKS = (
('External Tools', _check_git_and_rg), (None, _check_terminal_backend), (None, _check_node_and_browser), ('External Tools', _check_git_and_rg), (None, _check_terminal_backend), (None, _check_node_and_browser),
(None, _check_npm_audit), ('API Connectivity', _check_api_connectivity), (None, _check_npm_audit), ('API Connectivity', _check_api_connectivity),
('Tool Availability', _check_tool_availability), ('Skills Hub', _check_skills_hub), ('Tool Availability', _check_tool_availability), ('Skills Hub', _check_skills_hub),
('Memory Provider', _check_memory_provider), (None, _check_profiles), (None, _check_cross_profile_gateway_credentials), ('Memory Provider', _check_memory_provider), (None, _check_profiles),
) )

View File

@@ -572,27 +572,9 @@ def _check_profiles(should_fix: bool, f: Finding) -> None:
_m = _re.search(r"hermes -p (\S+)", wrapper.read_text(encoding="utf-8")) _m = _re.search(r"hermes -p (\S+)", wrapper.read_text(encoding="utf-8"))
if _m and not profile_exists(_m.group(1)): if _m and not profile_exists(_m.group(1)):
check_warn(f"Orphan alias: {wrapper.name} → profile '{_m.group(1)}' no longer exists") check_warn(f"Orphan alias: {wrapper.name} → profile '{_m.group(1)}' no longer exists")
# Same helper as the multiplex migration preflight, so doctor names the duplicates that make
# `hermes gateway migrate --multiplex` refuse (and made pre-multiplex standalone gateways race).
@doctor_check("") # Diagnostics must still finish if one profile's files are unreadable. from hermes_cli.gateway_migrate import duplicate_credential_findings
def _check_cross_profile_gateway_credentials(should_fix: bool, f: Finding) -> None: for line in duplicate_credential_findings():
"""Report bot credentials that would make local profile gateways fight each other.""" check_warn("Duplicate platform credential across profiles", f"({line})")
from hermes_cli.profile_channels import scan_local_profile_credential_collisions f.manual_issues.append(line)
report = scan_local_profile_credential_collisions()
if not report.collisions and not report.unreadable_paths:
return
_section("Gateway Profile Credentials")
if report.collisions:
check_warn("Duplicate platform credentials across local profile homes", f"({report.format_for_display()})")
f.manual_issues.append(
"Duplicate gateway platform credentials across local profiles — give each profile its own bot "
"credential or stop/remove the other profile gateway before starting it."
)
if report.unreadable_paths:
paths = ", ".join(str(path) for path in report.unreadable_paths)
check_warn("Could not inspect gateway credentials in local profile homes", f"({paths})")
f.manual_issues.append(
"Could not inspect one or more local profile homes for duplicate gateway credentials; "
"check their path permissions and configuration files."
)

View File

@@ -1600,23 +1600,16 @@ def _print_other_profiles_gateway_status() -> None:
pass pass
def _print_cross_profile_credential_warnings() -> None: def _print_duplicate_credential_warnings() -> None:
"""Surface bot-token collisions even when profile gateways run standalone.""" """The migrate preflight's duplicate-credential findings, so ``gateway status`` explains a parked
try: or racing bot (and why the fleet will not fold) with the same words as ``migrate --dry-run``."""
from hermes_cli.profile_channels import scan_local_profile_credential_collisions with contextlib.suppress(Exception):
from hermes_cli.gateway_migrate import duplicate_credential_findings
report = scan_local_profile_credential_collisions() lines = duplicate_credential_findings()
except Exception: if lines:
return print()
if not report.collisions and not report.unreadable_paths: for line in lines:
return print(f"⚠ {line}")
print()
if report.collisions:
print("⚠ Duplicate platform credentials across local profile homes")
print(f" {report.format_for_display()}")
print(" Give each profile its own bot credential, or stop/remove the other profile gateway.")
if report.unreadable_paths:
print("⚠ Could not inspect gateway credentials in: " + ", ".join(map(str, report.unreadable_paths)))
def _gateway_list() -> None: def _gateway_list() -> None:
@@ -5040,7 +5033,7 @@ def _cmd_status(args):
print(" hermes gateway run # Run in foreground") print(" hermes gateway run # Run in foreground")
_print_lines(*_STATUS_STOPPED_HINTS[_status_host_kind()]) _print_lines(*_STATUS_STOPPED_HINTS[_status_host_kind()])
_print_cross_profile_credential_warnings() _print_duplicate_credential_warnings()
_print_other_profiles_gateway_status() _print_other_profiles_gateway_status()

View File

@@ -396,24 +396,54 @@ def _credential_claims(config) -> dict[tuple, str]:
return claims return claims
def _credential_key_names(platform_value: str) -> str:
"""Env key NAMES (never values) that make ``platform_value`` connect as a bot, e.g.
``TELEGRAM_BOT_TOKEN``; the platform id when no key is registered (config.yaml-only token)."""
from hermes_cli.profile_channels import credential_env_keys
names = sorted(key for key, pid in credential_env_keys().items() if pid == platform_value)
return "/".join(names) or f"the {platform_value} token"
def duplicate_credential_lines(configs: list[tuple[str, object]]) -> list[str]:
"""One finding per platform credential two profiles both hold, with the remedy. The SINGLE
source for the migrate preflight, ``hermes doctor`` and ``hermes gateway status``, so all three
name the same duplicates the same way: profile names + key names only, never a value or hash."""
owners: dict[tuple, str] = {}
lines: list[str] = []
for name, cfg in configs: # default first: it wins the claim, like at multiplexer startup
for claim, platform_value in _credential_claims(cfg).items():
owner = owners.setdefault(claim, name)
if owner == name:
continue
key = _credential_key_names(platform_value)
lines.append(
f"Profiles '{owner}' and '{name}' both hold the same {platform_value} credential ({key}): "
f"one platform token can serve only one gateway, so the bot answers from whichever profile "
f"claims it first and the other's adapter is parked. Give '{name}' its own bot token, or remove "
f"{key} from the profile that should not own it (or keep it in {owner} and route {name}'s "
f"chats with profile_routes — gateway.profile_routes in {owner}'s config.yaml), then run "
f"{MIGRATE_COMMAND}."
)
return lines
def duplicate_credential_findings() -> list[str]:
"""The preflight's duplicate-credential check read straight from the local profile homes, for
diagnostics that have no migration plan (doctor, gateway status). A profile whose gateway config
does not load is skipped here — ``build_migration_plan`` reports that one as its own blocker."""
configs: list[tuple[str, object]] = []
with _multiplex_read_mode():
for name, home in _profile_homes():
with contextlib.suppress(Exception):
configs.append((name, _profile_gateway_config(home)))
return duplicate_credential_lines(configs)
def _check_duplicate_credentials(plan: MigrationPlan, configs: dict[str, object]) -> None: def _check_duplicate_credentials(plan: MigrationPlan, configs: dict[str, object]) -> None:
"""BLOCKER: the same bot credential configured on two profiles — the multiplexer would park """BLOCKER: the same bot credential configured on two profiles — the multiplexer would park
the duplicate adapter, so one profile's bot would go silent after migration.""" the duplicate adapter, so one profile's bot would go silent after migration."""
owners: dict[tuple, str] = {} plan.blockers.extend(duplicate_credential_lines(
for profile in plan.profiles: # default first: it wins the claim, like at multiplexer startup [(p.name, configs[p.name]) for p in plan.profiles if p.name in configs]))
cfg = configs.get(profile.name)
if cfg is None:
continue
for claim, platform_value in _credential_claims(cfg).items():
owner = owners.setdefault(claim, profile.name)
if owner == profile.name:
continue
plan.blockers.append(
f"Profiles '{owner}' and '{profile.name}' both configure {platform_value} with the same "
f"credential: the bot can only belong to one profile; remove the token from "
f"'{profile.name}' or keep it in {owner} and route {profile.name}'s chats with "
f"profile_routes (gateway.profile_routes in {owner}'s config.yaml)."
)
def platform_serves_profile_prefix(platform_value: str) -> bool: def platform_serves_profile_prefix(platform_value: str) -> bool:

View File

@@ -22,7 +22,6 @@ from __future__ import annotations
import contextlib import contextlib
import logging import logging
import re import re
from dataclasses import dataclass
from functools import partial from functools import partial
from pathlib import Path from pathlib import Path
from typing import Dict, Iterable, List, Optional, Set, Tuple from typing import Dict, Iterable, List, Optional, Set, Tuple
@@ -483,87 +482,6 @@ def shared_credential_warning(profile: str, platforms: List[str], source: str =
) )
@dataclass(frozen=True)
class ProfileCredentialCollision:
"""One platform credential claimed by more than one local profile home.
``paths`` intentionally names homes, not the credential or its fingerprint: doctor and
gateway status must make the operator's next action clear without leaking a secret.
"""
platform: str
paths: tuple[Path, ...]
@dataclass(frozen=True)
class LocalProfileCredentialCollisionReport:
"""Best-effort credential scan used by non-mutating diagnostic commands."""
collisions: tuple[ProfileCredentialCollision, ...] = ()
unreadable_paths: tuple[Path, ...] = ()
def format_for_display(self) -> str:
"""Render only platform names and paths; never expose credentials or fingerprints."""
return "; ".join(
f"{collision.platform}: {', '.join(str(path) for path in collision.paths)}"
for collision in self.collisions
)
def _local_profile_homes() -> tuple[tuple[str, Path], ...]:
"""Distinct default/named homes rooted at the local Hermes installation."""
from hermes_cli.profiles import profiles_to_serve
homes: list[tuple[str, Path]] = []
seen: set[Path] = set()
for name, home in profiles_to_serve(multiplex=True):
try:
key = home.resolve(strict=False)
except OSError:
key = home
if key not in seen:
seen.add(key)
homes.append((name, home))
return tuple(homes)
def _profile_credential_claims(home: Path) -> set[tuple[str, str]]:
"""Read claims through the migration preflight's gateway fingerprint implementation."""
from hermes_cli import gateway_migrate
with gateway_migrate._multiplex_read_mode():
return set(gateway_migrate._credential_claims(gateway_migrate._profile_gateway_config(home)))
def scan_local_profile_credential_collisions() -> LocalProfileCredentialCollisionReport:
"""Find duplicate enabled platform credentials across local profile homes.
The gateway migration preflight is the canonical source for both enabled-platform
resolution and credential fingerprints. This scan deliberately uses the same path so a
diagnostic warning agrees with the gateway's collision behaviour. A malformed or unreadable
profile is reported separately while healthy homes continue to be scanned.
"""
owners: dict[tuple[str, str], list[Path]] = {}
unreadable: list[Path] = []
for _name, home in _local_profile_homes():
try:
claims = _profile_credential_claims(home)
except (OSError, ValueError, RuntimeError):
unreadable.append(home)
continue
for claim in claims:
owners.setdefault(claim, []).append(home)
collisions = tuple(
ProfileCredentialCollision(platform=platform, paths=tuple(paths))
for (platform, _fingerprint), paths in sorted(owners.items())
if len(paths) > 1
)
return LocalProfileCredentialCollisionReport(
collisions=collisions,
unreadable_paths=tuple(unreadable),
)
def format_stripped_notice(profile: str, platforms: List[str], clone_flag: str = "--clone") -> List[str]: def format_stripped_notice(profile: str, platforms: List[str], clone_flag: str = "--clone") -> List[str]:
"""Lines printed after a channel-less clone so the user knows what was left behind and how to """Lines printed after a channel-less clone so the user knows what was left behind and how to
configure the new profile's own bots.""" configure the new profile's own bots."""

View File

@@ -1,4 +1,7 @@
"""Regression coverage for cross-profile gateway credential diagnostics (#118388).""" """Regression for #118388: a platform singleton secret (``TELEGRAM_BOT_TOKEN``) duplicated across
local profile homes was invisible to ``hermes doctor`` / ``hermes gateway status``; the only signal
was the losing standalone gateway's log. Doctor, status and the migrate preflight now share one
duplicate-credential helper, so all three name the same profiles + key names (never the value)."""
from __future__ import annotations from __future__ import annotations
@@ -6,89 +9,59 @@ import io
from contextlib import redirect_stdout from contextlib import redirect_stdout
from pathlib import Path from pathlib import Path
import pytest
import hermes_constants import hermes_constants
from hermes_cli import doctor_state, profile_channels from hermes_cli import doctor_state, gateway, gateway_migrate as gm
SECRET = "123456:shared-secret-value"
def _profile_home(root: Path, name: str, token: str) -> Path: @pytest.fixture
home = root if name == "default" else root / "profiles" / name def homes(tmp_path, monkeypatch):
home.mkdir(parents=True, exist_ok=True)
(home / ".env").write_text(f"TELEGRAM_BOT_TOKEN={token}\n", encoding="utf-8")
return home
def test_scan_reports_duplicate_platform_credential_with_paths_not_secret(monkeypatch, tmp_path):
root = tmp_path / ".hermes" root = tmp_path / ".hermes"
default = _profile_home(root, "default", "123:shared-secret") (root / "profiles" / "worker").mkdir(parents=True)
worker = _profile_home(root, "worker", "123:shared-secret") monkeypatch.setattr(Path, "home", lambda: tmp_path)
monkeypatch.setenv("HERMES_HOME", str(root)) monkeypatch.setenv("HERMES_HOME", str(root))
for name in ("TELEGRAM_BOT_TOKEN", "DISCORD_BOT_TOKEN", "OPENAI_API_KEY"):
monkeypatch.delenv(name, raising=False)
monkeypatch.setattr(hermes_constants, "_default_hermes_root_memo", None) monkeypatch.setattr(hermes_constants, "_default_hermes_root_memo", None)
monkeypatch.setattr(gm, "_live_gateway_pid", lambda home: None)
report = profile_channels.scan_local_profile_credential_collisions() monkeypatch.setattr(gm, "_installed_services", lambda home: [])
return root, root / "profiles" / "worker"
assert [(collision.platform, collision.paths) for collision in report.collisions] == [
("telegram", (default, worker)),
]
assert "shared-secret" not in report.format_for_display()
assert str(default) in report.format_for_display()
assert str(worker) in report.format_for_display()
def test_scan_ignores_distinct_platform_credentials(monkeypatch, tmp_path): def _surfaces() -> str:
root = tmp_path / ".hermes" out = io.StringIO()
_profile_home(root, "default", "123:default-secret") with redirect_stdout(out):
_profile_home(root, "worker", "123:worker-secret") doctor_state._check_profiles(False)
monkeypatch.setenv("HERMES_HOME", str(root)) gateway._print_duplicate_credential_warnings()
monkeypatch.setattr(hermes_constants, "_default_hermes_root_memo", None) return out.getvalue()
report = profile_channels.scan_local_profile_credential_collisions()
assert report.collisions == ()
assert report.unreadable_paths == ()
def test_scan_reports_unreadable_profile_without_hiding_other_collisions(monkeypatch, tmp_path): def test_duplicate_singleton_secret_named_identically_by_doctor_status_and_preflight(homes):
default = tmp_path / ".hermes" default, worker = homes
worker = default / "profiles" / "worker" (default / ".env").write_text(f"TELEGRAM_BOT_TOKEN={SECRET}\nOPENAI_API_KEY=sk-shared\n", encoding="utf-8")
broken = default / "profiles" / "broken" (worker / ".env").write_text(f"TELEGRAM_BOT_TOKEN={SECRET}\nOPENAI_API_KEY=sk-shared\n", encoding="utf-8")
monkeypatch.setattr(
profile_channels,
"_local_profile_homes",
lambda: (("default", default), ("worker", worker), ("broken", broken)),
)
monkeypatch.setattr(
profile_channels,
"_profile_credential_claims",
lambda home: (
{("telegram", "fingerprint")} if home != broken else (_ for _ in ()).throw(OSError("denied"))
),
)
report = profile_channels.scan_local_profile_credential_collisions() findings = gm.duplicate_credential_findings()
assert len(findings) == 1
assert report.collisions[0].platform == "telegram" line = findings[0]
assert report.unreadable_paths == (broken,) assert "'default'" in line and "'worker'" in line and "TELEGRAM_BOT_TOKEN" in line
assert "only one gateway" in line and gm.MIGRATE_COMMAND in line
# Same helper, same words: the preflight blocker IS the doctor/status finding.
assert gm.build_migration_plan().blockers == findings
out = _surfaces()
assert out.count(line) == 2 # once from doctor, once from gateway status
# Absence: the value (or a hash a user could mistake for it) never reaches any surface.
assert SECRET not in out and "shared-secret" not in out and "sk-shared" not in out
def test_doctor_and_gateway_status_surface_collision_without_secret(monkeypatch, tmp_path): def test_distinct_tokens_and_shared_non_singleton_keys_are_not_findings(homes):
default = tmp_path / ".hermes" default, worker = homes
worker = default / "profiles" / "worker" (default / ".env").write_text(f"TELEGRAM_BOT_TOKEN={SECRET}\nOPENAI_API_KEY=sk-shared\n", encoding="utf-8")
report = profile_channels.LocalProfileCredentialCollisionReport( (worker / ".env").write_text("TELEGRAM_BOT_TOKEN=999999:other-token\nOPENAI_API_KEY=sk-shared\n", encoding="utf-8")
collisions=(profile_channels.ProfileCredentialCollision("telegram", (default, worker)),),
)
monkeypatch.setattr(profile_channels, "scan_local_profile_credential_collisions", lambda: report)
doctor_output = io.StringIO() assert gm.duplicate_credential_findings() == []
with redirect_stdout(doctor_output): assert not gm.build_migration_plan().blocked
findings = doctor_state._check_cross_profile_gateway_credentials(False) assert "both hold" not in _surfaces()
assert "Duplicate platform credentials" in doctor_output.getvalue()
assert str(worker) in doctor_output.getvalue()
assert findings.manual_issues
from hermes_cli import gateway
status_output = io.StringIO()
with redirect_stdout(status_output):
gateway._print_cross_profile_credential_warnings()
assert "Duplicate platform credentials" in status_output.getvalue()
assert "shared-secret" not in status_output.getvalue()