···11+bin "zitadel-invite"
22+about "Invite a person to Zitadel projects: create the user, send the invite mail, add user grants"
33+44+arg "<email>" help="Email address, also the username of a new user"
55+arg "<project>..." help="Project name"
66+flag "--name" help="\"First Last\" of a new user (default: the email's local part)" {
77+ arg "<name>"
88+}
99+flag "--resend" help="Send a new invite code to an existing user without a login"
1010+flag "--app-name" help="Application name in the invite mail" {
1111+ arg "<name>"
1212+}
1313+flag "--url" help="Zitadel URL (env ZITADEL_URL)" {
1414+ arg "<url>"
1515+}
1616+flag "--token-file" help="File with a personal access token (env ZITADEL_TOKEN_FILE, or the token in ZITADEL_TOKEN)" {
1717+ arg "<path>"
1818+}
···11+# Invites a person to projects of a Zitadel organization: creates a human
22+# user for the email address (username = the address), lets Zitadel mail
33+# them an invite code to set up a password, and adds a user grant on each
44+# project. An existing user with that address only gets the missing grants.
55+import argparse
66+import json
77+import os
88+import sys
99+import urllib.error
1010+import urllib.request
1111+1212+parser = argparse.ArgumentParser(
1313+ prog="zitadel-invite",
1414+ description="Invite a person to Zitadel projects.",
1515+ epilog="It needs the token of a user with the ORG_USER_MANAGER role.",
1616+)
1717+parser.add_argument("email")
1818+parser.add_argument("projects", nargs="+", metavar="project")
1919+parser.add_argument("--name", help='"First Last", default: the local part')
2020+parser.add_argument(
2121+ "--resend",
2222+ action="store_true",
2323+ help="send a new invite code to an existing user who has not set up "
2424+ "a login yet",
2525+)
2626+parser.add_argument(
2727+ "--app-name", help="application name in the invite mail"
2828+)
2929+parser.add_argument(
3030+ "--url",
3131+ default=os.environ.get("ZITADEL_URL"),
3232+ help="Zitadel URL, e.g. https://id.example.org (env ZITADEL_URL)",
3333+)
3434+parser.add_argument(
3535+ "--token-file",
3636+ default=os.environ.get("ZITADEL_TOKEN_FILE"),
3737+ help="file with a personal access token (env ZITADEL_TOKEN_FILE, or "
3838+ "the token itself in ZITADEL_TOKEN)",
3939+)
4040+args = parser.parse_args()
4141+4242+if not args.url:
4343+ parser.error("--url or ZITADEL_URL is needed")
4444+if args.token_file:
4545+ with open(args.token_file) as f:
4646+ token = f.read().strip()
4747+else:
4848+ token = os.environ.get("ZITADEL_TOKEN")
4949+if not token:
5050+ parser.error("--token-file, ZITADEL_TOKEN_FILE or ZITADEL_TOKEN is needed")
5151+issuer = args.url.rstrip("/")
5252+5353+5454+def fetch(req, what):
5555+ try:
5656+ with urllib.request.urlopen(req) as res:
5757+ return json.load(res)
5858+ except urllib.error.HTTPError as e:
5959+ sys.exit(f"{what}: {e.code} {e.read().decode()}")
6060+6161+6262+def call(method, path, body=None):
6363+ return fetch(urllib.request.Request(
6464+ issuer + path,
6565+ method=method,
6666+ headers={
6767+ "Authorization": "Bearer " + token,
6868+ "Content-Type": "application/json",
6969+ },
7070+ data=None if body is None else json.dumps(body).encode(),
7171+ ), f"{method} {path}")
7272+7373+7474+def invite(user_id):
7575+ send_code = {}
7676+ if args.app_name:
7777+ send_code["applicationName"] = args.app_name
7878+ call("POST", f"/v2/users/{user_id}/invite_code", {"sendCode": send_code})
7979+ print(f"user {args.email}: invite mail sent")
8080+8181+8282+# check the projects before changing anything
8383+project_ids = {}
8484+for name in args.projects:
8585+ found = [
8686+ p for p in call("POST", "/management/v1/projects/_search", {
8787+ "queries": [{"nameQuery": {
8888+ "name": name,
8989+ "method": "TEXT_QUERY_METHOD_EQUALS",
9090+ }}],
9191+ }).get("result", [])
9292+ if p["name"] == name
9393+ ]
9494+ if len(found) != 1:
9595+ sys.exit(f"project {name}: {len(found)} matches")
9696+ project_ids[name] = found[0]["id"]
9797+9898+users = call("POST", "/management/v1/users/_search", {"queries": [
9999+ {"emailQuery": {
100100+ "emailAddress": args.email,
101101+ "method": "TEXT_QUERY_METHOD_EQUALS_IGNORE_CASE",
102102+ }},
103103+ {"typeQuery": {"type": "TYPE_HUMAN"}},
104104+]}).get("result", [])
105105+if len(users) > 1:
106106+ sys.exit(f"user {args.email}: {len(users)} users with this address")
107107+108108+if users:
109109+ user = users[0]
110110+ user_id = user["id"]
111111+ # Anyone can register with any address. Grants go only to users who
112112+ # verified theirs, or who were created by an admin and have no login yet.
113113+ initial = user.get("state") == "USER_STATE_INITIAL"
114114+ verified = user["human"].get("email", {}).get("isEmailVerified", False)
115115+ if not initial and not verified:
116116+ sys.exit(
117117+ f"user {args.email}: exists ({user['userName']}) with the "
118118+ "address unverified, not granting"
119119+ )
120120+ print(f"user {args.email}: exists ({user['userName']})")
121121+ if args.resend:
122122+ if initial:
123123+ invite(user_id)
124124+ else:
125125+ print(f"user {args.email}: has a login already, no invite")
126126+else:
127127+ full_name = args.name or args.email.split("@")[0]
128128+ given, _, family = full_name.partition(" ")
129129+ org_id = call("GET", "/management/v1/orgs/me")["org"]["id"]
130130+ user_id = call("POST", "/v2/users/new", {
131131+ "organizationId": org_id,
132132+ "human": {
133133+ "profile": {
134134+ "givenName": given,
135135+ # Zitadel requires one
136136+ "familyName": family.strip() or given,
137137+ "displayName": full_name,
138138+ },
139139+ # verified by the invite code, without a separate mail
140140+ "email": {"email": args.email, "returnCode": {}},
141141+ },
142142+ })["id"]
143143+ print(f"user {args.email}: created")
144144+ invite(user_id)
145145+146146+granted = {
147147+ g["projectId"]
148148+ for g in call("POST", "/management/v1/users/grants/_search", {
149149+ "queries": [{"userIdQuery": {"userId": user_id}}],
150150+ }).get("result", [])
151151+}
152152+for name, project_id in project_ids.items():
153153+ if project_id in granted:
154154+ print(f"project {name}: granted already")
155155+ else:
156156+ call("POST", f"/management/v1/users/{user_id}/grants", {
157157+ "projectId": project_id,
158158+ })
159159+ print(f"project {name}: granted")