forwarder.py

import os
import sys
import logging
import signal
from time import (
    sleep,
    time,
    )
from threading import Thread
from configparser import ConfigParser
from waitress.server import create_server
from pyramid.config import Configurator
from pyramid.paster import setup_logging
from pyramid_rpc.jsonrpc import jsonrpc_method
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 opensipkd.jsonrpc.exc import (
    JsonRpcInvalidParams,
    JsonRpcBankNotFound,
    JsonRpcBillerNetwork,
    )
from ..read_conf import (
    read_conf,
    ip_conf,
    web as web_conf,
    listen_ports,
    allowed_ips,
    get_conf,
    name_conf,
    )


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 log_debug(s):
    log = get_log()
    log.debug(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_debug(self, msg):
        msg = self.log_message(msg)
        log_debug(msg)

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

    def log_ack(self, iso):
        if iso.ack_message:
            self.log_error(iso.ack_message)

    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):
        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):
        self.log_raw_to_iso(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
        self.created_time = time()
        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
    def process(self, raw):
        CommonConnection.process(self, raw)

    # Override
    def set_timeout(self):
        if self.conf['echo']:
            BaseRequestHandler.set_timeout(self)

    # Override
    def is_timeout(self):
        if not self.conf['echo']:
            return False
        return BaseRequestHandler.is_timeout(self)

    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)
        return BaseRequestHandler.raw_for_send(self, raw)

    def just_send(self, raw):
        self.log_send(raw)
        BaseRequestHandler.just_send(self, 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]
        log_debug('ISO8583 server shutdown')
        server.shutdown()
        log_debug('ISO8583 server thread join')
        thread.join()
        log_debug('ISO8583 server thread joined')
        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)
        return BaseClient.raw_for_send(self, raw)

    def just_send(self, raw):
        self.log_send(raw)
        BaseClient.just_send(self, 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]
        msg = 'Thread {} join'.format(ip_port)
        log_debug(msg)
        thread.join()
        msg = 'Thread {} joined'.format(ip_port)
        log_debug(msg)


#######################
# 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)
        try:
            iso = self.connection.job.process(from_iso)
        except:
            self.log_unknown()
            self.running = False
            return
        if iso:
            self.log_ack(iso)
            self.log_encode(iso)
            raw = iso.getRawIso()
            self.connection.send(raw)
        else:  # dapat response
            ip_port = join_ip_port(self.conf['ip'], self.conf['port'])
            if ip_port in web_process:
                stan_list = web_process[ip_port]
                stan = from_iso.get_stan()
                if stan in stan_list:
                    i = stan_list.index(stan)
                    del stan_list[i]
                    append_web_response(ip_port, stan, from_iso)
        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_msg(s):
    return 'Web server {}'.format(s)


def log_web_info(s):
    msg = log_web_msg(s)
    log = get_log()
    log.info(msg)


def log_web_error(s):
    msg = log_web_msg(s)
    log = get_log()
    log.error(msg)


def log_web_debug(s):
    msg = log_web_msg(s)
    log = get_log()
    log.debug(msg)


def conn_by_name(name):
    conf = name_conf[name]
    found_conn = None
    for ip_port, conn in conn_mgr:
        ip, port = ip_port.split(':')
        if conf['ip'] != ip:
            continue
        port = int(port)
        if conf['port'] != port:
            continue
        found_conn = conn
    if not found_conn:
        raise JsonRpcBankNotFound()
    if not found_conn.running:
        raise JsonRpcBankNotFound(message='Disconnected')
    return found_conn


# Daftar job dari web request, berisi iso request
# key: ip:port, value: list of iso
web_request = {}

# Daftar job yang sedang diproses, yaitu menunggu iso response
# key: ip:port, value: list of stan (bit 11)
web_process = {}

# Daftar job yang sudah selesai, berisi iso response
# key: ip:port, value: dict of (key: stan, value: iso)
web_response = {}


def append_web_process(ip_port, iso):
    stan = iso.get_stan()
    if ip_port in web_process:
        web_process[ip_port].append(stan)
    else:
        web_process[ip_port] = [stan]


def append_web_response(ip_port, stan, iso):
    if ip_port in web_response:
        web_response[ip_port][stan] = iso
    else:
        web_response[ip_port] = {stan: iso}


def web_job(conn, iso):
    ip_port = join_ip_port(conn.conf['ip'], conn.conf['port'])
    if ip_port in web_request:
        web_request[ip_port].append(iso)
    else:
        web_request[ip_port] = [iso]
    stan = iso.get_stan()
    awal = time()
    while True:
        sleep(0.001)
        if time() - awal > 5:
            raise JsonRpcBillerNetwork(message='Timeout')
        if ip_port not in web_response:
            continue
        result = web_response[ip_port]
        if stan in result:
            iso = result[stan]
            del result[stan]
            data = iso_to_dict(iso)
            return dict(code=0, message='OK', data=data)


def validate_rpc(p):
    if 'host' not in p:
        raise JsonRpcInvalidParams()
    return conn_by_name(p['host'])


def log_web_receive(request, method, p, flow='Receive'):
    msg = '{} {} {} {}'.format(request.client_addr, flow, method, p)
    log_web_info(msg)


def log_web_send(request, method, p):
    log_web_receive(request, method, p, 'Send')


@jsonrpc_method(endpoint='rpc')
def echo(request, p):
    log_web_receive(request, 'echo', p)
    conn = validate_rpc(p)
    iso = conn.job.echo_request()
    r = web_job(conn, iso)
    log_web_send(request, 'echo', r)
    return r


@jsonrpc_method(endpoint='rpc')
def inquiry(request, p):
    log_web_receive(request, 'inquiry', p)
    conn = validate_rpc(p)
    iso = conn.job.inquiry(p)
    r = web_job(conn, iso)
    log_web_send(request, 'inquiry', r)
    return r


@jsonrpc_method(endpoint='rpc')
def payment(request, p):
    log_web_receive(request, 'payment', p)
    conn = validate_rpc(p)
    iso = conn.job.payment(p)
    r = web_job(conn, iso)
    log_web_send(request, 'payment', r)
    return r


@jsonrpc_method(endpoint='rpc')
def reversal(request, p):
    log_web_receive(request, 'reversal', p)
    conn = validate_rpc(p)
    iso = conn.job.reversal(p)
    r = web_job(conn, iso)
    log_web_send(request, 'reversal', r)
    return r


web_server = {}


def start_web_server():
    port = web_conf.get('port')
    if not port:
        return
    thread_count = web_conf.get('threads')
    host = '0.0.0.0'
    with Configurator() as config:
        config.include('pyramid_tm')
        config.include('pyramid_rpc.jsonrpc')
        config.add_jsonrpc_endpoint('rpc', '/rpc')
        config.scan(__name__)
        app = config.make_wsgi_app()
    web_server['listener'] = server = create_server(
        app, host=host, port=port, threads=thread_count)
    web_server['thread'] = create_thread(server.run)
    web_server['thread'].start()
    log_web_info('listen at {}:{} with {} workers'.format(
        host, port, thread_count))


def stop_web_server(reason):
    if 'listener' not in web_server:
        return
    msg = 'stop because {}'.format(reason)
    log_web_info(msg)
    srv = web_server['listener']
    srv.close()
    while srv._map:
        triggers = list(srv._map.values())
        for trigger in triggers:
            trigger.handle_close()
        srv.maintenance(0)
        srv.task_dispatcher.shutdown()
    log_web_debug('thread join')
    web_server['thread'].join()
    log_web_debug('thread joined')


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_if_old(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 iso:
            iso_list = [iso]
        else:
            if ip_port not in web_request:
                continue
            web_iso_list = web_request[ip_port]
            if not web_iso_list:
                continue
            iso_list = []
            while web_iso_list:
                iso = web_iso_list[0]
                del web_iso_list[0]
                append_web_process(ip_port, iso)
                iso_list.append(iso)
        with_headers = []
        for iso in iso_list:
            connection.log_encode(iso)
            raw = iso.getRawIso()
            with_header = connection.raw_for_send(raw)
            with_headers.append(with_header)
        raw = b''.join(with_headers)
        connection.just_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


def set_log(conf_file):
    conf = ConfigParser()
    conf.read(conf_file)
    log = logging.getLogger(__file__)
    log.setLevel(conf.get('logger_root', 'level'))
    logs.append(log)


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)
    set_log(config_uri)
    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(0.5)
    except KeyboardInterrupt:
        out()