From 4b2e5578cd050657f2fd0de6817c697693183117 Mon Sep 17 00:00:00 2001 From: agent Date: Sun, 18 Oct 2026 15:29:14 +0300 Subject: [PATCH 1/5] Add resumable corpus inventory verification --- docs/operations/CORPUS_HANDOFF.md | 91 ++ docs/operations/CURRENT_STATE.md | 106 +- docs/operations/DEPLOYMENT.md | 25 +- ...njust-document-normalization-agent-task.md | 9 +- infra/deploy/test_verify_corpus.py | 164 +++ infra/deploy/verify_corpus.py | 1026 +++++++++++++++++ 6 files changed, 1372 insertions(+), 49 deletions(-) create mode 100644 docs/operations/CORPUS_HANDOFF.md create mode 100644 infra/deploy/test_verify_corpus.py create mode 100644 infra/deploy/verify_corpus.py diff --git a/docs/operations/CORPUS_HANDOFF.md b/docs/operations/CORPUS_HANDOFF.md new file mode 100644 index 0000000..47cdba9 --- /dev/null +++ b/docs/operations/CORPUS_HANDOFF.md @@ -0,0 +1,91 @@ +# Corpus inventory and production handoff + +The repository contains a read-only verifier for the ignored local corpus at +`data/minjust-cbd/` and `data/minjust-normalized/`. It does not download, +normalize, copy, index, or modify source files. Keep inventory artifacts outside +the repository and both source trees. + +## Run the inventory + +From the repository root, create a private artifact directory and run: + +```bash +artifact_dir=$(mktemp -d /tmp/akyldash-corpus-audit.XXXXXXXX) +python3 infra/deploy/verify_corpus.py \ + --raw-root data/minjust-cbd \ + --normalized-root data/minjust-normalized \ + --output-dir "$artifact_dir" +``` + +The tool writes `result.json` and `inventory.sqlite3` into that directory. Keep +the directory private: the inventory contains source paths and checksums. Exit +status `0` means the scan and structural checks completed without errors; `1` +means findings were recorded or the run is incomplete; `2` means the arguments +or artifact path were invalid. Read `result.json` to distinguish those states. +Warnings, including documents with zero text fragments, are reported +separately and do not turn the scan into a failure. + +For an interrupted run, use the same artifact directory and add `--resume`. +Resume rehashes every source file instead of trusting cached timestamps, then +reports added, changed, and removed paths relative to entries already stored +in the inventory. After resuming an interrupted first pass, those counters are +relative to the partial saved inventory; the full rehash and final checks still +cover every path. A successful result therefore describes the latest complete +pass. Run against a quiescent corpus; do not +normalize, fetch, or copy files into either source tree during the scan. Do not +point `--output-dir` inside either +source tree or through a symlink. The verifier rejects such paths before +creating or changing them; it also flags symlinks, reparse points, hard links, +unsupported file types, and files that change while being read. + +The audit checks SQLite integrity, manifest/document directory agreement, +normalized schema and normalizer versions, source inventory checksums, +fragment identities and text hashes, required output files, and text assembled +from fragments. It does not establish the legal authority, completeness, or +semantic correctness of MinJust content. + +## Current data disposition + +As measured on 2026-10-18, both local manifests contain 209,958 documents and +pass SQLite `quick_check`. All normalized manifest entries are `success` at +schema `1` and normalizer `1.0.0`. The raw manifest records one fetch error for +code `6`, which is absent from its `documents` table; this does not prove that +the source document is absent. On 2026-10-18, bounded calls through the +repository's rate-limited MinJust client received HTTP `404` for +`GetDocumentListByQuery`, `GetDocument(6)`, control `GetDocument(1096)`, and +`CheckAvailable`. The listing call failed before returning a code list, and +both document calls failed identically. The repeated `404` is consistent with +the OpenData endpoint being unavailable or its route having changed, but does +not identify which. It is not evidence that code `6` is absent. The [public +document listing page](https://cbd.minjust.gov.kg/list-docs/kg) was reachable, +but it does not confirm this numeric API code. Do not describe the corpus as +fully verified or silently waive this error. Recheck the official listing and +API when the endpoint responds, then record an explicit owner disposition. +From the VPS, a metadata-only HTTPS check to `CheckAvailable` failed at the +HTTP/2 transport layer (`curl` exit 92, `INTERNAL_ERROR`, HTTP status `000`, +TLS verification `0`). An HTTP/1.1 retry was not completed. + +The full local file inventory and normalized cross-check are still in progress +until a `result.json` reports `status: complete` or `status: failed` and +`scan_status: complete`. Do not treat progress output as a completed audit. + +## Production transfer and indexing gates + +Production deployment is not enabled. Before any transfer, the owner must +review the root-only host audit, confirm the target data location is empty or +backed up, resolve the raw code `6` finding, and approve the exact normalized +corpus to use. Transfer only `minjust-normalized/`; keep the raw source archive +offline. Do not preserve the source's Windows/WSL file permission bits. Copy +to a new staging location with owner-approved permissions, verify the copied +manifest and files there, and only then consider placing it under `DATA_ROOT`. +Never overwrite an existing corpus or OpenSearch volume. + +The current bulk importer checks individual Bulk API item results, but it can +switch an alias after a zero-fragment input and does not enforce a refresh, +count comparison, or representative-query check before that switch. Until a +separate change adds those gates, do not pass `--alias` during production +loading. Load into a new versioned index only after the corpus transfer is +verified; require a nonzero fragment count, successful item results, an +explicit refresh, a matching OpenSearch document count, and representative +Russian/Kyrgyz queries before an owner performs an alias change. Take and +verify the required backup before any production index or service change. diff --git a/docs/operations/CURRENT_STATE.md b/docs/operations/CURRENT_STATE.md index a15d875..2151960 100644 --- a/docs/operations/CURRENT_STATE.md +++ b/docs/operations/CURRENT_STATE.md @@ -1,25 +1,26 @@ # Infrastructure current state -Last host audit: 2026-10-12. Gitea Actions status checked 2026-10-18. Source: read-only SSH session as `codex`, owner-reported -UFW state, local repository inspection, and Gitea API. Root-only Docker state, -UFW rule details, and protected file metadata remain unverified. +Read-only host recheck: 2026-10-18. Source: SSH as `codex`, local repository +inspection, and Gitea API. Root-only Docker state, UFW rule details, and +protected file metadata remain unverified. ## VPS facts - Host: `avprogs.hlab.kz`, `206.223.249.118`, SSH port `22`, user `codex`. - OS: Ubuntu Server `24.04.5 LTS`; kernel `6.8.0-142-generic` from the prior inventory audit. -- Resources: 4 vCPU, 7.7 GiB RAM, 3.8 GiB swap, 99 GiB disk with 92 GiB free - at this audit. +- Resources: 4 vCPU, 7.7 GiB RAM, 3.8 GiB swap, 99 GiB root filesystem with + 84 GiB free on 2026-10-18. Runner storage is a separate 15 GiB ext4 loop + mount with 12 GiB free at that check. - Network: interface `ens3`, IPv4 `206.223.249.118/32`. - `codex` is UID/GID 1000, groups `codex` and `users`; `sudo -n true` requires a password. Docker CLI cannot access `/var/run/docker.sock` as `codex`. - SSH service is running. The inspected sshd configuration has public-key and password authentication enabled and `PermitRootLogin yes`; no SSH changes were made. -- Bootstrap completed. Docker CE/CLI `29.9.0`, Compose `5.6.0`, containerd - `2.4.1`, and runc `1.5.2` are installed; package versions and active Docker - service were independently checked. The Docker daemon API remains root-only. +- Bootstrap completed. Docker CLI `29.9.0` and Compose `5.6.0` were rechecked + over SSH on 2026-10-18; containerd `2.4.1` and runc `1.5.2` are from the + prior package audit. The Docker daemon API remains root-only. - `/etc/docker/daemon.json` has the expected `local` log driver and `10m` / `3` rotation limits. Bootstrap paths exist with root ownership: `/opt/akyldash` mode `0755`, `/etc/akyldash` and `/var/backups/akyldash` mode `0700`, and @@ -27,7 +28,10 @@ UFW rule details, and protected file metadata remain unverified. - `/opt/akyldash/current`, verifier/rollback binaries, and the sudoers file are absent. `codex` has not received sudo or Docker-group access. Metadata for `/etc/akyldash/production.env` and `/etc/akyldash/review_secret` could not be - checked because `/etc/akyldash` is root-only. + checked because access is denied. The target + `/var/lib/akyldash/data/minjust-normalized` and its manifest metadata were + also denied; their presence, ownership, and contents are unknown. No sudo was + used. - TCP listeners observed: SSH `22` and loopback DNS only; no listener on `1500`, `80`, or `443`. No ISPmanager package or service unit was found through the unprivileged checks; root can confirm with the audit script below. @@ -39,8 +43,10 @@ UFW rule details, and protected file metadata remain unverified. `https://git.vakanaut.ru/api/v1/version` returned Gitea `1.25.5`; runner registration and Gitea Actions protocol connectivity are not confirmed. - `akyldash.kg` resolves to Cloudflare IPv4/IPv6 edge addresses. From the VPS, - HTTP, HTTPS, and `/review` still return Cloudflare `521`; origin TLS and - Cloudflare Full (strict) are not verified. + HTTPS returned Cloudflare `521` with TLS verification result `0` on + 2026-10-18. TCP listeners were SSH `22` and local DNS only, so no origin + listener on `80`/`443` was observed. Origin TLS and Cloudflare Full (strict) + are not verified. ## Application and Gitea facts @@ -49,13 +55,12 @@ UFW rule details, and protected file metadata remain unverified. PRs enabled. `main` is protected and requires one approval; the current API principal cannot push or merge directly to `main`. - The repository runner-list API returns `403` because this API principal is - not the repository owner. Gitea status checked 2026-10-18 showed smoke run 62 - successful and CI run 61 pending on PR head `69c62466`. That workflow used - `ubuntu-latest`, while the smoke workflow used `akyldash-ci`; this label - difference is a plausible explanation, not a confirmed scheduler diagnosis. - The owner reports that the runner is registered and Idle and rootless Docker - works (report date not recorded). Runner inventory and package publishing - permissions remain unverified. + not the repository owner. The latest verified CI run 75 and smoke run 76 both + succeeded on `main` head `0e276ae6`. The Actions run-list endpoint returned + `403` for this API principal during this recheck, so no newer run state was + independently retrieved. The owner reports the runner is registered and + Idle; repository runner inventory and package publishing permissions remain + unverified. - No public application listener is observed on TCP `80`/`443`. Docker container/volume state still requires the root audit. Repository Compose describes Caddy `2.11.6` pinned by digest, Python Search API `0.11.3`, SQLite @@ -89,11 +94,15 @@ UFW rule details, and protected file metadata remain unverified. ext4 loop-backed image/workspace storage, capacity 1, per-job limits (2 CPU, 3 GiB RAM, 512 processes), account slice limits (2.5 CPU, 4 GiB RAM, 2048 tasks), rotated file logs, and a systemd health timer. Jobs target the - `akyldash-ci` label and do not receive either Docker socket. The owner reports - that runner registration and rootless Docker work (report date not recorded). - Gitea status checked 2026-10-18 showed smoke run 62 successful and CI run 61 - pending on PR head `69c62466`. The workflow/label difference may explain that - pending status, but the scheduler reason was not independently verified. On + `akyldash-ci` label and do not receive either Docker socket. On 2026-10-18, + SSH showed `akyldash-act-runner.service` active under `akyldash-runner`, with + zero restarts and about 115 MiB current memory. Its rootless Docker store is + mounted from `/dev/loop0` as ext4 with `rw,nosuid,nodev`; the 15 GiB mount had + 12 GiB free. The separate `akyldash-runner-restart.service` remains failed + with exit status 1 since 2026-10-14; its journal had no entries, so the cause + is unknown. The latest verified main CI run 75 and smoke run 76 succeeded on + head `0e276ae6`. The Actions API returned 403 during this recheck, so no newer + run state was independently retrieved. On head `3a89dc21`, CI run 63 failed at the host audit root guard and smoke run 64 succeeded. On head `436806a8`, CI run 65 failed in the Caddy container action with `exec: "validate": executable file not found`; Kaniko was skipped and @@ -113,6 +122,37 @@ UFW rule details, and protected file metadata remain unverified. backend `0.11.3`, frontend `0.1.2`, Telegram bot `0.2.2` (the bot is not deployed on this VPS). +## Corpus readiness (2026-10-18) + +- The local raw archive and normalized corpus are ignored under `data/`; no + corpus files were changed. Their read-only SQLite manifests both pass + `quick_check` and each report 209,958 documents. The raw manifest is + 28,295,168 bytes with SHA-256 + `f8e3dd2fdec607583ca09cfdb5b76171bb7bfa1e41a0e07ee5aca8481e18e17e`; it has + one fetch error, code `6`, absent from `documents`. The normalized manifest + is 31,571,968 bytes with SHA-256 + `1679c20e6d0f3e5a9b50737c006c4417a28ebcaf7467d194d3bf8243538f83bf`; all + entries are `success`, schema `1`, normalizer `1.0.0`. +- A streaming full-file inventory and cross-check is running read-only with + artifacts only in `/tmp/akyldash-corpus-audit.Mjr6aa`. Latest progress was + run `b4a4556e30cc464d90bc8d75ef1673e9`; live counters are in its private + `result.json`, updated every 1,000 entries. The normalized tree has not yet + started. Progress is not a completed audit; record the final result and full + manifest comparison when the run finishes. +- Through the repository client, bounded MinJust OpenData requests for + `GetDocumentListByQuery`, `GetDocument(6)`, control `GetDocument(1096)`, and + `CheckAvailable` each returned HTTP `404`. The list API returned no code list; + the direct requests do not establish whether code `6` exists. The public + listing page is reachable, but it does not map the numeric API code. A + metadata-only `CheckAvailable` request from the VPS failed in HTTP/2 with + `curl` exit 92 (`INTERNAL_ERROR`), HTTP status `000`, and TLS verification 0; + the HTTP/1.1 retry was not completed. Treat the code `6` disposition and + full corpus readiness as unresolved. +- On the VPS, read-only `stat` of `/var/lib/akyldash/data/minjust-normalized` + and its manifest was denied. Production corpus presence and metadata remain + unknown. No corpus was transferred and no production service, Docker socket, + runner policy, or VPS configuration was changed. + ## Checks not completed - Docker Engine version from the server API, container and volume inventory, @@ -149,17 +189,11 @@ UFW rule details, and protected file metadata remain unverified. 5. Test backup/restore and rollback only with synthetic data in an isolated environment before any production deployment. -## CI workflow change in this PR +## Change in this PR -The branch targets `akyldash-ci` and uses digest-pinned isolated actions for -Caddy validation and the two production Dockerfile builds. Kaniko image -publishing and cache publishing are disabled. This repository change does not -modify the VPS or grant workflow containers a Docker socket. CI run 65 failed -in the Caddy action because the runner tried to execute `validate` directly; -Kaniko was skipped. CI run 67 then found that the action's fixed -`/github/workspace` path did not contain the repository; Kaniko was skipped. -The branch now passes Gitea's `${{ github.workspace }}` value into all three -container action invocations. CI run 69 validated Caddy and completed both -image builds; smoke run 70 passed. Actual per-action resource-limit enforcement -and peak usage were not independently measured or verified. Host-wide runner -metrics, image publishing, and production deployment remain unverified. +Adds a read-only, resumable full-file corpus inventory and normalized-data +cross-check, synthetic regression tests, and an operator handoff runbook. CI's +existing `infra/deploy` unittest discovery includes the new tests. No workflow, +runner, VPS, production data, or Docker settings are changed. The real corpus +scan is still in progress and the source API's code `6` outcome remains +unconfirmed. diff --git a/docs/operations/DEPLOYMENT.md b/docs/operations/DEPLOYMENT.md index 167947c..a59882c 100644 --- a/docs/operations/DEPLOYMENT.md +++ b/docs/operations/DEPLOYMENT.md @@ -34,8 +34,9 @@ backup/restore have been tested. It creates `/etc/akyldash/production.env` from the tracked example and generates `/etc/akyldash/review_secret`; both are root-owned mode `0600`. It refuses to overwrite either file and does not start services or create - volumes. Review `DATA_ROOT` and place the approved normalized corpus there - before deployment. + volumes. Review `DATA_ROOT` and follow the + [corpus inventory and handoff gates](CORPUS_HANDOFF.md) before placing the + approved normalized corpus there. 3. After reviewing the root-owned checkout and audit output, run `sudo bash /opt/akyldash/current/infra/deploy/install-helper.sh`. It installs the backup reader, read-only verifier, `akyldash-compose config|status`, and @@ -65,13 +66,13 @@ backup/restore have been tested. sudo chmod 0600 /opt/akyldash/deployed-image.txt ``` -Application deployment remains owner-run. A repository-level Actions runner -installation is prepared in [CI_RUNNER.md](CI_RUNNER.md), but it is not yet -installed or registered; its Online state and smoke workflow are unverified. -The runner's job containers have no Docker socket or production secrets. CI has -no image publishing or production deployment step. The installed sudo helpers -support fixed verification, read-only Compose `config`/`status`, and guarded -rollback. They do not start or stop services. +Application deployment remains owner-run. The repository-scoped `act_runner` +`0.2.13` service is active under `akyldash-runner` and uses rootless Docker on a +separate loop-backed filesystem. CI run 75 and smoke run 76 succeeded on main +head `0e276ae6`. Runner job containers have neither Docker socket nor production +secrets. CI has no image publishing or production deployment step. The +installed sudo helpers support fixed verification, read-only Compose +`config`/`status`, and guarded rollback. They do not start or stop services. ## CI @@ -79,5 +80,7 @@ rollback. They do not start or stop services. production images on pull requests and pushes to `main`. It does not publish an image or deploy production. `.gitea/workflows/runner-smoke.yaml` exercises the `akyldash-ci` runner on pull requests, pushes to `main`, or manual dispatch. -Runner registration is incomplete, so neither workflow execution on that -runner nor current `ubuntu-latest` runner availability has been confirmed. +Both workflows passed on main head `0e276ae6` in runs 75 and 76 respectively. +The repository runner-list endpoint is not available to the current API +principal, so its registration details cannot be independently read through +Gitea API. diff --git a/docs/product/minjust-document-normalization-agent-task.md b/docs/product/minjust-document-normalization-agent-task.md index 1491bd4..baf5006 100644 --- a/docs/product/minjust-document-normalization-agent-task.md +++ b/docs/product/minjust-document-normalization-agent-task.md @@ -11,8 +11,13 @@ - загрузчик находится в `backend/ingestion/minjust_cbd.py`; - сырой архив хранится в `data/minjust-cbd` и исключён из Git; -- в манифесте 209 811 успешно загруженных документов; -- 13 кодов остаются в таблице `errors` и отсутствуют в таблице `documents`; +- в read-only проверенном манифесте 209 958 документов; +- один код, `6`, остаётся в таблице `errors` и отсутствует в `documents`; + это не подтверждает отсутствие документа в источнике: при последней + проверке OpenData API возвращал HTTP `404` для списка, `GetDocument(6)`, + контрольного `GetDocument(1096)` и `CheckAvailable`; +- в нормализованном манифесте 209 958 успешных записей, schema `1`, + normalizer `1.0.0`; полная потоковая проверка всех файлов ещё выполняется. - документ содержит `metadata.json` и каталог `editions`; - редакция содержит `metadata.json`, `ru.html` и/или `ky.html`, иногда изображения; diff --git a/infra/deploy/test_verify_corpus.py b/infra/deploy/test_verify_corpus.py new file mode 100644 index 0000000..3970dbd --- /dev/null +++ b/infra/deploy/test_verify_corpus.py @@ -0,0 +1,164 @@ +import hashlib +import json +import os +import sqlite3 +import tempfile +import unittest +from pathlib import Path + +from verify_corpus import ( + InventoryError, + database, + safe_relative_path, + validate_fragment_file, + verify_corpus, +) + + +class CorpusInventoryTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.base = Path(self.temp.name) + self.raw = self.base / "raw" + self.raw.mkdir() + self._manifest() + + def tearDown(self): + self.temp.cleanup() + + def _manifest(self): + connection = sqlite3.connect(self.raw / "manifest.sqlite3") + connection.executescript( + "CREATE TABLE documents (code INTEGER PRIMARY KEY, source_sha256 TEXT, fetched_at TEXT);" + "CREATE TABLE errors (code INTEGER PRIMARY KEY, message TEXT, failed_at TEXT);" + ) + connection.close() + + def test_output_inside_source_is_rejected_without_creating_it(self): + output = self.raw / "artifacts" + before_mode = self.raw.stat().st_mode + with self.assertRaisesRegex(InventoryError, "separate"): + verify_corpus(self.raw, None, output) + self.assertFalse(output.exists()) + self.assertEqual(self.raw.stat().st_mode, before_mode) + self.assertEqual({path.name for path in self.raw.iterdir()}, {"manifest.sqlite3"}) + + def test_symlink_output_is_rejected_without_changing_target(self): + target = self.base / "target" + target.mkdir(mode=0o755) + link = self.base / "output-link" + try: + link.symlink_to(target, target_is_directory=True) + except (NotImplementedError, OSError): + self.skipTest("directory symlinks are unavailable") + before_mode = target.stat().st_mode + with self.assertRaisesRegex(InventoryError, "link or file"): + verify_corpus(self.raw, None, link) + self.assertEqual(target.stat().st_mode, before_mode) + self.assertEqual(list(target.iterdir()), []) + + def test_public_existing_output_is_rejected_without_chmod(self): + output = self.base / "public-output" + output.mkdir(mode=0o755) + output.chmod(0o755) + before_mode = output.stat().st_mode + with self.assertRaisesRegex(InventoryError, "must be private"): + verify_corpus(self.raw, None, output) + self.assertEqual(output.stat().st_mode, before_mode) + self.assertEqual(list(output.iterdir()), []) + + def test_resume_rejects_symlink_inventory_without_touching_target(self): + output = self.base / "artifacts" + output.mkdir(mode=0o700) + target = self.base / "do-not-touch.sqlite3" + target.write_bytes(b"sentinel") + link = output / "inventory.sqlite3" + try: + link.symlink_to(target) + except (NotImplementedError, OSError): + self.skipTest("file symlinks are unavailable") + with self.assertRaisesRegex(InventoryError, "private regular file"): + verify_corpus(self.raw, None, output, resume=True) + self.assertEqual(target.read_bytes(), b"sentinel") + + def test_rejects_unsafe_cross_platform_relative_paths(self): + for value in ("../outside", "C:\\outside", "folder\\file", "//server/share"): + with self.subTest(value=value), self.assertRaises(InventoryError): + safe_relative_path(value) + + def test_interrupted_inventory_resumes_and_detects_changed_input(self): + payload = self.raw / "payload.txt" + payload.write_text("before", encoding="utf-8") + output = self.base / "artifacts" + stopped = verify_corpus(self.raw, None, output, stop_after_entries=1) + self.assertEqual(stopped["status"], "incomplete") + + resumed = verify_corpus(self.raw, None, output, resume=True) + self.assertEqual(resumed["status"], "complete") + self.assertEqual(resumed["input_changes"]["raw"]["added"], 1) + + payload.write_text("after", encoding="utf-8") + changed = verify_corpus(self.raw, None, output, resume=True) + self.assertEqual(changed["status"], "complete") + self.assertEqual(changed["input_changes"]["raw"]["changed"], 1) + rows = sqlite3.connect(output / "inventory.sqlite3").execute( + "SELECT sha256 FROM files WHERE relative_path='payload.txt'" + ).fetchone() + self.assertEqual(rows[0], hashlib.sha256(b"after").hexdigest()) + + def test_manifest_error_missing_document_and_extra_directory_are_reported(self): + connection = sqlite3.connect(self.raw / "manifest.sqlite3") + connection.execute("INSERT INTO documents VALUES (1, ?, 'now')", ("a" * 64,)) + connection.execute("INSERT INTO errors VALUES (6, 'fetch failed', 'now')") + connection.commit() + connection.close() + (self.raw / "documents" / "999").mkdir(parents=True) + + result = verify_corpus(self.raw, None, self.base / "artifacts") + kinds = {finding["kind"] for finding in result["finding_examples"]} + self.assertEqual(result["status"], "failed") + self.assertIn("raw_manifest_error", kinds) + self.assertIn("missing_raw_document_directory", kinds) + self.assertIn("extra_raw_document_directory", kinds) + + def test_fragment_text_hash_mismatch_is_reported(self): + normalized = self.base / "normalized" + fragment_rel = "documents/1/editions/2/ru/fragments.json" + fragment_path = normalized / fragment_rel + fragment_path.parent.mkdir(parents=True) + fragment = { + "id": "document:1:edition:2:lang:ru:fragment:1", + "document_code": "1", + "edition_code": "2", + "language": "ru", + "position": 1, + "text": "text", + "text_sha256": "0" * 64, + "source_path": "documents/1/source.txt", + "source_sha256": "1" * 64, + } + fragment_path.write_text(json.dumps([fragment]), encoding="utf-8") + content_path = fragment_path.with_name("content.txt") + content_path.write_text("text\n", encoding="utf-8") + connection = database(self.base / "test.sqlite3") + content_digest = hashlib.sha256(b"text\n").hexdigest() + connection.execute( + "INSERT INTO files(root_id,relative_path,sha256,status,last_seen_run) VALUES('normalized',?,?, 'hashed','run')", + (content_path.relative_to(normalized).as_posix(), content_digest), + ) + validate_fragment_file( + connection, + "run", + normalized, + fragment_path, + fragment_rel, + hashlib.sha256(fragment_path.read_bytes()).hexdigest(), + {"documents/1/source.txt": "1" * 64}, + ) + kinds = {row[0] for row in connection.execute("SELECT kind FROM findings WHERE run_id='run'")} + connection.close() + self.assertIn("fragment_text_checksum_mismatch", kinds) + + +if __name__ == "__main__": + unittest.main() diff --git a/infra/deploy/verify_corpus.py b/infra/deploy/verify_corpus.py new file mode 100644 index 0000000..6d9c2bc --- /dev/null +++ b/infra/deploy/verify_corpus.py @@ -0,0 +1,1026 @@ +#!/usr/bin/env python3 +"""Build a resumable, read-only inventory and validate AkylDash corpus trees.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import ntpath +import os +import sqlite3 +import stat +import sys +import uuid +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Callable, Iterable + +INVENTORY_SCHEMA_VERSION = 1 +CHECKPOINT_INTERVAL = 1000 +CHUNK_SIZE = 1024 * 1024 +MAX_JSON_BYTES = 64 * 1024 * 1024 +REPARSE_POINT = getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0x400) +NORMALIZER_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(NORMALIZER_ROOT)) +from backend.normalization.minjust_cbd import NORMALIZER_VERSION, SCHEMA_VERSION as NORMALIZED_SCHEMA_VERSION # noqa: E402 + + +class InventoryError(RuntimeError): + pass + + +def utc_now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def file_signature(value: os.stat_result) -> tuple[int, int, int, int, int, int]: + return ( + value.st_dev, + value.st_ino, + value.st_size, + value.st_mtime_ns, + value.st_ctime_ns, + value.st_nlink, + ) + + +def is_reparse(value: os.stat_result) -> bool: + return bool(getattr(value, "st_file_attributes", 0) & REPARSE_POINT) + + +def safe_relative_path(value: str) -> str: + path = Path(value) + drive, _ = ntpath.splitdrive(value) + if ( + path.is_absolute() + or Path(value.replace("\\", "/")).is_absolute() + or drive + or "\\" in value + or "\x00" in value + or any(part in {"", ".", ".."} for part in value.replace("\\", "/").split("/")) + ): + raise InventoryError(f"unsafe relative path: {value!r}") + return path.as_posix() + + +def open_regular_file(path: Path, root: Path) -> tuple[int, os.stat_result, str]: + relative = path.relative_to(root).as_posix() + safe_relative_path(relative) + before = path.lstat() + if stat.S_ISLNK(before.st_mode) or is_reparse(before): + raise InventoryError("symlink or reparse point") + if not stat.S_ISREG(before.st_mode): + raise InventoryError("not a regular file") + if before.st_nlink > 1: + raise InventoryError("unexpected hard link") + flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) | getattr(os, "O_NOFOLLOW", 0) + descriptor = os.open(path, flags) + opened = os.fstat(descriptor) + if not stat.S_ISREG(opened.st_mode) or file_signature(opened) != file_signature(before): + os.close(descriptor) + raise InventoryError("file changed or was replaced before reading") + return descriptor, opened, relative + + +def hash_file(path: Path, root: Path) -> tuple[int, str, tuple[int, int, int, int, int, int], str]: + descriptor, before, relative = open_regular_file(path, root) + digest = hashlib.sha256() + try: + with os.fdopen(descriptor, "rb", closefd=False) as source: + for chunk in iter(lambda: source.read(CHUNK_SIZE), b""): + digest.update(chunk) + after = os.fstat(descriptor) + path_after = path.lstat() + finally: + os.close(descriptor) + if ( + file_signature(before) != file_signature(after) + or file_signature(after) != file_signature(path_after) + or stat.S_ISLNK(path_after.st_mode) + or is_reparse(path_after) + ): + raise InventoryError("file changed during inventory") + return after.st_size, digest.hexdigest(), file_signature(after), relative + + +def read_json_checked(path: Path, root: Path, expected_sha256: str | None) -> Any: + descriptor, before, _ = open_regular_file(path, root) + payload = bytearray() + digest = hashlib.sha256() + try: + with os.fdopen(descriptor, "rb", closefd=False) as source: + while chunk := source.read(CHUNK_SIZE): + payload.extend(chunk) + digest.update(chunk) + if len(payload) > MAX_JSON_BYTES: + raise InventoryError("JSON file exceeds the 64 MiB validation limit") + after = os.fstat(descriptor) + path_after = path.lstat() + finally: + os.close(descriptor) + if file_signature(before) != file_signature(after) or file_signature(after) != file_signature(path_after): + raise InventoryError("file changed during validation") + if expected_sha256 is not None and digest.hexdigest() != expected_sha256: + raise InventoryError("file changed after the inventory pass") + return json.loads(payload) + + +def atomic_json(path: Path, value: dict[str, Any]) -> None: + temporary = path.with_name(f".{path.name}.{uuid.uuid4().hex}.tmp") + with temporary.open("x", encoding="utf-8") as target: + json.dump(value, target, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + target.write("\n") + target.flush() + os.fsync(target.fileno()) + os.replace(temporary, path) + + +def database(path: Path) -> sqlite3.Connection: + connection = sqlite3.connect(path) + connection.row_factory = sqlite3.Row + connection.execute("PRAGMA journal_mode=DELETE") + connection.execute("PRAGMA synchronous=FULL") + connection.executescript( + """ + CREATE TABLE IF NOT EXISTS metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL); + CREATE TABLE IF NOT EXISTS files ( + root_id TEXT NOT NULL, + relative_path TEXT NOT NULL, + size INTEGER, + sha256 TEXT, + status TEXT NOT NULL, + device INTEGER, + inode INTEGER, + mtime_ns INTEGER, + ctime_ns INTEGER, + last_seen_run TEXT NOT NULL, + PRIMARY KEY (root_id, relative_path) + ); + CREATE TABLE IF NOT EXISTS directories ( + root_id TEXT NOT NULL, + relative_path TEXT NOT NULL, + last_seen_run TEXT NOT NULL, + PRIMARY KEY (root_id, relative_path) + ); + CREATE TABLE IF NOT EXISTS document_dirs ( + root_id TEXT NOT NULL, + code TEXT NOT NULL, + last_seen_run TEXT NOT NULL, + PRIMARY KEY (root_id, code) + ); + CREATE TABLE IF NOT EXISTS manifest_docs ( + root_id TEXT NOT NULL, + code TEXT NOT NULL, + source_sha256 TEXT, + state TEXT, + schema_version TEXT, + normalizer_version TEXT, + PRIMARY KEY (root_id, code) + ); + CREATE TABLE IF NOT EXISTS manifest_errors ( + root_id TEXT NOT NULL, + code TEXT NOT NULL, + PRIMARY KEY (root_id, code) + ); + CREATE TABLE IF NOT EXISTS findings ( + run_id TEXT NOT NULL, + root_id TEXT NOT NULL, + severity TEXT NOT NULL, + kind TEXT NOT NULL, + relative_path TEXT NOT NULL, + detail TEXT NOT NULL + ); + CREATE INDEX IF NOT EXISTS findings_run ON findings(run_id, severity, kind); + CREATE INDEX IF NOT EXISTS files_run ON files(root_id, last_seen_run); + CREATE INDEX IF NOT EXISTS directories_run ON directories(root_id, last_seen_run); + """ + ) + return connection + + +def finding( + connection: sqlite3.Connection, + run_id: str, + root_id: str, + kind: str, + relative_path: str = "", + detail: str = "", + severity: str = "error", +) -> None: + connection.execute( + "INSERT INTO findings VALUES (?, ?, ?, ?, ?, ?)", + (run_id, root_id, severity, kind, relative_path, detail[:500]), + ) + + +def result_summary( + connection: sqlite3.Connection, + run_id: str, + status: str, + scan_status: str, + verification_status: str, + roots: dict[str, Path], + progress: dict[str, dict[str, Any]], + changed: dict[str, dict[str, int]], +) -> dict[str, Any]: + errors = connection.execute( + "SELECT severity,kind,count(*) AS count FROM findings WHERE run_id=? GROUP BY severity,kind ORDER BY severity,kind", + (run_id,), + ).fetchall() + examples = connection.execute( + "SELECT root_id,kind,relative_path,detail FROM findings WHERE run_id=? ORDER BY rowid LIMIT 25", + (run_id,), + ).fetchall() + return { + "schema_version": INVENTORY_SCHEMA_VERSION, + "run_id": run_id, + "updated_at": utc_now(), + "status": status, + "scan_status": scan_status, + "verification_status": verification_status, + "roots": {key: str(value) for key, value in roots.items()}, + "progress": progress, + "input_changes": changed, + "finding_counts": [dict(row) for row in errors], + "finding_examples": [dict(row) for row in examples], + } + + +def save_progress( + connection: sqlite3.Connection, + result_path: Path, + run_id: str, + roots: dict[str, Path], + progress: dict[str, dict[str, Any]], + changed: dict[str, dict[str, int]], + scan_status: str = "running", + verification_status: str = "not_started", +) -> None: + connection.commit() + atomic_json( + result_path, + result_summary( + connection, run_id, "running", scan_status, verification_status, roots, progress, changed + ), + ) + + +def path_metadata(root: Path) -> dict[str, Any]: + value = root.stat() + return {"path": str(root), "device": value.st_dev, "inode": value.st_ino} + + +def validate_output_dir(output: Path, roots: dict[str, Path]) -> Path: + candidate = output.absolute() + parent = candidate.parent.resolve(strict=True) + candidate = parent / candidate.name + existed = os.path.lexists(candidate) + if existed: + metadata = candidate.lstat() + if stat.S_ISLNK(metadata.st_mode) or is_reparse(metadata) or not stat.S_ISDIR(metadata.st_mode): + raise InventoryError("artifact output must be a real directory, not a link or file") + if stat.S_IMODE(metadata.st_mode) & 0o077: + raise InventoryError("existing artifact output must be private (mode 0700 or stricter)") + for root in roots.values(): + if candidate == root or candidate.is_relative_to(root) or root.is_relative_to(candidate): + raise InventoryError("artifact output must be separate from all source roots") + if not existed: + try: + candidate.mkdir(mode=0o700) + except FileExistsError as error: + raise InventoryError("artifact output appeared during validation; choose a private new path") from error + metadata = candidate.lstat() + if stat.S_ISLNK(metadata.st_mode) or is_reparse(metadata) or not stat.S_ISDIR(metadata.st_mode): + raise InventoryError("artifact output must be a real directory, not a link or file") + if stat.S_IMODE(metadata.st_mode) & 0o077: + raise InventoryError("artifact output must be private (mode 0700 or stricter)") + resolved = candidate.resolve(strict=True) + for root in roots.values(): + if resolved == root or resolved.is_relative_to(root) or root.is_relative_to(resolved): + raise InventoryError("artifact output must be separate from all source roots") + return resolved + + +def insert_or_update_file( + connection: sqlite3.Connection, + run_id: str, + root_id: str, + relative_path: str, + size: int, + sha256: str, + signature: tuple[int, int, int, int, int, int], + changed: dict[str, dict[str, int]], +) -> None: + old = connection.execute( + "SELECT sha256 FROM files WHERE root_id=? AND relative_path=?", + (root_id, relative_path), + ).fetchone() + if old is None: + changed[root_id]["added"] += 1 + elif old["sha256"] is not None and old["sha256"] != sha256: + changed[root_id]["changed"] += 1 + connection.execute( + """INSERT INTO files + (root_id,relative_path,size,sha256,status,device,inode,mtime_ns,ctime_ns,last_seen_run) + VALUES (?,?,?,?,?,?,?,?,?,?) + ON CONFLICT(root_id,relative_path) DO UPDATE SET + size=excluded.size,sha256=excluded.sha256,status=excluded.status, + device=excluded.device,inode=excluded.inode,mtime_ns=excluded.mtime_ns, + ctime_ns=excluded.ctime_ns,last_seen_run=excluded.last_seen_run""", + ( + root_id, relative_path, size, sha256, "hashed", + signature[0], signature[1], signature[3], signature[4], run_id, + ), + ) + + +def record_directory( + connection: sqlite3.Connection, + run_id: str, + root_id: str, + relative_path: str, +) -> None: + connection.execute( + """INSERT INTO directories(root_id,relative_path,last_seen_run) VALUES(?,?,?) + ON CONFLICT(root_id,relative_path) DO UPDATE SET last_seen_run=excluded.last_seen_run""", + (root_id, relative_path, run_id), + ) + parent, _, code = relative_path.rpartition("/") + if parent == "documents" and code: + connection.execute( + """INSERT INTO document_dirs(root_id,code,last_seen_run) VALUES(?,?,?) + ON CONFLICT(root_id,code) DO UPDATE SET last_seen_run=excluded.last_seen_run""", + (root_id, code, run_id), + ) + + +def scan_root( + connection: sqlite3.Connection, + run_id: str, + root_id: str, + root: Path, + output_path: Path, + roots: dict[str, Path], + progress: dict[str, dict[str, Any]], + changed: dict[str, dict[str, int]], + checkpoint_every: int, + stop_after_entries: int | None, + processed_before: int, + progress_callback: Callable[[dict[str, Any]], None] | None, +) -> int: + record_directory(connection, run_id, root_id, "") + stack: list[os.ScandirIterator[str]] = [os.scandir(root)] + since_checkpoint = 0 + try: + while stack: + iterator = stack[-1] + try: + entry = next(iterator) + except StopIteration: + iterator.close() + stack.pop() + continue + relative = Path(entry.path).relative_to(root).as_posix() + try: + safe_relative_path(relative) + before = entry.stat(follow_symlinks=False) + if entry.is_symlink() or is_reparse(before): + finding(connection, run_id, root_id, "unsafe_link", relative, "symlink or reparse point") + elif stat.S_ISDIR(before.st_mode): + record_directory(connection, run_id, root_id, relative) + stack.append(os.scandir(entry.path)) + progress[root_id]["directories"] += 1 + elif stat.S_ISREG(before.st_mode): + if before.st_nlink > 1: + finding(connection, run_id, root_id, "unexpected_hard_link", relative) + else: + try: + size, sha256, signature, relative = hash_file(Path(entry.path), root) + insert_or_update_file( + connection, run_id, root_id, relative, size, sha256, signature, changed + ) + progress[root_id]["files"] += 1 + progress[root_id]["bytes"] += size + except (OSError, InventoryError) as error: + finding(connection, run_id, root_id, "file_read_error", relative, str(error)) + connection.execute( + """INSERT INTO files(root_id,relative_path,size,sha256,status,last_seen_run) + VALUES(?,?,NULL,NULL,'error',?) + ON CONFLICT(root_id,relative_path) DO UPDATE SET + size=NULL,sha256=NULL,status='error',last_seen_run=excluded.last_seen_run""", + (root_id, relative, run_id), + ) + progress[root_id]["errors"] += 1 + else: + finding(connection, run_id, root_id, "unsupported_file_type", relative) + except (OSError, InventoryError) as error: + finding(connection, run_id, root_id, "entry_read_error", relative, str(error)) + progress[root_id]["errors"] += 1 + progress[root_id]["entries"] += 1 + processed_before += 1 + since_checkpoint += 1 + if since_checkpoint >= checkpoint_every: + save_progress(connection, output_path, run_id, roots, progress, changed) + if progress_callback: + progress_callback(result_summary(connection, run_id, "running", "running", "not_started", roots, progress, changed)) + since_checkpoint = 0 + if stop_after_entries is not None and processed_before >= stop_after_entries: + raise KeyboardInterrupt + finally: + for iterator in stack: + iterator.close() + if since_checkpoint: + save_progress(connection, output_path, run_id, roots, progress, changed) + stale_files = connection.execute( + "SELECT relative_path FROM files WHERE root_id=? AND last_seen_run<>?", + (root_id, run_id), + ).fetchall() + for row in stale_files: + finding(connection, run_id, root_id, "removed_file", row["relative_path"], "file absent from resumed input") + changed[root_id]["removed"] += 1 + connection.execute("DELETE FROM files WHERE root_id=? AND last_seen_run<>?", (root_id, run_id)) + stale_dirs = connection.execute( + "SELECT relative_path FROM directories WHERE root_id=? AND last_seen_run<>?", + (root_id, run_id), + ).fetchall() + for row in stale_dirs: + changed[root_id]["removed_directories"] += 1 + connection.execute("DELETE FROM directories WHERE root_id=? AND last_seen_run<>?", (root_id, run_id)) + connection.execute("DELETE FROM document_dirs WHERE root_id=? AND last_seen_run<>?", (root_id, run_id)) + connection.commit() + return processed_before + + +def readonly_manifest(path: Path) -> sqlite3.Connection: + sidecars = [Path(f"{path}-{suffix}") for suffix in ("wal", "shm", "journal")] + present = [sidecar.name for sidecar in sidecars if sidecar.exists()] + if present: + raise InventoryError(f"SQLite sidecar files prevent an immutable read: {', '.join(present)}") + uri = path.resolve(strict=True).as_uri() + "?mode=ro&immutable=1" + connection = sqlite3.connect(uri, uri=True) + connection.row_factory = sqlite3.Row + return connection + + +def load_manifest( + connection: sqlite3.Connection, + run_id: str, + root_id: str, + root: Path, +) -> None: + manifest = root / "manifest.sqlite3" + expected = connection.execute( + "SELECT sha256,status FROM files WHERE root_id=? AND relative_path='manifest.sqlite3'", + (root_id,), + ).fetchone() + if expected is None or expected["status"] != "hashed": + finding(connection, run_id, root_id, "manifest_missing_or_unhashed", "manifest.sqlite3") + return + try: + source = readonly_manifest(manifest) + try: + check = source.execute("PRAGMA quick_check").fetchone()[0] + if check != "ok": + finding(connection, run_id, root_id, "manifest_quick_check_failed", "manifest.sqlite3", str(check)) + tables = {row["name"] for row in source.execute("SELECT name FROM sqlite_master WHERE type='table'")} + required = {"documents"} | ({"errors"} if root_id == "raw" else set()) + if not required.issubset(tables): + finding(connection, run_id, root_id, "manifest_schema_missing", "manifest.sqlite3", ",".join(sorted(required - tables))) + return + columns = {row["name"] for row in source.execute("PRAGMA table_info(documents)")} + if root_id == "raw": + required_columns = {"code", "source_sha256", "fetched_at"} + selected = "code,source_sha256" + else: + required_columns = {"code", "source_sha256", "schema_version", "normalizer_version", "state"} + selected = "code,source_sha256,schema_version,normalizer_version,state" + if not required_columns.issubset(columns): + finding(connection, run_id, root_id, "manifest_columns_missing", "manifest.sqlite3", ",".join(sorted(required_columns - columns))) + return + connection.execute("DELETE FROM manifest_docs WHERE root_id=?", (root_id,)) + for row in source.execute(f"SELECT {selected} FROM documents"): + fields = list(row) + code = str(fields[0]) + try: + safe_relative_path(f"documents/{code}") + except InventoryError: + finding(connection, run_id, root_id, "unsafe_manifest_code", f"documents/{code}") + continue + source_sha256 = fields[1] + state = fields[4] if root_id == "normalized" else "success" + schema_version = fields[2] if root_id == "normalized" else None + normalizer_version = fields[3] if root_id == "normalized" else None + connection.execute( + "INSERT INTO manifest_docs VALUES (?,?,?,?,?,?)", + (root_id, code, source_sha256, state, schema_version, normalizer_version), + ) + if root_id == "normalized" and state != "success": + 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: + 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'") + for row in source.execute("SELECT code FROM errors"): + code = str(row[0]) + connection.execute("INSERT INTO manifest_errors VALUES ('raw',?)", (code,)) + finding(connection, run_id, "raw", "raw_manifest_error", f"documents/{code}", "source fetch error recorded") + finally: + source.close() + except (OSError, sqlite3.Error, InventoryError) as error: + finding(connection, run_id, root_id, "manifest_read_error", "manifest.sqlite3", str(error)) + try: + _, digest, _, _ = hash_file(manifest, root) + if digest != expected["sha256"]: + finding(connection, run_id, root_id, "manifest_changed_during_validation", "manifest.sqlite3") + except (OSError, InventoryError) as error: + finding(connection, run_id, root_id, "manifest_changed_during_validation", "manifest.sqlite3", str(error)) + connection.commit() + + +def dir_code_sets(connection: sqlite3.Connection, root_id: str) -> tuple[set[str], set[str]]: + manifest_codes = { + row[0] + for row in connection.execute("SELECT code FROM manifest_docs WHERE root_id=?", (root_id,)) + } + disk_codes = { + row[0] + for row in connection.execute("SELECT code FROM document_dirs WHERE root_id=?", (root_id,)) + } + return manifest_codes, disk_codes + + +def raw_file_rows(connection: sqlite3.Connection, code: str) -> list[dict[str, str]]: + prefix = f"documents/{code}/" + return [ + {"path": row[0], "sha256": row[1]} + for row in connection.execute( + "SELECT relative_path,sha256 FROM files WHERE root_id='raw' AND relative_path LIKE ? ORDER BY relative_path", + (prefix + "%",), + ) + ] + + +def combined_source_hash(files: Iterable[dict[str, str]]) -> str: + digest = hashlib.sha256() + for item in files: + digest.update(item["path"].encode("utf-8")) + digest.update(b"\0") + digest.update(item["sha256"].encode("ascii")) + digest.update(b"\0") + return digest.hexdigest() + + +def validate_raw( + connection: sqlite3.Connection, + run_id: str, + root: Path, +) -> dict[str, int]: + manifest_codes, disk_codes = dir_code_sets(connection, "raw") + for code in sorted(manifest_codes - disk_codes): + finding(connection, run_id, "raw", "missing_raw_document_directory", f"documents/{code}") + for code in sorted(disk_codes - manifest_codes): + finding(connection, run_id, "raw", "extra_raw_document_directory", f"documents/{code}") + zero_editions = 0 + for code in sorted(manifest_codes & disk_codes): + path = root / "documents" / code / "metadata.json" + expected = connection.execute( + "SELECT sha256 FROM files WHERE root_id='raw' AND relative_path=? AND status='hashed'", + (f"documents/{code}/metadata.json",), + ).fetchone() + if expected is None: + finding(connection, run_id, "raw", "missing_raw_metadata", f"documents/{code}/metadata.json") + continue + try: + metadata = read_json_checked(path, root, expected[0]) + if not isinstance(metadata, dict): + raise InventoryError("raw document metadata must contain an object") + if str(metadata.get("Code")) != code: + finding(connection, run_id, "raw", "raw_document_identity_mismatch", f"documents/{code}/metadata.json") + except (OSError, ValueError, InventoryError) as error: + finding(connection, run_id, "raw", "raw_document_metadata_invalid", f"documents/{code}/metadata.json", str(error)) + edition_dirs = connection.execute( + "SELECT relative_path FROM directories WHERE root_id='raw' AND relative_path LIKE ?", + (f"documents/{code}/editions/%",), + ).fetchall() + edition_codes = [row[0].rsplit("/", 1)[-1] for row in edition_dirs if row[0].count("/") == 3] + if not edition_codes: + zero_editions += 1 + for edition_code in edition_codes: + edition_path = f"documents/{code}/editions/{edition_code}/metadata.json" + row = connection.execute( + "SELECT sha256 FROM files WHERE root_id='raw' AND relative_path=? AND status='hashed'", + (edition_path,), + ).fetchone() + if row is None: + finding(connection, run_id, "raw", "missing_raw_edition_metadata", edition_path) + continue + try: + edition = read_json_checked(root / edition_path, root, row[0]) + if not isinstance(edition, dict): + raise InventoryError("raw edition metadata must contain an object") + if str(edition.get("Code")) != edition_code: + finding(connection, run_id, "raw", "raw_edition_identity_mismatch", edition_path) + except (OSError, ValueError, InventoryError) as error: + finding(connection, run_id, "raw", "raw_edition_metadata_invalid", edition_path, str(error)) + return {"documents": len(manifest_codes), "directories": len(disk_codes), "zero_edition_documents": zero_editions} + + +def validate_fragment_file( + connection: sqlite3.Connection, + run_id: str, + root: Path, + path: Path, + relative: str, + expected_sha256: str, + raw_files: dict[str, str], +) -> tuple[int, str]: + parts = Path(relative).parts + code, edition_code, language = parts[1], parts[3], parts[4] + try: + fragments = read_json_checked(path, root, expected_sha256) + if not isinstance(fragments, list): + raise InventoryError("fragments.json must contain a list") + texts: list[str] = [] + for position, fragment in enumerate(fragments, 1): + if not isinstance(fragment, dict): + finding(connection, run_id, "normalized", "fragment_invalid", relative, f"position {position}") + continue + expected_id = f"document:{code}:edition:{edition_code}:lang:{language}:fragment:{position}" + if ( + fragment.get("id") != expected_id + or str(fragment.get("document_code")) != code + or str(fragment.get("edition_code")) != edition_code + or fragment.get("language") != language + or fragment.get("position") != position + ): + finding(connection, run_id, "normalized", "fragment_identity_mismatch", relative, f"position {position}") + text = fragment.get("text") + if not isinstance(text, str): + finding(connection, run_id, "normalized", "fragment_text_invalid", relative, f"position {position}") + continue + texts.append(text) + text_sha = hashlib.sha256(text.encode("utf-8")).hexdigest() + if fragment.get("text_sha256") != text_sha: + finding(connection, run_id, "normalized", "fragment_text_checksum_mismatch", relative, f"position {position}") + source_path = fragment.get("source_path") + try: + safe_source = safe_relative_path(source_path) + except (TypeError, InventoryError): + finding(connection, run_id, "normalized", "unsafe_fragment_source_path", relative, f"position {position}") + continue + if raw_files and raw_files.get(safe_source) != fragment.get("source_sha256"): + finding(connection, run_id, "normalized", "fragment_source_checksum_mismatch", relative, f"position {position}") + text_path = path.parent / "content.txt" + text_rel = text_path.relative_to(root).as_posix() + text_row = connection.execute( + "SELECT sha256 FROM files WHERE root_id='normalized' AND relative_path=? AND status='hashed'", + (text_rel,), + ).fetchone() + if text_row is None: + finding(connection, run_id, "normalized", "missing_content_text", text_rel) + else: + try: + size, digest, _, _ = hash_file(text_path, root) + if digest != text_row[0]: + finding(connection, run_id, "normalized", "content_text_changed", text_rel) + expected_text = "\n\n".join(texts) + ("\n" if texts else "") + if size <= 128 * 1024 * 1024: + descriptor, before, _ = open_regular_file(text_path, root) + try: + with os.fdopen(descriptor, "rb", closefd=False) as source_file: + actual_text = source_file.read(128 * 1024 * 1024 + 1).decode("utf-8") + after = os.fstat(descriptor) + path_after = text_path.lstat() + finally: + os.close(descriptor) + if file_signature(before) != file_signature(after) or file_signature(after) != file_signature(path_after): + raise InventoryError("file changed during validation") + if actual_text != expected_text: + finding(connection, run_id, "normalized", "content_text_fragment_mismatch", text_rel) + except (OSError, UnicodeError, InventoryError) as error: + finding(connection, run_id, "normalized", "content_text_invalid", text_rel, str(error)) + return len(fragments), "ok" + except (OSError, ValueError, InventoryError) as error: + finding(connection, run_id, "normalized", "fragment_file_invalid", relative, str(error)) + return 0, "error" + + +def validate_normalized( + connection: sqlite3.Connection, + run_id: str, + root: Path, + raw_present: bool, +) -> dict[str, int]: + manifest_codes, disk_codes = dir_code_sets(connection, "normalized") + for code in sorted(manifest_codes - disk_codes): + finding(connection, run_id, "normalized", "missing_normalized_document_directory", f"documents/{code}") + for code in sorted(disk_codes - manifest_codes): + finding(connection, run_id, "normalized", "extra_normalized_document_directory", f"documents/{code}") + total_fragments = zero_fragment_files = zero_fragment_docs = 0 + for code in sorted(manifest_codes & disk_codes): + document_rel = f"documents/{code}/document.json" + row = connection.execute( + "SELECT sha256 FROM files WHERE root_id='normalized' AND relative_path=? AND status='hashed'", + (document_rel,), + ).fetchone() + manifest = connection.execute( + "SELECT source_sha256,state,schema_version,normalizer_version FROM manifest_docs WHERE root_id='normalized' AND code=?", + (code,), + ).fetchone() + if row is None or manifest is None: + finding(connection, run_id, "normalized", "missing_document_json", document_rel) + continue + try: + document = read_json_checked(root / document_rel, root, row[0]) + if not isinstance(document, dict): + raise InventoryError("normalized document must contain an object") + if str(document.get("source_code")) != code: + finding(connection, run_id, "normalized", "document_identity_mismatch", document_rel) + if str(document.get("schema_version")) != str(manifest["schema_version"]): + finding(connection, run_id, "normalized", "document_schema_manifest_mismatch", document_rel) + normalizer = document.get("normalizer") or {} + if not isinstance(normalizer, dict) or normalizer.get("version") != manifest["normalizer_version"]: + finding(connection, run_id, "normalized", "document_normalizer_manifest_mismatch", document_rel) + source = document.get("source") or {} + if not isinstance(source, dict): + finding(connection, run_id, "normalized", "document_source_invalid", document_rel) + source = {} + if source.get("sha256") != manifest["source_sha256"]: + finding(connection, run_id, "normalized", "document_source_manifest_mismatch", document_rel) + if source.get("path") != f"documents/{code}": + finding(connection, run_id, "normalized", "document_source_path_mismatch", document_rel) + raw_entries = raw_file_rows(connection, code) if raw_present else [] + raw_files = {item["path"]: item["sha256"] for item in raw_entries} + if raw_present: + computed_source = combined_source_hash(raw_entries) + if computed_source != manifest["source_sha256"]: + finding(connection, run_id, "normalized", "normalized_source_hash_mismatch", document_rel) + if source.get("files") != raw_entries: + finding(connection, run_id, "normalized", "document_source_file_list_mismatch", document_rel) + summaries = document.get("editions") + if not isinstance(summaries, list): + finding(connection, run_id, "normalized", "document_editions_invalid", document_rel) + continue + summary_codes = [str(item.get("source_code")) for item in summaries if isinstance(item, dict)] + edition_dir_prefix = f"documents/{code}/editions/" + edition_dirs = { + row[0][len(edition_dir_prefix):] + for row in connection.execute( + "SELECT relative_path FROM directories WHERE root_id='normalized' AND relative_path LIKE ?", + (edition_dir_prefix + "%",), + ) + if row[0].count("/") == 3 and row[0].startswith(edition_dir_prefix) + } + if set(summary_codes) != edition_dirs or len(summary_codes) != len(set(summary_codes)): + finding(connection, run_id, "normalized", "edition_directory_list_mismatch", f"documents/{code}/editions") + doc_fragments = 0 + for edition_code in sorted(edition_dirs): + edition_rel = f"documents/{code}/editions/{edition_code}/edition.json" + edition_row = connection.execute( + "SELECT sha256 FROM files WHERE root_id='normalized' AND relative_path=? AND status='hashed'", + (edition_rel,), + ).fetchone() + if edition_row is None: + finding(connection, run_id, "normalized", "missing_edition_json", edition_rel) + continue + try: + edition = read_json_checked(root / edition_rel, root, edition_row[0]) + if not isinstance(edition, dict): + raise InventoryError("normalized edition must contain an object") + if str(edition.get("source_code")) != edition_code: + finding(connection, run_id, "normalized", "edition_identity_mismatch", edition_rel) + if edition.get("source_code") not in summary_codes: + finding(connection, run_id, "normalized", "edition_not_in_document", edition_rel) + if raw_present: + source_prefix = f"documents/{code}/editions/{edition_code}/" + expected_edition_files = [item for item in raw_entries if item["path"].startswith(source_prefix)] + edition_source = edition.get("source") or {} + if ( + not isinstance(edition_source, dict) + or edition_source.get("path") != source_prefix.rstrip("/") + or edition_source.get("files") != expected_edition_files + ): + finding(connection, run_id, "normalized", "edition_source_file_list_mismatch", edition_rel) + images = edition.get("images") + if not isinstance(images, list): + finding(connection, run_id, "normalized", "edition_images_invalid", edition_rel) + else: + for image in images: + if not isinstance(image, dict): + finding(connection, run_id, "normalized", "edition_image_invalid", edition_rel) + continue + image_path = image.get("source_path") + if image_path is None: + continue + try: + safe_image_path = safe_relative_path(image_path) + except (TypeError, InventoryError): + finding(connection, run_id, "normalized", "unsafe_image_source_path", edition_rel) + continue + if raw_present and raw_files.get(safe_image_path) != image.get("source_sha256"): + finding(connection, run_id, "normalized", "image_source_checksum_mismatch", edition_rel, safe_image_path) + languages = edition.get("available_languages") + if not isinstance(languages, list): + finding(connection, run_id, "normalized", "edition_languages_invalid", edition_rel) + continue + for language in languages: + if language not in {"ru", "ky"}: + finding(connection, run_id, "normalized", "unexpected_language", edition_rel, str(language)) + continue + fragment_rel = f"documents/{code}/editions/{edition_code}/{language}/fragments.json" + fragment_row = connection.execute( + "SELECT sha256 FROM files WHERE root_id='normalized' AND relative_path=? AND status='hashed'", + (fragment_rel,), + ).fetchone() + for suffix in ("content.html", "content.txt", "fragments.json"): + required_rel = f"documents/{code}/editions/{edition_code}/{language}/{suffix}" + if connection.execute( + "SELECT 1 FROM files WHERE root_id='normalized' AND relative_path=? AND status='hashed'", + (required_rel,), + ).fetchone() is None: + finding(connection, run_id, "normalized", "missing_language_output", required_rel) + if fragment_row is None: + continue + count, status = validate_fragment_file( + connection, run_id, root, root / fragment_rel, fragment_rel, fragment_row[0], raw_files + ) + if status == "ok": + doc_fragments += count + total_fragments += count + if count == 0: + zero_fragment_files += 1 + except (OSError, ValueError, InventoryError) as error: + finding(connection, run_id, "normalized", "edition_json_invalid", edition_rel, str(error)) + if doc_fragments == 0: + zero_fragment_docs += 1 + finding(connection, run_id, "normalized", "zero_fragment_document", f"documents/{code}", severity="warning") + except (OSError, ValueError, InventoryError) as error: + finding(connection, run_id, "normalized", "document_json_invalid", document_rel, str(error)) + return { + "documents": len(manifest_codes), + "directories": len(disk_codes), + "fragments": total_fragments, + "zero_fragment_documents": zero_fragment_docs, + "zero_fragment_files": zero_fragment_files, + } + + +def setup_run( + output_dir: Path, + roots: dict[str, Path], + resume: bool, +) -> tuple[sqlite3.Connection, str, Path]: + output_dir = validate_output_dir(output_dir, roots) + database_path = output_dir / "inventory.sqlite3" + result_path = output_dir / "result.json" + database_exists = os.path.lexists(database_path) + if database_exists: + metadata = database_path.lstat() + if ( + stat.S_ISLNK(metadata.st_mode) + or is_reparse(metadata) + or not stat.S_ISREG(metadata.st_mode) + or metadata.st_nlink > 1 + ): + raise InventoryError("inventory.sqlite3 must be a private regular file") + if database_exists and not resume: + raise InventoryError("inventory already exists; use --resume or choose a new artifact directory") + if resume and not database_exists: + raise InventoryError("--resume requested but inventory.sqlite3 is missing") + if not resume and next(output_dir.iterdir(), None) is not None: + raise InventoryError("artifact output must be empty for a new inventory") + connection = database(database_path) + root_metadata = {key: path_metadata(value) for key, value in roots.items()} + if resume: + stored = {row["key"]: json.loads(row["value"]) for row in connection.execute("SELECT key,value FROM metadata")} + if stored.get("schema_version") != INVENTORY_SCHEMA_VERSION: + raise InventoryError("inventory schema differs; use a new artifact directory") + if stored.get("roots") != root_metadata: + raise InventoryError("--resume roots differ from the saved inventory") + else: + connection.executemany( + "INSERT INTO metadata VALUES (?,?)", + (("schema_version", json.dumps(INVENTORY_SCHEMA_VERSION)), ("roots", json.dumps(root_metadata, sort_keys=True))), + ) + run_id = uuid.uuid4().hex + connection.execute( + "INSERT INTO metadata(key,value) VALUES('last_run_id',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value", + (json.dumps(run_id),), + ) + connection.execute("DELETE FROM manifest_docs") + connection.execute("DELETE FROM manifest_errors") + connection.commit() + return connection, run_id, result_path + + +def verify_corpus( + raw_root: Path | None, + normalized_root: Path | None, + output_dir: Path, + resume: bool = False, + checkpoint_every: int = CHECKPOINT_INTERVAL, + stop_after_entries: int | None = None, + progress_callback: Callable[[dict[str, Any]], None] | None = None, +) -> dict[str, Any]: + if raw_root is None and normalized_root is None: + raise InventoryError("at least one of --raw-root or --normalized-root is required") + roots: dict[str, Path] = {} + for key, value in (("raw", raw_root), ("normalized", normalized_root)): + if value is None: + continue + metadata = value.lstat() + if stat.S_ISLNK(metadata.st_mode) or is_reparse(metadata) or not stat.S_ISDIR(metadata.st_mode): + raise InventoryError(f"source root is not a real directory: {value}") + path = value.resolve(strict=True) + roots[key] = path + if checkpoint_every <= 0: + raise InventoryError("checkpoint interval must be greater than zero") + connection, run_id, result_path = setup_run(output_dir, roots, resume) + progress = {key: {"entries": 0, "files": 0, "directories": 0, "bytes": 0, "errors": 0} for key in roots} + changed = {key: {"added": 0, "changed": 0, "removed": 0, "removed_directories": 0} for key in roots} + atomic_json(result_path, result_summary(connection, run_id, "running", "running", "not_started", roots, progress, changed)) + processed = 0 + try: + for root_id, root in roots.items(): + processed = scan_root( + connection, run_id, root_id, root, result_path, roots, progress, changed, + checkpoint_every, stop_after_entries, processed, progress_callback, + ) + load_manifest(connection, run_id, root_id, root) + save_progress(connection, result_path, run_id, roots, progress, changed, "running", "validating") + raw_result = validate_raw(connection, run_id, roots["raw"]) if "raw" in roots else None + normalized_result = ( + validate_normalized(connection, run_id, roots["normalized"], "raw" in roots) + if "normalized" in roots else None + ) + connection.commit() + finding_counts = connection.execute( + "SELECT severity,count(*) FROM findings WHERE run_id=? GROUP BY severity", + (run_id,), + ).fetchall() + fatal = any(row[0] == "error" and row[1] for row in finding_counts) + status = "failed" if fatal else "complete" + summary = result_summary( + connection, run_id, status, "complete", "failed" if fatal else "passed", roots, progress, changed + ) + summary["manifest_summary"] = {"raw": raw_result, "normalized": normalized_result} + atomic_json(result_path, summary) + return summary + except KeyboardInterrupt: + connection.commit() + summary = result_summary( + connection, run_id, "incomplete", "incomplete", "not_finished", roots, progress, changed + ) + atomic_json(result_path, summary) + return summary + except Exception as error: + connection.commit() + finding(connection, run_id, "inventory", "run_failed", "", str(error)) + connection.commit() + summary = result_summary( + connection, run_id, "failed", "incomplete", "not_finished", roots, progress, changed + ) + atomic_json(result_path, summary) + return summary + finally: + connection.close() + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--raw-root", type=Path) + parser.add_argument("--normalized-root", type=Path) + parser.add_argument("--output-dir", required=True, type=Path) + parser.add_argument("--resume", action="store_true") + parser.add_argument("--checkpoint-every", type=int, default=CHECKPOINT_INTERVAL) + arguments = parser.parse_args() + + def progress(value: dict[str, Any]) -> None: + print(json.dumps({ + "run_id": value["run_id"], + "status": value["status"], + "progress": value["progress"], + "updated_at": value["updated_at"], + }, ensure_ascii=False), flush=True) + + try: + result = verify_corpus( + arguments.raw_root, + arguments.normalized_root, + arguments.output_dir, + resume=arguments.resume, + checkpoint_every=arguments.checkpoint_every, + progress_callback=progress, + ) + except (InventoryError, OSError) as error: + print(f"ERROR {error}", file=sys.stderr) + return 2 + print(json.dumps(result, ensure_ascii=False, sort_keys=True), flush=True) + return 0 if result["status"] == "complete" else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) -- 2.54.0 From 4d106903e113c33e588e45dfe5a3e806312a9289 Mon Sep 17 00:00:00 2001 From: agent Date: Sun, 18 Oct 2026 19:22:00 +0300 Subject: [PATCH 2/5] Fix corpus inventory indexing and provenance --- docs/operations/CORPUS_HANDOFF.md | 75 +++++++++++-- docs/operations/CURRENT_STATE.md | 23 ++-- docs/operations/DEPLOYMENT.md | 5 +- infra/deploy/test_verify_corpus.py | 164 +++++++++++++++++++++++++++++ infra/deploy/verify_corpus.py | 88 +++++++++++++--- 5 files changed, 319 insertions(+), 36 deletions(-) diff --git a/docs/operations/CORPUS_HANDOFF.md b/docs/operations/CORPUS_HANDOFF.md index 47cdba9..ea803a2 100644 --- a/docs/operations/CORPUS_HANDOFF.md +++ b/docs/operations/CORPUS_HANDOFF.md @@ -17,8 +17,10 @@ python3 infra/deploy/verify_corpus.py \ --output-dir "$artifact_dir" ``` -The tool writes `result.json` and `inventory.sqlite3` into that directory. Keep -the directory private: the inventory contains source paths and checksums. Exit +The tool writes `result.json`, `inventory.sqlite3`, and a persistent lock file +into that directory. Keep the directory private: the inventory contains source +paths and checksums. A second run against the same artifact directory is +rejected while a scan is active. Exit status `0` means the scan and structural checks completed without errors; `1` means findings were recorded or the run is incomplete; `2` means the arguments or artifact path were invalid. Read `result.json` to distinguish those states. @@ -31,19 +33,65 @@ reports added, changed, and removed paths relative to entries already stored in the inventory. After resuming an interrupted first pass, those counters are relative to the partial saved inventory; the full rehash and final checks still cover every path. A successful result therefore describes the latest complete -pass. Run against a quiescent corpus; do not -normalize, fetch, or copy files into either source tree during the scan. Do not -point `--output-dir` inside either -source tree or through a symlink. The verifier rejects such paths before +pass. Run against a quiescent corpus; do not normalize, fetch, or copy files +into either source tree during the scan. Do not point `--output-dir` inside +either source tree or through a symlink. The verifier rejects such paths before creating or changing them; it also flags symlinks, reparse points, hard links, unsupported file types, and files that change while being read. +The inventory database schema is versioned. Resume requires the same root +metadata and verifier SHA-256 recorded in `result.json`; start a new artifact +directory after changing the verifier. `entries` counts files plus directories; +`files` and `directories` are separate counts. Document-prefix checks use +indexed path ranges, and stale-path reporting streams rows from SQLite. + The audit checks SQLite integrity, manifest/document directory agreement, normalized schema and normalizer versions, source inventory checksums, fragment identities and text hashes, required output files, and text assembled from fragments. It does not establish the legal authority, completeness, or semantic correctness of MinJust content. +## Compare source and staging before transfer + +Run a complete inventory of the approved normalized source and the copied +staging tree into separate private artifact directories, using +`--normalized-root` for each. Require `status: complete` and +`verification_status: passed` in both results. Then compare every relative +path, size, and SHA-256 from the two `inventory.sqlite3` files. This streaming +comparison uses constant memory and reports only a mismatch count and the first +few differing paths: + +```bash +python3 - "$source_artifacts/inventory.sqlite3" "$staging_artifacts/inventory.sqlite3" <<'PY' +import itertools +import sqlite3 +import sys + +def rows(path): + db = sqlite3.connect(f"file:{path}?mode=ro&immutable=1", uri=True) + yield from db.execute( + "SELECT relative_path,size,sha256 FROM files " + "WHERE root_id='normalized' AND status='hashed' ORDER BY relative_path" + ) + db.close() + +differences = 0 +for source, staging in itertools.zip_longest(rows(sys.argv[1]), rows(sys.argv[2])): + if source != staging: + differences += 1 + if differences <= 20: + print("mismatch:", source, staging) +print("mismatch_rows:", differences) +sys.exit(1 if differences else 0) +PY +``` + +Use absolute artifact database paths in the command. A zero exit status confirms +that both inventories contain the same ordered path/size/SHA-256 rows. Keep the +source and staging trees quiescent during their scans and comparison. This is +an equality check only; the owner still reviews the code `6` disposition and +approves the transfer destination. + ## Current data disposition As measured on 2026-10-18, both local manifests contain 209,958 documents and @@ -65,9 +113,18 @@ From the VPS, a metadata-only HTTPS check to `CheckAvailable` failed at the HTTP/2 transport layer (`curl` exit 92, `INTERNAL_ERROR`, HTTP status `000`, TLS verification `0`). An HTTP/1.1 retry was not completed. -The full local file inventory and normalized cross-check are still in progress -until a `result.json` reports `status: complete` or `status: failed` and -`scan_status: complete`. Do not treat progress output as a completed audit. +The first full pass (run `b4a4556e30cc464d90bc8d75ef1673e9`) stopped before +validation with `Python int too large to convert to SQLite INTEGER` after +2,120,420 raw entries and 397,856 normalized entries. This was an inventory +metadata storage failure, not a corpus finding. The verifier now stores file +identity values as text and records its own SHA-256. A complete rehash with the +updated verifier is required; do not treat the failed pass or progress output +as a completed audit. + +A fresh full rehash is now running in `/tmp/akyldash-corpus-audit-final.gtsD87vc`, +run `f4dd5e53020f442bbbecee2d0e4986a6`, verifier SHA-256 +`83dac230099fc91972cd14e86e9f8ec72d13bca08f53d402f13a169e096d98f5`. Check +its private `result.json` for final counters, findings, and manifest comparison. ## Production transfer and indexing gates diff --git a/docs/operations/CURRENT_STATE.md b/docs/operations/CURRENT_STATE.md index 2151960..308d710 100644 --- a/docs/operations/CURRENT_STATE.md +++ b/docs/operations/CURRENT_STATE.md @@ -133,12 +133,18 @@ protected file metadata remain unverified. is 31,571,968 bytes with SHA-256 `1679c20e6d0f3e5a9b50737c006c4417a28ebcaf7467d194d3bf8243538f83bf`; all entries are `success`, schema `1`, normalizer `1.0.0`. -- A streaming full-file inventory and cross-check is running read-only with - artifacts only in `/tmp/akyldash-corpus-audit.Mjr6aa`. Latest progress was - run `b4a4556e30cc464d90bc8d75ef1673e9`; live counters are in its private - `result.json`, updated every 1,000 entries. The normalized tree has not yet - started. Progress is not a completed audit; record the final result and full - manifest comparison when the run finishes. +- The read-only full-file inventory in `/tmp/akyldash-corpus-audit.Mjr6aa` + stopped before validation in run `b4a4556e30cc464d90bc8d75ef1673e9` with + `Python int too large to convert to SQLite INTEGER`. At stop it had processed + 2,120,420 raw entries (1,081,428 files, 1,038,992 directories) and 397,856 + normalized entries (246,321 files, 151,535 directories). This is a verifier + metadata storage failure, not a corpus finding. A full rehash with the + corrected, versioned verifier is required before readiness can be stated. +- A fresh full rehash is running in `/tmp/akyldash-corpus-audit-final.gtsD87vc`, + run `f4dd5e53020f442bbbecee2d0e4986a6`, with verifier SHA-256 + `83dac230099fc91972cd14e86e9f8ec72d13bca08f53d402f13a169e096d98f5`. + `result.json` currently reports `scan_status: running`; its live counters + must be recorded after completion. Sources remain read-only. - Through the repository client, bounded MinJust OpenData requests for `GetDocumentListByQuery`, `GetDocument(6)`, control `GetDocument(1096)`, and `CheckAvailable` each returned HTTP `404`. The list API returned no code list; @@ -194,6 +200,7 @@ protected file metadata remain unverified. Adds a read-only, resumable full-file corpus inventory and normalized-data cross-check, synthetic regression tests, and an operator handoff runbook. CI's existing `infra/deploy` unittest discovery includes the new tests. No workflow, -runner, VPS, production data, or Docker settings are changed. The real corpus -scan is still in progress and the source API's code `6` outcome remains +runner, VPS, production data, or Docker settings are changed. The first real +corpus scan stopped on an integer-storage bug; full validation with the +corrected verifier is still required. The source API's code `6` outcome remains unconfirmed. diff --git a/docs/operations/DEPLOYMENT.md b/docs/operations/DEPLOYMENT.md index a59882c..6e44d8b 100644 --- a/docs/operations/DEPLOYMENT.md +++ b/docs/operations/DEPLOYMENT.md @@ -71,8 +71,9 @@ Application deployment remains owner-run. The repository-scoped `act_runner` separate loop-backed filesystem. CI run 75 and smoke run 76 succeeded on main head `0e276ae6`. Runner job containers have neither Docker socket nor production secrets. CI has no image publishing or production deployment step. The -installed sudo helpers support fixed verification, read-only Compose -`config`/`status`, and guarded rollback. They do not start or stop services. +repository defines sudo helpers for fixed verification, read-only Compose +`config`/`status`, and guarded rollback, but none is installed on the VPS. They +do not start or stop services. ## CI diff --git a/infra/deploy/test_verify_corpus.py b/infra/deploy/test_verify_corpus.py index 3970dbd..ca68ceb 100644 --- a/infra/deploy/test_verify_corpus.py +++ b/infra/deploy/test_verify_corpus.py @@ -1,6 +1,7 @@ import hashlib import json import os +import fcntl import sqlite3 import tempfile import unittest @@ -9,6 +10,10 @@ from pathlib import Path from verify_corpus import ( InventoryError, database, + insert_or_update_file, + prefix_upper_bound, + raw_file_rows, + combined_source_hash, safe_relative_path, validate_fragment_file, verify_corpus, @@ -34,6 +39,70 @@ class CorpusInventoryTests(unittest.TestCase): ) connection.close() + def _nonempty_corpus(self): + document = self.raw / "documents/1" + edition = document / "editions/2" + edition.mkdir(parents=True) + (document / "metadata.json").write_text('{"Code":"1"}\n', encoding="utf-8") + (edition / "metadata.json").write_text('{"Code":"2"}\n', encoding="utf-8") + html = edition / "ru.html" + html.write_text("

Фрагмент

\n", encoding="utf-8") + source_files = [] + for path in sorted(document.rglob("*")): + if path.is_file(): + digest = hashlib.sha256(path.read_bytes()).hexdigest() + source_files.append({"path": path.relative_to(self.raw).as_posix(), "sha256": digest}) + source_sha = combined_source_hash(source_files) + raw_manifest = sqlite3.connect(self.raw / "manifest.sqlite3") + raw_manifest.execute("INSERT INTO documents VALUES (1, ?, 'now')", (source_sha,)) + raw_manifest.commit() + raw_manifest.close() + + normalized = self.base / "normalized" + language = normalized / "documents/1/editions/2/ru" + language.mkdir(parents=True) + fragment_text = "Фрагмент" + fragments = [{ + "id": "document:1:edition:2:lang:ru:fragment:1", + "document_code": "1", + "edition_code": "2", + "language": "ru", + "position": 1, + "text": fragment_text, + "text_sha256": hashlib.sha256(fragment_text.encode("utf-8")).hexdigest(), + "source_path": "documents/1/editions/2/ru.html", + "source_sha256": next(item["sha256"] for item in source_files if item["path"].endswith("/ru.html")), + }] + (language / "fragments.json").write_text(json.dumps(fragments, ensure_ascii=False), encoding="utf-8") + (language / "content.txt").write_text(fragment_text + "\n", encoding="utf-8") + (language / "content.html").write_text("

Фрагмент

\n", encoding="utf-8") + document_json = { + "schema_version": "1", + "source_code": "1", + "normalizer": {"version": "1.0.0"}, + "source": {"path": "documents/1", "files": source_files, "sha256": source_sha}, + "editions": [{"source_code": "2"}], + } + (normalized / "documents/1/document.json").write_text(json.dumps(document_json), encoding="utf-8") + edition_json = { + "source_code": "2", + "available_languages": ["ru"], + "source": { + "path": "documents/1/editions/2", + "files": [item for item in source_files if item["path"].startswith("documents/1/editions/2/")], + }, + "images": [], + } + (normalized / "documents/1/editions/2/edition.json").write_text(json.dumps(edition_json), encoding="utf-8") + normalized_manifest = sqlite3.connect(normalized / "manifest.sqlite3") + normalized_manifest.execute( + "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', '1.0.0', 'success')", (source_sha,)) + normalized_manifest.commit() + normalized_manifest.close() + return normalized, source_sha + def test_output_inside_source_is_rejected_without_creating_it(self): output = self.raw / "artifacts" before_mode = self.raw.stat().st_mode @@ -106,6 +175,68 @@ class CorpusInventoryTests(unittest.TestCase): ).fetchone() self.assertEqual(rows[0], hashlib.sha256(b"after").hexdigest()) + def test_large_filesystem_identity_values_are_stored_without_integer_overflow(self): + connection = database(self.base / "identity.sqlite3") + insert_or_update_file( + connection, "run", "raw", "payload", 1, "a" * 64, + (2**100, 2**120, 1, 2**110, 2**115, 1), {"raw": {"added": 0, "changed": 0}}, + ) + device, inode, mtime_ns, ctime_ns = connection.execute( + "SELECT device,inode,mtime_ns,ctime_ns FROM files" + ).fetchone() + self.assertEqual((device, inode, mtime_ns, ctime_ns), tuple(map(str, (2**100, 2**120, 2**110, 2**115)))) + connection.close() + + def test_document_prefix_uses_primary_key_range_and_excludes_neighbor_codes(self): + connection = database(self.base / "paths.sqlite3") + prefix = "documents/1/" + upper = prefix_upper_bound(prefix) + plan = connection.execute( + "EXPLAIN QUERY PLAN SELECT relative_path FROM files WHERE root_id='raw' AND relative_path>=? AND relative_path sqlite3.Connection: size INTEGER, sha256 TEXT, status TEXT NOT NULL, - device INTEGER, - inode INTEGER, - mtime_ns INTEGER, - ctime_ns INTEGER, + device TEXT, + inode TEXT, + mtime_ns TEXT, + ctime_ns TEXT, last_seen_run TEXT NOT NULL, PRIMARY KEY (root_id, relative_path) ); @@ -199,6 +202,35 @@ def database(path: Path) -> sqlite3.Connection: return connection +def prefix_upper_bound(prefix: str) -> str: + for index in range(len(prefix) - 1, -1, -1): + codepoint = ord(prefix[index]) + if codepoint < 0x10FFFF: + return prefix[:index] + chr(codepoint + 1) + raise InventoryError("path prefix has no upper bound") + + +@contextmanager +def inventory_lock(output_dir: Path): + path = output_dir / ".inventory.lock" + descriptor = os.open(path, os.O_CREAT | os.O_RDWR | getattr(os, "O_NOFOLLOW", 0), 0o600) + try: + metadata = os.fstat(descriptor) + if ( + not stat.S_ISREG(metadata.st_mode) + or metadata.st_nlink != 1 + or stat.S_IMODE(metadata.st_mode) & 0o077 + ): + raise InventoryError("inventory lock must be a regular private file") + try: + fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError as error: + raise InventoryError("another inventory run is using this artifact directory") from error + yield + finally: + os.close(descriptor) + + def finding( connection: sqlite3.Connection, run_id: str, @@ -234,6 +266,7 @@ def result_summary( ).fetchall() return { "schema_version": INVENTORY_SCHEMA_VERSION, + "verifier_sha256": TOOL_SHA256, "run_id": run_id, "updated_at": utc_now(), "status": status, @@ -330,7 +363,7 @@ def insert_or_update_file( ctime_ns=excluded.ctime_ns,last_seen_run=excluded.last_seen_run""", ( root_id, relative_path, size, sha256, "hashed", - signature[0], signature[1], signature[3], signature[4], run_id, + *(str(value) for value in (signature[0], signature[1], signature[3], signature[4])), run_id, ), ) @@ -435,7 +468,7 @@ def scan_root( stale_files = connection.execute( "SELECT relative_path FROM files WHERE root_id=? AND last_seen_run<>?", (root_id, run_id), - ).fetchall() + ) for row in stale_files: finding(connection, run_id, root_id, "removed_file", row["relative_path"], "file absent from resumed input") changed[root_id]["removed"] += 1 @@ -443,7 +476,7 @@ def scan_root( stale_dirs = connection.execute( "SELECT relative_path FROM directories WHERE root_id=? AND last_seen_run<>?", (root_id, run_id), - ).fetchall() + ) for row in stale_dirs: changed[root_id]["removed_directories"] += 1 connection.execute("DELETE FROM directories WHERE root_id=? AND last_seen_run<>?", (root_id, run_id)) @@ -557,8 +590,8 @@ def raw_file_rows(connection: sqlite3.Connection, code: str) -> list[dict[str, s return [ {"path": row[0], "sha256": row[1]} for row in connection.execute( - "SELECT relative_path,sha256 FROM files WHERE root_id='raw' AND relative_path LIKE ? ORDER BY relative_path", - (prefix + "%",), + "SELECT relative_path,sha256 FROM files WHERE root_id='raw' AND relative_path>=? AND relative_path=? AND relative_path=? AND relative_path dict[str, Any]: connection, run_id, result_path = setup_run(output_dir, roots, resume) progress = {key: {"entries": 0, "files": 0, "directories": 0, "bytes": 0, "errors": 0} for key in roots} changed = {key: {"added": 0, "changed": 0, "removed": 0, "removed_directories": 0} for key in roots} -- 2.54.0 From 0979b8f544ac711ed1ce72e19028343a27ca8529 Mon Sep 17 00:00:00 2001 From: agent Date: Sun, 18 Oct 2026 19:28:27 +0300 Subject: [PATCH 3/5] Reject empty corpus staging comparisons --- docs/operations/CORPUS_HANDOFF.md | 43 ++---- docs/operations/CURRENT_STATE.md | 3 +- infra/deploy/compare_corpus_inventories.py | 141 ++++++++++++++++++ .../deploy/test_compare_corpus_inventories.py | 94 ++++++++++++ 4 files changed, 249 insertions(+), 32 deletions(-) create mode 100644 infra/deploy/compare_corpus_inventories.py create mode 100644 infra/deploy/test_compare_corpus_inventories.py diff --git a/docs/operations/CORPUS_HANDOFF.md b/docs/operations/CORPUS_HANDOFF.md index ea803a2..63e9176 100644 --- a/docs/operations/CORPUS_HANDOFF.md +++ b/docs/operations/CORPUS_HANDOFF.md @@ -56,41 +56,22 @@ semantic correctness of MinJust content. Run a complete inventory of the approved normalized source and the copied staging tree into separate private artifact directories, using `--normalized-root` for each. Require `status: complete` and -`verification_status: passed` in both results. Then compare every relative -path, size, and SHA-256 from the two `inventory.sqlite3` files. This streaming -comparison uses constant memory and reports only a mismatch count and the first -few differing paths: +`verification_status: passed` in both results. The comparison command also +requires each manifest to contain at least one document and each normalized +tree to contain at least one fragment; an empty source or staging tree is a +failure. ```bash -python3 - "$source_artifacts/inventory.sqlite3" "$staging_artifacts/inventory.sqlite3" <<'PY' -import itertools -import sqlite3 -import sys - -def rows(path): - db = sqlite3.connect(f"file:{path}?mode=ro&immutable=1", uri=True) - yield from db.execute( - "SELECT relative_path,size,sha256 FROM files " - "WHERE root_id='normalized' AND status='hashed' ORDER BY relative_path" - ) - db.close() - -differences = 0 -for source, staging in itertools.zip_longest(rows(sys.argv[1]), rows(sys.argv[2])): - if source != staging: - differences += 1 - if differences <= 20: - print("mismatch:", source, staging) -print("mismatch_rows:", differences) -sys.exit(1 if differences else 0) -PY +python3 infra/deploy/compare_corpus_inventories.py \ + "$source_artifacts/result.json" "$staging_artifacts/result.json" ``` -Use absolute artifact database paths in the command. A zero exit status confirms -that both inventories contain the same ordered path/size/SHA-256 rows. Keep the -source and staging trees quiescent during their scans and comparison. This is -an equality check only; the owner still reviews the code `6` disposition and -approves the transfer destination. +The checker streams every relative path, size, and SHA-256 from both SQLite +inventories, and requires the same verifier/schema version and different +normalized roots. A zero exit status confirms a non-empty staging tree matches +the source. Keep both trees quiescent during their scans and comparison. This +is an equality check only; the owner still reviews the code `6` disposition +and approves the transfer destination. ## Current data disposition diff --git a/docs/operations/CURRENT_STATE.md b/docs/operations/CURRENT_STATE.md index 308d710..373b961 100644 --- a/docs/operations/CURRENT_STATE.md +++ b/docs/operations/CURRENT_STATE.md @@ -198,7 +198,8 @@ protected file metadata remain unverified. ## Change in this PR Adds a read-only, resumable full-file corpus inventory and normalized-data -cross-check, synthetic regression tests, and an operator handoff runbook. CI's +cross-check, a source/staging comparator requiring nonzero documents and +fragments, synthetic regression tests, and an operator handoff runbook. CI's existing `infra/deploy` unittest discovery includes the new tests. No workflow, runner, VPS, production data, or Docker settings are changed. The first real corpus scan stopped on an integer-storage bug; full validation with the diff --git a/infra/deploy/compare_corpus_inventories.py b/infra/deploy/compare_corpus_inventories.py new file mode 100644 index 0000000..22d6c4b --- /dev/null +++ b/infra/deploy/compare_corpus_inventories.py @@ -0,0 +1,141 @@ +#!/usr/bin/env python3 +"""Require non-empty completed normalized audits and compare every file row.""" + +from __future__ import annotations + +import argparse +import itertools +import json +import sqlite3 +import stat +import sys +from pathlib import Path +from typing import Any + + +class ComparisonError(ValueError): + pass + + +def read_audit(result_path: Path, label: str) -> tuple[sqlite3.Connection, dict[str, Any]]: + result_path = result_path.absolute() + try: + metadata = result_path.lstat() + if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode): + raise ComparisonError(f"{label}: result.json must be a regular file") + result = json.loads(result_path.read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError) as error: + raise ComparisonError(f"{label}: cannot read result.json: {error}") from error + + if not isinstance(result, dict): + raise ComparisonError(f"{label}: result.json must contain an object") + if result.get("status") != "complete" or result.get("verification_status") != "passed": + raise ComparisonError(f"{label}: inventory must be complete and passed") + verifier_sha256 = result.get("verifier_sha256") + if not isinstance(verifier_sha256, str) or len(verifier_sha256) != 64: + raise ComparisonError(f"{label}: verifier SHA-256 is missing") + normalized_root = result.get("roots", {}).get("normalized") + if not isinstance(normalized_root, str): + raise ComparisonError(f"{label}: normalized root is missing") + summary = result.get("manifest_summary", {}).get("normalized") + if not isinstance(summary, dict): + raise ComparisonError(f"{label}: normalized manifest summary is missing") + for key in ("documents", "fragments"): + if not isinstance(summary.get(key), int) or summary[key] <= 0: + raise ComparisonError(f"{label}: normalized {key} must be greater than zero") + + database_path = result_path.with_name("inventory.sqlite3") + try: + db_metadata = database_path.lstat() + if ( + stat.S_ISLNK(db_metadata.st_mode) + or not stat.S_ISREG(db_metadata.st_mode) + or db_metadata.st_nlink != 1 + ): + raise ComparisonError(f"{label}: inventory.sqlite3 must be a regular unlinked file") + for suffix in ("-wal", "-shm", "-journal"): + if Path(f"{database_path}{suffix}").exists(): + raise ComparisonError(f"{label}: SQLite sidecar files are present") + connection = sqlite3.connect(database_path.resolve().as_uri() + "?mode=ro&immutable=1", uri=True) + connection.row_factory = sqlite3.Row + if connection.execute("PRAGMA quick_check").fetchone()[0] != "ok": + raise ComparisonError(f"{label}: inventory database failed quick_check") + manifest_documents = connection.execute( + "SELECT count(*) FROM manifest_docs WHERE root_id='normalized'" + ).fetchone()[0] + if manifest_documents != summary["documents"]: + raise ComparisonError(f"{label}: normalized document count does not match inventory") + files, byte_count, unfinished = connection.execute( + "SELECT count(*),coalesce(sum(size),0)," + "sum(CASE WHEN status!='hashed' OR size IS NULL OR sha256 IS NULL THEN 1 ELSE 0 END) " + "FROM files WHERE root_id='normalized'" + ).fetchone() + progress = result.get("progress", {}).get("normalized", {}) + if unfinished or files != progress.get("files") or byte_count != progress.get("bytes"): + raise ComparisonError(f"{label}: inventory rows do not match completed scan counters") + return connection, { + "root": normalized_root, + "documents": summary["documents"], + "fragments": summary["fragments"], + "files": files, + "bytes": byte_count, + "verifier_sha256": verifier_sha256, + "schema_version": result.get("schema_version"), + } + except sqlite3.Error as error: + raise ComparisonError(f"{label}: cannot validate inventory database: {error}") from error + + +def compare_inventories(source_result: Path, staging_result: Path) -> dict[str, int]: + source_db, source = read_audit(source_result, "source") + try: + staging_db, staging = read_audit(staging_result, "staging") + except Exception: + source_db.close() + raise + try: + if source["root"] == staging["root"]: + raise ComparisonError("source and staging must be different normalized roots") + if source["verifier_sha256"] != staging["verifier_sha256"] or source["schema_version"] != staging["schema_version"]: + raise ComparisonError("source and staging audits must use the same verifier and schema") + if (source["documents"], source["fragments"]) != (staging["documents"], staging["fragments"]): + raise ComparisonError("source and staging document or fragment counts differ") + + query = ( + "SELECT relative_path,size,sha256 FROM files " + "WHERE root_id='normalized' ORDER BY relative_path" + ) + left = iter(source_db.execute(query)) + right = iter(staging_db.execute(query)) + mismatch_count = 0 + for source_row, staging_row in itertools.zip_longest(left, right): + if source_row != staging_row: + mismatch_count += 1 + if mismatch_count: + raise ComparisonError(f"source and staging differ in {mismatch_count} inventory rows") + return {"documents": source["documents"], "fragments": source["fragments"], "files": source["files"], "bytes": source["bytes"]} + finally: + source_db.close() + staging_db.close() + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("source_result", type=Path) + parser.add_argument("staging_result", type=Path) + arguments = parser.parse_args() + try: + summary = compare_inventories(arguments.source_result, arguments.staging_result) + except (ComparisonError, OSError) as error: + print(f"FAIL: {error}", file=sys.stderr) + return 1 + print( + "PASS: matched " + f"{summary['documents']} documents, {summary['fragments']} fragments, " + f"{summary['files']} files, {summary['bytes']} bytes" + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/infra/deploy/test_compare_corpus_inventories.py b/infra/deploy/test_compare_corpus_inventories.py new file mode 100644 index 0000000..d77a731 --- /dev/null +++ b/infra/deploy/test_compare_corpus_inventories.py @@ -0,0 +1,94 @@ +import json +import sqlite3 +import tempfile +import unittest +from pathlib import Path + +from compare_corpus_inventories import ComparisonError, compare_inventories + + +class CompareCorpusInventoryTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.base = Path(self.temp.name) + self.source = self._artifact("source", "/data/source") + self.staging = self._artifact("staging", "/data/staging") + + def tearDown(self): + self.temp.cleanup() + + def _artifact(self, name, root, documents=1, fragments=2, rows=None): + directory = self.base / name + directory.mkdir(mode=0o700) + database = directory / "inventory.sqlite3" + connection = sqlite3.connect(database) + connection.execute( + "CREATE TABLE files (root_id TEXT, relative_path TEXT, size INTEGER, sha256 TEXT, status TEXT)" + ) + connection.execute("CREATE TABLE manifest_docs (root_id TEXT, code TEXT)") + connection.executemany( + "INSERT INTO manifest_docs VALUES ('normalized',?)", + [(str(code),) for code in range(documents)], + ) + if rows is None: + rows = [ + ("manifest.sqlite3", 16, "a" * 64), + ("documents/1/document.json", 24, "b" * 64), + ] + connection.executemany( + "INSERT INTO files VALUES ('normalized',?,?,?,'hashed')", rows + ) + connection.commit() + connection.close() + result = { + "status": "complete", + "verification_status": "passed", + "schema_version": 2, + "verifier_sha256": "c" * 64, + "roots": {"normalized": root}, + "manifest_summary": { + "normalized": {"documents": documents, "fragments": fragments} + }, + "progress": { + "normalized": { + "files": len(rows), + "bytes": sum(row[1] for row in rows), + } + }, + } + result_path = directory / "result.json" + result_path.write_text(json.dumps(result), encoding="utf-8") + return result_path + + def test_nonempty_source_and_staging_inventories_match(self): + self.assertEqual( + compare_inventories(self.source, self.staging), + {"documents": 1, "fragments": 2, "files": 2, "bytes": 40}, + ) + + def test_empty_source_is_rejected(self): + empty = self._artifact("empty-source", "/data/empty-source", documents=0, fragments=0) + with self.assertRaisesRegex(ComparisonError, "source: normalized documents"): + compare_inventories(empty, self.staging) + + def test_empty_staging_is_rejected(self): + empty = self._artifact("empty-staging", "/data/empty-staging", documents=0, fragments=0) + with self.assertRaisesRegex(ComparisonError, "staging: normalized documents"): + compare_inventories(self.source, empty) + + def test_zero_fragments_are_rejected_even_with_documents(self): + empty = self._artifact("no-fragments", "/data/no-fragments", fragments=0) + with self.assertRaisesRegex(ComparisonError, "normalized fragments"): + compare_inventories(self.source, empty) + + def test_different_file_sha_is_rejected(self): + changed = self._artifact( + "changed", "/data/changed", + rows=[("manifest.sqlite3", 16, "a" * 64), ("documents/1/document.json", 24, "d" * 64)], + ) + with self.assertRaisesRegex(ComparisonError, "differ in 1 inventory rows"): + compare_inventories(self.source, changed) + + +if __name__ == "__main__": + unittest.main() -- 2.54.0 From 0da2bcd82de73afc198ebf9b444545f97f7a8d67 Mon Sep 17 00:00:00 2001 From: agent Date: Sun, 18 Oct 2026 19:33:52 +0300 Subject: [PATCH 4/5] Lock corpus artifacts during comparison --- docs/operations/CORPUS_HANDOFF.md | 10 ++- docs/operations/CURRENT_STATE.md | 3 +- infra/deploy/compare_corpus_inventories.py | 90 +++++++++++++------ .../deploy/test_compare_corpus_inventories.py | 38 ++++++++ 4 files changed, 107 insertions(+), 34 deletions(-) diff --git a/docs/operations/CORPUS_HANDOFF.md b/docs/operations/CORPUS_HANDOFF.md index 63e9176..dc1fb10 100644 --- a/docs/operations/CORPUS_HANDOFF.md +++ b/docs/operations/CORPUS_HANDOFF.md @@ -68,10 +68,12 @@ python3 infra/deploy/compare_corpus_inventories.py \ The checker streams every relative path, size, and SHA-256 from both SQLite inventories, and requires the same verifier/schema version and different -normalized roots. A zero exit status confirms a non-empty staging tree matches -the source. Keep both trees quiescent during their scans and comparison. This -is an equality check only; the owner still reviews the code `6` disposition -and approves the transfer destination. +normalized roots. It opens both existing private `.inventory.lock` files +read-only and holds shared nonblocking locks through comparison; a missing, +unsafe, or busy lock fails closed. A zero exit status confirms a non-empty +staging tree matches the source. Keep both trees quiescent during their scans +and comparison. This is an equality check only; the owner still reviews the +code `6` disposition and approves the transfer destination. ## Current data disposition diff --git a/docs/operations/CURRENT_STATE.md b/docs/operations/CURRENT_STATE.md index 373b961..b77ddcf 100644 --- a/docs/operations/CURRENT_STATE.md +++ b/docs/operations/CURRENT_STATE.md @@ -199,7 +199,8 @@ protected file metadata remain unverified. Adds a read-only, resumable full-file corpus inventory and normalized-data cross-check, a source/staging comparator requiring nonzero documents and -fragments, synthetic regression tests, and an operator handoff runbook. CI's +fragments plus held shared artifact locks, synthetic regression tests, and an +operator handoff runbook. CI's existing `infra/deploy` unittest discovery includes the new tests. No workflow, runner, VPS, production data, or Docker settings are changed. The first real corpus scan stopped on an integer-storage bug; full validation with the diff --git a/infra/deploy/compare_corpus_inventories.py b/infra/deploy/compare_corpus_inventories.py index 22d6c4b..93028b6 100644 --- a/infra/deploy/compare_corpus_inventories.py +++ b/infra/deploy/compare_corpus_inventories.py @@ -4,11 +4,14 @@ from __future__ import annotations import argparse +import fcntl import itertools import json +import os import sqlite3 import stat import sys +from contextlib import ExitStack, contextmanager from pathlib import Path from typing import Any @@ -17,6 +20,32 @@ class ComparisonError(ValueError): pass +@contextmanager +def shared_inventory_lock(result_path: Path, label: str): + lock_path = result_path.absolute().parent / ".inventory.lock" + if not hasattr(os, "O_NOFOLLOW"): + raise ComparisonError(f"{label}: safe inventory lock opens are unsupported") + try: + descriptor = os.open(lock_path, os.O_RDONLY | os.O_NOFOLLOW) + except OSError as error: + raise ComparisonError(f"{label}: cannot open inventory lock: {error}") from error + try: + metadata = os.fstat(descriptor) + if ( + not stat.S_ISREG(metadata.st_mode) + or metadata.st_nlink != 1 + or stat.S_IMODE(metadata.st_mode) & 0o077 + ): + raise ComparisonError(f"{label}: inventory lock must be a private regular file") + try: + fcntl.flock(descriptor, fcntl.LOCK_SH | fcntl.LOCK_NB) + except BlockingIOError as error: + raise ComparisonError(f"{label}: inventory is active (lock busy)") from error + yield + finally: + os.close(descriptor) + + def read_audit(result_path: Path, label: str) -> tuple[sqlite3.Connection, dict[str, Any]]: result_path = result_path.absolute() try: @@ -87,36 +116,39 @@ def read_audit(result_path: Path, label: str) -> tuple[sqlite3.Connection, dict[ def compare_inventories(source_result: Path, staging_result: Path) -> dict[str, int]: - source_db, source = read_audit(source_result, "source") - try: - staging_db, staging = read_audit(staging_result, "staging") - except Exception: - source_db.close() - raise - try: - if source["root"] == staging["root"]: - raise ComparisonError("source and staging must be different normalized roots") - if source["verifier_sha256"] != staging["verifier_sha256"] or source["schema_version"] != staging["schema_version"]: - raise ComparisonError("source and staging audits must use the same verifier and schema") - if (source["documents"], source["fragments"]) != (staging["documents"], staging["fragments"]): - raise ComparisonError("source and staging document or fragment counts differ") + with ExitStack() as locks: + locks.enter_context(shared_inventory_lock(source_result, "source")) + locks.enter_context(shared_inventory_lock(staging_result, "staging")) + source_db, source = read_audit(source_result, "source") + try: + staging_db, staging = read_audit(staging_result, "staging") + except Exception: + source_db.close() + raise + try: + if source["root"] == staging["root"]: + raise ComparisonError("source and staging must be different normalized roots") + if source["verifier_sha256"] != staging["verifier_sha256"] or source["schema_version"] != staging["schema_version"]: + raise ComparisonError("source and staging audits must use the same verifier and schema") + if (source["documents"], source["fragments"]) != (staging["documents"], staging["fragments"]): + raise ComparisonError("source and staging document or fragment counts differ") - query = ( - "SELECT relative_path,size,sha256 FROM files " - "WHERE root_id='normalized' ORDER BY relative_path" - ) - left = iter(source_db.execute(query)) - right = iter(staging_db.execute(query)) - mismatch_count = 0 - for source_row, staging_row in itertools.zip_longest(left, right): - if source_row != staging_row: - mismatch_count += 1 - if mismatch_count: - raise ComparisonError(f"source and staging differ in {mismatch_count} inventory rows") - return {"documents": source["documents"], "fragments": source["fragments"], "files": source["files"], "bytes": source["bytes"]} - finally: - source_db.close() - staging_db.close() + query = ( + "SELECT relative_path,size,sha256 FROM files " + "WHERE root_id='normalized' ORDER BY relative_path" + ) + left = iter(source_db.execute(query)) + right = iter(staging_db.execute(query)) + mismatch_count = 0 + for source_row, staging_row in itertools.zip_longest(left, right): + if source_row != staging_row: + mismatch_count += 1 + if mismatch_count: + raise ComparisonError(f"source and staging differ in {mismatch_count} inventory rows") + return {"documents": source["documents"], "fragments": source["fragments"], "files": source["files"], "bytes": source["bytes"]} + finally: + source_db.close() + staging_db.close() def main() -> int: diff --git a/infra/deploy/test_compare_corpus_inventories.py b/infra/deploy/test_compare_corpus_inventories.py index d77a731..a346249 100644 --- a/infra/deploy/test_compare_corpus_inventories.py +++ b/infra/deploy/test_compare_corpus_inventories.py @@ -1,4 +1,6 @@ +import fcntl import json +import os import sqlite3 import tempfile import unittest @@ -20,6 +22,7 @@ class CompareCorpusInventoryTests(unittest.TestCase): def _artifact(self, name, root, documents=1, fragments=2, rows=None): directory = self.base / name directory.mkdir(mode=0o700) + (directory / ".inventory.lock").touch(mode=0o600) database = directory / "inventory.sqlite3" connection = sqlite3.connect(database) connection.execute( @@ -89,6 +92,41 @@ class CompareCorpusInventoryTests(unittest.TestCase): with self.assertRaisesRegex(ComparisonError, "differ in 1 inventory rows"): compare_inventories(self.source, changed) + def test_busy_source_lock_is_rejected(self): + self._assert_busy_lock(self.source, "source") + + def test_busy_staging_lock_is_rejected(self): + self._assert_busy_lock(self.staging, "staging") + + def _assert_busy_lock(self, result_path, label): + descriptor = os.open(result_path.parent / ".inventory.lock", os.O_RDWR) + try: + fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB) + with self.assertRaisesRegex(ComparisonError, rf"{label}: inventory is active"): + compare_inventories(self.source, self.staging) + finally: + os.close(descriptor) + + def test_missing_lock_is_rejected_without_creating_it(self): + lock = self.source.parent / ".inventory.lock" + lock.unlink() + with self.assertRaisesRegex(ComparisonError, "source: cannot open inventory lock"): + compare_inventories(self.source, self.staging) + self.assertFalse(lock.exists()) + + def test_symlink_lock_is_rejected(self): + lock = self.source.parent / ".inventory.lock" + lock.unlink() + target = self.base / "external-lock" + target.touch(mode=0o600) + lock.symlink_to(target) + with self.assertRaisesRegex(ComparisonError, "source: cannot open inventory lock"): + compare_inventories(self.source, self.staging) + + def test_read_only_lock_can_be_shared(self): + (self.source.parent / ".inventory.lock").chmod(0o400) + self.assertEqual(compare_inventories(self.source, self.staging)["documents"], 1) + if __name__ == "__main__": unittest.main() -- 2.54.0 From 6bd4b05af2c2b535a04d12ed570a3a6df3bc64ae Mon Sep 17 00:00:00 2001 From: agent Date: Sun, 18 Oct 2026 19:39:34 +0300 Subject: [PATCH 5/5] Document corpus inventory safety gates --- CHANGELOG.md | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6bac0d9..994eee8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,13 @@ ## Не выпущено +- Добавлен read-only возобновляемый inventory корпуса с полным хешированием, + потоковой сверкой, индексируемыми диапазонами путей и provenance SHA-256 + проверяющего; значения `stat` хранятся без ограничения SQLite INTEGER. + Сравнение source/staging сверяет путь, размер и SHA-256, удерживает shared + locks и отклоняет пустые корпуса без документов или фрагментов. Runbook и + `CURRENT_STATE.md` отражают, что полный аудит корпуса ещё выполняется; это + изменение не означает готовность корпуса или production. - CI actions теперь используют workspace, переданный Gitea, для Caddyfile и Kaniko build context; CI run 69 и smoke run 70 прошли на self-hosted runner. - Runner installer сохраняет `HOME` и XDG-пути в user-unit `docker.service`, перезапускает активный daemon и проверяет окружение его процесса. - Runner installer защищает image и mountpoint root-owned home-каталогом, оставляет регистрацию и workspace в runner-owned каталоге, безопасно форматирует образ по открытому FD и проверяет `nosuid,nodev`. -- 2.54.0