#!/usr/bin/env python3 """Signed, version-bound translation memory and publication-readiness gate. The host owns the SQLite connection, quality-receipt verifier, and isolated approval authority. This module never reads a signing key and never publishes. """ from __future__ import annotations import hashlib import importlib.util import json import re import sqlite3 import sys import unicodedata from contextlib import contextmanager from dataclasses import dataclass from pathlib import Path from typing import Any, Iterator, Protocol SCHEMA_VERSION = 1 APPROVAL_SCHEMA = "blun.website-localization-approval.v10" RECEIPT_BINDING_SCHEMA = "blun.localization-quality-receipt-binding.v9" INDEPENDENT_MODEL_REVIEW_SCHEMA = "blun.independent-model-review.v1" PUBLICATION_EVIDENCE_SCHEMA = "blun.website-localization-release-evidence.v14" PUBLICATION_EVIDENCE_CAPABILITIES_SCHEMA = ( "blun.website-localization-release-evidence-capabilities.v8" ) PUBLICATION_EVIDENCE_FIELDS = ( "schema", "release_evidence_contract_sha256", "job_id", "target_locale", "target_sha256", "approval_id", "content_type", "result_sha256", "approval_sha256", "quality_receipt_sha256", "evidence_request_id", "evidence_revision", "quality_profile", "commercial_profile", "commercial_quality_profile", "commercial_review", "commercial_review_routing_contract_sha256", "commercial_review_resolution_contract_sha256", "commercial_review_resolution", ) MAX_TEXT_BYTES = 2_000_000 MAX_RECEIPT_LENGTH = 16_384 MAX_TTL_SECONDS = 31_536_000.0 HEX64 = re.compile(r"^[0-9a-f]{64}$") TOKEN = re.compile(r"^[A-Za-z0-9_.:-]{1,256}$") EVIDENCE_REQUEST_ID = re.compile(r"^blun-l10n-evidence-[0-9a-f]{64}$") _COLUMNS = ( "approval_id", "job_id", "target_locale", "target_sha256", "result_json", "result_sha256", "approval_json", "approval_sha256", "signature_algorithm", "key_id", "signature", "approved_at", "expires_at", ) def _load_module(name: str, path: Path): spec = importlib.util.spec_from_file_location(name, path) if spec is None or spec.loader is None: raise RuntimeError(f"cannot load required release dependency: {path.name}") module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) return module _ROOT = Path(__file__).resolve().parents[1] _QUEUE = _load_module( "blun_website_localization_release_queue", _ROOT / "integrations" / "website_localization_queue.py", ) _WORKER = _load_module( "blun_website_localization_release_worker", _ROOT / "integrations" / "website_localization_worker.py", ) COMMERCIAL_REVIEW_RESOLUTION_SCHEMA = ( _WORKER._COMMERCIAL.REVIEW_RESOLUTION_SCHEMA ) class LocalizationReleaseBlocked(RuntimeError): """Stable, content-free release failure.""" def __init__(self, code: str, *, retryable: bool = False): if not isinstance(code, str) or TOKEN.fullmatch(code) is None: raise ValueError("release failure code is invalid") if not isinstance(retryable, bool): raise ValueError("release failure retryability must be boolean") super().__init__(code) self.code = code self.retryable = retryable @dataclass(frozen=True) class ApprovalSignature: algorithm: str key_id: str signature: str class ApprovalAuthority(Protocol): def sign(self, payload: bytes) -> ApprovalSignature: ... def verify(self, payload: bytes, signature: ApprovalSignature) -> bool: ... class QualityReceiptVerifier(Protocol): def verify( self, *, binding: dict[str, Any], receipt: str, ) -> bool: ... class IndependentModelReviewVerifier(Protocol): def verify( self, *, binding: dict[str, Any], receipt: str, ) -> bool: ... @dataclass(frozen=True) class ApprovedLocalization: approval_id: str job_id: str target_locale: str target_sha256: str candidate: str release_evidence: dict[str, Any] approved_at: float expires_at: float @dataclass(frozen=True) class WebsiteReadiness: plan_id: str ready: bool required_locales: tuple[str, ...] approved_locales: tuple[str, ...] blocked: tuple[tuple[str, str], ...] def validate_publication_evidence( value: Any, *, require_current_locale_quality: bool = True, ) -> dict[str, Any]: """Validate the content-free release proof carried to a CMS.""" if ( not isinstance(require_current_locale_quality, bool) or not isinstance(value, dict) or set(value) != set(PUBLICATION_EVIDENCE_FIELDS) ): raise LocalizationReleaseBlocked("publication.evidence.invalid") if ( value.get("schema") != PUBLICATION_EVIDENCE_SCHEMA or value.get("release_evidence_contract_sha256") != publication_evidence_contract()["sha256"] or any( not isinstance(value.get(field), str) or TOKEN.fullmatch(value[field]) is None for field in ("job_id", "target_locale", "approval_id", "content_type") ) or any( HEX64.fullmatch(str(value.get(field))) is None for field in ( "target_sha256", "result_sha256", "approval_sha256", "quality_receipt_sha256", ) ) or not isinstance(value.get("evidence_request_id"), str) or EVIDENCE_REQUEST_ID.fullmatch(value["evidence_request_id"]) is None or not isinstance(value.get("evidence_revision"), str) or TOKEN.fullmatch(value["evidence_revision"]) is None ): raise LocalizationReleaseBlocked("publication.evidence.invalid") quality_profile = value.get("quality_profile") if ( not isinstance(quality_profile, dict) or set(quality_profile) != {"locale", "version", "sha256"} or quality_profile.get("locale") != value["target_locale"] or not isinstance(quality_profile.get("version"), str) or TOKEN.fullmatch(quality_profile["version"]) is None or HEX64.fullmatch(str(quality_profile.get("sha256"))) is None ): raise LocalizationReleaseBlocked("publication.evidence.invalid") if require_current_locale_quality: try: current_quality = _WORKER._PLANNER.quality_profile_for( value["target_locale"], ) expected_quality_profile = { "locale": current_quality["locale"], "version": current_quality["version"], "sha256": current_quality["sha256"], } except Exception: raise LocalizationReleaseBlocked( "publication.evidence.invalid" ) from None if quality_profile != expected_quality_profile: raise LocalizationReleaseBlocked("publication.evidence.invalid") profile = value.get("commercial_profile") commercial_quality_profile = value.get("commercial_quality_profile") review = value.get("commercial_review") routing_contract_sha256 = value.get( "commercial_review_routing_contract_sha256" ) resolution_contract_sha256 = value.get( "commercial_review_resolution_contract_sha256" ) resolution = value.get("commercial_review_resolution") if ( len({ profile is None, commercial_quality_profile is None, review is None, }) != 1 or (profile is None) != (routing_contract_sha256 is None) or (profile is None) != (resolution_contract_sha256 is None) or (value["content_type"] == "commercial") != (profile is not None) ): raise LocalizationReleaseBlocked("publication.evidence.invalid") if profile is not None: if not isinstance(profile, str) or TOKEN.fullmatch(profile) is None: raise LocalizationReleaseBlocked("publication.evidence.invalid") expected_commercial_quality = None if require_current_locale_quality: try: current_commercial_quality = ( _WORKER._PLANNER.commercial_quality_profile_for( value["target_locale"], ) ) expected_commercial_quality = { "profile": current_commercial_quality[ "commercial_profile" ], "version": current_commercial_quality["version"], "sha256": current_commercial_quality["sha256"], } except Exception: raise LocalizationReleaseBlocked( "publication.evidence.invalid" ) from None if ( profile != _WORKER._PLANNER.COMMERCIAL_PROFILE or ( require_current_locale_quality and commercial_quality_profile != expected_commercial_quality ) or HEX64.fullmatch(str(routing_contract_sha256)) is None or routing_contract_sha256 != _WORKER._COMMERCIAL.public_review_routing_contract(profile)[ "sha256" ] or HEX64.fullmatch(str(resolution_contract_sha256)) is None or resolution_contract_sha256 != _WORKER._COMMERCIAL.public_review_resolution_contract(profile)[ "sha256" ] ): raise LocalizationReleaseBlocked("publication.evidence.invalid") if ( not isinstance(commercial_quality_profile, dict) or set(commercial_quality_profile) != {"profile", "version", "sha256"} or commercial_quality_profile.get("profile") != profile or not isinstance(commercial_quality_profile.get("version"), str) or TOKEN.fullmatch(commercial_quality_profile["version"]) is None or HEX64.fullmatch( str(commercial_quality_profile.get("sha256")) ) is None ): raise LocalizationReleaseBlocked("publication.evidence.invalid") try: review = _WORKER._COMMERCIAL.validate_summary( review, profile, review_required=( isinstance(review, dict) and review.get("status") == "review_required" ), ) except _WORKER._COMMERCIAL.CommercialReviewBlocked: raise LocalizationReleaseBlocked("publication.evidence.invalid") from None if review["status"] == "verified": if resolution is not None: raise LocalizationReleaseBlocked("publication.evidence.invalid") else: try: resolution = _WORKER._COMMERCIAL.validate_review_resolution( resolution, review, profile, ) except _WORKER._COMMERCIAL.CommercialReviewBlocked: raise LocalizationReleaseBlocked("publication.evidence.invalid") elif resolution is not None: raise LocalizationReleaseBlocked("publication.evidence.invalid") return json.loads(_canonical_json({ **value, "commercial_review": review, "commercial_review_resolution": resolution, })) def _commercial_review_resolution( result: dict[str, Any], approval: dict[str, Any], ) -> dict[str, Any] | None: review = result["commercial_review"] if ( result["content_type"] != "commercial" or review["status"] == "verified" ): return None human_receipt_sha256 = approval["human_review_receipt_sha256"] independent = approval["independent_model_review"] if human_receipt_sha256 is not None: method = "qualified_human" receipt_sha256 = human_receipt_sha256 provider = None else: method = "independent_model" receipt_sha256 = independent["receipt_sha256"] provider = json.loads(_canonical_json(independent["provider"])) return { "schema": COMMERCIAL_REVIEW_RESOLUTION_SCHEMA, "profile": review["profile"], "contract_sha256": approval[ "commercial_review_resolution_contract_sha256" ], "status": "resolved", "reviewed_dimensions": json.loads(_canonical_json( review["review_required_dimensions"] )), "reviewed_offer_count": review["offer_count"], "reviewed_offers": json.loads(_canonical_json( review["review_required_offers"] )), "method": method, "receipt_sha256": receipt_sha256, "primary_provider": json.loads(_canonical_json(result["provider"])), "provider": provider, } @dataclass(frozen=True) class VerifiedResultCache: """Runner adapter for exact, signed translation-memory results.""" store: Any authority: ApprovalAuthority def resolve( self, job_payload: dict[str, Any], *, now: float, ) -> dict[str, Any] | None: return self.store.cached_result(job_payload, self.authority, now=now) def _canonical_json(value: Any) -> str: try: return json.dumps( value, ensure_ascii=False, allow_nan=False, sort_keys=True, separators=(",", ":"), ) except (TypeError, ValueError) as error: raise LocalizationReleaseBlocked("release.json.invalid") from error def _hash_text(value: str) -> str: return hashlib.sha256(value.encode("utf-8")).hexdigest() def _publication_evidence_contract_body() -> dict[str, Any]: return { "schema": PUBLICATION_EVIDENCE_CAPABILITIES_SCHEMA, "release_evidence_schema": PUBLICATION_EVIDENCE_SCHEMA, "required_fields": list(PUBLICATION_EVIDENCE_FIELDS), "bindings": { "sha256_fields": [ "release_evidence_contract_sha256", "target_sha256", "result_sha256", "approval_sha256", "quality_receipt_sha256", "commercial_review_routing_contract_sha256", "commercial_review_resolution_contract_sha256", ], "lineage_fields": ["evidence_request_id", "evidence_revision"], "target_identity_fields": [ "job_id", "target_locale", "target_sha256", "approval_id", ], "quality_profile": "exact-current-target-locale-quality-profile", "signed_container": "blun.cms-localization-publication.v3", }, "commercial_scope": { "content_type": "commercial", "required_non_null": [ "commercial_profile", "commercial_quality_profile", "commercial_review", "commercial_review_routing_contract_sha256", "commercial_review_resolution_contract_sha256", ], "review_evidence_contract_sha256": ( "exact-current-public-commercial-evidence-contract" ), "review_resolution_contract_sha256": ( "exact-current-public-commercial-resolution-contract" ), "quality_profile": ( "exact-current-target-locale-commercial-quality-profile" ), "resolution": "required-only-when-review-required", "non_commercial_fields": "all-null", }, "content_policy": { "source_text": False, "target_text": False, "raw_receipt": False, "reviewer_identity": False, "reviewer_prose": False, "commercial_review_routing": False, "project_prices": False, "project_brands": False, "authentication_material": False, }, "tamper_policy": "reject-complete-publication-before-host-commit", } def publication_evidence_contract() -> dict[str, Any]: """Return the exact content-free contract advertised to CMS adapters.""" body = _publication_evidence_contract_body() return {**body, "sha256": _hash_text(_canonical_json(body))} def validate_publication_evidence_contract(value: Any) -> dict[str, Any]: """Reject a partial, reordered, altered, or merely rehashed contract.""" body = _publication_evidence_contract_body() expected = {**body, "sha256": _hash_text(_canonical_json(body))} if value != expected: raise LocalizationReleaseBlocked("publication.evidence_contract.invalid") return json.loads(_canonical_json(expected)) def _text(value: Any, code: str, *, limit: int = 256) -> str: if not isinstance(value, str) or not value or value != value.strip(): raise LocalizationReleaseBlocked(code) if len(value) > limit or "\x00" in value or not unicodedata.is_normalized("NFC", value): raise LocalizationReleaseBlocked(code) return value def _timestamp(value: Any, code: str) -> float: if isinstance(value, bool) or not isinstance(value, (int, float)): raise LocalizationReleaseBlocked(code) value = float(value) if value < 0 or value != value or value in {float("inf"), float("-inf")}: raise LocalizationReleaseBlocked(code) return value def _receipt(value: Any, code: str) -> str: return _text(value, code, limit=MAX_RECEIPT_LENGTH) def _independent_model_review(value: Any, primary_provider: dict[str, Any]) -> tuple[dict[str, str], str]: if not isinstance(value, dict) or set(value) != {"schema", "provider", "receipt"}: raise LocalizationReleaseBlocked("independent_model_review.invalid") provider = value.get("provider") if not isinstance(provider, dict) or set(provider) != {"id", "model_id", "model_version"}: raise LocalizationReleaseBlocked("independent_model_review.provider.invalid") normalized = { name: _text(provider.get(name), "independent_model_review.provider.invalid") for name in ("id", "model_id", "model_version") } if value.get("schema") != INDEPENDENT_MODEL_REVIEW_SCHEMA: raise LocalizationReleaseBlocked("independent_model_review.schema.invalid") if normalized["id"] == primary_provider.get("id"): raise LocalizationReleaseBlocked("independent_model_review.not_independent") return normalized, _receipt(value.get("receipt"), "independent_model_review.receipt.required") def _commercial_resolution_contract_sha256( job: dict[str, Any], result: dict[str, Any], ) -> str | None: commercial_profile = job.get("commercial_profile") commercial_review = result["commercial_review"] if commercial_profile is None or commercial_review["status"] != "review_required": return None try: expected = _WORKER._COMMERCIAL.public_review_resolution_contract( commercial_profile, )["sha256"] except (KeyError, TypeError, ValueError): raise LocalizationReleaseBlocked("approval.binding_mismatch") from None if ( job.get("commercial_review_resolution_contract_sha256") != expected or result.get("commercial_review_resolution_contract_sha256") != expected ): raise LocalizationReleaseBlocked("approval.binding_mismatch") return expected def _receipt_binding( job_id: str, job: dict[str, Any], result: dict[str, Any], result_sha256: str, *, evidence_request_id: str, evidence_revision: str, review_kind: str, review_provider: dict[str, str] | None = None, ) -> dict[str, Any]: """Build the exact canonical context that an evidence receipt must bind.""" if review_kind not in {"quality", "qualified_human", "independent_model"}: raise LocalizationReleaseBlocked("review.binding.invalid") if (review_kind == "independent_model") != (review_provider is not None): raise LocalizationReleaseBlocked("review.binding.invalid") if ( not isinstance(evidence_request_id, str) or EVIDENCE_REQUEST_ID.fullmatch(evidence_request_id) is None or not isinstance(evidence_revision, str) or TOKEN.fullmatch(evidence_revision) is None ): raise LocalizationReleaseBlocked("review.evidence_context.invalid") commercial_profile = job.get("commercial_profile") commercial_review = result["commercial_review"] resolution_contract_sha256 = _commercial_resolution_contract_sha256( job, result, ) binding = { "schema": RECEIPT_BINDING_SCHEMA, "review_kind": review_kind, "evidence_request_id": evidence_request_id, "evidence_revision": evidence_revision, "job_id": job_id, "result_sha256": result_sha256, "source_text": job["source"]["text"], "target_text": result["candidate"], "source_sha256": result["source_sha256"], "target_sha256": result["target_sha256"], "source_locale": result["source_locale"], "target_locale": result["target_locale"], "content_type": result["content_type"], "glossary_version": result["glossary_version"], "policy_version": result["policy_version"], "primary_provider": result["provider"], "review_provider": review_provider, "software_version": result["software_version"], "review_confidence": result["review_confidence"], "quality_profile": result["quality_profile"], "commercial_profile": commercial_profile, "commercial_review": commercial_review, "commercial_review_routing": result["commercial_review_routing"], "commercial_review_routing_contract_sha256": result[ "commercial_review_routing_contract_sha256" ], "commercial_review_resolution_contract_sha256": ( resolution_contract_sha256 ), "human_review_required": result["human_review_required"], "independent_review_required": result["independent_review_required"], } return json.loads(_canonical_json(binding)) def _verification_failure(error: Exception, verifier_kind: str) -> None: if getattr(type(error), "localization_receipt_verification_failure", None) is not True: return code = getattr(error, "code", None) retryable = getattr(error, "retryable", None) combined = f"{verifier_kind}.verifier.{code}" if ( isinstance(code, str) and TOKEN.fullmatch(code) is not None and isinstance(retryable, bool) and len(combined) <= 256 ): raise LocalizationReleaseBlocked(combined, retryable=retryable) from None def _signature(value: Any) -> ApprovalSignature: if not isinstance(value, ApprovalSignature): raise LocalizationReleaseBlocked("approval.signature.invalid") for item in (value.algorithm, value.key_id, value.signature): if not isinstance(item, str) or TOKEN.fullmatch(item) is None: raise LocalizationReleaseBlocked("approval.signature.invalid") return value @contextmanager def _transaction(connection: sqlite3.Connection) -> Iterator[None]: if connection.in_transaction: raise LocalizationReleaseBlocked("release.transaction.external") try: connection.execute("BEGIN IMMEDIATE") yield except Exception: connection.rollback() raise else: connection.commit() def _plan_job(plan: Any, job_id: str) -> tuple[str, dict[str, Any]]: plan_id = _text(getattr(plan, "plan_id", None), "plan.invalid") jobs = getattr(plan, "jobs", None) if not isinstance(jobs, tuple) or not jobs: raise LocalizationReleaseBlocked("plan.invalid") matches = [job for job in jobs if getattr(job, "job_id", None) == job_id] if len(matches) != 1 or not callable(getattr(matches[0], "as_payload", None)): raise LocalizationReleaseBlocked("plan.job.missing") payload = matches[0].as_payload() if not isinstance(payload, dict) or payload.get("job_id") != job_id: raise LocalizationReleaseBlocked("plan.job.invalid") return plan_id, _validated_job_payload(payload) def _validated_job_payload(payload: Any) -> dict[str, Any]: try: return _WORKER._validated_job(payload) except _WORKER.LocalizationWorkerBlocked: raise LocalizationReleaseBlocked("plan.job.invalid") from None def _validate_result(job: dict[str, Any], result: Any) -> dict[str, Any]: if not isinstance(result, dict): raise LocalizationReleaseBlocked("result.invalid") expected = { "schema", "worker_schema", "job_id", "source_sha256", "target_sha256", "source_locale", "target_locale", "content_type", "glossary_version", "policy_version", "provider", "software_version", "candidate", "quality_passes", "integrity", "review_confidence", "quality_profile", "commercial_review", "commercial_review_routing", "commercial_review_routing_contract_sha256", "commercial_review_resolution_contract_sha256", "human_review_required", "independent_review_required", "release_required", } if set(result) != expected: raise LocalizationReleaseBlocked("result.invalid") candidate = result.get("candidate") if not isinstance(candidate, str) or not candidate or len(candidate.encode("utf-8")) > MAX_TEXT_BYTES: raise LocalizationReleaseBlocked("result.invalid") if "\x00" in candidate or not unicodedata.is_normalized("NFC", candidate): raise LocalizationReleaseBlocked("result.invalid") binding = { "job_id": job["job_id"], "source_sha256": job["source"]["sha256"], "source_locale": job["source"]["locale"], "target_locale": job["target"]["locale"], "content_type": job["content_type"], "glossary_version": job["glossary_version"], "policy_version": job["policy_version"], "provider": job["provider"], "software_version": job["software_version"], } if any(result.get(name) != value for name, value in binding.items()): raise LocalizationReleaseBlocked("result.binding_mismatch") expected_quality_profile = { "locale": job["target"]["locale"], "version": job["target"]["quality_profile_version"], "sha256": job["target"]["quality_profile_sha256"], } if job["content_type"] == "commercial": expected_quality_profile["commercial"] = { "profile": job["commercial_profile"], "version": job["commercial_quality_profile"]["version"], "sha256": job["commercial_quality_profile"]["sha256"], } if result.get("quality_profile") != expected_quality_profile: raise LocalizationReleaseBlocked("result.quality_profile.invalid") if result.get("schema") != _WORKER.RESULT_SCHEMA or result.get("worker_schema") != _WORKER.WORKER_SCHEMA: raise LocalizationReleaseBlocked("result.schema.invalid") if result.get("target_sha256") != _hash_text(candidate): raise LocalizationReleaseBlocked("result.target_hash.invalid") phases = result.get("quality_passes") if not isinstance(phases, list) or [item.get("phase") for item in phases if isinstance(item, dict)] != list(_WORKER.PHASES): raise LocalizationReleaseBlocked("result.quality.invalid") for item in phases: if set(item) != {"phase", "request_sha256", "response_sha256", "status"}: raise LocalizationReleaseBlocked("result.quality.invalid") if item["status"] != "PASS" or HEX64.fullmatch(str(item["request_sha256"])) is None: raise LocalizationReleaseBlocked("result.quality.invalid") if HEX64.fullmatch(str(item["response_sha256"])) is None: raise LocalizationReleaseBlocked("result.quality.invalid") if result.get("integrity") != {"status": "PASS", "guard": "translate-native-structure-and-token-gate"}: raise LocalizationReleaseBlocked("result.integrity.invalid") if result.get("release_required") is not True: raise LocalizationReleaseBlocked("result.release.invalid") review_confidence = result.get("review_confidence") if ( not isinstance(review_confidence, dict) or set(review_confidence) != {"target_native", "source_fidelity"} or any(value not in {"high", "low"} for value in review_confidence.values()) ): raise LocalizationReleaseBlocked("result.review_confidence.invalid") human_review_required = result.get("human_review_required") expected_human_review = job["content_type"] == "legal" if not isinstance(human_review_required, bool) or human_review_required is not expected_human_review: raise LocalizationReleaseBlocked("result.human_review.invalid") independent_review_required = result.get("independent_review_required") expected_independent_review = ( job["content_type"] != "legal" and "low" in review_confidence.values() ) if ( not isinstance(independent_review_required, bool) or independent_review_required is not expected_independent_review ): raise LocalizationReleaseBlocked("result.independent_review.invalid") commercial_review = result.get("commercial_review") commercial_review_routing = result.get("commercial_review_routing") commercial_routing_contract_sha256 = result.get( "commercial_review_routing_contract_sha256" ) commercial_resolution_contract_sha256 = result.get( "commercial_review_resolution_contract_sha256" ) if job["content_type"] == "commercial": expected_routing_contract_sha256 = job[ "commercial_review_routing_contract_sha256" ] if commercial_routing_contract_sha256 != expected_routing_contract_sha256: raise LocalizationReleaseBlocked("result.commercial_review.invalid") if commercial_resolution_contract_sha256 != job[ "commercial_review_resolution_contract_sha256" ]: raise LocalizationReleaseBlocked("result.commercial_review.invalid") try: commercial_review = _WORKER._COMMERCIAL.validate_summary( commercial_review, job["commercial_profile"], review_required=independent_review_required, ) if commercial_review["status"] == "review_required": _WORKER._COMMERCIAL.validate_review_routing_context( commercial_review_routing, job["source"]["text"], candidate, commercial_review, job["commercial_profile"], ) elif commercial_review_routing is not None: raise _WORKER._COMMERCIAL.CommercialReviewBlocked( "review.commercial.routing_invalid", ) except _WORKER._COMMERCIAL.CommercialReviewBlocked: raise LocalizationReleaseBlocked("result.commercial_review.invalid") from None elif any(value is not None for value in ( commercial_review, commercial_review_routing, commercial_routing_contract_sha256, commercial_resolution_contract_sha256, )): raise LocalizationReleaseBlocked("result.commercial_review.invalid") return result class LocalizationReleaseStore: """Append-only approvals and exact-version translation-memory lookups.""" def __init__(self, connection: sqlite3.Connection, queue: Any): if not isinstance(connection, sqlite3.Connection): raise LocalizationReleaseBlocked("release.connection.invalid") if not isinstance(queue, _QUEUE.LocalizationQueue): raise LocalizationReleaseBlocked("release.queue.invalid") self.connection = connection self.connection.row_factory = sqlite3.Row self.queue = queue version = int(connection.execute("PRAGMA user_version").fetchone()[0]) if version not in {0, SCHEMA_VERSION}: raise LocalizationReleaseBlocked("release.schema.unsupported") if version == 0: with _transaction(connection): connection.execute(""" CREATE TABLE localization_approvals ( approval_id TEXT PRIMARY KEY, job_id TEXT NOT NULL, target_locale TEXT NOT NULL, target_sha256 TEXT NOT NULL, result_json TEXT NOT NULL, result_sha256 TEXT NOT NULL, approval_json TEXT NOT NULL, approval_sha256 TEXT NOT NULL, signature_algorithm TEXT NOT NULL, key_id TEXT NOT NULL, signature TEXT NOT NULL, approved_at REAL NOT NULL, expires_at REAL NOT NULL ) """) connection.execute(""" CREATE INDEX localization_approvals_job ON localization_approvals (job_id, approved_at DESC) """) connection.execute(f"PRAGMA user_version = {SCHEMA_VERSION}") columns = tuple(row["name"] for row in connection.execute("PRAGMA table_info(localization_approvals)")) if columns != _COLUMNS: raise LocalizationReleaseBlocked("release.schema.altered") def validated_result(self, plan: Any, job_id: str) -> dict[str, Any]: """Return an exact completed result only after all release checks pass.""" job_id = _text(job_id, "job.id.invalid") _, job = _plan_job(plan, job_id) result = _validate_result(job, self.queue.result(job_id)) return json.loads(_canonical_json(result)) def approve( self, plan: Any, job_id: str, quality_receipt: str, quality_verifier: QualityReceiptVerifier, authority: ApprovalAuthority, *, evidence_request_id: str, evidence_revision: str, now: float | int, ttl_seconds: float | int = 2_592_000, human_review_receipt: str | None = None, human_review_verifier: QualityReceiptVerifier | None = None, independent_model_review: dict[str, Any] | None = None, independent_model_review_verifier: IndependentModelReviewVerifier | None = None, ) -> ApprovedLocalization: job_id = _text(job_id, "job.id.invalid") _, job = _plan_job(plan, job_id) result = self.validated_result(plan, job_id) now = _timestamp(now, "approval.time.invalid") ttl = _timestamp(ttl_seconds, "approval.ttl.invalid") if ttl <= 0 or ttl > MAX_TTL_SECONDS: raise LocalizationReleaseBlocked("approval.ttl.invalid") result_json = _canonical_json(result) result_hash = _hash_text(result_json) quality_receipt = _receipt(quality_receipt, "quality.receipt.invalid") if ( not isinstance(evidence_request_id, str) or EVIDENCE_REQUEST_ID.fullmatch(evidence_request_id) is None or not isinstance(evidence_revision, str) or TOKEN.fullmatch(evidence_revision) is None ): raise LocalizationReleaseBlocked("review.evidence_context.invalid") verifier = getattr(quality_verifier, "verify", None) if not callable(verifier): raise LocalizationReleaseBlocked("quality.verifier.invalid") try: quality_ok = verifier( binding=_receipt_binding( job_id, job, result, result_hash, evidence_request_id=evidence_request_id, evidence_revision=evidence_revision, review_kind="quality", ), receipt=quality_receipt, ) is True except Exception as error: _verification_failure(error, "quality") quality_ok = False if not quality_ok: raise LocalizationReleaseBlocked("quality.receipt.rejected") human_hash = None independent_binding = None if result["human_review_required"]: if independent_model_review is not None: raise LocalizationReleaseBlocked("independent_model_review.legal_forbidden") human_review_receipt = _receipt(human_review_receipt, "human.receipt.required") human_verify = getattr(human_review_verifier, "verify", None) if not callable(human_verify): raise LocalizationReleaseBlocked("human.verifier.required") try: human_ok = human_verify( binding=_receipt_binding( job_id, job, result, result_hash, evidence_request_id=evidence_request_id, evidence_revision=evidence_revision, review_kind="qualified_human", ), receipt=human_review_receipt, ) is True except Exception as error: _verification_failure(error, "human") human_ok = False if not human_ok: raise LocalizationReleaseBlocked("human.receipt.rejected") human_hash = _hash_text(human_review_receipt) elif result["independent_review_required"]: if human_review_receipt is not None and independent_model_review is not None: raise LocalizationReleaseBlocked("review.escalation.ambiguous") if independent_model_review is not None: provider, model_receipt = _independent_model_review( independent_model_review, result["provider"], ) model_verify = getattr(independent_model_review_verifier, "verify", None) if not callable(model_verify): raise LocalizationReleaseBlocked("independent_model_review.verifier.required") try: model_ok = model_verify( binding=_receipt_binding( job_id, job, result, result_hash, evidence_request_id=evidence_request_id, evidence_revision=evidence_revision, review_kind="independent_model", review_provider=provider, ), receipt=model_receipt, ) is True except Exception as error: _verification_failure(error, "independent_model_review") model_ok = False if not model_ok: raise LocalizationReleaseBlocked("independent_model_review.receipt.rejected") independent_binding = { "schema": INDEPENDENT_MODEL_REVIEW_SCHEMA, "provider": provider, "receipt_sha256": _hash_text(model_receipt), } else: human_review_receipt = _receipt(human_review_receipt, "human.receipt.required") human_verify = getattr(human_review_verifier, "verify", None) if not callable(human_verify): raise LocalizationReleaseBlocked("human.verifier.required") try: human_ok = human_verify( binding=_receipt_binding( job_id, job, result, result_hash, evidence_request_id=evidence_request_id, evidence_revision=evidence_revision, review_kind="qualified_human", ), receipt=human_review_receipt, ) is True except Exception as error: _verification_failure(error, "human") human_ok = False if not human_ok: raise LocalizationReleaseBlocked("human.receipt.rejected") human_hash = _hash_text(human_review_receipt) elif any(value is not None for value in ( human_review_receipt, human_review_verifier, independent_model_review, independent_model_review_verifier, )): raise LocalizationReleaseBlocked("review.escalation.unexpected") immutable = { "schema": APPROVAL_SCHEMA, "release_evidence_contract_sha256": ( publication_evidence_contract()["sha256"] ), "evidence_request_id": evidence_request_id, "evidence_revision": evidence_revision, "job_id": job_id, "source_sha256": result["source_sha256"], "target_sha256": result["target_sha256"], "source_locale": result["source_locale"], "target_locale": result["target_locale"], "content_type": result["content_type"], "glossary_version": result["glossary_version"], "policy_version": result["policy_version"], "provider": result["provider"], "software_version": result["software_version"], "worker_schema": result["worker_schema"], "review_confidence": result["review_confidence"], "quality_profile": result["quality_profile"], "commercial_review_routing_contract_sha256": result[ "commercial_review_routing_contract_sha256" ], "result_sha256": result_hash, "quality_receipt_sha256": _hash_text(quality_receipt), "human_review_receipt_sha256": human_hash, "independent_model_review": independent_binding, "commercial_review_resolution_contract_sha256": ( _commercial_resolution_contract_sha256(job, result) ), } approval_id = "blun-l10n-approval-" + _hash_text(_canonical_json(immutable)) payload = { **immutable, "approval_id": approval_id, "approved_at": now, "expires_at": now + ttl, } approval_json = _canonical_json(payload) approval_bytes = approval_json.encode("utf-8") prior = self.connection.execute( "SELECT target_sha256 FROM localization_approvals WHERE job_id = ? LIMIT 1", (job_id,), ).fetchone() if prior is not None and prior["target_sha256"] != result["target_sha256"]: raise LocalizationReleaseBlocked("translation_memory.immutable") sign = getattr(authority, "sign", None) verify = getattr(authority, "verify", None) if not callable(sign) or not callable(verify): raise LocalizationReleaseBlocked("approval.authority.invalid") try: signed = _signature(sign(approval_bytes)) verified = verify(approval_bytes, signed) is True except LocalizationReleaseBlocked: raise except Exception: raise LocalizationReleaseBlocked("approval.signing.failed") from None if not verified: raise LocalizationReleaseBlocked("approval.signature.rejected") with _transaction(self.connection): prior = self.connection.execute( "SELECT target_sha256 FROM localization_approvals WHERE job_id = ? LIMIT 1", (job_id,), ).fetchone() if prior is not None and prior["target_sha256"] != result["target_sha256"]: raise LocalizationReleaseBlocked("translation_memory.immutable") existing = self.connection.execute( "SELECT * FROM localization_approvals WHERE approval_id = ?", (approval_id,), ).fetchone() if existing is None: self.connection.execute(""" INSERT INTO localization_approvals VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( approval_id, job_id, result["target_locale"], result["target_sha256"], result_json, result_hash, approval_json, _hash_text(approval_json), signed.algorithm, signed.key_id, signed.signature, now, now + ttl, )) else: return self._approved_from_row(existing, plan, authority, now) return self.lookup(plan, job_id, authority, now=now) def _approved_from_row( self, row: sqlite3.Row, plan: Any, authority: ApprovalAuthority, now: float, ) -> ApprovedLocalization: _, job = _plan_job(plan, row["job_id"]) approved, _ = self._verified_row(row, job, authority, now) return approved def _verified_row( self, row: sqlite3.Row, job: dict[str, Any], authority: ApprovalAuthority, now: float, ) -> tuple[ApprovedLocalization, dict[str, Any]]: if _hash_text(row["result_json"]) != row["result_sha256"]: raise LocalizationReleaseBlocked("translation_memory.result_tampered") if _hash_text(row["approval_json"]) != row["approval_sha256"]: raise LocalizationReleaseBlocked("approval.payload.tampered") try: result = json.loads(row["result_json"]) payload = json.loads(row["approval_json"]) except json.JSONDecodeError: raise LocalizationReleaseBlocked("translation_memory.json.invalid") from None result = _validate_result(job, result) expected_keys = { "schema", "release_evidence_contract_sha256", "evidence_request_id", "evidence_revision", "job_id", "source_sha256", "target_sha256", "source_locale", "target_locale", "content_type", "glossary_version", "policy_version", "provider", "software_version", "worker_schema", "result_sha256", "review_confidence", "quality_profile", "commercial_review_routing_contract_sha256", "quality_receipt_sha256", "human_review_receipt_sha256", "independent_model_review", "approval_id", "commercial_review_resolution_contract_sha256", "approved_at", "expires_at", } if set(payload) != expected_keys: raise LocalizationReleaseBlocked("approval.binding_mismatch") binding = { "schema": APPROVAL_SCHEMA, "release_evidence_contract_sha256": ( publication_evidence_contract()["sha256"] ), "evidence_request_id": payload.get("evidence_request_id"), "evidence_revision": payload.get("evidence_revision"), "job_id": row["job_id"], "source_sha256": result["source_sha256"], "target_sha256": result["target_sha256"], "source_locale": result["source_locale"], "target_locale": result["target_locale"], "content_type": result["content_type"], "glossary_version": result["glossary_version"], "policy_version": result["policy_version"], "provider": result["provider"], "software_version": result["software_version"], "worker_schema": result["worker_schema"], "review_confidence": result["review_confidence"], "quality_profile": result["quality_profile"], "commercial_review_routing_contract_sha256": result[ "commercial_review_routing_contract_sha256" ], "result_sha256": row["result_sha256"], "commercial_review_resolution_contract_sha256": ( _commercial_resolution_contract_sha256(job, result) ), "approved_at": row["approved_at"], "expires_at": row["expires_at"], } if any(payload.get(name) != value for name, value in binding.items()): raise LocalizationReleaseBlocked("approval.binding_mismatch") if ( EVIDENCE_REQUEST_ID.fullmatch( str(payload.get("evidence_request_id")) ) is None or not isinstance(payload.get("evidence_revision"), str) or TOKEN.fullmatch(payload["evidence_revision"]) is None ): raise LocalizationReleaseBlocked("approval.binding_mismatch") if payload.get("approval_id") != row["approval_id"]: raise LocalizationReleaseBlocked("approval.binding_mismatch") if HEX64.fullmatch(str(payload.get("quality_receipt_sha256"))) is None: raise LocalizationReleaseBlocked("approval.binding_mismatch") human_hash = payload.get("human_review_receipt_sha256") independent = payload.get("independent_model_review") if not ( result["human_review_required"] or result["independent_review_required"] ) and (human_hash is not None or independent is not None): raise LocalizationReleaseBlocked("approval.binding_mismatch") if result["human_review_required"]: if HEX64.fullmatch(str(human_hash)) is None or independent is not None: raise LocalizationReleaseBlocked("approval.binding_mismatch") elif result["independent_review_required"]: if (human_hash is None) == (independent is None): raise LocalizationReleaseBlocked("approval.binding_mismatch") elif human_hash is not None and HEX64.fullmatch(str(human_hash)) is None: raise LocalizationReleaseBlocked("approval.binding_mismatch") elif independent is not None: if not isinstance(independent, dict) or set(independent) != { "schema", "provider", "receipt_sha256", }: raise LocalizationReleaseBlocked("approval.binding_mismatch") if independent.get("schema") != INDEPENDENT_MODEL_REVIEW_SCHEMA: raise LocalizationReleaseBlocked("approval.binding_mismatch") provider = independent.get("provider") if not isinstance(provider, dict) or set(provider) != {"id", "model_id", "model_version"}: raise LocalizationReleaseBlocked("approval.binding_mismatch") if any(not isinstance(value, str) or not value for value in provider.values()): raise LocalizationReleaseBlocked("approval.binding_mismatch") if provider["id"] == result["provider"]["id"]: raise LocalizationReleaseBlocked("approval.binding_mismatch") if HEX64.fullmatch(str(independent.get("receipt_sha256"))) is None: raise LocalizationReleaseBlocked("approval.binding_mismatch") immutable = { key: value for key, value in payload.items() if key not in {"approval_id", "approved_at", "expires_at"} } expected_id = "blun-l10n-approval-" + _hash_text(_canonical_json(immutable)) if expected_id != row["approval_id"]: raise LocalizationReleaseBlocked("approval.binding_mismatch") if row["target_locale"] != result["target_locale"] or row["target_sha256"] != result["target_sha256"]: raise LocalizationReleaseBlocked("approval.binding_mismatch") signature = _signature(ApprovalSignature( row["signature_algorithm"], row["key_id"], row["signature"], )) verify = getattr(authority, "verify", None) try: valid = callable(verify) and verify(row["approval_json"].encode("utf-8"), signature) is True except Exception: valid = False if not valid: raise LocalizationReleaseBlocked("approval.signature.invalid") if row["expires_at"] <= now: raise LocalizationReleaseBlocked("approval.expired") return ApprovedLocalization( approval_id=row["approval_id"], job_id=row["job_id"], target_locale=row["target_locale"], target_sha256=row["target_sha256"], candidate=result["candidate"], release_evidence=validate_publication_evidence({ "schema": PUBLICATION_EVIDENCE_SCHEMA, "release_evidence_contract_sha256": payload[ "release_evidence_contract_sha256" ], "job_id": row["job_id"], "target_locale": row["target_locale"], "target_sha256": row["target_sha256"], "approval_id": row["approval_id"], "content_type": result["content_type"], "result_sha256": row["result_sha256"], "approval_sha256": row["approval_sha256"], "quality_receipt_sha256": payload["quality_receipt_sha256"], "evidence_request_id": payload["evidence_request_id"], "evidence_revision": payload["evidence_revision"], "quality_profile": { "locale": result["quality_profile"]["locale"], "version": result["quality_profile"]["version"], "sha256": result["quality_profile"]["sha256"], }, "commercial_profile": job.get("commercial_profile"), "commercial_quality_profile": ( json.loads(_canonical_json( result["quality_profile"].get("commercial") )) if result["content_type"] == "commercial" else None ), "commercial_review": json.loads( _canonical_json(result["commercial_review"]) ), "commercial_review_routing_contract_sha256": result[ "commercial_review_routing_contract_sha256" ], "commercial_review_resolution_contract_sha256": result[ "commercial_review_resolution_contract_sha256" ], "commercial_review_resolution": _commercial_review_resolution( result, payload, ), }), approved_at=float(row["approved_at"]), expires_at=float(row["expires_at"]), ), result def verified_result_cache( self, authority: ApprovalAuthority, ) -> VerifiedResultCache: """Bind the exact approval verifier for use by a queue runner.""" return VerifiedResultCache(self, authority) def cached_result( self, job_payload: dict[str, Any], authority: ApprovalAuthority, *, now: float | int, ) -> dict[str, Any] | None: """Return an exact verified result, or ``None`` for a clean cache miss.""" job = _validated_job_payload(job_payload) now = _timestamp(now, "approval.time.invalid") row = self.connection.execute(""" SELECT * FROM localization_approvals WHERE job_id = ? ORDER BY approved_at DESC, approval_id DESC LIMIT 1 """, (job["job_id"],)).fetchone() if row is None: return None try: _, result = self._verified_row(row, job, authority, now) except LocalizationReleaseBlocked as error: if error.code == "approval.expired": return None raise return json.loads(_canonical_json(result)) def lookup( self, plan: Any, job_id: str, authority: ApprovalAuthority, *, now: float | int, ) -> ApprovedLocalization: job_id = _text(job_id, "job.id.invalid") _plan_job(plan, job_id) now = _timestamp(now, "approval.time.invalid") row = self.connection.execute(""" SELECT * FROM localization_approvals WHERE job_id = ? ORDER BY approved_at DESC, approval_id DESC LIMIT 1 """, (job_id,)).fetchone() if row is None: raise LocalizationReleaseBlocked("approval.missing") return self._approved_from_row(row, plan, authority, now) def readiness( self, plan: Any, authority: ApprovalAuthority, *, now: float | int, ) -> WebsiteReadiness: plan_id = _text(getattr(plan, "plan_id", None), "plan.invalid") jobs = getattr(plan, "jobs", None) if not isinstance(jobs, tuple) or not jobs: raise LocalizationReleaseBlocked("plan.invalid") now = _timestamp(now, "approval.time.invalid") required = tuple(sorted(job.as_payload()["target"]["locale"] for job in jobs)) approved: list[str] = [] blocked: list[tuple[str, str]] = [] for job in jobs: locale = job.as_payload()["target"]["locale"] try: self.lookup(plan, job.job_id, authority, now=now) except LocalizationReleaseBlocked as error: blocked.append((locale, error.code)) else: approved.append(locale) return WebsiteReadiness( plan_id=plan_id, ready=not blocked and tuple(sorted(approved)) == required, required_locales=required, approved_locales=tuple(sorted(approved)), blocked=tuple(sorted(blocked)), ) def publication_bundle( self, plan: Any, authority: ApprovalAuthority, *, now: float | int, ) -> tuple[ApprovedLocalization, ...]: status = self.readiness(plan, authority, now=now) if not status.ready: raise LocalizationReleaseBlocked("website.not_ready") return tuple(sorted( ( self.lookup(plan, job.job_id, authority, now=now) for job in plan.jobs ), key=lambda item: item.target_locale, ))