This repository has no description
8.6 kB
246 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}
91IDP_PATHS = {"github": "github", "google": "google", "oidc": "generic_oidc"}
92for idp in config["idps"].values():
93 name = idp["name"]
94 client_id = idp["clientId"]
95 if client_id is None:
96 with open(idp["clientIdFile"]) as f:
97 client_id = f.read().strip()
98 with open(idp["clientSecretFile"]) as f:
99 client_secret = f.read().strip()
100 if not client_id or not client_secret:
101 print(f"idp {name}: no credentials, skipped")
102 continue
103 body = {
104 "name": name,
105 "clientId": client_id,
106 "clientSecret": client_secret,
107 "scopes": idp["scopes"],
108 "providerOptions": idp["providerOptions"],
109 }
110 if idp["type"] == "oidc":
111 body |= {
112 "issuer": idp["issuer"],
113 # user info from the ID token, without a userinfo request
114 "isIdTokenMapping": True,
115 "usePkce": True,
116 }
117 path = f"/admin/v1/idps/{IDP_PATHS[idp['type']]}"
118 existing = by_name(providers, name)
119 if existing:
120 idp_id = existing["id"]
121 call("PUT", f"{path}/{idp_id}", body)
122 else:
123 idp_id = call("POST", path, body)["id"]
124 print(f"idp {name}: created")
125 if idp_id not in linked:
126 call("POST", "/admin/v1/policies/login/idps", {"idpId": idp_id})
127 print(f"idp {name}: added to the login policy")
128
129
130# projects, OIDC apps and user grants (default organization)
131
132def find_user(name):
133 method = "TEXT_QUERY_METHOD_EQUALS_IGNORE_CASE"
134 users = search("/management/v1/users/_search", {"queries": [
135 {"loginNameQuery": {"loginName": name, "method": method}},
136 ]})
137 if not users:
138 # only verified addresses: anyone can register with any address
139 users = [
140 u for u in search("/management/v1/users/_search", {"queries": [
141 {"emailQuery": {"emailAddress": name, "method": method}},
142 ]})
143 if u.get("human", {}).get("email", {}).get("isEmailVerified")
144 ]
145 if len(users) != 1:
146 print(f"user {name}: {len(users)} matches, skipped", file=sys.stderr)
147 return None
148 return users[0]["id"]
149
150
151def write_file(path, uid, content):
152 tmp = path + ".tmp"
153 fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o400)
154 with os.fdopen(fd, "w") as f:
155 f.write(content)
156 os.chown(tmp, uid, -1)
157 os.rename(tmp, path)
158
159
160projects = search("/management/v1/projects/_search")
161for project_name, project in config["projects"].items():
162 check = project["access"] != "public"
163 existing = by_name(projects, project_name)
164 if existing is None:
165 project_id = call("POST", "/management/v1/projects", {
166 "name": project_name,
167 "projectRoleCheck": check,
168 })["id"]
169 print(f"project {project_name}: created")
170 else:
171 project_id = existing["id"]
172 if differs(existing, {"projectRoleCheck": check}):
173 call("PUT", f"/management/v1/projects/{project_id}", {
174 **existing,
175 "projectRoleCheck": check,
176 })
177 print(f"project {project_name}: projectRoleCheck={check}")
178
179 base = f"/management/v1/projects/{project_id}/apps"
180 apps = search(f"{base}/_search")
181 for app_name, app in project["apps"].items():
182 oidc_config = {
183 "redirectUris": app["redirectUris"],
184 "responseTypes": ["OIDC_RESPONSE_TYPE_CODE"],
185 "grantTypes": ["OIDC_GRANT_TYPE_AUTHORIZATION_CODE"],
186 "appType": "OIDC_APP_TYPE_WEB",
187 "authMethodType": app["authMethodType"],
188 }
189 state_file = f"{config['stateDir']}/{project_name}-{app_name}.json"
190 secret = None
191 existing = by_name(apps, app_name)
192 if existing is None:
193 new = {"name": app_name, **oidc_config}
194 res = call("POST", f"{base}/oidc", new)
195 app_id, client_id = res["appId"], res["clientId"]
196 secret = res["clientSecret"]
197 print(f"app {project_name}/{app_name}: created")
198 else:
199 app_id = existing["id"]
200 client_id = existing["oidcConfig"]["clientId"]
201 if differs(existing["oidcConfig"], oidc_config):
202 call("PUT", f"{base}/{app_id}/oidc_config", oidc_config)
203 print(f"app {project_name}/{app_name}: updated")
204 if os.path.exists(state_file):
205 with open(state_file) as f:
206 saved = json.load(f)
207 if saved["clientId"] == client_id:
208 secret = saved["clientSecret"]
209 # Zitadel shows a secret only when it creates it, so without a saved
210 # copy for this client the app gets a new one.
211 if secret is None:
212 path = f"{base}/{app_id}/oidc_config/_generate_client_secret"
213 secret = call("POST", path, {})["clientSecret"]
214 print(f"app {project_name}/{app_name}: new client secret")
215 write_file(state_file, 0, json.dumps({
216 "clientId": client_id,
217 "clientSecret": secret,
218 }))
219 env = app["envFile"]
220 uid = pwd.getpwnam(env["owner"]).pw_uid
221 write_file(env["path"], uid, (
222 f"{env['clientIdVariable']}={client_id}\n"
223 f"{env['clientSecretVariable']}={secret}\n"
224 ))
225
226 if project["access"] != "declarative":
227 continue
228 users = {find_user(name): name for name in project["users"]}
229 users.pop(None, None)
230 grants = search("/management/v1/users/grants/_search", {"queries": [
231 {"projectIdQuery": {"projectId": project_id}},
232 ]})
233 for grant in grants:
234 if grant["userId"] in users:
235 del users[grant["userId"]]
236 else:
237 call(
238 "DELETE",
239 f"/management/v1/users/{grant['userId']}/grants/{grant['id']}",
240 )
241 print(f"project {project_name}: revoked {grant['userName']}")
242 for user_id, name in users.items():
243 call("POST", f"/management/v1/users/{user_id}/grants", {
244 "projectId": project_id,
245 })
246 print(f"project {project_name}: granted {name}")