mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-31 19:16:29 +00:00
feat(managed-scope): add managed_scope module (resolver, loaders, key helpers)
New hermes_cli/managed_scope.py resolves a system-level managed directory (HERMES_MANAGED_DIR override > /etc/hermes), parses managed config.yaml/.env with fail-open semantics, and exposes is_key_managed/is_env_managed helpers. The system default is ignored under pytest and HERMES_MANAGED_DIR is added to the conftest env scrub so a real managed scope can't leak into the suite. Not wired into the load paths yet (Phases 2-3).
This commit is contained in:
parent
bf9a0481fa
commit
9cbcc0c9c8
3 changed files with 317 additions and 0 deletions
171
hermes_cli/managed_scope.py
Normal file
171
hermes_cli/managed_scope.py
Normal file
|
|
@ -0,0 +1,171 @@
|
|||
"""Managed scope — IT-pushed, user-immutable config & env layer.
|
||||
|
||||
A system-level directory (default ``/etc/hermes``, root-owned and not
|
||||
user-writable) supplies ``config.yaml`` and ``.env`` values that WIN over the
|
||||
user's ``~/.hermes/config.yaml`` and ``~/.hermes/.env`` on a per-leaf-key basis.
|
||||
|
||||
This is DISTINCT from ``hermes_cli.config.is_managed()`` / ``HERMES_MANAGED``,
|
||||
which is a coarse package-manager write-lock (declarative-distro / formula
|
||||
installs). That lock blocks all mutation; this layer injects specific immutable
|
||||
values. The two are independent and may coexist.
|
||||
|
||||
v1 enforcement is filesystem permissions only — see
|
||||
``docs/design/managed-scope.md`` §7. v1 is Linux/POSIX-first; ``get_managed_dir()``
|
||||
is the single seam for adding macOS / Windows native locations later.
|
||||
|
||||
Attribution: do not reference any third-party product by name in this file.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import logging
|
||||
import os
|
||||
import threading
|
||||
from pathlib import Path
|
||||
from typing import Dict, Optional
|
||||
|
||||
import yaml
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# POSIX default. Other-platform locations are a deliberate v2 item; when added,
|
||||
# they belong ONLY inside get_managed_dir().
|
||||
_DEFAULT_MANAGED_DIR = Path("/etc/hermes")
|
||||
|
||||
_CACHE_LOCK = threading.Lock()
|
||||
# path_key -> (mtime_ns, size, parsed)
|
||||
_CONFIG_CACHE: Dict[str, tuple] = {}
|
||||
_ENV_CACHE: Dict[str, tuple] = {}
|
||||
|
||||
|
||||
def _under_pytest() -> bool:
|
||||
"""True when running inside the test suite.
|
||||
|
||||
Used to ignore the system default ``/etc/hermes`` during tests so a real
|
||||
managed scope on a developer/CI box can't leak policy into the suite. Tests
|
||||
that exercise managed scope set ``HERMES_MANAGED_DIR`` explicitly, which is
|
||||
still honored (the override path below runs before this guard takes effect).
|
||||
"""
|
||||
return "PYTEST_CURRENT_TEST" in os.environ
|
||||
|
||||
|
||||
def get_managed_dir() -> Optional[Path]:
|
||||
"""Resolve the managed-scope directory, or None when no scope is present.
|
||||
|
||||
Resolution (highest priority first):
|
||||
1. ``$HERMES_MANAGED_DIR`` — deployment/bootstrap path override (IT-only;
|
||||
never persisted to any .env). Honored only when set to a non-empty value
|
||||
AND the directory exists.
|
||||
2. ``/etc/hermes`` — POSIX default, when it exists. Ignored under pytest so
|
||||
a real system managed scope can't leak into the test suite.
|
||||
|
||||
A non-existent directory at either tier resolves to None (no managed scope),
|
||||
which is the common case and must be cheap + side-effect-free.
|
||||
"""
|
||||
override = os.environ.get("HERMES_MANAGED_DIR", "").strip()
|
||||
if override:
|
||||
p = Path(override)
|
||||
return p if p.is_dir() else None
|
||||
if _under_pytest():
|
||||
return None
|
||||
return _DEFAULT_MANAGED_DIR if _DEFAULT_MANAGED_DIR.is_dir() else None
|
||||
|
||||
|
||||
def invalidate_managed_cache() -> None:
|
||||
"""Drop cached managed config/env. For tests and post-edit reloads."""
|
||||
with _CACHE_LOCK:
|
||||
_CONFIG_CACHE.clear()
|
||||
_ENV_CACHE.clear()
|
||||
|
||||
|
||||
def _cached_read(path: Path, cache: Dict[str, tuple], parse):
|
||||
"""Shared (mtime_ns, size)-keyed read. Returns a deepcopy of the parsed value.
|
||||
|
||||
Returns ``None`` when the file is absent or fails to parse (fail-open). A
|
||||
parse failure is logged LOUDLY — the admin needs to know their policy isn't
|
||||
being applied — but never raises, so a malformed managed file can't brick
|
||||
startup.
|
||||
"""
|
||||
try:
|
||||
st = path.stat()
|
||||
except OSError:
|
||||
return None # absent
|
||||
key = (st.st_mtime_ns, st.st_size)
|
||||
path_key = str(path)
|
||||
with _CACHE_LOCK:
|
||||
hit = cache.get(path_key)
|
||||
if hit is not None and hit[:2] == key:
|
||||
return copy.deepcopy(hit[2])
|
||||
try:
|
||||
with open(path, encoding="utf-8") as f:
|
||||
parsed = parse(f)
|
||||
except Exception as exc: # noqa: BLE001 — fail-open, but LOUD
|
||||
logger.warning(
|
||||
"managed scope: failed to parse %s: %s — IGNORING this managed file. "
|
||||
"Admin policy from this file is NOT being applied. Fix and restart.",
|
||||
path,
|
||||
exc,
|
||||
)
|
||||
return None
|
||||
with _CACHE_LOCK:
|
||||
cache[path_key] = (key[0], key[1], copy.deepcopy(parsed))
|
||||
return parsed
|
||||
|
||||
|
||||
def load_managed_config() -> dict:
|
||||
"""Parsed managed config.yaml, or {} when absent/malformed (fail-open)."""
|
||||
managed_dir = get_managed_dir()
|
||||
if managed_dir is None:
|
||||
return {}
|
||||
parsed = _cached_read(
|
||||
managed_dir / "config.yaml",
|
||||
_CONFIG_CACHE,
|
||||
lambda f: yaml.safe_load(f) or {},
|
||||
)
|
||||
return parsed if isinstance(parsed, dict) else {}
|
||||
|
||||
|
||||
def load_managed_env() -> Dict[str, str]:
|
||||
"""Parsed managed .env (KEY=VALUE), or {} when absent (fail-open)."""
|
||||
managed_dir = get_managed_dir()
|
||||
if managed_dir is None:
|
||||
return {}
|
||||
parsed = _cached_read(managed_dir / ".env", _ENV_CACHE, _parse_env)
|
||||
return parsed if isinstance(parsed, dict) else {}
|
||||
|
||||
|
||||
def _parse_env(f) -> Dict[str, str]:
|
||||
out: Dict[str, str] = {}
|
||||
for line in f:
|
||||
line = line.strip()
|
||||
if not line or line.startswith("#") or "=" not in line:
|
||||
continue
|
||||
key, _, value = line.partition("=")
|
||||
out[key.strip()] = value.strip().strip("\"'")
|
||||
return out
|
||||
|
||||
|
||||
def _flatten_keys(d: dict, prefix: str = "") -> set:
|
||||
keys: set = set()
|
||||
for k, v in d.items():
|
||||
dotted = f"{prefix}.{k}" if prefix else str(k)
|
||||
if isinstance(v, dict) and v:
|
||||
keys |= _flatten_keys(v, dotted)
|
||||
else:
|
||||
keys.add(dotted)
|
||||
return keys
|
||||
|
||||
|
||||
def managed_config_keys() -> set:
|
||||
"""Dotted leaf keys pinned by the managed config (e.g. {'model.default'})."""
|
||||
return _flatten_keys(load_managed_config())
|
||||
|
||||
|
||||
def is_key_managed(dotted_key: str) -> bool:
|
||||
"""True if the exact dotted config key is pinned by the managed layer."""
|
||||
return dotted_key in managed_config_keys()
|
||||
|
||||
|
||||
def is_env_managed(name: str) -> bool:
|
||||
"""True if the env var name is pinned by the managed .env layer."""
|
||||
return name in load_managed_env()
|
||||
|
|
@ -190,6 +190,7 @@ _HERMES_BEHAVIORAL_VARS = frozenset({
|
|||
"HERMES_INFERENCE_PROVIDER",
|
||||
"HERMES_TUI_PROVIDER",
|
||||
"HERMES_MANAGED",
|
||||
"HERMES_MANAGED_DIR",
|
||||
"HERMES_DEV",
|
||||
"HERMES_CONTAINER",
|
||||
"HERMES_EPHEMERAL_SYSTEM_PROMPT",
|
||||
|
|
|
|||
145
tests/hermes_cli/test_managed_scope.py
Normal file
145
tests/hermes_cli/test_managed_scope.py
Normal file
|
|
@ -0,0 +1,145 @@
|
|||
"""Unit tests for hermes_cli.managed_scope (resolver + loaders + key helpers)."""
|
||||
import textwrap
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
# ── Directory resolver ───────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_get_managed_dir_env_override(tmp_path, monkeypatch):
|
||||
from hermes_cli import managed_scope
|
||||
|
||||
managed = tmp_path / "managed"
|
||||
managed.mkdir()
|
||||
monkeypatch.setenv("HERMES_MANAGED_DIR", str(managed))
|
||||
assert managed_scope.get_managed_dir() == managed
|
||||
|
||||
|
||||
def test_get_managed_dir_absent_override_returns_none(tmp_path, monkeypatch):
|
||||
from hermes_cli import managed_scope
|
||||
|
||||
monkeypatch.setenv("HERMES_MANAGED_DIR", str(tmp_path / "nope"))
|
||||
# Override points at a non-existent dir → no managed scope.
|
||||
assert managed_scope.get_managed_dir() is None
|
||||
|
||||
|
||||
def test_get_managed_dir_empty_override_falls_through(tmp_path, monkeypatch):
|
||||
from hermes_cli import managed_scope
|
||||
|
||||
monkeypatch.setenv("HERMES_MANAGED_DIR", " ") # whitespace = unset
|
||||
# Under pytest the /etc/hermes default is ignored, so this is None; the
|
||||
# assertion that matters is that it does NOT raise.
|
||||
result = managed_scope.get_managed_dir()
|
||||
assert result is None or result.exists()
|
||||
|
||||
|
||||
def test_get_managed_dir_default_ignored_under_pytest(monkeypatch):
|
||||
"""The system default must be inert in the test suite (isolation guard)."""
|
||||
from hermes_cli import managed_scope
|
||||
|
||||
monkeypatch.delenv("HERMES_MANAGED_DIR", raising=False)
|
||||
assert managed_scope.get_managed_dir() is None
|
||||
|
||||
|
||||
# ── Loaders + key helpers ────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _write_managed(tmp_path, monkeypatch, *, config=None, env=None):
|
||||
from hermes_cli import managed_scope
|
||||
|
||||
managed = tmp_path / "managed"
|
||||
managed.mkdir(exist_ok=True)
|
||||
if config is not None:
|
||||
(managed / "config.yaml").write_text(textwrap.dedent(config), encoding="utf-8")
|
||||
if env is not None:
|
||||
(managed / ".env").write_text(textwrap.dedent(env), encoding="utf-8")
|
||||
monkeypatch.setenv("HERMES_MANAGED_DIR", str(managed))
|
||||
managed_scope.invalidate_managed_cache()
|
||||
return managed
|
||||
|
||||
|
||||
def test_load_managed_config(tmp_path, monkeypatch):
|
||||
from hermes_cli import managed_scope
|
||||
|
||||
_write_managed(
|
||||
tmp_path,
|
||||
monkeypatch,
|
||||
config="""
|
||||
model:
|
||||
default: managed/model
|
||||
""",
|
||||
)
|
||||
assert managed_scope.load_managed_config() == {"model": {"default": "managed/model"}}
|
||||
|
||||
|
||||
def test_load_managed_config_absent_is_empty(tmp_path, monkeypatch):
|
||||
from hermes_cli import managed_scope
|
||||
|
||||
monkeypatch.setenv("HERMES_MANAGED_DIR", str(tmp_path / "nope"))
|
||||
managed_scope.invalidate_managed_cache()
|
||||
assert managed_scope.load_managed_config() == {}
|
||||
|
||||
|
||||
def test_load_managed_config_malformed_fails_open(tmp_path, monkeypatch):
|
||||
from hermes_cli import managed_scope
|
||||
|
||||
_write_managed(tmp_path, monkeypatch, config="model: : : not yaml :")
|
||||
assert managed_scope.load_managed_config() == {} # fail-open, no raise
|
||||
|
||||
|
||||
def test_managed_config_keys_are_dotted_leaves(tmp_path, monkeypatch):
|
||||
from hermes_cli import managed_scope
|
||||
|
||||
_write_managed(
|
||||
tmp_path,
|
||||
monkeypatch,
|
||||
config="""
|
||||
model:
|
||||
default: m
|
||||
security:
|
||||
redact_secrets: true
|
||||
""",
|
||||
)
|
||||
assert managed_scope.managed_config_keys() == {
|
||||
"model.default",
|
||||
"security.redact_secrets",
|
||||
}
|
||||
|
||||
|
||||
def test_is_key_managed(tmp_path, monkeypatch):
|
||||
from hermes_cli import managed_scope
|
||||
|
||||
_write_managed(tmp_path, monkeypatch, config="model:\n default: m\n")
|
||||
assert managed_scope.is_key_managed("model.default") is True
|
||||
assert managed_scope.is_key_managed("model.fallback") is False
|
||||
|
||||
|
||||
def test_load_managed_env_and_is_env_managed(tmp_path, monkeypatch):
|
||||
from hermes_cli import managed_scope
|
||||
|
||||
_write_managed(
|
||||
tmp_path, monkeypatch, env="OPENAI_API_BASE=https://org.example/v1\n"
|
||||
)
|
||||
assert managed_scope.load_managed_env() == {
|
||||
"OPENAI_API_BASE": "https://org.example/v1"
|
||||
}
|
||||
assert managed_scope.is_env_managed("OPENAI_API_BASE") is True
|
||||
assert managed_scope.is_env_managed("OTHER") is False
|
||||
|
||||
|
||||
def test_editing_managed_config_invalidates_cache(tmp_path, monkeypatch):
|
||||
from hermes_cli import managed_scope
|
||||
|
||||
managed = _write_managed(tmp_path, monkeypatch, config="model:\n default: v1\n")
|
||||
assert managed_scope.load_managed_config()["model"]["default"] == "v1"
|
||||
(managed / "config.yaml").write_text("model:\n default: v2\n", encoding="utf-8")
|
||||
managed_scope.invalidate_managed_cache()
|
||||
assert managed_scope.load_managed_config()["model"]["default"] == "v2"
|
||||
|
||||
|
||||
def test_managed_dir_env_scrubbed_by_default():
|
||||
"""conftest must scrub HERMES_MANAGED_DIR so a dev-shell value can't leak in."""
|
||||
import os
|
||||
|
||||
assert "HERMES_MANAGED_DIR" not in os.environ
|
||||
Loading…
Add table
Add a link
Reference in a new issue