This repository has no description
8.3 kB
238 lines
1# Applies lass.zitadel (modules/zitadel/default.nix) to Zitadel through its
2# admin and management APIs. Objects are found by name, created when missing
3# and updated when they differ. Projects and apps missing from the config
4# are left alone.
5import json
6import os
7import pwd
8import sys
9import time
10import urllib.error
11import urllib.request
12
13[config_path] = sys.argv[1:]
14with open(config_path) as f:
15 config = json.load(f)
16
17
18def request(method, path, body=None, token=None):
19 headers = {"Host": config["domain"], "Content-Type": "application/json"}
20 if token:
21 headers["Authorization"] = "Bearer " + token
22 req = urllib.request.Request(
23 config["url"] + path,
24 method=method,
25 headers=headers,
26 data=None if body is None else json.dumps(body).encode(),
27 )
28 with urllib.request.urlopen(req) as res:
29 return json.load(res)
30
31
32# Goes through the HTTP gateway to the gRPC API, which comes up after
33# /debug/ready already answers.
34while True:
35 try:
36 request("GET", "/admin/v1/healthz")
37 break
38 except (urllib.error.URLError, ConnectionError):
39 time.sleep(1)
40
41with open(config["patFile"]) as f:
42 pat = f.read().strip()
43
44
45def call(method, path, body=None):
46 try:
47 return request(method, path, body, pat)
48 except urllib.error.HTTPError as e:
49 sys.exit(f"{method} {path}: {e.code} {e.read().decode()}")
50
51
52def search(path, body={}):
53 results = []
54 while True:
55 res = call("POST", path, {**body, "query": {"offset": len(results)}})
56 page = res.get("result", [])
57 results += page
58 total = int(res.get("details", {}).get("totalResult", 0))
59 if not page or len(results) >= total:
60 return results
61
62
63def by_name(objects, name):
64 return next((o for o in objects if o["name"] == name), None)
65
66
67# The APIs omit fields with zero values.
68ZERO = {
69 "appType": "OIDC_APP_TYPE_WEB",
70 "authMethodType": "OIDC_AUTH_METHOD_TYPE_BASIC",
71}
72
73
74def differs(current, wanted):
75 return any(
76 current.get(k, ZERO.get(k, False)) != v for k, v in wanted.items()
77 )
78
79
80# login policy and external IdPs (instance defaults)
81
82policy = call("GET", "/admin/v1/policies/login")["policy"]
83wanted = config["loginPolicy"]
84if differs(policy, wanted):
85 call("PUT", "/admin/v1/policies/login", {**policy, **wanted})
86
87providers = search("/admin/v1/idps/templates/_search")
88linked = {
89 i["idpId"] for i in search("/admin/v1/policies/login/idps/_search")
90}
91for name, idp in config["idps"].items():
92 with open(idp["clientIdFile"]) as f:
93 client_id = f.read().strip()
94 with open(idp["clientSecretFile"]) as f:
95 client_secret = f.read().strip()
96 if not client_id or not client_secret:
97 print(f"idp {name}: no credentials, skipped")
98 continue
99 body = {
100 "name": name,
101 "clientId": client_id,
102 "clientSecret": client_secret,
103 "scopes": idp["scopes"],
104 "providerOptions": {
105 "isLinkingAllowed": True,
106 "isCreationAllowed": True,
107 "isAutoUpdate": True,
108 },
109 }
110 existing = by_name(providers, name)
111 if existing:
112 idp_id = existing["id"]
113 call("PUT", f"/admin/v1/idps/{idp['type']}/{idp_id}", body)
114 else:
115 idp_id = call("POST", f"/admin/v1/idps/{idp['type']}", body)["id"]
116 print(f"idp {name}: created")
117 if idp_id not in linked:
118 call("POST", "/admin/v1/policies/login/idps", {"idpId": idp_id})
119 print(f"idp {name}: added to the login policy")
120
121
122# projects, OIDC apps and user grants (default organization)
123
124def find_user(name):
125 method = "TEXT_QUERY_METHOD_EQUALS_IGNORE_CASE"
126 users = search("/management/v1/users/_search", {"queries": [
127 {"loginNameQuery": {"loginName": name, "method": method}},
128 ]})
129 if not users:
130 # only verified addresses: anyone can register with any address
131 users = [
132 u for u in search("/management/v1/users/_search", {"queries": [
133 {"emailQuery": {"emailAddress": name, "method": method}},
134 ]})
135 if u.get("human", {}).get("email", {}).get("isEmailVerified")
136 ]
137 if len(users) != 1:
138 print(f"user {name}: {len(users)} matches, skipped", file=sys.stderr)
139 return None
140 return users[0]["id"]
141
142
143def write_file(path, uid, content):
144 tmp = path + ".tmp"
145 fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o400)
146 with os.fdopen(fd, "w") as f:
147 f.write(content)
148 os.chown(tmp, uid, -1)
149 os.rename(tmp, path)
150
151
152projects = search("/management/v1/projects/_search")
153for project_name, project in config["projects"].items():
154 check = project["access"] != "public"
155 existing = by_name(projects, project_name)
156 if existing is None:
157 project_id = call("POST", "/management/v1/projects", {
158 "name": project_name,
159 "projectRoleCheck": check,
160 })["id"]
161 print(f"project {project_name}: created")
162 else:
163 project_id = existing["id"]
164 if differs(existing, {"projectRoleCheck": check}):
165 call("PUT", f"/management/v1/projects/{project_id}", {
166 **existing,
167 "projectRoleCheck": check,
168 })
169 print(f"project {project_name}: projectRoleCheck={check}")
170
171 base = f"/management/v1/projects/{project_id}/apps"
172 apps = search(f"{base}/_search")
173 for app_name, app in project["apps"].items():
174 oidc_config = {
175 "redirectUris": app["redirectUris"],
176 "responseTypes": ["OIDC_RESPONSE_TYPE_CODE"],
177 "grantTypes": ["OIDC_GRANT_TYPE_AUTHORIZATION_CODE"],
178 "appType": "OIDC_APP_TYPE_WEB",
179 "authMethodType": app["authMethodType"],
180 }
181 state_file = f"{config['stateDir']}/{project_name}-{app_name}.json"
182 secret = None
183 existing = by_name(apps, app_name)
184 if existing is None:
185 new = {"name": app_name, **oidc_config}
186 res = call("POST", f"{base}/oidc", new)
187 app_id, client_id = res["appId"], res["clientId"]
188 secret = res["clientSecret"]
189 print(f"app {project_name}/{app_name}: created")
190 else:
191 app_id = existing["id"]
192 client_id = existing["oidcConfig"]["clientId"]
193 if differs(existing["oidcConfig"], oidc_config):
194 call("PUT", f"{base}/{app_id}/oidc_config", oidc_config)
195 print(f"app {project_name}/{app_name}: updated")
196 if os.path.exists(state_file):
197 with open(state_file) as f:
198 saved = json.load(f)
199 if saved["clientId"] == client_id:
200 secret = saved["clientSecret"]
201 # Zitadel shows a secret only when it creates it, so without a saved
202 # copy for this client the app gets a new one.
203 if secret is None:
204 path = f"{base}/{app_id}/oidc_config/_generate_client_secret"
205 secret = call("POST", path, {})["clientSecret"]
206 print(f"app {project_name}/{app_name}: new client secret")
207 write_file(state_file, 0, json.dumps({
208 "clientId": client_id,
209 "clientSecret": secret,
210 }))
211 env = app["envFile"]
212 uid = pwd.getpwnam(env["owner"]).pw_uid
213 write_file(env["path"], uid, (
214 f"{env['clientIdVariable']}={client_id}\n"
215 f"{env['clientSecretVariable']}={secret}\n"
216 ))
217
218 if project["access"] != "declarative":
219 continue
220 users = {find_user(name): name for name in project["users"]}
221 users.pop(None, None)
222 grants = search("/management/v1/users/grants/_search", {"queries": [
223 {"projectIdQuery": {"projectId": project_id}},
224 ]})
225 for grant in grants:
226 if grant["userId"] in users:
227 del users[grant["userId"]]
228 else:
229 call(
230 "DELETE",
231 f"/management/v1/users/{grant['userId']}/grants/{grant['id']}",
232 )
233 print(f"project {project_name}: revoked {grant['userName']}")
234 for user_id, name in users.items():
235 call("POST", f"/management/v1/users/{user_id}/grants", {
236 "projectId": project_id,
237 })
238 print(f"project {project_name}: granted {name}")