seb/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(auth): Disable Nous legacy session key fallback

Browse Source
This commit is contained in:
Robin Fernandes
2026-05-29 02:24:48 -07:00
committed by kshitij
parent a87f0a82a5
commit 41ff6e5937
17 changed files with 484 additions and 860 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/hermes_cli/test_auth_commands.py
+6 -6
View File
@@ -107,15 +107,15 @@ def test_auth_add_nous_oauth_persists_pool_entry(tmp_path, monkeypatch):
"portal_base_url": "https://portal.example.com",
"inference_base_url": "https://inference.example.com/v1",
"client_id": "hermes-cli",
"scope": "inference:invoke inference:mint_agent_key",
"scope": "inference:invoke",
"token_type": "Bearer",
"access_token": token,
"refresh_token": "refresh-token",
"obtained_at": "2026-03-23T10:00:00+00:00",
"expires_at": "2026-03-23T11:00:00+00:00",
"expires_in": 3600,
"agent_key": "ak-test",
"agent_key_id": "ak-id",
"agent_key": token,
"agent_key_id": None,
"agent_key_expires_at": "2026-03-23T10:30:00+00:00",
"agent_key_expires_in": 1800,
"agent_key_reused": False,
@@ -228,15 +228,15 @@ def test_auth_add_nous_oauth_honors_custom_label(tmp_path, monkeypatch):
"portal_base_url": "https://portal.example.com",
"inference_base_url": "https://inference.example.com/v1",
"client_id": "hermes-cli",
"scope": "inference:invoke inference:mint_agent_key",
"scope": "inference:invoke",
"token_type": "Bearer",
"access_token": token,
"refresh_token": "refresh-token",
"obtained_at": "2026-03-23T10:00:00+00:00",
"expires_at": "2026-03-23T11:00:00+00:00",
"expires_in": 3600,
"agent_key": "ak-test",
"agent_key_id": "ak-id",
"agent_key": token,
"agent_key_id": None,
"agent_key_expires_at": "2026-03-23T10:30:00+00:00",
"agent_key_expires_in": 1800,
"agent_key_reused": False,
tests/hermes_cli/test_auth_nous_provider.py
+230 -261
View File
@@ -1,4 +1,4 @@
"""Regression tests for Nous OAuth refresh + agent-key mint interactions."""
"""Regression tests for Nous OAuth refresh and inference JWT interactions."""
import base64
import json
@@ -126,14 +126,15 @@ class TestResolveVerifyFallback:
def _setup_nous_auth(
hermes_home: Path,
*,
access_token: str = "access-old",
access_token: str = "",
refresh_token: str = "refresh-old",
scope: str = "inference:mint_agent_key",
scope: str = "inference:invoke",
expires_at: str = "2026-02-01T00:00:00+00:00",
expires_in: int = 0,
agent_key: str | None = None,
agent_key_expires_at: str | None = None,
) -> None:
access_token = access_token or _invoke_jwt(seconds=3600, scope=scope)
hermes_home.mkdir(parents=True, exist_ok=True)
auth_store = {
"version": 1,
@@ -162,16 +163,6 @@ def _setup_nous_auth(
(hermes_home / "auth.json").write_text(json.dumps(auth_store, indent=2))
def _mint_payload(api_key: str = "agent-key") -> dict:
return {
"api_key": api_key,
"key_id": "key-id-1",
"expires_at": datetime.now(timezone.utc).isoformat(),
"expires_in": 1800,
"reused": False,
}
def _jwt_with_claims(claims: dict) -> str:
def _part(payload: dict) -> str:
raw = json.dumps(payload, separators=(",", ":")).encode("utf-8")
@@ -184,7 +175,7 @@ def _future_iso(seconds: int = 3600) -> str:
return datetime.fromtimestamp(time.time() + seconds, tz=timezone.utc).isoformat()
def _invoke_jwt(*, seconds: int = 3600, scope: object = "inference:invoke inference:mint_agent_key") -> str:
def _invoke_jwt(*, seconds: int = 3600, scope: object = "inference:invoke") -> str:
return _jwt_with_claims({
"sub": "test-user",
"scope": scope,
@@ -209,11 +200,6 @@ def test_resolve_nous_runtime_credentials_prefers_invoke_jwt_and_mirrors(
)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
def _unexpected_mint(*args, **kwargs):
raise AssertionError("legacy agent-key mint should not run for invoke JWT")
monkeypatch.setattr(auth_mod, "_mint_agent_key", _unexpected_mint)
creds = auth_mod.resolve_nous_runtime_credentials(min_key_ttl_seconds=300)
assert creds["api_key"] == token
@@ -278,15 +264,11 @@ def test_resolve_nous_runtime_credentials_invoke_jwt_is_idempotent(
before_mtime = auth_path.stat().st_mtime_ns
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
def _unexpected_mint(*args, **kwargs):
raise AssertionError("stable invoke JWT should not mint a legacy key")
def _unexpected_shared_write(*args, **kwargs):
raise AssertionError("unchanged invoke JWT resolution should not sync shared store")
sync_calls = []
monkeypatch.setattr(auth_mod, "_mint_agent_key", _unexpected_mint)
monkeypatch.setattr(auth_mod, "_write_shared_nous_state", _unexpected_shared_write)
monkeypatch.setattr(
auth_mod,
@@ -330,11 +312,7 @@ def test_resolve_nous_runtime_credentials_trusts_invoke_jwt_exp_over_stale_metad
def _unexpected_refresh(*args, **kwargs):
raise AssertionError("valid invoke JWT should not be refreshed because metadata is stale")
def _unexpected_mint(*args, **kwargs):
raise AssertionError("valid invoke JWT should not fall back to legacy mint")
monkeypatch.setattr(auth_mod, "_refresh_access_token", _unexpected_refresh)
monkeypatch.setattr(auth_mod, "_mint_agent_key", _unexpected_mint)
creds = auth_mod.resolve_nous_runtime_credentials(min_key_ttl_seconds=300)
@@ -347,7 +325,7 @@ def test_resolve_nous_runtime_credentials_trusts_invoke_jwt_exp_over_stale_metad
assert datetime.fromisoformat(singleton["agent_key_expires_at"]).timestamp() > time.time() + 300
def test_resolve_nous_runtime_credentials_does_not_apply_legacy_ttl_to_invoke_jwt(
def test_resolve_nous_runtime_credentials_does_not_apply_agent_key_ttl_to_invoke_jwt(
tmp_path,
monkeypatch,
):
@@ -364,11 +342,6 @@ def test_resolve_nous_runtime_credentials_does_not_apply_legacy_ttl_to_invoke_jw
)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
def _unexpected_mint(*args, **kwargs):
raise AssertionError("1800s legacy min TTL should not force opaque mint for invoke JWT")
monkeypatch.setattr(auth_mod, "_mint_agent_key", _unexpected_mint)
creds = auth_mod.resolve_nous_runtime_credentials(min_key_ttl_seconds=1800)
assert creds["api_key"] == token
@@ -378,7 +351,56 @@ def test_resolve_nous_runtime_credentials_does_not_apply_legacy_ttl_to_invoke_jw
assert payload["credential_pool"]["nous"][0]["agent_key"] == token
def test_legacy_auth_mode_bypasses_usable_invoke_jwt(tmp_path, monkeypatch):
def test_resolve_nous_runtime_credentials_refreshes_legacy_agent_key_to_invoke_jwt(
tmp_path,
monkeypatch,
):
import hermes_cli.auth as auth_mod
hermes_home = tmp_path / "hermes"
refreshed_token = _invoke_jwt(seconds=3600)
_setup_nous_auth(
hermes_home,
access_token="legacy-access-token",
refresh_token="refresh-old",
scope=auth_mod.DEFAULT_NOUS_SCOPE,
expires_at=_future_iso(3600),
expires_in=3600,
agent_key="legacy-opaque-session-key",
agent_key_expires_at=_future_iso(3600),
)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
refresh_calls = []
def _fake_refresh_access_token(*, client, portal_base_url, client_id, refresh_token):
del client, portal_base_url, client_id
refresh_calls.append(refresh_token)
return {
"access_token": refreshed_token,
"refresh_token": "refresh-new",
"expires_in": 3600,
"token_type": "Bearer",
"scope": auth_mod.DEFAULT_NOUS_SCOPE,
}
monkeypatch.setattr(auth_mod, "_refresh_access_token", _fake_refresh_access_token)
creds = auth_mod.resolve_nous_runtime_credentials(min_key_ttl_seconds=300)
assert refresh_calls == ["refresh-old"]
assert creds["api_key"] == refreshed_token
assert creds["source"] == auth_mod.NOUS_AUTH_PATH_INVOKE_JWT
payload = json.loads((hermes_home / "auth.json").read_text())
singleton = payload["providers"]["nous"]
assert singleton["access_token"] == refreshed_token
assert singleton["refresh_token"] == "refresh-new"
assert singleton["agent_key"] == refreshed_token
assert singleton["agent_key_id"] is None
assert payload["credential_pool"]["nous"][0]["agent_key"] == refreshed_token
def test_legacy_auth_mode_is_rejected(tmp_path, monkeypatch):
import hermes_cli.auth as auth_mod
hermes_home = tmp_path / "hermes"
@@ -392,28 +414,14 @@ def test_legacy_auth_mode_bypasses_usable_invoke_jwt(tmp_path, monkeypatch):
)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
mint_calls = []
def _fake_mint_agent_key(*, client, portal_base_url, access_token, min_ttl_seconds):
del client, portal_base_url, min_ttl_seconds
mint_calls.append(access_token)
return _mint_payload(api_key="legacy-after-jwt-401")
monkeypatch.setattr(auth_mod, "_mint_agent_key", _fake_mint_agent_key)
creds = auth_mod.resolve_nous_runtime_credentials(
min_key_ttl_seconds=300,
inference_auth_mode=auth_mod.NOUS_INFERENCE_AUTH_MODE_LEGACY,
)
assert mint_calls == [token]
assert creds["api_key"] == "legacy-after-jwt-401"
assert creds["auth_path"] == auth_mod.NOUS_AUTH_PATH_LEGACY_SESSION_KEY_MINT
payload = json.loads((hermes_home / "auth.json").read_text())
assert payload["providers"]["nous"]["agent_key"] == "legacy-after-jwt-401"
with pytest.raises(ValueError, match="Invalid Nous inference auth mode"):
auth_mod.resolve_nous_runtime_credentials(
min_key_ttl_seconds=300,
inference_auth_mode="legacy",
)
def test_resolve_nous_runtime_credentials_falls_back_when_invoke_scope_missing(
def test_resolve_nous_runtime_credentials_reauths_when_invoke_scope_missing(
tmp_path,
monkeypatch,
):
@@ -428,32 +436,24 @@ def test_resolve_nous_runtime_credentials_falls_back_when_invoke_scope_missing(
_setup_nous_auth(
hermes_home,
access_token=token,
scope=auth_mod.NOUS_LEGACY_AGENT_KEY_SCOPE,
refresh_token="",
scope="inference:mint_agent_key",
expires_at=_future_iso(3600),
expires_in=3600,
)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
calls = []
with pytest.raises(AuthError) as exc:
auth_mod.resolve_nous_runtime_credentials(min_key_ttl_seconds=300)
def _fake_mint_agent_key(*, client, portal_base_url, access_token, min_ttl_seconds):
del client, portal_base_url, min_ttl_seconds
calls.append(access_token)
return _mint_payload(api_key="opaque-agent-key")
monkeypatch.setattr(auth_mod, "_mint_agent_key", _fake_mint_agent_key)
creds = auth_mod.resolve_nous_runtime_credentials(min_key_ttl_seconds=300)
assert calls == [token]
assert creds["api_key"] == "opaque-agent-key"
assert creds["source"] == "portal"
assert exc.value.code == "missing_inference_invoke_scope"
assert exc.value.relogin_required is True
payload = json.loads((hermes_home / "auth.json").read_text())
assert payload["providers"]["nous"]["agent_key"] == "opaque-agent-key"
assert payload["credential_pool"]["nous"][0]["agent_key"] == "opaque-agent-key"
assert payload["providers"]["nous"]["agent_key"] is None
assert "credential_pool" not in payload or not payload["credential_pool"].get("nous")
def test_nous_device_code_login_retries_legacy_scope_when_invoke_refused(monkeypatch):
def test_nous_device_code_login_does_not_retry_legacy_scope_when_invoke_refused(monkeypatch):
import hermes_cli.auth as auth_mod
scopes = []
@@ -461,59 +461,31 @@ def test_nous_device_code_login_retries_legacy_scope_when_invoke_refused(monkeyp
def _fake_request_device_code(*, client, portal_base_url, client_id, scope):
del client, portal_base_url, client_id
scopes.append(scope)
if len(scopes) == 1:
request = httpx.Request("POST", "https://portal.example.com/api/oauth/device/code")
response = httpx.Response(
400,
json={
"error": "invalid_scope",
"error_description": "unsupported inference:invoke",
},
request=request,
)
raise httpx.HTTPStatusError("invalid_scope", request=request, response=response)
return {
"device_code": "device",
"user_code": "user",
"verification_uri": "https://portal.example.com/device",
"verification_uri_complete": "https://portal.example.com/device?code=user",
"expires_in": 600,
"interval": 1,
}
def _fake_poll_for_token(**kwargs):
del kwargs
return {
"access_token": "access-legacy",
"refresh_token": "refresh-legacy",
"expires_in": 900,
"scope": auth_mod.NOUS_LEGACY_AGENT_KEY_SCOPE,
}
def _fake_refresh(state, **kwargs):
del kwargs
refreshed = dict(state)
refreshed["agent_key"] = "opaque-agent-key"
refreshed["agent_key_expires_at"] = _future_iso(1800)
return refreshed
request = httpx.Request("POST", "https://portal.example.com/api/oauth/device/code")
response = httpx.Response(
400,
json={
"error": "invalid_scope",
"error_description": "unsupported inference:invoke",
},
request=request,
)
raise httpx.HTTPStatusError("invalid_scope", request=request, response=response)
monkeypatch.setattr(auth_mod, "_request_device_code", _fake_request_device_code)
monkeypatch.setattr(auth_mod, "_poll_for_token", _fake_poll_for_token)
monkeypatch.setattr(auth_mod, "refresh_nous_oauth_from_state", _fake_refresh)
result = auth_mod._nous_device_code_login(
portal_base_url="https://portal.example.com",
inference_base_url="https://inference.example.com/v1",
open_browser=False,
timeout_seconds=1,
)
with pytest.raises(httpx.HTTPStatusError):
auth_mod._nous_device_code_login(
portal_base_url="https://portal.example.com",
inference_base_url="https://inference.example.com/v1",
open_browser=False,
timeout_seconds=1,
)
assert scopes == [auth_mod.DEFAULT_NOUS_SCOPE, auth_mod.NOUS_LEGACY_AGENT_KEY_SCOPE]
assert result["scope"] == auth_mod.NOUS_LEGACY_AGENT_KEY_SCOPE
assert result["agent_key"] == "opaque-agent-key"
assert scopes == [auth_mod.DEFAULT_NOUS_SCOPE]
def test_forced_legacy_env_skips_invoke_scope_and_jwt_storage(tmp_path, monkeypatch):
def test_legacy_session_env_is_ignored_for_invoke_scope_and_jwt_storage(tmp_path, monkeypatch):
import hermes_cli.auth as auth_mod
hermes_home = tmp_path / "hermes"
@@ -526,25 +498,16 @@ def test_forced_legacy_env_skips_invoke_scope_and_jwt_storage(tmp_path, monkeypa
expires_in=3600,
)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setenv(auth_mod.NOUS_LEGACY_SESSION_KEYS_ENV, "true")
mint_calls = []
def _fake_mint_agent_key(*, client, portal_base_url, access_token, min_ttl_seconds):
del client, portal_base_url, min_ttl_seconds
mint_calls.append(access_token)
return _mint_payload(api_key="forced-legacy-key")
monkeypatch.setattr(auth_mod, "_mint_agent_key", _fake_mint_agent_key)
monkeypatch.setenv("HERMES_AGENT_USE_LEGACY_SESSION_KEYS", "true")
creds = auth_mod.resolve_nous_runtime_credentials(min_key_ttl_seconds=300)
assert mint_calls == [token]
assert creds["api_key"] == "forced-legacy-key"
assert creds["api_key"] == token
payload = json.loads((hermes_home / "auth.json").read_text())
assert payload["providers"]["nous"]["agent_key"] == "forced-legacy-key"
assert payload["providers"]["nous"]["agent_key"] == token
requested_scopes = []
login_token = _invoke_jwt(seconds=3600)
def _fake_request_device_code(*, client, portal_base_url, client_id, scope):
del client, portal_base_url, client_id
@@ -561,31 +524,24 @@ def test_forced_legacy_env_skips_invoke_scope_and_jwt_storage(tmp_path, monkeypa
def _fake_poll_for_token(**kwargs):
del kwargs
return {
"access_token": "access-legacy",
"refresh_token": "refresh-legacy",
"access_token": login_token,
"refresh_token": "refresh-token",
"expires_in": 900,
"scope": auth_mod.NOUS_LEGACY_AGENT_KEY_SCOPE,
"scope": auth_mod.DEFAULT_NOUS_SCOPE,
}
def _fake_refresh(state, **kwargs):
del kwargs
refreshed = dict(state)
refreshed["agent_key"] = "forced-legacy-login-key"
refreshed["agent_key_expires_at"] = _future_iso(1800)
return refreshed
monkeypatch.setattr(auth_mod, "_request_device_code", _fake_request_device_code)
monkeypatch.setattr(auth_mod, "_poll_for_token", _fake_poll_for_token)
monkeypatch.setattr(auth_mod, "refresh_nous_oauth_from_state", _fake_refresh)
auth_mod._nous_device_code_login(
result = auth_mod._nous_device_code_login(
portal_base_url="https://portal.example.com",
inference_base_url="https://inference.example.com/v1",
open_browser=False,
timeout_seconds=1,
)
assert requested_scopes == [auth_mod.NOUS_LEGACY_AGENT_KEY_SCOPE]
assert requested_scopes == [auth_mod.DEFAULT_NOUS_SCOPE]
assert result["agent_key"] == login_token
def test_nous_inference_auth_logs_do_not_include_secret_values(
@@ -596,37 +552,42 @@ def test_nous_inference_auth_logs_do_not_include_secret_values(
import hermes_cli.auth as auth_mod
hermes_home = tmp_path / "hermes"
token = _jwt_with_claims({
"sub": "secret-user",
"scope": "inference:mint_agent_key",
"exp": int(time.time() + 3600),
})
token = _invoke_jwt(seconds=3600)
refreshed_token = _invoke_jwt(seconds=7200)
refresh_token = "refresh-secret-token"
opaque_key = "opaque-secret-agent-key"
_setup_nous_auth(
hermes_home,
access_token=token,
refresh_token=refresh_token,
scope=auth_mod.NOUS_LEGACY_AGENT_KEY_SCOPE,
scope=auth_mod.DEFAULT_NOUS_SCOPE,
expires_at=_future_iso(3600),
expires_in=3600,
)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
def _fake_mint_agent_key(*, client, portal_base_url, access_token, min_ttl_seconds):
del client, portal_base_url, access_token, min_ttl_seconds
return _mint_payload(api_key=opaque_key)
def _fake_refresh_access_token(*, client, portal_base_url, client_id, refresh_token):
del client, portal_base_url, client_id, refresh_token
return {
"access_token": refreshed_token,
"refresh_token": "refresh-new",
"expires_in": 7200,
"token_type": "Bearer",
"scope": auth_mod.DEFAULT_NOUS_SCOPE,
}
monkeypatch.setattr(auth_mod, "_mint_agent_key", _fake_mint_agent_key)
monkeypatch.setattr(auth_mod, "_refresh_access_token", _fake_refresh_access_token)
caplog.set_level(logging.INFO, logger="hermes_cli.auth")
auth_mod.resolve_nous_runtime_credentials(min_key_ttl_seconds=300)
auth_mod.resolve_nous_runtime_credentials(
min_key_ttl_seconds=300,
force_refresh=True,
)
logged = caplog.text
assert "legacy session key path" in logged
assert "using NAS invoke JWT" in logged
assert token not in logged
assert refreshed_token not in logged
assert refresh_token not in logged
assert opaque_key not in logged
def test_get_nous_auth_status_checks_credential_pool(tmp_path, monkeypatch):
@@ -805,68 +766,75 @@ def test_get_nous_auth_status_empty_returns_not_logged_in(tmp_path, monkeypatch)
assert status["logged_in"] is False
def test_refresh_token_persisted_when_mint_returns_insufficient_credits(tmp_path, monkeypatch):
def test_refresh_token_persisted_when_refreshed_jwt_lacks_invoke_scope(tmp_path, monkeypatch):
hermes_home = tmp_path / "hermes"
_setup_nous_auth(hermes_home, refresh_token="refresh-old")
_setup_nous_auth(
hermes_home,
access_token="access-old",
refresh_token="refresh-old",
)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
refresh_calls = []
mint_calls = {"count": 0}
bad_jwt = _jwt_with_claims({
"sub": "test-user",
"scope": "profile",
"exp": int(time.time() + 3600),
})
good_jwt = _invoke_jwt(seconds=3600)
def _fake_refresh_access_token(*, client, portal_base_url, client_id, refresh_token):
refresh_calls.append(refresh_token)
idx = len(refresh_calls)
if len(refresh_calls) == 1:
token = bad_jwt
else:
token = good_jwt
return {
"access_token": f"access-{idx}",
"refresh_token": f"refresh-{idx}",
"expires_in": 0,
"access_token": token,
"refresh_token": f"refresh-{len(refresh_calls)}",
"expires_in": 3600,
"token_type": "Bearer",
"scope": "profile" if len(refresh_calls) == 1 else "inference:invoke",
}
def _fake_mint_agent_key(*, client, portal_base_url, access_token, min_ttl_seconds):
mint_calls["count"] += 1
if mint_calls["count"] == 1:
raise AuthError("credits exhausted", provider="nous", code="insufficient_credits")
return _mint_payload(api_key="agent-key-2")
monkeypatch.setattr("hermes_cli.auth._refresh_access_token", _fake_refresh_access_token)
monkeypatch.setattr("hermes_cli.auth._mint_agent_key", _fake_mint_agent_key)
with pytest.raises(AuthError) as exc:
resolve_nous_runtime_credentials(min_key_ttl_seconds=300)
assert exc.value.code == "insufficient_credits"
assert exc.value.code == "missing_inference_invoke_scope"
state_after_failure = get_provider_auth_state("nous")
assert state_after_failure is not None
assert state_after_failure["refresh_token"] == "refresh-1"
assert state_after_failure["access_token"] == "access-1"
assert state_after_failure["access_token"] == bad_jwt
creds = resolve_nous_runtime_credentials(min_key_ttl_seconds=300)
assert creds["api_key"] == "agent-key-2"
assert creds["api_key"] == good_jwt
assert refresh_calls == ["refresh-old", "refresh-1"]
def test_refresh_token_persisted_when_mint_times_out(tmp_path, monkeypatch):
def test_refresh_token_persisted_when_refreshed_token_is_not_jwt(tmp_path, monkeypatch):
hermes_home = tmp_path / "hermes"
_setup_nous_auth(hermes_home, refresh_token="refresh-old")
_setup_nous_auth(
hermes_home,
access_token="access-old",
refresh_token="refresh-old",
)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
def _fake_refresh_access_token(*, client, portal_base_url, client_id, refresh_token):
return {
"access_token": "access-1",
"refresh_token": "refresh-1",
"expires_in": 0,
"expires_in": 3600,
"token_type": "Bearer",
}
def _fake_mint_agent_key(*, client, portal_base_url, access_token, min_ttl_seconds):
raise httpx.ReadTimeout("mint timeout")
monkeypatch.setattr("hermes_cli.auth._refresh_access_token", _fake_refresh_access_token)
monkeypatch.setattr("hermes_cli.auth._mint_agent_key", _fake_mint_agent_key)
with pytest.raises(httpx.ReadTimeout):
with pytest.raises(AuthError) as exc:
resolve_nous_runtime_credentials(min_key_ttl_seconds=300)
assert exc.value.code == "access_token_not_jwt"
state_after_failure = get_provider_auth_state("nous")
assert state_after_failure is not None
@@ -881,7 +849,11 @@ def test_terminal_refresh_failure_quarantines_tokens(
from hermes_cli import auth as auth_mod
hermes_home = tmp_path / "hermes"
_setup_nous_auth(hermes_home, refresh_token="refresh-old")
_setup_nous_auth(
hermes_home,
access_token="access-old",
refresh_token="refresh-old",
)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
from agent.credential_pool import load_pool
@@ -967,35 +939,36 @@ def test_managed_access_token_refresh_failure_quarantines_tokens(
assert refresh_calls == ["refresh-old"]
def test_mint_retry_uses_latest_rotated_refresh_token(tmp_path, monkeypatch):
def test_unusable_access_token_refresh_uses_latest_rotated_refresh_token(tmp_path, monkeypatch):
hermes_home = tmp_path / "hermes"
_setup_nous_auth(hermes_home, refresh_token="refresh-old")
_setup_nous_auth(
hermes_home,
access_token="access-old",
refresh_token="refresh-old",
)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
refresh_calls = []
mint_calls = {"count": 0}
good_jwt = _invoke_jwt(seconds=3600)
def _fake_refresh_access_token(*, client, portal_base_url, client_id, refresh_token):
refresh_calls.append(refresh_token)
idx = len(refresh_calls)
token = "access-still-not-jwt" if len(refresh_calls) == 1 else good_jwt
return {
"access_token": f"access-{idx}",
"refresh_token": f"refresh-{idx}",
"expires_in": 0,
"access_token": token,
"refresh_token": f"refresh-{len(refresh_calls)}",
"expires_in": 3600,
"token_type": "Bearer",
"scope": "inference:invoke",
}
def _fake_mint_agent_key(*, client, portal_base_url, access_token, min_ttl_seconds):
mint_calls["count"] += 1
if mint_calls["count"] == 1:
raise AuthError("stale access token", provider="nous", code="invalid_token")
return _mint_payload(api_key="agent-key")
monkeypatch.setattr("hermes_cli.auth._refresh_access_token", _fake_refresh_access_token)
monkeypatch.setattr("hermes_cli.auth._mint_agent_key", _fake_mint_agent_key)
with pytest.raises(AuthError) as exc:
resolve_nous_runtime_credentials(min_key_ttl_seconds=300)
assert exc.value.code == "access_token_not_jwt"
creds = resolve_nous_runtime_credentials(min_key_ttl_seconds=300)
assert creds["api_key"] == "agent-key"
assert creds["api_key"] == good_jwt
assert refresh_calls == ["refresh-old", "refresh-1"]
@@ -1170,21 +1143,23 @@ class TestLoginNousSkipKeepsCurrent:
def _full_state_fixture() -> dict:
"""Shape of the dict returned by _nous_device_code_login /
refresh_nous_oauth_from_state. Used as helper input."""
token = _invoke_jwt(seconds=3600)
expires_at = _future_iso(3600)
return {
"portal_base_url": "https://portal.example.com",
"inference_base_url": "https://inference.example.com/v1",
"client_id": "hermes-cli",
"scope": "inference:mint_agent_key",
"scope": "inference:invoke",
"token_type": "Bearer",
"access_token": "access-tok",
"access_token": token,
"refresh_token": "refresh-tok",
"obtained_at": "2026-04-17T22:00:00+00:00",
"expires_at": "2026-04-17T22:15:00+00:00",
"expires_in": 900,
"agent_key": "agent-key-value",
"agent_key_id": "ak-id",
"agent_key_expires_at": "2026-04-18T22:00:00+00:00",
"agent_key_expires_in": 86400,
"expires_at": expires_at,
"expires_in": 3600,
"agent_key": token,
"agent_key_id": None,
"agent_key_expires_at": expires_at,
"agent_key_expires_in": 3600,
"agent_key_reused": False,
"agent_key_obtained_at": "2026-04-17T22:00:10+00:00",
"tls": {"insecure": False, "ca_bundle": None},
@@ -1210,7 +1185,8 @@ def test_persist_nous_credentials_writes_both_pool_and_providers(tmp_path, monke
}))
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
entry = persist_nous_credentials(_full_state_fixture())
state = _full_state_fixture()
entry = persist_nous_credentials(state)
assert entry is not None
assert entry.provider == "nous"
@@ -1220,17 +1196,17 @@ def test_persist_nous_credentials_writes_both_pool_and_providers(tmp_path, monke
# providers.nous populated with the full state (new behaviour)
singleton = payload["providers"]["nous"]
assert singleton["access_token"] == "access-tok"
assert singleton["access_token"] == state["access_token"]
assert singleton["refresh_token"] == "refresh-tok"
assert singleton["agent_key"] == "agent-key-value"
assert singleton["agent_key_expires_at"] == "2026-04-18T22:00:00+00:00"
assert singleton["agent_key"] == state["agent_key"]
assert singleton["agent_key_expires_at"] == state["agent_key_expires_at"]
# credential_pool.nous has exactly one canonical device_code entry
pool_entries = payload["credential_pool"]["nous"]
assert len(pool_entries) == 1, pool_entries
pool_entry = pool_entries[0]
assert pool_entry["source"] == NOUS_DEVICE_CODE_SOURCE
assert pool_entry["agent_key"] == "agent-key-value"
assert pool_entry["agent_key"] == state["agent_key"]
assert pool_entry["inference_base_url"] == "https://inference.example.com/v1"
@@ -1243,7 +1219,6 @@ def test_persist_nous_credentials_allows_recovery_from_401(tmp_path, monkeypatch
providers.nous was empty.
"""
from hermes_cli.auth import (
NOUS_INFERENCE_AUTH_MODE_FRESH,
persist_nous_credentials,
resolve_nous_runtime_credentials,
)
@@ -1256,29 +1231,27 @@ def test_persist_nous_credentials_allows_recovery_from_401(tmp_path, monkeypatch
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
persist_nous_credentials(_full_state_fixture())
new_jwt = _invoke_jwt(seconds=3600)
# Stub the network-touching steps so we don't actually contact the
# portal — the point of this test is that state lookup succeeds and
# doesn't raise "Hermes is not logged into Nous Portal".
def _fake_refresh_access_token(*, client, portal_base_url, client_id, refresh_token):
return {
"access_token": "access-new",
"access_token": new_jwt,
"refresh_token": "refresh-new",
"expires_in": 900,
"expires_in": 3600,
"token_type": "Bearer",
"scope": "inference:invoke",
}
def _fake_mint_agent_key(*, client, portal_base_url, access_token, min_ttl_seconds):
return _mint_payload(api_key="new-agent-key")
monkeypatch.setattr("hermes_cli.auth._refresh_access_token", _fake_refresh_access_token)
monkeypatch.setattr("hermes_cli.auth._mint_agent_key", _fake_mint_agent_key)
creds = resolve_nous_runtime_credentials(
min_key_ttl_seconds=300,
inference_auth_mode=NOUS_INFERENCE_AUTH_MODE_FRESH,
force_refresh=True,
)
assert creds["api_key"] == "new-agent-key"
assert creds["api_key"] == new_jwt
def test_persist_nous_credentials_idempotent_no_duplicate_pool_entries(tmp_path, monkeypatch):
@@ -1304,21 +1277,23 @@ def test_persist_nous_credentials_idempotent_no_duplicate_pool_entries(tmp_path,
persist_nous_credentials(first)
second = _full_state_fixture()
second["access_token"] = "access-second"
second["agent_key"] = "agent-key-second"
second_token = _invoke_jwt(seconds=7200)
second["access_token"] = second_token
second["agent_key"] = second_token
second["agent_key_expires_at"] = _future_iso(7200)
persist_nous_credentials(second)
payload = json.loads((hermes_home / "auth.json").read_text())
# providers.nous reflects the latest write (singleton semantics)
assert payload["providers"]["nous"]["access_token"] == "access-second"
assert payload["providers"]["nous"]["agent_key"] == "agent-key-second"
assert payload["providers"]["nous"]["access_token"] == second_token
assert payload["providers"]["nous"]["agent_key"] == second_token
# credential_pool.nous has exactly one entry, carrying the latest agent_key
pool_entries = payload["credential_pool"]["nous"]
assert len(pool_entries) == 1, pool_entries
assert pool_entries[0]["source"] == NOUS_DEVICE_CODE_SOURCE
assert pool_entries[0]["agent_key"] == "agent-key-second"
assert pool_entries[0]["agent_key"] == second_token
# And no stray `manual:device_code` / `manual:dashboard_device_code` rows
assert not any(
e["source"].startswith("manual:") for e in pool_entries
@@ -1339,13 +1314,14 @@ def test_persist_nous_credentials_reloads_pool_after_singleton_write(tmp_path, m
}))
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
entry = persist_nous_credentials(_full_state_fixture())
state = _full_state_fixture()
entry = persist_nous_credentials(state)
assert entry is not None
assert entry.source == NOUS_DEVICE_CODE_SOURCE
# Label derived by _seed_from_singletons via label_from_token; we don't
# assert its exact value, just that the helper returned a real entry.
assert entry.access_token == "access-tok"
assert entry.agent_key == "agent-key-value"
assert entry.access_token == state["access_token"]
assert entry.agent_key == state["agent_key"]
def test_persist_nous_credentials_embeds_custom_label(tmp_path, monkeypatch):
@@ -1658,7 +1634,8 @@ def test_shared_store_write_and_read_roundtrip(shared_store_env):
_write_shared_nous_state,
)
_write_shared_nous_state(_full_state_fixture())
state = _full_state_fixture()
_write_shared_nous_state(state)
path = _nous_shared_store_path()
assert path.is_file()
@@ -1670,7 +1647,7 @@ def test_shared_store_write_and_read_roundtrip(shared_store_env):
loaded = _read_shared_nous_state()
assert loaded is not None
assert loaded["refresh_token"] == "refresh-tok"
assert loaded["access_token"] == "access-tok"
assert loaded["access_token"] == state["access_token"]
assert loaded["portal_base_url"] == "https://portal.example.com"
assert loaded["inference_base_url"] == "https://inference.example.com/v1"
# Volatile agent_key MUST NOT be persisted to the shared store
@@ -1760,12 +1737,12 @@ def test_try_import_shared_returns_none_on_refresh_failure(
assert auth_mod._read_shared_nous_state() is None
def test_try_import_shared_persists_rotated_token_when_mint_fails(
def test_try_import_shared_persists_rotated_token_when_jwt_validation_fails(
shared_store_env, monkeypatch,
):
"""A forced shared import refresh rotates the single-use token before minting.
"""A forced shared import refresh rotates the single-use token before validation.
If the later agent-key mint fails, the shared store must still keep the
If the later inference-JWT validation fails, the shared store must still keep the
rotated refresh token; otherwise the next import attempt replays the
consumed token and trips refresh-token reuse.
"""
@@ -1785,12 +1762,7 @@ def test_try_import_shared_persists_rotated_token_when_mint_fails(
"token_type": "Bearer",
}
def _fake_mint_agent_key(*, client, portal_base_url, access_token, min_ttl_seconds):
assert access_token == "access-new"
raise AuthError("credits exhausted", provider="nous", code="insufficient_credits")
monkeypatch.setattr(auth_mod, "_refresh_access_token", _fake_refresh_access_token)
monkeypatch.setattr(auth_mod, "_mint_agent_key", _fake_mint_agent_key)
assert auth_mod._try_import_shared_nous_state() is None
@@ -1801,16 +1773,17 @@ def test_try_import_shared_persists_rotated_token_when_mint_fails(
def test_try_import_shared_rehydrates_on_success(shared_store_env, monkeypatch):
"""Happy path: stored refresh_token is accepted, forced refresh+mint
returns a fresh access_token + agent_key, and the returned dict has
"""Happy path: stored refresh_token is accepted, forced refresh
returns a fresh access_token JWT, and the returned dict has
every field persist_nous_credentials() needs.
"""
from hermes_cli import auth as auth_mod
auth_mod._write_shared_nous_state(_full_state_fixture())
fresh_jwt = _invoke_jwt(seconds=7200)
def _fake_refresh(state, **kwargs):
# Simulate portal returning fresh tokens + a new agent_key
# Simulate portal returning a fresh inference JWT.
assert kwargs.get("force_refresh") is True
assert (
kwargs.get("inference_auth_mode")
@@ -1818,10 +1791,10 @@ def test_try_import_shared_rehydrates_on_success(shared_store_env, monkeypatch):
)
return {
**state,
"access_token": "fresh-access-tok",
"access_token": fresh_jwt,
"refresh_token": "fresh-refresh-tok", # rotated
"agent_key": "new-agent-key",
"agent_key_expires_at": "2026-04-19T22:00:00+00:00",
"agent_key": fresh_jwt,
"agent_key_expires_at": _future_iso(7200),
}
monkeypatch.setattr(auth_mod, "refresh_nous_oauth_from_state", _fake_refresh)
@@ -1829,9 +1802,9 @@ def test_try_import_shared_rehydrates_on_success(shared_store_env, monkeypatch):
result = auth_mod._try_import_shared_nous_state()
assert result is not None
assert result["access_token"] == "fresh-access-tok"
assert result["access_token"] == fresh_jwt
assert result["refresh_token"] == "fresh-refresh-tok"
assert result["agent_key"] == "new-agent-key"
assert result["agent_key"] == fresh_jwt
# Preserved from shared state
assert result["portal_base_url"] == "https://portal.example.com"
assert result["client_id"] == "hermes-cli"
@@ -1878,13 +1851,15 @@ def test_shared_store_survives_across_profile_switch(
assert shared["refresh_token"] == "refresh-tok"
# And a successful rehydrate + persist lands nous into profile B
b_jwt = _invoke_jwt(seconds=7200)
def _fake_refresh(state, **kwargs):
return {
**state,
"access_token": "b-access-tok",
"access_token": b_jwt,
"refresh_token": "b-refresh-tok",
"agent_key": "b-agent-key",
"agent_key_expires_at": "2026-04-19T22:00:00+00:00",
"agent_key": b_jwt,
"agent_key_expires_at": _future_iso(7200),
}
monkeypatch.setattr(auth_mod, "refresh_nous_oauth_from_state", _fake_refresh)
@@ -1924,35 +1899,29 @@ def test_runtime_refresh_uses_newer_shared_token_before_local_stale_token(
monkeypatch.setenv("HERMES_HOME", str(profile_b))
shared_state = _full_state_fixture()
shared_state["access_token"] = "shared-fresh-access"
shared_token = _invoke_jwt(seconds=3600)
shared_state["access_token"] = shared_token
shared_state["refresh_token"] = "shared-fresh-refresh"
shared_state["expires_at"] = "2099-01-01T00:00:00+00:00"
shared_state["scope"] = "inference:invoke"
auth_mod._write_shared_nous_state(shared_state)
def _refresh_should_not_happen(**_kwargs):
raise AssertionError("stale profile-local refresh token was used")
minted_with: list[str] = []
def _fake_mint_agent_key(*, client, portal_base_url, access_token, min_ttl_seconds):
minted_with.append(access_token)
return _mint_payload(api_key="agent-key-from-shared-token")
monkeypatch.setattr(auth_mod, "_refresh_access_token", _refresh_should_not_happen)
monkeypatch.setattr(auth_mod, "_mint_agent_key", _fake_mint_agent_key)
creds = auth_mod.resolve_nous_runtime_credentials(
min_key_ttl_seconds=300,
inference_auth_mode=auth_mod.NOUS_INFERENCE_AUTH_MODE_FRESH,
)
assert creds["api_key"] == "agent-key-from-shared-token"
assert minted_with == ["shared-fresh-access"]
assert creds["api_key"] == shared_token
profile_state = auth_mod.get_provider_auth_state("nous")
assert profile_state is not None
assert profile_state["refresh_token"] == "shared-fresh-refresh"
assert profile_state["access_token"] == "shared-fresh-access"
assert profile_state["access_token"] == shared_token
def test_managed_gateway_access_token_uses_newer_shared_token(
tests/hermes_cli/test_proxy.py
+3 -11
View File
@@ -144,7 +144,7 @@ def test_nous_adapter_get_credential_uses_runtime_resolver(tmp_path, monkeypatch
assert cred.token_type == "Bearer"
def test_nous_adapter_retry_credential_forces_legacy_mint(tmp_path, monkeypatch):
def test_nous_adapter_retry_credential_does_not_fallback_on_jwt_401(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_write_auth_store(tmp_path, {
"access_token": "jwt-access",
@@ -155,15 +155,8 @@ def test_nous_adapter_retry_credential_forces_legacy_mint(tmp_path, monkeypatch)
"agent_key": "jwt-access",
})
refreshed_state = {
"api_key": "legacy-bearer",
"base_url": "https://inference-api.nousresearch.com/v1",
"expires_at": "2099-01-01T00:00:00Z",
}
with patch(
"hermes_cli.proxy.adapters.nous_portal.resolve_nous_runtime_credentials",
return_value=refreshed_state,
) as mock_resolve:
adapter = NousPortalAdapter()
cred = adapter.get_retry_credential(
@@ -174,9 +167,8 @@ def test_nous_adapter_retry_credential_forces_legacy_mint(tmp_path, monkeypatch)
status_code=401,
)
assert cred is not None
assert cred.bearer == "legacy-bearer"
assert mock_resolve.call_args.kwargs["inference_auth_mode"] == "legacy"
assert cred is None
mock_resolve.assert_not_called()
def test_nous_adapter_retry_credential_skips_opaque_bearer(tmp_path, monkeypatch):
tests/hermes_cli/test_web_oauth_dispatch.py
+14 -24
View File
@@ -25,6 +25,7 @@ from datetime import datetime, timezone
from unittest.mock import patch
import httpx
import pytest
from fastapi.testclient import TestClient
from hermes_cli.web_server import _SESSION_TOKEN, app
@@ -99,7 +100,7 @@ def test_minimax_login_does_not_launch_anthropic_flow():
assert body["expires_in"] == 600
def test_nous_dashboard_device_flow_honors_legacy_scope_override(monkeypatch):
def test_nous_dashboard_device_flow_ignores_legacy_scope_override(monkeypatch):
from hermes_cli import auth as auth_mod
from hermes_cli import web_server as ws
@@ -109,24 +110,24 @@ def test_nous_dashboard_device_flow_honors_legacy_scope_override(monkeypatch):
requested_scopes.append(kwargs["scope"])
return _fake_nous_device_data()
monkeypatch.setenv(auth_mod.NOUS_LEGACY_SESSION_KEYS_ENV, "true")
monkeypatch.setenv("HERMES_AGENT_USE_LEGACY_SESSION_KEYS", "true")
monkeypatch.setattr(auth_mod, "_request_device_code", fake_request_device_code)
monkeypatch.setattr(ws, "_nous_poller", lambda sid: None)
result = asyncio.run(ws._start_device_code_flow("nous"))
try:
assert requested_scopes == [auth_mod.NOUS_LEGACY_AGENT_KEY_SCOPE]
assert requested_scopes == [auth_mod.DEFAULT_NOUS_SCOPE]
assert result["flow"] == "device_code"
assert result["user_code"] == "NOUS-1234"
assert (
ws._oauth_sessions[result["session_id"]]["scope"]
== auth_mod.NOUS_LEGACY_AGENT_KEY_SCOPE
== auth_mod.DEFAULT_NOUS_SCOPE
)
finally:
ws._oauth_sessions.pop(result["session_id"], None)
def test_nous_dashboard_device_flow_retries_legacy_scope_on_invoke_refusal(monkeypatch):
def test_nous_dashboard_device_flow_does_not_retry_legacy_scope_on_invoke_refusal(monkeypatch):
from hermes_cli import auth as auth_mod
from hermes_cli import web_server as ws
@@ -134,26 +135,15 @@ def test_nous_dashboard_device_flow_retries_legacy_scope_on_invoke_refusal(monke
def fake_request_device_code(**kwargs):
requested_scopes.append(kwargs["scope"])
if len(requested_scopes) == 1:
raise _invoke_scope_refusal()
return _fake_nous_device_data()
raise _invoke_scope_refusal()
monkeypatch.delenv(auth_mod.NOUS_LEGACY_SESSION_KEYS_ENV, raising=False)
monkeypatch.delenv("HERMES_AGENT_USE_LEGACY_SESSION_KEYS", raising=False)
monkeypatch.setattr(auth_mod, "_request_device_code", fake_request_device_code)
monkeypatch.setattr(ws, "_nous_poller", lambda sid: None)
result = asyncio.run(ws._start_device_code_flow("nous"))
try:
assert requested_scopes == [
auth_mod.DEFAULT_NOUS_SCOPE,
auth_mod.NOUS_LEGACY_AGENT_KEY_SCOPE,
]
assert (
ws._oauth_sessions[result["session_id"]]["scope"]
== auth_mod.NOUS_LEGACY_AGENT_KEY_SCOPE
)
finally:
ws._oauth_sessions.pop(result["session_id"], None)
with pytest.raises(httpx.HTTPStatusError):
asyncio.run(ws._start_device_code_flow("nous"))
assert requested_scopes == [auth_mod.DEFAULT_NOUS_SCOPE]
def test_nous_dashboard_poller_preserves_effective_scope_when_token_omits_scope(monkeypatch):
@@ -173,13 +163,13 @@ def test_nous_dashboard_poller_preserves_effective_scope_when_token_omits_scope(
"device_code": "device-code",
"interval": 5,
"expires_at": time.time() + 600,
"scope": auth_mod.NOUS_LEGACY_AGENT_KEY_SCOPE,
"scope": auth_mod.DEFAULT_NOUS_SCOPE,
}
captured_state = {}
def fake_refresh_nous_oauth_from_state(state, **kwargs):
captured_state.update(state)
return {**state, "agent_key": "legacy-agent-key"}
return {**state, "agent_key": "jwt-agent-key"}
monkeypatch.setattr(
auth_mod,
@@ -200,7 +190,7 @@ def test_nous_dashboard_poller_preserves_effective_scope_when_token_omits_scope(
try:
ws._nous_poller(session_id)
assert captured_state["scope"] == auth_mod.NOUS_LEGACY_AGENT_KEY_SCOPE
assert captured_state["scope"] == auth_mod.DEFAULT_NOUS_SCOPE
assert ws._oauth_sessions[session_id]["status"] == "approved"
finally:
ws._oauth_sessions.pop(session_id, None)