Фрагмент
\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 + 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_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 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 TEXT, + inode TEXT, + mtime_ns TEXT, + ctime_ns TEXT, + 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 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, + 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, + "verifier_sha256": TOOL_SHA256, + "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", + *(str(value) for value in (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), + ) + 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), + ) + 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>=? AND relative_path 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>=? AND relative_path 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>=? AND relative_path 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 any(path.name != ".inventory.lock" for path in output_dir.iterdir()): + 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("verifier_sha256") != TOOL_SHA256: + raise InventoryError("verifier code 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)), + ("verifier_sha256", json.dumps(TOOL_SHA256)), + ("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") + output_dir = validate_output_dir(output_dir, roots) + with inventory_lock(output_dir): + return _run_inventory( + roots, output_dir, resume, checkpoint_every, stop_after_entries, progress_callback + ) + + +def _run_inventory( + roots: dict[str, Path], + output_dir: Path, + resume: bool, + checkpoint_every: int, + stop_after_entries: int | None, + progress_callback: Callable[[dict[str, Any]], None] | None, +) -> 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} + 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())