tessl/pypi-ncclient

Python library for NETCONF clients with support for SSH, TLS, and device-specific operations

—

Pending

Overview

Eval results

Files

Transport Layer

Name: tessl/pypi-ncclient
Author: tessl

Low-level transport implementations providing the communication foundation for NETCONF sessions. NCClient supports multiple transport protocols with connection multiplexing, session management, and protocol-specific optimizations.

Capabilities

Base Transport Classes

Foundation classes providing common transport functionality and interfaces.

class Session:
    """Base transport session class providing common functionality."""
    
    def connect(self, *args, **kwargs):
        """Establish transport connection."""
        
    def close(self):
        """Close transport connection."""
        
    @property
    def connected(self):
        """bool: Connection status."""
        
    @property
    def id(self):
        """str: Session ID assigned by server."""
        
    def send(self, message):
        """Send message over transport."""
        
    def take_notification(self, block=True, timeout=None):
        """Retrieve notification from queue."""

class SessionListener:
    """Interface for session event handling."""
    
    def callback(self, root, raw):
        """Handle received message callback."""
        
    def errback(self, ex):
        """Handle error callback."""

class NetconfBase:
    """Base NETCONF protocol functionality."""

SSH Transport

Primary transport implementation using SSH for secure NETCONF communication.

class SSHSession(Session):
    """SSH transport session with NETCONF over SSH subsystem."""
    
    def __init__(self, device_handler):
        """
        Initialize SSH session.
        
        Parameters:
        - device_handler: device-specific handler
        """
    
    def connect(self, host, port=830, timeout=None, 
               username=None, password=None, key_filename=None,
               allow_agent=True, hostkey_verify=True, 
               look_for_keys=True, sock=None):
        """
        Connect via SSH transport.
        
        Parameters:
        - host: str, hostname or IP address
        - port: int, SSH port (default 830)
        - timeout: float, connection timeout
        - username: str, SSH username
        - password: str, SSH password
        - key_filename: str or list, SSH private key file(s)
        - allow_agent: bool, use SSH agent
        - hostkey_verify: bool, verify host key
        - look_for_keys: bool, search for SSH keys
        - sock: socket, existing socket connection
        """
    
    def scp(self):
        """Create SCP client for file transfer."""
        
    @property
    def transport(self):
        """SSH transport object."""

LibSSH Transport

Alternative SSH implementation using ssh-python library for enhanced performance.

class LibSSHSession(Session):
    """LibSSH-based transport session (requires ssh-python)."""
    
    def __init__(self, device_handler):
        """
        Initialize LibSSH session.
        
        Parameters:
        - device_handler: device-specific handler
        """
    
    def connect(self, host, port=830, username=None, 
               password=None, **kwargs):
        """
        Connect via LibSSH transport.
        
        Parameters:
        - host: str, hostname or IP address
        - port: int, SSH port (default 830)
        - username: str, SSH username
        - password: str, SSH password
        """

TLS Transport

Secure transport implementation using TLS for NETCONF over TLS communication.

class TLSSession(Session):
    """TLS transport session for NETCONF over TLS."""
    
    def __init__(self, device_handler):
        """
        Initialize TLS session.
        
        Parameters:
        - device_handler: device-specific handler
        """
    
    def connect(self, host, port=6513, certfile=None, 
               keyfile=None, ca_certs=None, cert_reqs=None,
               ssl_version=None, ciphers=None, **kwargs):
        """
        Connect via TLS transport.
        
        Parameters:
        - host: str, hostname or IP address
        - port: int, TLS port (default 6513)
        - certfile: str, client certificate file
        - keyfile: str, client private key file
        - ca_certs: str, CA certificates file
        - cert_reqs: int, certificate verification requirements
        - ssl_version: int, SSL/TLS protocol version
        - ciphers: str, cipher suite specification
        """

Unix Domain Socket Transport

Local transport implementation using Unix domain sockets (Unix platforms only).

class UnixSocketSession(Session):
    """Unix domain socket transport session."""
    
    def __init__(self, device_handler):
        """
        Initialize Unix socket session.
        
        Parameters:
        - device_handler: device-specific handler
        """
    
    def connect(self, path, **kwargs):
        """
        Connect via Unix domain socket.
        
        Parameters:
        - path: str, socket file path
        """

Transport Errors

Exception hierarchy for transport-specific error conditions.

class TransportError(NCClientError):
    """Base exception for transport layer errors."""

class AuthenticationError(TransportError):
    """Authentication failure during transport setup."""

class SessionCloseError(TransportError):
    """Error during session close operation."""

class SSHError(TransportError):
    """SSH transport specific error."""

class SSHUnknownHostError(SSHError):
    """SSH unknown host key error."""

Protocol Support

NETCONF Protocol Features

# NETCONF base capabilities
NETCONF_BASE_10 = "urn:ietf:params:netconf:base:1.0"
NETCONF_BASE_11 = "urn:ietf:params:netconf:base:1.1"

# Chunked framing support (NETCONF 1.1)
def supports_chunked_framing(capabilities):
    """Check if server supports chunked framing."""

# EOM framing support (NETCONF 1.0) 
def supports_eom_framing(capabilities):
    """Check if server supports end-of-message framing."""

Message Parsing

class Parser:
    """NETCONF message parser for framing and XML processing."""
    
    def __init__(self, session):
        """Initialize parser for session."""
    
    def parse(self, data):
        """Parse incoming NETCONF message data."""

class DefaultParser(Parser):
    """Standard NETCONF message parser."""

class ChunkedFramingParser(Parser):
    """Parser for NETCONF 1.1 chunked framing."""

Notification Handling

class NotificationListener:
    """Handles NETCONF event notifications."""
    
    def __init__(self, session):
        """Initialize notification listener."""
    
    def add_listener(self, callback):
        """Add notification callback."""
    
    def remove_listener(self, callback):
        """Remove notification callback."""

class Notification:
    """NETCONF event notification."""
    
    @property
    def event_time(self):
        """str: Notification timestamp."""
        
    @property
    def notification_xml(self):
        """str: Notification content as XML."""

Usage Examples

Direct SSH Transport Usage

from ncclient.transport import SSHSession
from ncclient.devices.default import DefaultDeviceHandler

# Create device handler
device_handler = DefaultDeviceHandler({}, [])

# Initialize SSH session
session = SSHSession(device_handler) 

try:
    # Connect with SSH parameters
    session.connect(
        host='192.168.1.1',
        port=830,
        username='netconf-user',
        password='netconf-pass',
        hostkey_verify=False,
        timeout=30
    )
    
    print(f"Connected with session ID: {session.id}")
    print(f"Connection status: {session.connected}")
    
finally:
    session.close()

TLS Transport with Certificates

from ncclient.transport import TLSSession
from ncclient.devices.default import DefaultDeviceHandler

device_handler = DefaultDeviceHandler({}, [])
session = TLSSession(device_handler)

try:
    session.connect(
        host='secure-device.example.com',
        port=6513,
        certfile='/path/to/client.crt',
        keyfile='/path/to/client.key',
        ca_certs='/path/to/ca.crt'
    )
    
    print("Secure TLS connection established")
    
finally:
    session.close()

Unix Socket Transport

from ncclient.transport import UnixSocketSession
from ncclient.devices.default import DefaultDeviceHandler
import sys

# Unix sockets not available on Windows
if sys.platform != 'win32':
    device_handler = DefaultDeviceHandler({}, [])
    session = UnixSocketSession(device_handler)
    
    try:
        session.connect(path='/var/run/netconf.sock')
        print("Connected via Unix domain socket")
    finally:
        session.close()

Custom Session Listener

from ncclient.transport import SessionListener
from ncclient import manager

class MySessionListener(SessionListener):
    def callback(self, root, raw):
        print(f"Received message: {raw[:100]}...")
    
    def errback(self, ex):
        print(f"Session error: {ex}")

# Use with manager (implementation detail)
listener = MySessionListener()

Error Handling

from ncclient.transport import (
    TransportError, AuthenticationError, 
    SSHError, SSHUnknownHostError
)

try:
    with manager.connect_ssh(
        host='untrusted-device.example.com',
        username='admin',
        password='admin',
        hostkey_verify=True
    ) as m:
        pass
        
except SSHUnknownHostError as e:
    print("Unknown host key - add to known_hosts")
except AuthenticationError as e:
    print("Authentication failed - check credentials")
except SSHError as e:
    print(f"SSH error: {e}")
except TransportError as e:
    print(f"Transport error: {e}")

Install with Tessl CLI