forwarder.py

import os
import sys
import logging
import signal
from time import sleep
from threading import Thread
from wsgiref.simple_server import make_server
from pyramid.config import Configurator
from pyramid.response import Response
from pyramid.paster import setup_logging
from opensipkd.tcp.connection import (
    ConnectionManager as BaseConnectionManager,
    join_ip_port,
    )
from opensipkd.string import (
    exception_message,
    dict_to_str,
    )
from opensipkd.tcp.server import (
    Server as BaseServer,
    RequestHandler as BaseRequestHandler,
    )
from opensipkd.tcp.client import Client as BaseClient
from ..read_conf import (
    read_conf,
    ip_conf,
    get_web_port,
    listen_ports,
    allowed_ips,
    get_conf,
    )


def debug(s):
    msg = 'DEBUG {}'.format(s)
    print(msg)


def get_log():
    return logs[0]


def log_info(s):
    log = get_log()
    log.info(s)


def log_error(s):
    log = get_log()
    log.error(s)


def iso_to_dict(iso):
    data = iso.get_values()
    return dict_to_str(data)


def to_str(s):
    if sys.version_info.major > 2:
        return s
    if isinstance(s, unicode):
        return str(s)


class Log:
    def log_message(self, msg):
        return '{ip} {name} {mem_id} {msg}'.format(
            ip=self.conf['ip'], name=self.conf['name'], mem_id=id(self),
            msg=msg)

    def log_info(self, msg):
        msg = self.log_message(msg)
        log_info(msg)

    def log_error(self, msg):
        msg = self.log_message(msg)
        log_error(msg)

    def log_unknown(self):
        msg = exception_message()
        self.log_error(msg)

    def log_decode(self, iso):
        data = iso_to_dict(iso)
        self.log_info('Decode MTI {} Data {}'.format(iso.getMTI(), data))

    def log_encode(self, iso):
        data = iso_to_dict(iso)
        self.log_info('Encode MTI {} Data {}'.format(iso.getMTI(), data))


parser_threads = []


class CommonConnection(Log):
    def log_receive_raw(self, raw):
        raw = to_str(raw)
        self.log_info('Receive {}'.format([raw]))

    def log_send(self, raw):
        if isinstance(raw, bytes):
            raw = raw.decode('utf-8')
        raw = to_str(raw)
        self.log_info('Send {}'.format([raw]))

    def log_close(self, reason):
        msg = 'close connection because {}'.format(reason)
        self.log_info(msg)

    def log_timeout(self):
        self.log_close('timeout')

    def log_raw_to_iso(self, raw):
        self.log_info('Raw to ISO8583 {}'.format([raw]))

    def log_iso_to_raw(self, raw):
        raw = to_str(raw)
        self.log_info('ISO8583 to raw {}'.format([raw]))

    def process(self, raw):
        parser = Parser(self, raw)
        thread = create_thread(parser.run)
        parser_threads.append((parser, thread))
        thread.start()


def create_thread(func):
    thread = Thread(target=func)
    # Exit the server thread when the main thread terminates
    thread.daemon = True
    return thread


###################
# ISO 8583 Server #
###################
class Server(BaseServer):
    # Override
    def verify_request(self, request, client_address):
        client_ip = client_address[0]
        log = get_log()
        if client_ip in allowed_ips:
            log.info('{} allowed'.format(client_ip))
            return True
        log.error('{} denied'.format(client_ip))


class RequestHandler(BaseRequestHandler, CommonConnection):
    def handle(self):
        ip = self.client_address[0]
        port = self.server.server_address[1]
        self.conf = get_conf(ip, port)
        self.running = False 
        conn_mgr.add(self)
        BaseRequestHandler.handle(self)

    def on_receive_raw(self, raw):
        self.log_receive_raw(raw)
        BaseRequestHandler.on_receive_raw(self, raw)

    # Override BaseRequestHandler.process()
    def process(self, raw):
        CommonConnection.process(self, raw)

    def close_because_timeout(self):
        self.log_timeout()
        BaseRequestHandler.close_because_timeout(self)

    def on_socket_error(self, err):
        self.log_error(err)
        BaseRequestHandler.on_socket_error(self, err)

    def raw_for_send(self, raw):
        self.log_iso_to_raw(raw)
        raw = BaseRequestHandler.raw_for_send(self, raw)
        self.log_send(raw)
        return raw

    def run(self):
        try:
            BaseRequestHandler.run(self)
        except:
            self.log_unknown()


servers = {}


def start_servers():
    for listen_port in listen_ports:
        listen_address = ('0.0.0.0', listen_port)
        log_info('ISO8583 server listen at {}:{}'.format(*listen_address))
        server = Server(listen_address, RequestHandler)
        thread = create_thread(server.serve_forever)
        servers[listen_port] = (server, thread) 
        thread.start()


def stop_servers(reason):
    for listen_port in listen_ports:
        server, thread = servers[listen_port]
        server.shutdown()
        thread.join()
        sleep(1)


###################
# ISO 8583 Client #
###################
class Client(BaseClient, CommonConnection):
    def connect(self):
        ip, port = self.address
        self.log_info('connect to port {}'.format(port))
        BaseClient.connect(self)

    def on_receive_raw(self, raw):
        self.log_receive_raw(raw)
        BaseClient.on_receive_raw(self, raw)

    # Override BaseClient.process()
    def process(self, raw):
        CommonConnection.process(self, raw)

    def close_because_timeout(self):
        self.log_timeout()
        BaseClient.close_because_timeout(self)

    def on_refused(self, err):
        ip, port = self.address
        self.log_error('port {} {}'.format(port, err))
        BaseClient.on_refused(self, err)

    def on_socket_error(self, err):
        self.log_error(err)
        BaseClient.on_socket_error(self, err)

    def raw_for_send(self, raw):
        self.log_iso_to_raw(raw)
        raw = BaseClient.raw_for_send(self, raw)
        self.log_send(raw)
        return raw

    def run(self):
        try:
            BaseClient.run(self)
        except:
            self.log_unknown()


def start_client(conf):
    client = Client(conf)
    thread = create_thread(client.run)
    ip_port = join_ip_port(conf['ip'], conf['port'])
    clients[ip_port] = (client, thread)
    thread.start()
    conn_mgr.add(client)


clients = {}


def stop_connections(reason):
    for ip_port, connection in conn_mgr:
        connection.log_close(reason)
        connection.close()
        sleep(1)
    for ip_port in clients:
        client, thread = clients[ip_port]
        thread.join()
 

#######################
# Raw ISO 8583 parser #
#######################
class Parser(Log):
    def __init__(self, connection, raw):
        self.connection = connection
        self.raw = raw
        self.conf = connection.conf
        self.conn_id = id(connection)
        self.parser_id = id(self)
        self.running = True

    # Override
    def log_message(self, msg):
        return '{ip} {name} {conn_id} -> {parser_id} {msg}'.format(
            ip=self.conf['ip'], name=self.conf['name'], conn_id=self.conn_id,
            parser_id=self.parser_id, msg=msg)

    def run(self):
        from_iso = self.connection.job.raw_to_iso(self.raw)
        self.log_decode(from_iso)
        iso = self.connection.job.process(from_iso)
        if iso:
            self.log_encode(iso)
            raw = iso.getRawIso()
            self.connection.send(raw)
        self.running = False


######################
# Connection Manager #
######################
class ConnectionManager(BaseConnectionManager):
    def close_old_connection(self, old_conn):
        old_conn.log_close('new connection found')
        BaseConnectionManager.close_old_connection(self, old_conn)


conn_mgr = ConnectionManager()
    
#######
# Web #
#######
def log_web_info(s):
    log = get_log()
    msg = 'Web server {}'.format(s)
    log.info(msg)


def hello_world(request):
    return Response('Hello World!')


web_server = {}


def start_web_server():
    port = get_web_port()
    with Configurator() as config:
        config.add_route('hello', '/')
        config.add_view(hello_world, route_name='hello')
        app = config.make_wsgi_app()
        web_server['listener'] = server = make_server('0.0.0.0', port, app)
    web_server['thread'] = Thread(target=server.serve_forever) 
    web_server['thread'].start()
    ip_port = web_server['listener'].server_address
    log_web_info('listen at {}:{}'.format(*ip_port))


def stop_web_server(reason):
    msg = 'stop because {}'.format(reason)
    log_web_info(msg)
    # shutdown() ini kadang tidak segera mengakhiri web server. Akan cepat
    # berakhir bila ada client yang akses.
    web_server['listener'].shutdown()
    web_server['thread'].join()


MSG_KILL_BY_SIGNAL = 'kill by signal {}'
MSG_KILL_BY_KEYBOARD = 'kill by keyboard interrupt'


def out(sig=None, func=None):
    if running:
        del running[0]  # Akhiri loop utama
    if sig:
        reason = MSG_KILL_BY_SIGNAL.format(sig)
    else:
        reason = MSG_KILL_BY_KEYBOARD 
    stop_servers(reason)
    stop_connections(reason)
    stop_web_server(reason)


def usage(argv):
    cmd = os.path.basename(argv[0])
    print('usage: %s <config_uri>\n'
          '(example: "%s test.ini")' % (cmd, cmd))
    sys.exit(1)


def check_connection():
    for ip_port in ip_conf:
        if ip_port in conn_mgr:
            index = -1
            while True:
                index += 1
                if not conn_mgr[index:]:
                    break
                this_ip_port, conn = conn_mgr[index]
                if this_ip_port != ip_port:
                    continue
                if conn.running:
                    continue
                conn_mgr.remove(index)
                break
            continue
        cfg = ip_conf[ip_port]
        if cfg['listen']:
            continue
        start_client(cfg)
        sleep(5)


def check_job():
    for ip_port, connection in conn_mgr:
        if not connection.running:
            continue
        if not connection.is_connected():
            continue
        iso = connection.job.get_iso()
        if not iso:
            continue
        connection.log_encode(iso)
        raw = iso.getRawIso()
        connection.send(raw)


def check_parser():
    i = -1
    while True:
        i += 1
        if not parser_threads[i:]:
            break
        parser, thread = parser_threads[i]
        if not parser.running:
            thread.join()
            del parser_threads[i]
            i -= 1 


logs = []
running = []


def main(argv=sys.argv):
    if len(argv) != 2:
        usage(argv)
    config_uri = argv[1]
    setup_logging(config_uri)
    read_conf(config_uri)
    log = logging.getLogger(__file__)
    logs.append(log)
    running.append(True)
    start_web_server()
    start_servers()
    # Antisipasi kill
    signal.signal(signal.SIGTERM, out)
    try:
        while running:
            check_connection()
            check_job()
            check_parser()
            sleep(5)
    except KeyboardInterrupt:
        out()