# Copyright (C) 2023-2026 Pain001. All rights reserved.
# SPDX-License-Identifier: Apache-2.0 OR MIT
#
# Licensed under either of the Apache License, Version 2.0 or the MIT
# License, at your option. You may not use this file except in
# compliance with one of those licences. Copies are provided in
# LICENSE-APACHE and LICENSE-MIT.
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the Licences is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied. See the applicable Licence for the specific language
# governing permissions and limitations.
# pylint: disable=duplicate-code
"""Generate and XSD-validate ISO 20022 XML payment messages."""
import logging
import os
import re
import time
import warnings
from collections.abc import Iterator
from decimal import Decimal, InvalidOperation
from typing import Any
from jinja2 import Template, select_autoescape
from jinja2.sandbox import SandboxedEnvironment
from pain001.exceptions import PaymentValidationError
from pain001.observability import emit_metric_event
from pain001.observability.otel import set_span_attributes, traced
from pain001.security import validate_path
from pain001.xml.generate_updated_xml_file_path import (
generate_updated_xml_file_path,
)
from pain001.xml.message_registry import MESSAGE_REGISTRY, prepare_xml_data
from pain001.xml.validate_via_xsd import (
collect_xsd_validation_errors,
validate_xml_string_via_xsd,
)
logger = logging.getLogger(__name__)
# Templates are trusted, but these directives would let one pull in
# arbitrary files from disk, so they are rejected before rendering.
_TEMPLATE_DIRECTIVE_PATTERN = re.compile(
r"{%\s*(include|import|from|extends)\b", re.IGNORECASE
)
# ISO 20022 amounts carry at most two decimal places.
_AMOUNT_EXPONENT = Decimal("0.01")
# Common caller-side field aliases -> canonical field names. LLM agents and
# ad-hoc integrations naturally say "amount"/"currency"; the schemas and
# templates use payment_amount / payment_currency (the JSON Schemas document
# "currency", the templates read "payment_currency" - both are canonical
# spellings of the same value, so the pair is mirrored bidirectionally).
_FIELD_ALIASES: dict[str, str] = {
"amount": "payment_amount",
"instructed_amount": "payment_amount",
"payment_currency": "currency",
"currency": "payment_currency",
"execution_date": "requested_execution_date",
}
# Keys whose canonical spelling carries an upper-case identifier suffix;
# callers frequently produce the all-lower-case form.
_IDENTIFIER_KEYS: dict[str, str] = {
"debtor_account_iban": "debtor_account_IBAN",
"creditor_account_iban": "creditor_account_IBAN",
"debtor_agent_bic": "debtor_agent_BIC",
"creditor_agent_bic": "creditor_agent_BIC",
"forwarding_agent_bic": "forwarding_agent_BIC",
"charge_agent_bicfi": "charge_agent_BICFI",
"creditor_agent_bicfi": "creditor_agent_BICFI",
"debtor_agent_account_iban": "debtor_agent_account_IBAN",
"charge_account_iban": "charge_account_IBAN",
}
_DATE_ONLY_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$")
_DATETIME_SPACE_RE = re.compile(r"^\d{4}-\d{2}-\d{2} \d{2}:\d{2}(:\d{2})?$")
_DATETIME_RE = re.compile(r"^\d{4}-\d{2}-\d{2}[T ]")
# Fields rendered into xs:dateTime XML content (e.g. GrpHdr/CreDtTm): a bare
# date is coerced to midnight so "2026-07-18" does not fail XSD validation.
_DATETIME_FIELDS = ("date",)
# Fields rendered into xs:date XML content (e.g. ReqdExctnDt/Dt): a full
# datetime is truncated to its date part.
_DATE_FIELDS = ("requested_execution_date", "reference_date")
# Fields rendered into xs:boolean XML content.
_BOOLEAN_FIELDS = ("batch_booking",)
def _canonicalize_keys(row: dict[str, Any]) -> dict[str, Any]:
"""Return a copy of ``row`` with alias keys mapped to canonical names.
An alias never overwrites an explicitly-provided canonical key with a
non-empty value.
Args:
row: The raw input row.
Returns:
A new dict whose keys are the canonical field names.
"""
updated: dict[str, Any] = dict(row)
for key, value in row.items():
canonical = _IDENTIFIER_KEYS.get(key.lower())
if canonical is None:
canonical = _FIELD_ALIASES.get(key)
if canonical is None or canonical == key:
continue
existing = updated.get(canonical)
if existing is None or str(existing).strip() == "":
updated[canonical] = value
return updated
def _coerce_temporal_and_boolean(row: dict[str, Any]) -> dict[str, Any]:
"""Coerce date/datetime/boolean values into their XSD lexical forms.
Args:
row: A canonical-key row (mutated in place and returned).
Returns:
The same row with temporal and boolean fields in XSD form.
"""
for key in _DATETIME_FIELDS:
value = row.get(key)
if isinstance(value, str):
text = value.strip()
if _DATE_ONLY_RE.match(text):
row[key] = f"{text}T00:00:00"
elif _DATETIME_SPACE_RE.match(text):
row[key] = text.replace(" ", "T", 1)
for key in _DATE_FIELDS:
value = row.get(key)
if isinstance(value, str):
text = value.strip()
if _DATETIME_RE.match(text):
row[key] = text[:10]
for key in _BOOLEAN_FIELDS:
value = row.get(key)
if isinstance(value, str) and value.strip().lower() in (
"true",
"false",
):
row[key] = value.strip().lower()
return row
def _format_amount(value: Any, row_index: int) -> str:
"""Normalize a monetary amount to an exact two-decimal string.
Amounts are handled with Decimal end-to-end: floats are converted via
their shortest string representation so binary artifacts are surfaced
(and rejected) rather than silently rounded into the payment file.
Args:
value: The raw amount from the input row.
row_index: 1-based row number, used in error messages.
Returns:
The amount as an exact two-decimal string (e.g. "10.00").
Raises:
PaymentValidationError: If the amount is missing, not a number,
not positive, or carries more than two decimal places.
"""
if value is None or (isinstance(value, str) and not value.strip()):
raise PaymentValidationError(
f"Row {row_index}: payment_amount is required",
field="payment_amount",
)
try:
amount = Decimal(str(value).strip())
except InvalidOperation as e:
raise PaymentValidationError(
f"Row {row_index}: payment_amount {value!r} is not a number",
field="payment_amount",
) from e
if not amount.is_finite() or amount <= 0:
raise PaymentValidationError(
f"Row {row_index}: payment_amount must be a positive amount, "
f"got {value!r}",
field="payment_amount",
)
quantized = amount.quantize(_AMOUNT_EXPONENT)
if amount != quantized:
raise PaymentValidationError(
f"Row {row_index}: payment_amount {value!r} has more than two "
"decimal places; round it explicitly before generating",
field="payment_amount",
)
return f"{quantized:.2f}"
def _normalize_financial_fields(
data: list[dict[str, Any]],
) -> tuple[list[dict[str, Any]], str, str]:
"""Normalize amounts and booleans, computing batch totals from rows.
Args:
data: Payment rows, each carrying a "payment_amount" value.
Returns:
Tuple of (normalized rows, nb_of_txs, ctrl_sum) where amounts are
two-decimal strings, booleans are XSD-style "true"/"false", and
nb_of_txs/ctrl_sum are computed from the data instead of trusting
caller-provided header fields.
"""
normalized: list[dict[str, Any]] = []
total = Decimal("0.00")
for index, row in enumerate(data, start=1):
updated = _canonicalize_keys(row)
formatted = _format_amount(updated.get("payment_amount"), index)
total += Decimal(formatted)
# XSD xs:boolean only accepts "true"/"false"; Python bools from
# typed sources (JSON, SQLite) would otherwise render as "True".
updated = {
key: (
("true" if value else "false")
if isinstance(value, bool)
else value
)
for key, value in updated.items()
}
updated = _coerce_temporal_and_boolean(updated)
updated["payment_amount"] = formatted
normalized.append(updated)
for updated in normalized:
# Batch totals are always computed from the rows; caller-supplied
# values are overwritten so the message stays internally consistent
# and the header fields never have to be provided by hand. Kept as
# int/float so the rows also satisfy the JSON Schemas' types.
updated["nb_of_txs"] = len(normalized)
updated["ctrl_sum"] = float(total)
return normalized, str(len(normalized)), f"{total:.2f}"
[docs]
def canonicalize_payment_record(row: dict[str, Any]) -> dict[str, Any]:
"""Map a record's alias keys to canonical field names, preserving values.
The key-mapping half of :func:`normalize_payment_records`: ``amount``
becomes ``payment_amount``, ``currency`` and ``payment_currency``
mirror each other, and lower-case ``*_iban`` / ``*_bic`` spellings are
canonicalized - without reformatting any value. Use this before
JSON-Schema validation, where typed values (numbers, booleans) must
keep their types.
Args:
row: The raw input record.
Returns:
A new dict with canonical keys; the input is not mutated.
"""
return _canonicalize_keys(row)
[docs]
def normalize_payment_records(
data: list[dict[str, Any]],
) -> list[dict[str, Any]]:
"""Normalize caller records into the canonical pain.001 input shape.
Applies the same ergonomics the generators use internally, so records
can also be pre-normalized before JSON-Schema validation:
* field aliases (``amount`` -> ``payment_amount``, ``currency`` <->
``payment_currency``, lower-case IBAN/BIC key spellings, ...);
* amounts formatted as exact two-decimal strings;
* Python/JSON booleans (and ``"True"``/``"FALSE"`` strings) rendered
in XSD form (``"true"``/``"false"``);
* bare dates coerced to the XSD lexical form each field requires
(``date`` -> ``YYYY-MM-DDT00:00:00``, ``requested_execution_date``
truncated to ``YYYY-MM-DD``);
* ``nb_of_txs`` and ``ctrl_sum`` computed from the rows themselves.
Args:
data: The raw payment rows.
Returns:
A new list of normalized rows; the input is not mutated.
A ``PaymentValidationError`` from amount normalization (missing,
non-numeric, non-positive, or over-precise payment amounts)
propagates unchanged.
"""
normalized, _, _ = _normalize_financial_fields(data)
return normalized
def _load_trusted_template_source(xml_template_path: str) -> str:
"""Read a trusted template and reject filesystem-expanding directives."""
if not str(xml_template_path).endswith(".xml"):
raise ValueError("Template path must point to an .xml file")
with open(xml_template_path, encoding="utf-8") as handle: # nosec B108
template_source = handle.read()
if _TEMPLATE_DIRECTIVE_PATTERN.search(template_source):
raise ValueError(
"Template contains disabled Jinja filesystem directives"
)
return template_source
def _validate_generation_inputs(
data: list[dict[str, Any]],
payment_initiation_message_type: str,
xml_template_path: str,
) -> str:
"""Validate message type, data presence, and template path.
Args:
data: List of dictionaries containing payment data.
payment_initiation_message_type: Message type (e.g., "pain.001.001.03").
xml_template_path: Path to the Jinja2 XML template file.
Returns:
The validated template path.
Raises:
ValueError: If message type is invalid, data is empty, or the
template path fails validation.
"""
if payment_initiation_message_type not in MESSAGE_REGISTRY:
raise ValueError(
f"Invalid XML message type: {payment_initiation_message_type}"
)
if not data:
raise ValueError("No data to process - data list is empty")
try:
return str(validate_path(xml_template_path, must_exist=True))
except Exception as e:
raise ValueError(f"Invalid template path: {e}") from e
def _prepare_template_and_data(
data: list[dict[str, Any]],
payment_initiation_message_type: str,
safe_template_path: str,
) -> tuple[Template, dict[str, Any], int]:
"""Prepare normalized XML template and data dictionary.
Args:
data: List of dictionaries containing payment data.
payment_initiation_message_type: Message type (e.g., "pain.001.001.03").
safe_template_path: Pre-validated path to XML template file.
Returns:
A tuple of (compiled Jinja2 template, prepared XML data dict, row count).
"""
# Normalize amounts (Decimal, 2dp) and compute batch totals from the
# rows themselves — header fields like NbOfTxs/CtrlSum must never be
# trusted from input or the message becomes internally inconsistent.
normalized_data, nb_of_txs, ctrl_sum = _normalize_financial_fields(data)
# Prepare XML data using the registry-driven pipeline
prepare_started = time.time()
xml_data = prepare_xml_data(
normalized_data, payment_initiation_message_type
)
xml_data["nb_of_txs"] = nb_of_txs
if "payment_nb_of_txs" in xml_data:
xml_data["payment_nb_of_txs"] = nb_of_txs
if "ctrl_sum" in xml_data:
xml_data["ctrl_sum"] = ctrl_sum
emit_metric_event(
"xml_prepared",
message_type=payment_initiation_message_type,
record_count=len(normalized_data),
duration_ms=int((time.time() - prepare_started) * 1000),
)
template_source = _load_trusted_template_source(safe_template_path)
env = SandboxedEnvironment(
autoescape=select_autoescape(
enabled_extensions=("xml",),
default_for_string=True,
),
)
template = env.from_string(template_source)
return template, xml_data, len(normalized_data)
[docs]
@traced("pain001.stream")
def stream_xml_chunks(
data: list[dict[str, Any]],
payment_initiation_message_type: str,
xml_template_path: str,
buffer_size: int = 10,
) -> Iterator[str]:
"""Stream ISO 20022 pain.001 XML content as string chunks (generator).
Provides an iterator of string chunks with constant memory overhead,
enabling O(1) memory serialization of large payment batches directly
to disk or network sockets without materializing the full XML document
in memory.
Args:
data: List of dictionaries containing payment data.
payment_initiation_message_type: Message type (e.g., "pain.001.001.03").
xml_template_path: Path to the Jinja2 XML template file.
buffer_size: Buffer size for Jinja template streaming (default 10).
Yields:
str: Successive string chunks of the rendered XML document.
A ``ValueError`` from invalid message type, empty data, or invalid
template path propagates unchanged.
"""
safe_template_path = _validate_generation_inputs(
data, payment_initiation_message_type, xml_template_path
)
template, xml_data, row_count = _prepare_template_and_data(
data, payment_initiation_message_type, safe_template_path
)
set_span_attributes(
**{
"pain001.message_type": payment_initiation_message_type,
"pain001.row_count": row_count,
"pain001.streaming": True,
}
)
render_started = time.time()
stream = template.stream(**xml_data)
if buffer_size >= 5:
stream.enable_buffering(buffer_size)
yield from stream
emit_metric_event(
"xml_stream_rendered",
message_type=payment_initiation_message_type,
duration_ms=int((time.time() - render_started) * 1000),
)
[docs]
@traced("pain001.write.stream")
def stream_xml_to_file(
data: list[dict[str, Any]],
payment_initiation_message_type: str,
xml_template_path: str,
output_path: str,
buffer_size: int = 10,
) -> str:
"""Stream ISO 20022 pain.001 XML directly to a file with constant memory.
Args:
data: List of dictionaries containing payment data.
payment_initiation_message_type: Message type identifier.
xml_template_path: Path to the Jinja2 XML template file.
output_path: Destination filesystem path for the XML file.
buffer_size: Buffer size for streaming chunks (default 10).
Returns:
The canonical path the XML file was written to.
A ``ValueError`` from invalid message type, empty data, or invalid
template path propagates unchanged.
"""
safe_xml_path = os.path.realpath(str(output_path))
os.makedirs(os.path.dirname(safe_xml_path) or ".", exist_ok=True)
chunks = stream_xml_chunks(
data,
payment_initiation_message_type,
xml_template_path,
buffer_size=buffer_size,
)
total_bytes = 0
with open(safe_xml_path, "w", encoding="utf-8") as handle:
for chunk in chunks:
handle.write(chunk)
total_bytes += len(chunk.encode("utf-8"))
emit_metric_event(
"xml_generated",
message_type=payment_initiation_message_type,
output_path=str(safe_xml_path),
file_size_bytes=total_bytes,
streaming=True,
)
logger.info(
"XML file streamed to %s (%d bytes)", safe_xml_path, total_bytes
)
return safe_xml_path
[docs]
@traced("pain001.render")
def generate_xml_string(
data: list[dict[str, Any]],
payment_initiation_message_type: str,
xml_template_path: str,
xsd_schema_path: str,
) -> str:
"""Generate ISO 20022 pain.001 XML content as a string (in-memory).
This function is ideal for serverless architectures, REST APIs, and
microservices where XML needs to be returned without writing to disk.
Args:
data: List of dictionaries containing payment data.
payment_initiation_message_type: Message type (e.g., "pain.001.001.03").
xml_template_path: Path to the Jinja2 XML template file.
xsd_schema_path: Path to XSD schema file for validation.
Returns:
str: The generated and validated XML content.
Raises:
ValueError: If message type is invalid, data is empty, the
template or schema path fails validation, or the template
contains disabled Jinja filesystem directives.
RuntimeError: If XML validation fails against XSD schema.
PaymentValidationError from amount normalization (missing,
non-numeric, non-positive, or over-precise payment_amount values)
propagates unchanged.
Examples:
>>> data = [{"id": "MSG001", "date": "2026-01-15", ...}]
>>> xml_str = generate_xml_string(
... data,
... "pain.001.001.03",
... "templates/pain.001.001.03/template.xml",
... "templates/pain.001.001.03/pain.001.001.03.xsd"
... ) # doctest: +SKIP
>>> xml_str.startswith('<?xml')
True
"""
safe_template_path = _validate_generation_inputs(
data,
payment_initiation_message_type,
xml_template_path,
)
# Validate schema path
try:
xsd_schema_path = validate_path(xsd_schema_path, must_exist=True)
except Exception as e:
raise ValueError(f"Invalid schema path: {e}") from e
template, xml_data, row_count = _prepare_template_and_data(
data,
payment_initiation_message_type,
safe_template_path,
)
set_span_attributes(
**{
"pain001.message_type": payment_initiation_message_type,
"pain001.row_count": row_count,
}
)
# Render the template to string
render_started = time.time()
xml_content = template.render(**xml_data)
emit_metric_event(
"xml_rendered",
message_type=payment_initiation_message_type,
duration_ms=int((time.time() - render_started) * 1000),
)
# Validate the XML content against the XSD schema
validation_started = time.time()
is_valid = validate_xml_string_via_xsd(xml_content, xsd_schema_path)
if not is_valid:
emit_metric_event(
"validation_failed",
message_type=payment_initiation_message_type,
schema_path=str(xsd_schema_path),
)
# Report every violation at once (element path + reason) so a
# caller can fix all inputs in a single retry instead of probing
# one opaque failure per attempt.
reasons = collect_xsd_validation_errors(xml_content, xsd_schema_path)
detail = "; ".join(reasons) if reasons else "unknown reason"
raise RuntimeError(
f"Generated XML failed validation against "
f"{xsd_schema_path}: {detail}"
)
emit_metric_event(
"xsd_validation_passed",
message_type=payment_initiation_message_type,
schema_path=str(xsd_schema_path),
duration_ms=int((time.time() - validation_started) * 1000),
)
return str(xml_content)
[docs]
@traced("pain001.write")
def generate_xml(
data: list[dict[str, Any]],
payment_initiation_message_type: str,
xml_file_path: str,
xsd_file_path: str,
output_path: str | None = None,
envelop_bah: bool = False,
bah_header: Any | None = None,
xml_sign_key: str | bytes | None = None,
xml_sign_cert: str | bytes | None = None,
xml_sign_passphrase: str | bytes | None = None,
) -> str:
"""Generates an ISO 20022 pain.001 XML file from input data.
This function writes XML to a file. For in-memory XML generation
(serverless/API use cases), use generate_xml_string() instead.
Args:
data: List of dictionaries containing payment data
payment_initiation_message_type: Message type identifier. Any
supported version from "pain.001.001.03" through
"pain.001.001.12", or "pain.008.001.02" for direct debits.
xml_file_path: Path to the Jinja2 XML template file
xsd_file_path: Path to XML schema file for validation
output_path: Explicit path to write the generated XML file to.
Parent directories are created if needed. When omitted, the
file is written next to the template (deprecated behavior
that only works when the template lives under the current
working directory).
envelop_bah: If True, wrap generated XML in an ISO 20022 BAH envelope.
bah_header: Optional customized BusinessApplicationHeader instance.
xml_sign_key: Optional RSA private key in PEM format to sign the XML.
xml_sign_cert: Optional X.509 certificate in PEM format for KeyInfo.
xml_sign_passphrase: Optional passphrase for the private key.
Returns:
The path the XML file was written to.
Raises:
ValueError: If message type is invalid, data is empty, or the
output path fails validation. RuntimeError from XSD
validation in generate_xml_string propagates unchanged.
"""
# Generate XML content as string
xml_content = generate_xml_string(
data, payment_initiation_message_type, xml_file_path, xsd_file_path
)
if envelop_bah or bah_header is not None:
from pain001.xml.envelope import (
BusinessApplicationHeader,
envelop_iso20022_message,
)
hdr = bah_header or BusinessApplicationHeader()
xml_content = envelop_iso20022_message(xml_content, hdr)
if xml_sign_key is not None:
from pain001.security.dsig import sign_xml_document
xml_content = sign_xml_document(
xml_content,
private_key_pem=xml_sign_key,
cert_pem=xml_sign_cert,
passphrase=xml_sign_passphrase,
)
if output_path is not None:
safe_xml_path = os.path.realpath(str(output_path))
os.makedirs(os.path.dirname(safe_xml_path) or ".", exist_ok=True)
else:
warnings.warn(
"Calling generate_xml without output_path writes next to the "
"template and requires it to be under the current working "
"directory; pass output_path explicitly instead.",
DeprecationWarning,
stacklevel=2,
)
# Legacy behavior: derive the output path from the template path.
updated_xml_file_path = generate_updated_xml_file_path(
xml_file_path, payment_initiation_message_type
)
try:
safe_xml_path = validate_path(updated_xml_file_path) # nosec B108
except (
Exception
) as e: # pragma: no cover - the template path was already validated
raise ValueError(f"Path validation failed: {e}") from e
# Explicit startswith guard for CodeQL CWE-22 sanitiser recognition.
cwd_prefix = str(os.path.realpath(os.getcwd()))
if not safe_xml_path.startswith(
cwd_prefix + os.sep
): # pragma: no cover - CodeQL barrier; validate_path enforces it
raise ValueError(
f"Output path outside working directory: {safe_xml_path}"
)
from pain001.plugins.registry import registry as plugin_registry
writer = plugin_registry.get_writer("xml-file")
if writer is None:
raise ValueError("The xml-file writer plugin is disabled")
safe_xml_path = writer.write(xml_content, safe_xml_path)
emit_metric_event(
"xml_generated",
message_type=payment_initiation_message_type,
output_path=str(safe_xml_path),
file_size_bytes=len(xml_content.encode("utf-8")),
)
logger.info("XML file created at %s", safe_xml_path)
return safe_xml_path