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

refactor(plugins/dashboard_auth): JwtOAuthProvider base (complete/refresh/verify + JWKS client) and NonInteractiveMixin in _shared; nous/self_hosted/basic/drain become thin subclasses

Browse Source
This commit is contained in:
Teknium
2026-09-02 22:45:15 -07:00
parent d3a429ce42
commit d20d37e5e0
5 changed files with 133 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/dashboard_auth/_shared.py
+75 -1
View File
@@ -17,7 +17,9 @@ from typing import Any, Callable, Dict, Optional
import httpx
from hermes_cli.dashboard_auth import InvalidCodeError, LoginStart, ProviderError, Session, classify_jwks_lookup_error
from hermes_cli.dashboard_auth import (
DashboardAuthProvider, InvalidCodeError, LoginStart, ProviderError, RefreshExpiredError, Session,
classify_jwks_lookup_error)
# JWKS Cache-Control max-age (nous contract C7); self-hosted mirrors it.
JWKS_CACHE_SECONDS = 300
@@ -221,3 +223,75 @@ def verify_jwt(
except Exception:
pass
raise ProviderError(f"{label} verification failed: {exc}{details}") from exc
# ---- Shared provider skeletons ----
class NonInteractiveMixin:
"""OAuth-redirect stubs for providers without a browser login flow (password / service
credential). ``_NOT_INTERACTIVE`` is the operator-facing reason; ``_NO_START_LOGIN``
optionally overrides the ``start_login`` message."""
_NOT_INTERACTIVE: str = ""
_NO_START_LOGIN: str = ""
def start_login(self, *, redirect_uri: str) -> LoginStart:
raise NotImplementedError(self._NO_START_LOGIN or self._NOT_INTERACTIVE)
def complete_login(self, *, code: str, state: str, code_verifier: str, redirect_uri: str) -> Session:
raise NotImplementedError(self._NOT_INTERACTIVE)
class JwtOAuthProvider(DashboardAuthProvider):
"""Authorization-code + PKCE provider whose session token is a JWT we verify ourselves
(nous: Portal access token; self-hosted: OIDC ID token). Subclasses implement
``_jwks_uri``, ``_claims_for``, ``_grant``, ``_refresh_request`` and ``_session``."""
_jwks_client: Any = None
def _jwks_uri(self) -> str:
raise NotImplementedError
def _claims_for(self, token: str) -> Dict[str, Any]:
raise NotImplementedError
def _grant(
self, data: Dict[str, str], *, bad_request_exc: type[Exception], headers: Optional[Dict[str, str]] = None,
previous_refresh_token: str = "",
) -> Session:
raise NotImplementedError
def _refresh_request(self, refresh_token: str) -> tuple[Dict[str, str], Optional[Dict[str, str]]]:
"""``(form_data, extra_headers)`` for the refresh grant."""
raise NotImplementedError
def _session(self, token: str, refresh_token: str, claims: Dict[str, Any]) -> Session:
raise NotImplementedError
def _get_jwks_client(self) -> Any:
if self._jwks_client is None:
self._jwks_client = make_jwks_client(self._jwks_uri())
return self._jwks_client
def complete_login(self, *, code: str, state: str, code_verifier: str, redirect_uri: str) -> Session:
# ``state`` is verified by the auth-route layer before this call.
return self._grant(
{"grant_type": "authorization_code", "code": code, "redirect_uri": redirect_uri,
"client_id": self._client_id, "code_verifier": code_verifier},
bad_request_exc=InvalidCodeError)
def refresh_session(self, *, refresh_token: str) -> Session:
if not refresh_token:
raise RefreshExpiredError("no refresh token present in session")
data, headers = self._refresh_request(refresh_token)
return self._grant(
data, headers=headers, bad_request_exc=RefreshExpiredError, previous_refresh_token=refresh_token)
def verify_session(self, *, access_token: str) -> Optional[Session]:
# None on expiry/invalidity (middleware then tries refresh); a ProviderError
# (JWKS unreachable) bubbles up so middleware emits 503.
try:
claims = self._claims_for(access_token)
except InvalidCodeError:
return None
return self._session(access_token, "", claims)
plugins/dashboard_auth/basic/__init__.py
+8 -14
View File
@@ -20,9 +20,9 @@ import secrets
import time
from typing import Optional
from hermes_cli.dashboard_auth import (
DashboardAuthProvider, InvalidCredentialsError, LoginStart, RefreshExpiredError, Session)
from plugins.dashboard_auth._shared import SkipRegistration, load_config_section, register_provider, resolve_env_or_cfg
from hermes_cli.dashboard_auth import DashboardAuthProvider, InvalidCredentialsError, RefreshExpiredError, Session
from plugins.dashboard_auth._shared import (
NonInteractiveMixin, SkipRegistration, load_config_section, register_provider, resolve_env_or_cfg)
logger = logging.getLogger(__name__)
_TAG = "dashboard-auth-basic"
@@ -111,12 +111,16 @@ def _unsign(token: str, secret: bytes, kind: str) -> Optional[dict]:
# ---- Provider ----
class BasicAuthProvider(DashboardAuthProvider):
class BasicAuthProvider(NonInteractiveMixin, DashboardAuthProvider):
"""Username/password provider with stateless HMAC-signed sessions."""
name = "basic"
display_name = "Username & Password"
supports_password = True
_NOT_INTERACTIVE = "BasicAuthProvider is password-only; use complete_password_login."
_NO_START_LOGIN = (
"BasicAuthProvider is password-only; there is no OAuth redirect flow. "
"The login page POSTs to /auth/password-login instead.")
def __init__(self, *, username: str, password_hash: str, secret: bytes, ttl_seconds: int = _DEFAULT_TTL_SECONDS) -> None:
if not username:
@@ -130,16 +134,6 @@ class BasicAuthProvider(DashboardAuthProvider):
self._secret = secret
self._ttl = max(60, int(ttl_seconds))
# ---- OAuth methods: not used (pure-password provider) ------------------
def start_login(self, *, redirect_uri: str) -> LoginStart:
raise NotImplementedError(
"BasicAuthProvider is password-only; there is no OAuth redirect flow. "
"The login page POSTs to /auth/password-login instead.")
def complete_login(self, *, code: str, state: str, code_verifier: str, redirect_uri: str) -> Session:
raise NotImplementedError("BasicAuthProvider is password-only; use complete_password_login.")
# ---- password login ----------------------------------------------------
def complete_password_login(self, *, username: str, password: str) -> Session:
plugins/dashboard_auth/drain/__init__.py
+6 -10
View File
@@ -16,8 +16,8 @@ import os
from collections import Counter
from typing import Optional
from hermes_cli.dashboard_auth import DashboardAuthProvider, LoginStart, Session, TokenPrincipal
from plugins.dashboard_auth._shared import SkipRegistration, load_config_section, register_provider
from hermes_cli.dashboard_auth import DashboardAuthProvider, Session, TokenPrincipal
from plugins.dashboard_auth._shared import NonInteractiveMixin, SkipRegistration, load_config_section, register_provider
logger = logging.getLogger(__name__)
_TAG = "dashboard-auth-drain"
@@ -65,13 +65,15 @@ def assess_secret_strength(secret: str, *, min_chars: int = _DEFAULT_MIN_SECRET_
return None
class DrainSecretProvider(DashboardAuthProvider):
class DrainSecretProvider(NonInteractiveMixin, DashboardAuthProvider):
"""Non-interactive shared-bearer-secret provider for drain control."""
name = "drain-secret"
display_name = "Drain Control (service credential)"
supports_token = True
supports_session = False
_NOT_INTERACTIVE = "DrainSecretProvider is a non-interactive service credential."
_NO_START_LOGIN = "DrainSecretProvider is a non-interactive service credential; there is no login flow."
def __init__(self, *, secret: str, scope: str = "drain") -> None:
# Defence in depth: construction enforces the entropy bar too, so a
@@ -93,19 +95,13 @@ class DrainSecretProvider(DashboardAuthProvider):
# ---- interactive methods: unsupported (service credential only) --------
def start_login(self, *, redirect_uri: str) -> LoginStart:
raise NotImplementedError("DrainSecretProvider is a non-interactive service credential; there is no login flow.")
def complete_login(self, *, code: str, state: str, code_verifier: str, redirect_uri: str) -> Session:
raise NotImplementedError("DrainSecretProvider is a non-interactive service credential.")
def verify_session(self, *, access_token: str) -> Optional[Session]:
# Never mints a Session, so never recognises a cookie. Return None (don't raise)
# so it stacks harmlessly in the cookie-verify loop.
return None
def refresh_session(self, *, refresh_token: str) -> Session:
raise NotImplementedError("DrainSecretProvider is a non-interactive service credential.")
raise NotImplementedError(self._NOT_INTERACTIVE)
def revoke_session(self, *, refresh_token: str) -> None:
return None
plugins/dashboard_auth/nous/__init__.py
+21 -44
View File
@@ -13,13 +13,12 @@ from __future__ import annotations
import logging
from typing import Any, Dict, Optional
from hermes_cli.dashboard_auth import (
DashboardAuthProvider, InvalidCodeError, LoginStart, ProviderError, RefreshExpiredError, Session)
from hermes_cli.dashboard_auth import LoginStart, ProviderError, Session
from plugins.dashboard_auth._shared import (
JwtOAuthProvider,
SkipRegistration,
exchange_token,
load_config_section,
make_jwks_client,
pkce_login_start,
refresh_token_from,
register_provider,
@@ -38,7 +37,7 @@ _EXPECTED_CONTRACT_VERSION = 1 # contract C11
LAST_SKIP_REASON: str = "" # cleared on every register() so restarts don't leak stale reasons
class NousDashboardAuthProvider(DashboardAuthProvider):
class NousDashboardAuthProvider(JwtOAuthProvider):
"""Nous Portal OAuth via authorization-code + PKCE (S256)."""
name = "nous"
@@ -56,63 +55,41 @@ class NousDashboardAuthProvider(DashboardAuthProvider):
self._token_url = f"{self._portal_url}/api/oauth/token"
self._jwks_client: Any = None # lazily built (crypto import cost)
# ---- public API (DashboardAuthProvider) -------------------------------
def start_login(self, *, redirect_uri: str) -> LoginStart:
validate_redirect_uri(redirect_uri)
return pkce_login_start(self._authorize_url, client_id=self._client_id, scope=_SCOPE, redirect_uri=redirect_uri)
def complete_login(self, *, code: str, state: str, code_verifier: str, redirect_uri: str) -> Session:
# ``state`` is verified by the auth-route layer; Portal doesn't re-check it at the token endpoint.
return self._token_grant(
{
"grant_type": "authorization_code", "code": code, "redirect_uri": redirect_uri,
"client_id": self._client_id, "code_verifier": code_verifier},
bad_request_exc=InvalidCodeError)
def refresh_session(self, *, refresh_token: str) -> Session:
if not refresh_token:
raise RefreshExpiredError("no refresh token present in session")
# The RT goes in BOTH the body (Portal's request schema requires it) and the
# ``x-nous-refresh-token`` header (Portal reconciles the two and keeps the value
# out of body access logs). Header-only → 400.
return self._token_grant(
{"grant_type": "refresh_token", "client_id": self._client_id, "refresh_token": refresh_token},
headers={"x-nous-refresh-token": refresh_token},
bad_request_exc=RefreshExpiredError)
def verify_session(self, *, access_token: str) -> Optional[Session]:
# None on expiry/invalidity (middleware then tries refresh); a ProviderError
# (JWKS unreachable) bubbles up so middleware emits 503.
try:
claims = self._verify_jwt(access_token)
except InvalidCodeError:
return None
return self._session(access_token, "", claims)
def revoke_session(self, *, refresh_token: str) -> None:
# Portal exposes no token-endpoint revocation grant; logout is client-side cookie
# clearing and the RT expires within its 24h TTL.
return None
# ---- internals --------------------------------------------------------
# ---- JwtOAuthProvider hooks -------------------------------------------
def _token_grant(
def _jwks_uri(self) -> str:
return self._jwks_url
def _refresh_request(self, refresh_token: str) -> tuple[Dict[str, str], Optional[Dict[str, str]]]:
# The RT goes in BOTH the body (Portal's request schema requires it) and the
# ``x-nous-refresh-token`` header (Portal reconciles the two and keeps the value
# out of body access logs). Header-only → 400.
return (
{"grant_type": "refresh_token", "client_id": self._client_id, "refresh_token": refresh_token},
{"x-nous-refresh-token": refresh_token})
def _grant(
self, data: Dict[str, str], *, bad_request_exc: type[Exception], headers: Optional[Dict[str, str]] = None,
previous_refresh_token: str = "",
) -> Session:
access_token, payload = exchange_token(
self._token_url, data, headers=headers, bad_request_exc=bad_request_exc,
idp="Portal", endpoint="Portal token endpoint", token_key="access_token",
missing_msg="Portal token response missing access_token")
# Rotating RT the caller MUST persist back to the cookie.
return self._session(access_token, refresh_token_from(payload), self._verify_jwt(access_token))
return self._session(access_token, refresh_token_from(payload), self._claims_for(access_token))
def _get_jwks_client(self) -> Any:
if self._jwks_client is None:
self._jwks_client = make_jwks_client(self._jwks_url)
return self._jwks_client
def _verify_jwt(self, access_token: str) -> Dict[str, Any]:
def _claims_for(self, access_token: str) -> Dict[str, Any]:
claims = verify_jwt(
access_token, self._get_jwks_client(), algorithms=["RS256"],
audience=self._client_id, # contract C2: bare client_id
@@ -133,11 +110,11 @@ class NousDashboardAuthProvider(DashboardAuthProvider):
f"unsupported oauth_contract_version={contract_version!r}, expected {_EXPECTED_CONTRACT_VERSION}")
return claims
def _session(self, access_token: str, refresh_token: str, claims: Dict[str, Any]) -> Session:
# Contract C4: no email / display_name in tokens.
return session_from_claims(
self.name, claims, access_token=access_token, refresh_token=refresh_token, org_id=str(claims.get("org_id") or ""),
)
self.name, claims, access_token=access_token, refresh_token=refresh_token, org_id=str(claims.get("org_id") or ""))
# ---- Plugin entry point ----
plugins/dashboard_auth/self_hosted/__init__.py
+23 -47
View File
@@ -19,15 +19,14 @@ from typing import Any, Dict, Optional
import httpx
from hermes_cli.dashboard_auth import (
DashboardAuthProvider, InvalidCodeError, LoginStart, ProviderError, RefreshExpiredError, Session)
from hermes_cli.dashboard_auth import LoginStart, ProviderError, Session
from plugins.dashboard_auth._shared import (
JSON_HEADERS,
TOKEN_ENDPOINT_TIMEOUT_SEC as _TOKEN_ENDPOINT_TIMEOUT_SEC,
JwtOAuthProvider,
SkipRegistration,
exchange_token,
load_config_section,
make_jwks_client,
parse_json_body,
pkce_login_start,
refresh_token_from,
@@ -65,7 +64,7 @@ def _require_https_or_loopback(url: str, *, field: str) -> str:
raise ProviderError(f"OIDC {field} must be https:// (or http on localhost), got {url!r}")
class SelfHostedOIDCProvider(DashboardAuthProvider):
class SelfHostedOIDCProvider(JwtOAuthProvider):
"""Generic self-hosted OpenID Connect provider (authorization-code + PKCE)."""
name = "self-hosted"
@@ -92,8 +91,6 @@ class SelfHostedOIDCProvider(DashboardAuthProvider):
self._discovery_lock = threading.Lock()
self._jwks_client: Any = None
# ---- public API (DashboardAuthProvider) -------------------------------
def start_login(self, *, redirect_uri: str) -> LoginStart:
# Validate the redirect before discovery so a bad redirect_uri surfaces even when the IDP is unreachable.
validate_redirect_uri(redirect_uri)
@@ -101,36 +98,6 @@ class SelfHostedOIDCProvider(DashboardAuthProvider):
return pkce_login_start(
disco["authorization_endpoint"], client_id=self._client_id, scope=self._scopes, redirect_uri=redirect_uri)
def complete_login(self, *, code: str, state: str, code_verifier: str, redirect_uri: str) -> Session:
# ``state`` is verified by the auth-route layer before this call.
return self._exchange(
{
"grant_type": "authorization_code", "code": code, "redirect_uri": redirect_uri,
"client_id": self._client_id, "code_verifier": code_verifier},
bad_request_exc=InvalidCodeError)
def refresh_session(self, *, refresh_token: str) -> Session:
if not refresh_token:
raise RefreshExpiredError("no refresh token present in session")
return self._exchange(
{
"grant_type": "refresh_token", "client_id": self._client_id, "refresh_token": refresh_token,
# Re-request the same scopes so the rotated ID token keeps its identity
# claims (some IDPs narrow scope on refresh otherwise).
"scope": self._scopes},
bad_request_exc=RefreshExpiredError,
previous_refresh_token=refresh_token)
def verify_session(self, *, access_token: str) -> Optional[Session]:
# The session cookie carries the ID token in the access-token slot (see _session)
# so this per-request check verifies a real JWT. None on expiry/invalidity;
# ProviderError if IDP/JWKS unreachable.
try:
claims = self._verify_id_token(access_token)
except InvalidCodeError:
return None
return self._session(access_token, "", claims)
def revoke_session(self, *, refresh_token: str) -> None:
# Best-effort RFC 7009 revocation when the IDP advertises an endpoint.
# Must never raise — logout is client-side cookie clearing regardless.
@@ -152,7 +119,7 @@ class SelfHostedOIDCProvider(DashboardAuthProvider):
logger.debug("self-hosted OIDC: revoke failed (ignored): %s", exc)
return None
# ---- internals: token exchange ----------------------------------------
# ---- JwtOAuthProvider hooks: token exchange ---------------------------
def _token_endpoint_auth(self, disco: Dict[str, Any]) -> tuple[Dict[str, str], Dict[str, str]]:
"""``(extra_data, extra_headers)`` for token-endpoint client auth. Public client →
@@ -169,7 +136,18 @@ class SelfHostedOIDCProvider(DashboardAuthProvider):
userpass = f"{urllib.parse.quote(self._client_id, safe='')}:{urllib.parse.quote(self._client_secret, safe='')}"
return {}, {"Authorization": f"Basic {base64.b64encode(userpass.encode('utf-8')).decode('ascii')}"}
def _exchange(self, data: Dict[str, str], *, bad_request_exc: type[Exception], previous_refresh_token: str = "") -> Session:
def _refresh_request(self, refresh_token: str) -> tuple[Dict[str, str], Optional[Dict[str, str]]]:
# Re-request the same scopes so the rotated ID token keeps its identity claims
# (some IDPs narrow scope on refresh otherwise).
return (
{"grant_type": "refresh_token", "client_id": self._client_id, "refresh_token": refresh_token,
"scope": self._scopes},
None)
def _grant(
self, data: Dict[str, str], *, bad_request_exc: type[Exception], headers: Optional[Dict[str, str]] = None,
previous_refresh_token: str = "",
) -> Session:
"""POST the discovered token endpoint and turn the response into a Session.
Confidential-client auth (body field or Basic header) is added for both grants —
the IDP rejects an unauthenticated refresh with ``invalid_client``."""
@@ -234,8 +212,7 @@ class SelfHostedOIDCProvider(DashboardAuthProvider):
if advertised_issuer and advertised_issuer.rstrip("/") != self._issuer:
raise ProviderError(
f"OIDC discovery issuer mismatch: document advertises {advertised_issuer!r} "
f"but configured issuer is {self._issuer!r}"
)
f"but configured issuer is {self._issuer!r}")
for key, url in endpoints.items():
_require_https_or_loopback(url, field=key)
# Absent/garbage auth-methods → [] → OIDC default (basic) applies.
@@ -247,12 +224,10 @@ class SelfHostedOIDCProvider(DashboardAuthProvider):
"token_endpoint_auth_methods_supported": (
[str(m) for m in auth_methods_raw] if isinstance(auth_methods_raw, list) else [])}
# ---- internals: JWT verification + mapping ----------------------------
# ---- JwtOAuthProvider hooks: verification + mapping -------------------
def _get_jwks_client(self) -> Any:
if self._jwks_client is None:
self._jwks_client = make_jwks_client(self._get_discovery()["jwks_uri"])
return self._jwks_client
def _jwks_uri(self) -> str:
return self._get_discovery()["jwks_uri"]
def _verify_id_token(self, id_token: str) -> Dict[str, Any]:
issuer = self._get_discovery()["issuer"]
@@ -260,6 +235,8 @@ class SelfHostedOIDCProvider(DashboardAuthProvider):
id_token, self._get_jwks_client(), algorithms=list(_ALLOWED_ID_TOKEN_ALGS),
audience=self._client_id, issuer=issuer, label="ID token")
_claims_for = _verify_id_token
def _session(self, id_token: str, refresh_token: str, claims: Dict[str, Any]) -> Session:
"""Map verified OIDC claims onto a Session. The verified ID token is stored in
``Session.access_token`` so the per-request ``verify_session`` re-verifies a real
@@ -315,5 +292,4 @@ def register(ctx) -> None:
if kw is not None:
logger.info(
"dashboard-auth-self-hosted: registered provider (issuer=%s, client_id=%s, scopes=%r, confidential=%s)",
kw["issuer"], kw["client_id"], kw["scopes"], bool(kw["client_secret"]), # never log the secret itself
)
kw["issuer"], kw["client_id"], kw["scopes"], bool(kw["client_secret"])) # never log the secret itself