forked from
lass.pds.lassul.us/superconfig
This repository has no description
8.5 kB
243 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(email):
133 # Only verified addresses: anyone can register with any address and any
134 # free username, including one that looks like someone's address.
135 method = "TEXT_QUERY_METHOD_EQUALS_IGNORE_CASE"
136 users = [
137 u for u in search("/management/v1/users/_search", {"queries": [
138 {"emailQuery": {"emailAddress": email, "method": method}},
139 ]})
140 if u.get("human", {}).get("email", {}).get("isEmailVerified")
141 ]
142 if len(users) != 1:
143 print(f"user {email}: {len(users)} matches, skipped", file=sys.stderr)
144 return None
145 return users[0]["id"]
146
147
148def write_file(path, uid, content):
149 tmp = path + ".tmp"
150 fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o400)
151 with os.fdopen(fd, "w") as f:
152 f.write(content)
153 os.chown(tmp, uid, -1)
154 os.rename(tmp, path)
155
156
157projects = search("/management/v1/projects/_search")
158for project_name, project in config["projects"].items():
159 check = project["access"] != "public"
160 existing = by_name(projects, project_name)
161 if existing is None:
162 project_id = call("POST", "/management/v1/projects", {
163 "name": project_name,
164 "projectRoleCheck": check,
165 })["id"]
166 print(f"project {project_name}: created")
167 else:
168 project_id = existing["id"]
169 if differs(existing, {"projectRoleCheck": check}):
170 call("PUT", f"/management/v1/projects/{project_id}", {
171 **existing,
172 "projectRoleCheck": check,
173 })
174 print(f"project {project_name}: projectRoleCheck={check}")
175
176 base = f"/management/v1/projects/{project_id}/apps"
177 apps = search(f"{base}/_search")
178 for app_name, app in project["apps"].items():
179 oidc_config = {
180 "redirectUris": app["redirectUris"],
181 "responseTypes": ["OIDC_RESPONSE_TYPE_CODE"],
182 "grantTypes": ["OIDC_GRANT_TYPE_AUTHORIZATION_CODE"],
183 "appType": "OIDC_APP_TYPE_WEB",
184 "authMethodType": app["authMethodType"],
185 }
186 state_file = f"{config['stateDir']}/{project_name}-{app_name}.json"
187 secret = None
188 existing = by_name(apps, app_name)
189 if existing is None:
190 new = {"name": app_name, **oidc_config}
191 res = call("POST", f"{base}/oidc", new)
192 app_id, client_id = res["appId"], res["clientId"]
193 secret = res["clientSecret"]
194 print(f"app {project_name}/{app_name}: created")
195 else:
196 app_id = existing["id"]
197 client_id = existing["oidcConfig"]["clientId"]
198 if differs(existing["oidcConfig"], oidc_config):
199 call("PUT", f"{base}/{app_id}/oidc_config", oidc_config)
200 print(f"app {project_name}/{app_name}: updated")
201 if os.path.exists(state_file):
202 with open(state_file) as f:
203 saved = json.load(f)
204 if saved["clientId"] == client_id:
205 secret = saved["clientSecret"]
206 # Zitadel shows a secret only when it creates it, so without a saved
207 # copy for this client the app gets a new one.
208 if secret is None:
209 path = f"{base}/{app_id}/oidc_config/_generate_client_secret"
210 secret = call("POST", path, {})["clientSecret"]
211 print(f"app {project_name}/{app_name}: new client secret")
212 write_file(state_file, 0, json.dumps({
213 "clientId": client_id,
214 "clientSecret": secret,
215 }))
216 env = app["envFile"]
217 uid = pwd.getpwnam(env["owner"]).pw_uid
218 write_file(env["path"], uid, (
219 f"{env['clientIdVariable']}={client_id}\n"
220 f"{env['clientSecretVariable']}={secret}\n"
221 ))
222
223 if project["access"] != "declarative":
224 continue
225 users = {find_user(name): name for name in project["users"]}
226 users.pop(None, None)
227 grants = search("/management/v1/users/grants/_search", {"queries": [
228 {"projectIdQuery": {"projectId": project_id}},
229 ]})
230 for grant in grants:
231 if grant["userId"] in users:
232 del users[grant["userId"]]
233 else:
234 call(
235 "DELETE",
236 f"/management/v1/users/{grant['userId']}/grants/{grant['id']}",
237 )
238 print(f"project {project_name}: revoked {grant['userName']}")
239 for user_id, name in users.items():
240 call("POST", f"/management/v1/users/{user_id}/grants", {
241 "projectId": project_id,
242 })
243 print(f"project {project_name}: granted {name}")