trigger.netdevices — Network device metadata library

The heart and soul of Trigger, NetDevices is an abstract interface to network device metadata and ACL associations.

Parses NETDEVICES_SOURCE and makes available a dictionary of NetDevice objects, which is keyed by the FQDN of every network device.

Other interfaces are non-public.

Example:

>>> from trigger.netdevices import NetDevices
>>> nd = NetDevices()
>>> dev = nd['test1-abc.net.aol.com']
>>> dev.vendor, dev.make
(<Vendor: Juniper>, 'MX960-BASE-AC')
>>> dev.bounce.next_ok('green')
datetime.datetime(2010, 4, 9, 9, 0, tzinfo=<UTC>)
trigger.netdevices.device_match(name, production_only=True)

Return a matching NetDevice object based on partial name. Return None if no match or if multiple matches is cancelled:

>>> device_match('test')
2 possible matches found for 'test':
  [ 1] test1-abc.net.aol.com
  [ 2] test2-abc.net.aol.com
  [ 0] Exit

Enter a device number: 2
<NetDevice: test2-abc.net.aol.com>

If there is only a single match, that device object is returned without a prompt:

>>> device_match('fw')
Matched 'fw1-xyz.net.aol.com'.
<NetDevice: fw1-xyz.net.aol.com>
class trigger.netdevices.NetDevice(data=None, with_acls=None)

An object that represents a distinct network device and its metadata.

Almost all of the attributes are populated by _populate() and are mostly dependent upon the source data. This is prone to implementation problems and should be revisited in the long-run as there are certain fields that are baked into the core functionality of Trigger.

Users usually won’t create these objects directly! Rely instead upon NetDevices to do this for you.

allowable(action, when=None)

Return whether it’s okay to perform the specified action.

False means a bounce window conflict. For now 'load-acl' is the only valid action and moratorium status is not checked.

Parameters:
  • action – The action to check.
  • when – A datetime object.
can_ssh_async()

Am I enabled to use SSH async?

can_ssh_pty()

Am I enabled to use SSH pty?

close()

Close an open NetDevice object.

dump()

Prints details for a device.

has_ssh()

Am I even listening on SSH?

is_brocade_vdx()

Am I a Brocade VDX switch?

This is used to account for the disparity between the Brocade FCX switches (which behave like Foundry devices) and the Brocade VDX switches (which behave differently from classic Foundry devices).

is_cisco_asa()

Am I a Cisco ASA Firewall?

This is used to account for slight differences in the commands that may be used between Cisco’s ASA and IOS platforms. Cisco ASA is still very IOS-like, but there are still several gotcha’s between the platforms.

Will return True if vendor is Cisco and platform is Firewall. This is to allow operability if using .csv NetDevices and pretty safe to assume considering ASA (was PIX) are Cisco’s flagship(if not only) Firewalls.

is_cisco_nexus()

Am I a Cisco Nexus device?

is_cumulus()

Am I running Cumulus?

is_firewall()

Am I a firewall?

is_ioslike()

Am I an IOS-like device (as determined by IOSLIKE_VENDORS)?

is_netscaler()

Am I a NetScaler?

is_netscreen()

Am I a NetScreen running ScreenOS?

is_pica8()

Am I a Pica8?

is_reachable()

Do I respond to a ping?

is_router()

Am I a router?

is_switch()

Am I a switch?

next_ok(action, when=None)

Return the next time at or after the specified time (default now) that it will be ok to perform the specified action.

Parameters:
  • action – The action to check.
  • when – A datetime object.
open()

Open new session with NetDevice.

Example:
>>> nd = NetDevices()
>>> dev = nd.find('arista-sw1.demo.local')
>>> dev.open()
run_channeled_commands(commands, on_error=None)

Public method for scheduling commands onto device.

This variant allows for efficient multiplexing of commands across multiple vty lines where supported ie Arista and Cumulus.

Parameters:
  • commands (list) – List containing commands to schedule onto device loop.
  • on_error (func) – Error handler
Example:
>>> ...
>>> dev.open()
>>> dev.run_channeled_commands(['show ip int brief', 'show version'], on_error=lambda x: handle(x))
run_commands(commands, on_error=None)

Public method for scheduling commands onto device.

Default implementation that schedules commands onto a Device loop. This implementation ensures commands are executed sequentially.

Parameters:
  • commands (list) – List containing commands to schedule onto device loop.
  • on_error (func) – Error handler
Example:
>>> ...
>>> dev.open()
>>> dev.run_commands(['show ip int brief', 'show version'], on_error=lambda x: handle(x))
class trigger.netdevices.Vendor(manufacturer=None)

Map a manufacturer name to Trigger’s canonical name.

Given a manufacturer name like ‘CISCO SYSTEMS’, this will attempt to map it to the canonical vendor name specified in settings.VENDOR_MAP. If this can’t be done, attempt to split the name up (‘CISCO, ‘SYSTEMS’) and see if any of the words map. An exception is raised as a last resort.

This exposes a normalized name that can be used in the event of a multi-word canonical name.

determine_vendor(manufacturer)

Try to turn the provided vendor name into the cname.

normalized

Return the normalized name for the vendor.

class trigger.netdevices.NetDevices(production_only=True, with_acls=None)

Returns an immutable Singleton dictionary of NetDevice objects.

By default it will only return devices for which adminStatus=='PRODUCTION'.

There are hardly any use cases where NON-PRODUCTION devices are needed, and it can cause real bugs of two sorts:

  1. trying to contact unreachable devices and reporting spurious failures,
  2. hot spares with the same nodeName.

You may override this by passing production_only=False.

class _actual(production_only=True, with_acls=None)

This is the real class that stays active upon instantiation. All attributes are inherited by NetDevices from this object. This means you do NOT reference _actual itself, and instead call the methods from the parent object.

Right:

>>> nd = NetDevices()
>>> nd.search('fw')
[<NetDevice: fw1-xyz.net.aol.com>]

Wrong:

>>> nd._actual.search('fw')
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
TypeError: unbound method match() must be called with _actual
instance as first argument (got str instance instead)
add_device(device)

Add a device object to the store.

Parameters:deviceNetDevice object
all()

Returns all NetDevice objects.

This method can be overloaded in NetDevices loader plugins to customize the behavior as dictated by the plugin.

find(key)

Return either the exact nodename, or a unique dot-delimited prefix. For example, if there is a node ‘test1-abc.net.aol.com’, then any of find(‘test1-abc’) or find(‘test1-abc.net’) or find(‘test1-abc.net.aol.com’) will match, but not find(‘test1’).

This method can be overloaded in NetDevices loader plugins to customize the behavior as dictated by the plugin.

Parameters:key (string) – Hostname prefix to find.
Returns:NetDevice object
get_devices_by_type(devtype)

Returns a list of NetDevice objects with deviceType matching type.

Known deviceTypes: [‘FIREWALL’, ‘ROUTER’, ‘SWITCH’]

list_firewalls()

Returns a list of NetDevice objects with deviceType of FIREWALL

list_routers()

Returns a list of NetDevice objects with deviceType of ROUTER

list_switches()

Returns a list of NetDevice objects with deviceType of SWITCH

match(**kwargs)

Attempt to match values to all keys in @kwargs by dynamically building a list comprehension. Will throw errors if the keys don’t match legit NetDevice attributes.

Keys and values are case IN-senstitive. Matches against non-string values will FAIL.

This method can be overloaded in NetDevices loader plugins to customize the behavior as dictated by the plugin. If skip_loader=True the built-in method will be used instead.

Example by reference:

>>> nd = NetDevices()
>>> myargs = {'onCallName':'Data Center', 'model':'FCSLB'}
>>> mydevices = nd(**myargs)

Example by keyword arguments:

>>> mydevices = nd(oncallname='data center', model='fcslb')
Returns:List of NetDevice objects
search(token, field='nodeName')

Returns a list of NetDevice objects where other is in dev.nodeName. The getattr call in the search will allow a AttributeError from a bogus field lookup so that you don’t get an empty list thinking you performed a legit query.

For example, this:

>>> field = 'bacon'
>>> [x for x in nd.all() if 'ash' in getattr(x, field)]
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
AttributeError: 'NetDevice' object has no attribute 'bacon'

Is better than this:

>>> [x for x in nd.all() if 'ash' in getattr(x, field, '')]
[]

Because then you know that ‘bacon’ isn’t a field you can search on.

Parameters:
  • token (string) – Token to search match on in @field
  • field (string) – The field to match on when searching
Returns:

List of NetDevice objects

set_loader(loader)

Set the NetDevices loader and initialize internal dictionary.

Parameters:loader – A BaseLoader plugin instance