feat(auth): log forensic detail at Nous quarantine so terminal auth death is visible

A NAS-hosted Fly agent's Nous bootstrap session can take a terminal
invalid_grant and get quarantined in _quarantine_nous_oauth_state, which
clears the dead tokens from auth.json. Until now this quarantine was
completely silent: the only signal was a downstream "No access token found"
WARNING once the credential pool was already empty, which is too late to
root-cause. Because the Fly log drain is WARNING-only, nothing about the
terminal death reached centralized logging, and a real incident could not be
diagnosed because the evidence was never recorded.

Emit a WARNING+ forensic record AT the quarantine point, before the token
material is cleared. Fields: refresh_token hash prefix (12-char SHA-256 hex,
correlates to NAS's refreshTokenHash), client_id, agent_key_id, error code,
reason, auth.json path/size/mtime/exists, and whether the token was already
past its own expiry. WARNING level is deliberate — INFO never reaches the Fly
drain.

Redaction safety (load-bearing): the log dict is built only from computed
values (hash prefix, sizes, booleans). No raw refresh_token, access_token, or
agent_key bytes are ever passed into the log call, avoiding Hermes's known
credential-literal corruption bug class. A test asserts the raw refresh token
substring is absent from all emitted log output.

Note: no session_id field exists on Nous auth state; provenance is captured
via client_id + agent_key_id, which are non-secret routing identifiers.
This commit is contained in:
Ben 2026-07-07 13:51:22 +10:00 committed by Teknium
parent 536ffedbf4
commit 444dc0da89
2 changed files with 165 additions and 0 deletions

View file

@ -4860,6 +4860,58 @@ def _quarantine_nous_oauth_state(
reason: str,
) -> None:
"""Keep routing metadata but remove dead OAuth material so it is not replayed."""
# Forensic logging BEFORE we clear the token material. A NAS-hosted Fly agent
# can take a terminal invalid_grant and get quarantined here silently: the
# only downstream signal is a "No access token found" WARNING once the pool
# is already empty, which is too late to root-cause. The Fly log drain is
# WARNING-only, so this MUST be logger.warning (INFO never reaches the drain).
#
# Redaction safety: emit ONLY the 12-char SHA-256 hex prefix of the refresh
# token (correlates to NAS's refreshTokenHash without leaking the secret) plus
# sizes/booleans. NEVER pass a raw token/agent_key into the log call — Hermes
# has a known bug class where credential-shaped literals get corrupted in logs.
forensic: Dict[str, Any] = {
"reason": reason,
"error_code": error.code,
# No session_id field exists on Nous state; provenance is client_id +
# agent_key_id (both non-secret routing identifiers).
"client_id": state.get("client_id"),
"agent_key_id": state.get("agent_key_id"),
"refresh_token_fp": _token_fingerprint(state.get("refresh_token")),
}
# On-disk integrity of the auth store at the moment of quarantine.
try:
auth_path = _auth_file_path()
forensic["auth_json_path"] = str(auth_path)
try:
st = os.stat(auth_path)
forensic["auth_json_size"] = st.st_size
forensic["auth_json_mtime"] = st.st_mtime
forensic["auth_json_exists"] = True
except FileNotFoundError:
forensic["auth_json_exists"] = False
except Exception as exc: # pragma: no cover - never let logging break quarantine
forensic["auth_json_stat_error"] = repr(exc)
# Was the token already past its own expiry when it was rejected?
already_expired: Optional[bool] = None
expires_at_raw = state.get("expires_at")
if isinstance(expires_at_raw, str) and expires_at_raw:
try:
parsed = datetime.fromisoformat(expires_at_raw)
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=timezone.utc)
already_expired = parsed < datetime.now(timezone.utc)
except ValueError:
already_expired = None
forensic["token_already_expired"] = already_expired
logger.warning(
"Nous OAuth state quarantined (terminal auth death): %s",
json.dumps(forensic, sort_keys=True, ensure_ascii=False),
)
for key in (
"access_token",
"refresh_token",

View file

@ -0,0 +1,113 @@
"""Redaction-safe forensic logging at the Nous OAuth quarantine path.
A NAS-hosted Fly agent's Nous bootstrap session can take a terminal
``invalid_grant`` and get quarantined (dead tokens cleared from auth.json).
Historically this was completely silent no WARNING+ record at the terminal
rejection, only a downstream "No access token found" warning once the pool was
already empty. The Fly log drain is WARNING-only, so nothing about the terminal
death reached centralized logging. These tests lock in that
``_quarantine_nous_oauth_state`` now emits a WARNING+ forensic record, and the
load-bearing assertion that the raw refresh token never appears in that output.
"""
import hashlib
import logging
import hermes_cli.auth as auth
from hermes_cli.auth import AuthError, _quarantine_nous_oauth_state
# Sanity guard: make sure we are exercising THIS worktree's module and not the
# editable-installed main checkout that shares the venv.
def test_module_resolves_to_this_worktree():
assert "worktrees/bootstrap-h2-logging" in auth.__file__, (
f"auth module resolved to unexpected path: {auth.__file__}"
)
# A distinctive, obviously-fake refresh token so the redaction assertion is
# unambiguous if it ever leaks.
_FAKE_RT = "nous_rt_LEAK_CANARY_do_not_log_raw_0123456789abcdef"
_EXPECTED_FP = hashlib.sha256(_FAKE_RT.encode("utf-8")).hexdigest()[:12]
def _make_state(**overrides):
state = {
"portal_base_url": "https://portal.example.com",
"client_id": "test-client-id",
"access_token": "nous_at_SECRET_access_token_material",
"refresh_token": _FAKE_RT,
"agent_key": "nous_agent_key_SECRET_material",
"agent_key_id": "ak-12345",
"expires_at": "2020-01-01T00:00:00+00:00", # in the past
"obtained_at": "2019-12-31T00:00:00+00:00",
}
state.update(overrides)
return state
def _error():
return AuthError(
"invalid_grant: token expired or revoked",
provider="nous",
code="invalid_grant",
relogin_required=True,
)
def test_quarantine_emits_warning(caplog):
state = _make_state()
with caplog.at_level(logging.WARNING, logger="hermes_cli.auth"):
_quarantine_nous_oauth_state(state, _error(), reason="unit_test_quarantine")
warnings = [r for r in caplog.records if r.levelno >= logging.WARNING]
assert warnings, "expected at least one WARNING+ record from quarantine"
assert any("quarantined" in r.getMessage() for r in warnings)
def test_warning_contains_hash_prefix_and_error_code(caplog):
state = _make_state()
with caplog.at_level(logging.WARNING, logger="hermes_cli.auth"):
_quarantine_nous_oauth_state(state, _error(), reason="unit_test_quarantine")
text = caplog.text
assert _EXPECTED_FP in text, (
f"expected refresh-token hash prefix {_EXPECTED_FP} in log output"
)
assert "invalid_grant" in text, "expected error.code in log output"
assert "unit_test_quarantine" in text, "expected reason in log output"
def test_raw_refresh_token_never_logged(caplog):
"""Load-bearing redaction-safety test: the raw secret must never appear."""
state = _make_state()
with caplog.at_level(logging.DEBUG, logger="hermes_cli.auth"):
_quarantine_nous_oauth_state(state, _error(), reason="unit_test_quarantine")
text = caplog.text
assert _FAKE_RT not in text, "RAW refresh token leaked into log output!"
# Belt-and-suspenders: the access token and agent key must not leak either.
assert "nous_at_SECRET_access_token_material" not in text
assert "nous_agent_key_SECRET_material" not in text
def test_quarantine_no_refresh_token_does_not_throw(caplog):
state = _make_state()
state.pop("refresh_token", None)
with caplog.at_level(logging.WARNING, logger="hermes_cli.auth"):
# Must not raise even when there is no refresh token to fingerprint.
_quarantine_nous_oauth_state(state, _error(), reason="unit_test_no_rt")
warnings = [r for r in caplog.records if r.levelno >= logging.WARNING]
assert warnings, "expected a WARNING even when refresh_token is absent"
# Fingerprint should be null/None, and definitely not the canary prefix.
assert _EXPECTED_FP not in caplog.text
def test_quarantine_clears_token_material():
"""Regression guard: the quarantine still clears dead token keys."""
state = _make_state()
_quarantine_nous_oauth_state(state, _error(), reason="unit_test_quarantine")
for key in ("access_token", "refresh_token", "agent_key", "agent_key_id", "expires_at"):
assert key not in state, f"{key} should have been cleared by quarantine"
assert state["last_auth_error"]["code"] == "invalid_grant"