Skip to main content

Nx Witness Authentication APIs

Christian Sahar
Updated

This article will describe each authentication API and demonstrate how developers can utilize their functionality in a brief code example.

All owner-level API requests in the /rest section of the API documentation require session authentication with a fresh bearer token to enforce password checks before executing API requests. Please review the VMS Server API documentation for more information on how to properly utilize parameters and execute the API requests.

HTTP Bearer/Session Token Authentication

Nx Witness 5.0 introduced HTTP Bearer/Session Token Authentication, an improved authentication method recommended over the deprecated methods from prior versions of Nx Witness.

A bearer token is a cryptic string generated by Nx Witness Server in response to a login request. An Nx Witness client must send this token in the authorization header when making requests to protected resources. This method can only be used over HTTPS or RTSPS. Raw HTTP or RTSP requests cannot be made. 

Local and LDAP Users

The following steps demonstrate how to authenticate API requests for Local and LDAP users:

  1. [Optional] Check user type on the VMS Server to confirm they are not a Cloud user.
     
  2. Execute a login request to the VMS Server to obtain two bearer tokens: one to initiate the API call and another to delete the first. For cookie-based authentication, setCookie should be set to true. Afterward, all web browser requests to the VMS Server will be authorized automatically.
     
  3. [Optional] Check if the bearer token is valid on the VMS Server.
     
  4. Execute any request which requires authentication with a bearer token on the VMS Server.
    Note: Requests involving Video Walls require the Video Wall ID in the request. See the API documentation for more details.
     
  5. Terminate a session that is no longer needed on the VMS Server.
Click to see the code example
import requests
from pprint import pprint

LOCAL_USER = 'admin' # local account username
LOCAL_PASSWORD = 'pass123' # local account password
LOCAL_URL = 'https://localhost:7001' # https://<server_ip>:<sever_port> or https://{system_id}.relay.vmsproxy.com


def check_status(request, verbose):
    if request.status_code == requests.codes.ok:
        if verbose:
            print("Request successful\n{0}".format(request.text))
        return True
    print(request.url + " Request error {0}\n{1}".format(request.status_code, request.text))
    return False


def request_api(url, uri, method, **kwargs):
    server_url = f'{url}{uri}'
    response = requests.request(
        method,
        server_url,
        **kwargs
    )
    if not check_status(response, False):
        exit(1)
    if method == 'DELETE':
        return response
    return response.json()


def create_header(bearer_token):
    header = {"Authorization": f"Bearer {bearer_token}"}
    return header


def print_system_info(response):
    if 'reply' in response:
        system_info = response['reply']
        number_of_servers = len(system_info)
        system_name = system_info[0]['systemName']
    else:
        system_info = response
        number_of_servers = len(system_info)
        system_name = system_info[0]['systemName']
    print(f'System {system_name} contains {number_of_servers} server(s):')
    pprint(system_info)

def create_local_payload():
        payload = {
        'username': LOCAL_USER,
        'password': LOCAL_PASSWORD,
        'setCookie': False
    }
    return payload


def is_local_user(api_response):
    if api_response['username'] == 'admin':
        return True
    elif api_response['type'] == 'cloud':
        return False


def get_cloud_system_id(api_response):
    cloud_system_id = api_response['cloudId']
    return cloud_system_id


def main():
    # STEP 1
    cloud_state = request_api(LOCAL_URL, f'/rest/v1/login/users/{LOCAL_USER}', 'GET', verify=False)
    if not is_local_user(cloud_state):
        print(LOCAL_USER + ' is not a local user.')
        exit(1)

    # STEP 2
    payload = create_local_payload()
    primary_session = request_api(LOCAL_URL, '/rest/v1/login/sessions', 'POST', verify=False, json=payload)
    primary_token = primary_session['token']

    secondary_session = request_api(LOCAL_URL, '/rest/v1/login/sessions', 'POST', verify=False, json=payload)
    secondary_token = secondary_session['token']

    # STEP 3
    primary_token_info = request_api(LOCAL_URL, f'/rest/v1/login/sessions/{primary_token}', 'GET', verify=False)
    if is_expired(primary_token_info):
        print('Expired token')
        exit(1)

    secondary_token_info = request_api(LOCAL_URL, f'/rest/v1/login/sessions/{secondary_token}', 'GET', verify=False)
    if is_expired(secondary_token_info):
        print('Expired token')
        exit(1)

    # STEP 4
    get_method_header = create_header(primary_token)
    system_info = request_api(LOCAL_URL, f'/rest/v1/servers/*/info', 'GET', verify=False,
                              headers=get_method_header)
    print_system_info(system_info)

    # STEP 5
    delete_method_header = create_header(secondary_token)

    request_api(LOCAL_URL, f'/rest/v1/login/sessions/{secondary_token}', 'DELETE', verify=False,
                headers=delete_method_header)


if __name__ == '__main__':
    main()

 

You can also find this example in the Nx GitHub repository.

Cloud Users

The following steps demonstrate how to authenticate API requests for Cloud users:

  1. Check if the system is connected to the Cloud and get its Cloud System ID from the VMS Server. If the Cloud System ID is not listed, the system is not connected to the Cloud and does not have any Cloud users.
     
  2. [Optional] Check the user type on the VMS Server. Only Cloud users may use Cloud Sessions.
    Note: Field methods will not contain sessions because the VMS Server does not give sessions for Cloud users.
     
  3. Cloud users must obtain two bearer tokens from the Cloud: one to initiate the API calls and another to delete the first token when finished.
     
  4. [Optional] Check if the session token is valid on the VMS Server. For cookie-based authentication, add the ?setCookie=true URL parameter. Afterward, all web browser requests to the VMS Server will be authorized automatically.
     
  5. Execute any request which requires authentication with a bearer token on the VMS Server.
     
  6. Cloud sessions can be terminated on the Cloud when it's no longer needed.
Click to see the code example
import requests
from pprint import pprint

LOCAL_USER = 'admin'  # local account username
LOCAL_PASSWORD = 'pass123'  # local account password
LOCAL_URL = 'https://localhost:7001'  # https://<server_ip>:<sever_port> or https://{system_id}.relay.vmsproxy.com
CLOUD_USER = 'user@gmail.com'  # cloud account email
CLOUD_PASSWORD = 'pass123'  # cloud account password
CLOUD_DOMAIN_NAME = 'nxvms.com'  # cloud service domain name
CLOUD_URL = 'https://' + CLOUD_DOMAIN_NAME


def check_status(request, verbose):
    if request.status_code == requests.codes.ok:
        if verbose:
            print("Request successful\n{0}".format(request.text))
        return True
    print(request.url + " Request error {0}\n{1}".format(request.status_code, request.text))
    return False


def request_api(url, uri, method, **kwargs):
    server_url = f'{url}{uri}'
    response = requests.request(
        method,
        server_url,
        **kwargs
    )
    if not check_status(response, False):
        exit(1)
    if method == 'DELETE':
        return response
    return response.json()


def create_header(bearer_token):
    header = {"Authorization": f"Bearer {bearer_token}"}
    return header


def print_system_info(response):
    if 'reply' in response:
        system_info = response['reply']
        number_of_servers = len(system_info)
        system_name = system_info[0]['systemName']
    else:
        system_info = response
        number_of_servers = len(system_info)
        system_name = system_info[0]['systemName']
    print(f'System {system_name} contains {number_of_servers} server(s):')
    pprint(system_info)


def create_cloud_payload(cloud_system_id):
        return {
            'grant_type': 'password', 'response_type': 'token', 'client_id': '3rdParty',
            'scope': f'{CLOUD_URL}/cdb/oauth2/token cloudSystemId={cloud_system_id}',
            'username': CLOUD_USER, 'password': CLOUD_PASSWORD
        }


def get_cloud_system_id(api_response):
    cloud_system_id = api_response['cloudId']
    return cloud_system_id


def has_cloud_host(api_response):
    return api_response['cloudHost'] == CLOUD_DOMAIN_NAME


def is_cloud_user(api_response):
    if api_response['type'] != 'cloud':
        return False
    else:
        return True


def get_token(api_response):
    if int(api_response['expires_in']) < 1:
        print('Expired token')
        exit(1)
    bearer_token = api_response['access_token']
    return bearer_token


def is_expired(api_response):
    if int(api_response['expiresInS']) < 1:
        return True
    else:
        return False


def main():
    # STEP 1
    system_info = request_api(LOCAL_URL, '/rest/v1/system/info', 'GET', verify=False)
    if not has_cloud_host(system_info):
        print('Not a cloud-connected system')
        exit(1)
    cloud_system_id = get_cloud_system_id(system_info)

    # STEP 2
    user_info = request_api(LOCAL_URL, f'/rest/v1/login/users/{CLOUD_USER}', 'GET', verify=False)

    if not is_cloud_user(user_info):
        print(CLOUD_USER + ' is not a cloud user.')
        exit(1)

    # STEP 3
    oauth_payload = create_cloud_payload(cloud_system_id)
    oath_response = request_api(CLOUD_URL, f'/cdb/oauth2/token', 'POST', json=oauth_payload)
    primary_token = get_token(oath_response)

    oauth_payload = create_cloud_payload(cloud_system_id)
    oath_response = request_api(CLOUD_URL, f'/cdb/oauth2/token', 'POST', json=oauth_payload)
    secondary_token = get_token(oath_response)

    # STEP 4
    token_info = request_api(LOCAL_URL, f'/rest/v1/login/sessions/{primary_token}', 'GET', verify=False)

    if is_expired(token_info):
        print('Expired token')
        exit(1)

    # STEP 5
    primary_token_header = create_header(primary_token)
    system_info = request_api(LOCAL_URL, f'/rest/v1/servers/*/info', 'GET',
                              headers=primary_token_header, verify=False)
    print_system_info(system_info)
    secondary_token_header = create_header(secondary_token)

    # STEP 6
    request_api(CLOUD_URL, f'/cdb/oauth2/token/{primary_token}', 'DELETE', headers=secondary_token_header)


if __name__ == '__main__':
    main()

 

You can also find this example in the Nx GitHub repository.

Deprecated Authentication in Nx Witness

While examples for Digest Authentication are included in the Nx Github Repository, this method has been deprecated since version 5.0 and is kept solely for legacy support. It will be removed in a future release, so we strongly recommend using only the supported authentication methods.

Related articles

Comments

0 comments

Article is closed for comments.