forwarder.py

import os
import sys
import logging
import signal
from time import (
    sleep,
    time,
    )
from threading import Thread
from wsgiref.simple_server import make_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,
    get_web_port,
    listen_ports,
    allowed_ips,
    get_conf,
    name_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()
    print("DEBUG data", data)
    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)
        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 conn_by_name(name):
    # Tambahan Exceptions
    if name not in name_conf:
        raise ValueError("host %s not found" %name)

    conf = name_conf[name]
    found_conn = None
    for ip_port, conn in conn_mgr:
        ip, port = ip_port.split(':')
        print("DEBUG>>", ip, port, conf['ip'])
        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(1)
        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)
    result = web_job(conn, iso)
    return result


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


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


web_server = {}


def start_web_server():
    port = get_web_port()
    if not port:
        return
    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 = make_server('0.0.0.0', port, app)
    web_server['thread'] = create_thread(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):
    if 'listener' not in web_server:
        return
    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():
    # log_info("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():
    # log_info("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:
            if ip_port not in web_request:
                continue
            jobs = web_request[ip_port]
            if not jobs:
                continue
            iso = jobs[0]
            del jobs[0]
            append_web_process(ip_port, iso)
        connection.log_encode(iso)
        raw = iso.getRawIso()
        connection.send(raw)


def check_parser():
    # log_info("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()