forwarder.py

import os
import sys
import signal
from time import (
    sleep,
    time,
    )
from threading import Thread
from waitress.server import create_server
from pyramid.config import Configurator
from pyramid.paster import setup_logging
from opensipkd.tcp.connection import join_ip_port
from opensipkd.string import exception_message
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,
    web as web_conf,
    web_path_conf,
    listen_ports,
    allowed_ips,
    get_conf,
    name_conf,
    )
from .tools import iso_to_dict
from .connection import conn_mgr
from .views import (
    web_request,
    web_process,
    append_web_process,
    append_web_response,
    )
from .logger import (
    get_log,
    set_log,
    log_info,
    log_error,
    log_debug,
    log_web_info,
    log_web_debug,
    )


# Jika kosong berarti perintah untuk mengakhiri aplikasi ini
running = []

# Daftar server ISO8583, key: port
servers = {}

# Daftar client ISO8583, key: ip:port
clients = {}

# Daftar penerjemah dokumen ISO8583
parser_threads = []

# Web server profile
web_server = {}

# Daftar antrian saat sebagai ISO8583 forwarder
# key: ip:port
# value: dict of (
#            key: STAN forwarder
#            value: list of [
#                       ip_port aggregator,
#                       iso response for aggregator])
iso_process = {}


def append_iso_process(
        ip_port_target, iso_response, iso_for_host, ip_port_source):
    stan = iso_for_host.get_stan()
    data = [ip_port_source, iso_response]
    if ip_port_target in iso_process:
        iso_process[ip_port_target][stan] = data 
    else:
        iso_process[ip_port_target] = {stan: 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))


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()


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)


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:
            try:
                for_host = getattr(iso, 'for_host')
            except AttributeError:
                for_host = None
            if for_host:
                try:
                    self.send_iso_for_host(iso)
                except:
                    self.log_unknown()
            else:
                self.send(iso)
        else:
            self.on_response(from_iso)
        self.running = False

    def on_response(self, from_iso):
        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)
        # as forwarder
        if ip_port not in iso_process:
            return
        stan = from_iso.get_stan()
        if stan not in iso_process[ip_port]:
            return
        ip_port_source, iso = iso_process[ip_port][stan]
        try:
            iso.continue_response(from_iso)
            ok = True
        except:
            self.log_unknown()
            ok = False
        if ok:
            conn = get_connection_from_ip_port(ip_port_source)
            self.log_ack(iso)
            self.send_by_conn(iso, conn)
        del iso_process[ip_port][stan]

    def send_by_conn(self, iso, conn):
        conn.log_encode(iso)
        raw = iso.getRawIso()
        conn.send(raw)
 
    def send(self, iso):
        self.log_ack(iso)
        self.send_by_conn(iso, self.connection)

    def send_iso_for_host(self, iso):
        conn = get_connection_from_name(iso.for_host)
        if not conn:
            iso.ack_link_conf()
            self.send(iso)
            return
        if not conn.running:
            iso.ack_link_down()
            self.send(iso)
            return
        iso_for_host = iso.create_iso_for_host(conn.job.conf)
        ip_port_source = join_ip_port(self.conf['ip'], self.conf['port'])
        ip_port_target = join_ip_port(conn.conf['ip'], conn.conf['port'])
        append_iso_process(ip_port_target, iso, iso_for_host, ip_port_source)
        self.send_by_conn(iso_for_host, conn)


def get_connection_from_name(name):
    conf = name_conf[name]
    for ip_port, conn in conn_mgr:
        ip, port = ip_port.split(':')
        if conf['ip'] != ip:
            continue
        port = int(port)
        if conf['port'] == port:
            return conn


def get_connection_from_ip_port(ip_port):
    for this_ip_port, conn in conn_mgr:
        if ip_port == this_ip_port:
            return conn


#######
# Web #
#######
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')
        for path, cfg in web_path_conf.items():
            cfg['module_obj'].pyramid_init(config)
            config.scan(cfg['module'])
        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 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, __file__)
    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()