74 lines
3.0 KiB
Python
74 lines
3.0 KiB
Python
"""keycloak — recipe-specific functional test (Phase 2 P3, ≥2 beyond parity).
|
|
|
|
The defining keycloak behavior is issuing JWTs. This test does the canonical password-grant flow
|
|
against the live keycloak (real admin user, real abra-generated admin_password secret), then
|
|
decodes the returned JWT and asserts the standard claims: `iss` (issuer URL), `azp` (authorized
|
|
party), `typ == "Bearer"`, and `exp` in the future. Non-vacuous: a wrongly-configured realm,
|
|
broken signing key, or wrong issuer would fail the claim assertions, not just status==200.
|
|
|
|
Reuses the per-recipe `tests/keycloak/kc_admin.py` helpers (admin_password + admin_token shape)
|
|
and the harness `lifecycle.exec_in_app` (already polled+raising per Phase 1e F1e-1 fix).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
|
|
# kc_admin.py lives in tests/keycloak/, one level up from this file in tests/keycloak/custom/
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "..", "runner"))
|
|
import kc_admin # noqa: E402
|
|
|
|
|
|
def _b64url_decode(seg: str) -> bytes:
|
|
"""JWT segments are URL-safe base64 without padding — add padding before decode."""
|
|
pad = "=" * ((4 - len(seg) % 4) % 4)
|
|
return base64.urlsafe_b64decode(seg + pad)
|
|
|
|
|
|
def _decode_jwt_payload(token: str) -> dict:
|
|
parts = token.split(".")
|
|
assert len(parts) == 3, f"JWT must have 3 segments (header.payload.signature), got {len(parts)}"
|
|
payload = json.loads(_b64url_decode(parts[1]))
|
|
return payload
|
|
|
|
|
|
def test_password_grant_issues_valid_jwt(live_app):
|
|
"""Obtain a token via the password grant, validate the JWT shape + standard claims."""
|
|
password = kc_admin.admin_password(live_app)
|
|
token = kc_admin.admin_token(live_app, password)
|
|
|
|
# Shape: a JWT is exactly 3 base64url segments
|
|
assert (
|
|
isinstance(token, str) and token.count(".") == 2
|
|
), f"access_token does not look like a JWT (no 3 segments): len={len(token) if token else 0}"
|
|
|
|
payload = _decode_jwt_payload(token)
|
|
|
|
# iss = the issuer URL, must be the per-run domain's /realms/master endpoint
|
|
expected_iss = f"https://{live_app}/realms/master"
|
|
assert (
|
|
payload.get("iss") == expected_iss
|
|
), f"JWT iss claim {payload.get('iss')!r} != {expected_iss!r}"
|
|
|
|
# azp = authorized party (which client requested this token)
|
|
assert payload.get("azp") == "admin-cli", f"JWT azp claim {payload.get('azp')!r} != 'admin-cli'"
|
|
|
|
# typ = token type
|
|
assert payload.get("typ") == "Bearer", f"JWT typ claim {payload.get('typ')!r} != 'Bearer'"
|
|
|
|
# exp must be in the future (the token must be usable, not pre-expired)
|
|
exp = payload.get("exp")
|
|
assert isinstance(exp, int), f"JWT exp claim missing or not int: {exp!r}"
|
|
assert exp > time.time(), f"JWT exp {exp} not in the future (now={time.time():.0f})"
|
|
|
|
# iat (issued at) is also a standard claim
|
|
iat = payload.get("iat")
|
|
assert (
|
|
isinstance(iat, int) and iat <= time.time() + 60
|
|
), f"JWT iat {iat!r} not a reasonable past timestamp"
|