encrypt.py 3.77 KB
import sys
import json
import logging.config
from datetime import datetime
from argparse import ArgumentParser
from base64 import b64encode
from pprint import pprint
from opensipkd.waktu import create_now
from winpay.encrypt import (
    sign,
    sign_without_salt,
    verify,
    verify_without_salt,
    )
from winpay.signature import generator
from winpay.va import (
    create,
    HttpErr,
    )


def encrypt(argv=sys.argv[1:]):
    pars = ArgumentParser()
    pars.add_argument('--private-file', required=True)
    pars.add_argument('--data-file', required=True)
    pars.add_argument('--without-salt', action='store_true')
    pars.add_argument('--public-file')
    option = pars.parse_args(argv)

    with open(option.private_file, 'rb') as f:
        private_key = f.read()

    with open(option.data_file, 'rb') as f:
        data = f.read()

    sign_func = option.without_salt and sign_without_salt or sign
    data_signed = sign_func(private_key, data)
    data_b64 = b64encode(data_signed).decode('utf-8')
    print(data_b64)
    if not option.public_file:
        return
    with open(option.public_file, 'rb') as f:
        public_key = f.read()
    verify_func = option.without_salt and verify_without_salt or verify
    verify_func(public_key, data_signed, data)


def signature(argv=sys.argv[1:]):
    pars = ArgumentParser()
    pars.add_argument('--private-file', required=True)
    pars.add_argument('--json-file', required=True)
    option = pars.parse_args(sys.argv[1:])

    with open(option.private_file, 'rb') as f:
        private_bytes = f.read()

    with open(option.json_file) as f:
        data = json.loads(f.read())

    timestamp = create_now()
    sign_bytes = generator(private_bytes, data, timestamp)
    sign_b64 = b64encode(sign_bytes).decode('utf-8')
    print(sign_b64)


def setup_logging():
    logging.config.dictConfig({
        'version': 1,
        'formatters': {
            'generic': {
                'format': '%(asctime)s %(levelname)s %(name)s %(message)s',
                },
            },
        'handlers': {
            'console': {
                'class': 'logging.StreamHandler',
                'stream': 'ext://sys.stdout',
                'formatter': 'generic'},
            },
        'loggers': {
            '': {
                'handlers': ['console'],
                'level': 'DEBUG',
                },
            },
        })


def create_va(argv=sys.argv[1:]):
    url = 'https://sandbox-api.bmstaging.id/snap/v1.0/transfer-va/create-va'
    help_url = f'default {url}'

    expired_days = 1
    help_expired = f'default {expired_days}'

    channels = [
        'BRI', 'BNI', 'MANDIRI', 'PERMATA', 'BSI', 'MUAMALAT', 'BCA', 'CIMB',
        'SINARMAS', 'BNC', 'INDOMARET', 'ALFAMART']

    pars = ArgumentParser()
    pars.add_argument('--private-file', required=True)
    pars.add_argument('--partner-id', required=True)
    pars.add_argument('--customer-no', required=True)
    pars.add_argument('--va-name', required=True)
    pars.add_argument('--amount', type=int, required=True)
    pars.add_argument('--channel', choices=channels)
    pars.add_argument('--url', default=url, help=help_url)
    pars.add_argument(
        '--expired-days', type=int, default=expired_days, help=help_expired)
    pars.add_argument('--trx-id')
    option = pars.parse_args(sys.argv[1:])

    with open(option.private_file, 'rb') as f:
        private_key = f.read()

    if option.trx_id:
        trx_id = option.trx_id
    else:
        trx_id = datetime.now().strftime('%H%M%S')

    setup_logging()
    try:
        r = create(
                option.url, private_key, option.partner_id, option.customer_no,
                option.va_name, option.amount, option.channel, trx_id,
                option.expired_days)
        pprint(r)
    except HttpErr as e:
        print('ERROR', e)