dynatrace_users_crud_for_admins.py

import json
from typing import List

import requests


class DynatraceOauthClient:

    def __init__(self, client_id: str, client_secret: str, sso_url: str = "https://sso.dynatrace.com/sso/oauth2/token"):
        self._client_id = client_id
        self._client_secret = client_secret
        self._sso_url = sso_url
        self.bearer = None

    def authorise(self):
        headers = {
            'Content-Type': "application/x-www-form-urlencoded"
        }
        payload = {
            'grant_type': 'client_credentials',
            'client_id': self._client_id,
            'client_secret': self._client_secret
        }

        _r = requests.post(self._sso_url, headers=headers, data=payload)
        try:
            self.bearer = _r.json()["access_token"]
        except:
            raise Exception(_r.status_code, _r.content, _r.reason)
        return self


def _response_logger(response_obj):
    """in case you don't want to use standard log"""
    text = response_obj.text if response_obj.text else response_obj.reason
    print(f'response  url:   {response_obj.url}\n'
          f'response code:   {response_obj.status_code}\n'
          f'elapsed  time:   {response_obj.elapsed}\n'
          f'response body:   {text}\n')


def _get_user_by_email(user_acc_uuid: str, user_email: str, access_token: str):
    headers = {
        'Authorization': f'Bearer {access_token}'
    }
    _r = requests.get(f'https://api.dynatrace.com/iam/v1/accounts/{user_acc_uuid}/users/{user_email}', headers=headers)
    return _r


def _set_default_groups(user_acc_uuid: str, user_email: str, access_token: str):
    headers = {
        'Content-Type': "application/json; charset=utf-8",
        'Accept': '*/*',
        'Authorization': f'Bearer {access_token}'
    }
    # e.g monitoring viewer, logs viewer
    data = json.dumps(["xxxxx-xxx-xxx-xx-xxxxx", "xxxxx-xxx-xxx-xx-xxxxx"])
    url = f'https://api.dynatrace.com/iam/v1/accounts/{user_acc_uuid}/users/{user_email}/groups'
    _r = requests.put(url=url, data=data, headers=headers)
    return _r


def _create_new_user(user_acc_uuid: str, user_email: str, access_token: str):
    headers = {
        'Content-Type': "application/json; charset=utf-8",
        'Accept': '*/*',
        'Authorization': f'Bearer {access_token}'
    }
    # monitoring viewer, logs viewer
    data = json.dumps({
        "email": user_email
    })
    url = f'https://api.dynatrace.com/iam/v1/accounts/{user_acc_uuid}/users'
    _r = requests.post(url=url, data=data, headers=headers)
    return _r


def _delete_user(user_acc_uuid: str, user_email: str, access_token: str):
    headers = {
        'Authorization': f'Bearer {access_token}'
    }
    url = f'https://api.dynatrace.com/iam/v1/accounts/{user_acc_uuid}/users/{user_email}'
    _r = requests.delete(url=url, headers=headers)
    return _r


'''crud users  https://www.dynatrace.com/support/help/dynatrace-api/basics/dynatrace-api-authentication/account-api-authentication'''
if __name__ == '__main__':
    # acc_uuid, oauth_acc_client, oauth_acc_secret - store it in secure place!
    CLIENT_CFG: List[str] = [acc_uuid, oauth_acc_client, oauth_acc_secret]

    USER_CRUD = True
    CREATE_NEW_USERS = False
    DELETE_USERS = False
    SET_BASE_PERMISSIONS = False
    CHECK_RESULTS = True

    EMAILS_FOR_ACTION: List[str] = [
        "xxx@gmail.com"
    ]

    if USER_CRUD:
        token = DynatraceOauthClient(
            client_id=oauth_acc_client, client_secret=oauth_acc_secret
        ).authorise().bearer

        if CREATE_NEW_USERS:
            print("\ncreate new users")
            for email in EMAILS_FOR_ACTION:
                resp = _create_new_user(user_acc_uuid=acc_uuid, user_email=email, access_token=token)
                _response_logger(response_obj=resp)

        if SET_BASE_PERMISSIONS:
            print("\nset permissions")
            for email in EMAILS_FOR_ACTION:
                resp = _set_default_groups(user_acc_uuid=acc_uuid, user_email=email, access_token=token)
                _response_logger(response_obj=resp)

        if DELETE_USERS:
            """bulk delete email pattern"""
            keyword = 'gmail.com'
            print("\ndelete users")
            for email in EMAILS_FOR_ACTION:
                if keyword in email:
                    resp = _delete_user(user_acc_uuid=acc_uuid, user_email=email, access_token=token)
                    _response_logger(response_obj=resp)

        if CHECK_RESULTS:
            print("\ncheck results")
            for email in EMAILS_FOR_ACTION:
                resp = _get_user_by_email(user_acc_uuid=acc_uuid, user_email=email, access_token=token)
                _response_logger(response_obj=resp)