Collection of NETCONF tasks and connection plugin for Nornir
pip install nornir_netconf- netconf - Connect to network devices using ncclient
- netconf_capabilities - Return server capabilities from target ->
Result.result -> RpcResult - netconf_commit - Commits a change ->
Result.result -> RpcResult - netconf_edit_config - Edits configuration on specified datastore (default="running") ->
Result.result -> RpcResult - netconf_get - Returns state data based on the supplied xpath ->
Result.result -> RpcResult - netconf_get_config - Returns configuration from specified configuration store (default="running") ->
Result.result -> RpcResult - netconf_get_schemas - Retrieves schemas and saves aggregates content into a directory with schema output ->
Result.result -> SchemaResult - netconf_lock - Locks or Unlocks a specified datastore (default="lock") ->
Result.result -> RpcResult - netconf_validate - Validates configuration datastore. Requires the
validatecapability. ->Result.result -> RpcResult
The goal of the task results is to put the NETCONF RPC-reply back in your hands. In most cases, the Nornir Result.result attribute will return back a dataclass depending on the task operation. It's important that you understand the object you will be working with. Please see the dataclasses section below and review the code if you want to see what attributes to expect.
Defined in
nornir_netconf/plugins/helpers/models.py
RpcResult-> This will return an attribute ofrpcandmanager. You will encounter this object in most NornirResultsas the return value to theresultattribute. NETCONF / XML payloads can be overwhelming, especially with large configurations and it's just not efficient or useful to display thousands of lines of code in any result.SchemaResult-> An aggregation of interesting information when grabbing schemas from NETCONF servers.
The netconf_lock task will always return the Manager object, which is the established (and locked) agent used to send RPC's back and forth. The idea of retrieving the Manager is to carry this established locked session from task to task and only lock and unlock once during a run of tasks. Please review the examples below to see how to extract the manager and store it under the task.host dictionary as a variable that can be used across multiple tasks. The Manager is passed into other tasks and re-used to send RPCs to the remote server.
Head over to the Examples directory if you'd like to review the files.
Directory Structure
├── example-project
│ ├── config.yml
│ ├── inventory
│ │ ├── groups.yml
│ │ ├── hosts-local.yml
│ │ └── ssh_config
│ ├── logs
│ │ └── nornir.log
│ └── nr-get-config.py
└── README.mdNetconf Connection Plugin
Below is the snippet of a host inside the host-local.yml file and its associated group, sros.
nokia_rtr:
hostname: "192.168.1.205"
port: 830
groups:
- "sros"sros:
username: "netconf"
password: "NCadmin123"
port: 830
platform: "sros"
connection_options:
netconf:
extras:
hostkey_verify: false
timeout: 300
allow_agent: false
look_for_keys: falseTask: Get Config
"""Nornir NETCONF Example Task: 'get-config'."""
from nornir import InitNornir
from nornir.core.task import Task
from nornir_utils.plugins.functions import print_result
from nornir_netconf.plugins.tasks import netconf_get_config
__author__ = "Hugo Tinoco"
__email__ = "[email protected]"
nr = InitNornir("config.yml")
# Filter the hosts by 'west-region' assignment
west_region = nr.filter(region="west-region")
def example_netconf_get_config(task: Task) -> str:
"""Test get config."""
config = task.run(
netconf_get_config,
source="running",
path="""
<configure xmlns="urn:nokia.com:sros:ns:yang:sr:conf">
<router>
<router-name>Base</router-name>
</router>
</configure>
""",
filter_type="subtree",
)
return config.result.rpc.data_xml
def main():
"""Execute Nornir Script."""
print_result(west_region.run(task=example_netconf_get_config))
if __name__ == "__main__":
main()This returns the following
vvvv example_netconf_get_config ** changed : False vvvvvvvvvvvvvvvvvvvvvvvvvvvvv INFO
<?xml version="1.0" encoding="UTF-8"?><rpc-reply xmlns="urn:ietf:params:xml:ns:netconf:base:1.0" message-id="urn:uuid:918bf169-523f-4bb0-b00c-c97c01a48ecd">
<data>
<configure xmlns="urn:nokia.com:sros:ns:yang:sr:conf">
<router>
<router-name>Base</router-name>
<interface>
<interface-name>L3-OAM-eNodeB069420-X1</interface-name>
<admin-state>disable</admin-state>
<ingress-stats>false</ingress-stats>
</interface>
</router>
</configure>
</data>
</rpc-reply>
---- netconf_get_config ** changed : False ------------------------------------- INFO
RpcResult(rpc=<ncclient.xml_.NCElement object at 0x7f4b1e08a440>)
^^^^ END example_netconf_get_config ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
(nornir-netconf-Ky5gYI2O-py3.10) ➜ example-project git:(feature/validate-tasks) ✗ Task: Get Capabilities
"""Nornir NETCONF Example Task: 'capabilities'."""
from nornir import InitNornir
from nornir.core.task import Task
from nornir_utils.plugins.functions import print_result
from nornir_netconf.plugins.tasks import netconf_capabilities
__author__ = "Hugo Tinoco"
__email__ = "[email protected]"
nr = InitNornir("config.yml")
# Filter the hosts by 'west-region' assignment
west_region = nr.filter(region="west-region")
def example_netconf_get_capabilities(task: Task) -> str:
"""Test get capabilities."""
capabilities = task.run(netconf_capabilities)
# This may be a lot, so for example we'll just print the first one
return [cap for cap in capabilities.result.rpc][0]
def main():
"""Execute Nornir Script."""
print_result(west_region.run(task=example_netconf_get_capabilities))
if __name__ == "__main__":
main()This returns the following
(nornir-netconf-Ky5gYI2O-py3.10) ➜ example-project git:(feature/validate-tasks) ✗ python3 nr_get_capabilities.py
example_netconf_get_capabilities************************************************
* nokia_rtr ** changed : False *************************************************
vvvv example_netconf_get_capabilities ** changed : False vvvvvvvvvvvvvvvvvvvvvvv INFO
urn:ietf:params:netconf:base:1.0
---- netconf_capabilities ** changed : False ----------------------------------- INFO
RpcResult(rpc=<dict_keyiterator object at 0x7f7111328c70>)
^^^^ END example_netconf_get_capabilities ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
(nornir-netconf-Ky5gYI2O-py3.10) ➜ example-project git:(feature/validate-tasks) ✗ Task: Edit-Config with Global Lock
"""Nornir NETCONF Example Task: 'edit-config', 'netconf_lock'."""
from nornir import InitNornir
from nornir_utils.plugins.functions import print_result
from nornir_netconf.plugins.tasks import netconf_edit_config, netconf_lock, netconf_commit
__author__ = "Hugo Tinoco"
__email__ = "[email protected]"
nr = InitNornir("config.yml")
# Filter the hosts by 'west-region' assignment
west_region = nr.filter(region="west-region")
def example_global_lock(task):
"""Test global lock operation of 'candidate' datastore."""
lock = task.run(netconf_lock, datastore="candidate", operation="lock")
# Retrieve the Manager(agent) from lock operation and store for further
# operations.
task.host["manager"] = lock.result.manager
def example_edit_config(task):
"""Test edit-config with global lock using manager agent."""
config_payload = """
<config>
<configure xmlns="urn:nokia.com:sros:ns:yang:sr:conf">
<router>
<router-name>Base</router-name>
<interface>
<interface-name>L3-OAM-eNodeB069420-X1</interface-name>
<admin-state>disable</admin-state>
<ingress-stats>false</ingress-stats>
</interface>
</router>
</configure>
</config>
"""
result = task.run(
netconf_edit_config, config=config_payload, target="candidate", manager=task.host["manager"]
)
# Validate configuration
task.run(netconf_validate)
# Commit
task.run(netconf_commit, manager=task.host["manager"])
def example_unlock(task):
"""Unlock candidate datastore."""
task.run(netconf_lock, datastore="candidate", operation="unlock", manager=task.host["manager"])
def main():
"""Execute Nornir Script."""
print_result(west_region.run(task=example_global_lock))
print_result(west_region.run(task=example_edit_config))
print_result(west_region.run(task=example_unlock))
if __name__ == "__main__":
main()Task: Get Schemas
"""Get Schemas from NETCONF device."""
from nornir import InitNornir
from nornir.core import Task
from nornir.core.task import Result
from nornir_utils.plugins.functions import print_result
from nornir_netconf.plugins.tasks import netconf_get, netconf_get_schemas
from tests.conftest import xml_dict
__author__ = "Hugo Tinoco"
__email__ = "[email protected]"
nr = InitNornir("config.yml")
# Filter the hosts by 'west-region' assignment
west_region = nr.filter(region="west-region")
SCHEMA_FILTER = """
<netconf-state xmlns="urn:ietf:params:xml:ns:yang:ietf-netconf-monitoring">
<schemas>
</schemas>
</netconf-state>
"""
def example_task_get_schemas(task: Task) -> Result:
"""Get Schemas from NETCONF device."""
result = task.run(netconf_get, path=SCHEMA_FILTER, filter_type="subtree")
# xml_dict is a custom function to convert XML to Python dictionary. Not part of Nornir Plugin.
# See the code example if you want to use it.
parsed = xml_dict(result.result.rpc.data_xml)
first_schema = parsed["rpc-reply"]["data"]["netconf-state"]["schemas"]["schema"][0]
return task.run(netconf_get_schemas, schemas=[first_schema["identifier"]], schema_path="./output/schemas")
def main():
"""Execute Nornir Script."""
print_result(west_region.run(task=example_task_get_schemas))
if __name__ == "__main__":
main()This returns the following
(nornir-netconf-Ky5gYI2O-py3.10) ➜ example-project git:(feature/validate-tasks) ✗ python3 nr_get_schemas.py
example_task_get_schemas********************************************************
* nokia_rtr ** changed : False *************************************************
vvvv example_task_get_schemas ** changed : False vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv INFO
---- netconf_get ** changed : False -------------------------------------------- INFO
RpcResult(rpc=<ncclient.xml_.NCElement object at 0x7f36391540d0>)
---- netconf_get_schemas ** changed : False ------------------------------------ INFO
SchemaResult(directory='./output/schemas')
^^^^ END example_task_get_schemas ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^Github actions spins up a Containerlab instance to do full integration tests once linting has been satisfied.
No line of code shall go untested! Any contribution will need to be accounted for by the coverage report and satisfy all linting.
Linters:
- Ruff (Flake8/Pydocstyle)
- Black
- Yamllint
- Pylint
- Bandit
- MyPy
To test within a local docker environment
git clone https://github.com/h4ndzdatm0ld/nornir_netconfdocker-compose build && docker-compose run testTo test locally with pytest
If you'd like to run integration tests with ContainerLab
export SKIP_INTEGRATION_TESTS=Falsedocker-compose up -dpoetry install && poetry shellpytest --cov=nornir_netconf --color=yes --disable-pytest-warnings -vvvDevices with full integration tests with ContainerLab
- Nokia SROS - TiMOS-B-21.2.R1
- Cisco IOSxR - Cisco IOS XR Software, Version 6.1.3
- Cisco IOSXE - Cisco IOS XE Software, Version 17.03.02
- Arista CEOS - 4.28.0F-26924507.4280F (engineering build)
Documentation is generated with Sphinx and hosted with Github Pages. Documentation