in lambdas/kf_profile_manager/index.py [0:0]
def apply_rolebinding(user_name: str, user_email: str, group: str):
custom_objects_api = client.CustomObjectsApi(api_client)
rolebindings = custom_objects_api.list_namespaced_custom_object(
group="rbac.authorization.k8s.io",
version="v1",
namespace=f"{group}",
plural="rolebindings",
pretty="true"
)
create_rb = True # by default create profile
# iterate over rolebindings to find if new one should be created or old one exists
for rb in rolebindings['items']:
rb_metadata_name = rb['metadata']['name']
if rb_metadata_name == f"user-{user_name}-kubeflow-org-clusterrole-edit":
logger.info(
f"rolebinding user-{user_name}-kubeflow-org-clusterrole-edit in namespace={group} already exists")
create_rb = False
if create_rb:
manifest_rolebinding = {
"apiVersion": "rbac.authorization.k8s.io/v1",
"kind": "RoleBinding",
"metadata": {
"annotations": {
"role": "edit",
"user": user_email
},
"name": f"user-{user_name}-kubeflow-org-clusterrole-edit",
"namespace": f"{group}"
},
"roleRef": {
"apiGroup": "rbac.authorization.k8s.io",
"kind": "ClusterRole",
"name": "kubeflow-edit"
},
"subjects": [
{
"apiGroup": "rbac.authorization.k8s.io",
"kind": "User",
"name": user_email
}
]
}
rolebindings = custom_objects_api.create_namespaced_custom_object(
group="rbac.authorization.k8s.io",
version="v1",
namespace=f"{group}",
plural="rolebindings",
body=manifest_rolebinding,
pretty="true"
)