From e11650feb96e304a247d60e5ba69860ef605526f Mon Sep 17 00:00:00 2001 From: chidanandpujar Date: Mon, 6 Oct 2025 02:24:19 -0700 Subject: [PATCH 1/2] support for allow_agent paramter --- lib/jnpr/junos/device.py | 23 +++++- tests/functional/test_device_ssh.py | 108 ++++++++++++++++++++++++++-- 2 files changed, 122 insertions(+), 9 deletions(-) diff --git a/lib/jnpr/junos/device.py b/lib/jnpr/junos/device.py index 27133661c..50db84293 100644 --- a/lib/jnpr/junos/device.py +++ b/lib/jnpr/junos/device.py @@ -1217,6 +1217,14 @@ def __init__(self, *vargs, **kvargs): *OPTIONAL* To disable public key authentication. default is ``None``. + :param bool allow_agent: + *OPTIONAL* Specifies whether to use keys provided by an SSH agent for authentication. + If set to ``True``, the SSH connection will use any keys loaded in the agent. + If set to ``False``, keys from the SSH agent will not be used. + If set to ``None``, the default behavior is applied: agent keys are used only if + both password and private key file are not provided. + Default is ``None``. + :param str bind_addr: *OPTIONAL* To use (local) source IP address. default is ``None``. @@ -1243,6 +1251,7 @@ def __init__(self, *vargs, **kvargs): self._huge_tree = kvargs.get("huge_tree", False) self._conn_open_timeout = kvargs.get("conn_open_timeout", 30) self._look_for_keys = kvargs.get("look_for_keys", None) + self._allow_agent = kvargs.get("allow_agent", None) self._bind_addr = kvargs.get("bind_addr", None) self._hostkey_verify = kvargs.get("hostkey_verify", False) if self._fact_style != "new": @@ -1367,9 +1376,17 @@ def open(self, *vargs, **kvargs): # in this condition it means we want to query the agent # for available ssh keys - allow_agent = bool( - (self._auth_password is None) and (self._ssh_private_key_file is None) - ) + if self._allow_agent is True: + # If set to True by the user, override with value as per user. + allow_agent = self._allow_agent + elif self._allow_agent is False: + # If set to False by the user, override with value as per user. + allow_agent = self._allow_agent + else: + # Default behaviour if allow_agent is None + allow_agent = bool( + (self._auth_password is None) and (self._ssh_private_key_file is None) + ) # option to disable ncclient transport ssh authentication # using public keys look_for_keys=False diff --git a/tests/functional/test_device_ssh.py b/tests/functional/test_device_ssh.py index 1302a05ce..87b545bcd 100644 --- a/tests/functional/test_device_ssh.py +++ b/tests/functional/test_device_ssh.py @@ -1,9 +1,11 @@ __author__ = "rsherman, vnitinv" -import unittest - from jnpr.junos import Device +try: + import unittest2 as unittest +except ImportError: + import unittest class TestDeviceSsh(unittest.TestCase): def tearDown(self): @@ -11,10 +13,104 @@ def tearDown(self): def test_device_open_key_pass(self): self.dev = Device( - host="xxxx", - user="jenkins", - ssh_private_key_file="/var/lib/jenkins/.ssh/passkey", - passwd="password", + host="x.x.x.x", + user="netops", + ssh_private_key_file="~/.ssh/id_rsa", + passwd="net123", + ) + self.dev.open() + self.assertEqual(self.dev.connected, True) + + def test_device_open_password(self): + self.dev = Device( + host="x.x.x.x", + user="netops", + passwd="net123", + ) + self.dev.open() + self.assertEqual(self.dev.connected, True) + + def test_device_open_ssh_agent_true(self): + self.dev = Device( + host="x.x.x.x", + user="netops", + allow_agent=True + ) + self.dev.open() + self.assertEqual(self.dev.connected, True) + + def test_device_open_ssh_agent_false(self): + self.dev = Device( + host="x.x.x.x", + user="netops", + allow_agent=False, + ) + self.dev.open() + self.assertEqual(self.dev.connected, True) + + def test_device_open_key_file(self): + self.dev = Device( + host="x.x.x.x", + user="netops", + ssh_private_key_file="~/.ssh/id_rsa", + ) + self.dev.open() + self.assertEqual(self.dev.connected, True) + + def test_device_open_key_file(self): + self.dev = Device( + host="x.x.x.x", + user="netops", + ssh_private_key_file="~/.ssh/id_rsa", + ) + self.dev.open() + self.assertEqual(self.dev.connected, True) + + def test_device_open_proxy(self): + self.dev = Device( + host="x.x.x.x", + user="netops", + proxy_command="ssh -J netops@y.y.y.y" + ) + self.dev.open() + self.assertEqual(self.dev.connected, True) + + def test_device_open_ssh_agent_proxy(self): + self.dev = Device( + host="x.x.x.x", + user="netops", + proxy_command="ssh -J netops@y.y.y.y", + allow_agent=True, + ) + self.dev.open() + self.assertEqual(self.dev.connected, True) + + def test_device_open_key_file_proxy(self): + self.dev = Device( + host="x.x.x.x", + user="netops", + proxy_command="ssh -J netops@y.y.y.y", + ssh_private_key_file="~/.ssh/id_rsa", + ) + self.dev.open() + self.assertEqual(self.dev.connected, True) + + def test_device_open_ssh_agent_proxy(self): + self.dev = Device( + host="x.x.x.x", + user="netops", + proxy_command="ssh -J netops@y.y.y.y", + allow_agent=True, + ) + self.dev.open() + self.assertEqual(self.dev.connected, True) + + def test_device_open_key_file_proxy(self): + self.dev = Device( + host="x.x.x.x", + user="netops", + proxy_command="ssh -J netops@y.y.y.y", + ssh_private_key_file="~/.ssh/id_rsa", ) self.dev.open() self.assertEqual(self.dev.connected, True) From eb7836cb916c374408f85cc3b187f297dc5be4fb Mon Sep 17 00:00:00 2001 From: chidanandpujar Date: Mon, 6 Oct 2025 02:34:25 -0700 Subject: [PATCH 2/2] black tool error --- lib/jnpr/junos/device.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/lib/jnpr/junos/device.py b/lib/jnpr/junos/device.py index 50db84293..0c48be0e2 100644 --- a/lib/jnpr/junos/device.py +++ b/lib/jnpr/junos/device.py @@ -1385,7 +1385,8 @@ def open(self, *vargs, **kvargs): else: # Default behaviour if allow_agent is None allow_agent = bool( - (self._auth_password is None) and (self._ssh_private_key_file is None) + (self._auth_password is None) + and (self._ssh_private_key_file is None) ) # option to disable ncclient transport ssh authentication