Add native Spotify tools with PKCE auth

This commit is contained in:
Dilee 2026-04-24 14:17:44 +03:00 committed by Teknium
parent 3392d1e422
commit 7e9dd9ca45
9 changed files with 1936 additions and 3 deletions

View file

@ -33,8 +33,10 @@ import webbrowser
from contextlib import contextmanager
from dataclasses import dataclass, field
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path
from typing import Any, Dict, List, Optional
from urllib.parse import parse_qs, urlencode, urlparse
import httpx
import yaml
@ -81,6 +83,25 @@ CODEX_ACCESS_TOKEN_REFRESH_SKEW_SECONDS = 120
QWEN_OAUTH_CLIENT_ID = "f0304373b74a44d2b584a3fb70ca9e56"
QWEN_OAUTH_TOKEN_URL = "https://chat.qwen.ai/api/v1/oauth2/token"
QWEN_ACCESS_TOKEN_REFRESH_SKEW_SECONDS = 120
DEFAULT_SPOTIFY_ACCOUNTS_BASE_URL = "https://accounts.spotify.com"
DEFAULT_SPOTIFY_API_BASE_URL = "https://api.spotify.com/v1"
DEFAULT_SPOTIFY_REDIRECT_URI = "http://127.0.0.1:43827/spotify/callback"
SPOTIFY_ACCESS_TOKEN_REFRESH_SKEW_SECONDS = 120
DEFAULT_SPOTIFY_SCOPE = " ".join((
"user-modify-playback-state",
"user-read-playback-state",
"user-read-currently-playing",
"user-read-recently-played",
"playlist-read-private",
"playlist-read-collaborative",
"playlist-modify-public",
"playlist-modify-private",
"user-library-read",
"user-library-modify",
))
SERVICE_PROVIDER_NAMES: Dict[str, str] = {
"spotify": "Spotify",
}
# Google Gemini OAuth (google-gemini-cli provider, Cloud Code Assist backend)
DEFAULT_GEMINI_CLOUDCODE_BASE_URL = "cloudcode-pa://google"
@ -795,6 +816,34 @@ def _save_provider_state(auth_store: Dict[str, Any], provider_id: str, state: Di
auth_store["active_provider"] = provider_id
def _store_provider_state(
auth_store: Dict[str, Any],
provider_id: str,
state: Dict[str, Any],
*,
set_active: bool = True,
) -> None:
providers = auth_store.setdefault("providers", {})
if not isinstance(providers, dict):
auth_store["providers"] = {}
providers = auth_store["providers"]
providers[provider_id] = state
if set_active:
auth_store["active_provider"] = provider_id
def is_known_auth_provider(provider_id: str) -> bool:
normalized = (provider_id or "").strip().lower()
return normalized in PROVIDER_REGISTRY or normalized in SERVICE_PROVIDER_NAMES
def get_auth_provider_display_name(provider_id: str) -> str:
normalized = (provider_id or "").strip().lower()
if normalized in PROVIDER_REGISTRY:
return PROVIDER_REGISTRY[normalized].name
return SERVICE_PROVIDER_NAMES.get(normalized, provider_id)
def read_credential_pool(provider_id: Optional[str] = None) -> Dict[str, Any]:
"""Return the persisted credential pool, or one provider slice."""
auth_store = _load_auth_store()
@ -1429,8 +1478,520 @@ def get_gemini_oauth_auth_status() -> Dict[str, Any]:
"email": creds.email,
"project_id": creds.project_id,
}
# Spotify auth — PKCE tokens stored in ~/.hermes/auth.json
# =============================================================================
def _spotify_scope_list(raw_scope: Optional[str] = None) -> List[str]:
scope_text = (raw_scope or DEFAULT_SPOTIFY_SCOPE).strip()
scopes = [part for part in scope_text.split() if part]
seen: set[str] = set()
ordered: List[str] = []
for scope in scopes:
if scope not in seen:
seen.add(scope)
ordered.append(scope)
return ordered
def _spotify_scope_string(raw_scope: Optional[str] = None) -> str:
return " ".join(_spotify_scope_list(raw_scope))
def _spotify_client_id(
explicit: Optional[str] = None,
state: Optional[Dict[str, Any]] = None,
) -> str:
from hermes_cli.config import get_env_value
candidates = (
explicit,
get_env_value("HERMES_SPOTIFY_CLIENT_ID"),
get_env_value("SPOTIFY_CLIENT_ID"),
state.get("client_id") if isinstance(state, dict) else None,
)
for candidate in candidates:
cleaned = str(candidate or "").strip()
if cleaned:
return cleaned
raise AuthError(
"Spotify client_id is required. Set HERMES_SPOTIFY_CLIENT_ID or pass --client-id.",
provider="spotify",
code="spotify_client_id_missing",
)
def _spotify_redirect_uri(
explicit: Optional[str] = None,
state: Optional[Dict[str, Any]] = None,
) -> str:
from hermes_cli.config import get_env_value
candidates = (
explicit,
get_env_value("HERMES_SPOTIFY_REDIRECT_URI"),
get_env_value("SPOTIFY_REDIRECT_URI"),
state.get("redirect_uri") if isinstance(state, dict) else None,
DEFAULT_SPOTIFY_REDIRECT_URI,
)
for candidate in candidates:
cleaned = str(candidate or "").strip()
if cleaned:
return cleaned
return DEFAULT_SPOTIFY_REDIRECT_URI
def _spotify_api_base_url(state: Optional[Dict[str, Any]] = None) -> str:
from hermes_cli.config import get_env_value
candidates = (
get_env_value("HERMES_SPOTIFY_API_BASE_URL"),
state.get("api_base_url") if isinstance(state, dict) else None,
DEFAULT_SPOTIFY_API_BASE_URL,
)
for candidate in candidates:
cleaned = str(candidate or "").strip().rstrip("/")
if cleaned:
return cleaned
return DEFAULT_SPOTIFY_API_BASE_URL
def _spotify_accounts_base_url(state: Optional[Dict[str, Any]] = None) -> str:
from hermes_cli.config import get_env_value
candidates = (
get_env_value("HERMES_SPOTIFY_ACCOUNTS_BASE_URL"),
state.get("accounts_base_url") if isinstance(state, dict) else None,
DEFAULT_SPOTIFY_ACCOUNTS_BASE_URL,
)
for candidate in candidates:
cleaned = str(candidate or "").strip().rstrip("/")
if cleaned:
return cleaned
return DEFAULT_SPOTIFY_ACCOUNTS_BASE_URL
def _spotify_code_verifier(length: int = 64) -> str:
raw = base64.urlsafe_b64encode(os.urandom(length)).decode("ascii")
return raw.rstrip("=")[:128]
def _spotify_code_challenge(code_verifier: str) -> str:
digest = hashlib.sha256(code_verifier.encode("utf-8")).digest()
return base64.urlsafe_b64encode(digest).decode("ascii").rstrip("=")
def _spotify_build_authorize_url(
*,
client_id: str,
redirect_uri: str,
scope: str,
state: str,
code_challenge: str,
accounts_base_url: str,
) -> str:
query = urlencode({
"client_id": client_id,
"response_type": "code",
"redirect_uri": redirect_uri,
"scope": scope,
"state": state,
"code_challenge_method": "S256",
"code_challenge": code_challenge,
})
return f"{accounts_base_url}/authorize?{query}"
def _spotify_validate_redirect_uri(redirect_uri: str) -> tuple[str, int, str]:
parsed = urlparse(redirect_uri)
if parsed.scheme != "http":
raise AuthError(
"Spotify PKCE redirect_uri must use http://localhost or http://127.0.0.1.",
provider="spotify",
code="spotify_redirect_invalid",
)
host = parsed.hostname or ""
if host not in {"127.0.0.1", "localhost"}:
raise AuthError(
"Spotify PKCE redirect_uri must point to localhost or 127.0.0.1.",
provider="spotify",
code="spotify_redirect_invalid",
)
if not parsed.port:
raise AuthError(
"Spotify PKCE redirect_uri must include an explicit localhost port.",
provider="spotify",
code="spotify_redirect_invalid",
)
return host, parsed.port, parsed.path or "/"
def _make_spotify_callback_handler(expected_path: str) -> tuple[type[BaseHTTPRequestHandler], dict[str, Any]]:
result: dict[str, Any] = {
"code": None,
"state": None,
"error": None,
"error_description": None,
}
class _SpotifyCallbackHandler(BaseHTTPRequestHandler):
def do_GET(self) -> None: # noqa: N802
parsed = urlparse(self.path)
if parsed.path != expected_path:
self.send_response(404)
self.end_headers()
self.wfile.write(b"Not found.")
return
params = parse_qs(parsed.query)
result["code"] = params.get("code", [None])[0]
result["state"] = params.get("state", [None])[0]
result["error"] = params.get("error", [None])[0]
result["error_description"] = params.get("error_description", [None])[0]
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.end_headers()
if result["error"]:
body = "<html><body><h1>Spotify authorization failed.</h1>You can close this tab.</body></html>"
else:
body = "<html><body><h1>Spotify authorization received.</h1>You can close this tab.</body></html>"
self.wfile.write(body.encode("utf-8"))
def log_message(self, format: str, *args: Any) -> None: # noqa: A003
return
return _SpotifyCallbackHandler, result
def _spotify_wait_for_callback(
redirect_uri: str,
*,
timeout_seconds: float = 180.0,
) -> dict[str, Any]:
host, port, path = _spotify_validate_redirect_uri(redirect_uri)
handler_cls, result = _make_spotify_callback_handler(path)
class _ReuseHTTPServer(HTTPServer):
allow_reuse_address = True
try:
server = _ReuseHTTPServer((host, port), handler_cls)
except OSError as exc:
raise AuthError(
f"Could not bind Spotify callback server on {host}:{port}: {exc}",
provider="spotify",
code="spotify_callback_bind_failed",
) from exc
thread = threading.Thread(target=server.serve_forever, kwargs={"poll_interval": 0.1}, daemon=True)
thread.start()
deadline = time.time() + max(5.0, timeout_seconds)
try:
while time.time() < deadline:
if result["code"] or result["error"]:
return result
time.sleep(0.1)
finally:
server.shutdown()
server.server_close()
thread.join(timeout=1.0)
raise AuthError(
"Spotify authorization timed out waiting for the local callback.",
provider="spotify",
code="spotify_callback_timeout",
)
def _spotify_token_payload_to_state(
token_payload: Dict[str, Any],
*,
client_id: str,
redirect_uri: str,
requested_scope: str,
accounts_base_url: str,
api_base_url: str,
previous_state: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
now = datetime.now(timezone.utc)
expires_in = _coerce_ttl_seconds(token_payload.get("expires_in", 0))
expires_at = datetime.fromtimestamp(now.timestamp() + expires_in, tz=timezone.utc)
state = dict(previous_state or {})
state.update({
"client_id": client_id,
"redirect_uri": redirect_uri,
"accounts_base_url": accounts_base_url,
"api_base_url": api_base_url,
"scope": requested_scope,
"granted_scope": str(token_payload.get("scope") or requested_scope).strip(),
"token_type": str(token_payload.get("token_type", "Bearer") or "Bearer").strip() or "Bearer",
"access_token": str(token_payload.get("access_token", "") or "").strip(),
"refresh_token": str(
token_payload.get("refresh_token")
or state.get("refresh_token")
or ""
).strip(),
"obtained_at": now.isoformat(),
"expires_at": expires_at.isoformat(),
"expires_in": expires_in,
"auth_type": "oauth_pkce",
})
return state
def _spotify_exchange_code_for_tokens(
*,
client_id: str,
code: str,
redirect_uri: str,
code_verifier: str,
accounts_base_url: str,
timeout_seconds: float = 20.0,
) -> Dict[str, Any]:
try:
response = httpx.post(
f"{accounts_base_url}/api/token",
headers={"Content-Type": "application/x-www-form-urlencoded"},
data={
"client_id": client_id,
"grant_type": "authorization_code",
"code": code,
"redirect_uri": redirect_uri,
"code_verifier": code_verifier,
},
timeout=timeout_seconds,
)
except Exception as exc:
raise AuthError(
f"Spotify token exchange failed: {exc}",
provider="spotify",
code="spotify_token_exchange_failed",
) from exc
if response.status_code >= 400:
detail = response.text.strip()
raise AuthError(
"Spotify token exchange failed."
+ (f" Response: {detail}" if detail else ""),
provider="spotify",
code="spotify_token_exchange_failed",
)
payload = response.json()
if not isinstance(payload, dict) or not str(payload.get("access_token", "") or "").strip():
raise AuthError(
"Spotify token response did not include an access_token.",
provider="spotify",
code="spotify_token_exchange_invalid",
)
return payload
def _refresh_spotify_oauth_state(
state: Dict[str, Any],
*,
timeout_seconds: float = 20.0,
) -> Dict[str, Any]:
refresh_token = str(state.get("refresh_token", "") or "").strip()
if not refresh_token:
raise AuthError(
"Spotify refresh token missing. Run `hermes auth spotify` again.",
provider="spotify",
code="spotify_refresh_token_missing",
relogin_required=True,
)
client_id = _spotify_client_id(state=state)
accounts_base_url = _spotify_accounts_base_url(state)
try:
response = httpx.post(
f"{accounts_base_url}/api/token",
headers={"Content-Type": "application/x-www-form-urlencoded"},
data={
"grant_type": "refresh_token",
"refresh_token": refresh_token,
"client_id": client_id,
},
timeout=timeout_seconds,
)
except Exception as exc:
raise AuthError(
f"Spotify token refresh failed: {exc}",
provider="spotify",
code="spotify_refresh_failed",
) from exc
if response.status_code >= 400:
detail = response.text.strip()
raise AuthError(
"Spotify token refresh failed. Run `hermes auth spotify` again."
+ (f" Response: {detail}" if detail else ""),
provider="spotify",
code="spotify_refresh_failed",
relogin_required=True,
)
payload = response.json()
if not isinstance(payload, dict) or not str(payload.get("access_token", "") or "").strip():
raise AuthError(
"Spotify refresh response did not include an access_token.",
provider="spotify",
code="spotify_refresh_invalid",
relogin_required=True,
)
return _spotify_token_payload_to_state(
payload,
client_id=client_id,
redirect_uri=_spotify_redirect_uri(state=state),
requested_scope=str(state.get("scope") or DEFAULT_SPOTIFY_SCOPE),
accounts_base_url=accounts_base_url,
api_base_url=_spotify_api_base_url(state),
previous_state=state,
)
def resolve_spotify_runtime_credentials(
*,
force_refresh: bool = False,
refresh_if_expiring: bool = True,
refresh_skew_seconds: int = SPOTIFY_ACCESS_TOKEN_REFRESH_SKEW_SECONDS,
) -> Dict[str, Any]:
with _auth_store_lock():
auth_store = _load_auth_store()
state = _load_provider_state(auth_store, "spotify")
if not state:
raise AuthError(
"Spotify is not authenticated. Run `hermes auth spotify` first.",
provider="spotify",
code="spotify_auth_missing",
relogin_required=True,
)
should_refresh = bool(force_refresh)
if not should_refresh and refresh_if_expiring:
should_refresh = _is_expiring(state.get("expires_at"), refresh_skew_seconds)
if should_refresh:
state = _refresh_spotify_oauth_state(state)
_store_provider_state(auth_store, "spotify", state, set_active=False)
_save_auth_store(auth_store)
access_token = str(state.get("access_token", "") or "").strip()
if not access_token:
raise AuthError(
"Spotify access token missing. Run `hermes auth spotify` again.",
provider="spotify",
code="spotify_access_token_missing",
relogin_required=True,
)
return {
"provider": "spotify",
"access_token": access_token,
"api_key": access_token,
"token_type": str(state.get("token_type", "Bearer") or "Bearer"),
"base_url": _spotify_api_base_url(state),
"scope": str(state.get("granted_scope") or state.get("scope") or "").strip(),
"client_id": _spotify_client_id(state=state),
"redirect_uri": _spotify_redirect_uri(state=state),
"expires_at": state.get("expires_at"),
"refresh_token": str(state.get("refresh_token", "") or "").strip(),
}
def get_spotify_auth_status() -> Dict[str, Any]:
state = get_provider_auth_state("spotify")
if not state:
return {"logged_in": False}
expires_at = state.get("expires_at")
refresh_token = str(state.get("refresh_token", "") or "").strip()
return {
"logged_in": bool(refresh_token or not _is_expiring(expires_at, 0)),
"auth_type": state.get("auth_type", "oauth_pkce"),
"client_id": state.get("client_id"),
"redirect_uri": state.get("redirect_uri"),
"scope": state.get("granted_scope") or state.get("scope"),
"expires_at": expires_at,
"api_base_url": state.get("api_base_url"),
"has_refresh_token": bool(refresh_token),
}
def login_spotify_command(args) -> None:
existing_state = get_provider_auth_state("spotify") or {}
client_id = _spotify_client_id(getattr(args, "client_id", None), existing_state)
redirect_uri = _spotify_redirect_uri(getattr(args, "redirect_uri", None), existing_state)
scope = _spotify_scope_string(getattr(args, "scope", None) or existing_state.get("scope"))
accounts_base_url = _spotify_accounts_base_url(existing_state)
api_base_url = _spotify_api_base_url(existing_state)
open_browser = not getattr(args, "no_browser", False)
code_verifier = _spotify_code_verifier()
code_challenge = _spotify_code_challenge(code_verifier)
state_nonce = uuid.uuid4().hex
authorize_url = _spotify_build_authorize_url(
client_id=client_id,
redirect_uri=redirect_uri,
scope=scope,
state=state_nonce,
code_challenge=code_challenge,
accounts_base_url=accounts_base_url,
)
print("Starting Spotify PKCE login...")
print(f"Client ID: {client_id}")
print(f"Redirect URI: {redirect_uri}")
print("Make sure this redirect URI is allow-listed in your Spotify app settings.")
print()
print("Open this URL to authorize Hermes:")
print(authorize_url)
print()
if open_browser and not _is_remote_session():
try:
opened = webbrowser.open(authorize_url)
except Exception:
opened = False
if opened:
print("Browser opened for Spotify authorization.")
else:
print("Could not open the browser automatically; use the URL above.")
callback = _spotify_wait_for_callback(
redirect_uri,
timeout_seconds=float(getattr(args, "timeout", None) or 180.0),
)
if callback.get("error"):
detail = callback.get("error_description") or callback["error"]
raise SystemExit(f"Spotify authorization failed: {detail}")
if callback.get("state") != state_nonce:
raise SystemExit("Spotify authorization failed: state mismatch.")
token_payload = _spotify_exchange_code_for_tokens(
client_id=client_id,
code=str(callback.get("code") or ""),
redirect_uri=redirect_uri,
code_verifier=code_verifier,
accounts_base_url=accounts_base_url,
timeout_seconds=float(getattr(args, "timeout", None) or 20.0),
)
spotify_state = _spotify_token_payload_to_state(
token_payload,
client_id=client_id,
redirect_uri=redirect_uri,
requested_scope=scope,
accounts_base_url=accounts_base_url,
api_base_url=api_base_url,
)
with _auth_store_lock():
auth_store = _load_auth_store()
_store_provider_state(auth_store, "spotify", spotify_state, set_active=False)
saved_to = _save_auth_store(auth_store)
print("Spotify login successful!")
print(f" Auth state: {saved_to}")
print(" Provider state saved under providers.spotify")
# =============================================================================
# SSH / remote session detection
@ -2744,6 +3305,8 @@ def get_external_process_provider_status(provider_id: str) -> Dict[str, Any]:
def get_auth_status(provider_id: Optional[str] = None) -> Dict[str, Any]:
"""Generic auth status dispatcher."""
target = provider_id or get_active_provider()
if target == "spotify":
return get_spotify_auth_status()
if target == "nous":
return get_nous_auth_status()
if target == "openai-codex":
@ -3658,7 +4221,7 @@ def logout_command(args) -> None:
"""Clear auth state for a provider."""
provider_id = getattr(args, "provider", None)
if provider_id and provider_id not in PROVIDER_REGISTRY:
if provider_id and not is_known_auth_provider(provider_id):
print(f"Unknown provider: {provider_id}")
raise SystemExit(1)
@ -3669,8 +4232,8 @@ def logout_command(args) -> None:
print("No provider is currently logged in.")
return
provider_name = PROVIDER_REGISTRY[target].name if target in PROVIDER_REGISTRY else target
config_matches = _config_provider_matches(target)
provider_name = get_auth_provider_display_name(target)
if clear_provider_auth(target) or config_matches:
_reset_config_provider()

View file

@ -408,6 +408,44 @@ def auth_reset_command(args) -> None:
print(f"Reset status on {count} {provider} credentials")
def auth_status_command(args) -> None:
provider = _normalize_provider(getattr(args, "provider", "") or "")
if not provider:
raise SystemExit("Provider is required. Example: `hermes auth status spotify`.")
status = auth_mod.get_auth_status(provider)
if not status.get("logged_in"):
reason = status.get("error")
if reason:
print(f"{provider}: logged out ({reason})")
else:
print(f"{provider}: logged out")
return
print(f"{provider}: logged in")
for key in ("auth_type", "client_id", "redirect_uri", "scope", "expires_at", "api_base_url"):
value = status.get(key)
if value:
print(f" {key}: {value}")
def auth_logout_command(args) -> None:
auth_mod.logout_command(SimpleNamespace(provider=getattr(args, "provider", None)))
def auth_spotify_command(args) -> None:
action = str(getattr(args, "spotify_action", "") or "login").strip().lower()
if action in {"", "login"}:
auth_mod.login_spotify_command(args)
return
if action == "status":
auth_status_command(SimpleNamespace(provider="spotify"))
return
if action == "logout":
auth_logout_command(SimpleNamespace(provider="spotify"))
return
raise SystemExit(f"Unknown Spotify auth action: {action}")
def _interactive_auth() -> None:
"""Interactive credential pool management when `hermes auth` is called bare."""
# Show current pool status first
@ -605,5 +643,14 @@ def auth_command(args) -> None:
if action == "reset":
auth_reset_command(args)
return
if action == "status":
auth_status_command(args)
return
if action == "logout":
auth_logout_command(args)
return
if action == "spotify":
auth_spotify_command(args)
return
# No subcommand — launch interactive mode
_interactive_auth()

View file

@ -7316,7 +7316,7 @@ For more help on a command:
)
logout_parser.add_argument(
"--provider",
choices=["nous", "openai-codex"],
choices=["nous", "openai-codex", "spotify"],
default=None,
help="Provider to log out from (default: active provider)",
)
@ -7373,6 +7373,17 @@ For more help on a command:
"reset", help="Clear exhaustion status for all credentials for a provider"
)
auth_reset.add_argument("provider", help="Provider id")
auth_status = auth_subparsers.add_parser("status", help="Show auth status for a provider")
auth_status.add_argument("provider", help="Provider id")
auth_logout = auth_subparsers.add_parser("logout", help="Log out a provider and clear stored auth state")
auth_logout.add_argument("provider", help="Provider id")
auth_spotify = auth_subparsers.add_parser("spotify", help="Authenticate Hermes with Spotify via PKCE")
auth_spotify.add_argument("spotify_action", nargs="?", choices=["login", "status", "logout"], default="login")
auth_spotify.add_argument("--client-id", help="Spotify app client_id (or set HERMES_SPOTIFY_CLIENT_ID)")
auth_spotify.add_argument("--redirect-uri", help="Allow-listed localhost redirect URI for your Spotify app")
auth_spotify.add_argument("--scope", help="Override requested Spotify scopes")
auth_spotify.add_argument("--no-browser", action="store_true", help="Do not attempt to open the browser automatically")
auth_spotify.add_argument("--timeout", type=float, help="Callback/token exchange timeout in seconds")
auth_parser.set_defaults(func=cmd_auth)
# =========================================================================