`services/ingestion/naupa3.py`

128 lines
"""NAUPA III XML parser adapter — the third ingestion format.

NAUPA III replaces NAUPA II fixed-width with XML validated against an XSD. Real element
names/namespaces vary by jurisdiction version; this is a representative, namespace-TOLERANT
subset sufficient to prove the adapter and its security posture on synthetic data.

Registered lazily by ingest._get_parser under format_name 'naupa3_v1'. Emits the same
CanonicalProperty as the CSV/NAUPA II adapters.

SECURITY — XXE / entity-expansion defense:
  Legitimate NAUPA III feeds are XSD-validated DATA documents with NO DTD. We REJECT any
  document containing a DOCTYPE/ENTITY declaration before parsing, which closes external-
  entity (XXE) and billion-laughs expansion attacks at the door. Production should also use
  defusedxml. Python's stdlib ElementTree does not resolve external entities, but a malicious
  internal-entity bomb is still worth refusing outright.
"""
from __future__ import annotations

import hashlib
import re
from typing import Iterable
from xml.etree import ElementTree as ET

from services.common.normalize import (
    amount_band, mask_name, normalize_business, normalize_postal, normalize_text,
    parse_decimal,
)

# Case-insensitive scan for a DTD/entity declaration anywhere in the document prolog/body.
_DTD_RE = re.compile(rb"<!\s*(DOCTYPE|ENTITY)", re.IGNORECASE)


class XmlSecurityError(ValueError):
    """The XML declared a DTD/ENTITY — refused to protect against XXE / entity expansion."""


def _localname(tag: str) -> str:
    return tag.split("}", 1)[-1] if "}" in tag else tag


def _find_text(elem: ET.Element, localname: str) -> str:
    """First descendant whose local (namespace-stripped) name matches; '' if absent."""
    for child in elem.iter():
        if _localname(child.tag) == localname:
            return (child.text or "").strip()
    return ""


def _looks_like_business(name: str) -> bool:
    from services.common.normalize import CORPORATE_SUFFIXES
    return any(tok in CORPORATE_SUFFIXES for tok in normalize_text(name).split())


def build_naupa3_xml(records: list[dict], namespace: str | None = None) -> bytes:
    """Assemble a valid NAUPA-III-style XML document (used by tests/tools).

    If `namespace` is given, it's applied as the default xmlns so the parser's
    namespace-tolerance is exercised.
    """
    ns_attr = f' xmlns="{namespace}"' if namespace else ""
    parts = [f"<UnclaimedProperty{ns_attr}>"]
    for r in records:
        parts.append("  <Property>")
        parts.append(f"    <PropertyId>{r.get('property_id','')}</PropertyId>")
        parts.append(f"    <HolderName>{r.get('holder_name','')}</HolderName>")
        parts.append("    <Owner>")
        parts.append(f"      <FirstName>{r.get('first_name','')}</FirstName>")
        parts.append(f"      <LastName>{r.get('last_name','')}</LastName>")
        parts.append("    </Owner>")
        parts.append("    <Address>")
        parts.append(f"      <Street>{r.get('street','')}</Street>")
        parts.append(f"      <City>{r.get('city','')}</City>")
        parts.append(f"      <State>{r.get('state','')}</State>")
        parts.append(f"      <Zip>{r.get('zip','')}</Zip>")
        parts.append("    </Address>")
        parts.append(f"    <PropertyType>{r.get('property_type','')}</PropertyType>")
        parts.append(f"    <Amount>{r.get('amount','')}</Amount>")
        parts.append("  </Property>")
    parts.append("</UnclaimedProperty>")
    return ("\n".join(parts) + "\n").encode("utf-8")


def parse_naupa3_feed(data: bytes, jurisdiction: str) -> Iterable["object"]:
    from services.ingestion.ingest import CanonicalProperty  # lazy: avoid import cycle

    if _DTD_RE.search(data):
        raise XmlSecurityError(
            "XML declares a DOCTYPE/ENTITY — refused (XXE / entity-expansion protection)"
        )

    root = ET.fromstring(data)  # no external-entity resolution in stdlib ElementTree
    for prop in root.iter():
        if _localname(prop.tag) != "Property":
            continue
        pid = _find_text(prop, "PropertyId")
        first = _find_text(prop, "FirstName")
        last = _find_text(prop, "LastName")
        owner_name = f"{first} {last}".strip()
        is_business = _looks_like_business(owner_name)
        amount_raw = _find_text(prop, "Amount")
        try:
            amount = parse_decimal(amount_raw)
        except ValueError:
            amount = None
        owner_norm = (
            normalize_business(owner_name) if is_business else normalize_text(owner_name)
        )
        # canonical raw payload: the serialized element (stable across whitespace/ns)
        raw = ET.tostring(prop, encoding="unicode")
        raw_hash = hashlib.sha256(raw.encode("utf-8")).hexdigest()
        yield CanonicalProperty(
            jurisdiction=jurisdiction,
            source_property_id=pid,
            holder_name_raw=_find_text(prop, "HolderName"),
            owner_type="business" if is_business else "person",
            owner_name_raw=owner_name,
            owner_name_normalized=owner_norm,
            owner_name_masked=mask_name(owner_name),
            amount=amount,
            amount_band=amount_band(amount),
            raw_payload=raw,
            raw_record_hash=raw_hash,
            city_normalized=normalize_text(_find_text(prop, "City")) or None,
            region=normalize_text(_find_text(prop, "State")) or None,
            postal_code=normalize_postal(_find_text(prop, "Zip")),
            property_type=normalize_text(_find_text(prop, "PropertyType")) or None,
        )