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