Crosswork Cloud APIs are for programmers who want to use the APIs with their network management and operations applications.

The Crosswork Cloud Network Insights API allows you to perform configuration tasks such as subscribing to prefixes or ASNs, configuring notification endpoints, and specifying conditions under which an alarm is triggered. The Crosswork Cloud Traffic Analysis API retrieves traffic statistics.

You must be logged into Crosswork Cloud to access the Crosswork Cloud API documentation. To view API call definitions and documentation, either navigate to > APIs or go to https://crosswork.cisco.com/apiDoc/CiscoCrossworkCloudAPI.

Join the Cisco Community Here to access the Crosswork Developer Hub. You can also access the Cisco Community by navigating to > Support > Community Forum. Make sure to use and subscribe to the "Crosswork" label to help identify Crosswork Cloud discussions.

A Crosswork Cloud API Key consists of:

• An API Key, which is a hex encoded, 32-byte symmetric key. Client applications use the API Key to sign REST API requests destined for Crosswork Cloud Network Insights or Crosswork Cloud Traffic Analysis.

• An API Key identifier (ID), which is a unique value for the key and must be included with each signed request. Crosswork Cloud services use the Key ID to retrieve a copy of the API Key to verify the incoming request.

Note	Protect your API Key as you would a password. Because the API Key provides access to your account, make sure you store it securely.

A client application uses the API Key to sign all requests that are sent to Crosswork Cloud. Each request includes:

• The request signature

• The API Key ID

• Metadata detailing the fields used to determine the signature

After Crosswork Cloud receives a REST API request, it performs the following steps:

1. Extracts the requested parameters.

2. Uses the API Key ID to retrieve the API Key and associated metadata.

3. Recalculates the signature.

4. Compares the calculated signature with the requested signature.

5. If the calculated and requested signatures match, Crosswork Cloud forwards the request. If the signatures do not match, Crosswork Cloud rejects the request.

The following script examples are written in Python. You will need python/get_traffic_example.py and python/cctraffic/cctraffic.py to execute the Crosswork Traffic Analysis APIs. Prior to running get_traffic_example.py, you must do the following:

1. Install python dependencies: pip3 install -r requirements.txt

2. Set the API bearer token: export TOKEN=<token string>

3. Edit the get_traffic_example.py file. Replace the following with the correct values: api_version, device_name, start and end.

After editing the get_traffic_example.py file, run the script: python3 get_traffic_example.py

Script Example: get_traffic_example.py

# get_traffic_example.py

import os
import sys
from cctraffic import CCTrafficRestClient

host = "https://crosswork.cisco.com"
api_version = "beta"
device_name = "flow-automation-1"

# start and end may be supplied as:
#   - ISO 8601 datetime string
#   - unix timestamp in seconds since 1970
#   - now
#   - "<number> <unit> ago" where unit can be: "seconds", "minutes", "hours", "days".
start = "7 days ago"
end = "now"

if "TOKEN" in os.environ:
    token = os.environ["TOKEN"]
else:
    print("Bearer token not found. Set bearer token with: export TOKEN=<token string>")
    sys.exit(-1)

client = CCTrafficRestClient(host, token, version=api_version, debug=False)

print(f"GetDevice for {device_name}")
device_info = client.GetDevice(device_name)
device_id = device_info["deviceId"]
print(f"Found device ID for {device_name}: {device_id}")

print(f"Traffic by interface for {device_name}")
traffic_for_my_device = client.GetInterfaceCounterTrafficTotals(start, end, device_id)
interface_name = traffic_for_my_device[0]["interfaces"][0]["interfaceName"]

print(f"Traffic by ASN for {device_name}/{interface_name}")
asn_traffic_for_my_device_interface = client.GetNetFlowTrafficTotalsByDevice(start, end, device_id, interface=interface_name, asn_breakdown=True)

print(f"Traffic by Prefix for {device_name}/{interface_name}")
prefix_traffic_for_my_device_interface = client.GetNetFlowTrafficTotalsByDevice(start, end, device_id, interface=interface_name, prefix_breakdown=True)

asn = asn_traffic_for_my_device_interface[0]["interfaces"][0]["asns"][0]["asn"]
device_prefix = prefix_traffic_for_my_device_interface[0]["interfaces"][0]["prefixes"][0]["prefix"]

print(f"Traffic by Prefix for {device_name}/{interface_name} ASN {asn}")
prefix_traffic_for_my_device_interface_asn = client.GetNetFlowTrafficTotalsByDevice(
    start, end, device_id, interface=interface_name, asn=asn, asn_breakdown=True)

print(f"Traffic by Prefix")
prefix_traffic = client.GetNetFlowTrafficTotalsByPrefix(start, end)
prefix = prefix_traffic[0]["prefix"]

print(f"Traffic by Device for {prefix}")
device_traffic_for_prefix = client.GetNetFlowTrafficTotalsByPrefix(start, end, prefix)

print(f"Time series for {device_name}")
time_series_for_device = client.GetInterfaceCounterTrafficTimeSeries(start, end, device_id)

print(f"Time series for {device_name}/{interface_name}")
time_series_for_interface = client.GetInterfaceCounterTrafficTimeSeries(start, end, device_id, interface=interface_name)

print(f"Time series for {device_name}/{interface_name} {device_prefix}")
time_series_for_prefix = client.GetNetFlowTrafficTimeSeriesByDevice(start, end, device_id, interface=interface_name, prefix=device_prefix)

print(f"Time series for {device_name}/{interface_name} {asn}")
time_series_for_asn = client.GetNetFlowTrafficTimeSeriesByDevice(start, end, device_id, interface=interface_name, asn=asn)

Script Example: cctraffic.py

# cctraffic.py
# Contains a very simple REST client to demonstrate how to call the Crosswork Cloud Traffic APIs
# Copyright (c) 2021 Cisco Systems, Inc. and others.  All rights reserved.

import requests
from .util import UrlEncode

import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


class CCTrafficRestClient:

    def __init__(self, host: str, token, version: str = "v1", debug: bool = False):
        self.version = version
        self.host = host
        self.debug = debug
        self.headers = {"content-type": "application/json", "Authorization": f"Bearer {token}"}

    def DoApiCall(self, url):
        if self.debug == True:
            print(url)
        response = requests.get(url, headers=self.headers, verify=False)
        if self.debug == True:
            print(response.status_code)
            print(response.content)
        return response

    def GetDevice(self, device_name: str):
        url = f"{self.host}/api/{self.version}/devices?name={device_name}"
        response = self.DoApiCall(url)
        if response.status_code != 200:
            return ""
        return response.json()["devices"][0]["deviceInfo"]

    def GetInterfaceCounterTrafficTotals(self, start: str, end: str, device_id: str = ""):
        start = UrlEncode(start)
        end = UrlEncode(end)

        if device_id == "":
            url = f"{self.host}/api/{self.version}/devices/statistics/totals"
        else:
            url = f"{self.host}/api/{self.version}/devices/{device_id}/interfaces/statistics/totals"

        url += f"?format=totals&timeStart={start}&timeEnd={end}"
        response = self.DoApiCall(url)
        if response.status_code != 200:
            return ""
        return response.json()["devices"]

    def GetNetFlowTrafficTotalsByDevice(self, start: str, end: str, device_id: str, interface: str,
                                        asn: int = 0, prefix: str = "", asn_breakdown: bool = False, prefix_breakdown: bool = False):
        interface = UrlEncode(UrlEncode(interface))
        prefix = UrlEncode(UrlEncode(prefix))
        start = UrlEncode(start)
        end = UrlEncode(end)

        if asn != 0:
            url = f"{self.host}/api/{self.version}/traffic/devices/{device_id}/interfaces/{interface}/asns/{asn}/prefixes"
        elif asn_breakdown:
            url = f"{self.host}/api/{self.version}/traffic/devices/{device_id}/interfaces/{interface}/asns"
        elif prefix_breakdown:
            url = f"{self.host}/api/{self.version}/traffic/devices/{device_id}/interfaces/{interface}/prefixes"

        url += f"?format=totals&timeStart={start}&timeEnd={end}"
        response = self.DoApiCall(url)
        if response.status_code != 200:
            return ""
        return response.json()["devices"]

    def GetNetFlowTrafficTotalsByPrefix(self, start: str, end: str, prefix: str = "", device_id: str = ""):
        prefix = UrlEncode(UrlEncode(prefix))
        start = UrlEncode(start)
        end = UrlEncode(end)

        if prefix == "":
            url = f"{self.host}/api/{self.version}/traffic/prefixes"
        elif device_id == "":
            url = f"{self.host}/api/{self.version}/traffic/prefixes/{prefix}/devices"
        else:
            url = f"{self.host}/api/{self.version}/traffic/prefixes/{prefix}/devices/{device_id}/interfaces"

        url += f"?format=totals&timeStart={start}&timeEnd={end}"
        response = requests.get(url, headers=self.headers, verify=False)
        if response.status_code != 200:
            return ""
        return response.json()["prefixes"]

    def GetInterfaceCounterTrafficTimeSeries(self, start: str, end: str, device_id: str, interface: str = ""):
        interface = UrlEncode(UrlEncode(interface))
        start = UrlEncode(start)
        end = UrlEncode(end)

        if interface == "":
            url = f"{self.host}/api/{self.version}/devices/{device_id}/statistics/totals"
        else:
            url = f"{self.host}/api/{self.version}/devices/{device_id}/interfaces/{interface}/statistics/totals"

        url += f"?format=timeseries&timeStart={start}&timeEnd={end}"
        response = requests.get(url, headers=self.headers, verify=False)
        if response.status_code != 200:
            return ""
        return response.json()["devices"]

    def GetNetFlowTrafficTimeSeriesByDevice(self, start: str, end: str, device_id: str, interface: str, asn: int = 0, prefix: str = ""):
        interface = UrlEncode(UrlEncode(interface))
        prefix = UrlEncode(UrlEncode(prefix))
        start = UrlEncode(start)
        end = UrlEncode(end)

        if asn == 0 and prefix != "":
            url = f"{self.host}/api/{self.version}/traffic/devices/{device_id}/interfaces/{interface}/prefixes/{prefix}"
        elif asn != 0 and prefix == "":
            url = f"{self.host}/api/{self.version}/traffic/devices/{device_id}/interfaces/{interface}/asns/{asn}"
        else:
            url = f"{self.host}/api/{self.version}/traffic/devices/{device_id}/interfaces/{interface}/asns/{asn}/prefixes/{prefix}"

        url += f"?format=timeseries&timeStart={start}&timeEnd={end}"
        response = self.DoApiCall(url)
        if response.status_code != 200:
            return ""
        return response.json()["devices"]