openbsc/contrib/twisted_ipa.py

#!/usr/bin/python3
# -*- mode: python-mode; py-indent-tabs-mode: nil -*-
"""
/*
 * Copyright (C) 2016 sysmocom s.f.m.c. GmbH
 *
 * All Rights Reserved
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License along
 * with this program; if not, write to the Free Software Foundation, Inc.,
 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
 */
"""

from ipa import Ctrl, IPA
from twisted.internet.protocol import ReconnectingClientFactory
from twisted.internet import reactor
from twisted.protocols import basic
import argparse

class IPACommon(basic.Int16StringReceiver):
    """
    Generic IPA protocol handler: include some routines for simpler subprotocols.
    It's not intended as full implementation of all subprotocols, rather common ground and example code.
    """
    def dbg(self, line):
        """
        Debug print helper
        """
        if self.factory.debug:
            print(line)

    def osmo_CTRL(self, data):
        """
        OSMO CTRL protocol
        Placeholder, see corresponding derived class
        """
        pass

    def osmo_MGCP(self, data):
        """
        OSMO MGCP extension
        """
        self.dbg('OSMO MGCP received %s' % data)

    def osmo_LAC(self, data):
        """
        OSMO LAC extension
        """
        self.dbg('OSMO LAC received %s' % data)

    def osmo_SMSC(self, data):
        """
        OSMO SMSC extension
        """
        self.dbg('OSMO SMSC received %s' % data)

    def osmo_ORC(self, data):
        """
        OSMO ORC extension
        """
        self.dbg('OSMO ORC received %s' % data)

    def osmo_GSUP(self, data):
        """
        OSMO GSUP extension
        """
        self.dbg('OSMO GSUP received %s' % data)

    def osmo_OAP(self, data):
        """
        OSMO OAP extension
        """
        self.dbg('OSMO OAP received %s' % data)

    def osmo_UNKNOWN(self, data):
        """
        OSMO defaul extension handler
        """
        self.dbg('OSMO unknown extension received %s' % data)

    def handle_RSL(self, data, proto, extension):
        """
        RSL protocol handler
        """
        self.dbg('IPA RSL received message with extension %s' % extension)

    def handle_CCM(self, data, proto, msgt):
        """
        CCM (IPA Connection Management)
        Placeholder, see corresponding derived class
        """
        pass

    def handle_SCCP(self, data, proto, extension):
        """
        SCCP protocol handler
        """
        self.dbg('IPA SCCP received message with extension %s' % extension)

    def handle_OML(self, data, proto, extension):
        """
        OML protocol handler
        """
        self.dbg('IPA OML received message with extension %s' % extension)

    def handle_OSMO(self, data, proto, extension):
        """
        Dispatcher point for OSMO subprotocols based on extension name, lambda default should never happen
        """
        method = getattr(self, 'osmo_' + IPA().ext(extension), lambda: "extension dispatch failure")
        method(data)

    def handle_MGCP(self, data, proto, extension):
        """
        MGCP protocol handler
        """
        self.dbg('IPA MGCP received message with attribute %s' % extension)

    def handle_UNKNOWN(self, data, proto, extension):
        """
        Default protocol handler
        """
        self.dbg('IPA received message for %s (%s) protocol with attribute %s' % (IPA().proto(proto), proto, extension))

    def process_chunk(self, data):
        """
        Generic message dispatcher for IPA (sub)protocols based on protocol name, lambda default should never happen
        """
        (_, proto, extension, content) = IPA().del_header(data)
        if content is not None:
            self.dbg('IPA received %s::%s [%d/%d] %s' % (IPA().proto(proto), IPA().ext_name(proto, extension), len(data), len(content), content))
            method = getattr(self, 'handle_' + IPA().proto(proto), lambda: "protocol dispatch failure")
            method(content, proto, extension)

    def dataReceived(self, data):
        """
        Override for dataReceived from Int16StringReceiver because of inherently incompatible interpretation of length
        If default handler is used than we would always get off-by-1 error (Int16StringReceiver use equivalent of l + 2)
        """
        if len(data):
            (head, tail) = IPA().split_combined(data)
            self.process_chunk(head)
            self.dataReceived(tail)

    def connectionMade(self):
        """
        We have to resetDelay() here to drop internal state to default values to make reconnection logic work
        Make sure to call this via super() if overriding to keep reconnection logic intact
        """
        self.dbg('IPA connection made!')
        self.factory.resetDelay()


class CCM(IPACommon):
    """
    Implementation of CCM protocol for IPA multiplex
    """
    def ack(self):
        self.transport.write(IPA().id_ack())

    def ping(self):
        self.transport.write(IPA().ping())

    def pong(self):
        self.transport.write(IPA().pong())

    def handle_CCM(self, data, proto, msgt):
        """
        CCM (IPA Connection Management)
        Only basic logic necessary for tests is implemented (ping-pong, id ack etc)
        """
        if msgt == IPA.MSGT['ID_GET']:
            self.transport.getHandle().sendall(IPA().id_resp(self.factory.ccm_id))
            # if we call
            # self.transport.write(IPA().id_resp(self.factory.test_id))
            # instead, than we would have to also call
            # reactor.callLater(1, self.ack)
            # instead of self.ack()
            # otherwise the writes will be glued together - hence the necessity for ugly hack with 1s timeout
            # Note: this still might work depending on the IPA implementation details on the other side
            self.ack()
            # schedule PING in 4s
            reactor.callLater(4, self.ping)
        if msgt == IPA.MSGT['PING']:
            self.pong()


class CTRL(IPACommon):
    """
    Implementation of Osmocom control protocol for IPA multiplex
    """
    def ctrl_SET(self, data, op_id, v):
        """
        Handle CTRL SET command
        """
        self.dbg('CTRL SET [%s] %s' % (op_id, v))

    def ctrl_SET_REPLY(self, data, op_id, v):
        """
        Handle CTRL SET reply
        """
        self.dbg('CTRL SET REPLY [%s] %s' % (op_id, v))

    def ctrl_GET(self, data, op_id, v):
        """
        Handle CTRL GET command
        """
        self.dbg('CTRL GET [%s] %s' % (op_id, v))

    def ctrl_GET_REPLY(self, data, op_id, v):
        """
        Handle CTRL GET reply
        """
        self.dbg('CTRL GET REPLY [%s] %s' % (op_id, v))

    def ctrl_TRAP(self, data, op_id, v):
        """
        Handle CTRL TRAP command
        """
        self.dbg('CTRL TRAP [%s] %s' % (op_id, v))

    def ctrl_ERROR(self, data, op_id, v):
        """
        Handle CTRL ERROR reply
        """
        self.dbg('CTRL ERROR [%s] %s' % (op_id, v))

    def osmo_CTRL(self, data):
        """
        OSMO CTRL message dispatcher, lambda default should never happen
        For basic tests only, appropriate handling routines should be replaced: see CtrlServer for example
        """
        self.dbg('OSMO CTRL received %s::%s' % Ctrl().parse(data.decode('utf-8')))
        (cmd, op_id, v) = data.decode('utf-8').split(' ', 2)
        method = getattr(self, 'ctrl_' + cmd, lambda: "CTRL unknown command")
        method(data, op_id, v)


class IPAServer(CCM):
    """
    Test implementation of IPA server
    Demonstrate CCM opearation by overriding necessary bits from CCM
    """
    def connectionMade(self):
        """
        Keep reconnection logic working by calling routine from CCM
        Initiate CCM upon connection
        """
        print('IPA server connection made!')
        super(IPAServer, self).connectionMade()
        self.transport.write(IPA().id_get())


class CtrlServer(CTRL):
    """
    Test implementation of CTRL server
    Demonstarte CTRL handling by overriding simpler routines from CTRL
    """
    def connectionMade(self):
        """
        Keep reconnection logic working by calling routine from CTRL
        Send TRAP upon connection
        Note: we can't use sendString() because of it's incompatibility with IPA interpretation of length prefix
        """
        print('CTRL server connection made!')
        super(CtrlServer, self).connectionMade()
        self.transport.write(Ctrl().trap('LOL', 'what'))
        self.transport.write(Ctrl().trap('rulez', 'XXX'))

    def reply(self, r):
        self.transport.write(Ctrl().add_header(r))

    def ctrl_SET(self, data, op_id, v):
        """
        CTRL SET command: always succeed
        """
        print('SET [%s] %s' % (op_id, v))
        self.reply('SET_REPLY %s %s' % (op_id, v))

    def ctrl_GET(self, data, op_id, v):
        """
        CTRL GET command: always fail
        """
        print('GET [%s] %s' % (op_id, v))
        self.reply('ERROR %s No variable found' % op_id)


class IPAFactory(ReconnectingClientFactory):
    """
    Generic IPA Client Factory which can be used to store state for various subprotocols and manage connections
    Note: so far we do not really need separate Factory for acting as a server due to protocol simplicity
    """
    protocol = IPACommon
    debug = False
    ccm_id = IPA().identity(unit=b'1515/0/1', mac=b'b0:0b:fa:ce:de:ad:be:ef', utype=b'sysmoBTS', name=b'StingRay', location=b'hell', sw=IPA.version.encode('utf-8'))

    def __init__(self, proto=None, debug=False, ccm_id=None):
        if proto:
            self.protocol = proto
        if debug:
            self.debug = debug
        if ccm_id:
            self.ccm_id = ccm_id

    def clientConnectionFailed(self, connector, reason):
        """
        Only necessary for as debugging aid - if we can somehow set parent's class noisy attribute then we can omit this method
        """
        if self.debug:
            print('IPAFactory connection failed:', reason.getErrorMessage())
        ReconnectingClientFactory.clientConnectionFailed(self, connector, reason)

    def clientConnectionLost(self, connector, reason):
        """
        Only necessary for as debugging aid - if we can somehow set parent's class noisy attribute then we can omit this method
        """
        if self.debug:
            print('IPAFactory connection lost:', reason.getErrorMessage())
        ReconnectingClientFactory.clientConnectionLost(self, connector, reason)


if __name__ == '__main__':
    p = argparse.ArgumentParser("Twisted IPA (module v%s) app" % IPA.version)
    p.add_argument('-v', '--version', action='version', version='%(prog)s v0.3')
    p.add_argument('-p', '--port', type=int, default=4250, help="Port to use for CTRL interface")
    p.add_argument('-d', '--host', default='localhost', help="Adress to use for CTRL interface")
    cs = p.add_mutually_exclusive_group()
    cs.add_argument("-c", "--client", action='store_true', help="asume client role")
    cs.add_argument("-s", "--server", action='store_true', help="asume server role")
    ic = p.add_mutually_exclusive_group()
    ic.add_argument("--ipa", action='store_true', help="use IPA protocol")
    ic.add_argument("--ctrl", action='store_true', help="use CTRL protocol")
    args = p.parse_args()
    test = False
    if args.ctrl:
        if args.client:
            # Start osmo-bsc to receive TRAP messages when osmo-bts-* connects to it
            print('CTRL client, connecting to %s:%d' % (args.host, args.port))
            reactor.connectTCP(args.host, args.port, IPAFactory(CTRL, debug=True))
            test = True
        if args.server:
            # Use bsc_control.py to issue set/get commands
            print('CTRL server, listening on port %d' % args.port)
            reactor.listenTCP(args.port, IPAFactory(CtrlServer, debug=True))
            test = True
    if args.ipa:
        if args.client:
            # Start osmo-nitb which would initiate A-bis/IP session
            print('IPA client, connecting to %s ports %d and %d' % (args.host, IPA.TCP_PORT_OML, IPA.TCP_PORT_RSL))
            reactor.connectTCP(args.host, IPA.TCP_PORT_OML, IPAFactory(CCM, debug=True))
            reactor.connectTCP(args.host, IPA.TCP_PORT_RSL, IPAFactory(CCM, debug=True))
            test = True
        if args.server:
            # Start osmo-bts-* which would attempt to connect to us
            print('IPA server, listening on ports %d and %d' % (IPA.TCP_PORT_OML, IPA.TCP_PORT_RSL))
            reactor.listenTCP(IPA.TCP_PORT_RSL, IPAFactory(IPAServer, debug=True))
            reactor.listenTCP(IPA.TCP_PORT_OML, IPAFactory(IPAServer, debug=True))
            test = True
    if test:
        reactor.run()
    else:
        print("Please specify which protocol in which role you'd like to test.")