diff --git a/backend/services/hwpx_recognition.py b/backend/services/hwpx_recognition.py
index 325f5417c..a5b6292b9 100644
--- a/backend/services/hwpx_recognition.py
+++ b/backend/services/hwpx_recognition.py
@@ -1,4 +1,11 @@
-"""Safely extract bounded text and graph records from an HWPX package."""
+"""Safely recognize HWPX section text into Naruon's content graph.
+
+The importer retains HWPX bytes for deferred processing. This module performs the
+next deterministic worker-side boundary: it revalidates the ZIP package,
+follows the OPF spine order from ``Contents/content.hpf``, parses section XML
+with defused XML semantics, and emits paragraph-level provenance without
+extracting files, executing active content, or fetching external resources.
+"""
from __future__ import annotations
@@ -7,144 +14,292 @@
import re
import zipfile
from dataclasses import dataclass
+from pathlib import PurePosixPath
+from xml.etree.ElementTree import ParseError
-from defusedxml import ElementTree
-from services.content_graph import ParseResult, PdfDomSection, parse_pdf_dom
+from defusedxml import ElementTree as DefusedElementTree
+from defusedxml.common import DefusedXmlException
-HWPX_CONTENT_TYPE = "application/hwp+zip"
-MAX_HWPX_SECTION_XML_BYTES = 8 * 1024 * 1024
-MAX_HWPX_TOTAL_XML_BYTES = 32 * 1024 * 1024
-MAX_HWPX_TEXT_CHARS = 1_000_000
-_SECTION_NAME_PATTERN = re.compile(r"Contents/section[0-9]+\.xml\Z")
-_FORBIDDEN_XML_DECLARATION = re.compile(
- br" str:
+ """Return the local XML name without trusting a particular namespace URI."""
+
+ return tag.rsplit("}", 1)[-1]
+
+
+def _safe_member_name(name: str) -> str:
+ """Validate a ZIP member as one normalized package-internal POSIX path."""
+ if not name or "\\" in name or "\x00" in name:
+ raise ValueError("HWPX package contains an unsafe member path")
+ path = PurePosixPath(name)
+ if path.is_absolute() or any(part in {"", ".", ".."} for part in path.parts):
+ raise ValueError("HWPX package contains an unsafe member path")
+ return path.as_posix()
-def recognize_hwpx(
+
+def _package_entries(archive: zipfile.ZipFile) -> dict[str, zipfile.ZipInfo]:
+ """Return unique, unencrypted HWPX members within metadata budgets."""
+
+ entries = archive.infolist()
+ if not 0 < len(entries) <= MAX_HWPX_PACKAGE_ENTRIES:
+ raise ValueError("HWPX package exceeds the member-count limit")
+
+ result: dict[str, zipfile.ZipInfo] = {}
+ aggregate_name_bytes = 0
+ for entry in entries:
+ name = _safe_member_name(entry.filename)
+ aggregate_name_bytes += len(name.encode("utf-8", errors="surrogatepass"))
+ if aggregate_name_bytes > MAX_HWPX_MEMBER_NAME_BYTES:
+ raise ValueError("HWPX package exceeds the member-name metadata limit")
+ if name in result:
+ raise ValueError("HWPX package contains duplicate member paths")
+ if entry.flag_bits & 0x1:
+ raise ValueError("HWPX package contains encrypted members")
+ result[name] = entry
+ return result
+
+
+def _read_member(
+ archive: zipfile.ZipFile,
+ entry: zipfile.ZipInfo,
*,
- hwpx_bytes: bytes,
+ label: str,
+ max_bytes: int = MAX_HWPX_XML_MEMBER_BYTES,
+) -> bytes:
+ """Read one already-selected XML member within its expansion budget."""
+
+ if entry.is_dir() or entry.file_size > max_bytes:
+ raise ValueError(f"HWPX {label} XML member exceeds the expansion limit")
+ try:
+ payload = archive.read(entry)
+ except (NotImplementedError, zipfile.BadZipFile) as exc:
+ raise ValueError(f"HWPX {label} XML member could not be read") from exc
+ if len(payload) != entry.file_size or len(payload) > max_bytes:
+ raise ValueError(f"HWPX {label} XML member exceeds the expansion limit")
+ return payload
+
+
+def _parse_xml(payload: bytes, *, label: str):
+ """Parse one bounded XML member with entity/DTD defenses enabled."""
+
+ try:
+ return DefusedElementTree.fromstring(payload)
+ except (DefusedXmlException, ParseError, ValueError) as exc:
+ raise ValueError(f"HWPX {label} contains unsafe XML") from exc
+
+
+def _resolve_manifest_href(href: str) -> str:
+ """Resolve an OPF item href only to the standard HWPX section namespace."""
+
+ value = href.strip()
+ if not value or "\\" in value or "\x00" in value or "?" in value or "#" in value:
+ raise ValueError("HWPX content.hpf contains an unsafe manifest href")
+ raw_path = PurePosixPath(value)
+ if raw_path.is_absolute() or any(part in {"", ".", ".."} for part in raw_path.parts):
+ raise ValueError("HWPX content.hpf contains an unsafe manifest href")
+
+ if raw_path.parts and raw_path.parts[0] == "Contents":
+ resolved = raw_path.as_posix()
+ else:
+ resolved = (PurePosixPath("Contents") / raw_path).as_posix()
+ if not _SECTION_PATH_RE.fullmatch(resolved):
+ raise ValueError("HWPX content.hpf contains a non-section spine target")
+ return resolved
+
+
+def _section_paths_from_spine(content_hpf_root) -> tuple[str, ...]:
+ """Resolve section member paths using OPF manifest identity and spine order."""
+
+ manifest_element = next(
+ (child for child in content_hpf_root if _local_name(child.tag) == "manifest"),
+ None,
+ )
+ spine_element = next(
+ (child for child in content_hpf_root if _local_name(child.tag) == "spine"),
+ None,
+ )
+ if manifest_element is None or spine_element is None:
+ raise ValueError("HWPX content.hpf is missing manifest or spine metadata")
+
+ manifest: dict[str, str] = {}
+ for item in manifest_element:
+ if _local_name(item.tag) != "item":
+ continue
+ item_id = (item.get("id") or "").strip()
+ href = (item.get("href") or "").strip()
+ if not item_id or not href or item_id in manifest:
+ raise ValueError("HWPX content.hpf contains ambiguous manifest identity")
+ manifest[item_id] = href
+
+ section_paths: list[str] = []
+ for itemref in spine_element:
+ if _local_name(itemref.tag) != "itemref":
+ continue
+ item_id = (itemref.get("idref") or "").strip()
+ if not item_id or item_id not in manifest:
+ raise ValueError("HWPX spine item cannot be resolved through the manifest")
+ section_paths.append(_resolve_manifest_href(manifest[item_id]))
+
+ if not section_paths or len(section_paths) != len(set(section_paths)):
+ raise ValueError("HWPX spine must reference unique section members")
+ return tuple(section_paths)
+
+
+def _paragraph_text(paragraph) -> str:
+ """Extract text controls from one paragraph without duplicating nested paragraphs."""
+
+ parts: list[str] = []
+
+ def visit(element) -> None:
+ for child in element:
+ local_name = _local_name(child.tag)
+ if local_name == "p":
+ continue
+ if local_name == "t":
+ parts.append("".join(child.itertext()))
+ elif local_name in {"lineBreak", "br"}:
+ parts.append("\n")
+ elif local_name == "tab":
+ parts.append("\t")
+ else:
+ visit(child)
+
+ visit(paragraph)
+ return "".join(parts).strip()
+
+
+def _section_paragraphs(section_root) -> tuple[str, ...]:
+ """Return non-empty paragraphs in document order from one HWPX section."""
+
+ paragraphs: list[str] = []
+ for element in section_root.iter():
+ if _local_name(element.tag) != "p":
+ continue
+ text = _paragraph_text(element)
+ if text:
+ paragraphs.append(text)
+ return tuple(paragraphs)
+
+
+def recognize_hwpx_package(
+ payload: bytes,
+ *,
+ filename: str,
+ source_kind: str,
source_record_uid: str,
- display_name: str = "",
) -> HwpxRecognitionRecords:
- """Extract text from bounded HWPX section XML without executing content.
+ """Recognize bounded HWPX section text and paragraph provenance.
- Only canonical ``Contents/sectionN.xml`` members are read. Encrypted or
- unsupported compression members, XML entity declarations, oversized
- members, and empty documents fail closed before graph records are created.
+ ``Contents/content.hpf`` is the ordering authority: manifest IDs resolve
+ package paths and spine references define reading order. Each selected XML
+ member is bounded before decompression and parsed with ``defusedxml``.
+ Images, OLE objects, external resources, macros, and non-section package
+ members are intentionally not interpreted by this slice.
"""
- sections: list[PdfDomSection] = []
- total_xml_bytes = 0
+
+ if not isinstance(payload, bytes) or not payload.startswith(b"PK"):
+ raise ValueError("Pending attachment payload is not a HWPX package")
+
try:
- with zipfile.ZipFile(io.BytesIO(hwpx_bytes)) as archive:
- section_infos = sorted(
- (
- info
- for info in archive.infolist()
- if _SECTION_NAME_PATTERN.fullmatch(info.filename)
- ),
- key=lambda info: int(
- info.filename.removeprefix("Contents/section").removesuffix(
- ".xml"
- )
- ),
- )
- if not section_infos:
- raise HwpxRecognitionError("HWPX package has no section XML")
- if len({info.filename for info in section_infos}) != len(section_infos):
- raise HwpxRecognitionError("HWPX package has duplicate sections")
-
- for info in section_infos:
- if info.flag_bits & 0x1 or info.compress_type not in (
- zipfile.ZIP_STORED,
- zipfile.ZIP_DEFLATED,
- ):
- raise HwpxRecognitionError("HWPX section compression is not allowed")
- if info.file_size > MAX_HWPX_SECTION_XML_BYTES:
- raise HwpxRecognitionError("HWPX section exceeds the XML size limit")
- total_xml_bytes += info.file_size
- if total_xml_bytes > MAX_HWPX_TOTAL_XML_BYTES:
- raise HwpxRecognitionError("HWPX XML exceeds the total size limit")
- xml_bytes = archive.read(info)
- if len(xml_bytes) != info.file_size:
- raise HwpxRecognitionError("HWPX section size changed while reading")
- sections.extend(_parse_section_xml(xml_bytes))
- except HwpxRecognitionError:
- raise
- except (
- OSError,
- RuntimeError,
- ValueError,
- zipfile.BadZipFile,
- ElementTree.ParseError,
- ) as exc:
- raise HwpxRecognitionError("HWPX package could not be safely read") from exc
-
- paragraphs = tuple(
- paragraph
- for section in sections
- for paragraph in section.paragraphs
- if paragraph.strip()
- )
- parse_text = "\n\n".join(paragraphs)
- if not parse_text.strip():
- raise HwpxRecognitionError("HWPX package contains no readable text")
- if len(parse_text) > MAX_HWPX_TEXT_CHARS:
- raise HwpxRecognitionError("HWPX text exceeds the parse size limit")
+ archive = zipfile.ZipFile(io.BytesIO(payload))
+ except (zipfile.BadZipFile, ValueError) as exc:
+ raise ValueError("Pending attachment payload is not a HWPX package") from exc
+
+ with archive:
+ entries = _package_entries(archive)
+ mimetype_entry = entries.get("mimetype")
+ version_entry = entries.get("version.xml")
+ content_hpf_entry = entries.get(_CONTENT_HPF_PATH)
+ if (
+ mimetype_entry is None
+ or version_entry is None
+ or content_hpf_entry is None
+ ):
+ raise ValueError("HWPX package is missing required identity metadata")
+ mimetype_payload = _read_member(
+ archive,
+ mimetype_entry,
+ label="mimetype",
+ max_bytes=128,
+ )
+ if mimetype_payload != _HWPX_MIMETYPE:
+ raise ValueError("HWPX package has an invalid mimetype member")
+
+ content_hpf_payload = _read_member(
+ archive,
+ content_hpf_entry,
+ label="content.hpf",
+ )
+ content_hpf_root = _parse_xml(content_hpf_payload, label="content.hpf")
+ section_paths = _section_paths_from_spine(content_hpf_root)
+
+ selected_entries: list[zipfile.ZipInfo] = []
+ expanded_total = len(content_hpf_payload)
+ for section_path in section_paths:
+ entry = entries.get(section_path)
+ if entry is None:
+ raise ValueError("HWPX spine section member is missing from the package")
+ if entry.file_size > MAX_HWPX_XML_MEMBER_BYTES:
+ raise ValueError("HWPX section XML member exceeds the expansion limit")
+ expanded_total += entry.file_size
+ if expanded_total > MAX_HWPX_TOTAL_XML_BYTES:
+ raise ValueError("HWPX selected XML exceeds the total expansion limit")
+ selected_entries.append(entry)
+
+ sections: list[PdfDomSection] = []
+ paragraph_count = 0
+ for entry in selected_entries:
+ section_payload = _read_member(archive, entry, label="section")
+ section_root = _parse_xml(section_payload, label="section")
+ paragraphs = _section_paragraphs(section_root)
+ paragraph_count += len(paragraphs)
+ sections.append(PdfDomSection(heading="", paragraphs=paragraphs))
- source_content_hash = hashlib.sha256(hwpx_bytes).hexdigest()
+ if paragraph_count == 0:
+ raise ValueError("HWPX package contains no readable paragraph text")
+
+ source_content_hash = hashlib.sha256(payload).hexdigest()
parse_result = parse_pdf_dom(
- source_kind="attachment",
+ source_kind=source_kind,
source_record_uid=source_record_uid,
sections=sections,
source_content_hash=source_content_hash,
- display_name=display_name,
- content_type=HWPX_CONTENT_TYPE,
+ display_name=filename,
+ content_type=HWPX_PARSE_CONTENT_TYPE,
+ )
+ parse_text = "\n\n".join(
+ segment.safe_text_content
+ for segment in parse_result.segments
+ if segment.segment_kind == "paragraph"
)
return HwpxRecognitionRecords(
parse_text=parse_text,
- source_content_hash=source_content_hash,
parse_result=parse_result,
+ section_count=len(sections),
+ paragraph_count=paragraph_count,
)
-
-
-def _parse_section_xml(xml_bytes: bytes) -> list[PdfDomSection]:
- """Parse one section into paragraph units without resolving declarations."""
- if _FORBIDDEN_XML_DECLARATION.search(xml_bytes):
- raise HwpxRecognitionError("HWPX XML declarations are not allowed")
- root = ElementTree.fromstring(xml_bytes)
- paragraphs: list[str] = []
- for element in root.iter():
- if _local_name(element.tag) != "p":
- continue
- text = "".join(
- node.text or ""
- for node in element.iter()
- if _local_name(node.tag) == "t"
- )
- normalized = " ".join(text.split())
- if normalized:
- paragraphs.append(normalized)
- if not paragraphs:
- return []
- return [PdfDomSection(heading="", paragraphs=tuple(paragraphs))]
-
-
-def _local_name(tag: str) -> str:
- """Return an XML local name while rejecting non-element tags."""
- if not isinstance(tag, str):
- return ""
- return tag.rsplit("}", maxsplit=1)[-1]
diff --git a/backend/services/newsdom_worker.py b/backend/services/newsdom_worker.py
index 469482ecd..951df16f9 100644
--- a/backend/services/newsdom_worker.py
+++ b/backend/services/newsdom_worker.py
@@ -1,18 +1,19 @@
-"""Background worker glue for NewsDOM PDF DOM recognition.
+"""Background worker glue for deferred PDF and HWPX recognition.
-Attachments and workspace documents whose PDF recognition was deferred at
-import time are processed here: the sidecar is called (via
-:mod:`services.newsdom_pdf_recognition`) and the returned tree is landed into
-the persisted attachment/document content and, for attachments, the
-``content_nodes`` / ``content_segments`` graph.
+PDF attachments and workspace documents whose NewsDOM recognition was deferred
+at import time are processed through the configured sidecar. HWPX attachments
+are processed locally through the bounded OWPML recognizer. Both paths land
+recognized attachment text and provenance into ``content_nodes`` /
+``content_segments`` without allowing a pending payload to masquerade as parsed.
The apply functions are deliberately session-free so they can be unit tested
-with in-memory model instances and a mocked NewsDOM client.
+with in-memory model instances and injected adapters.
"""
from __future__ import annotations
import asyncio
+import hashlib
import logging
import random
from collections.abc import Awaitable, Callable
@@ -31,15 +32,15 @@
from db.session import AsyncSessionLocal
from services.attachment_parser import (
HWP_CONVERSION_PENDING_STATUS,
- HWPX_XML_PACKAGE_PENDING_STATUS,
decode_deferred_attachment_payload,
)
from services.content_graph import ParseResult
from services.hwpx_recognition import (
- HWPX_CONTENT_TYPE,
- HwpxRecognitionError,
+ HWPX_FAILED_STATUS,
+ HWPX_PARSED_STATUS,
+ HWPX_PARSE_CONTENT_TYPE,
HwpxRecognitionRecords,
- recognize_hwpx,
+ recognize_hwpx_package,
)
from services.newsdom_client import (
NewsdomConfigurationError,
@@ -62,6 +63,9 @@
logger = logging.getLogger(__name__)
_sysrand = random.SystemRandom()
+HWPX_PENDING_STATUS = "hwpx_xml_package_pending"
+HWPX_PARSER_KEY = "hwpx"
+
# How the worker resolves a runtime config for an organization. Injectable so
# the per-item processors are unit-testable without a database.
ConfigResolver = Callable[
@@ -69,6 +73,20 @@
]
+def _attachment_source_record_uid(email: Email, attachment: Attachment) -> str:
+ """Return opaque provenance derived from stable message and source data."""
+ identity = "\x00".join(
+ (
+ email.message_id or "",
+ email.thread_id or "",
+ attachment.filename or "",
+ attachment.content or "",
+ )
+ )
+ digest = hashlib.sha256(identity.encode("utf-8", errors="surrogatepass"))
+ return f"attachment:{digest.hexdigest()[:32]}"
+
+
def _append_parse_result_to_attachment(
*,
email: Email,
@@ -139,11 +157,12 @@ def apply_hwpx_recognition_to_attachment(
attachment: Attachment,
records: HwpxRecognitionRecords,
) -> None:
- """Land locally extracted HWPX text and graph records on an attachment."""
+ """Land recognized HWPX text and provenance on one attachment."""
+
attachment.content = records.parse_text
- attachment.parse_content_type = HWPX_CONTENT_TYPE
- attachment.parser_key = "hwpx"
- attachment.parse_status = PDF_DOM_RECOGNITION_PARSED_STATUS
+ attachment.parse_content_type = HWPX_PARSE_CONTENT_TYPE
+ attachment.parser_key = HWPX_PARSER_KEY
+ attachment.parse_status = HWPX_PARSED_STATUS
attachment.parse_error_code = None
_append_parse_result_to_attachment(
email=email,
@@ -186,6 +205,29 @@ async def recognize_attachment_pdf(
return records
+def recognize_attachment_hwpx(
+ *,
+ email: Email,
+ attachment: Attachment,
+ hwpx_bytes: bytes,
+ source_record_uid: str,
+) -> HwpxRecognitionRecords:
+ """Recognize a HWPX package and land its ordered text and graph."""
+
+ records = recognize_hwpx_package(
+ hwpx_bytes,
+ filename=attachment.filename or "attachment.hwpx",
+ source_kind="attachment",
+ source_record_uid=source_record_uid,
+ )
+ apply_hwpx_recognition_to_attachment(
+ email=email,
+ attachment=attachment,
+ records=records,
+ )
+ return records
+
+
async def recognize_document_pdf(
*,
document: Document,
@@ -227,6 +269,14 @@ async def recognize_document_pdf(
}
+def _attachment_failed_status(attachment: Attachment) -> str:
+ """Return the parser-family-specific visible failure status."""
+
+ if attachment.parse_status == HWPX_PENDING_STATUS:
+ return HWPX_FAILED_STATUS
+ return PDF_DOM_RECOGNITION_FAILED_STATUS
+
+
async def process_pending_attachment(
*,
session: AsyncSession,
@@ -234,23 +284,59 @@ async def process_pending_attachment(
config_resolver: ConfigResolver = resolve_newsdom_config_from_db,
request_fn: ParseRequestFn = request_pdf_dom,
) -> str:
- """Recognize one pending attachment PDF, or record a safe outcome.
+ """Recognize one pending PDF or HWPX attachment, or record a safe outcome.
- Returns ``RESULT_RECOGNIZED`` on success, ``RESULT_PENDING`` when no active
- provider is configured yet (left pending to retry later), or
- ``RESULT_FAILED`` when the payload or the sidecar response is unusable (a
- visible failure status is recorded - never a false ``parsed``).
+ HWPX recognition is deterministic and local, so it never resolves a
+ NewsDOM provider. PDF recognition retains the provider-aware retry behavior:
+ ``RESULT_PENDING`` means no active provider is usable yet. Invalid retained
+ bytes or recognition failures are always recorded explicitly and never
+ reported as parsed.
"""
email = attachment.email
if email is None:
- attachment.parse_status = PDF_DOM_RECOGNITION_FAILED_STATUS
+ attachment.parse_status = _attachment_failed_status(attachment)
attachment.parse_error_code = "orphan_attachment"
return RESULT_FAILED
- expected_content_type = attachment.parse_content_type or "application/pdf"
if attachment.parse_status == HWP_CONVERSION_PENDING_STATUS:
- # HWP binary conversion requires a separately sandboxed converter. Keep
- # the source pending rather than sending a non-PDF payload to NewsDOM.
+ # Binary HWP needs a separately sandboxed converter and must not be
+ # handed to the PDF or HWPX recognizers.
return RESULT_PENDING
+
+ if attachment.parse_status == HWPX_PENDING_STATUS:
+ try:
+ hwpx_bytes = decode_deferred_attachment_payload(
+ attachment.content,
+ attachment.parse_content_type or HWPX_PARSE_CONTENT_TYPE,
+ )
+ except ValueError as exc:
+ attachment.parse_status = HWPX_FAILED_STATUS
+ attachment.parse_error_code = "invalid_pending_payload"
+ logger.warning(
+ "HWPX attachment %s rejected before recognition: %s",
+ getattr(attachment, "id", "?"),
+ exc,
+ )
+ return RESULT_FAILED
+
+ try:
+ recognize_attachment_hwpx(
+ email=email,
+ attachment=attachment,
+ hwpx_bytes=hwpx_bytes,
+ source_record_uid=_attachment_source_record_uid(email, attachment),
+ )
+ except ValueError as exc:
+ attachment.parse_status = HWPX_FAILED_STATUS
+ attachment.parse_error_code = "recognition_failed"
+ logger.warning(
+ "HWPX attachment %s recognition failed: %s",
+ getattr(attachment, "id", "?"),
+ exc,
+ )
+ return RESULT_FAILED
+ return RESULT_RECOGNIZED
+
+ expected_content_type = attachment.parse_content_type or "application/pdf"
try:
deferred_bytes = decode_deferred_attachment_payload(
attachment.content,
@@ -266,29 +352,6 @@ async def process_pending_attachment(
)
return RESULT_FAILED
- if attachment.parse_status == HWPX_XML_PACKAGE_PENDING_STATUS:
- try:
- records = recognize_hwpx(
- hwpx_bytes=deferred_bytes,
- source_record_uid=f"attachment-{attachment.id}",
- display_name=attachment.filename or "",
- )
- except HwpxRecognitionError as exc:
- attachment.parse_status = PDF_DOM_RECOGNITION_FAILED_STATUS
- attachment.parse_error_code = "hwpx_recognition_failed"
- logger.warning(
- "HWPX attachment %s recognition failed: %s",
- getattr(attachment, "id", "?"),
- exc,
- )
- return RESULT_FAILED
- apply_hwpx_recognition_to_attachment(
- email=email,
- attachment=attachment,
- records=records,
- )
- return RESULT_RECOGNIZED
-
pdf_bytes = deferred_bytes
config = await config_resolver(session, email.organization_id)
@@ -308,7 +371,7 @@ async def process_pending_attachment(
attachment=attachment,
pdf_bytes=pdf_bytes,
config=config,
- source_record_uid=f"attachment-{attachment.id}",
+ source_record_uid=_attachment_source_record_uid(email, attachment),
request_fn=request_fn,
)
except NewsdomConfigurationError as exc:
@@ -427,13 +490,14 @@ async def _release_sweep_lease(session: AsyncSession) -> None:
class NewsdomRecognitionWorker:
- """Periodically recognize pending PDF attachments and workspace documents.
+ """Periodically recognize pending PDF/HWPX attachments and PDF documents.
Mirrors :class:`ReplySlaScheduler`: a jittered periodic loop, a PostgreSQL
advisory-lock lease so only one replica sweeps per cycle, and per-item
- error isolation. Items whose organization has no active NewsDOM provider are
- left pending (they recognize once a provider is configured); unusable
- payloads/responses are marked failed rather than parsed.
+ error isolation. HWPX work is deterministic/local. PDF items whose
+ organization has no active NewsDOM provider remain pending until a provider
+ is configured; unusable payloads/responses are marked failed rather than
+ parsed.
"""
def __init__(
@@ -506,8 +570,7 @@ async def _sweep(self) -> None:
lease = await _try_acquire_sweep_lease(session)
if lease is False:
logger.debug(
- "NewsDOM recognition sweep skipped: another replica holds "
- "the lease."
+ "Recognition sweep skipped: another replica holds the lease."
)
return
try:
@@ -533,26 +596,23 @@ async def _sweep_attachments(self, session: AsyncSession) -> None:
await session.commit()
if result != RESULT_PENDING:
logger.info(
- "NewsDOM attachment %s recognition result: %s",
+ "Attachment %s recognition result: %s",
attachment.id,
result,
)
except Exception:
await session.rollback()
logger.error(
- "NewsDOM attachment %s recognition raised.",
+ "Attachment %s recognition raised.",
getattr(attachment, "id", "?"),
exc_info=True,
)
def _pending_attachment_statement(self, after_id: int | None):
- """Build the next deterministic attachment batch query."""
+ """Build the next deterministic PDF/HWPX attachment batch query."""
statement = select(Attachment).where(
Attachment.parse_status.in_(
- (
- PDF_DOM_RECOGNITION_PENDING_STATUS,
- HWPX_XML_PACKAGE_PENDING_STATUS,
- )
+ (PDF_DOM_RECOGNITION_PENDING_STATUS, HWPX_PENDING_STATUS)
)
)
if after_id is not None:
diff --git a/backend/tests/test_hwpx_recognition.py b/backend/tests/test_hwpx_recognition.py
index 7a1048dc7..976f1b7ba 100644
--- a/backend/tests/test_hwpx_recognition.py
+++ b/backend/tests/test_hwpx_recognition.py
@@ -1,138 +1,261 @@
-"""Regression tests for bounded HWPX XML extraction."""
+"""Executable contracts for safe HWPX section-text recognition."""
+from __future__ import annotations
+
+import hashlib
import io
import zipfile
import pytest
-from services import hwpx_recognition as hwpx_module
-
-HwpxRecognitionError = hwpx_module.HwpxRecognitionError
-recognize_hwpx = hwpx_module.recognize_hwpx
+from services import hwpx_recognition as recognition
-def _package(
- section_xml: str,
- *,
- compression: int = zipfile.ZIP_DEFLATED,
- duplicate_section: bool = False,
-) -> bytes:
- """Build a package with the same members admitted by attachment parsing."""
- stream = io.BytesIO()
- with zipfile.ZipFile(stream, "w") as archive:
- archive.writestr("mimetype", "application/hwp+zip")
- archive.writestr("version.xml", "")
- archive.writestr("Contents/content.hpf", "")
- archive.writestr("Contents/section0.xml", section_xml, compress_type=compression)
- if duplicate_section:
- with pytest.warns(UserWarning, match="Duplicate name"):
- archive.writestr(
- "Contents/section0.xml",
- section_xml,
- compress_type=compression,
- )
- return stream.getvalue()
+_OPF_NS = "http://www.idpf.org/2007/opf/"
+_HP_NS = "http://www.owpml.org/owpml/2021/paragraph"
+_HS_NS = "http://www.owpml.org/owpml/2021/section"
-def test_recognize_hwpx_extracts_paragraphs_and_graph_records() -> None:
- payload = _package(
- ""
- "첫 문단"
- "둘째 문단"
- ""
+def _section_xml(*paragraphs: str) -> str:
+ """Return a minimal standards-shaped HWPX section XML document."""
+ rendered = "".join(
+ f'{text}'
+ for text in paragraphs
)
+ return (
+ f''
+ f"{rendered}"
+ )
+
- result = recognize_hwpx(
- hwpx_bytes=payload,
- source_record_uid="attachment-1",
- display_name="policy.hwpx",
+def _content_hpf(
+ *,
+ manifest: tuple[tuple[str, str], ...],
+ spine: tuple[str, ...],
+) -> str:
+ """Return a minimal OPF manifest/spine used by an HWPX package."""
+ manifest_xml = "".join(
+ f''
+ for item_id, href in manifest
+ )
+ spine_xml = "".join(f'' for item_id in spine)
+ return (
+ f'{manifest_xml}'
+ f"{spine_xml}"
)
- assert result.parse_text == "첫 문단\n\n둘째 문단"
- assert result.parse_result.content_type == "application/hwp+zip"
- assert len(result.parse_result.segments) == 2
- assert result.source_content_hash
+def _hwpx_package(
+ *,
+ sections: dict[str, str],
+ spine: tuple[str, ...],
+ manifest_hrefs: dict[str, str] | None = None,
+ include_version: bool = True,
+) -> bytes:
+ """Build a small HWPX package with explicit manifest and spine order."""
+ hrefs = manifest_hrefs or {
+ section_id: f"Contents/{section_id}.xml" for section_id in sections
+ }
+ buffer = io.BytesIO()
+ with zipfile.ZipFile(buffer, "w", compression=zipfile.ZIP_DEFLATED) as archive:
+ archive.writestr("mimetype", b"application/hwp+zip")
+ if include_version:
+ archive.writestr("version.xml", b'')
+ archive.writestr(
+ "Contents/content.hpf",
+ _content_hpf(
+ manifest=tuple((section_id, hrefs[section_id]) for section_id in sections),
+ spine=spine,
+ ),
+ )
+ for section_id, xml in sections.items():
+ archive.writestr(f"Contents/{section_id}.xml", xml)
+ return buffer.getvalue()
+
+
+def test_recognize_hwpx_follows_spine_and_preserves_paragraph_provenance() -> None:
+ """Read sections in OPF spine order and retain semantic source positions."""
+ payload = _hwpx_package(
+ sections={
+ "section0": _section_xml("첫 번째 구역", "둘째 문단"),
+ "section1": _section_xml("두 번째 구역"),
+ },
+ spine=("section1", "section0"),
+ )
-def test_recognize_hwpx_rejects_missing_sections() -> None:
- stream = io.BytesIO()
- with zipfile.ZipFile(stream, "w") as archive:
- archive.writestr("mimetype", "application/hwp+zip")
- archive.writestr("version.xml", "")
- archive.writestr("Contents/content.hpf", "")
+ records = recognition.recognize_hwpx_package(
+ payload,
+ filename="proposal.hwpx",
+ source_kind="attachment",
+ source_record_uid="attachment-42",
+ )
- with pytest.raises(HwpxRecognitionError, match="no section"):
- recognize_hwpx(hwpx_bytes=stream.getvalue(), source_record_uid="attachment-2")
+ assert records.parse_text == "두 번째 구역\n\n첫 번째 구역\n\n둘째 문단"
+ assert records.parse_result.source_content_hash == hashlib.sha256(payload).hexdigest()
+ assert [segment.safe_text_content for segment in records.parse_result.segments] == [
+ "두 번째 구역",
+ "첫 번째 구역",
+ "둘째 문단",
+ ]
+ assert [segment.segment_path for segment in records.parse_result.segments] == [
+ "/document[1]/section[1]/paragraph[1]",
+ "/document[1]/section[2]/paragraph[1]",
+ "/document[1]/section[2]/paragraph[2]",
+ ]
+
+ changed_records = recognition.recognize_hwpx_package(
+ _hwpx_package(
+ sections={"section0": _section_xml("변경된 문서")},
+ spine=("section0",),
+ ),
+ filename="proposal.hwpx",
+ source_kind="attachment",
+ source_record_uid="attachment-42",
+ )
+ assert changed_records.parse_result.nodes[0].content_node_uid != (
+ records.parse_result.nodes[0].content_node_uid
+ )
+ assert changed_records.parse_result.segments[0].content_segment_uid != (
+ records.parse_result.segments[0].content_segment_uid
+ )
-def test_recognize_hwpx_rejects_duplicate_sections() -> None:
- payload = _package(
- "text
",
- duplicate_section=True,
+def test_recognize_hwpx_rejects_missing_version_member() -> None:
+ """Revalidate the required version member before section recognition."""
+ payload = _hwpx_package(
+ sections={"section0": _section_xml("safe")},
+ spine=("section0",),
+ include_version=False,
)
- with pytest.raises(HwpxRecognitionError, match="duplicate"):
- recognize_hwpx(hwpx_bytes=payload, source_record_uid="attachment-3")
+ with pytest.raises(ValueError, match="missing required identity"):
+ recognition.recognize_hwpx_package(
+ payload,
+ filename="missing-version.hwpx",
+ source_kind="attachment",
+ source_record_uid="attachment-47",
+ )
-def test_recognize_hwpx_rejects_entity_declarations_and_malformed_xml() -> None:
- entity_payload = _package(
- "]>&x;"
+@pytest.mark.parametrize("member_name", ("mimetype", "Contents/content.hpf"))
+def test_recognize_hwpx_normalizes_zip_read_failures(
+ monkeypatch: pytest.MonkeyPatch,
+ member_name: str,
+) -> None:
+ """Convert ZIP read implementation failures into bounded parse errors."""
+ payload = _hwpx_package(
+ sections={"section0": _section_xml("safe")},
+ spine=("section0",),
)
- with pytest.raises(HwpxRecognitionError, match="declarations"):
- recognize_hwpx(hwpx_bytes=entity_payload, source_record_uid="attachment-4")
+ original_read = zipfile.ZipFile.read
+
+ def broken_read(archive, member, *args, **kwargs):
+ if getattr(member, "filename", member) == member_name:
+ raise zipfile.BadZipFile("simulated read failure")
+ return original_read(archive, member, *args, **kwargs)
+
+ monkeypatch.setattr(zipfile.ZipFile, "read", broken_read)
- malformed_payload = _package("")
- with pytest.raises(HwpxRecognitionError, match="safely read"):
- recognize_hwpx(
- hwpx_bytes=malformed_payload,
- source_record_uid="attachment-5",
+ with pytest.raises(ValueError, match="could not be read"):
+ recognition.recognize_hwpx_package(
+ payload,
+ filename="read-failure.hwpx",
+ source_kind="attachment",
+ source_record_uid="attachment-48",
)
-def test_recognize_hwpx_rejects_unsupported_compression() -> None:
- if not hasattr(zipfile, "ZIP_BZIP2"):
- pytest.skip("Python zipfile has no BZIP2 support")
- payload = _package(
- "text
",
- compression=zipfile.ZIP_BZIP2,
+def test_paragraph_text_does_not_duplicate_nested_paragraphs() -> None:
+ """Nested paragraph nodes are skipped by the containing paragraph."""
+ nested_section = (
+ f''
+ "nested"
+ ""
+ )
+ payload = _hwpx_package(
+ sections={"section0": nested_section},
+ spine=("section0",),
)
- with pytest.raises(HwpxRecognitionError, match="compression"):
- recognize_hwpx(hwpx_bytes=payload, source_record_uid="attachment-6")
+ records = recognition.recognize_hwpx_package(
+ payload,
+ filename="nested.hwpx",
+ source_kind="attachment",
+ source_record_uid="attachment-49",
+ )
+ assert records.parse_text == "nested"
+ assert records.paragraph_count == 1
-def test_recognize_hwpx_enforces_section_and_total_xml_limits(monkeypatch) -> None:
- payload = _package("text
")
- monkeypatch.setattr(hwpx_module, "MAX_HWPX_SECTION_XML_BYTES", 1)
- with pytest.raises(HwpxRecognitionError, match="section exceeds"):
- recognize_hwpx(hwpx_bytes=payload, source_record_uid="attachment-7")
- monkeypatch.setattr(hwpx_module, "MAX_HWPX_SECTION_XML_BYTES", 10_000)
- monkeypatch.setattr(hwpx_module, "MAX_HWPX_TOTAL_XML_BYTES", 1)
- with pytest.raises(HwpxRecognitionError, match="total size"):
- recognize_hwpx(hwpx_bytes=payload, source_record_uid="attachment-8")
+def test_recognize_hwpx_rejects_manifest_path_traversal() -> None:
+ """Never resolve an OPF manifest href outside the HWPX package root."""
+ payload = _hwpx_package(
+ sections={"section0": _section_xml("safe")},
+ spine=("section0",),
+ manifest_hrefs={"section0": "../section0.xml"},
+ )
+ with pytest.raises(ValueError, match="unsafe manifest href"):
+ recognition.recognize_hwpx_package(
+ payload,
+ filename="unsafe.hwpx",
+ source_kind="attachment",
+ source_record_uid="attachment-43",
+ )
-def test_recognize_hwpx_rejects_changed_read_size(monkeypatch) -> None:
- payload = _package("text
")
- monkeypatch.setattr(zipfile.ZipFile, "read", lambda _archive, _info: b"")
- with pytest.raises(HwpxRecognitionError, match="size changed"):
- recognize_hwpx(hwpx_bytes=payload, source_record_uid="attachment-9")
+def test_recognize_hwpx_bounds_expanded_xml_before_reading(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ """Reject a ZIP expansion budget violation before XML parsing."""
+ payload = _hwpx_package(
+ sections={"section0": _section_xml("x" * 256)},
+ spine=("section0",),
+ )
+ monkeypatch.setattr(recognition, "MAX_HWPX_XML_MEMBER_BYTES", 64)
+
+ with pytest.raises(ValueError, match="XML member exceeds"):
+ recognition.recognize_hwpx_package(
+ payload,
+ filename="oversized.hwpx",
+ source_kind="attachment",
+ source_record_uid="attachment-44",
+ )
-def test_recognize_hwpx_enforces_text_limit(monkeypatch) -> None:
- payload = _package("text
")
- monkeypatch.setattr(hwpx_module, "MAX_HWPX_TEXT_CHARS", 1)
+def test_recognize_hwpx_rejects_unsafe_xml_entities() -> None:
+ """Defused XML parsing must reject entity-bearing section payloads."""
+ dangerous_section = (
+ ']>'
+ f''
+ "&exfil;"
+ )
+ payload = _hwpx_package(
+ sections={"section0": dangerous_section},
+ spine=("section0",),
+ )
- with pytest.raises(HwpxRecognitionError, match="text exceeds"):
- recognize_hwpx(hwpx_bytes=payload, source_record_uid="attachment-10")
+ with pytest.raises(ValueError, match="unsafe XML"):
+ recognition.recognize_hwpx_package(
+ payload,
+ filename="entity.hwpx",
+ source_kind="attachment",
+ source_record_uid="attachment-45",
+ )
-def test_recognize_hwpx_rejects_empty_text_and_non_string_xml_tags() -> None:
- payload = _package("
")
- with pytest.raises(HwpxRecognitionError, match="no readable"):
- recognize_hwpx(hwpx_bytes=payload, source_record_uid="attachment-11")
+def test_recognize_hwpx_requires_spine_referenced_section() -> None:
+ """Fail closed when OPF spine metadata cannot resolve a section member."""
+ payload = _hwpx_package(
+ sections={"section0": _section_xml("safe")},
+ spine=("missing-section",),
+ )
- assert hwpx_module._local_name(None) == ""
+ with pytest.raises(ValueError, match="spine item"):
+ recognition.recognize_hwpx_package(
+ payload,
+ filename="missing.hwpx",
+ source_kind="attachment",
+ source_record_uid="attachment-46",
+ )
diff --git a/backend/tests/test_hwpx_worker.py b/backend/tests/test_hwpx_worker.py
new file mode 100644
index 000000000..3a93fccd7
--- /dev/null
+++ b/backend/tests/test_hwpx_worker.py
@@ -0,0 +1,226 @@
+"""Contract tests for deferred HWPX background recognition.
+
+The attachment importer deliberately stores validated HWPX bytes as a pending
+base64 payload. These tests require the existing recognition worker to consume
+that pending state locally, without requiring a NewsDOM provider, while keeping
+failure states explicit and bounded.
+"""
+
+from __future__ import annotations
+
+import base64
+import io
+import zipfile
+
+import pytest
+from sqlalchemy.dialects import postgresql
+
+from db.models import Attachment, Email
+from services.newsdom_worker import (
+ RESULT_FAILED,
+ RESULT_RECOGNIZED,
+ NewsdomRecognitionWorker,
+ process_pending_attachment,
+)
+
+HWPX_PENDING_STATUS = "hwpx_xml_package_pending"
+HWPX_PARSED_STATUS = "hwpx_xml_package_parsed"
+HWPX_FAILED_STATUS = "hwpx_xml_package_failed"
+
+
+def _hwpx_payload(
+ *, include_section: bool = True, include_version: bool = True
+) -> bytes:
+ """Build one minimal standards-shaped HWPX package for worker tests."""
+
+ package = io.BytesIO()
+ with zipfile.ZipFile(package, "w", compression=zipfile.ZIP_STORED) as archive:
+ archive.writestr("mimetype", b"application/hwp+zip")
+ # Parent #1353 admission requires version.xml before a payload can be pending.
+ if include_version:
+ archive.writestr("version.xml", b'')
+ archive.writestr(
+ "Contents/content.hpf",
+ b"""
+
+
+
+
+
+
+
+
+""",
+ )
+ if include_section:
+ archive.writestr(
+ "Contents/section0.xml",
+ b"""
+
+ Quarterly decision record
+ Approve the next action.
+
+""",
+ )
+ return package.getvalue()
+
+
+def _pending_hwpx_attachment(payload: bytes) -> Attachment:
+ """Create an in-memory HWPX attachment with its owning email relationship."""
+
+ email = Email()
+ email.organization_id = "org-hwpx"
+ attachment = Attachment(
+ id=73,
+ filename="decision.hwpx",
+ content=base64.b64encode(payload).decode("ascii"),
+ content_type="application/hwp+zip",
+ parse_content_type="application/hwp+zip",
+ parser_key="hwpx",
+ parse_status=HWPX_PENDING_STATUS,
+ )
+ email.attachments.append(attachment)
+ return attachment
+
+
+async def _must_not_resolve_provider(*_args, **_kwargs):
+ """Fail when deterministic HWPX recognition tries to resolve NewsDOM."""
+
+ raise AssertionError("HWPX recognition must not require a NewsDOM provider")
+
+
+async def _must_not_call_newsdom(**_kwargs):
+ """Fail when deterministic HWPX recognition reaches the NewsDOM sidecar."""
+
+ raise AssertionError("HWPX recognition must not call the NewsDOM sidecar")
+
+
+@pytest.mark.asyncio
+async def test_pending_hwpx_attachment_is_recognized_without_provider() -> None:
+ """A pending HWPX package becomes searchable text plus graph provenance."""
+
+ attachment = _pending_hwpx_attachment(_hwpx_payload())
+ attachment.parse_content_type = ""
+
+ result = await process_pending_attachment(
+ session=object(),
+ attachment=attachment,
+ config_resolver=_must_not_resolve_provider,
+ request_fn=_must_not_call_newsdom,
+ )
+
+ assert result == RESULT_RECOGNIZED
+ assert attachment.parse_status == HWPX_PARSED_STATUS
+ assert attachment.parse_error_code is None
+ assert attachment.parse_content_type == "application/hwp+zip"
+ assert attachment.parser_key == "hwpx"
+ assert attachment.content == (
+ "Quarterly decision record\n\nApprove the next action."
+ )
+ assert [segment.safe_text_content for segment in attachment.content_segments] == [
+ "Quarterly decision record",
+ "Approve the next action.",
+ ]
+ assert attachment.content_nodes
+ assert attachment.content_nodes[0].source_record_uid.startswith("attachment:")
+ assert attachment.content_nodes[0].source_record_uid != "attachment-73"
+
+
+@pytest.mark.asyncio
+async def test_pending_hwpx_attachment_revalidates_retained_bytes() -> None:
+ """Tampered retained bytes fail closed before any provider or XML work."""
+
+ attachment = _pending_hwpx_attachment(b"not-a-zip")
+
+ result = await process_pending_attachment(
+ session=object(),
+ attachment=attachment,
+ config_resolver=_must_not_resolve_provider,
+ request_fn=_must_not_call_newsdom,
+ )
+
+ assert result == RESULT_FAILED
+ assert attachment.parse_status == HWPX_FAILED_STATUS
+ assert attachment.parse_error_code == "invalid_pending_payload"
+ assert attachment.content_nodes == []
+ assert attachment.content_segments == []
+
+
+@pytest.mark.asyncio
+async def test_pending_hwpx_attachment_records_recognizer_failure() -> None:
+ """A valid HWPX identity with a broken spine never masquerades as parsed."""
+
+ attachment = _pending_hwpx_attachment(_hwpx_payload(include_section=False))
+
+ result = await process_pending_attachment(
+ session=object(),
+ attachment=attachment,
+ config_resolver=_must_not_resolve_provider,
+ request_fn=_must_not_call_newsdom,
+ )
+
+ assert result == RESULT_FAILED
+ assert attachment.parse_status == HWPX_FAILED_STATUS
+ assert attachment.parse_error_code == "recognition_failed"
+ assert attachment.content_nodes == []
+ assert attachment.content_segments == []
+
+
+@pytest.mark.asyncio
+async def test_pending_hwpx_attachment_revalidates_version_member() -> None:
+ """A retained package missing version.xml cannot become parsed."""
+ attachment = _pending_hwpx_attachment(_hwpx_payload(include_version=False))
+
+ result = await process_pending_attachment(
+ session=object(),
+ attachment=attachment,
+ config_resolver=_must_not_resolve_provider,
+ request_fn=_must_not_call_newsdom,
+ )
+
+ assert result == RESULT_FAILED
+ assert attachment.parse_status == HWPX_FAILED_STATUS
+ assert attachment.parse_error_code == "invalid_pending_payload"
+
+
+@pytest.mark.asyncio
+async def test_orphan_pending_hwpx_attachment_uses_hwpx_failure_status() -> None:
+ """An orphan HWPX row fails visibly without being mislabeled as PDF."""
+
+ attachment = Attachment(
+ id=74,
+ filename="orphan.hwpx",
+ content=base64.b64encode(_hwpx_payload()).decode("ascii"),
+ content_type="application/hwp+zip",
+ parse_content_type="application/hwp+zip",
+ parser_key="hwpx",
+ parse_status=HWPX_PENDING_STATUS,
+ )
+
+ result = await process_pending_attachment(
+ session=object(),
+ attachment=attachment,
+ config_resolver=_must_not_resolve_provider,
+ request_fn=_must_not_call_newsdom,
+ )
+
+ assert result == RESULT_FAILED
+ assert attachment.parse_status == HWPX_FAILED_STATUS
+ assert attachment.parse_error_code == "orphan_attachment"
+
+
+def test_worker_selects_pdf_and_hwpx_pending_attachments() -> None:
+ """The bounded sweep must include both deferred attachment families."""
+
+ worker = NewsdomRecognitionWorker(batch_limit=7)
+ statement = worker._pending_attachment_statement(None)
+ sql = str(
+ statement.compile(
+ dialect=postgresql.dialect(),
+ compile_kwargs={"literal_binds": True},
+ )
+ )
+
+ assert "pdf_dom_recognition_pending" in sql
+ assert HWPX_PENDING_STATUS in sql
+ assert "LIMIT 7" in sql
diff --git a/backend/tests/test_newsdom_worker.py b/backend/tests/test_newsdom_worker.py
index fd5bcb18b..f904d37df 100644
--- a/backend/tests/test_newsdom_worker.py
+++ b/backend/tests/test_newsdom_worker.py
@@ -29,6 +29,8 @@
RESULT_FAILED = newsdom_worker_module.RESULT_FAILED
RESULT_PENDING = newsdom_worker_module.RESULT_PENDING
RESULT_RECOGNIZED = newsdom_worker_module.RESULT_RECOGNIZED
+HWPX_FAILED_STATUS = newsdom_worker_module.HWPX_FAILED_STATUS
+HWPX_PARSED_STATUS = newsdom_worker_module.HWPX_PARSED_STATUS
process_pending_attachment = newsdom_worker_module.process_pending_attachment
process_pending_document = newsdom_worker_module.process_pending_document
@@ -101,7 +103,13 @@ def _hwpx_payload(*, include_text: bool = True) -> bytes:
compress_type=zipfile.ZIP_STORED,
)
archive.writestr("version.xml", "")
- archive.writestr("Contents/content.hpf", "")
+ archive.writestr(
+ "Contents/content.hpf",
+ """
+