Files
zeekay 28035b6721 hanzo-iam: one client, one env reader, one verifier
The credential rode in the query string on AsyncIAMClient too — the
exported one — across seventeen call sites. Same fix as the sync client:
Authorization, Basic for a confidential client.

AsyncIAMClient was DEFINED TWICE, in client.py and async_client.py, and
the copies had drifted to nineteen methods against thirty-eight. __init__
exports the one in async_client.py, so the other was unreachable; it is
gone.

Three readers of IAM_* disagreed about the tenant: IAMConfig.from_env and
fastapi.configure both defaulted the org to the literal "hanzo", so a Lux
or Zoo process that forgot IAM_ORG did not fail — it addressed the hanzo
tenant and reported hanzo's users as its own. Now one reader,
IAMConfig.from_env(org), which seeds the endpoint and the tenant from the
org it is given; organization has no default at all. A test walks the
package and asserts config.py is the only module that reads the
environment.

IAMClient.validate_token and validate_token_with_cert were a second,
weaker verifier — no issuer check, no require:["exp"]. tokens.verify does
both and fails closed, so the strict one is what survives.

118 pass, and the seventeen tests of the two consumers that build on it.
2026-08-23 12:51:30 -07:00

105 lines
3.7 KiB
Python

"""Configuration for Hanzo IAM client."""
from __future__ import annotations
import os
from typing import ClassVar
from pydantic import BaseModel, ConfigDict, Field
from hanzo_iam.models import (
OIDC_AUTHORIZE_PATH,
Organization,
OIDC_DEVICE_PATH,
OIDC_JWKS_PATH,
OIDC_TOKEN_PATH,
OIDC_USERINFO_PATH,
)
class IAMConfig(BaseModel):
"""Configuration for Hanzo IAM client.
Can be initialized directly or from environment variables via from_env().
"""
model_config = ConfigDict(frozen=True)
# Canonical environment variable prefix. There is exactly one prefix —
# no upstream-brand aliases, no per-org fallbacks. See
# ~/work/hanzo/iam/CLAUDE.md "Configuration" section.
ENV_PREFIX: ClassVar[str] = "IAM_"
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(description="IAM organization name — the tenant this client acts in")
application: str = Field(default="app", description="IAM application name")
certificate: str = Field(
default="", description="JWT verification certificate (PEM)"
)
@classmethod
def from_env(cls, org: Organization = Organization.HANZO, prefix: str | None = None) -> IAMConfig:
"""Read configuration from the environment. The one reader.
The canonical prefix is ``IAM_``: no upstream-brand aliases, no per-org
variants. ``org`` seeds the endpoint and the tenant when the environment
names neither, so a Zoo process that exports nothing reaches zoo.id as
zoo rather than hanzo.id as hanzo. ``prefix`` exists for a process that
scopes more than one IAM client; new code leaves it alone.
IAM_ENDPOINT server URL, default ``org.iam_url``
IAM_CLIENT_ID OAuth2 client id
IAM_CLIENT_SECRET OAuth2 client secret
IAM_ORG tenant, default ``org.value``
IAM_APP application, default ``app``
IAM_CERT JWT verification certificate, content or path
"""
p = prefix or cls.ENV_PREFIX
cert = os.environ.get(f"{p}CERT", "")
if cert and not cert.startswith("-----BEGIN"):
path = os.path.expanduser(cert)
if os.path.isfile(path):
cert = open(path).read()
return cls(
server_url=os.environ.get(f"{p}ENDPOINT", org.iam_url),
client_id=os.environ.get(f"{p}CLIENT_ID", ""),
client_secret=os.environ.get(f"{p}CLIENT_SECRET", ""),
organization=os.environ.get(f"{p}ORG", org.value),
application=os.environ.get(f"{p}APP", "app"),
certificate=cert,
)
@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."""
return f"{self.base_url}{OIDC_TOKEN_PATH}"
@property
def authorize_endpoint(self) -> str:
"""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.base_url}{OIDC_USERINFO_PATH}"