Cisco Prime 401 unauthorized error from Cisco Prime API

AartiKailasam3215 · ‎06-22-2020

We have 5 Cisco Prime instances with 3.7 which we upgraded very recently from 3.4, after the upgrade we see 401 unauthorized exception from all the 5 PI instances, which we never encountered when we consumed them with Prime 3.4.

I am using below Python script to consume the API, as per Cisco's suggestion I have closed the connection in headers as well as with r.connection.close but it simply does not work

import requests, json, time, sys, traceback, math, pyodbc, re, socket
#only needed to supress warnings
import urllib3
urllib3.disable_warnings()
#max number of results allowed by Prime
MAX_PAGE_SIZE = 1000
TRUSTED_CONNECTION = ""
DRIVER = ""
TEST_MODE = False
"""
AP Inventory function returns a list of APs in the format defined by ap class.
Not all information is found with this query. Data to be merged when radio_details() is called.
"""
def ap_inventory():
mySecrets = get_secrets()
ap_list = []
server_list = mySecrets.get_servers().split(',')

for p_server in server_list:
#first get the @count to know how many to loop through
print('retrieving from',p_server)
r = requests.get('https://'+p_server+'/webacs/api/v3/data/AccessPointDetails.json', verify = False, auth = (mySecrets.get_user(), mySecrets.get_pass()))
if (not check_status_code(r.status_code)):
return
myjson = json.loads(r.text)
total_aps = int(myjson["queryResponse"]["@count"])
print(f'total aps: {total_aps}')

bottom_range = 0

#total number of queries is total items divided by 1000. Each iteration queries a page range.
for i in range(0,math.ceil(total_aps/MAX_PAGE_SIZE)):
print('Querying from ', bottom_range, " to ", bottom_range+MAX_PAGE_SIZE)
r = requests.get('https://'+p_server+'/webacs/api/v3/data/AccessPointDetails/.json?.full=true&.maxResults='+str(MAX_PAGE_SIZE)+'&.firstResult='+str(bottom_range)+'', verify = False, auth = (mySecrets.get_user(), mySecrets.get_pass()))
except requests.exceptions.ConnectionError as e:
r = "No response in connecting to server"
if (not check_status_code(r.status_code)):
return
myjson = json.loads(r.text)
ap_response = myjson["queryResponse"]["entity"]
for cur_ap in ap_response:
try:
#create AP using ID as init value
currentAP = ap(cur_ap["accessPointDetailsDTO"]["@id"])
except KeyError:
#add some logging?
print('no id')
continue
try:
currentAP.set_server(p_server)
except:
pass
try:
currentAP.set_name(cur_ap["accessPointDetailsDTO"]["name"])
except KeyError:
pass
try:
currentAP.set_ipaddress(cur_ap["accessPointDetailsDTO"]["ipAddress"])
except KeyError:
pass
try:
currentAP.set_base_mac(cur_ap["accessPointDetailsDTO"]["macAddress"])
except KeyError:
pass
try:
currentAP.set_ap_type(cur_ap["accessPointDetailsDTO"]["apType"])
except KeyError:
pass
try:
currentAP.set_ap_group_name(cur_ap["accessPointDetailsDTO"]["unifiedApInfo"]["apGroupName"])
except KeyError:
pass
try:
currentAP.set_current_controller_ip(cur_ap["accessPointDetailsDTO"]["unifiedApInfo"]["controllerIpAddress"])
except KeyError:
pass
try:
currentAP.set_current_controller(cur_ap["accessPointDetailsDTO"]["unifiedApInfo"]["controllerName"])
except KeyError:
pass
try:
currentAP.set_pri_controller(cur_ap["accessPointDetailsDTO"]["unifiedApInfo"]["primaryMwar"])
except KeyError:
pass
try:
currentAP.set_sec_controller(cur_ap["accessPointDetailsDTO"]["unifiedApInfo"]["secondaryMwar"])
except KeyError:
pass
try:
currentAP.set_ter_controller(cur_ap["accessPointDetailsDTO"]["unifiedApInfo"]["tertiaryMwar"])
except KeyError:
pass
try:
currentAP.set_ethernet_mac(cur_ap["accessPointDetailsDTO"]["ethernetMac"])
except KeyError:
pass
try:
currentAP.set_client_count(cur_ap["accessPointDetailsDTO"]["clientCount"])
except KeyError:
pass
try:
currentAP.set_client_count_2_4ghz(cur_ap["accessPointDetailsDTO"]["clientCount_2_4GHz"])
except KeyError:
pass
try:
currentAP.set_client_count_5ghz(cur_ap["accessPointDetailsDTO"]["clientCount_5GHz"])
except KeyError:
pass
try:
currentAP.set_capwap_uptime(cur_ap["accessPointDetailsDTO"]["unifiedApInfo"]["capwapUpTime"])
except KeyError:
pass
try:
currentAP.set_serial(cur_ap["accessPointDetailsDTO"]["serialNumber"])
except KeyError:
pass
try:
currentAP.set_software_version(cur_ap["accessPointDetailsDTO"]["softwareVersion"])
except KeyError:
pass
try:
currentAP.set_location_hierarchy(cur_ap["accessPointDetailsDTO"]["locationHierarchy"])
except KeyError:
pass
try:
currentAP.set_reachability_status(cur_ap["accessPointDetailsDTO"]["reachabilityStatus"])
except KeyError:
pass
"""
CDP neigh object looks like this:
"cdpNeighbors": {
"cdpNeighbor": [
{
"capabilities": "Switch IGMP ",
"duplex": "Full Duplex",
"interfaceSpeed": "1Gbps",
"localPort": "2",
"neighborIpAddress": "x.x.x.x",
"neighborName": "xxxxx",
"neighborPort": "GigabitEthernet1/0/2",
"platform": "xxxxx"
}
]
},
"""
try:
cdpNeigh = cur_ap["accessPointDetailsDTO"]["cdpNeighbors"]["cdpNeighbor"][0]
currentAP.set_cdpneigh_port_speed(cdpNeigh["interfaceSpeed"])
currentAP.set_cdpneigh_name(cdpNeigh["neighborName"])
currentAP.set_cdpneigh_ip(cdpNeigh["neighborIpAddress"])
currentAP.set_cdpneigh_port_name(cdpNeigh["neighborPort"])
currentAP.set_cdpneigh_duplex(cdpNeigh["duplex"])
currentAP.set_cdpneigh_platform(cdpNeigh["platform"])
except KeyError:
pass
"""
wlanProfiles (for ssid list) looks like this:
"wlanProfiles": {
"wlanProfile": [
{
"broadcastSsidEnabled": true,
"profileName": "XX",
"ssid": "XX"
},
{
"broadcastSsidEnabled": true,
"profileName": "XX",
"ssid": "XX"
},
#more profiles
]
}
"""
try:
wlanProfiles = cur_ap["accessPointDetailsDTO"]["unifiedApInfo"]["wlanProfiles"]["wlanProfile"]
#create comma seperated string of ssid
ssid = ""
for i in range(0,len(wlanProfiles)):
ssid += str(wlanProfiles[i]["ssid"])
if(i < len(wlanProfiles)-1):
ssid += ","
currentAP.set_ssid_list(ssid)
except KeyError:
pass
#add ap object to ap list
ap_list.append(currentAP)
#increment page
bottom_range += MAX_PAGE_SIZE
r.connection.close() //connection close as per cisco suggestion
return ap_list

'''
Defines a access point with data from the following source:
#https://{{hostname}}/webacs/api/v3/data/AccessPointDetails/.json?.full=true

'''
class ap(object):
def __init__(self, id):
self.id = id
#init all values to Null for exception handling
self.name = None
self.ip_address = None
self.base_mac = None
self.ap_type = None
self.ap_group_name = None
self.current_controller_name = None
self.current_controller_ip = None
self.pri_controller = None
self.sec_controller = None
self.ter_controller = None
self.ethernet_mac = None
self.client_count = None
self.client_count_2_4ghz = None
self.client_count_5ghz = None
self.cdpneigh_name = None
self.cdpneigh_ip = None
self.cdpneigh_port_name = None
self.cdpneigh_port_speed = None
self.cdpneigh_duplex = None
self.cdpneigh_platform = None
self.radio_base_mac = None
self.channel_number = None
self.channel_width = None
self.radio_type = None
self.tx_power_output = None
self.cdpneigh_platform = None
self.antenna_name = None
self.ethernet_mac = None
self.capwap_uptime = None
self.software_version = None
self.serial = None
self.ssid_list = None
self.location_hierarchy = None
self.reachability_status = None
self.server = None
def get_id(self):
return self.id
def set_name(self, name):
self.name = name
def get_name(self):
return self.name
def set_ipaddress(self, ip_address):
self.ip_address = ip_address
def get_ipaddress(self):
return self.ip_address
def set_base_mac(self,mac):
self.base_mac = mac
def get_base_mac(self):
return self.base_mac
def set_ap_type(self,aptype):
self.ap_type = aptype
def get_ap_type(self):
return self.ap_type
def set_ap_group_name(self, ap_group_name):
self.ap_group_name = ap_group_name
def get_ap_group_name(self):
return self.ap_group_name
def set_current_controller(self,controller):
self.current_controller_name = controller
def get_current_controller(self):
return self.current_controller_name
def set_current_controller_ip(self,controller_ip):
self.current_controller_ip = controller_ip
def get_current_controller_ip(self):
return self.current_controller_ip
def set_pri_controller(self, pri_controller):
self.pri_controller = pri_controller
def get_pri_controller(self):
return self.pri_controller
def set_sec_controller(self, sec_controller):
self.sec_controller = sec_controller
def get_sec_controller(self):
return self.sec_controller
def set_ter_controller(self,ter_controller):
self.ter_controller = ter_controller
def get_ter_controller(self):
return self.ter_controller
def set_ethernet_mac(self, ethernet_mac):
self.ethernet_mac = ethernet_mac
def get_ethernet_mac(self):
return self.ethernet_mac
def set_client_count(self,client_count):
self.client_count = client_count
def get_client_count(self):
return self.client_count
def set_client_count_2_4ghz(self, client_count_2_4ghz):
self.client_count_2_4ghz = client_count_2_4ghz
def get_client_count_2_4ghz(self):
return self.client_count_2_4ghz
def set_client_count_5ghz(self,client_count_5ghz):
self.client_count_5ghz = client_count_5ghz
def get_client_count_5ghz(self):
return self.client_count_5ghz
def set_cdpneigh_name(self, name):
self.cdpneigh_name = name
def get_cdpneigh_name(self):
return self.cdpneigh_name
def set_cdpneigh_ip(self, ipaddress):
self.cdpneigh_ip = ipaddress
def get_cdpneigh_ip(self):
return self.cdpneigh_ip
def set_cdpneigh_port_name(self, port_name):
self.cdpneigh_port_name = port_name
def get_cdpneigh_port_name(self):
return self.cdpneigh_port_name
def set_cdpneigh_port_speed(self, speed):
self.cdpneigh_port_speed = speed
def get_cdpneigh_port_speed(self):
return self.cdpneigh_port_speed
def set_cdpneigh_duplex(self, duplex):
self.cdpneigh_duplex = duplex
def get_cdpneigh_duplex(self):
return self.cdpneigh_duplex
def set_cdpneigh_platform(self, platform):
self.cdpneigh_platform = platform
def get_cdpneigh_patform(self):
return self.cdpneigh_platform
def set_capwap_uptime(self,uptime):
self.capwap_uptime = uptime
def get_capwap_uptime(self):
return self.capwap_uptime
def set_software_version(self,version):
self.software_version = version
def get_software_version(self):
return self.software_version
def set_serial(self,serial):
self.serial = serial
def get_serial(self):
return self.serial
def set_ssid_list(self,ssid_list):
self.ssid_list = ssid_list
def get_ssid_list(self):
return self.ssid_list
def set_location_hierarchy(self,location):
self.location_hierarchy = location
def get_location_heirarchy(self):
return self.location_hierarchy
def set_reachability_status(self,status):
self.reachability_status = status
def get_reachability_status(self):
return self.reachability_status
def set_server(self,server):
self.server = server
def get_server(self):
return self.server

#Shows all variables for debugging purposes
def ap_display_object(self):
return {
"id " : self.id,
"name " : self.name,
"mac " : self.base_mac,
"ipaddress " : self.ip_address,
"ap_type " : self.ap_type,
"ap_group_name " : self.ap_group_name,
"current_controller" : self.current_controller_name,
"current_controller_ip " : self.current_controller_ip,
"pri_controller " : self.pri_controller,
"sec_controller " : self.sec_controller,
"ter_controller " : self.ter_controller,
"ethernet_mac " : self.ethernet_mac,
"client_count " : self.client_count,
"client_count_2_4ghz " : self.client_count_2_4ghz,
"client_count_5ghz " : self.client_count_5ghz,
"cdpneigh_name " : self.cdpneigh_name,
"cdpneigh_ip " : self.cdpneigh_ip,
"cdpneigh_port_name " : self.cdpneigh_port_name,
"cdpneigh_port_speed " : self.cdpneigh_port_speed,
"cdpneigh_duplex " : self.cdpneigh_duplex,
"cdpneigh_platform " : self.cdpneigh_platform,
"ethernet_mac" : self.ethernet_mac,
"capwap_uptime " : self.capwap_uptime,
"software_version " : self.software_version,
"serial " : self.serial,
"ssid list " : self.ssid_list,
"location hierarchy " : self.location_hierarchy,
"reachability status " : self.reachability_status
}

class Secrets(object):
def __init__(self,api_user,api_pass,prime_servers,db_name,db_username,db_pass,db_server):
self.username = api_user
self.password = api_pass
self.servers_csv = prime_servers
self.db_name = db_name
self.db_username = db_username
self.db_pass = db_pass
self.db_server = db_server
def get_user(self):
return self.username
def get_pass(self):
return self.password
def get_servers(self):
return self.servers_csv
def get_db_server(self):
return self.db_server
def get_db_name(self):
return self.db_name
def get_db_username(self):
return self.db_username
def get_db_pass(self):
return self.db_pass

## Helper functions ##

def get_secrets():
global TRUSTED_CONNECTION, DRIVER
if(TEST_MODE):
from secrets import LocalSecrets
DRIVER='{SQL Server}'
return LocalSecrets()
else:
TRUSTED_CONNECTION = "Trusted_connection={yes};"
my_secrets = Secrets(sys.argv[1],sys.argv[2],sys.argv[3],sys.argv[4],sys.argv[5],sys.argv[6],sys.argv[7])
DRIVER='{SQL Server Native Client 11.0}'
return my_secrets

def check_status_code(code):
if(code == 200):
print('success')
return True
else:
print('no such luck')
return False

def insert_database(ap_list):
global TRUSTED_CONNECTION,DRIVER
mySecrets = get_secrets()
try:
conn = pyodbc.connect('DRIVER='+DRIVER+';SERVER={'+mySecrets.get_db_server()+'};DATABASE={'+mySecrets.get_db_name()+'};'+TRUSTED_CONNECTION+'uid={'+ mySecrets.get_db_username()+'};pwd={'+mySecrets.get_db_pass()+'}')
cursor = conn.cursor()
if TEST_MODE:
print('truncating table')
#otherwise handled in SSIS package
cursor.execute("truncate table dbo.aptable")
cursor.commit()
print('inserting db\n\n')
for ap in ap_list:
cursor.execute("Insert Into dbo.aptable (manager,ap_id,icmp_address,name,specific_model,ap_group_name,ap_sysuptime,enet_duplex,enet_in,cdp_neigh_name,cdp_neigh_port,lan_mac,current_controller,current_controller_ip,primary_controller,secondary_controller,tertiary_controller,version,device_serial,ssid_list,client_count,client_count_2_4ghz,client_count_5ghz,ap_folder_path,reachability_status) values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", \
ap.get_server(),ap.get_id(),ap.get_ipaddress(),ap.get_name(),ap.get_ap_type(),ap.get_ap_group_name(),ap.get_capwap_uptime(),ap.get_cdpneigh_duplex(),ap.get_cdpneigh_port_speed(),ap.get_cdpneigh_name(),ap.get_cdpneigh_port_name(),ap.get_ethernet_mac(),ap.get_current_controller(),ap.get_current_controller_ip(),ap.get_pri_controller(),ap.get_sec_controller(),ap.get_ter_controller(),ap.get_software_version(),ap.get_serial(),ap.get_ssid_list(),ap.get_client_count(),ap.get_client_count_2_4ghz(),ap.get_client_count_5ghz(),ap.get_location_heirarchy(),ap.get_reachability_status())
cursor.commit()
cursor.close()
conn.close()
except:
print('could not connect')
print(traceback.format_exc())

if __name__ == "__main__":
ap_list = ap_inventory()
insert_database(ap_list)

when I tried checking the 401 xml, I see below error possibilities

JR-AG-2024 · ‎09-30-2024

For anyone that want to help.
Create a new user with the permission and getting the same output.
Not able to get a simple get

response = requests.get('https://x.x.x.x/webacs/api/v4/data/Devices?.full=true&.sort=deviceName', auth=('USERLOCAL', 'xxxxxxxxxxxxxx'), verify=False)
output = response.text
print(output)