-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
139 lines (110 loc) · 5.07 KB
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
import random
import string
import uuid
import survey
from kubernetes import client, config
# Configs can be set in Configuration class directly or using helper utility
config.load_kube_config()
v1 = client.CoreV1Api()
custom = client.CustomObjectsApi()
def select_resource(resource_type, filter_function, message):
ret = custom.list_cluster_custom_object(group="kubermatic.k8c.io", version="v1", plural=resource_type)
resources = [resource for resource in ret['items'] if filter_function(resource)]
resource_names = [resource['spec'][resource_type == 'users' and 'email' or 'name'] for resource in resources]
if not resources:
print(f"There are no {resource_type} that meet your criteria!")
exit()
index = survey.routines.select(message, options=resource_names, focus_mark='> ',
evade_color=survey.colors.basic('yellow'))
return resources[index]
def create_binding(project, user, user_type):
return {
"apiVersion": "kubermatic.k8c.io/v1",
"kind": "UserProjectBinding",
"metadata": {"generation": 1, "name": generate_random_name()},
"spec": {
"group": f"{user_type}-{project['metadata']['name']}",
"projectID": project['metadata']['name'],
"userEmail": user['spec']['email']
}
}
def generate_random_name():
return ''.join(random.choices(string.ascii_lowercase + string.digits, k=10))
def select_admin():
return select_resource("users", lambda user: user['spec']['admin'], 'Which admin do you want to add? ')
def select_non_admin():
return select_resource("users", lambda user: not user['spec']['admin'], 'Which user do you want to promote? ')
def select_project(filter=[]):
return select_resource("projects", lambda project: project['metadata']['name'] not in filter,
'Which project do you want to work on? ')
def add_member():
admin = select_admin()
ret = custom.list_cluster_custom_object(group="kubermatic.k8c.io", version="v1", plural="userprojectbindings")
bindings = [binding for binding in ret['items'] if binding['spec']['userEmail'] == admin['spec']['email']]
existing_projects = [project['spec']['projectID'] for project in bindings]
project = select_project(existing_projects)
binding = {
"apiVersion": "kubermatic.k8c.io/v1",
"kind": "UserProjectBinding",
"metadata": {
"finalizers": [
"kubermatic.k8c.io/cleanup-seed-user-project-bindings"
],
"generation": 1,
"name": ''.join(random.choices(string.ascii_lowercase + string.digits, k=10)),
"ownerReferences": [
{
"apiVersion": "kubermatic.k8c.io/v1",
"kind": "Project",
"name": project['metadata']['name'],
"uid": project['metadata']['uid']
}
],
"uid": str(uuid.uuid4())
},
"spec": {
"group": f"owners-{project['metadata']['name']}",
"projectID": project['metadata']['name'],
"userEmail": admin['spec']['email']
}
}
custom.create_cluster_custom_object(group="kubermatic.k8c.io", version="v1", plural="userprojectbindings",
body=binding)
def make_admin():
user = select_non_admin()
patch = {
"spec": {"admin": True}
}
custom.patch_cluster_custom_object(group="kubermatic.k8c.io", version="v1", plural="users",
name=user['metadata']['name'], body=patch)
def make_non_admin():
admin = select_admin()
patch = {
"spec": {"admin": False}
}
custom.patch_cluster_custom_object(group="kubermatic.k8c.io", version="v1", plural="users",
name=admin['metadata']['name'], body=patch)
def remove_member():
project = select_project()
ret = custom.list_cluster_custom_object(group="kubermatic.k8c.io", version="v1", plural="userprojectbindings")
bindings = [binding for binding in ret['items'] if binding['spec']['projectID'] == project['metadata']['name']]
user_names = [binding['spec']['userEmail'] for binding in bindings]
index = survey.routines.select('Which user do you want to remove? ', options=user_names, focus_mark='> ',
evade_color=survey.colors.basic('yellow'))
binding = bindings[index]
custom.delete_cluster_custom_object(group="kubermatic.k8c.io", version="v1", plural="userprojectbindings",
name=binding['metadata']['name'])
mode_list = ["Make someone admin", "Make someone a regular user", "Add an admin to a project",
"Remove user from project"]
index = survey.routines.select('What do you want to do? ', options=mode_list, focus_mark='> ',
evade_color=survey.colors.basic('yellow'))
if index == 0:
make_admin()
elif index == 1:
make_non_admin()
elif index == 2:
add_member()
elif index == 3:
remove_member()
else:
print("Bad choice!")