#!/usr/bin/env python3 """Offline auth-state migration to the actual SQLite runtime authority. Publication requires real AuthState/webhook readers, exact typed data and verified encrypted custody. Standalone webhook authorities are never discarded. AGE requires pyrage, a flinttrade-core dependency (not an optional extra). Missing crypto or reader support is a failure, never skipped work. Passwords enter only through a TTY or --master-password-fd; auth.db password/MFA material is not changed by this tool. """ from __future__ import annotations import argparse import base64 import getpass import hashlib import logging import os import sqlite3 import sys from contextlib import ExitStack, closing from pathlib import Path from flinttrade_core.auth_state_migration import record_webhook_selection, validate_staged_selection from flinttrade_core.legacy_sqlite_migration import ( ArchivePending, MigrationRefused, database_digest, file_digest, fsync_directory, fsync_file, migrate_database, publish_noreplace, require_regular, rows_digest, sqlite_connection, sqlite_value, ) from flinttrade_core.secure_file import harden from flinttrade_core.workspace import workspace_dir _TABLES = ("revoked_jtis", "rate_limit_events", "otp_requests", "webhooks", "webhook_nonces") _SCHEMA_DDL = """ CREATE TABLE IF NOT EXISTS revoked_jtis ( jti TEXT PRIMARY KEY, revoked_at REAL NOT NULL ); CREATE TABLE IF NOT EXISTS rate_limit_events ( user_id TEXT NOT NULL, endpoint TEXT NOT NULL, bucket_ts REAL NOT NULL, count INTEGER NOT NULL, PRIMARY KEY (user_id, endpoint, bucket_ts) ); CREATE TABLE IF NOT EXISTS otp_requests ( request_id TEXT PRIMARY KEY, user_id TEXT NOT NULL, purpose TEXT NOT NULL, code_hash TEXT NOT NULL, attempts_left INTEGER NOT NULL DEFAULT 3, expires_at REAL NOT NULL, created_at REAL NOT NULL ); CREATE INDEX IF NOT EXISTS idx_otp_requests_expires_at ON otp_requests(expires_at) WHERE attempts_left > 0; CREATE TABLE IF NOT EXISTS consent_ledger ( id INTEGER PRIMARY KEY, purpose TEXT NOT NULL, granted_at REAL NOT NULL, revoked_at REAL, artifact_hash TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_consent_ledger_purpose_active ON consent_ledger(purpose, revoked_at); CREATE TABLE IF NOT EXISTS agent_runs ( run_id TEXT PRIMARY KEY, agent_name TEXT NOT NULL, parent_jti TEXT NOT NULL, started_at REAL NOT NULL, ended_at REAL, status TEXT NOT NULL CHECK (status IN ( 'running', 'completed', 'failed', 'cancelled' )), final_decision_count INTEGER, pnl_delta REAL ); CREATE INDEX IF NOT EXISTS idx_agent_runs_parent_jti ON agent_runs(parent_jti); CREATE INDEX IF NOT EXISTS idx_agent_runs_agent_started ON agent_runs(agent_name, started_at DESC); CREATE TABLE IF NOT EXISTS webhooks ( webhook_id TEXT PRIMARY KEY, source TEXT NOT NULL, name TEXT NOT NULL, user_id TEXT NOT NULL, secret_encrypted_aes_gcm_siv BLOB NOT NULL, dek_id TEXT NOT NULL, key_version INTEGER NOT NULL, created_at REAL NOT NULL, last_rotated_at REAL NOT NULL, revoked_at REAL, CHECK (length(webhook_id) > 0) ); CREATE INDEX IF NOT EXISTS idx_webhooks_source_active ON webhooks(source) WHERE revoked_at IS NULL; CREATE INDEX IF NOT EXISTS idx_webhooks_user_source ON webhooks(user_id, source) WHERE revoked_at IS NULL; CREATE TABLE IF NOT EXISTS webhook_nonces ( webhook_id TEXT NOT NULL REFERENCES webhooks(webhook_id) ON DELETE CASCADE, nonce TEXT NOT NULL, seen_at REAL NOT NULL, source_ip_hash TEXT, PRIMARY KEY (webhook_id, nonce) ); CREATE INDEX IF NOT EXISTS idx_webhook_nonces_gc ON webhook_nonces(seen_at); CREATE TABLE IF NOT EXISTS login_sessions ( consent_app_id TEXT PRIMARY KEY, broker_id TEXT NOT NULL, encrypted_api_key BLOB NOT NULL, encrypted_api_secret BLOB NOT NULL, nonce BLOB NOT NULL, dek_salt BLOB NOT NULL, created_at REAL NOT NULL, expires_at REAL NOT NULL, CHECK (length(consent_app_id) > 0), CHECK (length(dek_salt) >= 32), CHECK (length(nonce) >= 12) ); CREATE INDEX IF NOT EXISTS idx_login_sessions_expires ON login_sessions(expires_at); CREATE TABLE IF NOT EXISTS audit_chain_head ( chain_id TEXT PRIMARY KEY, last_event_hash TEXT NOT NULL, last_event_ts REAL NOT NULL, updated_at REAL NOT NULL ); CREATE TABLE IF NOT EXISTS auth_sessions ( jti TEXT PRIMARY KEY, sub TEXT NOT NULL, actor_type TEXT NOT NULL CHECK (actor_type IN ( 'human', 'agent', 'external_intent' )), run_id TEXT REFERENCES agent_runs(run_id), issued_at REAL NOT NULL, last_seen_at REAL NOT NULL, ip_hash TEXT NOT NULL, ua_hash TEXT NOT NULL, location_summary TEXT, device_summary TEXT, fp_cookie_hash TEXT, created_via TEXT NOT NULL CHECK (created_via IN ( 'password_totp', 'password_only', 'broker_sso', 'magic_link', 'jwt_refresh', 'agent_spawn' )), revoked_at REAL ); CREATE INDEX IF NOT EXISTS idx_auth_sessions_sub_active ON auth_sessions( sub, actor_type, last_seen_at DESC, jti, ip_hash, ua_hash, location_summary, device_summary, created_via, issued_at ) WHERE revoked_at IS NULL; """ def create_schema(sqlite_path: str | os.PathLike[str]) -> None: """Retain the historical schema utility; NOT a runtime migration guarantee.""" with closing(sqlite3.connect(sqlite_path)) as conn: conn.executescript(_SCHEMA_DDL) def _create_staging_schema(sqlite_path: str | os.PathLike[str]) -> None: """Create a staging schema from the actual AuthState and webhook owners.""" from flinttrade_core.auth_state import _CREATE_SCHEMA_SQL from flinttrade_webhooks.webhook_replay import ensure_nonce_schema from flinttrade_webhooks.webhook_secret_store import _WEBHOOK_SECRET_SCHEMA with closing(sqlite3.connect(sqlite_path)) as conn: conn.executescript(_CREATE_SCHEMA_SQL) conn.executescript(_WEBHOOK_SECRET_SCHEMA) ensure_nonce_schema(conn) def _read_master_password(args: argparse.Namespace) -> str: if args.master_password_fd is not None: with os.fdopen(args.master_password_fd, "r") as stream: password = stream.readline().rstrip("\n") elif sys.stdin.isatty(): password = getpass.getpass("master password: ") else: raise MigrationRefused("master password requires a TTY or --master-password-fd") if not password: raise MigrationRefused("empty master password") return password def _plaintext(blob: bytes, webhook_id: str, version: int, master_dek: bytes, legacy_key: bytes | None) -> bytes: from cryptography.exceptions import InvalidTag from cryptography.fernet import Fernet, InvalidToken from flinttrade_webhooks.webhook_keys import decrypt_webhook_secret if not isinstance(blob, bytes) or type(version) is not int or version < 1: raise MigrationRefused("invalid webhook ciphertext metadata") try: return decrypt_webhook_secret(blob, webhook_id, master_dek, version) except (InvalidTag, ValueError): if legacy_key is None: raise MigrationRefused("webhook decryption failed; source retained") from None # Normal Fernet tokens are base64, not a leading raw 0x80 byte. token = base64.urlsafe_b64encode(blob) if blob[:1] == b"\x80" else blob try: return Fernet(legacy_key).decrypt(token) except (InvalidToken, ValueError): raise MigrationRefused("webhook decryption failed; source retained") from None def _transcode_webhook_secrets(sqlite_path: Path, legacy_fernet_key: bytes | None, master_password: str) -> int: """Authenticate every row and transcode verified Fernet in one transaction.""" from cryptography.exceptions import InvalidTag from flinttrade_webhooks.webhook_keys import decrypt_webhook_secret, encrypt_webhook_secret from flinttrade_webhooks.webhook_secret_store import WebhookSecretStore, canonical_webhook_id master_dek = WebhookSecretStore._derive_master_dek(master_password) updates = [] with closing(sqlite_connection(sqlite_path)) as conn: rows = conn.execute("SELECT webhook_id,path,key_version,secret_encrypted_aes_gcm_siv FROM webhooks").fetchall() for webhook_id, path, version, blob in rows: if not isinstance(path, str) or not path or canonical_webhook_id(path) != webhook_id: raise MigrationRefused("webhook identity/path mismatch") plaintext = _plaintext(blob, webhook_id, version, master_dek, legacy_fernet_key) plaintext.decode("utf-8") # actual get_secret contract try: decrypt_webhook_secret(blob, webhook_id, master_dek, version) except (InvalidTag, ValueError): encrypted = encrypt_webhook_secret(plaintext, webhook_id, master_dek, version) if decrypt_webhook_secret(encrypted, webhook_id, master_dek, version) != plaintext: raise MigrationRefused("webhook transcode verification failed") updates.append((encrypted, webhook_id)) # Do not reset versions, identities or rotation times during transcoding. with conn: conn.executemany("UPDATE webhooks SET secret_encrypted_aes_gcm_siv=? WHERE webhook_id=?", updates) return len(updates) def _archive_legacy(source: Path | None, archive: Path, expected: str | None, master_password: str) -> None: """Verify a durable real AGE archive before the coordinator unlinks source.""" from pyrage import passphrase if source is not None and file_digest(source) != expected: raise MigrationRefused("source changed before encrypted archival") require_regular(archive) temporary = archive.with_name(archive.name + ".tmp") require_regular(temporary) def verify(path: Path) -> None: if hashlib.sha256(passphrase.decrypt(path.read_bytes(), master_password)).hexdigest() != expected: raise MigrationRefused("encrypted archive does not match original") if temporary.exists(): # Interrupted encryption is evidence, not permission to truncate a # possibly unrelated archive. Reuse only an authenticated exact copy. verify(temporary) if not archive.exists(): if not temporary.exists(): if source is None: raise MigrationRefused("source and archive both missing") payload = source.read_bytes() encrypted = passphrase.encrypt(payload, master_password) if passphrase.decrypt(encrypted, master_password) != payload: raise MigrationRefused("AGE encryption round-trip failed") fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0), 0o600) with os.fdopen(fd, "wb") as stream: stream.write(encrypted) stream.flush() os.fsync(stream.fileno()) harden(temporary) verify(temporary) fsync_file(temporary) fsync_directory(temporary.parent) publish_noreplace(temporary, archive) verify(archive) fsync_file(archive) fsync_directory(archive.parent) if temporary.exists(): temporary.unlink() fsync_directory(archive.parent) def _source_rows(source: Path | None, target: Path) -> dict[str, tuple[list[str], list[tuple]]]: if source is None: return {} import duckdb with duckdb.connect(str(source), read_only=True, config={"threads": 1}) as duck: actual = {r[0] for r in duck.execute("SHOW TABLES").fetchall()} required, optional = set(_TABLES[:3]), set(_TABLES[3:]) if not required <= actual or actual - required - optional or (actual & optional not in (set(), optional)): raise MigrationRefused("unsupported/incomplete auth schema; no tables may be discarded") if duck.execute("SELECT count(*) FROM information_schema.tables WHERE table_schema <> 'main' " "OR table_type <> 'BASE TABLE'").fetchone()[0]: raise MigrationRefused("unsupported auth source objects") result = {} with closing(sqlite_connection(target, readonly=True)) as conn: for table in _TABLES: if table not in actual: continue info = conn.execute(f"PRAGMA table_info({table})").fetchall() columns = [r[1] for r in info] if {r[0] for r in duck.execute(f"DESCRIBE {table}").fetchall()} != set(columns): raise MigrationRefused("auth column set mismatch; no fields may be discarded") affinities = {"VARCHAR": "TEXT", "DOUBLE": "REAL"} rows = duck.execute(f"SELECT {','.join(columns)} FROM {table}").fetchall() converted = [tuple(sqlite_value(value, affinities.get(info[i][2], info[i][2])) for i, value in enumerate(row)) for row in rows] if any(row[i] is None and (col[3] or col[5]) for row in converted for i, col in enumerate(info)): raise MigrationRefused("null auth key or required value") result[table] = (columns, converted) return result def _validate_rows(source: Path | None, target: Path, password: str | None, legacy_key: bytes | None) -> None: from flinttrade_webhooks.webhook_secret_store import WebhookSecretStore tables = _source_rows(source, target) database_digest(target) with closing(sqlite_connection(target, readonly=True)) as conn: for table in _TABLES: columns, rows = tables.get(table, ([], [])) actual = conn.execute(f"SELECT {','.join(columns) if columns else '*'} FROM {table}").fetchall() if table == "webhooks" and rows: master_dek = WebhookSecretStore._derive_master_dek(password) def plaintext_rows(values, key, columns=columns, master_dek=master_dek): converted = [] for value in values: value = list(value) i = columns.index("secret_encrypted_aes_gcm_siv") value[i] = _plaintext(value[i], value[columns.index("webhook_id")], value[columns.index("key_version")], master_dek, key) converted.append(tuple(value)) return converted rows, actual = plaintext_rows(rows, legacy_key), plaintext_rows(actual, None) if rows_digest(actual) != rows_digest(rows): raise MigrationRefused("auth data/crypto relationship mismatch") def _validate_current_readers(target: Path, password: str | None) -> None: from flinttrade_core.auth_state import AuthState, _default_db_path from flinttrade_webhooks.webhook_secret_store import WebhookSecretStore if _default_db_path().name != "auth_state.sqlite": raise MigrationRefused("auth_state_reader_cutover_required; active DuckDB authority retained") # Constructors can perform expiry GC. Test a backup, never change the stage. probe = target.with_name("reader-probe.sqlite") require_regular(probe) probe.unlink(missing_ok=True) with closing(sqlite_connection(target, readonly=True)) as conn, closing(sqlite3.connect(probe)) as copy: conn.backup(copy) jtis = conn.execute("SELECT jti FROM revoked_jtis").fetchall() hooks = conn.execute("SELECT path,webhook_id,key_version,secret_encrypted_aes_gcm_siv FROM webhooks").fetchall() try: reader = AuthState(probe) try: if not all(reader.is_jti_revoked(row[0]) for row in jtis): raise MigrationRefused("current AuthState reader lost revocation") finally: reader.close() if hooks: store = WebhookSecretStore(probe, password) for path, webhook_id, version, blob in hooks: if store.get_secret(path).encode() != _plaintext(blob, webhook_id, version, store._master_dek, None): raise MigrationRefused("current webhook reader mismatch") finally: for suffix in ("", "-wal", "-shm"): probe.with_name(probe.name + suffix).unlink(missing_ok=True) def migrate(workspace: Path, args: argparse.Namespace) -> int: """Stage exact authority; publish only after real reader and crypto parity.""" password, legacy_key = None, None def credentials(): nonlocal password, legacy_key if password is None: password = _read_master_password(args) if args.legacy_fernet_key_path: key_path = Path(args.legacy_fernet_key_path) require_regular(key_path) legacy_key = key_path.read_bytes() return password def build(source, target): if args.accept_unknown_tables: raise MigrationRefused("discarding unknown auth tables is not supported") _create_staging_schema(target) tables = _source_rows(source, target) with closing(sqlite_connection(target)) as conn: conn.execute("PRAGMA synchronous=FULL") with conn: for table, (columns, rows) in tables.items(): conn.executemany(f"INSERT INTO {table} ({','.join(columns)}) VALUES " f"({','.join('?' for _ in columns)})", rows) if tables.get("webhooks", ([], []))[1]: credentials() _transcode_webhook_secrets(target, legacy_key, password) record_webhook_selection(workspace, target) def validate(source, target): if source is not None: credentials() _validate_rows(source, target, password, legacy_key) _validate_current_readers(target, password) validate_staged_selection(workspace, target) if source is not None: from pyrage import passphrase # Import and check actual crypto BEFORE publication, not afterwards. if not callable(passphrase.encrypt) or not callable(passphrase.decrypt): raise MigrationRefused("incompatible pyrage dependency") credentials() def archive(source, target, expected): validate_staged_selection(workspace, workspace / "auth_state.sqlite") _archive_legacy(source, target, expected, credentials()) with ExitStack() as fences: standalone = workspace / "webhook_secrets.db" require_regular(standalone) if standalone.exists(): # Exclude standalone SQLite writers too, without changing their # schema, credentials, nonces or journal mode. fence = fences.enter_context(closing(sqlite_connection(standalone))) fence.execute("BEGIN IMMEDIATE") migrate_database(workspace, source_name="auth_state.duckdb", destination_name="auth_state.sqlite", build=build, validate=validate, archive_codec=archive, archive_suffix=".age") return 0 def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--workspace", type=Path, default=None) parser.add_argument("--master-password-fd", type=int, default=None) parser.add_argument("--accept-unknown-tables", action="store_true", help="Retired: unknown data is never discarded") parser.add_argument("--legacy-fernet-key-path", type=str, default=None) args = parser.parse_args() logging.basicConfig(level=logging.CRITICAL) try: migrate(args.workspace if args.workspace is not None else workspace_dir(), args) except ArchivePending: print("published/archive-pending: destination retained; rerun to resume encrypted archival") return 3 except Exception as exc: code = "auth_state_reader_cutover_required" if isinstance(exc, MigrationRefused) and str(exc).startswith( "auth_state_reader_cutover_required") else type(exc).__name__ print(f"migration refused ({code}); originals retained; reconcile source/dependencies") return 2 print("auth migration complete; destination and encrypted archive verified") return 0 if __name__ == "__main__": raise SystemExit(main())