ISO 8583: Specifiche Tecniche e Guida all'Implementazione

Acquirer -> Network -> Issuer
    |          |         |
    |          |         |
    └----------└---------┘
      Risposta

Bit 1: Presenza bitmap secondaria
Bit 2-64: Presenza dei campi dati 1-63

Vantaggi:
- Leggibilità umana
- Compatibilità universale

Svantaggi:
- Maggiore occupazione di spazio
- Overhead di conversione

Vantaggi:
- Compatibilità mainframe
- Efficienza elaborativa

Svantaggi:
- Minore portabilità
- Complessità di debug

Vantaggi:
- Massima efficienza spaziale
- Performance ottimali

Svantaggi:
- Difficoltà di debug
- Problemi di endianness

from dataclasses import dataclass
from typing import Optional, Any

@dataclass
class ISO8583Field:
    """Rappresentazione di un campo ISO 8583"""
    length: int
    type: str
    data: bytes
    encoding: str

from typing import Dict, List, Optional
from dataclasses import dataclass
import struct

@dataclass
class ISO8583Message:
    mti: str
    fields: Dict[int, str]
    bitmap: bytes

class ISO8583Parser:
    def __init__(self):
        self.mti = None
        self.bitmap = None
        self.fields = {}
        self._field_lengths = self._init_field_lengths()

    def _init_field_lengths(self) -> Dict[int, int]:
        """Inizializza le lunghezze dei campi standard"""
        return {
            2: 19,  # PAN
            3: 6,   # Processing Code
            4: 12,  # Amount
            7: 10,  # Date and Time
            11: 6,  # STAN
            # ... altri campi
        }

    def parse(self, data: bytes) -> ISO8583Message:
        """Parse raw ISO 8583 message"""
        try:
            position = 0

            # Parse MTI (4 bytes in ASCII)
            self.mti = data[position:position+4].decode('ascii')
            position += 4

            # Parse primary bitmap (8 bytes)
            self.bitmap = data[position:position+8]
            bitmap_int = int.from_bytes(self.bitmap, byteorder='big')
            position += 8

            # Check for secondary bitmap
            if bitmap_int & (1 << 63):
                self.bitmap += data[position:position+8]
                position += 8

            # Parse fields based on bitmap
            present_fields = self._get_present_fields(bitmap_int)
            for field_num in present_fields:
                field_data = self._parse_field(data, position, field_num)
                self.fields[field_num] = field_data
                position += len(field_data)

            return ISO8583Message(self.mti, self.fields, self.bitmap)

        except Exception as e:
            raise ISO8583ParseError(f"Parsing error at position {position}: {str(e)}")

    def _get_present_fields(self, bitmap_int: int) -> List[int]:
        """Determine which fields are present based on bitmap"""
        fields = []
        for i in range(1, 65):  # Primary bitmap fields
            if bitmap_int & (1 << (64 - i)):
                fields.append(i)
        return fields

    def _parse_field(self, data: bytes, position: int, field_num: int) -> str:
        """Parse individual field based on field number"""
        if field_num in self._field_lengths:
            # Fixed length field
            length = self._field_lengths[field_num]
            field_data = data[position:position+length]
            return field_data.decode('ascii')
        else:
            # Variable length field
            length_indicator = int(data[position:position+2].decode('ascii'))
            field_data = data[position+2:position+2+length_indicator]
            return field_data.decode('ascii')

class ISO8583Builder:
    """Builder for creating ISO 8583 messages"""
    def __init__(self):
        self.fields = {}
        self.mti = "0100"  # Default to authorization request

    def set_mti(self, mti: str) -> 'ISO8583Builder':
        """Set Message Type Indicator"""
        if not (len(mti) == 4 and mti.isdigit()):
            raise ValueError("MTI must be 4 digits")
        self.mti = mti
        return self

    def add_field(self, field_num: int, value: str) -> 'ISO8583Builder':
        """Add a field to the message"""
        if not 1 <= field_num <= 128:
            raise ValueError("Field number must be between 1 and 128")
        self.fields[field_num] = value
        return self

    def build(self) -> bytes:
        """Build the complete ISO 8583 message"""
        # Start with MTI
        message = self.mti.encode('ascii')

        # Generate bitmap
        bitmap = self._generate_bitmap()
        message += bitmap

        # Add fields in order
        fields = sorted(self.fields.keys())
        for field_num in fields:
            field_data = self._encode_field(field_num, self.fields[field_num])
            message += field_data

        return message

    def _generate_bitmap(self) -> bytes:
        """Generate bitmap based on present fields"""
        bitmap = 0
        for field_num in self.fields.keys():
            bitmap |= (1 << (64 - field_num))
        return bitmap.to_bytes(8, byteorder='big')

    def _encode_field(self, field_num: int, value: str) -> bytes:
        """Encode a field value according to ISO 8583 rules"""
        # Implement field-specific encoding logic here
        return value.encode('ascii')

class ISO8583Error(Exception):
    """Base exception for ISO8583 parsing errors"""
    pass

class InvalidMTIError(ISO8583Error):
    """Invalid Message Type Indicator"""
    pass

class InvalidBitmapError(ISO8583Error):
    """Invalid bitmap format"""
    pass

class FieldLengthError(ISO8583Error):
    """Field length mismatch"""
    pass

from cryptography.fernet import Fernet

class SecureISO8583:
    def __init__(self, key: bytes):
        self.fernet = Fernet(key)

    def encrypt_pan(self, pan: str) -> bytes:
        """Encrypt Primary Account Number"""
        return self.fernet.encrypt(pan.encode())

    def decrypt_pan(self, encrypted_pan: bytes) -> str:
        """Decrypt Primary Account Number"""
        return self.fernet.decrypt(encrypted_pan).decode()

def validate_field(field_num: int, value: str) -> bool:
    """Validate field content"""
    validators = {
        2: lambda x: len(x) <= 19 and x.isdigit(),  # PAN
        3: lambda x: len(x) == 6 and x.isdigit(),   # Processing code
        4: lambda x: len(x) == 12 and x.isdigit(),  # Amount
    }
    return validators.get(field_num, lambda x: True)(value)

from array import array
from typing import Dict, Optional
import mmap

class OptimizedParser:
    """Parser ottimizzato per alte performance"""
    def __init__(self):
        # Pre-allocate buffers per evitare allocazioni dinamiche
        self.field_cache = array('B', [0] * 1024)
        self.bitmap_cache = array('B', [0] * 16)
        self._setup_lookup_tables()

    def _setup_lookup_tables(self):
        """Inizializza tabelle di lookup per parsing veloce"""
        self.field_lengths = array('H', [0] * 128)  # Lunghezze campi fisse
        self.field_encodings = array('B', [0] * 128)  # Encoding per campo

        # Populate lookup tables
        for i in range(128):
            self.field_lengths[i] = self._get_standard_length(i)
            self.field_encodings[i] = self._get_standard_encoding(i)

    def parse_optimized(self, data: bytes) -> Dict[int, str]:
        """Parsing ottimizzato con memoria pre-allocata"""
        # Usa mmap per file grandi
        if len(data) > 1024 * 1024:  # 1MB
            with mmap.mmap(-1, len(data)) as mm:
                mm.write(data)
                return self._parse_mmap(mm)

        # Fast path per messaggi piccoli
        return self._parse_memory(memoryview(data))

    def _parse_mmap(self, mm: mmap.mmap) -> Dict[int, str]:
        """Parsing ottimizzato per grandi volumi di dati"""
        fields = {}
        position = 0

        # Read MTI (sempre 4 bytes)
        mti = bytes(mm[position:position+4])
        position += 4

        # Read and process bitmap
        self.bitmap_cache[0:8] = mm[position:position+8]
        present_fields = self._fast_bitmap_scan(self.bitmap_cache)
        position += 8

        # Process fields
        for field_num in present_fields:
            length = self.field_lengths[field_num]
            if length > 0:  # Fixed length
                end = position + length
                fields[field_num] = mm[position:end]
                position = end
            else:  # Variable length
                length = int(mm[position:position+2])
                position += 2
                fields[field_num] = mm[position:position+length]
                position += length

        return fields

    def _fast_bitmap_scan(self, bitmap: array) -> List[int]:
        """Scansione bitmap ottimizzata usando lookup table"""
        present = []
        # Lookup table pre-calcolata per bit processing
        for byte_index, byte in enumerate(bitmap[:8]):
            if byte == 0:  # Skip empty bytes
                continue
            for bit_index in range(8):
                if byte & (1 << (7 - bit_index)):
                    field_num = byte_index * 8 + bit_index + 1
                    present.append(field_num)
        return present

import pytest
from typing import Dict, List
from datetime import datetime

class TestISO8583:
    @pytest.fixture
    def parser(self):
        return ISO8583Parser()

    @pytest.fixture
    def builder(self):
        return ISO8583Builder()

    def test_parse_mti(self, parser):
        """Test parsing of Message Type Indicator"""
        raw_message = b'0100' + b'\x00' * 20  # MTI seguito da padding
        message = parser.parse(raw_message)
        assert message.mti == '0100'

    def test_parse_bitmap(self, parser):
        """Test parsing della bitmap primaria"""
        # Bitmap con campi 2 e 7 presenti
        bitmap = b'\x42\x00\x00\x00\x00\x00\x00\x00'
        raw_message = b'0100' + bitmap + b'test'
        message = parser.parse(raw_message)
        assert 2 in message.fields
        assert 7 in message.fields

    def test_parse_full_message(self, parser):

def test_end_to_end_flow():
    # Setup
    acquirer = AcquirerSimulator()
    network = NetworkSimulator()
    issuer = IssuerSimulator()

    # Test transaction flow
    response = process_transaction(
        acquirer=acquirer,
        network=network,
        issuer=issuer,
        amount=100.00,
        pan="4111111111111111"
    )

    assert response.is_approved()