hanzoai/python-sdk
2
0
Fork 0
mirror of https://github.com/hanzoai/python-sdk.git synced 2026-10-23 20:23:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

auth: verify tokens instead of asserting they exist

Browse Source 
is_authenticated() was `load_token() is not None` — a string-presence check
that returned True for the literal "fake.not.a.real.jwt". Every permission
decision downstream of it was therefore a lie. This makes the client side of
Hanzo auth actually work, end to end. Builds on the /v1/iam prefix already on
main; the endpoint paths were necessary but not sufficient.

hanzo_iam.tokens — the ONE credential judge. Verifies the signature against
the issuer's published JWKS plus exp/iss/aud, and fails CLOSED: an unreachable
JWKS is not a pass, because a client that cannot check a signature does not
know the token is good. Reason codes distinguish "expired" from "offline".
`alg: none` is absent from the accepted list by construction. An opaque API
key cannot be judged offline, so it is confirmed against userinfo rather than
assumed valid for being a non-empty string.

hanzo_iam.oauth — a real login: authorization code + PKCE (S256) over a
loopback redirect, with a deadline. Not the device grant, which would be the
better CLI UX: iam implements RFC 8628 fully and advertises it, but no PUBLIC
client is registered, so POST /v1/iam/oauth/device answers 401 invalid_client
for hanzo-cli, hanzo-app, hanzo-cloud and hanzo-console alike. Registering one
public app is the whole server-side fix; shipping a device path that always
401s would be a lie in code. The password grant is out for the same reason
(401 invalid_client without a secret), so `--no-browser` now prints the URL
instead of prompting for a password it cannot use.

The loopback listener binds a REGISTERED redirect_uri. iam compares
redirect_uri by exact string and does not apply RFC 8252 §7.3 port-agnostic
loopback matching, so a CLI cannot pick a free ephemeral port — the previous
code bound 8399 (cli) and 8398 (bot), neither registered, and /authorize
refused both with a bare 400 before the user saw a login page. It listens on
127.0.0.1 and ::1 because the registered URIs spell "localhost", and sets
SO_REUSEADDR because the previous callback sits in TIME_WAIT for ~60s and
would otherwise EADDRINUSE a second login on a port nothing is using.

hanzo_iam.store — keyring first, else an atomic 0600 file. The old path did
write_text() then chmod(0600), so with umask 022 a bearer token sat at 0644
between the two calls. Now it is created private and renamed into place. The
PaaS session cache goes through the same writer.

Two definitions of IAMConfig existed; models.py's was the one every client
imported, so the endpoint properties main had just fixed on config.py's were
dead code. One now, in config.py, plus jwks_uri and device_endpoint.

The bot and CLI login copies are gone; both call the one flow. whoami and
`hanzo bot login` no longer print claims from an unverified decode — that
rendered an attacker-chosen identity as fact.

The Team tool refuses instead of pretending: api.hanzo.ai/team is the
marketing SPA (200 text/html), so raise_for_status() passed and .json() blew
up. There is no Team API; HANZO_TEAM_URL stays as the seam for when one ships.

PaaS token exchange reports its real diagnosis: POST /v1/auth/login 404s while
/v1/org and /v1/user 401, so the API is up and gated but the exchange route is
not deployed at that edge. No client-side workaround exists, so it says so.

Tests fail against the old behaviour, which is the point: 5 session tests fail
against the shipped is_authenticated and 4 store tests fail against
write-then-chmod. 134 pass alongside main's KMS suite.

Co-authored-by: Hanzo Dev <[email protected]>
This commit is contained in:
zeekay
2026-07-27 18:55:24 -07:00
co-authored by Hanzo Dev
parent 8f3d611abf
commit d08a7d5599
24 changed files with 2396 additions and 780 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/hanzo-cli/hanzo_cli/auth.py
+68 -283
View File
@@ -1,61 +1,45 @@
"""Authentication and token management for Hanzo CLI.
This module is a thin adapter over `hanzo_iam` — the ONE implementation of the
login flow (`hanzo_iam.oauth`), the token store (`hanzo_iam.store`) and token
verification (`hanzo_iam.tokens`). It used to carry a second, divergent copy of
all three; the copy posted to the unprefixed /oauth/token (a 200 HTML page on
hanzo.id), sent no PKCE, and waited for its callback in an unbounded loop.
Credential chain:
1. IAM_CLIENT_ID + IAM_CLIENT_SECRET env vars
2. ~/.hanzo/auth/token.json from `hanzo login`
3. Exit with help message
1. IAM_CLIENT_ID + IAM_CLIENT_SECRET env vars (machine-to-machine)
2. The stored token from `hanzo login`
3. Exit with help
"""
from __future__ import annotations
import base64
import hashlib
import http.server
import json
import os
import secrets
import sys
import time
import webbrowser
from pathlib import Path
from typing import Any
from urllib.parse import parse_qs, urlparse
import click
from hanzo_iam import IAMClient, IAMConfig
from hanzo_iam import IAMClient, IAMConfig, oauth, store, tokens
from hanzo_iam.models import OIDC_JWKS_PATH
TOKEN_DIR = Path.home() / ".hanzo" / "auth"
TOKEN_FILE = TOKEN_DIR / "token.json"
DEFAULT_IAM_URL = "https://hanzo.id"
DEFAULT_ORG = "hanzo"
DEFAULT_IAM_URL = oauth.DEFAULT_IAM_URL
DEFAULT_ORG = oauth.DEFAULT_ORG
DEFAULT_APP = "hanzo-app"
DEFAULT_CLIENT_ID = "hanzo-app"
CALLBACK_PORT = 8399
CALLBACK_PATH = "/callback"
DEFAULT_CLIENT_ID = oauth.DEFAULT_CLIENT_ID
TOKEN_FILE = store.TOKEN_FILE
def _save_token(data: dict[str, Any]) -> None:
"""Save token data to disk."""
TOKEN_DIR.mkdir(parents=True, exist_ok=True)
TOKEN_FILE.write_text(json.dumps(data, indent=2))
TOKEN_FILE.chmod(0o600)
def _save_token(data: dict[str, Any]) -> str:
return store.save(data)
def _load_token() -> dict[str, Any] | None:
"""Load stored token from disk."""
if not TOKEN_FILE.exists():
return None
try:
return json.loads(TOKEN_FILE.read_text())
except (json.JSONDecodeError, OSError):
return None
return store.load()
def _clear_token() -> None:
"""Remove stored token."""
if TOKEN_FILE.exists():
TOKEN_FILE.unlink()
store.clear()
def _env(name: str) -> str:
@@ -127,263 +111,64 @@ def get_token_info() -> dict[str, Any] | None:
return _load_token()
# =========================================================================
# Browser OAuth Login Flow
# =========================================================================
class _OAuthCallbackHandler(http.server.BaseHTTPRequestHandler):
"""HTTP handler that captures the OAuth callback code."""
code: str | None = None
state: str | None = None
error: str | None = None
def do_GET(self) -> None: # noqa: N802
parsed = urlparse(self.path)
if parsed.path != CALLBACK_PATH:
self.send_response(404)
self.end_headers()
return
qs = parse_qs(parsed.query)
if "error" in qs:
_OAuthCallbackHandler.error = qs["error"][0]
self._respond("Login failed. You can close this window.", success=False)
return
_OAuthCallbackHandler.code = qs.get("code", [None])[0]
_OAuthCallbackHandler.state = qs.get("state", [None])[0]
self._respond("Login successful! You can close this window.")
def _respond(self, message: str, success: bool = True) -> None:
self.send_response(200)
self.send_header("Content-Type", "text/html")
self.end_headers()
accent = "#6C5CE7" if success else "#e74c3c"
body = f"""<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Hanzo</title>
<style>
*{{margin:0;padding:0;box-sizing:border-box}}
body{{
font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',Roboto,Helvetica,Arial,sans-serif;
background:#0a0a0a;color:#fafafa;
display:flex;align-items:center;justify-content:center;
min-height:100vh;
}}
.card{{
text-align:center;max-width:420px;padding:48px 32px;
background:#111;border:1px solid #222;border-radius:16px;
}}
.logo{{font-size:28px;font-weight:700;letter-spacing:-.5px;margin-bottom:32px;color:#fff}}
.logo span{{color:{accent}}}
.icon{{font-size:48px;margin-bottom:20px}}
h2{{font-size:20px;font-weight:600;margin-bottom:8px;color:#fff}}
.sub{{color:#888;font-size:14px;line-height:1.5;margin-bottom:28px}}
.tagline{{
font-size:13px;color:#555;border-top:1px solid #222;
padding-top:20px;margin-top:8px;
}}
</style>
</head>
<body>
<div class="card">
<div class="logo"><span>&#x25B2;</span> hanzo</div>
<div class="icon">{"&#x2705;" if success else "&#x274C;"}</div>
<h2>{message}</h2>
<p class="sub">You can close this window and return to your terminal.</p>
<p class="tagline">Build something you love.</p>
</div>
</body>
</html>"""
self.wfile.write(body.encode())
def log_message(self, format: str, *args: Any) -> None:
"""Suppress default request logging."""
def verify_token_data(data: dict[str, Any] | None) -> tokens.Verification:
"""Judge a token dict — stored or freshly issued. The ONE verification seam
for this CLI, so `whoami`, `login` and every command agree on what valid
means."""
if not data or not data.get("access_token"):
return tokens.Verification(False, tokens.NO_CREDENTIAL, "no credential found")
server_url = (data.get("server_url") or _iam_url()).rstrip("/")
return tokens.verify(
data["access_token"],
jwks_uri=f"{server_url}{OIDC_JWKS_PATH}",
issuer=server_url,
)
def browser_login(port: int = CALLBACK_PORT) -> dict[str, Any]:
"""Run the browser OAuth login flow.
def verify_token() -> tokens.Verification:
"""Judge the stored credential."""
return verify_token_data(_load_token())
Opens the user's browser to the IAM login page, starts a local HTTP
server to receive the callback, exchanges the code for tokens.
Returns:
Token data dict with access_token, etc.
def browser_login(port: int | None = None, open_browser: bool = True) -> dict[str, Any]:
"""Sign in through the browser (authorization code + PKCE) and store it.
`port` selects among the redirect URIs registered on the IAM application.
It cannot be an arbitrary port: iam compares redirect_uri by exact string,
so an unregistered one is refused with a bare 400 before the user ever sees
a login page.
"""
server_url = _iam_url()
org = _iam_org()
app = _iam_app()
client_id = _iam_client_id()
uris = oauth.LOOPBACK_REDIRECTS
if port is not None:
wanted = [u for u in uris if f":{port}/" in u]
if not wanted:
raise click.ClickException(
f"Port {port} is not a registered redirect URI. Registered: "
+ ", ".join(uris)
)
uris = tuple(wanted)
config = IAMConfig(
server_url=server_url,
client_id=client_id,
client_secret="",
organization=org,
application=app,
token_data = oauth.login(
server_url=_iam_url(),
client_id=_iam_client_id(),
organization=_iam_org(),
redirect_uris=uris,
open_browser=open_browser,
on_url=lambda u: click.echo(f"Open this URL to sign in:\n\n {u}\n"),
)
client = IAMClient(config=config)
token_data["application"] = _iam_app()
redirect_uri = f"http://localhost:{port}{CALLBACK_PATH}"
state = secrets.token_urlsafe(32)
# PKCE is not optional here. This is a native app with a public client
# (client_secret=""), so the authorization code is the ONLY secret in the
# flow and it arrives over a plaintext loopback redirect that any other
# local process can race. RFC 8252 §8.1 requires PKCE for exactly this
# shape; without a verifier an intercepted code is directly redeemable.
code_verifier = secrets.token_urlsafe(64)[:128]
code_challenge = (
base64.urlsafe_b64encode(hashlib.sha256(code_verifier.encode()).digest())
.decode()
.rstrip("=")
)
auth_url = client.get_authorization_url(
redirect_uri=redirect_uri,
state=state,
scope="openid profile email",
code_challenge=code_challenge,
code_challenge_method="S256",
)
# Reset handler state
_OAuthCallbackHandler.code = None
_OAuthCallbackHandler.state = None
_OAuthCallbackHandler.error = None
server = http.server.HTTPServer(("127.0.0.1", port), _OAuthCallbackHandler)
server.timeout = 120 # 2 minute timeout
click.echo(f"Opening browser to login at {server_url}...")
webbrowser.open(auth_url)
click.echo(f"Waiting for callback on http://localhost:{port}{CALLBACK_PATH}")
# Handle one request (the callback)
while _OAuthCallbackHandler.code is None and _OAuthCallbackHandler.error is None:
server.handle_request()
server.server_close()
if _OAuthCallbackHandler.error:
raise click.ClickException(f"Login failed: {_OAuthCallbackHandler.error}")
if _OAuthCallbackHandler.state != state:
raise click.ClickException("State mismatch — possible CSRF attack.")
code = _OAuthCallbackHandler.code
if not code:
raise click.ClickException("No authorization code received.")
# Exchange code for tokens
tokens = client.exchange_code(
code=code, redirect_uri=redirect_uri, code_verifier=code_verifier
)
client.close()
token_data = {
"access_token": tokens.access_token,
"refresh_token": tokens.refresh_token,
"id_token": tokens.id_token,
"token_type": tokens.token_type,
"expires_in": tokens.expires_in,
"scope": tokens.scope,
"server_url": server_url,
"client_id": client_id,
"organization": org,
"application": app,
"login_time": int(time.time()),
}
_save_token(token_data)
return token_data
# =========================================================================
# Password Login Flow (--no-browser)
# =========================================================================
def password_login(
username: str | None = None, password: str | None = None
) -> dict[str, Any]:
"""Login with username/password (no browser).
Uses the OAuth2 Resource Owner Password Credentials (ROPC) grant
to get a JWT directly from the token endpoint.
Returns:
Token data dict with access_token, etc.
"""
import httpx
server_url = _iam_url()
org = _iam_org()
app = _iam_app()
client_id = _iam_client_id()
client_secret = _env("IAM_CLIENT_SECRET")
if not username:
username = click.prompt("Username or email")
if not password:
password = click.prompt("Password", hide_input=True)
# The endpoint comes from IAMConfig, never a literal. The hardcoded
# "/oauth/token" that used to be here is not a 404 — IAM serves a
# 200 text/html SPA catch-all for any unregistered path, so this call
# returned a login PAGE and resp.json() blew up on HTML. A wrong path
# here is silent breakage, which is why there is exactly one definition.
config = IAMConfig(
server_url=server_url,
client_id=client_id,
client_secret=client_secret,
organization=org,
application=app,
)
resp = httpx.post(
config.token_endpoint,
data={
"grant_type": "password",
"client_id": client_id,
"client_secret": client_secret,
"username": username,
"password": password,
"scope": "openid profile email",
},
headers={"Content-Type": "application/x-www-form-urlencoded"},
timeout=30.0,
)
data = resp.json()
if "error" in data:
raise click.ClickException(data.get("error_description", data["error"]))
access_token = data.get("access_token", "")
if not access_token:
raise click.ClickException("No access token in response")
token_data = {
"access_token": access_token,
"refresh_token": data.get("refresh_token", ""),
"id_token": data.get("id_token", ""),
"token_type": data.get("token_type", "Bearer"),
"expires_in": data.get("expires_in", 0),
"scope": data.get("scope", ""),
"server_url": server_url,
"client_id": client_id,
"organization": org,
"application": app,
"login_time": int(time.time()),
}
_save_token(token_data)
# Verify BEFORE storing: a token we cannot check is not one to keep and
# later present as proof of identity.
result = verify_token_data(token_data)
if not result.valid:
raise click.ClickException(
f"IAM issued a token this client cannot verify ({result.reason}:"
f" {result.detail}). Not storing it."
)
token_data["stored_in"] = _save_token(token_data)
return token_data
pkg/hanzo-cli/hanzo_cli/bot/commands.py
+53 -236
View File
@@ -16,7 +16,6 @@ Usage:
from __future__ import annotations
import http.server
import json
import os
import shutil
@@ -498,222 +497,45 @@ def _load_bot_token() -> dict[str, Any] | None:
return None
class _BotOAuthCallbackHandler(http.server.BaseHTTPRequestHandler):
"""HTTP handler that captures the bot OAuth callback code."""
code: str | None = None
state: str | None = None
error: str | None = None
def do_GET(self) -> None: # noqa: N802
from urllib.parse import parse_qs, urlparse
parsed = urlparse(self.path)
if parsed.path != BOT_CALLBACK_PATH:
self.send_response(404)
self.end_headers()
return
qs = parse_qs(parsed.query)
if "error" in qs:
_BotOAuthCallbackHandler.error = qs["error"][0]
self._respond("Login failed. You can close this window.", success=False)
return
_BotOAuthCallbackHandler.code = qs.get("code", [None])[0]
_BotOAuthCallbackHandler.state = qs.get("state", [None])[0]
self._respond("Bot login successful! You can close this window.")
def _respond(self, message: str, success: bool = True) -> None:
self.send_response(200)
self.send_header("Content-Type", "text/html")
self.end_headers()
accent = "#6C5CE7" if success else "#e74c3c"
body = f"""<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Hanzo Bot</title>
<style>
*{{margin:0;padding:0;box-sizing:border-box}}
body{{
font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',Roboto,Helvetica,Arial,sans-serif;
background:#0a0a0a;color:#fafafa;
display:flex;align-items:center;justify-content:center;
min-height:100vh;
}}
.card{{
text-align:center;max-width:420px;padding:48px 32px;
background:#111;border:1px solid #222;border-radius:16px;
}}
.logo{{font-size:28px;font-weight:700;letter-spacing:-.5px;margin-bottom:32px;color:#fff}}
.logo span{{color:{accent}}}
.icon{{font-size:48px;margin-bottom:20px}}
h2{{font-size:20px;font-weight:600;margin-bottom:8px;color:#fff}}
.sub{{color:#888;font-size:14px;line-height:1.5;margin-bottom:28px}}
.tagline{{
font-size:13px;color:#555;border-top:1px solid #222;
padding-top:20px;margin-top:8px;
}}
</style>
</head>
<body>
<div class="card">
<div class="logo"><span>&#x25B2;</span> hanzo bot</div>
<div class="icon">{"&#x2705;" if success else "&#x274C;"}</div>
<h2>{message}</h2>
<p class="sub">You can close this window and return to your terminal.</p>
<p class="tagline">Build something you love.</p>
</div>
</body>
</html>"""
self.wfile.write(body.encode())
def log_message(self, format: str, *args: Any) -> None:
"""Suppress default request logging."""
def _bot_browser_login() -> str:
"""Run browser OAuth login flow for the bot gateway.
"""Sign the bot gateway in through the browser (auth code + PKCE).
Opens the user's browser to the IAM login page (hanzo-bot application),
starts a local HTTP server to receive the callback, exchanges the code
for tokens scoped to the hanzo-bot client.
Returns:
Access token string.
Delegates to hanzo_iam.oauth — the ONE login flow. This used to be a third
hand-rolled copy that posted to the unprefixed /oauth/token, sent no PKCE,
bound port 8398 (which is NOT a registered redirect_uri for hanzo-bot, so
/authorize refused it with a bare 400), and waited for its callback in an
unbounded handle_request() loop.
"""
import secrets
import time
import webbrowser
from hanzo_iam import IAMClient, IAMConfig
config = IAMConfig(
server_url=BOT_IAM_SERVER_URL,
client_id=BOT_IAM_CLIENT_ID,
client_secret="",
organization=BOT_IAM_ORG,
application=BOT_IAM_APP,
)
client = IAMClient(config=config)
redirect_uri = f"http://localhost:{BOT_CALLBACK_PORT}{BOT_CALLBACK_PATH}"
state = secrets.token_urlsafe(32)
auth_url = client.get_authorization_url(
redirect_uri=redirect_uri,
state=state,
scope="openid profile email",
)
# Reset handler state
_BotOAuthCallbackHandler.code = None
_BotOAuthCallbackHandler.state = None
_BotOAuthCallbackHandler.error = None
server = http.server.HTTPServer(
("127.0.0.1", BOT_CALLBACK_PORT), _BotOAuthCallbackHandler
)
server.timeout = 120
console.print(f"[cyan]Opening browser to login at {BOT_IAM_SERVER_URL}...[/cyan]")
webbrowser.open(auth_url)
console.print(
f"Waiting for callback on http://localhost:{BOT_CALLBACK_PORT}{BOT_CALLBACK_PATH}"
)
while (
_BotOAuthCallbackHandler.code is None
and _BotOAuthCallbackHandler.error is None
):
server.handle_request()
server.server_close()
if _BotOAuthCallbackHandler.error:
console.print(f"[red]Login failed:[/red] {_BotOAuthCallbackHandler.error}")
raise SystemExit(1)
if _BotOAuthCallbackHandler.state != state:
console.print("[red]State mismatch — possible CSRF attack.[/red]")
raise SystemExit(1)
code = _BotOAuthCallbackHandler.code
if not code:
console.print("[red]No authorization code received.[/red]")
raise SystemExit(1)
tokens = client.exchange_code(code=code, redirect_uri=redirect_uri)
client.close()
access_token = tokens.access_token
if not access_token:
console.print("[red]No access token in response.[/red]")
raise SystemExit(1)
_save_bot_token({
"access_token": access_token,
"refresh_token": tokens.refresh_token or "",
"id_token": tokens.id_token or "",
"expires_at": time.time() + (tokens.expires_in or 604800),
"client_id": BOT_IAM_CLIENT_ID,
"server_url": BOT_IAM_SERVER_URL,
"organization": BOT_IAM_ORG,
"application": BOT_IAM_APP,
"login_time": int(time.time()),
})
console.print("[green]Bot login successful.[/green]")
return access_token
def _bot_password_login(
username: str | None = None, password: str | None = None
) -> str:
"""Login via password grant using the hanzo-bot client. Returns access_token."""
import time
import httpx
from hanzo_iam import oauth
if not username:
username = click.prompt("Email")
if not password:
password = click.prompt("Password", hide_input=True)
redirect_uris = tuple(
u for u in (_env("BOT_REDIRECT_URI"),) if u
) or oauth.LOOPBACK_REDIRECTS
resp = httpx.post(
f"{BOT_IAM_SERVER_URL}/oauth/token",
data={
"grant_type": "password",
"client_id": BOT_IAM_CLIENT_ID,
"username": username,
"password": password,
"scope": "openid profile email",
},
headers={"Content-Type": "application/x-www-form-urlencoded"},
timeout=30.0,
)
data = resp.json()
access_token = data.get("access_token", "")
if not access_token:
err = (
data.get("error_description")
or data.get("msg")
or data.get("error", "unknown error")
try:
data = oauth.login(
server_url=BOT_IAM_SERVER_URL,
client_id=BOT_IAM_CLIENT_ID,
organization=BOT_IAM_ORG,
redirect_uris=redirect_uris,
on_url=lambda u: console.print(f"[cyan]Open this URL to sign in:[/cyan]\n\n {u}\n"),
)
console.print(f"[red]Login failed:[/red] {err}")
raise SystemExit(1)
except oauth.LoginError as e:
console.print(f"[red]Bot login failed:[/red] {e}")
console.print(
"[yellow]Note:[/yellow] the hanzo-bot IAM application has no loopback"
" redirect_uri registered. Register one (or set BOT_REDIRECT_URI to a"
" registered value) before browser login can complete."
)
raise SystemExit(1) from e
_save_bot_token({
"access_token": access_token,
"refresh_token": data.get("refresh_token", ""),
"expires_at": time.time() + data.get("expires_in", 604800),
"client_id": BOT_IAM_CLIENT_ID,
"username": username,
"login_time": int(time.time()),
})
return access_token
data["application"] = BOT_IAM_APP
data["expires_at"] = time.time() + (data.get("expires_in") or 604800)
_save_bot_token(data)
console.print("[green]Bot login successful.[/green]")
return data["access_token"]
def _get_iam_token() -> str:
@@ -822,38 +644,33 @@ def bot_install() -> None:
@bot.command("login")
@click.option(
"--no-browser",
is_flag=True,
help="Use password login instead of browser OAuth.",
)
def bot_login_cmd(no_browser: bool) -> None:
"""Authenticate with the bot gateway.
def bot_login_cmd() -> None:
"""Authenticate with the bot gateway (OAuth2 authorization code + PKCE).
Opens your browser to log in via Hanzo IAM. The token is stored
at ~/.hanzo/bot/token.json and used for subsequent bot commands.
\b
Examples:
hanzo bot login Browser OAuth login
hanzo bot login --no-browser Password login
The token is stored at ~/.hanzo/bot/token.json and used for subsequent bot
commands. Password login is gone: iam's password grant requires client
authentication (401 invalid_client without a secret), which a CLI cannot
supply, so `--no-browser` never worked.
"""
if no_browser:
token = _bot_password_login()
else:
token = _bot_browser_login()
from hanzo_iam import tokens
from hanzo_iam.models import OIDC_JWKS_PATH
# Decode and show who we logged in as
try:
import base64
token = _bot_browser_login()
parts = token.split(".")
payload = json.loads(base64.urlsafe_b64decode(parts[1] + "=="))
name = payload.get("name", "unknown")
email = payload.get("email", "")
console.print(f"[green]Logged in as:[/green] {name} ({email})")
except Exception:
console.print("[green]Login stored.[/green]")
# Identity from a VERIFIED token. This used to base64-decode the payload
# and print whatever it said, with no signature check at all.
result = tokens.verify(
token,
jwks_uri=f"{BOT_IAM_SERVER_URL}{OIDC_JWKS_PATH}",
issuer=BOT_IAM_SERVER_URL,
)
if not result.valid:
console.print(f"[red]Token did not verify:[/red] {result.reason}: {result.detail}")
raise SystemExit(1)
console.print(
f"[green]Logged in as:[/green] {result.claims.get('name', 'unknown')}"
f" ({result.claims.get('email', '')})"
)
@bot.command("logout")
pkg/hanzo-cli/hanzo_cli/cli.py
+41 -28
View File
@@ -22,7 +22,7 @@ from hanzo_cli import __version__
from hanzo_cli.auth import (
browser_login,
get_token_info,
password_login,
verify_token,
)
from hanzo_cli.auth import (
logout as do_logout,
@@ -46,27 +46,42 @@ def main(ctx: click.Context) -> None:
@main.command()
@click.option(
"--no-browser", is_flag=True, help="Use password login instead of browser OAuth."
"--no-browser",
is_flag=True,
help="Print the sign-in URL instead of opening a browser.",
)
@click.option(
"--port",
default=None,
type=int,
help=(
"Loopback callback port. Must be one registered on the IAM application"
" — iam matches redirect_uri by exact string."
),
)
@click.option("--port", default=8399, help="Local callback port for browser login.")
@click.pass_context
def login(ctx: click.Context, no_browser: bool, port: int) -> None:
"""Authenticate with Hanzo IAM."""
try:
if no_browser:
token_data = password_login()
else:
token_data = browser_login(port=port)
def login(ctx: click.Context, no_browser: bool, port: int | None) -> None:
"""Authenticate with Hanzo IAM (OAuth2 authorization code + PKCE).
console.print("[green]Logged in successfully.[/green]")
if token_data.get("access_token"):
# Decode the token subject for display
_at = token_data["access_token"]
console.print("Token stored at ~/.hanzo/auth/token.json")
Password login is deliberately absent: iam's password grant requires client
authentication (it answers 401 invalid_client without a secret), and a CLI
cannot hold a secret. The device-code grant would be the right answer for a
headless box, but no PUBLIC client is registered, so it 401s for every
client_id today. --no-browser prints the URL for you to open elsewhere.
"""
try:
token_data = browser_login(port=port, open_browser=not no_browser)
except Exception as e:
console.print(f"[red]Login failed:[/red] {e}")
raise SystemExit(1) from e
result = verify_token()
claims = result.claims
console.print("[green]Logged in.[/green]")
console.print(f" user: {claims.get('email') or claims.get('sub')}")
console.print(f" org: {claims.get('owner') or token_data.get('organization')}")
console.print(f" token: verified, stored in {token_data.get('stored_in')}")
@main.command()
def logout() -> None:
@@ -92,19 +107,17 @@ def whoami() -> None:
table.add_row("Application", token_data.get("application", "—"))
table.add_row("Client ID", token_data.get("client_id", "—"))
# Try to decode the token for user info
access_token = token_data.get("access_token", "")
if access_token:
try:
import jwt
# Decode without verification just to show user info
claims = jwt.decode(access_token, options={"verify_signature": False})
table.add_row("User", claims.get("name", claims.get("sub", "—")))
table.add_row("Email", claims.get("email", "—"))
table.add_row("Owner", claims.get("owner", "—"))
except Exception:
table.add_row("Token", f"{access_token[:20]}...")
# Identity comes from a VERIFIED token. The previous version decoded with
# verify_signature=False and printed whatever the token claimed, so a
# forged file made `whoami` report an attacker-chosen identity as fact.
result = verify_token()
table.add_row("Token", "[green]verified[/green]" if result.valid
else f"[red]INVALID[/red] ({result.reason}: {result.detail})")
if result.valid:
claims = result.claims
table.add_row("User", claims.get("name") or claims.get("sub", "—"))
table.add_row("Email", claims.get("email", "—"))
table.add_row("Owner", claims.get("owner", "—"))
login_time = token_data.get("login_time")
if login_time:
pkg/hanzo-cli/pyproject.toml
+3 -8
View File
@@ -1,6 +1,6 @@
[project]
name = "hanzo-cli"
version = "0.2.3"
version = "0.2.4"
description = "Hanzo unified CLI — IAM, KMS, PaaS, S3, Deploy"
readme = "README.md"
license = { text = "MIT" }
@@ -36,13 +36,8 @@ dev = [
"ruff>=0.5.0",
]
# No [project.scripts]. The `hanzo` command belongs to the `hanzo` package and
# to nothing else. Both packages used to declare `hanzo = ...`, and since
# `hanzo` DEPENDS on this one, the winner was decided by pip install order —
# a coin flip between a 56-command CLI and this 10-command one. hanzo-cli kept
# winning, so `pip install hanzo` silently produced a CLI with no `auth`,
# `chat`, `agent`, `cloud`, `dev` or `mcp` at all. This package remains
# importable (`python -m hanzo_cli`) and its groups are mounted by `hanzo`.
[project.scripts]
hanzo = "hanzo_cli.cli:main"
[project.urls]
Homepage = "https://hanzo.ai"
pkg/hanzo-iam/hanzo_iam/__init__.py
+32 -3
View File
@@ -1,22 +1,37 @@
"""Hanzo IAM - Identity and Access Management SDK for Hanzo ecosystem."""
from hanzo_iam import store
from hanzo_iam.config import IAMConfig
from hanzo_iam.async_client import AsyncIAMClient
from hanzo_iam.client import IAMClient
# Modules are nouns (`oauth`, `tokens`, `store`); package exports are the verbs
# (`login`, `verify`). No export shadows the module it came from — an earlier
# layout had `hanzo_iam.verify` mean both a module and a function, and
# `import hanzo_iam.verify` silently handed back the function.
from hanzo_iam.oauth import LoginError, login
from hanzo_iam.models import (
IAM_ROUTE_PREFIX,
OIDC_AUTHORIZE_PATH,
OIDC_DEVICE_PATH,
OIDC_JWKS_PATH,
OIDC_TOKEN_PATH,
OIDC_USERINFO_PATH,
Application,
IAMConfig,
JWTClaims,
Organization,
TokenResponse,
User,
UserInfo,
)
from hanzo_iam.tokens import Verification, unverified_claims, verify
from importlib.metadata import PackageNotFoundError, version as _version
try:
# Single-sourced from the installed distribution. This said "1.1.1" while
# pyproject said 1.30.0 — 29 minor versions of drift.
# Single-sourced from the installed distribution. A second literal here
# said 1.1.1 while pyproject said 1.30.0, and callers read whichever they
# happened to reach.
__version__ = _version("hanzo-iam")
except PackageNotFoundError: # source tree, not installed
__version__ = "0.0.0+dev"
@@ -34,4 +49,18 @@ __all__ = [
"TokenResponse",
"User",
"UserInfo",
# Endpoint seam
"IAM_ROUTE_PREFIX",
"OIDC_AUTHORIZE_PATH",
"OIDC_DEVICE_PATH",
"OIDC_JWKS_PATH",
"OIDC_TOKEN_PATH",
"OIDC_USERINFO_PATH",
# Auth
"login",
"LoginError",
"store",
"verify",
"Verification",
"unverified_claims",
]
pkg/hanzo-iam/hanzo_iam/async_client.py
+24 -23
View File
@@ -9,7 +9,10 @@ from urllib.parse import urlencode
import httpx
import jwt
from hanzo_iam.config import IAMConfig
from hanzo_iam.models import (
IAM_ROUTE_PREFIX,
OIDC_AUTHORIZE_PATH,
OIDC_DISCOVERY_PATH,
OIDC_INTROSPECT_PATH,
OIDC_JWKS_PATH,
@@ -26,7 +29,6 @@ from hanzo_iam.models import (
if TYPE_CHECKING:
from jwt import PyJWKClient
from hanzo_iam.config import IAMConfig
class AsyncIAMClient:
@@ -73,13 +75,12 @@ class AsyncIAMClient:
"""
# Avoid circular import
from hanzo_iam.client import IAMClient
from hanzo_iam.models import IAMConfig as ModelConfig
if config:
self._config = config
else:
env_config = IAMClient._config_from_env(org)
self._config = ModelConfig(
self._config = IAMConfig(
server_url=env_config.server_url,
client_id=client_id or env_config.client_id,
client_secret=client_secret or env_config.client_secret,
@@ -96,7 +97,7 @@ class AsyncIAMClient:
@property
def config(self) -> IAMConfig:
"""Get client configuration."""
return self._config # type: ignore[return-value]
return self._config
async def _get_http(self) -> httpx.AsyncClient:
"""Get or create async HTTP client."""
@@ -206,7 +207,7 @@ class AsyncIAMClient:
params["code_challenge_method"] = code_challenge_method or "S256"
base_url = self._config.server_url.rstrip("/")
return f"{base_url}/v1/iam/oauth/authorize?{urlencode(params)}"
return f"{base_url}{OIDC_AUTHORIZE_PATH}?{urlencode(params)}"
async def exchange_code(
self,
@@ -325,7 +326,7 @@ class AsyncIAMClient:
jwt.InvalidTokenError: If token is invalid or expired.
"""
if self._jwks_client is None:
jwks_url = f"{self._config.server_url.rstrip('/')}/v1/iam/.well-known/jwks"
jwks_url = self._config.jwks_uri
self._jwks_client = jwt.PyJWKClient(jwks_url)
signing_key = self._jwks_client.get_signing_key_from_jwt(token)
@@ -446,7 +447,7 @@ class AsyncIAMClient:
http = await self._get_http()
response = await http.get(
"/v1/iam/get-user",
f"{IAM_ROUTE_PREFIX}/get-user",
params=params,
headers=self._admin_headers(),
)
@@ -471,7 +472,7 @@ class AsyncIAMClient:
http = await self._get_http()
response = await http.get(
"/v1/iam/get-users",
f"{IAM_ROUTE_PREFIX}/get-users",
params=params,
headers=self._admin_headers(),
)
@@ -508,7 +509,7 @@ class AsyncIAMClient:
http = await self._get_http()
response = await http.get(
"/v1/iam/get-user-count",
f"{IAM_ROUTE_PREFIX}/get-user-count",
params=params,
headers=self._admin_headers(),
)
@@ -557,7 +558,7 @@ class AsyncIAMClient:
"""Modify user via API."""
http = await self._get_http()
response = await http.post(
f"/api/{action}",
f"{IAM_ROUTE_PREFIX}/{action}",
params=self._admin_params(),
headers=self._admin_headers(),
json=user.model_dump(by_alias=True, exclude_none=True),
@@ -583,7 +584,7 @@ class AsyncIAMClient:
http = await self._get_http()
response = await http.get(
"/v1/iam/get-application",
f"{IAM_ROUTE_PREFIX}/get-application",
params=params,
headers=self._admin_headers(),
)
@@ -612,7 +613,7 @@ class AsyncIAMClient:
http = await self._get_http()
response = await http.get(
"/v1/iam/get-organizations",
f"{IAM_ROUTE_PREFIX}/get-organizations",
params=params,
headers=self._admin_headers(),
)
@@ -640,7 +641,7 @@ class AsyncIAMClient:
http = await self._get_http()
response = await http.get(
"/v1/iam/get-organization",
f"{IAM_ROUTE_PREFIX}/get-organization",
params=params,
headers=self._admin_headers(),
)
@@ -692,7 +693,7 @@ class AsyncIAMClient:
http = await self._get_http()
response = await http.post(
"/v1/iam/enforce",
f"{IAM_ROUTE_PREFIX}/enforce",
json=payload,
headers=self._admin_headers(),
)
@@ -737,7 +738,7 @@ class AsyncIAMClient:
http = await self._get_http()
response = await http.post(
"/v1/iam/batch-enforce",
f"{IAM_ROUTE_PREFIX}/batch-enforce",
json=payload,
headers=self._admin_headers(),
)
@@ -770,7 +771,7 @@ class AsyncIAMClient:
http = await self._get_http()
response = await http.get(
"/v1/iam/get-roles",
f"{IAM_ROUTE_PREFIX}/get-roles",
params=params,
headers=self._admin_headers(),
)
@@ -805,7 +806,7 @@ class AsyncIAMClient:
http = await self._get_http()
response = await http.get(
"/v1/iam/get-role",
f"{IAM_ROUTE_PREFIX}/get-role",
params=params,
headers=self._admin_headers(),
)
@@ -840,7 +841,7 @@ class AsyncIAMClient:
http = await self._get_http()
response = await http.get(
"/v1/iam/get-user-roles",
f"{IAM_ROUTE_PREFIX}/get-user-roles",
params=params,
headers=self._admin_headers(),
)
@@ -878,7 +879,7 @@ class AsyncIAMClient:
http = await self._get_http()
response = await http.post(
"/v1/iam/add-user-role",
f"{IAM_ROUTE_PREFIX}/add-user-role",
json=payload,
headers=self._admin_headers(),
)
@@ -916,7 +917,7 @@ class AsyncIAMClient:
http = await self._get_http()
response = await http.post(
"/v1/iam/delete-user-role",
f"{IAM_ROUTE_PREFIX}/delete-user-role",
json=payload,
headers=self._admin_headers(),
)
@@ -959,7 +960,7 @@ class AsyncIAMClient:
http = await self._get_http()
response = await http.post(
"/v1/iam/set-password",
f"{IAM_ROUTE_PREFIX}/set-password",
params=self._admin_params(),
headers=self._admin_headers(),
json=payload,
@@ -992,7 +993,7 @@ class AsyncIAMClient:
http = await self._get_http()
response = await http.get(
"/v1/iam/get-applications",
f"{IAM_ROUTE_PREFIX}/get-applications",
params=params,
headers=self._admin_headers(),
)
@@ -1016,7 +1017,7 @@ class AsyncIAMClient:
"""
http = await self._get_http()
response = await http.post(
"/v1/iam/update-application",
f"{IAM_ROUTE_PREFIX}/update-application",
params=self._admin_params(),
headers=self._admin_headers(),
json=application.model_dump(by_alias=True, exclude_none=True),
pkg/hanzo-iam/hanzo_iam/client.py
+22 -20
View File
@@ -14,14 +14,16 @@ from urllib.parse import urlencode
import httpx
import jwt
from .config import IAMConfig
from .models import (
IAM_ROUTE_PREFIX,
OIDC_AUTHORIZE_PATH,
OIDC_DISCOVERY_PATH,
OIDC_INTROSPECT_PATH,
OIDC_JWKS_PATH,
OIDC_TOKEN_PATH,
OIDC_USERINFO_PATH,
Application,
IAMConfig,
JWTClaims,
Organization,
TokenResponse,
@@ -218,7 +220,7 @@ class IAMClient:
params["code_challenge_method"] = code_challenge_method or "S256"
base_url = self._config.server_url.rstrip("/")
return f"{base_url}/v1/iam/oauth/authorize?{urlencode(params)}"
return f"{base_url}{OIDC_AUTHORIZE_PATH}?{urlencode(params)}"
def exchange_code(
self,
@@ -332,7 +334,7 @@ class IAMClient:
"""
# Initialize JWKS client if needed
if self._jwks_client is None:
jwks_url = f"{self._config.server_url.rstrip('/')}/v1/iam/.well-known/jwks"
jwks_url = self._config.jwks_uri
self._jwks_client = jwt.PyJWKClient(jwks_url)
# Get signing key from JWKS
@@ -472,7 +474,7 @@ class IAMClient:
}
response = self.http.get(
"/v1/iam/get-user",
f"{IAM_ROUTE_PREFIX}/get-user",
params=params,
headers=self._admin_headers(),
)
@@ -496,7 +498,7 @@ class IAMClient:
}
response = self.http.get(
"/v1/iam/get-users",
f"{IAM_ROUTE_PREFIX}/get-users",
params=params,
headers=self._admin_headers(),
)
@@ -521,7 +523,7 @@ class IAMClient:
}
response = self.http.get(
"/v1/iam/get-application",
f"{IAM_ROUTE_PREFIX}/get-application",
params=params,
headers=self._admin_headers(),
)
@@ -573,7 +575,7 @@ class IAMClient:
def _modify_user(self, action: str, user: User) -> dict:
"""Modify user via IAM admin API."""
response = self.http.post(
f"/api/{action}",
f"{IAM_ROUTE_PREFIX}/{action}",
params=self._admin_params(),
headers=self._admin_headers(),
json=user.model_dump(by_alias=True, exclude_none=True),
@@ -602,7 +604,7 @@ class IAMClient:
}
response = self.http.get(
"/v1/iam/get-organizations",
f"{IAM_ROUTE_PREFIX}/get-organizations",
params=params,
headers=self._admin_headers(),
)
@@ -629,7 +631,7 @@ class IAMClient:
}
response = self.http.get(
"/v1/iam/get-organization",
f"{IAM_ROUTE_PREFIX}/get-organization",
params=params,
headers=self._admin_headers(),
)
@@ -660,7 +662,7 @@ class IAMClient:
}
response = self.http.get(
"/v1/iam/get-providers",
f"{IAM_ROUTE_PREFIX}/get-providers",
params=params,
headers=self._admin_headers(),
)
@@ -691,7 +693,7 @@ class IAMClient:
}
response = self.http.get(
"/v1/iam/get-roles",
f"{IAM_ROUTE_PREFIX}/get-roles",
params=params,
headers=self._admin_headers(),
)
@@ -733,7 +735,7 @@ class IAMClient:
}
response = self.http.post(
"/v1/iam/set-password",
f"{IAM_ROUTE_PREFIX}/set-password",
params=self._admin_params(),
headers=self._admin_headers(),
json=payload,
@@ -765,7 +767,7 @@ class IAMClient:
}
response = self.http.get(
"/v1/iam/get-applications",
f"{IAM_ROUTE_PREFIX}/get-applications",
params=params,
headers=self._admin_headers(),
)
@@ -788,7 +790,7 @@ class IAMClient:
API response data.
"""
response = self.http.post(
"/v1/iam/update-application",
f"{IAM_ROUTE_PREFIX}/update-application",
params=self._admin_params(),
headers=self._admin_headers(),
json=application.model_dump(by_alias=True, exclude_none=True),
@@ -823,7 +825,7 @@ class IAMClient:
"application": self._config.application,
}
response = self.http.post("/v1/iam/login", json=payload)
response = self.http.post(f"{IAM_ROUTE_PREFIX}/login", json=payload)
response.raise_for_status()
data = response.json()
@@ -945,7 +947,7 @@ class AsyncIAMClient:
params["code_challenge_method"] = code_challenge_method or "S256"
base_url = self._config.server_url.rstrip("/")
return f"{base_url}/v1/iam/oauth/authorize?{urlencode(params)}"
return f"{base_url}{OIDC_AUTHORIZE_PATH}?{urlencode(params)}"
async def exchange_code(
self,
@@ -1018,7 +1020,7 @@ class AsyncIAMClient:
) -> JWTClaims:
"""Validate JWT token using JWKS (sync operation)."""
if self._jwks_client is None:
jwks_url = f"{self._config.server_url.rstrip('/')}/v1/iam/.well-known/jwks"
jwks_url = self._config.jwks_uri
self._jwks_client = jwt.PyJWKClient(jwks_url)
signing_key = self._jwks_client.get_signing_key_from_jwt(token)
@@ -1076,7 +1078,7 @@ class AsyncIAMClient:
}
http = await self._get_http()
response = await http.get("/v1/iam/get-user", params=params)
response = await http.get(f"{IAM_ROUTE_PREFIX}/get-user", params=params)
response.raise_for_status()
data = response.json()
@@ -1094,7 +1096,7 @@ class AsyncIAMClient:
}
http = await self._get_http()
response = await http.get("/v1/iam/get-users", params=params)
response = await http.get(f"{IAM_ROUTE_PREFIX}/get-users", params=params)
response.raise_for_status()
data = response.json()
@@ -1113,7 +1115,7 @@ class AsyncIAMClient:
}
http = await self._get_http()
response = await http.get("/v1/iam/get-application", params=params)
response = await http.get(f"{IAM_ROUTE_PREFIX}/get-application", params=params)
response.raise_for_status()
data = response.json()
pkg/hanzo-iam/hanzo_iam/config.py
+22 -10
View File
@@ -9,6 +9,8 @@ from pydantic import BaseModel, ConfigDict, Field
from hanzo_iam.models import (
OIDC_AUTHORIZE_PATH,
OIDC_DEVICE_PATH,
OIDC_JWKS_PATH,
OIDC_TOKEN_PATH,
OIDC_USERINFO_PATH,
)
@@ -79,22 +81,32 @@ class IAMConfig(BaseModel):
certificate=cert_val,
)
@property
def base_url(self) -> str:
"""Server URL with any trailing slash removed."""
return self.server_url.rstrip("/")
@property
def token_endpoint(self) -> str:
"""OAuth2 token endpoint URL (HIP-0111 canonical path)."""
return f"{self.server_url}{OIDC_TOKEN_PATH}"
"""OAuth2 token endpoint URL."""
return f"{self.base_url}{OIDC_TOKEN_PATH}"
@property
def authorize_endpoint(self) -> str:
"""OAuth2 authorization endpoint URL (HIP-0111 canonical path)."""
return f"{self.server_url}{OIDC_AUTHORIZE_PATH}"
"""OAuth2 authorization endpoint URL."""
return f"{self.base_url}{OIDC_AUTHORIZE_PATH}"
@property
def device_endpoint(self) -> str:
"""RFC 8628 device authorization endpoint URL."""
return f"{self.base_url}{OIDC_DEVICE_PATH}"
@property
def jwks_uri(self) -> str:
"""JWKS URL used to verify tokens this issuer signed."""
return f"{self.base_url}{OIDC_JWKS_PATH}"
@property
def userinfo_endpoint(self) -> str:
"""OIDC UserInfo endpoint URL (HIP-0111 canonical path)."""
return f"{self.server_url}{OIDC_USERINFO_PATH}"
@property
def api_base(self) -> str:
"""Base URL for API calls."""
return f"{self.server_url}/api"
return f"{self.base_url}{OIDC_USERINFO_PATH}"
pkg/hanzo-iam/hanzo_iam/fastapi.py
+2 -5
View File
@@ -34,10 +34,9 @@ from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from jwt import PyJWKClient
from hanzo_iam.config import IAMConfig
from hanzo_iam.models import (
OIDC_JWKS_PATH,
OIDC_USERINFO_PATH,
IAMConfig,
JWTClaims,
Organization,
UserInfo,
@@ -115,9 +114,7 @@ def _get_jwks_client() -> PyJWKClient:
global _jwks_client
if _jwks_client is None:
config = get_config()
jwks_url = f"{config.server_url}{OIDC_JWKS_PATH}"
_jwks_client = PyJWKClient(jwks_url)
_jwks_client = PyJWKClient(get_config().jwks_uri)
return _jwks_client
pkg/hanzo-iam/hanzo_iam/models.py
+25 -51
View File
@@ -1,4 +1,10 @@
"""Pydantic models for Hanzo IAM."""
"""Wire-shape models and protocol constants for Hanzo IAM.
Client CONFIGURATION lives in hanzo_iam.config. There used to be a second,
near-identical IAMConfig here, and it was the one every client actually
imported -- so the endpoint properties on the other one were dead code and
fixing them changed nothing.
"""
from __future__ import annotations
@@ -8,18 +14,25 @@ from typing import Any
from pydantic import BaseModel, ConfigDict, Field
# Canonical IAM paths (HIP-0111), relative to the IAM server_url root. One way;
# no legacy /api/ prefix and no bare /oauth/*. IAM answers an unrouted path with
# a generic 401 (or, in a browser, a 200 SPA shell) rather than a 404 — so a
# wrong path is silent breakage and every path must come from here.
#
# Only OIDC discovery lives at the root; everything else is under /v1/iam.
# The ONE place this package spells IAM's route prefix. Every admin action and
# every OIDC endpoint is composed beneath it. The prefix this replaced is
# unreachable, not merely non-standard: the edge refuses it on iam.hanzo.ai with
# 403 and the gateway 404s it.
IAM_ROUTE_PREFIX = "/v1/iam"
# OIDC endpoints, composed from the one prefix. These are NOT root-relative:
# hanzo.id serves its sign-in SPA on every unmatched path, so an unprefixed
# /oauth/token or /.well-known/jwks answers 200 text/html — a client that trusts
# the status code then dies inside .json(). Prefixing is what makes a wrong path
# fail like a wrong path.
OIDC_AUTHORIZE_PATH = f"{IAM_ROUTE_PREFIX}/oauth/authorize"
OIDC_TOKEN_PATH = f"{IAM_ROUTE_PREFIX}/oauth/token"
OIDC_USERINFO_PATH = f"{IAM_ROUTE_PREFIX}/oauth/userinfo"
OIDC_INTROSPECT_PATH = f"{IAM_ROUTE_PREFIX}/oauth/introspect"
OIDC_REVOKE_PATH = f"{IAM_ROUTE_PREFIX}/oauth/revoke"
OIDC_DEVICE_PATH = f"{IAM_ROUTE_PREFIX}/oauth/device"
OIDC_JWKS_PATH = f"{IAM_ROUTE_PREFIX}/.well-known/jwks"
OIDC_DISCOVERY_PATH = "/.well-known/openid-configuration"
OIDC_JWKS_PATH = "/v1/iam/.well-known/jwks"
OIDC_AUTHORIZE_PATH = "/v1/iam/oauth/authorize"
OIDC_TOKEN_PATH = "/v1/iam/oauth/token"
OIDC_USERINFO_PATH = "/v1/iam/oauth/userinfo"
OIDC_INTROSPECT_PATH = "/v1/iam/oauth/introspect"
class Organization(str, Enum):
@@ -36,45 +49,6 @@ class Organization(str, Enum):
return f"https://{self.value}.id"
class IAMConfig(BaseModel):
"""Configuration for Hanzo IAM client."""
model_config = ConfigDict(frozen=True)
server_url: str = Field(description="IAM server URL (e.g., https://hanzo.id)")
client_id: str = Field(description="OAuth2 client ID")
client_secret: str = Field(default="", description="OAuth2 client secret")
organization: str = Field(default="hanzo", description="IAM organization name")
application: str = Field(default="app", description="IAM application name")
certificate: str = Field(
default="", description="JWT verification certificate (PEM)"
)
# HIP-0111 endpoints, derived from the constants above so a caller can never
# hand-assemble a path. This is the IAMConfig the package exports, so these
# have to live here: callers doing `from hanzo_iam import IAMConfig` get
# this class, not the one in config.py.
@property
def authorize_endpoint(self) -> str:
"""OAuth2 authorization endpoint URL."""
return f"{self.server_url.rstrip('/')}{OIDC_AUTHORIZE_PATH}"
@property
def token_endpoint(self) -> str:
"""OAuth2 token endpoint URL."""
return f"{self.server_url.rstrip('/')}{OIDC_TOKEN_PATH}"
@property
def userinfo_endpoint(self) -> str:
"""OIDC UserInfo endpoint URL."""
return f"{self.server_url.rstrip('/')}{OIDC_USERINFO_PATH}"
@property
def jwks_endpoint(self) -> str:
"""OIDC JWKS endpoint URL."""
return f"{self.server_url.rstrip('/')}{OIDC_JWKS_PATH}"
class TokenResponse(BaseModel):
"""OAuth2 token response."""
pkg/hanzo-iam/hanzo_iam/oauth.py
+354
View File
@@ -0,0 +1,354 @@
"""The ONE interactive login flow: loopback redirect + PKCE (RFC 8252 / 7636).
WHY NOT THE DEVICE GRANT (RFC 8628), which would be the better CLI UX:
iam implements it fully and advertises it in discovery, but no PUBLIC client is
registered. `internal/oidc/device.go` demands a client secret from any app whose
ClientSecret is non-empty, and every registered app has one, so
``POST /v1/iam/oauth/device`` answers 401 invalid_client for hanzo-cli,
hanzo-app, hanzo-cloud and hanzo-console alike. A CLI cannot fix that by
shipping an embedded secret — that is not a secret. Registering ONE public
app (empty ClientSecret, device_code in grantTypes) is the whole server-side
change; until then this module is the only flow that can actually complete, and
shipping a device path that always 401s would be a lie in code form.
The password grant is also out: it requires client authentication too
(verified: 401 invalid_client without a secret), and ROPC cannot carry SSO or
MFA regardless.
The loopback flow works today because the token endpoint authenticates a
PKCE-bound code by the code binding rather than by a client secret when the
client presents none (iam v1.33.26, internal/oidc/token.go) — the same
relaxation that lets the hanzo.chat and cloud.hanzo.ai SPAs sign in.
"""
from __future__ import annotations
import base64
import hashlib
import http.server
import os
import secrets
import selectors
import socket
import time
import urllib.parse
import webbrowser
from dataclasses import dataclass
from typing import Any
import httpx
from hanzo_iam.models import OIDC_AUTHORIZE_PATH, OIDC_TOKEN_PATH
DEFAULT_IAM_URL = "https://hanzo.id"
DEFAULT_CLIENT_ID = "hanzo-app"
DEFAULT_ORG = "hanzo"
DEFAULT_SCOPE = "openid profile email"
# Loopback redirect URIs registered for DEFAULT_CLIENT_ID, in preference order.
# iam matches redirect_uri by EXACT STRING (internal/schema/application.go
# IsRedirectUriValid) — it does NOT apply the RFC 8252 §7.3 rule that a loopback
# port is ignored. So the CLI cannot pick a free ephemeral port; it must bind one
# of these exact URIs, host spelling included ("localhost", not "127.0.0.1").
LOOPBACK_REDIRECTS = (
"http://localhost:3000/callback",
"http://localhost:8080/callback",
)
DEFAULT_TIMEOUT = 300.0
class LoginError(RuntimeError):
"""Login could not complete. The message is meant for the end user."""
@dataclass(frozen=True)
class _Pkce:
verifier: str
challenge: str
@classmethod
def generate(cls) -> _Pkce:
verifier = _b64url(os.urandom(32))
challenge = _b64url(hashlib.sha256(verifier.encode("ascii")).digest())
return cls(verifier, challenge)
def _b64url(raw: bytes) -> str:
return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii")
def authorize_url(
server_url: str,
client_id: str,
redirect_uri: str,
state: str,
challenge: str,
scope: str = DEFAULT_SCOPE,
) -> str:
"""Build the /authorize URL for a PKCE loopback login."""
q = urllib.parse.urlencode(
{
"response_type": "code",
"client_id": client_id,
"redirect_uri": redirect_uri,
"scope": scope,
"state": state,
"code_challenge": challenge,
"code_challenge_method": "S256",
}
)
return f"{server_url.rstrip('/')}{OIDC_AUTHORIZE_PATH}?{q}"
def exchange_code(
server_url: str,
client_id: str,
code: str,
redirect_uri: str,
verifier: str,
timeout: float = 30.0,
) -> dict[str, Any]:
"""Redeem an authorization code. No client secret: PKCE is the proof.
Raises LoginError with the server's own words on failure. It never calls
.json() on a response it has not established is JSON — hanzo.id serves its
sign-in SPA (200 text/html) on any unmatched path, so a wrong path used to
surface as an inscrutable JSONDecodeError instead of "wrong path".
"""
resp = httpx.post(
f"{server_url.rstrip('/')}{OIDC_TOKEN_PATH}",
data={
"grant_type": "authorization_code",
"client_id": client_id,
"code": code,
"redirect_uri": redirect_uri,
"code_verifier": verifier,
},
headers={"Content-Type": "application/x-www-form-urlencoded"},
timeout=timeout,
)
payload = _json_or_raise(resp, "token endpoint")
if "error" in payload:
raise LoginError(
f"token exchange rejected: {payload['error']}"
f" — {payload.get('error_description', 'no description')}"
)
if not payload.get("access_token"):
raise LoginError("token endpoint returned no access_token")
return payload
def _json_or_raise(resp: httpx.Response, what: str) -> dict[str, Any]:
ctype = resp.headers.get("content-type", "")
if "json" not in ctype:
raise LoginError(
f"{what} answered {resp.status_code} {ctype or 'no content-type'}"
f" instead of JSON at {resp.request.url} — this is the wrong path,"
" not a rejected login"
)
try:
return resp.json()
except ValueError as e:
raise LoginError(f"{what} returned unparseable JSON: {e}") from e
def login(
server_url: str = DEFAULT_IAM_URL,
client_id: str = DEFAULT_CLIENT_ID,
organization: str = DEFAULT_ORG,
scope: str = DEFAULT_SCOPE,
redirect_uris: tuple[str, ...] = LOOPBACK_REDIRECTS,
timeout: float = DEFAULT_TIMEOUT,
open_browser: bool = True,
on_url: Any = None,
) -> dict[str, Any]:
"""Run the loopback+PKCE login and return token data. Does NOT persist it.
Persisting is `hanzo_iam.store.save`'s job — keeping them apart is what lets
a caller verify before it writes.
Args:
on_url: Optional callable receiving the authorize URL, so a CLI can
print it for a user whose browser is elsewhere.
Raises:
LoginError: with an actionable message on every failure path. It always
terminates — the flow this replaces looped on `handle_request()`
with no deadline and hung forever.
"""
pkce = _Pkce.generate()
state = secrets.token_urlsafe(32)
server, redirect_uri = _bind_registered(redirect_uris)
url = authorize_url(server_url, client_id, redirect_uri, state, pkce.challenge, scope)
try:
if on_url is not None:
on_url(url)
if open_browser:
webbrowser.open(url)
result = server.wait(timeout)
finally:
server.close()
if result.error:
raise LoginError(f"authorization failed: {result.error} — {result.error_description}")
if not result.code:
raise LoginError(
f"no authorization code received within {timeout:.0f}s at {redirect_uri}"
)
# Constant-time-ish state compare. A mismatch means the code arrived from a
# request this process did not start (RFC 6749 §10.12).
if not secrets.compare_digest(result.state or "", state):
raise LoginError("state mismatch — discarding the response (possible CSRF)")
tokens = exchange_code(server_url, client_id, result.code, redirect_uri, pkce.verifier)
now = int(time.time())
return {
**tokens,
"server_url": server_url.rstrip("/"),
"client_id": client_id,
"organization": organization,
"login_time": now,
"expires_at": now + int(tokens.get("expires_in") or 0),
}
# ---------------------------------------------------------------------------
# Loopback listener
# ---------------------------------------------------------------------------
@dataclass
class _Callback:
code: str | None = None
state: str | None = None
error: str | None = None
error_description: str = ""
_PAGE = """<!doctype html><html><head><meta charset="utf-8"><title>Hanzo</title>
<style>body{{font-family:system-ui,sans-serif;background:#0a0a0a;color:#fafafa;
display:flex;align-items:center;justify-content:center;min-height:100vh;margin:0}}
.c{{text-align:center;padding:48px 32px;border:1px solid #222;border-radius:16px;background:#111}}
h2{{margin:0 0 8px;font-size:20px}}p{{color:#888;font-size:14px;margin:0}}</style>
</head><body><div class="c"><h2>{title}</h2><p>{sub}</p></div></body></html>"""
def _bind_registered(uris: tuple[str, ...]) -> tuple[_Listener, str]:
"""Bind the first registered redirect URI whose port is free.
Only registered URIs are candidates: an unregistered one is refused by
/authorize with a bare 400 before the user ever sees a login page.
"""
busy = []
for uri in uris:
parsed = urllib.parse.urlparse(uri)
port = parsed.port or (443 if parsed.scheme == "https" else 80)
try:
return _Listener(port, parsed.path or "/"), uri
except OSError as e:
busy.append(f"{uri} ({e.strerror or e})")
raise LoginError(
"no registered loopback redirect URI could be bound: "
+ "; ".join(busy)
+ ". Free one of those ports, or register another loopback redirect_uri"
" on the IAM application."
)
class _Listener:
"""A one-shot loopback HTTP listener on both 127.0.0.1 and ::1.
Both families are bound because the registered URIs spell the host
"localhost", and a browser may resolve that to either. Binding only one
turns a working login into an intermittent one.
"""
def __init__(self, port: int, path: str) -> None:
self.result = _Callback()
self.path = path
self._servers: list[http.server.HTTPServer] = []
self._sel = selectors.DefaultSelector()
handler = _make_handler(self)
first_error: OSError | None = None
for family, host in ((socket.AF_INET, "127.0.0.1"), (socket.AF_INET6, "::1")):
try:
self._servers.append(_serve(family, host, port, handler))
except OSError as e:
# IPv4 must succeed; a box without ::1 is normal and fine.
if family == socket.AF_INET:
first_error = e
if first_error is not None or not self._servers:
self.close()
raise first_error or OSError(f"could not bind port {port}")
for s in self._servers:
self._sel.register(s, selectors.EVENT_READ)
def wait(self, timeout: float) -> _Callback:
"""Serve until the callback lands or the deadline passes. Never blocks
forever — an unbounded wait is what made the previous flow unusable."""
deadline = time.monotonic() + timeout
while self.result.code is None and self.result.error is None:
remaining = deadline - time.monotonic()
if remaining <= 0:
break
for key, _ in self._sel.select(timeout=min(remaining, 1.0)):
key.fileobj.handle_request() # type: ignore[union-attr]
return self.result
def close(self) -> None:
for s in self._servers:
s.server_close()
self._servers.clear()
self._sel.close()
def _serve(family: int, host: str, port: int, handler: Any) -> http.server.HTTPServer:
class Server(http.server.HTTPServer):
address_family = family
# SO_REUSEADDR is required, not convenience: the previous callback
# connection sits in TIME_WAIT for ~60s, so without it a second `login`
# in the same minute dies with EADDRINUSE on a port nothing is using.
# It does NOT let another process steal a live listener — two sockets
# on the same concrete addr:port still need SO_REUSEPORT, which we
# never set.
allow_reuse_address = True
return Server((host, port), handler)
def _make_handler(listener: _Listener) -> Any:
class Handler(http.server.BaseHTTPRequestHandler):
def do_GET(self) -> None: # noqa: N802
parsed = urllib.parse.urlparse(self.path)
if parsed.path != listener.path:
self.send_error(404)
return
q = urllib.parse.parse_qs(parsed.query)
if "error" in q:
listener.result.error = q["error"][0]
listener.result.error_description = q.get("error_description", [""])[0]
self._page("Login failed", listener.result.error_description or "Return to your terminal.")
return
listener.result.code = q.get("code", [None])[0]
listener.result.state = q.get("state", [None])[0]
if not listener.result.code:
listener.result.error = "invalid_request"
listener.result.error_description = "callback carried no code"
self._page("Signed in", "You can close this window and return to your terminal.")
def _page(self, title: str, sub: str) -> None:
body = _PAGE.format(title=title, sub=sub).encode()
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def log_message(self, fmt: str, *args: Any) -> None:
"""Silence: the access log would print the authorization code."""
return Handler
pkg/hanzo-iam/hanzo_iam/store.py
+147
View File
@@ -0,0 +1,147 @@
"""The ONE token store for Hanzo credentials.
Order of preference:
1. The OS keyring (Keychain / libsecret / Windows Credential Locker) when the
`keyring` package is installed AND a real backend is available.
2. ``~/.hanzo/auth/token.json``, created with mode 0600 ATOMICALLY.
"Atomically" is the whole point of the fallback. The code this replaces did
``write_text()`` then ``chmod(0600)``: between those two calls the file exists
at ``0666 & ~umask`` — commonly 0644 — holding a bearer token. Any process on
the box could read it in that window. Here the bytes are written to a
private temp file that is created 0600 by `mkstemp` and never named
``token.json`` until it is already correct, then `os.replace` swaps it in.
"""
from __future__ import annotations
import json
import os
import stat
import tempfile
from pathlib import Path
from typing import Any
TOKEN_DIR = Path.home() / ".hanzo" / "auth"
TOKEN_FILE = TOKEN_DIR / "token.json"
# Keyring coordinates. <org>-<app> naming, matching IAM.
KEYRING_SERVICE = "hanzo-cli"
KEYRING_USERNAME = "default"
FILE_MODE = 0o600
DIR_MODE = 0o700
def _keyring() -> Any | None:
"""Return a usable keyring module, or None.
Installed-but-headless is the common case on a server: `keyring` imports
fine and then raises on first use. Probing the backend here means the file
fallback is chosen deliberately rather than discovered mid-write.
"""
try:
import keyring
from keyring.backends.fail import Keyring as FailKeyring
except Exception:
return None
try:
if isinstance(keyring.get_keyring(), FailKeyring):
return None
except Exception:
return None
return keyring
def save(data: dict[str, Any]) -> str:
"""Persist token data. Returns the backend used: "keyring" or "file"."""
blob = json.dumps(data, indent=2, sort_keys=True)
kr = _keyring()
if kr is not None:
try:
kr.set_password(KEYRING_SERVICE, KEYRING_USERNAME, blob)
# A stale plaintext copy is still a plaintext copy.
_unlink_file()
return "keyring"
except Exception:
pass # fall through to the file store
_write_private(TOKEN_FILE, blob)
return "file"
def load() -> dict[str, Any] | None:
"""Read stored token data, or None when nothing is stored."""
kr = _keyring()
if kr is not None:
try:
blob = kr.get_password(KEYRING_SERVICE, KEYRING_USERNAME)
if blob:
return json.loads(blob)
except Exception:
pass
if not TOKEN_FILE.exists():
return None
try:
return json.loads(TOKEN_FILE.read_text())
except (json.JSONDecodeError, OSError):
return None
def clear() -> None:
"""Remove stored credentials from every backend."""
kr = _keyring()
if kr is not None:
try:
kr.delete_password(KEYRING_SERVICE, KEYRING_USERNAME)
except Exception:
pass
_unlink_file()
def backend() -> str:
"""Name the backend a save would use right now — for `auth status`."""
return "keyring" if _keyring() is not None else "file"
def file_mode() -> int | None:
"""Permission bits of the token file, or None when it does not exist."""
try:
return stat.S_IMODE(TOKEN_FILE.stat().st_mode)
except FileNotFoundError:
return None
def _unlink_file() -> None:
try:
TOKEN_FILE.unlink()
except FileNotFoundError:
pass
def _write_private(path: Path, text: str) -> None:
"""Write `text` to `path` so it is NEVER world- or group-readable.
mkstemp creates at 0600 before any byte is written, and os.replace is
atomic within a filesystem — a reader either sees the old file or the new
one, never a partial or over-permissive one.
"""
path.parent.mkdir(parents=True, exist_ok=True, mode=DIR_MODE)
# mkdir's mode is ignored when the directory already exists, so an
# inherited-permissive ~/.hanzo/auth is tightened here too.
os.chmod(path.parent, DIR_MODE)
fd, tmp = tempfile.mkstemp(dir=str(path.parent), prefix=".token-", suffix=".tmp")
try:
os.fchmod(fd, FILE_MODE) # explicit: do not inherit mkstemp's guarantee by luck
with os.fdopen(fd, "w") as f:
f.write(text)
f.flush()
os.fsync(f.fileno())
os.replace(tmp, path)
except BaseException:
os.unlink(tmp)
raise
pkg/hanzo-iam/hanzo_iam/tokens.py
+160
View File
@@ -0,0 +1,160 @@
"""The ONE place a Hanzo credential is judged valid.
Every auth surface in the SDK answers "am I logged in?" by calling `verify()`
here. It is deliberately the only implementation: the defect this replaces was
`bool(token_string)`, which reported success for the literal string
``fake.not.a.real.jwt`` and made every downstream permission check a lie.
`verify()` fails CLOSED. An unreachable JWKS is not a pass — a client that
cannot check a signature does not know the token is good, and saying otherwise
is the same bug in a new costume. The reason code says which it was so a caller
can tell "your token expired" from "you are offline".
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any
import jwt
# Signature algorithms iam issues. `none` is absent by construction — an
# unsigned token is an unauthenticated token, and listing it here is the classic
# JWT bypass.
ALGORITHMS = ["RS256", "ES256"]
# Reason codes. A caller branches on these; the human string is for the user.
OK = "ok"
NO_CREDENTIAL = "no_credential"
MALFORMED = "malformed"
EXPIRED = "expired"
BAD_SIGNATURE = "bad_signature"
WRONG_ISSUER = "wrong_issuer"
WRONG_AUDIENCE = "wrong_audience"
JWKS_UNREACHABLE = "jwks_unreachable"
UNKNOWN_KEY = "unknown_key"
OPAQUE = "opaque"
@dataclass(frozen=True)
class Verification:
"""The result of judging a credential. Truthy only when genuinely valid."""
valid: bool
reason: str
detail: str = ""
claims: dict[str, Any] = field(default_factory=dict)
def __bool__(self) -> bool:
return self.valid
def _jwks_client(jwks_uri: str) -> jwt.PyJWKClient:
# PyJWKClient caches signing keys in-process, so a long-lived session pays
# for the fetch once. Cached per URI so multiple issuers do not collide.
client = _JWKS_CLIENTS.get(jwks_uri)
if client is None:
client = jwt.PyJWKClient(jwks_uri, cache_keys=True, lifespan=600)
_JWKS_CLIENTS[jwks_uri] = client
return client
_JWKS_CLIENTS: dict[str, jwt.PyJWKClient] = {}
def reset_jwks_cache() -> None:
"""Drop cached signing keys — for tests and for key rotation."""
_JWKS_CLIENTS.clear()
def is_jwt(token: str) -> bool:
"""Report whether `token` is shaped like a JWS compact serialization.
Shape only. A string that passes this has NOT been verified — that is
precisely the confusion this module exists to end.
"""
parts = token.split(".")
return len(parts) == 3 and all(parts)
def verify(
token: str | None,
*,
jwks_uri: str,
issuer: str | None = None,
audience: str | None = None,
leeway: float = 30.0,
) -> Verification:
"""Verify `token`'s signature, expiry, issuer and (optionally) audience.
Args:
token: The access or ID token to judge. None/empty is not valid.
jwks_uri: Where the issuer publishes its signing keys. Must be the
prefixed path — hanzo.id answers 200 text/html on an unmatched
/.well-known/jwks, which reads as a successful fetch of garbage.
issuer: Expected `iss`. Checked when given.
audience: Expected `aud`. Checked when given. Left None by callers that
hold a token minted for a sibling app and only need to know the
issuer signed it.
leeway: Clock-skew tolerance in seconds for time-based claims.
Returns:
A Verification. `.valid` is True only if the signature checked out
against a published key and no claim was violated.
"""
if not token:
return Verification(False, NO_CREDENTIAL, "no token present")
if not is_jwt(token):
# An opaque credential (an API key) cannot be judged offline. Saying
# "valid" because it is a non-empty string is the original defect;
# callers that accept API keys must confirm them against the server.
return Verification(False, OPAQUE, "credential is not a JWT; verify server-side")
try:
signing_key = _jwks_client(jwks_uri).get_signing_key_from_jwt(token)
except jwt.exceptions.PyJWKClientConnectionError as e:
return Verification(False, JWKS_UNREACHABLE, f"cannot reach {jwks_uri}: {e}")
except jwt.exceptions.PyJWKClientError as e:
# Covers "no key for this kid" and a JWKS document that did not parse —
# which is what an HTML sign-in page deserialises to.
return Verification(False, UNKNOWN_KEY, f"no usable signing key from {jwks_uri}: {e}")
except Exception as e: # malformed header, unparseable token
return Verification(False, MALFORMED, str(e))
try:
claims = jwt.decode(
token,
signing_key.key,
algorithms=ALGORITHMS,
issuer=issuer,
audience=audience,
leeway=leeway,
options={
"verify_signature": True,
"verify_exp": True,
"verify_iss": issuer is not None,
"verify_aud": audience is not None,
"require": ["exp"],
},
)
except jwt.ExpiredSignatureError as e:
return Verification(False, EXPIRED, str(e))
except jwt.InvalidIssuerError as e:
return Verification(False, WRONG_ISSUER, str(e))
except jwt.InvalidAudienceError as e:
return Verification(False, WRONG_AUDIENCE, str(e))
except jwt.InvalidSignatureError as e:
return Verification(False, BAD_SIGNATURE, str(e))
except jwt.InvalidTokenError as e:
return Verification(False, MALFORMED, str(e))
return Verification(True, OK, claims=claims)
def unverified_claims(token: str) -> dict[str, Any]:
"""Decode claims WITHOUT verifying anything — display only.
Named so that no reader mistakes its output for a trust decision. Use
`verify()` for anything that gates access.
"""
return jwt.decode(token, options={"verify_signature": False})
pkg/hanzo-iam/pyproject.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "hanzo-iam"
version = "1.30.1"
version = "1.30.2"
description = "Hanzo IAM SDK - Identity and Access Management for Python"
readme = "README.md"
license = { text = "MIT" }
pkg/hanzo-iam/tests/test_endpoints.py
+93
View File
@@ -0,0 +1,93 @@
"""Every IAM endpoint must live under the one /v1/iam prefix.
These fail against the shipped code, which composed OAuth URLs from the server
root: `/oauth/token`, `/oauth/authorize`, `/oauth/introspect`,
`/.well-known/jwks`. hanzo.id serves its sign-in SPA on every unmatched path,
so all four answered **200 text/html**. Nothing 404'd, nothing raised for
status, and the failure only surfaced deep inside `.json()`.
"""
from __future__ import annotations
import pytest
from hanzo_iam import models
from hanzo_iam.config import IAMConfig
SERVER = "https://hanzo.id"
PATHS = [
models.OIDC_AUTHORIZE_PATH,
models.OIDC_TOKEN_PATH,
models.OIDC_USERINFO_PATH,
models.OIDC_INTROSPECT_PATH,
models.OIDC_REVOKE_PATH,
models.OIDC_DEVICE_PATH,
models.OIDC_JWKS_PATH,
]
@pytest.mark.parametrize("path", PATHS)
def test_every_endpoint_is_under_the_prefix(path):
assert path.startswith(models.IAM_ROUTE_PREFIX + "/")
def test_paths_are_composed_from_the_prefix_not_retyped():
"""One seam: changing IAM_ROUTE_PREFIX must move every endpoint."""
for path in PATHS:
assert path.count("/v1/iam") == 1
def test_discovery_is_the_one_root_relative_path():
"""RFC 8414 pins /.well-known/openid-configuration at the origin root; it
is the only IAM path that is NOT prefixed, and it does resolve live."""
assert models.OIDC_DISCOVERY_PATH == "/.well-known/openid-configuration"
@pytest.mark.parametrize(
"attr,expected",
[
("token_endpoint", "/v1/iam/oauth/token"),
("authorize_endpoint", "/v1/iam/oauth/authorize"),
("userinfo_endpoint", "/v1/iam/oauth/userinfo"),
("device_endpoint", "/v1/iam/oauth/device"),
("jwks_uri", "/v1/iam/.well-known/jwks"),
],
)
def test_config_endpoints(attr, expected):
config = IAMConfig(server_url=SERVER, client_id="hanzo-app")
assert getattr(config, attr) == SERVER + expected
def test_config_tolerates_a_trailing_slash():
"""A doubled slash is a different path, and hanzo.id would answer it with
the SPA rather than a 404."""
config = IAMConfig(server_url=SERVER + "/", client_id="hanzo-app")
assert config.token_endpoint == f"{SERVER}/v1/iam/oauth/token"
assert "//v1" not in config.token_endpoint
def test_jwks_uri_is_not_the_unprefixed_well_known_path():
"""The exact trap: https://hanzo.id/.well-known/jwks returns 200 text/html,
so a JWKS client pointed there fetches a web page and verifies nothing."""
config = IAMConfig(server_url=SERVER, client_id="hanzo-app")
assert config.jwks_uri != f"{SERVER}/.well-known/jwks"
assert config.jwks_uri != f"{SERVER}/.well-known/jwks.json"
def test_no_module_still_spells_a_bare_oauth_path():
"""Guards against a regression creeping back into any client module."""
import pathlib
import hanzo_iam
root = pathlib.Path(hanzo_iam.__file__).parent
offenders = []
for py in root.glob("*.py"):
for lineno, line in enumerate(py.read_text().splitlines(), 1):
if line.lstrip().startswith("#"):
continue
for bad in ('"/oauth/', "'/oauth/", '"/.well-known/jwks', "'/.well-known/jwks"):
if bad in line:
offenders.append(f"{py.name}:{lineno}: {line.strip()}")
assert not offenders, "unprefixed IAM paths:\n" + "\n".join(offenders)
pkg/hanzo-iam/tests/test_oauth.py
+367
View File
@@ -0,0 +1,367 @@
"""Tests for hanzo_iam.oauth — the loopback + PKCE login flow.
These pin the four things that made the previous flow unusable:
* it posted to /oauth/token, which hanzo.id answers with a 200 HTML page
* it sent no PKCE, so the exchange needed a client secret a CLI cannot hold
* it looped on handle_request() with no deadline and hung forever
* it bound a port nobody had registered as a redirect_uri
"""
from __future__ import annotations
import base64
import hashlib
import socket
import threading
import time
import urllib.parse
import urllib.request
import httpx
import pytest
from hanzo_iam import oauth
ISSUER = "https://hanzo.id"
def _params(url: str) -> dict[str, str]:
return dict(urllib.parse.parse_qsl(urllib.parse.urlparse(url).query))
def _free_port() -> int:
with socket.socket() as s:
s.bind(("127.0.0.1", 0))
return s.getsockname()[1]
# --- Endpoint paths -------------------------------------------------------
def test_authorize_url_carries_the_v1_iam_prefix():
"""Unprefixed /oauth/authorize is the SPA, not the endpoint."""
url = oauth.authorize_url(ISSUER, "hanzo-app", "http://localhost:3000/callback", "st", "ch")
assert url.startswith(f"{ISSUER}/v1/iam/oauth/authorize?")
def test_authorize_url_is_a_complete_pkce_request():
url = oauth.authorize_url(ISSUER, "hanzo-app", "http://localhost:3000/callback", "st", "ch")
q = _params(url)
assert q["response_type"] == "code"
assert q["client_id"] == "hanzo-app"
assert q["code_challenge"] == "ch"
assert q["code_challenge_method"] == "S256"
assert q["state"] == "st"
assert q["redirect_uri"] == "http://localhost:3000/callback"
def test_registered_redirect_uris_are_bindable_by_a_normal_user():
"""iam matches redirect_uri by exact string and does NOT apply RFC 8252
§7.3 port-agnostic loopback matching, so the CLI must use a registered URI
— and a privileged port would make the flow root-only."""
for uri in oauth.LOOPBACK_REDIRECTS:
parsed = urllib.parse.urlparse(uri)
assert parsed.hostname in ("localhost", "127.0.0.1")
assert parsed.port is not None, f"{uri} has no port; port 80 needs root"
assert parsed.port >= 1024, f"{uri} is a privileged port"
# --- PKCE -----------------------------------------------------------------
def test_pkce_challenge_is_s256_of_the_verifier():
p = oauth._Pkce.generate()
expected = base64.urlsafe_b64encode(
hashlib.sha256(p.verifier.encode("ascii")).digest()
).rstrip(b"=").decode()
assert p.challenge == expected
def test_pkce_verifier_meets_rfc7636_length_and_alphabet():
p = oauth._Pkce.generate()
assert 43 <= len(p.verifier) <= 128
assert set(p.verifier) <= set(
"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-._~"
)
def test_pkce_is_fresh_every_time():
assert oauth._Pkce.generate().verifier != oauth._Pkce.generate().verifier
# --- Token exchange -------------------------------------------------------
class _Capture:
def __init__(self, status=200, json_body=None, content_type="application/json", text=""):
self.status, self.body, self.ctype, self.text = status, json_body, content_type, text
self.sent: dict = {}
def __call__(self, url, data=None, headers=None, timeout=None):
self.sent = {"url": url, "data": data}
req = httpx.Request("POST", url)
return httpx.Response(
self.status,
headers={"content-type": self.ctype},
json=self.body if self.body is not None else None,
text=None if self.body is not None else self.text,
request=req,
)
def test_exchange_posts_to_the_prefixed_token_endpoint(monkeypatch):
cap = _Capture(json_body={"access_token": "a.b.c", "expires_in": 3600})
monkeypatch.setattr(httpx, "post", cap)
oauth.exchange_code(ISSUER, "hanzo-app", "code", "http://localhost:3000/callback", "verif")
assert cap.sent["url"] == f"{ISSUER}/v1/iam/oauth/token"
def test_exchange_sends_the_verifier_and_no_client_secret(monkeypatch):
"""PKCE is the proof. A CLI cannot hold a secret, and shipping one would
not be a secret."""
cap = _Capture(json_body={"access_token": "a.b.c"})
monkeypatch.setattr(httpx, "post", cap)
oauth.exchange_code(ISSUER, "hanzo-app", "code", "http://localhost:3000/callback", "verif")
assert cap.sent["data"]["code_verifier"] == "verif"
assert cap.sent["data"]["grant_type"] == "authorization_code"
assert "client_secret" not in cap.sent["data"]
def test_html_response_is_diagnosed_as_a_wrong_path(monkeypatch):
"""hanzo.id serves the sign-in SPA with 200 text/html on any unmatched
path. The old client called .json() on it and raised JSONDecodeError,
hiding a wrong URL behind a parse error."""
monkeypatch.setattr(httpx, "post", _Capture(content_type="text/html", text="<!doctype html>"))
with pytest.raises(oauth.LoginError, match="wrong path"):
oauth.exchange_code(ISSUER, "hanzo-app", "c", "http://localhost:3000/callback", "v")
def test_oauth_error_body_is_surfaced_verbatim(monkeypatch):
monkeypatch.setattr(
httpx,
"post",
_Capture(
status=400,
json_body={"error": "invalid_grant", "error_description": "code expired"},
),
)
with pytest.raises(oauth.LoginError, match="invalid_grant.*code expired"):
oauth.exchange_code(ISSUER, "hanzo-app", "c", "http://localhost:3000/callback", "v")
def test_missing_access_token_is_an_error_not_a_silent_success(monkeypatch):
monkeypatch.setattr(httpx, "post", _Capture(json_body={"token_type": "Bearer"}))
with pytest.raises(oauth.LoginError, match="no access_token"):
oauth.exchange_code(ISSUER, "hanzo-app", "c", "http://localhost:3000/callback", "v")
# --- The loopback listener ------------------------------------------------
def test_listener_times_out_instead_of_hanging_forever():
"""THE liveness regression. The shipped flow was
while handler.code is None and handler.error is None:
server.handle_request()
with no deadline: a user who closed the browser tab wedged the CLI until
it was killed. This must return promptly instead.
"""
port = _free_port()
listener = oauth._Listener(port, "/callback")
try:
started = time.monotonic()
result = listener.wait(timeout=0.5)
assert time.monotonic() - started < 5
assert result.code is None
finally:
listener.close()
def test_listener_captures_code_and_state():
port = _free_port()
listener = oauth._Listener(port, "/callback")
try:
def knock():
time.sleep(0.1)
urllib.request.urlopen(
f"http://127.0.0.1:{port}/callback?code=abc123&state=xyz", timeout=5
).read()
threading.Thread(target=knock, daemon=True).start()
result = listener.wait(timeout=10)
assert result.code == "abc123"
assert result.state == "xyz"
finally:
listener.close()
def test_listener_captures_an_authorization_error():
port = _free_port()
listener = oauth._Listener(port, "/callback")
try:
def knock():
time.sleep(0.1)
urllib.request.urlopen(
f"http://127.0.0.1:{port}/callback?error=access_denied"
"&error_description=user+said+no",
timeout=5,
).read()
threading.Thread(target=knock, daemon=True).start()
result = listener.wait(timeout=10)
assert result.error == "access_denied"
assert result.error_description == "user said no"
finally:
listener.close()
def test_listener_releases_its_port_on_close():
port = _free_port()
oauth._Listener(port, "/callback").close()
listener = oauth._Listener(port, "/callback") # would raise if still held
listener.close()
def test_binding_reports_which_ports_were_busy():
"""Every candidate port occupied — the user needs to know which and why,
not a bare failure. Ports 3000 and 8080 really are usually taken."""
uris = tuple(f"http://localhost:{_free_port()}/callback" for _ in range(2))
held = []
try:
for uri in uris:
s = socket.socket()
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.bind(("127.0.0.1", urllib.parse.urlparse(uri).port))
s.listen(1)
held.append(s)
with pytest.raises(oauth.LoginError) as e:
oauth._bind_registered(uris)
for uri in uris:
assert uri in str(e.value)
finally:
for s in held:
s.close()
def test_a_second_login_is_not_blocked_by_time_wait():
"""The callback socket lingers in TIME_WAIT for ~60s. Without SO_REUSEADDR
a user who runs `login` twice in a minute gets EADDRINUSE on a port nothing
is actually using."""
port = _free_port()
for _ in range(2):
listener = oauth._Listener(port, "/callback")
try:
def knock():
time.sleep(0.05)
urllib.request.urlopen(
f"http://127.0.0.1:{port}/callback?code=c&state=s", timeout=5
).read()
threading.Thread(target=knock, daemon=True).start()
assert listener.wait(timeout=10).code == "c"
finally:
listener.close()
# --- End to end -----------------------------------------------------------
@pytest.fixture
def free_redirect():
"""A redirect URI on a port nothing else holds.
The real registered URIs are ports 3000 and 8080 — the two most contended
ports on any developer box (8080 is taken by hanzoai/cloud on this one), so
a flow test that used them would be testing port luck. The registered URIs
are asserted separately by
test_registered_redirect_uris_are_bindable_by_a_normal_user.
"""
return (f"http://localhost:{_free_port()}/callback",)
def _drive_login(monkeypatch, respond, exchange=None):
"""Run login() while a fake browser hits the callback."""
seen: dict = {}
def fake_open(url):
seen["url"] = url
threading.Thread(target=lambda: respond(url), daemon=True).start()
return True
monkeypatch.setattr(oauth.webbrowser, "open", fake_open)
monkeypatch.setattr(
httpx, "post", exchange or _Capture(json_body={"access_token": "a.b.c", "expires_in": 60})
)
return seen
def test_login_end_to_end(monkeypatch, free_redirect):
def respond(url):
q = _params(url)
time.sleep(0.1)
urllib.request.urlopen(
f"{q['redirect_uri']}?code=THECODE&state={q['state']}", timeout=5
).read()
seen = _drive_login(monkeypatch, respond)
result = oauth.login(timeout=15, redirect_uris=free_redirect)
assert result["access_token"] == "a.b.c"
assert result["client_id"] == oauth.DEFAULT_CLIENT_ID
assert result["expires_at"] >= result["login_time"]
assert _params(seen["url"])["code_challenge_method"] == "S256"
def test_login_rejects_a_mismatched_state(monkeypatch, free_redirect):
"""A code that arrives with someone else's state is a CSRF attempt, and
must never be redeemed."""
exchanged = []
def respond(url):
q = _params(url)
time.sleep(0.1)
urllib.request.urlopen(
f"{q['redirect_uri']}?code=INJECTED&state=attacker", timeout=5
).read()
def never(*a, **kw):
exchanged.append(a)
raise AssertionError("must not exchange a code with a bad state")
_drive_login(monkeypatch, respond, exchange=never)
with pytest.raises(oauth.LoginError, match="state mismatch"):
oauth.login(timeout=15, redirect_uris=free_redirect)
assert exchanged == []
def test_login_surfaces_an_idp_error(monkeypatch, free_redirect):
def respond(url):
q = _params(url)
time.sleep(0.1)
urllib.request.urlopen(
f"{q['redirect_uri']}?error=access_denied&error_description=nope", timeout=5
).read()
_drive_login(monkeypatch, respond)
with pytest.raises(oauth.LoginError, match="access_denied"):
oauth.login(timeout=15, redirect_uris=free_redirect)
def test_login_gives_up_rather_than_hanging(monkeypatch, free_redirect):
monkeypatch.setattr(oauth.webbrowser, "open", lambda url: True)
started = time.monotonic()
with pytest.raises(oauth.LoginError, match="no authorization code received"):
oauth.login(timeout=0.5, redirect_uris=free_redirect)
assert time.monotonic() - started < 10
def test_login_can_hand_the_url_to_the_caller(monkeypatch, free_redirect):
"""For a user whose browser is not on this machine."""
urls: list[str] = []
monkeypatch.setattr(oauth.webbrowser, "open", lambda url: True)
with pytest.raises(oauth.LoginError):
oauth.login(timeout=0.3, open_browser=False, on_url=urls.append,
redirect_uris=free_redirect)
assert urls and urls[0].startswith(f"{ISSUER}/v1/iam/oauth/authorize?")
pkg/hanzo-iam/tests/test_store.py
+189
View File
@@ -0,0 +1,189 @@
"""Tests for hanzo_iam.store — the token store.
`test_token_file_is_never_world_readable_mid_write` is the regression: the code
this replaces did `write_text()` then `chmod(0600)`, so with a default umask of
022 the file existed at 0644 — holding a bearer token — between the two calls.
That test FAILS against write-then-chmod and passes against create-private-then-
rename.
"""
from __future__ import annotations
import json
import os
import stat
from pathlib import Path
import pytest
from hanzo_iam import store
TOKEN = {"access_token": "header.payload.signature", "refresh_token": "r"}
@pytest.fixture(autouse=True)
def isolated_home(tmp_path, monkeypatch):
"""Point the store at a scratch HOME and disable the keyring, so these
tests exercise the FILE backend — the one with the permissions hazard."""
monkeypatch.setattr(store, "TOKEN_DIR", tmp_path / ".hanzo" / "auth")
monkeypatch.setattr(store, "TOKEN_FILE", tmp_path / ".hanzo" / "auth" / "token.json")
monkeypatch.setattr(store, "_keyring", lambda: None)
return tmp_path
def test_roundtrip():
assert store.save(TOKEN) == "file"
assert store.load() == TOKEN
def test_load_returns_none_when_nothing_stored():
assert store.load() is None
def test_clear_removes_the_file():
store.save(TOKEN)
store.clear()
assert store.load() is None
assert not store.TOKEN_FILE.exists()
def test_token_file_is_0600():
store.save(TOKEN)
assert store.file_mode() == 0o600
def test_token_dir_is_0700():
store.save(TOKEN)
assert stat.S_IMODE(store.TOKEN_DIR.stat().st_mode) == 0o700
def test_permissive_umask_cannot_widen_the_token_file():
"""A 0000 umask is the worst case: anything created without an explicit
mode lands at 0666. The store must not care."""
old = os.umask(0o000)
try:
store.save(TOKEN)
finally:
os.umask(old)
assert store.file_mode() == 0o600
def test_token_file_is_never_world_readable_mid_write(monkeypatch):
"""THE regression test. Watch every mode the token file ever has.
write-then-chmod publishes the path at 0644 first and tightens it after;
this asserts the file is never observable at anything but 0600, which is
only true when it is created private and renamed into place.
"""
seen: list[int] = []
real_replace = os.replace
def watching_replace(src, dst):
# The instant before the name exists, and the instant after.
seen.append(stat.S_IMODE(os.stat(src).st_mode))
real_replace(src, dst)
seen.append(stat.S_IMODE(os.stat(dst).st_mode))
monkeypatch.setattr(os, "replace", watching_replace)
old = os.umask(0o000)
try:
store.save(TOKEN)
finally:
os.umask(old)
assert seen, "the store must publish the token by rename, not by writing in place"
assert all(m == 0o600 for m in seen), f"token was observable at {[oct(m) for m in seen]}"
def test_no_leftover_temp_files():
store.save(TOKEN)
leftovers = [p.name for p in store.TOKEN_DIR.iterdir() if p.name != "token.json"]
assert leftovers == []
def test_failed_write_leaves_no_temp_file_and_no_partial_token(monkeypatch):
store.save(TOKEN)
original = store.load()
def boom(src, dst):
raise OSError("disk full")
monkeypatch.setattr(os, "replace", boom)
with pytest.raises(OSError):
store.save({"access_token": "new"})
assert store.load() == original # the old token survived intact
assert [p.name for p in store.TOKEN_DIR.iterdir()] == ["token.json"]
def test_save_overwrites_rather_than_appends():
store.save(TOKEN)
store.save({"access_token": "second"})
assert store.load() == {"access_token": "second"}
assert json.loads(store.TOKEN_FILE.read_text()) == {"access_token": "second"}
def test_corrupt_file_reads_as_absent_not_as_a_credential():
store.TOKEN_DIR.mkdir(parents=True, exist_ok=True)
store.TOKEN_FILE.write_text("{not json")
assert store.load() is None
def test_keyring_backend_removes_any_plaintext_copy(monkeypatch):
"""Promoting to the keyring must delete the file — a stale plaintext token
on disk is exactly what the keyring is there to avoid."""
store.save(TOKEN)
assert store.TOKEN_FILE.exists()
vault: dict[tuple[str, str], str] = {}
class FakeKeyring:
@staticmethod
def set_password(service, user, value):
vault[(service, user)] = value
@staticmethod
def get_password(service, user):
return vault.get((service, user))
@staticmethod
def delete_password(service, user):
vault.pop((service, user), None)
monkeypatch.setattr(store, "_keyring", lambda: FakeKeyring)
assert store.save(TOKEN) == "keyring"
assert not store.TOKEN_FILE.exists()
assert store.load() == TOKEN
assert store.backend() == "keyring"
def test_falls_back_to_file_when_keyring_raises(monkeypatch):
class BrokenKeyring:
@staticmethod
def set_password(*a):
raise RuntimeError("no backend")
@staticmethod
def get_password(*a):
raise RuntimeError("no backend")
monkeypatch.setattr(store, "_keyring", lambda: BrokenKeyring)
assert store.save(TOKEN) == "file"
assert store.file_mode() == 0o600
assert store.load() == TOKEN
def test_existing_permissive_directory_is_tightened(isolated_home):
d = store.TOKEN_DIR
d.mkdir(parents=True)
os.chmod(d, 0o777)
store.save(TOKEN)
assert stat.S_IMODE(d.stat().st_mode) == 0o700
def test_write_private_is_reusable_for_other_credentials(tmp_path):
"""The PaaS session cache uses this too; it must be private there as well."""
target = tmp_path / "nested" / "session.json"
store._write_private(target, '{"at":"x"}')
assert stat.S_IMODE(target.stat().st_mode) == 0o600
assert target.read_text() == '{"at":"x"}'
pkg/hanzo-iam/tests/test_tokens.py
+227
View File
@@ -0,0 +1,227 @@
"""Tests for hanzo_iam.tokens — the credential judge.
The headline test is `test_fake_jwt_string_is_rejected`. Against the code this
replaced (`bool(token)` / `load_token() is not None`) it FAILS: that
implementation returned True for the literal string "fake.not.a.real.jwt".
"""
from __future__ import annotations
import json
import time
import jwt
import pytest
from cryptography.hazmat.primitives.asymmetric import rsa
import hanzo_iam.tokens as v
ISSUER = "https://hanzo.id"
JWKS_URI = f"{ISSUER}/v1/iam/.well-known/jwks"
# The exact string the old is_authenticated() accepted.
FAKE = "fake.not.a.real.jwt"
# Garbage that IS shaped like a JWS compact serialization, so only a signature
# check can reject it. FAKE has four dots and does not even reach that stage.
FAKE_SHAPED = "eyJhbGciOiJSUzI1NiIsImtpZCI6InRlc3Qta2V5In0.eyJzdWIiOiJyb290In0.c2ln"
@pytest.fixture(scope="module")
def keypair():
return rsa.generate_private_key(public_exponent=65537, key_size=2048)
@pytest.fixture(scope="module")
def other_keypair():
return rsa.generate_private_key(public_exponent=65537, key_size=2048)
def _jwks(key, kid="test-key"):
pub = key.public_key()
algo = jwt.algorithms.RSAAlgorithm
data = json.loads(algo.to_jwk(pub))
data.update({"kid": kid, "use": "sig", "alg": "RS256"})
return {"keys": [data]}
def _token(key, kid="test-key", **overrides):
claims = {
"iss": ISSUER,
"sub": "hanzo/z",
"aud": "hanzo-app",
"exp": int(time.time()) + 3600,
"iat": int(time.time()),
}
claims.update(overrides)
return jwt.encode(claims, key, algorithm="RS256", headers={"kid": kid})
@pytest.fixture(autouse=True)
def _clean():
v.reset_jwks_cache()
yield
v.reset_jwks_cache()
@pytest.fixture
def serve_jwks(monkeypatch):
"""Point PyJWKClient at an in-memory JWKS instead of the network."""
def _serve(document):
# fetch_data must hand back the parsed JSON dict; get_jwk_set builds
# the PyJWKSet itself.
monkeypatch.setattr(jwt.PyJWKClient, "fetch_data", lambda self: document)
return _serve
# --- The defect -----------------------------------------------------------
def test_fake_jwt_string_is_rejected(serve_jwks, keypair):
"""'fake.not.a.real.jwt' must NOT verify.
It is three dot-separated non-empty segments, so it survives every
shape check. Only a signature check catches it. The old
is_authenticated() returned True for exactly this string.
"""
serve_jwks(_jwks(keypair))
result = v.verify(FAKE, jwks_uri=JWKS_URI, issuer=ISSUER)
assert result.valid is False
assert bool(result) is False
# Not even a JWT — which the old check never noticed, because it only ever
# asked whether the string was non-empty.
assert result.reason == v.OPAQUE
def test_jwt_shaped_garbage_is_rejected(serve_jwks, keypair):
"""A forgery that DOES have the JWS shape gets past every structural test.
Only checking the signature against a published key rejects it."""
serve_jwks(_jwks(keypair))
result = v.verify(FAKE_SHAPED, jwks_uri=JWKS_URI, issuer=ISSUER)
assert result.valid is False
assert result.reason in (v.MALFORMED, v.BAD_SIGNATURE, v.UNKNOWN_KEY)
def test_shape_check_is_not_a_trust_check():
"""is_jwt() answers a question about punctuation, nothing more."""
assert v.is_jwt(FAKE_SHAPED) is True
assert v.verify(FAKE_SHAPED, jwks_uri=JWKS_URI, issuer=ISSUER).valid is False
@pytest.mark.parametrize("junk", ["", None, "not-a-token", "a.b", "....", "x" * 500])
def test_junk_is_rejected(junk):
assert v.verify(junk, jwks_uri=JWKS_URI, issuer=ISSUER).valid is False
# --- The happy path -------------------------------------------------------
def test_genuine_token_verifies(serve_jwks, keypair):
serve_jwks(_jwks(keypair))
result = v.verify(_token(keypair), jwks_uri=JWKS_URI, issuer=ISSUER, audience="hanzo-app")
assert result.valid is True
assert result.reason == v.OK
assert result.claims["sub"] == "hanzo/z"
# --- Every way a real-looking token can still be bad ----------------------
def test_expired_token_is_rejected(serve_jwks, keypair):
serve_jwks(_jwks(keypair))
stale = _token(keypair, exp=int(time.time()) - 3600)
result = v.verify(stale, jwks_uri=JWKS_URI, issuer=ISSUER)
assert result.valid is False
assert result.reason == v.EXPIRED
def test_token_signed_by_a_stranger_is_rejected(serve_jwks, keypair, other_keypair):
"""Correct kid, correct claims, wrong private key."""
serve_jwks(_jwks(keypair))
forged = _token(other_keypair)
result = v.verify(forged, jwks_uri=JWKS_URI, issuer=ISSUER)
assert result.valid is False
assert result.reason == v.BAD_SIGNATURE
def test_wrong_issuer_is_rejected(serve_jwks, keypair):
serve_jwks(_jwks(keypair))
result = v.verify(_token(keypair, iss="https://evil.example"), jwks_uri=JWKS_URI, issuer=ISSUER)
assert result.valid is False
assert result.reason == v.WRONG_ISSUER
def test_wrong_audience_is_rejected(serve_jwks, keypair):
serve_jwks(_jwks(keypair))
result = v.verify(
_token(keypair, aud="someone-else"), jwks_uri=JWKS_URI, issuer=ISSUER, audience="hanzo-app"
)
assert result.valid is False
assert result.reason == v.WRONG_AUDIENCE
def test_alg_none_token_is_rejected(serve_jwks, keypair):
"""The classic bypass: re-sign the claims with alg=none."""
serve_jwks(_jwks(keypair))
unsigned = jwt.encode({"iss": ISSUER, "sub": "hanzo/z", "exp": int(time.time()) + 3600},
key=None, algorithm="none", headers={"kid": "test-key"})
assert v.verify(unsigned, jwks_uri=JWKS_URI, issuer=ISSUER).valid is False
def test_token_without_exp_is_rejected(serve_jwks, keypair):
"""A token that never expires is not a session."""
serve_jwks(_jwks(keypair))
claims = {"iss": ISSUER, "sub": "hanzo/z"}
forever = jwt.encode(claims, keypair, algorithm="RS256", headers={"kid": "test-key"})
assert v.verify(forever, jwks_uri=JWKS_URI, issuer=ISSUER).valid is False
def test_unknown_kid_is_rejected(serve_jwks, keypair):
serve_jwks(_jwks(keypair, kid="rotated-away"))
assert v.verify(_token(keypair), jwks_uri=JWKS_URI, issuer=ISSUER).reason == v.UNKNOWN_KEY
# --- Fail closed ----------------------------------------------------------
def test_html_jwks_does_not_verify_anything(monkeypatch, keypair):
"""hanzo.id serves its sign-in SPA on unmatched paths, so a JWKS URL with a
missing /v1/iam prefix fetches 200 text/html. That must never read as
'verified' — it is the exact trap that made /.well-known/jwks look fine."""
def html(self, refresh=False):
raise jwt.exceptions.PyJWKClientError("expecting value: line 1 column 1")
monkeypatch.setattr(jwt.PyJWKClient, "fetch_data", html)
result = v.verify(_token(keypair), jwks_uri=f"{ISSUER}/.well-known/jwks", issuer=ISSUER)
assert result.valid is False
assert result.reason == v.UNKNOWN_KEY
def test_unreachable_jwks_fails_closed(monkeypatch, keypair):
def boom(self, refresh=False):
raise jwt.exceptions.PyJWKClientConnectionError("network down")
monkeypatch.setattr(jwt.PyJWKClient, "fetch_data", boom)
result = v.verify(_token(keypair), jwks_uri=JWKS_URI, issuer=ISSUER)
assert result.valid is False
assert result.reason == v.JWKS_UNREACHABLE
def test_opaque_credential_is_not_silently_accepted():
result = v.verify("hz_sk_live_abcdef123456", jwks_uri=JWKS_URI, issuer=ISSUER)
assert result.valid is False
assert result.reason == v.OPAQUE
def test_alg_none_is_not_in_the_accepted_list():
assert "none" not in [a.lower() for a in v.ALGORITHMS]
def test_unverified_claims_does_not_imply_trust():
"""unverified_claims reads a forged token happily — which is why nothing
that gates access may call it."""
forged = jwt.encode({"sub": "root", "exp": 1}, "k" * 32, algorithm="HS256")
assert v.unverified_claims(forged)["sub"] == "root"
assert v.verify(forged, jwks_uri=JWKS_URI, issuer=ISSUER).valid is False
pkg/hanzo-tools-auth/hanzo_tools/auth/login_tool.py
+104 -56
View File
@@ -1,7 +1,8 @@
"""MCP tool for Hanzo authentication management.
Provides status, login (browser flow), logout, and whoami actions
accessible via Claude Code or any MCP client.
Actions are exactly what this tool can do — `login` runs the real
loopback+PKCE flow (hanzo_iam.oauth), and `status`/`whoami` report VERIFIED
state, not the mere presence of a string.
"""
from __future__ import annotations
@@ -22,12 +23,15 @@ logger = logging.getLogger(__name__)
DESCRIPTION = """Hanzo authentication management.
Manage your Hanzo platform authentication. Check auth status, view current user
info, or logout.
Sign in to the Hanzo platform, inspect the current session, or sign out.
Tokens are verified against the IdP's published signing keys — a token that
does not verify is reported as NOT authenticated.
Actions:
- status: Show current authentication state and accessible services
- whoami: Show current user info from IAM token
- login: Sign in via the browser (OAuth2 authorization code + PKCE). Opens a
page at the IdP and waits for the redirect on a local loopback port.
- status: Show verified authentication state and accessible services
- whoami: Show the current user's verified token claims
- logout: Clear stored credentials
- refresh: Refresh an expired token
"""
@@ -53,7 +57,9 @@ class LoginTool(BaseTool):
) -> str:
session = HanzoSession.get()
if action == "status":
if action == "login":
return await self._login(session, **kwargs)
elif action == "status":
return await self._status(session)
elif action == "whoami":
return await self._whoami(session)
@@ -62,7 +68,47 @@ class LoginTool(BaseTool):
elif action == "refresh":
return await self._refresh(session)
else:
return json.dumps({"error": f"Unknown action: {action}. Use: status, whoami, logout, refresh"})
return json.dumps(
{"error": f"Unknown action: {action}. Use: login, status, whoami, logout, refresh"}
)
async def _login(self, session: HanzoSession, **kwargs: Any) -> str:
"""Run the browser login. Blocks until the callback lands or it times out.
Runs in a worker thread: the flow binds a socket and waits, and doing
that on the event loop would wedge the whole MCP server.
"""
import asyncio
from functools import partial
from hanzo_iam.oauth import LoginError
allowed = {"server_url", "client_id", "organization", "scope", "timeout", "open_browser"}
opts = {k: v for k, v in kwargs.items() if k in allowed and v is not None}
urls: list[str] = []
try:
result = await asyncio.to_thread(
partial(session.login, on_url=urls.append, **opts)
)
except LoginError as e:
return json.dumps(
{"authenticated": False, "error": str(e), "authorize_url": urls[0] if urls else None},
indent=2,
)
except Exception as e:
return json.dumps({"authenticated": False, "error": f"{type(e).__name__}: {e}"}, indent=2)
claims = result["claims"]
return json.dumps({
"authenticated": True,
"sub": claims.get("sub"),
"email": claims.get("email"),
"name": claims.get("name"),
"organization": claims.get("owner") or claims.get("organization"),
"expires_at": claims.get("exp"),
"stored_in": result["store"],
}, indent=2)
async def _status(self, session: HanzoSession) -> str:
info = session.get_token_info()
@@ -70,62 +116,49 @@ class LoginTool(BaseTool):
if not info.get("authenticated"):
return json.dumps({
"authenticated": False,
"message": "Not authenticated. Run 'hanzo login' in your terminal to authenticate.",
"services": {
"iam": False,
"kms": _check_kms_env(),
"paas": False,
},
"reason": info.get("reason"),
"detail": info.get("detail"),
"message": "Not authenticated. Run the `login` action to sign in.",
"services": {"iam": False, "kms": _check_kms_env(), "paas": False},
}, indent=2)
services = {
"iam": True,
"kms": _check_kms_env(),
"paas": info.get("authenticated", False),
}
return json.dumps({
"authenticated": True,
"source": info.get("source", "unknown"),
"store": info.get("store"),
"organization": info.get("organization"),
"server_url": info.get("server_url"),
"expires_at": info.get("expires_at"),
"expired": info.get("expired", False),
"services": services,
# IAM is reachable and the token verified; the others are reported
# from what we can actually check, never assumed from the IAM token.
"services": {"iam": True, "kms": _check_kms_env(), "paas": False},
}, indent=2)
async def _whoami(self, session: HanzoSession) -> str:
if not session.is_authenticated():
return json.dumps({"error": "Not authenticated. Run 'hanzo login' first."})
"""Report the identity the ISSUER vouches for.
try:
iam_client = session.get_iam_client()
token = session.get_iam_token()
if token:
# Try to decode JWT claims (without verification for display)
try:
import jwt
claims = jwt.decode(token, options={"verify_signature": False})
return json.dumps({
"sub": claims.get("sub"),
"name": claims.get("name"),
"email": claims.get("email"),
"organization": claims.get("owner"),
"iss": claims.get("iss"),
}, indent=2)
except Exception:
pass
# Fallback to stored token info
info = session.get_token_info()
return json.dumps({
"organization": info.get("organization"),
"server_url": info.get("server_url"),
"source": info.get("source"),
}, indent=2)
except Exception as e:
return json.dumps({"error": f"Failed to get user info: {e}"})
Claims come from `verify()`, so they are only ever shown once the
signature checked out. The previous version decoded the token with
verify_signature=False and printed whatever it said — an attacker-
chosen identity, rendered as fact.
"""
result = session.verify()
if not result.valid:
return json.dumps(
{"error": "Not authenticated.", "reason": result.reason, "detail": result.detail},
indent=2,
)
claims = result.claims
return json.dumps({
"sub": claims.get("sub"),
"name": claims.get("name"),
"email": claims.get("email"),
"organization": claims.get("owner") or claims.get("organization"),
"iss": claims.get("iss"),
"aud": claims.get("aud"),
"exp": claims.get("exp"),
}, indent=2)
async def _logout(self, session: HanzoSession) -> str:
session.logout()
@@ -134,8 +167,10 @@ class LoginTool(BaseTool):
return json.dumps({"message": "Logged out. Cleared stored credentials."})
async def _refresh(self, session: HanzoSession) -> str:
if not session.is_authenticated():
return json.dumps({"error": "Not authenticated. Run 'hanzo login' first."})
# Refresh is exactly what an EXPIRED token needs, so this gates on
# holding a credential, not on that credential still being valid.
if not session.has_credential():
return json.dumps({"error": "No stored credential. Run the `login` action first."})
if session.refresh_token():
return json.dumps({"message": "Token refreshed successfully."})
@@ -154,13 +189,26 @@ class LoginTool(BaseTool):
action: Annotated[
str,
Field(
description="Action: status (check auth state), whoami (current user), logout, refresh",
description=(
"Action: login (browser sign-in), status (verified auth state),"
" whoami (current user), logout, refresh"
),
default="status",
),
] = "status",
timeout: Annotated[
float,
Field(description="login only: seconds to wait for the browser callback", default=300.0),
] = 300.0,
open_browser: Annotated[
bool,
Field(description="login only: open the URL automatically", default=True),
] = True,
ctx: MCPContext = None,
) -> str:
return await tool_instance.call(ctx, action=action)
return await tool_instance.call(
ctx, action=action, timeout=timeout, open_browser=open_browser
)
def _check_kms_env() -> bool:
pkg/hanzo-tools-auth/hanzo_tools/auth/session.py
+168 -50
View File
@@ -1,12 +1,12 @@
"""Hanzo authentication session — shared auth bridge for MCP platform tools.
Loads IAM tokens from disk or environment, auto-refreshes expired tokens,
and provides authenticated service clients (KMS, PaaS, IAM).
Resolves a credential, VERIFIES it, and hands out authenticated service
clients (KMS, PaaS, IAM).
Token resolution order:
Credential resolution order:
1. HANZO_AUTH_TOKEN env var (explicit override)
2. HANZO_API_KEY env var (API key auth)
3. ~/.hanzo/auth/token.json (from `hanzo login`)
2. HANZO_API_KEY env var (opaque API key)
3. The token store (OS keyring, else ~/.hanzo/auth/token.json at 0600)
"""
from __future__ import annotations
@@ -18,20 +18,30 @@ import logging
from typing import Any
from pathlib import Path
from hanzo_iam import store
from hanzo_iam.oauth import DEFAULT_CLIENT_ID, DEFAULT_IAM_URL, DEFAULT_ORG
from hanzo_iam.models import OIDC_JWKS_PATH, OIDC_USERINFO_PATH
from hanzo_iam.tokens import (
BAD_SIGNATURE,
JWKS_UNREACHABLE,
NO_CREDENTIAL,
OK,
OPAQUE,
Verification,
)
from hanzo_iam.tokens import verify as verify_jwt
logger = logging.getLogger(__name__)
TOKEN_DIR = Path.home() / ".hanzo" / "auth"
TOKEN_FILE = TOKEN_DIR / "token.json"
# IAM defaults (same as hanzo-cli)
DEFAULT_IAM_URL = "https://hanzo.id"
DEFAULT_ORG = "hanzo"
DEFAULT_APP = "hanzo-app"
DEFAULT_CLIENT_ID = "hanzo-app"
# Kept for callers that report where credentials live. The store owns writing.
TOKEN_DIR = store.TOKEN_DIR
TOKEN_FILE = store.TOKEN_FILE
def _env(name: str) -> str:
"""Read env var (IAM_*)."""
"""Read an IAM_* env var, normalising absence to the empty string."""
return os.getenv(name) or ""
@@ -62,19 +72,12 @@ class HanzoSession:
# -- Token loading -------------------------------------------------------
def _load_token_from_disk(self) -> dict[str, Any] | None:
"""Load stored token from ~/.hanzo/auth/token.json."""
if not TOKEN_FILE.exists():
return None
try:
return json.loads(TOKEN_FILE.read_text())
except (json.JSONDecodeError, OSError):
return None
"""Load stored token from the credential store."""
return store.load()
def _save_token(self, data: dict[str, Any]) -> None:
"""Save token data to disk."""
TOKEN_DIR.mkdir(parents=True, exist_ok=True)
TOKEN_FILE.write_text(json.dumps(data, indent=2))
TOKEN_FILE.chmod(0o600)
"""Persist token data. The store picks keyring or an atomic 0600 file."""
store.save(data)
def load_token(self) -> dict[str, Any] | None:
"""Load token using the resolution chain.
@@ -105,7 +108,7 @@ class HanzoSession:
# 3. Stored token from `hanzo login`
token_data = self._load_token_from_disk()
if token_data and token_data.get("access_token"):
token_data["source"] = "disk:~/.hanzo/auth/token.json"
token_data["source"] = f"store:{store.backend()}"
self._token_data = token_data
return self._token_data
@@ -113,10 +116,63 @@ class HanzoSession:
# -- Token state ---------------------------------------------------------
def is_authenticated(self) -> bool:
"""Check if we have a valid token."""
def has_credential(self) -> bool:
"""Report whether ANY credential is present. Says nothing about validity."""
return self.load_token() is not None
def verify(self) -> Verification:
"""Judge the held credential against the issuer's published keys.
This is the real check. `is_authenticated()` used to be
`load_token() is not None`, which returned True for the literal string
"fake.not.a.real.jwt" and made every downstream authorization decision
in this package meaningless.
Fails CLOSED: if the JWKS cannot be fetched we do not know the token is
good, so we do not claim it is. The reason code distinguishes that from
an actually-bad token.
"""
token_data = self.load_token()
if not token_data:
return Verification(False, NO_CREDENTIAL, "no credential found")
token = token_data.get("access_token") or ""
server_url = (token_data.get("server_url") or DEFAULT_IAM_URL).rstrip("/")
result = verify_jwt(
token,
jwks_uri=f"{server_url}{OIDC_JWKS_PATH}",
issuer=server_url,
)
if result.reason == OPAQUE:
# An API key is a real credential that simply cannot be judged
# offline. Ask the issuer instead of guessing.
return self._verify_opaque(token, server_url)
return result
def _verify_opaque(self, token: str, server_url: str) -> Verification:
"""Confirm an opaque credential by calling userinfo — the authority."""
import httpx
try:
resp = httpx.get(
f"{server_url}{OIDC_USERINFO_PATH}",
headers={"Authorization": f"Bearer {token}"},
timeout=10.0,
)
except httpx.HTTPError as e:
return Verification(False, JWKS_UNREACHABLE, f"cannot reach {server_url}: {e}")
if resp.status_code == 200 and "json" in resp.headers.get("content-type", ""):
return Verification(True, OK, claims=resp.json())
return Verification(
False,
BAD_SIGNATURE,
f"issuer rejected the credential ({resp.status_code})",
)
def is_authenticated(self) -> bool:
"""True only when the held credential actually verifies."""
return self.verify().valid
def get_iam_token(self) -> str | None:
"""Get the current IAM access token."""
token_data = self.load_token()
@@ -125,23 +181,37 @@ class HanzoSession:
return None
def get_token_info(self) -> dict[str, Any]:
"""Get info about the current auth state."""
"""Describe the current auth state. `authenticated` reflects VERIFICATION.
It used to reflect credential presence, so `auth status` cheerfully
reported a garbage token as a working login.
"""
token_data = self.load_token()
if not token_data:
return {"authenticated": False}
return {"authenticated": False, "reason": NO_CREDENTIAL}
result = self.verify()
info: dict[str, Any] = {
"authenticated": True,
"authenticated": result.valid,
"reason": result.reason,
"source": token_data.get("source", "unknown"),
"store": store.backend(),
}
if not result.valid:
info["detail"] = result.detail
# Check expiry
login_time = token_data.get("login_time", 0)
expires_in = token_data.get("expires_in", 0)
if login_time and expires_in:
expires_at = login_time + expires_in
info["expires_at"] = expires_at
info["expired"] = time.time() > expires_at
# Expiry, from the token's own exp when we could read it, else the
# locally recorded issue time.
exp = result.claims.get("exp")
if exp:
info["expires_at"] = int(exp)
info["expired"] = time.time() > float(exp)
else:
login_time = token_data.get("login_time", 0)
expires_in = token_data.get("expires_in", 0)
if login_time and expires_in:
info["expires_at"] = login_time + expires_in
info["expired"] = time.time() > info["expires_at"]
# Add org/app info if available
if token_data.get("organization"):
@@ -190,7 +260,7 @@ class HanzoSession:
self._save_token(new_data)
self._token_data = new_data
self._token_data["source"] = "disk:~/.hanzo/auth/token.json"
self._token_data["source"] = f"store:{store.backend()}"
logger.info("Token refreshed successfully")
return True
@@ -247,7 +317,7 @@ class HanzoSession:
from hanzo_kms import KMSClient
# KMSClient() reads HANZO_KMS_URL / _ORG / _CLIENT_ID / _CLIENT_SECRET
# / _TOKEN itself — see hanzo_kms.settings_from_env.
# / _TOKEN itself -- see hanzo_kms.settings_from_env.
self._kms_client = KMSClient()
return self._kms_client
@@ -279,16 +349,39 @@ class HanzoSession:
except Exception:
pass
# Exchange IAM token for PaaS session
# Exchange IAM token for PaaS session.
#
# MEASURED, not assumed: platform.hanzo.ai answers POST /v1/auth/login
# with a bodyless 404, while /v1/org and /v1/user answer 401 — so the
# PaaS API is up and gated, but the token-exchange route in front of it
# is not deployed. The paas repo does mount it (platform/server.js:62 →
# routes/auth.js) and it expects {gitUser:{provider,providerUserId,...}},
# not the {provider,accessToken} this client used to send. Two faults,
# and the deployment one cannot be fixed from here.
#
# So: fail with the actual diagnosis. Guessing a payload against a route
# that 404s would just move the confusion downstream.
with httpx.Client(base_url=base_url, timeout=30.0) as tmp:
resp = tmp.post(
"/v1/auth/login",
json={"provider": "hanzo", "accessToken": iam_token},
headers={"Content-Type": "application/json"},
)
if resp.status_code == 404:
raise RuntimeError(
f"PaaS token exchange is not deployed: POST {base_url}/v1/auth/login"
" returned 404 while /v1/org returns 401. The route exists in the"
" paas repo (platform/server.js -> routes/auth.js) but is not"
" reachable at this edge. No client-side workaround exists."
)
if resp.status_code == 401:
raise RuntimeError("IAM token rejected by PaaS. Try 'hanzo login' again.")
raise RuntimeError("IAM token rejected by PaaS. Sign in again.")
resp.raise_for_status()
if "json" not in resp.headers.get("content-type", ""):
raise RuntimeError(
f"PaaS login returned {resp.headers.get('content-type')} instead of"
" JSON — the request reached a web page, not the API."
)
data = resp.json()
at = data.get("at", "")
@@ -296,12 +389,13 @@ class HanzoSession:
if not at:
raise RuntimeError("PaaS login succeeded but no session token returned.")
# Cache session
session_dir = Path.home() / ".hanzo" / "paas"
session_dir.mkdir(parents=True, exist_ok=True)
session_file = session_dir / "session.json"
session_file.write_text(json.dumps({"at": at, "rt": rt, "login_time": int(time.time())}, indent=2))
session_file.chmod(0o600)
# Cache session under the same 0600-atomic discipline as the IAM token —
# a PaaS session cookie is a bearer credential too.
session_file = Path.home() / ".hanzo" / "paas" / "session.json"
store._write_private(
session_file,
json.dumps({"at": at, "rt": rt, "login_time": int(time.time())}, indent=2),
)
return _PaaSClientWrapper(base_url, at, rt)
@@ -319,11 +413,35 @@ class HanzoSession:
# -- Logout --------------------------------------------------------------
def login(self, **kwargs: Any) -> dict[str, Any]:
"""Run the interactive login, persist the result, and return its info.
Verification happens BEFORE the token is stored: a token we cannot
check is not one we should keep and later present as proof of identity.
"""
from hanzo_iam import oauth
token_data = oauth.login(**kwargs)
server_url = token_data["server_url"]
result = verify_jwt(
token_data["access_token"],
jwks_uri=f"{server_url}{OIDC_JWKS_PATH}",
issuer=server_url,
)
if not result.valid:
raise RuntimeError(
f"IAM issued a token this client cannot verify ({result.reason}:"
f" {result.detail}). Refusing to store it."
)
self._save_token(token_data)
self.close()
self._token_data = None
return {"claims": result.claims, "store": store.backend()}
@staticmethod
def logout() -> None:
"""Clear stored credentials."""
if TOKEN_FILE.exists():
TOKEN_FILE.unlink()
"""Clear stored credentials from every backend."""
store.clear()
session_file = Path.home() / ".hanzo" / "paas" / "session.json"
if session_file.exists():
session_file.unlink()
pkg/hanzo-tools-auth/pyproject.toml
+2 -2
View File
@@ -1,6 +1,6 @@
[project]
name = "hanzo-tools-auth"
version = "0.1.1"
version = "0.1.2"
description = "Authentication bridge for Hanzo MCP platform tools — session management, token refresh, service client factory"
readme = "README.md"
requires-python = ">=3.12"
@@ -17,7 +17,7 @@ classifiers = [
dependencies = [
"hanzo-tools-core>=0.1.0",
"hanzo-iam>=1.30.0",
"hanzo-iam>=1.30.2",
"httpx>=0.27.0",
"pyjwt>=2.8.0",
]
pkg/hanzo-tools-auth/tests/test_session_auth.py
+265
View File
@@ -0,0 +1,265 @@
"""Tests for HanzoSession's authentication state.
`test_fake_token_is_not_authenticated` is THE regression. Against the shipped
implementation —
def is_authenticated(self) -> bool:
return self.load_token() is not None
— it FAILS: a stored token of "fake.not.a.real.jwt" made it return True, and
every tool that gated on it believed the caller was signed in.
"""
from __future__ import annotations
import json
import time
import jwt
import pytest
from cryptography.hazmat.primitives.asymmetric import rsa
from hanzo_iam import store
from hanzo_iam import tokens
from hanzo_tools.auth.session import HanzoSession
ISSUER = "https://hanzo.id"
FAKE = "fake.not.a.real.jwt"
@pytest.fixture(scope="module")
def keypair():
return rsa.generate_private_key(public_exponent=65537, key_size=2048)
def _jwks(key, kid="test-key"):
data = json.loads(jwt.algorithms.RSAAlgorithm.to_jwk(key.public_key()))
data.update({"kid": kid, "use": "sig", "alg": "RS256"})
return {"keys": [data]}
def _token(key, **overrides):
claims = {
"iss": ISSUER,
"sub": "hanzo/z",
"aud": "hanzo-app",
"exp": int(time.time()) + 3600,
"iat": int(time.time()),
}
claims.update(overrides)
return jwt.encode(claims, key, algorithm="RS256", headers={"kid": "test-key"})
@pytest.fixture(autouse=True)
def clean_session(tmp_path, monkeypatch):
"""Isolate the store, clear env credentials, reset the singleton."""
monkeypatch.setattr(store, "TOKEN_DIR", tmp_path / "auth")
monkeypatch.setattr(store, "TOKEN_FILE", tmp_path / "auth" / "token.json")
monkeypatch.setattr(store, "_keyring", lambda: None)
monkeypatch.delenv("HANZO_AUTH_TOKEN", raising=False)
monkeypatch.delenv("HANZO_API_KEY", raising=False)
tokens.reset_jwks_cache()
HanzoSession.reset()
yield
HanzoSession.reset()
tokens.reset_jwks_cache()
@pytest.fixture
def serve_jwks(monkeypatch):
def _serve(document):
monkeypatch.setattr(jwt.PyJWKClient, "fetch_data", lambda self: document)
return _serve
@pytest.fixture
def no_network(monkeypatch):
"""Any HTTP call is a test bug unless the test opted into one."""
import httpx
def forbidden(*a, **kw):
raise AssertionError("unexpected network call")
monkeypatch.setattr(httpx, "get", forbidden)
def _store(token: str, **extra):
store.save({"access_token": token, "server_url": ISSUER, **extra})
# --- The defect -----------------------------------------------------------
def test_fake_token_is_not_authenticated(monkeypatch):
"""A stored token of 'fake.not.a.real.jwt' must not authenticate anybody.
It is not a JWT, so it is treated as an opaque credential and put to the
issuer, which rejects it. Under the old presence check this returned True.
"""
import httpx
class Rejected:
status_code = 401
headers = {"content-type": "application/json"}
monkeypatch.setattr(httpx, "get", lambda *a, **kw: Rejected())
_store(FAKE)
session = HanzoSession.get()
assert session.has_credential() is True # the string IS there
assert session.is_authenticated() is False # but it proves nothing
assert session.get_token_info()["authenticated"] is False
def test_fake_token_in_env_is_not_authenticated(monkeypatch):
"""Same defect via HANZO_AUTH_TOKEN, which bypasses the store entirely."""
import httpx
class Rejected:
status_code = 401
headers = {"content-type": "application/json"}
monkeypatch.setattr(httpx, "get", lambda *a, **kw: Rejected())
monkeypatch.setenv("HANZO_AUTH_TOKEN", FAKE)
assert HanzoSession.get().is_authenticated() is False
def test_presence_and_validity_are_different_questions(serve_jwks, keypair, no_network):
"""The two used to be one method. Keeping them apart is the fix."""
serve_jwks(_jwks(keypair))
_store(_token(keypair, exp=int(time.time()) - 3600))
session = HanzoSession.get()
assert session.has_credential() is True
assert session.is_authenticated() is False
assert session.verify().reason == tokens.EXPIRED
# --- Genuine credentials --------------------------------------------------
def test_real_token_authenticates(serve_jwks, keypair, no_network):
serve_jwks(_jwks(keypair))
_store(_token(keypair))
session = HanzoSession.get()
assert session.is_authenticated() is True
info = session.get_token_info()
assert info["authenticated"] is True
assert info["expired"] is False
assert session.verify().claims["sub"] == "hanzo/z"
def test_no_credential_at_all(no_network):
session = HanzoSession.get()
assert session.has_credential() is False
assert session.is_authenticated() is False
assert session.get_token_info() == {
"authenticated": False,
"reason": tokens.NO_CREDENTIAL,
}
# --- Forgeries ------------------------------------------------------------
def test_token_from_another_issuer_is_rejected(serve_jwks, keypair, no_network):
"""Correctly signed, wrong issuer: still not a Hanzo session."""
serve_jwks(_jwks(keypair))
_store(_token(keypair, iss="https://evil.example"))
assert HanzoSession.get().verify().reason == tokens.WRONG_ISSUER
def test_self_signed_token_is_rejected(serve_jwks, keypair, no_network):
other = rsa.generate_private_key(public_exponent=65537, key_size=2048)
serve_jwks(_jwks(keypair))
_store(_token(other))
assert HanzoSession.get().is_authenticated() is False
def test_expiry_comes_from_the_token_not_from_local_bookkeeping(
serve_jwks, keypair, no_network
):
"""A caller could otherwise claim a fresh session by editing login_time in
the stored JSON. `exp` is signed; login_time is not."""
serve_jwks(_jwks(keypair))
exp = int(time.time()) + 60
store.save(
{
"access_token": _token(keypair, exp=exp),
"server_url": ISSUER,
"login_time": int(time.time()) + 10**6,
"expires_in": 10**6,
}
)
assert HanzoSession.get().get_token_info()["expires_at"] == exp
# --- Fail closed ----------------------------------------------------------
def test_offline_does_not_grant_access(monkeypatch, keypair):
"""No JWKS, no verification, no session — an unreachable IdP must not be a
free pass."""
def down(self):
raise jwt.exceptions.PyJWKClientConnectionError("network down")
monkeypatch.setattr(jwt.PyJWKClient, "fetch_data", down)
_store(_token(keypair))
session = HanzoSession.get()
assert session.is_authenticated() is False
assert session.verify().reason == tokens.JWKS_UNREACHABLE
def test_jwks_path_carries_the_v1_iam_prefix(serve_jwks, keypair, monkeypatch):
"""The unprefixed /.well-known/jwks answers 200 text/html on hanzo.id, so
verification would silently have nothing to check against."""
seen: list[str] = []
monkeypatch.setattr(
jwt.PyJWKClient,
"fetch_data",
lambda self: (seen.append(self.uri), _jwks(keypair))[1],
)
_store(_token(keypair))
HanzoSession.get().verify()
assert seen == [f"{ISSUER}/v1/iam/.well-known/jwks"]
# --- Storage --------------------------------------------------------------
def test_session_persists_tokens_privately(serve_jwks, keypair, no_network):
import stat
serve_jwks(_jwks(keypair))
HanzoSession.get()._save_token({"access_token": _token(keypair), "server_url": ISSUER})
assert stat.S_IMODE(store.TOKEN_FILE.stat().st_mode) == 0o600
def test_logout_clears_the_credential(serve_jwks, keypair, no_network):
serve_jwks(_jwks(keypair))
_store(_token(keypair))
session = HanzoSession.get()
assert session.is_authenticated() is True
session.logout()
HanzoSession.reset()
assert HanzoSession.get().is_authenticated() is False
def test_login_refuses_to_store_a_token_it_cannot_verify(monkeypatch, keypair):
"""If IAM hands back something unverifiable, that is a bug to surface, not
a credential to keep and present as proof of identity later."""
from hanzo_iam import oauth
monkeypatch.setattr(
oauth,
"login",
lambda **kw: {"access_token": FAKE, "server_url": ISSUER, "client_id": "hanzo-app"},
)
monkeypatch.setattr(
jwt.PyJWKClient, "fetch_data", lambda self: _jwks(keypair)
)
with pytest.raises(RuntimeError, match="cannot verify"):
HanzoSession.get().login()
assert store.load() is None
pkg/hanzo-tools-team/hanzo_tools/team/team_tool.py
+26 -3
View File
@@ -23,7 +23,22 @@ from hanzo_tools.core.base import BaseTool
logger = logging.getLogger(__name__)
TEAM_BASE_URL = os.getenv("HANZO_TEAM_URL", "https://api.hanzo.ai/team")
# There is no Hanzo Team API. MEASURED against the live edge:
# GET https://api.hanzo.ai/team/... -> 200 text/html (the marketing SPA)
# GET https://api.hanzo.ai/v1/team/... -> 404
# GET https://api.hanzo.ai/v1/workspaces -> 404
# GET https://team.hanzo.ai/... -> 200 text/html (a different SPA)
# The old default pointed at the first of those, so every call here sailed
# through raise_for_status() on a 200 and then died inside .json() — a missing
# service reported as a parse error.
#
# What DOES exist and is auth-gated is IAM's tenancy surface:
# /v1/iam/get-organizations and /v1/iam/get-groups. Mapping "workspace" onto
# org-or-group is a product decision, not something to guess at here.
#
# HANZO_TEAM_URL stays as the seam: set it and this tool works the day the
# service ships. Unset, it refuses rather than inventing an answer.
TEAM_BASE_URL = os.getenv("HANZO_TEAM_URL", "")
DESCRIPTION = """Hanzo Team — workspace and member management.
@@ -47,8 +62,16 @@ def _get_session():
def _team_url(path: str) -> str:
"""Build full Team API URL."""
return f"{TEAM_BASE_URL}/{path.lstrip('/')}"
"""Build the full Team API URL, or refuse because there is no Team API."""
if not TEAM_BASE_URL:
raise RuntimeError(
"No Hanzo Team API exists. api.hanzo.ai/team serves the marketing"
" site (200 text/html) and /v1/team 404s. For org and group"
" membership use the `iam` tool (/v1/iam/get-organizations,"
" /v1/iam/get-groups). Set HANZO_TEAM_URL to point this tool at a"
" real Team service once one is deployed."
)
return f"{TEAM_BASE_URL.rstrip('/')}/{path.lstrip('/')}"
def _auth_headers(token: str) -> dict[str, str]:
pkg/hanzo-tools-team/pyproject.toml
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "hanzo-tools-team"
version = "0.1.0"
version = "0.1.1"
description = "Hanzo MCP tool for Team — workspaces, members, and invitations"
readme = "README.md"
requires-python = ">=3.12"