Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 43 additions & 0 deletions tests/test_auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -203,6 +203,49 @@ def test_read_internal_token_returns_none_when_missing(
assert _read_internal_token() is None


# ---------------------------------------------------------------------------
# Regression — touch_last_used stamps E2E connects
# ---------------------------------------------------------------------------


class TestTouchLastUsed:
"""E2E nodes never call validate(), so last_used_at needs a direct stamp."""

def _store(self, tmp_path: Path):
from cryptography.fernet import Fernet
from hermes_node_plugin.tokens import TokenStore

key = Fernet.generate_key().decode("ascii")
store = TokenStore(path=tmp_path / "tokens.json", key=key)
return store

def test_touch_last_used_stamps_existing_record(self, tmp_path: Path) -> None:
store = self._store(tmp_path)
store.create("workmac")
before = store.list()[0]
assert before.last_used_at is None

store.touch_last_used("workmac")

after = store.list()[0]
assert after.last_used_at is not None, "last_used_at should be stamped"

def test_touch_last_used_skips_revoked(self, tmp_path: Path) -> None:
store = self._store(tmp_path)
store.create("workmac")
store.revoke("workmac")

# Revoked records must not be stamped (mirrors validate() semantics).
store.touch_last_used("workmac")
assert store.list()[0].last_used_at is None

def test_touch_last_used_unknown_name_is_noop(self, tmp_path: Path) -> None:
store = self._store(tmp_path)
# Should not raise on a name with no record.
store.touch_last_used("ghost")
assert store.list() == []


# ---------------------------------------------------------------------------
# Smoke — create_app
# ---------------------------------------------------------------------------
Expand Down
39 changes: 39 additions & 0 deletions tokens.py
Original file line number Diff line number Diff line change
Expand Up @@ -528,6 +528,45 @@ def validate(self, name: str, presented_token: str) -> bool:
pass
return True

def touch_last_used(self, name: str) -> None:
"""Stamp ``last_used_at`` for ``name`` without validating a token.

The legacy auth path records the connect time inside
:meth:`validate` (it has the presented token to check). E2E nodes
authenticate via proof instead — :meth:`validate` is never called
for them — so ``last_used_at`` would otherwise stay ``None`` and
``hermes node list`` renders it as ``-``. This method lets the E2E
path record the successful connect explicitly, using the node name
only (the proof was already verified by the caller).

Best-effort, mirroring :meth:`validate`: a write failure is
swallowed — the connect already succeeded, the stamp is a
nice-to-have audit field. No token comparison happens here.

Args:
name: Node name whose record to stamp.

Raises:
TokenStoreError: ``name`` fails the character / length policy.
"""
name = _validate_name(name)
now = _now_iso()

def _write(records: list[_StoredRecord]) -> list[_StoredRecord]:
for rec in records:
if rec.name == name and not rec.revoked:
rec.last_used_at = now
break
return records

try:
self._mutate(_write)
except Exception: # noqa: S110
# Broad on purpose: a transient disk/permission failure must
# not convert a successful auth into a server crash. Same
# contract as validate()'s best-effort write.
pass

def resolve_token(self, name: str) -> str:
"""Return the raw pairing token for ``name``, or raise TokenStoreError.

Expand Down
10 changes: 10 additions & 0 deletions wsserver/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -955,6 +955,16 @@ async def ws_nodes(websocket: WebSocket) -> None:
server_proof = _e2e.compute_server_proof(e2e_handshake_key)
is_valid = True # proof verified — skip token store check
logger.warning("E2E proof verified for %r session=%s", auth.node_name, session_id)
# E2E bypasses token_store.validate() (the only place that
# stamps last_used_at), so record the successful connect here
# or `hermes node list` shows "-" forever for E2E nodes.
# Best-effort; offloaded to a thread like the legacy path.
try:
await asyncio.to_thread(token_store.touch_last_used, auth.node_name)
except Exception as exc:
logger.debug(
"touch_last_used failed for %r: %s", auth.node_name, exc
)
else:
# Legacy auth — validate token
is_valid = False
Expand Down
Loading