test-iso8583_send.py

"""Minimal ISO8583 sender.

Sends one optional sign-on (0800/301) then one transaction (0200) using a single invoice
profile (field 61 / bit61).

Example:
  python test-iso8583_send.py --host localhost --port 8592 --inv-id 3275050002008001801990
    python test-iso8583_send.py --host 36.95.7.36 --port 8595 --inv-file invoice_bks.txt --threads 50 --timeout 30 --no-login --no-etx
"""

from __future__ import annotations

import argparse
import socket
import threading
import time
from datetime import datetime, timedelta
from dataclasses import dataclass
from typing import Dict, List, Optional, Tuple
from ISO8583.ISO8583 import ISO8583
ETX = b"\x03"


ASCII_FIELD_SPECS: Dict[int, Tuple[str, int]] = {
    2: ("llvar", 2),
    3: ("fixed", 6),
    4: ("fixed", 12),
    7: ("fixed", 10),
    11: ("fixed", 6),
    12: ("fixed", 6),
    13: ("fixed", 4),
    15: ("fixed", 4),
    18: ("fixed", 4),
    22: ("fixed", 3),
    32: ("llvar", 2),
    33: ("llvar", 2),
    35: ("llvar", 2),
    37: ("fixed", 12),
    39: ("fixed", 2),
    41: ("fixed", 8),
    42: ("fixed", 15),
    43: ("fixed", 40),
    49: ("fixed", 3),
    59: ("lllvar", 3),
    60: ("lllvar", 3),
    61: ("lllvar", 3),
    62: ("lllvar", 3),
    63: ("lllvar", 3),
    70: ("fixed", 3),
    102: ("llvar", 2),
    107: ("lllvar", 3),
}


def load_invoices_from_file(path: str) -> List[str]:
    raw = open(path, "rb").read()

    if raw.startswith(b"\xff\xfe") or raw.startswith(b"\xfe\xff"):
        text = raw.decode("utf-16")
    elif raw.startswith(b"\xef\xbb\xbf"):
        text = raw.decode("utf-8-sig")
    else:
        try:
            text = raw.decode("utf-8")
        except UnicodeDecodeError:
            text = raw.decode("utf-16")

    invoices: List[str] = []
    for line in text.splitlines():
        s = line.strip()
        if not s or s.startswith("#"):
            continue
        invoices.append(s)
    return invoices


def recv_exact(sock: socket.socket, n: int) -> bytes:
    chunks: List[bytes] = []
    remaining = n
    while remaining > 0:
        chunk = sock.recv(remaining)
        if not chunk:
            raise ConnectionError("socket closed while reading")
        chunks.append(chunk)
        remaining -= len(chunk)
    return b"".join(chunks)


def send_frame(sock: socket.socket, payload: bytes, *, use_etx: bool) -> None:
    trailer = ETX if use_etx else b""
    frame_len = len(payload) + len(trailer)
    length = str(frame_len).zfill(4).encode("ascii")
    sock.sendall(length + payload + trailer)


def recv_frame(sock: socket.socket, *, expect_etx: bool) -> bytes:
    length_raw = recv_exact(sock, 4)
    try:
        length = int(length_raw.decode("ascii"))
    except ValueError as e:
        raise ValueError(f"invalid length prefix: {length_raw!r}") from e

    data = recv_exact(sock, length-4)
    if expect_etx:
        if not data.endswith(ETX):
            raise ValueError(f"invalid frame trailer (expected ETX 0x03): {data[-1:]!r}")
        return data[:-1]
    return data


def bits_from_hex_bitmap(hex16: str, offset: int) -> List[int]:
    b = bytes.fromhex(hex16)
    out: List[int] = []
    for byte_index, byte_value in enumerate(b):
        for bit_index in range(8):
            mask = 1 << (7 - bit_index)
            if byte_value & mask:
                out.append(offset + byte_index * 8 + bit_index + 1)
    return out


def parse_until_39(payload: bytes) -> Dict[int, str]:
    """Parse MTI + fields up to (and including) bit 39."""
    s = payload.decode("ascii", errors="replace")
    if len(s) < 4 + 16:
        raise ValueError(f"payload too short: {payload!r}")

    mti = s[:4]
    pmap = s[4:20]
    bits = bits_from_hex_bitmap(pmap, 0)
    index = 20

    if 1 in bits:
        if len(s) < index + 16:
            raise ValueError("payload too short for secondary bitmap")
        smap = s[index : index + 16]
        bits += bits_from_hex_bitmap(smap, 64)
        index += 16

    values: Dict[int, str] = {0: mti}
    for bit in sorted(bits):
        if bit == 1:
            continue
        spec = ASCII_FIELD_SPECS.get(bit)
        if spec is None:
            raise ValueError(f"unsupported field bit={bit}; add spec to ASCII_FIELD_SPECS")

        kind, n = spec
        if kind == "fixed":
            values[bit] = s[index : index + n]
            index += n
        elif kind == "llvar":
            length = int(s[index : index + n])
            index += n
            values[bit] = s[index : index + length].rstrip()
            index += length
        elif kind == "lllvar":
            length = int(s[index : index + n])
            index += n
            values[bit] = s[index : index + length].rstrip()
            index += length
        else:
            raise ValueError(f"unsupported field kind: {kind}")

        if bit == 39:
            break

    return values


def now_bit7(dt: Optional[datetime] = None) -> str:
    if dt is None:
        dt = datetime.now()
    return dt.strftime("%m%d%H%M%S")


def ts() -> str:
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]


_print_lock = threading.Lock()


def log_line(line: str) -> None:
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
    line = f"{now} {line}"
    with _print_lock:
        print(line, flush=True)


def format_ok_line(*, kind: str, inv_id: Optional[str], rc39: Optional[str], send_ts: str, t0: float, stan: Optional[str] = None) -> str:
    dt_ms = (time.perf_counter() - t0) * 1000.0
    inv_part = "" if inv_id is None else f" inv_id={inv_id}"
    stan_part = f" stan={stan}" if stan is not None else ""
    dt = datetime.strptime(send_ts, '%Y-%m-%d %H:%M:%S.%f') 
    akhir= datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')

    return f"{send_ts} {kind}{inv_part}{stan_part} rc39={rc39} duration_ms={dt_ms:.2f} elapsed={akhir}s"


def format_err_line(*, kind: str, inv_id: Optional[str], exc: BaseException, send_ts: str, t0: float, stan: Optional[str] = None) -> str:
    dt_ms = (time.perf_counter() - t0) * 1000.0
    inv_part = "" if inv_id is None else f" inv_id={inv_id}"
    stan_part = f" stan={stan}" if stan is not None else ""
    return f"{send_ts} {kind}{inv_part}{stan_part} ERROR {type(exc).__name__}:{exc} duration_ms={dt_ms:.2f}"


@dataclass
class TxResult:
    inv_id: str
    ok: bool
    rc39: Optional[str]
    duration_ms: Optional[float]
    error: Optional[str] = None


@dataclass
class _Pending:
    kind: str  # "TX" or "SIGNON"
    inv_id: Optional[str]
    stan: str
    send_ts: str
    t0: float
    event: threading.Event
    result: Optional[TxResult] = None
    ok: Optional[bool] = None
    error: Optional[str] = None



# Standalone receipt thread for STAN dispatch
def log_send(kind: str, inv_id: str|None, stan: str, send_ts: str):
    inv_part = f" inv_id={inv_id}" if inv_id else ""
    log_line(f"SEND   {send_ts} {kind}{inv_part} stan={stan}")

def log_receipt(kind: str, inv_id: str|None, stan: str, rc39: str|None, recv_ts: str):
    inv_part = f" inv_id={inv_id}" if inv_id else ""
    log_line(f"RECEIPT {recv_ts} {kind}{inv_part} stan={stan} rc39={rc39}")

class ReceiptThread(threading.Thread):
    # def log_send(kind: str, inv_id: str|None, stan: str, send_ts: str):
    #     inv_part = f" inv_id={inv_id}" if inv_id else ""
    #     log_line(f"SEND   {send_ts} {kind}{inv_part} stan={stan}")

    # def log_receipt(kind: str, inv_id: str|None, stan: str, rc39: str|None, recv_ts: str):
    #     inv_part = f" inv_id={inv_id}" if inv_id else ""
    #     log_line(f"RECEIPT {recv_ts} {kind}{inv_part} stan={stan} rc39={rc39}")
    def __init__(self, sock, use_etx, pending, pending_lock, closed):
        super().__init__(daemon=True)
        self.sock = sock
        self.use_etx = use_etx
        self.pending = pending
        self.pending_lock = pending_lock
        self.closed = closed
        self.sock.settimeout(1.0)

    def run(self):
        while not self.closed.is_set():
            try:
                resp = recv_frame(self.sock, expect_etx=self.use_etx)
            except socket.timeout:
                continue
            except (OSError, ValueError) as e:
                with self.pending_lock:
                    pendings = list(self.pending.values())
                    self.pending.clear()
                for p in pendings:
                    p.error = type(e).__name__
                    if p.kind == "TX":
                        p.result = TxResult(inv_id=p.inv_id or "", ok=False, rc39=None, duration_ms=None, error=p.error)
                    else:
                        p.ok = False
                    p.event.set()
                return

            # Print/log the raw response bytes received
            log_line(f"[RECEIPT RAW] {resp.hex()}")

            try:
                parsed = parse_until_39(resp)
            except (ValueError, UnicodeError) as e:
                with self.pending_lock:
                    pendings = list(self.pending.values())
                    self.pending.clear()
                for p in pendings:
                    p.error = type(e).__name__
                    if p.kind == "TX":
                        p.result = TxResult(inv_id=p.inv_id or "", ok=False, rc39=None, duration_ms=None, error=p.error)
                    else:
                        p.ok = False
                    p.event.set()
                return

            stan = parsed.get(11)
            if not stan:
                continue
            with self.pending_lock:
                p = self.pending.pop(stan, None)
            if p is None:
                continue
            rc = parsed.get(39)
            dt_ms = (time.perf_counter() - p.t0) * 1000.0
            recv_ts = ts()
            log_receipt(p.kind, p.inv_id, stan, rc, recv_ts)
            if p.kind == "SIGNON":
                ok = parsed.get(0) == "0810" and rc == "00"
                log_line(format_ok_line(kind="SIGNON", inv_id=None, rc39=rc, send_ts=p.send_ts, t0=p.t0, stan=stan))
                p.ok = ok
            else:
                ok = parsed.get(0) == "0210" and rc == "00"
                log_line(format_ok_line(kind="TX", inv_id=p.inv_id, rc39=rc, send_ts=p.send_ts, t0=p.t0, stan=stan))
                p.result = TxResult(inv_id=p.inv_id or "", ok=ok, rc39=rc, duration_ms=dt_ms)
            p.event.set()


def send_signon(sock: socket.socket, *, use_etx: bool, stan_counter: "StanCounter") -> bool:
    """Minimal sign-on: send 0800/301 and check for 0810/00 response."""
    stan = stan_counter.next()
    payload = build_0800_network(stan, "301")
    send_ts = ts()
    t0 = time.perf_counter()
    try:
        send_frame(sock, payload, use_etx=use_etx)
        resp = recv_frame(sock, expect_etx=use_etx)
        parsed = parse_until_39(resp)
        rc = parsed.get(39)
        log_line(format_ok_line(kind="SIGNON", inv_id=None, rc39=rc, send_ts=send_ts, t0=t0, stan=stan))
        return parsed.get(0) == "0810" and rc == "00"
    except Exception as e:
        log_line(format_err_line(kind="SIGNON", inv_id=None, exc=e, send_ts=send_ts, t0=t0, stan=stan))
        return False


def send_tx(sock: socket.socket, *, use_etx: bool, stan_counter: "StanCounter", inv_id: str) -> TxResult:
    stan = stan_counter.next()
    payload = build_0200_transaction(stan, bit61=inv_id)
    send_ts = ts()
    t0 = time.perf_counter()
    try:
        print(f"DEBUG: sending transaction for inv_id={inv_id} with stan={stan} at {send_ts} {datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')}")
        send_frame(sock, payload, use_etx=use_etx)
        resp = recv_frame(sock, expect_etx=use_etx)
        print(f"DEBUG: received transaction for inv_id={inv_id} with stan={stan} at {send_ts} {datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')}")
        dt_ms = (time.perf_counter() - t0) * 1000.0
        parsed = parse_until_39(resp)
        rc = parsed.get(39)
        ok = parsed.get(0) == "0210" and rc == "00"
        log_line(format_ok_line(kind="TX", inv_id=inv_id, rc39=rc, send_ts=send_ts, t0=t0, stan=stan))
        return TxResult(inv_id=inv_id, ok=ok, rc39=rc, duration_ms=dt_ms)
    except (OSError, ValueError) as e:
        dt_ms = (time.perf_counter() - t0) * 1000.0
        log_line(format_err_line(kind="TX", inv_id=inv_id, exc=e, send_ts=send_ts, t0=t0, stan=stan))
        return TxResult(inv_id=inv_id, ok=False, rc39=None, duration_ms=dt_ms, error=type(e).__name__)


def percentile(sorted_values: List[float], p: float) -> float:
    if not sorted_values:
        return 0.0
    if p <= 0:
        return sorted_values[0]
    if p >= 100:
        return sorted_values[-1]
    k = (len(sorted_values) - 1) * (p / 100.0)
    f = int(k)
    c = min(f + 1, len(sorted_values) - 1)
    if f == c:
        return sorted_values[f]
    d0 = sorted_values[f] * (c - k)
    d1 = sorted_values[c] * (k - f)
    return d0 + d1


def print_summary(results: List[TxResult]) -> None:
    total = len(results)
    ok = sum(1 for r in results if r.ok)
    fail = total - ok

    rc_counts: Dict[str, int] = {}
    for r in results:
        rc = r.rc39 or "NA"
        rc_counts[rc] = rc_counts.get(rc, 0) + 1

    durations = [r.duration_ms for r in results if r.duration_ms is not None]
    durations_f = [float(x) for x in durations]
    durations_sorted = sorted(durations_f)

    log_line("\nSummary")
    log_line(f"  total: {total}")
    log_line(f"  ok:    {ok}")
    log_line(f"  fail:  {fail}")
    if durations_sorted:
        avg = sum(durations_sorted) / len(durations_sorted)
        log_line(
            "  duration_ms: "
            f"avg={avg:.2f} p50={percentile(durations_sorted, 50):.2f} "
            f"p95={percentile(durations_sorted, 95):.2f} max={durations_sorted[-1]:.2f}"
        )

    log_line("  rc39_counts:")
    for rc, cnt in sorted(rc_counts.items(), key=lambda kv: (-kv[1], kv[0])):
        log_line(f"    {rc}: {cnt}")


def print_run_start(
    *,
    start_ts: str,
    host: str,
    port: int,
    use_etx: bool,
    do_signon: bool,
    invoices_count: int,
    threads: int,
    timeout: float,
) -> None:
    mode = "single-connection"
    frame = "etx" if use_etx else "no-etx"
    signon = "on" if do_signon else "off"
    log_line(
        f"Start ts={start_ts} mode={mode} frame={frame} signon={signon} "
        f"invoices={invoices_count} threads={threads} timeout={timeout}s target={host}:{port}"
    )


def print_run_end(*, end_ts: str, elapsed_s: float) -> None:
    log_line(f"End ts={end_ts} elapsed_s={elapsed_s:.3f}")


class StanCounter:
    def __init__(self) -> None:
        self._value = int(datetime.now().strftime("%H%M%S")) % 1_000_000

    def next(self) -> str:
        self._value = (self._value + 1) % 1_000_000
        return str(self._value).zfill(6)


def build_0800_network(stan: str, code70: str) -> bytes:
    mti = "0800"
    primary = "8220000000000000"  # bits 1,7,11
    secondary = "0400000000000000"  # bit 70
    bit7 = now_bit7()
    return f"{mti}{primary}{secondary}{bit7}{stan}{code70}".encode("ascii")


def build_0200_transaction(stan: str, *, bit61: str) -> bytes:
    dt = datetime.now()
    mti = "0200"
    bitmap = "F23A4401A8E0803A0000000004200000"

    pan = "622011444444444444"
    proc_code = "341019"
    bit7 = dt.strftime("%m%d%H%M%S")
    bit11 = stan
    bit12 = dt.strftime("%H%M%S")
    bit13 = dt.strftime("%m%d")
    bit15 = (dt + timedelta(days=1)).strftime("%m%d")

    bit18 = "6010"
    bit22 = "021"
    bit32 = "110"
    bit33 = "00110"
    bit35 = "622011444444444444=9912"
    bit37 = ("000000" + stan)[-12:]
    bit41 = "N703".ljust(8)
    bit42 = "02N703".ljust(15)
    bit43 = "TLR     N703".ljust(40)
    bit49 = "360"
    bit59 = "PAY"
    bit60 = "120"
    bit63 = "214"
    bit102 = "0010823214360".ljust(20)
    bit107 = "0010"

    amount12 = "000000000000"
    parts = [
        mti,
        bitmap,
        f"{len(pan):02d}{pan}",
        proc_code,
        amount12,
        bit7,
        bit11,
        bit12,
        bit13,
        bit15,
        bit18,
        bit22,
        f"{len(bit32):02d}{bit32}",
        f"{len(bit33):02d}{bit33}",
        f"{len(bit35):02d}{bit35}",
        bit37,
        bit41,
        bit42,
        bit43,
        bit49,
        f"{len(bit59):03d}{bit59}",
        f"{len(bit60):03d}{bit60}",
        f"{len(bit61):03d}{bit61}",
        f"{len(bit63):03d}{bit63}",
        f"{len(bit102):02d}{bit102}",
        f"{len(bit107):03d}{bit107}",
    ]
    return "".join(parts).encode("ascii")


def main(argv: Optional[List[str]] = None) -> int:
    p = argparse.ArgumentParser(description="Minimal ISO8583 sender")
    p.add_argument("--host", default="127.0.0.1", help="Server host (default: 127.0.0.1)")
    p.add_argument("--port", default=8585, type=int, help="Server port (default: 8585)")
    p.add_argument("--inv-id", dest="bit61", default=None, help="invoice profile string for bit 61")
    p.add_argument(
        "--inv-file",
        default="demo.txt",
        help="path to invoices.txt (one invoice per line; blank/# ignored) (default: demo.txt)",
    )
    p.add_argument(
        "--limit",
        type=int,
        default=None,
        help="optional max number of invoices to send from --inv-file",
    )
    p.add_argument(
        "--threads",
        type=int,
        default=1,
        help="max concurrent invoice threads (still uses one TCP connection)",
    )
    p.add_argument("--timeout", type=float, default=25.0)
    p.add_argument("--no-etx", action="store_true", help="length prefix only (no trailing ETX)")
    p.add_argument("--no-signon", action="store_true", help="skip initial 0800/301")

    p.add_argument(
        "--network-only",
        action="store_true",
        help="Only send a network management (0800/301) message and exit."
    )
    p.add_argument(
        "--network-count",
        type=int,
        default=1,
        help="Number of network management (sign-on) messages to send if --network-only is set (default: 1)"
    )
    args = p.parse_args(argv)


    if args.network_only:
        use_etx = not args.no_etx
        results = []
        count = args.network_count
        with socket.create_connection((args.host, args.port), timeout=args.timeout) as s:
            stan_counter = StanCounter()
            threads = []
            results = [None] * count

            def send_one(idx):
                ok = send_signon(s, use_etx=use_etx, stan_counter=stan_counter)
                print(f"Network management test {idx+1}/{count} (0800/301) sent, response OK: {ok}")
                results[idx] = ok

            for i in range(count):
                t = threading.Thread(target=send_one, args=(i,))
                threads.append(t)
                t.start()
            for t in threads:
                t.join()
        return 0 if all(results) else 1

    invoices: List[str] = []
    if args.inv_file:
        invoices = load_invoices_from_file(args.inv_file)
        if args.limit is not None:
            if args.limit <= 0:
                raise SystemExit("--limit must be > 0")
            invoices = invoices[: args.limit]
    elif args.bit61:
        invoices = [args.bit61]
    else:
        raise SystemExit("provide --inv-id or --inv-file")

    if not invoices:
        raise SystemExit("no invoices loaded")

    if args.threads <= 0:
        raise SystemExit("--threads must be > 0")

    use_etx = not args.no_etx
    do_signon = not args.no_signon

    run_start_ts = ts()
    run_t0 = time.perf_counter()
    print_run_start(
        start_ts=run_start_ts,
        host=args.host,
        port=args.port,
        use_etx=use_etx,
        do_signon=do_signon,
        invoices_count=len(invoices),
        threads=args.threads,
        timeout=args.timeout,
    )

    # Server closes old connections when a new connection is opened from the same
    # configured client key. So keep exactly one TCP connection for the whole run.
    results: List[TxResult] = []
    any_fail = False


    try:
        with socket.create_connection((args.host, args.port), timeout=args.timeout) as s:
            # Shared state for receipt thread
            pending: Dict[str, _Pending] = {}
            pending_lock = threading.Lock()
            closed = threading.Event()
            stan_counter = StanCounter()
            send_lock = threading.Lock()

            # Start receipt thread
            receipt_thread = ReceiptThread(s, use_etx, pending, pending_lock, closed)
            receipt_thread.start()


            # Minimal sign-on using the reduced function
            def send_signon_stan():
                return send_signon(s, use_etx=use_etx, stan_counter=stan_counter)

            def send_tx_stan(inv_id: str) -> TxResult:
                stan = stan_counter.next()
                payload = build_0200_transaction(stan, bit61=inv_id)
                send_ts = ts()
                t0 = time.perf_counter()
                p = _Pending(kind="TX", inv_id=inv_id, stan=stan, send_ts=send_ts, t0=t0, event=threading.Event())
                with pending_lock:
                    pending[stan] = p
                log_send("TX", inv_id, stan, send_ts)
                try:
                    with send_lock:
                        send_frame(s, payload, use_etx=use_etx)
                except (OSError, ValueError) as e:
                    with pending_lock:
                        pending.pop(stan, None)
                    log_line(format_err_line(kind="TX", inv_id=inv_id, exc=e, send_ts=send_ts, t0=t0, stan=stan))
                    return TxResult(inv_id=inv_id, ok=False, rc39=None, duration_ms=None, error=type(e).__name__)
                if not p.event.wait(args.timeout):
                    with pending_lock:
                        pending.pop(stan, None)
                    log_line(format_err_line(kind="TX", inv_id=inv_id, exc=TimeoutError("timeout"), send_ts=send_ts, t0=t0, stan=stan))
                    return TxResult(inv_id=inv_id, ok=False, rc39=None, duration_ms=None, error="timeout")
                return p.result or TxResult(inv_id=inv_id, ok=False, rc39=None, duration_ms=None, error=p.error or "unknown")

            if do_signon:
                ok_signon = send_signon_stan()
                if not ok_signon:
                    any_fail = True

            results = [
                TxResult(inv_id=inv_id, ok=False, rc39=None, duration_ms=None, error="not_processed")
                for inv_id in invoices
            ]

            sem = threading.Semaphore(args.threads)

            def run_one(idx: int, inv_id: str) -> None:
                nonlocal any_fail
                with sem:
                    r = send_tx_stan(inv_id)
                    results[idx] = r
                    if not r.ok:
                        any_fail = True

            threads: List[threading.Thread] = []
            for i, inv_id in enumerate(invoices):
                t = threading.Thread(target=run_one, args=(i, inv_id), daemon=True)
                threads.append(t)
                t.start()
            for t in threads:
                t.join()

            closed.set()
            receipt_thread.join(timeout=2)

    except (OSError, ValueError) as e:
        log_line(format_err_line(kind="CONNECT", inv_id=None, exc=e, send_ts=ts(), t0=time.perf_counter()))
        if not results:
            results = [TxResult(inv_id=inv_id, ok=False, rc39=None, duration_ms=None, error=type(e).__name__) for inv_id in invoices]
            any_fail = True

    print_summary(results)
    print_run_end(end_ts=ts(), elapsed_s=time.perf_counter() - run_t0)
    return 1 if any_fail else 0


if __name__ == "__main__":
    raise SystemExit(main())