Retrieve Access Bearer Token

/**
 * Delivers the csp-auth-token (CSP = Cloud Services Platform) or the
 * refresh_token with the bearer token for the HOL in the log with
 * fixed values of user name and user password.
 *
 * @author Stefan Schnell
 *
 * Checked with VMware Aria Automation 8.5.1, 8.12.0, 8.16.2 and 8.18.0
 */

var restHost = RESTHostManager.createHost("dynamicRequest");
var httpRestHost = RESTHostManager.createTransientHostFrom(restHost);

httpRestHost.operationTimeout = 60;
httpRestHost.connectionTimeout = 30;
httpRestHost.hostVerification = false;

/**
 * Access data for HOL with release 8.5.1
 *
 * httpRestHost.url = "https://vr-automation.corp.local";
 * var oLogin = {
 *   "username": "holadmin@corp.local",
 *   "password": "VMware1!"
 * }
 */

/**
 * Access data for HOL with release 8.12.0
 * httpRestHost.url = "https://aria-auto.corp.vmbeans.com";
 * var oLogin = {
 *   "username": "holadmin@corp.vmbeans.com",
 *   "password": "VMware1!"
 * }
 */

// Access data for HOL with release 8.16.2 and 8.18.0
httpRestHost.url = "https://auto.vcf.sddc.lab";
var oLogin = {
  "username": "holadmin@vcf.holo.lab",
  "password": "VMware123!"
}

var jsonLogin = JSON.stringify(oLogin);

// csp-auth-token
// var request = httpRestHost.createRequest("POST",
//   "/csp/gateway/am/api/login", jsonLogin);

// refresh_token
var request = httpRestHost.createRequest("POST",
  "/csp/gateway/am/api/login?access_token", jsonLogin);

request.contentType = "application/json";

// Execute REST Request
System.log("REST Request: POST " + request.fullUrl);
var response = request.execute();

// REST call output
System.log("REST Response Status Code: " + response.statusCode);
System.log("REST Response Content: " + response.contentAsString);

// Bearer token
var oRefreshToken = JSON.parse(response.contentAsString);
var refreshToken = "{\"refreshToken\":\"" + oRefreshToken.refresh_token +
  "\"}";

request = httpRestHost.createRequest("POST",
  "/iaas/api/login", refreshToken);
request.contentType = "application/json";

// Execute REST Request
System.log("REST Request: POST " + request.fullUrl);
response = request.execute();

// REST call output
System.log("REST Response Status Code: " + response.statusCode);
System.log("REST Response Content: " + response.contentAsString);


var oBearerToken = JSON.parse(response.contentAsString);
var bearerToken = oBearerToken.token;
System.log(bearerToken);

import json

def handler(context: dict, inputs: dict) -> dict:
    """ Delivers the bearer token of the logged in user in the log
    and as return value.

    Checked with Aria Automation 8.18.0 and 9.0.0
    """

    bearerToken: str = context['getToken']()
    print(bearerToken)

    outputs: dict = {
        "status": "done",
        "bearerToken": bearerToken
    }

    return outputs

"""
@module de.stschnell

@version 0.1.0

@runtime python:3.10

@inputType in_url {string}
@inputType in_userName {string}
@inputType in_password {string}

@outputType Properties

Checked with Aria Automation 8.18.0
"""


import base64
import json
import ssl
import urllib.request


def request(
    url: str,
    user: str | None = None,
    password: str | None = None,
    bearerToken: str | None = None,
    method: str = "GET",
    body: dict = {},
    contentType: str = "application/json;charset=utf-8",
    accept: str = "application/json"
) -> dict | bytes | None:

    """ Executes a REST request

    @param {string} url - URL to execute the request
    @param {string} user
    @param {string} password
    @param {string} bearerToken
    @param {string} method - Method of request, e.g. GET, POST, etc
    @param {dictionary} body - Body of request
    @param {string} contentType - MIME type of the body for the request
    @param {string} accept - MIME type of response content
    @returns {dictionary or bytes or None}
    """

    returnValue: dict | bytes | None = None

    try:

        request: urllib.request.Request = urllib.request.Request(
            url = url,
            method = method,
            data = bytes(json.dumps(body).encode("utf-8"))
        )

        if user and password:
            authorization: str = base64.b64encode(
                bytes(user + ":" + password, "UTF-8")
            ).decode("UTF-8")
            request.add_header(
                "Authorization", "Basic " + authorization
            )

        if bearerToken:
            request.add_header(
                "Authorization", "Bearer " + bearerToken
            )

        request.add_header(
            "Content-Type", contentType
        )

        request.add_header(
            "Accept", accept
        )

        response = urllib.request.urlopen(
            request,
            context = ssl._create_unverified_context()
        )

        responseCode: int = response.status
        responseRead: bytes = response.read()

        if responseCode == 200 or responseCode == 202:
            if "json" in accept:
                returnValue = json.loads(responseRead)
            else:
                returnValue = responseRead

    except Exception as err:
        raise Exception(f"An error occurred at request - {err}") \
          from err

    return returnValue


def getBearerToken(
    url: str,
    username: str,
    password: str
) -> str | None:
    """ Gets the Bearer token

    @param {string} url
    @param {string} username
    @param {string} password
    """

    returnValue: str | None = None

    try:

        login: dict = {
            "username": username,
            "password": password
        }

        refreshToken: str = request(
            url = url + "/csp/gateway/am/api/login?access_token",
            method = "POST",
            body = login
        )["refresh_token"]

        bearerToken: str = request(
            url = url + "/iaas/api/login",
            method = "POST",
            body = { "refreshToken": refreshToken }
        )["token"]

        returnValue = bearerToken

    except Exception as err:
        raise ValueError(f"An error occurred at get Bearer token - {err}") \
            from err

    return returnValue


def handler(context: dict, inputs: dict) -> dict:
    """ Aria Automation standard handler, the main function.
    """

    url: str = "https://auto.vcf.sddc.lab"
    username: str = "holadmin@vcf.holo.lab"
    password: str = "VMware123!"

    # url: str = inputs["in_url"]
    # username: str = inputs["in_userName"]
    # password: str = inputs["in_password"]

    outputs: dict = {}

    try:

        bearerToken: str | None = getBearerToken(
            url,
            username,
            password
        )

        outputs = {
            "status": "done",
            "error": None,
            "result": bearerToken
        }

    except Exception as err:

        outputs = {
            "status": "incomplete",
            "error": repr(err),
            "result": None
        }

    return outputs