#!/usr/bin/env python3 # -*- mode: python-mode; py-indent-tabs-mode: nil -*- """ /* * Copyright (C) 2019 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. */ """ __version__ = "0.0.2" # bump this on every non-trivial change from functools import partial import configparser, argparse, time, os, asyncio, aiohttp from osmopy.trap_helper import make_params, gen_hash, log_init, comm_proc from osmopy.osmo_ipa import Ctrl def log_bsc_time(l, rq, task, ts, bsc, msg, *args, **kwargs): """ Logging contextual wrapper. FIXME: remove task parameter once we bump requirements to Python 3.7+ """ # FIXME: following function is deprecated and will be removed in Python 3.9 # Use the asyncio.all_tasks() function instead when available (Python 3.7+). num_tasks = len(task.all_tasks()) num_req = len(rq) delta = time.perf_counter() - ts if delta < 1: l('[%d/%d] BSC %s: ' + msg, num_req, num_tasks, bsc, *args, **kwargs) else: l('[%d/%d] BSC %s, %.2f sec: ' + msg, num_req, num_tasks, bsc, time.perf_counter() - ts, *args, **kwargs) def check_h_val(ctrl, h, v, t, exp): """ Check for header inconsistencies. """ if v != exp: ctrl.log.error('Unexpected %s value %x (instead of %x) in |%s| header', t, v, exp, h.hex()) def get_ctrl_len(ctrl, header): """ Obtain expected message length. """ (dlen, p, e, _) = ctrl.del_header(header) check_h_val(ctrl, header, p, "protocol", ctrl.PROTO['OSMO']) check_h_val(ctrl, header, e, "extension", ctrl.EXT['CTRL']) return dlen - 1 class Proxy(Ctrl): """ Wrapper class to implement per-type message dispatch and keep BSC <-> http Task mapping. N. B: keep async/await semantics out of it. """ def __init__(self, log): super().__init__() self.req = {} self.log = log self.conf = configparser.ConfigParser(interpolation = None) self.conf.read(self.config_file) self.timeout = self.conf['main'].getint('timeout', 30) self.location = self.conf['main'].get('location') self.ctrl_addr = self.conf['main'].get('addr_ctrl', 'localhost') self.ctrl_port = self.conf['main'].getint('port_ctrl', 4250) self.concurrency = self.conf['main'].getint('num_max_conn', 5) # FIXME: use timeout parameter when available (aiohttp version 3.3) as follows #self.http_client = aiohttp.ClientSession(connector = aiohttp.TCPConnector(limit = self.concurrency), timeout = self.timeout) self.http_client = aiohttp.ClientSession(connector = aiohttp.TCPConnector(limit = self.concurrency)) def dispatch(self, w, data): """ Basic dispatcher: the expected entry point for CTRL messages. """ (cmd, _, v) = data.decode('utf-8').split(' ', 2) method = getattr(self, cmd, lambda *_: self.log.info('CTRL %s is unhandled by dispatch: ignored.', cmd)) method(w, v.split()) def ERROR(self, _, k): """ Handle CTRL ERROR messages. """ self.log_ignore('ERROR', k) def SET_REPLY(self, _, k): """ Handle CTRL SET_REPLY messages. """ self.log_ignore('SET_REPLY', k) def TRAP(self, w, k): """ Handle incoming TRAPs. """ p = k[0].split('.') if p[-1] == 'location-state': self.handle_locationstate(w, p[1], p[3], p[5], k[1]) else: self.log_ignore('TRAP', k[0]) def handle_locationstate(self, w, net, bsc, bts, data): """ Handle location-state TRAP: parse trap content, build HTTP request and setup async handlers. """ ts = time.perf_counter() self.cleanup_task(bsc) params = make_params(bsc, data) params['h'] = gen_hash(params, self.conf['main'].get('secret_key')) # FIXME: use asyncio.create_task() when available (Python 3.7+). t = asyncio.ensure_future(self.http_client.post(self.location, data = params)) log_bsc_time(self.log.info, self.req, t, ts, bsc, 'location-state@%s => %s', params['time_stamp'], data) t.add_done_callback(partial(self.reply_callback, w, bsc, ts)) self.req[bsc] = (t, ts) log_bsc_time(self.log.info, self.req, t, ts, bsc, 'request added (net %s, BTS %s)', net, bts) def cleanup_task(self, bsc): """ It's ok to cancel() task which is done() but if either of the checks above fires it means that Proxy() is in inconsistent state which should never happen as long as we keep async/await semantics out of it. """ if bsc in self.req: (task, ts) = self.req[bsc] log_bsc = partial(log_bsc_time, self.log.error, self.req, task, ts, bsc) if task.done(): log_bsc('task is done but not removed') if task.cancelled(): log_bsc('task is cancelled without update') task.cancel() def log_ignore(self, kind, m): """ Log ignored CTRL message. """ self.log.error('Ignoring CTRL %s: %s', kind, ' '.join(m) if type(m) is list else m) def reply_callback(self, w, bsc, ts, task): """ Process per-BSC response status and prepare async handler if necessary. We don't have to delete cancel()ed task from self.req explicitly because it will be replaced by new one in handle_locationstate() """ log_bsc = partial(log_bsc_time, self.log.info, self.req, task, ts, bsc) if task.cancelled(): log_bsc('request cancelled') else: exp = task.exception() if exp: log_bsc('exception %s triggered', repr(exp)) else: resp = task.result() if resp.status != 200: log_bsc('unexpected HTTP response %d', resp.status) else: log_bsc('request completed') # FIXME: use asyncio.create_task() when available (Python 3.7+). asyncio.ensure_future(recv_response(self.log, w, bsc, resp.json())) del self.req[bsc] async def recv_response(log, w, bsc, resp): """ Process json response asynchronously. """ js = await resp if js.get('error'): log.info('BSC %s response error: %s', bsc, repr(js.get('error'))) else: comm_proc(js.get('commands'), bsc, w.write, log) await w.drain() # Trigger Writer's flow control async def recon_reader(proxy, reader, num_bytes): """ Read requested amount of bytes, reconnect if necessary. """ try: return await reader.readexactly(num_bytes) except asyncio.IncompleteReadError: proxy.log.info('Failed to read %d bytes reconnecting to %s:%d...', num_bytes, proxy.ctrl_addr, proxy.ctrl_port) raise async def ctrl_client(proxy, rd, wr): """ Read CTRL stream and handle selected messages. """ while True: header = await recon_reader(proxy, rd, 4) data = await recon_reader(proxy, rd, get_ctrl_len(proxy, header)) proxy.dispatch(wr, data) async def conn_client(proxy): """ (Re)establish connection with CTRL server and pass Reader/Writer to CTRL handler. """ while True: try: reader, writer = await asyncio.open_connection(proxy.ctrl_addr, proxy.ctrl_port) proxy.log.info('Connected to %s:%d', proxy.ctrl_addr, proxy.ctrl_port) await ctrl_client(proxy, reader, writer) except OSError as e: proxy.log.info('%s: %d seconds delayed retrying...', e, proxy.timeout) await asyncio.sleep(proxy.timeout) except asyncio.IncompleteReadError: pass proxy.log.info('Reconnecting...') if __name__ == '__main__': a = argparse.ArgumentParser(description = 'Proxy between given GCI service and Osmocom CTRL protocol.') a.add_argument('-v', '--version', action = 'version', version = ("%(prog)s v" + __version__)) a.add_argument('-d', '--debug', action = 'store_true', help = "Enable debug log") a.add_argument('-c', '--config-file', required = True, help = "Path to mandatory config file (in INI format).") P = Proxy(log_init('TRAP2CGI', a.parse_args(namespace=Proxy).debug)) P.log.info('CGI proxy v%s starting with PID %d:', __version__, os.getpid()) P.log.info('Destination %s (concurrency %d)', P.location, P.concurrency) P.log.info('Connecting to TRAP source %s:%d...', P.ctrl_addr, P.ctrl_port) loop = asyncio.get_event_loop() loop.run_until_complete(conn_client(P)) # FIXME: use loop.run() function instead when available (Python 3.7+).