#!/usr/bin/env python3 """Synthetic-only restic 0.19.1 restoration experiment, Windows amd64. Python 3.11+; standard library only. Accepts one pinned public release ZIP, never an existing repository, source directory, or restore destination. All mutation occurs below a fresh temporary directory in the working directory. The public dummy password is a test fixture, not a usable security credential. """ from __future__ import annotations import argparse import csv import hashlib import io import json import os from pathlib import Path import platform import re import shutil import stat import struct import subprocess import sys import tempfile import tomllib import wave import xml.etree.ElementTree as ET import zipfile VERSION = "0.19.1" ARCHIVE_NAME = "restic_0.19.1_windows_amd64.zip" ARCHIVE_URL = f"https://github.com/restic/restic/releases/download/v{VERSION}/{ARCHIVE_NAME}" CHECKSUM_URL = f"https://github.com/restic/restic/releases/download/v{VERSION}/SHA256SUMS" ARCHIVE_SHA256 = "da948ad707ed690426473aaba2046cd61f8f90f6f0e7dab6be0d5796531de67d" BINARY_SHA256 = "b0dd1fd21eea5d8fe1325f55f7118213c21f36de8a261e04c0624a5ab9fd7830" PUBLIC_DUMMY_PASSWORD = "PUBLIC-SYNTHETIC-LAB-ONLY-NOT-A-REAL-PASSWORD-v1" PUBLIC_WRONG_PASSWORD = "PUBLIC-SYNTHETIC-LAB-ONLY-WRONG-PASSWORD-v1" FIXED_MTIME = 978307200 # 2001-01-01 UTC; synthetic fixture, not an activity date. OMITTED_PATH = "documents/contacts.csv" VERSIONED_PATH = "documents/status.json" UNREADABLE_PATH = "documents/records.json" def sha256(data: bytes) -> str: return hashlib.sha256(data).hexdigest() def refuse_links(path: Path) -> None: """シンボリックリンク、junctionなどのreparse pointを拒否する。""" for item in (path, *path.parents): if item.exists() or item.is_symlink(): info = item.lstat() if stat.S_ISLNK(info.st_mode) or getattr(info, "st_file_attributes", 0) & 0x400: raise RuntimeError(f"Link/reparse point refused: {item}") def corpus() -> dict[str, bytes]: """実在の人、画像、録音、秘密情報を含まない12個の人工ファイル。""" zipped = io.BytesIO() with zipfile.ZipFile(zipped, "w", compression=zipfile.ZIP_STORED) as archive: member = zipfile.ZipInfo("read-me.txt", date_time=(2001, 1, 1, 0, 0, 0)) archive.writestr(member, b"Synthetic archive member. No real records.\n") sound = io.BytesIO() with wave.open(sound, "wb") as audio: audio.setnchannels(1) audio.setsampwidth(2) audio.setframerate(8000) audio.writeframes(b"".join(struct.pack("synthetic\n', "documents/reading-list.xml": b"Synthetic document\n", "documents/archive.zip": zipped.getvalue(), "media/tone.wav": sound.getvalue(), "records/settings.toml": b'synthetic = true\nformat_version = 1\nlabel = "fixture"\n', } def parse_file(path: Path) -> dict: """ファイル形式を標準parserで読む。全アプリでの利用可能性は判定しない。""" parser = { ".txt": "strict UTF-8 decoder", ".json": "json.loads", ".csv": "csv.reader (strict)", ".xml": "xml.etree.ElementTree.parse", ".zip": "zipfile.ZipFile.testzip + member reads", ".wav": "wave.open + all frame reads", ".toml": "tomllib.loads", }[path.suffix] try: if path.suffix == ".txt": path.read_bytes().decode("utf-8", errors="strict") elif path.suffix == ".json": json.loads(path.read_text(encoding="utf-8")) elif path.suffix == ".csv": with path.open(encoding="utf-8", newline="") as handle: rows = list(csv.reader(handle, strict=True)) if not rows or any(len(row) != len(rows[0]) for row in rows): raise ValueError("Unexpected fixture row shape") elif path.suffix == ".xml": ET.parse(path) elif path.suffix == ".zip": with zipfile.ZipFile(path) as archive: if archive.testzip() is not None: raise ValueError("ZIP CRC check failed") for member in archive.infolist(): archive.read(member) elif path.suffix == ".wav": with wave.open(str(path), "rb") as audio: expected = audio.getnframes() * audio.getnchannels() * audio.getsampwidth() if len(audio.readframes(audio.getnframes())) != expected: raise ValueError("WAV payload length mismatch") elif path.suffix == ".toml": tomllib.loads(path.read_text(encoding="utf-8")) return {"parser": parser, "pass": True} except (OSError, ValueError, csv.Error, ET.ParseError, zipfile.BadZipFile, wave.Error, EOFError) as error: return {"parser": parser, "pass": False, "error_type": type(error).__name__} class Lab: def __init__(self, archive: Path): refuse_links(Path.cwd()) refuse_links(archive) if archive.stat().st_size > 50_000_000: raise RuntimeError("Archive exceeds the pinned release size ceiling") archive_data = archive.read_bytes() if sha256(archive_data) != ARCHIVE_SHA256: raise RuntimeError("Archive SHA256 differs from the pinned official release") self.root = Path(tempfile.mkdtemp(prefix="c07-restore-lab-", dir=Path.cwd())).resolve() print(f"LAB_ROOT={self.root}", flush=True) self.binary = self.root / "restic.exe" with zipfile.ZipFile(io.BytesIO(archive_data)) as zipped: executable = zipped.read(f"restic_{VERSION}_windows_amd64.exe") if sha256(executable) != BINARY_SHA256: raise RuntimeError("Binary SHA256 differs from the measured official binary") self.binary.write_bytes(executable) self.logs: list[dict] = [] self.results: list[dict] = [] self.manifests: dict[str, dict] = {} self.script_sha256 = sha256(Path(__file__).read_bytes()) self.env = {key: value for key, value in os.environ.items() if key.upper() in {"SYSTEMROOT", "WINDIR", "COMSPEC"}} subprocess_temp = self.guard(self.root / "subprocess-temp") subprocess_temp.mkdir() self.env["TEMP"] = str(subprocess_temp) self.env["TMP"] = str(subprocess_temp) self.env["RESTIC_PASSWORD"] = PUBLIC_DUMMY_PASSWORD self.version = subprocess.run([str(self.binary), "version"], cwd=self.root, env=self.env, stdin=subprocess.DEVNULL, capture_output=True, text=True, check=True).stdout.strip() if not self.version.startswith(f"restic {VERSION} "): raise RuntimeError("Unexpected restic version") def guard(self, path: Path) -> Path: refuse_links(path) resolved = path.resolve() if resolved == self.root or not resolved.is_relative_to(self.root): raise RuntimeError("Mutation must be strictly below the newly created lab root") return resolved def write_json(self, name: str, value: object) -> None: self.guard(self.root / name).write_text( json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def source(self, case: str, files: dict[str, bytes]) -> Path: source = self.guard(self.root / case / "source") source.mkdir(parents=True, exist_ok=False) for name, contents in files.items(): target = self.guard(source / name) target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(contents) os.utime(target, (FIXED_MTIME, FIXED_MTIME)) return source def manifest(self, case: str, source: Path, role: str = "intended") -> dict: # 保存対象を絞る前に、利用者が必要とする全ファイルを記録する。 entries = [] for path in sorted(source.rglob("*")): refuse_links(path) if path.is_file(): payload = path.read_bytes() entries.append({"path": path.relative_to(source).as_posix(), "size": len(payload), "sha256": sha256(payload), "source_parser": parse_file(path)}) result = {"role": role, "file_count": len(entries), "files": entries} self.manifests[f"{case}/{role}"] = result return result def sanitize(self, text: str) -> str: # 公開ログから実行場所を除く。日付・時刻フィールドはfixture時刻でも除く。 for path in (str(self.root), self.root.as_posix(), str(self.root).replace("\\", "\\\\")): text = text.replace(path, "") text = re.sub(r'"(mtime|atime|ctime|time|backup_start|backup_end)"\s*:\s*"[^"]*"', lambda match: f'"{match.group(1)}":""', text) text = re.sub(r'"username"\s*:\s*"[^"]*"', '"username":""', text) return text def run(self, case: str, repo: Path, args: list[str], *, wrong: bool = False, must_succeed: bool = False) -> subprocess.CompletedProcess: self.guard(repo) env = self.env.copy() if wrong: env["RESTIC_PASSWORD"] = PUBLIC_WRONG_PASSWORD command = [str(self.binary), "--repo", str(repo), "--no-cache", *args] process = subprocess.run(command, cwd=self.root / case, env=env, stdin=subprocess.DEVNULL, capture_output=True, encoding="utf-8", errors="replace", timeout=90) log = {"case": case, "arguments": [self.sanitize(arg) for arg in command[1:]], "public_password_fixture": "wrong" if wrong else "correct", "exit_code": process.returncode, "stdout": self.sanitize(process.stdout), "stderr": self.sanitize(process.stderr)} self.logs.append(log) self.write_json("command-log.json", self.logs) if must_succeed and process.returncode != 0: raise RuntimeError(f"{case}: setup failed: {log}") return process def init_repo(self, case: str) -> Path: repo = self.guard(self.root / case / "repository") self.run(case, repo, ["init", "--repository-version", "2"], must_succeed=True) return repo def backup(self, case: str, repo: Path, tag: str, exclude: str | None = None) -> str: args = ["backup", "--json", "--force", "--host", "synthetic-restore-lab", "--tag", tag] if exclude is not None: args.extend(["--exclude", f"source/{exclude}"]) args.append("source") process = self.run(case, repo, args, must_succeed=True) records = [json.loads(line) for line in process.stdout.splitlines() if line.startswith("{")] summary = next(record for record in records if record.get("message_type") == "summary") # バックアップ完了時のIDを固定し、latestを復元対象にしない。 snapshot = summary["snapshot_id"] if not re.fullmatch(r"[0-9a-f]{64}", snapshot): raise RuntimeError("Expected a full immutable snapshot ID") return snapshot def copy_repo(self, case: str, source_repo: Path) -> Path: for path in (source_repo, *source_repo.rglob("*")): refuse_links(path) destination = self.guard(self.root / case / "repository") destination.parent.mkdir(parents=True, exist_ok=True) shutil.copytree(source_repo, destination, symlinks=False) return destination def inspect_restore(self, target: Path, intended: dict) -> dict: # snapshotのsourceフォルダだけを固定subfolder構文で復元する。 paths = {} for path in target.rglob("*"): refuse_links(path) if path.is_file(): paths[path.relative_to(target).as_posix()] = path wanted = {item["path"]: item for item in intended["files"]} missing = sorted(set(wanted) - set(paths)) unexpected = sorted(set(paths) - set(wanted)) details = [] for name, item in wanted.items(): if name in paths: data = paths[name].read_bytes() details.append({"path": name, "actual_size": len(data), "actual_sha256": sha256(data), "hash_match": sha256(data) == item["sha256"], "restored_parser": parse_file(paths[name])}) hashes = sum(item["hash_match"] for item in details) parsed = sum(item["restored_parser"]["pass"] for item in details) return {"expected_count": len(wanted), "restored_count": len(paths), "inventory_pass": not missing and not unexpected, "missing": missing, "unexpected": unexpected, "matching_hash_count": hashes, "all_expected_hashes_match": hashes == len(wanted) and not unexpected, "parser_attempted_count": len(details), "parser_pass_count": parsed, "parser_failures": [item["path"] for item in details if not item["restored_parser"]["pass"]], "file_results": details} def examine(self, case: str, condition: str, repo: Path, snapshot: str, intended: dict, *, wrong: bool = False, extra: dict | None = None) -> None: check = self.run(case, repo, ["check"], wrong=wrong) read = self.run(case, repo, ["check", "--read-data"], wrong=wrong) target = self.guard(self.root / case / "restored-empty-target") target.mkdir(exist_ok=False) if any(target.iterdir()): raise RuntimeError("Restore target must be newly created and empty") listing = self.run(case, repo, ["ls", "--json", snapshot], wrong=wrong) subfolder = None if listing.returncode == 0: nodes = [json.loads(line) for line in listing.stdout.splitlines() if line.startswith("{")] candidates = [node["path"] for node in nodes if node.get("type") == "dir" and node.get("name") == "source"] if len(candidates) != 1: raise RuntimeError(f"Cannot identify unique synthetic source directory: {candidates}") subfolder = candidates[0] snapshot_argument = f"{snapshot}:{subfolder}" else: # 誤パスワードでは一覧が得られない。固定IDへの復元を実際に試す。 snapshot_argument = snapshot restored = self.run(case, repo, ["restore", "--json", snapshot_argument, "--target", str(target), "--overwrite", "never"], wrong=wrong) result = {"case": case, "condition": condition, "snapshot_id": snapshot, "source_subfolder": self.sanitize(subfolder) if subfolder else None, "check_exit": check.returncode, "read_data_exit": read.returncode, "snapshot_list_exit": listing.returncode, "restore_exit": restored.returncode, **self.inspect_restore(target, intended), **(extra or {})} self.results.append(result) self.write_json("partial-results.json", self.results) print(json.dumps({key: value for key, value in result.items() if key in {"case", "check_exit", "read_data_exit", "restore_exit", "restored_count", "matching_hash_count", "parser_pass_count"}}), flush=True) def corrupt_payload(self, case: str, repo: Path) -> dict: # 新規実験root内の使い捨てcopyにあるdata blobだけを1 byte反転する。 target_blob_id = sha256(corpus()[UNREADABLE_PATH]) indexes = self.run(case, repo, ["list", "index"], must_succeed=True).stdout.splitlines() for index_id in sorted(indexes): index = json.loads(self.run(case, repo, ["cat", "index", index_id], must_succeed=True).stdout) for pack in index["packs"]: for blob in pack["blobs"]: if blob["type"] == "data" and blob["id"] == target_blob_id and blob["length"] > 48: path = self.guard(repo / "data" / pack["id"][:2] / pack["id"]) before = path.read_bytes() # nonce(16 bytes)の次のciphertext byte。pack headerは変更しない。 offset = blob["offset"] + 16 after = bytearray(before) after[offset] ^= 1 os.chmod(path, stat.S_IREAD | stat.S_IWRITE) with path.open("r+b") as handle: handle.seek(offset) handle.write(bytes([after[offset]])) measured = path.read_bytes() if len(measured) != len(before) or measured != bytes(after): raise RuntimeError("Corruption fixture did not match the planned single-byte change") return {"target_path": UNREADABLE_PATH, "pack_id": pack["id"], "data_blob_id": blob["id"], "byte_offset": offset, "changed_byte_count": 1, "pack_size_unchanged": True, "pack_bytes": len(before), "pack_sha256_before": sha256(before), "pack_sha256_after": sha256(measured)} raise RuntimeError("The known single-blob JSON fixture was not found in the data index") def execute(self) -> None: files = corpus() if len(files) != 12: raise RuntimeError("Unexpected corpus size") source = self.source("S1", files) baseline = self.manifest("S1", source) if not all(item["source_parser"]["pass"] for item in baseline["files"]): raise RuntimeError("Baseline source parser failed") repo = self.init_repo("S1") snapshot = self.backup("S1", repo, "synthetic-baseline") self.examine("S1", "normal", repo, snapshot, baseline) wrong_repo = self.copy_repo("S2", repo) self.manifests["S2/intended"] = baseline self.examine("S2", "wrong public dummy password", wrong_repo, snapshot, baseline, wrong=True) damaged_repo = self.copy_repo("S3", repo) self.manifests["S3/intended"] = baseline damage = self.corrupt_payload("S3", damaged_repo) self.examine("S3", "one encrypted data payload byte changed in disposable copy", damaged_repo, snapshot, baseline, extra={"corruption": damage}) omitted_source = self.source("S4", files) omitted_expected = self.manifest("S4", omitted_source) omitted_repo = self.init_repo("S4") omitted_snapshot = self.backup("S4", omitted_repo, "synthetic-exclusion", OMITTED_PATH) self.examine("S4", "one required file excluded at backup", omitted_repo, omitted_snapshot, omitted_expected, extra={"excluded_path": OMITTED_PATH}) version_source = self.source("S5", files) latest_expected = self.manifest("S5", version_source) version_path = self.guard(version_source / VERSIONED_PATH) version_path.write_bytes(b'{"synthetic":true,"revision":1,"state":"previous"}\n') os.utime(version_path, (FIXED_MTIME, FIXED_MTIME)) old_expected = self.manifest("S5", version_source, "old_snapshot_source") version_repo = self.init_repo("S5") old_snapshot = self.backup("S5", version_repo, "synthetic-old-revision") version_path.write_bytes(files[VERSIONED_PATH]) os.utime(version_path, (FIXED_MTIME, FIXED_MTIME)) newer_snapshot = self.backup("S5", version_repo, "synthetic-current-revision") self.examine("S5", "older fixed snapshot restored against required current manifest", version_repo, old_snapshot, latest_expected, extra={"newer_snapshot_id": newer_snapshot, "versioned_path": VERSIONED_PATH}) target = self.root / "S5" / "restored-empty-target" self.results[-1]["old_snapshot_hash_match_count"] = self.inspect_restore(target, old_expected)["matching_hash_count"] bad_files = dict(files) bad_files[UNREADABLE_PATH] = b'{"synthetic":true,"records":[BROKEN-FIXTURE-NOT-JSON]}\n' bad_source = self.source("S6", bad_files) bad_expected = self.manifest("S6", bad_source) bad_repo = self.init_repo("S6") bad_snapshot = self.backup("S6", bad_repo, "synthetic-invalid-source") self.examine("S6", "JSON is invalid before backup", bad_repo, bad_snapshot, bad_expected, extra={"preexisting_invalid_path": UNREADABLE_PATH}) self.finish() def finish(self) -> None: if sha256(Path(__file__).read_bytes()) != self.script_sha256: raise RuntimeError("Script changed during the run; rerun the final source before publication") metadata = {"experiment": "synthetic-restic-restore-six-cases", "schema_version": 1, "restic_version_output": self.version, "python": platform.python_version(), "platform": {"system": platform.system(), "release": platform.release(), "machine": platform.machine()}, "release_archive_url": ARCHIVE_URL, "official_checksum_url": CHECKSUM_URL, "release_archive_sha256": ARCHIVE_SHA256, "binary_sha256": BINARY_SHA256, "release_checksum_match": True, "release_signature_verified": False, "script_sha256": self.script_sha256, "repository_format": 2, "cache": "disabled", "compression": "restic default auto", "corpus_file_count": 12, "public_dummy_password": PUBLIC_DUMMY_PASSWORD, "public_wrong_dummy_password": PUBLIC_WRONG_PASSWORD, "log_redactions": ["fresh lab absolute path", "JSON filesystem/snapshot/backup timestamps", "execution-account username"], "scope": "Single local Windows run; not a product ranking, media-life study, offsite test, account-recovery test, or whole-system recovery test."} self.write_json("results.json", {"metadata": metadata, "cases": self.results}) self.write_json("synthetic-manifests.json", self.manifests) fields = ["case", "condition", "check_exit", "read_data_exit", "restore_exit", "expected_count", "restored_count", "inventory_pass", "matching_hash_count", "all_expected_hashes_match", "parser_attempted_count", "parser_pass_count"] with self.guard(self.root / "results.csv").open("w", encoding="utf-8", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=fields, extrasaction="ignore") writer.writeheader() writer.writerows(self.results) print("COMPLETE: results.json, results.csv, synthetic-manifests.json, command-log.json", flush=True) print("No cleanup performed. All repositories and restored fixtures remain in LAB_ROOT.", flush=True) def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--restic-zip", type=Path, required=True, help="Pinned official restic 0.19.1 Windows amd64 ZIP; SHA256 is enforced") args = parser.parse_args() if sys.platform != "win32" or sys.version_info < (3, 11): parser.error("This measured reproduction script requires Windows and Python 3.11+") Lab(args.restic_zip.absolute()).execute() if __name__ == "__main__": main()