dc73e9cbf6
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade
95 lines
4.5 KiB
Text
95 lines
4.5 KiB
Text
"""Pre-open schema/integrity/count proof; stdlib only, never constructs Store."""
|
|
import argparse
|
|
from contextlib import closing
|
|
import hashlib
|
|
import os
|
|
import re
|
|
import stat
|
|
import json
|
|
from pathlib import Path
|
|
import sqlite3
|
|
|
|
|
|
def inspect_readonly(path, *, _closed_snapshot=False):
|
|
path = Path(path)
|
|
if not path.is_file() or path.is_symlink():
|
|
raise ValueError('existing nonsymlink private database required')
|
|
uri = path.resolve().as_uri() + '?mode=ro' + ('&immutable=1' if _closed_snapshot else '')
|
|
# sqlite3.Connection context management ends transactions; it does not close.
|
|
with closing(sqlite3.connect(uri, uri=True, timeout=5)) as db:
|
|
db.execute('PRAGMA query_only=ON')
|
|
db.execute('BEGIN')
|
|
version = db.execute('PRAGMA user_version').fetchone()[0]
|
|
if version != 5:
|
|
raise ValueError('pre-open database is not schema5; refuse migration')
|
|
if db.execute('PRAGMA integrity_check').fetchall() != [('ok',)]:
|
|
raise ValueError('read-only integrity check refused')
|
|
lifecycle = {name: db.execute('SELECT COUNT(*) FROM ' + table).fetchone()[0]
|
|
for name, table in (('records', 'lifecycle_records'),
|
|
('events', 'lifecycle_events'),
|
|
('pending', 'lifecycle_pending'))}
|
|
result = {'schema_version': version, 'lifecycle': lifecycle,
|
|
'outbox': dict(db.execute('SELECT state,COUNT(*) FROM outbox GROUP BY state')),
|
|
'submissions': dict(db.execute('SELECT state,COUNT(*) FROM submissions GROUP BY state')),
|
|
'integrity': 'ok', 'read_only_pre_open': True, 'store_constructed': False}
|
|
db.rollback()
|
|
return result
|
|
|
|
|
|
def inspect_closed_snapshot(path, expected_sha256):
|
|
"""Only a privately retained, frozen main-file snapshot with no WAL frames.
|
|
|
|
Caller must establish exclusive snapshot creation/admin exit/no writer, and
|
|
supply the recorded frozen main digest. Never use for the live database.
|
|
Immutable SQLite reads ignore stale SHM and never create/change sidecars.
|
|
Existing sidecars remain custody evidence; nothing is deleted/checkpointed.
|
|
"""
|
|
path = Path(path)
|
|
if re.fullmatch(r'[0-9a-f]{64}', expected_sha256) is None:
|
|
raise ValueError('exact frozen snapshot digest required')
|
|
def metadata(target):
|
|
info = target.lstat()
|
|
if not stat.S_ISREG(info.st_mode) or stat.S_IMODE(info.st_mode) != 0o600 or info.st_nlink != 1:
|
|
raise ValueError('private single-link snapshot custody required')
|
|
return info
|
|
main = metadata(path)
|
|
if not 0 < main.st_size <= 512 * 1024**2:
|
|
raise ValueError('bounded frozen snapshot required')
|
|
def sidecars():
|
|
identities = {}
|
|
for suffix in ('-wal', '-shm'):
|
|
target = Path(str(path) + suffix)
|
|
if os.path.lexists(target):
|
|
info = metadata(target)
|
|
if info.st_uid != main.st_uid or info.st_gid != main.st_gid:
|
|
raise ValueError('snapshot sidecar custody changed')
|
|
if suffix == '-wal' and info.st_size != 0:
|
|
raise ValueError('snapshot has WAL frames; immutable read refused')
|
|
identities[suffix] = (info.st_dev, info.st_ino, info.st_size, info.st_mtime_ns)
|
|
return identities
|
|
before_sidecars = sidecars()
|
|
def digest():
|
|
value = hashlib.sha256()
|
|
with path.open('rb') as stream:
|
|
for block in iter(lambda: stream.read(1024**2), b''):
|
|
value.update(block)
|
|
return value.hexdigest()
|
|
if digest() != expected_sha256:
|
|
raise ValueError('frozen main-file digest changed before inspection')
|
|
proof = inspect_readonly(path, _closed_snapshot=True)
|
|
after = metadata(path)
|
|
if (after.st_dev, after.st_ino, after.st_uid, after.st_gid, after.st_size) != (main.st_dev, main.st_ino, main.st_uid, main.st_gid, main.st_size):
|
|
raise ValueError('frozen main-file identity changed')
|
|
if digest() != expected_sha256 or sidecars() != before_sidecars:
|
|
raise ValueError('frozen snapshot changed during inspection')
|
|
return proof
|
|
|
|
|
|
if __name__ == '__main__':
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--db', type=Path, required=True)
|
|
args = parser.parse_args()
|
|
try:
|
|
print(json.dumps(inspect_readonly(args.db), sort_keys=True))
|
|
except Exception:
|
|
raise SystemExit('read-only schema5/integrity/count gate refused; no Store/admin open') from None
|