wehub-resource-sync
b4fbd6fe9f
Deploy Site / deploy-vercel (push) Has been skipped
Deploy Site / deploy-docs (push) Has been skipped
Build Skills Index / build-index (push) Has been skipped
CI / Deny unrelated histories (push) Has been skipped
CI / Detect affected areas (push) Successful in 27m35s
CI / OSV scan (push) Failing after 4s
CI / Build&Test Docker image (push) Successful in 9s
CI / Supply-chain scan (push) Has been skipped
CI / Lint Docker scripts (push) Failing after 5m13s
CI / Check contributors (push) Failing after 12m8s
CI / Docs Site (push) Failing after 12m8s
CI / TypeScript (push) Failing after 12m8s
CI / Python lints (push) Failing after 12m9s
CI / Python tests (push) Failing after 12m9s
CI / Check uv.lock (push) Failing after 23m22s
CI / CI timing report (push) Has been cancelled
Build Skills Index / trigger-deploy (push) Has been cancelled
CI / All required checks pass (push) Has been cancelled
255 lines
11 KiB
Python
255 lines
11 KiB
Python
"""Tests for plugins/memory/honcho/oauth.py — OAuth grant storage + refresh."""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from plugins.memory.honcho import oauth
|
|
from plugins.memory.honcho.oauth import OAuthCredential
|
|
|
|
|
|
def _host_block(refresh="hch-rt-old", expires_at=10_000):
|
|
return {
|
|
"apiKey": "hch-at-old",
|
|
"oauth": {
|
|
"refreshToken": refresh,
|
|
"expiresAt": expires_at,
|
|
"clientId": "hermes-desktop",
|
|
"tokenEndpoint": "http://localhost:8000/oauth/token",
|
|
"scope": "write",
|
|
"tokenType": "Bearer",
|
|
},
|
|
}
|
|
|
|
|
|
def _write(path: Path, raw: dict) -> None:
|
|
path.write_text(json.dumps(raw), encoding="utf-8")
|
|
|
|
|
|
class TestTokenDetection:
|
|
def test_access_token_prefix(self):
|
|
assert oauth.is_oauth_access_token("hch-at-abc")
|
|
assert not oauth.is_oauth_access_token("hch-v3-abc")
|
|
assert not oauth.is_oauth_access_token("hch-rt-abc")
|
|
assert not oauth.is_oauth_access_token(None)
|
|
|
|
|
|
class TestCredentialModel:
|
|
def test_roundtrip(self):
|
|
cred = OAuthCredential.from_host_block(_host_block())
|
|
assert cred is not None
|
|
block = cred.oauth_block()
|
|
assert block["refreshToken"] == "hch-rt-old"
|
|
assert block["expiresAt"] == 10_000
|
|
assert block["clientId"] == "hermes-desktop"
|
|
|
|
def test_incomplete_block_returns_none(self):
|
|
# plain API key (no oauth sub-block)
|
|
assert OAuthCredential.from_host_block({"apiKey": "hch-v3-x"}) is None
|
|
# oauth block missing refreshToken
|
|
bad = _host_block()
|
|
del bad["oauth"]["refreshToken"]
|
|
assert OAuthCredential.from_host_block(bad) is None
|
|
|
|
def test_is_expired_respects_skew(self):
|
|
cred = OAuthCredential.from_host_block(_host_block(expires_at=1000))
|
|
assert not cred.is_expired(now=800, skew=120) # 1000-120=880 > 800
|
|
assert cred.is_expired(now=900, skew=120) # 900 >= 880
|
|
|
|
|
|
class TestEnsureFreshToken:
|
|
def test_no_oauth_credential_is_noop(self, tmp_path):
|
|
path = tmp_path / "honcho.json"
|
|
_write(path, {"hosts": {"hermes": {"apiKey": "hch-v3-static"}}})
|
|
token, refreshed = oauth.ensure_fresh_token(path, "hermes", now=0)
|
|
assert token is None and refreshed is False
|
|
|
|
def test_fresh_token_skips_refresh(self, tmp_path, monkeypatch):
|
|
path = tmp_path / "honcho.json"
|
|
_write(path, {"hosts": {"hermes": _host_block(expires_at=10_000)}})
|
|
monkeypatch.setattr(
|
|
oauth, "_http_post_form",
|
|
lambda *a, **k: pytest.fail("refresh must not be called when fresh"),
|
|
)
|
|
token, refreshed = oauth.ensure_fresh_token(path, "hermes", now=0)
|
|
assert token == "hch-at-old" and refreshed is False
|
|
|
|
def test_fresh_token_served_from_cache_without_disk(self, tmp_path, monkeypatch):
|
|
path = tmp_path / "honcho.json"
|
|
_write(path, {"hosts": {"hermes": _host_block(expires_at=10_000)}})
|
|
oauth._expiry_cache.clear()
|
|
# First call seeds the cache from disk.
|
|
oauth.ensure_fresh_token(path, "hermes", now=0)
|
|
# Second call must not touch disk while the token is well clear of expiry.
|
|
monkeypatch.setattr(
|
|
oauth, "_read_config",
|
|
lambda *a, **k: pytest.fail("disk must not be read while token is fresh"),
|
|
)
|
|
token, refreshed = oauth.ensure_fresh_token(path, "hermes", now=100)
|
|
assert token == "hch-at-old" and refreshed is False
|
|
|
|
def test_expired_token_refreshes_and_persists_rotation(self, tmp_path, monkeypatch):
|
|
path = tmp_path / "honcho.json"
|
|
_write(path, {"hosts": {"hermes": _host_block(expires_at=100)}})
|
|
|
|
def fake_post(url, data, timeout):
|
|
assert data["grant_type"] == "refresh_token"
|
|
assert data["refresh_token"] == "hch-rt-old"
|
|
assert data["client_id"] == "hermes-desktop"
|
|
return {
|
|
"access_token": "hch-at-new",
|
|
"refresh_token": "hch-rt-new",
|
|
"expires_in": 3600,
|
|
"scope": "write",
|
|
"token_type": "Bearer",
|
|
}
|
|
|
|
monkeypatch.setattr(oauth, "_http_post_form", fake_post)
|
|
token, refreshed = oauth.ensure_fresh_token(path, "hermes", now=1000)
|
|
assert token == "hch-at-new" and refreshed is True
|
|
|
|
# Rotated refresh token + new access token + absolute expiry persisted.
|
|
saved = json.loads(path.read_text())["hosts"]["hermes"]
|
|
assert saved["apiKey"] == "hch-at-new"
|
|
assert saved["oauth"]["refreshToken"] == "hch-rt-new"
|
|
assert saved["oauth"]["expiresAt"] == 1000 + 3600
|
|
|
|
def test_refresh_failure_fails_open(self, tmp_path, monkeypatch):
|
|
path = tmp_path / "honcho.json"
|
|
_write(path, {"hosts": {"hermes": _host_block(expires_at=100)}})
|
|
|
|
def boom(*a, **k):
|
|
raise RuntimeError("network down")
|
|
|
|
monkeypatch.setattr(oauth, "_http_post_form", boom)
|
|
token, refreshed = oauth.ensure_fresh_token(path, "hermes", now=1000)
|
|
# Stale token returned, no crash, file untouched.
|
|
assert token == "hch-at-old" and refreshed is False
|
|
assert json.loads(path.read_text())["hosts"]["hermes"]["apiKey"] == "hch-at-old"
|
|
|
|
def test_double_check_uses_disk_when_already_rotated(self, tmp_path, monkeypatch):
|
|
# Simulates a concurrent thread that rotated the token on disk after our
|
|
# stale in-memory snapshot: the locked re-read must skip the HTTP call.
|
|
path = tmp_path / "honcho.json"
|
|
_write(path, {"hosts": {"hermes": _host_block(refresh="hch-rt-fresh", expires_at=10_000)}})
|
|
stale_raw = {"hosts": {"hermes": _host_block(refresh="hch-rt-old", expires_at=100)}}
|
|
stale_raw["hosts"]["hermes"]["apiKey"] = "hch-at-stale"
|
|
monkeypatch.setattr(
|
|
oauth, "_http_post_form",
|
|
lambda *a, **k: pytest.fail("must not refresh; disk token is fresh"),
|
|
)
|
|
token, refreshed = oauth.ensure_fresh_token(path, "hermes", stale_raw, now=1000)
|
|
assert token == "hch-at-old" # the on-disk fresh credential's access token
|
|
|
|
def test_refresh_holds_cross_process_lock(self, tmp_path, monkeypatch):
|
|
# A second opener must not grab <config>.lock mid-refresh — proving the
|
|
# rotation is serialized machine-wide so peers can't replay the token.
|
|
fcntl = pytest.importorskip("fcntl")
|
|
path = tmp_path / "honcho.json"
|
|
_write(path, {"hosts": {"hermes": _host_block(expires_at=100)}})
|
|
seen = {}
|
|
|
|
def fake_post(url, data, timeout):
|
|
with open(f"{path}.lock", "a+b") as other:
|
|
try:
|
|
fcntl.flock(other.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
fcntl.flock(other.fileno(), fcntl.LOCK_UN)
|
|
seen["held"] = False
|
|
except OSError:
|
|
seen["held"] = True
|
|
return {"access_token": "hch-at-new", "refresh_token": "hch-rt-new",
|
|
"expires_in": 3600, "scope": "write", "token_type": "Bearer"}
|
|
|
|
monkeypatch.setattr(oauth, "_http_post_form", fake_post)
|
|
token, refreshed = oauth.ensure_fresh_token(path, "hermes", now=1000)
|
|
assert refreshed is True and seen.get("held") is True
|
|
# Released afterward: a non-blocking acquire now succeeds.
|
|
with open(f"{path}.lock", "a+b") as fh:
|
|
fcntl.flock(fh.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
fcntl.flock(fh.fileno(), fcntl.LOCK_UN)
|
|
|
|
def test_refresh_degrades_when_lock_unavailable(self, tmp_path, monkeypatch):
|
|
# No flock (unsupported FS/platform) must not block refresh — it falls
|
|
# back to in-process serialization only.
|
|
fcntl = pytest.importorskip("fcntl")
|
|
path = tmp_path / "honcho.json"
|
|
_write(path, {"hosts": {"hermes": _host_block(expires_at=100)}})
|
|
|
|
def no_flock(*a, **k):
|
|
raise OSError("flock unsupported")
|
|
|
|
monkeypatch.setattr(fcntl, "flock", no_flock)
|
|
monkeypatch.setattr(
|
|
oauth, "_http_post_form",
|
|
lambda *a, **k: {"access_token": "hch-at-new", "refresh_token": "hch-rt-new",
|
|
"expires_in": 3600, "scope": "write", "token_type": "Bearer"},
|
|
)
|
|
token, refreshed = oauth.ensure_fresh_token(path, "hermes", now=1000)
|
|
assert token == "hch-at-new" and refreshed is True
|
|
|
|
|
|
class TestInstallGrant:
|
|
def test_deep_merges_config_and_preserves_other_hosts(self, tmp_path):
|
|
path = tmp_path / "honcho.json"
|
|
_write(path, {
|
|
"apiKey": "hch-v3-root", # root static key preserved
|
|
"hosts": {
|
|
"obsidian": {"workspace": "obsidian"},
|
|
"hermes": {"workspace": "hermes", "saveMessages": False},
|
|
},
|
|
})
|
|
grant = {
|
|
"access_token": "hch-at-fresh",
|
|
"refresh_token": "hch-rt-fresh",
|
|
"expires_in": 3600,
|
|
"scope": "write",
|
|
"config": {
|
|
"environment": "production",
|
|
"hosts": {"hermes": {"saveMessages": True, "recallMode": "hybrid"}},
|
|
},
|
|
}
|
|
cred = oauth.install_grant(
|
|
path, "hermes", grant,
|
|
client_id="hermes-desktop",
|
|
token_endpoint="http://localhost:8000/oauth/token",
|
|
now=1000,
|
|
)
|
|
assert cred.expires_at == 1000 + 3600
|
|
|
|
saved = json.loads(path.read_text())
|
|
assert saved["apiKey"] == "hch-v3-root" # untouched
|
|
assert saved["hosts"]["obsidian"] == {"workspace": "obsidian"} # untouched
|
|
h = saved["hosts"]["hermes"]
|
|
assert h["apiKey"] == "hch-at-fresh"
|
|
assert h["oauth"]["refreshToken"] == "hch-rt-fresh"
|
|
assert h["saveMessages"] is True # grant config won the deep-merge
|
|
assert h["recallMode"] == "hybrid" # new key added
|
|
assert h["workspace"] == "hermes" # pre-existing key preserved
|
|
assert saved["environment"] == "production" # root key from grant
|
|
|
|
def test_rejects_grant_without_tokens(self, tmp_path):
|
|
path = tmp_path / "honcho.json"
|
|
_write(path, {})
|
|
with pytest.raises(ValueError):
|
|
oauth.install_grant(
|
|
path, "hermes", {"access_token": "hch-at-x"}, # no refresh_token
|
|
client_id="c", token_endpoint="e",
|
|
)
|
|
|
|
|
|
class TestApplyTokenToClient:
|
|
def test_mutates_live_bearer(self):
|
|
class FakeHttp:
|
|
api_key = "hch-at-old"
|
|
|
|
class FakeClient:
|
|
_http = FakeHttp()
|
|
|
|
client = FakeClient()
|
|
assert oauth.apply_token_to_client(client, "hch-at-new") is True
|
|
assert client._http.api_key == "hch-at-new"
|
|
|
|
def test_returns_false_when_shape_unknown(self):
|
|
assert oauth.apply_token_to_client(object(), "hch-at-new") is False
|