mirror of
https://github.com/hanzoai/python-sdk.git
synced 2026-10-27 04:20:09 +00:00
zeekay
28035b6721
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.
118 lines
3.0 KiB
Python
118 lines
3.0 KiB
Python
"""Where the confidential client's credential rides.
|
|
|
|
IAM reads a credential from Authorization and nowhere else — its `app()` calls
|
|
httpx.Basic(c) and never looks at the query. A pair in the query string
|
|
therefore authenticates nothing and lands in every log on the path.
|
|
"""
|
|
|
|
import base64
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from hanzo_iam.client import IAMClient, basic
|
|
from hanzo_iam.config import IAMConfig
|
|
|
|
|
|
CONFIG = IAMConfig(
|
|
server_url="https://hanzo.id",
|
|
client_id="app-id",
|
|
client_secret="app-secret",
|
|
organization="hanzo",
|
|
application="hanzo-cloud",
|
|
)
|
|
|
|
|
|
def client(handler):
|
|
c = IAMClient(config=CONFIG)
|
|
c._http = httpx.Client(
|
|
base_url=CONFIG.base_url, transport=httpx.MockTransport(handler)
|
|
)
|
|
return c
|
|
|
|
|
|
def seen(call):
|
|
"""Run one admin call against a mock and return the request it sent.
|
|
|
|
Only the request matters here, so a reply the caller cannot deserialize is
|
|
still an answer.
|
|
"""
|
|
captured = {}
|
|
|
|
def handler(request):
|
|
captured["request"] = request
|
|
return httpx.Response(200, json={"status": "ok", "data": {}, "data2": []})
|
|
|
|
try:
|
|
call(client(handler))
|
|
except Exception:
|
|
pass
|
|
return captured["request"]
|
|
|
|
|
|
def test_basic_renders_the_pair():
|
|
assert basic("id", "secret") == "Basic " + base64.b64encode(b"id:secret").decode()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"call",
|
|
[
|
|
lambda c: c.get_user("z"),
|
|
lambda c: c.get_users(),
|
|
lambda c: c.get_application(),
|
|
lambda c: c.get_organizations(),
|
|
],
|
|
)
|
|
def test_credential_is_in_authorization_never_the_query(call):
|
|
request = seen(call)
|
|
|
|
assert request.headers["Authorization"] == basic("app-id", "app-secret")
|
|
|
|
query = str(request.url.query)
|
|
assert "app-secret" not in query
|
|
assert "clientSecret" not in query
|
|
assert "clientId" not in query
|
|
|
|
|
|
def test_a_bearer_replaces_the_pair():
|
|
seen_auth = {}
|
|
|
|
def handler(request):
|
|
seen_auth["value"] = request.headers["Authorization"]
|
|
return httpx.Response(200, json={"status": "ok", "data": {}})
|
|
|
|
c = client(handler)
|
|
c._bearer_token = "tok"
|
|
try:
|
|
c.get_user("z")
|
|
except Exception:
|
|
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)
|