#!/usr/bin/env python3 """Mutate only disposable, self-created PostgreSQL clusters. Never a live database.""" import argparse from datetime import datetime, timezone import hashlib import json import os from pathlib import Path import platform import shutil import shlex import subprocess import tempfile MARKER = b"HESELA_TOAST_TAIL_0123456789_abcdefghijklmnop" PAYLOAD = b"A" * 24000 + MARKER ENV = {k: v for k, v in os.environ.items() if not k.startswith("PG") and k not in {"LD_PRELOAD", "LD_LIBRARY_PATH"}} ENV.update(LC_ALL="C", LANG="C") PORT = "55439" def digest(value): return hashlib.sha256(value).hexdigest() def run(args, required=False): proc = subprocess.run([str(x) for x in args], text=True, capture_output=True, timeout=60, env=ENV) result = dict(returncode=proc.returncode, stdout=proc.stdout.strip(), stderr=proc.stderr.strip()) if required and proc.returncode: raise RuntimeError(f"{args[0]} failed: {result}") return result def mutate(data): if data.count(MARKER) != 1: raise ValueError("Expected exactly one complete fixture marker") offset = data.index(MARKER) changed = bytearray(data) changed[offset] ^= 1 changed = bytes(changed) assert sum(a != b for a, b in zip(data, changed)) == 1 return changed, offset def verify_case(case): clean, damaged = case["clean"], case["damaged"] for name, item in clean["probes"].items(): assert item["returncode"] == 0, (name, item) assert clean["probes"]["whole_sha256"]["stdout"] == digest(PAYLOAD) for name in ("row_count", "byte_length", "prefix", "heap_without_toast", "amcheck_neither"): assert damaged["probes"][name]["returncode"] == 0, (name, damaged["probes"][name]) assert damaged["probes"][name]["stdout"] == clean["probes"][name]["stdout"], name for state in (clean, damaged): assert state["probes"]["row_count"]["stdout"] == "1" assert state["probes"]["byte_length"]["stdout"] == str(len(PAYLOAD)) assert state["probes"]["prefix"]["stdout"] == (b"A" * 32).hex() assert state["probes"]["heap_without_toast"]["stdout"] == "[]" if case["checksums"]: assert clean["offline_checksums"]["returncode"] == 0 for name in ("whole_sha256", "heap_with_toast", "amcheck_default", "amcheck_no_pointers", "amcheck_no_dependent_toast"): item = damaged["probes"][name] assert item["returncode"] != 0, (name, item) assert "checksum" in item["stderr"].lower() or "invalid page" in item["stderr"].lower(), item assert damaged["offline_checksums"]["returncode"] != 0 assert "Bad checksums: 1" in damaged["offline_checksums"]["stdout"] else: for name, item in damaged["probes"].items(): assert item["returncode"] == 0, (name, item) expected = bytearray(PAYLOAD) expected[24000] ^= 1 assert damaged["probes"]["whole_sha256"]["stdout"] == digest(expected) assert damaged["probes"]["whole_sha256"]["stdout"] != digest(PAYLOAD) assert damaged["probes"]["heap_with_toast"]["stdout"] == "[]" def case_run(server_bin, client_bin, scratch, checksums): root = Path(tempfile.mkdtemp(prefix="ht-", dir=scratch)) data, sock = root / "data", root / "s" sock.mkdir(mode=0o700) running = False conn = ["-h", sock, "-p", PORT, "-U", "hesela_lab", "-d", "postgres"] def sql(statement, required=True): return run([client_bin / "psql", "-X", "-A", "-t", "-v", "ON_ERROR_STOP=1", *conn, "-c", statement], required) def start(): nonlocal running options = shlex.join(["-k", str(sock), "-p", PORT, "-h", "", "-c", "shared_buffers=16MB", "-c", "max_connections=10", "-c", "autovacuum=off"]) run([server_bin / "pg_ctl", "-D", data, "-l", root / "server.log", "-o", options, "-w", "start"], True) running = True def stop(): nonlocal running run([server_bin / "pg_ctl", "-D", data, "-m", "fast", "-w", "stop"], True) running = False def probes(): statements = { "row_count": "SELECT count(*) FROM public.payloads;", "byte_length": "SELECT octet_length(body) FROM public.payloads;", "prefix": "SELECT encode(substring(body FROM 1 FOR 32), 'hex') FROM public.payloads;", "whole_sha256": "SELECT encode(sha256(body), 'hex') FROM public.payloads;", "heap_without_toast": "SELECT coalesce(json_agg(v), '[]') FROM verify_heapam('public.payloads', false, false, 'none') v;", "heap_with_toast": "SELECT coalesce(json_agg(v), '[]') FROM verify_heapam('public.payloads', false, true, 'none') v;", } output = {name: dict(sql=statement, **sql(statement, False)) for name, statement in statements.items()} variants = { "amcheck_default": [], "amcheck_no_pointers": ["--exclude-toast-pointers"], "amcheck_no_dependent_toast": ["--no-dependent-toast"], "amcheck_neither": ["--exclude-toast-pointers", "--no-dependent-toast"], } for name, flags in variants.items(): options = ["--table=public.payloads", "--no-dependent-indexes", *flags] output[name] = dict(options=options, **run([client_bin / "pg_amcheck", *conn, *options])) return output try: init_args = [server_bin / "initdb", "-D", data, "--username=hesela_lab", "--auth-local=trust", "--auth-host=reject", "--no-locale", "-E", "UTF8"] if checksums: init_args.append("--data-checksums") run(init_args, True) start() sql("CREATE EXTENSION amcheck; CREATE TABLE public.payloads(id integer PRIMARY KEY, body bytea); ALTER TABLE public.payloads ALTER COLUMN body SET STORAGE EXTERNAL;") sql(f"INSERT INTO public.payloads VALUES (1, convert_to(repeat('A', 24000) || '{MARKER.decode()}', 'UTF8'));") sql("VACUUM FREEZE public.payloads;") settings = sql("SELECT json_build_object('version', version(), 'checksums', current_setting('data_checksums'), 'block_size', current_setting('block_size'), 'server_encoding', current_setting('server_encoding'), 'ignore_checksum_failure', current_setting('ignore_checksum_failure'));")["stdout"] relpath = sql("SELECT pg_relation_filepath(reltoastrelid) FROM pg_class WHERE oid='public.payloads'::regclass;")["stdout"] chunks = sql("SELECT format('SELECT count(*) FROM %s', reltoastrelid::regclass) FROM pg_class WHERE oid='public.payloads'::regclass;")["stdout"] chunk_count = int(sql(chunks)["stdout"]) clean = dict(probes=probes()) stop() clean["offline_checksums"] = run([server_bin / "pg_checksums", "--check", "-D", data]) relation = (data / relpath).resolve() if not relation.is_relative_to(data.resolve()): raise ValueError("Unexpected relation path") original = relation.read_bytes() changed, offset = mutate(original) assert offset // 8192 > 0, "Damage must be beyond the first TOAST page" relation.write_bytes(changed) mutation = dict(relation=relpath, byte_offset=offset, block=offset // 8192, byte_within_block=offset % 8192, before=original[offset], after=changed[offset], xor_mask=1, changed_bytes=1, relation_bytes=len(original), sha256_before=digest(original), sha256_after=digest(changed), checksum_recomputed=False) start() damaged = dict(probes=probes()) stop() damaged["offline_checksums"] = run([server_bin / "pg_checksums", "--check", "-D", data]) result = dict(checksums=checksums, settings=json.loads(settings), toast_chunks=chunk_count, mutation=mutation, clean=clean, damaged=damaged) verify_case(result) return result finally: if running or (data / "postmaster.pid").exists(): stop() # Remove only this invocation's privately created, stopped fixture. shutil.rmtree(root) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--server-bin", type=Path, required=True) parser.add_argument("--client-bin", type=Path, required=True) parser.add_argument("--scratch-root", type=Path, required=True) parser.add_argument("--output", type=Path, required=True) args = parser.parse_args() if not __debug__: parser.error("Run without -O: assertions are required") if os.geteuid() == 0: parser.error("Run unprivileged; no existing cluster is accepted") server_bin, client_bin = args.server_bin.resolve(), args.client_bin.resolve() scratch = args.scratch_root.resolve() for directory, names in ((server_bin, ("postgres", "initdb", "pg_ctl", "pg_checksums")), (client_bin, ("psql", "pg_amcheck"))): for name in names: if not (directory / name).is_file(): parser.error(f"Missing binary {name}") if not scratch.is_dir() or " " in str(scratch) or len(str(scratch)) > 65: parser.error("Use an existing short scratch path without spaces") cases = [case_run(server_bin, client_bin, scratch, enabled) for enabled in (True, False)] result = dict( schema_version=1, generated_at=datetime.now(timezone.utc).isoformat(), experiment="postgresql-toast-coverage", harness_sha256=digest(Path(__file__).read_bytes()), evidence_type="Actual PostgreSQL on synthetic data with one deliberate offline bit flip", environment=dict(python=platform.python_version(), system=platform.system(), kernel=platform.release(), architecture=platform.machine(), postgres=run([server_bin / "postgres", "--version"], True)["stdout"], psql=run([client_bin / "psql", "--version"], True)["stdout"], pg_amcheck=run([client_bin / "pg_amcheck", "--version"], True)["stdout"], postgres_sha256=digest((server_bin / "postgres").read_bytes())), fixture=dict(rows=1, payload_bytes=len(PAYLOAD), storage="EXTERNAL", expected_sha256=digest(PAYLOAD), prefix_bytes=32, marker=MARKER.decode(), synthetic=True), cases=cases, assertions_passed=True, limits=["Not a hardware-failure rate or exhaustive corruption campaign.", "Uncompressed bytea only; not compressed text, indexes, concurrency or recovery.", "Successful structural checking is not an expected-content comparison.", "Checksum settings differ across separately initialized clusters."]) args.output.write_text(json.dumps(result, indent=2) + "\n") print(json.dumps({"assertions_passed": True, "states": 4, "online_probes": 40, "output": str(args.output)})) if __name__ == "__main__": main()