forked from
lass.pds.lassul.us/superconfig
This repository has no description
5.3 kB
159 lines
1# Invites a person to projects of a Zitadel organization: creates a human
2# user for the email address (username = the address), lets Zitadel mail
3# them an invite code to set up a password, and adds a user grant on each
4# project. An existing user with that address only gets the missing grants.
5import argparse
6import json
7import os
8import sys
9import urllib.error
10import urllib.request
11
12parser = argparse.ArgumentParser(
13 prog="zitadel-invite",
14 description="Invite a person to Zitadel projects.",
15 epilog="It needs the token of a user with the ORG_USER_MANAGER role.",
16)
17parser.add_argument("email")
18parser.add_argument("projects", nargs="+", metavar="project")
19parser.add_argument("--name", help='"First Last", default: the local part')
20parser.add_argument(
21 "--resend",
22 action="store_true",
23 help="send a new invite code to an existing user who has not set up "
24 "a login yet",
25)
26parser.add_argument(
27 "--app-name", help="application name in the invite mail"
28)
29parser.add_argument(
30 "--url",
31 default=os.environ.get("ZITADEL_URL"),
32 help="Zitadel URL, e.g. https://id.example.org (env ZITADEL_URL)",
33)
34parser.add_argument(
35 "--token-file",
36 default=os.environ.get("ZITADEL_TOKEN_FILE"),
37 help="file with a personal access token (env ZITADEL_TOKEN_FILE, or "
38 "the token itself in ZITADEL_TOKEN)",
39)
40args = parser.parse_args()
41
42if not args.url:
43 parser.error("--url or ZITADEL_URL is needed")
44if args.token_file:
45 with open(args.token_file) as f:
46 token = f.read().strip()
47else:
48 token = os.environ.get("ZITADEL_TOKEN")
49if not token:
50 parser.error("--token-file, ZITADEL_TOKEN_FILE or ZITADEL_TOKEN is needed")
51issuer = args.url.rstrip("/")
52
53
54def fetch(req, what):
55 try:
56 with urllib.request.urlopen(req) as res:
57 return json.load(res)
58 except urllib.error.HTTPError as e:
59 sys.exit(f"{what}: {e.code} {e.read().decode()}")
60
61
62def call(method, path, body=None):
63 return fetch(urllib.request.Request(
64 issuer + path,
65 method=method,
66 headers={
67 "Authorization": "Bearer " + token,
68 "Content-Type": "application/json",
69 },
70 data=None if body is None else json.dumps(body).encode(),
71 ), f"{method} {path}")
72
73
74def invite(user_id):
75 send_code = {}
76 if args.app_name:
77 send_code["applicationName"] = args.app_name
78 call("POST", f"/v2/users/{user_id}/invite_code", {"sendCode": send_code})
79 print(f"user {args.email}: invite mail sent")
80
81
82# check the projects before changing anything
83project_ids = {}
84for name in args.projects:
85 found = [
86 p for p in call("POST", "/management/v1/projects/_search", {
87 "queries": [{"nameQuery": {
88 "name": name,
89 "method": "TEXT_QUERY_METHOD_EQUALS",
90 }}],
91 }).get("result", [])
92 if p["name"] == name
93 ]
94 if len(found) != 1:
95 sys.exit(f"project {name}: {len(found)} matches")
96 project_ids[name] = found[0]["id"]
97
98users = call("POST", "/management/v1/users/_search", {"queries": [
99 {"emailQuery": {
100 "emailAddress": args.email,
101 "method": "TEXT_QUERY_METHOD_EQUALS_IGNORE_CASE",
102 }},
103 {"typeQuery": {"type": "TYPE_HUMAN"}},
104]}).get("result", [])
105if len(users) > 1:
106 sys.exit(f"user {args.email}: {len(users)} users with this address")
107
108if users:
109 user = users[0]
110 user_id = user["id"]
111 # Anyone can register with any address. Grants go only to users who
112 # verified theirs, or who were created by an admin and have no login yet.
113 initial = user.get("state") == "USER_STATE_INITIAL"
114 verified = user["human"].get("email", {}).get("isEmailVerified", False)
115 if not initial and not verified:
116 sys.exit(
117 f"user {args.email}: exists ({user['userName']}) with the "
118 "address unverified, not granting"
119 )
120 print(f"user {args.email}: exists ({user['userName']})")
121 if args.resend:
122 if initial:
123 invite(user_id)
124 else:
125 print(f"user {args.email}: has a login already, no invite")
126else:
127 full_name = args.name or args.email.split("@")[0]
128 given, _, family = full_name.partition(" ")
129 org_id = call("GET", "/management/v1/orgs/me")["org"]["id"]
130 user_id = call("POST", "/v2/users/new", {
131 "organizationId": org_id,
132 "human": {
133 "profile": {
134 "givenName": given,
135 # Zitadel requires one
136 "familyName": family.strip() or given,
137 "displayName": full_name,
138 },
139 # verified by the invite code, without a separate mail
140 "email": {"email": args.email, "returnCode": {}},
141 },
142 })["id"]
143 print(f"user {args.email}: created")
144 invite(user_id)
145
146granted = {
147 g["projectId"]
148 for g in call("POST", "/management/v1/users/grants/_search", {
149 "queries": [{"userIdQuery": {"userId": user_id}}],
150 }).get("result", [])
151}
152for name, project_id in project_ids.items():
153 if project_id in granted:
154 print(f"project {name}: granted already")
155 else:
156 call("POST", f"/management/v1/users/{user_id}/grants", {
157 "projectId": project_id,
158 })
159 print(f"project {name}: granted")