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
"""

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

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)
    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:
    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) -> str:
    dt_ms = (time.perf_counter() - t0) * 1000.0
    inv_part = "" if inv_id is None else f" inv_id={inv_id}"
    return f"{send_ts} {kind}{inv_part} rc39={rc39} duration_ms={dt_ms:.2f}"


def format_err_line(*, kind: str, inv_id: Optional[str], exc: BaseException, send_ts: str, t0: float) -> str:
    dt_ms = (time.perf_counter() - t0) * 1000.0
    inv_part = "" if inv_id is None else f" inv_id={inv_id}"
    return f"{send_ts} {kind}{inv_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


class StanDispatcher:
    """Single TCP connection dispatcher keyed by STAN (field 11).

    - Many threads may call .tx() concurrently.
    - Only one receiver thread reads from socket.
    - Responses are routed to the right waiter using bit 11 (STAN).
    """

    def __init__(self, sock: socket.socket, *, use_etx: bool, timeout_s: float):
        self._sock = sock
        self._use_etx = use_etx
        self._timeout_s = timeout_s

        self._closed = threading.Event()
        self._send_lock = threading.Lock()
        self._pending_lock = threading.Lock()
        self._pending: Dict[str, _Pending] = {}

        self._stan_lock = threading.Lock()
        self._stan_counter = StanCounter()

        # Receiver loop uses a short timeout to notice close.
        self._sock.settimeout(1.0)
        self._rx_thread = threading.Thread(target=self._rx_loop, daemon=True)
        self._rx_thread.start()

    def close(self) -> None:
        self._closed.set()

        try:
            self._sock.shutdown(socket.SHUT_RDWR)
        except OSError:
            pass
        try:
            self._sock.close()
        except OSError:
            pass

        # Unblock any waiters.
        with self._pending_lock:
            pendings = list(self._pending.values())
            self._pending.clear()
        for p in pendings:
            p.error = p.error or "closed"
            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()

    def _next_stan(self) -> str:
        with self._stan_lock:
            return self._stan_counter.next()

    def _fail_all(self, err: BaseException) -> None:
        with self._pending_lock:
            pendings = list(self._pending.values())
            self._pending.clear()
        for p in pendings:
            p.error = type(err).__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()

    def _rx_loop(self) -> None:
        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:
                self._fail_all(e)
                return

            try:
                parsed = parse_until_39(resp)
            except (ValueError, UnicodeError) as e:
                self._fail_all(e)
                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

            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))
                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))
                p.result = TxResult(inv_id=p.inv_id or "", ok=ok, rc39=rc, duration_ms=dt_ms)

            p.event.set()

    def signon(self) -> bool:
        stan = self._next_stan()
        payload = build_0800_network(stan, "301")
        send_ts = ts()
        t0 = time.perf_counter()

        p = _Pending(kind="SIGNON", inv_id=None, stan=stan, send_ts=send_ts, t0=t0, event=threading.Event())
        with self._pending_lock:
            self._pending[stan] = p

        try:
            with self._send_lock:
                send_frame(self._sock, payload, use_etx=self._use_etx)
        except (OSError, ValueError) as e:
            with self._pending_lock:
                self._pending.pop(stan, None)
            log_line(format_err_line(kind="SIGNON", inv_id=None, exc=e, send_ts=send_ts, t0=t0))
            return False

        if not p.event.wait(self._timeout_s):
            with self._pending_lock:
                self._pending.pop(stan, None)
            log_line(format_err_line(kind="SIGNON", inv_id=None, exc=TimeoutError("timeout"), send_ts=send_ts, t0=t0))
            return False
        return bool(p.ok)

    def tx(self, inv_id: str) -> TxResult:
        stan = self._next_stan()
        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 self._pending_lock:
            self._pending[stan] = p

        try:
            with self._send_lock:
                send_frame(self._sock, payload, use_etx=self._use_etx)
        except (OSError, ValueError) as e:
            with self._pending_lock:
                self._pending.pop(stan, None)
            log_line(format_err_line(kind="TX", inv_id=inv_id, exc=e, send_ts=send_ts, t0=t0))
            return TxResult(inv_id=inv_id, ok=False, rc39=None, duration_ms=None, error=type(e).__name__)

        if not p.event.wait(self._timeout_s):
            with self._pending_lock:
                self._pending.pop(stan, None)
            log_line(format_err_line(kind="TX", inv_id=inv_id, exc=TimeoutError("timeout"), send_ts=send_ts, t0=t0))
            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")


def send_signon(sock: socket.socket, *, use_etx: bool, stan_counter: "StanCounter") -> bool:
    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))
        return parsed.get(0) == "0810" and rc == "00"
    except (OSError, ValueError) as e:
        log_line(format_err_line(kind="SIGNON", inv_id=None, exc=e, send_ts=send_ts, t0=t0))
        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:
        send_frame(sock, payload, use_etx=use_etx)
        resp = recv_frame(sock, expect_etx=use_etx)
        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))
        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))
        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", required=True)
    p.add_argument("--port", required=True, type=int)
    p.add_argument("--inv-id", dest="bit61", default=None, help="invoice profile string for bit 61")
    p.add_argument(
        "--inv-file",
        default=None,
        help="path to invoices.txt (one invoice per line; blank/# ignored)",
    )
    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")
    args = p.parse_args(argv)

    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:
            dispatcher = StanDispatcher(s, use_etx=use_etx, timeout_s=args.timeout)

            if do_signon:
                ok_signon = dispatcher.signon()
                if not ok_signon:
                    any_fail = True
                    # Still continue to print summary/run end.

            # Prepare result slots to preserve invoice ordering in summary.
            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 = dispatcher.tx(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()

            dispatcher.close()
    except (OSError, ValueError) as e:
        log_line(format_err_line(kind="CONNECT", inv_id=None, exc=e, send_ts=ts(), t0=time.perf_counter()))
        # Mark all as failed if we never managed to connect.
        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())