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);

/**
 * @module de.stschnell
 *
 * @version 0.1.0
 *
 * @param {string} in_userName
 * @param {SecureString} in_password
 *
 * @outputType String
 */
function getBearerToken(in_userName, in_password) {

// Begin ---------------------------------------------------------------

/**
 * Delivers the bearer token in the log with the user name and user
 * password as parameters.
 *
 * @author Stefan Schnell <mail@stefan-schnell.de>
 * @license MIT
 * @version 0.1.0
 *
 * Set com.vmware.scripting.javascript.allow-native-object in the
 * system properties to true.
 *
 * Checked with VMware Aria Automation 8.12.0 and 8.18.0.
 *
 * @example
 * // Release 8.12.0
 * var token = System.getModule("de.stschnell").getBearerToken(
 *   "holadmin@corp.vmbeans.com",
 *   "VMware1!"
 * );
 *
 * @example
 * // Release 8.18.0
 * var token = System.getModule("de.stschnell").getBearerToken(
 *   "holadmin@vcf.holo.lab",
 *   "VMware123!"
 * );
 */

function getFQDN() {
  var fqdn = "";
  var jvmOpts = java.lang.System.getenv("JVM_OPTS");
  if (jvmOpts !== null) {
    var options = jvmOpts.split(" ");
    options.forEach( function(option) {
      if (option.substring(0, 19) === "-Dvco.app.hostname=") {
        fqdn = option.substring(19, option.length);
      }
    });
  }
  if (fqdn !== "") {
    return fqdn;
  } else {
    return null;
  }
}

function retrieveBearerToken(url, username, password) {
  var httpRestHost = RESTHostManager.createTransientHostFrom(
    RESTHostManager.createHost("dynamicRequest")
  );
  httpRestHost.operationTimeout = 60;
  httpRestHost.connectionTimeout = 30;
  httpRestHost.hostVerification = false;
  httpRestHost.url = url;
  var oLogin = {
    "username": username,
    "password": password
  };
  var jsonLogin = JSON.stringify(oLogin);
  var request = httpRestHost.createRequest(
    "POST",
    "/csp/gateway/am/api/login?access_token",
    jsonLogin
  );
  request.contentType = "application/json";
  var response = request.execute();
  if (response.statusCode === 200) {
    var oRefreshToken = JSON.parse(response.contentAsString);
    var refreshToken = "{\"refreshToken\":\"" +
      oRefreshToken.refresh_token + "\"}";
    request = httpRestHost.createRequest(
      "POST",
      "/iaas/api/login",
      refreshToken
    );
    request.contentType = "application/json";
    response = request.execute();
    if (response.statusCode === 200) {
      var oBearerToken = JSON.parse(response.contentAsString);
      var bearerToken = oBearerToken.token;
      return bearerToken;
    }
  }
}

// Main
if (
  String(in_userName).trim() !== "" &&
  String(in_password).trim() !== ""
) {
  var fqdn = getFQDN();
  if (fqdn !== null) {
    var bearerToken = retrieveBearerToken(
      "https://" + fqdn,
      in_userName,
      in_password
    );
    if (bearerToken !== null) {
      System.log(bearerToken);
      return bearerToken;
    }
  }
} else {
  throw new Error(
    "in_userName or in_password argument can not be null"
  );
}

// End -----------------------------------------------------------------

}

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
    """

    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