diff --git a/openbsc/contrib/twisted_ipa.py b/openbsc/contrib/twisted_ipa.py new file mode 100755 index 000000000..8d0ef9fff --- /dev/null +++ b/openbsc/contrib/twisted_ipa.py @@ -0,0 +1,373 @@ +#!/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.")