tessl/pypi-junos-eznc

Junos 'EZ' automation library for remotely managing and automating Juniper Networks Junos devices through NETCONF protocol

—

Pending

Overview

Eval results

Files

File System Operations

Name: tessl/pypi-junos-eznc
Author: tessl

File system and transfer utilities for secure file operations, including SCP and FTP transfers, directory operations, file management, and shell session management for advanced file system operations.

Capabilities

File System Utility Class

The FS utility provides comprehensive file system operations including directory management, file operations, and storage information retrieval.

class FS:
    def __init__(self, dev):
        """
        Initialize FS utility bound to a device.
        
        Parameters:
        - dev (Device): Device object to bind to
        """

Secure Copy (SCP) Operations

The SCP utility provides secure file transfer capabilities for copying files to and from Junos devices over SSH connections.

class SCP:
    def __init__(self, dev):
        """
        Initialize SCP utility bound to a device.
        
        Parameters:
        - dev (Device): Device object to bind to
        """
    
    def put(self, local_file, remote_path='.', progress=None):
        """
        Copy local file to remote device using SCP.
        
        Parameters:
        - local_file (str): Local file path to upload
        - remote_path (str): Remote directory or file path
        - progress (callable): Progress callback function
        
        Returns:
        - bool: True if transfer successful
        
        Raises:
        - ScpError: SCP transfer failed
        - FileNotFoundError: Local file not found
        """
    
    def get(self, remote_file, local_path='.', progress=None):
        """
        Copy remote file from device to local system using SCP.
        
        Parameters:
        - remote_file (str): Remote file path to download
        - local_path (str): Local directory or file path
        - progress (callable): Progress callback function
        
        Returns:
        - bool: True if transfer successful
        
        Raises:
        - ScpError: SCP transfer failed
        """

FTP Operations

The FTP utility provides FTP file transfer capabilities for devices that support FTP file transfers.

class FTP:
    def __init__(self, dev):
        """
        Initialize FTP utility bound to a device.
        
        Parameters:
        - dev (Device): Device object to bind to
        """
    
    def put(self, local_file, remote_path='.', progress=None):
        """
        Upload local file to device using FTP.
        
        Parameters:
        - local_file (str): Local file path to upload
        - remote_path (str): Remote directory or file path
        - progress (callable): Progress callback function
        
        Returns:
        - bool: True if transfer successful
        
        Raises:
        - FtpError: FTP transfer failed
        """
    
    def get(self, remote_file, local_path='.', progress=None):
        """
        Download remote file from device using FTP.
        
        Parameters:
        - remote_file (str): Remote file path to download
        - local_path (str): Local directory or file path
        - progress (callable): Progress callback function
        
        Returns:
        - bool: True if transfer successful
        
        Raises:
        - FtpError: FTP transfer failed
        """

Shell Session Management

The StartShell utility provides shell session management for advanced file system operations and command execution outside of the NETCONF session.

class StartShell:
    def __init__(self, dev):
        """
        Initialize StartShell utility bound to a device.
        
        Parameters:
        - dev (Device): Device object to bind to
        """
    
    def open(self, **kwargs):
        """
        Open shell session on the device.
        
        Parameters:
        - **kwargs: Shell session parameters
        
        Returns:
        - bool: True if shell session opened successfully
        
        Raises:
        - ConnectError: Shell session failed to open
        """
    
    def close(self):
        """
        Close shell session.
        
        Returns:
        - bool: True if shell session closed successfully
        """
    
    def run(self, command, timeout=None):
        """
        Execute command in shell session.
        
        Parameters:
        - command (str): Shell command to execute
        - timeout (int): Command timeout in seconds
        
        Returns:
        - tuple: (status, output) where status is True/False and output is command result
        
        Raises:
        - ShellTimeoutError: Command execution timed out
        """

Utility Binding

Bind file system utilities to Device instances to enable file operations.

# Binding utilities to device
dev.bind(fs=FS)    # File system operations
dev.bind(scp=SCP)  # Secure copy operations
dev.bind(ftp=FTP)  # FTP operations
dev.bind(shell=StartShell)  # Shell operations

# Access bound utilities
dev.scp.put('/local/file.txt', '/var/tmp/')
dev.fs.method()  # File system operations

Usage Examples

Basic SCP File Transfer

from jnpr.junos import Device
from jnpr.junos.utils.scp import SCP

dev = Device(host='router1.example.com', user='admin', passwd='secret')
dev.open()

# Bind SCP utility
dev.bind(scp=SCP)

# Upload file to device
local_file = '/home/user/config-backup.txt'
remote_path = '/var/tmp/'

try:
    result = dev.scp.put(local_file, remote_path)
    if result:
        print("File uploaded successfully")
    else:
        print("File upload failed")
except Exception as e:
    print(f"SCP error: {e}")

# Download file from device
remote_file = '/var/log/messages'
local_path = '/home/user/device-logs/'

try:
    result = dev.scp.get(remote_file, local_path)
    if result:
        print("File downloaded successfully")
except Exception as e:
    print(f"SCP download error: {e}")

dev.close()

SCP with Progress Tracking

from jnpr.junos import Device
from jnpr.junos.utils.scp import SCP

def scp_progress(filename, size, sent):
    """Progress callback for SCP operations."""
    percent = (sent / size) * 100
    print(f"Transferring {filename}: {percent:.1f}% ({sent}/{size} bytes)")

dev = Device(host='router1.example.com', user='admin', passwd='secret')
dev.open()
dev.bind(scp=SCP)

# Upload large file with progress tracking
large_file = '/home/user/large-backup.tgz'

dev.scp.put(
    large_file, 
    '/var/tmp/', 
    progress=scp_progress
)

dev.close()

Multiple File Operations

from jnpr.junos import Device
from jnpr.junos.utils.scp import SCP
import os

dev = Device(host='router1.example.com', user='admin', passwd='secret')
dev.open()
dev.bind(scp=SCP)

# Upload multiple configuration files
config_files = [
    '/home/user/configs/interfaces.conf',
    '/home/user/configs/protocols.conf', 
    '/home/user/configs/policies.conf'
]

for config_file in config_files:
    if os.path.exists(config_file):
        filename = os.path.basename(config_file)
        print(f"Uploading {filename}...")
        
        result = dev.scp.put(config_file, '/var/tmp/')
        if result:
            print(f"  {filename} uploaded successfully")
        else:
            print(f"  {filename} upload failed")
    else:
        print(f"  {config_file} not found")

dev.close()

FTP File Transfer

from jnpr.junos import Device
from jnpr.junos.utils.ftp import FTP

dev = Device(host='router1.example.com', user='admin', passwd='secret')
dev.open()

# Bind FTP utility
dev.bind(ftp=FTP)

# Upload file using FTP
local_file = '/home/user/firmware.img'

try:
    result = dev.ftp.put(local_file, '/var/tmp/')
    if result:
        print("FTP upload successful")
except Exception as e:
    print(f"FTP error: {e}")

dev.close()

Shell Session Operations

from jnpr.junos import Device
from jnpr.junos.utils.start_shell import StartShell

dev = Device(host='router1.example.com', user='admin', passwd='secret')
dev.open()

# Bind shell utility
dev.bind(shell=StartShell)

try:
    # Open shell session
    dev.shell.open()
    
    # Execute shell commands
    status, output = dev.shell.run('ls -la /var/tmp/')
    if status:
        print("Directory listing:")
        print(output)
    
    # Check disk space
    status, output = dev.shell.run('df -h')
    if status:
        print("Disk usage:")
        print(output)
    
    # Create directory
    status, output = dev.shell.run('mkdir -p /var/tmp/backups')
    if status:
        print("Directory created successfully")
    
except Exception as e:
    print(f"Shell error: {e}")
finally:
    # Always close shell session
    dev.shell.close()

dev.close()

Backup and Restore Operations

from jnpr.junos import Device
from jnpr.junos.utils.scp import SCP
from jnpr.junos.utils.config import Config
import datetime

def backup_configuration(dev, backup_dir):
    """Backup device configuration."""
    hostname = dev.facts['hostname']
    timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
    
    # Get configuration
    config_xml = dev.rpc.get_config(format='xml')
    config_text = dev.rpc.get_config(format='text')
    
    # Save configurations locally
    xml_file = f"{backup_dir}/{hostname}_config_{timestamp}.xml"
    text_file = f"{backup_dir}/{hostname}_config_{timestamp}.txt"
    
    with open(xml_file, 'w') as f:
        f.write(str(config_xml))
    
    with open(text_file, 'w') as f:
        f.write(str(config_text))
    
    print(f"Configuration backed up to {xml_file} and {text_file}")
    return xml_file, text_file

dev = Device(host='router1.example.com', user='admin', passwd='secret')
dev.open()
dev.bind(scp=SCP)
dev.bind(cu=Config)

# Backup configuration
backup_files = backup_configuration(dev, '/home/user/backups')

# Upload backup to device for remote storage
for backup_file in backup_files:
    dev.scp.put(backup_file, '/var/tmp/backups/')
    print(f"Backup uploaded: {backup_file}")

dev.close()

Log File Collection

from jnpr.junos import Device
from jnpr.junos.utils.scp import SCP
import os

dev = Device(host='router1.example.com', user='admin', passwd='secret')
dev.open()
dev.bind(scp=SCP)

# Define log files to collect
log_files = [
    '/var/log/messages',
    '/var/log/chassisd',
    '/var/log/rpd',
    '/var/log/dcd'
]

# Create local directory for logs
hostname = dev.facts['hostname']
log_dir = f'/home/user/logs/{hostname}'
os.makedirs(log_dir, exist_ok=True)

# Download log files
for log_file in log_files:
    try:
        filename = os.path.basename(log_file)
        local_path = os.path.join(log_dir, filename)
        
        print(f"Downloading {log_file}...")
        result = dev.scp.get(log_file, local_path)
        
        if result:
            print(f"  Downloaded to {local_path}")
        else:
            print(f"  Failed to download {log_file}")
            
    except Exception as e:
        print(f"  Error downloading {log_file}: {e}")

print(f"Log collection completed. Files saved to {log_dir}")

dev.close()

Error Handling

from jnpr.junos import Device
from jnpr.junos.utils.scp import SCP
from jnpr.junos.exception import ScpError

dev = Device(host='router1.example.com', user='admin', passwd='secret')
dev.open()
dev.bind(scp=SCP)

local_file = '/home/user/test-file.txt'
remote_path = '/var/tmp/'

try:
    result = dev.scp.put(local_file, remote_path)
    print("File transfer successful")
    
except ScpError as e:
    print(f"SCP transfer failed: {e}")
    
except FileNotFoundError:
    print(f"Local file not found: {local_file}")
    
except PermissionError:
    print("Permission denied - check file permissions")
    
except Exception as e:
    print(f"Unexpected error: {e}")

dev.close()

Types

# File path types
LocalPath = str   # Local file or directory path
RemotePath = str  # Remote file or directory path

# Progress callback type
ProgressCallback = callable  # Function(filename, size, sent) for progress updates

# Transfer result type
TransferResult = bool  # True if transfer successful

# Shell command types
ShellCommand = str      # Shell command string
ShellOutput = str       # Shell command output
ShellStatus = bool      # Shell command execution status

# Shell result type
ShellResult = tuple[bool, str]  # (status, output) tuple

Install with Tessl CLI