ncclient:NETCONF Python client

# Copyright 2009 Shikhar Bhushan

#

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#

#    http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

import logging

from xml.etree import cElementTree as ElementTree

from cStringIO import StringIO

logger = logging.getLogger('ncclient.content')

def qualify(tag, ns=None):

    if ns is None:

        return tag

    else:

        return '{%s}%s' % (ns, tag)

_ = qualify

################################################################################

class Hello:

    NS = 'urn:ietf:params:xml:ns:netconf:base:1.0'

    @staticmethod

    def build(capabilities, encoding='utf-8'):

        hello = ElementTree.Element(_('hello', Hello.NS))

        caps = ElementTree.Element('capabilities')

        for uri in capabilities:

            cap = ElementTree.Element('capability')

            cap.text = uri

            caps.append(cap)

        hello.append(caps)

        tree = ElementTree.ElementTree(hello)

        fp = StringIO()

        tree.write(fp, encoding)

        return fp.getvalue()

    @staticmethod

    def parse(raw):

        'Returns tuple of (session-id, ["capability_uri", ...])'

        id, capabilities = 0, []

        root = ElementTree.fromstring(raw)

        if root.tag == _('hello', Hello.NS):

            for child in root.getchildren():

                if child.tag == _('session-id', Hello.NS):

                    id = int(child.text)

                elif child.tag == _('capabilities', Hello.NS):

                    for cap in child.getiterator(_('capability', Hello.NS)):

                        capabilities.append(cap.text)

        return id, capabilities

################################################################################

class RootElementParser:

    '''Parse the root element of an XML document. The tag and namespace of

    recognized elements, and attributes of interest can be customized.

    RootElementParser does not parse any sub-elements.

    '''

    def __init__(self, recognize=[]):

        self._recognize = recognize

    def recognize(self, element):

        '''Specify an element that should be successfully parsed.

        element should be a string that represents a qualified name of the form

        *{namespace}tag*.

        '''

        self._recognize.append((element, attrs))

    def parse(self, raw):

        '''Parse the root element from a string representing an XML document.

        Returns a (tag, attributes) tuple. tag is a string representing

        the qualified name of the recognized element. attributes is a

        {'attr': value} dictionary.

        '''

        fp = StringIO(raw)

        for event, element in ElementTree.iterparse(fp, events=('start',)):

            for e in self._recognize:

                if element.tag == e:

                    return (element.tag, element.attrib)

            break

        return None

################################################################################

class XMLBuilder:

    @staticmethod

    def _element(spec):

        element = ElementTree.Element(spec['tag'], spec.get('attrib', {}))

        for child in spec.get('children', []):

            element.append(XMLBuilder._element(child))

        return element

    @staticmethod

    def _etree(spec):

        return ElementTree.ElementTree(XMLBuilder._element(spec))

    @staticmethod

    def build(spec, encoding='utf-8'):

        fp = StringIO()

        XMLBuilder._etree(spec).write(fp, encoding)

        return fp.get_value()

# Copyright 2009 Shikhar Bhushan

#

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#

#    http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

import logging

from weakref import WeakValueDictionary

import content

from session import SessionCloseError

logger = logging.getLogger('ncclient.listeners')

################################################################################

# {session-id: SessionListener}

session_listeners = WeakValueDictionary

def session_listener_factory(session):

    try:

        return session_listeners[session]

    except KeyError:

        session_listeners[session] = SessionListener()

        return session_listeners[session]

class SessionListener:

    def __init__(self):

        # {message-id: RPC}

        self._rpc = WeakValueDictionary()

        # if the session gets closed by remote endpoint,

        # need to know if it is an error event or was requested through

        # a NETCONF operation i.e. CloseSession

        self._expecting_close = False

        # other recognized names 

        self._recognized = []

    def __str__(self):

        return 'SessionListener'

    def expect_close(self):

        self._expecting_close = True

    def register(self, id, op):

        self._id2rpc[id] = op

    ### Events

    def reply(self, raw):

        try:

            id = content.parse_message_root(raw)

            if id is None:

                pass

            elif id == 'notification':

                self._id2rpc[self._sub_id]._notify(raw)

            else:

                self._id2rpc[id]._response_cb(raw)

        except Exception as e:

            logger.warning(e)

    def error(self, err):

        if err is SessionCloseError:

            logger.debug('session closed by remote endpoint, expecting_close=%s' %

                         self._expecting_close)

            if not self._expecting_close:

                raise err

################################################################################

class HelloListener:

    def __str__(self):

        return 'HelloListener'

    def __init__(self, session):

        self._session = session

    ### Events

    def reply(self, data):

        try:

            id, capabilities = content.Hello.parse(data)

            logger.debug('HelloListener: session_id: %s; capabilities: %s', id, capabilities)

            self._session.initialize(id, capabilities)

        except Exception as e:

            self._session.initialize_error(e)

    def error(self, err):

        self._session.initialize_error(err)

################################################################################

class DebugListener:

    def __str__(self): return 'DebugListener'

    def reply(self, raw): logger.debug('DebugListener:reply:\n%s' % raw)

    def error(self, err): logger.debug('DebugListener:error:\n%s' % err)

# Copyright 2009 Shikhar Bhushan

#

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#

#    http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

from threading import Event, Lock

from uuid import uuid1

import content

from listeners import session_listener_factory

class RPC:

    metadata = {

        'tag': 'rpc',

        'xmlns': 'urn:ietf:params:xml:ns:netconf:base:1.0',

        }

    def __init__(self, session, async=False, parse=True):

        self._session = session

        self._async = async

        self._id = uuid1().urn

        listener = session_listener_factory(self._session)

        listener.register(self._id, self)

        session.add_listener(listener)

        self._reply = None

        self._reply_event = Event()

    def _response_cb(self, reply):

        self._reply = reply

        self._event.set()

    def _do_request(self, op):

        self._session.send(content.make_rpc(self._id, op))

        # content.make(RPC, attrs={'message-id': self._id}, children=(op,))

        if not self._async:

            self._reply_event.wait()

        return self._reply

    def request(self):

        raise NotImplementedError

    def wait_for_reply(self, timeout=None):

        self._reply_event.wait(timeout)

    @property

    def has_reply(self):

        return self._reply_event.isSet()

    @property

    def is_async(self):

        return self._async

    @property

    def reply(self):

        return self._reply

    @property

    def id(self):

        return self._id

    @property

    def session(self):

        return self._session

class RPCReply:

    class RPCError:

        pass

# Copyright 2009 Shikhar Bhushan

#

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#

#    http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

import logging

from threading import Thread, Event

from Queue import Queue

import content

from capabilities import Capabilities, CAPABILITIES

from error import ClientError

from listeners import HelloListener

from subject import Subject

logger = logging.getLogger('ncclient.session')

class SessionError(ClientError): pass

class Session(Thread, Subject):

    def __init__(self):

        Thread.__init__(self, name='session')

        Subject.__init__(self)

        self._client_capabilities = CAPABILITIES

        self._server_capabilities = None # yet

        self._id = None # session-id

        self._error = None

        self._init_event = Event()

        self._q = Queue()

        self._connected = False # to be set/cleared by subclass implementation

    def _post_connect(self):

        # start the subclass' main loop

        listener = HelloListener(self)

        self.add_listener(listener)

        self.start()

        # queue client's hello message for sending

        self.send(content.Hello.build(self._client_capabilities))

        # we expect server's hello message, wait for _init_event to be set

        self._init_event.wait()

        self.remove_listener(listener)

        # there may have been an error

        if self._error:

            self._close()

            raise self._error

    def hello(self, id, capabilities):

        self._id, self._capabilities = id, Capabilities(capabilities)

        self._init_event.set()

    def hello_error(self, err):

        self._error = err

        self._init_event.set()

    def send(self, message):

        logger.debug('queueing message: \n%s' % message)

        self._q.put(message)

    def connect(self):

        raise NotImplementedError

    def run(self):

        raise NotImplementedError

    def capabilities(self, whose='client'):

        if whose == 'client':

            return self._client_capabilities

        elif whose == 'server':

            return self._server_capabilities

    ### Properties

    @property

    def client_capabilities(self):

        return self._client_capabilities

    @property

    def server_capabilities(self):

        return self._server_capabilities

    @property

    def connected(self):

        return self._connected

    @property

    def id(self):

        return self._id

# Copyright 2009 Shikhar Bhushan

#

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#

#    http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

import logging

from cStringIO import StringIO

from os import SEEK_CUR

import socket

import paramiko

from session import Session, SessionError

logger = logging.getLogger('ncclient.ssh')

class SessionCloseError(SessionError):

    def __str__(self):

        return 'RECEIVED: %s | UNSENT: %s' % (self._in_buf, self._out_buf)

    def __init__(self, in_buf, out_buf=None):

        SessionError.__init__(self)

        self._in_buf, self._out_buf = in_buf, out_buf

class SSHSession(Session):

    BUF_SIZE = 4096

    MSG_DELIM = ']]>]]>'

    def __init__(self, load_known_hosts=True,

                 missing_host_key_policy=paramiko.RejectPolicy()):

        Session.__init__(self)

        self._client = paramiko.SSHClient()

        self._channel = None

        if load_known_hosts:

            self._client.load_system_host_keys()

        self._client.set_missing_host_key_policy(missing_host_key_policy)

        self._in_buf = StringIO()

        self._parsing_state = 0

        self._parsing_pos = 0

    def _close(self):

        self._channel.close()

        self._connected = False

    def _fresh_data(self):

        delim = SSHSession.MSG_DELIM

        n = len(delim) - 1

        state = self._parsing_state

        buf = self._in_buf

        buf.seek(self._parsing_pos)

        while True:

            x = buf.read(1)

            if not x: # done reading

                break

            elif x == delim[state]:

                state += 1

            else:

                continue

            # loop till last delim char expected, break if other char encountered

            for i in range(state, n):

                x = buf.read(1)

                if not x: # done reading

                    break

                if x==delim[i]: # what we expected

                    state += 1 # expect the next delim char

                else:

                    state = 0 # reset

                    break

            else: # if we didn't break out of above loop, full delim parsed

                till = buf.tell() - n

                buf.seek(0)

                msg = buf.read(till)

                self.dispatch('reply', msg)

                buf.seek(n+1, SEEK_CUR)

                rest = buf.read()

                buf = StringIO()

                buf.write(rest)

                buf.seek(0)

                state = 0

        self._in_buf = buf

        self._parsing_state = state

        self._parsing_pos = self._in_buf.tell()

    #def load_host_keys(self, filename):

    #    self._client.load_host_keys(filename)

    #

    #def set_missing_host_key_policy(self, policy):

    #    self._client.set_missing_host_key_policy(policy)

    #

    #def connect(self, hostname, port=830, username=None, password=None,

    #            key_filename=None, timeout=None, allow_agent=True,

    #            look_for_keys=True):

    #    self._client.connect(hostname, port=port, username=username,

    #                        password=password, key_filename=key_filename,

    #                        timeout=timeout, allow_agent=allow_agent,

    #                        look_for_keys=look_for_keys)    

    #    transport = self._client.get_transport()

    #    self._channel = transport.open_session()

    #    self._channel.invoke_subsystem('netconf')

    #    self._channel.set_name('netconf')

    #    self._connected = True

    #    self._post_connect()

    def connect(self, hostname, port=830, username=None, password=None,

                key_filename=None, timeout=None, allow_agent=True,

                look_for_keys=True):

        self._transport = paramiko.Transport()

    def run(self):

        chan = self._channel

        chan.setblocking(0)

        q = self._q

        try:

            while True:    

                if chan.closed:

                    raise SessionCloseError(self._in_buf.getvalue())         

                if chan.send_ready() and not q.empty():

                    data = q.get() + SSHSession.MSG_DELIM

                    while data:

                        n = chan.send(data)

                        if n <= 0:

                            raise SessionCloseError(self._in_buf.getvalue(), data)

                        data = data[n:]

                if chan.recv_ready():

                    data = chan.recv(SSHSession.BUF_SIZE)

                    if data:

                        self._in_buf.write(data)

                        self._fresh_data()

                    else:

                        raise SessionCloseError(self._in_buf.getvalue())

        except Exception as e:

            logger.debug('*** broke out of main loop ***')

            self.dispatch('error', e)

class MissingHostKeyPolicy(paramiko.MissingHostKeyPolicy):

    def __init__(self, cb):

        self._cb = cb

    def missing_host_key(self, client, hostname, key):

        if not self._cb(hostname, key):

            raise SSHError

# Copyright 2009 Shikhar Bhushan

#

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#

#    http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

from threading import Lock

import logging

logger = logging.getLogger('ncclient.subject')

class Subject:

    def __init__(self, listeners=[]):

        self._listeners = set(listeners)

        self._lock = Lock()

    def has_listener(self, listener):

        with self._lock:

            return (listener in self._listeners)

    def add_listener(self, listener):

        with self._lock:

            self._listeners.add(listener)

    def remove_listener(self, listener):

        with self._lock:

            self._listeners.discard(listener)

    def dispatch(self, event, *args, **kwds):

        # holding the lock while doing callbacks could lead to a deadlock

        # if one of the above methods is called

        with self._lock:

            listeners = list(self._listeners)

        for l in listeners:

            try:

                logger.debug('dispatching [%s] to [%s]' % (event, l))

                getattr(l, event)(*args, **kwds)

            except Exception as e:

                logger.warning(e)