Thank you! I fixed up the errors and made a few more tweaks. I'm sure it's nowhere near as good as a professional developer would write but everything works. I have made it available here - OnePK Python scripting for Network Engineers - An example Kind regards, Ryan Chaney ... View more

Hi, Ok, I've gone away and spent some time rewriting my script. What I'm trying to achieve is this: 1. Provide IPs in the format of the hosts.txt file 2. I want to be able to provide a '-f  hosts.txt' argument to my script from the command line 3. I want the script to check whether -a or -f has been used 4. if -f was used iterate through true_hosts in the OpenAndCloseVTY class 5. Otherwise if -a was used, run OpenAndCloseVTY against the single IP address The code I want to use for parsing the hosts.txt is: hosts_file = open('hosts.txt') lines = hosts_file.readlines() for line in lines:         line_check = str(line)         if line_check.startswith('#') is True:             pass         elif line_check.startswith(' ') is True:             pass         else:             line_check = line.strip('\n')             line_check = line.strip(' ')             line_check = line.split(".")             if 1 <= int(line_check[0]) <= 223 and 0 <= int(line_check[1]) <= 255 and 0 <= int(line_check[2]) <= 255 and 1 <= int(line_check[3]) <= 254:                 line = line.strip('\n')                 line = line.strip(' ')                 true_hosts = []                 true_hosts.append(line)                 print true_hosts             else:                 print "Error: Invalid IP address" Can you please give me some points as what I should put where in dev.py? Kind regards, Ryan Chaney ... View more

Hi, Thank you, this works. I made one change as I wanted to capture the received bytes not packets: From - print intf_stats.get_param(InterfaceStatistics.InterfaceStatisticsParameter.ONEP_IF_STAT_RX_PKTS To - print intf_stats.get_param(InterfaceStatistics.InterfaceStatisticsParameter.ONEP_IF_STAT_RX_ BYTES I agree this script is *very* ugly. Once I get it working how I want then I will tidy it up significantly. Thanks again, Ryan ... View more

That makes sense. I moved this down and when I run the script I now get: INFO:onep:VTYTutorial:ParserState prompt - chaney-xtr# INFO:onep:VTYTutorial:ParserState overallrc - 0 INFO:onep:VTYTutorial:ParserState::cmdresult:inputline - show int atm0 | in packets input INFO:onep:VTYTutorial:ParserState::cmdresult:parsereturncode - 0 INFO:onep:VTYTutorial:ParserState::cmdresult:errorlocation - 14 For me this is of no use so I have just commented it out, but at least we fixed the error and I understand what the config was intended to do. Thanks again. ... View more

Hi, I'd like to keep the file to a plain text format as that is more familiar to network engineers. The following config gives me a list of IPs from the host file template I posted above: import re expression = re.compile('^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$') hosts_file = open('hosts.txt') lines = hosts_file.readlines() for line in lines:     if expression.match(line):         line = line.strip('\n')         true_hosts = []         true_hosts.append(line) What I am trying to get my head around now is how I can add an option for -f to the attached script so that if I specify -f instead of -a I can iterate through the list of IPs and run the script against each host. Any ideas? ... View more

Ok that has helped enormously, thank you. I'm now getting this message instead: ERROR:onep:VTYTutorial:Error in getting parser state ERROR:onep:VTYTutorial:No results in queue, ONEP_FAIL I'm trying to work out if this is bad or not or what is causing this, any ideas? I could just comment out this section but I haven't enough detail on what it does yet. Ryan ... View more

I'm getting this error in my script: ERROR:onep:VTYTutorial:name 'ne' is not defined I suspected it was to do with where I put the from statements but I've tried a few places and it doesn't seem to matter. ------------------------------ #!/usr/bin/env python """ Connect to a network element to obtain a session handle. """ from onep.element.NetworkApplication import NetworkApplication from onep.interfaces import InterfaceStatus from onep.element import SessionConfig from onep.interfaces import NetworkInterface import getopt import sys import getpass import onep.interfaces.InterfaceFilter import os import socket from onep.core.util import tlspinning from onep.vty import VtyService from onep.core.exception import OnepRemoteProcedureException, OnepException #  START SNIPPET: createLogger import logging logging.basicConfig(level=logging.WARNING) logger = logging.getLogger('onep:BaseTutorial') logger.setLevel(logging.INFO) #  END SNIPPET: createLogger class BaseTutorial(object):     #  START SNIPPET: variables     element_hostname = None     username = None     password = None     transport = "tls"     network_element = None     session_handle = None     root_cert_path = None     client_cert_path = None     client_key_path = None     key_passphrase = None     tls_pinning_file = None         args = None     #  END SNIPPET: variables     def __init__(self, args):         """         Instantiates the Logger instance using the classname of the implementing class.         """         self.args = args     def connect(self, applicationName):         """         Obtains a NetworkApplication instance, sets the name to applicationName,         gets a network element for the hostname or address in the command line         arguments or tutorial.properties file and then tries to connect to the         Network Element with the username and password supplied, or from the         tutorial.properties file.                 @param applicationName: The NetworkApplication name is set to this value.         @return True if the connection succeeded without exception, else false.         @throws OnepException         """                 #  START SNIPPET: init_myapp         network_application = NetworkApplication.get_instance()         #  END SNIPPET: init_myapp                 #  START SNIPPET: name_myapp           network_application.name = applicationName         #  END SNIPPET: name_myapp                 #  START SNIPPET: get_ne_opt1         self.network_element = network_application.get_network_element(self.element_hostname)         #  END SNIPPET: get_ne_opt1         if self.network_element == None:             logger.error("Failed to get network element")             sys.exit(1) #        logger.info("We have a NetworkElement : " + self.network_element.__str__())                 #  START SNIPPET: connect         session_config = SessionConfig(SessionConfig.SessionTransportMode.TLS) #default is TLS         if self.transport.lower() == "tipc" or self.transport == "2":             session_config = SessionConfig(SessionConfig.SessionTransportMode.TIPC)         #Set all the TLS properties in session_config         session_config.ca_certs = self.root_cert_path         session_config.keyfile = self.client_key_path         session_config.certfile = self.client_cert_path         session_config.set_tls_pinning(self.tls_pinning_file, PinningHandler(self.tls_pinning_file))         self.session_handle = self.network_element.connect(self.username, self.password, session_config)         #  END SNIPPET: connect                 if self.session_handle == None:             #  START SNIPPET: loggerError             logger.error("Failed to connect to NetworkElement - ")             #  END SNIPPET: loggerError             return False #        logger.info("Successful connection to NetworkElement - " )         return True     def disconnect(self):         """         Disconnects the application from the Network Element.                 @return True if the disconnect succeeded without an exception, else false if the application failed to disconnect         from the Network Element.         """         try:             if self.network_element.is_connected():                 self.network_element.disconnect()         except Exception, e:             logger.error("Failed to disconnect from Network Element")             logger.error(e)             return False         return True     def get_element_hostname(self):         """         Gets the IP address or hostname of the Network Element we are interested in.                 @return The element_hostname.                """         return self.element_hostname     #  START SNIPPET: getLogger     def get_logger(self):         """         Implements the logger, which sends all enabled log messages.                 @return The logger.         """         return self.logger     #  END SNIPPET: getLogger         def get_network_element(self):         """         Gets the NetworkElement that is being connected to by the application.                 @return The NetworkElement that is being connected to by the application.         """         return self.network_element     def get_session_handle(self):         """         Gets the session_handle that identifies the connection to this Network Element.                 @return The session_handle that identifies the connection to this NetworkElement.         """         return self.session_handle     def get_username(self):         """         Gets the username on whose behalf the connection will be made. The username is specified via the command line or         in the tutorial.properties file.                 @return The username that is specified via the command line or in the tutorial.properties file.         """         return self.username         def get_password(self):         """         Gets the password with which connection will be made. The password is specified via the command line or         in the tutorial.properties file.                 @return The password that is specified via the command line or in the tutorial.properties file.         """                  return self.password         def get_transport(self):         """         Gets the transport mode with which connection will be made. The transport is specified via the command line or         in the tutorial.properties file.                 @return The transport that is specified via the command line or in the tutorial.properties file.         """         return self.transport         def get_usage(self):         return " Usage: -a <address or FQDN> [-t <transport>] [-C <client cert file>] \n [-K <client private key file>] \n [-R <root certificates file>] \n [-P <tls pinning file>]"     def parse_command_line(self, skip_login_for_tipc=True):         """         Parse the command line options.         @param args  The args string passed into the main(...) method.                 @return true if parsing the command line succeeds, false otherwise.         """         try:             opts, args = getopt.getopt(self.args[1:],"ha:t:R:C:K:P:",["hostname=","transport=", "rootcert=", "clientcert=", "key=", "tlspinningfile="])         except getopt.GetoptError as err:             print str(err)             logger.info(self.get_usage())             sys.exit(2)                 """          * options:          *       -a, --hostname <network element address or FQDN>          *       -t, --transport <transport type> default is tls          *       -C, --clientcert <client certificate file>          *       -K, --clientkey <client private key file>          *       -R, --rootcert <root certificates file>          *       -P, --pinfile <tls pinning file>         """             for option, arg in opts:             if option == '-h':                 logger.info(self.get_usage())                 sys.exit()             elif option in ("-a", "--hostname"):                 self.element_hostname = arg             elif option in ("-t", "--transport"):                 self.transport = arg             elif option in ("-R", "--rootcert"):                 self.root_cert_path = arg             elif option in ("-C", "--clientcert"):                 self.client_cert_path = arg             elif option in ("-K", "--key"):                 self.client_key_path = arg             elif option in ("-P", "--pinfile"):                 self.tls_pinning_file = arg                 if self.element_hostname == None:             logger.error(self.get_usage())             return False         if self.transport != "tipc" or not skip_login_for_tipc:             self.username = raw_input('Enter Username : ')             self.password = getpass.getpass('Enter Password : ')             if self.username == None or self.password == None:                 logger.error("Username and password are required.")                 return False                 return True     def set_element_hostname(self, element_hostname):         """         """         self.element_hostname = element_hostname     def set_logger(self, logger):         """         """         self.logger = logger     def set_network_element(self, network_element):         """         """                self.network_element = network_element     def set_password(self, password):         """         """                self.password = password     def set_session_handle(self, session_handle):         """         """                self.session_handle = session_handle     def set_username(self, username):         """         """                self.username = username     def get_all_interfaces(self):         """         """                interfaceList = None         try:             interfaceList = self.network_element.get_interface_list(onep.interfaces.InterfaceFilter())         except Exception, e:             logger.error(e)         return interfaceList     def get_an_interface(self):         """           Get an interface which has an IP addr set and which is not the interface that the application is connected to         """         interfaceList = None         networkInterface = None         skipFirst = False         try:             interfaceList = self.network_element.get_interface_list(onep.interfaces.InterfaceFilter())             elemAddress = socket.gethostbyname(self.network_element.host_address)             for networkInterface in interfaceList:   if networkInterface.interface_type != NetworkInterface.InterfaceTypes.ONEP_IF_TYPE_TUNNEL:                 addresses = networkInterface.get_address_list()                 if addresses and elemAddress not in addresses:                     if skipFirst:                         return networkInterface                     else:                         skipFirst = True         except Exception, e:             logger.error(e)         return None     def get_an_up_interface(self):         """           Get an interface which has its LINK status as UP         """         interfaceList = None         networkInterface = None         try:             interfaceList = self.network_element.get_interface_list(onep.interfaces.InterfaceFilter())             for networkInterface in interfaceList:                 if networkInterface.get_status().link == InterfaceStatus.InterfaceState.ONEP_IF_STATE_OPER_UP:                     return networkInterface                                  except Exception, e:             logger.error(e)         return None     class PinningHandler(tlspinning.TLSUnverifiedElementHandler):         def __init__(self, pinning_file):         self.pinning_file = pinning_file             def handle_verify(self, host,hashtype, finger_print, changed):         """         Callback to the app to determine whether to add a host to pinning DB         Upon receipt of a certificate which fails to match based on server-name or         IP address, and for which there is no match in the pinning database,         this callback asks the application whether to accept the         connection and/or whether to add the server to the pinning database.         By default, the connection will be terminated and the pinning db will         remain unchanged.         """         if  changed:             msg = "\n@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@\n"             msg +="WARNING: THE CERTIFICATE PRESENTED BY REMOTE HOST '%s'\n IS DIFFERENT FROM THE ONE PREVIOUSLY ACCEPTED" %(host)             msg +="\n@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@"         else:             msg = "WARNING: Certificate presented by remote host '%s' is not verified."%(host)         msg += "\n\nThe %s fingerprint sent by the remote host(%s) is:\n%s" %(hashtype, host, finger_print)         msg += "\n\nYou MUST verify the certificate on remote host before proceeding! \n"         msg += "\nChoose from following options:"         if self.pinning_file:             prompt = "\nAccept and Pin (p), Accept Once (o), Reject (r) (default) :"         else:             prompt = "\nAccept Once (o), Reject (r) (default) :"         sys.stdout.write(msg)         self.decision = raw_input(prompt)                 while True:             if not self.decision or self.decision.lower() == 'r':                 return tlspinning.DecisionType.REJECT             elif self.decision.lower() == 'p' and self.pinning_file:                 return tlspinning.DecisionType.ACCEPT_AND_PIN             elif self.decision.lower() == 'o':                 return tlspinning.DecisionType.ACCEPT_ONCE             else:                 self.decision = raw_input(prompt)             # if __name__ == '__main__': #     tutorial = BaseTutorial(sys.argv) #     # #    logger.info("Reading arguments...") #     if not tutorial.parse_command_line(): #         logger.error("Error in parsing arguments") #         sys.exit(1) #     # #    logger.info("Connecting to Network Element...") #     #try: #     tutorial.connect("BaseTutorial") #     #except Exception, e: #         #logger.error("Error in connecting to network element - %s", e) #     #logger.error(e) #         #     tutorial.disconnect() # #    logger.info("Done.") # # VTYTutorial starts here # logger = logging.getLogger('onep:VTYTutorial') logger.setLevel(logging.INFO) class VTYTutorial(BaseTutorial):     """     This tutorial shows how to create VTY Service on a ONEP Application     and demonstrates the capability to communicate with a Network Element via virtual terminal. .     @author The onePK Team ( onepk-feedback@cisco.com )     """     def show_parser_state_attributes(self, parser_state):         """         This method prints the attributes of the ParserState and iterates over the command results                 @param parser_state         """         #  START SNIPPET: vty_parser_state_attribs         logger.info("ParserState prompt - %s", parser_state.prompt)         logger.info("ParserState overallrc - %s", parser_state.overallRC)         cmd_results = parser_state.results         for cmd_result in cmd_results:             logger.info("ParserState::cmdresult:inputline - %s", cmd_result.input_line)             logger.info("ParserState::cmdresult:parsereturncode - %s", cmd_result.parse_return_code)             logger.info("ParserState::cmdresult:errorlocation - %s", cmd_result.error_location)         #  END SNIPPET: vty_parser_state_attribs if __name__ == '__main__':     """     Invokes the tutorial via the command line. This main method attempts to use arguments from the command line.     """     import sys     tutorial = VTYTutorial(sys.argv)     #    logger.info("Reading arguments...")     if not tutorial.parse_command_line():         logger.error("Error in parsing arguments")         sys.exit(1)     try: #        logger.info("Connecting to Network Element...")         if not tutorial.connect("VTYTutorial"):             logger.error("Error in connecting to network element")             sys.exit(1) #        logger.info("Done")                 """Get a VTYService from the NE"""         #  START SNIPPET: create_vty         vtyService = VtyService(tutorial.get_network_element())         #  END SNIPPET: create_vty                 """Open the VTY on the NE with the default command interpreter"""         #  START SNIPPET: open_vty         vtyService.open()         #  END SNIPPET: open_vty         """Get a timeout of the VTY on NE"""         #  START SNIPPET: get_vty_timeout #        logger.info("VTY Time Out - %s", vtyService.timeout)         #  END SNIPPET: get_vty_timeout                """Get the parser state and its attributes"""         try:             #  START SNIPPET: get_vty_parser_state             parser_state = vtyService.get_parser_state()             #  END SNIPPET: get_vty_parser_state                         """Show the parser state attributes"""             tutorial.show_parser_state_attributes(parser_state)         except OnepRemoteProcedureException as re:             logger.error("Error in getting parser state : ", re)                 """Get the VTY State"""         #  START SNIPPET: get_vty_state            #        logger.info("State - %s", vtyService.state)         #  END SNIPPET: get_vty_state                 """Set the max response length"""         #  START SNIPPET: set_vty_max_response                    MAX_RESPONSE_LENGTH = 110         vtyService.max_response = MAX_RESPONSE_LENGTH #        logger.info("MaxResponse - %s", vtyService.max_response)         #  END SNIPPET: set_vty_max_response         # ----------------------------------------- # Run CLI commands and format their output # -----------------------------------------                 from onep.element import NetworkElement,SessionConfig         from onep.interfaces import InterfaceStatistics         print ne.properties.sys_name         intf_stats = ne.get_interface_by_name("ATM0").get_statistics()         print intf_stats.get_param(InterfaceStatistics.InterfaceStatisticsParameter.ONEP_IF_STAT_RX_PKTS)                 """ Get hostname """         GET_HOSTNAME = "show run | in hostname "         cli_result = vtyService.write(GET_HOSTNAME)         HOSTNAME = str(cli_result)         """ Format HOSTNAME variable for banner """         HOSTNAME = HOSTNAME.split(' ')[1]         HOSTNAME = HOSTNAME.strip() #        print HOSTNAME         """ Get number of full inits """         GET_DROPS = "show control vd 0 | in Full"         cli_result = vtyService.write(GET_DROPS)         DROPS = str(cli_result)         """ Format DROPS variable for banner """          DROPS = DROPS.strip('Full inits:')         DROPS = DROPS.strip()         DROPS = int(float(DROPS))         if DROPS == 1:             DROPS = 0 #        print DROPS       #        logger.info("Test Command : %s", DROPS) #        logger.info("CLI Result for Test Command : %s", cli_result)         """ Get DSL noise margin """         GET_NOISE_MARGIN = "show control vd 0 | in Noise"         cli_result = vtyService.write(GET_NOISE_MARGIN)         NOISE_MARGIN = str(cli_result)  #        print NOISE_MARGIN                 """ Format NOISE_MARGIN variable for banner """          NOISE_MARGIN = NOISE_MARGIN.strip('Noise Margin:')         NOISE_MARGIN = NOISE_MARGIN.strip()         NOISE_MARGIN = NOISE_MARGIN.split(' ')[0]     #        print NOISE_MARGIN #        logger.info("Test Command : %s", NOISE_MARGIN) #        logger.info("CLI Result for Test Command : %s", cli_result)              """ Get DSL sync speed (down) """         GET_SPEED_DOWN = "show control vd 0 | in Speed | exl Previous"              cli_result = vtyService.write(GET_SPEED_DOWN)         SPEED_DOWN = str(cli_result) #        print SPEED_DOWN         """ Format SPEED_DOWN variable for banner """                MBPS_DOWN = 0         KBPS_DOWN = 0         SPEED_DOWN = SPEED_DOWN.strip ('Speed (kbps):          0')         SPEED_DOWN = SPEED_DOWN.strip()         SPEED_DOWN = SPEED_DOWN.split(' ')[0]         SPEED_DOWN = SPEED_DOWN.strip() #        print SPEED_DOWN  #        print len(SPEED_DOWN)             if len(SPEED_DOWN) < 4:             KBPS_DOWN = SPEED_DOWN         elif len(SPEED_DOWN) == 4:             MBPS_DOWN = SPEED_DOWN[:1]             KBPS_DOWN = SPEED_DOWN[-3:-2]         elif len(SPEED_DOWN) == 5:             MBPS_DOWN = SPEED_DOWN[:2]             KBPS_DOWN = SPEED_DOWN[-3:-2] #        print '{}.{}'.format(MBPS_DOWN, KBPS_DOWN) #        logger.info("Test Command : %s", SPEED_DOWN) #        logger.info("CLI Result for Test Command : %s", cli_result)         """ Get DSL sync speed (up) """         GET_SPEED_UP = "show control vd 0 | in Speed | exl Previous"         cli_result = vtyService.write(GET_SPEED_UP)         """ Format SPEED_UP variable for banner """         SPEED_UP = str(cli_result)         MBPS_UP = 0         KBPS_UP = 0         SPEED_UP = SPEED_UP[-7:]         SPEED_UP = SPEED_UP.strip()         if len(SPEED_UP) < 4:             KBPS_UP = SPEED_UP         elif len(SPEED_UP) == 4:             MBPS_UP = SPEED_UP[:1]             KBPS_UP = SPEED_UP[-3:-2]         elif len(SPEED_UP) == 5:             MBPS_UP = SPEED_UP[:2]             KBPS_UP = SPEED_UP[-3:-2] #        print '{}.{}'.format(MBPS_UP, KBPS_UP) #        logger.info("Test Command : %s", SPEED_UP) #        logger.info("CLI Result for Test Command : %s", cli_result)         """ Get total amount of data downloaded """         GET_DOWNLOADS = "show int atm0 | in packets input"         cli_result = vtyService.write(GET_DOWNLOADS)         DOWNLOADS = str(cli_result)         """ Format DOWNLOADS variable for banner """         MB = 0         GB = 0         DOWNLOADS = DOWNLOADS.strip()         DOWNLOADS = DOWNLOADS.split(' ')[3]         if DOWNLOADS < 1000000000:             MB = DOWNLOADS[-9:-8]         if DOWNLOADS >= 1000000000:             GB = DOWNLOADS[:-9]             MB = DOWNLOADS[-9:-8] #        print '{}.{}'.format(GB, MB)                          """ Print banner """         print('\nHostname is {}\nCurrent download speed is {}.{} Mbps, current upload speed is {}.{} Mbps \nNoise margin is {} dB, {} WAN interface resets since counters last cleared\n{}.{} GB downloaded this month \n '.format(HOSTNAME, MBPS_DOWN, KBPS_DOWN, MBPS_UP, KBPS_UP, NOISE_MARGIN, DROPS, GB, MB)) # ----------------------------------------- # End of CLI commands # -----------------------------------------                 """Cancel the command execution"""         vtyService.cancel()                           """Close the VTY connection on NE"""                   vtyService.close()                """Check if the VTY is still open""" #        logger.info("Is Open - %s", vtyService.is_open())         """Destroy the VTY"""                 vtyService.destroy()     except Exception, e:         tutorial.disconnect()         logger.error(str(e))     tutorial.disconnect()     sys.exit(0) ---------------------------------------------------- ... View more

Thanks Karen, I thought this would be the case. In relation to your config, I am already connected to the device earlier in the script (not shown), this is how I can run the VTY commands. I assume therefore that I only need: from onep.element import NetworkElement,SessionConfig from onep.interfaces import InterfaceStatistics print ne.properties.sys_name #Print input packets on ATM0 intf_stats = ne.get_interface_by_name("ATM0").get_statistics() print intf_stats.get_param(InterfaceStatistics.InterfaceStatisticsParameter.ONEP_IF_STAT_RX_PKTS) Is this correct? ... View more

Hi Joseph, Here is my current script. Kind regards, Ryan Chaney ... View more

Hey, You make a very good point, I hadn't considered how commands might change across platforms. As I have a network engineering background my approach was to start with VTY commands, get the hang of the scripting, and then changeover to the Cisco API where possible. I have another post relating to this (Which VTY commands can I replace with the native Python API?), would you mind taking a look? Kind regards, Ryan ... View more