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

hanzo-iam: one client, one env reader, one verifier

Browse Source 
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.
This commit is contained in:
zeekay
2026-08-23 12:51:30 -07:00
parent aa18787d58
commit 28035b6721
7 changed files with 127 additions and 510 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/hanzo-iam/hanzo_iam/async_client.py
+8 -29
View File
@@ -9,6 +9,7 @@ from urllib.parse import urlencode
import httpx
import jwt
from hanzo_iam.client import basic
from hanzo_iam.config import IAMConfig
from hanzo_iam import routes
from hanzo_iam.models import (
@@ -80,7 +81,7 @@ class AsyncIAMClient:
if config:
self._config = config
else:
env_config = IAMClient._config_from_env(org)
env_config = IAMConfig.from_env(org)
self._config = IAMConfig(
server_url=env_config.server_url,
client_id=client_id or env_config.client_id,
@@ -117,20 +118,15 @@ class AsyncIAMClient:
# Admin Auth Helpers
# =========================================================================
def _admin_params(self) -> dict[str, str]:
"""Return query params for admin API auth (empty if using bearer token)."""
if self._bearer_token:
return {}
return {
"clientId": self._config.client_id,
"clientSecret": self._config.client_secret,
}
def _admin_headers(self) -> dict[str, str]:
"""Return extra headers for admin API auth (Authorization if bearer token)."""
"""The credential, in Authorization — the only place IAM reads one.
A bearer if we hold one, otherwise the confidential client's own pair as
HTTP Basic (RFC 6749 2.3.1).
"""
if self._bearer_token:
return {"Authorization": f"Bearer {self._bearer_token}"}
return {}
return {"Authorization": basic(self._config.client_id, self._config.client_secret)}
# =========================================================================
# OIDC Discovery
@@ -443,7 +439,6 @@ class AsyncIAMClient:
"""
params = {
"id": f"{self._config.organization}/{user_id}",
**self._admin_params(),
}
http = await self._get_http()
@@ -468,7 +463,6 @@ class AsyncIAMClient:
"""
params = {
"owner": self._config.organization,
**self._admin_params(),
}
http = await self._get_http()
@@ -503,7 +497,6 @@ class AsyncIAMClient:
"""
params: dict[str, Any] = {
"owner": owner or self._config.organization,
**self._admin_params(),
}
if is_online is not None:
params["isOnline"] = str(is_online).lower()
@@ -560,7 +553,6 @@ class AsyncIAMClient:
http = await self._get_http()
response = await http.post(
f"{IAM_ROUTE_PREFIX}/{action}",
params=self._admin_params(),
headers=self._admin_headers(),
json=user.model_dump(by_alias=True, exclude_none=True),
)
@@ -580,7 +572,6 @@ class AsyncIAMClient:
"""
params = {
"id": f"{self._config.organization}/{self._config.application}",
**self._admin_params(),
}
http = await self._get_http()
@@ -609,7 +600,6 @@ class AsyncIAMClient:
"""
params = {
"owner": "admin",
**self._admin_params(),
}
http = await self._get_http()
@@ -637,7 +627,6 @@ class AsyncIAMClient:
"""
params = {
"id": f"admin/{name}",
**self._admin_params(),
}
http = await self._get_http()
@@ -687,7 +676,6 @@ class AsyncIAMClient:
"resourceId": resource_id,
"enforceId": enforce_id,
"owner": owner or self._config.organization,
**self._admin_params(),
}
if request:
payload["casbinRequest"] = request
@@ -732,7 +720,6 @@ class AsyncIAMClient:
"modelId": model_id,
"enforceId": enforce_id,
"owner": owner or self._config.organization,
**self._admin_params(),
}
if requests:
payload["casbinRequest"] = requests
@@ -767,7 +754,6 @@ class AsyncIAMClient:
"""
params = {
"owner": owner or self._config.organization,
**self._admin_params(),
}
http = await self._get_http()
@@ -802,7 +788,6 @@ class AsyncIAMClient:
org = owner or self._config.organization
params = {
"id": f"{org}/{role_name}",
**self._admin_params(),
}
http = await self._get_http()
@@ -837,7 +822,6 @@ class AsyncIAMClient:
org = owner or self._config.organization
params = {
"id": f"{org}/{username}",
**self._admin_params(),
}
http = await self._get_http()
@@ -875,7 +859,6 @@ class AsyncIAMClient:
payload = {
"user": f"{org}/{username}",
"role": f"{org}/{role_name}",
**self._admin_params(),
}
http = await self._get_http()
@@ -913,7 +896,6 @@ class AsyncIAMClient:
payload = {
"user": f"{org}/{username}",
"role": f"{org}/{role_name}",
**self._admin_params(),
}
http = await self._get_http()
@@ -962,7 +944,6 @@ class AsyncIAMClient:
http = await self._get_http()
response = await http.post(
f"{IAM_ROUTE_PREFIX}/set-password",
params=self._admin_params(),
headers=self._admin_headers(),
json=payload,
)
@@ -989,7 +970,6 @@ class AsyncIAMClient:
"""
params = {
"owner": owner,
**self._admin_params(),
}
http = await self._get_http()
@@ -1019,7 +999,6 @@ class AsyncIAMClient:
http = await self._get_http()
response = await http.post(
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
+1 -382
View File
@@ -97,7 +97,7 @@ class IAMClient:
if config:
self._config = config
else:
env_config = self._config_from_env(org)
env_config = IAMConfig.from_env(org)
self._config = IAMConfig(
server_url=env_config.server_url,
client_id=client_id or env_config.client_id,
@@ -112,33 +112,6 @@ class IAMClient:
self._jwks_client: PyJWKClient | None = None
self._openid_config: dict | None = None
@staticmethod
def _config_from_env(org: Organization = Organization.HANZO) -> IAMConfig:
"""Read configuration from environment variables.
The canonical contract is ``IAM_*`` only — no upstream-brand
aliases, no per-org fallbacks (see ~/work/hanzo/iam/CLAUDE.md
"Configuration"). The ``org`` argument seeds the default server
URL when ``IAM_ENDPOINT`` is unset and the default organization
name when ``IAM_ORG`` is unset; it does not influence which env
vars are read.
Environment variables:
IAM_ENDPOINT - IAM server URL (defaults to ``org.iam_url``)
IAM_CLIENT_ID - OAuth2 client ID
IAM_CLIENT_SECRET - OAuth2 client secret
IAM_ORG - Organization name (defaults to ``org.value``)
IAM_APP - Application name (defaults to ``app``)
IAM_CERT - JWT verification certificate
"""
return IAMConfig(
server_url=os.getenv("IAM_ENDPOINT", org.iam_url),
client_id=os.getenv("IAM_CLIENT_ID", ""),
client_secret=os.getenv("IAM_CLIENT_SECRET", ""),
organization=os.getenv("IAM_ORG", org.value),
application=os.getenv("IAM_APP", "app"),
certificate=os.getenv("IAM_CERT", ""),
)
@property
def http(self) -> httpx.Client:
@@ -324,83 +297,7 @@ class IAMClient:
# Token Validation
# =========================================================================
def validate_token(
self,
token: str,
verify_exp: bool = True,
verify_aud: bool = True,
) -> JWTClaims:
"""Validate JWT token using JWKS.
Args:
token: JWT access token or ID token
verify_exp: Verify expiration (default: True)
verify_aud: Verify audience matches client_id (default: True)
Returns:
JWTClaims with decoded token claims.
Raises:
jwt.InvalidTokenError: If token is invalid or expired.
"""
# Initialize JWKS client if needed
if self._jwks_client is None:
jwks_url = self._config.jwks_uri
self._jwks_client = jwt.PyJWKClient(jwks_url)
# Get signing key from JWKS
signing_key = self._jwks_client.get_signing_key_from_jwt(token)
# Build verification options
options = {
"verify_exp": verify_exp,
"verify_aud": verify_aud,
}
audience = self._config.client_id if verify_aud else None
# Decode and validate
claims = jwt.decode(
token,
signing_key.key,
algorithms=["RS256", "ES256"],
audience=audience,
options=options,
)
return JWTClaims.model_validate(claims)
def validate_token_with_cert(
self,
token: str,
verify_exp: bool = True,
) -> JWTClaims:
"""Validate JWT token using configured certificate.
Use this when you have the public certificate configured.
Args:
token: JWT access token or ID token
verify_exp: Verify expiration (default: True)
Returns:
JWTClaims with decoded token claims.
"""
if not self._config.certificate:
raise ValueError(
"Certificate not configured. Use validate_token() with JWKS instead."
)
options = {"verify_exp": verify_exp}
claims = jwt.decode(
token,
self._config.certificate,
algorithms=["RS256"],
options=options,
)
return JWTClaims.model_validate(claims)
def introspect_token(self, token: str) -> dict:
"""Introspect token at IAM server.
@@ -847,281 +744,3 @@ class IAMClient:
self.close()
class AsyncIAMClient:
"""
Async OAuth2/OIDC client for Hanzo IAM.
Same interface as IAMClient but uses async/await.
"""
def __init__(
self,
client_id: str | None = None,
client_secret: str | None = None,
org: Organization = Organization.HANZO,
config: IAMConfig | None = None,
):
"""Initialize async IAM client."""
if config:
self._config = config
else:
env_config = IAMClient._config_from_env(org)
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,
organization=env_config.organization,
application=env_config.application,
certificate=env_config.certificate,
)
self._http: httpx.AsyncClient | None = None
self._jwks_client: PyJWKClient | None = None
self._openid_config: dict | None = None
@property
def config(self) -> IAMConfig:
"""Get client configuration."""
return self._config
async def _get_http(self) -> httpx.AsyncClient:
"""Get or create async HTTP client."""
if self._http is None:
self._http = httpx.AsyncClient(
base_url=self._config.server_url.rstrip("/"),
timeout=30.0,
headers={
"User-Agent": "hanzo-iam-python/1.0",
"Content-Type": "application/json",
},
)
return self._http
async def get_openid_configuration(self) -> dict:
"""Get OpenID Connect discovery document."""
if self._openid_config is None:
http = await self._get_http()
response = await http.get(OIDC_DISCOVERY_PATH)
response.raise_for_status()
self._openid_config = response.json()
return self._openid_config
async def get_jwks(self) -> dict:
"""Get JSON Web Key Set."""
http = await self._get_http()
response = await http.get(OIDC_JWKS_PATH)
response.raise_for_status()
return response.json()
def get_authorization_url(
self,
redirect_uri: str,
state: str | None = None,
scope: str = "openid profile email",
response_type: str = "code",
nonce: str | None = None,
code_challenge: str | None = None,
code_challenge_method: str | None = None,
) -> str:
"""Build authorization URL."""
if state is None:
state = secrets.token_urlsafe(32)
params = {
"client_id": self._config.client_id,
"redirect_uri": redirect_uri,
"response_type": response_type,
"scope": scope,
"state": state,
}
if nonce:
params["nonce"] = nonce
if code_challenge:
params["code_challenge"] = code_challenge
params["code_challenge_method"] = code_challenge_method or "S256"
base_url = self._config.server_url.rstrip("/")
return f"{base_url}{OIDC_AUTHORIZE_PATH}?{urlencode(params)}"
async def exchange_code(
self,
code: str,
redirect_uri: str,
code_verifier: str | None = None,
) -> TokenResponse:
"""Exchange authorization code for tokens."""
data = {
"grant_type": "authorization_code",
"client_id": self._config.client_id,
"client_secret": self._config.client_secret,
"code": code,
"redirect_uri": redirect_uri,
}
if code_verifier:
data["code_verifier"] = code_verifier
http = await self._get_http()
response = await http.post(
OIDC_TOKEN_PATH,
data=data,
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
response.raise_for_status()
return TokenResponse.model_validate(response.json())
async def refresh_token(self, refresh_token: str) -> TokenResponse:
"""Refresh access token."""
data = {
"grant_type": "refresh_token",
"client_id": self._config.client_id,
"client_secret": self._config.client_secret,
"refresh_token": refresh_token,
}
http = await self._get_http()
response = await http.post(
OIDC_TOKEN_PATH,
data=data,
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
response.raise_for_status()
return TokenResponse.model_validate(response.json())
async def client_credentials(self, scope: str = "openid") -> TokenResponse:
"""Get access token using client credentials."""
data = {
"grant_type": "client_credentials",
"client_id": self._config.client_id,
"client_secret": self._config.client_secret,
"scope": scope,
}
http = await self._get_http()
response = await http.post(
OIDC_TOKEN_PATH,
data=data,
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
response.raise_for_status()
return TokenResponse.model_validate(response.json())
def validate_token(
self,
token: str,
verify_exp: bool = True,
verify_aud: bool = True,
) -> JWTClaims:
"""Validate JWT token using JWKS (sync operation)."""
if self._jwks_client is None:
jwks_url = self._config.jwks_uri
self._jwks_client = jwt.PyJWKClient(jwks_url)
signing_key = self._jwks_client.get_signing_key_from_jwt(token)
options = {
"verify_exp": verify_exp,
"verify_aud": verify_aud,
}
audience = self._config.client_id if verify_aud else None
claims = jwt.decode(
token,
signing_key.key,
algorithms=["RS256", "ES256"],
audience=audience,
options=options,
)
return JWTClaims.model_validate(claims)
async def introspect_token(self, token: str) -> dict:
"""Introspect token at IAM server."""
data = {
"token": token,
"client_id": self._config.client_id,
"client_secret": self._config.client_secret,
}
http = await self._get_http()
response = await http.post(
OIDC_INTROSPECT_PATH,
data=data,
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
response.raise_for_status()
return response.json()
async def get_user_info(self, access_token: str) -> UserInfo:
"""Get user info from userinfo endpoint."""
http = await self._get_http()
response = await http.get(
OIDC_USERINFO_PATH,
headers={"Authorization": f"Bearer {access_token}"},
)
response.raise_for_status()
return UserInfo.model_validate(response.json())
def _admin_headers(self) -> dict:
"""The credential, in Authorization — the only place IAM reads one."""
if self._bearer_token:
return {"Authorization": f"Bearer {self._bearer_token}"}
return {"Authorization": basic(self._config.client_id, self._config.client_secret)}
async def get_user(self, user_id: str) -> User:
"""Get user by ID."""
params = {"id": f"{self._config.organization}/{user_id}"}
http = await self._get_http()
response = await http.get(routes.USER, params=params, headers=self._admin_headers())
response.raise_for_status()
data = response.json()
if data.get("status") == "error":
raise ValueError(data.get("msg", "Failed to get user"))
return User.model_validate(routes.unwrap(data))
async def get_users(self) -> list[User]:
"""Get all users in organization."""
params = {"owner": self._config.organization}
http = await self._get_http()
response = await http.get(routes.USERS, params=params, headers=self._admin_headers())
response.raise_for_status()
data = response.json()
if data.get("status") == "error":
raise ValueError(data.get("msg", "Failed to get users"))
users_data = routes.unwrap(data, "users")
return [User.model_validate(u) for u in users_data]
async def get_application(self) -> Application:
"""Get current application configuration."""
params = {"id": f"{self._config.organization}/{self._config.application}"}
http = await self._get_http()
response = await http.get(routes.APPLICATION, params=params, headers=self._admin_headers())
response.raise_for_status()
data = response.json()
if data.get("status") == "error":
raise ValueError(data.get("msg", "Failed to get application"))
return Application.model_validate(routes.unwrap(data))
async def close(self) -> None:
"""Close HTTP client."""
if self._http:
await self._http.aclose()
self._http = None
self._jwks_client = None
async def __aenter__(self) -> AsyncIAMClient:
return self
async def __aexit__(self, *args) -> None:
await self.close()
pkg/hanzo-iam/hanzo_iam/config.py
+26 -34
View File
@@ -9,6 +9,7 @@ from pydantic import BaseModel, ConfigDict, Field
from hanzo_iam.models import (
OIDC_AUTHORIZE_PATH,
Organization,
OIDC_DEVICE_PATH,
OIDC_JWKS_PATH,
OIDC_TOKEN_PATH,
@@ -32,53 +33,44 @@ class IAMConfig(BaseModel):
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")
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, prefix: str | None = None) -> IAMConfig:
"""Create config from environment variables.
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). The ``prefix`` argument exists for advanced
callers that scope multiple IAM clients in a single process; new
code should leave it at the default.
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.
Environment variables (with the default prefix):
IAM_ENDPOINT - IAM server URL (preferred)
IAM_CLIENT_ID - OAuth2 client ID
IAM_CLIENT_SECRET - OAuth2 client secret
IAM_ORG - Organization name
IAM_APP - Application name
IAM_CERT - JWT verification certificate (PEM content or file path)
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
server_url = os.environ.get(f"{p}ENDPOINT", "")
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", "hanzo")
application = os.environ.get(f"{p}APP", "app")
# Certificate can be content or file path
cert_val = os.environ.get(f"{p}CERT", "")
if cert_val and not cert_val.startswith("-----BEGIN"):
# Treat as file path
cert_path = os.path.expanduser(cert_val)
if os.path.isfile(cert_path):
with open(cert_path) as f:
cert_val = f.read()
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=server_url,
client_id=client_id,
client_secret=client_secret,
organization=organization,
application=application,
certificate=cert_val,
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
pkg/hanzo-iam/hanzo_iam/fastapi.py
+8 -8
View File
@@ -75,18 +75,18 @@ def configure(
if isinstance(org, str):
org = Organization(org)
# Get credentials from args or environment (canonical IAM_* only)
resolved_client_id = client_id or os.getenv("IAM_CLIENT_ID", "")
resolved_client_secret = client_secret or os.getenv("IAM_CLIENT_SECRET", "")
# One reader of the environment, the same one both clients use.
env = IAMConfig.from_env(org)
resolved_client_id = client_id or env.client_id
if not resolved_client_id:
raise ValueError("client_id required (or set IAM_CLIENT_ID)")
_config = IAMConfig(
server_url=os.getenv("IAM_ENDPOINT", org.iam_url),
client_id=resolved_client_id,
client_secret=resolved_client_secret,
organization=os.getenv("IAM_ORG", org.value),
_config = env.model_copy(
update={
"client_id": resolved_client_id,
"client_secret": client_secret or env.client_secret,
}
)
# Reset JWKS client to pick up new config
pkg/hanzo-iam/tests/test_credential_location.py
+26
View File
@@ -89,3 +89,29 @@ def test_a_bearer_replaces_the_pair():
pass
assert seen_auth["value"] == "Bearer tok"
@pytest.mark.asyncio
async def test_the_async_client_puts_it_in_the_same_place():
"""AsyncIAMClient is the exported one, and it had the same leak."""
from hanzo_iam import AsyncIAMClient
captured = {}
def handler(request):
captured["request"] = request
return httpx.Response(200, json={"status": "ok", "data": {}, "data2": []})
c = AsyncIAMClient(config=CONFIG)
c._http = httpx.AsyncClient(
base_url=CONFIG.base_url, transport=httpx.MockTransport(handler)
)
try:
await c.get_users()
except Exception:
pass
request = captured["request"]
assert request.headers["Authorization"] == basic("app-id", "app-secret")
assert "app-secret" not in str(request.url.query)
assert "clientSecret" not in str(request.url.query)
pkg/hanzo-iam/tests/test_endpoints.py
+3 -3
View File
@@ -55,14 +55,14 @@ def test_discovery_is_the_one_root_relative_path():
],
)
def test_config_endpoints(attr, expected):
config = IAMConfig(server_url=SERVER, client_id="hanzo-app")
config = IAMConfig(server_url=SERVER, client_id="hanzo-app", organization="hanzo")
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")
config = IAMConfig(server_url=SERVER + "/", client_id="hanzo-app", organization="hanzo")
assert config.token_endpoint == f"{SERVER}/v1/iam/oauth/token"
assert "//v1" not in config.token_endpoint
@@ -70,7 +70,7 @@ def test_config_tolerates_a_trailing_slash():
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")
config = IAMConfig(server_url=SERVER, client_id="hanzo-app", organization="hanzo")
assert config.jwks_uri != f"{SERVER}/.well-known/jwks"
assert config.jwks_uri != f"{SERVER}/.well-known/jwks.json"
pkg/hanzo-iam/tests/test_env.py
+55 -54
View File
@@ -7,6 +7,7 @@ per-org variants. See ~/work/hanzo/iam/CLAUDE.md "Configuration".
from __future__ import annotations
import os
from pathlib import Path
import pytest
@@ -58,7 +59,7 @@ class TestCanonicalPrefix:
class TestConfigFromEnv:
"""IAMConfig.from_env reads only IAM_* vars."""
"""IAMConfig.from_env is the one reader of the environment."""
def test_reads_iam_vars(self, monkeypatch):
monkeypatch.setenv("IAM_ENDPOINT", "https://iam.example")
@@ -74,72 +75,72 @@ class TestConfigFromEnv:
assert cfg.organization == "acme"
assert cfg.application == "myapp"
def test_ignores_hanzo_iam_legacy(self, monkeypatch):
# Legacy aliases must NOT be honored.
monkeypatch.setenv("HANZO_IAM_CLIENT_ID", "legacy")
monkeypatch.setenv("HANZO_IAM_CLIENT_SECRET", "legacy-secret")
monkeypatch.setenv("HANZO_IAM_ENDPOINT", "https://legacy.example")
def test_org_seeds_the_endpoint_and_the_tenant(self):
"""A process that exports nothing still reaches its own org.
cfg = IAMConfig.from_env()
assert cfg.client_id == ""
assert cfg.client_secret == ""
assert cfg.server_url == ""
This is why from_env takes an org: the default is not a literal.
"""
assert IAMConfig.from_env(Organization.ZOO).server_url == "https://zoo.id"
assert IAMConfig.from_env(Organization.ZOO).organization == "zoo"
assert IAMConfig.from_env(Organization.LUX).server_url == "https://lux.id"
assert IAMConfig.from_env(Organization.LUX).organization == "lux"
def test_ignores_org_prefixed_legacy(self, monkeypatch):
# {ORG}_IAM_* aliases must NOT be honored.
monkeypatch.setenv("LUX_IAM_CLIENT_ID", "leak")
monkeypatch.setenv("ZOO_IAM_CLIENT_SECRET", "leak2")
cfg = IAMConfig.from_env()
assert cfg.client_id == ""
assert cfg.client_secret == ""
def test_defaults_when_empty(self, monkeypatch):
cfg = IAMConfig.from_env()
assert cfg.server_url == ""
assert cfg.client_id == ""
assert cfg.organization == "hanzo"
assert cfg.application == "app"
class TestClientConfigFromEnv:
"""IAMClient._config_from_env reads only IAM_* vars."""
def test_reads_iam_vars(self, monkeypatch):
def test_the_environment_beats_the_seed(self, monkeypatch):
monkeypatch.setenv("IAM_ENDPOINT", "https://iam.example")
monkeypatch.setenv("IAM_CLIENT_ID", "cid")
monkeypatch.setenv("IAM_CLIENT_SECRET", "csec")
monkeypatch.setenv("IAM_ORG", "acme")
monkeypatch.setenv("IAM_APP", "myapp")
monkeypatch.setenv("IAM_CERT", "")
cfg = IAMClient._config_from_env(Organization.HANZO)
cfg = IAMConfig.from_env(Organization.ZOO)
assert cfg.server_url == "https://iam.example"
assert cfg.client_id == "cid"
assert cfg.client_secret == "csec"
assert cfg.organization == "acme"
assert cfg.application == "myapp"
def test_endpoint_defaults_to_org_url(self, monkeypatch):
cfg = IAMClient._config_from_env(Organization.ZOO)
assert cfg.server_url == "https://zoo.id"
assert cfg.organization == "zoo"
@pytest.mark.parametrize(
"name",
[
"HANZO_IAM_CLIENT_ID",
"HANZO_IAM_CLIENT_SECRET",
"HANZO_IAM_ENDPOINT",
"HANZO_IAM_URL",
"LUX_IAM_CLIENT_ID",
"ZOO_IAM_CLIENT_SECRET",
],
)
def test_no_alias_is_honoured(self, monkeypatch, name):
"""IAM_ is the prefix. A brand alias or an org prefix reads as nothing."""
monkeypatch.setenv(name, "leak")
def test_ignores_hanzo_iam_legacy(self, monkeypatch):
monkeypatch.setenv("HANZO_IAM_CLIENT_ID", "legacy")
monkeypatch.setenv("HANZO_IAM_URL", "https://legacy.example")
cfg = IAMClient._config_from_env(Organization.HANZO)
cfg = IAMConfig.from_env(Organization.LUX)
assert cfg.client_id == ""
assert cfg.server_url == "https://hanzo.id" # org default, NOT legacy URL
assert cfg.client_secret == ""
assert cfg.server_url == "https://lux.id"
def test_ignores_org_prefixed_legacy(self, monkeypatch):
monkeypatch.setenv("LUX_IAM_CLIENT_ID", "leak")
monkeypatch.setenv("ZOO_IAM_CLIENT_ID", "leak2")
def test_it_is_the_only_reader(self):
"""No module but config.py reads the environment.
cfg = IAMClient._config_from_env(Organization.LUX)
assert cfg.client_id == ""
Three readers disagreed about the tenant once: two 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.
"""
pkg = Path(__file__).resolve().parent.parent / "hanzo_iam"
readers = [
f.name
for f in pkg.glob("*.py")
if ("os.getenv" in f.read_text() or "os.environ" in f.read_text())
]
assert readers == ["config.py"], readers
if __name__ == "__main__":
pytest.main([__file__, "-v"])
def test_the_tenant_has_no_default():
"""IAMConfig will not guess a tenant.
It defaulted to the literal "hanzo", so a Lux or Zoo service that forgot
IAM_ORG addressed the hanzo tenant and reported hanzo's users as its own.
"""
with pytest.raises(Exception):
IAMConfig(server_url="https://lux.id", client_id="app")
assert IAMConfig(server_url="https://lux.id", client_id="app", organization="lux").organization == "lux"