from __future__ import annotations from collections import Counter from dataclasses import dataclass from typing import Literal from xml.etree.ElementTree import ParseError from defusedxml.ElementTree import fromstring from defusedxml.common import DefusedXmlException BPMN_MODEL_NAMESPACE = "http://www.omg.org/spec/BPMN/20100524/MODEL" BPMN_DI_NAMESPACE = "http://www.omg.org/spec/BPMN/20100524/DI" OMG_DI_NAMESPACE = "http://www.omg.org/spec/DD/20100524/DI" OMG_DC_NAMESPACE = "http://www.omg.org/spec/DD/20100524/DC" MAX_BPMN_XML_BYTES = 1_048_576 MAX_BPMN_ELEMENTS = 20_000 SupportLevel = Literal[ "interchange_only", "native_mapping", "native_execution", ] NATIVE_EXECUTION_ELEMENTS = frozenset( { "startEvent", "endEvent", "manualTask", "userTask", "serviceTask", "businessRuleTask", "receiveTask", "sendTask", "exclusiveGateway", "parallelGateway", "sequenceFlow", "intermediateCatchEvent", "intermediateThrowEvent", } ) NATIVE_MAPPING_ELEMENTS = NATIVE_EXECUTION_ELEMENTS | frozenset( { "task", "scriptTask", "callActivity", "subProcess", "transaction", "adHocSubProcess", "inclusiveGateway", "eventBasedGateway", "complexGateway", "boundaryEvent", "eventSubProcess", "dataObject", "dataObjectReference", "dataStoreReference", "messageFlow", "participant", "lane", "laneSet", "textAnnotation", "association", "group", } ) @dataclass(frozen=True, slots=True) class BpmnDiagnostic: severity: Literal["error", "warning", "info"] code: str message: str element_id: str | None = None @dataclass(frozen=True, slots=True) class BpmnElementInventoryItem: element_type: str element_id: str | None name: str | None parent_type: str | None parent_id: str | None support_level: SupportLevel @dataclass(frozen=True, slots=True) class BpmnInspection: definitions_id: str | None target_namespace: str | None process_count: int executable_process_count: int collaboration_count: int choreography_count: int element_counts: dict[str, int] support_counts: dict[str, int] elements: tuple[BpmnElementInventoryItem, ...] diagnostics: tuple[BpmnDiagnostic, ...] @property def valid_xml(self) -> bool: return not any(item.severity == "error" for item in self.diagnostics) class BpmnInspectionError(ValueError): pass def bpmn_support_level(element_type: str) -> SupportLevel: if element_type in NATIVE_EXECUTION_ELEMENTS: return "native_execution" if element_type in NATIVE_MAPPING_ELEMENTS: return "native_mapping" return "interchange_only" def inspect_bpmn_xml(xml: str) -> BpmnInspection: encoded = xml.encode("utf-8") if not encoded: raise BpmnInspectionError("BPMN XML is empty") if len(encoded) > MAX_BPMN_XML_BYTES: raise BpmnInspectionError( f"BPMN XML exceeds the {MAX_BPMN_XML_BYTES}-byte inspection limit" ) try: root = fromstring(encoded) except (DefusedXmlException, ParseError, ValueError) as exc: raise BpmnInspectionError(f"BPMN XML is not safe and well formed: {exc}") from exc namespace, local_name = _qualified_name(root.tag) if namespace != BPMN_MODEL_NAMESPACE or local_name != "definitions": raise BpmnInspectionError( "BPMN document root must be bpmn:definitions in the BPMN 2.0 model namespace" ) diagnostics: list[BpmnDiagnostic] = [] elements: list[BpmnElementInventoryItem] = [] ids: dict[str, str] = {} references: list[tuple[str, str | None, str]] = [] process_count = 0 executable_process_count = 0 collaboration_count = 0 choreography_count = 0 stack = [(root, None, None)] visited = 0 while stack: element, parent_type, parent_id = stack.pop() visited += 1 if visited > MAX_BPMN_ELEMENTS: raise BpmnInspectionError( f"BPMN document exceeds the {MAX_BPMN_ELEMENTS}-element inspection limit" ) element_namespace, element_type = _qualified_name(element.tag) element_id = _bounded_attribute(element.attrib.get("id"), 255) if element_namespace == BPMN_MODEL_NAMESPACE: name = _bounded_attribute(element.attrib.get("name"), 300) support = bpmn_support_level(element_type) elements.append( BpmnElementInventoryItem( element_type=element_type, element_id=element_id, name=name, parent_type=parent_type, parent_id=parent_id, support_level=support, ) ) if element_id: previous = ids.get(element_id) if previous is not None: diagnostics.append( BpmnDiagnostic( severity="error", code="duplicate_bpmn_id", message=( f"BPMN id {element_id!r} is used by both " f"{previous} and {element_type}" ), element_id=element_id, ) ) else: ids[element_id] = element_type if element_type == "process": process_count += 1 if element.attrib.get("isExecutable", "").strip().lower() == "true": executable_process_count += 1 elif element_type == "collaboration": collaboration_count += 1 elif element_type in {"choreography", "globalChoreographyTask"}: choreography_count += 1 _collect_references(element_type, element_id, element.attrib, references) next_parent_type = element_type next_parent_id = element_id else: next_parent_type = parent_type next_parent_id = parent_id for child in reversed(list(element)): stack.append((child, next_parent_type, next_parent_id)) if process_count == 0 and collaboration_count == 0 and choreography_count == 0: diagnostics.append( BpmnDiagnostic( severity="warning", code="no_bpmn_process_or_collaboration", message="BPMN definitions contain no process, collaboration, or choreography", ) ) for reference, element_id, field in references: if reference not in ids: diagnostics.append( BpmnDiagnostic( severity="error", code="dangling_bpmn_reference", message=( f"{field} references unknown BPMN id {reference!r}" ), element_id=element_id, ) ) unsupported = [ item for item in elements if item.support_level == "interchange_only" and item.element_type not in {"definitions", "process"} ] if unsupported: diagnostics.append( BpmnDiagnostic( severity="info", code="interchange_only_elements", message=( f"{len(unsupported)} BPMN element(s) can be inventoried and " "round-tripped by a BPMN modeler but are not executable by " "the native GovOPlaN workflow runtime" ), ) ) element_counts = dict( sorted(Counter(item.element_type for item in elements).items()) ) support_counts = dict( sorted(Counter(item.support_level for item in elements).items()) ) return BpmnInspection( definitions_id=_bounded_attribute(root.attrib.get("id"), 255), target_namespace=_bounded_attribute( root.attrib.get("targetNamespace"), 1000, ), process_count=process_count, executable_process_count=executable_process_count, collaboration_count=collaboration_count, choreography_count=choreography_count, element_counts=element_counts, support_counts=support_counts, elements=tuple(elements), diagnostics=tuple(diagnostics), ) def _collect_references( element_type: str, element_id: str | None, attributes: dict[str, str], references: list[tuple[str, str | None, str]], ) -> None: fields: tuple[str, ...] if element_type == "sequenceFlow": fields = ("sourceRef", "targetRef") elif element_type == "messageFlow": fields = ("sourceRef", "targetRef", "messageRef") elif element_type == "participant": fields = ("processRef",) elif element_type == "lane": fields = ("partitionElementRef",) else: fields = () for field in fields: value = attributes.get(field) if value: references.append((value, element_id, field)) def _qualified_name(tag: str) -> tuple[str | None, str]: if tag.startswith("{") and "}" in tag: namespace, local_name = tag[1:].split("}", 1) return namespace, local_name return None, tag def _bounded_attribute(value: str | None, limit: int) -> str | None: if value is None: return None text = value.strip() return text[:limit] if text else None __all__ = [ "BPMN_MODEL_NAMESPACE", "BpmnInspectionError", "NATIVE_EXECUTION_ELEMENTS", "NATIVE_MAPPING_ELEMENTS", "bpmn_support_level", "inspect_bpmn_xml", ]