From 228dfdaca0947dd4c240ada12b7beaccb68abd3b Mon Sep 17 00:00:00 2001 From: Codex Agent Date: Tue, 20 Oct 2026 01:59:32 +0300 Subject: [PATCH] Harden MinJust staging and legacy corpus verification --- backend/README.md | 5 + backend/ingestion/minjust_current.py | 174 ++++++++++++++++++++++----- backend/test_minjust_current.py | 80 +++++++++++- docs/operations/CURRENT_STATE.md | 6 +- infra/deploy/test_verify_corpus.py | 15 ++- infra/deploy/verify_corpus.py | 3 +- 6 files changed, 243 insertions(+), 40 deletions(-) diff --git a/backend/README.md b/backend/README.md index 32303b7..9144e7f 100644 --- a/backend/README.md +++ b/backend/README.md @@ -73,6 +73,11 @@ PYTHONPATH=backend python3 -m ingestion.minjust_current \ `data/minjust-cbd` и `data/minjust-normalized`. Адаптер сохраняет исходные ответы и checkpoints, публикует папку документа атомарно и не заменяет существующий код в staging. +Каталог staging должен быть отдельной директорией режима 0700, принадлежащей +текущему пользователю; его родитель должен быть существующим доверенным +каталогом. Символические ссылки в generation, checkpoints и каталогах +документов отклоняются. Legacy-корни вычисляются от корня репозитория, а не +от текущего рабочего каталога. Для `/GetEdition?lang=kg` официальный ответ помещает кыргызский HTML в `contentRu` при `contentKg=null`; архивный язык определяется запрошенным `lang`, а не именем поля ответа. diff --git a/backend/ingestion/minjust_current.py b/backend/ingestion/minjust_current.py index cb53966..4f9944d 100644 --- a/backend/ingestion/minjust_current.py +++ b/backend/ingestion/minjust_current.py @@ -9,6 +9,7 @@ import json import logging import os import sqlite3 +import stat import time import urllib.error import urllib.parse @@ -35,6 +36,7 @@ MAX_RESPONSE_BYTES = 16 * 1024 * 1024 DEFAULT_DOCUMENT_BUDGET = 4 * 1024 * 1024 * 1024 LANGUAGES = (("ru", "ru", "textRusType"), ("kg", "ky", "textKyrType")) LOGGER = logging.getLogger(__name__) +REPOSITORY_ROOT = Path(__file__).resolve().parents[2] class SourceError(RuntimeError): @@ -217,26 +219,118 @@ def open_manifest(path: Path) -> sqlite3.Connection: return connection -def validate_stage_root(output: Path, resume: bool) -> None: - destination = output.resolve() +def reject_symlink_components(path: Path) -> None: + current = Path(path.anchor) + for part in path.parts[1:]: + current /= part + try: + mode = current.lstat().st_mode + except FileNotFoundError: + continue + if stat.S_ISLNK(mode): + raise ValueError(f"staging path must not contain a symbolic link: {current}") + if current != path and not stat.S_ISDIR(mode): + raise ValueError(f"staging parent must be a directory: {current}") + + +def ensure_stage_path(root: Path, path: Path) -> Path: + root = root.absolute() + path = path.absolute() + try: + relative = path.relative_to(root) + except ValueError as error: + raise ValueError("staging file escaped its private generation") from error + current = root + if current.is_symlink(): + raise ValueError("staging root must not be a symbolic link") + for part in relative.parts: + current /= part + try: + mode = current.lstat().st_mode + except FileNotFoundError: + continue + if stat.S_ISLNK(mode): + raise ValueError(f"staging path must not contain a symbolic link: {relative}") + if current != path and not stat.S_ISDIR(mode): + raise ValueError(f"staging parent must be a directory: {relative}") + return path + + +def stage_write(root: Path, path: Path, content: bytes) -> None: + path = ensure_stage_path(root, path) + atomic_write(path, content) + ensure_stage_path(root, path) + + +def stage_mkdir(root: Path, path: Path) -> None: + path = ensure_stage_path(root, path) + path.mkdir(parents=True, exist_ok=True) + ensure_stage_path(root, path) + if not path.is_dir(): + raise ValueError("staging path must be a directory") + + +def validate_stage_tree(output: Path) -> None: + for relative in ( + Path("current-api.sqlite3"), Path("current-api.sqlite3-journal"), + Path("current-api.sqlite3-wal"), Path("current-api.sqlite3-shm"), + Path("manifest.sqlite3"), Path("manifest.sqlite3-journal"), + Path("manifest.sqlite3-wal"), Path("manifest.sqlite3-shm"), + Path("request-contract.json"), + ): + path = ensure_stage_path(output, output / relative) + if path.exists() and not path.is_file(): + raise ValueError(f"staging checkpoint must be a regular file: {relative}") + for relative in (Path(".work"), Path("documents")): + root = ensure_stage_path(output, output / relative) + if root.exists() and not root.is_dir(): + raise ValueError(f"staging path must be a directory: {relative}") + if root.is_dir(): + for directory, directories, files in os.walk(root, followlinks=False): + for name in directories + files: + ensure_stage_path(output, Path(directory) / name) + + +def validate_stage_root(output: Path, resume: bool) -> Path: + requested = Path(output) + if ".." in requested.parts: + raise ValueError("staging output path must not contain '..'") + destination = Path(os.path.abspath(requested)) + reject_symlink_components(destination) + parent = destination.parent + if not parent.is_dir(): + raise ValueError("staging output parent must already exist") + for ancestor in (parent, *parent.parents): + info = ancestor.lstat() + mode = stat.S_IMODE(info.st_mode) + if stat.S_ISLNK(info.st_mode) or not stat.S_ISDIR(info.st_mode): + raise ValueError("staging output ancestors must be real directories") + sticky_parent = bool(mode & stat.S_ISVTX) + if mode & 0o022 and not sticky_parent: + raise ValueError("staging output parent is not trusted") for relative in (Path("data/minjust-cbd"), Path("data/minjust-normalized")): - legacy = relative.resolve() + legacy = (REPOSITORY_ROOT / relative).resolve() if destination == legacy or destination.is_relative_to(legacy) or legacy.is_relative_to(destination): raise ValueError("current API staging must not overlap the existing corpus") - if output.exists(): - if output.is_symlink() or not output.is_dir(): + if destination.exists(): + if destination.is_symlink() or not destination.is_dir(): raise ValueError("staging output must be a real directory") - if any(output.iterdir()) and not resume: - raise ValueError("staging output is not empty; use --resume for this generation") else: - output.mkdir(mode=0o700, parents=True) - for relative in (Path(".work"), Path(".work/documents"), Path("documents")): - path = output / relative - if path.is_symlink() or (path.exists() and not path.is_dir()): - raise ValueError(f"staging path must be a real directory: {relative}") - contract_path = output / "request-contract.json" - if contract_path.is_symlink(): - raise ValueError("staging request contract must not be a symbolic link") + destination.mkdir(mode=0o700) + root_info = destination.lstat() + root_mode = stat.S_IMODE(root_info.st_mode) + if ( + stat.S_ISLNK(root_info.st_mode) + or not stat.S_ISDIR(root_info.st_mode) + or root_info.st_uid != os.geteuid() + or root_mode & 0o077 + or (root_mode & 0o700) != 0o700 + ): + raise ValueError("staging output must be a private directory owned by the current user") + validate_stage_tree(destination) + if any(destination.iterdir()) and not resume: + raise ValueError("staging output is not empty; use --resume for this generation") + contract_path = destination / "request-contract.json" if contract_path.exists(): try: contract = json.loads(contract_path.read_text(encoding="utf-8")) @@ -245,9 +339,10 @@ def validate_stage_root(output: Path, resume: bool) -> None: if contract != REQUEST_CONTRACT: raise SourceChanged("staging generation uses a different request contract") else: - if any(output.iterdir()): + if any(destination.iterdir()): raise SourceChanged("staging generation has no exact-edition request contract; start a new generation") - atomic_write(contract_path, json_bytes(REQUEST_CONTRACT)) + stage_write(destination, contract_path, json_bytes(REQUEST_CONTRACT)) + return destination def stage_document( @@ -263,7 +358,7 @@ def stage_document( raise ValueError("document code is invalid") if max_document_bytes <= 0: raise ValueError("document byte budget must be positive") - validate_stage_root(output, resume) + output = validate_stage_root(output, resume) work_root = output / ".work" / "documents" target_root = output / "documents" connection = open_manifest(output / "current-api.sqlite3") @@ -293,20 +388,28 @@ def stage_document( raise SourceChanged("document metadata changed; start a new staging generation") working_document = work_root / str(canonical) final_document = target_root / str(canonical) + ensure_stage_path(output, working_document) + ensure_stage_path(output, final_document) if final_document.is_symlink(): raise ValueError("staging target must not be a symbolic link") if final_document.exists() and not previous: raise ValueError("staging target already exists without a matching checkpoint") if final_document.exists(): - metadata = json.loads((final_document / "metadata.json").read_text(encoding="utf-8")) + final_metadata = ensure_stage_path(output, final_document / "metadata.json") + source_document = ensure_stage_path(output, final_document / "source-document.json") + metadata = json.loads(final_metadata.read_text(encoding="utf-8")) if ( metadata.get("SourceApi", {}).get("response_sha256") != record_sha - or hashlib.sha256((final_document / "source-document.json").read_bytes()).hexdigest() != record_sha + or hashlib.sha256(source_document.read_bytes()).hexdigest() != record_sha ): raise SourceChanged("published staging document does not match the current source record") for edition in editions: identifier = edition_id(edition) for request_language, archive_language, type_field in LANGUAGES: + html_path = ensure_stage_path( + output, + final_document / "editions" / str(identifier) / f"{archive_language}.html", + ) checkpoint = connection.execute( "SELECT state, content_sha256 FROM languages WHERE canonical_id = ? AND edition_id = ? AND request_language = ?", (canonical, identifier, request_language), @@ -317,8 +420,8 @@ def stage_document( elif ( not checkpoint or checkpoint[0] != "complete" - or not (final_document / "editions" / str(identifier) / f"{archive_language}.html").is_file() - or sha256_file(final_document / "editions" / str(identifier) / f"{archive_language}.html") != checkpoint[1] + or not html_path.is_file() + or sha256_file(html_path) != checkpoint[1] ): raise SourceError(f"published staging document failed verification for edition {identifier} {request_language}") document_sha = hash_tree(final_document) @@ -351,10 +454,10 @@ def stage_document( "state": "text_complete_incomplete_attachments", "resumed": True, } - working_document.mkdir(parents=True, exist_ok=True) + stage_mkdir(output, working_document) document_metadata = map_document_metadata(source, alias, response.sha256) - atomic_write(working_document / "metadata.json", json_bytes(document_metadata)) - atomic_write(working_document / "source-document.json", response.body) + stage_write(output, working_document / "metadata.json", json_bytes(document_metadata)) + stage_write(output, working_document / "source-document.json", response.body) with connection: connection.execute( "INSERT INTO documents (alias, canonical_id, response_sha256, expected_editions, state, updated_at) VALUES (?, ?, ?, ?, 'fetching', ?) " @@ -379,10 +482,10 @@ def stage_document( for order, edition in enumerate(editions): identifier = edition_id(edition) directory = working_document / "editions" / str(identifier) - directory.mkdir(parents=True, exist_ok=True) - atomic_write(directory / "metadata.json", json_bytes(map_edition_metadata(edition, order))) + stage_mkdir(output, directory) + stage_write(output, directory / "metadata.json", json_bytes(map_edition_metadata(edition, order))) for request_language, archive_language, type_field in LANGUAGES: - html_path = directory / f"{archive_language}.html" + html_path = ensure_stage_path(output, directory / f"{archive_language}.html") checkpoint = connection.execute( "SELECT state, content_sha256, byte_count FROM languages WHERE canonical_id = ? AND edition_id = ? AND request_language = ?", (canonical, identifier, request_language), @@ -425,11 +528,12 @@ def stage_document( response_metadata["requested_edition_id"] = identifier response_metadata["exact"] = True response_metadata["response_sha256"] = language_response.sha256 - atomic_write( + stage_write( + output, directory / "source" / f"{request_language}.json", json_bytes(response_metadata), ) - atomic_write(html_path, encoded) + stage_write(output, html_path, encoded) save_language_checkpoint( connection, canonical, identifier, request_language, archive_language, language_response.sha256, hashlib.sha256(encoded).hexdigest(), len(encoded), "complete", None, @@ -452,16 +556,20 @@ def stage_document( ) return {"alias": alias, "canonical_id": canonical, "editions": len(editions), "incomplete": incomplete} - metadata = json.loads((working_document / "metadata.json").read_text(encoding="utf-8")) + working_metadata = ensure_stage_path(output, working_document / "metadata.json") + metadata = json.loads(working_metadata.read_text(encoding="utf-8")) metadata["SourceCompleteness"] = { "text": "complete", "attachments": "not independently verified", } - atomic_write(working_document / "metadata.json", json_bytes(metadata)) - target_root.mkdir(parents=True, exist_ok=True) + stage_write(output, working_metadata, json_bytes(metadata)) + stage_mkdir(output, target_root) + ensure_stage_path(output, working_document) + ensure_stage_path(output, final_document) if final_document.exists(): raise ValueError("refusing to replace a document in this staging generation") os.replace(working_document, final_document) + validate_stage_tree(output) document_sha = hash_tree(final_document) fetched_at = utc_now() with raw_manifest: diff --git a/backend/test_minjust_current.py b/backend/test_minjust_current.py index b6beb33..19f1353 100644 --- a/backend/test_minjust_current.py +++ b/backend/test_minjust_current.py @@ -1,12 +1,14 @@ import json +import os import sqlite3 import tempfile import unittest import urllib.error from email.message import Message from pathlib import Path -from unittest.mock import Mock +from unittest.mock import Mock, patch +from ingestion import minjust_current from ingestion.minjust_current import ( ApiResponse, CurrentCbdClient, @@ -45,6 +47,80 @@ def document_record(alias="6", canonical=6, editions=None): class CurrentCbdClientTests(unittest.TestCase): + def test_rejects_legacy_roots_when_called_from_an_unrelated_working_directory(self): + with tempfile.TemporaryDirectory() as temporary: + base = Path(temporary) + repository = base / "repository" + elsewhere = base / "elsewhere" + elsewhere.mkdir() + legacy_roots = [ + repository / "data/minjust-cbd", + repository / "data/minjust-normalized", + ] + for legacy in legacy_roots: + legacy.mkdir(parents=True) + original_cwd = Path.cwd() + client = Mock() + try: + os.chdir(elsewhere) + with patch.object(minjust_current, "REPOSITORY_ROOT", repository): + for legacy in legacy_roots: + with self.subTest(legacy=legacy), self.assertRaisesRegex(ValueError, "must not overlap"): + stage_document(legacy / "staging", "6", client) + self.assertFalse((legacy / "staging").exists()) + finally: + os.chdir(original_cwd) + client.document.assert_not_called() + + def test_requires_private_owned_staging_root(self): + with tempfile.TemporaryDirectory() as temporary: + output = Path(temporary) / "shared" + output.mkdir(mode=0o755) + output.chmod(0o755) + + with self.assertRaisesRegex(ValueError, "private directory"): + stage_document(output, "6", Mock()) + + def test_resume_rejects_symlinked_checkpoints_and_document_paths(self): + with tempfile.TemporaryDirectory() as temporary: + base = Path(temporary) + targets = [ + ("request-contract.json", False), + ("current-api.sqlite3", False), + ("manifest.sqlite3", False), + (".work/documents/6", True), + (".work/documents/6/editions/920492/ru.html", False), + ("documents/6", True), + ] + client = Mock() + for index, (relative, directory) in enumerate(targets): + with self.subTest(relative=relative): + output = base / f"generation-{index}" + output.mkdir(mode=0o700) + contract = output / "request-contract.json" + if relative != "request-contract.json": + contract.write_text(json.dumps(minjust_current.REQUEST_CONTRACT), encoding="utf-8") + outside = base / f"outside-{index}" + if directory: + outside.mkdir() + else: + outside.write_text("preserve", encoding="utf-8") + link = output / relative + link.parent.mkdir(parents=True, exist_ok=True) + try: + link.symlink_to(outside, target_is_directory=directory) + except (NotImplementedError, OSError): + self.skipTest("symlinks are unavailable") + + with self.assertRaisesRegex(ValueError, "symbolic link"): + stage_document(output, "6", client, resume=True) + + client.document.assert_not_called() + if not directory: + self.assertEqual(outside.read_text(encoding="utf-8"), "preserve") + else: + self.assertEqual(list(outside.iterdir()), []) + def test_respects_retry_after_and_retries_server_errors(self): headers = Message() headers["Retry-After"] = "7" @@ -269,7 +345,7 @@ class CurrentCbdClientTests(unittest.TestCase): def test_does_not_resume_generation_without_exact_request_contract(self): with tempfile.TemporaryDirectory() as temporary: output = Path(temporary) / "legacy-generation" - output.mkdir() + output.mkdir(mode=0o700) sentinel = output / "old-response.json" sentinel.write_text("preserve", encoding="utf-8") client = Mock() diff --git a/docs/operations/CURRENT_STATE.md b/docs/operations/CURRENT_STATE.md index a2d199a..75db2c3 100644 --- a/docs/operations/CURRENT_STATE.md +++ b/docs/operations/CURRENT_STATE.md @@ -193,7 +193,11 @@ protected file metadata remain unverified. - `backend/ingestion/minjust_current.py` stages one document into a separate generation, at no more than one request per second, with a 16 MiB response limit, a 4 GiB text budget per document, contract-bound checkpoints, and - atomic publication. The earlier non-exact pass is quarantined at + atomic publication. Staging requires a private 0700 directory owned by the + current user and rejects symlinks under checkpoints/document paths; legacy + corpus exclusions are anchored to the repository root. The corpus verifier + continues to validate unchanged normalized manifests from versions 1.0.0 + and 1.1.0. The earlier non-exact pass is quarantined at `/tmp/akyldash-current-code6.ZAMC37u4ex/qualification.json`: its 240 language responses (304,493,520 bytes) remain unverified and are not reusable. The retained work-tree SHA-256 is diff --git a/infra/deploy/test_verify_corpus.py b/infra/deploy/test_verify_corpus.py index d6c4af6..deff0c3 100644 --- a/infra/deploy/test_verify_corpus.py +++ b/infra/deploy/test_verify_corpus.py @@ -40,7 +40,7 @@ class CorpusInventoryTests(unittest.TestCase): ) connection.close() - def _nonempty_corpus(self): + def _nonempty_corpus(self, normalizer_version=NORMALIZER_VERSION): document = self.raw / "documents/1" edition = document / "editions/2" edition.mkdir(parents=True) @@ -80,7 +80,7 @@ class CorpusInventoryTests(unittest.TestCase): document_json = { "schema_version": "1", "source_code": "1", - "normalizer": {"version": NORMALIZER_VERSION}, + "normalizer": {"version": normalizer_version}, "source": {"path": "documents/1", "files": source_files, "sha256": source_sha}, "editions": [{"source_code": "2"}], } @@ -100,7 +100,7 @@ class CorpusInventoryTests(unittest.TestCase): "CREATE TABLE documents (code TEXT PRIMARY KEY, source_sha256 TEXT, schema_version TEXT, normalizer_version TEXT, state TEXT)" ) normalized_manifest.execute( - "INSERT INTO documents VALUES ('1', ?, '1', ?, 'success')", (source_sha, NORMALIZER_VERSION) + "INSERT INTO documents VALUES ('1', ?, '1', ?, 'success')", (source_sha, normalizer_version) ) normalized_manifest.commit() normalized_manifest.close() @@ -276,6 +276,15 @@ class CorpusInventoryTests(unittest.TestCase): self.assertEqual(result["manifest_summary"]["raw"]["documents"], 1) self.assertEqual(result["manifest_summary"]["normalized"]["fragments"], 1) + def test_verifies_legacy_normalizer_1_0_0_manifest(self): + normalized, _ = self._nonempty_corpus(normalizer_version="1.0.0") + result = verify_corpus(self.raw, normalized, self.base / "artifacts") + self.assertEqual(result["status"], "complete", result["finding_examples"]) + self.assertNotIn( + "normalizer_version_mismatch", + {item["kind"] for item in result["finding_examples"]}, + ) + def test_raw_normalized_source_mismatch_is_reported(self): normalized, _ = self._nonempty_corpus() manifest = sqlite3.connect(normalized / "manifest.sqlite3") diff --git a/infra/deploy/verify_corpus.py b/infra/deploy/verify_corpus.py index 7855128..0018b4b 100644 --- a/infra/deploy/verify_corpus.py +++ b/infra/deploy/verify_corpus.py @@ -27,6 +27,7 @@ NORMALIZER_ROOT = Path(__file__).resolve().parents[2] TOOL_SHA256 = hashlib.sha256(Path(__file__).read_bytes()).hexdigest() sys.path.insert(0, str(NORMALIZER_ROOT)) from backend.normalization.minjust_cbd import NORMALIZER_VERSION, SCHEMA_VERSION as NORMALIZED_SCHEMA_VERSION # noqa: E402 +SUPPORTED_NORMALIZER_VERSIONS = frozenset({"1.0.0", NORMALIZER_VERSION}) class InventoryError(RuntimeError): @@ -552,7 +553,7 @@ def load_manifest( finding(connection, run_id, root_id, "normalized_manifest_state", f"documents/{code}", str(state)) if root_id == "normalized" and schema_version != NORMALIZED_SCHEMA_VERSION: finding(connection, run_id, root_id, "schema_version_mismatch", f"documents/{code}", str(schema_version)) - if root_id == "normalized" and normalizer_version != NORMALIZER_VERSION: + if root_id == "normalized" and normalizer_version not in SUPPORTED_NORMALIZER_VERSIONS: finding(connection, run_id, root_id, "normalizer_version_mismatch", f"documents/{code}", str(normalizer_version)) if root_id == "raw": connection.execute("DELETE FROM manifest_errors WHERE root_id='raw'")