python-sdk/pkg/hanzo-iam/hanzo_iam/models.py at main

mirror of https://github.com/hanzoai/python-sdk.git synced 2026-10-27 04:20:09 +00:00

Files

T

zeekayandHanzo Dev d08a7d5599 auth: verify tokens instead of asserting they exist

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]>

2026-07-27 18:55:24 -07:00

222 lines

8.5 KiB

Python

Raw Permalink Blame History

 """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
 from datetime import datetime
 from enum import Enum
 from typing import Any
 from pydantic import BaseModel, ConfigDict, Field
 # 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"
 class Organization(str, Enum):
     """Hanzo IAM organizations with their identity domains."""
     HANZO = "hanzo"
     ZOO = "zoo"
     LUX = "lux"
     PARS = "pars"
     @property
     def iam_url(self) -> str:
         """Return the IAM URL for this organization."""
         return f"https://{self.value}.id"
 class TokenResponse(BaseModel):
     """OAuth2 token response."""
     model_config = ConfigDict(populate_by_name=True)
     access_token: str = Field(alias="access_token")
     token_type: str = Field(default="Bearer", alias="token_type")
     expires_in: int = Field(alias="expires_in")
     refresh_token: str | None = Field(default=None, alias="refresh_token")
     id_token: str | None = Field(default=None, alias="id_token")
     scope: str | None = Field(default=None)
 class JWTClaims(BaseModel):
     """JWT claims with standard and Hanzo-specific fields."""
     model_config = ConfigDict(populate_by_name=True, extra="allow")
     # Standard JWT claims
     iss: str | None = Field(default=None, description="Issuer")
     sub: str | None = Field(default=None, description="Subject (user ID)")
     aud: str | list[str] | None = Field(default=None, description="Audience")
     exp: int | None = Field(
         default=None, description="Expiration time (Unix timestamp)"
     )
     iat: int | None = Field(default=None, description="Issued at (Unix timestamp)")
     nbf: int | None = Field(default=None, description="Not before (Unix timestamp)")
     jti: str | None = Field(default=None, description="JWT ID")
     # Hanzo-specific claims
     name: str | None = Field(default=None, description="User display name")
     email: str | None = Field(default=None, description="User email")
     owner: str | None = Field(default=None, description="Organization owner")
     roles: list[str] = Field(default_factory=list, description="User roles")
     @property
     def is_expired(self) -> bool:
         """Check if token is expired."""
         if self.exp is None:
             return False
         return datetime.now().timestamp() > self.exp
 class UserInfo(BaseModel):
     """OIDC UserInfo response with Hanzo extensions."""
     model_config = ConfigDict(populate_by_name=True, extra="allow")
     # Standard OIDC claims
     sub: str = Field(description="Subject identifier")
     name: str | None = Field(default=None)
     given_name: str | None = Field(default=None, alias="given_name")
     family_name: str | None = Field(default=None, alias="family_name")
     preferred_username: str | None = Field(default=None, alias="preferred_username")
     email: str | None = Field(default=None)
     email_verified: bool = Field(default=False, alias="email_verified")
     picture: str | None = Field(default=None)
     locale: str | None = Field(default=None)
     updated_at: int | None = Field(default=None, alias="updated_at")
     # Hanzo extensions
     owner: str | None = Field(default=None, description="Organization owner")
     balance: float = Field(default=0.0, description="Account balance")
     is_admin: bool = Field(default=False, alias="isAdmin", description="Admin status")
     roles: list[str] = Field(default_factory=list, description="User roles")
 class User(BaseModel):
     """Full user object from IAM API."""
     model_config = ConfigDict(populate_by_name=True, extra="allow")
     # Identity
     id: str | None = Field(default=None)
     owner: str = Field(description="Organization owner")
     name: str = Field(description="Username")
     display_name: str = Field(default="", alias="displayName")
     # Contact
     email: str | None = Field(default=None)
     phone: str | None = Field(default=None)
     country_code: str | None = Field(default=None, alias="countryCode")
     # Profile
     avatar: str | None = Field(default=None)
     avatar_type: str | None = Field(default=None, alias="avatarType")
     bio: str | None = Field(default=None)
     location: str | None = Field(default=None)
     homepage: str | None = Field(default=None)
     # Status
     is_admin: bool = Field(default=False, alias="isAdmin")
     is_deleted: bool = Field(default=False, alias="isDeleted")
     is_forbidden: bool = Field(default=False, alias="isForbidden")
     is_online: bool = Field(default=False, alias="isOnline")
     # Account
     type: str | None = Field(default=None)
     password: str | None = Field(default=None)
     password_salt: str | None = Field(default=None, alias="passwordSalt")
     password_type: str | None = Field(default=None, alias="passwordType")
     # Verification
     email_verified: bool = Field(default=False, alias="emailVerified")
     phone_verified: bool = Field(default=False, alias="phoneVerified")
     # Hanzo extensions
     balance: float = Field(default=0.0)
     score: int = Field(default=0)
     karma: int = Field(default=0)
     ranking: int = Field(default=0)
     signup_application: str | None = Field(default=None, alias="signupApplication")
     # Permissions
     roles: list[str] = Field(default_factory=list)
     permissions: list[str] = Field(default_factory=list)
     groups: list[str] = Field(default_factory=list)
     # Timestamps
     created_time: str | None = Field(default=None, alias="createdTime")
     updated_time: str | None = Field(default=None, alias="updatedTime")
     # Additional properties
     properties: dict[str, Any] = Field(default_factory=dict)
 class Application(BaseModel):
     """IAM application configuration."""
     model_config = ConfigDict(populate_by_name=True, extra="allow")
     # Identity
     owner: str = Field(description="Organization owner")
     name: str = Field(description="Application name")
     display_name: str = Field(default="", alias="displayName")
     description: str = Field(default="")
     logo: str = Field(default="")
     homepage_url: str = Field(default="", alias="homepageUrl")
     # OAuth2 configuration
     client_id: str = Field(alias="clientId")
     client_secret: str = Field(default="", alias="clientSecret")
     redirect_uris: list[str] = Field(default_factory=list, alias="redirectUris")
     grant_types: list[str] = Field(default_factory=list, alias="grantTypes")
     response_types: list[str] = Field(default_factory=list, alias="responseTypes")
     # Token settings
     expire_in_hours: int = Field(default=168, alias="expireInHours")
     refresh_expire_in_hours: int = Field(default=0, alias="refreshExpireInHours")
     # Providers
     providers: list[dict[str, Any]] = Field(default_factory=list)
     signup_items: list[dict[str, Any]] = Field(
         default_factory=list, alias="signupItems"
     )
     # Features
     enable_password: bool = Field(default=True, alias="enablePassword")
     enable_signup: bool = Field(default=True, alias="enableSignUp")
     enable_signin_session: bool = Field(default=False, alias="enableSigninSession")
     enable_code_signin: bool = Field(default=False, alias="enableCodeSignin")
     # Organization
     organization: str = Field(default="")
     # Certificate for JWT verification
     cert: str = Field(default="")
     # Timestamps
     created_time: str | None = Field(default=None, alias="createdTime")

222 lines 8.5 KiB Python Raw Permalink Blame History

222 lines

8.5 KiB

Python

Raw Permalink Blame History