#!/usr/bin/env python3
"""Create a private SQLite inspection copy without opening the source in SQLite.
Python 3.10+, POSIX, standard library. Copies DB + WAL, never source SHM.
Two observations can detect changes; they do NOT establish an atomic snapshot.
No production writes, cleanup, checkpoint, decryption, or application control.
"""
import argparse
import hashlib
import json
import os
from pathlib import Path
import sqlite3
import stat
import struct
import sys
import time
from urllib.parse import quote
LIMIT = 256 * 1024 * 1024
DIR_FLAGS = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW
READ_FLAGS = os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK
SUFFIXES = ("", "-wal", "-shm", "-journal")
COPY_NAMES = {"": "source.sqlite", "-wal": "source.sqlite-wal"}
class Refused(Exception):
pass
def require(condition, code):
if not condition:
raise Refused(code)
def absolute(value):
path = Path(value)
require(path.is_absolute() and ".." not in path.parts, "INVALID_PATH")
require(path.name not in ("", ".", ".."), "INVALID_PATH")
return path
def open_directory(path):
"""Walk anchored descriptors; do not silently resolve symlink ancestors."""
require(path.is_absolute(), "INVALID_PATH")
fd = os.open("/", DIR_FLAGS)
try:
for part in path.parts[1:]:
require(part not in ("", ".", ".."), "INVALID_PATH")
child = os.open(part, DIR_FLAGS, dir_fd=fd)
os.close(fd)
fd = child
return fd
except BaseException:
os.close(fd)
raise
def fingerprint(s):
return (s.st_dev, s.st_ino, s.st_mode, s.st_uid, s.st_gid,
s.st_nlink, s.st_size, s.st_mtime_ns, s.st_ctime_ns)
def same_directory(path, fd):
current = open_directory(path)
try:
a, b = os.fstat(current), os.fstat(fd)
require((a.st_dev, a.st_ino) == (b.st_dev, b.st_ino), "SOURCE_CHANGED")
finally:
os.close(current)
def inventory(parent, base):
result = {}
for suffix in SUFFIXES:
try:
s = os.stat(base + suffix, dir_fd=parent, follow_symlinks=False)
except FileNotFoundError:
continue
require(stat.S_ISREG(s.st_mode), "UNSAFE_SOURCE")
require(s.st_uid == os.getuid(), "SOURCE_NOT_OWNED")
result[suffix] = fingerprint(s)
require("" in result, "SOURCE_MISSING")
require("-journal" not in result or result["-journal"][6] == 0,
"ROLLBACK_JOURNAL_PRESENT")
require(sum(s[6] for s in result.values()) <= LIMIT, "SIZE_LIMIT")
return result
def read_member(parent, name, expected, output=None, on_write=None):
fd = os.open(name, READ_FLAGS, dir_fd=parent)
try:
require(fingerprint(os.fstat(fd)) == expected, "SOURCE_CHANGED")
digest = hashlib.sha256()
size = 0
while True:
block = os.read(fd, 1024 * 1024)
if not block:
break
size += len(block)
require(size <= LIMIT, "SIZE_LIMIT")
digest.update(block)
if output is not None:
view = memoryview(block)
while view:
n = os.write(output, view)
require(n > 0, "OUTPUT_IO_ERROR")
view = view[n:]
if on_write:
on_write()
require(size == expected[6], "SOURCE_CHANGED")
require(fingerprint(os.fstat(fd)) == expected, "SOURCE_CHANGED")
require(fingerprint(os.stat(name, dir_fd=parent, follow_symlinks=False))
== expected, "SOURCE_CHANGED")
return digest.hexdigest()
finally:
os.close(fd)
def write_json(parent, name, value, owned):
fd = os.open(name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
0o600, dir_fd=parent)
try:
owned[name] = fingerprint(os.fstat(fd))
data = memoryview((json.dumps(value, ensure_ascii=True, indent=2) + "\n").encode())
while data:
count = os.write(fd, data)
require(count > 0, "OUTPUT_IO_ERROR")
data = data[count:]
owned[name] = fingerprint(os.fstat(fd))
os.fsync(fd)
finally:
os.close(fd)
def check_wal_header(directory, size):
"""Reject plainly unsupported WAL, without claiming to validate frame replay."""
if size == 0:
return "empty"
require(size >= 32, "WAL_HEADER_UNSUPPORTED")
fd = os.open("source.sqlite-wal", READ_FLAGS, dir_fd=directory)
try:
header = os.read(fd, 32)
finally:
os.close(fd)
require(len(header) == 32, "WAL_HEADER_UNSUPPORTED")
magic, version, page_size = struct.unpack(">III", header[:12])
require(magic in (0x377F0682, 0x377F0683) and version == 3007000 and
512 <= page_size <= 65536 and page_size & (page_size - 1) == 0,
"WAL_HEADER_UNSUPPORTED")
return "basic_header_only"
def verify_copy(path, observe_engine_files=None):
deadline = time.monotonic() + 15
conn = None
try:
conn = sqlite3.connect("file:" + quote(str(path), safe="/") + "?mode=ro",
uri=True, timeout=1)
conn.set_progress_handler(lambda: int(time.monotonic() > deadline), 1000)
conn.execute("PRAGMA query_only=ON")
conn.execute("PRAGMA trusted_schema=OFF")
require(conn.execute("PRAGMA quick_check").fetchall() == [("ok",)],
"SQLITE_CHECK_FAILED")
finally:
if conn is not None:
conn.close()
if observe_engine_files is not None:
observe_engine_files()
def remove_own_output(parent, name, directory, owned):
"""Remove only known regular output members, never a replacement directory."""
a = os.stat(name, dir_fd=parent, follow_symlinks=False)
b = os.fstat(directory)
require((a.st_dev, a.st_ino) == (b.st_dev, b.st_ino), "CLEANUP_INCOMPLETE")
members = os.listdir(directory)
for member in members:
require(member in owned, "CLEANUP_INCOMPLETE")
item = os.stat(member, dir_fd=directory, follow_symlinks=False)
require(stat.S_ISREG(item.st_mode) and fingerprint(item) == owned[member],
"CLEANUP_INCOMPLETE")
for member in members:
# Recheck before unlink; an observed later write must survive failure.
require(fingerprint(os.stat(member, dir_fd=directory, follow_symlinks=False))
== owned[member], "CLEANUP_INCOMPLETE")
os.unlink(member, dir_fd=directory)
os.rmdir(name, dir_fd=parent)
def snapshot(source, output, expected_sha256=None):
source, output = absolute(source), absolute(output)
require(not output.is_relative_to(source.parent), "OUTPUT_INSIDE_SOURCE")
if expected_sha256 is not None:
require(len(expected_sha256) == 64 and
all(c in "0123456789abcdef" for c in expected_sha256),
"INVALID_EXPECTED_HASH")
src = out_parent = dest = None
created = success = False
owned = {}
try:
src = open_directory(source.parent)
out_parent = open_directory(output.parent)
parent_stat = os.fstat(out_parent)
require(parent_stat.st_uid == os.getuid() and
stat.S_IMODE(parent_stat.st_mode) == 0o700, "OUTPUT_NOT_PRIVATE")
before = inventory(src, source.name)
shm_hash = (read_member(src, source.name + "-shm", before["-shm"])
if "-shm" in before else None)
os.mkdir(output.name, 0o700, dir_fd=out_parent)
created = True
dest = os.open(output.name, DIR_FLAGS, dir_fd=out_parent)
copied = {}
for suffix, name in COPY_NAMES.items():
if suffix not in before:
continue
fd = os.open(name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
0o600, dir_fd=dest)
try:
owned[name] = fingerprint(os.fstat(fd))
def remember_write():
owned[name] = fingerprint(os.fstat(fd))
digest = read_member(src, source.name + suffix, before[suffix], fd,
remember_write)
os.fsync(fd)
finally:
os.close(fd)
copied[suffix] = {"file": name, "sha256": digest, "bytes": before[suffix][6]}
require(expected_sha256 is None or copied[""]["sha256"] == expected_sha256,
"SOURCE_HASH_MISMATCH")
require(inventory(src, source.name) == before, "SOURCE_CHANGED")
same_directory(output.parent, out_parent)
same_directory(output, dest)
wal_check = (check_wal_header(dest, copied["-wal"]["bytes"])
if "-wal" in copied else "absent")
def remember_engine_files():
for name in ("source.sqlite-shm", "source.sqlite-wal"):
if name in owned:
continue
try:
info = os.stat(name, dir_fd=dest, follow_symlinks=False)
except FileNotFoundError:
continue
require(stat.S_ISREG(info.st_mode), "UNEXPECTED_OUTPUT")
owned[name] = fingerprint(info)
verify_copy(output / "source.sqlite", remember_engine_files)
# The engine may create private SHM; source DB/WAL are never engine-opened.
require(inventory(src, source.name) == before, "SOURCE_CHANGED")
for suffix, item in copied.items():
require(read_member(src, source.name + suffix, before[suffix]) == item["sha256"],
"SOURCE_CHANGED")
current = os.stat(item["file"], dir_fd=dest, follow_symlinks=False)
require(stat.S_ISREG(current.st_mode), "COPY_CHANGED")
require(read_member(dest, item["file"], fingerprint(current)) == item["sha256"],
"COPY_CHANGED")
if "-shm" in before:
require(read_member(src, source.name + "-shm", before["-shm"]) == shm_hash,
"SOURCE_CHANGED")
require(inventory(src, source.name) == before, "SOURCE_CHANGED")
same_directory(source.parent, src)
same_directory(output.parent, out_parent)
same_directory(output, dest)
# SQLite-created files inherit process umask, but explicitly restrict them too.
for name in os.listdir(dest):
require(name in {"source.sqlite", "source.sqlite-wal", "source.sqlite-shm"},
"UNEXPECTED_OUTPUT")
item = os.stat(name, dir_fd=dest, follow_symlinks=False)
require(stat.S_ISREG(item.st_mode) and name in owned and
fingerprint(item) == owned[name], "OUTPUT_CHANGED")
if stat.S_IMODE(item.st_mode) != 0o600:
fd = os.open(name, READ_FLAGS, dir_fd=dest)
try:
require(fingerprint(os.fstat(fd)) == owned[name], "OUTPUT_CHANGED")
os.fchmod(fd, 0o600)
owned[name] = fingerprint(os.fstat(fd))
finally:
os.close(fd)
report = {
"schema": 1, "status": "verified_inspection_copy", "sqlite_version": sqlite3.sqlite_version,
"atomic_snapshot": False, "source_opened_by_sqlite": False,
"source_shm_copied": False, "observed_source_stable": True,
"wal_check": wal_check, "wal_frame_semantics_verified": False,
"quick_check": "ok", "copies": list(copied.values()),
"source_sidecars_present": [s for s in SUFFIXES[1:] if s in before],
"contains_private_source_data": True,
}
write_json(dest, "report.json", report, owned)
os.fsync(dest)
success = True
return {"status": report["status"], "copied_files": len(copied),
"bytes": sum(i["bytes"] for i in copied.values()), "atomic_snapshot": False}
finally:
cleanup_failed = False
if created and not success:
try:
if dest is None:
os.rmdir(output.name, dir_fd=out_parent)
else:
remove_own_output(out_parent, output.name, dest, owned)
except (OSError, Refused):
cleanup_failed = True
for fd in (dest, out_parent, src):
if fd is not None:
os.close(fd)
if cleanup_failed:
raise Refused("CLEANUP_INCOMPLETE") from None
class Parser(argparse.ArgumentParser):
def error(self, message):
raise Refused("INVALID_ARGUMENTS")
def main():
parser = Parser(description=__doc__)
parser.add_argument("--source", required=True)
parser.add_argument("--out-dir", required=True)
parser.add_argument("--expected-sha256")
try:
args = parser.parse_args()
os.umask(0o077)
print(json.dumps(snapshot(args.source, args.out_dir, args.expected_sha256)))
return 0
except Refused as exc:
print(str(exc), file=sys.stderr)
except FileExistsError:
print("OUTPUT_EXISTS", file=sys.stderr)
except sqlite3.Error:
print("SQLITE_CHECK_FAILED", file=sys.stderr)
except OSError:
print("FILESYSTEM_ERROR", file=sys.stderr)
return 2
if __name__ == "__main__":
sys.exit(main())
← 返回 Skill 说明
v2.1.0 / Python 工具
工具源码