Repository navigation
feat(realm): tenant translation contract + single-tenant guard (Phase 3, #2196) #2239
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,42 @@ | ||
| name: Keycloak realm — tenant claim contract | ||
|
|
||
| # Guards the pinned-tenant contract (ADR-0003 / DD-AUTH-04) against the | ||
| # canonical broker realm: fail-closed with no pinned attribute, the pin | ||
| # emitted verbatim as a single string, tenant-bearing groups inert AND | ||
| # flagged as a policy violation. Boots a throwaway Keycloak in docker. | ||
|
|
||
| on: | ||
| pull_request: | ||
| branches: [main] | ||
| paths: | ||
| - "deploy/gitops/environments/*/keycloak/realms/**" | ||
| - "deploy/compose/keycloak/tests/**" | ||
| - ".github/workflows/keycloak-realm-contract.yml" | ||
| workflow_dispatch: | ||
|
|
||
| concurrency: | ||
| group: ${{ github.workflow }}-${{ github.ref }} | ||
| cancel-in-progress: true | ||
|
|
||
| permissions: | ||
| contents: read | ||
|
|
||
| jobs: | ||
| tenant-contract: | ||
| name: tenant contract guard | ||
| runs-on: ubuntu-latest | ||
| timeout-minutes: 15 | ||
| steps: | ||
| - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 | ||
| with: | ||
| persist-credentials: false | ||
|
|
||
| - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 | ||
| with: | ||
| python-version: "3.12" | ||
|
|
||
| - name: Install PyYAML | ||
| run: python3 -m pip install --quiet pyyaml | ||
|
|
||
| - name: Tenant translation guard | ||
| run: python3 deploy/compose/keycloak/tests/tenant_contract_guard.py | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,197 @@ | ||
| #!/usr/bin/env python3 | ||
| """Pinned-tenant contract guard (ADR-0003, insight#2196). | ||
|
|
||
| The tenant is ALWAYS the pinned per-registration value (a user attribute | ||
| stamped by each IdP mapper); an IdP's own tenancy assertions are never | ||
| consulted, and group-sourced tenants are rejected by policy. Imports the | ||
| canonical broker realm into a throwaway Keycloak and asserts: | ||
|
|
||
| 1. clean baseline -> the canonical realm ships zero tenant groups | ||
| 2. no pinned attribute -> token carries NO tenant_id (fail closed) | ||
| 3. pinned attribute -> token tenant_id == the pin, a single string | ||
| 4. tenant-bearing group -> must NOT influence the token (no aggregation) | ||
| AND the realm scan flags it as a violation | ||
|
|
||
| Requires docker and PyYAML. Boots quay.io/keycloak/keycloak, converts the | ||
| canonical realm YAML to a realm representation (env placeholders substituted | ||
| with synthetic values), creates it via the admin API, and evaluates example | ||
| tokens. Exit 0 = contract holds. | ||
| """ | ||
|
|
||
| # ruff: noqa: T201 — stdout IS this script's CI report. | ||
|
|
||
| import json | ||
| import re | ||
| import subprocess | ||
| import sys | ||
| import time | ||
| import urllib.error | ||
| import urllib.parse | ||
| import urllib.request | ||
| from pathlib import Path | ||
|
|
||
| import yaml | ||
|
|
||
| REPO_ROOT = Path(__file__).resolve().parents[4] | ||
| CANONICAL_REALM = REPO_ROOT / "deploy/gitops/environments/local/keycloak/realms/insight-broker.yaml" | ||
| KC_IMAGE = "quay.io/keycloak/keycloak:26.4" | ||
| KC_PORT = 18086 | ||
| BASE = f"http://127.0.0.1:{KC_PORT}" | ||
| CONTAINER = "tenant-guard-kc" | ||
|
|
||
| PLACEHOLDER = re.compile(r"\$\(env:([A-Za-z_][A-Za-z0-9_]*)\)") | ||
| SYNTHETIC = { | ||
| "INSIGHT_AUTHENTICATOR_CLIENT_SECRET": "guard-secret", | ||
| "INSIGHT_AUTHENTICATOR_REDIRECT_URI": "http://localhost/auth/callback", | ||
| "INSIGHT_TENANT_ID": "00000000-0000-0000-0000-00000000feed", | ||
| } | ||
|
|
||
| TENANT_GROUP = "aaaaaaaa-0000-0000-0000-000000000001" | ||
| TENANT_PIN = "cccccccc-0000-0000-0000-000000000003" | ||
|
|
||
|
|
||
| def sh(*args: str) -> None: | ||
| subprocess.run(args, check=True, capture_output=True) | ||
|
|
||
|
|
||
| def substitute(node): | ||
| if isinstance(node, str): | ||
| return PLACEHOLDER.sub(lambda m: SYNTHETIC.get(m.group(1), f"missing-{m.group(1)}"), node) | ||
| if isinstance(node, list): | ||
| return [substitute(v) for v in node] | ||
| if isinstance(node, dict): | ||
| return {k: substitute(v) for k, v in node.items()} | ||
| return node | ||
|
|
||
|
|
||
| class Admin: | ||
| def __init__(self) -> None: | ||
| self.token = self._call( | ||
| "/realms/master/protocol/openid-connect/token", | ||
| method="POST", | ||
| raw=urllib.parse.urlencode( | ||
| {"grant_type": "password", "client_id": "admin-cli", "username": "admin", "password": "admin"} | ||
| ).encode(), | ||
| )["access_token"] | ||
|
|
||
| def _call(self, path: str, method: str = "GET", body=None, raw: bytes | None = None): | ||
| headers = {"Content-Type": "application/x-www-form-urlencoded" if raw else "application/json"} | ||
| if hasattr(self, "token"): | ||
| headers["Authorization"] = f"Bearer {self.token}" | ||
| req = urllib.request.Request( | ||
| f"{BASE}{path}", | ||
| method=method, | ||
| headers=headers, | ||
| data=raw if raw is not None else (json.dumps(body).encode() if body is not None else None), | ||
| ) | ||
| with urllib.request.urlopen(req) as resp: | ||
| payload = resp.read() | ||
| return json.loads(payload) if payload else None | ||
|
coderabbitai[bot] marked this conversation as resolved.
Resolved
|
||
|
|
||
| def realm(self, path: str, method: str = "GET", body=None): | ||
| return self._call(f"/admin/realms/insight-broker{path}", method, body) | ||
|
|
||
|
|
||
| def wait_for_keycloak(timeout_s: int = 180) -> None: | ||
| deadline = time.monotonic() + timeout_s | ||
| while time.monotonic() < deadline: | ||
| try: | ||
| urllib.request.urlopen(f"{BASE}/realms/master/.well-known/openid-configuration", timeout=3) | ||
| return | ||
| except (urllib.error.URLError, OSError): | ||
| time.sleep(3) | ||
| raise TimeoutError("Keycloak did not come up") | ||
|
|
||
|
|
||
| def tenant_claim(admin: Admin, client_id: str, user_id: str): | ||
| token = admin.realm(f"/clients/{client_id}/evaluate-scopes/generate-example-id-token?userId={user_id}&scope=openid") | ||
| return token.get("tenant_id") | ||
|
|
||
|
|
||
| def tenant_bearing_groups(admin: Admin) -> list[str]: | ||
| """Policy scan: NO group in the realm may carry a tenant_id attribute.""" | ||
| offenders = [] | ||
| for g in admin.realm("/groups"): | ||
| detail = admin.realm(f"/groups/{g['id']}") | ||
|
coderabbitai[bot] marked this conversation as resolved.
Resolved
|
||
| if (detail.get("attributes") or {}).get("tenant_id"): | ||
| offenders.append(g["name"]) | ||
|
Comment on lines
+114
to
+130
There was a problem hiding this comment. 🎯 Functional Correctness | 🟠 Major | ⚡ Quick win 🧩 Analysis chain🏁 Script executed: #!/bin/bash
set -euo pipefail
echo "Files:"
git ls-files | rg 'deploy/compose/keycloak/tests/tenant_contract_guard.py|compose|keycloak|tenant_contract_guard' || true
echo
echo "Target file outline/stat:"
wc -l deploy/compose/keycloak/tests/tenant_contract_guard.py
sed -n '1,220p' deploy/compose/keycloak/tests/tenant_contract_guard.py
echo
echo "Search tenant_id and group checks:"
rg -n "tenant_id|all_groups|tenant_bearing_groups|registrations|clean-baseline|subGroups|groups" deploy/compose/keycloak/tests/tenant_contract_guard.py deploy/compose/keycloak 2>/dev/null || trueRepository: constructorfabric/insight Length of output: 20625 🏁 Script executed: #!/bin/bash
set -euo pipefail
# Inspect likely Keycloak API shape and behavior from available docs/repo examples.
echo "Relevant API docs/search:"
rg -n "subGroups|tenant_id|groups/\\{" . 2>/dev/null | head -200 || true
echo
echo "Read only structural verifier for tenant_bearing_groups implementation:"
python3 - <<'PY'
from pathlib import Path
p = Path('deploy/compose/keycloak/tests/tenant_contract_guard.py')
txt = p.read_text()
start = txt.index('def tenant_bearing_groups')
end = txt.index('\ndef ', start+1) if '\ndef ' in txt[start+1:] else len(txt)
body = txt[start:end]
print(body)
print("references subGroups:", 'subGroups' in txt)
print("recursive group traversal present:", any(tok in body for tok in ['iterate_recursive', 'subGroups', 'all_groups(admin)']))
PYRepository: constructorfabric/insight Length of output: 22676 🌐 Web query:
💡 Result: In Keycloak, group management via the REST API is handled through the Groups and Group resources. Group Representation and Subgroups The GroupRepresentation object represents a group and includes a subGroups field, which is a list of nested GroupRepresentation objects [1]. While the data model supports a hierarchical structure, retrieving this hierarchy depends on the API endpoint and parameters used [2]. When listing groups using GET /admin/realms/{realm}/groups, you can control the detail level of the response using the briefRepresentation query parameter [2][3]. Setting briefRepresentation to false includes full details, such as attributes and role mappings, whereas true (the default) limits the response to basic information like ID, name, path, and parentId [4][2][3]. Retrieving subgroups can be done in a few ways: 1. When listing all groups or searching, subgroups may be populated based on the parameters used [2]. 2. You can explicitly list subgroups for a specific parent group by calling the children endpoint: GET /admin/realms/{realm}/groups/{parentId}/children [5]. 3. The GroupRepresentation includes a subGroupCount field and a corresponding getter/setter, which can be useful for managing or displaying hierarchical data [1]. Attributes and tenant_id The GroupRepresentation class contains an attributes field, which is a Map<String, List> used to store custom key-value pairs [6]. There is no specialized "tenant_id" field hardcoded into the GroupRepresentation model; instead, if you need to associate a group with a specific tenant, you should add a "tenant_id" key to the attributes map [6]. Key Administrative Actions - Create a child group: POST /admin/realms/{realm}/groups/{parentId}/children [5]. - Update a child group: POST /admin/realms/{realm}/groups/{id}/children [5]. - Retrieve group details: GET /admin/realms/{realm}/groups/{group-id} [2]. Note that for large hierarchies, retrieving full representations for every nested group can be resource-intensive, so using briefRepresentation: true is generally recommended unless full attribute data is required for all nodes [2][3][7]. Citations:
🌐 Web query:
💡 Result: In modern Keycloak versions (since release 23), the behavior of the Citations:
Traverse nested groups before accepting the policy.
🤖 Prompt for AI Agents |
||
| return offenders | ||
|
|
||
|
|
||
| def main() -> int: | ||
| failures: list[str] = [] | ||
|
|
||
| def check(name: str, ok: bool, detail: str) -> None: | ||
| print(f"{'ok ' if ok else 'FAIL'} {name}: {detail}") | ||
| if not ok: | ||
| failures.append(name) | ||
|
|
||
| realm = substitute(yaml.safe_load(CANONICAL_REALM.read_text())) | ||
|
|
||
| subprocess.run(["docker", "rm", "-f", CONTAINER], capture_output=True, check=False) | ||
| sh( | ||
| "docker", | ||
| "run", | ||
| "-d", | ||
| "--name", | ||
| CONTAINER, | ||
| "-p", | ||
| f"127.0.0.1:{KC_PORT}:8080", | ||
| "-e", | ||
| "KC_BOOTSTRAP_ADMIN_USERNAME=admin", | ||
| "-e", | ||
| "KC_BOOTSTRAP_ADMIN_PASSWORD=admin", | ||
| KC_IMAGE, | ||
| "start-dev", | ||
| ) | ||
| try: | ||
| wait_for_keycloak() | ||
| admin = Admin() | ||
| admin._call("/admin/realms", "POST", realm) | ||
|
|
||
| profile = admin.realm("/users/profile") | ||
| profile["unmanagedAttributePolicy"] = "ADMIN_EDIT" | ||
| admin.realm("/users/profile", "PUT", profile) | ||
|
|
||
| check("clean-baseline", not tenant_bearing_groups(admin), "canonical realm ships zero tenant groups") | ||
|
|
||
| admin.realm( | ||
| "/users", | ||
| "POST", | ||
| {"username": "guard@example.com", "email": "guard@example.com", "enabled": True, "emailVerified": True}, | ||
| ) | ||
| user = admin.realm("/users?username=guard@example.com&exact=true")[0]["id"] | ||
| client = admin.realm("/clients?clientId=insight-authenticator")[0]["id"] | ||
|
|
||
| claim = tenant_claim(admin, client, user) | ||
| check("fail-closed", claim is None, f"no pinned attribute -> claim {claim!r}") | ||
|
|
||
| u = admin.realm(f"/users/{user}") | ||
| u["attributes"] = {"tenant_id": [TENANT_PIN]} | ||
| admin.realm(f"/users/{user}", "PUT", u) | ||
| claim = tenant_claim(admin, client, user) | ||
| check("pinned-tenant", claim == TENANT_PIN, f"pinned attribute -> claim {claim!r}") | ||
| check("claim-is-scalar", isinstance(claim, str), f"claim type {type(claim).__name__}") | ||
|
coderabbitai[bot] marked this conversation as resolved.
Resolved
|
||
|
|
||
| admin.realm("/groups", "POST", {"name": "tenant-a", "attributes": {"tenant_id": [TENANT_GROUP]}}) | ||
| groups = {g["name"]: g["id"] for g in admin.realm("/groups")} | ||
| admin.realm(f"/users/{user}/groups/{groups['tenant-a']}", "PUT") | ||
| claim = tenant_claim(admin, client, user) | ||
| check("group-inert", claim == TENANT_PIN, f"tenant group must not affect the token -> claim {claim!r}") | ||
| check( | ||
| "group-policy-detected", | ||
| bool(tenant_bearing_groups(admin)), | ||
| "a tenant-bearing group exists and the realm scan flags it", | ||
| ) | ||
| finally: | ||
| subprocess.run(["docker", "rm", "-f", CONTAINER], capture_output=True, check=False) | ||
|
|
||
| if failures: | ||
| print(f"\ntenant-contract guard FAILED: {failures}") | ||
| return 1 | ||
| print("\ntenant-contract guard OK") | ||
| return 0 | ||
|
|
||
|
|
||
| if __name__ == "__main__": | ||
| sys.exit(main()) | ||
Uh oh!
There was an error while loading. Please reload this page.