mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
Teknium
5dc648c2af
Greptile on #141510: platform secrets (MATRIX_PASSWORD, DINGTALK_CLIENT_SECRET) passed the client-variable check, so it now uses is_registered_env_name like the rest of config. Uninstall deleted grants by manifest key while sign-in stores them by directory key; both use plugin_key now. A core resolver's own refresh POST (Codex) could raise httpx with the refresh-token request attached; that is rebuilt too. Logout no longer needs a valid oauth: entry, a numeric id is refused instead of accepted and unusable, and the host-domain warning keeps three labels for two-letter country suffixes (vendor.co.uk).
201 lines
9.1 KiB
Python
201 lines
9.1 KiB
Python
"""Signed-in provider requests that core makes FOR plugins; the token never reaches plugin code.
|
|
|
|
A plugin that needs the user's subscription sign-in (today: Codex / ChatGPT) must not read, refresh
|
|
or rewrite Hermes's auth store or ``~/.codex/auth.json``. Codex refresh tokens are single-use, so a
|
|
second refresher races Hermes's locked refresh and can sign the user out. Instead the plugin declares
|
|
``requires_auth: [openai-codex]`` in its ``plugin.yaml`` and calls :func:`credentialed_provider_request`.
|
|
Hermes resolves the credential through its own locked-refresh path, attaches ``Authorization``
|
|
itself, sends only to that provider's origins, never follows a redirect, and hands back
|
|
status, headers and body.
|
|
|
|
The caller is identified by file location: the nearest stack frame inside an installed plugin
|
|
directory. Like capabilities this is consent and visibility, not a sandbox. In-process plugin code
|
|
can still read any file the user can.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json as _json
|
|
import sys
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any, Callable, Mapping, Optional
|
|
|
|
from hermes_cli.urllib_security import url_origin
|
|
|
|
|
|
class ProviderNotSignedIn(RuntimeError):
|
|
"""The user has no usable sign-in for the provider in this Hermes profile."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ProviderResponse:
|
|
status: int
|
|
headers: dict[str, str]
|
|
body: bytes
|
|
|
|
@property
|
|
def text(self) -> str:
|
|
return self.body.decode("utf-8", "replace")
|
|
|
|
def json(self) -> Any:
|
|
return _json.loads(self.body)
|
|
|
|
|
|
def _codex_auth(url: str) -> tuple[dict[str, str], str]:
|
|
"""``(auth headers, credential base URL)`` from Hermes's own locked-refresh resolver."""
|
|
from agent.codex_headers import codex_account_headers
|
|
from agent.turn_failure_copy import oauth_relogin_command
|
|
from hermes_cli.auth_codex import resolve_codex_runtime_credentials
|
|
from hermes_cli.auth_constants import AuthError
|
|
|
|
try:
|
|
creds = resolve_codex_runtime_credentials()
|
|
except AuthError as exc:
|
|
if not exc.relogin_required:
|
|
raise
|
|
creds = {}
|
|
token = str(creds.get("api_key") or "").strip()
|
|
if not token:
|
|
raise ProviderNotSignedIn(
|
|
f"Not signed in to Codex in Hermes. Run `{oauth_relogin_command('openai-codex')}` "
|
|
"or sign in to OpenAI Codex under Providers in the Hermes desktop app.")
|
|
headers = codex_account_headers(token) if url_origin(url)[1] == "chatgpt.com" else {}
|
|
headers["Authorization"] = f"Bearer {token}"
|
|
return headers, str(creds.get("base_url") or "")
|
|
|
|
|
|
# provider -> (origins its token may be sent to, auth resolver). chatgpt.com is the Codex backend
|
|
# core itself calls; api.openai.com hosts the Realtime client-secret endpoint Codex clients use.
|
|
_PROVIDERS: dict[str, tuple[frozenset, Callable[[str], tuple[dict[str, str], str]]]] = {
|
|
"openai-codex": (frozenset({("https", "chatgpt.com", 443), ("https", "api.openai.com", 443)}), _codex_auth),
|
|
}
|
|
|
|
|
|
def _plugin_roots() -> set[Path]:
|
|
"""Every plugins directory an installed plugin can live under, as given and resolved."""
|
|
from hermes_cli.plugins import get_bundled_plugins_dir
|
|
from hermes_constants import get_default_hermes_root, get_hermes_home, get_process_hermes_home
|
|
|
|
bases = (get_hermes_home() / "plugins", get_process_hermes_home() / "plugins",
|
|
get_default_hermes_root() / "plugins", get_bundled_plugins_dir())
|
|
return {r for base in bases for r in (base.absolute(), base.resolve())}
|
|
|
|
|
|
def _calling_plugin_dir() -> Optional[Path]:
|
|
"""Directory of the plugin whose code is nearest on the calling stack, or None."""
|
|
roots = _plugin_roots()
|
|
frame = sys._getframe(2)
|
|
while frame is not None:
|
|
raw = Path(frame.f_code.co_filename)
|
|
for path in (raw.absolute(), raw.resolve()):
|
|
root = next((r for r in roots if path.is_relative_to(r) and path != r), None)
|
|
if root is None:
|
|
continue
|
|
for parent in path.parents:
|
|
if parent == root:
|
|
break
|
|
if (parent / "plugin.yaml").is_file() or (parent / "plugin.yml").is_file():
|
|
return parent
|
|
return root / path.relative_to(root).parts[0]
|
|
frame = frame.f_back
|
|
return None
|
|
|
|
|
|
def _detached(exc: BaseException) -> BaseException:
|
|
"""*exc* without traceback or chain: those frames hold the token, refresh token or client secret."""
|
|
exc.__cause__ = exc.__context__ = None
|
|
return exc.with_traceback(None)
|
|
|
|
|
|
def declared_auth_providers(manifest: Mapping[str, Any]) -> list[str]:
|
|
"""``requires_auth`` from a manifest dict: the providers whose sign-in the plugin uses."""
|
|
raw = (manifest or {}).get("requires_auth")
|
|
return [str(p).strip() for p in raw if str(p).strip()] if isinstance(raw, list) else []
|
|
|
|
|
|
def requires_auth_notice(manifest: Mapping[str, Any]) -> str:
|
|
"""One line for install/show surfaces; empty when the plugin declares no sign-in use."""
|
|
from hermes_cli.plugin_oauth import oauth_notice
|
|
|
|
providers = declared_auth_providers(manifest)
|
|
sign_in = (f"Uses your sign-in for: {', '.join(providers)} (Hermes sends the requests; "
|
|
"the plugin never receives the token)") if providers else ""
|
|
return "; ".join(part for part in (sign_in, oauth_notice(manifest)) if part)
|
|
|
|
|
|
def _declared_providers(plugin_dir: Path) -> list[str]:
|
|
from pm.plugin_declarations import native_manifest_file, read_native_manifest
|
|
|
|
manifest = native_manifest_file(plugin_dir)
|
|
return declared_auth_providers(read_native_manifest(manifest)) if manifest else []
|
|
|
|
|
|
def credentialed_provider_request(
|
|
provider: str, method: str, url: str, *, json: Any = None,
|
|
headers: Optional[Mapping[str, str]] = None, timeout: float = 30.0,
|
|
) -> ProviderResponse:
|
|
"""Send ``method url`` with the user's *provider* sign-in attached by Hermes.
|
|
|
|
*provider* is a Hermes sign-in (listed under ``requires_auth``) or an ``id`` the calling plugin
|
|
declares under ``oauth:`` (:mod:`hermes_cli.plugin_oauth`); the plugin's own entry wins.
|
|
Refuses (``PermissionError``) a caller outside an installed plugin, a plugin whose ``plugin.yaml``
|
|
does not list *provider* under ``requires_auth``, and any origin outside the provider's
|
|
allowlist; raises :class:`ProviderNotSignedIn` when this profile has no sign-in. Redirects are
|
|
returned as-is, never followed. The result carries no credential.
|
|
"""
|
|
from hermes_cli.plugin_oauth import _plugin_oauth_spec
|
|
|
|
plugin_dir = _calling_plugin_dir()
|
|
if plugin_dir is None:
|
|
raise PermissionError("credentialed_provider_request must be called from an installed plugin's code")
|
|
# The plugin's own oauth: entry first, so a core provider added later under the same id cannot take it over.
|
|
spec = _plugin_oauth_spec(plugin_dir, provider)
|
|
if spec is None:
|
|
spec = _PROVIDERS.get(provider)
|
|
if spec is None:
|
|
raise ValueError(f"No signed-in requests for provider {provider!r} (supported: {', '.join(_PROVIDERS)}, "
|
|
"or an id the plugin declares under oauth: in plugin.yaml)")
|
|
if provider not in _declared_providers(plugin_dir):
|
|
raise PermissionError(
|
|
f"Plugin '{plugin_dir.name}' must declare `requires_auth: [{provider}]` in plugin.yaml")
|
|
origins, resolve = spec
|
|
if url_origin(url) not in origins:
|
|
raise PermissionError(f"Refusing to send the {provider} sign-in to {url_origin(url)[1] or url!r}")
|
|
import httpx
|
|
|
|
# Every error the plugin receives past this point is rebuilt without traceback, chain or request: those
|
|
# reach the frames and httpx objects that hold the token, refresh token or client secret.
|
|
refusal: Optional[BaseException] = None
|
|
with httpx.Client(timeout=timeout, follow_redirects=False) as client:
|
|
request = client.build_request(method, url, json=json, headers=dict(headers or {})) # no credential yet
|
|
try:
|
|
auth_headers, credential_base = resolve(url)
|
|
except httpx.HTTPError as exc: # a resolver's own refresh POST failed: its request holds the refresh token
|
|
refusal = type(exc)(str(exc))
|
|
except (RuntimeError, OSError, ValueError, LookupError, TypeError) as exc:
|
|
refusal = _detached(exc)
|
|
else:
|
|
if url_origin(credential_base) not in origins:
|
|
# A pooled gateway key belongs to its own host only (#121486).
|
|
refusal = PermissionError(
|
|
f"This profile's {provider} credential routes to a custom endpoint; refusing")
|
|
else:
|
|
for name in auth_headers:
|
|
request.headers.pop(name, None)
|
|
request.headers.update(auth_headers)
|
|
try:
|
|
response = client.send(request)
|
|
except httpx.HTTPError as exc:
|
|
refusal = type(exc)(str(exc))
|
|
auth_headers = request = None
|
|
if refusal is not None:
|
|
raise refusal
|
|
return ProviderResponse(response.status_code, dict(response.headers), response.content)
|
|
|
|
|
|
__all__ = [
|
|
"ProviderNotSignedIn", "ProviderResponse", "credentialed_provider_request", "declared_auth_providers",
|
|
"requires_auth_notice",
|
|
]
|