diff --git a/CHANGELOG.md b/CHANGELOG.md index 04c883e..de21d8c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## unreleased +* Add support for service accounts. Thanks, @einar-lanfranco and @feerbau. + ## 3.10.0 (2023-10-30) diff --git a/examples/service-account.py b/examples/service-account.py new file mode 100644 index 0000000..a5df8d3 --- /dev/null +++ b/examples/service-account.py @@ -0,0 +1,71 @@ +""" +About +===== + +Example program for interacting with the Grafana Service Account. + +Usage +===== + +Make sure to adjust `GrafanaApi` options at the bottom of the +file. +And replace aaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa with the uid of the folder where you want to add the service account. + +Synopsis +======== +:: + + source .venv/bin/activate + python examples/service-account.py + +""" +import uuid + +from grafana_client import GrafanaApi +from grafana_client.client import GrafanaClientError + + +def run_conversation(grafana: GrafanaApi): + # print("Grafana address") + # print(grafana.client.url) + + # print("Health check") + # print(grafana.health.check()) + + print("Create Service Account") + sa = {"name": "Sytem Account DEMO", "role": "Viewer"} + + # Add a new service account + try: + response = grafana.serviceaccount.create(sa) + print(response) + except GrafanaClientError as ex: + print(f"ERROR: {ex}") + + # Add a new token to service account, you need to writedown the token['key'] to use it later + # Genera an UUID + uuid_to_token = uuid.uuid4() + token_name = "%s" % (uuid_to_token) + + try: + token = grafana.serviceaccount.create_token(response["id"], {"name": token_name}) + print(token) + except GrafanaClientError as ex: + print(f"ERROR: {ex}") + + # Add Viewer permissions to service account created to folder with uid aaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa + uid_folder = "aaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa" + permissions = {"permission": "View"} + sa_id = response["id"] + grafana.folder.update_folder_permissions_for_user(uid_folder, sa_id, permissions) + + # Search service account by name + search_sa = grafana.serviceaccount.search_one(service_account_name="Sytem Account DEMO") + print(search_sa) + + +if __name__ == "__main__": + # Connect to custom Grafana instance. + grafana = GrafanaApi.from_url(url="localhost:3000", credential=("admin", "admin")) + + run_conversation(grafana) diff --git a/grafana_client/api.py b/grafana_client/api.py index 17b9748..7b58f3b 100644 --- a/grafana_client/api.py +++ b/grafana_client/api.py @@ -26,6 +26,7 @@ Plugin, Rbac, Search, + ServiceAccount, Snapshots, Teams, User, @@ -80,6 +81,7 @@ def __init__( self.snapshots = Snapshots(self.client) self.notifications = Notifications(self.client) self.plugin = Plugin(self.client) + self.serviceaccount = ServiceAccount(self.client) def connect(self): try: diff --git a/grafana_client/elements/__init__.py b/grafana_client/elements/__init__.py index e6d058d..bf98122 100644 --- a/grafana_client/elements/__init__.py +++ b/grafana_client/elements/__init__.py @@ -13,6 +13,31 @@ from .plugin import Plugin from .rbac import Rbac from .search import Search +from .service_account import ServiceAccount from .snapshots import Snapshots from .team import Teams from .user import User, Users + +__all__ = ( + "Admin", + "Alerting", + "AlertingProvisioning", + "Annotations", + "Base", + "Dashboard", + "DashboardVersions", + "Datasource", + "Folder", + "Health", + "Notifications", + "Organization", + "Organizations", + "Plugin", + "Rbac", + "Search", + "ServiceAccount", + "Snapshots", + "Teams", + "User", + "Users", +) diff --git a/grafana_client/elements/folder.py b/grafana_client/elements/folder.py index f2514d5..ba7b3f9 100644 --- a/grafana_client/elements/folder.py +++ b/grafana_client/elements/folder.py @@ -100,3 +100,17 @@ def update_folder_permissions(self, uid, items): update_folder_permissions_path = "/folders/%s/permissions" % uid r = self.client.POST(update_folder_permissions_path, json=items) return r + + def update_folder_permissions_for_user(self, uid, user_id, items): + """ + + :param uid: + :param user_id: + :param items: + {"permission": "View"} or {"permission": "Edit"} or {"permission": ""} + :return: + """ + + update_folder_permissions_path_for_user = "/access-control/folders/%s/users/%s" % (uid, user_id) + r = self.client.POST(update_folder_permissions_path_for_user, json=items) + return r diff --git a/grafana_client/elements/service_account.py b/grafana_client/elements/service_account.py new file mode 100644 index 0000000..eb700de --- /dev/null +++ b/grafana_client/elements/service_account.py @@ -0,0 +1,145 @@ +""" +https://grafana.com/docs/grafana/latest/developers/http_api/serviceaccount/ +https://grafana.com/docs/grafana/latest/administration/service-accounts/ +https://grafana.com/docs/grafana/latest/developers/http_api/create-api-tokens-for-org/ +""" +from .base import Base + + +class ServiceAccount(Base): + def __init__(self, client): + super(ServiceAccount, self).__init__(client) + self.client = client + self.path = "/serviceaccounts" + + def get(self, service_account_id): + """ + Get service account by id. + https://grafana.com/docs/grafana/latest/developers/http_api/serviceaccount/#get-a-service-account-by-id + + :param service_account_id: + :return: + """ + get_actual_user_path = "/serviceaccounts/%s?accesscontrol=true" % (service_account_id) + r = self.client.GET(get_actual_user_path) + return r + + def create(self, service_account): + """ + Create service account. + https://grafana.com/docs/grafana/latest/developers/http_api/serviceaccount/#create-service-account + + :param service_account: {"name": "string", "role": "string"} + :return: + """ + create_service_account_path = "/serviceaccounts/" + r = self.client.POST(create_service_account_path, json=service_account) + return r + + def delete(self, service_account_id): + """ + Delete service account. + https://grafana.com/docs/grafana/latest/developers/http_api/serviceaccount/#delete-service-account + + :param service_account: + :return: + """ + + delete_service_account_path = "/serviceaccounts/%s" % (service_account_id) + r = self.client.DELETE(delete_service_account_path) + return r + + def get_tokens(self, service_account_id): + """ + Get service account tokens. + https://grafana.com/docs/grafana/latest/developers/http_api/serviceaccount/#get-service-account-tokens + + :param service_account_id: + :return: + """ + service_account_tokens_path = "/serviceaccounts/%s/tokens" % (service_account_id) + r = self.client.GET(service_account_tokens_path) + return r + + def create_token(self, service_account_id, content): + """ + Create service account tokens + https://grafana.com/docs/grafana/latest/developers/http_api/serviceaccount/#create-service-account-tokens + + :param service_account_id: + :param service_account_name: + :return: + """ + create_service_account_token_path = "/serviceaccounts/%s/tokens" % (service_account_id) + r = self.client.POST(create_service_account_token_path, json=content) + return r + + def delete_token(self, service_account_id, service_account_token_id): + """ + Delete service account tokens. + https://grafana.com/docs/grafana/latest/developers/http_api/serviceaccount/#delete-service-account-tokens + + :param service_account_id: + :param service_account_token_id: + :return: + """ + delete_service_account_token_path = "/serviceaccounts/%s/tokens/%s" % ( + service_account_id, + service_account_token_id, + ) + r = self.client.DELETE(delete_service_account_token_path) + return r + + def search(self, query=None, page=None, perpage=None): + """ + + :return: + """ + list_of_sa = [] + sa_on_page = None + show_sa_path = "/serviceaccounts/search" + params = [] + if query: + params.append("query=%s" % query) + + if page: + iterate = False + params.append("page=%s" % page) + else: + iterate = True + params.append("page=%s") + page = 1 + + if perpage: + params.append("perpage=%s" % perpage) + + show_sa_path += "?" + show_sa_path += "&".join(params) + if iterate: + while True: + url = show_sa_path % page + sa_on_page = self.client.GET(url) + list_of_sa.append(sa_on_page) + if not sa_on_page["serviceAccounts"]: + break + page += 1 + else: + sa_on_page = self.client.GET(show_sa_path) + list_of_sa.append(sa_on_page) + + return list_of_sa + + def search_one(self, service_account_name=""): + """ + Find a single service account by name. Raises errors on multiple or no matches. + + :param service_account_name: + :return: + """ + s = self.search(query=service_account_name)[0] + if s["totalCount"] == 1: + return s["serviceAccounts"][0] + elif s["totalCount"] > 1: + raise ValueError("More than one service account matched") + else: + raise ValueError("No service account matched") diff --git a/test/elements/__init__.py b/test/elements/__init__.py index e69de29..8b13789 100644 --- a/test/elements/__init__.py +++ b/test/elements/__init__.py @@ -0,0 +1 @@ + diff --git a/test/elements/test_folder.py b/test/elements/test_folder.py index 732ca64..62f8aa5 100644 --- a/test/elements/test_folder.py +++ b/test/elements/test_folder.py @@ -246,6 +246,22 @@ def test_update_folder_permissions(self, m): ) self.assertEqual(folder["message"], "Folder permissions updated") + @requests_mock.Mocker() + def test_update_folder_permissions_for_user(self, m): + m.post( + "http://localhost/api/access-control/folders/nErXDvCkzz/users/12345", + json={"message": "Folder permissions updated"}, + ) + folder = self.grafana.folder.update_folder_permissions_for_user( + uid="nErXDvCkzz", + user_id="12345", + items=[ + {"permission": "View"}, + {"permission": "Edit"}, + ], + ) + self.assertEqual(folder["message"], "Folder permissions updated") + @requests_mock.Mocker() def test_delete_folder(self, m): m.delete("http://localhost/api/folders/nErXDvCkzz", json={"message": "Folder deleted"}) diff --git a/test/elements/test_service_account.py b/test/elements/test_service_account.py new file mode 100644 index 0000000..c280ae3 --- /dev/null +++ b/test/elements/test_service_account.py @@ -0,0 +1,138 @@ +import unittest + +import requests_mock + +from grafana_client import GrafanaApi + + +class ServiceAccountsTestCase(unittest.TestCase): + def setUp(self): + self.grafana = GrafanaApi(("admin", "admin"), host="localhost", url_path_prefix="", protocol="http") + + @requests_mock.Mocker() + def test_get(self, m): + m.get( + "http://localhost/api/serviceaccounts/42?accesscontrol=true", + json={ + "id": 42, + "name": "rjuan", + "login": "sa-juan", + "orgId": 1, + "isDisabled": False, + "createdAt": "2023-12-15T21:38:59Z", + "updatedAt": "2023-12-19T12:46:06Z", + "avatarUrl": "/avatar/06b4d9db72de4d293813135da09fd736", + "role": "Viewer", + "teams": None, + "accessControl": { + "serviceaccounts.permissions:read": True, + "serviceaccounts.permissions:write": True, + "serviceaccounts:delete": True, + "serviceaccounts:read": True, + "serviceaccounts:write": True, + }, + }, + ) + result = self.grafana.serviceaccount.get(42) + self.assertEqual( + result, + { + "id": 42, + "name": "rjuan", + "login": "sa-juan", + "orgId": 1, + "isDisabled": False, + "createdAt": "2023-12-15T21:38:59Z", + "updatedAt": "2023-12-19T12:46:06Z", + "avatarUrl": "/avatar/06b4d9db72de4d293813135da09fd736", + "role": "Viewer", + "teams": None, + "accessControl": { + "serviceaccounts.permissions:read": True, + "serviceaccounts.permissions:write": True, + "serviceaccounts:delete": True, + "serviceaccounts:read": True, + "serviceaccounts:write": True, + }, + }, + ) + + @requests_mock.Mocker() + def test_create(self, m): + m.post("http://localhost/api/serviceaccounts/", json={"message": "Service account created"}) + user = self.grafana.serviceaccount.create({"name": "foo", "role": "Admin"}) + self.assertEqual(user["message"], "Service account created") + + @requests_mock.Mocker() + def test_delete(self, m): + m.delete("http://localhost/api/serviceaccounts/42", json={"message": "Service account deleted"}) + user = self.grafana.serviceaccount.delete(42) + self.assertEqual(user["message"], "Service account deleted") + + @requests_mock.Mocker() + def test_create_token(self, m): + m.post("http://localhost/api/serviceaccounts/42/tokens", json={"message": "Service account token created"}) + user = self.grafana.serviceaccount.create_token(42, {"name": "some-uuid"}) + self.assertEqual(user["message"], "Service account token created") + + @requests_mock.Mocker() + def test_delete_token(self, m): + m.delete("http://localhost/api/serviceaccounts/42/tokens/2", json={"message": "Service account token deleted"}) + user = self.grafana.serviceaccount.delete_token(42, 2) + self.assertEqual(user["message"], "Service account token deleted") + + @requests_mock.Mocker() + def test_get_tokens_some(self, m): + m.get("http://localhost/api/serviceaccounts/42/tokens", json=["token1", "token2"]) + result = self.grafana.serviceaccount.get_tokens(42) + self.assertEqual(len(result), 2) + + @requests_mock.Mocker() + def test_get_tokens_zero(self, m): + m.get("http://localhost/api/serviceaccounts/42/tokens", json=[]) + result = self.grafana.serviceaccount.get_tokens(42) + self.assertEqual(len(result), 0) + + @requests_mock.Mocker() + def test_search(self, m): + # TODO: Don't know how the shape of the response looks like. + m.get("http://localhost/api/serviceaccounts/search?query=foo&page=3&perpage=10", json={"foo": "bar"}) + result = self.grafana.serviceaccount.search("foo", page=3, perpage=10) + self.assertEqual(result, [{"foo": "bar"}]) + + @requests_mock.Mocker() + def test_search_one_success(self, m): + m.get( + "http://localhost/api/serviceaccounts/search?query=foo&page=1", + json={"totalCount": 1, "serviceAccounts": [{"foo": "bar"}]}, + ) + m.get( + "http://localhost/api/serviceaccounts/search?query=foo&page=2", + json={"totalCount": 1, "serviceAccounts": []}, + ) + result = self.grafana.serviceaccount.search_one("foo") + self.assertEqual(result, {"foo": "bar"}) + + @requests_mock.Mocker() + def test_search_one_find_two(self, m): + m.get( + "http://localhost/api/serviceaccounts/search?query=foo&page=1", + json={"totalCount": 2, "serviceAccounts": [{"foo": "bar"}, {"foo": "baz"}]}, + ) + m.get( + "http://localhost/api/serviceaccounts/search?query=foo&page=2", + json={"totalCount": 2, "serviceAccounts": []}, + ) + with self.assertRaises(ValueError) as ex: + self.grafana.serviceaccount.search_one("foo") + self.assertEqual("More than one service account matched", str(ex.exception)) + + @requests_mock.Mocker() + def test_search_one_find_zero(self, m): + m.get( + "http://localhost/api/serviceaccounts/search?query=foo&page=1", + json={"totalCount": 0, "serviceAccounts": []}, + ) + with self.assertRaises(ValueError) as ex: + self.grafana.serviceaccount.search_one("foo") + self.assertEqual("No service account matched", str(ex.exception))