|
| 1 | +import os |
| 2 | +import functools |
| 3 | + |
| 4 | +import requests |
| 5 | + |
| 6 | + |
| 7 | +class KeyCloakAPIClient: |
| 8 | + def __init__(self): |
| 9 | + self.base_url = os.getenv("KEYCLOAK_BASE_URL") |
| 10 | + self.realm = os.getenv("KEYCLOAK_REALM") |
| 11 | + self.client_id = os.getenv("KEYCLOAK_CLIENT_ID", "coldfront") |
| 12 | + self.client_secret = os.getenv("KEYCLOAK_CLIENT_SECRET", "nomoresecret") |
| 13 | + |
| 14 | + self.token_url = ( |
| 15 | + f"{self.base_url}/realms/{self.realm}/protocol/openid-connect/token" |
| 16 | + ) |
| 17 | + |
| 18 | + @functools.cached_property |
| 19 | + def api_client(self): |
| 20 | + params = { |
| 21 | + "grant_type": "client_credentials", |
| 22 | + "client_id": self.client_id, |
| 23 | + "client_secret": self.client_secret, |
| 24 | + } |
| 25 | + r = requests.post(self.token_url, data=params).json() |
| 26 | + headers = { |
| 27 | + "Authorization": ("Bearer %s" % r["access_token"]), |
| 28 | + "Content-Type": "application/json", |
| 29 | + } |
| 30 | + session = requests.session() |
| 31 | + session.headers.update(headers) |
| 32 | + return session |
| 33 | + |
| 34 | + def create_group(self, group_name): |
| 35 | + url = f"{self.base_url}/admin/realms/{self.realm}/groups" |
| 36 | + payload = {"name": group_name} |
| 37 | + response = self.api_client.post(url, json=payload) |
| 38 | + |
| 39 | + # If group already exists, ignore and move on |
| 40 | + if response.status_code not in (201, 409): |
| 41 | + response.raise_for_status() |
| 42 | + |
| 43 | + def create_user(self, cf_username): |
| 44 | + """Helper function to create user in Keycloak, for testing purposes only""" |
| 45 | + url = f"{self.base_url}/admin/realms/{self.realm}/users" |
| 46 | + payload = { |
| 47 | + "username": cf_username, |
| 48 | + "enabled": True, |
| 49 | + "email": cf_username, |
| 50 | + } |
| 51 | + r = self.api_client.post(url, json=payload) |
| 52 | + r.raise_for_status() |
| 53 | + |
| 54 | + def get_group_id(self, group_name) -> str | None: |
| 55 | + """Return None if group not found""" |
| 56 | + query = f"search={group_name}&exact=true" |
| 57 | + url = f"{self.base_url}/admin/realms/{self.realm}/groups?{query}" |
| 58 | + r = self.api_client.get(url).json() |
| 59 | + return r[0]["id"] if r else None |
| 60 | + |
| 61 | + def get_user_id(self, cf_username) -> str | None: |
| 62 | + """Return None if user not found""" |
| 63 | + # TODO (Quan): Confirm that Coldfront usernames map to Keycloak emails, not email, or something else? |
| 64 | + query = f"username={cf_username}&exact=true" |
| 65 | + url = f"{self.base_url}/admin/realms/{self.realm}/users?{query}" |
| 66 | + r = self.api_client.get(url).json() |
| 67 | + return r[0]["id"] if r else None |
| 68 | + |
| 69 | + def add_user_to_group(self, user_id, group_id): |
| 70 | + url = f"{self.base_url}/admin/realms/{self.realm}/users/{user_id}/groups/{group_id}" |
| 71 | + r = self.api_client.put(url) |
| 72 | + r.raise_for_status() |
| 73 | + |
| 74 | + def remove_user_from_group(self, user_id, group_id): |
| 75 | + url = f"{self.base_url}/admin/realms/{self.realm}/users/{user_id}/groups/{group_id}" |
| 76 | + r = self.api_client.delete(url) |
| 77 | + r.raise_for_status() |
| 78 | + |
| 79 | + def get_user_groups(self, user_id) -> list[str]: |
| 80 | + url = f"{self.base_url}/admin/realms/{self.realm}/users/{user_id}/groups" |
| 81 | + r = self.api_client.get(url) |
| 82 | + r.raise_for_status() |
| 83 | + return [group["name"] for group in r.json()] |
0 commit comments