Harden MinJust staging and legacy corpus verification

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`, а не именем поля ответа.

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:

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()

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

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")

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'")