feat(xrechnung): validate inbound invoices fail closed
Module Package Release / publish-packages (push) Successful in 12s
Module Package Release / publish-packages (push) Successful in 12s
This commit is contained in:
@@ -0,0 +1,4 @@
|
||||
"""GovOPlaN XRechnung connector."""
|
||||
|
||||
__version__ = "0.1.19"
|
||||
|
||||
@@ -0,0 +1,2 @@
|
||||
"""Inbound XRechnung validation boundary."""
|
||||
|
||||
@@ -0,0 +1,144 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from govoplan_core.core.access import (
|
||||
CAPABILITY_AUTH_PERMISSION_EVALUATOR,
|
||||
CAPABILITY_AUTH_PRINCIPAL_RESOLVER,
|
||||
)
|
||||
from govoplan_core.core.modules import (
|
||||
DocumentationCondition,
|
||||
DocumentationLink,
|
||||
DocumentationTopic,
|
||||
ModuleManifest,
|
||||
PermissionDefinition,
|
||||
RoleTemplate,
|
||||
)
|
||||
from govoplan_core.core.provider_governance import declared_module_architecture
|
||||
|
||||
|
||||
MODULE_ID = "xrechnung"
|
||||
MODULE_VERSION = "0.1.19"
|
||||
READ_SCOPE = "xrechnung:validation:read"
|
||||
EXECUTE_SCOPE = "xrechnung:validation:execute"
|
||||
HANDOFF_SCOPE = "xrechnung:handoff:create"
|
||||
|
||||
|
||||
def _permission(scope: str, label: str, description: str) -> PermissionDefinition:
|
||||
module_id, resource, action = scope.split(":", 2)
|
||||
return PermissionDefinition(
|
||||
scope=scope,
|
||||
label=label,
|
||||
description=description,
|
||||
category="XRechnung",
|
||||
level="tenant",
|
||||
module_id=module_id,
|
||||
resource=resource,
|
||||
action=action,
|
||||
)
|
||||
|
||||
|
||||
manifest = ModuleManifest(
|
||||
id=MODULE_ID,
|
||||
name="XRechnung",
|
||||
version=MODULE_VERSION,
|
||||
dependencies=("access",),
|
||||
optional_dependencies=("files", "procurement", "payments", "records", "audit"),
|
||||
required_capabilities=(
|
||||
CAPABILITY_AUTH_PRINCIPAL_RESOLVER,
|
||||
CAPABILITY_AUTH_PERMISSION_EVALUATOR,
|
||||
),
|
||||
permissions=(
|
||||
_permission(READ_SCOPE, "View XRechnung validation", "Read validation profiles, outcomes, and non-secret diagnostics."),
|
||||
_permission(EXECUTE_SCOPE, "Validate inbound XRechnung", "Run a pinned local KoSIT profile against an inbound invoice."),
|
||||
_permission(HANDOFF_SCOPE, "Handoff validated XRechnung", "Create a digest-bound handoff only from technically complete valid evidence."),
|
||||
),
|
||||
role_templates=(
|
||||
RoleTemplate(
|
||||
slug="xrechnung_processor",
|
||||
name="XRechnung processor",
|
||||
description="Validate inbound invoices and create governed valid-invoice handoffs.",
|
||||
permissions=(READ_SCOPE, EXECUTE_SCOPE, HANDOFF_SCOPE),
|
||||
),
|
||||
RoleTemplate(
|
||||
slug="xrechnung_reviewer",
|
||||
name="XRechnung reviewer",
|
||||
description="Inspect validation profiles, results, and diagnostics without creating effects.",
|
||||
permissions=(READ_SCOPE,),
|
||||
),
|
||||
),
|
||||
documentation=(
|
||||
DocumentationTopic(
|
||||
id="xrechnung.inbound-validation",
|
||||
title="Validate and hand off an inbound XRechnung",
|
||||
summary="Verify safe invoice XML with an exact locally pinned KoSIT engine and rule tree, then create a digest-bound handoff only from complete valid evidence.",
|
||||
body=(
|
||||
"An administrator must approve an exact XRechnung version, KoSIT validator release, configuration release, configuration-tree digest, engine digest, scenario file, and minimum validation-step count. The connector verifies those artifacts before each run and never downloads rules while processing an invoice. It rejects dangerous XML before invoking Java. Runner exit, bounded output, technical error markers, VARL identity, matched scenario, configured validation-step count, validity flag, and exactly one accept or reject assessment are checked independently. Technical failure or partial reports yield unknown conformance and can never be handed off. A valid handoff binds tenant, source reference, invoice digest, report digest, and profile digest for an owning Procurement or Payments workflow."
|
||||
),
|
||||
layer="configured",
|
||||
documentation_types=("admin", "user"),
|
||||
audience=("user", "operator", "module_admin", "auditor"),
|
||||
related_modules=("files", "procurement", "payments", "records", "audit"),
|
||||
conditions=(
|
||||
DocumentationCondition(
|
||||
any_scopes=(READ_SCOPE, EXECUTE_SCOPE, HANDOFF_SCOPE),
|
||||
),
|
||||
),
|
||||
links=(
|
||||
DocumentationLink(
|
||||
label="Inbound validation profile",
|
||||
href="docs/INBOUND_VALIDATION.md",
|
||||
kind="repository",
|
||||
),
|
||||
),
|
||||
translations={
|
||||
"de": {
|
||||
"title": "Eingehende XRechnung prüfen und übergeben",
|
||||
"summary": "Sicheres Rechnungs-XML mit exakt lokal festgelegter KoSIT-Engine und Regelstruktur prüfen und nur aus vollständigem gültigem Nachweis eine prüfsummengebundene Übergabe erzeugen.",
|
||||
"body": "Administrierende müssen eine exakte XRechnungs-Version, KoSIT-Validator-Version, Konfigurationsversion, Prüfsumme der Konfigurationsstruktur, Engine-Prüfsumme, Szenariodatei und Mindestzahl der Prüfschritte freigeben. Der Konnektor prüft diese Artefakte vor jedem Lauf und lädt während der Rechnungsverarbeitung keine Regeln herunter. Gefährliches XML wird vor dem Java-Aufruf abgewiesen. Prozessende, begrenzte Ausgabe, technische Fehlermarker, VARL-Identität, erkanntes Szenario, konfigurierte Prüfschrittzahl, Gültigkeitskennzeichen und genau eine Annahme- oder Ablehnungsempfehlung werden unabhängig geprüft. Technische Fehler oder Teilberichte ergeben unbekannte Konformität und dürfen niemals übergeben werden. Eine gültige Übergabe bindet Mandant, Quellreferenz, Rechnungsprüfsumme, Berichtsprüfsumme und Profilprüfsumme für einen fachlich verantwortlichen Procurement- oder Payments-Ablauf.",
|
||||
}
|
||||
},
|
||||
metadata={
|
||||
"kind": "workflow",
|
||||
"prerequisites": [
|
||||
"An exact KoSIT engine and self-contained XRechnung configuration have been installed locally.",
|
||||
"The administrator has approved all artifact digests and the expected validation-step count.",
|
||||
"The owning invoice workflow supplies a stable tenant and source reference.",
|
||||
],
|
||||
"steps": [
|
||||
"Verify the executable, engine digest, complete configuration-tree digest, and scenario location.",
|
||||
"Reject unsafe or oversized XML, then run the pinned profile with bounded time and output.",
|
||||
"Check technical completeness separately from the report's semantic validity.",
|
||||
"Create a handoff only when the report is complete, formally valid, and accepts the invoice.",
|
||||
],
|
||||
"limitations": [
|
||||
"No XRechnung/configuration version is activated by default in this release.",
|
||||
"The connector creates a handoff contract but does not own payable approval or booking.",
|
||||
"A syntactically valid report is not trusted when runner output signals a technical failure.",
|
||||
],
|
||||
"consequences": [
|
||||
"Changing any pinned artifact changes the profile digest and requires a new validation.",
|
||||
"Invalid invoices retain diagnostics but cannot enter the valid-invoice handoff.",
|
||||
"Technical failure produces unknown conformance, never a semantic rejection or acceptance.",
|
||||
],
|
||||
"verification": "Confirm artifact and profile digests, technical outcome, validation-step count, formal validity, VARL assessment, invoice digest, and report digest before following the handoff reference.",
|
||||
},
|
||||
order=100,
|
||||
),
|
||||
),
|
||||
architecture=declared_module_architecture(
|
||||
layer="data_reporting_integration",
|
||||
kind="integration",
|
||||
maturity="vertical_slice",
|
||||
documentation_ref="docs/INBOUND_VALIDATION.md",
|
||||
test_ref="tests/test_validation.py",
|
||||
known_limits=("A deployment-approved XRechnung and KoSIT configuration profile is required; none is activated by default.",),
|
||||
owned_concepts=("XRechnung validation profile", "validation result", "validated invoice handoff"),
|
||||
non_owned_concepts=("invoice payable", "procurement approval", "booking status", "invoice file storage"),
|
||||
recovery_docs=("docs/INBOUND_VALIDATION.md",),
|
||||
security_docs=("docs/INBOUND_VALIDATION.md",),
|
||||
operations_docs=("docs/INBOUND_VALIDATION.md",),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def get_manifest() -> ModuleManifest:
|
||||
return manifest
|
||||
@@ -0,0 +1,441 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from datetime import UTC, datetime
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import re
|
||||
import subprocess
|
||||
import tempfile
|
||||
from typing import Literal
|
||||
|
||||
from defusedxml import ElementTree
|
||||
|
||||
|
||||
VARL_NAMESPACE = "http://www.xoev.de/de/validator/varl/1"
|
||||
MAX_INVOICE_BYTES = 20 * 1024 * 1024
|
||||
MAX_RUNNER_OUTPUT_BYTES = 2 * 1024 * 1024
|
||||
MAX_REPORT_BYTES = 16 * 1024 * 1024
|
||||
_TECHNICAL_OUTPUT = re.compile(
|
||||
rb"(?im)(?:^|\n)\s*(?:\[[^\]]*\]\s*)?(?:ERROR|FATAL)\b|\b(?:Exception|StackOverflowError|OutOfMemoryError)\b"
|
||||
)
|
||||
|
||||
TechnicalOutcome = Literal["complete", "failed", "incomplete"]
|
||||
ConformanceOutcome = Literal["valid", "invalid", "unknown"]
|
||||
AssessmentOutcome = Literal["accept", "reject", "unknown"]
|
||||
|
||||
|
||||
class XRechnungValidationError(RuntimeError):
|
||||
"""Stable validation or configuration error without invoice contents."""
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class KoSITValidationProfile:
|
||||
"""Exact locally installed validator and rule-set selection."""
|
||||
|
||||
profile_id: str
|
||||
xrechnung_version: str
|
||||
validator_version: str
|
||||
configuration_version: str
|
||||
java_executable: Path
|
||||
validator_jar: Path
|
||||
validator_jar_sha256: str
|
||||
configuration_root: Path
|
||||
configuration_tree_sha256: str
|
||||
scenarios_file: Path
|
||||
minimum_validation_steps: int
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
for name in (
|
||||
"profile_id",
|
||||
"xrechnung_version",
|
||||
"validator_version",
|
||||
"configuration_version",
|
||||
):
|
||||
value = str(getattr(self, name) or "").strip()
|
||||
if not value or len(value) > 255:
|
||||
raise ValueError(f"XRechnung {name.replace('_', ' ')} is required and limited to 255 characters.")
|
||||
object.__setattr__(self, name, value)
|
||||
for name in ("validator_jar_sha256", "configuration_tree_sha256"):
|
||||
object.__setattr__(self, name, _sha256_text(getattr(self, name), name))
|
||||
for name in ("java_executable", "validator_jar", "configuration_root", "scenarios_file"):
|
||||
path = Path(getattr(self, name))
|
||||
if not path.is_absolute():
|
||||
raise ValueError(f"XRechnung {name.replace('_', ' ')} must be an absolute path.")
|
||||
object.__setattr__(self, name, path)
|
||||
if self.minimum_validation_steps < 1:
|
||||
raise ValueError("XRechnung minimum_validation_steps must be positive.")
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class InboundInvoice:
|
||||
tenant_id: str
|
||||
source_reference: str
|
||||
document: bytes
|
||||
received_at: datetime
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if not self.tenant_id.strip() or not self.source_reference.strip():
|
||||
raise ValueError("Inbound invoices require tenant and source references.")
|
||||
if not self.document or len(self.document) > MAX_INVOICE_BYTES:
|
||||
raise ValueError(f"Inbound invoice XML must contain 1 to {MAX_INVOICE_BYTES} bytes.")
|
||||
|
||||
@property
|
||||
def document_sha256(self) -> str:
|
||||
return hashlib.sha256(self.document).hexdigest()
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ValidationDiagnostic:
|
||||
level: str
|
||||
code: str | None
|
||||
text: str
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class XRechnungValidationResult:
|
||||
profile_id: str
|
||||
profile_sha256: str
|
||||
document_sha256: str
|
||||
technical_outcome: TechnicalOutcome
|
||||
conformance: ConformanceOutcome
|
||||
assessment: AssessmentOutcome
|
||||
observed_at: datetime
|
||||
report_sha256: str | None
|
||||
validation_step_count: int
|
||||
diagnostics: tuple[ValidationDiagnostic, ...]
|
||||
technical_reason: str | None = None
|
||||
|
||||
@property
|
||||
def handoff_allowed(self) -> bool:
|
||||
return (
|
||||
self.technical_outcome == "complete"
|
||||
and self.conformance == "valid"
|
||||
and self.assessment == "accept"
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ValidatedInvoiceHandoff:
|
||||
tenant_id: str
|
||||
source_reference: str
|
||||
document_sha256: str
|
||||
report_sha256: str
|
||||
validation_profile_id: str
|
||||
validation_profile_sha256: str
|
||||
validated_at: datetime
|
||||
handoff_sha256: str
|
||||
|
||||
|
||||
class KoSITValidator:
|
||||
def __init__(self, profile: KoSITValidationProfile, *, timeout_seconds: int = 120) -> None:
|
||||
if not 1 <= timeout_seconds <= 600:
|
||||
raise ValueError("KoSIT timeout must be between 1 and 600 seconds.")
|
||||
self.profile = profile
|
||||
self.timeout_seconds = timeout_seconds
|
||||
|
||||
def validate(self, invoice: InboundInvoice) -> XRechnungValidationResult:
|
||||
profile_sha256 = verify_profile(self.profile)
|
||||
_verify_safe_xml(invoice.document)
|
||||
with tempfile.TemporaryDirectory(prefix="govoplan-xrechnung-") as temporary:
|
||||
workdir = Path(temporary)
|
||||
input_file = workdir / "invoice.xml"
|
||||
input_file.write_bytes(invoice.document)
|
||||
command = [
|
||||
str(self.profile.java_executable),
|
||||
"-jar",
|
||||
str(self.profile.validator_jar),
|
||||
"-s",
|
||||
str(self.profile.scenarios_file),
|
||||
"-r",
|
||||
str(workdir),
|
||||
"-h",
|
||||
str(input_file),
|
||||
]
|
||||
env = {"LANG": "C.UTF-8", "LC_ALL": "C.UTF-8", "TZ": "UTC"}
|
||||
if os.environ.get("JAVA_HOME"):
|
||||
env["JAVA_HOME"] = os.environ["JAVA_HOME"]
|
||||
try:
|
||||
completed = subprocess.run( # noqa: S603 - absolute, digest-pinned executable and artifacts; no shell.
|
||||
command,
|
||||
cwd=self.profile.configuration_root,
|
||||
env=env,
|
||||
stdin=subprocess.DEVNULL,
|
||||
capture_output=True,
|
||||
check=False,
|
||||
timeout=self.timeout_seconds,
|
||||
)
|
||||
except subprocess.TimeoutExpired:
|
||||
return _technical_result(
|
||||
self.profile,
|
||||
profile_sha256,
|
||||
invoice.document_sha256,
|
||||
"failed",
|
||||
"KoSIT validation timed out; no handoff is allowed.",
|
||||
)
|
||||
output = completed.stdout + b"\n" + completed.stderr
|
||||
if len(output) > MAX_RUNNER_OUTPUT_BYTES:
|
||||
return _technical_result(
|
||||
self.profile,
|
||||
profile_sha256,
|
||||
invoice.document_sha256,
|
||||
"failed",
|
||||
"KoSIT runner output exceeded the safety limit.",
|
||||
)
|
||||
report_path = workdir / "invoice-report.xml"
|
||||
report = report_path.read_bytes() if report_path.is_file() else None
|
||||
return interpret_kosit_result(
|
||||
profile=self.profile,
|
||||
profile_sha256=profile_sha256,
|
||||
document_sha256=invoice.document_sha256,
|
||||
return_code=completed.returncode,
|
||||
runner_output=output,
|
||||
report=report,
|
||||
)
|
||||
|
||||
|
||||
def verify_profile(profile: KoSITValidationProfile) -> str:
|
||||
if not profile.java_executable.is_file() or not os.access(profile.java_executable, os.X_OK):
|
||||
raise XRechnungValidationError("The pinned Java executable is unavailable or not executable.")
|
||||
if not profile.validator_jar.is_file():
|
||||
raise XRechnungValidationError("The pinned KoSIT validator JAR is unavailable.")
|
||||
if _file_sha256(profile.validator_jar) != profile.validator_jar_sha256:
|
||||
raise XRechnungValidationError("The KoSIT validator JAR digest does not match the approved profile.")
|
||||
root = profile.configuration_root.resolve()
|
||||
if not root.is_dir():
|
||||
raise XRechnungValidationError("The pinned KoSIT configuration root is unavailable.")
|
||||
scenarios = profile.scenarios_file.resolve()
|
||||
if not scenarios.is_file() or not scenarios.is_relative_to(root):
|
||||
raise XRechnungValidationError("The scenarios file must exist inside the pinned configuration root.")
|
||||
observed_tree = configuration_tree_sha256(root)
|
||||
if observed_tree != profile.configuration_tree_sha256:
|
||||
raise XRechnungValidationError("The KoSIT configuration tree digest does not match the approved profile.")
|
||||
descriptor = {
|
||||
"profile_id": profile.profile_id,
|
||||
"xrechnung_version": profile.xrechnung_version,
|
||||
"validator_version": profile.validator_version,
|
||||
"configuration_version": profile.configuration_version,
|
||||
"validator_jar_sha256": profile.validator_jar_sha256,
|
||||
"configuration_tree_sha256": profile.configuration_tree_sha256,
|
||||
"scenarios_file": str(scenarios.relative_to(root)),
|
||||
"minimum_validation_steps": profile.minimum_validation_steps,
|
||||
}
|
||||
return hashlib.sha256(_canonical_json(descriptor)).hexdigest()
|
||||
|
||||
|
||||
def configuration_tree_sha256(root: Path) -> str:
|
||||
resolved = root.resolve()
|
||||
if not resolved.is_dir():
|
||||
raise XRechnungValidationError("KoSIT configuration tree is unavailable.")
|
||||
digest = hashlib.sha256()
|
||||
files = sorted(resolved.rglob("*"), key=lambda item: item.relative_to(resolved).as_posix())
|
||||
for path in files:
|
||||
if path.is_symlink():
|
||||
raise XRechnungValidationError("KoSIT configuration trees must not contain symbolic links.")
|
||||
if path.is_dir():
|
||||
continue
|
||||
if not path.is_file():
|
||||
raise XRechnungValidationError("KoSIT configuration trees may contain only regular files and directories.")
|
||||
relative = path.relative_to(resolved).as_posix().encode("utf-8")
|
||||
digest.update(len(relative).to_bytes(4, "big"))
|
||||
digest.update(relative)
|
||||
digest.update(bytes.fromhex(_file_sha256(path)))
|
||||
return digest.hexdigest()
|
||||
|
||||
|
||||
def interpret_kosit_result(
|
||||
*,
|
||||
profile: KoSITValidationProfile,
|
||||
profile_sha256: str,
|
||||
document_sha256: str,
|
||||
return_code: int,
|
||||
runner_output: bytes,
|
||||
report: bytes | None,
|
||||
) -> XRechnungValidationResult:
|
||||
if return_code != 0:
|
||||
return _technical_result(
|
||||
profile,
|
||||
profile_sha256,
|
||||
document_sha256,
|
||||
"failed",
|
||||
f"KoSIT runner exited with status {return_code}; report semantics are not trusted.",
|
||||
)
|
||||
if _TECHNICAL_OUTPUT.search(runner_output):
|
||||
return _technical_result(
|
||||
profile,
|
||||
profile_sha256,
|
||||
document_sha256,
|
||||
"incomplete",
|
||||
"KoSIT runner reported a technical error; a generated report is not trusted.",
|
||||
)
|
||||
if report is None or not report or len(report) > MAX_REPORT_BYTES:
|
||||
return _technical_result(
|
||||
profile,
|
||||
profile_sha256,
|
||||
document_sha256,
|
||||
"incomplete",
|
||||
"KoSIT did not produce a bounded XML report.",
|
||||
)
|
||||
report_sha256 = hashlib.sha256(report).hexdigest()
|
||||
try:
|
||||
root = ElementTree.fromstring(report)
|
||||
except ElementTree.ParseError:
|
||||
return _technical_result(
|
||||
profile,
|
||||
profile_sha256,
|
||||
document_sha256,
|
||||
"incomplete",
|
||||
"KoSIT produced malformed report XML.",
|
||||
report_sha256=report_sha256,
|
||||
)
|
||||
if root.tag != f"{{{VARL_NAMESPACE}}}report" or root.get("valid") not in {"true", "false"}:
|
||||
return _technical_result(
|
||||
profile,
|
||||
profile_sha256,
|
||||
document_sha256,
|
||||
"incomplete",
|
||||
"KoSIT report is missing the expected VARL report identity or validity flag.",
|
||||
report_sha256=report_sha256,
|
||||
)
|
||||
scenario = root.find(f"{{{VARL_NAMESPACE}}}scenarioMatched")
|
||||
steps = root.findall(f".//{{{VARL_NAMESPACE}}}validationStepResult")
|
||||
assessment = root.find(f"{{{VARL_NAMESPACE}}}assessment")
|
||||
accepts = [] if assessment is None else assessment.findall(f"{{{VARL_NAMESPACE}}}accept")
|
||||
rejects = [] if assessment is None else assessment.findall(f"{{{VARL_NAMESPACE}}}reject")
|
||||
steps_well_formed = all(item.get("valid") in {"true", "false"} for item in steps)
|
||||
if (
|
||||
scenario is None
|
||||
or len(steps) < profile.minimum_validation_steps
|
||||
or not steps_well_formed
|
||||
or (len(accepts), len(rejects)) not in {(1, 0), (0, 1)}
|
||||
):
|
||||
return _technical_result(
|
||||
profile,
|
||||
profile_sha256,
|
||||
document_sha256,
|
||||
"incomplete",
|
||||
"KoSIT report is partial or lacks the configured validation-step and assessment evidence.",
|
||||
report_sha256=report_sha256,
|
||||
validation_step_count=len(steps),
|
||||
)
|
||||
accepted = len(accepts) == 1
|
||||
valid_flag = root.get("valid") == "true"
|
||||
diagnostics = tuple(
|
||||
ValidationDiagnostic(
|
||||
level=(item.get("level") or "unknown")[:50],
|
||||
code=(item.get("code") or "")[:100] or None,
|
||||
text=" ".join("".join(item.itertext()).split())[:1000],
|
||||
)
|
||||
for item in root.findall(f".//{{{VARL_NAMESPACE}}}message")[:1000]
|
||||
)
|
||||
return XRechnungValidationResult(
|
||||
profile_id=profile.profile_id,
|
||||
profile_sha256=profile_sha256,
|
||||
document_sha256=document_sha256,
|
||||
technical_outcome="complete",
|
||||
conformance="valid" if valid_flag else "invalid",
|
||||
assessment="accept" if accepted else "reject",
|
||||
observed_at=datetime.now(UTC),
|
||||
report_sha256=report_sha256,
|
||||
validation_step_count=len(steps),
|
||||
diagnostics=diagnostics,
|
||||
)
|
||||
|
||||
|
||||
def create_validated_handoff(
|
||||
invoice: InboundInvoice,
|
||||
result: XRechnungValidationResult,
|
||||
) -> ValidatedInvoiceHandoff:
|
||||
if result.document_sha256 != invoice.document_sha256:
|
||||
raise XRechnungValidationError("Validation result belongs to another invoice document.")
|
||||
if not result.handoff_allowed or not result.report_sha256:
|
||||
raise XRechnungValidationError(
|
||||
"Invoice handoff requires technically complete and semantically valid XRechnung evidence."
|
||||
)
|
||||
payload = {
|
||||
"tenant_id": invoice.tenant_id,
|
||||
"source_reference": invoice.source_reference,
|
||||
"document_sha256": invoice.document_sha256,
|
||||
"report_sha256": result.report_sha256,
|
||||
"validation_profile_id": result.profile_id,
|
||||
"validation_profile_sha256": result.profile_sha256,
|
||||
"validated_at": result.observed_at.isoformat(),
|
||||
}
|
||||
return ValidatedInvoiceHandoff(
|
||||
tenant_id=invoice.tenant_id,
|
||||
source_reference=invoice.source_reference,
|
||||
document_sha256=invoice.document_sha256,
|
||||
report_sha256=result.report_sha256,
|
||||
validation_profile_id=result.profile_id,
|
||||
validation_profile_sha256=result.profile_sha256,
|
||||
validated_at=result.observed_at,
|
||||
handoff_sha256=hashlib.sha256(_canonical_json(payload)).hexdigest(),
|
||||
)
|
||||
|
||||
|
||||
def _verify_safe_xml(document: bytes) -> None:
|
||||
try:
|
||||
ElementTree.fromstring(document)
|
||||
except ElementTree.ParseError as exc:
|
||||
raise XRechnungValidationError("Inbound invoice is not well-formed safe XML.") from exc
|
||||
|
||||
|
||||
def _technical_result(
|
||||
profile: KoSITValidationProfile,
|
||||
profile_sha256: str,
|
||||
document_sha256: str,
|
||||
outcome: Literal["failed", "incomplete"],
|
||||
reason: str,
|
||||
*,
|
||||
report_sha256: str | None = None,
|
||||
validation_step_count: int = 0,
|
||||
) -> XRechnungValidationResult:
|
||||
return XRechnungValidationResult(
|
||||
profile_id=profile.profile_id,
|
||||
profile_sha256=profile_sha256,
|
||||
document_sha256=document_sha256,
|
||||
technical_outcome=outcome,
|
||||
conformance="unknown",
|
||||
assessment="unknown",
|
||||
observed_at=datetime.now(UTC),
|
||||
report_sha256=report_sha256,
|
||||
validation_step_count=validation_step_count,
|
||||
diagnostics=(),
|
||||
technical_reason=reason,
|
||||
)
|
||||
|
||||
|
||||
def _file_sha256(path: Path) -> str:
|
||||
digest = hashlib.sha256()
|
||||
with path.open("rb") as handle:
|
||||
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
||||
digest.update(chunk)
|
||||
return digest.hexdigest()
|
||||
|
||||
|
||||
def _sha256_text(value: str, label: str) -> str:
|
||||
normalized = str(value or "").strip().lower().removeprefix("sha256:")
|
||||
if len(normalized) != 64 or any(character not in "0123456789abcdef" for character in normalized):
|
||||
raise ValueError(f"XRechnung {label.replace('_', ' ')} must be a SHA-256 digest.")
|
||||
return normalized
|
||||
|
||||
|
||||
def _canonical_json(value: object) -> bytes:
|
||||
return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
|
||||
|
||||
|
||||
__all__ = [
|
||||
"InboundInvoice",
|
||||
"KoSITValidationProfile",
|
||||
"KoSITValidator",
|
||||
"ValidatedInvoiceHandoff",
|
||||
"ValidationDiagnostic",
|
||||
"XRechnungValidationError",
|
||||
"XRechnungValidationResult",
|
||||
"configuration_tree_sha256",
|
||||
"create_validated_handoff",
|
||||
"interpret_kosit_result",
|
||||
"verify_profile",
|
||||
]
|
||||
@@ -0,0 +1 @@
|
||||
|
||||
Reference in New Issue
Block a user