1#!/usr/bin/env python3
2import argparse
3import importlib
4import json
5from pathlib import Path
6import subprocess
7import sys
8import time
9import urllib.error
10import urllib.parse
11import urllib.request
12import uuid
13
14
15def main():
16 parser = argparse.ArgumentParser()
17 parser.add_argument("--url", required=True)
18 parser.add_argument("--socket", required=True)
19 parser.add_argument("--proof-file", type=Path, required=True)
20 parser.add_argument("--user", default="studio-dashboard")
21 parser.add_argument("--output", type=Path)
22 args = parser.parse_args()
23 if args.output:
24 args.output.unlink(missing_ok=True)
25 sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "service/keycloak"))
26 from api import Keycloak
27 keycloak = Keycloak("keycloak.studio.test", importlib.import_module("dashboard-run").secret("get", "keycloak", "password"), attempts=1)
28 keycloak.configure_profile()
29 name = "iam-fixture-" + uuid.uuid4().hex
30 password = uuid.uuid4().hex + "A1!"
31 identity = None
32 proof = args.proof_file.read_text().strip()
33 admin = keycloak.request("/admin/realms/master/users?username=admin&exact=true")[0]
34 admin = keycloak.request("/admin/realms/master/users/" + admin["id"])
35 admin_roles = keycloak.request("/admin/realms/master/users/" + admin["id"] + "/role-mappings/realm")
36 server_role = next(role for role in admin_roles if role["name"] == "admin")
37
38 def http(path, method="GET", body=None, status=200, actor="fixture-operator", groups="infra-admin"):
39 request = urllib.request.Request(args.url + "/api/" + path, method=method,
40 data=json.dumps(body).encode() if body is not None else None,
41 headers={"User-Name": actor, "User-Groups": groups, "Studio-Proxy-Token": proof, "Content-Type": "application/json"})
42 try:
43 response = urllib.request.urlopen(request, timeout=70)
44 except urllib.error.HTTPError as error:
45 response = error
46 with response:
47 content = response.read()
48 assert response.status == status, (path, response.status, content[:200])
49 return json.loads(content) if content and response.headers.get("Content-Type", "").startswith("application/json") else None
50
51 def broker(path=None, method="GET", body=None, status=200, **fields):
52 payload = {"operation": "iam.request", "path": path, "method": method, "body": body} if path is not None else fields
53 process = subprocess.run([sys.executable, str(Path(__file__).with_name("dashboard-host-vm-test.py")), args.socket,
54 "--client", "--user", args.user], input=json.dumps(payload).encode() + b"\n", capture_output=True, timeout=75)
55 assert process.returncode == 0, "broker client failed"
56 result = json.loads(process.stdout)
57 if status != 200:
58 assert result.get("status") == status, (path, result)
59 return None
60 assert "value" in result, (path, result)
61 return result["value"]
62
63 def sign_in(secret):
64 body = urllib.parse.urlencode({"client_id": "admin-cli", "grant_type": "password", "username": name, "password": secret}).encode()
65 return keycloak.request("/realms/master/protocol/openid-connect/token", "POST", body)
66
67 try:
68 started = time.monotonic()
69 directory = http("users")
70 directory_seconds = time.monotonic() - started
71 assert all(user["username"] != "admin" for user in directory["users"])
72 roles = {role["name"]: role for role in directory["groups"]}
73 assert set(roles) == {"infra-admin", "media", "media-manage"}
74 http("users", status=403, groups="media")
75 created = http("users", "POST", {"profile": {"username": name, "email": name + "@fixture.invalid", "firstName": "Native", "lastName": "Fixture"},
76 "groups": [roles["media"]["id"]], "setup": {"kind": "password", "password": password}}, status=201)
77 identity = created["id"]
78 path = "users/" + identity
79 assert keycloak.request("/admin/realms/master/" + path)["username"] == name
80 http(path, "PATCH", {"requiredActions": [], "emailVerified": True, "firstName": "Verified"}, status=204)
81 http(path + "/password", "PUT", {"password": password, "temporary": False}, status=204)
82 tokens = sign_in(password)
83 assert tokens["access_token"]
84 http(path + "/credentials")
85 http(path + "/groups/" + roles["infra-admin"]["id"], "PUT", status=204)
86 http(path + "/groups/" + roles["infra-admin"]["id"], "DELETE", status=400, actor=name)
87 http(path + "/groups/" + roles["infra-admin"]["id"], "DELETE", status=204)
88 mapped = {role["name"] for role in keycloak.request("/admin/realms/master/" + path + "/role-mappings/realm")}
89 assert mapped & {"infra-admin", "media", "media-manage", "admin"} == {"media"}
90 http("account", "PATCH", {"firstName": "Account"}, status=204, actor=name, groups="media")
91 assert http("account", actor=name, groups="media")["firstName"] == "Account"
92 broker("/" + path, "PUT", {"attributes": {"picture": ["https://fixture.invalid/picture"]}})
93 profile = keycloak.request("/admin/realms/master/" + path)
94 assert profile["attributes"]["picture"] == ["https://fixture.invalid/picture"]
95 assert profile["firstName"] == "Account" and profile["email"] == name + "@fixture.invalid"
96 broker("/" + path, "PUT", {"attributes": {"picture": None}})
97 profile = keycloak.request("/admin/realms/master/" + path)
98 assert "picture" not in profile.get("attributes", {})
99 assert profile["firstName"] == "Account" and profile["email"] == name + "@fixture.invalid"
100 http(path + "/logout", "POST", status=204)
101 try:
102 keycloak.request("/realms/master/protocol/openid-connect/token", "POST", urllib.parse.urlencode({
103 "client_id": "admin-cli", "grant_type": "refresh_token", "refresh_token": tokens["refresh_token"]}).encode())
104 except urllib.error.HTTPError as error:
105 assert error.code == 400
106 else:
107 raise AssertionError("logged-out refresh token remained usable")
108 changed = uuid.uuid4().hex + "A1!"
109 http(path + "/password", "PUT", {"password": changed, "temporary": False}, status=204)
110 assert sign_in(changed)["access_token"]
111 http(path, "PATCH", {"enabled": False}, status=204)
112 try:
113 sign_in(changed)
114 except urllib.error.HTTPError as error:
115 assert error.code == 400
116 else:
117 raise AssertionError("disabled fixture signed in")
118 http(path, "PATCH", {"enabled": True}, status=204)
119 for operation in ["get", "set", "rotate"]:
120 fields = {"operation": "deploy.secret." + operation, "service": "keycloak", "key": "password"}
121 if operation == "set":
122 fields["value"] = "fixture-not-applied"
123 broker(status=403, **fields)
124 for forbidden in ["/clients", "/realm-settings", "/users/" + identity + "/../../clients", "/users?max=1000&first=0"]:
125 broker(forbidden, status=400)
126 broker("/users/" + admin["id"] + "/reset-password", "PUT", {"type": "password", "value": "fixture-not-applied", "temporary": False}, status=403)
127 broker("/users/" + admin["id"], "DELETE", status=403)
128 broker("/" + path + "/role-mappings/realm", "POST", [server_role], status=403)
129 broker("/" + path + "/role-mappings/realm", "POST", [{"id": server_role["id"], "name": "infra-admin"}], status=403)
130 assert keycloak.request("/admin/realms/master/users/" + admin["id"]) == admin
131 assert keycloak.request("/admin/realms/master/users/" + admin["id"] + "/role-mappings/realm") == admin_roles
132 Keycloak("keycloak.studio.test", importlib.import_module("dashboard-run").secret("get", "keycloak", "password"), attempts=1)
133 http(path, "DELETE", status=204)
134 identity = None
135 result = {"native_user_crud": True, "password_sign_in_and_reset": True, "disable_rejects_sign_in": True,
136 "dashboard_role_grants": True, "self_lockout_refused": True, "account_profile": True,
137 "logout_revokes_refresh": True, "non_admin_refused": True, "bootstrap_user_hidden_and_protected": True,
138 "credential_export_and_change_refused": True, "admin_paths_and_forged_roles_refused": True,
139 "existing_realm_preserved": True, "directory_seconds": round(directory_seconds, 3)}
140 result["picture_persists_and_preserves_profile"] = True
141 finally:
142 if identity is not None:
143 keycloak.request("/admin/realms/master/users/" + identity, "DELETE")
144 remaining = keycloak.request("/admin/realms/master/users?username=" + name + "&exact=true")
145 for user in remaining:
146 keycloak.request("/admin/realms/master/users/" + user["id"], "DELETE")
147 assert not keycloak.request("/admin/realms/master/users?username=" + name + "&exact=true")
148 result["owned_fixture_removed"] = True
149 if args.output:
150 args.output.parent.mkdir(parents=True, exist_ok=True)
151 args.output.write_text(json.dumps(result, indent=2) + "\n")
152 print(json.dumps(result))
153
154
155if __name__ == "__main__":
156 main()