Junos 'EZ' automation library for remotely managing and automating Juniper Networks Junos devices through NETCONF protocol
—
Core device connection functionality for managing NETCONF and console sessions with Junos devices, including authentication, session management, connectivity testing, and connection lifecycle operations.
Main class for NETCONF connections to Junos devices supporting SSH transport, authentication methods, session management, and automatic facts gathering.
class Device:
def __init__(self, *vargs, **kvargs):
"""
Initialize Device connection object.
Parameters (passed as keyword arguments):
- host (str): Target device hostname or IP address (can also be first positional arg)
- user (str): Login username
- passwd (str): Login password
- port (int): NETCONF SSH port (default: 830)
- ssh_private_key_file (str): Path to SSH private key file
- ssh_config (str): Path to SSH config file
- timeout (int): Default RPC timeout in seconds
- gather_facts (bool): Automatically gather device facts on connect
- auto_probe (int): Auto-probe timeout before connection attempt
- sock_fd (socket): Existing socket file descriptor
- session_listener: NETCONF session event listener
- conn_open_timeout (int): Connection establishment timeout
- mode (str): Connection mode for console connections
- baud (int): Baud rate for serial console connections
- cs_user (str): Console server username
- cs_passwd (str): Console server password
- fact_style (str): Facts gathering style ('new' or 'old')
- cs_timeout (int): Console connection timeout
"""Methods for establishing, managing, and terminating device connections with comprehensive error handling and connection state management.
def open(self, **kwargs):
"""
Open connection to the device.
Parameters:
- auto_probe (int): Probe timeout before connection
- gather_facts (bool): Gather facts after connection
- normalize (bool): Normalize RPC responses
Raises:
- ConnectError: Base connection error
- ConnectAuthError: Authentication failure
- ConnectTimeoutError: Connection timeout
- ConnectUnknownHostError: Unknown host
- ConnectRefusedError: Connection refused
"""
def close(self):
"""
Close the device connection.
Cleanly terminates the NETCONF session and releases resources.
"""
def probe(self, timeout=5):
"""
Test connectivity to device without establishing NETCONF session.
Parameters:
- timeout (int): Probe timeout in seconds
Returns:
- bool: True if device is reachable
Raises:
- ProbeError: Probe operation failed
"""Console connection class for telnet and serial connections to Junos devices, supporting out-of-band management and console server connections.
class Console:
def __init__(
self,
host=None,
user='root',
mode='telnet',
baud=9600,
timeout=0.5,
attempts=10,
ssh_private_key_file=None,
port=23,
cs_user=None,
cs_passwd=None,
gather_facts=True,
auto_probe=0,
fact_style='new',
cs_timeout=90
):
"""
Initialize Console connection object.
Parameters:
- host (str): Console server hostname or IP address
- user (str): Device login username (default: 'root')
- mode (str): Connection mode ('telnet' or 'serial')
- baud (int): Serial connection baud rate
- timeout (float): Operation timeout in seconds
- attempts (int): Connection retry attempts
- ssh_private_key_file (str): SSH private key for console server
- port (int): Console server port (default: 23 for telnet)
- cs_user (str): Console server username
- cs_passwd (str): Console server password
- gather_facts (bool): Gather facts after connection
- auto_probe (int): Auto-probe timeout
- fact_style (str): Facts gathering style
- cs_timeout (int): Console operation timeout
"""
def open(self, **kwargs):
"""
Open console connection to device.
Establishes telnet or serial console connection and performs
device login authentication.
"""
def close(self):
"""
Close console connection.
Cleanly terminates the console session.
"""
def zeroize(self):
"""
Perform device zeroize operation.
WARNING: This completely erases device configuration and
returns device to factory defaults.
"""Properties for accessing connection state, device information, and session details.
# Read-only properties
@property
def connected(self) -> bool:
"""Connection status (True if connected, False otherwise)"""
@property
def hostname(self) -> str:
"""Device hostname from facts or connection parameters"""
@property
def user(self) -> str:
"""Login username for the connection"""
@property
def master(self) -> bool:
"""True if connected to master routing engine"""
@property
def re_name(self) -> str:
"""Current routing engine name"""
# Read-write properties
@property
def timeout(self) -> int:
"""Default RPC timeout in seconds"""
@timeout.setter
def timeout(self, value: int):
"""Set default RPC timeout"""
@property
def transform(self):
"""XML transformation function for RPC responses"""
@transform.setter
def transform(self, func):
"""Set XML transformation function"""from jnpr.junos import Device
# Basic connection with username/password
dev = Device(host='router1.example.com', user='admin', passwd='secret')
dev.open()
print(f"Connected to {dev.hostname}")
print(f"Connection status: {dev.connected}")
dev.close()from jnpr.junos import Device
# Connection using SSH private key
dev = Device(
host='192.168.1.1',
user='admin',
ssh_private_key_file='/home/user/.ssh/id_rsa'
)
dev.open()
# Device is now connected
dev.close()from jnpr.junos import Console
# Telnet console connection
console = Console(
host='console-server.example.com',
port=2001,
user='root',
mode='telnet'
)
console.open()
# Perform console operations
console.close()from jnpr.junos import Device
# Test connectivity before connecting
dev = Device(host='router1.example.com', user='admin', passwd='secret')
# Probe first
if dev.probe(timeout=10):
print("Device is reachable")
dev.open()
else:
print("Device is not reachable")from jnpr.junos import Device
from jnpr.junos.exception import ConnectError, ConnectAuthError, ConnectTimeoutError
dev = Device(host='router1.example.com', user='admin', passwd='secret')
try:
dev.open()
print("Successfully connected")
except ConnectAuthError:
print("Authentication failed")
except ConnectTimeoutError:
print("Connection timed out")
except ConnectError as e:
print(f"Connection failed: {e}")
finally:
if dev.connected:
dev.close()# Connection parameters type
ConnectionParams = dict[str, any] # Dict with connection parameters
# Socket file descriptor type
SocketFD = int # File descriptor for existing socket connection
# Session listener type
SessionListener = object # NETCONF session event listener objectInstall with Tessl CLI
npx tessl i tessl/pypi-junos-eznc