REST API

  • 3 minute read
Prev Next

The admin UI for AppGate ZTNA exclusively uses REST API calls to the Controller. AppGate provides a sub-set of these Controller APIs written in OpenApi v3 format for general usage. If you specifically need to use an undocumented API call, contact Appgate support.

The Controller APIs will let you do pretty much everything the Admin can do through the UI. So if you want to add an Entitlement to a Policy then this can be scripted on an external system and actioned using the appropriate REST API call. For instance; this would allow some provisioning system to add the related Entitlement for a new server instance it has just created.

Getting started with the Controller REST API

You will need:

NOTE

Use only one Controller  for HA administration when using scripts that make repeated API REST calls. This avoids possible database conflicts when the same form is being edited at the same time on different Controllers.

The same admin rights that have been set up for that user in the admin UI apply to these REST APIs also. So an admin who only has admin rights for Entitlements will only be able to use get /entitlements, get /entitlements/{id}, put /entitlements/{id} and post /entitlements.

The code example below will listen to AppGate ZTNA audit logs (via rsyslog messages) and act on ALERT messages, such as removing users' on-boarding cookie or adding the user to the denylist.

Code Example using Controller APIs

import json
        import socketserver
        import sys
        import os
        import requests

        # edit the values below
        CONTROLLER_URL = "https://controller1.company.com:8443"
        CA_PATH = os.path.join(os.path.dirname(os.path.realpath(__file__)), "ca.pem")

        PROVIDER = "local"
        USERNAME = "username"
        PASSWORD = "password"
        API_VERSION = 22

        # do not edit anything below
        LISTEN_HOST, LISTEN_PORT = "127.0.0.1", 5140

        class SyslogHandler(socketserver.BaseRequestHandler):
        def handle(self):
        data = bytes.decode(self.request[0].strip())

        if len(data) == 0:
        print("zero len message received")
        return

        # parse to JSON
        jlog = json.loads(data)
        # print(jlog["event_type"] + " received")

        if jlog["event_type"] == "ip_access" and jlog["action"] == "alert":
        dn = jlog["distinguished_name"]
        print(f"ALERT received for {dn}")
        blacklist_user(dn)
        revoke_tokens(dn)
        else:
        # print("skipping " + jlog["event_type"])
        pass

        def login():
        headers = {"Accept": f"application/vnd.appgate.peer-v{API_VERSION}+json",
        "Content-Type": "application/JSON"}

        # authenticate
        data = {"machineId": "f0031c00-0522-43b3-a642-ae23cfd1bc22",
        "providerName": PROVIDER, "username": USERNAME, "password": PASSWORD}

        res = requests.post(f"{CONTROLLER_URL}/admin/login",
        verify=CA_PATH, headers=headers, data=json.dumps(data), timeout=5)

        if res.status_code != 200:
        print(f"CONTROLLER_URL login failed with http {str(res.status_code)}")
        sys.exit(-1)

        token = json.loads(res.text)["token"]
        headers["Authorization"] = f"Bearer {token}"

        return headers

        def blacklist_user(dn):
        auth_headers = login()

        # dn without deviceid
        user_dn = dn.split(",", 1)[1]
        data = {"userDistinguishedName": user_dn, "reason": "suspicious traffic"}

        res = requests.post(f"{CONTROLLER_URL}/admin/blacklist",
        data=json.dumps(data), verify=CA_PATH, headers=auth_headers)

        print(f"{user_dn} blacklisted {str(res.status_code)}")
        return res.status_code

        def revoke_tokens(dn):
        auth_headers = login()
        data = {"distinguishedNameFilter": dn, "revocationReason": "api revocation for alert", "delayMinutes": 0}
        res = requests.post(f"{CONTROLLER_URL}/admin/on-boarded-devices/revoke-tokens",
        data=json.dumps(data), verify=CA_PATH, headers=auth_headers)
        print(f"{dn} tokens revoked {str(res.status_code)}")
        return res.status_code

        if __name__ == "__main__":
        print("========================================================")
        print(f"Appgate SDP Alert Handler Started on {str(LISTEN_PORT)}")
        print("========================================================")

        with socketserver.UDPServer((LISTEN_HOST, LISTEN_PORT), SyslogHandler) as server:
        server.serve_forever(poll_interval=0.5)

AppGate reserves the right to change any of the REST API calls when a new version of AppGate ZTNA is released. Where any changes are made, the documentation will be updated accordingly.  Any script provided via AppGate Professional Service team or as example scripts from the manual are delivered as is, without warranty of any kind.

AppGate disclaims all implied warranties including, without limitation, any implied warranties of merchantability or of fitness for a particular purpose. Any maintenance needs arising out of the use or performance of these sample scripts (and related documentation) remains the responsibility of the customer.

Related articles