mirror of
https://github.com/hanzoai/datastore.git
synced 2026-09-26 13:55:07 +00:00
zeekayandHanzo Dev
c7c5ab4f73
`owner` names the org of the APPLICATION a token was minted through, so reading it let a person pick their tenant, and reach SuperAdmin, by picking an application. The session's org is now the first entry of the signed `orgs` membership set (authz.Claims.Home), empty for a token without one; a program's org is a membership IAM granted its application, never its owner. `owner` is read for nothing. SuperAdmin is a person whose home org is admin, in a token signed with the admin org's own key (superadmin_kid, default cert-admin). A program never is, and neither is an admin membership next to another home org. The predicates are hanzoai/authz testdata/claims.json, embedded from the contrib/authz submodule: authz's tests and these run the same cases, so the two readers cannot drift apart silently. Migrations run as the schema's owning application under a scoped `schema` role, declared with superadmin in users.d so both exist before any SQL: DDL and row policies on event and attic, SELECT and INSERT on event with grant option, CREATE ROLE on the four event-plane roles by name. The image now starts its server once at build time and checks the directory, the two roles and a refused non-token login. Co-authored-by: Hanzo Dev <[email protected]>
114 lines
4.0 KiB
Python
114 lines
4.0 KiB
Python
"""A stand-in for Hanzo IAM, run inside the test container on http://localhost:8000.
|
|
|
|
It serves what the datastore reads from an issuer (the OIDC discovery document and the JWKS)
|
|
and, for the test only, mints RS256 access tokens on GET /token. Each org signs with its own
|
|
key, as IAM does: `cert-hanzo` and `cert-admin`. The keys are generated at start with the
|
|
openssl CLI, so the container needs nothing beyond python3 and openssl.
|
|
|
|
GET /token?sub=acme/alice&orgs=acme,zoo[&owner=...&azp=...&type=application&aud=...&exp_in=...&kid=...]
|
|
|
|
`orgs` is the membership set, home org first, as IAM signs it; `owner` is the org of the
|
|
application the token was minted through and should decide nothing.
|
|
"""
|
|
|
|
import base64
|
|
import http.server
|
|
import json
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
import urllib.parse
|
|
|
|
ISSUER = "http://localhost:8000"
|
|
KIDS = ["cert-hanzo", "cert-admin"]
|
|
|
|
|
|
def b64url(data: bytes) -> str:
|
|
return base64.urlsafe_b64encode(data).rstrip(b"=").decode()
|
|
|
|
|
|
def int_bytes(value: int) -> bytes:
|
|
return value.to_bytes((value.bit_length() + 7) // 8, "big")
|
|
|
|
|
|
def generate() -> str:
|
|
path = tempfile.NamedTemporaryFile(suffix=".pem", delete=False).name
|
|
subprocess.run(
|
|
["openssl", "genpkey", "-algorithm", "RSA", "-pkeyopt", "rsa_keygen_bits:2048", "-out", path],
|
|
check=True,
|
|
capture_output=True,
|
|
)
|
|
return path
|
|
|
|
|
|
def modulus(path: str) -> int:
|
|
out = subprocess.run(["openssl", "rsa", "-in", path, "-noout", "-modulus"], check=True, capture_output=True, text=True)
|
|
return int(out.stdout.strip().split("=", 1)[1], 16)
|
|
|
|
|
|
KEY_FILES = {kid: generate() for kid in KIDS}
|
|
JWKS = {
|
|
"keys": [
|
|
{"kty": "RSA", "use": "sig", "alg": "RS256", "kid": kid, "n": b64url(int_bytes(modulus(path))), "e": b64url(int_bytes(65537))}
|
|
for kid, path in KEY_FILES.items()
|
|
]
|
|
}
|
|
DISCOVERY = {"issuer": ISSUER, "jwks_uri": ISSUER + "/v1/iam/.well-known/jwks"}
|
|
|
|
|
|
def mint(query: dict) -> str:
|
|
now = int(time.time())
|
|
kid = query.get("kid", "cert-hanzo")
|
|
header = {"alg": "RS256", "typ": "JWT", "kid": kid}
|
|
payload = {
|
|
"iss": query.get("iss", ISSUER),
|
|
"aud": query.get("aud", "datastore"),
|
|
"sub": query["sub"],
|
|
"iat": now,
|
|
"nbf": now,
|
|
"exp": now + int(query.get("exp_in", "600")),
|
|
"tokenType": "access-token",
|
|
}
|
|
for claim in ("owner", "azp", "type"):
|
|
if claim in query:
|
|
payload[claim] = query[claim]
|
|
if "orgs" in query:
|
|
payload["orgs"] = [{"org": org, "role": "member"} for org in query["orgs"].split(",") if org]
|
|
signing_input = b64url(json.dumps(header).encode()) + "." + b64url(json.dumps(payload).encode())
|
|
signature = subprocess.run(
|
|
# A kid nobody published is signed with some real key, so only the kid is wrong.
|
|
["openssl", "dgst", "-sha256", "-sign", KEY_FILES.get(kid, KEY_FILES["cert-hanzo"])],
|
|
input=signing_input.encode(),
|
|
check=True,
|
|
capture_output=True,
|
|
).stdout
|
|
return signing_input + "." + b64url(signature)
|
|
|
|
|
|
class Handler(http.server.BaseHTTPRequestHandler):
|
|
def answer(self, body: str, content_type: str = "application/json") -> None:
|
|
raw = body.encode()
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", content_type)
|
|
self.send_header("Content-Length", str(len(raw)))
|
|
self.end_headers()
|
|
self.wfile.write(raw)
|
|
|
|
def do_GET(self) -> None:
|
|
url = urllib.parse.urlparse(self.path)
|
|
if url.path == "/health":
|
|
self.answer("OK", "text/plain")
|
|
elif url.path == "/.well-known/openid-configuration":
|
|
self.answer(json.dumps(DISCOVERY))
|
|
elif url.path == "/v1/iam/.well-known/jwks":
|
|
self.answer(json.dumps(JWKS))
|
|
elif url.path == "/token":
|
|
self.answer(mint(dict(urllib.parse.parse_qsl(url.query))), "text/plain")
|
|
else:
|
|
self.send_response(404)
|
|
self.end_headers()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
http.server.HTTPServer(("localhost", 8000), Handler).serve_forever()
|